fix(server): bound and scope routine projection recovery

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

doc/plans/2026-10-30-routine-status-sleep.md

@@ -7,14 +7,16 @@ scanned every five minutes, even with no routines or only completed history.
 ## Ownership and trigger
 
 `routine-run-status.ts` owns the projection, custody checks and repair transaction.
-The worker loads actual mismatches at startup. It accepts issue hints from issue
+The worker loads actual mismatches in keyset pages of 50 at startup. It accepts issue hints from issue
 creation, status/origin changes, checkout/release, routine issue association,
 execution control writes and review state writes. It uses the existing
 transaction-aware work notification protocol. An outer transaction owns the hint;
 a savepoint release does not wake the repair worker before commit.
 
-A standard execution must name the same routine, invocation, company and linked
-task. Plugin-managed executions may use their declared custom origin. Repair
+An execution must retain reciprocal task/invocation links in the same company
+and belong to a routine in that company. This is the existing routine status
+owner contract. Origin labels can change when a plugin declaration changes or
+is removed; retained reciprocal links preserve historical execution custody. Repair
 locks the routine and task and rechecks the linked invocation before writing.
 It preserves skipped/coalesced history and unrelated dispatch failures. A task
 that resumes clears an execution-task transient failure through the same owner
@@ -33,8 +35,9 @@ history leave no recurring worker SQL. The existing Cloud and heartbeat schedule
 enablement gates remain in place. Ordinary routine scheduling is unchanged.
 
 Supported app writers register `notifyRoutineExecutionStatus(tx, issueId)` before
-their mutation. Common execution writers without an issue hint request debt
-reconstruction. Backups and independent SQL import/repair tools must restore
+their mutation. Common execution writers without an issue hint request debt reconstruction only
+for their recorded company. Projection debt settles even when that company is
+paused; company activity does not create or cancel this obligation. Backups and independent SQL import/repair tools must restore
 before startup or call that same transaction contract; process-local hints do
 not observe independent SQL or another process. No permanent listener or periodic
 configured-routine scan is added.

server/src/__tests__/execution-control-work.test.ts

@@ -383,11 +383,11 @@ describe.skipIf(!support.supported)("execution-control durable work", () => {
     const due = new Date(Date.now() + 120_000), earlier = new Date(Date.now() + 60_000);
     for (const at of [due, earlier]) {
       await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns)
-        .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)));
+        .set({ executionControlDeadlineAt: at }).where(eq(heartbeatRuns.id, run.id)), { companyId });
       await settle(); expect(c.nextWakeAt()).toBe(at.getTime());
     }
     await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx.update(heartbeatRuns)
