feat(recovery): settle runtime-control operations of proven-gone owners on the shared path (#20016)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A managed runtime control (start, stop, restart, repair, run) on an
execution workspace writes a `running` operation row that blocks every
other control on that workspace
> - If the server that ran it crashes, the row stays `running` until the
next control request sweeps it. The sweep had its own `kill(pid, 0)`
check, with no process start time, so a reused PID looked alive until 60
s of heartbeat silence passed
> - The durable ownership plan proves owners by identity through one
shared prover and settles them from one reconciler
> - This pull request stamps the owner's boot id and start time, judges
it with `proveOwnerGone`, and registers a settle for the kind
> - The benefit is that a workspace stranded by a crashed server frees
itself in the next reconciler pass, and a reused PID no longer holds it

## Linked Issues or Issue Description

Refs #19831 (plan, Phase 2 sections A and C), #19838 (inventory:
"Managed runtime-control operation", and the "One fact, many proofs"
table row `ownerProcessIsAlive` workspace-operations.ts).

**What happened?**
`workspace_runtime_control_operation` was a report-only kind with a
bespoke liveness check: `ownerProcessProvenGone` probed the PID with
signal 0 and had no start time. A server restart that reused the PID
kept the operation "alive" until its heartbeat aged out, and nothing
settled the row until another control request arrived.

**Expected behavior**
The owner is proven by the shared prover: a dead or reused PID in the
stamp's PID space is gone. Another host, a stamp without a PID space, or
a live PID without a recorded start time stays unprovable and keeps the
heartbeat lease as its fenced takeover at the next control.

## What Changed

- `workspace-operations.ts`:
- The owner stamp adds the server's `bootId` and `processStartedAt`
(`currentRuntimeControlOwner`, no I/O).
- The new `runtimeControlOperationOwnerStamp` is shared with the
registry.
- The sweep proves the owner with `proveOwnerGone`.
`ownerProcessProvenGone` is deleted.
- The terminal CAS and log append are factored out
(`claimRuntimeControlTermination`,
`appendRuntimeControlReconciliationLog`).
- New `settleRuntimeControlOperation` for the reconciler: it skips
operations this process is driving, requires the same owner as the
proof, CAS-fails the row and writes a `workspace_operation.reconciled`
activity entry.
- `recovery/ownership-registry.ts`: the kind uses the shared stamp and
registers the settle.
- The settling-kinds list in `ownership-reconciliation-report.test.ts`
is updated.

## Verification

- `workspace-runtime-control-recovery.test.ts`: 21 pass. The 6 new tests
are in "shared ownership path". 5 fail on `master` and pass here:
  - a reused PID is recovered at once;
  - the reconciler settles a dead owner with activity and frees control;
- an unprovable owner is never settled, and a read-only pass changes
nothing;
  - a re-stamped owner is kept;
  - new operations carry the boot id and start time.

  The live-owner guard passes on both.
- `ownership-reconciliation-report.test.ts` (25),
`workspace-operations-reconciliation.test.ts` (9) and
`execution-workspace-runtime-control-conflict.test.ts` (24) pass.
- The server package typecheck passes.

## Risks

- Rows written before this change have no start time. A live PID on such
a row stays unprovable, and the 60 s heartbeat lease still applies at
the next control. A dead PID is still proof.
- The reconciler now terminalizes these rows. It does so only on `gone`,
and only with the same owner and an `updated_at` CAS.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code
execution in Claude Code.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

server/src/__tests__/ownership-reconciliation-report.test.ts

@@ -143,10 +143,7 @@ describe("report-only ownership reconciliation", () => {
     expect(new Set(names).size).toBe(names.length);
     expect(names.length).toBeGreaterThanOrEqual(40);
     expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id))
-      .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent",
-        "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox",
-        "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "wake_request_claimed", "legacy_controller_lease",
-        "issue_execution_lock", "issue_checkout_lock"]);
+      .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "wake_request_claimed", "legacy_controller_lease", "issue_execution_lock", "issue_checkout_lock"]);
     expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true);
   });
 

server/src/__tests__/workspace-runtime-control-recovery.test.ts

