From b6f58dc6e2ae8e0e2ad9009bc65a150e8336d565 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 05:07:58 -0600 Subject: [PATCH] feat(recovery): settle runtime-control operations of proven-gone owners on the shared path (#20016) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A 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 --- .../ownership-reconciliation-report.test.ts | 5 +- ...workspace-runtime-control-recovery.test.ts | 133 ++++++++- .../services/recovery/ownership-registry.ts | 25 +- server/src/services/workspace-operations.ts | 262 ++++++++++++------ 4 files changed, 317 insertions(+), 108 deletions(-) diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 4b0be8d6226..2c8cd3031a8 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/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); }); diff --git a/server/src/__tests__/workspace-runtime-control-recovery.test.ts b/server/src/__tests__/workspace-runtime-control-recovery.test.ts index 62d3a6fce05..299bd746a13 100644 --- a/server/src/__tests__/workspace-runtime-control-recovery.test.ts +++ b/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 | 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).runtimeControlOwner as Record; + return { status: "succeeded" }; + } }); + const controller = currentControllerProcessStamp(); + expect(seen).toMatchObject({ pid: process.pid, hostHash: controller.hostHash, bootId: controller.bootId, + processStartedAt: controller.processStartedAt }); + }); + }); }); diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 10b62ea54e6..bea0f867eb5 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/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 diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index b05ba90a915..3d50a8866a4 100644 --- a/server/src/services/workspace-operations.ts +++ b/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).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; } +/** + * 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; +}): Promise { + const claimed = await db + .update(workspaceOperations) + .set({ + status: "failed", + stderrExcerpt: appendExcerpt(candidate.stderrExcerpt ?? "", input.message), + metadata: combineMetadata(candidate.metadata as Record | 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 | 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 | 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)