fix(watchdogs): atomically claim self reviews

Prevent concurrent watchdog evaluations from duplicating comments, trigger
bookkeeping, activity, and wake requests for the same stopped fingerprint.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-07-29 19:07:29 +00:00
1 parent 7b4722e26c
commit 3a5e75425b
2 files changed
+53 -14

No files matched your search

@@ -371,6 +371,34 @@ describeEmbeddedPostgres("task watchdog scheduler", () => {
expect(commentsAfter).toHaveLength(1);
});
it("self mode atomically claims a stopped fingerprint across concurrent reconciliations", async () => {
const companyId = await seedCompany();
const agentId = await seedAgent(companyId);
const sourceId = await seedIssue(companyId, {
identifier: "WDOG-SELF-RACE",
status: "done",
assigneeAgentId: agentId,
});
await seedWatchdog(companyId, sourceId, agentId, {
mode: "self",
instructions: "Verify the report exists and passes lint.",
});
const { service, wakes } = createService();
const results = await Promise.all([
service.reconcileTaskWatchdogs({ companyId }),
service.reconcileTaskWatchdogs({ companyId }),
]);
expect(results.reduce((total, result) => total + result.triggered, 0)).toBe(1);
expect(results.reduce((total, result) => total + result.alreadyReviewed, 0)).toBe(1);
expect(wakes).toHaveLength(1);
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, sourceId));
expect(comments).toHaveLength(1);
const [watchdog] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId));
expect(watchdog?.triggerCount).toBe(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" });
+25 -14
View File
@@ -1527,6 +1527,31 @@ 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({
@@ -1547,20 +1572,6 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
},
);
const now = new Date();
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(eq(issueWatchdogs.id, watchdog.id));
await logActivity(db, {
companyId: sourceIssue.companyId,
actorType: "system",