diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index e95910b8f0..e78fccfcbf 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -1279,6 +1279,7 @@ export { requestConfirmationCustomTargetSchema, requestConfirmationTargetSchema, requestConfirmationPayloadSchema, + requestConfirmationResumeFailureSchema, requestConfirmationResultSchema, requestCheckboxConfirmationOptionSchema, requestCheckboxConfirmationPayloadSchema, diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index 4a792e0093..f651e482df 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -1088,6 +1088,16 @@ export interface RequestConfirmationResult { reason?: string | null; commentId?: string | null; staleTarget?: RequestConfirmationTarget | null; + resumeFailure?: { + status: "retrying" | "needs_attention"; + errorCode: string | null; + attempt: number; + maxAttempts: number; + runId?: string | null; + retryRunId?: string | null; + recoveryActionId?: string | null; + updatedAt?: string | null; + } | null; } export interface RequestCheckboxConfirmationResult extends RequestConfirmationResult { diff --git a/packages/shared/src/validators/index.ts b/packages/shared/src/validators/index.ts index eb610f933a..8b14b7e62c 100644 --- a/packages/shared/src/validators/index.ts +++ b/packages/shared/src/validators/index.ts @@ -359,6 +359,7 @@ export { requestConfirmationCustomTargetSchema, requestConfirmationTargetSchema, requestConfirmationPayloadSchema, + requestConfirmationResumeFailureSchema, requestConfirmationResultSchema, requestCheckboxConfirmationOptionSchema, requestCheckboxConfirmationPayloadSchema, diff --git a/packages/shared/src/validators/issue.ts b/packages/shared/src/validators/issue.ts index 0b94aa8aee..a89116a38b 100644 --- a/packages/shared/src/validators/issue.ts +++ b/packages/shared/src/validators/issue.ts @@ -855,12 +855,24 @@ export const requestCheckboxConfirmationPayloadSchema = z.object({ } }); +export const requestConfirmationResumeFailureSchema = z.object({ + status: z.enum(["retrying", "needs_attention"]), + errorCode: z.string().trim().min(1).max(120).nullable(), + attempt: z.number().int().min(0).max(100), + maxAttempts: z.number().int().min(0).max(100), + runId: z.string().uuid().nullable().optional(), + retryRunId: z.string().uuid().nullable().optional(), + recoveryActionId: z.string().uuid().nullable().optional(), + updatedAt: z.string().trim().min(1).nullable().optional(), +}); + export const requestConfirmationResultSchema = z.object({ version: z.literal(1), outcome: z.enum(["accepted", "rejected", "superseded_by_comment", "stale_target"]), reason: z.string().trim().max(4000).nullable().optional(), commentId: z.string().uuid().nullable().optional(), staleTarget: requestConfirmationTargetSchema.nullable().optional(), + resumeFailure: requestConfirmationResumeFailureSchema.nullable().optional(), }); export const requestCheckboxConfirmationResultSchema = requestConfirmationResultSchema.extend({ diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index d0af367378..db068b1d7f 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -95,6 +95,8 @@ vi.mock("../adapters/index.ts", async () => { }); import { + INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + INTERACTION_CONTINUATION_INFRA_WAKE_REASON, heartbeatService, redactDetectedSuccessfulRunProgressSummaryForBoard, } from "../services/heartbeat.ts"; @@ -1812,7 +1814,408 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(comments).toHaveLength(0); }); + it("schedules bounded retries for failed accepted interaction continuation wakes", async () => { + const { companyId, agentId, runId, wakeupRequestId, issueId } = await seedQueuedIssueRunFixture(); + const interactionId = randomUUID(); + + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt: new Date("2026-03-19T00:00:00.000Z"), + payload: { + version: 1, + prompt: "Approve the plan?", + target: { + type: "issue_document", + issueId, + key: "plan", + revisionId: randomUUID(), + }, + }, + result: { version: 1, outcome: "accepted" }, + }); + + await db + .update(agentWakeupRequests) + .set({ + source: "automation", + reason: "issue_commented", + payload: { + issueId, + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + mutation: "interaction", + }, + }) + .where(eq(agentWakeupRequests.id, wakeupRequestId)); + await db + .update(heartbeatRuns) + .set({ + invocationSource: "automation", + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + await db + .update(issues) + .set({ status: "in_review" }) + .where(eq(issues.id, issueId)); + + mockAdapterExecute.mockRejectedValueOnce( + new Error('Failed to start command "codex" in "/workspace". Verify adapter command, working directory, and PATH.'), + ); + + const heartbeat = heartbeatService(db); + await heartbeat.resumeQueuedRuns(); + + const runs = await waitForValue(async () => { + const rows = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + return rows.length >= 2 ? rows : null; + }); + expect(runs).toHaveLength(2); + + const failedRun = runs?.find((row) => row.id === runId); + const retryRun = runs?.find((row) => row.id !== runId); + expect(failedRun).toMatchObject({ + status: "failed", + errorCode: "adapter_failed", + }); + expect(retryRun).toMatchObject({ + status: "scheduled_retry", + retryOfRunId: runId, + scheduledRetryAttempt: 1, + scheduledRetryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + }); + expect(retryRun?.contextSnapshot).toMatchObject({ + issueId, + interactionId, + interactionStatus: "accepted", + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + scheduledRetryAttempt: 1, + }); + + const wakeups = await db + .select({ + id: agentWakeupRequests.id, + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + runId: agentWakeupRequests.runId, + payload: agentWakeupRequests.payload, + }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakeups.find((row) => row.id === wakeupRequestId)).toMatchObject({ + status: "failed", + reason: "issue_commented", + runId, + }); + expect(wakeups.find((row) => row.runId === retryRun?.id)).toMatchObject({ + status: "queued", + reason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + payload: expect.objectContaining({ + issueId, + interactionId, + retryOfRunId: runId, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + scheduledRetryAttempt: 1, + }), + }); + + const issue = await db + .select({ status: issues.status, executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue).toEqual({ + status: "in_review", + executionRunId: retryRun?.id ?? null, + }); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]).toMatchObject({ + authorType: "system", + createdByRunId: runId, + body: "Agent failed to resume after approval: `adapter_failed` — retrying (attempt 1/3)", + }); + + const interaction = await db + .select({ result: issueThreadInteractions.result }) + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, interactionId)) + .then((rows) => rows[0] ?? null); + expect(interaction?.result).toMatchObject({ + version: 1, + outcome: "accepted", + resumeFailure: { + status: "retrying", + errorCode: "adapter_failed", + attempt: 1, + maxAttempts: 3, + runId, + retryRunId: retryRun?.id ?? null, + }, + }); + mockAdapterExecute.mockClear(); + }); + + it("escalates exhausted plan approval resume failures with a system comment and recovery action", async () => { + const { companyId, agentId, runId, issueId } = await seedQueuedIssueRunFixture(); + const interactionId = randomUUID(); + + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt: new Date("2026-03-19T00:00:00.000Z"), + payload: { + version: 1, + prompt: "Approve the plan?", + target: { + type: "issue_document", + issueId, + key: "plan", + revisionId: randomUUID(), + }, + }, + result: { version: 1, outcome: "accepted" }, + }); + await db + .update(heartbeatRuns) + .set({ + status: "failed", + error: "Failed to start command", + errorCode: "adapter_failed", + scheduledRetryAttempt: 3, + scheduledRetryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + finishedAt: new Date("2026-03-19T00:10:00.000Z"), + }) + .where(eq(heartbeatRuns.id, runId)); + await db + .update(issues) + .set({ status: "in_review", executionRunId: runId }) + .where(eq(issues.id, issueId)); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.scheduleBoundedRetry(runId, { + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }); + + expect(result).toMatchObject({ + outcome: "retry_exhausted", + maxAttempts: 3, + }); + + const issue = await db + .select({ status: issues.status }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue?.status).toBe("blocked"); + + const recoveryAction = await db + .select({ id: issueRecoveryActions.id, status: issueRecoveryActions.status, sourceIssueId: issueRecoveryActions.sourceIssueId }) + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)) + .then((rows) => rows[0] ?? null); + expect(recoveryAction).toMatchObject({ + status: "active", + sourceIssueId: issueId, + }); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]).toMatchObject({ + authorType: "system", + body: expect.stringContaining("Agent failed to resume after approval: `adapter_failed` — needs attention"), + }); + expect(comments[0]?.body).toContain("Recovery action:"); + + const interaction = await db + .select({ result: issueThreadInteractions.result }) + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, interactionId)) + .then((rows) => rows[0] ?? null); + expect(interaction?.result).toMatchObject({ + version: 1, + outcome: "accepted", + resumeFailure: { + status: "needs_attention", + errorCode: "adapter_failed", + attempt: 3, + maxAttempts: 3, + runId, + recoveryActionId: recoveryAction?.id ?? null, + }, + }); + }); + + // Scenario 4: `process_lost` before the agent started is retried like + // other infrastructure failures. Distinct from the pid-based process-loss retry + // ("queues exactly one retry when the recorded local pid is dead"): here no pid was ever + // recorded (the process died before producing output), so the reaper falls through to the + // accepted-interaction infra-retry path. Pre-P1 `process_lost` was not retry-eligible there. + it("retries a plan-approval continuation lost as process_lost before agent start as an infrastructure failure", async () => { + const { companyId, agentId, runId, wakeupRequestId, issueId } = await seedQueuedIssueRunFixture(); + const interactionId = randomUUID(); + + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt: new Date("2026-03-19T00:00:00.000Z"), + payload: { + version: 1, + prompt: "Approve the plan?", + target: { type: "issue_document", issueId, key: "plan", revisionId: randomUUID() }, + }, + result: { version: 1, outcome: "accepted" }, + }); + + // The continuation wake was claimed and a run spawned, but the process was lost before + // the agent produced any output — no pid/process-group was ever recorded. + await db + .update(agentWakeupRequests) + .set({ + source: "automation", + reason: "issue_commented", + status: "claimed", + payload: { + issueId, + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + mutation: "interaction", + }, + }) + .where(eq(agentWakeupRequests.id, wakeupRequestId)); + await db + .update(heartbeatRuns) + .set({ + status: "running", + invocationSource: "automation", + processPid: null, + processGroupId: null, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + startedAt: new Date("2026-03-19T00:00:00.000Z"), + updatedAt: new Date("2026-03-19T00:00:00.000Z"), + }) + .where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, issueId)); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reapOrphanedRuns(); + expect(result.reaped).toBe(1); + expect(result.runIds).toEqual([runId]); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(2); + + const failedRun = runs.find((row) => row.id === runId); + const retryRun = runs.find((row) => row.id !== runId); + expect(failedRun).toMatchObject({ status: "failed", errorCode: "process_lost" }); + expect(retryRun).toMatchObject({ + status: "scheduled_retry", + retryOfRunId: runId, + scheduledRetryAttempt: 1, + scheduledRetryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + }); + expect(retryRun?.contextSnapshot).toMatchObject({ + issueId, + interactionId, + interactionStatus: "accepted", + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + scheduledRetryAttempt: 1, + }); + + const retryWakeup = await db + .select({ reason: agentWakeupRequests.reason, status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.runId, retryRun?.id ?? "")) + .then((rows) => rows[0] ?? null); + expect(retryWakeup?.reason).toBe(INTERACTION_CONTINUATION_INFRA_WAKE_REASON); + + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(1); + expect(comments[0]).toMatchObject({ + authorType: "system", + body: "Agent failed to resume after approval: `process_lost` — retrying (attempt 1/3)", + }); + + const interaction = await db + .select({ result: issueThreadInteractions.result }) + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, interactionId)) + .then((rows) => rows[0] ?? null); + expect(interaction?.result).toMatchObject({ + version: 1, + outcome: "accepted", + resumeFailure: { + status: "retrying", + errorCode: "process_lost", + attempt: 1, + maxAttempts: 3, + runId, + retryRunId: retryRun?.id ?? null, + }, + }); + mockAdapterExecute.mockClear(); + }); + it("blocks a git-sensitive local adapter before launch when a project-workspace-linked issue is missing its project id", async () => { + mockAdapterExecute.mockClear(); const { companyId, agentId, runId, issueId } = await seedQueuedIssueRunFixture(); const projectId = randomUUID(); const projectWorkspaceId = randomUUID(); @@ -3489,6 +3892,361 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(issue?.status).toBe("in_progress"); }); + it("requeues accepted interaction continuations stranded in_review without execution state", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const interactionId = randomUUID(); + const resolvedAt = new Date("2026-03-19T00:05:00.000Z"); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + defaultResponsibleUserId: "responsible-user", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Accepted plan never resumed", + status: "in_review", + priority: "medium", + assigneeAgentId: agentId, + responsibleUserId: "responsible-user", + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt, + updatedAt: resolvedAt, + payload: { version: 1, prompt: "Approve the plan?" }, + result: { outcome: "accepted" }, + }); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reconcileStrandedAssignedIssues(); + + expect(result.continuationRequeued).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const run = await db + .select({ + agentId: heartbeatRuns.agentId, + contextSnapshot: heartbeatRuns.contextSnapshot, + retryOfRunId: heartbeatRuns.retryOfRunId, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)) + .then((rows) => rows[0] ?? null); + expect(run?.agentId).toBe(agentId); + expect(run?.retryOfRunId).toBeNull(); + expect(run?.contextSnapshot).toMatchObject({ + issueId, + taskId: issueId, + wakeReason: "issue_continuation_needed", + retryReason: "issue_continuation_needed", + source: "issue.interaction_continuation_recovery", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + interactionContinuationPolicy: "wake_assignee_on_accept", + interactionResolvedAt: resolvedAt.toISOString(), + }); + }); + + it("requeues accepted interaction continuations even when a later successful run is unrelated", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const interactionId = randomUUID(); + const resolvedAt = new Date("2026-03-19T00:05:00.000Z"); + const unrelatedRunAt = new Date("2026-03-19T00:06:00.000Z"); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + defaultResponsibleUserId: "responsible-user", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Accepted plan masked by unrelated run", + status: "in_review", + priority: "medium", + assigneeAgentId: agentId, + responsibleUserId: "responsible-user", + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt, + updatedAt: resolvedAt, + payload: { version: 1, prompt: "Approve the plan?" }, + result: { outcome: "accepted" }, + }); + await db.insert(heartbeatRuns).values({ + id: randomUUID(), + companyId, + agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "succeeded", + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_assigned", + source: "unrelated_followup", + }, + startedAt: unrelatedRunAt, + finishedAt: unrelatedRunAt, + createdAt: unrelatedRunAt, + updatedAt: unrelatedRunAt, + }); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reconcileStrandedAssignedIssues(); + + expect(result.continuationRequeued).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const runs = await db + .select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + const recoveryRun = runs.find( + (row) => (row.contextSnapshot as Record | null)?.source === "issue.interaction_continuation_recovery", + ); + expect(recoveryRun?.contextSnapshot).toMatchObject({ + issueId, + interactionId, + interactionStatus: "accepted", + source: "issue.interaction_continuation_recovery", + }); + }); + + // Scenario 5: enqueue-failure at accept time is no longer a silent permanent + // stall. When the accept-time continuation wake is dropped (routes/issues.ts fire-and-forget + // enqueue swallowed the error), the issue is left in_review with an accepted interaction but + // *no* wake request and *no* run at all. Pre-P1 the recovery sweep skipped in_review issues + // lacking an execution policy, so this limbo persisted forever. The sweep now requeues it. + it("recovers a plan approval whose accept-time continuation wake enqueue was silently dropped", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const interactionId = randomUUID(); + const resolvedAt = new Date("2026-03-19T00:05:00.000Z"); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + defaultResponsibleUserId: "responsible-user", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "idle", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Approved plan whose wake enqueue was dropped", + status: "in_review", + priority: "medium", + assigneeAgentId: agentId, + responsibleUserId: "responsible-user", + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt, + updatedAt: resolvedAt, + payload: { version: 1, prompt: "Approve the plan?" }, + result: { outcome: "accepted" }, + }); + + // Precondition of the silent-enqueue-drop bug: the accept produced no wake and no run. + const priorWakeups = await db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.companyId, companyId)); + expect(priorWakeups).toHaveLength(0); + const priorRuns = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.companyId, companyId)); + expect(priorRuns).toHaveLength(0); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reconcileStrandedAssignedIssues(); + + expect(result.continuationRequeued).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const run = await db + .select({ agentId: heartbeatRuns.agentId, contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)) + .then((rows) => rows[0] ?? null); + expect(run?.agentId).toBe(agentId); + expect(run?.contextSnapshot).toMatchObject({ + issueId, + interactionId, + interactionStatus: "accepted", + source: "issue.interaction_continuation_recovery", + }); + + const wakeup = await db + .select({ payload: agentWakeupRequests.payload }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, agentId)) + .then((rows) => rows[0] ?? null); + expect(wakeup).not.toBeNull(); + expect((wakeup?.payload as Record | null)?.issueId).toBe(issueId); + }); + + // Scenario 3 (restart durability): a bounded continuation retry scheduled + // before a server restart survives it. Promotion is DB-driven (scheduled_retry rows + + // promoteDueScheduledRetries), not an in-memory setTimeout — so a brand-new heartbeat + // service instance with empty in-memory state still promotes the due retry. + it("promotes a scheduled plan-approval continuation retry after a simulated server restart", async () => { + const { companyId, agentId, runId, issueId } = await seedQueuedIssueRunFixture(); + const interactionId = randomUUID(); + const now = new Date("2026-03-19T00:10:00.000Z"); + + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId, + kind: "request_confirmation", + status: "accepted", + continuationPolicy: "wake_assignee_on_accept", + createdByAgentId: agentId, + resolvedByUserId: "responsible-user", + resolvedAt: now, + payload: { + version: 1, + prompt: "Approve the plan?", + target: { type: "issue_document", issueId, key: "plan", revisionId: randomUUID() }, + }, + result: { version: 1, outcome: "accepted" }, + }); + await db + .update(heartbeatRuns) + .set({ + status: "failed", + error: "workspace validation failed before dispatch", + errorCode: "workspace_validation_failed", + resultJson: {}, + finishedAt: now, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + await db + .update(issues) + .set({ status: "in_review", executionRunId: runId }) + .where(eq(issues.id, issueId)); + + // Service instance that scheduled the retry (pre-restart). + const preRestart = heartbeatService(db); + const scheduled = await preRestart.scheduleBoundedRetry(runId, { + now, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }); + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + + const beforePromotion = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .then((rows) => rows[0] ?? null); + expect(beforePromotion?.status).toBe("scheduled_retry"); + + // Simulate a server restart: no in-memory process/timer state carries over. + runningProcesses.clear(); + const restarted = heartbeatService(db); + const promotion = await restarted.promoteDueScheduledRetries(scheduled.dueAt); + expect(promotion).toEqual({ promoted: 1, runIds: [scheduled.run.id] }); + + const promoted = await db + .select({ status: heartbeatRuns.status, retryOfRunId: heartbeatRuns.retryOfRunId }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .then((rows) => rows[0] ?? null); + expect(promoted).toMatchObject({ status: "queued", retryOfRunId: runId }); + }); + it("still re-enqueues stranded assigned todo recovery when an old queued wake exists", async () => { const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "todo", diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 26e9fd664d..3dada976b5 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { and, eq, sql } from "drizzle-orm"; +import { and, eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { agents, @@ -11,10 +11,12 @@ import { companySkills, createDb, environmentLeases, + executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueRelations, issues, + projects, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, @@ -23,6 +25,8 @@ import { import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts"; import { BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS, + INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + INTERACTION_CONTINUATION_INFRA_WAKE_REASON, MAX_TURN_CONTINUATION_RETRY_REASON, MAX_TURN_CONTINUATION_WAKE_REASON, heartbeatService, @@ -92,6 +96,8 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { await db.delete(environmentLeases); await db.delete(issueRelations); await db.delete(issues); + await db.delete(executionWorkspaces); + await db.delete(projects); await db.delete(activityLog); await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); @@ -496,6 +502,641 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { }); }); + it("schedules accepted interaction continuation infra retries while the issue is in_review", async () => { + const { issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" }); + const interactionId = randomUUID(); + + await db + .update(heartbeatRuns) + .set({ + error: "workspace validation failed before dispatch", + errorCode: "workspace_validation_failed", + resultJson: {}, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const scheduled = await heartbeat.scheduleBoundedRetry(runId, { + now, + random: () => 0.5, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }); + + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + expect(scheduled.attempt).toBe(1); + expect(scheduled.maxAttempts).toBe(3); + + const retryRun = await db + .select({ + retryOfRunId: heartbeatRuns.retryOfRunId, + status: heartbeatRuns.status, + scheduledRetryAttempt: heartbeatRuns.scheduledRetryAttempt, + scheduledRetryReason: heartbeatRuns.scheduledRetryReason, + contextSnapshot: heartbeatRuns.contextSnapshot, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .then((rows) => rows[0] ?? null); + + expect(retryRun).toMatchObject({ + retryOfRunId: runId, + status: "scheduled_retry", + scheduledRetryAttempt: 1, + scheduledRetryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + }); + expect(retryRun?.contextSnapshot).toMatchObject({ + issueId, + interactionId, + interactionStatus: "accepted", + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + scheduledRetryAttempt: 1, + }); + + const wakeupRequest = await db + .select({ reason: agentWakeupRequests.reason, payload: agentWakeupRequests.payload }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, retryRun?.wakeupRequestId ?? "")) + .then((rows) => rows[0] ?? null); + expect(wakeupRequest?.reason).toBe(INTERACTION_CONTINUATION_INFRA_WAKE_REASON); + expect(wakeupRequest?.payload).toMatchObject({ + issueId, + interactionId, + retryOfRunId: runId, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + scheduledRetryAttempt: 1, + }); + + const issue = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue?.executionRunId).toBe(scheduled.run.id); + }); + + it("coalesces duplicate accepted interaction continuation infra retry schedules", async () => { + const { issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" }); + const interactionId = randomUUID(); + await db + .update(heartbeatRuns) + .set({ + error: "workspace validation failed before dispatch", + errorCode: "workspace_validation_failed", + resultJson: {}, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const retryOptions = { + now, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }; + const [first, second] = await Promise.all([ + heartbeat.scheduleBoundedRetry(runId, retryOptions), + heartbeat.scheduleBoundedRetry(runId, retryOptions), + ]); + + expect(first.outcome).toBe("scheduled"); + expect(second.outcome).toBe("scheduled"); + if (first.outcome !== "scheduled" || second.outcome !== "scheduled") return; + expect(new Set([first.run.id, second.run.id]).size).toBe(1); + + const retryRuns = await db + .select({ id: heartbeatRuns.id, wakeupRequestId: heartbeatRuns.wakeupRequestId }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.retryOfRunId, runId), + eq(heartbeatRuns.scheduledRetryReason, INTERACTION_CONTINUATION_INFRA_RETRY_REASON), + eq(heartbeatRuns.scheduledRetryAttempt, 1), + )); + expect(retryRuns).toHaveLength(1); + + const wakeups = await db + .select({ + id: agentWakeupRequests.id, + coalescedCount: agentWakeupRequests.coalescedCount, + idempotencyKey: agentWakeupRequests.idempotencyKey, + }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.reason, INTERACTION_CONTINUATION_INFRA_WAKE_REASON)); + expect(wakeups).toHaveLength(1); + expect(wakeups[0]).toMatchObject({ + id: retryRuns[0]?.wakeupRequestId, + coalescedCount: 1, + }); + expect(wakeups[0]?.idempotencyKey).toContain(`:${issueId}:${runId}:1`); + }); + + it.each([ + { + name: "renamed branch", + workspaceValidation: (workspaceId: string) => ({ + reason: "git_worktree_branch_incoherence", + fingerprint: "workspace_incoherence:v1:sha256:renamed", + executionWorkspaceId: workspaceId, + expectedBranch: "stale-plan-approval-workspace", + actualBranch: "feat/skill-studio-test-runs", + cleanliness: "clean", + }), + }, + { + name: "dirty worktree", + workspaceValidation: (workspaceId: string) => ({ + reason: "git_worktree_branch_incoherence", + fingerprint: "workspace_incoherence:v1:sha256:dirty", + executionWorkspaceId: workspaceId, + expectedBranch: "stale-plan-approval-workspace", + actualBranch: "feat/skill-studio-test-runs", + cleanliness: "dirty", + safeRepair: { + eligible: false, + attempted: false, + succeeded: false, + reason: "worktree is not clean", + }, + }), + }, + ])("quarantines a failed $name workspace before scheduling the accepted interaction retry", async ({ workspaceValidation }) => { + const { companyId, agentId, issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" }); + const projectId = randomUUID(); + const executionWorkspaceId = randomUUID(); + const validation = workspaceValidation(executionWorkspaceId); + + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Paperclip App", + status: "in_progress", + }); + await db.insert(executionWorkspaces).values({ + id: executionWorkspaceId, + companyId, + projectId, + sourceIssueId: issueId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "stale-plan-approval-workspace", + status: "active", + cwd: "/workspace/stale-plan-approval-workspace", + baseRef: "origin/master", + branchName: "stale-plan-approval-workspace", + providerType: "git_worktree", + providerRef: "/workspace/stale-plan-approval-workspace", + metadata: { existing: true }, + }); + await db + .update(issues) + .set({ + projectId, + executionWorkspaceId, + executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "isolated_workspace" }, + }) + .where(eq(issues.id, issueId)); + + const interactionId = randomUUID(); + await db + .update(heartbeatRuns) + .set({ + error: "workspace validation failed before dispatch", + errorCode: "workspace_validation_failed", + resultJson: { workspaceValidation: validation }, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const scheduled = await heartbeat.scheduleBoundedRetry(runId, { + now, + random: () => 0.5, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }); + + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + + const issue = await db + .select({ + executionRunId: issues.executionRunId, + executionWorkspaceId: issues.executionWorkspaceId, + executionWorkspacePreference: issues.executionWorkspacePreference, + executionWorkspaceSettings: issues.executionWorkspaceSettings, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue).toMatchObject({ + executionRunId: scheduled.run.id, + executionWorkspaceId: null, + executionWorkspacePreference: null, + executionWorkspaceSettings: { mode: "isolated_workspace" }, + }); + + const workspace = await db + .select({ + status: executionWorkspaces.status, + closedAt: executionWorkspaces.closedAt, + cleanupEligibleAt: executionWorkspaces.cleanupEligibleAt, + cleanupReason: executionWorkspaces.cleanupReason, + metadata: executionWorkspaces.metadata, + }) + .from(executionWorkspaces) + .where(eq(executionWorkspaces.id, executionWorkspaceId)) + .then((rows) => rows[0] ?? null); + expect(workspace).toMatchObject({ + status: "archived", + cleanupEligibleAt: null, + cleanupReason: "workspace_validation_failed", + }); + expect(workspace?.closedAt?.toISOString()).toBe(now.toISOString()); + expect(workspace?.metadata).toMatchObject({ + existing: true, + workspaceValidationQuarantine: { + reason: "workspace_validation_failed", + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + sourceRunId: runId, + retryRunId: scheduled.run.id, + issueId, + sourceIssueId: issueId, + workspaceValidation: validation, + }, + }); + + const retryRun = await db + .select({ contextSnapshot: heartbeatRuns.contextSnapshot }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .then((rows) => rows[0] ?? null); + expect(retryRun?.contextSnapshot).toMatchObject({ + workspaceValidationRecovery: { + strategy: "quarantine_failed_workspace_and_retry_clean", + sourceRunId: runId, + reason: "git_worktree_branch_incoherence", + fingerprint: validation.fingerprint, + failedExecutionWorkspaceId: executionWorkspaceId, + }, + }); + + const activity = await db + .select({ action: activityLog.action, entityId: activityLog.entityId, details: activityLog.details }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "execution_workspace.workspace_validation_quarantined"), + )) + .then((rows) => rows[0] ?? null); + expect(activity).toMatchObject({ + action: "execution_workspace.workspace_validation_quarantined", + entityId: executionWorkspaceId, + details: expect.objectContaining({ + retryRunId: scheduled.run.id, + workspaceValidation: validation, + }), + }); + + const agent = await db + .select({ id: agents.id }) + .from(agents) + .where(eq(agents.id, agentId)) + .then((rows) => rows[0] ?? null); + expect(agent?.id).toBe(agentId); + }); + + it("does not quarantine another issue's workspace when validation payload is stale", async () => { + const { companyId, issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" }); + const projectId = randomUUID(); + const currentWorkspaceId = randomUUID(); + const foreignIssueId = randomUUID(); + const foreignWorkspaceId = randomUUID(); + const issuePrefix = `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + const validation = { + reason: "git_worktree_branch_incoherence", + fingerprint: "workspace_incoherence:v1:sha256:stale", + executionWorkspaceId: foreignWorkspaceId, + expectedBranch: "current-issue-branch", + actualBranch: "foreign-issue-branch", + cleanliness: "clean", + }; + + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Paperclip App", + status: "in_progress", + }); + await db.insert(issues).values({ + id: foreignIssueId, + companyId, + title: "Other active issue", + status: "in_progress", + priority: "medium", + responsibleUserId: "responsible-user", + issueNumber: 2, + identifier: `${issuePrefix}-2`, + }); + await db.insert(executionWorkspaces).values([ + { + id: currentWorkspaceId, + companyId, + projectId, + sourceIssueId: issueId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "current-issue-branch", + status: "active", + cwd: "/workspace/current-issue-branch", + baseRef: "origin/master", + branchName: "current-issue-branch", + providerType: "git_worktree", + providerRef: "/workspace/current-issue-branch", + metadata: { current: true }, + }, + { + id: foreignWorkspaceId, + companyId, + projectId, + sourceIssueId: foreignIssueId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "foreign-issue-branch", + status: "active", + cwd: "/workspace/foreign-issue-branch", + baseRef: "origin/master", + branchName: "foreign-issue-branch", + providerType: "git_worktree", + providerRef: "/workspace/foreign-issue-branch", + metadata: { foreign: true }, + }, + ]); + await db + .update(issues) + .set({ + projectId, + executionWorkspaceId: foreignWorkspaceId, + executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "isolated_workspace" }, + }) + .where(eq(issues.id, issueId)); + + const interactionId = randomUUID(); + await db + .update(heartbeatRuns) + .set({ + error: "workspace validation failed before dispatch", + errorCode: "workspace_validation_failed", + resultJson: { workspaceValidation: validation }, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const scheduled = await heartbeat.scheduleBoundedRetry(runId, { + now, + random: () => 0.5, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }); + + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + + const issue = await db + .select({ + executionRunId: issues.executionRunId, + executionWorkspaceId: issues.executionWorkspaceId, + executionWorkspacePreference: issues.executionWorkspacePreference, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue).toMatchObject({ + executionRunId: scheduled.run.id, + executionWorkspaceId: foreignWorkspaceId, + executionWorkspacePreference: "reuse_existing", + }); + + const workspaces = await db + .select({ id: executionWorkspaces.id, status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(inArray(executionWorkspaces.id, [currentWorkspaceId, foreignWorkspaceId])); + expect(workspaces).toEqual(expect.arrayContaining([ + expect.objectContaining({ id: currentWorkspaceId, status: "active", metadata: { current: true } }), + expect.objectContaining({ id: foreignWorkspaceId, status: "active", metadata: { foreign: true } }), + ])); + + const activity = await db + .select({ id: activityLog.id }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "execution_workspace.workspace_validation_quarantined"), + )); + expect(activity).toHaveLength(0); + }); + + it("does not quarantine an owned workspace that is no longer attached to the issue", async () => { + const { companyId, issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" }); + const projectId = randomUUID(); + const staleWorkspaceId = randomUUID(); + const currentWorkspaceId = randomUUID(); + const validation = { + reason: "git_worktree_branch_incoherence", + fingerprint: "workspace_incoherence:v1:sha256:stale-owned", + executionWorkspaceId: staleWorkspaceId, + expectedBranch: "old-plan-approval-workspace", + actualBranch: "current-plan-approval-workspace", + cleanliness: "clean", + }; + + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Paperclip App", + status: "in_progress", + }); + await db.insert(executionWorkspaces).values([ + { + id: staleWorkspaceId, + companyId, + projectId, + sourceIssueId: issueId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "old-plan-approval-workspace", + status: "active", + cwd: "/workspace/old-plan-approval-workspace", + baseRef: "origin/master", + branchName: "old-plan-approval-workspace", + providerType: "git_worktree", + providerRef: "/workspace/old-plan-approval-workspace", + metadata: { stale: true }, + }, + { + id: currentWorkspaceId, + companyId, + projectId, + sourceIssueId: issueId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "current-plan-approval-workspace", + status: "active", + cwd: "/workspace/current-plan-approval-workspace", + baseRef: "origin/master", + branchName: "current-plan-approval-workspace", + providerType: "git_worktree", + providerRef: "/workspace/current-plan-approval-workspace", + metadata: { current: true }, + }, + ]); + await db + .update(issues) + .set({ + projectId, + executionWorkspaceId: currentWorkspaceId, + executionWorkspacePreference: "reuse_existing", + executionWorkspaceSettings: { mode: "isolated_workspace" }, + }) + .where(eq(issues.id, issueId)); + + const interactionId = randomUUID(); + await db + .update(heartbeatRuns) + .set({ + error: "workspace validation failed before dispatch", + errorCode: "workspace_validation_failed", + resultJson: { workspaceValidation: validation }, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId, + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const scheduled = await heartbeat.scheduleBoundedRetry(runId, { + now, + random: () => 0.5, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }); + + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + + const issue = await db + .select({ + executionRunId: issues.executionRunId, + executionWorkspaceId: issues.executionWorkspaceId, + executionWorkspacePreference: issues.executionWorkspacePreference, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue).toMatchObject({ + executionRunId: scheduled.run.id, + executionWorkspaceId: currentWorkspaceId, + executionWorkspacePreference: "reuse_existing", + }); + + const workspaces = await db + .select({ id: executionWorkspaces.id, status: executionWorkspaces.status, metadata: executionWorkspaces.metadata }) + .from(executionWorkspaces) + .where(inArray(executionWorkspaces.id, [staleWorkspaceId, currentWorkspaceId])); + expect(workspaces).toEqual(expect.arrayContaining([ + expect.objectContaining({ id: staleWorkspaceId, status: "active", metadata: { stale: true } }), + expect.objectContaining({ id: currentWorkspaceId, status: "active", metadata: { current: true } }), + ])); + + const activity = await db + .select({ id: activityLog.id }) + .from(activityLog) + .where(and( + eq(activityLog.companyId, companyId), + eq(activityLog.action, "execution_workspace.workspace_validation_quarantined"), + )); + expect(activity).toHaveLength(0); + }); + + it("does not schedule accepted interaction continuation infra retries after terminal issue status", async () => { + const { issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "done" }); + + await db + .update(heartbeatRuns) + .set({ + error: "workspace validation failed before dispatch", + errorCode: "workspace_validation_failed", + resultJson: {}, + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "issue_commented", + mutation: "interaction", + interactionId: randomUUID(), + interactionKind: "request_confirmation", + interactionStatus: "accepted", + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const scheduled = await heartbeat.scheduleBoundedRetry(runId, { + now, + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: 3, + }); + + expect(scheduled).toMatchObject({ + outcome: "not_scheduled", + errorCode: "issue_terminal_status", + issueId, + }); + }); + it("coalesces duplicate max-turn continuation schedules for the same source run and attempt", async () => { const { issueId, runId, now } = await seedMaxTurnFixture(); const retryOptions = { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 620fb16d06..c1b45e3b77 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -20,6 +20,7 @@ import { type IssueExecutionMonitorPolicy, type IssueExecutionMonitorRecoveryPolicy, type ModelProfileKey, + type RequestConfirmationResult, type RoutineRevisionSnapshotV1, type RunLivenessState, type SourceTrustMetadata, @@ -48,6 +49,7 @@ import { issueApprovals, issueComments, issuePlanDecompositions, + issueRecoveryActions, issueRelations, issueThreadInteractions, issues, @@ -302,6 +304,10 @@ const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_JITTER_RATIO = 0.25; const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON = "transient_failure"; const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_WAKE_REASON = "transient_failure_retry"; const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS = BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS.length; +export const INTERACTION_CONTINUATION_INFRA_RETRY_REASON = "interaction_continuation_infra_retry"; +export const INTERACTION_CONTINUATION_INFRA_WAKE_REASON = "interaction_continuation_infra_retry"; +const INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS = 3; +const RESOLVED_INTERACTION_CONTINUATION_STATUSES = new Set(["accepted", "answered", "rejected"]); const WORKSPACE_VALIDATION_FAILURE_CODE = "workspace_validation_failed"; const WORKSPACE_VALIDATION_RECOVERY_CAUSE = "workspace_validation_failed"; const CONFIGURATION_INCOMPLETE_FAILURE_CODE = "configuration_incomplete"; @@ -422,6 +428,45 @@ function readTransientRecoveryContractFromRun( : null; } +function isResolvedInteractionContinuationWakeContext(contextSnapshot: unknown) { + const context = parseObject(contextSnapshot); + const interactionId = readNonEmptyString(context.interactionId); + const interactionStatus = readNonEmptyString(context.interactionStatus); + if (!interactionId || !interactionStatus) return false; + if (!RESOLVED_INTERACTION_CONTINUATION_STATUSES.has(interactionStatus)) return false; + + const mutation = readNonEmptyString(context.mutation); + const wakeReason = readNonEmptyString(context.wakeReason); + const retryReason = readNonEmptyString(context.retryReason); + return ( + (mutation === "interaction" && wakeReason === "issue_commented") || + wakeReason === INTERACTION_CONTINUATION_INFRA_WAKE_REASON || + retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON + ); +} + +function isSpawnLikeFailureMessage(value: unknown) { + if (typeof value !== "string") return false; + return /failed to start command|spawn\b|\bENOENT\b/i.test(value); +} + +function isRetryableInteractionContinuationInfrastructureFailure( + run: Pick, +) { + if (run.errorCode === WORKSPACE_VALIDATION_FAILURE_CODE || run.errorCode === "process_lost") { + return true; + } + + if (run.errorCode !== "adapter_failed" && run.errorCode !== "setup_failed") return false; + + const resultJson = parseObject(run.resultJson); + return ( + isSpawnLikeFailureMessage(run.error) || + isSpawnLikeFailureMessage(resultJson.errorMessage) || + isSpawnLikeFailureMessage(resultJson.message) + ); +} + function mergeAdapterRecoveryMetadata(input: { resultJson: Record | null | undefined; errorFamily?: string | null; @@ -1305,6 +1350,12 @@ function isWorkspaceValidationFailedRun( return run?.errorCode === WORKSPACE_VALIDATION_FAILURE_CODE; } +function readWorkspaceValidationPayloadFromRun( + run: Pick | null | undefined, +) { + return parseObject(parseObject(run?.resultJson).workspaceValidation); +} + function stableStringifyForFingerprint(value: unknown): string { if (Array.isArray(value)) { return `[${value.map((entry) => stableStringifyForFingerprint(entry)).join(",")}]`; @@ -5015,6 +5066,225 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; const budgets = budgetService(db, budgetHooks); const recovery = recoveryService(db, { enqueueWakeup }); + + function isPlanApprovalConfirmationPayload(payload: unknown) { + const target = parseObject(parseObject(payload).target); + return readNonEmptyString(target.type) === "issue_document" && + readNonEmptyString(target.key) === "plan"; + } + + async function getAcceptedPlanApprovalInteractionForRun( + run: typeof heartbeatRuns.$inferSelect, + issueId: string | null, + ) { + const context = parseObject(run.contextSnapshot); + const interactionId = readNonEmptyString(context.interactionId); + if (!issueId || !interactionId) return null; + + const interaction = await db + .select({ + id: issueThreadInteractions.id, + kind: issueThreadInteractions.kind, + status: issueThreadInteractions.status, + payload: issueThreadInteractions.payload, + result: issueThreadInteractions.result, + }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.companyId, run.companyId), + eq(issueThreadInteractions.issueId, issueId), + eq(issueThreadInteractions.id, interactionId), + ), + ) + .then((rows) => rows[0] ?? null); + + if (!interaction) return null; + if (interaction.kind !== "request_confirmation" || interaction.status !== "accepted") return null; + return isPlanApprovalConfirmationPayload(interaction.payload) ? interaction : null; + } + + function planApprovalResumeFailureErrorCode(run: typeof heartbeatRuns.$inferSelect) { + return readNonEmptyString(run.errorCode) ?? "unknown_error"; + } + + function buildPlanApprovalResumeFailureComment(input: { + run: typeof heartbeatRuns.$inferSelect; + status: "retrying" | "needs_attention"; + attempt: number; + maxAttempts: number; + }) { + const errorCode = planApprovalResumeFailureErrorCode(input.run); + if (input.status === "retrying") { + return `Agent failed to resume after approval: \`${errorCode}\` — retrying (attempt ${input.attempt}/${input.maxAttempts})`; + } + return `Agent failed to resume after approval: \`${errorCode}\` — needs attention`; + } + + function buildPlanApprovalResumeFailureResult(input: { + run: typeof heartbeatRuns.$inferSelect; + status: "retrying" | "needs_attention"; + attempt: number; + maxAttempts: number; + retryRunId?: string | null; + recoveryActionId?: string | null; + }): NonNullable { + return { + status: input.status, + errorCode: planApprovalResumeFailureErrorCode(input.run), + attempt: input.attempt, + maxAttempts: input.maxAttempts, + runId: input.run.id, + retryRunId: input.retryRunId ?? null, + recoveryActionId: input.recoveryActionId ?? null, + updatedAt: new Date().toISOString(), + }; + } + + async function updatePlanApprovalInteractionResumeFailure(input: { + interaction: NonNullable>>; + failure: NonNullable; + }) { + const result = parseObject(input.interaction.result); + const nextResult = { + ...result, + version: 1 as const, + outcome: "accepted" as const, + resumeFailure: input.failure, + } satisfies RequestConfirmationResult; + + await db + .update(issueThreadInteractions) + .set({ + result: nextResult, + updatedAt: new Date(), + }) + .where(eq(issueThreadInteractions.id, input.interaction.id)); + } + + async function addPlanApprovalResumeFailureCommentOnce(input: { + issueId: string; + run: typeof heartbeatRuns.$inferSelect; + body: string; + }) { + const existing = await db + .select({ id: issueComments.id }) + .from(issueComments) + .where( + and( + eq(issueComments.companyId, input.run.companyId), + eq(issueComments.issueId, input.issueId), + or( + eq(issueComments.body, input.body), + sql`${issueComments.body} like ${`${input.body}\n%`}`, + ), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + if (existing) return null; + return issuesSvc.addComment(input.issueId, input.body, { runId: input.run.id }, { authorType: "system" }); + } + + async function getActiveRecoveryActionId(companyId: string, sourceIssueId: string) { + return db + .select({ id: issueRecoveryActions.id }) + .from(issueRecoveryActions) + .where( + and( + eq(issueRecoveryActions.companyId, companyId), + eq(issueRecoveryActions.sourceIssueId, sourceIssueId), + inArray(issueRecoveryActions.status, ["active", "escalated"]), + ), + ) + .orderBy(desc(issueRecoveryActions.updatedAt)) + .limit(1) + .then((rows) => rows[0]?.id ?? null); + } + + async function recordPlanApprovalResumeFailureRetry(input: { + run: typeof heartbeatRuns.$inferSelect; + issueId: string | null; + retryRunId: string | null; + attempt: number; + maxAttempts: number; + }) { + const interaction = await getAcceptedPlanApprovalInteractionForRun(input.run, input.issueId); + if (!interaction || !input.issueId) return null; + + const body = buildPlanApprovalResumeFailureComment({ + run: input.run, + status: "retrying", + attempt: input.attempt, + maxAttempts: input.maxAttempts, + }); + await addPlanApprovalResumeFailureCommentOnce({ + issueId: input.issueId, + run: input.run, + body, + }); + await updatePlanApprovalInteractionResumeFailure({ + interaction, + failure: buildPlanApprovalResumeFailureResult({ + run: input.run, + status: "retrying", + attempt: input.attempt, + maxAttempts: input.maxAttempts, + retryRunId: input.retryRunId, + }), + }); + return interaction.id; + } + + async function escalatePlanApprovalResumeFailureNeedsAttention(input: { + run: typeof heartbeatRuns.$inferSelect; + issueId: string | null; + attempt: number; + maxAttempts: number; + }) { + const interaction = await getAcceptedPlanApprovalInteractionForRun(input.run, input.issueId); + if (!interaction || !input.issueId) return null; + + const issue = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, input.run.companyId), eq(issues.id, input.issueId))) + .then((rows) => rows[0] ?? null); + if (!issue) return null; + if (issue.status !== "todo" && issue.status !== "in_progress" && issue.status !== "in_review") return null; + + const body = buildPlanApprovalResumeFailureComment({ + run: input.run, + status: "needs_attention", + attempt: input.attempt, + maxAttempts: input.maxAttempts, + }); + await recovery.escalateStrandedAssignedIssue({ + issue, + previousStatus: issue.status, + latestRun: input.run, + comment: body, + }); + await addPlanApprovalResumeFailureCommentOnce({ + issueId: issue.id, + run: input.run, + body, + }); + + const recoveryActionId = await getActiveRecoveryActionId(issue.companyId, issue.id); + await updatePlanApprovalInteractionResumeFailure({ + interaction, + failure: buildPlanApprovalResumeFailureResult({ + run: input.run, + status: "needs_attention", + attempt: input.attempt, + maxAttempts: input.maxAttempts, + recoveryActionId, + }), + }); + return interaction.id; + } + const productivityReviews = productivityReviewService(db, { enqueueWakeup }); const taskWatchdogs = taskWatchdogService(db, { enqueueWakeup }); let unsafeTextProjectionPromise: Promise | null = null; @@ -8485,7 +8755,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const wakeReason = opts?.wakeReason ?? BOUNDED_TRANSIENT_HEARTBEAT_RETRY_WAKE_REASON; const maxAttempts = Math.max(0, Math.floor(opts?.maxAttempts ?? BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS)); const nextAttempt = (run.scheduledRetryAttempt ?? 0) + 1; - const baseSchedule = opts?.delayMs != null + const computedBaseSchedule = opts?.delayMs != null ? nextAttempt <= maxAttempts ? { attempt: nextAttempt, @@ -8498,6 +8768,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) : nextAttempt <= maxAttempts ? computeBoundedTransientHeartbeatRetrySchedule(nextAttempt, now, opts?.random) : null; + const baseSchedule = computedBaseSchedule ? { ...computedBaseSchedule, maxAttempts } : null; const transientRecovery = retryReason === BOUNDED_TRANSIENT_HEARTBEAT_RETRY_REASON ? readTransientRecoveryContractFromRun(run) @@ -8507,6 +8778,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ? resolveCodexTransientFallbackMode(nextAttempt) : null; const transientRetryNotBefore = transientRecovery?.retryNotBefore ?? null; + const contextSnapshot = parseObject(run.contextSnapshot); + const issueId = readNonEmptyString(contextSnapshot.issueId); if (!baseSchedule) { await appendRunEvent(run, await nextRunEventSeq(run.id), { @@ -8520,6 +8793,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) maxAttempts, }, }); + if (retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON) { + await escalatePlanApprovalResumeFailureNeedsAttention({ + run, + issueId, + attempt: Math.min(run.scheduledRetryAttempt ?? maxAttempts, maxAttempts), + maxAttempts, + }).catch((error) => { + logger.warn( + { err: error, runId: run.id, issueId }, + "failed to escalate exhausted plan-approval resume failure", + ); + }); + } return { outcome: "retry_exhausted" as const, attempt: nextAttempt, @@ -8530,8 +8816,6 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (retryReason !== MAX_TURN_CONTINUATION_RETRY_REASON) { const invokability = await getAgentInvokability(agent); if (!invokability.invokable) { - const contextSnapshot = parseObject(run.contextSnapshot); - const issueId = readNonEmptyString(contextSnapshot.issueId); await appendRunEvent(run, await nextRunEventSeq(run.id), { eventType: "lifecycle", stream: "system", @@ -8564,10 +8848,17 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } : baseSchedule; - const contextSnapshot = parseObject(run.contextSnapshot); - const issueId = readNonEmptyString(contextSnapshot.issueId); - if (retryReason === MAX_TURN_CONTINUATION_RETRY_REASON) { - const gate = await evaluateScheduledRetryGate({ run, agent, contextSnapshot, retryReason }); + const requiresIssueGate = + retryReason === MAX_TURN_CONTINUATION_RETRY_REASON || + retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON; + if (requiresIssueGate) { + const gate = await evaluateScheduledRetryGate({ + run, + agent, + contextSnapshot, + retryReason, + enforceIssueExecutionLock: retryReason === MAX_TURN_CONTINUATION_RETRY_REASON, + }); if (!gate.allowed) { await appendRunEvent(run, await nextRunEventSeq(run.id), { eventType: "lifecycle", @@ -8591,11 +8882,38 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null); const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey); + const interactionContinuationPayload = retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON + ? { + mutation: "interaction", + interactionId: readNonEmptyString(contextSnapshot.interactionId), + interactionKind: readNonEmptyString(contextSnapshot.interactionKind), + interactionStatus: readNonEmptyString(contextSnapshot.interactionStatus), + continuationPolicy: readNonEmptyString(contextSnapshot.continuationPolicy), + } + : {}; + const workspaceValidationRetryPayload = + retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON && isWorkspaceValidationFailedRun(run) + ? readWorkspaceValidationPayloadFromRun(run) + : null; + const shouldQuarantineWorkspaceForRetry = + workspaceValidationRetryPayload !== null && + Object.keys(workspaceValidationRetryPayload).length > 0; const retryContextSnapshot: Record = withRecoveryModelProfileHint({ ...contextSnapshot, retryOfRunId: run.id, wakeReason, retryReason, + ...(shouldQuarantineWorkspaceForRetry + ? { + workspaceValidationRecovery: { + strategy: "quarantine_failed_workspace_and_retry_clean", + sourceRunId: run.id, + reason: readNonEmptyString(workspaceValidationRetryPayload?.reason) ?? WORKSPACE_VALIDATION_FAILURE_CODE, + fingerprint: readNonEmptyString(workspaceValidationRetryPayload?.fingerprint), + failedExecutionWorkspaceId: readNonEmptyString(workspaceValidationRetryPayload?.executionWorkspaceId), + }, + } + : {}), ...(transientRecovery ? { errorFamily: transientRecovery.errorFamily } : {}), scheduledRetryAttempt: schedule.attempt, scheduledRetryAt: schedule.dueAt.toISOString(), @@ -8606,9 +8924,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ...(codexTransientFallbackMode ? { codexTransientFallbackMode } : {}), }, "normal_model"); const responsibleUserId = await resolveResponsibleUserIdForRunContext(run, retryContextSnapshot); - const maxTurnContinuationIdempotencyKey = retryReason === MAX_TURN_CONTINUATION_RETRY_REASON + const continuationRetryIdempotencyKey = retryReason === MAX_TURN_CONTINUATION_RETRY_REASON ? `max-turn-continuation:${run.companyId}:${issueId ?? "no-issue"}:${run.id}:${schedule.attempt}` - : null; + : retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON + ? `interaction-continuation:${run.companyId}:${issueId ?? "no-issue"}:${run.id}:${schedule.attempt}` + : null; type ScheduledRetryTransactionResult = | { @@ -8631,6 +8951,61 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; const scheduleResult = await db.transaction(async (tx): Promise => { + if (retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON) { + if (issueId) { + await tx.execute( + sql`select id from issues where company_id = ${run.companyId} and id = ${issueId} for update`, + ); + } else { + await tx.execute( + sql`select id from heartbeat_runs where company_id = ${run.companyId} and id = ${run.id} for update`, + ); + } + + const existingContinuation = await tx + .select() + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, run.companyId), + eq(heartbeatRuns.retryOfRunId, run.id), + eq(heartbeatRuns.scheduledRetryReason, retryReason), + eq(heartbeatRuns.scheduledRetryAttempt, schedule.attempt), + inArray(heartbeatRuns.status, [...MAX_TURN_CONTINUATION_LIVE_RUN_STATUSES]), + issueId + ? sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}` + : sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' is null`, + ), + ) + .orderBy(asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + + if (existingContinuation) { + if (existingContinuation.wakeupRequestId) { + const existingWakeup = await tx + .select({ coalescedCount: agentWakeupRequests.coalescedCount }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, existingContinuation.wakeupRequestId)) + .then((rows) => rows[0] ?? null); + + await tx + .update(agentWakeupRequests) + .set({ + coalescedCount: (existingWakeup?.coalescedCount ?? 0) + 1, + updatedAt: now, + }) + .where(eq(agentWakeupRequests.id, existingContinuation.wakeupRequestId)); + } + + return { + outcome: "scheduled", + run: existingContinuation, + reusedExisting: true, + }; + } + } + if (retryReason === MAX_TURN_CONTINUATION_RETRY_REASON) { if (issueId) { await tx.execute( @@ -8769,6 +9144,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) payload: withRecoveryModelProfileHint({ ...(issueId ? { issueId } : {}), retryOfRunId: run.id, + ...interactionContinuationPayload, retryReason, ...(transientRecovery ? { errorFamily: transientRecovery.errorFamily } : {}), scheduledRetryAttempt: schedule.attempt, @@ -8782,7 +9158,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) status: "queued", requestedByActorType: "system", requestedByActorId: null, - idempotencyKey: maxTurnContinuationIdempotencyKey, + idempotencyKey: continuationRetryIdempotencyKey, updatedAt: now, }) .returning() @@ -8818,6 +9194,94 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }) .where(eq(agentWakeupRequests.id, wakeupRequest.id)); + let detachWorkspaceFromIssue = false; + if (issueId && shouldQuarantineWorkspaceForRetry) { + const issueWorkspace = await tx + .select({ + id: issues.id, + companyId: issues.companyId, + executionWorkspaceId: issues.executionWorkspaceId, + }) + .from(issues) + .where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId))) + .for("update") + .then((rows) => rows[0] ?? null); + const failedExecutionWorkspaceId = + readNonEmptyString(workspaceValidationRetryPayload?.executionWorkspaceId) ?? + readNonEmptyString(issueWorkspace?.executionWorkspaceId); + + if (issueWorkspace && failedExecutionWorkspaceId) { + const failedWorkspace = await tx + .select({ + id: executionWorkspaces.id, + companyId: executionWorkspaces.companyId, + sourceIssueId: executionWorkspaces.sourceIssueId, + status: executionWorkspaces.status, + metadata: executionWorkspaces.metadata, + }) + .from(executionWorkspaces) + .where(and( + eq(executionWorkspaces.id, failedExecutionWorkspaceId), + eq(executionWorkspaces.companyId, run.companyId), + )) + .for("update") + .then((rows) => rows[0] ?? null); + + const workspaceBelongsToIssue = + failedWorkspace + ? failedWorkspace.sourceIssueId === issueId + : false; + + if ( + failedWorkspace && + workspaceBelongsToIssue && + issueWorkspace.executionWorkspaceId === failedExecutionWorkspaceId + ) { + const existingMetadata = parseObject(failedWorkspace.metadata); + const quarantine = { + reason: WORKSPACE_VALIDATION_FAILURE_CODE, + retryReason, + sourceRunId: run.id, + retryRunId: scheduledRun.id, + issueId, + sourceIssueId: failedWorkspace.sourceIssueId ?? null, + quarantinedAt: now.toISOString(), + workspaceValidation: workspaceValidationRetryPayload ?? {}, + }; + await tx + .update(executionWorkspaces) + .set({ + status: "archived", + closedAt: now, + cleanupEligibleAt: null, + cleanupReason: WORKSPACE_VALIDATION_FAILURE_CODE, + metadata: { + ...existingMetadata, + workspaceValidationQuarantine: quarantine, + }, + updatedAt: now, + }) + .where(and( + eq(executionWorkspaces.id, failedWorkspace.id), + eq(executionWorkspaces.companyId, run.companyId), + )); + + await logActivity(tx as unknown as Db, { + companyId: run.companyId, + actorType: "system", + actorId: "heartbeat", + agentId: run.agentId, + runId: run.id, + action: "execution_workspace.workspace_validation_quarantined", + entityType: "execution_workspace", + entityId: failedWorkspace.id, + details: quarantine, + }); + detachWorkspaceFromIssue = issueWorkspace.executionWorkspaceId === failedExecutionWorkspaceId; + } + } + } + if (issueId) { await tx .update(issues) @@ -8825,6 +9289,12 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) executionRunId: scheduledRun.id, executionAgentNameKey: normalizeAgentNameKey(agent.name), executionLockedAt: now, + ...(detachWorkspaceFromIssue + ? { + executionWorkspaceId: null, + executionWorkspacePreference: null, + } + : {}), updatedAt: now, }) .where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId), eq(issues.executionRunId, run.id))); @@ -8866,11 +9336,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eventType: "lifecycle", stream: "system", level: "info", - message: `Reused existing max-turn continuation ${retryRun.scheduledRetryAttempt}/${schedule.maxAttempts}`, + message: `Reused existing continuation retry ${retryRun.scheduledRetryAttempt}/${schedule.maxAttempts}`, payload: { retryRunId: retryRun.id, retryReason, - idempotencyKey: maxTurnContinuationIdempotencyKey, + idempotencyKey: continuationRetryIdempotencyKey, scheduledRetryAttempt: retryRun.scheduledRetryAttempt, scheduledRetryAt: dueAt.toISOString(), }, @@ -8907,6 +9377,21 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }, }); + if (retryReason === INTERACTION_CONTINUATION_INFRA_RETRY_REASON) { + await recordPlanApprovalResumeFailureRetry({ + run, + issueId, + retryRunId: retryRun.id, + attempt: schedule.attempt, + maxAttempts: schedule.maxAttempts, + }).catch((error) => { + logger.warn( + { err: error, runId: run.id, issueId, retryRunId: retryRun.id }, + "failed to record plan-approval resume retry failure", + ); + }); + } + return { outcome: "scheduled" as const, run: retryRun, @@ -8916,6 +9401,36 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; } + async function scheduleInteractionContinuationInfrastructureRetryIfEligible( + run: typeof heartbeatRuns.$inferSelect, + agent: typeof agents.$inferSelect, + ) { + if (!run.wakeupRequestId) return null; + if (!isResolvedInteractionContinuationWakeContext(run.contextSnapshot)) return null; + if (!isRetryableInteractionContinuationInfrastructureFailure(run)) { + const context = parseObject(run.contextSnapshot); + const issueId = readNonEmptyString(context.issueId); + await escalatePlanApprovalResumeFailureNeedsAttention({ + run, + issueId, + attempt: Math.min(run.scheduledRetryAttempt ?? INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS, INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS), + maxAttempts: INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS, + }).catch((error) => { + logger.warn( + { err: error, runId: run.id, issueId }, + "failed to escalate non-retryable plan-approval resume failure", + ); + }); + return null; + } + + return scheduleBoundedRetryForRun(run, agent, { + retryReason: INTERACTION_CONTINUATION_INFRA_RETRY_REASON, + wakeReason: INTERACTION_CONTINUATION_INFRA_WAKE_REASON, + maxAttempts: INTERACTION_CONTINUATION_INFRA_MAX_ATTEMPTS, + }); + } + async function promoteDueScheduledRetries(now = new Date()) { const dueRuns = await db .select() @@ -10138,12 +10653,17 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); let retriedRun: typeof heartbeatRuns.$inferSelect | null = null; + const retryAgent = await getAgent(run.agentId); if (shouldRetry) { - const agent = await getAgent(run.agentId); - if (agent) { - retriedRun = await enqueueProcessLossRetry(finalizedRun, agent, now); + if (retryAgent) { + retriedRun = await enqueueProcessLossRetry(finalizedRun, retryAgent, now); } - } else { + } else if (retryAgent) { + const scheduled = await scheduleInteractionContinuationInfrastructureRetryIfEligible(finalizedRun, retryAgent); + retriedRun = scheduled?.outcome === "scheduled" ? scheduled.run : null; + } + + if (!retriedRun) { await releaseIssueExecutionAndPromote(finalizedRun); } @@ -12653,6 +13173,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 updateRuntimeState(agent, livenessRun, { @@ -12764,6 +13285,12 @@ 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 releaseIssueExecutionAndPromote(livenessRun).catch((releaseError) => { logger.error( diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 336f5a8fb0..2ef1c1d051 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -715,6 +715,83 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) .then((rows) => Boolean(rows[0])); } + async function getLatestAcceptedContinuationInteraction(companyId: string, issueId: string) { + return db + .select({ + id: issueThreadInteractions.id, + kind: issueThreadInteractions.kind, + status: issueThreadInteractions.status, + continuationPolicy: issueThreadInteractions.continuationPolicy, + sourceRunId: issueThreadInteractions.sourceRunId, + resolvedAt: issueThreadInteractions.resolvedAt, + updatedAt: issueThreadInteractions.updatedAt, + }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.companyId, companyId), + eq(issueThreadInteractions.issueId, issueId), + eq(issueThreadInteractions.status, "accepted"), + inArray(issueThreadInteractions.continuationPolicy, ["wake_assignee", "wake_assignee_on_accept"]), + ), + ) + .orderBy(desc(sql`coalesce(${issueThreadInteractions.resolvedAt}, ${issueThreadInteractions.updatedAt})`), desc(issueThreadInteractions.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function hasSuccessfulIssueRunSince( + companyId: string, + issueId: string, + agentId: string, + since: Date, + interactionId?: string | null, + ) { + return db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, agentId), + eq(heartbeatRuns.status, "succeeded"), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + interactionId + ? sql`${heartbeatRuns.contextSnapshot} ->> 'interactionId' = ${interactionId}` + : sql`true`, + or(gte(heartbeatRuns.createdAt, since), gte(heartbeatRuns.finishedAt, since)), + ), + ) + .limit(1) + .then((rows) => Boolean(rows[0])); + } + + async function getLatestIssueRunSince(companyId: string, issueId: string, agentId: string, since: Date): Promise { + return db + .select({ + id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + contextSnapshot: heartbeatRuns.contextSnapshot, + livenessState: heartbeatRuns.livenessState, + resultJson: heartbeatRuns.resultJson, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, agentId), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + or(gte(heartbeatRuns.createdAt, since), gte(heartbeatRuns.finishedAt, since)), + ), + ) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + // 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 @@ -2949,6 +3026,67 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } + const acceptedContinuationInteraction = await getLatestAcceptedContinuationInteraction(issue.companyId, issue.id); + const acceptedInteractionResolvedAt = acceptedContinuationInteraction + ? acceptedContinuationInteraction.resolvedAt ?? acceptedContinuationInteraction.updatedAt + : null; + if (acceptedContinuationInteraction && acceptedInteractionResolvedAt && !pendingExecutionState) { + const successfulRunSinceResolution = await hasSuccessfulIssueRunSince( + issue.companyId, + issue.id, + agentId, + acceptedInteractionResolvedAt, + acceptedContinuationInteraction.id, + ); + + if (!successfulRunSinceResolution) { + if (!agentInvokable) { + result.skipped += 1; + continue; + } + + if (await hasQueuedIssueWake(issue.companyId, issue.id, agentId)) { + result.skipped += 1; + continue; + } + + if (await isInvocationBudgetBlocked(issue, agentId)) { + result.skipped += 1; + continue; + } + + const latestPostResolutionRun = await getLatestIssueRunSince( + issue.companyId, + issue.id, + agentId, + acceptedInteractionResolvedAt, + ); + const queued = await enqueueStrandedIssueRecovery({ + issueId: issue.id, + agentId, + reason: "issue_continuation_needed", + retryReason: "issue_continuation_needed", + source: "issue.interaction_continuation_recovery", + retryOfRunId: latestPostResolutionRun?.id ?? acceptedContinuationInteraction.sourceRunId ?? latestRun?.id ?? null, + extraContext: { + mutation: "interaction", + interactionId: acceptedContinuationInteraction.id, + interactionKind: acceptedContinuationInteraction.kind, + interactionStatus: acceptedContinuationInteraction.status, + interactionContinuationPolicy: acceptedContinuationInteraction.continuationPolicy, + interactionResolvedAt: acceptedInteractionResolvedAt.toISOString(), + }, + }); + if (queued) { + result.continuationRequeued += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + } + if (issue.status === "in_review") { if (!participantAgentId || !pendingExecutionState) { result.skipped += 1; diff --git a/ui/src/components/IssueThreadInteractionCard.test.tsx b/ui/src/components/IssueThreadInteractionCard.test.tsx index 20a9177689..7e5957b94c 100644 --- a/ui/src/components/IssueThreadInteractionCard.test.tsx +++ b/ui/src/components/IssueThreadInteractionCard.test.tsx @@ -14,6 +14,7 @@ import { disabledDeclineReasonRequestConfirmationInteraction, failedRequestConfirmationInteraction, pendingRequestConfirmationInteraction, + planApprovalResumeFailedRequestConfirmationInteraction, pendingSuggestedTasksInteraction, staleTargetRequestConfirmationInteraction, rejectedSuggestedTasksInteraction, @@ -419,6 +420,19 @@ describe("IssueThreadInteractionCard", () => { accepted.remove(); root = null; + const resumeFailed = renderCard({ + interaction: planApprovalResumeFailedRequestConfirmationInteraction, + }); + expect((resumeFailed.firstElementChild as HTMLElement).className).toContain("border-amber-500/70"); + expect(resumeFailed.textContent).toContain("Approved — agent resume failed"); + expect(resumeFailed.textContent).toContain("Agent resume failed"); + expect(resumeFailed.textContent).toContain("Paperclip needs attention before the agent can resume this approved work."); + expect(resumeFailed.textContent).toContain("adapter_failed"); + + act(() => root?.unmount()); + resumeFailed.remove(); + root = null; + const rejected = renderCard({ interaction: { ...pendingRequestConfirmationInteraction, diff --git a/ui/src/components/IssueThreadInteractionCard.tsx b/ui/src/components/IssueThreadInteractionCard.tsx index 65ddb6ba7f..a66b1e08ca 100644 --- a/ui/src/components/IssueThreadInteractionCard.tsx +++ b/ui/src/components/IssueThreadInteractionCard.tsx @@ -171,10 +171,26 @@ function isPlanConfirmation(interaction: IssueThreadInteraction): boolean { return target?.type === "issue_document" && target?.key === "plan"; } -function planStatusClasses(status: IssueThreadInteraction["status"]) { +function requestConfirmationResumeFailure(interaction: IssueThreadInteraction) { + if (interaction.kind !== "request_confirmation" && interaction.kind !== "request_checkbox_confirmation") return null; + return interaction.result?.resumeFailure ?? null; +} + +function planStatusClasses( + status: IssueThreadInteraction["status"], + resumeFailure?: ReturnType, +) { switch (status) { case "accepted": case "answered": + if (resumeFailure) { + return { + shell: "border-2 border-amber-500/70 bg-transparent", + badge: "border-amber-500/60 bg-amber-500/10 text-amber-900 dark:bg-amber-500/15 dark:text-amber-100", + label: "Approved — agent resume failed", + Icon: AlertTriangle, + }; + } return { shell: "border-2 border-green-500/80 bg-transparent", badge: "border-green-500/60 bg-green-500/10 text-green-900 dark:bg-green-500/15 dark:text-green-100", @@ -1119,6 +1135,32 @@ function RequestConfirmationResolution({ const staleTarget = interaction.result?.staleTarget ?? null; if (interaction.status === "accepted") { + const resumeFailure = requestConfirmationResumeFailure(interaction); + if (resumeFailure) { + return ( +
+
+ Confirmed + +
+
+
+ Agent resume failed +
+

