From b5fa4bf5c56db92febbdba2414d7c70f47d5793f Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 11 Sep 2026 18:06:07 -0500 Subject: [PATCH] fix(recovery): fence deferred admission against renewed blockers Reproduce a blocker appearing during admission, recheck current recovery before queuing, and consume the authorizing receipt atomically even when aggregation is suppressed. Retain the same saved receipt under daily-cap retries. Co-Authored-By: Paperclip --- .../explicit-native-continuation.test.ts | 50 ++++++++++++++++++- server/src/services/heartbeat.ts | 28 +++++++++-- 2 files changed, 72 insertions(+), 6 deletions(-) diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 47faf765d8..8a855140b6 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -3,7 +3,7 @@ import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_ import { remoteTerminationReceipt } from "./remote-execution-termination.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; -import { beforeAll, afterAll, describe, it, expect } from "vitest"; +import { beforeAll, afterAll, describe, it, expect, vi } from "vitest"; import { approvals, issueApprovals, issueThreadInteractions, agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, @@ -11,6 +11,7 @@ import { } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; +import * as continuationAdmission from "./explicit-native-continuation.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnvironmentRuntime } from "./heartbeat.js"; import { getExecutionBlocker } from "./execution-blocker.js"; @@ -113,6 +114,16 @@ const support = await getEmbeddedPostgresTestSupport(); expect(held.status).toBe("deferred_issue_execution"); expect(held.payload?.executionWait).toMatchObject({ reason: "issue_tree_hold_active" }); await db.update(issueTreeHolds).set({ status: "released" }).where(eq(issueTreeHolds.id, holdId)); + if (state === "available") { + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1, maxDailyRuns: 0 } } }).where(eq(agents.id, f.agentId)); + for (let attempt = 0; attempt < 2; attempt++) { + await makeDue(); + await heartbeatService(db).resumeExecutionWaitComments(); + } + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))).toHaveLength(0); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agentId)); + } await makeDue(); await Promise.all([heartbeatService(db).resumeExecutionWaitComments(), heartbeatService(db).resumeExecutionWaitComments()]); const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); @@ -122,6 +133,43 @@ const support = await getEmbeddedPostgresTestSupport(); if (state === "available") expect(after.runId).toBe(runs[0].id); }); + it("retains one saved receipt when a blocker reappears during transactional admission", async () => { + const f = await seed(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await heartbeatService(db).wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId }, + contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId } }); + const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + await db.update(issueRecoveryActions).set({ status: "resolved", evidence: { runId: f.sourceRunId } }) + .where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + const original = continuationAdmission.admitExplicitNativeContinuation; + let injected = false; + const admission = vi.spyOn(continuationAdmission, "admitExplicitNativeContinuation").mockImplementation(async input => { + if (input.issueId === f.issueId && !input.dryRun && !injected) { + injected = true; + await db.update(issueRecoveryActions).set({ status: "active" }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + } + return original(input); + }); + try { + await db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id)); + await heartbeatService(db).resumeExecutionWaitComments(); + expect(injected).toBe(true); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))).toHaveLength(0); + const [after] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id)); + expect(after.status).toBe("deferred_issue_execution"); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1); + } finally { admission.mockRestore(); } + await db.update(issueRecoveryActions).set({ status: "resolved" }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id)); + await Promise.all([heartbeatService(db).resumeExecutionWaitComments(), heartbeatService(db).resumeExecutionWaitComments()]); + const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(runs).toHaveLength(1); + const [after] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id)); + expect(after).toMatchObject({ status: "coalesced", runId: runs[0].id }); + }); + it.each(["live", "remote", "provider_event"])("does not accept invalid local stop proof: %s", async kind => { const f = await seed(); if (kind === "remote") { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index eac438cdd5..9eb65f7e1d 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -25842,9 +25842,9 @@ export function heartbeatService( let continuationWait = { reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." }; const deferBlockedExecution = async ( - executionBlocker: NonNullable>>, + executionBlocker: Awaited>, ) => { - const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait }; + const condition = { recoveryActionId: executionBlocker?.recoveryActionId ?? null, ...continuationWait }; if (executionWaitRequestId) { await tx.update(agentWakeupRequests).set({ payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${JSON.stringify(condition)}::jsonb)`, @@ -25877,7 +25877,7 @@ export function heartbeatService( ...durableReceiptFields, companyId: agent.companyId, agentId, source, triggerDetail, reason: "execution_reconciliation_required", - error: executionBlocker.nextAction, + error: executionBlocker?.nextAction ?? continuationWait.message, payload, requestedByActorType: opts.requestedByActorType ?? null, requestedByActorId: opts.requestedByActorId ?? null, @@ -26615,7 +26615,7 @@ export function heartbeatService( tx, ); if (dailyCapBlock) { - if (executionWaitRequestId && executionBlocker) { + if (executionWaitRequestId) { continuationWait = { reason: dailyCapBlock.reason, message: "The agent has reached its daily limit. Your message is saved until work can resume." }; return deferBlockedExecution(executionBlocker); @@ -26660,7 +26660,12 @@ export function heartbeatService( agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId, reason, commentId: wakeCommentId ?? null, successorRunId: explicitContinuationRunId, }); - if (!explicitContinuation && executionBlocker) return deferBlockedExecution(executionBlocker); + // Recovery can change while earlier admission gates await I/O. Use + // the current blocker, not the snapshot from the start of admission. + const remainingExecutionBlocker = await getExecutionBlocker( + tx as unknown as Db, issue.companyId, issue.id, + ); + if (remainingExecutionBlocker) return deferBlockedExecution(remainingExecutionBlocker); if (explicitContinuation) { enrichedContextSnapshot.forceFreshSession = true; enrichedContextSnapshot.previousRunId = explicitContinuation.previousRunId; @@ -26763,6 +26768,19 @@ export function heartbeatService( }) .where(eq(agentWakeupRequests.id, wakeupRequest.id)); + // The receipt that authorized this run is consumed independently of + // optional aggregation of other comments. A later blocker may suppress + // aggregation, but must never leave this message eligible for replay. + if (executionWaitRequestId) { + const [consumed] = await tx.update(agentWakeupRequests).set({ + status: "coalesced", runId: newRun.id, finishedAt: new Date(), updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.id, executionWaitRequestId), + eq(agentWakeupRequests.companyId, issue.companyId), eq(agentWakeupRequests.agentId, agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + )).returning({ id: agentWakeupRequests.id }); + if (!consumed) throw conflict("Saved message changed before admission"); + } + if (adoptedComments.length) { await tx .update(agentWakeupRequests)