1. 1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2. 2
3. 3use std::collections::VecDeque;
4. 4use std::sync::{Arc, Mutex};
5. 5use std::time::Duration;
6. 6
7. 7use tokio_util::sync::CancellationToken;
8. 8
9. 9use vak_core::Core;
10. 10use vak_llm::stream;
11. 11use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
12. 12use vak_llm::{EventStream, LlmError, Provider};
13. 13
14. 14struct Scripted {
15. 15 responses: Mutex<VecDeque<AssistantMessage>>,
16. 16}
17. 17
18. 18#[async_trait::async_trait]
19. 19impl Provider for Scripted {
20. 20 fn name(&self) -> &str {
21. 21 "scripted"
22. 22 }
23. 23
24. 24 async fn stream(
25. 25 &self,
26. 26 _request: ChatRequest,
27. 27 _cancel: CancellationToken,
28. 28 ) -> Result<EventStream, LlmError> {
29. 29 let next = self.responses.lock().unwrap().pop_front();
30. 30 let (mut sink, rx) = stream::channel(64);
31. 31 match next {
32. 32 Some(m) => {
33. 33 sink.push(stream::StreamEvent::Start { partial: m.clone() });
34. 34 sink.close_message(m).await;
35. 35 }
36. 36 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
37. 37 }
38. 38 Ok(rx)
39. 39 }
40. 40}
41. 41
42. 42fn text(t: &str) -> AssistantMessage {
43. 43 AssistantMessage {
44. 44 content: vec![ContentBlock::text(t)],
45. 45 stop_reason: StopReason::EndTurn,
46. 46 usage: Usage {
47. 47 input_tokens: 7,
48. 48 output_tokens: 3,
49. 49 ..Default::default()
50. 50 },
51. 51 model: "test-model".into(),
52. 52 response_id: None,
53. 53 }
54. 54}
55. 55
56. 56fn tool_call(id: &str, name: &str, input: serde_json::Value) -> AssistantMessage {
57. 57 AssistantMessage {
58. 58 content: vec![ContentBlock::ToolUse {
59. 59 id: id.into(),
60. 60 name: name.into(),
61. 61 input,
62. 62 }],
63. 63 stop_reason: StopReason::ToolUse,
64. 64 usage: Usage::default(),
65. 65 model: "test-model".into(),
66. 66 response_id: None,
67. 67 }
68. 68}
69. 69
70. 70/// Pin `VAK_HOME` to an empty, process-stable tempdir so `spawn_server`'s
71. 71/// `Core::new` does not inherit the operator's real user-global config — which
72. 72/// here is `permission_mode = fullaccess` with `approval_mode` != `ask` and
73. 73/// would, via `refresh_persisted_preferences`, clobber the test's runtime
74. 74/// `set_permission_mode` pin and auto-approve the `bash` -> `Ask` arm so no
75. 75/// `ApprovalRequested` is ever published. With the global layer absent,
76. 76/// `Config::default()` applies (`approval_mode = Ask`, vak-config lib.rs:891)
77. 77/// and the per-test `mode` pin plus `WorkspaceWrite` default restore the
78. 78/// intended human-in-the-loop path. Installed once via `Once`; harmless to the
79. 79/// sibling tests because the temp never holds a global config.
80. 80fn isolate_global_config() {
81. 81 vak_config::paths::isolate_home_for_tests();
82. 82}
83. 83
84. 84async fn spawn_server(
85. 85 provider: Arc<dyn Provider>,
86. 86 mode: vak_config::PermissionMode,
87. 87) -> (String, tokio::task::JoinHandle<()>) {
88. 88 isolate_global_config();
89. 89 let dir = tempfile::tempdir().unwrap();
90. 90 vak_config::paths::isolate_home_for_tests();
91. 91 let core = Core::new(dir.path().to_path_buf()).unwrap();
92. 92 core.set_sessions_home(dir.path().join("home"));
93. 93 core.set_permission_mode(mode);
94. 94 core.set_tool_worker_exe(std::path::PathBuf::from(env!(
95. 95 "CARGO_BIN_EXE_vak-tool-worker"
96. 96 )));
97. 97 core.set_provider_instance(provider);
98. 98 // keep tempdir alive for the process lifetime of the test
99. 99 std::mem::forget(dir);
100. 100
101. 101 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
102. 102 let addr = listener.local_addr().unwrap();
103. 103 let app = vak_server::router(core);
104. 104 let handle = tokio::spawn(async move {
105. 105 axum::serve(listener, app).await.unwrap();
106. 106 });
107. 107 (format!("http://{addr}"), handle)
108. 108}
109. 109
110. 110#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
111. 111async fn http_lifecycle_run_events_transcript() {
112. 112 let provider = Arc::new(Scripted {
113. 113 responses: Mutex::new(VecDeque::from(vec![
114. 114 tool_call("t1", "bash", serde_json::json!({"command": "echo served"})),
115. 115 text("all served"),
116. 116 ])),
117. 117 });
118. 118 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::FullAccess).await;
119. 119 let client = reqwest::Client::new();
120. 120
121. 121 // health
122. 122 let health = client.get(format!("{base}/health")).send().await.unwrap();
123. 123 assert_eq!(health.status(), 200);
124. 124 let body_health: serde_json::Value = health.json().await.unwrap();
125. 125 assert_eq!(body_health["status"], "ok");
126. 126
127. 127 // create session
128. 128 let res = client
129. 129 .post(format!("{base}/sessions"))
130. 130 .send()
131. 131 .await
132. 132 .unwrap();
133. 133 assert_eq!(res.status(), 200);
134. 134 let session_id: String = res.json::<serde_json::Value>().await.unwrap()["session_id"]
135. 135 .as_str()
136. 136 .unwrap()
137. 137 .to_string();
138. 138
139. 139 // subscribe to SSE BEFORE running so no events are missed
140. 140 let sse_url = format!("{base}/sessions/{session_id}/events");
141. 141 let (opened_tx, opened_rx) = tokio::sync::oneshot::channel::<()>();
142. 142 let mut opened_tx = Some(opened_tx);
143. 143 let sse_task = tokio::spawn(async move {
144. 144 let res = reqwest::get(&sse_url).await.unwrap();
145. 145 assert_eq!(res.status(), 200);
146. 146 let mut collected = Vec::new();
147. 147 let mut opened_done = false;
148. 148 use futures::StreamExt;
149. 149 let mut stream = res.bytes_stream();
150. 150 let deadline = std::time::Instant::now() + Duration::from_secs(10);
151. 151 while std::time::Instant::now() < deadline {
152. 152 if let Some(Ok(chunk)) = stream.next().await {
153. 153 let text = String::from_utf8_lossy(&chunk).into_owned();
154. 154 for line in text.lines() {
155. 155 if let Some(data) = line.strip_prefix("data:") {
156. 156 collected.push(data.trim().to_string());
157. 157 }
158. 158 }
159. 159 }
160. 160 if !opened_done && collected.iter().any(|c| c.contains("StreamOpened")) {
161. 161 opened_done = true;
162. 162 if let Some(t) = opened_tx.take() {
163. 163 let _ = t.send(());
164. 164 }
165. 165 #[allow(unused_assignments)]
166. 166 {
167. 167 // oneshot send consumes; guard against loop re-entry
168. 168 }
169. 169 }
170. 170 if collected
171. 171 .iter()
172. 172 .any(|c| c.contains("__done__") || c.contains("RunFinished"))
173. 173 {
174. 174 break;
175. 175 }
176. 176 }
177. 177 collected
178. 178 });
179. 179
180. 180 // fire the run only once the event stream is confirmed open
181. 181 let _ = tokio::time::timeout(Duration::from_secs(5), opened_rx).await;
182. 182
183. 183 // run a prompt
184. 184 let res = client
185. 185 .post(format!("{base}/sessions/{session_id}/run"))
186. 186 .json(&serde_json::json!({"prompt": "serve it"}))
187. 187 .send()
188. 188 .await
189. 189 .unwrap();
190. 190 assert_eq!(res.status(), 202);
191. 191
192. 192 let events = tokio::time::timeout(Duration::from_secs(10), sse_task)
193. 193 .await
194. 194 .expect("sse timed out")
195. 195 .unwrap();
196. 196
197. 197 // The transcript must show the full loop.
198. 198 let transcript: serde_json::Value = client
199. 199 .get(format!("{base}/sessions/{session_id}/transcript"))
200. 200 .send()
201. 201 .await
202. 202 .unwrap()
203. 203 .json()
204. 204 .await
205. 205 .unwrap();
206. 206 assert_eq!(transcript["count"].as_u64(), Some(4)); // user, assistant(tool), user(result), assistant(final)
207. 207 assert!(
208. 208 serde_json::to_string(&transcript)
209. 209 .unwrap()
210. 210 .contains("served"),
211. 211 "transcript must contain the run"
212. 212 );
213. 213
214. 214 // Events must include streamed text and the terminal marker.
215. 215 let joined = events.join("\n");
216. 216 assert!(joined.contains("TurnStart"), "events: {joined}");
217. 217 assert!(
218. 218 joined.contains("RunFinished") || joined.contains("__done__"),
219. 219 "terminal event missing: {joined}"
220. 220 );
221. 221
222. 222 // Phase R forensics: receipts must be served over HTTP with leg
223. 223 // attribution (the desktop drill-down panel reads exactly this).
224. 224 let receipts: serde_json::Value = client
225. 225 .get(format!("{base}/sessions/{session_id}/receipts"))
226. 226 .send()
227. 227 .await
228. 228 .unwrap()
229. 229 .json()
230. 230 .await
231. 231 .unwrap();
232. 232 let list = receipts.as_array().expect("receipts array");
233. 233 assert!(!list.is_empty(), "a settled run must leave receipts");
234. 234 let r = &list[list.len() - 1];
235. 235 assert_eq!(r["purpose"], "execute");
236. 236 assert_eq!(r["provider"], "scripted");
237. 237 // The receipt stamps the CONTRACT leg (what was dispatched), which is
238. 238 // the core's effective model -- not the mock's self-reported id.
239. 239 let expected_model = body_health["model"].as_str().unwrap().to_string();
240. 240 assert_eq!(r["model"], expected_model);
241. 241 assert_eq!(r["winning_attempt"], 0);
242. 242 let attempts = r["attempts"].as_array().unwrap();
243. 243 assert_eq!(attempts.len(), 1);
244. 244 assert_eq!(attempts[0]["settlement"], "ok");
245. 245 assert_eq!(attempts[0]["reason"], "initial");
246. 246}
247. 247
248. 248#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
249. 249async fn operations_center_is_a_real_evidence_projection() {
250. 250 let provider = Arc::new(Scripted {
251. 251 responses: Mutex::new(VecDeque::new()),
252. 252 });
253. 253 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::ReadOnly).await;
254. 254 let body: serde_json::Value = reqwest::get(format!("{base}/ops/center"))
255. 255 .await
256. 256 .unwrap()
257. 257 .error_for_status()
258. 258 .unwrap()
259. 259 .json()
260. 260 .await
261. 261 .unwrap();
262. 262 assert!(body["server"]["pid"].as_u64().is_some());
263. 263 assert!(body["server"]["uptime_secs"].as_u64().is_some());
264. 264 assert!(body["health"]["checks"].is_array());
265. 265 assert!(body["pool"]["entries"].is_array());
266. 266 assert!(body["runs"].is_array());
267. 267 assert!(body["outbox"]["records"].is_array());
268. 268 assert!(body["incidents"].is_array());
269. 269}
270. 270
271. 271#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
272. 272async fn secured_operations_center_uses_the_bound_port() {
273. 273 let dir = tempfile::tempdir().unwrap();
274. 274 vak_config::paths::isolate_home_for_tests();
275. 275 let core = Core::new(dir.path().to_path_buf()).unwrap();
276. 276 core.set_sessions_home(dir.path().join("home"));
277. 277 std::mem::forget(dir);
278. 278
279. 279 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
280. 280 let addr = listener.local_addr().unwrap();
281. 281 let (app, token) = vak_server::secured_router_with_port(core, false, addr.port());
282. 282 let handle = tokio::spawn(async move {
283. 283 axum::serve(listener, app).await.unwrap();
284. 284 });
285. 285 let client = reqwest::Client::new();
286. 286
287. 287 let unauthenticated = client
288. 288 .get(format!("http://{addr}/ops/center"))
289. 289 .send()
290. 290 .await
291. 291 .unwrap();
292. 292 assert_eq!(unauthenticated.status(), reqwest::StatusCode::UNAUTHORIZED);
293. 293
294. 294 let body: serde_json::Value = client
295. 295 .get(format!("http://{addr}/ops/center"))
296. 296 .bearer_auth(&token)
297. 297 .send()
298. 298 .await
299. 299 .unwrap()
300. 300 .error_for_status()
301. 301 .unwrap()
302. 302 .json()
303. 303 .await
304. 304 .unwrap();
305. 305 assert_eq!(body["ops_port"], addr.port());
306. 306 assert_eq!(body["server"]["pid"], std::process::id());
307. 307 assert!(body["server"]["uptime_secs"].as_u64().is_some());
308. 308 assert_eq!(body["services"]["gateway_healthy"], true);
309. 309
310. 310 let outbox: serde_json::Value = client
311. 311 .get(format!("http://{addr}/ops/outbox"))
312. 312 .bearer_auth(&token)
313. 313 .send()
314. 314 .await
315. 315 .unwrap()
316. 316 .error_for_status()
317. 317 .unwrap()
318. 318 .json()
319. 319 .await
320. 320 .unwrap();
321. 321 assert!(outbox["records"].is_array());
322. 322
323. 323 let replay_missing = client
324. 324 .post(format!("http://{addr}/ops/outbox/not-a-real-job/replay"))
325. 325 .bearer_auth(&token)
326. 326 .send()
327. 327 .await
328. 328 .unwrap();
329. 329 assert_eq!(replay_missing.status(), reqwest::StatusCode::CONFLICT);
330. 330
331. 331 handle.abort();
332. 332}
333. 333
334. 334#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
335. 335async fn approval_flow_resolves_over_http() {
336. 336 // First turn requests bash (workspace-write => ask); we approve over HTTP.
337. 337 let provider = Arc::new(Scripted {
338. 338 responses: Mutex::new(VecDeque::from(vec![
339. 339 tool_call(
340. 340 "t1",
341. 341 "bash",
342. 342 serde_json::json!({"command": "echo approved-run"}),
343. 343 ),
344. 344 text("done after approval"),
345. 345 ])),
346. 346 });
347. 347 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::WorkspaceWrite).await;
348. 348 let client = reqwest::Client::new();
349. 349
350. 350 let session_id: String = client
351. 351 .post(format!("{base}/sessions"))
352. 352 .send()
353. 353 .await
354. 354 .unwrap()
355. 355 .json::<serde_json::Value>()
356. 356 .await
357. 357 .unwrap()["session_id"]
358. 358 .as_str()
359. 359 .unwrap()
360. 360 .to_string();
361. 361
362. 362 // SSE collector answers approvals inline so the run can proceed.
363. 363 let sse_session = session_id.clone();
364. 364 let sse_url = format!("{base}/sessions/{sse_session}/events");
365. 365 let answer_url_base = base.clone();
366. 366 let (opened_tx, opened_rx) = tokio::sync::oneshot::channel::<()>();
367. 367 let mut opened_tx = Some(opened_tx);
368. 368 let sse_task = tokio::spawn(async move {
369. 369 let res = reqwest::get(&sse_url).await.unwrap();
370. 370 use futures::StreamExt;
371. 371 let mut stream = res.bytes_stream();
372. 372 let mut approval_answered = false;
373. 373 let mut saw_finish = false;
374. 374 let mut opened_done = false;
375. 375 let deadline = std::time::Instant::now() + Duration::from_secs(10);
376. 376 while std::time::Instant::now() < deadline {
377. 377 if let Some(Ok(chunk)) = stream.next().await {
378. 378 let text = String::from_utf8_lossy(&chunk).into_owned();
379. 379 for line in text.lines() {
380. 380 if let Some(data) = line.strip_prefix("data:")
381. 381 && let Ok(v) = serde_json::from_str::<serde_json::Value>(data.trim())
382. 382 {
383. 383 if !opened_done && v["StreamOpened"].is_object() {
384. 384 opened_done = true;
385. 385 if let Some(t) = opened_tx.take() {
386. 386 let _ = t.send(());
387. 387 }
388. 388 }
389. 389 if !approval_answered && v["ApprovalRequested"]["id"].is_string() {
390. 390 let rid = v["ApprovalRequested"]["id"].as_str().unwrap().to_string();
391. 391 let _ = reqwest::Client::new()
392. 392 .post(format!(
393. 393 "{answer_url_base}/sessions/{sse_session}/approvals/{rid}"
394. 394 ))
395. 395 .json(&serde_json::json!({"approve": true}))
396. 396 .send()
397. 397 .await;
398. 398 approval_answered = true;
399. 399 }
400. 400 if v["RunFinished"].is_object() {
401. 401 saw_finish = true;
402. 402 }
403. 403 }
404. 404 }
405. 405 }
406. 406 if saw_finish {
407. 407 break;
408. 408 }
409. 409 }
410. 410 (approval_answered, saw_finish)
411. 411 });
412. 412
413. 413 // fire the run only once the event stream is confirmed open
414. 414 let _ = tokio::time::timeout(Duration::from_secs(5), opened_rx).await;
415. 415 client
416. 416 .post(format!("{base}/sessions/{session_id}/run"))
417. 417 .json(&serde_json::json!({"prompt": "needs approval"}))
418. 418 .send()
419. 419 .await
420. 420 .unwrap();
421. 421
422. 422 let (answered, saw_finish) = tokio::time::timeout(Duration::from_secs(10), sse_task)
423. 423 .await
424. 424 .expect("sse timed out")
425. 425 .unwrap();
426. 426 assert!(answered, "an approval request must have been published");
427. 427 assert!(saw_finish, "run must finish after inline approval");
428. 428}
429. 429
430. 430#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
431. 431async fn mode_change_revokes_run_waiting_for_approval() {
432. 432 let provider = Arc::new(Scripted {
433. 433 responses: Mutex::new(VecDeque::from(vec![tool_call(
434. 434 "t1",
435. 435 "bash",
436. 436 serde_json::json!({"command": "echo must-not-run"}),
437. 437 )])),
438. 438 });
439. 439 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::WorkspaceWrite).await;
440. 440 let client = reqwest::Client::new();
441. 441 let session_id: String = client
442. 442 .post(format!("{base}/sessions"))
443. 443 .send()
444. 444 .await
445. 445 .unwrap()
446. 446 .json::<serde_json::Value>()
447. 447 .await
448. 448 .unwrap()["session_id"]
449. 449 .as_str()
450. 450 .unwrap()
451. 451 .to_string();
452. 452
453. 453 let _events = client
454. 454 .get(format!("{base}/sessions/{session_id}/events"))
455. 455 .send()
456. 456 .await
457. 457 .unwrap();
458. 458 let run = client
459. 459 .post(format!("{base}/sessions/{session_id}/run"))
460. 460 .json(&serde_json::json!({"prompt": "request approval"}))
461. 461 .send()
462. 462 .await
463. 463 .unwrap();
464. 464 assert_eq!(run.status(), 202);
465. 465 tokio::time::sleep(Duration::from_millis(250)).await;
466. 466
467. 467 let switched = client
468. 468 .post(format!("{base}/config/mode"))
469. 469 .json(&serde_json::json!({"mode": "read-only"}))
470. 470 .send()
471. 471 .await
472. 472 .unwrap();
473. 473 assert_eq!(switched.status(), 200);
474. 474
475. 475 let deadline = std::time::Instant::now() + Duration::from_secs(5);
476. 476 loop {
477. 477 assert!(std::time::Instant::now() < deadline, "run was not revoked");
478. 478 let transcript: serde_json::Value = client
479. 479 .get(format!("{base}/sessions/{session_id}/transcript"))
480. 480 .send()
481. 481 .await
482. 482 .unwrap()
483. 483 .json()
484. 484 .await
485. 485 .unwrap();
486. 486 if transcript.get("error").is_none() {
487. 487 break;
488. 488 }
489. 489 tokio::time::sleep(Duration::from_millis(50)).await;
490. 490 }
491. 491}
492. 492
493. 493#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
494. 494async fn config_endpoint_exposes_route_policy() {
495. 495 // Phase R: the desktop Settings panel reads routing policy from
496. 496 // /config; the section must exist with resolved defaults.
497. 497 struct Empty;
498. 498 #[async_trait::async_trait]
499. 499 impl Provider for Empty {
500. 500 fn name(&self) -> &str {
501. 501 "empty"
502. 502 }
503. 503 async fn stream(
504. 504 &self,
505. 505 _r: ChatRequest,
506. 506 _c: CancellationToken,
507. 507 ) -> Result<EventStream, LlmError> {
508. 508 Err(LlmError::Network("unused".into()))
509. 509 }
510. 510 }
511. 511 let (base, _server) =
512. 512 spawn_server(Arc::new(Empty), vak_config::PermissionMode::FullAccess).await;
513. 513 let client = reqwest::Client::new();
514. 514 let cfg: serde_json::Value = client
515. 515 .get(format!("{base}/config"))
516. 516 .send()
517. 517 .await
518. 518 .unwrap()
519. 519 .json()
520. 520 .await
521. 521 .unwrap();
522. 522 let route = cfg
523. 523 .get("route")
524. 524 .expect("/config must carry the route section");
525. 525 assert_eq!(route["objective"], "auto");
526. 526 assert_eq!(route["max_fallbacks"], 4);
527. 527 assert!(route["fallback_models"].is_array());
528. 528 assert!(route["quality_hints"].is_array());
529. 529}
530. 530
531. 531#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
532. 532async fn cancel_endpoint_stops_a_running_session() {
533. 533 // A provider that hangs until cancelled — mirrors a stalled stream.
534. 534 struct Hung;
535. 535 #[async_trait::async_trait]
536. 536 impl Provider for Hung {
537. 537 fn name(&self) -> &str {
538. 538 "hung"
539. 539 }
540. 540 async fn stream(
541. 541 &self,
542. 542 _r: ChatRequest,
543. 543 cancel: CancellationToken,
544. 544 ) -> Result<EventStream, LlmError> {
545. 545 let (mut sink, rx) = stream::channel(8);
546. 546 sink.push(stream::StreamEvent::Start {
547. 547 partial: AssistantMessage::empty("m"),
548. 548 });
549. 549 tokio::spawn(async move {
550. 550 let _keep = sink;
551. 551 tokio::select! {
552. 552 _ = cancel.cancelled() => {}
553. 553 _ = std::future::pending::<()>() => {}
554. 554 }
555. 555 });
556. 556 Ok(rx)
557. 557 }
558. 558 }
559. 559
560. 560 let (base, _server) =
561. 561 spawn_server(Arc::new(Hung), vak_config::PermissionMode::FullAccess).await;
562. 562 let client = reqwest::Client::new();
563. 563
564. 564 let session_id: String = client
565. 565 .post(format!("{base}/sessions"))
566. 566 .send()
567. 567 .await
568. 568 .unwrap()
569. 569 .json::<serde_json::Value>()
570. 570 .await
571. 571 .unwrap()["session_id"]
572. 572 .as_str()
573. 573 .unwrap()
574. 574 .to_string();
575. 575
576. 576 // Wait for the stream-open marker, then start the run.
577. 577 let sse_url = format!("{base}/sessions/{session_id}/events");
578. 578 let opened = reqwest::get(&sse_url).await.unwrap();
579. 579 use futures::StreamExt;
580. 580 let mut events = opened.bytes_stream();
581. 581 let mut saw_cancelled = false;
582. 582
583. 583 let reader = tokio::spawn(async move {
584. 584 let deadline = std::time::Instant::now() + Duration::from_secs(8);
585. 585 while std::time::Instant::now() < deadline {
586. 586 if let Some(Ok(chunk)) = events.next().await {
587. 587 let text = String::from_utf8_lossy(&chunk);
588. 588 // `cancel_run` no longer synthesizes its own `RunFinished`
589. 589 // (it raced the real one); the terminal event now comes
590. 590 // from the run itself unwinding as `TurnOutcome::Aborted`,
591. 591 // which `run_prompt`/`http_settle` summarize as the internal
592. 592 // sentinel "aborted". The wire's `RunFinished` carries a
593. 593 // typed `outcome` and a human `message`, never that raw
594. 594 // internal summary string (`vak-server/src/
595. 595 // client_events.rs`) -- a cancelled run's outcome is
596. 596 // `"Stopped"`.
597. 597 if text.contains("RunFinished") && text.contains("\"Stopped\"") {
598. 598 saw_cancelled = true;
599. 599 break;
600. 600 }
601. 601 }
602. 602 }
603. 603 saw_cancelled
604. 604 });
605. 605
606. 606 tokio::time::sleep(Duration::from_millis(150)).await;
607. 607 client
608. 608 .post(format!("{base}/sessions/{session_id}/run"))
609. 609 .json(&serde_json::json!({"prompt": "hang forever"}))
610. 610 .send()
611. 611 .await
612. 612 .unwrap();
613. 613
614. 614 tokio::time::sleep(Duration::from_millis(300)).await;
615. 615 let res = client
616. 616 .post(format!("{base}/sessions/{session_id}/cancel"))
617. 617 .send()
618. 618 .await
619. 619 .unwrap();
620. 620 assert_eq!(res.status(), 202);
621. 621
622. 622 let saw = tokio::time::timeout(Duration::from_secs(9), reader)
623. 623 .await
624. 624 .expect("reader timed out")
625. 625 .unwrap();
626. 626 assert!(saw, "RunFinished(aborted) must be observed after /cancel");
627. 627}
628. 628
629. 629#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
630. 630async fn worker_endpoints_scope_and_wire() {
631. 631 let (base, _server) = spawn_server(
632. 632 Arc::new(Scripted {
633. 633 responses: Mutex::new(VecDeque::from(vec![text("no children")])),
634. 634 }),
635. 635 vak_config::PermissionMode::WorkspaceWrite,
636. 636 )
637. 637 .await;
638. 638
639. 639 // Mint a session.
640. 640 let res = reqwest::Client::new()
641. 641 .post(format!("{base}/sessions"))
642. 642 .send()
643. 643 .await
644. 644 .unwrap();
645. 645 assert_eq!(res.status(), 200);
646. 646 let sid: serde_json::Value = res.json().await.unwrap();
647. 647 let sid = sid["session_id"].as_str().unwrap().to_string();
648. 648
649. 649 // No live children.
650. 650 let res = reqwest::Client::new()
651. 651 .get(format!("{base}/sessions/{sid}/workers"))
652. 652 .send()
653. 653 .await
654. 654 .unwrap();
655. 655 assert_eq!(res.status(), 200);
656. 656 let body: serde_json::Value = res.json().await.unwrap();
657. 657 assert_eq!(body["workers"], serde_json::json!([]));
658. 658
659. 659 // Steering/stopping a child this session does not own is 404 — never a
660. 660 // cross-session capability leak, and never a silent no-op.
661. 661 let client = reqwest::Client::new();
662. 662 for path in [
663. 663 format!("/sessions/{sid}/workers/child-nope/steer"),
664. 664 format!("/sessions/{sid}/workers/child-nope/stop"),
665. 665 ] {
666. 666 let res = client
667. 667 .post(format!("{base}{path}"))
668. 668 .json(&serde_json::json!({"text": "hi"}))
669. 669 .send()
670. 670 .await
671. 671 .unwrap();
672. 672 assert_eq!(res.status(), 404, "{path}");
673. 673 }
674. 674}
675. 675
676. 676/// Responds after a fixed delay, so a caller has a window to observe the
677. 677/// session as busy before the leg settles.
678. 678struct DelayedThenScripted {
679. 679 responses: Mutex<VecDeque<AssistantMessage>>,
680. 680 delay: Duration,
681. 681}
682. 682
683. 683#[async_trait::async_trait]
684. 684impl Provider for DelayedThenScripted {
685. 685 fn name(&self) -> &str {
686. 686 "delayed"
687. 687 }
688. 688
689. 689 async fn stream(
690. 690 &self,
691. 691 _request: ChatRequest,
692. 692 _cancel: CancellationToken,
693. 693 ) -> Result<EventStream, LlmError> {
694. 694 tokio::time::sleep(self.delay).await;
695. 695 let next = self.responses.lock().unwrap().pop_front();
696. 696 let (mut sink, rx) = stream::channel(64);
697. 697 match next {
698. 698 Some(m) => {
699. 699 sink.push(stream::StreamEvent::Start { partial: m.clone() });
700. 700 sink.close_message(m).await;
701. 701 }
702. 702 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
703. 703 }
704. 704 Ok(rx)
705. 705 }
706. 706}
707. 707
708. 708async fn poll_transcript_contains(
709. 709 client: &reqwest::Client,
710. 710 base: &str,
711. 711 sid: &str,
712. 712 needle: &str,
713. 713 deadline: std::time::Instant,
714. 714) -> String {
715. 715 let mut raw = String::new();
716. 716 while std::time::Instant::now() < deadline {
717. 717 if let Ok(res) = client
718. 718 .get(format!("{base}/sessions/{sid}/transcript"))
719. 719 .send()
720. 720 .await
721. 721 {
722. 722 let t: serde_json::Value = res.json().await.unwrap_or_default();
723. 723 raw = serde_json::to_string(&t).unwrap_or_default();
724. 724 if raw.contains(needle) {
725. 725 return raw;
726. 726 }
727. 727 }
728. 728 tokio::time::sleep(Duration::from_millis(50)).await;
729. 729 }
730. 730 raw
731. 731}
732. 732
733. 733/// Finding 1: `/run` on a busy session used to return a bare 202 with the
734. 734/// prompt silently dropped (invariant 30 / docs/design/64, "Request
735. 735/// durability and delivery" — busy input is queued durably, never
736. 736/// discarded). It must now be queued and actually run as the chain's next
737. 737/// leg once the first run settles.
738. 738#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
739. 739async fn busy_run_is_queued_and_runs_as_a_continuation_leg() {
740. 740 let provider = Arc::new(DelayedThenScripted {
741. 741 responses: Mutex::new(VecDeque::from(vec![
742. 742 text("first done"),
743. 743 text("second done"),
744. 744 ])),
745. 745 delay: Duration::from_millis(400),
746. 746 });
747. 747 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::FullAccess).await;
748. 748 let client = reqwest::Client::new();
749. 749 let session_id: String = client
750. 750 .post(format!("{base}/sessions"))
751. 751 .send()
752. 752 .await
753. 753 .unwrap()
754. 754 .json::<serde_json::Value>()
755. 755 .await
756. 756 .unwrap()["session_id"]
757. 757 .as_str()
758. 758 .unwrap()
759. 759 .to_string();
760. 760
761. 761 let first = client
762. 762 .post(format!("{base}/sessions/{session_id}/run"))
763. 763 .json(&serde_json::json!({"prompt": "first", "request_id": "run-first"}))
764. 764 .send()
765. 765 .await
766. 766 .unwrap();
767. 767 assert_eq!(first.status(), 202);
768. 768 let first_body: serde_json::Value = first.json().await.unwrap();
769. 769 assert_eq!(first_body["state"], "started");
770. 770
771. 771 // The provider is still asleep, so this lands while busy.
772. 772 let second = client
773. 773 .post(format!("{base}/sessions/{session_id}/run"))
774. 774 .json(&serde_json::json!({"prompt": "second", "request_id": "run-second"}))
775. 775 .send()
776. 776 .await
777. 777 .unwrap();
778. 778 assert_eq!(second.status(), 202);
779. 779 let second_body: serde_json::Value = second.json().await.unwrap();
780. 780 assert_eq!(
781. 781 second_body["state"], "queued",
782. 782 "a busy /run must be queued, not silently dropped: {second_body}"
783. 783 );
784. 784 assert_eq!(second_body["request_id"], "run-second");
785. 785
786. 786 let deadline = std::time::Instant::now() + Duration::from_secs(10);
787. 787 let raw = poll_transcript_contains(&client, &base, &session_id, "second done", deadline).await;
788. 788 assert!(raw.contains("first done"), "first leg missing: {raw}");
789. 789 assert!(
790. 790 raw.contains("second"),
791. 791 "queued prompt must reach the model as the next leg: {raw}"
792. 792 );
793. 793 assert!(
794. 794 raw.contains("second done"),
795. 795 "queued run must actually execute as a continuation leg: {raw}"
796. 796 );
797. 797}
798. 798
799. 799/// Same fix, the `/steering` admission path: a steer that arrives while
800. 800/// busy must be queued and drained into a continuation leg — never
801. 801/// acknowledged and left in the queue with nothing left to drain it
802. 802/// (finding 1b).
803. 803#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
804. 804async fn busy_steering_is_queued_and_runs_as_a_continuation_leg() {
805. 805 let provider = Arc::new(DelayedThenScripted {
806. 806 responses: Mutex::new(VecDeque::from(vec![
807. 807 text("first done"),
808. 808 text("steered done"),
809. 809 ])),
810. 810 delay: Duration::from_millis(400),
811. 811 });
812. 812 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::FullAccess).await;
813. 813 let client = reqwest::Client::new();
814. 814 let session_id: String = client
815. 815 .post(format!("{base}/sessions"))
816. 816 .send()
817. 817 .await
818. 818 .unwrap()
819. 819 .json::<serde_json::Value>()
820. 820 .await
821. 821 .unwrap()["session_id"]
822. 822 .as_str()
823. 823 .unwrap()
824. 824 .to_string();
825. 825
826. 826 let run = client
827. 827 .post(format!("{base}/sessions/{session_id}/run"))
828. 828 .json(&serde_json::json!({"prompt": "first"}))
829. 829 .send()
830. 830 .await
831. 831 .unwrap();
832. 832 assert_eq!(run.status(), 202);
833. 833
834. 834 let steer = client
835. 835 .post(format!("{base}/sessions/{session_id}/steering"))
836. 836 .json(&serde_json::json!({"text": "please continue with this"}))
837. 837 .send()
838. 838 .await
839. 839 .unwrap();
840. 840 assert_eq!(steer.status(), 202);
841. 841 let steer_body: serde_json::Value = steer.json().await.unwrap();
842. 842 assert_eq!(steer_body["state"], "steering_queued");
843. 843
844. 844 let deadline = std::time::Instant::now() + Duration::from_secs(10);
845. 845 let raw = poll_transcript_contains(&client, &base, &session_id, "steered done", deadline).await;
846. 846 assert!(raw.contains("first done"), "first leg missing: {raw}");
847. 847 assert!(
848. 848 raw.contains("please continue with this"),
849. 849 "queued steering text must reach the model: {raw}"
850. 850 );
851. 851 assert!(
852. 852 raw.contains("steered done"),
853. 853 "queued steering must actually execute as a continuation leg: {raw}"
854. 854 );
855. 855}
856. 856
857. 857/// Hangs on its first call (until cancelled), then answers normally on any
858. 858/// later call — models a run that gets stopped and immediately resent.
859. 859struct HungOnceThenScripted {
860. 860 hung_once: std::sync::atomic::AtomicBool,
861. 861 responses: Mutex<VecDeque<AssistantMessage>>,
862. 862}
863. 863
864. 864#[async_trait::async_trait]
865. 865impl Provider for HungOnceThenScripted {
866. 866 fn name(&self) -> &str {
867. 867 "hung-once"
868. 868 }
869. 869
870. 870 async fn stream(
871. 871 &self,
872. 872 _request: ChatRequest,
873. 873 cancel: CancellationToken,
874. 874 ) -> Result<EventStream, LlmError> {
875. 875 if !self
876. 876 .hung_once
877. 877 .swap(true, std::sync::atomic::Ordering::SeqCst)
878. 878 {
879. 879 let (mut sink, rx) = stream::channel(8);
880. 880 sink.push(stream::StreamEvent::Start {
881. 881 partial: AssistantMessage::empty("m"),
882. 882 });
883. 883 tokio::spawn(async move {
884. 884 let _keep = sink;
885. 885 tokio::select! {
886. 886 _ = cancel.cancelled() => {}
887. 887 _ = std::future::pending::<()>() => {}
888. 888 }
889. 889 });
890. 890 return Ok(rx);
891. 891 }
892. 892 let next = self.responses.lock().unwrap().pop_front();
893. 893 let (mut sink, rx) = stream::channel(64);
894. 894 match next {
895. 895 Some(m) => {
896. 896 sink.push(stream::StreamEvent::Start { partial: m.clone() });
897. 897 sink.close_message(m).await;
898. 898 }
899. 899 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
900. 900 }
901. 901 Ok(rx)
902. 902 }
903. 903}
904. 904
905. 905/// Finding 1c: `cancel_run` no longer synthesizes its own `RunFinished`, and
906. 906/// input arriving after the stop but before the run unwinds is queued and
907. 907/// runs as the next leg of the chain — a stop must never eat the resend
908. 908/// that follows it.
909. 909#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
910. 910async fn stop_then_resend_runs_the_resend() {
911. 911 let provider = Arc::new(HungOnceThenScripted {
912. 912 hung_once: std::sync::atomic::AtomicBool::new(false),
913. 913 responses: Mutex::new(VecDeque::from(vec![text("resend done")])),
914. 914 });
915. 915 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::FullAccess).await;
916. 916 let client = reqwest::Client::new();
917. 917 let session_id: String = client
918. 918 .post(format!("{base}/sessions"))
919. 919 .send()
920. 920 .await
921. 921 .unwrap()
922. 922 .json::<serde_json::Value>()
923. 923 .await
924. 924 .unwrap()["session_id"]
925. 925 .as_str()
926. 926 .unwrap()
927. 927 .to_string();
928. 928
929. 929 client
930. 930 .post(format!("{base}/sessions/{session_id}/run"))
931. 931 .json(&serde_json::json!({"prompt": "hang forever"}))
932. 932 .send()
933. 933 .await
934. 934 .unwrap();
935. 935
936. 936 // Give the hung leg a moment to actually take the ledger.
937. 937 tokio::time::sleep(Duration::from_millis(200)).await;
938. 938 let cancelled = client
939. 939 .post(format!("{base}/sessions/{session_id}/cancel"))
940. 940 .send()
941. 941 .await
942. 942 .unwrap();
943. 943 assert_eq!(cancelled.status(), 202);
944. 944
945. 945 // Resend immediately — this may land while the cancelled leg is still
946. 946 // unwinding (queued) or just after (started); both must eventually run.
947. 947 let resend = client
948. 948 .post(format!("{base}/sessions/{session_id}/run"))
949. 949 .json(&serde_json::json!({"prompt": "resend", "request_id": "resend-1"}))
950. 950 .send()
951. 951 .await
952. 952 .unwrap();
953. 953 assert_eq!(resend.status(), 202);
954. 954 let resend_body: serde_json::Value = resend.json().await.unwrap();
955. 955 assert!(
956. 956 matches!(
957. 957 resend_body["state"].as_str(),
958. 958 Some("started") | Some("queued")
959. 959 ),
960. 960 "resend must be admitted, not rejected: {resend_body}"
961. 961 );
962. 962
963. 963 let deadline = std::time::Instant::now() + Duration::from_secs(10);
964. 964 let raw = poll_transcript_contains(&client, &base, &session_id, "resend done", deadline).await;
965. 965 assert!(
966. 966 raw.contains("resend done"),
967. 967 "a resend right after stop must actually run: {raw}"
968. 968 );
969. 969}
970. 970
971. 971/// Finding 2: `run_prompt` used to wait up to 2s on `handle.subscribed`
972. 972/// unconditionally, but `Notify`'s single permit only ever satisfies the
973. 973/// FIRST run — every later run paid the full timeout even with a client
974. 974/// already attached. With an SSE consumer already connected, `/run` must
975. 975/// return promptly.
976. 976#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
977. 977async fn no_wait_when_an_sse_subscriber_is_already_attached() {
978. 978 let provider = Arc::new(Scripted {
979. 979 responses: Mutex::new(VecDeque::from(vec![text("fast reply")])),
980. 980 });
981. 981 let (base, _server) = spawn_server(provider, vak_config::PermissionMode::FullAccess).await;
982. 982 let client = reqwest::Client::new();
983. 983 let session_id: String = client
984. 984 .post(format!("{base}/sessions"))
985. 985 .send()
986. 986 .await
987. 987 .unwrap()
988. 988 .json::<serde_json::Value>()
989. 989 .await
990. 990 .unwrap()["session_id"]
991. 991 .as_str()
992. 992 .unwrap()
993. 993 .to_string();
994. 994
995. 995 // Attach and confirm the stream is open before timing the run.
996. 996 let sse_url = format!("{base}/sessions/{session_id}/events");
997. 997 let (opened_tx, opened_rx) = tokio::sync::oneshot::channel::<()>();
998. 998 let mut opened_tx = Some(opened_tx);
999. 999 let _sse_task = tokio::spawn(async move {
1000. 1000 let res = reqwest::get(&sse_url).await.unwrap();