From a6c28fa41430088729908a082f65a03e65a1890f Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 16 Aug 2026 19:00:24 +0000 Subject: [PATCH] fix(server): recover lost human feedback delivery Replay feedback lost before process launch, reconcile stranded review feedback, and project durable delivery state onto source comments. Co-Authored-By: Paperclip --- packages/shared/src/constants.ts | 18 + packages/shared/src/index.ts | 4 + packages/shared/src/types/index.ts | 2 + packages/shared/src/types/issue.ts | 55 ++ .../heartbeat-process-recovery.test.ts | 870 ++++++++++++++++++ server/src/index.ts | 4 + server/src/routes/issues.ts | 165 +++- .../feedback-delivery-comment-states.ts | 228 +++++ server/src/services/heartbeat.ts | 786 +++++++++++++++- server/src/services/index.ts | 1 + .../feedback-delivery-projection.test.ts | 293 ++++++ .../recovery/feedback-delivery-projection.ts | 258 ++++++ .../recovery/feedback-delivery.test.ts | 333 +++++++ .../services/recovery/feedback-delivery.ts | 362 ++++++++ server/src/services/recovery/index.ts | 23 + server/src/services/recovery/service.ts | 393 +++++++- 16 files changed, 3762 insertions(+), 33 deletions(-) create mode 100644 server/src/services/feedback-delivery-comment-states.ts create mode 100644 server/src/services/recovery/feedback-delivery-projection.test.ts create mode 100644 server/src/services/recovery/feedback-delivery-projection.ts create mode 100644 server/src/services/recovery/feedback-delivery.test.ts create mode 100644 server/src/services/recovery/feedback-delivery.ts diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index d4660812352..36ef970e6f3 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -376,6 +376,7 @@ export const ISSUE_SURFACE_VISIBILITIES = ["default", "plugin_operation"] as con export type IssueSurfaceVisibility = (typeof ISSUE_SURFACE_VISIBILITIES)[number]; export const ISSUE_RECOVERY_ACTION_KINDS = [ + "feedback_delivery", "missing_disposition", "stranded_assigned_issue", "workspace_validation", @@ -401,6 +402,23 @@ export const ISSUE_RECOVERY_ACTION_OWNER_TYPES = [ ] as const; export type IssueRecoveryActionOwnerType = (typeof ISSUE_RECOVERY_ACTION_OWNER_TYPES)[number]; +/** + * Delivery state of one human feedback comment, as projected onto that comment + * for operator surfaces. This is deliberately *not* a task status: the task + * keeps its own workflow status while a delivery attempt is in flight. + * + * - `retrying` — automatic replay is still in flight; no operator action. + * - `exhausted_needs_attention` — automatic replay gave up; a human must act. + * - `recovered_after_attention` — delivery succeeded after it had needed + * attention, so the resolution stays visible in thread history. + */ +export const ISSUE_FEEDBACK_DELIVERY_STATES = [ + "retrying", + "exhausted_needs_attention", + "recovered_after_attention", +] as const; +export type IssueFeedbackDeliveryState = (typeof ISSUE_FEEDBACK_DELIVERY_STATES)[number]; + export const ISSUE_RECOVERY_ACTION_OUTCOMES = [ "restored", "handed_back", diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index b98c0db9de9..6d8bdae6760 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -299,6 +299,7 @@ export { ONBOARDING_FIRST_TASK_ORIGIN_KIND, ISSUE_WATCHDOG_DISCOVERY_KINDS, ISSUE_SURFACE_VISIBILITIES, + ISSUE_FEEDBACK_DELIVERY_STATES, ISSUE_RECOVERY_ACTION_KINDS, ISSUE_RECOVERY_ACTION_STATUSES, ISSUE_RECOVERY_ACTION_OWNER_TYPES, @@ -492,6 +493,7 @@ export { type IssueOriginKind, type IssueWatchdogDiscoveryKind, type IssueSurfaceVisibility, + type IssueFeedbackDeliveryState, type IssueRecoveryActionKind, type IssueRecoveryActionStatus, type IssueRecoveryActionOwnerType, @@ -1008,6 +1010,8 @@ export type { IssueExecutionStagePrincipal, IssueExecutionDecision, IssueComment, + IssueCommentFeedbackDelivery, + IssueFeedbackDeliveryRetryResponse, IssueCommentDerivedAuthorSource, IssueCommentMetadata, IssueCommentMetadataSection, diff --git a/packages/shared/src/types/index.ts b/packages/shared/src/types/index.ts index 1f4971b940d..77149f2e85b 100644 --- a/packages/shared/src/types/index.ts +++ b/packages/shared/src/types/index.ts @@ -649,6 +649,8 @@ export type { IssueReviewRequest, IssueExecutionDecision, IssueComment, + IssueCommentFeedbackDelivery, + IssueFeedbackDeliveryRetryResponse, IssueCommentDerivedAuthorSource, IssueCommentMetadata, IssueCommentMetadataSection, diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index 0dc0bc030ac..485f217e246 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -14,6 +14,7 @@ import type { IssueReferenceSourceKind, IssueExecutionStageType, IssueExecutionStateStatus, + IssueFeedbackDeliveryState, IssueHarnessKind, IssueOriginKind, IssuePriority, @@ -941,6 +942,54 @@ export type IssueCommentDerivedAuthorSource = | "run_id" | "run_log_comment_post"; +/** + * Delivery state projected onto a single human feedback comment. + * + * Every field here is safe to render to any operator who can already read the + * comment: raw adapter errors, stack traces, request bodies, credentials, and + * filesystem paths are mapped to a short `safeFailureReason` code server-side + * and never passed through verbatim. + */ +export interface IssueCommentFeedbackDelivery { + state: IssueFeedbackDeliveryState; + /** The human comment this delivery obligation belongs to. */ + sourceCommentId: string; + /** Completed automatic replay generations, or `null` when unknown. */ + attempt: number | null; + /** Automatic replay budget, or `null` when unknown. */ + maxAttempts: number | null; + nextAttemptAt: string | null; + lastAttemptAt: string | null; + deliveredAt: string | null; + /** + * Run the operator may open for context. Omitted (`null`) when no run exists + * or the viewer is not authorized to read it — never rendered as a disabled + * mystery action. + */ + targetRunId: string | null; + /** True when the viewer may ask for one more explicit delivery attempt. */ + canRetry: boolean; + /** + * True when a replay is already queued for this obligation. Keeps the + * "Retrying…" affordance durable across reloads instead of living only in + * button-local component state. + */ + retryInFlight: boolean; + /** Short operator-safe reason code, never a raw adapter error. */ + safeFailureReason: string | null; + /** + * How many human comments share this one delivery obligation (a batched wake + * can cover several). Used for the thread rollup copy. + */ + batchedCommentCount: number; +} + +export interface IssueFeedbackDeliveryRetryResponse { + outcome: "queued" | "already_queued" | "not_exhausted" | "no_invokable_assignee"; + message: string; + feedbackDelivery: IssueCommentFeedbackDelivery | null; +} + export interface IssueComment { id: string; companyId: string; @@ -964,6 +1013,12 @@ export interface IssueComment { deletedByRunId?: string | null; sourceTrust?: SourceTrustMetadata | null; followUpRequested?: boolean; + /** + * Present only while this human comment's delivery is retrying, exhausted, or + * freshly recovered. Absent for the overwhelming majority of comments, which + * were delivered on the first attempt. + */ + feedbackDelivery?: IssueCommentFeedbackDelivery | null; createdAt: Date; updatedAt: Date; } diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 7ada8c774b1..75eea5673b7 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -100,12 +100,18 @@ vi.mock("../adapters/index.ts", async () => { }); import { + FEEDBACK_DELIVERY_RETRY_REASON, + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, INTERACTION_CONTINUATION_INFRA_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_WAKE_REASON, heartbeatService, redactDetectedSuccessfulRunProgressSummaryForBoard, redactSuccessfulRunHandoffEvidence, } from "../services/heartbeat.ts"; +import { + buildFeedbackDeliveryFingerprint, + buildFeedbackDeliveryRetryIdempotencyKey, +} from "../services/recovery/feedback-delivery.ts"; import { readHotRestartIntent, resolveLegacyHotRestartIntentPath, @@ -569,6 +575,69 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { return { companyId, agentId, runId, wakeupRequestId, issueId }; } + async function prepareFeedbackWake( + fixture: Awaited>, + input?: { + commentId?: string; + retryGeneration?: number; + rootWakeupRequestId?: string; + issueStatus?: "in_progress" | "in_review" | "done"; + }, + ) { + const commentId = input?.commentId ?? randomUUID(); + const rootWakeupRequestId = input?.rootWakeupRequestId ?? fixture.wakeupRequestId; + const retryGeneration = input?.retryGeneration ?? 0; + const createdAt = new Date("2026-03-19T00:00:01.000Z"); + + await db.insert(issueComments).values({ + id: commentId, + companyId: fixture.companyId, + issueId: fixture.issueId, + authorType: "user", + authorUserId: "responsible-user", + body: "Please incorporate this review feedback.", + createdAt, + updatedAt: createdAt, + }); + await db + .update(agentWakeupRequests) + .set({ + source: "automation", + triggerDetail: "user", + reason: retryGeneration > 0 ? FEEDBACK_DELIVERY_RETRY_WAKE_REASON : "issue_commented", + payload: { issueId: fixture.issueId, commentId }, + requestedByActorType: "user", + requestedByActorId: "responsible-user", + }) + .where(eq(agentWakeupRequests.id, rootWakeupRequestId)); + await db + .update(heartbeatRuns) + .set({ + contextSnapshot: { + issueId: fixture.issueId, + taskId: fixture.issueId, + wakeReason: retryGeneration > 0 ? FEEDBACK_DELIVERY_RETRY_WAKE_REASON : "issue_commented", + commentId, + wakeCommentId: commentId, + wakeCommentIds: [commentId], + ...(retryGeneration > 0 + ? { + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId, + feedbackDeliveryRetryGeneration: retryGeneration, + } + : {}), + }, + }) + .where(eq(heartbeatRuns.id, fixture.runId)); + await db + .update(issues) + .set({ status: input?.issueStatus ?? "in_progress" }) + .where(eq(issues.id, fixture.issueId)); + + return { commentId, rootWakeupRequestId }; + } + async function seedEnvironmentLeaseFixture(input: { companyId: string; runId: string; @@ -911,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"; }) { @@ -1368,6 +1626,337 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(checkoutReleasedIssue?.checkoutRunId).toBeNull(); }); + it("replays a user feedback wake once when restart loses it before process identity exists", async () => { + let releaseAdapter: (() => void) | null = null; + const adapterStarted = new Promise((resolve) => { + mockAdapterExecute.mockImplementationOnce(async () => { + resolve(); + await new Promise((release) => { + releaseAdapter = release; + }); + return { + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Handled the recovered feedback.", + provider: "test", + model: "test-model", + }; + }); + }); + const fixture = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + }); + const { commentId } = await prepareFeedbackWake(fixture); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] }); + await Promise.race([ + adapterStarted, + new Promise((_, reject) => setTimeout(() => reject(new Error("feedback retry did not start")), 3_000)), + ]); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, fixture.agentId)); + expect(runs).toHaveLength(2); + const retryRun = runs.find((run) => run.id !== fixture.runId); + expect(retryRun).toMatchObject({ + retryOfRunId: fixture.runId, + scheduledRetryAttempt: 1, + scheduledRetryReason: FEEDBACK_DELIVERY_RETRY_REASON, + }); + expect(retryRun?.contextSnapshot).toMatchObject({ + issueId: fixture.issueId, + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + feedbackDeliveryRootWakeupRequestId: fixture.wakeupRequestId, + feedbackDeliverySourceRunId: fixture.runId, + feedbackDeliveryRetryGeneration: 1, + wakeCommentIds: [commentId], + }); + + const retryWake = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.runId, retryRun?.id ?? "")) + .then((rows) => rows[0] ?? null); + expect(retryWake).toMatchObject({ + reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + requestedByActorType: "system", + }); + const auditRows = await db + .select() + .from(activityLog) + .where(and( + eq(activityLog.runId, fixture.runId), + eq(activityLog.action, "issue.feedback_retry_queued"), + )); + expect(auditRows).toHaveLength(1); + expect(auditRows[0]?.details).toMatchObject({ + rootWakeupRequestId: fixture.wakeupRequestId, + sourceCommentIds: [commentId], + retryRunId: retryRun?.id, + retryGeneration: 1, + }); + + if (!retryRun || !releaseAdapter) throw new Error("feedback retry did not reach the adapter"); + await db.insert(issueComments).values({ + companyId: fixture.companyId, + issueId: fixture.issueId, + authorType: "agent", + authorAgentId: fixture.agentId, + createdByRunId: retryRun.id, + body: "Applied the requested feedback.", + }); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, fixture.issueId)); + releaseAdapter(); + expect((await waitForRunToSettle(heartbeat, retryRun.id, 5_000))?.status).toBe("succeeded"); + await heartbeat.waitForRunExecutionDrain(retryRun.id); + + const finalRuns = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, fixture.agentId)); + expect(finalRuns).toHaveLength(2); + }); + + it("creates one source-scoped recovery action when the feedback replay is exhausted", async () => { + const fixture = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + }); + const { commentId } = await prepareFeedbackWake(fixture, { retryGeneration: 1 }); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] }); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, fixture.agentId)); + expect(runs).toHaveLength(1); + const recoveryRows = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId)); + expect(recoveryRows).toHaveLength(1); + expect(recoveryRows[0]).toMatchObject({ + kind: "feedback_delivery", + status: "active", + ownerAgentId: fixture.agentId, + cause: "feedback_delivery_exhausted", + maxAttempts: 1, + }); + expect(recoveryRows[0]?.evidence).toMatchObject({ + rootWakeupRequestId: fixture.wakeupRequestId, + outstandingCommentIds: [commentId], + retryGeneration: 1, + }); + const recoveryAudits = await db + .select() + .from(activityLog) + .where(and( + eq(activityLog.runId, fixture.runId), + eq(activityLog.action, "issue.feedback_recovery_exhausted"), + )); + expect(recoveryAudits).toHaveLength(1); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] }); + expect(await db + .select({ id: issueRecoveryActions.id }) + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toHaveLength(1); + }); + + it("preserves terminal issue gates instead of replaying stale feedback", async () => { + const fixture = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + }); + await prepareFeedbackWake(fixture, { issueStatus: "done" }); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] }); + + expect(await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, fixture.agentId))).toHaveLength(1); + expect(await db + .select({ id: issueRecoveryActions.id }) + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toHaveLength(0); + const issue = await db + .select({ status: issues.status, executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, fixture.issueId)) + .then((rows) => rows[0] ?? null); + expect(issue).toEqual({ status: "done", executionRunId: null }); + }); + + it("preserves pause gates and records recovery instead of invoking the agent", async () => { + const fixture = await seedRunFixture({ + agentStatus: "paused", + processPid: null, + processGroupId: null, + }); + await prepareFeedbackWake(fixture); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] }); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + expect(await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, fixture.agentId))).toHaveLength(1); + const recovery = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId)) + .then((rows) => rows[0] ?? null); + expect(recovery).toMatchObject({ + kind: "feedback_delivery", + cause: "feedback_delivery_exhausted", + evidence: { + gateErrorCode: "agent_not_invokable", + }, + }); + }); + + it("coalesces a recovered feedback wake ahead of a newer queued comment without another handler", async () => { + let releaseAdapter: (() => void) | null = null; + const adapterStarted = new Promise((resolve) => { + mockAdapterExecute.mockImplementationOnce(async () => { + resolve(); + await new Promise((release) => { + releaseAdapter = release; + }); + return { + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Handled the coalesced feedback.", + provider: "test", + model: "test-model", + }; + }); + }); + const fixture = await seedRunFixture({ + agentStatus: "idle", + processPid: null, + processGroupId: null, + }); + const first = await prepareFeedbackWake(fixture); + const secondCommentId = randomUUID(); + const secondWakeId = randomUUID(); + const secondRunId = randomUUID(); + const secondCreatedAt = new Date("2026-03-19T00:00:02.000Z"); + await db.insert(issueComments).values({ + id: secondCommentId, + companyId: fixture.companyId, + issueId: fixture.issueId, + authorType: "user", + authorUserId: "responsible-user", + body: "One more adjustment after the first review.", + createdAt: secondCreatedAt, + updatedAt: secondCreatedAt, + }); + await db.insert(agentWakeupRequests).values({ + id: secondWakeId, + companyId: fixture.companyId, + agentId: fixture.agentId, + source: "automation", + triggerDetail: "user", + reason: "issue_commented", + payload: { issueId: fixture.issueId, commentId: secondCommentId }, + status: "queued", + runId: secondRunId, + requestedByActorType: "user", + requestedByActorId: "responsible-user", + requestedAt: secondCreatedAt, + updatedAt: secondCreatedAt, + }); + await db.insert(heartbeatRuns).values({ + id: secondRunId, + companyId: fixture.companyId, + agentId: fixture.agentId, + invocationSource: "automation", + triggerDetail: "user", + status: "queued", + wakeupRequestId: secondWakeId, + contextSnapshot: { + issueId: fixture.issueId, + taskId: fixture.issueId, + wakeReason: "issue_commented", + commentId: secondCommentId, + wakeCommentId: secondCommentId, + wakeCommentIds: [secondCommentId], + }, + createdAt: secondCreatedAt, + updatedAt: secondCreatedAt, + }); + await db + .update(issues) + .set({ executionRunId: secondRunId, updatedAt: secondCreatedAt }) + .where(eq(issues.id, fixture.issueId)); + const heartbeat = heartbeatService(db); + + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 1, runIds: [fixture.runId] }); + await Promise.race([ + adapterStarted, + new Promise((_, reject) => setTimeout(() => reject(new Error("coalesced retry did not start")), 3_000)), + ]); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, fixture.agentId)); + expect(runs).toHaveLength(2); + const coalescedRun = runs.find((run) => run.id === secondRunId); + expect(coalescedRun?.contextSnapshot).toMatchObject({ + feedbackDeliveryRootWakeupRequestId: fixture.wakeupRequestId, + feedbackDeliveryRetryGeneration: 1, + wakeCommentIds: [first.commentId, secondCommentId], + commentId: secondCommentId, + wakeCommentId: secondCommentId, + }); + const auditRows = await db + .select() + .from(activityLog) + .where(and( + eq(activityLog.runId, fixture.runId), + eq(activityLog.action, "issue.feedback_retry_queued"), + )); + expect(auditRows).toHaveLength(1); + expect(auditRows[0]?.details).toMatchObject({ + retryRunId: secondRunId, + disposition: "coalesced", + sourceCommentIds: [first.commentId], + }); + + if (!releaseAdapter) throw new Error("coalesced feedback run did not reach the adapter"); + await db.insert(issueComments).values({ + companyId: fixture.companyId, + issueId: fixture.issueId, + authorType: "agent", + authorAgentId: fixture.agentId, + createdByRunId: secondRunId, + body: "Applied both feedback comments in order.", + }); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, fixture.issueId)); + releaseAdapter(); + expect((await waitForRunToSettle(heartbeat, secondRunId, 5_000))?.status).toBe("succeeded"); + await heartbeat.waitForRunExecutionDrain(secondRunId); + }); + it("restores one lost monitor dispatch before escalating a second process loss", async () => { const { companyId, agentId, runId, issueId } = await seedRunFixture({ adapterType: "openclaw_gateway", @@ -5345,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/routes/issues.ts b/server/src/routes/issues.ts index cff29679612..f02ec664dc5 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -88,6 +88,7 @@ import { type IssueWakeDiagnosticWakeFailureClass, type IssueWakeDiagnosticWakeRequest, type IssueWakeDiagnosticsResponse, + type IssueFeedbackDeliveryRetryResponse, type IssueRelationIssueSummary, type IssueReviewPolicy, type IssueThreadInteractionCanonicalResolverPolicy, @@ -104,6 +105,14 @@ import { type IssueWriteDenialContext, } from "@paperclipai/shared"; import { trackAgentTaskCompleted } from "@paperclipai/shared/telemetry"; +import { + buildFeedbackDeliveryManualRetryIdempotencyKey, + FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + FEEDBACK_DELIVERY_RECOVERY_CAUSE, + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY, +} from "../services/recovery/feedback-delivery.js"; import { getTelemetryClient } from "../telemetry.js"; import { isUniqueViolation } from "../db-errors.js"; import type { StorageService } from "../storage/types.js"; @@ -116,6 +125,7 @@ import { companyService, companySearchService, executionWorkspaceService, + feedbackDeliveryCommentStateService, goalService, heartbeatService, issueApprovalService, @@ -2756,6 +2766,7 @@ export function issueRoutes( const goalsSvc = goalService(db); const issueApprovalsSvc = issueApprovalService(db); const recoveryActionsSvc = issueRecoveryActionService(db); + const feedbackDeliveryStatesSvc = feedbackDeliveryCommentStateService(db); const executionWorkspacesSvc = executionWorkspaceServiceDirect(db); const workProductsSvc = workProductService(db); const documentsSvc = documentService(db); @@ -10763,7 +10774,159 @@ export function issueRoutes( order, limit, }); - res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, comments)); + // Operator surfaces only: an agent reading its own thread has no delivery + // banner to act on, and the run reference stays gated behind the same + // board/user authority that guards run detail. + const isOperatorViewer = req.actor.type !== "agent"; + const feedbackDeliveryByComment = await feedbackDeliveryStatesSvc.listForIssue({ + companyId: issue.companyId, + issueId: issue.id, + viewerCanRetry: isOperatorViewer, + viewerCanReadRuns: isOperatorViewer, + }); + const withFeedbackDelivery = feedbackDeliveryByComment.size === 0 + ? comments + : comments.map((comment) => { + const feedbackDelivery = feedbackDeliveryByComment.get(comment.id) ?? null; + return feedbackDelivery ? { ...comment, feedbackDelivery } : comment; + }); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, withFeedbackDelivery)); + }); + + /** + * Explicit operator "Retry delivery" for a human comment whose automatic + * replay budget is exhausted. Board-only: this is the operator escape hatch + * from the exhausted-needs-attention state, not an agent-callable API. + * + * The recovery action stays open. It is discharged by the normal + * revalidation path once the replay actually produces a disposition, which is + * what turns the banner into the recovered notice — a retry request alone must + * not clear the attention item. + */ + router.post("/issues/:id/feedback-delivery/retry", async (req, res) => { + assertBoard(req); + const id = req.params.id as string; + const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found"); + if (!issue) return; + + const actor = getActorInfo(req); + const readStates = () => + feedbackDeliveryStatesSvc.listForIssue({ + companyId: issue.companyId, + issueId: issue.id, + viewerCanRetry: true, + viewerCanReadRuns: true, + }); + + const activeAction = await recoveryActionsSvc.getActiveForIssue(issue.companyId, issue.id); + const isFeedbackDeliveryAction = + activeAction?.kind === FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND + && activeAction.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE; + if (!activeAction || !isFeedbackDeliveryAction) { + const states = await readStates(); + res.json({ + outcome: "not_exhausted", + message: "This task has no exhausted feedback delivery to retry.", + feedbackDelivery: states.values().next().value ?? null, + } satisfies IssueFeedbackDeliveryRetryResponse); + return; + } + + const evidence = (typeof activeAction.evidence === "object" && activeAction.evidence !== null + ? activeAction.evidence + : {}) as Record; + const rootWakeupRequestId = typeof evidence.rootWakeupRequestId === "string" + ? evidence.rootWakeupRequestId + : null; + const outstandingCommentIds = Array.isArray(evidence.outstandingCommentIds) + ? evidence.outstandingCommentIds.filter((entry): entry is string => typeof entry === "string") + : []; + // The original assignee owns the feedback: hand the replay back to it rather + // than to whichever manager the recovery ladder parked the action on. + const targetAgentId = activeAction.returnOwnerAgentId + ?? activeAction.previousOwnerAgentId + ?? issue.assigneeAgentId; + if (!targetAgentId || !rootWakeupRequestId) { + res.json({ + outcome: "no_invokable_assignee", + message: "This delivery has no agent to retry against. Reassign the task first.", + feedbackDelivery: (await readStates()).get(outstandingCommentIds[0] ?? "") ?? null, + } satisfies IssueFeedbackDeliveryRetryResponse); + return; + } + + const statesBefore = await readStates(); + const alreadyInFlight = outstandingCommentIds.some( + (commentId) => statesBefore.get(commentId)?.retryInFlight === true, + ); + if (alreadyInFlight) { + res.json({ + outcome: "already_queued", + message: "A delivery retry is already queued.", + feedbackDelivery: statesBefore.get(outstandingCommentIds[0] ?? "") ?? null, + } satisfies IssueFeedbackDeliveryRetryResponse); + return; + } + + const latestCommentId = outstandingCommentIds.at(-1) ?? null; + const deliveryContext = { + issueId: issue.id, + taskId: issue.id, + feedbackDeliveryRootWakeupRequestId: rootWakeupRequestId, + feedbackDeliveryRetryGeneration: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + feedbackDeliveryOperatorRequested: true, + [FEEDBACK_DELIVERY_WAKE_COMMENT_IDS_KEY]: outstandingCommentIds, + ...(latestCommentId ? { commentId: latestCommentId, wakeCommentId: latestCommentId } : {}), + }; + await heartbeat.wakeup(targetAgentId, { + source: "on_demand", + triggerDetail: "manual", + reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + payload: deliveryContext, + idempotencyKey: buildFeedbackDeliveryManualRetryIdempotencyKey({ + fingerprint: activeAction.fingerprint, + attempt: (activeAction.attemptCount ?? 0) + 1, + }), + requestedByActorType: actor.actorType === "user" ? "user" : "system", + requestedByActorId: actor.actorId, + contextSnapshot: { + ...deliveryContext, + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + source: "issue.operator_feedback_delivery_retry", + instruction: + "An operator asked Paperclip to deliver human feedback on this task again. Handle the outstanding " + + "comment(s) now and record a valid disposition.", + }, + }); + + await logActivity(db, { + companyId: issue.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: targetAgentId, + action: "issue.feedback_delivery_retry_requested", + entityType: "issue", + entityId: issue.id, + details: { + recoveryActionId: activeAction.id, + rootWakeupRequestId, + sourceCommentIds: outstandingCommentIds, + attempt: (activeAction.attemptCount ?? 0) + 1, + }, + }); + + const statesAfter = await readStates(); + const projected = latestCommentId ? statesAfter.get(latestCommentId) ?? null : null; + res.json({ + outcome: "queued", + message: "Retrying feedback delivery.", + // Deferred behind a live run still reads as in-flight to the operator, so + // fall back to a synthetic in-flight projection rather than reporting a + // state that would re-enable the button it just disabled. + feedbackDelivery: projected + ? { ...projected, retryInFlight: true, canRetry: false } + : null, + } satisfies IssueFeedbackDeliveryRetryResponse); }); router.get("/issues/:id/interactions", async (req, res) => { diff --git a/server/src/services/feedback-delivery-comment-states.ts b/server/src/services/feedback-delivery-comment-states.ts new file mode 100644 index 00000000000..32f2a82b8c7 --- /dev/null +++ b/server/src/services/feedback-delivery-comment-states.ts @@ -0,0 +1,228 @@ +import { and, eq, gte, inArray, or, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions } from "@paperclipai/db"; +import type { IssueCommentFeedbackDelivery } from "@paperclipai/shared"; + +import { + buildFeedbackDeliveryFingerprint, + FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + readFeedbackDeliveryWakeCommentIds, +} from "./recovery/feedback-delivery.js"; +import { + FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS, + projectFeedbackDeliveryByComment, + type FeedbackDeliveryPendingReplay, +} from "./recovery/feedback-delivery-projection.js"; + +/** Run statuses that still represent a delivery attempt the operator is waiting on. */ +const PENDING_REPLAY_RUN_STATUSES = ["queued", "scheduled_retry", "running"] as const; + +/** + * Wake statuses that still represent an undelivered replay. A replay deferred + * behind a live run on the same issue is still something the operator is + * waiting on, so it reads as `retrying` just like a queued one. + */ +const PENDING_REPLAY_WAKE_STATUSES = ["queued", "deferred_issue_execution"] as const; + +function readRecord(value: unknown): Record { + return typeof value === "object" && value !== null ? (value as Record) : {}; +} + +function readNonEmptyString(value: unknown) { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function readGeneration(value: unknown) { + return typeof value === "number" && Number.isFinite(value) ? Math.max(0, Math.floor(value)) : 0; +} + +/** + * The two replay lanes spell the wake payload slightly differently — the + * immediate lane writes `rootWakeupRequestId`/`commentIds`, the backstop lane + * writes the run-context spelling. Read both so neither lane goes invisible. + */ +function readReplayRootWakeupRequestId(payload: Record) { + return ( + readNonEmptyString(payload.feedbackDeliveryRootWakeupRequestId) + ?? readNonEmptyString(payload.rootWakeupRequestId) + ); +} + +function readReplayCommentIds(payload: Record) { + const merged = [...readFeedbackDeliveryWakeCommentIds(payload)]; + const legacy = payload.commentIds; + if (Array.isArray(legacy)) { + for (const entry of legacy) { + const normalized = readNonEmptyString(entry); + if (normalized && !merged.includes(normalized)) merged.push(normalized); + } + } + return merged; +} + +/** + * Reads the durable feedback-delivery signals for one issue and projects them + * onto the individual human comments they belong to. + * + * Three cheap, index-backed reads, all scoped to the issue: + * - `feedback_delivery` recovery actions (active plus recently resolved), + * - queued replay wakes (`payload->>'issueId'` index), + * - non-terminal runs carrying replay context (`contextSnapshot->>'issueId'`). + * + * The projection itself is pure and unit-tested separately. + */ +export function feedbackDeliveryCommentStateService(db: Db) { + async function loadRecoveryActions(input: { + companyId: string; + issueId: string; + resolvedSince: Date; + }) { + return db + .select({ + id: issueRecoveryActions.id, + kind: issueRecoveryActions.kind, + cause: issueRecoveryActions.cause, + status: issueRecoveryActions.status, + fingerprint: issueRecoveryActions.fingerprint, + evidence: issueRecoveryActions.evidence, + attemptCount: issueRecoveryActions.attemptCount, + maxAttempts: issueRecoveryActions.maxAttempts, + lastAttemptAt: issueRecoveryActions.lastAttemptAt, + resolvedAt: issueRecoveryActions.resolvedAt, + createdAt: issueRecoveryActions.createdAt, + }) + .from(issueRecoveryActions) + .where( + and( + eq(issueRecoveryActions.companyId, input.companyId), + eq(issueRecoveryActions.sourceIssueId, input.issueId), + eq(issueRecoveryActions.kind, FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND), + or( + inArray(issueRecoveryActions.status, ["active", "escalated"]), + gte(issueRecoveryActions.resolvedAt, input.resolvedSince), + ), + ), + ); + } + + async function loadPendingReplays(input: { + companyId: string; + issueId: string; + }): Promise { + const [wakes, runs] = await Promise.all([ + db + .select({ + agentId: agentWakeupRequests.agentId, + payload: agentWakeupRequests.payload, + runId: agentWakeupRequests.runId, + requestedAt: agentWakeupRequests.requestedAt, + }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, input.companyId), + inArray(agentWakeupRequests.status, [...PENDING_REPLAY_WAKE_STATUSES]), + eq(agentWakeupRequests.reason, FEEDBACK_DELIVERY_RETRY_WAKE_REASON), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${input.issueId}`, + ), + ), + db + .select({ + id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + contextSnapshot: heartbeatRuns.contextSnapshot, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + scheduledRetryAt: heartbeatRuns.scheduledRetryAt, + createdAt: heartbeatRuns.createdAt, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, input.companyId), + inArray(heartbeatRuns.status, [...PENDING_REPLAY_RUN_STATUSES]), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${input.issueId}`, + sql`${heartbeatRuns.contextSnapshot} ->> 'feedbackDeliveryRootWakeupRequestId' is not null`, + ), + ), + ]); + + const replays: FeedbackDeliveryPendingReplay[] = []; + + for (const wake of wakes) { + const payload = readRecord(wake.payload); + const rootWakeupRequestId = readReplayRootWakeupRequestId(payload); + if (!rootWakeupRequestId) continue; + replays.push({ + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId: input.companyId, + issueId: input.issueId, + agentId: wake.agentId, + rootWakeupRequestId, + }), + commentIds: readReplayCommentIds(payload), + generation: readGeneration(payload.feedbackDeliveryRetryGeneration), + runId: wake.runId ?? null, + nextAttemptAt: null, + requestedAt: wake.requestedAt, + }); + } + + for (const run of runs) { + const context = readRecord(run.contextSnapshot); + const rootWakeupRequestId = readReplayRootWakeupRequestId(context) ?? run.wakeupRequestId; + if (!rootWakeupRequestId) continue; + replays.push({ + fingerprint: buildFeedbackDeliveryFingerprint({ + companyId: input.companyId, + issueId: input.issueId, + agentId: run.agentId, + rootWakeupRequestId, + }), + commentIds: readReplayCommentIds(context), + generation: readGeneration(context.feedbackDeliveryRetryGeneration), + runId: run.id, + nextAttemptAt: run.scheduledRetryAt, + requestedAt: run.createdAt, + }); + } + + return replays; + } + + return { + /** + * Returns a comment-id keyed map of delivery state, or an empty map when + * this issue has no feedback-delivery history at all (the common case). + */ + async listForIssue(input: { + companyId: string; + issueId: string; + viewerCanRetry: boolean; + viewerCanReadRuns: boolean; + now?: Date; + }): Promise> { + const now = input.now ?? new Date(); + // Issued together: the first automatic replay is `retrying` *before* any + // recovery action exists, so recovery actions cannot gate the replay read. + // All three predicates are index-backed and scoped to one issue. + const [recoveryActions, pendingReplays] = await Promise.all([ + loadRecoveryActions({ + companyId: input.companyId, + issueId: input.issueId, + resolvedSince: new Date(now.getTime() - FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS), + }), + loadPendingReplays({ companyId: input.companyId, issueId: input.issueId }), + ]); + if (recoveryActions.length === 0 && pendingReplays.length === 0) return new Map(); + + return projectFeedbackDeliveryByComment({ + recoveryActions, + pendingReplays, + viewerCanRetry: input.viewerCanRetry, + viewerCanReadRuns: input.viewerCanReadRuns, + now, + }); + }, + }; +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 358fe7357f5..4150426ebb9 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -151,6 +151,7 @@ import { WORKTREE_INSTANCE_ROOT_METADATA_KEY, } from "./workspace-instance-cleanup.js"; import { issueService } from "./issues.js"; +import { issueRecoveryActionService } from "./issue-recovery-actions.js"; import { projectService } from "./projects.js"; import { authorizationService, type AuthorizationActor } from "./authorization.js"; import { createToolGatewayService } from "./tool-gateway.js"; @@ -223,6 +224,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 { @@ -485,6 +496,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 { 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 @@ -6739,6 +6751,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; const budgets = budgetService(db, budgetHooks); const recovery = recoveryService(db, { enqueueWakeup }); + const recoveryActions = issueRecoveryActionService(db); function isPlanApprovalConfirmationPayload(payload: unknown) { const target = parseObject(parseObject(payload).target); @@ -10065,7 +10078,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { outcome: "satisfied" as const, queuedRun: null }; } - if (readNonEmptyString(contextSnapshot.retryReason) === "missing_issue_comment") { + const issueCommentRetryReason = readNonEmptyString(contextSnapshot.retryReason); + if ( + issueCommentRetryReason === "missing_issue_comment" || + issueCommentRetryReason === FEEDBACK_DELIVERY_RETRY_REASON + ) { await patchRunIssueCommentStatus(run.id, { issueCommentStatus: "retry_exhausted", issueCommentSatisfiedByCommentId: null, @@ -10074,7 +10091,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eventType: "lifecycle", stream: "system", level: "warn", - message: "Run ended without an issue comment after one retry; no further comment wake will be queued", + message: + issueCommentRetryReason === FEEDBACK_DELIVERY_RETRY_REASON + ? "Recovered feedback run ended without handling evidence; no separate missing-comment retry will be queued" + : "Run ended without an issue comment after one retry; no further comment wake will be queued", }); return { outcome: "retry_exhausted" as const, queuedRun: null }; } @@ -10123,6 +10143,663 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { outcome: "retry_exhausted" as const, queuedRun: null }; } + type FeedbackDeliveryContext = { + issueId: string; + rootWakeupRequestId: string; + commentIds: string[]; + generation: number; + fingerprint: string; + }; + + async function readFeedbackDeliveryContext( + run: typeof heartbeatRuns.$inferSelect, + ): Promise { + const context = parseObject(run.contextSnapshot); + const issueId = readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId); + const rootWakeupRequestId = + readNonEmptyString(context.feedbackDeliveryRootWakeupRequestId) ?? run.wakeupRequestId; + if (!issueId || !rootWakeupRequestId) return null; + + const wakeReason = readNonEmptyString(context.wakeReason); + const commentIds = mergeWakeCommentIds(context, deriveCommentId(context, null)); + const carriesExplicitResume = carriesExplicitFeedbackResume(context); + const eligibleWake = isEligibleFeedbackDeliveryWake({ wakeReason, carriesExplicitResume }); + if (!eligibleWake || (commentIds.length === 0 && !carriesExplicitResume)) return null; + + const rootWake = await db + .select({ requestedByActorType: agentWakeupRequests.requestedByActorType }) + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.id, rootWakeupRequestId), + eq(agentWakeupRequests.companyId, run.companyId), + eq(agentWakeupRequests.agentId, run.agentId), + )) + .then((rows) => rows[0] ?? null); + // Durable feedback delivery is a human-input contract. System-created + // generation-1 wakes retain their original user wake id in context. + if (rootWake?.requestedByActorType !== "user") return null; + + return { + issueId, + rootWakeupRequestId, + commentIds, + 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, + agentId: run.agentId, + rootWakeupRequestId, + }), + }; + } + + async function findFeedbackHandlingEvidence( + run: typeof heartbeatRuns.$inferSelect, + issueId: string, + ) { + const [comment, documentRevision, workProduct, issueActivity] = await Promise.all([ + findRunIssueComment(run.id, run.companyId, issueId), + db + .select({ id: documentRevisions.id }) + .from(documentRevisions) + .innerJoin(issueDocuments, eq(issueDocuments.documentId, documentRevisions.documentId)) + .where(and( + eq(documentRevisions.companyId, run.companyId), + eq(documentRevisions.createdByRunId, run.id), + eq(issueDocuments.companyId, run.companyId), + eq(issueDocuments.issueId, issueId), + )) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: issueWorkProducts.id }) + .from(issueWorkProducts) + .where(and( + eq(issueWorkProducts.companyId, run.companyId), + eq(issueWorkProducts.issueId, issueId), + eq(issueWorkProducts.createdByRunId, run.id), + )) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: activityLog.id, action: activityLog.action }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, run.companyId), + eq(activityLog.entityType, "issue"), + eq(activityLog.entityId, issueId), + eq(activityLog.runId, run.id), + inArray(activityLog.action, ISSUE_PROGRESS_ACTIVITY_ACTIONS), + )) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + if (comment) return { kind: "comment", id: comment.id } as const; + if (documentRevision) return { kind: "document_revision", id: documentRevision.id } as const; + if (workProduct) return { kind: "work_product", id: workProduct.id } as const; + if (issueActivity) return { kind: "issue_activity", id: issueActivity.id, action: issueActivity.action } as const; + return null; + } + + async function logFeedbackRetryQueuedOnce(input: { + run: typeof heartbeatRuns.$inferSelect; + delivery: FeedbackDeliveryContext; + retryRunId: string | null; + retryWakeupRequestId: string | null; + disposition: "queued" | "coalesced" | "deferred"; + }) { + const existing = await db + .select({ id: activityLog.id }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, input.run.companyId), + eq(activityLog.entityType, "issue"), + eq(activityLog.entityId, input.delivery.issueId), + eq(activityLog.action, "issue.feedback_retry_queued"), + sql`${activityLog.details} ->> 'rootWakeupRequestId' = ${input.delivery.rootWakeupRequestId}`, + )) + .limit(1) + .then((rows) => rows[0] ?? null); + if (existing) return; + await logActivity(db, { + companyId: input.run.companyId, + actorType: "system", + actorId: "heartbeat", + agentId: input.run.agentId, + runId: input.run.id, + action: "issue.feedback_retry_queued", + entityType: "issue", + entityId: input.delivery.issueId, + details: { + sourceRunId: input.run.id, + sourceWakeupRequestId: input.run.wakeupRequestId, + rootWakeupRequestId: input.delivery.rootWakeupRequestId, + sourceCommentIds: input.delivery.commentIds, + retryRunId: input.retryRunId, + retryWakeupRequestId: input.retryWakeupRequestId, + retryGeneration: 1, + disposition: input.disposition, + }, + }); + } + + async function ensureFeedbackDeliveryRecovery(input: { + run: typeof heartbeatRuns.$inferSelect; + agent: typeof agents.$inferSelect; + delivery: FeedbackDeliveryContext; + gateReason?: string | null; + gateErrorCode?: string | null; + }) { + const issue = await db + .select({ + id: issues.id, + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + }) + .from(issues) + .where(and(eq(issues.id, input.delivery.issueId), eq(issues.companyId, input.run.companyId))) + .then((rows) => rows[0] ?? null); + if (!issue) return { kind: "suppressed" as const, reason: "issue_not_found" }; + if (issue.status === "done" || issue.status === "cancelled") { + return { kind: "suppressed" as const, reason: "issue_terminal" }; + } + if (issue.assigneeAgentId !== input.run.agentId || issue.assigneeUserId) { + return { kind: "suppressed" as const, reason: "ownership_changed" }; + } + + const existing = await recoveryActions.getActiveForIssue(input.run.companyId, issue.id); + if ( + 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 }; + } + + const action = await recoveryActions.upsertSourceScoped({ + companyId: input.run.companyId, + sourceIssueId: issue.id, + kind: FEEDBACK_DELIVERY_RECOVERY_ACTION_KIND, + ownerType: "agent", + ownerAgentId: input.agent.id, + previousOwnerAgentId: input.agent.id, + returnOwnerAgentId: input.agent.id, + cause: FEEDBACK_DELIVERY_RECOVERY_CAUSE, + fingerprint: input.delivery.fingerprint, + evidence: { + sourceRunId: input.run.id, + sourceWakeupRequestId: input.run.wakeupRequestId, + rootWakeupRequestId: input.delivery.rootWakeupRequestId, + outstandingCommentIds: input.delivery.commentIds, + retryGeneration: input.delivery.generation, + lastFailureCode: input.run.errorCode, + lastFailureReason: input.run.error, + gateErrorCode: input.gateErrorCode ?? null, + gateReason: input.gateReason ?? null, + }, + nextAction: [ + input.gateReason, + "Restore an invokable current assignee and valid workspace, then retry delivery explicitly or record a valid manual issue disposition.", + ].filter(Boolean).join(" "), + wakePolicy: { + type: "manual_repair_required", + reason: FEEDBACK_DELIVERY_RECOVERY_CAUSE, + ownerAgentId: input.agent.id, + }, + maxAttempts: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + lastAttemptAt: new Date(), + }); + + await logActivity(db, { + companyId: input.run.companyId, + actorType: "system", + actorId: "heartbeat", + agentId: input.run.agentId, + runId: input.run.id, + action: "issue.feedback_recovery_exhausted", + entityType: "issue", + entityId: issue.id, + details: { + recoveryActionId: action.id, + sourceRunId: input.run.id, + sourceWakeupRequestId: input.run.wakeupRequestId, + rootWakeupRequestId: input.delivery.rootWakeupRequestId, + sourceCommentIds: input.delivery.commentIds, + retryGeneration: input.delivery.generation, + errorCode: input.run.errorCode, + gateErrorCode: input.gateErrorCode ?? null, + }, + }); + return { kind: "recovery" as const, action, reusedExisting: false }; + } + + async function enqueueFeedbackDeliveryRetry(input: { + run: typeof heartbeatRuns.$inferSelect; + agent: typeof agents.$inferSelect; + delivery: FeedbackDeliveryContext; + }) { + const now = new Date(); + const sourceContext = parseObject(input.run.contextSnapshot); + const retryContext = withRecoveryModelProfileHint({ + ...sourceContext, + issueId: input.delivery.issueId, + taskId: input.delivery.issueId, + wakeReason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + retryOfRunId: input.run.id, + feedbackDeliveryRootWakeupRequestId: input.delivery.rootWakeupRequestId, + feedbackDeliverySourceRunId: input.run.id, + feedbackDeliveryRetryGeneration: 1, + [WAKE_COMMENT_IDS_KEY]: input.delivery.commentIds, + ...(input.delivery.commentIds.length > 0 + ? { + commentId: input.delivery.commentIds.at(-1), + wakeCommentId: input.delivery.commentIds.at(-1), + } + : {}), + }, "normal_model"); + const taskKey = deriveTaskKeyWithHeartbeatFallback(retryContext, null); + const sessionBefore = await resolveSessionBeforeForWakeup(input.agent, taskKey); + const responsibleUserId = await resolveResponsibleUserIdForRunContext(input.run, retryContext); + const idempotencyKey = buildFeedbackDeliveryRetryIdempotencyKey({ + fingerprint: input.delivery.fingerprint, + generation: 1, + }); + + const result = await db.transaction(async (tx) => { + const companyIsActive = await tx + .select({ status: companies.status }) + .from(companies) + .where(eq(companies.id, input.run.companyId)) + .then((rows) => rows[0]?.status === "active"); + if (!companyIsActive) return { kind: "suppressed" as const }; + + await tx.execute(sql` + select id from issues + where id = ${input.delivery.issueId} and company_id = ${input.run.companyId} + for update + `); + const issue = await tx + .select({ + id: issues.id, + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(and( + eq(issues.id, input.delivery.issueId), + eq(issues.companyId, input.run.companyId), + )) + .then((rows) => rows[0] ?? null); + if ( + !issue || + issue.assigneeAgentId !== input.run.agentId || + issue.assigneeUserId || + issue.status === "done" || + issue.status === "cancelled" + ) { + return { kind: "suppressed" as const }; + } + + const liveRun = await tx + .select() + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, input.run.companyId), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${input.delivery.issueId}`, + ne(heartbeatRuns.id, input.run.id), + )) + .orderBy( + sql`case when ${heartbeatRuns.id} = ${issue.executionRunId} then 0 when ${heartbeatRuns.status} = 'running' then 1 else 2 end`, + desc(heartbeatRuns.createdAt), + desc(heartbeatRuns.id), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + + const orderCommentIds = async (ids: string[]) => { + if (ids.length === 0) return ids; + const rows = await tx + .select({ id: issueComments.id }) + .from(issueComments) + .where(and( + eq(issueComments.companyId, input.run.companyId), + eq(issueComments.issueId, input.delivery.issueId), + inArray(issueComments.id, ids), + )) + .orderBy(asc(issueComments.createdAt), asc(issueComments.id)); + const ordered = rows.map((row) => row.id); + return [...ordered, ...ids.filter((id) => !ordered.includes(id))]; + }; + + const liveContext = parseObject(liveRun?.contextSnapshot); + if ( + liveRun && + readNonEmptyString(liveContext.feedbackDeliveryRootWakeupRequestId) === + input.delivery.rootWakeupRequestId && + Number(liveContext.feedbackDeliveryRetryGeneration) >= 1 + ) { + return { + kind: "coalesced" as const, + run: liveRun, + wakeupRequestId: liveRun.wakeupRequestId, + }; + } + + if ( + liveRun && + liveRun.agentId === input.run.agentId && + liveRun.status !== "running" + ) { + const mergedContext = mergeCoalescedContextSnapshot(retryContext, parseObject(liveRun.contextSnapshot)); + const orderedCommentIds = await orderCommentIds(mergeWakeCommentIds(retryContext, liveRun.contextSnapshot)); + if (orderedCommentIds.length > 0) { + mergedContext[WAKE_COMMENT_IDS_KEY] = orderedCommentIds; + mergedContext.commentId = orderedCommentIds.at(-1); + mergedContext.wakeCommentId = orderedCommentIds.at(-1); + } + mergedContext.feedbackDeliveryRootWakeupRequestId = input.delivery.rootWakeupRequestId; + mergedContext.feedbackDeliverySourceRunId = input.run.id; + mergedContext.feedbackDeliveryRetryGeneration = 1; + const mergedRun = await tx + .update(heartbeatRuns) + .set({ contextSnapshot: mergedContext, updatedAt: now }) + .where(eq(heartbeatRuns.id, liveRun.id)) + .returning() + .then((rows) => rows[0] ?? liveRun); + return { + kind: "coalesced" as const, + run: mergedRun, + wakeupRequestId: mergedRun.wakeupRequestId, + }; + } + + const existingDeferred = await tx + .select() + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.companyId, input.run.companyId), + eq(agentWakeupRequests.agentId, input.run.agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${input.delivery.issueId}`, + )) + .orderBy(asc(agentWakeupRequests.requestedAt), asc(agentWakeupRequests.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + + if (liveRun) { + const existingPayload = parseObject(existingDeferred?.payload); + const existingDeferredContext = parseObject(existingPayload[DEFERRED_WAKE_CONTEXT_KEY]); + const mergedDeferredContext = mergeCoalescedContextSnapshot(retryContext, existingDeferredContext); + const orderedCommentIds = await orderCommentIds( + mergeWakeCommentIds(retryContext, existingDeferredContext), + ); + if (orderedCommentIds.length > 0) { + mergedDeferredContext[WAKE_COMMENT_IDS_KEY] = orderedCommentIds; + mergedDeferredContext.commentId = orderedCommentIds.at(-1); + mergedDeferredContext.wakeCommentId = orderedCommentIds.at(-1); + } + mergedDeferredContext.feedbackDeliveryRootWakeupRequestId = input.delivery.rootWakeupRequestId; + mergedDeferredContext.feedbackDeliverySourceRunId = input.run.id; + mergedDeferredContext.feedbackDeliveryRetryGeneration = 1; + const payload = { + ...existingPayload, + issueId: input.delivery.issueId, + retryOfRunId: input.run.id, + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + [DEFERRED_WAKE_CONTEXT_KEY]: mergedDeferredContext, + }; + if (existingDeferred) { + const updated = await tx + .update(agentWakeupRequests) + .set({ + payload, + coalescedCount: (existingDeferred.coalescedCount ?? 0) + 1, + updatedAt: now, + }) + .where(eq(agentWakeupRequests.id, existingDeferred.id)) + .returning() + .then((rows) => rows[0] ?? existingDeferred); + return { kind: "deferred" as const, run: liveRun, wakeupRequestId: updated.id }; + } + const deferred = await tx + .insert(agentWakeupRequests) + .values({ + companyId: input.run.companyId, + agentId: input.run.agentId, + source: "automation", + triggerDetail: "system", + reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + payload, + status: "deferred_issue_execution", + requestedByActorType: "system", + requestedByActorId: null, + idempotencyKey, + updatedAt: now, + }) + .returning() + .then((rows) => rows[0]); + return { kind: "deferred" as const, run: liveRun, wakeupRequestId: deferred.id }; + } + + const wakeupRequest = await tx + .insert(agentWakeupRequests) + .values({ + companyId: input.run.companyId, + agentId: input.run.agentId, + source: "automation", + triggerDetail: "system", + reason: FEEDBACK_DELIVERY_RETRY_WAKE_REASON, + payload: { + issueId: input.delivery.issueId, + retryOfRunId: input.run.id, + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + rootWakeupRequestId: input.delivery.rootWakeupRequestId, + commentIds: input.delivery.commentIds, + }, + status: "queued", + requestedByActorType: "system", + requestedByActorId: null, + idempotencyKey, + updatedAt: now, + }) + .returning() + .then((rows) => rows[0]); + const retryRun = await tx + .insert(heartbeatRuns) + .values({ + companyId: input.run.companyId, + agentId: input.run.agentId, + invocationSource: "automation", + triggerDetail: "system", + status: "queued", + wakeupRequestId: wakeupRequest.id, + contextSnapshot: retryContext, + responsibleUserId, + sessionIdBefore: sessionBefore, + retryOfRunId: input.run.id, + scheduledRetryAttempt: 1, + scheduledRetryReason: FEEDBACK_DELIVERY_RETRY_REASON, + updatedAt: now, + }) + .returning() + .then((rows) => rows[0]); + await tx + .update(agentWakeupRequests) + .set({ runId: retryRun.id, updatedAt: now }) + .where(eq(agentWakeupRequests.id, wakeupRequest.id)); + await tx + .update(issues) + .set({ + executionRunId: retryRun.id, + executionAgentNameKey: normalizeAgentNameKey(input.agent.name), + executionLockedAt: now, + updatedAt: now, + }) + .where(and(eq(issues.id, issue.id), eq(issues.companyId, input.run.companyId))); + return { kind: "queued" as const, run: retryRun, wakeupRequestId: wakeupRequest.id }; + }); + + if (result.kind === "queued") { + publishLiveEvent({ + companyId: result.run.companyId, + type: "heartbeat.run.queued", + payload: { + runId: result.run.id, + agentId: result.run.agentId, + invocationSource: result.run.invocationSource, + triggerDetail: result.run.triggerDetail, + wakeupRequestId: result.run.wakeupRequestId, + }, + }); + await startNextQueuedRunForAgent(result.run.agentId); + } + return result; + } + + async function handleFeedbackDeliveryDisposition(input: { + run: typeof heartbeatRuns.$inferSelect; + agent: typeof agents.$inferSelect; + existingRetryRun?: typeof heartbeatRuns.$inferSelect | null; + }) { + const delivery = await readFeedbackDeliveryContext(input.run); + if (!delivery) return { kind: "not_applicable" as const }; + + const handlingEvidence = await findFeedbackHandlingEvidence(input.run, delivery.issueId); + if (handlingEvidence) { + return { kind: "handled" as const, delivery, handlingEvidence }; + } + + const preLaunchProcessLoss = + input.run.errorCode === "process_lost" && + !input.run.processPid && + !input.run.processGroupId; + const automaticReplayExhausted = delivery.generation >= FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES; + if (automaticReplayExhausted) { + const recovery = await ensureFeedbackDeliveryRecovery({ + run: input.run, + agent: input.agent, + delivery, + }); + return { kind: "recovery" as const, delivery, recovery }; + } + if (!preLaunchProcessLoss && !input.existingRetryRun) { + return { kind: "not_applicable" as const }; + } + + const gate = await evaluateScheduledRetryGate({ + run: input.run, + agent: input.agent, + contextSnapshot: parseObject(input.run.contextSnapshot), + retryReason: FEEDBACK_DELIVERY_RETRY_REASON, + }); + if (!gate.allowed) { + if ( + gate.errorCode === "issue_not_found" || + gate.errorCode === "issue_reassigned" || + gate.errorCode === "issue_cancelled" || + gate.errorCode === "issue_terminal_status" + ) { + return { kind: "suppressed" as const, delivery, reason: gate.errorCode }; + } + const recovery = await ensureFeedbackDeliveryRecovery({ + run: input.run, + agent: input.agent, + delivery, + gateReason: gate.reason, + gateErrorCode: gate.errorCode, + }); + return { kind: "recovery" as const, delivery, recovery }; + } + + if (input.existingRetryRun) { + const retryContext: Record = { + ...parseObject(input.existingRetryRun.contextSnapshot), + feedbackDeliveryRootWakeupRequestId: delivery.rootWakeupRequestId, + feedbackDeliverySourceRunId: input.run.id, + feedbackDeliveryRetryGeneration: 1, + [WAKE_COMMENT_IDS_KEY]: delivery.commentIds, + }; + if (delivery.commentIds.length > 0) { + retryContext.commentId = delivery.commentIds.at(-1); + retryContext.wakeCommentId = delivery.commentIds.at(-1); + } + const updatedRetry = await db + .update(heartbeatRuns) + .set({ contextSnapshot: retryContext, updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, input.existingRetryRun.id)) + .returning() + .then((rows) => rows[0] ?? input.existingRetryRun!); + await logFeedbackRetryQueuedOnce({ + run: input.run, + delivery, + retryRunId: updatedRetry.id, + retryWakeupRequestId: updatedRetry.wakeupRequestId, + disposition: "queued", + }); + return { kind: "retry" as const, delivery, run: updatedRetry, disposition: "queued" as const }; + } + + const retry = await enqueueFeedbackDeliveryRetry({ run: input.run, agent: input.agent, delivery }); + if (retry.kind === "suppressed") { + return { kind: "suppressed" as const, delivery, reason: "state_changed" }; + } + await logFeedbackRetryQueuedOnce({ + run: input.run, + delivery, + retryRunId: retry.run.id, + retryWakeupRequestId: retry.wakeupRequestId, + disposition: retry.kind, + }); + return { kind: "retry" as const, delivery, run: retry.run, disposition: retry.kind }; + } + + async function applyPostRunDisposition(input: { + run: typeof heartbeatRuns.$inferSelect; + agent: typeof agents.$inferSelect; + existingRetryRun?: typeof heartbeatRuns.$inferSelect | null; + scheduleInteractionRetry?: boolean; + suppressImmediateRecovery?: boolean; + }) { + const feedback = await handleFeedbackDeliveryDisposition({ + run: input.run, + agent: input.agent, + existingRetryRun: input.existingRetryRun, + }); + let interactionRetry = null; + if ( + input.scheduleInteractionRetry !== false && + feedback.kind === "not_applicable" && + UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES.includes( + input.run.status as (typeof UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES)[number], + ) + ) { + interactionRetry = await scheduleInteractionContinuationInfrastructureRetryIfEligible( + input.run, + input.agent, + ); + } + await releaseIssueExecutionAndPromote(input.run, { + suppressImmediateRecovery: + input.suppressImmediateRecovery === true || + feedback.kind === "retry" || + feedback.kind === "recovery" || + interactionRetry?.outcome === "scheduled", + }); + await handleIssueReviewPathDisposition(input.run); + return { feedback, interactionRetry }; + } + async function enqueueProcessLossRetry( run: typeof heartbeatRuns.$inferSelect, agent: typeof agents.$inferSelect, @@ -10454,6 +11131,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .select({ run: heartbeatRuns, adapterType: agents.adapterType, + agent: agents, }) .from(heartbeatRuns) .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) @@ -10611,6 +11289,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) classify(candidate, "adopted", processPidAlive ? "process_pid_alive" : "process_group_alive", patch); } + // A run can finalize while the server is down (for example through an + // external callback). Re-run the same idempotent disposition pipeline used + // by live execution and startup reaping so restart reconciliation cannot + // strand feedback or a consumed review path. + for (const runId of new Set(finalizedWhileDownRunIds)) { + const current = currentByRunId.get(runId); + if (!current || !isHeartbeatRunTerminalStatus(current.run.status)) continue; + await applyPostRunDisposition({ + run: current.run, + agent: current.agent, + }).catch((error) => { + logger.error( + { err: error, runId }, + "failed to apply post-run disposition during hot-restart reconciliation", + ); + }); + } + const report = await writeHotRestartReport({ version: 1, requestedAt: intent.requestedAt, @@ -10727,11 +11423,15 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); const retry = await enqueueProcessLossRetry(interrupted, agent, now); - if (!retry) { - await releaseIssueExecutionAndPromote(interrupted); - } else { + if (retry) { retryRunIds.push(retry.id); } + await applyPostRunDisposition({ + run: interrupted, + agent, + existingRetryRun: retry, + scheduleInteractionRetry: false, + }); await appendRunEvent(interrupted, await nextRunEventSeq(interrupted.id), { eventType: "lifecycle", @@ -13507,13 +14207,23 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (retryAgent) { retriedRun = await enqueueProcessLossRetry(finalizedRun, retryAgent, now); } - } else if (retryAgent) { - const scheduled = await scheduleInteractionContinuationInfrastructureRetryIfEligible(finalizedRun, retryAgent); - retriedRun = scheduled?.outcome === "scheduled" ? scheduled.run : null; } - if (!retriedRun) { + if (retryAgent) { + const disposition = await applyPostRunDisposition({ + run: finalizedRun, + agent: retryAgent, + existingRetryRun: retriedRun, + scheduleInteractionRetry: !shouldRetry, + }); + if (!retriedRun && disposition.feedback.kind === "retry") { + retriedRun = disposition.feedback.run; + } else if (!retriedRun && disposition.interactionRetry?.outcome === "scheduled") { + retriedRun = disposition.interactionRetry.run; + } + } else { await releaseIssueExecutionAndPromote(finalizedRun); + await handleIssueReviewPathDisposition(finalizedRun); } await appendRunEvent(finalizedRun, await nextRunEventSeq(finalizedRun.id), { @@ -16253,9 +16963,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) await scheduleBoundedRetryForRun(livenessRun, agent); } const issueCommentPolicyResult = await finalizeIssueCommentPolicy(livenessRun, agent); - await releaseIssueExecutionAndPromote(livenessRun); + await applyPostRunDisposition({ run: livenessRun, agent }); await handleRunLivenessContinuation(livenessRun); - await handleIssueReviewPathDisposition(livenessRun); await handleSuccessfulRunHandoff( issueCommentPolicyResult.outcome === "retry_queued" || issueCommentPolicyResult.outcome === "retry_exhausted" ? { @@ -16426,9 +17135,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (!isWorkspaceValidationFailedRun(livenessRun) && !isConfigurationIncompleteFailedRun(livenessRun)) { await finalizeIssueCommentPolicy(livenessRun, agent); } - await scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, agent); - await releaseIssueExecutionAndPromote(livenessRun); - await handleIssueReviewPathDisposition(livenessRun); + await applyPostRunDisposition({ run: livenessRun, agent }); await updateRuntimeState(agent, livenessRun, { exitCode: null, @@ -16554,25 +17261,27 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (!isWorkspaceValidationFailedRun(livenessRun) && !isConfigurationIncompleteFailedRun(livenessRun)) { await finalizeIssueCommentPolicy(livenessRun, failedAgent).catch(() => undefined); } - await scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, failedAgent).catch((retryError) => { - logger.warn( - { err: retryError, runId: livenessRun.id }, - "failed to schedule interaction continuation retry after setup failure", + await applyPostRunDisposition({ run: livenessRun, agent: failedAgent }).catch((dispositionError) => { + logger.error( + { err: dispositionError, runId: livenessRun.id }, + "failed to apply post-run disposition after setup failure", + ); + }); + } + if (!failedAgent) { + await releaseIssueExecutionAndPromote(livenessRun).catch((releaseError) => { + logger.error( + { err: releaseError, runId }, + "failed to release issue execution after heartbeat setup failure", + ); + }); + await handleIssueReviewPathDisposition(livenessRun).catch((reviewPathError) => { + logger.error( + { err: reviewPathError, runId }, + "failed to evaluate review-path disposition after heartbeat setup failure", ); }); } - await releaseIssueExecutionAndPromote(livenessRun).catch((releaseError) => { - logger.error( - { err: releaseError, runId }, - "failed to release issue execution after heartbeat setup failure", - ); - }); - await handleIssueReviewPathDisposition(livenessRun).catch((reviewPathError) => { - logger.error( - { err: reviewPathError, runId }, - "failed to evaluate review-path disposition after heartbeat setup failure", - ); - }); } // Ensure the agent is not left stuck in "running" if the setup-failure // path owned the terminal transition. If another path already finalized @@ -16585,6 +17294,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } } finally { let latestRun = await getRun(run.id).catch(() => null); + const dispositionNeededAfterLeaseTerminalization = + latestRun?.status === "running" || latestRun?.status === "queued"; // Close the invariant "environment lease released implies the run is // terminal". When the teardown reaches this point with the run still // running or queued, force a terminal status before the lease is @@ -16598,6 +17309,21 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return latestRun; }); } + if ( + dispositionNeededAfterLeaseTerminalization + && latestRun + && isHeartbeatRunTerminalStatus(latestRun.status) + ) { + const dispositionAgent = await getAgent(latestRun.agentId).catch(() => null); + if (dispositionAgent) { + await applyPostRunDisposition({ run: latestRun, agent: dispositionAgent }).catch((dispositionError) => { + logger.error( + { err: dispositionError, runId: latestRun.id }, + "failed to apply post-run disposition after lease-release terminalization", + ); + }); + } + } await releaseEnvironmentLeasesForRun({ runId: run.id, companyId: run.companyId, diff --git a/server/src/services/index.ts b/server/src/services/index.ts index 2ab2994c246..3ff658fd209 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -49,6 +49,7 @@ export { issueTreeControlService } from "./issue-tree-control.js"; export { issueApprovalService } from "./issue-approvals.js"; export { issueReferenceService } from "./issue-references.js"; export { issueRecoveryActionService } from "./issue-recovery-actions.js"; +export { feedbackDeliveryCommentStateService } from "./feedback-delivery-comment-states.js"; export { stalledReviewDecisionService, type DecideStalledReviewInput, diff --git a/server/src/services/recovery/feedback-delivery-projection.test.ts b/server/src/services/recovery/feedback-delivery-projection.test.ts new file mode 100644 index 00000000000..b8973a397f7 --- /dev/null +++ b/server/src/services/recovery/feedback-delivery-projection.test.ts @@ -0,0 +1,293 @@ +import { describe, expect, it } from "vitest"; + +import { + FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS, + projectFeedbackDeliveryByComment, + toSafeFeedbackDeliveryFailureReason, + type FeedbackDeliveryPendingReplay, + type FeedbackDeliveryRecoveryActionRow, +} from "./feedback-delivery-projection.js"; +import { FEEDBACK_DELIVERY_RECOVERY_CAUSE } from "./feedback-delivery.js"; + +const NOW = new Date("2026-08-14T18:00:00.000Z"); +const FINGERPRINT = "feedback_delivery:company-1:issue-1:agent-1:root-wake-1"; + +function action( + overrides: Partial = {}, +): FeedbackDeliveryRecoveryActionRow { + return { + id: "action-1", + kind: "feedback_delivery", + cause: FEEDBACK_DELIVERY_RECOVERY_CAUSE, + status: "active", + fingerprint: FINGERPRINT, + evidence: { + outstandingCommentIds: ["comment-1"], + rootWakeupRequestId: "root-wake-1", + retryGeneration: 1, + maxAutomaticRetries: 1, + sourceRunId: "run-1", + lastFailureCode: "process_lost", + // Deliberately present: the projection must never surface it. + lastFailureReason: "Error: ECONNREFUSED /home/agent/.secret/token.json", + }, + attemptCount: 1, + maxAttempts: 1, + lastAttemptAt: new Date("2026-08-14T17:58:00.000Z"), + resolvedAt: null, + createdAt: new Date("2026-08-14T17:57:00.000Z"), + ...overrides, + }; +} + +function replay( + overrides: Partial = {}, +): FeedbackDeliveryPendingReplay { + return { + fingerprint: FINGERPRINT, + commentIds: ["comment-1"], + generation: 1, + runId: "run-2", + nextAttemptAt: new Date("2026-08-14T18:00:12.000Z"), + requestedAt: new Date("2026-08-14T17:59:30.000Z"), + ...overrides, + }; +} + +function project(input: { + recoveryActions?: FeedbackDeliveryRecoveryActionRow[]; + pendingReplays?: FeedbackDeliveryPendingReplay[]; + viewerCanRetry?: boolean; + viewerCanReadRuns?: boolean; + now?: Date; +}) { + return projectFeedbackDeliveryByComment({ + recoveryActions: input.recoveryActions ?? [], + pendingReplays: input.pendingReplays ?? [], + viewerCanRetry: input.viewerCanRetry ?? true, + viewerCanReadRuns: input.viewerCanReadRuns ?? true, + now: input.now ?? NOW, + }); +} + +describe("projectFeedbackDeliveryByComment", () => { + it("returns nothing for an issue with no delivery history", () => { + expect(project({}).size).toBe(0); + }); + + it("projects a queued replay as quiet retrying state with no operator action", () => { + const states = project({ pendingReplays: [replay()] }); + const state = states.get("comment-1"); + expect(state).toMatchObject({ + state: "retrying", + sourceCommentId: "comment-1", + attempt: 1, + maxAttempts: 1, + canRetry: false, + retryInFlight: true, + safeFailureReason: null, + nextAttemptAt: "2026-08-14T18:00:12.000Z", + targetRunId: "run-2", + }); + }); + + it("projects an active recovery action as exhausted and offers retry", () => { + const state = project({ recoveryActions: [action()] }).get("comment-1"); + expect(state).toMatchObject({ + state: "exhausted_needs_attention", + attempt: 1, + maxAttempts: 1, + canRetry: true, + retryInFlight: false, + targetRunId: "run-1", + safeFailureReason: "The agent's process stopped before it read the comment.", + batchedCommentCount: 1, + }); + }); + + it("never leaks a raw failure reason into the projection", () => { + const state = project({ + recoveryActions: [ + action({ + evidence: { + outstandingCommentIds: ["comment-1"], + lastFailureCode: "some_unmapped_adapter_code", + lastFailureReason: "Error: ECONNREFUSED /home/agent/.secret/token.json", + gateReason: "workspace /srv/secret/path is dirty", + }, + }), + ], + }).get("comment-1"); + expect(state?.safeFailureReason).toBeNull(); + expect(JSON.stringify(state)).not.toContain("ECONNREFUSED"); + expect(JSON.stringify(state)).not.toContain("/srv/secret/path"); + }); + + it("keeps the actionable banner while an explicit retry is in flight", () => { + const state = project({ + recoveryActions: [action()], + pendingReplays: [replay()], + }).get("comment-1"); + // Still exhausted: clearing it here would drop the attention item before the + // replay has actually delivered anything. + expect(state).toMatchObject({ + state: "exhausted_needs_attention", + retryInFlight: true, + canRetry: false, + }); + }); + + it("projects a recently resolved action as recovered after attention", () => { + const state = project({ + recoveryActions: [ + action({ + status: "resolved", + resolvedAt: new Date("2026-08-14T17:59:00.000Z"), + }), + ], + }).get("comment-1"); + expect(state).toMatchObject({ + state: "recovered_after_attention", + deliveredAt: "2026-08-14T17:59:00.000Z", + canRetry: false, + retryInFlight: false, + safeFailureReason: null, + }); + }); + + it("drops the recovered notice once it ages out of the window", () => { + const states = project({ + recoveryActions: [ + action({ + status: "resolved", + resolvedAt: new Date(NOW.getTime() - FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS - 1000), + }), + ], + }); + expect(states.size).toBe(0); + }); + + it("lets a re-opened obligation win over its own earlier resolution", () => { + const state = project({ + recoveryActions: [ + action({ + id: "action-old", + status: "resolved", + resolvedAt: new Date("2026-08-14T17:30:00.000Z"), + }), + action({ id: "action-new", status: "active" }), + ], + }).get("comment-1"); + expect(state?.state).toBe("exhausted_needs_attention"); + }); + + it("keeps a newer comment's obligation from clearing an older one", () => { + const states = project({ + recoveryActions: [ + action({ + id: "action-old", + fingerprint: `${FINGERPRINT}:old`, + status: "active", + evidence: { outstandingCommentIds: ["comment-1"], retryGeneration: 1 }, + }), + action({ + id: "action-new", + fingerprint: `${FINGERPRINT}:new`, + status: "resolved", + resolvedAt: new Date("2026-08-14T17:59:00.000Z"), + evidence: { outstandingCommentIds: ["comment-2"], retryGeneration: 1 }, + }), + ], + }); + expect(states.get("comment-1")?.state).toBe("exhausted_needs_attention"); + expect(states.get("comment-2")?.state).toBe("recovered_after_attention"); + }); + + it("rolls one batched obligation onto every comment it covers", () => { + const states = project({ + recoveryActions: [ + action({ + evidence: { + outstandingCommentIds: ["comment-1", "comment-2", "comment-3"], + retryGeneration: 1, + }, + }), + ], + }); + expect([...states.keys()]).toEqual(["comment-1", "comment-2", "comment-3"]); + for (const state of states.values()) { + expect(state.batchedCommentCount).toBe(3); + expect(state.state).toBe("exhausted_needs_attention"); + } + }); + + it("omits the run reference for a viewer who cannot read runs", () => { + const state = project({ + recoveryActions: [action()], + pendingReplays: [replay()], + viewerCanReadRuns: false, + }).get("comment-1"); + expect(state?.targetRunId).toBeNull(); + // The recovery action stays available even without a run link. + expect(state?.state).toBe("exhausted_needs_attention"); + }); + + it("withholds retry from a viewer without operator authority", () => { + const state = project({ + recoveryActions: [action()], + viewerCanRetry: false, + }).get("comment-1"); + expect(state?.canRetry).toBe(false); + expect(state?.state).toBe("exhausted_needs_attention"); + }); + + it("ignores recovery actions of other kinds and causes", () => { + const states = project({ + recoveryActions: [ + action({ kind: "stranded_assigned_issue" }), + action({ id: "action-2", cause: "other_cause" }), + ], + }); + expect(states.size).toBe(0); + }); + + it("keeps the highest generation when several replays share a fingerprint", () => { + const state = project({ + pendingReplays: [ + replay({ generation: 0, runId: "run-old" }), + replay({ generation: 1, runId: "run-new" }), + ], + }).get("comment-1"); + expect(state).toMatchObject({ attempt: 1, targetRunId: "run-new" }); + }); + + it("ignores replays that carry no fingerprint", () => { + expect(project({ pendingReplays: [replay({ fingerprint: null })] }).size).toBe(0); + }); +}); + +describe("toSafeFeedbackDeliveryFailureReason", () => { + it("maps known codes to plain operator language", () => { + expect(toSafeFeedbackDeliveryFailureReason("agent_not_invokable")).toBe( + "The assigned agent can't be started right now.", + ); + expect(toSafeFeedbackDeliveryFailureReason(" budget_blocked ")).toBe( + "The agent is over its budget.", + ); + }); + + it("collapses anything unrecognized rather than echoing it", () => { + expect(toSafeFeedbackDeliveryFailureReason("Error: connect ECONNREFUSED 127.0.0.1:5432")).toBeNull(); + expect(toSafeFeedbackDeliveryFailureReason("")).toBeNull(); + expect(toSafeFeedbackDeliveryFailureReason(null)).toBeNull(); + expect(toSafeFeedbackDeliveryFailureReason(undefined)).toBeNull(); + }); + + it("uses no transport or harness jargon in operator copy", () => { + const jargon = ["wake", "adapter", "heartbeat", "pid", "stack"]; + for (const code of ["process_lost", "server_shutdown_interrupted", "agent_not_invokable"]) { + const copy = toSafeFeedbackDeliveryFailureReason(code)?.toLowerCase() ?? ""; + for (const term of jargon) expect(copy).not.toContain(term); + } + }); +}); diff --git a/server/src/services/recovery/feedback-delivery-projection.ts b/server/src/services/recovery/feedback-delivery-projection.ts new file mode 100644 index 00000000000..8a32568bb12 --- /dev/null +++ b/server/src/services/recovery/feedback-delivery-projection.ts @@ -0,0 +1,258 @@ +import type { IssueCommentFeedbackDelivery, IssueRecoveryActionStatus } from "@paperclipai/shared"; + +import { + FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + FEEDBACK_DELIVERY_RECOVERY_CAUSE, +} from "./feedback-delivery.js"; + +/** + * How long a recovered-after-attention notice keeps its own projected state. + * After this the comment falls back to no delivery state at all; the audit + * trail lives in activity history, not in a permanently decorated comment. + */ +export const FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS = 24 * 60 * 60 * 1000; + +/** + * Operator-safe failure vocabulary. Run/gate failure codes are mapped through + * this whitelist so no adapter message, stack frame, request body, credential, + * or filesystem path can reach a comment surface. An unrecognized code is + * deliberately collapsed to `null` rather than echoed: the banner copy already + * states the operator-relevant fact ("Paperclip stopped retrying"). + */ +const SAFE_FAILURE_REASONS: Record = { + agent_not_found: "The assigned agent no longer exists.", + agent_not_invokable: "The assigned agent can't be started right now.", + budget_blocked: "The agent is over its budget.", + issue_paused: "Work on this task is paused.", + issue_dependencies_blocked: "This task is waiting on a blocker.", + issue_assignee_changed: "The assignee changed while delivering.", + issue_reassigned: "The assignee changed while delivering.", + issue_execution_lock_changed: "Another run took over this task.", + issue_review_participant_changed: "The review owner changed while delivering.", + issue_terminal_status: "The task was closed while delivering.", + process_lost: "The agent's process stopped before it read the comment.", + server_shutdown_interrupted: "Paperclip restarted before the agent read the comment.", + workspace_validation_failed: "The task's workspace couldn't be prepared.", + configuration_validation_failed: "Required agent configuration is missing.", +}; + +export function toSafeFeedbackDeliveryFailureReason( + code: string | null | undefined, +): string | null { + if (typeof code !== "string") return null; + const normalized = code.trim(); + if (!normalized) return null; + return SAFE_FAILURE_REASONS[normalized] ?? null; +} + +/** Minimal recovery-action shape this projection needs. */ +export type FeedbackDeliveryRecoveryActionRow = { + id: string; + kind: string; + cause: string; + /** Raw DB column: narrowed against the active-status list rather than cast. */ + status: string; + fingerprint: string; + evidence: unknown; + attemptCount: number; + maxAttempts: number | null; + lastAttemptAt: Date | string | null; + resolvedAt: Date | string | null; + createdAt: Date | string; +}; + +/** + * A replay this issue is currently waiting on: either a queued wake or a + * queued/scheduled run that carries feedback-delivery context. + */ +export type FeedbackDeliveryPendingReplay = { + fingerprint: string | null; + commentIds: readonly string[]; + generation: number; + runId: string | null; + nextAttemptAt: Date | string | null; + requestedAt: Date | string | null; +}; + +export type FeedbackDeliveryProjectionInput = { + recoveryActions: readonly FeedbackDeliveryRecoveryActionRow[]; + pendingReplays: readonly FeedbackDeliveryPendingReplay[]; + /** Whether this viewer may request one more explicit delivery attempt. */ + viewerCanRetry: boolean; + /** Whether this viewer may open run detail. Gates `targetRunId` entirely. */ + viewerCanReadRuns: boolean; + now: Date; + recoveredWindowMs?: number; +}; + +const ACTIVE_STATUSES: readonly string[] = [ + "active", + "escalated", +] satisfies readonly IssueRecoveryActionStatus[]; + +function toIso(value: Date | string | null | undefined): string | null { + if (!value) return null; + const date = value instanceof Date ? value : new Date(value); + return Number.isNaN(date.getTime()) ? null : date.toISOString(); +} + +function toMillis(value: Date | string | null | undefined): number | null { + if (!value) return null; + const date = value instanceof Date ? value : new Date(value); + return Number.isNaN(date.getTime()) ? null : date.getTime(); +} + +function readEvidenceCommentIds(evidence: unknown): string[] { + if (typeof evidence !== "object" || evidence === null) return []; + const raw = (evidence as Record).outstandingCommentIds; + if (!Array.isArray(raw)) return []; + const ids: string[] = []; + for (const entry of raw) { + if (typeof entry !== "string") continue; + const trimmed = entry.trim(); + if (trimmed && !ids.includes(trimmed)) ids.push(trimmed); + } + return ids; +} + +function readEvidenceString(evidence: unknown, key: string): string | null { + if (typeof evidence !== "object" || evidence === null) return null; + const raw = (evidence as Record)[key]; + return typeof raw === "string" && raw.trim().length > 0 ? raw.trim() : null; +} + +function readEvidenceNumber(evidence: unknown, key: string): number | null { + if (typeof evidence !== "object" || evidence === null) return null; + const raw = (evidence as Record)[key]; + return typeof raw === "number" && Number.isFinite(raw) ? raw : null; +} + +function isFeedbackDeliveryAction(action: FeedbackDeliveryRecoveryActionRow) { + return action.kind === "feedback_delivery" && action.cause === FEEDBACK_DELIVERY_RECOVERY_CAUSE; +} + +/** + * Projects feedback-delivery state onto the individual human comments it + * belongs to. + * + * Precedence is deliberate: + * 1. An active/escalated recovery action always wins, because the operator must + * keep seeing the actionable banner until the obligation is discharged — an + * in-flight explicit retry surfaces as `retryInFlight`, not as a state change + * that would make the banner (and its attention item) disappear. + * 2. Otherwise a queued replay renders the quiet `retrying` footer line. + * 3. Otherwise a recently resolved action renders the historical recovered + * notice, so attention disappearing is explained rather than silent. + * + * A newer feedback comment never clears an older obligation: each comment id is + * keyed independently off the evidence of the action that owns it. + */ +export function projectFeedbackDeliveryByComment( + input: FeedbackDeliveryProjectionInput, +): Map { + const byComment = new Map(); + const recoveredWindowMs = input.recoveredWindowMs ?? FEEDBACK_DELIVERY_RECOVERED_WINDOW_MS; + const nowMs = input.now.getTime(); + + const actions = input.recoveryActions.filter(isFeedbackDeliveryAction); + const activeActions = actions.filter((action) => ACTIVE_STATUSES.includes(action.status)); + const activeFingerprints = new Set(activeActions.map((action) => action.fingerprint)); + + // Recovered notices come first so an active action or queued replay for the + // same comment overwrites them, never the other way round. + const resolvedActions = actions + .filter((action) => !ACTIVE_STATUSES.includes(action.status)) + // A fingerprint that resolved and then re-opened is owned by the open one. + .filter((action) => !activeFingerprints.has(action.fingerprint)) + .filter((action) => { + const resolvedMs = toMillis(action.resolvedAt); + return resolvedMs !== null && nowMs - resolvedMs <= recoveredWindowMs; + }) + // Oldest first: the freshest resolution wins a shared comment. + .sort((a, b) => (toMillis(a.resolvedAt) ?? 0) - (toMillis(b.resolvedAt) ?? 0)); + + for (const action of resolvedActions) { + const commentIds = readEvidenceCommentIds(action.evidence); + const targetRunId = input.viewerCanReadRuns + ? readEvidenceString(action.evidence, "sourceRunId") + : null; + for (const commentId of commentIds) { + byComment.set(commentId, { + state: "recovered_after_attention", + sourceCommentId: commentId, + attempt: readEvidenceNumber(action.evidence, "retryGeneration"), + maxAttempts: action.maxAttempts ?? FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + nextAttemptAt: null, + lastAttemptAt: toIso(action.lastAttemptAt), + deliveredAt: toIso(action.resolvedAt), + targetRunId, + canRetry: false, + retryInFlight: false, + safeFailureReason: null, + batchedCommentCount: commentIds.length, + }); + } + } + + const replaysByFingerprint = new Map(); + for (const replay of input.pendingReplays) { + if (!replay.fingerprint) continue; + const existing = replaysByFingerprint.get(replay.fingerprint); + if (!existing || replay.generation > existing.generation) { + replaysByFingerprint.set(replay.fingerprint, replay); + } + } + + for (const replay of replaysByFingerprint.values()) { + if (activeFingerprints.has(replay.fingerprint as string)) continue; + for (const commentId of replay.commentIds) { + byComment.set(commentId, { + state: "retrying", + sourceCommentId: commentId, + attempt: replay.generation, + maxAttempts: FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + nextAttemptAt: toIso(replay.nextAttemptAt), + lastAttemptAt: toIso(replay.requestedAt), + deliveredAt: null, + targetRunId: input.viewerCanReadRuns ? replay.runId : null, + canRetry: false, + retryInFlight: true, + safeFailureReason: null, + batchedCommentCount: replay.commentIds.length, + }); + } + } + + for (const action of activeActions) { + const commentIds = readEvidenceCommentIds(action.evidence); + const replay = replaysByFingerprint.get(action.fingerprint) ?? null; + const retryInFlight = replay !== null; + const targetRunId = input.viewerCanReadRuns + ? replay?.runId ?? readEvidenceString(action.evidence, "sourceRunId") + : null; + const safeFailureReason = + toSafeFeedbackDeliveryFailureReason(readEvidenceString(action.evidence, "lastFailureCode")) + ?? toSafeFeedbackDeliveryFailureReason(readEvidenceString(action.evidence, "gateErrorCode")); + for (const commentId of commentIds) { + byComment.set(commentId, { + state: "exhausted_needs_attention", + sourceCommentId: commentId, + attempt: readEvidenceNumber(action.evidence, "retryGeneration"), + maxAttempts: + readEvidenceNumber(action.evidence, "maxAutomaticRetries") + ?? action.maxAttempts + ?? FEEDBACK_DELIVERY_MAX_AUTOMATIC_RETRIES, + nextAttemptAt: toIso(replay?.nextAttemptAt ?? null), + lastAttemptAt: toIso(action.lastAttemptAt ?? action.createdAt), + deliveredAt: null, + targetRunId, + canRetry: input.viewerCanRetry && !retryInFlight, + retryInFlight, + safeFailureReason, + batchedCommentCount: commentIds.length, + }); + } + } + + return byComment; +} 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..cbb75aefa46 --- /dev/null +++ b/server/src/services/recovery/feedback-delivery.ts @@ -0,0 +1,362 @@ +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}`; +} + +/** + * Explicit operator "Retry delivery" replays get their own key namespace: the + * automatic `:generation:N` keys are already spent by the time delivery is + * exhausted, so reusing them would silently swallow the operator's request. + * Keying on the recovery action's attempt count still collapses a double-click + * into one replay while allowing a fresh attempt after a failure. + */ +export function buildFeedbackDeliveryManualRetryIdempotencyKey(input: { + fingerprint: string; + attempt: number; +}) { + return `${input.fingerprint}:manual:${input.attempt}`; +} + +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;