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

fix: retain saved messages while decisions are pending 

Recheck pending approvals and questions when cleanup already cleared the recovery action, and distinguish final admission rejection from a non-applicable continuation. Add deterministic cross-connection admission coverage for held and resolved recovery, receipt preservation, and exactly-once resumption.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta 2026-09-12 13:29:51 -05:00
parent 5637dc2ab6
commit 91645fe400
3 changed files with 77 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -236,6 +236,69 @@ const support = await getEmbeddedPostgresTestSupport();
expect(after).toMatchObject({ status: "coalesced", runId: runs[0].id });
});
it.each([
["approval", "held"], ["question", "held"],
["approval", "resolved"], ["question", "resolved"],
] as const)("retains a saved message when a %s appears at final admission after recovery is %s", async (kind, recovery) => {
const f = await seed();
// Occupy the agent so a regression queues work without invoking a provider.
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));
expect(waiting.status).toBe("deferred_issue_execution");
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId));
if (recovery === "resolved") await db.update(issueRecoveryActions).set({ status: "resolved", evidence: { runId: f.sourceRunId } })
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
const decisionId = randomUUID();
if (kind === "question") await db.insert(issueThreadInteractions).values({
id: decisionId, companyId: f.companyId, issueId: f.issueId,
kind: "ask_user_questions", status: "resolved", payload: { version: 1, questions: [] },
});
else {
await db.insert(approvals).values({ id: decisionId, companyId: f.companyId, type: "hire_agent", status: "approved", payload: {} });
await db.insert(issueApprovals).values({ companyId: f.companyId, issueId: f.issueId, approvalId: decisionId });
}
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;
// Change decision state on another connection after the early reads.
// Final transactional admission must observe that committed change.
if (kind === "question") await db.update(issueThreadInteractions).set({ status: "pending" }).where(eq(issueThreadInteractions.id, decisionId));
else await db.update(approvals).set({ status: "pending" }).where(eq(approvals.id, decisionId));
}
return original(input);
});
const makeDue = () => db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id));
try {
await makeDue();
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).toMatchObject({ status: "deferred_issue_execution", runId: null });
expect(after.payload?.executionWait).toMatchObject({ reason: "decision_pending" });
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1);
// Unchanged retries must preserve the same receipt, including after the
// recovery blocker itself has been cleared.
await makeDue();
await heartbeatService(db).resumeExecutionWaitComments();
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))).toHaveLength(0);
} finally { admission.mockRestore(); }
if (kind === "question") await db.update(issueThreadInteractions).set({ status: "resolved" }).where(eq(issueThreadInteractions.id, decisionId));
else await db.update(approvals).set({ status: "approved" }).where(eq(approvals.id, decisionId));
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")));
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") {

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

@ -34,6 +34,7 @@ export async function admitExplicitNativeContinuation(input: {
reason: string | null; commentId: string | null; successorRunId: string;
failedRunId?: string | null;
dryRun?: boolean;
resumingSavedMessage?: boolean;
onBlocked?: (reason: string, message: string) => void;
}): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string } | null> {
const { db, companyId, issueId, agentId, actorId, commentId } = input;
@ -61,7 +62,9 @@ export async function admitExplicitNativeContinuation(input: {
eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
executionBlockerPredicate(),
)).for("update");
if (!actions.length) return null;
// Cleanup can remove the recovery action before a saved message is retried.
// Its pending decisions still gate admission, even without a hold to retire.
if (!actions.length && !input.resumingSavedMessage) return null;
const blocker = await getExecutionBlocker(db, companyId, issueId);
if (blocker && blocker.recoveryActionId === null) return null;
const [pendingInteraction] = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and(
@ -73,6 +76,7 @@ export async function admitExplicitNativeContinuation(input: {
)).where(and(eq(issueApprovals.companyId, companyId), eq(issueApprovals.issueId, issueId),
inArray(approvals.status, ["pending", "revision_requested"]))).limit(1);
if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start.");
if (!actions.length) return null;
const sources: Run[] = [];
const cancelledStartupIds = new Set<string>();

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

@ -26961,17 +26961,25 @@ export function heartbeatService(
return { kind: "skipped" as const };
}
let continuationRejected = false;
const explicitContinuation = await admitExplicitNativeContinuation({
db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id,
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId,
resumingSavedMessage: Boolean(executionWaitRequestId),
onBlocked: (reason, message) => {
continuationRejected = true;
continuationWait = { reason, message };
},
});
// 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);
// A decision can reject a saved message after cleanup has removed
// every recovery blocker; null can also mean no applicable hold to retire.
if (continuationRejected || remainingExecutionBlocker) return deferBlockedExecution(remainingExecutionBlocker);
if (explicitContinuation) {
enrichedContextSnapshot.forceFreshSession = true;
enrichedContextSnapshot.previousRunId = explicitContinuation.previousRunId;