From 27e8d75829f609660334475fb6ec2a807165157f Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 22 Apr 2026 21:04:19 -0500 Subject: [PATCH] Move heartbeat recovery side effects into service --- server/src/services/heartbeat.ts | 1257 +------------------- server/src/services/recovery/index.ts | 3 + server/src/services/recovery/service.ts | 1404 +++++++++++++++++++++++ 3 files changed, 1411 insertions(+), 1253 deletions(-) create mode 100644 server/src/services/recovery/service.ts diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7257bc46ad8..1f1447383f2 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -72,10 +72,6 @@ import { classifyRunLiveness, type RunLivenessClassificationInput, } from "./run-liveness.js"; -import { - classifyIssueGraphLiveness, - type IssueLivenessFinding, -} from "./recovery/index.js"; import { logActivity, publishPluginDomainEvent, type LogActivityInput } from "./activity-log.js"; import { buildWorkspaceReadyComment, @@ -115,13 +111,12 @@ import { instanceSettingsService } from "./instance-settings.js"; import { RECOVERY_ORIGIN_KINDS, RUN_LIVENESS_CONTINUATION_REASON, - buildIssueGraphLivenessLeafKey, buildRunLivenessContinuationIdempotencyKey, decideRunLivenessContinuation, findExistingRunLivenessContinuationWake, - parseIssueGraphLivenessIncidentKey, readContinuationAttempt, } from "./recovery/index.js"; +import { recoveryService } from "./recovery/service.js"; import { redactCurrentUserText, redactCurrentUserValue } from "../log-redaction.js"; import { hasSessionCompactionThresholds, @@ -162,12 +157,10 @@ 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; -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; export const ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS = 60 * 1000; const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation; -const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueRecovery; const ACTIVE_RUN_OUTPUT_EVIDENCE_TAIL_BYTES = 8 * 1024; export const BOUNDED_TRANSIENT_HEARTBEAT_RETRY_DELAYS_MS = [ 2 * 60 * 1000, @@ -1969,6 +1962,7 @@ export function heartbeatService(db: Db) { cancelWorkForScope: cancelBudgetScopeWork, }; const budgets = budgetService(db, budgetHooks); + const recovery = recoveryService(db, { enqueueWakeup }); let unsafeTextProjectionPromise: Promise | null = null; async function hasUnsafeTextProjectionDatabase() { @@ -4128,600 +4122,8 @@ export function heartbeatService(db: Db) { } } - async function getLatestIssueRun(companyId: string, issueId: string) { - return db - .select({ - id: heartbeatRuns.id, - agentId: heartbeatRuns.agentId, - status: heartbeatRuns.status, - error: heartbeatRuns.error, - errorCode: heartbeatRuns.errorCode, - contextSnapshot: heartbeatRuns.contextSnapshot, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.companyId, companyId), - sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, - ), - ) - .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) - .limit(1) - .then((rows) => rows[0] ?? null); - } - - async function hasActiveExecutionPath(companyId: string, issueId: string) { - const [run, deferredWake] = await Promise.all([ - db - .select({ id: heartbeatRuns.id }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.companyId, companyId), - inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), - sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null), - db - .select({ id: agentWakeupRequests.id }) - .from(agentWakeupRequests) - .where( - and( - eq(agentWakeupRequests.companyId, companyId), - eq(agentWakeupRequests.status, "deferred_issue_execution"), - sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null), - ]); - - return Boolean(run || deferredWake); - } - - async function enqueueStrandedIssueRecovery(input: { - issueId: string; - agentId: string; - reason: "issue_assignment_recovery" | "issue_continuation_needed"; - retryReason: "assignment_recovery" | "issue_continuation_needed"; - source: string; - retryOfRunId?: string | null; - }) { - const queued = await enqueueWakeup(input.agentId, { - source: "automation", - triggerDetail: "system", - reason: input.reason, - payload: { - issueId: input.issueId, - ...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}), - }, - requestedByActorType: "system", - requestedByActorId: null, - contextSnapshot: { - issueId: input.issueId, - taskId: input.issueId, - wakeReason: input.reason, - retryReason: input.retryReason, - source: input.source, - ...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}), - }, - }); - - if (queued && input.retryOfRunId) { - return db - .update(heartbeatRuns) - .set({ - retryOfRunId: input.retryOfRunId, - updatedAt: new Date(), - }) - .where(eq(heartbeatRuns.id, queued.id)) - .returning() - .then((rows) => rows[0] ?? queued); - } - - return queued; - } - - function formatIssueLinksForComment(relations: Array<{ identifier?: string | null }>) { - const identifiers = [ - ...new Set( - relations - .map((relation) => relation.identifier) - .filter((identifier): identifier is string => Boolean(identifier)), - ), - ]; - if (identifiers.length === 0) return "another open issue"; - return identifiers - .slice(0, 5) - .map((identifier) => { - const prefix = identifier.split("-")[0] || "PAP"; - return `[${identifier}](/${prefix}/issues/${identifier})`; - }) - .join(", "); - } - - async function reconcileUnassignedBlockingIssues() { - const candidates = await db - .select({ - id: issues.id, - companyId: issues.companyId, - identifier: issues.identifier, - status: issues.status, - createdByAgentId: issues.createdByAgentId, - }) - .from(issueRelations) - .innerJoin(issues, eq(issueRelations.issueId, issues.id)) - .where( - and( - eq(issueRelations.type, "blocks"), - inArray(issues.status, ["todo", "blocked"]), - isNull(issues.assigneeAgentId), - isNull(issues.assigneeUserId), - sql`${issues.createdByAgentId} is not null`, - sql`exists ( - select 1 - from issues blocked_issue - where blocked_issue.id = ${issueRelations.relatedIssueId} - and blocked_issue.company_id = ${issues.companyId} - and blocked_issue.status not in ('done', 'cancelled') - )`, - ), - ); - - let assigned = 0; - let skipped = 0; - const issueIds: string[] = []; - const seen = new Set(); - - for (const candidate of candidates) { - if (seen.has(candidate.id)) continue; - seen.add(candidate.id); - - const creatorAgentId = candidate.createdByAgentId; - if (!creatorAgentId) { - skipped += 1; - continue; - } - const creatorAgent = await getAgent(creatorAgentId); - if ( - !creatorAgent || - creatorAgent.companyId !== candidate.companyId || - creatorAgent.status === "paused" || - creatorAgent.status === "terminated" || - creatorAgent.status === "pending_approval" - ) { - skipped += 1; - continue; - } - - const relations = await issuesSvc.getRelationSummaries(candidate.id); - const blockingLinks = formatIssueLinksForComment(relations.blocks); - const updated = await issuesSvc.update(candidate.id, { - assigneeAgentId: creatorAgent.id, - assigneeUserId: null, - }); - if (!updated) { - skipped += 1; - continue; - } - - await issuesSvc.addComment( - candidate.id, - [ - "## Assigned Orphan Blocker", - "", - `Paperclip found this issue is blocking ${blockingLinks} but had no assignee, so no heartbeat could pick it up.`, - "", - "- Assigned it back to the agent that created the blocker.", - "- Next action: resolve this blocker or reassign it to the right owner.", - ].join("\n"), - {}, - ); - - await logActivity(db, { - companyId: candidate.companyId, - actorType: "system", - actorId: "system", - agentId: null, - runId: null, - action: "issue.updated", - entityType: "issue", - entityId: candidate.id, - details: { - identifier: candidate.identifier, - assigneeAgentId: creatorAgent.id, - source: "heartbeat.reconcile_unassigned_blocking_issue", - }, - }); - - const queued = await enqueueWakeup(creatorAgent.id, { - source: "automation", - triggerDetail: "system", - reason: "issue_assigned", - payload: { - issueId: candidate.id, - mutation: "unassigned_blocker_recovery", - }, - requestedByActorType: "system", - requestedByActorId: null, - contextSnapshot: { - issueId: candidate.id, - taskId: candidate.id, - wakeReason: "issue_assigned", - source: "issue.unassigned_blocker_recovery", - }, - }); - - if (queued) { - assigned += 1; - issueIds.push(candidate.id); - } else { - skipped += 1; - } - } - - return { assigned, skipped, issueIds }; - } - - async function getCompanyIssuePrefix(companyId: string) { - return db - .select({ issuePrefix: companies.issuePrefix }) - .from(companies) - .where(eq(companies.id, companyId)) - .then((rows) => rows[0]?.issuePrefix ?? "PAP"); - } - - async function findOpenStrandedIssueRecoveryIssue(companyId: string, sourceIssueId: string) { - return db - .select() - .from(issues) - .where( - and( - eq(issues.companyId, companyId), - eq(issues.originKind, STRANDED_ISSUE_RECOVERY_ORIGIN_KIND), - eq(issues.originId, sourceIssueId), - isNull(issues.hiddenAt), - notInArray(issues.status, ["done", "cancelled"]), - ), - ) - .orderBy(desc(issues.createdAt)) - .limit(1) - .then((rows) => rows[0] ?? null); - } - - async function resolveStrandedIssueRecoveryOwnerAgentId(issue: typeof issues.$inferSelect) { - const candidateIds: string[] = []; - if (issue.assigneeAgentId) { - const assignee = await getAgent(issue.assigneeAgentId); - if (assignee?.reportsTo) candidateIds.push(assignee.reportsTo); - } - if (issue.createdByAgentId) { - const creator = await getAgent(issue.createdByAgentId); - if (creator?.reportsTo) candidateIds.push(creator.reportsTo); - candidateIds.push(issue.createdByAgentId); - } - - const roleCandidates = await db - .select() - .from(agents) - .where(and(eq(agents.companyId, issue.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)); - if (issue.assigneeAgentId) candidateIds.push(issue.assigneeAgentId); - - 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 !== issue.companyId) continue; - const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.id, { - issueId: issue.id, - projectId: issue.projectId, - }); - if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; - } - - return null; - } - - function buildStrandedIssueRecoveryDescription(input: { - issue: typeof issues.$inferSelect; - latestRun: Pick< - typeof heartbeatRuns.$inferSelect, - "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" - > | null; - previousStatus: "todo" | "in_progress"; - prefix: string; - }) { - const sourceIssue = issueUiLink({ identifier: input.issue.identifier, id: input.issue.id }, input.prefix); - const runLink = input.latestRun - ? `[\`${input.latestRun.id}\`](/${input.prefix}/agents/${input.latestRun.agentId}/runs/${input.latestRun.id})` - : "none"; - const retryReason = readNonEmptyString(parseObject(input.latestRun?.contextSnapshot)?.retryReason) ?? "unknown"; - const failureSummary = summarizeRunFailureForIssueComment(input.latestRun); - - return [ - "Paperclip exhausted automatic recovery for an assigned issue and created this explicit recovery task.", - "", - "## Source", - "", - `- Source issue: ${sourceIssue}`, - `- Previous source status: \`${input.previousStatus}\``, - `- Latest retry run: ${runLink}`, - `- Latest retry status: \`${input.latestRun?.status ?? "unknown"}\``, - `- Retry reason: \`${retryReason}\``, - failureSummary ? `- Failure: ${failureSummary.trim()}` : "- Failure: none recorded", - "", - "## Required Action", - "", - "- Inspect the latest run and source issue state.", - "- Fix the runtime/adapter problem, reassign the source issue, or convert the source issue into a clear manual-review state.", - "- When the source issue has a live execution path or has been intentionally resolved, mark this recovery issue done.", - ].join("\n"); - } - - async function ensureStrandedIssueRecoveryIssue(input: { - issue: typeof issues.$inferSelect; - latestRun: Pick< - typeof heartbeatRuns.$inferSelect, - "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" - > | null; - previousStatus: "todo" | "in_progress"; - }) { - const existing = await findOpenStrandedIssueRecoveryIssue(input.issue.companyId, input.issue.id); - if (existing) return existing; - - const ownerAgentId = await resolveStrandedIssueRecoveryOwnerAgentId(input.issue); - if (!ownerAgentId) return null; - - const prefix = await getCompanyIssuePrefix(input.issue.companyId); - const recovery = await issuesSvc.create(input.issue.companyId, { - title: `Recover stalled issue ${input.issue.identifier ?? input.issue.title}`, - description: buildStrandedIssueRecoveryDescription({ - issue: input.issue, - latestRun: input.latestRun, - previousStatus: input.previousStatus, - prefix, - }), - status: "todo", - priority: input.issue.priority, - parentId: input.issue.id, - projectId: input.issue.projectId, - goalId: input.issue.goalId, - assigneeAgentId: ownerAgentId, - originKind: STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, - originId: input.issue.id, - originRunId: input.latestRun?.id ?? null, - originFingerprint: [ - STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, - input.issue.companyId, - input.issue.id, - input.latestRun?.id ?? "no-run", - ].join(":"), - billingCode: input.issue.billingCode, - inheritExecutionWorkspaceFromIssueId: input.issue.id, - }); - - await enqueueWakeup(ownerAgentId, { - source: "assignment", - triggerDetail: "system", - reason: "issue_assigned", - payload: { - issueId: recovery.id, - sourceIssueId: input.issue.id, - strandedRunId: input.latestRun?.id ?? null, - }, - requestedByActorType: "system", - requestedByActorId: null, - contextSnapshot: { - issueId: recovery.id, - taskId: recovery.id, - wakeReason: "issue_assigned", - source: STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, - sourceIssueId: input.issue.id, - strandedRunId: input.latestRun?.id ?? null, - }, - }); - - return recovery; - } - - async function escalateStrandedAssignedIssue(input: { - issue: typeof issues.$inferSelect; - previousStatus: "todo" | "in_progress"; - latestRun: Pick< - typeof heartbeatRuns.$inferSelect, - "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" - > | null; - comment: string; - }) { - const recoveryIssue = await ensureStrandedIssueRecoveryIssue({ - issue: input.issue, - previousStatus: input.previousStatus, - latestRun: input.latestRun, - }); - const blockerIds = await existingBlockerIssueIds(input.issue.companyId, input.issue.id); - const nextBlockerIds = recoveryIssue - ? [...new Set([...blockerIds, recoveryIssue.id])] - : blockerIds; - const updated = await issuesSvc.update(input.issue.id, { - status: "blocked", - blockedByIssueIds: nextBlockerIds, - }); - if (!updated) return null; - - const prefix = await getCompanyIssuePrefix(input.issue.companyId); - const recoveryLine = recoveryIssue - ? [ - "", - `- Recovery issue: ${issueUiLink({ identifier: recoveryIssue.identifier, id: recoveryIssue.id }, prefix)}`, - "- Next action: the recovery owner should either restore a live execution path or record the manual resolution, then mark the recovery issue done.", - ].join("\n") - : ""; - - await issuesSvc.addComment(input.issue.id, `${input.comment}${recoveryLine}`, {}); - - await logActivity(db, { - companyId: input.issue.companyId, - actorType: "system", - actorId: "system", - agentId: null, - runId: null, - action: "issue.updated", - entityType: "issue", - entityId: input.issue.id, - details: { - identifier: input.issue.identifier, - status: "blocked", - previousStatus: input.previousStatus, - source: "heartbeat.reconcile_stranded_assigned_issue", - latestRunId: input.latestRun?.id ?? null, - latestRunStatus: input.latestRun?.status ?? null, - latestRunErrorCode: input.latestRun?.errorCode ?? null, - recoveryIssueId: recoveryIssue?.id ?? null, - blockerIssueIds: nextBlockerIds, - }, - }); - - return updated; - } - async function reconcileStrandedAssignedIssues() { - const candidates = await db - .select() - .from(issues) - .where( - and( - isNull(issues.assigneeUserId), - inArray(issues.status, ["todo", "in_progress"]), - sql`${issues.assigneeAgentId} is not null`, - ), - ); - - const result = { - dispatchRequeued: 0, - continuationRequeued: 0, - orphanBlockersAssigned: 0, - escalated: 0, - skipped: 0, - issueIds: [] as string[], - }; - - for (const issue of candidates) { - const agentId = issue.assigneeAgentId; - if (!agentId) { - result.skipped += 1; - continue; - } - - const agent = await getAgent(agentId); - if (!agent || agent.companyId !== issue.companyId) { - result.skipped += 1; - continue; - } - if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") { - result.skipped += 1; - continue; - } - - if (await hasActiveExecutionPath(issue.companyId, issue.id)) { - result.skipped += 1; - continue; - } - - const latestRun = await getLatestIssueRun(issue.companyId, issue.id); - if (issue.status === "todo") { - if (!latestRun || latestRun.status === "succeeded") { - result.skipped += 1; - continue; - } - - if (didAutomaticRecoveryFail(latestRun, "assignment_recovery")) { - const failureSummary = summarizeRunFailureForIssueComment(latestRun); - const updated = await escalateStrandedAssignedIssue({ - issue, - previousStatus: "todo", - latestRun, - comment: - "Paperclip automatically retried dispatch for this assigned `todo` issue after a lost wake/run, " + - `but it still has no live execution path.${failureSummary ?? ""} ` + - "Moving it to `blocked` so it is visible for intervention.", - }); - if (updated) { - result.escalated += 1; - result.issueIds.push(issue.id); - } else { - result.skipped += 1; - } - continue; - } - - const queued = await enqueueStrandedIssueRecovery({ - issueId: issue.id, - agentId, - reason: "issue_assignment_recovery", - retryReason: "assignment_recovery", - source: "issue.assignment_recovery", - retryOfRunId: latestRun.id, - }); - if (queued) { - result.dispatchRequeued += 1; - result.issueIds.push(issue.id); - } else { - result.skipped += 1; - } - continue; - } - - if (!latestRun && !issue.checkoutRunId && !issue.executionRunId) { - result.skipped += 1; - continue; - } - if (didAutomaticRecoveryFail(latestRun, "issue_continuation_needed")) { - const failureSummary = summarizeRunFailureForIssueComment(latestRun); - const updated = await escalateStrandedAssignedIssue({ - issue, - previousStatus: "in_progress", - latestRun, - comment: - "Paperclip automatically retried continuation for this assigned `in_progress` issue after its live " + - `execution disappeared, but it still has no live execution path.${failureSummary ?? ""} ` + - "Moving it to `blocked` so it is visible for intervention.", - }); - if (updated) { - result.escalated += 1; - result.issueIds.push(issue.id); - } else { - result.skipped += 1; - } - continue; - } - - const queued = await enqueueStrandedIssueRecovery({ - issueId: issue.id, - agentId, - reason: "issue_continuation_needed", - retryReason: "issue_continuation_needed", - source: "issue.continuation_recovery", - retryOfRunId: latestRun?.id ?? issue.checkoutRunId ?? null, - }); - if (queued) { - result.continuationRequeued += 1; - result.issueIds.push(issue.id); - } else { - result.skipped += 1; - } - } - - const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues(); - result.orphanBlockersAssigned = orphanBlockerRecovery.assigned; - result.skipped += orphanBlockerRecovery.skipped; - result.issueIds.push(...orphanBlockerRecovery.issueIds); - - return result; + return recovery.reconcileStrandedAssignedIssues(); } function issueIdFromRunContext(contextSnapshot: unknown) { @@ -5282,183 +4684,6 @@ export function heartbeatService(db: Db) { return result; } - async function collectIssueGraphLivenessFindings() { - const [issueRows, relationRows, agentRows, activeRunRows, activeIssueRunRows, wakeRows] = await Promise.all([ - db - .select({ - id: issues.id, - companyId: issues.companyId, - identifier: issues.identifier, - title: issues.title, - status: issues.status, - projectId: issues.projectId, - goalId: issues.goalId, - parentId: issues.parentId, - assigneeAgentId: issues.assigneeAgentId, - assigneeUserId: issues.assigneeUserId, - createdByAgentId: issues.createdByAgentId, - createdByUserId: issues.createdByUserId, - executionState: issues.executionState, - }) - .from(issues) - .where( - and( - isNull(issues.hiddenAt), - notInArray(issues.originKind, [RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation]), - ), - ), - db - .select({ - companyId: issueRelations.companyId, - blockerIssueId: issueRelations.issueId, - blockedIssueId: issueRelations.relatedIssueId, - }) - .from(issueRelations) - .where(eq(issueRelations.type, "blocks")), - db - .select({ - id: agents.id, - companyId: agents.companyId, - name: agents.name, - role: agents.role, - title: agents.title, - status: agents.status, - reportsTo: agents.reportsTo, - }) - .from(agents), - db - .select({ - companyId: heartbeatRuns.companyId, - agentId: heartbeatRuns.agentId, - status: heartbeatRuns.status, - contextSnapshot: heartbeatRuns.contextSnapshot, - }) - .from(heartbeatRuns) - .where(inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES])), - db - .select({ - companyId: issues.companyId, - agentId: heartbeatRuns.agentId, - status: heartbeatRuns.status, - issueId: issues.id, - }) - .from(issues) - .innerJoin(heartbeatRuns, eq(issues.executionRunId, heartbeatRuns.id)) - .where( - and( - isNull(issues.hiddenAt), - notInArray(issues.originKind, [RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation]), - inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), - ), - ), - db - .select({ - companyId: agentWakeupRequests.companyId, - agentId: agentWakeupRequests.agentId, - status: agentWakeupRequests.status, - payload: agentWakeupRequests.payload, - }) - .from(agentWakeupRequests) - .where(inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"])), - ]); - - return classifyIssueGraphLiveness({ - issues: issueRows, - relations: relationRows, - agents: agentRows, - activeRuns: activeRunRows.map((row) => ({ - companyId: row.companyId, - agentId: row.agentId, - status: row.status, - issueId: issueIdFromRunContext(row.contextSnapshot), - })).concat(activeIssueRunRows.map((row) => ({ - companyId: row.companyId, - agentId: row.agentId, - status: row.status, - issueId: row.issueId, - }))), - queuedWakeRequests: wakeRows.map((row) => ({ - companyId: row.companyId, - agentId: row.agentId, - status: row.status, - issueId: issueIdFromWakePayload(row.payload), - })), - }); - } - - async function findOpenLivenessEscalation(companyId: string, incidentKey: string) { - return db - .select() - .from(issues) - .where( - and( - eq(issues.companyId, companyId), - eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), - eq(issues.originId, incidentKey), - isNull(issues.hiddenAt), - notInArray(issues.status, ["done", "cancelled"]), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - } - - function parseLivenessIncidentKey(incidentKey: string | null | undefined) { - if (!incidentKey) return null; - return parseIssueGraphLivenessIncidentKey(incidentKey); - } - - function livenessRecoveryLeafIssueId(finding: IssueLivenessFinding) { - return finding.recoveryIssueId; - } - - function livenessRecoveryLeafFingerprint(finding: IssueLivenessFinding) { - return buildIssueGraphLivenessLeafKey({ - companyId: finding.companyId, - state: finding.state, - leafIssueId: livenessRecoveryLeafIssueId(finding), - }); - } - - function livenessRecoveryLeafKey(companyId: string, state: string, leafIssueId: string) { - return buildIssueGraphLivenessLeafKey({ companyId, state, leafIssueId }); - } - - async function findOpenLivenessRecoveryIssueForLeaf(finding: IssueLivenessFinding) { - const byFingerprint = await db - .select() - .from(issues) - .where( - and( - eq(issues.companyId, finding.companyId), - eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), - eq(issues.originFingerprint, livenessRecoveryLeafFingerprint(finding)), - isNull(issues.hiddenAt), - notInArray(issues.status, ["done", "cancelled"]), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - if (byFingerprint) return byFingerprint; - - const leafIssueId = livenessRecoveryLeafIssueId(finding); - const openRecoveries = await db - .select() - .from(issues) - .where( - and( - eq(issues.companyId, finding.companyId), - eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), - isNull(issues.hiddenAt), - notInArray(issues.status, ["done", "cancelled"]), - ), - ); - return openRecoveries.find((row) => { - const parsed = parseLivenessIncidentKey(row.originId); - return parsed?.state === finding.state && parsed.leafIssueId === leafIssueId; - }) ?? null; - } - async function existingBlockerIssueIds(companyId: string, issueId: string) { return db .select({ blockerIssueId: issueRelations.issueId }) @@ -5473,482 +4698,8 @@ export function heartbeatService(db: Db) { .then((rows) => rows.map((row) => row.blockerIssueId)); } - async function removeRecoveryBlockerFromSource(recovery: typeof issues.$inferSelect) { - const parsed = parseLivenessIncidentKey(recovery.originId); - if (!parsed) return false; - const sourceIssue = await db - .select() - .from(issues) - .where(and(eq(issues.companyId, recovery.companyId), eq(issues.id, parsed.issueId))) - .then((rows) => rows[0] ?? null); - if (!sourceIssue) return false; - - const blockerIds = await existingBlockerIssueIds(sourceIssue.companyId, sourceIssue.id); - if (!blockerIds.includes(recovery.id)) return false; - await issuesSvc.update(sourceIssue.id, { - blockedByIssueIds: blockerIds.filter((blockerId) => blockerId !== recovery.id), - }); - return true; - } - - async function hasActiveRunForIssueId(companyId: string, issueId: string) { - const [contextRun, issueRun] = await Promise.all([ - db - .select({ id: heartbeatRuns.id }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.companyId, companyId), - inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), - sql`(${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId} - OR ${heartbeatRuns.contextSnapshot}->>'taskId' = ${issueId})`, - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null), - db - .select({ id: heartbeatRuns.id }) - .from(issues) - .innerJoin(heartbeatRuns, eq(issues.executionRunId, heartbeatRuns.id)) - .where( - and( - eq(issues.companyId, companyId), - eq(issues.id, issueId), - inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null), - ]); - return Boolean(contextRun || issueRun); - } - - async function retireObsoleteLivenessRecoveryIssues(findings: IssueLivenessFinding[]) { - const currentIncidentKeys = new Set(findings.map((finding) => finding.incidentKey)); - const currentLeafKeys = new Set( - findings.map((finding) => - livenessRecoveryLeafKey( - finding.companyId, - finding.state, - livenessRecoveryLeafIssueId(finding), - ), - ), - ); - const openRecoveries = await db - .select() - .from(issues) - .where( - and( - eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), - isNull(issues.hiddenAt), - notInArray(issues.status, ["done", "cancelled"]), - ), - ); - const result = { - retired: 0, - activeSkipped: 0, - blockerRelationsRemoved: 0, - retiredIssueIds: [] as string[], - }; - - for (const recovery of openRecoveries) { - if (recovery.originId && currentIncidentKeys.has(recovery.originId)) continue; - const parsed = parseLivenessIncidentKey(recovery.originId); - if (!parsed) continue; - if ( - currentLeafKeys.has( - livenessRecoveryLeafKey(parsed.companyId, parsed.state, parsed.leafIssueId), - ) - ) { - continue; - } - if (await removeRecoveryBlockerFromSource(recovery)) { - result.blockerRelationsRemoved += 1; - } - if (await hasActiveRunForIssueId(recovery.companyId, recovery.id)) { - result.activeSkipped += 1; - continue; - } - await issuesSvc.update(recovery.id, { status: "cancelled" }); - result.retired += 1; - result.retiredIssueIds.push(recovery.id); - } - - return result; - } - - async function isLivenessFindingOldEnoughForAutoRecovery(finding: IssueLivenessFinding, now = new Date()) { - const issueIds = [...new Set(finding.dependencyPath.map((entry) => entry.issueId))]; - if (issueIds.length === 0) return false; - const rows = await db - .select({ id: issues.id, updatedAt: issues.updatedAt }) - .from(issues) - .where(and(eq(issues.companyId, finding.companyId), inArray(issues.id, issueIds))); - if (rows.length !== issueIds.length) return false; - const latestUpdatedAt = rows.reduce((latest, row) => - row.updatedAt > latest ? row.updatedAt : latest, - rows[0]!.updatedAt); - return now.getTime() - latestUpdatedAt.getTime() >= ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_MIN_STALE_MS; - } - - function isUniqueLivenessRecoveryConflict(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_liveness_recovery_incident_uq" || - maybe.constraint === "issues_active_liveness_recovery_leaf_uq" || - typeof maybe.message === "string" && - ( - maybe.message.includes("issues_active_liveness_recovery_incident_uq") || - maybe.message.includes("issues_active_liveness_recovery_leaf_uq") - ) - ); - } - - function formatDependencyPath(finding: IssueLivenessFinding) { - return finding.dependencyPath - .map((entry) => entry.identifier ?? entry.issueId) - .join(" -> "); - } - - function buildLivenessEscalationDescription(finding: IssueLivenessFinding) { - return [ - "Paperclip detected a harness-level issue graph liveness incident.", - "", - `- Incident key: \`${finding.incidentKey}\``, - `- Finding: \`${finding.state}\``, - `- Dependency path: ${formatDependencyPath(finding)}`, - `- Reason: ${finding.reason}`, - `- Requested action: ${finding.recommendedAction}`, - "", - "Resolve the blocked chain, then mark this escalation issue done so the original issue can resume when all blockers are cleared.", - ].join("\n"); - } - - function buildLivenessOriginalIssueComment(finding: IssueLivenessFinding, escalation: typeof issues.$inferSelect) { - return [ - "Paperclip detected a harness-level liveness incident in this issue's dependency graph.", - "", - `- Escalation issue: ${escalation.identifier ?? escalation.id}`, - `- Incident key: \`${finding.incidentKey}\``, - `- Finding: \`${finding.state}\``, - `- Dependency path: ${formatDependencyPath(finding)}`, - `- Reason: ${finding.reason}`, - `- Manager action requested: ${finding.recommendedAction}`, - "", - "This issue now keeps its existing blockers and is also blocked by the escalation issue so dependency wakeups remain explicit.", - ].join("\n"); - } - - async function resolveEscalationOwnerAgentId( - finding: IssueLivenessFinding, - issue: typeof issues.$inferSelect, - ) { - const detailedCandidates = finding.recommendedOwnerCandidates.length > 0 - ? finding.recommendedOwnerCandidates - : finding.recommendedOwnerCandidateAgentIds.map((agentId) => ({ - agentId, - reason: "ordered_invokable_fallback" as const, - sourceIssueId: finding.recoveryIssueId, - })); - const seenCandidates = new Set(); - const candidates = detailedCandidates.filter((candidate) => { - if (seenCandidates.has(candidate.agentId)) return false; - seenCandidates.add(candidate.agentId); - return true; - }); - const budgetBlockedCandidateAgentIds: string[] = []; - - for (const candidate of candidates) { - const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.agentId, { - issueId: issue.id, - projectId: issue.projectId, - }); - if (!budgetBlock) { - return { - agentId: candidate.agentId, - reason: candidate.reason, - sourceIssueId: candidate.sourceIssueId, - candidateAgentIds: candidates.map((entry) => entry.agentId), - candidateReasons: candidates.map((entry) => ({ - agentId: entry.agentId, - reason: entry.reason, - sourceIssueId: entry.sourceIssueId, - })), - budgetBlockedCandidateAgentIds, - }; - } - budgetBlockedCandidateAgentIds.push(candidate.agentId); - } - - return null; - } - - function shouldReuseRecoveryExecutionWorkspace(input: { - finding: IssueLivenessFinding; - recoveryIssue: typeof issues.$inferSelect; - ownerAgentId: string; - }) { - if (input.finding.recoveryIssueId === input.finding.issueId) return false; - return input.recoveryIssue.assigneeAgentId === input.ownerAgentId; - } - - async function ensureIssueBlockedByEscalation(input: { - issue: typeof issues.$inferSelect; - escalationIssueId: string; - finding: IssueLivenessFinding; - runId?: string | null; - }) { - const blockerIds = await existingBlockerIssueIds(input.issue.companyId, input.issue.id); - const nextBlockerIds = [...new Set([...blockerIds, input.escalationIssueId])]; - const update: Partial & { blockedByIssueIds: string[] } = { - blockedByIssueIds: nextBlockerIds, - }; - if (input.issue.status !== "blocked") { - update.status = "blocked"; - } - - const updated = await issuesSvc.update(input.issue.id, update); - if (!updated) return null; - - await logActivity(db, { - companyId: input.issue.companyId, - actorType: "system", - actorId: "system", - agentId: null, - runId: input.runId ?? null, - action: "issue.blockers.updated", - entityType: "issue", - entityId: input.issue.id, - details: { - source: "heartbeat.reconcile_issue_graph_liveness", - incidentKey: input.finding.incidentKey, - findingState: input.finding.state, - blockerIssueIds: nextBlockerIds, - escalationIssueId: input.escalationIssueId, - status: update.status ?? input.issue.status, - previousStatus: input.issue.status, - }, - }); - - return updated; - } - - async function createIssueGraphLivenessEscalation(input: { - finding: IssueLivenessFinding; - runId?: string | null; - }) { - const issue = await db - .select() - .from(issues) - .where(eq(issues.id, input.finding.issueId)) - .then((rows) => rows[0] ?? null); - if (!issue || issue.companyId !== input.finding.companyId) return { kind: "skipped" as const }; - - const recoveryIssue = await db - .select() - .from(issues) - .where(and(eq(issues.id, input.finding.recoveryIssueId), eq(issues.companyId, issue.companyId))) - .then((rows) => rows[0] ?? null); - if (!recoveryIssue) return { kind: "skipped" as const }; - - const existing = - await findOpenLivenessEscalation(issue.companyId, input.finding.incidentKey) ?? - await findOpenLivenessRecoveryIssueForLeaf(input.finding); - if (existing) { - await ensureIssueBlockedByEscalation({ - issue, - escalationIssueId: existing.id, - finding: input.finding, - runId: input.runId ?? null, - }); - return { kind: "existing" as const, escalationIssueId: existing.id }; - } - - const ownerSelection = await resolveEscalationOwnerAgentId(input.finding, recoveryIssue); - if (!ownerSelection) return { kind: "skipped" as const }; - const reuseRecoveryExecutionWorkspace = shouldReuseRecoveryExecutionWorkspace({ - finding: input.finding, - recoveryIssue, - ownerAgentId: ownerSelection.agentId, - }); - - let escalation: Awaited>; - try { - escalation = await issuesSvc.create(issue.companyId, { - title: `Unblock liveness incident for ${recoveryIssue.identifier ?? recoveryIssue.title}`, - description: buildLivenessEscalationDescription(input.finding), - status: "todo", - priority: "high", - parentId: recoveryIssue.id, - projectId: recoveryIssue.projectId, - goalId: recoveryIssue.goalId, - assigneeAgentId: ownerSelection.agentId, - originKind: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation, - originId: input.finding.incidentKey, - originFingerprint: livenessRecoveryLeafFingerprint(input.finding), - billingCode: recoveryIssue.billingCode, - ...(reuseRecoveryExecutionWorkspace - ? { inheritExecutionWorkspaceFromIssueId: recoveryIssue.id } - : { - executionWorkspaceId: null, - executionWorkspacePreference: null, - executionWorkspaceSettings: null, - }), - }); - } catch (error) { - if (!isUniqueLivenessRecoveryConflict(error)) throw error; - const raced = - await findOpenLivenessEscalation(issue.companyId, input.finding.incidentKey) ?? - await findOpenLivenessRecoveryIssueForLeaf(input.finding); - if (!raced) throw error; - await ensureIssueBlockedByEscalation({ - issue, - escalationIssueId: raced.id, - finding: input.finding, - runId: input.runId ?? null, - }); - return { kind: "existing" as const, escalationIssueId: raced.id }; - } - - await ensureIssueBlockedByEscalation({ - issue, - escalationIssueId: escalation.id, - finding: input.finding, - runId: input.runId ?? null, - }); - - await issuesSvc.addComment( - issue.id, - buildLivenessOriginalIssueComment(input.finding, escalation), - { runId: input.runId ?? null }, - ); - - await logActivity(db, { - companyId: issue.companyId, - actorType: "system", - actorId: "system", - agentId: ownerSelection.agentId, - runId: input.runId ?? null, - action: "issue.harness_liveness_escalation_created", - entityType: "issue", - entityId: escalation.id, - details: { - source: "heartbeat.reconcile_issue_graph_liveness", - incidentKey: input.finding.incidentKey, - findingState: input.finding.state, - sourceIssueId: issue.id, - sourceIdentifier: issue.identifier, - recoveryIssueId: recoveryIssue.id, - recoveryIdentifier: recoveryIssue.identifier, - escalationIssueId: escalation.id, - escalationIdentifier: escalation.identifier, - dependencyPath: input.finding.dependencyPath, - ownerSelection: { - selectedAgentId: ownerSelection.agentId, - selectedReason: ownerSelection.reason, - selectedSourceIssueId: ownerSelection.sourceIssueId, - candidateAgentIds: ownerSelection.candidateAgentIds, - candidateReasons: ownerSelection.candidateReasons, - budgetBlockedCandidateAgentIds: ownerSelection.budgetBlockedCandidateAgentIds, - }, - workspaceSelection: { - reuseRecoveryExecutionWorkspace, - inheritedExecutionWorkspaceFromIssueId: reuseRecoveryExecutionWorkspace ? recoveryIssue.id : null, - projectWorkspaceSourceIssueId: recoveryIssue.id, - }, - }, - }); - - const wake = await enqueueWakeup(ownerSelection.agentId, { - source: "assignment", - triggerDetail: "system", - reason: "issue_assigned", - payload: { - issueId: escalation.id, - sourceIssueId: issue.id, - recoveryIssueId: recoveryIssue.id, - incidentKey: input.finding.incidentKey, - }, - requestedByActorType: "system", - requestedByActorId: null, - contextSnapshot: { - issueId: escalation.id, - taskId: escalation.id, - wakeReason: "issue_assigned", - source: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation, - sourceIssueId: issue.id, - recoveryIssueId: recoveryIssue.id, - incidentKey: input.finding.incidentKey, - }, - }); - - logger.warn({ - incidentKey: input.finding.incidentKey, - findingState: input.finding.state, - sourceIssueId: issue.id, - recoveryIssueId: recoveryIssue.id, - escalationIssueId: escalation.id, - ownerAgentId: ownerSelection.agentId, - ownerSelectionReason: ownerSelection.reason, - wakeupRunId: wake?.id ?? null, - }, "created issue graph liveness escalation"); - - return { kind: "created" as const, escalationIssueId: escalation.id }; - } - async function reconcileIssueGraphLiveness(opts?: { runId?: string | null }) { - const findings = await collectIssueGraphLivenessFindings(); - const experimentalSettings = await instanceSettings.getExperimental(); - const autoRecoveryEnabled = experimentalSettings.enableIssueGraphLivenessAutoRecovery === true; - const obsoleteRecoveryCleanup = await retireObsoleteLivenessRecoveryIssues(findings); - const result = { - findings: findings.length, - autoRecoveryEnabled, - escalationsCreated: 0, - existingEscalations: 0, - skipped: 0, - skippedAutoRecoveryDisabled: 0, - skippedAutoRecoveryTooYoung: 0, - obsoleteRecoveriesRetired: obsoleteRecoveryCleanup.retired, - obsoleteRecoveriesActiveSkipped: obsoleteRecoveryCleanup.activeSkipped, - obsoleteRecoveryBlockerRelationsRemoved: obsoleteRecoveryCleanup.blockerRelationsRemoved, - issueIds: [] as string[], - escalationIssueIds: [] as string[], - retiredRecoveryIssueIds: obsoleteRecoveryCleanup.retiredIssueIds, - }; - - if (!autoRecoveryEnabled) { - result.skippedAutoRecoveryDisabled = findings.length; - return result; - } - - const now = new Date(); - for (const finding of findings) { - if (!await isLivenessFindingOldEnoughForAutoRecovery(finding, now)) { - result.skippedAutoRecoveryTooYoung += 1; - result.skipped += 1; - continue; - } - const escalation = await createIssueGraphLivenessEscalation({ - finding, - runId: opts?.runId ?? null, - }); - if (escalation.kind === "created") { - result.escalationsCreated += 1; - result.issueIds.push(finding.issueId); - result.escalationIssueIds.push(escalation.escalationIssueId); - } else if (escalation.kind === "existing") { - result.existingEscalations += 1; - result.issueIds.push(finding.issueId); - result.escalationIssueIds.push(escalation.escalationIssueId); - } else { - result.skipped += 1; - } - } - - return result; + return recovery.reconcileIssueGraphLiveness(opts); } async function updateRuntimeState( diff --git a/server/src/services/recovery/index.ts b/server/src/services/recovery/index.ts index 8c3b6ff2014..521287f161a 100644 --- a/server/src/services/recovery/index.ts +++ b/server/src/services/recovery/index.ts @@ -27,6 +27,9 @@ export type { IssueLivenessSeverity, IssueLivenessState, } from "./issue-graph-liveness.js"; +export { + recoveryService, +} from "./service.js"; export { DEFAULT_MAX_LIVENESS_CONTINUATION_ATTEMPTS, RUN_LIVENESS_CONTINUATION_REASON, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts new file mode 100644 index 00000000000..9f5f61c41b9 --- /dev/null +++ b/server/src/services/recovery/service.ts @@ -0,0 +1,1404 @@ +import { and, asc, desc, eq, inArray, isNull, notInArray, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { + agents, + agentWakeupRequests, + companies, + heartbeatRuns, + issueRelations, + issues, +} from "@paperclipai/db"; +import { parseObject, asBoolean, asNumber } from "../../adapters/utils.js"; +import { logger } from "../../middleware/logger.js"; +import { logActivity } from "../activity-log.js"; +import { budgetService } from "../budgets.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { issueService } from "../issues.js"; +import { + RECOVERY_ORIGIN_KINDS, + buildIssueGraphLivenessLeafKey, + parseIssueGraphLivenessIncidentKey, +} from "./origins.js"; +import { + classifyIssueGraphLiveness, + type IssueLivenessFinding, +} from "./issue-graph-liveness.js"; + +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; +const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueRecovery; +const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; + +type RecoveryWakeupOptions = { + source?: "timer" | "assignment" | "on_demand" | "automation"; + triggerDetail?: "manual" | "ping" | "callback" | "system"; + reason?: string | null; + payload?: Record | null; + idempotencyKey?: string | null; + requestedByActorType?: "user" | "agent" | "system"; + requestedByActorId?: string | null; + contextSnapshot?: Record; +}; + +type RecoveryWakeup = ( + agentId: string, + opts?: RecoveryWakeupOptions, +) => Promise; + +type LatestIssueRun = Pick< + typeof heartbeatRuns.$inferSelect, + "id" | "agentId" | "status" | "error" | "errorCode" | "contextSnapshot" +> | null; + +function readNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +function summarizeRunFailureForIssueComment(run: LatestIssueRun) { + if (!run) return null; + + const errorCode = readNonEmptyString(run.errorCode)?.trim() ?? null; + const rawError = readNonEmptyString(run.error)?.trim() ?? null; + const apiMessageMatch = rawError?.match(/"message"\s*:\s*"([^"]+)"/); + const firstLine = rawError + ?.split(/\r?\n/) + .map((line) => line.trim()) + .find(Boolean) ?? null; + const summarySource = apiMessageMatch?.[1] ?? firstLine; + const summary = + summarySource && summarySource.length > 240 + ? `${summarySource.slice(0, 237)}...` + : summarySource; + + if (errorCode && summary) return ` Latest retry failure: \`${errorCode}\` - ${summary}.`; + if (errorCode) return ` Latest retry failure: \`${errorCode}\`.`; + if (summary) return ` Latest retry failure: ${summary}.`; + return null; +} + +function didAutomaticRecoveryFail( + latestRun: LatestIssueRun, + expectedRetryReason: "assignment_recovery" | "issue_continuation_needed", +) { + if (!latestRun) return false; + + const latestContext = parseObject(latestRun.contextSnapshot); + const latestRetryReason = readNonEmptyString(latestContext.retryReason); + return latestRetryReason === expectedRetryReason && + UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES.includes( + latestRun.status as (typeof UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES)[number], + ); +} + +function issueIdFromRunContext(contextSnapshot: unknown) { + const context = parseObject(contextSnapshot); + return readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId); +} + +function issueIdFromWakePayload(payload: unknown) { + const parsed = parseObject(payload); + const nestedContext = parseObject(parsed[DEFERRED_WAKE_CONTEXT_KEY]); + return readNonEmptyString(parsed.issueId) ?? + readNonEmptyString(nestedContext.issueId) ?? + readNonEmptyString(nestedContext.taskId); +} + +function issueUiLink(issue: { identifier: string | null; id: string }, prefix: string) { + const label = issue.identifier ?? issue.id; + return `[${label}](/${prefix}/issues/${label})`; +} + +function formatIssueLinksForComment(relations: Array<{ identifier?: string | null }>) { + const identifiers = [ + ...new Set( + relations + .map((relation) => relation.identifier) + .filter((identifier): identifier is string => Boolean(identifier)), + ), + ]; + if (identifiers.length === 0) return "another open issue"; + return identifiers + .slice(0, 5) + .map((identifier) => { + const prefix = identifier.split("-")[0] || "PAP"; + return `[${identifier}](/${prefix}/issues/${identifier})`; + }) + .join(", "); +} + +function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) { + return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status)); +} + +function parseLivenessIncidentKey(incidentKey: string | null | undefined) { + if (!incidentKey) return null; + return parseIssueGraphLivenessIncidentKey(incidentKey); +} + +function livenessRecoveryLeafIssueId(finding: IssueLivenessFinding) { + return finding.recoveryIssueId; +} + +function livenessRecoveryLeafFingerprint(finding: IssueLivenessFinding) { + return buildIssueGraphLivenessLeafKey({ + companyId: finding.companyId, + state: finding.state, + leafIssueId: livenessRecoveryLeafIssueId(finding), + }); +} + +function livenessRecoveryLeafKey(companyId: string, state: string, leafIssueId: string) { + return buildIssueGraphLivenessLeafKey({ companyId, state, leafIssueId }); +} + +function isUniqueLivenessRecoveryConflict(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_liveness_recovery_incident_uq" || + maybe.constraint === "issues_active_liveness_recovery_leaf_uq" || + typeof maybe.message === "string" && + ( + maybe.message.includes("issues_active_liveness_recovery_incident_uq") || + maybe.message.includes("issues_active_liveness_recovery_leaf_uq") + ) + ); +} + +function formatDependencyPath(finding: IssueLivenessFinding) { + return finding.dependencyPath + .map((entry) => entry.identifier ?? entry.issueId) + .join(" -> "); +} + +function buildLivenessEscalationDescription(finding: IssueLivenessFinding) { + const source = finding.dependencyPath[0]; + const recovery = finding.dependencyPath.find((entry) => entry.issueId === finding.recoveryIssueId); + const selectedOwner = finding.recommendedOwnerAgentId ?? "none"; + + return [ + "Paperclip detected a harness-level issue graph liveness incident.", + "", + "## Source", + "", + `- Source issue: ${source?.identifier ?? source?.issueId ?? finding.issueId}`, + `- Recovery target issue: ${recovery?.identifier ?? recovery?.issueId ?? finding.recoveryIssueId}`, + `- Incident key: \`${finding.incidentKey}\``, + `- Detected invariant: \`${finding.state}\``, + `- Dependency path: ${formatDependencyPath(finding)}`, + `- Reason: ${finding.reason}`, + "", + "## Ownership", + "", + `- Selected owner agent: \`${selectedOwner}\``, + `- Candidate owner agents: ${finding.recommendedOwnerCandidateAgentIds.length > 0 ? finding.recommendedOwnerCandidateAgentIds.map((id) => `\`${id}\``).join(", ") : "none"}`, + "", + "## Next Action", + "", + finding.recommendedAction, + "", + "Resolve the blocked chain, then mark this escalation issue done so the original issue can resume when all blockers are cleared.", + ].join("\n"); +} + +function buildLivenessOriginalIssueComment(finding: IssueLivenessFinding, escalation: typeof issues.$inferSelect) { + return [ + "Paperclip detected a harness-level liveness incident in this issue's dependency graph.", + "", + `- Escalation issue: ${escalation.identifier ?? escalation.id}`, + `- Incident key: \`${finding.incidentKey}\``, + `- Finding: \`${finding.state}\``, + `- Dependency path: ${formatDependencyPath(finding)}`, + `- Reason: ${finding.reason}`, + `- Manager action requested: ${finding.recommendedAction}`, + "", + "This issue now keeps its existing blockers and is also blocked by the escalation issue so dependency wakeups remain explicit.", + ].join("\n"); +} + +export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) { + const issuesSvc = issueService(db); + const budgets = budgetService(db); + const instanceSettings = instanceSettingsService(db); + + async function getAgent(agentId: string) { + return db.select().from(agents).where(eq(agents.id, agentId)).then((rows) => rows[0] ?? null); + } + + async function getLatestIssueRun(companyId: string, issueId: string): Promise { + return db + .select({ + id: heartbeatRuns.id, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + error: heartbeatRuns.error, + errorCode: heartbeatRuns.errorCode, + contextSnapshot: heartbeatRuns.contextSnapshot, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + ), + ) + .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function hasActiveExecutionPath(companyId: string, issueId: string) { + const [run, deferredWake] = await Promise.all([ + db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + + return Boolean(run || deferredWake); + } + + async function enqueueStrandedIssueRecovery(input: { + issueId: string; + agentId: string; + reason: "issue_assignment_recovery" | "issue_continuation_needed"; + retryReason: "assignment_recovery" | "issue_continuation_needed"; + source: string; + retryOfRunId?: string | null; + }) { + const queued = await deps.enqueueWakeup(input.agentId, { + source: "automation", + triggerDetail: "system", + reason: input.reason, + payload: { + issueId: input.issueId, + ...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}), + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: input.issueId, + taskId: input.issueId, + wakeReason: input.reason, + retryReason: input.retryReason, + source: input.source, + ...(input.retryOfRunId ? { retryOfRunId: input.retryOfRunId } : {}), + }, + }); + + if (queued && input.retryOfRunId) { + return db + .update(heartbeatRuns) + .set({ + retryOfRunId: input.retryOfRunId, + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, queued.id)) + .returning() + .then((rows) => rows[0] ?? queued); + } + + return queued; + } + + async function reconcileUnassignedBlockingIssues() { + const candidates = await db + .select({ + id: issues.id, + companyId: issues.companyId, + identifier: issues.identifier, + status: issues.status, + createdByAgentId: issues.createdByAgentId, + }) + .from(issueRelations) + .innerJoin(issues, eq(issueRelations.issueId, issues.id)) + .where( + and( + eq(issueRelations.type, "blocks"), + inArray(issues.status, ["todo", "blocked"]), + isNull(issues.assigneeAgentId), + isNull(issues.assigneeUserId), + sql`${issues.createdByAgentId} is not null`, + sql`exists ( + select 1 + from issues blocked_issue + where blocked_issue.id = ${issueRelations.relatedIssueId} + and blocked_issue.company_id = ${issues.companyId} + and blocked_issue.status not in ('done', 'cancelled') + )`, + ), + ); + + let assigned = 0; + let skipped = 0; + const issueIds: string[] = []; + const seen = new Set(); + + for (const candidate of candidates) { + if (seen.has(candidate.id)) continue; + seen.add(candidate.id); + + const creatorAgentId = candidate.createdByAgentId; + if (!creatorAgentId) { + skipped += 1; + continue; + } + const creatorAgent = await getAgent(creatorAgentId); + if (!creatorAgent || creatorAgent.companyId !== candidate.companyId || !isAgentInvokable(creatorAgent)) { + skipped += 1; + continue; + } + + const relations = await issuesSvc.getRelationSummaries(candidate.id); + const blockingLinks = formatIssueLinksForComment(relations.blocks); + const updated = await issuesSvc.update(candidate.id, { + assigneeAgentId: creatorAgent.id, + assigneeUserId: null, + }); + if (!updated) { + skipped += 1; + continue; + } + + await issuesSvc.addComment( + candidate.id, + [ + "## Assigned Orphan Blocker", + "", + `Paperclip found this issue is blocking ${blockingLinks} but had no assignee, so no heartbeat could pick it up.`, + "", + "- Assigned it back to the agent that created the blocker.", + "- Next action: resolve this blocker or reassign it to the right owner.", + ].join("\n"), + {}, + ); + + await logActivity(db, { + companyId: candidate.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: null, + action: "issue.updated", + entityType: "issue", + entityId: candidate.id, + details: { + identifier: candidate.identifier, + assigneeAgentId: creatorAgent.id, + source: "recovery.reconcile_unassigned_blocking_issue", + }, + }); + + const queued = await deps.enqueueWakeup(creatorAgent.id, { + source: "automation", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId: candidate.id, + mutation: "unassigned_blocker_recovery", + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: candidate.id, + taskId: candidate.id, + wakeReason: "issue_assigned", + source: "issue.unassigned_blocker_recovery", + }, + }); + + if (queued) { + assigned += 1; + issueIds.push(candidate.id); + } else { + skipped += 1; + } + } + + return { assigned, skipped, issueIds }; + } + + async function getCompanyIssuePrefix(companyId: string) { + return db + .select({ issuePrefix: companies.issuePrefix }) + .from(companies) + .where(eq(companies.id, companyId)) + .then((rows) => rows[0]?.issuePrefix ?? "PAP"); + } + + async function findOpenStrandedIssueRecoveryIssue(companyId: string, sourceIssueId: string) { + return db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, STRANDED_ISSUE_RECOVERY_ORIGIN_KIND), + eq(issues.originId, sourceIssueId), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .orderBy(desc(issues.createdAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function resolveStrandedIssueRecoveryOwnerAgentId(issue: typeof issues.$inferSelect) { + const candidateIds: string[] = []; + if (issue.assigneeAgentId) { + const assignee = await getAgent(issue.assigneeAgentId); + if (assignee?.reportsTo) candidateIds.push(assignee.reportsTo); + } + if (issue.createdByAgentId) { + const creator = await getAgent(issue.createdByAgentId); + if (creator?.reportsTo) candidateIds.push(creator.reportsTo); + candidateIds.push(issue.createdByAgentId); + } + + const roleCandidates = await db + .select() + .from(agents) + .where(and(eq(agents.companyId, issue.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)); + if (issue.assigneeAgentId) candidateIds.push(issue.assigneeAgentId); + + 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 !== issue.companyId) continue; + const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.id, { + issueId: issue.id, + projectId: issue.projectId, + }); + if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id; + } + + return null; + } + + function buildStrandedIssueRecoveryDescription(input: { + issue: typeof issues.$inferSelect; + latestRun: LatestIssueRun; + previousStatus: "todo" | "in_progress"; + prefix: string; + }) { + const sourceIssue = issueUiLink({ identifier: input.issue.identifier, id: input.issue.id }, input.prefix); + const runLink = input.latestRun + ? `[\`${input.latestRun.id}\`](/${input.prefix}/agents/${input.latestRun.agentId}/runs/${input.latestRun.id})` + : "none"; + const retryReason = readNonEmptyString(parseObject(input.latestRun?.contextSnapshot)?.retryReason) ?? "unknown"; + const failureSummary = summarizeRunFailureForIssueComment(input.latestRun); + + return [ + "Paperclip exhausted automatic recovery for an assigned issue and created this explicit recovery task.", + "", + "## Source", + "", + `- Source issue: ${sourceIssue}`, + `- Previous source status: \`${input.previousStatus}\``, + `- Latest retry run: ${runLink}`, + `- Latest retry status: \`${input.latestRun?.status ?? "unknown"}\``, + `- Detected invariant: \`stranded_assigned_issue\``, + `- Retry reason: \`${retryReason}\``, + failureSummary ? `- Failure: ${failureSummary.trim()}` : "- Failure: none recorded", + "", + "## Ownership", + "", + "- Selected owner: the first invokable manager/creator/executive candidate with budget available.", + "", + "## Required Action", + "", + "- Inspect the latest run and source issue state.", + "- Fix the runtime/adapter problem, reassign the source issue, or convert the source issue into a clear manual-review state.", + "- When the source issue has a live execution path or has been intentionally resolved, mark this recovery issue done.", + ].join("\n"); + } + + async function ensureStrandedIssueRecoveryIssue(input: { + issue: typeof issues.$inferSelect; + latestRun: LatestIssueRun; + previousStatus: "todo" | "in_progress"; + }) { + const existing = await findOpenStrandedIssueRecoveryIssue(input.issue.companyId, input.issue.id); + if (existing) return existing; + + const ownerAgentId = await resolveStrandedIssueRecoveryOwnerAgentId(input.issue); + if (!ownerAgentId) return null; + + const prefix = await getCompanyIssuePrefix(input.issue.companyId); + const recovery = await issuesSvc.create(input.issue.companyId, { + title: `Recover stalled issue ${input.issue.identifier ?? input.issue.title}`, + description: buildStrandedIssueRecoveryDescription({ + issue: input.issue, + latestRun: input.latestRun, + previousStatus: input.previousStatus, + prefix, + }), + status: "todo", + priority: input.issue.priority, + parentId: input.issue.id, + projectId: input.issue.projectId, + goalId: input.issue.goalId, + assigneeAgentId: ownerAgentId, + originKind: STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, + originId: input.issue.id, + originRunId: input.latestRun?.id ?? null, + originFingerprint: [ + STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, + input.issue.companyId, + input.issue.id, + input.latestRun?.id ?? "no-run", + ].join(":"), + billingCode: input.issue.billingCode, + inheritExecutionWorkspaceFromIssueId: input.issue.id, + }); + + await deps.enqueueWakeup(ownerAgentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId: recovery.id, + sourceIssueId: input.issue.id, + strandedRunId: input.latestRun?.id ?? null, + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: recovery.id, + taskId: recovery.id, + wakeReason: "issue_assigned", + source: STRANDED_ISSUE_RECOVERY_ORIGIN_KIND, + sourceIssueId: input.issue.id, + strandedRunId: input.latestRun?.id ?? null, + }, + }); + + return recovery; + } + + 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)); + } + + async function escalateStrandedAssignedIssue(input: { + issue: typeof issues.$inferSelect; + previousStatus: "todo" | "in_progress"; + latestRun: LatestIssueRun; + comment: string; + }) { + const recoveryIssue = await ensureStrandedIssueRecoveryIssue({ + issue: input.issue, + previousStatus: input.previousStatus, + latestRun: input.latestRun, + }); + const blockerIds = await existingBlockerIssueIds(input.issue.companyId, input.issue.id); + const nextBlockerIds = recoveryIssue + ? [...new Set([...blockerIds, recoveryIssue.id])] + : blockerIds; + const updated = await issuesSvc.update(input.issue.id, { + status: "blocked", + blockedByIssueIds: nextBlockerIds, + }); + if (!updated) return null; + + const prefix = await getCompanyIssuePrefix(input.issue.companyId); + const recoveryLine = recoveryIssue + ? [ + "", + `- Recovery issue: ${issueUiLink({ identifier: recoveryIssue.identifier, id: recoveryIssue.id }, prefix)}`, + "- Next action: the recovery owner should either restore a live execution path or record the manual resolution, then mark the recovery issue done.", + ].join("\n") + : ""; + + await issuesSvc.addComment(input.issue.id, `${input.comment}${recoveryLine}`, {}); + + await logActivity(db, { + companyId: input.issue.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: null, + action: "issue.updated", + entityType: "issue", + entityId: input.issue.id, + details: { + identifier: input.issue.identifier, + status: "blocked", + previousStatus: input.previousStatus, + source: "recovery.reconcile_stranded_assigned_issue", + latestRunId: input.latestRun?.id ?? null, + latestRunStatus: input.latestRun?.status ?? null, + latestRunErrorCode: input.latestRun?.errorCode ?? null, + recoveryIssueId: recoveryIssue?.id ?? null, + blockerIssueIds: nextBlockerIds, + }, + }); + + return updated; + } + + async function reconcileStrandedAssignedIssues() { + const candidates = await db + .select() + .from(issues) + .where( + and( + isNull(issues.assigneeUserId), + inArray(issues.status, ["todo", "in_progress"]), + sql`${issues.assigneeAgentId} is not null`, + ), + ); + + const result = { + dispatchRequeued: 0, + continuationRequeued: 0, + orphanBlockersAssigned: 0, + escalated: 0, + skipped: 0, + issueIds: [] as string[], + }; + + for (const issue of candidates) { + const agentId = issue.assigneeAgentId; + if (!agentId) { + result.skipped += 1; + continue; + } + + const agent = await getAgent(agentId); + if (!agent || agent.companyId !== issue.companyId || !isAgentInvokable(agent)) { + result.skipped += 1; + continue; + } + + if (await hasActiveExecutionPath(issue.companyId, issue.id)) { + result.skipped += 1; + continue; + } + + const latestRun = await getLatestIssueRun(issue.companyId, issue.id); + if (issue.status === "todo") { + if (!latestRun || latestRun.status === "succeeded") { + result.skipped += 1; + continue; + } + + if (didAutomaticRecoveryFail(latestRun, "assignment_recovery")) { + const failureSummary = summarizeRunFailureForIssueComment(latestRun); + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "todo", + latestRun, + comment: + "Paperclip automatically retried dispatch for this assigned `todo` issue after a lost wake/run, " + + `but it still has no live execution path.${failureSummary ?? ""} ` + + "Moving it to `blocked` so it is visible for intervention.", + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + + const queued = await enqueueStrandedIssueRecovery({ + issueId: issue.id, + agentId, + reason: "issue_assignment_recovery", + retryReason: "assignment_recovery", + source: "issue.assignment_recovery", + retryOfRunId: latestRun.id, + }); + if (queued) { + result.dispatchRequeued += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + + if (!latestRun && !issue.checkoutRunId && !issue.executionRunId) { + result.skipped += 1; + continue; + } + if (didAutomaticRecoveryFail(latestRun, "issue_continuation_needed")) { + const failureSummary = summarizeRunFailureForIssueComment(latestRun); + const updated = await escalateStrandedAssignedIssue({ + issue, + previousStatus: "in_progress", + latestRun, + comment: + "Paperclip automatically retried continuation for this assigned `in_progress` issue after its live " + + `execution disappeared, but it still has no live execution path.${failureSummary ?? ""} ` + + "Moving it to `blocked` so it is visible for intervention.", + }); + if (updated) { + result.escalated += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + continue; + } + + const queued = await enqueueStrandedIssueRecovery({ + issueId: issue.id, + agentId, + reason: "issue_continuation_needed", + retryReason: "issue_continuation_needed", + source: "issue.continuation_recovery", + retryOfRunId: latestRun?.id ?? issue.checkoutRunId ?? null, + }); + if (queued) { + result.continuationRequeued += 1; + result.issueIds.push(issue.id); + } else { + result.skipped += 1; + } + } + + const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues(); + result.orphanBlockersAssigned = orphanBlockerRecovery.assigned; + result.skipped += orphanBlockerRecovery.skipped; + result.issueIds.push(...orphanBlockerRecovery.issueIds); + + return result; + } + + async function collectIssueGraphLivenessFindings() { + const [issueRows, relationRows, agentRows, activeRunRows, activeIssueRunRows, wakeRows] = await Promise.all([ + db + .select({ + id: issues.id, + companyId: issues.companyId, + identifier: issues.identifier, + title: issues.title, + status: issues.status, + projectId: issues.projectId, + goalId: issues.goalId, + parentId: issues.parentId, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + createdByAgentId: issues.createdByAgentId, + createdByUserId: issues.createdByUserId, + executionState: issues.executionState, + }) + .from(issues) + .where( + and( + isNull(issues.hiddenAt), + notInArray(issues.originKind, [RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation]), + ), + ), + db + .select({ + companyId: issueRelations.companyId, + blockerIssueId: issueRelations.issueId, + blockedIssueId: issueRelations.relatedIssueId, + }) + .from(issueRelations) + .where(eq(issueRelations.type, "blocks")), + db + .select({ + id: agents.id, + companyId: agents.companyId, + name: agents.name, + role: agents.role, + title: agents.title, + status: agents.status, + reportsTo: agents.reportsTo, + }) + .from(agents), + db + .select({ + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + contextSnapshot: heartbeatRuns.contextSnapshot, + }) + .from(heartbeatRuns) + .where(inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES])), + db + .select({ + companyId: issues.companyId, + agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, + issueId: issues.id, + }) + .from(issues) + .innerJoin(heartbeatRuns, eq(issues.executionRunId, heartbeatRuns.id)) + .where( + and( + isNull(issues.hiddenAt), + notInArray(issues.originKind, [RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation]), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + ), + ), + db + .select({ + companyId: agentWakeupRequests.companyId, + agentId: agentWakeupRequests.agentId, + status: agentWakeupRequests.status, + payload: agentWakeupRequests.payload, + }) + .from(agentWakeupRequests) + .where(inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"])), + ]); + + return classifyIssueGraphLiveness({ + issues: issueRows, + relations: relationRows, + agents: agentRows, + activeRuns: activeRunRows.map((row) => ({ + companyId: row.companyId, + agentId: row.agentId, + status: row.status, + issueId: issueIdFromRunContext(row.contextSnapshot), + })).concat(activeIssueRunRows.map((row) => ({ + companyId: row.companyId, + agentId: row.agentId, + status: row.status, + issueId: row.issueId, + }))), + queuedWakeRequests: wakeRows.map((row) => ({ + companyId: row.companyId, + agentId: row.agentId, + status: row.status, + issueId: issueIdFromWakePayload(row.payload), + })), + }); + } + + async function findOpenLivenessEscalation(companyId: string, incidentKey: string) { + return db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), + eq(issues.originId, incidentKey), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + } + + async function findOpenLivenessRecoveryIssueForLeaf(finding: IssueLivenessFinding) { + const byFingerprint = await db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, finding.companyId), + eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), + eq(issues.originFingerprint, livenessRecoveryLeafFingerprint(finding)), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + if (byFingerprint) return byFingerprint; + + const leafIssueId = livenessRecoveryLeafIssueId(finding); + const openRecoveries = await db + .select() + .from(issues) + .where( + and( + eq(issues.companyId, finding.companyId), + eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ); + return openRecoveries.find((row) => { + const parsed = parseLivenessIncidentKey(row.originId); + return parsed?.state === finding.state && parsed.leafIssueId === leafIssueId; + }) ?? null; + } + + async function removeRecoveryBlockerFromSource(recovery: typeof issues.$inferSelect) { + const parsed = parseLivenessIncidentKey(recovery.originId); + if (!parsed) return false; + const sourceIssue = await db + .select() + .from(issues) + .where(and(eq(issues.companyId, recovery.companyId), eq(issues.id, parsed.issueId))) + .then((rows) => rows[0] ?? null); + if (!sourceIssue) return false; + + const blockerIds = await existingBlockerIssueIds(sourceIssue.companyId, sourceIssue.id); + if (!blockerIds.includes(recovery.id)) return false; + await issuesSvc.update(sourceIssue.id, { + blockedByIssueIds: blockerIds.filter((blockerId) => blockerId !== recovery.id), + }); + return true; + } + + async function hasActiveRunForIssueId(companyId: string, issueId: string) { + const [contextRun, issueRun] = await Promise.all([ + db + .select({ id: heartbeatRuns.id }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.companyId, companyId), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + sql`(${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId} + OR ${heartbeatRuns.contextSnapshot}->>'taskId' = ${issueId})`, + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + db + .select({ id: heartbeatRuns.id }) + .from(issues) + .innerJoin(heartbeatRuns, eq(issues.executionRunId, heartbeatRuns.id)) + .where( + and( + eq(issues.companyId, companyId), + eq(issues.id, issueId), + inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null), + ]); + return Boolean(contextRun || issueRun); + } + + async function retireObsoleteLivenessRecoveryIssues(findings: IssueLivenessFinding[]) { + const currentIncidentKeys = new Set(findings.map((finding) => finding.incidentKey)); + const currentLeafKeys = new Set( + findings.map((finding) => + livenessRecoveryLeafKey( + finding.companyId, + finding.state, + livenessRecoveryLeafIssueId(finding), + ), + ), + ); + const openRecoveries = await db + .select() + .from(issues) + .where( + and( + eq(issues.originKind, RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation), + isNull(issues.hiddenAt), + notInArray(issues.status, ["done", "cancelled"]), + ), + ); + const result = { + retired: 0, + activeSkipped: 0, + blockerRelationsRemoved: 0, + retiredIssueIds: [] as string[], + }; + + for (const recovery of openRecoveries) { + if (recovery.originId && currentIncidentKeys.has(recovery.originId)) continue; + const parsed = parseLivenessIncidentKey(recovery.originId); + if (!parsed) continue; + if ( + currentLeafKeys.has( + livenessRecoveryLeafKey(parsed.companyId, parsed.state, parsed.leafIssueId), + ) + ) { + continue; + } + if (await removeRecoveryBlockerFromSource(recovery)) { + result.blockerRelationsRemoved += 1; + } + if (await hasActiveRunForIssueId(recovery.companyId, recovery.id)) { + result.activeSkipped += 1; + continue; + } + await issuesSvc.update(recovery.id, { status: "cancelled" }); + result.retired += 1; + result.retiredIssueIds.push(recovery.id); + } + + return result; + } + + async function isLivenessFindingOldEnoughForAutoRecovery(finding: IssueLivenessFinding, now = new Date()) { + const issueIds = [...new Set(finding.dependencyPath.map((entry) => entry.issueId))]; + if (issueIds.length === 0) return false; + const rows = await db + .select({ id: issues.id, updatedAt: issues.updatedAt }) + .from(issues) + .where(and(eq(issues.companyId, finding.companyId), inArray(issues.id, issueIds))); + if (rows.length !== issueIds.length) return false; + const latestUpdatedAt = rows.reduce((latest, row) => + row.updatedAt > latest ? row.updatedAt : latest, + rows[0]!.updatedAt); + return now.getTime() - latestUpdatedAt.getTime() >= ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_MIN_STALE_MS; + } + + async function resolveEscalationOwnerAgentId( + finding: IssueLivenessFinding, + issue: typeof issues.$inferSelect, + ) { + const detailedCandidates = finding.recommendedOwnerCandidates.length > 0 + ? finding.recommendedOwnerCandidates + : finding.recommendedOwnerCandidateAgentIds.map((agentId) => ({ + agentId, + reason: "ordered_invokable_fallback" as const, + sourceIssueId: finding.recoveryIssueId, + })); + const seenCandidates = new Set(); + const candidates = detailedCandidates.filter((candidate) => { + if (seenCandidates.has(candidate.agentId)) return false; + seenCandidates.add(candidate.agentId); + return true; + }); + const budgetBlockedCandidateAgentIds: string[] = []; + + for (const candidate of candidates) { + const budgetBlock = await budgets.getInvocationBlock(issue.companyId, candidate.agentId, { + issueId: issue.id, + projectId: issue.projectId, + }); + if (!budgetBlock) { + return { + agentId: candidate.agentId, + reason: candidate.reason, + sourceIssueId: candidate.sourceIssueId, + candidateAgentIds: candidates.map((entry) => entry.agentId), + candidateReasons: candidates.map((entry) => ({ + agentId: entry.agentId, + reason: entry.reason, + sourceIssueId: entry.sourceIssueId, + })), + budgetBlockedCandidateAgentIds, + }; + } + budgetBlockedCandidateAgentIds.push(candidate.agentId); + } + + return null; + } + + function shouldReuseRecoveryExecutionWorkspace(input: { + finding: IssueLivenessFinding; + recoveryIssue: typeof issues.$inferSelect; + ownerAgentId: string; + }) { + if (input.finding.recoveryIssueId === input.finding.issueId) return false; + return input.recoveryIssue.assigneeAgentId === input.ownerAgentId; + } + + async function ensureIssueBlockedByEscalation(input: { + issue: typeof issues.$inferSelect; + escalationIssueId: string; + finding: IssueLivenessFinding; + runId?: string | null; + }) { + const blockerIds = await existingBlockerIssueIds(input.issue.companyId, input.issue.id); + const nextBlockerIds = [...new Set([...blockerIds, input.escalationIssueId])]; + const update: Partial & { blockedByIssueIds: string[] } = { + blockedByIssueIds: nextBlockerIds, + }; + if (input.issue.status !== "blocked") { + update.status = "blocked"; + } + + const updated = await issuesSvc.update(input.issue.id, update); + if (!updated) return null; + + await logActivity(db, { + companyId: input.issue.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: input.runId ?? null, + action: "issue.blockers.updated", + entityType: "issue", + entityId: input.issue.id, + details: { + source: "recovery.reconcile_issue_graph_liveness", + incidentKey: input.finding.incidentKey, + findingState: input.finding.state, + blockerIssueIds: nextBlockerIds, + escalationIssueId: input.escalationIssueId, + status: update.status ?? input.issue.status, + previousStatus: input.issue.status, + }, + }); + + return updated; + } + + async function createIssueGraphLivenessEscalation(input: { + finding: IssueLivenessFinding; + runId?: string | null; + }) { + const issue = await db + .select() + .from(issues) + .where(eq(issues.id, input.finding.issueId)) + .then((rows) => rows[0] ?? null); + if (!issue || issue.companyId !== input.finding.companyId) return { kind: "skipped" as const }; + + const recoveryIssue = await db + .select() + .from(issues) + .where(and(eq(issues.id, input.finding.recoveryIssueId), eq(issues.companyId, issue.companyId))) + .then((rows) => rows[0] ?? null); + if (!recoveryIssue) return { kind: "skipped" as const }; + + const existing = + await findOpenLivenessEscalation(issue.companyId, input.finding.incidentKey) ?? + await findOpenLivenessRecoveryIssueForLeaf(input.finding); + if (existing) { + await ensureIssueBlockedByEscalation({ + issue, + escalationIssueId: existing.id, + finding: input.finding, + runId: input.runId ?? null, + }); + return { kind: "existing" as const, escalationIssueId: existing.id }; + } + + const ownerSelection = await resolveEscalationOwnerAgentId(input.finding, recoveryIssue); + if (!ownerSelection) return { kind: "skipped" as const }; + const reuseRecoveryExecutionWorkspace = shouldReuseRecoveryExecutionWorkspace({ + finding: input.finding, + recoveryIssue, + ownerAgentId: ownerSelection.agentId, + }); + + let escalation: Awaited>; + try { + escalation = await issuesSvc.create(issue.companyId, { + title: `Unblock liveness incident for ${recoveryIssue.identifier ?? recoveryIssue.title}`, + description: buildLivenessEscalationDescription(input.finding), + status: "todo", + priority: "high", + parentId: recoveryIssue.id, + projectId: recoveryIssue.projectId, + goalId: recoveryIssue.goalId, + assigneeAgentId: ownerSelection.agentId, + originKind: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation, + originId: input.finding.incidentKey, + originFingerprint: livenessRecoveryLeafFingerprint(input.finding), + billingCode: recoveryIssue.billingCode, + ...(reuseRecoveryExecutionWorkspace + ? { inheritExecutionWorkspaceFromIssueId: recoveryIssue.id } + : { + executionWorkspaceId: null, + executionWorkspacePreference: null, + executionWorkspaceSettings: null, + }), + }); + } catch (error) { + if (!isUniqueLivenessRecoveryConflict(error)) throw error; + const raced = + await findOpenLivenessEscalation(issue.companyId, input.finding.incidentKey) ?? + await findOpenLivenessRecoveryIssueForLeaf(input.finding); + if (!raced) throw error; + await ensureIssueBlockedByEscalation({ + issue, + escalationIssueId: raced.id, + finding: input.finding, + runId: input.runId ?? null, + }); + return { kind: "existing" as const, escalationIssueId: raced.id }; + } + + await ensureIssueBlockedByEscalation({ + issue, + escalationIssueId: escalation.id, + finding: input.finding, + runId: input.runId ?? null, + }); + + await issuesSvc.addComment( + issue.id, + buildLivenessOriginalIssueComment(input.finding, escalation), + { runId: input.runId ?? null }, + ); + + await logActivity(db, { + companyId: issue.companyId, + actorType: "system", + actorId: "system", + agentId: ownerSelection.agentId, + runId: input.runId ?? null, + action: "issue.harness_liveness_escalation_created", + entityType: "issue", + entityId: escalation.id, + details: { + source: "recovery.reconcile_issue_graph_liveness", + incidentKey: input.finding.incidentKey, + findingState: input.finding.state, + sourceIssueId: issue.id, + sourceIdentifier: issue.identifier, + recoveryIssueId: recoveryIssue.id, + recoveryIdentifier: recoveryIssue.identifier, + escalationIssueId: escalation.id, + escalationIdentifier: escalation.identifier, + dependencyPath: input.finding.dependencyPath, + ownerSelection: { + selectedAgentId: ownerSelection.agentId, + selectedReason: ownerSelection.reason, + selectedSourceIssueId: ownerSelection.sourceIssueId, + candidateAgentIds: ownerSelection.candidateAgentIds, + candidateReasons: ownerSelection.candidateReasons, + budgetBlockedCandidateAgentIds: ownerSelection.budgetBlockedCandidateAgentIds, + }, + workspaceSelection: { + reuseRecoveryExecutionWorkspace, + inheritedExecutionWorkspaceFromIssueId: reuseRecoveryExecutionWorkspace ? recoveryIssue.id : null, + projectWorkspaceSourceIssueId: recoveryIssue.id, + }, + }, + }); + + const wake = await deps.enqueueWakeup(ownerSelection.agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { + issueId: escalation.id, + sourceIssueId: issue.id, + recoveryIssueId: recoveryIssue.id, + incidentKey: input.finding.incidentKey, + }, + requestedByActorType: "system", + requestedByActorId: null, + contextSnapshot: { + issueId: escalation.id, + taskId: escalation.id, + wakeReason: "issue_assigned", + source: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation, + sourceIssueId: issue.id, + recoveryIssueId: recoveryIssue.id, + incidentKey: input.finding.incidentKey, + }, + }); + + logger.warn({ + incidentKey: input.finding.incidentKey, + findingState: input.finding.state, + sourceIssueId: issue.id, + recoveryIssueId: recoveryIssue.id, + escalationIssueId: escalation.id, + ownerAgentId: ownerSelection.agentId, + ownerSelectionReason: ownerSelection.reason, + wakeupRunId: wake?.id ?? null, + }, "created issue graph liveness escalation"); + + return { kind: "created" as const, escalationIssueId: escalation.id }; + } + + async function reconcileIssueGraphLiveness(opts?: { runId?: string | null }) { + const findings = await collectIssueGraphLivenessFindings(); + const experimentalSettings = await instanceSettings.getExperimental(); + const autoRecoveryEnabled = asBoolean( + experimentalSettings.enableIssueGraphLivenessAutoRecovery, + false, + ); + const obsoleteRecoveryCleanup = await retireObsoleteLivenessRecoveryIssues(findings); + const result = { + findings: findings.length, + autoRecoveryEnabled, + escalationsCreated: 0, + existingEscalations: 0, + skipped: 0, + skippedAutoRecoveryDisabled: 0, + skippedAutoRecoveryTooYoung: 0, + obsoleteRecoveriesRetired: obsoleteRecoveryCleanup.retired, + obsoleteRecoveriesActiveSkipped: obsoleteRecoveryCleanup.activeSkipped, + obsoleteRecoveryBlockerRelationsRemoved: obsoleteRecoveryCleanup.blockerRelationsRemoved, + issueIds: [] as string[], + escalationIssueIds: [] as string[], + retiredRecoveryIssueIds: obsoleteRecoveryCleanup.retiredIssueIds, + }; + + if (!autoRecoveryEnabled) { + result.skippedAutoRecoveryDisabled = findings.length; + return result; + } + + const now = new Date(); + for (const finding of findings) { + if (!await isLivenessFindingOldEnoughForAutoRecovery(finding, now)) { + result.skippedAutoRecoveryTooYoung += 1; + result.skipped += 1; + continue; + } + const escalation = await createIssueGraphLivenessEscalation({ + finding, + runId: opts?.runId ?? null, + }); + if (escalation.kind === "created") { + result.escalationsCreated += 1; + result.issueIds.push(finding.issueId); + result.escalationIssueIds.push(escalation.escalationIssueId); + } else if (escalation.kind === "existing") { + result.existingEscalations += 1; + result.issueIds.push(finding.issueId); + result.escalationIssueIds.push(escalation.escalationIssueId); + } else { + result.skipped += 1; + } + } + + return result; + } + + function readRecoveryTimerIntervalMs(raw: unknown, fallback: number) { + return Math.max(1, Math.floor(asNumber(raw, fallback))); + } + + return { + reconcileStrandedAssignedIssues, + reconcileIssueGraphLiveness, + readRecoveryTimerIntervalMs, + }; +}