fix(heartbeat): save a comment refused by a temporary hold and route post-quarantine input to its own gate (#19829)

## 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 <noreply@anthropic.com>

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.

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");

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.");

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,
       });

server/src/services/native-local-process-stop.ts

@@ -134,6 +134,21 @@ export async function recordUnidentifiedLocalRunnerStop(db: Db, run: Pick<typeof
   return true;
 }
 
+/** A native run that an admission fence refused before it requested any
+ * runner launch (a quarantined prior-session cleanup is checked before the
+ * provider starts) never had a process: there is nothing to stop. Any launch
+ * request or identity record keeps the normal stop proofs. */
+export async function nativeRunRefusedBeforeLaunch(db: Db, run: Pick<typeof heartbeatRuns.$inferSelect,
+  "id" | "companyId" | "runtimeMode" | "status" | "errorCode" | "processPid" | "processGroupId">): Promise<boolean> {
+  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.

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),

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;