fix(retry): wait for admitted source cleanup before dispatch (#18898)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Connected-task Retry keeps the original request and its current authority. > - One durable Retry can already own a queued successor. > - The old native environment can still have pending cleanup before that successor starts. > - The task execution lock points to the successor, so the old cleanup can escape the queue claim check. > - Dispatch then refuses the source state and fails the new run before provider work. > - This pull request keeps that same run queued and wakes dispatch when cleanup changes. ## Linked Issues or Issue Description Refs #18845, #18853, #18860, #18877, #18882 and #18932. These repairs cover preservation binding, interrupted preparation, old native ownership, manual native holds or other setup guards. This change covers an already admitted connected-task Retry whose native predecessor still owes cleanup. It uses the shared cleanup predicate so the manual-hold repair remains with its existing owner. Search found no duplicate queued-source cleanup repair. **What happened?** A board Retry saved its exact chat request and admitted one queued successor. The source lease was still pending cleanup when that successor reached dispatch. A retained stop receipt did not settle the full cleanup contract. The queue checked the task's current execution owner, which was already the successor. The native authorization guard then rejected the predecessor state and failed the new run before provider dispatch. **Expected behavior** Keep the same canonical Retry and queued run while the exact source cleanup is pending. After authoritative settlement, submit that run through normal dispatch once. Do not infer safe execution from a stop-and-retain receipt alone. **Steps to reproduce** 1. Retry an authorized connected request from a failed native run. 2. Admit one successor while capacity keeps it queued. 3. Leave or restore the source lease to pending cleanup with a matching retained stop receipt. 4. Release capacity and resume the ordinary queue. 5. The old claim path starts the successor and the native source guard refuses it. **Paperclip version or commit** Reproduced against master at `f77499283270d2e966aabd58493f9d35c6bbdf5f`. **Deployment mode** The regression uses the real HTTP route, chat authorization, queue claims, disposable PostgreSQL and the production work coordinator on macOS arm64. A controlled adapter supplies the final result. No hosted provider replay or deployment is claimed. ## What Changed - Validate the exact canonical Retry, chat action, company, task, agent, source and successor before testing source cleanup. - Keep the successor queued under the existing issue claim lock while the shared cleanup predicate remains pending. - Wake only the scoped queued-dispatch worker after cleanup or exact workspace-owner release commits. Add no timer or background scan. - Verify pending cleanup at intake, later cleanup invalidation, one final execution, idempotency, Stop, changed input, unknown stop receipts, a different live controller and signed sandbox absence followed by owner release. - Document the queued-wait contract. No schema, endpoint, credential, model or provider policy changes are required. - Align five native preparation acceptance fixtures with the existing workspace publication contract. Use the real reservation and publication owner. Retain a raw-root refusal case with no provider launch and preserved files. ## Verification - Latest master sync at `e26cad76b8f20ba823d702ed60e7b7294f8f5a7c` includes master `396b7fc62a0a4b1d8aadbda8221c05879d54177c`. The shared test helper preserves #18944's canonical retained-directory default, the real publisher for the five fresh-workspace cases, and the explicit raw-root refusal. Independent remerge review found no dropped upstream behavior, guard or assertion. - The latest head passes the focused six cases, all 580 complete workspace/continuation tests, all 38 selected HTTP Retry cases, fresh full recursive Node 24.22.2 types and full build. No affected tests skipped. An initial cached test-layer attempt read old budget declarations; regenerating them resolved all seven errors without test/mock or production-export changes. - All current-head CI is green at `e26cad76b8f20ba823d702ed60e7b7294f8f5a7c`, including required [ci / verify](https://github.com/paperclipai/paperclip/actions/runs/41846304017/job/126636751314) and [ci / e2e](https://github.com/paperclipai/paperclip/actions/runs/41846304017/job/126636638086). Greptile and Paperclip Review both rate this exact head 5/5, with no findings or open review threads. - Master sync at `3e4f266c2da172a84afd0b9cb4a91ea3b95f3c0d` merges master `b891e69bd23d6dab8ce536569e15bc4a7a31c35c`. The only conflicts were adjacent queue imports and two HTTP Retry test matrices. Both imports and both complete matrices remain unchanged. Independent review found no relaxed guard or assertion. - Node 24.22.2 and pnpm 9.15.4: all 38 selected real HTTP Retry cases, 194 queue/cleanup/finalizer cases, module boundaries, recursive types and build pass at `3e4f266c2da172a84afd0b9cb4a91ea3b95f3c0d`. - That sync head's [server check](https://github.com/paperclipai/paperclip/actions/runs/41838741852/job/126604742761) failed five native acceptance cases in CI and locally with `remote_workspace_published_path_invalid`. The test fixture supplied a raw directory instead of a fresh publisher-generated workspace. The guard from #18932 remains unchanged. The same five cases passed earlier Linux CI before that guard was present. - The test-only alignment at `dcdee1bb6c2e2e6dd494382499d17c23a3ce71a8` obtains the saved authority, path and run reference from real `reserveRemoteWorkspace` and `prepareRemoteWorkspace`. All five original cases and a raw-root no-launch/no-result/files-preserved control pass. Their native result, file, catalog and single-execution assertions remain intact. An initial setup attempt was refused for a noncanonical allocation root; the fixture now resolves that owned root before publication. - At previous fixture head `dcdee1bb6c2e2e6dd494382499d17c23a3ce71a8`, the generated server test-layer typecheck passed. All 580 tests in the complete workspace/continuation family pass, with no skips. That test-only commit left production files byte-identical to the sync head. - Previous published-head CI was green at `dcdee1bb6c2e2e6dd494382499d17c23a3ce71a8`, including required `ci / verify` and `ci / e2e`. The [Linux Node 24.22.2 server shard](https://github.com/paperclipai/paperclip/actions/runs/41842665765/job/126618334638) executed all five original acceptance cases and the new raw-root refusal case successfully. Its whole workspace recovery file passed all 245 tests. - Greptile reviewed previous published head `dcdee1bb6c2e2e6dd494382499d17c23a3ce71a8` at 5/5, with no findings or open review threads. No unchanged-job rerun was used for the reproduced sync-head failure. - Causal red: the real HTTP regression started and failed the successor at the native source-state guard while cleanup remained pending. A second red proved that signed sandbox absence followed by copyback-owner release lacked a queued-dispatch wake. - Green: seven HTTP/PostgreSQL cleanup cases, the exact confirmed-owner release notification control and seven existing retry lineage controls pass. The real coordinator resumes from committed cleanup or workspace-owner release, without a manual queue restart. One controlled adapter execution produces the result after settlement. - Earlier branch verification passed full repository `pnpm -r typecheck`, `pnpm build`, module boundaries and `git diff --check`. - The full stale-queue file control passed all 64 cases with unchanged observer/hook limits. The initial broader affected run had 192 passes and one non-assignee native-owner case miss its existing five-second adapter observer while already running with no error. The isolated case passed; the unchanged base passed all 64 cases but hit its existing 30-second teardown timeout. Those original results are retained. The new queue hint has no dispatch consumer in that fixture. - Earlier-head CI was green: 54 success, three skipped and two neutral checks at `6adb9f1ba682282c512d5e751ba2aa663f614ca2`. One chat job was cancelled by an external runner shutdown; its single unchanged-job rerun passed. No source was changed to obtain that result. - Greptile and Paperclip Review both rated earlier head `6adb9f1ba682282c512d5e751ba2aa663f614ca2` 5/5 with no findings or unresolved threads. These results do not qualify the master-sync head. - At initial verification, a complete local `pnpm test:run` was not repeated while the related setup repair's full local run was active. The full earlier-head CI suites passed. A full local suite has not been repeated for the mechanical master sync. Local verification uses the affected HTTP, producer, lineage and queue suites. No full local-suite pass is claimed. - The fresh clone's missing plugin SDK links were repaired with the repository's own postinstall helper before verification. The embedded PostgreSQL package's own symlink helper was also required. - No staging task was submitted again with this repair, and no deployment is claimed. ## Risks - A pending source cleanup can keep the queued run waiting. Its cleanup owner still owns reconciliation and its existing limits. - Stop, changed source, unknown termination and live ownership retain their existing authority checks. A cleanup notification grants no execution authority. - A new cleanup change after claim still faces the existing final dispatch guard. - Cleanup notifications can change dispatch timing in a running controller. The real coordinator cases verify commit-bound wakeups; the broader current queue-file control passed all 64 cases. The earlier loaded-run and baseline teardown failures are retained in verification. ## Model Used OpenAI GPT-6 through Codex, with reasoning, code editing, tool use, PostgreSQL and browser verification. This session does not expose a more specific deployment model ID or context-window size. ## 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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-11-08 12:33:05 +01:00 · 2026-11-04 08:33:39 -06:00
1 parent 5d226ca08a
commit 84e9c8f364
@@ -65177,6 +65177,161 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
      } finally { await f.service.shutdown(); }
    },
  );
