Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 114 additions & 0 deletions packages/amico-run/src/cost.ts
Original file line number Diff line number Diff line change
@@ -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-<id>.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 <unit>`.
//
// 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<ReceiptRecord, "cpu_seconds" | "cost_usd" | "cost_basis"> {
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 };
}
56 changes: 54 additions & 2 deletions packages/amico-run/src/ledger.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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;
Expand All @@ -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<string, CostBucket>;
by_lab: Record<string, CostBucket>;
}

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
Expand Down
39 changes: 38 additions & 1 deletion packages/amico-run/src/local_executor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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<SubmitOpts["cost"]>,
): 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
Expand Down Expand Up @@ -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<RunEvent>();
const logStream = createWriteStream(join(runDir, "run.log"), { flags: "a" });
let resolveFinished!: (f: Finished) => void;
Expand All @@ -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.
//
Expand Down Expand Up @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion packages/amico-run/src/placement.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
6 changes: 6 additions & 0 deletions packages/amico-run/src/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
}

Expand Down
Loading
Loading