diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 3d85828fbc4..173a13e8e3b 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -2339,4 +2339,75 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); expect(runs).toHaveLength(1); }); + + it("suppresses bounded liveness continuation wakes when the issue is held by an active pause hold", async () => { + const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "plan_only", + livenessReason: "Run made no concrete progress", + nextAction: "Take the first concrete action now.", + activePauseHold: true, + }); + const heartbeat = heartbeatService(db); + + const sourceRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + if (!sourceRun) throw new Error("Expected source run"); + + await heartbeat.handleRunLivenessContinuationForTest(sourceRun); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(0); + + const refreshedRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(refreshedRun?.continuationAttempt).toBe(0); + expect(refreshedRun?.livenessReason).toContain("continuation suppressed by active pause hold"); + + const recoveryIssues = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.originKind, "stranded_issue_recovery"))); + expect(recoveryIssues).toHaveLength(0); + + const refreshedIssue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); + expect(refreshedIssue?.status).toBe("in_progress"); + expect(refreshedIssue?.assigneeAgentId).toBe(agentId); + }); + + it("queues a bounded liveness continuation when no pause hold suppresses it", async () => { + const { companyId, runId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "succeeded", + livenessState: "plan_only", + livenessReason: "Run made no concrete progress", + nextAction: "Take the first concrete action now.", + }); + const heartbeat = heartbeatService(db); + + const sourceRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + if (!sourceRun) throw new Error("Expected source run"); + + await heartbeat.handleRunLivenessContinuationForTest(sourceRun); + + const wakeups = await db + .select() + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "run_liveness_continuation"))); + expect(wakeups).toHaveLength(1); + }); }); diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 9999c05e523..edf94d6eb4e 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { + activityLog, agents, agentRuntimeState, agentWakeupRequests, @@ -10,6 +11,8 @@ import { environmentLeases, heartbeatRunEvents, heartbeatRuns, + issueTreeHoldMembers, + issueTreeHolds, issues, } from "@paperclipai/db"; import { @@ -42,8 +45,11 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { }, 20_000); afterEach(async () => { + await db.delete(activityLog); await db.delete(heartbeatRunEvents); await db.delete(environmentLeases); + await db.delete(issueTreeHoldMembers); + await db.delete(issueTreeHolds); await db.delete(issues); await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); @@ -834,4 +840,132 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { retryNotBefore.toISOString(), ); }); + + it("cancels a due scheduled retry whose issue is held by an active pause hold", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const sourceRunId = randomUUID(); + const now = new Date("2026-04-20T20:00:00.000Z"); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }); + + await db.insert(heartbeatRuns).values({ + id: sourceRunId, + companyId, + agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "failed", + error: "upstream overload", + errorCode: "adapter_failed", + finishedAt: now, + contextSnapshot: { + issueId, + wakeReason: "issue_assigned", + }, + updatedAt: now, + createdAt: now, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Pause-hold scheduled retry", + status: "todo", + priority: "medium", + assigneeAgentId: agentId, + executionRunId: sourceRunId, + executionAgentNameKey: "codexcoder", + executionLockedAt: now, + issueNumber: 1, + identifier: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}-1`, + }); + + const scheduled = await heartbeat.scheduleBoundedRetry(sourceRunId, { + now, + random: () => 0.5, + }); + expect(scheduled.outcome).toBe("scheduled"); + if (scheduled.outcome !== "scheduled") return; + + await db.insert(issueTreeHolds).values({ + companyId, + rootIssueId: issueId, + mode: "pause", + status: "active", + reason: "pause scheduled retry promotion", + releasePolicy: { strategy: "manual" }, + }); + + const promotion = await heartbeat.promoteDueScheduledRetries(scheduled.dueAt); + expect(promotion).toEqual({ promoted: 0, runIds: [] }); + + const cancelled = await db + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + error: heartbeatRuns.error, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, scheduled.run.id)) + .then((rows) => rows[0] ?? null); + expect(cancelled).toMatchObject({ + status: "cancelled", + errorCode: "issue_paused", + }); + expect(cancelled?.error).toContain("active subtree pause hold"); + + const wakeupRequestId = cancelled?.wakeupRequestId ?? null; + if (wakeupRequestId) { + const wakeup = await db + .select({ status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null); + expect(wakeup?.status).toBe("cancelled"); + } + + const issue = await db + .select({ executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(issue?.executionRunId).toBeNull(); + + const event = await db + .select({ + message: heartbeatRunEvents.message, + payload: heartbeatRunEvents.payload, + }) + .from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, scheduled.run.id)) + .orderBy(sql`${heartbeatRunEvents.id} desc`) + .then((rows) => rows[0] ?? null); + expect(event?.message).toContain("active subtree pause hold"); + expect(event?.payload).toMatchObject({ issueId }); + }); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 963ad5bfb7a..f04bd090caa 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2833,6 +2833,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); return; } + + if (await isAutomaticRecoverySuppressedByPauseHold(db, issue.companyId, issue.id, treeControlSvc)) { + await setRunStatus(run.id, run.status, { + livenessReason: + `${run.livenessReason ?? "Run ended without concrete progress"}; continuation suppressed by active pause hold`, + }); + return; + } } const nextAttempt = readContinuationAttempt(run.continuationAttempt) + 1; @@ -3657,6 +3665,88 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); continue; } + + const activePauseHold = await treeControlSvc.getActivePauseHoldGate(dueRun.companyId, dueRunIssueId); + if (activePauseHold) { + const reason = "Cancelled because issue is held by an active subtree pause hold"; + const cancelled = await db + .update(heartbeatRuns) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + errorCode: "issue_paused", + updatedAt: now, + }) + .where( + and( + eq(heartbeatRuns.id, dueRun.id), + eq(heartbeatRuns.status, "scheduled_retry"), + lte(heartbeatRuns.scheduledRetryAt, now), + ), + ) + .returning() + .then((rows) => rows[0] ?? null); + + if (!cancelled) continue; + + if (cancelled.wakeupRequestId) { + await db + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + updatedAt: now, + }) + .where(eq(agentWakeupRequests.id, cancelled.wakeupRequestId)); + } + + if (issue && issue.executionRunId === cancelled.id) { + await db + .update(issues) + .set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: now, + }) + .where(and(eq(issues.id, issue.id), eq(issues.executionRunId, cancelled.id))); + } + + await appendRunEvent(cancelled, await nextRunEventSeq(cancelled.id), { + eventType: "lifecycle", + stream: "system", + level: "warn", + message: "Scheduled retry cancelled because issue is held by an active subtree pause hold", + payload: { + issueId: dueRunIssueId, + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + scheduledRetryAttempt: cancelled.scheduledRetryAttempt, + scheduledRetryAt: cancelled.scheduledRetryAt ? new Date(cancelled.scheduledRetryAt).toISOString() : null, + scheduledRetryReason: cancelled.scheduledRetryReason, + }, + }); + + await logActivity(db, { + companyId: cancelled.companyId, + actorType: "system", + actorId: "system", + agentId: cancelled.agentId, + runId: cancelled.id, + action: "issue.tree_hold_run_interrupted", + entityType: "heartbeat_run", + entityId: cancelled.id, + details: { + issueId: dueRunIssueId, + holdId: activePauseHold.holdId, + rootIssueId: activePauseHold.rootIssueId, + source: "heartbeat.promote_due_scheduled_retries", + }, + }); + continue; + } } const promoted = await db @@ -7497,6 +7587,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) reapOrphanedRuns, + handleRunLivenessContinuationForTest: handleRunLivenessContinuation, + promoteDueScheduledRetries, resumeQueuedRuns,