Vakyartha documentation

1//! `Core::reflect_after_turn` — the background reflection seam shared by
2//! every surface (docs/design/29 P1). Contract under test: best-effort by
3//! design; flag-off / budget-denied / concurrent / failing paths all skip
4//! cleanly and never dispatch outside their allowed envelope.
5#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
6
7use std::sync::Arc;
8use std::sync::atomic::{AtomicUsize, Ordering};
9
10use tokio_util::sync::CancellationToken;
11
12use vak_llm::stream;
13use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
14use vak_llm::{EventStream, LlmError, Provider};
15
16/// Provider stub that counts dispatches, optionally fails, and optionally
17/// holds the response open until released (to force real overlap between
18/// two concurrent reflection passes).
19struct Counting {
20 calls: Arc<AtomicUsize>,
21 reply: &'static str,
22 fail: bool,
23 hold: Option<Arc<tokio::sync::Notify>>,
24}
25
26fn msg(text: &str) -> AssistantMessage {
27 AssistantMessage {
28 content: vec![ContentBlock::text(text)],
29 stop_reason: StopReason::EndTurn,
30 usage: Usage::default(),
31 model: "test-model".into(),
32 response_id: None,
33 }
34}
35
36#[async_trait::async_trait]
37impl Provider for Counting {
38 fn name(&self) -> &str {
39 "counting"
40 }
41
42 async fn stream(
43 &self,
44 _req: ChatRequest,
45 _cancel: CancellationToken,
46 ) -> Result<EventStream, LlmError> {
47 self.calls.fetch_add(1, Ordering::SeqCst);
48 if let Some(hold) = &self.hold {
49 hold.notified().await;
50 }
51 let (mut sink, rx) = stream::channel(8);
52 if self.fail {
53 sink.close_error(LlmError::Parse("reflection boom".into()))
54 .await;
55 } else {
56 sink.push(stream::StreamEvent::Start {
57 partial: msg(self.reply),
58 });
59 sink.close_message(msg(self.reply)).await;
60 }
61 Ok(rx)
62 }
63}
64
65const GOOD_REPLY: &str = r#"{"notes":[{"note":"the release pipeline pauses before every rollback window","kind":"decision","tag":"releases"}],"skill":{"name":"ship-guarded","description":"Ship with rollbacks guarded","instructions":"Run scripts/ship.sh after checks"}}"#;
66
67fn core_in(dir: &tempfile::TempDir, project_config: &str) -> vak_core::Core {
68 let project = dir.path().join(".vak");
69 std::fs::create_dir_all(&project).unwrap();
70 std::fs::write(project.join("config.toml"), project_config).unwrap();
71 vak_config::paths::isolate_home_for_tests();
72 let core = vak_core::Core::new_with_trust(dir.path().to_path_buf(), true).unwrap();
73 core.set_sessions_home(dir.path().join("home"));
74 core
75}
76
77#[tokio::test]
78async fn reflection_disabled_skips_without_dispatch() {
79 let dir = tempfile::tempdir().unwrap();
80 // Project layer pins reflection off so a developer's global config
81 // cannot leak into this hermetic test.
82 let core = core_in(&dir, "[memory]\nreflection = false\n");
83 let calls = Arc::new(AtomicUsize::new(0));
84 core.set_provider_instance(Arc::new(Counting {
85 calls: calls.clone(),
86 reply: GOOD_REPLY,
87 fail: false,
88 hold: None,
89 }));
90 let session = core.start_session().await.unwrap();
91
92 let out = core.reflect_after_turn(&session, "all done").await;
93
94 assert_eq!(
95 out,
96 vak_core::reflection::ReflectionOutcome::Skipped {
97 reason: "reflection-disabled"
98 }
99 );
100 assert_eq!(calls.load(Ordering::SeqCst), 0, "no provider dispatch");
101 assert!(vak_core::memory::list_notes(&core.sessions_home(), core.cwd()).is_empty());
102}
103
104#[tokio::test]
105async fn memory_writes_disabled_skips_without_dispatch() {
106 let dir = tempfile::tempdir().unwrap();
107 let core = core_in(&dir, "[memory]\nreflection = true\nwrite_enabled = false\n");
108 let calls = Arc::new(AtomicUsize::new(0));
109 core.set_provider_instance(Arc::new(Counting {
110 calls: calls.clone(),
111 reply: GOOD_REPLY,
112 fail: false,
113 hold: None,
114 }));
115 let session = core.start_session().await.unwrap();
116
117 let out = core.reflect_after_turn(&session, "all done").await;
118
119 assert_eq!(
120 out,
121 vak_core::reflection::ReflectionOutcome::Skipped {
122 reason: "memory-writes-disabled"
123 }
124 );
125 assert_eq!(calls.load(Ordering::SeqCst), 0);
126}
127
128#[tokio::test]
129async fn read_only_reflection_skips_without_writing_memory() {
130 let dir = tempfile::tempdir().unwrap();
131 let core = core_in(
132 &dir,
133 "permission_mode = \"read-only\"\n[memory]\nreflection = true\n",
134 );
135 let calls = Arc::new(AtomicUsize::new(0));
136 core.set_provider_instance(Arc::new(Counting {
137 calls: calls.clone(),
138 reply: GOOD_REPLY,
139 fail: false,
140 hold: None,
141 }));
142 let session = core.start_session().await.unwrap();
143 let out = core.reflect_after_turn(&session, "all done").await;
144 assert_eq!(
145 out,
146 vak_core::reflection::ReflectionOutcome::Skipped {
147 reason: "permission-mode-read-only"
148 }
149 );
150 assert_eq!(calls.load(Ordering::SeqCst), 0);
151 assert!(vak_core::memory::list_notes(&core.sessions_home(), core.cwd()).is_empty());
152}
153
154#[tokio::test]
155async fn budget_denied_skips_before_any_dispatch() {
156 let dir = tempfile::tempdir().unwrap();
157 let home = dir.path().join("home");
158 // Priced test model + a day cap already blown by seeded spend: the
159 // gate must deny BEFORE the provider is ever touched.
160 let core = core_in(
161 &dir,
162 "[memory]\nreflection = true\n[finops]\nmax_day_usd = 1.0\n\n[finops.price_overrides.test-model]\ninput = 1.25\noutput = 6.0\n",
163 );
164 core.set_model("test-model".into());
165 let ledger = vak_core::finops::FinOpsLedger::new(&home);
166 ledger
167 .append(&vak_core::finops::CostRow {
168 ts: chrono::Utc::now(),
169 model: "test-model".into(),
170 provider: "counting".into(),
171 input_tokens: 1,
172 output_tokens: 1,
173 cache_read_input_tokens: None,
174 usd: Some(50.0),
175 source: "test-seed".into(),
176 session_id: "seed".into(),
177 })
178 .unwrap();
179 let calls = Arc::new(AtomicUsize::new(0));
180 core.set_provider_instance(Arc::new(Counting {
181 calls: calls.clone(),
182 reply: GOOD_REPLY,
183 fail: false,
184 hold: None,
185 }));
186 let session = core.start_session().await.unwrap();
187
188 let out = core.reflect_after_turn(&session, "all done").await;
189
190 assert_eq!(
191 out,
192 vak_core::reflection::ReflectionOutcome::Skipped { reason: "budget" }
193 );
194 assert_eq!(
195 calls.load(Ordering::SeqCst),
196 0,
197 "denial must precede dispatch"
198 );
199}
200
201#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
202async fn concurrent_turns_reflect_once_and_never_duplicate() {
203 let dir = tempfile::tempdir().unwrap();
204 // Explicit generous cap keeps the gate deterministic even when the
205 // developer's global config carries tight finops knobs.
206 let core = core_in(
207 &dir,
208 "[memory]\nreflection = true\n[finops]\nmax_day_usd = 1000.0\n\n[finops.price_overrides.test-model]\ninput = 1.25\noutput = 6.0\n",
209 );
210 core.set_model("test-model".into());
211 let calls = Arc::new(AtomicUsize::new(0));
212 let hold = Arc::new(tokio::sync::Notify::new());
213 core.set_provider_instance(Arc::new(Counting {
214 calls: calls.clone(),
215 reply: GOOD_REPLY,
216 fail: false,
217 hold: Some(hold.clone()),
218 }));
219 let session = core.start_session().await.unwrap();
220 let header_b = session.header().unwrap().clone();
221
222 // A starts first and parks inside the aux call while holding the
223 // in-flight marker (dispatch happens strictly after acquisition).
224 let core_a = core.clone();
225 let ha = tokio::spawn(async move { core_a.reflect_after_turn(&session, "all done").await });
226 for _ in 0..200 {
227 if calls.load(Ordering::SeqCst) == 1 {
228 break;
229 }
230 tokio::time::sleep(std::time::Duration::from_millis(5)).await;
231 }
232 assert_eq!(calls.load(Ordering::SeqCst), 1, "A dispatched exactly once");
233
234 // B races in on the same session id while A is mid-flight. Ledger
235 // handles are exclusively locked, so this second surface carries its
236 // own log file cloned from A's header — legal here because a skipped
237 // pass exits at the marker before touching any content or disk.
238 let s_b =
239 vak_session::SessionLog::create(dir.path().join("second-surface.jsonl"), header_b).unwrap();
240 let core_b = core.clone();
241 let hb = tokio::spawn(async move { core_b.reflect_after_turn(&s_b, "all done").await });
242 // ...and must be skipped without its own dispatch.
243 let b = hb.await.unwrap();
244 assert_eq!(
245 b,
246 vak_core::reflection::ReflectionOutcome::Skipped {
247 reason: "already-in-flight"
248 }
249 );
250
251 // Release A; it completes as the single reflected pass.
252 hold.notify_one();
253 let a = ha.await.unwrap();
254 assert_eq!(
255 a,
256 vak_core::reflection::ReflectionOutcome::Reflected {
257 notes_added: 1,
258 skills_proposed: true
259 }
260 );
261 assert_eq!(calls.load(Ordering::SeqCst), 1, "no second dispatch ever");
262 let notes = vak_core::memory::list_notes(&core.sessions_home(), core.cwd());
263 assert_eq!(notes.len(), 1, "one note total — no duplicates");
264}
265
266#[tokio::test]
267async fn provider_error_yields_skipped_not_panic() {
268 let dir = tempfile::tempdir().unwrap();
269 let core = core_in(&dir, "[memory]\nreflection = true\n");
270 let calls = Arc::new(AtomicUsize::new(0));
271 core.set_provider_instance(Arc::new(Counting {
272 calls: calls.clone(),
273 reply: GOOD_REPLY,
274 fail: true,
275 hold: None,
276 }));
277 let session = core.start_session().await.unwrap();
278
279 let out = core.reflect_after_turn(&session, "all done").await;
280
281 assert_eq!(calls.load(Ordering::SeqCst), 1, "dispatch attempted");
282 assert_eq!(
283 out,
284 vak_core::reflection::ReflectionOutcome::Skipped {
285 reason: "reflect-call-failed"
286 }
287 );
288 assert!(vak_core::memory::list_notes(&core.sessions_home(), core.cwd()).is_empty());
289}
290

