Merge commit '649861b5f9bfa76c3ba55336343b27231c4516f1' into codex/sleep-09-aggregate

* commit '649861b5f9bfa76c3ba55336343b27231c4516f1':
  Align process recovery authority and narrow controller renewal work

doc/plans/2026-10-30-live-recovery-sleep.md

@@ -7,7 +7,7 @@ This is the final recovery slice after run dispatch, issue state recovery, and d
 | Resultless native session | Native running/failed run; finalization `retryable_failure`; no result | Execution/finalization writes, checkpoint writes, restart claim writes, process identity events, and existing run-dispatch/deferred-eligibility boundaries for agent, task, review, dependency and pause-hold changes | Maximum saved retry/claim lease deadline. Fresh claim owner validates provider/session state. Accepted result, terminal failure, or explicit ownership gate removes the entry. |
 | Native question cancellation | Cancellable run with committed `nativeQuestionCancellation` marker | Authorized task-completion interaction transaction | Immediate. Existing cancellation owner consumes exact company/run marker. Terminal status removes debt. |
 | Active execution | Running run | Run lifecycle, controller renewal/revocation, native authority and process evidence writes | Fresh native/legacy controller expiry, saved five-minute preparation grace, or bounded physical-process probe. Local execution promises retain execution ownership. Explicit evidence/manual gates park until their owning producer changes truth. |
-| Settled native process identity | Terminal native failure/cancellation with persisted process identity | Terminal execution writes, server process identity/stop receipts, environment settlement | Immediate inspection. Only matched local host/PID/group/birth or confirmed remote stop evidence clears identity. A live matched process or local execution owner keeps a bounded probe. Missing authority waits for committed evidence. Startup reconstructs rows independently of the old capped process Map. Successful release removes any obsolete compatibility Map entry. |
+| Settled native process identity | Terminal native failure/cancellation with persisted process identity | Terminal execution writes, server process identity/stop receipts, environment settlement | Immediate inspection. Only matched local host/PID/group/birth or confirmed remote stop evidence clears identity. A live matched process or local execution owner keeps a bounded probe. Missing authority waits for committed evidence. Startup reconstructs rows independently of the old capped process Map. Legacy local PID rows lacking birth evidence remain unchanged and excluded until the actual process-metadata writer records matching identity; they grant no automatic continuation authority. Remote allocations remain eligible without local birth evidence because their exact provider stop receipts own release. Successful release removes any obsolete compatibility Map entry. |
 | Orphan active environment lease | Active lease with NULL run reference or terminal same-company run | Environment owning transactions and indexed physical-resource peer notifications | Saved lease updatedAt plus five minutes, re-read before due work. Existing export, shutdown, and shared-resource fences remain. Workspace preservation lock/statement timeouts retain a bounded thirty-second retry; claim/evidence gates still park. Exact-lease dispatch does not mutate timestamps for broad-page fairness. Successful CAS hands work to the existing pending-cleanup owner. |
 | Legacy shutdown settlement | Exact lease with pending shutdown marker | Begin, renew, finish, expiry settlement owning transactions | Exact saved marker deadline. Existing owner revalidates boot/controller, source, competing allocations and stop/copyback authority. Expiry records preserved workspace recovery; it never infers file export from process loss. |
 | Stranded agent status | Ready agent with status running | Agent configuration/lifecycle/instruction settlement, run/resource transitions, accounting acknowledgement, canonical wake status writes, and direct existing-receipt terminal writers | Saved updatedAt plus five minutes, then actual surviving controller expiry when present. Fresh settlement rechecks live execution, native controller, working copy, and lease truth. Settled status removes debt. |