-      .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)));
+      .set({ executionControlDeadlineAt: null }).where(eq(heartbeatRuns.id, run.id)), { companyId });
     await quiet(c);
   });
   it("delivers a reconciled continuation when marking and resolving commit together", async () => {

server/src/__tests__/heartbeat-workspace-branch-containment.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDatabaseWork } from "@paperclipai/db";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
 import { execFile } from "node:child_process";
 import { createHash, randomUUID } from "node:crypto";
@@ -888,7 +890,7 @@ describeEmbeddedPostgres("heartbeat workspace branch containment", () => {
   const tempRoots: string[] = [];
 
   beforeAll(async () => {
-    if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl);
+    if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl, { maxConnections: 6 });
     else {
       tempDb = await startEmbeddedPostgresTestDatabase("paperclip-branch-containment-");
       db = createDb(tempDb.connectionString);
@@ -1173,15 +1175,15 @@ describeEmbeddedPostgres("heartbeat workspace branch containment", () => {
       },
     });
 
+    const routineEvents: string[] = [];
+    const unsubscribeRoutine = subscribeDatabaseWork(db, DELIVERY_QUEUES.routineExecutionStatus, event => routineEvents.push(event));
     const heartbeat = heartbeatService(db);
-    const run = await heartbeat.wakeup(agentId, {
-      source: "assignment",
-      triggerDetail: "system",
-      reason: "issue_assigned",
-      payload: { issueId },
-      contextSnapshot: { issueId, wakeReason: "issue_assigned" },
-    });
-
+    let run;
+    try {
+      run = await heartbeat.wakeup(agentId, { source: "assignment", triggerDetail: "system", reason: "issue_assigned",
+        payload: { issueId }, contextSnapshot: { issueId, wakeReason: "issue_assigned" } });
+    } finally { unsubscribeRoutine(); }
+    expect(routineEvents).toContain("intent"); expect(routineEvents).toContain("settled");
     expect(run).toBeNull();
     expect(adapterExecute).not.toHaveBeenCalled();
 

server/src/__tests__/routine-execution-status-work.test.ts

@@ -1,3 +1,5 @@
+import { writeExecutionControl } from "../services/execution-control-notifications.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import { eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -122,6 +124,56 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await settled(); await quiet();
     expect(await saved(run.id)).toEqual(before); expect((await saved(otherRun.id)).status).toBe("issue_created"); expect(errors).toEqual([]);
   });
+  it.each(["changed", "removed"])("keeps historical custom-origin custody after its plugin declaration is %s", async mode => {
+    const f = await fixture(), pluginKey = `routine-history-${randomUUID()}`, originId = `operation:${randomUUID()}`;
+    const [plugin] = await db.insert(plugins).values({ pluginKey, packageName: pluginKey, version: "0.1.0",
+      manifestJson: { id: pluginKey, apiVersion: 1, version: "0.1.0", displayName: "Routine test", description: "Routine test",
+        author: "Paperclip", categories: ["automation"], capabilities: ["routines.managed"], entrypoints: { worker: "./worker.js" } } }).returning();
+    const [resource] = await db.insert(pluginManagedResources).values({ companyId: f.companyId, pluginId: plugin.id, pluginKey,
+      resourceKind: "routine", resourceKey: "history", resourceId: f.routine.id, defaultsJson: { issueTemplate: { originId } } }).returning();
+    const run = await dispatch(f);
+    expect((await db.select().from(issues).where(eq(issues.id, run.linkedIssueId!)))[0].originId).toBe(originId);
+    if (mode === "changed") await db.update(pluginManagedResources).set({ defaultsJson: { issueTemplate: { originId: "new-origin" } } }).where(eq(pluginManagedResources.id, resource.id));
+    else await db.delete(pluginManagedResources).where(eq(pluginManagedResources.id, resource.id));
+    await db.update(issues).set({ status: "done" }).where(eq(issues.id, run.linkedIssueId!));
+    await start(); await projected(run.id, "completed"); await quiet(); expect(errors).toEqual([]);
+  });
+  it("reconstructs more than one page and does not lose a committed issue hint behind the startup cursor", async () => {
+    const f = await fixture(), ids = Array.from({ length: 53 }, () => randomUUID()).sort();
+    const issueIds = ids.map(() => randomUUID());
+    await db.insert(issues).values(ids.map((id, index) => ({ id: issueIds[index], companyId: f.companyId, title: "Legacy execution",
+      status: index === 0 ? "todo" : "done", originKind: "routine_execution", originRunId: id, originId: f.routine.id })));
+    await db.insert(routineRuns).values(ids.map((id, index) => ({ id, companyId: f.companyId, routineId: f.routine.id,
+      source: "schedule", linkedIssueId: issueIds[index], status: "issue_created" })));
+    const service = routineExecutionStatusService(db), load = service.load;
+    let entered!: () => void, release!: () => void;
+    const enteredGate = new Promise<void>(resolve => { entered = resolve; });
+    const releaseGate = new Promise<void>(resolve => { release = resolve; });
+    const loaded = vi.spyOn(service, "load").mockImplementationOnce(async (scope, afterId) => {
+      const page = await load(scope, afterId); expect(page).toHaveLength(50); entered(); await releaseGate; return page;
+    });
+    const starting = start(service); await enteredGate;
+    await issueService(db).update(issueIds[0]!, { status: "done" }); release(); await starting;
+    await vi.advanceTimersByTimeAsync(3000);
+    await vi.waitFor(async () => expect((await db.select().from(routineRuns)).every(row => row.status === "completed")).toBe(true));
+    expect(loaded.mock.calls.some(([scope, afterId]) => scope === undefined && afterId !== undefined)).toBe(true);
+    expect(loaded.mock.calls.some(([scope]) => scope?.issueId === issueIds[0])).toBe(true);
+    await quiet(); expect(errors).toEqual([]);
+  });
+  it("scopes shared execution writes to their company and settles paused-company projection debt", async () => {
+    const f = await fixture(), other = await fixture(), run = await dispatch(f), otherRun = await dispatch(other);
+    const service = routineExecutionStatusService(db), load = vi.spyOn(service, "load");
+    await start(service); await quiet(); load.mockClear();
+    await db.update(companies).set({ status: "paused" }).where(eq(companies.id, f.companyId));
+    await db.update(issues).set({ status: "done" }).where(eq(issues.id, otherRun.linkedIssueId!));
+    await writeExecutionControl(db, DELIVERY_QUEUES.executionStatus,
+      tx => tx.update(issues).set({ status: "done" }).where(eq(issues.id, run.linkedIssueId!)), { companyId: f.companyId });
+    await projected(run.id, "completed");
+    expect(load.mock.calls.every(([scope]) => scope?.companyId === f.companyId)).toBe(true);
+    expect((await saved(otherRun.id)).status).toBe("issue_created"); await quiet();
+    await db.transaction(async tx => { await notifyRoutineExecutionStatus(tx, otherRun.linkedIssueId!); });
+    await projected(otherRun.id, "completed"); await quiet(); expect(errors).toEqual([]);
+  });
   it("waits for the real outer transaction and forgets a rolled-back nested status write", async () => {
     const f = await fixture(), run = await dispatch(f); await start(); await quiet();
     let release!: () => void, entered!: () => void;

server/src/__tests__/routine-execution-status-worker.test.ts

@@ -1,3 +1,4 @@
+import type { RoutineExecutionStatusScope } from "../services/routine-execution-status-notifications.js";
 import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
 import { afterEach, beforeEach, expect, it, vi } from "vitest";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
@@ -18,19 +19,19 @@ afterEach(async () => { await coordinator.stop(); vi.useRealTimers(); });
 it("loads actual debt once and leaves no recurring owner work", async () => {
   const service = { load: vi.fn(async () => []), settle: vi.fn(async () => true) };
   await registerRoutineExecutionStatusWork(coordinator, service, () => canRun).ready;
-  expect(service.load).toHaveBeenCalledExactlyOnceWith(undefined);
+  expect(service.load).toHaveBeenCalledExactlyOnceWith(undefined, undefined);
   expect(coordinator.nextWakeAt()).toBeNull();
   await vi.advanceTimersByTimeAsync(20 * 60_000);
   expect(service.load).toHaveBeenCalledOnce(); expect(service.settle).not.toHaveBeenCalled(); expect(errors).toEqual([]);
 });
 
 it("reconstructs only an invalidated issue and coalesces duplicate requests", async () => {
-  const service = { load: vi.fn(async (_issueId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) };
+  const service = { load: vi.fn(async (_scope?: RoutineExecutionStatusScope, _afterId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) };
   const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready;
   service.load.mockResolvedValueOnce([entry]);
-  worker.invalidate("issue"); worker.invalidate("issue");
+  worker.invalidate(JSON.stringify({ issueId: "issue" })); worker.invalidate(JSON.stringify({ issueId: "issue" }));
   await vi.advanceTimersByTimeAsync(1000);
-  expect(service.load).toHaveBeenLastCalledWith("issue"); expect(service.load).toHaveBeenCalledTimes(2);
+  expect(service.load).toHaveBeenLastCalledWith({ issueId: "issue" }, undefined); expect(service.load).toHaveBeenCalledTimes(2);
   expect(service.settle).toHaveBeenCalledExactlyOnceWith(expect.objectContaining(entry));
   expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]);
 });
@@ -50,7 +51,7 @@ it("retains startup and first work during drain without passive owner calls", as
   canRun = false;
   const service = { load: vi.fn(async () => [entry]), settle: vi.fn(async () => true) };
   const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready;
-  worker.invalidate("issue");
+  worker.invalidate(JSON.stringify({ issueId: "issue" }));
   await vi.advanceTimersByTimeAsync(20 * 60_000);
   expect(service.load).not.toHaveBeenCalled(); expect(service.settle).not.toHaveBeenCalled();
   canRun = true; await vi.advanceTimersByTimeAsync(30_000);
@@ -69,11 +70,11 @@ it("keeps a failed owner across drain and retries after resume", async () => {
 it("retains a notification that arrives while the startup reconstruction is in flight", async () => {
   let release!: () => void;
   const gate = new Promise<void>(resolve => { release = resolve; });
-  const service = { load: vi.fn(async (_issueId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) };
+  const service = { load: vi.fn(async (_scope?: RoutineExecutionStatusScope, _afterId?: string) => [] as typeof entry[]), settle: vi.fn(async () => true) };
   service.load.mockImplementationOnce(async () => { await gate; return []; });
   const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun);
   await vi.waitFor(() => expect(service.load).toHaveBeenCalledOnce());
-  worker.invalidate("issue"); service.load.mockResolvedValueOnce([entry]); release(); await worker.ready;
+  worker.invalidate(JSON.stringify({ issueId: "issue" })); service.load.mockResolvedValueOnce([entry]); release(); await worker.ready;
   await vi.advanceTimersByTimeAsync(1000);
   expect(service.settle).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]);
 });
@@ -81,6 +82,26 @@ it("retains a notification that arrives while the startup reconstruction is in f
 it("cancels retries and subscriptions at shutdown", async () => {
   const service = { load: vi.fn(async () => [entry]), settle: vi.fn(async () => true).mockRejectedValue(new Error("offline")) };
   const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready;
-  await worker.stop(); worker.invalidate("issue"); await vi.advanceTimersByTimeAsync(20 * 60_000);
+  await worker.stop(); worker.invalidate(JSON.stringify({ issueId: "issue" })); await vi.advanceTimersByTimeAsync(20 * 60_000);
   expect(service.settle).toHaveBeenCalledOnce(); expect(coordinator.nextWakeAt()).toBeNull();
 });
+
+
+it("reconstructs a backlog in bounded keyset pages while retaining a scoped hint behind the cursor", async () => {
+  const firstPage = Array.from({ length: 50 }, (_, i) => ({ ...entry, id: `run-${String(i).padStart(3, "0")}` }));
+  const late = { ...entry, id: "run-late", issueId: "late-issue" };
+  const service = {
+    load: vi.fn(async (scope?: RoutineExecutionStatusScope, afterId?: string) => scope?.issueId ? [late]
+      : afterId ? [{ ...entry, id: "run-050" }] : firstPage),
+    settle: vi.fn(async () => true),
+  };
+  const worker = registerRoutineExecutionStatusWork(coordinator, service, () => canRun); await worker.ready;
+  expect(service.load).toHaveBeenCalledExactlyOnceWith(undefined, undefined);
+  worker.invalidate(JSON.stringify({ issueId: "late-issue" }));
+  await vi.advanceTimersByTimeAsync(3000);
+  expect(service.load).toHaveBeenCalledWith(undefined, "run-049");
+  expect(service.load).toHaveBeenCalledWith({ issueId: "late-issue" }, undefined);
+  expect(service.settle).toHaveBeenCalledTimes(52);
+  expect(service.settle).toHaveBeenCalledWith(expect.objectContaining(late));
+  expect(coordinator.nextWakeAt()).toBeNull(); expect(errors).toEqual([]);
+});

server/src/__tests__/routine-status-reconciliation.test.ts

@@ -38,11 +38,10 @@ if (!support.supported) console.warn(`Skipping routine status database tests: ${
       expect((await db.select().from(routineRuns))[0]).toMatchObject({ status: "issue_created", completedAt: null });
     }
   });
-  it.each(["active_issue", "wrong_origin", "wrong_routine", "wrong_run", "wrong_company", "wrong_routine_company", "different_status"])("does not settle an unsupported handoff: %s", async mode => {
+  it.each(["active_issue", "wrong_origin", "wrong_run", "wrong_company", "wrong_routine_company", "different_status"])("does not settle an unsupported handoff: %s", async mode => {
     const f = await seed();
     if (mode === "active_issue") await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id));
     if (mode === "wrong_origin") await db.update(issues).set({ originKind: "manual" }).where(eq(issues.id, f.issue.id));
-    if (mode === "wrong_routine") await db.update(issues).set({ originId: randomUUID() }).where(eq(issues.id, f.issue.id));
     if (mode === "wrong_run") await db.update(issues).set({ originRunId: randomUUID() }).where(eq(issues.id, f.issue.id));
     if (mode === "wrong_company") { const other = await seed("done"); await db.update(routineRuns).set({ companyId: other.companyId }).where(eq(routineRuns.id, f.run.id)); }
     if (mode === "wrong_routine_company") { const other = await seed("done");
@@ -55,6 +54,20 @@ if (!support.supported) console.warn(`Skipping routine status database tests: ${
     expect((await db.select().from(routineRuns)).find(row => row.id === f.run.id)).toEqual(before.find(row => row.id === f.run.id));
     if (mode === "wrong_routine_company") await expect(routineService(db).syncRunStatusForIssue(f.issue.id)).resolves.toBeNull();
   });
+  it("uses reciprocal durable custody when a retained origin no longer names the current routine declaration", async () => {
+    const f = await seed("done");
+    await db.update(issues).set({ originId: "previous-declaration-origin" }).where(eq(issues.id, f.issue.id));
+    expect(await routineService(db).reconcileExecutionRunStatuses()).toEqual({ scanned: 1, settled: 1, deferred: 0 });
+    expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.run.id)))[0].status).toBe("completed");
+  });
+  it.each(["unknown-status", "other-code"])("preserves a failed invocation with an unrelated transient marker: %s", async mode => {
+    const f = await seed("in_progress");
+    await db.update(routineRuns).set({ status: "failed", failureReason: null,
+      triggerPayload: { transientFailure: { code: mode === "other-code" ? "unrelated_failure" : "execution_issue_status",
+        status: mode === "unknown-status" ? "unrecognized" : "blocked" } } }).where(eq(routineRuns.id, f.run.id));
+    expect(await routineService(db).reconcileExecutionRunStatuses()).toEqual({ scanned: 0, settled: 0, deferred: 0 });
+    expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.run.id)))[0].status).toBe("failed");
+  });
   it.each(["resume", "retarget", "routine_transfer"])("rechecks %s after candidate selection and before repair", async mode => {
     const f = await seed(); const transaction = db.transaction.bind(db);
     vi.spyOn(db, "transaction").mockImplementationOnce(async callback => {

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

@@ -1,3 +1,4 @@
+import { routineExecutionStatusService } from "../services/routine-run-status.js";
 import { nativeHarnessBackupExpirationService } from "../services/native-runtime/native-harness-backup-expiration.js";
 import { skillVersionRetentionService } from "../services/skill-version-retention.js";
 import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js";
@@ -392,6 +393,11 @@ vi.mock("../services/approval-retention.js", async importOriginal => ({
   approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })),
 }));
 
+vi.mock("../services/routine-run-status.js", async importOriginal => ({
+  ...await importOriginal<typeof import("../services/routine-run-status.js")>(),
+  routineExecutionStatusService: vi.fn(() => ({ load: vi.fn(async () => []), settle: vi.fn(async () => false) })),
+}));
+
 vi.mock("../services/index.js", () => ({
   backfillLegacyToolOAuthTokens: vi.fn(async () => ({
     scannedConnections: 0,
@@ -1366,6 +1372,27 @@ describe("startServer feedback export wiring", () => {
     }
   });
 
+  it.each([{ cloud: false, enabled: true }, { cloud: true, enabled: false }, { cloud: true, enabled: true }])(
+    "registers routine status only for enabled Cloud scheduling ($cloud/$enabled)", async ({ cloud, enabled }) => {
+      if (cloud) vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", "startup-test-token");
+      else vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", undefined);
+      loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+      const intervalCallbacks: Array<() => void> = [];
+      const intervals = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void) => {
+        intervalCallbacks.push(callback); return 1 as unknown as ReturnType<typeof setInterval>;
+      }) as typeof setInterval);
+      try {
+        await startServer(); await settleStartupHousekeeping();
+        expect(deliveryWorkRegisterMock.mock.calls.filter(([queue]) => queue === DELIVERY_QUEUES.routineExecutionStatus)).toHaveLength(cloud && enabled ? 1 : 0);
+        const service = vi.mocked(routineExecutionStatusService).mock.results.at(-1)?.value;
+        if (cloud && enabled) expect(service?.load).toHaveBeenCalledExactlyOnceWith(undefined, undefined);
+        for (let index = 0; index < 4; index++) for (const callback of intervalCallbacks) callback();
+        await settleStartupHousekeeping();
+        if (cloud && enabled) expect(service?.load).toHaveBeenCalledOnce();
+      } finally { intervals.mockRestore(); vi.unstubAllEnvs(); }
+    },
+  );
+
   it("keeps routine ticks and setup cleanup active when heartbeat scheduling is suppressed", async () => {
     loadConfigMock.mockReturnValue(buildTestConfig({
       heartbeatSchedulerEnabled: true,

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

@@ -2,8 +2,10 @@ import { notifyRoutineExecutionStatus } from "./routine-execution-status-notific
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js";
 
-export async function notifyExecutionRunWork(transaction: object, issueId?: string) {
-  await notifyRoutineExecutionStatus(transaction, issueId);
+export type ExecutionWorkScope = { companyId: string; agentId?: string; runId?: string; issueId?: string };
+
+export async function notifyExecutionRunWork(transaction: object, scope?: ExecutionWorkScope) {
+  if (scope) await notifyRoutineExecutionStatus(transaction, scope.issueId, scope.companyId);
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
     await notifyDeliveryWork(transaction, queue);
@@ -14,9 +16,10 @@ export function writeExecutionControl<T>(
   db: Db | Parameters<Parameters<Db["transaction"]>[0]>[0],
   queue: DeliveryQueue | readonly DeliveryQueue[],
   write: (tx: Db) => PromiseLike<T>,
+  scope: ExecutionWorkScope | undefined,
 ): Promise<T> {
   return db.transaction(async transaction => {
-    await notifyRoutineExecutionStatus(transaction);
+    if (scope) await notifyRoutineExecutionStatus(transaction, scope.issueId, scope.companyId);
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
     return write(transaction as unknown as Db);

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

@@ -196,7 +196,7 @@ export async function reconcileAbandonedExecutionControl(
           await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id });
           await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
           await notifyUnsafeWorkspaceExportRunChange(tx, run);
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           const [updatedRun] = await tx
             .update(heartbeatRuns)
             .set({

server/src/services/execution-recovery-resolution.ts

@@ -1,3 +1,4 @@
+import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { notifyChatResponseAvailability } from "./chat-response-availability.js";
@@ -546,7 +547,7 @@ export async function settleUnrecoverableExecutions(
             companyId: run.companyId, issueId: task.id, runId: run.id,
           }).limit(1).for("update");
           if (!hold) return;
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           const note = "Exact session recovery stopped. The recorded workspace recovery holds this execution without replaying its actions.";
           await tx.update(issueRecoveryActions).set({ status: "cancelled", outcome: "cancelled",
             resolvedAt: now, updatedAt: now, wakePolicy: null, monitorPolicy: null,
@@ -596,6 +597,7 @@ export async function settleUnrecoverableExecutions(
           if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id });
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+          await notifyRoutineExecutionStatus(tx, task.id, task.companyId);
           const [projected] = await tx
             .update(issues)
             .set({
@@ -654,7 +656,7 @@ export async function settleUnrecoverableExecutions(
             replay: "not_authorized",
           },
         });
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await tx
           .update(heartbeatRuns)
           .set({ executionStatusDeliveryId: randomUUID() })

server/src/services/explicit-native-continuation.ts

@@ -475,7 +475,7 @@ export async function admitExplicitNativeContinuation(input: {
   }).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.id, input.queuedCommentInterruptId),
     eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution")));
   if (cancelledStartupIds.size) await db.transaction(async tx => {
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId, agentId });
     for (const runId of cancelledStartupIds) {
       await tx.update(nativeRunFinalizations).set({
         phase: "terminal_failure", failureCode: "native_startup_cancelled", nextAttemptAt: null,

server/src/services/heartbeat.ts

@@ -3724,7 +3724,7 @@ export function heartbeatService(
           await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, tx => tx
             .update(heartbeatRuns)
             .set({ executionControlDeadlineAt: null })
-            .where(eq(heartbeatRuns.id, run.id)));
+            .where(eq(heartbeatRuns.id, run.id)), { companyId: run.companyId });
         }
         nativeOwnershipHeld =
           nativeOwnershipHeld ||

server/src/services/heartbeat/queue.ts

@@ -1,3 +1,4 @@
+import { notifyRoutineExecutionStatus } from "../routine-execution-status-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
@@ -3919,6 +3920,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
               await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
               await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
               if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id });
+              await notifyRoutineExecutionStatus(tx, issue.id, issue.companyId);
               await tx
                 .update(issues)
                 .set({

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

@@ -1044,8 +1044,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
       if (previous && isHeartbeatRunTerminalStatus(status))
         await notifyRetainedNativeSessionCleanup(tx, { companyId: previous.companyId, runId: previous.id });
       if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId);
-      await notifyExecutionRunWork(tx, previous?.nativeIssueId ?? previous?.issueId
-        ?? readNonEmptyString(parseObject(previous?.contextSnapshot).issueId) ?? undefined);
+      await notifyExecutionRunWork(tx, previous ? { companyId: previous.companyId, agentId: previous.agentId, runId: previous.id,
+        issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined } : undefined);
       if (previous) await notifyChatRunMilestones(tx, { companyId: previous.companyId, runId: previous.id,
         issueId: previous.nativeIssueId ?? previous.issueId ?? readNonEmptyString(parseObject(previous.contextSnapshot).issueId) ?? undefined });
       return write(tx);

server/src/services/heartbeat/runtime-execution.ts

@@ -966,7 +966,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
               eq(heartbeatRuns.id, run.id),
               eq(heartbeatRuns.status, "running"),
             ),
-          ));
+          ), { companyId: run.companyId });
         const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded";
         await recordWorkspaceFinalize(workspaceFinalizeStatus);
         if (adapterResult.nativeFinalization) {
@@ -1042,7 +1042,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe
           eq(heartbeatRuns.id, run.id),
           eq(heartbeatRuns.status, "running"),
         ),
-      ));
+      ), { companyId: run.companyId });
     if (
       issueRef &&
       context.resumeSessionGoalHeartbeat === true &&

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

@@ -392,7 +392,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], tx => tx.update(issueRecoveryActions).set({ nextAction: input.nextAction, updatedAt: now })
             .where(and(eq(issueRecoveryActions.id, existing.id),
               inArray(issueRecoveryActions.status, [...ACTIVE_RECOVERY_ACTION_STATUSES]),
-              sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning());
+              sql`${issueRecoveryActions.evidence}->'workspaceOwnerStop' is null`)).returning(), { companyId: input.companyId });
           if (updated) return toReadModel(updated);
         }
         return existing;
@@ -467,7 +467,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
               ),
             )
             .returning();
