diff --git a/doc/TASK-WATCHDOG.md b/doc/TASK-WATCHDOG.md index c638a6bc0cf..99f2bbaad56 100644 --- a/doc/TASK-WATCHDOG.md +++ b/doc/TASK-WATCHDOG.md @@ -84,7 +84,7 @@ Paperclip runs a watchdog reconciliation tick at server startup, at the end of e For each active watchdog the tick: -1. **Walks the watched subtree.** Starts at the configured issue and follows `parent_id` downward, excluding every issue whose `originKind = 'task_watchdog'` and everything below it. This excludes the watchdog's own review tasks so it cannot trigger itself. +1. **Walks the watched subtree.** Starts at the configured issue and follows `parent_id` downward, excluding every issue whose `originKind = 'task_watchdog'` and everything below it. This excludes the watchdog's own review tasks so it cannot trigger itself. The scan includes at most 100 levels, with the watched issue at level 1. If an eligible child remains beyond that limit, the result is **incomplete**. Paperclip skips review dispatch and rejects watchdog mutation revalidation until it can read the full subtree. A leaf at level 100 does not by itself make the scan incomplete. Reconciliation reports these cases separately through `incomplete` and `incompleteIssueIds`. Startup and periodic reconciliation warn when a watched issue first becomes incomplete. The warning contains the newly affected issue IDs. An unchanged incomplete issue does not produce repeated warnings. Once a successful scan no longer reports that issue as incomplete, a later incomplete scan can warn again. 2. **Checks for live paths.** If any included issue has a live run (`queued`, `running`, `scheduled_retry`), a queued wake request, or a scheduled retry, the subtree is **live** and the watchdog does not fire. 3. **Computes a stop fingerprint.** A SHA-256 hash over the stopped leaves' identifiers, statuses, blockers, pending interactions, and the current watchdog configuration. The configuration is part of the fingerprint, so changing the agent or instructions invalidates the previously reviewed state. 4. **Compares against `lastReviewedFingerprint`.** Match → suppress (the watchdog already saw this exact stopped state). New → proceed. diff --git a/server/src/__tests__/task-watchdogs-classifier.test.ts b/server/src/__tests__/task-watchdogs-classifier.test.ts index b44d0c76c52..16ef800c088 100644 --- a/server/src/__tests__/task-watchdogs-classifier.test.ts +++ b/server/src/__tests__/task-watchdogs-classifier.test.ts @@ -1,6 +1,8 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { classifyTaskWatchdogSubtree, type TaskWatchdogClassifierIssue } from "../services/task-watchdogs.ts"; +import { createIncompleteTaskWatchdogScanReporter } from "../services/task-watchdog-diagnostics.ts"; + const companyId = "company-1"; const sourceId = "source-1"; const childId = "child-1"; @@ -35,6 +37,45 @@ function classify(overrides: Partial { + it("reports an unchanged incomplete scan once across startup and a day of periodic ticks", () => { + const warn = vi.fn(); + const report = createIncompleteTaskWatchdogScanReporter(warn); + report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "startup"); + for (let tick = 0; tick < 2880; tick += 1) { + report({ incomplete: 2, incompleteIssueIds: [childId, sourceId] }, "periodic"); + } + expect(warn).toHaveBeenCalledTimes(1); + expect(warn).toHaveBeenCalledWith( + { incomplete: 2, newIncompleteIssueIds: [sourceId, childId] }, + "startup task-watchdog reconciliation found incomplete subtree scans", + ); + }); + + it("reports new incomplete issues and reports a resolved issue if it becomes incomplete again", () => { + const warn = vi.fn(); + const report = createIncompleteTaskWatchdogScanReporter(warn); + report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "startup"); + report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "periodic"); + expect(warn).toHaveBeenLastCalledWith( + { incomplete: 2, newIncompleteIssueIds: [childId] }, + "periodic task-watchdog reconciliation found incomplete subtree scans", + ); + report({ incomplete: 0, incompleteIssueIds: [] }, "periodic"); + expect(warn).toHaveBeenCalledTimes(2); + report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "periodic"); + expect(warn).toHaveBeenCalledTimes(3); + expect(warn).toHaveBeenLastCalledWith( + { incomplete: 1, newIncompleteIssueIds: [sourceId] }, + "periodic task-watchdog reconciliation found incomplete subtree scans", + ); + }); + + it("does not produce a stopped fingerprint from an incomplete scan", () => { + const result = classify({ subtreeIncomplete: true }); + expect(result.state).toBe("incomplete"); + expect(result).not.toHaveProperty("stopFingerprint"); + }); + it("suppresses watchdog wakeups while watched subtree work has a live path", () => { const result = classify({ issues: [ diff --git a/server/src/__tests__/task-watchdogs-scheduler.test.ts b/server/src/__tests__/task-watchdogs-scheduler.test.ts index 1124c163eb2..e5fbf969ab4 100644 --- a/server/src/__tests__/task-watchdogs-scheduler.test.ts +++ b/server/src/__tests__/task-watchdogs-scheduler.test.ts @@ -437,6 +437,81 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { expect(wakes).toHaveLength(1); }); + async function seedDeepSubtree() { + const companyId = await seedCompany(); + const sourceId = await seedIssue(companyId, { status: "blocked", issueNumber: 1000 }); + let boundaryId = sourceId; + for (let depth = 1; depth < 100; depth += 1) { + boundaryId = await seedIssue(companyId, { + parentId: boundaryId, status: "blocked", identifier: `DEEP-${depth}`, issueNumber: 1000 + depth, + }); + } + const agentId = await seedAgent(companyId); + const watchdog = await seedWatchdog(companyId, sourceId, agentId); + return { companyId, sourceId, boundaryId, agentId, watchdog, ...createService() }; + } + + it("rejects an incomplete subtree before dispatching a stopped review", async () => { + const x = await seedDeepSubtree(); + const omittedId = await seedIssue(x.companyId, { + parentId: x.boundaryId, status: "blocked", issueNumber: 1100, + }); + await db.insert(heartbeatRuns).values({ + companyId: x.companyId, agentId: x.agentId, status: "running", + contextSnapshot: { issueId: omittedId }, + }); + expect(await x.service.reconcileTaskWatchdogs({ companyId: x.companyId })) + .toMatchObject({ checked: 1, triggered: 0, incomplete: 1, incompleteIssueIds: [x.sourceId], skipped: 0 }); + expect(x.wakes).toHaveLength(0); + const [persisted] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.id, x.watchdog.id)); + expect(persisted!.watchdogIssueId).toBeNull(); + expect(persisted!.lastObservedFingerprint).toBeNull(); + }); + + it("allows a complete boundary leaf but revokes its scope when an omitted child appears", async () => { + const x = await seedDeepSubtree(); + expect((await x.service.reconcileTaskWatchdogs({ companyId: x.companyId })).triggered).toBe(1); + const [persisted] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.id, x.watchdog.id)); + const scope = { + kind: "watchdog" as const, watchdogId: x.watchdog.id, companyId: x.companyId, + watchedIssueId: x.sourceId, stopFingerprint: persisted!.lastObservedFingerprint, + }; + expect((await x.service.revalidateMutationScope(scope)).allowed).toBe(true); + const omittedId = await seedIssue(x.companyId, { + parentId: x.boundaryId, status: "blocked", issueNumber: 1100, + }); + await db.insert(heartbeatRuns).values({ + companyId: x.companyId, agentId: x.agentId, status: "running", + contextSnapshot: { issueId: omittedId }, + }); + const incomplete = await x.service.revalidateMutationScope(scope); + expect(incomplete.allowed).toBe(false); + expect(incomplete.classification?.state).toBe("incomplete"); + expect(incomplete.reason).toContain("incomplete"); + expect(await x.service.reconcileForIssueAndAncestors(x.companyId, x.sourceId)) + .toMatchObject({ triggered: 0, incomplete: 1, incompleteIssueIds: [x.sourceId], skipped: 0 }); + expect(x.wakes).toHaveLength(1); + await db.update(issues).set({ parentId: x.sourceId }).where(eq(issues.id, omittedId)); + const visible = await x.service.revalidateMutationScope(scope); + expect(visible.allowed).toBe(false); + expect(visible.classification?.state).toBe("live"); + }); + + it("does not treat excluded boundary children as a truncated watched subtree", async () => { + const x = await seedDeepSubtree(); + const hiddenId = await seedIssue(x.companyId, { parentId: x.boundaryId, issueNumber: 1100 }); + await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, hiddenId)); + const harnessId = await seedIssue(x.companyId, { parentId: x.boundaryId, issueNumber: 1103 }); + await db.update(issues).set({ harnessKind: "test" }).where(eq(issues.id, harnessId)); + await seedIssue(x.companyId, { + parentId: x.boundaryId, issueNumber: 1101, originKind: "task_watchdog", originId: x.boundaryId, + }); + const otherCompanyId = await seedCompany(); + await seedIssue(otherCompanyId, { parentId: x.boundaryId, issueNumber: 1102 }); + expect((await x.service.reconcileTaskWatchdogs({ companyId: x.companyId })).triggered).toBe(1); + expect(x.wakes).toHaveLength(1); + }); + it("reconciles ancestor watchdogs for a descendant issue mutation", async () => { const companyId = await seedCompany(); const sourceId = await seedIssue(companyId, { identifier: "WDOG-ANCESTOR", status: "done" }); diff --git a/server/src/index.ts b/server/src/index.ts index b4e814959e1..23b13375283 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -85,6 +85,7 @@ import detectPort from "detect-port"; import { createApp } from "./app.js"; import { loadConfig } from "./config.js"; import { logger } from "./middleware/logger.js"; +import { createIncompleteTaskWatchdogScanReporter } from "./services/task-watchdog-diagnostics.js"; import { setStartupRecoveryPhase } from "./startup-recovery-state.js"; import { StartupRefusalError, @@ -927,6 +928,9 @@ async function startServerWithDatabaseTeardown( } }; const pluginWorkerManager = createPluginWorkerManager(); + const reportIncompleteWatchdogScans = createIncompleteTaskWatchdogScanReporter( + (details, message) => logger.warn(details, message), + ); const heartbeat = config.heartbeatSchedulerEnabled ? heartbeatService(db as any, { pluginWorkerManager }) : null; @@ -1733,6 +1737,7 @@ async function startServerWithDatabaseTeardown( } const taskWatchdogsReconciled = await heartbeat.reconcileTaskWatchdogs(); + reportIncompleteWatchdogScans(taskWatchdogsReconciled, "startup"); if (taskWatchdogsReconciled.triggered > 0) { logger.warn( { ...taskWatchdogsReconciled }, @@ -1940,6 +1945,7 @@ async function startServerWithDatabaseTeardown( }) .then(async () => { const reconciled = await heartbeat.reconcileTaskWatchdogs(); + reportIncompleteWatchdogScans(reconciled, "periodic"); if (reconciled.triggered > 0) { logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work"); } diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 2035e1099da..dccefa9a355 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -528,6 +528,8 @@ function noopTaskWatchdogService(): TaskWatchdogService { live: 0, pendingFirstRun: 0, alreadyReviewed: 0, + incomplete: 0, + incompleteIssueIds: [], skipped: 0, watchdogIssueIds: [], }), @@ -535,6 +537,8 @@ function noopTaskWatchdogService(): TaskWatchdogService { checked: 0, triggered: 0, pendingFirstRun: 0, + incomplete: 0, + incompleteIssueIds: [], skipped: 0, watchdogIssueIds: [], }), diff --git a/server/src/services/task-watchdog-diagnostics.ts b/server/src/services/task-watchdog-diagnostics.ts new file mode 100644 index 00000000000..ea08eebe2b4 --- /dev/null +++ b/server/src/services/task-watchdog-diagnostics.ts @@ -0,0 +1,16 @@ +export function createIncompleteTaskWatchdogScanReporter( + warn: (details: { incomplete: number; newIncompleteIssueIds: string[] }, message: string) => void, +) { + let previousIncompleteIssueIds = new Set(); + return (result: { incomplete: number; incompleteIssueIds: string[] }, phase: "startup" | "periodic") => { + const currentIssueIds = new Set(result.incompleteIssueIds); + const newIncompleteIssueIds = [...currentIssueIds].filter((id) => !previousIncompleteIssueIds.has(id)); + previousIncompleteIssueIds = currentIssueIds; + if (newIncompleteIssueIds.length === 0) return; + warn( + { incomplete: result.incomplete, newIncompleteIssueIds }, + `${phase} task-watchdog reconciliation found incomplete subtree scans`, + ); + }; +} + diff --git a/server/src/services/task-watchdogs.ts b/server/src/services/task-watchdogs.ts index e9627cf74bb..cf8d92d39df 100644 --- a/server/src/services/task-watchdogs.ts +++ b/server/src/services/task-watchdogs.ts @@ -149,6 +149,11 @@ type TaskWatchdogPendingInteractionsByIssueId = Record>; export type TaskWatchdogClassifierResult = + | { + state: "incomplete"; + reason: string; + includedIssueIds: string[]; + } | { state: "not_applicable"; reason: string; @@ -188,6 +193,7 @@ export type TaskWatchdogClassifierResult = export type TaskWatchdogClassifierInput = { watchdog: TaskWatchdogClassifierConfig; issues: TaskWatchdogClassifierIssue[]; + subtreeIncomplete?: boolean; activeRuns?: TaskWatchdogClassifierPath[]; queuedWakeRequests?: TaskWatchdogClassifierPath[]; blockers?: TaskWatchdogClassifierRelation[]; @@ -382,6 +388,15 @@ export function classifyTaskWatchdogSubtree(input: TaskWatchdogClassifierInput): }; } + // A partial scan cannot prove that every execution path has stopped. + if (input.subtreeIncomplete) { + return { + state: "incomplete", + reason: "Watched subtree scan is incomplete because it exceeds the traversal depth limit.", + includedIssueIds: [...issuesById.keys()], + }; + } + const childrenByParentId = new Map(); for (const issue of input.issues) { if (issue.companyId !== input.watchdog.companyId || !issue.parentId) continue; @@ -927,20 +942,35 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) assignee_user_id AS "assigneeUserId", origin_kind AS "originKind", updated_at AS "updatedAt", - created_at AS "createdAt" + created_at AS "createdAt", + EXISTS ( + SELECT 1 + FROM watched_issues boundary + JOIN issues child ON child.parent_id = boundary.id + WHERE boundary.depth = ${TASK_WATCHDOG_SUBTREE_MAX_DEPTH - 1} + AND child.company_id = ${companyId} + AND child.hidden_at IS NULL + AND child.harness_kind IS NULL + AND child.origin_kind <> ${TASK_WATCHDOG_ORIGIN_KIND} + ) AS "subtreeIncomplete" FROM watched_issues `); - return (Array.isArray(rows) ? rows : []) as TaskWatchdogClassifierIssue[]; + const issueRows = (Array.isArray(rows) ? rows : []) as Array< + TaskWatchdogClassifierIssue & { subtreeIncomplete: boolean } + >; + return { issues: issueRows, incomplete: issueRows.some((row) => row.subtreeIncomplete) }; } async function collectClassifierInput(companyId: string, watchdog: IssueWatchdogRow) { - const issueRows = await loadWatchdogSubtreeIssues(companyId, watchdog.issueId); + const subtree = await loadWatchdogSubtreeIssues(companyId, watchdog.issueId); + const issueRows = subtree.issues; const subtreeIssueIds = issueRows.map((issue) => issue.id); - if (subtreeIssueIds.length === 0) { + if (subtreeIssueIds.length === 0 || subtree.incomplete) { return { watchdog: summarizeIssueWatchdog(watchdog), - issues: [], + issues: issueRows, + subtreeIncomplete: subtree.incomplete, activeRuns: [], queuedWakeRequests: [], blockers: [], @@ -1665,6 +1695,8 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) allowed: false as const, reason: classification.state === "stopped" ? "Task-watchdog review is stale because the watched subtree stop fingerprint changed; refresh the source state before mutating it." + : classification.state === "incomplete" + ? "Task-watchdog review cannot authorize mutations because the watched subtree scan is incomplete." : "Task-watchdog review is stale because the watched subtree now has a live, waiting, already-reviewed, or not-applicable path; refresh the source state before mutating it.", classification, }; @@ -1763,6 +1795,8 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) live: 0, pendingFirstRun: 0, alreadyReviewed: 0, + incomplete: 0, + incompleteIssueIds: [] as string[], skipped: 0, watchdogIssueIds: [] as string[], }; @@ -1782,6 +1816,9 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) result.pendingFirstRun += 1; } else if (evaluated.state === "already_reviewed") { result.alreadyReviewed += 1; + } else if (evaluated.state === "incomplete") { + result.incomplete += 1; + result.incompleteIssueIds.push(row.issueId); } else { result.skipped += 1; } @@ -1799,6 +1836,8 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) checked: 0, triggered: 0, pendingFirstRun: 0, + incomplete: 0, + incompleteIssueIds: [] as string[], skipped: 0, watchdogIssueIds: [] as string[], }; @@ -1810,6 +1849,9 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) result.watchdogIssueIds.push(evaluated.watchdogIssueId); } else if (evaluated.state === "pending_first_run") { result.pendingFirstRun += 1; + } else if (evaluated.state === "incomplete") { + result.incomplete += 1; + result.incompleteIssueIds.push(row.issueId); } else if ( evaluated.state === "watchdog_review_open" || evaluated.state === "watchdog_live" ||