diff --git a/doc/idle-issue-monitors.md b/doc/idle-issue-monitors.md new file mode 100644 index 00000000000..db0903a914b --- /dev/null +++ b/doc/idle-issue-monitors.md @@ -0,0 +1,38 @@ +# Issue monitor scheduling + +Paperclip reconstructs monitor deadlines from saved issue rows at server startup. +While running, supported issue, execution, company, and native-run services +notify the monitor worker when their database transactions commit. The worker +uses those notifications to re-read durable state. It does not depend on a +process-local queue count for correctness. + +Notifications are batched for up to the configured heartbeat interval (normally +30 seconds). This bounds global monitor scans during frequent task edits. An +already-known monitor deadline can fire earlier than the batching window. +After all work settles, no periodic monitor scan remains. Pending claims and +failed or deferred admissions retain their bounded deadlines. + +Both idle drains and ordinary maintenance drains pause monitor dispatch. The +worker retains its pending request without repeatedly reading monitor tables. +After an explicit drain stop or TTL expiry, a bounded in-memory retry resumes +that request. Shutdown cancels local timers; restart reconstructs durable work. + +## Supported writers + +Use Paperclip's API and services to create or change monitor work. These writers +share the serving process's database-work notification scope, including nested +transactions and uncertain commit reconciliation. Supported import paths also +use these service boundaries or startup reconstruction. + +Independent SQL clients, other service processes, and additional server replicas +do not publish into that process-local scope. Their direct writes are not +supported live monitor producers. Restart reconstructs saved rows, but that is +not a live cross-process delivery guarantee. Such integrations need an explicit +ingress or durable notification contract before they can rely on monitor wakes. +There is no unconditional recovery poll for unsupported writers. + +## Scope of the sleep claim + +An idle monitor worker can stop its SQL activity. This alone does not prove that +the host or database will suspend. Cloud's existing external alarm and sleep +safety checks remain in place; an in-process timer cannot wake a suspended host. diff --git a/doc/plans/2026-10-27-issue-monitor-sleep-audit.md b/doc/plans/2026-10-27-issue-monitor-sleep-audit.md new file mode 100644 index 00000000000..4f6b1d52919 --- /dev/null +++ b/doc/plans/2026-10-27-issue-monitor-sleep-audit.md @@ -0,0 +1,190 @@ +# Issue monitor deadline migration + +Parent: slice 61, PR #17840, `68339f4162f26f5e12df0848d92f2ed62af8841f`. +Fresh remote master: `0cb3fc7db591ba23e21fbbc70042bc44ec7d6f20`. + +This slice owns only issue-monitor scheduling. Legacy agent interval policy stays +on the existing heartbeat cadence for its own subsequent slice. Manual combined +`tickTimers` behavior must remain compatible. No Cloud sleep proof is relaxed. + +Truth is already in issues: monitorNextCheckAt, monitorWakeRequestedAt, assignee, +status, and execution policy. Native runs and finalization phase suppress a new +monitor while the prior native owner remains active or unfinished. The scheduling +module keeps dispatch authorization, exact-claim CAS, quota recovery, attempt +limits, diagnostics, and monitor consumption. Index wiring selects workers only. + +The next deadline must include already-due work left by the 50-item pass and a +claim's five-minute stale threshold (including its strict comparison boundary). +Failed/deferred dispatch keeps concrete due debt with the shared deadline worker's +bounded retry. Empty/history-only state becomes unscheduled. Suppressed work must +reactivate through settings and lifecycle notifications, including after drain. + +Producer audit: issue service create/update, native set_task_monitor, native +preserved-monitor decisions, scheduler manual/automatic claim/release/consumption, +issue lifecycle state, company pause/reactivation/budget transitions/deletion, and +native run/finalization changes. Import deliberately leaves monitor deadlines null +until explicitly activated. Existing native projection and issue review-state +notification boundaries already cover their corresponding transition owners. +Signals must precede writes in the committing transaction; filesystem/provider +work must stay outside those transactions. Process-local writer coverage does not +claim support for unrelated SQL writers or another server replica. + +Acceptance must measure quiet SQL, first real producer, future and stale-claim +boundaries, capped overdue work, deferred/error retry, manual compatibility, +outer commit/rollback, native settlement reactivation, company pause/resume, +startup/restart, drain/standby, and shutdown. Existing monitor/admission integration +coverage remains required. This audit is a plan, not evidence of completion. + +## Implemented boundary + +`createHeartbeatScheduling` now exposes `tickIssueMonitors` and +`nextIssueMonitorAt`. It also exposes `tickAgentTimers`. The public `tickTimers` +method still combines both for existing manual callers. `registerScheduledWork` +uses the existing deadline worker for monitors and a separate coordinator topic +for legacy agent ticks. The heartbeat interval only wakes the latter. + +The monitor query includes overdue work, including a capped pass's remainder. +Its deadline is the greater of the saved deadline and the claim timestamp plus +five minutes plus one millisecond. The extra millisecond matches the existing +strict stale-claim comparison. A native queued/running/retry owner or unfinished +native finalization excludes the monitor until a state change wakes inspection. + +The worker uses the configured heartbeat interval as a minimum retry spacing +while overdue debt remains. The spacing applies to the monitor worker, so new +notifications during that backoff do not start another immediate scan. Future +saved deadlines run directly when no overdue retry is pending. State-change +bursts use a coalescing window equal to the configured heartbeat interval. No per-monitor +queue, new table, independent timer, or shared scheduler modification was added. + +## Producer coverage + +| Owning path | Notification boundary | +| --- | --- | +| Issue create/update | Existing monitor intent before the issue write | +| Native `set_task_monitor`; preserved-monitor decisions | Existing monitor intent before the issue write | +| Manual/automatic claim, release, diagnostic fallback, consumption | Added intent in the existing or short write transaction; provider admission remains outside it | +| Issue checkout and stale checkout adoption | Added a short transaction around the existing CAS and intent; captured the expected execution ID before entering the callback | +| Issue removal, release, and force release | Added intent in the existing transaction | +| Tree restoration, member reassignment, agent deletion | Added intent before changing monitor eligibility | +| Agent conversation activation/settlement and external conversation settlement | Added intent before direct issue status writes | +| Workspace reconstruction and execution-recovery blocking | Added intent before changing eligibility | +| Native projection/finalization and issue review transitions | Extended existing notification fanout; no service invocation inside the notifier | +| Execution-control recovery writes and active-run watchdog folding | Extended transaction notifications because these can release native ownership | +| Company status/deletion/budget changes | Added monitor topic beside existing schedule notifications | +| Worktree execution suppression setting | Existing native projection notification now also wakes monitor inspection | + +The next-deadline predicate deliberately has the same issue/company/native-owner +eligibility as the existing due-monitor query. Admission still owns authorization, +quota behavior, exact claim guards, current assignment, and client-error handling. +An abort or drain observed after a claim leaves that claim durable for later stale +claim recovery. It does not dispatch new work after that lifecycle check fails. + +## Verification on the slice + +- Baseline: existing monitor admission and scheduled-work suites, 25 passed. +- Final combined run: eight complete suites, 219 passed, no skipped tests. The + suites were `heartbeat/scheduling`, `scheduled-work`, `scheduled-work-postgres`, + `routine-schedule-work`, `issue-monitor-scheduler`, `server-startup-feedback-export`, + `execution-control-work`, and `native-run-projection-work`. +- Six isolated external PostgreSQL 17 databases had all 403 migrations applied. + This avoids treating unavailable embedded-Postgres suites as passing evidence. +- New lifecycle cases verify zero worker select/execute/transaction calls over + multiple old poll periods after settlement, with completed monitor history + retained. They also verify the first ordinary issue producer, nested outer + commit/rollback, first work during drain, deferred admission, stale claims, + pause/resume, checkout reactivation, earlier deadline/cancellation, and a + 51-monitor backlog that needs a second bounded pass. +- Native finalization integration verifies that a dormant monitor worker wakes + when workspace settlement finishes and applies an expired monitor's existing + recovery policy without another provider turn. Existing manual and combined + scheduler cases still pass. The assembled heartbeat test exercises the new + monitor-specific entry point and its persisted wake/admission result. +- The lifecycle fixture waits for both finished reconciliation and no remaining + deadline before measuring quiet. Rollback may first require XID reconciliation; + drain may retain an intent hold until resume. Neither is counted as quiet + completion before the shared protocol actually finishes. +- `pnpm -r typecheck` passed. The server package build passed. `git diff --check` + passed. Repository-wide tests/build and live provider sleep qualification have + not been completed for this slice; the PR's CI gates remain separate evidence. + +## Limits and next duty + +The supported writers share one server process's database-work scope. Independent +SQL writers and other server replicas need their own ingress/reconciliation +contract. Active events can still cause inspection; the quiet claim concerns an +unchanged subsystem after settlement, not zero queries during active work. + +This slice does not prove host or database suspension. Cloud's existing durable +routine/monitor alarm proof remains unchanged. Legacy agent interval policy and +other remaining recurring duties still run. Migrate the legacy agent timer as +its own slice, preserving due claims, worktree cutoff, eligibility, and budgets. + +## Review corrections + +The original lifecycle fixture covered idle drain but missed ordinary maintenance +drain. Monitor registration now checks all task drains through the existing +admission owner. The shared deadline worker retains an unfinished request if +admission closes during either the run or the deadline read. It clears the old +local deadline in that state, so a past deadline cannot create an immediate spin; +a bounded in-memory retry remains. Explicit stop and TTL expiry restore dispatch. +Other deadline-worker consumers keep their existing admission predicates. + +Ordinary issue and execution-control writes can wake this queue. Their global +scans are now bounded by the configured heartbeat interval instead of one second. +Known saved deadlines still preempt that batching window. New notifications can +therefore take up to one interval to be inspected, matching the previous polling +latency. This does not add a periodic scan after the queue settles. + +New tests reproduce both findings before the fix. They cover mid-run and +mid-read drain transitions, ordinary drain stop/expiry, repeated notification +traffic, and repeated real checkout/release transactions. Integration fixtures +advance the scheduler clock between real I/O completions and disable driver +socket timers in that fixture; DB disconnect/reconnect has separate tests. The +operator writer contract is in [idle-issue-monitors.md](../idle-issue-monitors.md). + +Validation after review: eight complete suites passed (211 tests). The routine +suite initially had five failures because its unused monitor stub retained the +new batching deadline. That fixture now registers only the real routine worker; +all 17 routine tests pass separately. An intermediate test setup used a worker +stop method that is not exposed; it was removed before final verification. The +combined affected coverage is 228 passing tests, with no skips in the final +complete suite results. Server typechecking and build pass. These are local +results; current-head CI and remaining reviews are separate gates. + +## October 29 master reconciliation and current acceptance + +This existing issue-monitor slice was reconciled with the accepted routine parent +and its landed prerequisites through ordinary merges. Two conflicts required +explicit composition: monitor consumption keeps its new transaction and intent +while preserving master's `triggered` result after an already-admitted check loses +the replacement consumption fence; issue deletion keeps the scoped decision-target +removal notifier together with the new monitor intent. All remaining paths match +the automatic merge, and the package tree is identical to the qualified parent. + +Current source `c24aadfc91e9394d73c799f9e05fc67202160557` passed **1,233 cases +in 35 complete files**, without skips, in 458.16 seconds. Coverage includes +heartbeat scheduling and monitor admission, scheduled deadline queries, retained +routine lifecycles, native monitor preservation and run projection, execution +control, issue and tree services, conversation producers, budget/company changes, +startup, and the shared deadline helper's plugin, credential expiry, tool review, +decision, retention, custom-image and Grok consumers. Database build, production +server typechecking/build, module boundaries, migration order and Runner profile +identity pass. All selected hashes and temporary external-fixture imports were +restored exactly. Twenty-four fresh PostgreSQL 17 schemas, plus helper-owned +per-file databases, applied all 412 journal entries through index 413. + +The old remote head's CI failures are preserved separately from current acceptance. +Its delegated Slack rate assertion admitted an unexpected second send. A minute +boundary is consistent with the old live-clock fixture; the precise historical +cause is not established. The inherited parent fixture now freezes Date for all +rate variants, retaining the same denial and send-count assertions, and is byte +identical to merged #17835. The old Runner fixture also failed its initial +provider-drain deadline, before its deliberate timeout stage. Current packages +are byte identical to merged #17835's qualified profile-80 tree. These inherited +changes and local passes do not waive fresh final-head CI or security gates. + +The final routine-parent publication changed only its audit; all production trees +and these 35 test hashes remain identical to accepted source. Publication still +requires adopting the actual merged routine squash and new exact-head review. +No full local repository-suite, live provider, external alarm delivery, measured +instance/database suspension, production load or fleet cost claim is made. diff --git a/server/src/__tests__/issue-monitor-scheduler.test.ts b/server/src/__tests__/issue-monitor-scheduler.test.ts index 81ca8e227f3..23f951e3f1b 100644 --- a/server/src/__tests__/issue-monitor-scheduler.test.ts +++ b/server/src/__tests__/issue-monitor-scheduler.test.ts @@ -261,9 +261,9 @@ describeEmbeddedPostgres("issue monitor scheduler", () => { const unrelatedPolicy = { commentRequired: false, futureAuthorization: { preserve: true } }; await db.update(issues).set({ executionPolicy: { ...before.executionPolicy, ...unrelatedPolicy } }).where(eq(issues.id, issueId)); - const result = await heartbeat.tickTimers(tickAt); + const result = await heartbeat.tickIssueMonitors(tickAt); - expect(result.enqueued).toBe(1); + expect(result.triggered).toBe(1); const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0]!); expect(issue.monitorNextCheckAt).toBeNull(); diff --git a/server/src/__tests__/routine-schedule-work.test.ts b/server/src/__tests__/routine-schedule-work.test.ts index e9cb49de446..a2822bb6b8d 100644 --- a/server/src/__tests__/routine-schedule-work.test.ts +++ b/server/src/__tests__/routine-schedule-work.test.ts @@ -7,7 +7,8 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro 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 { registerDeadlineWork } from "../services/deadline-work.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -54,14 +55,16 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu const svc = service(); coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); - const workers = registerScheduledWork(coordinator, db, { retryMs: 30_000, + + + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.routineSchedules, { + retryMs: 30_000, canRun: () => !draining, - tickTimers: async () => {}, - tickRoutines: options.run ?? (signal => svc.tickScheduledTriggers(new Date(), () => !draining && !signal.aborted)), - nextRoutineRunAt: options.next ?? (() => svc.nextScheduledTriggerAt()), + run: options.run ?? (signal => svc.tickScheduledTriggers(new Date(), () => !draining && !signal.aborted)), + next: options.next ?? (() => svc.nextScheduledTriggerAt()), }); - await Promise.all([workers.routines.ready, workers.timers.ready]); - return workers; + await worker.ready; + return worker; } async function settled() { await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); diff --git a/server/src/__tests__/scheduled-work-postgres.test.ts b/server/src/__tests__/scheduled-work-postgres.test.ts index 46fd20fc4c7..13ce385ecb2 100644 --- a/server/src/__tests__/scheduled-work-postgres.test.ts +++ b/server/src/__tests__/scheduled-work-postgres.test.ts @@ -5,6 +5,7 @@ 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 { heartbeatService } from "../services/heartbeat.js"; import { routineService } from "../services/routines.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { registerScheduledWork } from "../services/scheduled-work.js"; @@ -60,12 +61,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 () => {}, canRun: () => true, + } }, { retryMs: 60_000, tickAgentTimers: async () => {}, tickIssueMonitors: async () => {}, + nextIssueMonitorAt: () => heartbeatService(db).nextIssueMonitorAt(), tickRoutines: async () => {}, canRun: () => true, nextRoutineRunAt: () => routineService(db).nextScheduledTriggerAt() }); for (const task of tasks.values()) await task.run(new AbortController().signal); 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()); + expect(await heartbeatService(db).nextIssueMonitorAt()).toBe(at(-1).getTime()); + expect(tasks.get(DELIVERY_QUEUES.issueMonitors)!.nextRunAt!()).toBe(now + 60_000); 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); diff --git a/server/src/__tests__/scheduled-work.test.ts b/server/src/__tests__/scheduled-work.test.ts index 79567e83899..63286eabf4c 100644 --- a/server/src/__tests__/scheduled-work.test.ts +++ b/server/src/__tests__/scheduled-work.test.ts @@ -1,19 +1,19 @@ import { PgDialect } from "drizzle-orm/pg-core"; -import type { Db } from "@paperclipai/db"; import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { afterEach, describe, expect, it, vi } from "vitest"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js"; import { registerScheduledWork } from "../services/scheduled-work.js"; -import { idleWorkSnapshot } from "../services/task-admission.js"; +import { idleWorkSnapshot, startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; const coordinators: ReturnType[] = []; afterEach(async () => { for (const coordinator of coordinators.splice(0)) await coordinator.stop(); + stopTaskDrain(); vi.useRealTimers(); }); -function fixture(input: { routine?: number | null; monitor?: number | null; tickTimers?: () => Promise; tickRoutines?: () => Promise } = {}) { +function fixture(input: { routine?: number | null; monitor?: number | null; tickIssueMonitors?: () => Promise; tickRoutines?: () => Promise } = {}) { vi.useFakeTimers(); vi.setSystemTime(new Date("2020-01-01T00:00:00.000Z")); const dialect = new PgDialect(); @@ -26,8 +26,7 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick const text = compiled.sql; if (text.includes("pg_current_xact_id")) return [{ xid: "42" }]; if (failRead) throw new Error("database unavailable"); - const at = deadlines[text.includes("routine_triggers") ? "routine" : "monitor"]; - return [{ next_at: at !== null && at > Date.parse(String(compiled.params[0])) ? new Date(at) : null }]; + throw new Error(`Unexpected database query: ${text}`); }); const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise) { return callback({ execute, transaction: owner.transaction }); @@ -35,7 +34,14 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick const errors = vi.fn(); const coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, canReconcile: () => canReconcile, onError: errors }); coordinators.push(coordinator); - const tickTimers = vi.fn(input.tickTimers ?? (async () => {})); + const tickIssueMonitors = vi.fn(input.tickIssueMonitors ?? (async () => { + if (deadlines.monitor !== null && deadlines.monitor <= Date.now()) deadlines.monitor = null; + })); + const tickAgentTimers = vi.fn(async () => {}); + const nextIssueMonitorAt = vi.fn(async () => { + if (failRead) throw new Error("database unavailable"); + return deadlines.monitor; + }); const tickRoutines = vi.fn(input.tickRoutines ?? (async () => { if (deadlines.routine !== null && deadlines.routine <= Date.now()) deadlines.routine = null; })); @@ -43,14 +49,60 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick if (failRead) throw new Error("database unavailable"); return deadlines.routine; }); - const workers = registerScheduledWork(coordinator, owner as unknown as Db, - { 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]), + const workers = registerScheduledWork(coordinator, + { retryMs: 60_000, tickAgentTimers, tickIssueMonitors, nextIssueMonitorAt, tickRoutines, nextRoutineRunAt, canRun: () => canRun }); + return { coordinator, workers, owner, deadlines, execute, nextRoutineRunAt, nextIssueMonitorAt, tickAgentTimers, tickIssueMonitors, tickRoutines, errors, + ready: () => Promise.all([workers.routines.ready, workers.monitors.ready, workers.timers.ready]), canRun: (value: boolean) => { canRun = value; }, canReconcile: (value: boolean) => { canReconcile = value; }, failRead: (value: boolean) => { failRead = value; } }; } describe("scheduled routine and monitor deadlines", () => { + it.each(["run", "deadline read"])("retains a monitor pass interrupted during %s by a maintenance drain", async boundary => { + const f = fixture({ tickIssueMonitors: async () => { if (boundary === "run") startTaskDrain(); } }); + if (boundary === "deadline read") f.nextIssueMonitorAt.mockImplementationOnce(async () => { startTaskDrain(); return null; }); + await f.ready(); + expect(f.coordinator.nextWakeAt()).not.toBeNull(); + expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(boundary === "run" ? 0 : 1); + await vi.advanceTimersByTimeAsync(120_000); + expect(f.tickIssueMonitors).toHaveBeenCalledOnce(); + f.tickIssueMonitors.mockImplementation(async () => {}); + stopTaskDrain(); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(2); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + + it("bounds notification scans to the configured cadence while preserving known deadlines", async () => { + const origin = Date.UTC(2020, 0, 1); + const f = fixture({ monitor: origin + 150_000 }); + await f.ready(); + for (let second = 0; second < 120; second++) { + await f.owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors)); + await vi.advanceTimersByTimeAsync(1000); + } + expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(3); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(3); + await vi.advanceTimersByTimeAsync(30_000); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(4); + expect(f.coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(120_000); + expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(4); + }); + + it("legacy agent timer wakes do not query or dispatch issue monitors", async () => { + const f = fixture(); + await f.ready(); + const before = f.nextIssueMonitorAt.mock.calls.length; + for (let i = 0; i < 5; i++) { + f.workers.timers.wake(); + await vi.advanceTimersByTimeAsync(60_000); + } + expect(f.tickAgentTimers).toHaveBeenCalledTimes(6); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(1); + expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(before); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("dispatches at each saved deadline without waiting for a periodic tick", async () => { const now = Date.UTC(2020, 0, 1); const f = fixture({ routine: now + 10_000, monitor: now + 20_000 }); @@ -61,16 +113,16 @@ describe("scheduled routine and monitor deadlines", () => { expect(f.tickRoutines).toHaveBeenCalledTimes(1); await vi.advanceTimersByTimeAsync(1); expect(f.tickRoutines).toHaveBeenCalledTimes(2); - expect(f.tickTimers).toHaveBeenCalledTimes(1); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(1); expect(f.coordinator.nextWakeAt()).toBe(now + 20_000); await vi.advanceTimersByTimeAsync(10_000); - expect(f.tickTimers).toHaveBeenCalledTimes(2); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(2); expect(f.coordinator.nextWakeAt()).toBeNull(); - expect(f.execute).toHaveBeenCalledTimes(2); + expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(2); expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(2); expect(idleWorkSnapshot().active).toBe(0); await vi.advanceTimersByTimeAsync(3_600_000); - expect(f.execute).toHaveBeenCalledTimes(2); + expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(2); expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(2); }); @@ -78,16 +130,16 @@ describe("scheduled routine and monitor deadlines", () => { let release!: () => void; const held = new Promise(resolve => { release = resolve; }); const tick = vi.fn().mockImplementationOnce(() => held).mockImplementation(async () => { - if (kind === "routine") f.deadlines.routine = null; + f.deadlines[kind] = null; }); const now = Date.UTC(2020, 0, 1); const f = fixture({ [kind]: now + 10_000, - ...(kind === "routine" ? { tickRoutines: tick } : { tickTimers: tick }) }); + ...(kind === "routine" ? { tickRoutines: tick } : { tickIssueMonitors: tick }) }); try { - await (kind === "routine" ? f.workers.timers.ready : f.workers.routines.ready); + await (kind === "routine" ? f.workers.monitors.ready : f.workers.routines.ready); await vi.advanceTimersByTimeAsync(12_000); } finally { release(); await f.ready(); } - const expected = kind === "routine" ? now + 60_000 : Date.now(); + const expected = now + 60_000; expect(f.coordinator.nextWakeAt()).toBe(expected); await vi.advanceTimersByTimeAsync(expected - Date.now() + 1); expect(tick).toHaveBeenCalledTimes(2); @@ -95,15 +147,15 @@ describe("scheduled routine and monitor deadlines", () => { expect(idleWorkSnapshot().active).toBe(0); }); - it("retains capped overdue work with bounded spacing, then disarms after completion", async () => { + it.each(["routine", "monitor"] as const)("retains capped overdue %s work with bounded spacing, then disarms after completion", async kind => { const tick = vi.fn(async () => {}); const now = Date.UTC(2020, 0, 1); - const f = fixture({ routine: now - 10_000, tickRoutines: tick }); + const f = fixture({ [kind]: now - 10_000, ...(kind === "routine" ? { tickRoutines: tick } : { tickIssueMonitors: 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; + f.deadlines[kind] = null; await vi.advanceTimersByTimeAsync(1); expect(tick).toHaveBeenCalledTimes(2); expect(f.coordinator.nextWakeAt()).toBeNull(); @@ -133,7 +185,7 @@ describe("scheduled routine and monitor deadlines", () => { expect(idleWorkSnapshot().active).toBe(0); await vi.advanceTimersByTimeAsync(8999); expect(f.tickRoutines).toHaveBeenCalledTimes(3); - expect(f.tickTimers).toHaveBeenCalledTimes(1); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(1); }); it("keeps monitor dispatch responsive while routine work is slow", async () => { @@ -141,9 +193,9 @@ describe("scheduled routine and monitor deadlines", () => { const held = new Promise(resolve => { release = resolve; }); const f = fixture({ monitor: Date.UTC(2020, 0, 1) + 10_000, tickRoutines: () => held }); try { - await f.workers.timers.ready; + await f.workers.monitors.ready; await vi.advanceTimersByTimeAsync(10_000); - expect(f.tickTimers).toHaveBeenCalledTimes(2); + expect(f.tickIssueMonitors).toHaveBeenCalledTimes(2); expect(f.tickRoutines).toHaveBeenCalledTimes(1); } finally { release(); await f.workers.routines.ready; } }); @@ -157,7 +209,7 @@ describe("scheduled routine and monitor deadlines", () => { expect(idleWorkSnapshot().active).toBe(0); f.canReconcile(false); await vi.advanceTimersByTimeAsync(60_000); - expect(f.execute).toHaveBeenCalledTimes(1); + expect(f.nextIssueMonitorAt).toHaveBeenCalledTimes(1); expect(f.nextRoutineRunAt).toHaveBeenCalledTimes(1); await f.coordinator.stop(); f.canRun(true); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bd46bc6b45d..786ee772bd2 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -154,7 +154,9 @@ const { sweepStaleIssueLocks: vi.fn(async () => ({ cleared: 0 })), sweepPendingCleanupLeases: vi.fn(async () => ({ swept: 0, destroyed: 0, capped: 0 })), sweepExpiredRuntimeStatuses: vi.fn(() => 0), - tickTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })), + tickAgentTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })), + tickIssueMonitors: vi.fn(async () => ({ checked: 0, triggered: 0, skipped: 0 })), + nextIssueMonitorAt: vi.fn(async () => null), }; const heartbeatServiceFactoryMock = vi.fn(() => heartbeatServiceMock); const issueThreadInteractionServiceMock = { @@ -1168,7 +1170,7 @@ describe("startServer feedback export wiring", () => { await settleStartupHousekeeping(); expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled(); - expect(heartbeatServiceMock.tickTimers).not.toHaveBeenCalled(); + expect(heartbeatServiceMock.tickAgentTimers).not.toHaveBeenCalled(); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); expect(intervalCallback).not.toBeNull(); @@ -1176,7 +1178,7 @@ describe("startServer feedback export wiring", () => { await Promise.resolve(); await Promise.resolve(); - expect(heartbeatServiceMock.tickTimers).not.toHaveBeenCalled(); + expect(heartbeatServiceMock.tickAgentTimers).not.toHaveBeenCalled(); expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); diff --git a/server/src/index.ts b/server/src/index.ts index 6424a50d991..9e49cfb6a17 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -2078,21 +2078,24 @@ async function startServerWithDatabaseTeardown( })); }; const startHeartbeatScheduler = () => { - const scheduled = registerScheduledWork(deliveryWork, db, { + const scheduled = registerScheduledWork(deliveryWork, { retryMs: config.heartbeatSchedulerIntervalMs, - tickTimers: async () => { + tickAgentTimers: async () => { if ((await heartbeat.resolveSchedulingSuppression()).suppressed) return; - const result = await heartbeat.tickTimers(new Date()); + const result = await heartbeat.tickAgentTimers(new Date()); if (result.enqueued > 0) logger.info({ ...result }, "heartbeat timer tick enqueued runs"); }, canRun: () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, + nextIssueMonitorAt: () => heartbeat.nextIssueMonitorAt(), + tickIssueMonitors: signal => heartbeat.tickIssueMonitors(new Date(), + () => !signal.aborted && !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"); }), }); - trackHeartbeatSchedulerWork(Promise.all([scheduled.timers.ready, scheduled.routines.ready])); + trackHeartbeatSchedulerWork(Promise.all([scheduled.timers.ready, scheduled.monitors.ready, scheduled.routines.ready])); startHeartbeatSchedulerInterval(createHeartbeatSchedulerTick(scheduled)); }; trackHeartbeatSchedulerWork(runStartupHousekeeping().then(() => { diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index fee9d5510ee..66c63898d73 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; @@ -347,6 +348,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId: input.sourceIssue.id }); await notifyNativeStatusReassessment(tx, { companyId, issueId: input.sourceIssue.id }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const [updatedRun] = await tx .update(heartbeatRuns) .set({ diff --git a/server/src/services/access.ts b/server/src/services/access.ts index 81a0c7222c7..77b2e648d6c 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatCompanyAuthorityChange } from "./chat-run-milestone-notifications.js"; @@ -742,6 +743,7 @@ export function accessService(db: Db) { sql`${issues.status} not in ('done', 'cancelled')`, ); await notifyNativeStatusReassessment(tx, { companyId }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const resetInProgress = await tx .update(issues) .set({ diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index 4cafdf507b1..974572df305 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared"; import { @@ -268,6 +269,7 @@ export async function prepareConversationTurn( context.conversationReplayThroughCommentId = replayThrough?.id ?? commentId; } } + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await tx .update(issues) .set({ @@ -386,6 +388,7 @@ export async function settleConversationTurn( issue.conversationState === conversationState ) return true; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); if (status === "in_review") await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); await tx .update(issues) diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index 32afaee0865..602e534dcfc 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; @@ -35,6 +36,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str }); await notifyExternalConversationSettlement(tx, { companyId }); await issueThreadInteractionService(tx).cancelPendingForDeletedAddressee(companyId, id); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await tx.update(agents).set({ reportsTo: null }).where(eq(agents.reportsTo, id)); await tx .update(issues) diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index f32e28dbaf1..cbe88f735ed 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -302,6 +302,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (scope.paused) return; await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); + await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); await db @@ -360,6 +361,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (scope.pauseReason !== "budget") return; await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); + await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); 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 8e1f34b889d..999c89e17b3 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -334,6 +334,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 notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyCompanyRetentionWork(tx, id); await notifyChatInboundWakeNotices(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); @@ -488,6 +489,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); 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 2a4fb16e37d..7a9b0e55052 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -75,6 +75,7 @@ export async function deleteCompany(db: Db, id: string) { await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); 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 14c5cad5465..b552dc38598 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -23,7 +23,7 @@ export function registerDeadlineWork( pausePassiveMaintenance: true, onSettled: () => { requested++; }, run: async signal => { - if (!task.canRun()) { requested++; return; } + if (!task.canRun()) { requested++; next = null; return; } if (Date.now() < earliest) { next = earliest; return; } if (Date.now() < coalesceUntil && (next === null || next > Date.now())) { next = next === null ? coalesceUntil : Math.min(next, coalesceUntil); @@ -33,8 +33,12 @@ export function registerDeadlineWork( const startedAt = Date.now(); earliest = startedAt + task.retryMs; await task.run(signal); - if (signal.aborted) return; + if (signal.aborted || !task.canRun()) { next = null; return; } const deadline = await task.next(); + + + + if (!task.canRun()) { next = null; return; } // Back off failed or still-overdue work, not a successfully read future deadline. earliest = deadline === null || deadline > Date.now() ? 0 : startedAt + (task.continuationMs ?? task.retryMs); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index f5e68e69690..39f2d1d762f 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -110,6 +110,7 @@ export const DELIVERY_QUEUES = { terminalWorkspaces: "terminal-workspaces", routineSchedules: "routine-schedules", issueMonitors: "issue-monitors", + agentTimers: "agent-timers", caseHandoffs: "case-handoffs", nativeCheckpointExpiration: "native-checkpoint-expiration", } as const; diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index fb59deafbe8..22c8913bbcd 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -3,7 +3,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./deliv export async function notifyExecutionRunWork(transaction: object) { for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, - DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition]) { + DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { await notifyDeliveryWork(transaction, queue); } } @@ -14,6 +14,7 @@ export function writeExecutionControl( write: (tx: Db) => PromiseLike, ): Promise { return db.transaction(async transaction => { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors); for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); return write(transaction as unknown as Db); }); diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index 725fba0eb55..b7c86dcc40f 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -594,6 +594,7 @@ export async function settleUnrecoverableExecutions( if (current) { await notifyNativeStatusReassessment(tx, { companyId: task.companyId, issueId: task.id }); if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const [projected] = await tx .update(issues) .set({ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index dd894aeb586..1076558001b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1560,6 +1560,9 @@ export function heartbeatService( const { triggerIssueMonitor, tickTimers, + tickAgentTimers, + tickIssueMonitors, + nextIssueMonitorAt, recoverActiveSessionGoals, recoverPendingSessionGoalActions, } = createHeartbeatScheduling(db, { @@ -4264,6 +4267,9 @@ export function heartbeatService( buildRunOutputSilence, tickTimers, + tickAgentTimers, + tickIssueMonitors, + nextIssueMonitorAt, cancelRun: (runId: string, reason?: string, options?: CancelRunOptions) => cancelRunInternal(runId, reason, options), diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 03e95a7f463..bf1dad379ee 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery, notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; @@ -3904,6 +3905,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ].join("\n"); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id }); await tx .update(issues) diff --git a/server/src/services/heartbeat/scheduling.test.ts b/server/src/services/heartbeat/scheduling.test.ts index 26b84d48b82..4e7535c8455 100644 --- a/server/src/services/heartbeat/scheduling.test.ts +++ b/server/src/services/heartbeat/scheduling.test.ts @@ -1,12 +1,22 @@ import { randomUUID } from "node:crypto"; +import { setTimeout as realDelay } from "node:timers/promises"; import type { Db } from "@paperclipai/db"; import { - activityLog, agents, agentSessionGoalActions, agentTaskSessions, companies, createDb, + activityLog, applyPendingMigrations, reconcileDatabaseWork, agents, agentSessionGoalActions, agentTaskSessions, companies, createDb, completionContracts, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; -import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { heartbeatService } from "../heartbeat.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { companyService } from "../companies.js"; +import { issueService } from "../issues.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { registerScheduledWork } from "../scheduled-work.js"; +import { idleWorkSnapshot, readTaskDrain, startTaskDrain, stopTaskDrain } from "../task-admission.js"; import { conflict } from "../../errors.js"; import { createHeartbeatRunPreparation } from "./run-preparation.js"; import { createHeartbeatRunState } from "./run-state.js"; @@ -63,7 +73,7 @@ describe("heartbeat scheduling module boundary", () => { for (const callback of Object.values(deps)) if (vi.isMockFunction(callback)) expect(callback).not.toHaveBeenCalled(); }); - it.each(["tickTimers", "recoverActiveSessionGoals", "recoverPendingSessionGoalActions"] as const)( + it.each(["tickTimers", "tickAgentTimers", "tickIssueMonitors", "recoverActiveSessionGoals", "recoverPendingSessionGoalActions"] as const)( "honors suppression before %s reads or admits work", async (operation) => { const database = guardedDatabase(); const deps = callbacks(database.db); @@ -77,19 +87,24 @@ describe("heartbeat scheduling module boundary", () => { ); }); -const postgres = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HEARTBEAT_SCHEDULING_TEST_DATABASE_URL"); +const postgres = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", () => { let database: Awaited>; let db: Db; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-scheduling-module"); - db = createDb(database.connectionString); + if (external) await applyPendingMigrations(external); + else database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-scheduling-module"); + + + db = createDb(external ?? database.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0, connectTimeoutSeconds: 0 }); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); beforeEach(async () => { + await reconcileDatabaseWork(db, DELIVERY_QUEUES.issueMonitors); await db.execute(sql`truncate table companies restart identity cascade`); }); @@ -319,7 +334,7 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", ( expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(1); }); - it.each(["scheduled", "manual"] as const)("applies expired monitor recovery after workspace settlement without another turn (%s)", async (dispatch) => { + it.each(["scheduled", "manual", "worker"] as const)("applies expired monitor recovery after workspace settlement without another turn (%s)", async (dispatch) => { const f = await fixture(); const start = new Date(); const nextCheckAt = new Date(start.getTime() + 60_000).toISOString(); @@ -364,17 +379,31 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", ( const expiredAt = new Date(start.getTime() + 180_000); vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(expiredAt); + let monitorWorker: ReturnType | undefined; try { await expect(nativeCompletionFeedback(db, run!.id, result)).rejects.toThrow("persisted, eligible monitor"); expect(await f.scheduler.tickTimers(expiredAt)).toEqual({ checked: 0, enqueued: 0, skipped: 0 }); await expect(f.scheduler.triggerIssueMonitor(f.issue.id, { now: expiredAt })).rejects.toThrow("waiting for the previous run to settle"); + if (dispatch === "worker") { + monitorWorker = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const workers = registerScheduledWork(monitorWorker, { retryMs: 1000, canRun: () => true, + tickAgentTimers: async () => {}, tickRoutines: async () => {}, nextRoutineRunAt: async () => null, + tickIssueMonitors: signal => f.scheduler.tickIssueMonitors(new Date(), () => !signal.aborted), + nextIssueMonitorAt: () => f.scheduler.nextIssueMonitorAt(), + }); + await Promise.all([workers.monitors.ready, workers.routines.ready, workers.timers.ready]); + expect(monitorWorker.nextWakeAt()).toBeNull(); + } await finalizeNativeRun({ db, runId: run!.id, workspaceFinalizeStatus: "succeeded" }); expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run!.id)))[0]) .toMatchObject({ phase: "committed", failureCode: null }); expect(await readIssue(f.issue.id)).toMatchObject({ executionRunId: null, checkoutRunId: null, monitorNextCheckAt: new Date(nextCheckAt) }); if (dispatch === "manual") await f.scheduler.triggerIssueMonitor(f.issue.id, { now: expiredAt }); - else await f.scheduler.tickTimers(expiredAt); + else if (dispatch === "worker") { + vi.setSystemTime(new Date(expiredAt.getTime() + 2000)); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull(), { timeout: 5000 }); + } else await f.scheduler.tickTimers(expiredAt); expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: null, executionState: { monitor: { status: "cleared", clearReason: "timeout_exceeded" } } }); expect(await db.select().from(issueComments).where(eq(issueComments.issueId, f.issue.id))).toHaveLength(1); @@ -384,6 +413,7 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", ( expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, saved!.id)))[0]!.resultJson).toEqual(saved!.resultJson); } finally { + await monitorWorker?.stop(); vi.useRealTimers(); } }); @@ -482,4 +512,300 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", ( const failed = await db.select().from(agentSessionGoalActions).where(eq(agentSessionGoalActions.id, action!.id)); expect(failed[0]).toMatchObject({ status: "failed", error: "session_goal_control_payload_invalid" }); }); + describe("issue monitor deadline lifecycle", () => { + let coordinator: ReturnType | undefined; + let draining = false; + const errors: unknown[] = []; + beforeEach(() => { + draining = false; errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + vi.setSystemTime(now); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + stopTaskDrain(); + vi.useRealTimers(); vi.restoreAllMocks(); + expect(errors).toEqual([]); + }); + async function start(f: { scheduler: Pick, "tickIssueMonitors" | "nextIssueMonitorAt"> }) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + const workers = registerScheduledWork(coordinator, { retryMs: 30_000, canRun: () => !draining, + tickAgentTimers: async () => {}, tickRoutines: async () => {}, nextRoutineRunAt: async () => null, + tickIssueMonitors: signal => f.scheduler.tickIssueMonitors(new Date(), () => !draining && !signal.aborted), + nextIssueMonitorAt: () => f.scheduler.nextIssueMonitorAt(), + }); + await Promise.all([workers.monitors.ready, workers.routines.ready, workers.timers.ready]); + return workers; + } + async function advance(milliseconds: number) { + + + for (let remaining = milliseconds; remaining > 0; remaining -= 1000) { + await vi.advanceTimersByTimeAsync(Math.min(1000, remaining)); + await realDelay(5); + } + } + async function quiet() { + + await advance(31_000); + if (!draining) await vi.waitFor(() => { + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction"); + try { + await advance(120_000); + expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); execute.mockRestore(); transaction.mockRestore(); } + } + async function schedule(f: Awaited>, at: Date, owner = db) { + return issueService(owner).update(f.issue.id, { monitorNextCheckAt: at, executionPolicy: { + mode: "normal", commentRequired: true, stages: [], monitor: { nextCheckAt: at.toISOString() }, + } }); + } + + it.each(["commit", "rollback"] as const)("reconciles worktree activation only after its outer transaction %s", async outcome => { + const f = await fixture(); + const deadline = now.getTime() + 3_600_000; + await db.update(issues).set({ monitorNextCheckAt: new Date(deadline), + executionPolicy: { monitor: { nextCheckAt: new Date(deadline).toISOString() } } }).where(eq(issues.id, f.issue.id)); + const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "monitor-activation-fixture" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const heartbeat = heartbeatService(db, { runtimeEnv }); + await start({ scheduler: heartbeat }); + await quiet(); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const writing = db.transaction(async tx => { + await instanceSettingsService(tx as unknown as Db, { runtimeEnv }).updateExperimental({ enableWorktreeRunExecution: true }); + entered(); + await held; + if (outcome === "rollback") throw new Error("rollback activation"); + }).then(() => undefined, error => error as Error); + try { + await ready; + await advance(31_000); + + + expect(await heartbeat.nextIssueMonitorAt()).toBeNull(); + expect((await readIssue(f.issue.id)).monitorAttemptCount).toBe(0); + } finally { release(); } + const error = await writing; + if (outcome === "rollback") { + expect(error?.message).toBe("rollback activation"); + await quiet(); + } else { + expect(error).toBeUndefined(); + await advance(31_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(deadline), { timeout: 5000 }); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + await quiet(); + } + expect(await readIssue(f.issue.id)).toMatchObject({ monitorNextCheckAt: new Date(deadline), monitorAttemptCount: 0 }); + }); + + it("uses fresh worktree activation for immediate monitor deadline inspection", async () => { + const f = await fixture(); + const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "monitor-fresh-fixture" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const heartbeat = heartbeatService(db, { runtimeEnv }); + expect(await heartbeat.nextIssueMonitorAt()).toBeNull(); + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + expect(await heartbeat.nextIssueMonitorAt()).toBe(dueAt.getTime()); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + expect(await heartbeat.nextIssueMonitorAt()).toBeNull(); + expect(await heartbeat.tickIssueMonitors()).toEqual({ checked: 0, triggered: 0, skipped: 0 }); + expect((await readIssue(f.issue.id)).monitorNextCheckAt).toEqual(dueAt); + }); + + it("stays quiet with retained history, then a real producer schedules and consumes its deadline", async () => { + const f = await fixture(); + await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id)); + await db.insert(issues).values({ companyId: f.company.id, title: "Old monitor", status: "done", + assigneeAgentId: f.agent.id, monitorNextCheckAt: dueAt }); + await start(f); await quiet(); + expect(coordinator!.nextWakeAt()).toBeNull(); + const at = new Date(Date.now() + 10_000); + await schedule(f, at); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(at.getTime())); + await advance(10_000); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull()); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + await quiet(); expect(coordinator!.nextWakeAt()).toBeNull(); + }); + + it.each([false, true])("waits for the outer producer transaction and handles rollback=%s", async rollback => { + const f = await fixture(); + await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id)); + await start(f); + let release!: () => void, entered!: () => void; + const hold = new Promise(r => { release = r; }); + const written = new Promise(r => { entered = r; }); + const at = new Date(Date.now() + 60_000); + const writing = db.transaction(async tx => { + await schedule(f, at, tx as unknown as Db); + entered(); await hold; + if (rollback) throw new Error("rollback monitor"); + }).catch(error => { if (!rollback) throw error; expect(error.message).toBe("rollback monitor"); }); + try { + await written; + await advance(2000); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull(); + } finally { release(); await writing; } + if (rollback) { + await advance(30_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await quiet(); + } + else { + await advance(30_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(at.getTime())); + } + }); + + it.each(["stop", "expiry"] as const)("resumes a saved monitor after maintenance drain %s", async end => { + const f = await fixture(); + const at = new Date(Date.now() + 10_000); + await db.update(issues).set({ monitorNextCheckAt: at }).where(eq(issues.id, f.issue.id)); + f.deps.getSchedulingSuppression.mockImplementation(async () => ({ suppressed: readTaskDrain(new Date()) !== null })); + await start(f); + startTaskDrain({ ttlMs: end === "expiry" ? 90_000 : null }); + const select = vi.spyOn(db, "select"); + try { + await advance(60_000); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + expect(select).not.toHaveBeenCalled(); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + } finally { select.mockRestore(); } + if (end === "stop") stopTaskDrain(); + await advance(60_000); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull()); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + await quiet(); + }); + + it("retains a first producer during drain without passive SQL, then resumes", async () => { + const f = await fixture(); + await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id)); + await start(f); draining = true; + await schedule(f, new Date(Date.now() + 10_000)); + await quiet(); expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + draining = false; + await advance(30_000); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull()); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + await quiet(); + }); + + it("backs off a deferred admission and disarms after the retry succeeds", async () => { + const f = await fixture(); f.deps.enqueueWakeup.mockResolvedValueOnce(null); + await start(f); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + await advance(5000); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await advance(30_000); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull()); + expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(2); + await quiet(); + }); + + it("retains the overdue remainder of a capped pass until the next bounded drain", async () => { + const f = await fixture(); + await db.insert(issues).values(Array.from({ length: 50 }, (_, index) => ({ + companyId: f.company.id, title: `Due monitor ${index}`, status: "in_progress", assigneeAgentId: f.agent.id, + monitorNextCheckAt: dueAt, executionPolicy: f.issue.executionPolicy, + }))); + await start(f); + expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(50); + expect(await f.scheduler.nextIssueMonitorAt()).toBe(dueAt.getTime()); + await advance(30_000); + await vi.waitFor(() => expect(f.deps.enqueueWakeup).toHaveBeenCalledTimes(51)); + await quiet(); + expect(await f.scheduler.nextIssueMonitorAt()).toBeNull(); + }, 30_000); + + it("replaces an earlier deadline and cancels it through the ordinary issue producer", async () => { + const f = await fixture(); + await db.update(issues).set({ monitorNextCheckAt: null, executionPolicy: null }).where(eq(issues.id, f.issue.id)); + await start(f); + const later = new Date(Date.now() + 600_000); + await schedule(f, later); + await advance(30_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(later.getTime()), { timeout: 5000 }); + const earlier = new Date(Date.now() + 60_000); + await schedule(f, earlier); + await advance(30_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(earlier.getTime()), { timeout: 5000 }); + await issueService(db).update(f.issue.id, { monitorNextCheckAt: null, executionPolicy: null }); + await quiet(); + await advance(600_000); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("reconstructs a claim deadline and respects its strict five-minute fence", async () => { + const f = await fixture(); + const claimedAt = new Date(); + await db.update(issues).set({ monitorWakeRequestedAt: claimedAt }).where(eq(issues.id, f.issue.id)); + expect(await f.scheduler.nextIssueMonitorAt()).toBe(claimedAt.getTime() + 300_001); + expect(await f.scheduler.tickIssueMonitors(new Date(claimedAt.getTime() + 300_000))).toMatchObject({ triggered: 0 }); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(claimedAt.getTime() + 300_001); + await advance(300_001); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull()); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + await quiet(); + }); + + it("disarms while the company is paused and reactivates on its committed resume", async () => { + const f = await fixture(); + await companyService(db).update(f.company.id, { status: "paused" }); + await start(f); await quiet(); + expect(coordinator!.nextWakeAt()).toBeNull(); + await companyService(db).update(f.company.id, { status: "active" }); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull()); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + await quiet(); + }); + + it("bounds global scans during repeated ordinary checkout and release writes", async () => { + const f = await fixture(); + const at = new Date(Date.now() + 300_000); + await db.update(issues).set({ status: "todo", monitorNextCheckAt: at }).where(eq(issues.id, f.issue.id)); + const next = vi.spyOn(f.scheduler, "nextIssueMonitorAt"); + const tick = vi.spyOn(f.scheduler, "tickIssueMonitors"); + await start(f); + const service = issueService(db); + for (let i = 0; i < 12; i++) { + await service.checkout(f.issue.id, f.agent.id, ["todo"], null); + await service.release(f.issue.id, f.agent.id, null); + await advance(2000); + } + expect(next).toHaveBeenCalledOnce(); + expect(tick).toHaveBeenCalledOnce(); + await service.checkout(f.issue.id, f.agent.id, ["todo"], null); + await advance(6000); + await vi.waitFor(() => expect(next).toHaveBeenCalledTimes(2)); + expect(tick).toHaveBeenCalledTimes(2); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(at.getTime())); + expect(f.deps.enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("checkout reactivates a retained monitor on a formerly ineligible task", async () => { + const f = await fixture(); + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.issue.id)); + await start(f); await quiet(); + expect(coordinator!.nextWakeAt()).toBeNull(); + await issueService(db).checkout(f.issue.id, f.agent.id, ["todo"], null); + await vi.waitFor(async () => expect((await readIssue(f.issue.id)).monitorNextCheckAt).toBeNull()); + expect(f.deps.enqueueWakeup).toHaveBeenCalledOnce(); + await quiet(); + }); + }); + }); diff --git a/server/src/services/heartbeat/scheduling.ts b/server/src/services/heartbeat/scheduling.ts index c627fdc0e99..e6d8af61a70 100644 --- a/server/src/services/heartbeat/scheduling.ts +++ b/server/src/services/heartbeat/scheduling.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyIssueReviewState } from "../issue-review-state-notifications.js"; import { insertIssueComment } from "../issue-comment-writes.js"; import { @@ -72,7 +73,7 @@ export interface HeartbeatSchedulingDependencies Pick { issuesSvc: Pick, "listReviewAttention" | "create">; scheduleBoundedRetryForRun: ReturnType["scheduleBoundedRetryForRun"]; - getSchedulingSuppression: () => Promise<{ suppressed: boolean }>; + getSchedulingSuppression: (fresh?: boolean) => Promise<{ suppressed: boolean }>; } function readNonEmptyString(value: unknown): string | null { @@ -591,6 +592,7 @@ export function createHeartbeatScheduling( const publications: ActivityPublication[] = []; try { await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const released = await tx.update(issues) .set({ monitorWakeRequestedAt: null, updatedAt: input.now }) .where(issueMonitorClaimCondition(claimed)).returning({ id: issues.id }); @@ -622,8 +624,11 @@ export function createHeartbeatScheduling( } catch (err) { // Diagnostics are optional. Release the original claim even if audit storage // failed, and never clobber a monitor that was replaced in the meantime. - await db.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now }) - .where(issueMonitorClaimCondition(claimed)); + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await tx.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now }) + .where(issueMonitorClaimCondition(claimed)); + }); logger.warn({ err, issueId: claimed.id }, "issue monitor diagnostic could not be recorded"); } } @@ -819,8 +824,9 @@ export function createHeartbeatScheduling( } } - const consumed = await db - .update(issues) + const consumed = await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + return tx.update(issues) .set({ ...monitorOnlyDispatchPatch(buildIssueMonitorTriggeredPatch({ issue: claimed, @@ -831,8 +837,7 @@ export function createHeartbeatScheduling( }) .where(issueMonitorClaimCondition(claimed)) .returning({ id: issues.id }); - // Admission already accepted the check. A replacement schedule must stay - // intact, but losing its consumption fence does not undo that accepted wake. + }); if (consumed.length === 0) return { outcome: "triggered" as const }; await logActivity(db, { @@ -970,6 +975,7 @@ export function createHeartbeatScheduling( const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000); const claimed = await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const [updated] = await tx .update(issues) .set({ @@ -1012,7 +1018,7 @@ export function createHeartbeatScheduling( }); } - async function tickDueIssueMonitors(now = new Date()) { + async function tickDueIssueMonitors(now: Date, canContinue: () => boolean) { const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000); const dueMonitors = await db .select(issueMonitorDispatchColumns) @@ -1040,7 +1046,9 @@ export function createHeartbeatScheduling( let skipped = 0; for (const due of dueMonitors) { + if (!canContinue()) break; const claimed = await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const [updated] = await tx .update(issues) .set({ @@ -1067,6 +1075,7 @@ export function createHeartbeatScheduling( }); if (!claimed) continue; + if (!canContinue()) break; try { const result = await dispatchClaimedIssueMonitor(claimed, { @@ -1095,6 +1104,26 @@ export function createHeartbeatScheduling( }; } + async function tickIssueMonitors(now = new Date(), canContinue: () => boolean = () => true) { + if (!canContinue() || (await getSchedulingSuppression(true)).suppressed) { + return { checked: 0, triggered: 0, skipped: 0 }; + } + return tickDueIssueMonitors(now, canContinue); + } + + async function nextIssueMonitorAt(): Promise { + if ((await getSchedulingSuppression(true)).suppressed) return null; + const [row] = await db.select({ at: sql`min(greatest(${issues.monitorNextCheckAt}, + ${issues.monitorWakeRequestedAt} + interval '5 minutes' + interval '1 millisecond'))` }) + .from(issues).innerJoin(companies, eq(companies.id, issues.companyId)) + .where(and(eq(companies.status, "active"), noActiveNativeMonitorRun(), + sql`${issues.monitorNextCheckAt} is not null`, isNull(issues.assigneeUserId), + sql`${issues.assigneeAgentId} is not null`, inArray(issues.status, ["in_progress", "in_review"]))); + + + return row?.at == null ? null : new Date(row.at).getTime(); + } + async function recoverActiveSessionGoals() { if ((await getSchedulingSuppression()).suppressed) { return { scanned: 0, enqueued: 0 }; @@ -1256,7 +1285,7 @@ export function createHeartbeatScheduling( return { scanned: pending.length, enqueued, alreadyQueued, invalid }; } - async function tickTimers(now = new Date()) { + async function tickAgentTimers(now = new Date()) { if ((await getSchedulingSuppression()).suppressed) { return { checked: 0, @@ -1334,18 +1363,24 @@ export function createHeartbeatScheduling( else skipped += 1; } - const issueMonitors = await tickDueIssueMonitors(now); - - return { - checked: checked + issueMonitors.checked, - enqueued: enqueued + issueMonitors.triggered, - skipped: skipped + issueMonitors.skipped, - }; + return { checked, enqueued, skipped }; } + + + async function tickTimers(now = new Date()) { + const timers = await tickAgentTimers(now); + const monitors = await tickIssueMonitors(now); + return { checked: timers.checked + monitors.checked, + enqueued: timers.enqueued + monitors.triggered, skipped: timers.skipped + monitors.skipped }; + } + return { triggerIssueMonitor, tickTimers, + tickAgentTimers, + tickIssueMonitors, + nextIssueMonitorAt, recoverActiveSessionGoals, recoverPendingSessionGoalActions, }; diff --git a/server/src/services/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index 035dbf10546..679b7e37475 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -1,8 +1,10 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyExternalConversationSettlement(tx, scope); await notifyNativeReviewResponse(tx, scope); await notifyNativeStatusReassessment(tx, scope); diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index d66d3f0732a..f024052ffa5 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; @@ -999,6 +1000,7 @@ export function issueTreeControlService(db: Db) { const now = new Date(); const releasedCancelHoldIds = activeCancelHolds.map((hold) => hold.id); const updatedIssues = await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const restored: TreeStatusUpdateResult["updatedIssues"] = []; for (const [status, issueIdsForStatus] of issueIdsByStatus) { if (issueIdsForStatus.length === 0) continue; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index dfed5e40c6d..52002f47cd7 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -11937,6 +11937,7 @@ export function issueService(db: Db) { await notifyApprovalWork(tx, owner.companyId); } await notifyDecisionTargetIssueRemoval(tx, id); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); removedIssue = await tx .delete(issues) .where(eq(issues.id, id)) @@ -12062,27 +12063,30 @@ export function issueService(db: Db) { eq(issues.executionRunId, checkoutRunId), ) : isNull(issues.executionRunId); - const updated = await db - .update(issues) - .set({ - assigneeAgentId: agentId, - assigneeUserId: null, - checkoutRunId, - executionRunId: checkoutRunId, - status: "in_progress", - startedAt: now, - updatedAt: now, - }) - .where( - and( - eq(issues.id, id), - inArray(issues.status, expectedStatuses), - or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition), - executionLockCondition, - ), - ) - .returning() - .then((rows) => rows[0] ?? null); + const updated = await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + return tx + .update(issues) + .set({ + assigneeAgentId: agentId, + assigneeUserId: null, + checkoutRunId, + executionRunId: checkoutRunId, + status: "in_progress", + startedAt: now, + updatedAt: now, + }) + .where( + and( + eq(issues.id, id), + inArray(issues.status, expectedStatuses), + or(isNull(issues.assigneeAgentId), sameRunAssigneeCondition), + executionLockCondition, + ), + ) + .returning() + .then((rows) => rows[0] ?? null); + }); if (updated) { const [enriched] = await withIssueLabels(db, [updated]); @@ -12186,22 +12190,26 @@ export function issueService(db: Db) { if (current.status !== "in_progress") { adoptionSet.startedAt = now; } - const adopted = await db - .update(issues) - .set(adoptionSet) - .where( - and( - eq(issues.id, id), - inArray(issues.status, expectedStatuses), - eq(issues.executionRunId, current.executionRunId), - or( - isNull(issues.assigneeAgentId), - eq(issues.assigneeAgentId, agentId), + const staleExecutionRunId = current.executionRunId; + const adopted = await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + return tx + .update(issues) + .set(adoptionSet) + .where( + and( + eq(issues.id, id), + inArray(issues.status, expectedStatuses), + eq(issues.executionRunId, staleExecutionRunId), + or( + isNull(issues.assigneeAgentId), + eq(issues.assigneeAgentId, agentId), + ), ), - ), - ) - .returning() - .then((rows) => rows[0] ?? null); + ) + .returning() + .then((rows) => rows[0] ?? null); + }); if (adopted) { const [enriched] = await withIssueLabels(db, [adopted]); return enriched; @@ -12438,6 +12446,7 @@ export function issueService(db: Db) { existing.status === "in_progress" ? "todo" : existing.status; await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const updated = await tx .update(issues) .set({ @@ -12496,6 +12505,7 @@ export function issueService(db: Db) { if (existing.checkoutRunId || existing.executionRunId) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id }); } + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); const updated = await tx .update(issues) .set(patch) diff --git a/server/src/services/native-run-projection-notifications.ts b/server/src/services/native-run-projection-notifications.ts index 7e23de97a1f..d745ec73cb6 100644 --- a/server/src/services/native-run-projection-notifications.ts +++ b/server/src/services/native-run-projection-notifications.ts @@ -2,7 +2,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type NativeRunProjectionScope = { companyId: string; issueId?: string; runId?: string }; -export function notifyNativeRunProjection(tx: object, scope?: NativeRunProjectionScope) { +export async function notifyNativeRunProjection(tx: object, scope?: NativeRunProjectionScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeRunProjection, scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId }) : "*"); } diff --git a/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts b/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts index 1cedcb28be8..ebd1eee0a14 100644 --- a/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts +++ b/server/src/services/native-runtime/native-preserved-monitor-wait.test.ts @@ -1,26 +1,31 @@ import { randomUUID } from "node:crypto"; -import { and, eq } from "drizzle-orm"; -import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, agentWakeupRequests, companies, completionContracts, createDb, heartbeatRuns, +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, agentWakeupRequests, companies, completionContracts, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, subscribeDatabaseWork } from "@paperclipai/db"; import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-run-finalizer.js"; import { nativeSha256 } from "./canonical.js"; import * as monitorDelivery from "./native-monitor-delivery.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { heartbeatService } from "../heartbeat.js"; import { hasPendingRunCleanup } from "../execution-blocker.js"; -const support = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PRESERVED_MONITOR_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)("released native monitor results", () => { let temporary: Awaited>; let db: ReturnType; beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("native-preserved-monitor-"); - db = createDb(temporary.connectionString); - }); + if (external) await applyPendingMigrations(external); + else temporary = await startEmbeddedPostgresTestDatabase("native-preserved-monitor-"); + db = createDb(external ?? temporary.connectionString); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + beforeEach(async () => { await db.execute(sql`truncate table companies restart identity cascade`); }); async function fixture() { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); @@ -160,18 +165,32 @@ describe.skipIf(!support.supported)("released native monitor results", () => { expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].finishedAt).toEqual(f.finishedAt); }); - it("notifies the existing monitor scheduler once when finalization stops blocking it", async () => { - const f = await fixture(), events: string[] = []; + it("makes the saved deadline eligible after finalization commits without replaying the result", async () => { + const f = await fixture(); + const scheduler = heartbeatService(db, { runtimeEnv: {} }); + const observedDeadlines: Promise[] = []; expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(true); - const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { events.push(event); }); + expect(await scheduler.nextIssueMonitorAt()).toBeNull(); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { + if (event === "settled") observedDeadlines.push(scheduler.nextIssueMonitorAt()); + }); try { await f.finalize(); - expect(events).toEqual(["intent", "settled"]); + const deadline = new Date(f.nextCheckAt).getTime(); + + + expect(await Promise.all(observedDeadlines)).toContain(deadline); + expect(await scheduler.nextIssueMonitorAt()).toBe(deadline); expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(false); - expect((await f.readIssue()).monitorNextCheckAt?.toISOString()).toBe(f.nextCheckAt); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); await f.finalize(); - expect(events).toEqual(["intent", "settled"]); - } finally { unsubscribe(); } + expect(await scheduler.nextIssueMonitorAt()).toBe(deadline); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId))).toEqual(decisions); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]); + } finally { + unsubscribe(); + await Promise.all(observedDeadlines); + } }); it.each(["live", "recovered"])("publishes the accepted monitor explanation once (%s)", async mode => { @@ -207,8 +226,8 @@ describe.skipIf(!support.supported)("released native monitor results", () => { const [owner] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }).returning(); const nextCheckAt = new Date(Date.now() + 7_200_000).toISOString(); const original = monitorDelivery.readNativePreservedMonitorWait; - const events: string[] = []; - const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.issueMonitors, event => { events.push(event); }); + const scheduler = heartbeatService(db, { runtimeEnv: {} }); + expect(await scheduler.nextIssueMonitorAt()).toBeNull(); const race = vi.spyOn(monitorDelivery, "readNativePreservedMonitorWait").mockImplementationOnce(async (...args) => { const proof = await original(...args); expect(proof).not.toBeNull(); @@ -220,10 +239,10 @@ describe.skipIf(!support.supported)("released native monitor results", () => { } }).where(eq(issues.id, f.issueId)); return proof; }); - try { await f.finalize(); } finally { race.mockRestore(); unsubscribe(); } + try { await f.finalize(); } finally { race.mockRestore(); } const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); expect(coordinator.phase).toBe(state === "rescheduled" ? "committed" : "retryable_failure"); - expect(events).toEqual(state === "rescheduled" ? ["intent", "settled"] : []); + expect(await scheduler.nextIssueMonitorAt()).toBe(state === "rescheduled" ? new Date(nextCheckAt).getTime() : null); const task = await f.readIssue(); expect(task.executionRunId).toBe(state === "owner" ? owner!.id : null); expect(task.monitorNextCheckAt?.toISOString() ?? null).toBe(state === "cleared" ? null : state === "rescheduled" ? nextCheckAt : f.nextCheckAt); diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.ts b/server/src/services/native-runtime/native-workspace-reconstruction.ts index b8ed3709fc3..d65ba978bb7 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { and, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm"; import { agentWakeupRequests, assets, environmentLeases, heartbeatRuns, issueAttachments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; @@ -184,6 +185,7 @@ export async function reconstructNativeWorkspace(input: { evidence: sql`${issueRecoveryActions.evidence} || ${JSON.stringify({ workspaceReconstructionRequest: requestSha256 })}::jsonb`, wakePolicy: null, monitorPolicy: null, }).where(eq(issueRecoveryActions.id, source.action.id)); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null, blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id)); const publication = await persistActivity(db, { companyId: input.companyId, actorType: "user", actorId: input.actorId, diff --git a/server/src/services/scheduled-work.ts b/server/src/services/scheduled-work.ts index e46bcca4f66..b079aa5c647 100644 --- a/server/src/services/scheduled-work.ts +++ b/server/src/services/scheduled-work.ts @@ -1,49 +1,40 @@ +import { readTaskDrain } from "./task-admission.js"; 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"; -import { DELIVERY_QUEUES, type DeliveryQueue } from "./delivery-work-notifications.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; export function registerScheduledWork( coordinator: Pick, "register">, - db: Pick, options: { retryMs: number; - tickTimers: () => Promise; + tickAgentTimers: () => Promise; + tickIssueMonitors: (signal: AbortSignal) => Promise; + nextIssueMonitorAt: () => Promise; tickRoutines: (signal: AbortSignal) => Promise; nextRoutineRunAt: () => Promise; canRun: () => boolean; }, ) { - function register(queue: DeliveryQueue, work: () => Promise, query: (now: Date) => SQL) { - let next: number | null = null; - return coordinator.register(queue, { - retryMs: options.retryMs, - run: async () => { - const cutoff = new Date(); - await work(); - const [row] = await db.execute<{ next_at: Date | string | null }>(query(cutoff)); - if (!row || row.next_at === undefined) throw new Error("Scheduled deadline query returned no result"); - next = row.next_at === null ? null - : row.next_at instanceof Date ? row.next_at.getTime() : Date.parse(row.next_at); - if (next !== null && !Number.isFinite(next)) throw new Error("Scheduled deadline is invalid"); - }, - hasPending: async () => false, - nextRunAt: () => next, - }); - } - 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 - WHERE c.status = 'active' AND i.status IN ('in_progress', 'in_review') - AND i.assignee_agent_id IS NOT NULL AND i.assignee_user_id IS NULL - AND i.monitor_next_check_at > ${now.toISOString()}::timestamptz`); - return { routines, timers }; + const monitors = registerDeadlineWork(coordinator, DELIVERY_QUEUES.issueMonitors, { + run: options.tickIssueMonitors, + next: options.nextIssueMonitorAt, + canRun: () => options.canRun() && readTaskDrain(new Date()) === null, + retryMs: options.retryMs, + notificationCoalesceMs: options.retryMs, + }); + + + const timers = coordinator.register(DELIVERY_QUEUES.agentTimers, { + retryMs: options.retryMs, + run: options.tickAgentTimers, + hasPending: async () => false, + nextRunAt: () => null, + }); + return { routines, monitors, timers }; } diff --git a/server/src/services/slack-conversation-lifecycle.ts b/server/src/services/slack-conversation-lifecycle.ts index 537530a0de5..05369b13526 100644 --- a/server/src/services/slack-conversation-lifecycle.ts +++ b/server/src/services/slack-conversation-lifecycle.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { and, asc, desc, eq, gt, inArray, isNull, sql } from "drizzle-orm"; import { chatConversations, chatEndpoints, heartbeatRuns, issueComments, issues, type Db } from "@paperclipai/db"; @@ -130,6 +131,7 @@ export async function settleExternalConversation(db: Db, companyId: string, issu .returning({ id: chatConversations.id }); if (!changed.length) return false; await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await tx.update(issues).set({ status: "in_review", updatedAt: new Date() }) .where(and(eq(issues.id, issueId), eq(issues.companyId, companyId))); activity = (await persistActivity(tx as unknown as Db, {