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 <noreply@paperclip.ing>

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]);
 

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);

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<string, unknown>;
+    },
+  ) => Promise<IssueHeartbeatRunCancellationResult | null>;
+};
 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<typeof issues.$inferSelect>) => 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",