From 1a7eb70a37bf455e1164ca64306d54d184f842cd Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:39:48 -0600 Subject: [PATCH] fix(run-dispatch): save a redirect sent while a stopped run settles (#19122) 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 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 e107a757556d3c74dcb5594b5eca9639cd3951e3. 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 Co-authored-by: Claude Opus 5.5 --- .../modules/run-dispatch/adapters/postgres.ts | 11 ++- .../src/services/workspace-recovery.test.ts | 79 +++++++++++++++++-- 2 files changed, 79 insertions(+), 11 deletions(-) diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index f9a33f7f9f2..8e487f1f0ea 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/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[])] : []; diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index a38c0881f22..921cb2c4355 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/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);