1. 1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2. 2
3. 3use std::collections::VecDeque;
4. 4use std::sync::{Arc, Mutex};
5. 5
6. 6use tokio_util::sync::CancellationToken;
7. 7
8. 8use vak_core::Core;
9. 9use vak_llm as _;
10. 10use vak_llm::stream;
11. 11use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, Usage};
12. 12use vak_llm::{EventStream, LlmError, Provider}; // keep types import path stable for future edits
13. 13
14. 14struct Scripted {
15. 15 responses: Mutex<VecDeque<AssistantMessage>>,
16. 16}
17. 17
18. 18#[async_trait::async_trait]
19. 19impl Provider for Scripted {
20. 20 fn name(&self) -> &str {
21. 21 "scripted"
22. 22 }
23. 23
24. 24 async fn stream(
25. 25 &self,
26. 26 _request: ChatRequest,
27. 27 _cancel: CancellationToken,
28. 28 ) -> Result<EventStream, LlmError> {
29. 29 let next = self.responses.lock().unwrap().pop_front();
30. 30 let (mut sink, rx) = stream::channel(64);
31. 31 match next {
32. 32 Some(m) => {
33. 33 sink.push(stream::StreamEvent::Start { partial: m.clone() });
34. 34 sink.close_message(m).await;
35. 35 }
36. 36 None => sink.close_error(LlmError::Parse("exhausted".into())).await,
37. 37 }
38. 38 Ok(rx)
39. 39 }
40. 40}
41. 41
42. 42fn text(t: &str) -> AssistantMessage {
43. 43 AssistantMessage {
44. 44 content: vec![ContentBlock::text(t)],
45. 45 stop_reason: vak_llm::types::StopReason::EndTurn,
46. 46 usage: Usage {
47. 47 input_tokens: 7,
48. 48 output_tokens: 3,
49. 49 ..Default::default()
50. 50 },
51. 51 model: "test-model".into(),
52. 52 response_id: None,
53. 53 }
54. 54}
55. 55
56. 56fn tool_call(id: &str, name: &str, input: serde_json::Value) -> AssistantMessage {
57. 57 AssistantMessage {
58. 58 content: vec![ContentBlock::ToolUse {
59. 59 id: id.into(),
60. 60 name: name.into(),
61. 61 input,
62. 62 }],
63. 63 stop_reason: vak_llm::types::StopReason::ToolUse,
64. 64 usage: Usage::default(),
65. 65 model: "test-model".into(),
66. 66 response_id: None,
67. 67 }
68. 68}
69. 69
70. 70#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
71. 71async fn remember_propose_recall_promote_loop() {
72. 72 let dir = tempfile::tempdir().unwrap();
73. 73 let home = dir.path().join("home");
74. 74 let cwd = dir.path().to_path_buf();
75. 75
76. 76 vak_config::paths::isolate_home_for_tests();
77. 77 let core = Core::new(cwd.clone()).unwrap();
78. 78 core.set_sessions_home(home.clone());
79. 79 core.set_permission_mode(vak_config::PermissionMode::WorkspaceWrite);
80. 80 core.set_provider_instance(Arc::new(Scripted {
81. 81 responses: Mutex::new(VecDeque::from(vec![
82. 82 // Turn 1 of run A: the model journals a decision and drafts a skill.
83. 83 tool_call(
84. 84 "r1",
85. 85 "remember",
86. 86 serde_json::json!({
87. 87 "note": "the deploy script must pause before rollback windows",
88. 88 "kind": "decision",
89. 89 "tag": "deploy-rollbacks"
90. 90 }),
91. 91 ),
92. 92 tool_call(
93. 93 "p1",
94. 94 "propose_skill",
95. 95 serde_json::json!({
96. 96 "name": "deploy-safely",
97. 97 "description": "Run deploys with rollback pauses",
98. 98 "instructions": "1. read scripts/deploy.sh\n2. pause before rollback windows"
99. 99 }),
100. 100 ),
101. 101 text("Noted both."),
102. 102 ])),
103. 103 }));
104. 104
105. 105 // ---- Run A: remember + propose --------------------------------------
106. 106 let s_a = core.start_session().await.unwrap();
107. 107 let sid_a = s_a.header().unwrap().session_id.clone();
108. 108 let (tx_a, _rx_a) = tokio::sync::mpsc::channel(256);
109. 109 let (outcome_a, log_a) = core
110. 110 .run_turn_with(
111. 111 s_a,
112. 112 "save what we learned",
113. 113 CancellationToken::new(),
114. 114 None,
115. 115 None,
116. 116 None,
117. 117 tx_a,
118. 118 )
119. 119 .await
120. 120 .unwrap();
121. 121 assert!(matches!(
122. 122 outcome_a,
123. 123 vak_agent::TurnOutcome::Completed { .. }
124. 124 ));
125. 125
126. 126 // Notes landed in this Agent's own memory, with provenance...
127. 127 let notes = vak_core::memory::list_notes(&core.sessions_home(), &cwd);
128. 128 assert_eq!(notes.len(), 1);
129. 129 assert_eq!(notes[0].kind, "decision");
130. 130 assert_eq!(notes[0].tag, "deploy-rollbacks");
131. 131 assert_eq!(notes[0].session_id, sid_a);
132. 132
133. 133 // ...and the confirmations are logged on the ledger (invariant 1).
134. 134 let logged = log_a
135. 135 .message_chain()
136. 136 .iter()
137. 137 .map(|(_, m)| {
138. 138 m.content
139. 139 .iter()
140. 140 .filter_map(|b| match b {
141. 141 ContentBlock::ToolResult { content, .. } => Some(content.clone()),
142. 142 _ => None,
143. 143 })
144. 144 .collect::<Vec<_>>()
145. 145 .join("\n")
146. 146 })
147. 147 .collect::<Vec<_>>()
148. 148 .join("\n");
149. 149 assert!(logged.contains("remembered"), "{logged}");
150. 150 assert!(logged.contains("queued for review"), "{logged}");
151. 151
152. 152 // Proposal is pending, not installed.
153. 153 let pending = vak_core::learning::list_proposals(&home, &cwd);
154. 154 assert_eq!(pending.len(), 1);
155. 155 assert_eq!(pending[0].name, "deploy-safely");
156. 156 assert!(!home.join("skills/deploy-safely/SKILL.md").exists());
157. 157
158. 158 // ---- Run B: recall ranks memory first --------------------------------
159. 159 core.set_provider_instance(Arc::new(Scripted {
160. 160 responses: Mutex::new(VecDeque::from(vec![
161. 161 tool_call(
162. 162 "s1",
163. 163 "session_search",
164. 164 serde_json::json!({"query": "deploy rollback"}),
165. 165 ),
166. 166 text("Recalled from memory."),
167. 167 ])),
168. 168 }));
169. 169 let s_b = core.start_session().await.unwrap();
170. 170 let (_tx_b, _rx_b) = tokio::sync::mpsc::channel::<vak_agent::AgentEvent>(256);
171. 171 let (tx_b2, _rx_b2) = tokio::sync::mpsc::channel(256);
172. 172 let _ = (&_tx_b, &_rx_b);
173. 173 let (outcome_b, log_b) = core
174. 174 .run_turn_with(
175. 175 s_b,
176. 176 "what do we know about deploys?",
177. 177 CancellationToken::new(),
178. 178 None,
179. 179 None,
180. 180 None,
181. 181 tx_b2,
182. 182 )
183. 183 .await
184. 184 .unwrap();
185. 185 assert!(matches!(
186. 186 outcome_b,
187. 187 vak_agent::TurnOutcome::Completed { .. }
188. 188 ));
189. 189 let search_output = log_b
190. 190 .message_chain()
191. 191 .iter()
192. 192 .flat_map(|(_, m)| m.content.iter())
193. 193 .filter_map(|b| match b {
194. 194 ContentBlock::ToolResult { content, .. } => Some(content.clone()),
195. 195 _ => None,
196. 196 })
197. 197 .find(|c| c.contains("hit(s)"))
198. 198 .expect("search result logged");
199. 199 assert!(
200. 200 search_output.contains("deploy-rollbacks"),
201. 201 "memory hit present: {search_output}"
202. 202 );
203. 203 assert!(
204. 204 search_output.contains("memory"),
205. 205 "role surfaces as memory: {search_output}"
206. 206 );
207. 207
208. 208 // ---- Promotion closes the loop ---------------------------------------
209. 209 let id = pending[0].id.clone();
210. 210 let promoted = vak_core::learning::promote(&home, &cwd, &id).unwrap();
211. 211 assert_eq!(promoted, "deploy-safely");
212. 212 assert!(home.join("skills/deploy-safely/SKILL.md").exists());
213. 213 assert!(vak_core::learning::list_proposals(&home, &cwd).is_empty());
214. 214 let discovered = vak_core::skills::discover(&cwd, &home);
215. 215 assert!(
216. 216 discovered
217. 217 .iter()
218. 218 .any(|s| s.name == "deploy-safely" && s.description.contains("rollback")),
219. 219 "promoted skill enters discovery: {discovered:?}"
220. 220 );
221. 221
222. 222 // Promoting twice or promoting after rejection fails cleanly.
223. 223 assert!(vak_core::learning::promote(&home, &cwd, &id).is_err());
224. 224}
225. 225