From 7bfaaadcb80a2138cf71c5581e08ce8f8bd5ab1c Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 3 Jul 2026 18:30:51 -0700 Subject: [PATCH] Add dependency wake reconciliation backstop (#8943) --- .../heartbeat-dependency-scheduling.test.ts | 116 +++++++ ...eartbeat-issue-liveness-escalation.test.ts | 286 ++++++++++++++++++ server/src/index.ts | 8 +- server/src/routes/issues.ts | 226 +++++++++++--- server/src/services/heartbeat.ts | 59 +++- .../src/services/issue-dependency-wakeups.ts | 72 +++++ server/src/services/recovery/service.ts | 232 ++++++++++++++ ui/src/api/instanceSettings.ts | 11 + 8 files changed, 959 insertions(+), 51 deletions(-) create mode 100644 server/src/services/issue-dependency-wakeups.ts diff --git a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts index ea9987d19c..6bf716d508 100644 --- a/server/src/__tests__/heartbeat-dependency-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-dependency-scheduling.test.ts @@ -401,6 +401,122 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () = expect(noActiveRuns).toBe(true); }); + it("defers issue_blockers_resolved as a follow-up when the same issue is already running", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const blockerId = randomUUID(); + const blockedIssueId = randomUUID(); + const activeRunId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `D${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: { + heartbeat: { + wakeOnDemand: true, + maxConcurrentRuns: 1, + }, + }, + permissions: {}, + }); + await db.insert(heartbeatRuns).values({ + id: activeRunId, + companyId, + agentId, + status: "running", + invocationSource: "on_demand", + contextSnapshot: { + issueId: blockedIssueId, + wakeReason: "manual_test_active_run", + }, + }); + await db.insert(issues).values([ + { + id: blockerId, + companyId, + title: "Completed prerequisite", + status: "done", + priority: "medium", + }, + { + id: blockedIssueId, + companyId, + title: "Blocked dependent", + status: "blocked", + priority: "medium", + assigneeAgentId: agentId, + executionRunId: activeRunId, + executionLockedAt: new Date(), + }, + ]); + await db.insert(issueRelations).values({ + companyId, + issueId: blockerId, + relatedIssueId: blockedIssueId, + type: "blocks", + }); + runningProcesses.set(activeRunId, { + child: {} as import("node:child_process").ChildProcess, + graceSec: 1, + processGroupId: null, + }); + + const idempotencyKey = `issue_blockers_resolved:${blockedIssueId}:${blockerId}`; + const wake = await heartbeat.wakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { + issueId: blockedIssueId, + resolvedBlockerIssueId: blockerId, + }, + idempotencyKey, + contextSnapshot: { + issueId: blockedIssueId, + wakeReason: "issue_blockers_resolved", + resolvedBlockerIssueId: blockerId, + }, + }); + + expect(wake).toBeNull(); + + const wakeRequests = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + idempotencyKey: agentWakeupRequests.idempotencyKey, + runId: agentWakeupRequests.runId, + }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.idempotencyKey, idempotencyKey)); + + expect(wakeRequests).toEqual([ + expect.objectContaining({ + status: "deferred_issue_execution", + reason: "issue_execution_deferred", + idempotencyKey, + runId: null, + }), + ]); + + runningProcesses.delete(activeRunId); + await db + .update(heartbeatRuns) + .set({ status: "succeeded", finishedAt: new Date(), updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, activeRunId)); + }); + it("honors maxConcurrentRuns 1 by leaving a second assignment wake queued", async () => { const companyId = randomUUID(); const agentId = randomUUID(); diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index ecf70fa173..a1352b6e08 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -4,6 +4,7 @@ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest import { activityLog, agents, + agentWakeupRequests, budgetPolicies, companies, costEvents, @@ -16,6 +17,7 @@ import { issues, projects, projectWorkspaces, + workspaceOperations, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, @@ -61,6 +63,7 @@ vi.mock("../adapters/index.ts", async () => { import { heartbeatService } from "../services/heartbeat.ts"; import { instanceSettingsService } from "../services/instance-settings.ts"; +import { issueService } from "../services/issues.ts"; import { runningProcesses } from "../adapters/index.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -206,6 +209,117 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { return { companyId, managerId, coderId, blockedIssueId, blockerIssueId }; } + async function seedResolvedDependencyBackstopFixture(opts: { + workspaceState?: "none" | "not_finalized" | "finalized"; + } = {}) { + const workspaceState = opts.workspaceState ?? "none"; + const companyId = randomUUID(); + const agentId = randomUUID(); + const blockedIssueId = randomUUID(); + const blockerIssueId = randomUUID(); + const projectId = randomUUID(); + const projectWorkspaceId = randomUUID(); + const executionWorkspaceId = randomUUID(); + const issuePrefix = `R${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`; + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Priya", + role: "engineer", + status: "idle", + adapterType: "test_adapter", + adapterConfig: {}, + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } }, + permissions: {}, + }); + + if (workspaceState !== "none") { + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Synthetic dependency project", + status: "in_progress", + }); + await db.insert(projectWorkspaces).values({ + id: projectWorkspaceId, + companyId, + projectId, + name: "Synthetic workspace", + sourceType: "git_worktree", + }); + await db.insert(executionWorkspaces).values({ + id: executionWorkspaceId, + companyId, + projectId, + projectWorkspaceId, + mode: "isolated_workspace", + strategyType: "git_worktree", + name: "Synthetic execution workspace", + providerType: "git_worktree", + }); + } + + await db.insert(issues).values([ + { + id: blockedIssueId, + companyId, + projectId: workspaceState === "none" ? null : projectId, + title: "Synthetic blocked dependent", + status: "blocked", + priority: "medium", + assigneeAgentId: agentId, + issueNumber: 1, + identifier: `${issuePrefix}-1`, + }, + { + id: blockerIssueId, + companyId, + projectId: workspaceState === "none" ? null : projectId, + title: "Synthetic completed blocker", + status: "done", + priority: "medium", + executionWorkspaceId: workspaceState === "none" ? null : executionWorkspaceId, + issueNumber: 2, + identifier: `${issuePrefix}-2`, + }, + ]); + await db.insert(issueRelations).values({ + companyId, + issueId: blockerIssueId, + relatedIssueId: blockedIssueId, + type: "blocks", + }); + + if (workspaceState === "not_finalized") { + await db.insert(workspaceOperations).values({ + companyId, + executionWorkspaceId, + issueId: blockerIssueId, + phase: "adapter_execute", + status: "succeeded", + startedAt: new Date(Date.now() - 60_000), + }); + } else if (workspaceState === "finalized") { + await db.insert(workspaceOperations).values({ + companyId, + executionWorkspaceId, + issueId: blockerIssueId, + phase: "workspace_finalize", + status: "succeeded", + startedAt: new Date(), + }); + } + + return { companyId, agentId, blockedIssueId, blockerIssueId, executionWorkspaceId }; + } + it("keeps liveness findings advisory when auto recovery is disabled", async () => { await instanceSettingsService(db).updateExperimental({ enableIssueGraphLivenessAutoRecovery: false, @@ -227,6 +341,178 @@ describeEmbeddedPostgres("heartbeat issue graph liveness escalation", () => { expect(escalations).toHaveLength(0); }); + it("heals a blocked dependent whose done blocker has no workspace finalize obligation", async () => { + await enableAutoRecovery(); + const { companyId, agentId, blockedIssueId, blockerIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + + const result = await heartbeatService(db).reconcileIssueGraphLiveness(); + + expect(result.findings).toBe(0); + expect(result.dependencyWakesHealed).toBe(1); + expect(result.dependencyWakeIssueIds).toEqual([blockedIssueId]); + expect(result.escalationsCreated).toBe(0); + + const wake = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + idempotencyKey: agentWakeupRequests.idempotencyKey, + }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, agentId)) + .orderBy(agentWakeupRequests.requestedAt) + .then((rows) => rows[0] ?? null); + + expect(wake?.reason).toBe("issue_blockers_resolved"); + expect(wake?.idempotencyKey).toBe(`issue_blockers_resolved:${blockedIssueId}:${blockerIssueId}`); + expect(["queued", "claimed", "completed"]).toContain(wake?.status); + + const events = await db + .select({ action: activityLog.action, entityId: activityLog.entityId, details: activityLog.details }) + .from(activityLog) + .where(and(eq(activityLog.companyId, companyId), eq(activityLog.action, "issue.blockers_resolved_wake_emitted"))); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ entityId: blockedIssueId }); + }); + + it("waits for workspace finalize before healing a resolved blocked dependent", async () => { + await enableAutoRecovery(); + const { companyId, agentId, blockedIssueId, blockerIssueId, executionWorkspaceId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "not_finalized" }); + const heartbeat = heartbeatService(db); + + const beforeFinalize = await heartbeat.reconcileIssueGraphLiveness(); + + expect(beforeFinalize.findings).toBe(0); + expect(beforeFinalize.dependencyWakesHealed).toBe(0); + expect(beforeFinalize.dependencyWakeNotReadySkipped).toBe(1); + + const wakesBeforeFinalize = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, agentId)); + expect(wakesBeforeFinalize).toHaveLength(0); + + await db.insert(workspaceOperations).values({ + companyId, + executionWorkspaceId, + issueId: blockerIssueId, + phase: "workspace_finalize", + status: "succeeded", + startedAt: new Date(), + }); + + const afterFinalize = await heartbeat.reconcileIssueGraphLiveness(); + + expect(afterFinalize.dependencyWakesHealed).toBe(1); + expect(afterFinalize.dependencyWakeIssueIds).toEqual([blockedIssueId]); + + const wake = await db + .select({ + reason: agentWakeupRequests.reason, + idempotencyKey: agentWakeupRequests.idempotencyKey, + }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.agentId, agentId)) + .orderBy(agentWakeupRequests.requestedAt) + .then((rows) => rows[0] ?? null); + expect(wake).toMatchObject({ + reason: "issue_blockers_resolved", + idempotencyKey: `issue_blockers_resolved:${blockedIssueId}:${blockerIssueId}`, + }); + }); + + it("does not duplicate an existing dependency wake keyed to any resolved blocker", async () => { + await enableAutoRecovery(); + const { companyId, agentId, blockedIssueId, blockerIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const secondBlockerIssueId = randomUUID(); + await db.insert(issues).values({ + id: secondBlockerIssueId, + companyId, + title: "Second completed blocker", + status: "done", + priority: "medium", + issueNumber: 3, + identifier: "R-MULTI-3", + }); + await db.insert(issueRelations).values({ + companyId, + issueId: secondBlockerIssueId, + relatedIssueId: blockedIssueId, + type: "blocks", + }); + + const readiness = await issueService(db).getDependencyReadiness(blockedIssueId); + const blockerIdNotUsedByBackstop = readiness.blockerIssueIds.find((id) => id !== blockerIssueId); + if (!blockerIdNotUsedByBackstop) { + throw new Error("Expected a second blocker id in dependency readiness"); + } + expect(blockerIdNotUsedByBackstop).toBe(secondBlockerIssueId); + await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_blockers_resolved", + payload: { + issueId: blockedIssueId, + resolvedBlockerIssueId: blockerIdNotUsedByBackstop, + }, + status: "queued", + idempotencyKey: `issue_blockers_resolved:${blockedIssueId}:${blockerIdNotUsedByBackstop}`, + }); + + const result = await heartbeatService(db).reconcileIssueGraphLiveness(); + + expect(result.dependencyWakesHealed).toBe(0); + expect(result.dependencyWakeExistingSkipped).toBe(1); + + const wakes = await db + .select({ + id: agentWakeupRequests.id, + idempotencyKey: agentWakeupRequests.idempotencyKey, + }) + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "issue_blockers_resolved"))); + expect(wakes).toHaveLength(1); + expect(wakes[0]?.idempotencyKey).toBe( + `issue_blockers_resolved:${blockedIssueId}:${blockerIdNotUsedByBackstop}`, + ); + }); + + it("counts null dependency wake returns as deferred instead of enqueue failures", async () => { + await enableAutoRecovery(); + const { companyId, agentId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + await db + .update(agents) + .set({ + runtimeConfig: { heartbeat: { wakeOnDemand: false, maxConcurrentRuns: 1 } }, + }) + .where(eq(agents.id, agentId)); + + const result = await heartbeatService(db).reconcileIssueGraphLiveness(); + + expect(result.dependencyWakesHealed).toBe(0); + expect(result.dependencyWakeDeferredOrFailed).toBe(1); + expect(result.dependencyWakeEnqueueFailed).toBe(0); + + const skippedWake = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + }) + .from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId))) + .then((rows) => rows[0] ?? null); + expect(skippedWake).toMatchObject({ + status: "skipped", + reason: "heartbeat.wakeOnDemand.disabled", + }); + }); + it("does not create recovery issues outside the configured lookback window", async () => { await enableAutoRecovery(); const { companyId } = await seedBlockedChain({ outsideLookback: true }); diff --git a/server/src/index.ts b/server/src/index.ts index 7254ee5366..88894a204f 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -824,10 +824,10 @@ export async function startServer(): Promise { } const issueGraphReconciled = await heartbeat.reconcileIssueGraphLiveness(); - if (issueGraphReconciled.escalationsCreated > 0) { + if (issueGraphReconciled.escalationsCreated > 0 || issueGraphReconciled.dependencyWakesHealed > 0) { logger.warn( { ...issueGraphReconciled }, - "startup issue-graph liveness reconciliation created escalations", + "startup issue-graph liveness reconciliation changed issue graph state", ); } @@ -928,8 +928,8 @@ export async function startServer(): Promise { }) .then(async () => { const reconciled = await heartbeat.reconcileIssueGraphLiveness(); - if (reconciled.escalationsCreated > 0) { - logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation created escalations"); + if (reconciled.escalationsCreated > 0 || reconciled.dependencyWakesHealed > 0) { + logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation changed issue graph state"); } }) .then(async () => { diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index cf9f37ae79..9dc4324a6b 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -119,6 +119,11 @@ import { SVG_CONTENT_TYPE, } from "../attachment-types.js"; import { queueIssueAssignmentWakeup } from "../services/issue-assignment-wakeup.js"; +import { + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + buildIssueBlockersResolvedWakeIdempotencyKey, + findExistingIssueBlockersResolvedWake, +} from "../services/issue-dependency-wakeups.js"; import { assertEnvironmentSelectionForCompany } from "./environment-selection.js"; import { executionWorkspaceService as executionWorkspaceServiceDirect } from "../services/execution-workspaces.js"; import { feedbackService } from "../services/feedback.js"; @@ -6616,6 +6621,10 @@ export function issueRoutes( // Merge all wakeups from this update into one enqueue per agent to avoid duplicate runs. void (async () => { type WakeupRequest = NonNullable[1]>; + type DependencyReadinessProvider = { + getDependencyReadiness?: typeof svc.getDependencyReadiness; + }; + const dependencyReadinessSvc = svc as DependencyReadinessProvider; const wakeups = new Map(); const addWakeup = (agentId: string, wakeup: WakeupRequest) => { const wakeIssueId = @@ -6624,6 +6633,53 @@ export function issueRoutes( : issue.id; wakeups.set(`${agentId}:${wakeIssueId}`, { agentId, wakeup }); }; + const addDependencyResolvedWakeup = async (input: { + agentId: string; + dependentIssueId: string; + resolvedBlockerIssueId: string; + blockerIssueIds: string[]; + source: string; + mutation: string; + }) => { + const idempotencyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + dependentIssueId: input.dependentIssueId, + resolvedBlockerIssueId: input.resolvedBlockerIssueId, + }); + try { + const existingWake = await findExistingIssueBlockersResolvedWake(db, { + companyId: issue.companyId, + idempotencyKey, + }); + if (existingWake) return; + } catch (err) { + logger.warn( + { err, issueId: input.dependentIssueId, idempotencyKey }, + "failed to check existing dependency wake before issue update wake", + ); + } + addWakeup(input.agentId, { + source: "automation", + triggerDetail: "system", + reason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + payload: { + issueId: input.dependentIssueId, + resolvedBlockerIssueId: input.resolvedBlockerIssueId, + blockerIssueIds: input.blockerIssueIds, + mutation: input.mutation, + }, + idempotencyKey, + requestedByActorType: actor.actorType, + requestedByActorId: actor.actorId, + contextSnapshot: { + issueId: input.dependentIssueId, + taskId: input.dependentIssueId, + wakeReason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + source: input.source, + resolvedBlockerIssueId: input.resolvedBlockerIssueId, + blockerIssueIds: input.blockerIssueIds, + }, + }); + }; if (executionStageWakeup) { addWakeup(executionStageWakeup.agentId, executionStageWakeup.wakeup); @@ -6750,25 +6806,40 @@ export function issueRoutes( if (becameDone) { const dependents = await svc.listWakeableBlockedDependents(issue.id); for (const dependent of dependents) { - addWakeup(dependent.assigneeAgentId, { - source: "automation", - triggerDetail: "system", - reason: "issue_blockers_resolved", - payload: { - issueId: dependent.id, - resolvedBlockerIssueId: issue.id, - blockerIssueIds: dependent.blockerIssueIds, - }, - requestedByActorType: actor.actorType, - requestedByActorId: actor.actorId, - contextSnapshot: { - issueId: dependent.id, - taskId: dependent.id, - wakeReason: "issue_blockers_resolved", - source: "issue.blockers_resolved", - resolvedBlockerIssueId: issue.id, - blockerIssueIds: dependent.blockerIssueIds, - }, + await addDependencyResolvedWakeup({ + agentId: dependent.assigneeAgentId, + dependentIssueId: dependent.id, + resolvedBlockerIssueId: issue.id, + blockerIssueIds: dependent.blockerIssueIds, + source: "issue.blockers_resolved", + mutation: "blocker_done", + }); + } + } + + const restoredBlockedReadyDependency = + issue.status === "blocked" && + issue.assigneeAgentId && + ( + existing.status !== "blocked" || + Array.isArray(req.body.blockedByIssueIds) || + existing.assigneeAgentId !== issue.assigneeAgentId + ); + if (restoredBlockedReadyDependency && typeof dependencyReadinessSvc.getDependencyReadiness === "function") { + const readiness = await dependencyReadinessSvc.getDependencyReadiness(issue.id); + const resolvedBlockerIssueId = readiness.blockerIssueIds[0] ?? null; + if ( + resolvedBlockerIssueId && + readiness.isDependencyReady && + readiness.blockerIssueIds.length > 0 + ) { + await addDependencyResolvedWakeup({ + agentId: issue.assigneeAgentId!, + dependentIssueId: issue.id, + resolvedBlockerIssueId, + blockerIssueIds: readiness.blockerIssueIds, + source: "issue.blockers_restored", + mutation: "blocked_dependency_restored", }); } } @@ -6811,6 +6882,30 @@ export function issueRoutes( for (const { agentId, wakeup } of wakeups.values()) { heartbeat .wakeup(agentId, wakeup) + .then((wakeRun) => { + if (wakeup.reason !== ISSUE_BLOCKERS_RESOLVED_WAKE_REASON) return; + const payload = wakeup.payload && typeof wakeup.payload === "object" ? wakeup.payload : {}; + const dependentIssueId = typeof payload.issueId === "string" ? payload.issueId : issue.id; + return logActivity(db, { + companyId: issue.companyId, + actorType: "system", + actorId: "issue_update", + agentId, + runId: actor.runId, + action: "issue.blockers_resolved_wake_emitted", + entityType: "issue", + entityId: dependentIssueId, + details: { + source: wakeup.contextSnapshot?.source ?? "issue.update", + wakeupRunId: wakeRun?.id ?? null, + idempotencyKey: wakeup.idempotencyKey ?? null, + resolvedBlockerIssueId: typeof payload.resolvedBlockerIssueId === "string" + ? payload.resolvedBlockerIssueId + : null, + blockerIssueIds: Array.isArray(payload.blockerIssueIds) ? payload.blockerIssueIds : [], + }, + }); + }) .catch((err) => logger.warn({ err, issueId: issue.id, agentId }, "failed to wake agent on issue update")); } })(); @@ -8049,6 +8144,51 @@ export function issueRoutes( if (wakeups.has(key)) return; wakeups.set(key, { agentId, wakeup }); }; + const addDependencyResolvedWakeup = async (input: { + agentId: string; + dependentIssueId: string; + resolvedBlockerIssueId: string; + blockerIssueIds: string[]; + }) => { + const idempotencyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + dependentIssueId: input.dependentIssueId, + resolvedBlockerIssueId: input.resolvedBlockerIssueId, + }); + try { + const existingWake = await findExistingIssueBlockersResolvedWake(db, { + companyId: currentIssue.companyId, + idempotencyKey, + }); + if (existingWake) return; + } catch (err) { + logger.warn( + { err, issueId: input.dependentIssueId, idempotencyKey }, + "failed to check existing dependency wake before issue comment wake", + ); + } + addWakeup(input.agentId, { + source: "automation", + triggerDetail: "system", + reason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + payload: { + issueId: input.dependentIssueId, + resolvedBlockerIssueId: input.resolvedBlockerIssueId, + blockerIssueIds: input.blockerIssueIds, + mutation: "comment", + }, + idempotencyKey, + requestedByActorType: actor.actorType, + requestedByActorId: actor.actorId, + contextSnapshot: { + issueId: input.dependentIssueId, + taskId: input.dependentIssueId, + wakeReason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + source: "issue.blockers_resolved", + resolvedBlockerIssueId: input.resolvedBlockerIssueId, + blockerIssueIds: input.blockerIssueIds, + }, + }); + }; if (commentDecisionStageWakeup) { addWakeup(commentDecisionStageWakeup.agentId, commentDecisionStageWakeup.wakeup); @@ -8148,25 +8288,11 @@ export function issueRoutes( if (becameDone) { const dependents = await svc.listWakeableBlockedDependents(currentIssue.id); for (const dependent of dependents) { - addWakeup(dependent.assigneeAgentId, { - source: "automation", - triggerDetail: "system", - reason: "issue_blockers_resolved", - payload: { - issueId: dependent.id, - resolvedBlockerIssueId: currentIssue.id, - blockerIssueIds: dependent.blockerIssueIds, - }, - requestedByActorType: actor.actorType, - requestedByActorId: actor.actorId, - contextSnapshot: { - issueId: dependent.id, - taskId: dependent.id, - wakeReason: "issue_blockers_resolved", - source: "issue.blockers_resolved", - resolvedBlockerIssueId: currentIssue.id, - blockerIssueIds: dependent.blockerIssueIds, - }, + await addDependencyResolvedWakeup({ + agentId: dependent.assigneeAgentId, + dependentIssueId: dependent.id, + resolvedBlockerIssueId: currentIssue.id, + blockerIssueIds: dependent.blockerIssueIds, }); } } @@ -8210,6 +8336,30 @@ export function issueRoutes( for (const { agentId, wakeup } of wakeups.values()) { heartbeat .wakeup(agentId, wakeup) + .then((wakeRun) => { + if (wakeup.reason !== ISSUE_BLOCKERS_RESOLVED_WAKE_REASON) return; + const payload = wakeup.payload && typeof wakeup.payload === "object" ? wakeup.payload : {}; + const dependentIssueId = typeof payload.issueId === "string" ? payload.issueId : currentIssue.id; + return logActivity(db, { + companyId: currentIssue.companyId, + actorType: "system", + actorId: "issue_comment", + agentId, + runId: actor.runId, + action: "issue.blockers_resolved_wake_emitted", + entityType: "issue", + entityId: dependentIssueId, + details: { + source: wakeup.contextSnapshot?.source ?? "issue.comment", + wakeupRunId: wakeRun?.id ?? null, + idempotencyKey: wakeup.idempotencyKey ?? null, + resolvedBlockerIssueId: typeof payload.resolvedBlockerIssueId === "string" + ? payload.resolvedBlockerIssueId + : null, + blockerIssueIds: Array.isArray(payload.blockerIssueIds) ? payload.blockerIssueIds : [], + }, + }); + }) .catch((err) => logger.warn({ err, issueId: currentIssue.id, agentId }, "failed to wake agent on issue comment")); } })(); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 00f9f3ea26..856b01129f 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -109,6 +109,11 @@ import { sanitizeRuntimeServiceBaseEnv, } from "./workspace-runtime.js"; import { issueService } from "./issues.js"; +import { + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + buildIssueBlockersResolvedWakeIdempotencyKey, + findExistingIssueBlockersResolvedWake, +} from "./issue-dependency-wakeups.js"; import { buildIssueMonitorClearedPatch, buildIssueMonitorTriggeredPatch, @@ -417,7 +422,10 @@ function mergeAdapterRecoveryMetadata(input: { : {}), }; } -const RUNNING_ISSUE_WAKE_REASONS_REQUIRING_FOLLOWUP = new Set(["approval_approved"]); +const RUNNING_ISSUE_WAKE_REASONS_REQUIRING_FOLLOWUP = new Set([ + "approval_approved", + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, +]); const SESSIONED_LOCAL_ADAPTERS = new Set([ "claude_local", "codex_local", @@ -11439,11 +11447,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) agent, ); - // Workspace-finalize wake re-fire: if this run's issue was marked done - // mid-run (so the original `issue_blockers_resolved` wake was gated by - // the readiness check waiting for workspace_finalize), the finalize - // row we just recorded now lets dependents proceed. Fire wakes here. - if (issueId && adapterFinalizeOutcome === "succeeded") { + // Dependency wake re-check: if this run's issue was marked done mid-run, + // the route-time `issue_blockers_resolved` wake may have been gated by + // workspace finalization or merged into this run. Re-evaluate after any + // run completion, including failed adapter outcomes; this is safe because + // `listWakeableBlockedDependents` delegates to dependency readiness, which + // only returns dependents whose done blockers have crossed the successful + // `workspace_finalize` barrier. + if (issueId && finalizedRun) { try { const blockerIssueStatus = await db .select({ status: issues.status }) @@ -11453,25 +11464,55 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (blockerIssueStatus === "done") { const dependents = await issuesSvc.listWakeableBlockedDependents(issueId); for (const dependent of dependents) { + const idempotencyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + dependentIssueId: dependent.id, + resolvedBlockerIssueId: issueId, + }); + const existingWake = await findExistingIssueBlockersResolvedWake(db, { + companyId: finalizedRun.companyId, + idempotencyKey, + }); + if (existingWake) continue; await enqueueWakeup(dependent.assigneeAgentId, { source: "automation", triggerDetail: "system", - reason: "issue_blockers_resolved", + reason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, payload: { issueId: dependent.id, resolvedBlockerIssueId: issueId, blockerIssueIds: dependent.blockerIssueIds, deferredFor: "workspace_finalize", }, + idempotencyKey, + requestedByActorType: "system", + requestedByActorId: "heartbeat_finalize", contextSnapshot: { issueId: dependent.id, taskId: dependent.id, - wakeReason: "issue_blockers_resolved", + wakeReason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, source: "workspace.finalize", resolvedBlockerIssueId: issueId, blockerIssueIds: dependent.blockerIssueIds, }, - }).catch((wakeErr) => { + }).then((wakeRun) => + logActivity(db, { + companyId: finalizedRun.companyId, + actorType: "system", + actorId: "heartbeat_finalize", + agentId: dependent.assigneeAgentId, + runId: finalizedRun.id, + action: "issue.blockers_resolved_wake_emitted", + entityType: "issue", + entityId: dependent.id, + details: { + source: "workspace.finalize", + wakeupRunId: wakeRun?.id ?? null, + idempotencyKey, + resolvedBlockerIssueId: issueId, + blockerIssueIds: dependent.blockerIssueIds, + }, + }) + ).catch((wakeErr) => { logger.warn( { err: wakeErr, issueId, dependentIssueId: dependent.id, agentId: dependent.assigneeAgentId }, "failed to fire deferred dependent wake after workspace_finalize", diff --git a/server/src/services/issue-dependency-wakeups.ts b/server/src/services/issue-dependency-wakeups.ts new file mode 100644 index 0000000000..019a9a9cd5 --- /dev/null +++ b/server/src/services/issue-dependency-wakeups.ts @@ -0,0 +1,72 @@ +import { and, eq, inArray } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { agentWakeupRequests } from "@paperclipai/db"; + +export const ISSUE_BLOCKERS_RESOLVED_WAKE_REASON = "issue_blockers_resolved"; + +const IDEMPOTENT_DEPENDENCY_WAKE_STATUSES = [ + "queued", + "deferred_issue_execution", + "claimed", + "completed", +] as const; + +export function buildIssueBlockersResolvedWakeIdempotencyKey(input: { + dependentIssueId: string; + resolvedBlockerIssueId: string; +}) { + return [ + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + input.dependentIssueId, + input.resolvedBlockerIssueId, + ].join(":"); +} + +export async function findExistingIssueBlockersResolvedWake( + db: Db, + input: { + companyId: string; + idempotencyKey: string; + }, +) { + return db + .select({ id: agentWakeupRequests.id, status: agentWakeupRequests.status }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, input.companyId), + eq(agentWakeupRequests.idempotencyKey, input.idempotencyKey), + inArray(agentWakeupRequests.status, [...IDEMPOTENT_DEPENDENCY_WAKE_STATUSES]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); +} + +export async function findExistingIssueBlockersResolvedWakeForAnyKey( + db: Db, + input: { + companyId: string; + idempotencyKeys: string[]; + }, +) { + const idempotencyKeys = [...new Set(input.idempotencyKeys.filter(Boolean))]; + if (idempotencyKeys.length === 0) return null; + + return db + .select({ + id: agentWakeupRequests.id, + status: agentWakeupRequests.status, + idempotencyKey: agentWakeupRequests.idempotencyKey, + }) + .from(agentWakeupRequests) + .where( + and( + eq(agentWakeupRequests.companyId, input.companyId), + inArray(agentWakeupRequests.idempotencyKey, idempotencyKeys), + inArray(agentWakeupRequests.status, [...IDEMPOTENT_DEPENDENCY_WAKE_STATUSES]), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); +} diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 607c8e8f7c..24c32899a6 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -37,6 +37,11 @@ import { instanceSettingsService } from "../instance-settings.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueTreeControlService } from "../issue-tree-control.js"; import { TERMINAL_HEARTBEAT_RUN_STATUSES, issueService } from "../issues.js"; +import { + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + buildIssueBlockersResolvedWakeIdempotencyKey, + findExistingIssueBlockersResolvedWakeForAnyKey, +} from "../issue-dependency-wakeups.js"; import { parseIssueExecutionState } from "../issue-execution-policy.js"; import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; import { getRunLogStore } from "../run-log-store.js"; @@ -74,6 +79,7 @@ const STRANDED_ISSUE_RECOVERY_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.strandedIssueR const STALE_ACTIVE_RUN_EVALUATION_ORIGIN_KIND = RECOVERY_ORIGIN_KINDS.staleActiveRunEvaluation; const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; const EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON = "execution_review_participant_recovery"; +const RESOLVED_DEPENDENCY_WAKE_BACKSTOP_CANDIDATE_LIMIT = 500; const SESSIONED_LOCAL_ADAPTERS = new Set([ "claude_local", "codex_local", @@ -516,6 +522,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) const budgets = budgetService(db); const instanceSettings = instanceSettingsService(db); const runLogStore = getRunLogStore(); + let resolvedDependencyWakeBackstopCandidateCursor: string | null = null; const getCurrentUserRedactionOptions = async () => ({ enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs, @@ -4066,6 +4073,205 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return { kind: "created" as const, escalationIssueId: escalation.id }; } + async function reconcileResolvedDependencyWakeBackstop(opts?: { runId?: string | null }) { + const result = { + checked: 0, + healed: 0, + existingWakeSkipped: 0, + livePathSkipped: 0, + interactionSkipped: 0, + pauseHoldSkipped: 0, + notReadySkipped: 0, + candidateLimitSkipped: 0, + deferredOrFailed: 0, + enqueueFailed: 0, + issueIds: [] as string[], + }; + + const queryCandidates = (afterIssueId: string | null) => { + const filters = [ + eq(issues.status, "blocked"), + isNull(issues.hiddenAt), + sql`${issues.assigneeAgentId} is not null`, + ]; + if (afterIssueId) filters.push(gt(issues.id, afterIssueId)); + + return db + .select({ + id: issues.id, + companyId: issues.companyId, + identifier: issues.identifier, + assigneeAgentId: issues.assigneeAgentId, + totalCount: sql`count(*) over()::int`, + }) + .from(issues) + .where(and(...filters)) + .orderBy(asc(issues.id)) + .limit(RESOLVED_DEPENDENCY_WAKE_BACKSTOP_CANDIDATE_LIMIT); + }; + + let candidateRows = await queryCandidates(resolvedDependencyWakeBackstopCandidateCursor); + if (candidateRows.length === 0 && resolvedDependencyWakeBackstopCandidateCursor) { + resolvedDependencyWakeBackstopCandidateCursor = null; + candidateRows = await queryCandidates(null); + } + const totalCandidateCount = candidateRows[0]?.totalCount ?? 0; + const candidates = candidateRows.map(({ totalCount: _totalCount, ...candidate }) => candidate); + result.checked = candidates.length; + result.candidateLimitSkipped = Math.max(0, totalCandidateCount - candidates.length); + const lastCandidate = candidates[candidates.length - 1] ?? null; + resolvedDependencyWakeBackstopCandidateCursor = + result.candidateLimitSkipped > 0 && lastCandidate ? lastCandidate.id : null; + if (result.candidateLimitSkipped > 0) { + logger.warn( + { + processed: candidates.length, + skipped: result.candidateLimitSkipped, + limit: RESOLVED_DEPENDENCY_WAKE_BACKSTOP_CANDIDATE_LIMIT, + nextCursor: resolvedDependencyWakeBackstopCandidateCursor, + }, + "issue graph liveness backstop deferred resolved dependency wake candidates past page limit", + ); + } + + const candidatesByCompany = new Map(); + for (const candidate of candidates) { + const companyCandidates = candidatesByCompany.get(candidate.companyId) ?? []; + companyCandidates.push(candidate); + candidatesByCompany.set(candidate.companyId, companyCandidates); + } + + for (const [companyId, companyCandidates] of candidatesByCompany.entries()) { + const readinessMap = await issuesSvc.listDependencyReadiness( + companyId, + companyCandidates.map((candidate) => candidate.id), + ); + + for (const candidate of companyCandidates) { + const agentId = candidate.assigneeAgentId; + if (!agentId) continue; + + const readiness = readinessMap.get(candidate.id); + const resolvedBlockerIssueId = readiness?.blockerIssueIds[0] ?? null; + if ( + !readiness || + !readiness.isDependencyReady || + readiness.blockerIssueIds.length === 0 || + !resolvedBlockerIssueId + ) { + result.notReadySkipped += 1; + continue; + } + + const idempotencyKeys = readiness.blockerIssueIds.map((blockerIssueId) => + buildIssueBlockersResolvedWakeIdempotencyKey({ + dependentIssueId: candidate.id, + resolvedBlockerIssueId: blockerIssueId, + }) + ); + const idempotencyKey = buildIssueBlockersResolvedWakeIdempotencyKey({ + dependentIssueId: candidate.id, + resolvedBlockerIssueId, + }); + const existingWake = await findExistingIssueBlockersResolvedWakeForAnyKey(db, { + companyId, + idempotencyKeys, + }); + if (existingWake) { + result.existingWakeSkipped += 1; + continue; + } + + if ( + await hasActiveExecutionPath(companyId, candidate.id, agentId) || + await hasQueuedIssueWake(companyId, candidate.id, agentId) + ) { + result.livePathSkipped += 1; + continue; + } + + if (await hasPendingWakeInteraction(companyId, candidate.id)) { + result.interactionSkipped += 1; + continue; + } + + if (await isAutomaticRecoverySuppressedByPauseHold(db, companyId, candidate.id, treeControlSvc)) { + result.pauseHoldSkipped += 1; + continue; + } + + try { + const wake = await deps.enqueueWakeup(agentId, { + source: "automation", + triggerDetail: "system", + reason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + payload: { + issueId: candidate.id, + resolvedBlockerIssueId, + blockerIssueIds: readiness.blockerIssueIds, + backstop: "issue_graph_liveness_reconciliation", + }, + idempotencyKey, + requestedByActorType: "system", + requestedByActorId: "issue_graph_liveness_backstop", + contextSnapshot: { + issueId: candidate.id, + taskId: candidate.id, + wakeReason: ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + source: "issue_graph_liveness.backstop", + resolvedBlockerIssueId, + blockerIssueIds: readiness.blockerIssueIds, + }, + }); + if (!wake) { + // enqueueWakeup returns null for normal deferred/skipped paths + // such as disabled wake-on-demand or concurrency gating. That is + // not an enqueue error, but the backstop still did not heal now. + result.deferredOrFailed += 1; + continue; + } + + result.healed += 1; + result.issueIds.push(candidate.id); + + await logActivity(db, { + companyId, + actorType: "system", + actorId: "issue_graph_liveness_backstop", + agentId, + runId: opts?.runId ?? null, + action: "issue.blockers_resolved_wake_emitted", + entityType: "issue", + entityId: candidate.id, + details: { + source: "issue_graph_liveness.backstop", + wakeupRunId: wake.id, + idempotencyKey, + resolvedBlockerIssueId, + blockerIssueIds: readiness.blockerIssueIds, + }, + }); + } catch (err) { + result.deferredOrFailed += 1; + result.enqueueFailed += 1; + logger.warn( + { err, issueId: candidate.id, agentId, idempotencyKey }, + "failed to enqueue dependency wake from issue graph liveness backstop", + ); + } + } + } + + if (result.healed > 0) { + logger.warn( + { healed: result.healed, issueIds: result.issueIds }, + "issue graph liveness backstop healed resolved blocked dependency wakes", + ); + } + + return result; + } + async function reconcileIssueGraphLiveness(opts?: { runId?: string | null; force?: boolean; @@ -4099,6 +4305,17 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) obsoleteRecoveriesActiveSkipped: obsoleteRecoveryCleanup.activeSkipped, obsoleteRecoveryBlockerRelationsRemoved: obsoleteRecoveryCleanup.blockerRelationsRemoved, doneRecoveryBlockerRelationsRemoved: doneRecoveryBlockerCleanup.blockerRelationsRemoved, + dependencyWakeBackstopChecked: 0, + dependencyWakesHealed: 0, + dependencyWakeExistingSkipped: 0, + dependencyWakeLivePathSkipped: 0, + dependencyWakeInteractionSkipped: 0, + dependencyWakePauseHoldSkipped: 0, + dependencyWakeNotReadySkipped: 0, + dependencyWakeCandidateLimitSkipped: 0, + dependencyWakeDeferredOrFailed: 0, + dependencyWakeEnqueueFailed: 0, + dependencyWakeIssueIds: [] as string[], issueIds: [] as string[], escalationIssueIds: [] as string[], retiredRecoveryIssueIds: obsoleteRecoveryCleanup.retiredIssueIds, @@ -4109,6 +4326,21 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return result; } + const dependencyWakeBackstop = await reconcileResolvedDependencyWakeBackstop({ + runId: opts?.runId ?? null, + }); + result.dependencyWakeBackstopChecked = dependencyWakeBackstop.checked; + result.dependencyWakesHealed = dependencyWakeBackstop.healed; + result.dependencyWakeExistingSkipped = dependencyWakeBackstop.existingWakeSkipped; + result.dependencyWakeLivePathSkipped = dependencyWakeBackstop.livePathSkipped; + result.dependencyWakeInteractionSkipped = dependencyWakeBackstop.interactionSkipped; + result.dependencyWakePauseHoldSkipped = dependencyWakeBackstop.pauseHoldSkipped; + result.dependencyWakeNotReadySkipped = dependencyWakeBackstop.notReadySkipped; + result.dependencyWakeCandidateLimitSkipped = dependencyWakeBackstop.candidateLimitSkipped; + result.dependencyWakeDeferredOrFailed = dependencyWakeBackstop.deferredOrFailed; + result.dependencyWakeEnqueueFailed = dependencyWakeBackstop.enqueueFailed; + result.dependencyWakeIssueIds = dependencyWakeBackstop.issueIds; + for (const finding of findings) { if (!isLivenessFindingInsideAutoRecoveryLookback(finding, cutoff, updatedAtByIssueKey)) { result.skippedOutsideLookback += 1; diff --git a/ui/src/api/instanceSettings.ts b/ui/src/api/instanceSettings.ts index ca8b7841b5..8ae0b80a4d 100644 --- a/ui/src/api/instanceSettings.ts +++ b/ui/src/api/instanceSettings.ts @@ -38,6 +38,17 @@ export const instanceSettingsApi = { skipped: number; skippedAutoRecoveryDisabled: number; skippedOutsideLookback: number; + dependencyWakeBackstopChecked: number; + dependencyWakesHealed: number; + dependencyWakeExistingSkipped: number; + dependencyWakeLivePathSkipped: number; + dependencyWakeInteractionSkipped: number; + dependencyWakePauseHoldSkipped: number; + dependencyWakeNotReadySkipped: number; + dependencyWakeCandidateLimitSkipped: number; + dependencyWakeDeferredOrFailed: number; + dependencyWakeEnqueueFailed: number; + dependencyWakeIssueIds: string[]; escalationIssueIds: string[]; }>( "/instance/settings/experimental/issue-graph-liveness-auto-recovery/run",