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, 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. 42/// Spawn `secured_router` exactly like the desktop shell will: bind an
43. 43/// ephemeral loopback port and keep the token for authenticated calls.
44. 44async fn spawn_secured(
45. 45 provider: Arc<dyn Provider>,
46. 46) -> (
47. 47 String,
48. 48 String,
49. 49 std::path::PathBuf,
50. 50 tokio::task::JoinHandle<()>,
51. 51) {
52. 52 let dir = tempfile::tempdir().unwrap();
53. 53 let cwd = dir.path().to_path_buf();
54. 54 // Isolate from the developer's real global config: layered config
55. 55 // loads $HOME/.config/vak/config.toml, and a personal
56. 56 // `[memory] reflection = true` would make the post-turn reflection
57. 57 // seam consume scripted provider responses mid-test.
58. 58 let project = cwd.join(".vak");
59. 59 std::fs::create_dir_all(&project).unwrap();
60. 60 std::fs::write(
61. 61 project.join("config.toml"),
62. 62 "[memory]\nreflection = false\n",
63. 63 )
64. 64 .unwrap();
65. 65 vak_config::paths::isolate_home_for_tests();
66. 66 let core = Core::new(cwd.clone()).unwrap();
67. 67 core.set_sessions_home(dir.path().join("home"));
68. 68 // The real brokered-tool worker: under `cargo test`, `current_exe()` is
69. 69 // the test harness, which answers a preview launch with "0 tests" and
70. 70 // exits (as in gateway.rs and scheduler_personal_os.rs).
71. 71 core.set_tool_worker_exe(std::path::PathBuf::from(env!(
72. 72 "CARGO_BIN_EXE_vak-tool-worker"
73. 73 )));
74. 74 core.set_provider_instance(provider);
75. 75 // keep tempdir alive for the process lifetime of the test
76. 76 std::mem::forget(dir);
77. 77
78. 78 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
79. 79 let addr = listener.local_addr().unwrap();
80. 80 let (app, token) = vak_server::secured_router(core);
81. 81 let handle = tokio::spawn(async move {
82. 82 axum::serve(listener, app).await.unwrap();
83. 83 });
84. 84 (format!("http://{addr}"), token, cwd, handle)
85. 85}
86. 86
87. 87fn client_with(token: &str) -> reqwest::Client {
88. 88 reqwest::ClientBuilder::new()
89. 89 .default_headers({
90. 90 let mut h = reqwest::header::HeaderMap::new();
91. 91 h.insert(
92. 92 reqwest::header::AUTHORIZATION,
93. 93 format!("Bearer {token}").parse().unwrap(),
94. 94 );
95. 95 h
96. 96 })
97. 97 .build()
98. 98 .unwrap()
99. 99}
100. 100
101. 101fn text(t: &str) -> AssistantMessage {
102. 102 AssistantMessage {
103. 103 content: vec![vak_llm::types::ContentBlock::text(t)],
104. 104 stop_reason: vak_llm::types::StopReason::EndTurn,
105. 105 usage: Usage {
106. 106 input_tokens: 7,
107. 107 output_tokens: 3,
108. 108 ..Default::default()
109. 109 },
110. 110 model: "test-model".into(),
111. 111 response_id: None,
112. 112 }
113. 113}
114. 114
115. 115async fn wait_transcript(client: &reqwest::Client, base: &str, id: &str) -> serde_json::Value {
116. 116 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
117. 117 loop {
118. 118 assert!(std::time::Instant::now() < deadline, "transcript timeout");
119. 119 if let Ok(res) = client
120. 120 .get(format!("{base}/sessions/{id}/transcript"))
121. 121 .send()
122. 122 .await
123. 123 && res.status() == reqwest::StatusCode::OK
124. 124 {
125. 125 let body: serde_json::Value = res.json().await.unwrap();
126. 126 // While a run is live the endpoint answers 200 with an error
127. 127 // payload; keep polling until the real transcript lands.
128. 128 if body.get("error").is_none() {
129. 129 return body;
130. 130 }
131. 131 }
132. 132 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
133. 133 }
134. 134}
135. 135
136. 136#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
137. 137async fn secured_router_enforces_token_and_cors() {
138. 138 let (base, token, _cwd, _server) = spawn_secured(Arc::new(Scripted {
139. 139 responses: Mutex::new(VecDeque::new()),
140. 140 }))
141. 141 .await;
142. 142
143. 143 // /health is open by design.
144. 144 let health = reqwest::get(format!("{base}/health")).await.unwrap();
145. 145 assert_eq!(health.status(), 200);
146. 146
147. 147 // Missing or wrong bearer tokens are rejected.
148. 148 let anon = reqwest::get(format!("{base}/sessions")).await.unwrap();
149. 149 assert_eq!(anon.status(), 401);
150. 150 let wrong = reqwest::Client::new()
151. 151 .get(format!("{base}/sessions"))
152. 152 .bearer_auth("vk_wrong")
153. 153 .send()
154. 154 .await
155. 155 .unwrap();
156. 156 assert_eq!(wrong.status(), 401);
157. 157
158. 158 // Both Tauri webview origins must be allowed through CORS preflight.
159. 159 for origin in ["tauri://localhost", "http://tauri.localhost"] {
160. 160 let preflight = reqwest::Client::new()
161. 161 .request(reqwest::Method::OPTIONS, format!("{base}/sessions"))
162. 162 .header(reqwest::header::ORIGIN, origin)
163. 163 .header(reqwest::header::ACCESS_CONTROL_REQUEST_METHOD, "POST")
164. 164 .send()
165. 165 .await
166. 166 .unwrap();
167. 167 assert_eq!(
168. 168 preflight
169. 169 .headers()
170. 170 .get(reqwest::header::ACCESS_CONTROL_ALLOW_ORIGIN),
171. 171 Some(&reqwest::header::HeaderValue::from_bytes(origin.as_bytes()).unwrap()),
172. 172 );
173. 173 }
174. 174
175. 175 // The real token grants access.
176. 176 let ok = client_with(&token)
177. 177 .get(format!("{base}/sessions"))
178. 178 .send()
179. 179 .await
180. 180 .unwrap();
181. 181 assert_eq!(ok.status(), 200);
182. 182}
183. 183
184. 184#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
185. 185async fn sessions_list_attach_and_title_roundtrip() {
186. 186 // Build the persisted ledger directly, then drop it so the lock is
187. 187 // released — mirroring "TUI wrote this session earlier and exited".
188. 188 let dir = tempfile::tempdir().unwrap();
189. 189 let cwd = dir.path().to_path_buf();
190. 190 vak_config::paths::isolate_home_for_tests();
191. 191 let core_a = Core::new(cwd.clone()).unwrap();
192. 192 core_a.set_sessions_home(cwd.join("home"));
193. 193 std::mem::forget(dir);
194. 194
195. 195 let mut log = core_a.start_session().await.unwrap();
196. 196 let session_id = log.header().unwrap().session_id.clone();
197. 197 use vak_llm::types::{ContentBlock as CB, Role};
198. 198 log.append_message(vak_session::MessageRecord {
199. 199 message: vak_llm::Message {
200. 200 role: Role::User,
201. 201 content: vec![CB::text("title probe here")],
202. 202 },
203. 203 meta: None,
204. 204 })
205. 205 .unwrap();
206. 206 log.append_message(vak_session::MessageRecord {
207. 207 message: vak_llm::Message {
208. 208 role: Role::Assistant,
209. 209 content: vec![CB::text("all done")],
210. 210 },
211. 211 meta: None,
212. 212 })
213. 213 .unwrap();
214. 214 drop(log);
215. 215
216. 216 // A fresh server over the same store: lists the session, resumes it.
217. 217 vak_config::paths::isolate_home_for_tests();
218. 218 let core = Core::new(cwd.clone()).unwrap();
219. 219 core.set_sessions_home(cwd.join("home"));
220. 220 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
221. 221 let addr = listener.local_addr().unwrap();
222. 222 let (app, token) = vak_server::secured_router(core);
223. 223 tokio::spawn(async move {
224. 224 axum::serve(listener, app).await.unwrap();
225. 225 });
226. 226 let base = format!("http://{addr}");
227. 227 let client = client_with(&token);
228. 228
229. 229 // The sidebar projection must see the persisted file with a usable title.
230. 230 let listed = client.get(format!("{base}/sessions")).send().await.unwrap();
231. 231 let body: serde_json::Value = listed.json().await.unwrap();
232. 232 let entry = body["sessions"]
233. 233 .as_array()
234. 234 .unwrap()
235. 235 .iter()
236. 236 .find(|s| s["session_id"] == session_id.as_str())
237. 237 .expect("persisted session must be listed");
238. 238 assert_eq!(entry["title"], "title probe here");
239. 239 assert!(entry["updated_at"].is_string());
240. 240 assert_eq!(entry["running"], false);
241. 241
242. 242 let attached = client
243. 243 .post(format!("{base}/sessions/{session_id}/attach"))
244. 244 .json(&serde_json::json!({"session_id": session_id}))
245. 245 .send()
246. 246 .await
247. 247 .unwrap();
248. 248 assert_eq!(attached.status(), 200);
249. 249
250. 250 let resumed: serde_json::Value = client
251. 251 .get(format!("{base}/sessions/{session_id}/transcript"))
252. 252 .send()
253. 253 .await
254. 254 .unwrap()
255. 255 .json()
256. 256 .await
257. 257 .unwrap();
258. 258 assert_eq!(resumed["count"].as_u64(), Some(2));
259. 259}
260. 260
261. 261#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
262. 262async fn persisted_conversation_accepts_followup_and_streams_without_explicit_attach() {
263. 263 let dir = tempfile::tempdir().unwrap();
264. 264 let cwd = dir.path().to_path_buf();
265. 265 vak_config::paths::isolate_home_for_tests();
266. 266 let core = Core::new(cwd.clone()).unwrap();
267. 267 core.set_sessions_home(cwd.join("home"));
268. 268 let mut session = core.start_session().await.unwrap();
269. 269 let id = session.header().unwrap().session_id.clone();
270. 270 session
271. 271 .append_message(vak_session::MessageRecord {
272. 272 message: vak_llm::Message {
273. 273 role: vak_llm::types::Role::User,
274. 274 content: vec![vak_llm::types::ContentBlock::text("Earlier question")],
275. 275 },
276. 276 meta: None,
277. 277 })
278. 278 .unwrap();
279. 279 session
280. 280 .append_message(vak_session::MessageRecord {
281. 281 message: vak_llm::Message {
282. 282 role: vak_llm::types::Role::Assistant,
283. 283 content: vec![vak_llm::types::ContentBlock::text("Earlier answer")],
284. 284 },
285. 285 meta: None,
286. 286 })
287. 287 .unwrap();
288. 288 drop(session);
289. 289
290. 290 // A new process has no live handle. EventSources reconnect before the
291. 291 // browser sends a follow-up, so both streams and /run must recover it.
292. 292 let resumed = Core::new(cwd.clone()).unwrap();
293. 293 resumed.set_sessions_home(cwd.join("home"));
294. 294 resumed.set_provider_instance(Arc::new(Scripted {
295. 295 responses: Mutex::new(VecDeque::from(vec![text("Follow-up answer")])),
296. 296 }));
297. 297 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
298. 298 let addr = listener.local_addr().unwrap();
299. 299 let (app, token) = vak_server::secured_router(resumed);
300. 300 tokio::spawn(async move {
301. 301 axum::serve(listener, app).await.unwrap();
302. 302 });
303. 303 let base = format!("http://{addr}");
304. 304 let client = client_with(&token);
305. 305
306. 306 let events = client
307. 307 .get(format!("{base}/sessions/{id}/events"))
308. 308 .send()
309. 309 .await
310. 310 .unwrap();
311. 311 assert_eq!(events.status(), 200);
312. 312 let presentation = client
313. 313 .get(format!("{base}/sessions/{id}/presentation/events"))
314. 314 .send()
315. 315 .await
316. 316 .unwrap();
317. 317 assert_eq!(presentation.status(), 200);
318. 318 drop(events);
319. 319 drop(presentation);
320. 320
321. 321 let started = client
322. 322 .post(format!("{base}/sessions/{id}/run"))
323. 323 .json(&serde_json::json!({"prompt": "Follow-up question"}))
324. 324 .send()
325. 325 .await
326. 326 .unwrap();
327. 327 assert_eq!(started.status(), 202);
328. 328 let deadline = std::time::Instant::now() + Duration::from_secs(15);
329. 329 loop {
330. 330 assert!(std::time::Instant::now() < deadline, "follow-up timeout");
331. 331 let transcript = wait_transcript(&client, &base, &id).await;
332. 332 if transcript["count"].as_u64() == Some(4) {
333. 333 break;
334. 334 }
335. 335 tokio::time::sleep(Duration::from_millis(50)).await;
336. 336 }
337. 337}
338. 338
339. 339#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
340. 340async fn sessions_list_hides_abandoned_header_only_drafts() {
341. 341 let dir = tempfile::tempdir().unwrap();
342. 342 let cwd = dir.path().to_path_buf();
343. 343 vak_config::paths::isolate_home_for_tests();
344. 344 let core = Core::new(cwd.clone()).unwrap();
345. 345 core.set_sessions_home(cwd.join("home"));
346. 346 let draft = core.start_session().await.unwrap();
347. 347 let draft_id = draft.header().unwrap().session_id.clone();
348. 348 drop(draft);
349. 349
350. 350 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
351. 351 let addr = listener.local_addr().unwrap();
352. 352 let (app, token) = vak_server::secured_router(core);
353. 353 tokio::spawn(async move {
354. 354 axum::serve(listener, app).await.unwrap();
355. 355 });
356. 356
357. 357 let body: serde_json::Value = client_with(&token)
358. 358 .get(format!("http://{addr}/sessions"))
359. 359 .send()
360. 360 .await
361. 361 .unwrap()
362. 362 .json()
363. 363 .await
364. 364 .unwrap();
365. 365 assert!(
366. 366 body["sessions"]
367. 367 .as_array()
368. 368 .unwrap()
369. 369 .iter()
370. 370 .all(|session| session["session_id"] != draft_id)
371. 371 );
372. 372}
373. 373
374. 374#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
375. 375async fn fs_endpoints_are_confined_to_workspace() {
376. 376 let (base, token, cwd, _server) = spawn_secured(Arc::new(Scripted {
377. 377 responses: Mutex::new(VecDeque::new()),
378. 378 }))
379. 379 .await;
380. 380 let client = client_with(&token);
381. 381
382. 382 // Write inside the workspace.
383. 383 let put = client
384. 384 .put(format!("{base}/fs/file"))
385. 385 .json(&serde_json::json!({"path": "notes/hello.txt", "content": "hi"}))
386. 386 .send()
387. 387 .await
388. 388 .unwrap();
389. 389 assert_eq!(put.status(), 200);
390. 390 assert_eq!(
391. 391 tokio::fs::read_to_string(cwd.join("notes/hello.txt"))
392. 392 .await
393. 393 .unwrap(),
394. 394 "hi"
395. 395 );
396. 396
397. 397 let got: serde_json::Value = client
398. 398 .get(format!("{base}/fs/file?path=notes/hello.txt"))
399. 399 .send()
400. 400 .await
401. 401 .unwrap()
402. 402 .json()
403. 403 .await
404. 404 .unwrap();
405. 405 assert_eq!(got["content"], "hi");
406. 406
407. 407 // Traversal escape is rejected.
408. 408 let escape_put = client
409. 409 .put(format!("{base}/fs/file"))
410. 410 .json(&serde_json::json!({"path": "../evil.txt", "content": "nope"}))
411. 411 .send()
412. 412 .await
413. 413 .unwrap();
414. 414 assert_eq!(escape_put.status(), 403);
415. 415
416. 416 let escape_get = client
417. 417 .get(format!("{base}/fs/file?path=/etc/hostname"))
418. 418 .send()
419. 419 .await
420. 420 .unwrap();
421. 421 assert_eq!(escape_get.status(), 403);
422. 422
423. 423 let missing = client
424. 424 .get(format!("{base}/fs/file?path=no/such.txt"))
425. 425 .send()
426. 426 .await
427. 427 .unwrap();
428. 428 assert_eq!(missing.status(), 404);
429. 429}
430. 430
431. 431#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
432. 432async fn mode_switch_and_diff_endpoint() {
433. 433 let (base, token, cwd, _server) = spawn_secured(Arc::new(Scripted {
434. 434 responses: Mutex::new(VecDeque::new()),
435. 435 }))
436. 436 .await;
437. 437 let client = client_with(&token);
438. 438
439. 439 let session_id: String = client
440. 440 .post(format!("{base}/sessions"))
441. 441 .send()
442. 442 .await
443. 443 .unwrap()
444. 444 .json::<serde_json::Value>()
445. 445 .await
446. 446 .unwrap()["session_id"]
447. 447 .as_str()
448. 448 .unwrap()
449. 449 .to_string();
450. 450
451. 451 // Unknown mode is a bad request.
452. 452 let bad = client
453. 453 .post(format!("{base}/config/mode"))
454. 454 .json(&serde_json::json!({"mode": "yolo"}))
455. 455 .send()
456. 456 .await
457. 457 .unwrap();
458. 458 assert_eq!(bad.status(), 400);
459. 459
460. 460 let switched = client
461. 461 .post(format!("{base}/config/mode"))
462. 462 .json(&serde_json::json!({"mode": "read-only"}))
463. 463 .send()
464. 464 .await
465. 465 .unwrap();
466. 466 assert_eq!(switched.status(), 200);
467. 467 let health: serde_json::Value = reqwest::get(format!("{base}/health"))
468. 468 .await
469. 469 .unwrap()
470. 470 .json()
471. 471 .await
472. 472 .unwrap();
473. 473 assert_eq!(health["permission_mode"], "ReadOnly");
474. 474
475. 475 let config: serde_json::Value = client
476. 476 .get(format!("{base}/config"))
477. 477 .send()
478. 478 .await
479. 479 .unwrap()
480. 480 .json()
481. 481 .await
482. 482 .unwrap();
483. 483 assert_eq!(config["permission_mode"], "ReadOnly");
484. 484 assert!(config["paths"]["project_config"].is_string());
485. 485 assert!(config["integrations"]["skills"].is_array());
486. 486
487. 487 let patched = client
488. 488 .patch(format!("{base}/config"))
489. 489 .json(&serde_json::json!({
490. 490 "provider": "google",
491. 491 "model": "gemini-test",
492. 492 "max_turns": 17,
493. 493 "permission_mode": "workspace-write"
494. 494 }))
495. 495 .send()
496. 496 .await
497. 497 .unwrap();
498. 498 assert_eq!(patched.status(), 200);
499. 499 let updated: serde_json::Value = client
500. 500 .get(format!("{base}/config"))
501. 501 .send()
502. 502 .await
503. 503 .unwrap()
504. 504 .json()
505. 505 .await
506. 506 .unwrap();
507. 507 assert_eq!(updated["provider"], "google");
508. 508 assert_eq!(updated["model"], "gemini-test");
509. 509 assert_eq!(updated["max_turns"], 17);
510. 510 assert_eq!(updated["permission_mode"], "WorkspaceWrite");
511. 511 assert_eq!(updated["provider_source"], "project_config");
512. 512 assert_eq!(updated["model_source"], "project_config");
513. 513 assert!(
514. 514 updated["route_revision"]
515. 515 .as_str()
516. 516 .is_some_and(|r| r.starts_with('r'))
517. 517 );
518. 518 let persisted = tokio::fs::read_to_string(cwd.join(".vak/config.toml"))
519. 519 .await
520. 520 .unwrap();
521. 521 assert!(persisted.contains("provider = \"google\""));
522. 522 assert!(persisted.contains("model = \"gemini-test\""));
523. 523 vak_config::paths::isolate_home_for_tests();
524. 524 let restarted = Core::new_with_trust(cwd, true).unwrap();
525. 525 assert_eq!(restarted.effective_provider(), "google");
526. 526 assert_eq!(restarted.effective_model(), "gemini-test");
527. 527
528. 528 // Tempdir is not a git repo: diff endpoint reports that as a value.
529. 529 let diff: serde_json::Value = client
530. 530 .get(format!("{base}/sessions/{session_id}/diff"))
531. 531 .send()
532. 532 .await
533. 533 .unwrap()
534. 534 .json()
535. 535 .await
536. 536 .unwrap();
537. 537 assert_eq!(diff["error"], "not a git repository");
538. 538}
539. 539
540. 540#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
541. 541async fn fs_tree_lists_and_skips_vendored_dirs() {
542. 542 let (base, token, cwd, _server) = spawn_secured(Arc::new(Scripted {
543. 543 responses: Mutex::new(VecDeque::new()),
544. 544 }))
545. 545 .await;
546. 546 let client = client_with(&token);
547. 547
548. 548 tokio::fs::create_dir_all(cwd.join("src/deep"))
549. 549 .await
550. 550 .unwrap();
551. 551 tokio::fs::write(cwd.join("src/main.rs"), "fn main() {}")
552. 552 .await
553. 553 .unwrap();
554. 554 tokio::fs::write(cwd.join("src/deep/util.rs"), "pub fn u() {}")
555. 555 .await
556. 556 .unwrap();
557. 557 tokio::fs::create_dir_all(cwd.join("node_modules/left-pad"))
558. 558 .await
559. 559 .unwrap();
560. 560 tokio::fs::write(cwd.join("node_modules/left-pad/index.js"), "x")
561. 561 .await
562. 562 .unwrap();
563. 563
564. 564 let body: serde_json::Value = client
565. 565 .get(format!("{base}/fs/tree"))
566. 566 .send()
567. 567 .await
568. 568 .unwrap()
569. 569 .json()
570. 570 .await
571. 571 .unwrap();
572. 572
573. 573 let files: Vec<String> = body["files"]
574. 574 .as_array()
575. 575 .unwrap()
576. 576 .iter()
577. 577 .map(|f| f.as_str().unwrap().to_string())
578. 578 .collect();
579. 579 assert!(
580. 580 files.contains(&"src/main.rs".to_string()),
581. 581 "files: {files:?}"
582. 582 );
583. 583 assert!(files.contains(&"src/deep/util.rs".to_string()));
584. 584 assert!(
585. 585 !files.iter().any(|f| f.starts_with("node_modules")),
586. 586 "vendored dirs must be skipped"
587. 587 );
588. 588}
589. 589
590. 590#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
591. 591async fn failed_side_turn_does_not_wedge_the_session() {
592. 592 // Regression (v0.6 deployment gate): whatever way a side turn ends
593. 593 // without a restorable log — provider exhausted into endless endurance
594. 594 // retries, or an explicit cancel — the session handle must become
595. 595 // usable again instead of answering "run in progress" forever.
596. 596 let provider = Arc::new(Scripted {
597. 597 responses: Mutex::new(VecDeque::from(vec![
598. 598 text("main answer"),
599. 599 text("side answer"),
600. 600 ])),
601. 601 });
602. 602 let (base, token, _cwd, _server) = spawn_secured(provider).await;
603. 603 let client = client_with(&token);
604. 604
605. 605 let session_id: String = client
606. 606 .post(format!("{base}/sessions"))
607. 607 .send()
608. 608 .await
609. 609 .unwrap()
610. 610 .json::<serde_json::Value>()
611. 611 .await
612. 612 .unwrap()["session_id"]
613. 613 .as_str()
614. 614 .unwrap()
615. 615 .to_string();
616. 616
617. 617 client
618. 618 .post(format!("{base}/sessions/{session_id}/run"))
619. 619 .json(&serde_json::json!({"prompt": "say main answer"}))
620. 620 .send()
621. 621 .await
622. 622 .unwrap();
623. 623 let before = wait_transcript(&client, &base, &session_id).await;
624. 624 assert_eq!(before["count"].as_u64(), Some(2));
625. 625
626. 626 // Start a side turn and cancel it mid-flight.
627. 627 let started = client
628. 628 .post(format!("{base}/sessions/{session_id}/side"))
629. 629 .json(&serde_json::json!({"question": "long aside?"}))
630. 630 .send()
631. 631 .await
632. 632 .unwrap();
633. 633 assert_eq!(started.status(), 202);
634. 634 let cancelled = client
635. 635 .post(format!("{base}/sessions/{session_id}/side/cancel"))
636. 636 .send()
637. 637 .await
638. 638 .unwrap();
639. 639 assert_eq!(cancelled.status(), 202);
640. 640
641. 641 // The handle must come back: transcript answers with real content and
642. 642 // a follow-up side turn is accepted rather than 409-conflicted.
643. 643 let after = wait_transcript(&client, &base, &session_id).await;
644. 644 assert_eq!(
645. 645 after["count"].as_u64(),
646. 646 Some(2),
647. 647 "cancelled side turn must leave the main chain intact"
648. 648 );
649. 649 let again = client
650. 650 .post(format!("{base}/sessions/{session_id}/side"))
651. 651 .json(&serde_json::json!({"question": "still here?"}))
652. 652 .send()
653. 653 .await
654. 654 .unwrap();
655. 655 assert_eq!(again.status(), 202, "session handle must be restorable");
656. 656}
657. 657
658. 658#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
659. 659async fn side_chat_branches_off_and_restores_main_chain() {
660. 660 let provider = Arc::new(Scripted {
661. 661 responses: Mutex::new(VecDeque::from(vec![
662. 662 text("main answer"),
663. 663 text("side answer"),
664. 664 ])),
665. 665 });
666. 666 let (base, token, cwd, _server) = spawn_secured(provider).await;
667. 667 let client = client_with(&token);
668. 668
669. 669 let session_id: String = client
670. 670 .post(format!("{base}/sessions"))
671. 671 .send()
672. 672 .await
673. 673 .unwrap()
674. 674 .json::<serde_json::Value>()
675. 675 .await
676. 676 .unwrap()["session_id"]
677. 677 .as_str()
678. 678 .unwrap()
679. 679 .to_string();
680. 680
681. 681 client
682. 682 .post(format!("{base}/sessions/{session_id}/run"))
683. 683 .json(&serde_json::json!({"prompt": "say main answer"}))
684. 684 .send()
685. 685 .await
686. 686 .unwrap();
687. 687 let before = wait_transcript(&client, &base, &session_id).await;
688. 688 assert_eq!(before["count"].as_u64(), Some(2));
689. 689
690. 690 let started = client
691. 691 .post(format!("{base}/sessions/{session_id}/side"))
692. 692 .json(&serde_json::json!({"question": "quick aside?"}))
693. 693 .send()
694. 694 .await
695. 695 .unwrap();
696. 696 assert_eq!(started.status(), 202);
697. 697
698. 698 // Once the side turn completes, the main view is exactly what it was.
699. 699 let after = wait_transcript(&client, &base, &session_id).await;
700. 700 assert_eq!(after["count"].as_u64(), Some(2), "main chain untouched");
701. 701 assert!(
702. 702 !serde_json::to_string(&after)
703. 703 .unwrap()
704. 704 .contains("side answer"),
705. 705 "side answer must not leak into derive_messages()"
706. 706 );
707. 707 assert!(
708. 708 serde_json::to_string(&after)
709. 709 .unwrap()
710. 710 .contains("main answer"),
711. 711 "main context must remain intact"
712. 712 );
713. 713
714. 714 // ...but the sibling branch stays reconstructable in the JSONL.
715. 715 tokio::time::sleep(Duration::from_millis(100)).await;
716. 716 let agent_home = cwd.join("home").join("agents").join("vak");
717. 717 let home = if agent_home.exists() {
718. 718 agent_home
719. 719 } else {
720. 720 cwd.join("home")
721. 721 };
722. 722 let mut dir = std::fs::read_dir(vak_session::SessionPath::sessions_dir(&home, &cwd)).unwrap();
723. 723 let path = dir.next().unwrap().unwrap().path();
724. 724 let raw = std::fs::read_to_string(path).unwrap();
725. 725 assert!(raw.contains("quick aside?"), "question persisted");
726. 726 assert!(
727. 727 raw.contains("side answer"),
728. 728 "answer persisted as branch entry"
729. 729 );
730. 730}
731. 731
732. 732fn git_seed(cwd: &std::path::Path) {
733. 733 let run = |args: &[&str]| {
734. 734 let out = std::process::Command::new("git")
735. 735 .args(args)
736. 736 .current_dir(cwd)
737. 737 .env("GIT_AUTHOR_NAME", "t")
738. 738 .env("GIT_AUTHOR_EMAIL", "t@t")
739. 739 .env("GIT_COMMITTER_NAME", "t")
740. 740 .env("GIT_COMMITTER_EMAIL", "t@t")
741. 741 .output()
742. 742 .unwrap();
743. 743 assert!(
744. 744 out.status.success(),
745. 745 "git {args:?} failed: {}",
746. 746 String::from_utf8_lossy(&out.stderr)
747. 747 );
748. 748 };
749. 749 run(&["init", "-q"]);
750. 750 std::fs::write(cwd.join("README.md"), "seed\n").unwrap();
751. 751 run(&["add", "."]);
752. 752 run(&["commit", "-q", "-m", "seed"]);
753. 753}
754. 754
755. 755#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
756. 756async fn bestofn_fans_out_keep_and_discard() {
757. 757 let provider = Arc::new(Scripted {
758. 758 responses: Mutex::new(VecDeque::from(vec![
759. 759 text("candidate A"),
760. 760 text("candidate B"),
761. 761 ])),
762. 762 });
763. 763 let (base, token, cwd, _server) = spawn_secured(provider).await;
764. 764 let client = client_with(&token);
765. 765 git_seed(&cwd);
766. 766
767. 767 let anchor: String = client
768. 768 .post(format!("{base}/sessions"))
769. 769 .send()
770. 770 .await
771. 771 .unwrap()
772. 772 .json::<serde_json::Value>()
773. 773 .await
774. 774 .unwrap()["session_id"]
775. 775 .as_str()
776. 776 .unwrap()
777. 777 .to_string();
778. 778
779. 779 // Not a repo guard would have fired before seed; now it must succeed.
780. 780 let res = client
781. 781 .post(format!("{base}/sessions/{anchor}/bestofn"))
782. 782 .json(&serde_json::json!({"prompt": "solve it", "n": 2}))
783. 783 .send()
784. 784 .await
785. 785 .unwrap();
786. 786 let status = res.status();
787. 787 let body: serde_json::Value = res.json().await.unwrap();
788. 788 assert_eq!(status, 200, "start failed: {body}");
789. 789 let runs = body["runs"].as_array().unwrap().clone();
790. 790 assert_eq!(runs.len(), 2);
791. 791
792. 792 let wt_root = cwd.join(".vak/worktrees");
793. 793 let wait_child = |cid: String| {
794. 794 let client = client.clone();
795. 795 let base = base.clone();
796. 796 async move { wait_transcript(&client, &base, &cid).await }
797. 797 };
798. 798
799. 799 let mut child_ids = Vec::new();
800. 800 for r in &runs {
801. 801 let cid = r["session_id"].as_str().unwrap().to_string();
802. 802 let t = wait_child(cid.clone()).await;
803. 803 assert_eq!(t["count"].as_u64(), Some(2));
804. 804 child_ids.push(cid);
805. 805 }
806. 806
807. 807 // Both worktrees exist on disk.
808. 808 assert_eq!(std::fs::read_dir(&wt_root).unwrap().count(), 2);
809. 809
810. 810 // Discard one: worktree + branch gone.
811. 811 let disc = client
812. 812 .post(format!("{base}/sessions/{}/discard", child_ids[1]))
813. 813 .send()
814. 814 .await
815. 815 .unwrap();
816. 816 assert_eq!(disc.status(), 200);
817. 817 assert_eq!(std::fs::read_dir(&wt_root).unwrap().count(), 1);
818. 818
819. 819 // Keep the other: endpoint merges (up-to-date is fine for text-only
820. 820 // candidates) and always cleans up the worktree + branch.
821. 821 let keep = client
822. 822 .post(format!("{base}/sessions/{}/keep", child_ids[0]))
823. 823 .send()
824. 824 .await
825. 825 .unwrap();
826. 826 assert_eq!(
827. 827 keep.status(),
828. 828 200,
829. 829 "keep failed: {}",
830. 830 keep.text().await.unwrap_or_default()
831. 831 );
832. 832
833. 833 assert_eq!(
834. 834 std::fs::read_dir(&wt_root).unwrap().count(),
835. 835 0,
836. 836 "worktrees all cleaned"
837. 837 );
838. 838 let branches = {
839. 839 let o = std::process::Command::new("git")
840. 840 .args(["branch", "--list", "vak/*"])
841. 841 .current_dir(&cwd)
842. 842 .output()
843. 843 .unwrap();
844. 844 String::from_utf8_lossy(&o.stdout).trim().to_string()
845. 845 };
846. 846 assert!(
847. 847 branches.is_empty(),
848. 848 "candidate branches must be gone: {branches}"
849. 849 );
850. 850}
851. 851
852. 852#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
853. 853async fn pr_endpoints_surface_structured_results() {
854. 854 let (base, token, cwd, _server) = spawn_secured(Arc::new(Scripted {
855. 855 responses: Mutex::new(VecDeque::new()),
856. 856 }))
857. 857 .await;
858. 858 let client = client_with(&token);
859. 859 git_seed_main(&cwd);
860. 860
861. 861 let anchor: String = client
862. 862 .post(format!("{base}/sessions"))
863. 863 .send()
864. 864 .await
865. 865 .unwrap()
866. 866 .json::<serde_json::Value>()
867. 867 .await
868. 868 .unwrap()["session_id"]
869. 869 .as_str()
870. 870 .unwrap()
871. 871 .to_string();
872. 872
873. 873 // Status is always structured JSON: either a PR view, a no_pr/gh reason,
874. 874 // or an error — never a hang.
875. 875 let body: serde_json::Value = client
876. 876 .get(format!("{base}/sessions/{anchor}/pr"))
877. 877 .send()
878. 878 .await
879. 879 .unwrap()
880. 880 .json()
881. 881 .await
882. 882 .unwrap();
883. 883 assert_eq!(body["branch"], "main");
884. 884 assert!(body.get("pr").is_some(), "must carry pr slot: {body}");
885. 885 if body["pr"].is_null() {
886. 886 assert!(body.get("reason").is_some() || body.get("error").is_some());
887. 887 }
888. 888
889. 889 // Merge endpoint surfaces gh failures as values (409), not hangs.
890. 890 let merged = client
891. 891 .post(format!("{base}/sessions/{anchor}/pr/merge"))
892. 892 .json(&serde_json::json!({"number": 999}))
893. 893 .send()
894. 894 .await
895. 895 .unwrap();
896. 896 assert!(
897. 897 merged.status() == 200 || merged.status() == 409,
898. 898 "merge must resolve deterministically"
899. 899 );
900. 900
901. 901 // Unknown session → 404 on merge.
902. 902 let missing = client
903. 903 .post(format!("{base}/sessions/nope/pr/merge"))
904. 904 .json(&serde_json::json!({"number": 1}))
905. 905 .send()
906. 906 .await
907. 907 .unwrap();
908. 908 assert_eq!(missing.status(), 404);
909. 909}
910. 910
911. 911fn git_seed_main(cwd: &std::path::Path) {
912. 912 let run = |args: &[&str]| {
913. 913 let out = std::process::Command::new("git")
914. 914 .args(args)
915. 915 .current_dir(cwd)
916. 916 .env("GIT_AUTHOR_NAME", "t")
917. 917 .env("GIT_AUTHOR_EMAIL", "t@t")
918. 918 .env("GIT_COMMITTER_NAME", "t")
919. 919 .env("GIT_COMMITTER_EMAIL", "t@t")
920. 920 .output()
921. 921 .unwrap();
922. 922 assert!(out.status.success(), "git {args:?} failed");
923. 923 };
924. 924 run(&["init", "-q", "-b", "main"]);
925. 925 std::fs::write(cwd.join("README.md"), "seed\n").unwrap();
926. 926 run(&["add", "."]);
927. 927 run(&["commit", "-q", "-m", "seed"]);
928. 928}
929. 929
930. 930#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
931. 931async fn scheduled_tasks_crud_runnow_and_worktree_churn() {
932. 932 let provider = Arc::new(Scripted {
933. 933 responses: Mutex::new(VecDeque::from(vec![
934. 934 text("task ran"),
935. 935 text("task ran again"),
936. 936 ])),
937. 937 });
938. 938 let (base, token, cwd, _server) = spawn_secured(provider).await;
939. 939 let client = client_with(&token);
940. 940 git_seed_main(&cwd);
941. 941
942. 942 // Interval validation is a value, not a panic.
943. 943 let bad = client
944. 944 .post(format!("{base}/tasks"))
945. 945 .json(&serde_json::json!({"name":"too hot","prompt":"x","interval_secs":10}))
946. 946 .send()
947. 947 .await
948. 948 .unwrap();
949. 949 assert_eq!(bad.status(), 400);
950. 950
951. 951 let created = client
952. 952 .post(format!("{base}/tasks"))
953. 953 .json(&serde_json::json!({"name":"nightly sweep","prompt":"sweep the repo","interval_secs":3600}))
954. 954 .send()
955. 955 .await
956. 956 .unwrap();
957. 957 assert_eq!(created.status(), 200);
958. 958
959. 959 let listed: serde_json::Value = client
960. 960 .get(format!("{base}/tasks"))
961. 961 .send()
962. 962 .await
963. 963 .unwrap()
964. 964 .json()
965. 965 .await
966. 966 .unwrap();
967. 967 let tasks = listed["tasks"].as_array().unwrap();
968. 968 assert_eq!(tasks.len(), 1);
969. 969 assert_eq!(tasks[0]["name"], "nightly sweep");
970. 970 assert_eq!(tasks[0]["enabled"], true);
971. 971 let tid = tasks[0]["id"].as_str().unwrap().to_string();
972. 972
973. 973 // Run now: child runs in a worktree; transcript completes; task records it.
974. 974 let fired = client
975. 975 .post(format!("{base}/tasks/{tid}/run-now"))
976. 976 .send()
977. 977 .await
978. 978 .unwrap();
979. 979 assert_eq!(fired.status(), 202);
980. 980
981. 981 #[allow(unused_assignments)]
982. 982 let mut child_id = String::new();
983. 983 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
984. 984 loop {
985. 985 assert!(std::time::Instant::now() < deadline, "task run timeout");
986. 986 let t: serde_json::Value = client
987. 987 .get(format!("{base}/tasks"))
988. 988 .send()
989. 989 .await
990. 990 .unwrap()
991. 991 .json()
992. 992 .await
993. 993 .unwrap();
994. 994 let t0 = &t["tasks"][0];
995. 995 if let Some(cid) = t0["last_session_id"].as_str() {
996. 996 child_id = cid.to_string();
997. 997 let _ = &child_id;
998. 998 // The watcher records the run's real final answer, not a
999. 999 // status word (docs/design/22-gateway.md).
1000. 1000 if t0["last_summary"] == "task ran" {