1. 1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2. 2
3. 3use tokio_util::sync::CancellationToken;
4. 4
5. 5use vak_llm::Provider;
6. 6use vak_llm::error::LlmError;
7. 7use vak_llm::openai_responses::{OpenAiResponsesConfig, OpenAiResponsesProvider, build_body};
8. 8use vak_llm::stream::StreamEvent;
9. 9use vak_llm::types::{ChatRequest, ContentBlock, Message, Role, StopReason, ToolDefinition};
10. 10
11. 11fn sample_request() -> ChatRequest {
12. 12 let mut req = ChatRequest::new("gpt-5.6");
13. 13 req.system = Some("You are a coding agent.".into());
14. 14 req.messages = vec![
15. 15 Message::user_text("list the files"),
16. 16 Message::assistant(vec![
17. 17 ContentBlock::text("Checking."),
18. 18 ContentBlock::ToolUse {
19. 19 id: "call_abc".into(),
20. 20 name: "bash".into(),
21. 21 input: serde_json::json!({"command": "ls"}),
22. 22 },
23. 23 ]),
24. 24 Message {
25. 25 role: Role::User,
26. 26 content: vec![ContentBlock::tool_result("call_abc", "src\nREADME.md")],
27. 27 },
28. 28 ];
29. 29 req.tools = vec![ToolDefinition::new(
30. 30 "bash",
31. 31 "Run a shell command",
32. 32 serde_json::json!({"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}),
33. 33 )];
34. 34 req
35. 35}
36. 36
37. 37#[test]
38. 38fn body_maps_neutral_history_to_responses_shape() {
39. 39 let body = build_body(&OpenAiResponsesConfig::default(), &sample_request()).unwrap();
40. 40 assert_eq!(body["model"], "gpt-5.6");
41. 41 assert_eq!(body["stream"], true);
42. 42 assert_eq!(body["instructions"], "You are a coding agent.");
43. 43
44. 44 let input = body["input"].as_array().unwrap();
45. 45 // user msg, function_call item, assistant text item, function_call_output
46. 46 assert_eq!(input.len(), 4);
47. 47 assert_eq!(input[0]["role"], "user");
48. 48 assert_eq!(input[0]["content"][0]["type"], "input_text");
49. 49
50. 50 assert_eq!(input[1]["type"], "function_call");
51. 51 assert_eq!(
52. 52 input[1]["call_id"], "call_abc",
53. 53 "ids survive provider switches"
54. 54 );
55. 55 assert_eq!(input[1]["name"], "bash");
56. 56 let args = input[1]["arguments"].as_str().unwrap();
57. 57 assert!(serde_json::from_str::<serde_json::Value>(args).is_ok());
58. 58
59. 59 assert_eq!(input[2]["role"], "assistant");
60. 60 assert_eq!(input[2]["content"][0]["type"], "output_text");
61. 61
62. 62 assert_eq!(input[3]["type"], "function_call_output");
63. 63 assert_eq!(input[3]["call_id"], "call_abc");
64. 64
65. 65 let tools = body["tools"].as_array().unwrap();
66. 66 assert_eq!(tools[0]["type"], "function");
67. 67 assert_eq!(tools[0]["name"], "bash");
68. 68}
69. 69
70. 70const FIXTURE_STREAM: &str = "\
71. 71event: response.output_text.delta\n\
72. 72data: {\"type\":\"response.output_text.delta\",\"delta\":\"Read\"}\n\
73. 73\n\
74. 74event: response.output_text.delta\n\
75. 75data: {\"type\":\"response.output_text.delta\",\"delta\":\"ing now.\"}\n\
76. 76\n\
77. 77event: response.output_item.added\n\
78. 78data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"function_call\",\"id\":\"fc_1\",\"call_id\":\"call_9\",\"name\":\"read\",\"arguments\":\"\"}}\n\
79. 79\n\
80. 80event: response.function_call_arguments.delta\n\
81. 81data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"{\\\"path\\\":\"}\n\
82. 82\n\
83. 83event: response.function_call_arguments.delta\n\
84. 84data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"\\\"a.txt\\\"}\"}\n\
85. 85\n\
86. 86event: response.completed\n\
87. 87data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_123\",\"usage\":{\"input_tokens\":33,\"output_tokens\":9,\"input_tokens_details\":{\"cached_tokens\":4}}}}\n\
88. 88\n";
89. 89
90. 90#[tokio::test]
91. 91async fn full_stream_accumulates_text_and_tool_calls() {
92. 92 let provider = OpenAiResponsesProvider::new(OpenAiResponsesConfig {
93. 93 api_key: "k".into(),
94. 94 base_url: mock_url(FIXTURE_STREAM).await,
95. 95 ..Default::default()
96. 96 })
97. 97 .unwrap();
98. 98
99. 99 let mut es = provider
100. 100 .stream(sample_request(), CancellationToken::new())
101. 101 .await
102. 102 .unwrap();
103. 103 let mut text = String::new();
104. 104 while let Some(ev) = futures::StreamExt::next(&mut es).await {
105. 105 if let StreamEvent::TextDelta { delta, .. } = ev {
106. 106 text.push_str(&delta);
107. 107 }
108. 108 }
109. 109 let msg = es.result().await.unwrap();
110. 110
111. 111 assert_eq!(text, "Reading now.");
112. 112 // A function_call item means the model wants tools — the loop must
113. 113 // continue even though response.completed arrived.
114. 114 assert_eq!(msg.stop_reason, StopReason::ToolUse);
115. 115 // The fixture's `input_tokens: 33` already includes the 4 cached
116. 116 // tokens (Responses API semantics), so the normalized, non-cached
117. 117 // `input_tokens` is 29 — the split, not the raw total.
118. 118 assert_eq!(msg.usage.input_tokens, 29);
119. 119 assert_eq!(msg.usage.output_tokens, 9);
120. 120 assert_eq!(msg.usage.cache_read_input_tokens, Some(4));
121. 121 // The provider's original total is reconstructible from the split.
122. 122 assert_eq!(msg.usage.prompt_tokens(), 33);
123. 123 assert_eq!(msg.response_id, Some("resp_123".to_string()));
124. 124
125. 125 let calls: Vec<&ContentBlock> = msg
126. 126 .content
127. 127 .iter()
128. 128 .filter(|b| matches!(b, ContentBlock::ToolUse { .. }))
129. 129 .collect();
130. 130 assert_eq!(calls.len(), 1);
131. 131 if let ContentBlock::ToolUse { id, name, input } = calls[0] {
132. 132 assert_eq!(id, "call_9");
133. 133 assert_eq!(name, "read");
134. 134 assert_eq!(input, &serde_json::json!({"path": "a.txt"}));
135. 135 } else {
136. 136 unreachable!()
137. 137 }
138. 138}
139. 139
140. 140const FIXTURE_FULL_CACHE_HIT: &str = "\
141. 141event: response.completed\n\
142. 142data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_456\",\"usage\":{\"input_tokens\":12000,\"output_tokens\":50,\"input_tokens_details\":{\"cached_tokens\":12000}}}}\n\
143. 143\n";
144. 144
145. 145#[tokio::test]
146. 146async fn a_full_cache_hit_reports_zero_fresh_input_tokens_not_the_raw_total() {
147. 147 let provider = OpenAiResponsesProvider::new(OpenAiResponsesConfig {
148. 148 api_key: "k".into(),
149. 149 base_url: mock_url(FIXTURE_FULL_CACHE_HIT).await,
150. 150 ..Default::default()
151. 151 })
152. 152 .unwrap();
153. 153 let mut req = ChatRequest::new("m");
154. 154 req.messages = vec![Message::user_text("hi")];
155. 155 let mut es = provider
156. 156 .stream(req, CancellationToken::new())
157. 157 .await
158. 158 .unwrap();
159. 159 while futures::StreamExt::next(&mut es).await.is_some() {}
160. 160 let msg = es.result().await.unwrap();
161. 161 assert_eq!(msg.usage.input_tokens, 0);
162. 162 assert_eq!(msg.usage.cache_read_input_tokens, Some(12_000));
163. 163 assert_eq!(msg.usage.prompt_tokens(), 12_000);
164. 164}
165. 165
166. 166const FIXTURE_INCOMPLETE: &str = "\
167. 167data: {\"type\":\"response.incomplete\",\"response\":{\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\
168. 168\n";
169. 169
170. 170#[tokio::test]
171. 171async fn incomplete_maps_to_max_tokens() {
172. 172 let provider = OpenAiResponsesProvider::new(OpenAiResponsesConfig {
173. 173 api_key: "k".into(),
174. 174 base_url: mock_url(FIXTURE_INCOMPLETE).await,
175. 175 ..Default::default()
176. 176 })
177. 177 .unwrap();
178. 178 let mut req = ChatRequest::new("m");
179. 179 req.messages = vec![Message::user_text("hi")];
180. 180 let mut es = provider
181. 181 .stream(req, CancellationToken::new())
182. 182 .await
183. 183 .unwrap();
184. 184 while futures::StreamExt::next(&mut es).await.is_some() {}
185. 185 let msg = es.result().await.unwrap();
186. 186 assert_eq!(msg.stop_reason, StopReason::MaxTokens);
187. 187}
188. 188
189. 189#[tokio::test]
190. 190async fn http_error_maps_to_typed_value() {
191. 191 let url = error_url(401, r#"{"error":{"message":"bad key"}}"#).await;
192. 192 let provider = OpenAiResponsesProvider::new(OpenAiResponsesConfig {
193. 193 api_key: "bad".into(),
194. 194 base_url: url,
195. 195 ..Default::default()
196. 196 })
197. 197 .unwrap();
198. 198 let err = provider
199. 199 .stream(ChatRequest::new("m"), CancellationToken::new())
200. 200 .await
201. 201 .err()
202. 202 .expect("expected error");
203. 203 assert!(matches!(err, LlmError::Auth(_)), "got {err:?}");
204. 204}
205. 205
206. 206async fn spawn_server(body: Vec<u8>) -> String {
207. 207 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
208. 208 let addr = listener.local_addr().unwrap();
209. 209 tokio::spawn(async move {
210. 210 let (mut sock, _) = listener.accept().await.unwrap();
211. 211 drain_headers(&mut sock).await;
212. 212 use tokio::io::AsyncWriteExt;
213. 213 let _ = sock.write_all(&body).await;
214. 214 let _ = sock.flush().await;
215. 215 let _ = sock.shutdown().await;
216. 216 });
217. 217 format!("http://{addr}")
218. 218}
219. 219
220. 220async fn drain_headers<S>(sock: &mut S)
221. 221where
222. 222 S: tokio::io::AsyncRead + Unpin,
223. 223{
224. 224 use tokio::io::AsyncReadExt;
225. 225 let mut buf = Vec::new();
226. 226 let mut chunk = [0u8; 1024];
227. 227 loop {
228. 228 let n = sock.read(&mut chunk).await.unwrap_or(0);
229. 229 if n == 0 {
230. 230 return;
231. 231 }
232. 232 buf.extend_from_slice(&chunk[..n]);
233. 233 if buf.windows(4).any(|w| w == b"\r\n\r\n") {
234. 234 return;
235. 235 }
236. 236 }
237. 237}
238. 238
239. 239async fn mock_url(sse_body: &str) -> String {
240. 240 let owned = format!(
241. 241 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\nconnection: close\r\n\r\n{sse_body}"
242. 242 );
243. 243 spawn_server(owned.into_bytes()).await
244. 244}
245. 245
246. 246async fn error_url(status: u16, json: &str) -> String {
247. 247 let owned = format!(
248. 248 "HTTP/1.1 {status} Err\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n{json}"
249. 249 );
250. 250 spawn_server(owned.into_bytes()).await
251. 251}
252. 252