+  it.each(["settled", "pending_at_intake", "source_edited", "operator_stopped", "unknown_stop", "live_owner", "absent_sandbox"] as const)(
+    "keeps the admitted chat Retry queued until its source lease cleanup settles: %s", async mode => {
+    const f = await currentGitHubTaskRetryFixture();
+    const companyId = f.fixture.companyId, issueId = f.issue.id;
+    const heartbeat = heartbeatService(db);
+    const errors: unknown[] = [];
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let releaseWorkspaceOwner: () => void = () => undefined;
+    let workspaceOwner: Promise<unknown> | undefined;
+    const app = express();
+    app.use(express.json());
+    app.use((req, _res, next) => { req.actor = boardActor(companyId); next(); });
+    app.use("/api", issueRoutes(db, createStorageService().storage, { chatRunRetries: f.service }));
+    app.use(errorHandler);
+    const lease = { id: randomUUID(), companyId, issueId, heartbeatRunId: f.runId,
+      provider: "daytona", providerLeaseId: randomUUID(), status: "released", cleanupStatus: "success",
+      releasedAt: new Date(), leasePolicy: "retain_on_failure" };
+    const termination = remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" })!;
+    const metadata = { remoteExecutionTermination: termination, sandboxStopAndRetainReceipt: {
+      schema: "paperclip.sandbox-stop-and-retain-receipt.v1", companyId, runId: f.runId,
+      leaseId: lease.id, provider: lease.provider, providerLeaseId: lease.providerLeaseId,
+      requestId: randomUUID(), confirmedAt: termination.confirmedAt, builtinProvider: "daytona", method: "builtin.stopLease",
+    } };
+    await db.insert(environmentLeases).values({ ...lease, metadata,
+      ...(mode === "pending_at_intake" ? { status: "pending_cleanup", cleanupStatus: "failed" } : {}) });
+    await db.update(nativeRunFinalizations).set({ failureCode: "native_session_interrupted",
+      failureDetail: { recoveryMode: "ambiguous_state", providerSessionEstablished: false,
+        providerEventsExist: false, checkpointExists: false } }).where(eq(nativeRunFinalizations.runId, f.runId));
+    await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
+      kind: "active_run_watchdog", status: "resolved", outcome: "blocked", ownerType: "system",
+      cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), nextAction: "Verify source cleanup",
+      evidence: { runId: f.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" },
+        systemRecovery: { ...createSystemRecovery(f.runId), state: mode === "pending_at_intake" ? "running" : "complete" } } });
+    const adapterType = `retained-chat-retry-${randomUUID()}`;
+    const execute = vi.fn(async () => ({ exitCode: 0, signal: null, timedOut: false, errorMessage: null,
+      summary: "Completed the exact saved request after source cleanup", provider: "test", model: "test-model" }));
+    registerServerAdapter({ type: adapterType, supportsLocalAgentJwt: false, execute,
+      testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }) });
+    try {
+      const body = { observedRunId: f.runId, idempotencyKey: randomUUID() };
+      const response = await request(app).post(`/api/issues/${issueId}/retry`).send(body);
+      expect(response.status, JSON.stringify(response.body)).toBe(202);
+      if (mode === "pending_at_intake") {
+        await expect.poll(async () => (await readAgentWakeupRequestsById(response.body.chatActionId))[0]?.status)
+          .toBe("deferred_issue_execution");
+        expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ runId: null });
+        expect(execute).not.toHaveBeenCalled();
+        await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success" })
+          .where(eq(environmentLeases.id, lease.id));
+        await reconcileTaskChatRetryReceipts(db, { companyId });
+        await f.service.processFailedChatRunRetry(response.body.chatActionId);
+      }
+      await expect.poll(async () => (await readAgentWakeupRequestsById(response.body.requestId))[0]?.runId).toBeTruthy();
+      const [intent] = await readAgentWakeupRequestsById(response.body.requestId);
+      const [successor] = await readHeartbeatRunsById(intent.runId!);
+      expect(successor).toMatchObject({ status: "queued", retryOfRunId: f.runId,
+        contextSnapshot: { taskRetryRequestId: intent.id, explicitUserContinuation: {
+          previousRunId: f.runId, readiness: { mode: "fresh_isolated_workspace" } } } });
+      await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed",
+        metadata: { ...metadata, pendingCleanupInFlight: true } }).where(eq(environmentLeases.id, lease.id));
+      await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(and(
+        eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "running"), isNull(heartbeatRuns.wakeupRequestId)));
+      await db.update(agents).set({ status: "idle" }).where(eq(agents.id, f.fixture.assignedAgentId));
+      await heartbeat.resumeQueuedRuns();
+      await heartbeat.drainActiveRunExecutions();
+      expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "queued", startedAt: null });
+      expect(execute).not.toHaveBeenCalled();
+      expect((await readAgentWakeupRequestsById(response.body.requestId))[0]).toMatchObject({ runId: successor.id });
+      if (mode === "operator_stopped") await heartbeat.cancelRun(successor.id);
+      if (mode === "source_edited") await db.update(issueComments).set({ body: "Changed after admission" })
+        .where(eq(issueComments.id, String(f.action.payload.commentId)));
+      if (mode === "live_owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "another-controller",
+        leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
+      await db.update(agents).set({ adapterType }).where(eq(agents.id, f.fixture.assignedAgentId));
+      const { registerRunDispatchWork } = await import("../services/heartbeat/dispatch-work.js");
+      const { writeEnvironmentCleanup } = await import("../services/environment-cleanup-notifications.js");
+      let dispatchAttempts = 0;
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+        onError: error => errors.push(error) });
+      await registerRunDispatchWork(coordinator, { promoteDueScheduledRetries: async () => undefined,
+        nextScheduledRetryAt: async () => null, schedulingSuppressed: async () => false,
+        resumeQueuedRunsAfterDrain: async () => { await heartbeat.resumeQueuedRunsAfterDrain({ companyId }); dispatchAttempts++; },
+      }, () => true).ready;
+      await expect.poll(() => dispatchAttempts).toBeGreaterThan(0);
+      await expect.poll(() => coordinator!.nextWakeAt(), { timeout: 10_000 }).toBeNull();
+      const beforeSettlement = dispatchAttempts;
+      const settledMetadata = { ...metadata, pendingCleanupInFlight: false,
+        remoteExecutionTermination: mode === "unknown_stop" ? { ...termination, providerLeaseId: randomUUID() } : termination };
+      if (mode === "absent_sandbox") {
+        const { withNativeWorkspaceFinalizationOwnership } = await import("../services/native-runtime/native-workspace-finalization-ownership.js");
+        const { settleAbsentSandbox } = await import("../services/sandbox-stop-and-retain.js");
+        let entered!: (token: string) => void;
+        const ownerEntered = new Promise<string>(resolve => { entered = resolve; });
+        const ownerMayFinish = new Promise<void>(resolve => { releaseWorkspaceOwner = resolve; });
+        workspaceOwner = withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: f.runId }, async ownership => {
+          entered(ownership.token);
+          await ownerMayFinish;
+        });
+        const ownerToken = await ownerEntered;
+        const [currentLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id));
+        expect(await settleAbsentSandbox(db, currentLease, { receipt: "verified-upstream", proof: {
+          companyId, issueId, runId: f.runId, leaseId: lease.id, provider: "daytona", providerLeaseId: lease.providerLeaseId,
+          ownerToken, v: 1, iss: "paperclip-cloud", aud: "paperclip-sandbox-absence/v1", sub: "fixture", absent: true,
+          iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 60,
+        } })).toBe(true);
+        expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "queued", startedAt: null });
+        expect(execute).not.toHaveBeenCalled();
+        releaseWorkspaceOwner();
+        await workspaceOwner;
+      } else {
+        await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "released", cleanupStatus: "success",
+          metadata: settledMetadata }).where(eq(environmentLeases.id, lease.id)), { companyId, runId: f.runId, leaseId: lease.id });
+      }
+      await expect.poll(() => dispatchAttempts, { timeout: 10_000 }).toBeGreaterThan(beforeSettlement);
+      if (mode === "source_edited") await expect.poll(async () =>
+        (await readHeartbeatRunsById(successor.id))[0]?.status, { timeout: 10_000 }).toBe("failed");
+      await heartbeat.drainActiveRunExecutions();
+      expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1);
+      expect((await readHeartbeatRunsById(f.runId))[0]).toMatchObject({ status: "failed" });
+      expect((await request(app).post(`/api/issues/${issueId}/retry`).send(body)).body.requestId).toBe(intent.id);
+      expect(errors).toEqual([]);
+      if (mode === "operator_stopped") {
+        expect(execute).not.toHaveBeenCalled();
+        expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "cancelled" });
+        return;
+      }
+      if (mode === "source_edited" || mode === "unknown_stop") {
+        expect(execute).not.toHaveBeenCalled();
+        const [withheld] = await readHeartbeatRunsById(successor.id);
+        if (mode === "unknown_stop") expect(["queued", "failed"]).toContain(withheld.status);
+        if (withheld.status === "failed") expect(withheld.errorCode).toBe("chat_failed_run_retry_not_authorized");
+        return;
+      }
+      if (mode === "live_owner") {
+        await expect.poll(() => coordinator!.nextWakeAt(), { timeout: 10_000 }).toBeNull();
+        expect(execute).not.toHaveBeenCalled();
+        expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "queued", startedAt: null });
+        const { writeExecutionControl } = await import("../services/execution-control-notifications.js");
+        const { DELIVERY_QUEUES } = await import("../services/delivery-work-notifications.js");
+        await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(nativeRunFinalizations)
+          .set({ leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)), { companyId, runId: f.runId });
+      }
+      await expect.poll(() => execute.mock.calls.length, { timeout: 10_000 }).toBe(1);
+      await heartbeat.drainActiveRunExecutions();
+      expect(execute).toHaveBeenCalledTimes(1);
+      expect((await readHeartbeatRunsById(successor.id))[0]).toMatchObject({ status: "succeeded" });
+      expect((await readHeartbeatRunsByCompanyId(companyId)).filter(run => run.retryOfRunId === f.runId)).toHaveLength(1);
+      expect((await readHeartbeatRunsById(f.runId))[0]).toMatchObject({ status: "failed" });
+      expect((await request(app).post(`/api/issues/${issueId}/retry`).send(body)).body.requestId).toBe(intent.id);
+      expect(execute).toHaveBeenCalledTimes(1);
+      await expect.poll(() => coordinator!.nextWakeAt(), { timeout: 10_000 }).toBeNull();
+      expect(errors).toEqual([]);
+    } finally { releaseWorkspaceOwner(); await workspaceOwner?.catch(() => undefined); await coordinator?.stop();
+      await heartbeat.drainActiveRunExecutions(); unregisterServerAdapter(adapterType); await f.service.shutdown(); }
+  });

  it("retries saved GitHub input after a successful run without replaying that run", async () => {
    const f = await successfulGitHubQueuedInputFixture();
@@ -5,7 +5,7 @@ import { readTaskRetryIntent } from "../task-retry.js";
 import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped } from "../native-local-process-stop.js";
 import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
 import { adapterExecutionControls } from "../adapter-execution-control.js";
-import { getExecutionBlocker } from "../execution-blocker.js";
+import { getExecutionBlocker, hasPendingRunCleanup } from "../execution-blocker.js";
 import { readSystemRecovery } from "../system-recovery-state.js";

 /** This receipt changes the workspace strategy, never the external principal,
@@ -34,6 +34,23 @@ export async function taskChatRetryExecutionStopped(db: Db, run: typeof heartbea
  return taskChatRetryExecutionContained(db, run, false);
 }

+export async function taskChatRetrySourceCleanupPending(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) {
+  const hint = run.contextSnapshot?.chatFailedRunRetry as Record<string, unknown> | undefined;
+  const intentId = run.contextSnapshot?.taskRetryRequestId;
+  if (typeof intentId !== "string" || !run.retryOfRunId || !run.wakeupRequestId ||
+      run.contextSnapshot?.issueId !== issueId || hint?.version !== 1 ||
+      hint.actionId !== run.wakeupRequestId || hint.failedRunId !== run.retryOfRunId) return false;
+  const retry = await readTaskChatRetryIntent(db, { companyId: run.companyId, issueId, agentId: run.agentId,
+    sourceRunId: run.retryOfRunId, intentId, chatActionId: run.wakeupRequestId });
+  if (retry?.wake.status !== "coalesced" || retry.wake.runId !== run.id) return false;
+  const [source] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+    eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId),
+    eq(heartbeatRuns.nativeIssueId, issueId), eq(heartbeatRuns.id, run.retryOfRunId),
+    eq(heartbeatRuns.runtimeMode, "native"),
+  ));
+  return Boolean(source && await hasPendingRunCleanup(db, run.companyId, source.id));
+}
+
 async function taskChatRetryExecutionContained(db: Db, run: typeof heartbeatRuns.$inferSelect, admittedFailure: boolean) {
  if (adapterExecutionControls.has(run.id) ||
      (run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date())) return false;
@@ -4,6 +4,7 @@ import { notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js
 import { notifyNativeEnvironmentSettlement, type NativeEnvironmentSettlementScope } from "./native-environment-settlement-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+import { notifyQueuedRunDispatch } from "./run-dispatch-notifications.js";

 export type EnvironmentCleanupWorkScope = NativeEnvironmentSettlementScope & { leaseId?: string };

@@ -28,6 +29,7 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Envi
    }
  }
  await notifyNativeEnvironmentSettlement(transaction, scope);
+  await notifyQueuedRunDispatch(transaction, scope);
  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup);
  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup);
 }
@@ -51,6 +51,7 @@ import {
 import { canRetryStoppedRun } from "../cancelled-native-startup.js";
 import { getExecutionBlocker, hasPendingRunCleanup } from "../execution-blocker.js";
 import { nativeManualAdmissionHoldsExecution } from "../native-runtime/native-runner-ownership.js";
+import { taskChatRetrySourceCleanupPending } from "../chat-channels/task-retry-authority.js";
 import { validateNativeWorkspaceReconstruction } from "../native-runtime/native-workspace-reconstruction.js";
 import { isConversationAdapter } from "../conversation-continuation.js";
 import { recordExecutionWait } from "../execution-wait.js";
@@ -1075,6 +1076,9 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
          owner.executionRunId !== run.id && owner.executionRunId !== run.retryOfRunId) {
        return { ownsIssue, blocked: true };
      }
+      if (ownsIssue && issueId && await taskChatRetrySourceCleanupPending(tx, run, issueId)) {
+        return { ownsIssue, blocked: true };
+      }
      const previousRunId = run.scheduledRetryReason === "native_provider_overloaded"
        ? run.retryOfRunId : ownsIssue ? owner?.executionRunId : null;
      if (previousRunId && previousRunId !== run.id) {
@@ -1,6 +1,7 @@
 import { notifyNativeResultCommit } from "../native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization, releaseNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
+import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js";
 import { beginRunPhase } from "../execution-preparation-timing.js";
 import { logActivity } from "../activity-log.js";
 import { issueRecoveryActionService } from "../issue-recovery-actions.js";
@@ -302,6 +303,7 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
          await notifyNativeResultCommit(tx, { companyId: input.companyId, runId: input.runId });
          await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, runId: input.runId });
          await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.runId });
+          await notifyQueuedRunDispatch(tx, { companyId: input.companyId });
          await tx.update(heartbeatRuns).set({
            runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`,
          }).where(owned);
@@ -219,12 +219,19 @@ describe("native workspace finalization recovery", () => {
      observed.push(db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns)
        .where(eq(heartbeatRuns.id, seed.runId)).then(rows => rows[0].profile));
    });
+    const queued: Array<Promise<unknown>> = [];
+    const stopQueued = subscribeDeliveryWork(db, DELIVERY_QUEUES.queuedRunDispatch, () => {
+      queued.push(db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, seed.runId)).then(rows => rows[0].profile));
+    });
    let receipt: Awaited<ReturnType<typeof resumeNativeWorkspaceAfterOwnerStop>>;
    try {
      receipt = await resumeNativeWorkspaceAfterOwnerStop(seed);
      await vi.waitFor(() => expect(observed).toHaveLength(1));
      expect(await Promise.all(observed)).toEqual([{ preserved: "workspace descriptor" }]);
-    } finally { stop(); }
+      await vi.waitFor(() => expect(queued).toHaveLength(1));
+      expect(await Promise.all(queued)).toEqual([{ preserved: "workspace descriptor" }]);
+    } finally { stop(); stopQueued(); }
    expect(receipt).toEqual({ runId: seed.runId, resultId: before.resultId, status: "queued" });
    const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId));
    expect(after).toEqual(before);
@@ -2,6 +2,7 @@ import { notifyNativeResultCommit } from "../native-result-commit-notifications.
 import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js";
 import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
+import { notifyQueuedRunDispatch } from "../run-dispatch-notifications.js";
 import os from "node:os";
 import { and, eq, sql } from "drizzle-orm";
 import { heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
@@ -71,6 +72,7 @@ export async function resumeNativeWorkspaceAfterOwnerStop(input: {
    await notifyNativeResultCommit(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId });
    await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId });
    await notifyNativeStatusReassessment(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId });
+    await notifyQueuedRunDispatch(tx, { companyId: input.companyId });
    await tx.update(heartbeatRuns).set({
      runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - ${OWNER_KEY}`, updatedAt: now,
    }).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, run.id),
@@ -3,11 +3,16 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio

 export type RunDispatchScope = { companyId: string };

+export async function notifyQueuedRunDispatch(transaction: object, scope?: RunDispatchScope) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch,
+    scope ? JSON.stringify({ companyId: scope.companyId }) : "*");
+}
+
 export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) {
  await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined);
  const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativeProcessSessions, hint);
-  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);
+  await notifyQueuedRunDispatch(transaction, scope);
 }

 /** Only a retry row or deadline change needs to re-read the saved retry schedule. */
@@ -361,14 +361,48 @@ describe("automatic retained workspace recovery", () => {
    return app;
  }

-  async function acceptedRemoteExport() {
+  type RecoverySourceLayout = "retained" | "published" | "raw";
+
+  async function acceptedRemoteExport(sourceLayout: RecoverySourceLayout = "retained") {
    const f = await seed("native", false);
-    const source = await run(f.runId), workspaceId = randomUUID();
-    const publishedRoot = path.join(path.dirname(f.remote), ".paperclip-workspaces", workspaceId);
-    await mkdir(path.dirname(publishedRoot), { recursive: true });
-    await rename(f.remote, publishedRoot);
-    f.remote = publishedRoot;
-    const remoteCwd = await realpath(f.remote);
+    const source = await run(f.runId);
+    let workspaceId: string = randomUUID(), remoteCwd = await realpath(f.remote);
+    let preparedReference: Awaited<ReturnType<typeof prepareRemoteWorkspace>>["reference"] | undefined;
+    if (sourceLayout === "published") {
+      // Fresh modern sources retain the actual publisher's path and authority.
+      // The explicit raw fixture remains available for refusal coverage.
+      targets.get(f.lease.id)!.remoteCwd = remoteCwd;
+      const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
+      await db.update(issues).set({ status: "in_progress", executionRunId: f.runId, executionLockedAt: new Date() })
+        .where(eq(issues.id, f.issueId));
+      await db.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null })
+        .where(eq(heartbeatRuns.id, f.runId));
+      const leaseMetadata = { ...f.lease.metadata };
+      delete leaseMetadata.remoteExecutionTermination;
+      const [activeLease] = await db.update(environmentLeases).set({ status: "active", releasedAt: null,
+        cleanupStatus: null, metadata: leaseMetadata }).where(eq(environmentLeases.id, f.lease.id)).returning();
+      const reserved = await reserveRemoteWorkspace({ db, companyId: f.companyId, issueId: f.issueId,
+        runId: f.runId, agentId: source.agentId, environmentId: activeLease.environmentId!,
+        projectId: null, projectWorkspaceId: null, existingWorkspaceId: null, source: { kind: "empty" } });
+      const prepared = await prepareRemoteWorkspace({ db, workspace: reserved, runId: f.runId,
+        lease: activeLease as EnvironmentLease, target: targets.get(f.lease.id)! });
+      workspaceId = prepared.workspace.id; remoteCwd = prepared.remoteRoot; preparedReference = prepared.reference;
+      for (const name of ["existing.txt", "saved.txt"]) await cp(path.join(f.remote, name), path.join(remoteCwd, name));
+      await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success", releasedAt: new Date(),
+        metadata: { ...prepared.lease.metadata,
+          remoteExecutionTermination: remoteTerminationReceipt(prepared.lease, { providerLeaseId: f.providerLeaseId, state: "stopped" }) },
+      }).where(eq(environmentLeases.id, f.lease.id));
+      await db.update(heartbeatRuns).set({ status: source.status, finishedAt: source.finishedAt,
+        error: source.error, errorCode: source.errorCode }).where(eq(heartbeatRuns.id, f.runId));
+      await db.update(issues).set({ status: issue.status, executionRunId: issue.executionRunId,
+        executionLockedAt: issue.executionLockedAt }).where(eq(issues.id, f.issueId));
+    } else if (sourceLayout === "retained") {
+      const publishedRoot = path.join(path.dirname(f.remote), ".paperclip-workspaces", workspaceId);
+      await mkdir(path.dirname(publishedRoot), { recursive: true });
+      await rename(f.remote, publishedRoot);
+      f.remote = publishedRoot;
+      remoteCwd = await realpath(f.remote);
+    }
    const git = (args: string[], cwd = remoteCwd) => new Promise<string>((resolve, reject) =>
      execFile("git", args, { cwd }, (error, stdout) => error ? reject(error) : resolve(stdout.trim())));
    await git(["init"]);
