diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 326a85c1687..8916c01d62e 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -842,6 +842,8 @@ Use it for: Blocked issues should stay idle while blockers remain unresolved. Paperclip should not create a queued heartbeat run for that issue until the final blocker is done and the `issue_blockers_resolved` wake can start real work. +Once the last blocker is resolved, the dependency sweep wakes the assignee once for that blocked cycle, whether the blocker cleared by reaching `done` or by settling its workspace finalize (including a "Recover saved files" finalize that has no workspace). The sweep normally skips an issue that its assignee agent blocked itself while its blockers were already resolved, because that agent is waiting on something outside the blocker graph. A native completion that Paperclip refused because dependencies were unresolved (`durable_dependency_blocker_bound`) is not such a wait: the dependency owns its wake, so the sweep wakes it once readiness holds, including for issues already sitting `blocked` when a release that changes readiness is deployed. + `cancelled` is terminal for the blocker issue itself, but it does not satisfy the dependency. A cancelled blocker edge remains unresolved until the edge is removed or replaced, and Paperclip must surface blocker attention on the dependent regardless of whether that dependent is currently displayed as `blocked`, `todo`, `backlog`, or another non-terminal agent-owned status. If a parent is truly waiting on a child, model that with blockers. Do not rely on the parent/child relationship alone. diff --git a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts index d620a21a275..5c36dac61ce 100644 --- a/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts +++ b/server/src/__tests__/heartbeat-issue-liveness-escalation.test.ts @@ -1283,6 +1283,69 @@ describeEmbeddedPostgres("heartbeat resolved dependency wake reconciliation", () expect(result.selfBlockedSkipped).toBe(0); }); + it("wakes once when a completion refused for unresolved dependencies later becomes valid", async () => { + const { companyId, agentId, blockedIssueId, blockerIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date(Date.now() - 120_000); + await db.update(issues).set({ blockedTransitionAt }).where(eq(issues.id, blockedIssueId)); + await db.update(issues).set({ completedAt: new Date(blockedTransitionAt.getTime() - 60_000) }).where(eq(issues.id, blockerIssueId)); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, issueId: blockedIssueId, status: "succeeded", + runtimeMode: "native", contextSnapshot: { issueId: blockedIssueId } }).returning(); + await db.insert(activityLog).values({ companyId, actorType: "system", actorId: "native-status-committer", + runId: run.id, action: "issue.updated", entityType: "issue", entityId: blockedIssueId, + details: { source: "native_status_decision", fromStatus: "in_progress", toStatus: "blocked", + reasonCode: "durable_dependency_blocker_bound" }, createdAt: blockedTransitionAt }); + + const first = await heartbeatService(db).reconcileResolvedDependencyWakes({ companyId, issueId: blockedIssueId }); + expect(first.selfBlockedSkipped).toBe(0); + expect(first.healed).toBe(1); + const second = await heartbeatService(db).reconcileResolvedDependencyWakes({ companyId, issueId: blockedIssueId }); + expect(second.healed).toBe(0); + const wakes = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), + eq(agentWakeupRequests.reason, "issue_blockers_resolved"))); + expect(wakes).toHaveLength(1); + }); + + it("still holds a native block the agent chose while its dependencies were ready", async () => { + const { companyId, agentId, blockedIssueId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "none" }); + const blockedTransitionAt = new Date(Date.now() - 120_000); + await db.update(issues).set({ blockedTransitionAt }).where(eq(issues.id, blockedIssueId)); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, issueId: blockedIssueId, status: "succeeded", + runtimeMode: "native", contextSnapshot: { issueId: blockedIssueId } }).returning(); + await db.insert(activityLog).values({ companyId, actorType: "system", actorId: "native-status-committer", + runId: run.id, action: "issue.updated", entityType: "issue", entityId: blockedIssueId, + details: { source: "native_status_decision", fromStatus: "in_progress", toStatus: "blocked", + reasonCode: "task_wide_blocker_bound" }, createdAt: blockedTransitionAt }); + const result = await heartbeatService(db).reconcileResolvedDependencyWakes({ companyId, issueId: blockedIssueId }); + expect(result.healed).toBe(0); + expect(result.selfBlockedSkipped).toBe(1); + }); + + it("heals a self-block once a later recovery finalize without a workspace saves the blocker", async () => { + const { companyId, agentId, blockedIssueId, blockerIssueId, executionWorkspaceId } = + await seedResolvedDependencyBackstopFixture({ workspaceState: "finalized" }); + await db.delete(workspaceOperations).where(eq(workspaceOperations.issueId, blockerIssueId)); + const blockedTransitionAt = new Date(Date.now() - 120_000); + await db.update(issues).set({ blockedTransitionAt }).where(eq(issues.id, blockedIssueId)); + const [blockerRun] = await db.insert(heartbeatRuns).values({ companyId, agentId, issueId: blockerIssueId, + status: "succeeded", contextSnapshot: { issueId: blockerIssueId } }).returning(); + await db.insert(workspaceOperations).values({ companyId, executionWorkspaceId, issueId: blockerIssueId, + heartbeatRunId: blockerRun.id, phase: "workspace_finalize", status: "failed", + startedAt: new Date(blockedTransitionAt.getTime() - 60_000), finishedAt: new Date(blockedTransitionAt.getTime() - 59_000) }); + await db.insert(activityLog).values({ companyId, actorType: "agent", actorId: agentId, agentId, + action: "issue.updated", entityType: "issue", entityId: blockedIssueId, details: { status: "blocked" }, createdAt: blockedTransitionAt }); + await expect(issueService(db).getDependencyReadiness(blockedIssueId)).resolves.toMatchObject({ isDependencyReady: false }); + await db.insert(workspaceOperations).values({ companyId, executionWorkspaceId: null, issueId: blockerIssueId, + heartbeatRunId: blockerRun.id, phase: "workspace_finalize", status: "succeeded", startedAt: new Date(), finishedAt: new Date(), + metadata: { owningService: "workspace_recovery", preservationOperationIds: [] } }); + await expect(issueService(db).getDependencyReadiness(blockedIssueId)).resolves.toMatchObject({ isDependencyReady: true }); + + const result = await heartbeatService(db).reconcileResolvedDependencyWakes({ companyId, issueId: blockedIssueId }); + expect(result.selfBlockedSkipped).toBe(0); + expect(result.healed).toBe(1); + }); + it.each(["dependent", "blocker", "unattributed"] as const)("attributes shared-workspace export to its %s issue", async (exporter) => { const { companyId, agentId, blockedIssueId, blockerIssueId, executionWorkspaceId } = await seedResolvedDependencyBackstopFixture({ workspaceState: "finalized" }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 4456079b96f..65292d727b1 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1189,7 +1189,8 @@ export function recoveryService( const selfBlocked = row?.actorType === "agent" && row.agentId === agentId && details.status === "blocked"; const nativeSelfBlocked = row?.actorType === "system" && row.actorId === "native-status-committer" && row.runAgentId === agentId && - details.source === "native_status_decision" && details.toStatus === "blocked"; + details.source === "native_status_decision" && details.toStatus === "blocked" && + details.reasonCode !== "durable_dependency_blocker_bound"; if (!selfBlocked && !nativeSelfBlocked) return false; const newlyCompletedBlocker = await db.select({ id: issues.id }).from(issues).where(and( eq(issues.companyId, companyId), @@ -1199,8 +1200,10 @@ export function recoveryService( if (newlyCompletedBlocker.length > 0) return false; const newlyFinalizedBlocker = await db.select({ id: workspaceOperations.id }).from(workspaceOperations) .innerJoin(issues, and(eq(issues.companyId, companyId), inArray(issues.id, blockerIssueIds), - eq(issues.executionWorkspaceId, workspaceOperations.executionWorkspaceId), - eq(workspaceOperations.issueId, issues.id))) + eq(workspaceOperations.issueId, issues.id), + or(eq(issues.executionWorkspaceId, workspaceOperations.executionWorkspaceId), + and(isNull(workspaceOperations.executionWorkspaceId), + sql`${workspaceOperations.metadata}->>'owningService' = 'workspace_recovery'`)))) .where(and(eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded"), gt(workspaceOperations.finishedAt, blockedTransitionAt))) .limit(1); diff --git a/server/src/services/workspace-recovery-dependency-finalize.test.ts b/server/src/services/workspace-recovery-dependency-finalize.test.ts index 5f1210649db..daa70d2848e 100644 --- a/server/src/services/workspace-recovery-dependency-finalize.test.ts +++ b/server/src/services/workspace-recovery-dependency-finalize.test.ts @@ -1,13 +1,24 @@ import { and, eq } from "drizzle-orm"; import { describe, expect, it, vi } from "vitest"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; -import { issueRelations, issues, workspaceOperations } from "@paperclipai/db"; +import { randomUUID } from "node:crypto"; +import { activityLog, heartbeatRuns, issueRelations, issues, workspaceOperations } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { issueService } from "./issues.js"; +import { recoveryService } from "./recovery/service.js"; import { recoverRetainedWorkspace } from "./workspace-recovery.js"; import { readWorkspaceRecovery } from "./workspace-recovery-receipt.js"; import { useWorkspaceRecoveryFixture, acceptedRemoteExport, action, db } from "./__tests__/workspace-recovery-fixture.js"; const targets = vi.hoisted(() => new Map()); +const deliveryNotices = vi.hoisted(() => [] as Array<{ queue: string; hint: string | undefined }>); +vi.mock("./delivery-work-notifications.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, notifyDeliveryWork: async (...args: Parameters) => { + deliveryNotices.push({ queue: args[1], hint: args[2] }); + return actual.notifyDeliveryWork(...args); + } }; +}); vi.mock("./environment-execution-target.js", async importOriginal => { const actual = await importOriginal(); return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters[0]) => @@ -45,6 +56,40 @@ describe("dependency readiness after a done child's work is recovered", () => { expect(listed?.blockerAttention?.pendingFinalizeBlockerIssueIds ?? []).toEqual([]); }, 30_000); + it("wakes a parent whose completion was refused once the child's recovery saves its files", async () => { + const f = await acceptedRemoteExport(); + const [child] = await db.update(issues).set({ status: "done", completedAt: new Date(Date.now() - 120_000), executionWorkspaceId: f.workspaceId }) + .where(eq(issues.id, f.issueId)).returning(); + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId, + executionWorkspaceId: f.workspaceId, phase: "workspace_finalize", status: "failed", + startedAt: new Date(Date.now() - 90_000), finishedAt: new Date(Date.now() - 89_000) }); + const blockedTransitionAt = new Date(Date.now() - 60_000); + const [parent] = await db.insert(issues).values({ companyId: f.companyId, title: "Parent", status: "blocked", + priority: "medium", assigneeAgentId: child.assigneeAgentId, blockedTransitionAt }).returning(); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: f.issueId, relatedIssueId: parent.id, type: "blocks" }); + const [parentRun] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: child.assigneeAgentId!, + issueId: parent.id, status: "succeeded", runtimeMode: "native", contextSnapshot: { issueId: parent.id } }).returning(); + await db.insert(activityLog).values({ companyId: f.companyId, actorType: "system", actorId: "native-status-committer", + runId: parentRun.id, action: "issue.updated", entityType: "issue", entityId: parent.id, + details: { source: "native_status_decision", fromStatus: "in_progress", toStatus: "blocked", + reasonCode: "durable_dependency_blocker_bound" }, createdAt: blockedTransitionAt }); + await expect(readiness(parent.id)).resolves.toMatchObject({ isDependencyReady: false, pendingFinalizeBlockerIssueIds: [f.issueId] }); + + deliveryNotices.length = 0; + await recoverRetainedWorkspace(f.input); + expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered"); + expect(deliveryNotices).toContainEqual({ queue: DELIVERY_QUEUES.resolvedDependencies, + hint: JSON.stringify({ companyId: f.companyId }) }); + await expect(readiness(parent.id)).resolves.toMatchObject({ isDependencyReady: true }); + + const enqueueWakeup = vi.fn(async () => ({ id: randomUUID() }) as never); + const result = await recoveryService(db, { enqueueWakeup }).reconcileResolvedDependencyWakeBackstop({ + companyId: f.companyId, issueId: parent.id }); + expect(result).toMatchObject({ selfBlockedSkipped: 0, healed: 1 }); + expect(enqueueWakeup).toHaveBeenCalledWith(child.assigneeAgentId, expect.objectContaining({ + reason: "issue_blockers_resolved", payload: expect.objectContaining({ issueId: parent.id }) })); + }, 30_000); + it("releases the parent when the recovery saved through a snapshot without a workspace", async () => { const f = await recoveredDoneChildWithParent(); const finalize = await recoveryFinalize(f);