From 1fedad05b93ae953c6c1704844a3bcf2aca26869 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 21 Apr 2026 14:03:30 -0500 Subject: [PATCH] Remove terminal issue run cleanup --- .../heartbeat-process-recovery.test.ts | 99 ------------- server/src/routes/issues.ts | 18 --- server/src/services/heartbeat.ts | 139 ------------------ 3 files changed, 256 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 9b33738e466..4f68d803d0d 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -551,105 +551,6 @@ 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("completes active runs discovered by terminal issue context", async () => { - const { agentId, runId, wakeupRequestId, issueId } = await seedRunFixture({ - agentStatus: "running", - }); - 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.completeRunsForTerminalIssue({ - issueId, - issueStatus: "done", - }); - - expect(result.completed).toBe(1); - expect(result.results.map((item) => item.outcome)).toEqual(["completed"]); - - const run = await heartbeat.getRun(runId, { unsafeFullResultJson: true }); - expect(run?.status).toBe("succeeded"); - expect(run?.finishedAt).toBeTruthy(); - 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 9e860716956..a50531502df 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -73,7 +73,6 @@ import { } from "../services/issue-execution-policy.js"; const MAX_ISSUE_COMMENT_LIMIT = 500; -const TERMINAL_ISSUE_RUN_COMPLETION_GRACE_MS = 10_000; const updateIssueRouteSchema = updateIssueSchema.extend({ interrupt: z.boolean().optional(), }); @@ -2043,23 +2042,6 @@ export function issueRoutes( }; } - if ((issue.status === "done" || issue.status === "cancelled") && !isClosedIssueStatus(existing.status)) { - const terminalIssueStatus = issue.status; - setTimeout(() => { - void heartbeat - .completeRunsForTerminalIssue({ - issueId: issue.id, - issueStatus: terminalIssueStatus, - }) - .catch((err) => - logger.warn( - { err, issueId: issue.id }, - "failed to complete heartbeat runs after terminal issue update", - ), - ); - }, TERMINAL_ISSUE_RUN_COMPLETION_GRACE_MS); - } - 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 0b0382dd318..7873b500e34 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1670,14 +1670,6 @@ function isHeartbeatRunTerminalStatus( ); } -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. @@ -6901,133 +6893,6 @@ 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 completeRunsForTerminalIssue(input: { - issueId: string; - issueStatus: "done" | "cancelled"; - }) { - const outcome = terminalIssueStatusToRunOutcome(input.issueStatus); - if (!outcome) return { completed: 0, results: [] }; - - const activeRuns = await db - .select({ id: heartbeatRuns.id }) - .from(heartbeatRuns) - .where( - and( - inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]), - sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${input.issueId}`, - ), - ) - .orderBy(asc(heartbeatRuns.createdAt)); - - const results: Array>> = []; - for (const run of activeRuns) { - results.push( - await completeRunForTerminalIssue({ - runId: run.id, - issueId: input.issueId, - issueStatus: input.issueStatus, - }), - ); - } - - return { - completed: results.filter((result) => result.outcome === "completed").length, - results, - }; - } - async function cancelRunInternal(runId: string, reason = "Cancelled by control plane") { const run = await getRun(runId); if (!run) throw notFound("Heartbeat run not found"); @@ -7448,10 +7313,6 @@ export function heartbeatService(db: Db) { cancelRun: (runId: string) => cancelRunInternal(runId), - completeRunForTerminalIssue, - - completeRunsForTerminalIssue, - cancelActiveForAgent: (agentId: string) => cancelActiveForAgentInternal(agentId), cancelBudgetScopeWork,