From c481be44e30e71e034591375491a32fc4da2f9ae Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 25 Jul 2026 08:34:41 -0500 Subject: [PATCH] fix(task-watchdogs): deduplicate unchanged stopped-state wakes (#10207) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Task watchdogs review issue subtrees when no run or queued wake keeps work live > - Pending human interactions and approvals are valid stopped states that still need one watchdog review > - The existing fingerprint included volatile activity timestamps, so unchanged stopped trees could wake repeatedly after comments, documents, work products, or sibling completions > - This pull request fingerprints only review-material leaf and wait state, persists the reviewed snapshot, and suppresses shrink-only repeats > - The benefit is one review per materially new stopped state without weakening liveness classification or hiding human waits ## Linked Issues or Issue Description ### What happened? Task-watchdog stop fingerprints changed for metadata-only activity and completed siblings, producing duplicate wakes after an unchanged stop had already been reviewed. ### Expected behavior Pending interactions and approvals remain classified as stopped, but a reviewed stopped state only wakes again when waits, non-terminal leaves, status, assignment, or blockers gain material changes. ### Steps to reproduce 1. Review a stopped watched subtree with a pending human wait or multiple non-terminal stopped leaves. 2. Add only comment/document/work-product activity, or complete one stopped sibling without changing the wait set. 3. Observe a duplicate wake from the timestamp-heavy fingerprint. Related public work: Refs #9452 for overlapping task-watchdog service edits and #10043 for related no-op fingerprint suppression. ## What Changed - Added fingerprint v2 over non-terminal material leaves plus subtree-wide pending wait ids, excluding volatile timestamps while retaining them in wake context. - Added nullable observed/reviewed JSONB stop snapshots and shrink-only reviewed-state suppression with legacy exact-fingerprint fallback. - Added pending interaction kinds and approval ids to watchdog wake context, review comments, and comment metadata. - Added classifier and scheduler coverage for waiting-leaf liveness, metadata stability, sibling shrink suppression, material changes, snapshot promotion, legacy rows, and unchanged idempotency keys. ## Verification - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/task-watchdogs-classifier.test.ts src/__tests__/task-watchdogs-scheduler.test.ts` — 2 files, 36 tests passed. - `pnpm --filter @paperclipai/server typecheck` — passed. - `git diff --check origin/master...HEAD` — passed. ## Risks - Fingerprint version 2 intentionally re-fingerprints every currently stopped watched tree once after deployment, causing a one-time wake burst before the new reviewed snapshots are established. - Migration `0191_task_watchdog_stop_snapshots.sql` only adds two nullable JSONB columns with no backfill; legacy rows continue exact-fingerprint behavior until a post-deploy review promotes a snapshot. - PR #9452 edits the same service file; whichever lands second may need a trivial rebase. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex, exact model ID `gpt-5.6-sol`, high reasoning mode, with repository tool use and code execution. The runtime did not expose a context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../0192_task_watchdog_stop_snapshots.sql | 2 + packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/issue_watchdogs.ts | 4 +- .../task-watchdogs-classifier.test.ts | 187 +++++++++++++++++- .../task-watchdogs-scheduler.test.ts | 132 ++++++++++++- server/src/routes/issues.ts | 7 + server/src/services/task-watchdogs.ts | 186 +++++++++++++++-- 7 files changed, 507 insertions(+), 18 deletions(-) create mode 100644 packages/db/src/migrations/0192_task_watchdog_stop_snapshots.sql diff --git a/packages/db/src/migrations/0192_task_watchdog_stop_snapshots.sql b/packages/db/src/migrations/0192_task_watchdog_stop_snapshots.sql new file mode 100644 index 0000000000..4fbd8c4d39 --- /dev/null +++ b/packages/db/src/migrations/0192_task_watchdog_stop_snapshots.sql @@ -0,0 +1,2 @@ +ALTER TABLE "issue_watchdogs" ADD COLUMN IF NOT EXISTS "last_observed_stop_snapshot" jsonb;--> statement-breakpoint +ALTER TABLE "issue_watchdogs" ADD COLUMN IF NOT EXISTS "last_reviewed_stop_snapshot" jsonb; diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 04b70adf89..c99cd8db4e 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -1331,6 +1331,13 @@ "when": 1784916885227, "tag": "0191_status_card_mentioned_issue_ids", "breakpoints": true + }, + { + "idx": 192, + "version": "7", + "when": 1784916886226, + "tag": "0192_task_watchdog_stop_snapshots", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/issue_watchdogs.ts b/packages/db/src/schema/issue_watchdogs.ts index f3863a87ba..ca5c1aa87c 100644 --- a/packages/db/src/schema/issue_watchdogs.ts +++ b/packages/db/src/schema/issue_watchdogs.ts @@ -1,5 +1,5 @@ import { sql } from "drizzle-orm"; -import { index, integer, pgTable, text, timestamp, uniqueIndex, uuid } from "drizzle-orm/pg-core"; +import { index, integer, jsonb, pgTable, text, timestamp, uniqueIndex, uuid } from "drizzle-orm/pg-core"; import { agents } from "./agents.js"; import { companies } from "./companies.js"; import { heartbeatRuns } from "./heartbeat_runs.js"; @@ -17,6 +17,8 @@ export const issueWatchdogs = pgTable( watchdogIssueId: uuid("watchdog_issue_id").references(() => issues.id, { onDelete: "set null" }), lastObservedFingerprint: text("last_observed_fingerprint"), lastReviewedFingerprint: text("last_reviewed_fingerprint"), + lastObservedStopSnapshot: jsonb("last_observed_stop_snapshot"), + lastReviewedStopSnapshot: jsonb("last_reviewed_stop_snapshot"), lastTriggeredAt: timestamp("last_triggered_at", { withTimezone: true }), lastCompletedAt: timestamp("last_completed_at", { withTimezone: true }), triggerCount: integer("trigger_count").notNull().default(0), diff --git a/server/src/__tests__/task-watchdogs-classifier.test.ts b/server/src/__tests__/task-watchdogs-classifier.test.ts index aca41e096d..b44d0c76c5 100644 --- a/server/src/__tests__/task-watchdogs-classifier.test.ts +++ b/server/src/__tests__/task-watchdogs-classifier.test.ts @@ -56,7 +56,13 @@ describe("task watchdog subtree classifier", () => { issue({ status: "done" }), issue({ id: childId, identifier: "PAP-2", parentId: sourceId, status: "in_review" }), ], - pendingInteractions: [{ companyId, issueId: childId, id: "interaction-1", status: "pending" }], + pendingInteractions: [{ + companyId, + issueId: childId, + id: "interaction-1", + kind: "request_confirmation", + status: "pending", + }], }); expect(result.state).toBe("stopped"); @@ -69,6 +75,185 @@ describe("task watchdog subtree classifier", () => { pendingInteractionIds: ["interaction-1"], }), ]); + expect(result.stopSnapshot.waitsByIssueId).toEqual({ + [childId]: { + pendingInteractionIds: ["interaction-1"], + pendingApprovalIds: [], + }, + }); + expect(result.pendingInteractionsByIssueId).toEqual({ + [childId]: [{ id: "interaction-1", kind: "request_confirmation" }], + }); + }); + + it("keeps the material fingerprint stable across metadata-only ticks", () => { + const initial = classify({ + issues: [issue({ + status: "blocked", + latestCommentAt: "2026-06-17T20:01:00.000Z", + latestDocumentAt: "2026-06-17T20:02:00.000Z", + latestWorkProductAt: "2026-06-17T20:03:00.000Z", + })], + }); + const ticked = classify({ + issues: [issue({ + status: "blocked", + updatedAt: "2026-06-18T20:00:00.000Z", + latestCommentAt: "2026-06-18T20:01:00.000Z", + latestDocumentAt: "2026-06-18T20:02:00.000Z", + latestWorkProductAt: "2026-06-18T20:03:00.000Z", + })], + }); + + expect(initial.state).toBe("stopped"); + expect(ticked.state).toBe("stopped"); + if (initial.state !== "stopped" || ticked.state !== "stopped") return; + expect(ticked.stopFingerprint).toBe(initial.stopFingerprint); + expect(ticked.stoppedLeaves[0]?.updatedAt).not.toBe(initial.stoppedLeaves[0]?.updatedAt); + }); + + it("suppresses a shrink-only stopped state after a sibling completes", () => { + const siblingId = "child-2"; + const initial = classify({ + issues: [ + issue({ status: "in_progress" }), + issue({ id: childId, parentId: sourceId, status: "in_review" }), + issue({ id: siblingId, parentId: sourceId, status: "blocked" }), + ], + pendingInteractions: [{ companyId, issueId: childId, id: "interaction-1", status: "pending" }], + }); + expect(initial.state).toBe("stopped"); + if (initial.state !== "stopped") return; + + const shrunk = classify({ + watchdog: { + companyId, + issueId: sourceId, + lastReviewedFingerprint: initial.stopFingerprint, + lastReviewedStopSnapshot: initial.stopSnapshot, + }, + issues: [ + issue({ status: "in_progress" }), + issue({ id: childId, parentId: sourceId, status: "in_review" }), + issue({ id: siblingId, parentId: sourceId, status: "done" }), + ], + pendingInteractions: [{ companyId, issueId: childId, id: "interaction-1", status: "pending" }], + }); + + expect(shrunk.state).toBe("already_reviewed"); + if (shrunk.state !== "already_reviewed") return; + expect(shrunk.stopFingerprint).not.toBe(initial.stopFingerprint); + }); + + it("uses exact-hash behavior when the legacy reviewed snapshot is null", () => { + const siblingId = "child-2"; + const initial = classify({ + issues: [ + issue({ status: "in_progress" }), + issue({ id: childId, parentId: sourceId, status: "blocked" }), + issue({ id: siblingId, parentId: sourceId, status: "blocked" }), + ], + }); + expect(initial.state).toBe("stopped"); + if (initial.state !== "stopped") return; + + const changed = classify({ + watchdog: { + companyId, + issueId: sourceId, + lastReviewedFingerprint: initial.stopFingerprint, + lastReviewedStopSnapshot: null, + }, + issues: [ + issue({ status: "in_progress" }), + issue({ id: childId, parentId: sourceId, status: "blocked" }), + issue({ id: siblingId, parentId: sourceId, status: "done" }), + ], + }); + + expect(changed.state).toBe("stopped"); + }); + + it("includes waits on non-leaf issues in the material snapshot", () => { + const initial = classify({ + issues: [ + issue({ status: "in_review" }), + issue({ id: childId, parentId: sourceId, status: "blocked" }), + ], + pendingApprovals: [{ companyId, issueId: sourceId, id: "approval-1", status: "pending" }], + }); + + expect(initial.state).toBe("stopped"); + if (initial.state !== "stopped") return; + expect(initial.stopSnapshot.waitsByIssueId).toEqual({ + [sourceId]: { + pendingInteractionIds: [], + pendingApprovalIds: ["approval-1"], + }, + }); + + const changed = classify({ + watchdog: { + companyId, + issueId: sourceId, + lastReviewedFingerprint: initial.stopFingerprint, + lastReviewedStopSnapshot: initial.stopSnapshot, + }, + issues: [ + issue({ status: "in_review" }), + issue({ id: childId, parentId: sourceId, status: "blocked" }), + ], + pendingApprovals: [{ companyId, issueId: sourceId, id: "approval-2", status: "pending" }], + }); + + expect(changed.state).toBe("stopped"); + }); + + it.each([ + ["wait set", { + pendingInteractions: [{ companyId, issueId: childId, id: "interaction-2", status: "pending" }], + }], + ["leaf status", { + issues: [issue({ status: "in_progress" }), issue({ id: childId, parentId: sourceId, status: "todo" })], + }], + ["leaf assignee", { + issues: [ + issue({ status: "in_progress" }), + issue({ id: childId, parentId: sourceId, status: "blocked", assigneeAgentId: "agent-2" }), + ], + }], + ["leaf blocker", { + blockers: [{ companyId, blockedIssueId: childId, blockerIssueId: "blocker-2" }], + }], + ["new stopped leaf", { + issues: [ + issue({ status: "in_progress" }), + issue({ id: childId, parentId: sourceId, status: "blocked" }), + issue({ id: "child-2", parentId: sourceId, status: "blocked" }), + ], + }], + ])("triggers a fresh stop when the %s changes", (_label, overrides) => { + const baseInput = { + issues: [issue({ status: "in_progress" }), issue({ id: childId, parentId: sourceId, status: "blocked" })], + pendingInteractions: [{ companyId, issueId: childId, id: "interaction-1", status: "pending" }], + blockers: [{ companyId, blockedIssueId: childId, blockerIssueId: "blocker-1" }], + }; + const initial = classify(baseInput); + expect(initial.state).toBe("stopped"); + if (initial.state !== "stopped") return; + + const changed = classify({ + ...baseInput, + ...overrides, + watchdog: { + companyId, + issueId: sourceId, + lastReviewedFingerprint: initial.stopFingerprint, + lastReviewedStopSnapshot: initial.stopSnapshot, + }, + }); + + expect(changed.state).toBe("stopped"); }); it("suppresses an unchanged stopped fingerprint once the watchdog reviewed it", () => { diff --git a/server/src/__tests__/task-watchdogs-scheduler.test.ts b/server/src/__tests__/task-watchdogs-scheduler.test.ts index 325d45ff63..9c407002a5 100644 --- a/server/src/__tests__/task-watchdogs-scheduler.test.ts +++ b/server/src/__tests__/task-watchdogs-scheduler.test.ts @@ -5,12 +5,15 @@ import { activityLog, agentWakeupRequests, agents, + approvals, companies, createDb, documents, heartbeatRuns, issueComments, issueDocuments, + issueApprovals, + issueThreadInteractions, issueWorkProducts, issues, issueWatchdogs, @@ -41,6 +44,9 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { afterEach(async () => { await db.delete(activityLog); + await db.delete(issueApprovals); + await db.delete(approvals); + await db.delete(issueThreadInteractions); await db.delete(issueWorkProducts); await db.delete(issueDocuments); await db.delete(documents); @@ -173,6 +179,7 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { expect(wakes).toHaveLength(1); expect(wakes[0]?.agentId).toBe(agentId); expect(wakes[0]?.opts?.reason).toBe("task_watchdog_stopped_subtree"); + expect(wakes[0]?.opts?.idempotencyKey).toMatch(/^task_watchdog:[^:]+:task_watchdog_stop:/); expect(wakes[0]?.opts?.contextSnapshot).toMatchObject({ taskWatchdog: { watchedIssueId: sourceId, @@ -213,6 +220,12 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { const [watchdog] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId)); expect(watchdog?.watchdogIssueId).toBe(watchdogIssues[0]?.id); expect(watchdog?.lastObservedFingerprint).toMatch(/^task_watchdog_stop:/); + expect(watchdog?.lastObservedStopSnapshot).toMatchObject({ + version: 2, + fingerprint: watchdog?.lastObservedFingerprint, + materialLeaves: [], + waitsByIssueId: {}, + }); expect(watchdog?.triggerCount).toBe(1); }); @@ -408,6 +421,7 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { expect(reviewed).toMatchObject({ checked: 1, triggered: 0, alreadyReviewed: 1 }); const [reviewedWatchdog] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId)); expect(reviewedWatchdog?.lastReviewedFingerprint).toBe(firstWatchdog?.lastObservedFingerprint); + expect(reviewedWatchdog?.lastReviewedStopSnapshot).toEqual(firstWatchdog?.lastObservedStopSnapshot); await db .update(issues) @@ -430,6 +444,50 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { expect(wakes.length).toBe(2); }); + it("suppresses a shrink-only stop after review when the snapshot round-trips through jsonb", async () => { + const companyId = await seedCompany(); + const sourceId = await seedIssue(companyId, { identifier: "WDOG-SHRINK", status: "in_review" }); + const waitingLeafId = await seedIssue(companyId, { parentId: sourceId, status: "in_review" }); + const siblingLeafId = await seedIssue(companyId, { parentId: sourceId, status: "in_progress" }); + const agentId = await seedAgent(companyId); + await db.insert(issueThreadInteractions).values({ + id: randomUUID(), + companyId, + issueId: waitingLeafId, + kind: "request_confirmation", + status: "pending", + payload: { version: 1, prompt: "Confirm the stop." }, + createdByAgentId: agentId, + }); + await seedWatchdog(companyId, sourceId, agentId); + const { service, wakes } = createService(); + + const first = await service.reconcileTaskWatchdogs({ companyId }); + expect(first).toMatchObject({ checked: 1, triggered: 1 }); + + const [triggeredWatchdog] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId)); + await db + .update(issues) + .set({ status: "done", updatedAt: new Date() }) + .where(eq(issues.id, triggeredWatchdog!.watchdogIssueId!)); + const reviewed = await service.reconcileTaskWatchdogs({ companyId }); + expect(reviewed).toMatchObject({ checked: 1, triggered: 0, alreadyReviewed: 1 }); + const [reviewedWatchdog] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId)); + expect(reviewedWatchdog?.lastReviewedStopSnapshot).not.toBeNull(); + + // The sibling completing shrinks the material leaf set while the wait set + // is unchanged; the reviewed snapshot loaded back from jsonb (which does + // not preserve object key order) must still suppress the wake. + await db + .update(issues) + .set({ status: "done", updatedAt: new Date(Date.now() + 60_000) }) + .where(eq(issues.id, siblingLeafId)); + const afterShrink = await service.reconcileTaskWatchdogs({ companyId }); + + expect(afterShrink).toMatchObject({ checked: 1, triggered: 0, alreadyReviewed: 1 }); + expect(wakes).toHaveLength(1); + }); + it("does not let an old terminal watchdog review mark a newer observed fingerprint reviewed", async () => { const companyId = await seedCompany(); const sourceId = await seedIssue(companyId, { identifier: "WDOG-STALE", status: "done" }); @@ -473,6 +531,7 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { const [reviewedWatchdog] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId)); expect(reviewedWatchdog?.lastReviewedFingerprint).toBe(oldFingerprint); expect(reviewedWatchdog?.lastReviewedFingerprint).not.toBe(newerFingerprint); + expect(reviewedWatchdog?.lastReviewedStopSnapshot).toBeNull(); const [reopenedWatchdogIssue] = await db.select().from(issues).where(eq(issues.id, watchdogIssueId)); expect(reopenedWatchdogIssue).toMatchObject({ status: "todo", @@ -490,7 +549,7 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { expect(wakes.length).toBe(2); }); - it("revalidates stale watchdog reviews against current source evidence before allowing mutations", async () => { + it("keeps watchdog mutation scope valid across metadata-only source evidence", async () => { const companyId = await seedCompany(); const sourceId = await seedIssue(companyId, { identifier: "WDOG-REVALIDATE", status: "blocked" }); const agentId = await seedAgent(companyId); @@ -522,11 +581,10 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { stopFingerprint: originalFingerprint, }); - expect(revalidated.allowed).toBe(false); - expect(revalidated.reason).toContain("stop fingerprint changed"); + expect(revalidated.allowed).toBe(true); expect(revalidated.classification?.state).toBe("stopped"); if (revalidated.classification?.state !== "stopped") throw new Error("Expected stopped classification"); - expect(revalidated.classification.stopFingerprint).not.toBe(originalFingerprint); + expect(revalidated.classification.stopFingerprint).toBe(originalFingerprint); expect(revalidated.classification.stoppedLeaves[0]).toMatchObject({ latestCommentAt: later.toISOString(), latestDocumentAt: new Date(later.getTime() + 1_000).toISOString(), @@ -534,6 +592,72 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { }); }); + it("surfaces pending interaction kinds and approval ids in the wake and watchdog comment", async () => { + const companyId = await seedCompany(); + const sourceId = await seedIssue(companyId, { identifier: "WDOG-WAITS", status: "in_review" }); + const agentId = await seedAgent(companyId); + await seedWatchdog(companyId, sourceId, agentId); + const interactionId = randomUUID(); + const approvalId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: interactionId, + companyId, + issueId: sourceId, + kind: "request_confirmation", + status: "pending", + payload: { version: 1, prompt: "Confirm the reviewed stop." }, + createdByAgentId: agentId, + }); + await db.insert(approvals).values({ + id: approvalId, + companyId, + type: "request_board_approval", + requestedByAgentId: agentId, + status: "pending", + payload: { summary: "Approve the reviewed stop." }, + }); + await db.insert(issueApprovals).values({ + companyId, + issueId: sourceId, + approvalId, + linkedByAgentId: agentId, + }); + const { service, wakes } = createService(); + + const result = await service.reconcileTaskWatchdogs({ companyId }); + + expect(result).toMatchObject({ checked: 1, triggered: 1 }); + expect(wakes[0]?.opts?.contextSnapshot).toMatchObject({ + taskWatchdog: { + pendingInteractions: { + [sourceId]: [{ id: interactionId, kind: "request_confirmation" }], + }, + pendingApprovals: { + [sourceId]: [approvalId], + }, + }, + }); + const [watchdog] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.issueId, sourceId)); + expect(watchdog?.lastObservedStopSnapshot).toMatchObject({ + waitsByIssueId: { + [sourceId]: { + pendingInteractionIds: [interactionId], + pendingApprovalIds: [approvalId], + }, + }, + }); + const comments = await db + .select() + .from(issueComments) + .where(eq(issueComments.issueId, watchdog!.watchdogIssueId!)); + expect(comments.at(-1)?.body).toContain(`pending request_confirmation ${interactionId.slice(0, 8)}…`); + expect(comments.at(-1)?.body).toContain(`approval ${approvalId.slice(0, 8)}…`); + const metadata = comments.at(-1)?.metadata as { sections?: Array<{ rows?: unknown[] }> } | null; + expect(metadata?.sections?.[0]?.rows).toEqual(expect.arrayContaining([ + expect.objectContaining({ label: "Pending waits", text: "2" }), + ])); + }); + it("revalidates a stale watchdog review as live when the source gets a fresh run path", async () => { const companyId = await seedCompany(); const sourceId = await seedIssue(companyId, { identifier: "WDOG-LIVE-REVALIDATE", status: "blocked" }); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 36d74da9c3..6af7134cda 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -349,6 +349,13 @@ function noopTaskWatchdogService(): TaskWatchdogService { includedIssueIds: [], stopFingerprint: "task_watchdog_stop:unavailable", stoppedLeaves: [], + stopSnapshot: { + version: 2, + fingerprint: "task_watchdog_stop:unavailable", + materialLeaves: [], + waitsByIssueId: {}, + }, + pendingInteractionsByIssueId: {}, }, }), }; diff --git a/server/src/services/task-watchdogs.ts b/server/src/services/task-watchdogs.ts index ed708d530c..44a95f0490 100644 --- a/server/src/services/task-watchdogs.ts +++ b/server/src/services/task-watchdogs.ts @@ -87,6 +87,7 @@ export type TaskWatchdogClassifierWaitingPath = { companyId: string; issueId: string; id?: string | null; + kind?: string | null; status: string; }; @@ -99,7 +100,9 @@ export type TaskWatchdogClassifierRelation = { export type TaskWatchdogClassifierConfig = Pick< IssueWatchdogSummary, "companyId" | "issueId" | "lastReviewedFingerprint" ->; +> & { + lastReviewedStopSnapshot?: TaskWatchdogStopSnapshot | null; +}; export type TaskWatchdogStoppedLeaf = { issueId: string; @@ -117,6 +120,34 @@ export type TaskWatchdogStoppedLeaf = { latestWorkProductAt: string | null; }; +export type TaskWatchdogMaterialLeaf = Pick< + TaskWatchdogStoppedLeaf, + | "issueId" + | "status" + | "assigneeAgentId" + | "assigneeUserId" + | "blockerIssueIds" + | "pendingInteractionIds" + | "pendingApprovalIds" +>; + +export type TaskWatchdogWaitsByIssueId = Record; + +export type TaskWatchdogStopSnapshot = { + version: 2; + fingerprint: string; + materialLeaves: TaskWatchdogMaterialLeaf[]; + waitsByIssueId: TaskWatchdogWaitsByIssueId; +}; + +type TaskWatchdogPendingInteractionsByIssueId = Record>; + export type TaskWatchdogClassifierResult = | { state: "not_applicable"; @@ -141,6 +172,8 @@ export type TaskWatchdogClassifierResult = includedIssueIds: string[]; stopFingerprint: string; stoppedLeaves: TaskWatchdogStoppedLeaf[]; + stopSnapshot: TaskWatchdogStopSnapshot; + pendingInteractionsByIssueId: TaskWatchdogPendingInteractionsByIssueId; } | { state: "stopped"; @@ -148,6 +181,8 @@ export type TaskWatchdogClassifierResult = includedIssueIds: string[]; stopFingerprint: string; stoppedLeaves: TaskWatchdogStoppedLeaf[]; + stopSnapshot: TaskWatchdogStopSnapshot; + pendingInteractionsByIssueId: TaskWatchdogPendingInteractionsByIssueId; }; export type TaskWatchdogClassifierInput = { @@ -269,17 +304,70 @@ function waitingPathIds( function stableStopFingerprint(input: { companyId: string; watchedIssueId: string; - leaves: TaskWatchdogStoppedLeaf[]; + materialLeaves: TaskWatchdogMaterialLeaf[]; + waitsByIssueId: TaskWatchdogWaitsByIssueId; }) { const payload = JSON.stringify({ - version: 1, + version: 2, companyId: input.companyId, watchedIssueId: input.watchedIssueId, - leaves: input.leaves, + materialLeaves: input.materialLeaves, + waitsByIssueId: input.waitsByIssueId, }); return `task_watchdog_stop:${createHash("sha256").update(payload).digest("hex")}`; } +function materialLeaf(leaf: TaskWatchdogStoppedLeaf): TaskWatchdogMaterialLeaf { + return { + issueId: leaf.issueId, + status: leaf.status, + assigneeAgentId: leaf.assigneeAgentId, + assigneeUserId: leaf.assigneeUserId, + blockerIssueIds: leaf.blockerIssueIds, + pendingInteractionIds: leaf.pendingInteractionIds, + pendingApprovalIds: leaf.pendingApprovalIds, + }; +} + +function parseStopSnapshot(value: unknown): TaskWatchdogStopSnapshot | null { + if (!value || typeof value !== "object") return null; + const candidate = value as Partial; + if ( + candidate.version !== 2 || + typeof candidate.fingerprint !== "string" || + !Array.isArray(candidate.materialLeaves) || + !candidate.waitsByIssueId || + typeof candidate.waitsByIssueId !== "object" + ) return null; + return candidate as TaskWatchdogStopSnapshot; +} + +// Snapshots loaded from jsonb columns come back with Postgres's normalized key +// order, so equality checks against freshly built snapshots must not depend on +// object key order. +function canonicalJson(value: unknown): string { + return JSON.stringify(value, (_key, val) => + val && typeof val === "object" && !Array.isArray(val) + ? Object.fromEntries( + Object.entries(val as Record).sort(([left], [right]) => + left < right ? -1 : left > right ? 1 : 0 + ), + ) + : val); +} + +function isShrinkOfReviewedSnapshot( + current: TaskWatchdogStopSnapshot, + reviewed: TaskWatchdogStopSnapshot | null | undefined, +) { + if (!reviewed || canonicalJson(current.waitsByIssueId) !== canonicalJson(reviewed.waitsByIssueId)) return false; + const reviewedLeaves = new Map(reviewed.materialLeaves.map((leaf) => [leaf.issueId, leaf])); + return current.materialLeaves.every((leaf) => { + const previous = reviewedLeaves.get(leaf.issueId); + return previous != null && canonicalJson(previous) === canonicalJson(leaf); + }); +} + export function classifyTaskWatchdogSubtree(input: TaskWatchdogClassifierInput): TaskWatchdogClassifierResult { const issuesById = new Map(input.issues.map((issue) => [issue.id, issue])); const root = issuesById.get(input.watchdog.issueId); @@ -380,8 +468,25 @@ export function classifyTaskWatchdogSubtree(input: TaskWatchdogClassifierInput): blockersByIssueId.set(relation.blockedIssueId, list); } + const nonTerminalIssues = included + .filter((issue) => !isTerminalIssueStatus(issue.status)) + .sort((left, right) => left.id.localeCompare(right.id)); + const waitsByIssueId = Object.fromEntries(nonTerminalIssues + .map((issue) => [issue.id, { + pendingInteractionIds: waitingPathIds(input.pendingInteractions, input.watchdog.companyId, issue.id), + pendingApprovalIds: waitingPathIds(input.pendingApprovals, input.watchdog.companyId, issue.id), + }] as const) + .filter(([, waits]) => waits.pendingInteractionIds.length > 0 || waits.pendingApprovalIds.length > 0)); + const pendingInteractionsByIssueId = Object.fromEntries(nonTerminalIssues + .map((issue) => [issue.id, (input.pendingInteractions ?? []) + .filter((path) => path.companyId === input.watchdog.companyId && path.issueId === issue.id) + .map((path) => ({ id: path.id ?? `${path.status}:${path.issueId}`, kind: path.kind ?? null })) + .sort((left, right) => left.id.localeCompare(right.id))] as const) + .filter(([, waits]) => waits.length > 0)); + const leaves = included .filter((issue) => (includedChildrenByParentId.get(issue.id) ?? []).length === 0) + .filter((issue) => !isTerminalIssueStatus(issue.status)) .sort((left, right) => left.id.localeCompare(right.id)) .map((issue) => ({ issueId: issue.id, @@ -398,19 +503,32 @@ export function classifyTaskWatchdogSubtree(input: TaskWatchdogClassifierInput): latestDocumentAt: optionalIso(issue.latestDocumentAt), latestWorkProductAt: optionalIso(issue.latestWorkProductAt), })); + const materialLeaves = leaves.map(materialLeaf); const stopFingerprint = stableStopFingerprint({ companyId: input.watchdog.companyId, watchedIssueId: input.watchdog.issueId, - leaves, + materialLeaves, + waitsByIssueId, }); + const currentStopSnapshot: TaskWatchdogStopSnapshot = { + version: 2, + fingerprint: stopFingerprint, + materialLeaves, + waitsByIssueId, + }; - if (input.watchdog.lastReviewedFingerprint === stopFingerprint) { + if ( + input.watchdog.lastReviewedFingerprint === stopFingerprint || + isShrinkOfReviewedSnapshot(currentStopSnapshot, input.watchdog.lastReviewedStopSnapshot) + ) { return { state: "already_reviewed", reason: "The current stopped subtree fingerprint was already reviewed by the watchdog.", includedIssueIds: includedIds, stopFingerprint, stoppedLeaves: leaves, + stopSnapshot: currentStopSnapshot, + pendingInteractionsByIssueId, }; } @@ -420,6 +538,8 @@ export function classifyTaskWatchdogSubtree(input: TaskWatchdogClassifierInput): includedIssueIds: includedIds, stopFingerprint, stoppedLeaves: leaves, + stopSnapshot: currentStopSnapshot, + pendingInteractionsByIssueId, }; } @@ -501,11 +621,20 @@ function buildStoppedFingerprintComment(input: { sourceIssue: Pick; stopFingerprint: string; stoppedLeaves: TaskWatchdogStoppedLeaf[]; + pendingInteractionsByIssueId: TaskWatchdogPendingInteractionsByIssueId; resumed: boolean; }) { - const leafLines = input.stoppedLeaves.slice(0, 12).map((leaf) => - `- ${leaf.identifier ?? leaf.issueId}: ${leaf.status} (updated ${leaf.updatedAt})` - ); + const shortId = (id: string) => id.length > 8 ? `${id.slice(0, 8)}…` : id; + const leafLines = input.stoppedLeaves.slice(0, 12).map((leaf) => { + const interactionKinds = new Map( + (input.pendingInteractionsByIssueId[leaf.issueId] ?? []).map((wait) => [wait.id, wait.kind]), + ); + const waits = [ + ...leaf.pendingInteractionIds.map((id) => `${interactionKinds.get(id) ?? "interaction"} ${shortId(id)}`), + ...leaf.pendingApprovalIds.map((id) => `approval ${shortId(id)}`), + ]; + return `- ${leaf.identifier ?? leaf.issueId}: ${leaf.status}${waits.length > 0 ? ` (pending ${waits.join(", ")})` : ""}`; + }); const more = input.stoppedLeaves.length > leafLines.length ? `\n- ...and ${input.stoppedLeaves.length - leafLines.length} more stopped leaves` : ""; @@ -524,8 +653,13 @@ function buildStoppedFingerprintComment(input: { function stoppedFingerprintMetadata(input: { sourceIssueId: string; stopFingerprint: string; + waitsByIssueId: TaskWatchdogWaitsByIssueId; resumed: boolean; }) { + const pendingWaitCount = Object.values(input.waitsByIssueId).reduce( + (count, waits) => count + waits.pendingInteractionIds.length + waits.pendingApprovalIds.length, + 0, + ); return { version: 1 as const, sections: [ @@ -534,6 +668,7 @@ function stoppedFingerprintMetadata(input: { rows: [ { type: "text" as const, label: "Watched issue", text: input.sourceIssueId }, { type: "text" as const, label: "Stopped fingerprint", text: input.stopFingerprint }, + { type: "text" as const, label: "Pending waits", text: String(pendingWaitCount) }, { type: "text" as const, label: "Resume intent", text: input.resumed ? "true" : "false" }, ], }, @@ -557,6 +692,10 @@ function watchdogWakeContext(input: { watchedIssueIdentifier: input.sourceIssue.identifier, watchedIssueTitle: input.sourceIssue.title, stopFingerprint: input.classification.stopFingerprint, + pendingInteractions: input.classification.pendingInteractionsByIssueId, + pendingApprovals: Object.fromEntries(Object.entries(input.classification.stopSnapshot.waitsByIssueId) + .filter(([, waits]) => waits.pendingApprovalIds.length > 0) + .map(([issueId, waits]) => [issueId, waits.pendingApprovalIds])), capabilities: { targetScope: { watchedIssueId: input.sourceIssue.id, @@ -877,6 +1016,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) companyId: issueThreadInteractions.companyId, issueId: issueThreadInteractions.issueId, id: issueThreadInteractions.id, + kind: issueThreadInteractions.kind, status: issueThreadInteractions.status, }) .from(issueThreadInteractions) @@ -953,7 +1093,10 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) const completedRunIssueIds = await collectCompletedRunIssueIds(companyId, freshIssueIds); return { - watchdog: summarizeIssueWatchdog(watchdog), + watchdog: { + ...summarizeIssueWatchdog(watchdog), + lastReviewedStopSnapshot: parseStopSnapshot(watchdog.lastReviewedStopSnapshot), + }, issues: issueRows.map((issue) => ({ ...issue, latestCommentAt: latestCommentByIssueId.get(issue.id) ?? null, @@ -1136,11 +1279,20 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) : false; if (!isWatchdogReviewDisposition(watchdogIssue, hasPendingReviewPath)) return watchdog; const reviewedFingerprint = reviewedFingerprintForWatchdogIssue(watchdogIssue); - if (!reviewedFingerprint || watchdog.lastReviewedFingerprint === reviewedFingerprint) return watchdog; + if (!reviewedFingerprint) return watchdog; + const observedSnapshot = parseStopSnapshot(watchdog.lastObservedStopSnapshot); + const reviewedStopSnapshot = observedSnapshot?.fingerprint === reviewedFingerprint + ? observedSnapshot + : null; + if ( + watchdog.lastReviewedFingerprint === reviewedFingerprint && + canonicalJson(parseStopSnapshot(watchdog.lastReviewedStopSnapshot)) === canonicalJson(reviewedStopSnapshot) + ) return watchdog; const [updated] = await db .update(issueWatchdogs) .set({ lastReviewedFingerprint: reviewedFingerprint, + lastReviewedStopSnapshot: reviewedStopSnapshot, lastCompletedAt: new Date(), updatedAt: new Date(), }) @@ -1161,6 +1313,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) watchdogIssueId: watchdogIssue.id, reviewedFingerprint, lastObservedFingerprint: watchdog.lastObservedFingerprint, + reviewedStopSnapshot, watchdogIssueStatus: watchdogIssue.status, }, }); @@ -1214,6 +1367,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) sourceIssue: input.sourceIssue, stopFingerprint: input.classification.stopFingerprint, stoppedLeaves: input.classification.stoppedLeaves, + pendingInteractionsByIssueId: input.classification.pendingInteractionsByIssueId, resumed: true, }), { runId: input.runId ?? null }, @@ -1222,6 +1376,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) metadata: stoppedFingerprintMetadata({ sourceIssueId: input.sourceIssue.id, stopFingerprint: input.classification.stopFingerprint, + waitsByIssueId: input.classification.stopSnapshot.waitsByIssueId, resumed: true, }), }, @@ -1263,6 +1418,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) sourceIssue: input.sourceIssue, stopFingerprint: input.classification.stopFingerprint, stoppedLeaves: input.classification.stoppedLeaves, + pendingInteractionsByIssueId: input.classification.pendingInteractionsByIssueId, resumed: false, }), { runId: input.runId ?? null }, @@ -1271,6 +1427,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) metadata: stoppedFingerprintMetadata({ sourceIssueId: input.sourceIssue.id, stopFingerprint: input.classification.stopFingerprint, + waitsByIssueId: input.classification.stopSnapshot.waitsByIssueId, resumed: false, }), }, @@ -1305,6 +1462,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) .set({ watchdogIssueId: existingWatchdogIssueId, lastObservedFingerprint: classification.stopFingerprint, + lastObservedStopSnapshot: classification.stopSnapshot, updatedAt: new Date(), }) .where(eq(issueWatchdogs.id, watchdog.id)); @@ -1324,13 +1482,15 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) if (await sameFingerprintWatchdogReviewIsStillOpen(existingWatchdogIssue, classification.stopFingerprint)) { if ( watchdog.watchdogIssueId !== existingWatchdogIssue!.id || - watchdog.lastObservedFingerprint !== classification.stopFingerprint + watchdog.lastObservedFingerprint !== classification.stopFingerprint || + canonicalJson(parseStopSnapshot(watchdog.lastObservedStopSnapshot)) !== canonicalJson(classification.stopSnapshot) ) { await db .update(issueWatchdogs) .set({ watchdogIssueId: existingWatchdogIssue!.id, lastObservedFingerprint: classification.stopFingerprint, + lastObservedStopSnapshot: classification.stopSnapshot, updatedAt: new Date(), }) .where(eq(issueWatchdogs.id, watchdog.id)); @@ -1354,6 +1514,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) .set({ watchdogIssueId: watchdogIssue.id, lastObservedFingerprint: classification.stopFingerprint, + lastObservedStopSnapshot: classification.stopSnapshot, lastTriggeredAt: now, triggerCount: sql`${issueWatchdogs.triggerCount} + 1`, updatedAt: now, @@ -1374,6 +1535,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) watchdogId: watchdog.id, watchdogIssueId: watchdogIssue.id, stopFingerprint: classification.stopFingerprint, + stopSnapshot: classification.stopSnapshot, stoppedLeaves: classification.stoppedLeaves, }, });