fix(recovery): continue a message sent during Stop settlement after recovery (#19355)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A user can stop or interrupt a running native task and then send a
message to give new direction
> - When the stopped run left a workspace hold, the dispatch gate
withholds the follow-up run until recovery saves the work
> - The gate must save the user's message on its receipt so that the
deferred-receipt worker continues it after recovery
> - Two conditions made the gate skip the message instead of saving it,
so the task stayed in progress with "Saved work recovered. Send a
message to continue." although the user already sent one
> - This pull request saves the message in both conditions
> - The benefit is that the message the user already sent continues
once, automatically, after recovery

## Linked Issues or Issue Description

**What happened?**
After a user stopped or interrupted a native task with an isolated
sandbox workspace and sent a message, the follow-up run was cancelled
with `execution_reconciliation_required`. Its error said "Saved work
recovered. Send a message to continue." Nothing re-woke the task. In the
interrupt case, a periodic sweep started a run about four minutes later.
In the case where the follow-up first waited for the stopped run's
workspace save (`workspace_busy`), each later retry parked on the same
resolved recovery action, and the task did not continue.

**Expected behavior**
When the recovery that the follow-up waited on resolves, the message
that the user already sent continues once, automatically. The product
does not ask for a second message.

**Steps to reproduce**
1. Run a native task in a managed sandbox with an isolated workspace.
2. Stop the run (or send a comment with interrupt), then send a message
while the stop and workspace save are still settling.
3. Wait for the workspace recovery to resolve.
4. See that the follow-up run is cancelled with
`execution_reconciliation_required` and that no run continues the
message.

**Paperclip version or commit**
`master` before this change.

**Root cause**
`deferUnstartedUserComment` (run dispatch) refused to save the withheld
message in two cases:
1. It compared the message time with the source run's `finishedAt`. A
native run finishes only after its provider stop and workspace save
settle, which takes seconds. A message written after the Stop request
but before that settlement was treated as input written before the Stop.
The receipt was marked `skipped`.
2. It required the run's own receipt to be user-owned. A workspace-busy
retry or an admission retry carries the user's message under a
system-owned receipt, so the message was dropped with the retry.

## What Changed

- Add `userStopBoundaryAt(run)` in `run-cancellation.ts`. For a run
cancelled by a user Stop or interrupt, it returns the recorded Stop
request time (`cancellation.recordedAt`) when that is earlier than
`finishedAt`. For all other runs it returns `finishedAt`.
- Use this boundary in the dispatch gate's message deferral, in native
explicit-continuation admission, and in the remote-Stop continuation
check, so the three places agree on "message written after the Stop".
- When the withheld run is a system-owned retry, walk a bounded chain
(max 8) of retries that did not reach the provider (`workspace_busy`
with `providerWorkStarted: false`, or a verified pre-dispatch
`execution_reconciliation_required`) to the originating user receipt.
Save the message on that receipt with the user's original wake context.
Mark the retry's system receipt `skipped`.
- Update the "input written before the Stop" test so that the Stop
request, and not only the settlement, comes after the message.

## Verification

- New regressions in `server/src/services/workspace-recovery.test.ts`
(`continues a message ... without a second message, after recovery
already resolved`): a direct follow-up and a workspace-busy retry. Each
runs recovery to completion first, then lets the gate withhold the
follow-up, then runs only the deferred-receipt worker. Each asserts that
exactly one successor run is queued with the user's message and the
user's authority, and that no provider work starts.
- Red before the fix: both tests fail because the receipt is `skipped`
instead of `deferred_issue_execution`. Green after the fix.
- `npx vitest run src/services/workspace-recovery.test.ts` (384 passed).
- `npx vitest run src/services/explicit-native-continuation.test.ts
src/modules/run-dispatch src/services/execution-continuation.test.ts
src/services/run-cancellation.test.ts` (581 passed).
- 19 more heartbeat, retry, queued-comment, and native recovery suites
(1272 passed). One failure, `heartbeat-process-recovery > redacts opaque
environment-bound credentials from Sentry diagnostics: result`, also
fails on unchanged `master` and is not related.

## Risks

- A message written after the Stop request but before the stopped run
settled now continues. The stopping run cannot consume new input after
its Stop request, and admission still removes comments that a run
already delivered (`undeliveredLegacyUserCommentIds`). All admission
gates (holds, stop proofs, approvals, budget) still apply.
- Only user Stops and interrupts move the boundary. Other cancellations
keep `finishedAt`.
- The retry walk is bounded, stays in one company, agent, and issue, and
accepts only retries without provider work. It revives a user receipt
only from `cancelled` or `skipped`, so a second retry cannot save the
same receipt twice.
- Not in scope: a run that fails during native preparation with no user
message (for example `runner_state_identity_mismatch:
prior_owner_terminal_state_indeterminate;
retirement_refused=unexpected_root_entry` on a system wake). After its
recovery resolves, the task still waits for a user message. That is a
separate root cause in the native session executor's prior-owner
retirement proof.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use (shell, file edits, local test runs).

## 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
- [x] All Paperclip CI gates are green
- [x] 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/modules/run-dispatch/adapters/postgres.ts

@@ -4,7 +4,7 @@ import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restor
 import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js";
 import { isPreDispatchExecutionWaitVerified } from "../../../services/pre-dispatch-execution-wait.js";
 import { isUserCommentContinuationReason } from "../../../services/explicit-native-continuation.js";
-import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js";
+import { queuedCommentIdsFromRunContext, withQueuedCommentIdsInRunContext, withQueuedCommentIdsInWakePayload } from "../../../services/issue-queued-comment-queue.js";
 import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js";
@@ -12,6 +12,7 @@ import { notifyInstructionCollection } from "../../../services/instruction-colle
 import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js";
 import { readClaimedNativeStopRecovery } from "../../../services/native-runtime/native-cancellation-request.js";
 import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js";
+import { userStopBoundaryAt } from "../../../services/run-cancellation.js";
 import { getExecutionBlocker } from "../../../services/execution-blocker.js";
 import { validateReconstructedWorkspaceDispatch } from "../../../services/native-runtime/native-workspace-reconstruction.js";
 import { getNativeReviewAssignment } from "../../../services/native-runtime/native-review-participant.js";
@@ -67,6 +68,7 @@ import type {
 } from "../domain/policy.js";
 import {
   MAX_TURN_CONTINUATION_RETRY_REASON,
+  WORKSPACE_BUSY_RETRY_REASON,
   allowsIssueInteractionWake,
   deriveCommentId,
   isNonAssigneeWorkspaceBusyRetry,
@@ -949,6 +951,42 @@ export function createPostgresRunDispatchAdapter(
     return transactionResult.outcome;
   }
 
+  /**
+   * A system retry of an unstarted user turn (a workspace wait or an
+   * admission retry) still carries that user's message, but its own receipt
+   * is system-owned. Find the originating user receipt through a bounded chain
+   * of retries that never reached the provider, so the message can be saved
+   * under the user's authority instead of being dropped with the retry.
+   */
+  async function unstartedUserTurnOrigin(tx: Db, run: HeartbeatRun, issueId: string) {
+    let current = run;
+    // Comments can coalesce into any retry in the chain while it waits.
+    const chainRunIds: string[] = [], chainWakeIds: string[] = [];
+    for (let hop = 0; hop < 8 && current.retryOfRunId; hop++) {
+      const [parent] = await tx.select().from(heartbeatRuns).where(and(
+        eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId),
+        eq(heartbeatRuns.id, current.retryOfRunId), eq(heartbeatRuns.status, "cancelled"),
+        sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
+      )).for("update");
+      if (!parent?.wakeupRequestId) return null;
+      const recovery = parseObject(parent.resultJson?.executionRecovery);
+      const unstarted = parent.errorCode === WORKSPACE_BUSY_RETRY_REASON
+        ? recovery.kind === "workspace_wait" && recovery.providerWorkStarted === false
+        : parent.errorCode === "execution_reconciliation_required" && await isPreDispatchExecutionWaitVerified(tx, parent);
+      if (!unstarted) return null;
+      const [wake] = await tx.select().from(agentWakeupRequests).where(and(
+        eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
+        eq(agentWakeupRequests.id, parent.wakeupRequestId), eq(agentWakeupRequests.runId, parent.id),
+      )).for("update");
+      if (!wake) return null;
+      chainRunIds.push(parent.id); chainWakeIds.push(wake.id);
+      if (wake.requestedByActorType === "user")
+        return ["cancelled", "skipped"].includes(wake.status) ? { run: parent, wake, chainRunIds, chainWakeIds } : null;
+      current = parent;
+    }
+    return null;
+  }
+
   async function deferUnstartedUserComment(tx: Db, run: HeartbeatRun, issueId: string, now: Date) {
     if (!run.wakeupRequestId || !await isPreDispatchExecutionWaitVerified(tx, run)) return null;
     const wait = parseObject(run.resultJson?.executionWait);
@@ -967,18 +1005,27 @@ export function createPostgresRunDispatchAdapter(
     // 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;
-    const [wake] = await tx.select().from(agentWakeupRequests).where(and(
+    const runContext = parseObject(run.contextSnapshot), commentId = deriveCommentId(runContext);
+    const commentIds = queuedCommentIdsFromRunContext(runContext);
+    if (!commentId || !commentIds.includes(commentId)) return null;
+    const [ownWake] = await tx.select().from(agentWakeupRequests).where(and(
       eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
       eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id),
       inArray(agentWakeupRequests.status, ["queued", "claimed"]),
-      eq(agentWakeupRequests.requestedByActorType, "user"),
     )).for("update");
-    const payload = parseObject(wake?.payload);
+    if (!ownWake) return null;
+    const origin = ownWake.requestedByActorType === "system" ? await unstartedUserTurnOrigin(tx, run, issueId) : null;
+    if (ownWake.requestedByActorType !== "user" && !origin) return null;
+    const wake = origin?.wake ?? ownWake;
+    const receiptRunId = origin?.run.id ?? run.id;
+    const runIds = [run.id, ...(origin?.chainRunIds ?? [])];
+    // A retry's context names the retry reason. Save the user's original wake
+    // context together with every message the retry carried.
+    const context = origin ? withQueuedCommentIdsInRunContext(parseObject(origin.run.contextSnapshot), commentIds) : runContext;
+    if (!isUserCommentContinuationReason(readNonEmptyString(context.wakeReason))) return null;
+    const payload = parseObject(wake.payload);
     const originalCommentId = readNonEmptyString(payload.commentId);
-    if (!wake?.requestedByActorId || payload.issueId !== issueId ||
+    if (!wake.requestedByActorId || payload.issueId !== issueId ||
         !originalCommentId || !commentIds.includes(originalCommentId)) return null;
     const comments = await tx.select().from(issueComments).where(and(
       eq(issueComments.companyId, run.companyId), eq(issueComments.issueId, issueId),
@@ -989,24 +1036,25 @@ export function createPostgresRunDispatchAdapter(
       !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;
+    const stoppedAt = userStopBoundaryAt(source);
+    if (!copybackReplay && (!stoppedAt || !latest || latest.createdAt <= stoppedAt)) return null;
+    const sourceIds = runContext.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[])] : [];
     const adopted = sourceWakeIds.length ? await tx.select().from(agentWakeupRequests).where(and(
       eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
-      eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "coalesced"),
+      inArray(agentWakeupRequests.runId, runIds), eq(agentWakeupRequests.status, "coalesced"),
       inArray(agentWakeupRequests.id, sourceWakeIds),
       sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
     )).for("update") : [];
     if (adopted.length !== sourceWakeIds.length) return null;
-    const mergeTargets = [wake.id, ...adopted.map(receipt => receipt.id)];
+    const mergeTargets = [...new Set([wake.id, ownWake.id, ...(origin?.chainWakeIds ?? []), ...adopted.map(receipt => receipt.id)])];
     const coalesced = await tx.select().from(agentWakeupRequests).where(and(
       eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
       eq(agentWakeupRequests.status, "coalesced"), eq(agentWakeupRequests.requestedByActorType, "user"),
       sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
-      or(and(eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.reason, "issue_execution_same_name")),
-        and(or(isNull(agentWakeupRequests.runId), eq(agentWakeupRequests.runId, run.id)),
+      or(and(inArray(agentWakeupRequests.runId, runIds), eq(agentWakeupRequests.reason, "issue_execution_same_name")),
+        and(or(isNull(agentWakeupRequests.runId), inArray(agentWakeupRequests.runId, runIds)),
           inArray(sql<string>`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId'`, mergeTargets))),
     )).for("update");
     const authors = new Map(comments.map(comment => [comment.id, comment.authorUserId]));
@@ -1024,8 +1072,9 @@ export function createPostgresRunDispatchAdapter(
         executionWait: { ...parseObject(run.resultJson?.executionWait), reason: "execution_recovery",
           message: "Waiting for execution recovery. Your message is saved." } }, commentIds), updatedAt: now,
     }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId),
-      eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id),
-      inArray(agentWakeupRequests.status, ["queued", "claimed"]))).returning({ id: agentWakeupRequests.id });
+      eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, receiptRunId),
+      inArray(agentWakeupRequests.status, origin ? ["cancelled", "skipped"] : ["queued", "claimed"])))
+      .returning({ id: agentWakeupRequests.id });
     return deferred?.id ?? null;
   }
 
