fix(server): schedule plugin jobs from saved deadlines

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

doc/idle-plugin-jobs.md

@@ -0,0 +1,75 @@
+# Plugin job scheduling without idle polling
+
+Plugin cron scheduling uses the instance delivery coordinator. It no longer
+queries `plugin_jobs` every 30 seconds when there is no eligible work.
+
+## Ownership and durable state
+
+The plugin scheduler owns cron interpretation, dispatch eligibility, concurrency,
+and run settlement. `plugin_jobs` stores schedules and their next deadlines;
+`plugin_job_runs` records admitted calls and their outcomes. No table or migration
+is added. The application passes its existing coordinator to `scheduler.start`.
+The coordinator owns timer execution, transaction uncertainty, and drain gating.
+
+The job store and registry signal before changes inside their transactions.
+Definition sync, pause/resume, changed schedules, run creation/settlement, job
+removal, and plugin removal invalidate the scheduler. Nested transactions wake
+only after their outer transaction settles. An empty manifest is a real sync:
+both the loader and lifecycle coordinator pause removed definitions.
+Registration and worker status changes provide local lifecycle wakes. The shared
+`registerDeadlineWork` helper retains explicit wakes received during drain,
+including a worker becoming available without a database mutation.
+
+## Idle and dispatch behavior
+
+- Empty instances, manual-only jobs, invalid cron definitions, paused jobs, and
+  unavailable workers have no recurring scheduler SQL after reconciliation.
+- Configured future jobs arm their saved next deadline. No SQL is needed between
+  reconciliation and that deadline unless a supported producer changes state.
+- Startup reconstructs deadlines. An overdue saved timestamp is owed work and
+  is not replaced merely because the plugin registered again. A missed schedule
+  dispatches once, then computes its next cron time after completion.
+- A missing timestamp is initialized from the current cron expression. A changed
+  cron clears the old timestamp so the new schedule can be initialized.
+- Claims recheck the definition under a job-row lock and reject outstanding
+  queued, pending, or running records. Manual and scheduled dispatch share this
+  boundary. Worker RPC calls run outside transactions.
+- Active calls consume concurrency slots and remain tracked until settlement.
+  Their completion wakes waiting work without an interval. Slow calls do not
+  prevent other available slots from being used.
+- Completion and the scheduled job's next timestamp commit together. Failed
+  persistence retries the retained settlement, not the plugin side effect.
+  A concurrent schedule edit is preserved by the settlement check.
+- Stop rejects new admissions, detaches worker subscriptions, and waits for
+  admitted work. Existing calls retain their RPC timeout. The application then
+  stops the shared coordinator before ending the database pool.
+
+## Limits
+
+This is the existing single-process writer model. Independent SQL writers and
+other app replicas do not emit these local signals. The row lock prevents an
+overlapping claim, but does not establish a supported multi-replica wake protocol.
+
+A process crash after a plugin call but before durable completion leaves an
+unresolved run. The scheduler does not assume that an unknown external effect is
+safe to repeat. Such a run remains outstanding until explicit lifecycle/operator
+reconciliation. During a live process, failed completion writes retain their
+exact result and retry. This does not invent exactly-once external execution.
+
+Plugin definitions remain instance-scoped, as on the base branch. The separate
+company-scoped job proposal is not implemented here. Plugin configuration and
+worker authorization retain their existing owners.
+
+The Cloud sleep proof remains conservative for plugin work. An internal cron
+deadline cannot wake a stopped host; this PR does not add external alarm coverage,
+allow configured plugins to bypass idle checks, or claim measured host sleep.
+
+## Verification
+
+`plugin-scheduled-work.test.ts` uses real PostgreSQL and controlled worker calls.
+It checks idle SQL silence with no jobs and configured future/manual-only jobs;
+first writes, pause/resume, worker availability, outer commit and rollback;
+restart and overdue execution; removed schedules; concurrent claims; failed
+completion persistence without repeated RPC; capacity refill; drain; and shutdown.
+Worker responses are simulated. Existing startup, authorization, loader, shared
+coordinator, decision-expiry, and warm-standby tests cover affected consumers.

doc/plans/2026-10-24-plugin-job-sleep-audit.md

@@ -70,3 +70,21 @@ The current health checker returns immediately for AI-purpose connections, while
 the sweep does not explicitly exclude that purpose. Audit inert due rows and
 pre-probe setup failures before deriving deadlines from `healthCheckedAt`.
 No tool-health migration is claimed by this plugin-job slice.
+
+## Implementation decisions
+
+The migration now uses existing job/run rows and the shared deadline worker.
+Local lifecycle wakes needed the helper to retain an explicit wake across drain;
+the plugin tests exercise that condition through a real status event. The worker
+handle already exposes status subscriptions, so no new worker-manager API is
+needed. The job store accepts the query/transaction subset shared by root clients
+and transactions instead of casting transaction handles to full pools.
+
+Claims recheck current state with a job-row lock. Worker calls stay outside the
+transaction. Completion and schedule advancement commit atomically; failed
+persistence retries only settlement. Unresolved old runs are not blindly replayed.
+This preserves the conservative Cloud sleep blocker for unknown external effects.
+
+Related proposals: #7905 covers empty manifest sync; this migration needs that
+behavior on both activation paths. #11571 proposes company-scoped jobs; this
+migration retains the current instance-scoped contract and does not implement it.

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -931,3 +931,24 @@ after its single justified retry. The duplicate cancelled workflows for #17490
 and #17497 were each rerun once to clear stale cancelled contexts. The #17490
 retry has a new E2E shard 3 failure under investigation; do not call it green or
 rerun again without a repair. #17497's retry was still running at this observation.
+
+### Plugin scheduling implementation, 2026-10-24
+
+The next slice replaces the plugin scheduler's private 30-second query loop with
+shared deadlines and transaction-aware producer wakes. It uses existing job/run
+rows, preserves overdue schedules, handles empty manifest sync, and tracks calls
+through durable settlement and shutdown. See [the implementation contract](../idle-plugin-jobs.md)
+and [producer audit](2026-10-24-plugin-job-sleep-audit.md).
+
+The baseline empty-instance regression failed before the migration. The first
+expanded local run passed 59 tests across plugin scheduling, warm standby, the
+shared coordinator, and decision expiry. The consumer run exposed an incomplete
+loader mock after empty-manifest synchronization became mandatory; that fixture
+was corrected; all 14 loader recovery cases passed. Workspace typechecks and the
+production build passed. Final server typecheck and build also passed after the
+lifecycle wiring edits. Remote review/checks remain open.
+
+Separate stack gates remain: #17490 E2E cleanup hit a cancellation lock timeout;
+#17510 has a security finding about unscoped approval queue scans plus CI failures.
+Neither PR is claimed green. Approval retention #17497 is green at `2c5babcabc`.
+The complete original inventory remains in scope; this slice does not finish it.

server/src/__tests__/cloud-warm-standby.test.ts

@@ -140,16 +140,18 @@ describe("unclaimed Cloud background work", () => {
       tickIntervalMs: 1000,
       isBackgroundWorkEnabled: () => enabled,
     });
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, onError: () => {} });
     try {
       email.start();
-      scheduler.start();
+      await scheduler.start(coordinator);
       await vi.advanceTimersByTimeAsync(10 * 60_000);
       expect(select).not.toHaveBeenCalled();
       enabled = true;
       await vi.advanceTimersByTimeAsync(1000);
       expect(select.mock.calls.length).toBeGreaterThanOrEqual(2);
     } finally {
-      scheduler.stop();
+      await scheduler.stop();
+      await coordinator.stop();
       await email.shutdown();
     }
   });

server/src/__tests__/plugin-environment-driver-ready-recovery.test.ts

