fix(recovery): preserve failed remote reattach workspaces (#18936)

## Thinking Path

> - Paperclip manages AI agents and their work.
> - Native runs can edit files in a remote execution workspace.
> - A failed reattach can leave an observed coordinator and unpublished
files.
> - The failure handler recognizes the older workspace-sync descriptor
but misses the modern workspace reference.
> - The released task lock then prevents ordinary session recovery from
claiming that old run.
> - This pull request retains the files through the existing Stop,
cleanup and workspace recovery owners.
> - Exact ownership and Stop evidence permit file preservation while
unknown provider effects remain under manual reconciliation.

## Linked Issues or Issue Description

Related: #18877 protects native manual admission holds. #18879 handles
scheduled native retry lock reclaim. #18918 fixes remote workspace
row-lock order. Those changes do not record the missing modern
failure-retention obligation.

**What happened?**

A native reattach failure can leave a failed run, a dirty modern remote
workspace, an expired observed coordinator and a free task execution
lock. The run has no accepted result. The failure handler skips file
retention because it checks only the older descriptor. Recovery cannot
export while the old coordinator remains owned. A mismatched workspace
generation can also make orphan cleanup miss the existing file claim.

**Expected behavior**

Retain the exact original allocation and files. Stop the allocation
through the existing lifecycle owner before retiring the stale
coordinator. Keep checkpoints and unknown external effects available for
manual reconciliation. Unsafe or changed source authority must prevent
physical cleanup.

**Steps to reproduce**

1. Prepare a modern remote workspace for a native run.
2. Record a provider checkpoint and an active interruption incident.
3. Fail reattach preparation and release the task execution lock.
4. Let the coordinator lease expire.
5. Run the existing orphan, environment cleanup and workspace recovery
workers.
6. Verify the original files from a cold archive read and verify that no
new native session starts.

## What Changed

- Include modern prepared references in the existing heartbeat
failure-retention producer.
- Add a file-only historical intake for an exact failed observed run in
the normal expired reattach state.
- Require an untouched active ephemeral lease, a free blocked task,
matching workspace writer and generation, no accepted result and
complete single-owner allocation history.
- Persist the existing StopAndRetain intent before provider I/O. Use
nonblocking row locks and classify busy ownership only after rollback.
- Use the existing native environment settlement owner after exact Stop
and retention receipts arrive. Keep its single scoped discovery query.
- Preserve source and coordinator fingerprints across the handoff. Keep
original checkpoints, errors and manual unknown-effect incidents.
- Keep stale or malformed modern workspace claims held when preservation
intake refuses them.
- Add actual PostgreSQL, heartbeat, allocator, worker restart, archive
and safety controls. Document the narrow recovery contract.

## Verification

Current published head `b1f49691187a398415cf9b16b723e3ab473f0f91`: all
required CI checks pass (54 success, 3 skipped, 1 neutral); Greptile is
5/5 with no open threads. The separate Paperclip assessment is queued
and its current response says work is paused. That assessment is
unavailable; no additional review nudge or hosted retry was issued.

All local commands use Node 24.22.2 on macOS arm64.

- Exact published head: `b1f49691187a398415cf9b16b723e3ab473f0f91`.
Required CI passes (54 success, 3 skipped, 1 neutral); Greptile is 5/5
with zero open threads. The optional Paperclip assessment remains
queued/paused without a completed score.
- Final recursive `pnpm -r typecheck`: passed.
- Final `pnpm build`: passed.
- Final three native producer, containment and scheduling files: **67/67
passed**. These include the real coordinator and worker restart, the
production heartbeat recovery callback, exact Stop receipts, cold
archive restore, and zero native backend, new allocation or destructive
release calls.
- The plugin allocator cases use the actual selector with `reuseLease:
true`. Active and released ephemeral leases receive a fresh allocation.
- The assembled reattach producer checks that the task and coordinator
claim commit before physical resume. A historical pin refuses that
current owner. A later claim behind a committed pin remains held.
- The safety controls cover reusable allocations, manual and unknown
states, cleanup in flight, accepted but unbound results, late allocation
owners, wrong or malformed workspace authority, changed
generation/checkpoint, lost Stop acknowledgement and a real
cleanup-producer lock conflict.
- Untouched base `e7549ab6131555f655736e6b544892a3b1e323cb`: four
expected behavioral failures and one passing legacy-descriptor control.
The failures cover the modern producer, original provider history,
historical retention and wrong-epoch cleanup. Fixed production files
were restored byte-for-byte after this comparison.
- The first affected six-file run completed **562 passed / 1 failed**.
The failure was an existing HTTP case with `socket hang up`. That exact
case passes separately on both untouched base and this branch. The
original connection failure remains unexplained. At `0e92de8c`, the
complete workspace family passes **188/188** with unchanged assertions.
The environment cleanup owner family also passes all 23 cases. The later
guard changes are qualified by the complete 67-case native owner group.
A full repository test run was not repeated locally; the requested
full-ci workflow runs the complete repository suites.
- Published `0e92de8c` fails both new explicit-null Stop/export marker
controls because it creates a new retention pin. The follow-up uses the
shared presence-aware Stop guard. Both controls now preserve the marker
and record no new retention pin, Stop or destructive release.
- Paperclip's review found that malformed workspace authority could
throw and abort an orphan page. Published `5f6cd4f5` fails all three new
controls: intake, the actual page and after-Stop settlement. The same
strict schema now uses a safe parse, and the existing OBSERVED hold runs
before the prepared-source lookup. The malformed source stays held while
the next valid source receives its retention pin.
- Two existing scheduling tests caught the draft's second global
discovery query. The final single-query repair passes all 36 unchanged
scheduling tests.
- An earlier restarted archive assertion stopped after 20 seconds. Its
recorded Stop was valid, and the coordinator had settled. The existing
evidence recheck deadline was still 13 seconds ahead. The final test
observes the unchanged 24–36 second policy within its existing 60 second
case limit.

The external fake provider Stop and file transport are controlled test
boundaries. Database ownership, lifecycle workers, archive bytes,
encrypted local storage and cold restore are real. This does not claim a
live cloud sandbox, S3 or model run. No hosted task was changed for this
PR.

## Risks

- Historical intake is deliberately narrow. Reusable, already released,
mismatched or competing allocations remain held.
- File preservation does not settle unknown provider effects or
authorize a repeated provider turn. The task can still require manual
reconciliation.
- A busy or changed owner delays preservation. The original files and
claims remain retained.
- This does not repair the underlying sandbox readiness failure or
establish the cause of a separate archive-capture failure.
- Deployment and recovery of original hosted tasks remain unverified.

## Model Used

OpenAI Codex, GPT-6 family, with repository inspection, code execution
and GitHub tools. The runtime does not expose the exact serving model ID
or context window size. A separate Codex agent and the coordinator
performed read-only source review.

## 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 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/native-workspace-finalization-recovery.md

@@ -19,6 +19,37 @@ cleanup run. This includes descriptor reads that fail inside runtime selection.
 preparation failure; another owner or a stale claim prevents cleanup. Settling
 that claim and recording file retention share a transaction.
 
+Modern `remoteWorkspace` preparation references use the same failure-retention
+producer as workspace-sync descriptors. A failed native run can also retain an
+expired `awaiting_runner_reattach` coordinator after its task execution lock was
+released. The orphan-lease owner may record a file-only preservation obligation
+for that historical state only when the exact allocation remains active,
+unreleased and ephemeral. It rechecks the task, run, coordinator generation,
+prepared workspace writer, complete allocation history, current controller and
+cleanup state under short row locks. A reusable allocation, accepted result,
+competing writer, manual recovery state or ambiguous cleanup stays held. The
+presence of a Stop-only marker remains a hold even when its value is invalid.
+
+The transaction pins the existing StopAndRetain intent before provider I/O. The
+ordinary cleanup owner stops only the recorded allocation and rechecks all its
+current owners before dispatch. Normal checkpoint restart must first publish its
+task and coordinator claim; that claim prevents preservation intake. After a pin,
+the released task execution lock prevents an observed run from resuming itself.
+No row lock spans provider or archive I/O. Conflicting native or cleanup owners
+produce a nonblocking busy result after transaction rollback.
+
+Only matching source and generation fingerprints plus the exact stopped-and-retained
+receipt let the native environment settlement owner retire that stale coordinator
+into manual terminal reconciliation. The existing workspace worker then verifies
+and saves the original files. The interrupted provider checkpoint and its unknown
+effects remain unknown; this path admits no provider turn, accepts no result and
+changes no task status. A changed or malformed modern workspace claim remains a
+cleanup hold even when it is ineligible for automatic preservation. It cannot fall
+through to destructive orphan cleanup. Invalid workspace authority is refused
+without throwing or stopping preservation of the next source in the batch.
+Already released or mismatched historical
+sources require their owning reconciliation and are not rearmed by this path.
+
 The reconciliation candidate scan selects the complete workspace-sync record and
 the native status wake identifier from each saved run. It keeps their JSON types
 and reads full evidence later where finalization or recovery needs it. Saved

server/src/__tests__/environment-runtime.test.ts

@@ -857,6 +857,23 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
     return { pluginId, companyId, agentId, environment, runId, executionWorkspaceId, reusableLease };
   }
 
