fix(native): recover saved listener startup failures (#19479)

## Thinking Path

> - Paperclip manages AI agents and their work.
> - Native recovery must preserve saved work and respect current task
authority.
> - A closed local API listener can stop startup before a provider
starts.
> - #19462 gives new failures a bounded bootstrap retry.
> - Older saved failures still have a generic terminal code and a
recovery hold.
> - This pull request verifies the old producer outcome and uses the
existing safe replacement path.
> - Eligible tasks continue once. Unknown provider outcomes and user
holds remain blocked.

## Linked Issues or Issue Description

Refs: #19462

**What happened?**

A restart can leave an old listener startup failure in
`terminal_failure` with `native_session_interrupted`. Startup recovery
does not claim it after #19462. A completed file recovery can also
retain its no-replay hold.

**Expected behavior**

Continue automatically when durable records prove that no provider
started, the source stopped, and the current files were saved. Keep the
existing attempt budget and normal admission checks.

**Steps to reproduce**

1. Close the actual API listener after a fresh native coordinator claim.
2. Persist the old executor's generic failure. Save the workspace and
confirm source termination.
3. Restart recovery. The baseline schedules no successor. The repair
schedules one successor after all receipts pass.

**Paperclip version or commit**

Baseline: `0252421d5c`. Old failure producer: `2f35eb1080`.

**Deployment mode**

Built from source. Local macOS arm64, Node 26.4.0, real embedded
PostgreSQL, independent filesystem execution. No external AI service is
used by the controlled acceptance.

## What Changed

- Recognize the exact saved local-listener guard through one proof
predicate shared by discovery and the locked transaction.
- Require a fresh first attempt, matching execution bindings, no prior
provider lineage, no provider events or tool calls, no checkpoint or
admission receipt, and no cancellation.
- Reuse current snapshot, exact stop, source block, attempt budget, and
successor idempotency checks.
- Wait for completed source file-recovery receipts. Retire only that
same source's verified preservation hold in the successor transaction.
- Bound discovery to 100 candidates per pass, keep completed history
quiet, and wake through existing transactional producer notifications.
- Retain retries after transient database and final ownership-proof
errors; advance the bounded cursor and force a rescan so a failing page
cannot hide later work.
- Let only the replacement duty discharge a proved-empty notification
during drain through its existing read-only readiness check, once per
notification generation. Keep real work and concurrent changes
outstanding without running replacement during drain.
- Wake replacement at canonical local-stop proof writes and the existing
dispatch eligibility boundary; these hints grant no new execution
authority.
- Add database regressions, scheduler quietness checks, real controller
handoff coverage, and an opt-in controlled Runner continuation.

## Verification

Final production head: `0df41b2fef2644195d1224d36984af3827da3e3b`.

Local environment: macOS arm64, Node 26.4.0, TypeScript 7.0.2, Vitest
5.0.3, frozen-lockfile dependencies. Tests execute freshly;
`--fsModuleCache` reuses only content-validated module transforms.

- Native safe replacement: 62/62 PASS on isolated PostgreSQL 17 (63.18
seconds), including clean per-suite database disposal. All 62 assertions
also passed on embedded PostgreSQL 18 beta, but its Darwin cleanup hook
timed out; that invocation is not counted as a pass. Docker's disk was
full, so the disposable PostgreSQL 17 data used a size-limited RAM
mount, with no host mounts. The owned container was removed afterward.
- Actual stopped-process producer suite: 31/31 PASS. A real exited child
produces the stop receipt, and its replacement notification settles only
after the outer transaction commits.
- Workspace and separate-process drain regressions: 33/33 PASS (75.90
seconds).
- Shared coordinator and execution-control lifecycle checks: 39/39 PASS
(16.43 seconds). Pending work, concurrent changes, cold startup,
shutdown, and passive drain boundaries remain covered.
- Actual suppressed Cloud routine/drain workflow: PASS (34.64 seconds),
using the unchanged test and timeouts. Its matching local RED reproduced
CI's 462-query, zero-pending-query, one-idle-hold timeout after an
admitted write committed during drain (69.46 seconds).
- Actual historical controller handoff: PASS (48.40 seconds). A separate
controller loads the original executor from `2f35eb1080`, writes its
generic listener failure, and exits. Recovery at the final head admits
exactly one successor. The real Rust Runner and controlled Codex process
complete governed `paperclip_finish`; attempt two commits and the saved
file remains intact. Only the child loads the old artifact, and both
isolated artifacts are restored afterward. Rust source subtree
`77961b70f177ec3c38d11abcaec92fd3d0e96ec0` matches the built binary. No
external AI service is used.
- Seven corrected existing fixtures pass: transactional stop-proof calls
(5 cases) and notification inventory (2 cases). Their earlier CI
failures exposed fixture calls outside the documented transaction and an
outdated expected queue set.
- Server typecheck: PASS.

Red/green proofs also cover the original saved failure on baseline
`0252421d5c` (zero successors instead of one), held-history quietness,
bounded forward scanning through 100 parked rows and concurrent wakes,
real snapshot/cleanup producer notifications, actual issue-row lock
timeout, and task/history eligibility notifications. An obsolete
prior-run question can be skipped to request another proof check; actual
skip while the task is paused or the source owns provider tool activity
schedules no successor.

The failed-page regression starts with 25 persistent verifier errors and
a healthy candidate 26. Baseline RED repeats the first page. Repair
GREEN advances the healthy candidate, revisits the original failures
after their fault clears, creates all 26 successors exactly once, and
becomes quiet.

Greptile is SUCCESS, 5/5 on the exact final head (2026-11-06 17:09:21
UTC). Final-head CI is GREEN: 56 successful checks, three intentional
skips, one neutral result, and no pending or failing checks. [Attempt
two and
results](https://github.com/paperclipai/paperclip/actions/runs/42258352876/attempts/2)
include passing build, typecheck, Rust, native server, package, browser
E2E, serialized server, general server/chat tests, and final
verification. Five jobs in the first attempt were interrupted together
at 17:13:12 UTC; the native-server log records an explicit runner
shutdown signal. The single infrastructure retry reran only those five
jobs and dependent verification, preserving the 49 successful checks on
the unchanged head. No timeout was weakened, and setup failures or
interrupted invocations are not counted as passes.

Focused commands from the repository root:

```sh
./server/node_modules/.bin/vitest run --project @paperclipai/server server/src/services/remote-workspaces.integration.test.ts server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts -t 'native safe-replacement decision|reconciles the drained remote backend' --fsModuleCache --reporter=dot
./server/node_modules/.bin/vitest run --project @paperclipai/server server/src/__tests__/execution-control-work.test.ts server/src/__tests__/delivery-work-coordinator.test.ts -t 'execution-control lifecycle|work coordinator' --fsModuleCache --reporter=dot
node node_modules/typescript/bin/tsc -p server/tsconfig.json --noEmit --pretty false
```

The historical acceptance sets `PAPERCLIP_TEST_HISTORICAL_LISTENER=1`
and uses an original-executor child artifact with a `try/finally`
restore. Normal regressions require no Rust binary.

## Risks

- Compatibility is deliberately limited to positively proved fresh
remote startup failures. Other old ambiguous failures remain held.
- Existing provider work, a changed task block, incomplete preservation,
or revoked authority must prevent execution. Normal retry admission
remains in control.
- This PR does not deploy the change or mutate a hosted task. Hosted
recovery still requires the repaired revision and all current proof
gates.

## Model Used

OpenAI GPT-6 through Codex, with reasoning, tool use, and code
execution. The exact backend model identifier and context window are not
exposed in this session.

## 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>

doc/execution-semantics.md

@@ -3633,6 +3633,34 @@ re-read dependency readiness, since new dependencies need not change the
 displayed task status. Old blocked rows without a receipt remain held; no
 historical status backfill is performed.
 
+Older native executors recorded the local callback-listener startup guard as
+`native_session_interrupted`. A saved exact guard outcome can enter the same
+bounded safe-replacement path after restart. This requires the first attempt,
+an unused provider-session identity, matching saved execution bindings, no
+provider events, checkpoint, admission receipt, tool call, or cancellation, and
+the existing current snapshot and confirmed stop receipts. A generic error or
+a pending provider-pack receipt alone grants no recovery authority. The same
+transaction can retire that run's completed file-preservation hold only when
+its bound recovery and finalization receipts prove completion. Unrelated holds,
+later task blocks, normal dispatch admission, and the three-attempt budget remain
+effective.
+
+Historical replacement discovery reads at most 100 candidates per pass and
+keeps a cursor only for the remaining scan. A work notification preserves forward
+progress and requests a final rescan, so held earlier rows cannot hide a later
+eligible run. Completed scans of held history become quiet. Final snapshot
+publication, verified lease cleanup, local process-stop proof, completed workspace
+recovery, and dispatch eligibility changes notify the existing replacement worker
+in their transactions. A cleared historical interaction only requests another
+proof check; it grants no new continuation authority. Source provider effects,
+task pause or cancellation, and the original authority checks still prevent replay.
+Temporary decision failures retain the existing worker's retry and advance the
+bounded cursor, with a forced rescan of earlier failures. A failed page cannot
+hide later ready runs. During drain, the initialized replacement worker can read
+readiness once per notification generation to discharge an empty notification
+without starting replacement. Concrete pending work and changes arriving during
+that read remain outstanding. Restart reconstructs the scan from durable records.
+
 For legacy GitHub input without a saved App ID or its own response receipt,
 Task Retry may prove the endpoint's earlier non-null App claim from a processed
 response ledger created before that input. The proof stays within the same

server/src/__tests__/execution-control-work.test.ts

@@ -61,6 +61,41 @@ describe("execution-control lifecycle", () => {
     expect(run).toHaveBeenCalledTimes(runs + 1);
     expect(c.nextWakeAt()).toBeNull();
   });
+  it.each(["empty", "pending", "concurrent"] as const)("settles only proved-empty replacement notifications during drain (%s)", async kind => {
+    vi.useFakeTimers(); const owner = fakeOwner(); let draining = false;
+    const c = coordinator(owner, () => !draining);
+    const run = vi.fn(async () => {}), next = vi.fn(async (): Promise<number | null> => null);
+    await registerExecutionControlWork(c, [{ queue: DELIVERY_QUEUES.executionReplacement, run, next,
+      settleEmptyWhilePaused: true }]).ready;
+    const before = idleWorkSnapshot().active;
+    draining = true;
+    let release: (() => void) | undefined;
+    if (kind === "pending") next.mockResolvedValue(Date.now() + 15_000);
+    if (kind === "concurrent") next.mockImplementationOnce(() => new Promise(resolve => {
+      release = () => resolve(null);
+    }));
+    await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement));
+    await vi.advanceTimersByTimeAsync(15_000);
+    if (kind === "concurrent") {
+      expect(release).toBeTypeOf("function");
+      next.mockResolvedValue(Date.now() + 15_000);
+      await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement));
+      release!(); await vi.advanceTimersByTimeAsync(1);
+    }
+    expect(run).toHaveBeenCalledTimes(1);
+    expect(idleWorkSnapshot().active).toBe(before + (kind === "empty" ? 0 : 1));
+    const reads = next.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(next).toHaveBeenCalledTimes(reads);
+    expect(run).toHaveBeenCalledTimes(1);
+    if (kind !== "empty") {
+      draining = false; next.mockResolvedValue(null);
+      await vi.advanceTimersByTimeAsync(15_000);
+      expect(run).toHaveBeenCalledTimes(2);
+    }
+    expect(idleWorkSnapshot().active).toBe(before);
+    expect(c.nextWakeAt()).toBeNull();
+  });
   it("keeps a change arriving during a sweep and does not overlap it", async () => {
     vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner);
     let release!: () => void;

server/src/__tests__/helpers/historical-listener-continuation.ts

@@ -0,0 +1,134 @@
+import { createServer } from "node:http";
+import { spawn } from "node:child_process";
+import { createRequire } from "node:module";
+import { join, resolve } from "node:path";
+import { mkdir, readFile, writeFile } from "node:fs/promises";
+import { eq } from "drizzle-orm";
+import { completionContracts, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
+import { expect } from "vitest";
+import { createRunDispatch } from "../../modules/run-dispatch/index.js";
+import { prepareNativeHeartbeatRun } from "../../services/native-runtime/prepare-native-run.js";
+import { buildNativeExecutionInput } from "../../services/native-runtime/native-execution-input.js";
+import { buildNativeCompletionContract } from "../../services/native-runtime/completion-contracts.js";
+import { nativeRuntimeContextFixture } from "../../services/native-runtime/runtime-context.test-fixture.js";
+import { materializeAsset } from "../../services/native-runtime/runtime-context.js";
+import { executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js";
+import { reconcileNativeFinalizations } from "../../services/native-runtime/native-finalization-reconciler.js";
+import { canonicalNativeRuntimeContextDigest, createNativeSessionBackend, createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "../../vendor/paperclip-runner/index.js";
+import { registerLocalRuntimeApiServer } from "../../runtime-api.js";
+import { registerRunnerPrpAuthority, runnerPrpWebSocketInternals, setupRunnerPrpWebSocketServer } from "../../realtime/runner-prp-ws.js";
+
+async function prepare(db: Db, runId: string, cwd: string, environmentLeaseId: string) {
+  const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+  const [issue] = await db.select().from(issues).where(eq(issues.id, run.nativeIssueId ?? String(run.contextSnapshot?.issueId)));
+  await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, issue.id));
+  const native = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId });
+  const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issue.id));
+  const runtimeContext = nativeRuntimeContextFixture();
+  runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444,
+    content: Buffer.from("Complete the saved task once.\n") }]);
+  runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext);
+  const execution = buildNativeExecutionInput({ companyId: run.companyId, agentId: run.agentId, runId, issue,
+    taskPrompt: "Preserve the saved file and complete the task once.", provider: "codex",
+    workspace: { id: issue.executionWorkspaceId!, cwd, repoUrl: null, repoRef: null, branchName: null },
+    normalizedSessionId: native.normalizedSessionId, runtimeContext,
+    completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
+      contract: buildNativeCompletionContract(issue, { revision: contract.revision }) },
+  });
+  const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+  await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, ...current.runnerProfileJson, nativeExecutionInput: execution } })
+    .where(eq(heartbeatRuns.id, runId));
+  return { run: current, native, execution };
+}
+
+export async function recordHistoricalListenerFailure(input: {
+  db: Db; connectionString: string; runId: string; cwd: string; environmentId: string; leaseId: string;
+}) {
+  await input.db.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null })
+    .where(eq(heartbeatRuns.id, input.runId));
+  const { execution, native } = await prepare(input.db, input.runId, input.cwd, input.leaseId);
+  await input.db.insert(nativeRunFinalizations).values({ runId: input.runId, companyId: execution.binding.companyId,
+    issueId: execution.binding.issueId, phase: "observed" });
+  const child = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"),
+    resolve(import.meta.dirname, "native-drain-controller.test-worker.ts")], { stdio: ["ignore", "pipe", "pipe", "ipc"], env: process.env });
+  const logs: string[] = [];
+  child.stdout?.on("data", data => logs.push(String(data)));
+  child.stderr?.on("data", data => logs.push(String(data)));
+  const exited = new Promise(resolveExit => child.once("exit", resolveExit));
+  try {
+    const result = await new Promise((resolveResult, reject) => {
+      const timeout = setTimeout(() => reject(new Error(logs.join("").slice(-3000))), 60_000);
+      child.once("error", reject);
+      child.on("message", (message: { ready?: boolean; done?: boolean; error?: string }) => {
+        if (message.ready) child.send({ connectionString: input.connectionString, options: { execution,
+          runnerInstanceId: native.runnerInstanceId, useRunnerd: true,
+          runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId: input.environmentId,
+            leaseId: input.leaseId, remoteCwd: input.cwd } } });
+        if (message.done || message.error) { clearTimeout(timeout); message.error ? reject(new Error(message.error)) : resolveResult(message); }
+      });
+    });
+    expect(result).toMatchObject({ commands: 0, errorMessage: "local_runtime_api_listener_unavailable" });
+    expect(await exited).toBe(0);
+    const [failed] = await input.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, input.runId));
+    expect(failed).toMatchObject({ phase: "terminal_failure", attempt: 1, failureCode: "native_session_interrupted",
+      failureDetail: { message: "local_runtime_api_listener_unavailable", recoveryMode: "ambiguous_state",
+        providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false } });
+  } finally { child.kill("SIGKILL"); await exited; }
+}
+
+export async function completeHistoricalListenerSuccessor(input: { db: Db; runId: string; root: string;
+  prepareWorkspace: () => Promise<{ cwd: string; finalize: () => Promise<void> }> }) {
+  const [scheduled] = await input.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId));
+  const dispatch = createRunDispatch(input.db);
+  expect(await dispatch.promoteScheduledRetry({ companyId: scheduled.companyId, runId: scheduled.id,
+    now: new Date(scheduled.scheduledRetryAt!.getTime() + 1) })).toMatchObject({ outcome: "promoted" });
+  expect(await dispatch.cancelStaleQueuedRun({ companyId: scheduled.companyId, runId: scheduled.id, expectedStatus: "queued" }))
+    .toMatchObject({ outcome: "not_stale" });
+  await input.db.update(heartbeatRuns).set({ status: "running", runtimeMode: "native",
+    nativeIssueId: String(scheduled.contextSnapshot?.issueId) }).where(eq(heartbeatRuns.id, scheduled.id));
+  await input.db.update(issues).set({ executionRunId: scheduled.id }).where(eq(issues.id, String(scheduled.contextSnapshot?.issueId)));
+  const workspace = await input.prepareWorkspace();
+  const { execution, native } = await prepare(input.db, scheduled.id, workspace.cwd, `controlled-${scheduled.id}`);
+  await input.db.insert(nativeRunFinalizations).values({ runId: scheduled.id, companyId: scheduled.companyId,
+    issueId: execution.binding.issueId, phase: "observed" });
+  const home = join(input.root, "codex-home"), report = join(input.root, "result.json"), calls = join(input.root, "calls.log");
+  await mkdir(home, { recursive: true });
+  await writeFile(report, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
+    summary: "Saved task completed after historical controller failure.", completionClaim: {
+      contractRevision: execution.completionContract.contract.revision, objectiveSatisfied: true,
+      criteria: execution.completionContract.contract.criteria.map(criterion => ({ criterionId: criterion.id, status: "satisfied", evidenceRefs: [] })),
+      remainingWork: [] }, evidence: [], verification: [], artifacts: [], attentionRequests: [] }));
+  const server = createServer();
+  registerLocalRuntimeApiServer(server);
+  await new Promise<void>(resolveListen => server.listen(0, "127.0.0.1", resolveListen));
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("TCP listener required");
+  setupRunnerPrpWebSocketServer(server, { apiUrl: `http://127.0.0.1:${address.port}` });
+  const bundle = createRunnerdCodexTransport({ runnerBinary: defaultCapabilityRunnerdBinary(),
+    codexCommand: resolve(import.meta.dirname, "../../../../packages/paperclip-runner/runner/target/debug/fake-codex-app-server"),
+    codexArgs: ["--state-file", join(input.root, "provider.json"), "--terminal-result-file", report, "--call-log", calls, "--record-process-start"],
+    stateDirectory: input.root, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
+    runtimeContext: execution.runtimeContext, runnerRuntimeContext: execution.runtimeContext,
+    sourceCodexHome: home, environment: { CODEX_HOME: home, PATH: process.env.PATH },
+    prpIdentity: { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: native.environmentLeaseId,
+      runId: scheduled.id, normalizedSessionId: native.normalizedSessionId, turnId: native.turnId, itemId: native.itemId },
+    controlPlaneRegistration: authority => registerRunnerPrpAuthority({ companyId: scheduled.companyId, runId: scheduled.id, authority }),
+  });
+  const backend = createNativeSessionBackend(execution, { runnerInstanceId: native.runnerInstanceId,
+    environment: { CODEX_HOME: home, PATH: process.env.PATH }, codexTransportFactory: () => bundle.transport });
+  try {
+    expect(await executePaperclipNativeSession({ db: input.db, execution, backend, useRunnerd: true,
+      runnerInstanceId: native.runnerInstanceId, turnTimeoutMs: 15_000 })).toMatchObject({ exitCode: 0, timedOut: false });
+    await workspace.finalize();
+    await reconcileNativeFinalizations(input.db, [scheduled.id]);
+    const [finished] = await input.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, scheduled.id));
+    expect(finished, JSON.stringify(finished.failureDetail)).toMatchObject({ phase: "committed", attempt: 2 });
+    expect(await input.db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, scheduled.id)))
+      .toMatchObject([{ schemaStatus: "accepted" }]);
+    expect((await readFile(calls, "utf8")).split("\n").filter(line => line === "process-start")).toHaveLength(1);
+    expect(await readFile(join(workspace.cwd, "saved.txt"), "utf8")).toBe("work survives a failed provider session\n");
+  } finally {
+    await bundle.transport.close(); runnerPrpWebSocketInternals.resetForTests();
+    server.closeAllConnections(); await new Promise<void>(resolveClose => server.close(() => resolveClose()));
+  }
+}

