mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-23 09:07:46 +02:00
fix: address accounting recovery and budget review findings
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
dce0c0f265
commit
05035f9f65
22 files changed
+1552
-50001
No files matched your search
@@ -155,3 +155,20 @@ Responses carry an opaque account identity and successful capture timestamp. The
|
||||
Both Costs page variants retain loaded reports and expanded rows during background refresh failures, with a small stale-data notice. Initial failures remain visible. Company/date changes isolate cached reports. The run-count label is `runs: 0 api · 11 sub`: distinct recorded runs, not tasks or individual model requests. Input totals include cache reads, with the cached portion shown explicitly.
|
||||
|
||||
Receipt identity and supplemental charges: `heartbeat:` idempotency keys are reserved for internally generated run receipts. Additional API-reported charges may link to the same run with their own keys. They contribute to monthly and acknowledged-run lifetime totals, while receipt-integrity checks compare only the original provider receipt. Recovery isolates each budget scope so a deleted target or failed scope does not prevent recovery elsewhere.
|
||||
|
||||
### Review hardening
|
||||
|
||||
Native restart recovery reuses a held reservation only when the coordinator's
|
||||
current, unexpired lease owns the same run and project. Ordinary duplicate
|
||||
provider dispatch remains rejected. A process adapter with no tokens, provider,
|
||||
price, or explicit unknown-price marker does not create an unpriced charge;
|
||||
provider adapters with missing prices still do.
|
||||
|
||||
Budget policy updates may omit unchanged settings, including the amount for an
|
||||
existing policy. Omitted settings are read under the accounting transaction lock.
|
||||
Creating a policy still requires an amount. Generic agent and company budget
|
||||
edits deliver hard-stop cancellation after their transaction commits.
|
||||
|
||||
Status-card update costs retain the ledger's fractional-cent storage precision.
|
||||
The mutation gate retains partial results and per-run reports/logs under
|
||||
`coverage/accounting/` when its baseline fails or a mutation survives.
|
||||
@@ -0,0 +1,2 @@
|
||||
-- Preserve the ledger precision in status-card update snapshots. Reapplying this type widening is safe.
|
||||
ALTER TABLE "status_card_updates" ALTER COLUMN "cost_cents" SET DATA TYPE numeric(24, 7);
|
||||
-49948
File diff suppressed because it is too large.
Load diff
+1392
-12
File diff suppressed because it is too large.
Load diff
+7
@@ -2059,6 +2059,13 @@
|
||||
"when": 1790983662974,
|
||||
"tag": "0295_lean_baron_strucker",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 296,
|
||||
"version": "7",
|
||||
"when": 1791086889296,
|
||||
"tag": "0296_perpetual_black_cat",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import { boolean, index, integer, jsonb, pgTable, text, timestamp, uuid } from "drizzle-orm/pg-core";
|
||||
import { boolean, index, integer, numeric, jsonb, pgTable, text, timestamp, uuid } from "drizzle-orm/pg-core";
|
||||
import type { CompanySearchQuery, StatusCardRefreshPolicy } from "@paperclipai/shared";
|
||||
import { agents } from "./agents.js";
|
||||
import { companies } from "./companies.js";
|
||||
@@ -81,7 +81,7 @@ export const statusCardUpdates = pgTable(
|
||||
changes: jsonb("changes").$type<StatusCardUpdateChange[]>().notNull().default(sql`'[]'::jsonb`),
|
||||
inputTokens: integer("input_tokens").notNull().default(0),
|
||||
outputTokens: integer("output_tokens").notNull().default(0),
|
||||
costCents: integer("cost_cents").notNull().default(0),
|
||||
costCents: numeric("cost_cents", { precision: 24, scale: 7, mode: "number" }).notNull().default(0),
|
||||
model: text("model"),
|
||||
queryVersion: integer("query_version"),
|
||||
changeSummary: text("change_summary"),
|
||||
|
||||
@@ -92,7 +92,7 @@ export interface BudgetPolicyUpsertInput {
|
||||
scopeId: string;
|
||||
metric?: BudgetMetric;
|
||||
windowKind?: BudgetWindowKind;
|
||||
amount: number;
|
||||
amount?: number;
|
||||
reservationCents?: string | number;
|
||||
warnPercent?: number;
|
||||
hardStopEnabled?: boolean;
|
||||
|
||||
@@ -12,13 +12,13 @@ export const upsertBudgetPolicySchema = z.object({
|
||||
scopeId: z.string().guid(),
|
||||
metric: z.enum(BUDGET_METRICS).optional().default("billed_cents"),
|
||||
windowKind: z.enum(BUDGET_WINDOW_KINDS).optional().default("calendar_month_utc"),
|
||||
amount: z.number().int().nonnegative().max(2_147_483_647),
|
||||
amount: z.number().int().nonnegative().max(2_147_483_647).optional(),
|
||||
reservationCents: exactCentsSchema.optional(),
|
||||
warnPercent: z.number().int().min(1).max(99).optional().default(80),
|
||||
hardStopEnabled: z.boolean().optional().default(true),
|
||||
warnPercent: z.number().int().min(1).max(99).optional(),
|
||||
hardStopEnabled: z.boolean().optional(),
|
||||
unpricedUsagePolicy: z.enum(["block", "allow"]).optional(),
|
||||
notifyEnabled: z.boolean().optional().default(true),
|
||||
isActive: z.boolean().optional().default(true),
|
||||
notifyEnabled: z.boolean().optional(),
|
||||
isActive: z.boolean().optional(),
|
||||
});
|
||||
|
||||
export type UpsertBudgetPolicy = z.infer<typeof upsertBudgetPolicySchema>;
|
||||
|
||||
@@ -1,34 +1,47 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { mkdtemp, readFile, rm, writeFile, mkdir } from "node:fs/promises";
|
||||
import { mkdtemp, readFile, rm, writeFile, mkdir, copyFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
const root = fileURLToPath(new URL("..", import.meta.url));
|
||||
const directory = await mkdtemp(path.join(tmpdir(), "accounting-mutations-"));
|
||||
const artifactDirectory = path.join(root, "coverage/accounting/mutations");
|
||||
const targets = { deduplication: "deduplicates a retried receipt", company: "isolates company reporting", projection: "conserves agent projections", threshold: "stops at the exact budget boundary" };
|
||||
const evidence = [];
|
||||
const persist = () => writeFile(path.join(root, "coverage/accounting/mutations.json"), JSON.stringify({ generatedAt: new Date().toISOString(), evidence }, null, 2));
|
||||
await mkdir(artifactDirectory, { recursive: true });
|
||||
try {
|
||||
for (const name of ["baseline", ...Object.keys(targets)]) {
|
||||
const report = path.join(directory, `${name}.json`);
|
||||
const result = await new Promise((resolve, reject) => {
|
||||
const entry = { mutation: name, result: "error", assertions: [] };
|
||||
evidence.push(entry);
|
||||
await persist();
|
||||
const result = await new Promise((resolve) => {
|
||||
const child = spawn(process.execPath, ["node_modules/vitest/vitest.mjs", "run", "--config", "server/vitest.accounting-mutations.config.ts", "--reporter=json", `--outputFile=${report}`], {
|
||||
cwd: root, env: { ...process.env, PAPERCLIP_ACCOUNTING_MUTATION: name === "baseline" ? "" : name }, stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
let log = ""; const capture = data => { log += data; };
|
||||
child.stdout.on("data", capture); child.stderr.on("data", capture);
|
||||
child.once("error", reject); child.once("exit", code => resolve({ code, log }));
|
||||
child.once("error", error => resolve({ code: null, log: `${log}\n${error.message}` }));
|
||||
child.once("close", code => resolve({ code, log }));
|
||||
});
|
||||
await writeFile(path.join(artifactDirectory, `${name}.log`), result.log);
|
||||
await copyFile(report, path.join(artifactDirectory, `${name}.json`)).catch(() => {});
|
||||
const json = JSON.parse(await readFile(report, "utf8").catch(() => { throw new Error(`${name} produced no test evidence: ${result.log}`); }));
|
||||
const assertions = json.testResults.flatMap(file => file.assertionResults);
|
||||
entry.assertions = assertions.map(({ title, status }) => ({ title, status }));
|
||||
if (name === "baseline") {
|
||||
if (result.code !== 0 || assertions.length !== 4 || assertions.some(test => test.status !== "passed")) throw new Error(`Mutation baseline failed or skipped: ${result.log}`);
|
||||
entry.result = result.code === 0 && assertions.length === 4 && assertions.every(test => test.status === "passed") ? "passed" : "baseline_failed";
|
||||
if (entry.result !== "passed") throw new Error(`Mutation baseline failed or skipped: ${result.log}`);
|
||||
} else {
|
||||
const sentinel = assertions.find(test => test.title === targets[name]);
|
||||
if (result.code === 0 || !result.log.includes(`ACCOUNTING_MUTATION_APPLIED ${name}`) || sentinel?.status !== "failed") throw new Error(`Mutation ${name} survived or failed without its sentinel: ${result.log}`);
|
||||
entry.result = result.code !== 0 && result.log.includes(`ACCOUNTING_MUTATION_APPLIED ${name}`) && sentinel?.status === "failed" ? "killed" : "survived_or_invalid";
|
||||
if (entry.result !== "killed") throw new Error(`Mutation ${name} survived or failed without its sentinel: ${result.log}`);
|
||||
}
|
||||
evidence.push({ mutation: name, result: name === "baseline" ? "passed" : "killed", assertions: assertions.map(({ title, status }) => ({ title, status })) });
|
||||
process.stdout.write(`${name}: ${name === "baseline" ? "passed" : "killed by expected assertion"}\n`);
|
||||
await persist();
|
||||
process.stdout.write(`${name}: ${entry.result}\n`);
|
||||
}
|
||||
await mkdir(path.join(root, "coverage/accounting"), { recursive: true });
|
||||
await writeFile(path.join(root, "coverage/accounting/mutations.json"), JSON.stringify({ generatedAt: new Date().toISOString(), evidence }, null, 2));
|
||||
} finally { await rm(directory, { recursive: true, force: true }); }
|
||||
} finally {
|
||||
// Failed baselines, surviving mutants, and missing reports are evidence too.
|
||||
try { await persist(); } finally { await rm(directory, { recursive: true, force: true }); }
|
||||
}
|
||||
@@ -4,7 +4,7 @@ import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { agents, agentRuntimeState, budgetReservations, companies, costEvents, costAdjustments, createDb, heartbeatRuns, runUsageReceipts } from "@paperclipai/db";
|
||||
import { agents, agentRuntimeState, budgetReservations, companies, costEvents, costAdjustments, createDb, heartbeatRuns, runUsageReceipts, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { accountingIntegrityService } from "../services/accounting-integrity.js";
|
||||
import { billingReconciliationService } from "../services/billing-reconciliation.js";
|
||||
import { budgetService } from "../services/budgets.js";
|
||||
@@ -65,6 +65,37 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
expect((await accountingIntegrityService(db).inspect(f.company.id)).findings).toEqual([]);
|
||||
});
|
||||
|
||||
it("reuses a held native reservation only for its current recovery lease", async () => {
|
||||
const f = await fixture(); const run = await runFor(f);
|
||||
const [issue] = await db.insert(issues).values({ companyId: f.company.id, title: "Recover" }).returning();
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, run.id));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId: f.company.id, runId: run.id, issueId: issue.id,
|
||||
phase: "running", leaseOwner: "successor", leaseExpiresAt: new Date(Date.now() + 60_000) });
|
||||
await budgetService(db).upsertPolicy(f.company.id, { scopeType: "company", scopeId: f.company.id, amount: 10, reservationCents: "6" }, "board");
|
||||
const original = await reserveRunBudget(db, f.company.id, run.id, null);
|
||||
await expect(reserveRunBudget(db, f.company.id, run.id, null)).rejects.toThrow("already reserved");
|
||||
await expect(reserveRunBudget(db, f.company.id, run.id, null, {}, "stale-owner")).rejects.toThrow("already reserved");
|
||||
expect(await reserveRunBudget(db, f.company.id, run.id, null, {}, "successor")).toEqual(original);
|
||||
expect((await accountingIntegrityService(db).health(f.company.id)).heldReservationCents).toBe("6.0000000");
|
||||
await db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
await expect(reserveRunBudget(db, f.company.id, run.id, null, {}, "successor")).rejects.toThrow("already reserved");
|
||||
});
|
||||
|
||||
it.each([false, true])("settles absent usage without hiding explicit unknown pricing (%s)", async (unpriced) => {
|
||||
const f = await fixture(); const run = await runFor(f);
|
||||
await budgetService(db).upsertPolicy(f.company.id, { scopeType: "agent", scopeId: f.agent.id, amount: 100, reservationCents: "10" }, "board");
|
||||
await reserveRunBudget(db, f.company.id, run.id, null);
|
||||
const recorder = await createRunUsageRecorder(db, { companyId: f.company.id, runId: run.id, adapterType: "process" }, directory);
|
||||
await recorder.complete({ exitCode: 0, signal: null, timedOut: false, ...(unpriced ? { provider: "moonshot", costStatus: "unpriced" as const, costUsd: null } : {}) });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await accountRunCost(db, run.id)).toBe(true);
|
||||
expect((await accountingIntegrityService(db).health(f.company.id)).heldReservationCents).toBe("0.0000000");
|
||||
const summary = await costService(db).summary(f.company.id);
|
||||
expect(summary.unpricedEventCount).toBe(unpriced ? 1 : 0);
|
||||
const block = await budgetService(db).getInvocationBlock(f.company.id, f.agent.id);
|
||||
if (unpriced) expect(block).not.toBeNull(); else expect(block).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps an interrupted partial receipt pending and settles its reservation when evidence becomes complete", async () => {
|
||||
const f = await fixture(); const run = await runFor(f);
|
||||
await budgetService(db).upsertPolicy(f.company.id, { scopeType: "agent", scopeId: f.agent.id, amount: 100, reservationCents: "10" }, "board");
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { upsertBudgetPolicySchema } from "@paperclipai/shared";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
@@ -288,6 +289,31 @@ databaseDescribe("cost accounting reliability (PostgreSQL)", () => {
|
||||
expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).toMatchObject({ scopeType: "company" });
|
||||
});
|
||||
|
||||
it("delivers cancellation after generic budget updates commit", async () => {
|
||||
const f = await fixture();
|
||||
await costService(db).createEvent(f.company.id, f.receipt);
|
||||
const cancelWorkForScope = vi.fn(async (scope) => {
|
||||
const policies = await budgetService(db).listPolicies(f.company.id);
|
||||
expect(policies.some(policy => policy.scopeType === scope.scopeType && policy.amount < 100)).toBe(true);
|
||||
});
|
||||
await services.value.agentService(db, { cancelWorkForScope }).update(f.agent.id, { budgetMonthlyCents: 50 });
|
||||
await services.value.companyService(db, { cancelWorkForScope }).update(f.company.id, { budgetMonthlyCents: 75 });
|
||||
expect(cancelWorkForScope).toHaveBeenCalledWith(expect.objectContaining({ scopeType: "agent", scopeId: f.agent.id }));
|
||||
expect(cancelWorkForScope).toHaveBeenCalledWith(expect.objectContaining({ scopeType: "company", scopeId: f.company.id }));
|
||||
const rows = await db.select().from(budgetPolicies).where(eq(budgetPolicies.companyId, f.company.id));
|
||||
expect(rows.every(row => row.enforcementVersion === row.enforcementDeliveredVersion)).toBe(true);
|
||||
});
|
||||
|
||||
it("merges partial policy edits with current settings under the accounting lock", async () => {
|
||||
const f = await fixture(); const budgets = budgetService(db);
|
||||
const identity = { scopeType: "agent" as const, scopeId: f.agent.id };
|
||||
await budgets.upsertPolicy(f.company.id, { ...identity, amount: 100, hardStopEnabled: true }, "board");
|
||||
await budgets.upsertPolicy(f.company.id, { ...identity, amount: 200, hardStopEnabled: false, warnPercent: 60, notifyEnabled: false }, "other-operator");
|
||||
const saved = await budgets.upsertPolicy(f.company.id, upsertBudgetPolicySchema.parse({ ...identity, reservationCents: "10" }), "board");
|
||||
expect(saved).toMatchObject({ amount: 200, hardStopEnabled: false, warnPercent: 60, notifyEnabled: false, reservationCents: "10.0000000" });
|
||||
await expect(budgets.upsertPolicy(f.company.id, { scopeType: "project", scopeId: f.project.id, reservationCents: "1" }, "board")).rejects.toThrow("Amount is required");
|
||||
});
|
||||
|
||||
it("conserves both spend and tokens across complete per-model receipts", async () => {
|
||||
const f = await fixture();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), costAccountingPending: true, usageJson: {
|
||||
|
||||
@@ -1078,7 +1078,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
model: "gpt-5.4",
|
||||
inputTokens: 5200,
|
||||
outputTokens: 980,
|
||||
costCents: 2,
|
||||
costCents: 0.1234567,
|
||||
occurredAt: new Date(),
|
||||
});
|
||||
const summaryWrite = await request(writerApp).put(`/api/status-cards/${created.body.id}/summary`).send({
|
||||
@@ -1111,7 +1111,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
summaryBody: "**Decide:** unblock launch approval.\n\n**Recent work:** launch review is waiting.",
|
||||
watchedIssueCount: 1,
|
||||
todayTokens: 6180,
|
||||
todayCostCents: 2,
|
||||
todayCostCents: 0.1234567,
|
||||
};
|
||||
const detail = await request(boardApp).get(`/api/status-cards/${created.body.id}`);
|
||||
expect(detail.status).toBe(200);
|
||||
@@ -1143,7 +1143,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
model: "gpt-5.4",
|
||||
inputTokens: 1300,
|
||||
outputTokens: 410,
|
||||
costCents: 1,
|
||||
costCents: 3_000_000_000.25,
|
||||
occurredAt: new Date(),
|
||||
});
|
||||
const incrementalWrite = await request(createApp(db, agentActor(company.id, summarizer.id, updateRun.id)))
|
||||
@@ -1155,7 +1155,7 @@ describeEmbeddedPostgres("status card routes", () => {
|
||||
model: "gpt-5.4",
|
||||
});
|
||||
expect(incrementalWrite.status).toBe(200);
|
||||
expect(await db.select().from(statusCardUpdates).then((rows) => rows.find((row) => row.kind === "incremental"))).toMatchObject({ inputTokens: 1300, outputTokens: 410 });
|
||||
expect(await db.select().from(statusCardUpdates).then((rows) => rows.find((row) => row.kind === "incremental"))).toMatchObject({ inputTokens: 1300, outputTokens: 410, costCents: 3_000_000_000.25 });
|
||||
const revisions = await request(boardApp).get(`/api/status-cards/${created.body.id}/summary-revisions`);
|
||||
expect(revisions.status).toBe(200);
|
||||
expect(revisions.body.map((row: { revisionNumber: number }) => row.revisionNumber)).toEqual([2, 1]);
|
||||
|
||||
@@ -539,7 +539,7 @@ export function agentRoutes(
|
||||
const KNOWN_INSTRUCTIONS_BUNDLE_KEY_SET: ReadonlySet<string> = new Set(KNOWN_INSTRUCTIONS_BUNDLE_KEYS);
|
||||
|
||||
const router = Router();
|
||||
const svc = agentService(db);
|
||||
const svc = agentService(db, { cancelWorkForScope: scope => heartbeat.cancelBudgetScopeWork(scope) });
|
||||
const access = accessService(db);
|
||||
const approvalsSvc = approvalService(db);
|
||||
const budgets = budgetService(db);
|
||||
|
||||
@@ -55,6 +55,7 @@ import {
|
||||
accessService,
|
||||
agentService,
|
||||
budgetService,
|
||||
heartbeatService,
|
||||
buildExportFidelityReport,
|
||||
collectExportFidelityCounts,
|
||||
companyArtifactsService,
|
||||
@@ -278,7 +279,8 @@ export function companyRoutes(db: Db, storage?: StorageService, options?: Compan
|
||||
const router = Router();
|
||||
const importTransferSpoolRoot =
|
||||
options?.importTransferSpoolRoot ?? resolveDefaultImportTransferSpoolRoot();
|
||||
const svc = companyService(db);
|
||||
const heartbeat = heartbeatService(db);
|
||||
const svc = companyService(db, { cancelWorkForScope: heartbeat.cancelBudgetScopeWork });
|
||||
const agents = agentService(db);
|
||||
const portability = companyPortabilityService(db, storage);
|
||||
const access = accessService(db);
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { budgetServiceInTransaction } from "./budgets.js";
|
||||
import { budgetServiceInTransaction, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js";
|
||||
import { withAccountingTransaction } from "./accounting-transaction.js";
|
||||
import type { ActivityPublication } from "./activity-log.js";
|
||||
import { agentAppearanceSchema, randomAgentAppearance, resolveAgentAppearance, agentAvatarUrl } from "@paperclipai/shared";
|
||||
@@ -341,7 +341,7 @@ export function deduplicateAgentName(
|
||||
return `${candidateName} ${Date.now()}`;
|
||||
}
|
||||
|
||||
export function agentService(db: Db) {
|
||||
export function agentService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
|
||||
const secretsSvc = secretService(db);
|
||||
|
||||
function currentUtcMonthWindow(now = new Date()) {
|
||||
@@ -860,7 +860,11 @@ export function agentService(db: Db) {
|
||||
return normalizedUpdated;
|
||||
};
|
||||
|
||||
if (normalizedPatch.budgetMonthlyCents !== undefined) return withAccountingTransaction(db, existing.companyId, applyUpdate);
|
||||
if (normalizedPatch.budgetMonthlyCents !== undefined) {
|
||||
const result = await withAccountingTransaction(db, existing.companyId, applyUpdate);
|
||||
await deliverBudgetEnforcement(db, budgetHooks, existing.companyId);
|
||||
return result;
|
||||
}
|
||||
|
||||
const transaction = (db as unknown as {
|
||||
transaction?: (callback: (tx: unknown) => Promise<AgentUpdateResult>) => Promise<AgentUpdateResult>;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { budgetPolicies, budgetReservations, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { budgetPolicies, budgetReservations, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { centsToUnits, unitsToCents } from "@paperclipai/shared";
|
||||
import { conflict, notFound } from "../errors.js";
|
||||
import { withAccountingTransaction } from "./accounting-transaction.js";
|
||||
@@ -8,13 +8,23 @@ import { budgetServiceInTransaction, computeObservedSpend } from "./budgets.js";
|
||||
/** Reserve before dispatch, under the same company lock as charges and policy
|
||||
* changes. Estimates constrain admission; they cannot cap a provider's bill.
|
||||
* A reservation survives timeouts and restarts until accounting proves closure. */
|
||||
export async function reserveRunBudget(db: Db, companyId: string, runId: string, projectId: string | null, ledgerScope: Record<string, unknown> = {}) {
|
||||
export async function reserveRunBudget(db: Db, companyId: string, runId: string, projectId: string | null, ledgerScope: Record<string, unknown> = {}, recoveryLeaseOwner?: string) {
|
||||
return withAccountingTransaction(db, companyId, async (tx, publications) => {
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).for("update");
|
||||
if (!run) throw notFound("Run not found");
|
||||
if (run.costAccountedAt || !["queued", "running"].includes(run.status)) throw conflict("Run can no longer start provider work");
|
||||
const [existing] = await tx.select().from(budgetReservations).where(and(eq(budgetReservations.companyId, companyId), eq(budgetReservations.runId, runId)));
|
||||
if (existing) throw conflict("Provider dispatch has already reserved this run");
|
||||
if (existing) {
|
||||
// The native coordinator fences recovery ownership. Reuse the original
|
||||
// hold only for that live owner, never for an ordinary duplicate dispatch.
|
||||
const [owner] = recoveryLeaseOwner && run.runtimeMode === "native" && existing.state === "held"
|
||||
&& existing.projectId === projectId ? await tx.select({ runId: nativeRunFinalizations.runId })
|
||||
.from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId),
|
||||
eq(nativeRunFinalizations.runId, runId), eq(nativeRunFinalizations.leaseOwner, recoveryLeaseOwner),
|
||||
sql`${nativeRunFinalizations.leaseExpiresAt} > now()`)).for("update") : [];
|
||||
if (owner) return existing;
|
||||
throw conflict("Provider dispatch has already reserved this run");
|
||||
}
|
||||
const block = await budgetServiceInTransaction(tx, publications).getInvocationBlock(companyId, run.agentId, { projectId });
|
||||
if (block) throw conflict(block.reason);
|
||||
const candidates = await tx.select().from(budgetPolicies).where(and(eq(budgetPolicies.companyId, companyId), eq(budgetPolicies.isActive, true), eq(budgetPolicies.hardStopEnabled, true)));
|
||||
|
||||
@@ -611,8 +611,6 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
if (input.reservationCents !== undefined && compareCents(input.reservationCents, 0) < 0) throw unprocessable("Reservation cannot be negative");
|
||||
const metric = input.metric ?? "billed_cents";
|
||||
const windowKind = input.windowKind ?? (input.scopeType === "project" ? "lifetime" : "calendar_month_utc");
|
||||
const amount = Math.max(0, Math.floor(input.amount));
|
||||
const nextIsActive = amount > 0 && (input.isActive ?? true);
|
||||
const existing = await db
|
||||
.select()
|
||||
.from(budgetPolicies)
|
||||
@@ -627,6 +625,9 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (!existing && input.amount === undefined) throw unprocessable("Amount is required for a new budget policy");
|
||||
const amount = Math.max(0, Math.floor(input.amount ?? existing!.amount));
|
||||
const nextIsActive = amount > 0 && (input.isActive ?? (input.amount !== undefined ? true : existing?.isActive ?? true));
|
||||
const now = new Date();
|
||||
const row = existing
|
||||
? await db
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { publishAccountingActivities } from "./accounting-transaction.js";
|
||||
import { budgetServiceInTransaction } from "./budgets.js";
|
||||
import { budgetServiceInTransaction, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js";
|
||||
import { and, count, eq, gte, inArray, isNull, lt, notInArray, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
@@ -68,7 +68,7 @@ const SYSTEM_COMPANY_ACTOR: CompanyActivityActor = {
|
||||
runId: null,
|
||||
};
|
||||
|
||||
export function companyService(db: Db) {
|
||||
export function companyService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
|
||||
const environmentsSvc = environmentService(db);
|
||||
const heartbeat = heartbeatService(db);
|
||||
const builtInAgents = builtInAgentService(db);
|
||||
@@ -462,6 +462,7 @@ export function companyService(db: Db) {
|
||||
});
|
||||
if (!result) return null;
|
||||
publishAccountingActivities(id, budgetPublications);
|
||||
if (data.budgetMonthlyCents !== undefined) await deliverBudgetEnforcement(db, budgetHooks, id);
|
||||
// Post-commit, fire-and-forget, and BEFORE any finalization that
|
||||
// could throw: a Cloud-pinned primary company that crossed the
|
||||
// archived boundary (either direction) rings the harness so the
|
||||
|
||||
@@ -24657,7 +24657,7 @@ export function heartbeatService(
|
||||
const guardedDispatch =
|
||||
await dispatchResolvedInteractionContinuationWithAtomicGate(
|
||||
async (markDispatchStarted) => {
|
||||
await reserveRunBudget(db, run.companyId, run.id, readNonEmptyString(runLedgerScope.projectId), runLedgerScope);
|
||||
await reserveRunBudget(db, run.companyId, run.id, readNonEmptyString(runLedgerScope.projectId), runLedgerScope, runOptions.nativeLeaseOwner);
|
||||
return executePaperclipNativeSession({
|
||||
db,
|
||||
execution: nativeExecution,
|
||||
|
||||
@@ -50,6 +50,10 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
|
||||
const billingType = text(usage.billingType) ?? "unknown";
|
||||
const costUnits = billingType === "subscription_included" ? 0n : usdToUnits(costUsd ?? 0);
|
||||
const costCents = unitsToCents(costUnits);
|
||||
const [agent] = await tx.select({ adapterType: agents.adapterType }).from(agents).where(eq(agents.id, run.agentId));
|
||||
const hasBillableEvidence = agent?.adapterType !== "process" || costUsd !== null || usage.costStatus === "unpriced"
|
||||
|| ["inputTokens", "cachedInputTokens", "outputTokens"].some(key => (amount(usage[key]) ?? 0) > 0)
|
||||
|| (text(usage.provider) !== null && usage.provider !== "unknown");
|
||||
const receipt = {
|
||||
idempotencyKey: `heartbeat:${run.id}:final`,
|
||||
agentId: run.agentId, heartbeatRunId: run.id,
|
||||
@@ -58,7 +62,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
|
||||
providerRequestId: text(usage.providerRequestId),
|
||||
pricingProvenance: object(usage.pricingProvenance),
|
||||
model: text(usage.model) ?? "unknown", billingType,
|
||||
costStatus: costUsd === null && billingType !== "subscription_included" ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported",
|
||||
costStatus: costUsd === null && billingType !== "subscription_included" && hasBillableEvidence ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported",
|
||||
inputTokens: amount(usage.inputTokens) ?? 0, cachedInputTokens: amount(usage.cachedInputTokens) ?? 0, outputTokens: amount(usage.outputTokens) ?? 0,
|
||||
costCents, occurredAt: run.finishedAt ?? run.createdAt,
|
||||
};
|
||||
@@ -93,7 +97,6 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
|
||||
outputTokens: sql<string>`coalesce(sum(${costEvents.outputTokens}),0)::text`,
|
||||
costCents: sql<string>`coalesce(sum(${costEvents.costCents}),0)::text`,
|
||||
}).from(costEvents).where(and(eq(costEvents.companyId, run.companyId), eq(costEvents.heartbeatRunId, run.id)));
|
||||
const [agent] = await tx.select({ adapterType: agents.adapterType }).from(agents).where(eq(agents.id, run.agentId));
|
||||
await tx.insert(agentRuntimeState).values({ agentId: run.agentId, companyId: run.companyId, adapterType: agent.adapterType }).onConflictDoNothing();
|
||||
await tx.execute(sql`insert into accounting_runtime_baselines (agent_id,company_id,cost_cents,input_tokens,cached_input_tokens,output_tokens)
|
||||
select agent_id,company_id,total_cost_cents,total_input_tokens,total_cached_input_tokens,total_output_tokens
|
||||
|
||||
@@ -179,7 +179,7 @@ export function statusCardService(
|
||||
: Promise.resolve(null),
|
||||
db.select({
|
||||
tokens: sql<number>`coalesce(sum(coalesce(${statusCardUpdates.inputTokens}, 0) + coalesce(${statusCardUpdates.outputTokens}, 0)), 0)::int`,
|
||||
costCents: sql<number>`coalesce(sum(${statusCardUpdates.costCents}), 0)::int`,
|
||||
costCents: sql<number>`coalesce(sum(${statusCardUpdates.costCents}), 0)::double precision`,
|
||||
})
|
||||
.from(statusCardUpdates)
|
||||
.where(and(eq(statusCardUpdates.cardId, card.id), gte(statusCardUpdates.startedAt, dayStart)))
|
||||
@@ -877,7 +877,7 @@ export function statusCardService(
|
||||
? await tx.select({
|
||||
inputTokens: sql<number>`coalesce(sum(${costEvents.inputTokens}), 0)::int`,
|
||||
outputTokens: sql<number>`coalesce(sum(${costEvents.outputTokens}), 0)::int`,
|
||||
costCents: sql<number>`coalesce(sum(${costEvents.costCents}), 0)::double precision`,
|
||||
costCents: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
|
||||
}).from(costEvents).where(eq(costEvents.heartbeatRunId, actor.runId))
|
||||
: [];
|
||||
const existingUpdate = await tx.select().from(statusCardUpdates)
|
||||
@@ -892,7 +892,7 @@ export function statusCardService(
|
||||
changeSummary: input.changeSummary,
|
||||
inputTokens: Number(usage[0]?.inputTokens ?? 0),
|
||||
outputTokens: Number(usage[0]?.outputTokens ?? 0),
|
||||
costCents: Number(usage[0]?.costCents ?? 0),
|
||||
costCents: sql`${usage[0]?.costCents ?? "0"}::numeric`,
|
||||
};
|
||||
if (existingUpdate) {
|
||||
await tx.update(statusCardUpdates).set(updateValues).where(eq(statusCardUpdates.id, existingUpdate.id));
|
||||
|
||||
@@ -16,6 +16,7 @@ import { api } from "./client";
|
||||
|
||||
function dateParams(from?: string, to?: string): string {
|
||||
const params = new URLSearchParams();
|
||||
if (!from && !to) params.set("period", "all");
|
||||
if (from) params.set("from", from);
|
||||
if (to) params.set("to", to);
|
||||
const qs = params.toString();
|
||||
@@ -51,6 +52,7 @@ export const costsApi = {
|
||||
|
||||
function dateParamsWithLimit(from?: string, to?: string, limit?: number): string {
|
||||
const params = new URLSearchParams();
|
||||
if (!from && !to) params.set("period", "all");
|
||||
if (from) params.set("from", from);
|
||||
if (to) params.set("to", to);
|
||||
if (limit) params.set("limit", String(limit));
|
||||
|
||||
Reference in new issue
Block a user