diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index 414ac9d5abf..76fe58ed5dd 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -185,6 +185,34 @@ describe("deferred receipt durable dispatch", () => { return { ...f, receipt, blocker }; } + it("starts a deferred children-completed wake left behind when its parent's run blocked the parent (restart chaos t06)", async () => { + 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_assigned" } }).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 [child] = await db.insert(issues).values({ companyId: f.companyId, title: "Part 1", parentId: f.issue.id, + status: "done", completedAt: new Date() }).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", idempotencyKey: `issue_children_completed:${f.issue.id}`, + payload: { issueId: f.issue.id, childIssueIds: [child.id], _paperclipWakeContext: { issueId: f.issue.id, + taskId: f.issue.id, wakeReason: "issue_children_completed", source: "native_status_decision", childIssueIds: [child.id] } }, + }).returning(); + await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; + await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready; + await waitForReal(async () => { + await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions(); + const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.id)); + expect(saved.status, JSON.stringify({ 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)); + expect(runs.filter(run => run.contextSnapshot?.wakeReason === "issue_children_completed")).toHaveLength(1); + await quiet(); + }); + 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; diff --git a/server/src/services/heartbeat/deferred-receipt-work.ts b/server/src/services/heartbeat/deferred-receipt-work.ts index 915d72b7b07..5a3e3bddd64 100644 --- a/server/src/services/heartbeat/deferred-receipt-work.ts +++ b/server/src/services/heartbeat/deferred-receipt-work.ts @@ -5,7 +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"; +import { dependencyResolvedWakePayloadSql } from "../issue-dependency-wakeups.js"; type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean }; type NativeIntentOptions = DeferredExecutionScope & { @@ -75,7 +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, + dependencyResolvedWakePayloadSql, 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 dfeba70e0a1..be86fd31205 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -153,8 +153,8 @@ import { import { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; import { ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, - isIssueBlockersResolvedWakePayload, - issueBlockersResolvedWakePayloadSql, + isDependencyResolvedWakePayload, + dependencyResolvedWakePayloadSql, } from "../issue-dependency-wakeups.js"; import { resumeIfPauseAlreadyReleased } from "../issue-tree-resume.js"; import { @@ -1901,7 +1901,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende sql`jsonb_typeof(${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}') = 'array'`, sql`${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`, ), sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`, - issueBlockersResolvedWakePayloadSql), + dependencyResolvedWakePayloadSql), 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} @@ -1912,7 +1912,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50); const releasedNativeSources = new Set(); for (const { wake } of strandedQueues) { - const dependencyWake = isIssueBlockersResolvedWakePayload(wake.payload); + const dependencyWake = isDependencyResolvedWakePayload(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({ diff --git a/server/src/services/issue-dependency-wakeups.ts b/server/src/services/issue-dependency-wakeups.ts index 4d891d583a8..48591daaedc 100644 --- a/server/src/services/issue-dependency-wakeups.ts +++ b/server/src/services/issue-dependency-wakeups.ts @@ -5,13 +5,18 @@ 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 const ISSUE_CHILDREN_COMPLETED_WAKE_REASON = "issue_children_completed"; -export function isIssueBlockersResolvedWakePayload(payload: Record | null | undefined): boolean { +const DEPENDENCY_RESOLVED_WAKE_REASONS = [ISSUE_BLOCKERS_RESOLVED_WAKE_REASON, ISSUE_CHILDREN_COMPLETED_WAKE_REASON]; + +export const dependencyResolvedWakePayloadSql = + sql`${agentWakeupRequests.payload}#>>'{_paperclipWakeContext,wakeReason}' in (${sql.join( + DEPENDENCY_RESOLVED_WAKE_REASONS.map(reason => sql`${reason}`), sql`, `)})`; + +export function isDependencyResolvedWakePayload(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; + && DEPENDENCY_RESOLVED_WAKE_REASONS.includes(String((context as Record).wakeReason)); } /**