Vakyartha documentation

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3//! Incremental compaction (docs/design/68-context-engine.md §4): compaction
4//! fires only when the `WorkingSetPlanner` collapses real, card-carrying
5//! turns into a packet — so these tests drive the agent through several
6//! REAL turns (each producing a genuine `TurnCard` at close) rather than
7//! seeding raw messages directly.
8
9use std::collections::VecDeque;
10use std::sync::{Arc, Mutex};
11
12use tokio::sync::mpsc;
13use tokio_util::sync::CancellationToken;
14
15use tempfile::tempdir;
16
17use vak_agent::{Agent, AgentConfig, TurnOutcome};
18use vak_llm::stream;
19use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
20use vak_llm::{EventStream, LlmError, Provider};
21use vak_session::SessionLog;
22use vak_session::types::{EntryPayload, FrozenContract, SessionHeader};
23
24/// Routes by system-prompt marker: a compaction call carries
25/// `COMPACTION_SYSTEM`; every other request gets the next scripted step
26/// answer, cycling once exhausted (each turn here is a single no-tool-call
27/// step, so the same short pool of answers covers many turns).
28struct TaggedScripted {
29 steps: Mutex<VecDeque<AssistantMessage>>,
30 seen_systems: Arc<Mutex<Vec<String>>>,
31}
32
33impl TaggedScripted {
34 fn is_compaction(request: &ChatRequest) -> bool {
35 request
36 .system
37 .as_deref()
38 .is_some_and(|s| s.contains("context compactor"))
39 }
40}
41
42#[async_trait::async_trait]
43impl Provider for TaggedScripted {
44 fn name(&self) -> &str {
45 "scripted"
46 }
47
48 async fn stream(
49 &self,
50 request: ChatRequest,
51 _cancel: CancellationToken,
52 ) -> Result<EventStream, LlmError> {
53 self.seen_systems
54 .lock()
55 .unwrap()
56 .push(request.system.clone().unwrap_or_default());
57 let msg = if Self::is_compaction(&request) {
58 Some(text(
59 "Summary: prior turns condensed; open items carried forward.",
60 ))
61 } else {
62 let mut steps = self.steps.lock().unwrap();
63 let next = steps.pop_front();
64 if let Some(m) = &next {
65 steps.push_back(m.clone());
66 }
67 next
68 };
69 let (mut sink, rx) = stream::channel(64);
70 match msg {
71 Some(m) => {
72 sink.push(stream::StreamEvent::Start { partial: m.clone() });
73 sink.close_message(m).await;
74 }
75 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
76 }
77 Ok(rx)
78 }
79}
80
81fn text(t: &str) -> AssistantMessage {
82 AssistantMessage {
83 content: vec![ContentBlock::text(t)],
84 stop_reason: StopReason::EndTurn,
85 usage: Usage::default(),
86 model: "test-model".into(),
87 response_id: None,
88 }
89}
90
91fn header(session_id: &str, dir: &std::path::Path) -> SessionHeader {
92 SessionHeader {
93 agent: None,
94 session_id: session_id.into(),
95 created_at: chrono::Utc::now(),
96 cwd: dir.to_path_buf(),
97 parent_session_id: None,
98 contract_id: None,
99 work_item_id: None,
100 conversation: None,
101 contract: FrozenContract {
102 app_version: "0".into(),
103 provider: "scripted".into(),
104 model: "test-model".into(),
105 route_ladder: Vec::new(),
106 route_objective: String::new(),
107 route_annotations: Vec::new(),
108 system_prompt: "sys".into(),
109 permission_mode: "full-access".into(),
110 capabilities: Vec::new(),
111 prompt_layers: Vec::new(),
112 },
113 }
114}
115
116#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
117async fn a_tiny_horizon_eventually_triggers_incremental_compaction() {
118 let dir = tempdir().unwrap();
119 let log =
120 SessionLog::create(dir.path().join("s.jsonl"), header("compact", dir.path())).unwrap();
121
122 let seen = Arc::new(Mutex::new(Vec::new()));
123 let provider = Arc::new(TaggedScripted {
124 steps: Mutex::new(VecDeque::from(vec![text(
125 "a padded no-tool-call answer with enough filler text to accumulate real tokens over several turns of conversation",
126 )])),
127 seen_systems: seen.clone(),
128 });
129
130 let mut cfg = AgentConfig::new("sys");
131 // No live CapacityProfile is wired in, so the loop falls back to a
132 // metadata-only one built from these two fields (docs/design/68 §4).
133 // Small enough that recency alone cannot hold every turn at Full, and
134 // eventually not even at Card, forcing a packet.
135 cfg.declared_window = 600;
136 cfg.max_output = 20;
137 let mut agent = Agent::new(provider, log, cfg);
138
139 let mut compaction_seen = false;
140 for i in 0..20 {
141 let outcome = agent
142 .run(
143 &format!(
144 "padded question number {i} with enough filler text to accumulate real tokens across turns"
145 ),
146 &Default::default(),
147 CancellationToken::new(),
148 mpsc::channel(256).0,
149 )
150 .await;
151 assert!(
152 matches!(outcome, TurnOutcome::Completed { .. }),
153 "turn {i} did not complete: {outcome:?}"
154 );
155 if seen
156 .lock()
157 .unwrap()
158 .iter()
159 .any(|s| s.contains("context compactor"))
160 {
161 compaction_seen = true;
162 break;
163 }
164 }
165 assert!(
166 compaction_seen,
167 "expected an incremental compaction call within 20 turns of a 600-token horizon"
168 );
169
170 let session = agent.session.lock().await;
171 let packet = session
172 .chain_to_root()
173 .iter()
174 .find_map(|e| match &e.payload {
175 EntryPayload::Compaction(c) if !c.reset_all => Some(c.clone()),
176 _ => None,
177 })
178 .expect("a packet entry must be on the ledger");
179 assert!(
180 !packet.first_turn_id.is_empty() && !packet.last_turn_id.is_empty(),
181 "a packet is keyed by the turn range it covers: {packet:?}"
182 );
183 assert!(
184 session
185 .packet_for(&packet.first_turn_id, &packet.last_turn_id)
186 .is_some()
187 );
188
189 // The packet is a cache for the plan that asked for it, not a boundary:
190 // the plan-free (all-Full) projection carries every turn and no
191 // summary, while a plan asking for exactly that range renders it.
192 let plan_free: String = session
193 .derive_messages()
194 .iter()
195 .map(|m| m.text_content())
196 .collect::<Vec<_>>()
197 .join("\n");
198 assert!(
199 !plan_free.contains("<context_summary>") && plan_free.contains("padded question number 0"),
200 "a packet must not hide turns from the plan-free projection: {plan_free}"
201 );
202 let plan = vak_session::WorkingSetPlan {
203 packet_range: Some((packet.first_turn_id.clone(), packet.last_turn_id.clone())),
204 ..vak_session::WorkingSetPlan::default()
205 };
206 let planned: String = session
207 .derive_with_plan(&plan)
208 .iter()
209 .map(|m| m.text_content())
210 .collect::<Vec<_>>()
211 .join("\n");
212 assert!(
213 planned.contains("<context_summary>"),
214 "the plan that asked for the packet must see it: {planned}"
215 );
216}
217
218#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
219async fn no_usable_horizon_fails_closed_when_handoff_is_disabled() {
220 let dir = tempdir().unwrap();
221 let log = SessionLog::create(dir.path().join("s.jsonl"), header("over", dir.path())).unwrap();
222
223 let provider = Arc::new(TaggedScripted {
224 steps: Mutex::new(VecDeque::from(vec![text("never reached")])),
225 seen_systems: Arc::new(Mutex::new(Vec::new())),
226 });
227
228 let mut cfg = AgentConfig::new("sys");
229 // A horizon far too small to hold even the current directive plus
230 // output reserve: budget saturates to 0 (no usable horizon).
231 cfg.declared_window = 5;
232 cfg.max_output = 5;
233 cfg.handoff_reset = false;
234 cfg.run_retry_attempts = 0;
235 let mut agent = Agent::new(provider, log, cfg);
236
237 let outcome = agent
238 .run(
239 &"x".repeat(2_000),
240 &Default::default(),
241 CancellationToken::new(),
242 mpsc::channel(64).0,
243 )
244 .await;
245 match outcome {
246 TurnOutcome::Failed { error } => {
247 assert!(
248 error.to_string().contains("no usable horizon"),
249 "got: {error}"
250 );
251 }
252 other => panic!("expected fail-closed, got {other:?}"),
253 }
254}
255

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3//! Incremental compaction (docs/design/68-context-engine.md §4): compaction
4//! fires only when the `WorkingSetPlanner` collapses real, card-carrying
5//! turns into a packet — so these tests drive the agent through several
6//! REAL turns (each producing a genuine `TurnCard` at close) rather than
7//! seeding raw messages directly.
8
9use std::collections::VecDeque;
10use std::sync::{Arc, Mutex};
11
12use tokio::sync::mpsc;
13use tokio_util::sync::CancellationToken;
14
15use tempfile::tempdir;
16
17use vak_agent::{Agent, AgentConfig, TurnOutcome};
18use vak_llm::stream;
19use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
20use vak_llm::{EventStream, LlmError, Provider};
21use vak_session::SessionLog;
22use vak_session::types::{EntryPayload, FrozenContract, SessionHeader};
23
24/// Routes by system-prompt marker: a compaction call carries
25/// `COMPACTION_SYSTEM`; every other request gets the next scripted step
26/// answer, cycling once exhausted (each turn here is a single no-tool-call
27/// step, so the same short pool of answers covers many turns).
28struct TaggedScripted {
29 steps: Mutex<VecDeque<AssistantMessage>>,
30 seen_systems: Arc<Mutex<Vec<String>>>,
31}
32
33impl TaggedScripted {
34 fn is_compaction(request: &ChatRequest) -> bool {
35 request
36 .system
37 .as_deref()
38 .is_some_and(|s| s.contains("context compactor"))
39 }
40}
41
42#[async_trait::async_trait]
43impl Provider for TaggedScripted {
44 fn name(&self) -> &str {
45 "scripted"
46 }
47
48 async fn stream(
49 &self,
50 request: ChatRequest,
51 _cancel: CancellationToken,
52 ) -> Result<EventStream, LlmError> {
53 self.seen_systems
54 .lock()
55 .unwrap()
56 .push(request.system.clone().unwrap_or_default());
57 let msg = if Self::is_compaction(&request) {
58 Some(text(
59 "Summary: prior turns condensed; open items carried forward.",
60 ))
61 } else {
62 let mut steps = self.steps.lock().unwrap();
63 let next = steps.pop_front();
64 if let Some(m) = &next {
65 steps.push_back(m.clone());
66 }
67 next
68 };
69 let (mut sink, rx) = stream::channel(64);
70 match msg {
71 Some(m) => {
72 sink.push(stream::StreamEvent::Start { partial: m.clone() });
73 sink.close_message(m).await;
74 }
75 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
76 }
77 Ok(rx)
78 }
79}
80
81fn text(t: &str) -> AssistantMessage {
82 AssistantMessage {
83 content: vec![ContentBlock::text(t)],
84 stop_reason: StopReason::EndTurn,
85 usage: Usage::default(),
86 model: "test-model".into(),
87 response_id: None,
88 }
89}
90
91fn header(session_id: &str, dir: &std::path::Path) -> SessionHeader {
92 SessionHeader {
93 agent: None,
94 session_id: session_id.into(),
95 created_at: chrono::Utc::now(),
96 cwd: dir.to_path_buf(),
97 parent_session_id: None,
98 contract_id: None,
99 work_item_id: None,
100 conversation: None,
101 contract: FrozenContract {
102 app_version: "0".into(),
103 provider: "scripted".into(),
104 model: "test-model".into(),
105 route_ladder: Vec::new(),
106 route_objective: String::new(),
107 route_annotations: Vec::new(),
108 system_prompt: "sys".into(),
109 permission_mode: "full-access".into(),
110 capabilities: Vec::new(),
111 prompt_layers: Vec::new(),
112 },
113 }
114}
115
116#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
117async fn a_tiny_horizon_eventually_triggers_incremental_compaction() {
118 let dir = tempdir().unwrap();
119 let log =
120 SessionLog::create(dir.path().join("s.jsonl"), header("compact", dir.path())).unwrap();
121
122 let seen = Arc::new(Mutex::new(Vec::new()));
123 let provider = Arc::new(TaggedScripted {
124 steps: Mutex::new(VecDeque::from(vec![text(
125 "a padded no-tool-call answer with enough filler text to accumulate real tokens over several turns of conversation",
126 )])),
127 seen_systems: seen.clone(),
128 });
129
130 let mut cfg = AgentConfig::new("sys");
131 // No live CapacityProfile is wired in, so the loop falls back to a
132 // metadata-only one built from these two fields (docs/design/68 §4).
133 // Small enough that recency alone cannot hold every turn at Full, and
134 // eventually not even at Card, forcing a packet.
135 cfg.declared_window = 600;
136 cfg.max_output = 20;
137 let mut agent = Agent::new(provider, log, cfg);
138
139 let mut compaction_seen = false;
140 for i in 0..20 {
141 let outcome = agent
142 .run(
143 &format!(
144 "padded question number {i} with enough filler text to accumulate real tokens across turns"
145 ),
146 &Default::default(),
147 CancellationToken::new(),
148 mpsc::channel(256).0,
149 )
150 .await;
151 assert!(
152 matches!(outcome, TurnOutcome::Completed { .. }),
153 "turn {i} did not complete: {outcome:?}"
154 );
155 if seen
156 .lock()
157 .unwrap()
158 .iter()
159 .any(|s| s.contains("context compactor"))
160 {
161 compaction_seen = true;
162 break;
163 }
164 }
165 assert!(
166 compaction_seen,
167 "expected an incremental compaction call within 20 turns of a 600-token horizon"
168 );
169
170 let session = agent.session.lock().await;
171 let packet = session
172 .chain_to_root()
173 .iter()
174 .find_map(|e| match &e.payload {
175 EntryPayload::Compaction(c) if !c.reset_all => Some(c.clone()),
176 _ => None,
177 })
178 .expect("a packet entry must be on the ledger");
179 assert!(
180 !packet.first_turn_id.is_empty() && !packet.last_turn_id.is_empty(),
181 "a packet is keyed by the turn range it covers: {packet:?}"
182 );
183 assert!(
184 session
185 .packet_for(&packet.first_turn_id, &packet.last_turn_id)
186 .is_some()
187 );
188
189 // The packet is a cache for the plan that asked for it, not a boundary:
190 // the plan-free (all-Full) projection carries every turn and no
191 // summary, while a plan asking for exactly that range renders it.
192 let plan_free: String = session
193 .derive_messages()
194 .iter()
195 .map(|m| m.text_content())
196 .collect::<Vec<_>>()
197 .join("\n");
198 assert!(
199 !plan_free.contains("<context_summary>") && plan_free.contains("padded question number 0"),
200 "a packet must not hide turns from the plan-free projection: {plan_free}"
201 );
202 let plan = vak_session::WorkingSetPlan {
203 packet_range: Some((packet.first_turn_id.clone(), packet.last_turn_id.clone())),
204 ..vak_session::WorkingSetPlan::default()
205 };
206 let planned: String = session
207 .derive_with_plan(&plan)
208 .iter()
209 .map(|m| m.text_content())
210 .collect::<Vec<_>>()
211 .join("\n");
212 assert!(
213 planned.contains("<context_summary>"),
214 "the plan that asked for the packet must see it: {planned}"
215 );
216}
217
218#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
219async fn no_usable_horizon_fails_closed_when_handoff_is_disabled() {
220 let dir = tempdir().unwrap();
221 let log = SessionLog::create(dir.path().join("s.jsonl"), header("over", dir.path())).unwrap();
222
223 let provider = Arc::new(TaggedScripted {
224 steps: Mutex::new(VecDeque::from(vec![text("never reached")])),
225 seen_systems: Arc::new(Mutex::new(Vec::new())),
226 });
227
228 let mut cfg = AgentConfig::new("sys");
229 // A horizon far too small to hold even the current directive plus
230 // output reserve: budget saturates to 0 (no usable horizon).
231 cfg.declared_window = 5;
232 cfg.max_output = 5;
233 cfg.handoff_reset = false;
234 cfg.run_retry_attempts = 0;
235 let mut agent = Agent::new(provider, log, cfg);
236
237 let outcome = agent
238 .run(
239 &"x".repeat(2_000),
240 &Default::default(),
241 CancellationToken::new(),
242 mpsc::channel(64).0,
243 )
244 .await;
245 match outcome {
246 TurnOutcome::Failed { error } => {
247 assert!(
248 error.to_string().contains("no usable horizon"),
249 "got: {error}"
250 );
251 }
252 other => panic!("expected fail-closed, got {other:?}"),
253 }
254}
255

Indexing the workspace…