Checkpoint scoped live recovery owners and durable cohorts

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

server/src/modules/active-run-watchdog/adapters/deadline-work.ts

@@ -0,0 +1,62 @@
+import { and, asc, eq, gt, sql } from "drizzle-orm";
+import { heartbeatRuns, type Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "../../../services/delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.js";
+import type { LiveRunRecoveryScope } from "../../../services/live-run-recovery-notifications.js";
+import { registerScopedDeadlineWork } from "../../../services/scoped-deadline-work.js";
+import type { ScanSilentActiveRunsOptions } from "../application/use-cases.js";
+import type { WatchdogRunReader } from "../application/ports.js";
+
+type Entry = { id: string; companyId: string; issueId: string | null; dueAt: number };
+
+export function registerActiveRunSilenceWork(db: Db,
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  owner: { reader: WatchdogRunReader; scan(input: ScanSilentActiveRunsOptions): Promise<unknown>;
+    suspicionThresholdMs: number; enabled(): Promise<boolean> },
+  canRun: () => boolean,
+) {
+  const cursors = new Map<string, string>();
+  let enabled = false;
+  return registerScopedDeadlineWork<Entry, LiveRunRecoveryScope>(coordinator, {
+    queue: DELIVERY_QUEUES.activeRunSilence,
+    reconstructionIntervalMs: 30_000,
+    beforePass: async () => { enabled = await owner.enabled(); },
+    decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint),
+    matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId) && (!scope.issueId || entry.issueId === scope.issueId),
+    load: async scope => {
+      const key = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*";
+      if (!enabled) { cursors.delete(key); return []; }
+      const cursor = cursors.get(key);
+      const issueId = sql<string | null>`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`;
+      const rows = await db.select({ issueId, id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
+        lastActivity: sql<Date>`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.processStartedAt},
+          ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt})`.mapWith(heartbeatRuns.createdAt),
+      }).from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "running"),
+        scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
+        scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
+        scope?.issueId ? eq(issueId, scope.issueId) : undefined,
+        cursor ? gt(heartbeatRuns.id, cursor) : undefined,
+      )).orderBy(asc(heartbeatRuns.id)).limit(100);
+      const complete = rows.length < 100;
+      if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
+      return { complete, rows: rows.map(row => ({ id: row.id, companyId: row.companyId, issueId: row.issueId,
+        dueAt: row.lastActivity.getTime() + owner.suspicionThresholdMs })) };
+    },
+    check: async entry => {
+      if (!enabled) return null;
+      const run = await owner.reader.findRunForCompany(entry.companyId, entry.id);
+      if (!run || run.status !== "running") return null;
+      const started = run.lastOutputAt ?? run.processStartedAt ?? run.startedAt ?? run.createdAt;
+      if (!started) return null;
+      const due = started.getTime() + owner.suspicionThresholdMs;
+      if (due > Date.now()) return { ...entry, dueAt: due };
+      const decision = await owner.reader.findLatestDecision(entry.companyId, entry.id, new Date());
+      if (decision.quietUntilDecision) return { ...entry, dueAt: decision.quietUntilDecision.snoozedUntil.getTime() };
+      let next: number | null = null;
+      await owner.scan({ companyId: entry.companyId, runId: entry.id, onDeferredDeadline: at => {
+        next = next === null ? at : Math.min(next, at);
+      } });
+      return next === null ? null : { ...entry, dueAt: next };
+    },
+  }, canRun);
+}

server/src/modules/active-run-watchdog/adapters/postgres.ts

@@ -1,3 +1,4 @@
+import { notifyActiveRunSilence } from "../../../services/live-run-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
 import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
 import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
