fix(watchdogs): serialize self-review reassignment

Lock active self watchdogs across dispatch and assignee mutations, reset the reviewed claim on ownership changes, and scope wake idempotency to the reviewer.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/task-watchdogs-scheduler.test.ts

@@ -22,6 +22,7 @@ import {
   getEmbeddedPostgresTestSupport,
   startEmbeddedPostgresTestDatabase,
 } from "./helpers/embedded-postgres.js";
+import { issueService } from "../services/issues.ts";
 import { taskWatchdogService } from "../services/task-watchdogs.ts";
 
 const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
@@ -428,6 +429,57 @@ describeEmbeddedPostgres("task watchdog scheduler", () => {
     expect(watchdog?.watchdogAgentId).toBe(currentAssigneeId);
   });
 
+  it("re-wakes the current assignee when reassignment overlaps self-review dispatch", async () => {
+    const companyId = await seedCompany();
+    const formerAssigneeId = await seedAgent(companyId, { name: "Former Assignee" });
+    const currentAssigneeId = await seedAgent(companyId, { name: "Current Assignee" });
+    const sourceId = await seedIssue(companyId, {
+      identifier: "WDOG-SELF-REASSIGN-RACE",
+      status: "done",
+      assigneeAgentId: formerAssigneeId,
+    });
+    await seedWatchdog(companyId, sourceId, formerAssigneeId, { mode: "self" });
+
+    let signalWakeStarted!: () => void;
+    const wakeStarted = new Promise<void>((resolve) => {
+      signalWakeStarted = resolve;
+    });
+    let releaseWake!: () => void;
+    const wakeCanFinish = new Promise<void>((resolve) => {
+      releaseWake = resolve;
+    });
+    const firstWakes: string[] = [];
+    const firstService = taskWatchdogService(db, {
+      enqueueWakeup: async (agentId) => {
+        firstWakes.push(agentId);
+        signalWakeStarted();
+        await wakeCanFinish;
+        return { id: randomUUID() };
+      },
+    });
+
+    const firstReconcile = firstService.reconcileTaskWatchdogs({ companyId });
+    await wakeStarted;
+    const reassignment = issueService(db).update(sourceId, { assigneeAgentId: currentAssigneeId });
+    releaseWake();
+    await Promise.all([firstReconcile, reassignment]);
+
+    expect(firstWakes).toEqual([formerAssigneeId]);
+    const [retargeted] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId));
+    expect(retargeted).toMatchObject({
+      watchdogAgentId: currentAssigneeId,
+      lastReviewedFingerprint: null,
+    });
+
+    const { service, wakes } = createService();
+    const second = await service.reconcileTaskWatchdogs({ companyId });
+
+    expect(second).toMatchObject({ checked: 1, triggered: 1 });
+    expect(wakes).toHaveLength(1);
+    expect(wakes[0]?.agentId).toBe(currentAssigneeId);
+    expect(wakes[0]?.opts?.idempotencyKey).toContain(currentAssigneeId);
+  });
+
   it("retries a self review when the wake was not queued", async () => {
     const companyId = await seedCompany();
     const agentId = await seedAgent(companyId);

server/src/services/issues.ts

@@ -6617,6 +6617,8 @@ export function issueService(db: Db) {
         issueData.assigneeAgentId !== undefined ? issueData.assigneeAgentId : existing.assigneeAgentId;
       const nextAssigneeUserId =
         issueData.assigneeUserId !== undefined ? issueData.assigneeUserId : existing.assigneeUserId;
+      const assigneeWillChange =
+        nextAssigneeAgentId !== existing.assigneeAgentId || nextAssigneeUserId !== existing.assigneeUserId;
 
       if (nextAssigneeAgentId && nextAssigneeUserId) {
         throw unprocessable("Issue can only have one assignee");
@@ -6708,10 +6710,7 @@ export function issueService(db: Db) {
         patch.executionAgentNameKey = null;
         patch.executionLockedAt = null;
       }
-      if (
-        (issueData.assigneeAgentId !== undefined && issueData.assigneeAgentId !== existing.assigneeAgentId) ||
-        (issueData.assigneeUserId !== undefined && issueData.assigneeUserId !== existing.assigneeUserId)
-      ) {
+      if (assigneeWillChange) {
         patch.checkoutRunId = null;
         patch.executionRunId = null;
         patch.executionAgentNameKey = null;
@@ -6719,6 +6718,22 @@ export function issueService(db: Db) {
       }
 
       const runUpdate = async (tx: any) => {
+        // Self-watchdog dispatch takes this row lock across wake creation and
+        // the reviewed-fingerprint claim. Taking it before changing the issue
+        // assignee closes the read/dispatch race in both directions.
+        const lockedSelfWatchdog = assigneeWillChange
+          ? await tx
+            .select({ id: issueWatchdogs.id })
+            .from(issueWatchdogs)
+            .where(and(
+              eq(issueWatchdogs.companyId, existing.companyId),
+              eq(issueWatchdogs.issueId, existing.id),
+              eq(issueWatchdogs.status, "active"),
+              eq(issueWatchdogs.mode, "self"),
+            ))
+            .for("update")
+            .then((rows: Array<{ id: string }>) => rows[0] ?? null)
+          : null;
         const defaultCompanyGoal = await getDefaultCompanyGoal(tx, existing.companyId);
         const [currentProjectGoalId, nextProjectGoalId] = await Promise.all([
           getProjectDefaultGoalId(tx, existing.companyId, existing.projectId),
@@ -6745,6 +6760,17 @@ export function issueService(db: Db) {
           .returning()
           .then((rows: Array<typeof issues.$inferSelect>) => rows[0] ?? null);
         if (!updated) return null;
+        if (lockedSelfWatchdog) {
+          await tx
+            .update(issueWatchdogs)
+            .set({
+              ...(nextAssigneeAgentId ? { watchdogAgentId: nextAssigneeAgentId } : {}),
+              lastReviewedFingerprint: null,
+              lastReviewedStopSnapshot: null,
+              updatedAt: new Date(),
+            })
+            .where(eq(issueWatchdogs.id, lockedSelfWatchdog.id));
+        }
         if (existing.status !== updated.status) {
           if (updated.status === "done" || updated.status === "cancelled") {
             await finalizeSummarySlotsForTerminalIssue(tx, updated);

server/src/services/task-watchdogs.ts

@@ -624,8 +624,12 @@ function taskWatchdogWakeIdempotencyKey(watchdogId: string, stopFingerprint: str
   return `task_watchdog:${watchdogId}:${stopFingerprint}`;
 }
 
-function taskWatchdogSelfWakeIdempotencyKey(watchdogId: string, stopFingerprint: string) {
-  return `task_watchdog_self:${watchdogId}:${stopFingerprint}`;
+function taskWatchdogSelfWakeIdempotencyKey(
+  watchdogId: string,
+  watchdogAgentId: string,
+  stopFingerprint: string,
+) {
+  return `task_watchdog_self:${watchdogId}:${watchdogAgentId}:${stopFingerprint}`;
 }
 
 function buildStoppedFingerprintComment(input: {
@@ -1543,127 +1547,158 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
     classification: Extract<TaskWatchdogClassifierResult, { state: "stopped" }>;
     runId: string | null;
   }) {
-    const { watchdog, sourceIssue, classification } = input;
-    const context = selfWatchdogWakeContext({ watchdog, sourceIssue, classification });
-    const idempotencyKey = taskWatchdogSelfWakeIdempotencyKey(watchdog.id, classification.stopFingerprint);
-    const wake = deps.enqueueWakeup
-      ? await deps.enqueueWakeup(watchdog.watchdogAgentId, {
-        source: "automation",
-        triggerDetail: "system",
-        reason: "task_watchdog_self_review",
-        payload: context,
-        contextSnapshot: context,
-        idempotencyKey,
-        requestedByActorType: "system",
-        requestedByActorId: null,
-      }).catch(async (error: unknown) => {
-        if (!isTaskWatchdogSelfWakeUniqueConflict(error)) throw error;
-        return db
-          .select({ id: agentWakeupRequests.runId })
-          .from(agentWakeupRequests)
-          .where(and(
-            eq(agentWakeupRequests.companyId, watchdog.companyId),
-            eq(agentWakeupRequests.idempotencyKey, idempotencyKey),
-            sql`${agentWakeupRequests.runId} is not null`,
-          ))
-          .then((rows) => rows[0]?.id ? { id: rows[0].id } : null);
-      })
-      : null;
-    if (!wake) {
-      return {
-        ...classification,
-        state: "wake_not_queued" as const,
-        reason: "Self-review wake was not queued.",
-      };
-    }
-
-    const now = new Date();
-    const claimed = await db.transaction(async (tx) => {
-      const rows = await tx
-        .update(issueWatchdogs)
-      .set({
-        lastObservedFingerprint: classification.stopFingerprint,
-        lastObservedStopSnapshot: classification.stopSnapshot,
-        lastReviewedFingerprint: classification.stopFingerprint,
-        lastReviewedStopSnapshot: classification.stopSnapshot,
-        lastTriggeredAt: now,
-        triggerCount: sql`${issueWatchdogs.triggerCount} + 1`,
-        updatedAt: now,
-      })
-      .where(and(
-        eq(issueWatchdogs.id, watchdog.id),
-        sql`${issueWatchdogs.lastReviewedFingerprint} is distinct from ${classification.stopFingerprint}`,
-      ))
-      .returning({ id: issueWatchdogs.id });
-      if (rows.length > 0) {
-        await logActivity(tx as unknown as Db, {
-          companyId: sourceIssue.companyId,
-          actorType: "system",
-          actorId: "system",
-          agentId: watchdog.watchdogAgentId,
-          runId: input.runId,
-          action: "issue.task_watchdog_triggered",
-          entityType: "issue",
-          entityId: sourceIssue.id,
-          details: {
-            source: "task_watchdogs.evaluate",
-            mode: "self",
-            watchdogId: watchdog.id,
-            stopFingerprint: classification.stopFingerprint,
-            stopSnapshot: classification.stopSnapshot,
-            stoppedLeaves: classification.stoppedLeaves,
-          },
-        });
+    const { classification } = input;
+    return db.transaction(async (tx) => {
+      // Assignee changes take this same row lock before updating the issue. It
+      // therefore covers the full read -> wake -> reviewed-claim interval and
+      // prevents a wake from being durably routed to an owner that changed in
+      // the middle of dispatch.
+      const lockedWatchdog = await tx
+        .select()
+        .from(issueWatchdogs)
+        .where(eq(issueWatchdogs.id, input.watchdog.id))
+        .for("update")
+        .then((rows) => rows[0] ?? null);
+      if (!lockedWatchdog || lockedWatchdog.status !== "active" || issueWatchdogMode(lockedWatchdog) !== "self") {
+        return { state: "skipped" as const, reason: "self_watchdog_not_active" };
       }
-      return rows;
-    });
 
-    await db.transaction(async (tx) => {
-      await tx.execute(sql`select pg_advisory_xact_lock(hashtext(${idempotencyKey}))`);
-      const exists = await tx
+      if (lockedWatchdog.lastReviewedFingerprint === classification.stopFingerprint) {
+        return {
+          ...classification,
+          state: "already_reviewed" as const,
+          reason: "The current stopped subtree fingerprint was already claimed for self review.",
+        };
+      }
+
+      const sourceIssue = await tx
+        .select()
+        .from(issues)
+        .where(and(
+          eq(issues.companyId, lockedWatchdog.companyId),
+          eq(issues.id, lockedWatchdog.issueId),
+          visibleIssueCondition(),
+        ))
+        .then((rows) => rows[0] ?? null);
+      if (!sourceIssue?.assigneeAgentId) {
+        return { state: "skipped" as const, reason: "self_watchdog_issue_unassigned" };
+      }
+
+      let watchdog = lockedWatchdog;
+      if (watchdog.watchdogAgentId !== sourceIssue.assigneeAgentId) {
+        await assertWatchdogAgentInvokable(tx as unknown as Db, watchdog.companyId, sourceIssue.assigneeAgentId);
+        watchdog = await tx
+          .update(issueWatchdogs)
+          .set({ watchdogAgentId: sourceIssue.assigneeAgentId, updatedAt: new Date() })
+          .where(eq(issueWatchdogs.id, watchdog.id))
+          .returning()
+          .then((rows) => rows[0] ?? watchdog);
+      }
+
+      const context = selfWatchdogWakeContext({ watchdog, sourceIssue, classification });
+      const idempotencyKey = taskWatchdogSelfWakeIdempotencyKey(
+        watchdog.id,
+        watchdog.watchdogAgentId,
+        classification.stopFingerprint,
+      );
+      const wake = deps.enqueueWakeup
+        ? await deps.enqueueWakeup(watchdog.watchdogAgentId, {
+          source: "automation",
+          triggerDetail: "system",
+          reason: "task_watchdog_self_review",
+          payload: context,
+          contextSnapshot: context,
+          idempotencyKey,
+          requestedByActorType: "system",
+          requestedByActorId: null,
+        }).catch(async (error: unknown) => {
+          if (!isTaskWatchdogSelfWakeUniqueConflict(error)) throw error;
+          return tx
+            .select({ id: agentWakeupRequests.runId })
+            .from(agentWakeupRequests)
+            .where(and(
+              eq(agentWakeupRequests.companyId, watchdog.companyId),
+              eq(agentWakeupRequests.idempotencyKey, idempotencyKey),
+              sql`${agentWakeupRequests.runId} is not null`,
+            ))
+            .then((rows) => rows[0]?.id ? { id: rows[0].id } : null);
+        })
+        : null;
+      if (!wake) {
+        return {
+          ...classification,
+          state: "wake_not_queued" as const,
+          reason: "Self-review wake was not queued.",
+        };
+      }
+
+      const now = new Date();
+      await tx
+        .update(issueWatchdogs)
+        .set({
+          lastObservedFingerprint: classification.stopFingerprint,
+          lastObservedStopSnapshot: classification.stopSnapshot,
+          lastReviewedFingerprint: classification.stopFingerprint,
+          lastReviewedStopSnapshot: classification.stopSnapshot,
+          lastTriggeredAt: now,
+          triggerCount: sql`${issueWatchdogs.triggerCount} + 1`,
+          updatedAt: now,
+        })
+        .where(eq(issueWatchdogs.id, watchdog.id));
+      await logActivity(tx as unknown as Db, {
+        companyId: sourceIssue.companyId,
+        actorType: "system",
+        actorId: "system",
+        agentId: watchdog.watchdogAgentId,
+        runId: input.runId,
+        action: "issue.task_watchdog_triggered",
+        entityType: "issue",
+        entityId: sourceIssue.id,
+        details: {
+          source: "task_watchdogs.evaluate",
+          mode: "self",
+          watchdogId: watchdog.id,
+          stopFingerprint: classification.stopFingerprint,
+          stopSnapshot: classification.stopSnapshot,
+          stoppedLeaves: classification.stoppedLeaves,
+        },
+      });
+
+      const commentExists = await tx
         .select({ body: issueComments.body })
         .from(issueComments)
         .where(eq(issueComments.issueId, sourceIssue.id))
         .then((rows) => rows.some((row) => row.body.includes(classification.stopFingerprint)));
-      if (exists) return;
-      await issuesSvc.addComment(
-        sourceIssue.id,
-        buildSelfReviewComment({
-          stopFingerprint: classification.stopFingerprint,
-          stoppedLeaves: classification.stoppedLeaves,
-          pendingInteractionsByIssueId: classification.pendingInteractionsByIssueId,
-          goalInstructions: watchdog.instructions,
-        }),
-        { runId: input.runId },
-        {
-          authorType: "system",
-          metadata: stoppedFingerprintMetadata({
-            sourceIssueId: sourceIssue.id,
+      if (!commentExists) {
+        await issuesSvc.addComment(
+          sourceIssue.id,
+          buildSelfReviewComment({
             stopFingerprint: classification.stopFingerprint,
-            waitsByIssueId: classification.stopSnapshot.waitsByIssueId,
-            resumed: true,
+            stoppedLeaves: classification.stoppedLeaves,
+            pendingInteractionsByIssueId: classification.pendingInteractionsByIssueId,
+            goalInstructions: watchdog.instructions,
           }),
-        },
-        tx,
-      );
-    });
+          { runId: input.runId },
+          {
+            authorType: "system",
+            metadata: stoppedFingerprintMetadata({
+              sourceIssueId: sourceIssue.id,
+              stopFingerprint: classification.stopFingerprint,
+              waitsByIssueId: classification.stopSnapshot.waitsByIssueId,
+              resumed: true,
+            }),
+          },
+          tx,
+        );
+      }
 
-    if (claimed.length === 0) {
       return {
-        ...classification,
-        state: "already_reviewed" as const,
-        reason: "The current stopped subtree fingerprint was already claimed for self review.",
+        state: "triggered" as const,
+        classification,
+        watchdogIssueId: sourceIssue.id,
+        wakeupRunId: wake.id,
       };
-    }
-
-
-
-    return {
-      state: "triggered" as const,
-      classification,
-      watchdogIssueId: sourceIssue.id,
-      wakeupRunId: wake.id,
-    };
+    });
   }
 
   async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null } = {}) {