1. 1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2. 2
3. 3//! Request tail assembly (docs/design/68-context-engine.md §6/§10): the
4. 4//! per-turn control block lands as the final text block of the last user
5. 5//! message, stays byte-identical across every step of one turn, carries
6. 6//! cache breakpoints at the three documented positions, and a change in the
7. 7//! stable prefix is surfaced as a `prefix-changed` activity exactly when the
8. 8//! digest actually changes.
9. 9
10. 10use std::collections::VecDeque;
11. 11use std::sync::{Arc, Mutex};
12. 12
13. 13use tempfile::tempdir;
14. 14use tokio::sync::mpsc;
15. 15use tokio_util::sync::CancellationToken;
16. 16
17. 17use vak_agent::{Agent, AgentConfig, AutoApprove, SteeringQueues, TailInput, TurnOutcome};
18. 18use vak_llm::stream;
19. 19use vak_llm::types::{
20. 20 AssistantMessage, ChatRequest, ContentBlock, Message, Role, StopReason, Usage,
21. 21};
22. 22use vak_llm::{EventStream, LlmError, Provider};
23. 23use vak_permission::{Mode, PermissionEngine};
24. 24use vak_session::types::{FrozenContract, MessageRecord, SessionHeader};
25. 25use vak_session::{SessionLog, SessionPath};
26. 26use vak_tools::bash::BashTool;
27. 27
28. 28/// Records every request it receives (for tail/cache inspection) and
29. 29/// replays a fixed queue of responses.
30. 30struct Recording {
31. 31 requests: Mutex<Vec<ChatRequest>>,
32. 32 responses: Mutex<VecDeque<AssistantMessage>>,
33. 33}
34. 34
35. 35impl Recording {
36. 36 fn new(responses: Vec<AssistantMessage>) -> Self {
37. 37 Recording {
38. 38 requests: Mutex::new(Vec::new()),
39. 39 responses: Mutex::new(responses.into_iter().collect()),
40. 40 }
41. 41 }
42. 42
43. 43 fn requests(&self) -> Vec<ChatRequest> {
44. 44 self.requests.lock().unwrap().clone()
45. 45 }
46. 46}
47. 47
48. 48#[async_trait::async_trait]
49. 49impl Provider for Recording {
50. 50 fn name(&self) -> &str {
51. 51 "recording"
52. 52 }
53. 53
54. 54 async fn stream(
55. 55 &self,
56. 56 request: ChatRequest,
57. 57 _cancel: CancellationToken,
58. 58 ) -> Result<EventStream, LlmError> {
59. 59 self.requests.lock().unwrap().push(request);
60. 60 let next = self.responses.lock().unwrap().pop_front();
61. 61 let (mut sink, rx) = stream::channel(64);
62. 62 match next {
63. 63 Some(m) => {
64. 64 sink.push(stream::StreamEvent::Start { partial: m.clone() });
65. 65 sink.close_message(m).await;
66. 66 }
67. 67 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
68. 68 }
69. 69 Ok(rx)
70. 70 }
71. 71}
72. 72
73. 73fn bash_call(id: &str, cmd: &str) -> AssistantMessage {
74. 74 AssistantMessage {
75. 75 content: vec![ContentBlock::ToolUse {
76. 76 id: id.into(),
77. 77 name: "bash".into(),
78. 78 input: serde_json::json!({"command": cmd}),
79. 79 }],
80. 80 stop_reason: StopReason::ToolUse,
81. 81 usage: Usage::default(),
82. 82 model: "test-model".into(),
83. 83 response_id: None,
84. 84 }
85. 85}
86. 86
87. 87fn text_msg(t: &str) -> AssistantMessage {
88. 88 AssistantMessage {
89. 89 content: vec![ContentBlock::text(t)],
90. 90 stop_reason: StopReason::EndTurn,
91. 91 usage: Usage {
92. 92 input_tokens: 50,
93. 93 output_tokens: 3,
94. 94 ..Default::default()
95. 95 },
96. 96 model: "test-model".into(),
97. 97 response_id: None,
98. 98 }
99. 99}
100. 100
101. 101fn header(cwd: &std::path::Path) -> SessionHeader {
102. 102 SessionHeader {
103. 103 agent: None,
104. 104 session_id: "tail-test".into(),
105. 105 created_at: chrono::Utc::now(),
106. 106 cwd: cwd.to_path_buf(),
107. 107 parent_session_id: None,
108. 108 contract_id: None,
109. 109 work_item_id: None,
110. 110 conversation: None,
111. 111 contract: FrozenContract {
112. 112 app_version: "0".into(),
113. 113 provider: "recording".into(),
114. 114 model: "test-model".into(),
115. 115 route_ladder: Vec::new(),
116. 116 route_objective: String::new(),
117. 117 route_annotations: Vec::new(),
118. 118 system_prompt: "sys".into(),
119. 119 permission_mode: "workspace-write".into(),
120. 120 capabilities: Vec::new(),
121. 121 prompt_layers: Vec::new(),
122. 122 },
123. 123 }
124. 124}
125. 125
126. 126fn build_agent(provider: Arc<Recording>, dir: &std::path::Path) -> Agent {
127. 127 let home = dir.join(".vak-home");
128. 128 std::fs::create_dir_all(&home).unwrap();
129. 129 let path = SessionPath::new_session_file(&home, dir, "tail-test");
130. 130 let log = SessionLog::create(path, header(dir)).unwrap();
131. 131 let mut cfg = AgentConfig::new("sys");
132. 132 cfg.model = "test-model".into();
133. 133 cfg.mode = Mode::FullAccess;
134. 134 cfg.permission = Some(Arc::new(PermissionEngine::default()));
135. 135 cfg.approver = Some(Arc::new(AutoApprove));
136. 136 cfg.tools = vec![Arc::new(BashTool)];
137. 137 cfg.tool_definitions = Some(vak_tools::definitions(&cfg.tools));
138. 138 cfg.retry_base_backoff_ms = 1;
139. 139 cfg.run_retry_base_backoff_ms = 1;
140. 140 cfg.tail = TailInput {
141. 141 temporal: "current UTC instant 2026-09-19T00:00:00Z".into(),
142. 142 stance: "Provide a clear, direct answer.".into(),
143. 143 };
144. 144 Agent::new(provider, log, cfg)
145. 145}
146. 146
147. 147async fn run(agent: &mut Agent, prompt: &str) -> TurnOutcome {
148. 148 let (ev_tx, mut ev_rx) = mpsc::channel(256);
149. 149 tokio::spawn(async move { while ev_rx.recv().await.is_some() {} });
150. 150 let cancel = CancellationToken::new();
151. 151 let steering = SteeringQueues::new();
152. 152 agent.run(prompt, &steering, cancel, ev_tx).await
153. 153}
154. 154
155. 155/// The tail is appended as a final text block on the turn's DIRECTIVE
156. 156/// message — never re-homed onto whatever message a later step happens to
157. 157/// end with — so the directive (tail included) stays byte-identical across
158. 158/// every step of the same turn: an append-only request
159. 159/// (docs/design/68-context-engine.md §6/§7).
160. 160#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
161. 161async fn tail_precedes_the_users_words_and_is_stable_within_a_turn() {
162. 162 let dir = tempdir().unwrap();
163. 163 let provider = Arc::new(Recording::new(vec![
164. 164 bash_call("call-1", "echo hi"),
165. 165 text_msg("done"),
166. 166 ]));
167. 167 let mut agent = build_agent(provider.clone(), dir.path());
168. 168 let outcome = run(&mut agent, "run the check").await;
169. 169 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
170. 170
171. 171 let requests = provider.requests();
172. 172 assert_eq!(requests.len(), 2, "one step per response");
173. 173
174. 174 // Step 1: only the directive exists yet, so it is also the last (and
175. 175 // only) message; the tail sits BEFORE it so the user's own words are
176. 176 // the last thing the model reads (docs/design/68-context-engine.md
177. 177 // §6), and there is no echo because the directive itself follows.
178. 178 let step1_directive = requests[0].messages.last().expect("a message").clone();
179. 179 assert_eq!(step1_directive.role, Role::User);
180. 180 let texts: Vec<&str> = step1_directive
181. 181 .content
182. 182 .iter()
183. 183 .filter_map(|b| match b {
184. 184 ContentBlock::Text { text } => Some(text.as_str()),
185. 185 _ => None,
186. 186 })
187. 187 .collect();
188. 188 assert_eq!(texts.len(), 2, "tail block then directive: {texts:?}");
189. 189 let tail_step_1 = texts[0].to_string();
190. 190 assert!(tail_step_1.contains("<turn_context>"));
191. 191 assert!(tail_step_1.contains("current UTC instant 2026-09-19T00:00:00Z"));
192. 192 assert!(tail_step_1.contains("<stance>"));
193. 193 assert!(tail_step_1.contains("Provide a clear, direct answer."));
194. 194 assert!(!tail_step_1.contains("<directive>"));
195. 195 assert_eq!(texts[1], "run the check");
196. 196
197. 197 // Step 2: the tool call and its result are appended AFTER the
198. 198 // directive. The directive -- tail included -- is byte-identical to
199. 199 // step 1's own first message, and the tail is never re-attached to the
200. 200 // trailing tool-result message.
201. 201 assert_eq!(
202. 202 requests[1].messages[0], step1_directive,
203. 203 "the directive (with its tail) must be byte-identical across steps"
204. 204 );
205. 205 let step2_last = requests[1].messages.last().unwrap();
206. 206 assert!(matches!(
207. 207 step2_last.content.first(),
208. 208 Some(ContentBlock::ToolResult { .. })
209. 209 ));
210. 210 assert!(
211. 211 !step2_last
212. 212 .content
213. 213 .iter()
214. 214 .any(|b| matches!(b, ContentBlock::Text { .. })),
215. 215 "the tail must not be re-homed onto the tool-result message: {:?}",
216. 216 step2_last.content
217. 217 );
218. 218}
219. 219
220. 220/// Append-only requests within a turn (docs/design/68-context-engine.md
221. 221/// §7): across every step of one multi-step tool turn, a later request's
222. 222/// `system`, `tools`, and its messages up to the length of an earlier
223. 223/// request are byte-identical to that earlier request — nothing already
224. 224/// sent ever silently changes shape underneath a replayed thinking block.
225. 225#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
226. 226async fn later_requests_in_a_turn_are_byte_identical_prefixes_of_earlier_ones() {
227. 227 let dir = tempdir().unwrap();
228. 228 let provider = Arc::new(Recording::new(vec![
229. 229 bash_call("call-1", "echo one"),
230. 230 bash_call("call-2", "echo two"),
231. 231 bash_call("call-3", "echo three"),
232. 232 text_msg("done"),
233. 233 ]));
234. 234 let mut agent = build_agent(provider.clone(), dir.path());
235. 235 let outcome = run(&mut agent, "run three checks").await;
236. 236 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
237. 237
238. 238 let requests = provider.requests();
239. 239 assert_eq!(requests.len(), 4, "one step per response");
240. 240 for k in 0..requests.len() - 1 {
241. 241 assert_eq!(
242. 242 requests[k + 1].system,
243. 243 requests[k].system,
244. 244 "system prompt must not change mid-turn (step {k} -> {})",
245. 245 k + 1
246. 246 );
247. 247 assert_eq!(
248. 248 requests[k + 1].tools,
249. 249 requests[k].tools,
250. 250 "tools array must not change mid-turn (step {k} -> {})",
251. 251 k + 1
252. 252 );
253. 253 assert_eq!(
254. 254 requests[k + 1].messages[..requests[k].messages.len()],
255. 255 requests[k].messages[..],
256. 256 "request {}'s messages must carry request {k}'s as a byte-identical prefix",
257. 257 k + 1
258. 258 );
259. 259 }
260. 260}
261. 261
262. 262/// Cache breakpoints land after the stable prefix, after the last message of
263. 263/// any previous turn, and on the last message of the request being built.
264. 264#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
265. 265async fn cache_breakpoints_mark_prefix_previous_turn_and_current_step() {
266. 266 let dir = tempdir().unwrap();
267. 267 let provider = Arc::new(Recording::new(vec![text_msg("first answer")]));
268. 268 let mut agent = build_agent(provider.clone(), dir.path());
269. 269 let outcome = run(&mut agent, "first question").await;
270. 270 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
271. 271
272. 272 // First-ever turn: no earlier turn exists, so only two positions apply
273. 273 // (after the prefix, and on the last — and only — message).
274. 274 let first_turn_requests = provider.requests();
275. 275 let breakpoints = first_turn_requests[0]
276. 276 .cache
277. 277 .as_ref()
278. 278 .expect("cache hints present")
279. 279 .breakpoints
280. 280 .clone();
281. 281 assert_eq!(
282. 282 breakpoints
283. 283 .iter()
284. 284 .map(|b| b.after_message)
285. 285 .collect::<Vec<_>>(),
286. 286 vec![None, Some(0)]
287. 287 );
288. 288
289. 289 provider
290. 290 .responses
291. 291 .lock()
292. 292 .unwrap()
293. 293 .push_back(text_msg("second answer"));
294. 294 let outcome = run(&mut agent, "second question").await;
295. 295 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
296. 296
297. 297 let all_requests = provider.requests();
298. 298 let second_turn_request = all_requests.last().expect("second turn request");
299. 299 // messages: [user1, assistant1, user2] — three positions now apply.
300. 300 assert_eq!(second_turn_request.messages.len(), 3);
301. 301 let hints = second_turn_request
302. 302 .cache
303. 303 .as_ref()
304. 304 .expect("cache hints present");
305. 305 assert!(!hints.session_key.is_empty());
306. 306 assert_eq!(
307. 307 hints
308. 308 .breakpoints
309. 309 .iter()
310. 310 .map(|b| b.after_message)
311. 311 .collect::<Vec<_>>(),
312. 312 vec![None, Some(1), Some(2)],
313. 313 "prefix, end of previous turn, and the current step's last message"
314. 314 );
315. 315}
316. 316
317. 317/// A digest that differs from the previous receipt's is surfaced as a
318. 318/// `prefix-changed` Activity; an unchanged digest across turns writes none.
319. 319#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
320. 320async fn prefix_changed_activity_recorded_only_when_digest_changes() {
321. 321 let dir = tempdir().unwrap();
322. 322 let provider = Arc::new(Recording::new(vec![
323. 323 text_msg("first"),
324. 324 text_msg("second"),
325. 325 text_msg("third"),
326. 326 ]));
327. 327 let mut agent = build_agent(provider.clone(), dir.path());
328. 328
329. 329 // Turn 1: first digest ever seen — nothing to compare against yet.
330. 330 assert!(matches!(
331. 331 run(&mut agent, "one").await,
332. 332 TurnOutcome::Completed { .. }
333. 333 ));
334. 334 // Turn 2: same prefix and tools — digest unchanged.
335. 335 assert!(matches!(
336. 336 run(&mut agent, "two").await,
337. 337 TurnOutcome::Completed { .. }
338. 338 ));
339. 339 {
340. 340 let session = agent.session.lock().await;
341. 341 assert!(
342. 342 session
343. 343 .activities()
344. 344 .iter()
345. 345 .all(|(_, _, activity)| activity.label != "prefix-changed"),
346. 346 "an unchanged prefix must never write a prefix-changed activity"
347. 347 );
348. 348 }
349. 349
350. 350 // Turn 3: the prefix changes — this must be the one and only regression
351. 351 // surfaced in the ledger.
352. 352 agent.config.system_prefix = "a different system prefix".into();
353. 353 assert!(matches!(
354. 354 run(&mut agent, "three").await,
355. 355 TurnOutcome::Completed { .. }
356. 356 ));
357. 357
358. 358 let session = agent.into_session().await;
359. 359 let prefix_changed: Vec<_> = session
360. 360 .activities()
361. 361 .into_iter()
362. 362 .filter(|(_, _, activity)| activity.label == "prefix-changed")
363. 363 .collect();
364. 364 assert_eq!(
365. 365 prefix_changed.len(),
366. 366 1,
367. 367 "exactly one prefix change happened"
368. 368 );
369. 369 let (_, _, activity) = &prefix_changed[0];
370. 370 assert!(activity.data.contains_key("previous"));
371. 371 assert!(activity.data.contains_key("current"));
372. 372 assert_ne!(activity.data["previous"], activity.data["current"]);
373. 373
374. 374 // Every receipt still carries a digest, and only the third run's usage
375. 375 // measured a fresh `prefix_tokens` (the first two share one digest).
376. 376 let receipts = session.receipts();
377. 377 assert_eq!(receipts.len(), 3);
378. 378 assert!(receipts.iter().all(|r| !r.prefix_digest.is_empty()));
379. 379 assert!(
380. 380 receipts[0].prefix_tokens.is_some(),
381. 381 "the first request with a digest measures prefix_tokens"
382. 382 );
383. 383 assert!(
384. 384 receipts[1].prefix_tokens.is_none(),
385. 385 "the digest repeats, so it is not re-measured"
386. 386 );
387. 387 assert!(
388. 388 receipts[2].prefix_tokens.is_some(),
389. 389 "a new digest is measured again"
390. 390 );
391. 391}
392. 392
393. 393/// The conversation thread in the assembled request lists only directives
394. 394/// the projection leaves out (docs/design/68-context-engine.md §6/§10): a
395. 395/// directive hidden behind a reset-with-handoff resurfaces there, but one
396. 396/// still present — including the turn's own new directive — is never
397. 397/// repeated. (The plan-driven case, a packeted turn, is covered at the
398. 398/// session level: `tail_sections(Some(&plan))` in vak-session's tests.)
399. 399#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
400. 400async fn thread_in_the_assembled_request_lists_only_non_verbatim_directives() {
401. 401 let dir = tempdir().unwrap();
402. 402 let home = dir.path().join(".vak-home");
403. 403 std::fs::create_dir_all(&home).unwrap();
404. 404 let path = SessionPath::new_session_file(&home, dir.path(), "tail-test");
405. 405 let mut log = SessionLog::create(path, header(dir.path())).unwrap();
406. 406
407. 407 log.append_goal_update(vak_intent::GoalUpdate {
408. 408 revision: 1,
409. 409 relation: vak_intent::GoalRelation::New,
410. 410 request: "research on WEF".into(),
411. 411 supersedes_revision: None,
412. 412 explicit: false,
413. 413 })
414. 414 .unwrap();
415. 415 log.append_message(MessageRecord {
416. 416 message: Message::user_text("research on WEF"),
417. 417 meta: None,
418. 418 })
419. 419 .unwrap();
420. 420 log.append_message(MessageRecord {
421. 421 message: Message::assistant(vec![ContentBlock::text("here are findings")]),
422. 422 meta: None,
423. 423 })
424. 424 .unwrap();
425. 425 // Reset-with-handoff after the first turn: everything before it is
426. 426 // invisible to the model; the second turn, appended after, stays
427. 427 // verbatim.
428. 428 log.append_handoff_reset("summary of the WEF research turn".into(), 999)
429. 429 .unwrap();
430. 430 log.append_goal_update(vak_intent::GoalUpdate {
431. 431 revision: 2,
432. 432 relation: vak_intent::GoalRelation::AddsTo,
433. 433 request: "use python sandbox".into(),
434. 434 supersedes_revision: None,
435. 435 explicit: false,
436. 436 })
437. 437 .unwrap();
438. 438 log.append_message(MessageRecord {
439. 439 message: Message::user_text("use python sandbox"),
440. 440 meta: None,
441. 441 })
442. 442 .unwrap();
443. 443 log.append_message(MessageRecord {
444. 444 message: Message::assistant(vec![ContentBlock::text("running in sandbox")]),
445. 445 meta: None,
446. 446 })
447. 447 .unwrap();
448. 448
449. 449 let provider = Arc::new(Recording::new(vec![text_msg(
450. 450 "the global economy grew modestly",
451. 451 )]));
452. 452 let mut cfg = AgentConfig::new("sys");
453. 453 cfg.model = "test-model".into();
454. 454 cfg.mode = Mode::FullAccess;
455. 455 cfg.permission = Some(Arc::new(PermissionEngine::default()));
456. 456 cfg.approver = Some(Arc::new(AutoApprove));
457. 457 cfg.retry_base_backoff_ms = 1;
458. 458 cfg.run_retry_base_backoff_ms = 1;
459. 459 let mut agent = Agent::new(provider.clone(), log, cfg);
460. 460
461. 461 let outcome = run(&mut agent, "now evaluate global GDP past 5 years").await;
462. 462 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
463. 463
464. 464 let requests = provider.requests();
465. 465 assert_eq!(requests.len(), 1);
466. 466 let last = requests[0].messages.last().unwrap();
467. 467 // The tail is the text block before the directive (§6).
468. 468 let tail = last
469. 469 .content
470. 470 .iter()
471. 471 .filter_map(|b| match b {
472. 472 ContentBlock::Text { text } => Some(text.clone()),
473. 473 _ => None,
474. 474 })
475. 475 .find(|text| text.contains("<conversation_thread"))
476. 476 .expect("thread section present for a reset-hidden directive");
477. 477
478. 478 let thread_start = tail.find("<conversation_thread").unwrap_or(0);
479. 479 let thread_text = &tail[thread_start..];
480. 480 assert!(
481. 481 thread_text.contains("research on WEF"),
482. 482 "the reset-hidden directive must resurface: {thread_text}"
483. 483 );
484. 484 assert!(
485. 485 !thread_text.contains("use python sandbox"),
486. 486 "a directive still verbatim in the working set must not repeat: {thread_text}"
487. 487 );
488. 488 assert!(
489. 489 !thread_text.contains("now evaluate global GDP"),
490. 490 "the turn's own directive is already in `messages` and must not repeat: {thread_text}"
491. 491 );
492. 492}
493. 493