Checkpoint residual recovery deadline rechecks and owning producers

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

server/src/services/agent-configuration.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifySkillVersionPinChange } from "./skill-version-retention-notifications.js";
@@ -85,6 +86,7 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura
       existing.adapterConfig,
     );
   }
+  await notifyStrandedAgentStatus(db, { companyId: existing.companyId, agentId: existing.id });
   await notifyRunDispatch(db, { companyId: existing.companyId });
   await notifyDeferredExecutionEligibility(db, { companyId: existing.companyId });
   if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) {

server/src/services/agent-instruction-working-copies.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js";
 import { notifyInstructionCollection, type InstructionCollectionScope } from "./instruction-collection-notifications.js";
 import { notifyDirectoryCleanup, type DirectoryCleanupScope } from "./directory-cleanup-notifications.js";
 import { notifyInstructionCommit, type InstructionCommitScope } from "./instruction-commit-notifications.js";
@@ -101,6 +102,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
     const signalsCleanup = directoryCleanupEntry(row) !== null || directoryCleanupEntry({ ...row, ...values }) !== null;
     const signalsCommit = row.state === "pending_commit" || values.state === "pending_commit";
     const write = async (tx: Db) => {
+      await notifyStrandedAgentStatus(tx, { companyId: row.companyId, agentId: row.agentId });
       if (signalsCommit)
         await notifyInstructionCommit(tx, { companyId: row.companyId, runId: row.runId });
       if (signalsCleanup) await notifyDirectoryCleanup(tx, { companyId: row.companyId, runId: row.runId });
@@ -112,7 +114,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
       )).returning();
       return updated;
     };