@@ -1083,7 +1132,9 @@ export function createPostgresRunDispatchAdapter(
 
       const deferredWakeRequestId = expectedStatus === "queued" && decision.errorCode === "execution_reconciliation_required"
         ? await deferUnstartedUserComment(tx, row, issueId, now) : null;
-      if (row.wakeupRequestId && !deferredWakeRequestId) {
+      // A retry's system receipt still settles when its user's message was
+      // saved on the originating user receipt.
+      if (row.wakeupRequestId && deferredWakeRequestId !== row.wakeupRequestId) {
         await tx
           .update(agentWakeupRequests)
           .set({ status: "skipped", finishedAt: now, error: decision.reason, updatedAt: now })

server/src/services/explicit-native-continuation.ts

@@ -12,7 +12,7 @@ import { readNativeWorkspaceSyncReference } from "./native-runtime/native-worksp
 import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js";
 import { createHash } from "node:crypto";
 import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
-import { canContinueCancelledRun } from "./run-cancellation.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 } from "./native-local-process-stop.js";
@@ -104,6 +104,11 @@ export async function undeliveredLegacyUserCommentIds(
     // native identity/process metadata is not proof that its prompt was sent.
     // Admission separately verifies process termination before a new turn.
     sql`not (${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.startedAt} is null)`,
+    // A claimed turn parked for a busy workspace stopped before provider
+    // dispatch; the host records that it sent no prompt.
+    sql`not coalesce(${heartbeatRuns.status} = 'cancelled' and ${heartbeatRuns.errorCode} = 'workspace_busy'
+      and ${heartbeatRuns.resultJson}#>>'{executionRecovery,kind}' = 'workspace_wait'
+      and ${heartbeatRuns.resultJson}#>'{executionRecovery,providerWorkStarted}' = 'false'::jsonb, false)`,
     or(...commentIds.map(id => or(
       sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${id}`,
       sql`${heartbeatRuns.contextSnapshot}->'wakeCommentIds' @> ${JSON.stringify([id])}::jsonb`,
@@ -295,7 +300,7 @@ export async function admitExplicitNativeContinuation(input: {
     if (!run || run.agentId !== agentId || !terminal.includes(run.status) ||
         (run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId) !== issueId ||
         !run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved.");
-    if (!queuedInterrupt && !queuedRequest && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
+    if (!queuedInterrupt && !queuedRequest && authorizedAt <= userStopBoundaryAt(run)!) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
     if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically.");
     const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
       eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
@@ -306,6 +311,9 @@ export async function admitExplicitNativeContinuation(input: {
     if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId ||
         lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null;
     run = lockedRun;
+    // A user's message written after their Stop request is new direction even
+    // when the stopped native run settled later.
+    const stoppedAt = userStopBoundaryAt(run)!;
     const failedPreparation = await isFailedNativePreparation(db, run, coordinator);
     const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
       run.errorCode === "execution_reconciliation_required" &&
@@ -378,14 +386,14 @@ export async function admitExplicitNativeContinuation(input: {
     if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission && !cancelledStartup && !failedPreparation) return null;
     if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission &&
         !(queuedRequest && run.runtimeMode === "native" &&
-          (run.status !== "cancelled" || authorizedAt > run.finishedAt!)) &&
+          (run.status !== "cancelled" || authorizedAt > stoppedAt)) &&
         !(queuedRequest && cancelledStartup && authorizedAt > run.finishedAt!) &&
         !(nativeQueuedSend && (run.runtimeMode === "native" || cancelledStartup) && authorizedAt > run.finishedAt!) &&
         !(queuedInterrupt && response?.source.requiresFreshSession && run.runtimeMode === "native")) return null;
     // A saved native message sent after Stop is a new user direction. Older
     // input cannot restart a cancelled run merely because it is still queued.
     if (queuedRequest && !queuedInterrupt && run.status === "cancelled" && !unusedAdmission &&
-        !canContinueCancelledRun(run) && !(run.runtimeMode === "native" && authorizedAt > run.finishedAt!) &&
+        !canContinueCancelledRun(run) && !(run.runtimeMode === "native" && authorizedAt > stoppedAt) &&
         !(cancelledStartup && authorizedAt > run.finishedAt!)) return null;
     if (retry && run.status === "cancelled" && !canContinueCancelledRun(run) && !cancelledStartup)
       return blocked("cancelled_by_operator", "Inspect the stopped run and send a new message to continue.");

server/src/services/heartbeat/run-control.ts

@@ -34,6 +34,7 @@ import {
   canContinueCancelledRun,
   readRunCancellation,
   requestedRunCancellation,
+  userStopBoundaryAt,
 } from "../run-cancellation.js";
 import { settleSlackConversation } from "../slack-conversation-lifecycle.js";
 import { readQueuedInteractionResponse } from "../queued-interaction-response.js";
@@ -665,7 +666,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
           sql`${issueComments.id}::text = ${commentId}`, eq(issueComments.authorType, "user"),
           eq(issueComments.authorUserId, requestedByActorId), isNull(issueComments.deletedAt),
           isNull(issueComments.createdByRunId),
-          gt(issueComments.createdAt, run.finishedAt),
+          gt(issueComments.createdAt, userStopBoundaryAt(run)!),
         ));
         if (!comment?.body.trim()) continue;
       } else {

server/src/services/run-cancellation.ts

@@ -135,6 +135,21 @@ export function canContinueCancelledRun(run: Run): boolean {
     run.resultJson?.acpToolInventoryComplete === true && run.resultJson?.acpPendingToolCount === 0;
 }
 
+/** When a user's Stop or interrupt took effect for message ordering. A native
+ * run finishes only after its provider stop and workspace save settle, which
+ * can take many seconds. A message written after the Stop request is a new
+ * direction even when that settlement finished later, so compare against the
+ * recorded request rather than `finishedAt`. Other cancellations keep
+ * `finishedAt`. */
+export function userStopBoundaryAt(run: Pick<Run, "status" | "finishedAt" | "resultJson">): Date | null {
+  if (!run.finishedAt) return null;
+  const cancellation = run.status === "cancelled" ? readRunCancellation(run.resultJson) : null;
+  if (!cancellation || cancellation.initiator.type !== "user" ||
+      !["operator", "queued_message"].includes(cancellation.source)) return run.finishedAt;
+  const requestedAt = new Date(cancellation.recordedAt);
+  return Number.isFinite(requestedAt.getTime()) && requestedAt < run.finishedAt ? requestedAt : run.finishedAt;
+}
+
 export function isUnexpectedRunCancellation(run: Run): boolean {
   const bootstrap = run.resultJson?.executionRecovery as Record<string, unknown> | undefined;
   return run.status === "cancelled" && Boolean(run.startedAt) &&

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

@@ -4747,9 +4747,14 @@ describe("automatic retained workspace recovery", () => {
     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));
+    // Older queued input is not a new direction and must not undo the Stop:
+    // the user requested Stop only after writing it.
+    if (!sentAfterStop) {
+      const stoppedAt = new Date(redirect.createdAt.getTime() + 1_000), stopped = await run(f.runId);
+      await db.update(heartbeatRuns).set({ finishedAt: stoppedAt, resultJson: { ...stopped.resultJson,
+        cancellation: { ...(stopped.resultJson?.cancellation as Record<string, unknown>), recordedAt: stoppedAt.toISOString() } } })
+        .where(eq(heartbeatRuns.id, f.runId));
+    }
     try {
       expect(await queue.claimQueuedRun(queued)).toBeNull();
       expect((await run(queued.id)).status).toBe("queued");
@@ -4802,6 +4807,108 @@ describe("automatic retained workspace recovery", () => {
     } finally { deps.activeRunExecutions.delete(f.runId); execution.finish(); adapterExecutionControls.delete(f.runId); }
   }, 30_000);
 
+  it.each([
+    { retried: false, title: "continues a message written during Stop settlement once, without a second message, after recovery already resolved" },
+    { retried: true, title: "continues a message carried by a workspace-busy retry once, without a second message, after recovery already resolved" },
+  ])("$title", async ({ retried }) => {
+    // S47: the user stops (or interrupts) a native run and sends a message.
+    // The native source finishes only after its stop and workspace save
+    // settle, so its finishedAt is later than the message. Recovery resolves
+    // first, then the dispatch gate withholds the unstarted follow-up behind
+    // the recovered hold ("Saved work recovered. Send a message to continue.").
+    // The message the user already sent must be saved and continued by the
+    // ordinary deferred-receipt worker, not dropped.
+    const t = await queuedResetFixture(false);
+    const { f, original, reset: message, wake, queued, deps, control, execution, queueDeps, queue } = t;
+    queueDeps.resumeExecutionWaitComments.mockImplementation(control.resumeExecutionWaitComments);
+    queueDeps.releaseIssueExecutionAndPromote.mockImplementation(control.releaseIssueExecutionAndPromote);
+    await db.update(issueComments).set({ body: "Actually, give me a 5-bullet summary instead." }).where(eq(issueComments.id, message.id));
+    // The Stop was requested before the message (fixture); settlement finished after it.
+    await db.update(heartbeatRuns).set({ finishedAt: new Date(message.createdAt.getTime() + 1_000) })
+      .where(eq(heartbeatRuns.id, f.runId));
+    let pending = queued;
+    let systemWakeId: string | null = null;
+    let expectedIds = [message.id];
+    if (retried) {
+      // The first follow-up was claimed, met the stopped run's unsettled
+      // workspace save, and was parked as a system-owned workspace-busy retry
+      // carrying the message. While that retry waited, the user added a second
+      // message, which coalesced into it; a further retry carried both.
+      const busy = () => ({ status: "cancelled" as const, startedAt: new Date(), finishedAt: new Date(), errorCode: "workspace_busy",
+        resultJson: { executionRecovery: { kind: "workspace_wait", providerWorkStarted: false },
+          cancellation: { source: "control_plane", expected: true, initiator: { type: "system" }, reason: "Waiting for the shared execution workspace", recordedAt: new Date().toISOString() } } });
+      await db.update(heartbeatRuns).set(busy()).where(eq(heartbeatRuns.id, queued.id));
+      await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, wake.id));
+      const retry = async (parentId: string, context: Record<string, unknown>, status: "queued" | "cancelled") => {
+        const [systemWake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId,
+          source: "automation", triggerDetail: "system", reason: "workspace_busy_retry", status: status === "queued" ? "claimed" : "cancelled",
+          claimedAt: new Date(), requestedByActorType: "system", payload: { issueId: f.issueId, retryOfRunId: parentId } }).returning();
+        const [retryRun] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId,
+          invocationSource: "automation", triggerDetail: "system", status: "queued", runtimeMode: "legacy", responsibleUserId: "operator",
+          wakeupRequestId: systemWake.id, retryOfRunId: parentId, scheduledRetryReason: "workspace_busy", scheduledRetryAttempt: 1,
+          contextSnapshot: { ...context, wakeReason: "workspace_busy_retry", retryOfRunId: parentId, retryReason: "workspace_busy" } }).returning();
+        await db.update(agentWakeupRequests).set({ runId: retryRun.id }).where(eq(agentWakeupRequests.id, systemWake.id));
+        return { retryRun, systemWake };
+      };
+      const first = await retry(queued.id, queued.contextSnapshot ?? {}, "cancelled");
+      const [second] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId,
+        authorType: "user", authorUserId: "operator", body: "Also keep it under 100 words." }).returning();
+      await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId,
+        source: "on_demand", triggerDetail: "manual", reason: "issue_commented", status: "coalesced", runId: first.retryRun.id,
+        requestedByActorType: "user", requestedByActorId: "operator",
+        payload: { issueId: f.issueId, commentId: second.id, mutation: "comment", coalescedIntoWakeupRequestId: first.systemWake.id } });
+      await db.update(heartbeatRuns).set(busy()).where(eq(heartbeatRuns.id, first.retryRun.id));
+      expectedIds = [message.id, second.id];
+      const carried = { ...queued.contextSnapshot, wakeCommentId: second.id, wakeCommentIds: expectedIds };
+      const last = await retry(first.retryRun.id, carried, "queued");
+      pending = last.retryRun;
+      await db.update(issues).set({ executionRunId: pending.id }).where(eq(issues.id, f.issueId));
+      systemWakeId = last.systemWake.id;
+    }
+    try {
+      expect(await queue.claimQueuedRun(pending)).toBeNull();
+      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);
+      // Recovery resolves and its completion hooks run before the follow-up is withheld.
+      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 });
+      expect(await queue.claimQueuedRun(await run(pending.id))).toBeNull();
+      expect(await run(pending.id)).toMatchObject({ status: "cancelled", errorCode: "execution_reconciliation_required", startedAt: null });
+      const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));
+      expect(saved).toMatchObject({ status: "deferred_issue_execution", runId: null,
+        payload: { executionWait: { reason: "execution_recovery", recoveryActionId: hold.id } } });
+      expect(saved.payload?._paperclipWakeContext).toMatchObject({ wakeReason: "issue_commented", wakeCommentIds: expectedIds });
+      if (systemWakeId) {
+        const [systemWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, systemWakeId));
+        expect(systemWake.status).toBe("skipped");
+      }
+
+      // Only the deferred-receipt worker runs (its 30-second wait has elapsed).
+      await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 31_000) }).where(eq(agentWakeupRequests.id, wake.id));
+      await queue.resumeDeferredExecutionReceipts({ companyId: f.companyId, requestId: wake.id });
+      await queue.resumeDeferredExecutionReceipts({ companyId: f.companyId, requestId: wake.id });
+      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].contextSnapshot?.wakeCommentIds).toEqual(expectedIds);
+      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);