diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 3365a89faea..75eea5673b7 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -108,6 +108,10 @@ import { redactDetectedSuccessfulRunProgressSummaryForBoard, redactSuccessfulRunHandoffEvidence, } from "../services/heartbeat.ts"; +import { + buildFeedbackDeliveryFingerprint, + buildFeedbackDeliveryRetryIdempotencyKey, +} from "../services/recovery/feedback-delivery.ts"; import { readHotRestartIntent, resolveLegacyHotRestartIntentPath, @@ -976,6 +980,195 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { return { companyId, agentId, runId, wakeupRequestId, issueId, stageId }; } + /** + * PAP-17302: the exact shape immediate terminalization missed on PAP-17271 — + * agent-assigned `in_review` whose only review path was a Board comment wake, + * whose confirmation the comment expired, and whose comment run failed + * pre-launch leaving no participant, interaction, run, wake, or recovery path. + */ + async function seedStrandedFeedbackReviewFixture(input?: { + retryGeneration?: number; + rootWakeActorType?: "user" | "agent" | "system"; + agentStatus?: "idle" | "paused"; + assignToUser?: boolean; + monitorNextCheckAt?: Date | null; + interactionStatus?: "pending" | "expired" | null; + runStatus?: "failed" | "succeeded"; + typedReviewParticipant?: boolean; + }) { + const companyId = randomUUID(); + const agentId = randomUUID(); + const runId = randomUUID(); + const rootWakeupRequestId = randomUUID(); + const issueId = randomUUID(); + const commentId = randomUUID(); + const interactionId = randomUUID(); + const now = new Date("2026-03-19T00:00:00.000Z"); + const finishedAt = new Date("2026-03-19T00:05:00.000Z"); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const retryGeneration = input?.retryGeneration ?? 0; + const replayWakeupRequestId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + defaultResponsibleUserId: "responsible-user", + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoderPro", + role: "engineer", + status: input?.agentStatus ?? "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + + await db.insert(agentWakeupRequests).values({ + id: rootWakeupRequestId, + companyId, + agentId, + source: "on_demand", + triggerDetail: "callback", + reason: "issue_commented", + payload: { issueId, commentId }, + status: "failed", + runId: retryGeneration === 0 ? runId : null, + requestedByActorType: input?.rootWakeActorType ?? "user", + requestedByActorId: "responsible-user", + requestedAt: now, + claimedAt: now, + finishedAt, + updatedAt: finishedAt, + }); + + if (retryGeneration > 0) { + // The generation-1 replay wake the immediate lane already queued and which + // itself failed; it keeps the original user wake id in its context. + await db.insert(agentWakeupRequests).values({ + id: replayWakeupRequestId, + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "feedback_delivery_retry", + payload: { issueId, rootWakeupRequestId, commentIds: [commentId] }, + status: "failed", + runId, + requestedByActorType: "system", + requestedAt: finishedAt, + claimedAt: finishedAt, + finishedAt: new Date("2026-03-19T00:10:00.000Z"), + updatedAt: new Date("2026-03-19T00:10:00.000Z"), + }); + } + + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "on_demand", + triggerDetail: "callback", + status: input?.runStatus ?? "failed", + wakeupRequestId: retryGeneration === 0 ? rootWakeupRequestId : replayWakeupRequestId, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: retryGeneration === 0 ? "issue_commented" : "feedback_delivery_retry", + wakeCommentIds: [commentId], + commentId, + wakeCommentId: commentId, + ...(retryGeneration > 0 + ? { + feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId, + feedbackDeliverySourceRunId: randomUUID(), + feedbackDeliveryRetryGeneration: retryGeneration, + retryReason: "feedback_delivery_retry", + } + : {}), + }, + startedAt: now, + finishedAt, + updatedAt: finishedAt, + errorCode: input?.runStatus === "succeeded" ? null : "process_lost", + error: input?.runStatus === "succeeded" ? null : "Process lost -- server may have restarted", + // Pre-launch loss: no process identity was ever persisted. + processPid: null, + processGroupId: null, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Plan feedback never reached the assignee", + status: "in_review", + priority: "medium", + assigneeAgentId: input?.assignToUser ? null : agentId, + assigneeUserId: input?.assignToUser ? "user-1" : null, + executionRunId: null, + monitorNextCheckAt: input?.monitorNextCheckAt ?? null, + responsibleUserId: "responsible-user", + issueNumber: 1, + identifier: `${issuePrefix}-1`, + startedAt: now, + ...(input?.typedReviewParticipant + ? { + executionState: { + status: "pending", + currentStageId: randomUUID(), + currentStageIndex: 0, + currentStageType: "review", + currentParticipant: { type: "agent", agentId, userId: null }, + returnAssignee: { type: "agent", agentId, userId: null }, + reviewRequest: null, + completedStageIds: [], + lastDecisionId: null, + lastDecisionOutcome: null, + }, + } + : {}), + }); + + await db.insert(issueComments).values({ + id: commentId, + companyId, + issueId, + authorType: "user", + authorUserId: "responsible-user", + body: "Please revise the plan before I approve it.", + createdAt: now, + updatedAt: now, + }); + + if (input?.interactionStatus) { + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: input.interactionStatus, + continuationPolicy: "wake_assignee", + payload: { + version: 1, + prompt: "Approve the plan?", + target: { type: "issue_document", issueId, key: "plan" }, + }, + createdByAgentId: agentId, + sourceRunId: runId, + ...(input.interactionStatus === "expired" + ? { resolvedAt: now, result: { version: 1, outcome: "superseded_by_comment" } } + : {}), + }); + } + + return { companyId, agentId, runId, rootWakeupRequestId, issueId, commentId, interactionId }; + } + async function seedAssignedTodoNoRunFixture(input?: { agentStatus?: "paused" | "idle" | "running"; }) { @@ -5741,6 +5934,287 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("requeues one normal-model recovery for stranded in-review feedback and never loops", async () => { + const { companyId, agentId, issueId, runId, rootWakeupRequestId, commentId } = + await seedStrandedFeedbackReviewFixture({ interactionStatus: "expired" }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.strandedFeedbackRequeued).toBe(1); + expect(result.strandedFeedbackRecoveryActions).toBe(0); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const replayRuns = await db + .select() + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.id} <> ${runId}`)); + expect(replayRuns).toHaveLength(1); + expect(replayRuns[0]).toMatchObject({ retryOfRunId: runId }); + expect(replayRuns[0]?.contextSnapshot).toMatchObject({ + issueId, + taskId: issueId, + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + source: "issue.stranded_feedback_delivery_backstop", + feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId, + feedbackDeliveryRetryGeneration: 1, + feedbackDeliveryBackstop: true, + wakeCommentIds: [commentId], + commentId, + }); + // Recovery must be a normal-model replay, not a cheap status-only nudge. + expect((replayRuns[0]?.contextSnapshot as Record).modelProfile).toBeUndefined(); + + const replayWake = await db + .select() + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.reason, FEEDBACK_DELIVERY_RETRY_WAKE_REASON), + )); + expect(replayWake).toHaveLength(1); + + const activity = await db + .select() + .from(activityLog) + .where(and(eq(activityLog.entityId, issueId), eq(activityLog.action, "issue.feedback_retry_queued"))); + expect(activity).toHaveLength(1); + expect(activity[0]?.details).toMatchObject({ + detectedBy: "issue.stranded_feedback_delivery_backstop", + rootWakeupRequestId, + sourceCommentIds: [commentId], + retryGeneration: 1, + }); + + // Bounded: a second sweep while the replay is still live must not open a + // parallel lane, and no recovery action is created yet. + const second = await heartbeat.reconcileStrandedAssignedIssues(); + expect(second.strandedFeedbackRequeued).toBe(0); + expect(second.strandedFeedbackRecoveryActions).toBe(0); + const runsAfter = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runsAfter).toHaveLength(2); + const actionsAfter = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actionsAfter).toHaveLength(0); + }); + + it("opens one explicit recovery action when stranded feedback replay is exhausted", async () => { + const { companyId, agentId, issueId, runId, rootWakeupRequestId, commentId } = + await seedStrandedFeedbackReviewFixture({ retryGeneration: 1, interactionStatus: "expired" }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.strandedFeedbackRequeued).toBe(0); + expect(result.strandedFeedbackRecoveryActions).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + // Exhaustion must not queue another replay run. + const runs = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(1); + expect(runs[0]?.id).toBe(runId); + + const actions = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actions).toHaveLength(1); + expect(actions[0]).toMatchObject({ + companyId, + kind: "feedback_delivery", + cause: "feedback_delivery_exhausted", + status: "active", + ownerType: "agent", + ownerAgentId: agentId, + previousOwnerAgentId: agentId, + returnOwnerAgentId: agentId, + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId, + issueId, + agentId, + rootWakeupRequestId, + }), + }); + expect(actions[0]?.nextAction).toContain("CodexCoderPro"); + expect(actions[0]?.nextAction).toContain("record a valid disposition"); + expect(actions[0]?.evidence).toMatchObject({ + detectedBy: "issue.stranded_feedback_delivery_backstop", + backstopReason: "retry_exhausted", + rootWakeupRequestId, + outstandingCommentIds: [commentId], + retryGeneration: 1, + }); + expect(actions[0]?.wakePolicy).toMatchObject({ type: "manual_repair_required" }); + + // The issue stays in_review with the recovery action as its explicit path. + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0]); + expect(issue?.status).toBe("in_review"); + + const activity = await db + .select() + .from(activityLog) + .where(and(eq(activityLog.entityId, issueId), eq(activityLog.action, "issue.feedback_recovery_exhausted"))); + expect(activity).toHaveLength(1); + + // No loop: repeated sweeps neither duplicate the action nor add a replay. + const second = await heartbeat.reconcileStrandedAssignedIssues(); + expect(second.strandedFeedbackRecoveryActions).toBe(0); + expect(second.strandedFeedbackRequeued).toBe(0); + const actionsAfter = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actionsAfter).toHaveLength(1); + expect(actionsAfter[0]?.id).toBe(actions[0]?.id); + const activityAfter = await db + .select({ id: activityLog.id }) + .from(activityLog) + .where(and(eq(activityLog.entityId, issueId), eq(activityLog.action, "issue.feedback_recovery_exhausted"))); + expect(activityAfter).toHaveLength(1); + }); + + it("does not open a second feedback lane when immediate terminalization already queued the replay", async () => { + const { companyId, agentId, issueId, rootWakeupRequestId } = + await seedStrandedFeedbackReviewFixture({ interactionStatus: "expired" }); + // Immediate lane already claimed generation 1 with the shared key. + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + payload: { issueId }, + status: "skipped", + requestedByActorType: "system", + idempotencyKey: buildFeedbackDeliveryRetryIdempotencyKey({ + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId, + issueId, + agentId, + rootWakeupRequestId, + }), + generation: 1, + }), + finishedAt: new Date("2026-03-19T00:06:00.000Z"), + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.strandedFeedbackRequeued).toBe(0); + expect(result.strandedFeedbackRecoveryActions).toBe(0); + expect(result.issueIds).not.toContain(issueId); + }); + + it("does not open a second feedback lane when a recovery action already owns the issue", async () => { + const { companyId, agentId, issueId, rootWakeupRequestId } = + await seedStrandedFeedbackReviewFixture({ retryGeneration: 1, interactionStatus: "expired" }); + const existingActionId = randomUUID(); + await db.insert(issueRecoveryActions).values({ + id: existingActionId, + companyId, + sourceIssueId: issueId, + kind: "feedback_delivery", + status: "active", + ownerType: "agent", + ownerAgentId: agentId, + cause: "feedback_delivery_exhausted", + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId, + issueId, + agentId, + rootWakeupRequestId, + }), + evidence: {}, + nextAction: "Immediate lane already owns this feedback batch.", + attemptCount: 1, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.strandedFeedbackRequeued).toBe(0); + expect(result.strandedFeedbackRecoveryActions).toBe(0); + + const actions = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actions).toHaveLength(1); + expect(actions[0]?.id).toBe(existingActionId); + expect(actions[0]?.attemptCount).toBe(1); + }); + + it.each([ + ["a pending Board confirmation", { interactionStatus: "pending" as const }], + ["a human-owned review", { assignToUser: true }], + ["a monitor wait", { + interactionStatus: "expired" as const, + monitorNextCheckAt: new Date("2036-03-19T00:00:00.000Z"), + }], + ["a paused assignee", { interactionStatus: "expired" as const, agentStatus: "paused" as const }], + ["a system-originated wake", { interactionStatus: "expired" as const, rootWakeActorType: "agent" as const }], + ["a successful latest run", { interactionStatus: "expired" as const, runStatus: "succeeded" as const }], + ])("leaves stranded-feedback lookalikes untouched: %s", async (_label, overrides) => { + const { issueId } = await seedStrandedFeedbackReviewFixture(overrides); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.strandedFeedbackRequeued).toBe(0); + expect(result.issueIds).not.toContain(issueId); + + const actions = await db + .select({ id: issueRecoveryActions.id, kind: issueRecoveryActions.kind }) + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actions.filter((action) => action.kind === "feedback_delivery")).toHaveLength(0); + }); + + it("leaves an in-review issue with a typed participant to the existing participant lane", async () => { + const { issueId } = await seedStrandedFeedbackReviewFixture({ + interactionStatus: "expired", + typedReviewParticipant: true, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.strandedFeedbackRequeued).toBe(0); + expect(result.strandedFeedbackRecoveryActions).toBe(0); + // The pre-existing execution-review participant lane still owns the issue. + expect(result.reviewParticipantRequeued).toBe(1); + + const actions = await db + .select({ kind: issueRecoveryActions.kind }) + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actions.filter((action) => action.kind === "feedback_delivery")).toHaveLength(0); + }); + + it.each([ + ["done", "done"], + ["cancelled", "cancelled"], + ])("leaves %s issues with stranded feedback context untouched", async (_label, status) => { + const { issueId } = await seedStrandedFeedbackReviewFixture({ interactionStatus: "expired" }); + await db.update(issues).set({ status }).where(eq(issues.id, issueId)); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.strandedFeedbackRequeued).toBe(0); + expect(result.strandedFeedbackRecoveryActions).toBe(0); + + const actions = await db + .select({ id: issueRecoveryActions.id }) + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actions).toHaveLength(0); + }); + it("escalates accepted interaction continuation recovery after three review-park cancellations", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/index.ts b/server/src/index.ts index c338564356d..6c581eeb666 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1130,6 +1130,8 @@ export async function startServer(): Promise { reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || reconciled.successfulRunHandoffEscalated > 0 || + reconciled.strandedFeedbackRequeued > 0 || + reconciled.strandedFeedbackRecoveryActions > 0 || reconciled.escalated > 0 ) { logger.warn( @@ -1344,6 +1346,8 @@ export async function startServer(): Promise { reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || reconciled.successfulRunHandoffEscalated > 0 || + reconciled.strandedFeedbackRequeued > 0 || + reconciled.strandedFeedbackRecoveryActions > 0 || reconciled.escalated > 0 ) { logger.warn( diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index b47dc5fea2c..d5cf25c9195 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -223,6 +223,16 @@ import { isSuccessfulRunHandoffValidPathSkip, SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY, readContinuationAttempt, + FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + FEEDBACK_DELIVERY_RECOVERY_CAUSE, + FEEDBACK_DELIVERY_RETRY_REASON, + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + buildFeedbackDeliveryFingerprint, + buildFeedbackDeliveryRetryIdempotencyKey, + carriesExplicitFeedbackResume, + isEligibleFeedbackDeliveryWake, + readFeedbackDeliveryRetryGeneration, } from "./recovery/index.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "./recovery/pause-hold-guard.js"; import { @@ -424,10 +434,7 @@ export const WORKSPACE_BUSY_RETRY_WAKE_REASON = "workspace_busy_retry"; export const WORKSPACE_BUSY_ERROR_CODE = "workspace_busy"; export const WORKSPACE_BUSY_RETRY_BASE_DELAY_MS = 60 * 1000; export const WORKSPACE_BUSY_RETRY_JITTER_MS = 60 * 1000; -export const FEEDBACK_DELIVERY_RETRY_REASON = "feedback_delivery_retry"; -export const FEEDBACK_DELIVERY_RETRY_WAKE_REASON = "feedback_delivery_retry"; -const FEEDBACK_DELIVERY_RECOVERY_CAUSE = "feedback_delivery_exhausted"; -const FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES = 1; +export { FEEDBACK_DELIVERY_RETRY_REASON, FEEDBACK_DELIVERY_RETRY_WAKE_REASON }; // A running run stops counting as a shared-workspace holder once it has been // silent this long. This is recovery's own "suspicious silence" bar for active // runs (scanSilentActiveRuns escalates such runs), so a zombie holder cannot @@ -10093,17 +10100,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const wakeReason = readNonEmptyString(context.wakeReason); const commentIds = mergeWakeCommentIds(context, deriveCommentId(context, null)); - const carriesExplicitResume = context.resumeIntent === true || context.followUpRequested === true; - const eligibleWake = - wakeReason === "issue_commented" || - wakeReason === "issue_reopened_via_comment" || - wakeReason === FEEDBACK_DELIVERY_RETRY_WAKE_REASON || - (carriesExplicitResume && ( - wakeReason === "issue_assigned" || - wakeReason === "issue_status_changed" || - wakeReason === "issue_commented" || - wakeReason === "issue_reopened_via_comment" - )); + const carriesExplicitResume = carriesExplicitFeedbackResume(context); + const eligibleWake = isEligibleFeedbackDeliveryWake({ wakeReason, carriesExplicitResume }); if (!eligibleWake || (commentIds.length === 0 && !carriesExplicitResume)) return null; const rootWake = await db @@ -10119,28 +10117,23 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) // generation-1 wakes retain their original user wake id in context. if (rootWake?.requestedByActorType !== "user") return null; - const generation = Math.max( - 0, - Math.floor( - typeof context.feedbackDeliveryRetryGeneration === "number" - ? context.feedbackDeliveryRetryGeneration - : run.scheduledRetryReason === FEEDBACK_DELIVERY_RETRY_REASON - ? run.scheduledRetryAttempt ?? 0 - : 0, - ), - ); return { issueId, rootWakeupRequestId, commentIds, - generation, - fingerprint: [ - "feedback_delivery", - run.companyId, + generation: readFeedbackDeliveryRetryGeneration({ + contextSnapshot: context, + scheduledRetryReason: run.scheduledRetryReason, + scheduledRetryAttempt: run.scheduledRetryAttempt, + }), + // Shared with the periodic backstop so neither lane can drift into a + // second parallel delivery lane for the same feedback batch. + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId: run.companyId, issueId, - run.agentId, + agentId: run.agentId, rootWakeupRequestId, - ].join(":"), + }), }; } @@ -10261,7 +10254,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const existing = await recoveryActions.getActiveForIssue(input.run.companyId, issue.id); if ( - existing?.kind === "feedback_delivery" + existing?.kind === FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND && existing.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE && existing.fingerprint === input.delivery.fingerprint ) { @@ -10271,7 +10264,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const action = await recoveryActions.upsertSourceScoped({ companyId: input.run.companyId, sourceIssueId: issue.id, - kind: "feedback_delivery", + kind: FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, ownerType: "agent", ownerAgentId: input.agent.id, previousOwnerAgentId: input.agent.id, @@ -10353,7 +10346,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const taskKey = deriveTaskKeyWithHeartbeatFallback(retryContext, null); const sessionBefore = await resolveSessionBeforeForWakeup(input.agent, taskKey); const responsibleUserId = await resolveResponsibleUserIdForRunContext(input.run, retryContext); - const idempotencyKey = `${input.delivery.fingerprint}:generation:1`; + const idempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({ + fingerprint: input.delivery.fingerprint, + generation: 1, + }); const result = await db.transaction(async (tx) => { const companyIsActive = await tx diff --git a/server/src/services/recovery/feedback-delivery.test.ts b/server/src/services/recovery/feedback-delivery.test.ts new file mode 100644 index 00000000000..b3e65e5f7e0 --- /dev/null +++ b/server/src/services/recovery/feedback-delivery.test.ts @@ -0,0 +1,333 @@ +import { describe, expect, it } from "vitest"; +import { + FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + buildFeedbackDeliveryFingerprint, + buildFeedbackDeliveryRetryIdempotencyKey, + decideStrandedFeedbackDeliveryBackstop, + readFeedbackDeliveryRunContext, + type StrandedFeedbackDeliveryBackstopProbe, +} from "./feedback-delivery.js"; + +const COMPANY_ID = "company-1"; +const ISSUE_ID = "issue-1"; +const AGENT_ID = "agent-1"; +const ROOT_WAKE_ID = "wake-root-1"; +const COMMENT_ID = "comment-1"; + +const openProbe: StrandedFeedbackDeliveryBackstopProbe = { + hasActiveExecutionPath: false, + hasPendingWakeInteraction: false, + hasPendingApproval: false, + hasPersistedDurableWaitPath: false, + hasQueuedIssueWake: false, + hasActiveRecoveryAction: false, + hasExistingRetryWake: false, + assigneeInvokable: true, + assigneeHold: null, + budgetBlocked: false, + pauseHeld: false, +}; + +function failedFeedbackRun(overrides: Record = {}) { + const { + contextSnapshot: contextOverrides, + ...runOverrides + } = overrides as { contextSnapshot?: Record } & Record; + return { + id: "run-1", + agentId: AGENT_ID, + status: "failed", + errorCode: "process_lost", + wakeupRequestId: ROOT_WAKE_ID, + contextSnapshot: { + issueId: ISSUE_ID, + taskId: ISSUE_ID, + wakeReason: "issue_commented", + wakeCommentIds: [COMMENT_ID], + ...(contextOverrides ?? {}), + }, + scheduledRetryReason: null, + scheduledRetryAttempt: null, + ...runOverrides, + }; +} + +function decide(input: { + probe?: Partial; + issue?: Partial<{ + id: string; + status: string; + assigneeAgentId: string | null; + assigneeUserId: string | null; + }>; + latestRun?: ReturnType | null; + hasTypedReviewParticipant?: boolean; + rootWakeRequestedByActorType?: string | null; +} = {}) { + return decideStrandedFeedbackDeliveryBackstop({ + companyId: COMPANY_ID, + issue: { + id: ISSUE_ID, + status: "in_review", + assigneeAgentId: AGENT_ID, + assigneeUserId: null, + ...(input.issue ?? {}), + }, + hasTypedReviewParticipant: input.hasTypedReviewParticipant ?? false, + latestRun: input.latestRun === undefined ? failedFeedbackRun() : input.latestRun, + rootWakeRequestedByActorType: input.rootWakeRequestedByActorType ?? "user", + probe: { ...openProbe, ...(input.probe ?? {}) }, + }); +} + +const FINGERPRINT = buildFeedbackDeliveryFingerprint({ + companyId: COMPANY_ID, + issueId: ISSUE_ID, + agentId: AGENT_ID, + rootWakeupRequestId: ROOT_WAKE_ID, +}); + +describe("stranded feedback delivery backstop", () => { + it("requeues exactly one normal-model recovery for the original assignee", () => { + const decision = decide(); + + expect(decision).toMatchObject({ + kind: "requeue", + generation: 1, + idempotencyKey: buildFeedbackDeliveryRetryIdempotencyKey({ + fingerprint: FINGERPRINT, + generation: 1, + }), + payload: { + issueId: ISSUE_ID, + retryOfRunId: "run-1", + feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID, + feedbackDeliveryRetryGeneration: 1, + feedbackDeliveryBackstop: true, + wakeCommentIds: [COMMENT_ID], + commentId: COMMENT_ID, + }, + contextSnapshot: { + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + source: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + }, + }); + if (decision.kind !== "requeue") throw new Error("expected a requeue decision"); + // Normal-model recovery must not carry the cheap/status-only guard hints. + expect(decision.contextSnapshot.modelProfile).toBeUndefined(); + expect(decision.contextSnapshot.recoveryIntent).toBeUndefined(); + expect(decision.delivery.agentId).toBe(AGENT_ID); + }); + + it("shares the immediate lane's fingerprint and replay key so there is never a second lane", () => { + const immediate = readFeedbackDeliveryRunContext({ + companyId: COMPANY_ID, + run: failedFeedbackRun(), + }); + const decision = decide(); + + expect(immediate?.fingerprint).toBe(FINGERPRINT); + if (decision.kind !== "requeue") throw new Error("expected a requeue decision"); + expect(decision.idempotencyKey).toBe( + buildFeedbackDeliveryRetryIdempotencyKey({ fingerprint: immediate!.fingerprint, generation: 1 }), + ); + }); + + it("skips when the immediate lane already queued the same replay generation", () => { + expect(decide({ probe: { hasExistingRetryWake: true } })).toEqual({ + kind: "skip", + reason: "feedback delivery replay wake already exists", + }); + }); + + it("skips when any recovery action already owns the issue", () => { + expect(decide({ probe: { hasActiveRecoveryAction: true } })).toEqual({ + kind: "skip", + reason: "issue already has an active recovery action", + }); + }); + + it("opens one explicit recovery action once automatic replay is exhausted", () => { + const exhausted = failedFeedbackRun({ + contextSnapshot: { + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID, + feedbackDeliveryRetryGeneration: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + wakeCommentIds: [COMMENT_ID], + }, + wakeupRequestId: "wake-retry-1", + }); + + const decision = decide({ latestRun: exhausted }); + expect(decision).toMatchObject({ kind: "recovery_action", reason: "retry_exhausted" }); + if (decision.kind !== "recovery_action") throw new Error("expected a recovery action"); + // The exhausted generation keeps the original root wake fingerprint so it + // lands on the immediate lane's action instead of creating a new one. + expect(decision.delivery.fingerprint).toBe(FINGERPRINT); + expect(decision.delivery.generation).toBe(FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES); + }); + + it("reads the exhausted generation from a scheduled retry run that lost its context hint", () => { + const decision = decide({ + latestRun: failedFeedbackRun({ + contextSnapshot: { + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID, + wakeCommentIds: [COMMENT_ID], + }, + scheduledRetryReason: "feedback_delivery_retry", + scheduledRetryAttempt: 1, + }), + }); + + expect(decision).toMatchObject({ kind: "recovery_action", reason: "retry_exhausted" }); + }); + + it("cannot loop: a replayed generation never requeues another replay", () => { + let run = failedFeedbackRun(); + const generations: number[] = []; + for (let round = 0; round < 5; round += 1) { + const decision = decide({ latestRun: run }); + if (decision.kind !== "requeue") { + expect(decision).toMatchObject({ kind: "recovery_action" }); + break; + } + generations.push(decision.generation); + // Simulate the replay run also failing without handling the feedback. + run = failedFeedbackRun({ + contextSnapshot: decision.contextSnapshot as Record, + wakeupRequestId: `wake-retry-${decision.generation}`, + }); + } + + expect(generations).toEqual([1]); + }); + + it.each([ + ["a paused assignee", { assigneeInvokable: false, assigneeHold: "paused" as const }], + ["a budget-blocked assignee", { budgetBlocked: true }], + ])("still records exhausted feedback for %s, because the action never wakes anyone", (_label, probe) => { + const decision = decide({ + probe, + latestRun: failedFeedbackRun({ + contextSnapshot: { + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + feedbackDeliveryRootWakeupRequestId: ROOT_WAKE_ID, + feedbackDeliveryRetryGeneration: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + wakeCommentIds: [COMMENT_ID], + }, + wakeupRequestId: "wake-retry-1", + }), + }); + + expect(decision).toMatchObject({ kind: "recovery_action", reason: "retry_exhausted" }); + }); + + it("routes to an explicit recovery action when the original assignee is not invokable", () => { + expect(decide({ probe: { assigneeInvokable: false, assigneeHold: null } })).toMatchObject({ + kind: "recovery_action", + reason: "assignee_not_invokable", + }); + }); + + it.each([ + ["a pending Board interaction", { hasPendingWakeInteraction: true }, "issue has a pending waking interaction"], + ["a pending approval", { hasPendingApproval: true }, "issue has a pending approval"], + ["a monitor or blocker wait", { hasPersistedDurableWaitPath: true }, "issue has a persisted durable wait path"], + ["an active run", { hasActiveExecutionPath: true }, "issue already has an active execution path"], + ["a queued wake", { hasQueuedIssueWake: true }, "issue already has a queued wake"], + ["a pause hold", { pauseHeld: true }, "automatic recovery is suppressed by a pause hold"], + ["a budget block", { budgetBlocked: true }, "assignee invocation is budget blocked"], + [ + "a paused assignee", + { assigneeInvokable: false, assigneeHold: "paused" as const }, + "assignee is held (paused)", + ], + [ + "an assignee awaiting approval", + { assigneeInvokable: false, assigneeHold: "pending_approval" as const }, + "assignee is held (pending_approval)", + ], + ])("leaves a legitimate wait untouched: %s", (_label, probe, reason) => { + expect(decide({ probe })).toEqual({ kind: "skip", reason }); + }); + + it.each([ + ["done", "done"], + ["cancelled", "cancelled"], + ["in_progress", "in_progress"], + ["blocked", "blocked"], + ])("leaves %s issues untouched", (_label, status) => { + expect(decide({ issue: { status } })).toEqual({ kind: "skip", reason: "issue is not in review" }); + }); + + it("leaves human-owned reviews untouched", () => { + expect(decide({ issue: { assigneeUserId: "user-1" } })).toEqual({ + kind: "skip", + reason: "review is owned by a user", + }); + }); + + it("leaves reviews with a typed execution participant to the participant lane", () => { + expect(decide({ hasTypedReviewParticipant: true })).toEqual({ + kind: "skip", + reason: "review issue has a typed execution participant", + }); + }); + + it("ignores a latest run that carried no feedback context", () => { + expect( + decide({ + latestRun: failedFeedbackRun({ + contextSnapshot: { issueId: ISSUE_ID, wakeReason: "issue_assigned" }, + }), + }), + ).toEqual({ kind: "skip", reason: "latest failed run carried no feedback delivery context" }); + }); + + it("ignores feedback whose root wake did not come from a user", () => { + expect(decide({ rootWakeRequestedByActorType: "agent" })).toEqual({ + kind: "skip", + reason: "feedback wake did not originate from a user", + }); + }); + + it.each([ + ["succeeded", "succeeded"], + ["running", "running"], + ["cancelled", "cancelled"], + ["interrupted", "interrupted"], + ])("ignores a latest run in status %s", (_label, status) => { + expect(decide({ latestRun: failedFeedbackRun({ status }) })).toEqual({ + kind: "skip", + reason: "latest run did not fail", + }); + }); + + it("ignores a failed feedback run owned by a different agent", () => { + expect(decide({ latestRun: failedFeedbackRun({ agentId: "agent-2" }) })).toEqual({ + kind: "skip", + reason: "latest failed run belongs to a different agent", + }); + }); + + it("ignores a review issue with no runs at all", () => { + expect(decide({ latestRun: null })).toEqual({ kind: "skip", reason: "review issue has no latest run" }); + }); + + it("accepts an explicit resume wake with no comment ids", () => { + expect( + decide({ + latestRun: failedFeedbackRun({ + contextSnapshot: { + issueId: ISSUE_ID, + wakeReason: "issue_status_changed", + resumeIntent: true, + }, + }), + }), + ).toMatchObject({ kind: "requeue", generation: 1 }); + }); +}); diff --git a/server/src/services/recovery/feedback-delivery.ts b/server/src/services/recovery/feedback-delivery.ts new file mode 100644 index 00000000000..e62c0fdd41e --- /dev/null +++ b/server/src/services/recovery/feedback-delivery.ts @@ -0,0 +1,348 @@ +import { withRecoveryModelProfileHint } from "./model-profile-hint.js"; + +export const FEEDBACK_DELIVERY_RETRY_REASON = "feedback_delivery_retry"; +export const FEEDBACK_DELIVERY_RETRY_WAKE_REASON = "feedback_delivery_retry"; +export const FEEDBACK_DELIVERY_RECOVERY_CAUSE = "feedback_delivery_exhausted"; +export const FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND = "feedback_delivery" as const; +export const FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES = 1; +export const FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY = "wakeCommentIds"; + +export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE = "issue.stranded_feedback_delivery_backstop"; +export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION = + "Human feedback on this in-review issue was never durably handled: the run created for it failed and no reviewer, " + + "interaction, approval, monitor, active run, or queued wake remained. Handle the outstanding comment(s) now and " + + "record a valid disposition. This is the only automatic backstop wake for this feedback batch."; +export const STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION = + "Handle the outstanding human feedback comment(s) on this in-review issue and record a valid disposition, " + + "or state the exact blocker and named unblock owner. Automatic feedback replay is exhausted."; + +// Feedback delivery is a human-input contract, so only wakes whose root request +// came from a user create a delivery obligation. +export const FEEDBACK_DELIVERY_ROOT_WAKE_ACTOR_TYPE = "user"; + +const FEEDBACK_DELIVERY_WAKE_REASONS = [ + "issue_commented", + "issue_reopened_via_comment", + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, +] as const; +const FEEDBACK_DELIVERY_RESUME_WAKE_REASONS = [ + "issue_assigned", + "issue_status_changed", + "issue_commented", + "issue_reopened_via_comment", +] as const; + +function readNonEmptyString(value: unknown) { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +export function buildFeedbackDeliveryFingerprint(input: { + companyId: string; + issueId: string; + agentId: string; + rootWakeupRequestId: string; +}) { + return [ + "feedback_delivery", + input.companyId, + input.issueId, + input.agentId, + input.rootWakeupRequestId, + ].join(":"); +} + +/** + * Both delivery lanes (immediate terminalization in heartbeat and the periodic + * backstop in recovery) derive the replay wake key from this function, so a + * generation can only ever be queued once no matter which lane notices first. + */ +export function buildFeedbackDeliveryRetryIdempotencyKey(input: { + fingerprint: string; + generation: number; +}) { + return `${input.fingerprint}:generation:${input.generation}`; +} + +export function readFeedbackDeliveryWakeCommentIds( + contextSnapshot: Record | null | undefined, +): string[] { + const merged: string[] = []; + const append = (value: unknown) => { + const normalized = readNonEmptyString(value); + if (!normalized || merged.includes(normalized)) return; + merged.push(normalized); + }; + + const batched = contextSnapshot?.[FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY]; + if (Array.isArray(batched)) { + for (const entry of batched) append(entry); + } + append(contextSnapshot?.wakeCommentId); + append(contextSnapshot?.commentId); + return merged; +} + +export function carriesExplicitFeedbackResume( + contextSnapshot: Record | null | undefined, +) { + return contextSnapshot?.resumeIntent === true || contextSnapshot?.followUpRequested === true; +} + +export function isEligibleFeedbackDeliveryWake(input: { + wakeReason: string | null; + carriesExplicitResume: boolean; +}) { + if (!input.wakeReason) return false; + if ((FEEDBACK_DELIVERY_WAKE_REASONS as readonly string[]).includes(input.wakeReason)) return true; + return ( + input.carriesExplicitResume && + (FEEDBACK_DELIVERY_RESUME_WAKE_REASONS as readonly string[]).includes(input.wakeReason) + ); +} + +export function readFeedbackDeliveryRetryGeneration(input: { + contextSnapshot: Record | null | undefined; + scheduledRetryReason?: string | null; + scheduledRetryAttempt?: number | null; +}) { + const raw = input.contextSnapshot?.feedbackDeliveryRetryGeneration; + const fromContext = typeof raw === "number" ? raw : null; + const fromRun = input.scheduledRetryReason === FEEDBACK_DELIVERY_RETRY_REASON + ? input.scheduledRetryAttempt ?? 0 + : 0; + return Math.max(0, Math.floor(fromContext ?? fromRun)); +} + +export type FeedbackDeliveryRunContext = { + issueId: string; + agentId: string; + rootWakeupRequestId: string; + commentIds: string[]; + generation: number; + fingerprint: string; +}; + +/** + * Pure half of feedback-delivery context extraction: everything both lanes can + * decide from the run row alone. Callers still confirm the root wake was + * user-requested, which needs a lookup. + */ +export function readFeedbackDeliveryRunContext(input: { + companyId: string; + run: { + id: string; + agentId: string; + wakeupRequestId: string | null; + contextSnapshot: unknown; + scheduledRetryReason?: string | null; + scheduledRetryAttempt?: number | null; + }; +}): FeedbackDeliveryRunContext | null { + const context = (typeof input.run.contextSnapshot === "object" && input.run.contextSnapshot !== null + ? input.run.contextSnapshot + : {}) as Record; + const issueId = readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId); + const rootWakeupRequestId = + readNonEmptyString(context.feedbackDeliveryRootWakeupRequestId) ?? input.run.wakeupRequestId; + if (!issueId || !rootWakeupRequestId) return null; + + const commentIds = readFeedbackDeliveryWakeCommentIds(context); + const carriesExplicitResume = carriesExplicitFeedbackResume(context); + if ( + !isEligibleFeedbackDeliveryWake({ + wakeReason: readNonEmptyString(context.wakeReason), + carriesExplicitResume, + }) + ) { + return null; + } + if (commentIds.length === 0 && !carriesExplicitResume) return null; + + return { + issueId, + agentId: input.run.agentId, + rootWakeupRequestId, + commentIds, + generation: readFeedbackDeliveryRetryGeneration({ + contextSnapshot: context, + scheduledRetryReason: input.run.scheduledRetryReason ?? null, + scheduledRetryAttempt: input.run.scheduledRetryAttempt ?? null, + }), + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId: input.companyId, + issueId, + agentId: input.run.agentId, + rootWakeupRequestId, + }), + }; +} + +export type StrandedFeedbackDeliveryBackstopDecision = + | { + kind: "requeue"; + delivery: FeedbackDeliveryRunContext; + generation: number; + idempotencyKey: string; + payload: Record; + contextSnapshot: Record; + } + | { + kind: "recovery_action"; + delivery: FeedbackDeliveryRunContext; + reason: "retry_exhausted" | "assignee_not_invokable"; + } + | { kind: "skip"; reason: string }; + +export type StrandedFeedbackDeliveryBackstopProbe = { + /** Any live or queued-for-execution run on the issue, regardless of agent. */ + hasActiveExecutionPath: boolean; + hasPendingWakeInteraction: boolean; + hasPendingApproval: boolean; + /** Monitor check or an unresolved `blocks` dependency. */ + hasPersistedDurableWaitPath: boolean; + hasQueuedIssueWake: boolean; + /** Any active/escalated recovery action already owns this issue. */ + hasActiveRecoveryAction: boolean; + /** A wake already exists for the replay idempotency key this lane would use. */ + hasExistingRetryWake: boolean; + assigneeInvokable: boolean; + /** + * Set when the assignee is deliberately held (operator pause, or still + * awaiting approval). That is a legitimate wait, not stranded feedback, so it + * is left alone rather than routed to a takeover owner. + */ + assigneeHold: "paused" | "pending_approval" | null; + budgetBlocked: boolean; + pauseHeld: boolean; +}; + +/** + * Narrow reconciliation backstop for review feedback that immediate + * terminalization missed: agent-assigned `in_review` with no typed review + * participant, no interaction/approval/monitor/user owner, no active run or + * queued wake, and a failed latest run that carried user feedback context. + */ +export function decideStrandedFeedbackDeliveryBackstop(input: { + companyId: string; + issue: { + id: string; + status: string; + assigneeAgentId: string | null; + assigneeUserId: string | null; + }; + hasTypedReviewParticipant: boolean; + latestRun: { + id: string; + agentId: string; + status: string; + errorCode: string | null; + wakeupRequestId: string | null; + contextSnapshot: unknown; + scheduledRetryReason?: string | null; + scheduledRetryAttempt?: number | null; + } | null; + rootWakeRequestedByActorType: string | null; + probe: StrandedFeedbackDeliveryBackstopProbe; + maxAutomaticRetries?: number; +}): StrandedFeedbackDeliveryBackstopDecision { + if (input.issue.status !== "in_review") { + return { kind: "skip", reason: "issue is not in review" }; + } + if (input.issue.assigneeUserId) { + return { kind: "skip", reason: "review is owned by a user" }; + } + if (!input.issue.assigneeAgentId) { + return { kind: "skip", reason: "review issue has no agent assignee" }; + } + if (input.hasTypedReviewParticipant) { + return { kind: "skip", reason: "review issue has a typed execution participant" }; + } + if (!input.latestRun) { + return { kind: "skip", reason: "review issue has no latest run" }; + } + if (input.latestRun.status !== "failed" && input.latestRun.status !== "timed_out") { + return { kind: "skip", reason: "latest run did not fail" }; + } + if (input.latestRun.agentId !== input.issue.assigneeAgentId) { + return { kind: "skip", reason: "latest failed run belongs to a different agent" }; + } + + const delivery = readFeedbackDeliveryRunContext({ + companyId: input.companyId, + run: input.latestRun, + }); + if (!delivery || delivery.issueId !== input.issue.id) { + return { kind: "skip", reason: "latest failed run carried no feedback delivery context" }; + } + if (input.rootWakeRequestedByActorType !== FEEDBACK_DELIVERY_ROOT_WAKE_ACTOR_TYPE) { + return { kind: "skip", reason: "feedback wake did not originate from a user" }; + } + + const probe = input.probe; + if (probe.hasActiveExecutionPath) return { kind: "skip", reason: "issue already has an active execution path" }; + if (probe.hasPendingWakeInteraction) return { kind: "skip", reason: "issue has a pending waking interaction" }; + if (probe.hasPendingApproval) return { kind: "skip", reason: "issue has a pending approval" }; + if (probe.hasPersistedDurableWaitPath) return { kind: "skip", reason: "issue has a persisted durable wait path" }; + if (probe.hasQueuedIssueWake) return { kind: "skip", reason: "issue already has a queued wake" }; + if (probe.pauseHeld) return { kind: "skip", reason: "automatic recovery is suppressed by a pause hold" }; + // The immediate lane owns this issue; never open a parallel second lane. + if (probe.hasActiveRecoveryAction) return { kind: "skip", reason: "issue already has an active recovery action" }; + + // Exhaustion is checked before the hold/budget gates: the recovery action is a + // visibility record with a `manual_repair_required` wake policy, so recording + // it never wakes a held or budget-gated assignee, and exhausted feedback must + // not stay silently `in_review`. + const maxAutomaticRetries = input.maxAutomaticRetries ?? FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES; + if (delivery.generation >= maxAutomaticRetries) { + return { kind: "recovery_action", delivery, reason: "retry_exhausted" }; + } + + // Replay would wake the assignee, so a deliberate hold or budget gate is a + // legitimate wait that keeps its own lane as the single owner. + if (probe.assigneeHold) { + return { kind: "skip", reason: `assignee is held (${probe.assigneeHold})` }; + } + if (probe.budgetBlocked) return { kind: "skip", reason: "assignee invocation is budget blocked" }; + + if (!probe.assigneeInvokable) { + return { kind: "recovery_action", delivery, reason: "assignee_not_invokable" }; + } + + const generation = delivery.generation + 1; + const idempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({ + fingerprint: delivery.fingerprint, + generation, + }); + if (probe.hasExistingRetryWake) { + return { kind: "skip", reason: "feedback delivery replay wake already exists" }; + } + + const latestCommentId = delivery.commentIds.at(-1); + const payload = withRecoveryModelProfileHint({ + issueId: input.issue.id, + taskId: input.issue.id, + sourceIssueId: input.issue.id, + retryOfRunId: input.latestRun.id, + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + feedbackDeliveryRootWakeupRequestId: delivery.rootWakeupRequestId, + feedbackDeliverySourceRunId: input.latestRun.id, + feedbackDeliveryRetryGeneration: generation, + feedbackDeliveryBackstop: true, + [FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY]: delivery.commentIds, + ...(latestCommentId ? { commentId: latestCommentId, wakeCommentId: latestCommentId } : {}), + }, "normal_model"); + + return { + kind: "requeue", + delivery, + generation, + idempotencyKey, + payload, + contextSnapshot: withRecoveryModelProfileHint({ + ...payload, + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + source: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + instruction: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION, + }, "normal_model"), + }; +} diff --git a/server/src/services/recovery/index.ts b/server/src/services/recovery/index.ts index 6d35159ed37..b989e3aba85 100644 --- a/server/src/services/recovery/index.ts +++ b/server/src/services/recovery/index.ts @@ -28,6 +28,29 @@ export type { IssueLivenessSeverity, IssueLivenessState, } from "./issue-graph-liveness.js"; +export { + FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + FEEDBACK_DELIVERY_RECOVERY_CAUSE, + FEEDBACK_DELIVERY_RETRY_REASON, + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + STRANDED_FEEDBACK_DELIVERY_BACKSTOP_INSTRUCTION, + STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION, + STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + buildFeedbackDeliveryFingerprint, + buildFeedbackDeliveryRetryIdempotencyKey, + carriesExplicitFeedbackResume, + decideStrandedFeedbackDeliveryBackstop, + isEligibleFeedbackDeliveryWake, + readFeedbackDeliveryRetryGeneration, + readFeedbackDeliveryRunContext, + readFeedbackDeliveryWakeCommentIds, +} from "./feedback-delivery.js"; +export type { + FeedbackDeliveryRunContext, + StrandedFeedbackDeliveryBackstopDecision, + StrandedFeedbackDeliveryBackstopProbe, +} from "./feedback-delivery.js"; export { recoveryService, } from "./service.js"; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 9cfe3921970..81fa3917b4b 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -51,7 +51,7 @@ import { buildIssueBlockersResolvedWakeIdempotencyKey, findExistingIssueBlockersResolvedWakeForAnyKey, } from "../issue-dependency-wakeups.js"; -import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; +import { evaluateAgentInvokabilityFromDb, type AgentInvokability } from "../agent-invokability.js"; import { getRunLogStore } from "../run-log-store.js"; import { DEFAULT_MAX_SUCCESSFUL_RUN_HANDOFF_ATTEMPTS, @@ -82,6 +82,18 @@ import { withRecoveryModelProfileHint, } from "./model-profile-hint.js"; import { isAutomaticRecoverySuppressedByPauseHold } from "./pause-hold-guard.js"; +import { + FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + FEEDBACK_DELIVERY_RECOVERY_CAUSE, + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION, + STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + buildFeedbackDeliveryRetryIdempotencyKey, + decideStrandedFeedbackDeliveryBackstop, + readFeedbackDeliveryRunContext, + type StrandedFeedbackDeliveryBackstopDecision, +} from "./feedback-delivery.js"; const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["interrupted", "failed", "cancelled", "timed_out"] as const; @@ -3638,6 +3650,366 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) readNonEmptyString(monitor.externalRef) === latestRun.id; } + async function getLatestFeedbackDeliveryRun(companyId: string, issueId: string) { + return db + .select({ + id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + error: heartbeatRuns.error, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + contextSnapshot: heartbeatRuns.contextSnapshot, + livenessState: heartbeatRuns.livenessState, + resultJson: heartbeatRuns.resultJson, + startedAt: heartbeatRuns.startedAt, + createdAt: heartbeatRuns.createdAt, + scheduledRetryReason: heartbeatRuns.scheduledRetryReason, + scheduledRetryAttempt: heartbeatRuns.scheduledRetryAttempt, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + ), + ) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function evaluateIssueAssigneeInvokability( + issue: typeof issues.$inferSelect, + ): Promise { + if (!issue.assigneeAgentId) { + return { invokable: false, reason: "missing", message: "Issue has no agent assignee", details: {}, invalidOrgChain: false }; + } + const agent = await getAgent(issue.assigneeAgentId); + if (!agent || agent.companyId !== issue.companyId) { + return { invokable: false, reason: "missing", message: "Agent no longer exists", details: {}, invalidOrgChain: false }; + } + return evaluateAgentInvokabilityFromDb(db, agent); + } + + async function hasPendingIssueApproval(companyId: string, issueId: string) { + return db + .select({ id: issueApprovals.approvalId }) + .from(issueApprovals) + .innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id)) + .where( + and( + eq(issueApprovals.companyId, companyId), + eq(issueApprovals.issueId, issueId), + inArray(approvals.status, ["pending", "revision_requested"]), + ), + ) + .limit(1) + .then((rows) => Boolean(rows[0])); + } + + async function readRootFeedbackWakeActorType(input: { + companyId: string; + agentId: string; + rootWakeupRequestId: string; + }) { + return db + .select({ requestedByActorType: agentWakeupRequests.requestedByActorType }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.id, input.rootWakeupRequestId), + eq(agentWakeupRequests.companyId, input.companyId), + eq(agentWakeupRequests.agentId, input.agentId), + ), + ) + .limit(1) + .then((rows) => rows[0]?.requestedByActorType ?? null); + } + + async function hasFeedbackDeliveryWakeForIdempotencyKey(companyId: string, idempotencyKey: string) { + return db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.idempotencyKey, idempotencyKey), + ), + ) + .limit(1) + .then((rows) => Boolean(rows[0])); + } + + /** + * The immediate lane logs `issue.feedback_retry_queued` keyed on the root wake + * id. Re-using the same dedupe check keeps the audit trail single-entry per + * feedback batch no matter which lane queued the replay. + */ + async function hasFeedbackDeliveryActivity(input: { + companyId: string; + issueId: string; + action: string; + rootWakeupRequestId: string; + }) { + return db + .select({ id: activityLog.id }) + .from(activityLog) + .where( + and( + eq(activityLog.companyId, input.companyId), + eq(activityLog.entityType, "issue"), + eq(activityLog.entityId, input.issueId), + eq(activityLog.action, input.action), + sql`${activityLog.details} ->> 'rootWakeupRequestId' = ${input.rootWakeupRequestId}`, + ), + ) + .limit(1) + .then((rows) => Boolean(rows[0])); + } + + async function ensureStrandedFeedbackDeliveryRecoveryAction(input: { + issue: typeof issues.$inferSelect; + latestRun: Awaited>; + decision: Extract; + }) { + const delivery = input.decision.delivery; + const ownerAgentId = input.decision.reason === "assignee_not_invokable" + ? await resolveStrandedIssueRecoveryOwnerAgentId(input.issue) + : await resolveInvokableRecoveryAgentId(input.issue, delivery.agentId) + ?? await resolveStrandedIssueRecoveryOwnerAgentId(input.issue); + const ownerName = ownerAgentId + ? (await getAgent(ownerAgentId))?.name ?? null + : null; + const routingFallbackReason = ownerAgentId === delivery.agentId + ? null + : "The original assignee is not invokable; stranded feedback recovery fell through to the manager ladder."; + + const action = await recoveryActionsSvc.upsertSourceScoped({ + companyId: input.issue.companyId, + sourceIssueId: input.issue.id, + kind: FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + ownerType: ownerAgentId ? "agent" : "board", + ownerAgentId, + previousOwnerAgentId: delivery.agentId, + returnOwnerAgentId: delivery.agentId, + cause: FEEDBACK_DELIVERY_RECOVERY_CAUSE, + fingerprint: delivery.fingerprint, + evidence: { + sourceIssueId: input.issue.id, + sourceIdentifier: input.issue.identifier, + previousStatus: "in_review", + detectedBy: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + backstopReason: input.decision.reason, + sourceRunId: input.latestRun?.id ?? null, + latestRunStatus: input.latestRun?.status ?? null, + lastFailureCode: input.latestRun?.errorCode ?? null, + rootWakeupRequestId: delivery.rootWakeupRequestId, + outstandingCommentIds: delivery.commentIds, + retryGeneration: delivery.generation, + maxAutomaticRetries: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + failureSummary: summarizeRunFailureForIssueComment(input.latestRun ?? null)?.trim() ?? null, + routingFallbackReason, + }, + nextAction: [ + ownerName ? `Owner: ${ownerName}.` : "Owner: the Board (no invokable recovery agent).", + STRANDED_FEEDBACK_DELIVERY_BACKSTOP_NEXT_ACTION, + ].join(" "), + wakePolicy: { + type: "manual_repair_required", + reason: FEEDBACK_DELIVERY_RECOVERY_CAUSE, + ownerAgentId, + }, + maxAttempts: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + lastAttemptAt: new Date(), + }); + + if ( + !(await hasFeedbackDeliveryActivity({ + companyId: input.issue.companyId, + issueId: input.issue.id, + action: "issue.feedback_recovery_exhausted", + rootWakeupRequestId: delivery.rootWakeupRequestId, + })) + ) { + await logActivity(db, { + companyId: input.issue.companyId, + actorType: "system", + actorId: "recovery", + agentId: delivery.agentId, + action: "issue.feedback_recovery_exhausted", + entityType: "issue", + entityId: input.issue.id, + details: { + recoveryActionId: action.id, + detectedBy: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + backstopReason: input.decision.reason, + sourceRunId: input.latestRun?.id ?? null, + rootWakeupRequestId: delivery.rootWakeupRequestId, + sourceCommentIds: delivery.commentIds, + retryGeneration: delivery.generation, + ownerAgentId, + }, + }); + } + + return action; + } + + /** + * PAP-17302: periodic backstop for the narrow review-feedback shape immediate + * terminalization can miss — agent-assigned `in_review` with no typed review + * participant, no interaction/approval/monitor/user owner, no active run or + * queued wake, and a failed latest run that carried user feedback context. + * + * Dedupes against the immediate retry/recovery lane by sharing its wake + * idempotency key, recovery-action fingerprint, and activity keys, so this can + * never open a parallel second delivery lane. + */ + async function reconcileStrandedFeedbackDelivery(input: { + issue: typeof issues.$inferSelect; + hasTypedReviewParticipant: boolean; + }): Promise<{ outcome: "requeued" | "recovery_action" | "skipped"; reason?: string }> { + const latestRun = await getLatestFeedbackDeliveryRun(input.issue.companyId, input.issue.id); + const candidate = latestRun + ? readFeedbackDeliveryRunContext({ companyId: input.issue.companyId, run: latestRun }) + : null; + // Cheap pure shape gate first: most `in_review` issues without a typed + // participant are legitimate waits, and they must not pay for the probes. + if (!latestRun || !candidate || candidate.issueId !== input.issue.id) { + return { outcome: "skipped", reason: "not_stranded_feedback_shape" }; + } + + const retryIdempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({ + fingerprint: candidate.fingerprint, + generation: candidate.generation + 1, + }); + const [ + rootWakeRequestedByActorType, + activeExecutionPath, + pendingWakeInteraction, + pendingApproval, + durableWaitPath, + queuedIssueWake, + activeRecoveryAction, + existingRetryWake, + assigneeInvokability, + budgetBlocked, + ] = await Promise.all([ + readRootFeedbackWakeActorType({ + companyId: input.issue.companyId, + agentId: candidate.agentId, + rootWakeupRequestId: candidate.rootWakeupRequestId, + }), + hasActiveExecutionPath(input.issue.companyId, input.issue.id), + hasPendingWakeInteraction(input.issue.companyId, input.issue.id), + hasPendingIssueApproval(input.issue.companyId, input.issue.id), + hasPersistedDurableWaitPath(input.issue), + hasQueuedIssueWake(input.issue.companyId, input.issue.id), + recoveryActionsSvc.getActiveForIssue(input.issue.companyId, input.issue.id), + hasFeedbackDeliveryWakeForIdempotencyKey(input.issue.companyId, retryIdempotencyKey), + evaluateIssueAssigneeInvokability(input.issue), + input.issue.assigneeAgentId + ? isInvocationBudgetBlocked(input.issue, input.issue.assigneeAgentId) + : Promise.resolve(false), + ]); + + const decision = decideStrandedFeedbackDeliveryBackstop({ + companyId: input.issue.companyId, + issue: { + id: input.issue.id, + status: input.issue.status, + assigneeAgentId: input.issue.assigneeAgentId, + assigneeUserId: input.issue.assigneeUserId, + }, + hasTypedReviewParticipant: input.hasTypedReviewParticipant, + latestRun, + rootWakeRequestedByActorType, + probe: { + hasActiveExecutionPath: activeExecutionPath, + hasPendingWakeInteraction: pendingWakeInteraction, + hasPendingApproval: pendingApproval, + hasPersistedDurableWaitPath: durableWaitPath, + hasQueuedIssueWake: queuedIssueWake, + hasActiveRecoveryAction: Boolean(activeRecoveryAction), + hasExistingRetryWake: existingRetryWake, + assigneeInvokable: assigneeInvokability.invokable, + assigneeHold: !assigneeInvokability.invokable && + (assigneeInvokability.reason === "paused" || assigneeInvokability.reason === "pending_approval") + ? assigneeInvokability.reason + : null, + budgetBlocked, + // The caller gates the whole scan on pause holds before reaching here. + pauseHeld: false, + }, + }); + + if (decision.kind === "skip") { + return { outcome: "skipped", reason: decision.reason }; + } + + if (decision.kind === "recovery_action") { + await ensureStrandedFeedbackDeliveryRecoveryAction({ + issue: input.issue, + latestRun, + decision, + }); + return { outcome: "recovery_action", reason: decision.reason }; + } + + const queued = await deps.enqueueWakeup(decision.delivery.agentId, { + source: "automation", + triggerDetail: "system", + reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + idempotencyKey: decision.idempotencyKey, + payload: decision.payload, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: decision.contextSnapshot, + }); + if (!queued) { + return { outcome: "skipped", reason: "replay_wake_not_queued" }; + } + + await db + .update(heartbeatRuns) + .set({ retryOfRunId: latestRun.id, updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, queued.id)); + + if ( + !(await hasFeedbackDeliveryActivity({ + companyId: input.issue.companyId, + issueId: input.issue.id, + action: "issue.feedback_retry_queued", + rootWakeupRequestId: decision.delivery.rootWakeupRequestId, + })) + ) { + await logActivity(db, { + companyId: input.issue.companyId, + actorType: "system", + actorId: "recovery", + agentId: decision.delivery.agentId, + runId: queued.id, + action: "issue.feedback_retry_queued", + entityType: "issue", + entityId: input.issue.id, + details: { + detectedBy: STRANDED_FEEDBACK_DELIVERY_BACKSTOP_SOURCE, + sourceRunId: latestRun.id, + sourceWakeupRequestId: latestRun.wakeupRequestId, + rootWakeupRequestId: decision.delivery.rootWakeupRequestId, + sourceCommentIds: decision.delivery.commentIds, + retryRunId: queued.id, + retryWakeupRequestId: queued.wakeupRequestId, + retryGeneration: decision.generation, + disposition: "queued", + }, + }); + } + + return { outcome: "requeued" }; + } + async function reconcileStrandedAssignedIssues(opts?: { issueCreatedAtGte?: Date | null }) { const candidates = await db .select() @@ -3663,6 +4035,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) orphanBlockersAssigned: 0, successfulRunHandoffEscalated: 0, reviewParticipantRequeued: 0, + strandedFeedbackRequeued: 0, + strandedFeedbackRecoveryActions: 0, escalated: 0, waitingOnReviewResolved: 0, providerQuotaMonitored: 0, @@ -3893,7 +4267,22 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) if (issue.status === "in_review") { if (!participantAgentId || !pendingExecutionState) { - result.skipped += 1; + // PAP-17302: an `in_review` issue with no typed participant used to be + // skipped unconditionally, which stranded review feedback whose only + // path was a failed comment run. Everything else stays skipped. + const feedbackBackstop = await reconcileStrandedFeedbackDelivery({ + issue, + hasTypedReviewParticipant: Boolean(participantAgentId && pendingExecutionState), + }); + if (feedbackBackstop.outcome === "requeued") { + result.strandedFeedbackRequeued += 1; + result.issueIds.push(issue.id); + } else if (feedbackBackstop.outcome === "recovery_action") { + result.strandedFeedbackRecoveryActions += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } continue; } const participantLatestRun = participantLatestRunForRecovery;