1. 1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2. 2
3. 3use std::collections::VecDeque;
4. 4use std::path::{Path, PathBuf};
5. 5use std::sync::{Arc, Mutex};
6. 6use std::time::Duration;
7. 7
8. 8use tokio_util::sync::CancellationToken;
9. 9
10. 10use vak_core::Core;
11. 11use vak_llm::stream;
12. 12use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, Usage};
13. 13use vak_llm::{EventStream, LlmError, Provider};
14. 14
15. 15struct Scripted {
16. 16 responses: Mutex<VecDeque<AssistantMessage>>,
17. 17}
18. 18
19. 19#[async_trait::async_trait]
20. 20impl Provider for Scripted {
21. 21 fn name(&self) -> &str {
22. 22 "scripted"
23. 23 }
24. 24
25. 25 async fn stream(
26. 26 &self,
27. 27 _request: ChatRequest,
28. 28 _cancel: CancellationToken,
29. 29 ) -> Result<EventStream, LlmError> {
30. 30 let next = self.responses.lock().unwrap().pop_front();
31. 31 let (mut sink, rx) = stream::channel(64);
32. 32 match next {
33. 33 Some(m) => {
34. 34 sink.push(stream::StreamEvent::Start { partial: m.clone() });
35. 35 sink.close_message(m).await;
36. 36 }
37. 37 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
38. 38 }
39. 39 Ok(rx)
40. 40 }
41. 41}
42. 42
43. 43fn text(t: &str) -> AssistantMessage {
44. 44 AssistantMessage {
45. 45 content: vec![ContentBlock::text(t)],
46. 46 stop_reason: vak_llm::types::StopReason::EndTurn,
47. 47 usage: Usage {
48. 48 input_tokens: 7,
49. 49 output_tokens: 3,
50. 50 ..Default::default()
51. 51 },
52. 52 model: "test-model".into(),
53. 53 response_id: None,
54. 54 }
55. 55}
56. 56
57. 57fn tool_call(id: &str, name: &str, input: serde_json::Value) -> AssistantMessage {
58. 58 AssistantMessage {
59. 59 content: vec![ContentBlock::ToolUse {
60. 60 id: id.into(),
61. 61 name: name.into(),
62. 62 input,
63. 63 }],
64. 64 stop_reason: vak_llm::types::StopReason::ToolUse,
65. 65 usage: Usage::default(),
66. 66 model: "test-model".into(),
67. 67 response_id: None,
68. 68 }
69. 69}
70. 70
71. 71struct Gateway {
72. 72 base: String,
73. 73 token: String,
74. 74 home: PathBuf,
75. 75 _server: tokio::task::JoinHandle<()>,
76. 76}
77. 77
78. 78/// Gateway enabled purely through trusted project config. The agent modes
79. 79/// (permission/approval) are pinned in the project config by [`config`] so an
80. 80/// ambient global `fullaccess` profile cannot silently collapse the Ask gate
81. 81/// these forwarded-approval tests assert on, or hang the deny test on a
82. 82/// blocking approver. Project overrides global per the layered-config merge;
83. 83/// note a runtime `set_approval_mode` would be clobbered by the control-plane
84. 84/// refresh, so the modes live in the persisted config instead.
85. 85async fn spawn_with_config(provider: Arc<dyn Provider>, gateway_toml: &str) -> Gateway {
86. 86 let dir = tempfile::tempdir().unwrap();
87. 87 let cwd = dir.path().to_path_buf();
88. 88 let project = cwd.join(".vak");
89. 89 std::fs::create_dir_all(&project).unwrap();
90. 90 std::fs::write(project.join("config.toml"), gateway_toml).unwrap();
91. 91
92. 92 vak_config::paths::isolate_home_for_tests();
93. 93 let core = Core::new_with_trust(cwd.clone(), true).unwrap();
94. 94 let home = dir.path().join("home");
95. 95 core.set_sessions_home(home.clone());
96. 96 core.set_provider_instance(provider);
97. 97 core.set_tool_worker_exe(std::path::PathBuf::from(env!(
98. 98 "CARGO_BIN_EXE_vak-tool-worker"
99. 99 )));
100. 100 std::mem::forget(dir);
101. 101
102. 102 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
103. 103 let addr = listener.local_addr().unwrap();
104. 104 let (app, token) = vak_server::secured_router_with(core, false);
105. 105 let server = tokio::spawn(async move {
106. 106 axum::serve(listener, app).await.unwrap();
107. 107 });
108. 108 Gateway {
109. 109 base: format!("http://{addr}"),
110. 110 token,
111. 111 home,
112. 112 _server: server,
113. 113 }
114. 114}
115. 115
116. 116fn client_with(token: &str) -> reqwest::Client {
117. 117 reqwest::ClientBuilder::new()
118. 118 .default_headers({
119. 119 let mut h = reqwest::header::HeaderMap::new();
120. 120 h.insert(
121. 121 reqwest::header::AUTHORIZATION,
122. 122 format!("Bearer {token}").parse().unwrap(),
123. 123 );
124. 124 h
125. 125 })
126. 126 .build()
127. 127 .unwrap()
128. 128}
129. 129
130. 130fn config(toml_body: &str) -> String {
131. 131 config_with_modes(toml_body, "workspace-write", "ask")
132. 132}
133. 133
134. 134fn config_with_modes(toml_body: &str, permission_mode: &str, approval_mode: &str) -> String {
135. 135 format!(
136. 136 "permission_mode = \"{permission_mode}\"\n\
137. 137 approval_mode = \"{approval_mode}\"\n\
138. 138 [gateway]\n\
139. 139 enabled = true\n\
140. 140 chat_allowlist_open = true\n\
141. 141 {toml_body}\n\n\
142. 142 [memory]\n\
143. 143 reflection = false\n"
144. 144 )
145. 145}
146. 146
147. 147async fn inbound(
148. 148 client: &reqwest::Client,
149. 149 base: &str,
150. 150 surface: &str,
151. 151 chat: &str,
152. 152 text: &str,
153. 153 wait: bool,
154. 154) -> reqwest::Response {
155. 155 client
156. 156 .post(format!("{base}/gateway/inbound"))
157. 157 .json(&serde_json::json!({
158. 158 "surface": surface, "chat": chat, "text": text, "wait": wait
159. 159 }))
160. 160 .send()
161. 161 .await
162. 162 .unwrap()
163. 163}
164. 164
165. 165fn deliveries(home: &Path) -> PathBuf {
166. 166 home.join("gateway/deliveries.jsonl")
167. 167}
168. 168
169. 169async fn wait_for_delivery(home: &Path, needle: &str, secs: u64) -> bool {
170. 170 let deadline = std::time::Instant::now() + Duration::from_secs(secs);
171. 171 while std::time::Instant::now() < deadline {
172. 172 if let Ok(raw) = std::fs::read_to_string(deliveries(home))
173. 173 && raw.contains(needle)
174. 174 {
175. 175 return true;
176. 176 }
177. 177 tokio::time::sleep(Duration::from_millis(150)).await;
178. 178 }
179. 179 false
180. 180}
181. 181
182. 182#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
183. 183async fn forwarded_gate_resolves_from_approver_chat() {
184. 184 let gw = spawn_with_config(
185. 185 Arc::new(Scripted {
186. 186 responses: Mutex::new(VecDeque::from(vec![
187. 187 tool_call(
188. 188 "t1",
189. 189 "bash",
190. 190 serde_json::json!({"command": "echo approved-run"}),
191. 191 ),
192. 192 text("done after yes"),
193. 193 ])),
194. 194 }),
195. 195 &config("approvals = \"forward\"\napprover = \"log:ops\"\n"),
196. 196 )
197. 197 .await;
198. 198 let client = client_with(&gw.token);
199. 199
200. 200 // Kick off an unattended turn that needs escalation.
201. 201 let res = inbound(&client, &gw.base, "webhook", "ci", "run it", false).await;
202. 202 assert_eq!(res.status(), 202);
203. 203
204. 204 // The gate lands on the approver surface's delivery transport.
205. 205 let announced = wait_for_delivery(&gw.home, "Approval requested", 15).await;
206. 206 if !announced {
207. 207 let status: serde_json::Value = client
208. 208 .get(format!("{}/gateway/status", gw.base))
209. 209 .send()
210. 210 .await
211. 211 .unwrap()
212. 212 .json()
213. 213 .await
214. 214 .unwrap();
215. 215 eprintln!("DBG status={status}");
216. 216 if let Ok(raw) = std::fs::read_to_string(deliveries(&gw.home)) {
217. 217 eprintln!("DBG deliveries={raw}");
218. 218 } else {
219. 219 eprintln!("DBG deliveries=<missing> home={:?}", gw.home);
220. 220 }
221. 221 }
222. 222 assert!(announced, "gate must be announced on the approver surface");
223. 223
224. 224 // Answering from the approver chat resolves it.
225. 225 let res = inbound(&client, &gw.base, "log", "ops", "yes", false).await;
226. 226 assert_eq!(res.status(), 200);
227. 227 let body: serde_json::Value = res.json().await.unwrap();
228. 228 assert_eq!(body["state"], "approval_resolved");
229. 229 assert_eq!(body["approved"], true);
230. 230
231. 231 // The tool then really runs and the turn completes.
232. 232 let sid: String = {
233. 233 let status: serde_json::Value = client
234. 234 .get(format!("{}/gateway/status", gw.base))
235. 235 .send()
236. 236 .await
237. 237 .unwrap()
238. 238 .json()
239. 239 .await
240. 240 .unwrap();
241. 241 status["bindings"][0]["session_id"]
242. 242 .as_str()
243. 243 .unwrap()
244. 244 .to_string()
245. 245 };
246. 246 let deadline = std::time::Instant::now() + Duration::from_secs(20);
247. 247 let mut raw = String::new();
248. 248 while std::time::Instant::now() < deadline {
249. 249 if let Ok(res) = client
250. 250 .get(format!("{}/sessions/{sid}/transcript", gw.base))
251. 251 .send()
252. 252 .await
253. 253 && res.status() == reqwest::StatusCode::OK
254. 254 {
255. 255 let t: serde_json::Value = res.json().await.unwrap();
256. 256 if t.get("error").is_none() {
257. 257 raw = serde_json::to_string(&t).unwrap();
258. 258 if raw.contains("approved-run") && raw.contains("done after yes") {
259. 259 break;
260. 260 }
261. 261 }
262. 262 }
263. 263 tokio::time::sleep(Duration::from_millis(150)).await;
264. 264 }
265. 265 assert!(
266. 266 raw.contains("approved-run"),
267. 267 "bash ran after approval: {raw}"
268. 268 );
269. 269 assert!(raw.contains("done after yes"), "turn completed: {raw}");
270. 270
271. 271 // Pending count back to zero.
272. 272 let status: serde_json::Value = client
273. 273 .get(format!("{}/gateway/status", gw.base))
274. 274 .send()
275. 275 .await
276. 276 .unwrap()
277. 277 .json()
278. 278 .await
279. 279 .unwrap();
280. 280 assert_eq!(status["approvals"]["pending"], 0);
281. 281}
282. 282
283. 283#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
284. 284async fn unanswered_gate_times_out_and_fails_closed() {
285. 285 let gw = spawn_with_config(
286. 286 Arc::new(Scripted {
287. 287 responses: Mutex::new(VecDeque::from(vec![
288. 288 tool_call(
289. 289 "t1",
290. 290 "bash",
291. 291 serde_json::json!({"command": "echo never-runs"}),
292. 292 ),
293. 293 text("bash denied: moving on without it"),
294. 294 ])),
295. 295 }),
296. 296 &config("approvals = \"forward\"\napprover = \"log:ops\"\napproval_timeout_secs = 5\n"),
297. 297 )
298. 298 .await;
299. 299 let client = client_with(&gw.token);
300. 300
301. 301 let res = inbound(&client, &gw.base, "webhook", "ci", "go", false).await;
302. 302 assert_eq!(res.status(), 202);
303. 303 assert!(
304. 304 wait_for_delivery(&gw.home, "Approval requested", 15).await,
305. 305 "gate announced"
306. 306 );
307. 307
308. 308 // Let the 5s window lapse; the late reply must resolve nothing.
309. 309 tokio::time::sleep(Duration::from_secs(6)).await;
310. 310 let res = inbound(&client, &gw.base, "log", "ops", "yes", false).await;
311. 311 let body: serde_json::Value = res.json().await.unwrap();
312. 312 assert_eq!(body["state"], "no_pending_approvals");
313. 313
314. 314 // The turn finishes anyway; the tool never executed.
315. 315 let sid: String = {
316. 316 let status: serde_json::Value = client
317. 317 .get(format!("{}/gateway/status", gw.base))
318. 318 .send()
319. 319 .await
320. 320 .unwrap()
321. 321 .json()
322. 322 .await
323. 323 .unwrap();
324. 324 status["bindings"][0]["session_id"]
325. 325 .as_str()
326. 326 .unwrap()
327. 327 .to_string()
328. 328 };
329. 329 let deadline = std::time::Instant::now() + Duration::from_secs(20);
330. 330 let mut raw = String::new();
331. 331 while std::time::Instant::now() < deadline {
332. 332 if let Ok(res) = client
333. 333 .get(format!("{}/sessions/{sid}/transcript", gw.base))
334. 334 .send()
335. 335 .await
336. 336 && res.status() == reqwest::StatusCode::OK
337. 337 {
338. 338 let t: serde_json::Value = res.json().await.unwrap();
339. 339 if t.get("error").is_none() {
340. 340 raw = serde_json::to_string(&t).unwrap();
341. 341 if raw.contains("moving on without it") {
342. 342 break;
343. 343 }
344. 344 }
345. 345 }
346. 346 tokio::time::sleep(Duration::from_millis(150)).await;
347. 347 }
348. 348 assert!(raw.contains("moving on without it"), "run finished: {raw}");
349. 349}
350. 350
351. 351#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
352. 352async fn addressed_yes_resolves_only_that_gate_and_reports_it() {
353. 353 let gw = spawn_with_config(
354. 354 Arc::new(Scripted {
355. 355 responses: Mutex::new(VecDeque::from(vec![
356. 356 tool_call(
357. 357 "t1",
358. 358 "bash",
359. 359 serde_json::json!({"command": "echo addressed-run"}),
360. 360 ),
361. 361 text("done after addressed yes"),
362. 362 ])),
363. 363 }),
364. 364 &config("approvals = \"forward\"\napprover = \"log:ops\"\n"),
365. 365 )
366. 366 .await;
367. 367 let client = client_with(&gw.token);
368. 368
369. 369 let res = inbound(&client, &gw.base, "webhook", "ci", "run it", false).await;
370. 370 assert_eq!(res.status(), 202);
371. 371 assert!(
372. 372 wait_for_delivery(&gw.home, "Approval requested", 15).await,
373. 373 "gate announced"
374. 374 );
375. 375 let raw = std::fs::read_to_string(deliveries(&gw.home)).unwrap();
376. 376 let announcement = raw
377. 377 .lines()
378. 378 .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
379. 379 .find_map(|line| line["text"].as_str().map(str::to_string))
380. 380 .unwrap();
381. 381 let start = announcement.find('[').unwrap() + 1;
382. 382 let short = announcement[start..start + 8].to_string();
383. 383
384. 384 // A verdict addressed to a nonexistent gate must leave the live gate
385. 385 // untouched.
386. 386 let res = inbound(&client, &gw.base, "log", "ops", "yes 00000000", false).await;
387. 387 let body: serde_json::Value = res.json().await.unwrap();
388. 388 assert_eq!(body["state"], "no_pending_approvals");
389. 389 let status: serde_json::Value = client
390. 390 .get(format!("{}/gateway/status", gw.base))
391. 391 .send()
392. 392 .await
393. 393 .unwrap()
394. 394 .json()
395. 395 .await
396. 396 .unwrap();
397. 397 assert_eq!(
398. 398 status["approvals"]["pending"], 1,
399. 399 "live gate survives a mis-addressed yes"
400. 400 );
401. 401
402. 402 // Addressing the real gate resolves exactly it, and the reply says so.
403. 403 let res = inbound(
404. 404 &client,
405. 405 &gw.base,
406. 406 "log",
407. 407 "ops",
408. 408 &format!("yes {short}"),
409. 409 false,
410. 410 )
411. 411 .await;
412. 412 let body: serde_json::Value = res.json().await.unwrap();
413. 413 assert_eq!(body["state"], "approval_resolved");
414. 414 assert_eq!(body["approved"], true);
415. 415 assert!(
416. 416 body["gate"].as_str().unwrap().starts_with(&short),
417. 417 "resolved gate id reported: {body}"
418. 418 );
419. 419 assert_eq!(body["remaining"], 0);
420. 420 assert!(
421. 421 body["session_id"].as_str().is_some_and(|s| !s.is_empty()),
422. 422 "session attribution present"
423. 423 );
424. 424
425. 425 // The tool then really runs.
426. 426 let sid = body["session_id"].as_str().unwrap().to_string();
427. 427 let deadline = std::time::Instant::now() + Duration::from_secs(20);
428. 428 let mut transcript = String::new();
429. 429 while std::time::Instant::now() < deadline {
430. 430 if let Ok(res) = client
431. 431 .get(format!("{}/sessions/{sid}/transcript", gw.base))
432. 432 .send()
433. 433 .await
434. 434 && res.status() == reqwest::StatusCode::OK
435. 435 {
436. 436 let t: serde_json::Value = res.json().await.unwrap();
437. 437 if t.get("error").is_none()
438. 438 && let Ok(raw) = serde_json::to_string(&t)
439. 439 && raw.contains("addressed-run")
440. 440 {
441. 441 transcript = raw;
442. 442 break;
443. 443 }
444. 444 }
445. 445 tokio::time::sleep(Duration::from_millis(150)).await;
446. 446 }
447. 447 assert!(
448. 448 transcript.contains("addressed-run"),
449. 449 "bash ran: {transcript}"
450. 450 );
451. 451}
452. 452
453. 453#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
454. 454async fn default_policy_denies_without_forwarding() {
455. 455 let gw = spawn_with_config(
456. 456 Arc::new(Scripted {
457. 457 // Denial feeds an error result back; the completion guard may
458. 458 // ask for one more pass, so keep a spare response queued.
459. 459 responses: Mutex::new(VecDeque::from(vec![
460. 460 tool_call(
461. 461 "t1",
462. 462 "bash",
463. 463 serde_json::json!({"command": "echo denied-run"}),
464. 464 ),
465. 465 text("skipping that step"),
466. 466 text("skipping that step"),
467. 467 ])),
468. 468 }),
469. 469 &config_with_modes("", "read-only", "ask"), // approvals unset => deny
470. 470 )
471. 471 .await;
472. 472 let client = client_with(&gw.token);
473. 473
474. 474 // No approver configured: forward_mode() must be false even if someone
475. 475 // sends verdict-shaped text to any chat.
476. 476 let res = inbound(&client, &gw.base, "log", "ops", "yes", false).await;
477. 477 let body: serde_json::Value = res.json().await.unwrap();
478. 478 assert_ne!(body["state"], "approval_resolved");
479. 479
480. 480 // The unattended turn completes via auto-deny; nothing was forwarded.
481. 481 let res = inbound(&client, &gw.base, "webhook", "ci", "go", true).await;
482. 482 assert_eq!(res.status(), 200);
483. 483 let body: serde_json::Value = res.json().await.unwrap();
484. 484 assert_eq!(body["text"], "skipping that step");
485. 485 assert!(
486. 486 !deliveries(&gw.home).exists(),
487. 487 "deny mode must not announce anything"
488. 488 );
489. 489}
490. 490