mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 05:16:50 +01:00
fix(server): schedule routines without heartbeat polling (#17840)
<!-- entire-trail-link-start --> https://entire.com/gh/paperclipai/-/trails/47/changes/paperclip/47 <!-- entire-trail-link-end --> ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Routines create work at saved cron deadlines. > - The heartbeat still wakes the routine worker when nothing is due. > - Its old next-deadline query also drops overdue work left by capped catch-up. > - Removing the heartbeat alone can strand that work and miss company reactivation. > - This pull request uses saved deadlines and committed producer changes. > - Idle routine scheduling stops querying until a change, retry, or real deadline. ## Linked Issues or Issue Description Refs #16514. Merged prerequisite: #17836. Related Cloud wake contract: #16215. **What happened?** Every heartbeat wakes the routine scheduler. The next-deadline query excludes overdue triggers. A 27-tick hourly backlog loses its deadline after the first 25-run pass. Company and budget lifecycle changes also lack routine schedule signals. **Expected behavior** Use the actual next saved deadline. Retain overdue work. Refresh after supported producer transactions commit. Keep idle schedules quiet between obligations. **Steps to reproduce** 1. Create an hourly routine with capped catch-up and 27 overdue ticks. 2. Run the registered worker once. The base processes 25 ticks and reports no next wake. 3. Create a future schedule and pause or reactivate its company. The base does not refresh its registered deadline until another wake. ## What Changed - Remove only the routine worker wake from the heartbeat tick. - Put routine eligibility and the next-deadline read in `routineService`. Include overdue triggers. Keep trigger/routine company scope consistent and require an active company. - Reuse `registerDeadlineWork` for retries, request generations, and passive drain behavior. Narrow its coordinator type to the `register` method it uses. Existing consumers keep the same runtime behavior. - Signal routine work from manual schedule invocation, company pause/resume/archive/delete, and budget-induced company pause/resume. Existing routine and trigger producers retain their transaction signals. - Stop between trigger batches during drain/shutdown. Finish a claimed batch so its advanced timestamp does not discard admitted work. - Add real PostgreSQL lifecycle tests and extend startup/query fixtures. Add a validated external database option to the company suite, including migrations and client closure. - Record producer coverage and limits in `doc/plans/2026-10-27-routine-schedule-sleep.md`. ## Verification - **417 passing unique cases across 14 complete files**, no skips, on accepted source `4659ef76d398193b15e7812d4a97f3da3b743226`. Coverage includes the real deadline worker, database eligibility queries, routine services/HTTP/routes/plugins, budget/company producers, git/device fixtures, startup and shared coordinator. - The first combined run passed 416 cases and had one UUID HTTP authorization timeout at its unchanged 20-second limit. One fresh complete 13-case E2E diagnostic passed (the affected case took 425 ms). No assertions, timeouts or production source were changed; the initial timeout's cause is not established. - Database build, production server typechecking/build, module boundaries, target migration order and Runner profile identity pass. Fresh PostgreSQL 17 fixtures applied all 412 journal entries through index 413; all selected test hashes and temporary fixture imports were restored exactly. - Normal parent merges preserve the tested production and test source. Historical acceptance and review corrections remain in [the routine audit](doc/plans/2026-10-27-routine-schedule-sleep.md). - No full local repository-suite, deployed suspend/wake, external alarm, live provider or fleet-cost claim. Final-head CI, security, exact-head Greptile 5/5 and actionable Ponytail findings remain separate merge gates. ## Risks - Company pause now prevents due routine claims, matching the prior deadline query. Rows remain saved for reactivation. Existing routine reactivation, project-pause skip records, cron, timezone, catch-up, and concurrency policies remain with the routine owner. - Signals cover supported service writers in one process. Independent SQL writers and other replicas need their own integration. - This proves worker quiet, not host/database suspension or fleet savings. Cloud idle proof and external wake coverage are unchanged. - Issue-monitor and legacy-agent-timer polling remain separate duties. This slice does not finish the full migration stack or its CI/review work. ## Model Used OpenAI Codex (GPT-6-based coding agent), using repository inspection, ordinary Git merges, source-composition proofs, PostgreSQL fixtures and local lifecycle tests. No subagents were used in the current reconciliation. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
16 files changed
+662
-54
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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<typeof import("../routes/agents.js")>("../routes/agents.js"),
|
||||
vi.importActual<typeof import("../middleware/index.js")>("../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();
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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();
|
||||
});
|
||||
|
||||
|
||||
@@ -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<typeof createDb>;
|
||||
@@ -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) {
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let companyId: string, agentId: string;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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<number | null>;
|
||||
run?: (signal: AbortSignal) => Promise<unknown>;
|
||||
} = {}) {
|
||||
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<void>(resolve => { release = resolve; });
|
||||
const written = new Promise<void>(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<void>(resolve => { release = resolve; });
|
||||
const read = new Promise<void>(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();
|
||||
});
|
||||
|
||||
});
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
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();
|
||||
});
|
||||
|
||||
@@ -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<void>(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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
+4
-4
@@ -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");
|
||||
}),
|
||||
});
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -2,7 +2,7 @@ import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.
|
||||
import type { DeliveryQueue } from "./delivery-work-notifications.js";
|
||||
|
||||
export function registerDeadlineWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
coordinator: Pick<ReturnType<typeof createDeliveryWorkCoordinator>, "register">,
|
||||
queue: DeliveryQueue,
|
||||
task: {
|
||||
run: (signal: AbortSignal) => Promise<unknown>;
|
||||
|
||||
@@ -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<number | null> => {
|
||||
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
|
||||
|
||||
@@ -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<unknown>;
|
||||
tickRoutines: () => Promise<unknown>;
|
||||
tickRoutines: (signal: AbortSignal) => Promise<unknown>;
|
||||
nextRoutineRunAt: () => Promise<number | null>;
|
||||
canRun: () => boolean;
|
||||
},
|
||||
) {
|
||||
function register(queue: DeliveryQueue, work: () => Promise<unknown>, 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
|
||||
|
||||
Reference in new issue
Block a user