diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 212f19d00bd..ccf848eb184 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -5841,6 +5841,171 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { ).toBe(true); }); + it("retries an interrupted corrective successful-run handoff instead of escalating it", async () => { + // A graceful server shutdown (a deploy restart) interrupted the single + // corrective run before the agent could choose a disposition. That is + // not a finished attempt, so the sweeper must give it the bounded + // transient retry any interrupted run gets, not a board escalation. + const { agentId, runId, issueId } = await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + runErrorCode: "server_shutdown_interrupted", + runError: "Interrupted by graceful server shutdown (SIGTERM)", + }); + const sourceRunId = randomUUID(); + await db + .update(heartbeatRuns) + .set({ + status: "interrupted", + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "finish_successful_run_handoff", + sourceRunId, + resumeFromRunId: sourceRunId, + handoffRequired: true, + handoffReason: "successful_run_missing_state", + missingDisposition: "clear_next_step", + handoffAttempt: 1, + maxHandoffAttempts: 1, + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.successfulRunHandoffRetried).toBe(1); + expect(result.successfulRunHandoffEscalated).toBe(0); + expect(result.escalated).toBe(0); + expect(result.issueIds).toEqual([issueId]); + + const retryRun = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, runId)) + .then((rows) => rows[0] ?? null); + expect(retryRun).toMatchObject({ + agentId, + status: "scheduled_retry", + scheduledRetryAttempt: 1, + scheduledRetryReason: "transient_failure", + }); + // The retry keeps the handoff contract: it is still the corrective run, + // so a finished attempt that leaves no disposition still escalates. + expect(retryRun?.contextSnapshot).toMatchObject({ + issueId, + retryOfRunId: runId, + handoffRequired: true, + handoffReason: "successful_run_missing_state", + missingDisposition: "clear_next_step", + handoffAttempt: 1, + maxHandoffAttempts: 1, + }); + + const sourceIssue = await db + .select() + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(sourceIssue?.status).toBe("in_progress"); + expect(sourceIssue?.assigneeAgentId).toBe(agentId); + const recoveryActions = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(recoveryActions).toHaveLength(0); + const comments = await db + .select() + .from(issueComments) + .where(eq(issueComments.issueId, issueId)); + expect(comments).toHaveLength(0); + + // A second sweep sees the scheduled retry as the live path and leaves + // the issue alone rather than escalating or double-scheduling. + const again = await heartbeat.reconcileStrandedAssignedIssues(); + expect(again.successfulRunHandoffRetried).toBe(0); + expect(again.successfulRunHandoffEscalated).toBe(0); + const retryRuns = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, runId)); + expect(retryRuns).toHaveLength(1); + }); + + it("escalates an interrupted corrective successful-run handoff once its transient retry budget is spent", async () => { + const { companyId, agentId, runId, issueId } = + await seedStrandedIssueFixture({ + status: "in_progress", + runStatus: "failed", + runErrorCode: "server_shutdown_interrupted", + runError: "Interrupted by graceful server shutdown (SIGTERM)", + // The real interrupted run carries the conversation-continuation + // marker the shutdown path writes; it also keeps a spent retry chain + // out of legacy reconciliation so the sweeper reaches the handoff + // branch with an exhausted budget. + resultJson: { + stopReason: "interrupted", + conversationContinuation: "continue_conversation_v1", + }, + }); + const sourceRunId = randomUUID(); + await db + .update(heartbeatRuns) + .set({ + status: "interrupted", + scheduledRetryAttempt: 2, + scheduledRetryReason: "transient_failure", + contextSnapshot: { + issueId, + taskId: issueId, + wakeReason: "transient_failure_retry", + sourceRunId, + resumeFromRunId: sourceRunId, + handoffRequired: true, + handoffReason: "successful_run_missing_state", + missingDisposition: "clear_next_step", + handoffAttempt: 1, + maxHandoffAttempts: 1, + }, + }) + .where(eq(heartbeatRuns.id, runId)); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reconcileStrandedAssignedIssues(); + expect(result.successfulRunHandoffRetried).toBe(0); + expect(result.successfulRunHandoffEscalated).toBe(1); + expect(result.issueIds).toEqual([issueId]); + const retryRuns = await db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, runId)); + expect(retryRuns).toHaveLength(0); + + const recoveryAction = await expectSourceScopedStrandedRecoveryAction({ + companyId, + agentId, + issueId, + runId, + previousStatus: "in_progress", + retryReason: null, + cause: SUCCESSFUL_RUN_MISSING_STATE_REASON, + kind: "missing_disposition", + }); + expect(recoveryAction.evidence).toMatchObject({ + sourceRunId, + missingDisposition: "clear_next_step", + latestRunStatus: "interrupted", + latestRunErrorCode: "server_shutdown_interrupted", + recoveryCause: SUCCESSFUL_RUN_MISSING_STATE_REASON, + }); + const sourceIssue = await db + .select() + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + expect(sourceIssue?.status).toBe("blocked"); + }); + it("escalates an exhausted successful handoff run that still leaves no disposition", async () => { const { companyId, agentId, runId, issueId } = await seedStrandedIssueFixture({ diff --git a/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts b/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts index 32ddc0de94b..8de38150f94 100644 --- a/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts +++ b/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts @@ -75,12 +75,10 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => { await tempDb?.cleanup(); }); - async function seedQueuedRun() { + async function seedAgentAndIssue() { const companyId = randomUUID(); const agentId = randomUUID(); const issueId = randomUUID(); - const runId = randomUUID(); - const wakeupRequestId = randomUUID(); await db.insert(companies).values({ id: companyId, @@ -120,6 +118,14 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => { responsibleUserId: "responsible-user", }); + return { companyId, agentId, issueId }; + } + + async function seedQueuedRun() { + const { companyId, agentId, issueId } = await seedAgentAndIssue(); + const runId = randomUUID(); + const wakeupRequestId = randomUUID(); + await db.insert(agentWakeupRequests).values({ id: wakeupRequestId, companyId, @@ -219,6 +225,78 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => { expect(finished?.status).toBe("succeeded"); }, 20_000); + it("keeps a wake that arrives during a task drain queued and runs it once the drain lifts", async () => { + const { agentId, issueId } = await seedAgentAndIssue(); + const heartbeat = heartbeatService(db); + + // The drain holds ADMISSION, not the request. The wake lands in the + // durable queue, so the restart that clears the process-local drain + // resumes it instead of losing it. (Before this, the wake was written + // as `skipped` and the issue sat in `todo` with no run and no path.) + startTaskDrain({}); + await heartbeat.wakeup(agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId }, + contextSnapshot: { issueId, wakeReason: "issue_assigned" }, + requestedByActorType: "system", + requestedByActorId: "issue_assignment", + }); + await heartbeat.drainActiveRunExecutions(); + + const wakeup = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + payload: agentWakeupRequests.payload, + }) + .from(agentWakeupRequests) + .then((rows) => rows[0] ?? null); + expect(wakeup).toMatchObject({ status: "queued", reason: "issue_assigned" }); + expect(wakeup?.payload).not.toHaveProperty("heartbeatSkip"); + const queuedRuns = await db + .select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) + .from(heartbeatRuns); + expect(queuedRuns).toHaveLength(1); + expect(queuedRuns[0]).toMatchObject({ status: "queued" }); + const runId = queuedRuns[0]!.id; + + // Admission stays held for as long as the drain is active, and the + // queued row does not count against quiescence: the restart may proceed. + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + const held = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(held?.status).toBe("queued"); + expect(getTaskDrainStatus()).toMatchObject({ + draining: true, + activeRuns: 0, + pendingWakes: 0, + quiescent: true, + }); + + // The drain lifts (a restart clears it the same way): the queued wake + // runs to completion through the normal admission path. + stopTaskDrain(); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + const finished = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(finished?.status).toBe("succeeded"); + const completedWakeup = await db + .select({ status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .then((rows) => rows[0] ?? null); + expect(completedWakeup?.status).toBe("completed"); + }, 20_000); + // Inject one rollback only after the test starts task drain. Admission also // uses transactions, so transaction call numbers do not identify release. function withFailingTransactionalUpdate(realDb: typeof db, failingTable: unknown, armed: () => boolean) { diff --git a/server/src/index.ts b/server/src/index.ts index ba8cde96eb9..dc47462a483 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1529,6 +1529,7 @@ async function startServerWithDatabaseTeardown( reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || reconciled.successfulRunHandoffEscalated > 0 || + reconciled.successfulRunHandoffRetried > 0 || reconciled.escalated > 0 ) { logger.warn( @@ -1769,6 +1770,7 @@ async function startServerWithDatabaseTeardown( reconciled.dispatchRequeued > 0 || reconciled.continuationRequeued > 0 || reconciled.successfulRunHandoffEscalated > 0 || + reconciled.successfulRunHandoffRetried > 0 || reconciled.escalated > 0 ) { logger.warn( diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index abf7e8a3425..48a252089aa 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -26339,7 +26339,19 @@ export function heartbeatService( }; const schedulingSuppression = await getSchedulingSuppression(); - if (schedulingSuppression.suppressed) { + // A task drain holds ADMISSION, not the request. The drain is a + // process-local pre-restart hold, so a wake that arrives while it is + // active still names real work that must run once the process comes + // back: leave it in the durable queue and let the dispatch-side checks + // (startNextQueuedRunForAgent / executeRun) keep it from starting until + // the drain lifts or the restart clears it. Writing it as `skipped` + // here dropped the wake permanently — an accepted plan whose + // continuation wake landed mid-drain left its issue in `todo` with no + // run and no path until a person noticed. + if ( + schedulingSuppression.suppressed && + schedulingSuppression.reason !== "task_drain" + ) { await writeSkippedHeartbeatRequest("heartbeat.scheduling_suppressed", { reason: schedulingSuppression.reason, }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 19316440a2a..98be0347ae0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -4174,6 +4174,7 @@ export function recoveryService( successfulContinuationObserved: 0, orphanBlockersAssigned: 0, successfulRunHandoffEscalated: 0, + successfulRunHandoffRetried: 0, reviewParticipantRequeued: 0, escalated: 0, waitingOnReviewResolved: 0, @@ -4995,6 +4996,39 @@ export function recoveryService( continue; } + // An interrupted corrective run is not evidence that the agent could + // not choose a disposition: a graceful server shutdown (a deploy + // restart, a lost process) ended the attempt before the agent + // finished. The attempt cap counts attempts the agent got to finish, + // so give the interrupted run the same bounded transient retry any + // interrupted run gets — the retry keeps the handoff context, so it + // is still the corrective run — and escalate only once that retry + // budget is spent or a finished attempt still leaves no disposition. + // Native-runtime corrective runs have no process-loss retry lane + // (a graceful shutdown suspends their controller for reattach + // instead of interrupting them; other native interruptions are + // reconciled by their own finalizer), so the recovery enqueue + // returns null for them and they escalate exactly as before. + if (latestRun?.status === "interrupted") { + if (await isInvocationBudgetBlocked(issue, agentId)) { + result.skipped += 1; + continue; + } + const retried = await enqueueStrandedIssueRecovery({ + issueId: issue.id, + agentId, + reason: "issue_continuation_needed", + retryReason: "issue_continuation_needed", + source: "issue.successful_run_handoff_interrupted_retry", + retryOfRunId: latestRun.id, + }); + if (retried) { + result.successfulRunHandoffRetried += 1; + result.issueIds.push(issue.id); + continue; + } + } + const updated = await escalateStrandedAssignedIssue({ issue, previousStatus: "in_progress",