@@ -439,7 +439,7 @@ describe("listReadyPluginEnvironmentDrivers worker recovery", () => {
         clearPlugin: vi.fn(),
         subscriptionCount: vi.fn(() => 0),
       };
-      const jobScheduler = { unregisterPlugin: vi.fn().mockResolvedValue(undefined) };
+      const jobScheduler = { registerPlugin: vi.fn().mockResolvedValue(undefined), unregisterPlugin: vi.fn().mockResolvedValue(undefined) };
       const toolDispatcher = { unregisterPluginTools: vi.fn(), registerPluginTools: vi.fn() };
 
       const loader = pluginLoader(
@@ -454,7 +454,7 @@ describe("listReadyPluginEnvironmentDrivers worker recovery", () => {
           workerManager,
           eventBus,
           jobScheduler,
-          jobStore: {},
+          jobStore: { syncJobDeclarations: vi.fn().mockResolvedValue(undefined) },
           toolDispatcher,
           buildHostHandlers: vi.fn(() => ({})),
           instanceInfo: {

server/src/__tests__/plugin-scheduled-work.test.ts

@@ -1,33 +1,213 @@
-import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+import { EventEmitter } from "node:events";
-import { sql } from "drizzle-orm";
+import { randomUUID } from "node:crypto";
-import { createDb } from "@paperclipai/db";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { eq, sql } from "drizzle-orm";
+import { createDb, plugins, pluginJobs, pluginJobRuns } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { createPluginJobScheduler } from "../services/plugin-job-scheduler.js";
 import { pluginJobStore } from "../services/plugin-job-store.js";
+import { pluginRegistryService } from "../services/plugin-registry.js";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
 
 const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PLUGIN_SCHEDULE_TEST_DATABASE_URL");
 const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 (support.supported ? describe : describe.skip)("plugin job scheduling without idle polling", () => {
   let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
-  let db: ReturnType<typeof createDb>;
+  let db: ReturnType<typeof createDb>, pluginId: string;
   let scheduler: ReturnType<typeof createPluginJobScheduler> | undefined;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let available = true, draining = false;
+  let events: EventEmitter;
+  const rpc = vi.fn<(...args: unknown[]) => Promise<unknown>>();
+  const errors: unknown[] = [];
   beforeAll(async () => {
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("plugin-schedule-");
-    db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
-    await db.execute(sql`select 1`);
   }, 30_000);
-  afterEach(() => { scheduler?.stop(); vi.useRealTimers(); vi.restoreAllMocks(); });
+  beforeEach(async () => {
-  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+    pluginId = randomUUID(); available = true; draining = false; events = new EventEmitter(); errors.length = 0;
-
+    rpc.mockReset(); rpc.mockResolvedValue({});
-  it("does not query repeatedly after an empty startup reconciliation", async () => {
+    await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: pluginId, version: "1.0.0", manifestJson: {} as never });
-    const workerManager = { isRunning: () => false, call: vi.fn() } as unknown as PluginWorkerManager;
+    await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
-    scheduler = createPluginJobScheduler({ db, jobStore: pluginJobStore(db), workerManager });
     vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
-    scheduler.start(); await scheduler.tick();
+  });
-    const reads = vi.spyOn(db, "select");
+  afterEach(async () => {
+    await scheduler?.stop(); await coordinator?.stop(); scheduler = undefined; coordinator = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks();
+    await db.execute(sql`drop trigger if exists plugin_settlement_failure on plugin_job_runs`);
+    await db.execute(sql`drop function if exists plugin_settlement_failure()`);
+    await db.delete(plugins).where(eq(plugins.id, pluginId));
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  async function start(maxConcurrentJobs = 10) {
+    const workerManager = { isRunning: () => available, call: rpc, getWorker: () => events } as unknown as PluginWorkerManager;
+    scheduler = createPluginJobScheduler({ db, jobStore: pluginJobStore(db), workerManager,
+      isBackgroundWorkEnabled: () => !draining, maxConcurrentJobs });
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
+      onError: error => errors.push(error) });
+    await scheduler.start(coordinator);
+    await scheduler.registerPlugin(pluginId);
+  }
+  const job = (key = "hourly", schedule = "0 * * * *") => pluginRegistryService(db).upsertJob(pluginId, key, { schedule });
+  const runs = () => db.select().from(pluginJobRuns).where(eq(pluginJobRuns.pluginId, pluginId));
+  async function settled() {
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(idleWorkSnapshot().active).toBe(0);
+    }, { timeout: 5000 });
+  }
+  async function quiet() {
+    await settled(); expect(coordinator!.nextWakeAt()).toBeNull();
+    const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute");
     await vi.advanceTimersByTimeAsync(180_000);
-    expect(reads).not.toHaveBeenCalled();
+    expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled();
+    reads.mockRestore(); writes.mockRestore(); executes.mockRestore();
+  }
+  async function due(key = "hourly") {
+    const row = await job(key);
+    await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), new Date(Date.now() - 1000));
+    return row;
+  }
+  async function completed(count = 1) {
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect((await runs()).filter(run => run.status === "succeeded" || run.status === "failed")).toHaveLength(count);
+    }, { timeout: 5000 });
+  }
+
+  it("does not query repeatedly after an empty startup reconciliation", async () => { await start(); await quiet(); });
+  it("keeps configured manual-only and invalid schedules quiet", async () => {
+    await job("manual", ""); await job("invalid", "not cron"); await start(); await quiet(); expect(rpc).not.toHaveBeenCalled();
+  });
+  it("initializes a first declaration after empty startup and runs only at its deadline", async () => {
+    await start(); await quiet(); const row = await job(); await settled();
+    const current = await pluginJobStore(db).getJobById(row.id);
+    expect(current?.nextRunAt!.getTime()).toBeGreaterThan(Date.now());
+    expect(rpc).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(current!.nextRunAt!.getTime() - Date.now()); await completed(); expect(rpc).toHaveBeenCalledTimes(1);
+  });
+  it("catches up an overdue startup job without moving its owed timestamp", async () => {
+    const row = await due(); const at = (await pluginJobStore(db).getJobById(row.id))!.nextRunAt!;
+    await start(); await completed();
+    expect(rpc.mock.calls[0][2]).toMatchObject({ job: { scheduledAt: at.toISOString(), trigger: "schedule" } });
+    expect((await pluginJobStore(db).getJobById(row.id))!.nextRunAt!.getTime()).toBeGreaterThan(Date.now());
+  });
+  it("pauses a due schedule and reactivates from the real status producer", async () => {
+    const row = await due(); await pluginJobStore(db).updateJobStatus(row.id, "paused"); await start(); await quiet();
+    await pluginJobStore(db).updateJobStatus(row.id, "active"); await completed(); expect(rpc).toHaveBeenCalledTimes(1);
+  });
+  it("parks an unavailable worker and wakes overdue work from its lifecycle event", async () => {
+    await due(); available = false; await start(); await quiet();
+    available = true; events.emit("status", { status: "ready" }); await completed(); expect(rpc).toHaveBeenCalledTimes(1);
+  });
+  it("retains a worker-availability wake received during drain", async () => {
+    await due(); available = false; await start(); await quiet();
+    draining = true; available = true; events.emit("status", { status: "ready" });
+    await vi.advanceTimersByTimeAsync(10_000); expect(rpc).not.toHaveBeenCalled();
+    draining = false; await completed();
+  });
+  it("retains the first committed job created during drain", async () => {
+    await start(); await quiet(); draining = true; await due();
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    expect(rpc).not.toHaveBeenCalled(); draining = false; await completed();
+  });
+  it("does not dispatch a nested write until the outer transaction commits", async () => {
+    await start(); await quiet();
+    await db.transaction(async tx => {
+      await pluginJobStore(tx).syncJobDeclarations(pluginId, [{ jobKey: "outer", schedule: "0 * * * *" }]);
+      const row = await pluginJobStore(tx).getJobByKey(pluginId, "outer");
+      await pluginJobStore(tx).updateRunTimestamps(row!.id, new Date(0), new Date(Date.now() - 1000));
+      await vi.advanceTimersByTimeAsync(5000); expect(rpc).not.toHaveBeenCalled();
+    });
+    await completed();
+  });
+  it("disarms after a rolled-back declaration", async () => {
+    await start(); await quiet();
+    await expect(db.transaction(async tx => {
+      await pluginJobStore(tx).syncJobDeclarations(pluginId, [{ jobKey: "rollback", schedule: "* * * * *" }]);
+      throw new Error("abort");
+    })).rejects.toThrow("abort");
+    await quiet(); expect(rpc).not.toHaveBeenCalled();
+  });
+  it("parks removed declarations while retaining their history", async () => {
+    await due(); await start(); await completed();
+    await pluginJobStore(db).syncJobDeclarations(pluginId, []); await quiet(); expect(await runs()).toHaveLength(1);
+  });
+  it("recomputes a changed cron schedule", async () => {
+    await job("changed", "0 0 1 1 *"); await start(); await settled();
+    await job("changed", "* * * * *"); await settled();
+    expect(coordinator!.nextWakeAt()! - Date.now()).toBeLessThanOrEqual(60_000);
+    await vi.advanceTimersByTimeAsync(60_000); await completed();
+  });
+  it("retries failed completion persistence without repeating the plugin side effect", async () => {
+    await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`);
+    await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`);
+    await due(); await start();
+    await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); expect(rpc).toHaveBeenCalledTimes(1);
+    await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0;
+    await completed(); expect(rpc).toHaveBeenCalledTimes(1);
+  });
+  it("records provider failure and advances the cron obligation", async () => {
+    rpc.mockRejectedValueOnce(new Error("plugin rejected job")); await due(); await start(); await completed();
+    expect((await runs())[0]).toMatchObject({ status: "failed", error: "plugin rejected job" });
+    await settled(); expect(coordinator!.nextWakeAt()! > Date.now()).toBe(true);
+  });
+  it("does not replay an unresolved run left by another process", async () => {
+    const row = await due(); await pluginJobStore(db).createRun({ jobId: row.id, pluginId, trigger: "schedule" });
+    await start(); await quiet(); expect(rpc).not.toHaveBeenCalled();
+    const [run] = await runs(); await pluginJobStore(db).completeRun(run.id, { status: "cancelled" }); await completed();
+  });
+  it("fills released capacity from completion without a poll", async () => {
+    await due("one"); await due("two"); let release!: () => void;
+    rpc.mockImplementationOnce(() => new Promise<void>(resolve => { release = resolve; }));
+    const starting = start(1); await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1));
+    release(); await starting; await completed(2); expect(rpc).toHaveBeenCalledTimes(2);
+  });
+  it("tracks manual work and waits for admitted calls during shutdown", async () => {
+    const row = await job("manual", ""); await start(); await quiet(); let release!: () => void;
+    rpc.mockImplementationOnce(() => new Promise<void>(resolve => { release = resolve; }));
+    await scheduler!.triggerJob(row.id); await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1));
+    expect(idleWorkSnapshot().active).toBeGreaterThan(0);
+    let stopped = false; const stopping = scheduler!.stop().then(() => { stopped = true; });
+    await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false);
+    release(); await stopping; expect((await runs())[0].status).toBe("succeeded");
+  });
+  it("does no SQL while waiting for a future configured deadline", async () => {
+    const row = await job(); const at = new Date(Date.now() + 600_000);
+    await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at);
+    await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime());
+    const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+    await vi.advanceTimersByTimeAsync(180_000);
+    expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+    reads.mockRestore(); writes.mockRestore(); expect(rpc).not.toHaveBeenCalled();
+  });
+  it("reconstructs the saved deadline after scheduler restart", async () => {
+    const row = await job(); const at = new Date(Date.now() + 600_000);
+    await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at);
+    await start(); await settled(); await scheduler!.stop(); await coordinator!.stop();
+    await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime());
+    await vi.advanceTimersByTimeAsync(at.getTime() - Date.now()); await completed(); expect(rpc).toHaveBeenCalledTimes(1);
+  });
+  it("rechecks a paused definition after waiting for its lock", async () => {
+    const row = await due(); let startup!: Promise<void>;
+    await db.transaction(async tx => {
+      await pluginJobStore(tx).updateJobStatus(row.id, "paused");
+      startup = start(); await vi.advanceTimersByTimeAsync(1000); expect(rpc).not.toHaveBeenCalled();
+    });
+    await startup; await quiet(); expect(rpc).not.toHaveBeenCalled();
+  });
+  it("prevents overlapping concurrent manual triggers", async () => {
+    const row = await job("manual", ""); await start(); await quiet(); let release!: () => void;
+    rpc.mockImplementation(() => new Promise<void>(resolve => { release = resolve; }));
+    const attempts = await Promise.allSettled([scheduler!.triggerJob(row.id), scheduler!.triggerJob(row.id)]);
+    await vi.waitFor(() => expect(rpc).toHaveBeenCalledTimes(1));
+    release(); await completed(); expect(attempts.filter(result => result.status === "fulfilled")).toHaveLength(1);
+  });
+  it("deleting the last job cancels its future deadline", async () => {
+    await job(); await start(); await settled(); await pluginJobStore(db).deleteAllJobs(pluginId); await quiet();
   });
 });

