Allow recovery owners to cancel stale runs

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-04-29 14:00:36 -05:00
1 parent 229a4c60a2
commit 33bcaa5a4c
8 files changed
+373 -11

No files matched your search

@@ -66,6 +66,10 @@ const mockHeartbeatService = vi.hoisted(() => ({
cancelRun: vi.fn(),
}));
const mockRecoveryService = vi.hoisted(() => ({
recordWatchdogDecision: vi.fn(),
}));
const mockIssueApprovalService = vi.hoisted(() => ({
linkManyForApproval: vi.fn(),
}));
@@ -142,6 +146,10 @@ function registerModuleMocks() {
heartbeatService: () => mockHeartbeatService,
}));
vi.doMock("../services/recovery/service.js", () => ({
recoveryService: () => mockRecoveryService,
}));
vi.doMock("../services/issue-approvals.js", () => ({
issueApprovalService: () => mockIssueApprovalService,
}));
@@ -271,6 +279,7 @@ describe.sequential("agent permission routes", () => {
vi.doUnmock("../services/budgets.js");
vi.doUnmock("../services/company-skills.js");
vi.doUnmock("../services/heartbeat.js");
vi.doUnmock("../services/recovery/service.js");
vi.doUnmock("../services/index.js");
vi.doUnmock("../services/instance-settings.js");
vi.doUnmock("../services/issue-approvals.js");
@@ -306,6 +315,7 @@ describe.sequential("agent permission routes", () => {
mockHeartbeatService.resetRuntimeSession.mockReset();
mockHeartbeatService.getRun.mockReset();
mockHeartbeatService.cancelRun.mockReset();
mockRecoveryService.recordWatchdogDecision.mockReset();
mockIssueApprovalService.linkManyForApproval.mockReset();
mockIssueService.list.mockReset();
mockSecretService.normalizeAdapterConfigForPersistence.mockReset();
@@ -374,6 +384,13 @@ describe.sequential("agent permission routes", () => {
censorUsernameInLogs: false,
});
mockLogActivity.mockResolvedValue(undefined);
mockRecoveryService.recordWatchdogDecision.mockResolvedValue({
id: "watchdog-decision-1",
runId: "run-1",
evaluationIssueId: "evaluation-1",
decision: "cancel_run",
reason: "The stale run is stuck",
});
});
it("redacts agent detail for authenticated company members without agent admin permission", async () => {
@@ -1194,4 +1211,114 @@ describe.sequential("agent permission routes", () => {
expect(res.status).toBe(403);
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
});
it("keeps generic heartbeat cancellation board-only", async () => {
const app = await createApp({
type: "agent",
agentId,
companyId,
source: "agent_key",
runId: "actor-run-1",
});
const res = await requestApp(app, (baseUrl) => request(baseUrl)
.post("/api/heartbeat-runs/run-1/cancel")
.send({}));
expect(res.status).toBe(403);
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
});
it("lets an agent use the watchdog recovery cancel route without board access", async () => {
mockHeartbeatService.getRun.mockResolvedValue({
id: "run-1",
companyId,
agentId,
status: "running",
});
mockHeartbeatService.cancelRun.mockResolvedValue({
id: "run-1",
companyId,
agentId,
status: "cancelled",
});
const app = await createApp({
type: "agent",
agentId,
companyId,
source: "agent_key",
runId: "actor-run-1",
});
const res = await requestApp(app, (baseUrl) => request(baseUrl)
.post("/api/heartbeat-runs/run-1/watchdog-decisions")
.send({
decision: "cancel_run",
evaluationIssueId: "evaluation-1",
reason: "The stale run is stuck",
}));
expect(res.status).toBe(200);
expect(mockRecoveryService.recordWatchdogDecision).toHaveBeenCalledWith({
runId: "run-1",
actor: expect.objectContaining({
type: "agent",
agentId,
runId: "actor-run-1",
}),
decision: "cancel_run",
evaluationIssueId: "evaluation-1",
reason: "The stale run is stuck",
snoozedUntil: null,
createdByRunId: "actor-run-1",
});
expect(mockHeartbeatService.cancelRun).toHaveBeenCalledWith("run-1", "The stale run is stuck");
expect(mockLogActivity).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
actorType: "agent",
actorId: agentId,
agentId,
runId: "run-1",
action: "heartbeat.watchdog_cancelled",
entityType: "heartbeat_run",
entityId: "run-1",
details: expect.objectContaining({
decision: "cancel_run",
evaluationIssueId: "evaluation-1",
reason: "The stale run is stuck",
createdByRunId: "actor-run-1",
targetRunId: "run-1",
previousStatus: "running",
resultStatus: "cancelled",
}),
}));
});
it("requires an evaluation issue on watchdog recovery cancellation", async () => {
mockHeartbeatService.getRun.mockResolvedValue({
id: "run-1",
companyId,
agentId,
status: "running",
});
const app = await createApp({
type: "agent",
agentId,
companyId,
source: "agent_key",
runId: "actor-run-1",
});
const res = await requestApp(app, (baseUrl) => request(baseUrl)
.post("/api/heartbeat-runs/run-1/watchdog-decisions")
.send({
decision: "cancel_run",
reason: "The stale run is stuck",
}));
expect(res.status).toBe(400);
expect(mockRecoveryService.recordWatchdogDecision).not.toHaveBeenCalled();
expect(mockHeartbeatService.cancelRun).not.toHaveBeenCalled();
});
});
@@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto";
import { and, eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import {
activityLog,
agents,
companies,
createDb,
@@ -353,6 +354,87 @@ describeEmbeddedPostgres("active-run output watchdog", () => {
});
});
it("lets the assigned recovery owner record cancel_run and uses the normal cancellation path", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId, issueId } = await seedRunningRun({
now,
ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000,
});
const heartbeat = heartbeatService(db);
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
const scan = await heartbeat.scanSilentActiveRuns({ now, companyId });
const evaluationIssueId = scan.evaluationIssueIds[0];
expect(evaluationIssueId).toBeTruthy();
const managerRunId = randomUUID();
await db.insert(heartbeatRuns).values({
id: managerRunId,
companyId,
agentId: managerId,
status: "succeeded",
invocationSource: "assignment",
triggerDetail: "system",
startedAt: now,
finishedAt: now,
processStartedAt: now,
lastOutputAt: now,
lastOutputSeq: 1,
lastOutputStream: "stdout",
contextSnapshot: {},
logBytes: 0,
});
const decision = await recovery.recordWatchdogDecision({
runId,
actor: { type: "agent", agentId: managerId, runId: managerRunId },
decision: "cancel_run",
evaluationIssueId,
reason: "Run is silent and evidence shows it is stuck",
});
expect(decision).toMatchObject({
runId,
evaluationIssueId,
decision: "cancel_run",
reason: "Run is silent and evidence shows it is stuck",
createdByAgentId: managerId,
createdByRunId: managerRunId,
});
const cancelled = await heartbeat.cancelRun(runId, decision.reason ?? undefined);
expect(cancelled).toMatchObject({
id: runId,
status: "cancelled",
error: "Run is silent and evidence shows it is stuck",
errorCode: "cancelled",
});
const [sourceIssue] = await db.select().from(issues).where(eq(issues.id, issueId));
expect(sourceIssue?.executionRunId).toBeNull();
expect(sourceIssue?.executionLockedAt).toBeNull();
const [audit] = await db
.select()
.from(activityLog)
.where(and(eq(activityLog.companyId, companyId), eq(activityLog.action, "heartbeat.watchdog_cancel_requested")));
expect(audit).toMatchObject({
actorType: "agent",
actorId: managerId,
agentId: managerId,
runId,
entityType: "heartbeat_run",
entityId: runId,
});
expect(audit?.details).toMatchObject({
decision: "cancel_run",
evaluationIssueId,
targetRunId: runId,
reason: "Run is silent and evidence shows it is stuck",
createdByRunId: managerRunId,
});
});
it("re-arms continue decisions after the default quiet window", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId } = await seedRunningRun({
@@ -470,6 +552,7 @@ describeEmbeddedPostgres("active-run output watchdog", () => {
snoozedUntil: new Date(now.getTime() + 60 * 60 * 1000),
},
{ decision: "continue" as const, evaluationIssueId: otherEvaluationIssueId },
{ decision: "cancel_run" as const, evaluationIssueId: otherEvaluationIssueId },
];
for (const attempt of attempts) {
@@ -495,6 +578,61 @@ describeEmbeddedPostgres("active-run output watchdog", () => {
).rejects.toMatchObject({ status: 403 });
});
it("rejects cancel_run for terminal runs before writing a decision", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId } = await seedRunningRun({
now,
ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000,
});
const heartbeat = heartbeatService(db);
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
const scan = await heartbeat.scanSilentActiveRuns({ now, companyId });
const evaluationIssueId = scan.evaluationIssueIds[0];
expect(evaluationIssueId).toBeTruthy();
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: now }).where(eq(heartbeatRuns.id, runId));
await expect(
recovery.recordWatchdogDecision({
runId,
actor: { type: "agent", agentId: managerId },
decision: "cancel_run",
evaluationIssueId,
reason: "Too late to cancel",
}),
).rejects.toMatchObject({ status: 409 });
const rows = await db
.select()
.from(heartbeatRunWatchdogDecisions)
.where(and(eq(heartbeatRunWatchdogDecisions.runId, runId), eq(heartbeatRunWatchdogDecisions.decision, "cancel_run")));
expect(rows).toHaveLength(0);
});
it("requires an evaluation issue for cancel_run decisions", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { managerId, runId } = await seedRunningRun({
now,
ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000,
});
const recovery = recoveryService(db, { enqueueWakeup: vi.fn() });
await expect(
recovery.recordWatchdogDecision({
runId,
actor: { type: "agent", agentId: managerId },
decision: "cancel_run",
reason: "No evaluation issue binding",
}),
).rejects.toMatchObject({ status: 400 });
const rows = await db
.select()
.from(heartbeatRunWatchdogDecisions)
.where(and(eq(heartbeatRunWatchdogDecisions.runId, runId), eq(heartbeatRunWatchdogDecisions.decision, "cancel_run")));
expect(rows).toHaveLength(0);
});
it("validates createdByRunId before storing watchdog decisions", async () => {
const now = new Date("2026-04-22T20:00:00.000Z");
const { companyId, managerId, runId } = await seedRunningRun({
+48 -4
View File
@@ -85,6 +85,9 @@ import { recoveryService } from "../services/recovery/service.js";
const RUN_LOG_DEFAULT_LIMIT_BYTES = 256_000;
const RUN_LOG_MAX_LIMIT_BYTES = 1024 * 1024;
const LIVE_RUNS_DEFAULT_LIMIT = 50;
const WATCHDOG_DECISIONS = ["snooze", "continue", "dismissed_false_positive", "cancel_run"] as const;
type WatchdogDecision = (typeof WATCHDOG_DECISIONS)[number];
function readRunLogLimitBytes(value: unknown) {
const parsed = Number(value ?? RUN_LOG_DEFAULT_LIMIT_BYTES);
@@ -2572,7 +2575,7 @@ export function agentRoutes(
assertCompanyAccess(req, companyId);
const minCount = readLiveRunsQueryInt(req.query.minCount, 50);
const limit = readLiveRunsQueryInt(req.query.limit, 50);
const limit = readLiveRunsQueryInt(req.query.limit, LIVE_RUNS_DEFAULT_LIMIT, LIVE_RUNS_DEFAULT_LIMIT);
const columns = {
id: heartbeatRuns.id,
@@ -2695,12 +2698,20 @@ export function agentRoutes(
}
assertCompanyAccess(req, existing.companyId);
const decision = typeof req.body?.decision === "string" ? req.body.decision : "";
if (!["snooze", "continue", "dismissed_false_positive"].includes(decision)) {
if (!WATCHDOG_DECISIONS.includes(decision as WatchdogDecision)) {
res.status(400).json({ error: "Unsupported watchdog decision" });
return;
}
const evaluationIssueId = typeof req.body?.evaluationIssueId === "string" ? req.body.evaluationIssueId : null;
const reason = typeof req.body?.reason === "string" ? req.body.reason.slice(0, 4000) : null;
if (decision === "cancel_run" && !reason?.trim()) {
res.status(400).json({ error: "Cancellation reason is required" });
return;
}
if (decision === "cancel_run" && !evaluationIssueId) {
res.status(400).json({ error: "evaluationIssueId is required for cancellation" });
return;
}
const snoozedUntil = decision === "snooze"
? new Date(String(req.body?.snoozedUntil ?? ""))
: null;
@@ -2712,14 +2723,47 @@ export function agentRoutes(
const row = await recovery.recordWatchdogDecision({
runId: existing.id,
actor: req.actor,
decision: decision as "snooze" | "continue" | "dismissed_false_positive",
decision: decision as WatchdogDecision,
evaluationIssueId,
reason,
snoozedUntil,
createdByRunId: req.actor.runId ?? null,
});
res.json(row);
if (decision !== "cancel_run") {
res.json(row);
return;
}
const cancelledRun = await heartbeat.cancelRun(existing.id, reason?.trim());
if (cancelledRun) {
await logActivity(db, {
companyId: cancelledRun.companyId,
actorType: req.actor.type === "agent" ? "agent" : "user",
actorId: req.actor.type === "agent"
? req.actor.agentId ?? "agent"
: req.actor.type === "board"
? req.actor.userId ?? "board"
: "unknown",
agentId: req.actor.type === "agent" ? req.actor.agentId ?? null : null,
runId: cancelledRun.id,
action: "heartbeat.watchdog_cancelled",
entityType: "heartbeat_run",
entityId: cancelledRun.id,
details: {
source: "routes.heartbeat_runs.watchdog_decisions",
decision,
evaluationIssueId,
reason: reason?.trim() ?? null,
createdByRunId: req.actor.runId ?? null,
targetRunId: cancelledRun.id,
previousStatus: existing.status,
resultStatus: cancelledRun.status,
},
});
}
res.json({ decision: row, run: cancelledRun });
});
router.get("/heartbeat-runs/:runId/events", async (req, res) => {
+1 -1
View File
@@ -7732,7 +7732,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return { checked, enqueued, skipped };
},
cancelRun: (runId: string) => cancelRunInternal(runId),
cancelRun: (runId: string, reason?: string) => cancelRunInternal(runId, reason),
cancelActiveForAgent: (agentId: string) => cancelActiveForAgentInternal(agentId),
+23 -5
View File
@@ -24,7 +24,7 @@ import {
} from "@paperclipai/db";
import { parseObject, asBoolean, asNumber } from "../../adapters/utils.js";
import { runningProcesses } from "../../adapters/index.js";
import { forbidden, notFound } from "../../errors.js";
import { badRequest, conflict, forbidden, notFound } from "../../errors.js";
import { logger } from "../../middleware/logger.js";
import { redactCurrentUserText } from "../../log-redaction.js";
import { redactSensitiveText } from "../../redaction.js";
@@ -64,6 +64,7 @@ import {
} from "../issue-execution-disposition.js";
const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
const CANCELLABLE_ACTIVE_RUN_RECOVERY_STATUSES = ["queued", "running", "scheduled_retry"] as const;
const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const;
export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000;
export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000;
@@ -112,6 +113,8 @@ type WatchdogDecisionActor =
| { type: "agent"; agentId?: string | null; runId?: string | null }
| { type: "none" };
type WatchdogDecision = "snooze" | "continue" | "dismissed_false_positive" | "cancel_run";
export type RunOutputSilenceSummary = {
lastOutputAt: Date | null;
lastOutputSeq: number;
@@ -1342,7 +1345,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
async function recordWatchdogDecision(input: {
runId: string;
actor: WatchdogDecisionActor;
decision: "snooze" | "continue" | "dismissed_false_positive";
decision: WatchdogDecision;
evaluationIssueId?: string | null;
reason?: string | null;
snoozedUntil?: Date | null;
@@ -1356,6 +1359,15 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
.limit(1);
if (!run) throw notFound("Heartbeat run not found");
const normalizedReason = typeof input.reason === "string" ? input.reason.trim() : "";
if (input.decision === "cancel_run") {
if (!normalizedReason) throw badRequest("Cancellation reason is required");
if (!input.evaluationIssueId) throw badRequest("evaluationIssueId is required for cancellation");
if (!CANCELLABLE_ACTIVE_RUN_RECOVERY_STATUSES.includes(run.status as (typeof CANCELLABLE_ACTIVE_RUN_RECOVERY_STATUSES)[number])) {
throw conflict("Heartbeat run is not cancellable", { status: run.status });
}
}
let evaluationIssue: {
id: string;
assigneeAgentId: string | null;
@@ -1442,7 +1454,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
evaluationIssueId: input.evaluationIssueId ?? null,
decision: input.decision,
snoozedUntil: effectiveSnoozedUntil,
reason: input.reason ?? null,
reason: input.decision === "cancel_run" ? normalizedReason : input.reason ?? null,
createdByAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null,
createdByUserId: input.actor.type === "board" ? input.actor.userId ?? null : null,
createdByRunId,
@@ -1459,15 +1471,21 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
: "unknown",
agentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null,
runId: run.id,
action: input.decision === "snooze" ? "heartbeat.watchdog_snoozed" : "heartbeat.watchdog_decision_recorded",
action: input.decision === "snooze"
? "heartbeat.watchdog_snoozed"
: input.decision === "cancel_run"
? "heartbeat.watchdog_cancel_requested"
: "heartbeat.watchdog_decision_recorded",
entityType: "heartbeat_run",
entityId: run.id,
details: {
source: "recovery.record_watchdog_decision",
decision: input.decision,
evaluationIssueId: input.evaluationIssueId ?? null,
targetRunId: run.id,
snoozedUntil: effectiveSnoozedUntil?.toISOString() ?? null,
reason: input.reason ?? null,
reason: input.decision === "cancel_run" ? normalizedReason : input.reason ?? null,
createdByRunId,
},
});
+1 -1
View File
@@ -60,7 +60,7 @@ export interface LiveRunForIssue {
export interface WatchdogDecisionInput {
runId: string;
decision: "snooze" | "continue" | "dismissed_false_positive";
decision: "snooze" | "continue" | "dismissed_false_positive" | "cancel_run";
evaluationIssueId?: string | null;
reason?: string | null;
snoozedUntil?: string | null;
+16
View File
@@ -411,6 +411,7 @@ describe("IssueRunLedger", () => {
expect(container.textContent).toContain("Stale-run watchdog alert");
expect(container.textContent).toContain("PAP-404");
expect(container.textContent).toContain("Stale run");
expect(container.textContent).toContain("assigned recovery owner can cancel this bound run");
const watchdogBanner = Array.from(container.querySelectorAll("p"))
.find((node) => node.textContent?.includes("Stale-run watchdog alert"))
?.closest("div");
@@ -429,6 +430,20 @@ describe("IssueRunLedger", () => {
decision: "continue",
evaluationIssueId: "issue-eval-1",
});
const cancelButton = Array.from(container.querySelectorAll("button")).find(
(button) => button.textContent?.includes("Cancel run"),
);
expect(cancelButton).not.toBeUndefined();
act(() => {
cancelButton?.dispatchEvent(new MouseEvent("click", { bubbles: true }));
});
expect(onWatchdogDecision).toHaveBeenCalledWith({
runId: "run-live-1",
decision: "cancel_run",
evaluationIssueId: "issue-eval-1",
reason: "Cancelled from stale-run recovery banner",
});
});
it("hides watchdog decision actions for known non-owner viewers", () => {
@@ -445,6 +460,7 @@ describe("IssueRunLedger", () => {
expect(container.textContent).not.toContain("Continue monitoring");
expect(container.textContent).not.toContain("Snooze 1h");
expect(container.textContent).not.toContain("Mark false positive");
expect(container.textContent).not.toContain("Cancel run");
expect(container.querySelectorAll("button")).toHaveLength(0);
expect(onWatchdogDecision).not.toHaveBeenCalled();
});
+19
View File
@@ -575,6 +575,9 @@ export function IssueRunLedgerContent({
{" "}for recovery context.
</>
) : null}
{latestSilentRun.outputSilence.level === "critical" ? (
<> The assigned recovery owner can cancel this bound run if the evidence shows it is stuck.</>
) : null}
</p>
{onWatchdogDecision && canRecordWatchdogDecisions ? (
<div className="mt-2 flex flex-wrap gap-1.5">
@@ -620,6 +623,22 @@ export function IssueRunLedgerContent({
>
Mark false positive
</button>
{latestSilentRun.outputSilence.level === "critical" ? (
<button
type="button"
className="rounded-md border border-red-500/40 bg-red-500/10 px-2 py-1 text-[11px] text-red-900 hover:bg-red-500/15 dark:text-red-200"
onClick={() =>
onWatchdogDecision({
runId: latestSilentRun.runId,
decision: "cancel_run",
evaluationIssueId: latestSilentRun.outputSilence?.evaluationIssueId ?? null,
reason: "Cancelled from stale-run recovery banner",
})}
disabled={pendingWatchdogDecision != null}
>
Cancel run
</button>
) : null}
</div>
) : null}
{watchdogDecisionError ? (