Integrate event-driven issue recovery and source fanout

Preserve routine restore, card, retention, and issue recovery notifications in their original writer transactions.

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

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

@@ -0,0 +1,18 @@
+# Issue recovery without periodic scans
+
+Four existing owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database.
+
+| Owner | Durable truth and producer boundary | Deadline or event | Exit |
+| --- | --- | --- | --- |
+| Stranded assignment | Visible assigned todo/in-progress/review issues; issue create/update/checkout/release; run lifecycle and execution control; agent/company/budget/configuration; interaction, approval and subtree hold changes | A committed eligibility change. A transient continuation uses the latest finish plus the existing exponential delay. | An execution path, durable wait, escalation, or current ineligibility. |
+| Resolved dependency | Blocked assigned issues and company-scoped blocker relations; blocker status or relation changes through issue service and import; run/control and pause gate changes | The committing change. A failed enqueue retains that issue with error backoff. | Existing idempotent wake, live path, unresolved dependency or deliberate self-block. |
+| Task watchdog | Active issue watchdogs and their current source subtree; normal watchdog upsert/disable; issue/comment/document/work-product and wait changes | Exact source creation plus the existing 15-second first-run grace. An incomplete scan retains the watchdog with error backoff. | Live path, open/reviewed reusable review, disabled configuration, or ineligible source. |
+| Stale issue lock | Existing checkout/execution run references; issue and run lifecycle changes | A committing ownership change, or the fresh owning controller's saved lease expiry. | No stale reference, a valid owner, or the existing guarded terminalization and lock clear. |
+
+The owner methods accept company and issue/watchdog identity. A narrowed lock check still reads other same-company references before using terminal-issue authority. A stale terminal alias cannot end an active issue's shared run. The controller lease owner supplies its fresh expiry; the existing write predicate revalidates the observed boot ID and database-clock expiry.
+
+Producer coverage uses the existing run-dispatch, execution-control, issue-review-state, native status reassessment and external conversation settlement boundaries. Direct gaps have explicit signals inside their owning transaction: document writes/import, attachment removal, blocker import, recovery-action adoption/mutation, conversation pause release, Telegram failure comment replacement, native artifact title/metadata updates and company deletion. Metadata-only comment classification that does not change timestamp, visibility or execution eligibility needs no additional watchdog activation. Approval links and decisions use their existing issue-review/continuation effects. Work-product create/update/delete/import uses its existing native status boundary.
+
+The new owners replace only these four startup and periodic calls. Deferred comments, silent active runs and residual orphan/resource recovery retain their existing backstop until the next slice. Every previously migrated native-owner skip remains unchanged.
+
+A controlled process test prints an execution proof but cannot apply a task disposition. The test operator completes that task through issue service. If the existing continuation policy already saved a disposition-repair retry, the test preserves it until its stored deadline. Only after all provider work has joined does it advance the clock; dispatch then revalidates the completed task and cancels the retry. The final quiet proof requires no registered deadlines, no tracked in-flight work and no owner SQL over twenty former polling minutes. It does not claim a saved future retry is already drained.

server/src/__tests__/issue-recovery-deadlines.test.ts

