Vakyartha documentation

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3//! Model drift (docs/design/68-context-engine.md §7b) and the over-length
4//! replan (§5): both are runtime-enforcement paths with no live model, so
5//! they are covered here the same way the other repair-nudge regressions
6//! are — a scripted provider driving `Agent::run` end to end.
7
8use std::collections::VecDeque;
9use std::sync::{Arc, Mutex};
10
11use tokio::sync::mpsc;
12use tokio_util::sync::CancellationToken;
13
14use tempfile::tempdir;
15
16use vak_agent::{Agent, AgentConfig, TurnOutcome};
17use vak_llm::stream;
18use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
19use vak_llm::{EventStream, LlmError, Provider};
20use vak_session::types::{
21 EntryPayload, FrozenContract, MessageRecord, SessionHeader, TurnCardRecord,
22};
23use vak_session::{SessionLog, TurnIndex};
24
25struct Scripted {
26 responses: Mutex<VecDeque<AssistantMessage>>,
27}
28
29#[async_trait::async_trait]
30impl Provider for Scripted {
31 fn name(&self) -> &str {
32 "scripted"
33 }
34
35 async fn stream(
36 &self,
37 _request: ChatRequest,
38 _cancel: CancellationToken,
39 ) -> Result<EventStream, LlmError> {
40 let next = self.responses.lock().unwrap().pop_front();
41 let (mut sink, rx) = stream::channel(64);
42 match next {
43 Some(m) => {
44 sink.push(stream::StreamEvent::Start { partial: m.clone() });
45 sink.close_message(m).await;
46 }
47 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
48 }
49 Ok(rx)
50 }
51}
52
53/// A provider whose first N calls reject with `LlmError::Context`, then
54/// succeeds — for the over-length replan (§5).
55struct OverLengthThenOk {
56 context_errors_remaining: Mutex<u32>,
57 final_answer: Mutex<Option<AssistantMessage>>,
58}
59
60#[async_trait::async_trait]
61impl Provider for OverLengthThenOk {
62 fn name(&self) -> &str {
63 "scripted"
64 }
65
66 async fn stream(
67 &self,
68 _request: ChatRequest,
69 _cancel: CancellationToken,
70 ) -> Result<EventStream, LlmError> {
71 let (mut sink, rx) = stream::channel(64);
72 let should_reject = {
73 let mut remaining = self.context_errors_remaining.lock().unwrap();
74 if *remaining > 0 {
75 *remaining -= 1;
76 true
77 } else {
78 false
79 }
80 };
81 if should_reject {
82 sink.close_error(LlmError::Context("request too long for this model".into()))
83 .await;
84 } else {
85 let msg = self
86 .final_answer
87 .lock()
88 .unwrap()
89 .take()
90 .unwrap_or_else(|| text("done"));
91 sink.push(stream::StreamEvent::Start {
92 partial: msg.clone(),
93 });
94 sink.close_message(msg).await;
95 }
96 Ok(rx)
97 }
98}
99
100fn text(t: &str) -> AssistantMessage {
101 AssistantMessage {
102 content: vec![ContentBlock::text(t)],
103 stop_reason: StopReason::EndTurn,
104 usage: Usage::default(),
105 model: "test-model".into(),
106 response_id: None,
107 }
108}
109
110fn header(session_id: &str, dir: &std::path::Path) -> SessionHeader {
111 SessionHeader {
112 agent: None,
113 session_id: session_id.into(),
114 created_at: chrono::Utc::now(),
115 cwd: dir.to_path_buf(),
116 parent_session_id: None,
117 contract_id: None,
118 work_item_id: None,
119 conversation: None,
120 contract: FrozenContract {
121 app_version: "0".into(),
122 provider: "scripted".into(),
123 model: "test-model".into(),
124 route_ladder: Vec::new(),
125 route_objective: String::new(),
126 route_annotations: Vec::new(),
127 system_prompt: "sys".into(),
128 permission_mode: "full-access".into(),
129 capabilities: Vec::new(),
130 prompt_layers: Vec::new(),
131 },
132 }
133}
134
135/// Three consecutive verbatim repeats of a prior turn's answer -- the one
136/// surviving drift trigger (docs/design/68-context-engine.md §7: subject
137/// domains must never drive control flow, so a tool's declared capability
138/// domain is no longer compared against the reading's domains here) -- end
139/// the turn at the third drift event without even needing a fourth
140/// scripted response.
141#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
142async fn three_consecutive_drift_events_end_the_turn_degraded() {
143 let dir = tempdir().unwrap();
144 let mut log =
145 SessionLog::create(dir.path().join("s.jsonl"), header("drift3", dir.path())).unwrap();
146
147 let t1 = log
148 .append_message(MessageRecord {
149 message: vak_llm::Message::user_text("what is the capital of France"),
150 meta: None,
151 })
152 .unwrap()
153 .id;
154 log.append_message(MessageRecord {
155 message: vak_llm::Message::assistant(vec![ContentBlock::text("Paris is the capital.")]),
156 meta: None,
157 })
158 .unwrap();
159 let card = TurnIndex::from_log(&log)
160 .turn_by_id(&t1)
161 .unwrap()
162 .build_card("completed", "Paris is the capital.".to_string(), &|s| {
163 s.len() as u64 / 4
164 });
165 log.append_turn_card(TurnCardRecord { turn_id: t1, card })
166 .unwrap();
167
168 let provider = Arc::new(Scripted {
169 responses: Mutex::new(VecDeque::from(vec![
170 text("Paris is the capital."),
171 text("Paris is the capital."),
172 text("Paris is the capital."),
173 ])),
174 });
175 let cfg = AgentConfig::new("sys");
176 let mut agent = Agent::new(provider, log, cfg);
177
178 let outcome = agent
179 .run(
180 "how many people live in Tokyo",
181 &Default::default(),
182 CancellationToken::new(),
183 mpsc::channel(64).0,
184 )
185 .await;
186 match outcome {
187 TurnOutcome::Completed { response } => {
188 assert!(
189 response.text_content().contains("drifting"),
190 "expected the degraded drift message, got: {}",
191 response.text_content()
192 );
193 }
194 other => panic!("expected a degraded Completed outcome, got {other:?}"),
195 }
196
197 let session = agent.session.lock().await;
198 let saw_exhaustion = session.chain_to_root().iter().any(
199 |e| matches!(&e.payload, EntryPayload::Activity(a) if a.label == "model-drift-exhausted"),
200 );
201 assert!(
202 saw_exhaustion,
203 "expected a model-drift-exhausted diagnostic"
204 );
205}
206
207#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
208async fn a_verbatim_repeat_of_a_past_answer_is_treated_as_drift_and_redone() {
209 let dir = tempdir().unwrap();
210 let mut log = SessionLog::create(
211 dir.path().join("s.jsonl"),
212 header("drift-repeat", dir.path()),
213 )
214 .unwrap();
215
216 // A prior closed turn whose narration is the exact string the model
217 // will later repeat verbatim for an unrelated new directive.
218 let t1 = log
219 .append_message(MessageRecord {
220 message: vak_llm::Message::user_text("what is the capital of France"),
221 meta: None,
222 })
223 .unwrap()
224 .id;
225 log.append_message(MessageRecord {
226 message: vak_llm::Message::assistant(vec![ContentBlock::text("Paris is the capital.")]),
227 meta: None,
228 })
229 .unwrap();
230 let card = TurnIndex::from_log(&log)
231 .turn_by_id(&t1)
232 .unwrap()
233 .build_card("completed", "Paris is the capital.".to_string(), &|s| {
234 s.len() as u64 / 4
235 });
236 log.append_turn_card(TurnCardRecord { turn_id: t1, card })
237 .unwrap();
238
239 let provider = Arc::new(Scripted {
240 responses: Mutex::new(VecDeque::from(vec![
241 text("Paris is the capital."),
242 text("Tokyo has about 14 million people."),
243 ])),
244 });
245 let cfg = AgentConfig::new("sys");
246 let mut agent = Agent::new(provider, log, cfg);
247
248 let outcome = agent
249 .run(
250 "how many people live in Tokyo",
251 &Default::default(),
252 CancellationToken::new(),
253 mpsc::channel(64).0,
254 )
255 .await;
256 match outcome {
257 TurnOutcome::Completed { response } => {
258 assert_eq!(
259 response.text_content(),
260 "Tokyo has about 14 million people."
261 );
262 }
263 other => panic!("expected the redone answer to complete, got {other:?}"),
264 }
265}
266
267#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
268async fn over_length_rejection_lowers_the_horizon_and_retries_once() {
269 let dir = tempdir().unwrap();
270 let log =
271 SessionLog::create(dir.path().join("s.jsonl"), header("overlength", dir.path())).unwrap();
272
273 let provider = Arc::new(OverLengthThenOk {
274 context_errors_remaining: Mutex::new(1),
275 final_answer: Mutex::new(Some(text("recovered after replan"))),
276 });
277 let cfg = AgentConfig::new("sys");
278 let mut agent = Agent::new(provider, log, cfg);
279
280 let outcome = agent
281 .run(
282 "hello",
283 &Default::default(),
284 CancellationToken::new(),
285 mpsc::channel(64).0,
286 )
287 .await;
288 match outcome {
289 TurnOutcome::Completed { response } => {
290 assert_eq!(response.text_content(), "recovered after replan");
291 }
292 other => panic!("expected recovery after one replanned retry, got {other:?}"),
293 }
294
295 assert!(
296 agent.config.capacity.is_some(),
297 "the over-length feedback must persist a CapacityProfile"
298 );
299 let profile = agent.config.capacity.as_ref().unwrap();
300 assert!(
301 profile.verified_window.is_some(),
302 "verified_window must be set from the rejection"
303 );
304
305 let session = agent.session.lock().await;
306 let saw_feedback = session.chain_to_root().iter().any(
307 |e| matches!(&e.payload, EntryPayload::Activity(a) if a.label.contains("over-length")),
308 );
309 assert!(saw_feedback, "expected a capacity-feedback activity");
310}
311
312#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
313async fn a_second_consecutive_over_length_rejection_fails_the_turn() {
314 let dir = tempdir().unwrap();
315 let log = SessionLog::create(
316 dir.path().join("s.jsonl"),
317 header("overlength2", dir.path()),
318 )
319 .unwrap();
320
321 let provider = Arc::new(OverLengthThenOk {
322 // Never runs out: every call rejects.
323 context_errors_remaining: Mutex::new(u32::MAX),
324 final_answer: Mutex::new(None),
325 });
326 let mut cfg = AgentConfig::new("sys");
327 cfg.run_retry_attempts = 0;
328 let mut agent = Agent::new(provider, log, cfg);
329
330 let outcome = agent
331 .run(
332 "hello",
333 &Default::default(),
334 CancellationToken::new(),
335 mpsc::channel(64).0,
336 )
337 .await;
338 match outcome {
339 TurnOutcome::Failed { error } => {
340 assert!(matches!(error, LlmError::Context(_)), "got: {error}");
341 }
342 other => panic!("expected the second Context error to fail the turn, got {other:?}"),
343 }
344}
345

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3//! Model drift (docs/design/68-context-engine.md §7b) and the over-length
4//! replan (§5): both are runtime-enforcement paths with no live model, so
5//! they are covered here the same way the other repair-nudge regressions
6//! are — a scripted provider driving `Agent::run` end to end.
7
8use std::collections::VecDeque;
9use std::sync::{Arc, Mutex};
10
11use tokio::sync::mpsc;
12use tokio_util::sync::CancellationToken;
13
14use tempfile::tempdir;
15
16use vak_agent::{Agent, AgentConfig, TurnOutcome};
17use vak_llm::stream;
18use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
19use vak_llm::{EventStream, LlmError, Provider};
20use vak_session::types::{
21 EntryPayload, FrozenContract, MessageRecord, SessionHeader, TurnCardRecord,
22};
23use vak_session::{SessionLog, TurnIndex};
24
25struct Scripted {
26 responses: Mutex<VecDeque<AssistantMessage>>,
27}
28
29#[async_trait::async_trait]
30impl Provider for Scripted {
31 fn name(&self) -> &str {
32 "scripted"
33 }
34
35 async fn stream(
36 &self,
37 _request: ChatRequest,
38 _cancel: CancellationToken,
39 ) -> Result<EventStream, LlmError> {
40 let next = self.responses.lock().unwrap().pop_front();
41 let (mut sink, rx) = stream::channel(64);
42 match next {
43 Some(m) => {
44 sink.push(stream::StreamEvent::Start { partial: m.clone() });
45 sink.close_message(m).await;
46 }
47 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
48 }
49 Ok(rx)
50 }
51}
52
53/// A provider whose first N calls reject with `LlmError::Context`, then
54/// succeeds — for the over-length replan (§5).
55struct OverLengthThenOk {
56 context_errors_remaining: Mutex<u32>,
57 final_answer: Mutex<Option<AssistantMessage>>,
58}
59
60#[async_trait::async_trait]
61impl Provider for OverLengthThenOk {
62 fn name(&self) -> &str {
63 "scripted"
64 }
65
66 async fn stream(
67 &self,
68 _request: ChatRequest,
69 _cancel: CancellationToken,
70 ) -> Result<EventStream, LlmError> {
71 let (mut sink, rx) = stream::channel(64);
72 let should_reject = {
73 let mut remaining = self.context_errors_remaining.lock().unwrap();
74 if *remaining > 0 {
75 *remaining -= 1;
76 true
77 } else {
78 false
79 }
80 };
81 if should_reject {
82 sink.close_error(LlmError::Context("request too long for this model".into()))
83 .await;
84 } else {
85 let msg = self
86 .final_answer
87 .lock()
88 .unwrap()
89 .take()
90 .unwrap_or_else(|| text("done"));
91 sink.push(stream::StreamEvent::Start {
92 partial: msg.clone(),
93 });
94 sink.close_message(msg).await;
95 }
96 Ok(rx)
97 }
98}
99
100fn text(t: &str) -> AssistantMessage {
101 AssistantMessage {
102 content: vec![ContentBlock::text(t)],
103 stop_reason: StopReason::EndTurn,
104 usage: Usage::default(),
105 model: "test-model".into(),
106 response_id: None,
107 }
108}
109
110fn header(session_id: &str, dir: &std::path::Path) -> SessionHeader {
111 SessionHeader {
112 agent: None,
113 session_id: session_id.into(),
114 created_at: chrono::Utc::now(),
115 cwd: dir.to_path_buf(),
116 parent_session_id: None,
117 contract_id: None,
118 work_item_id: None,
119 conversation: None,
120 contract: FrozenContract {
121 app_version: "0".into(),
122 provider: "scripted".into(),
123 model: "test-model".into(),
124 route_ladder: Vec::new(),
125 route_objective: String::new(),
126 route_annotations: Vec::new(),
127 system_prompt: "sys".into(),
128 permission_mode: "full-access".into(),
129 capabilities: Vec::new(),
130 prompt_layers: Vec::new(),
131 },
132 }
133}
134
135/// Three consecutive verbatim repeats of a prior turn's answer -- the one
136/// surviving drift trigger (docs/design/68-context-engine.md §7: subject
137/// domains must never drive control flow, so a tool's declared capability
138/// domain is no longer compared against the reading's domains here) -- end
139/// the turn at the third drift event without even needing a fourth
140/// scripted response.
141#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
142async fn three_consecutive_drift_events_end_the_turn_degraded() {
143 let dir = tempdir().unwrap();
144 let mut log =
145 SessionLog::create(dir.path().join("s.jsonl"), header("drift3", dir.path())).unwrap();
146
147 let t1 = log
148 .append_message(MessageRecord {
149 message: vak_llm::Message::user_text("what is the capital of France"),
150 meta: None,
151 })
152 .unwrap()
153 .id;
154 log.append_message(MessageRecord {
155 message: vak_llm::Message::assistant(vec![ContentBlock::text("Paris is the capital.")]),
156 meta: None,
157 })
158 .unwrap();
159 let card = TurnIndex::from_log(&log)
160 .turn_by_id(&t1)
161 .unwrap()
162 .build_card("completed", "Paris is the capital.".to_string(), &|s| {
163 s.len() as u64 / 4
164 });
165 log.append_turn_card(TurnCardRecord { turn_id: t1, card })
166 .unwrap();
167
168 let provider = Arc::new(Scripted {
169 responses: Mutex::new(VecDeque::from(vec![
170 text("Paris is the capital."),
171 text("Paris is the capital."),
172 text("Paris is the capital."),
173 ])),
174 });
175 let cfg = AgentConfig::new("sys");
176 let mut agent = Agent::new(provider, log, cfg);
177
178 let outcome = agent
179 .run(
180 "how many people live in Tokyo",
181 &Default::default(),
182 CancellationToken::new(),
183 mpsc::channel(64).0,
184 )
185 .await;
186 match outcome {
187 TurnOutcome::Completed { response } => {
188 assert!(
189 response.text_content().contains("drifting"),
190 "expected the degraded drift message, got: {}",
191 response.text_content()
192 );
193 }
194 other => panic!("expected a degraded Completed outcome, got {other:?}"),
195 }
196
197 let session = agent.session.lock().await;
198 let saw_exhaustion = session.chain_to_root().iter().any(
199 |e| matches!(&e.payload, EntryPayload::Activity(a) if a.label == "model-drift-exhausted"),
200 );
201 assert!(
202 saw_exhaustion,
203 "expected a model-drift-exhausted diagnostic"
204 );
205}
206
207#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
208async fn a_verbatim_repeat_of_a_past_answer_is_treated_as_drift_and_redone() {
209 let dir = tempdir().unwrap();
210 let mut log = SessionLog::create(
211 dir.path().join("s.jsonl"),
212 header("drift-repeat", dir.path()),
213 )
214 .unwrap();
215
216 // A prior closed turn whose narration is the exact string the model
217 // will later repeat verbatim for an unrelated new directive.
218 let t1 = log
219 .append_message(MessageRecord {
220 message: vak_llm::Message::user_text("what is the capital of France"),
221 meta: None,
222 })
223 .unwrap()
224 .id;
225 log.append_message(MessageRecord {
226 message: vak_llm::Message::assistant(vec![ContentBlock::text("Paris is the capital.")]),
227 meta: None,
228 })
229 .unwrap();
230 let card = TurnIndex::from_log(&log)
231 .turn_by_id(&t1)
232 .unwrap()
233 .build_card("completed", "Paris is the capital.".to_string(), &|s| {
234 s.len() as u64 / 4
235 });
236 log.append_turn_card(TurnCardRecord { turn_id: t1, card })
237 .unwrap();
238
239 let provider = Arc::new(Scripted {
240 responses: Mutex::new(VecDeque::from(vec![
241 text("Paris is the capital."),
242 text("Tokyo has about 14 million people."),
243 ])),
244 });
245 let cfg = AgentConfig::new("sys");
246 let mut agent = Agent::new(provider, log, cfg);
247
248 let outcome = agent
249 .run(
250 "how many people live in Tokyo",
251 &Default::default(),
252 CancellationToken::new(),
253 mpsc::channel(64).0,
254 )
255 .await;
256 match outcome {
257 TurnOutcome::Completed { response } => {
258 assert_eq!(
259 response.text_content(),
260 "Tokyo has about 14 million people."
261 );
262 }
263 other => panic!("expected the redone answer to complete, got {other:?}"),
264 }
265}
266
267#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
268async fn over_length_rejection_lowers_the_horizon_and_retries_once() {
269 let dir = tempdir().unwrap();
270 let log =
271 SessionLog::create(dir.path().join("s.jsonl"), header("overlength", dir.path())).unwrap();
272
273 let provider = Arc::new(OverLengthThenOk {
274 context_errors_remaining: Mutex::new(1),
275 final_answer: Mutex::new(Some(text("recovered after replan"))),
276 });
277 let cfg = AgentConfig::new("sys");
278 let mut agent = Agent::new(provider, log, cfg);
279
280 let outcome = agent
281 .run(
282 "hello",
283 &Default::default(),
284 CancellationToken::new(),
285 mpsc::channel(64).0,
286 )
287 .await;
288 match outcome {
289 TurnOutcome::Completed { response } => {
290 assert_eq!(response.text_content(), "recovered after replan");
291 }
292 other => panic!("expected recovery after one replanned retry, got {other:?}"),
293 }
294
295 assert!(
296 agent.config.capacity.is_some(),
297 "the over-length feedback must persist a CapacityProfile"
298 );
299 let profile = agent.config.capacity.as_ref().unwrap();
300 assert!(
301 profile.verified_window.is_some(),
302 "verified_window must be set from the rejection"
303 );
304
305 let session = agent.session.lock().await;
306 let saw_feedback = session.chain_to_root().iter().any(
307 |e| matches!(&e.payload, EntryPayload::Activity(a) if a.label.contains("over-length")),
308 );
309 assert!(saw_feedback, "expected a capacity-feedback activity");
310}
311
312#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
313async fn a_second_consecutive_over_length_rejection_fails_the_turn() {
314 let dir = tempdir().unwrap();
315 let log = SessionLog::create(
316 dir.path().join("s.jsonl"),
317 header("overlength2", dir.path()),
318 )
319 .unwrap();
320
321 let provider = Arc::new(OverLengthThenOk {
322 // Never runs out: every call rejects.
323 context_errors_remaining: Mutex::new(u32::MAX),
324 final_answer: Mutex::new(None),
325 });
326 let mut cfg = AgentConfig::new("sys");
327 cfg.run_retry_attempts = 0;
328 let mut agent = Agent::new(provider, log, cfg);
329
330 let outcome = agent
331 .run(
332 "hello",
333 &Default::default(),
334 CancellationToken::new(),
335 mpsc::channel(64).0,
336 )
337 .await;
338 match outcome {
339 TurnOutcome::Failed { error } => {
340 assert!(matches!(error, LlmError::Context(_)), "got: {error}");
341 }
342 other => panic!("expected the second Context error to fail the turn, got {other:?}"),
343 }
344}
345

Drift And Replan

Indexing the workspace…

Drift And Replan