From b0f53888a2298a7474e2c89bb7eb3abba77b5cbb Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 29 Jul 2026 18:50:46 +0000 Subject: [PATCH] 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 --- .../task-watchdogs-scheduler.test.ts | 52 ++++ server/src/services/issues.ts | 34 ++- server/src/services/task-watchdogs.ts | 259 ++++++++++-------- 3 files changed, 229 insertions(+), 116 deletions(-) diff --git a/server/src/__tests__/task-watchdogs-scheduler.test.ts b/server/src/__tests__/task-watchdogs-scheduler.test.ts index f74f1a4fba6..b71d34988a6 100644 --- a/server/src/__tests__/task-watchdogs-scheduler.test.ts +++ b/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((resolve) => { + signalWakeStarted = resolve; + }); + let releaseWake!: () => void; + const wakeCanFinish = new Promise((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); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index b5e6811cf30..ab5a0763497 100644 --- a/server/src/services/issues.ts +++ b/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) => 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); diff --git a/server/src/services/task-watchdogs.ts b/server/src/services/task-watchdogs.ts index 0c599a350b8..5701f0cc6d8 100644 --- a/server/src/services/task-watchdogs.ts +++ b/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; 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 } = {}) {