diff --git a/.github/workflows/pr-trusted.yml b/.github/workflows/pr-trusted.yml index 4bcaaa279a..13c90ad810 100644 --- a/.github/workflows/pr-trusted.yml +++ b/.github/workflows/pr-trusted.yml @@ -318,6 +318,13 @@ jobs: - name: Test no-git-push check run: node --test ./scripts/check-no-git-push.test.mjs + + - name: Validate feature module boundaries + run: pnpm check:module-boundaries + + - name: Test feature module boundary check + run: node --test ./scripts/check-module-boundaries.test.mjs + - name: Test PR quality-gate scripts run: node --test '.github/scripts/tests/*.test.mjs' diff --git a/package.json b/package.json index 0a7fb82bab..3a27287be9 100644 --- a/package.json +++ b/package.json @@ -44,6 +44,7 @@ "check:token-gates": "node scripts/check-token-gates.mjs", "check:node-version": "node scripts/check-node-version-policy.mjs", "check:no-git-push": "node scripts/check-no-git-push.mjs", + "check:module-boundaries": "node scripts/check-module-boundaries.mjs", "test:check-no-git-push": "node --test scripts/check-no-git-push.test.mjs", "test:install-sh-docker": "./scripts/test-install-sh-docker.sh", "test:hermes-gateway-smoke": "node --test scripts/smoke/hermes-gateway-smoke.test.mjs", diff --git a/scripts/check-module-boundaries.mjs b/scripts/check-module-boundaries.mjs new file mode 100644 index 0000000000..4be852f8c1 --- /dev/null +++ b/scripts/check-module-boundaries.mjs @@ -0,0 +1,158 @@ +#!/usr/bin/env node + +import { readdirSync, readFileSync } from "node:fs"; +import { dirname, relative, resolve, sep } from "node:path"; +import { fileURLToPath } from "node:url"; + +const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); +const defaultServerSrc = resolve(repoRoot, "server/src"); +const defaultModulesRoot = resolve(defaultServerSrc, "modules"); +const layerNames = new Set(["domain", "application", "adapters"]); +const databasePackages = ["@paperclipai/db", "drizzle-orm", "embedded-postgres", "postgres"]; + +function normalizedRelative(from, to) { + return relative(from, to).split(sep).join("/"); +} + +function isInside(root, candidate) { + const rel = relative(root, candidate); + return rel === "" || (!rel.startsWith(`..${sep}`) && rel !== ".."); +} + +function isPackageOrSubpath(specifier, packageName) { + return specifier === packageName || specifier.startsWith(`${packageName}/`); +} + +function isDatabasePackage(specifier) { + return databasePackages.some((packageName) => isPackageOrSubpath(specifier, packageName)); +} + +export function extractImportSpecifiers(sourceText) { + const specifiers = new Set(); + const patterns = [ + /\bimport\s+(?:type\s+)?(?:[^"'`;]*?\s+from\s+)?["']([^"']+)["']/g, + /\bexport\s+(?:type\s+)?(?:[^"'`;]*?\s+from\s+)?["']([^"']+)["']/g, + /\bimport\s*\(\s*["']([^"']+)["']\s*\)/g, + /\brequire\s*\(\s*["']([^"']+)["']\s*\)/g, + ]; + + for (const pattern of patterns) { + for (const match of sourceText.matchAll(pattern)) specifiers.add(match[1]); + } + return [...specifiers]; +} + +function listProductionSourceFiles(root) { + const files = []; + const walk = (directory) => { + for (const entry of readdirSync(directory, { withFileTypes: true })) { + if (entry.name === "node_modules" || entry.name === "dist") continue; + const entryPath = resolve(directory, entry.name); + if (entry.isDirectory()) walk(entryPath); + else if ( + entry.isFile() && + /\.(?:ts|tsx)$/.test(entry.name) && + !/\.(?:test|spec)\.(?:ts|tsx)$/.test(entry.name) && + !entry.name.endsWith(".d.ts") + ) { + files.push(entryPath); + } + } + }; + walk(root); + return files.sort(); +} + +function moduleLocation(modulesRoot, filePath) { + if (!isInside(modulesRoot, filePath)) return null; + const [moduleName, layer] = normalizedRelative(modulesRoot, filePath).split("/"); + if (!moduleName) return null; + return { moduleName, layer: layerNames.has(layer) ? layer : null }; +} + +function resolvedTarget(sourceFile, specifier) { + return specifier.startsWith(".") ? resolve(dirname(sourceFile), specifier) : null; +} + +function targetServerSegments(serverSrc, target) { + if (!target || !isInside(serverSrc, target)) return []; + return normalizedRelative(serverSrc, target).split("/"); +} + +function addViolation(violations, file, layer, specifier, reason) { + violations.push({ file, layer, specifier, reason }); +} + +export function scanModuleBoundaries({ + serverSrc = defaultServerSrc, + modulesRoot = defaultModulesRoot, +} = {}) { + const violations = []; + + for (const sourceFile of listProductionSourceFiles(serverSrc)) { + const sourceLocation = moduleLocation(modulesRoot, sourceFile); + const sourceLabel = normalizedRelative(repoRoot, sourceFile); + const sourceText = readFileSync(sourceFile, "utf8"); + + for (const specifier of extractImportSpecifiers(sourceText)) { + const target = resolvedTarget(sourceFile, specifier); + const targetSegments = targetServerSegments(serverSrc, target); + const targetLocation = target ? moduleLocation(modulesRoot, target) : null; + + if (sourceLocation?.layer === "domain") { + if (isDatabasePackage(specifier)) { + addViolation(violations, sourceLabel, "domain", specifier, "domain cannot import database packages"); + } else if (specifier.startsWith("node:")) { + addViolation(violations, sourceLabel, "domain", specifier, "domain cannot import Node.js runtime modules"); + } else if (targetSegments.includes("services") || targetSegments.includes("routes")) { + addViolation(violations, sourceLabel, "domain", specifier, "domain cannot import server services or routes"); + } else if (targetLocation?.layer === "application" || targetLocation?.layer === "adapters") { + addViolation(violations, sourceLabel, "domain", specifier, "domain cannot depend on outer module layers"); + } + } + + if (sourceLocation?.layer === "application") { + if (isDatabasePackage(specifier)) { + addViolation(violations, sourceLabel, "application", specifier, "application cannot import database packages"); + } else if (targetLocation?.layer === "adapters") { + addViolation(violations, sourceLabel, "application", specifier, "application cannot import concrete adapters"); + } else if (targetSegments.join("/") === "errors.js" || targetSegments.join("/") === "errors.ts") { + addViolation(violations, sourceLabel, "application", specifier, "application cannot import HTTP error helpers"); + } + } + + if (targetLocation && sourceLocation?.moduleName !== targetLocation.moduleName) { + const targetRelative = normalizedRelative(resolve(modulesRoot, targetLocation.moduleName), target); + if (targetRelative !== "index.js" && targetRelative !== "index.ts") { + addViolation( + violations, + sourceLabel, + sourceLocation?.layer ?? null, + specifier, + `imports inside module ${targetLocation.moduleName} instead of its index`, + ); + } + } + } + } + + return violations; +} + +export function formatViolation(violation) { + const layer = violation.layer ? ` (${violation.layer})` : ""; + return `${violation.file}${layer}: ${violation.reason}: ${JSON.stringify(violation.specifier)}`; +} + +function main() { + const violations = scanModuleBoundaries(); + if (violations.length > 0) { + console.error("Feature module boundary check failed:"); + for (const violation of violations) console.error(`- ${formatViolation(violation)}`); + process.exitCode = 1; + return; + } + console.log("Feature module boundary check passed."); +} + +if (resolve(process.argv[1] ?? "") === fileURLToPath(import.meta.url)) main(); diff --git a/scripts/check-module-boundaries.test.mjs b/scripts/check-module-boundaries.test.mjs new file mode 100644 index 0000000000..9d270df848 --- /dev/null +++ b/scripts/check-module-boundaries.test.mjs @@ -0,0 +1,72 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; +import { extractImportSpecifiers, scanModuleBoundaries } from "./check-module-boundaries.mjs"; + +test("extractImportSpecifiers recognizes supported TypeScript dependency forms", () => { + assert.deepEqual( + extractImportSpecifiers([ + 'import type { Db } from "@paperclipai/db";', + 'export { helper } from "./helper.js";', + 'const adapter = await import("../adapters/postgres.js");', + 'const postgres = require("postgres");', + 'import fs = require("node:fs");', + ].join("\n")), + ["@paperclipai/db", "./helper.js", "../adapters/postgres.js", "postgres", "node:fs"], + ); +}); + +test("scanModuleBoundaries rejects outward dependencies and module-internal imports", () => { + const serverSrc = mkdtempSync(join(tmpdir(), "paperclip-module-boundaries-")); + const modulesRoot = join(serverSrc, "modules"); + + const write = (relativePath, source) => { + const filePath = join(serverSrc, relativePath); + mkdirSync(join(filePath, ".."), { recursive: true }); + writeFileSync(filePath, source); + }; + + try { + write("modules/watchdog/domain/policy.ts", [ + 'import { eq } from "drizzle-orm";', + 'import { service } from "../../../services/example.js";', + 'import { run } from "../application/run.js";', + ].join("\n")); + write( + "modules/watchdog/application/run.ts", + [ + 'import { adapter } from "../adapters/postgres.js";', + 'import { forbidden } from "../../../errors.js";', + 'import db = require("@paperclipai/db");', + ].join("\n"), + ); + write("modules/watchdog/adapters/postgres.ts", 'import { eq } from "drizzle-orm";\n'); + write("modules/watchdog/index.ts", 'export { run } from "./application/run.js";\n'); + write("services/example.ts", 'import { run } from "../modules/watchdog/application/run.js";\n'); + + const violations = scanModuleBoundaries({ serverSrc, modulesRoot }); + assert.deepEqual( + violations.map(({ specifier, reason }) => ({ specifier, reason })), + [ + { specifier: "../adapters/postgres.js", reason: "application cannot import concrete adapters" }, + { specifier: "../../../errors.js", reason: "application cannot import HTTP error helpers" }, + { specifier: "@paperclipai/db", reason: "application cannot import database packages" }, + { specifier: "drizzle-orm", reason: "domain cannot import database packages" }, + { specifier: "../../../services/example.js", reason: "domain cannot import server services or routes" }, + { specifier: "../application/run.js", reason: "domain cannot depend on outer module layers" }, + { + specifier: "../modules/watchdog/application/run.js", + reason: "imports inside module watchdog instead of its index", + }, + ], + ); + } finally { + rmSync(serverSrc, { recursive: true, force: true }); + } +}); + +test("the repository's feature modules satisfy their import boundaries", () => { + assert.deepEqual(scanModuleBoundaries(), []); +}); diff --git a/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts b/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts index 1c886ee632..8542bd365f 100644 --- a/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts +++ b/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts @@ -19,6 +19,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; import { ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS, ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS, @@ -26,6 +27,13 @@ import { recoveryService, } from "../services/recovery/service.js"; +vi.mock("../services/heartbeat-run-events.js", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, appendHeartbeatRunEvent: vi.fn(actual.appendHeartbeatRunEvent) }; +}); + +const mockedAppendHeartbeatRunEvent = vi.mocked(appendHeartbeatRunEvent); + const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -68,6 +76,7 @@ describeEmbeddedPostgres("active-run output watchdog", () => { }, 30_000); afterEach(async () => { + mockedAppendHeartbeatRunEvent.mockClear(); await truncateCompaniesWithDeadlockRetry(db); }); @@ -221,49 +230,6 @@ describeEmbeddedPostgres("active-run output watchdog", () => { expect(manager?.status).toBe("idle"); } - it.each([ - { - level: "suspicious" as const, - ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000, - }, - { - level: "critical" as const, - ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000, - }, - ])("surfaces $level silence without creating recovery work", async ({ level, ageMs }) => { - const now = new Date("2026-04-22T20:00:00.000Z"); - const seeded = await seedRunningRun({ now, ageMs }); - const { enqueueWakeup, recovery } = createRecovery(); - - await expect(recovery.buildRunOutputSilence( - (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0]!, - now, - )).resolves.toMatchObject({ - level, - silenceAgeMs: ageMs, - suspicionThresholdMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, - criticalThresholdMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS, - evaluationIssueId: null, - evaluationIssueIdentifier: null, - evaluationIssueAssigneeAgentId: null, - }); - - const first = await recovery.scanSilentActiveRuns({ now, companyId: seeded.companyId }); - const second = await recovery.scanSilentActiveRuns({ now, companyId: seeded.companyId }); - expect(first).toMatchObject({ scanned: 1, created: 0, existing: 0, escalated: 0, skipped: 1 }); - expect(second).toMatchObject({ scanned: 1, created: 0, existing: 0, escalated: 0, skipped: 1 }); - expect(first.evaluationIssueIds).toEqual([]); - expect(second.evaluationIssueIds).toEqual([]); - expect(enqueueWakeup).not.toHaveBeenCalled(); - await expectNoReviewArtifacts(seeded); - - const decisions = await db - .select() - .from(heartbeatRunWatchdogDecisions) - .where(eq(heartbeatRunWatchdogDecisions.runId, seeded.runId)); - expect(decisions).toHaveLength(0); - }); - it("keeps blocked and recovery-origin sources artifact-free", async () => { const now = new Date("2026-04-22T20:00:00.000Z"); const blocked = await seedRunningRun({ @@ -293,106 +259,135 @@ describeEmbeddedPostgres("active-run output watchdog", () => { expect(enqueueWakeup).not.toHaveBeenCalled(); }); - it("stores board snooze decisions directly on the run", async () => { + it("scopes candidates, readers, and writers to one company", async () => { + const now = new Date("2026-04-22T20:00:00.000Z"); + const companyA = await seedRunningRun({ now, ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000 }); + const companyB = await seedRunningRun({ now, ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000 }); + const healthyRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: healthyRunId, + companyId: companyA.companyId, + agentId: companyA.coderId, + status: "running", + invocationSource: "assignment", + triggerDetail: "system", + startedAt: now, + processStartedAt: now, + lastOutputAt: now, + lastOutputSeq: 1, + lastOutputStream: "stdout", + contextSnapshot: {}, + logBytes: 0, + }); + const { recovery } = createRecovery(); + + const result = await recovery.scanSilentActiveRuns({ now, companyId: companyA.companyId }); + + // The company filter, the SQL timestamp expression, and the healthy-run + // exclusion together keep the scan to the one silent run in company A. + expect(result.scanned).toBe(1); + + const evaluationIssueIdInCompanyB = randomUUID(); + await db.insert(issues).values({ + id: evaluationIssueIdInCompanyB, + companyId: companyB.companyId, + title: "Evaluation issue in the other company", + status: "todo", + priority: "medium", + assigneeAgentId: companyB.managerId, + issueNumber: 2, + identifier: `${companyB.issuePrefix}-2`, + originKind: "stale_active_run_evaluation", + originId: companyB.runId, + originRunId: companyB.runId, + originFingerprint: `stale_active_run:${companyB.companyId}:${companyB.runId}`, + }); + + await expect(recovery.recordWatchdogDecision({ + runId: companyA.runId, + actor: { type: "agent", agentId: companyA.managerId }, + decision: "continue", + evaluationIssueId: evaluationIssueIdInCompanyB, + reason: "Cross-company evaluation issue must be rejected", + now, + })).rejects.toMatchObject({ status: 404 }); + + await expect(recovery.recordWatchdogDecision({ + runId: companyA.runId, + actor: { type: "board" }, + decision: "continue", + reason: "Cross-company createdByRunId must be rejected", + createdByRunId: companyB.runId, + now, + })).rejects.toMatchObject({ status: 403 }); + }); + + it("leaves no partial state when the fold transaction fails", async () => { const now = new Date("2026-04-22T20:00:00.000Z"); const seeded = await seedRunningRun({ now, ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000, + sourceStatus: "done", + sameRunTerminalEvidence: true, + }); + const evaluationIssueId = randomUUID(); + await db.insert(issues).values({ + id: evaluationIssueId, + companyId: seeded.companyId, + title: "Existing stale evaluation", + status: "todo", + priority: "high", + assigneeAgentId: seeded.managerId, + issueNumber: 2, + identifier: `${seeded.issuePrefix}-2`, + originKind: "stale_active_run_evaluation", + originId: seeded.runId, + originRunId: seeded.runId, + originFingerprint: `stale_active_run:${seeded.companyId}:${seeded.runId}`, + }); + await db.insert(issueRecoveryActions).values({ + companyId: seeded.companyId, + sourceIssueId: seeded.issueId, + recoveryIssueId: evaluationIssueId, + kind: "active_run_watchdog", + status: "active", + ownerType: "agent", + ownerAgentId: seeded.managerId, + cause: "active_run_watchdog", + fingerprint: `active-run-watchdog:${seeded.companyId}:${seeded.runId}:${seeded.issueId}`, + evidence: { runId: seeded.runId }, + nextAction: "Review stale active run", }); const { recovery } = createRecovery(); - const snoozedUntil = new Date(now.getTime() + 60 * 60 * 1000); + mockedAppendHeartbeatRunEvent.mockRejectedValueOnce(new Error("injected fold transaction fault")); - const decision = await recovery.recordWatchdogDecision({ - runId: seeded.runId, - actor: { type: "board" }, - decision: "snooze", - snoozedUntil, - reason: "Known quiet compile", - now, - }); - expect(decision).toMatchObject({ - runId: seeded.runId, - evaluationIssueId: null, - decision: "snooze", - snoozedUntil, - }); - await expect(buildSummary(seeded.runId, now)).resolves.toMatchObject({ - level: "snoozed", - snoozedUntil, - evaluationIssueId: null, - }); - await expect(buildSummary(seeded.runId, new Date(snoozedUntil.getTime() + 1))).resolves.toMatchObject({ - level: "critical", - snoozedUntil: null, - }); - await expectNoReviewArtifacts(seeded); + await expect(recovery.scanSilentActiveRuns({ now, companyId: seeded.companyId })) + .rejects.toThrow("injected fold transaction fault"); + + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + expect(run?.status).toBe("running"); + expect(await db.select().from(heartbeatRunWatchdogDecisions).where(eq( + heartbeatRunWatchdogDecisions.runId, + seeded.runId, + ))).toHaveLength(0); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, seeded.runId))).toHaveLength(0); + // The seed itself planted one activity-log row as the fake same-run + // terminal evidence; the fold must not add a second one. + expect(await db.select().from(activityLog).where(eq(activityLog.runId, seeded.runId))).toHaveLength(1); + const [source] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + const [evaluation] = await db.select().from(issues).where(eq(issues.id, evaluationIssueId)); + const [action] = await db + .select() + .from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId)); + expect(source?.executionRunId).toBe(seeded.runId); + expect(evaluation?.status).toBe("todo"); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, evaluationIssueId))).toHaveLength(0); + expect(action).toMatchObject({ status: "active", outcome: null }); + const [agent] = await db.select().from(agents).where(eq(agents.id, seeded.coderId)); + expect(agent?.status).toBe("running"); }); - it("re-arms board continue decisions after 30 minutes without creating artifacts", async () => { - const now = new Date("2026-04-22T20:00:00.000Z"); - const seeded = await seedRunningRun({ - now, - ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000, - }); - const { enqueueWakeup, recovery } = createRecovery(); - const decision = await recovery.recordWatchdogDecision({ - runId: seeded.runId, - actor: { type: "board" }, - decision: "continue", - reason: "Keep watching this run", - now, - }); - const rearmAt = new Date(now.getTime() + ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS); - - expect(decision.evaluationIssueId).toBeNull(); - expect(decision.snoozedUntil?.toISOString()).toBe(rearmAt.toISOString()); - await expect(buildSummary(seeded.runId, new Date(rearmAt.getTime() - 1))).resolves.toMatchObject({ - level: "snoozed", - evaluationIssueId: null, - }); - await expect(buildSummary(seeded.runId, new Date(rearmAt.getTime() + 1))).resolves.toMatchObject({ - level: "suspicious", - evaluationIssueId: null, - }); - await expect(recovery.scanSilentActiveRuns({ now: new Date(rearmAt.getTime() - 1), companyId: seeded.companyId })) - .resolves.toMatchObject({ snoozed: 1, created: 0 }); - await expect(recovery.scanSilentActiveRuns({ now: new Date(rearmAt.getTime() + 1), companyId: seeded.companyId })) - .resolves.toMatchObject({ skipped: 1, created: 0 }); - expect(enqueueWakeup).not.toHaveBeenCalled(); - await expectNoReviewArtifacts(seeded); - }); - - it("permanently suppresses a run after a board false-positive decision", async () => { - const now = new Date("2026-04-22T20:00:00.000Z"); - const seeded = await seedRunningRun({ - now, - ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000, - }); - const { enqueueWakeup, recovery } = createRecovery(); - - const decision = await recovery.recordWatchdogDecision({ - runId: seeded.runId, - actor: { type: "board" }, - decision: "dismissed_false_positive", - reason: "This run is expected to remain quiet", - now, - }); - expect(decision.evaluationIssueId).toBeNull(); - await expect(buildSummary(seeded.runId, now)).resolves.toMatchObject({ - level: "not_applicable", - snoozedUntil: null, - evaluationIssueId: null, - }); - const muchLater = new Date(now.getTime() + 24 * 60 * 60 * 1000); - await expect(buildSummary(seeded.runId, muchLater)).resolves.toMatchObject({ - level: "not_applicable", - snoozedUntil: null, - }); - await expect(recovery.scanSilentActiveRuns({ now: muchLater, companyId: seeded.companyId })) - .resolves.toMatchObject({ created: 0, skipped: 1 }); - expect(enqueueWakeup).not.toHaveBeenCalled(); - await expectNoReviewArtifacts(seeded); - }); it("folds a terminal source with same-run evidence without creating review work", async () => { const now = new Date("2026-04-22T20:00:00.000Z"); @@ -598,17 +593,4 @@ describeEmbeddedPostgres("active-run output watchdog", () => { ))).toHaveLength(0); }); - it("ignores healthy runs that produced recent output", async () => { - const now = new Date("2026-04-22T20:00:00.000Z"); - const seeded = await seedRunningRun({ - now, - ageMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS + 60_000, - withOutput: true, - }); - const { recovery } = createRecovery(); - - await expect(buildSummary(seeded.runId, now)).resolves.toMatchObject({ level: "ok" }); - await expect(recovery.scanSilentActiveRuns({ now, companyId: seeded.companyId })) - .resolves.toMatchObject({ scanned: 0, created: 0 }); - }); }); diff --git a/server/src/modules/README.md b/server/src/modules/README.md new file mode 100644 index 0000000000..05ff555034 --- /dev/null +++ b/server/src/modules/README.md @@ -0,0 +1,31 @@ +# Feature modules + +A feature module under `server/src/modules//` uses three layers. Each +layer has one rule: which layer below it, it may import from. + +``` +adapters → application → domain +``` + +- **`domain/`** holds pure business rules. A domain file takes plain data in + and returns plain data out. A domain file must not import `drizzle-orm`, + `@paperclipai/db`, a service under `server/src/services/`, a route under + `server/src/routes/`, or a Node.js I/O module (`node:child_process`, + `node:fs`, `node:net`). A domain function must not read the system clock; + the caller passes `now` as an explicit `Date` value. +- **`application/`** holds use cases and the ports they need. A use case + takes its ports as constructor arguments and calls domain functions for + policy decisions. An application file must not import `drizzle-orm`, a SQL + client, a concrete adapter, or the server's HTTP error helpers. The outer + service or route translates application errors into transport responses. +- **`adapters/`** holds the concrete implementations of the ports: + Postgres queries, transactions, and process control. An adapter file may + import `drizzle-orm`, `@paperclipai/db`, and Node.js I/O modules. + +A module exposes one entry point, `index.ts`, which composes the adapters +and the use cases behind a factory function. Code outside the module imports +only that entry point, never a file inside `domain/`, `application/`, or +`adapters/` directly. + +`pnpm check:module-boundaries` enforces these rules for production source +files. It also rejects imports that bypass another module's `index.ts`. diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts new file mode 100644 index 0000000000..126bd70a39 --- /dev/null +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -0,0 +1,486 @@ +import { and, asc, desc, eq, gt, gte, inArray, notInArray, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { + agents, + agentWakeupRequests, + activityLog, + heartbeatRunWatchdogDecisions, + heartbeatRuns, + issues, +} from "@paperclipai/db"; +import { parseObject } from "../../../adapters/utils.js"; +import { visibleIssueCondition } from "../../../services/issue-visibility.js"; +import { logActivity } from "../../../services/activity-log.js"; +import { appendHeartbeatRunEvent } from "../../../services/heartbeat-run-events.js"; +import { emitAgentTaskRun } from "../../../services/agent-task-run-telemetry.js"; +import { + executeIssuePostCommitActions, + issueService, + type IssuePostCommitAction, +} from "../../../services/issues.js"; +import { issueRecoveryActionService } from "../../../services/issue-recovery-actions.js"; +import { RECOVERY_ORIGIN_KINDS } from "../../../services/recovery/origins.js"; +import { isTerminalIssueStatus } from "../domain/policy.js"; +import type { WatchdogRunReader, WatchdogWriter } from "../application/ports.js"; +import type { + EvaluationIssueSnapshot, + FoldOutcome, + FoldSourceResolvedRunInput, + RecordDecisionInput, + RunSnapshot, + SourceIssueSnapshot, + WatchdogDecisionRecord, +} from "../application/types.js"; + +const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation; +const RECOVERY_ORIGIN_KIND_VALUES = new Set(Object.values(RECOVERY_ORIGIN_KINDS)); + +function isRecoveryOriginKind(originKind: string | null): boolean { + return originKind !== null && RECOVERY_ORIGIN_KIND_VALUES.has(originKind); +} + +function issueContextId(contextSnapshot: unknown): string | null { + const context = parseObject(contextSnapshot); + const issueId = context.issueId ?? context.taskId; + return typeof issueId === "string" && issueId.length > 0 ? issueId : null; +} + +function toRunSnapshot(row: typeof heartbeatRuns.$inferSelect): RunSnapshot { + return { + id: row.id, + companyId: row.companyId, + agentId: row.agentId, + status: row.status, + lastOutputAt: row.lastOutputAt, + lastOutputSeq: row.lastOutputSeq, + lastOutputStream: row.lastOutputStream, + processStartedAt: row.processStartedAt, + startedAt: row.startedAt, + createdAt: row.createdAt, + sourceIssueId: issueContextId(row.contextSnapshot), + resultJson: row.resultJson, + wakeupRequestId: row.wakeupRequestId, + processPid: row.processPid, + processGroupId: row.processGroupId, + }; +} + +export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & WatchdogWriter { + const issuesSvc = issueService(db); + const recoveryActionsSvc = issueRecoveryActionService(db); + + async function findCandidateSilentRuns(input: { + companyId?: string; + suspicionBefore: Date; + issueCreatedAtGte?: Date | null; + }): Promise { + const rows = await db + .select() + .from(heartbeatRuns) + .where( + and( + input.companyId ? eq(heartbeatRuns.companyId, input.companyId) : undefined, + eq(heartbeatRuns.status, "running"), + sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.processStartedAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) <= ${input.suspicionBefore.toISOString()}::timestamptz`, + ), + ) + .orderBy(asc(heartbeatRuns.createdAt)) + .limit(100); + + let candidates = rows.map(toRunSnapshot); + + if (input.issueCreatedAtGte) { + const issueCreatedAtGte = input.issueCreatedAtGte; + const issueIds = [...new Set(candidates.flatMap((run) => { + return run.sourceIssueId ? [run.sourceIssueId] : []; + }))]; + const eligibleIssueIds = new Set( + issueIds.length > 0 + ? (await db.select({ id: issues.id }).from(issues).where(and( + inArray(issues.id, issueIds), + gte(issues.createdAt, issueCreatedAtGte), + ))).map((issue) => issue.id) + : [], + ); + candidates = candidates.filter((run) => { + return run.sourceIssueId !== null && eligibleIssueIds.has(run.sourceIssueId); + }); + } + + return candidates; + } + + async function findRunForCompany(companyId: string, runId: string): Promise { + const [row] = await db + .select() + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))) + .limit(1); + return row ? toRunSnapshot(row) : null; + } + + async function findLatestDecision(companyId: string, runId: string, now: Date) { + const [quietUntilRows, dismissedRows] = await Promise.all([ + db + .select({ + decision: heartbeatRunWatchdogDecisions.decision, + snoozedUntil: heartbeatRunWatchdogDecisions.snoozedUntil, + }) + .from(heartbeatRunWatchdogDecisions) + .where( + and( + eq(heartbeatRunWatchdogDecisions.companyId, companyId), + eq(heartbeatRunWatchdogDecisions.runId, runId), + inArray(heartbeatRunWatchdogDecisions.decision, ["snooze", "continue"]), + gt(heartbeatRunWatchdogDecisions.snoozedUntil, now), + ), + ) + .orderBy(desc(heartbeatRunWatchdogDecisions.createdAt)) + .limit(1), + db + .select({ id: heartbeatRunWatchdogDecisions.id }) + .from(heartbeatRunWatchdogDecisions) + .where( + and( + eq(heartbeatRunWatchdogDecisions.companyId, companyId), + eq(heartbeatRunWatchdogDecisions.runId, runId), + eq(heartbeatRunWatchdogDecisions.decision, "dismissed_false_positive"), + ), + ) + .limit(1), + ]); + const quietUntilRow = quietUntilRows[0]; + return { + dismissedFalsePositive: dismissedRows.length > 0, + quietUntilDecision: quietUntilRow && quietUntilRow.snoozedUntil + ? { decision: quietUntilRow.decision as "snooze" | "continue", snoozedUntil: quietUntilRow.snoozedUntil } + : null, + }; + } + + function selectEvaluationIssueSnapshot() { + return { + id: issues.id, + identifier: issues.identifier, + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + companyId: issues.companyId, + originKind: issues.originKind, + originId: issues.originId, + hiddenAt: issues.hiddenAt, + }; + } + + async function findOpenStaleRunEvaluation(companyId: string, runId: string): Promise { + const [row] = await db + .select(selectEvaluationIssueSnapshot()) + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND), + eq(issues.originId, runId), + visibleIssueCondition(), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .limit(1); + return row ?? null; + } + + async function findEvaluationIssueById(companyId: string, issueId: string): Promise { + const [row] = await db + .select(selectEvaluationIssueSnapshot()) + .from(issues) + .where(and(eq(issues.id, issueId), eq(issues.companyId, companyId))) + .limit(1); + return row ?? null; + } + + async function findLatestSameRunTerminalEvidence(companyId: string, input: { + runId: string; + sourceIssueId: string; + sourceIssueStatus: string; + evidenceAfter: Date | null; + }) { + if (!isTerminalIssueStatus(input.sourceIssueStatus)) return null; + const activityPredicates = [ + eq(activityLog.companyId, companyId), + eq(activityLog.runId, input.runId), + eq(activityLog.action, "issue.updated"), + eq(activityLog.entityType, "issue"), + eq(activityLog.entityId, input.sourceIssueId), + sql`${activityLog.details} ->> 'status' = ${input.sourceIssueStatus}`, + ]; + if (input.evidenceAfter) { + activityPredicates.push(gte(activityLog.createdAt, input.evidenceAfter)); + } + + const activity = await db + .select({ id: activityLog.id, createdAt: activityLog.createdAt, action: activityLog.action }) + .from(activityLog) + .where(and(...activityPredicates)) + .orderBy(desc(activityLog.createdAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + + return activity ? { kind: "activity" as const, id: activity.id, createdAt: activity.createdAt, action: activity.action } : null; + } + + async function findSourceIssue(companyId: string, issueId: string): Promise { + const [row] = await db + .select({ id: issues.id, identifier: issues.identifier, status: issues.status, originKind: issues.originKind }) + .from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.id, issueId), visibleIssueCondition())) + .limit(1); + return row ? { ...row, isRecoveryOriginKind: isRecoveryOriginKind(row.originKind) } : null; + } + + async function findRunningAgent(companyId: string, agentId: string) { + const [row] = await db + .select({ id: agents.id, companyId: agents.companyId, adapterType: agents.adapterType }) + .from(agents) + .where(and(eq(agents.id, agentId), eq(agents.companyId, companyId))) + .limit(1); + return row ?? null; + } + + async function recordDecision(companyId: string, input: RecordDecisionInput): Promise { + const [row] = await db + .insert(heartbeatRunWatchdogDecisions) + .values({ + companyId, + runId: input.runId, + evaluationIssueId: input.evaluationIssueId, + decision: input.decision, + snoozedUntil: input.snoozedUntil, + reason: input.reason, + createdByAgentId: input.createdByAgentId, + createdByUserId: input.createdByUserId, + createdByRunId: input.createdByRunId, + }) + .returning(); + + await logActivity(db, { + companyId, + actorType: input.actor.type === "agent" ? "agent" : "user", + actorId: input.actor.type === "agent" + ? input.actor.agentId ?? "agent" + : input.actor.type === "board" + ? input.actor.userId ?? "board" + : "unknown", + agentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, + runId: input.runId, + action: input.decision === "snooze" ? "heartbeat.watchdog_snoozed" : "heartbeat.watchdog_decision_recorded", + entityType: "heartbeat_run", + entityId: input.runId, + details: { + source: "recovery.record_watchdog_decision", + decision: input.decision, + evaluationIssueId: input.evaluationIssueId, + snoozedUntil: input.snoozedUntil?.toISOString() ?? null, + reason: input.reason, + }, + }); + + return { ...row, decision: row.decision as WatchdogDecisionRecord["decision"] }; + } + + async function foldSourceResolvedRun(companyId: string, input: FoldSourceResolvedRunInput): Promise { + const finalRunStatus = input.sourceIssue.status === "cancelled" ? "cancelled" : "succeeded"; + const postCommitIssueActions: IssuePostCommitAction[] = []; + const resultJson = { + ...parseObject(input.run.resultJson), + sourceResolvedWatchdogFold: { + sourceIssueId: input.sourceIssue.id, + sourceIssueIdentifier: input.sourceIssue.identifier, + sourceIssueStatus: input.sourceIssue.status, + sameRunEvidenceKind: input.evidence.kind, + sameRunEvidenceId: input.evidence.id, + sameRunEvidenceAt: input.evidence.createdAt.toISOString(), + silenceStartedAt: input.silenceStartedAt?.toISOString() ?? null, + silenceAgeMs: input.silenceAgeMs, + evaluationIssueId: input.existingEvaluation?.id ?? null, + evaluationIssueIdentifier: input.existingEvaluation?.identifier ?? null, + cleanup: input.cleanup, + }, + }; + + const transactionResult = await db.transaction(async (tx) => { + const [updatedRun] = await tx + .update(heartbeatRuns) + .set({ + status: finalRunStatus, + finishedAt: input.now, + error: null, + errorCode: null, + resultJson, + updatedAt: input.now, + }) + .where(and( + eq(heartbeatRuns.id, input.run.id), + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.status, "running"), + )) + .returning(); + if (!updatedRun) return null; + + if (input.run.wakeupRequestId) { + await tx + .update(agentWakeupRequests) + .set({ + status: finalRunStatus === "succeeded" ? "completed" : "cancelled", + finishedAt: input.now, + error: null, + updatedAt: input.now, + }) + .where(and( + eq(agentWakeupRequests.id, input.run.wakeupRequestId), + eq(agentWakeupRequests.companyId, companyId), + )); + } + + await tx + .update(issues) + .set({ + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: input.now, + }) + .where(and( + eq(issues.id, input.sourceIssue.id), + eq(issues.companyId, companyId), + eq(issues.executionRunId, input.run.id), + )); + + if (input.existingEvaluation && !isTerminalIssueStatus(input.existingEvaluation.status)) { + const updatedEvaluation = await issuesSvc.update( + input.existingEvaluation.id, + { status: "done" }, + tx, + undefined, + postCommitIssueActions, + ); + if (!updatedEvaluation) { + throw new Error("Evaluation issue disappeared during source-resolved watchdog fold"); + } + await issuesSvc.addComment(input.existingEvaluation.id, [ + "Source-resolved watchdog fold.", + "", + `- Source issue: ${input.sourceIssue.identifier ?? input.sourceIssue.id}`, + `- Run: \`${input.run.id}\``, + `- Same-run evidence: \`${input.evidence.kind}:${input.evidence.id}\` at ${input.evidence.createdAt.toISOString()}`, + "- Outcome: false positive; the source issue already reached a terminal disposition from this run.", + ].join("\n"), { runId: input.run.id }, undefined, tx); + } + + const activeRecoveryAction = await recoveryActionsSvc.getActiveForIssue( + companyId, + input.sourceIssue.id, + tx, + ); + if (activeRecoveryAction?.kind === "active_run_watchdog") { + await recoveryActionsSvc.resolveActiveForIssue({ + companyId, + sourceIssueId: input.sourceIssue.id, + actionId: activeRecoveryAction.id, + status: "resolved", + outcome: "false_positive", + resolutionNote: "Source issue reached a terminal disposition through durable same-run activity; watchdog folded as source-resolved.", + }, tx); + } + + const [decision] = await tx + .insert(heartbeatRunWatchdogDecisions) + .values({ + companyId, + runId: input.run.id, + evaluationIssueId: input.existingEvaluation?.id ?? null, + decision: "dismissed_false_positive", + reason: "Source issue already reached a terminal disposition through durable same-run activity.", + createdByRunId: input.run.id, + }) + .returning(); + + await appendHeartbeatRunEvent(tx as unknown as Db, { + companyId, + runId: input.run.id, + agentId: input.run.agentId, + eventType: "lifecycle", + stream: "system", + level: input.cleanup.outcome === "failed" ? "warn" : "info", + message: "Source-resolved watchdog fold finalized stale active run", + payload: resultJson.sourceResolvedWatchdogFold, + }); + + await logActivity(tx as unknown as Db, { + companyId, + actorType: "system", + actorId: "system", + agentId: input.run.agentId, + runId: input.run.id, + action: "heartbeat.output_stale_source_resolved", + entityType: "heartbeat_run", + entityId: input.run.id, + details: { + source: "recovery.scan_silent_active_runs", + sourceIssueId: input.sourceIssue.id, + sourceIssueIdentifier: input.sourceIssue.identifier, + sourceIssueStatus: input.sourceIssue.status, + evaluationIssueId: input.existingEvaluation?.id ?? null, + watchdogDecisionId: decision.id, + sameRunEvidenceKind: input.evidence.kind, + sameRunEvidenceId: input.evidence.id, + sameRunEvidenceAt: input.evidence.createdAt.toISOString(), + cleanup: input.cleanup, + }, + }); + + const [runningCountRow] = await tx + .select({ count: sql`count(*)::int` }) + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.agentId, input.run.agentId), + eq(heartbeatRuns.status, "running"), + )); + const runningCount = Number(runningCountRow?.count ?? 0); + const nextAgentStatus = runningCount > 0 ? "running" : "idle"; + await tx + .update(agents) + .set({ status: nextAgentStatus, lastHeartbeatAt: new Date(), updatedAt: new Date() }) + .where(and( + eq(agents.id, input.run.agentId), + eq(agents.companyId, companyId), + notInArray(agents.status, ["paused", "terminated"]), + )); + + return updatedRun; + }); + + if (!transactionResult) return { kind: "stale" }; + const finalizedRun = transactionResult; + + await executeIssuePostCommitActions(db, postCommitIssueActions); + + // Telemetry is best-effort background work; it must not delay the + // watchdog fold's caller, so fire it and do not await it. + void emitAgentTaskRun(db, finalizedRun); + + return { kind: "folded", evaluationIssueId: input.existingEvaluation?.id ?? null }; + } + + return { + findCandidateSilentRuns, + findRunForCompany, + findLatestDecision, + findOpenStaleRunEvaluation, + findEvaluationIssueById, + findLatestSameRunTerminalEvidence, + findSourceIssue, + findRunningAgent, + recordDecision, + foldSourceResolvedRun, + }; +} diff --git a/server/src/modules/active-run-watchdog/adapters/process.test.ts b/server/src/modules/active-run-watchdog/adapters/process.test.ts new file mode 100644 index 0000000000..dbb51b9e08 --- /dev/null +++ b/server/src/modules/active-run-watchdog/adapters/process.test.ts @@ -0,0 +1,178 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { runningProcesses } from "../../../adapters/utils.js"; +import { isPidAlive, isProcessGroupAlive, terminateLocalService } from "../../../services/local-service-supervisor.js"; +import { createProcessAdapter } from "./process.js"; + +vi.mock("../../../services/local-service-supervisor.js", () => ({ + isPidAlive: vi.fn(), + isProcessGroupAlive: vi.fn(), + terminateLocalService: vi.fn(), +})); + +const mockedIsPidAlive = vi.mocked(isPidAlive); +const mockedIsProcessGroupAlive = vi.mocked(isProcessGroupAlive); +const mockedTerminateLocalService = vi.mocked(terminateLocalService); + +describe("adapters", () => { + describe("createProcessAdapter", () => { + beforeEach(() => { + mockedIsPidAlive.mockReset(); + mockedIsProcessGroupAlive.mockReset(); + mockedTerminateLocalService.mockReset(); + runningProcesses.clear(); + }); + + it("reports skipped_non_local_adapter for a non-sessioned adapter type", async () => { + const adapter = createProcessAdapter(); + + const outcome = await adapter.cleanupRunProcess({ + runId: "run-1", + adapterType: "hermes_gateway", + fallbackPid: 4242, + fallbackProcessGroupId: null, + }); + + expect(outcome).toEqual({ attempted: false, outcome: "skipped_non_local_adapter", adapterType: "hermes_gateway" }); + expect(mockedIsPidAlive).not.toHaveBeenCalled(); + }); + + it("reports no_process_metadata when no pid or process group is known", async () => { + const adapter = createProcessAdapter(); + + const outcome = await adapter.cleanupRunProcess({ + runId: "run-1", + adapterType: "codex_local", + fallbackPid: null, + fallbackProcessGroupId: null, + }); + + expect(outcome).toEqual({ attempted: false, outcome: "no_process_metadata", adapterType: "codex_local" }); + }); + + it("reports not_running when the process is dead", async () => { + mockedIsPidAlive.mockReturnValue(false); + mockedIsProcessGroupAlive.mockReturnValue(false); + const adapter = createProcessAdapter(); + + const outcome = await adapter.cleanupRunProcess({ + runId: "run-1", + adapterType: "codex_local", + fallbackPid: 4242, + fallbackProcessGroupId: null, + }); + + expect(outcome).toEqual({ + attempted: false, + outcome: "not_running", + adapterType: "codex_local", + pid: 4242, + processGroupId: null, + }); + expect(mockedTerminateLocalService).not.toHaveBeenCalled(); + }); + + it("reports terminated when the live process stops after termination", async () => { + mockedIsPidAlive.mockReturnValueOnce(true).mockReturnValueOnce(false); + mockedIsProcessGroupAlive.mockReturnValue(false); + mockedTerminateLocalService.mockResolvedValue(undefined); + const adapter = createProcessAdapter(); + + const outcome = await adapter.cleanupRunProcess({ + runId: "run-1", + adapterType: "codex_local", + fallbackPid: 4242, + fallbackProcessGroupId: null, + }); + + expect(outcome).toEqual({ + attempted: true, + outcome: "terminated", + adapterType: "codex_local", + pid: 4242, + processGroupId: null, + }); + expect(mockedTerminateLocalService).toHaveBeenCalledTimes(1); + }); + + it("reports failed when termination throws", async () => { + mockedIsPidAlive.mockReturnValue(true); + mockedIsProcessGroupAlive.mockReturnValue(false); + mockedTerminateLocalService.mockRejectedValue(new Error("kill failed")); + const adapter = createProcessAdapter(); + + const outcome = await adapter.cleanupRunProcess({ + runId: "run-1", + adapterType: "codex_local", + fallbackPid: 4242, + fallbackProcessGroupId: null, + }); + + expect(outcome).toEqual({ + attempted: true, + outcome: "failed", + adapterType: "codex_local", + pid: 4242, + processGroupId: null, + error: "kill failed", + }); + }); + + it("uses a valid process group when no pid is available", async () => { + mockedIsProcessGroupAlive.mockReturnValueOnce(true).mockReturnValueOnce(false); + mockedTerminateLocalService.mockResolvedValue(undefined); + const adapter = createProcessAdapter(); + + const outcome = await adapter.cleanupRunProcess({ + runId: "run-1", + adapterType: "codex_local", + fallbackPid: null, + fallbackProcessGroupId: 4242, + }); + + expect(outcome).toEqual({ + attempted: true, + outcome: "terminated", + adapterType: "codex_local", + pid: null, + processGroupId: 4242, + }); + expect(mockedTerminateLocalService).toHaveBeenCalledWith( + { pid: 4242, processGroupId: 4242 }, + undefined, + ); + }); + + it.each([ + { fallbackPid: 0, fallbackProcessGroupId: null }, + { fallbackPid: -7, fallbackProcessGroupId: null }, + { fallbackPid: 4.5, fallbackProcessGroupId: null }, + { fallbackPid: null, fallbackProcessGroupId: 0 }, + { fallbackPid: null, fallbackProcessGroupId: -7 }, + { fallbackPid: null, fallbackProcessGroupId: 4.5 }, + ])( + "reports no_process_metadata for invalid identifiers ($fallbackPid, $fallbackProcessGroupId)", + async ({ fallbackPid, fallbackProcessGroupId }) => { + mockedIsPidAlive.mockReturnValue(true); + mockedIsProcessGroupAlive.mockReturnValue(false); + mockedTerminateLocalService.mockResolvedValue(undefined); + const adapter = createProcessAdapter(); + + const outcome = await adapter.cleanupRunProcess({ + runId: "run-1", + adapterType: "codex_local", + fallbackPid, + fallbackProcessGroupId, + }); + + expect(outcome).toEqual({ + attempted: false, + outcome: "no_process_metadata", + adapterType: "codex_local", + }); + expect(mockedIsPidAlive).not.toHaveBeenCalled(); + expect(mockedIsProcessGroupAlive).not.toHaveBeenCalled(); + expect(mockedTerminateLocalService).not.toHaveBeenCalled(); + }, + ); + }); +}); diff --git a/server/src/modules/active-run-watchdog/adapters/process.ts b/server/src/modules/active-run-watchdog/adapters/process.ts new file mode 100644 index 0000000000..17fcbb95c4 --- /dev/null +++ b/server/src/modules/active-run-watchdog/adapters/process.ts @@ -0,0 +1,85 @@ +import { runningProcesses } from "../../../adapters/utils.js"; +import { isPidAlive, isProcessGroupAlive, terminateLocalService } from "../../../services/local-service-supervisor.js"; +import type { RunProcessController } from "../application/ports.js"; +import type { RunProcessCleanupOutcome, RunProcessMetadata } from "../application/types.js"; + +const SESSIONED_LOCAL_ADAPTERS = new Set([ + "claude_local", + "codex_local", + "cursor", + "gemini_local", + "hermes_local", + "kimi_local", + "opencode_local", + "pi_local", +]); + +function isValidPositivePid(value: number | null): value is number { + return typeof value === "number" && Number.isInteger(value) && value > 0; +} + +export function createProcessAdapter(): RunProcessController { + return { + async cleanupRunProcess(input: RunProcessMetadata): Promise { + if (!SESSIONED_LOCAL_ADAPTERS.has(input.adapterType)) { + return { attempted: false, outcome: "skipped_non_local_adapter", adapterType: input.adapterType }; + } + + const running = runningProcesses.get(input.runId); + const registeredPid = running?.child.pid ?? null; + const registeredProcessGroupId = running?.processGroupId ?? null; + const pid = isValidPositivePid(registeredPid) + ? registeredPid + : isValidPositivePid(input.fallbackPid) + ? input.fallbackPid + : null; + const processGroupId = isValidPositivePid(registeredProcessGroupId) + ? registeredProcessGroupId + : isValidPositivePid(input.fallbackProcessGroupId) + ? input.fallbackProcessGroupId + : null; + const terminationPid = pid ?? processGroupId; + if (terminationPid === null) { + return { attempted: false, outcome: "no_process_metadata", adapterType: input.adapterType }; + } + + const wasAlive = + (pid !== null && isPidAlive(pid)) || + (processGroupId !== null && isProcessGroupAlive(processGroupId)); + if (!wasAlive) { + runningProcesses.delete(input.runId); + return { attempted: false, outcome: "not_running", adapterType: input.adapterType, pid, processGroupId }; + } + + try { + await terminateLocalService( + { + pid: terminationPid, + processGroupId, + }, + running ? { forceAfterMs: Math.max(1, running.graceSec) * 1000 } : undefined, + ); + runningProcesses.delete(input.runId); + const stillAlive = + (pid !== null && isPidAlive(pid)) || + (processGroupId !== null && isProcessGroupAlive(processGroupId)); + return { + attempted: true, + outcome: stillAlive ? "termination_sent_still_running" : "terminated", + adapterType: input.adapterType, + pid, + processGroupId, + }; + } catch (error) { + return { + attempted: true, + outcome: "failed", + adapterType: input.adapterType, + pid, + processGroupId, + error: error instanceof Error ? error.message : String(error), + }; + } + }, + }; +} diff --git a/server/src/modules/active-run-watchdog/application/ports.ts b/server/src/modules/active-run-watchdog/application/ports.ts new file mode 100644 index 0000000000..d4f73b2a62 --- /dev/null +++ b/server/src/modules/active-run-watchdog/application/ports.ts @@ -0,0 +1,82 @@ +import type { + EvaluationIssueSnapshot, + FoldOutcome, + FoldSourceResolvedRunInput, + RecordDecisionInput, + RunProcessCleanupOutcome, + RunProcessMetadata, + RunSnapshot, + SourceIssueSnapshot, + TerminalEvidence, + WatchdogDecisionRecord, +} from "./types.js"; + +export type FindCandidateSilentRunsInput = { + companyId?: string; + suspicionBefore: Date; + issueCreatedAtGte?: Date | null; +}; + +export type FindLatestDecisionResult = { + quietUntilDecision: { decision: "snooze" | "continue"; snoozedUntil: Date } | null; + dismissedFalsePositive: boolean; +}; + +export type FindLatestSameRunTerminalEvidenceInput = { + runId: string; + sourceIssueId: string; + sourceIssueStatus: string; + evidenceAfter: Date | null; +}; + +export type RunningAgentSnapshot = { + id: string; + companyId: string; + adapterType: string; +}; + +/** + * Reads active-run watchdog state. Every method takes the company scope of + * the caller, except `findCandidateSilentRuns`: the periodic recovery scan + * runs system-wide, so its scope is optional. Every candidate that method + * returns carries its own authoritative `companyId`, and every later read + * or write must use that value, never a value from the caller. + * + * `findRunningAgent` and `findEvaluationIssueById` are not two of the ports + * the design review named; each is a narrow, company-scoped addition that + * carries a read the recovery service already performed before the + * extraction. `findRunningAgent` guards against a run whose `agentId` + * foreign key does not match the run's own company. `findEvaluationIssueById` + * loads a caller-named evaluation issue by its own id, in any status, which + * `findOpenStaleRunEvaluation`'s open-only, run-bound query cannot do. + */ +export interface WatchdogRunReader { + findCandidateSilentRuns(input: FindCandidateSilentRunsInput): Promise; + findRunForCompany(companyId: string, runId: string): Promise; + findLatestDecision(companyId: string, runId: string, now: Date): Promise; + findOpenStaleRunEvaluation(companyId: string, runId: string): Promise; + findLatestSameRunTerminalEvidence( + companyId: string, + input: FindLatestSameRunTerminalEvidenceInput, + ): Promise; + findSourceIssue(companyId: string, issueId: string): Promise; + findRunningAgent(companyId: string, agentId: string): Promise; + findEvaluationIssueById(companyId: string, issueId: string): Promise; +} + +/** + * Writes active-run watchdog state. `foldSourceResolvedRun` is one + * semantic operation: it owns the `running` compare-and-set, the cleared + * source-issue execution fields, evaluation closure and comment, recovery + * action resolution, decision, run event, activity record, and agent-status + * update inside one transaction. + */ +export interface WatchdogWriter { + recordDecision(companyId: string, input: RecordDecisionInput): Promise; + foldSourceResolvedRun(companyId: string, input: FoldSourceResolvedRunInput): Promise; +} + +/** Controls the local operating-system process backing a run. */ +export interface RunProcessController { + cleanupRunProcess(input: RunProcessMetadata): Promise; +} diff --git a/server/src/modules/active-run-watchdog/application/types.ts b/server/src/modules/active-run-watchdog/application/types.ts new file mode 100644 index 0000000000..63fc3a8995 --- /dev/null +++ b/server/src/modules/active-run-watchdog/application/types.ts @@ -0,0 +1,166 @@ +export type RunStatus = string; + +export type RunSnapshot = { + id: string; + companyId: string; + agentId: string; + status: RunStatus; + lastOutputAt: Date | null; + lastOutputSeq: number | null; + lastOutputStream: string | null; + processStartedAt: Date | null; + startedAt: Date | null; + createdAt: Date | null; + sourceIssueId: string | null; + resultJson: unknown; + wakeupRequestId: string | null; + processPid: number | null; + processGroupId: number | null; +}; + +export type SourceIssueSnapshot = { + id: string; + identifier: string | null; + status: string; + originKind: string | null; + isRecoveryOriginKind: boolean; +}; + +export type EvaluationIssueSnapshot = { + id: string; + identifier: string | null; + status: string; + assigneeAgentId: string | null; + companyId: string; + originKind: string; + originId: string | null; + hiddenAt: Date | null; +}; + +export type TerminalEvidence = { + kind: "activity"; + id: string; + createdAt: Date; + action: string; +}; + +export type RunOutputSilenceSummary = { + lastOutputAt: Date | null; + lastOutputSeq: number; + lastOutputStream: "stdout" | "stderr" | null; + silenceStartedAt: Date | null; + silenceAgeMs: number | null; + level: "not_applicable" | "ok" | "suspicious" | "critical" | "snoozed"; + suspicionThresholdMs: number; + criticalThresholdMs: number; + snoozedUntil: Date | null; + evaluationIssueId: string | null; + evaluationIssueIdentifier: string | null; + evaluationIssueAssigneeAgentId: string | null; +}; + +export type ScanSilentActiveRunsResult = { + scanned: number; + created: number; + existing: number; + escalated: number; + folded: number; + snoozed: number; + skipped: number; + evaluationIssueIds: string[]; +}; + +export type RunProcessMetadata = { + runId: string; + adapterType: string; + fallbackPid: number | null; + fallbackProcessGroupId: number | null; +}; + +export type RunProcessCleanupOutcome = + | { + attempted: false; + outcome: "skipped_non_local_adapter" | "no_process_metadata" | "not_running"; + adapterType: string; + pid?: number | null; + processGroupId?: number | null; + } + | { + attempted: true; + outcome: "terminated" | "termination_sent_still_running"; + adapterType: string; + pid: number | null; + processGroupId: number | null; + } + | { + attempted: true; + outcome: "failed"; + adapterType: string; + pid: number | null; + processGroupId: number | null; + error: string; + }; + +export type FoldSourceResolvedRunInput = { + run: RunSnapshot; + sourceIssue: SourceIssueSnapshot; + evidence: TerminalEvidence; + existingEvaluation: EvaluationIssueSnapshot | null; + silenceStartedAt: Date | null; + silenceAgeMs: number | null; + cleanup: RunProcessCleanupOutcome; + now: Date; +}; + +export type FoldOutcome = + | { kind: "folded"; evaluationIssueId: string | null } + | { kind: "stale" }; + +export type WatchdogDecisionActor = + | { type: "board"; userId?: string | null; runId?: string | null } + | { type: "agent"; agentId?: string | null; runId?: string | null } + | { type: "none" }; + +export type RecordDecisionInput = { + runId: string; + actor: WatchdogDecisionActor; + evaluationIssueId: string | null; + decision: "snooze" | "continue" | "dismissed_false_positive"; + snoozedUntil: Date | null; + reason: string | null; + createdByAgentId: string | null; + createdByUserId: string | null; + createdByRunId: string | null; +}; + +export type WatchdogDecisionRecord = { + id: string; + companyId: string; + runId: string; + evaluationIssueId: string | null; + decision: "snooze" | "continue" | "dismissed_false_positive"; + snoozedUntil: Date | null; + reason: string | null; + createdByAgentId: string | null; + createdByUserId: string | null; + createdByRunId: string | null; + createdAt: Date; +}; + +export type WatchdogDecisionApplicationErrorCode = + | "run_not_found" + | "evaluation_issue_not_found" + | "not_authorized" + | "evaluation_issue_mismatch" + | "evaluation_issue_required" + | "creator_run_invalid"; + +export class WatchdogDecisionApplicationError extends Error { + constructor( + readonly code: WatchdogDecisionApplicationErrorCode, + message: string, + ) { + super(message); + this.name = "WatchdogDecisionApplicationError"; + } +} diff --git a/server/src/modules/active-run-watchdog/application/use-cases.test.ts b/server/src/modules/active-run-watchdog/application/use-cases.test.ts new file mode 100644 index 0000000000..c8d1c1570f --- /dev/null +++ b/server/src/modules/active-run-watchdog/application/use-cases.test.ts @@ -0,0 +1,497 @@ +import { describe, expect, it } from "vitest"; +import { + createBuildRunOutputSilence, + createFoldSourceResolvedRun, + createRecordWatchdogDecision, + createScanSilentActiveRuns, +} from "./use-cases.js"; +import type { RunProcessController, WatchdogRunReader, WatchdogWriter } from "./ports.js"; +import type { RunSnapshot } from "./types.js"; + +describe("application", () => { + const SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; + const CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; + const CONTINUE_REARM_MS = 30 * 60 * 1000; + + function makeRun(overrides: Partial = {}): RunSnapshot { + return { + id: "run-1", + companyId: "company-1", + agentId: "agent-1", + status: "running", + lastOutputAt: null, + lastOutputSeq: 0, + lastOutputStream: null, + processStartedAt: new Date("2026-01-01T00:00:00.000Z"), + startedAt: new Date("2026-01-01T00:00:00.000Z"), + createdAt: new Date("2026-01-01T00:00:00.000Z"), + sourceIssueId: null, + resultJson: null, + wakeupRequestId: null, + processPid: null, + processGroupId: null, + ...overrides, + }; + } + + function makeReader(overrides: Partial = {}): WatchdogRunReader { + return { + findCandidateSilentRuns: async () => [], + findRunForCompany: async () => null, + findLatestDecision: async () => ({ quietUntilDecision: null, dismissedFalsePositive: false }), + findOpenStaleRunEvaluation: async () => null, + findLatestSameRunTerminalEvidence: async () => null, + findSourceIssue: async () => null, + findRunningAgent: async () => null, + findEvaluationIssueById: async () => null, + ...overrides, + }; + } + + describe("createBuildRunOutputSilence", () => { + it("builds a summary from the reader ports and the domain level", async () => { + const now = new Date("2026-01-01T05:00:00.000Z"); + const run = makeRun(); + const reader = makeReader({ + findOpenStaleRunEvaluation: async () => ({ + id: "eval-1", + identifier: "PAP-9", + status: "todo", + assigneeAgentId: "agent-2", + companyId: "company-1", + originKind: "stale_active_run_evaluation", + originId: run.id, + hiddenAt: null, + }), + }); + const buildRunOutputSilence = createBuildRunOutputSilence({ + reader, + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + criticalThresholdMs: CRITICAL_THRESHOLD_MS, + }); + + const summary = await buildRunOutputSilence(run, now); + + expect(summary).toMatchObject({ + level: "critical", + silenceAgeMs: 5 * 60 * 60 * 1000, + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + criticalThresholdMs: CRITICAL_THRESHOLD_MS, + evaluationIssueId: "eval-1", + evaluationIssueIdentifier: "PAP-9", + evaluationIssueAssigneeAgentId: "agent-2", + }); + }); + + it("reports snoozed while a snooze decision is active", async () => { + const now = new Date("2026-01-01T05:00:00.000Z"); + const run = makeRun(); + const snoozedUntil = new Date("2026-01-01T06:00:00.000Z"); + const reader = makeReader({ + findLatestDecision: async () => ({ + quietUntilDecision: { decision: "snooze", snoozedUntil }, + dismissedFalsePositive: false, + }), + }); + const buildRunOutputSilence = createBuildRunOutputSilence({ + reader, + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + criticalThresholdMs: CRITICAL_THRESHOLD_MS, + }); + + const summary = await buildRunOutputSilence(run, now); + + expect(summary).toMatchObject({ level: "snoozed", snoozedUntil }); + }); + + it("reports not_applicable for a run that is not running", async () => { + const now = new Date("2026-01-01T05:00:00.000Z"); + const run = makeRun({ status: "succeeded" }); + const buildRunOutputSilence = createBuildRunOutputSilence({ + reader: makeReader(), + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + criticalThresholdMs: CRITICAL_THRESHOLD_MS, + }); + + const summary = await buildRunOutputSilence(run, now); + + expect(summary).toMatchObject({ level: "not_applicable", silenceAgeMs: null }); + }); + }); + + const foldSourceIssue = { + id: "issue-1", + identifier: "PAP-1", + status: "done", + originKind: "manual", + isRecoveryOriginKind: false, + }; + const foldEvidence = { + kind: "activity" as const, + id: "activity-1", + createdAt: new Date("2026-01-01T00:10:00.000Z"), + action: "issue.updated", + }; + + describe("createFoldSourceResolvedRun", () => { + it("cleans up the process before the folding persistence call", async () => { + const calls: string[] = []; + const processController: RunProcessController = { + cleanupRunProcess: async (input) => { + calls.push("cleanup"); + expect(input).toMatchObject({ runId: "run-1", adapterType: "codex_local", fallbackPid: 4242 }); + return { attempted: true, outcome: "terminated", adapterType: "codex_local", pid: 4242, processGroupId: null }; + }, + }; + const writer: WatchdogWriter = { + recordDecision: async () => { + throw new Error("not used in this test"); + }, + foldSourceResolvedRun: async (companyId, input) => { + calls.push("fold"); + expect(companyId).toBe("company-1"); + expect(input.cleanup).toMatchObject({ outcome: "terminated" }); + return { kind: "folded", evaluationIssueId: null }; + }, + }; + const foldSourceResolvedRun = createFoldSourceResolvedRun({ writer, processController }); + + const outcome = await foldSourceResolvedRun({ + run: makeRun({ processPid: 4242 }), + runningAgentAdapterType: "codex_local", + sourceIssue: foldSourceIssue, + evidence: foldEvidence, + existingEvaluation: null, + silenceStartedAt: new Date("2026-01-01T00:00:00.000Z"), + silenceAgeMs: 5 * 60 * 60 * 1000, + now: new Date("2026-01-01T05:00:00.000Z"), + }); + + expect(calls).toEqual(["cleanup", "fold"]); + expect(outcome).toEqual({ kind: "folded", evaluationIssueId: null }); + }); + + it("returns a stale outcome when the compare-and-set fails after termination", async () => { + const processController: RunProcessController = { + cleanupRunProcess: async () => ({ + attempted: true, + outcome: "terminated", + adapterType: "codex_local", + pid: 4242, + processGroupId: null, + }), + }; + const writer: WatchdogWriter = { + recordDecision: async () => { + throw new Error("not used in this test"); + }, + foldSourceResolvedRun: async () => ({ kind: "stale" }), + }; + const foldSourceResolvedRun = createFoldSourceResolvedRun({ writer, processController }); + + const outcome = await foldSourceResolvedRun({ + run: makeRun({ processPid: 4242 }), + runningAgentAdapterType: "codex_local", + sourceIssue: foldSourceIssue, + evidence: foldEvidence, + existingEvaluation: null, + silenceStartedAt: null, + silenceAgeMs: null, + now: new Date("2026-01-01T05:00:00.000Z"), + }); + + expect(outcome).toEqual({ kind: "stale" }); + }); + }); + + function makeDecisionWriter(overrides: Partial = {}): WatchdogWriter { + return { + recordDecision: async (companyId, input) => ({ + id: "decision-1", + companyId, + runId: input.runId, + evaluationIssueId: input.evaluationIssueId, + decision: input.decision, + snoozedUntil: input.snoozedUntil, + reason: input.reason, + createdByAgentId: input.createdByAgentId, + createdByUserId: input.createdByUserId, + createdByRunId: input.createdByRunId, + createdAt: new Date("2026-01-01T00:00:00.000Z"), + }), + foldSourceResolvedRun: async () => { + throw new Error("not used in this test"); + }, + ...overrides, + }; + } + + describe("createRecordWatchdogDecision", () => { + it("records snooze, continue, and false-positive decisions with company scope", async () => { + const now = new Date("2026-01-01T05:00:00.000Z"); + const snoozedUntil = new Date("2026-01-01T06:00:00.000Z"); + const recordWatchdogDecision = createRecordWatchdogDecision({ + reader: makeReader({ findRunForCompany: async () => makeRun() }), + writer: makeDecisionWriter(), + continueRearmMs: CONTINUE_REARM_MS, + }); + + const snooze = await recordWatchdogDecision({ + companyId: "company-1", + runId: "run-1", + actor: { type: "board" }, + decision: "snooze", + snoozedUntil, + now, + }); + expect(snooze).toMatchObject({ companyId: "company-1", runId: "run-1", decision: "snooze", snoozedUntil }); + + const continueDecision = await recordWatchdogDecision({ + companyId: "company-1", + runId: "run-1", + actor: { type: "board" }, + decision: "continue", + now, + }); + expect(continueDecision.snoozedUntil?.toISOString()).toBe(new Date(now.getTime() + CONTINUE_REARM_MS).toISOString()); + + const dismissed = await recordWatchdogDecision({ + companyId: "company-1", + runId: "run-1", + actor: { type: "board" }, + decision: "dismissed_false_positive", + now, + }); + expect(dismissed).toMatchObject({ decision: "dismissed_false_positive", snoozedUntil: null }); + }); + + it("rejects an agent decision with no target evaluation issue", async () => { + const recordWatchdogDecision = createRecordWatchdogDecision({ + reader: makeReader({ findRunForCompany: async () => makeRun() }), + writer: makeDecisionWriter(), + continueRearmMs: CONTINUE_REARM_MS, + }); + + await expect(recordWatchdogDecision({ + companyId: "company-1", + runId: "run-1", + actor: { type: "agent", agentId: "agent-9" }, + decision: "continue", + now: new Date("2026-01-01T05:00:00.000Z"), + })).rejects.toMatchObject({ code: "evaluation_issue_required" }); + }); + + it("rejects an agent decision when the agent is not the evaluation issue's assignee", async () => { + const reader = makeReader({ + findRunForCompany: async () => makeRun(), + findEvaluationIssueById: async () => ({ + id: "eval-1", + identifier: "PAP-2", + status: "todo", + assigneeAgentId: "agent-1", + companyId: "company-1", + originKind: "stale_active_run_evaluation", + originId: "run-1", + hiddenAt: null, + }), + }); + const recordWatchdogDecision = createRecordWatchdogDecision({ + reader, + writer: makeDecisionWriter(), + continueRearmMs: CONTINUE_REARM_MS, + }); + + await expect(recordWatchdogDecision({ + companyId: "company-1", + runId: "run-1", + actor: { type: "agent", agentId: "agent-9" }, + decision: "continue", + evaluationIssueId: "eval-1", + now: new Date("2026-01-01T05:00:00.000Z"), + })).rejects.toMatchObject({ code: "not_authorized" }); + }); + + it("accepts an agent decision when the agent is the assigned recovery owner", async () => { + const reader = makeReader({ + findRunForCompany: async () => makeRun(), + findEvaluationIssueById: async () => ({ + id: "eval-1", + identifier: "PAP-2", + status: "todo", + assigneeAgentId: "agent-1", + companyId: "company-1", + originKind: "stale_active_run_evaluation", + originId: "run-1", + hiddenAt: null, + }), + }); + const recordWatchdogDecision = createRecordWatchdogDecision({ + reader, + writer: makeDecisionWriter(), + continueRearmMs: CONTINUE_REARM_MS, + }); + + const decision = await recordWatchdogDecision({ + companyId: "company-1", + runId: "run-1", + actor: { type: "agent", agentId: "agent-1" }, + decision: "continue", + evaluationIssueId: "eval-1", + now: new Date("2026-01-01T05:00:00.000Z"), + }); + + expect(decision).toMatchObject({ evaluationIssueId: "eval-1", createdByAgentId: "agent-1" }); + }); + + it("reports run_not_found when the run does not exist in the company", async () => { + const recordWatchdogDecision = createRecordWatchdogDecision({ + reader: makeReader({ findRunForCompany: async () => null }), + writer: makeDecisionWriter(), + continueRearmMs: CONTINUE_REARM_MS, + }); + + await expect(recordWatchdogDecision({ + companyId: "company-1", + runId: "missing-run", + actor: { type: "board" }, + decision: "snooze", + snoozedUntil: new Date("2026-01-01T06:00:00.000Z"), + now: new Date("2026-01-01T05:00:00.000Z"), + })).rejects.toMatchObject({ code: "run_not_found" }); + }); + + it("reports evaluation_issue_not_found without encoding an HTTP status", async () => { + const recordWatchdogDecision = createRecordWatchdogDecision({ + reader: makeReader({ findRunForCompany: async () => makeRun() }), + writer: makeDecisionWriter(), + continueRearmMs: CONTINUE_REARM_MS, + }); + + const error = await recordWatchdogDecision({ + companyId: "company-1", + runId: "run-1", + actor: { type: "board" }, + decision: "continue", + evaluationIssueId: "missing-evaluation", + now: new Date("2026-01-01T05:00:00.000Z"), + }).catch((caught: unknown) => caught); + + expect(error).toMatchObject({ code: "evaluation_issue_not_found" }); + expect(error).not.toHaveProperty("status"); + }); + }); + + function makeScanFoldUseCase(writerOverrides: Partial = {}) { + const processController: RunProcessController = { + cleanupRunProcess: async () => ({ + attempted: false, + outcome: "no_process_metadata", + adapterType: "codex_local", + }), + }; + const writer: WatchdogWriter = { + recordDecision: async () => { + throw new Error("not used in this test"); + }, + foldSourceResolvedRun: async () => ({ kind: "folded", evaluationIssueId: null }), + ...writerOverrides, + }; + return createFoldSourceResolvedRun({ writer, processController }); + } + + describe("createScanSilentActiveRuns", () => { + it("skips suppressed runs and keeps the created and escalated counters at zero", async () => { + const run = makeRun({ sourceIssueId: "issue-1" }); + const reader = makeReader({ + findCandidateSilentRuns: async () => [run], + findLatestDecision: async () => ({ + quietUntilDecision: { decision: "snooze", snoozedUntil: new Date("2026-01-01T06:00:00.000Z") }, + dismissedFalsePositive: false, + }), + }); + const scanSilentActiveRuns = createScanSilentActiveRuns({ + reader, + foldSourceResolvedRun: makeScanFoldUseCase(), + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + }); + + const result = await scanSilentActiveRuns({ now: new Date("2026-01-01T05:00:00.000Z") }); + + expect(result).toMatchObject({ scanned: 1, snoozed: 1, created: 0, escalated: 0, skipped: 0 }); + }); + + it("applies the optional company scope and the issue-date cutoff to the candidate query", async () => { + let seenInput: unknown = null; + const now = new Date("2026-01-01T05:00:00.000Z"); + const cutoff = new Date("2026-01-01T00:00:00.000Z"); + const reader = makeReader({ + findCandidateSilentRuns: async (input) => { + seenInput = input; + return []; + }, + }); + const scanSilentActiveRuns = createScanSilentActiveRuns({ + reader, + foldSourceResolvedRun: makeScanFoldUseCase(), + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + }); + + await scanSilentActiveRuns({ now, companyId: "company-9", issueCreatedAtGte: cutoff }); + + expect(seenInput).toMatchObject({ + companyId: "company-9", + suspicionBefore: new Date(now.getTime() - SUSPICION_THRESHOLD_MS), + issueCreatedAtGte: cutoff, + }); + }); + + it("skips a run whose running agent belongs to a different company", async () => { + const run = makeRun({ sourceIssueId: "issue-1" }); + const reader = makeReader({ + findCandidateSilentRuns: async () => [run], + findRunningAgent: async () => ({ id: "agent-1", companyId: "other-company", adapterType: "codex_local" }), + }); + const scanSilentActiveRuns = createScanSilentActiveRuns({ + reader, + foldSourceResolvedRun: makeScanFoldUseCase(), + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + }); + + const result = await scanSilentActiveRuns({ now: new Date("2026-01-01T05:00:00.000Z") }); + + expect(result).toMatchObject({ scanned: 1, skipped: 1 }); + }); + + it("folds a run whose source issue has same-run terminal evidence", async () => { + const run = makeRun({ sourceIssueId: "issue-1" }); + const reader = makeReader({ + findCandidateSilentRuns: async () => [run], + findRunningAgent: async () => ({ id: "agent-1", companyId: "company-1", adapterType: "codex_local" }), + findSourceIssue: async () => ({ + id: "issue-1", + identifier: "PAP-1", + status: "done", + originKind: "manual", + isRecoveryOriginKind: false, + }), + findLatestSameRunTerminalEvidence: async () => ({ + kind: "activity", + id: "activity-1", + createdAt: new Date("2026-01-01T00:10:00.000Z"), + action: "issue.updated", + }), + }); + const scanSilentActiveRuns = createScanSilentActiveRuns({ + reader, + foldSourceResolvedRun: makeScanFoldUseCase({ + foldSourceResolvedRun: async () => ({ kind: "folded", evaluationIssueId: null }), + }), + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + }); + + const result = await scanSilentActiveRuns({ now: new Date("2026-01-01T05:00:00.000Z") }); + + expect(result).toMatchObject({ scanned: 1, folded: 1, skipped: 0 }); + }); + }); +}); diff --git a/server/src/modules/active-run-watchdog/application/use-cases.ts b/server/src/modules/active-run-watchdog/application/use-cases.ts new file mode 100644 index 0000000000..478c10fefc --- /dev/null +++ b/server/src/modules/active-run-watchdog/application/use-cases.ts @@ -0,0 +1,340 @@ +import { + classifySilenceLevel, + evaluateSuppression, + isTerminalIssueStatus, + shouldFoldTerminalSource, + silenceAgeMs, + silenceStartedAt, +} from "../domain/policy.js"; +import { WatchdogDecisionApplicationError } from "./types.js"; +import type { RunProcessController, WatchdogRunReader, WatchdogWriter } from "./ports.js"; +import type { + RunOutputSilenceSummary, + RunSnapshot, + ScanSilentActiveRunsResult, + SourceIssueSnapshot, + EvaluationIssueSnapshot, + FoldOutcome, + TerminalEvidence, + WatchdogDecisionActor, + WatchdogDecisionRecord, +} from "./types.js"; + +export type BuildRunOutputSilenceDeps = { + reader: WatchdogRunReader; + suspicionThresholdMs: number; + criticalThresholdMs: number; +}; + +export type BuildRunOutputSilenceRunInput = Pick< + RunSnapshot, + "id" | "companyId" | "status" | "lastOutputAt" | "lastOutputSeq" | "lastOutputStream" | "processStartedAt" | "startedAt" | "createdAt" +>; + +export function createBuildRunOutputSilence(deps: BuildRunOutputSilenceDeps) { + return async function buildRunOutputSilence( + run: BuildRunOutputSilenceRunInput, + now: Date, + ): Promise { + const [decisionState, evaluation] = await Promise.all([ + deps.reader.findLatestDecision(run.companyId, run.id, now), + deps.reader.findOpenStaleRunEvaluation(run.companyId, run.id), + ]); + const { dismissedFalsePositive, quietUntilDecision } = decisionState; + const isRunningRun = run.status === "running"; + const silenceAgeMsValue = isRunningRun ? silenceAgeMs(run, now) : null; + const level = classifySilenceLevel({ + isRunningRun, + silenceAgeMs: silenceAgeMsValue, + dismissedFalsePositive, + snoozed: Boolean(quietUntilDecision), + suspicionThresholdMs: deps.suspicionThresholdMs, + criticalThresholdMs: deps.criticalThresholdMs, + }); + + return { + lastOutputAt: run.lastOutputAt ?? null, + lastOutputSeq: run.lastOutputSeq ?? 0, + lastOutputStream: run.lastOutputStream === "stdout" || run.lastOutputStream === "stderr" + ? run.lastOutputStream + : null, + silenceStartedAt: silenceStartedAt(run), + silenceAgeMs: silenceAgeMsValue, + level, + suspicionThresholdMs: deps.suspicionThresholdMs, + criticalThresholdMs: deps.criticalThresholdMs, + snoozedUntil: dismissedFalsePositive ? null : quietUntilDecision?.snoozedUntil ?? null, + evaluationIssueId: evaluation?.id ?? null, + evaluationIssueIdentifier: evaluation?.identifier ?? null, + evaluationIssueAssigneeAgentId: evaluation?.assigneeAgentId ?? null, + }; + }; +} + +export type FoldSourceResolvedRunDeps = { + writer: WatchdogWriter; + processController: RunProcessController; +}; + +export type FoldSourceResolvedRunUseCaseInput = { + run: RunSnapshot; + runningAgentAdapterType: string; + sourceIssue: SourceIssueSnapshot; + evidence: TerminalEvidence; + existingEvaluation: EvaluationIssueSnapshot | null; + silenceStartedAt: Date | null; + silenceAgeMs: number | null; + now: Date; +}; + +export function createFoldSourceResolvedRun(deps: FoldSourceResolvedRunDeps) { + return async function foldSourceResolvedRun(input: FoldSourceResolvedRunUseCaseInput): Promise { + const cleanup = await deps.processController.cleanupRunProcess({ + runId: input.run.id, + adapterType: input.runningAgentAdapterType, + fallbackPid: input.run.processPid, + fallbackProcessGroupId: input.run.processGroupId, + }); + + return deps.writer.foldSourceResolvedRun(input.run.companyId, { + run: input.run, + sourceIssue: input.sourceIssue, + evidence: input.evidence, + existingEvaluation: input.existingEvaluation, + silenceStartedAt: input.silenceStartedAt, + silenceAgeMs: input.silenceAgeMs, + cleanup, + now: input.now, + }); + }; +} + +const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = "stale_active_run_evaluation"; + +export type RecordWatchdogDecisionDeps = { + reader: WatchdogRunReader; + writer: WatchdogWriter; + continueRearmMs: number; +}; + +export type RecordWatchdogDecisionUseCaseInput = { + companyId: string; + runId: string; + actor: WatchdogDecisionActor; + decision: "snooze" | "continue" | "dismissed_false_positive"; + evaluationIssueId?: string | null; + reason?: string | null; + snoozedUntil?: Date | null; + createdByRunId?: string | null; + now?: Date; +}; + +export function createRecordWatchdogDecision(deps: RecordWatchdogDecisionDeps) { + return async function recordWatchdogDecision( + input: RecordWatchdogDecisionUseCaseInput, + ): Promise { + const run = await deps.reader.findRunForCompany(input.companyId, input.runId); + if (!run) throw new WatchdogDecisionApplicationError("run_not_found", "Heartbeat run not found"); + + const evaluationIssue = input.evaluationIssueId + ? await deps.reader.findEvaluationIssueById(input.companyId, input.evaluationIssueId) + : null; + if (input.evaluationIssueId && !evaluationIssue) { + throw new WatchdogDecisionApplicationError("evaluation_issue_not_found", "Evaluation issue not found"); + } + if (input.actor.type === "agent" && !evaluationIssue) { + throw new WatchdogDecisionApplicationError( + "evaluation_issue_required", + "Agent watchdog decisions require the target evaluation issue", + ); + } + + const boardActor = input.actor.type === "board"; + const assignedRecoveryOwner = + input.actor.type === "agent" && + Boolean(input.actor.agentId) && + evaluationIssue !== null && + evaluationIssue.originKind === STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND && + evaluationIssue.originId === run.id && + evaluationIssue.hiddenAt === null && + !["done", "cancelled"].includes(evaluationIssue.status) && + evaluationIssue.assigneeAgentId === input.actor.agentId; + if (!boardActor && !assignedRecoveryOwner) { + throw new WatchdogDecisionApplicationError( + "not_authorized", + "Only the board or the assigned recovery owner can record watchdog decisions", + ); + } + + if (evaluationIssue && ( + evaluationIssue.originKind !== STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND || + evaluationIssue.originId !== run.id + )) { + throw new WatchdogDecisionApplicationError( + "evaluation_issue_mismatch", + "Watchdog decision evaluation issue is not bound to the target run", + ); + } + + const createdByRunId = input.actor.type === "agent" + ? input.actor.runId ?? input.createdByRunId ?? null + : input.actor.type === "board" + ? input.actor.runId ?? input.createdByRunId ?? null + : null; + if (createdByRunId) { + const creatorRun = await deps.reader.findRunForCompany(input.companyId, createdByRunId); + const sameAgent = input.actor.type !== "agent" || creatorRun?.agentId === input.actor.agentId; + if (!creatorRun || !sameAgent) { + throw new WatchdogDecisionApplicationError( + "creator_run_invalid", + "createdByRunId is not valid for this watchdog decision actor", + ); + } + } + + const decisionNow = input.now ?? new Date(); + const effectiveSnoozedUntil = input.decision === "snooze" + ? input.snoozedUntil ?? null + : input.decision === "continue" + ? input.snoozedUntil && input.snoozedUntil > decisionNow + ? input.snoozedUntil + : new Date(decisionNow.getTime() + deps.continueRearmMs) + : null; + + return deps.writer.recordDecision(input.companyId, { + runId: run.id, + actor: input.actor, + evaluationIssueId: input.evaluationIssueId ?? null, + decision: input.decision, + snoozedUntil: effectiveSnoozedUntil, + reason: input.reason ?? null, + createdByAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, + createdByUserId: input.actor.type === "board" ? input.actor.userId ?? null : null, + createdByRunId, + }); + }; +} + +export type ScanSilentActiveRunsDeps = { + reader: WatchdogRunReader; + foldSourceResolvedRun: ReturnType; + suspicionThresholdMs: number; +}; + +export type ScanSilentActiveRunsOptions = { + now?: Date; + companyId?: string; + issueCreatedAtGte?: Date | null; +}; + +type InspectOutcome = + | { kind: "skipped" } + | { kind: "existing"; evaluationIssueId: string } + | { kind: "folded"; evaluationIssueId: string | null }; + +export function createScanSilentActiveRuns(deps: ScanSilentActiveRunsDeps) { + async function resolveSourceIssue(run: RunSnapshot): Promise { + if (!run.sourceIssueId) return null; + return deps.reader.findSourceIssue(run.companyId, run.sourceIssueId); + } + + async function inspectSilentActiveRun(input: { + run: RunSnapshot; + now: Date; + dismissedFalsePositive: boolean; + }): Promise { + const runningAgent = await deps.reader.findRunningAgent(input.run.companyId, input.run.agentId); + if (!runningAgent || runningAgent.companyId !== input.run.companyId) return { kind: "skipped" }; + + const sourceIssue = await resolveSourceIssue(input.run); + const existing = await deps.reader.findOpenStaleRunEvaluation(input.run.companyId, input.run.id); + + if (evaluateSuppression({ recoveryOriginSource: sourceIssue?.isRecoveryOriginKind === true }).suppressed) { + return { kind: "skipped" }; + } + + const silenceStartedAtValue = silenceStartedAt(input.run); + if (sourceIssue) { + const terminalEvidence = isTerminalIssueStatus(sourceIssue.status) + ? await deps.reader.findLatestSameRunTerminalEvidence(input.run.companyId, { + runId: input.run.id, + sourceIssueId: sourceIssue.id, + sourceIssueStatus: sourceIssue.status, + evidenceAfter: silenceStartedAtValue, + }) + : null; + if (shouldFoldTerminalSource({ + sourceIssueStatus: sourceIssue.status, + hasSameRunTerminalEvidence: terminalEvidence !== null, + })) { + const foldOutcome = await deps.foldSourceResolvedRun({ + run: input.run, + runningAgentAdapterType: runningAgent.adapterType, + sourceIssue, + evidence: terminalEvidence!, + existingEvaluation: existing, + silenceStartedAt: silenceStartedAtValue, + silenceAgeMs: silenceAgeMs(input.run, input.now), + now: input.now, + }); + return foldOutcome.kind === "folded" + ? { kind: "folded", evaluationIssueId: foldOutcome.evaluationIssueId } + : { kind: "skipped" }; + } + } + + // Blocked source work can be intentionally quiet. The issue state already carries + // the durable waiting signal, so the scan has nothing to do. + if (evaluateSuppression({ blockedSource: sourceIssue?.status === "blocked" }).suppressed) { + return { kind: "skipped" }; + } + + if (evaluateSuppression({ dismissedFalsePositive: input.dismissedFalsePositive }).suppressed) { + return { kind: "skipped" }; + } + + return existing ? { kind: "existing", evaluationIssueId: existing.id } : { kind: "skipped" }; + } + + return async function scanSilentActiveRuns(opts?: ScanSilentActiveRunsOptions): Promise { + const now = opts?.now ?? new Date(); + const suspicionBefore = new Date(now.getTime() - deps.suspicionThresholdMs); + const candidates = await deps.reader.findCandidateSilentRuns({ + companyId: opts?.companyId, + suspicionBefore, + issueCreatedAtGte: opts?.issueCreatedAtGte, + }); + + const result: ScanSilentActiveRunsResult = { + scanned: candidates.length, + created: 0, + existing: 0, + escalated: 0, + folded: 0, + snoozed: 0, + skipped: 0, + evaluationIssueIds: [], + }; + + for (const run of candidates) { + const decisionState = await deps.reader.findLatestDecision(run.companyId, run.id, now); + if (evaluateSuppression({ snoozedOrContinued: Boolean(decisionState.quietUntilDecision) }).suppressed) { + result.snoozed += 1; + continue; + } + const outcome = await inspectSilentActiveRun({ + run, + now, + dismissedFalsePositive: decisionState.dismissedFalsePositive, + }); + if (outcome.kind === "existing") result.existing += 1; + else if (outcome.kind === "folded") result.folded += 1; + else result.skipped += 1; + if ("evaluationIssueId" in outcome && outcome.evaluationIssueId) { + result.evaluationIssueIds.push(outcome.evaluationIssueId); + } + } + + return result; + }; +} diff --git a/server/src/modules/active-run-watchdog/domain/policy.test.ts b/server/src/modules/active-run-watchdog/domain/policy.test.ts new file mode 100644 index 0000000000..6520afd27f --- /dev/null +++ b/server/src/modules/active-run-watchdog/domain/policy.test.ts @@ -0,0 +1,236 @@ +import { describe, expect, it } from "vitest"; +import { + classifySilenceLevel, + evaluateSuppression, + isTerminalIssueStatus, + shouldFoldTerminalSource, + silenceAgeMs, + silenceStartedAt, +} from "./policy.js"; + +describe("domain", () => { + const SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; + const CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; + + describe("silenceStartedAt / silenceAgeMs", () => { + it.each([ + { + name: "prefers the last output time over every other timestamp", + run: { + lastOutputAt: new Date("2026-01-01T00:10:00.000Z"), + processStartedAt: new Date("2026-01-01T00:05:00.000Z"), + startedAt: new Date("2026-01-01T00:04:00.000Z"), + createdAt: new Date("2026-01-01T00:00:00.000Z"), + }, + expected: "2026-01-01T00:10:00.000Z", + }, + { + name: "falls back to the process start time when there is no output", + run: { + lastOutputAt: null, + processStartedAt: new Date("2026-01-01T00:05:00.000Z"), + startedAt: new Date("2026-01-01T00:04:00.000Z"), + createdAt: new Date("2026-01-01T00:00:00.000Z"), + }, + expected: "2026-01-01T00:05:00.000Z", + }, + { + name: "falls back to the run start time when there is no process start", + run: { + lastOutputAt: null, + processStartedAt: null, + startedAt: new Date("2026-01-01T00:04:00.000Z"), + createdAt: new Date("2026-01-01T00:00:00.000Z"), + }, + expected: "2026-01-01T00:04:00.000Z", + }, + { + name: "falls back to the run creation time last", + run: { + lastOutputAt: null, + processStartedAt: null, + startedAt: null, + createdAt: new Date("2026-01-01T00:00:00.000Z"), + }, + expected: "2026-01-01T00:00:00.000Z", + }, + ])("selects the latest usable timestamp: $name", ({ run, expected }) => { + expect(silenceStartedAt(run)?.toISOString()).toBe(expected); + }); + + it("returns null when every timestamp is null", () => { + expect(silenceStartedAt({ lastOutputAt: null, processStartedAt: null, startedAt: null, createdAt: null })).toBeNull(); + }); + + it("calculates the silence age from the selected timestamp", () => { + const run = { + lastOutputAt: new Date("2026-01-01T00:00:00.000Z"), + processStartedAt: null, + startedAt: null, + createdAt: null, + }; + const now = new Date("2026-01-01T01:00:00.000Z"); + + expect(silenceAgeMs(run, now)).toBe(60 * 60 * 1000); + }); + + it("floors the silence age at zero when the clock reads before the start", () => { + const run = { + lastOutputAt: new Date("2026-01-01T01:00:00.000Z"), + processStartedAt: null, + startedAt: null, + createdAt: null, + }; + const now = new Date("2026-01-01T00:00:00.000Z"); + + expect(silenceAgeMs(run, now)).toBe(0); + }); + + it("returns null silence age when there is no usable timestamp", () => { + const run = { lastOutputAt: null, processStartedAt: null, startedAt: null, createdAt: null }; + + expect(silenceAgeMs(run, new Date("2026-01-01T00:00:00.000Z"))).toBeNull(); + }); + }); + + describe("classifySilenceLevel", () => { + function classifyInput(overrides: { + isRunningRun?: boolean; + silenceAgeMs?: number | null; + dismissedFalsePositive?: boolean; + snoozed?: boolean; + } = {}) { + return { + isRunningRun: true, + silenceAgeMs: null, + dismissedFalsePositive: false, + snoozed: false, + suspicionThresholdMs: SUSPICION_THRESHOLD_MS, + criticalThresholdMs: CRITICAL_THRESHOLD_MS, + ...overrides, + }; + } + + it.each([ + { + name: "not-applicable when the run is not running", + input: classifyInput({ isRunningRun: false, silenceAgeMs: CRITICAL_THRESHOLD_MS + 1 }), + expected: "not_applicable", + }, + { + name: "not-applicable when the run has a permanent false-positive dismissal", + input: classifyInput({ dismissedFalsePositive: true, silenceAgeMs: CRITICAL_THRESHOLD_MS + 1 }), + expected: "not_applicable", + }, + { + name: "snoozed when a snooze or continue decision is active", + input: classifyInput({ snoozed: true, silenceAgeMs: CRITICAL_THRESHOLD_MS + 1 }), + expected: "snoozed", + }, + { + name: "healthy below the suspicion threshold", + input: classifyInput({ silenceAgeMs: SUSPICION_THRESHOLD_MS - 1 }), + expected: "ok", + }, + { + name: "suspicious at or above the suspicion threshold", + input: classifyInput({ silenceAgeMs: SUSPICION_THRESHOLD_MS }), + expected: "suspicious", + }, + { + name: "critical at or above the critical threshold", + input: classifyInput({ silenceAgeMs: CRITICAL_THRESHOLD_MS }), + expected: "critical", + }, + { + name: "healthy when there is no silence age yet", + input: classifyInput(), + expected: "ok", + }, + ])("classifies: $name", ({ input, expected }) => { + expect(classifySilenceLevel(input)).toBe(expected); + }); + }); + + describe("evaluateSuppression", () => { + it("suppresses a snoozed run until the snooze expires", () => { + expect(evaluateSuppression({ snoozedOrContinued: true })).toEqual({ + suppressed: true, + reason: "snoozed", + }); + }); + + it("re-arms the watchdog once the continue decision's snooze window has passed", () => { + expect(evaluateSuppression({ snoozedOrContinued: false })).toEqual({ suppressed: false }); + }); + + it("suppresses a run permanently after a false-positive decision", () => { + expect(evaluateSuppression({ dismissedFalsePositive: true })).toEqual({ + suppressed: true, + reason: "dismissed_false_positive", + }); + }); + + it("suppresses a blocked source", () => { + expect(evaluateSuppression({ blockedSource: true })).toEqual({ + suppressed: true, + reason: "blocked_source", + }); + }); + + it("suppresses a recovery-origin source", () => { + expect(evaluateSuppression({ recoveryOriginSource: true })).toEqual({ + suppressed: true, + reason: "recovery_origin_source", + }); + }); + + it("is not suppressed when no signal is set", () => { + expect(evaluateSuppression({})).toEqual({ suppressed: false }); + }); + + it("checks snoozed before recovery-origin, blocked-source, and dismissed-false-positive", () => { + expect( + evaluateSuppression({ + snoozedOrContinued: true, + recoveryOriginSource: true, + blockedSource: true, + dismissedFalsePositive: true, + }), + ).toEqual({ suppressed: true, reason: "snoozed" }); + }); + }); + + describe("isTerminalIssueStatus", () => { + it.each([ + { status: "done", expected: true }, + { status: "cancelled", expected: true }, + { status: "in_progress", expected: false }, + { status: "blocked", expected: false }, + { status: null, expected: false }, + { status: undefined, expected: false }, + ])("$status -> $expected", ({ status, expected }) => { + expect(isTerminalIssueStatus(status)).toBe(expected); + }); + }); + + describe("shouldFoldTerminalSource", () => { + it("folds a terminal source only with same-run terminal evidence", () => { + expect( + shouldFoldTerminalSource({ sourceIssueStatus: "done", hasSameRunTerminalEvidence: true }), + ).toBe(true); + }); + + it("does not fold a terminal source without same-run terminal evidence", () => { + expect( + shouldFoldTerminalSource({ sourceIssueStatus: "done", hasSameRunTerminalEvidence: false }), + ).toBe(false); + }); + + it("does not fold a non-terminal source even with evidence present", () => { + expect( + shouldFoldTerminalSource({ sourceIssueStatus: "in_progress", hasSameRunTerminalEvidence: true }), + ).toBe(false); + }); + }); +}); diff --git a/server/src/modules/active-run-watchdog/domain/policy.ts b/server/src/modules/active-run-watchdog/domain/policy.ts new file mode 100644 index 0000000000..e3a093f8b6 --- /dev/null +++ b/server/src/modules/active-run-watchdog/domain/policy.ts @@ -0,0 +1,91 @@ +export type RunSilenceTimestamps = { + lastOutputAt: Date | null; + processStartedAt: Date | null; + startedAt: Date | null; + createdAt: Date | null; +}; + +export type SilenceLevel = "not_applicable" | "ok" | "snoozed" | "suspicious" | "critical"; + +export type ClassifySilenceLevelInput = { + isRunningRun: boolean; + silenceAgeMs: number | null; + dismissedFalsePositive: boolean; + snoozed: boolean; + suspicionThresholdMs: number; + criticalThresholdMs: number; +}; + +/** + * Picks the timestamp the silence clock started counting from. The order is + * the last output time, then the process start time, then the run start + * time, then the run creation time. + */ +export function silenceStartedAt(run: RunSilenceTimestamps): Date | null { + return run.lastOutputAt ?? run.processStartedAt ?? run.startedAt ?? run.createdAt ?? null; +} + +export function silenceAgeMs(run: RunSilenceTimestamps, now: Date): number | null { + const startedAt = silenceStartedAt(run); + return startedAt ? Math.max(0, now.getTime() - startedAt.getTime()) : null; +} + +export function classifySilenceLevel(input: ClassifySilenceLevelInput): SilenceLevel { + if (!input.isRunningRun) return "not_applicable"; + if (input.dismissedFalsePositive) return "not_applicable"; + if (input.snoozed) return "snoozed"; + const age = input.silenceAgeMs ?? 0; + if (age >= input.criticalThresholdMs) return "critical"; + if (age >= input.suspicionThresholdMs) return "suspicious"; + return "ok"; +} + +export type SuppressionSignals = { + snoozedOrContinued?: boolean; + recoveryOriginSource?: boolean; + blockedSource?: boolean; + dismissedFalsePositive?: boolean; +}; + +export type SuppressionReason = + | "snoozed" + | "recovery_origin_source" + | "blocked_source" + | "dismissed_false_positive"; + +export type SuppressionResult = + | { suppressed: false } + | { suppressed: true; reason: SuppressionReason }; + +/** + * Decides whether a signal on a silent active run suppresses recovery work. + * The caller passes only the signals it has resolved at its call site; an + * unresolved signal must be left `undefined`, not `false`, so this function + * checks each rule in a fixed priority order and returns the first match. + */ +export function evaluateSuppression(signals: SuppressionSignals): SuppressionResult { + if (signals.snoozedOrContinued) return { suppressed: true, reason: "snoozed" }; + if (signals.recoveryOriginSource) return { suppressed: true, reason: "recovery_origin_source" }; + if (signals.blockedSource) return { suppressed: true, reason: "blocked_source" }; + if (signals.dismissedFalsePositive) return { suppressed: true, reason: "dismissed_false_positive" }; + return { suppressed: false }; +} + +export function isTerminalIssueStatus(status: string | null | undefined): boolean { + return status === "done" || status === "cancelled"; +} + +export type ShouldFoldTerminalSourceInput = { + sourceIssueStatus: string | null | undefined; + hasSameRunTerminalEvidence: boolean; +}; + +/** + * A terminal source issue folds the watchdog run only when durable, + * same-run evidence shows the issue already reached that terminal status + * from an action inside this run. A terminal status alone is not enough + * evidence; a different run or a different actor could have closed it. + */ +export function shouldFoldTerminalSource(input: ShouldFoldTerminalSourceInput): boolean { + return isTerminalIssueStatus(input.sourceIssueStatus) && input.hasSameRunTerminalEvidence; +} diff --git a/server/src/modules/active-run-watchdog/index.ts b/server/src/modules/active-run-watchdog/index.ts new file mode 100644 index 0000000000..0de9687ee3 --- /dev/null +++ b/server/src/modules/active-run-watchdog/index.ts @@ -0,0 +1,58 @@ +import type { Db } from "@paperclipai/db"; +import { createPostgresWatchdogAdapter } from "./adapters/postgres.js"; +import { createProcessAdapter } from "./adapters/process.js"; +import { + createBuildRunOutputSilence, + createFoldSourceResolvedRun, + createRecordWatchdogDecision, + createScanSilentActiveRuns, +} from "./application/use-cases.js"; + +export type ActiveRunWatchdogConfig = { + suspicionThresholdMs: number; + criticalThresholdMs: number; + continueRearmMs: number; +}; + +/** + * Composes the active-run output watchdog module: the Postgres adapter, + * the process adapter, and the four use cases. The recovery service holds + * the only caller. + */ +export function createActiveRunWatchdog(db: Db, config: ActiveRunWatchdogConfig) { + const postgresAdapter = createPostgresWatchdogAdapter(db); + const processController = createProcessAdapter(); + + const foldSourceResolvedRun = createFoldSourceResolvedRun({ + writer: postgresAdapter, + processController, + }); + + return { + buildRunOutputSilence: createBuildRunOutputSilence({ + reader: postgresAdapter, + suspicionThresholdMs: config.suspicionThresholdMs, + criticalThresholdMs: config.criticalThresholdMs, + }), + scanSilentActiveRuns: createScanSilentActiveRuns({ + reader: postgresAdapter, + foldSourceResolvedRun, + suspicionThresholdMs: config.suspicionThresholdMs, + }), + recordWatchdogDecision: createRecordWatchdogDecision({ + reader: postgresAdapter, + writer: postgresAdapter, + continueRearmMs: config.continueRearmMs, + }), + }; +} + +export type ActiveRunWatchdog = ReturnType; + +export type { + RunOutputSilenceSummary, + ScanSilentActiveRunsResult, + WatchdogDecisionActor, + WatchdogDecisionRecord, +} from "./application/types.js"; +export { WatchdogDecisionApplicationError } from "./application/types.js"; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index b3b1a92fd2..f5f31334c3 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -28,7 +28,7 @@ import { runningProcesses } from "../../adapters/index.js"; import { visibleIssueCondition } from "../issue-visibility.js"; import { forbidden, notFound } from "../../errors.js"; import { logger } from "../../middleware/logger.js"; -import { isPidAlive, isProcessGroupAlive, terminateLocalService } from "../local-service-supervisor.js"; +import { isPidAlive, isProcessGroupAlive } from "../local-service-supervisor.js"; import { redactSensitiveText } from "../../redaction.js"; import { isUniqueViolation } from "../../db-errors.js"; import { logActivity } from "../activity-log.js"; @@ -76,6 +76,12 @@ import { dispositionRepairDelayMs, DISPOSITION_REPAIR_MAX_ATTEMPTS, } from "./disposition-repair.js"; +import { + createActiveRunWatchdog, + WatchdogDecisionApplicationError, + type RunOutputSilenceSummary, + type WatchdogDecisionActor, +} from "../../modules/active-run-watchdog/index.js"; const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["interrupted", "failed", "cancelled", "timed_out"] as const; @@ -83,22 +89,11 @@ export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; export const ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS = 30 * 60 * 1000; const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueRecovery; -const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation; const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON = "execution_review_participant_recovery"; const STRANDED_BOARD_ESCALATION_POLICY = "board_escalation_no_takeover_v1"; const DISPOSITION_REPAIR_IDEMPOTENCY_INDEX = "agent_wakeup_requests_disposition_repair_idempotency_uq"; const RESOLVED_DEPENDENCY_WAKE_BACKSTOP_CANDIDATE_LIMIT = 500; -const SESSIONED_LOCAL_ADAPTERS = new Set([ - "claude_local", - "codex_local", - "cursor", - "gemini_local", - "hermes_local", - "kimi_local", - "opencode_local", - "pi_local", -]); // GGU-809: when a stranded `in_progress` issue would otherwise hit the // `isRepeatedProductiveContinuationRecovery` escalation path, exempt the @@ -307,25 +302,7 @@ function readConfigurationIncompleteFingerprint(latestRun: LatestIssueRun): stri return readNonEmptyString(payload?.fingerprint); } -type WatchdogDecisionActor = - | { type: "board"; userId?: string | null; runId?: string | null } - | { type: "agent"; agentId?: string | null; runId?: string | null } - | { type: "none" }; - -export type RunOutputSilenceSummary = { - lastOutputAt: Date | null; - lastOutputSeq: number; - lastOutputStream: "stdout" | "stderr" | null; - silenceStartedAt: Date | null; - silenceAgeMs: number | null; - level: "not_applicable" | "ok" | "suspicious" | "critical" | "snoozed"; - suspicionThresholdMs: number; - criticalThresholdMs: number; - snoozedUntil: Date | null; - evaluationIssueId: string | null; - evaluationIssueIdentifier: string | null; - evaluationIssueAssigneeAgentId: string | null; -}; +export type { RunOutputSilenceSummary, WatchdogDecisionActor }; function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; @@ -1250,76 +1227,11 @@ export function recoveryService( .then((rows) => rows[0]?.issuePrefix ?? "PAP"); } - function isTerminalIssueStatus(status: string | null | undefined) { - return status === "done" || status === "cancelled"; - } - - function silenceStartedAtForRun(run: Pick) { - return run.lastOutputAt ?? run.processStartedAt ?? run.startedAt ?? run.createdAt ?? null; - } - - function silenceAgeMsForRun(run: Pick, now = new Date()) { - const startedAt = silenceStartedAtForRun(run); - return startedAt ? Math.max(0, now.getTime() - startedAt.getTime()) : null; - } - - async function activeOutputDecisionState(companyId: string, runId: string, now = new Date()) { - const [quietUntilRows, dismissedRows] = await Promise.all([ - db - .select({ - decision: heartbeatRunWatchdogDecisions.decision, - snoozedUntil: heartbeatRunWatchdogDecisions.snoozedUntil, - }) - .from(heartbeatRunWatchdogDecisions) - .where( - and( - eq(heartbeatRunWatchdogDecisions.companyId, companyId), - eq(heartbeatRunWatchdogDecisions.runId, runId), - inArray(heartbeatRunWatchdogDecisions.decision, ["snooze", "continue"]), - gt(heartbeatRunWatchdogDecisions.snoozedUntil, now), - ), - ) - .orderBy(desc(heartbeatRunWatchdogDecisions.createdAt)) - .limit(1), - db - .select({ id: heartbeatRunWatchdogDecisions.id }) - .from(heartbeatRunWatchdogDecisions) - .where( - and( - eq(heartbeatRunWatchdogDecisions.companyId, companyId), - eq(heartbeatRunWatchdogDecisions.runId, runId), - eq(heartbeatRunWatchdogDecisions.decision, "dismissed_false_positive"), - ), - ) - .limit(1), - ]); - return { - dismissedFalsePositive: dismissedRows.length > 0, - quietUntilDecision: quietUntilRows[0] ?? null, - }; - } - - async function findOpenStaleRunEvaluation(companyId: string, runId: string) { - const [row] = await db - .select({ - id: issues.id, - identifier: issues.identifier, - status: issues.status, - assigneeAgentId: issues.assigneeAgentId, - }) - .from(issues) - .where( - and( - eq(issues.companyId, companyId), - eq(issues.originKind, STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND), - eq(issues.originId, runId), - visibleIssueCondition(), - notInArray(issues.status, ["done", "cancelled"]), - ), - ) - .limit(1); - return row ?? null; - } + const watchdog = createActiveRunWatchdog(db, { + suspicionThresholdMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, + criticalThresholdMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS, + continueRearmMs: ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS, + }); async function buildRunOutputSilence( run: Pick< @@ -1328,93 +1240,7 @@ export function recoveryService( >, now = new Date(), ): Promise { - const [decisionState, evaluation] = await Promise.all([ - activeOutputDecisionState(run.companyId, run.id, now), - findOpenStaleRunEvaluation(run.companyId, run.id), - ]); - const { dismissedFalsePositive, quietUntilDecision } = decisionState; - const silenceStartedAt = silenceStartedAtForRun(run); - const silenceAgeMs = run.status === "running" ? silenceAgeMsForRun(run, now) : null; - const level = run.status !== "running" - ? "not_applicable" - : dismissedFalsePositive - ? "not_applicable" - : quietUntilDecision - ? "snoozed" - : (silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS - ? "critical" - : (silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS - ? "suspicious" - : "ok"; - return { - lastOutputAt: run.lastOutputAt ?? null, - lastOutputSeq: run.lastOutputSeq ?? 0, - lastOutputStream: (run.lastOutputStream === "stdout" || run.lastOutputStream === "stderr") - ? run.lastOutputStream - : null, - silenceStartedAt, - silenceAgeMs, - level, - suspicionThresholdMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, - criticalThresholdMs: ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS, - snoozedUntil: dismissedFalsePositive ? null : quietUntilDecision?.snoozedUntil ?? null, - evaluationIssueId: evaluation?.id ?? null, - evaluationIssueIdentifier: evaluation?.identifier ?? null, - evaluationIssueAssigneeAgentId: evaluation?.assigneeAgentId ?? null, - }; - } - - async function resolveStaleRunSourceIssue(run: typeof heartbeatRuns.$inferSelect) { - const issueId = issueIdFromRunContext(run.contextSnapshot); - if (!issueId) return null; - const [issue] = await db - .select() - .from(issues) - .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId), visibleIssueCondition())) - .limit(1); - return issue ?? null; - } - - async function latestSameRunSourceTerminalEvidence(input: { - run: typeof heartbeatRuns.$inferSelect; - sourceIssue: typeof issues.$inferSelect; - evidenceAfter: Date | null; - }) { - if (!isTerminalIssueStatus(input.sourceIssue.status)) return null; - const after = input.evidenceAfter ?? input.run.startedAt ?? input.run.createdAt ?? null; - const activityPredicates = [ - eq(activityLog.companyId, input.run.companyId), - eq(activityLog.runId, input.run.id), - eq(activityLog.action, "issue.updated"), - eq(activityLog.entityType, "issue"), - eq(activityLog.entityId, input.sourceIssue.id), - sql`${activityLog.details} ->> 'status' = ${input.sourceIssue.status}`, - ]; - if (after) { - activityPredicates.push(gte(activityLog.createdAt, after)); - } - - const activity = await db - .select({ - id: activityLog.id, - createdAt: activityLog.createdAt, - action: activityLog.action, - }) - .from(activityLog) - .where(and(...activityPredicates)) - .orderBy(desc(activityLog.createdAt)) - .limit(1) - .then((rows) => rows[0] ?? null); - - if (activity) { - return { - kind: "activity" as const, - id: activity.id, - createdAt: activity.createdAt, - action: activity.action, - }; - } - return null; + return watchdog.buildRunOutputSilence(run, now); } async function appendRecoveryRunEvent( @@ -1437,360 +1263,8 @@ export function recoveryService( }); } - async function cleanupSourceResolvedRunProcess(input: { - run: typeof heartbeatRuns.$inferSelect; - runningAgent: typeof agents.$inferSelect; - }) { - if (!SESSIONED_LOCAL_ADAPTERS.has(input.runningAgent.adapterType)) { - return { - attempted: false, - outcome: "skipped_non_local_adapter", - adapterType: input.runningAgent.adapterType, - }; - } - - const running = runningProcesses.get(input.run.id); - const pid = running?.child.pid ?? input.run.processPid ?? null; - const processGroupId = running?.processGroupId ?? input.run.processGroupId ?? null; - if (typeof pid !== "number" && typeof processGroupId !== "number") { - return { - attempted: false, - outcome: "no_process_metadata", - adapterType: input.runningAgent.adapterType, - }; - } - - const wasAlive = - (typeof pid === "number" && isPidAlive(pid)) || - (typeof processGroupId === "number" && isProcessGroupAlive(processGroupId)); - if (!wasAlive) { - runningProcesses.delete(input.run.id); - return { - attempted: false, - outcome: "not_running", - adapterType: input.runningAgent.adapterType, - pid, - processGroupId, - }; - } - - try { - await terminateLocalService( - { - pid: typeof pid === "number" && Number.isInteger(pid) && pid > 0 - ? pid - : (processGroupId ?? 0), - processGroupId: typeof processGroupId === "number" && Number.isInteger(processGroupId) && processGroupId > 0 - ? processGroupId - : null, - }, - running ? { forceAfterMs: Math.max(1, running.graceSec) * 1000 } : undefined, - ); - runningProcesses.delete(input.run.id); - const stillAlive = - (typeof pid === "number" && isPidAlive(pid)) || - (typeof processGroupId === "number" && isProcessGroupAlive(processGroupId)); - return { - attempted: true, - outcome: stillAlive ? "termination_sent_still_running" : "terminated", - adapterType: input.runningAgent.adapterType, - pid, - processGroupId, - }; - } catch (error) { - return { - attempted: true, - outcome: "failed", - adapterType: input.runningAgent.adapterType, - pid, - processGroupId, - error: error instanceof Error ? error.message : String(error), - }; - } - } - - async function finalizeAgentAfterSourceResolvedRun(run: typeof heartbeatRuns.$inferSelect, status: "succeeded" | "cancelled") { - const [runningCountRow] = await db - .select({ count: sql`count(*)::int` }) - .from(heartbeatRuns) - .where(and(eq(heartbeatRuns.agentId, run.agentId), eq(heartbeatRuns.status, "running"))); - const runningCount = Number(runningCountRow?.count ?? 0); - const nextStatus = runningCount > 0 ? "running" : status === "succeeded" || status === "cancelled" ? "idle" : "error"; - await db - .update(agents) - .set({ - status: nextStatus, - lastHeartbeatAt: new Date(), - updatedAt: new Date(), - }) - .where(and(eq(agents.id, run.agentId), notInArray(agents.status, ["paused", "terminated"]))); - } - - async function foldSourceResolvedStaleRun(input: { - run: typeof heartbeatRuns.$inferSelect; - runningAgent: typeof agents.$inferSelect; - sourceIssue: typeof issues.$inferSelect; - evidence: Awaited>; - existingEvaluation: Awaited>; - silenceStartedAt: Date | null; - silenceAgeMs: number | null; - now: Date; - }) { - if (!input.evidence) return { kind: "skipped" as const }; - const cleanup = await cleanupSourceResolvedRunProcess({ run: input.run, runningAgent: input.runningAgent }); - const finalRunStatus = input.sourceIssue.status === "cancelled" ? "cancelled" : "succeeded"; - const resultJson = { - ...parseObject(input.run.resultJson), - sourceResolvedWatchdogFold: { - sourceIssueId: input.sourceIssue.id, - sourceIssueIdentifier: input.sourceIssue.identifier, - sourceIssueStatus: input.sourceIssue.status, - sameRunEvidenceKind: input.evidence.kind, - sameRunEvidenceId: input.evidence.id, - sameRunEvidenceAt: input.evidence.createdAt.toISOString(), - silenceStartedAt: input.silenceStartedAt?.toISOString() ?? null, - silenceAgeMs: input.silenceAgeMs, - evaluationIssueId: input.existingEvaluation?.id ?? null, - evaluationIssueIdentifier: input.existingEvaluation?.identifier ?? null, - cleanup, - }, - }; - const finalizedRun = await db.transaction(async (tx) => { - const [updatedRun] = await tx - .update(heartbeatRuns) - .set({ - status: finalRunStatus, - finishedAt: input.now, - error: null, - errorCode: null, - resultJson, - updatedAt: input.now, - }) - .where(and(eq(heartbeatRuns.id, input.run.id), eq(heartbeatRuns.companyId, input.run.companyId), eq(heartbeatRuns.status, "running"))) - .returning(); - if (!updatedRun) return null; - - if (input.run.wakeupRequestId) { - await tx - .update(agentWakeupRequests) - .set({ - status: finalRunStatus === "succeeded" ? "completed" : "cancelled", - finishedAt: input.now, - error: null, - updatedAt: input.now, - }) - .where(and(eq(agentWakeupRequests.id, input.run.wakeupRequestId), eq(agentWakeupRequests.companyId, input.run.companyId))); - } - - await tx - .update(issues) - .set({ - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, - updatedAt: input.now, - }) - .where( - and( - eq(issues.id, input.sourceIssue.id), - eq(issues.companyId, input.run.companyId), - eq(issues.executionRunId, input.run.id), - ), - ); - - return updatedRun; - }); - if (!finalizedRun) return { kind: "skipped" as const }; - // Telemetry is best-effort background work; it must not delay the - // watchdog fold below, so fire it and do not await it. - void emitAgentTaskRun(db, finalizedRun); - - if (input.existingEvaluation && !isTerminalIssueStatus(input.existingEvaluation.status)) { - await issuesSvc.update(input.existingEvaluation.id, { status: "done" }); - await issuesSvc.addComment(input.existingEvaluation.id, [ - "Source-resolved watchdog fold.", - "", - `- Source issue: ${input.sourceIssue.identifier ?? input.sourceIssue.id}`, - `- Run: \`${input.run.id}\``, - `- Same-run evidence: \`${input.evidence.kind}:${input.evidence.id}\` at ${input.evidence.createdAt.toISOString()}`, - "- Outcome: false positive; the source issue already reached a terminal disposition from this run.", - ].join("\n"), { runId: input.run.id }); - } - - const activeRecoveryAction = await recoveryActionsSvc.getActiveForIssue(input.run.companyId, input.sourceIssue.id); - if (activeRecoveryAction?.kind === "active_run_watchdog") { - await recoveryActionsSvc.resolveActiveForIssue({ - companyId: input.run.companyId, - sourceIssueId: input.sourceIssue.id, - actionId: activeRecoveryAction.id, - status: "resolved", - outcome: "false_positive", - resolutionNote: "Source issue reached a terminal disposition through durable same-run activity; watchdog folded as source-resolved.", - }); - } - - const [decision] = await db - .insert(heartbeatRunWatchdogDecisions) - .values({ - companyId: input.run.companyId, - runId: input.run.id, - evaluationIssueId: input.existingEvaluation?.id ?? null, - decision: "dismissed_false_positive", - reason: "Source issue already reached a terminal disposition through durable same-run activity.", - createdByRunId: input.run.id, - }) - .returning(); - - await appendRecoveryRunEvent(finalizedRun, { - level: cleanup.outcome === "failed" ? "warn" : "info", - message: "Source-resolved watchdog fold finalized stale active run", - payload: resultJson.sourceResolvedWatchdogFold, - }); - await logActivity(db, { - companyId: input.run.companyId, - actorType: "system", - actorId: "system", - agentId: input.run.agentId, - runId: input.run.id, - action: "heartbeat.output_stale_source_resolved", - entityType: "heartbeat_run", - entityId: input.run.id, - details: { - source: "recovery.scan_silent_active_runs", - sourceIssueId: input.sourceIssue.id, - sourceIssueIdentifier: input.sourceIssue.identifier, - sourceIssueStatus: input.sourceIssue.status, - evaluationIssueId: input.existingEvaluation?.id ?? null, - watchdogDecisionId: decision.id, - sameRunEvidenceKind: input.evidence.kind, - sameRunEvidenceId: input.evidence.id, - sameRunEvidenceAt: input.evidence.createdAt.toISOString(), - cleanup, - }, - }); - await finalizeAgentAfterSourceResolvedRun(finalizedRun, finalRunStatus); - return { kind: "folded" as const, evaluationIssueId: input.existingEvaluation?.id ?? null }; - } - - async function inspectSilentActiveRun(input: { - run: typeof heartbeatRuns.$inferSelect; - now: Date; - dismissedFalsePositive: boolean; - }) { - const runningAgent = await getAgent(input.run.agentId); - if (!runningAgent || runningAgent.companyId !== input.run.companyId) return { kind: "skipped" as const }; - const sourceIssue = await resolveStaleRunSourceIssue(input.run); - const existing = await findOpenStaleRunEvaluation(input.run.companyId, input.run.id); - if ( - sourceIssue && - Object.values(RECOVERY_ORIGIN_KINDS).includes( - sourceIssue.originKind as typeof RECOVERY_ORIGIN_KINDS[keyof typeof RECOVERY_ORIGIN_KINDS], - ) - ) { - return { kind: "skipped" as const }; - } - const silenceStartedAt = silenceStartedAtForRun(input.run); - if (sourceIssue && isTerminalIssueStatus(sourceIssue.status)) { - const terminalEvidence = await latestSameRunSourceTerminalEvidence({ - run: input.run, - sourceIssue, - evidenceAfter: silenceStartedAt, - }); - if (terminalEvidence) { - return foldSourceResolvedStaleRun({ - run: input.run, - runningAgent, - sourceIssue, - evidence: terminalEvidence, - existingEvaluation: existing, - silenceStartedAt, - silenceAgeMs: silenceAgeMsForRun(input.run, input.now), - now: input.now, - }); - } - } - - // Blocked source work can be intentionally quiet. The issue state already carries - // the durable waiting signal, so the cleanup scan has nothing to do. - if (sourceIssue?.status === "blocked") return { kind: "skipped" as const }; - - if (input.dismissedFalsePositive) { - return { kind: "skipped" as const }; - } - - return existing - ? { kind: "existing" as const, evaluationIssueId: existing.id } - : { kind: "skipped" as const }; - } - async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string; issueCreatedAtGte?: Date | null }) { - const now = opts?.now ?? new Date(); - const suspicionBefore = new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS); - let candidates = await db - .select() - .from(heartbeatRuns) - .where( - and( - opts?.companyId ? eq(heartbeatRuns.companyId, opts.companyId) : undefined, - eq(heartbeatRuns.status, "running"), - sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.processStartedAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) <= ${suspicionBefore.toISOString()}::timestamptz`, - ), - ) - .orderBy(asc(heartbeatRuns.createdAt)) - .limit(100); - - if (opts?.issueCreatedAtGte) { - const issueIds = [...new Set(candidates.flatMap((run) => { - const context = parseObject(run.contextSnapshot); - const issueId = context.issueId ?? context.taskId; - return typeof issueId === "string" && issueId.length > 0 ? [issueId] : []; - }))]; - const eligibleIssueIds = new Set( - issueIds.length > 0 - ? (await db.select({ id: issues.id }).from(issues).where(and( - inArray(issues.id, issueIds), - gte(issues.createdAt, opts.issueCreatedAtGte), - ))).map((issue) => issue.id) - : [], - ); - candidates = candidates.filter((run) => { - const context = parseObject(run.contextSnapshot); - const issueId = context.issueId ?? context.taskId; - return typeof issueId === "string" && eligibleIssueIds.has(issueId); - }); - } - - const result = { - scanned: candidates.length, - created: 0, - existing: 0, - escalated: 0, - folded: 0, - snoozed: 0, - skipped: 0, - evaluationIssueIds: [] as string[], - }; - - for (const run of candidates) { - const decisionState = await activeOutputDecisionState(run.companyId, run.id, now); - if (decisionState.quietUntilDecision) { - result.snoozed += 1; - continue; - } - const outcome = await inspectSilentActiveRun({ - run, - now, - dismissedFalsePositive: decisionState.dismissedFalsePositive, - }); - if (outcome.kind === "existing") result.existing += 1; - else if (outcome.kind === "folded") result.folded += 1; - else result.skipped += 1; - if ("evaluationIssueId" in outcome && outcome.evaluationIssueId) { - result.evaluationIssueIds.push(outcome.evaluationIssueId); - } - } - - return result; + return watchdog.scanSilentActiveRuns(opts); } async function recordWatchdogDecision(input: { @@ -1804,128 +1278,20 @@ export function recoveryService( now?: Date; }) { const [run] = await db - .select() + .select({ companyId: heartbeatRuns.companyId }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1); if (!run) throw notFound("Heartbeat run not found"); - - let evaluationIssue: { - id: string; - assigneeAgentId: string | null; - companyId: string; - originKind: string; - originId: string | null; - hiddenAt: Date | null; - status: string; - } | null = null; - if (input.evaluationIssueId) { - evaluationIssue = await db - .select({ - id: issues.id, - assigneeAgentId: issues.assigneeAgentId, - companyId: issues.companyId, - originKind: issues.originKind, - originId: issues.originId, - hiddenAt: issues.hiddenAt, - status: issues.status, - }) - .from(issues) - .where(and(eq(issues.id, input.evaluationIssueId), eq(issues.companyId, run.companyId))) - .then((rows) => rows[0] ?? null); - if (!evaluationIssue) throw notFound("Evaluation issue not found"); - } - - const boardActor = input.actor.type === "board"; - const assignedRecoveryOwner = - input.actor.type === "agent" && - Boolean(input.actor.agentId) && - evaluationIssue !== null && - evaluationIssue.originKind === STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND && - evaluationIssue.originId === run.id && - evaluationIssue.hiddenAt === null && - !["done", "cancelled"].includes(evaluationIssue.status) && - evaluationIssue?.assigneeAgentId === input.actor.agentId; - if (!boardActor && !assignedRecoveryOwner) { - throw forbidden("Only the board or the assigned recovery owner can record watchdog decisions"); - } - - if (evaluationIssue && ( - evaluationIssue.originKind !== STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND || - evaluationIssue.originId !== run.id - )) { - throw forbidden("Watchdog decision evaluation issue is not bound to the target run"); - } - - if (input.actor.type === "agent" && !evaluationIssue) { - throw forbidden("Agent watchdog decisions require the target evaluation issue"); - } - - const createdByRunId = input.actor.type === "agent" - ? input.actor.runId ?? input.createdByRunId ?? null - : input.actor.type === "board" - ? input.actor.runId ?? input.createdByRunId ?? null - : null; - if (createdByRunId) { - const [creatorRun] = await db - .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, createdByRunId)) - .limit(1); - const sameCompany = creatorRun?.companyId === run.companyId; - const sameAgent = input.actor.type !== "agent" || creatorRun?.agentId === input.actor.agentId; - if (!creatorRun || !sameCompany || !sameAgent) { - throw forbidden("createdByRunId is not valid for this watchdog decision actor"); + try { + return await watchdog.recordWatchdogDecision({ ...input, companyId: run.companyId }); + } catch (error) { + if (!(error instanceof WatchdogDecisionApplicationError)) throw error; + if (error.code === "run_not_found" || error.code === "evaluation_issue_not_found") { + throw notFound(error.message); } + throw forbidden(error.message); } - - const decisionNow = input.now ?? new Date(); - const effectiveSnoozedUntil = input.decision === "snooze" - ? input.snoozedUntil ?? null - : input.decision === "continue" - ? input.snoozedUntil && input.snoozedUntil > decisionNow - ? input.snoozedUntil - : new Date(decisionNow.getTime() + ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS) - : null; - - const [row] = await db - .insert(heartbeatRunWatchdogDecisions) - .values({ - companyId: run.companyId, - runId: run.id, - evaluationIssueId: input.evaluationIssueId ?? null, - decision: input.decision, - snoozedUntil: effectiveSnoozedUntil, - reason: input.reason ?? null, - createdByAgentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, - createdByUserId: input.actor.type === "board" ? input.actor.userId ?? null : null, - createdByRunId, - }) - .returning(); - - await logActivity(db, { - companyId: run.companyId, - actorType: input.actor.type === "agent" ? "agent" : "user", - actorId: input.actor.type === "agent" - ? input.actor.agentId ?? "agent" - : input.actor.type === "board" - ? input.actor.userId ?? "board" - : "unknown", - agentId: input.actor.type === "agent" ? input.actor.agentId ?? null : null, - runId: run.id, - action: input.decision === "snooze" ? "heartbeat.watchdog_snoozed" : "heartbeat.watchdog_decision_recorded", - entityType: "heartbeat_run", - entityId: run.id, - details: { - source: "recovery.record_watchdog_decision", - decision: input.decision, - evaluationIssueId: input.evaluationIssueId ?? null, - snoozedUntil: effectiveSnoozedUntil?.toISOString() ?? null, - reason: input.reason ?? null, - }, - }); - - return row; } function isStrandedIssueRecoveryIssue(issue: typeof issues.$inferSelect) {