+ {resumeFailure.status === "retrying" + ? `Paperclip is retrying the agent resume after approval (attempt ${resumeFailure.attempt}/${resumeFailure.maxAttempts}).` + : "Paperclip needs attention before the agent can resume this approved work."} +

+ {resumeFailure.errorCode ? ( +

+ Latest cause: {resumeFailure.errorCode} +

+ ) : null} +
+
+ ); + } return (
Confirmed @@ -1910,7 +1952,8 @@ export function IssueThreadInteractionCard({ externalReferences, }: IssueThreadInteractionCardProps) { const isPlan = isPlanConfirmation(interaction); - const planStyles = isPlan ? planStatusClasses(interaction.status) : null; + const resumeFailure = requestConfirmationResumeFailure(interaction); + const planStyles = isPlan ? planStatusClasses(interaction.status, resumeFailure) : null; const StatusIcon = planStyles ? planStyles.Icon : statusIcon(interaction.status); const styles = planStyles ?? statusClasses(interaction.status); const createdByLabel = resolveActorLabel({ diff --git a/ui/src/fixtures/issueThreadInteractionFixtures.ts b/ui/src/fixtures/issueThreadInteractionFixtures.ts index 68030c022a..6970f74e96 100644 --- a/ui/src/fixtures/issueThreadInteractionFixtures.ts +++ b/ui/src/fixtures/issueThreadInteractionFixtures.ts @@ -462,6 +462,24 @@ export const planApprovalAcceptedRequestConfirmationInteraction = createRequestC }, }); +export const planApprovalResumeFailedRequestConfirmationInteraction = createRequestConfirmationInteraction({ + ...planApprovalAcceptedRequestConfirmationInteraction, + id: "interaction-confirmation-plan-resume-failed", + result: { + version: 1, + outcome: "accepted", + resumeFailure: { + status: "needs_attention", + errorCode: "adapter_failed", + attempt: 3, + maxAttempts: 3, + runId: "11111111-1111-4111-8111-222222222222", + recoveryActionId: "33333333-3333-4333-8333-333333333333", + updatedAt: "2026-04-20T14:45:00.000Z", + }, + }, +}); + export const rejectedRequestConfirmationInteraction = createRequestConfirmationInteraction({ id: "interaction-confirmation-rejected", status: "rejected", diff --git a/ui/storybook/stories/issue-thread-interactions.stories.tsx b/ui/storybook/stories/issue-thread-interactions.stories.tsx index 9c2729d2c1..b3910078a4 100644 --- a/ui/storybook/stories/issue-thread-interactions.stories.tsx +++ b/ui/storybook/stories/issue-thread-interactions.stories.tsx @@ -26,6 +26,7 @@ import { pendingRequestConfirmationInteraction, pendingSuggestedTasksInteraction, planApprovalAcceptedRequestConfirmationInteraction, + planApprovalResumeFailedRequestConfirmationInteraction, rejectedNoReasonRequestConfirmationInteraction, rejectedRequestCheckboxConfirmationInteraction, rejectedRequestConfirmationInteraction, @@ -552,6 +553,24 @@ export const RequestConfirmationPlanApprovalConfirmed: Story = { ), }; +export const RequestConfirmationPlanApprovalResumeFailed: Story = { + render: () => ( + + + + + + ), +}; + export const RequestConfirmationFailed: Story = { render: () => (