@@ -43,3 +43,9 @@ Each producer regression starts with a worker already parked and no unrelated bu
 The exact-lease owner forwards an optional deadline callback through the existing preservation helper. Only PostgreSQL lock-unavailable (`55P03`) and statement-cancelled (`57014`) outcomes retain a thirty-second retry. A rolled-back write intent may cause earlier bounded reconciliation. `NativeWorkspaceFinalizationBusyError` remains an ownership refusal, and wrong-company or manual/evidence gates do not acquire periodic retries. Legacy explicit callers keep the existing boolean contract.
 
 The regression holds the real task row `FOR UPDATE` until the existing one-second lock timeout fires, then releases only the transaction, without a mutation or notification. The registered worker retains its deadline and records workspace preservation exactly once on retry, handing the original allocation to stop-only pending cleanup with `retain_on_failure`. A second fixture exceeds the actual fifteen-second statement timeout with a one-shot PostgreSQL trigger. Both require the twenty-minute zero-SQL idle proof after the known producer fanout settles; wrong-company, active-claim and manual-gate controls remain parked.
+
+## Controller renewal scope
+
+A successful legacy lease renewal signals only its active-execution run deadline. The private liveness pool reuses its already-owned transaction for statement timeout, intent, and update; the public renewal API still owns an atomic transaction. Other controller/run transitions retain their broader producer signals. Renewal still uses `clock_timestamp()` and cannot resurrect an expired lease.
+
+The actual private-pool control prewarms the connection and requires exactly five transaction statements, one XID, one active-execution topic, no savepoints, and no other recovery topics. The registered-owner control measures the bounded intent/settlement passes, proves no unrelated queue pass, crosses the prior deadline, and requires the fresh saved expiry. The existing controller suite covers real timeout retries, foreign/current boots, expired refusal, revocation, and shared-pool saturation. Dead aggregate guard branches remain for the final aggregate cutover PR to remove together.

server/src/__tests__/live-recovery-work.test.ts

@@ -17,18 +17,19 @@ import { notifyExecutionRunWork } from "../services/execution-control-notificati
 import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
 import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js";
 import { beginLegacyShutdownWorkspaceSettlement } from "../services/legacy-shutdown-workspace-settlement.js";
-import { legacyControllerBootId } from "../services/legacy-controller-lease.js";
+import { renewLegacyControllerLease, legacyControllerBootId } from "../services/legacy-controller-lease.js";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 import { createHeartbeatRunOutput } from "../services/heartbeat/run-output.js";
 import { recoveryService } from "../services/recovery/service.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
 import { persistActivity } from "../services/activity-log.js";
 import { accountRunCost } from "../services/run-cost-accounting.js";
-import { createHeartbeatLifecycle } from "../services/heartbeat/run-lifecycle.js";
+import { persistHeartbeatRunProcessMetadata, createHeartbeatLifecycle } from "../services/heartbeat/run-lifecycle.js";
 import { createAgentLifecycle, configureAgentLifecycle } from "../services/agent-lifecycle.js";
 import { issueTreeControlService } from "../services/issue-tree-control.js";
 import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
 import { issueService } from "../services/issues.js";
+import { remoteTerminationReceipt } from "../services/remote-execution-termination.js";
 import { logger } from "../middleware/logger.js";
 
 vi.setConfig({ testTimeout: 90_000 });
@@ -199,6 +200,74 @@ describe("residual recovery production matrix", () => {
       eq(heartbeatRunEvents.eventType, "native.local_process_stopped")))).toHaveLength(1);
     await quiet();
   });
