From a94881d1a71b1e24a7990c01699a9e9b8cc5b16e Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 08:54:14 -0600 Subject: [PATCH] fix(heartbeat): save a comment refused by a temporary hold and route post-quarantine input to its own gate (#19829) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A user's follow-up comment or Retry must be delivered once, eventually, even when a temporary company admission hold (accounting still settling) is active > - #19741 kept claimed runs queued through such a hold and woke held work when it cleared > - A restart-chaos wave still lost a comment, because its wake was refused before admission and written as a terminal skip > - The same wave left a Retry waiting forever for a stop record of a process that never existed > - This pull request saves the refused comment as a deferred receipt, and routes input after a pre-launch quarantine failure to the quarantine's own gate > - The benefit is that the comment is delivered after the hold, and the task asks for the action that actually unblocks it instead of offering a Retry that can never start ## Linked Issues or Issue Description Refs #19741 and #19702. **What happened?** In a restart-chaos wave, a failed run's accounting held the company for 600 s. 1. A user commented while the task's run was live. The enqueue budget gate refused the comment's wake before it reached the deferral behind the live run. Because it was a user request with no saved receipt, it was written as `skipped` (`budget.blocked`). The live run finished without it and the task ended `done`, so the comment was lost. 2. A Retry created a run that the native cleanup-quarantine fence refused before it requested any runner launch (`native_session_cleanup_quarantined`, no process events). A second Retry then waited with `process_identity_missing` ("The previous run has no verified stop record") for a process that never existed. The task kept offering Retry, and that Retry could never start. **Expected behavior** A comment refused by a temporary hold stays saved and is delivered once later. Input after a run that was refused before launch is handled by the quarantine gate: Retry is not offered (send a new message), and a new message gets the verified-stop check of the previous provider session. ## What Changed - `heartbeat/queue.ts`: when the enqueue budget gate refuses a user's comment wake and the block is temporary (not a reached hard stop, per `isBoardOwnedBudgetBlock`), the wake is saved as a `deferred_issue_execution` receipt. The receipt carries the comment ids and an `executionWait` of `budget.blocked`, and the gate notifies the deferred-receipt worker and throws `PersistedExecutionWaitRefusal`. It is not written as `skipped`. Normal promotion (when the live run ends) and #19741's hold-release wakeups deliver it. - `native-local-process-stop.ts`: new `nativeRunRefusedBeforeLaunch`. It returns true for a failed native run with `native_session_cleanup_quarantined`, no recorded process, and no launch-request or identity event. - `explicit-native-continuation.ts`: such a run passes the process-identity check, so the existing quarantine branch decides. A Retry gets `cleanup_quarantined` (shown with `canRetry: false` and "send a new message"). A new message gets `verifyStoppedNativeSessionForContinuation`. Any launch request or identity record keeps the original stop proofs. - `native-session-executor.ts` (`verifyStoppedNativeSessionForContinuation`): when the source has no process record and no launch event, the verifier relies on the previous owner's stop evidence. Before this change it required the source to share the owner's process identity, so a new message stayed held even after the owner stopped. A source that carries a process identity must still match the owner exactly. ## Verification Each new test fails on `master` and passes with this change, on embedded Postgres: - `server/src/__tests__/heartbeat-held-comment-wake.test.ts`: a comment whose wake the hold refuses at enqueue is saved with a `budget.blocked` wait and is not skipped. It is delivered exactly once after accounting settles. - `server/src/services/explicit-native-continuation.test.ts`: - A run refused by the quarantine before launch routes a new message to the quarantine's stop verification and a Retry to `cleanup_quarantined`. These failed on `master` with `process_identity_missing`. - Guards (green on both): a quarantined run that requested a launch, and another failure without a process record, still require `process_identity_missing`. - `server/src/services/quarantined-task-retry.test.ts`: a never-launched quarantined source is certified once the previous owner has stopped (red on `master`). It stays held if the source requested a launch or the owner is still running. All 31 tests pass, and the `stopped native conversation physical cleanup` suite passes (46). - Related suites pass locally (6 files, 620 tests): explicit native continuation, held comment wake, queued comments routes, comment wake batching, refused assignment wake, deferred receipt work. Two `deferred-execution-receipt-work` tests failed once under parallel load and passed when run alone. ## Risks - The saved-comment path applies only to user comment wakes refused by a temporary budget hold. Hard stops, system wakes, case handoffs, durable chat requests, and existing saved receipts are unchanged. - The pre-launch rule is limited to the quarantine failure code with no launch evidence. Every other missing process identity still requires the existing stop proof. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, run as an agent in Claude Code with tool use (shell, file editing, local test execution, kept restart-chaos instance inspection). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../heartbeat-held-comment-wake.test.ts | 27 +++++++++++++++++++ .../explicit-native-continuation.test.ts | 26 ++++++++++++++++++ .../services/explicit-native-continuation.ts | 3 ++- server/src/services/heartbeat/queue.ts | 25 +++++++++++++++++ .../src/services/native-local-process-stop.ts | 15 +++++++++++ .../native-runtime/native-session-executor.ts | 11 +++++--- .../services/quarantined-task-retry.test.ts | 25 +++++++++++++++++ 7 files changed, 128 insertions(+), 4 deletions(-) diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index 31b747e9c24..8f57d676e30 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -148,6 +148,33 @@ describePostgres("a promoted comment refused by a temporary admission hold", () expect(deliveries(f.commentId)).toHaveLength(1); }, 180_000); + it("saves a comment whose wake the hold refuses at enqueue and delivers it once after the hold clears", async () => { + // s7532/t06: the comment arrived during the hold; its wake was refused + // before admission and written as a terminal skip, so it was lost. + const f = await seedHeldComment(); + await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorUserId: f.userId, body: "Posted during the hold." }).returning(); + await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, + contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, + source: "issue.comment", wakeReason: "issue_commented" }, + requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toThrow(/await accounting/); + const saved = (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) + .filter(wake => (wake.payload as { commentId?: string }).commentId === comment.id); + expect(saved).toEqual([expect.objectContaining({ status: "deferred_issue_execution", runId: null, + payload: expect.objectContaining({ executionWait: expect.objectContaining({ reason: "budget.blocked" }) }) })]); + await settle(f); + expect(deliveries(comment.id)).toHaveLength(0); + + await settleAccounting(f); + await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }).where(eq(agentWakeupRequests.id, saved[0].id)); + await settle(f); await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + await settle(f); + expect(deliveries(comment.id)).toHaveLength(1); + }, 180_000); + it("keeps a Retry clicked during the hold pending and starts it exactly once after the hold clears", async () => { const f = await seedHeldComment(); // The run the user retries failed; its saved comment was already discarded. diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 376e00aac93..a1782ae298c 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -94,6 +94,32 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes onBlocked: reason => blocked.push(reason) }); return { result, blocked, reads }; } + it.each([ + { shape: "refused by the cleanup quarantine before launch", errorCode: "native_session_cleanup_quarantined", launched: false, + comment: "local_cleanup", retry: "cleanup_quarantined" }, + { shape: "quarantined after it requested a launch", errorCode: "native_session_cleanup_quarantined", launched: true, + comment: "process_identity_missing", retry: "process_identity_missing" }, + { shape: "failed for another reason without a process record", errorCode: "adapter_failed", launched: false, + comment: "process_identity_missing", retry: "process_identity_missing" }, + ])("routes input after a run $shape to the gate that can clear it", async ({ errorCode, launched, comment, retry }) => { + // s7532/t04: a Retry's run failed at the quarantine fence before it asked + // for a runner. Input then waited for a stop record of a process that + // never existed ("process_identity_missing", Retry still offered) instead + // of the quarantine's own gate, which names the action that clears it. + const f = await seed(); + await db.update(heartbeatRuns).set({ processPid: null, errorCode }).where(eq(heartbeatRuns.id, f.sourceRunId)); + if (launched) await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.sourceRunId, agentId: f.agentId, + seq: 1, eventType: "native.process_start_requested", stream: "system", level: "info", message: "launch requested", + payload: { hostname: "another-host" } }); + const { result, blocked } = await readAdmission(f, false); + expect(result).toBeNull(); + expect(blocked).toEqual([comment]); + const retryBlocked: string[] = []; + expect(await admitExplicitNativeContinuation({ ...f, db, dryRun: true, reason: "retry_failed_run", + failedRunId: f.sourceRunId, onBlocked: reason => retryBlocked.push(reason) })).toBeNull(); + expect(retryBlocked).toEqual([retry]); + }); + it("keeps admission and the complete locked session profile while omitting prompt data", async () => { const f = await seed(); const payload = randomBytes(128 * 1024).toString("base64"); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index d96797f44e9..d48da1101de 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -16,7 +16,7 @@ import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { canContinueCancelledRun, userStopBoundaryAt } from "./run-cancellation.js"; import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; import { isCancelledNativeStartup, isFailedNativePreparation, isFailedNativePreparationRetryAncestor } from "./cancelled-native-startup.js"; -import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped, recordUnidentifiedLocalRunnerStop } from "./native-local-process-stop.js"; +import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, nativeRunRefusedBeforeLaunch, recordedLocalExecutionStopped, recordUnidentifiedLocalRunnerStop } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup, nativeSessionCleanupBlock } from "../vendor/paperclip-runner/index.js"; import { hasRecoveredCleanupStopReceipt, hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; import { z } from "zod"; @@ -508,6 +508,7 @@ export async function admitExplicitNativeContinuation(input: { if (!run.processPid && !run.processGroupId && !await hasNativeLocalProcessStop(db, companyId, run.id) && !await hasHistoricalSuspendedNativeSession(db, run) && + !await nativeRunRefusedBeforeLaunch(db, run) && !(run.runtimeMode === "native" && await recordUnidentifiedLocalRunnerStop(db, run, { record: !input.dryRun }))) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet."); if (run.processPid && !processStopped(run.processPid)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically."); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index d5f85b0794e..8acb6924cf7 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -2729,6 +2729,31 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }, ); if (budgetBlock) { + // A user's message refused by a temporary hold (accounting still settling) + // is saved as a deferred receipt, not skipped. It is promoted through + // normal admission once the run it waits behind ends or the hold clears. + if (!executionWaitRequestId && !caseRequest && !durableRequest && issueId && isUuidLike(issueId) && + opts.requestedByActorType === "user" && wakeCommentId && + !(await isBoardOwnedBudgetBlock(db, agent.companyId, budgetBlock))) { + const savedIssueId = issueId; + const condition = { reason: "budget.blocked", message: `${budgetBlock.reason} Your message is saved.`, + scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId }; + await db.transaction(async tx => { + await tx.execute(sql`select id from issues where id = ${savedIssueId} and company_id = ${agent.companyId} for update`); + await notifyDeferredExecutionWork(tx, { companyId: agent.companyId }); + await tx.insert(agentWakeupRequests).values({ + companyId: agent.companyId, agentId, source, triggerDetail, reason, + payload: withQueuedCommentIdsInWakePayload({ + ...payload, issueId: savedIssueId, [DEFERRED_WAKE_CONTEXT_KEY]: enrichedContextSnapshot, executionWait: condition, + }, [...new Set([...queuedCommentIdsFromRunContext(enrichedContextSnapshot), wakeCommentId])]), + status: "deferred_issue_execution", + requestedByActorType: opts.requestedByActorType ?? null, + requestedByActorId: opts.requestedByActorId ?? null, + idempotencyKey: opts.idempotencyKey ?? null, + }); + }); + throw new PersistedExecutionWaitRefusal(budgetBlock.reason, { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId }); + } const wait = await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId, }); diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 70731d14533..5999ea523b6 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -134,6 +134,21 @@ export async function recordUnidentifiedLocalRunnerStop(db: Db, run: Pick): Promise { + if (run.runtimeMode !== "native" || run.status !== "failed" || run.processPid || run.processGroupId || + run.errorCode !== "native_session_cleanup_quarantined") return false; + const [launch] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED]), + )).limit(1); + return !launch; +} + /** Pre-receipt native runs can retain an exact suspended session after their * mutable process fields were cleared. This is admission evidence for a new * user turn only, never permission to replay the old run or infer its outcomes. diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 19e6a761dd5..e3b0d16f6ab 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -3145,9 +3145,14 @@ export async function verifyStoppedNativeSessionForContinuation( !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(retained.status) || !retained.finishedAt || retained.finishedAt > run.createdAt || retained.controllerLeaseExpiresAt && retained.controllerLeaseExpiresAt > new Date() || - !retained.processStartedAt || !run.processStartedAt || - retained.processStartedAt.getTime() !== run.processStartedAt.getTime() || - retained.processPid !== run.processPid || retained.processGroupId !== run.processGroupId) return null; + // A run refused before launch (no launch event, checked above) and + // with no process record owns no process: only the previous owner's + // stop evidence below can certify the session. A source that does + // carry a process identity must still match the owner's exactly. + (run.processPid || run.processGroupId || run.processStartedAt) && + (!retained.processStartedAt || !run.processStartedAt || + retained.processStartedAt.getTime() !== run.processStartedAt.getTime() || + retained.processPid !== run.processPid || retained.processGroupId !== run.processGroupId)) return null; const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.issueId, run.nativeIssueId), eq(nativeRunFinalizations.runId, retained.id), diff --git a/server/src/services/quarantined-task-retry.test.ts b/server/src/services/quarantined-task-retry.test.ts index 8a02a628709..f08669a6e84 100644 --- a/server/src/services/quarantined-task-retry.test.ts +++ b/server/src/services/quarantined-task-retry.test.ts @@ -244,6 +244,31 @@ describe("task Retry after inherited native cleanup quarantine", () => { } }); + it.each(["stopped", "source_launch_requested", "owner_running"])( + "certifies a quarantined source that never launched from the previous owner's stop (%s)", async mode => { + // s7532/t04: the quarantine fence refused the source before it asked for + // a runner, so it has no process record at all. Only the owner can prove + // the retained session stopped. + const base = await mkdtemp(join(tmpdir(), "quarantined-never-launched-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = base; + try { + const f = await seed(base); + await db.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null }) + .where(eq(heartbeatRuns.id, f.sourceId)); + if (mode === "source_launch_requested") await appendHeartbeatRunEvent(db, { + companyId: f.companyId, runId: f.sourceId, agentId: f.agentId, eventType: PROCESS_START_REQUESTED }); + if (mode === "owner_running") await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.ownerId)); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceId)); + const proof = await verifyStoppedNativeSessionForContinuation(db, source); + if (mode === "stopped") expect(proof).not.toBeNull(); + else expect(proof).toBeNull(); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(base, { recursive: true, force: true }); + } + }); + it("holds a Retry while the task's own provider session close is still retained, then starts it", async () => { const base = await mkdtemp(join(tmpdir(), "retained-close-retry-")); const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;