diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 3954ccb0..414aad1c 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -3418,6 +3418,78 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(retryRun?.contextSnapshot as Record).not.toHaveProperty("modelProfile"); }); + it("exempts stranded-recovery escalation when assignee posted a recent comment (GGU-809)", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + retryReason: "issue_continuation_needed", + runSource: "issue.productive_terminal_continuation_recovery", + livenessState: "advanced", + }); + // Recent agent-authored comment should suppress the repeat-productive + // escalation and let the normal continuation-retry path proceed. + await db.insert(issueComments).values({ + companyId, + issueId, + authorAgentId: agentId, + body: "frame 02/08 generated, attaching shortly", + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.escalated).toBe(0); + expect(result.recentProgressExempted).toBe(1); + expect(result.continuationRequeued).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("in_progress"); + + const recoveryIssues = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"))); + expect(recoveryIssues).toHaveLength(0); + + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(2); + const retryRun = runs.find((row) => row.id !== runId); + expect(retryRun?.contextSnapshot as Record | undefined).toMatchObject({ + issueId, + retryReason: "issue_continuation_needed", + source: "issue.productive_terminal_continuation_recovery", + }); + }); + + it("still escalates stranded-recovery work when the recent comment is older than the exemption window (GGU-809)", async () => { + const { companyId, agentId, issueId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + retryReason: "issue_continuation_needed", + runSource: "issue.productive_terminal_continuation_recovery", + livenessState: "advanced", + }); + // Comment older than the exemption window must NOT suppress escalation. + const stale = new Date(Date.now() - 24 * 60 * 60 * 1000); + await db.insert(issueComments).values({ + companyId, + issueId, + authorAgentId: agentId, + body: "old progress note", + createdAt: stale, + updatedAt: stale, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.escalated).toBe(1); + expect(result.recentProgressExempted).toBe(0); + expect(result.continuationRequeued).toBe(0); + + const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("blocked"); + }); + it("does not reconcile user-assigned work through the agent stranded-work recovery path", async () => { const { issueId, runId } = await seedStrandedIssueFixture({ status: "todo", diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 1a5fb8c5..5b93a7b8 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -16,6 +16,7 @@ import { heartbeatRunEvents, heartbeatRunWatchdogDecisions, heartbeatRuns, + issueAttachments, issueComments, issueApprovals, issueRecoveryActions, @@ -81,6 +82,18 @@ const SESSIONED_LOCAL_ADAPTERS = new Set([ "pi_local", ]); +// GGU-809: when a stranded `in_progress` issue would otherwise hit the +// `isRepeatedProductiveContinuationRecovery` escalation path, exempt the +// escalation if the assignee posted a comment or attachment within this window. +// Batch workflows (e.g. Image Spec multi-frame generation) make real progress +// every heartbeat and would otherwise trigger a recovery issue after just two +// productive heartbeats. Floor the override at 60s to keep the exemption from +// being effectively disabled by misconfiguration. +export const STRANDED_RECENT_PROGRESS_EXEMPTION_MS = Math.max( + 60_000, + Number(process.env.STRANDED_RECENT_PROGRESS_EXEMPTION_MS) || 30 * 60 * 1000, +); + type RecoveryWakeupOptions = { source?: "timer" | "assignment" | "on_demand" | "automation"; triggerDetail?: "manual" | "ping" | "callback" | "system"; @@ -584,6 +597,47 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) .then((rows) => Boolean(rows[0])); } + // GGU-809: visible-progress signal for stranded-recovery escalation guard. + // Returns true if the assignee posted a comment, OR any attachment was added + // to the issue, within `windowMs`. Used to suppress false-positive recovery + // issues for batch workflows that genuinely advance every heartbeat. + async function hasRecentVisibleProgress( + companyId: string, + issueId: string, + assigneeAgentId: string, + windowMs: number, + ) { + const since = new Date(Date.now() - windowMs); + const [comment, attachment] = await Promise.all([ + db + .select({ id: issueComments.id }) + .from(issueComments) + .where( + and( + eq(issueComments.companyId, companyId), + eq(issueComments.issueId, issueId), + eq(issueComments.authorAgentId, assigneeAgentId), + gt(issueComments.createdAt, since), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: issueAttachments.id }) + .from(issueAttachments) + .where( + and( + eq(issueAttachments.companyId, companyId), + eq(issueAttachments.issueId, issueId), + gt(issueAttachments.createdAt, since), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + return Boolean(comment || attachment); + } + async function enqueueStrandedIssueRecovery(input: { issueId: string; agentId: string; @@ -2475,6 +2529,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) orphanBlockersAssigned: 0, successfulRunHandoffEscalated: 0, escalated: 0, + recentProgressExempted: 0, skipped: 0, issueIds: [] as string[], }; @@ -2623,21 +2678,34 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } if (isRepeatedProductiveContinuationRecovery(successfulRun)) { - const updated = await escalateStrandedAssignedIssue({ - issue, - previousStatus: "in_progress", - latestRun: successfulRun, - comment: - "Paperclip automatically retried continuation for this assigned `in_progress` issue and the retry " + - "made progress, but it still has no live execution path. Moving it to `blocked` so it is visible for intervention.", - }); - if (updated) { - result.escalated += 1; - result.issueIds.push(issue.id); - } else { - result.skipped += 1; + // GGU-809: skip escalation if the assignee has shown visible progress + // (comment or attachment) within the exemption window. Falling + // through here lets the normal continuation-retry path enqueue the + // next wake, which is the correct behaviour for batch workflows. + const exempted = await hasRecentVisibleProgress( + issue.companyId, + issue.id, + agentId, + STRANDED_RECENT_PROGRESS_EXEMPTION_MS, + ); + if (!exempted) { + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "in_progress", + latestRun: successfulRun, + comment: + "Paperclip automatically retried continuation for this assigned `in_progress` issue and the retry " + + "made progress, but it still has no live execution path. Moving it to `blocked` so it is visible for intervention.", + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; } - continue; + result.recentProgressExempted += 1; } if (await isInvocationBudgetBlocked(issue, agentId)) {