mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
Remove terminal issue run cleanup
This commit is contained in:
1 parent
34307374b9
commit
1fedad05b9
3 files changed
-256
No files matched your search
@@ -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,
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -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<Awaited<ReturnType<typeof completeRunForTerminalIssue>>> = [];
|
||||
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,
|
||||
|
||||
Reference in new issue
Block a user