From 356708b0e1069e49239051def1433be4ca919411 Mon Sep 17 00:00:00 2001 From: Aaron Trowbridge Date: Wed, 4 Nov 2026 16:37:21 -0500 Subject: [PATCH] feat(amico-run): attribute CPU cost to principal and Lab from the solve unit cgroup Measure a local solve's real CPU from its transient unit's cgroup (cpu.stat usage_usec delta, startRun to settle) instead of the engine process clock. Emit a local receipt with cpu_seconds, principal, lab, and a cost_usd priced from ~/.amico/amicode/rates.json (list_price_estimate); unmeasured when no cgroup. Add costTotals rollup by principal and Lab. Accounting never fails a run. --- packages/amico-run/src/cost.ts | 114 +++++++++++++ packages/amico-run/src/ledger.ts | 56 +++++- packages/amico-run/src/local_executor.ts | 39 ++++- packages/amico-run/src/placement.ts | 2 +- packages/amico-run/src/types.ts | 6 + packages/amico-run/test/cost.test.ts | 161 ++++++++++++++++++ .../test/local_executor_cost.test.ts | 157 +++++++++++++++++ .../test/local_executor_ledger.test.ts | 6 +- packages/amico-run/test/placement.test.ts | 3 +- packages/amico-run/test/setup.ts | 5 + .../test/slow/cost_unit.slow.test.ts | 101 +++++++++++ .../schema/schemas/ledger-record.schema.json | 25 ++- packages/schema/test/ledger-record.test.ts | 43 ++++- 13 files changed, 707 insertions(+), 11 deletions(-) create mode 100644 packages/amico-run/src/cost.ts create mode 100644 packages/amico-run/test/cost.test.ts create mode 100644 packages/amico-run/test/local_executor_cost.test.ts create mode 100644 packages/amico-run/test/slow/cost_unit.slow.test.ts diff --git a/packages/amico-run/src/cost.ts b/packages/amico-run/src/cost.ts new file mode 100644 index 000000000..1f5387307 --- /dev/null +++ b/packages/amico-run/src/cost.ts @@ -0,0 +1,114 @@ +// CPU cost attribution for a local solve (central-hub cost ledger, ADR 0041 §5; #2133). +// +// WHAT IS MEASURED. A solve's real CPU is burned by the Julia child (and its threads), NOT by +// the engine/orchestrator process: `process.cpuUsage()` sees only the latter and would cost every +// solve at ~zero, so it is deliberately never used here (a test greps for it). Since #2045 each +// solve runs in its own transient unit `amico-run-.service`; that unit's cgroup accounts the +// CPU of every process in it, children and threads included. +// +// HOW. `cpu.stat` `usage_usec` of the unit's cgroup (cgroup v2), read once as early as practical +// (startRun, before Julia is spawned) and once at settle (before the unit is torn down); the +// delta is `cpu_seconds`. We read the file rather than `systemctl show -p CPUUsageNSec` because +// (a) it needs no CPUAccounting property on the unit — v2 always maintains usage_usec — and +// (b) after the first lookup it is one file read, with no subprocess at settle. The cgroup +// directory is resolved ONCE, via `systemctl --user show -p ControlGroup --value `. +// +// No unit / no cgroup (macOS, protection:"none"): nothing is measured and the record says +// `cost_basis: "unmeasured"` — no estimate, no fallback to engine-process CPU. +// +// Accounting never fails a run: every function here is total (returns undefined on any fault). +import { spawnSync } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { managerEnv } from "./placement.js"; +import type { ReceiptRecord } from "./ledger.js"; + +const CGROUP_ROOT = "/sys/fs/cgroup"; + +/** `usage_usec` from the text of a cgroup v2 `cpu.stat`, or undefined. */ +export function parseUsageUsec(cpuStat: string): number | undefined { + const m = /^usage_usec\s+(\d+)\s*$/m.exec(cpuStat); + return m ? Number(m[1]) : undefined; +} + +/** Absolute cgroup directory of a systemd user unit, or undefined (not Linux, unit gone, no bus). */ +export function resolveUnitCgroupDir(unit: string, env: NodeJS.ProcessEnv = process.env): string | undefined { + if (process.platform !== "linux") return undefined; + const r = spawnSync("systemctl", ["--user", "show", "-p", "ControlGroup", "--value", unit], { + env: managerEnv(env), + encoding: "utf8", + timeout: 10000, + }); + const cg = r.status === 0 ? r.stdout.trim() : ""; + return cg.startsWith("/") ? join(CGROUP_ROOT, cg) : undefined; +} + +export interface CpuMeter { + /** CPU-seconds consumed in the cgroup since the meter started; undefined if it cannot be read. */ + stop(): number | undefined; +} + +const NO_METER: CpuMeter = { stop: () => undefined }; + +/** Start metering. `cgroupDir` (a directory holding `cpu.stat`) wins over `unit`; it is the test + * seam and also lets a caller that already knows the cgroup skip the systemctl lookup. With + * neither, the run simply is not measured (silently: that is the macOS / unplaced case). A + * measurement that was expected and failed is reported on stderr, once per meter. */ +export function startCpuMeter(src: { unit?: string | null; cgroupDir?: string; env?: NodeJS.ProcessEnv }): CpuMeter { + if (!src.cgroupDir && !src.unit) return NO_METER; + let warned = false; + const warn = (why: string): undefined => { + if (!warned) process.stderr.write(`amico-run: cpu accounting unavailable (${why}); run unaffected\n`); + warned = true; + return undefined; + }; + try { + const dir = src.cgroupDir ?? resolveUnitCgroupDir(src.unit!, src.env); + if (!dir) return { stop: () => warn("no cgroup for the run's unit") }; + const read = (): number | undefined => parseUsageUsec(readFileSync(join(dir, "cpu.stat"), "utf8")); + const start = read(); + if (start === undefined) return { stop: () => warn("no usage_usec in cpu.stat") }; + return { + stop() { + try { + const end = read(); + return end === undefined || end < start ? warn("cpu.stat went backwards or lost usage_usec") : (end - start) / 1e6; + } catch (e) { + return warn((e as Error).message); + } + }, + }; + } catch (e) { + return { stop: () => warn((e as Error).message) }; + } +} + +/** `~/.amico/amicode/rates.json` (human-edited), or `$AMICO_RATES_FILE`. */ +export function ratesPath(env: NodeJS.ProcessEnv = process.env): string { + return env.AMICO_RATES_FILE || join(homedir(), ".amico", "amicode", "rates.json"); +} + +/** `cpu_usd_per_second` from the rates file; undefined when absent, malformed, or not a positive + * finite number — an unconfigured install records seconds and no invented dollars. */ +export function readCpuRate(path: string = ratesPath()): number | undefined { + try { + const v = (JSON.parse(readFileSync(path, "utf8")) as { cpu_usd_per_second?: unknown } | null)?.cpu_usd_per_second; + return typeof v === "number" && Number.isFinite(v) && v > 0 ? v : undefined; + } catch { + return undefined; + } +} + +/** The cost fields of a local receipt. Per-record values are NOT rounded (rollups round). */ +export function costFields( + cpuSeconds: number | undefined, + rate: number | undefined, +): Pick { + if (cpuSeconds === undefined) return { cost_basis: "unmeasured" }; + const cost = rate === undefined ? 0 : cpuSeconds * rate; + // schema: cost_usd exclusiveMinimum 0 — omit a zero (or non-finite) cost rather than violate it + return cost > 0 && Number.isFinite(cost) + ? { cpu_seconds: cpuSeconds, cost_usd: cost, cost_basis: "list_price_estimate" } + : { cpu_seconds: cpuSeconds }; +} diff --git a/packages/amico-run/src/ledger.ts b/packages/amico-run/src/ledger.ts index d71f951b7..e0362d913 100644 --- a/packages/amico-run/src/ledger.ts +++ b/packages/amico-run/src/ledger.ts @@ -274,10 +274,21 @@ export interface ReceiptRecord { type: "receipt"; ts: string; task_id: string; - executor: "remote"; + /** `local` = CPU burned by a solve on this host (#2133); `remote` = cloud GPU solve. */ + executor: "local" | "remote"; gpu_sku?: string; gpu_seconds?: number; + /** CPU-seconds of a local solve, measured from its unit's cgroup (never process.cpuUsage()). + * Absent when no cgroup was available — see cost_basis. */ + cpu_seconds?: number; cost_usd?: number; + /** `list_price_estimate` = cost_usd is cpu_seconds × the configured list rate (not an invoice); + * `unmeasured` = no cgroup, so no seconds and no dollars. */ + cost_basis?: "list_price_estimate" | "unmeasured"; + /** Person principal the run is charged to (central-hub attribution). */ + principal?: string; + /** Lab the run is charged to. */ + lab?: string; problem?: string; session?: string; status?: "completed" | "failed" | "aborted"; @@ -341,7 +352,7 @@ export function gpuTotals(file = ledgerPath()): GpuTotals { } catch { continue; // a torn line never breaks accounting } - if (rec.type !== "receipt") continue; + if (rec.type !== "receipt" || rec.executor === "local") continue; // local CPU rows: see costTotals t.receipts += 1; if (rec.gpu_seconds !== undefined) t.gpu_seconds += rec.gpu_seconds; if (rec.cost_usd !== undefined) t.cost_usd += rec.cost_usd; @@ -357,6 +368,47 @@ export function gpuTotals(file = ledgerPath()): GpuTotals { return t; } +export interface CostBucket { + receipts: number; + cpu_seconds: number; + cost_usd: number; + /** receipts with no cpu_seconds (nothing could be measured) */ + unmeasured: number; +} + +export interface CostTotals extends CostBucket { + by_principal: Record; + by_lab: Record; +} + +export const UNATTRIBUTED = "(unattributed)"; + +/** CPU cost rollup over LOCAL receipt rows, by principal and by Lab (#2133). Pure; never throws; + * zeros on empty. A row with no principal / lab rolls up under `(unattributed)`, so each + * breakdown sums to the total. Rounding to 1e-4 happens here only — per-record values stay raw. */ +export function costTotals(records: readonly LedgerRecord[]): CostTotals { + const bucket = (): CostBucket => ({ receipts: 0, cpu_seconds: 0, cost_usd: 0, unmeasured: 0 }); + const t: CostTotals = { ...bucket(), by_principal: {}, by_lab: {} }; + const add = (b: CostBucket, r: ReceiptRecord): void => { + b.receipts += 1; + if (r.cpu_seconds === undefined) b.unmeasured += 1; + else b.cpu_seconds += r.cpu_seconds; + if (r.cost_usd !== undefined) b.cost_usd += r.cost_usd; + }; + for (const rec of records) { + if (rec.type !== "receipt" || rec.executor !== "local") continue; + add(t, rec); + add((t.by_principal[rec.principal ?? UNATTRIBUTED] ??= bucket()), rec); + add((t.by_lab[rec.lab ?? UNATTRIBUTED] ??= bucket()), rec); + } + const round = (b: CostBucket): void => { + b.cpu_seconds = Math.round(b.cpu_seconds * 1e4) / 1e4; + b.cost_usd = Math.round(b.cost_usd * 1e4) / 1e4; + }; + [t, ...Object.values(t.by_principal), ...Object.values(t.by_lab)].forEach(round); + return t; +} + /** Append one record as a single JSONL line. Validates against the `ledger-record` * schema (Task 2 wires this in) before writing so no malformed line ever lands. * Throws when the serialized line exceeds PIPE_BUF (would break O_APPEND diff --git a/packages/amico-run/src/local_executor.ts b/packages/amico-run/src/local_executor.ts index 87bee13ce..f69d4a7f7 100644 --- a/packages/amico-run/src/local_executor.ts +++ b/packages/amico-run/src/local_executor.ts @@ -5,8 +5,9 @@ import { delimiter, join, resolve } from "node:path"; import * as readline from "node:readline"; import { parse as tomlParse, stringify as tomlStringify } from "smol-toml"; import { EventQueue } from "./event_queue.js"; -import { appendRecord, type SolveOutcome, type SolveRecord, type SolveSummary } from "./ledger.js"; +import { appendRecord, type ReceiptRecord, type SolveOutcome, type SolveRecord, type SolveSummary } from "./ledger.js"; import { classifyLine } from "./telemetry.js"; +import { costFields, readCpuRate, startCpuMeter } from "./cost.js"; import { LOCK_HEARTBEAT_MS, startRunLock } from "./run_lock.js"; import { writePendingVerification } from "./verify.js"; import { @@ -230,6 +231,34 @@ function emitSolveStanza(runDir: string): void { } } +/** The `local` receipt (#2133): CPU-seconds from the unit cgroup, priced by the human-edited rate + * file, attributed to a principal and Lab. Never throws and never touches the run's verdict — + * accounting failure is logged once and the run is unaffected. */ +function emitCpuReceipt( + runId: string, + status: RunStatus, + cpuSeconds: number | undefined, + cost: NonNullable, +): void { + try { + const rec: ReceiptRecord = { + type: "receipt", + ts: new Date().toISOString(), + task_id: runId, + executor: "local", + ...costFields(cpuSeconds, readCpuRate(cost.ratesPath)), + status, + }; + const principal = cost.principal ?? process.env.AMICO_PRINCIPAL; + const lab = cost.lab ?? process.env.AMICO_LAB; + if (principal) rec.principal = principal; + if (lab) rec.lab = lab; + appendRecord(rec); + } catch (e) { + process.stderr.write(`amico-run: failed to emit cpu receipt: ${(e as Error).message}\n`); + } +} + /** A run that has its directory, manifest, index entry and `latest` pointer — everything the * manifest-FIRST contract promises — but whose Julia child has not been spawned yet. * `prepareRun` produces it; `startRun` (in this process, or in an `--adopt` orchestrator @@ -346,6 +375,11 @@ export class LocalExecutor implements Executor { opts.heartbeatMs ?? (Number(process.env.AMICO_RUN_HEARTBEAT_MS) || LOCK_HEARTBEAT_MS), ); + // CPU meter (#2133): first reading before Julia exists, second at settle, both from the unit's + // cgroup. Total — a meter that cannot read yields undefined, never an exception. + const cost = opts.cost ?? {}; + const meter = startCpuMeter({ unit: opts.lock?.unit, cgroupDir: cost.cgroupDir }); + const events = new EventQueue(); const logStream = createWriteStream(join(runDir, "run.log"), { flags: "a" }); let resolveFinished!: (f: Finished) => void; @@ -358,6 +392,8 @@ export class LocalExecutor implements Executor { const settle = (status: RunStatus, exitCode: number): void => { if (settled) return; settled = true; + // Read the cgroup NOW: Julia has exited and the unit is still up (teardown follows FINISHED). + const cpuSeconds = meter.stop(); // FLUSH run.log BEFORE the verdict, then write the verdict, then announce. // @@ -390,6 +426,7 @@ export class LocalExecutor implements Executor { // any FINISHED a script faked (spec §5 step 8) process.stderr.write(`amico-run: failed to write FINISHED: ${(e as Error).message}\n`); } + emitCpuReceipt(runId, status, cpuSeconds, cost); // #2133 — never throws emitSolveStanza(runDir); // Plan 3 / L1 Task 5 — never throws; a ledger hiccup must never fail a run events.push({ kind: "finished", status, exitCode }); events.close(); diff --git a/packages/amico-run/src/placement.ts b/packages/amico-run/src/placement.ts index 6e0322bc5..2674dca92 100644 --- a/packages/amico-run/src/placement.ts +++ b/packages/amico-run/src/placement.ts @@ -28,7 +28,7 @@ export function unitPrefix(env: NodeJS.ProcessEnv = process.env): string { } /** Env for the systemd-run / setsid subprocess: the user bus must be discoverable. */ -function managerEnv(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv { +export function managerEnv(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv { const out = { ...env }; delete out.BUN_INSPECT; if (!out.XDG_RUNTIME_DIR) { diff --git a/packages/amico-run/src/types.ts b/packages/amico-run/src/types.ts index 9daf1c3fb..d0905afa2 100644 --- a/packages/amico-run/src/types.ts +++ b/packages/amico-run/src/types.ts @@ -15,6 +15,12 @@ export interface SubmitOpts { /** #2045: how the orchestrator that owns this run is placed, recorded in run.lock. * Default: in-process, unprotected (`none`, no unit). */ lock?: { protection: "unit" | "none"; unit?: string | null }; + /** #2133: cost attribution. `principal` / `lab` default to $AMICO_PRINCIPAL / $AMICO_LAB (the + * future hub frontdoor sets them); absent ⇒ rolled up as `(unattributed)`. NOTE `lab` here is the + * Lab the spend is charged to — not the hardware-lab pointer in `SubmitOpts.lab`. `ratesPath` + * overrides ~/.amico/amicode/rates.json; `cgroupDir` overrides the unit's cgroup (a dir holding + * `cpu.stat`) — both are test seams. */ + cost?: { principal?: string; lab?: string; ratesPath?: string; cgroupDir?: string }; heartbeatMs?: number; // run.lock heartbeat; default 30 s. Test knob, NOT exposed in the CLI. } diff --git a/packages/amico-run/test/cost.test.ts b/packages/amico-run/test/cost.test.ts new file mode 100644 index 000000000..ffd7eca01 --- /dev/null +++ b/packages/amico-run/test/cost.test.ts @@ -0,0 +1,161 @@ +// #2133 — cost attribution units: cgroup cpu.stat parsing + delta, the rate file, the cost fields, +// the rollups, and the guard that the engine-process clock is never the source of a solve's CPU. +import { describe, it, expect } from "vitest"; +import { mkdtempSync, readFileSync, writeFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { costFields, parseUsageUsec, readCpuRate, startCpuMeter } from "../src/cost.js"; +import { costTotals, gpuTotals, type LedgerRecord, type ReceiptRecord } from "../src/ledger.js"; + +const tmp = (): string => mkdtempSync(join(tmpdir(), "amico-cost-")); +const cpuStat = (usec: number): string => + `usage_usec ${usec}\nuser_usec ${usec - 10}\nsystem_usec 10\nnice_usec 0\ncore_sched.force_idle_usec 0\nnr_periods 0\n`; + +describe("cgroup cpu.stat reading", () => { + it("parses usage_usec from a real-shaped cpu.stat (and not user_usec/system_usec)", () => { + expect(parseUsageUsec(cpuStat(1844))).toBe(1844); + }); + it("returns undefined for text without usage_usec", () => { + expect(parseUsageUsec("")).toBeUndefined(); + expect(parseUsageUsec("user_usec 5\nsystem_usec 5\n")).toBeUndefined(); + expect(parseUsageUsec("usage_usec banana\n")).toBeUndefined(); + }); + it("the meter reports the delta between the start reading and the stop reading, in seconds", () => { + const dir = tmp(); + writeFileSync(join(dir, "cpu.stat"), cpuStat(2_000_000)); + const m = startCpuMeter({ cgroupDir: dir }); + writeFileSync(join(dir, "cpu.stat"), cpuStat(3_750_000)); + expect(m.stop()).toBeCloseTo(1.75, 6); + }); + it("a cgroup that already had CPU history only counts what accrued after the start reading", () => { + const dir = tmp(); + writeFileSync(join(dir, "cpu.stat"), cpuStat(9_000_000_000)); + const m = startCpuMeter({ cgroupDir: dir }); + expect(m.stop()).toBe(0); + }); + it("no unit and no cgroup dir ⇒ not measured (undefined), without a warning", () => { + expect(startCpuMeter({}).stop()).toBeUndefined(); + expect(startCpuMeter({ unit: null }).stop()).toBeUndefined(); + }); + it("a vanished or unreadable cpu.stat ⇒ undefined, never a throw", () => { + const dir = tmp(); + writeFileSync(join(dir, "cpu.stat"), cpuStat(1_000_000)); + const m = startCpuMeter({ cgroupDir: dir }); + rmSync(dir, { recursive: true }); + expect(m.stop()).toBeUndefined(); + expect(startCpuMeter({ cgroupDir: join(dir, "never-existed") }).stop()).toBeUndefined(); + }); + it("a cpu.stat that goes backwards (cgroup recreated) is not reported as negative CPU", () => { + const dir = tmp(); + writeFileSync(join(dir, "cpu.stat"), cpuStat(5_000_000)); + const m = startCpuMeter({ cgroupDir: dir }); + writeFileSync(join(dir, "cpu.stat"), cpuStat(100)); + expect(m.stop()).toBeUndefined(); + }); +}); + +describe("the rate file", () => { + const rates = (body: string): string => { + const f = join(tmp(), "rates.json"); + writeFileSync(f, body); + return f; + }; + it("reads cpu_usd_per_second", () => { + expect(readCpuRate(rates('{"cpu_usd_per_second": 0.00005}'))).toBe(0.00005); + }); + it("absent file ⇒ no rate", () => { + expect(readCpuRate(join(tmp(), "nope.json"))).toBeUndefined(); + }); + it("malformed or wrongly-typed content ⇒ no rate, never a throw", () => { + for (const body of ["{not json", "", "null", "[]", '{"cpu_usd_per_second": "0.1"}', '{"cpu_usd_per_second": -1}', '{"cpu_usd_per_second": 0}', '{"other": 1}']) + expect(readCpuRate(rates(body)), body).toBeUndefined(); + }); +}); + +describe("costFields", () => { + it("no measurement ⇒ cost_basis unmeasured, no seconds, no dollars — even with a rate", () => { + expect(costFields(undefined, 0.01)).toEqual({ cost_basis: "unmeasured" }); + }); + it("seconds but no rate ⇒ seconds only (no invented dollars, no basis)", () => { + expect(costFields(2.5, undefined)).toEqual({ cpu_seconds: 2.5 }); + }); + it("seconds and a rate ⇒ cost_usd = seconds × rate, basis list_price_estimate, unrounded", () => { + const f = costFields(1.234567, 0.00003); + expect(f.cpu_seconds).toBe(1.234567); + expect(f.cost_usd).toBe(1.234567 * 0.00003); + expect(f.cost_basis).toBe("list_price_estimate"); + }); + it("a zero cost is omitted (schema exclusiveMinimum), the measured seconds are kept", () => { + expect(costFields(0, 0.01)).toEqual({ cpu_seconds: 0 }); + }); +}); + +describe("costTotals", () => { + const rec = (over: Partial): ReceiptRecord => ({ + type: "receipt", + ts: "2026-11-04T00:00:00Z", + task_id: "t", + executor: "local", + ...over, + }); + const rows: LedgerRecord[] = [ + rec({ principal: "ana", lab: "labA", cpu_seconds: 10.00004, cost_usd: 0.50002 }), + rec({ principal: "ana", lab: "labB", cpu_seconds: 5, cost_usd: 0.25 }), + rec({ principal: "bo", lab: "labA", cpu_seconds: 1, cost_usd: 0.05 }), + rec({ cpu_seconds: 2 }), // unattributed, no rate + rec({ cost_basis: "unmeasured" }), // unattributed, nothing measured + rec({ executor: "remote", gpu_seconds: 100, cost_usd: 9 }), // GPU row: not part of the CPU rollup + ]; + it("principal and Lab breakdowns each sum to the total", () => { + const t = costTotals(rows); + expect(t.receipts).toBe(5); + for (const by of [t.by_principal, t.by_lab]) { + const vals = Object.values(by); + expect(vals.reduce((a, b) => a + b.receipts, 0)).toBe(t.receipts); + expect(vals.reduce((a, b) => a + b.cpu_seconds, 0)).toBeCloseTo(t.cpu_seconds, 4); + expect(vals.reduce((a, b) => a + b.cost_usd, 0)).toBeCloseTo(t.cost_usd, 4); + expect(vals.reduce((a, b) => a + b.unmeasured, 0)).toBe(t.unmeasured); + } + expect(t.cpu_seconds).toBeCloseTo(18.0, 3); + expect(t.cost_usd).toBeCloseTo(0.80002, 4); + }); + it("rows without a principal or Lab roll up under (unattributed)", () => { + const t = costTotals(rows); + expect(t.by_principal["(unattributed)"]).toMatchObject({ receipts: 2, cpu_seconds: 2, unmeasured: 1 }); + expect(t.by_lab["(unattributed)"]).toMatchObject({ receipts: 2, cpu_seconds: 2, unmeasured: 1 }); + expect(t.by_principal.ana.receipts).toBe(2); + expect(t.by_lab.labA.receipts).toBe(2); + }); + it("rounds to 1e-4 in the rollup only; the records themselves are untouched", () => { + const t = costTotals(rows); + expect(t.by_principal.ana.cost_usd).toBe(0.7500); + expect(t.by_principal.ana.cpu_seconds).toBe(15.0000); + expect((rows[0] as ReceiptRecord).cost_usd).toBe(0.50002); + }); + it("empty input ⇒ zeros and empty breakdowns", () => { + expect(costTotals([])).toEqual({ receipts: 0, cpu_seconds: 0, cost_usd: 0, unmeasured: 0, by_principal: {}, by_lab: {} }); + }); + it("local CPU receipts do not leak into the GPU totals the warrant fold reads", () => { + const f = join(tmp(), "runs.jsonl"); + writeFileSync(f, rows.map((r) => JSON.stringify(r)).join("\n") + "\n"); + const g = gpuTotals(f); + expect(g.receipts).toBe(1); + expect(g.gpu_seconds).toBe(100); + expect(g.cost_usd).toBe(9); + }); +}); + +describe("the engine process clock is never a source of solve CPU", () => { + it("neither local_executor.ts nor cost.ts calls the engine-process CPU clock", () => { + for (const f of ["local_executor.ts", "cost.ts"]) { + const code = readFileSync(join(__dirname, "..", "src", f), "utf8") + .split("\n") + .filter((l) => !/^\s*(\/\/|\*|\/\*)/.test(l)) + .join("\n"); + expect(code, f).not.toMatch(/cpuUsage|resourceUsage|getrusage/); + } + }); + it("the literal `process.cpuUsage` does not appear anywhere in local_executor.ts", () => { + expect(readFileSync(join(__dirname, "..", "src", "local_executor.ts"), "utf8")).not.toContain("process.cpuUsage"); + }); +}); diff --git a/packages/amico-run/test/local_executor_cost.test.ts b/packages/amico-run/test/local_executor_cost.test.ts new file mode 100644 index 000000000..4ebf428db --- /dev/null +++ b/packages/amico-run/test/local_executor_cost.test.ts @@ -0,0 +1,157 @@ +// #2133 — LocalExecutor emits a `local` receipt at settle with CPU measured from the run's cgroup. +// A fixture cgroup dir (a temp dir holding cpu.stat) stands in for the unit; the REAL transient-unit +// proof lives in test/slow/cost_unit.slow.test.ts. The fake julia plays the cgroup: it bumps +// usage_usec by exactly the CPU the "solve" burned, while the engine (this process) stays idle. +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { tmpRoot, fakeJulia } from "./helpers.js"; +import { LocalExecutor } from "../src/local_executor.js"; +import { readRecords, type ReceiptRecord } from "../src/ledger.js"; +import type { RunEvent, SubmitOpts } from "../src/types.js"; + +async function drain(events: AsyncIterable): Promise { + for await (const _ of events); +} +const receipts = (): ReceiptRecord[] => readRecords().filter((r): r is ReceiptRecord => r.type === "receipt"); + +describe("LocalExecutor settle() emits a local CPU receipt", () => { + let dir: string; + let cg: string; + let root: string; + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "amico-cost-le-")); + process.env.AMICO_LEDGER = join(dir, "runs.jsonl"); + cg = join(dir, "cgroup"); + root = tmpRoot(); + writeFileSync(join(dir, "placeholder"), ""); + mkdirSync(cg); + writeFileSync(join(cg, "cpu.stat"), "usage_usec 4000000\nuser_usec 4000000\nsystem_usec 0\n"); + }); + afterEach(() => { + rmSync(dir, { recursive: true, force: true }); + delete process.env.AMICO_LEDGER; + delete process.env.AMICO_PRINCIPAL; + delete process.env.AMICO_LAB; + }); + + // the "solve": accrues `usec` of CPU in the fixture cgroup, then exits with `code` + const solve = (usec: number, code = 0, extra = ""): string => + fakeJulia( + root, + `julia-${usec}-${code}`, + `const fs=require('fs');const p=${JSON.stringify(join(cg, "cpu.stat"))}; + const cur=Number(/usage_usec (\\d+)/.exec(fs.readFileSync(p,'utf8'))[1]); + fs.writeFileSync(p,'usage_usec '+(cur+${usec})+'\\nuser_usec 1\\nsystem_usec 1\\n'); + ${extra} + process.exit(${code});`, + ); + const script = (): string => fakeJulia(root, "s.jl", ""); + + async function submit(julia: string, cost: SubmitOpts["cost"] = {}) { + const h = await new LocalExecutor().submit(script(), { runsRoot: join(root, "runs"), julia: { julia }, cost: { cgroupDir: cg, ...cost } }); + await drain(h.events); + return h.finished; + } + + it("records the CPU the child burned in the cgroup (engine idle) — seconds only when no rate is configured", async () => { + const fin = await submit(solve(2_500_000)); + expect(fin).toEqual({ status: "completed", exitCode: 0 }); + const [r] = receipts(); + expect(receipts()).toHaveLength(1); + expect(r.executor).toBe("local"); + expect(r.cpu_seconds).toBeCloseTo(2.5, 6); + expect(r.cost_usd).toBeUndefined(); + expect(r.cost_basis).toBeUndefined(); + expect(r.status).toBe("completed"); + }); + + it("a rate file ⇒ cost_usd = cpu_seconds × rate, basis list_price_estimate", async () => { + const ratesPath = join(dir, "rates.json"); + writeFileSync(ratesPath, '{"cpu_usd_per_second": 0.0002}'); + await submit(solve(2_500_000), { ratesPath }); + const [r] = receipts(); + expect(r.cpu_seconds).toBeCloseTo(2.5, 6); + expect(r.cost_usd).toBeCloseTo(0.0005, 10); + expect(r.cost_usd).toBe(r.cpu_seconds! * 0.0002); + expect(r.cost_basis).toBe("list_price_estimate"); + }); + + it("a malformed rates.json ⇒ seconds but no dollars, and the run still completes", async () => { + const ratesPath = join(dir, "rates.json"); + writeFileSync(ratesPath, "{ this is not json"); + const fin = await submit(solve(1_000_000), { ratesPath }); + expect(fin.status).toBe("completed"); + const [r] = receipts(); + expect(r.cpu_seconds).toBeCloseTo(1, 6); + expect(r.cost_usd).toBeUndefined(); + }); + + it("a zero-CPU run with a rate omits cost_usd (schema exclusiveMinimum) but still records the receipt", async () => { + const ratesPath = join(dir, "rates.json"); + writeFileSync(ratesPath, '{"cpu_usd_per_second": 0.5}'); + await submit(solve(0), { ratesPath }); + const [r] = receipts(); + expect(r.cpu_seconds).toBe(0); + expect(r.cost_usd).toBeUndefined(); + }); + + it("principal and lab come from the options, else from AMICO_PRINCIPAL / AMICO_LAB, else are absent", async () => { + await submit(solve(100_000), { principal: "ana", lab: "labA" }); + process.env.AMICO_PRINCIPAL = "env-bo"; + process.env.AMICO_LAB = "env-labB"; + await submit(solve(100_000)); + delete process.env.AMICO_PRINCIPAL; + delete process.env.AMICO_LAB; + await submit(solve(100_000)); + const rs = receipts(); + expect(rs.map((r) => [r.principal, r.lab])).toEqual([ + ["ana", "labA"], + ["env-bo", "env-labB"], + [undefined, undefined], + ]); + }); + + it("no unit and no cgroup (macOS / protection none) ⇒ cpu_seconds absent, cost_basis unmeasured — never an engine-clock estimate", async () => { + const h = await new LocalExecutor().submit(script(), { runsRoot: join(root, "runs"), julia: { julia: solve(3_000_000) } }); + await drain(h.events); + expect(await h.finished).toEqual({ status: "completed", exitCode: 0 }); + const [r] = receipts(); + expect(r.cpu_seconds).toBeUndefined(); + expect(r.cost_usd).toBeUndefined(); + expect(r.cost_basis).toBe("unmeasured"); + }); + + it("a failed run keeps its exit code and records the CPU it burned, with status failed", async () => { + const fin = await submit(solve(1_500_000, 7)); + expect(fin).toEqual({ status: "failed", exitCode: 7 }); + const [r] = receipts(); + expect(r.cpu_seconds).toBeCloseTo(1.5, 6); + expect(r.status).toBe("failed"); + }); + + it("accounting failure — the cgroup vanishes mid-run — never fails the run or changes its exit code", async () => { + const gone = (code: number) => solve(1_000_000, code, `fs.rmSync(${JSON.stringify(cg)},{recursive:true,force:true});`); + expect(await submit(gone(0))).toEqual({ status: "completed", exitCode: 0 }); + const [r] = receipts(); + expect(r.cpu_seconds).toBeUndefined(); + expect(r.cost_basis).toBe("unmeasured"); + expect(r.status).toBe("completed"); + }); + + it("accounting failure — cpu.stat becomes unreadable garbage — same: exit code intact, unmeasured", async () => { + const fin = await submit( + fakeJulia(root, "julia-garbage", `require('fs').writeFileSync(${JSON.stringify(join(cg, "cpu.stat"))},'\\u0000not a cpu.stat'); process.exit(3)`), + ); + expect(fin).toEqual({ status: "failed", exitCode: 3 }); + const [r] = receipts(); + expect(r.cpu_seconds).toBeUndefined(); + expect(r.cost_basis).toBe("unmeasured"); + }); + + it("accounting failure — the ledger itself is unwritable — never fails the run", async () => { + process.env.AMICO_LEDGER = join(dir, "placeholder", "not-a-dir", "runs.jsonl"); // parent is a file + expect(await submit(solve(500_000))).toEqual({ status: "completed", exitCode: 0 }); + }); +}); diff --git a/packages/amico-run/test/local_executor_ledger.test.ts b/packages/amico-run/test/local_executor_ledger.test.ts index 77bce12c6..6c0de8440 100644 --- a/packages/amico-run/test/local_executor_ledger.test.ts +++ b/packages/amico-run/test/local_executor_ledger.test.ts @@ -165,7 +165,8 @@ describe("LocalExecutor.settle() emits the solve ledger stanza", () => { await collect(h.events); const fin = await h.finished; expect(fin.status).toBe("completed"); // the run itself is unaffected - expect(readRecords()).toEqual([]); + // #2133: every local run also appends a CPU `receipt`; the `solve` stanza is what is absent here. + expect(readRecords().filter((r) => r.type === "solve")).toEqual([]); }); // REGRESSION (found by adversarial spec review, 2026-07-28): the `max_solves` @@ -226,6 +227,7 @@ describe("LocalExecutor.settle() emits the solve ledger stanza", () => { const h = await new LocalExecutor().submit(script, { runsRoot: join(root, "runs"), julia: { julia } }); await collect(h.events); await h.finished; - expect(readRecords()).toEqual([]); + // #2133: every local run also appends a CPU `receipt`; the `solve` stanza is what is absent here. + expect(readRecords().filter((r) => r.type === "solve")).toEqual([]); }); }); diff --git a/packages/amico-run/test/placement.test.ts b/packages/amico-run/test/placement.test.ts index 7c37bc5bc..7a84f5440 100644 --- a/packages/amico-run/test/placement.test.ts +++ b/packages/amico-run/test/placement.test.ts @@ -123,7 +123,8 @@ describe("placement ladder (#2045 E.1/E.2)", () => { const root = tmpRoot(); const r = await runCli( ["run", fakeJulia(root, "s.jl", ""), "--runs-root", join(root, "runs"), "--julia", fakeJulia(root, "j", "process.exit(0)")], - cliEnv({ AMICO_RUN_PLACEMENT: "none" }), + // every local run now appends a CPU receipt (#2133): give it a writable ledger so "silently" holds + cliEnv({ AMICO_RUN_PLACEMENT: "none", AMICO_LEDGER: join(root, "ledger.jsonl") }), ); expect(r.code).toBe(0); expect(r.stderr).toBe(""); diff --git a/packages/amico-run/test/setup.ts b/packages/amico-run/test/setup.ts index 8c59c1038..40f42f9fd 100644 --- a/packages/amico-run/test/setup.ts +++ b/packages/amico-run/test/setup.ts @@ -48,3 +48,8 @@ if (!process.env.AMICO_RUN_PLACEMENT) process.env.AMICO_RUN_PLACEMENT = "none"; if (!process.env.SOLVE_GUARD_LOCK_DIR) { process.env.SOLVE_GUARD_LOCK_DIR = mkdtempSync(join(tmpdir(), "amico-run-restart-locks-")); } + +// #2133: the CPU rate file (~/.amico/amicode/rates.json) is human-edited; a test must neither read +// the developer's real rates nor depend on them. Default to an absent file (no rate ⇒ no dollars); +// the cost suites pass their own path. +if (!process.env.AMICO_RATES_FILE) process.env.AMICO_RATES_FILE = "/nonexistent/amico-test-guard/rates.json"; diff --git a/packages/amico-run/test/slow/cost_unit.slow.test.ts b/packages/amico-run/test/slow/cost_unit.slow.test.ts new file mode 100644 index 000000000..9f1ccb025 --- /dev/null +++ b/packages/amico-run/test/slow/cost_unit.slow.test.ts @@ -0,0 +1,101 @@ +// #2133 — REAL CPU attribution from a real transient unit's cgroup. +// +// Needs a working user service manager AND lingering; otherwise the whole file is skipped LOUDLY +// (console.warn below + the suite name says why). The fixture-cgroup logic is covered in the fast +// tier (test/cost.test.ts, test/local_executor_cost.test.ts) either way. +// +// SAFETY: only transient units named `amico-run-test-*` are created/stopped here +// (AMICO_RUN_UNIT_PREFIX=amico-run-test-); every unit this file creates is stopped in afterAll. +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { spawnSync } from "node:child_process"; +import { existsSync, mkdtempSync, readdirSync, readFileSync, writeFileSync } from "node:fs"; +import { randomBytes } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fakeJulia, tmpRoot } from "../helpers.js"; +import { probeUserUnits } from "../../src/placement.js"; +import { resolveUnitCgroupDir, startCpuMeter } from "../../src/cost.js"; +import { costTotals, type ReceiptRecord } from "../../src/ledger.js"; +import { buildBundle, cliEnv, runCli, waitUntil } from "../launch_harness.js"; + +const PREFIX = "amico-run-test-"; +const env0 = { ...process.env, AMICO_RUN_UNIT_PREFIX: PREFIX }; +const probe = probeUserUnits(env0); +if (!probe.ok) console.warn(`SKIPPED cost_unit.slow tests — real systemd placement unavailable: ${probe.reason}`); + +const created = new Set(); +const systemctl = (...args: string[]) => spawnSync("systemctl", ["--user", ...args], { encoding: "utf8", env: env0 }); + +// burns >= 1.5 s of CPU in THIS child (it measures itself only to know when to stop) +const BURN = ` +const t = () => { const u = process.cpuUsage(); return (u.user + u.system) / 1e6; }; +let x = 0; while (t() < 1.5) { for (let i = 0; i < 1e6; i++) x += Math.sqrt(i); } +`; + +beforeAll(buildBundle); +afterAll(() => { + for (const u of created) systemctl("stop", u); + const active = [...created].filter((u) => systemctl("is-active", u).stdout.trim() === "active"); + expect(active).toEqual([]); +}); + +describe.skipIf(!probe.ok)("real transient-unit CPU attribution", () => { + it("meters the CPU a child burns inside a unit's cgroup (read from the unit's own cpu.stat)", async () => { + const root = tmpRoot(); + const marker = join(root, "burned"); + const burner = join(root, "burner.js"); + writeFileSync(burner, `${BURN}require('fs').writeFileSync(${JSON.stringify(marker)},'x');setTimeout(()=>{},60000);`); + const unit = `${PREFIX}meter-${randomBytes(3).toString("hex")}.service`; + created.add(unit); + const started = spawnSync("systemd-run", ["--user", "--quiet", "--collect", `--unit=${unit}`, process.execPath, burner], { encoding: "utf8", env: env0 }); + expect(started.status, started.stderr).toBe(0); + + expect(resolveUnitCgroupDir(unit, env0)).toMatch(new RegExp(`/${unit}$`)); + const meter = startCpuMeter({ unit, env: env0 }); + await waitUntil("burned", () => existsSync(marker), 30000); + const secs = meter.stop(); + // the meter starts a few ms into the burn at worst: >= 1.0 s of the 1.5 s the child burned + expect(secs).toBeGreaterThanOrEqual(1.0); + expect(secs).toBeLessThan(30); + systemctl("stop", unit); + }, 60000); + + it("an `amico run` in a real unit whose idle engine only waits: the receipt carries the CHILD's >= 1 s, priced by the rate file, attributed to principal and Lab", async () => { + const root = tmpRoot(); + const ledger = join(root, "runs.jsonl"); + const rates = join(root, "rates.json"); + writeFileSync(rates, '{"cpu_usd_per_second": 0.001}'); + const julia = fakeJulia(root, "j", BURN); + const env = cliEnv({ + AMICO_RUN_PLACEMENT: "auto", + AMICO_RUN_UNIT_PREFIX: PREFIX, + AMICO_LEDGER: ledger, + AMICO_RATES_FILE: rates, + AMICO_PRINCIPAL: "ana", + AMICO_LAB: "labA", + }); + const r = await runCli(["run", fakeJulia(root, "s.jl", ""), "--runs-root", join(root, "runs"), "--julia", julia], env); + expect(r.code, r.stderr).toBe(0); + for (const d of readdirSync(join(root, "runs"))) { + try { + const lock = JSON.parse(readFileSync(join(root, "runs", d, "run.lock"), "utf8")); + if (lock.unit) { + created.add(lock.unit); + expect(lock.protection).toBe("unit"); + } + } catch { + /* not a run dir */ + } + } + const recs = readFileSync(ledger, "utf8").trim().split("\n").map((l) => JSON.parse(l) as ReceiptRecord); + const rec = recs.filter((x) => x.type === "receipt"); + expect(rec).toHaveLength(1); + expect(rec[0].executor).toBe("local"); + expect(rec[0].cpu_seconds).toBeGreaterThanOrEqual(1.0); + expect(rec[0].cpu_seconds).toBeLessThan(30); + expect(rec[0].cost_basis).toBe("list_price_estimate"); + expect(rec[0].cost_usd).toBeCloseTo(rec[0].cpu_seconds! * 0.001, 12); + expect([rec[0].principal, rec[0].lab]).toEqual(["ana", "labA"]); + expect(costTotals(rec).by_principal.ana.cpu_seconds).toBeGreaterThanOrEqual(1.0); + }, 90000); +}); diff --git a/packages/schema/schemas/ledger-record.schema.json b/packages/schema/schemas/ledger-record.schema.json index afb3b0bfe..a19682cab 100644 --- a/packages/schema/schemas/ledger-record.schema.json +++ b/packages/schema/schemas/ledger-record.schema.json @@ -861,7 +861,7 @@ }, { "title": "receipt", - "description": "GPU/compute accounting for a remote run (#425, GPU-plane spec): pure spend record \u2014 no fidelity, never feeds priors. Emitted by the remote executor at settle from the runner's finished payload; the campaign warrant fold sums these (unit-keyed bounds, telaio #5). A receipt names WHAT consumed (task_id) and WHERE (executor: remote) \u2014 local runs burn nothing billable and produce none.", + "description": "GPU/compute accounting for a remote run (#425, GPU-plane spec): pure spend record \u2014 no fidelity, never feeds priors. Emitted by the remote executor at settle from the runner's finished payload; the campaign warrant fold sums these (unit-keyed bounds, telaio #5). A receipt names WHAT consumed (task_id) and WHERE (executor: remote = cloud GPU; local = CPU of a solve on this host, #2133: cpu_seconds measured from the solve unit's cgroup, never process.cpuUsage(); cost_usd = cpu_seconds x the human-edited list rate, cost_basis names its provenance; principal and lab attribute the spend).", "type": "object", "additionalProperties": false, "required": [ @@ -882,7 +882,10 @@ "minLength": 1 }, "executor": { - "const": "remote" + "enum": [ + "local", + "remote" + ] }, "gpu_sku": { "type": "string", @@ -892,10 +895,28 @@ "type": "number", "exclusiveMinimum": 0 }, + "cpu_seconds": { + "type": "number", + "minimum": 0 + }, "cost_usd": { "type": "number", "exclusiveMinimum": 0 }, + "cost_basis": { + "enum": [ + "list_price_estimate", + "unmeasured" + ] + }, + "principal": { + "type": "string", + "minLength": 1 + }, + "lab": { + "type": "string", + "minLength": 1 + }, "problem": { "type": "string", "minLength": 1 diff --git a/packages/schema/test/ledger-record.test.ts b/packages/schema/test/ledger-record.test.ts index eee1a05e1..94fbde888 100644 --- a/packages/schema/test/ledger-record.test.ts +++ b/packages/schema/test/ledger-record.test.ts @@ -266,10 +266,49 @@ describe("the receipt record (GPU accounting, #425)", () => { expect(validate(receipt({ gpu_seconds: "1800" }), "ledger-record").ok).toBe(false); expect(validate(receipt({ cost_usd: -0.01 }), "ledger-record").ok).toBe(false); }); - it("executor is an enum — local receipts make no sense (nothing burns)", () => { - expect(validate(receipt({ executor: "local" }), "ledger-record").ok).toBe(false); + it("executor is an enum — local (CPU of a solve on this host, #2133) and remote; nothing else", () => { + expect(validate(receipt({ executor: "local" }), "ledger-record").ok).toBe(true); + expect(validate(receipt({ executor: "elsewhere" }), "ledger-record").ok).toBe(false); }); it("optional problem/session stamps for attribution", () => { expect(validate(receipt({ problem: "cx-cat-cat-cavity", session: "s1" }), "ledger-record")).toMatchObject({ ok: true }); }); }); + +// ── the local CPU receipt (#2133): cost attribution from the solve unit's cgroup ──── +describe("the local CPU receipt (cost attribution, #2133)", () => { + const local = (over: Record = {}) => ({ + type: "receipt", + ts: "2026-11-04T12:00:00.000Z", + task_id: "r20261104-120000Z-abc123", + executor: "local", + cpu_seconds: 1.5, + cost_usd: 0.000075, + cost_basis: "list_price_estimate", + principal: "aaron", + lab: "harmoniqs", + status: "completed", + ...over, + }); + it("accepts cpu_seconds, cost_usd, cost_basis, principal and lab", () => { + expect(validate(local(), "ledger-record")).toMatchObject({ ok: true }); + }); + it("accepts an unmeasured receipt: no cpu_seconds, no cost_usd, cost_basis unmeasured", () => { + expect( + validate(local({ cpu_seconds: undefined, cost_usd: undefined, cost_basis: "unmeasured", principal: undefined, lab: undefined }), "ledger-record"), + ).toMatchObject({ ok: true }); + }); + it("rejects bad types: string cpu_seconds, negative cpu_seconds, zero cost_usd, unknown cost_basis, empty/non-string principal or lab", () => { + for (const over of [ + { cpu_seconds: "1.5" }, + { cpu_seconds: -1 }, + { cost_usd: 0 }, + { cost_basis: "invoice" }, + { principal: "" }, + { principal: 7 }, + { lab: "" }, + { lab: ["a"] }, + ]) + expect(validate(local(over), "ledger-record").ok, JSON.stringify(over)).toBe(false); + }); +});