Finish runs when agent completes issue

This commit is contained in:
Dotta committed 2026-04-21 14:55:23 -05:00
1 parent 17a8ea9475
commit fbbac4f71f
3 files changed
+190 -2

No files matched your search

@@ -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,
+24
View File
@@ -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 =
+113 -2
View File
@@ -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,