-    return signalsCommit || signalsCleanup || signalsCollection ? db.transaction(tx => write(tx as unknown as Db)) : write(db);
+    return db.transaction(tx => write(tx as unknown as Db));
   }
   async function patch(row: Copy, values: Partial<typeof copies.$inferInsert>) {
     const updated = await tryPatch(row, values);
@@ -204,6 +206,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
     if (!existing) {
       await db.transaction(async tx => {
         await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
+        await notifyStrandedAgentStatus(tx, { companyId: input.companyId, agentId: input.agentId });
       await tx.insert(copies).values({
         runId: input.runId, companyId: input.companyId, agentId: input.agentId,
         responsibleUserId: bound.onBehalfOfUserId!, entryFile,

server/src/services/agent-lifecycle.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
@@ -41,6 +42,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li
     notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle),
     async beforeStatusChange(tx, companyId, agentId, before, after) {
       if (before !== after) {
+        await notifyStrandedAgentStatus(tx, { companyId, agentId });
         await notifyRunDispatch(tx, { companyId });
         await notifyDeferredExecutionEligibility(tx, { companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);

server/src/services/environment-cleanup-notifications.ts

@@ -1,8 +1,21 @@
+import { and, eq, sql } from "drizzle-orm";
+import { environmentLeases } from "@paperclipai/db";
+import { notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js";
 import { notifyNativeEnvironmentSettlement, type NativeEnvironmentSettlementScope } from "./native-environment-settlement-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 export async function notifyEnvironmentCleanup(transaction: object, scope?: NativeEnvironmentSettlementScope): Promise<void> {
+  await notifyLiveResourceRecovery(transaction, scope);
+  if (scope?.runId && typeof (transaction as Db).select === "function") {
+    const peers = await (transaction as Db).select({ id: environmentLeases.id, companyId: environmentLeases.companyId,
+      runId: environmentLeases.heartbeatRunId }).from(environmentLeases).where(and(eq(environmentLeases.status, "active"),
+      sql`exists (select 1 from environment_leases changed where changed.company_id = ${scope.companyId}
+        and changed.heartbeat_run_id = ${scope.runId} and changed.provider_lease_id is not null
+        and changed.provider = ${environmentLeases.provider} and changed.provider_lease_id = ${environmentLeases.providerLeaseId})`));
+    for (const peer of peers) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.orphanActiveLeases,
+      JSON.stringify({ companyId: peer.companyId, leaseId: peer.id }));
+  }
   await notifyNativeEnvironmentSettlement(transaction, scope);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup);

server/src/services/execution-control-notifications.ts

@@ -1,4 +1,4 @@
-import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
+import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import type { Db } from "@paperclipai/db";
@@ -10,7 +10,7 @@ export async function notifyExecutionRunWork(transaction: object, scope?: Execut
   if (scope) {
     await notifyRunDispatch(transaction, scope);
     await notifyDeferredExecutionEligibility(transaction, scope);
-    await notifyActiveRunSilence(transaction, scope);
+    await notifyLiveRunRecovery(transaction, scope);
   }
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
@@ -28,7 +28,7 @@ export function writeExecutionControl<T>(
     await notifyRunDispatch(transaction, scope);
     if (scope) {
       await notifyDeferredExecutionEligibility(transaction, scope);
-      await notifyActiveRunSilence(transaction, scope);
+      await notifyLiveRunRecovery(transaction, scope);
     }
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);

server/src/services/heartbeat-run-events.ts

@@ -1,3 +1,4 @@
+import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
 import { readNativeEventEvidenceOutcome } from "./native-runtime/evidence-classifier.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js";
@@ -74,10 +75,11 @@ export async function appendHeartbeatRunEvent(
   db: Db,
   input: AppendHeartbeatRunEventInput,
 ): Promise<AppendHeartbeatRunEventResult> {
+  const projectsRecovery = ["native.process_start_requested", "native.process_identity_recorded", "native.local_process_stopped"].includes(input.eventType);
   const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null;
   const projectsProgress = isSafeNativeChatProgressEventType(input.eventType);
   const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent);
-  if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment) {
+  if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment && !projectsRecovery) {
     const allocated = db.$with("allocated_run_event", {}).as(sql`
       update ${heartbeatRuns}
       set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1,
@@ -117,6 +119,8 @@ export async function appendHeartbeatRunEvent(
       throw new Error("heartbeat_run_event_binding_mismatch");
     }
 
+    if (projectsRecovery) await notifyLiveRunRecovery(tx, input);
+
     if (input.retryExhaustion && !input.nativeSource) {
       // The run lock also serializes concurrent recovery checks across server
       // instances. Reusing the receipt must not allocate a sequence or publish

server/src/services/heartbeat.ts

@@ -1026,6 +1026,7 @@ export function heartbeatService(
   });
 
   const {
+    registerResidualRecoveryWork: registerResidualRecoveryOwner,
     sweepPendingCleanupLeases,
     markNativeOwnershipUnverified,
     prepareHotRestartShutdown,
@@ -4245,6 +4246,9 @@ export function heartbeatService(
     computeTaskDrain,
     applyTaskDrain,
 
+    registerResidualRecoveryWork: (coordinator: Parameters<typeof registerResidualRecoveryOwner>[0], canRun: () => boolean) =>
+      registerResidualRecoveryOwner(coordinator, () => canRun() && readTaskDrain(new Date()) === null,
+        async () => !(await getSchedulingSuppression(true)).suppressed),
     registerActiveRunSilenceWork: (coordinator: Parameters<typeof recovery.registerActiveRunSilenceWork>[0], canRun: () => boolean) =>
       recovery.registerActiveRunSilenceWork(coordinator, () => canRun() && readTaskDrain(new Date()) === null,
         async scope => recovery.scanSilentActiveRuns({ ...scope, issueCreatedAtGte: await getWorktreeExecutionCutoff() }),

server/src/services/heartbeat/agent-status.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
 import { agents, heartbeatRuns, type Db } from "@paperclipai/db";
 import { and, asc, desc, eq, lte, sql } from "drizzle-orm";
@@ -91,6 +92,7 @@ export function createHeartbeatAgentStatus(db: Db) {
       )).limit(1);
       const nextStatus = running ? "running" : settledStatus(outcome, options?.keepIdleOnFailure);
       const now = new Date();
+      await notifyStrandedAgentStatus(tx, { companyId: existing.companyId, agentId });
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
       const [updated] = await tx.update(agents).set({
         status: nextStatus,
@@ -109,7 +111,7 @@ export function createHeartbeatAgentStatus(db: Db) {
     publishStatus(result.updated, outcome);
   }
 
-  async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet<string>, scope: { companyId?: string; agentId?: string } = {}) {
+  async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet<string>, scope: { companyId?: string; agentId?: string; onDeferredDeadline?: (at: number) => void } = {}) {
     const now = new Date();
     const cutoff = new Date(now.getTime() - RECONCILIATION_GRACE_MS);
     const candidates = await db.transaction(async tx => {
@@ -158,6 +160,21 @@ export function createHeartbeatAgentStatus(db: Db) {
       for (const publication of publications) publishActivity(publication);
       publishStatus(updated, "reconciled");
     }
+    if (scope.companyId && scope.agentId && reconciled.length === 0 && scope.onDeferredDeadline) {
+      const rows = await db.execute(sql`select min(deadline) as deadline from (
+        select r.controller_lease_expires_at as deadline from heartbeat_runs r
+          where r.company_id = ${scope.companyId} and r.agent_id = ${scope.agentId}
+        union all
+        select f.lease_expires_at as deadline from native_run_finalizations f
+          inner join heartbeat_runs r on r.id = f.run_id and r.company_id = f.company_id
+          where r.company_id = ${scope.companyId} and r.agent_id = ${scope.agentId}
+      ) authority where deadline > ${now.toISOString()}::timestamptz`);
+      const deadline = rows[0]?.deadline;
+      if (deadline) {
+        const at = deadline instanceof Date ? deadline.getTime() : Date.parse(String(deadline));
+        if (Number.isFinite(at) && at > Date.now()) scope.onDeferredDeadline(at);
+      }
+    }
     return reconciled;
   }
 

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

@@ -25,16 +25,7 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof
   function register(queue: DeliveryQueue, kind: RunKind | "orphanLease" | "shutdown" | "agent") {
     const cursors = new Map<string, string>();
     let enabled = false;
-    return registerScopedDeadlineWork<Entry, Scope>(coordinator, {
-      queue,
-      reconstructionIntervalMs: 30_000,
-      beforePass: async () => { enabled = await owner.enabled(); },
-      decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint),
-      matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (
-        kind === "agent" ? !scope.agentId || entry.id === scope.agentId
-          : kind === "orphanLease" || kind === "shutdown" ? (!scope.leaseId || entry.id === scope.leaseId) && (!scope.runId || entry.runId === scope.runId)
-            : !scope.runId || entry.id === scope.runId),
-      load: async scope => {
+    async function load(scope?: Scope) {
         const key = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId,
           leaseId: scope.leaseId, agentId: scope.agentId }) : "*";
         if (!enabled) { cursors.delete(key); return []; }
@@ -89,9 +80,26 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof
         const complete = rows.length < 100;
         if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
         return { complete, rows };
-      },
+    }
+    return registerScopedDeadlineWork<Entry, Scope>(coordinator, {
+      queue,
+      reconstructionIntervalMs: 30_000,
+      beforePass: async () => { enabled = await owner.enabled(); },
+      decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (
+        kind === "agent" ? !scope.agentId || entry.id === scope.agentId
+          : kind === "orphanLease" || kind === "shutdown" ? (!scope.leaseId || entry.id === scope.leaseId) && (!scope.runId || entry.runId === scope.runId)
+            : !scope.runId || entry.id === scope.runId),
+      load,
       check: async entry => {
         if (!enabled) return null;
+        const currentScope = { companyId: entry.companyId,
+          ...(kind === "agent" ? { agentId: entry.id } : kind === "orphanLease" || kind === "shutdown"
+            ? { leaseId: entry.id } : { runId: entry.id }) };
+        const loaded = await load(currentScope);
+        const current = (Array.isArray(loaded) ? loaded : loaded.rows).find(row => row.id === entry.id);
+        if (!current) return null;
+        if (current.dueAt > Date.now()) return current;
         let next: number | null = null;
         const errors: unknown[] = [];
         const scope: CheckScope = { companyId: entry.companyId,
@@ -102,6 +110,11 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof
         };
         await owner[kind](scope);
         if (errors.length) throw new AggregateError(errors, `Recovery obligation ${entry.id} failed`);
+        if (kind === "native") {
+          const refreshed = await load(currentScope);
+          const remaining = (Array.isArray(refreshed) ? refreshed : refreshed.rows).find(row => row.id === entry.id);
+          if (remaining && remaining.dueAt > Date.now()) next = next === null ? remaining.dueAt : Math.min(next, remaining.dueAt);
+        }
         return next === null ? null : { ...entry, dueAt: next };
       },
     }, canRun);

server/src/services/heartbeat/recovery.ts

@@ -1,3 +1,4 @@
+import { registerLiveRecoveryWork } from "./live-recovery-work.js";
 import { nativeEnvironmentSettlementDisposition, nativeEnvironmentSettlementWork } from "../native-runtime/native-environment-settlement-work.js";
 import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js";
 import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js";
@@ -18,7 +19,7 @@ import {
   hasStopOnlyCleanup,
   settleStopOnlyCleanup,
 } from "../sandbox-stop-and-retain.js";
-import { discoverSettledNativeRunProcesses, recordNativeLocalProcessStop, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js";
+import { discoverSettledNativeRunProcesses, recordNativeLocalProcessStop, releaseSettledNativeRunProcess, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js";
 import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js";
 import {
   legacyControllerBootId,
@@ -2010,6 +2011,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
           eq(heartbeatRuns.runtimeMode, "native"),
           eq(nativeRunFinalizations.phase, "retryable_failure"),
           isNull(nativeRunFinalizations.resultId),
+          sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`,
         ),
       );
     const claimableNativeRunIds = new Set<string>();
@@ -2260,6 +2262,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       nativeControllerLeaseExpiresAt,
       nativeWorkspaceFinalizationPending,
     } of activeRuns) {
+      if (parseObject(run.contextSnapshot).nativeQuestionCancellation) continue;
       // Native recovery owns these dispositions. Its writes happened after this
       // inventory was read; do not reap or overwrite them using the old row.
       if (remoteRecoveryRunIds.has(run.id)) continue;
@@ -2329,10 +2332,16 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
           processPidAlive: nativeProcessPidAlive,
           processGroupAlive: nativeProcessGroupAlive,
         });
+        if (nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now)
+          scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime());
         continue;
       }
       if (resumedRunIds.has(run.id)) continue;
-      if (locallyTracked) continue;
+      if (locallyTracked) {
+        if (!activeRunExecutions.has(run.id) && !runningProcesses.has(run.id) && nativeControllerLeaseExpiresAt)
+          scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime());
+        continue;
+      }
       const legacyLease = await liveLegacyControllerExpiresAt(db, run);
       if (legacyLease) { scope.onDeferredDeadline?.(legacyLease.getTime()); continue; }
 
@@ -2674,7 +2683,28 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     return { reaped: reaped.length, runIds: reaped };
   }
 
