Arnike/paperclip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add heartbeat preflight budget caps (#8347)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Heartbeats are the control-plane path that turns scheduled,
comment-driven, or on-demand wakeups into adapter executions.
> - Budgeting and recurring work need enforcement before an adapter
starts, not only after model usage is recorded.
> - Empty timer wakes also need an opt-in fast-exit path so operators
can keep routine schedules without paying for no-op model turns.
> - This pull request adds heartbeat preflight gates for daily run and
daily cost caps, plus an explicit timer no-work skip policy.
> - The benefit is safer autonomous operation: capped agents stop before
new execution, queued work is cancelled cleanly at claim time, and
proactive agents still run by default unless the operator opts into
no-work skipping.

## Linked Issues or Issue Description

No public issue exists for this change. Inline bug report:

### What happened?

Heartbeat execution can start without enforcing per-agent daily
invocation and spend limits at the heartbeat boundary. A run that was
queued before a cap was reached can also be claimed later and invoke the
adapter unless the cap is checked again immediately before execution.
Operators also do not have an explicit opt-in fast-exit policy for
generic timer wakes with no actionable assigned work.

### Expected behavior

Configured daily run and daily cost caps should stop new heartbeat runs
before adapter execution. Already queued runs should be rechecked at
claim time and cancelled cleanly when a cap is now reached. Queued issue
runs cancelled by daily caps should release their issue execution locks
and promote deferred wakeups without entering immediate recovery loops.
Generic timer no-work skipping should be opt-in so proactive agents
continue to run by default.

### Steps to reproduce

1. Configure an agent heartbeat policy with a one-run daily cap or a
daily cost cap.
2. Create or queue heartbeat wakeups for that agent after the cap has
already been consumed.
3. Observe that without preflight and claim-time checks, the heartbeat
path can still enqueue or claim work that should be blocked before
adapter execution.

### Paperclip version or commit

Reproduced against `master` before this branch.

### Deployment mode

Local dev (`pnpm dev`) / built from source.

### Installation method

Built from source (`pnpm dev` / `pnpm build`).

### Agent adapter(s) involved

Not adapter-specific (core heartbeat scheduling and claim logic).

### Database mode

External Postgres in tests via embedded test harness.

### Access context

Not applicable.

### Node.js version

Node 20 in CI-compatible local development.

### Operating system

macOS local development, Linux CI-compatible tests.

### Relevant logs or output

The regression suite added in this PR covers the failing paths:

```shell
pnpm exec vitest run server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts
```

### Relevant config (if applicable)

```json
{
  "heartbeat": {
    "maxDailyRuns": 1,
    "maxDailyCostCents": 1,
    "skipTimerWhenNoActionableWork": true
  }
}
```

### Additional context

This affects recurring/autonomous operation because the safest place to
stop excess work is before adapter execution starts.

### Privacy checklist

Reviewed for sensitive data; no private logs, credentials, or local
instance URLs are included.

## What Changed

- Added heartbeat policy parsing for per-agent daily run caps, daily
cost caps, and opt-in no-actionable-work timer skipping.
- Added pre-queue daily cap checks while preserving same-issue wake
coalescing.
- Added claim-time cap checks so already queued runs are cancelled
before adapter execution when a cap is reached.
- Added skipped wakeup metadata for cap and timer fast-exit decisions.
- Released issue execution locks for queued issue runs cancelled by
daily caps, with deferred wake promotion and without immediate recovery
loops while caps are active.
- Added regression coverage for timer skipping, proactive default
behavior, run caps, cost caps, queued-run cancellation, started
cancelled runs, and deferred issue wake promotion.

## Verification

- `git diff --check`
- `node -c server/src/services/heartbeat.ts`
- `pnpm exec vitest run
server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts`
- `pnpm --filter @paperclipai/server typecheck`
- Local autoreview: `skills/autoreview/scripts/autoreview --mode branch
--base origin/master --engine codex --model gpt-5.5 --thinking high`
  - Result: clean, no accepted/actionable findings

## Risks

- Medium operational risk because this changes heartbeat scheduling and
claim-time behavior.
- The no-actionable-work timer fast-exit is explicitly opt-in to avoid
suppressing proactive agents unexpectedly.
- Daily run caps count runs by `startedAt` so old queued rows do not
consume today’s cap, while started runs still count even if they later
end as cancelled.
- Queued issue-run cap cancellation uses the existing release/promotion
path with immediate recovery suppressed to avoid retry loops while caps
are active.

## Model Used

Codex with GPT-5.5 high reasoning assisted with implementation, local
testing, and autoreview. The final review gate used local autoreview
with `gpt-5.5` high reasoning.

## 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)
- [ ] 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] If this change affects the UI, I have included before/after
screenshots
- [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
This commit is contained in:
levineam
2026-06-20 00:39:50 -04:00
committed by GitHub
parent 323b6220cc
commit 631b7806ed
2 changed files with 1020 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/services/heartbeat.ts
+321 -40
View File
@@ -3,7 +3,7 @@ import path from "node:path";
import { execFile as execFileCallback } from "node:child_process";
import { promisify } from "node:util";
import { randomUUID } from "node:crypto";
import { and, asc, desc, eq, getTableColumns, gt, inArray, isNull, lt, lte, notInArray, or, sql } from "drizzle-orm";
import { and, asc, desc, eq, getTableColumns, gt, gte, inArray, isNull, lt, lte, notInArray, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
AGENT_DEFAULT_MAX_CONCURRENT_RUNS,
@@ -32,6 +32,7 @@ import {
approvals,
companySkills as companySkillsTable,
companies,
costEvents,
documentAnnotationComments,
documentAnnotationThreads,
documentRevisions,
@@ -238,6 +239,7 @@ const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_r
const CANCELLABLE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const;
const HEARTBEAT_RUN_TERMINAL_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const;
const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["failed", "cancelled", "timed_out"] as const;
const TIMER_ACTIONABLE_ISSUE_STATUSES = ["todo", "in_progress"] as const;
export {
ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS,
ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS,
@@ -6835,9 +6837,169 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
intervalSec: Math.max(0, asNumber(heartbeat.intervalSec, 0)),
wakeOnDemand: asBoolean(heartbeat.wakeOnDemand ?? heartbeat.wakeOnAssignment ?? heartbeat.wakeOnOnDemand ?? heartbeat.wakeOnAutomation, true),
maxConcurrentRuns: normalizeMaxConcurrentRuns(heartbeat.maxConcurrentRuns),
skipTimerWhenNoActionableWork: asBoolean(
heartbeat.skipTimerWhenNoActionableWork ??
heartbeat.requireActionableTimerWork ??
heartbeat.issueOnlyTimer,
false,
),
maxDailyRuns: normalizeOptionalNonNegativeInteger(
heartbeat.maxDailyRuns ?? heartbeat.dailyRunLimit ?? heartbeat.dailyRunCap ?? heartbeat.maxRunsPerDay,
),
maxDailyCostCents: normalizeOptionalNonNegativeInteger(
heartbeat.maxDailyCostCents ??
heartbeat.dailyCostCentsLimit ??
heartbeat.dailySpendCentsLimit ??
heartbeat.dailyBudgetCents,
),
};
}
function normalizeOptionalNonNegativeInteger(value: unknown) {
if (value === null || value === undefined || value === "") return null;
const normalized = Math.floor(asNumber(value, 0));
return normalized >= 0 ? normalized : null;
}
function currentUtcDayWindow(now = new Date()) {
const start = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate(), 0, 0, 0, 0));
const end = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() + 1, 0, 0, 0, 0));
return { start, end };
}
async function getHeartbeatDailyCapBlock(
agent: typeof agents.$inferSelect,
policy: ReturnType<typeof parseHeartbeatPolicy>,
options: { checkRunCap?: boolean; checkCostCap?: boolean; excludeRunId?: string | null } = {},
client: Pick<Db, "select"> = db,
) {
const checkRunCap = options.checkRunCap ?? true;
const checkCostCap = options.checkCostCap ?? true;
const { start, end } = currentUtcDayWindow();
if (checkRunCap && policy.maxDailyRuns !== null) {
const conditions = [
eq(heartbeatRuns.companyId, agent.companyId),
eq(heartbeatRuns.agentId, agent.id),
gte(heartbeatRuns.startedAt, start),
lt(heartbeatRuns.startedAt, end),
notInArray(heartbeatRuns.status, ["queued", "scheduled_retry"]),
];
if (options.excludeRunId) {
conditions.push(sql`${heartbeatRuns.id} <> ${options.excludeRunId}`);
}
const [row] = await client
.select({ total: sql<number>`count(*)::integer` })
.from(heartbeatRuns)
.where(and(...conditions));
const observed = Number(row?.total ?? 0);
if (observed >= policy.maxDailyRuns) {
return {
reason: "heartbeat.daily_run_limit",
observed,
limit: policy.maxDailyRuns,
};
}
}
if (checkCostCap && policy.maxDailyCostCents !== null) {
const [row] = await client
.select({ total: sql<number>`coalesce(sum(${costEvents.costCents})::bigint, 0)` })
.from(costEvents)
.where(
and(
eq(costEvents.companyId, agent.companyId),
eq(costEvents.agentId, agent.id),
gte(costEvents.occurredAt, start),
lt(costEvents.occurredAt, end),
),
);
const observed = Number(row?.total ?? 0);
if (observed >= policy.maxDailyCostCents) {
return {
reason: "heartbeat.daily_cost_limit",
observed,
limit: policy.maxDailyCostCents,
};
}
}
return null;
}
async function cancelQueuedRunForHeartbeatDailyCap(
run: typeof heartbeatRuns.$inferSelect,
dailyCapBlock: NonNullable<Awaited<ReturnType<typeof getHeartbeatDailyCapBlock>>>,
) {
const now = new Date();
const reason = "Cancelled because the agent reached a per-day heartbeat budget cap before adapter invocation";
const cancelled = await setRunStatus(run.id, "cancelled", {
finishedAt: now,
error: reason,
errorCode: dailyCapBlock.reason,
resultJson: {
...parseObject(run.resultJson),
stopReason: dailyCapBlock.reason,
observed: dailyCapBlock.observed,
limit: dailyCapBlock.limit,
effectiveTimeoutSec: 0,
timeoutConfigured: false,
timeoutSource: "heartbeat_daily_cap_gate",
timeoutFired: false,
},
});
if (!cancelled) return null;
await setWakeupStatus(run.wakeupRequestId, "skipped", {
finishedAt: now,
error: reason,
});
await appendRunEvent(cancelled, await nextRunEventSeq(cancelled.id), {
eventType: "lifecycle",
stream: "system",
level: "warn",
message: reason,
payload: {
reason: dailyCapBlock.reason,
observed: dailyCapBlock.observed,
limit: dailyCapBlock.limit,
},
});
await releaseIssueExecutionAndPromote(cancelled, { suppressImmediateRecovery: true });
return cancelled;
}
async function hasActionableTimerWork(agent: typeof agents.$inferSelect) {
const row = await db
.select({ id: issues.id })
.from(issues)
.where(
and(
eq(issues.companyId, agent.companyId),
eq(issues.assigneeAgentId, agent.id),
isNull(issues.assigneeUserId),
isNull(issues.hiddenAt),
inArray(issues.status, [...TIMER_ACTIONABLE_ISSUE_STATUSES]),
),
)
.limit(1)
.then((rows) => rows[0] ?? null);
return Boolean(row);
}
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));
}
function parseMaxTurnContinuationPolicy(agent: typeof agents.$inferSelect): MaxTurnContinuationPolicy {
const runtimeConfig = parseObject(agent.runtimeConfig);
const heartbeat = parseObject(runtimeConfig.heartbeat);
@@ -6915,6 +7077,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return null;
}
const dailyCapBlock = await getHeartbeatDailyCapBlock(agent, parseHeartbeatPolicy(agent), {
excludeRunId: run.id,
checkRunCap: true,
checkCostCap: true,
});
if (dailyCapBlock) {
await cancelQueuedRunForHeartbeatDailyCap(run, dailyCapBlock);
return null;
}
const issueId = readNonEmptyString(context.issueId);
if (issueId) {
const activePauseHold = await treeControlSvc.getActivePauseHoldGate(run.companyId, issueId);
@@ -9801,7 +9973,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
);
}
async function releaseIssueExecutionAndPromote(run: typeof heartbeatRuns.$inferSelect) {
async function releaseIssueExecutionAndPromote(
run: typeof heartbeatRuns.$inferSelect,
options: { suppressImmediateRecovery?: boolean } = {},
) {
const runContext = parseObject(run.contextSnapshot);
const contextIssueId = readNonEmptyString(runContext.issueId);
const taskKey = deriveTaskKeyWithHeartbeatFallback(runContext, null);
@@ -10188,6 +10363,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
if (!issueNeedsImmediateRecovery) {
return { kind: "released" as const };
}
if (options.suppressImmediateRecovery) {
return { kind: "released" as const };
}
const existingExecutionPath = await tx
.select({ id: heartbeatRuns.id })
@@ -10410,6 +10588,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
...patch,
});
};
const writeSkippedHeartbeatRequest = async (skipReason: string, details: Record<string, unknown>) => {
await writeSkippedRequest(skipReason, {
payload: {
...(payload ?? {}),
heartbeatSkip: details,
},
});
};
const company = await db
.select({ status: companies.status })
@@ -10523,6 +10709,20 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return null;
}
const genericTimerWake =
source === "timer" &&
!issueId &&
!wakeCommentId &&
!readNonEmptyString(enrichedContextSnapshot.taskId) &&
!readNonEmptyString(enrichedContextSnapshot.taskKey);
if (policy.skipTimerWhenNoActionableWork && genericTimerWake && !(await hasActionableTimerWork(agent))) {
await writeSkippedHeartbeatRequest("heartbeat.timer.no_actionable_work", {
reason: "No assigned todo or in_progress issue requires this agent before timer adapter invocation.",
});
await markTimerHeartbeatChecked(agentId, source);
return null;
}
if (issueId) {
const activePauseHold = await treeControlSvc.getActivePauseHoldGate(agent.companyId, issueId);
if (activePauseHold) {
@@ -10996,6 +11196,41 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
}
}
const dailyCapBlock = await getHeartbeatDailyCapBlock(agent, policy, {}, tx);
if (dailyCapBlock) {
const now = new Date();
await tx.insert(agentWakeupRequests).values({
companyId: agent.companyId,
agentId,
source,
triggerDetail,
reason: dailyCapBlock.reason,
payload: {
...(payload ?? {}),
heartbeatSkip: {
reason: "Per-agent heartbeat daily cap reached before adapter invocation.",
observed: dailyCapBlock.observed,
limit: dailyCapBlock.limit,
},
},
status: "skipped",
requestedByActorType: opts.requestedByActorType ?? null,
requestedByActorId: opts.requestedByActorId ?? null,
idempotencyKey: opts.idempotencyKey ?? null,
finishedAt: now,
});
if (source === "timer") {
await tx
.update(agents)
.set({
lastHeartbeatAt: now,
updatedAt: now,
})
.where(eq(agents.id, agentId));
}
return { kind: "skipped" as const };
}
const wakeupRequest = await tx
.insert(agentWakeupRequests)
.values({
@@ -11130,46 +11365,92 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
return mergedRun;
}
const wakeupRequest = await db
.insert(agentWakeupRequests)
.values({
companyId: agent.companyId,
agentId,
source,
triggerDetail,
reason,
payload,
status: "queued",
requestedByActorType: opts.requestedByActorType ?? null,
requestedByActorId: opts.requestedByActorId ?? null,
idempotencyKey: opts.idempotencyKey ?? null,
})
.returning()
.then((rows) => rows[0]);
const queueOutcome = await db.transaction(async (tx) => {
await tx.execute(
sql`select id from agents where id = ${agentId} and company_id = ${agent.companyId} for update`,
);
const newRun = await db
.insert(heartbeatRuns)
.values({
companyId: agent.companyId,
agentId,
invocationSource: source,
triggerDetail,
status: "queued",
wakeupRequestId: wakeupRequest.id,
contextSnapshot: enrichedContextSnapshot,
sessionIdBefore: sessionBefore,
continuationAttempt,
})
.returning()
.then((rows) => rows[0]);
const dailyCapBlock = await getHeartbeatDailyCapBlock(agent, policy, {}, tx);
if (dailyCapBlock) {
const now = new Date();
await tx.insert(agentWakeupRequests).values({
companyId: agent.companyId,
agentId,
source,
triggerDetail,
reason: dailyCapBlock.reason,
payload: {
...(payload ?? {}),
heartbeatSkip: {
reason: "Per-agent heartbeat daily cap reached before adapter invocation.",
observed: dailyCapBlock.observed,
limit: dailyCapBlock.limit,
},
},
status: "skipped",
requestedByActorType: opts.requestedByActorType ?? null,
requestedByActorId: opts.requestedByActorId ?? null,
idempotencyKey: opts.idempotencyKey ?? null,
finishedAt: now,
});
if (source === "timer") {
await tx
.update(agents)
.set({
lastHeartbeatAt: now,
updatedAt: now,
})
.where(eq(agents.id, agentId));
}
return { kind: "skipped" as const };
}
await db
.update(agentWakeupRequests)
.set({
runId: newRun.id,
updatedAt: new Date(),
})
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
const wakeupRequest = await tx
.insert(agentWakeupRequests)
.values({
companyId: agent.companyId,
agentId,
source,
triggerDetail,
reason,
payload,
status: "queued",
requestedByActorType: opts.requestedByActorType ?? null,
requestedByActorId: opts.requestedByActorId ?? null,
idempotencyKey: opts.idempotencyKey ?? null,
})
.returning()
.then((rows) => rows[0]);
const newRun = await tx
.insert(heartbeatRuns)
.values({
companyId: agent.companyId,
agentId,
invocationSource: source,
triggerDetail,
status: "queued",
wakeupRequestId: wakeupRequest.id,
contextSnapshot: enrichedContextSnapshot,
sessionIdBefore: sessionBefore,
continuationAttempt,
})
.returning()
.then((rows) => rows[0]);
await tx
.update(agentWakeupRequests)
.set({
runId: newRun.id,
updatedAt: new Date(),
})
.where(eq(agentWakeupRequests.id, wakeupRequest.id));
return { kind: "queued" as const, run: newRun };
});
if (queueOutcome.kind === "skipped") return null;
const newRun = queueOutcome.run;
publishLiveEvent({
companyId: newRun.companyId,