fix: persist controller ownership before run provisioning

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/DATABASE.md

@@ -386,3 +386,14 @@ pnpm secrets:migrate-inline-env --apply
 ```
 
 Hosted AWS provider notes live in [SECRETS-AWS-PROVIDER.md](./SECRETS-AWS-PROVIDER.md).
+
+## Legacy controller ownership
+
+Legacy run claims atomically record `controller_boot_id`, a database-clock
+`controller_lease_expires_at`, and `execution_stage` before workspace provisioning.
+The lease renews independently of output. A different container must not infer
+controller death from its own process map or numeric PIDs. Expiration grants
+cleanup authority; it does not prove that remote inference has stopped. Recovery
+revokes the previous boot identity with a conditional update. Its own claim also
+expires so another sweep can finish cleanup after a restart. Historical rows keep
+null ownership fields and follow the previous recovery path.

doc/SPEC-implementation.md

@@ -1261,6 +1261,11 @@ Scheduler must skip invocation when:
 - an existing run is active
 - hard budget limit has been hit
 
+Legacy execution records a renewable controller lease when claiming a queued run,
+before provisioning. A live lease protects the run during overlapping service
+deployments. An expired controller loses dispatch authority; a recovery worker
+must establish that the previous execution stopped before starting a successor.
+
 ## 11.7 Durable agent session goals
 
 Runner Protocol v2 negotiates a required `sessionGoals` capability and typed

packages/db/src/migrations/0273_aromatic_moondragon.sql

@@ -0,0 +1,3 @@
+ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "controller_boot_id" uuid;--> statement-breakpoint
+ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "controller_lease_expires_at" timestamp with time zone;--> statement-breakpoint
+ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "execution_stage" text;

packages/db/src/migrations/meta/_journal.json

@@ -1898,6 +1898,13 @@
       "when": 1789137216452,
       "tag": "0272_light_kate_bishop",
       "breakpoints": true
+    },
+    {
+      "idx": 273,
+      "version": "7",
+      "when": 1789164595203,
+      "tag": "0273_aromatic_moondragon",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/heartbeat_runs.ts

@@ -67,6 +67,10 @@ export const heartbeatRuns = pgTable(
     stderrExcerpt: text("stderr_excerpt"),
     errorCode: text("error_code"),
     externalRunId: text("external_run_id"),
+    // Legacy controller lease. A PID alone is not an identity across containers.
+    controllerBootId: uuid("controller_boot_id"),
+    controllerLeaseExpiresAt: timestamp("controller_lease_expires_at", { withTimezone: true }),
+    executionStage: text("execution_stage"),
     processPid: integer("process_pid"),
     processGroupId: integer("process_group_id"),
     processStartedAt: timestamp("process_started_at", { withTimezone: true }),

server/src/services/heartbeat.ts

@@ -1,4 +1,5 @@
 import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js";
+import { legacyControllerBootId, legacyControllerClaim, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
 import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
 import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
 import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
@@ -16911,6 +16912,7 @@ export function heartbeatService(
                   .update(heartbeatRuns)
                   .set({
                     status: "running",
+                    ...legacyControllerClaim(run.runtimeMode),
                     responsibleUserId,
                     startedAt: lockedRun.startedAt ?? claimedAt,
                     updatedAt: claimedAt,
@@ -17007,6 +17009,7 @@ export function heartbeatService(
                 .update(heartbeatRuns)
                 .set({
                   status: "running",
+                    ...legacyControllerClaim(run.runtimeMode),
                   responsibleUserId,
                   startedAt: lockedRun.startedAt ?? claimedAt,
                   contextSnapshot: withQueuedCommentIdsInRunContext(
@@ -17073,6 +17076,7 @@ export function heartbeatService(
             .update(heartbeatRuns)
             .set({
               status: "running",
+                    ...legacyControllerClaim(run.runtimeMode),
               responsibleUserId,
               startedAt: run.startedAt ?? claimedAt,
               updatedAt: claimedAt,
@@ -18343,6 +18347,7 @@ export function heartbeatService(
       }
       if (resumedRunIds.has(run.id)) continue;
       if (locallyTracked) continue;
+      if (await hasLiveLegacyController(db, run)) continue;
 
       // Apply staleness threshold to avoid false positives
       if (staleThresholdMs > 0) {
@@ -18414,6 +18419,7 @@ export function heartbeatService(
         ((tracksLegacyLocalChild &&
           (!!run.processPid || !!run.processGroupId)) ||
           monitorDispatchLostWithoutFutureWake);
+      if (!(await revokeExpiredLegacyController(db, run))) continue;
       const baseMessage = buildProcessLossMessage(run);
       const conversationContinuationEligible = await runUsedConversationAdapter(db, run);
 
@@ -19222,8 +19228,11 @@ export function heartbeatService(
       }
     }
 
+    if (run.runtimeMode === "legacy" && run.controllerBootId &&
+        run.controllerBootId !== legacyControllerBootId) return;
     activeRunExecutions.add(run.id);
     const executionControl = createAdapterExecutionControl();
+    const controllerLease = watchLegacyControllerLease(db, run, executionControl.controller);
     let runScratch: HeartbeatRunScratch | null = null;
     let githubLauncherLocation:
       Parameters<typeof cleanupGitHubOperationLaunchers>[0] | null = null;
@@ -21039,6 +21048,7 @@ export function heartbeatService(
         ReturnType<typeof envOrchestrator.acquireForRun>
       >;
       try {
+        await controllerLease.assertOwned();
         acquiredEnvironment = await envOrchestrator.acquireForRun({
           companyId: agent.companyId,
           selectedEnvironmentId,
@@ -21050,6 +21060,7 @@ export function heartbeatService(
           persistedExecutionWorkspace,
           executionWorkspaceSettings: environmentExecutionWorkspaceSettings,
         });
+        await controllerLease.assertOwned();
         nativeRunnerPreparationSpans.push({
           name: "environment.acquire",
           parentName: "task.run",
@@ -21189,6 +21200,7 @@ export function heartbeatService(
       ): Promise<
         { dispatched: true; resultPromise: Promise<T> } | { dispatched: false }
       > => {
+        await controllerLease.assertOwned("dispatching");
         // Recheck after workspace/credential preparation, immediately before the
         // provider handoff. Never hold validation locks while adapter code runs.
         await authorizeFailedChatRetryExecution();
@@ -22570,6 +22582,7 @@ export function heartbeatService(
               })
               .onConflictDoNothing();
           });
+          controllerLease.stop();
           nativeWorkspaceSync = await prepareNativeWorkspaceSync({
             db,
             runId: run.id,
@@ -24849,6 +24862,7 @@ export function heartbeatService(
           });
         }
       }
+      controllerLease.stop();
       activeRunExecutions.delete(run.id);
       // A failed owned Stop remains visible until this exact executor settles,
       // including a graceful exit result arriving after the cancellation error.

server/src/services/legacy-controller-lease.test.ts

@@ -0,0 +1,94 @@
+import { randomUUID } from "node:crypto";
+import { eq, sql } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
+import { heartbeatService } from "./heartbeat.js";
+import { hasLiveLegacyController, legacyControllerBootId, legacyControllerClaim,
+  renewLegacyControllerLease, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+(support.supported ? describe : describe.skip)("durable legacy controller ownership", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  beforeAll(async () => {
+    database = await startEmbeddedPostgresTestDatabase("legacy-controller-");
+    db = createDb(database.connectionString);
+  }, 30000);
+  afterAll(async () => { await database?.cleanup(); });
+  async function seed() {
+    const companyId = randomUUID(), agentId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Controller test", issuePrefix: `C${companyId.slice(0, 7)}` });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Agent", role: "general", adapterType: "claude_local", status: "idle" });
+    const [queued] = await db.insert(heartbeatRuns).values({ companyId, agentId }).returning();
+    const [run] = await db.update(heartbeatRuns).set({ status: "running", ...legacyControllerClaim("legacy") })
+      .where(eq(heartbeatRuns.id, queued.id)).returning();
+    return run;
+  }
+  async function expire(id: string) {
+    await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: sql`clock_timestamp() - interval '1 second'` })
+      .where(eq(heartbeatRuns.id, id));
+  }
+  it("commits ownership with the queued claim before provisioning or logs exist", async () => {
+    const run = await seed();
+    expect(run).toMatchObject({ status: "running", controllerBootId: legacyControllerBootId, executionStage: "preparing", processPid: null });
+    expect(await hasLiveLegacyController(db, run)).toBe(true);
+    expect(await revokeExpiredLegacyController(db, run)).toBe(false);
+  });
+  it("another deployment's startup reaper preserves an unexpired controller", async () => {
+    const run = await seed();
+    await db.update(heartbeatRuns).set({ controllerBootId: randomUUID() }).where(eq(heartbeatRuns.id, run.id));
+    await heartbeatService(db).reapOrphanedRuns({ staleThresholdMs: 0 });
+    const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
+    expect(saved.status).toBe("running");
+    expect(saved.errorCode).toBeNull();
+  });
+  it("a current controller renews and records the dispatch boundary", async () => {
+    const run = await seed();
+    expect(await renewLegacyControllerLease(db, run, "dispatching")).toBe(true);
+    const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
+    expect(saved.executionStage).toBe("dispatching");
+    expect(await revokeExpiredLegacyController(db, run)).toBe(false);
+  });
+  it("an expired controller cannot renew or dispatch even before a reaper claims it", async () => {
+    const run = await seed();
+    await expire(run.id);
+    expect(await renewLegacyControllerLease(db, run, "dispatching")).toBe(false);
+    const controller = new AbortController();
+    const watch = watchLegacyControllerLease(db, run, controller);
+    try {
+      await expect(watch.assertOwned("dispatching")).rejects.toThrow("lease lost");
+      expect(controller.signal.aborted).toBe(true);
+    } finally { watch.stop(); }
+  });
+  it("only one competing recovery revokes the observed expired owner", async () => {
+    const run = await seed();
+    await expire(run.id);
+    const attempts = await Promise.all([revokeExpiredLegacyController(db, run), revokeExpiredLegacyController(db, run)]);
+    expect(attempts.filter(Boolean)).toHaveLength(1);
+    expect(await renewLegacyControllerLease(db, run)).toBe(false);
+  });
+  it("a crash after revocation permits a later recovery claim", async () => {
+    const run = await seed();
+    await expire(run.id);
+    expect(await revokeExpiredLegacyController(db, run)).toBe(true);
+    await expire(run.id);
+    const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
+    expect(await revokeExpiredLegacyController(db, saved)).toBe(true);
+  });
+  it("rejects foreign company renewal and revocation", async () => {
+    const run = await seed();
+    expect(await renewLegacyControllerLease(db, { ...run, companyId: randomUUID() })).toBe(false);
+    await expire(run.id);
+    expect(await revokeExpiredLegacyController(db, { ...run, companyId: randomUUID() })).toBe(false);
+  });
+  it("never turns a terminal run back into owned execution", async () => {
+    const run = await seed();
+    await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, run.id));
+    expect(await renewLegacyControllerLease(db, run)).toBe(false);
+    expect(await revokeExpiredLegacyController(db, run)).toBe(false);
+  });
+  it("leaves native controller ownership to the native coordinator", () => {
+    expect(legacyControllerClaim("native")).toEqual({});
+  });
+});

server/src/services/legacy-controller-lease.ts

@@ -0,0 +1,101 @@
+import { randomUUID } from "node:crypto";
+import { and, eq, gt, lte, sql } from "drizzle-orm";
+import { heartbeatRuns, type Db } from "@paperclipai/db";
+
+// A boot UUID has meaning across containers; a numeric PID does not.
+export const legacyControllerBootId = randomUUID();
+export const LEGACY_CONTROLLER_LEASE_MS = 60_000;
+export const LEGACY_CONTROLLER_RENEW_MS = 10_000;
+
+type Run = typeof heartbeatRuns.$inferSelect;
+
+/** Commit these fields in the same UPDATE that claims a queued run. */
+export function legacyControllerClaim(runtimeMode: string) {
+  if (runtimeMode === "native") return {};
+  return {
+    controllerBootId: legacyControllerBootId,
+    controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
+    executionStage: "preparing",
+  };
+}
+
+export async function renewLegacyControllerLease(
+  db: Db,
+  run: Pick<Run, "id" | "companyId" | "controllerBootId">,
+  stage?: "dispatching",
+): Promise<boolean> {
+  const [renewed] = await db.update(heartbeatRuns).set({
+    controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
+    ...(stage ? { executionStage: stage } : {}),
+  }).where(and(
+    eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
+    eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"),
+    eq(heartbeatRuns.controllerBootId, legacyControllerBootId),
+    gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
+  )).returning({ id: heartbeatRuns.id });
+  return Boolean(renewed);
+}
+
+export async function hasLiveLegacyController(db: Db, run: Run): Promise<boolean> {
+  if (run.runtimeMode === "native" || !run.controllerBootId) return false;
+  const [owner] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+    eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
+    eq(heartbeatRuns.status, "running"),
+    gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
+  ));
+  return Boolean(owner);
+}
+
+/** Atomically revoke an expired controller. Renewal and revocation serialize on
+ * the run row. Expiry permits cleanup, never dispatch of a replacement agent. */
+export async function revokeExpiredLegacyController(db: Db, run: Run): Promise<boolean> {
+  if (run.runtimeMode === "native" || !run.controllerBootId) return true;
+  const [revoked] = await db.update(heartbeatRuns).set({
+    controllerBootId: randomUUID(),
+    controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
+  }).where(and(
+    eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
+    eq(heartbeatRuns.status, "running"),
+    eq(heartbeatRuns.controllerBootId, run.controllerBootId),
+    lte(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
+  )).returning({ id: heartbeatRuns.id });
+  return Boolean(revoked);
+}
+
+/** Abort the adapter if the controller cannot renew. Bound each check by the
+ * lease duration even when the database connection never settles. */
+export function watchLegacyControllerLease(db: Db, run: Run, controller: AbortController) {
+  if (run.runtimeMode === "native" || !run.controllerBootId) {
+    return { stop() {}, async assertOwned(_stage?: "dispatching") {} };
+  }
+  let stopped = false;
+  let pending = false;
+  const lost = () => controller.abort(new Error("Legacy controller lease lost"));
+  let deadline = setTimeout(lost, Math.max(0,
+    (run.controllerLeaseExpiresAt?.getTime() ?? 0) - Date.now()));
+  deadline.unref();
+  const assertOwned = async (stage?: "dispatching") => {
+    if (stopped) return;
+    controller.signal.throwIfAborted();
+    const startedAt = Date.now();
+    const renewed = await renewLegacyControllerLease(db, run, stage);
+    if (stopped) return;
+    if (!renewed) {
+      lost();
+      controller.signal.throwIfAborted();
+    }
+    controller.signal.throwIfAborted();
+    if (!stopped) {
+      clearTimeout(deadline);
+      deadline = setTimeout(lost, Math.max(0, LEGACY_CONTROLLER_LEASE_MS - (Date.now() - startedAt)));
+      deadline.unref();
+    }
+  };
+  const timer = setInterval(() => {
+    if (pending || stopped) return;
+    pending = true;
+    void assertOwned().catch(lost).finally(() => { pending = false; });
+  }, LEGACY_CONTROLLER_RENEW_MS);
+  timer.unref();
+  return { assertOwned, stop() { stopped = true; clearInterval(timer); clearTimeout(deadline); } };
+}