@@ -380,7 +414,7 @@ describe("automatic retained workspace recovery", () => {
    const authority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: null,
      source: { kind: "empty" }, writerRunId: f.runId, dirty: true,
      location: { environmentId: f.lease.environmentId!, leaseId: f.lease.id, providerLeaseId: f.providerLeaseId, remoteCwd } };
-    await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId, sourceIssueId: f.issueId,
+    if (sourceLayout !== "published") await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId, sourceIssueId: f.issueId,
      name: "Completed remote work", cwd: remoteCwd, providerType: "cloud_sandbox", mode: "isolated_workspace",
      strategyType: "project_primary", status: "active", metadata: { remoteWorkspace: authority } });
    await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, f.issueId));
@@ -396,7 +430,7 @@ describe("automatic retained workspace recovery", () => {
      canonicalSha256: "contract", createdByActorType: "system", createdByActorId: "test" });
    await db.update(heartbeatRuns).set({ completionContractId: contractId, resultJson: {}, runnerProfileJson: {
      adapterDispatch: { adapterType: "paperclip_runner" },
-      remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId: lease.id,
+      remoteWorkspace: preparedReference ?? { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId: lease.id,
        providerLeaseId: f.providerLeaseId, remoteCwd, epoch: 1, state: "prepared", snapshotId: null },
    } }).where(eq(heartbeatRuns.id, f.runId));
    await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId, runId: f.runId,
