From ba07f85cadede4673a06c96becc4c4eed2193b93 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 08:58:27 -0500 Subject: [PATCH] perf: schedule agent timers from durable deadlines (#17876) https://entire.com/gh/paperclipai/-/trails/58/changes/paperclip/58 ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agent timers must start due work and preserve queue admission rules. > - The server woke their scan on every heartbeat interval, even with no eligible timer. > - Idle scans keep the database and instance active. > - This pull request derives the next deadline from saved policy and work records. > - Supported producer transactions notify the shared coordinator when eligibility changes. > - An idle timer worker stops scanning and resumes when real work or a deadline requires it. ## Linked Issues or Issue Description Refs #17863. This is slice 63 of the recurring-work migration and follows the merged issue-monitor deadline PR. Related: #16632 changes timer issue scope; #16881 changes live-run eligibility. This PR preserves the current admission policy. Those policy changes are separate work. A future time-sensitive predicate must expose its next transition deadline. **What existing behavior does this improve?** Agent timer heartbeat scheduling in the server. **Current behavior** A recurring heartbeat wakes the timer scan even when companies are paused, timers are disabled, or no actionable task exists. **Proposed behavior** Register the earliest eligible interval or positive daily-cap reset deadline. Re-read on supported producer commits and startup. Leave no recurring database scan when no timer obligation remains. ## What Changed - Keep policy and deadline calculation in the heartbeat queue. Scheduling retains company, reporting-chain, worktree, and CAS gates. - Use the existing deadline coordinator. Remove only the timer wake from the server heartbeat loop. - Notify before existing transaction writes that change eligibility or deadlines. Cover configuration, lifecycle, company budgets, task ownership, recovery, conversations, accounting corrections, and completion. - Use the lifecycle status-change hook for saved hire rejection. Cover checkout's second adoption transaction after an initial failed CAS. - Preserve pending work through idle and maintenance drains. Check cancellation between due agents. Reconstruct on restart. - Document the supported writer model and producer inventory. Add no schema, table, or provider branch. ## Verification - **1,710 passing cases across 49 complete files**, no skips, on accepted source `32db86c1974dcbf794f01a8c5e42024e2ad071fc`. Coverage includes real timer policy/CAS/producers, monitor and shared deadline consumers, native reconstruction, Slack lifecycle, skills/settings/configuration, queue isolation and agent lifecycle. - Database build, production server typechecking/build, module boundaries, target migration order and Runner profile identity pass. Twenty-nine fresh PostgreSQL 17 schemas plus per-file fixtures applied all 412 journal entries through index 413; all selected source hashes and temporary fixture imports were restored exactly. - Four new deadline composition cases verify future/overdue timing, minimum scan spacing and disarming. A controlled mutation dropping only the inherited continuation option fails; the composed source passes with identical assertions and timers. - Normal parent reconciliation preserves task-drain expiry and both deadline contracts. The final monitor-parent publication changes only its audit, preserving every qualified runtime and selected test source. - Historical failures and review corrections remain in [the timer audit](doc/idle-agent-timers.md). No full local repository-suite, live provider, measured suspend/wake, external-alarm or fleet-savings claim. Final-head CI/security, exact-head Greptile 5/5 and actionable Ponytail findings remain separate merge gates. ## Risks - Process-local notifications cover supported service/API writers in one server process. Direct SQL and independent replicas need an explicit ingress contract. Startup reconstruction is not live cross-process notification. - Notifications can batch for up to the configured heartbeat interval. Known deadlines retain their scheduler path; failed reconciliation retains bounded retry. - Enabled timers without an actionable-work requirement remain real recurring obligations. Zero caps require policy changes; positive reached caps use UTC midnight. - The timer fixture substitutes adapter enqueue. Existing admission tests also pass, but this is not deployed provider or sleep qualification. - Cloud does not gain legacy agent timer alarms in this PR. Existing sleep-safety blockers remain. No database/host suspension or fleet savings claim is made. - Parent-stack CI and review must pass before merge. This PR does not authorize merging its dependencies. ## Model Used OpenAI Codex (GPT-6-based coding agent), using repository inspection, normal Git merges, composition proofs, a controlled red/green regression and complete PostgreSQL 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-agent-timers.md | 138 ++++ ...-instance-sleep-and-scheduler-inventory.md | 16 +- .../__tests__/agent-timer-deadlines.test.ts | 762 ++++++++++++++++++ server/src/__tests__/deadline-work.test.ts | 41 + ...heartbeat-stale-queue-invalidation.test.ts | 23 +- .../__tests__/scheduled-work-postgres.test.ts | 2 +- server/src/__tests__/scheduled-work.test.ts | 2 +- .../server-startup-feedback-export.test.ts | 7 +- .../slack-conversation-lifecycle.test.ts | 29 +- server/src/index.ts | 11 +- .../active-run-watchdog/adapters/postgres.ts | 1 + .../agent-lifecycle/adapters/records.ts | 4 + server/src/services/access.ts | 1 + server/src/services/agent-configuration.ts | 4 + server/src/services/agent-conversations.ts | 2 + server/src/services/agent-deletion.ts | 1 + server/src/services/agent-lifecycle.ts | 1 + server/src/services/billing-reconciliation.ts | 2 + server/src/services/budgets.ts | 2 + server/src/services/companies.ts | 2 + server/src/services/company-skills.ts | 2 + server/src/services/deadline-work.ts | 8 +- .../services/execution-recovery-resolution.ts | 1 + server/src/services/heartbeat.ts | 4 + server/src/services/heartbeat/agent-status.ts | 2 + server/src/services/heartbeat/queue.ts | 60 +- .../src/services/heartbeat/scheduling.test.ts | 5 +- server/src/services/heartbeat/scheduling.ts | 146 ++-- server/src/services/instance-settings.ts | 1 + server/src/services/issue-tree-control.ts | 1 + server/src/services/issues.ts | 10 + .../native-workspace-reconstruction.test.ts | 36 +- .../native-workspace-reconstruction.ts | 1 + server/src/services/scheduled-work.ts | 15 +- .../services/slack-conversation-lifecycle.ts | 1 + .../src/services/slack-conversation-state.ts | 2 + 36 files changed, 1206 insertions(+), 140 deletions(-) create mode 100644 doc/idle-agent-timers.md create mode 100644 server/src/__tests__/agent-timer-deadlines.test.ts create mode 100644 server/src/__tests__/deadline-work.test.ts diff --git a/doc/idle-agent-timers.md b/doc/idle-agent-timers.md new file mode 100644 index 00000000000..4034d8636e9 --- /dev/null +++ b/doc/idle-agent-timers.md @@ -0,0 +1,138 @@ +# Agent timer deadlines + +The agent timer worker reconstructs its next deadline from existing agent, +company, issue, run, and cost records. It no longer receives a wake from the +server's recurring heartbeat interval. The queue service owns timer policy and +admission; scheduling selects eligible agents and registers the earliest result +with the shared delivery coordinator. No table or independent scheduler is added. + +## Deadline and quiet conditions + +For an enabled timer with a positive interval, the ordinary deadline is +`(lastHeartbeatAt ?? createdAt) + intervalSec`. Reading it does not advance it. +A due timer retains the existing atomic timestamp claim and queue admission +checks. Successful finalization updates the last heartbeat and moves the next +deadline. An attempted admission still follows the existing claim semantics; +this change does not promise replay of every elapsed interval. + +Only active companies and invokable agents with valid reporting chains qualify. +Worktree execution suppression and the worktree issue creation cutoff still +apply. A disabled timer or zero interval has no deadline. When +`skipTimerWhenNoActionableWork` is enabled, the queue's existing actionable-work +predicate must match an assigned, visible, non-conversation task in Todo or +In Progress without a user assignee. Completed history does not qualify. + +A reached positive daily run or cost cap defers work until the next UTC day, +or the ordinary interval deadline if later. Either zero cap leaves no timed +obligation: only a policy change can make that timer runnable. An accounting +correction or released run claim can lower today's usage and bring a saved +midnight deadline forward. These producers notify the worker. + +Enabled timers that do not require actionable work remain real periodic +obligations. This migration does not silently disable those timers. With no +eligible timer, unresolved transaction, or pending reconciliation, the worker +has no recurring SQL or network scan. + +## Producer and transaction ownership + +Every notification is registered before the relevant write in its existing +transaction. Nested transactions retain outer commit/rollback ownership through +`packages/db/src/work-signals.ts`. Durable records are the source of truth; +notifications request another read and are not an in-memory queue length. + +| Changed facts | Owning producers | +| --- | --- | +| Timer configuration or reporting chain | Agent configuration service, including `updateAgentConfigurationInTransaction`; agent deletion's report reassignment | +| Agent eligibility | Lifecycle status transition hook, including rejection through a saved hire approval | +| Company eligibility | Company update/archive; company budget pause/resume | +| Last heartbeat and daily run claim | Queue timer claim/skip paths, suppression claim release, heartbeat finalization, active-run watchdog finalization | +| Actionable task creation/status/assignment/visibility | Issue create/update/checkout/release/admin release/delete, including checkout's second adoption transaction | +| Restored or reassigned tasks | Subtree cancellation restoration, member archival reassignment, native workspace reconstruction | +| Direct task disarm paths | Queue's invalid workspace block, unrecoverable execution disposition, expired skill harness hiding | +| Conversation/worktree eligibility | Agent conversation prepare/settle; external conversation resume/settle | +| Daily cost corrections | Accounting correction transaction | +| Worktree activation | Experimental settings owner; scheduling requests a fresh suppression read | + +Agent creation/approval remains governed by lifecycle preparation and its eventual +status transition. Company deletion requires completed agent termination, which +already removes timer eligibility. Ordinary run insertion, execution-lock changes, +monitor metadata, nonnegative cost accumulation, and terminal run retention do +not create earlier timer eligibility by themselves. In particular, a conversation +cannot satisfy the actionable-task predicate, but its status can change the +separate worktree cutoff gate, so those status writers are still covered. + +Supported live writers are the serving process's API and service boundaries. +Independent SQL writers and additional server replicas do not share process-local +notifications. Startup reconstructs their saved state, but there is no live +cross-process delivery promise or unconditional recovery polling fallback. +Such writers require an explicit ingress or durable notification contract. + +## Lifecycle and failure behavior + +Notifications coalesce for up to the configured heartbeat interval, normally +30 seconds; already-known deadlines can run sooner. Failed reconciliation keeps +bounded retries through the shared coordinator. Intent uncertainty is resolved +by the shared transaction protocol, never by assuming an empty query means a +write cannot commit later. + +Idle and maintenance drains pause timer work. The worker retains its request +without passive database polling, then retries in memory until admission resumes. +The timer scan checks cancellation/admission between agents. A skipped candidate +keeps its unclaimed deadline; shutdown stops local scheduling, and restart +reconstructs deadlines from the database. This uses the shared `registerDeadlineWork` +generation protocol rather than a timer-specific recovery loop. + +## Verification and limits + +`agent-timer-deadlines.test.ts` exercises real PostgreSQL producers, the queue's +deadline policy and CAS, and the delivery coordinator. It measures no SQL after +settlement for empty, disabled, completed-history, capped, and disarmed states; +first-work reactivation; outer commit and rollback; restart; failed reads; +concurrent claims; both drain modes; and interruption between due agents. +Native reconstruction and external conversation suites separately verify outer +transaction visibility and timer notification at their realistic boundaries. +The timer scheduling fixture substitutes adapter enqueue, so these tests do not +prove provider execution or replace existing queue/admission regressions. + +Worker quiet is not host sleep qualification. The Cloud alarm protocol still +covers only its explicitly supported durable schedules. Legacy agent timers +remain blocked by the existing sleep-safety proof; this change neither exports +them as Cloud alarms nor relaxes that proof. Deployed database suspension, +host suspension, and wake correctness require separate qualification. + +## October 29 master reconciliation and current acceptance + +The existing agent-timer slice was reconciled with the qualified issue-monitor +parent using normal merges. Four conflicts needed explicit composition. Startup +keeps both task-drain expiry registration and the timer worker's fresh drain read; +scheduled work keeps the new timer deadline owner; a comment-only conflict has +no runtime effect. The shared deadline helper retains the parent's explicit +continuation interval alongside the new per-worker minimum scan spacing. All +other paths match the automatic merge, and packages remain byte-identical to the +qualified profile-80 parent tree. + +Source `32db86c1974dcbf794f01a8c5e42024e2ad071fc` passed **1,710 cases in 49 +complete files**, without skips, in 653.08 seconds. This includes the 35 previously +qualified monitor/deadline-consumer files plus real timer policy, CAS and producer +coverage, stale-queue invalidation, native workspace reconstruction, Slack +conversation lifecycle, skill/settings/configuration producers, claim isolation, +five agent-lifecycle module suites, and the new deadline composition regression. +Database build, production server typechecking/build, module boundaries, migration +order and Runner profile identity pass. Twenty-nine 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. + +Four new deadline cases check future and overdue work with and without an explicit +minimum spacing, no dispatch before the selected time, and complete disarming +after settlement. A controlled mutation discarded only the inherited continuation +option while retaining timer spacing: the regression failed. Restoring the +composed source made all four cases pass with the same assertions and timers. +This guards against fixing the timer floor by restoring a 30-second delay for +other workers that deliberately use shorter bounded continuations. + +The final monitor-parent publication changed only its audit. All production trees +and 49 selected test hashes remain identical to this accepted source. Final +publication must also adopt the actual merged prerequisite and receive fresh +exact-head CI, security and reviews. No full local repository-suite, live provider, +external alarm delivery, measured instance/database suspension or fleet-cost claim +is made. Existing Cloud alarm coverage and timer sleep-safety gates remain intact. diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index 38dec000a0e..323e17c82ea 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -552,12 +552,16 @@ Longer polling intervals alone do not complete an item. [plugin-job-scheduler.ts](../../server/src/services/plugin-job-scheduler.ts), [plugin-job-coordinator.ts](../../server/src/services/plugin-job-coordinator.ts). -- [ ] **JOB-08 — Agent timer heartbeats and issue monitors.** Current: **30s** - agent/monitor scans; agent-level disabled heartbeats still require discovery. - Change: register each relevant deadline and aggregate locally. **Complete - when:** no timer work means no queries, and due work preserves pause, budget, - authorization, and atomic wake admission. Source: - [heartbeat.ts](../../server/src/services/heartbeat.ts), `tickTimers`. +- [ ] **JOB-08 — Agent timer heartbeats and issue monitors.** The stacked + implementation separates monitor deadlines (PR #17863) and agent timer + deadlines. Both reconstruct existing durable records and use transaction-aware + notifications instead of an unconditional heartbeat scan. Review/CI and deployed + sleep qualification remain outstanding; legacy agent timers still lack Cloud + alarm coverage and retain their sleep-safety blocker. **Complete when:** no + eligible work means no recurring queries, supported producers reactivate it, + and due work preserves pause, budget, authorization, and atomic wake admission. + Contracts: [issue monitors](../idle-issue-monitors.md), + [agent timers](../idle-agent-timers.md). - [ ] **JOB-09 — Routine cron triggers.** Current: **30s** due-trigger query. Change: use existing `nextRunAt` in the internal scheduler. **Complete when:** diff --git a/server/src/__tests__/agent-timer-deadlines.test.ts b/server/src/__tests__/agent-timer-deadlines.test.ts new file mode 100644 index 00000000000..4c4c0d324a9 --- /dev/null +++ b/server/src/__tests__/agent-timer-deadlines.test.ts @@ -0,0 +1,762 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, activityLog, approvals, authUsers, applyPendingMigrations, companies, companyMemberships, companySkills, companySkillVersions, companySkillTestRuns, issueRecoveryActions, chatEndpoints, chatConversations, toolApplications, toolConnections, costEvents, createDb, heartbeatRuns, issues, issueComments, reconcileDatabaseWork, subscribeDatabaseWork, type Db } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js"; +import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; +import { createHeartbeatRunPreparation } from "../services/heartbeat/run-preparation.js"; +import { createHeartbeatScheduling } from "../services/heartbeat/scheduling.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { registerScheduledWork } from "../services/scheduled-work.js"; +import { updateAgentConfigurationInTransaction } from "../services/agent-configuration-transaction.js"; +import { agentService } from "../services/agents.js"; +import { companyService } from "../services/companies.js"; +import { companySkillService } from "../services/company-skills.js"; +import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js"; +import { createPostgresWatchdogAdapter } from "../modules/active-run-watchdog/adapters/postgres.js"; +import { prepareConversationTurn, settleConversationTurn } from "../services/agent-conversations.js"; +import { approvalService } from "../services/approvals.js"; +import { budgetService } from "../services/budgets.js"; +import { billingReconciliationService } from "../services/billing-reconciliation.js"; +import { accessService } from "../services/access.js"; +import { issueTreeControlService } from "../services/issue-tree-control.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { createAgentLifecycle, startAgentLifecycle } from "../services/agent-lifecycle.js"; +import { createHeartbeatAgentStatus } from "../services/heartbeat/agent-status.js"; +import { issueService } from "../services/issues.js"; +import { resumeExternalConversation } from "../services/slack-conversation-state.js"; +import { idleWorkSnapshot, readTaskDrain, startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; + +vi.mock("../services/live-events.js", () => ({ publishLiveEvent: vi.fn() })); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_TIMER_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const now = new Date("2026-10-27T12:00:00Z"); + +describe.skipIf(!support.supported)("agent timer owned deadlines", () => { + let db: Db; + let temporary: Awaited> | undefined; + let coordinator: ReturnType | undefined; + const errors: unknown[] = []; + beforeAll(async () => { + if (external) await applyPendingMigrations(external); + else temporary = await startEmbeddedPostgresTestDatabase("agent-timer-deadlines-"); + db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + + + await reconcileDatabaseWork(db, DELIVERY_QUEUES.agentTimers); + await db.execute(sql`truncate companies cascade`); + 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([]); + }); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); + + async function fixture(heartbeat: Record = { enabled: true, intervalSec: 600 }) { + const [company] = await db.insert(companies).values({ name: "Timer company", issuePrefix: "TMR", + defaultResponsibleUserId: randomUUID(), requireBoardApprovalForNewAgents: false }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Timer agent", status: "idle", + lifecycleState: "ready", adapterType: "process", createdAt: now, runtimeConfig: { heartbeat } }).returning(); + const deps = createHeartbeatQueueTestDependencies(db); + const preparation = createHeartbeatRunPreparation(db); + const queue = createHeartbeatQueue(db, deps); + const enqueueWakeup = vi.fn(async () => ({ id: randomUUID() }) as typeof heartbeatRuns.$inferSelect); + const scheduling = createHeartbeatScheduling(db, { + getRun: deps.getRun, getAgent: deps.getAgent, getWorktreeExecutionCutoff: deps.getWorktreeExecutionCutoff, + getSchedulingSuppression: deps.getSchedulingSuppression, toAgentOrgRow: preparation.toAgentOrgRow, + groupAgentOrgRowsByCompany: preparation.groupAgentOrgRowsByCompany, issuesSvc: issueService(db), + scheduleBoundedRetryForRun: vi.fn(), enqueueWakeup, + parseHeartbeatPolicy: queue.parseHeartbeatPolicy, nextTimerHeartbeatAt: queue.nextTimerHeartbeatAt, + claimDueTimerHeartbeat: queue.claimDueTimerHeartbeat, + }); + return { company, agent, queue, scheduling, enqueueWakeup, deps }; + } + async function start(f: Awaited>, retryMs = 1000) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const workers = registerScheduledWork(coordinator, { retryMs, canRun: () => true, + tickAgentTimers: signal => f.scheduling.tickAgentTimers(new Date(), () => !signal.aborted && readTaskDrain(new Date()) === null), + nextAgentTimerAt: () => f.scheduling.nextAgentTimerAt(), + tickIssueMonitors: async () => {}, nextIssueMonitorAt: async () => null, + tickRoutines: async () => {}, nextRoutineRunAt: async () => null, + }); + await Promise.all([workers.timers.ready, workers.monitors.ready, workers.routines.ready]); + } + async function quiet() { + await vi.waitFor(() => { expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); }, { timeout: 3000 }); + const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); execute.mockRestore(); transaction.mockRestore(); } + } + + it("preserves the instance scan floor for short intervals and repeated policy writes", async () => { + const policy = { enabled: true, intervalSec: 1 }; + const f = await fixture(policy); + const tick = vi.spyOn(f.scheduling, "tickAgentTimers"); + const inspect = vi.spyOn(f.scheduling, "nextAgentTimerAt"); + const beforeStart = Date.now(); + await start(f, 30_000); + expect(tick).toHaveBeenCalledOnce(); + expect(inspect).toHaveBeenCalledOnce(); + const firstScan = coordinator!.nextWakeAt()!; + expect(firstScan).toBeGreaterThanOrEqual(beforeStart + 30_000); + const initialEnqueues = f.enqueueWakeup.mock.calls.length; + for (let update = 0; update < 10; update++) { + await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: policy } }); + await vi.advanceTimersByTimeAsync(1000); + } + expect(tick).toHaveBeenCalledOnce(); + expect(inspect).toHaveBeenCalledOnce(); + expect(f.enqueueWakeup).toHaveBeenCalledTimes(initialEnqueues); + + await vi.advanceTimersByTimeAsync(Math.max(0, firstScan - Date.now())); + await vi.waitFor(() => expect(f.enqueueWakeup).toHaveBeenCalledTimes(initialEnqueues + 1)); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeGreaterThanOrEqual(firstScan + 30_000)); + expect(tick).toHaveBeenCalledTimes(2); + expect(inspect).toHaveBeenCalledTimes(2); + await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { ...policy, enabled: false } } }); + await vi.advanceTimersByTimeAsync(31_000); + await quiet(); + expect(f.enqueueWakeup).toHaveBeenCalledTimes(initialEnqueues + 1); + }); + + it("derives a real interval deadline from the last checked heartbeat", async () => { + const f = await fixture(); + expect(await f.queue.nextTimerHeartbeatAt(f.agent, now)).toBe(now.getTime() + 600_000); + expect(await f.queue.nextTimerHeartbeatAt({ ...f.agent, lastHeartbeatAt: new Date(now.getTime() + 2000) }, now)) + .toBe(now.getTime() + 602_000); + }); + + it.each([{ enabled: false, intervalSec: 600 }, { enabled: true, intervalSec: 0 }, + { enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }])("is quiet for inactive timer policy %j", async policy => { + const f = await fixture(policy); + await db.insert(issues).values({ companyId: f.company.id, title: "Completed history", status: "done", assigneeAgentId: f.agent.id }); + await start(f); await quiet(); + expect(f.enqueueWakeup).not.toHaveBeenCalled(); + const [saved] = await db.select().from(agents).where(eq(agents.id, f.agent.id)); + expect(saved.lastHeartbeatAt).toBeNull(); + }); + + it("uses the next UTC day for a reached daily run cap, and no deadline for a zero cap", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, maxDailyRuns: 1 }); + await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: "succeeded", startedAt: now, finishedAt: now }); + expect(await f.queue.nextTimerHeartbeatAt(f.agent, now)).toBe(Date.parse("2026-10-28T00:00:00Z")); + expect(await f.queue.nextTimerHeartbeatAt({ ...f.agent, runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600, maxDailyRuns: 0 } } }, now)).toBeNull(); + expect(await f.queue.nextTimerHeartbeatAt(f.agent, new Date("2026-10-28T00:00:00Z"))).toBe(now.getTime() + 600_000); + }); + + it("defers a reached daily cost cap until UTC midnight", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, maxDailyCostCents: 10 }); + await db.insert(costEvents).values({ companyId: f.company.id, agentId: f.agent.id, + provider: "test", model: "test", costCents: 10, occurredAt: now }); + expect(await f.queue.nextTimerHeartbeatAt(f.agent, now)).toBe(Date.parse("2026-10-28T00:00:00Z")); + }); + + it("rearms before midnight when an accounting correction lowers today's cost", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, maxDailyCostCents: 10 }); + const [charge] = await db.insert(costEvents).values({ companyId: f.company.id, agentId: f.agent.id, + provider: "test", model: "test", costCents: 10, occurredAt: now }).returning(); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(Date.parse("2026-10-28T00:00:00Z")); + await billingReconciliationService(db).adjust(f.company.id, charge.id, { + idempotencyKey: randomUUID(), expectedCents: 10, correctedCents: 5, + reason: "Correct provider charge", pricing: { source: "operator" }, + }, "timer-test-operator"); + expect((await db.select().from(costEvents).where(eq(costEvents.id, charge.id)))[0].costCents).toBe(5); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("disarms on company archive even when no agent has a live run", async () => { + const f = await fixture(); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + await companyService(db).archive(f.company.id); + expect((await db.select().from(companies).where(eq(companies.id, f.company.id)))[0].status).toBe("archived"); + await quiet(); + }); + + it.each(["pause", "resume"] as const)("updates timer eligibility after company budget %s", async operation => { + const f = await fixture(); + const budgets = budgetService(db); + const policy = { scopeType: "company" as const, scopeId: f.company.id, + amount: 10, hardStopEnabled: true, notifyEnabled: false }; + await db.insert(costEvents).values({ companyId: f.company.id, agentId: f.agent.id, + provider: "test", model: "test", costCents: 15, occurredAt: now }); + if (operation === "resume") await budgets.upsertPolicy(f.company.id, policy, null); + await start(f); + if (operation === "pause") { + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + await budgets.upsertPolicy(f.company.id, policy, null); + expect((await db.select().from(companies).where(eq(companies.id, f.company.id)))[0].status).toBe("paused"); + await quiet(); + } else { + await quiet(); + await budgets.upsertPolicy(f.company.id, { ...policy, amount: 20 }, null); + expect((await db.select().from(companies).where(eq(companies.id, f.company.id)))[0].status).toBe("active"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + } + }); + + it("stays quiet when company deletion follows required agent termination", async () => { + const f = await fixture(); + const lifecycle = startAgentLifecycle(db, { + requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete", + }, () => true); + try { + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + await createAgentLifecycle(db).terminateAgent(f.agent.id); + await lifecycle.sweep(); + await quiet(); + await companyService(db).remove(f.company.id); + expect(await db.select().from(companies).where(eq(companies.id, f.company.id))).toEqual([]); + await quiet(); + } finally { await lifecycle.stop(); } + }); + + it("disarms a descendant timer when its manager hire approval is rejected", async () => { + const f = await fixture(); + const [manager] = await db.insert(agents).values({ companyId: f.company.id, name: "Pending manager", + status: "pending_approval", lifecycleState: "pending_approval", adapterType: "process" }).returning(); + await db.update(agents).set({ reportsTo: manager.id }).where(eq(agents.id, f.agent.id)); + const [approval] = await db.insert(approvals).values({ companyId: f.company.id, type: "hire_agent", + payload: { agentId: manager.id }, status: "pending" }).returning(); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + await approvalService(db).reject(approval.id, "timer-test-operator"); + expect((await db.select().from(agents).where(eq(agents.id, manager.id)))[0].status).toBe("terminated"); + await quiet(); + }); + + it.each(["prepare", "settle"] as const)("updates the worktree timer gate when a conversation turn does %s", async operation => { + const f = await fixture(); + f.deps.getWorktreeExecutionCutoff.mockResolvedValue(now); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Worktree conversation", + status: operation === "prepare" ? "in_review" : "in_progress", assigneeAgentId: f.agent.id, + conversationAgentId: f.agent.id, conversationUserId: "timer-test-operator", conversationSessionGeneration: 0, + conversationState: operation === "prepare" ? "waiting" : "active", createdAt: now }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: operation === "prepare" ? "running" : "succeeded", + contextSnapshot: { issueId: issue.id, conversationSessionGeneration: 0 } }).returning(); + await start(f); + if (operation === "prepare") { + await quiet(); + expect((await prepareConversationTurn(db, run)).conversation).toBe(true); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].status).toBe("in_progress"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + } else { + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + await db.insert(issueComments).values({ companyId: f.company.id, issueId: issue.id, + authorAgentId: f.agent.id, createdByRunId: run.id, body: "Turn completed" }); + expect(await settleConversationTurn(db, run)).toBe(true); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].status).toBe("in_review"); + await quiet(); + } + }); + + it("reschedules after the watchdog folds a completed source run", async () => { + const f = await fixture(); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Finished work", + status: "done", assigneeAgentId: f.agent.id }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: "running", startedAt: now, contextSnapshot: { issueId: issue.id } }).returning(); + await db.insert(activityLog).values({ companyId: f.company.id, actorType: "agent", actorId: f.agent.id, + runId: run.id, action: "issue.updated", entityType: "issue", entityId: issue.id, + details: { status: "done" }, createdAt: now }); + await start(f); + await vi.advanceTimersByTimeAsync(60_000); + const adapter = createPostgresWatchdogAdapter(db); + const snapshot = await adapter.findRunForCompany(f.company.id, run.id); + const sourceIssue = await adapter.findSourceIssue(f.company.id, issue.id); + const evidence = await adapter.findLatestSameRunTerminalEvidence(f.company.id, { + runId: run.id, sourceIssueId: issue.id, sourceIssueStatus: "done", evidenceAfter: now, + }); + expect(snapshot).not.toBeNull(); expect(sourceIssue).not.toBeNull(); expect(evidence).not.toBeNull(); + expect(await adapter.foldSourceResolvedRun(f.company.id, { run: snapshot!, sourceIssue: sourceIssue!, evidence: evidence!, + existingEvaluation: null, silenceStartedAt: now, silenceAgeMs: 60_000, + cleanup: { attempted: false, outcome: "no_process_metadata", adapterType: "process" }, now: new Date(), + })).toMatchObject({ kind: "folded" }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded"); + const [saved] = await db.select().from(agents).where(eq(agents.id, f.agent.id)); + const deadline = saved.lastHeartbeatAt!.getTime() + 600_000; + expect(deadline).toBeGreaterThan(now.getTime() + 600_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(deadline), { timeout: 3000 }); + }); + + it("disarms when an expired skill harness hides the last actionable task", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Expired harness", + status: "in_progress", assigneeAgentId: f.agent.id, harnessKind: "skill_test" }).returning(); + const [skill] = await db.insert(companySkills).values({ companyId: f.company.id, + key: "timer-test", slug: "timer-test", name: "Timer test", markdown: "Test" }).returning(); + const [version] = await db.insert(companySkillVersions).values({ companyId: f.company.id, + companySkillId: skill.id, revisionNumber: 1 }).returning(); + await db.insert(companySkillTestRuns).values({ companyId: f.company.id, skillId: skill.id, + skillVersionId: version.id, agentId: f.agent.id, issueId: issue.id, inputSnapshot: "Test", + harnessIssueExpiresAt: new Date(now.getTime() - 1000) }); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + expect(await companySkillService(db).pruneExpiredTestHarnessIssues(f.company.id, now)).toEqual({ pruned: 1 }); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].hiddenAt).toEqual(now); + await quiet(); + }); + + it("disarms when execution disposition blocks the last actionable task", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Unrecoverable task", + status: "in_progress", assigneeAgentId: f.agent.id }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: "failed", contextSnapshot: { issueId: issue.id } }).returning(); + await db.insert(issueRecoveryActions).values({ companyId: f.company.id, sourceIssueId: issue.id, + kind: "active_run_watchdog", status: "active", ownerType: "board", returnOwnerAgentId: f.agent.id, + cause: "legacy_execution_requires_reconciliation", fingerprint: run.id, evidence: { runId: run.id }, + nextAction: "Verify execution outcome" }); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + await settleUnrecoverableExecutions(db, now); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].status).toBe("blocked"); + await quiet(); + }); + + it("disarms when queue admission blocks a projectless isolated worktree task", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const settings = instanceSettingsService(db); + await settings.updateExperimental({ enableIsolatedWorkspaces: true }); + f.deps.instanceSettings.getExperimental.mockImplementation(() => settings.getExperimental()); + f.deps.getAgent.mockResolvedValue(f.agent); + const userId = f.company.defaultResponsibleUserId!; + await db.insert(authUsers).values({ id: userId, name: "Timer operator", email: `${userId}@example.test`, + createdAt: now, updatedAt: now }); + await db.insert(companyMemberships).values({ companyId: f.company.id, principalType: "user", + principalId: userId, status: "active", membershipRole: "member" }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Projectless isolated task", + status: "todo", assigneeAgentId: f.agent.id, responsibleUserId: userId, + executionWorkspaceSettings: { mode: "isolated_workspace", workspaceStrategy: { type: "git_worktree" } } }).returning(); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + expect(await f.queue.enqueueWakeup(f.agent.id, { source: "assignment", triggerDetail: "system", + reason: "issue_assigned", payload: { issueId: issue.id }, contextSnapshot: { issueId: issue.id, wakeReason: "issue_assigned" }, + })).toBeNull(); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].status).toBe("blocked"); + await quiet(); + }); + + it("reactivates after checkout adopts an execution that finished between its two write attempts", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [oldRun, newRun] = await db.insert(heartbeatRuns).values([ + { companyId: f.company.id, agentId: f.agent.id, status: "running" }, + { companyId: f.company.id, agentId: f.agent.id, status: "queued" }, + ]).returning(); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Checkout race", + status: "backlog", assigneeAgentId: f.agent.id, executionRunId: oldRun.id }).returning(); + await start(f); await quiet(); + const transaction = db.transaction.bind(db); + let firstAttemptSettled = false; + const intercepted = vi.spyOn(db, "transaction").mockImplementation(async (work, config) => { + const result = await transaction(work, config); + if (result === null && !firstAttemptSettled) { + firstAttemptSettled = true; + intercepted.mockRestore(); + + + await quiet(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, oldRun.id)); + } + return result; + }); + try { + expect(await issueService(db).checkout(issue.id, f.agent.id, ["backlog"], newRun.id)) + .toMatchObject({ status: "in_progress", executionRunId: newRun.id }); + expect(firstAttemptSettled).toBe(true); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + } finally { intercepted.mockRestore(); } + }); + + it("keeps a zero cost cap quiet even when today's positive run cap was also reached", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, maxDailyRuns: 1, maxDailyCostCents: 0 }); + await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: "succeeded", startedAt: now, finishedAt: now }); + await start(f); await quiet(); + }); + + it("leaves an empty instance quiet after discovery", async () => { + const f = await fixture(); + await db.execute(sql`truncate companies cascade`); + await start(f); await quiet(); + expect(f.enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("stops admission between due agents without consuming the remaining deadline", async () => { + const f = await fixture(); + await db.insert(agents).values({ companyId: f.company.id, name: "Second timer", status: "idle", + lifecycleState: "ready", adapterType: "process", createdAt: now, + runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } } }); + let canContinue = true; + f.enqueueWakeup.mockImplementation(async () => { + canContinue = false; + return { id: randomUUID() } as typeof heartbeatRuns.$inferSelect; + }); + const due = new Date(now.getTime() + 600_000); + expect(await f.scheduling.tickAgentTimers(due, () => canContinue)).toMatchObject({ checked: 1, enqueued: 1 }); + expect(f.enqueueWakeup).toHaveBeenCalledOnce(); + expect((await db.select().from(agents)).filter(agent => agent.lastHeartbeatAt === null)).toHaveLength(1); + expect(await f.scheduling.nextAgentTimerAt(due)).toBe(due.getTime()); + canContinue = true; + expect(await f.scheduling.tickAgentTimers(due, () => canContinue)).toMatchObject({ enqueued: 1 }); + expect(f.enqueueWakeup).toHaveBeenCalledTimes(2); + expect(await f.scheduling.nextAgentTimerAt(due)).toBe(due.getTime() + 600_000); + }); + + it.each([false, true])("honors the supported caller-owned configuration transaction, rollback=%s", async rollback => { + const f = await fixture({ enabled: false, intervalSec: 600 }); + await start(f); await quiet(); + const mutation = db.transaction(async tx => { + await updateAgentConfigurationInTransaction(tx as unknown as Db, f.agent.id, + { runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } } }); + await vi.advanceTimersByTimeAsync(2000); + expect(f.enqueueWakeup).not.toHaveBeenCalled(); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].runtimeConfig.heartbeat) + .toEqual({ enabled: false, intervalSec: 600 }); + if (rollback) throw new Error("rollback caller-owned configuration"); + }); + if (rollback) { + await expect(mutation).rejects.toThrow("rollback caller-owned configuration"); + await quiet(); + } else { + await mutation; + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + } + }); + + it("reactivates from creation of the first actionable issue", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + await start(f); await quiet(); + await issueService(db).create(f.company.id, { title: "First task", status: "todo", assigneeAgentId: f.agent.id }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it.each([false, true])("keeps agent configuration invisible until its outer transaction ends, rollback=%s", async rollback => { + const f = await fixture({ enabled: false, intervalSec: 600 }); + await start(f); await quiet(); + let release!: () => void, entered!: () => void; + const held = new Promise(r => { release = r; }), written = new Promise(r => { entered = r; }); + const transaction = db.transaction.bind(db); + const heldTransaction = vi.spyOn(db, "transaction").mockImplementationOnce((work, config) => + transaction(async tx => { + const result = await work(tx); + entered(); + await held; + if (rollback) throw new Error("rollback timer config"); + return result; + }, config)); + const mutation = agentService(db).update(f.agent.id, { + runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } }, + }).then(() => null, error => error as Error); + try { + + const earlyError = await Promise.race([written.then(() => null), mutation]); + expect(earlyError).toBeNull(); + await vi.advanceTimersByTimeAsync(2000); + expect(f.enqueueWakeup).not.toHaveBeenCalled(); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].runtimeConfig.heartbeat).toEqual({ enabled: false, intervalSec: 600 }); + } finally { + release(); + const error = await mutation; + heldTransaction.mockRestore(); + if (rollback) expect(error?.message).toBe("rollback timer config"); + else expect(error).toBeNull(); + } + if (rollback) await quiet(); + else await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("claims a due timer only once and publishes its timestamp change at outer commit", async () => { + const f = await fixture(); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event)); + const due = new Date(now.getTime() + 600_000); + try { + await db.transaction(async tx => { + const owner = tx as unknown as Db; + const queue = createHeartbeatQueue(owner, createHeartbeatQueueTestDependencies(owner)); + expect(await queue.claimDueTimerHeartbeat(f.agent, due, 600)).toEqual({ wasFirstHeartbeat: true }); + expect(events).toContain("intent"); + expect(events).not.toContain("settled"); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].lastHeartbeatAt).toBeNull(); + }); + expect(events).toContain("settled"); + expect(await f.queue.claimDueTimerHeartbeat(f.agent, due, 600)).toBeNull(); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].lastHeartbeatAt).toEqual(due); + } finally { unsubscribe(); } + }); + + it("disarms on pause and reactivates only after lifecycle resume completes", async () => { + const f = await fixture(); + const lifecycle = startAgentLifecycle(db, { + requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete", + }, () => true); + try { + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + await createAgentLifecycle(db).pauseAgent(f.agent.id); + await lifecycle.sweep(); + await quiet(); + await createAgentLifecycle(db).resumeAgent(f.agent.id); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("paused"); + await lifecycle.sweep(); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + } finally { await lifecycle.stop(); } + }); + + it("reactivates when the company becomes active", async () => { + const f = await fixture(); + await db.update(companies).set({ status: "paused" }).where(eq(companies.id, f.company.id)); + await start(f); await quiet(); + await companyService(db).update(f.company.id, { status: "active" }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("reschedules after a real heartbeat completion changes the last heartbeat", async () => { + const f = await fixture(); + await start(f); + await vi.advanceTimersByTimeAsync(60_000); + await createHeartbeatAgentStatus(db).finalizeAgentStatus(f.agent.id, "succeeded"); + const [saved] = await db.select().from(agents).where(eq(agents.id, f.agent.id)); + const deadline = saved.lastHeartbeatAt!.getTime() + 600_000; + expect(deadline).toBeGreaterThan(now.getTime() + 600_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(deadline), { timeout: 3000 }); + }); + + it("reactivates when checkout assigns the first actionable issue", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Unclaimed", status: "backlog" }).returning(); + await start(f); await quiet(); + await issueService(db).checkout(issue.id, f.agent.id, ["backlog"], null); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it.each(["release", "adminForceRelease", "remove"] as const)("disarms when %s removes the last actionable assignment", async operation => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Only task", status: "in_progress", assigneeAgentId: f.agent.id }).returning(); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000); + const service = issueService(db); + if (operation === "adminForceRelease") await service.adminForceRelease(issue.id, { clearAssignee: true }); + else await service[operation](issue.id); + await quiet(); + }); + + it("reactivates a descendant after its terminated manager is purged", async () => { + const f = await fixture(); + const [manager] = await db.insert(agents).values({ companyId: f.company.id, name: "Former manager", + status: "terminated", lifecycleState: "terminated", adapterType: "process" }).returning(); + await db.update(agents).set({ reportsTo: manager.id }).where(eq(agents.id, f.agent.id)); + await start(f); await quiet(); + await createAgentLifecycle(db).purgeAgent(manager.id); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("dispatches at the saved deadline and reconstructs the following deadline after restart", async () => { + const f = await fixture(); + await start(f); + await vi.advanceTimersByTimeAsync(599_000); + expect(f.enqueueWakeup).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(f.enqueueWakeup).toHaveBeenCalledOnce()); + expect(f.enqueueWakeup).toHaveBeenCalledWith(f.agent.id, expect.objectContaining({ + source: "timer", reason: "heartbeat_timer", + })); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + const [saved] = await db.select().from(agents).where(eq(agents.id, f.agent.id)); + expect(saved.lastHeartbeatAt!.getTime()).toBeGreaterThanOrEqual(now.getTime() + 600_000); + const next = saved.lastHeartbeatAt!.getTime() + 600_000; + await coordinator!.stop(); coordinator = undefined; + await start(f); + expect(coordinator!.nextWakeAt()).toBe(next); + expect(f.enqueueWakeup).toHaveBeenCalledOnce(); + await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: false } } }); + await quiet(); + }); + + it.each(["idle", "maintenance"] as const)("retains first work committed during %s drain without polling the database", async purpose => { + const f = await fixture({ enabled: false, intervalSec: 600 }); + await start(f); await quiet(); + startTaskDrain(purpose === "idle" ? { purpose } : {}); + await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } } }); + await vi.advanceTimersByTimeAsync(2000); + + + expect(idleWorkSnapshot().active).toBe(1); + const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + expect(f.enqueueWakeup).not.toHaveBeenCalled(); + } finally { select.mockRestore(); execute.mockRestore(); transaction.mockRestore(); } + stopTaskDrain(); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("does not inspect timer rows during startup drain, then reconstructs them on resume", async () => { + const f = await fixture(); + startTaskDrain(); + const select = vi.spyOn(db, "select"); + try { + await start(f); + await vi.advanceTimersByTimeAsync(60_000); + expect(select).not.toHaveBeenCalled(); + } finally { select.mockRestore(); } + stopTaskDrain(); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("retains a failed deadline inspection until a successful bounded retry", async () => { + const f = await fixture(); + const next = vi.spyOn(f.scheduling, "nextAgentTimerAt").mockRejectedValueOnce(new Error("deadline unavailable")); + await start(f); + expect(errors).toHaveLength(1); errors.length = 0; + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + expect(next).toHaveBeenCalledTimes(2); + }); + + it("releases a rolled-back timer claim so another claimant can own it", async () => { + const f = await fixture(); + const due = new Date(now.getTime() + 600_000); + await expect(db.transaction(async tx => { + const owner = tx as unknown as Db; + const queue = createHeartbeatQueue(owner, createHeartbeatQueueTestDependencies(owner)); + expect(await queue.claimDueTimerHeartbeat(f.agent, due, 600)).not.toBeNull(); + throw new Error("rollback claim"); + })).rejects.toThrow("rollback claim"); + const claims = await Promise.all([f.queue.claimDueTimerHeartbeat(f.agent, due, 600), f.queue.claimDueTimerHeartbeat(f.agent, due, 600)]); + expect(claims.filter(Boolean)).toHaveLength(1); + }); + + it.each(["no work", "daily cap", "issue daily cap"] as const)("notifies a timestamp change from manual timer admission skipped for %s", async reason => { + const f = await fixture({ enabled: true, intervalSec: 600, + ...(reason === "no work" ? { skipTimerWhenNoActionableWork: true } : { maxDailyRuns: 0 }) }); + f.deps.getAgent.mockResolvedValue(f.agent); + const [issue] = reason === "issue daily cap" ? await db.insert(issues).values({ + companyId: f.company.id, title: "Capped task", status: "todo", assigneeAgentId: f.agent.id, + }).returning() : []; + if (issue) f.deps.instanceSettings.getExperimental.mockResolvedValue({ enableIsolatedWorkspaces: false } as never); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event)); + try { + expect(await f.queue.enqueueWakeup(f.agent.id, { source: "timer", reason: "heartbeat_timer", + ...(issue ? { contextSnapshot: { issueId: issue.id } } : {}), + })).toBeNull(); + expect(events).toContain("settled"); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].lastHeartbeatAt).not.toBeNull(); + } finally { unsubscribe(); } + }); + + it("moves the deadline back from midnight when suppression releases a counted run claim", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, maxDailyRuns: 1 }); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, + status: "running", startedAt: now }).returning(); + await start(f); + expect(coordinator!.nextWakeAt()).toBe(Date.parse("2026-10-28T00:00:00Z")); + await f.queue.releaseRunClaimedJustBeforeSuppression(run.id); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toMatchObject({ status: "queued", startedAt: null }); + }); + + it("reactivates when a cancelled issue tree is restored", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Restorable", status: "todo", assigneeAgentId: f.agent.id }).returning(); + const tree = issueTreeControlService(db); + const actor = { actorType: "user" as const, actorId: "board", userId: "board" }; + const cancel = await tree.createHold(f.company.id, issue.id, { mode: "cancel", actor }); + await tree.cancelIssueStatusesForHold(f.company.id, issue.id, cancel.hold.id); + await start(f); await quiet(); + const restore = await tree.createHold(f.company.id, issue.id, { mode: "restore", actor }); + const result = await tree.restoreIssueStatusesForHold(f.company.id, issue.id, restore.hold.id, { actor }); + expect(result.updatedIssueIds).toEqual([issue.id]); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("reactivates when member archival reassigns work to the agent", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [member] = await db.insert(companyMemberships).values({ companyId: f.company.id, + principalType: "user", principalId: randomUUID(), status: "active", membershipRole: "operator" }).returning(); + await db.insert(issues).values({ companyId: f.company.id, title: "Reassigned", status: "in_progress", assigneeUserId: member.principalId }); + await start(f); await quiet(); + await accessService(db).archiveMember(f.company.id, member.id, { reassignment: { assigneeAgentId: f.agent.id } }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("reads fresh worktree activation before deciding that no timer is due", async () => { + const f = await fixture(); + const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "timer-fixture" }; + const settings = instanceSettingsService(db, { runtimeEnv, now: () => now }); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const heartbeat = heartbeatService(db, { runtimeEnv }); + expect((await heartbeat.resolveSchedulingSuppression(true)).suppressed).toBe(true); + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await db.insert(issues).values({ companyId: f.company.id, title: "New worktree work", status: "todo", + assigneeAgentId: f.agent.id, createdAt: now }); + expect(await heartbeat.nextAgentTimerAt()).toBe(now.getTime() + 600_000); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + expect(await heartbeat.nextAgentTimerAt()).toBeNull(); + }); + + it("publishes a timer wake when the worktree activation setting changes", async () => { + const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "timer-fixture" }; + const settings = instanceSettingsService(db, { runtimeEnv, now: () => now }); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event)); + try { + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + expect(events).toContain("settled"); + } finally { unsubscribe(); } + }); + + it("reactivates when an incoming message resumes an idle external conversation", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Conversation", status: "in_review", assigneeAgentId: f.agent.id }).returning(); + const [application] = await db.insert(toolApplications).values({ companyId: f.company.id, name: "Slack", type: "chat" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId: f.company.id, applicationId: application.id, + name: "Slack", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning(); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId: f.company.id, connectionId: connection.id, + provider: "slack", publicId: randomUUID(), assignedAgentId: f.agent.id, status: "active" }).returning(); + await db.insert(chatConversations).values({ companyId: f.company.id, endpointId: endpoint.id, issueId: issue.id, + externalConversationId: "channel", externalThreadId: "thread", externalLabel: "#timer-test", state: "waiting" }); + await start(f); await quiet(); + await db.transaction(tx => resumeExternalConversation(tx as unknown as Db, f.company.id, issue.id)); + expect((await db.select().from(issues).where(eq(issues.id, issue.id)))[0].status).toBe("todo"); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("reactivates through the real agent configuration producer", async () => { + const f = await fixture({ enabled: false, intervalSec: 600 }); + await start(f); await quiet(); + await agentService(db).update(f.agent.id, { runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } } }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); + + it("reactivates when the real issue producer creates actionable work", async () => { + const f = await fixture({ enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true }); + const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Upcoming task", status: "backlog", assigneeAgentId: f.agent.id }).returning(); + await start(f); await quiet(); + await issueService(db).update(issue.id, { status: "todo" }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(now.getTime() + 600_000), { timeout: 3000 }); + }); +}); diff --git a/server/src/__tests__/deadline-work.test.ts b/server/src/__tests__/deadline-work.test.ts new file mode 100644 index 00000000000..7b852977345 --- /dev/null +++ b/server/src/__tests__/deadline-work.test.ts @@ -0,0 +1,41 @@ +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { afterEach, expect, it, vi } from "vitest"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerDeadlineWork } from "../services/deadline-work.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; + +let coordinator: ReturnType | undefined; +afterEach(async () => { await coordinator?.stop(); vi.useRealTimers(); }); + +it.each([ + { due: "overdue", spacing: 0, delay: 250 }, + { due: "overdue", spacing: 1_000, delay: 1_000 }, + { due: "future", spacing: 0, delay: 5 }, + { due: "future", spacing: 1_000, delay: 1_000 }, +])("preserves $due continuation with $spacing ms minimum spacing and disarms after settlement", async ({ due, spacing, delay }) => { + vi.useFakeTimers(); + const onError = vi.fn(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError }); + const now = Date.now(); + const run = vi.fn(async () => {}); + const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { + run, + next: async () => run.mock.calls.length === 1 ? now + (due === "future" ? 5 : -1) : null, + canRun: () => true, + retryMs: 30_000, + continuationMs: 250, + minimumSpacingMs: spacing, + }); + await worker.ready; + expect(run).toHaveBeenCalledOnce(); + expect(coordinator.nextWakeAt()).toBe(now + delay); + await vi.advanceTimersByTimeAsync(delay - 1); + expect(run).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(1); + expect(run).toHaveBeenCalledTimes(2); + expect(coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(60_000); + expect(run).toHaveBeenCalledTimes(2); + expect(onError).not.toHaveBeenCalled(); +}); diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index df965f6dff0..65191c24d2f 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -10,6 +10,8 @@ import { companyMemberships, costEvents, createDb, + applyPendingMigrations, + EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, documentRevisions, documents, environmentLeases, @@ -25,6 +27,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { truncateTablesWithDeadlockRetry } from "./helpers/truncate-with-deadlock-retry.js"; import { @@ -59,7 +62,8 @@ vi.mock("../adapters/index.ts", async () => { }; }); -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STALE_QUEUE_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -147,8 +151,9 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { mockAdapterExecute.mock.calls.filter(([context]) => context?.runId === runId).length; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-stale-queue-"); - db = createDb(tempDb.connectionString); + if (external) await applyPendingMigrations(external); + else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-stale-queue-"); + db = createDb(external ?? tempDb!.connectionString); heartbeat = heartbeatService(db, { runtimeEnv: { ...process.env, PAPERCLIP_IN_WORKTREE: "false" }, beforeResolvedInteractionContinuationDispatchCheck: async (input) => { @@ -159,7 +164,7 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { }, }); await ensureIssueRelationsTable(db); - }, 20_000); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterEach(async () => { beforeContinuationDispatchCheck = null; @@ -942,7 +947,7 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { expect(countExecuteCallsForRun(runId)).toBe(0); }); - it("rate-limits skipped generic timer wakes by advancing the timer baseline", async () => { + it("leaves ineligible timers disarmed without writing skipped wakes or moving their baseline", async () => { const { agentId } = await seedCompanyAndAgent({ heartbeatConfig: { enabled: true, @@ -959,7 +964,8 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { const firstTick = await heartbeat.tickTimers(now); const secondTick = await heartbeat.tickTimers(now); - expect(firstTick.skipped).toBe(1); + expect(firstTick.skipped).toBe(0); + expect(await heartbeat.nextAgentTimerAt()).toBeNull(); expect(secondTick.skipped).toBe(0); expect(mockAdapterExecute).not.toHaveBeenCalled(); @@ -972,10 +978,9 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { .from(agents) .where(eq(agents.id, agentId)); - expect(wakeups).toHaveLength(1); - expect(wakeups[0]?.reason).toBe("heartbeat.timer.no_actionable_work"); + expect(wakeups).toEqual([]); expect(agent?.lastHeartbeatAt).toBeInstanceOf(Date); - expect(agent?.lastHeartbeatAt?.getTime()).toBeGreaterThan(now.getTime() - 120_000); + expect(agent?.lastHeartbeatAt?.getTime()).toBe(now.getTime() - 120_000); }); it("atomically claims a due timer interval across overlapping scheduler ticks", async () => { diff --git a/server/src/__tests__/scheduled-work-postgres.test.ts b/server/src/__tests__/scheduled-work-postgres.test.ts index 13ce385ecb2..ac5fda2e5ff 100644 --- a/server/src/__tests__/scheduled-work-postgres.test.ts +++ b/server/src/__tests__/scheduled-work-postgres.test.ts @@ -61,7 +61,7 @@ describe.skipIf(!support.supported)("scheduled deadline database queries", () => registerScheduledWork({ register: (queue, task) => { tasks.set(queue, task); return { ready: Promise.resolve(), wake: () => {} }; - } }, { retryMs: 60_000, tickAgentTimers: async () => {}, tickIssueMonitors: async () => {}, + } }, { retryMs: 60_000, tickAgentTimers: async () => {}, nextAgentTimerAt: async () => null, 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); diff --git a/server/src/__tests__/scheduled-work.test.ts b/server/src/__tests__/scheduled-work.test.ts index 63286eabf4c..561133f7436 100644 --- a/server/src/__tests__/scheduled-work.test.ts +++ b/server/src/__tests__/scheduled-work.test.ts @@ -50,7 +50,7 @@ function fixture(input: { routine?: number | null; monitor?: number | null; tick return deadlines.routine; }); const workers = registerScheduledWork(coordinator, - { retryMs: 60_000, tickAgentTimers, tickIssueMonitors, nextIssueMonitorAt, tickRoutines, nextRoutineRunAt, canRun: () => canRun }); + { retryMs: 60_000, nextAgentTimerAt: async () => null, 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; } }; diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 786ee772bd2..cffbde81177 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -155,6 +155,7 @@ const { sweepPendingCleanupLeases: vi.fn(async () => ({ swept: 0, destroyed: 0, capped: 0 })), sweepExpiredRuntimeStatuses: vi.fn(() => 0), tickAgentTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })), + nextAgentTimerAt: vi.fn(async () => null), tickIssueMonitors: vi.fn(async () => ({ checked: 0, triggered: 0, skipped: 0 })), nextIssueMonitorAt: vi.fn(async () => null), }; @@ -1170,7 +1171,9 @@ describe("startServer feedback export wiring", () => { await settleStartupHousekeeping(); expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled(); - expect(heartbeatServiceMock.tickAgentTimers).not.toHaveBeenCalled(); + + + expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); expect(intervalCallback).not.toBeNull(); @@ -1178,7 +1181,7 @@ describe("startServer feedback export wiring", () => { await Promise.resolve(); await Promise.resolve(); - expect(heartbeatServiceMock.tickAgentTimers).not.toHaveBeenCalled(); + expect(heartbeatServiceMock.tickAgentTimers).toHaveBeenCalledTimes(1); expect(externalObjectsServiceMock.refreshDueObjectsForActiveCompanies).toHaveBeenCalledTimes(1); expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); diff --git a/server/src/__tests__/slack-conversation-lifecycle.test.ts b/server/src/__tests__/slack-conversation-lifecycle.test.ts index 6fc3570fa2c..2a1e7778416 100644 --- a/server/src/__tests__/slack-conversation-lifecycle.test.ts +++ b/server/src/__tests__/slack-conversation-lifecycle.test.ts @@ -1,3 +1,6 @@ +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { createHeartbeatQueue } from "../services/heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js"; import { createHeartbeatScheduling, type HeartbeatSchedulingDependencies } from "../services/heartbeat/scheduling.js"; import { createHeartbeatRunState } from "../services/heartbeat/run-state.js"; import { createHeartbeatRunPreparation } from "../services/heartbeat/run-preparation.js"; @@ -30,7 +33,7 @@ import { activityLog, agents, agentWakeupRequests, approvals, chatConversations, chatEndpoints, chatMessageLinks, chatPublications, companies, createDb, heartbeatRuns, issueApprovals, issueComments, issueRelations, issueThreadInteractions, issues, nativeRunFinalizations, - toolApplications, toolConnections, + toolApplications, toolConnections, subscribeDatabaseWork, type Db, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { pendingExternalConversationSettlement, externalConversationSettlementCandidates, settleSlackConversation } from "../services/slack-conversation-lifecycle.js"; @@ -111,10 +114,32 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu toAgentOrgRow: preparation.toAgentOrgRow, groupAgentOrgRowsByCompany: preparation.groupAgentOrgRowsByCompany, issuesSvc: issueService(db), enqueueWakeup, getSchedulingSuppression: async () => ({ suppressed: false }), getWorktreeExecutionCutoff: async () => null, - scheduleBoundedRetryForRun: vi.fn(), claimDueTimerHeartbeat: vi.fn(), parseHeartbeatPolicy: vi.fn(), + scheduleBoundedRetryForRun: vi.fn(), claimDueTimerHeartbeat: vi.fn(), parseHeartbeatPolicy: vi.fn(), nextTimerHeartbeatAt: vi.fn(), }); } + it("notifies timer eligibility at outer commit when a published conversation becomes idle", async () => { + const f = await fixture(); + const [agent] = await db.update(agents).set({ runtimeConfig: { heartbeat: { + enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true, + } } }).where(eq(agents.id, f.agentId)).returning(); + const queue = createHeartbeatQueue(db, createHeartbeatQueueTestDependencies(db)); + expect(await queue.nextTimerHeartbeatAt(agent)).toBe(agent.createdAt.getTime() + 600_000); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event)); + try { + await db.transaction(async tx => { + expect(await settleSlackConversation(tx as unknown as Db, f.companyId, f.issueId)).toBe(true); + expect(events).toContain("intent"); + expect(events).not.toContain("settled"); + expect((await state(f)).status).toBe("in_progress"); + }); + expect(events).toContain("settled"); + expect((await state(f)).status).toBe("in_review"); + expect(await queue.nextTimerHeartbeatAt(agent)).toBeNull(); + } finally { unsubscribe(); } + }); + it("keeps monitor recovery pending until its wake handoff finishes", async () => { const f = await fixture(); let release!: () => void, entered!: () => void; diff --git a/server/src/index.ts b/server/src/index.ts index 9e49cfb6a17..a6978f577e5 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -47,7 +47,7 @@ import { caseHandoffDeliveryService } from "./services/case-handoff-delivery.js" import { registerScheduledWork } from "./services/scheduled-work.js"; import type { createDeliveryWorkCoordinator } from "./services/delivery-work-coordinator.js"; import { registerTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./services/terminal-workspace-work.js"; -import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork, registerTaskDrainExpiryWork } from "./services/task-admission.js"; +import { isIdleTaskDrainActive, readTaskDrain, beginIdleTrackedWork, trackIdleWork, registerTaskDrainExpiryWork } from "./services/task-admission.js"; import { markIdleStartupComplete, registerIdleLocalWorkProbe } from "./services/idle-local-work.js"; import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "./services/idle-database-backup.js"; import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js"; @@ -1980,8 +1980,6 @@ async function startServerWithDatabaseTeardown( ); } - scheduled.timers.wake(); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleExternalObjectRefreshSweep(new Date()); @@ -2080,12 +2078,13 @@ async function startServerWithDatabaseTeardown( const startHeartbeatScheduler = () => { const scheduled = registerScheduledWork(deliveryWork, { retryMs: config.heartbeatSchedulerIntervalMs, - tickAgentTimers: async () => { - if ((await heartbeat.resolveSchedulingSuppression()).suppressed) return; - const result = await heartbeat.tickAgentTimers(new Date()); + tickAgentTimers: async signal => { + const result = await heartbeat.tickAgentTimers(new Date(), + () => !signal.aborted && readTaskDrain(new Date()) === null && !isWarmStandby() && !heartbeatSchedulerStopped); if (result.enqueued > 0) logger.info({ ...result }, "heartbeat timer tick enqueued runs"); }, canRun: () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, + nextAgentTimerAt: () => heartbeat.nextAgentTimerAt(), nextIssueMonitorAt: () => heartbeat.nextIssueMonitorAt(), tickIssueMonitors: signal => heartbeat.tickIssueMonitors(new Date(), () => !signal.aborted && !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped), diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 66c63898d73..9531c449260 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -492,6 +492,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch )); const runningCount = Number(runningCountRow?.count ?? 0); const nextAgentStatus = runningCount > 0 ? "running" : "idle"; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await tx .update(agents) .set({ status: nextAgentStatus, lastHeartbeatAt: new Date(), updatedAt: new Date() }) diff --git a/server/src/modules/agent-lifecycle/adapters/records.ts b/server/src/modules/agent-lifecycle/adapters/records.ts index a6ccd8a0dbe..1c59758dd91 100644 --- a/server/src/modules/agent-lifecycle/adapters/records.ts +++ b/server/src/modules/agent-lifecycle/adapters/records.ts @@ -100,6 +100,10 @@ export function agentRecords(db: Db, effects: LifecycleEffects) { }, rejectPendingHire: async (id: string) => { + const [pending] = await db.select({ companyId: agents.companyId, status: agents.status }).from(agents) + .where(and(eq(agents.id, id), eq(agents.lifecycleState, "pending_approval"))).for("update"); + if (!pending) return []; + await effects.beforeStatusChange(db, pending.companyId, id, pending.status, "terminated"); const rows = await db.update(agents).set({ status: "terminated", lifecycleState: "rejected", lifecycleVersion: sql`${agents.lifecycleVersion} + 1`, lifecycleOperation: null, updatedAt: new Date() }).where(and(eq(agents.id, id), eq(agents.lifecycleState, "pending_approval"))).returning(); diff --git a/server/src/services/access.ts b/server/src/services/access.ts index 77b2e648d6c..aa0f9dcc828 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -744,6 +744,7 @@ export function accessService(db: Db) { ); await notifyNativeStatusReassessment(tx, { companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const resetInProgress = await tx .update(issues) .set({ diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index f9d485b9e84..16fae652c95 100644 --- a/server/src/services/agent-configuration.ts +++ b/server/src/services/agent-configuration.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { and, eq } from "drizzle-orm"; import { assets, agentConfigRevisions, agentTaskSessions, agentRuntimeState, type Db } from "@paperclipai/db"; @@ -81,6 +82,9 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura existing.adapterConfig, ); } + if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) { + await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); + } return normalizedPatch; } diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index 974572df305..af44a0355d8 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -270,6 +270,7 @@ export async function prepareConversationTurn( } } await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await tx .update(issues) .set({ @@ -389,6 +390,7 @@ export async function settleConversationTurn( ) return true; await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); 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 602e534dcfc..3650c40f1a6 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -37,6 +37,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 notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await tx.update(agents).set({ reportsTo: null }).where(eq(agents.reportsTo, id)); await tx .update(issues) diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 80d8a816fec..0a6dcaf0fc9 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -38,6 +38,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li }, notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), async beforeStatusChange(tx, companyId, agentId, before, after) { + if (before !== after) await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) { await notifyNativeResponsePresentation(tx, { companyId, agentId }); await notifyNativeReviewResponse(tx, { companyId, agentId }); diff --git a/server/src/services/billing-reconciliation.ts b/server/src/services/billing-reconciliation.ts index 2b146263c43..283e93e77e7 100644 --- a/server/src/services/billing-reconciliation.ts +++ b/server/src/services/billing-reconciliation.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { createFinanceEventInTransaction } from "./finance.js"; import { and, desc, eq, sql } from "drizzle-orm"; import { fastResponseRequests, decisionInvocations, promptInjectionInvocations, budgetReservations, agentRuntimeState, costEvents, costAdjustments, billingInvoices, billingInvoiceLines, financeEvents, heartbeatRuns, type Db } from "@paperclipai/db"; @@ -143,6 +144,7 @@ export function billingReconciliationService(db: Db, hooks: BudgetServiceHooks = reason: input.reason, actorId, pricing: input.pricing }).returning(); // Original provider value and fingerprint remain immutable. cost_cents // is the effective valuation, backed by this append-only correction. + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [updated] = await tx.update(costEvents).set({ costCents: sql`${input.correctedCents}::numeric`, reportedCostCents: event.event.reportedCostCents ?? normalizeCents(event.exact), costStatus: input.pricing.source === "rate_card" ? "estimated" : "reported", pricingProvenance: input.pricing, }).where(eq(costEvents.id, eventId)).returning(); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index cbe88f735ed..6777124ee90 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -301,6 +301,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); @@ -360,6 +361,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; await notifyCompanyRetentionWork(db, policy.companyId); + await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 999c89e17b3..0f2ed547a79 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -332,6 +332,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const renamedPrefix = await resolveRenamedIssuePrefix(tx, id, companyPatch); if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); @@ -487,6 +488,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { const wasAlreadyArchived = existing.status === "archived"; if (!wasAlreadyArchived) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index cafe24ee3ef..4a5925c484b 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { ensureSkillCacheDirectory, touchSkillVersionCache, withSkillVersionCacheLock } from "./skill-version-cache.js"; import { measureRunPhase } from "./execution-preparation-timing.js"; import { verifyExistingBundledSkillRelease } from "./bundled-skill-release-validation.js"; @@ -7305,6 +7306,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo )); for (const row of rows) { await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await tx .update(issues) .set({ hiddenAt: now, updatedAt: now }) diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index b552dc38598..392bb7fe095 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -12,6 +12,8 @@ export function registerDeadlineWork( continuationMs?: number; // Optional batching latency for notifications; already-known deadlines still preempt it. notificationCoalesceMs?: number; + + minimumSpacingMs?: number; }, ) { let requested = 1, reconciled = 0; @@ -31,7 +33,7 @@ export function registerDeadlineWork( } const generation = ++requested; const startedAt = Date.now(); - earliest = startedAt + task.retryMs; + earliest = startedAt + Math.max(task.retryMs, task.minimumSpacingMs ?? 0); await task.run(signal); if (signal.aborted || !task.canRun()) { next = null; return; } const deadline = await task.next(); @@ -39,9 +41,9 @@ export function registerDeadlineWork( 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); + ? (task.minimumSpacingMs ? Date.now() + task.minimumSpacingMs : 0) + : startedAt + Math.max(task.continuationMs ?? task.retryMs, task.minimumSpacingMs ?? 0); next = deadline === null ? null : Math.max(earliest, deadline); coalesceUntil = Date.now() + (task.notificationCoalesceMs ?? 0); reconciled = generation; diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index b7c86dcc40f..f3f5a3e4310 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -595,6 +595,7 @@ export async function settleUnrecoverableExecutions( 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); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [projected] = await tx .update(issues) .set({ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1076558001b..c40c6215a57 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1313,6 +1313,7 @@ export function heartbeatService( trackWakeup, resumeQueuedRuns, parseHeartbeatPolicy, + nextTimerHeartbeatAt, claimDueTimerHeartbeat, } = heartbeatQueue; @@ -1561,6 +1562,7 @@ export function heartbeatService( triggerIssueMonitor, tickTimers, tickAgentTimers, + nextAgentTimerAt, tickIssueMonitors, nextIssueMonitorAt, recoverActiveSessionGoals, @@ -1576,6 +1578,7 @@ export function heartbeatService( toAgentOrgRow, groupAgentOrgRowsByCompany, parseHeartbeatPolicy, + nextTimerHeartbeatAt, claimDueTimerHeartbeat, }); @@ -4268,6 +4271,7 @@ export function heartbeatService( tickTimers, tickAgentTimers, + nextAgentTimerAt, tickIssueMonitors, nextIssueMonitorAt, diff --git a/server/src/services/heartbeat/agent-status.ts b/server/src/services/heartbeat/agent-status.ts index 427e02121a8..f034cd06594 100644 --- a/server/src/services/heartbeat/agent-status.ts +++ b/server/src/services/heartbeat/agent-status.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { agents, heartbeatRuns, type Db } from "@paperclipai/db"; import { and, asc, desc, eq, lte, sql } from "drizzle-orm"; import { trackAgentFirstHeartbeat } from "@paperclipai/shared/telemetry"; @@ -90,6 +91,7 @@ export function createHeartbeatAgentStatus(db: Db) { )).limit(1); const nextStatus = running ? "running" : settledStatus(outcome, options?.keepIdleOnFailure); const now = new Date(); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [updated] = await tx.update(agents).set({ status: nextStatus, errorReason: nextStatus === "error" ? errorReason(failureReason) : null, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index bf1dad379ee..1feef52c3eb 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -481,12 +481,13 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende checkRunCap?: boolean; checkCostCap?: boolean; excludeRunId?: string | null; + now?: Date; } = {}, client: Pick = db, ) { const checkRunCap = options.checkRunCap ?? true; const checkCostCap = options.checkCostCap ?? true; - const { start, end } = currentUtcDayWindow(); + const { start, end } = currentUtcDayWindow(options.now); if (checkRunCap && policy.maxDailyRuns !== null) { const conditions = [ eq(heartbeatRuns.companyId, agent.companyId), @@ -609,18 +610,29 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende return Boolean(row); } + async function nextTimerHeartbeatAt(agent: typeof agents.$inferSelect, now = new Date()): Promise { + const policy = parseHeartbeatPolicy(agent); + if (!policy.enabled || policy.intervalSec <= 0) return null; + + if (policy.maxDailyRuns === 0 || policy.maxDailyCostCents === 0) return null; + if (policy.skipTimerWhenNoActionableWork && !(await hasActionableTimerWork(agent))) return null; + const dueAt = new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime() + policy.intervalSec * 1000; + const cap = await getHeartbeatDailyCapBlock(agent, policy, { now }); + if (!cap) return dueAt; + return Math.max(dueAt, currentUtcDayWindow(now).end.getTime()); + } + async function markTimerHeartbeatChecked( agentId: string, source: WakeupOptions["source"], ) { if (source !== "timer") return; - await db - .update(agents) - .set({ - lastHeartbeatAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(agents.id, agentId)); + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + const now = new Date(); + await tx.update(agents).set({ lastHeartbeatAt: now, updatedAt: now }) + .where(eq(agents.id, agentId)); + }); } async function claimDueTimerHeartbeat( @@ -629,27 +641,22 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende intervalSec: number, ) { const dueBefore = new Date(now.getTime() - intervalSec * 1000); - const claimed = await db - .update(agents) - .set({ - lastHeartbeatAt: now, - updatedAt: now, - }) - .where( - and( + const claimed = await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + return tx + .update(agents) + .set({ lastHeartbeatAt: now, updatedAt: now }) + .where(and( eq(agents.id, agent.id), eq(agents.companyId, agent.companyId), or( lte(agents.lastHeartbeatAt, dueBefore), - and( - isNull(agents.lastHeartbeatAt), - lte(agents.createdAt, dueBefore), - ), + and(isNull(agents.lastHeartbeatAt), lte(agents.createdAt, dueBefore)), ), - ), - ) - .returning({ id: agents.id }) - .then((rows) => rows[0] ?? null); + )) + .returning({ id: agents.id }) + .then((rows) => rows[0] ?? null); + }); if (!claimed) return null; return { wasFirstHeartbeat: !agent.lastHeartbeatAt }; } @@ -1734,6 +1741,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const [candidate] = await tx.select({ companyId: heartbeatRuns.companyId, contextSnapshot: heartbeatRuns.contextSnapshot }) .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.status, "running"))).for("update"); if (!candidate) return; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyChatRunMilestones(tx, { companyId: candidate.companyId, runId, issueId: readNonEmptyString(parseObject(candidate.contextSnapshot).issueId) ?? undefined }); const released = await tx @@ -3906,6 +3914,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id }); await tx .update(issues) @@ -4264,6 +4273,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende finishedAt: now, }); if (source === "timer") { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await tx .update(agents) .set({ @@ -4654,6 +4664,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende finishedAt: now, }); if (source === "timer") { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await tx .update(agents) .set({ @@ -5058,6 +5069,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende resumeQueuedRuns, resumeQueuedRunsAfterDrain, parseHeartbeatPolicy, + nextTimerHeartbeatAt, claimDueTimerHeartbeat, }; } diff --git a/server/src/services/heartbeat/scheduling.test.ts b/server/src/services/heartbeat/scheduling.test.ts index 4e7535c8455..99d23c00a2e 100644 --- a/server/src/services/heartbeat/scheduling.test.ts +++ b/server/src/services/heartbeat/scheduling.test.ts @@ -55,6 +55,7 @@ function callbacks(db: Db) { getSchedulingSuppression: vi.fn(async () => ({ suppressed: false })), getWorktreeExecutionCutoff: vi.fn(async () => null), parseHeartbeatPolicy: vi.fn(() => timerPolicy), + nextTimerHeartbeatAt: vi.fn(async agent => new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime() + 60_000), claimDueTimerHeartbeat: vi.fn(async () => ({ wasFirstHeartbeat: true })), } satisfies HeartbeatSchedulingDependencies; } @@ -387,7 +388,7 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", ( 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, + tickAgentTimers: async () => {}, nextAgentTimerAt: async () => null, tickRoutines: async () => {}, nextRoutineRunAt: async () => null, tickIssueMonitors: signal => f.scheduler.tickIssueMonitors(new Date(), () => !signal.aborted), nextIssueMonitorAt: () => f.scheduler.nextIssueMonitorAt(), }); @@ -531,7 +532,7 @@ describe.skipIf(!postgres.supported)("heartbeat scheduling database boundary", ( 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, + tickAgentTimers: async () => {}, nextAgentTimerAt: async () => null, tickRoutines: async () => {}, nextRoutineRunAt: async () => null, tickIssueMonitors: signal => f.scheduler.tickIssueMonitors(new Date(), () => !draining && !signal.aborted), nextIssueMonitorAt: () => f.scheduler.nextIssueMonitorAt(), }); diff --git a/server/src/services/heartbeat/scheduling.ts b/server/src/services/heartbeat/scheduling.ts index e6d8af61a70..2d499e216b9 100644 --- a/server/src/services/heartbeat/scheduling.ts +++ b/server/src/services/heartbeat/scheduling.ts @@ -69,7 +69,7 @@ type RunPreparation = ReturnType; export interface HeartbeatSchedulingDependencies extends Pick, - Pick, + Pick, Pick { issuesSvc: Pick, "listReviewAttention" | "create">; scheduleBoundedRetryForRun: ReturnType["scheduleBoundedRetryForRun"]; @@ -138,6 +138,7 @@ export function createHeartbeatScheduling( toAgentOrgRow, groupAgentOrgRowsByCompany, parseHeartbeatPolicy, + nextTimerHeartbeatAt, claimDueTimerHeartbeat, } = dependencies; @@ -1285,87 +1286,81 @@ export function createHeartbeatScheduling( return { scanned: pending.length, enqueued, alreadyQueued, invalid }; } - async function tickAgentTimers(now = new Date()) { - if ((await getSchedulingSuppression()).suppressed) { - return { - checked: 0, - enqueued: 0, - skipped: 0, - }; - } - const cutoff = await getWorktreeExecutionCutoff(); + async function eligibleTimerAgents() { + const cutoff = await getWorktreeExecutionCutoff(); - const allAgents = await db - .select({ ...getTableColumns(agents) }) - .from(agents) - .innerJoin(companies, eq(companies.id, agents.companyId)) - .where(eq(companies.status, "active")); - const agentsByCompany = groupAgentOrgRowsByCompany( - allAgents.map(toAgentOrgRow), + const allAgents = await db + .select({ ...getTableColumns(agents) }) + .from(agents) + .innerJoin(companies, eq(companies.id, agents.companyId)) + .where(eq(companies.status, "active")); + const agentsByCompany = groupAgentOrgRowsByCompany( + allAgents.map(toAgentOrgRow), + ); + const eligible: Array<{ agent: typeof agents.$inferSelect; intervalSec: number }> = []; + + for (const agent of allAgents) { + const invokability = evaluateAgentInvokability( + toAgentOrgRow(agent), + agentsByCompany.get(agent.companyId) ?? [], ); - let checked = 0; - let enqueued = 0; - let skipped = 0; + if (!invokability.invokable) continue; + const policy = parseHeartbeatPolicy(agent); + if (!policy.enabled || policy.intervalSec <= 0) continue; - for (const agent of allAgents) { - const invokability = evaluateAgentInvokability( - toAgentOrgRow(agent), - agentsByCompany.get(agent.companyId) ?? [], - ); - if (!invokability.invokable) continue; - const policy = parseHeartbeatPolicy(agent); - if (!policy.enabled || policy.intervalSec <= 0) continue; - - if (cutoff) { - const eligibleIssue = await db - .select({ id: issues.id }) - .from(issues) - .where( - and( - eq(issues.companyId, agent.companyId), - eq(issues.assigneeAgentId, agent.id), - inArray(issues.status, ["todo", "in_progress"]), - gte(issues.createdAt, cutoff), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - if (!eligibleIssue) continue; - } - - checked += 1; - const baseline = new Date( - agent.lastHeartbeatAt ?? agent.createdAt, - ).getTime(); - const elapsedMs = now.getTime() - baseline; - if (elapsedMs < policy.intervalSec * 1000) continue; - const timerClaim = await claimDueTimerHeartbeat( - agent, - now, - policy.intervalSec, - ); - if (!timerClaim) continue; - - const run = await enqueueWakeup(agent.id, { - source: "timer", - triggerDetail: "system", - reason: "heartbeat_timer", - requestedByActorType: "system", - requestedByActorId: "heartbeat_scheduler", - contextSnapshot: { - source: "scheduler", - reason: "interval_elapsed", - now: now.toISOString(), - timerClaimWasFirstHeartbeat: timerClaim.wasFirstHeartbeat, - }, - }); - if (run) enqueued += 1; - else skipped += 1; + if (cutoff) { + const eligibleIssue = await db + .select({ id: issues.id }) + .from(issues) + .where( + and( + eq(issues.companyId, agent.companyId), + eq(issues.assigneeAgentId, agent.id), + inArray(issues.status, ["todo", "in_progress"]), + gte(issues.createdAt, cutoff), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!eligibleIssue) continue; } - return { checked, enqueued, skipped }; + eligible.push({ agent, intervalSec: policy.intervalSec }); } + return eligible; + } + async function nextAgentTimerAt(now = new Date()): Promise { + if ((await getSchedulingSuppression(true)).suppressed) return null; + let next: number | null = null; + for (const { agent } of await eligibleTimerAgents()) { + const deadline = await nextTimerHeartbeatAt(agent, now); + if (deadline !== null) next = next === null ? deadline : Math.min(next, deadline); + } + return next; + } + + async function tickAgentTimers(now = new Date(), canContinue: () => boolean = () => true) { + if (!canContinue() || (await getSchedulingSuppression(true)).suppressed) return { checked: 0, enqueued: 0, skipped: 0 }; + let checked = 0, enqueued = 0, skipped = 0; + for (const { agent, intervalSec } of await eligibleTimerAgents()) { + if (!canContinue()) break; + checked++; + const deadline = await nextTimerHeartbeatAt(agent, now); + if (deadline === null || deadline > now.getTime()) continue; + if (!canContinue()) break; + const timerClaim = await claimDueTimerHeartbeat(agent, now, intervalSec); + if (!timerClaim) continue; + const run = await enqueueWakeup(agent.id, { + source: "timer", triggerDetail: "system", reason: "heartbeat_timer", + requestedByActorType: "system", requestedByActorId: "heartbeat_scheduler", + contextSnapshot: { source: "scheduler", reason: "interval_elapsed", now: now.toISOString(), + timerClaimWasFirstHeartbeat: timerClaim.wasFirstHeartbeat }, + }); + if (run) enqueued++; else skipped++; + } + return { checked, enqueued, skipped }; + } async function tickTimers(now = new Date()) { @@ -1379,6 +1374,7 @@ export function createHeartbeatScheduling( triggerIssueMonitor, tickTimers, tickAgentTimers, + nextAgentTimerAt, tickIssueMonitors, nextIssueMonitorAt, recoverActiveSessionGoals, diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 84b5840a77a..f7c40d13617 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -629,6 +629,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableWorktreeRunExecution !== undefined) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyExternalAgentSettlement(tx); await notifyInstructionCommit(tx); await notifyDirectoryCleanup(tx); diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index f024052ffa5..0d7633f846a 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1001,6 +1001,7 @@ export function issueTreeControlService(db: Db) { const releasedCancelHoldIds = activeCancelHolds.map((hold) => hold.id); const updatedIssues = await db.transaction(async (tx) => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); 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 52002f47cd7..4724dd66699 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -10600,6 +10600,7 @@ export function issueService(db: Db) { : values.parentId ? `issue:${values.parentId}` : undefined); } if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) { @@ -11476,6 +11477,10 @@ export function issueService(db: Db) { || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); } + if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined + || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + } if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt && (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined))) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); @@ -11938,6 +11943,7 @@ export function issueService(db: Db) { } await notifyDecisionTargetIssueRemoval(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); removedIssue = await tx .delete(issues) .where(eq(issues.id, id)) @@ -12065,6 +12071,7 @@ export function issueService(db: Db) { : isNull(issues.executionRunId); const updated = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) .set({ @@ -12193,6 +12200,7 @@ export function issueService(db: Db) { const staleExecutionRunId = current.executionRunId; const adopted = await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx .update(issues) .set(adoptionSet) @@ -12447,6 +12455,7 @@ export function issueService(db: Db) { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) .set({ @@ -12506,6 +12515,7 @@ export function issueService(db: Db) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: existing.companyId, issueId: id }); } await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx .update(issues) .set(patch) diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.test.ts b/server/src/services/native-runtime/native-workspace-reconstruction.test.ts index 3f97cfa67f4..fcff7eb58ec 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.test.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.test.ts @@ -3,7 +3,11 @@ import express from "express"; import request from "supertest"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, agentWakeupRequests, assets, authUsers, companies, companyMemberships, completionContracts, createDb, environmentLeases, heartbeatRuns, issueAttachments, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { activityLog, agents, agentWakeupRequests, assets, authUsers, companies, companyMemberships, completionContracts, createDb, applyPendingMigrations, subscribeDatabaseWork, type Db, environmentLeases, heartbeatRuns, issueAttachments, issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { createHeartbeatQueue } from "../heartbeat/queue.js"; +import { createHeartbeatQueueTestDependencies } from "../../__tests__/helpers/heartbeat-queue.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { errorHandler } from "../../middleware/index.js"; import { issueRoutes } from "../../routes/issues.js"; @@ -31,10 +35,16 @@ describe("reconstruction after confirmed native workspace loss", () => { let db: ReturnType; let runningHeartbeat: ReturnType | null = null; beforeAll(async () => { + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RECONSTRUCTION_TEST_DATABASE_URL"); + if (external) { + await applyPendingMigrations(external); + db = createDb(external); + return; + } temporary = await startEmbeddedPostgresTestDatabase("workspace-reconstruction-"); db = createDb(temporary.connectionString); }, 30_000); - afterAll(async () => { await temporary.cleanup(); }); + afterAll(async () => { await db?.$client.end(); await temporary?.cleanup(); }); afterEach(async () => { if (runningHeartbeat) await drainHeartbeatRunsToQuiescence(db, runningHeartbeat); runningHeartbeat = null; @@ -125,6 +135,28 @@ describe("reconstruction after confirmed native workspace loss", () => { } }); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, later.runId)))[0].status).toBe("cancelled"); }); + it("notifies timer eligibility at outer commit when reconstruction restores actionable work", async () => { + const f = await seed(); + const [agent] = await db.update(agents).set({ runtimeConfig: { heartbeat: { + enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true, + } } }).where(eq(agents.id, f.agentId)).returning(); + const queue = createHeartbeatQueue(db, createHeartbeatQueueTestDependencies(db)); + expect(await queue.nextTimerHeartbeatAt(agent)).toBeNull(); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event)); + try { + await db.transaction(async tx => { + await reconstructNativeWorkspace({ ...f, db: tx as unknown as Db }); + expect(events).toContain("intent"); + expect(events).not.toContain("settled"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("blocked"); + }); + expect(events).toContain("settled"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("todo"); + expect(await queue.nextTimerHeartbeatAt(agent)).toBe(agent.createdAt.getTime() + 600_000); + } finally { unsubscribe(); } + }); + it("pins a historical cancelled workspace through public heartbeat startup without replaying its turn", async () => { const f = await seed(); const later = await interveningCancelledRun(f, false); diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.ts b/server/src/services/native-runtime/native-workspace-reconstruction.ts index d65ba978bb7..5273b30216a 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.ts @@ -186,6 +186,7 @@ export async function reconstructNativeWorkspace(input: { wakePolicy: null, monitorPolicy: null, }).where(eq(issueRecoveryActions.id, source.action.id)); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); 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 b079aa5c647..be8c51b24fa 100644 --- a/server/src/services/scheduled-work.ts +++ b/server/src/services/scheduled-work.ts @@ -7,7 +7,8 @@ export function registerScheduledWork( coordinator: Pick, "register">, options: { retryMs: number; - tickAgentTimers: () => Promise; + tickAgentTimers: (signal: AbortSignal) => Promise; + nextAgentTimerAt: () => Promise; tickIssueMonitors: (signal: AbortSignal) => Promise; nextIssueMonitorAt: () => Promise; tickRoutines: (signal: AbortSignal) => Promise; @@ -28,13 +29,13 @@ export function registerScheduledWork( retryMs: options.retryMs, notificationCoalesceMs: options.retryMs, }); - - - const timers = coordinator.register(DELIVERY_QUEUES.agentTimers, { - retryMs: options.retryMs, + const timers = registerDeadlineWork(coordinator, DELIVERY_QUEUES.agentTimers, { run: options.tickAgentTimers, - hasPending: async () => false, - nextRunAt: () => null, + next: options.nextAgentTimerAt, + canRun: () => options.canRun() && readTaskDrain(new Date()) === null, + retryMs: options.retryMs, + notificationCoalesceMs: options.retryMs, + minimumSpacingMs: options.retryMs, }); return { routines, monitors, timers }; } diff --git a/server/src/services/slack-conversation-lifecycle.ts b/server/src/services/slack-conversation-lifecycle.ts index 05369b13526..0fd213dc084 100644 --- a/server/src/services/slack-conversation-lifecycle.ts +++ b/server/src/services/slack-conversation-lifecycle.ts @@ -132,6 +132,7 @@ export async function settleExternalConversation(db: Db, companyId: string, issu if (!changed.length) return false; await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); 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, { diff --git a/server/src/services/slack-conversation-state.ts b/server/src/services/slack-conversation-state.ts index c3f22cb37f1..dedaa8d6676 100644 --- a/server/src/services/slack-conversation-state.ts +++ b/server/src/services/slack-conversation-state.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { chatConversations, issues, type Db } from "@paperclipai/db"; import { and, eq, sql } from "drizzle-orm"; @@ -44,6 +45,7 @@ export function nonIdleSlackIssueCondition() { /** Called inside message admission while its transaction owns the issue lock. */ export async function resumeExternalConversation(tx: Db, companyId: string, issueId: string) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyExternalConversationSettlement(tx, { companyId, issueId }); // Only the verified idle convention may leave In Review automatically. // A real decision or delivery problem keeps the existing review semantics.