-        });
+        }, { companyId: input.companyId });
         if (!exhausted) {
           return retryUpsertSourceScoped(input, retryCount);
         }
@@ -539,7 +539,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
             ),
           )
           .returning();
-      });
+      }, { companyId: input.companyId });
       if (!updated) {
         return retryUpsertSourceScoped(input, retryCount);
       }
@@ -554,7 +554,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
           .insert(issueRecoveryActions)
           .values(buildInsertValues(input, ownerType, now))
           .returning();
-      });
+      }, { companyId: input.companyId });
       return toReadModel(created!);
     } catch (error) {
       if (!isUniqueRecoveryActionConflict(error)) throw error;
@@ -608,7 +608,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
         })
         .where(and(...predicates))
         .returning();
-    });
+    }, { companyId: input.companyId });
 
     return updated ? toReadModel(updated) : null;
   }

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

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

server/src/services/issues.ts

@@ -10607,7 +10607,7 @@ export function issueService(db: Db) {
         }
         if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
-        if (values.originKind === "routine_execution") await notifyRoutineExecutionStatus(tx, values.id);
+        if (values.originKind === "routine_execution") await notifyRoutineExecutionStatus(tx, values.id, companyId);
         const [issue] = await tx.insert(issues).values(values).returning();
         await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId });
         if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) {

server/src/services/legacy-execution-recovery.ts

@@ -135,7 +135,7 @@ export async function terminalizeLegacyExecution(input: {
     };
     if (issueId) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: current.companyId, issueId });
     await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id });
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: current.companyId });
     let [updated] = await tx
       .update(heartbeatRuns)
       .set({

server/src/services/native-runtime/native-failed-workspace-recovery.ts

@@ -65,7 +65,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
     let retained = !retainFiles || alreadyRetained ? run : await recordRequiredWorkspaceRestoreFailure(tx as unknown as Db, run, { workspaceRestoreFailure: "restore_failed" }, source);
     if (!retained || retainFiles && !hasRequiredWorkspaceRecovery(retained.resultJson)) return;
     if (options.allowPreterminal && coordinator.phase === "observed") {
-      await notifyExecutionRunWork(tx);
+      await notifyExecutionRunWork(tx, { companyId: run.companyId });
       const [settled] = await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure",
         leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, nextAttemptAt: null,
         recoveryState: "blocked", failureCode: coordinator.failureCode ?? "native_continuation_requires_reconciliation", updatedAt: new Date(),

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

@@ -899,7 +899,7 @@ export async function claimNativeRestartRecoveries(input: {
           recycledProviderPids: providerProcesses.recycledPids,
           stderrTail: row.run.stderrExcerpt,
         });
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: row.run.companyId });
         await tx
           .update(nativeRunFinalizations)
           .set({

server/src/services/native-runtime/native-run-finalizer.ts

@@ -401,7 +401,7 @@ async function recordRetryableFailure(input: {
     if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") {
       await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     }
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: input.run.companyId });
     await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     await tx
       .update(nativeRunFinalizations)
@@ -1659,7 +1659,7 @@ export async function finalizeNativeRun(input: {
         if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
         if (input.projectRunStatus) await notifyUnsafeWorkspaceExportRunChange(tx, run);
         if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
         const [after] = await projectRun(tx);
         return after ? { before, after } : null;

server/src/services/native-runtime/native-run-projection.ts

@@ -138,7 +138,7 @@ export async function projectCommittedRun(input: {
     await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id });
     await notifyUnsafeWorkspaceExportRunChange(tx, input.run);
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
-    await notifyExecutionRunWork(tx);
+    await notifyExecutionRunWork(tx, { companyId: input.run.companyId });
     await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
     const [after] = await tx
       .update(heartbeatRuns)

server/src/services/native-runtime/native-safe-replacement.ts

@@ -267,7 +267,7 @@ export async function reconcileSafeNativeReplacements(
             current.failureDetail?.replacementDenied
           )
             return;
-          await notifyExecutionRunWork(tx);
+          await notifyExecutionRunWork(tx, { companyId: run.companyId });
           await tx
             .update(nativeRunFinalizations)
             .set({
@@ -409,7 +409,7 @@ export async function reconcileSafeNativeReplacements(
             idempotencyKey: `native-safe-replacement:${run.id}`,
           })
           .returning();
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await tx.insert(heartbeatRuns).values({
           id: successorRunId,
           companyId: run.companyId,

server/src/services/native-runtime/native-session-executor.ts

@@ -9691,7 +9691,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.runId, input.execution.binding.runId),
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
           ),
-        ));
+        ), { companyId: input.execution.binding.companyId });
 
       const failedAtMs = Date.now();
       const executionFailureMessage = redactIdentityText(
@@ -9743,7 +9743,7 @@ async function executePaperclipNativeSessionWithinScope(
             eq(nativeRunFinalizations.leaseOwner, leaseOwner),
             eq(nativeRunFinalizations.attempt, attempt),
             isNull(nativeRunFinalizations.resultId),
-          )).returning({ runId: nativeRunFinalizations.runId }));
+          )).returning({ runId: nativeRunFinalizations.runId }), { companyId: input.execution.binding.companyId });
           if (settled) {
             await stoppedLeaseRenewal;
             if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
@@ -9830,7 +9830,7 @@ async function executePaperclipNativeSessionWithinScope(
             ),
           )
           .for("update");
