1. 1//! FinOps (docs/design/15-reliability.md): persisted cost ledger + pre-dispatch
2. 2//! budget admission. Every settled dispatch appends an estimated-USD row
3. 3//! keyed by durable attribution ids; caps are checked BEFORE each paid
4. 4//! call. Denial is a question (budget Ask), not a crash; unattended
5. 5//! surfaces auto-deny via the normal approver path.
6. 6
7. 7use std::collections::BTreeMap;
8. 8use std::io::{BufRead, BufReader, Write};
9. 9use std::path::PathBuf;
10. 10use std::sync::atomic::{AtomicBool, Ordering};
11. 11use std::sync::{Arc, Mutex};
12. 12
13. 13use serde::{Deserialize, Serialize};
14. 14use vak_agent::{SpendCheck, SpendGate};
15. 15use vak_llm::Usage;
16. 16
17. 17/// Above this size, `append` compacts the ledger before writing (see
18. 18/// [`FinOpsLedger::compact_if_large`]) instead of letting it grow forever —
19. 19/// every `authorize()` used to re-read the whole file from the start of
20. 20/// time on every paid dispatch, so an unbounded file meant unbounded
21. 21/// per-dispatch latency as well as unbounded disk use.
22. 22const COST_LOG_COMPACT_THRESHOLD_BYTES: u64 = 5 * 1024 * 1024;
23. 23/// How much history compaction keeps: comfortably more than the 14-day
24. 24/// admin trend chart and the `total_usd_since` callers in this codebase
25. 25/// use (7/30-day rollups), so compaction never changes a real answer.
26. 26const COST_LOG_RETENTION_DAYS: i64 = 90;
27. 27/// Same idea for the budget-alert log, which is read in full by
28. 28/// `last_alert`/`recent_budget_alerts` and only ever needs recent history.
29. 29const ALERTS_COMPACT_THRESHOLD_BYTES: u64 = 1024 * 1024;
30. 30const ALERTS_RETENTION_ROWS: usize = 2000;
31. 31
32. 32#[derive(Debug, Clone, Serialize, Deserialize)]
33. 33pub struct CostRow {
34. 34 pub ts: chrono::DateTime<chrono::Utc>,
35. 35 pub model: String,
36. 36 /// Serving provider of the frozen-ladder leg (Phase R per-provider
37. 37 /// FinOps rollups). Empty on legacy rows.
38. 38 #[serde(default)]
39. 39 pub provider: String,
40. 40 pub input_tokens: u64,
41. 41 pub output_tokens: u64,
42. 42 #[serde(default, skip_serializing_if = "Option::is_none")]
43. 43 pub cache_read_input_tokens: Option<u64>,
44. 44 /// Estimated USD. `None` when the model is unpriced — absent is
45. 45 /// UNKNOWN, never zero.
46. 46 #[serde(default, skip_serializing_if = "Option::is_none")]
47. 47 pub usd: Option<f64>,
48. 48 /// Always "estimated" today; reserved for providers that return real
49. 49 /// settlement data.
50. 50 pub source: String,
51. 51 pub session_id: String,
52. 52}
53. 53
54. 54#[derive(Debug, Clone, Serialize, Deserialize)]
55. 55pub struct ActivityRow {
56. 56 pub ts: chrono::DateTime<chrono::Utc>,
57. 57 pub kind: String,
58. 58 pub name: String,
59. 59 pub success: bool,
60. 60 #[serde(default, skip_serializing_if = "Option::is_none")]
61. 61 pub duration_ms: Option<u64>,
62. 62 #[serde(default, skip_serializing_if = "Option::is_none")]
63. 63 pub session_id: Option<String>,
64. 64 #[serde(default, skip_serializing_if = "Option::is_none")]
65. 65 pub plugin: Option<String>,
66. 66}
67. 67
68. 68pub struct ActivityLedger {
69. 69 path: PathBuf,
70. 70}
71. 71
72. 72impl ActivityLedger {
73. 73 pub fn new(sessions_home: &std::path::Path) -> Self {
74. 74 Self {
75. 75 path: sessions_home.join("activity-log.jsonl"),
76. 76 }
77. 77 }
78. 78
79. 79 pub fn append(&self, row: &ActivityRow) -> std::io::Result<()> {
80. 80 if let Some(parent) = self.path.parent() {
81. 81 std::fs::create_dir_all(parent)?;
82. 82 }
83. 83 let line = serde_json::to_string(row)
84. 84 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
85. 85 let mut file = std::fs::OpenOptions::new()
86. 86 .create(true)
87. 87 .append(true)
88. 88 .open(&self.path)?;
89. 89 writeln!(file, "{line}")
90. 90 }
91. 91
92. 92 pub fn all_rows(&self) -> Vec<ActivityRow> {
93. 93 let Ok(file) = std::fs::File::open(&self.path) else {
94. 94 return Vec::new();
95. 95 };
96. 96 BufReader::new(file)
97. 97 .lines()
98. 98 .map_while(Result::ok)
99. 99 .filter_map(|line| serde_json::from_str(&line).ok())
100. 100 .collect()
101. 101 }
102. 102}
103. 103
104. 104pub struct FinOpsLedger {
105. 105 path: PathBuf,
106. 106}
107. 107
108. 108impl FinOpsLedger {
109. 109 pub fn new(sessions_home: &std::path::Path) -> Self {
110. 110 FinOpsLedger {
111. 111 path: sessions_home.join("cost-log.jsonl"),
112. 112 }
113. 113 }
114. 114
115. 115 pub fn append(&self, row: &CostRow) -> std::io::Result<()> {
116. 116 if let Some(parent) = self.path.parent() {
117. 117 std::fs::create_dir_all(parent)?;
118. 118 }
119. 119 self.compact_if_large()?;
120. 120 let line = serde_json::to_string(row)
121. 121 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
122. 122 let mut f = std::fs::OpenOptions::new()
123. 123 .create(true)
124. 124 .append(true)
125. 125 .open(&self.path)?;
126. 126 writeln!(f, "{line}")
127. 127 }
128. 128
129. 129 /// Bound the ledger's on-disk size: cheap to skip on every call (one
130. 130 /// `metadata()` stat), and only reads+rewrites the whole file the rare
131. 131 /// time it actually crosses the threshold. Drops rows older than
132. 132 /// [`COST_LOG_RETENTION_DAYS`]; never touches today's numbers.
133. 133 fn compact_if_large(&self) -> std::io::Result<()> {
134. 134 self.compact_if_larger_than(
135. 135 COST_LOG_COMPACT_THRESHOLD_BYTES,
136. 136 chrono::Duration::days(COST_LOG_RETENTION_DAYS),
137. 137 )
138. 138 }
139. 139
140. 140 /// Parameterized so tests can exercise compaction without writing
141. 141 /// megabytes of fixture rows first.
142. 142 fn compact_if_larger_than(
143. 143 &self,
144. 144 threshold_bytes: u64,
145. 145 retention: chrono::Duration,
146. 146 ) -> std::io::Result<()> {
147. 147 let Ok(meta) = std::fs::metadata(&self.path) else {
148. 148 return Ok(());
149. 149 };
150. 150 if meta.len() < threshold_bytes {
151. 151 return Ok(());
152. 152 }
153. 153 let cutoff = chrono::Utc::now() - retention;
154. 154 let kept = self
155. 155 .all_rows()
156. 156 .into_iter()
157. 157 .filter(|r| r.ts >= cutoff)
158. 158 .collect::<Vec<_>>();
159. 159 let tmp = self.path.with_extension("jsonl.compact.tmp");
160. 160 {
161. 161 let mut f = std::fs::File::create(&tmp)?;
162. 162 for row in &kept {
163. 163 let line = serde_json::to_string(row)
164. 164 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
165. 165 writeln!(f, "{line}")?;
166. 166 }
167. 167 }
168. 168 std::fs::rename(&tmp, &self.path)
169. 169 }
170. 170
171. 171 /// USD total over rows at or after `since`. Unpriced rows contribute
172. 172 /// nothing but are never treated as evidence of zero spend elsewhere.
173. 173 pub fn total_usd_since(&self, since: chrono::DateTime<chrono::Utc>) -> f64 {
174. 174 let Ok(f) = std::fs::File::open(&self.path) else {
175. 175 return 0.0;
176. 176 };
177. 177 let mut total = 0.0;
178. 178 for line in BufReader::new(f).lines().map_while(Result::ok) {
179. 179 if let Ok(row) = serde_json::from_str::<CostRow>(&line)
180. 180 && row.ts >= since
181. 181 && let Some(usd) = row.usd
182. 182 {
183. 183 total += usd;
184. 184 }
185. 185 }
186. 186 total
187. 187 }
188. 188
189. 189 pub fn day_total_usd(&self, now: chrono::DateTime<chrono::Utc>) -> f64 {
190. 190 let day_start = now
191. 191 .date_naive()
192. 192 .and_hms_opt(0, 0, 0)
193. 193 .and_then(|t| t.and_local_timezone(chrono::Utc).single());
194. 194 match day_start {
195. 195 Some(start) => self.total_usd_since(start),
196. 196 None => 0.0,
197. 197 }
198. 198 }
199. 199
200. 200 /// Every row from the ledger, oldest first, for callers that need to
201. 201 /// bucket or roll them up themselves (admin console trend chart,
202. 202 /// provider/model breakdowns) rather than a single aggregate. Corrupt
203. 203 /// lines are skipped, same tolerance `total_usd_since` already has.
204. 204 pub fn all_rows(&self) -> Vec<CostRow> {
205. 205 let Ok(f) = std::fs::File::open(&self.path) else {
206. 206 return Vec::new();
207. 207 };
208. 208 BufReader::new(f)
209. 209 .lines()
210. 210 .map_while(Result::ok)
211. 211 .filter_map(|line| serde_json::from_str::<CostRow>(&line).ok())
212. 212 .collect()
213. 213 }
214. 214
215. 215 /// One USD total per UTC calendar day, oldest first, for the trailing
216. 216 /// `days` days including today — a fixed-length series so a chart never
217. 217 /// has to guess whether a missing day means "no spend" or "no data
218. 218 /// yet"; both render as `0.0`.
219. 219 pub fn daily_totals(
220. 220 &self,
221. 221 now: chrono::DateTime<chrono::Utc>,
222. 222 days: u32,
223. 223 ) -> Vec<(chrono::NaiveDate, f64)> {
224. 224 let today = now.date_naive();
225. 225 let mut totals: BTreeMap<chrono::NaiveDate, f64> = BTreeMap::new();
226. 226 for row in self.all_rows() {
227. 227 if let Some(usd) = row.usd {
228. 228 *totals.entry(row.ts.date_naive()).or_insert(0.0) += usd;
229. 229 }
230. 230 }
231. 231 (0..days)
232. 232 .rev()
233. 233 .filter_map(|offset| today.checked_sub_signed(chrono::Duration::days(offset as i64)))
234. 234 .map(|day| (day, totals.get(&day).copied().unwrap_or(0.0)))
235. 235 .collect()
236. 236 }
237. 237}
238. 238
239. 239/// Process-wide, cross-session admission state for the day cap, shared by
240. 240/// every [`CoreSpendGate`] built from the same `Core`. Closes two gaps a
241. 241/// gate-local `Mutex<f64>` can't: (1) a per-turn gate used to start the
242. 242/// run cap over from zero every turn (`Core::spend_gate_for` now hands
243. 243/// back the SAME gate for the life of a session instead, so `run_spent_usd`
244. 244/// finally means "this run", not "this turn"); (2) concurrent dispatches
245. 245/// (parallel tool calls, or multiple sessions sharing one `Core`) used to
246. 246/// all read the same stale on-disk day total and could jointly blow past
247. 247/// the cap before any of their rows landed — `reserved_usd` below is
248. 248/// credited at admission time, before the paid call happens, so a second
249. 249/// concurrent `authorize()` sees the first one's reservation immediately.
250. 250pub(crate) struct DayBudget {
251. 251 day: chrono::NaiveDate,
252. 252 /// Settled USD for `day`, read from the ledger once per day (not once
253. 253 /// per dispatch — this used to be a full-file re-scan on every single
254. 254 /// `authorize()` call).
255. 255 baseline_usd: f64,
256. 256 /// Admitted-but-not-yet-appended estimates for `day`. Rolled into
257. 257 /// `baseline_usd` (approximately — via the settled estimate, not the
258. 258 /// exact reservation) as each dispatch settles; a rollover to a new
259. 259 /// day always re-derives `baseline_usd` from the ledger, so any drift
260. 260 /// self-corrects at most once a day.
261. 261 reserved_usd: f64,
262. 262}
263. 263
264. 264impl DayBudget {
265. 265 /// A tracker with a deliberately-stale sentinel day, so the first
266. 266 /// `roll()` always re-derives `baseline_usd` from the ledger.
267. 267 pub(crate) fn new() -> Self {
268. 268 DayBudget {
269. 269 day: chrono::NaiveDate::from_ymd_opt(1970, 1, 1).unwrap_or_default(),
270. 270 baseline_usd: 0.0,
271. 271 reserved_usd: 0.0,
272. 272 }
273. 273 }
274. 274
275. 275 fn roll(&mut self, ledger: &FinOpsLedger, now: chrono::DateTime<chrono::Utc>) {
276. 276 let today = now.date_naive();
277. 277 if self.day != today {
278. 278 self.day = today;
279. 279 self.baseline_usd = ledger.day_total_usd(now);
280. 280 self.reserved_usd = 0.0;
281. 281 }
282. 282 }
283. 283}
284. 284
285. 285/// Core-side SpendGate: run/day caps + pricing-driven estimates.
286. 286pub struct CoreSpendGate {
287. 287 ledger: FinOpsLedger,
288. 288 max_run_usd: Mutex<Option<f64>>,
289. 289 max_day_usd: Mutex<Option<f64>>,
290. 290 overrides: Mutex<BTreeMap<String, vak_config::PriceEntry>>,
291. 291 run_spent_usd: Mutex<f64>,
292. 292 raised_once: AtomicBool,
293. 293 day_budget: Arc<Mutex<DayBudget>>,
294. 294}
295. 295
296. 296impl CoreSpendGate {
297. 297 pub fn new(sessions_home: &std::path::Path, finops: &vak_config::FinopsResolved) -> Self {
298. 298 Self::with_shared_day_budget(
299. 299 sessions_home,
300. 300 finops,
301. 301 Arc::new(Mutex::new(DayBudget::new())),
302. 302 )
303. 303 }
304. 304
305. 305 /// Same as [`Self::new`] but sharing the day-cap admission state with
306. 306 /// every other gate built from the same `Core` (see [`DayBudget`]).
307. 307 /// `Core::spend_gate_for` is the only caller that needs this; direct
308. 308 /// `new` (tests, the one-off reflection gate) is fine with its own
309. 309 /// isolated tracker since nothing else observes it.
310. 310 pub(crate) fn with_shared_day_budget(
311. 311 sessions_home: &std::path::Path,
312. 312 finops: &vak_config::FinopsResolved,
313. 313 day_budget: Arc<Mutex<DayBudget>>,
314. 314 ) -> Self {
315. 315 CoreSpendGate {
316. 316 ledger: FinOpsLedger::new(sessions_home),
317. 317 max_run_usd: Mutex::new(finops.max_run_usd),
318. 318 max_day_usd: Mutex::new(finops.max_day_usd),
319. 319 overrides: Mutex::new(finops.price_overrides.clone()),
320. 320 run_spent_usd: Mutex::new(0.0),
321. 321 raised_once: AtomicBool::new(false),
322. 322 day_budget,
323. 323 }
324. 324 }
325. 325
326. 326 /// Pick up a live `PATCH /finops` cap change on a gate that is being
327. 327 /// reused across turns (see [`DayBudget`] doc). Deliberately leaves
328. 328 /// `run_spent_usd`/`raised_once` untouched — a cap edit mid-run must
329. 329 /// not reset what's already been spent or re-arm a denial the
330. 330 /// approver already raised.
331. 331 pub(crate) fn refresh_caps(&self, finops: &vak_config::FinopsResolved) {
332. 332 *self
333. 333 .max_run_usd
334. 334 .lock()
335. 335 .unwrap_or_else(std::sync::PoisonError::into_inner) = finops.max_run_usd;
336. 336 *self
337. 337 .max_day_usd
338. 338 .lock()
339. 339 .unwrap_or_else(std::sync::PoisonError::into_inner) = finops.max_day_usd;
340. 340 *self
341. 341 .overrides
342. 342 .lock()
343. 343 .unwrap_or_else(std::sync::PoisonError::into_inner) = finops.price_overrides.clone();
344. 344 }
345. 345
346. 346 /// The approver answered "raise the cap for this run once".
347. 347 pub fn raise_once(&self) {
348. 348 self.raised_once.store(true, Ordering::SeqCst);
349. 349 }
350. 350
351. 351 /// Lower the run cap to `cap` for this gate's session, never raise it.
352. 352 ///
353. 353 /// An envelope's lifetime spend limit reaches the turn through here:
354. 354 /// the configured `max_run_usd` and the engagement's `spend_ceiling_usd`
355. 355 /// meet, and the smaller governs.
356. 356 pub fn narrow_run_cap(&self, cap: f64) {
357. 357 let mut current = self
358. 358 .max_run_usd
359. 359 .lock()
360. 360 .unwrap_or_else(std::sync::PoisonError::into_inner);
361. 361 *current = Some(current.map_or(cap, |existing| existing.min(cap)));
362. 362 }
363. 363
364. 364 /// Estimated cost of `usage` on `model` under the configured prices,
365. 365 /// or `None` when the model's price is unknown.
366. 366 pub fn estimate_usd(&self, model: &str, usage: &Usage) -> Option<f64> {
367. 367 self.estimate(model, usage)
368. 368 }
369. 369
370. 370 fn estimate(&self, model: &str, usage: &Usage) -> Option<f64> {
371. 371 let overrides = self
372. 372 .overrides
373. 373 .lock()
374. 374 .unwrap_or_else(std::sync::PoisonError::into_inner);
375. 375 vak_config::finops::estimate_cost_usd(model, usage, &overrides)
376. 376 }
377. 377}
378. 378
379. 379#[async_trait::async_trait]
380. 380impl SpendGate for CoreSpendGate {
381. 381 async fn authorize(&self, check: &SpendCheck<'_>) -> Result<(), String> {
382. 382 let planned = Usage {
383. 383 input_tokens: check.est_input_tokens,
384. 384 output_tokens: check.planned_output_tokens,
385. 385 ..Default::default()
386. 386 };
387. 387 let max_run_usd = *self
388. 388 .max_run_usd
389. 389 .lock()
390. 390 .unwrap_or_else(std::sync::PoisonError::into_inner);
391. 391 let run_spent = *self
392. 392 .run_spent_usd
393. 393 .lock()
394. 394 .unwrap_or_else(std::sync::PoisonError::into_inner);
395. 395 let max_day_usd = *self
396. 396 .max_day_usd
397. 397 .lock()
398. 398 .unwrap_or_else(std::sync::PoisonError::into_inner);
399. 399 let Some(est) = self.estimate(check.model, &planned) else {
400. 400 return if max_run_usd.is_some() || max_day_usd.is_some() {
401. 401 Err(format!(
402. 402 "cannot enforce dollar budget: model '{}' has unknown pricing; configure a price override before dispatch",
403. 403 check.model
404. 404 ))
405. 405 } else {
406. 406 Ok(())
407. 407 };
408. 408 };
409. 409 if let Some(cap) = max_run_usd
410. 410 && !self.raised_once.load(Ordering::SeqCst)
411. 411 && run_spent + est > cap
412. 412 {
413. 413 return Err(format!(
414. 414 "run budget ${cap:.2} would be exceeded by this dispatch (+${est:.2}, ${run_spent:.2} already spent)"
415. 415 ));
416. 416 }
417. 417 let max_day_usd = *self
418. 418 .max_day_usd
419. 419 .lock()
420. 420 .unwrap_or_else(std::sync::PoisonError::into_inner);
421. 421 if let Some(cap) = max_day_usd {
422. 422 let mut day = self
423. 423 .day_budget
424. 424 .lock()
425. 425 .unwrap_or_else(std::sync::PoisonError::into_inner);
426. 426 day.roll(&self.ledger, chrono::Utc::now());
427. 427 let projected = day.baseline_usd + day.reserved_usd + est;
428. 428 if projected > cap {
429. 429 return Err(format!(
430. 430 "day budget ${cap:.2} would be exceeded by this dispatch (+${est:.2}, ${:.2} spent today)",
431. 431 day.baseline_usd + day.reserved_usd
432. 432 ));
433. 433 }
434. 434 // Reserve immediately so a concurrent authorize() racing this
435. 435 // one sees the commitment before either dispatch settles.
436. 436 day.reserved_usd += est;
437. 437 }
438. 438 Ok(())
439. 439 }
440. 440
441. 441 fn record_settled(&self, provider: &str, model: &str, session_id: &str, usage: &Usage) {
442. 442 self.record_settled_with_latency(provider, model, session_id, usage, 0);
443. 443 }
444. 444
445. 445 fn record_settled_with_latency(
446. 446 &self,
447. 447 provider: &str,
448. 448 model: &str,
449. 449 session_id: &str,
450. 450 usage: &Usage,
451. 451 latency_ms: u64,
452. 452 ) {
453. 453 let usd = self.estimate(model, usage);
454. 454 // Fail-closed accounting: a run/day cap must still hold even if
455. 455 // the ledger write below fails (e.g. disk full) — an I/O error
456. 456 // must not silently re-open the budget it was there to enforce.
457. 457 if let Some(usd) = usd {
458. 458 let mut spent = self
459. 459 .run_spent_usd
460. 460 .lock()
461. 461 .unwrap_or_else(std::sync::PoisonError::into_inner);
462. 462 *spent += usd;
463. 463 let mut day = self
464. 464 .day_budget
465. 465 .lock()
466. 466 .unwrap_or_else(std::sync::PoisonError::into_inner);
467. 467 day.roll(&self.ledger, chrono::Utc::now());
468. 468 // Release this dispatch's reservation and fold the settled
469. 469 // amount into the baseline; clamp guards a same-day estimate
470. 470 // mismatch (planned vs. actual tokens) from going negative.
471. 471 day.reserved_usd = (day.reserved_usd - usd).max(0.0);
472. 472 day.baseline_usd += usd;
473. 473 }
474. 474 let row = CostRow {
475. 475 ts: chrono::Utc::now(),
476. 476 model: model.to_string(),
477. 477 provider: provider.to_string(),
478. 478 input_tokens: usage.input_tokens,
479. 479 output_tokens: usage.output_tokens,
480. 480 cache_read_input_tokens: usage.cache_read_input_tokens,
481. 481 usd,
482. 482 source: "estimated".to_string(),
483. 483 session_id: session_id.to_string(),
484. 484 };
485. 485 if let Err(e) = self.ledger.append(&row) {
486. 486 // The ledger write itself is still best-effort — the receipt
487. 487 // entries in the session log carry usage independently — but
488. 488 // the in-memory run/day counters above are already updated,
489. 489 // so caps stay enforced even when this fails.
490. 490 eprintln!("warning: cost ledger append failed: {e}");
491. 491 }
492. 492 let _ = ActivityLedger::new(
493. 493 self.ledger
494. 494 .path
495. 495 .parent()
496. 496 .unwrap_or(self.ledger.path.as_path()),
497. 497 )
498. 498 .append(&ActivityRow {
499. 499 ts: chrono::Utc::now(),
500. 500 kind: "provider".into(),
501. 501 name: format!("{provider}/{model}"),
502. 502 success: true,
503. 503 duration_ms: (latency_ms > 0).then_some(latency_ms),
504. 504 session_id: Some(session_id.to_string()),
505. 505 plugin: None,
506. 506 });
507. 507 }
508. 508}
509. 509
510. 510// ---- Budget alerts (docs/design/29-personal-os.md P2) ----------------------
511. 511
512. 512/// Proactive spend-alert thresholds, delivered to surfaces once per
513. 513/// threshold window instead of being discovered at denial time.
514. 514#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
515. 515#[serde(rename_all = "snake_case")]
516. 516pub enum AlertLevel {
517. 517 Eighty,
518. 518 Full,
519. 519}
520. 520
521. 521impl AlertLevel {
522. 522 pub fn as_str(&self) -> &'static str {
523. 523 match self {
524. 524 AlertLevel::Eighty => "eighty",
525. 525 AlertLevel::Full => "full",
526. 526 }
527. 527 }
528. 528}
529. 529
530. 530/// Which threshold `day_total_usd` has crossed against `cap`. Pure:
531. 531/// `>= 100%` → Full, `>= 80%` → Eighty, otherwise None. A non-positive or
532. 532/// non-finite cap means "no cap", which can never alert; a non-finite
533. 533/// total likewise.
534. 534pub fn alert_level(day_total_usd: f64, cap: f64) -> Option<AlertLevel> {
535. 535 if !cap.is_finite() || cap <= 0.0 || !day_total_usd.is_finite() {
536. 536 return None;
537. 537 }
538. 538 if day_total_usd >= cap {
539. 539 Some(AlertLevel::Full)
540. 540 } else if day_total_usd / cap >= 0.8 {
541. 541 Some(AlertLevel::Eighty)
542. 542 } else {
543. 543 None
544. 544 }
545. 545}
546. 546
547. 547/// One audit-only budget-alert ledger row. Lives beside the cost log in
548. 548/// `<home>/budget-alerts.jsonl`; never enters session logs, so projections
549. 549/// are untouched. The `"kind"` tag mirrors session receipt entries so
550. 550/// ledger consumers discriminate rows uniformly.
551. 551#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
552. 552pub struct BudgetAlertRow {
553. 553 #[serde(rename = "kind")]
554. 554 kind: String,
555. 555 pub ts: chrono::DateTime<chrono::Utc>,
556. 556 pub level: AlertLevel,
557. 557 /// Day spend as observed when the alert fired.
558. 558 pub day_total_usd: f64,
559. 559 pub session_id: String,
560. 560}
561. 561
562. 562fn alerts_path(home: &std::path::Path) -> PathBuf {
563. 563 home.join("budget-alerts.jsonl")
564. 564}
565. 565
566. 566/// Append an alert row for `level`, stamping it with the CURRENT day
567. 567/// spend from the cost ledger. Returns the row as written.
568. 568pub fn record_alert(
569. 569 home: &std::path::Path,
570. 570 level: AlertLevel,
571. 571 session_id: &str,
572. 572) -> std::io::Result<BudgetAlertRow> {
573. 573 std::fs::create_dir_all(home)?;
574. 574 compact_alerts_if_large(home)?;
575. 575 let row = BudgetAlertRow {
576. 576 kind: "budget_alert".to_string(),
577. 577 ts: chrono::Utc::now(),
578. 578 level,
579. 579 day_total_usd: FinOpsLedger::new(home).day_total_usd(chrono::Utc::now()),
580. 580 session_id: session_id.to_string(),
581. 581 };
582. 582 let line = serde_json::to_string(&row)
583. 583 .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
584. 584 let mut f = std::fs::OpenOptions::new()
585. 585 .create(true)
586. 586 .append(true)
587. 587 .open(alerts_path(home))?;
588. 588 writeln!(f, "{line}")?;
589. 589 Ok(row)
590. 590}
591. 591
592. 592/// Same bounded-growth treatment as [`FinOpsLedger::compact_if_large`]:
593. 593/// alerts are only ever read for "most recent N", so unbounded history
594. 594/// buys nothing but disk and scan time.
595. 595fn compact_alerts_if_large(home: &std::path::Path) -> std::io::Result<()> {
596. 596 let path = alerts_path(home);
597. 597 let Ok(meta) = std::fs::metadata(&path) else {
598. 598 return Ok(());
599. 599 };
600. 600 if meta.len() < ALERTS_COMPACT_THRESHOLD_BYTES {
601. 601 return Ok(());
602. 602 }
603. 603 let Ok(f) = std::fs::File::open(&path) else {
604. 604 return Ok(());
605. 605 };
606. 606 let mut lines: Vec<String> = BufReader::new(f).lines().map_while(Result::ok).collect();
607. 607 if lines.len() <= ALERTS_RETENTION_ROWS {
608. 608 return Ok(());
609. 609 }
610. 610 let drop = lines.len() - ALERTS_RETENTION_ROWS;
611. 611 lines.drain(0..drop);
612. 612 let tmp = path.with_extension("jsonl.compact.tmp");
613. 613 std::fs::write(&tmp, lines.join("\n") + "\n")?;
614. 614 std::fs::rename(&tmp, &path)
615. 615}
616. 616
617. 617/// Most recent recorded alert at exactly `level`, for once-per-window
618. 618/// firing decisions. Corrupt lines are skipped; a missing file is None.
619. 619pub fn last_alert(home: &std::path::Path, level: AlertLevel) -> Option<BudgetAlertRow> {
620. 620 let f = std::fs::File::open(alerts_path(home)).ok()?;
621. 621 let mut found = None;
622. 622 for line in BufReader::new(f).lines().map_while(Result::ok) {
623. 623 if let Ok(row) = serde_json::from_str::<BudgetAlertRow>(&line)
624. 624 && row.kind == "budget_alert"
625. 625 && row.level == level
626. 626 {
627. 627 found = Some(row);
628. 628 }
629. 629 }
630. 630 found
631. 631}
632. 632
633. 633#[cfg(test)]
634. 634mod tests {
635. 635
636. 636 #![allow(clippy::unwrap_used, clippy::expect_used)]
637. 637 use super::*;
638. 638 use tempfile::tempdir;
639. 639
640. 640 fn gate_with(caps: &[(&str, f64)]) -> (CoreSpendGate, tempfile::TempDir) {
641. 641 let dir = tempdir().unwrap();
642. 642 let mut finops = vak_config::FinopsResolved::default();
643. 643 for (k, v) in caps {
644. 644 match *k {
645. 645 "run" => finops.max_run_usd = Some(*v),
646. 646 "day" => finops.max_day_usd = Some(*v),
647. 647 _ => unreachable!(),
648. 648 }
649. 649 }
650. 650 (CoreSpendGate::new(dir.path(), &finops), dir)
651. 651 }
652. 652
653. 653 fn usage(in_tok: u64, out_tok: u64) -> Usage {
654. 654 Usage {
655. 655 input_tokens: in_tok,
656. 656 output_tokens: out_tok,
657. 657 ..Default::default()
658. 658 }
659. 659 }
660. 660
661. 661 fn check(model: &'static str) -> SpendCheck<'static> {
662. 662 SpendCheck {
663. 663 model,
664. 664 provider: "anthropic",
665. 665 session_id: "s1",
666. 666 est_input_tokens: 1_000_000,
667. 667 planned_output_tokens: 100_000,
668. 668 }
669. 669 }
670. 670
671. 671 #[tokio::test]
672. 672 async fn unpriced_model_admits_and_records_unknown_usd() {
673. 673 let (gate, dir) = gate_with(&[]);
674. 674 gate.authorize(&check("mystery-model")).await.unwrap();
675. 675 gate.record_settled(
676. 676 "anthropic",
677. 677 "mystery-model",
678. 678 "s1",
679. 679 &usage(1_000_000, 100_000),
680. 680 );
681. 681 let ledger = FinOpsLedger::new(dir.path());
682. 682 assert_eq!(ledger.day_total_usd(chrono::Utc::now()), 0.0);
683. 683 let text = std::fs::read_to_string(dir.path().join("cost-log.jsonl")).unwrap();
684. 684 assert!(text.contains("\"usd\":null") || !text.contains("\"usd\""));
685. 685 }
686. 686
687. 687 #[tokio::test]
688. 688 async fn unpriced_model_cannot_bypass_either_dollar_cap() {
689. 689 for cap in ["run", "day"] {
690. 690 let (gate, _dir) = gate_with(&[(cap, 0.01)]);
691. 691 let error = gate.authorize(&check("mystery-model")).await.unwrap_err();
692. 692 assert!(error.contains("unknown pricing"), "{error}");
693. 693 }
694. 694 }
695. 695
696. 696 #[tokio::test]
697. 697 async fn run_cap_denies_then_raise_once_admits() {
698. 698 // sonnet: $3/MTok in → est = 3*1 + 15*0.1 = $4.50 per dispatch.
699. 699 let (gate, _dir) = gate_with(&[("run", 5.0)]);
700. 700 gate.authorize(&check("claude-sonnet")).await.unwrap(); // 4.5 <= 5
701. 701 gate.record_settled(
702. 702 "anthropic",
703. 703 "claude-sonnet",
704. 704 "s1",
705. 705 &usage(1_000_000, 100_000),
706. 706 );
707. 707 let err = gate.authorize(&check("claude-sonnet")).await.unwrap_err();
708. 708 assert!(err.contains("run budget $5.00"), "{err}");
709. 709 gate.raise_once();
710. 710 gate.authorize(&check("claude-sonnet")).await.unwrap();
711. 711 }
712. 712
713. 713 #[tokio::test]
714. 714 async fn day_cap_counts_seeded_ledger_rows() {
715. 715 let (gate, dir) = gate_with(&[("day", 6.0)]);
716. 716 let ledger = FinOpsLedger::new(dir.path());
717. 717 ledger
718. 718 .append(&CostRow {
719. 719 ts: chrono::Utc::now(),
720. 720 model: "claude-sonnet".into(),
721. 721 provider: String::new(),
722. 722 input_tokens: 1_000_000,
723. 723 output_tokens: 100_000,
724. 724 cache_read_input_tokens: None,
725. 725 usd: Some(2.0),
726. 726 source: "estimated".into(),
727. 727 session_id: "seed".into(),
728. 728 })
729. 729 .unwrap();
730. 730 // est 4.50 + day 2.00 > 6.00 → denied with day wording.
731. 731 let err = gate.authorize(&check("claude-sonnet")).await.unwrap_err();
732. 732 assert!(err.contains("day budget $6.00"), "{err}");
733. 733 }
734. 734
735. 735 #[test]
736. 736 fn day_window_ignores_yesterday() {
737. 737 let dir = tempdir().unwrap();
738. 738 let ledger = FinOpsLedger::new(dir.path());
739. 739 let yesterday = chrono::Utc::now() - chrono::Duration::hours(30);
740. 740 ledger
741. 741 .append(&CostRow {
742. 742 ts: yesterday,
743. 743 model: "m".into(),
744. 744 provider: String::new(),
745. 745 input_tokens: 1,
746. 746 output_tokens: 1,
747. 747 cache_read_input_tokens: None,
748. 748 usd: Some(99.0),
749. 749 source: "estimated".into(),
750. 750 session_id: "old".into(),
751. 751 })
752. 752 .unwrap();
753. 753 assert_eq!(ledger.day_total_usd(chrono::Utc::now()), 0.0);
754. 754 }
755. 755
756. 756 fn row_on(day: chrono::NaiveDate, usd: Option<f64>) -> CostRow {
757. 757 CostRow {
758. 758 ts: day.and_hms_opt(12, 0, 0).unwrap().and_utc(),
759. 759 model: "m".into(),
760. 760 provider: "p".into(),
761. 761 input_tokens: 1,
762. 762 output_tokens: 1,
763. 763 cache_read_input_tokens: None,
764. 764 usd,
765. 765 source: "estimated".into(),
766. 766 session_id: "s".into(),
767. 767 }
768. 768 }
769. 769
770. 770 /// The chart-feeding series: always exactly `days` entries, oldest
771. 771 /// first ending at today, zero-filled for a day with no rows — a
772. 772 /// sparse map would leave a chart guessing which days are "no spend"
773. 773 /// versus simply absent.
774. 774 #[test]
775. 775 fn daily_totals_is_fixed_length_and_zero_fills_gaps() {
776. 776 let dir = tempdir().unwrap();
777. 777 let ledger = FinOpsLedger::new(dir.path());
778. 778 let now = chrono::Utc::now();
779. 779 let today = now.date_naive();
780. 780 let two_days_ago = today - chrono::Duration::days(2);
781. 781 ledger.append(&row_on(today, Some(3.0))).unwrap();
782. 782 ledger.append(&row_on(today, Some(1.5))).unwrap();
783. 783 ledger.append(&row_on(two_days_ago, Some(2.0))).unwrap();
784. 784 // Unpriced rows must not silently count as zero spend where a
785. 785 // priced row exists, nor crash the bucketing.
786. 786 ledger.append(&row_on(today, None)).unwrap();
787. 787
788. 788 let series = ledger.daily_totals(now, 3);
789. 789 assert_eq!(series.len(), 3);
790. 790 assert_eq!(series[2].0, today);
791. 791 assert!((series[2].1 - 4.5).abs() < 1e-9, "{:?}", series[2]);
792. 792 assert_eq!(series[1].0, today - chrono::Duration::days(1));
793. 793 assert_eq!(series[1].1, 0.0, "a day with no rows must zero-fill");
794. 794 assert_eq!(series[0].0, two_days_ago);
795. 795 assert!((series[0].1 - 2.0).abs() < 1e-9);
796. 796 }
797. 797
798. 798 #[test]
799. 799 fn alert_level_threshold_matrix() {
800. 800 assert_eq!(alert_level(0.0, 10.0), None);
801. 801 assert_eq!(alert_level(7.9, 10.0), None);
802. 802 // Exactly at the thresholds fires.
803. 803 assert_eq!(alert_level(8.0, 10.0), Some(AlertLevel::Eighty));
804. 804 assert_eq!(alert_level(9.99, 10.0), Some(AlertLevel::Eighty));
805. 805 assert_eq!(alert_level(10.0, 10.0), Some(AlertLevel::Full));
806. 806 assert_eq!(alert_level(150.0, 10.0), Some(AlertLevel::Full));
807. 807 // No cap / nonsense inputs never alert.
808. 808 assert_eq!(alert_level(100.0, 0.0), None);
809. 809 assert_eq!(alert_level(100.0, -5.0), None);
810. 810 assert_eq!(alert_level(f64::NAN, 10.0), None);
811. 811 }
812. 812
813. 813 #[test]
814. 814 fn record_and_last_alert_roundtrip_per_level() {
815. 815 let dir = tempdir().unwrap();
816. 816 let home = dir.path();
817. 817
818. 818 assert_eq!(last_alert(home, AlertLevel::Eighty), None);
819. 819
820. 820 let first = record_alert(home, AlertLevel::Eighty, "s1").unwrap();
821. 821 assert_eq!(first.kind, "budget_alert");
822. 822 assert_eq!(first.level, AlertLevel::Eighty);
823. 823
824. 824 let full = record_alert(home, AlertLevel::Full, "s1").unwrap();
825. 825 let later = record_alert(home, AlertLevel::Eighty, "s2").unwrap();
826. 826
827. 827 let eighty = last_alert(home, AlertLevel::Eighty).unwrap();
828. 828 assert_eq!(eighty.session_id, "s2");
829. 829 assert_eq!(eighty.ts, later.ts);
830. 830 let full_back = last_alert(home, AlertLevel::Full).unwrap();
831. 831 assert_eq!(full_back.ts, full.ts);
832. 832 assert_eq!(full_back.day_total_usd, full.day_total_usd);
833. 833 // Rows carry the current ledger day total (zero here).
834. 834 assert_eq!(first.day_total_usd, 0.0);
835. 835 }
836. 836
837. 837 #[test]
838. 838 fn corrupt_or_foreign_lines_are_ignored_by_readback() {
839. 839 let dir = tempdir().unwrap();
840. 840 std::fs::write(
841. 841 alerts_path(dir.path()),
842. 842 concat!(
843. 843 "{not json\n",
844. 844 "{\"kind\":\"receipt\",\"other\":1}\n",
845. 845 "{\"kind\":\"budget_alert\",\"ts\":\"2026-08-24T00:00:00Z\",\"level\":\"eighty\",\"day_total_usd\":1.5,\"session_id\":\"seed\"}\n",
846. 846 ),
847. 847 )
848. 848 .unwrap();
849. 849 let found = last_alert(dir.path(), AlertLevel::Eighty).unwrap();
850. 850 assert_eq!(found.session_id, "seed");
851. 851 assert!((found.day_total_usd - 1.5).abs() < 1e-9);
852. 852 assert!(last_alert(dir.path(), AlertLevel::Full).is_none());
853. 853 }
854. 854
855. 855 /// Audit fix: the ledger used to grow forever with no rotation, and
856. 856 /// every `authorize()` re-read the whole file from disk on every paid
857. 857 /// dispatch. Compaction should trim old rows once the file crosses
858. 858 /// its size threshold, and must never drop anything within the
859. 859 /// retention window.
860. 860 #[test]
861. 861 fn compaction_drops_only_rows_older_than_retention() {
862. 862 let dir = tempdir().unwrap();
863. 863 let ledger = FinOpsLedger::new(dir.path());
864. 864 let now = chrono::Utc::now();
865. 865 let old = now - chrono::Duration::days(400);
866. 866 ledger.append(&row_on(old.date_naive(), Some(1.0))).unwrap();
867. 867 ledger.append(&row_on(now.date_naive(), Some(2.0))).unwrap();
868. 868
869. 869 // Force compaction on the next append regardless of actual file
870. 870 // size, with a short retention window so the seeded old row falls
871. 871 // outside it.
872. 872 ledger
873. 873 .compact_if_larger_than(0, chrono::Duration::days(1))
874. 874 .unwrap();
875. 875
876. 876 let rows = ledger.all_rows();
877. 877 assert_eq!(
878. 878 rows.len(),
879. 879 1,
880. 880 "the old row must be dropped, not the fresh one"
881. 881 );
882. 882 assert_eq!(rows[0].usd, Some(2.0));
883. 883 // Aggregates must be unaffected by compaction for anything still
884. 884 // within the retention window.
885. 885 assert_eq!(ledger.day_total_usd(now), 2.0);
886. 886 }
887. 887
888. 888 #[test]
889. 889 fn compaction_is_a_noop_below_the_size_threshold() {
890. 890 let dir = tempdir().unwrap();
891. 891 let ledger = FinOpsLedger::new(dir.path());
892. 892 let now = chrono::Utc::now();
893. 893 let old = now - chrono::Duration::days(400);
894. 894 ledger.append(&row_on(old.date_naive(), Some(1.0))).unwrap();
895. 895
896. 896 // A generous threshold the tiny fixture file can never cross:
897. 897 // compaction must leave old-but-still-present rows alone.
898. 898 ledger
899. 899 .compact_if_larger_than(u64::MAX, chrono::Duration::days(1))
900. 900 .unwrap();
901. 901 assert_eq!(ledger.all_rows().len(), 1);
902. 902 }
903. 903
904. 904 /// Audit fix: concurrent dispatches used to each read the same stale
905. 905 /// on-disk day total before any of them settled, so a burst could
906. 906 /// jointly blow past `max_day_usd`. `DayBudget::reserved_usd` credits
907. 907 /// an admission immediately so a second concurrent `authorize()` sees
908. 908 /// the first one's reservation before either settles.
909. 909 #[tokio::test]
910. 910 async fn concurrent_authorize_calls_cannot_jointly_exceed_the_day_cap() {
911. 911 let dir = tempdir().unwrap();
912. 912 // sonnet est ~= $4.50/dispatch; cap admits exactly one.
913. 913 let finops = vak_config::FinopsResolved {
914. 914 max_day_usd: Some(5.0),
915. 915 ..Default::default()
916. 916 };
917. 917 let day_budget = Arc::new(Mutex::new(DayBudget::new()));
918. 918 let gate_a = CoreSpendGate::with_shared_day_budget(dir.path(), &finops, day_budget.clone());
919. 919 let gate_b = CoreSpendGate::with_shared_day_budget(dir.path(), &finops, day_budget);
920. 920
921. 921 // Both "concurrent" calls check against the ledger before either
922. 922 // has appended anything — a stale-read race would admit both.
923. 923 let first = gate_a.authorize(&check("claude-sonnet")).await;
924. 924 let second = gate_b.authorize(&check("claude-sonnet")).await;
925. 925 assert!(first.is_ok(), "{first:?}");
926. 926 assert!(
927. 927 second.is_err(),
928. 928 "the second concurrent dispatch must see the first one's reservation"
929. 929 );
930. 930 }
931. 931
932. 932 /// `record_settled` releases a dispatch's reservation and folds the
933. 933 /// settled amount into the day baseline; a same-day gate built after
934. 934 /// the first must see the earlier settlement.
935. 935 #[tokio::test]
936. 936 async fn record_settled_updates_the_shared_day_budget_for_later_gates() {
937. 937 let dir = tempdir().unwrap();
938. 938 // sonnet est ~= $4.50/dispatch; cap admits exactly one, so a
939. 939 // gate that doesn't see gate_a's settlement would wrongly admit
940. 940 // a second one at ~$9.00 total.
941. 941 let finops = vak_config::FinopsResolved {
942. 942 max_day_usd: Some(5.0),
943. 943 ..Default::default()
944. 944 };
945. 945 let day_budget = Arc::new(Mutex::new(DayBudget::new()));
946. 946 let gate_a = CoreSpendGate::with_shared_day_budget(dir.path(), &finops, day_budget.clone());
947. 947 gate_a.authorize(&check("claude-sonnet")).await.unwrap();
948. 948 gate_a.record_settled(
949. 949 "anthropic",
950. 950 "claude-sonnet",
951. 951 "s1",
952. 952 &usage(1_000_000, 100_000),
953. 953 );
954. 954
955. 955 // A later gate sharing the same tracker (e.g. the next turn's
956. 956 // session, or a different session on the same Core) must see
957. 957 // gate_a's settled spend, not a fresh zero.
958. 958 let gate_b = CoreSpendGate::with_shared_day_budget(dir.path(), &finops, day_budget);
959. 959 let err = gate_b
960. 960 .authorize(&check("claude-sonnet"))
961. 961 .await
962. 962 .expect_err("day cap must already reflect gate_a's settled spend");
963. 963 assert!(err.contains("day budget $5.00"), "{err}");
964. 964 }
965. 965
966. 966 #[test]
967. 967 fn activity_ledger_round_trips_observed_rows() {
968. 968 let dir = tempdir().unwrap();
969. 969 let ledger = ActivityLedger::new(dir.path());
970. 970 ledger
971. 971 .append(&ActivityRow {
972. 972 ts: chrono::Utc::now(),
973. 973 kind: "mcp".into(),
974. 974 name: "server/tool".into(),
975. 975 success: false,
976. 976 duration_ms: Some(42),
977. 977 session_id: Some("s1".into()),
978. 978 plugin: Some("demo".into()),
979. 979 })
980. 980 .unwrap();
981. 981 let rows = ledger.all_rows();
982. 982 assert_eq!(rows.len(), 1);
983. 983 assert_eq!(rows[0].duration_ms, Some(42));
984. 984 assert_eq!(rows[0].plugin.as_deref(), Some("demo"));
985. 985 }
986. 986}
987. 987