@@ -100,6 +101,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
 
   async function findCandidateSilentRuns(input: {
     companyId?: string;
+    runId?: string;
     suspicionBefore: Date;
     issueCreatedAtGte?: Date | null;
   }): Promise<RunSnapshot[]> {
@@ -109,6 +111,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
       .where(
         and(
           input.companyId ? eq(heartbeatRuns.companyId, input.companyId) : undefined,
+          input.runId ? eq(heartbeatRuns.id, input.runId) : undefined,
           eq(heartbeatRuns.status, "running"),
           sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.processStartedAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) <= ${input.suspicionBefore.toISOString()}::timestamptz`,
         ),
@@ -282,8 +285,9 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
   }
 
   async function recordDecision(companyId: string, input: RecordDecisionInput): Promise<WatchdogDecisionRecord> {
-    const [row] = await db
-      .insert(heartbeatRunWatchdogDecisions)
+    const [row] = await db.transaction(async tx => {
+      await notifyActiveRunSilence(tx, { companyId, runId: input.runId });
+      return tx.insert(heartbeatRunWatchdogDecisions)
       .values({
         companyId,
         runId: input.runId,
@@ -296,6 +300,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
         createdByRunId: input.createdByRunId,
       })
       .returning();
+    });
 
     await logActivity(db, {
       companyId,

server/src/modules/active-run-watchdog/application/ports.ts

@@ -13,6 +13,7 @@ import type {
 
 export type FindCandidateSilentRunsInput = {
   companyId?: string;
+  runId?: string;
   suspicionBefore: Date;
   issueCreatedAtGte?: Date | null;
 };

server/src/modules/active-run-watchdog/application/types.ts

@@ -114,7 +114,7 @@ export type FoldSourceResolvedRunInput = {
 
 export type FoldOutcome =
   | { kind: "folded"; evaluationIssueId: string | null }
-  | { kind: "stale" };
+  | { kind: "stale"; retryAt?: number };
 
 export type WatchdogDecisionActor =
   | { type: "board"; userId?: string | null; runId?: string | null }

server/src/modules/active-run-watchdog/application/use-cases.ts

@@ -100,7 +100,7 @@ export function createFoldSourceResolvedRun(deps: FoldSourceResolvedRunDeps) {
       fallbackProcessGroupId: input.run.processGroupId,
     });
 
-    return deps.writer.foldSourceResolvedRun(input.run.companyId, {
+    const outcome = await deps.writer.foldSourceResolvedRun(input.run.companyId, {
       run: input.run,
       sourceIssue: input.sourceIssue,
       evidence: input.evidence,
@@ -110,6 +110,8 @@ export function createFoldSourceResolvedRun(deps: FoldSourceResolvedRunDeps) {
       cleanup,
       now: input.now,
     });
+    return outcome.kind === "stale" && cleanup.attempted && cleanup.outcome === "termination_sent_still_running"
+      ? { ...outcome, retryAt: input.now.getTime() + 30_000 } : outcome;
   };
 }
 
@@ -227,12 +229,14 @@ export type ScanSilentActiveRunsDeps = {
 
 export type ScanSilentActiveRunsOptions = {
   now?: Date;
+  runId?: string;
+  onDeferredDeadline?: (at: number) => void;
   companyId?: string;
   issueCreatedAtGte?: Date | null;
 };
 
 type InspectOutcome =
-  | { kind: "skipped" }
+  | { kind: "skipped"; retryAt?: number }
   | { kind: "existing"; evaluationIssueId: string }
   | { kind: "folded"; evaluationIssueId: string | null };
 
@@ -283,7 +287,7 @@ export function createScanSilentActiveRuns(deps: ScanSilentActiveRunsDeps) {
         });
         return foldOutcome.kind === "folded"
           ? { kind: "folded", evaluationIssueId: foldOutcome.evaluationIssueId }
-          : { kind: "skipped" };
+          : { kind: "skipped", ...(foldOutcome.retryAt === undefined ? {} : { retryAt: foldOutcome.retryAt }) };
       }
     }
 
@@ -305,6 +309,7 @@ export function createScanSilentActiveRuns(deps: ScanSilentActiveRunsDeps) {
     const suspicionBefore = new Date(now.getTime() - deps.suspicionThresholdMs);
     const candidates = await deps.reader.findCandidateSilentRuns({
       companyId: opts?.companyId,
+      runId: opts?.runId,
       suspicionBefore,
       issueCreatedAtGte: opts?.issueCreatedAtGte,
     });
@@ -331,6 +336,7 @@ export function createScanSilentActiveRuns(deps: ScanSilentActiveRunsDeps) {
         now,
         dismissedFalsePositive: decisionState.dismissedFalsePositive,
       });
+      if (outcome.kind === "skipped" && outcome.retryAt !== undefined) opts?.onDeferredDeadline?.(outcome.retryAt);
       if (outcome.kind === "existing") result.existing += 1;
       else if (outcome.kind === "folded") result.folded += 1;
       else result.skipped += 1;

server/src/modules/active-run-watchdog/index.ts

@@ -1,3 +1,4 @@
+import { registerActiveRunSilenceWork } from "./adapters/deadline-work.js";
 import type { Db } from "@paperclipai/db";
 import { createPostgresWatchdogAdapter } from "./adapters/postgres.js";
 import { createProcessAdapter } from "./adapters/process.js";
@@ -29,17 +30,19 @@ export function createActiveRunWatchdog(db: Db, config: ActiveRunWatchdogConfig)
     processController,
   });
 
+  const scanSilentActiveRuns = createScanSilentActiveRuns({ reader: postgresAdapter, foldSourceResolvedRun,
+    suspicionThresholdMs: config.suspicionThresholdMs });
+
   return {
+    registerSilenceWork: (coordinator: Parameters<typeof registerActiveRunSilenceWork>[1], canRun: () => boolean,
+      scan = scanSilentActiveRuns, enabled = async () => true) => registerActiveRunSilenceWork(db, coordinator, { reader: postgresAdapter,
+        scan, enabled, suspicionThresholdMs: config.suspicionThresholdMs }, canRun),
     buildRunOutputSilence: createBuildRunOutputSilence({
       reader: postgresAdapter,
       suspicionThresholdMs: config.suspicionThresholdMs,
       criticalThresholdMs: config.criticalThresholdMs,
     }),
-    scanSilentActiveRuns: createScanSilentActiveRuns({
-      reader: postgresAdapter,
-      foldSourceResolvedRun,
-      suspicionThresholdMs: config.suspicionThresholdMs,
-    }),
+    scanSilentActiveRuns,
     recordWatchdogDecision: createRecordWatchdogDecision({
       reader: postgresAdapter,
       writer: postgresAdapter,

server/src/services/activity-log.ts

@@ -1,3 +1,4 @@
+import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { beginRunPhase } from "./execution-preparation-timing.js";
 import type { ActivityPublication } from "../types/activity-publication.js";
@@ -181,6 +182,8 @@ export async function persistActivity(db: Db, input: LogActivityInput) {
   const [activity] = input.entityType === "issue"
     ? await db.transaction(async tx => {
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, `task:${input.entityId}`);
+      if (input.runId && input.action === "issue.updated" && ["done", "cancelled"].includes(String(input.details?.status)))
+        await notifyActiveRunSilence(tx, { companyId: input.companyId, runId: input.runId });
       return insert(tx as unknown as Db);
     })
     : await insert(db);

server/src/services/delivery-work-notifications.ts

@@ -2,6 +2,14 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  nativeProcessSessions: "native-process-sessions",
+  nativeQuestionCancellations: "native-question-cancellations",
+  activeExecutionRecovery: "active-execution-recovery",
+  settledNativeProcesses: "settled-native-processes",
+  orphanActiveLeases: "orphan-active-leases",
+  legacyShutdownSettlement: "legacy-shutdown-settlement",
+  strandedAgentStatus: "stranded-agent-status",
+  activeRunSilence: "active-run-silence",
   deferredExecutionReceipts: "deferred-execution-receipts",
   taskRetryReceiptSettlement: "task-retry-receipt-settlement",
   nativeStatusWakeIntents: "native-status-wake-intents",

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

@@ -1,3 +1,4 @@
+import { notifyActiveRunSilence } 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";
@@ -9,6 +10,7 @@ export async function notifyExecutionRunWork(transaction: object, scope?: Execut
   if (scope) {
     await notifyRunDispatch(transaction, scope);
     await notifyDeferredExecutionEligibility(transaction, scope);
+    await notifyActiveRunSilence(transaction, scope);
   }
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
@@ -24,7 +26,10 @@ export function writeExecutionControl<T>(
 ): Promise<T> {
   return db.transaction(async transaction => {
     await notifyRunDispatch(transaction, scope);
-    if (scope) await notifyDeferredExecutionEligibility(transaction, scope);
+    if (scope) {
+      await notifyDeferredExecutionEligibility(transaction, scope);
+      await notifyActiveRunSilence(transaction, scope);
+    }
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
     return write(transaction as unknown as Db);

server/src/services/heartbeat.ts

@@ -4245,6 +4245,10 @@ export function heartbeatService(
     computeTaskDrain,
     applyTaskDrain,
 
+    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() }),
+        async () => !(await getSchedulingSuppression(true)).suppressed),
     registerDeferredReceiptWork: (coordinator: Parameters<typeof registerDeferredReceiptWork>[1], canRun: () => boolean) =>
       registerDeferredReceiptWork(db, coordinator, {
         resume: scope => heartbeatQueue.resumeDeferredExecutionReceipts(scope),

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

@@ -109,13 +109,15 @@ export function createHeartbeatAgentStatus(db: Db) {
     publishStatus(result.updated, outcome);
   }
 
-  async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet<string>) {
+  async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet<string>, scope: { companyId?: string; agentId?: string } = {}) {
     const now = new Date();
     const cutoff = new Date(now.getTime() - RECONCILIATION_GRACE_MS);
     const candidates = await db.transaction(async tx => {
       await tx.execute(sql`SET LOCAL statement_timeout = '2s'`);
       return tx.select({ id: agents.id, companyId: agents.companyId }).from(agents)
       .where(and(eq(agents.status, "running"), eq(agents.lifecycleState, "ready"),
+        scope.companyId ? eq(agents.companyId, scope.companyId) : undefined,
+        scope.agentId ? eq(agents.id, scope.agentId) : undefined,
         lte(agents.updatedAt, cutoff), settledAgentWorkCondition(liveRunIds(), now)))
       .orderBy(asc(agents.updatedAt), asc(agents.id)).limit(RECONCILIATION_BATCH_SIZE);
     });

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

@@ -0,0 +1,119 @@
+import { and, asc, eq, 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";
+import type { LiveResourceRecoveryScope, LiveRunRecoveryScope } from "../live-run-recovery-notifications.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+
+const TERMINAL = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"];
+type Scope = LiveRunRecoveryScope & LiveResourceRecoveryScope;
+type Entry = { id: string; companyId: string; runId?: string | null; dueAt: number };
+type CheckScope = { companyId: string; runId?: string; leaseId?: string; agentId?: string;
+  onDeferredDeadline(at: number): void; onError(error: unknown): void };
+type RunKind = "native" | "question" | "active" | "settled";
+
+export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, owner: {
+  native(scope: CheckScope): Promise<unknown>;
+  question(scope: CheckScope): Promise<unknown>;
+  active(scope: CheckScope): Promise<unknown>;
+  settled(scope: CheckScope): Promise<unknown>;
+  orphanLease(scope: CheckScope): Promise<unknown>;
+  shutdown(scope: CheckScope): Promise<unknown>;
+  agent(scope: CheckScope): Promise<unknown>;
+  enabled(): Promise<boolean>;
+}, canRun: () => boolean) {
+  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 => {
+        const key = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId,
+          leaseId: scope.leaseId, agentId: scope.agentId }) : "*";
+        if (!enabled) { cursors.delete(key); return []; }
+        const cursor = cursors.get(key);
+        let rows: Entry[];
+        if (kind === "agent") {
+          const candidates = await db.select({ id: agents.id, companyId: agents.companyId, updatedAt: agents.updatedAt })
+            .from(agents).where(and(eq(agents.status, "running"), eq(agents.lifecycleState, "ready"),
+              scope ? eq(agents.companyId, scope.companyId) : undefined,
+              scope?.agentId ? eq(agents.id, scope.agentId) : undefined,
+              cursor ? gt(agents.id, cursor) : undefined,
+            )).orderBy(asc(agents.id)).limit(100);
+          rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, dueAt: row.updatedAt.getTime() + 5 * 60_000 }));
+        } else if (kind === "orphanLease" || kind === "shutdown") {
+          const candidates = await db.select({ id: environmentLeases.id, companyId: environmentLeases.companyId,
+            runId: environmentLeases.heartbeatRunId, updatedAt: environmentLeases.updatedAt, metadata: environmentLeases.metadata }).from(environmentLeases)
+            .leftJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId),
+              eq(heartbeatRuns.companyId, environmentLeases.companyId)))
+            .where(and(scope ? eq(environmentLeases.companyId, scope.companyId) : undefined,
+              scope?.leaseId ? eq(environmentLeases.id, scope.leaseId) : undefined,
+              scope?.runId ? eq(environmentLeases.heartbeatRunId, scope.runId) : undefined,
+              cursor ? gt(environmentLeases.id, cursor) : undefined,
+              kind === "orphanLease" ? and(eq(environmentLeases.status, "active"),
+                or(isNull(environmentLeases.heartbeatRunId), inArray(heartbeatRuns.status, TERMINAL)))
+                : sql`${environmentLeases.metadata}#>>'{legacyShutdownWorkspaceSettlement,schema}' = 'paperclip.legacy-shutdown-workspace-settlement.v1'
+                  and ${environmentLeases.metadata}#>>'{legacyShutdownWorkspaceSettlement,state}' = 'pending'`,
+            )).orderBy(asc(environmentLeases.id)).limit(100);
+          rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, runId: row.runId,
+            dueAt: kind === "orphanLease" ? row.updatedAt.getTime() + 5 * 60_000
+              : Date.parse(String((row.metadata?.legacyShutdownWorkspaceSettlement as Record<string, unknown>)?.deadlineAt)) || Date.now(),
+          }));
+        } else {
+          const candidates = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
+            nextAttemptAt: nativeRunFinalizations.nextAttemptAt, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt })
+            .from(heartbeatRuns).leftJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id),
+              eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId)))
+            .where(and(scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
+              scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
+              cursor ? gt(heartbeatRuns.id, cursor) : undefined,
+              kind === "native" ? and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["running", "failed"]),
+                eq(nativeRunFinalizations.phase, "retryable_failure"), isNull(nativeRunFinalizations.resultId))
+                : kind === "question" ? and(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
+                  sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is not null`)
+                  : kind === "active" ? eq(heartbeatRuns.status, "running")
+                    : 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))),
+            )).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() }));
+        }
+        const complete = rows.length < 100;
+        if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
+        return { complete, rows };
+      },
+      check: async entry => {
+        if (!enabled) return null;
+        let next: number | null = null;
+        const errors: unknown[] = [];
+        const scope: CheckScope = { companyId: entry.companyId,
+          ...(kind === "agent" ? { agentId: entry.id } : kind === "orphanLease" || kind === "shutdown"
+            ? { leaseId: entry.id } : { runId: entry.id }),
+          onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); },
+          onError: error => { errors.push(error); },
+        };
+        await owner[kind](scope);
+        if (errors.length) throw new AggregateError(errors, `Recovery obligation ${entry.id} failed`);
+        return next === null ? null : { ...entry, dueAt: next };
+      },
+    }, canRun);
+  }
+  const workers = [register(DELIVERY_QUEUES.nativeProcessSessions, "native"),
+    register(DELIVERY_QUEUES.nativeQuestionCancellations, "question"),
+    register(DELIVERY_QUEUES.activeExecutionRecovery, "active"),
+    register(DELIVERY_QUEUES.settledNativeProcesses, "settled"),
+    register(DELIVERY_QUEUES.orphanActiveLeases, "orphanLease"),
+    register(DELIVERY_QUEUES.legacyShutdownSettlement, "shutdown"),
+    register(DELIVERY_QUEUES.strandedAgentStatus, "agent")];
+  return { ready: Promise.all(workers.map(worker => worker.ready)),
+    wake() { for (const worker of workers) worker.invalidate(); },
+    async stop() { await Promise.all(workers.map(worker => worker.stop())); } };
+}

server/src/services/heartbeat/recovery.ts

@@ -22,7 +22,7 @@ import { discoverSettledNativeRunProcesses, recordNativeLocalProcessStop, retryP
 import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js";
 import {
   legacyControllerBootId,
-  hasLiveLegacyController,
+  liveLegacyControllerExpiresAt,
   revokeExpiredLegacyControllerWithState,
 } from "../legacy-controller-lease.js";
 import { remoteTerminationReceipt } from "../remote-execution-termination.js";
@@ -1493,6 +1493,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
   // data already on the lease row.
   async function sweepOrphanedActiveLeases(opts: {
     backoffMs: number;
+    companyId?: string;
+    leaseId?: string;
   }): Promise<{ recovered: number }> {
     const cutoff = new Date(Date.now() - opts.backoffMs);
 
@@ -1506,6 +1508,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       .where(
         and(
           eq(environmentLeases.status, "active"),
+          opts.companyId ? eq(environmentLeases.companyId, opts.companyId) : undefined,
+          opts.leaseId ? eq(environmentLeases.id, opts.leaseId) : undefined,
           or(
             isNull(environmentLeases.heartbeatRunId),
             inArray(heartbeatRuns.status, [
@@ -1977,59 +1981,15 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     void cleanup.finally(() => activeRunExecutionPromises.delete(cleanup));
   }
 
-  let settledProcessesDiscovered = false;
+  type RecoveryScope = {
+    companyId?: string;
+    runId?: string;
+    onDeferredDeadline?: (at: number) => void;
+    onError?: (error: unknown) => void;
+  };
 
-  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipNativeStatusWakeIntents?: boolean; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean; skipNativeEnvironmentSettlement?: boolean }) {
-    if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db);
-    const staleThresholdMs = opts?.staleThresholdMs ?? 0;
+  async function resumeRetryableNativeProcessSessions(scope: RecoveryScope = {}, beforeDispatch?: () => Promise<void>) {
     const now = new Date();
-    if (!opts?.skipScheduledRetryCancellation) await scheduledRetryCancellation.sweep();
-    // Recovery never launches a provider or infers stopped ownership from
-    // terminal status. Uncaptured local copies require durable stop evidence.
-    if (!opts?.skipStoppedInstructionCollection) await instructionCopies.recoverStopped().catch(error => {
-      logger.warn({ err: error }, "failed to recover stopped instruction copies");
-    });
-    await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit, skipDirectoryCleanup: opts?.skipInstructionDirectoryCleanup }).catch(error => {
-      logger.warn({ err: error }, "failed to retry captured instruction revisions");
-    });
-
-    // Complete persisted native results before generic orphan recovery. The
-    // reconciler reads the durable workspace barrier and persisted runtime
-    // mode, never the current feature flag.
-    await reconcileNativeFinalizations(db, undefined, {
-      skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews,
-      skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews,
-      skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports,
-      skipNativePendingEffects: opts?.skipNativePendingEffects,
-      skipNativeResponsePresentation: opts?.skipNativeResponsePresentation,
-      skipNativeReviewResponse: opts?.skipNativeReviewResponse,
-      skipNativeStatusReassessment: opts?.skipNativeStatusReassessment,
-      skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization,
-      skipNativeResultCommit: opts?.skipNativeResultCommit,
-      skipNativeRunProjection: opts?.skipNativeRunProjection,
-      skipExternalConversationSettlement: opts?.skipExternalConversationSettlement,
-      skipExternalAgentSettlement: opts?.skipExternalAgentSettlement,
-      skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices,
-      environmentRuntime,
-      pluginWorkerManager: dependencies.pluginWorkerManager,
-      onWorkspaceSettled: input => settleRecoveredNativeWorkspace(input, {
-        skipEnvironmentSettlement: opts?.skipNativeEnvironmentSettlement,
-      }),
-    }).catch((error) => {
-      logger.warn(
-        { err: error },
-        "failed to reconcile persisted native finalizations before orphan reaping",
-      );
-    });
-    if (!opts?.skipRetainedNativeSessionCleanup) scheduleRetainedNativeSessionCleanup();
-    await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments);
-    if (!opts?.skipNativeStatusWakeIntents) await dispatchPendingNativeStatusWakeups().catch((error) => {
-      logger.warn(
-        { err: error },
-        "failed to dispatch persisted native status wake intents before orphan reaping",
-      );
-    });
-
     // A retryable native run can retain process identifiers from the failed
     // attempt. Inspect them before the recovery claim: a live identifier is
     // unowned and blocks recovery, while identifiers that are all dead can be
@@ -2044,6 +2004,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       )
       .where(
         and(
+          scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
+          scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
           inArray(heartbeatRuns.status, ["running", "failed"]),
           eq(heartbeatRuns.runtimeMode, "native"),
           eq(nativeRunFinalizations.phase, "retryable_failure"),
@@ -2067,6 +2029,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
           processPidAlive,
           processGroupAlive,
         });
+        scope.onDeferredDeadline?.(Date.now() + 30_000);
         continue;
       }
       const cleared = await db.transaction(async tx => {
@@ -2101,17 +2064,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       if (cleared) claimableNativeRunIds.add(cleared.id);
     }
 
-    const executingRun = (runId: string) => activeRunExecutions.has(runId) || runningProcesses.has(runId);
-    if (!settledProcessesDiscovered) {
-      settledProcessesDiscovered = await discoverSettledNativeRunProcesses(db, { skip: executingRun }).then(() => true, (error) => {
-        logger.warn({ err: error }, "failed to discover settled native process identities");
-        return false;
-      });
-    } else {
-      await retryPendingSettledNativeRunProcesses(db, { skip: executingRun }).catch((error) => {
-        logger.warn({ err: error }, "failed to retry settled native process identity release");
-      });
-    }
+    await beforeDispatch?.();
 
     // An explicit result-less retryable failure resumes on the original run.
     // The database lease is claimed before dispatch so concurrent service
@@ -2141,6 +2094,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
               );
             },
           }).catch((error) => {
+            scope.onError?.(error);
             logger.warn(
               { err: error },
               "failed to claim persisted native session resumptions",
@@ -2151,6 +2105,10 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       nativeResumeClaims.map((claim) => claim.runId),
     );
 
+    return resumedRunIds;
+  }
+
+  async function recoverQuestionCancellationRequests(scope: RecoveryScope = {}) {
     // A terminal issue transition writes this intent in the same transaction
     // that closes the question's task, even when its card is retained. Consume
     // it before generic orphan recovery so a restart preserves cancellation.
@@ -2162,6 +2120,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       .from(heartbeatRuns)
       .where(
         and(
+          scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
+          scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
           inArray(heartbeatRuns.status, [
             ...CANCELLABLE_HEARTBEAT_RUN_STATUSES,
           ]),
@@ -2200,6 +2160,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
           },
         });
       } catch (err) {
+        scope.onError?.(err);
         // Keep the marker intact for the next startup/periodic sweep.
         logger.warn(
           { err, runId: request.id },
@@ -2208,6 +2169,14 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       }
     }
 
+  }
+
+  async function recoverActiveExecutions(scope: RecoveryScope = {}, options: {
+    staleThresholdMs?: number; resumedRunIds?: Set<string>;
+  } = {}) {
+    const now = new Date();
+    const staleThresholdMs = options.staleThresholdMs ?? 5 * 60_000;
+    const resumedRunIds = options.resumedRunIds ?? new Set<string>();
     // Find all runs stuck in "running" state (queued runs are legitimately waiting; resumeQueuedRuns handles them)
     const activeRuns = await db
       .select({
@@ -2229,7 +2198,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
         nativeRunFinalizations,
         eq(nativeRunFinalizations.runId, heartbeatRuns.id),
       )
-      .where(eq(heartbeatRuns.status, "running"));
+      .where(and(eq(heartbeatRuns.status, "running"),
+        scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
+        scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined));
 
     // Reuse the orphan sweep's active-run inventory. Only an expired, untracked
     // controller creates recovery work; idle sweeps add no checkpoint query.
@@ -2362,12 +2333,16 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       }
       if (resumedRunIds.has(run.id)) continue;
       if (locallyTracked) continue;
-      if (await hasLiveLegacyController(db, run)) continue;
+      const legacyLease = await liveLegacyControllerExpiresAt(db, run);
+      if (legacyLease) { scope.onDeferredDeadline?.(legacyLease.getTime()); continue; }
 
       // Apply staleness threshold to avoid false positives
       if (staleThresholdMs > 0) {
         const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0;
-        if (now.getTime() - refTime < staleThresholdMs) continue;
+        if (now.getTime() - refTime < staleThresholdMs) {
+          scope.onDeferredDeadline?.(refTime + staleThresholdMs);
+          continue;
+        }
       }
 
       const currentAdapterTracksLocalChild =
@@ -2392,6 +2367,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
         (processPidAlive || processGroupAlive) &&
         readHotRestartAdoptionMetadata(parseObject(run.resultJson))
       ) {
+        scope.onDeferredDeadline?.(Date.now() + 30_000);
         continue;
       }
       if (processPidAlive || processGroupAlive) {
@@ -2417,6 +2393,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
             });
           }
         }
+        scope.onDeferredDeadline?.(Date.now() + 30_000);
         continue;
       }
 
@@ -2566,6 +2543,80 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       );
     }
 
+    return reaped;
+  }
+
+  let settledProcessesDiscovered = false;
+
+  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipNativeStatusWakeIntents?: boolean; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean; skipNativeEnvironmentSettlement?: boolean }) {
+    if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db);
+    const staleThresholdMs = opts?.staleThresholdMs ?? 0;
+    const now = new Date();
+    if (!opts?.skipScheduledRetryCancellation) await scheduledRetryCancellation.sweep();
+    // Recovery never launches a provider or infers stopped ownership from
+    // terminal status. Uncaptured local copies require durable stop evidence.
+    if (!opts?.skipStoppedInstructionCollection) await instructionCopies.recoverStopped().catch(error => {
+      logger.warn({ err: error }, "failed to recover stopped instruction copies");
+    });
+    await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit, skipDirectoryCleanup: opts?.skipInstructionDirectoryCleanup }).catch(error => {
+      logger.warn({ err: error }, "failed to retry captured instruction revisions");
+    });
+
+    // Complete persisted native results before generic orphan recovery. The
+    // reconciler reads the durable workspace barrier and persisted runtime
+    // mode, never the current feature flag.
+    await reconcileNativeFinalizations(db, undefined, {
+      skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews,
+      skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews,
+      skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports,
+      skipNativePendingEffects: opts?.skipNativePendingEffects,
+      skipNativeResponsePresentation: opts?.skipNativeResponsePresentation,
+      skipNativeReviewResponse: opts?.skipNativeReviewResponse,
+      skipNativeStatusReassessment: opts?.skipNativeStatusReassessment,
+      skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization,
+      skipNativeResultCommit: opts?.skipNativeResultCommit,
+      skipNativeRunProjection: opts?.skipNativeRunProjection,
+      skipExternalConversationSettlement: opts?.skipExternalConversationSettlement,
+      skipExternalAgentSettlement: opts?.skipExternalAgentSettlement,
+      skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices,
+      environmentRuntime,
+      pluginWorkerManager: dependencies.pluginWorkerManager,
+      onWorkspaceSettled: input => settleRecoveredNativeWorkspace(input, {
+        skipEnvironmentSettlement: opts?.skipNativeEnvironmentSettlement,
+      }),
+    }).catch((error) => {
+      logger.warn(
+        { err: error },
+        "failed to reconcile persisted native finalizations before orphan reaping",
+      );
+    });
+    if (!opts?.skipRetainedNativeSessionCleanup) scheduleRetainedNativeSessionCleanup();
+    await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments);
+    if (!opts?.skipNativeStatusWakeIntents) await dispatchPendingNativeStatusWakeups().catch((error) => {
+      logger.warn(
+        { err: error },
+        "failed to dispatch persisted native status wake intents before orphan reaping",
+      );
+    });
+
+    const resumedRunIds = await resumeRetryableNativeProcessSessions({}, async () => {
+    const executingRun = (runId: string) => activeRunExecutions.has(runId) || runningProcesses.has(runId);
+    if (!settledProcessesDiscovered) {
+      settledProcessesDiscovered = await discoverSettledNativeRunProcesses(db, { skip: executingRun }).then(() => true, (error) => {
+        logger.warn({ err: error }, "failed to discover settled native process identities");
+        return false;
+      });
+    } else {
+      await retryPendingSettledNativeRunProcesses(db, { skip: executingRun }).catch((error) => {
+        logger.warn({ err: error }, "failed to retry settled native process identity release");
+      });
+    }
+
+    });
+
+    await recoverQuestionCancellationRequests();
+    const reaped = await recoverActiveExecutions({}, { staleThresholdMs, resumedRunIds });
+
     // Recover an active lease whose run already ended before the same-tick
     // pending_cleanup sweep, so this tick can stop the recovered sandbox.
     // Isolate the sweep so its failure never hides the reaper result.
@@ -2634,6 +2685,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     reconcileHotRestartAdoption,
     recoverNativeRunsAfterRestart,
     reapOrphanedRuns,
+    resumeRetryableNativeProcessSessions,
+    recoverQuestionCancellationRequests,
+    recoverActiveExecutions,
     sweepOrphanedActiveLeases,
     drainRunningRunsForShutdown,
   };

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

@@ -1,3 +1,4 @@
+import { notifyActiveRunSilence } from "../live-run-recovery-notifications.js";
 import { eq } from "drizzle-orm";
 import { heartbeatRuns, type Db } from "@paperclipai/db";
 import type { AdapterInvocationMeta, AdapterRuntimeEvent } from "../../adapters/index.js";
@@ -94,8 +95,9 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
       pendingOutputProgress.at.getTime() - lastOutputFlushAt.getTime() >=
         ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS;
     if (!shouldFlush) return;
-    progressWrite = db
-      .update(heartbeatRuns)
+    progressWrite = db.transaction(async tx => {
+      await notifyActiveRunSilence(tx, { companyId: run.companyId, runId: run.id });
+      await tx.update(heartbeatRuns)
       .set({
         lastOutputAt: pendingOutputProgress.at,
         lastOutputSeq: pendingOutputProgress.seq,
@@ -103,7 +105,8 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput)
         lastOutputBytes: pendingOutputProgress.bytes,
         updatedAt: new Date(),
       })
-      .where(eq(heartbeatRuns.id, run.id))
+      .where(eq(heartbeatRuns.id, run.id));
+    })
       .then(() => {
         lastOutputFlushAt = pendingOutputProgress.at;
         if (outputProgressState.pending === pendingOutputProgress)

server/src/services/instance-settings.ts

@@ -1,3 +1,4 @@
+import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js";
@@ -634,6 +635,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
         if (patch.enableWorktreeRunExecution !== undefined) {
           await notifyRunDispatch(tx);
           await notifyDeferredExecutionEligibility(tx);
+          await notifyActiveRunSilence(tx);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           await notifyExternalAgentSettlement(tx);
           await notifyInstructionCommit(tx);

server/src/services/issues.ts

@@ -1,3 +1,4 @@
+import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
@@ -11507,6 +11508,7 @@ export function issueService(db: Db) {
           || blockedByIssueIds !== undefined) {
           await notifyRunDispatch(tx, { companyId: receiptExisting.companyId });
           await notifyDeferredExecutionEligibility(tx, { companyId: receiptExisting.companyId });
+          await notifyActiveRunSilence(tx, { companyId: receiptExisting.companyId, issueId: id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         }
         if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt &&
@@ -11972,7 +11974,10 @@ export function issueService(db: Db) {
           await notifyDecisionTargetIssueRemoval(tx, id);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
           await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined);
-          if (owner) await notifyDeferredExecutionEligibility(tx, owner);
+          if (owner) {
+            await notifyDeferredExecutionEligibility(tx, owner);
+            await notifyActiveRunSilence(tx, { companyId: owner.companyId, issueId: id });
+          }
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           removedIssue = await tx
             .delete(issues)

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

@@ -73,8 +73,10 @@ export async function legacyShutdownWorkspaceResourceProtected(db: Db,
 }
 
 /** Released stop receipts still need reconciliation if export never settled. */
-export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date()) {
+export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date(), scope: { companyId?: string; leaseId?: string } = {}) {
   const rows = await db.select().from(environmentLeases).where(and(
+    scope.companyId ? eq(environmentLeases.companyId, scope.companyId) : undefined,
+    scope.leaseId ? eq(environmentLeases.id, scope.leaseId) : undefined,
     sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'schema' = ${SCHEMA}`,
     sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`,
     sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt' <= ${now.toISOString()}`,

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

@@ -0,0 +1,25 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type LiveRunRecoveryScope = { companyId: string; runId?: string; issueId?: string; agentId?: string };
+
+export async function notifyActiveRunSilence(tx: object, scope?: LiveRunRecoveryScope) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.activeRunSilence, scope
+    ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*");
+}
+
+export async function notifyLiveRunRecovery(tx: object, scope?: LiveRunRecoveryScope) {
+  const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, agentId: scope.agentId }) : "*";
+  for (const queue of [DELIVERY_QUEUES.nativeProcessSessions, DELIVERY_QUEUES.nativeQuestionCancellations,
+    DELIVERY_QUEUES.activeExecutionRecovery, DELIVERY_QUEUES.settledNativeProcesses, DELIVERY_QUEUES.strandedAgentStatus])
+    await notifyDeliveryWork(tx, queue, hint);
+  await notifyActiveRunSilence(tx, scope);
+}
+
+export type LiveResourceRecoveryScope = { companyId: string; runId?: string; leaseId?: string };
+
+export async function notifyLiveResourceRecovery(tx: object, scope?: LiveResourceRecoveryScope) {
+  const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, leaseId: scope.leaseId }) : "*";
+  for (const queue of [DELIVERY_QUEUES.orphanActiveLeases, DELIVERY_QUEUES.legacyShutdownSettlement,
+    DELIVERY_QUEUES.strandedAgentStatus, DELIVERY_QUEUES.settledNativeProcesses])
+    await notifyDeliveryWork(tx, queue, hint);
+}

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