-        await notifyExecutionRunWork(tx);
+        await notifyExecutionRunWork(tx, { companyId: input.execution.binding.companyId });
         await notifyNativeFinalizationWork(tx, input.execution.binding);
         const updated = await tx
           .update(nativeRunFinalizations)

server/src/services/native-runtime/native-superseded-retry.ts

@@ -39,7 +39,7 @@ export async function retireSupersededNativeRetry(db: Db, input: {
   )).limit(1);
   if (!source) return false;
 
-  await notifyExecutionRunWork(db);
+  await notifyExecutionRunWork(db, { companyId: run.companyId });
   await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null,
     leaseExpiresAt: null, controlDeadlineAt: null, recoveryState: "blocked",
     failureCode: "native_retry_superseded", failureDetail: { ...coordinator.failureDetail,

server/src/services/native-runtime/native-workspace-reconstruction.test.ts

@@ -38,7 +38,7 @@ describe("reconstruction after confirmed native workspace loss", () => {
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RECONSTRUCTION_TEST_DATABASE_URL");
     if (external) {
       await applyPendingMigrations(external);
-      db = createDb(external);
+      db = createDb(external, { maxConnections: 6 });
       return;
     }
     temporary = await startEmbeddedPostgresTestDatabase("workspace-reconstruction-");
@@ -144,17 +144,21 @@ describe("reconstruction after confirmed native workspace loss", () => {
     expect(await queue.nextTimerHeartbeatAt(agent)).toBeNull();
     const events: string[] = [];
     const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event));
+    const routineEvents: string[] = [];
+    const unsubscribeRoutine = subscribeDatabaseWork(db, DELIVERY_QUEUES.routineExecutionStatus, event => routineEvents.push(event));
     try {
       await db.transaction(async tx => {
         await reconstructNativeWorkspace({ ...f, db: tx as unknown as Db });
         expect(events).toContain("intent");
         expect(events).not.toContain("settled");
+        expect(routineEvents).toContain("intent"); expect(routineEvents).not.toContain("settled");
         expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked");
       });
       expect(events).toContain("settled");
+      expect(routineEvents).toContain("settled");
       expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("todo");
       expect(await queue.nextTimerHeartbeatAt(agent)).toBe(agent.createdAt.getTime() + 600_000);
-    } finally { unsubscribe(); }
+    } finally { unsubscribe(); unsubscribeRoutine(); }
   });
 
   it("pins a historical cancelled workspace through public heartbeat startup without replaying its turn", async () => {

server/src/services/native-runtime/native-workspace-reconstruction.ts

@@ -1,3 +1,4 @@
+import { notifyRoutineExecutionStatus } from "../routine-execution-status-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { and, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm";
@@ -187,6 +188,7 @@ export async function reconstructNativeWorkspace(input: {
     }).where(eq(issueRecoveryActions.id, source.action.id));
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+    await notifyRoutineExecutionStatus(tx, input.issueId, input.companyId);
     await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null,
       blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id));
     const publication = await persistActivity(db, { companyId: input.companyId, actorType: "user", actorId: input.actorId,

server/src/services/routine-execution-status-notifications.ts

@@ -1,5 +1,8 @@
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export function notifyRoutineExecutionStatus(tx: object, issueId?: string) {
-  return notifyDeliveryWork(tx, DELIVERY_QUEUES.routineExecutionStatus, issueId ?? "*");
+export type RoutineExecutionStatusScope = { issueId?: string; companyId?: string };
+
+export function notifyRoutineExecutionStatus(tx: object, issueId?: string, companyId?: string) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.routineExecutionStatus,
+    issueId || companyId ? JSON.stringify({ issueId, companyId } satisfies RoutineExecutionStatusScope) : "*");
 }

server/src/services/routine-execution-status-work.ts

@@ -1,18 +1,27 @@
 import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
-import type { routineExecutionStatusService } from "./routine-run-status.js";
+import type { RoutineExecutionStatusScope } from "./routine-execution-status-notifications.js";
+import { ROUTINE_EXECUTION_STATUS_PAGE_SIZE, type routineExecutionStatusService } from "./routine-run-status.js";
 
 export function registerRoutineExecutionStatusWork(
   coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
   service: ReturnType<typeof routineExecutionStatusService>,
   canRun: () => boolean,
 ) {
+  const cursors = new Map<string, string>();
   return registerScopedDeadlineWork(coordinator, {
     queue: DELIVERY_QUEUES.routineExecutionStatus,
-    decodeScope: hint => hint === "*" ? undefined : hint,
-    matches: (entry, issueId?: string) => issueId === undefined || entry.issueId === issueId,
-    load: async (issueId?: string) => (await service.load(issueId)).map(row => ({ ...row, dueAt: Date.now() })),
+    decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RoutineExecutionStatusScope,
+    matches: (entry, scope?: RoutineExecutionStatusScope) => (!scope?.issueId || entry.issueId === scope.issueId)
+      && (!scope?.companyId || entry.companyId === scope.companyId),
+    load: async (scope?: RoutineExecutionStatusScope) => {
+      const key = JSON.stringify(scope) ?? "*";
+      const rows = await service.load(scope, cursors.get(key));
+      const complete = rows.length < ROUTINE_EXECUTION_STATUS_PAGE_SIZE;
+      if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
+      return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete };
+    },
     check: async entry => { await service.settle(entry); return null; },
   }, canRun);
 }

