From 2333d8fed0eda1f7d3ce530c984378d3dfb6c87a Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 28 Jul 2026 20:57:14 +0000 Subject: [PATCH] fix(watchdogs): preserve self-review wake retries Serialize self-review dispatch on the watchdog row, require a durable wake before marking the fingerprint reviewed, and roll back suppressed attempts so the same stopped state remains eligible for retry. Co-Authored-By: Paperclip --- .../task-watchdogs-scheduler.test.ts | 29 +++ server/src/services/task-watchdogs.ts | 185 ++++++++++-------- 2 files changed, 132 insertions(+), 82 deletions(-) diff --git a/server/src/__tests__/task-watchdogs-scheduler.test.ts b/server/src/__tests__/task-watchdogs-scheduler.test.ts index e828526f6e5..685183b3f6a 100644 --- a/server/src/__tests__/task-watchdogs-scheduler.test.ts +++ b/server/src/__tests__/task-watchdogs-scheduler.test.ts @@ -399,6 +399,35 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { expect(watchdog?.triggerCount).toBe(1); }); + it("retries a self review when the wake was not queued", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent(companyId); + const sourceId = await seedIssue(companyId, { + identifier: "WDOG-SELF-RETRY", + status: "done", + assigneeAgentId: agentId, + }); + await seedWatchdog(companyId, sourceId, agentId, { mode: "self" }); + const suppressedService = taskWatchdogService(db, { + enqueueWakeup: async () => null, + }); + + const suppressed = await suppressedService.reconcileTaskWatchdogs({ companyId }); + + expect(suppressed).toMatchObject({ checked: 1, triggered: 0, skipped: 1 }); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, sourceId))).toHaveLength(0); + const [unreviewed] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId)); + expect(unreviewed?.lastReviewedFingerprint).toBeNull(); + expect(unreviewed?.triggerCount).toBe(0); + + const { service, wakes } = createService(); + const retried = await service.reconcileTaskWatchdogs({ companyId }); + + expect(retried).toMatchObject({ checked: 1, triggered: 1 }); + expect(wakes).toHaveLength(1); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, sourceId))).toHaveLength(1); + }); + it("does not trigger while a non-watchdog descendant has live work", async () => { const companyId = await seedCompany(); const sourceId = await seedIssue(companyId, { identifier: "WDOG-2", status: "in_progress" }); diff --git a/server/src/services/task-watchdogs.ts b/server/src/services/task-watchdogs.ts index 4fb9de153c0..a2fe9b5f5c9 100644 --- a/server/src/services/task-watchdogs.ts +++ b/server/src/services/task-watchdogs.ts @@ -1527,90 +1527,111 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) runId: string | null; }) { const { watchdog, sourceIssue, classification } = input; - const now = new Date(); - const claimed = await db - .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 (claimed.length === 0) { - return { - ...classification, - state: "already_reviewed" as const, - reason: "The current stopped subtree fingerprint was already claimed for self review.", - }; - } - - 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, - stopFingerprint: classification.stopFingerprint, - waitsByIssueId: classification.stopSnapshot.waitsByIssueId, - resumed: true, - }), - }, - ); - - await logActivity(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 context = selfWatchdogWakeContext({ watchdog, sourceIssue, classification }); - const wake = deps.enqueueWakeup - ? await deps.enqueueWakeup(watchdog.watchdogAgentId, { - source: "automation", - triggerDetail: "system", - reason: "task_watchdog_self_review", - payload: context, - contextSnapshot: context, - idempotencyKey: taskWatchdogWakeIdempotencyKey(watchdog.id, classification.stopFingerprint), - requestedByActorType: "system", - requestedByActorId: null, - }) - : null; + const wakeNotQueued = new Error("Self-review wake was not queued."); - return { - state: "triggered" as const, - classification, - watchdogIssueId: sourceIssue.id, - wakeupRunId: wake?.id ?? null, - }; + try { + return await db.transaction(async (tx) => { + const txDb = tx as unknown as Db; + await tx.execute(sql`select id from issue_watchdogs where id = ${watchdog.id} for update`); + const persisted = await tx + .select({ lastReviewedFingerprint: issueWatchdogs.lastReviewedFingerprint }) + .from(issueWatchdogs) + .where(eq(issueWatchdogs.id, watchdog.id)) + .then((rows) => rows[0] ?? null); + if (!persisted || persisted.lastReviewedFingerprint === classification.stopFingerprint) { + return { + ...classification, + state: "already_reviewed" as const, + reason: "The current stopped subtree fingerprint was already claimed for self review.", + }; + } + + const wake = deps.enqueueWakeup + ? await deps.enqueueWakeup(watchdog.watchdogAgentId, { + source: "automation", + triggerDetail: "system", + reason: "task_watchdog_self_review", + payload: context, + contextSnapshot: context, + idempotencyKey: taskWatchdogWakeIdempotencyKey(watchdog.id, classification.stopFingerprint), + requestedByActorType: "system", + requestedByActorId: null, + }) + : null; + if (!wake) throw wakeNotQueued; + + 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, + stopFingerprint: classification.stopFingerprint, + waitsByIssueId: classification.stopSnapshot.waitsByIssueId, + resumed: true, + }), + }, + tx, + ); + + 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(txDb, { + 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, + }, + }); + + return { + state: "triggered" as const, + classification, + watchdogIssueId: sourceIssue.id, + wakeupRunId: wake.id, + }; + }); + } catch (error) { + if (error === wakeNotQueued) { + return { + ...classification, + state: "wake_not_queued" as const, + reason: wakeNotQueued.message, + }; + } + throw error; + } } async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null } = {}) {