server/src/__tests__/helpers/native-drain-controller.test-worker.ts

@@ -0,0 +1,50 @@
+import { createServer } from "node:http";
+import { closeRegisteredClients, createDb, heartbeatRunEvents } from "@paperclipai/db";
+import { registerLocalRuntimeApiServer } from "../../runtime-api.js";
+import { setStartupRecoveryPhase } from "../../startup-recovery-state.js";
+import { heartbeatService } from "../../services/heartbeat.js";
+import { executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js";
+
+process.send?.({ ready: true });
+process.once("message", async (input: {
+  connectionString: string;
+  options: Omit<Parameters<typeof executePaperclipNativeSession>[0], "db">;
+  providerEvidence?: boolean;
+}) => {
+  const api = createServer((_request, response) => response.writeHead(404).end());
+  registerLocalRuntimeApiServer(api);
+  let commands = 0;
+  let errorName: string | undefined;
+  let errorMessage: string | undefined;
+  try {
+    await new Promise<void>(resolve => api.listen(0, "127.0.0.1", resolve));
+    const db = createDb(input.connectionString);
+    try {
+      await executePaperclipNativeSession({ ...input.options, db,
+        runnerExecutionTarget: { ...input.options.runnerExecutionTarget!,
+          runner: { execute: async () => { commands++; throw new Error("Drained controller launched remote work"); } },
+        } as never,
+        onEvent: async event => {
+          if (event.payload?.span !== "native.coordinator.claim") return;
+          if (input.providerEvidence) await db.insert(heartbeatRunEvents).values({
+            companyId: input.options.execution.binding.companyId, agentId: input.options.execution.binding.agentId,
+            runId: input.options.execution.binding.runId, seq: 1_000_000, eventType: "turn.started",
+            sourceEventId: "controlled-provider-evidence" });
+          setStartupRecoveryPhase("draining");
+          await new Promise<void>((resolve, reject) => api.close(error => error ? reject(error) : resolve()));
+        },
+      });
+    } catch (error) {
+      errorName = error instanceof Error ? error.name : String(error);
+      errorMessage = error instanceof Error ? error.message : String(error);
+    }
+    const heartbeat = heartbeatService(db);
+    try {
+      await heartbeat.drainRunningRunsForShutdown("SIGTERM", new Date(), [input.options.execution.binding.runId]);
+    } finally { await heartbeat.drainActiveRunExecutions(); }
+    await closeRegisteredClients(input.connectionString, { scope: "database" });
+    process.send?.({ done: true, commands, errorName, errorMessage }, () => process.exit(0));
+  } catch (error) {
+    process.send?.({ error: error instanceof Error ? error.stack : String(error) }, () => process.exit(1));
+  }
+});

server/src/__tests__/status-card-notifications.test.ts

@@ -25,7 +25,7 @@ it.each([false, true])("keeps execution queue consumers intact and emits only a
     const write = vi.fn(async () => "written");
     expect(await writeExecutionControl(owner as never, DELIVERY_QUEUES.workspaceRecovery, write, scope)).toBe("written");
     expect(write).toHaveBeenCalledOnce();
-    expect(new Set(settled.map(row => row.queue))).toEqual(new Set([DELIVERY_QUEUES.issueMonitors, DELIVERY_QUEUES.workspaceRecovery,
+    expect(new Set(settled.map(row => row.queue))).toEqual(new Set([DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.issueMonitors, DELIVERY_QUEUES.workspaceRecovery,
       ...(scoped ? [DELIVERY_QUEUES.statusCardEvaluation] : [])]));
   } finally { for (const stop of unsubscribe) stop(); }
 });

server/src/index.ts

@@ -81,7 +81,7 @@ import { embeddedPostgresOwnerPort } from "./embedded-postgres-owner.js";
 import { deliverExecutionStatuses } from "./services/execution-status-delivery.js";
 import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js";
 import { deliverPendingPauseResumes } from "./services/issue-tree-resume.js";
-import { reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js";
+import { invalidateNativeReplacementDiscovery, reconcileSafeNativeReplacements } from "./services/native-runtime/native-safe-replacement.js";
 import { reconcileAbandonedExecutionControl } from "./services/execution-control-reconciliation.js";
 import { acknowledgeNativeStartupCleanup } from "./services/native-runtime/native-startup-cleanup.js";
 import { executionControlSchedule, registerExecutionControlWork } from "./services/execution-control-work.js";
@@ -1344,7 +1344,7 @@ async function startServerWithDatabaseTeardown(
     { queue: DELIVERY_QUEUES.executionStatus, run: () => deliverExecutionStatuses(db), next: executionSchedule.status },
     { queue: DELIVERY_QUEUES.executionDisposition, run: () => settleUnrecoverableExecutions(db), next: executionSchedule.disposition },
     ...(heartbeat ? [
-      { queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement },
+      { queue: DELIVERY_QUEUES.executionReplacement, run: () => reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession: run => verifyStoppedNativeSessionForReplacement(db, run) }), next: executionSchedule.replacement, onSettled: () => invalidateNativeReplacementDiscovery(db), settleEmptyWhilePaused: true },
       { queue: DELIVERY_QUEUES.executionContinuation, run: async () => {
         await deliverReconciledExecutions(db, heartbeat.wakeup);
         await deliverPendingPauseResumes(db, heartbeat.wakeup);

server/src/services/delivery-work-coordinator.ts

@@ -25,7 +25,7 @@ export function createDeliveryWorkCoordinator(input: {
       // do not postpone the obligation indefinitely.
       coalesceMs?: number;
       run: (signal: AbortSignal) => Promise<unknown>;
-      hasPending: () => Promise<boolean>;
+      hasPending: (runAllowed?: boolean) => Promise<boolean>;
       // Scoped hints belong to the outer transaction, including uncertainty recovery.
       onSettled?: (hints?: ReadonlySet<string>) => void;
       // Optional maintenance/retry deadline restored from durable state.
@@ -75,7 +75,7 @@ export function createDeliveryWorkCoordinator(input: {
         let runFailed = false;
         controller = attempt;
         running = withDatabaseOperation(`delivery.${queue}`, () => Promise.resolve().then(() => reconcileDatabaseWork(input.owner, queue))
-          .then(() => workerStopped || !(input.canRun() || (task.runDuringDrain && input.canReconcile?.())) ? undefined : task.run(attempt.signal).catch(error => { runFailed = true; throw error; })).then(() => workerStopped ? false : task.hasPending()).then(pending => {
+          .then(() => workerStopped || !(input.canRun() || (task.runDuringDrain && input.canReconcile?.())) ? undefined : task.run(attempt.signal).catch(error => { runFailed = true; throw error; })).then(() => workerStopped ? false : task.hasPending(input.canRun() || Boolean(task.runDuringDrain && input.canReconcile?.()))).then(pending => {
           pendingWork = pending || databaseWorkPending(input.owner, queue);
           if (pendingWork) schedule(task.retryMs);
           else { finishIntent?.(); finishIntent = undefined; }

server/src/services/environment-cleanup-notifications.ts

@@ -31,7 +31,10 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Envi
   await notifyNativeEnvironmentSettlement(transaction, scope);
   await notifyQueuedRunDispatch(transaction, scope);
   // A settled run lease can make that run's abandoned native owner releasable.
-  if (scope?.runId) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionFinalization);
+  if (scope?.runId) {
+    await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionFinalization);
+    await notifyDeliveryWork(transaction, DELIVERY_QUEUES.executionReplacement);
+  }
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup);
 }

server/src/services/execution-control-work.ts

@@ -5,7 +5,7 @@ import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.
 import { type DeliveryQueue } from "./delivery-work-notifications.js";
 import { EXECUTION_RECONCILIATION_INTERVAL_MS } from "./execution-control-deadline.js";
 import { obsoleteExecutionConversationHold, unresolvedExecutionDispositions } from "./execution-recovery-resolution.js";
-import { nativeReplacementPending } from "./native-runtime/native-safe-replacement.js";
+import { hasPendingNativeReplacements } from "./native-runtime/native-safe-replacement.js";
 import { nextAbandonedNativeOwnerReleaseAt } from "./native-runtime/native-failed-run-owner-release.js";
 import { hasPendingPauseResume } from "./issue-tree-resume.js";
 
@@ -16,6 +16,7 @@ export function registerExecutionControlWork(
     run: () => Promise<unknown>;
     next: () => Promise<number | null>;
     onSettled?: () => void;
+    settleEmptyWhilePaused?: boolean;
   }>,
   canRun: () => boolean = () => true,
 ) {
@@ -24,6 +25,7 @@ export function registerExecutionControlWork(
     let reconciled = 0;
     let next: number | null = null;
     let earliestRunAt = 0;
+    let pausedGeneration = 0, pausedPending = true;
     const worker = coordinator.register(duty.queue, {
       retryMs: EXECUTION_RECONCILIATION_INTERVAL_MS,
       pausePassiveMaintenance: true,
@@ -42,7 +44,18 @@ export function registerExecutionControlWork(
         next = deadline === null ? null : Math.max(deadline, earliestRunAt);
         reconciled = generation;
       },
-      hasPending: async () => reconciled !== requested,
+      hasPending: async (runAllowed = true) => {
+        if (reconciled === requested) return false;
+        if (runAllowed || !duty.settleEmptyWhilePaused || !canRun() || reconciled === 0) return true;
+        if (pausedGeneration !== requested) {
+          const generation = requested;
+          const deadline = await duty.next();
+          if (generation !== requested) return true;
+          pausedGeneration = generation; pausedPending = deadline !== null; next = deadline;
+          if (!pausedPending) reconciled = generation;
+        }
+        return pausedPending;
+      },
       nextRunAt: () => next,
     });
     return { ...worker, wake: () => { requested += 1; worker.wake(); } };
@@ -74,10 +87,7 @@ export function executionControlSchedule(db: Db) {
       return at === null ? null : deadline([new Date(at)]);
     },
     async replacement() {
-      const rows = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
-        .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
-        .where(nativeReplacementPending()).limit(1);
-      return rows.length ? retry() : null;
+      return await hasPendingNativeReplacements(db) ? retry() : null;
     },
     async continuation() {
       const rows = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(

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

@@ -1586,7 +1586,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
     await persistHeartbeatRunProcessMetadata(db, source.id, { pid: 999999999, processGroupId: null, startedAt: new Date().toISOString() });
     await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id));
     expect(await admit(f, true)).toBeNull();
-    expect(await recordNativeLocalProcessStop(db, source)).toBe(true);
+    expect(await db.transaction(tx => recordNativeLocalProcessStop(tx as unknown as typeof db, source))).toBe(true);
     await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: source.id, agentId: f.agentId,
       eventType: PROCESS_START_REQUESTED });
     // No PID was stored for the new launch, as when the server dies after spawn.
@@ -2440,7 +2440,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
         await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: "board", conversationState: "active", conversationBoundaryCommentId: boundaryId }).where(eq(issues.id, f.issueId));
         await db.update(issueRecoveryActions).set({ createdAt: new Date("2026-09-11T09:00:00Z") }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
         const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
-        expect(await recordNativeLocalProcessStop(db, source)).toBe(true);
+        expect(await db.transaction(tx => recordNativeLocalProcessStop(tx as unknown as typeof db, source))).toBe(true);
         await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, f.sourceRunId));
         if (scenario === "after_reset_workspace_hold") {
           await db.update(heartbeatRuns).set({ processPid: 999999999, resultJson: { workspaceRestoreFailure: "restore_unsafe_archive" } }).where(eq(heartbeatRuns.id, f.sourceRunId));

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

@@ -6,6 +6,7 @@ import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
 import { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js";
 import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
 import os from "node:os";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 export const PROCESS_START_REQUESTED = "native.process_start_requested";
 export const PROCESS_IDENTITY_RECORDED = "native.process_identity_recorded";
@@ -56,6 +57,8 @@ export async function recordNativeLocalProcessStop(db: Db, run: Pick<typeof hear
   if (leases.some(lease => lease.provider !== "local")) return false;
   if ((run.processPid && !absent(run.processPid)) ||
       (run.processGroupId && !absent(-run.processGroupId))) return false;
+  await notifyDeliveryWork(db, DELIVERY_QUEUES.executionReplacement,
+    JSON.stringify({ companyId: run.companyId, runId: run.id }));
   await appendHeartbeatRunEvent(db, {
     companyId: run.companyId,
     runId: run.id,

server/src/services/native-runtime/native-case-handoff.test.ts

@@ -84,7 +84,7 @@ describe("native persisted case handoff runtime", () => {
     expect(await isCaseHandoffSourceReleased(db, f.input)).toBe(false);
     const exited = once(child, "exit"); child.kill("SIGTERM"); await exited;
     const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
-    expect(await recordNativeLocalProcessStop(db, run!)).toBe(true);
+    expect(await db.transaction(tx => recordNativeLocalProcessStop(tx as unknown as typeof db, run!))).toBe(true);
     // A new worker has no process-local state and reconstructs the release.
     expect(await isCaseHandoffSourceReleased(db, f.input)).toBe(true);
     expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]?.resultJson).toHaveProperty("caseHandoffExecutionReleased");

server/src/services/native-runtime/native-runner-restart-preparation.integration.test.ts

@@ -1,11 +1,21 @@
 import { startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js";
 import { randomUUID } from "node:crypto";
-import { resolve } from "node:path";
+import { resolve, join } from "node:path";
+import { mkdir, readFile, writeFile } from "node:fs/promises";
 import { eq, sql } from "drizzle-orm";
-import { environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
+import { environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, issueRecoveryActions, issues, completionContracts, nativeRunResults, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
 import { beforeAll, expect, it, vi } from "vitest";
-import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
+import { spawn } from "node:child_process";
+import { createRequire } from "node:module";
+import { DurablePrpControlPlane, createNativeSessionBackend, createRunnerdCodexTransport, defaultCapabilityRunnerdBinary, canonicalNativeRuntimeContextDigest } from "../../vendor/paperclip-runner/index.js";
 import { buildNativeExecutionInput } from "./native-execution-input.js";
+import { buildNativeCompletionContract } from "./completion-contracts.js";
+import { reconcileNativeFinalizations, claimNativeSessionResumptions } from "./native-finalization-reconciler.js";
+import { prepareNativeWorkspaceSync } from "./native-workspace-sync.js";
+import { workspaceOperationService } from "../workspace-operations.js";
+import type { EnvironmentLease } from "@paperclipai/shared";
+import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
+import { materializeAsset } from "./runtime-context.js";
 import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
 import { executePaperclipNativeSession } from "./native-session-executor.js";
 import { registerNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
@@ -21,6 +31,147 @@ describeEmbeddedPostgres("native runner restart recovery: preparation", () => {
   let successor: typeof environment.current.successor;
   beforeAll(() => { ({ runtimeRoot, companyId, agentId, successor } = environment.current); });
 
+  it.each(["recover", "provider event", "cancelled task"] as const)("reconciles the drained remote backend with %s", async mode => {
+    const fixture = await seedRun(`DRAINED-BACKEND-${mode.replaceAll(" ", "-")}`);
+    const environmentId = randomUUID(), leaseId = randomUUID();
+    await fixture.db.insert(environments).values({ id: environmentId, name: `Drained backend ${environmentId}`, driver: "sandbox" });
+    const [lease] = await fixture.db.insert(environmentLeases).values({ id: leaseId, companyId, environmentId,
+      heartbeatRunId: fixture.runId, issueId: fixture.issueId, status: "active",
+      provider: "fixture", providerLeaseId: `drained-backend-${leaseId}` }).returning();
+    const [issue] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId));
+    const [contract] = await fixture.db.select().from(completionContracts).where(eq(completionContracts.issueId, fixture.issueId));
+    const runtimeContext = nativeRuntimeContextFixture();
+    runtimeContext.instructions.bundle = await materializeAsset([{ path: "AGENTS.md", mode: 0o444,
+      content: Buffer.from("Complete the controlled restart fixture task.\n") }]);
+    runtimeContext.aggregateDigest = canonicalNativeRuntimeContextDigest(runtimeContext);
+    const sandbox = join(runtimeRoot, `workspace-${fixture.runId}`);
+    await mkdir(sandbox, { recursive: true });
+    const savedFile = join(sandbox, "saved-before-drain.txt");
+    await writeFile(savedFile, "Preserve admitted work.\n");
+    const execution = buildNativeExecutionInput({ companyId, agentId, runId: fixture.runId,
+      issue: { id: fixture.issueId, identifier: "NRR-DRAINED", title: "Continue after drain", description: null, workMode: "standard" },
+      taskPrompt: "Continue the admitted work once.", provider: "codex",
+      workspace: { id: fixture.runId, cwd: sandbox, repoUrl: null, repoRef: null, branchName: null },
+      normalizedSessionId: fixture.native.normalizedSessionId, runtimeContext,
+      completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
+        contract: buildNativeCompletionContract(issue, { revision: contract.revision }) },
+    });
+    await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: {
+      ...fixture.preparedProfile, nativeExecutionInput: execution,
+    } }).where(eq(heartbeatRuns.id, fixture.runId));
+    const remoteSandbox = join(runtimeRoot, `remote-${fixture.runId}`);
+    const shell: CommandManagedRuntimeRunner = { execute: async input => {
+      const startedAt = new Date().toISOString();
+      const child = spawn(input.command, input.args ?? [], { cwd: input.cwd,
+        env: { ...process.env, ...input.env }, stdio: ["pipe", "pipe", "pipe"] });
+      let stdout = "", stderr = "", timedOut = false;
+      child.stdout.setEncoding("utf8").on("data", chunk => { stdout += chunk; });
+      child.stderr.setEncoding("utf8").on("data", chunk => { stderr += chunk; });
+      child.stdin.on("error", () => undefined);
+      child.stdin.end(input.stdin);
+      const timeout = setTimeout(() => { timedOut = true; child.kill("SIGKILL"); }, input.timeoutMs ?? 30_000);
+      try { return await new Promise((resolveResult, reject) => {
+        child.once("error", reject);
+        child.once("close", (exitCode, signal) => resolveResult({ stdout, stderr, exitCode, signal, timedOut,
+          pid: child.pid ?? null, startedAt }));
+      }); } finally { clearTimeout(timeout); }
+    } };
+    const workspaceSync = await prepareNativeWorkspaceSync({ db: fixture.db, runId: fixture.runId, companyId,
+      workspaceId: fixture.runId, workspaceLocalDir: sandbox, lease: lease as unknown as EnvironmentLease,
+      target: { kind: "remote", transport: "sandbox", environmentId, leaseId, remoteCwd: remoteSandbox, runner: shell } });
+    expect(workspaceSync).not.toBeNull();
+    const child = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"),
+      resolve(import.meta.dirname, "../../__tests__/helpers/native-drain-controller.test-worker.ts")],
+      { stdio: ["ignore", "pipe", "pipe", "ipc"], env: process.env });
+    const logs: string[] = [];
+    child.stdout?.on("data", data => logs.push(String(data)));
+    child.stderr?.on("data", data => logs.push(String(data)));
+    const exited = new Promise<number | null>(resolveExit => child.once("exit", resolveExit));
+    try {
+      const result = await new Promise<{ commands: number; errorName: string; errorMessage: string }>((resolveResult, reject) => {
+        const deadline = setTimeout(() => reject(new Error(`Controller handoff timed out: ${logs.join("").slice(-2000)}`)), 60_000);
+        child.once("error", error => { clearTimeout(deadline); reject(error); });
+        child.once("exit", code => { clearTimeout(deadline); reject(new Error(`Controller exited before handoff (${code}): ${logs.join("").slice(-2000)}`)); });
+        child.on("message", (message: { ready?: boolean; done?: boolean; commands: number; errorName: string; errorMessage: string; error?: string }) => {
+          if (message.ready) child.send({ connectionString: environment.current.temporary.connectionString, providerEvidence: mode === "provider event",
+            options: { execution, runnerInstanceId: fixture.native.runnerInstanceId, useRunnerd: true,
+              runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId, leaseId, remoteCwd: remoteSandbox } } });
+          if (message.done) { clearTimeout(deadline); resolveResult(message); }
+          if (message.error) { clearTimeout(deadline); reject(new Error(message.error)); }
+        });
+      });
+      expect(result).toMatchObject({ commands: 0, errorMessage: "local_runtime_api_listener_unavailable" });
+      expect(await exited).toBe(0);
+      const [handoff] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
+      if (mode === "provider event") {
+        expect(handoff).toMatchObject({ phase: "terminal_failure", nextAttemptAt: null,
+          failureDetail: { recoveryMode: "ambiguous_state", providerSessionEstablished: true } });
+        expect(await claimNativeSessionResumptions({ db: fixture.db, runnerInstanceId: "no-provider-replay",
+          runIds: [fixture.runId], deferIneligible: false })).toHaveLength(0);
+        return;
+      }
+      expect(handoff).toMatchObject({ phase: "retryable_failure", attempt: 1,
+        failureDetail: { originalFailureCode: "local_runtime_api_listener_unavailable", recoveryMode: "bootstrap_retry",
+          providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false } });
+      if (mode === "cancelled task") {
+        await fixture.db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, fixture.issueId));
+      }
+      const claims = (await Promise.all(["successor-a", "successor-b"].map(runnerInstanceId =>
+        claimNativeSessionResumptions({ db: fixture.db, runnerInstanceId, runIds: [fixture.runId],
+          now: new Date(handoff.nextAttemptAt!.getTime() + 1), deferIneligible: false }),
+      ))).flat();
+      if (mode === "cancelled task") { expect(claims).toHaveLength(0); return; }
+      expect(claims).toHaveLength(1);
+      const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+      expect(run).toMatchObject({ status: "running", nativeSessionId: fixture.native.normalizedSessionId });
+      const [action] = await fixture.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId));
+      expect(action).toMatchObject({ ownerType: "agent", ownerAgentId: agentId });
+      if (process.env.PAPERCLIP_TEST_DRAIN_CONTINUATION === "1") {
+        const claim = claims[0];
+        const root = join(runtimeRoot, `continued-${fixture.runId}`);
+        const home = join(root, "codex-home");
+        await mkdir(home, { recursive: true });
+        const report = join(root, "result.json");
+        const calls = join(root, "calls.log");
+        await writeFile(report, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
+          summary: "Admitted work completed after controller restart.", completionClaim: {
+            contractRevision: execution.completionContract.contract.revision, objectiveSatisfied: true,
+            criteria: execution.completionContract.contract.criteria.map(criterion => ({ criterionId: criterion.id,
+              status: "satisfied", evidenceRefs: [] })), remainingWork: [] },
+          evidence: [], verification: [], artifacts: [], attentionRequests: [] }));
+        const bundle = createRunnerdCodexTransport({ ...transportOptions(fixture, root),
+          runnerBinary: defaultCapabilityRunnerdBinary(),
+          codexArgs: ["--state-file", join(root, "provider.json"), "--terminal-result-file", report,
+            "--call-log", calls, "--record-process-start"],
+          runtimeContext: execution.runtimeContext, runnerRuntimeContext: execution.runtimeContext,
+          sourceCodexHome: home, environment: { CODEX_HOME: home, PATH: process.env.PATH },
+          lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
+        });
+        const backend = createNativeSessionBackend(execution, { runnerInstanceId: fixture.native.runnerInstanceId,
+          environment: { CODEX_HOME: home, PATH: process.env.PATH }, codexTransportFactory: () => bundle.transport });
+        try {
+          expect(await executePaperclipNativeSession({ db: fixture.db, execution, backend, useRunnerd: true,
+            runnerInstanceId: fixture.native.runnerInstanceId, leaseOwner: claim.leaseOwner,
+            runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId, leaseId, remoteCwd: remoteSandbox,
+              runner: { execute: async () => { throw new Error("Unexpected remote preparation after the controlled backend owns execution"); } } as never },
+            turnTimeoutMs: 15_000 })).toMatchObject({ exitCode: 0, timedOut: false });
+          const recorder = workspaceOperationService(fixture.db).createRecorder({ companyId,
+            issueId: fixture.issueId, heartbeatRunId: fixture.runId });
+          const saved = await recorder.recordOperation({ phase: "workspace_finalize", cwd: sandbox,
+            run: async () => { await workspaceSync!.restoreWorkspace(); return { status: "succeeded", exitCode: 0 }; } });
+          expect(saved.status).toBe("succeeded");
+          await reconcileNativeFinalizations(fixture.db, [fixture.runId]);
+          expect(await fixture.db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, fixture.runId)))
+            .toMatchObject([{ schemaStatus: "accepted" }]);
+          const [finished] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId));
+          expect(finished, JSON.stringify(finished.failureDetail)).toMatchObject({ phase: "committed", attempt: 2 });
+          expect(await readFile(savedFile, "utf8")).toBe("Preserve admitted work.\n");
+          expect((await readFile(calls, "utf8")).split("\n").filter(line => line === "process-start")).toHaveLength(1);
+        } finally { await bundle.transport.close(); }
+      }
+    } finally { child.kill("SIGKILL"); await exited; await workspaceSync?.cleanup(); }
+  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
+
   it.each(["preparation", "claim lock"])("gives a fresh remote claim its full lease through blocked %s", async (blockedPhase) => {
     const fixture = await seedRun(`FRESH-SLOW-${blockedPhase === "claim lock" ? "CLAIM-LOCK" : "PREPARATION"}`);
     const environmentId = randomUUID(), leaseId = randomUUID();

server/src/services/native-runtime/native-safe-replacement.test.ts

@@ -16,6 +16,11 @@ import {
   deliverReconciledExecutions,
 } from "../execution-recovery-resolution.js";
 import { randomUUID } from "node:crypto";
+import { spawn } from "node:child_process";
+import { persistHeartbeatRunProcessMetadata } from "../heartbeat/run-lifecycle.js";
+import { hasNativeLocalProcessStop, releaseSettledNativeRunProcess } from "../native-local-process-stop.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import { executionControlSchedule } from "../execution-control-work.js";
 import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
 import { tmpdir } from "node:os";
 import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
@@ -26,6 +31,7 @@ import {
   agents,
   companies,
   createDb,
+  subscribeDatabaseWork,
   heartbeatRunEvents,
   heartbeatRuns,
   issueRecoveryActions,
@@ -41,7 +47,7 @@ import {
   getEmbeddedPostgresTestSupport,
   startEmbeddedPostgresTestDatabase,
 } from "../../__tests__/helpers/embedded-postgres.js";
-import { reconcileSafeNativeReplacements } from "./native-safe-replacement.js";
+import { invalidateNativeReplacementDiscovery, reconcileSafeNativeReplacements } from "./native-safe-replacement.js";
 import { reconcileAbandonedExecutionControl } from "../execution-control-reconciliation.js";
 const externalDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL;
 const support = externalDatabaseUrl
@@ -219,6 +225,103 @@ const support = externalDatabaseUrl
         }
       },
     );
+    it("wakes a completed replacement scan after actual local stop proof commits", async () => {
+      const source = await seed(2);
+      const workspace = await mkdtemp(join(tmpdir(), "paperclip-late-stop-"));
+      const child = spawn(process.execPath, ["-e", "process.exit(0)"], { detached: true, stdio: "ignore" });
+      const pid = child.pid!;
+      await new Promise<void>((resolve, reject) => { child.once("exit", () => resolve()); child.once("error", reject); });
+      await persistHeartbeatRunProcessMetadata(db, source.runId, { pid, processGroupId: pid,
+        startedAt: new Date().toISOString(), targetKind: "local" });
+      await db.update(heartbeatRuns).set({ finishedAt: new Date(), runnerProfileJson: {
+        nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: workspace } },
+      } }).where(eq(heartbeatRuns.id, source.runId));
+      await db.update(nativeRunFinalizations).set({ failureCode: "provider_transport_failed" })
+        .where(eq(nativeRunFinalizations.runId, source.runId));
+      await appendHeartbeatRunEvent(db, { companyId: source.companyId, runId: source.runId, agentId: source.agentId,
+        eventType: "harness.ready", stream: "system", message: "Provider started before its transport failed." });
+      const verifyStoppedSession = async () => await hasNativeLocalProcessStop(db, source.companyId, source.runId)
+        ? { evidence: {}, retire: () => true } : null;
+      invalidateNativeReplacementDiscovery(db);
+      expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 0 });
+      expect(await executionControlSchedule(db).replacement()).toBeNull();
+      let wakes = 0;
+      const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => {
+        if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); }
+      });
+      try {
+        await db.transaction(async tx => {
+          expect(await releaseSettledNativeRunProcess(tx as unknown as typeof db, source.runId)).toBe("released");
+          expect(wakes).toBe(0);
+        });
+        expect(wakes).toBeGreaterThan(0);
+        expect(await executionControlSchedule(db).replacement()).not.toBeNull();
+        expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 1 });
+        expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 0 });
+        expect(await executionControlSchedule(db).replacement()).toBeNull();
+        const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId));
+        expect(successors).toHaveLength(1);
+      } finally { unsubscribe(); await rm(workspace, { recursive: true, force: true }); }
+    });
+    it("wakes completed recovery when the real task becomes eligible", async () => {
+      const source = await seed();
+      const workspace = await mkdtemp(join(tmpdir(), "paperclip-late-eligibility-"));
+      await db.update(heartbeatRuns).set({ runnerProfileJson: { recoveryEventInventoryVersion: 1,
+        nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: workspace } },
+      } }).where(eq(heartbeatRuns.id, source.runId));
+      await issueService(db).update(source.issueId, { status: "backlog" });
+      invalidateNativeReplacementDiscovery(db);
+      expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 });
+      expect(await executionControlSchedule(db).replacement()).toBeNull();
+      let wakes = 0;
+      const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => {
+        if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); }
+      });
+      try {
+        await issueService(db).update(source.issueId, { status: "in_progress" });
+        expect(wakes).toBeGreaterThan(0);
+        expect(await executionControlSchedule(db).replacement()).not.toBeNull();
+        expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 1 });
+        expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 });
+        expect(await executionControlSchedule(db).replacement()).toBeNull();
+        expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId))).toHaveLength(1);
+      } finally { unsubscribe(); await rm(workspace, { recursive: true, force: true }); }
+    });
+    it("advances past a failed full page and retries it after later work progresses", async () => {
+      const workspace = await mkdtemp(join(tmpdir(), "paperclip-failed-page-"));
+      const sources: Awaited<ReturnType<typeof seed>>[] = [];
+      for (let index = 0; index < 26; index++) {
+        const source = await seed(); sources.push(source);
+        await db.update(heartbeatRuns).set({ runnerProfileJson: {
+          nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: workspace } },
+        } }).where(eq(heartbeatRuns.id, source.runId));
+        await db.update(nativeRunFinalizations).set({ failureCode: "provider_transport_failed" })
+          .where(eq(nativeRunFinalizations.runId, source.runId));
+      }
+      sources.sort((a, b) => a.runId.localeCompare(b.runId));
+      const later = sources.at(-1)!;
+      let failing = true;
+      const verifyStoppedSession = vi.fn(async (run: typeof heartbeatRuns.$inferSelect) => {
+        if (failing && run.id !== later.runId) throw new Error("Controlled stopped-session verifier failure");
+        return { evidence: {}, retire: () => true };
+      });
+      invalidateNativeReplacementDiscovery(db);
+      try {
+        await expect(reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession }))
+          .rejects.toThrow("native_replacement_decision_retry_required");
+        expect(verifyStoppedSession).toHaveBeenCalledTimes(25);
+        expect(await executionControlSchedule(db).replacement()).not.toBeNull();
+        expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 1 });
+        expect(verifyStoppedSession).toHaveBeenCalledTimes(26);
+        expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, later.runId))).toHaveLength(1);
+        failing = false;
+        expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 25 });
+        expect(await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession })).toMatchObject({ scheduled: 0 });
+        expect(await executionControlSchedule(db).replacement()).toBeNull();
+        const successors = await db.select().from(heartbeatRuns).where(inArray(heartbeatRuns.retryOfRunId, sources.map(source => source.runId)));
+        expect(successors).toHaveLength(26);
+      } finally { await rm(workspace, { recursive: true, force: true }); }
+    }, 60_000);
     it.each(["unproven", "changed", "verified"] as const)("requires stopped-session proof through commit (%s)", async (mode) => {
       const source = await seed(2);
       await db.update(nativeRunFinalizations).set({ failureCode: "native_session_cleanup_quarantined" }).where(eq(nativeRunFinalizations.runId, source.runId));
@@ -235,8 +338,11 @@ const support = externalDatabaseUrl
         eventType: "tool.execution.started", stream: "system",
         payload: { name: "paperclip_finish", executionId: "completion", transport: "process" },
       });
-      await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession });
-      await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession });
+      for (let sweep = 0; sweep < 2; sweep++) {
+        const decision = reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession });
+        if (mode === "changed") await expect(decision).rejects.toThrow("native_replacement_decision_retry_required");
+        else await decision;
+      }
       const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId));
       expect(successors).toHaveLength(mode === "verified" ? 1 : 0);
       const [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, source.issueId));

server/src/services/native-runtime/native-safe-replacement.ts

@@ -5,7 +5,7 @@ import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { logger } from "../../middleware/logger.js";
 import { createHash, randomUUID } from "node:crypto";
 import { stat } from "node:fs/promises";
-import { and, eq, inArray, isNull, sql } from "drizzle-orm";
+import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm";
 import {
   agentWakeupRequests,
   environmentLeases,
@@ -27,6 +27,8 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js";
 import { buildExecutionContinuation } from "../execution-continuation.js";
 import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
 import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
+import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES } from "./native-retired-backup-source.js";
+import { completedWorkspaceRecoveryCondition } from "../workspace-recovery-receipt.js";
 
 export const NATIVE_SAFE_REPLACEMENT_REASON = "native_safe_replacement";
 const record = (v: unknown): Record<string, unknown> =>
@@ -45,6 +47,66 @@ const processAlive = (pid: number | null) => {
 
 type ReplacementDb = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
 
+const listenerUnavailable = "local_runtime_api_listener_unavailable";
+function historicalListenerGuardPending() {
+  const profile = heartbeatRuns.runnerProfileJson, detail = nativeRunFinalizations.failureDetail;
+  const providerEvents = sql.join([...NATIVE_PROVIDER_DURABLE_EVENT_TYPES].map(value => sql`${value}`), sql`, `);
+  return and(
+    eq(nativeRunFinalizations.failureCode, "native_session_interrupted"),
+    eq(heartbeatRuns.errorCode, "native_session_interrupted"), eq(heartbeatRuns.error, listenerUnavailable),
+    sql`${detail}->>'message' = ${listenerUnavailable}`,
+    sql`${detail}->>'originalFailureCode' = 'native_session_interrupted'`,
+    sql`${detail}->>'recoveryMode' = 'ambiguous_state'`,
+    sql`${detail}->'providerSessionEstablished' = 'false'::jsonb`,
+    sql`${detail}->'providerEventsExist' = 'false'::jsonb`,
+    sql`${detail}->'checkpointExists' = 'false'::jsonb`,
+    eq(nativeRunFinalizations.attempt, 1),
+    isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt),
+    isNull(nativeRunFinalizations.assessmentId), isNull(nativeRunFinalizations.decisionId),
+    isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId),
+    isNull(heartbeatRuns.sessionIdBefore), isNull(heartbeatRuns.sessionIdAfter), isNull(heartbeatRuns.retryOfRunId),
+    sql`not (coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ?| array['cancellation', 'nativeCancellation'])`,
+    sql`coalesce(${profile}->'sessionCheckpoint', 'null'::jsonb) = 'null'::jsonb`,
+    sql`coalesce(${profile}->'nativeProviderPackAdmission', 'null'::jsonb) = 'null'::jsonb`,
+    sql`coalesce(${profile}#>>'{nativeProviderPackAdmissionState,state}', 'pending') = 'pending'`,
+    sql`${profile}#>>'{nativeExecutionInput,binding,runId}' = ${heartbeatRuns.id}::text`,
+    sql`${profile}#>>'{nativeExecutionInput,binding,companyId}' = ${heartbeatRuns.companyId}::text`,
+    sql`${profile}#>>'{nativeExecutionInput,binding,agentId}' = ${heartbeatRuns.agentId}::text`,
+    sql`${profile}#>>'{nativeExecutionInput,binding,issueId}' = ${nativeRunFinalizations.issueId}::text`,
+    sql`${profile}#>>'{nativeExecutionInput,session,normalizedSessionId}' = ${heartbeatRuns.nativeSessionId}::text`,
+    sql`not exists (select 1 from ${heartbeatRuns} prior where prior.company_id = ${heartbeatRuns.companyId}
+      and prior.native_session_id = ${heartbeatRuns.nativeSessionId} and prior.id <> ${heartbeatRuns.id})`,
+    sql`not exists (select 1 from ${heartbeatRunEvents} event where event.company_id = ${heartbeatRuns.companyId}
+      and event.run_id = ${heartbeatRuns.id} and (event.source_event_id is not null
+        or event.event_type in (${providerEvents}) or event.event_type ~ '^(turn|item|tool|semantic_tool|run[.]result)[.]'))`,
+    sql`not exists (select 1 from ${toolInvocations} invocation where invocation.company_id = ${heartbeatRuns.companyId}
+      and invocation.run_id = ${heartbeatRuns.id})`,
+    sql`not exists (select 1 from ${issueRecoveryActions}
+      where ${issueRecoveryActions.companyId} = ${heartbeatRuns.companyId}
+        and ${issueRecoveryActions.sourceIssueId} = ${nativeRunFinalizations.issueId}
+        and ${issueRecoveryActions.cause} = 'native_continuation_requires_reconciliation'
+        and ${issueRecoveryActions.evidence}->>'runId' = ${heartbeatRuns.id}::text
+        and ${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'
+        and not coalesce((${completedWorkspaceRecoveryCondition(heartbeatRuns.id)}), false))`,
+    sql`exists (select 1 from ${issues} task where task.company_id = ${heartbeatRuns.companyId}
+      and task.id = ${nativeRunFinalizations.issueId} and task.assignee_agent_id = ${heartbeatRuns.agentId}
+      and task.status in ('todo', 'in_progress', 'in_review', 'blocked')
+      and (task.status <> 'blocked' or exists (select 1 from ${issueRecoveryActions} hold
+        where hold.company_id = task.company_id and hold.source_issue_id = task.id and hold.kind = 'active_run_watchdog'
+          and hold.cause = ${nativeRunFinalizations.failureCode} and hold.evidence->>'runId' = ${heartbeatRuns.id}::text
+          and hold.evidence#>>'{nativeFailureBlock,runId}' = ${heartbeatRuns.id}::text
+          and hold.evidence#>'{nativeFailureBlock,statusVersion}' = to_jsonb(task.status_version))))`,
+  );
+}
+
+async function historicalListenerGuardEvidence(db: ReplacementDb, run: typeof heartbeatRuns.$inferSelect) {
+  const [evidence] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns)
+    .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)).where(and(
+      eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), historicalListenerGuardPending(),
+    )).limit(1);
+  return Boolean(evidence);
+}
+
 /** A saved remote source is proved by its current ledger, never a controller path. */
 async function remoteWorkspaceReplacementEvidence(
   db: ReplacementDb,
@@ -113,16 +175,95 @@ export function nativeReplacementPending() {
   return and(
     eq(heartbeatRuns.status, "failed"),
     eq(nativeRunFinalizations.phase, "terminal_failure"),
-    inArray(
+    or(inArray(
       nativeRunFinalizations.failureCode,
       ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"],
-    ),
+    ), historicalListenerGuardPending()),
     isNull(nativeRunFinalizations.resultId),
     sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`,
     sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`,
   );
 }
 
+const REPLACEMENT_SCAN_LIMIT = 100;
+const replacementDiscovery = new WeakMap<object, { after: string | undefined; complete: boolean; generation: number; rescan: boolean }>();
+function replacementDiscoveryState(db: Db) {
+  let state = replacementDiscovery.get(db);
+  if (!state) { state = { after: undefined, complete: false, generation: 0, rescan: false }; replacementDiscovery.set(db, state); }
+  return state;
+}
+
+export function invalidateNativeReplacementDiscovery(db: Db) {
+  const state = replacementDiscoveryState(db);
+  state.complete = false; state.rescan = true; state.generation++;
+}
+
+async function replacementHistoryReady(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string, retryDatabaseErrors = false) {
+  try {
+    const continuation = await buildExecutionContinuation({ db, companyId: run.companyId, issueId,
+      agentId: run.agentId, context: record(run.contextSnapshot), summary: null, exposeLowTrustRaw: false });
+    return continuation.unresolvedInteractionIds.length === 0;
+  } catch (error) {
+    let cause: unknown = error;
+    for (let depth = 0; retryDatabaseErrors && depth < 4 && cause && typeof cause === "object"; depth++) {
+      const failure = cause as { code?: unknown; cause?: unknown };
+      if (typeof failure.code === "string" && /^[A-Z0-9_]+$/.test(failure.code)) throw error;
+      cause = failure.cause;
+    }
+    return null;
+  }
+}
+
+async function discoverNativeReplacements(db: Db, after: string | undefined, limit = 25) {
+  const candidates: Array<{ run: typeof heartbeatRuns.$inferSelect; coordinator: typeof nativeRunFinalizations.$inferSelect }> = [];
+  let scanned = 0;
+  for (;;) {
+    const rows = await db.select({ run: heartbeatRuns, coordinator: nativeRunFinalizations }).from(heartbeatRuns)
+      .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
+      .where(and(nativeReplacementPending(), after ? gt(heartbeatRuns.id, after) : undefined))
+      .orderBy(asc(heartbeatRuns.id)).limit(Math.min(25, REPLACEMENT_SCAN_LIMIT - scanned));
+    if (!rows.length) return { candidates, scanned, after: undefined };
+    for (const row of rows) {
+      after = row.run.id; scanned++;
+      let ready = true;
+      if (row.coordinator.failureCode === "native_session_interrupted") {
+        ready = false;
+        if (row.run.runnerProfileJson?.remoteWorkspace != null) {
+          const leases = await db.select().from(environmentLeases).where(and(
+            eq(environmentLeases.companyId, row.run.companyId), eq(environmentLeases.heartbeatRunId, row.run.id),
+          ));
+          if (leases.every(lease => lease.releasedAt !== null)) {
+            const evidence = await remoteWorkspaceReplacementEvidence(db, row.run, row.coordinator.issueId, leases);
+            ready = Boolean(evidence?.workspacePreserved && evidence.providerStopped)
+              && await replacementHistoryReady(db, row.run, row.coordinator.issueId, true) === true;
+          }
+        }
+      }
+      if (ready) candidates.push(row);
+      if (candidates.length >= limit || scanned >= REPLACEMENT_SCAN_LIMIT) return { candidates, scanned, after };
+    }
+    if (rows.length < 25) return { candidates, scanned, after: undefined };
+  }
+}
+
+function settleReplacementDiscovery(state: ReturnType<typeof replacementDiscoveryState>, after: string | undefined, generation: number) {
+  const rescan = state.rescan || generation !== state.generation;
+  state.after = after; state.complete = !after && !rescan;
+  state.rescan = Boolean(after && rescan);
+}
+
+export async function hasPendingNativeReplacements(db: Db) {
+  const state = replacementDiscoveryState(db);
+  if (state.complete) return false;
+  if (state.after) return true;
+  state.rescan = false;
+  const generation = state.generation;
+  const result = await discoverNativeReplacements(db, undefined, 1);
+  if (result.candidates.length) return true;
+  settleReplacementDiscovery(state, result.after, generation);
+  return !state.complete;
+}
+
 /** Retryable outbox: terminal_failure plus a new-format failure code is the durable candidate. */
 export async function reconcileSafeNativeReplacements(
   db: Db,
@@ -136,18 +277,14 @@ export async function reconcileSafeNativeReplacements(
     failpoint?: (phase: "successor_inserted" | "lineage_committed") => void;
   } = {},
 ) {
-  const candidates = await db
-    .select({ run: heartbeatRuns, coordinator: nativeRunFinalizations })
-    .from(heartbeatRuns)
-    .innerJoin(
-      nativeRunFinalizations,
-      eq(nativeRunFinalizations.runId, heartbeatRuns.id),
-    )
-    .where(
-      nativeReplacementPending(),
-    )
-    .limit(25);
-  let scheduled = 0;
+  const discovery = replacementDiscoveryState(db);
+  if (discovery.complete) discovery.after = undefined;
+  discovery.complete = false;
+  if (!discovery.after) discovery.rescan = false;
+  const generation = discovery.generation;
+  const selection = await discoverNativeReplacements(db, discovery.after);
+  const candidates = selection.candidates;
+  let scheduled = 0, pendingFailures = 0;
   for (const { run, coordinator } of candidates) {
     try {
       if (
@@ -170,11 +307,13 @@ export async function reconcileSafeNativeReplacements(
         );
       // Teardown is still in progress. The next sweep rechecks its durable outcome.
       if (leases.some((lease) => lease.releasedAt === null)) continue;
-      const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed"
+      const historicalListener = coordinator.failureCode === "native_session_interrupted";
+      if (historicalListener && !await historicalListenerGuardEvidence(db, run)) continue;
+      const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener
         ? await options.verifyStoppedSession?.(run) ?? null : null;
       // A transport label alone is not evidence. Keep inspecting these candidates
       // as process cleanup and the final transcript become durable.
-      if (coordinator.failureCode !== "native_provider_terminal_failed" && !stoppedSession) continue;
+      if (coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener && !stoppedSession) continue;
       const invocations = await db
         .select()
         .from(toolInvocations)
@@ -263,23 +402,11 @@ export async function reconcileSafeNativeReplacements(
           ? [String(p.kind ?? p.type)]
           : [];
       });
-      let historyComplete = false;
-      try {
-        const continuation = await buildExecutionContinuation({
-          db,
-          companyId: run.companyId,
-          issueId: coordinator.issueId,
-          agentId: run.agentId,
-          context: record(run.contextSnapshot),
-          summary: null,
-          exposeLowTrustRaw: false,
-        });
-        if (continuation.unresolvedInteractionIds.length > 0) continue;
-        historyComplete = true;
-      } catch {
-        /* The named operator outcome retains incomplete context. */
-      }
+      const history = await replacementHistoryReady(db, run, coordinator.issueId);
+      if (history === false) continue;
+      const historyComplete = history === true;
       const remoteWorkspace = await remoteWorkspaceReplacementEvidence(db, run, coordinator.issueId, leases);
+      if (historicalListener && (!remoteWorkspace?.workspacePreserved || !remoteWorkspace.providerStopped || invocations.length)) continue;
       const decision = decideNativeReplacement({
         failedSession: true,
         // A facade transport failure can conceal an authoritative protocol
@@ -413,6 +540,13 @@ export async function reconcileSafeNativeReplacements(
           current.attempt >= 3
         )
           return false;
+        let historicalRun: typeof heartbeatRuns.$inferSelect | undefined;
+        if (historicalListener) {
+          const [currentRun] = await tx.select().from(heartbeatRuns).where(and(
+            eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
+          )).for("update");
+          historicalRun = currentRun;
+        }
         if (task.status === "blocked") {
           const failureHolds = await tx.select().from(issueRecoveryActions).where(and(
             eq(issueRecoveryActions.companyId, run.companyId),
@@ -427,6 +561,14 @@ export async function reconcileSafeNativeReplacements(
           });
           if (!ownsBlock) return false;
         }
+        const preservationHolds = historicalListener ? await tx.select().from(issueRecoveryActions).where(and(
+          eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, task.id),
+          eq(issueRecoveryActions.kind, "active_run_watchdog"),
+          eq(issueRecoveryActions.cause, "native_continuation_requires_reconciliation"),
+          sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
+          sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`,
+        )).for("update") : [];
+        if (historicalListener && (!historicalRun || !await historicalListenerGuardEvidence(tx, historicalRun))) return false;
         if (remoteWorkspace) {
           const currentWorkspace = await remoteWorkspaceReplacementEvidence(tx, run, coordinator.issueId, leases, true);
           if (!currentWorkspace?.workspacePreserved || !currentWorkspace.providerStopped) return false;
@@ -548,6 +690,7 @@ export async function reconcileSafeNativeReplacements(
           eq(issueRecoveryActions.cause, current.failureCode!),
           sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
         )).for("update");
+        holds.push(...preservationHolds);
         if (holds.some(hold => hold.status === "active" || hold.status === "escalated")) {
           await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId: task.id });
         }
@@ -587,11 +730,17 @@ export async function reconcileSafeNativeReplacements(
       }
     } catch (error) {
       if (options.failpoint) throw error;
+      pendingFailures++;
       logger.warn(
         { runId: run.id },
         "Native replacement remains pending; another sweep will retry its durable decision",
       );
     }
   }
-  return { scanned: candidates.length, scheduled };
+  settleReplacementDiscovery(discovery, selection.after, generation);
+  if (pendingFailures) {
+    discovery.complete = false; discovery.rescan = true;
+    throw new Error("native_replacement_decision_retry_required");
+  }
+  return { scanned: selection.scanned, scheduled };
 }

server/src/services/remote-workspaces.integration.test.ts

@@ -13,7 +13,7 @@ import { pathToFileURL } from "node:url";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { and, eq, sql } from "drizzle-orm";
 import { activityLog, agents, agentTaskSessions, agentWakeupRequests, applyPendingMigrations, closeRegisteredClients, companies, createDb, environmentLeases, environments,
-  executionWorkspaceRepositories, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, livenessDb, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations } from "@paperclipai/db";
+  subscribeDatabaseWork, executionWorkspaceRepositories, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issueThreadInteractions, issues, livenessDb, nativeRunFinalizations, projects, projectWorkspaces, workspaceOperations } from "@paperclipai/db";
 import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
 import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
 import { buildProjectMentionHref, type Environment, type EnvironmentLease } from "@paperclipai/shared";
@@ -25,12 +25,13 @@ import { createSavedWorkspaceLoader, prepareRemoteWorkspace, readRemoteWorkspace
   clearRemoteWorkspaceLocationForReleasedLease, createRemoteWorkspaceGitAuth, prepareRemoteWorkspaceRepository,
   reserveRemoteWorkspace, saveRemoteWorkspace, shouldUseRemoteWorkspaces } from "./remote-workspaces.js";
 import type { createGitRemoteAuthProvider } from "./git-credentials.js";
-import { reconcileSafeNativeReplacements } from "./native-runtime/native-safe-replacement.js";
+import { invalidateNativeReplacementDiscovery, reconcileSafeNativeReplacements } from "./native-runtime/native-safe-replacement.js";
 import { remoteTerminationReceipt } from "./remote-execution-termination.js";
 import { RemoteWorkspaceRestoreError } from "./remote-workspace-transport.js";
 import { readRemoteRunnerState } from "./native-runtime/native-session-executor.js";
 import { notFound } from "../errors.js";
 import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
+import { issueThreadInteractionService } from "./issue-thread-interactions.js";
 import { createRemoteWorkspaceFiles } from "./remote-workspace-files.js";
 import { workspaceFileResourceService } from "./workspace-file-resources.js";
 import { nativeWorkspaceExportPendingForLease, preserveRemoteWorkspacePreparationFailure } from "./native-runtime/native-workspace-finalization-state.js";
@@ -47,6 +48,10 @@ import { instanceSettingsService } from "./instance-settings.js";
 import { resolveRemoteWorkspaceSource } from "./heartbeat/remote-workspace-source.js";
 import { toExecutionWorkspace } from "./execution-workspaces.js";
 import { recoveryPostgresSupport, startRecoveryPostgres } from "../__tests__/helpers/recovery-postgres.js";
+import { getExecutionBlocker } from "./execution-blocker.js";
+import { executionControlSchedule } from "./execution-control-work.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { createSystemRecovery } from "./system-recovery-state.js";
 
 const exec = promisify(execFile);
 let databaseUrl = process.env.PAPERCLIP_REMOTE_WORKSPACE_TEST_DATABASE_URL;
@@ -82,6 +87,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
     await db.insert(issues).values({ id: ids.issue, companyId: ids.company, title: "Use a remote workspace", status: "in_progress", assigneeAgentId: ids.agent });
     await db.insert(heartbeatRuns).values({ id: ids.run, companyId: ids.company, agentId: ids.agent, nativeIssueId: ids.issue,
       runtimeMode: "native", status: "running" });
+    invalidateNativeReplacementDiscovery(db);
     await db.insert(environmentLeases).values({ id: ids.lease, companyId: ids.company, environmentId: ids.environment,
       heartbeatRunId: ids.run, issueId: ids.issue, provider: "daytona", providerLeaseId: `allocation-${ids.lease}`, status: "active" });
   });
@@ -91,6 +97,10 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
       ? (row.metadata.uploads as { objectKey: string }[]).map(item => item.objectKey) : [])].filter((value): value is string => !!value))) {
       await storage.deleteObject(ids.company, key);
     }
+    if (temporary && process.env.PAPERCLIP_TEST_HISTORICAL_LISTENER === "1") await db.transaction(async tx => {
+      await tx.execute(sql`select set_config('client_min_messages', 'warning', true)`);
+      await tx.execute(sql`TRUNCATE companies CASCADE`);
+    });
     await db.delete(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.companyId, ids.company));
     await db.delete(workspaceOperations).where(eq(workspaceOperations.companyId, ids.company));
     await db.delete(activityLog).where(eq(activityLog.companyId, ids.company));
@@ -1712,14 +1722,20 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
     } finally { await recovered.cleanup(); }
   }, 60_000);
 
-  it.each(["saved", "dirty", "unsaved", "foreign_snapshot", "delete_requested", "deletion_race", "controller_claim_race", "stale_epoch", "unconfirmed_stop", "unknown_action"] as const)(
-    "checks durable remote work at the native safe-replacement decision (%s)", async scenario => {
+  it.each(["saved", "dirty", "unsaved", "foreign_snapshot", "delete_requested", "deletion_race", "controller_claim_race", "stale_epoch", "unconfirmed_stop", "unknown_action", "historical_listener_guard", "historical_listener_provider_event", "historical_listener_checkpoint", "historical_listener_resumed", "historical_listener_other_message", "historical_listener_cancelled", "historical_listener_prior_session", "historical_listener_unconfirmed_stop", "historical_listener_dirty", "historical_listener_stale_block", "historical_listener_unknown_event", "historical_listener_budget", "historical_listener_bad_preservation", "historical_listener_absent", "historical_listener_later_candidate", "historical_listener_late_proof", "historical_listener_lock_timeout", "historical_listener_late_interaction", "historical_listener_interaction_paused", "historical_listener_own_question"] as const)(
+    "checks durable remote work at the native safe-replacement decision (%s)", async kind => {
+      const historicalListener = kind.startsWith("historical_listener_");
+      const processAcceptance = kind === "historical_listener_guard" && process.env.PAPERCLIP_TEST_HISTORICAL_LISTENER === "1";
+      const scenario = kind === "historical_listener_late_proof" ? "dirty" : kind === "historical_listener_unconfirmed_stop" ? "unconfirmed_stop"
+        : kind === "historical_listener_dirty" ? "dirty" : historicalListener ? "saved" : kind;
+      const shouldSchedule = scenario === "saved" && (!historicalListener || ["historical_listener_guard", "historical_listener_later_candidate", "historical_listener_lock_timeout"].includes(kind));
       const reserved = await reserve({ kind: "empty" }), first = await target("native-replacement-sandbox");
       const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run,
         lease: await lease(), target: first, assertOwned: async () => undefined });
       await fs.writeFile(path.join(prepared.remoteRoot, "saved.txt"), "work survives a failed provider session\n");
       if (scenario !== "unsaved") await saveRemoteWorkspace({ db, storage, workspace: prepared.workspace,
-        runId: ids.run, lease: prepared.lease, target: first, assertOwned: async () => undefined });
+        runId: ids.run, lease: prepared.lease, target: first, assertOwned: async () => undefined,
+        boundary: kind === "historical_listener_late_proof" ? "initial" : "final" });
       const current = await workspace(reserved.id), authority = readRemoteWorkspaceAuthority(current.metadata)!;
       if (scenario === "dirty" || scenario === "stale_epoch") await db.update(executionWorkspaces).set({ metadata: {
         ...current.metadata, remoteWorkspace: { ...authority, ...(scenario === "dirty" ? { dirty: true, writerRunId: ids.run }
@@ -1729,29 +1745,137 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
         .where(eq(executionWorkspaceSnapshots.id, authority.snapshotId!));
       if (scenario === "delete_requested") await db.update(executionWorkspaceSnapshots).set({ status: "delete_requested" })
         .where(eq(executionWorkspaceSnapshots.id, authority.snapshotId!));
-      const stopped = (await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success",
+      if (processAcceptance) {
+        const { recordHistoricalListenerFailure } = await import("../__tests__/helpers/historical-listener-continuation.js");
+        await recordHistoricalListenerFailure({ db, connectionString: databaseUrl!, runId: ids.run, cwd: prepared.remoteRoot,
+          environmentId: ids.environment, leaseId: ids.lease });
+      }
+      const stopped = kind === "historical_listener_late_proof" ? prepared.lease : (await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success",
         metadata: { ...prepared.lease.metadata, ...(scenario !== "unconfirmed_stop" ? {
           remoteExecutionTermination: remoteTerminationReceipt(prepared.lease, {
             providerLeaseId: prepared.lease.providerLeaseId, state: "destroyed",
           }),
         } : {}) },
       }).where(eq(environmentLeases.id, ids.lease)).returning())[0]! as EnvironmentLease;
-      if (scenario === "saved") {
+      if (shouldSchedule) {
         await db.transaction(tx => clearRemoteWorkspaceLocationForReleasedLease({ db: tx, lease: stopped }));
         await fs.rm(prepared.remoteRoot, { recursive: true, force: true });
         await expect(fs.stat(prepared.remoteRoot)).rejects.toThrow();
       }
       const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, ids.run));
-      await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), runnerProfileJson: {
+      const sessionId = randomUUID();
+      await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(),
+        ...(historicalListener && !processAcceptance ? { nativeSessionId: sessionId } : {}),
+        ...(historicalListener ? { error: "local_runtime_api_listener_unavailable", errorCode: "native_session_interrupted" } : {}), runnerProfileJson: {
         ...source!.runnerProfileJson, recoveryEventInventoryVersion: 1,
-        nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: prepared.remoteRoot } },
+        nativeExecutionInput: processAcceptance ? source!.runnerProfileJson!.nativeExecutionInput : { provider: { kind: "codex" }, workspace: { cwd: prepared.remoteRoot },
+          ...(historicalListener ? { binding: { runId: ids.run, companyId: ids.company, agentId: ids.agent, issueId: ids.issue },
+            session: { normalizedSessionId: sessionId } } : {}) },
       } }).where(eq(heartbeatRuns.id, ids.run));
-      await db.insert(nativeRunFinalizations).values({ runId: ids.run, companyId: ids.company, issueId: ids.issue,
-        phase: "terminal_failure", attempt: 1, failureCode: "native_provider_terminal_failed",
-        failureDetail: { originalFailureCode: "fixture_checkpoint_unusable" } });
+      if (!processAcceptance) await db.insert(nativeRunFinalizations).values({ runId: ids.run, companyId: ids.company, issueId: ids.issue,
+        phase: "terminal_failure", attempt: 1, failureCode: historicalListener ? "native_session_interrupted" : "native_provider_terminal_failed",
+        failureDetail: historicalListener ? { originalFailureCode: "native_session_interrupted", message: "local_runtime_api_listener_unavailable",
+          recoveryMode: "ambiguous_state", providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false }
+          : { originalFailureCode: "fixture_checkpoint_unusable" } });
+      if (historicalListener && !processAcceptance) {
+        const blocked = await issueService(db).update(ids.issue, { status: "blocked" });
+        await db.insert(issueRecoveryActions).values({ companyId: ids.company, sourceIssueId: ids.issue,
+          kind: "active_run_watchdog", cause: "native_session_interrupted", ownerType: "board", status: "active",
+          fingerprint: createHash("sha256").update(`${ids.run}:native_session_interrupted`).digest("hex"),
+          evidence: { runId: ids.run, nativeFailureBlock: { runId: ids.run, statusVersion: blocked!.statusVersion } },
+          nextAction: "Reconcile the original startup failure." });
+      }
+      if (kind === "historical_listener_guard" || kind === "historical_listener_bad_preservation") {
+        const actionId = randomUUID(), preservationId = randomUUID();
+        await db.insert(issueRecoveryActions).values({ id: actionId, companyId: ids.company, sourceIssueId: ids.issue,
+          kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", ownerType: "system", status: "resolved", outcome: "blocked",
+          fingerprint: `native-workspace:${ids.run}`, evidence: { runId: ids.run,
+            workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [ids.lease] },
+            automaticRecovery: { replay: "blocked" }, continuationDelivery: "invalidated",
+            systemRecovery: { ...createSystemRecovery(ids.run), state: "complete", preservationOperationIds: [preservationId] },
+            workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: ids.run, status: "recovered", stage: "saving_receipts", attempt: 1, requestIds: [] },
+          }, nextAction: "Files were preserved without replay." });
+        await db.insert(workspaceOperations).values([{ id: preservationId, companyId: ids.company, heartbeatRunId: ids.run, issueId: ids.issue,
+          executionWorkspaceId: reserved.id, phase: "workspace_recovery", status: kind === "historical_listener_bad_preservation" ? "failed" : "succeeded", startedAt: new Date(),
+          metadata: { recoveryActionId: actionId, snapshotId: authority.snapshotId } },
+        { companyId: ids.company, heartbeatRunId: ids.run, issueId: ids.issue, phase: "workspace_finalize", status: "succeeded", startedAt: new Date(),
+          metadata: { owningService: "workspace_recovery", preservationOperationIds: [preservationId] } }]);
+      }
+      if (kind === "historical_listener_absent") await db.update(heartbeatRuns).set({
+        runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - 'remoteWorkspace'`,
+      }).where(eq(heartbeatRuns.id, ids.run));
+      if (kind === "historical_listener_stale_block") await issueService(db).update(ids.issue, { status: "blocked", actorUserId: "operator" });
+      if (kind === "historical_listener_unknown_event") await db.insert(heartbeatRunEvents).values({ companyId: ids.company, runId: ids.run,
+        agentId: ids.agent, seq: 100, eventType: "future.provider.activity", sourceEventId: "source-provider-event" });
+      if (kind === "historical_listener_budget") await db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, ids.run));
+      if (kind === "historical_listener_checkpoint") await db.update(heartbeatRuns).set({ runnerProfileJson: {
+        ...((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, ids.run)))[0]!.runnerProfileJson), sessionCheckpoint: { providerSessionId: "already-started" },
+      } }).where(eq(heartbeatRuns.id, ids.run));
+      if (kind === "historical_listener_resumed") await db.update(heartbeatRuns).set({ sessionIdBefore: "prior-session" }).where(eq(heartbeatRuns.id, ids.run));
+      if (kind === "historical_listener_prior_session") await db.insert(heartbeatRuns).values({ id: randomUUID(), companyId: ids.company,
+        agentId: ids.agent, status: "succeeded", nativeSessionId: sessionId });
+      if (kind === "historical_listener_other_message") await db.update(nativeRunFinalizations).set({ failureDetail: {
+        message: "provider reported local_runtime_api_listener_unavailable", originalFailureCode: "native_session_interrupted", recoveryMode: "ambiguous_state",
+        providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false,
+      } }).where(eq(nativeRunFinalizations.runId, ids.run));
+      if (kind === "historical_listener_cancelled") await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, ids.issue));
+      if (kind === "historical_listener_provider_event") await appendHeartbeatRunEvent(db, { companyId: ids.company, runId: ids.run,
+        agentId: ids.agent, eventType: "turn.started", stream: "system" });
+      const interactionWake = ["historical_listener_late_interaction", "historical_listener_interaction_paused", "historical_listener_own_question"].includes(kind);
+      const interactionId = randomUUID();
+      if (interactionWake) {
+        const priorRunId = randomUUID();
+        await db.insert(heartbeatRuns).values({ id: priorRunId, companyId: ids.company, agentId: ids.agent,
+          nativeIssueId: ids.issue, status: "succeeded", contextSnapshot: { issueId: ids.issue } });
+        await db.insert(issueThreadInteractions).values({ id: interactionId, companyId: ids.company, issueId: ids.issue,
+          kind: "request_confirmation", status: "pending", sourceRunId: kind === "historical_listener_own_question" ? ids.run : priorRunId,
+          payload: { version: 1, prompt: "An obsolete question from earlier work." } });
+        if (kind === "historical_listener_interaction_paused") await issueService(db).update(ids.issue, { status: "backlog" });
+        if (kind === "historical_listener_own_question") await appendHeartbeatRunEvent(db, { companyId: ids.company, runId: ids.run,
+          agentId: ids.agent, eventType: "tool.execution.started", stream: "system",
+          payload: { name: "request_confirmation", executionId: "source-question", transport: "process" } });
+      }
       if (scenario === "unknown_action") await appendHeartbeatRunEvent(db, { companyId: ids.company, runId: ids.run,
         agentId: ids.agent, eventType: "tool.execution.started", stream: "system",
         payload: { name: "send_email", executionId: "unconfirmed-external-action", transport: "process" } });
+      if (kind === "historical_listener_later_candidate") {
+        const sourceNumber = BigInt(`0x${ids.run.replaceAll("-", "")}`);
+        const parked = Array.from({ length: 100 }, (_, index) => {
+          const hex = (sourceNumber - BigInt(index + 1)).toString(16).padStart(32, "0");
+          return { runId: `${hex.slice(0,8)}-${hex.slice(8,12)}-${hex.slice(12,16)}-${hex.slice(16,20)}-${hex.slice(20)}`,
+            issueId: randomUUID(), sessionId: randomUUID() };
+        });
+        await db.insert(issues).values(parked.map(row => ({ id: row.issueId, companyId: ids.company,
+          title: "Preserve unavailable source", status: "in_progress", assigneeAgentId: ids.agent })));
+        await db.insert(heartbeatRuns).values(parked.map(row => ({ id: row.runId, companyId: ids.company, agentId: ids.agent,
+          nativeIssueId: row.issueId, nativeSessionId: row.sessionId, runtimeMode: "native", status: "failed",
+          errorCode: "native_session_interrupted", error: "local_runtime_api_listener_unavailable", runnerProfileJson: {
+            nativeExecutionInput: { binding: { runId: row.runId, companyId: ids.company, agentId: ids.agent, issueId: row.issueId },
+              session: { normalizedSessionId: row.sessionId } },
+          } })));
+        await db.insert(nativeRunFinalizations).values(parked.map(row => ({ runId: row.runId, companyId: ids.company, issueId: row.issueId,
+          phase: "terminal_failure", attempt: 1, failureCode: "native_session_interrupted", failureDetail: {
+            message: "local_runtime_api_listener_unavailable", originalFailureCode: "native_session_interrupted", recoveryMode: "ambiguous_state",
+            providerSessionEstablished: false, providerEventsExist: false, checkpointExists: false,
+          } })));
+        expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scanned: 100, scheduled: 0 });
+        expect(await executionControlSchedule(db).replacement()).not.toBeNull();
+        invalidateNativeReplacementDiscovery(db);
+      }
+      if (kind === "historical_listener_lock_timeout") {
+        let release!: () => void, announce!: () => void;
+        const locked = new Promise<void>(resolve => { announce = resolve; });
+        const released = new Promise<void>(resolve => { release = resolve; });
+        const holder = db.transaction(async tx => {
+          await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, ids.issue)).for("update");
+          announce(); await released;
+        });
+        await locked;
+        try {
+          await expect(reconcileSafeNativeReplacements(db)).rejects.toThrow();
+          expect(await executionControlSchedule(db).replacement()).not.toBeNull();
+        } finally { release(); await holder; }
+      }
       let replacement;
       if (scenario === "deletion_race" || scenario === "controller_claim_race") {
         let release!: () => void, announce!: () => void;
@@ -1783,12 +1907,102 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
           expect(blocked).toBe(true);
         } finally { release(); await deletion; replacement = await pending; }
       } else replacement = await reconcileSafeNativeReplacements(db);
-      expect(replacement).toMatchObject({ scheduled: scenario === "saved" ? 1 : 0 });
+      expect(replacement).toMatchObject({ scheduled: shouldSchedule ? 1 : 0 });
       const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, ids.run));
-      expect(successors).toHaveLength(scenario === "saved" ? 1 : 0);
-      if (scenario === "saved") expect(successors[0]).toMatchObject({ status: "scheduled_retry",
+      expect(successors).toHaveLength(shouldSchedule ? 1 : 0);
+      if (shouldSchedule) expect(successors[0]).toMatchObject({ status: "scheduled_retry",
         contextSnapshot: { retryOfRunId: ids.run, forceFreshSession: true } });
-      else expect(await fs.readFile(path.join(prepared.remoteRoot, "saved.txt"), "utf8"))
+      if (kind === "historical_listener_later_candidate") {
+        for (let pass = 0; pass < 3 && await executionControlSchedule(db).replacement() !== null; pass++) {
+          expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 });
+        }
+        expect(await executionControlSchedule(db).replacement()).toBeNull();
+      }
+      if (kind === "historical_listener_guard") expect(await getExecutionBlocker(db, ids.company, ids.issue)).toBeNull();
+      if (["historical_listener_provider_event", "historical_listener_checkpoint", "historical_listener_resumed", "historical_listener_other_message",
+        "historical_listener_cancelled", "historical_listener_prior_session", "historical_listener_unknown_event", "historical_listener_budget", "historical_listener_stale_block", "historical_listener_bad_preservation", "historical_listener_dirty", "historical_listener_unconfirmed_stop", "historical_listener_absent", "historical_listener_late_proof"].includes(kind)) {
+        expect(await executionControlSchedule(db).replacement()).toBeNull();
+      }
+      if (interactionWake) {
+        expect(await executionControlSchedule(db).replacement()).toBeNull();
+        let wakes = 0;
+        const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => {
+          if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); }
+        });
+        try {
+          await issueThreadInteractionService(db).skipInteraction({ id: ids.issue, companyId: ids.company,
+            status: kind === "historical_listener_interaction_paused" ? "backlog" : "blocked" }, interactionId, {}, { userId: "local-board" });
+          expect(wakes).toBeGreaterThan(0);
+          expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: kind === "historical_listener_late_interaction" ? 1 : 0 });
+          expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 });
+          expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, ids.run)))
+            .toHaveLength(kind === "historical_listener_late_interaction" ? 1 : 0);
+          expect(await executionControlSchedule(db).replacement()).toBeNull();
+        } finally { unsubscribe(); }
+      }
+      if (kind === "historical_listener_late_proof") {
+        let wakes = 0;
+        const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.executionReplacement, event => {
+          if (event === "settled") { wakes++; invalidateNativeReplacementDiscovery(db); }
+        });
+        try {
+          const saved = await saveRemoteWorkspace({ db, storage, workspace: await workspace(reserved.id), runId: ids.run,
+            lease: await lease(), target: first, assertOwned: async () => undefined });
+          expect(wakes).toBeGreaterThan(0);
+          expect(await executionControlSchedule(db).replacement()).toBeNull();
+          const priorWakes = wakes, active = await lease(), pluginId = randomUUID();
+          await db.update(environmentLeases).set({ metadata: { ...active.metadata, driver: "sandbox", pluginId,
+            provider: "daytona", sandboxProviderPlugin: true } }).where(eq(environmentLeases.id, ids.lease));
+          const runtime = environmentRuntimeService(db, { pluginWorkerManager: {
+            isRunning: () => true, getWorker: () => ({ supportedMethods: ["environmentReleaseLease"] }),
+            call: async (_id: string, method: string) => {
+              expect(method).toBe("environmentReleaseLease");
+              return { providerLeaseId: active.providerLeaseId, state: "destroyed" };
+            },
+          } as never });
+          await runtime.releaseRunLeases(ids.run, "failed");
+          expect(wakes).toBeGreaterThan(priorWakes);
+          expect(await executionControlSchedule(db).replacement()).not.toBeNull();
+          expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 1 });
+          expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 });
+          expect(await executionControlSchedule(db).replacement()).toBeNull();
+          const restored = await createSavedWorkspaceLoader(db, storage)({ companyId: ids.company, workspaceId: reserved.id, snapshotId: saved.snapshotId });
+          try { expect(await fs.readFile(path.join(restored.root, "saved.txt"), "utf8")).toBe("work survives a failed provider session\n"); }
+          finally { await restored.cleanup(); }
+        } finally { unsubscribe(); }
+      }
+      if (kind === "historical_listener_bad_preservation") {
+        expect(await getExecutionBlocker(db, ids.company, ids.issue)).toMatchObject({ cause: "native_continuation_requires_reconciliation" });
+
+      }
+      if (processAcceptance) {
+        expect(await reconcileSafeNativeReplacements(db)).toMatchObject({ scheduled: 0 });
+        const { completeHistoricalListenerSuccessor } = await import("../__tests__/helpers/historical-listener-continuation.js");
+        await completeHistoricalListenerSuccessor({ db, runId: successors[0]!.id, root: path.join(root, "successor-runtime"),
+          prepareWorkspace: async () => {
+            const leaseId = randomUUID(), runId = successors[0]!.id;
+            const [nextLease] = await db.insert(environmentLeases).values({ id: leaseId, companyId: ids.company,
+              environmentId: ids.environment, heartbeatRunId: runId, issueId: ids.issue, provider: "daytona",
+              providerLeaseId: `allocation-${leaseId}`, status: "active" }).returning();
+            const next = await reserveRemoteWorkspace({ db, storage, companyId: ids.company, issueId: ids.issue,
+              runId, agentId: ids.agent, environmentId: ids.environment, projectId: null, projectWorkspaceId: null,
+              existingWorkspaceId: reserved.id, source: { kind: "empty" } });
+            const nextTarget = { ...await target("historical-successor"), leaseId, sandboxId: nextLease.providerLeaseId! };
+            const nextPrepared = await prepareRemoteWorkspace({ db, storage, workspace: next, runId,
+              lease: nextLease as EnvironmentLease, target: nextTarget });
+            return { cwd: nextPrepared.remoteRoot, finalize: async () => {
+              const { workspaceOperationService } = await import("./workspace-operations.js");
+              const recorder = workspaceOperationService(db).createRecorder({ companyId: ids.company, issueId: ids.issue, heartbeatRunId: runId });
+              const saved = await recorder.recordOperation({ phase: "workspace_finalize", cwd: nextPrepared.remoteRoot, run: async () => {
+                await saveRemoteWorkspace({ db, storage, workspace: nextPrepared.workspace, runId, lease: nextPrepared.lease,
+                  target: nextTarget, assertOwned: async () => undefined });
+                return { status: "succeeded", exitCode: 0 };
+              } });
+              expect(saved.status).toBe("succeeded");
+            } };
+          } });
+      }
+      if (!shouldSchedule) expect(await fs.readFile(path.join(prepared.remoteRoot, "saved.txt"), "utf8"))
         .toBe("work survives a failed provider session\n");
       if (scenario === "unknown_action") {
         const [decision] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, ids.run));

server/src/services/remote-workspaces.ts

@@ -1,3 +1,4 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { workspaceAllocationFiles } from "./remote-workspace-allocation.js";
 import { readNativeWorkspaceMigration } from "./native-runtime/native-workspace-migration.js";
 import { createHash, randomUUID } from "node:crypto";
@@ -763,7 +764,10 @@ export async function saveRemoteWorkspace(input: {
         eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.status, "uploaded"))).returning({ id: executionWorkspaceSnapshots.id });
       if (!committed) throw new Error("remote_workspace_snapshot_publication_changed");
       await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`);
-      if (boundary === "final") await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`);
+      if (boundary === "final") {
+        await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`);
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement);
+      }
       const receipt = object(workspace.metadata?.remoteWorkspaceSnapshotRepositories);
       const priorRepositories = receipt.snapshotId === current.snapshotId && Array.isArray(receipt.repositories)
         ? remoteWorkspaceSourceSchema.parse({ kind: "empty", repositories: receipt.repositories }).repositories ?? [] : [];

server/src/services/run-dispatch-notifications.ts

@@ -12,6 +12,7 @@ export async function notifyRunDispatch(transaction: object, scope?: RunDispatch
   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.executionReplacement, hint);
   await notifyQueuedRunDispatch(transaction, scope);
 }
 

server/src/services/workspace-recovery.ts

@@ -390,6 +390,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ
       const activity = await db.transaction(async tx => {
         await notifyNativeResultCommit(tx, input);
         await notifyNativeWorkspaceFinalization(tx, input);
+        await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionReplacement);
         await notifyNativeStatusReassessment(tx, input);
         await tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,
           heartbeatRunId: input.runId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(),