server/src/services/routine-run-status.ts

@@ -1,7 +1,7 @@
-import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
+import { notifyRoutineExecutionStatus, type RoutineExecutionStatusScope } from "./routine-execution-status-notifications.js";
 import { and, eq, inArray, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
-import { issues, pluginManagedResources, routineRuns, routines } from "@paperclipai/db";
+import { issues, routineRuns, routines } from "@paperclipai/db";
 
 const EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE = "execution_issue_status";
 const EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES = ["blocked", "cancelled"] as const;
@@ -141,17 +141,11 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string
 }
 
 
-export function routineExecutionOriginCondition() {
-  return or(sql`${issues.originId} = ${routineRuns.routineId}::text`, sql`exists (
-    select 1 from ${pluginManagedResources} resource
-    where resource.company_id = ${routineRuns.companyId}
-      and resource.resource_kind = 'routine' and resource.resource_id = ${routineRuns.routineId}
-      and resource.defaults_json -> 'issueTemplate' ->> 'originId' = ${issues.originId}
-  )`);
-}
+export const ROUTINE_EXECUTION_STATUS_PAGE_SIZE = 50;
 
 export function routineExecutionStatusDebtCondition() {
-  const transientFailure = sql`(${routineRuns.triggerPayload} -> 'transientFailure' ->> 'code' = ${EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE}
+  const transientFailure = sql`((${routineRuns.triggerPayload} -> 'transientFailure' ->> 'code' = ${EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE}
+      and ${routineRuns.triggerPayload} -> 'transientFailure' ->> 'status' in ('blocked', 'cancelled'))
     or ${routineRuns.failureReason} in ('Execution issue moved to blocked', 'Execution issue moved to cancelled'))`;
   return or(
     and(eq(routineRuns.status, "issue_created"), inArray(issues.status, ["done", "blocked", "cancelled"])),
@@ -165,15 +159,18 @@ export function routineExecutionStatusDebtCondition() {
 }
 
 export function routineExecutionStatusService(db: Db) {
-  async function load(issueId?: string) {
+  async function load(scope?: RoutineExecutionStatusScope, afterId?: string) {
     return db.select({ id: routineRuns.id, companyId: routineRuns.companyId,
       issueId: issues.id, routineId: routineRuns.routineId })
       .from(routineRuns)
       .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, routineRuns.companyId)))
       .innerJoin(issues, and(eq(issues.id, routineRuns.linkedIssueId), eq(issues.companyId, routineRuns.companyId),
-        eq(issues.originKind, "routine_execution"), sql`${issues.originRunId} = ${routineRuns.id}::text`, routineExecutionOriginCondition()))
-      .where(and(routineExecutionStatusDebtCondition(), ...(issueId ? [eq(issues.id, issueId)] : [])))
-      .orderBy(routineRuns.id);
+        eq(issues.originKind, "routine_execution"), sql`${issues.originRunId} = ${routineRuns.id}::text`))
+      .where(and(routineExecutionStatusDebtCondition(),
+        ...(scope?.issueId ? [eq(issues.id, scope.issueId)] : []),
+        ...(scope?.companyId ? [eq(routineRuns.companyId, scope.companyId)] : []),
+        ...(afterId ? [sql`${routineRuns.id} > ${afterId}::uuid`] : [])))
+      .orderBy(routineRuns.id).limit(ROUTINE_EXECUTION_STATUS_PAGE_SIZE);
   }
   async function settle(candidate: Awaited<ReturnType<typeof load>>[number]) {
     return db.transaction(async tx => {
@@ -190,7 +187,7 @@ export function routineExecutionStatusService(db: Db) {
         .innerJoin(issues, eq(issues.id, routineRuns.linkedIssueId))
         .where(and(eq(routineRuns.id, candidate.id), eq(routineRuns.companyId, candidate.companyId),
           eq(routineRuns.routineId, candidate.routineId), eq(routineRuns.linkedIssueId, issue.id),
-          routineExecutionOriginCondition(), routineExecutionStatusDebtCondition())).for("update", { of: routineRuns });
+          routineExecutionStatusDebtCondition())).for("update", { of: routineRuns });
       if (!run) return false;
       return (await syncRoutineRunStatusForIssue(tx as unknown as Db, issue.id)) !== null;
     });

server/src/services/routines.ts

@@ -2002,6 +2002,8 @@ export function routineService(
   }
 
 
+  let statusReconcileCursor: string | undefined;
+
   async function syncRunStatusForIssue(issueId: string, executor: Db = db) {
     return syncRoutineRunStatusForIssue(executor, issueId);
   }
@@ -3246,7 +3248,8 @@ export function routineService(
 
     reconcileExecutionRunStatuses: async (): Promise<{ scanned: number; settled: number; deferred: number }> => {
       const service = routineExecutionStatusService(db);
-      const candidates = await service.load();
+      const candidates = await service.load(undefined, statusReconcileCursor);
+      statusReconcileCursor = candidates.length === 50 ? candidates.at(-1)!.id : undefined;
       let settled = 0, deferred = 0;
       for (const candidate of candidates) {
         try {

server/src/services/system-recovery.ts

@@ -71,7 +71,7 @@ export async function updateSystemRecovery(db: Db, source: WorkspaceRecoverySour
       sql`${issueRecoveryActions.evidence}#>>'{systemRecovery,revision}' = ${String(previous.revision)}`))
       .returning({ id: issueRecoveryActions.id });
     return Boolean(updated);
-  });
+  }, { companyId: source.companyId });
 }
 
 export async function loadSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity) {

server/src/services/task-retry.ts

@@ -334,7 +334,7 @@ export async function requestTaskRetry(input: {
       details: { requestId: active?.id ?? id, sourceRunId: latest.id, attachedToExistingWork: Boolean(active || busy) } });
     return { requestId: active?.id ?? id, accepted: true, stale: false, observedRunId: latest.id,
       chatActionId: intent.chatActionId };
-  });
+  }, { companyId: input.companyId });
 }
 
 /** Chat owns admission and authorization. The task receipt follows that