fix(run-dispatch): save a redirect sent while a stopped run settles (#19122)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A user can stop a running agent and then send a new message that
redirects the work
> - The stopped run keeps settling for some seconds after Stop (backup
and checkpoint persist)
> - A message sent in that window is queued as a normal run before
settlement records the workspace recovery hold
> - The dispatch gate then withholds the unstarted run, but it saved the
message only for sources with a native workspace-sync baseline
> - For other sources (remote workspace, snapshot-only) it marked the
wake `skipped`, so the message was lost and the task stayed in progress
with no run and no wake
> - This pull request saves the withheld message for any source when the
message was written after the Stop
> - The benefit is that a redirect after Stop is acted on exactly once
instead of disappearing

## Linked Issues or Issue Description

**What happened?**
After Stop, a user posted a redirect comment with resume intent. The
follow-up run was created while the stopped run was still persisting its
checkpoint. When settlement finished, the dispatch gate cancelled the
follow-up run with `execution_reconciliation_required` and an
empty-looking error ("Recovering saved work…"). The user's wake was
marked `skipped`. Nothing woke the task again. It stayed `in_progress`
with no active or queued run.

**Expected behavior**
The message sent after Stop runs exactly once after the stopped run
settles. The task never rests in progress with no run and no pending
wake.

**Steps to reproduce**
1. Start a native run whose stopped source has no native workspace-sync
baseline (for example a remote-workspace sandbox run).
2. Stop the run.
3. Post a comment with resume intent while the stopped run still
persists its checkpoint.
4. Let settlement record the workspace recovery hold.
5. Observe that the queued follow-up run is cancelled with
`execution_reconciliation_required` and its wake is `skipped`.

**Root cause**
`deferUnstartedUserComment` in
`server/src/modules/run-dispatch/adapters/postgres.ts` saved a withheld
user comment only when the stopped source had a native workspace-sync
baseline (the copyback replay case). The enqueue path saves or admits
the same message when it arrives after the hold exists. The gap was only
timing.

Refs #18332 (added the native-baseline-only deferral this PR widens).

**Paperclip version or commit**
Reproduced on `master` at e107a75755. The
touched code is unchanged on current `master`.

**Agent adapter(s) involved**
Not adapter-specific (core bug). Seen with the Paperclip runner adapter.

## What Changed

- `deferUnstartedUserComment` now saves a withheld user comment as a
`deferred_issue_execution` receipt for any stopped native source when
the latest saved comment was written after the source's Stop
(`finishedAt`).
- The native-baseline copyback path is unchanged.
- Input written before the Stop keeps the old behavior (`skipped`).
Older queued input cannot undo a Stop.
- Delivery uses the existing deferred-receipt path
(`resumeRemoteStopComments` / `resumeDeferredExecutionReceipts` →
`admitExplicitNativeContinuation` with `queuedCommentRequestId`). All
continuation gates still apply.
- New regression in `workspace-recovery.test.ts` that controls the
timing: the message is queued while the stopped run is still active,
then the hold is recorded, then the gate runs.
- The existing snapshot-only queued-reset tests asserted the dropped
message. They now assert that the message is saved and delivered once,
like the native-baseline case.

## Verification

- Red before the fix: the new test failed with the wake `status:
"skipped"`.
- Green after the fix:
- `cd server && npx vitest run src/services/workspace-recovery.test.ts`
(308 passed)
- `cd server && npx vitest run src/modules/run-dispatch
src/__tests__/heartbeat-undispatched-execution-wait.test.ts
src/__tests__/issue-queued-comments-routes.test.ts
src/services/explicit-native-continuation.test.ts
src/__tests__/heartbeat-process-recovery.test.ts
src/services/execution-continuation.test.ts`
- Safety cases in the new test:
- The redirect is saved, then delivered as exactly one successor run
with the original user's authority. The old provider run is not
executed.
- Input written before the Stop stays skipped and creates no successor.
- The UI Stop button uses the same `POST /heartbeat-runs/:id/cancel`
path. A comment with `interrupt: true` cancels through the same
`cancelRun` and then queues the comment wake. Both reach the same gate.
These were checked by code reading, not by a separate test.
- Not done: a live staging reproduction at the fixed revision.
Deployment is out of scope.

## Risks

- Behavior change: for sources without a native baseline, a post-Stop
message is now saved and delivered after recovery, instead of dropped.
Admission still blocks it until workspace repair is verified and the
previous execution has stopped. While blocked, the task shows "Your
message is saved."
- If admission never succeeds, the message stays saved and visible.
Before, it was silently lost.
- No schema or API change.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, in Claude Code with
tool use (shell, file edits, browser for read-only staging diagnostics).

## 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
- [ ] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Paperclip <noreply@paperclip.ing>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

server/src/modules/run-dispatch/adapters/postgres.ts

@@ -957,9 +957,12 @@ export function createPostgresRunDispatchAdapter(
       eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "cancelled"),
       sql`${heartbeatRuns.id}::text = ${preservation.evidence.runId}`,
     )) : [];
