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

fix(recovery): resume saved messages after cleanup resolves blockers 

Keep terminal stop receipts intact when a late process callback arrives. Reconsider saved execution-wait comments independently of recovery-action state, validate their author under the task lock, and retain admission gates and retry deduplication.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta 2026-09-11 17:56:04 -05:00
parent 18fcc893bc
commit 5809e5b97e
2 changed files with 85 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -73,6 +73,55 @@ const support = await getEmbeddedPostgresTestSupport();
expect(await admit(f, true)).toBeNull();
});
it("ignores late process callbacks without invalidating terminal stop evidence", async () => {
const f = await seed();
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
expect(await recordNativeLocalProcessStop(db, source)).toBe(true);
await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id));
expect(await persistHeartbeatRunProcessMetadata(db, source.id, {
pid: 999999999, processGroupId: null, startedAt: new Date().toISOString(),
})).toBeNull();
expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(true);
expect(await admit(f, true)).toMatchObject({ previousRunId: source.id });
});
it.each(["available", "deleted", "different-author"])("reconsiders saved messages after cleanup resolves the blocker: %s", async state => {
const f = await seed();
// Keep admission queued so the test never launches a real 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.payload?.executionWait).toMatchObject({ reason: "process_running" });
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId));
const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
expect(await recordNativeLocalProcessStop(db, stopped)).toBe(true);
await db.update(issueRecoveryActions).set({ status: "resolved", evidence: { runId: f.sourceRunId } })
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
if (state === "deleted") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, f.commentId));
if (state === "different-author") await db.update(issueComments).set({ authorUserId: "someone-else" }).where(eq(issueComments.id, f.commentId));
const makeDue = () => db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id));
const holdId = randomUUID();
await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause", status: "active" });
await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId, issueId: f.issueId, depth: 0, issueTitle: "Deploy", issueStatus: "blocked" });
await makeDue();
await heartbeatService(db).resumeExecutionWaitComments();
const [held] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id));
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));
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(state === "available" ? 1 : 0);
const [after] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id));
expect(after.status).toBe(state === "available" ? "coalesced" : "deferred_issue_execution");
if (state === "available") expect(after.runId).toBe(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") {

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

@ -8615,7 +8615,13 @@ export async function persistHeartbeatRunProcessMetadata(
: startedAt,
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, runId))
// Late callbacks must not rewrite completed process history or revoke
// its stop evidence after asynchronous identity lookup.
.where(and(
eq(heartbeatRuns.id, runId),
eq(heartbeatRuns.status, "running"),
isNull(heartbeatRuns.finishedAt),
))
.returning()
.then((rows) => rows[0] ?? null);
if (run?.runtimeMode === "native") await appendHeartbeatRunEvent(tx as unknown as Db, {
@ -10028,10 +10034,7 @@ export function heartbeatService(
sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`,
eq(issues.assigneeAgentId, agentWakeupRequests.agentId)))
.innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active")))
.where(and(exists(db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
eq(issueRecoveryActions.companyId, issues.companyId), eq(issueRecoveryActions.sourceIssueId, issues.id),
executionBlockerPredicate(),
))), eq(agentWakeupRequests.status, "deferred_issue_execution"),
.where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"),
eq(agentWakeupRequests.requestedByActorType, "user"),
sql`${agentWakeupRequests.payload}->'executionWait' is not null`,
lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)),
@ -10053,7 +10056,26 @@ export function heartbeatService(
// Match normal admission's deterministic current blocker selection. An
// arbitrary historical action must not choose the retry's source run.
const blocker = await getExecutionBlocker(db, wake.companyId, issueId);
const sourceId = blocker?.runId;
if (!blocker) {
// Cleanup may already have resolved the recovery action. The saved
// receipt remains work; normal admission rechecks ownership, holds,
// budgets and any blocker created since this read under the issue lock.
if (wake.idempotencyKey?.startsWith("chat-inbound:") ||
!["issue_commented", "issue_reopened_via_comment"].includes(wake.reason ?? "")) continue;
const payload = parseObject(wake.payload);
const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]);
await enqueueWakeup(wake.agentId, {
source: wake.source as WakeupOptions["source"],
triggerDetail: (wake.triggerDetail ?? undefined) as WakeupOptions["triggerDetail"],
reason: wake.reason, payload, contextSnapshot: context,
requestedByActorType: "user", requestedByActorId: wake.requestedByActorId,
idempotencyKey: `execution-wait-comment:${wake.id}`,
}, wake.id).catch(err => {
logger.warn({ err, requestId: wake.id }, "failed to resume saved execution-wait message after recovery");
});
continue;
}
const sourceId = blocker.runId;
if (!sourceId || !isUuidLike(sourceId)) continue;
const run = await getRun(sourceId);
if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue;
@ -25505,6 +25527,14 @@ export function heartbeatService(
if (!pending || !wakeCommentId || !queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) {
return { kind: "deferred" as const };
}
const [comment] = await tx.select({ id: issueComments.id }).from(issueComments).where(and(
eq(issueComments.companyId, agent.companyId), eq(issueComments.issueId, issueId),
sql`${issueComments.id}::text = ${wakeCommentId}`, eq(issueComments.authorType, "user"),
eq(issueComments.authorUserId, opts.requestedByActorId ?? ""),
isNull(issueComments.deletedAt), isNull(issueComments.createdByRunId),
sql`length(trim(${issueComments.body})) > 0`,
));
if (!comment) return { kind: "deferred" as const };
}
let automaticParentRunId: string | null = null;
if (