+  it("excludes unverifiable local birth history until actual process identity arrives", async () => {
+    const f = await fixture(), startedAt = new Date();
+    const child = spawn(process.execPath, ["-e", "console.log('legacy-identity-proof')"], { stdio: "ignore" });
+    const pid = child.pid!; await once(child, "exit");
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      runtimeMode: "native", nativeIssueId: f.issue.id, status: "failed", finishedAt: new Date(), processPid: pid,
+      runnerProfileJson: { sessionCheckpoint: { sessionId: "legacy-unverifiable-session" } } }).returning();
+    await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, phase: "committed" });
+    const checks: string[] = [], prior = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query) => {
+      if (query.includes("sessionCheckpoint") && query.includes("for update")) checks.push(query);
+    };
+    try {
+      await start();
+      expect(checks).toEqual([]);
+      await db.transaction(tx => notifyLiveRunRecovery(tx, { companyId: f.companyId }));
+      await advance(31_000); expect(checks).toEqual([]);
+    } finally { db.$client.options.debug = prior; }
+    expect(await getRun(run.id)).toMatchObject({ processPid: pid, processStartedAt: null });
+    await quiet();
+    await persistHeartbeatRunProcessMetadata(db, run.id, { pid, processGroupId: null, startedAt: startedAt.toISOString() });
+    await advance(1_000);
+    await vi.waitFor(async () => expect((await getRun(run.id)).processPid).toBeNull());
+    expect(await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, run.id),
+      eq(heartbeatRunEvents.eventType, "native.local_process_stopped")))).toHaveLength(1);
+    await quiet();
+  });
+  it("releases remote process identity without local birth after the exact provider stop receipt", async () => {
+    const f = await fixture();
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      runtimeMode: "native", nativeIssueId: f.issue.id, status: "failed", finishedAt: new Date(), processPid: 12345,
+      runnerProfileJson: { sessionCheckpoint: { sessionId: "remote-no-local-birth" } } }).returning();
+    await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, phase: "committed" });
+    const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id,
+      provider: "daytona", providerLeaseId: randomUUID(), status: "released", releasedAt: new Date() }).returning();
+    await start(); expect((await getRun(run.id)).processPid).toBe(12345);
+    const receipt = remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" });
+    expect(receipt).toBeDefined();
+    await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ cleanupStatus: "success",
+      metadata: { remoteExecutionTermination: receipt } }).where(eq(environmentLeases.id, lease.id)),
+    { companyId: f.companyId, runId: run.id, leaseId: lease.id });
+    await advance(1_000);
+    await vi.waitFor(async () => expect((await getRun(run.id)).processPid).toBeNull());
+    expect((await getRun(run.id)).processStartedAt).toBeNull();
+    expect(await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, run.id),
+      eq(heartbeatRunEvents.eventType, "native.local_process_stopped")))).toEqual([]);
+    await quiet();
+  });
+  it("renews only the active recovery owner and re-reads the moved controller deadline", async () => {
+    const f = await fixture(), old = new Date(Date.now() - 6 * 60_000), expiry = new Date(Date.now() + 20_000);
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      runtimeMode: "legacy", status: "running", controllerBootId: legacyControllerBootId,
+      controllerLeaseExpiresAt: expiry, executionStage: "preparing", startedAt: old, updatedAt: old,
+      lastOutputAt: new Date() }).returning();
+    await start(); expect(deadlines.get("active-execution-recovery")!()).toBe(expiry.getTime());
+    const before = timeline.length;
+    expect(await renewLegacyControllerLease(db, run)).toBe(true);
+    const moved = (await getRun(run.id)).controllerLeaseExpiresAt!;
+    expect(moved.getTime()).toBeGreaterThan(expiry.getTime());
+    await advance(1_000);
+    expect(timeline.slice(before).filter(event => Array.isArray(event) && event[0] === "pass")
+      .map(event => (event as unknown[])[1])).toEqual(["active-execution-recovery", "active-execution-recovery"]);
+    await advance(expiry.getTime() - Date.now());
+    expect((await getRun(run.id)).status).toBe("running");
+    expect(deadlines.get("active-execution-recovery")!()).toBe(moved.getTime());
+    await heartbeat.cancelRun(run.id, "Operator cancelled the controller renewal fixture");
+    await quiet();
+  });
   it("recovers orphan active leases after restart, including a deleted heartbeat reference, without starving behind 100 guarded rows", async () => {
     const f = await fixture(), foreign = await fixture(), resource = randomUUID(), old = new Date(Date.now() - 6 * 60_000);
     await db.insert(environmentLeases).values({ companyId: foreign.companyId, status: "retained", provider: "fixture", providerLeaseId: resource, updatedAt: old });

server/src/services/heartbeat/live-recovery-work.ts

@@ -1,4 +1,4 @@
-import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
+import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
 import { agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js";
@@ -74,7 +74,10 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof
                     sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`)
                     : and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["failed", "cancelled", "timed_out", "interrupted"]),
                       isNotNull(heartbeatRuns.finishedAt), or(isNotNull(heartbeatRuns.processPid),
-                        isNotNull(heartbeatRuns.processGroupId), isNotNull(heartbeatRuns.processStartedAt))),
+                        isNotNull(heartbeatRuns.processGroupId), isNotNull(heartbeatRuns.processStartedAt)),
+                      or(isNotNull(heartbeatRuns.processStartedAt), exists(db.select({ id: environmentLeases.id })
+                        .from(environmentLeases).where(and(eq(environmentLeases.companyId, heartbeatRuns.companyId),
+                          eq(environmentLeases.heartbeatRunId, heartbeatRuns.id), sql`${environmentLeases.provider} is distinct from 'local'`))))),
             )).orderBy(asc(heartbeatRuns.id)).limit(100);
           rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, dueAt: kind === "native"
             ? Math.max(Date.now(), row.nextAttemptAt?.getTime() ?? 0, row.leaseExpiresAt?.getTime() ?? 0) : Date.now() }));

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

@@ -1,8 +1,10 @@
 import { randomUUID } from "node:crypto";
 import { eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
-import { agents, closeLivenessDb, companies, createDb, heartbeatRuns, livenessDb } from "@paperclipai/db";
+import { agents, closeLivenessDb, companies, createDb, heartbeatRuns, livenessDb, subscribeDatabaseWork } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { heartbeatService } from "./heartbeat.js";
 import { hasLiveLegacyController, LEGACY_CONTROLLER_LEASE_MS, LEGACY_CONTROLLER_RENEW_MS, legacyControllerBootId, legacyControllerClaim, LegacyControllerLeaseLostError,
   renewLegacyControllerLease, revokeExpiredLegacyController, revokeExpiredLegacyControllerWithState,
@@ -16,15 +18,16 @@ function fakeRenewalDb(returning: () => Promise<unknown>) {
   return fake as unknown as WatchArgs[0];
 }
 
-const support = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_LEGACY_CONTROLLER_TEST_DATABASE_URL");
+const support = external ? { supported: true } : 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);
+    if (!external) database = await startEmbeddedPostgresTestDatabase("legacy-controller-");
+    db = createDb(external ?? database.connectionString);
   }, 30000);
-  afterAll(async () => { await database?.cleanup(); });
+  afterAll(async () => { await closeLivenessDb(db, 1); await db?.$client.end({ timeout: 1 }); 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)}` });
@@ -68,6 +71,26 @@ const support = await getEmbeddedPostgresTestSupport();
     expect(saved.executionStage).toBe("dispatching");
     expect(await revokeExpiredLegacyController(db, run)).toBe(false);
   });