@@ -4,6 +4,7 @@ import os from "node:os";
 import path from "node:path";
 import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
 import {
+  activityLog,
   companies,
   createDb,
   executionWorkspaces,
@@ -12,6 +13,8 @@ import {
 } from "@paperclipai/db";
 import { and, eq } from "drizzle-orm";
 import { setTimeout as delay } from "node:timers/promises";
+import { subscribeDatabaseWork } from "@paperclipai/db/work-signals";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import {
   RUNTIME_CONTROL_STALE_AFTER_MS,
   WorkspaceOperationTimeoutError,
@@ -19,7 +22,12 @@ import {
   workspaceOperationService,
   workspaceRuntimeControlOwnerIdForTests,
 } from "../services/workspace-operations.js";
-import { currentProcessSpaceHash } from "../services/controller-process-identity.js";
+import { currentControllerProcessStamp, currentProcessSpaceHash, currentProcessStartedAtMs } from "../services/controller-process-identity.js";
+import {
+  createOwnershipReportState,
+  runOwnershipReconciliationPass,
+} from "../services/recovery/ownership-invariant-report.js";
+import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js";
 import { getWorkspaceOperationLogStore } from "../services/workspace-operation-log-store.js";
 import {
   getEmbeddedPostgresTestSupport,
@@ -60,6 +68,7 @@ describeEmbeddedPostgres("managed runtime-control operation recovery", () => {
 
   afterEach(async () => {
     resetWorkspaceRuntimeControlStateForTests();
+    await db.delete(activityLog);
     await db.delete(workspaceOperations);
     await db.delete(executionWorkspaces);
     await db.delete(projects);
@@ -126,7 +135,8 @@ describeEmbeddedPostgres("managed runtime-control operation recovery", () => {
     companyId: string;
     executionWorkspaceId: string;
     /** hostHash defaults to this process's PID space; null omits it (a legacy stamp). */
-    owner?: { ownerId: string; pid: number; heartbeatAt: Date; hostHash?: string | null };
+    owner?: { ownerId: string; pid: number; heartbeatAt: Date; hostHash?: string | null;
+      bootId?: string; processStartedAt?: string };
     withLog?: boolean;
     startedAt?: Date;
     /**
@@ -162,6 +172,8 @@ describeEmbeddedPostgres("managed runtime-control operation recovery", () => {
                 ...(input.owner.hostHash === null ? {} : { hostHash: input.owner.hostHash ?? currentProcessSpaceHash() }),
                 action: "start",
                 heartbeatAt: input.owner.heartbeatAt.toISOString(),
+                ...(input.owner.bootId ? { bootId: input.owner.bootId } : {}),
+                ...(input.owner.processStartedAt ? { processStartedAt: input.owner.processStartedAt } : {}),
               },
             }
           : {}),
@@ -538,4 +550,121 @@ describeEmbeddedPostgres("managed runtime-control operation recovery", () => {
       .where(eq(workspaceOperations.executionWorkspaceId, executionWorkspaceId));
     expect(statuses.map((row) => row.status).sort()).toEqual(["failed", "succeeded"]);
   });
+
+  describe("shared ownership path (plan Phase 2)", () => {
+    const kind = OWNABLE_KINDS.find((entry) => entry.id === "workspace_runtime_control_operation")!;
+    // The server resolves its own start time at boot; the stamp reads it without I/O.
+    beforeAll(async () => { await currentProcessStartedAtMs(); });
+    const pass = (record = true) => runOwnershipReconciliationPass(db, { state: createOwnershipReportState(),
+      kinds: [kind], record });
+    // A previous server whose PID this test process now reuses: same PID space,
+    // same PID, but it started long before this process did.
+    const reusedPidOwner = () => ({ ownerId: randomUUID(), pid: process.pid, heartbeatAt: new Date(),
+      bootId: randomUUID(), processStartedAt: "2000-01-01T00:00:00.000Z" });
+
+    it("recovers at once a start whose owner's PID now belongs to a later process", async () => {
+      const seeded = await seedWorkspace();
+      const staleId = await seedRunningStart({ ...seeded, owner: reusedPidOwner() });
+      const service = workspaceOperationService(db);
+      await expect(service.reconcileStaleRuntimeControlOperations(seeded.executionWorkspaceId)).resolves.toEqual({
+        reconciled: 1, operationIds: [staleId] });
+    });
+
+    it("keeps a start whose owner PID and start time still match a live process", async () => {
+      const seeded = await seedWorkspace();
+      const controller = currentControllerProcessStamp();
+      expect(controller.processStartedAt).toBeTruthy();
+      await seedRunningStart({ ...seeded, owner: { ownerId: randomUUID(), pid: process.pid, heartbeatAt: new Date(),
+        bootId: randomUUID(), processStartedAt: controller.processStartedAt! } });
+      await expect(workspaceOperationService(db).reconcileStaleRuntimeControlOperations(seeded.executionWorkspaceId))
+        .resolves.toEqual({ reconciled: 0, operationIds: [] });
+      const report = await pass();
+      expect(report.kinds[0]).toMatchObject({ inProgress: 1, settled: 0 });
+    });
+
+    it("settles a start whose owner server is gone without waiting for the next control", async () => {
+      const seeded = await seedWorkspace();
+      const staleId = await seedRunningStart({ ...seeded, withLog: true,
+        owner: { ownerId: randomUUID(), pid: 4_194_303, heartbeatAt: new Date() } });
+
+      const report = await pass();
+      expect(report.kinds[0]).toMatchObject({ kind: "workspace_runtime_control_operation", inProgress: 1, settled: 1 });
+      const row = await workspaceOperationService(db).getById(staleId);
+      expect(row).toMatchObject({ status: "failed", metadata: { reconciled: true,
+        reconciliationReason: "orphaned_runtime_control", reconciledBy: "ownership-reconciler", ownerProof: "pid_dead" } });
+      expect(row?.stderrExcerpt).toContain("(pid 4194303) is gone (pid_dead)");
+      const activity = await db.select().from(activityLog).where(eq(activityLog.entityId, staleId));
+      expect(activity).toMatchObject([{ action: "workspace_operation.reconciled", actorId: "ownership-reconciler" }]);
+      await expect(workspaceOperationService(db).assertRuntimeControlAvailable({
+        executionWorkspaceId: seeded.executionWorkspaceId, action: "start" })).resolves.toBeUndefined();
+    });
+
+    it("never settles an owner it cannot prove, and a read-only pass changes nothing", async () => {
+      const seeded = await seedWorkspace();
+      const otherHost = await seedRunningStart({ ...seeded,
+        owner: { ownerId: randomUUID(), pid: 4_194_303, heartbeatAt: new Date(0), hostHash: "another-host-pid-space" } });
+      const proven = await seedRunningStart({ ...seeded,
+        owner: { ownerId: randomUUID(), pid: 4_194_303, heartbeatAt: new Date() } });
+
+      const readOnly = await pass(false);
+      expect(readOnly.kinds[0]).toMatchObject({ inProgress: 2, settled: 0 });
+      expect((await workspaceOperationService(db).getById(proven))?.status).toBe("running");
+
+      await pass();
+      // Even with a long-lapsed heartbeat, age is not proof on the shared path.
+      expect((await workspaceOperationService(db).getById(otherHost))?.status).toBe("running");
+      expect((await workspaceOperationService(db).getById(proven))?.status).toBe("failed");
+    });
+
+    it("keeps an operation whose owner re-stamped after the proof", async () => {
+      const seeded = await seedWorkspace();
+      const id = await seedRunningStart({ ...seeded, owner: { ownerId: randomUUID(), pid: 4_194_303, heartbeatAt: new Date() } });
+      const [record] = (await kind.listInProgress!(db, 10, 0)).records;
+      const controller = currentControllerProcessStamp();
+      await db.update(workspaceOperations).set({ metadata: { action: "start", runtimeControlOwner: {
+        ownerId: randomUUID(), pid: process.pid, hostHash: controller.hostHash, bootId: controller.bootId,
+        processStartedAt: controller.processStartedAt, action: "start", heartbeatAt: new Date().toISOString() } } })
+        .where(eq(workspaceOperations.id, id));
+      await expect(kind.settle!(db, record!, { now: new Date(), proof: { verdict: "gone", reason: "pid_dead" } }))
+        .resolves.toEqual({ settled: false, reason: "owner_changed_since_proof" });
+      expect((await workspaceOperationService(db).getById(id))?.status).toBe("running");
+    });
+
+    it("wakes the ownership worker when a runtime control starts, before its row is visible", async () => {
+      const seeded = await seedWorkspace();
+      const recorder = workspaceOperationService(db).createRecorder({ companyId: seeded.companyId,
+        executionWorkspaceId: seeded.executionWorkspaceId });
+      const hints: string[] = [];
+      const stop = subscribeDatabaseWork(db, DELIVERY_QUEUES.ownershipInvariantReport, (event, eventHints) => {
+        if (event === "settled") hints.push(...(eventHints ?? []));
+      });
+      let operationId: string | undefined;
+      try {
+        await recorder.recordOperation({ phase: "workspace_provision", metadata: { action: "start" }, run: async () => {
+          operationId = (await db.select().from(workspaceOperations).where(eq(workspaceOperations.status, "running")))[0]?.id;
+          return { status: "succeeded" };
+        } });
+        for (let attempt = 0; attempt < 50 && !hints.some(hint => operationId && hint.includes(operationId)); attempt += 1) {
+          await delay(20);
+        }
+      } finally { stop(); }
+      expect(operationId).toBeTruthy();
+      expect(hints.some(hint => hint.includes(operationId!))).toBe(true);
+    });
+
+    it("stamps new operations with the server's boot id and start time", async () => {
+      const seeded = await seedWorkspace();
+      const service = workspaceOperationService(db);
+      const recorder = service.createRecorder({ companyId: seeded.companyId, executionWorkspaceId: seeded.executionWorkspaceId });
+      let seen: Record<string, unknown> | null = null;
+      await recorder.recordOperation({ phase: "workspace_provision", metadata: { action: "start" }, run: async () => {
+        const [row] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.status, "running"));
+        seen = (row?.metadata as Record<string, unknown>).runtimeControlOwner as Record<string, unknown>;
+        return { status: "succeeded" };
+      } });
+      const controller = currentControllerProcessStamp();
+      expect(seen).toMatchObject({ pid: process.pid, hostHash: controller.hostHash, bootId: controller.bootId,
+        processStartedAt: controller.processStartedAt });
+    });
+  });
 });

server/src/services/recovery/ownership-registry.ts

@@ -33,6 +33,7 @@ import {
 } from "../native-runtime/native-accepted-result-lease-settlement.js";
 import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js";
 import { settleAbandonedWorkspaceFinalizeAttempt } from "./workspace-finalize-attempt-settlement.js";
+import { runtimeControlOperationOwnerStamp, settleRuntimeControlOperation } from "../workspace-operations.js";
 import { nativeControllerOwnerStamp, runProcessOwnerStamp } from "../native-runtime/native-owner-stamps.js";
 import { settleIssueLock } from "./issue-lock-settlement.js";
 import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js";
@@ -700,17 +701,19 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [
     row => controllerStamp(json(row.metadata)[ENVIRONMENT_LEASE_CONTROLLER_KEY], "environment_login_probe_lease")),
 
   // Workspaces
-  operationKind("workspace_runtime_control_operation", "Managed runtime-control operation", "ownership",
-    and(eq(workspaceOperations.status, "running"), sql`${workspaceOperations.metadata} ? 'runtimeControlOwner'`),
-    row => {
-      const owner = json(json(row.metadata).runtimeControlOwner);
-      const pid = int(owner.pid);
-      // The owner stores a per-process UUID, a PID and, since #19868, its PID
-      // space; no start time, so a live PID stays unprovable.
-      return pid ? { kind: "process", stamp: { kind: "local", scope: "workspace_runtime_control_operation",
-        generation: null, host: null, processSpace: str(owner.hostHash), bootId: str(owner.ownerId), pid,
-        processStartedAt: null, processGroupId: null } } : { kind: "none" };
-    }),
+  {
+    // Shared path since Phase 2: the owner stamp names the server process and,
+    // since this kind moved, its boot id and start time, so a dead or reused
+    // PID is proof. Another host or a stamp without a PID space stays
+    // unprovable; the pre-control sweep's heartbeat lease is its fenced takeover.
+    ...operationKind("workspace_runtime_control_operation", "Managed runtime-control operation", "ownership",
+      and(eq(workspaceOperations.status, "running"), sql`${workspaceOperations.metadata} ? 'runtimeControlOwner'`),
+      row => {
+        const stamp = runtimeControlOperationOwnerStamp(row.metadata);
+        return stamp ? { kind: "process", stamp } : { kind: "none" };
+      }),
+    settle: settleRuntimeControlOperation,
+  },
   {
     // Shared path since Phase 2: a finalize attempt whose process died (a
     // restart between start and copyback, or between copyback and done) is

server/src/services/workspace-operations.ts

@@ -11,9 +11,12 @@ import { asc, desc, eq, gte, inArray, isNull, lt, or, and } from "drizzle-orm";
 import { conflict, notFound } from "../errors.js";
 import { redactCurrentUserText, redactCurrentUserValue } from "../log-redaction.js";
 import { instanceSettingsService } from "./instance-settings.js";
-import { currentControllerProcessStamp, currentProcessSpaceHash, processExists, sameProcessSpace } from "./controller-process-identity.js";
+import { currentControllerProcessStamp } from "./controller-process-identity.js";
+import { proveOwnerGone, type LocalOwnerStamp, type OwnerProof } from "./owner-proof.js";
 import { getWorkspaceOperationLogStore } from "./workspace-operation-log-store.js";
+import { persistActivity, publishActivity, type ActivityPublication } from "./activity-log.js";
 import { crashPoint } from "@paperclipai/shared/crash-points";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 type WorkspaceOperationRow = typeof workspaceOperations.$inferSelect;
 
@@ -75,8 +78,21 @@ type RuntimeControlOwnerStamp = {
   heartbeatAt: string;
   /** PID space of `pid`. Absent on stamps written before it was recorded. */
   hostHash?: string;
+  /** Server boot id and process start time. Absent on stamps written before
+   * they were recorded; such an owner with a live PID stays unprovable. */
+  bootId?: string;
+  processStartedAt?: string | null;
 };
 
+function currentRuntimeControlOwner(action: string, heartbeatAt: Date): RuntimeControlOwnerStamp {
+  const controller = currentControllerProcessStamp();
+  return {
+    ownerId: RUNTIME_CONTROL_OWNER_ID, pid: controller.pid, hostHash: controller.hostHash,
+    bootId: controller.bootId, processStartedAt: controller.processStartedAt,
+    action, heartbeatAt: heartbeatAt.toISOString(),
+  };
+}
+
 function readRuntimeControlAction(metadata: unknown): string | null {
   if (!metadata || typeof metadata !== "object") return null;
   const action = (metadata as Record<string, unknown>).action;
@@ -96,6 +112,23 @@ function readRuntimeControlOwner(metadata: unknown): RuntimeControlOwnerStamp |
     action: typeof record.action === "string" ? record.action : "start",
     heartbeatAt: typeof record.heartbeatAt === "string" ? record.heartbeatAt : new Date(0).toISOString(),
     ...(typeof record.hostHash === "string" ? { hostHash: record.hostHash } : {}),
+    ...(typeof record.bootId === "string" ? { bootId: record.bootId } : {}),
+    ...(typeof record.processStartedAt === "string" ? { processStartedAt: record.processStartedAt } : {}),
+  };
+}
+
+/**
+ * The shared ownership stamp of a runtime-control operation's owner, or null
+ * when the row names no owner process. Used by the sweep below and by the
+ * ownership registry, so both judge the owner with the same prover.
+ */
+export function runtimeControlOperationOwnerStamp(metadata: unknown): LocalOwnerStamp | null {
+  const owner = readRuntimeControlOwner(metadata);
+  if (!owner || !Number.isInteger(owner.pid) || owner.pid <= 0) return null;
+  return {
+    kind: "local", scope: "workspace_runtime_control_operation", generation: null, host: null,
+    processSpace: owner.hostHash ?? null, bootId: owner.bootId ?? null, pid: owner.pid,
+    processStartedAt: owner.processStartedAt ?? null, processGroupId: null,
   };
 }
 
@@ -133,20 +166,6 @@ export function isNativeWorkspaceFinalizationOperationActive(scope: NativeFinali
     && active.runId === scope.runId && active.issueId === scope.issueId);
 }
 
-/**
- * Whether the process that stamped an operation has provably exited. A dead owner can never
- * heartbeat again, so its operation is recoverable immediately instead of after the staleness
- * window — which is what makes recovery after a crash or restart prompt. The PID is judged only
- * in the PID space that recorded it: an owner on another host or namespace, an unreadable
- * space, or a stamp without a recorded space is never proven gone here and waits out the
- * staleness window instead. Signal 0 only probes; EPERM means the process exists.
- */
-function ownerProcessProvenGone(owner: RuntimeControlOwnerStamp): boolean {
-  if (!sameProcessSpace(owner.hostHash)) return false;
-  if (!Number.isInteger(owner.pid) || owner.pid <= 0) return false;
-  return !processExists(owner.pid);
-}
-
 /**
  * Compare-and-swap predicate for an operation's `updated_at`.
  *
@@ -285,6 +304,114 @@ export interface WorkspaceOperationRecorder {
   }): Promise<WorkspaceOperation>;
 }
 
+/**
+ * Compare-and-swap a `running` runtime-control operation to `failed`. A live owner that
+ * heartbeats between the caller's read and this write moves `updated_at` and keeps it.
+ */
+async function claimRuntimeControlTermination(db: Db, candidate: WorkspaceOperationRow, input: {
+  reason: string; message: string; finishedAt: Date; provenance?: Record<string, unknown>;
+}): Promise<boolean> {
+  const claimed = await db
+    .update(workspaceOperations)
+    .set({
+      status: "failed",
+      stderrExcerpt: appendExcerpt(candidate.stderrExcerpt ?? "", input.message),
+      metadata: combineMetadata(candidate.metadata as Record<string, unknown> | null, {
+        reconciled: true,
+        reconciliationReason: input.reason,
+        reconciledAt: input.finishedAt.toISOString(),
+        ...(input.provenance ?? {}),
+      }),
+      finishedAt: input.finishedAt,
+      updatedAt: input.finishedAt,
+    })
+    .where(
+      and(
+        eq(workspaceOperations.id, candidate.id),
+        eq(workspaceOperations.status, "running"),
+        candidate.updatedAt
+          ? updatedAtUnchanged(candidate.updatedAt)
+          : isNull(workspaceOperations.updatedAt),
+      ),
+    )
+    .returning({ id: workspaceOperations.id })
+    .then((rows) => rows[0] ?? null);
+  if (claimed) liveRuntimeControlOperationIds.delete(candidate.id);
+  return claimed !== null;
+}
+
+async function appendRuntimeControlReconciliationLog(db: Db, candidate: WorkspaceOperationRow, message: string,
+  finishedAt: Date) {
+  if (!candidate.logStore || !candidate.logRef) return;
+  const logStore = getWorkspaceOperationLogStore();
+  const handle = { store: candidate.logStore as "local_file", logRef: candidate.logRef };
+  try {
+    await logStore.append(handle, { stream: "stderr", chunk: message, ts: finishedAt.toISOString() });
+    const finalized = await logStore.finalize(handle);
+    await db
+      .update(workspaceOperations)
+      .set({
+        logBytes: finalized.bytes,
+        logSha256: finalized.sha256,
+        logCompressed: finalized.compressed,
+        updatedAt: new Date(),
+      })
+      .where(eq(workspaceOperations.id, candidate.id));
+  } catch {
+    // The terminal row and stderr excerpt stay inspectable even when the external log
+    // file from the previous process is gone.
+  }
+}
+
+/**
+ * Settle of the `workspace_runtime_control_operation` ownership kind (plan Phase 2).
+ *
+ * The shared reconciler calls this only after the stamped owner process was proven gone
+ * (dead or reused PID in its own PID space). It fails exactly that operation, the same
+ * way the pre-control sweep does, so a workspace stranded by a crashed server frees
+ * itself without waiting for the next control request. It never touches an operation
+ * this process is driving, and an owner that re-stamped since the proof keeps its row.
+ */
+export async function settleRuntimeControlOperation(db: Db, record: { recordId: string; owner: { kind: string; stamp?: unknown } },
+  context: { now: Date; proof: OwnerProof }): Promise<{ settled: boolean; reason: string }> {
+  if (liveRuntimeControlOperationIds.has(record.recordId)) return { settled: false, reason: "operation_live_in_this_process" };
+  if (record.owner.kind !== "process") return { settled: false, reason: "owner_not_provable" };
+  const proven = record.owner.stamp as LocalOwnerStamp;
+  const candidate = await db.select().from(workspaceOperations)
+    .where(and(eq(workspaceOperations.id, record.recordId), eq(workspaceOperations.status, "running")))
+    .then((rows) => rows[0] ?? null);
+  if (!candidate) return { settled: false, reason: "operation_no_longer_running" };
+  const action = readRuntimeControlAction(candidate.metadata);
+  const current = runtimeControlOperationOwnerStamp(candidate.metadata);
+  if (!action || !current || current.pid !== proven.pid || current.bootId !== proven.bootId
+    || current.processSpace !== proven.processSpace || current.processStartedAt !== proven.processStartedAt) {
+    return { settled: false, reason: "owner_changed_since_proof" };
+  }
+  const finishedAt = context.now;
+  const message = `Reconciled stale managed runtime ${action} operation: the owning server process (pid ${current.pid}) `
+    + `is gone (${context.proof.reason}), so the operation was failed to release the execution workspace for retry or stop.\n`;
+  let publication: ActivityPublication | undefined;
+  const claimed = await db.transaction(async (transaction) => {
+    const tx = transaction as unknown as Db;
+    if (!await claimRuntimeControlTermination(tx, candidate, {
+      reason: "orphaned_runtime_control", message, finishedAt,
+      provenance: { reconciledBy: "ownership-reconciler", ownerProof: context.proof.reason },
+    })) return false;
+    publication = (await persistActivity(tx, {
+      companyId: candidate.companyId, actorType: "system", actorId: "ownership-reconciler",
+      action: "workspace_operation.reconciled", entityType: "workspace_operation", entityId: candidate.id,
+      runId: candidate.heartbeatRunId, issueId: candidate.issueId,
+      details: { reason: "orphaned_runtime_control", action, proof: context.proof.reason,
+        executionWorkspaceId: candidate.executionWorkspaceId },
+    })).publication;
+    return true;
+  });
+  if (!claimed) return { settled: false, reason: "operation_no_longer_running" };
+  if (publication) publishActivity(publication);
+  await appendRuntimeControlReconciliationLog(db, candidate, message, finishedAt);
+  return { settled: true, reason: context.proof.reason };
+}
+
 export function workspaceOperationService(db: Db) {
   const instanceSettings = instanceSettingsService(db);
   const logStore = getWorkspaceOperationLogStore();
@@ -348,9 +475,12 @@ export function workspaceOperationService(db: Db) {
       const ownedByThisProcess = owner?.ownerId === RUNTIME_CONTROL_OWNER_ID;
       const abandonedByThisProcess = ownedByThisProcess && !liveRuntimeControlOperationIds.has(candidate.id);
       const ownedHereAndLive = ownedByThisProcess && liveRuntimeControlOperationIds.has(candidate.id);
-      // A previous server process that no longer exists cannot heartbeat again, so its
-      // operations are recoverable at once instead of after the staleness window.
-      const ownerProcessGone = owner !== null && !ownedByThisProcess && ownerProcessProvenGone(owner);
+      // A previous server process proven gone (shared prover: dead or reused PID in
+      // this PID space) cannot heartbeat again, so its operation is recoverable at
+      // once. An unprovable owner (another host, no recorded PID space, a live PID
+      // with no start time) keeps the heartbeat lease below as its takeover fence.
+      const ownerStamp = owner && !ownedByThisProcess ? runtimeControlOperationOwnerStamp(candidate.metadata) : null;
+      const ownerProcessGone = ownerStamp !== null && (await proveOwnerGone(ownerStamp)).verdict === "gone";
 
       if (ownedHereAndLive) {
         // This process is driving it right now; its own time budget is what terminalizes it.
@@ -371,70 +501,18 @@ export function workspaceOperationService(db: Db) {
               ? "the owning request no longer exists"
               : `no live owner has reported progress for ${Math.round(silentForMs / 1000)}s`
         }, so the operation was failed to release the execution workspace for retry or stop.\n`;
-      // Compare-and-swap on `updated_at`: a live owner that heartbeats between the read and
-      // this write moves the column and keeps its operation.
-      const claimed = await db
-        .update(workspaceOperations)
-        .set({
-          status: "failed",
-          stderrExcerpt: appendExcerpt(candidate.stderrExcerpt ?? "", reconciliationMessage),
-          metadata: combineMetadata(candidate.metadata as Record<string, unknown> | null, {
-            reconciled: true,
-            reconciliationReason: abandonedByThisProcess
-              ? "abandoned_runtime_control"
-              : "orphaned_runtime_control",
-            reconciledAt: finishedAt.toISOString(),
-          }),
-          finishedAt,
-          updatedAt: finishedAt,
-        })
-        .where(
-          and(
-            eq(workspaceOperations.id, candidate.id),
-            eq(workspaceOperations.status, "running"),
-            candidate.updatedAt
-              ? updatedAtUnchanged(candidate.updatedAt)
-              : isNull(workspaceOperations.updatedAt),
-          ),
-        )
-        .returning({ id: workspaceOperations.id })
-        .then((rows) => rows[0] ?? null);
+      const claimed = await claimRuntimeControlTermination(db, candidate, {
+        reason: abandonedByThisProcess ? "abandoned_runtime_control" : "orphaned_runtime_control",
+        message: reconciliationMessage, finishedAt,
+      });
       if (!claimed) {
         // Someone else moved the row; re-read so callers see the current truth.
         const refreshed = await getById(candidate.id);
         if (refreshed?.status === "running") live.push(refreshed);
         continue;
       }
-
-      liveRuntimeControlOperationIds.delete(candidate.id);
       reconciledIds.push(candidate.id);
-
-      if (candidate.logStore && candidate.logRef) {
-        const handle = {
-          store: candidate.logStore as "local_file",
-          logRef: candidate.logRef,
-        };
-        try {
-          await logStore.append(handle, {
-            stream: "stderr",
-            chunk: reconciliationMessage,
-            ts: finishedAt.toISOString(),
-          });
-          const finalized = await logStore.finalize(handle);
-          await db
-            .update(workspaceOperations)
-            .set({
-              logBytes: finalized.bytes,
-              logSha256: finalized.sha256,
-              logCompressed: finalized.compressed,
-              updatedAt: new Date(),
-            })
-            .where(eq(workspaceOperations.id, candidate.id));
-        } catch {
-          // The terminal row and stderr excerpt stay inspectable even when the external log
-          // file from the previous process is gone.
-        }
-      }
+      await appendRuntimeControlReconciliationLog(db, candidate, reconciliationMessage, finishedAt);
     }
 
     return { reconciled: reconciledIds.length, operationIds: reconciledIds, live };
@@ -554,13 +632,7 @@ export function workspaceOperationService(db: Db) {
           let currentMetadata = runtimeControlAction
             ? {
                 ...(recordInput.metadata ?? {}),
-                runtimeControlOwner: {
-                  ownerId: RUNTIME_CONTROL_OWNER_ID,
-                  pid: process.pid,
-                  hostHash: currentProcessSpaceHash(),
-                  action: runtimeControlAction,
-                  heartbeatAt: startedAt.toISOString(),
-                } satisfies RuntimeControlOwnerStamp,
+                runtimeControlOwner: currentRuntimeControlOwner(runtimeControlAction, startedAt),
               }
             : recordInput.metadata ?? null;
           // Finalize attempts record the process that runs them (no I/O), so
@@ -578,7 +650,7 @@ export function workspaceOperationService(db: Db) {
 
           try {
             let published = false;
-            const publish = async () => { await writeState(database => database.insert(workspaceOperations).values({
+            const insertRunning = (database: Db) => database.insert(workspaceOperations).values({
               id,
               companyId: input.companyId,
               executionWorkspaceId,
@@ -596,7 +668,21 @@ export function workspaceOperationService(db: Db) {
               ) as Record<string, unknown> | null,
               startedAt,
               updatedAt: startedAt,
-            })); published = true; };
+            });
+            const publish = async () => {
+              if (runtimeControlAction) {
+                // Wake the ownership worker (on any server) before the row it
+                // announces, so a quiet worker starts watching this owner.
+                await db.transaction(async tx => {
+                  await notifyDeliveryWork(tx, DELIVERY_QUEUES.ownershipInvariantReport,
+                    JSON.stringify({ companyId: input.companyId, operationId: id }));
+                  await insertRunning(tx as unknown as Db);
+                });
+              } else {
+                await writeState(insertRunning);
+              }
+              published = true;
+            };
             if (recordInput.admit) await recordInput.admit(publish);
             else await publish();
             if (!published) throw new Error("Workspace operation admission did not publish its running row");
@@ -617,13 +703,7 @@ export function workspaceOperationService(db: Db) {
             heartbeatTimer = setInterval(() => {
               const heartbeatAt = new Date();
               currentMetadata = combineMetadata(currentMetadata, {
-                runtimeControlOwner: {
-                  ownerId: RUNTIME_CONTROL_OWNER_ID,
-                  pid: process.pid,
-                  hostHash: currentProcessSpaceHash(),
-                  action: runtimeControlAction,
-                  heartbeatAt: heartbeatAt.toISOString(),
-                } satisfies RuntimeControlOwnerStamp,
+                runtimeControlOwner: currentRuntimeControlOwner(runtimeControlAction, heartbeatAt),
               });
               void db
                 .update(workspaceOperations)