-    const baseline = readNativeWorkspaceSyncReference(source?.runnerProfileJson?.nativeWorkspaceSync);
-    if (!preservation || !source || !hasRequiredWorkspaceRecovery(source.resultJson) || !baseline || baseline.state === "finalized" ||
-        preservation.fingerprint !== `native-workspace:${source.id}`) return null;
+    if (!preservation || !source) return null;
+    const baseline = readNativeWorkspaceSyncReference(source.runnerProfileJson?.nativeWorkspaceSync);
+    // Copyback from a native baseline replays the saved turn. Without one, only
+    // a message sent after the Stop is saved; admission still applies all gates.
+    const copybackReplay = hasRequiredWorkspaceRecovery(source.resultJson) && !!baseline && baseline.state !== "finalized" &&
+      preservation.fingerprint === `native-workspace:${source.id}`;
     const context = parseObject(run.contextSnapshot), commentId = deriveCommentId(context);
     const commentIds = queuedCommentIdsFromRunContext(context);
     if (!commentId || !commentIds.includes(commentId) || !isUserCommentContinuationReason(readNonEmptyString(context.wakeReason))) return null;
@@ -981,6 +984,8 @@ export function createPostgresRunDispatchAdapter(
     if (comments.length !== commentIds.length || comments.some(comment =>
       !comment.body.trim() || !comment.authorUserId) ||
       comments.find(comment => comment.id === originalCommentId)?.authorUserId !== wake.requestedByActorId) return null;
+    const latest = comments.find(comment => comment.id === commentId);
+    if (!copybackReplay && (!source.finishedAt || !latest || latest.createdAt <= source.finishedAt)) return null;
     const sourceIds = context.queuedCommentSourceWakeIds;
     if (sourceIds !== undefined && (!Array.isArray(sourceIds) || sourceIds.some(id => typeof id !== "string" || !isUuidLike(id)))) return null;
     const sourceWakeIds = Array.isArray(sourceIds) ? [...new Set(sourceIds as string[])] : [];

server/src/services/workspace-recovery.test.ts

@@ -3721,13 +3721,9 @@ describe("automatic retained workspace recovery", () => {
       await Promise.all([control.resumeRemoteStopComments(await run(f.runId)),
         control.resumeRemoteStopComments(await run(f.runId))]);
       const [afterWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));
-      expect(heldWake.status).toBe(nativeDescriptor ? "deferred_issue_execution" : "skipped");
-      if (!nativeDescriptor) {
-        expect(afterWake.status).toBe("skipped");
-        expect(deps.enqueueWakeup).not.toHaveBeenCalled();
-        expect(queueDeps.executeRun).not.toHaveBeenCalled();
-        return;
-      }
+      // A post-Stop message is saved behind either source shape and delivered
+      // once through ordinary admission; it is never dropped by the gate.
+      expect(heldWake.status).toBe("deferred_issue_execution");
       expect(afterWake).toMatchObject({ status: "coalesced", runId: expect.any(String) });
       expect(afterWake.runId).not.toBe(queued.id);
       expect((await action(f.actionId)).evidence).toMatchObject({ automaticRecovery: { replay: "explicit_user_continuation" } });
@@ -3747,9 +3743,76 @@ describe("automatic retained workspace recovery", () => {
 
   it.each([
     { nativeDescriptor: true, title: "preserves an already queued reset until real native copyback replays its saved wake" },
-    { nativeDescriptor: false, title: "does not replay a snapshot-only source without a verified native baseline" },
+    { nativeDescriptor: false, title: "saves a queued reset behind a snapshot-only source and delivers it once after recovery" },
   ])("$title", async ({ nativeDescriptor }) => exerciseQueuedReset(nativeDescriptor), 30_000);
 
+  it.each([
+    { sentAfterStop: true, title: "saves and delivers once a redirect sent during Stop settlement when the stopped source has no native baseline" },
+    { sentAfterStop: false, title: "does not let input written before the Stop restart a source without a native baseline" },
+  ])("$title", async ({ sentAfterStop }) => {
+    // A message sent after Stop is queued while the stopped run is still
+    // persisting its checkpoint. Settlement then records a workspace hold, and
+    // the dispatch gate withholds the unstarted redirect. The gate must save it,
+    // not skip it: the task would otherwise stay in progress with no run or wake.
+    const t = await queuedResetFixture(false);
+    const { f, original, reset: redirect, wake, queued, deps, control, execution, queueDeps, queue } = t;
+    await db.update(issueComments).set({ body: "Actually, stop the essay. Instead give me a 5-bullet summary." })
+      .where(eq(issueComments.id, redirect.id));
+    // Older queued input is not a new direction and must not undo the Stop.
+    if (!sentAfterStop) await db.update(heartbeatRuns).set({ finishedAt: new Date(redirect.createdAt.getTime() + 1_000) })
+      .where(eq(heartbeatRuns.id, f.runId));
+    try {
+      expect(await queue.claimQueuedRun(queued)).toBeNull();
+      expect((await run(queued.id)).status).toBe("queued");
+      await recordRequiredWorkspaceRestoreFailure(db, await run(f.runId), { workspaceRestoreFailure: "restore_failed" }, f.lease);
+      const [hold] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId),
+        eq(issueRecoveryActions.sourceIssueId, f.issueId), eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`)));
+      f.actionId = hold.id; f.input.actionId = hold.id;
+      await recoverRetainedWorkspace(f.input);
+      await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", metadata: { ...f.lease.metadata,
+        remoteExecutionTermination: remoteTerminationReceipt(f.lease, { providerLeaseId: f.providerLeaseId, state: "stopped" }) } }).where(eq(environmentLeases.id, f.lease.id));
+      await control.acknowledgeRemoteStop(f.runId, f.companyId);
+      deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId);
+      await control.resumeRemoteStopComments(await run(f.runId));
+      await vi.waitFor(async () => {
+        await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async current => {
+          if (current.id === f.runId) await control.resumeRemoteStopComments(current);
+        }) ?? []);
+        expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete");
+      }, { timeout: 10_000, interval: 10 });
+
+      // The settled hold withholds the redirect that was queued before it existed.
+      expect(await queue.claimQueuedRun(await run(queued.id))).toBeNull();
+      expect(await run(queued.id)).toMatchObject({ status: "cancelled", errorCode: "execution_reconciliation_required", startedAt: null });
+      const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));
+      if (!sentAfterStop) {
+        expect(saved.status).toBe("skipped");
+        await control.resumeRemoteStopComments(await run(f.runId));
+        await queue.resumeDeferredExecutionReceipts({ companyId: f.companyId });
+        expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
+          sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${f.issueId}`, eq(heartbeatRuns.status, "queued")))).toHaveLength(0);
+        expect(deps.enqueueWakeup).not.toHaveBeenCalled();
+        expect(queueDeps.executeRun).not.toHaveBeenCalled();
+        return;
+      }
+      expect(saved).toMatchObject({ status: "deferred_issue_execution", runId: null,
+        payload: { executionWait: { reason: "execution_recovery", recoveryActionId: hold.id } } });
+      expect(saved.payload?._paperclipWakeContext).toMatchObject({ wakeCommentIds: [redirect.id] });
+
+      // Ordinary deferred-receipt delivery re-enters admission once.
+      await Promise.all([control.resumeRemoteStopComments(await run(f.runId)), control.resumeRemoteStopComments(await run(f.runId))]);
+      await queue.resumeDeferredExecutionReceipts({ companyId: f.companyId });
+      const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
+        sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${f.issueId}`, eq(heartbeatRuns.status, "queued")));
+      expect(successors).toHaveLength(1);
+      expect(successors[0].id).not.toBe(queued.id);
+      expect(successors[0].contextSnapshot?.wakeCommentIds).toEqual([redirect.id]);
+      expect(deps.enqueueWakeup).toHaveBeenCalledWith(original.agentId,
+        expect.objectContaining({ requestedByActorType: "user", requestedByActorId: "operator" }), wake.id);
+      expect(queueDeps.executeRun).not.toHaveBeenCalled();
+    } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); }
+  }, 30_000);
+
   it("reaches its queued snapshot-only reset after earlier recovery candidates without starting agent work", async () => {
     const previous = db;
     db = createDb(connectionString);