refactor(recovery): expose scoped issue recovery obligations

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

server/src/modules/wake-queue/adapters/postgres.ts

@@ -638,7 +638,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
         .returning()
         .then((rows) => rows[0]);
 
-      await notifyRunDispatch(tx);
+      await notifyRunDispatch(tx, { companyId });
       const queuedRun = await tx
         .insert(heartbeatRuns)
         .values({
@@ -717,7 +717,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
         .returning()
         .then((rows) => rows[0]);
 
-      await notifyRunDispatch(tx);
+      await notifyRunDispatch(tx, { companyId });
       const queuedRun = await tx
         .insert(heartbeatRuns)
         .values({

server/src/services/agent-configuration.ts

@@ -84,7 +84,7 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura
       existing.adapterConfig,
     );
   }
-  await notifyRunDispatch(db);
+  await notifyRunDispatch(db, { companyId: existing.companyId });
   if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) {
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
   }

server/src/services/agent-lifecycle.ts

@@ -40,7 +40,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 notifyRunDispatch(tx);
+        await notifyRunDispatch(tx, { companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
       }
       if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) {

server/src/services/budgets.ts

@@ -288,7 +288,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     }
 
     if (policy.scopeType === "project") {
-      await notifyRunDispatch(db);
+      await notifyRunDispatch(db, { companyId: policy.companyId });
       await db
         .update(projects)
         .set({
@@ -302,7 +302,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
 
     const scope = await resolveScopeRecord(db, "company", policy.scopeId);
     if (scope.paused) return;
-    await notifyRunDispatch(db);
+    await notifyRunDispatch(db, { companyId: policy.companyId });
     await notifyCompanyRetentionWork(db, policy.companyId);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
@@ -349,7 +349,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     }
 
     if (policy.scopeType === "project") {
-      await notifyRunDispatch(db);
+      await notifyRunDispatch(db, { companyId: policy.companyId });
       await db
         .update(projects)
         .set({
@@ -364,7 +364,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync);
     const scope = await resolveScopeRecord(db, "company", policy.scopeId);
     if (scope.pauseReason !== "budget") return;
-    await notifyRunDispatch(db);
+    await notifyRunDispatch(db, { companyId: policy.companyId });
     await notifyCompanyRetentionWork(db, policy.companyId);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);

server/src/services/chat-run-milestone-notifications.ts

@@ -12,7 +12,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type ChatRunMilestoneScope = { companyId: string; issueId?: string; runId?: string };
 
 export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) {
-  if (!scope.issueId && !scope.runId) await notifyRunDispatch(tx);
+  if (!scope.issueId && !scope.runId) await notifyRunDispatch(tx, scope);
   if (scope.issueId || scope.runId) {
     await notifyIssueReviewState(tx, { companyId: scope.companyId, issueId: scope.issueId,
       runId: scope.issueId ? undefined : scope.runId });

server/src/services/companies.ts

@@ -333,7 +333,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
         const renamedPrefix = await resolveRenamedIssuePrefix(tx, id, companyPatch);
 
         if (companyPatch.status !== undefined && companyPatch.status !== existing.status) {
-          await notifyRunDispatch(tx);
+          await notifyRunDispatch(tx, { companyId: id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
@@ -490,7 +490,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
         const wasAlreadyArchived = existing.status === "archived";
 
         if (!wasAlreadyArchived) {
-          await notifyRunDispatch(tx);
+          await notifyRunDispatch(tx, { companyId: id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);

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

@@ -2,6 +2,10 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  strandedAssignments: "stranded-assignments",
+  resolvedDependencies: "resolved-dependencies",
+  taskWatchdogs: "task-watchdogs",
+  staleIssueLocks: "stale-issue-locks",
   scheduledRetryDispatch: "scheduled-retry-dispatch",
   queuedRunDispatch: "queued-run-dispatch",
   runEvidenceMigration: "run-evidence-migration",

server/src/services/heartbeat.ts

@@ -1,3 +1,4 @@
+import { registerIssueRecoveryWork } from "./recovery/work.js";
 import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js";
 import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js";
 import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js";
@@ -1611,15 +1612,15 @@ export function heartbeatService(
     void reportRunFailure(db, updated, failureReport);
   }
 
-  async function reconcileStrandedAssignedIssues(opts?: { skipExternalConversationSettlement?: boolean }) {
+  async function reconcileStrandedAssignedIssues(opts?: Parameters<typeof recovery.reconcileStrandedAssignedIssues>[0]) {
     return recovery.reconcileStrandedAssignedIssues({
-      skipExternalConversationSettlement: opts?.skipExternalConversationSettlement,
+      ...opts,
       issueCreatedAtGte: await getWorktreeExecutionCutoff(),
     });
   }
 
-  async function sweepStaleIssueLocks() {
-    return recovery.sweepStaleIssueLocks();
+  async function sweepStaleIssueLocks(opts?: Parameters<typeof recovery.sweepStaleIssueLocks>[0]) {
+    return recovery.sweepStaleIssueLocks(opts);
   }
 
   function issueIdFromRunContext(contextSnapshot: unknown) {
@@ -1649,10 +1650,7 @@ export function heartbeatService(
     });
   }
 
-  async function reconcileTaskWatchdogs(opts?: {
-    companyId?: string | null;
-    runId?: string | null;
-  }) {
+  async function reconcileTaskWatchdogs(opts?: Parameters<typeof taskWatchdogs.reconcileTaskWatchdogs>[0]) {
     return taskWatchdogs.reconcileTaskWatchdogs({
       ...opts,
       issueCreatedAtGte: await getWorktreeExecutionCutoff(),
@@ -1677,10 +1675,7 @@ export function heartbeatService(
     return recovery.buildRunOutputSilence(run, now);
   }
 
-  async function reconcileResolvedDependencyWakes(opts?: {
-    runId?: string | null;
-    companyId?: string | null;
-  }) {
+  async function reconcileResolvedDependencyWakes(opts?: Parameters<typeof recovery.reconcileResolvedDependencyWakeBackstop>[0]) {
     return recovery.reconcileResolvedDependencyWakeBackstop(opts);
   }
 
@@ -4248,6 +4243,14 @@ export function heartbeatService(
     computeTaskDrain,
     applyTaskDrain,
 
+    registerIssueStateRecoveryWork: (coordinator: Parameters<typeof registerIssueRecoveryWork>[1], canRun: () => boolean) =>
+      registerIssueRecoveryWork(db, coordinator, {
+        stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true }),
+        dependencies: reconcileResolvedDependencyWakes,
+        locks: sweepStaleIssueLocks,
+        watchdogs: reconcileTaskWatchdogs,
+        schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed,
+      }, () => canRun() && readTaskDrain(new Date()) === null),
     registerRunDispatchWork: (coordinator: Parameters<typeof registerRunDispatchWork>[0], canRun: () => boolean) =>
       registerRunDispatchWork(coordinator, {
         promoteDueScheduledRetries,

server/src/services/heartbeat/queue.ts

@@ -4633,7 +4633,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
     }
 
     const queueOutcome = await db.transaction(async (tx) => {
-      await notifyRunDispatch(tx);
+      await notifyRunDispatch(tx, { companyId: agent.companyId });
       await tx.execute(
         sql`select id from agents where id = ${agentId} and company_id = ${agent.companyId} for update`,
       );

server/src/services/heartbeat/retries.ts

@@ -998,7 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
           retryContextSnapshot.previousRunId = continuation.previousRunId;
         }
 
-        await notifyRunDispatch(tx);
+        await notifyRunDispatch(tx, { companyId: run.companyId });
         const wakeupRequest = await tx
           .insert(agentWakeupRequests)
           .values({
@@ -1756,7 +1756,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
     };
 
     const updated = await db.transaction(async (tx) => {
-      await notifyRunDispatch(tx);
+      await notifyRunDispatch(tx, { companyId: scheduled.run.companyId });
       const row = await tx
         .update(heartbeatRuns)
         .set({

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

@@ -2448,7 +2448,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
         .returning()
         .then((rows) => rows[0]);
 
-      await notifyRunDispatch(tx);
+      await notifyRunDispatch(tx, { companyId: run.companyId });
       const queuedRun = await tx
         .insert(heartbeatRuns)
         .values({

server/src/services/issue-comment-writes.ts

@@ -1,3 +1,4 @@
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import type { PgInsertValue } from "drizzle-orm/pg-core";
 import { issueComments, type Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -6,6 +7,7 @@ type WriteDb = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
 
 export function insertIssueComment(db: WriteDb, input: PgInsertValue<typeof issueComments> & { companyId: string }) {
   return db.transaction(async tx => {
+    await notifyIssueRecovery(tx, { companyId: input.companyId });
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`);
     return tx.insert(issueComments).values(input).returning();
   });

server/src/services/issue-recovery-notifications.ts

@@ -0,0 +1,11 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type IssueRecoveryScope = { companyId: string };
+
+export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) {
+  const hint = scope ? JSON.stringify(scope) : "*";
+  for (const queue of [DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies,
+    DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) {
+    await notifyDeliveryWork(transaction, queue, hint);
+  }
+}

server/src/services/issue-review-state-notifications.ts

@@ -5,7 +5,7 @@ import { notifyNativeReviewResponse } from "./native-review-response-notificatio
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 
 export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) {
-  await notifyRunDispatch(tx);
+  await notifyRunDispatch(tx, scope);
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
   await notifyExternalConversationSettlement(tx, scope);
   await notifyNativeReviewResponse(tx, scope);

server/src/services/issues.ts

@@ -10606,7 +10606,7 @@ export function issueService(db: Db) {
             : values.parentId ? `issue:${values.parentId}` : undefined);
         }
         if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
-        await notifyRunDispatch(tx);
+        await notifyRunDispatch(tx, { companyId: values.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const [issue] = await tx.insert(issues).values(values).returning();
         await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId });
@@ -11494,8 +11494,10 @@ export function issueService(db: Db) {
           await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         }
         if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined
-          || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) {
-          await notifyRunDispatch(tx);
+          || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined || patch.parentId !== undefined
+          || patch.executionState !== undefined || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined
+          || blockedByIssueIds !== undefined) {
+          await notifyRunDispatch(tx, { companyId: receiptExisting.companyId });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         }
         if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt &&
@@ -11960,7 +11962,7 @@ export function issueService(db: Db) {
           }
           await notifyDecisionTargetIssueRemoval(tx, id);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
-          await notifyRunDispatch(tx);
+          await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
           removedIssue = await tx
             .delete(issues)
@@ -12089,7 +12091,7 @@ export function issueService(db: Db) {
         : isNull(issues.executionRunId);
       const updated = await db.transaction(async tx => {
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
-        await notifyRunDispatch(tx);
+        await notifyRunDispatch(tx, { companyId: issueCompany.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         return tx
           .update(issues)
@@ -12219,7 +12221,7 @@ export function issueService(db: Db) {
           const staleExecutionRunId = current.executionRunId;
           const adopted = await db.transaction(async tx => {
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
-            await notifyRunDispatch(tx);
+            await notifyRunDispatch(tx, { companyId: issueCompany.companyId });
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
             return tx
               .update(issues)
@@ -12475,7 +12477,7 @@ export function issueService(db: Db) {
         await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
-        await notifyRunDispatch(tx);
+        await notifyRunDispatch(tx, { companyId: existing.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const updated = await tx
           .update(issues)
@@ -12536,7 +12538,7 @@ export function issueService(db: Db) {
           await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id });
         }
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
-        await notifyRunDispatch(tx);
+        await notifyRunDispatch(tx, { companyId: existing.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const updated = await tx
           .update(issues)

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

@@ -77,14 +77,18 @@ async function renewOnLivenessPool(
   });
 }
 
-export async function hasLiveLegacyController(db: Db, run: Run): Promise<boolean> {
-  if (run.runtimeMode === "native" || !run.controllerBootId) return false;
-  const [owner] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
+export async function liveLegacyControllerExpiresAt(db: Db, run: Run): Promise<Date | null> {
+  if (run.runtimeMode === "native" || !run.controllerBootId) return null;
+  const [owner] = await db.select({ expiresAt: heartbeatRuns.controllerLeaseExpiresAt }).from(heartbeatRuns).where(and(
     eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
     eq(heartbeatRuns.status, "running"),
     gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
   ));
-  return Boolean(owner);
+  return owner?.expiresAt ?? null;
+}
+
+export async function hasLiveLegacyController(db: Db, run: Run): Promise<boolean> {
+  return (await liveLegacyControllerExpiresAt(db, run)) !== null;
 }
 
 export type RevokedLegacyControllerState = Pick<Run, "executionStage" | "processPid" | "processGroupId">;

server/src/services/native-runtime/native-provider-capacity-retry.ts

@@ -50,7 +50,7 @@ export async function materializeNativeProviderCapacityRetry(input: {
     scheduledRetryAt: dueAt.toISOString(),
     executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt),
   };
-  await notifyRunDispatch(input.tx);
+  await notifyRunDispatch(input.tx, { companyId: run.companyId });
   const [wake] = await input.tx.insert(agentWakeupRequests).values({
     companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system",
     reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId,

server/src/services/recovery/service.ts

@@ -27,7 +27,7 @@ import {
   decideLegacyContinuation, legacyDispositionEpisode, legacyDispositionFingerprint,
   LEGACY_DISPOSITION_REPAIR_INSTRUCTION, type LegacyDispositionEpisode,
 } from "./legacy-continuation.js";
-import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController } from "../legacy-controller-lease.js";
+import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController, liveLegacyControllerExpiresAt } from "../legacy-controller-lease.js";
 import { isCompletedOnboardingHandoffWake } from "../chat-completion-delivery.js";
 import { instanceSettingsService } from "../instance-settings.js";
 import { isWaitingConversation, settleConversationTurnForRun, deliverConversationComments } from "../agent-conversations.js";
@@ -245,6 +245,7 @@ type ResolvedDependencyWakeBackstopOptions = {
   runId?: string | null;
   companyId?: string | null;
   blockerIssueId?: string | null;
+  issueId?: string;
   source?: ResolvedDependencyWakeBackstopSource;
 };
 
@@ -2717,7 +2718,7 @@ export function recoveryService(
     const now = new Date();
     const retryAt = readProviderQuotaRetryAt(input.latestRun, now);
     return database.transaction(async (tx) => {
-      await notifyRunDispatch(tx);
+      await notifyRunDispatch(tx, { companyId: input.issue.companyId });
       const wakeup = await tx
         .insert(agentWakeupRequests)
         .values({
@@ -3318,7 +3319,7 @@ export function recoveryService(
           created = Boolean(enqueuedRun);
         } else {
           scheduledRun = await db.transaction(async (tx) => {
-            await notifyRunDispatch(tx);
+            await notifyRunDispatch(tx, { companyId: input.issue.companyId });
             const wakeup = await tx
               .insert(agentWakeupRequests)
               .values({
@@ -4526,6 +4527,9 @@ export function recoveryService(
   }
 
   async function reconcileStrandedAssignedIssues(opts?: {
+    companyId?: string;
+    issueId?: string;
+    onDeferredDeadline?: (at: number) => void;
     issueCreatedAtGte?: Date | null;
     skipExternalConversationSettlement?: boolean;
   }) {
@@ -4534,6 +4538,8 @@ export function recoveryService(
       .from(issues)
       .where(
         and(
+          opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined,
+          opts?.issueId ? eq(issues.id, opts.issueId) : undefined,
           isNull(issues.assigneeUserId),
           inArray(issues.status, ["todo", "in_progress", "in_review"]),
           or(
@@ -5714,6 +5720,7 @@ export function recoveryService(
               classification.baseBackoffMs *
               Math.pow(2, Math.max(0, consecutive - 1));
             if (elapsed < requiredDelay) {
+              opts?.onDeferredDeadline?.(latestFinishedAt.getTime() + requiredDelay);
               result.skipped += 1;
               continue;
             }
@@ -5815,7 +5822,7 @@ export function recoveryService(
       source === "workspace.finalize"
         ? "workspace_finalize_reconciliation"
         : "issue_graph_liveness_reconciliation";
-    const useCursor = !opts?.blockerIssueId;
+    const useCursor = !opts?.blockerIssueId && !opts?.issueId;
 
     const queryCandidates = (afterIssueId: string | null) => {
       const filters = [
@@ -5825,6 +5832,7 @@ export function recoveryService(
         sql`${issues.assigneeAgentId} is not null`,
       ];
       if (opts?.companyId) filters.push(eq(issues.companyId, opts.companyId));
+      if (opts?.issueId) filters.push(eq(issues.id, opts.issueId));
       if (afterIssueId) filters.push(gt(issues.id, afterIssueId));
 
       if (opts?.blockerIssueId) {
@@ -6115,6 +6123,7 @@ export function recoveryService(
       // fallback below. Without it, a terminal issue named in the run context
       // snapshot would still terminalize the shared run and defeat the guard.
       runReferencedByActiveIssue?: boolean;
+      onDeferredDeadline?: (at: number) => void;
     },
   ): Promise<{ terminalized: boolean; status: string }> {
     // Act only on a run in "running" status. A "queued" run has no process yet,
@@ -6130,8 +6139,11 @@ export function recoveryService(
     // Another controller may own a sandbox run whose PID has no meaning on
     // this host. Its live lease owns both execution and finalization, even if
     // the issue is already terminal or this process has no in-memory handle.
-    if (await hasLiveLegacyController(db, run))
+    const controllerDeadline = await liveLegacyControllerExpiresAt(db, run);
+    if (controllerDeadline) {
+      options?.onDeferredDeadline?.(controllerDeadline.getTime());
       return { terminalized: false, status: run.status };
+    }
 
     const pid = run.processPid ?? null;
     const processGroupId = run.processGroupId ?? null;
@@ -6352,7 +6364,7 @@ export function recoveryService(
   // still claims to be live but can no longer reach a terminal status on its
   // own, so a stuck "running" run can no longer block the sweep. Idempotent and
   // safe: clears at most one row's worth of lock columns per candidate.
-  async function sweepStaleIssueLocks() {
+  async function sweepStaleIssueLocks(opts?: { companyId?: string; issueId?: string; onDeferredDeadline?: (at: number) => void }) {
     const result = {
       cleared: 0,
       issueIds: [] as string[],
@@ -6369,7 +6381,9 @@ export function recoveryService(
       })
       .from(issues)
       .where(
-        sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`,
+        and(sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`,
+          opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined,
+          opts?.issueId ? eq(issues.id, opts.issueId) : undefined),
       );
 
     const referencedRunIds = [
@@ -6384,7 +6398,8 @@ export function recoveryService(
         ? await db
             .select()
             .from(heartbeatRuns)
-            .where(inArray(heartbeatRuns.id, referencedRunIds))
+            .where(and(inArray(heartbeatRuns.id, referencedRunIds),
+              opts?.companyId ? eq(heartbeatRuns.companyId, opts.companyId) : undefined))
         : [];
     const runStatusById = new Map<string, string>();
     for (const row of runRows) runStatusById.set(row.id, row.status);
@@ -6395,8 +6410,15 @@ export function recoveryService(
     // the same run id in a stale lock column. The terminal reference alone must
     // not terminalize a run that an active issue still owns, so exclude these
     // runs from the issue-terminal authority below.
+    const ownershipReferences = opts?.issueId && referencedRunIds.length > 0
+      ? await db.select({ companyId: issues.companyId, status: issues.status,
+          checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId }).from(issues).where(and(
+          opts.companyId ? eq(issues.companyId, opts.companyId) : undefined,
+          or(inArray(issues.checkoutRunId, referencedRunIds), inArray(issues.executionRunId, referencedRunIds)),
+        ))
+      : candidates;
     const runIdsReferencedByActiveIssue = new Set<string>();
-    for (const issue of candidates) {
+    for (const issue of ownershipReferences) {
       if (issue.status === "done" || issue.status === "cancelled") continue;
       for (const runId of [issue.checkoutRunId, issue.executionRunId]) {
         if (runId) runIdsReferencedByActiveIssue.add(runId);
@@ -6437,6 +6459,7 @@ export function recoveryService(
         referencingIssueTerminalStatus:
           issueTerminalStatusByRunId.get(row.id) ?? null,
         runReferencedByActiveIssue: runIdsReferencedByActiveIssue.has(row.id),
+        onDeferredDeadline: opts?.onDeferredDeadline,
       });
       runStatusById.set(row.id, outcome.status);
       if (outcome.terminalized) result.terminalizedRunIds.push(row.id);

server/src/services/recovery/work.ts

@@ -0,0 +1,58 @@
+import { registerTaskWatchdogWork } from "../task-watchdog-work.js";
+import { and, eq, inArray, isNotNull, isNull, or } from "drizzle-orm";
+import { issues, 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 { IssueRecoveryScope } from "../issue-recovery-notifications.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+
+type Entry = { id: string; companyId: string; dueAt: number };
+type Scope = { companyId: string; issueId: string; onDeferredDeadline: (at: number) => void };
+
+export function registerIssueRecoveryWork(
+  db: Db,
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  owner: {
+    stranded(scope: Scope): Promise<unknown>;
+    dependencies(scope: Scope): Promise<{ enqueueFailed: number }>;
+    locks(scope: Scope): Promise<unknown>;
+    watchdogs(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>;
+    schedulingSuppressed(): Promise<boolean>;
+  },
+  canRun: () => boolean,
+) {
+  function register(queue: DeliveryQueue, statuses: string[] | null, check: (scope: Scope) => Promise<unknown>) {
+    return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
+      queue,
+      decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
+      matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
+      load: async scope => {
+        const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and(
+          scope ? eq(issues.companyId, scope.companyId) : undefined,
+          statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)),
+          statuses ? isNull(issues.hiddenAt) : undefined,
+        ));
+        return rows.map(row => ({ ...row, dueAt: Date.now() }));
+      },
+      check: async entry => {
+        if (await owner.schedulingSuppressed()) return null;
+        let next: number | null = null;
+        await check({ companyId: entry.companyId, issueId: entry.id,
+          onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } });
+        return next === null ? null : { ...entry, dueAt: next };
+      },
+    }, canRun);
+  }
+  const assignments = register(DELIVERY_QUEUES.strandedAssignments, ["todo", "in_progress", "in_review"], owner.stranded);
+  const dependencies = register(DELIVERY_QUEUES.resolvedDependencies, ["blocked"], async scope => {
+    const result = await owner.dependencies(scope);
+    if (result.enqueueFailed > 0) throw new Error("Resolved dependency wake remains undelivered");
+  });
+  const locks = register(DELIVERY_QUEUES.staleIssueLocks, null, owner.locks);
+  const watchdogs = registerTaskWatchdogWork(db, coordinator, {
+    reconcile: owner.watchdogs, enabled: async () => !(await owner.schedulingSuppressed()),
+  }, canRun);
+  return { ready: Promise.all([assignments.ready, dependencies.ready, locks.ready, watchdogs.ready]),
+    wake() { assignments.invalidate(); dependencies.invalidate(); locks.invalidate(); watchdogs.invalidate(); },
+    async stop() { await Promise.all([assignments.stop(), dependencies.stop(), locks.stop(), watchdogs.stop()]); } };
+}

server/src/services/run-dispatch-notifications.ts

@@ -1,6 +1,8 @@
+import { notifyIssueRecovery, type IssueRecoveryScope } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export async function notifyRunDispatch(transaction: object) {
+export async function notifyRunDispatch(transaction: object, scope?: IssueRecoveryScope) {
+  await notifyIssueRecovery(transaction, scope);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch);
 }

server/src/services/scheduled-retry-cancellation-notifications.ts

@@ -4,6 +4,6 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type ScheduledRetryCancellationScope = { companyId: string; runId?: string };
 
 export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) {
-  await notifyRunDispatch(transaction);
+  await notifyRunDispatch(transaction, scope);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*");
 }

server/src/services/task-watchdog-work.ts

@@ -0,0 +1,36 @@
+import { and, eq } from "drizzle-orm";
+import { issueWatchdogs, type Db } from "@paperclipai/db";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import type { IssueRecoveryScope } from "./issue-recovery-notifications.js";
+import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
+
+type Entry = { id: string; companyId: string; dueAt: number };
+
+export function registerTaskWatchdogWork(
+  db: Db,
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  owner: {
+    enabled(): Promise<boolean>;
+    reconcile(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>;
+  },
+  canRun: () => boolean,
+) {
+  return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
+    queue: DELIVERY_QUEUES.taskWatchdogs,
+    decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
+    matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
+    load: async scope => (await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId })
+      .from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"),
+        scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined)))
+      .map(row => ({ ...row, dueAt: Date.now() })),
+    check: async entry => {
+      if (!(await owner.enabled())) return null;
+      let next: number | null = null;
+      const result = await owner.reconcile({ companyId: entry.companyId, watchdogId: entry.id,
+        onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } });
+      if (result.incomplete > 0) throw new Error("Task watchdog subtree remains incomplete");
+      return next === null ? null : { ...entry, dueAt: next };
+    },
+  }, canRun);
+}

server/src/services/task-watchdogs.ts

@@ -1,3 +1,4 @@
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { createHash } from "node:crypto";
 import { and, asc, eq, gte, inArray, isNull, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -831,7 +832,7 @@ async function updateIssueWatchdogRow(
   return updated;
 }
 
-export async function upsertIssueWatchdogForIssue(
+async function upsertIssueWatchdogInTransaction(
   dbOrTx: any,
   companyId: string,
   issueId: string,
@@ -852,7 +853,7 @@ export async function upsertIssueWatchdogForIssue(
     return { watchdog: toIssueWatchdog(updated), created: false };
   }
 
-  const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx
+  const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx.transaction(async (tx: Db) => tx
     .insert(issueWatchdogs)
     .values({
       companyId,
@@ -870,7 +871,7 @@ export async function upsertIssueWatchdogForIssue(
       updatedAt: now,
     })
     .returning()
-    .then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true }))
+    .then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true })))
     .catch(async (error: unknown) => {
       if (!isIssueWatchdogUniqueConflict(error)) throw error;
       const winner = await dbOrTx
@@ -885,6 +886,18 @@ export async function upsertIssueWatchdogForIssue(
   return { watchdog: toIssueWatchdog(insertResult.row), created: insertResult.created };
 }
 
+export function upsertIssueWatchdogForIssue(
+  dbOrTx: Db | Parameters<Parameters<Db["transaction"]>[0]>[0],
+  companyId: string,
+  issueId: string,
+  input: IssueWatchdogUpsertInput,
+): Promise<{ watchdog: IssueWatchdog; created: boolean }> {
+  return dbOrTx.transaction(async tx => {
+    await notifyIssueRecovery(tx, { companyId });
+    return upsertIssueWatchdogInTransaction(tx, companyId, issueId, input);
+  });
+}
+
 export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) {
   const issuesSvc = issueService(db);
 
@@ -1478,7 +1491,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
     return created;
   }
 
-  async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null } = {}) {
+  async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null; onDeferredDeadline?: (at: number) => void } = {}) {
     const watchdog = await markTerminalWatchdogIssueReviewed(row, opts);
     const sourceIssue = await db
       .select()
@@ -1491,6 +1504,13 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
 
     const input = await collectClassifierInput(watchdog.companyId, watchdog);
     const classification = classifyTaskWatchdogSubtree(input);
+    if (classification.state === "pending_first_run") {
+      const pending = new Set(classification.pendingIssueIds);
+      for (const issue of input.issues) if (pending.has(issue.id)) {
+        const createdAt = toEpochMs(issue.createdAt);
+        if (createdAt !== null) opts.onDeferredDeadline?.(createdAt + TASK_WATCHDOG_FIRST_RUN_GRACE_MS);
+      }
+    }
     if (classification.state !== "stopped") {
       return { state: classification.state, reason: classification.reason, classification };
     }
@@ -1610,13 +1630,14 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
     };
   }
 
-  async function listActiveWatchdogsForCompany(companyId?: string | null) {
+  async function listActiveWatchdogsForCompany(companyId?: string | null, watchdogId?: string) {
     return db
       .select()
       .from(issueWatchdogs)
       .where(and(
         eq(issueWatchdogs.status, "active"),
         ...(companyId ? [eq(issueWatchdogs.companyId, companyId)] : []),
+        watchdogId ? eq(issueWatchdogs.id, watchdogId) : undefined,
       ));
   }
 
@@ -1746,14 +1767,16 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
       issueId: string,
       actor: ActorFields = {},
     ): Promise<IssueWatchdog | null> => {
-      await assertWatchedIssue(db, companyId, issueId);
-      const existing = await db
+      return db.transaction(async tx => {
+      await assertWatchedIssue(tx, companyId, issueId);
+      const existing = await tx
         .select()
         .from(issueWatchdogs)
         .where(and(eq(issueWatchdogs.companyId, companyId), eq(issueWatchdogs.issueId, issueId)))
         .then((rows) => rows[0] ?? null);
       if (!existing || existing.status === "disabled") return null;
-      const [updated] = await db
+      await notifyIssueRecovery(tx, { companyId });
+      const [updated] = await tx
         .update(issueWatchdogs)
         .set({
           status: "disabled",
@@ -1765,14 +1788,17 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
         .where(eq(issueWatchdogs.id, existing.id))
         .returning();
       return toIssueWatchdog(updated);
+      });
     },
 
     reconcileTaskWatchdogs: async (opts: {
       companyId?: string | null;
       runId?: string | null;
+      watchdogId?: string;
+      onDeferredDeadline?: (at: number) => void;
       issueCreatedAtGte?: Date | null;
     } = {}) => {
-      let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null);
+      let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null, opts.watchdogId);
       if (opts.issueCreatedAtGte) {
         const watchdogIssueIds = [...new Set(rows.map((row) => row.issueId))];
         const eligibleIssueIds = new Set(
@@ -1802,7 +1828,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
       };
       for (const row of rows) {
         result.checked += 1;
-        const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null });
+        const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null, onDeferredDeadline: opts.onDeferredDeadline });
         if (evaluated.state === "triggered") {
           result.triggered += 1;
           result.watchdogIssueIds.push(evaluated.watchdogIssueId);