diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 4f68d803d0d..1a9732bef40 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -551,6 +551,59 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(wakeup?.status).toBe("claimed"); }); + it("completes a still-running local run when its issue is marked done", async () => { + const child = spawnAliveProcess(); + childProcesses.add(child); + expect(child.pid).toBeTypeOf("number"); + + const { agentId, runId, wakeupRequestId, issueId } = await seedRunFixture({ + agentStatus: "running", + processPid: child.pid ?? null, + }); + await db + .update(issues) + .set({ + status: "done", + completedAt: new Date("2026-03-19T00:05:00.000Z"), + checkoutRunId: null, + executionRunId: null, + }) + .where(eq(issues.id, issueId)); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.completeRunForTerminalIssue({ + runId, + issueId, + issueStatus: "done", + }); + + expect(result.outcome).toBe("completed"); + expect(await waitForPidExit(child.pid!, 2_000)).toBe(true); + + const run = await heartbeat.getRun(runId, { unsafeFullResultJson: true }); + expect(run?.status).toBe("succeeded"); + expect(run?.finishedAt).toBeTruthy(); + expect(run?.error).toBeNull(); + expect(run?.resultJson).toMatchObject({ + summary: "Run completed because its issue was marked done.", + stopReason: "completed", + }); + + const wakeup = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null); + expect(wakeup?.status).toBe("completed"); + + const agent = await db + .select() + .from(agents) + .where(eq(agents.id, agentId)) + .then((rows) => rows[0] ?? null); + expect(agent?.status).toBe("idle"); + }); + it("queues exactly one retry when the recorded local pid is dead", async () => { const { agentId, runId, issueId } = await seedRunFixture({ processPid: 999_999_999, diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 1dd87cb685c..c4081450e69 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -2041,6 +2041,30 @@ export function issueRoutes( ), }; } + + if ( + actor.actorType === "agent" && + actor.runId && + (issue.status === "done" || issue.status === "cancelled") && + !isClosedIssueStatus(existing.status) + ) { + const terminalIssueStatus = issue.status; + setTimeout(() => { + void heartbeat + .completeRunForTerminalIssue({ + runId: actor.runId!, + issueId: issue.id, + issueStatus: terminalIssueStatus, + }) + .catch((err) => + logger.warn( + { err, issueId: issue.id, runId: actor.runId }, + "failed to complete heartbeat run after terminal issue update", + ), + ); + }, 0); + } + const assigneeChanged = issue.assigneeAgentId !== existing.assigneeAgentId || issue.assigneeUserId !== existing.assigneeUserId; const statusChangedFromBacklog = diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 6a4dd65e303..fc1e05d7270 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -136,6 +136,7 @@ const MAX_INLINE_WAKE_COMMENT_BODY_TOTAL_CHARS = 12_000; const execFile = promisify(execFileCallback); const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; +const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const; export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [ 2 * 60 * 1000, @@ -1661,6 +1662,22 @@ function isTrackedLocalChildProcessAdapter(adapterType: string) { return SESSIONED_LOCAL_ADAPTERS.has(adapterType); } +function isHeartbeatRunTerminalStatus( + status: string | null | undefined, +): status is (typeof HEARTBEAT_RUN_TERMINAL_STATUSES)[number] { + return HEARTBEAT_RUN_TERMINAL_STATUSES.includes( + status as (typeof HEARTBEAT_RUN_TERMINAL_STATUSES)[number], + ); +} + +function terminalIssueStatusToRunOutcome( + issueStatus: string | null | undefined, +): "succeeded" | "cancelled" | null { + if (issueStatus === "done") return "succeeded"; + if (issueStatus === "cancelled") return "cancelled"; + return null; +} + // A positive liveness check means some process currently owns the PID. // On Linux, PIDs can be recycled, so this is a best-effort signal rather // than proof that the original child is still alive. @@ -5536,8 +5553,8 @@ export function heartbeatService(db: Db) { let outcome: "succeeded" | "failed" | "cancelled" | "timed_out"; const latestRun = await getRun(run.id); - if (latestRun?.status === "cancelled") { - outcome = "cancelled"; + if (isHeartbeatRunTerminalStatus(latestRun?.status)) { + outcome = latestRun.status; } else if (adapterResult.timedOut) { outcome = "timed_out"; } else if ((adapterResult.exitCode ?? 0) === 0 && !adapterResult.errorMessage) { @@ -6884,6 +6901,98 @@ export function heartbeatService(db: Db) { return wakeupIds.length; } + async function completeRunForTerminalIssue(input: { + runId: string; + issueId: string; + issueStatus: "done" | "cancelled"; + }) { + const outcome = terminalIssueStatusToRunOutcome(input.issueStatus); + if (!outcome) return { outcome: "not_terminal_issue" as const }; + + const run = await getRun(input.runId); + if (!run) return { outcome: "missing_run" as const }; + if (!CANCELLABLE_HEARTBEAT_RUN_STATUSES.includes(run.status as (typeof CANCELLABLE_HEARTBEAT_RUN_STATUSES)[number])) { + return { outcome: "run_not_active" as const, run }; + } + + const contextIssueId = readNonEmptyString(parseObject(run.contextSnapshot).issueId); + if (contextIssueId !== input.issueId) { + return { outcome: "issue_mismatch" as const }; + } + + const issue = await db + .select({ + id: issues.id, + companyId: issues.companyId, + status: issues.status, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(and(eq(issues.id, input.issueId), eq(issues.companyId, run.companyId))) + .then((rows) => rows[0] ?? null); + if (!issue) return { outcome: "missing_issue" as const }; + if (issue.status !== input.issueStatus) return { outcome: "issue_status_changed" as const }; + if (issue.executionRunId && issue.executionRunId !== run.id) { + return { outcome: "issue_owned_by_other_run" as const }; + } + + const agent = await getAgent(run.agentId); + const reason = + input.issueStatus === "done" + ? "Run completed because its issue was marked done." + : "Run cancelled because its issue was cancelled."; + + const running = runningProcesses.get(run.id); + if (running) { + await terminateHeartbeatRunProcess({ + pid: running.child.pid ?? run.processPid, + processGroupId: running.processGroupId ?? run.processGroupId, + graceMs: Math.max(1, running.graceSec) * 1000, + }); + } else if (run.processPid || run.processGroupId) { + await terminateHeartbeatRunProcess({ + pid: run.processPid, + processGroupId: run.processGroupId, + }); + } + + const completed = await setRunStatus(run.id, outcome, { + finishedAt: new Date(), + error: outcome === "cancelled" ? reason : null, + errorCode: outcome === "cancelled" ? "issue_cancelled" : null, + ...(agent ? { + resultJson: mergeRunStopMetadataForAgent(agent, outcome, { + resultJson: { + ...parseObject(run.resultJson), + summary: reason, + }, + errorCode: outcome === "cancelled" ? "issue_cancelled" : null, + errorMessage: outcome === "cancelled" ? reason : null, + }), + } : {}), + }); + + await setWakeupStatus(run.wakeupRequestId, outcome === "succeeded" ? "completed" : outcome, { + finishedAt: new Date(), + error: outcome === "cancelled" ? reason : null, + }); + + if (completed) { + await appendRunEvent(completed, await nextRunEventSeq(completed.id), { + eventType: "lifecycle", + stream: "system", + level: outcome === "succeeded" ? "info" : "warn", + message: reason, + }); + await releaseIssueExecutionAndPromote(completed); + } + + runningProcesses.delete(run.id); + await finalizeAgentStatus(run.agentId, outcome); + await startNextQueuedRunForAgent(run.agentId); + return { outcome: "completed" as const, run: completed }; + } + async function cancelRunInternal(runId: string, reason = "Cancelled by control plane") { const run = await getRun(runId); if (!run) throw notFound("Heartbeat run not found"); @@ -7304,6 +7413,8 @@ export function heartbeatService(db: Db) { cancelRun: (runId: string) => cancelRunInternal(runId), + completeRunForTerminalIssue, + cancelActiveForAgent: (agentId: string) => cancelActiveForAgentInternal(agentId), cancelBudgetScopeWork,