@@ -407,11 +441,11 @@ describe("automatic retained workspace recovery", () => {
    await db.update(issueRecoveryActions).set({ cause: "native_workspace_sync_out_retry_exhausted", evidence: { runId: f.runId } })
      .where(eq(issueRecoveryActions.id, f.actionId));
    await adoptSystemRecovery(db, f);
-    return { ...f, source, workspaceId, resultId, git, commit };
+    return { ...f, ...(sourceLayout === "published" ? { remote: remoteCwd } : {}), source, workspaceId, resultId, git, commit };
  }

-  async function interruptedRemotePreparation(interruption: "shutdown" | "lease" = "shutdown") {
-    const f = await acceptedRemoteExport();
+  async function interruptedRemotePreparation(interruption: "shutdown" | "lease" = "shutdown", sourceLayout: RecoverySourceLayout = "retained") {
+    const f = await acceptedRemoteExport(sourceLayout);
    await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
    await db.delete(nativeRunResults).where(eq(nativeRunResults.runId, f.runId));
    const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] };
@@ -437,8 +471,8 @@ describe("automatic retained workspace recovery", () => {
    return f;
  }

-  async function failedRemotePreparation() {
-    const f = await interruptedRemotePreparation();
+  async function failedRemotePreparation(sourceLayout: RecoverySourceLayout = "retained") {
+    const f = await interruptedRemotePreparation("shutdown", sourceLayout);
    const source = await run(f.runId);
    const error = "Repository access is no longer available; refresh the repository catalog";
    await terminalizeLegacyExecution({ db, run: source, status: "failed", reconcileIfNeeded: false,
@@ -691,9 +725,9 @@ describe("automatic retained workspace recovery", () => {
    expect(await readFile(path.join(cwd, "saved.txt"), "utf8")).toBe("existing task work\n");
  });

-  it.each(["complete", "catalog", "catalog_warning", "catalog_unknown_warning", "catalog_provider_warning"] as const)(
+  it.each(["complete", "catalog", "catalog_warning", "catalog_unknown_warning", "catalog_provider_warning", "invalid_published_root"] as const)(
    "dispatches the recovered failed-preparation Retry once through the native provider boundary (%s)", async mode => {
-    const unavailableCatalog = mode !== "complete", withWarning = mode.includes("warning");
+    const unavailableCatalog = mode !== "complete" && mode !== "invalid_published_root", withWarning = mode.includes("warning");
    const localStorage = getStorageService();
    const databaseUrl = process.env.DATABASE_URL, masterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY;
    const experimental = await instanceSettingsService(db).getExperimental();
@@ -705,7 +739,7 @@ describe("automatic retained workspace recovery", () => {
      storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" });
      process.env.DATABASE_URL = connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "63".repeat(32);
      await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableIsolatedWorkspaces: true });
-      const f = await failedRemotePreparation();
+      const f = await failedRemotePreparation(mode === "invalid_published_root" ? "raw" : "published");
      await recoverRetainedWorkspace(f.input);
      const source = await run(f.runId), nextDirectory = path.join(root, randomUUID(), "next");
      await mkdir(nextDirectory, { recursive: true });
@@ -809,6 +843,15 @@ describe("automatic retained workspace recovery", () => {
        eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "running")));
      await heartbeat.resumeQueuedRuns(); await heartbeat.drainActiveRunExecutions();
      const completed = await run(successors[0].id);
+      if (mode === "invalid_published_root") {
+        expect(completed).toMatchObject({ status: "failed", runtimeMode: "legacy", errorCode: "setup_failed",
+          error: "remote_workspace_published_path_invalid" });
+        expect(inputs).toHaveLength(0); expect(startTurn).not.toHaveBeenCalled();
+        expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, completed.id))).toHaveLength(0);
+        expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
+        expect((await run(f.runId)).error).toBe(source.error);
+        return;
+      }
      if (unavailableCatalog) {
        expect(completed).toMatchObject({ status: "failed", errorCode: "setup_failed",
          error: "Repository access is no longer available; refresh the repository catalog" });
@@ -855,6 +898,7 @@ describe("automatic retained workspace recovery", () => {
      expect({ status: completed.status, error: completed.error, runtimeMode: completed.runtimeMode })
        .toEqual({ status: "succeeded", error: null, runtimeMode: "native" });
      expect(inputs).toHaveLength(1); expect(startTurn).toHaveBeenCalledOnce();
+      expect(path.basename(path.dirname(inputs[0].workspace.cwd!))).toBe(".paperclip-workspaces");
      expect(completed.runnerProfileJson?.nativeExecutionInput).toMatchObject({ provider: { kind: "codex", model: "gpt-6.1-sol" } });
      expect(await readFile(path.join(inputs[0].workspace.cwd!, "continued.txt"), "utf8")).toBe("continued from preserved work\n");
      expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, completed.id)))