1//! `Core::reflect_after_turn` — the background reflection seam shared by
2//! every surface (docs/design/29 P1). Contract under test: best-effort by
3//! design; flag-off / budget-denied / concurrent / failing paths all skip
4//! cleanly and never dispatch outside their allowed envelope.
5#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
6
7use std::sync::Arc;
8use std::sync::atomic::{AtomicUsize, Ordering};
9
10use tokio_util::sync::CancellationToken;
11
12use vak_llm::stream;
13use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
14use vak_llm::{EventStream, LlmError, Provider};
15
16/// Provider stub that counts dispatches, optionally fails, and optionally
17/// holds the response open until released (to force real overlap between
18/// two concurrent reflection passes).
19struct Counting {
20 calls: Arc<AtomicUsize>,
21 reply: &'static str,
22 fail: bool,
23 hold: Option<Arc<tokio::sync::Notify>>,
24}
25
26fn msg(text: &str) -> AssistantMessage {
27 AssistantMessage {
28 content: vec![ContentBlock::text(text)],
29 stop_reason: StopReason::EndTurn,
30 usage: Usage::default(),
31 model: "test-model".into(),
32 response_id: None,
33 }
34}
35
36#[async_trait::async_trait]
37impl Provider for Counting {
38 fn name(&self) -> &str {
39 "counting"
40 }
41
42 async fn stream(
43 &self,
44 _req: ChatRequest,
45 _cancel: CancellationToken,
46 ) -> Result<EventStream, LlmError> {
47 self.calls.fetch_add(1, Ordering::SeqCst);
48 if let Some(hold) = &self.hold {
49 hold.notified().await;
50 }
51 let (mut sink, rx) = stream::channel(8);
52 if self.fail {
53 sink.close_error(LlmError::Parse("reflection boom".into()))
54 .await;
55 } else {
56 sink.push(stream::StreamEvent::Start {
57 partial: msg(self.reply),
58 });
59 sink.close_message(msg(self.reply)).await;
60 }
61 Ok(rx)
62 }
63}
64
65const GOOD_REPLY: &str = r#"{"notes":[{"note":"the release pipeline pauses before every rollback window","kind":"decision","tag":"releases"}],"skill":{"name":"ship-guarded","description":"Ship with rollbacks guarded","instructions":"Run scripts/ship.sh after checks"}}"#;
66
67fn core_in(dir: &tempfile::TempDir, project_config: &str) -> vak_core::Core {
68 let project = dir.path().join(".vak");
69 std::fs::create_dir_all(&project).unwrap();
70 std::fs::write(project.join("config.toml"), project_config).unwrap();
71 vak_config::paths::isolate_home_for_tests();
72 let core = vak_core::Core::new_with_trust(dir.path().to_path_buf(), true).unwrap();
73 core.set_sessions_home(dir.path().join("home"));
74 core
75}
76
77#[tokio::test]
78async fn reflection_disabled_skips_without_dispatch() {
79 let dir = tempfile::tempdir().unwrap();
80 // Project layer pins reflection off so a developer's global config
81 // cannot leak into this hermetic test.
82 let core = core_in(&dir, "[memory]\nreflection = false\n");
83 let calls = Arc::new(AtomicUsize::new(0));
84 core.set_provider_instance(Arc::new(Counting {
85 calls: calls.clone(),
86 reply: GOOD_REPLY,
87 fail: false,
88 hold: None,
89 }));
90 let session = core.start_session().await.unwrap();
91
92 let out = core.reflect_after_turn(&session, "all done").await;
93
94 assert_eq!(
95 out,
96 vak_core::reflection::ReflectionOutcome::Skipped {
97 reason: "reflection-disabled"
98 }
99 );
100 assert_eq!(calls.load(Ordering::SeqCst), 0, "no provider dispatch");
101 assert!(vak_core::memory::list_notes(&core.sessions_home(), core.cwd()).is_empty());
102}
103
104#[tokio::test]
105async fn memory_writes_disabled_skips_without_dispatch() {
106 let dir = tempfile::tempdir().unwrap();
107 let core = core_in(&dir, "[memory]\nreflection = true\nwrite_enabled = false\n");
108 let calls = Arc::new(AtomicUsize::new(0));
109 core.set_provider_instance(Arc::new(Counting {
110 calls: calls.clone(),
111 reply: GOOD_REPLY,
112 fail: false,
113 hold: None,
114 }));
115 let session = core.start_session().await.unwrap();
116
117 let out = core.reflect_after_turn(&session, "all done").await;
118
119 assert_eq!(
120 out,
121 vak_core::reflection::ReflectionOutcome::Skipped {
122 reason: "memory-writes-disabled"
123 }
124 );
125 assert_eq!(calls.load(Ordering::SeqCst), 0);
126}
127
128#[tokio::test]
129async fn read_only_reflection_skips_without_writing_memory() {
130 let dir = tempfile::tempdir().unwrap();
131 let core = core_in(
132 &dir,
133 "permission_mode = \"read-only\"\n[memory]\nreflection = true\n",
134 );
135 let calls = Arc::new(AtomicUsize::new(0));
136 core.set_provider_instance(Arc::new(Counting {
137 calls: calls.clone(),
138 reply: GOOD_REPLY,
139 fail: false,
140 hold: None,
141 }));
142 let session = core.start_session().await.unwrap();
143 let out = core.reflect_after_turn(&session, "all done").await;
144 assert_eq!(
145 out,
146 vak_core::reflection::ReflectionOutcome::Skipped {
147 reason: "permission-mode-read-only"
148 }
149 );
150 assert_eq!(calls.load(Ordering::SeqCst), 0);
151 assert!(vak_core::memory::list_notes(&core.sessions_home(), core.cwd()).is_empty());
152}
153
154#[tokio::test]
155async fn budget_denied_skips_before_any_dispatch() {
156 let dir = tempfile::tempdir().unwrap();
157 let home = dir.path().join("home");
158 // Priced test model + a day cap already blown by seeded spend: the
159 // gate must deny BEFORE the provider is ever touched.
160 let core = core_in(
161 &dir,
162 "[memory]\nreflection = true\n[finops]\nmax_day_usd = 1.0\n\n[finops.price_overrides.test-model]\ninput = 1.25\noutput = 6.0\n",
163 );
164 core.set_model("test-model".into());
165 let ledger = vak_core::finops::FinOpsLedger::new(&home);
166 ledger
167 .append(&vak_core::finops::CostRow {
168 ts: chrono::Utc::now(),
169 model: "test-model".into(),
170 provider: "counting".into(),
171 input_tokens: 1,
172 output_tokens: 1,
173 cache_read_input_tokens: None,
174 usd: Some(50.0),
175 source: "test-seed".into(),
176 session_id: "seed".into(),
177 })
178 .unwrap();
179 let calls = Arc::new(AtomicUsize::new(0));
180 core.set_provider_instance(Arc::new(Counting {
181 calls: calls.clone(),
182 reply: GOOD_REPLY,
183 fail: false,
184 hold: None,
185 }));
186 let session = core.start_session().await.unwrap();
187
188 let out = core.reflect_after_turn(&session, "all done").await;
189
190 assert_eq!(
191 out,
192 vak_core::reflection::ReflectionOutcome::Skipped { reason: "budget" }
193 );
194 assert_eq!(
195 calls.load(Ordering::SeqCst),
196 0,
197 "denial must precede dispatch"
198 );
199}
200
201#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
202async fn concurrent_turns_reflect_once_and_never_duplicate() {
203 let dir = tempfile::tempdir().unwrap();
204 // Explicit generous cap keeps the gate deterministic even when the
205 // developer's global config carries tight finops knobs.
206 let core = core_in(
207 &dir,
208 "[memory]\nreflection = true\n[finops]\nmax_day_usd = 1000.0\n\n[finops.price_overrides.test-model]\ninput = 1.25\noutput = 6.0\n",
209 );
210 core.set_model("test-model".into());
211 let calls = Arc::new(AtomicUsize::new(0));
212 let hold = Arc::new(tokio::sync::Notify::new());
213 core.set_provider_instance(Arc::new(Counting {
214 calls: calls.clone(),
215 reply: GOOD_REPLY,
216 fail: false,
217 hold: Some(hold.clone()),
218 }));
219 let session = core.start_session().await.unwrap();
220 let header_b = session.header().unwrap().clone();
221
222 // A starts first and parks inside the aux call while holding the
223 // in-flight marker (dispatch happens strictly after acquisition).
224 let core_a = core.clone();
225 let ha = tokio::spawn(async move { core_a.reflect_after_turn(&session, "all done").await });
226 for _ in 0..200 {
227 if calls.load(Ordering::SeqCst) == 1 {
228 break;
229 }
230 tokio::time::sleep(std::time::Duration::from_millis(5)).await;
231 }
232 assert_eq!(calls.load(Ordering::SeqCst), 1, "A dispatched exactly once");
233
234 // B races in on the same session id while A is mid-flight. Ledger
235 // handles are exclusively locked, so this second surface carries its
236 // own log file cloned from A's header — legal here because a skipped
237 // pass exits at the marker before touching any content or disk.
238 let s_b =
239 vak_session::SessionLog::create(dir.path().join("second-surface.jsonl"), header_b).unwrap();
240 let core_b = core.clone();
241 let hb = tokio::spawn(async move { core_b.reflect_after_turn(&s_b, "all done").await });
242 // ...and must be skipped without its own dispatch.
243 let b = hb.await.unwrap();
244 assert_eq!(
245 b,
246 vak_core::reflection::ReflectionOutcome::Skipped {
247 reason: "already-in-flight"
248 }
249 );
250
251 // Release A; it completes as the single reflected pass.
252 hold.notify_one();
253 let a = ha.await.unwrap();
254 assert_eq!(
255 a,
256 vak_core::reflection::ReflectionOutcome::Reflected {
257 notes_added: 1,
258 skills_proposed: true
259 }
260 );
261 assert_eq!(calls.load(Ordering::SeqCst), 1, "no second dispatch ever");
262 let notes = vak_core::memory::list_notes(&core.sessions_home(), core.cwd());
263 assert_eq!(notes.len(), 1, "one note total — no duplicates");
264}
265
266#[tokio::test]
267async fn provider_error_yields_skipped_not_panic() {
268 let dir = tempfile::tempdir().unwrap();
269 let core = core_in(&dir, "[memory]\nreflection = true\n");
270 let calls = Arc::new(AtomicUsize::new(0));
271 core.set_provider_instance(Arc::new(Counting {
272 calls: calls.clone(),
273 reply: GOOD_REPLY,
274 fail: true,
275 hold: None,
276 }));
277 let session = core.start_session().await.unwrap();
278
279 let out = core.reflect_after_turn(&session, "all done").await;
280
281 assert_eq!(calls.load(Ordering::SeqCst), 1, "dispatch attempted");
282 assert_eq!(
283 out,
284 vak_core::reflection::ReflectionOutcome::Skipped {
285 reason: "reflect-call-failed"
286 }
287 );
288 assert!(vak_core::memory::list_notes(&core.sessions_home(), core.cwd()).is_empty());
289}
290

Reflection Seam

Indexing the workspace…

Reflection Seam