From d75cb0d46baadff35d52867f7bb2235dc58843fe Mon Sep 17 00:00:00 2001 From: Jeremy McSpadden <211150+jeremymcs@users.noreply.github.com> Date: Sat, 29 Aug 2026 17:00:07 -0500 Subject: [PATCH] feat: meter coding-agent spend and cap it per hour MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PatchDeck could not say how much paid agent work it had done. `agent_runs` is a per-babysit-session record with a `pr_id` foreign key, so issue work, CI and deployment healing, release notes, social posts, and PR questions were invisible, and PR-scoped history was cascade-deleted with the PR. Adds a separate `agent_invocations` ledger — one row per `codex`/`claude` process spawn, with duration, exit code, resolved agent, model, and outcome, and no foreign keys so history outlives its target. Every spawn already funnelled through `runAgentCommand`, so metering hooks in there; an AsyncLocalStorage context established by each unit of work attributes the row. Adds `maxAgentInvocationsPerHour` (default 0, unlimited). When the rolling-hour ceiling is reached the dispatcher stops claiming agent-invoking job kinds, and any spawn started inside an already-running job — CI healing, agent fallback, conflict repair — is refused. Queued work is not failed: it waits and resumes as the window rolls. The refusal classifies as transient so it never consumes a paid retry attempt. Health-check probes are recorded but never counted, so opening Settings cannot exhaust a budget. Ledger rows are pruned after 30 days on the existing retention sweep, and rows left `running` by a hard shutdown are closed at boot. Surfaced through `GET /api/agent-spend`, the `get_agent_spend` MCP tool, a header pill shown once a ceiling is set, and a Settings field. Plan: docs/plans/agent-spend-metering.md Verified with: npm run check, npx eslint ., npm run build, npm run test:all (827) --- LOCAL_API.md | 39 +++ README.md | 6 + client/src/components/AppHeader.tsx | 51 ++- client/src/pages/settings.tsx | 9 + docs/plans/agent-spend-metering.md | 436 +++++++++++++++++++++++++ docs/public/configuration.md | 12 + server/agentRunner.ts | 60 ++-- server/agentSpend.test.ts | 277 ++++++++++++++++ server/agentSpend.ts | 282 ++++++++++++++++ server/appRuntime.ts | 14 + server/babysitter.ts | 31 +- server/backgroundJobDispatcher.test.ts | 63 ++++ server/backgroundJobDispatcher.ts | 44 ++- server/backgroundJobHandlers.ts | 25 +- server/ciHealingAgent.ts | 9 +- server/defaultConfig.test.ts | 7 + server/defaultConfig.ts | 1 + server/deploymentHealingAgent.ts | 10 +- server/failureRecovery.test.ts | 23 ++ server/failureRecovery.ts | 8 + server/issueWorkAgent.ts | 10 +- server/logsRetention.test.ts | 36 +- server/logsRetention.ts | 24 ++ server/mcp.ts | 10 + server/memoryStorage.ts | 80 +++++ server/prQuestionAgent.ts | 9 +- server/releaseAgent.ts | 15 +- server/releaseSocialPostAgent.ts | 10 +- server/routes.test.ts | 49 +++ server/routes.ts | 4 + server/sqliteStorage.ts | 172 +++++++++- server/storage.test.ts | 169 ++++++++++ server/storage.ts | 22 ++ shared/schema.ts | 72 ++++ 34 files changed, 2036 insertions(+), 53 deletions(-) create mode 100644 docs/plans/agent-spend-metering.md create mode 100644 server/agentSpend.test.ts create mode 100644 server/agentSpend.ts diff --git a/LOCAL_API.md b/LOCAL_API.md index 638db89..5f565be 100644 --- a/LOCAL_API.md +++ b/LOCAL_API.md @@ -248,6 +248,7 @@ compiled output instead: | `get_config` | Read current configuration | | `update_config` | Partially update configuration | | `get_runtime` | Get runtime state (drain mode, active queue handlers) | +| `get_agent_spend` | Get coding-agent spend for the rolling hour, with the ceiling and a breakdown by work kind | | `set_drain_mode` | Enable/disable drain mode for new queue claims | | `list_changelogs` | List social-media changelogs | | `get_changelog` | Get one changelog by ID | @@ -766,6 +767,44 @@ Get the current runtime state. --- +#### `GET /api/agent-spend` + +Get coding-agent spend for the rolling hour. Every spawn of the `codex` or +`claude` CLI is recorded, whichever path caused it — PR work, feedback +evaluation, issue work, issue decompose/verify, CI healing, deployment healing, +PR questions, release notes, and social posts. + +`max` is the `maxAgentInvocationsPerHour` setting. **`0` means unlimited**, and +`remaining` is then `null`. `used` excludes agent health-check probes, which are +recorded in `byKind` but never counted against the ceiling. + +`windowMs` is a rolling window, not a calendar hour: `resetsAt` is when the +oldest invocation inside the window ages out and the next slot frees up. + +When the ceiling is reached, the dispatcher stops claiming agent-invoking job +kinds and any agent spawn started inside an already-running job is refused. +Queued work is not failed — it waits and resumes on its own. + +**Response** `200` +```json +{ + "windowMs": 3600000, + "windowStartedAt": "2026-08-29T11:00:00.000Z", + "resetsAt": "2026-08-29T12:14:00.000Z", + "max": 30, + "used": 18, + "remaining": 12, + "exhausted": false, + "byKind": [ + { "workKind": "babysit_pr", "count": 11, "totalDurationMs": 1830000 }, + { "workKind": "work_issue", "count": 5, "totalDurationMs": 920000 }, + { "workKind": "heal_ci", "count": 2, "totalDurationMs": 240000 } + ] +} +``` + +--- + #### `POST /api/runtime/drain` Enable or disable drain mode. When enabled, the dispatcher stops claiming new diff --git a/README.md b/README.md index 3131dd2..bce6bc0 100644 --- a/README.md +++ b/README.md @@ -180,6 +180,7 @@ Most configuration is editable in Settings: - GitHub progress replies - CI healing - Agent retry attempts for failed work +- Hourly ceiling on agent runs - Theme Key defaults: @@ -195,6 +196,11 @@ Key defaults: needs a person — expired agent credentials, a missing CLI — is parked and shown under **Needs attention**, and is retried again roughly hourly so it resumes on its own once fixed. - Drain mode pauses new agent work without deleting tracked state. +- **Max agent runs per hour** (default 0, meaning unlimited) caps coding-agent runs + across every path — PR work, issues, CI and deployment healing, releases, and + questions. When the ceiling is reached, queued work waits instead of failing and + resumes on its own as the rolling hour moves forward. The header shows usage once + a ceiling is set, and `GET /api/agent-spend` reports the full breakdown. ## Authentication diff --git a/client/src/components/AppHeader.tsx b/client/src/components/AppHeader.tsx index ffe616b..1dab3e1 100644 --- a/client/src/components/AppHeader.tsx +++ b/client/src/components/AppHeader.tsx @@ -2,7 +2,7 @@ import type { ReactNode } from "react"; import { Link } from "wouter"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { apiRequest } from "@/lib/queryClient"; -import type { ActivitySnapshot, Config, RuntimeState } from "@shared/schema"; +import type { ActivitySnapshot, AgentSpendSummary, Config, RuntimeState } from "@shared/schema"; import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover"; import { Switch } from "@/components/ui/switch"; import { ToastAction } from "@/components/ui/toast"; @@ -371,6 +371,52 @@ function GitHubRateLimitNotice() { ); } +function AgentSpendNotice() { + const { data: config } = useQuery({ + queryKey: ["/api/config"], + }); + const uiPollIntervalMs = getUiPollIntervalMs(config); + const { data: spend } = useQuery({ + queryKey: ["/api/agent-spend"], + refetchInterval: uiPollIntervalMs, + }); + + // No ceiling configured means unlimited, and an unlimited budget has nothing + // worth taking up header space. + if (!spend || spend.max === 0) { + return null; + } + + const resetTime = new Date(spend.resetsAt).toLocaleTimeString("en-US", { + hour: "numeric", + minute: "2-digit", + }); + const nearingCeiling = spend.used >= Math.ceil(spend.max * 0.8); + + const label = spend.exhausted + ? `Agent runs paused until ${resetTime}` + : `${spend.used}/${spend.max} agent runs`; + + const tooltip = spend.exhausted + ? `The hourly ceiling of ${spend.max} agent runs is reached. Queued work resumes on its own around ${resetTime}.` + : `${spend.used} of ${spend.max} agent runs used in the last hour, across PR work, issues, healing, releases, and questions.`; + + const toneClass = spend.exhausted || nearingCeiling + ? "border-warning-border bg-warning-muted text-warning-foreground hover:border-warning hover:bg-warning-muted/80" + : "border-border text-muted-foreground hover:border-foreground/30 hover:text-foreground"; + + return ( + + {label} + + ); +} + export function AppHeader({ active, status, @@ -407,8 +453,9 @@ export function AppHeader({ })} -
+
+
{status ? ( diff --git a/client/src/pages/settings.tsx b/client/src/pages/settings.tsx index 934986f..8266d4c 100644 --- a/client/src/pages/settings.tsx +++ b/client/src/pages/settings.tsx @@ -40,6 +40,7 @@ const DEFAULT_SETTING_VALUES = { maxChangesPerRun: 20, maxConcurrentBabysitRuns: 3, maxAgentRetryAttempts: 3, + maxAgentInvocationsPerHour: 0, maxHealingAttemptsPerSession: 3, maxHealingAttemptsPerFingerprint: 2, maxConcurrentHealingRuns: 1, @@ -1358,6 +1359,14 @@ export default function Settings() { defaultValue={DEFAULT_SETTING_VALUES.maxAgentRetryAttempts} disabled={updateConfigMutation.isPending} /> + updateConfigMutation.mutate({ maxAgentInvocationsPerHour: v })} + defaultValue={DEFAULT_SETTING_VALUES.maxAgentInvocationsPerHour} + disabled={updateConfigMutation.isPending} + />
diff --git a/docs/plans/agent-spend-metering.md b/docs/plans/agent-spend-metering.md new file mode 100644 index 0000000..05642b7 --- /dev/null +++ b/docs/plans/agent-spend-metering.md @@ -0,0 +1,436 @@ +// PatchDeck + Agent Spend Metering Plan +// Copyright (c) 2026 Jeremy McSpadden + +# Agent Spend Metering + +## Goal + +PatchDeck knows how much paid agent work it has done, across every path that can +invoke one, and refuses to exceed a ceiling the user sets. Turning on Auto PRs, +Auto Issues, CI healing, and deployment healing at the same time becomes a +bounded decision instead of an open-ended one. + +This closes the follow-up tracked in `docs/plans/resilient-automation.md:71-73`: + +> A true rolling spend ceiling (max agent runs per hour, enforced before every +> invocation) was considered and deferred — `agentRuns` is PR-scoped today +> (`prId`, written only by `babysitter.ts`), so metering global spend needs +> accounting at every agent entry point. Tracked as a follow-up. + +Success criteria (verifiable): + +1. Every process spawn of `codex` or `claude` writes exactly one ledger row, + including the paths that never touch a PR (issue decompose, release notes, + social post, PR Q&A). +2. A ledger row survives deletion of the PR or issue it was working on. +3. With a ceiling configured and reached, the dispatcher stops claiming + agent-invoking jobs. Nothing fails; jobs stay `queued` and flow again when + the window rolls. +4. An agent invocation started *inside* an already-running job (CI healing, + agent fallback, conflict repair) is refused once the ceiling is reached, and + the refusal backs the job off rather than parking it. +5. `GET /api/agent-spend` reports used / remaining / reset time / breakdown by + work kind, and the dashboard shows it. +6. `maxAgentInvocationsPerHour: 0` (the default) reproduces today's behaviour + exactly. + +## Current Behaviour (why nothing is counted) + +`agent_runs` looks like a spend ledger and is not one. It is a **per-babysit-session +record** used for interrupted-run recovery and failure markers: + +| Fact | Code | Consequence | +|---|---|---| +| One row per babysit session, not per CLI invocation. `phase` is mutated as the session progresses. | `babysitter.ts:3256-3282` | A session that evaluates, applies, falls back to the other agent, then repairs a conflict is *one* row. | +| `pr_id TEXT NOT NULL` with `FOREIGN KEY(pr_id) REFERENCES prs(id) ON DELETE CASCADE` | `sqliteStorage.ts:683-696` | Non-PR agent work cannot be represented, and PR-scoped history is destroyed when the PR record goes away. | +| Written only by `babysitter.ts` | `grep upsertAgentRun server/` | Issue work, CI healing, deployment healing, release notes, social posts, and PR Q&A are invisible. | +| No duration, exit code, model, or outcome-vs-cost fields | `shared/schema.ts:179-192` | `createdAt`/`updatedAt` give a rough session span, nothing per invocation. | + +So `agent_runs` stays exactly as it is. This plan adds a separate ledger. + +### Every path that spawns a paid agent + +All of them funnel through **one function**, `runAgentCommand` +(`agentRunner.ts:132`), which is the whole reason this is tractable: + +``` +runAgentCommand(agent, args, options) <- the choke point + ├── evaluateFixNecessityWithAgent agentRunner.ts:205 + ├── applyFixesWithAgent agentRunner.ts:281 + │ └── runAgentOneShot agentRunner.ts:320 + ├── checkAgentHealth agentRunner.ts:164 (probe, see below) + └── called directly by three modules +``` + +| # | Caller | Site | Work kind | Job kind | +|---|---|---|---|---| +| 1 | `babysitter.ts` apply | `3694`, `4190`, `4211` | `babysit_pr` | `babysit_pr` | +| 2 | `babysitter.ts` evaluate | `3913`, `3921` | `evaluate_feedback` | `babysit_pr` | +| 3 | `issueWorkAgent.ts` | `482` | `work_issue` | `work_issue` | +| 4 | `issueDecompose.ts` | `189` | `decompose_issue` | `evaluate_issue` | +| 5 | `issueVerify.ts` | `147` | `verify_issue` | `verify_issue` | +| 6 | `ciHealingAgent.ts` | `332` | `heal_ci` | **none — runs inside `babysit_pr`** | +| 7 | `deploymentHealingAgent.ts` | `131` | `heal_deployment` | `heal_deployment` | +| 8 | `prQuestionAgent.ts` | `27` | `answer_pr_question` | `answer_pr_question` | +| 9 | `releaseAgent.ts` | `49`, `74` | `release_notes` | `process_release_run` | +| 10 | `releaseSocialPostAgent.ts` | `97` | `social_post` | `generate_social_changelog` | + +Row 6 is the reason a dispatcher-only gate is not enough: CI healing is started +from inside `babysitter.ts:55` while a `babysit_pr` job is already leased, so the +dispatcher has no further say. Same for the fallback-agent re-run at +`babysitter.ts:4211`. + +`checkAgentHealth` does spawn the CLI, but it is a fixed ~1-token probe used by +onboarding and diagnostics. It records as `outcome: "probe"` and is excluded +from the ceiling — otherwise opening Settings could exhaust a budget. + +## Decisions + +Recommended answers below. Items marked **CONFIRM** change the shape of the +feature and are worth a yes/no before implementation. + +### 1. Unit of account + +Count **invocations**, record **duration**, enforce on invocations. + +Not tokens: neither `codex exec` nor `claude -p` reports token usage on stdout in +the shapes this app parses (`parseEvaluationOutput`, `summarizeCommandResult`), +and scraping it would couple PatchDeck to CLI output formats that change. Not +wall-clock minutes as the primary unit either — a 90-minute `applyFixesWithAgent` +(`timeoutMs = 5400000`, `agentRunner.ts:291`) and a 20-second one-shot cost wildly +different amounts, but invocation count is the number a user can reason about and +the one that maps to "how many times did this thing decide to spend money". +`duration_ms` goes in the ledger so a duration-based ceiling can be added later +without a second migration. + +### 2. Window + +**Rolling hour**, computed as `started_at >= now - 3600000`. Not calendar-hour +buckets: a calendar window lets 2× the ceiling run across a boundary, which is +exactly the burst an unattended overnight run produces. + +### 3. Behaviour at the ceiling — **CONFIRM** + +**Park-and-auto-resume**, modelled on drain mode. Jobs stay `queued`, the +dispatcher stops claiming agent-invoking kinds, and work resumes on its own as +the window rolls forward. No manual clear, no failed jobs, no lost work. + +The alternative — hard stop requiring a manual reset — is safer against a +runaway but reintroduces exactly the "human clears errors" failure mode the +resilient-automation work removed. Recommending auto-resume. + +### 4. Default value — **CONFIRM** + +`maxAgentInvocationsPerHour: 0` meaning **unlimited**, matching how +`maxAgentRetryAttempts` shipped disabled-by-default in spirit. Upgrading changes +no behaviour; users opt in once the new spend view tells them what their normal +hour actually looks like. + +The alternative is shipping a generous default (30/hour) so the safety net +exists without being configured. That is a behaviour change on upgrade for +anyone running busy repos, and PatchDeck has no data yet on what a normal hour +is. Recommending 0, then revisiting after a release of real numbers. + +### 5. Context propagation — **CONFIRM** + +`runAgentCommand` is the only place that sees every invocation, and it has no +idea what work it is serving. Two ways to fix that: + +**(a) Explicit parameter.** Thread an `AgentInvocationContext` through all ten +call sites. Honest and greppable, matches the repo's dependency-injection +style. Costs: touches ten modules plus every injected seam +(`deps.applyFixesWithAgent` in `ciHealingAgent.ts:10`, `issueWorkAgent.ts:44`, +`deploymentHealingAgent.ts:35`, `runOneShot` in `issueDecompose.ts:20`, +`issueVerify.ts:22`), and an eleventh call site added next year silently escapes +the meter. + +**(b) `AsyncLocalStorage`.** A new `server/agentSpend.ts` owns an ALS store. +Each entry point wraps its work once — `withAgentWork({ kind, repo, targetId }, fn)` +— and `runAgentCommand` reads the ambient context. One choke point, impossible to +bypass, and an invocation arriving with *no* context is a detectable bug that +logs at `warn` and still records as `kind: "unattributed"`. + +Recommending **(b)**, with (a)'s discipline preserved by a test that asserts each +of the ten call sites runs inside a context. `AGENTS.md` favours simplicity and +surgical changes; (b) is ~40 lines and 10 one-line wrappers, (a) is a signature +change across ten modules and their test doubles. + +### 6. Ledger retention + +30 days, pruned by a sweep modelled on `logsRetention.ts`. Long enough for +week-over-week comparison, short enough that the table never becomes a +consideration. + +## Design + +### A. `shared/schema.ts` + +```ts +export const agentWorkKindEnum = z.enum([ + "babysit_pr", + "evaluate_feedback", + "work_issue", + "decompose_issue", + "verify_issue", + "heal_ci", + "heal_deployment", + "answer_pr_question", + "release_notes", + "social_post", + "probe", + "unattributed", +]); + +export const agentInvocationOutcomeEnum = z.enum([ + "running", "completed", "failed", "timeout", "refused", +]); + +export const agentInvocationSchema = z.object({ + id: z.string(), + workKind: agentWorkKindEnum, + agent: codingAgentSchema, + model: z.string().nullable(), + repo: z.string().nullable(), + targetId: z.string().nullable(), + agentRunId: z.string().nullable(), + startedAt: z.string(), + finishedAt: z.string().nullable(), + durationMs: z.number().int().nonnegative().nullable(), + exitCode: z.number().int().nullable(), + outcome: agentInvocationOutcomeEnum, + error: z.string().nullable(), +}); +``` + +Config gains one field, next to `maxAgentRetryAttempts` (`shared/schema.ts:733`): + +```ts +maxAgentInvocationsPerHour: z.number().int().nonnegative().default(0), // 0 = unlimited +``` + +### B. `sqliteStorage.ts` + +Additive `CREATE TABLE IF NOT EXISTS` in the same block as the others +(`sqliteStorage.ts:683`). **No foreign key on any target id** — deliberate, so +the ledger survives `prs` cascade deletion (criterion 2). + +```sql +CREATE TABLE IF NOT EXISTS agent_invocations ( + id TEXT PRIMARY KEY, + work_kind TEXT NOT NULL, + agent TEXT NOT NULL, + model TEXT, + repo TEXT, + target_id TEXT, + agent_run_id TEXT, + started_at TEXT NOT NULL, + finished_at TEXT, + duration_ms INTEGER, + exit_code INTEGER, + outcome TEXT NOT NULL, + error TEXT +); +CREATE INDEX IF NOT EXISTS idx_agent_invocations_started_at + ON agent_invocations(started_at); +CREATE INDEX IF NOT EXISTS idx_agent_invocations_kind_started_at + ON agent_invocations(work_kind, started_at); +``` + +`IStorage` (`storage.ts:232`) gains, mirroring the `agentRun` trio: + +```ts +recordAgentInvocationStart(row: AgentInvocation): Promise; +recordAgentInvocationEnd(id: string, end: { + finishedAt: string; durationMs: number; exitCode: number | null; + outcome: AgentInvocationOutcome; error: string | null; +}): Promise; +countAgentInvocationsSince(since: string, opts?: { excludeKinds?: AgentWorkKind[] }): Promise; +summarizeAgentInvocations(since: string): Promise; +pruneAgentInvocationsBefore(cutoff: string): Promise; +``` + +Implemented in both `sqliteStorage.ts` and `memoryStorage.ts` (`memoryStorage.ts:75` +already holds `agentRuns` in a `Map`; same pattern). + +### C. `server/agentSpend.ts` (new, the only new module) + +```ts +export type AgentWorkContext = { + kind: AgentWorkKind; + repo?: string | null; + targetId?: string | null; + agentRunId?: string | null; +}; + +export class AgentBudgetExhaustedError extends Error {} + +/** Wrap a unit of work so every agent spawn inside it is attributed. */ +export function withAgentWork(ctx: AgentWorkContext, fn: () => Promise): Promise; + +/** Ambient context, or null when an invocation escaped attribution. */ +export function currentAgentWork(): AgentWorkContext | null; + +/** Rolling-hour usage. */ +export function spendWindowStart(now: Date): string; +export async function readSpend(storage: IStorage, now: Date): Promise; + +/** Throws AgentBudgetExhaustedError when the ceiling is reached. 0 = unlimited. */ +export async function assertBudgetAvailable(storage: IStorage, now: Date): Promise; + +/** Installed once at boot; agentRunner calls into it. */ +export function installAgentSpendMeter(storage: IStorage, now: () => Date): void; +``` + +The meter is *installed* rather than imported directly by `agentRunner.ts` so the +agent runner keeps no storage dependency and its existing tests keep working with +no meter installed. + +### D. `server/agentRunner.ts` — the only hot-path change + +```ts +export async function runAgentCommand( + agent: CodingAgent, + args: string[], + options?: Parameters[2], +): Promise { + const meter = getInstalledMeter(); + if (!meter) { + return runCommand((await resolveCommandPath(agent)) ?? agent, args, options); + } + return meter.meter(agent, () => + runCommand((await resolveCommandPath(agent)) ?? agent, args, options)); +} +``` + +`meter.meter` does: read ambient context (warn + `unattributed` if absent) → +`assertBudgetAvailable` unless the kind is `probe` → write the `running` row → +run → write the terminal row in a `finally`, classifying `timeout` from +`CommandResult` and `failed` from a non-zero exit. + +A crash between start and end leaves a `running` row. The boot-time sweep that +already reconciles interrupted `agent_runs` (`babysitter.ts:2290-2319`) gets a +sibling that closes orphaned invocations as `failed`. Counting a `running` row +toward the ceiling is correct in the meantime — an in-flight agent is spend. + +### E. Gate 1 — dispatcher (cheap, correct, no failures) + +`resolveClaimableKinds` (`backgroundJobDispatcher.ts:198`) already filters kinds +by capacity for `maxConcurrentBabysitRuns`. Extend it: when the ceiling is +reached, drop every member of `AGENT_INVOKING_JOB_KINDS` +(`failureRecovery.ts:22`). Jobs stay `queued`, exactly as under drain mode. + +### F. Gate 2 — invocation (catches what the dispatcher cannot see) + +`assertBudgetAvailable` throws `AgentBudgetExhaustedError` inside +`runAgentCommand`. This covers CI healing started mid-`babysit_pr`, the +fallback-agent re-run at `babysitter.ts:4211`, and conflict repair. + +`classifyFailure` (`failureRecovery.ts`) classifies it **`transient`**, joining +the existing `/\bbudget is in the reserve band\b/i` pattern that already handles +the GitHub-budget analogue. Consequence: the job backs off and retries on the +free/generous cap rather than burning a paid `maxAgentRetryAttempts` slot — the +agent never ran, so it was not a paid attempt. + +### G. Surfacing + +- `GET /api/agent-spend` → `{ windowMs, max, used, remaining, resetsAt, byKind[], recent[] }`. +- Header pill beside the drain toggle, shown only when a ceiling is set: + `18/30 agent runs this hour`. Amber at 80%, red plus "paused until HH:MM" at + the ceiling. Reuses the drain-toggle affordance rather than adding a new one. +- Settings numeric field directly beneath **Max agent retry attempts** + (`settings.tsx:1356`), with `0 = unlimited` help text. +- Per-PR and per-issue detail: invocation count for that target, from + `targetId`. This is the first time either page can answer "how many agent runs + has this thing cost me". +- MCP tool `get_agent_spend` + a `LOCAL_API.md` section. + +### H. Retention + +`pruneAgentInvocationsBefore(now - 30d)` on the existing logs-retention sweep +schedule (`logsRetention.ts`). + +## Work Breakdown + +| Step | Change | Verify | +|------|--------|--------| +| 1 | `shared/schema.ts`: `agentInvocationSchema`, enums, `maxAgentInvocationsPerHour`; `defaultConfig.ts` | `defaultConfig.test.ts` — new key present, default 0 | +| 2 | `sqliteStorage.ts` + `memoryStorage.ts` ledger methods | `storage.test.ts` — round-trip; **row survives PR delete**; `countAgentInvocationsSince` respects the window boundary | +| 3 | `server/agentSpend.ts` + `agentSpend.test.ts` | ALS context nests correctly; `assertBudgetAvailable` no-ops at 0; throws at the ceiling; `probe` excluded | +| 4 | `agentRunner.ts` meter hook | `agentRunner.test.ts` — no meter installed ⇒ byte-identical behaviour; installed ⇒ one row per spawn, terminal row written on throw and on timeout | +| 5 | Wrap all ten call sites in `withAgentWork` | A test enumerating the call-site modules asserts none invokes unattributed | +| 6 | Dispatcher gate in `resolveClaimableKinds` | `backgroundJobDispatcher.test.ts` — at ceiling, agent kinds unclaimable and `sync_watched_repos` still claimed; resumes as the window rolls | +| 7 | `AgentBudgetExhaustedError` ⇒ `transient` | `failureRecovery.test.ts` — classified transient, takes the free cap, does not park | +| 8 | Orphaned-`running` reconciliation at boot | `appRuntime.test.ts` — orphan closed as `failed`, counted while open | +| 9 | `/api/agent-spend` + MCP tool + `LOCAL_API.md` | `routes.test.ts` — shape, and `max: 0` reports unlimited | +| 10 | Header pill, Settings field, PR/issue counts | `client/src/lib` test + screenshot | +| 11 | 30-day prune on the retention sweep | `logsRetention.test.ts` | +| 12 | `npm run check`, `npx eslint .`, `npm run build`, `npm run test:all` | Green | + +Steps 1–5 are the metering half and are independently shippable: they produce +the data with zero behaviour change. Steps 6–8 are the enforcement half. If the +ceiling decisions above need more thought, land 1–5 first and get a release of +real numbers before choosing a default. + +## Outcome + +All twelve steps landed. Decisions 3, 4, and 5 were confirmed as recommended: +park-and-auto-resume, `0` (unlimited) default, `AsyncLocalStorage` attribution. + +Two things changed during implementation: + +- **`babysitPR` was split rather than wrapped in place.** The public method is + now a thin wrapper that establishes the `babysit_pr` context and delegates to + a private `babysitPRInternal`. Wrapping the ~900-line body inline would have + reindented the whole method for no behavioural gain. The wrapper does one + extra `getPR` to attribute the repo, which is nothing next to a babysit run. +- **`decompose_issue` and `verify_issue` are attributed by their caller.** + `issueDecompose.ts` and `issueVerify.ts` take no repo or issue identity, and + their callers in `backgroundJobHandlers.ts` already have both in scope. The + ALS context flows down, so no signature changed. The step-5 guard test knows + about this exemption explicitly rather than silently. + +The pre-commit quality pass pulled out one more thing: six call sites had each +open-coded the "which model will this agent use" lookup, and three of them used +a `claudeModel ?? codexModel` fallback that would label a codex run with the +Claude model. That is now `resolveAgentModel(agent, settings)` in +`agentSpend.ts`, with its own test. + +Verification: `npm run check`, `npx eslint .`, `npm run build`, and +`npm run test:all` (827 tests) are green. + +## Risks + +- **A gate that strands work.** If the dispatcher filter is wrong, agent-invoking + jobs never get claimed and automation silently dies. Mitigated by: `0` default + (gate is inert until opted into), auto-resume on window roll, the header pill + making the paused state visible, and step 6's test asserting non-agent kinds + still flow. +- **ALS context lost across an await boundary.** Node's `AsyncLocalStorage` + survives promise chains but is lost across a manual `setTimeout` bridge or an + emitter hop. If any call site turns out to lose it, that invocation records as + `unattributed` and logs at `warn` — it is still counted, just not attributed. + Step 5's test is what catches it. +- **Double-counting a fallback.** `babysitter.ts:4190` and `:4211` are the + primary agent and its fallback in one session. Both spawn a CLI, so both are + genuinely spend and both should count. Noting it because the number will look + higher than "runs" intuitively suggests — the pill is labelled *agent runs*, + not *PRs worked*. +- **Ledger write on the agent hot path.** One INSERT before a process that runs + for seconds to 90 minutes. Negligible, and better-sqlite3 is synchronous + anyway. +- **`running` rows inflating the count after a hard kill.** Deliberate — an + in-flight agent is spend — but a crash-loop could accumulate phantom spend + until the boot sweep runs. The sweep runs at boot, so the exposure is one + process lifetime. + +## Out of scope + +- Token or dollar accounting. Neither CLI reports it in a shape this app parses. + `duration_ms` and `model` are recorded so it can be estimated later. +- Per-repo spend ceilings. `watchedRepoSchema` already carries per-repo + overrides if it is wanted; the global ceiling is the safety net. +- A duration-based ceiling ("max agent minutes per hour"). The column lands in + this work; the knob does not. +- Repurposing or migrating `agent_runs`. It stays a babysit-session record. +- Cost attribution across the fallback agent (which of the two produced the + merged commit). That is the outcomes-dashboard question, not this one. diff --git a/docs/public/configuration.md b/docs/public/configuration.md index ea083a8..3fb3cf3 100644 --- a/docs/public/configuration.md +++ b/docs/public/configuration.md @@ -163,6 +163,18 @@ patchdeck paces autonomous PR babysitter dispatch using a global in-flight cap f This value maps to `maxConcurrentBabysitRuns` in `GET /api/config` and `PATCH /api/config`. +## Agent Spend Settings + +Every spawn of the `codex` or `claude` CLI is recorded in a local ledger, whichever path caused it: PR work, feedback evaluation, issue work, issue decompose and verify, CI healing, deployment healing, PR questions, release notes, and social posts. Agent health-check probes are recorded too, but never counted against the ceiling. + +| Setting | Default | Description | +|---------|---------|-------------| +| `Max agent runs per hour` | `0` | Rolling-hour ceiling on coding-agent runs across every path. `0` means unlimited | + +The window rolls continuously rather than resetting at the top of the hour. When the ceiling is reached, the dispatcher stops claiming agent-invoking jobs and any agent run started inside an already-running job is refused. Queued work is not failed: it waits, and resumes on its own as the oldest run ages out of the window. The header shows current usage once a ceiling is set. + +This value maps to `maxAgentInvocationsPerHour` in `GET /api/config` and `PATCH /api/config`. `GET /api/agent-spend` reports usage, the ceiling, when the next slot frees up, and a breakdown by work kind. + ## Deployment Healing Settings patchdeck can monitor merged PRs for failed Vercel or Railway deployments and open a follow-up fix PR when the deployment breaks after merge. diff --git a/server/agentRunner.ts b/server/agentRunner.ts index 66a0522..9ef9a03 100644 --- a/server/agentRunner.ts +++ b/server/agentRunner.ts @@ -3,6 +3,7 @@ import { constants as fsConstants } from "fs"; import { homedir, tmpdir } from "os"; import path from "path"; import { spawn } from "child_process"; +import { getInstalledAgentSpendMeter, withAgentWork } from "./agentSpend"; export type CodingAgent = "codex" | "claude"; export type CodexReasoningEffort = "default" | "low" | "medium" | "high" | "xhigh"; @@ -129,12 +130,21 @@ export async function resolveCommandPath(command: string): Promise[2], ): Promise { - return runCommand((await resolveCommandPath(agent)) ?? agent, args, options); + const spawnAgent = async () => + runCommand((await resolveCommandPath(agent)) ?? agent, args, options); + + const meter = getInstalledAgentSpendMeter(); + return meter ? meter.meter(agent, spawnAgent) : spawnAgent(); } export async function resolveAgent( @@ -171,28 +181,32 @@ export async function checkAgentHealth(agent: CodingAgent): Promise ( + agent === "codex" + ? runAgentCommand( + "codex", + [ + "exec", + "--skip-git-repo-check", + "--sandbox", + "read-only", + prompt, + ], + { timeoutMs: 30000 }, + ) + : runAgentCommand( + "claude", + [ + "-p", + "--output-format", + "text", + prompt, + ], + { timeoutMs: 30000 }, + ) + )); if (result.code !== 0) { const detail = summarizeHealthFailure(result); diff --git a/server/agentSpend.test.ts b/server/agentSpend.test.ts new file mode 100644 index 0000000..349c4bf --- /dev/null +++ b/server/agentSpend.test.ts @@ -0,0 +1,277 @@ +// PatchDeck + Agent spend ledger, attribution, and ceiling tests +// Copyright (c) 2026 Jeremy McSpadden + +import assert from "node:assert/strict"; +import test from "node:test"; +import { MemStorage } from "./memoryStorage"; +import { + AGENT_SPEND_WINDOW_MS, + AgentBudgetExhaustedError, + assertAgentBudgetAvailable, + currentAgentWork, + installAgentSpendMeter, + isAgentBudgetExhausted, + isAgentBudgetExhaustedError, + readAgentSpend, + resolveAgentModel, + uninstallAgentSpendMeter, + withAgentWork, +} from "./agentSpend"; +import type { IStorage } from "./storage"; + +const NOW = new Date("2026-08-29T12:00:00.000Z"); + +async function storageWithCeiling(max: number): Promise { + const storage = new MemStorage(); + const config = await storage.getConfig(); + await storage.updateConfig({ ...config, maxAgentInvocationsPerHour: max }); + return storage; +} + +function ok() { + return Promise.resolve({ stdout: "done", stderr: "", code: 0 }); +} + +test("context nests, innermost wins, and unwinds cleanly", async () => { + assert.equal(currentAgentWork(), null); + + await withAgentWork({ kind: "babysit_pr", targetId: "pr-1" }, async () => { + assert.equal(currentAgentWork()?.kind, "babysit_pr"); + + await withAgentWork({ kind: "heal_ci", targetId: "pr-1" }, async () => { + assert.equal(currentAgentWork()?.kind, "heal_ci"); + }); + + assert.equal(currentAgentWork()?.kind, "babysit_pr"); + }); + + assert.equal(currentAgentWork(), null); +}); + +test("context survives await boundaries inside the wrapped work", async () => { + await withAgentWork({ kind: "work_issue", repo: "acme/app" }, async () => { + await new Promise((resolve) => setTimeout(resolve, 1)); + assert.equal(currentAgentWork()?.repo, "acme/app"); + }); +}); + +test("a ceiling of 0 means unlimited", async () => { + const storage = await storageWithCeiling(0); + const meter = installAgentSpendMeter(storage, () => NOW); + + try { + for (let index = 0; index < 25; index += 1) { + await withAgentWork({ kind: "babysit_pr" }, () => meter.meter("claude", ok)); + } + + assert.equal(await isAgentBudgetExhausted(storage, NOW), false); + await assertAgentBudgetAvailable(storage, NOW); + + const spend = await readAgentSpend(storage, NOW); + assert.equal(spend.max, 0); + assert.equal(spend.used, 25); + assert.equal(spend.remaining, null); + assert.equal(spend.exhausted, false); + } finally { + uninstallAgentSpendMeter(); + } +}); + +test("records one ledger row per spawn, with duration and outcome", async () => { + const storage = await storageWithCeiling(0); + let clock = NOW.getTime(); + const meter = installAgentSpendMeter(storage, () => new Date(clock)); + + try { + await withAgentWork({ kind: "work_issue", repo: "acme/app", targetId: "acme/app#7" }, () => + meter.meter("claude", async () => { + clock += 4000; + return { stdout: "", stderr: "", code: 0 }; + })); + + const rows = await storage.listAgentInvocationsSince(new Date(NOW.getTime() - 1000).toISOString()); + assert.equal(rows.length, 1); + assert.equal(rows[0].workKind, "work_issue"); + assert.equal(rows[0].agent, "claude"); + assert.equal(rows[0].repo, "acme/app"); + assert.equal(rows[0].targetId, "acme/app#7"); + assert.equal(rows[0].outcome, "completed"); + assert.equal(rows[0].durationMs, 4000); + assert.equal(rows[0].exitCode, 0); + } finally { + uninstallAgentSpendMeter(); + } +}); + +test("classifies non-zero exits, timeouts, and thrown errors", async () => { + const storage = await storageWithCeiling(0); + const meter = installAgentSpendMeter(storage, () => NOW); + + try { + await withAgentWork({ kind: "babysit_pr" }, () => + meter.meter("codex", async () => ({ stdout: "", stderr: "boom", code: 1 }))); + await withAgentWork({ kind: "babysit_pr" }, () => + meter.meter("codex", async () => ({ stdout: "", stderr: "", code: 124, timedOut: true }))); + await assert.rejects( + withAgentWork({ kind: "babysit_pr" }, () => + meter.meter("codex", async () => { + throw new Error("spawn failed"); + })), + /spawn failed/, + ); + + const rows = await storage.listAgentInvocationsSince(new Date(NOW.getTime() - 1000).toISOString()); + const outcomes = rows.map((row) => row.outcome).sort(); + assert.deepEqual(outcomes, ["failed", "failed", "timeout"]); + // Every row is terminal: a thrown spawn still closes its ledger entry. + assert.equal(rows.filter((row) => row.finishedAt === null).length, 0); + } finally { + uninstallAgentSpendMeter(); + } +}); + +test("refuses a spawn once the ceiling is reached and reports when it clears", async () => { + const storage = await storageWithCeiling(2); + const meter = installAgentSpendMeter(storage, () => NOW); + + try { + await withAgentWork({ kind: "babysit_pr" }, () => meter.meter("claude", ok)); + await withAgentWork({ kind: "work_issue" }, () => meter.meter("claude", ok)); + + assert.equal(await isAgentBudgetExhausted(storage, NOW), true); + + let refused: unknown; + try { + await withAgentWork({ kind: "heal_ci" }, () => meter.meter("claude", ok)); + } catch (error) { + refused = error; + } + + assert.ok(refused instanceof AgentBudgetExhaustedError); + assert.equal(isAgentBudgetExhaustedError(refused), true); + + // The refused spawn never ran, so it is not itself recorded as spend. + const spend = await readAgentSpend(storage, NOW); + assert.equal(spend.used, 2); + assert.equal(spend.remaining, 0); + assert.equal(spend.exhausted, true); + + // Rolling window: the ceiling clears as the oldest invocation ages out. + const later = new Date(NOW.getTime() + AGENT_SPEND_WINDOW_MS + 1000); + assert.equal(await isAgentBudgetExhausted(storage, later), false); + } finally { + uninstallAgentSpendMeter(); + } +}); + +test("health probes are recorded but never counted against the ceiling", async () => { + const storage = await storageWithCeiling(1); + const meter = installAgentSpendMeter(storage, () => NOW); + + try { + for (let index = 0; index < 5; index += 1) { + await withAgentWork({ kind: "probe" }, () => meter.meter("claude", ok)); + } + + assert.equal(await isAgentBudgetExhausted(storage, NOW), false); + + const spend = await readAgentSpend(storage, NOW); + assert.equal(spend.used, 0); + assert.equal(spend.byKind.find((entry) => entry.workKind === "probe")?.count, 5); + + // The one metered run still fits, and the next one does not. + await withAgentWork({ kind: "babysit_pr" }, () => meter.meter("claude", ok)); + await assert.rejects( + withAgentWork({ kind: "babysit_pr" }, () => meter.meter("claude", ok)), + AgentBudgetExhaustedError, + ); + } finally { + uninstallAgentSpendMeter(); + } +}); + +test("an unattributed spawn is still counted", async () => { + const storage = await storageWithCeiling(0); + const meter = installAgentSpendMeter(storage, () => NOW); + + try { + await meter.meter("claude", ok); + + const rows = await storage.listAgentInvocationsSince(new Date(NOW.getTime() - 1000).toISOString()); + assert.equal(rows.length, 1); + assert.equal(rows[0].workKind, "unattributed"); + } finally { + uninstallAgentSpendMeter(); + } +}); + +test("invocations outside the rolling window do not count", async () => { + const storage = await storageWithCeiling(1); + const stale = new Date(NOW.getTime() - AGENT_SPEND_WINDOW_MS - 60_000); + const meter = installAgentSpendMeter(storage, () => stale); + + try { + await withAgentWork({ kind: "babysit_pr" }, () => meter.meter("claude", ok)); + + assert.equal(await isAgentBudgetExhausted(storage, NOW), false); + const spend = await readAgentSpend(storage, NOW); + assert.equal(spend.used, 0); + } finally { + uninstallAgentSpendMeter(); + } +}); + +// Guard: the ALS context only works if every module that reaches an agent +// primitive establishes one. A new call site added without a context would +// silently record as `unattributed`, so the set is pinned here. +test("every module that invokes a coding agent establishes a work context", async () => { + const { readdir, readFile } = await import("node:fs/promises"); + const path = await import("node:path"); + + const AGENT_PRIMITIVES = [ + "runAgentCommand(", + "applyFixesWithAgent(", + "runAgentOneShot(", + "evaluateFixNecessityWithAgent(", + ]; + + // Modules that reach a primitive but are attributed by their caller instead, + // because the caller is where the repo and target are known. + const ATTRIBUTED_BY_CALLER = new Set(["issueDecompose.ts", "issueVerify.ts"]); + + const serverDir = path.join(import.meta.dirname, "."); + const files = (await readdir(serverDir)) + .filter((name) => name.endsWith(".ts") && !name.endsWith(".test.ts")) + .filter((name) => name !== "agentSpend.ts"); + + const missing: string[] = []; + for (const name of files) { + const source = await readFile(path.join(serverDir, name), "utf8"); + const callsAgent = AGENT_PRIMITIVES.some((primitive) => source.includes(primitive)); + if (!callsAgent || ATTRIBUTED_BY_CALLER.has(name)) { + continue; + } + + if (!source.includes("withAgentWork")) { + missing.push(name); + } + } + + assert.deepEqual( + missing, + [], + `these modules invoke a coding agent without establishing a spend context: ${missing.join(", ")}`, + ); +}); + +test("resolveAgentModel picks the model for the agent that will actually run", () => { + const settings = { claudeModel: "opus", codexModel: "gpt-5" }; + + assert.equal(resolveAgentModel("claude", settings), "opus"); + assert.equal(resolveAgentModel("codex", settings), "gpt-5"); + // An unset codexModel is the shipped default and must not fall through to + // the Claude model just because one is configured. + assert.equal(resolveAgentModel("codex", { claudeModel: "opus", codexModel: "" }), null); + assert.equal(resolveAgentModel("claude", undefined), null); + assert.equal(resolveAgentModel("claude", null), null); +}); diff --git a/server/agentSpend.ts b/server/agentSpend.ts new file mode 100644 index 0000000..7ab76db --- /dev/null +++ b/server/agentSpend.ts @@ -0,0 +1,282 @@ +// PatchDeck + Coding-agent spend ledger, attribution, and rolling-hour ceiling +// Copyright (c) 2026 Jeremy McSpadden + +import { AsyncLocalStorage } from "async_hooks"; +import { randomUUID } from "crypto"; +import type { + AgentInvocation, + AgentInvocationOutcome, + AgentSpendSummary, + AgentWorkKind, +} from "@shared/schema"; +import { childLogger } from "./logger"; +import type { IStorage } from "./storage"; + +const log = childLogger("agentSpend"); + +/** Rolling window the ceiling is measured over. */ +export const AGENT_SPEND_WINDOW_MS = 3_600_000; + +/** How long ledger rows are kept before the retention sweep drops them. */ +export const AGENT_SPEND_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; + +/** + * Work kinds that do not count against the ceiling. `probe` is the fixed + * one-token health check used by onboarding and diagnostics: opening Settings + * must never be able to exhaust a budget. + */ +export const UNMETERED_WORK_KINDS: AgentWorkKind[] = ["probe"]; + +export type AgentWorkContext = { + kind: AgentWorkKind; + repo?: string | null; + targetId?: string | null; + agentRunId?: string | null; + model?: string | null; +}; + +export class AgentBudgetExhaustedError extends Error { + readonly used: number; + readonly max: number; + readonly resetsAt: string; + + constructor(params: { used: number; max: number; resetsAt: string }) { + super( + `Agent budget exhausted: ${params.used}/${params.max} agent runs in the last hour. ` + + `Work resumes as the window rolls forward (next slot around ${params.resetsAt}).`, + ); + this.name = "AgentBudgetExhaustedError"; + this.used = params.used; + this.max = params.max; + this.resetsAt = params.resetsAt; + } +} + +export function isAgentBudgetExhaustedError(error: unknown): boolean { + return error instanceof AgentBudgetExhaustedError; +} + +const workContext = new AsyncLocalStorage(); + +/** + * Wrap a unit of work so every coding-agent spawn inside it is attributed to + * it. Nesting is allowed and the innermost context wins: CI healing runs inside + * a babysit session and should be billed as `heal_ci`. + */ +export function withAgentWork(context: AgentWorkContext, fn: () => Promise): Promise { + return workContext.run(context, fn); +} + +export function currentAgentWork(): AgentWorkContext | null { + return workContext.getStore() ?? null; +} + +/** Start of the rolling window, as an ISO timestamp comparable to `started_at`. */ +export function spendWindowStart(now: Date): string { + return new Date(now.getTime() - AGENT_SPEND_WINDOW_MS).toISOString(); +} + +/** The configured ceiling, normalised. 0 means unlimited. */ +export async function resolveAgentCeiling(storage: IStorage): Promise { + const config = await storage.getConfig(); + return Math.max(0, Math.floor(config.maxAgentInvocationsPerHour)); +} + +/** Invocations inside the rolling window that count against the ceiling. */ +export async function countMeteredAgentInvocations(storage: IStorage, now: Date): Promise { + return storage.countAgentInvocationsSince(spendWindowStart(now), { + excludeKinds: UNMETERED_WORK_KINDS, + }); +} + +/** + * Pick the model the agent will actually run with. The ledger records the + * resolved agent separately, so this is best-effort context for the row. + */ +export function resolveAgentModel( + agent: AgentInvocation["agent"], + settings?: { claudeModel?: string | null; codexModel?: string | null } | null, +): string | null { + const model = agent === "claude" ? settings?.claudeModel : settings?.codexModel; + return model && model.trim().length > 0 ? model : null; +} + +export async function readAgentSpend(storage: IStorage, now: Date): Promise { + const max = await resolveAgentCeiling(storage); + const windowStartedAt = spendWindowStart(now); + const invocations = await storage.listAgentInvocationsSince(windowStartedAt, { limit: 5000 }); + const metered = invocations.filter((invocation) => !UNMETERED_WORK_KINDS.includes(invocation.workKind)); + + const byKind = new Map(); + for (const invocation of invocations) { + const entry = byKind.get(invocation.workKind) ?? { count: 0, totalDurationMs: 0 }; + entry.count += 1; + entry.totalDurationMs += invocation.durationMs ?? 0; + byKind.set(invocation.workKind, entry); + } + + const used = metered.length; + + return { + windowMs: AGENT_SPEND_WINDOW_MS, + windowStartedAt, + resetsAt: nextSlotAt(metered, max, now), + max, + used, + remaining: max === 0 ? null : Math.max(0, max - used), + exhausted: max > 0 && used >= max, + byKind: Array.from(byKind.entries()) + .map(([workKind, entry]) => ({ workKind, ...entry })) + .sort((a, b) => b.count - a.count), + }; +} + +/** + * When the next slot frees up. With a rolling window that is the moment the + * oldest invocation still inside the window ages out, not the top of the hour. + */ +function nextSlotAt(metered: AgentInvocation[], max: number, now: Date): string { + if (max === 0 || metered.length < max) { + return now.toISOString(); + } + + // `metered` arrives newest-first, so the row that has to age out to free a + // slot is the one sitting on the ceiling boundary. + const boundary = metered[max - 1]; + return new Date(new Date(boundary.startedAt).getTime() + AGENT_SPEND_WINDOW_MS).toISOString(); +} + +/** + * True when a new agent spawn would exceed the ceiling. `max: 0` is unlimited. + * Pass `ceiling` when the caller has already read config, so a hot poll loop + * does not read it twice. + */ +export async function isAgentBudgetExhausted( + storage: IStorage, + now: Date, + ceiling?: number, +): Promise { + const max = ceiling ?? await resolveAgentCeiling(storage); + if (max === 0) { + return false; + } + + return (await countMeteredAgentInvocations(storage, now)) >= max; +} + +/** Throws {@link AgentBudgetExhaustedError} when the ceiling has been reached. */ +export async function assertAgentBudgetAvailable(storage: IStorage, now: Date): Promise { + const max = await resolveAgentCeiling(storage); + if (max === 0) { + return; + } + + const used = await countMeteredAgentInvocations(storage, now); + if (used < max) { + return; + } + + const summary = await readAgentSpend(storage, now); + throw new AgentBudgetExhaustedError({ used, max, resetsAt: summary.resetsAt }); +} + +export type AgentSpendMeter = { + /** + * Gate, record, and time a single coding-agent process spawn. `run` is the + * spawn itself; everything else is ledger bookkeeping. + */ + meter( + agent: AgentInvocation["agent"], + run: () => Promise, + ): Promise; +}; + +let installedMeter: AgentSpendMeter | null = null; + +export function getInstalledAgentSpendMeter(): AgentSpendMeter | null { + return installedMeter; +} + +/** Test seam. */ +export function uninstallAgentSpendMeter(): void { + installedMeter = null; +} + +/** + * Install the meter for the process. Done once at boot so `agentRunner` keeps + * no storage dependency and its existing tests run unmetered. + */ +export function installAgentSpendMeter(storage: IStorage, now: () => Date = () => new Date()): AgentSpendMeter { + const meter: AgentSpendMeter = { + async meter(agent, run) { + const context = currentAgentWork(); + if (!context) { + log.warn( + { agent }, + "Coding agent invoked outside an attributed work context; recording as unattributed", + ); + } + + const workKind: AgentWorkKind = context?.kind ?? "unattributed"; + const metered = !UNMETERED_WORK_KINDS.includes(workKind); + + if (metered) { + await assertAgentBudgetAvailable(storage, now()); + } + + const startedAtDate = now(); + const invocation: AgentInvocation = { + id: randomUUID(), + workKind, + agent, + model: context?.model ?? null, + repo: context?.repo ?? null, + targetId: context?.targetId ?? null, + agentRunId: context?.agentRunId ?? null, + startedAt: startedAtDate.toISOString(), + finishedAt: null, + durationMs: null, + exitCode: null, + outcome: "running", + error: null, + }; + + await storage.recordAgentInvocationStart(invocation); + + const finish = async (end: { + outcome: AgentInvocationOutcome; + exitCode: number | null; + error: string | null; + }) => { + const finishedAtDate = now(); + await storage.recordAgentInvocationEnd(invocation.id, { + finishedAt: finishedAtDate.toISOString(), + durationMs: Math.max(0, finishedAtDate.getTime() - startedAtDate.getTime()), + exitCode: end.exitCode, + outcome: end.outcome, + error: end.error, + }); + }; + + try { + const result = await run(); + await finish({ + outcome: result.timedOut ? "timeout" : result.code === 0 ? "completed" : "failed", + exitCode: result.code, + error: null, + }); + return result; + } catch (error) { + await finish({ + outcome: "failed", + exitCode: null, + error: error instanceof Error ? error.message : String(error), + }); + throw error; + } + }, + }; + + installedMeter = meter; + return meter; +} diff --git a/server/appRuntime.ts b/server/appRuntime.ts index 0f8baf4..c76f9ca 100644 --- a/server/appRuntime.ts +++ b/server/appRuntime.ts @@ -2,6 +2,7 @@ import { EventEmitter } from "node:events"; import type { ActivityItem, ActivitySnapshot, + AgentSpendSummary, BackgroundJob, Config, CurrentRunStatus, @@ -49,6 +50,7 @@ import { BackgroundJobDispatcher } from "./backgroundJobDispatcher"; import { BackgroundJobQueue, buildBackgroundJobDedupeKey } from "./backgroundJobQueue"; import { buildActivityPayload, readActivityPayload } from "./activityPayload"; import { createWatcherScheduler, type WatcherScheduler } from "./watcherScheduler"; +import { installAgentSpendMeter, readAgentSpend, uninstallAgentSpendMeter } from "./agentSpend"; import { startLogsRetentionJob, type RetentionJobHandle } from "./logsRetention"; import { getRateLimitState } from "./rateLimitState"; import { runWithRequestPriority } from "./requestPriority"; @@ -143,6 +145,7 @@ export type AppRuntime = { stop(): void; subscribe(listener: () => void): () => void; getRuntimeSnapshot(): Promise; + getAgentSpend(): Promise; getGitHubAuthStatus(): ReturnType; setDrainMode(input: DrainModeParams): Promise; listActivities(): Promise; @@ -2339,6 +2342,15 @@ export function createAppRuntime(dependencies: AppRuntimeDependencies = {}): App started = true; + // Every coding-agent spawn in this process is now recorded and gated. + installAgentSpendMeter(storage); + // An invocation left `running` by a hard shutdown would otherwise count + // against the ceiling forever. + const orphanedInvocations = await storage.closeOrphanedAgentInvocations(new Date().toISOString()); + if (orphanedInvocations > 0) { + log.warn({ orphanedInvocations }, "Closed agent invocations interrupted by the last shutdown"); + } + if (startBackgroundServices) { await backgroundJobDispatcher.start(); logsRetentionJob = startLogsRetentionJob(storage); @@ -2359,6 +2371,7 @@ export function createAppRuntime(dependencies: AppRuntimeDependencies = {}): App stop() { started = false; + uninstallAgentSpendMeter(); backgroundJobDispatcher.stop(); if (logsRetentionJob) { logsRetentionJob.stop(); @@ -2382,6 +2395,7 @@ export function createAppRuntime(dependencies: AppRuntimeDependencies = {}): App }, getRuntimeSnapshot, + getAgentSpend: () => readAgentSpend(storage, new Date()), async getGitHubAuthStatus() { const config = await storage.getConfig(); diff --git a/server/babysitter.ts b/server/babysitter.ts index c67a9c6..9d945cf 100644 --- a/server/babysitter.ts +++ b/server/babysitter.ts @@ -52,6 +52,7 @@ import { type ParsedRepoSlug, type StatusReplyRef, } from "./github"; +import { resolveAgentModel, withAgentWork } from "./agentSpend"; import { CIHealingManager, isTerminalHealingState } from "./ciHealingManager"; import { classifyCIFailures, type ClassifiedCIFailure } from "./ciFailureClassifier"; import { isFailingCheckSnapshot } from "./ciCheckIngestor"; @@ -3213,7 +3214,28 @@ export class PRBabysitter { return { status: "timeout", failures: [] }; } + /** + * Public entry point. Establishes the agent-spend work context so every + * coding-agent spawn inside the run — evaluate, apply, code-owner fallback, + * conflict repair, CI healing — is attributed and metered. Nested contexts + * win, so CI healing bills as `heal_ci` rather than `babysit_pr`. + */ async babysitPR( + prId: string, + preferredAgent: CodingAgent, + options?: Parameters[2], + ): Promise { + const pr = await this.storage.getPR(prId); + return withAgentWork({ + kind: "babysit_pr", + repo: pr?.repo ?? null, + targetId: prId, + agentRunId: options?.runId ?? null, + model: resolveAgentModel(preferredAgent, options?.agentSettings), + }, () => this.babysitPRInternal(prId, preferredAgent, options)); + } + + private async babysitPRInternal( prId: string, preferredAgent: CodingAgent, options?: { @@ -3908,7 +3930,12 @@ export class PRBabysitter { cwd: string; prompt: string; phase: string; - }) => { + }) => withAgentWork({ + kind: "evaluate_feedback", + repo: pr.repo, + targetId: pr.id, + agentRunId: runId, + }, async () => { try { return await this.runtime.evaluateFixNecessityWithAgent({ agent, @@ -3927,7 +3954,7 @@ export class PRBabysitter { } throw error; } - }; + }); const parsedPr: ParsedPRUrl = { owner: parsedRepo.owner, diff --git a/server/backgroundJobDispatcher.test.ts b/server/backgroundJobDispatcher.test.ts index c848dd2..1089eab 100644 --- a/server/backgroundJobDispatcher.test.ts +++ b/server/backgroundJobDispatcher.test.ts @@ -662,3 +662,66 @@ test("BackgroundJobDispatcher does not apply the agent spend cap to api-only wor dispatcher.stop(); } }); + +test("agent-invoking kinds stop being claimed once the agent ceiling is reached", async () => { + const storage = new MemStorage(); + const queue = new BackgroundJobQueue(storage); + const config = await storage.getConfig(); + await storage.updateConfig({ ...config, maxAgentInvocationsPerHour: 1 }); + + const startedAt = new Date().toISOString(); + await storage.recordAgentInvocationStart({ + id: "inv-1", + workKind: "babysit_pr", + agent: "claude", + model: null, + repo: "acme/app", + targetId: "pr-1", + agentRunId: null, + startedAt, + finishedAt: startedAt, + durationMs: 0, + exitCode: 0, + outcome: "completed", + error: null, + }); + + await queue.enqueue("babysit_pr", "pr-1", "babysit_pr:pr-1", { prId: "pr-1" }); + await queue.enqueue("sync_watched_repos", "all", "sync_watched_repos:all", {}); + + const handled: string[] = []; + const dispatcher = new BackgroundJobDispatcher({ + storage, + queue, + workerId: "dispatcher-budget", + pollIntervalMs: 5, + leaseMs: 30_000, + heartbeatIntervalMs: 10, + handlers: { + babysit_pr: async () => { + handled.push("babysit_pr"); + }, + sync_watched_repos: async () => { + handled.push("sync_watched_repos"); + }, + }, + }); + + await dispatcher.start(); + try { + // Free work still flows; paid work waits without failing. + await waitForCondition(() => handled.includes("sync_watched_repos")); + await new Promise((resolve) => setTimeout(resolve, 40)); + assert.equal(handled.includes("babysit_pr"), false); + + const parked = await storage.listBackgroundJobs({ kind: "babysit_pr" }); + assert.equal(parked[0]?.status, "queued"); + assert.equal(parked[0]?.attemptCount, 0); + + // Raising the ceiling releases the queued job on the next poll. + await storage.updateConfig({ ...config, maxAgentInvocationsPerHour: 5 }); + await waitForCondition(() => handled.includes("babysit_pr"), 1_000); + } finally { + dispatcher.stop(); + } +}); diff --git a/server/backgroundJobDispatcher.ts b/server/backgroundJobDispatcher.ts index 467465d..77b8f84 100644 --- a/server/backgroundJobDispatcher.ts +++ b/server/backgroundJobDispatcher.ts @@ -2,7 +2,8 @@ import { randomUUID } from "crypto"; import type { BackgroundJob, BackgroundJobKind } from "@shared/schema"; import type { IStorage } from "./storage"; import { BackgroundJobQueue } from "./backgroundJobQueue"; -import { classifyFailure, computeRetryDelayMs, resolveMaxAttempts, type FailureClass } from "./failureRecovery"; +import { isAgentBudgetExhausted } from "./agentSpend"; +import { classifyFailure, computeRetryDelayMs, resolveMaxAttempts, AGENT_INVOKING_JOB_KINDS, type FailureClass } from "./failureRecovery"; import { childLogger } from "./logger"; import { DEFAULT_CONFIG } from "./defaultConfig"; @@ -56,6 +57,7 @@ export class BackgroundJobDispatcher { private running = false; private polling = false; + private loggedBudgetPause = false; private pollTimer: NodeJS.Timeout | null = null; constructor(params: { @@ -198,21 +200,57 @@ export class BackgroundJobDispatcher { } private async resolveClaimableKinds(kinds: BackgroundJobKind[]): Promise { - if (!kinds.includes("babysit_pr")) { + // Nothing here can spend an agent, so neither gate applies and the poll + // stays free of a config read. + if (!kinds.some((kind) => AGENT_INVOKING_JOB_KINDS.has(kind))) { return kinds; } const config = await this.storage.getConfig(); + const withinBudget = await this.filterKindsWithinAgentBudget(kinds, config.maxAgentInvocationsPerHour); + if (!withinBudget.includes("babysit_pr")) { + return withinBudget; + } + const maxConcurrentBabysitRuns = Math.max(1, config.maxConcurrentBabysitRuns); const activeBabysitRuns = Array.from(this.activeJobs.keys()) .filter((jobId) => jobId.startsWith("babysit_pr:")) .length; if (activeBabysitRuns < maxConcurrentBabysitRuns) { + return withinBudget; + } + + return withinBudget.filter((kind) => kind !== "babysit_pr"); + } + + /** + * Stop claiming jobs that can spend a paid agent once the rolling-hour + * ceiling is reached. Jobs stay `queued` and flow again as the window rolls + * forward, exactly as under drain mode — nothing fails and nothing is lost. + */ + private async filterKindsWithinAgentBudget( + kinds: BackgroundJobKind[], + ceiling: number, + ): Promise { + if (!(await isAgentBudgetExhausted(this.storage, this.now(), ceiling))) { + if (this.loggedBudgetPause) { + this.loggedBudgetPause = false; + log.info("Agent invocation ceiling cleared; resuming agent-invoking job kinds"); + } return kinds; } - return kinds.filter((kind) => kind !== "babysit_pr"); + const remaining = kinds.filter((kind) => !AGENT_INVOKING_JOB_KINDS.has(kind)); + if (!this.loggedBudgetPause) { + this.loggedBudgetPause = true; + log.warn( + { stillClaimable: remaining }, + "Agent invocation ceiling reached; pausing agent-invoking job kinds until the window rolls", + ); + } + + return remaining; } private runJob(job: BackgroundJob): void { diff --git a/server/backgroundJobHandlers.ts b/server/backgroundJobHandlers.ts index e9721f8..ea64421 100644 --- a/server/backgroundJobHandlers.ts +++ b/server/backgroundJobHandlers.ts @@ -27,6 +27,7 @@ import { decomposeIssueBody, hashIssueBody } from "./issueDecompose"; import { verifySubtasksAgainstPr } from "./issueVerify"; import { runIssueWorkRepair } from "./issueWorkAgent"; import { answerPRQuestion, type PRQuestionRepositoryContext } from "./prQuestionAgent"; +import { withAgentWork } from "./agentSpend"; import { getRateLimitState } from "./rateLimitState"; import type { ReleaseManager } from "./releaseManager"; import { runWithRequestPriority } from "./requestPriority"; @@ -493,11 +494,15 @@ export function createBackgroundJobHandlers(params: { const agent = resolveRepoCodingAgent(config, repoSettings); const agentSettings = resolveRepoAgentRuntimeSettings(config, repoSettings); - const freshDecomposed = await decomposeIssueBody({ + const freshDecomposed = await withAgentWork({ + kind: "decompose_issue", + repo: issue.repoFullName, + targetId, + }, () => decomposeIssueBody({ body: issue.body, agent, settings: agentSettings, - }); + })); const existingSet = await storage.getIssueSubtasks(targetId); const subtasks = freshDecomposed.length >= 2 ? freshDecomposed @@ -510,14 +515,18 @@ export function createBackgroundJobHandlers(params: { status: "pending" as const, }]; - const result = await verifySubtasksAgainstPr({ + const result = await withAgentWork({ + kind: "verify_issue", + repo: issue.repoFullName, + targetId, + }, () => verifySubtasksAgainstPr({ issueTitle: issue.title, issueBody: issue.body, subtasks, prDiff: diff, agent, settings: agentSettings, - }); + })); await storage.upsertIssueSubtasks({ targetId, @@ -632,11 +641,15 @@ export function createBackgroundJobHandlers(params: { const existingSubtasks = await storage.getIssueSubtasks(targetId); let subtasks = existingSubtasks?.subtasks ?? []; if (!existingSubtasks || existingSubtasks.analyzedBodyHash !== bodyHash) { - subtasks = await decomposeIssueBody({ + subtasks = await withAgentWork({ + kind: "decompose_issue", + repo: issue.repoFullName, + targetId, + }, () => decomposeIssueBody({ body: issue.body, agent, settings: agentSettings, - }); + })); await storage.upsertIssueSubtasks({ targetId, repo: issue.repoFullName, diff --git a/server/ciHealingAgent.ts b/server/ciHealingAgent.ts index c51ede6..fcfc3b3 100644 --- a/server/ciHealingAgent.ts +++ b/server/ciHealingAgent.ts @@ -1,6 +1,7 @@ import { createHash } from "crypto"; import type { CodingAgent, CommandResult } from "./agentRunner"; import { applyFixesWithAgent, runCommand, summarizeCommandResult } from "./agentRunner"; +import { withAgentWork } from "./agentSpend"; import type { ClassifiedCIFailure } from "./ciFailureClassifier"; import { preparePrWorktree, removePrWorktree } from "./repoWorkspace"; @@ -329,12 +330,16 @@ export async function runCIHealingRepairAttempt(input: CIHealingWorktreeInput & }); try { - const agentResult = await deps.applyFixesWithAgent({ + const agentResult = await withAgentWork({ + kind: "heal_ci", + repo: input.repoFullName, + targetId: `${input.repoFullName}#${input.prNumber}`, + }, () => deps.applyFixesWithAgent({ agent: input.agent, cwd: worktree.worktreePath, prompt, env: input.env, - }); + })); if (agentResult.code === 0) { await commitWorktreeChanges(deps, worktree.worktreePath, input.prNumber); diff --git a/server/defaultConfig.test.ts b/server/defaultConfig.test.ts index 4db2936..5f11907 100644 --- a/server/defaultConfig.test.ts +++ b/server/defaultConfig.test.ts @@ -48,6 +48,8 @@ describe("DEFAULT_CONFIG", () => { "deploymentCheckPollIntervalMs", "maxConcurrentIssueEvaluations", "maxConcurrentIssueWork", + "maxAgentRetryAttempts", + "maxAgentInvocationsPerHour", "watchedRepos", "trustedReviewers", "priorityIssueAuthors", @@ -124,6 +126,11 @@ describe("DEFAULT_CONFIG", () => { assert.equal(DEFAULT_CONFIG.maxConcurrentIssueWork, 1); }); + it("leaves the hourly agent-run ceiling unlimited by default", () => { + // 0 means unlimited: upgrading must not start refusing work that used to run. + assert.equal(DEFAULT_CONFIG.maxAgentInvocationsPerHour, 0); + }); + it("uses rate-limit-safe install timer defaults", () => { assert.equal(DEFAULT_CONFIG.pollIntervalMs, 600000); assert.equal(DEFAULT_CONFIG.batchWindowMs, 600000); diff --git a/server/defaultConfig.ts b/server/defaultConfig.ts index d0a4565..c82531c 100644 --- a/server/defaultConfig.ts +++ b/server/defaultConfig.ts @@ -32,6 +32,7 @@ export const DEFAULT_CONFIG: Config = { deploymentCheckTimeoutMs: 900000, deploymentCheckPollIntervalMs: 60000, maxAgentRetryAttempts: 3, + maxAgentInvocationsPerHour: 0, maxConcurrentIssueEvaluations: 2, maxConcurrentIssueWork: 1, maxConcurrentBabysitRuns: 3, diff --git a/server/deploymentHealingAgent.ts b/server/deploymentHealingAgent.ts index f77985c..45050b8 100644 --- a/server/deploymentHealingAgent.ts +++ b/server/deploymentHealingAgent.ts @@ -1,5 +1,6 @@ import type { AgentRuntimeSettings, CodingAgent, CommandResult } from "./agentRunner"; import { applyFixesWithAgent, runCommand, summarizeCommandResult } from "./agentRunner"; +import { resolveAgentModel, withAgentWork } from "./agentSpend"; import { ensureRepoCache } from "./repoWorkspace"; import type { DeploymentPlatform } from "@shared/schema"; @@ -128,13 +129,18 @@ export async function runDeploymentHealingRepair( } try { - const agentResult = await deps.applyFixesWithAgent({ + const agentResult = await withAgentWork({ + kind: "heal_deployment", + repo: input.repo, + targetId: `${input.repo}@${input.mergeSha}`, + model: resolveAgentModel(input.agent, input.agentSettings), + }, () => deps.applyFixesWithAgent({ agent: input.agent, settings: input.agentSettings, cwd: repoCacheDir, prompt, env: input.env, - }); + })); if (agentResult.code !== 0) { return { diff --git a/server/failureRecovery.test.ts b/server/failureRecovery.test.ts index 90af979..b534610 100644 --- a/server/failureRecovery.test.ts +++ b/server/failureRecovery.test.ts @@ -3,6 +3,7 @@ import assert from "node:assert/strict"; import test from "node:test"; +import { AgentBudgetExhaustedError } from "./agentSpend"; import { AGENT_INVOKING_JOB_KINDS, classifyFailure, @@ -186,3 +187,25 @@ test("planFailedJobRecovery ignores jobs that are not parked", () => { assert.deepEqual(revivable, []); }); + +test("an exhausted agent budget is transient, not a spent retry attempt", () => { + const refusal = new AgentBudgetExhaustedError({ + used: 30, + max: 30, + resetsAt: "2026-08-29T13:00:00.000Z", + }); + + // The agent never ran, so this must not consume the paid retry budget. + assert.equal(classifyFailure(refusal), "transient"); + assert.notEqual( + resolveMaxAttempts({ + kind: "babysit_pr", + failureClass: "transient", + maxAgentRetryAttempts: 3, + }), + 3, + ); + + // The message survives a serialization boundary that loses the class. + assert.equal(classifyFailure(new Error(refusal.message)), "transient"); +}); diff --git a/server/failureRecovery.ts b/server/failureRecovery.ts index 8a4f689..2c75e7f 100644 --- a/server/failureRecovery.ts +++ b/server/failureRecovery.ts @@ -3,6 +3,7 @@ import type { BackgroundJobKind } from "@shared/schema"; import { detectAgentUnavailability } from "./agentRunner"; +import { isAgentBudgetExhaustedError } from "./agentSpend"; /** * How a background job failure should be treated by the retry policy. @@ -43,6 +44,7 @@ const TRANSIENT_PATTERNS: RegExp[] = [ /\bsecondary rate limit\b/i, /\babuse detection\b/i, /\bbudget is in the reserve band\b/i, + /\bagent budget exhausted\b/i, /\bserver error\b/i, ]; @@ -82,6 +84,12 @@ function matchesAny(message: string, patterns: RegExp[]): boolean { * expired credential. */ export function classifyFailure(error: unknown): FailureClass { + // The agent never ran, so this cost nothing and must not consume a paid + // retry attempt. Back off and try again once the spend window rolls. + if (isAgentBudgetExhaustedError(error)) { + return "transient"; + } + const message = toMessage(error).trim(); if (message.length === 0) { return "retryable"; diff --git a/server/issueWorkAgent.ts b/server/issueWorkAgent.ts index 0d7ff41..33ccb45 100644 --- a/server/issueWorkAgent.ts +++ b/server/issueWorkAgent.ts @@ -1,6 +1,7 @@ import { readFile } from "node:fs/promises"; import type { AgentRuntimeSettings, CodingAgent, CommandResult } from "./agentRunner"; import { applyFixesWithAgent, runCommand, summarizeAgentCommandFailure } from "./agentRunner"; +import { resolveAgentModel, withAgentWork } from "./agentSpend"; import { preparePrWorktree, removePrWorktree } from "./repoWorkspace"; import type { IssueSubtask, IssueSubtaskStatus } from "@shared/schema"; import path from "node:path"; @@ -479,13 +480,18 @@ export async function runIssueWorkRepair( await ensureGitIdentity(worktreePath, deps.runCommand); - const agentResult = await deps.applyFixesWithAgent({ + const agentResult = await withAgentWork({ + kind: "work_issue", + repo: input.repo, + targetId: `${input.repo}#${input.issueNumber}`, + model: resolveAgentModel(input.agent, input.agentSettings), + }, () => deps.applyFixesWithAgent({ agent: input.agent, settings: input.agentSettings, cwd: worktreePath, prompt: repoPrompt, env: input.env, - }); + })); if (agentResult.code !== 0) { return { diff --git a/server/logsRetention.test.ts b/server/logsRetention.test.ts index 9b7649e..309609b 100644 --- a/server/logsRetention.test.ts +++ b/server/logsRetention.test.ts @@ -1,7 +1,7 @@ import test from "node:test"; import assert from "node:assert/strict"; import { MemStorage } from "./storage"; -import { DEFAULT_LOG_RETENTION_DAYS, pruneLogsOnce, STDERR_LOG_MESSAGE_PREFIX, startLogsRetentionJob } from "./logsRetention"; +import { DEFAULT_LOG_RETENTION_DAYS, pruneAgentInvocationsOnce, pruneLogsOnce, STDERR_LOG_MESSAGE_PREFIX, startLogsRetentionJob } from "./logsRetention"; async function seedLog(storage: MemStorage, message: string, timestampOverride?: string) { const entry = await storage.addLog("pr-1", "info", message); @@ -57,3 +57,37 @@ test("startLogsRetentionJob ticks once at startup and on the interval", async () handle.stop(); } }); + +test("pruneAgentInvocationsOnce drops ledger rows past the retention horizon", async () => { + const storage = new MemStorage(); + const now = Date.now(); + + const rows = [ + { id: "fresh", startedAt: new Date(now - 60_000).toISOString() }, + { id: "stale", startedAt: new Date(now - 31 * 24 * 60 * 60 * 1000).toISOString() }, + ]; + + for (const row of rows) { + await storage.recordAgentInvocationStart({ + id: row.id, + workKind: "babysit_pr", + agent: "claude", + model: null, + repo: "acme/app", + targetId: "pr-1", + agentRunId: null, + startedAt: row.startedAt, + finishedAt: row.startedAt, + durationMs: 0, + exitCode: 0, + outcome: "completed", + error: null, + }); + } + + const removed = await pruneAgentInvocationsOnce(storage); + assert.equal(removed, 1); + + const remaining = await storage.listAgentInvocationsSince(new Date(0).toISOString()); + assert.deepEqual(remaining.map((row) => row.id), ["fresh"]); +}); diff --git a/server/logsRetention.ts b/server/logsRetention.ts index efb9bec..6b7bd58 100644 --- a/server/logsRetention.ts +++ b/server/logsRetention.ts @@ -1,3 +1,4 @@ +import { AGENT_SPEND_RETENTION_MS } from "./agentSpend"; import { childLogger } from "./logger"; import type { IStorage } from "./storage"; @@ -33,6 +34,23 @@ export async function pruneLogsOnce( return { byAge, byStderrPrefix }; } +/** + * Drop agent-spend ledger rows past the retention horizon. Kept separate from + * log pruning because the ledger has its own, much longer horizon. + */ +export async function pruneAgentInvocationsOnce( + storage: IStorage, + options: { retentionMs?: number } = {}, +): Promise { + const retentionMs = options.retentionMs ?? AGENT_SPEND_RETENTION_MS; + const cutoff = new Date(Date.now() - retentionMs).toISOString(); + const removed = await storage.pruneAgentInvocationsBefore(cutoff); + + log.info({ cutoff, removed }, "Pruned agent invocation ledger"); + + return removed; +} + export type RetentionJobHandle = { stop: () => void; }; @@ -56,6 +74,12 @@ export function startLogsRetentionJob( "Scheduled logs prune failed", ); }); + pruneAgentInvocationsOnce(storage).catch((err) => { + log.warn( + { err: err instanceof Error ? err.message : String(err) }, + "Scheduled agent invocation prune failed", + ); + }); }; // Run once at startup so existing bloat starts shrinking immediately, then on diff --git a/server/mcp.ts b/server/mcp.ts index eb3c9f4..23058c2 100644 --- a/server/mcp.ts +++ b/server/mcp.ts @@ -344,6 +344,14 @@ const TOOLS: Tool[] = [ "Get PatchDeck runtime state: drain mode status, active run count, and timestamps.", inputSchema: { type: "object", properties: {}, required: [] }, }, + { + name: "get_agent_spend", + description: + "Get coding-agent spend for the rolling hour: the ceiling, how many agent runs have been " + + "used, how many remain, when the next slot frees up, and a breakdown by work kind. " + + "A ceiling of 0 means unlimited.", + inputSchema: { type: "object", properties: {}, required: [] }, + }, { name: "set_drain_mode", description: @@ -503,6 +511,8 @@ async function callTool(name: string, args: ToolArgs): Promise { // Runtime case "get_runtime": return cfFetch("GET", "/api/runtime"); + case "get_agent_spend": + return cfFetch("GET", "/api/agent-spend"); case "set_drain_mode": return cfFetch("POST", "/api/runtime/drain", { enabled: args.enabled, diff --git a/server/memoryStorage.ts b/server/memoryStorage.ts index e47f4da..4905534 100644 --- a/server/memoryStorage.ts +++ b/server/memoryStorage.ts @@ -1,6 +1,9 @@ import type { + AgentInvocation, + AgentInvocationOutcome, AgentRun, AgentRunStatus, + AgentWorkKind, BackgroundJob, BackgroundJobKind, BackgroundJobStatus, @@ -73,6 +76,7 @@ export class MemStorage implements IStorage { private failureFingerprints: Map = new Map(); private releaseRuns: Map = new Map(); private agentRuns: Map = new Map(); + private agentInvocations: Map = new Map(); private socialChangelogs: Map = new Map(); private backgroundJobs: Map = new Map(); private deploymentHealingSessions: Map = new Map(); @@ -1016,6 +1020,82 @@ export class MemStorage implements IStorage { return { ...stored }; } + async recordAgentInvocationStart(invocation: AgentInvocation): Promise { + this.agentInvocations.set(invocation.id, { ...invocation }); + return { ...invocation }; + } + + async recordAgentInvocationEnd(id: string, end: { + finishedAt: string; + durationMs: number; + exitCode: number | null; + outcome: AgentInvocationOutcome; + error: string | null; + }): Promise { + const existing = this.agentInvocations.get(id); + if (!existing) { + return; + } + + this.agentInvocations.set(id, { ...existing, ...end }); + } + + async countAgentInvocationsSince(since: string, options?: { + excludeKinds?: AgentWorkKind[]; + }): Promise { + const excluded = new Set(options?.excludeKinds ?? []); + return Array.from(this.agentInvocations.values()) + .filter((invocation) => invocation.startedAt >= since && !excluded.has(invocation.workKind)) + .length; + } + + async listAgentInvocationsSince(since: string, options?: { + targetId?: string; + limit?: number; + }): Promise { + const limit = Math.max(1, Math.floor(options?.limit ?? 500)); + return Array.from(this.agentInvocations.values()) + .filter((invocation) => { + if (invocation.startedAt < since) return false; + if (options?.targetId && invocation.targetId !== options.targetId) return false; + return true; + }) + .sort((a, b) => (a.startedAt < b.startedAt ? 1 : a.startedAt > b.startedAt ? -1 : 0)) + .slice(0, limit) + .map((invocation) => ({ ...invocation })); + } + + async closeOrphanedAgentInvocations(finishedAt: string): Promise { + let closed = 0; + for (const [id, invocation] of Array.from(this.agentInvocations.entries())) { + if (invocation.outcome !== "running") { + continue; + } + + this.agentInvocations.set(id, { + ...invocation, + outcome: "failed", + finishedAt, + error: invocation.error ?? "PatchDeck restarted while the agent was running", + }); + closed += 1; + } + + return closed; + } + + async pruneAgentInvocationsBefore(cutoff: string): Promise { + let removed = 0; + for (const [id, invocation] of Array.from(this.agentInvocations.entries())) { + if (invocation.startedAt < cutoff) { + this.agentInvocations.delete(id); + removed += 1; + } + } + + return removed; + } + async getSocialChangelogs(): Promise { return Array.from(this.socialChangelogs.values()).sort( (a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime(), diff --git a/server/prQuestionAgent.ts b/server/prQuestionAgent.ts index 28bb1b8..3b9ab47 100644 --- a/server/prQuestionAgent.ts +++ b/server/prQuestionAgent.ts @@ -1,6 +1,7 @@ import type { IStorage } from "./storage"; import type { AgentRuntimeSettings, CodingAgent } from "./agentRunner"; import { buildAgentCommandArgs, resolveAgent, runAgentCommand, summarizeCommandResult } from "./agentRunner"; +import { resolveAgentModel, withAgentWork } from "./agentSpend"; /** * Answers a user question about a PR by gathering context (PR state, feedback, @@ -24,13 +25,17 @@ export async function answerPRQuestion(params: { const agent = await resolveAgent(preferredAgent); const prompt = buildPrompt(context, question); - const result = await runAgentCommand( + const result = await withAgentWork({ + kind: "answer_pr_question", + targetId: prId, + model: resolveAgentModel(agent, agentSettings), + }, () => runAgentCommand( agent, agent === "claude" ? buildAgentCommandArgs("claude", ["-p", "--output-format", "text", prompt], agentSettings) : buildAgentCommandArgs("codex", ["exec", "--skip-git-repo-check", "--sandbox", "read-only", prompt], agentSettings), { timeoutMs: 180_000 }, - ); + )); if (result.code !== 0) { const errorMsg = summarizeCommandResult(result, `Agent exited with code ${result.code}`); diff --git a/server/releaseAgent.ts b/server/releaseAgent.ts index d2985da..9d3c275 100644 --- a/server/releaseAgent.ts +++ b/server/releaseAgent.ts @@ -2,6 +2,7 @@ import { mkdtemp, readFile, rm } from "fs/promises"; import { tmpdir } from "os"; import path from "path"; import { buildAgentCommandArgs, resolveAgent, runAgentCommand, type AgentRuntimeSettings, type CodingAgent } from "./agentRunner"; +import { resolveAgentModel, withAgentWork, type AgentWorkContext } from "./agentSpend"; const DEFAULT_RELEASE_AGENT_TIMEOUT_MS = 120_000; @@ -40,13 +41,19 @@ export async function evaluateReleaseWorthinessWithAgent(params: { const timeoutMs = params.timeoutMs ?? DEFAULT_RELEASE_AGENT_TIMEOUT_MS; const agent = await resolveAgent(params.preferredAgent); const prompt = buildReleaseDecisionPrompt(params); + const work: AgentWorkContext = { + kind: "release_notes", + repo: params.repo, + targetId: `${params.repo}#${params.triggerPr.number}`, + model: resolveAgentModel(agent, params.agentSettings), + }; if (agent === "codex") { const tempDir = await mkdtemp(path.join(tmpdir(), "codex-release-eval-")); const outputFile = path.join(tempDir, "output.txt"); try { - const result = await runAgentCommand( + const result = await withAgentWork(work, () => runAgentCommand( "codex", buildAgentCommandArgs("codex", [ "exec", @@ -58,7 +65,7 @@ export async function evaluateReleaseWorthinessWithAgent(params: { prompt, ], params.agentSettings), { cwd, timeoutMs }, - ); + )); if (result.code !== 0) { throw new Error(`codex release evaluation failed (${result.code}): ${result.stderr || result.stdout}`); @@ -71,7 +78,7 @@ export async function evaluateReleaseWorthinessWithAgent(params: { } } - const result = await runAgentCommand( + const result = await withAgentWork(work, () => runAgentCommand( "claude", buildAgentCommandArgs("claude", [ "-p", @@ -80,7 +87,7 @@ export async function evaluateReleaseWorthinessWithAgent(params: { prompt, ], params.agentSettings), { cwd, timeoutMs }, - ); + )); if (result.code !== 0) { throw new Error(`claude release evaluation failed (${result.code}): ${result.stderr || result.stdout}`); diff --git a/server/releaseSocialPostAgent.ts b/server/releaseSocialPostAgent.ts index f3ac85d..6a9fc05 100644 --- a/server/releaseSocialPostAgent.ts +++ b/server/releaseSocialPostAgent.ts @@ -1,5 +1,6 @@ import type { CodingAgent } from "./agentRunner"; import { buildAgentCommandArgs, resolveAgent, runAgentCommand, summarizeCommandResult, type AgentRuntimeSettings } from "./agentRunner"; +import { resolveAgentModel, withAgentWork } from "./agentSpend"; export type ReleaseSocialPostInput = { repo: string; @@ -94,13 +95,18 @@ export async function generateReleaseSocialPost(params: { const agent = await resolveAgent(preferredAgent); const prompt = buildPrompt(input); - const result = await runAgentCommand( + const result = await withAgentWork({ + kind: "social_post", + repo: input.repo, + targetId: `${input.repo}@${input.tagName}`, + model: resolveAgentModel(agent, agentSettings), + }, () => runAgentCommand( agent, agent === "claude" ? buildAgentCommandArgs("claude", ["-p", "--output-format", "text", prompt], agentSettings) : buildAgentCommandArgs("codex", ["exec", "--skip-git-repo-check", "--sandbox", "read-only", prompt], agentSettings), { timeoutMs }, - ); + )); if (result.code !== 0) { throw new Error(summarizeCommandResult(result, `Agent exited with code ${result.code}`)); diff --git a/server/routes.test.ts b/server/routes.test.ts index 3832ed5..a7fba1e 100644 --- a/server/routes.test.ts +++ b/server/routes.test.ts @@ -1907,3 +1907,52 @@ test("GET and POST /api/issues proxy the runtime issue monitor and work action", await harness.close(); } }); + +test("GET /api/agent-spend reports the rolling hour, and 0 means unlimited", async () => { + const storage = new MemStorage(); + const harness = await createHarness(storage); + + try { + const unlimited = await fetch(`${harness.baseUrl}/api/agent-spend`); + assert.equal(unlimited.status, 200); + const unlimitedBody = await unlimited.json(); + assert.equal(unlimitedBody.max, 0); + assert.equal(unlimitedBody.used, 0); + assert.equal(unlimitedBody.remaining, null); + assert.equal(unlimitedBody.exhausted, false); + + const config = await storage.getConfig(); + await storage.updateConfig({ ...config, maxAgentInvocationsPerHour: 2 }); + + const startedAt = new Date().toISOString(); + for (const [id, workKind] of [["inv-1", "babysit_pr"], ["inv-2", "heal_ci"], ["inv-3", "probe"]] as const) { + await storage.recordAgentInvocationStart({ + id, + workKind, + agent: "claude", + model: "opus", + repo: "acme/widgets", + targetId: "pr-1", + agentRunId: null, + startedAt, + finishedAt: startedAt, + durationMs: 1000, + exitCode: 0, + outcome: "completed", + error: null, + }); + } + + const limited = await fetch(`${harness.baseUrl}/api/agent-spend`); + const body = await limited.json(); + assert.equal(body.max, 2); + // The probe is reported in the breakdown but never counted as spend. + assert.equal(body.used, 2); + assert.equal(body.remaining, 0); + assert.equal(body.exhausted, true); + assert.equal(body.byKind.find((entry: { workKind: string }) => entry.workKind === "probe")?.count, 1); + assert.ok(new Date(body.resetsAt).getTime() > Date.now()); + } finally { + await harness.close(); + } +}); diff --git a/server/routes.ts b/server/routes.ts index 3df62e5..c81feea 100644 --- a/server/routes.ts +++ b/server/routes.ts @@ -262,6 +262,10 @@ export async function registerRoutes( runtime.stop(); }); + app.get("/api/agent-spend", async (_req, res) => { + res.json(await runtime.getAgentSpend()); + }); + app.get("/api/runtime", async (_req, res) => { res.json(await runtime.getRuntimeSnapshot()); }); diff --git a/server/sqliteStorage.ts b/server/sqliteStorage.ts index d82548c..6591e99 100644 --- a/server/sqliteStorage.ts +++ b/server/sqliteStorage.ts @@ -3,8 +3,11 @@ import { DatabaseSync } from "node:sqlite"; import type { SQLInputValue } from "node:sqlite"; import { backgroundJobStatusEnum, docsAssessmentSchema, feedbackStatusEnum, prStageEnum, prWorkContractSchema } from "@shared/schema"; import type { + AgentInvocation, + AgentInvocationOutcome, AgentRun, AgentRunStatus, + AgentWorkKind, BackgroundJob, BackgroundJobKind, BackgroundJobStatus, @@ -91,6 +94,7 @@ type ConfigRow = { post_github_progress_replies: number; auto_heal_ci: number; max_agent_retry_attempts: number; + max_agent_invocations_per_hour: number; max_healing_attempts_per_session: number; max_healing_attempts_per_fingerprint: number; max_concurrent_healing_runs: number; @@ -243,6 +247,22 @@ type AgentRunRow = { updated_at: string; }; +type AgentInvocationRow = { + id: string; + work_kind: AgentWorkKind; + agent: Config["codingAgent"]; + model: string | null; + repo: string | null; + target_id: string | null; + agent_run_id: string | null; + started_at: string; + finished_at: string | null; + duration_ms: number | null; + exit_code: number | null; + outcome: AgentInvocationOutcome; + error: string | null; +}; + type QuestionRow = { id: string; pr_id: string; @@ -589,6 +609,7 @@ export class SqliteStorage implements IStorage { post_github_progress_replies INTEGER NOT NULL DEFAULT 0, auto_heal_ci INTEGER NOT NULL DEFAULT 0, max_agent_retry_attempts INTEGER NOT NULL DEFAULT 3, + max_agent_invocations_per_hour INTEGER NOT NULL DEFAULT 0, max_healing_attempts_per_session INTEGER NOT NULL DEFAULT 3, max_healing_attempts_per_fingerprint INTEGER NOT NULL DEFAULT 2, max_concurrent_healing_runs INTEGER NOT NULL DEFAULT 1, @@ -696,6 +717,22 @@ export class SqliteStorage implements IStorage { FOREIGN KEY(pr_id) REFERENCES prs(id) ON DELETE CASCADE ); + CREATE TABLE IF NOT EXISTS agent_invocations ( + id TEXT PRIMARY KEY, + work_kind TEXT NOT NULL, + agent TEXT NOT NULL, + model TEXT, + repo TEXT, + target_id TEXT, + agent_run_id TEXT, + started_at TEXT NOT NULL, + finished_at TEXT, + duration_ms INTEGER, + exit_code INTEGER, + outcome TEXT NOT NULL, + error TEXT + ); + CREATE TABLE IF NOT EXISTS background_jobs ( id TEXT PRIMARY KEY, kind TEXT NOT NULL, @@ -908,6 +945,9 @@ export class SqliteStorage implements IStorage { CREATE INDEX IF NOT EXISTS idx_feedback_items_pr_id ON feedback_items(pr_id); CREATE INDEX IF NOT EXISTS idx_logs_pr_id_timestamp ON logs(pr_id, timestamp); CREATE INDEX IF NOT EXISTS idx_agent_runs_status_updated_at ON agent_runs(status, updated_at); + CREATE INDEX IF NOT EXISTS idx_agent_invocations_started_at ON agent_invocations(started_at); + CREATE INDEX IF NOT EXISTS idx_agent_invocations_kind_started_at ON agent_invocations(work_kind, started_at); + CREATE INDEX IF NOT EXISTS idx_agent_invocations_target_started_at ON agent_invocations(target_id, started_at); CREATE INDEX IF NOT EXISTS idx_background_jobs_status_available_at ON background_jobs(status, available_at, priority, created_at); CREATE INDEX IF NOT EXISTS idx_background_jobs_lease_expires_at ON background_jobs(status, lease_expires_at); CREATE INDEX IF NOT EXISTS idx_background_jobs_kind_status ON background_jobs(kind, status); @@ -956,6 +996,7 @@ export class SqliteStorage implements IStorage { this.ensureColumn("config", "post_github_progress_replies", "INTEGER NOT NULL DEFAULT 0"); this.ensureColumn("config", "auto_heal_ci", "INTEGER NOT NULL DEFAULT 0"); this.ensureColumn("config", "max_agent_retry_attempts", "INTEGER NOT NULL DEFAULT 3"); + this.ensureColumn("config", "max_agent_invocations_per_hour", "INTEGER NOT NULL DEFAULT 0"); this.ensureColumn("config", "max_healing_attempts_per_session", "INTEGER NOT NULL DEFAULT 3"); this.ensureColumn("config", "max_healing_attempts_per_fingerprint", "INTEGER NOT NULL DEFAULT 2"); this.ensureColumn("config", "max_concurrent_healing_runs", "INTEGER NOT NULL DEFAULT 1"); @@ -1094,6 +1135,7 @@ export class SqliteStorage implements IStorage { ), autoHealCI: Boolean(row.auto_heal_ci ?? Number(DEFAULT_CONFIG.autoHealCI)), maxAgentRetryAttempts: row.max_agent_retry_attempts ?? DEFAULT_CONFIG.maxAgentRetryAttempts, + maxAgentInvocationsPerHour: row.max_agent_invocations_per_hour ?? DEFAULT_CONFIG.maxAgentInvocationsPerHour, maxHealingAttemptsPerSession: row.max_healing_attempts_per_session ?? DEFAULT_CONFIG.maxHealingAttemptsPerSession, maxHealingAttemptsPerFingerprint: row.max_healing_attempts_per_fingerprint ?? DEFAULT_CONFIG.maxHealingAttemptsPerFingerprint, maxConcurrentHealingRuns: row.max_concurrent_healing_runs ?? DEFAULT_CONFIG.maxConcurrentHealingRuns, @@ -1149,6 +1191,7 @@ export class SqliteStorage implements IStorage { post_github_progress_replies, auto_heal_ci, max_agent_retry_attempts, + max_agent_invocations_per_hour, max_healing_attempts_per_session, max_healing_attempts_per_fingerprint, max_concurrent_healing_runs, @@ -1163,7 +1206,7 @@ export class SqliteStorage implements IStorage { trusted_reviewers_json, priority_issue_authors_json, ignored_bots_json - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET github_token = excluded.github_token, github_tokens_json = excluded.github_tokens_json, @@ -1190,6 +1233,7 @@ export class SqliteStorage implements IStorage { post_github_progress_replies = excluded.post_github_progress_replies, auto_heal_ci = excluded.auto_heal_ci, max_agent_retry_attempts = excluded.max_agent_retry_attempts, + max_agent_invocations_per_hour = excluded.max_agent_invocations_per_hour, max_healing_attempts_per_session = excluded.max_healing_attempts_per_session, max_healing_attempts_per_fingerprint = excluded.max_healing_attempts_per_fingerprint, max_concurrent_healing_runs = excluded.max_concurrent_healing_runs, @@ -1231,6 +1275,7 @@ export class SqliteStorage implements IStorage { Number(config.postGitHubProgressReplies), Number(config.autoHealCI), config.maxAgentRetryAttempts, + config.maxAgentInvocationsPerHour, config.maxHealingAttemptsPerSession, config.maxHealingAttemptsPerFingerprint, config.maxConcurrentHealingRuns, @@ -1418,6 +1463,24 @@ export class SqliteStorage implements IStorage { }; } + private parseAgentInvocationRow(row: AgentInvocationRow): AgentInvocation { + return { + id: row.id, + workKind: row.work_kind, + agent: row.agent, + model: row.model, + repo: row.repo, + targetId: row.target_id, + agentRunId: row.agent_run_id, + startedAt: row.started_at, + finishedAt: row.finished_at, + durationMs: row.duration_ms, + exitCode: row.exit_code, + outcome: row.outcome, + error: row.error, + }; + } + private parseBackgroundJobRow(row: BackgroundJobRow): BackgroundJob { return { id: row.id, @@ -2045,7 +2108,7 @@ export class SqliteStorage implements IStorage { poll_interval_ms, max_changes_per_run, auto_resolve_merge_conflicts, auto_create_releases, auto_update_docs, auto_prs, auto_issues, include_repository_links_in_github_comments, github_comment_app_name, post_github_progress_replies, - auto_heal_ci, max_agent_retry_attempts, max_healing_attempts_per_session, + auto_heal_ci, max_agent_retry_attempts, max_agent_invocations_per_hour, max_healing_attempts_per_session, max_healing_attempts_per_fingerprint, max_concurrent_healing_runs, healing_cooldown_ms, auto_heal_deployments, deployment_check_delay_ms, deployment_check_timeout_ms, deployment_check_poll_interval_ms, max_concurrent_issue_evaluations, max_concurrent_issue_work, @@ -3373,6 +3436,111 @@ export class SqliteStorage implements IStorage { return stored; } + // ── Agent spend ledger ────────────────────────────────────────────────── + + async recordAgentInvocationStart(invocation: AgentInvocation): Promise { + this.run(` + INSERT INTO agent_invocations ( + id, work_kind, agent, model, repo, target_id, agent_run_id, + started_at, finished_at, duration_ms, exit_code, outcome, error + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + invocation.id, + invocation.workKind, + invocation.agent, + invocation.model, + invocation.repo, + invocation.targetId, + invocation.agentRunId, + invocation.startedAt, + invocation.finishedAt, + invocation.durationMs, + invocation.exitCode, + invocation.outcome, + invocation.error, + ); + + return invocation; + } + + async recordAgentInvocationEnd(id: string, end: { + finishedAt: string; + durationMs: number; + exitCode: number | null; + outcome: AgentInvocationOutcome; + error: string | null; + }): Promise { + this.run(` + UPDATE agent_invocations + SET finished_at = ?, duration_ms = ?, exit_code = ?, outcome = ?, error = ? + WHERE id = ? + `, end.finishedAt, end.durationMs, end.exitCode, end.outcome, end.error, id); + } + + async countAgentInvocationsSince(since: string, options?: { + excludeKinds?: AgentWorkKind[]; + }): Promise { + const excluded = options?.excludeKinds ?? []; + const placeholders = excluded.map(() => "?").join(", "); + const exclusion = excluded.length > 0 ? `AND work_kind NOT IN (${placeholders})` : ""; + + const row = this.get<{ count: number }>(` + SELECT COUNT(*) AS count + FROM agent_invocations + WHERE started_at >= ? ${exclusion} + `, since, ...excluded); + + return row ? Number(row.count) : 0; + } + + async listAgentInvocationsSince(since: string, options?: { + targetId?: string; + limit?: number; + }): Promise { + const values: (string | number)[] = [since]; + let clause = ""; + + if (options?.targetId) { + clause = "AND target_id = ?"; + values.push(options.targetId); + } + + const limit = Math.max(1, Math.floor(options?.limit ?? 500)); + values.push(limit); + + const rows = this.all(` + SELECT id, work_kind, agent, model, repo, target_id, agent_run_id, + started_at, finished_at, duration_ms, exit_code, outcome, error + FROM agent_invocations + WHERE started_at >= ? ${clause} + ORDER BY started_at DESC + LIMIT ? + `, ...values); + + return rows.map((row) => this.parseAgentInvocationRow(row)); + } + + async closeOrphanedAgentInvocations(finishedAt: string): Promise { + const result = this.run(` + UPDATE agent_invocations + SET outcome = 'failed', + finished_at = ?, + error = COALESCE(error, 'PatchDeck restarted while the agent was running') + WHERE outcome = 'running' + `, finishedAt); + + return Number(result.changes); + } + + async pruneAgentInvocationsBefore(cutoff: string): Promise { + const result = this.run(` + DELETE FROM agent_invocations + WHERE started_at < ? + `, cutoff); + + return Number(result.changes); + } + // ── Social changelogs ─────────────────────────────────────────────────── async getSocialChangelogs(): Promise { diff --git a/server/storage.test.ts b/server/storage.test.ts index d407c57..9434402 100644 --- a/server/storage.test.ts +++ b/server/storage.test.ts @@ -697,6 +697,175 @@ test("SqliteStorage upsertAgentRun preserves the original createdAt", async () = storage.close(); }); +test("agent invocation ledger round-trips, windows, and survives PR deletion", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "codefactory-storage-")); + const storage = new SqliteStorage(root); + + const pr = await storage.addPR({ + number: 61, + title: "Ledger survives deletion", + repo: "jeremymcs/patchdeck", + branch: "claude/spend-ledger", + author: "claude", + url: "https://github.com/jeremymcs/patchdeck/pull/61", + status: "watching", + feedbackItems: [], + accepted: 0, + rejected: 0, + flagged: 0, + testsPassed: null, + lintPassed: null, + lastChecked: null, + }); + + const inWindow = "2026-08-29T11:30:00.000Z"; + const outOfWindow = "2026-08-29T09:00:00.000Z"; + const windowStart = "2026-08-29T11:00:00.000Z"; + + await storage.recordAgentInvocationStart({ + id: "inv-1", + workKind: "babysit_pr", + agent: "claude", + model: "opus", + repo: pr.repo, + targetId: pr.id, + agentRunId: "run-1", + startedAt: inWindow, + finishedAt: null, + durationMs: null, + exitCode: null, + outcome: "running", + error: null, + }); + + await storage.recordAgentInvocationStart({ + id: "inv-2", + workKind: "probe", + agent: "claude", + model: null, + repo: null, + targetId: null, + agentRunId: null, + startedAt: inWindow, + finishedAt: null, + durationMs: null, + exitCode: null, + outcome: "running", + error: null, + }); + + await storage.recordAgentInvocationStart({ + id: "inv-3", + workKind: "work_issue", + agent: "codex", + model: null, + repo: pr.repo, + targetId: "jeremymcs/patchdeck#12", + agentRunId: null, + startedAt: outOfWindow, + finishedAt: null, + durationMs: null, + exitCode: null, + outcome: "running", + error: null, + }); + + await storage.recordAgentInvocationEnd("inv-1", { + finishedAt: "2026-08-29T11:31:00.000Z", + durationMs: 60_000, + exitCode: 0, + outcome: "completed", + error: null, + }); + + const stored = await storage.listAgentInvocationsSince(windowStart); + assert.deepEqual(stored.map((row) => row.id).sort(), ["inv-1", "inv-2"]); + const completed = stored.find((row) => row.id === "inv-1"); + assert.equal(completed?.outcome, "completed"); + assert.equal(completed?.durationMs, 60_000); + assert.equal(completed?.model, "opus"); + + // Probes are recorded but excluded from the metered count. + assert.equal(await storage.countAgentInvocationsSince(windowStart), 2); + assert.equal( + await storage.countAgentInvocationsSince(windowStart, { excludeKinds: ["probe"] }), + 1, + ); + + // Orphaned `running` rows are closed, not left counting forever. + assert.equal(await storage.closeOrphanedAgentInvocations("2026-08-29T12:00:00.000Z"), 2); + const closed = await storage.listAgentInvocationsSince(windowStart); + assert.equal(closed.every((row) => row.outcome !== "running"), true); + + // The whole point of carrying no foreign key: spend history outlives the PR. + assert.equal(await storage.removePR(pr.id), true); + assert.equal(await storage.getPR(pr.id), undefined); + const afterDelete = await storage.listAgentInvocationsSince(windowStart); + assert.equal(afterDelete.some((row) => row.id === "inv-1"), true); + + assert.equal(await storage.pruneAgentInvocationsBefore(windowStart), 1); + assert.equal((await storage.listAgentInvocationsSince(outOfWindow)).length, 2); + + storage.close(); +}); + +test("MemStorage agent invocation ledger matches the SQLite behaviour", async () => { + const storage = new MemStorage(); + const windowStart = "2026-08-29T11:00:00.000Z"; + + await storage.recordAgentInvocationStart({ + id: "inv-1", + workKind: "heal_ci", + agent: "claude", + model: null, + repo: "acme/app", + targetId: "acme/app#3", + agentRunId: null, + startedAt: "2026-08-29T11:30:00.000Z", + finishedAt: null, + durationMs: null, + exitCode: null, + outcome: "running", + error: null, + }); + await storage.recordAgentInvocationStart({ + id: "inv-2", + workKind: "probe", + agent: "claude", + model: null, + repo: null, + targetId: null, + agentRunId: null, + startedAt: "2026-08-29T10:00:00.000Z", + finishedAt: null, + durationMs: null, + exitCode: null, + outcome: "running", + error: null, + }); + + assert.equal(await storage.countAgentInvocationsSince(windowStart), 1); + assert.equal( + await storage.countAgentInvocationsSince(windowStart, { excludeKinds: ["heal_ci"] }), + 0, + ); + + await storage.recordAgentInvocationEnd("inv-1", { + finishedAt: "2026-08-29T11:35:00.000Z", + durationMs: 300_000, + exitCode: 1, + outcome: "failed", + error: "agent failed", + }); + const [row] = await storage.listAgentInvocationsSince(windowStart); + assert.equal(row.outcome, "failed"); + assert.equal(row.durationMs, 300_000); + + assert.equal(await storage.closeOrphanedAgentInvocations("2026-08-29T12:00:00.000Z"), 1); + assert.equal(await storage.pruneAgentInvocationsBefore(windowStart), 1); + assert.equal((await storage.listAgentInvocationsSince("2026-08-29T09:00:00.000Z")).length, 1); +}); + test("SqliteStorage persists background jobs and requeues expired leases", async () => { const root = await mkdtemp(path.join(os.tmpdir(), "codefactory-storage-")); const first = new SqliteStorage(root); diff --git a/server/storage.ts b/server/storage.ts index 798bc1c..138d1b0 100644 --- a/server/storage.ts +++ b/server/storage.ts @@ -1,6 +1,9 @@ import type { + AgentInvocation, + AgentInvocationOutcome, AgentRun, AgentRunStatus, + AgentWorkKind, BackgroundJob, BackgroundJobKind, BackgroundJobStatus, @@ -236,6 +239,25 @@ export interface IStorage { }): Promise; upsertAgentRun(run: AgentRun): Promise; + // Agent spend ledger (one row per coding-agent process spawn) + recordAgentInvocationStart(invocation: AgentInvocation): Promise; + recordAgentInvocationEnd(id: string, end: { + finishedAt: string; + durationMs: number; + exitCode: number | null; + outcome: AgentInvocationOutcome; + error: string | null; + }): Promise; + countAgentInvocationsSince(since: string, options?: { + excludeKinds?: AgentWorkKind[]; + }): Promise; + listAgentInvocationsSince(since: string, options?: { + targetId?: string; + limit?: number; + }): Promise; + closeOrphanedAgentInvocations(finishedAt: string): Promise; + pruneAgentInvocationsBefore(cutoff: string): Promise; + // Deployment healing getDeploymentHealingSession(id: string): Promise; getDeploymentHealingSessionByRepoAndMergeSha(repo: string, mergeSha: string): Promise; diff --git a/shared/schema.ts b/shared/schema.ts index 35f0507..c24aaca 100644 --- a/shared/schema.ts +++ b/shared/schema.ts @@ -192,6 +192,76 @@ export const agentRunSchema = z.object({ }); export type AgentRun = z.infer; +/** + * What a single coding-agent process spawn was doing. Distinct from + * `backgroundJobKind`: one job can spawn several agents (evaluate, then apply, + * then a fallback), and some agent work has no job of its own (CI healing runs + * inside `babysit_pr`). + */ +export const agentWorkKindEnum = z.enum([ + "babysit_pr", + "evaluate_feedback", + "work_issue", + "decompose_issue", + "verify_issue", + "heal_ci", + "heal_deployment", + "answer_pr_question", + "release_notes", + "social_post", + "probe", + "unattributed", +]); +export type AgentWorkKind = z.infer; + +export const agentInvocationOutcomeEnum = z.enum([ + "running", + "completed", + "failed", + "timeout", + "refused", +]); +export type AgentInvocationOutcome = z.infer; + +/** + * One row per coding-agent process spawn. This is the spend ledger; `agentRun` + * is a per-babysit-session record and is not interchangeable with it. Rows + * intentionally carry no foreign keys so history survives deletion of the PR or + * issue that caused the spend. + */ +export const agentInvocationSchema = z.object({ + id: z.string(), + workKind: agentWorkKindEnum, + agent: z.enum(["codex", "claude"]), + model: z.string().nullable(), + repo: z.string().nullable(), + targetId: z.string().nullable(), + agentRunId: z.string().nullable(), + startedAt: z.string(), + finishedAt: z.string().nullable(), + durationMs: z.number().int().nonnegative().nullable(), + exitCode: z.number().int().nullable(), + outcome: agentInvocationOutcomeEnum, + error: z.string().nullable(), +}); +export type AgentInvocation = z.infer; + +export const agentSpendSummarySchema = z.object({ + windowMs: z.number().int().positive(), + windowStartedAt: z.string(), + resetsAt: z.string(), + max: z.number().int().nonnegative(), + used: z.number().int().nonnegative(), + remaining: z.number().int().nonnegative().nullable(), + exhausted: z.boolean(), + byKind: z.array(z.object({ + workKind: agentWorkKindEnum, + count: z.number().int().nonnegative(), + totalDurationMs: z.number().int().nonnegative(), + })), +}); +export type AgentSpendSummary = z.infer; + export const runtimeStateSchema = z.object({ drainMode: z.boolean(), drainRequestedAt: z.string().nullable(), @@ -731,6 +801,8 @@ export const configSchema = z.object({ deploymentCheckTimeoutMs: z.number(), deploymentCheckPollIntervalMs: z.number(), maxAgentRetryAttempts: z.number().int().nonnegative().default(3), + /** Rolling-hour ceiling on coding-agent process spawns. 0 means unlimited. */ + maxAgentInvocationsPerHour: z.number().int().nonnegative().default(0), maxConcurrentIssueEvaluations: z.number().int().positive().default(2), maxConcurrentIssueWork: z.number().int().positive().default(1), maxConcurrentBabysitRuns: z.number().int().positive().default(3),