diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 09c247f0082..222044cbcdf 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -37,7 +37,7 @@ These decisions close open questions from `SPEC.md` for V1. | Visibility | Full visibility to board and all agents in same company | | Communication | Tasks + comments only (no separate chat system) | | Task ownership | Single assignee; atomic checkout required for `in_progress` transition | -| Recovery | No automatic reassignment; work recovery stays manual/explicit | +| Recovery | No automatic reassignment; control-plane recovery may retry lost execution continuity once, then uses explicit recovery issues or human escalation | | Agent adapters | Built-in `process` and `http` adapters | | Auth | Mode-dependent human auth (`local_trusted` implicit board in current code; authenticated mode uses sessions), API keys for agents | | Budget period | Monthly UTC calendar window | @@ -395,7 +395,7 @@ Side effects: - entering `done` sets `completed_at` - entering `cancelled` sets `cancelled_at` -Detailed ownership, execution, blocker, and crash-recovery semantics are documented in `doc/execution-semantics.md`. +Detailed ownership, execution, blocker, active-run watchdog, and crash-recovery semantics are documented in `doc/execution-semantics.md`. ## 8.3 Approval Status diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 9cfdf1b952f..15eb1d3917d 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1,7 +1,7 @@ # Execution Semantics Status: Current implementation guide -Date: 2026-04-13 +Date: 2026-04-23 Audience: Product and engineering This document explains how Paperclip interprets issue assignment, issue status, execution runs, wakeups, parent/sub-issue structure, and blocker relationships. @@ -218,15 +218,79 @@ This is an active-work continuity recovery. Startup recovery and periodic recovery are different from normal wakeup delivery. -On startup and on the periodic recovery loop, Paperclip now does three things in sequence: +On startup and on the periodic recovery loop, Paperclip now does four things in sequence: 1. reap orphaned `running` runs 2. resume persisted `queued` runs 3. reconcile stranded assigned work +4. scan silent active runs and create or update explicit watchdog review issues -That last step is what closes the gap where issue state survives a crash but the wake/run path does not. +The stranded-work pass closes the gap where issue state survives a crash but the wake/run path does not. The silent-run scan covers the separate case where a live process exists but has stopped producing observable output. -## 10. What This Does Not Mean +## 10. Silent Active-Run Watchdog + +An active run can still be unhealthy even when its process is `running`. Paperclip treats prolonged output silence as a watchdog signal, not as proof that the run is failed. + +The recovery service owns this contract: + +- classify active-run output silence as `ok`, `suspicious`, `critical`, `snoozed`, or `not_applicable` +- collect bounded evidence from run logs, recent run events, child issues, and blockers +- preserve redaction and truncation before evidence is written to issue descriptions +- create at most one open `stale_active_run_evaluation` issue per run +- honor active snooze decisions before creating more review work +- build the `outputSilence` summary shown by live-run and active-run API responses + +Suspicious silence creates a medium-priority review issue for the selected recovery owner. Critical silence raises that review issue to high priority and blocks the source issue on the explicit evaluation task without cancelling the active process. + +Watchdog decisions are explicit operator/recovery-owner decisions: + +- `snooze` records a future quiet-until time and suppresses scan-created review work during that window +- `continue` records that the current evidence is acceptable but does not cancel or mutate the active run +- `dismissed_false_positive` records why the review was not actionable + +The board can record watchdog decisions. The assigned owner of the watchdog evaluation issue can also record them. Other agents cannot. + +## 11. Auto-Recover vs Explicit Recovery vs Human Escalation + +Paperclip uses three different recovery outcomes, depending on how much it can safely infer. + +### Auto-Recover + +Auto-recovery is allowed when ownership is clear and the control plane only lost execution continuity. + +Examples: + +- requeue one dispatch wake for an assigned `todo` issue whose latest run failed, timed out, or was cancelled +- requeue one continuation wake for an assigned `in_progress` issue whose live execution path disappeared +- assign an orphan blocker back to its creator when that blocker is already preventing other work + +Auto-recovery preserves the existing owner. It does not choose a replacement agent. + +### Explicit Recovery Issue + +Paperclip creates an explicit recovery issue when the system can identify a problem but cannot safely complete the work itself. + +Examples: + +- automatic stranded-work retry was already exhausted +- a dependency graph has an invalid/uninvokable owner, unassigned blocker, or invalid review participant +- an active run is silent past the watchdog threshold + +The source issue remains visible and blocked on the recovery issue when blocking is necessary for correctness. The recovery owner must restore a live path, resolve the source issue manually, or record the reason it is a false positive. + +### Human Escalation + +Human escalation is required when the next safe action depends on board judgment, budget/approval policy, or information unavailable to the control plane. + +Examples: + +- all candidate recovery owners are paused, terminated, pending approval, or budget-blocked +- the issue is human-owned rather than agent-owned +- the run is intentionally quiet but needs an operator decision before cancellation or continuation + +In these cases Paperclip should leave a visible issue/comment trail instead of silently retrying. + +## 12. What This Does Not Mean These semantics do not change V1 into an auto-reassignment system. @@ -240,9 +304,10 @@ The recovery model is intentionally conservative: - preserve ownership - retry once when the control plane lost execution continuity +- create explicit recovery work when the system can identify a bounded recovery owner/action - escalate visibly when the system cannot safely keep going -## 11. Practical Interpretation +## 13. Practical Interpretation For a board operator, the intended meaning is: 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 d8594e1dbfc..a170002731d 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 { ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, heartbeatService, } from "../services/heartbeat.ts"; +import { recoveryService } from "../services/recovery/service.ts"; const mockAdapterExecute = vi.hoisted(() => vi.fn(async () => ({ @@ -248,4 +249,60 @@ describeEmbeddedPostgres("active-run output watchdog", () => { expect(staleResult).toMatchObject({ created: 0, snoozed: 1 }); expect(noisyResult).toMatchObject({ scanned: 0, created: 0 }); }); + + it("records watchdog decisions through recovery owner authorization", async () => { + const now = new Date("2026-04-22T20:00:00.000Z"); + const { companyId, managerId, runId } = await seedRunningRun({ + now, + ageMs: ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS + 60_000, + }); + const heartbeat = heartbeatService(db); + const recovery = recoveryService(db, { enqueueWakeup: vi.fn() }); + + const scan = await heartbeat.scanSilentActiveRuns({ now, companyId }); + const evaluationIssueId = scan.evaluationIssueIds[0]; + expect(evaluationIssueId).toBeTruthy(); + + await expect( + recovery.recordWatchdogDecision({ + runId, + actor: { type: "agent", agentId: randomUUID() }, + decision: "continue", + evaluationIssueId, + reason: "not my recovery issue", + }), + ).rejects.toMatchObject({ status: 403 }); + + const snoozedUntil = new Date(now.getTime() + 60 * 60 * 1000); + const decision = await recovery.recordWatchdogDecision({ + runId, + actor: { type: "agent", agentId: managerId }, + decision: "snooze", + evaluationIssueId, + reason: "Long compile with no output", + snoozedUntil, + }); + + expect(decision).toMatchObject({ + runId, + evaluationIssueId, + decision: "snooze", + createdByAgentId: managerId, + }); + await expect(recovery.buildRunOutputSilence({ + id: runId, + companyId, + status: "running", + lastOutputAt: null, + lastOutputSeq: 0, + lastOutputStream: null, + processStartedAt: new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS - 60_000), + startedAt: new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS - 60_000), + createdAt: new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS - 60_000), + }, now)).resolves.toMatchObject({ + level: "snoozed", + snoozedUntil, + evaluationIssueId, + }); + }); }); diff --git a/server/src/routes/agents.ts b/server/src/routes/agents.ts index 1724c3baf6e..6c29745343b 100644 --- a/server/src/routes/agents.ts +++ b/server/src/routes/agents.ts @@ -2,7 +2,7 @@ import { Router, type Request, type Response } from "express"; import { generateKeyPairSync, randomUUID } from "node:crypto"; import path from "node:path"; import type { Db } from "@paperclipai/db"; -import { agents as agentsTable, companies, heartbeatRunWatchdogDecisions, heartbeatRuns, issues as issuesTable } from "@paperclipai/db"; +import { agents as agentsTable, companies, heartbeatRuns, issues as issuesTable } from "@paperclipai/db"; import { and, desc, eq, inArray, not, sql } from "drizzle-orm"; import { agentSkillSyncSchema, @@ -80,6 +80,8 @@ import { } from "../services/default-agent-instructions.js"; import { getTelemetryClient } from "../telemetry.js"; import { assertEnvironmentSelectionForCompany } from "./environment-selection.js"; +import { recoveryService } from "../services/recovery/service.js"; +import { assertEnvironmentSelectionForCompany } from "./environment-selection.js"; const RUN_LOG_DEFAULT_LIMIT_BYTES = 256_000; const RUN_LOG_MAX_LIMIT_BYTES = 1024 * 1024; @@ -135,6 +137,7 @@ export function agentRoutes(db: Db) { const approvalsSvc = approvalService(db); const budgets = budgetService(db); const heartbeat = heartbeatService(db); + const recovery = recoveryService(db, { enqueueWakeup: heartbeat.wakeup }); const issueApprovalsSvc = issueApprovalService(db); const secretsSvc = secretService(db); const instructions = agentInstructionsService(); @@ -2618,62 +2621,14 @@ export function agentRoutes(db: Db) { return; } - let evaluationIssue: { id: string; assigneeAgentId: string | null; companyId: string } | null = null; - if (evaluationIssueId) { - evaluationIssue = await db - .select({ - id: issuesTable.id, - assigneeAgentId: issuesTable.assigneeAgentId, - companyId: issuesTable.companyId, - }) - .from(issuesTable) - .where(and(eq(issuesTable.id, evaluationIssueId), eq(issuesTable.companyId, existing.companyId))) - .then((rows) => rows[0] ?? null); - if (!evaluationIssue) { - res.status(404).json({ error: "Evaluation issue not found" }); - return; - } - } - - const boardActor = req.actor.type === "board"; - const assignedRecoveryOwner = - req.actor.type === "agent" && - req.actor.agentId && - evaluationIssue?.assigneeAgentId === req.actor.agentId; - if (!boardActor && !assignedRecoveryOwner) { - throw forbidden("Only the board or the assigned recovery owner can record watchdog decisions"); - } - - const [row] = await db - .insert(heartbeatRunWatchdogDecisions) - .values({ - companyId: existing.companyId, - runId: existing.id, - evaluationIssueId, - decision, - snoozedUntil, - reason, - createdByAgentId: req.actor.type === "agent" ? req.actor.agentId ?? null : null, - createdByUserId: req.actor.type === "board" ? req.actor.userId ?? null : null, - createdByRunId: typeof req.body?.createdByRunId === "string" ? req.body.createdByRunId : null, - }) - .returning(); - - await logActivity(db, { - companyId: existing.companyId, - actorType: req.actor.type === "agent" ? "agent" : "user", - actorId: req.actor.type === "agent" ? req.actor.agentId ?? "agent" : req.actor.userId ?? "board", - agentId: req.actor.type === "agent" ? req.actor.agentId ?? null : null, + const row = await recovery.recordWatchdogDecision({ runId: existing.id, - action: decision === "snooze" ? "heartbeat.watchdog_snoozed" : "heartbeat.watchdog_decision_recorded", - entityType: "heartbeat_run", - entityId: existing.id, - details: { - decision, - evaluationIssueId, - snoozedUntil: snoozedUntil?.toISOString() ?? null, - reason, - }, + actor: req.actor, + decision: decision as "snooze" | "continue" | "dismissed_false_positive", + evaluationIssueId, + reason, + snoozedUntil, + createdByRunId: typeof req.body?.createdByRunId === "string" ? req.body.createdByRunId : null, }); res.json(row); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1f1447383f2..322eb8b56b7 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -27,12 +27,10 @@ import { agentTaskSessions, agentWakeupRequests, activityLog, - companies, companySkills as companySkillsTable, documentRevisions, issueDocuments, heartbeatRunEvents, - heartbeatRunWatchdogDecisions, heartbeatRuns, issueComments, issueRelations, @@ -109,7 +107,6 @@ import { import { resolveEnvironmentDriverConfigForRuntime } from "./environment-config.js"; import { instanceSettingsService } from "./instance-settings.js"; import { - RECOVERY_ORIGIN_KINDS, RUN_LIVENESS_CONTINUATION_REASON, buildRunLivenessContinuationIdempotencyKey, decideRunLivenessContinuation, @@ -157,11 +154,11 @@ const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_r const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const; -export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; -export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; +export { + ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS, + ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS, +} from "./recovery/service.js"; export const ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS = 60 * 1000; -const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation; -const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024; export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [ 2 * 60 * 1000, 10 * 60 * 1000, @@ -4139,58 +4136,8 @@ export function heartbeatService(db: Db) { readNonEmptyString(nestedContext.taskId); } - function staleActiveRunOriginFingerprint(companyId: string, runId: string) { - return `stale_active_run:${companyId}:${runId}`; - } - - 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 latestActiveOutputSnooze(companyId: string, runId: string, now = new Date()) { - const [row] = await db - .select() - .from(heartbeatRunWatchdogDecisions) - .where( - and( - eq(heartbeatRunWatchdogDecisions.companyId, companyId), - eq(heartbeatRunWatchdogDecisions.runId, runId), - eq(heartbeatRunWatchdogDecisions.decision, "snooze"), - gt(heartbeatRunWatchdogDecisions.snoozedUntil, now), - ), - ) - .orderBy(desc(heartbeatRunWatchdogDecisions.createdAt)) - .limit(1); - return row ?? null; - } - - async function findOpenStaleRunEvaluation(companyId: string, runId: string) { - const [row] = await db - .select({ - id: issues.id, - identifier: issues.identifier, - status: issues.status, - priority: issues.priority, - assigneeAgentId: issues.assigneeAgentId, - updatedAt: issues.updatedAt, - }) - .from(issues) - .where( - and( - eq(issues.companyId, companyId), - eq(issues.originKind, STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND), - eq(issues.originId, runId), - isNull(issues.hiddenAt), - notInArray(issues.status, ["done", "cancelled"]), - ), - ) - .limit(1); - return row ?? null; + async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string }) { + return recovery.scanSilentActiveRuns(opts); } async function buildRunOutputSilence( @@ -4200,502 +4147,7 @@ export function heartbeatService(db: Db) { >, now = new Date(), ) { - const [snooze, evaluation] = await Promise.all([ - latestActiveOutputSnooze(run.companyId, run.id, now), - findOpenStaleRunEvaluation(run.companyId, run.id), - ]); - const silenceStartedAt = silenceStartedAtForRun(run); - const silenceAgeMs = run.status === "running" ? silenceAgeMsForRun(run, now) : null; - const level = run.status !== "running" - ? "not_applicable" - : snooze - ? "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: snooze?.snoozedUntil ?? null, - evaluationIssueId: evaluation?.id ?? null, - evaluationIssueIdentifier: evaluation?.identifier ?? null, - }; - } - - function redactWatchdogEvidenceText(value: string, currentUserRedactionOptions: Awaited>) { - return redactCurrentUserText(value, currentUserRedactionOptions) - .replace(/\b(sk-[a-zA-Z0-9_-]{12,})\b/g, "[REDACTED_API_KEY]") - .replace(/\b([A-Za-z0-9_]*TOKEN[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]") - .replace(/\b([A-Za-z0-9_]*KEY[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]") - .replace(/\b([A-Za-z0-9_]*SECRET[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]"); - } - - function truncateEvidenceText(value: string, maxChars = 4000) { - if (value.length <= maxChars) return value; - return `${value.slice(value.length - maxChars)}\n[truncated earlier evidence]`; - } - - function formatDuration(ms: number | null) { - if (ms === null) return "unknown"; - const minutes = Math.floor(ms / 60_000); - if (minutes < 60) return `${minutes}m`; - const hours = Math.floor(minutes / 60); - const remainingMinutes = minutes % 60; - return remainingMinutes > 0 ? `${hours}h ${remainingMinutes}m` : `${hours}h`; - } - - function issueUiLink(issue: { identifier: string | null; id: string }, prefix: string) { - const label = issue.identifier ?? issue.id; - return `[${label}](/${prefix}/issues/${label})`; - } - - function runUiLink(run: { id: string; agentId: string }, prefix: string) { - return `[${run.id}](/${prefix}/agents/${run.agentId}/runs/${run.id})`; - } - - async function readRunLogTailForEvidence(run: typeof heartbeatRuns.$inferSelect) { - if (!run.logStore || !run.logRef || !run.logBytes) return ""; - try { - const offset = Math.max(0, run.logBytes - ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES); - const result = await runLogStore.read( - { store: run.logStore as "local_file", logRef: run.logRef }, - { offset, limitBytes: ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES }, - ); - return result.content; - } catch (err) { - logger.warn({ err, runId: run.id }, "failed to read stale-run watchdog evidence tail"); - return ""; - } - } - - 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), isNull(issues.hiddenAt))) - .limit(1); - return issue ?? null; - } - - function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) { - return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status)); - } - - async function resolveStaleRunOwnerAgentId(input: { - run: typeof heartbeatRuns.$inferSelect; - runningAgent: typeof agents.$inferSelect; - sourceIssue: typeof issues.$inferSelect | null; - }) { - const candidateIds: string[] = []; - if (input.sourceIssue?.assigneeAgentId) { - const sourceAssignee = await getAgent(input.sourceIssue.assigneeAgentId); - if (sourceAssignee?.reportsTo) candidateIds.push(sourceAssignee.reportsTo); - } - if (input.runningAgent.reportsTo) candidateIds.push(input.runningAgent.reportsTo); - const roleCandidates = await db - .select() - .from(agents) - .where(and(eq(agents.companyId, input.run.companyId), inArray(agents.role, ["cto", "ceo"]))) - .orderBy(sql`case when ${agents.role} = 'cto' then 0 else 1 end`, asc(agents.createdAt)); - candidateIds.push(...roleCandidates.map((agent) => agent.id)); - - const seen = new Set(); - for (const agentId of candidateIds) { - if (seen.has(agentId)) continue; - seen.add(agentId); - const candidate = await getAgent(agentId); - if (!candidate || candidate.companyId !== input.run.companyId) continue; - const budgetBlock = await budgets.getInvocationBlock(input.run.companyId, candidate.id, { - issueId: input.sourceIssue?.id ?? null, - projectId: input.sourceIssue?.projectId ?? null, - }); - if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; - } - - return null; - } - - async function collectStaleRunEvidence(input: { - run: typeof heartbeatRuns.$inferSelect; - runningAgent: typeof agents.$inferSelect; - sourceIssue: typeof issues.$inferSelect | null; - prefix: string; - now: Date; - }) { - const [tail, recentEvents, childIssues, blockers] = await Promise.all([ - readRunLogTailForEvidence(input.run), - db - .select({ - eventType: heartbeatRunEvents.eventType, - level: heartbeatRunEvents.level, - message: heartbeatRunEvents.message, - createdAt: heartbeatRunEvents.createdAt, - }) - .from(heartbeatRunEvents) - .where(and(eq(heartbeatRunEvents.companyId, input.run.companyId), eq(heartbeatRunEvents.runId, input.run.id))) - .orderBy(desc(heartbeatRunEvents.id)) - .limit(8), - input.sourceIssue - ? db - .select({ id: issues.id, identifier: issues.identifier, title: issues.title, status: issues.status }) - .from(issues) - .where(and(eq(issues.companyId, input.run.companyId), eq(issues.parentId, input.sourceIssue.id), isNull(issues.hiddenAt))) - .orderBy(desc(issues.updatedAt)) - .limit(8) - : Promise.resolve([]), - input.sourceIssue - ? db - .select({ id: issues.id, identifier: issues.identifier, title: issues.title, status: issues.status }) - .from(issueRelations) - .innerJoin(issues, eq(issueRelations.issueId, issues.id)) - .where( - and( - eq(issueRelations.companyId, input.run.companyId), - eq(issueRelations.relatedIssueId, input.sourceIssue.id), - eq(issueRelations.type, "blocks"), - ), - ) - .limit(8) - : Promise.resolve([]), - ]); - const currentUserRedactionOptions = await getCurrentUserRedactionOptions(); - const safeTail = truncateEvidenceText(redactWatchdogEvidenceText(tail, currentUserRedactionOptions)); - const silenceAgeMs = silenceAgeMsForRun(input.run, input.now); - return { - safeTail, - silenceAgeMs, - recentEvents: recentEvents.reverse().map((event) => ({ - eventType: event.eventType, - level: event.level, - createdAt: event.createdAt.toISOString(), - message: event.message ? truncateEvidenceText(redactWatchdogEvidenceText(event.message, currentUserRedactionOptions), 300) : null, - })), - childIssues, - blockers, - }; - } - - function buildStaleRunEvaluationDescription(input: { - run: typeof heartbeatRuns.$inferSelect; - runningAgent: typeof agents.$inferSelect; - sourceIssue: typeof issues.$inferSelect | null; - prefix: string; - evidence: Awaited>; - level: "suspicious" | "critical"; - now: Date; - }) { - const sourceIssue = input.sourceIssue - ? issueUiLink({ identifier: input.sourceIssue.identifier, id: input.sourceIssue.id }, input.prefix) - : "none"; - const recentEvents = input.evidence.recentEvents.length > 0 - ? input.evidence.recentEvents.map((event) => - `- ${event.createdAt} \`${event.eventType}\`${event.level ? ` ${event.level}` : ""}: ${event.message ?? "(no message)"}`, - ).join("\n") - : "- none"; - const childIssues = input.evidence.childIssues.length > 0 - ? input.evidence.childIssues.map((issue) => - `- ${issueUiLink({ identifier: issue.identifier, id: issue.id }, input.prefix)} \`${issue.status}\`: ${issue.title}`, - ).join("\n") - : "- none detected"; - const blockers = input.evidence.blockers.length > 0 - ? input.evidence.blockers.map((issue) => - `- ${issueUiLink({ identifier: issue.identifier, id: issue.id }, input.prefix)} \`${issue.status}\`: ${issue.title}`, - ).join("\n") - : "- none detected"; - return [ - `Paperclip detected ${input.level} output silence on an active heartbeat run.`, - "", - "## Run", - "", - `- Run: ${runUiLink(input.run, input.prefix)}`, - `- Agent: ${input.runningAgent.name} (${input.runningAgent.adapterType})`, - `- Invocation: ${input.run.invocationSource}${input.run.triggerDetail ? ` / ${input.run.triggerDetail}` : ""}`, - `- Source issue: ${sourceIssue}`, - `- Started at: ${input.run.startedAt?.toISOString() ?? "unknown"}`, - `- Process started at: ${input.run.processStartedAt?.toISOString() ?? "unknown"}`, - `- Last output at: ${input.run.lastOutputAt?.toISOString() ?? "none recorded"}`, - `- Last output sequence: ${input.run.lastOutputSeq ?? 0}`, - `- Silent for: ${formatDuration(input.evidence.silenceAgeMs)}`, - `- Thresholds: suspicious after ${formatDuration(ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS)}, critical after ${formatDuration(ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS)}`, - `- Process metadata: pid \`${input.run.processPid ?? "unknown"}\`, process group \`${input.run.processGroupId ?? "unknown"}\`, in-memory handle \`${runningProcesses.has(input.run.id) ? "yes" : "no"}\``, - "", - "## Last Output Excerpt", - "", - input.evidence.safeTail ? `\`\`\`text\n${input.evidence.safeTail}\n\`\`\`` : "_No run-log tail was available._", - "", - "## Recent Run Events", - "", - recentEvents, - "", - "## Related Work", - "", - "Active child issues:", - childIssues, - "", - "Current source blockers:", - blockers, - "", - "## Decision Checklist", - "", - "- Continue or snooze if the run is intentionally quiet.", - "- Ask the run owner for context if work may be delegated outside the transcript.", - "- Preserve artifacts, branch state, and useful output before cancellation.", - "- Cancel or recover through the explicit run recovery controls when authorized.", - "- Close this issue as a false positive only after recording the reason.", - ].join("\n"); - } - - function isUniqueStaleRunEvaluationConflict(error: unknown) { - if (!error || typeof error !== "object") return false; - const maybe = error as { code?: string; constraint?: string; message?: string }; - return maybe.code === "23505" && - ( - maybe.constraint === "issues_active_stale_run_evaluation_uq" || - typeof maybe.message === "string" && maybe.message.includes("issues_active_stale_run_evaluation_uq") - ); - } - - async function ensureSourceIssueBlockedByStaleEvaluation(input: { - sourceIssue: typeof issues.$inferSelect | null; - evaluationIssue: { id: string; identifier: string | null }; - run: typeof heartbeatRuns.$inferSelect; - }) { - if (!input.sourceIssue || ["done", "cancelled"].includes(input.sourceIssue.status)) return false; - const blockerIds = await existingBlockerIssueIds(input.sourceIssue.companyId, input.sourceIssue.id); - if (blockerIds.includes(input.evaluationIssue.id)) return false; - const nextBlockerIds = [...blockerIds, input.evaluationIssue.id]; - await issuesSvc.update(input.sourceIssue.id, { - ...(input.sourceIssue.status === "blocked" ? {} : { status: "blocked" }), - blockedByIssueIds: nextBlockerIds, - }); - await issuesSvc.addComment(input.sourceIssue.id, [ - "Paperclip detected critical output silence on this issue's active run.", - "", - `- Evaluation issue: ${input.evaluationIssue.identifier ?? input.evaluationIssue.id}`, - `- Run: \`${input.run.id}\``, - "", - "This blocks the source issue on the explicit review task without cancelling the active process.", - ].join("\n"), { runId: input.run.id }); - await logActivity(db, { - companyId: input.sourceIssue.companyId, - actorType: "system", - actorId: "system", - agentId: null, - runId: input.run.id, - action: "heartbeat.output_stale_escalated", - entityType: "issue", - entityId: input.sourceIssue.id, - details: { - source: "heartbeat.scan_silent_active_runs", - evaluationIssueId: input.evaluationIssue.id, - blockerIssueIds: nextBlockerIds, - }, - }); - return true; - } - - async function createOrUpdateStaleRunEvaluation(input: { - run: typeof heartbeatRuns.$inferSelect; - now: Date; - }) { - 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 [company] = await db - .select({ issuePrefix: companies.issuePrefix }) - .from(companies) - .where(eq(companies.id, input.run.companyId)) - .limit(1); - const prefix = company?.issuePrefix ?? "PAP"; - const evidence = await collectStaleRunEvidence({ - run: input.run, - runningAgent, - sourceIssue, - prefix, - now: input.now, - }); - const level = (evidence.silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS ? "critical" : "suspicious"; - const existing = await findOpenStaleRunEvaluation(input.run.companyId, input.run.id); - if (existing) { - if (level === "critical" && existing.priority !== "high") { - await issuesSvc.update(existing.id, { - priority: "high", - }); - await issuesSvc.addComment(existing.id, [ - "Critical output silence threshold crossed.", - "", - `- Run: \`${input.run.id}\``, - `- Silent for: ${formatDuration(evidence.silenceAgeMs)}`, - `- Last output at: ${input.run.lastOutputAt?.toISOString() ?? "none recorded"}`, - ].join("\n"), { runId: input.run.id }); - await ensureSourceIssueBlockedByStaleEvaluation({ - sourceIssue, - evaluationIssue: existing, - run: input.run, - }); - return { kind: "escalated" as const, evaluationIssueId: existing.id }; - } - if (level === "critical") { - await ensureSourceIssueBlockedByStaleEvaluation({ - sourceIssue, - evaluationIssue: existing, - run: input.run, - }); - } - return { kind: "existing" as const, evaluationIssueId: existing.id }; - } - - const ownerAgentId = await resolveStaleRunOwnerAgentId({ run: input.run, runningAgent, sourceIssue }); - const description = buildStaleRunEvaluationDescription({ - run: input.run, - runningAgent, - sourceIssue, - prefix, - evidence, - level, - now: input.now, - }); - let evaluation: Awaited>; - try { - evaluation = await issuesSvc.create(input.run.companyId, { - title: `Review silent active run for ${runningAgent.name}`, - description, - status: "todo", - priority: level === "critical" ? "high" : "medium", - parentId: sourceIssue && !["done", "cancelled"].includes(sourceIssue.status) ? sourceIssue.id : null, - projectId: sourceIssue?.projectId ?? null, - goalId: sourceIssue?.goalId ?? null, - billingCode: sourceIssue?.billingCode ?? null, - assigneeAgentId: ownerAgentId, - originKind: STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND, - originId: input.run.id, - originRunId: input.run.id, - originFingerprint: staleActiveRunOriginFingerprint(input.run.companyId, input.run.id), - }); - } catch (error) { - if (!isUniqueStaleRunEvaluationConflict(error)) throw error; - const raced = await findOpenStaleRunEvaluation(input.run.companyId, input.run.id); - if (!raced) throw error; - return { kind: "existing" as const, evaluationIssueId: raced.id }; - } - - await logActivity(db, { - companyId: input.run.companyId, - actorType: "system", - actorId: "system", - agentId: ownerAgentId, - runId: input.run.id, - action: "heartbeat.output_stale_detected", - entityType: "issue", - entityId: evaluation.id, - details: { - source: "heartbeat.scan_silent_active_runs", - level, - sourceIssueId: sourceIssue?.id ?? null, - silenceAgeMs: evidence.silenceAgeMs, - lastOutputAt: input.run.lastOutputAt?.toISOString() ?? null, - }, - }); - if (level === "critical") { - await ensureSourceIssueBlockedByStaleEvaluation({ - sourceIssue, - evaluationIssue: evaluation, - run: input.run, - }); - } - if (ownerAgentId) { - await enqueueWakeup(ownerAgentId, { - source: "assignment", - triggerDetail: "system", - reason: "issue_assigned", - payload: { - issueId: evaluation.id, - staleRunId: input.run.id, - sourceIssueId: sourceIssue?.id ?? null, - }, - requestedByActorType: "system", - requestedByActorId: null, - contextSnapshot: { - issueId: evaluation.id, - taskId: evaluation.id, - wakeReason: "issue_assigned", - source: STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND, - staleRunId: input.run.id, - sourceIssueId: sourceIssue?.id ?? null, - }, - }); - } - return { kind: "created" as const, evaluationIssueId: evaluation.id }; - } - - async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string }) { - const now = opts?.now ?? new Date(); - const suspicionBefore = new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS); - const 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); - - const result = { - scanned: candidates.length, - created: 0, - existing: 0, - escalated: 0, - snoozed: 0, - skipped: 0, - evaluationIssueIds: [] as string[], - }; - - for (const run of candidates) { - if (await latestActiveOutputSnooze(run.companyId, run.id, now)) { - result.snoozed += 1; - continue; - } - const outcome = await createOrUpdateStaleRunEvaluation({ run, now }); - if (outcome.kind === "created") result.created += 1; - else if (outcome.kind === "existing") result.existing += 1; - else if (outcome.kind === "escalated") result.escalated += 1; - else result.skipped += 1; - if ("evaluationIssueId" in outcome && outcome.evaluationIssueId) { - result.evaluationIssueIds.push(outcome.evaluationIssueId); - } - } - - return result; - } - - async function existingBlockerIssueIds(companyId: string, issueId: string) { - return db - .select({ blockerIssueId: issueRelations.issueId }) - .from(issueRelations) - .where( - and( - eq(issueRelations.companyId, companyId), - eq(issueRelations.relatedIssueId, issueId), - eq(issueRelations.type, "blocks"), - ), - ) - .then((rows) => rows.map((row) => row.blockerIssueId)); + return recovery.buildRunOutputSilence(run, now); } async function reconcileIssueGraphLiveness(opts?: { runId?: string | null }) { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 9f5f61c41b9..ead0eadeee7 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,19 +1,25 @@ -import { and, asc, desc, eq, inArray, isNull, notInArray, sql } from "drizzle-orm"; +import { and, asc, desc, eq, gt, inArray, isNull, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agents, agentWakeupRequests, companies, + heartbeatRunEvents, + heartbeatRunWatchdogDecisions, heartbeatRuns, issueRelations, issues, } from "@paperclipai/db"; import { parseObject, asBoolean, asNumber } from "../../adapters/utils.js"; +import { runningProcesses } from "../../adapters/index.js"; +import { forbidden, notFound } from "../../errors.js"; import { logger } from "../../middleware/logger.js"; +import { redactCurrentUserText } from "../../log-redaction.js"; import { logActivity } from "../activity-log.js"; import { budgetService } from "../budgets.js"; import { instanceSettingsService } from "../instance-settings.js"; import { issueService } from "../issues.js"; +import { getRunLogStore } from "../run-log-store.js"; import { RECOVERY_ORIGIN_KINDS, buildIssueGraphLivenessLeafKey, @@ -27,7 +33,11 @@ import { const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const; const ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_MIN_STALE_MS = 24 * 60 * 60 * 1000; +export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; +export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; +const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024; 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"; type RecoveryWakeupOptions = { @@ -51,6 +61,25 @@ type LatestIssueRun = Pick< "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" > | null; +type WatchdogDecisionActor = + | { type: "board"; userId?: string | null } + | { type: "agent"; agentId?: 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; +}; + function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; } @@ -109,6 +138,19 @@ function issueUiLink(issue: { identifier: string | null; id: string }, prefix: s return `[${label}](/${prefix}/issues/${label})`; } +function runUiLink(run: { id: string; agentId: string }, prefix: string) { + return `[${run.id}](/${prefix}/agents/${run.agentId}/runs/${run.id})`; +} + +function formatDuration(ms: number | null) { + if (ms === null) return "unknown"; + const minutes = Math.floor(ms / 60_000); + if (minutes < 60) return `${minutes}m`; + const hours = Math.floor(minutes / 60); + const remainingMinutes = minutes % 60; + return remainingMinutes > 0 ? `${hours}h ${remainingMinutes}m` : `${hours}h`; +} + function formatIssueLinksForComment(relations: Array<{ identifier?: string | null }>) { const identifiers = [ ...new Set( @@ -222,6 +264,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) const issuesSvc = issueService(db); const budgets = budgetService(db); const instanceSettings = instanceSettingsService(db); + const runLogStore = getRunLogStore(); + + const getCurrentUserRedactionOptions = async () => ({ + enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs, + }); async function getAgent(agentId: string) { return db.select().from(agents).where(eq(agents.id, agentId)).then((rows) => rows[0] ?? null); @@ -448,6 +495,603 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) .then((rows) => rows[0]?.issuePrefix ?? "PAP"); } + function staleActiveRunOriginFingerprint(companyId: string, runId: string) { + return `stale_active_run:${companyId}:${runId}`; + } + + 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 latestActiveOutputSnooze(companyId: string, runId: string, now = new Date()) { + const [row] = await db + .select() + .from(heartbeatRunWatchdogDecisions) + .where( + and( + eq(heartbeatRunWatchdogDecisions.companyId, companyId), + eq(heartbeatRunWatchdogDecisions.runId, runId), + eq(heartbeatRunWatchdogDecisions.decision, "snooze"), + gt(heartbeatRunWatchdogDecisions.snoozedUntil, now), + ), + ) + .orderBy(desc(heartbeatRunWatchdogDecisions.createdAt)) + .limit(1); + return row ?? null; + } + + async function findOpenStaleRunEvaluation(companyId: string, runId: string) { + const [row] = await db + .select({ + id: issues.id, + identifier: issues.identifier, + status: issues.status, + priority: issues.priority, + assigneeAgentId: issues.assigneeAgentId, + updatedAt: issues.updatedAt, + }) + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND), + eq(issues.originId, runId), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .limit(1); + return row ?? null; + } + + async function buildRunOutputSilence( + run: Pick< + typeof heartbeatRuns.$inferSelect, + "id" | "companyId" | "status" | "lastOutputAt" | "lastOutputSeq" | "lastOutputStream" | "processStartedAt" | "startedAt" | "createdAt" + >, + now = new Date(), + ): Promise { + const [snooze, evaluation] = await Promise.all([ + latestActiveOutputSnooze(run.companyId, run.id, now), + findOpenStaleRunEvaluation(run.companyId, run.id), + ]); + const silenceStartedAt = silenceStartedAtForRun(run); + const silenceAgeMs = run.status === "running" ? silenceAgeMsForRun(run, now) : null; + const level = run.status !== "running" + ? "not_applicable" + : snooze + ? "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: snooze?.snoozedUntil ?? null, + evaluationIssueId: evaluation?.id ?? null, + evaluationIssueIdentifier: evaluation?.identifier ?? null, + }; + } + + function redactWatchdogEvidenceText(value: string, currentUserRedactionOptions: Awaited>) { + return redactCurrentUserText(value, currentUserRedactionOptions) + .replace(/\b(sk-[a-zA-Z0-9_-]{12,})\b/g, "[REDACTED_API_KEY]") + .replace(/\b([A-Za-z0-9_]*TOKEN[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]") + .replace(/\b([A-Za-z0-9_]*KEY[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]") + .replace(/\b([A-Za-z0-9_]*SECRET[A-Za-z0-9_]*\s*=\s*)[^\s]+/gi, "$1[REDACTED]"); + } + + function truncateEvidenceText(value: string, maxChars = 4000) { + if (value.length <= maxChars) return value; + return `${value.slice(value.length - maxChars)}\n[truncated earlier evidence]`; + } + + async function readRunLogTailForEvidence(run: typeof heartbeatRuns.$inferSelect) { + if (!run.logStore || !run.logRef || !run.logBytes) return ""; + try { + const offset = Math.max(0, run.logBytes - ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES); + const result = await runLogStore.read( + { store: run.logStore as "local_file", logRef: run.logRef }, + { offset, limitBytes: ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES }, + ); + return result.content; + } catch (err) { + logger.warn({ err, runId: run.id }, "failed to read stale-run watchdog evidence tail"); + return ""; + } + } + + 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), isNull(issues.hiddenAt))) + .limit(1); + return issue ?? null; + } + + async function resolveStaleRunOwnerAgentId(input: { + run: typeof heartbeatRuns.$inferSelect; + runningAgent: typeof agents.$inferSelect; + sourceIssue: typeof issues.$inferSelect | null; + }) { + const candidateIds: string[] = []; + if (input.sourceIssue?.assigneeAgentId) { + const sourceAssignee = await getAgent(input.sourceIssue.assigneeAgentId); + if (sourceAssignee?.reportsTo) candidateIds.push(sourceAssignee.reportsTo); + } + if (input.runningAgent.reportsTo) candidateIds.push(input.runningAgent.reportsTo); + const roleCandidates = await db + .select() + .from(agents) + .where(and(eq(agents.companyId, input.run.companyId), inArray(agents.role, ["cto", "ceo"]))) + .orderBy(sql`case when ${agents.role} = 'cto' then 0 else 1 end`, asc(agents.createdAt)); + candidateIds.push(...roleCandidates.map((agent) => agent.id)); + + const seen = new Set(); + for (const agentId of candidateIds) { + if (seen.has(agentId)) continue; + seen.add(agentId); + const candidate = await getAgent(agentId); + if (!candidate || candidate.companyId !== input.run.companyId) continue; + const budgetBlock = await budgets.getInvocationBlock(input.run.companyId, candidate.id, { + issueId: input.sourceIssue?.id ?? null, + projectId: input.sourceIssue?.projectId ?? null, + }); + if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; + } + + return null; + } + + async function collectStaleRunEvidence(input: { + run: typeof heartbeatRuns.$inferSelect; + runningAgent: typeof agents.$inferSelect; + sourceIssue: typeof issues.$inferSelect | null; + prefix: string; + now: Date; + }) { + const [tail, recentEvents, childIssues, blockers] = await Promise.all([ + readRunLogTailForEvidence(input.run), + db + .select({ + eventType: heartbeatRunEvents.eventType, + level: heartbeatRunEvents.level, + message: heartbeatRunEvents.message, + createdAt: heartbeatRunEvents.createdAt, + }) + .from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, input.run.companyId), eq(heartbeatRunEvents.runId, input.run.id))) + .orderBy(desc(heartbeatRunEvents.id)) + .limit(8), + input.sourceIssue + ? db + .select({ id: issues.id, identifier: issues.identifier, title: issues.title, status: issues.status }) + .from(issues) + .where(and(eq(issues.companyId, input.run.companyId), eq(issues.parentId, input.sourceIssue.id), isNull(issues.hiddenAt))) + .orderBy(desc(issues.updatedAt)) + .limit(8) + : Promise.resolve([]), + input.sourceIssue + ? db + .select({ id: issues.id, identifier: issues.identifier, title: issues.title, status: issues.status }) + .from(issueRelations) + .innerJoin(issues, eq(issueRelations.issueId, issues.id)) + .where( + and( + eq(issueRelations.companyId, input.run.companyId), + eq(issueRelations.relatedIssueId, input.sourceIssue.id), + eq(issueRelations.type, "blocks"), + ), + ) + .limit(8) + : Promise.resolve([]), + ]); + const currentUserRedactionOptions = await getCurrentUserRedactionOptions(); + const safeTail = truncateEvidenceText(redactWatchdogEvidenceText(tail, currentUserRedactionOptions)); + const silenceAgeMs = silenceAgeMsForRun(input.run, input.now); + return { + safeTail, + silenceAgeMs, + recentEvents: recentEvents.reverse().map((event) => ({ + eventType: event.eventType, + level: event.level, + createdAt: event.createdAt.toISOString(), + message: event.message ? truncateEvidenceText(redactWatchdogEvidenceText(event.message, currentUserRedactionOptions), 300) : null, + })), + childIssues, + blockers, + }; + } + + function buildStaleRunEvaluationDescription(input: { + run: typeof heartbeatRuns.$inferSelect; + runningAgent: typeof agents.$inferSelect; + sourceIssue: typeof issues.$inferSelect | null; + prefix: string; + evidence: Awaited>; + level: "suspicious" | "critical"; + now: Date; + }) { + const sourceIssue = input.sourceIssue + ? issueUiLink({ identifier: input.sourceIssue.identifier, id: input.sourceIssue.id }, input.prefix) + : "none"; + const recentEvents = input.evidence.recentEvents.length > 0 + ? input.evidence.recentEvents.map((event) => + `- ${event.createdAt} \`${event.eventType}\`${event.level ? ` ${event.level}` : ""}: ${event.message ?? "(no message)"}`, + ).join("\n") + : "- none"; + const childIssues = input.evidence.childIssues.length > 0 + ? input.evidence.childIssues.map((issue) => + `- ${issueUiLink({ identifier: issue.identifier, id: issue.id }, input.prefix)} \`${issue.status}\`: ${issue.title}`, + ).join("\n") + : "- none detected"; + const blockers = input.evidence.blockers.length > 0 + ? input.evidence.blockers.map((issue) => + `- ${issueUiLink({ identifier: issue.identifier, id: issue.id }, input.prefix)} \`${issue.status}\`: ${issue.title}`, + ).join("\n") + : "- none detected"; + return [ + `Paperclip detected ${input.level} output silence on an active heartbeat run.`, + "", + "## Run", + "", + `- Run: ${runUiLink(input.run, input.prefix)}`, + `- Agent: ${input.runningAgent.name} (${input.runningAgent.adapterType})`, + `- Invocation: ${input.run.invocationSource}${input.run.triggerDetail ? ` / ${input.run.triggerDetail}` : ""}`, + `- Source issue: ${sourceIssue}`, + `- Started at: ${input.run.startedAt?.toISOString() ?? "unknown"}`, + `- Process started at: ${input.run.processStartedAt?.toISOString() ?? "unknown"}`, + `- Last output at: ${input.run.lastOutputAt?.toISOString() ?? "none recorded"}`, + `- Last output sequence: ${input.run.lastOutputSeq ?? 0}`, + `- Silent for: ${formatDuration(input.evidence.silenceAgeMs)}`, + `- Thresholds: suspicious after ${formatDuration(ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS)}, critical after ${formatDuration(ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS)}`, + `- Process metadata: pid \`${input.run.processPid ?? "unknown"}\`, process group \`${input.run.processGroupId ?? "unknown"}\`, in-memory handle \`${runningProcesses.has(input.run.id) ? "yes" : "no"}\``, + "", + "## Last Output Excerpt", + "", + input.evidence.safeTail ? `\`\`\`text\n${input.evidence.safeTail}\n\`\`\`` : "_No run-log tail was available._", + "", + "## Recent Run Events", + "", + recentEvents, + "", + "## Related Work", + "", + "Active child issues:", + childIssues, + "", + "Current source blockers:", + blockers, + "", + "## Decision Checklist", + "", + "- Continue or snooze if the run is intentionally quiet.", + "- Ask the run owner for context if work may be delegated outside the transcript.", + "- Preserve artifacts, branch state, and useful output before cancellation.", + "- Cancel or recover through the explicit run recovery controls when authorized.", + "- Close this issue as a false positive only after recording the reason.", + ].join("\n"); + } + + function isUniqueStaleRunEvaluationConflict(error: unknown) { + if (!error || typeof error !== "object") return false; + const maybe = error as { code?: string; constraint?: string; message?: string }; + return maybe.code === "23505" && + ( + maybe.constraint === "issues_active_stale_run_evaluation_uq" || + typeof maybe.message === "string" && maybe.message.includes("issues_active_stale_run_evaluation_uq") + ); + } + + async function ensureSourceIssueBlockedByStaleEvaluation(input: { + sourceIssue: typeof issues.$inferSelect | null; + evaluationIssue: { id: string; identifier: string | null }; + run: typeof heartbeatRuns.$inferSelect; + }) { + if (!input.sourceIssue || ["done", "cancelled"].includes(input.sourceIssue.status)) return false; + const blockerIds = await existingBlockerIssueIds(input.sourceIssue.companyId, input.sourceIssue.id); + if (blockerIds.includes(input.evaluationIssue.id)) return false; + const nextBlockerIds = [...blockerIds, input.evaluationIssue.id]; + await issuesSvc.update(input.sourceIssue.id, { + ...(input.sourceIssue.status === "blocked" ? {} : { status: "blocked" }), + blockedByIssueIds: nextBlockerIds, + }); + await issuesSvc.addComment(input.sourceIssue.id, [ + "Paperclip detected critical output silence on this issue's active run.", + "", + `- Evaluation issue: ${input.evaluationIssue.identifier ?? input.evaluationIssue.id}`, + `- Run: \`${input.run.id}\``, + "", + "This blocks the source issue on the explicit review task without cancelling the active process.", + ].join("\n"), { runId: input.run.id }); + await logActivity(db, { + companyId: input.sourceIssue.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: input.run.id, + action: "heartbeat.output_stale_escalated", + entityType: "issue", + entityId: input.sourceIssue.id, + details: { + source: "recovery.scan_silent_active_runs", + evaluationIssueId: input.evaluationIssue.id, + blockerIssueIds: nextBlockerIds, + }, + }); + return true; + } + + async function createOrUpdateStaleRunEvaluation(input: { + run: typeof heartbeatRuns.$inferSelect; + now: Date; + }) { + 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 prefix = await getCompanyIssuePrefix(input.run.companyId); + const evidence = await collectStaleRunEvidence({ + run: input.run, + runningAgent, + sourceIssue, + prefix, + now: input.now, + }); + const level = (evidence.silenceAgeMs ?? 0) >= ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS ? "critical" : "suspicious"; + const existing = await findOpenStaleRunEvaluation(input.run.companyId, input.run.id); + if (existing) { + if (level === "critical" && existing.priority !== "high") { + await issuesSvc.update(existing.id, { + priority: "high", + }); + await issuesSvc.addComment(existing.id, [ + "Critical output silence threshold crossed.", + "", + `- Run: \`${input.run.id}\``, + `- Silent for: ${formatDuration(evidence.silenceAgeMs)}`, + `- Last output at: ${input.run.lastOutputAt?.toISOString() ?? "none recorded"}`, + ].join("\n"), { runId: input.run.id }); + await ensureSourceIssueBlockedByStaleEvaluation({ + sourceIssue, + evaluationIssue: existing, + run: input.run, + }); + return { kind: "escalated" as const, evaluationIssueId: existing.id }; + } + if (level === "critical") { + await ensureSourceIssueBlockedByStaleEvaluation({ + sourceIssue, + evaluationIssue: existing, + run: input.run, + }); + } + return { kind: "existing" as const, evaluationIssueId: existing.id }; + } + + const ownerAgentId = await resolveStaleRunOwnerAgentId({ run: input.run, runningAgent, sourceIssue }); + const description = buildStaleRunEvaluationDescription({ + run: input.run, + runningAgent, + sourceIssue, + prefix, + evidence, + level, + now: input.now, + }); + let evaluation: Awaited>; + try { + evaluation = await issuesSvc.create(input.run.companyId, { + title: `Review silent active run for ${runningAgent.name}`, + description, + status: "todo", + priority: level === "critical" ? "high" : "medium", + parentId: sourceIssue && !["done", "cancelled"].includes(sourceIssue.status) ? sourceIssue.id : null, + projectId: sourceIssue?.projectId ?? null, + goalId: sourceIssue?.goalId ?? null, + billingCode: sourceIssue?.billingCode ?? null, + assigneeAgentId: ownerAgentId, + originKind: STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND, + originId: input.run.id, + originRunId: input.run.id, + originFingerprint: staleActiveRunOriginFingerprint(input.run.companyId, input.run.id), + }); + } catch (error) { + if (!isUniqueStaleRunEvaluationConflict(error)) throw error; + const raced = await findOpenStaleRunEvaluation(input.run.companyId, input.run.id); + if (!raced) throw error; + return { kind: "existing" as const, evaluationIssueId: raced.id }; + } + + await logActivity(db, { + companyId: input.run.companyId, + actorType: "system", + actorId: "system", + agentId: ownerAgentId, + runId: input.run.id, + action: "heartbeat.output_stale_detected", + entityType: "issue", + entityId: evaluation.id, + details: { + source: "recovery.scan_silent_active_runs", + level, + sourceIssueId: sourceIssue?.id ?? null, + silenceAgeMs: evidence.silenceAgeMs, + lastOutputAt: input.run.lastOutputAt?.toISOString() ?? null, + }, + }); + if (level === "critical") { + await ensureSourceIssueBlockedByStaleEvaluation({ + sourceIssue, + evaluationIssue: evaluation, + run: input.run, + }); + } + if (ownerAgentId) { + await deps.enqueueWakeup(ownerAgentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId: evaluation.id, + staleRunId: input.run.id, + sourceIssueId: sourceIssue?.id ?? null, + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: evaluation.id, + taskId: evaluation.id, + wakeReason: "issue_assigned", + source: STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND, + staleRunId: input.run.id, + sourceIssueId: sourceIssue?.id ?? null, + }, + }); + } + return { kind: "created" as const, evaluationIssueId: evaluation.id }; + } + + async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string }) { + const now = opts?.now ?? new Date(); + const suspicionBefore = new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS); + const 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); + + const result = { + scanned: candidates.length, + created: 0, + existing: 0, + escalated: 0, + snoozed: 0, + skipped: 0, + evaluationIssueIds: [] as string[], + }; + + for (const run of candidates) { + if (await latestActiveOutputSnooze(run.companyId, run.id, now)) { + result.snoozed += 1; + continue; + } + const outcome = await createOrUpdateStaleRunEvaluation({ run, now }); + if (outcome.kind === "created") result.created += 1; + else if (outcome.kind === "existing") result.existing += 1; + else if (outcome.kind === "escalated") result.escalated += 1; + else result.skipped += 1; + if ("evaluationIssueId" in outcome && outcome.evaluationIssueId) { + result.evaluationIssueIds.push(outcome.evaluationIssueId); + } + } + + return result; + } + + async function recordWatchdogDecision(input: { + runId: string; + actor: WatchdogDecisionActor; + decision: "snooze" | "continue" | "dismissed_false_positive"; + evaluationIssueId?: string | null; + reason?: string | null; + snoozedUntil?: Date | null; + createdByRunId?: string | null; + }) { + const [run] = await db + .select() + .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 } | null = null; + if (input.evaluationIssueId) { + evaluationIssue = await db + .select({ + id: issues.id, + assigneeAgentId: issues.assigneeAgentId, + companyId: issues.companyId, + }) + .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?.assigneeAgentId === input.actor.agentId; + if (!boardActor && !assignedRecoveryOwner) { + throw forbidden("Only the board or the assigned recovery owner can record watchdog decisions"); + } + + const [row] = await db + .insert(heartbeatRunWatchdogDecisions) + .values({ + companyId: run.companyId, + runId: run.id, + evaluationIssueId: input.evaluationIssueId ?? null, + decision: input.decision, + snoozedUntil: input.snoozedUntil ?? null, + 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: input.createdByRunId ?? null, + }) + .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: input.snoozedUntil?.toISOString() ?? null, + reason: input.reason ?? null, + }, + }); + + return row; + } + async function findOpenStrandedIssueRecoveryIssue(companyId: string, sourceIssueId: string) { return db .select() @@ -1397,6 +2041,9 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } return { + buildRunOutputSilence, + recordWatchdogDecision, + scanSilentActiveRuns, reconcileStrandedAssignedIssues, reconcileIssueGraphLiveness, readRecoveryTimerIntervalMs,