@@ -0,0 +1,237 @@
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { randomUUID } from "node:crypto";
+import { and, eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issues, issueWatchdogs } from "@paperclipai/db";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { heartbeatService } from "../services/heartbeat.js";
+import { issueService } from "../services/issues.js";
+import { taskWatchdogService } from "../services/task-watchdogs.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_RECOVERY_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+describe.skipIf(!support.supported)("issue recovery production producers and deadlines", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  let heartbeat: ReturnType<typeof heartbeatService>;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
+  let worker: ReturnType<ReturnType<typeof heartbeatService>["registerIssueStateRecoveryWork"]>;
+  let dispatch: ReturnType<ReturnType<typeof heartbeatService>["registerRunDispatchWork"]>;
+  let cancellation: ReturnType<ReturnType<typeof heartbeatService>["registerScheduledRetryCancellationWork"]>;
+  let draining = false;
+  const tasks = new Map<string, { nextRunAt?: () => number | null; hasPending: () => Promise<boolean> }>();
+  const errors: unknown[] = [];
+  const extraStops: Array<() => Promise<void>> = [];
+  const extraDeadlines: Array<() => number | null> = [];
+  beforeAll(async () => {
+    if (!external) database = await startEmbeddedPostgresTestDatabase("issue-recovery-work-");
+    db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+    await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+  }, 90_000);
+  beforeEach(async () => {
+    await db.execute(sql`TRUNCATE companies CASCADE`);
+    heartbeat = heartbeatService(db); draining = false; errors.length = 0;
+  });
+  afterEach(async () => {
+    for (const stop of extraStops.splice(0)) await stop();
+    extraDeadlines.length = 0;
+    await worker?.stop(); await dispatch?.stop(); await cancellation?.stop(); await coordinator?.stop();
+    await heartbeat.drainActiveRunExecutions();
+    vi.useRealTimers(); vi.restoreAllMocks();
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
+  async function start() {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
+      onError: error => errors.push(error) });
+    tasks.clear();
+    const register = coordinator.register.bind(coordinator);
+    coordinator.register = (queue, task) => { tasks.set(queue, task); return register(queue, task); };
+    worker = heartbeat.registerIssueStateRecoveryWork(coordinator, () => !draining);
+    dispatch = heartbeat.registerRunDispatchWork(coordinator, () => !draining);
+    cancellation = heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !draining);
+    await Promise.all([worker.ready, dispatch.ready, cancellation.ready]);
+  }
+  async function compete() {
+    const other = heartbeatService(db);
+    const competing = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const work = other.registerIssueStateRecoveryWork(competing, () => true);
+    extraStops.push(() => competing.stop()); extraDeadlines.push(competing.nextWakeAt); await work.ready;
+  }
+  async function fixture() {
+    const companyId = randomUUID();
+    await db.insert(authUsers).values({ id: "recovery-user", name: "Recovery operator", email: "recovery-user@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing();
+    await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: `R${companyId.slice(0, 7)}`,
+      requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "recovery-user" });
+    await db.insert(companyMemberships).values({ companyId, principalId: "recovery-user", principalType: "user", status: "active", membershipRole: "owner" });
+    const [agent] = await db.insert(agents).values({ companyId, name: "Recovery process", role: "engineer", status: "idle",
+      adapterType: "process", adapterConfig: { command: process.execPath, args: ["-e", "console.log('recovery-proof')"], cwd: process.cwd() },
+      runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } },
+    }).returning();
+    return { companyId, agentId: agent.id };
+  }
+  async function settled() {
+    await vi.waitFor(async () => {
+      await heartbeat.drainActiveRunExecutions();
+      const wake = coordinator.nextWakeAt();
+      const detail = wake === null ? "" : JSON.stringify({ now: Date.now(), tasks: await Promise.all([...tasks].map(async ([queue, task]) => ({ queue, at: task.nextRunAt?.(), pending: await task.hasPending() }))),
+        runs: (await db.select().from(heartbeatRuns)).map(run => ({ id: run.id, status: run.status, error: run.errorCode, lease: run.controllerLeaseExpiresAt, reason: run.contextSnapshot?.wakeReason })) });
+      expect(wake, detail).toBeNull();
+      expect(extraDeadlines.map(next => next())).toEqual(extraDeadlines.map(() => null));
+      expect(idleWorkSnapshot().active).toBe(0);
+    }, { timeout: 10_000 });
+  }
+  async function quiet() {
+    await settled();
+    const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute");
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); }
+    finally { vi.useRealTimers(); reads.mockRestore(); executes.mockRestore(); }
+  }
+  async function finish(issueId: string) {
+    await issueService(db).update(issueId, { status: "done" });
+    await heartbeat.drainActiveRunExecutions();
+    const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
+      sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
+    if (saved.length > 0) {
+      const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime()));
+      expect(await db.select().from(heartbeatRuns).where(and(
+        sql`${heartbeatRuns.status} in ('queued', 'running')`,
+        sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`))).toHaveLength(0);
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+      await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1);
+      await vi.waitFor(async () => {
+        const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
+            sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
+        expect(remaining).toHaveLength(0);
+      });
+      for (const retry of saved) {
+        const [settledRetry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retry.id));
+        expect(settledRetry.status).toBe("cancelled");
+      }
+    }
+  }
+  async function completed(issueId: string) {
+    let runId: string | undefined;
+    await vi.waitFor(async () => {
+      const runs = await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`);
+      const run = runs.find(row => row.status === "running" || row.status === "succeeded");
+      expect(run).toBeDefined(); runId = run!.id;
+    }, { timeout: 15_000 });
+    await heartbeat.drainActiveRunExecutions();
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
+    expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode, stderr: run.stderrExcerpt, result: run.resultJson })).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("recovery-proof");
+    return run;
+  }
+  it("keeps empty, configured idle, and terminal history quiet", async () => {
+    await fixture(); await start(); await quiet();
+  });
+  it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => {
+    const f = await fixture(); await start(); await quiet();
+    const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId });
+    await completed(issue.id);
+    await finish(issue.id); await quiet();
+  }, 45_000);
+  it("reacts to blocker resolution once and dispatches the durable dependency wake", async () => {
+    const f = await fixture();
+    const blocker = await issueService(db).create(f.companyId, { title: "Blocker", status: "todo" });
+    const dependent = await issueService(db).create(f.companyId, { title: "Dependent", status: "blocked", assigneeAgentId: f.agentId,
+      blockedByIssueIds: [blocker.id] });
+    await start(); await quiet();
+    await compete();
+    await issueService(db).update(blocker.id, { status: "done" });
+    const run = await completed(dependent.id);
+    expect(run.contextSnapshot?.wakeReason).toBe("issue_blockers_resolved");
+    await finish(dependent.id); await quiet();
+    expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
+      sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'issue_blockers_resolved'`))).toHaveLength(1);
+  }, 45_000);
+  it("uses ordinary watchdog configuration and subtree closure to dispatch a reusable review", async () => {
+    const f = await fixture();
+    const source = await issueService(db).create(f.companyId, { title: "Finished subtree", status: "done" });
+    await start(); await quiet();
+    await compete();
+    const watchdog = await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId });
+    let reviewId: string | undefined;
+    await vi.waitFor(async () => {
+      const [row] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.id, watchdog.watchdog.id));
+      expect(row.watchdogIssueId).toBeTruthy(); reviewId = row.watchdogIssueId!;
+    }, { timeout: 15_000 });
+    await completed(reviewId!);
+    expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
+      sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'task_watchdog_stopped_subtree'`))).toHaveLength(1);
+    await finish(reviewId!);
+    await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet();
+  }, 45_000);
+  it("bounds ordinary writes to their company and reconstructs foreign stale debt after restart", async () => {
+    const selected = await fixture(); await start(); await quiet();
+    const foreign = await fixture();
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: foreign.companyId, agentId: foreign.agentId,
+      status: "failed", finishedAt: new Date() }).returning();
+    const [stale] = await db.insert(issues).values({ companyId: foreign.companyId, title: "Foreign stale lock", status: "done",
+      checkoutRunId: run.id }).returning();
+    await issueService(db).create(selected.companyId, { title: "Selected change", status: "done" }); await quiet();
+    expect((await db.select().from(issues).where(eq(issues.id, stale.id)))[0].checkoutRunId).toBe(run.id);
+    await worker.stop(); await dispatch.stop(); await cancellation.stop(); await coordinator.stop();
+    heartbeat = heartbeatService(db); await start(); await quiet();
+    expect((await db.select().from(issues).where(eq(issues.id, stale.id)))[0].checkoutRunId).toBeNull();
+  });
+  it("fences nested assignment writes until outer commit and ignores rollback", async () => {
+    const f = await fixture(); await start(); await quiet();
+    let committedId = "";
+    await db.transaction(async outer => {
+      await outer.transaction(async nested => {
+        const issue = await issueService(db).create(f.companyId, { title: "Nested commit", status: "todo", assigneeAgentId: f.agentId }, nested);
+        committedId = issue.id;
+      });
+      await new Promise(resolve => setTimeout(resolve, 100));
+      expect(await db.select().from(heartbeatRuns)).toHaveLength(0);
+    });
+    await completed(committedId); await finish(committedId); await settled();
+    await expect(db.transaction(async outer => {
+      await outer.transaction(async nested => {
+        await issueService(db).create(f.companyId, { title: "Nested rollback", status: "todo", assigneeAgentId: f.agentId }, nested);
+      });
+      throw new Error("rollback control");
+    })).rejects.toThrow("rollback control");
+    await quiet();
+    expect(await db.select().from(issues).where(eq(issues.title, "Nested rollback"))).toHaveLength(0);
+  }, 60_000);
+  it("retains the first committed assignment during drain and executes on resume", async () => {
+    const f = await fixture(); await start(); await quiet(); draining = true;
+    const issue = await issueService(db).create(f.companyId, { title: "Drained assignment", status: "todo", assigneeAgentId: f.agentId });
+    await new Promise(resolve => setTimeout(resolve, 100));
+    const reads = vi.spyOn(db, "select");
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try { await vi.advanceTimersByTimeAsync(90_000); expect(reads).not.toHaveBeenCalled(); }
+    finally { vi.useRealTimers(); reads.mockRestore(); }
+    expect(await db.select().from(heartbeatRuns)).toHaveLength(0);
+    draining = false; worker.wake(); dispatch.wake();
+    await completed(issue.id); await finish(issue.id); await quiet();
+  }, 45_000);
+  it("retains a fresh subtree's exact first-run grace then disarms on disable", async () => {
+    const f = await fixture();
+    const source = await issueService(db).create(f.companyId, { title: "Fresh unassigned subtree", status: "todo" });
+    await start(); await quiet();
+    await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId });
+    await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(source.createdAt.getTime() + 15_000), { timeout: 5_000 });
+    expect(await db.select().from(issues).where(eq(issues.originKind, "task_watchdog"))).toHaveLength(0);
+    await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet();
+  });
+  it("waits for a live foreign controller's exact stored lease before clearing a terminal lock", async () => {
+    const f = await fixture();
+    const source = await issueService(db).create(f.companyId, { title: "Remote owner", status: "done" });
+    const lease = new Date(Date.now() + 5_000);
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running",
+      runtimeMode: "legacy", controllerBootId: randomUUID(), controllerLeaseExpiresAt: lease,
+      contextSnapshot: { issueId: source.id }, processPid: 2_000_000_000, startedAt: new Date() }).returning();
+    await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, source.id));
+    await start();
+    await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(lease.getTime()), { timeout: 3_000 });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("running");
+    await vi.waitFor(async () => expect((await db.select().from(issues).where(eq(issues.id, source.id)))[0].executionRunId).toBeNull(), { timeout: 12_000 });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded"); await quiet();
+  }, 30_000);
+});

server/src/__tests__/issue-recovery-work-lifecycle.test.ts

@@ -0,0 +1,99 @@
+import { PgDialect } from "drizzle-orm/pg-core";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { issueWatchdogs } from "@paperclipai/db";
+import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js";
+import { registerIssueRecoveryWork } from "../services/recovery/work.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const stops: Array<() => Promise<void>> = [];
+afterEach(async () => { for (const stop of stops.splice(0)) await stop(); vi.useRealTimers(); });
+function fixture() {
+  vi.useFakeTimers(); vi.setSystemTime(1_000_000);
+  const dialect = new PgDialect();
+  const rows = { issues: [] as Array<{ id: string; companyId: string }>, watchdogs: [] as Array<{ id: string; companyId: string }> };
+  const execute = vi.fn(async (query: any) => {
+    if (dialect.sqlToQuery(query).sql.includes("pg_current_xact_id")) return [{ xid: "42" }];
+    throw new Error("unexpected SQL");
+  });
+  const load = vi.fn((table: unknown) => table === issueWatchdogs ? rows.watchdogs : rows.issues);
+  const db: any = installDatabaseWorkSignals({ execute,
+    select: () => ({ from: (table: unknown) => ({ where: async () => load(table) }) }),
+    transaction: async (callback: (tx: any) => Promise<unknown>) => callback({ execute, select: db.select, transaction: db.transaction }),
+  });
+  const state = { draining: false, standby: false, suppressed: false, deadline: null as number | null, fail: false, incomplete: false };
+  const check = vi.fn(async (scope: { onDeferredDeadline(at: number): void }) => {
+    if (state.fail) throw new Error("database unavailable");
+    if (state.deadline !== null && state.deadline > Date.now()) scope.onDeferredDeadline(state.deadline);
+  });
+  const owner = { stranded: check, dependencies: async (scope: any) => { await check(scope); return { enqueueFailed: 0 }; },
+    locks: check, watchdogs: async (scope: any) => { await check(scope); return { incomplete: state.incomplete ? 1 : 0 }; },
+    schedulingSuppressed: async () => state.suppressed };
+  const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !state.standby,
+    canReconcile: () => !state.standby, onError: vi.fn() });
+  const worker = registerIssueRecoveryWork(db, coordinator, owner, () => !state.draining && !state.standby);
+  stops.push(() => coordinator.stop());
+  return { db, rows, state, check, load, coordinator, worker,
+    notify: () => db.transaction((tx: any) => notifyIssueRecovery(tx, { companyId: "selected" })),
+    settle: () => vi.advanceTimersByTimeAsync(1_000) };
+}
+describe("issue recovery obligation lifecycle", () => {
+  it("has no timer, SQL, or owner probe after empty reconstruction", async () => {
+    const f = fixture(); await f.worker.ready;
+    expect(f.coordinator.nextWakeAt()).toBeNull(); const loaded = f.load.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(20 * 60_000);
+    expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled();
+  });
+  it("retains exact grace and lease deadlines without early owner probes", async () => {
+    const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" });
+    f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.deadline = 1_015_000;
+    await f.worker.ready; expect(f.coordinator.nextWakeAt()).toBe(1_015_000);
+    const checked = f.check.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(14_999); expect(f.check).toHaveBeenCalledTimes(checked);
+    await vi.advanceTimersByTimeAsync(1); expect(f.check.mock.calls.length).toBe(checked * 2);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("releases settled producer holds during domain drain while coordinator remains open", async () => {
+    const f = fixture(); await f.worker.ready; f.state.draining = true;
+    f.rows.issues.push({ id: "issue", companyId: "selected" });
+    await f.notify(); await f.settle(); const loaded = f.load.mock.calls.length;
+    await vi.advanceTimersByTimeAsync(90_000);
+    expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled();
+    expect(idleWorkSnapshot().active).toBe(0);
+    f.state.draining = false; f.worker.wake(); await f.settle();
+    expect(f.check).toHaveBeenCalled(); expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("retries concrete failed entries and then disarms without reloading the company", async () => {
+    const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); f.state.fail = true;
+    await f.worker.ready; const loaded = f.load.mock.calls.length;
+    expect(f.coordinator.nextWakeAt()).toBe(1_030_000);
+    await vi.advanceTimersByTimeAsync(29_999); const checked = f.check.mock.calls.length;
+    f.state.fail = false; await vi.advanceTimersByTimeAsync(1);
+    expect(f.check.mock.calls.length).toBe(checked * 2); expect(f.load).toHaveBeenCalledTimes(loaded);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("keeps an incomplete watchdog subtree as concrete retry debt", async () => {
+    const f = fixture(); f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.incomplete = true;
+    await f.worker.ready; expect(f.coordinator.nextWakeAt()).toBe(1_030_000);
+    const loaded = f.load.mock.calls.length; f.state.incomplete = false;
+    await vi.advanceTimersByTimeAsync(30_000);
+    expect(f.check).toHaveBeenCalledTimes(2); expect(f.load).toHaveBeenCalledTimes(loaded);
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("keeps a concurrent producer dirty until the in-flight owner has completed", async () => {
+    const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); await f.worker.ready;
+    let release!: () => void; const held = new Promise<void>(resolve => { release = resolve; });
+    f.check.mockImplementationOnce(() => held);
+    await f.notify(); await f.settle(); await f.notify(); release(); await f.settle(); await f.settle();
+    expect(f.coordinator.nextWakeAt()).toBeNull();
+    expect(f.check.mock.calls.length).toBeGreaterThan(3);
+  });
+  it("retains startup work on standby and cancels future deadlines on shutdown", async () => {
+    const f = fixture(); f.state.standby = true; f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" });
+    await f.worker.ready; await vi.advanceTimersByTimeAsync(90_000); expect(f.load).not.toHaveBeenCalled();
+    f.state.standby = false; f.state.deadline = Date.now() + 60_000; f.worker.wake(); await f.settle();
+    expect(f.check).toHaveBeenCalledOnce(); await f.worker.stop();
+    await vi.advanceTimersByTimeAsync(120_000); expect(f.check).toHaveBeenCalledOnce();
+  });
+});

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -137,6 +137,7 @@ const {
     })),
     reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })),
     reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })),
+    registerIssueStateRecoveryWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
     registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
@@ -980,6 +981,7 @@ describe("startServer feedback export wiring", () => {
   it.each([false, true])("registers retry and queued dispatch only when maintenance is enabled (%s)", async enabled => {
     loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
     await startServer(); await settleStartupHousekeeping();
+    expect(heartbeatServiceMock.registerIssueStateRecoveryWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
     expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
     if (enabled) {
       const [coordinator, admitted] = heartbeatServiceMock.registerRunDispatchWork.mock.calls[0];
@@ -1393,11 +1395,11 @@ describe("startServer feedback export wiring", () => {
       await startServer();
       await settleStartupHousekeeping();
       expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
-      expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(1);
+      expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled();
       expect(intervalCallback).not.toBeNull();
       intervalCallback?.();
       await new Promise<void>((resolve) => setImmediate(resolve));
-      expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
+      expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled();
       expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true });
       expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
       expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled();

server/src/index.ts

@@ -112,7 +112,6 @@ import { createApp } from "./app.js";
 import { initializeCloudRuntimeConfig } from "./services/cloud-runtime-config.js";
 import { loadCloudBootstrapConfig, loadConfig, type Config } from "./config.js";
 import { logger } from "./middleware/logger.js";
-import { createIncompleteTaskWatchdogScanReporter } from "./services/task-watchdog-diagnostics.js";
 import { setStartupRecoveryPhase } from "./startup-recovery-state.js";
 import {
   StartupRefusalError,
@@ -974,9 +973,6 @@ async function startServerWithDatabaseTeardown(
     }
   };
   const pluginWorkerManager = createPluginWorkerManager();
-  const reportIncompleteWatchdogScans = createIncompleteTaskWatchdogScanReporter(
-    (details, message) => logger.warn(details, message),
-  );
   const heartbeat = config.heartbeatSchedulerEnabled
     ? heartbeatService(db as any, { pluginWorkerManager })
     : null;
@@ -1748,47 +1744,10 @@ async function startServerWithDatabaseTeardown(
             "startup session-goal recovery resumed durable agent goals",
           );
         }
-        const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
-        if (
-          reconciled.assignmentDispatched > 0 ||
-          reconciled.dispatchRequeued > 0 ||
-          reconciled.continuationRequeued > 0 ||
-          reconciled.successfulRunHandoffEscalated > 0 ||
-          reconciled.successfulRunHandoffRetried > 0 ||
-          reconciled.escalated > 0
-        ) {
-          logger.warn(
-            { ...reconciled },
-            "startup heartbeat recovery changed assigned issue state",
-          );
-        }
-
-        const dependencyWakesReconciled = await heartbeat.reconcileResolvedDependencyWakes();
-        if (dependencyWakesReconciled.healed > 0) {
-          logger.warn(
-            { ...dependencyWakesReconciled },
-            "startup dependency-wake reconciliation restored task execution paths",
-          );
-        }
-
-        const taskWatchdogsReconciled = await heartbeat.reconcileTaskWatchdogs();
-        reportIncompleteWatchdogScans(taskWatchdogsReconciled, "startup");
-        if (taskWatchdogsReconciled.triggered > 0) {
-          logger.warn(
-            { ...taskWatchdogsReconciled },
-            "startup task-watchdog reconciliation triggered watchdog work",
-          );
-        }
-
         const scanned = await heartbeat.scanSilentActiveRuns();
         if (scanned.created > 0 || scanned.escalated > 0) {
           logger.warn({ ...scanned }, "startup active-run output watchdog created review work");
         }
-
-        const swept = await heartbeat.sweepStaleIssueLocks();
-        if (swept.cleared > 0) {
-          logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks");
-        }
       })().catch((err) => {
         logger.error({ err }, "startup heartbeat recovery failed");
         throw err;
@@ -1797,6 +1756,8 @@ async function startServerWithDatabaseTeardown(
       await startupHeartbeatRecovery;
     }
 
+    if (config.heartbeatSchedulerEnabled) await heartbeat.registerIssueStateRecoveryWork(deliveryWork,
+      () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready;
     if (config.heartbeatSchedulerEnabled) await heartbeat.registerRunDispatchWork(deliveryWork,
       () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready;
     if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork,
@@ -1957,33 +1918,6 @@ async function startServerWithDatabaseTeardown(
             .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true })
             .then(async () => {
               await heartbeat.resumeQueuedRuns();
-              const reconciled = await heartbeat.reconcileStrandedAssignedIssues({ skipExternalConversationSettlement: true });
-              if (
-                reconciled.assignmentDispatched > 0 ||
-                reconciled.dispatchRequeued > 0 ||
-                reconciled.continuationRequeued > 0 ||
-                reconciled.successfulRunHandoffEscalated > 0 ||
-                reconciled.successfulRunHandoffRetried > 0 ||
-                reconciled.escalated > 0
-              ) {
-                logger.warn(
-                  { ...reconciled },
-                  "periodic heartbeat recovery changed assigned issue state",
-                );
-              }
-            })
-            .then(async () => {
-              const reconciled = await heartbeat.reconcileResolvedDependencyWakes();
-              if (reconciled.healed > 0) {
-                logger.warn({ ...reconciled }, "periodic dependency-wake reconciliation restored task execution paths");
-              }
-            })
-            .then(async () => {
-              const reconciled = await heartbeat.reconcileTaskWatchdogs();
-              reportIncompleteWatchdogScans(reconciled, "periodic");
-              if (reconciled.triggered > 0) {
-                logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work");
-              }
             })
             .then(async () => {
               const scanned = await heartbeat.scanSilentActiveRuns();
@@ -1991,12 +1925,6 @@ async function startServerWithDatabaseTeardown(
                 logger.warn({ ...scanned }, "periodic active-run output watchdog created review work");
               }
             })
-            .then(async () => {
-              const swept = await heartbeat.sweepStaleIssueLocks();
-              if (swept.cleared > 0) {
-                logger.warn({ ...swept }, "periodic stale-lock sweeper cleared issue locks");
-              }
-            })
             .catch((err) => {
               logger.error({ err }, "periodic heartbeat recovery failed");
             }));

server/src/services/agent-conversations.ts

@@ -1,4 +1,5 @@
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared";
@@ -547,6 +548,7 @@ export async function resumeConversationForReset(db: Db, comment: typeof issueCo
       eq(issues.id, comment.issueId), eq(issues.companyId, comment.companyId),
     )).for("update");
     if (!isConversation(issue) || comment.conversationSessionGeneration != null) return;
+    await notifyIssueRecovery(tx, { companyId: issue.companyId });
     const released = await tx.update(issueTreeHolds).set({
       status: "released", releasedAt: new Date(), updatedAt: new Date(),
       releasedByActorType: "user", releasedByUserId: comment.authorUserId,

server/src/services/chat-channels/telegram/attachment-failure.ts

@@ -1,4 +1,5 @@
 import { notifyStatusCardIssueChange } from "../../status-card-notifications.js";
+import { notifyIssueRecovery } from "../../issue-recovery-notifications.js";
 import { createProviderEffects, type ProviderEffectTarget } from "../provider-effects.js";
 
 import type { EndpointRow, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js";
@@ -79,6 +80,7 @@ export function createTelegramAttachmentFailure(db: Db, options: TelegramAttachm
           })(),
       );
     const effect = await db.transaction(async (tx) => {
+      await notifyIssueRecovery(tx, { companyId: input.endpoint.companyId });
       // Replace the optimistic attachment placeholder with an explicit safe
       // Paperclip-visible failure. The message link stays authoritative for
       // exact provider redelivery, while no agent wakeup is enqueued for an

server/src/services/company-deletion.ts

@@ -1,5 +1,6 @@
 import { notifyStatusCardEvaluation } from "./status-card-notifications.js";
 import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
 import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
@@ -74,6 +75,7 @@ export async function deleteCompany(db: Db, id: string) {
           .where(eq(companies.id, id)).for("no key update");
         if (!existing) return null;
         await notifyExternalObjectRefresh(tx, id);
+        await notifyIssueRecovery(tx, { companyId: id });
         await notifyChatCompanyConfiguration(tx, id);
         await chatCompanyDeletion.deleteCompanyData(tx, id);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);

server/src/services/company-portability.ts

@@ -1,4 +1,5 @@
 import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { createAgentLifecycle } from "./agent-lifecycle.js";
 import { agentAppearanceSchema, agentHarnessType, paperclipRunnerProfileForHarness, agentRunner, type AgentRunnerChoice } from "@paperclipai/shared";
 import { resolveNewAgentRunnerForCompany } from "./agent-runner-selection.js";
@@ -6468,7 +6469,8 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
             const relationCompanyId = targetCompany.id;
             await db.transaction(async tx => {
               await notifyAttentionRetentionWork(tx, { companyId: relationCompanyId });
-            return tx
+              await notifyIssueRecovery(tx, { companyId: relationCompanyId });
+              await tx
               .insert(issueRelations)
               .values(relationRows.map((row) => ({
                 companyId: relationCompanyId,

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",
   statusCardEvaluation: "status-card-evaluation",

server/src/services/documents.ts

@@ -1,4 +1,5 @@
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, asc, desc, eq, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -277,6 +278,7 @@ export function documentService(db: Db) {
                   .where(eq(issueDocuments.issueId, issue.id));
                 const fallbackKey = nextAvailableDocumentKey(key, issueDocumentKeys.map((row) => row.key));
 
+                await notifyIssueRecovery(tx, { companyId: issue.companyId });
                 const [document] = await tx
                   .insert(documents)
                   .values({
@@ -378,6 +380,7 @@ export function documentService(db: Db) {
               });
             }
 
+            await notifyIssueRecovery(tx, { companyId: issue.companyId });
             const nextRevisionNumber = existing.latestRevisionNumber + 1;
             const [revision] = await tx
               .insert(documentRevisions)
@@ -440,6 +443,7 @@ export function documentService(db: Db) {
             throw conflict("Document does not exist yet", { key });
           }
 
+          await notifyIssueRecovery(tx, { companyId: issue.companyId });
           const [document] = await tx
             .insert(documents)
             .values({
@@ -594,7 +598,10 @@ export function documentService(db: Db) {
         });
       }
       await db.transaction(async (tx) => {
-        for (const companyId of new Set(rows.map(row => row.companyId))) await notifyStatusCardIssueChange(tx, companyId);
+        for (const companyId of new Set(rows.map(row => row.companyId))) {
+          await notifyStatusCardIssueChange(tx, companyId);
+          await notifyIssueRecovery(tx, { companyId });
+        }
         await insertRowsInChunks(tx, documents, documentRows);
         await insertRowsInChunks(tx, documentRevisions, revisionRows);
         await insertRowsInChunks(tx, issueDocuments, linkRows);
@@ -647,6 +654,7 @@ export function documentService(db: Db) {
           });
         }
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         const now = new Date();
         const nextRevisionNumber = existing.latestRevisionNumber + 1;
         const [restoredRevision] = await tx
@@ -726,6 +734,7 @@ export function documentService(db: Db) {
           };
         }
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         const now = new Date();
         await tx
           .update(documents)
@@ -773,6 +782,7 @@ export function documentService(db: Db) {
           };
         }
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         const now = new Date();
         await tx
           .update(documents)
@@ -821,6 +831,7 @@ export function documentService(db: Db) {
           });
         }
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         await tx.delete(issueDocuments).where(eq(issueDocuments.documentId, existing.id));
         await notifyStatusCardIssueChange(tx, existing.companyId);
         await tx.delete(documents).where(eq(documents.id, existing.id));

server/src/services/external-conversation-settlement-notifications.ts

@@ -1,7 +1,9 @@
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import type { ExternalConversationSettlementScope } from "./slack-conversation-lifecycle.js";
 
-export function notifyExternalConversationSettlement(tx: object, scope?: ExternalConversationSettlementScope) {
+export async function notifyExternalConversationSettlement(tx: object, scope?: ExternalConversationSettlementScope) {
+  await notifyIssueRecovery(tx, scope);
   return notifyDeliveryWork(tx, DELIVERY_QUEUES.externalConversationSettlement,
     scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*");
 }

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/issue-comment-writes.ts

@@ -1,4 +1,5 @@
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
+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";
@@ -8,6 +9,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 notifyStatusCardIssueChange(tx, input.companyId);
+    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-actions.ts

@@ -1,4 +1,5 @@
 import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { writeExecutionControl } from "./execution-control-notifications.js";
@@ -324,6 +325,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
     try {
       const created = await db.transaction(async (tx) => {
         await notifyAttentionRetentionWork(tx, { companyId: input.companyId });
+        await notifyIssueRecovery(tx, { companyId: input.companyId });
         await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
@@ -415,6 +417,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           Math.min(nextAttemptCount, effectiveMaxAttempts),
         );
         const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
+          await notifyIssueRecovery(tx, { companyId: input.companyId });
           await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
           if ((!input.preserveExistingOwner && existing.cause !== input.cause)
             || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) {
@@ -476,6 +479,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
         return toReadModel(exhausted);
       }
       const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
+        await notifyIssueRecovery(tx, { companyId: input.companyId });
         await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
         if ((!input.preserveExistingOwner && existing.cause !== input.cause)
           || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) {
@@ -550,6 +554,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
 
     try {
       const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
+        await notifyIssueRecovery(tx, { companyId: input.companyId });
         await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
         await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
         return tx
@@ -597,6 +602,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
       const pending = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions)
         .where(and(...predicates)).limit(1);
       if (!pending.length) return [];
+      await notifyIssueRecovery(tx, { companyId: input.companyId });
       await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
       await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
       return tx

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-tree-control.ts

@@ -1,5 +1,6 @@
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
@@ -760,6 +761,7 @@ export function issueTreeControlService(db: Db) {
       const releaseReason = input.reason ?? "Subtree resume applied.";
 
       const { hold: resumeHold } = await db.transaction(async (tx) => {
+        await notifyIssueRecovery(tx, { companyId });
         const [createdHold] = await tx
           .insert(issueTreeHolds)
           .values({
@@ -838,6 +840,7 @@ export function issueTreeControlService(db: Db) {
     }
 
     const { hold, members } = await db.transaction(async (tx) => {
+      await notifyIssueRecovery(tx, { companyId });
       const [createdHold] = await tx
         .insert(issueTreeHolds)
         .values({
@@ -1181,7 +1184,9 @@ export function issueTreeControlService(db: Db) {
       throw conflict("Issue tree hold is already released");
     }
 
-    const [updated] = await db
+    const [updated] = await db.transaction(async tx => {
+      await notifyIssueRecovery(tx, { companyId });
+      return tx
       .update(issueTreeHolds)
       .set({
         status: "released",
@@ -1199,6 +1204,7 @@ export function issueTreeControlService(db: Db) {
       })
       .where(and(eq(issueTreeHolds.id, holdId), eq(issueTreeHolds.companyId, companyId)))
       .returning();
+    });
 
     const members = await db
       .select()

server/src/services/issues.ts

@@ -1,3 +1,4 @@
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
@@ -7634,6 +7635,7 @@ export function issueService(db: Db) {
     }
 
     await notifyAttentionRetentionWork(dbOrTx, { companyId });
+    await notifyIssueRecovery(dbOrTx, { companyId });
     await dbOrTx
       .delete(issueRelations)
       .where(
@@ -11521,7 +11523,9 @@ export function issueService(db: Db) {
           await notifyMergeConfirmationWork(tx, { companyId: receiptExisting.companyId, issueId: id });
         }
         if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined
-          || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) {
+          || 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);
           await notifyRoutineExecutionStatus(tx, id);
@@ -13852,6 +13856,7 @@ export function issueService(db: Db) {
           .then((rows) => rows[0] ?? null);
         if (!existing) return null;
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         await tx.delete(issueAttachments).where(eq(issueAttachments.id, id));
         await tx.delete(assets).where(eq(assets.id, existing.assetId));
         return existing;

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/chat-attachment-reuse.test.ts

@@ -73,7 +73,7 @@ describe("native same-conversation chat attachment reuse", () => {
   beforeAll(async () => {
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_CHAT_REUSE_TEST_DATABASE_URL");
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-chat-reuse-");
-    db = createDb(external ?? temporary!.connectionString);
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 });
     const storageRoot = await mkdtemp(
       path.join(tmpdir(), "paperclip-chat-reuse-"),
     );

server/src/services/native-runtime/chat-attachment-reuse.ts

@@ -1,3 +1,4 @@
+import { notifyIssueRecovery } from "../issue-recovery-notifications.js";
 import { createHash } from "node:crypto";
 
 import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm";
@@ -1582,7 +1583,9 @@ export async function prepareReusedChatAttachment(input: {
     if (!workProduct) {
       throw new Error("paperclip_runner_chat_attachment_origin_not_persisted");
     }
-    await input.db
+    await input.db.transaction(async tx => {
+      await notifyIssueRecovery(tx, { companyId: input.binding.companyId });
+      await tx
       .update(issueWorkProducts)
       .set({
         title: input.title,
@@ -1602,6 +1605,7 @@ export async function prepareReusedChatAttachment(input: {
           eq(issueWorkProducts.createdByRunId, input.binding.runId),
         ),
       );
+    });
     const comment = await issueService(input.db).addComment(
       input.binding.issueId,
       `Prepared ${input.title} for this response.`,

server/src/services/native-runtime/native-runner-file-handoff.test.ts

@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js";
 import { nativeCompletionFeedback } from "./native-completion-feedback.js";
 import { createHash, randomUUID } from "node:crypto";
@@ -61,10 +62,9 @@ describe("native runner file handoff", () => {
   const runId = "00000000-0000-4000-8000-000000009104";
 
   beforeAll(async () => {
-    temporary = await startEmbeddedPostgresTestDatabase(
-      "native-runner-file-handoff-",
-    );
-    db = createDb(temporary.connectionString);
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_HANDOFF_TEST_DATABASE_URL");
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-runner-file-handoff-");
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 });
     temporaryRoot = await mkdtemp(
       path.join(tmpdir(), "paperclip-native-file-handoff-"),
     );

server/src/services/native-runtime/native-runner-file-handoff.ts

@@ -1,3 +1,4 @@
+import { notifyIssueRecovery } from "../issue-recovery-notifications.js";
 import { createHash, randomUUID } from "node:crypto";
 import { execFile } from "node:child_process";
 import { constants } from "node:fs";
@@ -1265,7 +1266,9 @@ export async function prepareNativeRunnerFileHandoff(input: {
     ) {
       throw new Error("paperclip_runner_file_handoff_origin_not_persisted");
     }
-    await input.db
+    await input.db.transaction(async tx => {
+      await notifyIssueRecovery(tx, { companyId: input.binding.companyId });
+      await tx
       .update(issueWorkProducts)
       .set({ title: verified.title, updatedAt: new Date() })
       .where(
@@ -1276,6 +1279,7 @@ export async function prepareNativeRunnerFileHandoff(input: {
           eq(issueWorkProducts.createdByRunId, input.binding.runId),
         ),
       );
+    });
     const comment = await issueService(input.db).addComment(
       input.binding.issueId,
       `Prepared ${verified.title} for this response.`,

server/src/services/native-status-reassessment-notifications.ts

@@ -1,5 +1,6 @@
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string };
@@ -7,6 +8,7 @@ export type NativeStatusReassessmentScope = { companyId: string; issueId?: strin
 export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
   await notifyStatusCardIssueChange(tx, scope.companyId);
   await notifyAttentionRetentionWork(tx, { companyId: scope.companyId });
+  await notifyIssueRecovery(tx, scope);
   return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment,
     JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId }));
 }

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;
 };
 
@@ -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,8 +1,10 @@
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 export type RunDispatchScope = { companyId: string };
 
 export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) {
+  await notifyIssueRecovery(transaction, scope);
   const hint = scope ? JSON.stringify(scope) : "*";
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);

server/src/services/system-recovery.ts

@@ -1,3 +1,4 @@
+import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { and, eq, sql } from "drizzle-orm";
 import { heartbeatRuns, issueComments, issueRecoveryActions, type Db } from "@paperclipai/db";
 import { writeExecutionControl } from "./execution-control-notifications.js";
@@ -11,6 +12,7 @@ export async function adoptSystemRecovery(db: Db, source: WorkspaceRecoverySourc
       .where(and(eq(issueRecoveryActions.companyId, source.companyId), eq(issueRecoveryActions.id, source.actionId),
         eq(issueRecoveryActions.sourceIssueId, source.issueId))).for("update");
     if (!existing || existing.evidence.systemRecovery) return;
+    await notifyIssueRecovery(tx, { companyId: source.companyId });
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
     const state = createSystemRecovery(source.runId);
     // Atomic, once per source. An exhausted previous execution budget is history;

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 ${entry.id} in company ${entry.companyId} has an incomplete subtree`);
+      return next === null ? null : { ...entry, dueAt: next };
+    },
+  }, canRun);
+}

server/src/services/task-watchdogs.ts

@@ -1,4 +1,5 @@
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
+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";
@@ -832,7 +833,7 @@ async function updateIssueWatchdogRow(
   return updated;
 }
 
-export async function upsertIssueWatchdogForIssue(
+async function upsertIssueWatchdogInTransaction(
   dbOrTx: any,
   companyId: string,
   issueId: string,
@@ -848,12 +849,13 @@ export async function upsertIssueWatchdogForIssue(
     .where(and(eq(issueWatchdogs.companyId, companyId), eq(issueWatchdogs.issueId, issueId)))
     .then((rows: IssueWatchdogRow[]) => rows[0] ?? null);
 
+  await notifyIssueRecovery(dbOrTx, { companyId });
   if (existing) {
     const updated = await updateIssueWatchdogRow(dbOrTx, existing, input, now);
     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,
@@ -871,7 +873,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
@@ -886,6 +888,15 @@ 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(tx => upsertIssueWatchdogInTransaction(tx, companyId, issueId, input));
+}
+
 export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) {
   const issuesSvc = issueService(db);
 
@@ -1482,7 +1493,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()
@@ -1495,6 +1506,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 };
     }
@@ -1614,13 +1632,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,
       ));
   }
 
@@ -1750,14 +1769,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",
@@ -1769,14 +1790,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(
@@ -1806,7 +1830,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);