server/src/app.ts

@@ -1267,7 +1267,6 @@ export async function createApp(
   app.use(errorHandler);
 
   jobCoordinator.start();
-  scheduler.start();
   const deliveryWork = createDeliveryWorkCoordinator({
     owner: db,
     canRun: () => !isWarmStandby() && !isIdleTaskDrainActive(),
@@ -1275,6 +1274,7 @@ export async function createApp(
     onError: (err, queue) => logger.error({ err, queue }, "Delivery reconciliation failed"),
   });
   app.locals.deliveryWork = deliveryWork;
+  await scheduler.start(deliveryWork);
   const agentLifecycleDelivery = deliveryWork.register(DELIVERY_QUEUES.agentLifecycle, {
     retryMs: 1_000,
     run: () => agentLifecycle.sweep(),
@@ -1458,7 +1458,7 @@ export async function createApp(
     appServicesShutdown = (async () => {
       // The scheduler tick queries the database. Stop it here, inside the
       // awaited teardown, so no tick runs after the caller ends the pool.
-      scheduler.stop();
+      await scheduler.stop();
       await agentLifecycle.stop();
       await publicMcpEvents?.stop();
       await dotMcpEvents?.stop();

server/src/services/deadline-work.ts

@@ -14,7 +14,7 @@ export function registerDeadlineWork(
   let requested = 1, reconciled = 0;
   let next: number | null = null;
   let earliest = 0;
-  return coordinator.register(queue, {
+  const worker = coordinator.register(queue, {
     retryMs: task.retryMs,
     pausePassiveMaintenance: true,
     onSettled: () => { requested++; },
@@ -32,4 +32,5 @@ export function registerDeadlineWork(
     hasPending: async () => requested !== reconciled,
     nextRunAt: () => next,
   });
+  return { ...worker, wake: () => { requested++; worker.wake(); } };
 }

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

@@ -59,6 +59,7 @@ export const DELIVERY_QUEUES = {
   gitRestorePoints: "git-restore-points",
   nativeWorkspaceSyncRetention: "native-workspace-sync-retention",
   question: "question-responses",
+  pluginSchedules: "plugin-schedules",
   pluginLifecycle: "plugin-resource-lifecycle",
   toolAction: "tool-action-receipts",
   terminalWorkspaces: "terminal-workspaces",

server/src/services/plugin-job-coordinator.ts

@@ -130,8 +130,8 @@ export function createPluginJobCoordinator(
           { pluginId, pluginKey, jobCount: jobDeclarations.length },
           "syncing job declarations from manifest",
         );
-        await jobStore.syncJobDeclarations(pluginId, jobDeclarations);
       }
+      await jobStore.syncJobDeclarations(pluginId, jobDeclarations);
 
       // Register with the scheduler (computes nextRunAt for active jobs)
       await scheduler.registerPlugin(pluginId);

server/src/services/plugin-job-scheduler.ts

@@ -1,755 +1,248 @@
-/**
+import { and, eq, inArray } from "drizzle-orm";
- * PluginJobScheduler — tick-based scheduler for plugin scheduled jobs.
+import { pluginJobs, pluginJobRuns, type Db } from "@paperclipai/db";
- *
+import { pluginJobStore, type PluginJobStore } from "./plugin-job-store.js";
- * The scheduler is the central coordinator for all plugin cron jobs. It
- * periodically ticks (default every 30 seconds), queries the `plugin_jobs`
- * table for jobs whose `nextRunAt` has passed, dispatches `runJob` RPC calls
- * to the appropriate worker processes, records each execution in the
- * `plugin_job_runs` table, and advances the scheduling pointer.
- *
- * ## Responsibilities
- *
- * 1. **Tick loop** — A `setInterval`-based loop fires every `tickIntervalMs`
- *    (default 30s). Each tick scans for due jobs and dispatches them.
- *
- * 2. **Cron parsing & next-run calculation** — Uses the lightweight built-in
- *    cron parser ({@link parseCron}, {@link nextCronTick}) to compute the
- *    `nextRunAt` timestamp after each run or when a new job is registered.
- *
- * 3. **Overlap prevention** — Before dispatching a job, the scheduler checks
- *    for an existing `running` run for the same job. If one exists, the job
- *    is skipped for that tick.
- *
- * 4. **Job run recording** — Every execution creates a `plugin_job_runs` row:
- *    `queued` → `running` → `succeeded` | `failed`. Duration and error are
- *    captured.
- *
- * 5. **Lifecycle integration** — The scheduler exposes `registerPlugin()` and
- *    `unregisterPlugin()` so the host lifecycle manager can wire up job
- *    scheduling when plugins start/stop. On registration, the scheduler
- *    computes `nextRunAt` for all active jobs that don't already have one.
- *
- * @see PLUGIN_SPEC.md §17 — Scheduled Jobs
- * @see ./plugin-job-store.ts — Persistence layer
- * @see ./cron.ts — Cron parsing utilities
- */
-
-import { and, eq, lte, or } from "drizzle-orm";
-import type { Db } from "@paperclipai/db";
-import { pluginJobs, pluginJobRuns } from "@paperclipai/db";
-import type { PluginJobStore } from "./plugin-job-store.js";
 import type { PluginWorkerManager } from "./plugin-worker-manager.js";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { registerDeadlineWork } from "./deadline-work.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { parseCron, nextCronTick, validateCron } from "./cron.js";
+import { trackIdleWork } from "./task-admission.js";
 import { logger } from "../middleware/logger.js";
 
-// ---------------------------------------------------------------------------
+type Job = typeof pluginJobs.$inferSelect;
-// Constants
+type Trigger = "schedule" | "manual" | "retry";
-// ---------------------------------------------------------------------------
+type Coordinator = ReturnType<typeof createDeliveryWorkCoordinator>;
 
-/** Default interval between scheduler ticks (30 seconds). */
-const DEFAULT_TICK_INTERVAL_MS = 30_000;
-
-/** Default timeout for a runJob RPC call (5 minutes). */
-const DEFAULT_JOB_TIMEOUT_MS = 5 * 60 * 1_000;
-
-/** Maximum number of concurrent job executions across all plugins. */
-const DEFAULT_MAX_CONCURRENT_JOBS = 10;
-
-// ---------------------------------------------------------------------------
-// Types
-// ---------------------------------------------------------------------------
-
-/**
- * Options for creating a PluginJobScheduler.
- */
 export interface PluginJobSchedulerOptions {
-  /** Suppress periodic database work while an unclaimed Cloud app stands by. */
   isBackgroundWorkEnabled?: () => boolean;
-  /** Drizzle database instance. */
   db: Db;
-  /** Persistence layer for jobs and runs. */
   jobStore: PluginJobStore;
-  /** Worker process manager for RPC calls. */
   workerManager: PluginWorkerManager;
-  /** Interval between scheduler ticks in ms (default: 30s). */
   tickIntervalMs?: number;
-  /** Timeout for individual job RPC calls in ms (default: 5min). */
   jobTimeoutMs?: number;
-  /** Maximum number of concurrent job executions (default: 10). */
   maxConcurrentJobs?: number;
 }
-
+export interface TriggerJobResult { runId: string; jobId: string }
-/**
- * Result of a manual job trigger.
- */
-export interface TriggerJobResult {
-  /** The created run ID. */
-  runId: string;
-  /** The job ID that was triggered. */
-  jobId: string;
-}
-
-/**
- * Diagnostic information about the scheduler.
- */
 export interface SchedulerDiagnostics {
-  /** Whether the tick loop is running. */
   running: boolean;
-  /** Number of jobs currently executing. */
   activeJobCount: number;
-  /** Set of job IDs currently in-flight. */
   activeJobIds: string[];
-  /** Total number of ticks executed since start. */
   tickCount: number;
-  /** Timestamp of the last tick (ISO 8601). */
   lastTickAt: string | null;
 }
-
-// ---------------------------------------------------------------------------
-// Scheduler
-// ---------------------------------------------------------------------------
-
-/**
- * The public interface of the job scheduler.
- */
 export interface PluginJobScheduler {
-  /**
+  start(coordinator: Coordinator): Promise<void>;
-   * Start the scheduler tick loop.
+  stop(): Promise<void>;
-   *
-   * Safe to call multiple times — subsequent calls are no-ops.
-   */
-  start(): void;
-
-  /**
-   * Stop the scheduler tick loop.
-   *
-   * In-flight job runs are NOT cancelled — they are allowed to finish
-   * naturally. The tick loop simply stops firing.
-   */
-  stop(): void;
-
-  /**
-   * Register a plugin with the scheduler.
-   *
-   * Computes `nextRunAt` for all active jobs that are missing it. This is
-   * typically called after a plugin's worker process starts and
-   * `syncJobDeclarations()` has been called.
-   *
-   * @param pluginId - UUID of the plugin
-   */
   registerPlugin(pluginId: string): Promise<void>;
-
-  /**
-   * Unregister a plugin from the scheduler.
-   *
-   * Cancels any in-flight runs for the plugin and removes tracking state.
-   *
-   * @param pluginId - UUID of the plugin
-   */
   unregisterPlugin(pluginId: string): Promise<void>;
-
-  /**
-   * Manually trigger a specific job (outside of the cron schedule).
-   *
-   * Creates a run with `trigger: "manual"` and dispatches immediately,
-   * respecting the overlap prevention check.
-   *
-   * @param jobId - UUID of the job to trigger
-   * @param trigger - What triggered this run (default: "manual")
-   * @returns The created run info
-   * @throws {Error} if the job is not found, not active, or already running
-   */
   triggerJob(jobId: string, trigger?: "manual" | "retry"): Promise<TriggerJobResult>;
-
+  tick(signal?: AbortSignal): Promise<void>;
-  /**
-   * Run a single scheduler tick immediately (for testing).
-   *
-   * @internal
-   */
-  tick(): Promise<void>;
-
-  /**
-   * Get diagnostic information about the scheduler state.
-   */
   diagnostics(): SchedulerDiagnostics;
 }
 
-// ---------------------------------------------------------------------------
+function nextCron(schedule: string, now: Date): Date | null {
-// Implementation
+  return schedule && !validateCron(schedule) ? nextCronTick(parseCron(schedule), now) : null;
-// ---------------------------------------------------------------------------
+}
-
-/**
- * Create a new PluginJobScheduler.
- *
- * @example
- * ```ts
- * const scheduler = createPluginJobScheduler({
- *   db,
- *   jobStore,
- *   workerManager,
- * });
- *
- * // Start the tick loop
- * scheduler.start();
- *
- * // When a plugin comes online, register it
- * await scheduler.registerPlugin(pluginId);
- *
- * // Manually trigger a job
- * const { runId } = await scheduler.triggerJob(jobId);
- *
- * // On server shutdown
- * scheduler.stop();
- * ```
- */
-export function createPluginJobScheduler(
-  options: PluginJobSchedulerOptions,
-): PluginJobScheduler {
-  const {
-    db,
-    jobStore,
-    workerManager,
-    tickIntervalMs = DEFAULT_TICK_INTERVAL_MS,
-    jobTimeoutMs = DEFAULT_JOB_TIMEOUT_MS,
-    maxConcurrentJobs = DEFAULT_MAX_CONCURRENT_JOBS,
-  } = options;
 
+export function createPluginJobScheduler(options: PluginJobSchedulerOptions): PluginJobScheduler {
+  const { db, jobStore, workerManager, tickIntervalMs = 1_000,
+    jobTimeoutMs = 5 * 60_000, maxConcurrentJobs = 10 } = options;
   const log = logger.child({ service: "plugin-job-scheduler" });
-
-  // -----------------------------------------------------------------------
-  // State
-  // -----------------------------------------------------------------------
-
-  /** Timer handle for the tick loop. */
-  let tickTimer: ReturnType<typeof setInterval> | null = null;
-
-  /** Whether the scheduler is running. */
-  let running = false;
-
-  /** Set of job IDs currently being executed (for overlap prevention). */
   const activeJobs = new Set<string>();
-
+  const pendingSettlements = new Map<string, { persist: () => Promise<void>; running?: Promise<void> }>();
-  /** Total number of ticks since start. */
+  const inFlight = new Set<Promise<unknown>>();
+  const subscriptions = new Map<string, () => void>();
+  const unregistered = new Set<string>();
+  let worker: ReturnType<typeof registerDeadlineWork> | undefined;
+  let running = false;
+  let next: number | null = null;
   let tickCount = 0;
-
-  /** Timestamp of the last tick. */
   let lastTickAt: Date | null = null;
+  let ticking: Promise<void> | undefined;
+  const canRun = () => running && options.isBackgroundWorkEnabled?.() !== false;
 
-  /** Guard against concurrent tick execution. */
+  function track<T>(promise: Promise<T>): Promise<T> {
-  let tickInProgress = false;
+    const tracked = trackIdleWork(promise);
-
+    inFlight.add(tracked);
-  // -----------------------------------------------------------------------
+    void tracked.then(() => inFlight.delete(tracked), () => inFlight.delete(tracked));
-  // Core: tick
+    return tracked;
-  // -----------------------------------------------------------------------
-
-  /**
-   * A single scheduler tick. Queries for due jobs and dispatches them.
-   */
-  async function tick(): Promise<void> {
-    if (options.isBackgroundWorkEnabled?.() === false) return;
-    // Prevent overlapping ticks (in case a tick takes longer than the interval)
-    if (tickInProgress) {
-      log.debug("skipping tick — previous tick still in progress");
-      return;
-    }
-
-    tickInProgress = true;
-    tickCount++;
-    lastTickAt = new Date();
-
-    try {
-      const now = new Date();
-
-      // Query for jobs whose nextRunAt has passed and are active.
-      // We include jobs with null nextRunAt since they may have just been
-      // registered and need their first run calculated.
-      const dueJobs = await db
-        .select()
-        .from(pluginJobs)
-        .where(
-          and(
-            eq(pluginJobs.status, "active"),
-            lte(pluginJobs.nextRunAt, now),
-          ),
-        );
-
-      if (dueJobs.length === 0) {
-        return;
-      }
-
-      log.debug({ count: dueJobs.length }, "found due jobs");
-
-      // Dispatch each due job (respecting concurrency limits)
-      const dispatches: Promise<void>[] = [];
-
-      for (const job of dueJobs) {
-        // Concurrency limit
-        if (activeJobs.size >= maxConcurrentJobs) {
-          log.warn(
-            { maxConcurrentJobs, activeJobCount: activeJobs.size },
-            "max concurrent jobs reached, deferring remaining jobs",
-          );
-          break;
-        }
-
-        // Overlap prevention: skip if this job is already running
-        if (activeJobs.has(job.id)) {
-          log.debug(
-            { jobId: job.id, jobKey: job.jobKey, pluginId: job.pluginId },
-            "skipping job — already running (overlap prevention)",
-          );
-          continue;
-        }
-
-        // Check if the worker is available
-        if (!workerManager.isRunning(job.pluginId)) {
-          log.debug(
-            { jobId: job.id, pluginId: job.pluginId },
-            "skipping job — worker not running",
-          );
-          continue;
-        }
-
-        // Validate cron expression before dispatching
-        if (!job.schedule) {
-          log.warn(
-            { jobId: job.id, jobKey: job.jobKey },
-            "skipping job — no schedule defined",
-          );
-          continue;
-        }
-
-        dispatches.push(dispatchJob(job));
-      }
-
-      if (dispatches.length > 0) {
-        await Promise.allSettled(dispatches);
-      }
-    } catch (err) {
-      log.error(
-        { err: err instanceof Error ? err.message : String(err) },
-        "scheduler tick error",
-      );
-    } finally {
-      tickInProgress = false;
-    }
   }
 
-  // -----------------------------------------------------------------------
+  async function settle(jobId: string) {
-  // Core: dispatch a single job
+    const entry = pendingSettlements.get(jobId);
-  // -----------------------------------------------------------------------
+    if (!entry) return;
+    entry.running ??= entry.persist().finally(() => { entry.running = undefined; });
+    await entry.running;
+    if (pendingSettlements.get(jobId) === entry) pendingSettlements.delete(jobId);
+  }
 
-  /**
+  async function execute(job: Job, runId: string, trigger: Trigger) {
-   * Dispatch a single job run — create the run record, call the worker,
-   * record the result, and advance the schedule pointer.
-   */
-  async function dispatchJob(
-    job: typeof pluginJobs.$inferSelect,
-  ): Promise<void> {
-    const { id: jobId, pluginId, jobKey, schedule } = job;
-    const jobLog = log.child({ jobId, pluginId, jobKey });
-
-    // Mark as active (overlap prevention)
-    activeJobs.add(jobId);
-
-    let runId: string | undefined;
     const startedAt = Date.now();
-
+    let error: string | undefined;
-    try {
-      // 1. Create run record
-      const run = await jobStore.createRun({
-        jobId,
-        pluginId,
-        trigger: "schedule",
-      });
-      runId = run.id;
-
-      jobLog.info({ runId }, "dispatching scheduled job");
-
-      // 2. Mark run as running
-      await jobStore.markRunning(runId);
-
-      // 3. Call worker via RPC
-      await workerManager.call(
-        pluginId,
-        "runJob",
-        {
-          job: {
-            jobKey,
-            runId,
-            trigger: "schedule" as const,
-            scheduledAt: (job.nextRunAt ?? new Date()).toISOString(),
-          },
-        },
-        jobTimeoutMs,
-      );
-
-      // 4. Mark run as succeeded
-      const durationMs = Date.now() - startedAt;
-      await jobStore.completeRun(runId, {
-        status: "succeeded",
-        durationMs,
-      });
-
-      jobLog.info({ runId, durationMs }, "job completed successfully");
-    } catch (err) {
-      const durationMs = Date.now() - startedAt;
-      const errorMessage = err instanceof Error ? err.message : String(err);
-
-      jobLog.error(
-        { runId, durationMs, err: errorMessage },
-        "job execution failed",
-      );
-
-      // Record the failure
-      if (runId) {
-        try {
-          await jobStore.completeRun(runId, {
-            status: "failed",
-            error: errorMessage,
-            durationMs,
-          });
-        } catch (completeErr) {
-          jobLog.error(
-            {
-              runId,
-              err: completeErr instanceof Error ? completeErr.message : String(completeErr),
-            },
-            "failed to record job failure",
-          );
-        }
-      }
-    } finally {
-      // Remove from active set
-      activeJobs.delete(jobId);
-
-      // 5. Always advance the schedule pointer (even on failure)
-      try {
-        await advanceSchedulePointer(job);
-      } catch (err) {
-        jobLog.error(
-          { err: err instanceof Error ? err.message : String(err) },
-          "failed to advance schedule pointer",
-        );
-      }
-    }
-  }
-
-  // -----------------------------------------------------------------------
-  // Core: manual trigger
-  // -----------------------------------------------------------------------
-
-  async function triggerJob(
-    jobId: string,
-    trigger: "manual" | "retry" = "manual",
-  ): Promise<TriggerJobResult> {
-    const job = await jobStore.getJobById(jobId);
-    if (!job) {
-      throw new Error(`Job not found: ${jobId}`);
-    }
-
-    if (job.status !== "active") {
-      throw new Error(
-        `Job "${job.jobKey}" is not active (status: ${job.status})`,
-      );
-    }
-
-    // Overlap prevention
-    if (activeJobs.has(jobId)) {
-      throw new Error(
-        `Job "${job.jobKey}" is already running — cannot trigger while in progress`,
-      );
-    }
-
-    // Also check DB for running runs (defensive — covers multi-instance)
-    const existingRuns = await db
-      .select()
-      .from(pluginJobRuns)
-      .where(
-        and(
-          eq(pluginJobRuns.jobId, jobId),
-          eq(pluginJobRuns.status, "running"),
-        ),
-      );
-
-    if (existingRuns.length > 0) {
-      throw new Error(
-        `Job "${job.jobKey}" already has a running execution — cannot trigger while in progress`,
-      );
-    }
-
-    // Check worker availability
-    if (!workerManager.isRunning(job.pluginId)) {
-      throw new Error(
-        `Worker for plugin "${job.pluginId}" is not running — cannot trigger job`,
-      );
-    }
-
-    // Create the run and dispatch (non-blocking)
-    const run = await jobStore.createRun({
-      jobId,
-      pluginId: job.pluginId,
-      trigger,
-    });
-
-    // Dispatch in background — don't block the caller
-    void dispatchManualRun(job, run.id, trigger);
-
-    return { runId: run.id, jobId };
-  }
-
-  /**
-   * Dispatch a manually triggered job run.
-   */
-  async function dispatchManualRun(
-    job: typeof pluginJobs.$inferSelect,
-    runId: string,
-    trigger: "manual" | "retry",
-  ): Promise<void> {
-    const { id: jobId, pluginId, jobKey } = job;
-    const jobLog = log.child({ jobId, pluginId, jobKey, runId, trigger });
-
-    activeJobs.add(jobId);
-    const startedAt = Date.now();
-
     try {
       await jobStore.markRunning(runId);
-
+      await workerManager.call(job.pluginId, "runJob", {
-      await workerManager.call(
+        job: { jobKey: job.jobKey, runId, trigger,
-        pluginId,
+          scheduledAt: (trigger === "schedule" ? job.nextRunAt! : new Date()).toISOString() },
-        "runJob",
+      }, jobTimeoutMs);
-        {
-          job: {
-            jobKey,
-            runId,
-            trigger,
-            scheduledAt: new Date().toISOString(),
-          },
-        },
-        jobTimeoutMs,
-      );
-
-      const durationMs = Date.now() - startedAt;
-      await jobStore.completeRun(runId, {
-        status: "succeeded",
-        durationMs,
-      });
-
-      jobLog.info({ durationMs }, "manual job completed successfully");
     } catch (err) {
-      const durationMs = Date.now() - startedAt;
+      error = err instanceof Error ? err.message : String(err);
-      const errorMessage = err instanceof Error ? err.message : String(err);
+      log.error({ jobId: job.id, runId, err }, "plugin job failed");
-      jobLog.error({ durationMs, err: errorMessage }, "manual job failed");
-
-      try {
-        await jobStore.completeRun(runId, {
-          status: "failed",
-          error: errorMessage,
-          durationMs,
-        });
-      } catch (completeErr) {
-        jobLog.error(
-          {
-            err: completeErr instanceof Error ? completeErr.message : String(completeErr),
-          },
-          "failed to record manual job failure",
-        );
-      }
-    } finally {
-      activeJobs.delete(jobId);
     }
+    const finishedAt = new Date();
+    pendingSettlements.set(job.id, { persist: () => db.transaction(async tx => {
+      const store = pluginJobStore(tx);
+      await store.completeRun(runId, {
+        status: error === undefined ? "succeeded" : "failed", error,
+        durationMs: finishedAt.getTime() - startedAt,
+      });
+      if (trigger === "schedule") {
+        const [current] = await tx.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)).for("update");
+        if (current && current.schedule === job.schedule && current.nextRunAt?.getTime() === job.nextRunAt?.getTime()) {
+          await store.updateRunTimestamps(job.id, finishedAt, nextCron(current.schedule, finishedAt));
+        }
+      }
+    }) });
+    try { await settle(job.id); }
+    finally { activeJobs.delete(job.id); worker?.wake(); }
   }
 
-  // -----------------------------------------------------------------------
+  async function dispatch(job: Job, trigger: Trigger, signal?: AbortSignal): Promise<TriggerJobResult | null> {
-  // Schedule pointer management
+    activeJobs.add(job.id);
-  // -----------------------------------------------------------------------
+    try {
+      const run = await db.transaction(async tx => {
+        const [current] = await tx.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)).for("update");
+        if (!current || current.status !== "active" || signal?.aborted || !canRun() || unregistered.has(job.pluginId)
+          || !workerManager.isRunning(job.pluginId)) return null;
+        if (trigger === "schedule" && (current.schedule !== job.schedule
+          || current.nextRunAt?.getTime() !== job.nextRunAt?.getTime())) return null;
+        const [outstanding] = await tx.select({ id: pluginJobRuns.id }).from(pluginJobRuns).where(and(
+          eq(pluginJobRuns.jobId, job.id), inArray(pluginJobRuns.status, ["pending", "queued", "running"]),
+        )).limit(1);
+        if (outstanding) return null;
+        return pluginJobStore(tx).createRun({ jobId: job.id, pluginId: job.pluginId, trigger });
+      });
+      if (!run) {
+        activeJobs.delete(job.id);
+        worker?.wake();
+        if (trigger !== "schedule") throw new Error(`Job "${job.jobKey}" is no longer available`);
+        return null;
+      }
+      const completion = track(execute(job, run.id, trigger));
+      void completion.catch(err => { log.error({ err, jobId: job.id }, "plugin job settlement deferred"); worker?.wake(); });
+      return { runId: run.id, jobId: job.id };
+    } catch (err) { activeJobs.delete(job.id); throw err; }
+  }
 
-  /**
+  async function refresh(signal?: AbortSignal) {
-   * Advance the `lastRunAt` and `nextRunAt` timestamps on a job after a run.
+    for (const jobId of pendingSettlements.keys()) await settle(jobId);
-   */
+    const jobs = await db.select().from(pluginJobs).where(eq(pluginJobs.status, "active"));
-  async function advanceSchedulePointer(
+    const runnable = jobs.filter(job => !unregistered.has(job.pluginId) && workerManager.isRunning(job.pluginId)
-    job: typeof pluginJobs.$inferSelect,
+      && !activeJobs.has(job.id) && job.schedule && !validateCron(job.schedule));
-  ): Promise<void> {
+    const outstanding = runnable.length === 0 ? [] : await db.select({ jobId: pluginJobRuns.jobId })
+      .from(pluginJobRuns).where(and(inArray(pluginJobRuns.jobId, runnable.map(job => job.id)),
+        inArray(pluginJobRuns.status, ["pending", "queued", "running"])));
+    const blocked = new Set(outstanding.map(run => run.jobId));
     const now = new Date();
-    let nextRunAt: Date | null = null;
+    const dispatches: Promise<unknown>[] = [];
-
+    next = null;
-    if (job.schedule) {
+    for (const job of runnable) {
-      const validationError = validateCron(job.schedule);
+      if (!canRun() || signal?.aborted) break;
-      if (validationError) {
+      if (blocked.has(job.id)) continue;
-        log.warn(
+      if (!job.nextRunAt) {
-          { jobId: job.id, schedule: job.schedule, error: validationError },
+        const deadline = nextCron(job.schedule, now);
-          "invalid cron schedule — cannot compute next run",
+        if (!deadline) continue;
-        );
+        await db.transaction(async tx => {
-      } else {
+          const [current] = await tx.select().from(pluginJobs).where(eq(pluginJobs.id, job.id)).for("update");
-        const cron = parseCron(job.schedule);
+          if (current && current.nextRunAt === null && current.schedule === job.schedule && current.status === "active") {
-        nextRunAt = nextCronTick(cron, now);
+            await pluginJobStore(tx).updateRunTimestamps(job.id, current.lastRunAt ?? new Date(0), deadline);
-      }
+          }
-    }
-
-    await jobStore.updateRunTimestamps(job.id, now, nextRunAt);
-  }
-
-  /**
-   * Ensure all active jobs for a plugin have a `nextRunAt` value.
-   * Called when a plugin is registered with the scheduler.
-   */
-  async function ensureNextRunTimestamps(pluginId: string): Promise<void> {
-    const jobs = await jobStore.listJobs(pluginId, "active");
-
-    for (const job of jobs) {
-      // Skip jobs that already have a valid nextRunAt in the future
-      if (job.nextRunAt && job.nextRunAt.getTime() > Date.now()) {
-        continue;
-      }
-
-      // Skip jobs without a schedule
-      if (!job.schedule) {
-        continue;
-      }
-
-      const validationError = validateCron(job.schedule);
-      if (validationError) {
-        log.warn(
-          { jobId: job.id, jobKey: job.jobKey, schedule: job.schedule, error: validationError },
-          "skipping job with invalid cron schedule",
-        );
-        continue;
-      }
-
-      const cron = parseCron(job.schedule);
-      const nextRunAt = nextCronTick(cron, new Date());
-
-      if (nextRunAt) {
-        await jobStore.updateRunTimestamps(
-          job.id,
-          job.lastRunAt ?? new Date(0),
-          nextRunAt,
-        );
-        log.debug(
-          { jobId: job.id, jobKey: job.jobKey, nextRunAt: nextRunAt.toISOString() },
-          "computed nextRunAt for job",
-        );
-      }
-    }
-  }
-
-  // -----------------------------------------------------------------------
-  // Plugin registration
-  // -----------------------------------------------------------------------
-
-  async function registerPlugin(pluginId: string): Promise<void> {
-    log.info({ pluginId }, "registering plugin with job scheduler");
-    await ensureNextRunTimestamps(pluginId);
-  }
-
-  async function unregisterPlugin(pluginId: string): Promise<void> {
-    log.info({ pluginId }, "unregistering plugin from job scheduler");
-
-    // Cancel any in-flight run records for this plugin that are still
-    // queued or running. Active jobs in-memory will finish naturally.
-    try {
-      const runningRuns = await db
-        .select()
-        .from(pluginJobRuns)
-        .where(
-          and(
-            eq(pluginJobRuns.pluginId, pluginId),
-            or(
-              eq(pluginJobRuns.status, "running"),
-              eq(pluginJobRuns.status, "queued"),
-            ),
-          ),
-        );
-
-      for (const run of runningRuns) {
-        await jobStore.completeRun(run.id, {
-          status: "cancelled",
-          error: "Plugin unregistered",
-          durationMs: run.startedAt
-            ? Date.now() - run.startedAt.getTime()
-            : null,
         });
+        job.nextRunAt = deadline;
+      }
+      const at = job.nextRunAt.getTime();
+      if (at <= now.getTime() && activeJobs.size < maxConcurrentJobs) {
+        dispatches.push(dispatch(job, "schedule", signal));
+      } else if (activeJobs.size < maxConcurrentJobs) {
+        next = next === null ? at : Math.min(next, at);
       }
-    } catch (err) {
-      log.error(
-        {
-          pluginId,
-          err: err instanceof Error ? err.message : String(err),
-        },
-        "error cancelling in-flight runs during unregister",
-      );
-    }
-
-    // Remove any active tracking for jobs owned by this plugin
-    const jobs = await jobStore.listJobs(pluginId);
-    for (const job of jobs) {
-      activeJobs.delete(job.id);
     }
+    const results = await Promise.allSettled(dispatches);
+    const failed = results.find(result => result.status === "rejected");
+    if (failed?.status === "rejected") throw failed.reason;
   }
 
-  // -----------------------------------------------------------------------
+  function tick(signal?: AbortSignal): Promise<void> {
-  // Lifecycle: start / stop
+    if (!canRun()) return Promise.resolve();
-  // -----------------------------------------------------------------------
+    if (ticking) return ticking;
-
+    tickCount++; lastTickAt = new Date();
-  function start(): void {
+    ticking = refresh(signal).finally(() => { ticking = undefined; });
-    if (running) {
+    return ticking;
-      log.debug("scheduler already running");
-      return;
-    }
-
-    running = true;
-    tickTimer = setInterval(() => {
-      void tick();
-    }, tickIntervalMs);
-
-    log.info(
-      { tickIntervalMs, maxConcurrentJobs },
-      "plugin job scheduler started",
-    );
   }
 
-  function stop(): void {
-    // Always clear the timer defensively, even if `running` is already false,
-    // to prevent leaked interval timers.
-    if (tickTimer !== null) {
-      clearInterval(tickTimer);
-      tickTimer = null;
-    }
-
-    if (!running) return;
-    running = false;
-
-    log.info(
-      { activeJobCount: activeJobs.size },
-      "plugin job scheduler stopped",
-    );
-  }
-
-  // -----------------------------------------------------------------------
-  // Diagnostics
-  // -----------------------------------------------------------------------
-
-  function diagnostics(): SchedulerDiagnostics {
-    return {
-      running,
-      activeJobCount: activeJobs.size,
-      activeJobIds: [...activeJobs],
-      tickCount,
-      lastTickAt: lastTickAt?.toISOString() ?? null,
-    };
-  }
-
-  // -----------------------------------------------------------------------
-  // Public API
-  // -----------------------------------------------------------------------
-
   return {
-    start,
+    async start(coordinator) {
-    stop,
+      if (worker) { await worker.ready; return; }
-    registerPlugin,
+      running = true;
-    unregisterPlugin,
+      worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.pluginSchedules, {
-    triggerJob,
+        run: tick, next: async () => next, canRun, retryMs: tickIntervalMs,
+      });
+      await worker.ready;
+    },
+    async stop() {
+      running = false;
+      for (const unsubscribe of subscriptions.values()) unsubscribe();
+      subscriptions.clear();
+      await Promise.allSettled([ticking]);
+      while (inFlight.size) await Promise.allSettled([...inFlight]);
+    },
+    async registerPlugin(pluginId) {
+      unregistered.delete(pluginId);
+      subscriptions.get(pluginId)?.();
+      const handle = workerManager.getWorker(pluginId);
+      if (handle) {
+        const wake = () => worker?.wake();
+        handle.on("status", wake);
+        subscriptions.set(pluginId, () => handle.off("status", wake));
+      }
+      worker?.wake();
+    },
+    async unregisterPlugin(pluginId) {
+      unregistered.add(pluginId);
+      subscriptions.get(pluginId)?.(); subscriptions.delete(pluginId);
+      worker?.wake();
+      const runs = await db.select().from(pluginJobRuns).where(and(
+        eq(pluginJobRuns.pluginId, pluginId), inArray(pluginJobRuns.status, ["pending", "queued", "running"]),
+      ));
+      for (const run of runs) {
+        if (!activeJobs.has(run.jobId) && ["running", "queued", "pending"].includes(run.status)) {
+          await jobStore.completeRun(run.id, { status: "cancelled", error: "Plugin unregistered",
+            durationMs: run.startedAt ? Date.now() - run.startedAt.getTime() : null });
+        }
+      }
+    },
+    triggerJob(jobId, trigger = "manual") {
+      return track((async () => {
+        if (!canRun()) throw new Error("Plugin job scheduler is not accepting work");
+        const job = await jobStore.getJobById(jobId);
+        if (!job) throw new Error(`Job not found: ${jobId}`);
+        if (job.status !== "active") throw new Error(`Job "${job.jobKey}" is not active (status: ${job.status})`);
+        if (activeJobs.has(jobId) || pendingSettlements.has(jobId)) throw new Error(`Job "${job.jobKey}" is already running`);
+        activeJobs.add(jobId);
+        try {
+          const runs = await db.select({ id: pluginJobRuns.id }).from(pluginJobRuns).where(and(
+            eq(pluginJobRuns.jobId, jobId), inArray(pluginJobRuns.status, ["pending", "queued", "running"]),
+          )).limit(1);
+          if (runs.length) throw new Error(`Job "${job.jobKey}" already has a running execution`);
+          if (unregistered.has(job.pluginId) || !workerManager.isRunning(job.pluginId)) throw new Error(`Worker for plugin "${job.pluginId}" is not running`);
+          const result = await dispatch(job, trigger);
+          if (!result) throw new Error(`Job "${job.jobKey}" is no longer available`);
+          return result;
+        } catch (err) { activeJobs.delete(jobId); throw err; }
+      })());
+    },
     tick,
-    diagnostics,
+    diagnostics: () => ({ running, activeJobCount: activeJobs.size, activeJobIds: [...activeJobs], tickCount,
+      lastTickAt: lastTickAt?.toISOString() ?? null }),
   };
 }

server/src/services/plugin-job-store.ts

@@ -1,3 +1,4 @@
+import { writePluginJobChange, type PluginJobDatabase } from "./plugin-job-work-notifications.js";
 /**
  * Plugin Job Store — persistence layer for scheduled plugin jobs and their
  * execution history.
@@ -100,12 +101,12 @@ export interface CompleteJobRunInput {
  * });
  * ```
  */
-export function pluginJobStore(db: Db) {
+export function pluginJobStore(db: PluginJobDatabase) {
   // -----------------------------------------------------------------------
   // Internal helpers
   // -----------------------------------------------------------------------
 
-  async function assertPluginExists(pluginId: string): Promise<void> {
+  async function assertPluginExists(db: PluginJobDatabase, pluginId: string): Promise<void> {
     const rows = await db
       .select({ id: plugins.id })
       .from(plugins)
@@ -145,63 +146,66 @@ export function pluginJobStore(db: Db) {
       pluginId: string,
       declarations: PluginJobDeclaration[],
     ): Promise<void> {
-      await assertPluginExists(pluginId);
+      return writePluginJobChange(db, async db => {
+        await assertPluginExists(db, pluginId);
 
-      // Fetch existing jobs for this plugin
+        // Fetch existing jobs for this plugin
-      const existingJobs = await db
+        const existingJobs = await db
-        .select()
+          .select()
-        .from(pluginJobs)
+          .from(pluginJobs)
-        .where(eq(pluginJobs.pluginId, pluginId));
+          .where(eq(pluginJobs.pluginId, pluginId));
 
-      const existingByKey = new Map(
+        const existingByKey = new Map(
-        existingJobs.map((j) => [j.jobKey, j]),
+          existingJobs.map((j) => [j.jobKey, j]),
-      );
+        );
 
-      const declaredKeys = new Set<string>();
+        const declaredKeys = new Set<string>();
 
-      // Upsert each declared job
+        // Upsert each declared job
-      for (const decl of declarations) {
+        for (const decl of declarations) {
-        declaredKeys.add(decl.jobKey);
+          declaredKeys.add(decl.jobKey);
 
-        const existing = existingByKey.get(decl.jobKey);
+          const existing = existingByKey.get(decl.jobKey);
-        const schedule = decl.schedule ?? "";
+          const schedule = decl.schedule ?? "";
 
-        if (existing) {
+          if (existing) {
-          // Update schedule if it changed; re-activate if it was paused
+            // Update schedule if it changed; re-activate if it was paused
-          const updates: Record<string, unknown> = {
+            const updates: Record<string, unknown> = {
-            updatedAt: new Date(),
+              updatedAt: new Date(),
-          };
+            };
-          if (existing.schedule !== schedule) {
+            if (existing.schedule !== schedule) {
-            updates.schedule = schedule;
+              updates.schedule = schedule;
+              updates.nextRunAt = null;
+            }
+            if (existing.status === "paused") {
+              updates.status = "active";
+            }
+
+            await db
+              .update(pluginJobs)
+              .set(updates)
+              .where(eq(pluginJobs.id, existing.id));
+          } else {
+            // Insert new job
+            await db.insert(pluginJobs).values({
+              pluginId,
+              jobKey: decl.jobKey,
+              schedule,
+              status: "active",
+            });
           }
-          if (existing.status === "paused") {
+        }
-            updates.status = "active";
+
+        // Pause jobs that are no longer declared in the manifest
+        for (const existing of existingJobs) {
+          if (!declaredKeys.has(existing.jobKey) && existing.status !== "paused") {
+            await db
+              .update(pluginJobs)
+              .set({ status: "paused", updatedAt: new Date() })
+              .where(eq(pluginJobs.id, existing.id));
           }
-
-          await db
-            .update(pluginJobs)
-            .set(updates)
-            .where(eq(pluginJobs.id, existing.id));
-        } else {
-          // Insert new job
-          await db.insert(pluginJobs).values({
-            pluginId,
-            jobKey: decl.jobKey,
-            schedule,
-            status: "active",
-          });
         }
-      }
+      });
-
-      // Pause jobs that are no longer declared in the manifest
-      for (const existing of existingJobs) {
-        if (!declaredKeys.has(existing.jobKey) && existing.status !== "paused") {
-          await db
-            .update(pluginJobs)
-            .set({ status: "paused", updatedAt: new Date() })
-            .where(eq(pluginJobs.id, existing.id));
-        }
-      }
     },
 
     /**
@@ -290,10 +294,12 @@ export function pluginJobStore(db: Db) {
       jobId: string,
       status: JobDefinitionStatus,
     ): Promise<void> {
-      await db
+      return writePluginJobChange(db, async db => {
-        .update(pluginJobs)
+        await db
-        .set({ status, updatedAt: new Date() })
+          .update(pluginJobs)
-        .where(eq(pluginJobs.id, jobId));
+          .set({ status, updatedAt: new Date() })
+          .where(eq(pluginJobs.id, jobId));
+      });
     },
 
     /**
@@ -311,14 +317,16 @@ export function pluginJobStore(db: Db) {
       lastRunAt: Date,
       nextRunAt: Date | null,
     ): Promise<void> {
-      await db
+      return writePluginJobChange(db, async db => {
-        .update(pluginJobs)
+        await db
-        .set({
+          .update(pluginJobs)
-          lastRunAt,
+          .set({
-          nextRunAt,
+            lastRunAt,
-          updatedAt: new Date(),
+            nextRunAt,
-        })
+            updatedAt: new Date(),
-        .where(eq(pluginJobs.id, jobId));
+          })
+          .where(eq(pluginJobs.id, jobId));
+      });
     },
 
     /**
@@ -329,9 +337,11 @@ export function pluginJobStore(db: Db) {
      * @param pluginId - UUID of the owning plugin
      */
     async deleteAllJobs(pluginId: string): Promise<void> {
-      await db
+      return writePluginJobChange(db, async db => {
-        .delete(pluginJobs)
+        await db
-        .where(eq(pluginJobs.pluginId, pluginId));
+          .delete(pluginJobs)
+          .where(eq(pluginJobs.pluginId, pluginId));
+      });
     },
 
     // =====================================================================
@@ -351,17 +361,19 @@ export function pluginJobStore(db: Db) {
     async createRun(
       input: CreateJobRunInput,
     ): Promise<typeof pluginJobRuns.$inferSelect> {
-      const rows = await db
+      return writePluginJobChange(db, async db => {
-        .insert(pluginJobRuns)
+        const rows = await db
-        .values({
+          .insert(pluginJobRuns)
-          jobId: input.jobId,
+          .values({
-          pluginId: input.pluginId,
+            jobId: input.jobId,
-          trigger: input.trigger,
+            pluginId: input.pluginId,
-          status: "queued",
+            trigger: input.trigger,
-        })
+            status: "queued",
-        .returning();
+          })
+          .returning();
 
-      return rows[0]!;
+        return rows[0]!;
+      });
     },
 
     /**
@@ -370,13 +382,15 @@ export function pluginJobStore(db: Db) {
      * @param runId - UUID of the run row
      */
     async markRunning(runId: string): Promise<void> {
-      await db
+      return writePluginJobChange(db, async db => {
-        .update(pluginJobRuns)
+        await db
-        .set({
+          .update(pluginJobRuns)
-          status: "running" as PluginJobRunStatus,
+          .set({
-          startedAt: new Date(),
+            status: "running" as PluginJobRunStatus,
-        })
+            startedAt: new Date(),
-        .where(eq(pluginJobRuns.id, runId));
+          })
+          .where(eq(pluginJobRuns.id, runId));
+      });
     },
 
     /**
@@ -390,15 +404,17 @@ export function pluginJobStore(db: Db) {
       runId: string,
       input: CompleteJobRunInput,
     ): Promise<void> {
-      await db
+      return writePluginJobChange(db, async db => {
-        .update(pluginJobRuns)
+        await db
-        .set({
+          .update(pluginJobRuns)
-          status: input.status,
+          .set({
-          error: input.error ?? null,
+            status: input.status,
-          durationMs: input.durationMs ?? null,
+            error: input.error ?? null,
-          finishedAt: new Date(),
+            durationMs: input.durationMs ?? null,
-        })
+            finishedAt: new Date(),
-        .where(eq(pluginJobRuns.id, runId));
+          })
+          .where(eq(pluginJobRuns.id, runId));
+      });
     },
 
     /**

server/src/services/plugin-job-work-notifications.ts

@@ -0,0 +1,11 @@
+import type { Db } from "@paperclipai/db";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type PluginJobDatabase = Pick<Db, "transaction" | "select" | "update" | "insert" | "delete">;
+
+export function writePluginJobChange<T>(db: PluginJobDatabase, write: (tx: PluginJobDatabase) => Promise<T>): Promise<T> {
+  return db.transaction(async tx => {
+    await notifyDeliveryWork(tx, DELIVERY_QUEUES.pluginSchedules);
+    return write(tx);
+  });
+}

server/src/services/plugin-loader.ts

@@ -2144,7 +2144,7 @@ export function pluginLoader(
       const { workerManager, jobScheduler } = runtimeServices;
 
       // 1. Stop the job scheduler tick loop
-      jobScheduler.stop();
+      await jobScheduler.stop();
 
       // 2. Stop all worker processes
       await workerManager.stopAll();
@@ -2416,9 +2416,9 @@ export function pluginLoader(
       // 6. Sync job declarations and register with scheduler
       // ------------------------------------------------------------------
       const jobDeclarations = manifest.jobs ?? [];
+      await jobStore.syncJobDeclarations(pluginId, jobDeclarations);
+      await jobScheduler.registerPlugin(pluginId);
       if (jobDeclarations.length > 0) {
-        await jobStore.syncJobDeclarations(pluginId, jobDeclarations);
-        await jobScheduler.registerPlugin(pluginId);
         registered.jobs = jobDeclarations.length;
 
         log.info(

server/src/services/plugin-registry.ts

@@ -1,3 +1,4 @@
+import { writePluginJobChange } from "./plugin-job-work-notifications.js";
 import { asc, eq, isNull, ne, sql, and } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
@@ -267,25 +268,27 @@ export function pluginRegistryService(db: Db) {
       const plugin = await getById(id);
       if (!plugin) throw notFound("Plugin not found");
 
-      if (removeData) {
+      return writePluginJobChange(db, async db => {
-        // Hard delete – plugin_config cascades via FK onDelete
+        if (removeData) {
+          // Hard delete – plugin_config cascades via FK onDelete
+          return db
+            .delete(plugins)
+            .where(eq(plugins.id, id))
+            .returning()
+            .then((rows) => rows[0] ?? null);
+        }
+
+        // Soft delete – mark as uninstalled
         return db
-          .delete(plugins)
+          .update(plugins)
+          .set({
+            status: "uninstalled" as PluginStatus,
+            updatedAt: new Date(),
+          })
           .where(eq(plugins.id, id))
           .returning()
           .then((rows) => rows[0] ?? null);
-      }
+      });
-
-      // Soft delete – mark as uninstalled
-      return db
-        .update(plugins)
-        .set({
-          status: "uninstalled" as PluginStatus,
-          updatedAt: new Date(),
-        })
-        .where(eq(plugins.id, id))
-        .returning()
-        .then((rows) => rows[0] ?? null);
     },
 
     // ----- Config ---------------------------------------------------------
@@ -670,35 +673,38 @@ export function pluginRegistryService(db: Db) {
       jobKey: string,
       input: { schedule: string; status?: PluginJobStatus },
     ) => {
-      const existing = await db
+      return writePluginJobChange(db, async db => {
-        .select()
+        const existing = await db
-        .from(pluginJobs)
+          .select()
-        .where(and(eq(pluginJobs.pluginId, pluginId), eq(pluginJobs.jobKey, jobKey)))
+          .from(pluginJobs)
-        .then((rows) => rows[0] ?? null);
+          .where(and(eq(pluginJobs.pluginId, pluginId), eq(pluginJobs.jobKey, jobKey)))
+          .then((rows) => rows[0] ?? null);
+
+        if (existing) {
+          return db
+            .update(pluginJobs)
+            .set({
+              schedule: input.schedule,
+              nextRunAt: existing.schedule === input.schedule ? existing.nextRunAt : null,
+              status: input.status ?? existing.status,
+              updatedAt: new Date(),
+            })
+            .where(eq(pluginJobs.id, existing.id))
+            .returning()
+            .then((rows) => rows[0]);
+        }
 
-      if (existing) {
         return db
-          .update(pluginJobs)
+          .insert(pluginJobs)
-          .set({
+          .values({
+            pluginId,
+            jobKey,
             schedule: input.schedule,
-            status: input.status ?? existing.status,
+            status: input.status ?? "active",
-            updatedAt: new Date(),
           })
-          .where(eq(pluginJobs.id, existing.id))
           .returning()
           .then((rows) => rows[0]);
-      }
+      });
-
-      return db
-        .insert(pluginJobs)
-        .values({
-          pluginId,
-          jobKey,
-          schedule: input.schedule,
-          status: input.status ?? "active",
-        })
-        .returning()
-        .then((rows) => rows[0]);
     },
 
     /**
@@ -721,17 +727,19 @@ export function pluginRegistryService(db: Db) {
       trigger: PluginJobRunTrigger,
       companyId: string | null,
     ) => {
-      return db
+      return writePluginJobChange(db, async db => {
-        .insert(pluginJobRuns)
+        return db
-        .values({
+          .insert(pluginJobRuns)
-          pluginId,
+          .values({
-          jobId,
+            pluginId,
-          companyId,
+            jobId,
-          trigger,
+            companyId,
-          status: "pending",
+            trigger,
-        })
+            status: "pending",
-        .returning()
+          })
-        .then((rows) => rows[0]);
+          .returning()
+          .then((rows) => rows[0]);
+      });
     },
 
     /**
@@ -752,12 +760,14 @@ export function pluginRegistryService(db: Db) {
         finishedAt?: Date;
       },
     ) => {
-      return db
+      return writePluginJobChange(db, async db => {
-        .update(pluginJobRuns)
+        return db
-        .set(input)
+          .update(pluginJobRuns)
-        .where(eq(pluginJobRuns.id, runId))
+          .set(input)
-        .returning()
+          .where(eq(pluginJobRuns.id, runId))
-        .then((rows) => rows[0] ?? null);
+          .returning()
+          .then((rows) => rows[0] ?? null);
+      });
     },
 
     // ----- Webhooks -------------------------------------------------------