1. 1//! Scheduler upgrade behaviors (docs/design/29-personal-os.md P2): cron
2. 2//! schedules with startup catch-up, zero-token watchdog script tasks over
3. 3//! the brokered bash path, per-task model pinning verified through work
4. 4//! receipts, and once-per-window budget alerts.
5. 5
6. 6#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
7. 7
8. 8use std::io::BufRead;
9. 9use std::path::{Path, PathBuf};
10. 10use std::sync::{
11. 11 Arc,
12. 12 atomic::{AtomicUsize, Ordering},
13. 13};
14. 14
15. 15use tokio_util::sync::CancellationToken;
16. 16
17. 17use vak_core::Core;
18. 18use vak_llm::stream;
19. 19use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, Usage};
20. 20use vak_llm::{EventStream, LlmError, Provider};
21. 21
22. 22struct Counting {
23. 23 dispatches: Arc<AtomicUsize>,
24. 24}
25. 25
26. 26#[async_trait::async_trait]
27. 27impl Provider for Counting {
28. 28 fn name(&self) -> &str {
29. 29 "counting"
30. 30 }
31. 31
32. 32 async fn stream(
33. 33 &self,
34. 34 _request: ChatRequest,
35. 35 _cancel: CancellationToken,
36. 36 ) -> Result<EventStream, LlmError> {
37. 37 self.dispatches.fetch_add(1, Ordering::SeqCst);
38. 38 let (mut sink, rx) = stream::channel(8);
39. 39 let done = AssistantMessage {
40. 40 content: vec![ContentBlock::text("done")],
41. 41 stop_reason: vak_llm::types::StopReason::EndTurn,
42. 42 usage: Usage {
43. 43 input_tokens: 9,
44. 44 output_tokens: 1,
45. 45 ..Default::default()
46. 46 },
47. 47 model: "counted-model".into(),
48. 48 response_id: None,
49. 49 };
50. 50 sink.push(stream::StreamEvent::Start {
51. 51 partial: done.clone(),
52. 52 });
53. 53 sink.close_message(done).await;
54. 54 Ok(rx)
55. 55 }
56. 56}
57. 57
58. 58struct Srv {
59. 59 base: String,
60. 60 token: String,
61. 61 home: PathBuf,
62. 62 dispatches: Arc<AtomicUsize>,
63. 63}
64. 64
65. 65impl Srv {
66. 66 fn client(&self) -> reqwest::Client {
67. 67 reqwest::ClientBuilder::new()
68. 68 .default_headers({
69. 69 let mut h = reqwest::header::HeaderMap::new();
70. 70 h.insert(
71. 71 reqwest::header::AUTHORIZATION,
72. 72 format!("Bearer {}", self.token).parse().unwrap(),
73. 73 );
74. 74 h
75. 75 })
76. 76 .build()
77. 77 .unwrap()
78. 78 }
79. 79
80. 80 async fn get_json(&self, path: &str) -> serde_json::Value {
81. 81 self.client()
82. 82 .get(format!("{}{path}", self.base))
83. 83 .send()
84. 84 .await
85. 85 .unwrap()
86. 86 .json()
87. 87 .await
88. 88 .unwrap()
89. 89 }
90. 90}
91. 91
92. 92fn git_seed(cwd: &Path) {
93. 93 let run = |args: &[&str]| {
94. 94 let out = std::process::Command::new("git")
95. 95 .args(args)
96. 96 .current_dir(cwd)
97. 97 .env("GIT_AUTHOR_NAME", "t")
98. 98 .env("GIT_AUTHOR_EMAIL", "t@t")
99. 99 .env("GIT_COMMITTER_NAME", "t")
100. 100 .env("GIT_COMMITTER_EMAIL", "t@t")
101. 101 .output()
102. 102 .unwrap();
103. 103 assert!(out.status.success(), "git {args:?} failed");
104. 104 };
105. 105 run(&["init", "-q"]);
106. 106 std::fs::write(cwd.join("README.md"), "seed\n").unwrap();
107. 107 run(&["add", "."]);
108. 108 run(&["commit", "-q", "-m", "seed"]);
109. 109}
110. 110
111. 111struct Fixture {
112. 112 srv: Srv,
113. 113 #[allow(dead_code)]
114. 114 dir: Arc<tempfile::TempDir>,
115. 115}
116. 116
117. 117/// Build a hermetic git workspace + home and start the FULL stack (bearer
118. 118/// auth + scheduler). `seed_tasks` receives `(home, ws)` BEFORE the server
119. 119/// starts — writing `<home>/tasks.json` there is the simulated-downtime
120. 120/// injection point, since the scheduler loads it once at startup.
121. 121async fn spawn_full(
122. 122 config_toml: &str,
123. 123 seed_tasks: Option<impl for<'a> FnOnce(&'a Path, &'a Path) -> serde_json::Value>,
124. 124) -> Fixture {
125. 125 let dir = Arc::new(tempfile::tempdir().unwrap());
126. 126 let ws = dir.path().join("ws");
127. 127 std::fs::create_dir_all(ws.join(".vak")).unwrap();
128. 128 std::fs::write(
129. 129 ws.join(".vak/config.toml"),
130. 130 format!("[memory]\nreflection = false\n{config_toml}"),
131. 131 )
132. 132 .unwrap();
133. 133 git_seed(&ws);
134. 134
135. 135 let home = dir.path().join("home");
136. 136 if let Some(build) = seed_tasks {
137. 137 std::fs::create_dir_all(&home).unwrap();
138. 138 std::fs::write(
139. 139 vak_core::tasks::tasks_file(&home),
140. 140 serde_json::to_string_pretty(&build(&home, &ws)).unwrap(),
141. 141 )
142. 142 .unwrap();
143. 143 }
144. 144
145. 145 let dispatches = Arc::new(AtomicUsize::new(0));
146. 146 vak_config::paths::isolate_home_for_tests();
147. 147 let core = Core::new_with_trust(ws.clone(), true).unwrap();
148. 148 core.set_sessions_home(home.clone());
149. 149 core.set_permission_mode(vak_config::PermissionMode::FullAccess);
150. 150 // A REAL worker executable: a cargo test harness cannot speak the
151. 151 // broker protocol.
152. 152 core.set_tool_worker_exe(PathBuf::from(env!("CARGO_BIN_EXE_vak-tool-worker")));
153. 153 core.set_provider_instance(Arc::new(Counting {
154. 154 dispatches: dispatches.clone(),
155. 155 }));
156. 156
157. 157 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
158. 158 let addr = listener.local_addr().unwrap();
159. 159 let (app, token) = vak_server::secured_router_with(core, false);
160. 160 tokio::spawn(async move {
161. 161 axum::serve(listener, app).await.unwrap();
162. 162 });
163. 163 Fixture {
164. 164 srv: Srv {
165. 165 base: format!("http://{addr}"),
166. 166 token,
167. 167 home,
168. 168 dispatches,
169. 169 },
170. 170 dir,
171. 171 }
172. 172}
173. 173
174. 174fn delivery_lines(home: &Path) -> Vec<(String, String)> {
175. 175 let Ok(f) = std::fs::File::open(home.join("gateway").join("deliveries.jsonl")) else {
176. 176 return Vec::new();
177. 177 };
178. 178 let mut out = Vec::new();
179. 179 for line in std::io::BufReader::new(f).lines().map_while(Result::ok) {
180. 180 if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) {
181. 181 out.push((
182. 182 v["target"].as_str().unwrap_or_default().to_string(),
183. 183 v["text"].as_str().unwrap_or_default().to_string(),
184. 184 ));
185. 185 }
186. 186 }
187. 187 out
188. 188}
189. 189
190. 190async fn wait_until(secs: u64, mut pred: impl FnMut() -> bool) -> bool {
191. 191 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs);
192. 192 while std::time::Instant::now() < deadline {
193. 193 if pred() {
194. 194 return true;
195. 195 }
196. 196 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
197. 197 }
198. 198 pred()
199. 199}
200. 200
201. 201async fn task_field(srv: &Srv, id: &str, field: &str) -> serde_json::Value {
202. 202 let body = srv.get_json("/tasks").await;
203. 203 body["tasks"]
204. 204 .as_array()
205. 205 .unwrap()
206. 206 .iter()
207. 207 .find(|t| t["id"] == id)
208. 208 .map(|t| t[field].clone())
209. 209 .unwrap_or(serde_json::Value::Null)
210. 210}
211. 211
212. 212async fn wait_summary(srv: &Srv, id: &str, secs: u64, pred: impl Fn(&str) -> bool) {
213. 213 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs);
214. 214 loop {
215. 215 assert!(
216. 216 std::time::Instant::now() < deadline,
217. 217 "task '{id}' summary never matched"
218. 218 );
219. 219 let v = task_field(srv, id, "last_summary").await;
220. 220 let matched = match v.as_str() {
221. 221 Some(s) => pred(s),
222. 222 None => false,
223. 223 };
224. 224 if matched {
225. 225 return;
226. 226 }
227. 227 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
228. 228 }
229. 229}
230. 230
231. 231fn stale_script_task(
232. 232 id: &str,
233. 233 name: &str,
234. 234 ws: &Path,
235. 235 deliver_to: &str,
236. 236 script: &str,
237. 237) -> serde_json::Value {
238. 238 serde_json::json!({
239. 239 "id": id,
240. 240 "name": name,
241. 241 "prompt": "",
242. 242 "interval_secs": 3600,
243. 243 "enabled": true,
244. 244 "cwd": ws.display().to_string(),
245. 245 "created_at": chrono::Utc::now().to_rfc3339(),
246. 246 // Two hours of downtime: every "* * * * *" slot since this instant
247. 247 // was missed.
248. 248 "last_run_at": (chrono::Utc::now() - chrono::Duration::hours(2)).to_rfc3339(),
249. 249 "last_session_id": null,
250. 250 "last_summary": null,
251. 251 "last_wt": null,
252. 252 "deliver_to": deliver_to,
253. 253 "schedule": "* * * * *",
254. 254 "script": script,
255. 255 "model_pin": null
256. 256 })
257. 257}
258. 258
259. 259// ---- Catch-up -----------------------------------------------------------------
260. 260
261. 261#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
262. 262async fn catch_up_fires_missed_cron_slot_exactly_once_with_zero_dispatch() {
263. 263 let fx = spawn_full(
264. 264 "",
265. 265 Some(|_home: &Path, ws: &Path| {
266. 266 serde_json::json!([stale_script_task(
267. 267 "cu-1",
268. 268 "nightly-watch",
269. 269 ws,
270. 270 "log:cu",
271. 271 "echo caught-up-42"
272. 272 )])
273. 273 }),
274. 274 )
275. 275 .await;
276. 276
277. 277 // The missed slot fires immediately at startup, stdout verbatim.
278. 278 assert!(
279. 279 wait_until(10, || delivery_lines(&fx.srv.home)
280. 280 .iter()
281. 281 .any(|(t, x)| t == "log:cu" && x.contains("caught-up-42")))
282. 282 .await,
283. 283 "catch-up never delivered"
284. 284 );
285. 285 // Exactly once: no duplicate delivery after things settle.
286. 286 tokio::time::sleep(std::time::Duration::from_millis(1200)).await;
287. 287 let hits = delivery_lines(&fx.srv.home)
288. 288 .iter()
289. 289 .filter(|(_, x)| x.contains("caught-up-42"))
290. 290 .count();
291. 291 assert_eq!(hits, 1, "catch-up must fire exactly once");
292. 292
293. 293 // Watchdogs never touch the LLM: zero provider dispatches overall.
294. 294 assert_eq!(fx.srv.dispatches.load(Ordering::SeqCst), 0);
295. 295
296. 296 // The record moved forward (no longer two hours stale).
297. 297 wait_summary(&fx.srv, "cu-1", 5, |s| s.contains("caught-up-42")).await;
298. 298 let last = task_field(&fx.srv, "cu-1", "last_run_at").await;
299. 299 let last = chrono::DateTime::parse_from_rfc3339(last.as_str().unwrap())
300. 300 .unwrap()
301. 301 .with_timezone(&chrono::Utc);
302. 302 assert!(
303. 303 chrono::Utc::now() - last < chrono::Duration::seconds(30),
304. 304 "last_run_at should be refreshed by the catch-up fire"
305. 305 );
306. 306}
307. 307
308. 308#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
309. 309async fn catch_up_disabled_waits_for_next_slot() {
310. 310 let fx = spawn_full(
311. 311 "[automation]\ncatch_up_missed = false\n",
312. 312 Some(|_home: &Path, ws: &Path| {
313. 313 serde_json::json!([stale_script_task(
314. 314 "cu-off",
315. 315 "quiet-watch",
316. 316 ws,
317. 317 "log:cuoff",
318. 318 "echo not-caught-up-99"
319. 319 )])
320. 320 }),
321. 321 )
322. 322 .await;
323. 323
324. 324 tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
325. 325 assert!(
326. 326 !delivery_lines(&fx.srv.home)
327. 327 .iter()
328. 328 .any(|(_, x)| x.contains("not-caught-up-99")),
329. 329 "disabled catch-up must not fire missed slots"
330. 330 );
331. 331 assert_eq!(fx.srv.dispatches.load(Ordering::SeqCst), 0);
332. 332}
333. 333
334. 334// ---- Watchdog script matrix -----------------------------------------------------
335. 335
336. 336async fn create_task(srv: &Srv, body: serde_json::Value) -> String {
337. 337 let res = srv
338. 338 .client()
339. 339 .post(format!("{}/tasks", srv.base))
340. 340 .json(&body)
341. 341 .send()
342. 342 .await
343. 343 .unwrap();
344. 344 assert_eq!(res.status(), 200, "create failed: {}", res.status());
345. 345 let list = srv.get_json("/tasks").await;
346. 346 list["tasks"]
347. 347 .as_array()
348. 348 .unwrap()
349. 349 .iter()
350. 350 .find(|t| t["name"] == body["name"])
351. 351 .map(|t| t["id"].as_str().unwrap().to_string())
352. 352 .unwrap()
353. 353}
354. 354
355. 355async fn run_now(srv: &Srv, id: &str) -> reqwest::StatusCode {
356. 356 srv.client()
357. 357 .post(format!("{}/tasks/{id}/run-now", srv.base))
358. 358 .send()
359. 359 .await
360. 360 .unwrap()
361. 361 .status()
362. 362}
363. 363
364. 364#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
365. 365async fn script_watchdog_matrix_silent_stdout_failure_and_delivery() {
366. 366 let fx = spawn_full("", None::<fn(&Path, &Path) -> serde_json::Value>).await;
367. 367 let srv = &fx.srv;
368. 368
369. 369 let ok_id = create_task(
370. 370 srv,
371. 371 serde_json::json!({
372. 372 "name": "ok-watch", "script": "echo watchdog-hello",
373. 373 "schedule": "0 0 29 2 *", "deliver_to": "log:w-ok"
374. 374 }),
375. 375 )
376. 376 .await;
377. 377 let silent_id = create_task(
378. 378 srv,
379. 379 serde_json::json!({
380. 380 "name": "silent-watch", "script": "true",
381. 381 "schedule": "0 0 29 2 *", "deliver_to": "log:w-silent"
382. 382 }),
383. 383 )
384. 384 .await;
385. 385 let fail_id = create_task(
386. 386 srv,
387. 387 serde_json::json!({
388. 388 "name": "fail-watch", "script": "echo broken >&2; exit 3",
389. 389 "schedule": "0 0 29 2 *"
390. 390 }),
391. 391 )
392. 392 .await;
393. 393
394. 394 // Success delivers trimmed stdout verbatim.
395. 395 assert_eq!(run_now(srv, &ok_id).await, 202);
396. 396 wait_summary(srv, &ok_id, 15, |s| s == "watchdog-hello").await;
397. 397 if !wait_until(10, || {
398. 398 delivery_lines(&srv.home)
399. 399 .iter()
400. 400 .any(|(t, x)| t == "log:w-ok" && x == "watchdog-hello")
401. 401 })
402. 402 .await
403. 403 {
404. 404 eprintln!(
405. 405 "[DBG-test] deliveries={:?} raw={:?} summary={:?}",
406. 406 delivery_lines(&srv.home),
407. 407 std::fs::read(srv.home.join("gateway").join("deliveries.jsonl")),
408. 408 task_field(srv, &ok_id, "last_summary").await
409. 409 );
410. 410 panic!("delivery never landed");
411. 411 }
412. 412
413. 413 // Empty stdout is a silent tick: recorded locally, never delivered.
414. 414 assert_eq!(run_now(srv, &silent_id).await, 202);
415. 415 wait_summary(srv, &silent_id, 15, |s| s == "(silent tick)").await;
416. 416 assert!(
417. 417 !delivery_lines(&srv.home)
418. 418 .iter()
419. 419 .any(|(t, _)| t == "log:w-silent"),
420. 420 "silent tick must not deliver"
421. 421 );
422. 422
423. 423 // Failure delivers an error alert EVEN without a configured target
424. 424 // (fallback log surface), carrying exit-code detail.
425. 425 assert_eq!(run_now(srv, &fail_id).await, 202);
426. 426 assert!(
427. 427 wait_until(15, || delivery_lines(&srv.home).iter().any(|(_, x)| {
428. 428 x.contains("watchdog 'fail-watch' alert") && x.contains("exit code: 3")
429. 429 }))
430. 430 .await,
431. 431 "failing watchdog must deliver a typed error alert"
432. 432 );
433. 433 wait_summary(srv, &fail_id, 10, |s| s.starts_with("script failed:")).await;
434. 434
435. 435 // Zero tokens by construction AND observed: no provider dispatch ever.
436. 436 assert_eq!(fx.srv.dispatches.load(Ordering::SeqCst), 0);
437. 437}
438. 438
439. 439#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
440. 440async fn missing_worker_delivers_typed_error_not_silence() {
441. 441 // Same fixture minus the real worker exe: the broker fails closed and
442. 442 // the failure must land as a DELIVERED alert, never silence.
443. 443 let dir = tempfile::tempdir().unwrap();
444. 444 let cwd = dir.path().join("ws");
445. 445 std::fs::create_dir_all(cwd.join(".vak")).unwrap();
446. 446 std::fs::write(cwd.join(".vak/config.toml"), "").unwrap();
447. 447 let home = dir.path().join("home");
448. 448 vak_config::paths::isolate_home_for_tests();
449. 449 let core = Core::new_with_trust(cwd, true).unwrap();
450. 450 core.set_sessions_home(home.clone());
451. 451 core.set_provider_instance(Arc::new(Counting {
452. 452 dispatches: Arc::new(AtomicUsize::new(0)),
453. 453 }));
454. 454 // Deliberately point the worker at something unusable.
455. 455 core.set_tool_worker_exe(dir.path().join("no-worker-here"));
456. 456 std::mem::forget(dir);
457. 457
458. 458 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
459. 459 let addr = listener.local_addr().unwrap();
460. 460 let app = vak_server::router(core);
461. 461 tokio::spawn(async move {
462. 462 axum::serve(listener, app).await.unwrap();
463. 463 });
464. 464 let base = format!("http://{addr}");
465. 465 let client = reqwest::Client::new();
466. 466
467. 467 client
468. 468 .post(format!("{base}/tasks"))
469. 469 .json(&serde_json::json!({
470. 470 "name": "brokerless", "script": "echo hi",
471. 471 "interval_secs": 3600, "deliver_to": "log:bw"
472. 472 }))
473. 473 .send()
474. 474 .await
475. 475 .unwrap();
476. 476 let list: serde_json::Value = client
477. 477 .get(format!("{base}/tasks"))
478. 478 .send()
479. 479 .await
480. 480 .unwrap()
481. 481 .json()
482. 482 .await
483. 483 .unwrap();
484. 484 let id = list["tasks"][0]["id"].as_str().unwrap().to_string();
485. 485 let res = client
486. 486 .post(format!("{base}/tasks/{id}/run-now"))
487. 487 .send()
488. 488 .await
489. 489 .unwrap();
490. 490 assert_eq!(res.status(), 202);
491. 491
492. 492 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
493. 493 loop {
494. 494 assert!(
495. 495 std::time::Instant::now() < deadline,
496. 496 "broker failure was silent; expected delivered alert"
497. 497 );
498. 498 if delivery_lines(&home)
499. 499 .iter()
500. 500 .any(|(_, x)| x.contains("watchdog 'brokerless' alert"))
501. 501 {
502. 502 break;
503. 503 }
504. 504 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
505. 505 }
506. 506}
507. 507
508. 508// ---- Model pin -------------------------------------------------------------------
509. 509
510. 510#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
511. 511async fn model_pin_receipts_show_pinned_model_only() {
512. 512 let fx = spawn_full("", None::<fn(&Path, &Path) -> serde_json::Value>).await;
513. 513 let srv = &fx.srv;
514. 514 let client = srv.client();
515. 515
516. 516 let res = client
517. 517 .post(format!("{}/tasks", srv.base))
518. 518 .json(&serde_json::json!({
519. 519 "name": "pinned-nightly",
520. 520 "prompt": "summarize the tree",
521. 521 "interval_secs": 3600,
522. 522 "model_pin": "counting/pinned-model-x"
523. 523 }))
524. 524 .send()
525. 525 .await
526. 526 .unwrap();
527. 527 assert_eq!(res.status(), 200);
528. 528 let list = srv.get_json("/tasks").await;
529. 529 let tid = list["tasks"][0]["id"].as_str().unwrap().to_string();
530. 530
531. 531 // A never-run interval task is due immediately, so the SCHEDULER fires
532. 532 // it on its first tick; wait for the child session id it records.
533. 533 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(25);
534. 534 let child = loop {
535. 535 assert!(
536. 536 std::time::Instant::now() < deadline,
537. 537 "task never recorded a child session"
538. 538 );
539. 539 if let Some(c) = task_field(srv, &tid, "last_session_id")
540. 540 .await
541. 541 .as_str()
542. 542 .map(String::from)
543. 543 {
544. 544 break c;
545. 545 }
546. 546 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
547. 547 };
548. 548 let receipts: serde_json::Value = {
549. 549 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
550. 550 loop {
551. 551 assert!(
552. 552 std::time::Instant::now() < deadline,
553. 553 "receipts never appeared for {child}"
554. 554 );
555. 555 match client
556. 556 .get(format!("{}/sessions/{child}/receipts", srv.base))
557. 557 .send()
558. 558 .await
559. 559 {
560. 560 Ok(res) if res.status() == 200 => {
561. 561 let body: serde_json::Value = res.json().await.unwrap();
562. 562 if body.as_array().is_some_and(|a| !a.is_empty()) {
563. 563 break body;
564. 564 }
565. 565 }
566. 566 _ => {}
567. 567 }
568. 568 tokio::time::sleep(std::time::Duration::from_millis(150)).await;
569. 569 }
570. 570 };
571. 571 for r in receipts.as_array().unwrap() {
572. 572 assert_eq!(
573. 573 r["model"], "pinned-model-x",
574. 574 "every receipt carries the pin"
575. 575 );
576. 576 assert_eq!(r["provider"], "counting");
577. 577 }
578. 578
579. 579 // And the run itself completed normally.
580. 580 wait_summary(srv, &tid, 15, |s| s == "done").await;
581. 581}
582. 582
583. 583// ---- Budget alerts ---------------------------------------------------------------
584. 584
585. 585#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
586. 586async fn budget_alert_fires_once_per_window_then_stops() {
587. 587 let fx = spawn_full(
588. 588 "[finops]\nmax_day_usd = 10.0\n",
589. 589 None::<fn(&Path, &Path) -> serde_json::Value>,
590. 590 )
591. 591 .await;
592. 592 let srv = &fx.srv;
593. 593
594. 594 // Day spend already past the 80% threshold of the $10 cap.
595. 595 let ledger = vak_core::finops::FinOpsLedger::new(&srv.home);
596. 596 ledger
597. 597 .append(&vak_core::finops::CostRow {
598. 598 ts: chrono::Utc::now(),
599. 599 model: "claude-sonnet".into(),
600. 600 provider: "anthropic".into(),
601. 601 input_tokens: 1000,
602. 602 output_tokens: 500,
603. 603 cache_read_input_tokens: None,
604. 604 usd: Some(9.0),
605. 605 source: "estimated".into(),
606. 606 session_id: "seed".into(),
607. 607 })
608. 608 .unwrap();
609. 609
610. 610 let alerts_path = srv.home.join("budget-alerts.jsonl");
611. 611
612. 612 let tid = create_task(
613. 613 srv,
614. 614 serde_json::json!({
615. 615 "name": "budget-probe", "script": "true",
616. 616 "schedule": "0 0 29 2 *", "deliver_to": "log:budget"
617. 617 }),
618. 618 )
619. 619 .await;
620. 620
621. 621 // First fire crosses the threshold: one audit row, one delivery.
622. 622 assert_eq!(run_now(srv, &tid).await, 202);
623. 623 assert!(
624. 624 wait_until(15, || std::fs::read_to_string(&alerts_path)
625. 625 .map(|c| c.lines().count())
626. 626 .unwrap_or(0)
627. 627 >= 1)
628. 628 .await,
629. 629 "no budget alert row recorded"
630. 630 );
631. 631 let rows = std::fs::read_to_string(&alerts_path).unwrap();
632. 632 assert_eq!(rows.lines().count(), 1, "{rows}");
633. 633 assert!(rows.contains("\"level\":\"eighty\""), "{rows}");
634. 634 assert!(
635. 635 wait_until(10, || delivery_lines(&srv.home).iter().any(|(t, x)| t
636. 636 == "log:budget"
637. 637 && x.contains("budget alert [eighty]")))
638. 638 .await
639. 639 );
640. 640
641. 641 // Second fire inside the same day window: no new row, no redelivery.
642. 642 assert_eq!(run_now(srv, &tid).await, 202);
643. 643 tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
644. 644 let rows = std::fs::read_to_string(&alerts_path).unwrap();
645. 645 assert_eq!(rows.lines().count(), 1, "same-level alert must not refire");
646. 646 let deliveries = delivery_lines(&srv.home)
647. 647 .iter()
648. 648 .filter(|(_, x)| x.contains("budget alert [eighty]"))
649. 649 .count();
650. 650 assert_eq!(deliveries, 1, "same-level alert must not redeliver");
651. 651}
652. 652
653. 653// ---- Task CRUD validation ---------------------------------------------------------
654. 654
655. 655#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
656. 656async fn task_api_validates_schedule_script_and_pin_fields() {
657. 657 let fx = spawn_full("", None::<fn(&Path, &Path) -> serde_json::Value>).await;
658. 658 let srv = &fx.srv;
659. 659 let client = srv.client();
660. 660
661. 661 // prompt XOR script.
662. 662 let res = client
663. 663 .post(format!("{}/tasks", srv.base))
664. 664 .json(&serde_json::json!({
665. 665 "name": "both", "prompt": "x", "script": "y", "interval_secs": 3600
666. 666 }))
667. 667 .send()
668. 668 .await
669. 669 .unwrap();
670. 670 assert_eq!(res.status(), 400);
671. 671 let body: serde_json::Value = res.json().await.unwrap();
672. 672 assert!(body["error"].as_str().unwrap().contains("exactly one"));
673. 673
674. 674 let res = client
675. 675 .post(format!("{}/tasks", srv.base))
676. 676 .json(&serde_json::json!({ "name": "neither", "interval_secs": 3600 }))
677. 677 .send()
678. 678 .await
679. 679 .unwrap();
680. 680 assert_eq!(res.status(), 400);
681. 681
682. 682 // Bad cron grammar is rejected with a typed message.
683. 683 let res = client
684. 684 .post(format!("{}/tasks", srv.base))
685. 685 .json(&serde_json::json!({
686. 686 "name": "badcron", "script": "true",
687. 687 "interval_secs": 3600, "schedule": "99 * * * *"
688. 688 }))
689. 689 .send()
690. 690 .await
691. 691 .unwrap();
692. 692 assert_eq!(res.status(), 400);
693. 693 let body: serde_json::Value = res.json().await.unwrap();
694. 694 assert!(body["error"].as_str().unwrap().contains("99"), "{body}");
695. 695
696. 696 // A valid cron+script watchdog is accepted; prompt stays optional.
697. 697 let res = client
698. 698 .post(format!("{}/tasks", srv.base))
699. 699 .json(&serde_json::json!({
700. 700 "name": "goodcron", "script": "true",
701. 701 "interval_secs": 3600, "schedule": "*/5 * * * *",
702. 702 "agent_id": "vak", "agent_revision": 1
703. 703 }))
704. 704 .send()
705. 705 .await
706. 706 .unwrap();
707. 707 assert_eq!(res.status(), 200);
708. 708 let list = srv.get_json("/tasks").await;
709. 709 let good = &list["tasks"][0];
710. 710 assert_eq!(good["schedule"], "*/5 * * * *");
711. 711 assert_eq!(good["agent_id"], "vak");
712. 712 assert_eq!(good["agent_revision"], 1);
713. 713
714. 714 // PATCH to an invalid schedule is rejected and leaves state untouched.
715. 715 let res = client
716. 716 .patch(format!(
717. 717 "{}/tasks/{}",
718. 718 srv.base,
719. 719 good["id"].as_str().unwrap()
720. 720 ))
721. 721 .json(&serde_json::json!({ "schedule": "* * * *" }))
722. 722 .send()
723. 723 .await
724. 724 .unwrap();
725. 725 assert_eq!(res.status(), 400);
726. 726 let list = srv.get_json("/tasks").await;
727. 727 assert_eq!(
728. 728 list["tasks"][0]["schedule"], "*/5 * * * *",
729. 729 "invalid patch ignored"
730. 730 );
731. 731
732. 732 // Clearing script then setting prompt keeps validation green.
733. 733 let res = client
734. 734 .patch(format!(
735. 735 "{}/tasks/{}",
736. 736 srv.base,
737. 737 good["id"].as_str().unwrap()
738. 738 ))
739. 739 .json(&serde_json::json!({ "script": null, "prompt": "now an agent task" }))
740. 740 .send()
741. 741 .await
742. 742 .unwrap();
743. 743 assert_eq!(res.status(), 200);
744. 744 let list = srv.get_json("/tasks").await;
745. 745 assert_eq!(list["tasks"][0]["prompt"], "now an agent task");
746. 746 assert!(list["tasks"][0]["script"].is_null());
747. 747
748. 748 // Agent ownership is patchable and clearing it also clears its revision.
749. 749 let res = client
750. 750 .patch(format!(
751. 751 "{}/tasks/{}",
752. 752 srv.base,
753. 753 good["id"].as_str().unwrap()
754. 754 ))
755. 755 .json(&serde_json::json!({ "agent_id": null }))
756. 756 .send()
757. 757 .await
758. 758 .unwrap();
759. 759 assert_eq!(res.status(), 200);
760. 760 let list = srv.get_json("/tasks").await;
761. 761 assert!(list["tasks"][0]["agent_id"].is_null());
762. 762 assert!(list["tasks"][0]["agent_revision"].is_null());
763. 763
764. 764 // Unknown task id on PATCH is a typed 404.
765. 765 let res = client
766. 766 .patch(format!("{}/tasks/nope", srv.base))
767. 767 .json(&serde_json::json!({ "name": "x" }))
768. 768 .send()
769. 769 .await
770. 770 .unwrap();
771. 771 assert_eq!(res.status(), 404);
772. 772}
773. 773