+  it.each(["active", "released"] as const)("does not select an ephemeral allocation for warm reuse (%s)", async status => {
+    const seeded = await seedReusablePluginSandboxLease("paperclip_runner");
+    await db.update(environmentLeases).set({ leasePolicy: "ephemeral", status,
+      releasedAt: status === "released" ? new Date() : null }).where(eq(environmentLeases.id, seeded.reusableLease.id));
+    const workerManager = createFreshPluginLeaseWorker(seeded.pluginId);
+    const runtimeWithPlugin = environmentRuntimeService(db, { pluginWorkerManager: workerManager });
+    const acquired = await runtimeWithPlugin.acquireRunLease({
+      companyId: seeded.companyId, environment: seeded.environment, agentId: seeded.agentId,
+      issueId: null, adapterType: "paperclip_runner", heartbeatRunId: seeded.runId,
+      persistedExecutionWorkspace: { id: seeded.executionWorkspaceId, mode: "shared_workspace" },
+    });
+    expect(seeded.environment.config.reuseLease).toBe(true);
+    expect(acquired.lease.providerLeaseId).toBe("fresh-plugin-lease");
+    expect(acquired.lease.providerLeaseId).not.toBe(seeded.reusableLease.providerLeaseId);
+    expect(workerManager.call).toHaveBeenCalledExactlyOnceWith(seeded.pluginId, "environmentAcquireLease", expect.any(Object), expect.any(Number));
+  });
+
   it("gives a bootstrap retry a fresh allocation even when the failed lease was reusable", async () => {
     const seeded = await seedReusablePluginSandboxLease("paperclip_runner");
     await environmentService(db).releaseLease(seeded.reusableLease.id, "expired");

server/src/__tests__/heartbeat-native-remote-failure-retention.test.ts

@@ -0,0 +1,156 @@
+import { randomUUID } from "node:crypto";
+import { mkdir, readFile, writeFile } from "node:fs/promises";
+import path from "node:path";
+import { eq } from "drizzle-orm";
+import { agents, authUsers, companyMemberships, environmentLeases, environments, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
+import { beforeAll, expect, it, vi } from "vitest";
+import { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js";
+import * as storageBackend from "../storage/index.js";
+import { getStorageService } from "../storage/index.js";
+import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
+import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
+import { pinFailedObservedWorkspace } from "../services/native-runtime/native-observed-workspace-preservation.js";
+import { heartbeatService } from "../services/heartbeat.js";
+import { environmentRuntimeService, classifyEnvironmentCapabilities } from "../services/environment-runtime.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
+import { buildNativeExecutionInput } from "../services/native-runtime/native-execution-input.js";
+import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js";
+import { registerNativeRunnerCheckpoint } from "../services/native-runtime/native-runner-checkpoint.js";
+import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js";
+
+describeEmbeddedPostgres("heartbeat native remote preparation failure retention", () => {
+  const fixture = nativeRestartFixture();
+  let runtimeRoot: string, companyId: string, agentId: string;
+  beforeAll(() => { ({ runtimeRoot, companyId, agentId } = fixture.current); });
+
+  it.each(["modern", "legacy", "provider_history"] as const)("preserves the allocated workspace when restart preparation fails (%s)", async mode => {
+    const f = await fixture.seedRun(`PREPARATION-FAILURE-${mode}`);
+    const db = f.db, workspaceId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID();
+    const settings = instanceSettingsService(db), previous = await settings.getExperimental();
+    const priorDatabase = process.env.DATABASE_URL, priorKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY;
+    const localStorage = getStorageService();
+    let storage: { mockRestore(): void } | undefined;
+    let heartbeat: ReturnType<typeof heartbeatService> | undefined;
+    try {
+    storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" });
+    process.env.DATABASE_URL = fixture.current.temporary.connectionString;
+    process.env.PAPERCLIP_SECRETS_MASTER_KEY = "71".repeat(32);
+    const userId = randomUUID();
+    await db.insert(authUsers).values({ id: userId, name: "Original operator", email: `${userId}@example.invalid`, createdAt: new Date(), updatedAt: new Date() });
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" });
+    const remoteCwd = path.join(runtimeRoot, workspaceId), providerLeaseId = `sandbox://fake/${randomUUID()}`;
+    await mkdir(remoteCwd, { recursive: true });
+    await writeFile(path.join(remoteCwd, "saved.txt"), "unfinished original work\n");
+    await db.insert(environments).values({ id: environmentId, name: `Bound original allocation ${mode}`, driver: "sandbox",
+      config: { provider: "fake", image: "fixture:test" } });
+    await db.update(agents).set({ defaultEnvironmentId: environmentId }).where(eq(agents.id, agentId));
+    const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId,
+      remoteCwd, epoch: 1, state: "prepared", snapshotId: null };
+    await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: f.issueId,
+      mode: "isolated_workspace", strategyType: "project_primary", providerType: "cloud_sandbox", name: "Original files",
+      cwd: `/workspaces/${workspaceId}`, metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1",
+        epoch: 1, snapshotId: null, source: { kind: "empty" }, location: { environmentId, leaseId, providerLeaseId, remoteCwd },
+        writerRunId: f.runId, dirty: true } } });
+    await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, f.issueId));
+    await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId: f.issueId, heartbeatRunId: f.runId,
+      executionWorkspaceId: workspaceId, environmentId, provider: "fake", providerLeaseId, status: "active",
+      leasePolicy: "ephemeral", metadata: { driver: "sandbox", provider: "fake", image: "fixture:test", remoteCwd, agentId } });
+    const execution = buildNativeExecutionInput({ companyId, agentId, runId: f.runId,
+      issue: { id: f.issueId, identifier: "NRR-MODERN-PREPARATION-FAILURE", title: "Preserve original work", description: null, workMode: "standard" },
+      taskPrompt: "Continue the original task.", provider: "codex",
+      workspace: { id: workspaceId, cwd: remoteCwd, repoUrl: null, repoRef: null, branchName: null },
+      normalizedSessionId: f.native.normalizedSessionId, runtimeContext: nativeRuntimeContextFixture(),
+      completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1",
+        contract: { revision: "1", objective: "Preserve original work", criteria: [{ id: "objective", requirement: "Retain files" }] } } });
+    const identity = { companyId, issueId: f.issueId, runId: f.runId, agentId, sessionId: f.native.normalizedSessionId };
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.preparedProfile, remoteWorkspace: reference,
+      nativeExecutionInput: execution, sessionCheckpoint: { identity, providerSessionId: "original-provider" },
+      ...(mode === "legacy" ? { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", state: "prepared",
+        descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null,
+        workspaceId, leaseId, providerLeaseId, remoteCwd, resourceDisposition: "stop_and_retain" } } : {}) },
+      responsibleUserId: userId, contextSnapshot: { issueId: f.issueId, executionWorkspaceId: workspaceId } }).where(eq(heartbeatRuns.id, f.runId));
+    await db.update(nativeRunFinalizations).set({ controllerBootId: fixture.current.successor.bootId,
+      controllerGeneration: 1, attempt: 1, leaseOwner: "previous-controller", leaseExpiresAt: new Date(Date.now() + 60_000) })
+      .where(eq(nativeRunFinalizations.runId, f.runId));
+    const authority = new DurablePrpControlPlane({ stateDirectory: path.join(runtimeRoot, f.runId, "control-plane"),
+      identity: fixture.transportOptions(f, runtimeRoot).prpIdentity,
+      expectedRunnerVersion: "retention-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
+    await registerNativeRunnerCheckpoint({ db, binding: { companyId, issueId: f.issueId, runId: f.runId },
+      bootId: fixture.current.successor.bootId, authority });
+    await authority.stop();
+    if (mode === "provider_history") {
+      const event = { schema: "paperclip.prp.event.v1", sourceInstanceId: f.native.runnerInstanceId,
+        sourceEventId: randomUUID(), sourceSeq: 1, schemaVersion: 1, sourceKind: "runner", runId: f.runId,
+        normalizedSessionId: f.native.normalizedSessionId, turnId: f.native.turnId, eventType: "tool.execution.started",
+        priority: 0, emittedAt: new Date().toISOString(), payload: { toolName: "fixture.external_effect", toolCallId: randomUUID() } };
+      await appendHeartbeatRunEvent(db, { companyId, agentId, runId: f.runId, eventType: event.eventType,
+        stream: "system", level: "info", message: "Original external action has an unknown outcome", payload: { prpEvent: event },
+        nativeSource: { sourceEventId: event.sourceEventId, sourceInstanceId: event.sourceInstanceId,
+          sourceSeq: event.sourceSeq, protocolSchemaVersion: 1, canonicalPayload: event } });
+    }
+    await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: f.issueId,
+      kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: agentId, cause: "native_session_interrupted",
+      fingerprint: `native-interruption:${f.runId}`, evidence: { runId: f.runId },
+      nextAction: "Inspect the interrupted source and preserve its work.", maxAttempts: 3, wakePolicy: null,
+      supersedeOnIdentityChange: true });
+    await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) })
+      .where(eq(nativeRunFinalizations.runId, f.runId));
+    const preparationFailure = new Error("Sandbox failed to become ready within the timeout period");
+    const resumed = vi.fn(async () => {
+      const [claimedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+      const [claimedOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
+      const [claimedTask] = await db.select().from(issues).where(eq(issues.id, f.issueId));
+      expect(claimedRun.status).toBe("running");
+      expect(claimedTask.executionRunId).toBe(f.runId);
+      expect(claimedOwner.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now());
+      expect(claimedOwner.controllerGeneration).toBe(2);
+      expect(await pinFailedObservedWorkspace(db, companyId, f.runId)).toBe(false);
+      throw preparationFailure;
+    });
+    const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox",
+      acquireRunLease: async () => { throw new Error("Unexpected new allocation"); },
+      resumeRunLease: async () => { throw new Error("Reattachment must not resume the allocation"); },
+      releaseRunLease: async () => { throw new Error("Unexpected destructive cleanup"); },
+      retryPendingSandboxTeardown: async () => { throw new Error("Stop must remain separately verified"); },
+      realizeWorkspace: async () => { throw new Error("Unexpected workspace realization"); },
+      execute: resumed,
+      syncIn: resumed,
+      syncOut: resumed,
+      supportsSync: () => true,
+      resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: ["environmentSyncIn", "environmentSyncOut", "environmentExecute"], declared: null, narrowing: null }),
+    }] });
+    await settings.updateExperimental({ enableNativeRunner: true });
+    heartbeat = heartbeatService(db, { environmentRuntime: runtime,
+      nativeSessionBackendFactory: () => { throw new Error("No provider backend may be created"); } });
+      await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]);
+      await heartbeat.drainActiveRunExecutions();
+      expect(resumed).toHaveBeenCalledOnce();
+      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+      expect(run.error).toBe(preparationFailure.message);
+      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
+      const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId));
+      const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId));
+      expect({ status: run.status, phase: coordinator.phase, taskExecution: task.executionRunId, dirty: workspace.metadata?.remoteWorkspace })
+        .toMatchObject({ status: "failed", phase: "terminal_failure", taskExecution: null, dirty: { writerRunId: f.runId, dirty: true } });
+      expect(run.resultJson).toMatchObject({ workspaceRestoreFailure: "restore_failed",
+        workspaceRestoreRecovery: { leaseIds: [leaseId] } });
+      const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId));
+      expect(lease).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure",
+        metadata: { sandboxStopAndRetain: { runId: f.runId, leaseId } } });
+      expect(await readFile(path.join(remoteCwd, "saved.txt"), "utf8")).toBe("unfinished original work\n");
+      expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)))
+        .toEqual(expect.arrayContaining([expect.objectContaining({ evidence: expect.objectContaining({ workspaceRestoreRecovery: expect.any(Object) }) })]));
+      expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1);
+      if (mode === "provider_history") expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.runId)))
+        .toEqual(expect.arrayContaining([expect.objectContaining({ eventType: "tool.execution.started", sourceSeq: 1 })]));
+    } finally {
+      try { await heartbeat?.drainActiveRunExecutions(); }
+      finally {
+        storage?.mockRestore();
+        if (priorDatabase === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = priorDatabase;
+        if (priorKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = priorKey;
+        await settings.updateExperimental(previous);
+      }
+    }
+  }, 60_000);
+});

server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts

@@ -0,0 +1,389 @@
+import { execFile } from "node:child_process";
+import { randomUUID } from "node:crypto";
+import { cp, mkdir, readFile, realpath, writeFile } from "node:fs/promises";
+import path from "node:path";
+import { promisify } from "node:util";
+import { and, eq, sql } from "drizzle-orm";
+import { agents, authUsers, companyMemberships, environmentLeases, environments, executionWorkspaceSnapshots, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db";
+import { beforeAll, expect, it, vi } from "vitest";
+import * as storageBackend from "../storage/index.js";
+import { getStorageService } from "../storage/index.js";
+import { buildNativeExecutionInput } from "../services/native-runtime/native-execution-input.js";
+import { nativeRuntimeContextFixture } from "../services/native-runtime/runtime-context.test-fixture.js";
+import { getSandboxProvider } from "../services/sandbox-provider-runtime.js";
+import { heartbeatService } from "../services/heartbeat.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { registerEnvironmentCleanupWork } from "../services/environment-cleanup-work.js";
+import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../services/environment-runtime.js";
+import { environmentService } from "../services/environments.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
+import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
+import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "../services/remote-workspace-snapshot-storage.js";
+import { restoreRemoteWorkspaceArchive } from "../services/remote-workspace-snapshot-archive.js";
+import { registerWorkspaceRecoveryWork } from "../services/workspace-recovery-work.js";
+import { hasVerifiedWorkspaceRepair } from "../services/workspace-recovery-receipt.js";
+import { pinFailedObservedWorkspace, settleStoppedObservedWorkspace, ObservedWorkspacePreservationBusyError } from "../services/native-runtime/native-observed-workspace-preservation.js";
+import { claimNativeRestartRecoveries } from "../services/native-runtime/native-restart-recovery.js";
+import { hasRemoteTerminationReceipt } from "../services/remote-execution-termination.js";
+import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js";
+
+const execute = promisify(execFile);
+function barrier() { let resolve!: () => void; const promise = new Promise<void>(ready => { resolve = ready; }); return { promise, resolve }; }
+
+describeEmbeddedPostgres("stopped observed native workspace preservation", () => {
+  const fixture = nativeRestartFixture();
+  let runtimeRoot: string, companyId: string, agentId: string;
+  beforeAll(() => { ({ runtimeRoot, companyId, agentId } = fixture.current); });
+
+  async function seedSource(label: string) {
+    const f = await fixture.seedRun(label), db = f.db;
+    const workspaceId = randomUUID(), environmentId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox://fake/${randomUUID()}`;
+    await mkdir(path.join(runtimeRoot, workspaceId));
+    const remoteCwd = await realpath(path.join(runtimeRoot, workspaceId));
+    await writeFile(path.join(remoteCwd, "saved.txt"), "unfinished work with an unknown provider effect\n");
+    await db.insert(environments).values({ id: environmentId, name: `Historical allocation ${environmentId}`, driver: "sandbox",
+      config: { provider: "fake", image: "fixture:test" } });
+    const userId = randomUUID();
+    await db.insert(authUsers).values({ id: userId, name: "Source operator", email: `${userId}@example.invalid`, createdAt: new Date(), updatedAt: new Date() });
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" });
+    await db.update(agents).set({ defaultEnvironmentId: environmentId, adapterConfig: { provider: "codex", model: "gpt-6.1-sol" } }).where(eq(agents.id, agentId));
+    const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, leaseId, providerLeaseId,
+      remoteCwd, epoch: 1, state: "prepared", snapshotId: null };
+    await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: f.issueId,
+      mode: "isolated_workspace", strategyType: "project_primary", providerType: "cloud_sandbox", name: "Unsaved original files",
+      cwd: `/workspaces/${workspaceId}`, metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1",
+        epoch: 1, snapshotId: null, source: { kind: "empty" }, writerRunId: f.runId, dirty: true,
+        location: { environmentId, leaseId, providerLeaseId, remoteCwd } } } });
+    await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null, executionWorkspaceId: workspaceId })
+      .where(eq(issues.id, f.issueId));
+    await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId: f.issueId, heartbeatRunId: f.runId,
+      executionWorkspaceId: workspaceId, environmentId, provider: "fake", providerLeaseId, status: "active", leasePolicy: "ephemeral",
+      metadata: { driver: "sandbox", provider: "fake", image: "fixture:test", remoteCwd, agentId }, updatedAt: new Date(Date.now() - 6 * 60_000) });
+    const checkpoint = { identity: { companyId, issueId: f.issueId, runId: f.runId, agentId, sessionId: f.native.normalizedSessionId },
+      providerSessionId: "historical-session", activeTurnId: "unfinished-turn", pendingRuntimeRequests: [{ requestId: "unknown-tool" }] };
+    const execution = buildNativeExecutionInput({ companyId, agentId, runId: f.runId,
+      issue: { id: f.issueId, identifier: "NRR-HISTORICAL-OBSERVED", title: "Preserve original work", description: null, workMode: "standard" },
+      taskPrompt: "Continue the original task.", provider: "codex", model: "gpt-6.1-sol",
+      workspace: { id: workspaceId, cwd: remoteCwd, repoUrl: null, repoRef: null, branchName: null },
+      normalizedSessionId: f.native.normalizedSessionId, runtimeContext: nativeRuntimeContextFixture(),
+      completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1",
+        contract: { revision: "1", objective: "Preserve original work", criteria: [{ id: "objective", requirement: "Retain files" }] } } });
+    const profile = { ...f.preparedProfile, remoteWorkspace: reference, sessionCheckpoint: checkpoint,
+      nativeExecutionInput: execution, nativeResourceDisposition: "stop_and_retain" };
+    await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(Date.now() - 6 * 60_000),
+      nativePhase: "observed", errorCode: "setup_failed", error: "Sandbox failed to become ready within the timeout period",
+      responsibleUserId: userId, runnerProfileJson: profile, contextSnapshot: { issueId: f.issueId, executionWorkspaceId: workspaceId } }).where(eq(heartbeatRuns.id, f.runId));
+    await db.update(nativeRunFinalizations).set({ phase: "observed", attempt: 1, controllerGeneration: 3, recoveryState: "awaiting_runner_reattach",
+        controllerBootId: randomUUID(), controllerPid: 2_147_483_640, controllerProcessStartedAt: new Date(0),
+      leaseOwner: "old-observed-controller", leaseExpiresAt: new Date(Date.now() - 5 * 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
+    await issueRecoveryActionService(db).upsertSourceScoped({ companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog",
+      ownerType: "board", returnOwnerAgentId: agentId, cause: "native_session_interrupted", fingerprint: `interrupted:${f.runId}`,
+      evidence: { runId: f.runId }, nextAction: "Reconcile the unknown provider outcome.", maxAttempts: 3, wakePolicy: null,
+      supersedeOnIdentityChange: true });
+    return { ...f, workspaceId, environmentId, leaseId, providerLeaseId, remoteCwd, checkpoint, profile };
+  }
+
+  it("stops the exact stale allocation, preserves a cold-readable archive and never creates another native session", async () => {
+    const f = await seedSource("HISTORICAL-OBSERVED"), db = f.db;
+    const settings = instanceSettingsService(db), previous = await settings.getExperimental();
+    const priorDatabase = process.env.DATABASE_URL, priorKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY;
+    const localStorage = getStorageService();
+    let storage: { mockRestore(): void } | undefined;
+    let heartbeat: ReturnType<typeof heartbeatService> | undefined;
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let workspaceWorker: ReturnType<typeof registerWorkspaceRecoveryWork> | undefined;
+    let providerStop: { mockRestore(): void } | undefined;
+    let providerDestroy: { mockRestore(): void } | undefined;
+    let cleanupWorker: ReturnType<typeof registerEnvironmentCleanupWork> | undefined;
+    const stopAllowed = barrier();
+    try {
+      storage = vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...localStorage, provider: "s3" });
+      process.env.DATABASE_URL = fixture.current.temporary.connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "72".repeat(32);
+      await settings.updateExperimental({ enableNativeRunner: true });
+      const { workspaceId, environmentId, leaseId, providerLeaseId, remoteCwd, checkpoint, profile } = f;
+      const acquire = vi.fn(async () => { throw new Error("A new allocation is forbidden"); });
+      const backend = vi.fn(() => { throw new Error("A native session is forbidden"); });
+      const copy = async (input: { operations: Array<{ files: Array<{ sourcePath: string; targetPath: string }> }> }) => {
+        for (const operation of input.operations) for (const file of operation.files) {
+          await mkdir(path.dirname(file.targetPath), { recursive: true }); await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false });
+        }
+        return { operations: [], filesTransferred: input.operations.reduce((sum, operation) => sum + operation.files.length, 0), bytesTransferred: 0 };
+      };
+      const provider = getSandboxProvider("fake")!;
+      const originalStop = provider.stopLease!.bind(provider);
+      const stopped = vi.spyOn(provider, "stopLease").mockImplementation(async input => {
+        await stopAllowed.promise; return originalStop(input);
+      });
+      providerStop = stopped;
+      const destroyed = vi.spyOn(provider, "destroyLease"); providerDestroy = destroyed;
+      const fileRuntime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", acquireRunLease: acquire,
+        resumeRunLease: async ({ lease }) => ({ providerLeaseId: lease.providerLeaseId!, metadata: { remoteCwd: lease.metadata?.remoteCwd } }),
+        releaseRunLease: async () => { throw new Error("Destructive release is forbidden"); },
+        retryPendingSandboxTeardown: async () => { throw new Error("Use the actual lifecycle owner"); }, realizeWorkspace: async () => { throw new Error("New realization is forbidden"); },
+        execute: async input => {
+          const result = await execute(input.command === "node" ? process.execPath : input.command, input.args ?? [],
+            { cwd: input.cwd, env: { ...process.env, ...input.env }, maxBuffer: 4 * 1024 * 1024 });
+          return { stdout: result.stdout, stderr: result.stderr, exitCode: 0, timedOut: false, signal: null };
+        },
+        syncIn: copy, syncOut: copy, supportsSync: () => true,
+        resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: ["environmentSyncIn", "environmentSyncOut", "environmentExecute"] }),
+      }] });
+      const runtime = { ...fileRuntime, retryPendingSandboxTeardown: environmentRuntimeService(db).retryPendingSandboxTeardown };
+      heartbeat = heartbeatService(db, { environmentRuntime: runtime, nativeSessionBackendFactory: backend });
+      const errors = vi.fn();
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors });
+      workspaceWorker = registerWorkspaceRecoveryWork(db, coordinator, runtime, heartbeat.resumeRemoteStopComments);
+      const environmentWorker = heartbeat.registerNativeEnvironmentSettlementWork(coordinator, () => true);
+      const orphanWorker = heartbeat.registerResidualRecoveryWork(coordinator, () => true);
+      await Promise.all([workspaceWorker.ready, environmentWorker.ready, orphanWorker.ready]);
+      await expect.poll(async () => (await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)))[0].status,
+        { timeout: 10_000 }).toBe("pending_cleanup");
+      const [beforeRestart] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId),
+        eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`)));
+      await coordinator.stop(); await workspaceWorker.stop();
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: errors });
+      workspaceWorker = registerWorkspaceRecoveryWork(db, coordinator, runtime, heartbeat.resumeRemoteStopComments);
+      const restartedEnvironment = heartbeat.registerNativeEnvironmentSettlementWork(coordinator, () => true);
+      const restartedOrphan = heartbeat.registerResidualRecoveryWork(coordinator, () => true);
+      await Promise.all([workspaceWorker.ready, restartedEnvironment.ready, restartedOrphan.ready]);
+      expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, beforeRestart.id)))[0].evidence.observedWorkspacePreservation)
+        .toEqual(beforeRestart.evidence.observedWorkspacePreservation);
+      cleanupWorker = registerEnvironmentCleanupWork(db, coordinator, { canRun: () => true,
+        sweepPending: backoffMs => heartbeat!.sweepPendingCleanupLeases({ backoffMs }), minimumSpacingMs: 1 });
+      await expect.poll(() => stopped.mock.calls.length, { timeout: 10_000 }).toBe(1);
+      const [pending] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId));
+      const [ownerBeforeStop] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
+      expect(ownerBeforeStop).toMatchObject({ phase: "observed", leaseOwner: "old-observed-controller", controllerGeneration: 3, attempt: 1 });
+      expect(pending).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure",
+        metadata: { sandboxStopAndRetain: { runId: f.runId, leaseId, providerLeaseId } } });
+      expect(hasRemoteTerminationReceipt(pending)).toBe(false);
+      expect(await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, workspaceId))).toHaveLength(0);
+      expect(backend).not.toHaveBeenCalled(); expect(acquire).not.toHaveBeenCalled();
+      stopAllowed.resolve();
+      await cleanupWorker.ready;
+      const [waiting] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId),
+        eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`)));
+      const wait = waiting.evidence.systemRecovery as { state?: string; nextAttemptAt?: number; failure?: { code?: string } } | undefined;
+      if (wait?.nextAttemptAt && wait.nextAttemptAt > Date.now()) {
+        expect(wait).toMatchObject({ state: "awaiting_evidence", failure: { code: "agent_stop_unconfirmed" } });
+        expect(wait.nextAttemptAt - Date.now()).toBeLessThanOrEqual(36_000);
+      }
+      await expect.poll(async () => hasVerifiedWorkspaceRepair(db,
+        (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]), { timeout: 40_000 }).toBe(true);
+      const [savedWorkspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId));
+      const [savedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+      const [savedOwner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
+      expect(savedOwner).toMatchObject({ phase: "terminal_failure", leaseOwner: null, attempt: 1, controllerGeneration: 3 });
+      expect(savedRun).toMatchObject({ status: "failed", errorCode: "setup_failed", runnerProfileJson: { sessionCheckpoint: checkpoint, nativeExecutionInput: profile.nativeExecutionInput } });
+      const snapshotId = (savedWorkspace.metadata?.remoteWorkspace as { snapshotId: string }).snapshotId;
+      const snapshot = await getWorkspaceSnapshot(db, { companyId, workspaceId, snapshotId });
+      expect(snapshot).not.toBeNull();
+      const cold = path.join(runtimeRoot, randomUUID()), archivePath = path.join(runtimeRoot, `${randomUUID()}.pcws`);
+      await loadWorkspaceSnapshotArchive({ storage: storageBackend.getStorageService(), reference: snapshot!, targetPath: archivePath });
+      await restoreRemoteWorkspaceArchive({ archivePath, targetDir: cold, expectedManifestDigest: snapshot!.manifestDigest });
+      expect(await readFile(path.join(cold, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n");
+      expect(await readFile(path.join(remoteCwd, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n");
+      expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, f.issueId))).toHaveLength(1);
+      expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toHaveLength(0);
+      const [preservation] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId),
+        eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`)));
+      expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked");
+      expect((await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId),
+        eq(issueRecoveryActions.fingerprint, `interrupted:${f.runId}`))))[0]).toMatchObject({ status: "active", cause: "native_session_interrupted" });
+      expect(preservation.evidence).toMatchObject({ automaticRecovery: { policy: "preserve_without_replay_v1", replay: "blocked", actionOutcome: "unknown" } });
+      expect(backend).not.toHaveBeenCalled(); expect(acquire).not.toHaveBeenCalled();
+      expect(destroyed).not.toHaveBeenCalled();
+      expect(errors).not.toHaveBeenCalled();
+    } finally {
+      stopAllowed.resolve(); cleanupWorker?.stop();
+      try { await coordinator?.stop(); await workspaceWorker?.stop(); await heartbeat?.drainActiveRunExecutions(); }
+      finally {
+        providerStop?.mockRestore(); providerDestroy?.mockRestore();
+        storage?.mockRestore();
+        if (priorDatabase === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = priorDatabase;
+        if (priorKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = priorKey;
+        await settings.updateExperimental(previous);
+      }
+    }
+  }, 60_000);
+
+  type Source = Awaited<ReturnType<typeof seedSource>>;
+  async function readSource(f: Source) {
+    const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+    const [lease] = await f.db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId));
+    const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
+    return { run, lease, owner };
+  }
+  async function changeLease(f: Source, patch: Partial<typeof environmentLeases.$inferInsert>, metadata?: Record<string, unknown>) {
+    const { lease } = await readSource(f);
+    await f.db.update(environmentLeases).set({ ...patch, ...(metadata ? { metadata: { ...lease.metadata, ...metadata } } : {}) })
+      .where(eq(environmentLeases.id, f.leaseId));
+  }
+  async function acceptedResult(f: Source) {
+    const { run } = await readSource(f);
+    await f.db.insert(nativeRunResults).values({ companyId, issueId: f.issueId, runId: f.runId,
+      completionContractId: run.completionContractId!, schemaStatus: "accepted", serverFingerprint: randomUUID(),
+      canonicalSha256: "a".repeat(64), resultJson: { outcome: "accepted" } });
+  }
+  async function otherAllocationOwner(f: Source) {
+    const other = await fixture.seedRun(`OTHER-${randomUUID()}`, f.db);
+    await f.db.insert(environmentLeases).values({ companyId, issueId: other.issueId, heartbeatRunId: other.runId,
+      environmentId: f.environmentId, provider: "fake", providerLeaseId: f.providerLeaseId,
+      status: "active", leasePolicy: "ephemeral", metadata: { driver: "sandbox", provider: "fake", image: "fixture:test" } });
+    return other;
+  }
+  async function sourceHeld(f: Source) {
+    const { run, owner } = await readSource(f);
+    expect(owner).toMatchObject({ phase: "observed", leaseOwner: "old-observed-controller", controllerGeneration: 3, attempt: 1 });
+    expect(run).toMatchObject({ status: "failed", runnerProfileJson: { sessionCheckpoint: f.checkpoint } });
+    expect(await readFile(path.join(f.remoteCwd, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n");
+    expect(await f.db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, f.workspaceId))).toHaveLength(0);
+  }
+
+  it.each(["reusable", "inflight_cleanup", "unknown_cleanup", "null_stop_intent", "null_export_intent", "unknown_recovery", "manual_failure", "live_controller",
+    "copyback_owner", "accepted_unbound", "other_allocation_owner", "wrong_epoch", "invalid_reference", "stopped_mismatch", "operator_stop"] as const)(
+    "keeps unsafe observed source held without Stop or Delete (%s)", async mode => {
+      const f = await seedSource(`HELD-${mode}`);
+      if (mode === "reusable") await changeLease(f, { leasePolicy: "reuse_by_environment" });
+      if (mode === "inflight_cleanup") await changeLease(f, {}, { pendingCleanupInFlight: true });
+      if (mode === "unknown_cleanup") await changeLease(f, {}, { pendingCleanupInFlight: "unknown" });
+      if (mode === "null_stop_intent") await changeLease(f, {}, { sandboxStopAndRetain: null });
+      if (mode === "null_export_intent") await changeLease(f, {}, { nativeWorkspaceExportResume: null });
+      if (mode === "unknown_recovery") await f.db.update(nativeRunFinalizations).set({ recoveryState: "awaiting_evidence" }).where(eq(nativeRunFinalizations.runId, f.runId));
+      if (mode === "manual_failure") await f.db.update(nativeRunFinalizations).set({ failureCode: "native_restart_recovery_blocked" }).where(eq(nativeRunFinalizations.runId, f.runId));
+      if (mode === "live_controller") await f.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId));
+      if (mode === "copyback_owner") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile,
+        nativeWorkspaceFinalizationOwner: { token: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId));
+      if (mode === "invalid_reference") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, remoteWorkspace: { schema: "unverified" } } }).where(eq(heartbeatRuns.id, f.runId));
+      if (mode === "accepted_unbound") await acceptedResult(f);
+      if (mode === "other_allocation_owner") await otherAllocationOwner(f);
+      if (mode === "wrong_epoch") await f.db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '2')` }).where(eq(executionWorkspaces.id, f.workspaceId));
+      if (mode === "stopped_mismatch") await changeLease(f, { status: "released", cleanupStatus: "success", releasedAt: new Date() }, {
+        remoteExecutionTermination: { schema: "paperclip.remote-execution-termination.v1", provider: "fake", providerLeaseId: `sandbox://fake/${randomUUID()}`, state: "stopped", confirmedAt: new Date().toISOString() },
+      });
+      if (mode === "operator_stop") await f.db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId));
+      const provider = getSandboxProvider("fake")!, stopped = vi.spyOn(provider, "stopLease"), destroyed = vi.spyOn(provider, "destroyLease");
+      const backend = vi.fn(() => { throw new Error("Unexpected native execution"); });
+      const heartbeat = heartbeatService(f.db, { nativeSessionBackendFactory: backend });
+      try {
+        await heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0, companyId, leaseId: f.leaseId });
+        await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
+        const current = await readSource(f);
+        expect(current.run.resultJson?.workspaceRestoreRecovery).toBeUndefined();
+        if (mode === "null_stop_intent") expect(current.lease.metadata?.sandboxStopAndRetain).toBeNull();
+        if (mode === "null_export_intent") expect(current.lease.metadata?.nativeWorkspaceExportResume).toBeNull();
+        await sourceHeld(f);
+        expect(stopped.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0);
+        expect(destroyed.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0); expect(backend).not.toHaveBeenCalled();
+      } finally { await heartbeat.drainActiveRunExecutions(); stopped.mockRestore(); destroyed.mockRestore(); }
+    });
+
+  it("holds malformed workspace authority without throwing or recording a new pin", async () => {
+    const f = await seedSource("MALFORMED-WORKSPACE");
+    await f.db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '0')` })
+      .where(eq(executionWorkspaces.id, f.workspaceId));
+    await expect(pinFailedObservedWorkspace(f.db, companyId, f.runId)).resolves.toBe(false);
+    expect((await readSource(f)).run.resultJson?.workspaceRestoreRecovery).toBeUndefined();
+    await sourceHeld(f);
+  });
+
+  it("keeps a poisoned workspace held and continues the actual orphan page to its next valid source", async () => {
+    const bad = await seedSource("POISONED-PAGE-FIRST"), good = await seedSource("VALID-PAGE-NEXT"), db = bad.db;
+    await db.update(environmentLeases).set({ updatedAt: new Date(Date.now() - 7 * 60_000) }).where(eq(environmentLeases.id, bad.leaseId));
+    await db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '0')` })
+      .where(eq(executionWorkspaces.id, bad.workspaceId));
+    const provider = getSandboxProvider("fake")!, stopped = vi.spyOn(provider, "stopLease"), destroyed = vi.spyOn(provider, "destroyLease");
+    const backend = vi.fn(() => { throw new Error("Unexpected native execution"); });
+    const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: backend });
+    try {
+      await expect(heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0, companyId })).resolves.toMatchObject({ recovered: 1 });
+      expect((await readSource(bad)).run.resultJson?.workspaceRestoreRecovery).toBeUndefined();
+      expect((await readSource(bad)).lease.status).toBe("active");
+      expect((await readSource(good)).run.resultJson?.workspaceRestoreRecovery).toMatchObject({ leaseIds: [good.leaseId] });
+      expect((await readSource(good)).lease).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" });
+      await sourceHeld(bad);
+      expect(stopped.mock.calls.filter(([input]) => input.providerLeaseId === bad.providerLeaseId)).toHaveLength(0);
+      expect(destroyed.mock.calls.filter(([input]) => input.providerLeaseId === bad.providerLeaseId)).toHaveLength(0);
+      expect(backend).not.toHaveBeenCalled();
+    } finally { await heartbeat.drainActiveRunExecutions(); stopped.mockRestore(); destroyed.mockRestore(); }
+  });
+
+  it.each(["before_pin", "after_pin"] as const)("serializes the real ephemeral checkpoint claim %s", async order => {
+    const f = await seedSource(`CLAIM-${order}`);
+    if (order === "before_pin") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
+    else expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true);
+    const dispositions = await claimNativeRestartRecoveries({ db: f.db, restartKind: "hard", runIds: [f.runId], controller: fixture.current.successor });
+    if (order === "before_pin") {
+      expect(dispositions).toMatchObject([{ kind: "reattach_remote_runner", runId: f.runId }]);
+      expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(false);
+      const { run, lease, owner } = await readSource(f);
+      expect(run.status).toBe("running"); expect(owner.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now());
+      expect(owner.controllerGeneration).toBe(4); expect(lease.status).toBe("active");
+      expect(lease.metadata?.sandboxStopAndRetain).toBeUndefined();
+    } else {
+      expect(dispositions).toMatchObject([{ kind: "blocked", runId: f.runId, reason: "issue_execution_lock_changed" }]);
+      expect((await readSource(f)).owner).toMatchObject({ phase: "observed", leaseOwner: "old-observed-controller", controllerGeneration: 3 });
+    }
+  });
+
+  it("rechecks a new physical allocation owner after pin before the actual Stop dispatch", async () => {
+    const f = await seedSource("ARRIVING-OWNER");
+    expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true);
+    const other = await otherAllocationOwner(f);
+    const provider = getSandboxProvider("fake")!, stopped = vi.spyOn(provider, "stopLease"), destroyed = vi.spyOn(provider, "destroyLease");
+    const heartbeat = heartbeatService(f.db);
+    try {
+      await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
+      expect(stopped.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0);
+        expect(destroyed.mock.calls.filter(([input]) => input.providerLeaseId === f.providerLeaseId)).toHaveLength(0);
+      expect(await settleStoppedObservedWorkspace(f.db, companyId, f.runId)).toBe(false);
+      expect((await readSource(f)).lease).toMatchObject({ status: "pending_cleanup", metadata: { pendingCleanupInFlight: false } });
+      expect((await f.db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, other.runId)))[0].status).toBe("active");
+      await sourceHeld(f);
+    } finally { await heartbeat.drainActiveRunExecutions(); stopped.mockRestore(); destroyed.mockRestore(); }
+  });
+
+  it.each(["generation", "checkpoint", "workspace_authority", "accepted_result", "other_owner", "lost_stop_ack"] as const)(
+    "does not settle observed ownership from stale or missing Stop evidence (%s)", async mode => {
+      const f = await seedSource(`STOP-FENCE-${mode}`);
+      expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true);
+      const original = (await f.db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId),
+        eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))))[0];
+      expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true);
+      expect((await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, original.id)))[0]).toEqual(original);
+      const provider = getSandboxProvider("fake")!;
+      const stop = mode === "lost_stop_ack" ? vi.spyOn(provider, "stopLease").mockRejectedValue(new Error("Controlled lost acknowledgement")) : null;
+      const heartbeat = heartbeatService(f.db);
+      try {
+        await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 });
+        expect(hasRemoteTerminationReceipt((await readSource(f)).lease)).toBe(mode !== "lost_stop_ack");
+        if (mode === "generation") await f.db.update(nativeRunFinalizations).set({ controllerGeneration: 4 }).where(eq(nativeRunFinalizations.runId, f.runId));
+        if (mode === "checkpoint") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, sessionCheckpoint: { ...f.checkpoint, activeTurnId: "changed" } } }).where(eq(heartbeatRuns.id, f.runId));
+        if (mode === "workspace_authority") await f.db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, '{remoteWorkspace,epoch}', '0')` }).where(eq(executionWorkspaces.id, f.workspaceId));
+        if (mode === "accepted_result") await acceptedResult(f);
+        if (mode === "other_owner") await otherAllocationOwner(f);
+        expect(await settleStoppedObservedWorkspace(f.db, companyId, f.runId)).toBe(false);
+        expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(mode === "lost_stop_ack");
+        const current = (await f.db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, original.id)))[0];
+        expect(current.evidence.observedWorkspacePreservation).toEqual(original.evidence.observedWorkspacePreservation);
+        expect((await readSource(f)).owner.phase).toBe("observed");
+        expect(await readFile(path.join(f.remoteCwd, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n");
+      } finally { await heartbeat.drainActiveRunExecutions(); stop?.mockRestore(); }
+    });
+
+  it("backs off a real cleanup producer conflict only after rolling the preservation transaction back", async () => {
+    const f = await seedSource("CLEANUP-ROW-BUSY"), entered = barrier(), release = barrier();
+    const peer = f.db.transaction(async tx => {
+      await environmentService(tx as unknown as typeof f.db).releaseLease(f.leaseId, "released", { cleanupStatus: "success" });
+      entered.resolve(); await release.promise;
+    });
+    try {
+      await entered.promise;
+      await expect(pinFailedObservedWorkspace(f.db, companyId, f.runId)).rejects.toBeInstanceOf(ObservedWorkspacePreservationBusyError);
+      expect((await readSource(f)).run.resultJson?.workspaceRestoreRecovery).toBeUndefined();
+    } finally { release.resolve(); await peer; }
+    expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(false);
+    await sourceHeld(f);
+  });
+
+});

server/src/services/heartbeat.ts

@@ -1993,7 +1993,8 @@ export function heartbeatService(
     const preserveNativeWorkspaceOnFailure = async () => {
       const failureRun = await getRun(run.id);
       if (failureRun?.runtimeMode === "native"
-        && parseObject(failureRun.runnerProfileJson?.nativeWorkspaceSync).state === "prepared") {
+        && (parseObject(failureRun.runnerProfileJson?.nativeWorkspaceSync).state === "prepared"
+          || parseObject(failureRun.runnerProfileJson?.remoteWorkspace).state === "prepared")) {
         if (!nativeRunHasNoWorkspaceAccess(failureRun)) providerResourceDispositionForRun = "stop_and_retain";
         try {
           await preserveNativeFailedWorkspace(db, failureRun, {

server/src/services/native-runtime/native-environment-settlement-work.ts

@@ -1,4 +1,4 @@
-import { and, asc, eq, exists, gt, inArray, or, sql } from "drizzle-orm";
+import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm";
 import { environmentLeases, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
 import type { ProviderResourceDisposition } from "../environment-runtime.js";
 import { nextRunLeaseReleaseAt, pendingRunLeaseReleaseCondition } from "../environment-lease-release-state.js";
@@ -7,6 +7,9 @@ import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
 import type { NativeEnvironmentSettlementScope } from "../native-environment-settlement-notifications.js";
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
 import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
+import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
+import { hasConfirmedSandboxStopAndRetain } from "../sandbox-stop-and-retain.js";
+import { settleStoppedObservedWorkspace } from "./native-observed-workspace-preservation.js";
 
 type Run = Pick<typeof heartbeatRuns.$inferSelect, "runnerProfileJson" | "resultJson">;
 export function nativeEnvironmentSettlementDisposition(run: Run, succeeded: boolean): ProviderResourceDisposition {
@@ -38,17 +41,23 @@ export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settleme
     }).from(nativeRunFinalizations)
       .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
         eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
-      .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId),
+      .leftJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId),
         eq(nativeRunResults.runId, nativeRunFinalizations.runId), eq(nativeRunResults.companyId, nativeRunFinalizations.companyId),
         eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), eq(nativeRunResults.schemaStatus, "accepted")))
       .where(and(eq(heartbeatRuns.runtimeMode, "native"),
         inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]),
-        or(eq(nativeRunFinalizations.phase, "committed"),
-          and(eq(nativeRunFinalizations.phase, "terminal_failure"),
-            sql`${nativeRunFinalizations.failureDetail} ? 'workspaceFinalizeAttempt'`)),
-        exists(db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
-          eq(environmentLeases.companyId, heartbeatRuns.companyId), eq(environmentLeases.heartbeatRunId, heartbeatRuns.id),
-          pendingRunLeaseReleaseCondition(true)))),
+        or(and(isNotNull(nativeRunResults.id),
+          or(eq(nativeRunFinalizations.phase, "committed"),
+            and(eq(nativeRunFinalizations.phase, "terminal_failure"),
+              sql`${nativeRunFinalizations.failureDetail} ? 'workspaceFinalizeAttempt'`)),
+          exists(db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
+            eq(environmentLeases.companyId, heartbeatRuns.companyId), eq(environmentLeases.heartbeatRunId, heartbeatRuns.id),
+            pendingRunLeaseReleaseCondition(true))))),
+          and(eq(heartbeatRuns.status, "failed"), eq(nativeRunFinalizations.phase, "observed"),
+            isNull(nativeRunFinalizations.resultId), isNotNull(nativeRunFinalizations.leaseOwner),
+            lte(nativeRunFinalizations.leaseExpiresAt, new Date()),
+            sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,schema}' = 'paperclip.remote-workspace-run.v1'`,
+            sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,state}' = 'prepared'`)),
         scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
         scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined,
         scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
@@ -56,6 +65,12 @@ export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settleme
       )).orderBy(asc(heartbeatRuns.id)).limit(100);
   }
   async function entry(row: Awaited<ReturnType<typeof candidates>>[number]): Promise<Entry | null> {
+    if (row.phase === "observed") {
+      const leases = await db.select().from(environmentLeases).where(and(
+        eq(environmentLeases.companyId, row.companyId), eq(environmentLeases.heartbeatRunId, row.id)));
+      return leases.length === 1 && leases.every(lease => hasRemoteTerminationReceipt(lease) && hasConfirmedSandboxStopAndRetain(lease))
+        ? { id: row.id, companyId: row.companyId, issueId: row.issueId, dueAt: 0 } : null;
+    }
     const dueAt = await nextRunLeaseReleaseAt(db, row.companyId, row.id,
       nativeEnvironmentSettlementDisposition(row, row.phase === "committed"));
     if (dueAt === null) { retryFloors.delete(row.id); return null; }
@@ -86,6 +101,10 @@ export function nativeEnvironmentSettlementWork(db: Db, settle: (input: Settleme
     const scope = { companyId: current.companyId, issueId: current.issueId, runId: current.id };
     const [row] = await candidates(scope);
     if (!row) { retryFloors.delete(current.id); return null; }
+    if (row.phase === "observed") {
+      await settleStoppedObservedWorkspace(db, current.companyId, current.id);
+      return null;
+    }
     const pending = await entry(row);
     if (!pending || pending.dueAt > Date.now()) return pending;
     if (signal?.aborted) throw new Error("Native environment settlement stopped");

server/src/services/native-runtime/native-failed-workspace-recovery.ts

@@ -15,6 +15,7 @@ import { hasAcknowledgedNativeRunStopReceipt } from "../acknowledged-native-stop
 import { logger } from "../../middleware/logger.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
 import { preserveRemoteWorkspacePreparationFailure } from "./native-workspace-finalization-state.js";
+import { pinFailedObservedWorkspace } from "./native-observed-workspace-preservation.js";
 
 
 function workspaceRecoveryReference(profile: Record<string, unknown> | null | undefined) {
@@ -143,6 +144,7 @@ export async function preserveOrphanedNativeWorkspace(db: Db, lease: Pick<typeof
       return Boolean(retained && hasRequiredWorkspaceRecovery(retained.resultJson));
     }
     if (run.status !== "failed") return false;
+    if (await pinFailedObservedWorkspace(db, run.companyId, run.id)) return true;
     const retained = await preserveNativeFailedWorkspace(db, run);
     return Boolean(retained && hasRequiredWorkspaceRecovery(retained.resultJson));
   } catch (error) {

server/src/services/native-runtime/native-observed-workspace-preservation.ts

@@ -0,0 +1,225 @@
+import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
+import { companies, environmentLeases, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db";
+import { remoteWorkspaceAuthoritySchema, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared";
+import { adapterExecutionControls } from "../adapter-execution-control.js";
+import { logActivity } from "../activity-log.js";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
+import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js";
+import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
+import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js";
+import { hasConfirmedSandboxStopAndRetain, hasStopOnlyCleanup } from "../sandbox-stop-and-retain.js";
+import { hasRequiredWorkspaceRecovery } from "../workspace-restore-recovery-state.js";
+import { nativeSha256 } from "./canonical.js";
+import { NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT, NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "./native-runner-ownership.js";
+
+const SCHEMA = "paperclip.observed-workspace-preservation.v1";
+const FIELD = "observedWorkspacePreservation";
+type Run = typeof heartbeatRuns.$inferSelect;
+type Coordinator = typeof nativeRunFinalizations.$inferSelect;
+type Fence = { schema: typeof SCHEMA; state: "stop_pending" | "stopped"; runId: string;
+  leaseOwner: string; leaseExpiresAt: string; controllerGeneration: number; providerAttempt: number; sourceSha256: string; coordinatorSha256: string;
+  leaseId: string; provider: string; providerLeaseId: string; environmentId: string; workspaceId: string };
+
+export class ObservedWorkspacePreservationBusyError extends Error {
+  readonly code = "55P03";
+  constructor(options?: ErrorOptions) { super("observed_workspace_preservation_busy", options); }
+}
+
+async function preservationTransaction<T>(db: Db, operation: (tx: Db) => Promise<T>): Promise<T> {
+  try {
+    return await db.transaction(async transaction => {
+      const tx = transaction as unknown as Db;
+      await tx.execute(sql`select set_config('statement_timeout', '15000', true), set_config('lock_timeout', '1000', true)`);
+      return operation(tx);
+    });
+  } catch (error) {
+    const failure = error as { code?: unknown; cause?: { code?: unknown } };
+    if (failure?.code === "55P03" || failure?.cause?.code === "55P03") throw new ObservedWorkspacePreservationBusyError({ cause: error });
+    throw error;
+  }
+}
+
+function sourceSha256(run: Run) {
+  const profile = run.runnerProfileJson ?? {};
+  return nativeSha256({ runId: run.id, companyId: run.companyId, agentId: run.agentId, issueId: run.nativeIssueId,
+    nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId,
+    processPid: run.processPid, processGroupId: run.processGroupId, processStartedAt: run.processStartedAt?.toISOString() ?? null,
+    remoteWorkspace: profile.remoteWorkspace, nativeExecutionInput: profile.nativeExecutionInput,
+    sessionCheckpoint: profile.sessionCheckpoint, nativeControllerRecovery: profile.nativeControllerRecovery,
+    errorCode: run.errorCode, error: run.error, finishedAt: run.finishedAt?.toISOString() ?? null });
+}
+
+function coordinatorSha256(coordinator: Coordinator) {
+  return nativeSha256({ controllerBootId: coordinator.controllerBootId, controllerPid: coordinator.controllerPid,
+    controllerProcessStartedAt: coordinator.controllerProcessStartedAt?.toISOString() ?? null,
+    controllerGeneration: coordinator.controllerGeneration, attempt: coordinator.attempt,
+    leaseOwner: coordinator.leaseOwner, leaseExpiresAt: coordinator.leaseExpiresAt?.toISOString() ?? null,
+    phase: coordinator.phase, resultId: coordinator.resultId, recoveryState: coordinator.recoveryState,
+    recoveryRequestId: coordinator.recoveryRequestId, recoveryHistory: coordinator.recoveryHistory,
+    failureCode: coordinator.failureCode, failureDetail: coordinator.failureDetail,
+    nextAttemptAt: coordinator.nextAttemptAt?.toISOString() ?? null, controlDeadlineAt: coordinator.controlDeadlineAt?.toISOString() ?? null });
+}
+
+function readFence(value: unknown): Fence | null {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
+  const fence = value as Fence;
+  return fence.schema === SCHEMA && ["stop_pending", "stopped"].includes(fence.state)
+    && [fence.runId, fence.leaseOwner, fence.leaseId, fence.provider, fence.providerLeaseId, fence.environmentId, fence.workspaceId]
+      .every(item => typeof item === "string" && Boolean(item))
+    && Number.isFinite(Date.parse(fence.leaseExpiresAt)) && /^[a-f0-9]{64}$/.test(fence.sourceSha256) && /^[a-f0-9]{64}$/.test(fence.coordinatorSha256)
+    && Number.isSafeInteger(fence.controllerGeneration) && fence.controllerGeneration >= 0
+    && Number.isSafeInteger(fence.providerAttempt) && fence.providerAttempt >= 0 ? fence : null;
+}
+
+function matchesObserved(run: Run, coordinator: Coordinator | undefined, now: Date) {
+  return run.runtimeMode === "native" && run.status === "failed" && run.nativePhase === "observed" && Boolean(run.nativeIssueId)
+    && run.errorCode !== NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE && run.errorCode !== NATIVE_ADOPTED_RUNNER_AUTHENTICATION_TIMEOUT
+    && !adapterExecutionControls.has(run.id)
+    && run.resultJson?.nativeCancellation == null && run.resultJson?.startupCancellation == null
+    && run.resultJson?.executionCancellation == null && run.resultJson?.cancellation == null
+    && !(run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > now)
+    && run.runnerProfileJson?.nativeWorkspaceSync == null && run.runnerProfileJson?.nativeWorkspaceFinalizationOwner == null
+    && coordinator?.companyId === run.companyId && coordinator.issueId === run.nativeIssueId
+    && coordinator.runId === run.id && coordinator.phase === "observed" && coordinator.resultId === null
+    && coordinator.assessmentId === null && coordinator.decisionId === null && coordinator.failureCode === null
+    && coordinator.recoveryState === "awaiting_runner_reattach" && coordinator.nextAttemptAt === null && coordinator.controlDeadlineAt === null
+    && Boolean(coordinator.leaseOwner) && Boolean(coordinator.leaseExpiresAt && coordinator.leaseExpiresAt <= now);
+}
+
+async function lockSource(db: Db, companyId: string, runId: string) {
+  await db.select({ id: companies.id }).from(companies).where(eq(companies.id, companyId)).for("key share", { noWait: true });
+  const [identity] = await db.select({ issueId: heartbeatRuns.nativeIssueId }).from(heartbeatRuns).where(and(
+    eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId)));
+  if (!identity?.issueId) return null;
+  const [task] = await db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, identity.issueId))).for("update", { noWait: true });
+  const [coordinator] = await db.select().from(nativeRunFinalizations).where(and(
+    eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId),
+    eq(nativeRunFinalizations.issueId, identity.issueId))).for("update", { noWait: true });
+  const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))).for("update", { noWait: true });
+  if (!task || !run || task.status !== "blocked" || task.assigneeAgentId !== run.agentId || task.assigneeUserId
+    || task.executionRunId || task.checkoutRunId || !matchesObserved(run, coordinator, new Date())) return null;
+  const [accepted] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and(
+    eq(nativeRunResults.companyId, companyId), eq(nativeRunResults.runId, runId), eq(nativeRunResults.schemaStatus, "accepted")))
+    .limit(1).for("update", { noWait: true });
+  if (accepted) return null;
+  const parsed = remoteWorkspaceRunReferenceSchema.safeParse(run.runnerProfileJson?.remoteWorkspace);
+  if (!parsed.success || parsed.data.state !== "prepared" || task.executionWorkspaceId !== parsed.data.workspaceId) return null;
+  const reference = parsed.data;
+  const [workspace] = await db.select().from(executionWorkspaces).where(and(
+    eq(executionWorkspaces.companyId, companyId), eq(executionWorkspaces.id, reference.workspaceId))).for("update", { noWait: true });
+  const parsedAuthority = remoteWorkspaceAuthoritySchema.safeParse(workspace?.metadata?.remoteWorkspace);
+  const authority = parsedAuthority.success ? parsedAuthority.data : null;
+  if (!workspace || workspace.sourceIssueId !== task.id || !authority?.location || authority.epoch !== reference.epoch
+    || authority.snapshotId !== reference.snapshotId || !authority.dirty || authority.writerRunId !== run.id
+    || authority.location.leaseId !== reference.leaseId || authority.location.providerLeaseId !== reference.providerLeaseId
+    || authority.location.remoteCwd !== reference.remoteCwd) return null;
+  const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
+    eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id),
+    inArray(workspaceOperations.status, ["pending", "running"]))).limit(1);
+  if (operation) return null;
+  const leases = await db.select().from(environmentLeases).where(and(
+    eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id))).orderBy(asc(environmentLeases.id));
+  if (leases.length !== 1) return null;
+  const selected = leases[0];
+  if (!selected.provider || !selected.providerLeaseId) return null;
+  const history = await db.select().from(environmentLeases).where(and(
+    eq(environmentLeases.provider, selected.provider), eq(environmentLeases.providerLeaseId, selected.providerLeaseId)))
+    .orderBy(asc(environmentLeases.id)).for("update", { noWait: true });
+  if (history.length !== 1 || history[0].id !== selected.id) return null;
+  const lease = history[0];
+  if (lease.id !== reference.leaseId || lease.issueId !== task.id || lease.executionWorkspaceId !== workspace.id
+    || lease.environmentId !== authority.location.environmentId || !lease.environmentId
+    || !lease.provider || lease.provider === "local" || lease.providerLeaseId !== reference.providerLeaseId
+    || lease.metadata?.remoteCwd !== reference.remoteCwd || lease.metadata?.agentId !== run.agentId
+    || (lease.metadata?.pendingCleanupInFlight != null && lease.metadata.pendingCleanupInFlight !== false)) return null;
+  const [writer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+    eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
+    sql`(${heartbeatRuns.nativeIssueId} = ${task.id}::uuid or ${heartbeatRuns.issueId} = ${task.id}::uuid
+      or ${heartbeatRuns.contextSnapshot}->>'issueId' = ${task.id}
+      or ${heartbeatRuns.contextSnapshot}->>'executionWorkspaceId' = ${workspace.id})`)).limit(1);
+  if (writer) return null;
+  return { run, task, coordinator: coordinator!, reference, workspace, lease };
+}
+
+function fenceMatches(fence: Fence, source: NonNullable<Awaited<ReturnType<typeof lockSource>>>, evidence: Record<string, unknown> | undefined) {
+  const { run, coordinator, lease } = source;
+  const recorded = run.resultJson?.workspaceRestoreRecovery as { schema?: unknown; leaseIds?: unknown } | undefined;
+  const action = evidence?.workspaceRestoreRecovery as { schema?: unknown; leaseIds?: unknown } | undefined;
+  const retained = lease.metadata?.workspaceRestoreRecovery as { schema?: unknown; runId?: unknown } | undefined;
+  const exact = (value: typeof recorded) => value?.schema === "paperclip.workspace-restore-recovery.v1"
+    && Array.isArray(value.leaseIds) && value.leaseIds.length === 1 && value.leaseIds[0] === lease.id;
+  return lease.leasePolicy === "retain_on_failure" && exact(recorded) && exact(action) && retained?.schema === "paperclip.workspace-restore-recovery.v1" && retained.runId === run.id
+    && fence.runId === run.id && fence.sourceSha256 === sourceSha256(run) && fence.coordinatorSha256 === coordinatorSha256(coordinator)
+    && fence.leaseOwner === coordinator.leaseOwner && fence.leaseExpiresAt === coordinator.leaseExpiresAt?.toISOString()
+    && fence.controllerGeneration === coordinator.controllerGeneration && fence.providerAttempt === coordinator.attempt
+    && fence.leaseId === lease.id && fence.provider === lease.provider && fence.providerLeaseId === lease.providerLeaseId
+    && fence.environmentId === lease.environmentId && fence.workspaceId === source.workspace.id;
+}
+
+export async function pinFailedObservedWorkspace(db: Db, companyId: string, runId: string): Promise<boolean> {
+  return preservationTransaction(db, async tx => {
+    const source = await lockSource(tx, companyId, runId);
+    if (!source) return false;
+    const { run, task, coordinator, workspace, lease } = source;
+    const [existing] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId),
+      eq(issueRecoveryActions.sourceIssueId, task.id), eq(issueRecoveryActions.fingerprint, `native-workspace:${run.id}`))).for("update", { noWait: true });
+    if (hasRequiredWorkspaceRecovery(run.resultJson)) {
+      const fence = readFence(existing?.evidence[FIELD]);
+      return Boolean(fence?.state === "stop_pending" && fenceMatches(fence, source, existing?.evidence));
+    }
+    const active = lease.leasePolicy === "ephemeral" && lease.status === "active" && lease.releasedAt === null && lease.cleanupStatus === null
+      && !hasStopOnlyCleanup(lease) && lease.metadata?.sandboxStopAndRetainReceipt == null
+      && lease.metadata?.remoteExecutionTermination == null && lease.metadata?.pendingCleanupAttemptId == null
+      && (lease.metadata?.pendingCleanupLeaseExpiresAtMs == null || lease.metadata.pendingCleanupLeaseExpiresAtMs === 0)
+      && lease.metadata?.workspaceRecoveryCleanup == null;
+    if (!active) return false;
+    await notifyNativeEnvironmentSettlement(tx, { companyId, runId, issueId: task.id });
+    const retained = await recordRequiredWorkspaceRestoreFailure(tx, run, { workspaceRestoreFailure: "restore_failed" }, lease);
+    if (!retained || !hasRequiredWorkspaceRecovery(retained.resultJson)) throw new Error("observed_workspace_retention_missing");
+    const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId),
+      eq(issueRecoveryActions.sourceIssueId, task.id), eq(issueRecoveryActions.fingerprint, `native-workspace:${run.id}`))).for("update", { noWait: true });
+    if (!action) throw new Error("observed_workspace_retention_action_missing");
+    const fence: Fence = { schema: SCHEMA, state: "stop_pending", runId, leaseOwner: coordinator.leaseOwner!,
+      leaseExpiresAt: coordinator.leaseExpiresAt!.toISOString(), controllerGeneration: coordinator.controllerGeneration,
+      providerAttempt: coordinator.attempt, sourceSha256: sourceSha256(run), coordinatorSha256: coordinatorSha256(coordinator), leaseId: lease.id,
+      provider: lease.provider!, providerLeaseId: lease.providerLeaseId!, environmentId: lease.environmentId!, workspaceId: workspace.id };
+    await tx.update(issueRecoveryActions).set({ evidence: { ...action.evidence, [FIELD]: fence }, updatedAt: new Date() })
+      .where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.id, action.id)));
+    await logActivity(tx, { companyId, actorType: "system", actorId: "native-workspace-recovery",
+      action: "issue.observed_workspace_preservation_pinned", entityType: "issue", entityId: task.id, runId,
+      details: { recoveryActionId: action.id, sourceRunId: runId, controllerGeneration: fence.controllerGeneration } });
+    return true;
+  });
+}
+
+export async function settleStoppedObservedWorkspace(db: Db, companyId: string, runId: string): Promise<boolean> {
+  return preservationTransaction(db, async tx => {
+    const source = await lockSource(tx, companyId, runId);
+    if (!source) return false;
+    const { run, task, coordinator, lease } = source;
+    const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId),
+      eq(issueRecoveryActions.sourceIssueId, task.id), eq(issueRecoveryActions.fingerprint, `native-workspace:${run.id}`))).for("update", { noWait: true });
+    const fence = readFence(action?.evidence[FIELD]);
+    if (!fence || fence.state !== "stop_pending" || !hasRequiredWorkspaceRecovery(run.resultJson)
+      || !fenceMatches(fence, source, action?.evidence)
+      || !hasRemoteTerminationReceipt(lease) || !hasConfirmedSandboxStopAndRetain(lease)) return false;
+    await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
+    await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionFinalization);
+    const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null,
+      leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null, recoveryState: "blocked",
+      failureCode: "native_continuation_requires_reconciliation", updatedAt: new Date() }).where(and(
+        eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId),
+        eq(nativeRunFinalizations.issueId, task.id), eq(nativeRunFinalizations.phase, "observed"), isNull(nativeRunFinalizations.resultId),
+        eq(nativeRunFinalizations.leaseOwner, fence.leaseOwner), eq(nativeRunFinalizations.controllerGeneration, fence.controllerGeneration),
+        eq(nativeRunFinalizations.attempt, fence.providerAttempt))).returning({ id: nativeRunFinalizations.runId });
+    if (!settled) throw new Error("observed_workspace_preservation_fence_changed");
+    await tx.update(heartbeatRuns).set({ nativePhase: "terminal_failure", nativePhaseUpdatedAt: new Date(), updatedAt: new Date() })
+      .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId)));
+    await tx.update(issueRecoveryActions).set({ evidence: { ...action!.evidence, [FIELD]: { ...fence, state: "stopped" } }, updatedAt: new Date() })
+      .where(and(eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.id, action!.id)));
+    await logActivity(tx, { companyId, actorType: "system", actorId: "native-workspace-recovery",
+      action: "issue.observed_workspace_stop_verified", entityType: "issue", entityId: task.id, runId,
+      details: { recoveryActionId: action!.id, sourceRunId: runId, controllerGeneration: fence.controllerGeneration } });
+    return true;
+  });
+}

server/src/services/native-runtime/native-workspace-finalization-state.ts

@@ -46,13 +46,20 @@ export async function preserveRemoteWorkspacePreparationFailure(db: Db, lease: W
 
 export async function nativeWorkspaceExportPendingForLease(db: Db, lease: WorkspaceSourceLease): Promise<boolean> {
   if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return false;
-  if (await preparedRemoteWorkspaceSource(db, lease)) return true;
   const input = sql`${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput'`;
   const [run] = await db.select({
     id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
     issueId: heartbeatRuns.issueId, nativeIssueId: heartbeatRuns.nativeIssueId,
     runtimeMode: heartbeatRuns.runtimeMode, driverKind: heartbeatRuns.driverKind,
     reference: sql<unknown>`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`,
+    observedRemoteWorkspace: sql<boolean>`${heartbeatRuns.status} = 'failed'
+      and ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace' is not null
+      and ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace' <> 'null'::jsonb
+      and exists (select 1 from ${nativeRunFinalizations}
+        where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId}
+          and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}
+          and ${nativeRunFinalizations.issueId} = ${heartbeatRuns.nativeIssueId}
+          and ${nativeRunFinalizations.phase} = 'observed' and ${nativeRunFinalizations.resultId} is null)`,
     executionScope: sql<Record<string, unknown>>`jsonb_build_object(
       'schema', ${input}->'schema', 'binding', ${input}->'binding', 'workspace', ${input}->'workspace',
       'provider', jsonb_build_object('kind', ${input}#>'{provider,kind}'))`,
@@ -60,6 +67,8 @@ export async function nativeWorkspaceExportPendingForLease(db: Db, lease: Worksp
     eq(heartbeatRuns.id, lease.heartbeatRunId), eq(heartbeatRuns.companyId, lease.companyId),
     eq(heartbeatRuns.runtimeMode, "native"),
   )).limit(1);
+  if (run?.observedRemoteWorkspace) return true;
+  if (await preparedRemoteWorkspaceSource(db, lease)) return true;
   if (!run || nativeRunHasNoWorkspaceAccess({ ...run, runnerProfileJson: { nativeExecutionInput: run.executionScope } })) return false;
   const reference = readNativeWorkspaceSyncReference(run.reference);
   return reference?.state === "prepared" && reference.leaseId === lease.id