+  it("renews through one private transaction with only the active execution topic", async () => {
+    const run = await seed(), settled: string[] = [], queries: string[] = [];
+    const queues = [DELIVERY_QUEUES.nativeProcessSessions, DELIVERY_QUEUES.nativeQuestionCancellations,
+      DELIVERY_QUEUES.activeExecutionRecovery, DELIVERY_QUEUES.settledNativeProcesses,
+      DELIVERY_QUEUES.strandedAgentStatus, DELIVERY_QUEUES.activeRunSilence];
+    const stops = queues.map(queue => subscribeDatabaseWork(db, queue, event => { if (event === "settled") settled.push(queue); }));
+    const pool = livenessDb(db), prior = pool.$client.options.debug;
+    await pool.execute(sql`select 1`);
+    pool.$client.options.debug = (_connection, query) => { queries.push(query); };
+    const watch = watchLegacyControllerLease(db, run, new AbortController());
+    try {
+      await watch.assertOwned();
+      expect(settled).toEqual([DELIVERY_QUEUES.activeExecutionRecovery]);
+      expect(queries.filter(query => /savepoint/i.test(query))).toEqual([]);
+      expect(queries.filter(query => /pg_current_xact_id/.test(query))).toHaveLength(1);
+      expect(queries.filter(query => /set_config.*statement_timeout/.test(query))).toHaveLength(1);
+      expect(queries.filter(query => /^update /i.test(query))).toHaveLength(1);
+      expect(queries).toHaveLength(5);
+    } finally { watch.stop(); pool.$client.options.debug = prior; stops.forEach(stop => stop()); }
+  });
   it("an expired controller cannot renew or dispatch even before a reaper claims it", async () => {
     const run = await seed();
     await expire(run.id);
@@ -139,7 +162,7 @@ const support = await getEmbeddedPostgresTestSupport();
   it("renews without queueing behind a slow statement on a saturated shared pool", async () => {
     const run = await seed();
     // postgres.js pipelines onto a busy connection when none is idle.
-    const shared = createDb(database.connectionString, { maxConnections: 1 });
+    const shared = createDb(external ?? database.connectionString, { maxConnections: 1 });
     const slow = shared.execute(sql`select pg_sleep(5)`).then(() => undefined); // drizzle sends on then()
     const controller = new AbortController();
     const watch = watchLegacyControllerLease(shared, run, controller);
@@ -155,7 +178,7 @@ const support = await getEmbeddedPostgresTestSupport();
 
   it("renews on one private pool per client that shutdown can end", async () => {
     const run = await seed();
-    const owner = createDb(database.connectionString);
+    const owner = createDb(external ?? database.connectionString);
     const pool = livenessDb(owner);
     expect(pool).not.toBe(owner);
     expect(livenessDb(owner)).toBe(pool);

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

@@ -1,4 +1,4 @@
-import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
+import { notifyActiveExecutionRecovery, notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, gt, lte, sql } from "drizzle-orm";
 import { heartbeatRuns, livenessDb, type Db } from "@paperclipai/db";
@@ -36,14 +36,13 @@ export function legacyControllerClaim(runtimeMode: string) {
   };
 }
 
-export async function renewLegacyControllerLease(
-  db: Db,
+async function renewLegacyControllerLeaseInTransaction(
+  tx: Db,
   run: Pick<Run, "id" | "companyId" | "controllerBootId">,
   stage?: "dispatching",
 ): Promise<boolean> {
-  const [renewed] = await db.transaction(async tx => {
-    await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id });
-    return tx.update(heartbeatRuns).set({
+  await notifyActiveExecutionRecovery(tx, { companyId: run.companyId, runId: run.id });
+  const [renewed] = await tx.update(heartbeatRuns).set({
     controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
     ...(stage ? { executionStage: stage } : {}),
   }).where(and(
@@ -52,10 +51,17 @@ export async function renewLegacyControllerLease(
     eq(heartbeatRuns.controllerBootId, legacyControllerBootId),
     gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
   )).returning({ id: heartbeatRuns.id });
-  });
   return Boolean(renewed);
 }
 
+export async function renewLegacyControllerLease(
+  db: Db,
+  run: Pick<Run, "id" | "companyId" | "controllerBootId">,
+  stage?: "dispatching",
+): Promise<boolean> {
+  return db.transaction(tx => renewLegacyControllerLeaseInTransaction(tx as unknown as Db, run, stage));
+}
+
 /** One attempt may wait at most this long; a timed-out attempt is retried
  * until the lease deadline instead of holding the renewal for the whole lease. */
 export const LEGACY_CONTROLLER_RENEWAL_STATEMENT_TIMEOUT_MS = 10_000;
@@ -77,7 +83,7 @@ async function renewOnLivenessPool(
 ): Promise<boolean> {
   return livenessDb(db).transaction(async (tx) => {
     await tx.execute(sql`select set_config('statement_timeout', ${`${statementTimeoutMs}ms`}, true)`);
-    return renewLegacyControllerLease(tx as unknown as Db, run, stage);
+    return renewLegacyControllerLeaseInTransaction(tx as unknown as Db, run, stage);
   });
 }
 

server/src/services/live-run-recovery-notifications.ts

@@ -7,6 +7,11 @@ export async function notifyActiveRunSilence(tx: object, scope?: LiveRunRecovery
     ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*");
 }
 
+export async function notifyActiveExecutionRecovery(tx: object, scope: { companyId: string; runId: string }) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.activeExecutionRecovery,
+    JSON.stringify({ companyId: scope.companyId, runId: scope.runId }));
+}
+
 export async function notifyLiveRunRecovery(tx: object, scope?: LiveRunRecoveryScope) {
   const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId }) : "*";
   for (const queue of [DELIVERY_QUEUES.nativeProcessSessions, DELIVERY_QUEUES.nativeQuestionCancellations,