diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 6310c54c46e..376f551f490 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -28,7 +28,7 @@ Do not mark an item done from a timer removal or a green check alone. | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | | 11 | Status-card evaluation | Settings/card mutations and evaluation deadlines | Pending | | 12 | Plugin scheduled jobs | Registration/update/removal and actual job deadlines | #17518 published; see latest-head gate records below | -| 13 | Routine, agent-timer and issue-monitor scheduling | Remove periodic wake backstops after complete producer coverage; routine execution-status reconciliation separately | Pending | +| 13 | Routine, agent-timer and issue-monitor scheduling | Remove periodic wake backstops after complete producer coverage; routine execution-status reconciliation separately | Routine schedules implemented in slice 61; issue/agent timers still pending | | 14 | Skill-version retention and native-harness backup expiration | Separate retention owners and existing retention guarantees | Pending | | 15 | Automatic database backups | Shared scheduling, actual backup deadlines, existing checkpoint contract | Slice 60 implemented; remote acceptance pending; see `2026-10-26-database-backup-sleep-schedule.md` | | 16 | Aggregate maintenance tick | Retire enabled and disabled branches after all constituent owners migrate | Pending | @@ -3489,3 +3489,23 @@ Full local test:run and remote acceptance remain outstanding. See the dedicated [backup scheduling audit](2026-10-26-database-backup-sleep-schedule.md) for limits. Skill retention, native harness expiration, and the other original inventory duties remain open; this is not completion of the full stack request. + + +## Slice 61: routine schedules + +Routine schedules now use their actual saved next-run timestamp and committed +producer changes without the periodic heartbeat wake. The owner includes overdue +catch-up debt, shares eligibility between dispatch and deadline reads, and stops +between claimed trigger batches on drain/shutdown. Added producer coverage includes +manual invocation of a schedule, company lifecycle and deletion, and budget-driven +company pause/resume. Existing cron behavior and Cloud sleep proofs stay with their +owners. Issue-monitor and legacy-agent-timer polling remain separate pending duties. + +Two original regressions and two budget-producer regressions failed before their +fixes. Final acceptance passed 224 tests in eight complete suites, workspace +typechecking, and the full build. Embedded PostgreSQL setup timed out, and the old +external disposable cluster later lost a catalog file; a fresh isolated PostgreSQL +cluster ran the same acceptance suites. No further IPC cleanup was performed. +Remote checks/review and full-stack acceptance remain pending. See the +[routine scheduling audit](2026-10-27-routine-schedule-sleep.md) for producer coverage, +lifecycle evidence, harness corrections and the supported single-process writer model. diff --git a/doc/plans/2026-10-27-routine-schedule-sleep.md b/doc/plans/2026-10-27-routine-schedule-sleep.md new file mode 100644 index 00000000000..59b2b51ef9d --- /dev/null +++ b/doc/plans/2026-10-27-routine-schedule-sleep.md @@ -0,0 +1,180 @@ +# Routine schedules without the heartbeat poll + +## Scope and baseline + +This is slice 61 of the recurring-server migration stack. Its parent is +`026fdef06a377f059fdb14bc80da1e906067bbb5` (PR #17836). Inspected +`origin/master` at `0cb3fc7db591ba23e21fbbc70042bc44ec7d6f20`; the routine owner +and schedule registrar have no newer upstream changes. The broader stack still +needs master reconciliation and remote acceptance. + +A heartbeat used to wake the routine worker on every tick. The worker therefore +queried schedules even when no routine owed work. It already registered future +deadlines, but its query excluded overdue rows. A capped catch-up batch could +leave more ticks due and depend on the next heartbeat to find them. + +This slice removes only that routine wake. Issue monitors, legacy agent timers, +other heartbeat duties, and the rest of the inventory remain separate work. + +## Obligation and owner + +- Trigger: startup/resume, an eligible cron deadline, or a committed producer change. +- Durable truth: the existing routine, trigger, and company rows. No new table. +- Eligibility: schedule triggers that are enabled, not archived, have saved cron, + timezone and next-run values, and belong to an active routine in an active + company. Trigger/routine company IDs must agree. +- Next deadline: the earliest eligible saved next-run timestamp, including overdue + timestamps left by capped catch-up. The owner returns a number or null. +- Retry: the configured heartbeat interval only after an error or while overdue + work remains. Successfully read future deadlines retain their exact timestamp. +- Disarm: no eligible schedule and no unresolved producer notification. + +`routineService` owns eligibility, database reads and cron dispatch. Its due and +next-deadline queries share the same eligibility predicate. The existing +`registerDeadlineWork` helper owns request generations, retries and lifecycle +behavior. The schedule registrar delegates to it. Its coordinator argument type +now requires only `register`, the only operation that helper uses; other consumers +keep the same runtime behavior. Server startup supplies the owner callbacks and +admission gate, then tracks readiness and shutdown through the existing owner. + +The next-deadline read orders by `next_run_at` and returns one row. The schema +already has `routine_triggers_next_run_idx`. The existing due dispatcher is not +redesigned or replaced with a new queue. + +## Producer audit + +| Writer | Wake coverage | +| --- | --- | +| Routine create/update, including status | Existing transaction-aware routine signal | +| Trigger create/update/delete | Existing transaction-aware routine signal | +| Revision restore and recreated triggers | Existing transaction-aware routine signal | +| Company import and built-in routines | Use routine-service create/update/trigger APIs; separate direct writes change metadata only | +| Manual run naming a schedule | New signal before the transaction changes its next-run timestamp | +| Company pause/reactivation/archive | New signal in the existing company transaction | +| Budget-induced company pause/resume | New signal in the existing accounting transaction | +| Company deletion | New signal before deleting routine data in the company transaction | +| Scheduler's claimed next-run update | Worker reloads after its own pass; no self-notification per catch-up run | +| Folder moves, revision pointers, webhook secret/delivery metadata | Do not change schedule eligibility or deadlines | + +Ordinary API, MCP, plugin and import callers use these owning services. The Grok +bot uses a different table and has its own migration. Company name/prefix repairs, +budget projections, issue counters and feedback settings do not change company +status. Process-local signals do not support an independent SQL writer or another +replica that bypasses these producers. + +## Preserved behavior and explicit differences + +Cron/timezone calculation, sub-hourly coalescing, the 25-run catch-up cap, +compare-and-set claims, activity gates, concurrency policies and suppressed-run +records stay with the existing dispatcher. Capped overdue work gets a retained +retry instead of becoming invisible. A paused company now matches the active +company eligibility already used by the old deadline query; due triggers remain +saved until company reactivation. Routine reactivation still uses its existing +next-cron reset policy. + +Project pause and worktree execution suppression still advance cron ticks and +record skipped runs. Those configured ticks remain real timed obligations. This +slice does not claim those cases require zero SQL forever. + +Drain/shutdown can stop between trigger candidates. Once a trigger's next-run +value is claimed, its admitted batch finishes; interrupting midway would abandon +runs whose timestamp was already advanced. Unclaimed trigger rows remain durable +for resume or restart. The shared coordinator joins in-flight work and cancels +subscriptions/deadlines on shutdown. + +Notifications attach to outer transaction settlement. Nested savepoint release +is not commit. Rollback removes the intent, and changes during a sweep retain a +new generation. Drain keeps deferred requests in memory without passive SQL. +Startup reconstructs the actual next deadline from the database. + +## Verification and observed failures + +The initial regression run failed both intended assertions: a 27-tick hourly +backlog lost its deadline after the first 25 runs, and company pause left the +old deadline armed. Both pass after the migration. Two additional budget tests +reproduced missing pause/resume notifications before the budget producer fix. + +The new integration suite checks no resources, an active routine with no trigger, +future configured work, retained run history, first producer, earlier replacement, +disable/archive/delete/restore, outer commit/rollback, manual invocation, company +lifecycle, budget lifecycle, restart catch-up, drain/resume, failed deadline reads, +concurrent mutation during a read, and interruption between claimed batches. +Quiet assertions observe database select/execute/transaction calls over former +poll periods. Existing cron, project pause, activity gate and coalescing tests +remain in the full owner acceptance set. Startup tests assert that the routine +worker's wake is not called by the recurring heartbeat. + +Harness failures are retained separately from product regressions: +- The company-deletion fixture initially lacked a running agent lifecycle driver. + Its deletion case now seeds completed termination before exercising the real + company deletion transaction; it does not mock deletion or weaken its guard. +- A drain test incorrectly required all tracked intent work to end before resume. + It now tests passive SQL quiet during drain and waits for settlement after resume. +- The SQL-query fixture called a retry-backed worker again before its retained + backoff elapsed. It now advances the test clock before the second pass. +- Embedded company-test PostgreSQL startup timed out. The same suite now supports + the validated external `_test` URL, applies migrations, and closes its client. +- The previous disposable external cluster subsequently lost `global/pg_filenode.map`. + That attempted budget regression did not reach assertions. A new loopback-only + disposable PostgreSQL 17 cluster and separate databases run the same suites. + No additional IPC cleanup was performed. The cause of the missing file is unproven. + +Final local acceptance: all 224 tests in eight complete suites passed, workspace +`pnpm -r typecheck` passed, and the full `pnpm build` passed. Those commands ran +against the final uncommitted source on parent `026fdef06a`; the build stamp names +that parent, not a clean parent-only build. The publication commit adds this audit +and ledger text without further source changes. Full local `test:run` was not +repeated because of the embedded PostgreSQL startup/capacity problem. Remote CI +and review are separate gates and remain pending at publication. + +## Sleep boundary and remaining work + +This establishes routine-worker quiet between obligations. It does not establish +provider suspension, database suspension, fleet savings, or new Cloud alarm +coverage. The existing durable routine/issue-monitor idle proof and admission +fencing remain unchanged. In-process `nextWakeAt()` is not a universal external +alarm registration API. Unknown or invalid durable schedule states keep their +existing conservative host-sleep proof. + +The next distinct scheduling duty is issue-monitor/legacy-agent-timer polling. +Its suppression and producer coverage need their own audit before removing +`scheduled.timers.wake()`. GitHub event/continuity polling, merged-PR confirmation, +external-object refresh, status cards, skill/native-harness retention, and other +heartbeat leftovers remain in the original inventory. + +## October 29 master reconciliation and current acceptance + +This existing slice is being reconciled through the backup parent #17836 and its +merged prerequisites. Normal merges preserve the routine worker, its predicate, +producer hooks and all existing test assertions. The package tree remains the +qualified profile-80 parent tree. Final publication must adopt the actual merged +backup parent; local preparation on its accepted source is not a review pass. + +Current source `4659ef76d398193b15e7812d4a97f3da3b743226` has **417 passing +unique cases across 14 complete files**, without skips: routine deadline work, +scheduled work and its PostgreSQL queries, routine service, routine HTTP E2E and +routes, plugin-managed routines, budgets, companies, git workspace expiration, +agent-device login, startup registration, scoped deadline work and the shared +delivery coordinator. Database build, production server typechecking and build, +module boundaries, target migration order and Runner profile identity pass. + +The initial combined run passed 416 cases and timed out in the UUID routine HTTP +authorization case at its existing 20-second limit. A fresh complete affected +13-case E2E file passed in 13.92 seconds; that case completed in 425 ms. A sanitized +PostgreSQL wait-state observer completed without errors. No test assertion, +timeout or production source changed for that diagnostic, and the cause of the +initial timeout is not claimed as established. The initial failure remains in +the acceptance record rather than being described as an uninterrupted green run. + +Six fresh PostgreSQL 17 schemas and fresh helper-owned per-file databases applied +all 412 migration journal entries through index 413. Temporary local fixture +imports allowed those existing assertions to run against the owned external +PostgreSQL process; every selected source hash and import was restored exactly. +The worker tests retain their direct zero-select/execute/transaction assertions +through multiple former polling periods and real committed-producer reactivation, +outer commit/rollback, pause/resume, catch-up, restart, drain and retry coverage. + +This is bounded local source acceptance, not a full repository-suite pass, +measured instance/database suspension, external alarm delivery or fleet savings. +Independent writers and replicas still require their own supported notification +contract. The final published head needs fresh CI, security and review gates. diff --git a/server/src/__tests__/agent-device-login-routes.test.ts b/server/src/__tests__/agent-device-login-routes.test.ts index 98128d9438c..7eae61d319d 100644 --- a/server/src/__tests__/agent-device-login-routes.test.ts +++ b/server/src/__tests__/agent-device-login-routes.test.ts @@ -362,11 +362,15 @@ function boardActor(userId: string, companyIds: string[] = [COMPANY_1, COMPANY_2 }; } -async function createApp() { - const [{ agentRoutes }, { errorHandler }] = await Promise.all([ +async function loadRouteModules() { + return Promise.all([ vi.importActual("../routes/agents.js"), vi.importActual("../middleware/index.js"), ]); +} + +async function createApp() { + const [{ agentRoutes }, { errorHandler }] = await loadRouteModules(); const app = express(); app.use(express.json()); app.use((req, _res, next) => { @@ -386,9 +390,11 @@ describe("adapter device-login routes", () => { // `afterEach` whether or not the route removed them itself. const accountHomeTestDirs: string[] = []; - beforeAll(() => { + beforeAll(async () => { vi.resetModules(); - }); + + await loadRouteModules(); + }, 60_000); beforeEach(() => { vi.clearAllMocks(); diff --git a/server/src/__tests__/companies-service.test.ts b/server/src/__tests__/companies-service.test.ts index 85fd814c20c..d173295f94b 100644 --- a/server/src/__tests__/companies-service.test.ts +++ b/server/src/__tests__/companies-service.test.ts @@ -1,7 +1,9 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { and, eq } from "drizzle-orm"; import { + applyPendingMigrations, activityLog, agentConfigRevisions, agents, @@ -41,7 +43,8 @@ import { deriveIssuePrefixBase } from "../services/issue-prefix.js"; import { readBuiltInAgentMarker } from "../lib/built-in-agent-metadata.js"; import { builtInAgentService, reconcileBuiltInAgentsOnStartup } from "../services/built-in-agents.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_COMPANIES_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -68,8 +71,9 @@ describeEmbeddedPostgres("companyService", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-company-service-"); - db = createDb(tempDb.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-company-service-"); + db = createDb(externalUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -93,6 +97,7 @@ describeEmbeddedPostgres("companyService", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/git-workspace-expiration.test.ts b/server/src/__tests__/git-workspace-expiration.test.ts index 7adbecd36d8..c3a9abfbec0 100644 --- a/server/src/__tests__/git-workspace-expiration.test.ts +++ b/server/src/__tests__/git-workspace-expiration.test.ts @@ -8,11 +8,12 @@ import { pathToFileURL } from "node:url"; import { createRequire } from "node:module"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; -import { companies, projects, projectWorkspaces, issues, executionWorkspaces, agents, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, workspaceRuntimeServices, executionWorkspaceRestorePoints, createDb } from "@paperclipai/db"; +import { companies, projects, projectWorkspaces, issues, executionWorkspaces, agents, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, workspaceRuntimeServices, executionWorkspaceRestorePoints, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { markGitRestorePointExpired } from "../services/git-restore-point-storage.js"; import { GIT_RESTORE_POINT_EXPIRED_WARNING } from "../services/execution-workspaces.js"; import { getStorageService } from "../storage/index.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { resolveManagedProjectWorkspaceDir } from "../home-paths.js"; import { materializeManagedProjectWorkspace } from "../services/managed-repository-checkout.js"; import { materializeIsolatedRemoteWorkspaceSource } from "../services/isolated-remote-workspace-source.js"; @@ -30,7 +31,8 @@ const gitEnv = { ...process.env, GIT_AUTHOR_NAME: "Fixture", GIT_AUTHOR_EMAIL: " const git = (cwd: string, ...args: string[]) => execFileSync("git", args, { cwd, env: gitEnv, encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] }).trim(); const exists = (target: string) => fs.lstat(target).then(() => true, () => false); -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_GIT_EXPIRATION_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = support.supported ? describe : describe.skip; describeEmbeddedPostgres("reconstructable Git expiration", () => { let db: ReturnType; @@ -39,17 +41,18 @@ describeEmbeddedPostgres("reconstructable Git expiration", () => { const previousHome = process.env.PAPERCLIP_HOME; const previousInstance = process.env.PAPERCLIP_INSTANCE_ID; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("git-cache-expiration-"); - db = createDb(database.connectionString); + if (external) await applyPendingMigrations(external); + else database = await startEmbeddedPostgresTestDatabase("git-cache-expiration-"); + db = createDb(external ?? database.connectionString); root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-git-expiration-")); process.env.PAPERCLIP_HOME = path.join(root, "home"); process.env.PAPERCLIP_INSTANCE_ID = "default"; - }); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterEach(async () => { vi.restoreAllMocks(); await db.execute(sql`TRUNCATE TABLE companies CASCADE`); }); afterAll(async () => { if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; if (previousInstance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = previousInstance; - await db.$client.end({ timeout: 1 }); await database.cleanup(); await fs.rm(root, { recursive: true, force: true }); + await db.$client.end({ timeout: 1 }); await database?.cleanup(); await fs.rm(root, { recursive: true, force: true }); }); async function fixture(managed = false, nested = false, withProject = managed) { diff --git a/server/src/__tests__/routine-schedule-work.test.ts b/server/src/__tests__/routine-schedule-work.test.ts new file mode 100644 index 00000000000..e9cb49de446 --- /dev/null +++ b/server/src/__tests__/routine-schedule-work.test.ts @@ -0,0 +1,321 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, applyPendingMigrations, companies, costEvents, createDb, routineRuns, routineTriggers, type Db } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { routineService } from "../services/routines.js"; +import { budgetService } from "../services/budgets.js"; +import { companyService } from "../services/companies.js"; +import { registerScheduledWork } from "../services/scheduled-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ROUTINE_SCHEDULE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("routine schedule deadline work", () => { + let db: Db; + let temporary: Awaited> | undefined; + let companyId: string, agentId: string; + let coordinator: ReturnType | undefined; + let draining = false; + const errors: unknown[] = []; + const wakeup = vi.fn(async () => null); + const service = (owner = db) => routineService(owner, { heartbeat: { wakeup }, runtimeEnv: {} }); + beforeAll(async () => { + if (external) await applyPendingMigrations(external); + else temporary = await startEmbeddedPostgresTestDatabase("routine-schedule-work-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 }); + }, 30_000); + beforeEach(async () => { + companyId = randomUUID(); agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Scheduled work", issuePrefix: `R${companyId.slice(0, 6)}`, + defaultResponsibleUserId: randomUUID(), requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Scheduled agent", status: "active", lifecycleState: "ready", adapterType: "codex_local" }); + errors.length = 0; wakeup.mockClear(); draining = false; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + vi.setSystemTime(new Date("2026-07-16T12:30:00Z")); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`truncate companies cascade`); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + const create = (owner = db, catchUpPolicy: "skip_missed" | "enqueue_missed_with_cap" = "skip_missed") => service(owner).create(companyId, { + title: "Scheduled routine", assigneeAgentId: agentId, priority: "medium", status: "active", + concurrencyPolicy: "coalesce_if_active", catchUpPolicy, + }, {}); + async function start(options: { + next?: () => Promise; + run?: (signal: AbortSignal) => Promise; + } = {}) { + const svc = service(); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, + onError: error => errors.push(error) }); + const workers = registerScheduledWork(coordinator, db, { retryMs: 30_000, + canRun: () => !draining, + tickTimers: async () => {}, + tickRoutines: options.run ?? (signal => svc.tickScheduledTriggers(new Date(), () => !draining && !signal.aborted)), + nextRoutineRunAt: options.next ?? (() => svc.nextScheduledTriggerAt()), + }); + await Promise.all([workers.routines.ready, workers.timers.ready]); + return workers; + } + async function settled() { + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + } + async function quiet(duration = 60_000, waitForIdle = true) { + if (waitForIdle) await settled(); + const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(duration); + expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); execute.mockRestore(); transaction.mockRestore(); } + } + + it("retains overdue catch-up work after the first capped pass", async () => { + const routine = await create(db, "enqueue_missed_with_cap"); + const { trigger } = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 * * * *", timezone: "UTC" }, {}); + await db.update(routineTriggers).set({ nextRunAt: new Date("2026-07-15T10:00:00Z") }).where(eq(routineTriggers.id, trigger.id)); + await start(); await settled(); + expect(await db.select().from(routineRuns)).toHaveLength(25); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(async () => expect(await db.select().from(routineRuns)).toHaveLength(27)); + await settled(); + expect(coordinator!.nextWakeAt()).toBe(Date.parse("2026-07-16T13:00:00Z")); + await quiet(); + }); + + it("disarms on company pause and rearms from the committed reactivation", async () => { + const routine = await create(); + const { trigger } = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + await start(); await settled(); + expect(coordinator!.nextWakeAt()).toBe(trigger.nextRunAt!.getTime()); + await companyService(db).update(companyId, { status: "paused" }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await quiet(); + await companyService(db).update(companyId, { status: "active" }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(trigger.nextRunAt!.getTime())); + await quiet(); + expect(wakeup).not.toHaveBeenCalled(); + }); + + it("reactivates from the first trigger, replaces earlier deadlines, and cancels disabled or archived schedules", async () => { + await start(); await quiet(); + const routine = await create(); + await quiet(); + const created = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(created.trigger.nextRunAt!.getTime())); + const updated = await service().updateTrigger(created.trigger.id, { cronExpression: "0 * * * *" }, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(updated.trigger.nextRunAt!.getTime())); + await service().updateTrigger(created.trigger.id, { enabled: false }, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await quiet(); + await service().updateTrigger(created.trigger.id, { enabled: true }, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).not.toBeNull()); + await service().update(routine.id, { status: "archived" }, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await quiet(); + expect(wakeup).not.toHaveBeenCalled(); + }); + + it("restores a deleted schedule from its revision through the ordinary producer", async () => { + const routine = await create(); + const created = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + await start(); + await service().deleteTrigger(created.trigger.id, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await quiet(); + await service().restoreRevision(routine.id, created.revision!.id, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(created.trigger.nextRunAt!.getTime())); + await quiet(); + }); + + it("waits for the outer transaction and forgets rolled-back nested schedules", async () => { + const routine = await create(); + await start(); await quiet(); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const written = new Promise(resolve => { entered = resolve; }); + let expected: number | null = null; + const writing = db.transaction(async tx => { + const created = await service(tx as unknown as Db).createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + expected = created.trigger.nextRunAt!.getTime(); entered(); await held; + }); + try { + await written; + expect(await service().nextScheduledTriggerAt()).toBeNull(); + expect(wakeup).not.toHaveBeenCalled(); + } finally { release(); await writing; } + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(expected)); + await expect(db.transaction(async tx => { + await service(tx as unknown as Db).createTrigger(routine.id, { kind: "schedule", cronExpression: "* * * * *", timezone: "UTC" }, {}); + throw new Error("rollback fixture"); + })).rejects.toThrow("rollback fixture"); + await settled(); + expect(await service().nextScheduledTriggerAt()).toBe(expected); + expect(coordinator!.nextWakeAt()).toBe(expected); + await quiet(); + }); + + it("retains the first committed schedule during drain without passive SQL and loads it on resume", async () => { + const routine = await create(); + await start(); await quiet(); + draining = true; + const created = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + await vi.advanceTimersByTimeAsync(1_000); + await quiet(60_000, false); + draining = false; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(created.trigger.nextRunAt!.getTime())); + await quiet(); + }); + + it("reconstructs an overdue schedule on restart and keeps completed history passive", async () => { + const routine = await create(); + await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "* * * * *", timezone: "UTC" }, {}); + await start(); await settled(); + await coordinator!.stop(); coordinator = undefined; + vi.setSystemTime(new Date("2026-07-16T12:35:30Z")); + await start(); await settled(); + expect(await db.select().from(routineRuns)).toHaveLength(1); + expect(coordinator!.nextWakeAt()).toBe(Date.parse("2026-07-16T12:36:00Z")); + await service().update(routine.id, { status: "paused" }, {}); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await quiet(); + expect(await db.select().from(routineRuns)).toHaveLength(1); + }); + + it("retries a failed deadline read after drain and becomes quiet after successful reconstruction", async () => { + const routine = await create(); + const created = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + const next = vi.fn().mockRejectedValueOnce(new Error("deadline unavailable")) + .mockImplementation(() => service().nextScheduledTriggerAt()); + await start({ next }); + expect(errors.splice(0)).toEqual([expect.objectContaining({ message: "deadline unavailable" })]); + expect(next).toHaveBeenCalledTimes(1); + draining = true; + await quiet(60_000, false); + expect(next).toHaveBeenCalledTimes(1); + draining = false; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(created.trigger.nextRunAt!.getTime())); + await quiet(); + }); + + it("reconciles a schedule changed while the previous deadline read was in flight", async () => { + const routine = await create(); + const created = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const read = new Promise(resolve => { entered = resolve; }); + const next = vi.fn().mockImplementationOnce(async () => { + const value = await service().nextScheduledTriggerAt(); entered(); await held; return value; + }).mockImplementation(() => service().nextScheduledTriggerAt()); + const starting = start({ next }); + let expected: number | undefined; + try { + await read; + const updated = await service().updateTrigger(created.trigger.id, { cronExpression: "0 * * * *" }, {}); + expected = updated.trigger.nextRunAt!.getTime(); + } finally { release(); await starting; } + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(expected)); + await quiet(); + }); + + it("stops between claimed trigger batches and leaves unclaimed work for restart", async () => { + const routine = await create(); + await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 * * * *", timezone: "UTC" }, {}); + await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 * * * *", timezone: "UTC" }, {}); + await db.update(routineTriggers).set({ nextRunAt: new Date("2026-07-16T12:00:00Z") }); + let candidates = 0; + await start({ run: () => service().tickScheduledTriggers(new Date(), () => ++candidates === 1) }); + await settled(); + expect(await db.select().from(routineRuns)).toHaveLength(1); + expect(await service().nextScheduledTriggerAt()).toBe(Date.parse("2026-07-16T12:00:00Z")); + await coordinator!.stop(); coordinator = undefined; + await start(); await settled(); + expect(await db.select().from(routineRuns)).toHaveLength(2); + expect(coordinator!.nextWakeAt()).toBe(Date.parse("2026-07-16T13:00:00Z")); + await quiet(); + }); + + + it("refreshes a saved deadline advanced by a manual run naming the trigger", async () => { + const routine = await create(); + const { trigger } = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + await db.update(routineTriggers).set({ nextRunAt: new Date("2026-08-01T00:00:00Z") }).where(eq(routineTriggers.id, trigger.id)); + await start(); await settled(); + expect(coordinator!.nextWakeAt()).toBe(Date.parse("2026-08-01T00:00:00Z")); + await service().runRoutine(routine.id, { source: "manual", triggerId: trigger.id }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(Date.parse("2026-07-17T00:00:00Z"))); + await quiet(); + }); + + it.each(["archive", "remove"] as const)("disarms a company schedule after %s", async operation => { + const routine = await create(); + await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + await start(); await settled(); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + if (operation === "remove") { + + + await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" }).where(eq(agents.id, agentId)); + } + await companyService(db)[operation](companyId); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await quiet(); + }); + + it("starts during drain without querying and reconstructs schedules when admission resumes", async () => { + const routine = await create(); + const { trigger } = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + draining = true; + const select = vi.spyOn(db, "select"); + try { await start(); expect(select).not.toHaveBeenCalled(); } finally { select.mockRestore(); } + await quiet(60_000, false); + draining = false; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(trigger.nextRunAt!.getTime())); + await quiet(); + }); + + it("does not consume overdue triggers in a paused company", async () => { + const routine = await create(); + const { trigger } = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 * * * *", timezone: "UTC" }, {}); + const due = new Date("2026-07-16T12:00:00Z"); + await db.update(routineTriggers).set({ nextRunAt: due }).where(eq(routineTriggers.id, trigger.id)); + await companyService(db).update(companyId, { status: "paused" }); + await start(); await quiet(); + expect(await db.select().from(routineRuns)).toHaveLength(0); + expect((await db.select().from(routineTriggers))[0]!.nextRunAt).toEqual(due); + expect(coordinator!.nextWakeAt()).toBeNull(); + }); + + + it.each(["pause", "resume"] as const)("refreshes schedules when budget enforcement changes company state: %s", async operation => { + const routine = await create(); + const { trigger } = await service().createTrigger(routine.id, { kind: "schedule", cronExpression: "0 0 * * *", timezone: "UTC" }, {}); + await db.insert(costEvents).values({ companyId, agentId, provider: "openai", biller: "openai", + billingType: "metered_api", model: "fixture", costCents: 125, occurredAt: new Date() }); + const budgets = budgetService(db); + const limit = (amount: number) => budgets.upsertPolicy(companyId, { + scopeType: "company", scopeId: companyId, amount, windowKind: "calendar_month_utc", notifyEnabled: false, + }, null); + if (operation === "resume") await limit(100); + await start(); await settled(); + if (operation === "pause") { + await limit(100); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + } else { + expect(coordinator!.nextWakeAt()).toBeNull(); + await limit(200); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(trigger.nextRunAt!.getTime())); + } + await quiet(); + }); + +}); diff --git a/server/src/__tests__/scheduled-work-postgres.test.ts b/server/src/__tests__/scheduled-work-postgres.test.ts index fb4d9a8d180..46fd20fc4c7 100644 --- a/server/src/__tests__/scheduled-work-postgres.test.ts +++ b/server/src/__tests__/scheduled-work-postgres.test.ts @@ -1,23 +1,29 @@ -import { agents, companies, createDb, issues, routines, routineTriggers } from "@paperclipai/db"; +import { agents, applyPendingMigrations, companies, createDb, issues, routines, routineTriggers } from "@paperclipai/db"; import { sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import type { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { DELIVERY_QUEUES, type DeliveryQueue } from "../services/delivery-work-notifications.js"; +import { routineService } from "../services/routines.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { registerScheduledWork } from "../services/scheduled-work.js"; -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_SCHEDULED_WORK_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)("scheduled deadline database queries", () => { let database: Awaited>; let db: Db; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-scheduled-deadlines"); - db = createDb(database.connectionString); + if (external) await applyPendingMigrations(external); + else database = await startEmbeddedPostgresTestDatabase("paperclip-scheduled-deadlines"); + db = createDb(external ?? database.connectionString); }, 30_000); + afterEach(() => vi.useRealTimers()); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); - it("selects future eligible deadlines and excludes inactive, mismatched and past work", async () => { + it("retains eligible overdue routines while excluding inactive and mismatched work", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); const now = Date.now(); const at = (minutes: number) => new Date(now + minutes * 60_000); const [company, pausedCompany] = await db.insert(companies).values([ @@ -30,7 +36,7 @@ describe.skipIf(!support.supported)("scheduled deadline database queries", () => { companyId: company!.id, title: "Paused routine", status: "paused" }, { companyId: pausedCompany!.id, title: "Paused company routine" }, ]).returning(); - const trigger = { companyId: company!.id, routineId: active!.id, kind: "schedule" }; + const trigger = { companyId: company!.id, routineId: active!.id, kind: "schedule", cronExpression: "0 * * * *", timezone: "UTC" }; await db.insert(routineTriggers).values([ { ...trigger, nextRunAt: at(10) }, { ...trigger, enabled: false, nextRunAt: at(1) }, @@ -54,11 +60,14 @@ describe.skipIf(!support.supported)("scheduled deadline database queries", () => registerScheduledWork({ register: (queue, task) => { tasks.set(queue, task); return { ready: Promise.resolve(), wake: () => {} }; - } }, db, { retryMs: 60_000, tickTimers: async () => {}, tickRoutines: async () => {} }); + } }, db, { retryMs: 60_000, tickTimers: async () => {}, tickRoutines: async () => {}, canRun: () => true, + nextRoutineRunAt: () => routineService(db).nextScheduledTriggerAt() }); for (const task of tasks.values()) await task.run(new AbortController().signal); - expect(tasks.get(DELIVERY_QUEUES.routineSchedules)!.nextRunAt!()).toBe(at(10).getTime()); + expect(await routineService(db).nextScheduledTriggerAt()).toBe(at(-1).getTime()); + expect(tasks.get(DELIVERY_QUEUES.routineSchedules)!.nextRunAt!()).toBeGreaterThan(now); expect(tasks.get(DELIVERY_QUEUES.issueMonitors)!.nextRunAt!()).toBe(at(15).getTime()); await db.execute(sql`truncate table companies restart identity cascade`); + vi.setSystemTime(at(2)); for (const task of tasks.values()) await task.run(new AbortController().signal); for (const task of tasks.values()) expect(task.nextRunAt!()).toBeNull(); }); diff --git a/server/src/__tests__/scheduled-work.test.ts b/server/src/__tests__/scheduled-work.test.ts index a2e3acb6161..79567e83899 100644 --- a/server/src/__tests__/scheduled-work.test.ts +++ b/server/src/__tests__/scheduled-work.test.ts @@ -36,10 +36,16 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick const coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, canReconcile: () => canReconcile, onError: errors }); coordinators.push(coordinator); const tickTimers = vi.fn(input.tickTimers ?? (async () => {})); - const tickRoutines = vi.fn(input.tickRoutines ?? (async () => {})); + const tickRoutines = vi.fn(input.tickRoutines ?? (async () => { + if (deadlines.routine !== null && deadlines.routine <= Date.now()) deadlines.routine = null; + })); + const nextRoutineRunAt = vi.fn(async () => { + if (failRead) throw new Error("database unavailable"); + return deadlines.routine; + }); const workers = registerScheduledWork(coordinator, owner as unknown as Db, - { retryMs: 60_000, tickTimers, tickRoutines }); - return { coordinator, workers, owner, deadlines, execute, tickTimers, tickRoutines, errors, + { retryMs: 60_000, tickTimers, tickRoutines, nextRoutineRunAt, canRun: () => canRun }); + return { coordinator, workers, owner, deadlines, execute, nextRoutineRunAt, tickTimers, tickRoutines, errors, ready: () => Promise.all([workers.routines.ready, workers.timers.ready]), canRun: (value: boolean) => { canRun = value; }, canReconcile: (value: boolean) => { canReconcile = value; }, failRead: (value: boolean) => { failRead = value; } }; } @@ -60,16 +66,20 @@ describe("scheduled routine and monitor deadlines", () => { await vi.advanceTimersByTimeAsync(10_000); expect(f.tickTimers).toHaveBeenCalledTimes(2); expect(f.coordinator.nextWakeAt()).toBeNull(); - expect(f.execute).toHaveBeenCalledTimes(4); + expect(f.execute).toHaveBeenCalledTimes(2); + expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(2); expect(idleWorkSnapshot().active).toBe(0); await vi.advanceTimersByTimeAsync(3_600_000); - expect(f.execute).toHaveBeenCalledTimes(4); + expect(f.execute).toHaveBeenCalledTimes(2); + expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(2); }); it.each(["routine", "monitor"] as const)("rescans a %s deadline crossed during an earlier scan", async kind => { let release!: () => void; const held = new Promise(resolve => { release = resolve; }); - const tick = vi.fn().mockImplementationOnce(() => held).mockResolvedValue(undefined); + const tick = vi.fn().mockImplementationOnce(() => held).mockImplementation(async () => { + if (kind === "routine") f.deadlines.routine = null; + }); const now = Date.UTC(2020, 0, 1); const f = fixture({ [kind]: now + 10_000, ...(kind === "routine" ? { tickRoutines: tick } : { tickTimers: tick }) }); @@ -77,19 +87,28 @@ describe("scheduled routine and monitor deadlines", () => { await (kind === "routine" ? f.workers.timers.ready : f.workers.routines.ready); await vi.advanceTimersByTimeAsync(12_000); } finally { release(); await f.ready(); } - expect(f.coordinator.nextWakeAt()).toBe(Date.now()); - await vi.advanceTimersByTimeAsync(1); + const expected = kind === "routine" ? now + 60_000 : Date.now(); + expect(f.coordinator.nextWakeAt()).toBe(expected); + await vi.advanceTimersByTimeAsync(expected - Date.now() + 1); expect(tick).toHaveBeenCalledTimes(2); expect(f.coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); - it("leaves work already overdue at scan start for bounded heartbeat recovery", async () => { - const f = fixture({ routine: Date.UTC(2020, 0, 1) - 10_000 }); + it("retains capped overdue work with bounded spacing, then disarms after completion", async () => { + const tick = vi.fn(async () => {}); + const now = Date.UTC(2020, 0, 1); + const f = fixture({ routine: now - 10_000, tickRoutines: tick }); await f.ready(); + expect(f.coordinator.nextWakeAt()).toBe(now + 60_000); + await vi.advanceTimersByTimeAsync(59_999); + expect(tick).toHaveBeenCalledTimes(1); + f.deadlines.routine = null; + await vi.advanceTimersByTimeAsync(1); + expect(tick).toHaveBeenCalledTimes(2); expect(f.coordinator.nextWakeAt()).toBeNull(); await vi.advanceTimersByTimeAsync(3_600_000); - expect(f.tickRoutines).toHaveBeenCalledTimes(1); + expect(tick).toHaveBeenCalledTimes(2); expect(idleWorkSnapshot().active).toBe(0); }); @@ -138,7 +157,8 @@ describe("scheduled routine and monitor deadlines", () => { expect(idleWorkSnapshot().active).toBe(0); f.canReconcile(false); await vi.advanceTimersByTimeAsync(60_000); - expect(f.execute).toHaveBeenCalledTimes(2); + expect(f.execute).toHaveBeenCalledTimes(1); + expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(1); await f.coordinator.stop(); f.canRun(true); await vi.advanceTimersByTimeAsync(3_600_000); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 94f22012fe9..bd46bc6b45d 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -190,6 +190,7 @@ const { }; const externalObjectsServiceFactoryMock = vi.fn(() => externalObjectsServiceMock); const routineServiceMock = { + nextScheduledTriggerAt: vi.fn(async () => null), tickScheduledTriggers: vi.fn(async () => ({ triggered: 0 })), }; const routineServiceFactoryMock = vi.fn(() => routineServiceMock); @@ -1180,6 +1181,9 @@ describe("startServer feedback export wiring", () => { expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); + const routineWorkerIndex = deliveryWorkRegisterMock.mock.calls.findIndex(([queue]) => queue === DELIVERY_QUEUES.routineSchedules); + expect(routineWorkerIndex).toBeGreaterThanOrEqual(0); + expect(deliveryWorkRegisterMock.mock.results[routineWorkerIndex]!.value.wake).not.toHaveBeenCalled(); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.customImageExpiry)).toBe(true); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(true); diff --git a/server/src/index.ts b/server/src/index.ts index d4112fe24b7..6424a50d991 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1990,9 +1990,6 @@ async function startServerWithDatabaseTeardown( scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - scheduled.routines.wake(); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork((async () => { const experimental = await instanceSettingsService(db).getExperimental(); @@ -2088,7 +2085,10 @@ async function startServerWithDatabaseTeardown( const result = await heartbeat.tickTimers(new Date()); if (result.enqueued > 0) logger.info({ ...result }, "heartbeat timer tick enqueued runs"); }, - tickRoutines: () => routines.tickScheduledTriggers(new Date()).then(result => { + canRun: () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, + nextRoutineRunAt: () => routines.nextScheduledTriggerAt(), + tickRoutines: signal => routines.tickScheduledTriggers(new Date(), + () => !signal.aborted && !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).then(result => { if (result.triggered > 0) logger.info({ ...result }, "routine scheduler tick enqueued runs"); }), }); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 9a84916484b..f32e28dbaf1 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -301,6 +301,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); await db @@ -358,6 +359,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await db .update(companies) diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 1534817fa2e..8e1f34b889d 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -333,6 +333,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); await notifyCompanyRetentionWork(tx, id); await notifyChatInboundWakeNotices(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); @@ -486,6 +487,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); await notifyCompanyRetentionWork(tx, id); await notifyChatInboundWakeNotices(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index f4e020bd3da..2a4fb16e37d 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -74,6 +74,7 @@ export async function deleteCompany(db: Db, id: string) { await notifyChatCompanyConfiguration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id); await notifyToolReviewChange(tx, id); await agentObservabilityService(tx as unknown as Db).removeForCompanyDeletion(id); diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index 50df3a48db5..14c5cad5465 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -2,7 +2,7 @@ import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator. import type { DeliveryQueue } from "./delivery-work-notifications.js"; export function registerDeadlineWork( - coordinator: ReturnType, + coordinator: Pick, "register">, queue: DeliveryQueue, task: { run: (signal: AbortSignal) => Promise; diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index bea7a1488c6..23787b45249 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -666,6 +666,22 @@ export function routineService( pluginWorkerManager: deps.pluginWorkerManager, }); + + const scheduledTriggerEligibility = () => and( + eq(routineTriggers.kind, "schedule"), + eq(routineTriggers.enabled, true), + eq(routineTriggers.archived, false), + eq(routines.status, "active"), + eq(companies.status, "active"), + isNotNull(routineTriggers.nextRunAt), + isNotNull(routineTriggers.cronExpression), + isNotNull(routineTriggers.timezone), + ); + const scheduledRoutineJoin = () => and( + eq(routineTriggers.routineId, routines.id), + eq(routineTriggers.companyId, routines.companyId), + ); + async function getRoutineById(id: string) { if (!isRoutineUuidInput(id)) return null; return db @@ -1772,6 +1788,11 @@ export function routineService( } } + + + if (input.trigger?.kind === "schedule" && input.nextRunAtOverride === undefined) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); + } const triggeredAt = new Date(); const manualRunnerUserId = input.source === "manual" ? input.actor?.userId ?? null : null; const latestRevisionResponsibleUserId = input.routine.latestRevisionId @@ -3092,7 +3113,18 @@ export function routineService( })); }, - tickScheduledTriggers: async (now: Date = new Date()) => { + nextScheduledTriggerAt: async (): Promise => { + const [row] = await db.select({ at: routineTriggers.nextRunAt }) + .from(routineTriggers) + .innerJoin(routines, scheduledRoutineJoin()) + .innerJoin(companies, eq(routines.companyId, companies.id)) + .where(scheduledTriggerEligibility()) + .orderBy(asc(routineTriggers.nextRunAt)).limit(1); + + return row?.at?.getTime() ?? null; + }, + + tickScheduledTriggers: async (now: Date = new Date(), canContinue: () => boolean = () => true) => { const worktreeActivation = isTruthyRuntimeEnvValue(runtimeEnv.PAPERCLIP_IN_WORKTREE) ? await resolveWorktreeRunExecutionActivationState({ getExperimental: instanceSettings.getExperimental, @@ -3106,15 +3138,12 @@ export function routineService( projectPausedAt: projects.pausedAt, }) .from(routineTriggers) - .innerJoin(routines, eq(routineTriggers.routineId, routines.id)) + .innerJoin(routines, scheduledRoutineJoin()) + .innerJoin(companies, eq(routines.companyId, companies.id)) .leftJoin(projects, eq(routines.projectId, projects.id)) .where( and( - eq(routineTriggers.kind, "schedule"), - eq(routineTriggers.enabled, true), - eq(routineTriggers.archived, false), - eq(routines.status, "active"), - isNotNull(routineTriggers.nextRunAt), + scheduledTriggerEligibility(), lte(routineTriggers.nextRunAt, now), ), ) @@ -3122,6 +3151,9 @@ export function routineService( let triggered = 0; for (const row of due) { + + + if (!canContinue()) break; if (!row.trigger.nextRunAt || !row.trigger.cronExpression || !row.trigger.timezone) continue; // Suppress scheduled firings while the routine's project is paused. The tick is still diff --git a/server/src/services/scheduled-work.ts b/server/src/services/scheduled-work.ts index 6f31e2e420a..e46bcca4f66 100644 --- a/server/src/services/scheduled-work.ts +++ b/server/src/services/scheduled-work.ts @@ -1,3 +1,4 @@ +import { registerDeadlineWork } from "./deadline-work.js"; import type { Db } from "@paperclipai/db"; import { sql, type SQL } from "drizzle-orm"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; @@ -9,7 +10,9 @@ export function registerScheduledWork( options: { retryMs: number; tickTimers: () => Promise; - tickRoutines: () => Promise; + tickRoutines: (signal: AbortSignal) => Promise; + nextRoutineRunAt: () => Promise; + canRun: () => boolean; }, ) { function register(queue: DeliveryQueue, work: () => Promise, query: (now: Date) => SQL) { @@ -30,12 +33,12 @@ export function registerScheduledWork( }); } - const routines = register(DELIVERY_QUEUES.routineSchedules, options.tickRoutines, now => sql` - SELECT min(t.next_run_at) AS next_at - FROM routine_triggers t JOIN routines r ON r.id = t.routine_id AND r.company_id = t.company_id - JOIN companies c ON c.id = r.company_id - WHERE t.kind = 'schedule' AND t.enabled AND NOT t.archived - AND r.status = 'active' AND c.status = 'active' AND t.next_run_at > ${now.toISOString()}::timestamptz`); + const routines = registerDeadlineWork(coordinator, DELIVERY_QUEUES.routineSchedules, { + run: options.tickRoutines, + next: options.nextRoutineRunAt, + canRun: options.canRun, + retryMs: options.retryMs, + }); const timers = register(DELIVERY_QUEUES.issueMonitors, options.tickTimers, now => sql` SELECT min(i.monitor_next_check_at) AS next_at FROM issues i JOIN companies c ON c.id = i.company_id