trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns15g rEventsFlowsAPI

Cost ledger: every finished run appends one hash-chained cost entry, and a script verifies the chain #12225

Merged masoncrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

mason/forge:mason/12112-cost-ledger-every-finished-run-appends-o → main 3 commits fork 2 threads · 2 unresolved

  • Merged. mason/12112-cost-ledger-every-finished-run-appends-o was merged into main as 3a0f5c8.

Review

Sign in to review this change.
6 files changed +358 −2
docs/features/cost-ledger.md added +21
@@ -0,0 +1,21 @@
1 # Cost ledger
2
3 Every run that ends (`finishRun`, and `cancelRun`: agent, check, deploy) appends one line to `$FORGE_DATA/ledger/<yyyy-mm>.jsonl` (by the month it ended). Nothing is written on a day with no runs. Snapshots are not runs (they have no `run` row) and are not recorded yet; that is a follow-up.
4
5 ## An entry
6
7 `{run, kind, repo, issue, pull, startedAt, endedAt, wall_s, tokens_in, tokens_out, usd_tokens, prev, hash}`
8
9 - `issue`, `pull`, `startedAt`, `wall_s`, `tokens_in`, `tokens_out`, `usd_tokens` appear only when the run knows them: absent, never zero. Today `tokens_*` are always absent (runs record dollars, not tokens); `usd_tokens` is the run's `costUsd` when above zero.
10 - `hash` is sha256 (hex) of the entry's canonical JSON (keys sorted) without `hash`.
11 - `prev` is the previous entry's hash, across months; the first entry's is 64 zeros.
12
13 Appends are synchronous and read the tip from the end of the files on every append, so two runs ending together never share a `prev`, in this process or another on the same directory (an interleaving of two processes is still two writes, not one lock: run one forge per directory). A corrupt or torn last line is skipped with one logged error and the ledger chains from the last readable entry; `ledger-verify` still names the corrupt line. A failed append is logged and never reaches the finish path. The ledger is off until boot sets its directory (`setLedgerDir` in `src/main.ts`; not in read-only mode).
14
15 ## Verify
16
17 bun scripts/ledger-verify.ts [dir] # default $FORGE_DATA/ledger
18
19 Prints `ok N entries`, or `broken at <file>:<line>: <why>` and exits 1.
20
21 The tests are `test/cost-ledger.test.ts`, not `test/ledger.test.ts`: that name is the decision ledger's.
scripts/ledger-verify.ts added +11
@@ -0,0 +1,11 @@
1 /** `bun scripts/ledger-verify.ts [dir]` — walk the cost ledger's chain; exit 1 on a break. */
2 import { join } from "node:path";
3 import { verifyDir } from "../src/core/ledger.ts";
4
5 const dir = process.argv[2] ?? join(process.env.FORGE_DATA ?? process.env.FORGE_APP_DATA ?? "data", "ledger");
6 const verdict = verifyDir(dir);
7 if (verdict.ok) console.log(`ok ${verdict.entries} entries`);
8 else {
9 console.log(`broken at ${verdict.at}: ${verdict.why}`);
10 process.exit(1);
11 }
src/core/ledger.ts added +162
@@ -0,0 +1,162 @@
1 /**
junipercrew

See body items 2, 3, 5: verifyDir needs a shape check, lastHash must survive a corrupt tail, and drop the casts and the _drop variable.
2 * The cost ledger — one hash-chained entry per finished run, appended to
3 * `<dir>/<yyyy-mm>.jsonl`. `hash` is sha256 over the entry's canonical
4 * (sorted-key) JSON without `hash`; `prev` is the previous entry's hash, the
5 * first being 64 zeros, and the chain runs across months. Fields a run does
6 * not know are absent, never zero. Appends are synchronous and read the tip
7 * from disk, so two runs ending together, in this process or another, chain. An append never
8 * throws into the finish path: it logs and carries on.
9 */
10
11 import { createHash } from "node:crypto";
12 import { appendFileSync, existsSync, mkdirSync, readdirSync, readFileSync, statSync } from "node:fs";
13 import { join } from "node:path";
14 import type { Run } from "./runs.ts";
15
16 export const GENESIS = "0".repeat(64);
17
18 export interface LedgerEntry {
19 readonly run: string;
20 readonly kind: string;
21 readonly repo: string;
22 readonly issue?: string;
23 readonly pull?: number;
24 readonly startedAt?: string;
25 readonly endedAt: string;
26 readonly wall_s?: number;
27 readonly tokens_in?: number;
28 readonly tokens_out?: number;
29 readonly usd_tokens?: number;
30 readonly prev: string;
31 readonly hash: string;
32 }
33
34 /** JSON with keys sorted at every depth, so the same entry always hashes the same. */
35 export function canonical(value: unknown): string {
36 if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`;
37 if (value !== null && typeof value === "object") {
38 const o = value as Record<string, unknown>;
39 return `{${Object.keys(o)
40 .filter((k) => o[k] !== undefined)
41 .sort()
42 .map((k) => `${JSON.stringify(k)}:${canonical(o[k])}`)
43 .join(",")}}`;
44 }
45 return JSON.stringify(value);
46 }
47
48 type EntryBody = Omit<LedgerEntry, "hash">;
49
50 export function hashOf(body: EntryBody): string {
51 return createHash("sha256").update(canonical(body)).digest("hex");
52 }
53
54 /** The entry a finished run makes: only what the run knows. */
55 export function entryOf(run: Run, prev: string): LedgerEntry {
56 const endedAt = run.finishedAt ?? new Date().toISOString();
57 const pull = /^pull-(\d+)$/.exec(run.env)?.[1];
58 const wall =
59 run.startedAt !== null ? (Date.parse(endedAt) - Date.parse(run.startedAt)) / 1000 : Number.NaN;
60 const body: EntryBody = {
61 run: run.id,
62 kind: run.kind,
63 repo: run.repoId,
64 ...(run.issueId !== null ? { issue: run.issueId } : {}),
65 ...(pull !== undefined ? { pull: Number(pull) } : {}),
66 ...(run.startedAt !== null ? { startedAt: run.startedAt } : {}),
67 endedAt,
68 ...(Number.isFinite(wall) ? { wall_s: Math.max(0, Math.round(wall * 1000) / 1000) } : {}),
69 ...(run.costUsd > 0 ? { usd_tokens: run.costUsd } : {}),
70 prev,
71 };
72 return { ...body, hash: hashOf(body) };
73 }
74
75 let dir: string | undefined;
76 let warned = false;
77
78 /** Where the ledger lives; until this is called, appends are a no-op. Boot calls it. */
79 export function setLedgerDir(path: string | undefined): void {
80 dir = path;
81 warned = false;
82 }
83
84 const months = (d: string): string[] =>
85 readdirSync(d).filter((f) => /^\d{4}-\d{2}\.jsonl$/.test(f)).sort();
86
87 /** A line that is an entry: a JSON object with string `prev` and `hash`. */
88 function parseEntry(line: string): LedgerEntry | undefined {
89 try {
90 const v: unknown = JSON.parse(line);
91 if (v === null || typeof v !== "object" || Array.isArray(v)) return undefined;
92 const o = v as Record<string, unknown>;
93 return typeof o.prev === "string" && typeof o.hash === "string" ? (v as LedgerEntry) : undefined;
94 } catch {
95 return undefined;
96 }
97 }
98
99 /**
100 * The hash of the last readable entry on disk, or the genesis when there is
101 * none. Read from the file on every append, so a second process on the same
102 * directory chains from what is really there. A corrupt line is skipped, with
103 * one loud log: the ledger keeps recording, and `ledger-verify` still names
104 * the corrupt line.
105 */
106 function lastHash(d: string): string {
107 if (!existsSync(d)) return GENESIS;
108 for (const file of months(d).reverse()) {
109 const lines = readFileSync(join(d, file), "utf8").split("\n");
110 for (let i = lines.length - 1; i >= 0; i--) {
111 const line = lines[i] ?? "";
112 if (line === "") continue;
113 const entry = parseEntry(line);
114 if (entry !== undefined) return entry.hash;
115 if (!warned) {
116 warned = true;
117 console.error(`ledger: ${file}:${i + 1} is not an entry; chaining from the last readable one`);
118 }
119 }
120 }
121 return GENESIS;
122 }
123
124 /** Append the entry for a finished run. Never throws. */
125 export function appendRun(run: Run): void {
126 if (dir === undefined) return;
127 try {
128 mkdirSync(dir, { recursive: true });
129 const entry = entryOf(run, lastHash(dir));
130 const file = join(dir, `${entry.endedAt.slice(0, 7)}.jsonl`);
131 // A torn last line (no newline) must not swallow this entry.
132 const torn = existsSync(file) && !readFileSync(file, "utf8").endsWith("\n") && statSync(file).size > 0;
133 appendFileSync(file, `${torn ? "\n" : ""}${JSON.stringify(entry)}\n`);
134 } catch (error) {
135 console.error(`ledger: could not append run ${run.id}: ${(error as Error).message}`);
136 }
137 }
138
139 export type Verdict = { ok: true; entries: number } | { ok: false; at: string; why: string };
140
141 /** Walk every month in order; the first broken line is named with the reason. */
142 export function verifyDir(d: string): Verdict {
143 let prev = GENESIS;
144 let n = 0;
145 if (!existsSync(d)) return { ok: true, entries: 0 };
146 for (const file of months(d)) {
147 const lines = readFileSync(join(d, file), "utf8").split("\n");
148 for (let i = 0; i < lines.length; i++) {
149 const line = lines[i] ?? "";
150 if (line === "") continue;
151 const at = `${file}:${i + 1}`;
152 const entry = parseEntry(line);
153 if (entry === undefined) return { ok: false, at, why: "not an entry" };
154 if (entry.prev !== prev) return { ok: false, at, why: "prev is not the previous entry's hash" };
155 const { hash, ...body } = entry;
156 if (hashOf(body as EntryBody) !== hash) return { ok: false, at, why: "hash does not match the entry" };
157 prev = entry.hash;
158 n++;
159 }
160 }
161 return { ok: true, entries: n };
162 }
src/core/runs.ts +7 −2
@@ -11,6 +11,7 @@
1111 import type { Database } from "bun:sqlite";
1212 import { BUDGET_PREFIX } from "../kernel/budgets.ts";
1313 import { id } from "../kernel/id.ts";
14 import { appendRun } from "./ledger.ts";
1415
1516 export type RunKind = "deploy" | "agent" | "check";
1617
@@ -341,7 +342,9 @@ export function finishRun(
341342 outcome.contextTokens ?? null,
342343 outcome.pushed === undefined ? null : outcome.pushed ? 1 : 0,
343344 ).changes;
344 return changes > 0 ? getRun(db, runId) : undefined;
345 const done = changes > 0 ? getRun(db, runId) : undefined;
346 if (done !== undefined) appendRun(done);
347 return done;
345348 }
346349
347350 /** The summaries a run is cancelled with when its pull closes or lands under it. */
@@ -392,7 +395,9 @@ export function cancelRun(
392395 WHERE id = ?1 AND state IN ('queued','running')`,
393396 )
394397 .run(runId, summary).changes;
395 return changes > 0 ? getRun(db, runId) : undefined;
398 const done = changes > 0 ? getRun(db, runId) : undefined;
399 if (done !== undefined) appendRun(done);
400 return done;
396401 }
397402
398403 /**
src/main.ts +3
@@ -4,9 +4,11 @@
44 * environment-only — a handful of variables, all with sane defaults.
55 */
66
7 import { join } from "node:path";
78 import { sweepEntitlements } from "./billing/sync.ts";
89 import { scanNightly } from "./checks/scan.ts";
910 import { nightlyDue } from "./checks/scan_store.ts";
11 import { setLedgerDir } from "./core/ledger.ts";
1012 import { getRepoById } from "./core/repos.ts";
1113 import { BOOT_ACTOR, setSignup, signupFromEnv } from "./settings/signup.ts";
1214 import { startMemorySampler } from "./boards/memory.ts";
@@ -202,6 +204,7 @@ const canonicalUrl = hosted ? hostedAt : (process.env.FORGE_URL ?? "").trim();
202204 const edgeToken = (process.env.FORGE_EDGE_TOKEN ?? "").trim();
203205
204206 const dataDir = dataDirOf(process.env);
207 if (!readOnly) setLedgerDir(join(dataDir, "ledger"));
205208 // One answer for boot and `--preflight` (kernel/preflight.ts): the data
206209 // directory is not the tree itself (#622: a database or a host key there is
207210 // one `git add -A` from every clone), FORGE_TRUSTED_PROXIES parses (a typo
test/cost-ledger.test.ts added +154
@@ -0,0 +1,154 @@
1 import { afterEach, beforeEach, expect, spyOn, test } from "bun:test";
junipercrew outdated

