Drive issue recovery from scoped source writes and saved deadlines

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

@@ -134,6 +134,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() })),
@@ -927,6 +928,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];
@@ -1334,11 +1336,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

@@ -109,7 +109,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,
@@ -973,9 +972,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;
@@ -1796,47 +1792,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;
@@ -1845,6 +1804,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,
@@ -2051,33 +2012,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();
@@ -2085,12 +2019,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,3 +1,4 @@
+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";
@@ -544,6 +545,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,3 +1,4 @@
+import { notifyIssueRecovery } from "../../issue-recovery-notifications.js";
 import { createProviderEffects, type ProviderEffectTarget } from "../provider-effects.js";
 
 import type { EndpointRow, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js";
@@ -78,6 +79,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,3 +1,4 @@
+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";
@@ -71,6 +72,7 @@ export async function deleteCompany(db: Db, id: string) {
         const [existing] = await tx.select({ id: companies.id }).from(companies)
           .where(eq(companies.id, id)).for("no key update");
         if (!existing) return null;
+        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,3 +1,4 @@
+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";
@@ -6465,7 +6466,9 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
           }
           if (relationRows.length > 0) {
             const relationCompanyId = targetCompany.id;
-            await db
+            await db.transaction(async tx => {
+              await notifyIssueRecovery(tx, { companyId: relationCompanyId });
+              await tx
               .insert(issueRelations)
               .values(relationRows.map((row) => ({
                 companyId: relationCompanyId,
@@ -6476,6 +6479,7 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
                 createdByUserId: actorUserId ?? null,
               })))
               .onConflictDoNothing();
+            });
           }
         }
 

server/src/services/documents.ts

@@ -1,3 +1,4 @@
+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";
@@ -275,6 +276,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({
@@ -376,6 +378,7 @@ export function documentService(db: Db) {
               });
             }
 
+            await notifyIssueRecovery(tx, { companyId: issue.companyId });
             const nextRevisionNumber = existing.latestRevisionNumber + 1;
             const [revision] = await tx
               .insert(documentRevisions)
@@ -438,6 +441,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({
@@ -592,6 +596,9 @@ export function documentService(db: Db) {
         });
       }
       await db.transaction(async (tx) => {
+        for (const companyId of new Set(rows.map(row => row.companyId))) {
+          await notifyIssueRecovery(tx, { companyId });
+        }
         await insertRowsInChunks(tx, documents, documentRows);
         await insertRowsInChunks(tx, documentRevisions, revisionRows);
         await insertRowsInChunks(tx, issueDocuments, linkRows);
@@ -644,6 +651,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
@@ -722,6 +730,7 @@ export function documentService(db: Db) {
           };
         }
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         const now = new Date();
         await tx
           .update(documents)
@@ -769,6 +778,7 @@ export function documentService(db: Db) {
           };
         }
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         const now = new Date();
         await tx
           .update(documents)
@@ -817,6 +827,7 @@ export function documentService(db: Db) {
           });
         }
 
+        await notifyIssueRecovery(tx, { companyId: existing.companyId });
         await tx.delete(issueDocuments).where(eq(issueDocuments.documentId, existing.id));
         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/issue-recovery-actions.ts

@@ -1,3 +1,4 @@
+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";
@@ -322,6 +323,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
   ): Promise<IssueRecoveryAction> {
     try {
       const created = await db.transaction(async (tx) => {
+        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);
@@ -413,6 +415,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)) {
@@ -474,6 +477,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)) {
@@ -548,6 +552,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
@@ -595,6 +600,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-tree-control.ts

@@ -1,3 +1,4 @@
+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";
@@ -758,6 +759,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({
@@ -836,6 +838,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({
@@ -1177,7 +1180,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",
@@ -1195,6 +1200,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 { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
@@ -7627,6 +7628,7 @@ export function issueService(db: Db) {
       await assertNoBlockingCycles(companyId, issueId, deduped, dbOrTx);
     }
 
+    await notifyIssueRecovery(dbOrTx, { companyId });
     await dbOrTx
       .delete(issueRelations)
       .where(
@@ -13799,6 +13801,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/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.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,8 +1,10 @@
+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 };
 
-export function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
+export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
+  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/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

@@ -29,7 +29,7 @@ export function registerTaskWatchdogWork(
       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");
+      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

@@ -848,6 +848,7 @@ async function upsertIssueWatchdogInTransaction(
     .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 };
@@ -892,10 +893,7 @@ export function upsertIssueWatchdogForIssue(
   issueId: string,
   input: IssueWatchdogUpsertInput,
 ): Promise<{ watchdog: IssueWatchdog; created: boolean }> {
-  return dbOrTx.transaction(async tx => {
-    await notifyIssueRecovery(tx, { companyId });
-    return upsertIssueWatchdogInTransaction(tx, companyId, issueId, input);
-  });
+  return dbOrTx.transaction(tx => upsertIssueWatchdogInTransaction(tx, companyId, issueId, input));
 }
 
 export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) {