perf: schedule agent timers from durable deadlines (#17876)

<!-- entire-trail-link-start -->
https://entire.com/gh/paperclipai/-/trails/58/changes/paperclip/58
<!-- entire-trail-link-end -->

## 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 <noreply@paperclip.ing>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
authored and GitHub committed 2026-10-29 08:58:27 -05:00
1 parent b81b870732
commit ba07f85cad
36 files changed
+1206 -140

No files matched your search

+138
View File
@@ -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.
@@ -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:**
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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<string, unknown> = { 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<ReturnType<typeof fixture>>, 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<void>(r => { release = r; }), written = new Promise<void>(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 });
});
});
@@ -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<typeof createDeliveryWorkCoordinator> | 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();
});
@@ -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 () => {
@@ -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);
+1 -1
View File
@@ -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; } };
@@ -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);
@@ -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;
+5 -6
View File
@@ -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),
@@ -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() })
@@ -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();
+1
View File
@@ -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({
@@ -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;
}
@@ -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)
+1
View File
@@ -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)
+1
View File
@@ -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 });
@@ -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();
+2
View File
@@ -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);
+2
View File
@@ -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);
+2
View File
@@ -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 })
+5 -3
View File
@@ -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;
@@ -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({
+4
View File
@@ -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,
@@ -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,
+36 -24
View File
@@ -481,12 +481,13 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
checkRunCap?: boolean;
checkCostCap?: boolean;
excludeRunId?: string | null;
now?: Date;
} = {},
client: Pick<Db, "select"> = 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<number | null> {
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,
};
}
@@ -55,6 +55,7 @@ function callbacks(db: Db) {
getSchedulingSuppression: vi.fn(async () => ({ suppressed: false })),
getWorktreeExecutionCutoff: vi.fn<HeartbeatSchedulingDependencies["getWorktreeExecutionCutoff"]>(async () => null),
parseHeartbeatPolicy: vi.fn<HeartbeatSchedulingDependencies["parseHeartbeatPolicy"]>(() => timerPolicy),
nextTimerHeartbeatAt: vi.fn<HeartbeatSchedulingDependencies["nextTimerHeartbeatAt"]>(async agent => new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime() + 60_000),
claimDueTimerHeartbeat: vi.fn<HeartbeatSchedulingDependencies["claimDueTimerHeartbeat"]>(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(),
});
+71 -75
View File
@@ -69,7 +69,7 @@ type RunPreparation = ReturnType<typeof createHeartbeatRunPreparation>;
export interface HeartbeatSchedulingDependencies
extends Pick<HeartbeatRetryDependencies, "getRun" | "getAgent" | "getWorktreeExecutionCutoff">,
Pick<HeartbeatQueue, "enqueueWakeup" | "parseHeartbeatPolicy" | "claimDueTimerHeartbeat">,
Pick<HeartbeatQueue, "enqueueWakeup" | "parseHeartbeatPolicy" | "nextTimerHeartbeatAt" | "claimDueTimerHeartbeat">,
Pick<RunPreparation, "toAgentOrgRow" | "groupAgentOrgRowsByCompany"> {
issuesSvc: Pick<ReturnType<typeof issueService>, "listReviewAttention" | "create">;
scheduleBoundedRetryForRun: ReturnType<typeof createHeartbeatRetries>["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<number | null> {
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,
+1
View File
@@ -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);
@@ -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;
+10
View File
@@ -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)
@@ -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<typeof createDb>;
let runningHeartbeat: ReturnType<typeof heartbeatService> | 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);
@@ -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,
+8 -7
View File
@@ -7,7 +7,8 @@ export function registerScheduledWork(
coordinator: Pick<ReturnType<typeof createDeliveryWorkCoordinator>, "register">,
options: {
retryMs: number;
tickAgentTimers: () => Promise<unknown>;
tickAgentTimers: (signal: AbortSignal) => Promise<unknown>;
nextAgentTimerAt: () => Promise<number | null>;
tickIssueMonitors: (signal: AbortSignal) => Promise<unknown>;
nextIssueMonitorAt: () => Promise<number | null>;
tickRoutines: (signal: AbortSignal) => Promise<unknown>;
@@ -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 };
}
@@ -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, {
@@ -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.