From 0f6f0a3fee01580459625ce3ad7c2e000fc75dc4 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 08:40:56 -0500 Subject: [PATCH] fix(server): schedule issue monitors without heartbeat polling (#17863) https://entire.com/gh/paperclipai/-/trails/56/changes/paperclip/56 ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Issue monitors save a future check on an existing issue. > - The heartbeat timer still scans monitors on every tick, even when none are due. > - The saved deadline and claim already provide durable scheduling state. > - This pull request gives monitors their own deadline worker and connects state changes to it. > - Idle monitor work stops querying. Existing admission and recovery rules remain in force. ## Linked Issues or Issue Description Refs #17840, #16325. Related: #17171 addresses rearming a monitor after its admitted run fails. This PR changes when monitor inspection runs and follows the merged routine prerequisite. **What existing behavior does this improve?** Issue-monitor scheduling and idle database use. **Current behavior** Heartbeat ticks call the combined agent and monitor sweep. The existing next-deadline query only retains future monitors. Polling covers overdue leftovers, stale claims, and changes that make a saved monitor eligible. **Proposed behavior** Use the shared deadline worker. Retain overdue work with bounded retries. Wake after committed changes to monitor state, issue eligibility, company state, or native ownership. Keep legacy agent ticks separate for the next slice. ## What Changed - Add monitor-specific sweep and deadline methods to the existing scheduling owner. Keep the manual combined method. - Include overdue work and the strict five-minute claim fence. Exclude active native owners until their state changes. - Register monitors with the shared deadline worker. Heartbeat ticks wake only the legacy agent timer worker. - Connect direct issue-state writers and existing execution-control notifications to monitor work. Checkout uses a short transaction around its existing CAS. Provider work stays outside it. - Add real database lifecycle tests and retain routine, startup, admission, execution-control, and native-projection coverage. - Record the obligation, producer inventory, recovery behavior, and limits in `doc/plans/2026-10-27-issue-monitor-sleep-audit.md`. ## Verification - **1,233 passing cases in 35 complete files**, no skips, on accepted source `c24aadfc91e9394d73c799f9e05fc67202160557`. Coverage includes real monitor scheduling/admission, native settlement, committed producers, retained routines and all shared deadline consumer integration suites. - Database build, production server typechecking/build, module boundaries, target migration order and Runner profile identity pass. Twenty-four fresh PostgreSQL 17 schemas plus per-file fixtures applied all 412 journal entries through index 413; every selected source hash and temporary fixture import was restored exactly. - Normal parent reconciliation preserves master's admitted-wake result and scoped issue deletion while keeping committed monitor notifications. All other paths match the automatic merge; packages remain byte-exact to the qualified parent. The final routine-parent publication changes only its audit. - Historical failures and corrections remain in [the monitor audit](doc/plans/2026-10-27-issue-monitor-sleep-audit.md). Old delegated-Slack rate and Runner drain CI failures are retained; the current inherited clock fixture and package source are byte-identical to merged #17835. Fresh final-head CI remains required. - No full local repository-suite, live provider, deployed suspend/wake, external-alarm, production-load or fleet-savings claim. Final-head CI/security, exact-head Greptile 5/5 and actionable Ponytail findings remain separate merge gates. ## Risks - A missing producer notification can strand a monitor after polling is removed. The audit covers ordinary writes, direct status/assignment writers, recovery, and native ownership changes. - The writer model is process-local. Independent SQL writers and other replicas need a supported notification or reconciliation contract. - Overdue work uses the configured heartbeat interval as a retry floor for the monitor worker. New notifications during that backoff wait for its next pass. Normal state-change bursts coalesce for one second. - Notification fanout can inspect monitors during other active work. An unchanged idle subsystem has no recurring monitor scan. - The legacy agent timer still polls. Cloud sleep proofs and external alarm coverage are unchanged. ## Model Used OpenAI Codex (GPT-6-based coding agent), using repository inspection, ordinary Git merges, composition proofs, PostgreSQL fixtures and complete local integration suites. No subagents were used in this reconciliation. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: Codex --- doc/idle-issue-monitors.md | 38 ++ .../2026-10-27-issue-monitor-sleep-audit.md | 190 ++++++++++ .../__tests__/issue-monitor-scheduler.test.ts | 4 +- .../__tests__/routine-schedule-work.test.ts | 17 +- .../__tests__/scheduled-work-postgres.test.ts | 7 +- server/src/__tests__/scheduled-work.test.ts | 102 ++++-- .../server-startup-feedback-export.test.ts | 8 +- server/src/index.ts | 11 +- .../active-run-watchdog/adapters/postgres.ts | 2 + server/src/services/access.ts | 2 + server/src/services/agent-conversations.ts | 3 + server/src/services/agent-deletion.ts | 2 + server/src/services/budgets.ts | 2 + server/src/services/companies.ts | 2 + server/src/services/company-deletion.ts | 1 + server/src/services/deadline-work.ts | 8 +- .../services/delivery-work-notifications.ts | 1 + .../execution-control-notifications.ts | 3 +- .../services/execution-recovery-resolution.ts | 1 + server/src/services/heartbeat.ts | 6 + server/src/services/heartbeat/queue.ts | 2 + .../src/services/heartbeat/scheduling.test.ts | 342 +++++++++++++++++- server/src/services/heartbeat/scheduling.ts | 67 +++- .../issue-review-state-notifications.ts | 2 + server/src/services/issue-tree-control.ts | 2 + server/src/services/issues.ts | 82 +++-- .../native-run-projection-notifications.ts | 3 +- .../native-preserved-monitor-wait.test.ts | 55 ++- .../native-workspace-reconstruction.ts | 2 + server/src/services/scheduled-work.ts | 51 ++- .../services/slack-conversation-lifecycle.ts | 2 + 31 files changed, 865 insertions(+), 155 deletions(-) create mode 100644 doc/idle-issue-monitors.md create mode 100644 doc/plans/2026-10-27-issue-monitor-sleep-audit.md 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, {