1. 1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2. 2
3. 3//! Goal mode + audited completion + regression obligations +
4. 4//! reset-with-handoff (docs/design/42-managed-work-contracts.md).
5. 5
6. 6use std::collections::VecDeque;
7. 7use std::sync::Arc;
8. 8
9. 9use tempfile::tempdir;
10. 10use tokio::sync::mpsc;
11. 11use tokio_util::sync::CancellationToken;
12. 12
13. 13use vak_agent::{
14. 14 Agent, AgentConfig, AutoApprove, SteeringQueues, TurnOutcome, workspace::WorkspaceDelta,
15. 15};
16. 16use vak_llm::stream;
17. 17use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
18. 18use vak_llm::{EventStream, LlmError, Provider};
19. 19use vak_permission::{Mode, PermissionEngine};
20. 20use vak_session::types::{EntryPayload, FrozenContract, SessionHeader};
21. 21use vak_session::{SessionLog, SessionPath};
22. 22use vak_tools::bash::BashTool;
23. 23
24. 24fn text_msg(t: &str) -> AssistantMessage {
25. 25 AssistantMessage {
26. 26 content: vec![ContentBlock::text(t)],
27. 27 stop_reason: StopReason::EndTurn,
28. 28 usage: Usage {
29. 29 input_tokens: 5,
30. 30 output_tokens: 3,
31. 31 ..Default::default()
32. 32 },
33. 33 model: "judge-model".into(),
34. 34 response_id: None,
35. 35 }
36. 36}
37. 37
38. 38fn tool_msg() -> AssistantMessage {
39. 39 AssistantMessage {
40. 40 content: vec![ContentBlock::ToolUse {
41. 41 id: "t1".into(),
42. 42 name: "bash".into(),
43. 43 input: serde_json::json!({"command": "echo green"}),
44. 44 }],
45. 45 stop_reason: StopReason::ToolUse,
46. 46 usage: Usage::default(),
47. 47 model: "test-model".into(),
48. 48 response_id: None,
49. 49 }
50. 50}
51. 51
52. 52/// Scripted responses consumed FIFO; empty queue => parse error.
53. 53struct Scripted {
54. 54 responses: std::sync::Mutex<VecDeque<AssistantMessage>>,
55. 55 requests: std::sync::Mutex<Vec<ChatRequest>>,
56. 56}
57. 57
58. 58impl Scripted {
59. 59 fn new(responses: Vec<AssistantMessage>) -> Arc<Self> {
60. 60 Arc::new(Scripted {
61. 61 responses: std::sync::Mutex::new(responses.into_iter().collect()),
62. 62 requests: std::sync::Mutex::new(Vec::new()),
63. 63 })
64. 64 }
65. 65}
66. 66
67. 67#[async_trait::async_trait]
68. 68impl Provider for Scripted {
69. 69 fn name(&self) -> &str {
70. 70 "scripted-goal"
71. 71 }
72. 72
73. 73 async fn stream(
74. 74 &self,
75. 75 request: ChatRequest,
76. 76 _cancel: CancellationToken,
77. 77 ) -> Result<EventStream, LlmError> {
78. 78 self.requests.lock().unwrap().push(request);
79. 79 let next = self.responses.lock().unwrap().pop_front();
80. 80 let (mut sink, rx) = stream::channel(64);
81. 81 match next {
82. 82 Some(m) => {
83. 83 sink.push(stream::StreamEvent::Start { partial: m.clone() });
84. 84 sink.close_message(m).await;
85. 85 }
86. 86 None => {
87. 87 sink.close_error(LlmError::Parse("script exhausted".into()))
88. 88 .await
89. 89 }
90. 90 }
91. 91 Ok(rx)
92. 92 }
93. 93}
94. 94
95. 95fn setup(
96. 96 provider: Arc<Scripted>,
97. 97 tools: Vec<Arc<dyn vak_tools::Tool>>,
98. 98) -> (Agent, tempfile::TempDir) {
99. 99 setup_with_delta(provider, tools, None)
100. 100}
101. 101
102. 102fn setup_with_delta(
103. 103 provider: Arc<Scripted>,
104. 104 tools: Vec<Arc<dyn vak_tools::Tool>>,
105. 105 delta: Option<Arc<dyn WorkspaceDelta>>,
106. 106) -> (Agent, tempfile::TempDir) {
107. 107 let dir = tempdir().unwrap();
108. 108 let cwd = dir.path().to_path_buf();
109. 109 let header = SessionHeader {
110. 110 agent: None,
111. 111 session_id: "goal".into(),
112. 112 created_at: chrono::Utc::now(),
113. 113 cwd: cwd.clone(),
114. 114 parent_session_id: None,
115. 115 contract_id: None,
116. 116 work_item_id: None,
117. 117 conversation: None,
118. 118 contract: FrozenContract {
119. 119 app_version: "0".into(),
120. 120 provider: "scripted-goal".into(),
121. 121 model: "test-model".into(),
122. 122 route_ladder: Vec::new(),
123. 123 route_objective: String::new(),
124. 124 route_annotations: Vec::new(),
125. 125 system_prompt: "sys".into(),
126. 126 permission_mode: "full-access".into(),
127. 127 capabilities: Vec::new(),
128. 128 prompt_layers: Vec::new(),
129. 129 },
130. 130 };
131. 131 let home = cwd.join(".vak-home");
132. 132 std::fs::create_dir_all(&home).unwrap();
133. 133 let log =
134. 134 SessionLog::create(SessionPath::new_session_file(&home, &cwd, "goal"), header).unwrap();
135. 135 let mut cfg = AgentConfig::new("sys");
136. 136 cfg.model = "test-model".into();
137. 137 cfg.tools = tools;
138. 138 cfg.mode = Mode::FullAccess;
139. 139 cfg.permission = Some(Arc::new(PermissionEngine::default()));
140. 140 cfg.approver = Some(Arc::new(AutoApprove));
141. 141 cfg.workspace_delta = delta;
142. 142 cfg.retry_base_backoff_ms = 1;
143. 143 cfg.run_retry_base_backoff_ms = 1;
144. 144 (Agent::new(provider, log, cfg), dir)
145. 145}
146. 146
147. 147async fn run(agent: &mut Agent, prompt: &str) -> TurnOutcome {
148. 148 let (ev_tx, mut ev_rx) = mpsc::channel(512);
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. 155fn goal_statuses(session: &SessionLog) -> Vec<String> {
156. 156 session
157. 157 .chain_to_root()
158. 158 .iter()
159. 159 .filter_map(|e| match &e.payload {
160. 160 EntryPayload::Goal(g) => Some(match &g.status {
161. 161 vak_session::types::GoalStatus::Active => "active".into(),
162. 162 vak_session::types::GoalStatus::Done { .. } => "done".into(),
163. 163 vak_session::types::GoalStatus::Unverified { .. } => "unverified".into(),
164. 164 }),
165. 165 _ => None,
166. 166 })
167. 167 .collect()
168. 168}
169. 169
170. 170/// Model claims done immediately; judge passes all criteria.
171. 171#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
172. 172async fn audited_done_when_judge_passes() {
173. 173 let provider = Scripted::new(vec![
174. 174 text_msg("haiku written"),
175. 175 text_msg(
176. 176 r#"{"results":[{"criterion":"summary mentions done","verdict":"pass","evidence":"said it"}]}"#,
177. 177 ),
178. 178 ]);
179. 179 struct StaticDelta;
180. 180 impl WorkspaceDelta for StaticDelta {
181. 181 fn summary(&self) -> Result<String, String> {
182. 182 Ok("M src/lib.rs\nA goal-live.txt\nWORKSPACE-DELTA-MARKER".into())
183. 183 }
184. 184 }
185. 185 let (mut agent, _dir) = setup_with_delta(provider.clone(), vec![], Some(Arc::new(StaticDelta)));
186. 186 agent.set_goal("write a haiku", vec!["summary mentions done".into()]);
187. 187 let outcome = run(&mut agent, "write a haiku about rust").await;
188. 188 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
189. 189
190. 190 // The judge call happened and was receipted as Verify work.
191. 191 let has_judge = provider.requests.lock().unwrap().iter().any(|r| {
192. 192 r.system
193. 193 .as_deref()
194. 194 .unwrap_or("")
195. 195 .contains("completion auditor")
196. 196 });
197. 197 assert!(has_judge, "judge call must carry the auditor system prompt");
198. 198 let session = agent.into_session().await;
199. 199 assert_eq!(goal_statuses(&session), vec!["active", "done"]);
200. 200 assert!(
201. 201 session
202. 202 .receipts()
203. 203 .iter()
204. 204 .any(|r| r.purpose == vak_llm::WorkPurpose::Verify),
205. 205 "audit dispatch must be receipted"
206. 206 );
207. 207}
208. 208
209. 209/// Judge fails a criterion first; findings reach the model; second claim
210. 210/// passes. Audit budget respected.
211. 211#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
212. 212async fn rejected_claim_returns_findings_then_passes() {
213. 213 let provider = Scripted::new(vec![
214. 214 text_msg("attempt one"),
215. 215 text_msg(
216. 216 r#"{"results":[{"criterion":"c1","verdict":"fail","evidence":"no file written"}]}"#,
217. 217 ),
218. 218 text_msg("attempt two, file written"),
219. 219 text_msg(r#"{"results":[{"criterion":"c1","verdict":"pass","evidence":"wrote it"}]}"#),
220. 220 ]);
221. 221 let (mut agent, _dir) = setup(provider.clone(), vec![]);
222. 222 agent.set_goal("create x", vec!["c1".into()]);
223. 223 let outcome = run(&mut agent, "create x").await;
224. 224 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
225. 225 let session = agent.into_session().await;
226. 226 let statuses = goal_statuses(&session);
227. 227 assert_eq!(statuses, vec!["active", "done"]);
228. 228 // Findings were injected model-visible (invariant 1): a [goal-audit]
229. 229 // user turn exists between claims. Raw ledger: the nudge is mid-turn
230. 230 // scaffolding and is dropped from the projection once the turn closes
231. 231 // (docs/design/68-context-engine.md §10) — this checks it was
232. 232 // recorded (and thus reached the model) at all.
233. 233 assert!(
234. 234 session
235. 235 .message_chain()
236. 236 .iter()
237. 237 .any(|(_, m)| m.text_content().contains("[goal-audit]"))
238. 238 );
239. 239 let judge_calls = provider
240. 240 .requests
241. 241 .lock()
242. 242 .unwrap()
243. 243 .iter()
244. 244 .filter(|r| {
245. 245 r.system
246. 246 .as_deref()
247. 247 .unwrap_or("")
248. 248 .contains("completion auditor")
249. 249 })
250. 250 .count();
251. 251 assert_eq!(judge_calls, 2, "two judge dispatches");
252. 252}
253. 253
254. 254/// A verify: criterion that fails rejects without any judge call.
255. 255#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
256. 256async fn shell_criterion_failure_blocks_without_judge() {
257. 257 let provider = Scripted::new(vec![text_msg("i ran things")]);
258. 258 let (mut agent, _dir) = setup(provider.clone(), vec![]);
259. 259 agent.config.max_audit_blocks = 0;
260. 260 agent.set_goal("run things", vec!["verify: exit 3".into()]);
261. 261 let outcome = run(&mut agent, "run things").await;
262. 262 assert!(
263. 263 matches!(outcome, TurnOutcome::Completed { .. }),
264. 264 "budget exhausts => unverified completion"
265. 265 );
266. 266 let judge_calls = provider
267. 267 .requests
268. 268 .lock()
269. 269 .unwrap()
270. 270 .iter()
271. 271 .filter(|r| {
272. 272 r.system
273. 273 .as_deref()
274. 274 .unwrap_or("")
275. 275 .contains("completion auditor")
276. 276 })
277. 277 .count();
278. 278 assert_eq!(judge_calls, 0, "deterministic failure needs no judge");
279. 279
280. 280 let session = agent.into_session().await;
281. 281 assert_eq!(goal_statuses(&session), vec!["active", "unverified"]);
282. 282}
283. 283
284. 284/// Green bash commands become obligations: a later failing re-run blocks
285. 285/// the claim until fixed.
286. 286#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
287. 287async fn regression_obligation_blocks_claim() {
288. 288 // Provider script: run echo-green (tool call), claim done, then after
289. 289 // rejection claim done again. Judge always passes.
290. 290 let provider = Scripted::new(vec![
291. 291 tool_msg(),
292. 292 text_msg("all done"),
293. 293 text_msg(r#"{"results":[{"criterion":"c1","verdict":"pass","evidence":"ok"}]}"#),
294. 294 ]);
295. 295 let (mut agent, _dir) = setup(provider, vec![Arc::new(BashTool)]);
296. 296 agent.set_goal("keep tests green", vec!["c1".into()]);
297. 297 let outcome = run(&mut agent, "echo something then finish").await;
298. 298 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
299. 299
300. 300 // The audit re-ran `echo green` as an obligation — visible as an extra
301. 301 // brokered request beyond the scripted model turns.
302. 302 let session = agent.into_session().await;
303. 303 let statuses = goal_statuses(&session);
304. 304 assert!(
305. 305 statuses
306. 306 .last()
307. 307 .map(|s| s == "done" || s == "unverified")
308. 308 .unwrap_or(false),
309. 309 "final status recorded: {statuses:?}"
310. 310 );
311. 311}
312. 312
313. 313/// Unparseable judge output fails closed: claim rejected, not accepted.
314. 314#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
315. 315async fn unparseable_judge_fails_closed() {
316. 316 let provider = Scripted::new(vec![
317. 317 text_msg("attempt one"),
318. 318 text_msg("I think it's probably fine honestly"),
319. 319 text_msg("attempt two"),
320. 320 text_msg(r#"{"results":[{"criterion":"c1","verdict":"pass","evidence":"now clear"}]}"#),
321. 321 ]);
322. 322 let (mut agent, _dir) = setup(provider, vec![]);
323. 323 agent.set_goal("do it", vec!["c1".into()]);
324. 324 let outcome = run(&mut agent, "do it").await;
325. 325 assert!(matches!(outcome, TurnOutcome::Completed { .. }));
326. 326 let session = agent.into_session().await;
327. 327 // Raw ledger: see the comment above on the same pattern.
328. 328 let msgs = session.message_chain();
329. 329 assert!(
330. 330 msgs.iter()
331. 331 .any(|(_, m)| m.text_content().contains("[goal-audit]")
332. 332 && m.text_content().contains("AUDIT UNAVAILABLE")),
333. 333 "unparseable verdict must inject fail-closed findings"
334. 334 );
335. 335}
336. 336
337. 337/// Handoff-reset: with a tiny window, the still-over path writes a handoff
338. 338/// and continues fresh instead of failing.
339. 339#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
340. 340async fn handoff_reset_rescues_still_over_context() {
341. 341 // Script: handoff writer draws first, then the recovered turn.
342. 342 let provider = Scripted::new(vec![
343. 343 text_msg(
344. 344 "# Objective\nfinish\n# Current State\nmid\n# Decisions Made\nnone\n# Open Items\ndone\n# Obligations\nnone",
345. 345 ),
346. 346 text_msg("recovered and finished"),
347. 347 ]);
348. 348 let (mut agent, _dir) = setup(provider, vec![]);
349. 349 // Tiny policy so the fixture triggers the over-budget path with too
350. 350 // few turns to compact.
351. 351 agent.config.declared_window = 900;
352. 352 agent.config.max_output = 64;
353. 353
354. 354 let filler = "x".repeat(4000);
355. 355 let prompt = format!("task {filler}");
356. 356 let outcome = run(&mut agent, &prompt).await;
357. 357 assert!(
358. 358 matches!(
359. 359 outcome,
360. 360 TurnOutcome::Completed { .. }
361. 361 | TurnOutcome::MaxTurnsReached
362. 362 | TurnOutcome::Failed { .. }
363. 363 ),
364. 364 "unexpected: {outcome:?}"
365. 365 );
366. 366
367. 367 let session = agent.into_session().await;
368. 368 let has_handoff = session
369. 369 .chain_to_root()
370. 370 .iter()
371. 371 .any(|e| matches!(&e.payload, EntryPayload::Compaction(c) if c.reset_all));
372. 372 assert!(has_handoff, "handoff entry must exist on disk");
373. 373
374. 374 // Projection after reset contains ONLY summaries — no verbatim filler.
375. 375 let projected = session.derive_messages();
376. 376 assert!(
377. 377 !projected.iter().any(|m| m.text_content().contains(&filler)),
378. 378 "reset must clear verbatim history"
379. 379 );
380. 380}
381. 381