RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add dependency wake reconciliation backstop (#8943)

This commit is contained in:
Nicky Leach 2026-07-03 18:30:51 -07:00 committed by GitHub
parent bcac517f3b
commit 7bfaaadcb8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 959 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
server/src/__tests__/heartbeat-dependency-scheduling.test.ts
View File

@ -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();

286
server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts
View File

@ -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 });

8
server/src/index.ts
View File

@ -824,10 +824,10 @@ export async function startServer(): Promise<StartedServer> {
}
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<StartedServer> {
})
.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 () => {

226
server/src/routes/issues.ts
View File

@ -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<Parameters<typeof heartbeat.wakeup>[1]>;
type DependencyReadinessProvider = {
getDependencyReadiness?: typeof svc.getDependencyReadiness;
};
const dependencyReadinessSvc = svc as DependencyReadinessProvider;
const wakeups = new Map<string, { agentId: string; wakeup: WakeupRequest }>();
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"));
}
})();

59
server/src/services/heartbeat.ts
View File

@ -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",

72
server/src/services/issue-dependency-wakeups.ts Normal file
View File

@ -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);
}

232
server/src/services/recovery/service.ts
View File

@ -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<number>`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<string, typeof candidates>();
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;

11
ui/src/api/instanceSettings.ts
View File

@ -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",