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

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 <noreply@paperclip.ing>
This commit is contained in:
Dotta 2026-09-11 18:06:07 -05:00
parent 5809e5b97e
commit b5fa4bf5c5
2 changed files with 72 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
server/src/services/explicit-native-continuation.test.ts
View File

@ -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") {

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

@ -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<Awaited<ReturnType<typeof getExecutionBlocker>>>,
executionBlocker: Awaited<ReturnType<typeof getExecutionBlocker>>,
) => {
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)