+  function registerResidualRecoveryWork(coordinator: Parameters<typeof registerLiveRecoveryWork>[1],
+    canRun: () => boolean, enabled: () => Promise<boolean>) {
+    return registerLiveRecoveryWork(db, coordinator, {
+      native: scope => resumeRetryableNativeProcessSessions(scope),
+      question: recoverQuestionCancellationRequests,
+      active: scope => recoverActiveExecutions(scope),
+      settled: async scope => {
+        if (!scope.runId || activeRunExecutions.has(scope.runId) || runningProcesses.has(scope.runId)
+          || adapterExecutionControls.has(scope.runId)) return;
+        await releaseSettledNativeRunProcess(db, scope.runId, scope);
+      },
+      orphanLease: scope => sweepOrphanedActiveLeases({ ...scope, backoffMs: 5 * 60_000 }),
+      shutdown: scope => sweepLegacyShutdownWorkspaceSettlements(db, new Date(), scope),
+      agent: scope => createHeartbeatAgentStatus(db).reconcileStrandedAgentStatuses(() => new Set([
+        ...activeRunExecutions, ...runningProcesses.keys(), ...adapterExecutionControls.keys(),
+      ]), scope),
+      enabled,
+    }, canRun);
+  }
+
   return {
+    registerResidualRecoveryWork,
     registerRetainedNativeSessionCleanupWork: retainedNativeCleanup.register,
     registerNativeWorkspaceFinalizationWork: nativeWorkspaceFinalization.register,
     registerNativeEnvironmentSettlementWork: nativeEnvironmentSettlement.register,

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

@@ -1,3 +1,4 @@
+import { 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";
@@ -40,7 +41,9 @@ export async function renewLegacyControllerLease(
   run: Pick<Run, "id" | "companyId" | "controllerBootId">,
   stage?: "dispatching",
 ): Promise<boolean> {
-  const [renewed] = await db.update(heartbeatRuns).set({
+  const [renewed] = await db.transaction(async tx => {
+    await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id });
+    return tx.update(heartbeatRuns).set({
     controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`,
     ...(stage ? { executionStage: stage } : {}),
   }).where(and(
@@ -49,6 +52,7 @@ export async function renewLegacyControllerLease(
     eq(heartbeatRuns.controllerBootId, legacyControllerBootId),
     gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
   )).returning({ id: heartbeatRuns.id });
+  });
   return Boolean(renewed);
 }
 
@@ -105,19 +109,23 @@ export async function revokeExpiredLegacyControllerWithState(
   if (run.runtimeMode === "native" || !run.controllerBootId) {
     return { executionStage: run.executionStage, processPid: run.processPid, processGroupId: run.processGroupId };
   }
-  const [revoked] = await db.update(heartbeatRuns).set({
+  const controllerBootId = run.controllerBootId;
+  const [revoked] = await db.transaction(async tx => {
+    await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id });
+    return tx.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),
+    eq(heartbeatRuns.controllerBootId, controllerBootId),
     lte(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
   )).returning({
     executionStage: heartbeatRuns.executionStage,
     processPid: heartbeatRuns.processPid,
     processGroupId: heartbeatRuns.processGroupId,
   });
+  });
   return revoked ?? null;
 }
 

server/src/services/legacy-shutdown-workspace-settlement.ts

@@ -1,3 +1,4 @@
+import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js";
 import { and, eq, gt, inArray, ne, or, sql } from "drizzle-orm";
 import { environmentLeases, heartbeatRuns, issues, type Db } from "@paperclipai/db";
 import { adapterExecutionControls } from "./adapter-execution-control.js";
@@ -33,6 +34,7 @@ export function maintainLegacyShutdownWorkspaceSettlement(db: Db, run: Run) {
     pending = db.transaction(async tx => {
       await tx.execute(sql`SET LOCAL lock_timeout = '1s'`);
       await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
+      await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
       await tx.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata},
         '{legacyShutdownWorkspaceSettlement,deadlineAt}', ${JSON.stringify(new Date(Date.now() + SETTLEMENT_LEASE_MS).toISOString())}::jsonb)` })
         .where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
@@ -86,23 +88,29 @@ export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new
 
 /** Commit the fence while the run is still live, before exposing interruption. */
 export async function beginLegacyShutdownWorkspaceSettlement(db: Db, run: Run, deadlineAt: Date) {
-  await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) ||
+  await db.transaction(async tx => {
+    await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
+    await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) ||
     ${JSON.stringify({ legacyShutdownWorkspaceSettlement: { schema: SCHEMA, state: "pending", runId: run.id,
       controllerBootId: legacyControllerBootId, deadlineAt: deadlineAt.toISOString() } })}::jsonb`,
   }).where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
     eq(environmentLeases.status, "active"), ne(environmentLeases.provider, "local"),
     or(sql`${environmentLeases.metadata}->>'driver' = 'sandbox'`, sql`${environmentLeases.metadata}->>'sandboxProviderPlugin' = 'true'`)));
+  });
 }
 
 /** The owner joins settlement, or records exact retention before timeout cleanup. */
 export async function finishLegacyShutdownWorkspaceSettlement(db: Db, run: Run, state: "settled" | "expired" = "settled") {
-  await db.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata},
+  await db.transaction(async tx => {
+    await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
+    await tx.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata},
     '{legacyShutdownWorkspaceSettlement,state}', ${JSON.stringify(state)}::jsonb)` }).where(and(
     eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
     sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'schema' = ${SCHEMA}`,
     sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'controllerBootId' = ${legacyControllerBootId}`,
     sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`,
   ));
+  });
 }
 
 /** A restart or another cleanup worker must never infer saved files from timeout. */
@@ -140,6 +148,7 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease,
     if (!Number.isFinite(deadline) || deadline > now.getTime()) return false;
     // A killed container cannot certify export. Preserve source and hold in
     // this same transaction before any worker may perform provider cleanup.
+    await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id });
     if (current.status !== "active") {
       const { hasConfirmedSandboxStopAndRetain, readStopOnlyCleanup } = await import("./sandbox-stop-and-retain.js");
       if (!(current.status === "released" && hasConfirmedSandboxStopAndRetain(current))

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

@@ -23,3 +23,8 @@ export async function notifyLiveResourceRecovery(tx: object, scope?: LiveResourc
     DELIVERY_QUEUES.strandedAgentStatus, DELIVERY_QUEUES.settledNativeProcesses])
     await notifyDeliveryWork(tx, queue, hint);
 }
+
+export async function notifyStrandedAgentStatus(tx: object, scope: { companyId: string; agentId: string }) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAgentStatus,
+    JSON.stringify({ companyId: scope.companyId, agentId: scope.agentId }));
+}

server/src/services/native-finalization-notifications.ts

@@ -1,4 +1,4 @@
-import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
+import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
 import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
@@ -16,7 +16,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i
 
 export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) {
   if (scope.issueId || scope.runId) {
-    await notifyActiveRunSilence(transaction, scope);
+    await notifyLiveRunRecovery(transaction, scope);
     await notifyNativeEnvironmentSettlement(transaction, scope);
     await notifyExternalAgentSettlement(transaction, scope);
     await notifyExternalConversationSettlement(transaction, scope);

server/src/services/native-local-process-stop.ts

@@ -1,3 +1,4 @@
+import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
 import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
 import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js";
@@ -178,6 +179,7 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string, opti
         return "pending";
       }
     }
+    await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id, agentId: run.agentId });
     await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id });
     await tx.update(heartbeatRuns).set({
       processPid: null, processGroupId: null, processStartedAt: null, updatedAt: new Date(),

server/src/services/native-runtime/native-execution-lease.ts

@@ -1,3 +1,4 @@
+import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { livenessDb, nativeRunFinalizations } from "@paperclipai/db";
 import { and, eq, gt, sql } from "drizzle-orm";
@@ -53,6 +54,7 @@ async function renewNativeSessionExecutionLeaseDeadline(
     // but conservatively include UPDATE/commit response time without assuming
     // that this server's wall clock agrees with the database's clock.
     confirmedDeadline = performance.now() + leaseTtlMs;
+    await notifyLiveRunRecovery(tx, { companyId: input.companyId, runId: input.runId });
     return tx.update(nativeRunFinalizations)
       .set({
         leaseExpiresAt: sql`clock_timestamp() + (${leaseTtlMs} * interval '1 millisecond')`,

server/src/services/native-runtime/native-finalization-reconciler.ts

@@ -1,3 +1,4 @@
+import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { pendingNativeRunProjectionCondition } from "./native-run-projection.js";
 import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js";
@@ -244,6 +245,7 @@ export async function claimNativeSessionResumptions(input: {
       ) {
         const failureCode = originalFailureCode;
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, row.run.companyId);
+        await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId });
         await tx.update(nativeRunFinalizations).set({
           phase: "terminal_failure",
           leaseOwner: null,
@@ -312,6 +314,7 @@ export async function claimNativeSessionResumptions(input: {
       }
 
       const leaseExpiresAt = new Date(now.getTime() + 20 * 60_000);
+      await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId });
       const updated = await tx.update(nativeRunFinalizations).set({
         phase: "observed",
         leaseOwner,
@@ -362,6 +365,7 @@ export async function claimNativeSessionResumptions(input: {
     if (claimed) claims.push({ runId: candidate.runId, leaseOwner });
     else await input.db.transaction(async tx => {
       await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`);
+      await notifyLiveRunRecovery(tx, { companyId: candidate.companyId, runId: candidate.runId });
       await tx.update(nativeRunFinalizations).set({
         nextAttemptAt: new Date(now.getTime() + 30_000),
         updatedAt: now,

server/src/services/native-runtime/native-question-bridge.ts

@@ -1,3 +1,4 @@
+import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
 import { and, eq, inArray, sql } from "drizzle-orm";
 
 import type { Db } from "@paperclipai/db";
@@ -348,6 +349,7 @@ export async function requestNativeQuestionRunCancellation(
     ...cause,
     requestedAt: new Date().toISOString(),
   });
+  await notifyLiveRunRecovery(db, { companyId: interaction.companyId, runId: run.id });
   return db.update(heartbeatRuns).set({
     contextSnapshot: sql`jsonb_set(
       case

server/src/services/native-runtime/native-restart-recovery.ts

@@ -1,3 +1,4 @@
+import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
 import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
@@ -649,6 +650,7 @@ export async function claimNativeRestartRecoveries(input: {
           hasProviderEvidence: false,
           stderrTail: row.run.stderrExcerpt,
         });
+        await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId });
         await tx
           .update(nativeRunFinalizations)
           .set({
@@ -1020,6 +1022,7 @@ export async function claimNativeRestartRecoveries(input: {
         recycledProviderPids: providerProcesses.recycledPids,
         stderrTail: row.run.stderrExcerpt,
       });
+      await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId });
       const claimed = await tx
         .update(nativeRunFinalizations)
         .set({

server/src/services/native-runtime/native-runner-checkpoint.ts

@@ -1,3 +1,4 @@
+import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
 import { and, eq, exists, gt, isNull, lte, inArray, notExists, or, sql } from "drizzle-orm";
 import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db";
 import { inspectDurablePrpCheckpointHeader, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js";
@@ -75,6 +76,7 @@ export function nativeRunnerCheckpointWriter(db: Db, owner: Owner) {
           eq(heartbeatRuns.nativeSessionId, checkpoint.predecessor.normalizedSessionId)));
         if (!prior) throw new Error("native_runner_checkpoint_binding_mismatch");
       }
+      await notifyLiveRunRecovery(tx, { companyId: owner.companyId, runId: owner.runId });
       if (checkpoint.reset) await tx.delete(nativeRunnerCheckpoints).where(and(
         eq(nativeRunnerCheckpoints.companyId, owner.companyId), eq(nativeRunnerCheckpoints.runId, owner.runId)));
       else if (checkpoint.removedRecords.length) await tx.delete(nativeRunnerCheckpoints).where(and(