@@ -1,3 +1,4 @@
+import { notifyActiveRunSilence } 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";
@@ -15,6 +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 notifyNativeEnvironmentSettlement(transaction, scope);
     await notifyExternalAgentSettlement(transaction, scope);
     await notifyExternalConversationSettlement(transaction, scope);

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

@@ -145,7 +145,7 @@ export async function readNativeLocalProcessStop(db: Db, companyId: string, runI
 
 type SettledProcessRelease = "released" | "pending" | "skipped";
 
-export async function releaseSettledNativeRunProcess(db: Db, runId: string): Promise<SettledProcessRelease> {
+export async function releaseSettledNativeRunProcess(db: Db, runId: string, options: { onDeferredDeadline?: (at: number) => void } = {}): Promise<SettledProcessRelease> {
   return db.transaction(async transaction => {
     const tx = transaction as unknown as Db;
     const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
@@ -173,7 +173,10 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string): Pro
       )).orderBy(desc(heartbeatRunEvents.seq)).limit(1);
       if (identity?.payload?.targetKind !== "local" || identity.payload.hostname !== os.hostname() ||
           identity.payload.processPid !== run.processPid || identity.payload.processGroupId !== run.processGroupId) return "skipped";
-      if (!(await recordNativeLocalProcessStop(tx, run))) return "pending";
+      if (!(await recordNativeLocalProcessStop(tx, run))) {
+        options.onDeferredDeadline?.(Date.now() + 30_000);
+        return "pending";
+      }
     }
     await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id });
     await tx.update(heartbeatRuns).set({

server/src/services/recovery/service.ts

@@ -2384,11 +2384,7 @@ export function recoveryService(
     });
   }
 
-  async function scanSilentActiveRuns(opts?: {
-    now?: Date;
-    companyId?: string;
-    issueCreatedAtGte?: Date | null;
-  }) {
+  async function scanSilentActiveRuns(opts?: Parameters<typeof watchdog.scanSilentActiveRuns>[0]) {
     return watchdog.scanSilentActiveRuns(opts);
   }
 
@@ -6568,6 +6564,7 @@ export function recoveryService(
     escalateStrandedAssignedIssue,
     recordWatchdogDecision,
     scanSilentActiveRuns,
+    registerActiveRunSilenceWork: watchdog.registerSilenceWork,
     reconcileStrandedAssignedIssues,
     reconcileUnassignedBlockingIssues,
     reconcileActiveRecoveryActions,