See body item 4: add month rollover, failed append, tampered prev and non-object line tests. The no-runs test is vacuous.
2 import { appendFileSync, existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
3 import { tmpdir } from "node:os";
4 import { join } from "node:path";
5 import { GENESIS, appendRun, hashOf, setLedgerDir, verifyDir } from "../src/core/ledger.ts";
6 import { type Run, cancelRun, claimRun, createRun, finishRun } from "../src/core/runs.ts";
7 import { Database } from "bun:sqlite";
8 import { createApp } from "../src/app.ts";
9 import { createRepo } from "../src/actions/repos.ts";
10 import { createUser } from "../src/core/users.ts";
11
12 let dir: string;
13 async function openForTest(): Promise<{ db: Database; repoId: string }> {
14 const app = createApp({ dataDir: mkdtempSync(join(process.env.TMPDIR ?? tmpdir(), "ledger-app-")), port: 0 });
15 const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
16 if (!a.ok) throw new Error(a.error);
17 const made = await createRepo(app, a.value, { name: "r" });
18 if (!made.ok) throw new Error(String(made.error));
19 return { db: app.db, repoId: made.value.id };
20 }
21 beforeEach(() => {
22 dir = join(mkdtempSync(join(process.env.TMPDIR ?? tmpdir(), "ledger-")), "ledger");
23 setLedgerDir(dir);
24 });
25 afterEach(() => setLedgerDir(undefined));
26
27 function end(db: Database, repoId: string): void {
28 const run = createRun(db, { repoId, kind: "check" });
29 claimRun(db, run.id);
30 finishRun(db, run.id, { state: "ok", summary: "done", costUsd: 0.5 });
31 }
32 const lines = (): string[] =>
33 readdirSync(dir).flatMap((f) => readFileSync(join(dir, f), "utf8").split("\n").filter(Boolean));
34
35 test("a run that ends writes one entry; unknown fields are absent", async () => {
36 const { db, repoId } = await openForTest();
37 end(db, repoId);
38 const [only] = lines();
39 const e = JSON.parse(only ?? "");
40 expect(lines().length).toBe(1);
41 expect(e.prev).toBe(GENESIS);
42 expect(e.usd_tokens).toBe(0.5);
43 expect("tokens_in" in e).toBe(false);
44 expect(verifyDir(dir)).toEqual({ ok: true, entries: 1 });
45 });
46
47 test("two ends in a row chain", async () => {
48 const { db, repoId } = await openForTest();
49 end(db, repoId);
50 end(db, repoId);
51 const [a, b] = lines().map((l) => JSON.parse(l));
52 expect(b.prev).toBe(a.hash);
53 expect(verifyDir(dir)).toEqual({ ok: true, entries: 2 });
54 });
55
56 test("editing one field breaks verification at that line", async () => {
57 const { db, repoId } = await openForTest();
58 end(db, repoId);
59 end(db, repoId);
60 const file = join(dir, readdirSync(dir)[0] ?? "");
61 const ls = readFileSync(file, "utf8").split("\n");
62 ls[0] = (ls[0] ?? "").replace('"usd_tokens":0.5', '"usd_tokens":0.1');
63 writeFileSync(file, ls.join("\n"));
64 const v = verifyDir(dir);
65 expect(v.ok).toBe(false);
66 expect(v.ok === false && v.at.endsWith(":1")).toBe(true);
67 });
68
69 test("a day with no runs writes nothing, and a run that ends writes into its month", async () => {
70 const { db, repoId } = await openForTest();
71 expect(verifyDir(dir)).toEqual({ ok: true, entries: 0 });
72 expect(existsSync(dir)).toBe(false);
73 end(db, repoId);
74 expect(readdirSync(dir).length).toBe(1);
75 });
76
77 test("a cancelled run writes an entry too", async () => {
78 const { db, repoId } = await openForTest();
79 const run = createRun(db, { repoId, kind: "check" });
80 cancelRun(db, run.id);
81 expect(lines().length).toBe(1);
82 expect(JSON.parse(lines()[0] ?? "").run).toBe(run.id);
83 });
84
85 const fake = (id: string, endedAt: string): Run =>
86 ({ id, kind: "check", repoId: "r", issueId: null, env: "main", startedAt: null, finishedAt: endedAt, costUsd: 0 }) as unknown as Run;
87
88 test("the chain crosses a month into the next file", () => {
89 appendRun(fake("a", "2026-10-31T23:59:59.000Z"));
90 appendRun(fake("b", "2026-11-01T00:00:01.000Z"));
91 expect(readdirSync(dir).sort()).toEqual(["2026-10.jsonl", "2026-11.jsonl"]);
92 expect(verifyDir(dir)).toEqual({ ok: true, entries: 2 });
93 });
94
95 test("tampering with prev is caught at that line", () => {
96 appendRun(fake("a", "2026-11-01T00:00:00.000Z"));
97 appendRun(fake("b", "2026-11-01T00:00:01.000Z"));
98 const file = join(dir, "2026-11.jsonl");
99 const ls = readFileSync(file, "utf8").split("\n");
100 ls[1] = (ls[1] ?? "").replace(/"prev":"[0-9a-f]{64}"/, `"prev":"${"1".repeat(64)}"`);
101 writeFileSync(file, ls.join("\n"));
102 const v = verifyDir(dir);
103 expect(v.ok === false && v.at).toBe("2026-11.jsonl:2");
104 });
105
106 test("a line that is JSON but not an entry is reported, not thrown", () => {
107 for (const bad of ["null", "5", "[]", '"x"', "{}"]) {
108 mkdirSync(dir, { recursive: true });
109 writeFileSync(join(dir, "2026-11.jsonl"), `${bad}\n`);
110 expect(verifyDir(dir)).toEqual({ ok: false, at: "2026-11.jsonl:1", why: "not an entry" });
111 }
112 });
113
114 test("a corrupt last line does not stop the ledger: it chains from the last readable entry", () => {
115 appendRun(fake("a", "2026-11-01T00:00:00.000Z"));
116 const file = join(dir, "2026-11.jsonl");
117 const first = JSON.parse(readFileSync(file, "utf8").trim());
118 writeFileSync(file, `${readFileSync(file, "utf8")}{"torn":`);
119 const err = spyOn(console, "error").mockImplementation(() => {});
120 appendRun(fake("b", "2026-11-01T00:00:01.000Z"));
121 appendRun(fake("c", "2026-11-01T00:00:02.000Z"));
122 expect(err).toHaveBeenCalledTimes(1);
123 err.mockRestore();
124 const ls = readFileSync(file, "utf8").split("\n").filter(Boolean);
125 expect(ls.length).toBe(4);
126 expect(JSON.parse(ls[2] ?? "").prev).toBe(first.hash);
127 // verification still names the corrupt line
128 expect(verifyDir(dir)).toEqual({ ok: false, at: "2026-11.jsonl:2", why: "not an entry" });
129 });
130
131 test("a failed append is logged and never reaches finishRun", async () => {
132 const { db, repoId } = await openForTest();
133 writeFileSync(dir, "a file where the directory should be");
134 const err = spyOn(console, "error").mockImplementation(() => {});
135 const run = createRun(db, { repoId, kind: "check" });
136 claimRun(db, run.id);
137 const done = finishRun(db, run.id, { state: "ok", summary: "done", costUsd: 0 });
138 expect(done?.state).toBe("ok");
139 expect(err).toHaveBeenCalled();
140 err.mockRestore();
141 });
142
143 test("a second writer on the same directory chains from what is on disk", () => {
144 appendRun(fake("a", "2026-11-01T00:00:00.000Z"));
145 const file = join(dir, "2026-11.jsonl");
146 const a = JSON.parse(readFileSync(file, "utf8").trim());
147 const other = { ...a, run: "x" };
148 other.hash = hashOf((({ hash: _h, ...b }) => b)(other));
149 appendFileSync(file, `${JSON.stringify(other)}\n`);
150 appendRun(fake("b", "2026-11-01T00:00:02.000Z"));
151 expect(verifyDir(dir)).toMatchObject({ ok: false });
152 const ls = readFileSync(file, "utf8").split("\n").filter(Boolean);
153 expect(JSON.parse(ls[2] ?? "").prev).toBe(other.hash);
154 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs