diff --git a/doc/dependency-wake-recovery.md b/doc/dependency-wake-recovery.md index 0187b99c6f3..41e94ef5ed5 100644 --- a/doc/dependency-wake-recovery.md +++ b/doc/dependency-wake-recovery.md @@ -13,6 +13,13 @@ unattributed operation in the same shared workspace does not remove its wait. Normal readiness, duplicate-wake, interaction, pause, and execution gates still apply. A board transition into blocked keeps the existing recovery behavior. +A dependency completion can arrive while the assignee answers a question about +the blocked task. Its saved wake waits behind that turn. Deferred-receipt recovery +reconstructs these dependency wakes after the turn and its cleanup finish, even +when they contain no comment or interaction response. It uses the existing +release and dispatch checks; task cancellation, agent pause, and pending cleanup +still prevent execution. + Cancelling a run or pausing its agent suppresses immediate stranded-work recovery. A queued user message remains a separate continuation request. The release path still preserves independently authorized deferred input; stopping the old run diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index be0135fc64e..414ac9d5abf 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -15,7 +15,7 @@ import { issueTreeControlService } from "../services/issue-tree-control.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, agentWakeupRequests, authUsers, chatActions, environmentLeases, nativeRunFinalizations, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, projects, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import { activityLog, agents, agentWakeupRequests, authUsers, chatActions, environmentLeases, nativeRunFinalizations, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, projects, reconcileDatabaseWork, heartbeatRuns, issueComments, issueRelations, issues } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -158,6 +158,96 @@ describe("deferred receipt durable dispatch", () => { source: "on_demand", status: "completed", reason: "task_retry", payload: { issueId: f.issue.id } }); await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; await quiet(); }); + async function strandedDependency() { + const f = await fixture(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: f.issue.id, + status: "succeeded", errorCode: null, contextSnapshot: { issueId: f.issue.id, + wakeReason: "issue_commented", dependencyBlockedInteraction: true } }).where(eq(heartbeatRuns.id, f.source.id)); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, + runId: f.source.id, phase: "committed" }); + await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issue.id)); + const [blocker] = await db.insert(issues).values({ companyId: f.companyId, title: "Completed review", + status: "done", completedAt: new Date() }).returning(); + await db.insert(issueRelations).values({ companyId: f.companyId, issueId: blocker.id, + relatedIssueId: f.issue.id, type: "blocks" }); + const [intent] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "automation", reason: "issue_blockers_resolved", status: "coalesced", + requestedByActorType: "system", requestedByActorId: "native-status-committer", + payload: { issueId: f.issue.id, resolvedBlockerIssueId: blocker.id }, + }).returning(); + const [receipt] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, + source: "automation", reason: "issue_execution_deferred", status: "deferred_issue_execution", + requestedByActorType: "system", requestedByActorId: `native-status-wake-dispatch:${intent.id}`, + payload: { issueId: f.issue.id, resolvedBlockerIssueId: blocker.id, _paperclipWakeContext: { + issueId: f.issue.id, wakeReason: "issue_blockers_resolved", statusDecisionSource: "native_status_decision", + nativeStatusWakeIntentId: intent.id } }, + }).returning(); + return { ...f, receipt, blocker }; + } + + it.each(["restart", "cleanup"] as const)("recovers a dependency wake after native audit %s into one continuation", async event => { + const f = await strandedDependency(); + const { receipt } = f; + const [lease] = event === "cleanup" ? await db.insert(environmentLeases).values({ companyId: f.companyId, + heartbeatRunId: f.source.id, issueId: f.issue.id, provider: "daytona", providerLeaseId: "dependency-finalizing", + status: "active", leasePolicy: "reuse_by_environment" }).returning() : []; + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + if (lease) { + await waitForReal(async () => { + await Promise.allSettled([...passes]); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + await environmentService(db).releaseLease(lease.id, "retained", { cleanupStatus: "success" }); + } + await waitForReal(async () => { + await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions(); + const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id)); + const pending = saved.runId ? (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, saved.runId)))[0] : null; + expect(saved.status, JSON.stringify({ status: pending?.status, error: pending?.error, errors })).toBe("completed"); + expect(saved.runId).not.toBeNull(); + }, { timeout: 15_000 }); + await finish(f.issue.id); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId)); + const dependencyRuns = runs.filter(run => run.contextSnapshot?.wakeReason === "issue_blockers_resolved"); + expect(dependencyRuns).toHaveLength(1); + const [resumed] = dependencyRuns; + expect(resumed).toMatchObject({ status: "succeeded", stdoutExcerpt: expect.stringContaining("receipt-proof"), + contextSnapshot: { wakeReason: "issue_blockers_resolved" } }); + expect(resumed.contextSnapshot?.dependencyBlockedInteraction).not.toBe(true); + await heartbeat.resumeQueuedRuns(); + const after = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId)); + expect(after.filter(run => run.contextSnapshot?.wakeReason === "issue_blockers_resolved")).toHaveLength(1); + expect(after.find(run => run.id === f.source.id)?.status).toBe("succeeded"); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]) + .toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null }); + await quiet(); + }); + it.each(["paused", "cleanup", "cancelled"] as const)("keeps the %s gate when recovering a dependency wake", async gate => { + const f = await strandedDependency(); + if (gate === "paused") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, f.agent.id)); + if (gate === "cancelled") await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issue.id)); + if (gate === "cleanup") await db.insert(environmentLeases).values({ companyId: f.companyId, + heartbeatRunId: f.source.id, issueId: f.issue.id, provider: "daytona", providerLeaseId: "dependency-cleanup", + status: "active", leasePolicy: "reuse_by_environment" }); + await heartbeat.resumeQueuedRuns(); + const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.receipt.id)); + expect(saved).toMatchObject({ status: gate === "cancelled" ? "cancelled" : "deferred_issue_execution", runId: null }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + }); + it("checks unresolved blockers before a recovered dependency wake can execute", async () => { + const f = await strandedDependency(); + await db.update(issues).set({ status: "blocked", completedAt: null }).where(eq(issues.id, f.blocker.id)); + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.receipt.id)); + expect(receipt).toMatchObject({ status: "skipped", runId: expect.any(String) }); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)); + expect(run).toMatchObject({ status: "cancelled", errorCode: "issue_dependencies_blocked", stdoutExcerpt: null }); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].status).toBe("blocked"); + }); it.each([false, true])("reconsiders every saved receipt during repeated company eligibility changes (expired wait: %s)", async expiredWait => { const f = await fixture(); const ids = Array.from({ length: 178 }, (_, index) => `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`); diff --git a/server/src/services/heartbeat/deferred-receipt-work.ts b/server/src/services/heartbeat/deferred-receipt-work.ts index b799f10b0a7..915d72b7b07 100644 --- a/server/src/services/heartbeat/deferred-receipt-work.ts +++ b/server/src/services/heartbeat/deferred-receipt-work.ts @@ -5,6 +5,7 @@ import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { issueBlockersResolvedWakePayloadSql } from "../issue-dependency-wakeups.js"; type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean }; type NativeIntentOptions = DeferredExecutionScope & { @@ -74,6 +75,7 @@ export function registerDeferredReceiptWork( sql`jsonb_typeof(${agentWakeupRequests.payload}#>'{_paperclipWakeContext,wakeCommentIds}') = 'array' and ${agentWakeupRequests.payload}#>'{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`, sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`, + issueBlockersResolvedWakePayloadSql, sql`exists (select 1 from heartbeat_runs r where r.company_id = ${agentWakeupRequests.companyId} and r.runtime_mode = 'legacy' and r.status = 'cancelled' and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 4da678e4b38..514d0a005b7 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -144,7 +144,11 @@ import { isThrottleCandidateIssueRewake, } from "../issue-rewake-throttle.js"; import { logActivity } from "../activity-log.js"; -import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON } from "../issue-dependency-wakeups.js"; +import { + ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, + isIssueBlockersResolvedWakePayload, + issueBlockersResolvedWakePayloadSql, +} from "../issue-dependency-wakeups.js"; import { resumeIfPauseAlreadyReleased } from "../issue-tree-resume.js"; import { ISSUE_PAUSED_ERROR_CODE, @@ -1757,7 +1761,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende or(and( sql`jsonb_typeof(${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}') = 'array'`, sql`${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`, - ), sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`), + ), sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`, + issueBlockersResolvedWakePayloadSql), sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`, sql`NOT EXISTS (SELECT 1 FROM heartbeat_runs live_issue_run WHERE live_issue_run.company_id = ${issues.companyId} @@ -1768,7 +1773,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); const releasedNativeSources = new Set(); for (const { wake } of strandedQueues) { - if (!queuedCommentIdsFromWakePayload(wake.payload).length && + const dependencyWake = isIssueBlockersResolvedWakePayload(wake.payload); + if (!dependencyWake && !queuedCommentIdsFromWakePayload(wake.payload).length && !await readQueuedInteractionResponse(db, wake.companyId, String(wake.payload?.issueId), wake.payload)) continue; const [latest] = await db.select({ id: heartbeatRuns.id, diff --git a/server/src/services/issue-dependency-wakeups.ts b/server/src/services/issue-dependency-wakeups.ts index ee09f3e55ff..4d891d583a8 100644 --- a/server/src/services/issue-dependency-wakeups.ts +++ b/server/src/services/issue-dependency-wakeups.ts @@ -1,10 +1,19 @@ -import { and, eq, inArray } from "drizzle-orm"; +import { and, eq, inArray, sql } from "drizzle-orm"; import { createHash } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { agentWakeupRequests } from "@paperclipai/db"; export const ISSUE_BLOCKERS_RESOLVED_WAKE_REASON = "issue_blockers_resolved"; +export const issueBlockersResolvedWakePayloadSql = + sql`${agentWakeupRequests.payload}#>>'{_paperclipWakeContext,wakeReason}' = ${ISSUE_BLOCKERS_RESOLVED_WAKE_REASON}`; + +export function isIssueBlockersResolvedWakePayload(payload: Record | null | undefined): boolean { + const context = payload?._paperclipWakeContext; + return context !== null && typeof context === "object" && !Array.isArray(context) + && (context as Record).wakeReason === ISSUE_BLOCKERS_RESOLVED_WAKE_REASON; +} + /** * Whether moving an issue into `blocked` should immediately wake its assignee * because every recorded blocker is already resolved ("blockers restored").