From bf2cf1e380ee41a22ff334b7c3fdf9dd48a3f572 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 29 Jul 2026 19:25:02 +0000 Subject: [PATCH] fix(watchdogs): cancel running self reviews Cancel former-owner self-review runs through the heartbeat service while reassignment holds the watchdog lock, and fail closed when active work cannot be stopped. Co-Authored-By: Paperclip --- .../task-watchdogs-scheduler.test.ts | 18 +- server/src/routes/issues.ts | 4 +- server/src/services/issues.ts | 192 +++++++++++++----- 3 files changed, 159 insertions(+), 55 deletions(-) diff --git a/server/src/__tests__/task-watchdogs-scheduler.test.ts b/server/src/__tests__/task-watchdogs-scheduler.test.ts index 6b5be593354..2e53bf262d2 100644 --- a/server/src/__tests__/task-watchdogs-scheduler.test.ts +++ b/server/src/__tests__/task-watchdogs-scheduler.test.ts @@ -9,6 +9,7 @@ import { companies, createDb, documents, + heartbeatRunEvents, heartbeatRuns, issueComments, issueDocuments, @@ -23,6 +24,7 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { issueService } from "../services/issues.ts"; +import { heartbeatService } from "../services/heartbeat.ts"; import { taskWatchdogService } from "../services/task-watchdogs.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -52,6 +54,7 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { await db.delete(issueDocuments); await db.delete(documents); await db.delete(issueComments); + await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); await db.delete(agentWakeupRequests); await db.delete(issueWatchdogs); @@ -347,7 +350,9 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { expect(wakes).toHaveLength(1); expect(wakes[0]?.agentId).toBe(agentId); expect(wakes[0]?.opts?.reason).toBe("task_watchdog_self_review"); - expect(wakes[0]?.opts?.idempotencyKey).toMatch(/^task_watchdog_self:[^:]+:task_watchdog_stop:/); + expect(wakes[0]?.opts?.idempotencyKey).toMatch( + new RegExp(`^task_watchdog_self:[^:]+:${agentId}:task_watchdog_stop:`), + ); expect(wakes[0]?.opts?.contextSnapshot).toMatchObject({ issueId: sourceId, taskId: sourceId, @@ -461,7 +466,8 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { triggerDetail: opts?.triggerDetail, reason: opts?.reason, payload: opts?.payload, - status: "queued", + status: "running", + startedAt: new Date(), idempotencyKey: opts?.idempotencyKey, }) .returning(); @@ -472,7 +478,8 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { agentId, invocationSource: opts?.source ?? "automation", triggerDetail: opts?.triggerDetail, - status: "queued", + status: "running", + startedAt: new Date(), wakeupRequestId: wakeupRequest!.id, contextSnapshot: opts?.contextSnapshot, }) @@ -489,7 +496,10 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { const firstReconcile = firstService.reconcileTaskWatchdogs({ companyId }); await wakeStarted; - const reassignment = issueService(db).update(sourceId, { assigneeAgentId: currentAssigneeId }); + const heartbeat = heartbeatService(db); + const reassignment = issueService(db, { + cancelHeartbeatRun: heartbeat.cancelRun, + }).update(sourceId, { assigneeAgentId: currentAssigneeId }); releaseWake(); await Promise.all([firstReconcile, reassignment]); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index ddd40f28612..3f563d5eff8 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -2588,11 +2588,13 @@ export function issueRoutes( } = {}, ) { const router = Router(); - const svc = issueService(db); const access = accessService(db); const heartbeat = heartbeatService(db, { pluginWorkerManager: opts.pluginWorkerManager, }); + const svc = issueService(db, { + cancelHeartbeatRun: heartbeat.cancelRun, + }); const enqueueRecoveryActionWakeup = opts.recoveryActionEnqueueWakeup ?? heartbeat.wakeup; const feedback = feedbackService(db); const companiesSvc = companyService(db); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 89124bfd3d2..742cdf895de 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -118,6 +118,25 @@ export const ISSUE_LIST_MAX_LIMIT = 1000; export const ISSUE_BLOCKER_DIAGNOSTICS_MAX_BLOCKERS = 100; export const ISSUE_WAKE_DIAGNOSTICS_MAX_WAKE_REQUESTS = 50; export const ISSUE_WAKE_DIAGNOSTICS_MAX_ACTIVITY_RECORDS = 50; + +type IssueHeartbeatRunCancellationResult = { + id: string; + agentId: string; + status: string; + wakeupRequestId: string | null; +}; + +type IssueServiceOptions = { + cancelHeartbeatRun?: ( + runId: string, + reason: string, + options: { + errorCode: string; + eventMessage: string; + eventPayload: Record; + }, + ) => Promise; +}; export const ISSUE_WAKE_DIAGNOSTICS_LOOKBACK_DAYS = 14; export const ISSUE_SUBTREE_DIAGNOSTICS_MAX_DEPTH = 8; export const ISSUE_SUBTREE_DIAGNOSTICS_MAX_NODES = 100; @@ -3801,7 +3820,7 @@ async function countBlockedInboxIssues(dbOrTx: any, companyId: string, filters?: }, 0); } -export function issueService(db: Db) { +export function issueService(db: Db, options: IssueServiceOptions = {}) { const instanceSettings = instanceSettingsService(db); const treeControlSvc = issueTreeControlService(db); @@ -6734,6 +6753,127 @@ export function issueService(db: Db) { .for("update") .then((rows: Array<{ id: string }>) => rows[0] ?? null) : null; + const cancelledSelfReviewRuns: Array<{ + id: string; + agentId: string; + wakeupRequestId: string | null; + }> = []; + if (lockedSelfWatchdog) { + const cancelledAt = new Date(); + const cancellationReason = "Cancelled because the self-review issue was reassigned to a different owner"; + const cancellableRuns = await tx + .select({ + id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, existing.companyId), + inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), + sql`${heartbeatRuns.contextSnapshot} -> 'taskWatchdogSelfReview' ->> 'watchdogId' = ${lockedSelfWatchdog.id}`, + )); + + const hasRunningReview = cancellableRuns.some((run: { status: string }) => run.status === "running"); + if (hasRunningReview && !options.cancelHeartbeatRun) { + throw conflict("Cannot reassign an issue while its self-review run is active", { + issueId: existing.id, + watchdogId: lockedSelfWatchdog.id, + runIds: cancellableRuns + .filter((run: { status: string }) => run.status === "running") + .map((run: { id: string }) => run.id), + }); + } + + if (options.cancelHeartbeatRun) { + for (const run of cancellableRuns) { + const cancelled = await options.cancelHeartbeatRun(run.id, cancellationReason, { + errorCode: "issue_reassigned", + eventMessage: "self-review run cancelled for issue reassignment", + eventPayload: { + source: "self_watchdog_reassignment", + issueId: existing.id, + watchdogId: lockedSelfWatchdog.id, + previousAssigneeAgentId: existing.assigneeAgentId, + currentAssigneeAgentId: nextAssigneeAgentId, + }, + }); + if (cancelled?.status === "cancelled") { + cancelledSelfReviewRuns.push({ + id: cancelled.id, + agentId: cancelled.agentId, + wakeupRequestId: cancelled.wakeupRequestId, + }); + } + } + } else { + const directlyCancelledRuns = await tx + .update(heartbeatRuns) + .set({ + status: "cancelled", + finishedAt: cancelledAt, + error: cancellationReason, + errorCode: "issue_reassigned", + updatedAt: cancelledAt, + }) + .where(and( + eq(heartbeatRuns.companyId, existing.companyId), + inArray(heartbeatRuns.status, ["queued", "scheduled_retry"]), + sql`${heartbeatRuns.contextSnapshot} -> 'taskWatchdogSelfReview' ->> 'watchdogId' = ${lockedSelfWatchdog.id}`, + )) + .returning({ + id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + wakeupRequestId: heartbeatRuns.wakeupRequestId, + }); + cancelledSelfReviewRuns.push(...directlyCancelledRuns); + } + + const remainingCancellableRuns = await tx + .select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, existing.companyId), + inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), + sql`${heartbeatRuns.contextSnapshot} -> 'taskWatchdogSelfReview' ->> 'watchdogId' = ${lockedSelfWatchdog.id}`, + )); + if (remainingCancellableRuns.length > 0) { + throw conflict("Cannot reassign an issue while its self-review work is still active", { + issueId: existing.id, + watchdogId: lockedSelfWatchdog.id, + runs: remainingCancellableRuns, + }); + } + + const wakeupRequestIds = cancelledSelfReviewRuns + .map((run) => run.wakeupRequestId) + .filter((requestId): requestId is string => Boolean(requestId)); + if (wakeupRequestIds.length > 0) { + await tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: cancelledAt, + error: cancellationReason, + updatedAt: cancelledAt, + }) + .where(inArray(agentWakeupRequests.id, wakeupRequestIds)); + } + await tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: cancelledAt, + error: cancellationReason, + updatedAt: cancelledAt, + }) + .where(and( + eq(agentWakeupRequests.companyId, existing.companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload} -> '_paperclipWakeContext' -> 'taskWatchdogSelfReview' ->> 'watchdogId' = ${lockedSelfWatchdog.id}`, + )); + } const defaultCompanyGoal = await getDefaultCompanyGoal(tx, existing.companyId); const [currentProjectGoalId, nextProjectGoalId] = await Promise.all([ getProjectDefaultGoalId(tx, existing.companyId, existing.projectId), @@ -6761,55 +6901,7 @@ export function issueService(db: Db) { .then((rows: Array) => rows[0] ?? null); if (!updated) return null; if (lockedSelfWatchdog) { - const cancelledAt = new Date(); - const cancellationReason = "Cancelled because the self-review issue was reassigned before the run started"; - const cancelledRuns = await tx - .update(heartbeatRuns) - .set({ - status: "cancelled", - finishedAt: cancelledAt, - error: cancellationReason, - errorCode: "issue_reassigned", - updatedAt: cancelledAt, - }) - .where(and( - eq(heartbeatRuns.companyId, existing.companyId), - inArray(heartbeatRuns.status, ["queued", "scheduled_retry"]), - sql`${heartbeatRuns.contextSnapshot} -> 'taskWatchdogSelfReview' ->> 'watchdogId' = ${lockedSelfWatchdog.id}`, - )) - .returning({ - id: heartbeatRuns.id, - agentId: heartbeatRuns.agentId, - wakeupRequestId: heartbeatRuns.wakeupRequestId, - }); - const wakeupRequestIds = cancelledRuns - .map((run: { wakeupRequestId: string | null }) => run.wakeupRequestId) - .filter((requestId: string | null): requestId is string => Boolean(requestId)); - if (wakeupRequestIds.length > 0) { - await tx - .update(agentWakeupRequests) - .set({ - status: "cancelled", - finishedAt: cancelledAt, - error: cancellationReason, - updatedAt: cancelledAt, - }) - .where(inArray(agentWakeupRequests.id, wakeupRequestIds)); - } - await tx - .update(agentWakeupRequests) - .set({ - status: "cancelled", - finishedAt: cancelledAt, - error: cancellationReason, - updatedAt: cancelledAt, - }) - .where(and( - eq(agentWakeupRequests.companyId, existing.companyId), - eq(agentWakeupRequests.status, "deferred_issue_execution"), - sql`${agentWakeupRequests.payload} -> '_paperclipWakeContext' -> 'taskWatchdogSelfReview' ->> 'watchdogId' = ${lockedSelfWatchdog.id}`, - )); - for (const cancelledRun of cancelledRuns) { + for (const cancelledRun of cancelledSelfReviewRuns) { await logActivity(tx as unknown as Db, { companyId: existing.companyId, actorType: actorAgentId ? "agent" : actorUserId ? "user" : "system",