Vakyartha documentation

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3use tokio_util::sync::CancellationToken;
4
5use vak_llm::Provider;
6use vak_llm::error::LlmError;
7use vak_llm::openai::{OpenAiCompletionsProvider, OpenAiConfig, build_body};
8use vak_llm::stream::StreamEvent;
9use vak_llm::types::{ChatRequest, ContentBlock, Message, Role, StopReason, ToolDefinition};
10
11fn sample_request() -> ChatRequest {
12 let mut req = ChatRequest::new("gpt-5.6");
13 req.system = Some("You are a coding agent.".into());
14 req.messages = vec![
15 Message::user_text("list the files"),
16 Message::assistant(vec![
17 ContentBlock::Thinking {
18 text: "hmm".into(),
19 signature: Some("sig".into()),
20 },
21 ContentBlock::text("Let me check."),
22 ContentBlock::ToolUse {
23 id: "toolu_abc".into(),
24 name: "bash".into(),
25 input: serde_json::json!({"command": "ls"}),
26 },
27 ]),
28 Message {
29 role: Role::User,
30 content: vec![ContentBlock::tool_result("toolu_abc", "src\nREADME.md")],
31 },
32 Message::user_text("and now?"),
33 ];
34 req.tools = vec![ToolDefinition::new(
35 "bash",
36 "Run a shell command",
37 serde_json::json!({"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}),
38 )];
39 req
40}
41
42#[test]
43fn body_maps_neutral_history_to_openai_shape() {
44 let body = build_body(&OpenAiConfig::default(), &sample_request()).unwrap();
45 assert_eq!(body["model"], "gpt-5.6");
46 assert_eq!(body["stream"], true);
47 assert_eq!(body["stream_options"]["include_usage"], true);
48
49 let msgs = body["messages"].as_array().unwrap();
50 assert_eq!(msgs[0]["role"], "system");
51 assert_eq!(msgs[1]["role"], "user");
52 assert_eq!(msgs[1]["content"], "list the files");
53
54 let assistant = &msgs[2];
55 assert_eq!(assistant["role"], "assistant");
56 assert!(
57 assistant.get("thinking").is_none(),
58 "thinking blocks must not leak into openai payloads"
59 );
60 let calls = assistant["tool_calls"].as_array().unwrap();
61 assert_eq!(
62 calls[0]["id"], "toolu_abc",
63 "tool ids survive provider switches"
64 );
65 assert_eq!(calls[0]["function"]["name"], "bash");
66 let args = calls[0]["function"]["arguments"].as_str().unwrap();
67 assert!(
68 serde_json::from_str::<serde_json::Value>(args).is_ok(),
69 "arguments must be a JSON string"
70 );
71
72 let tool_msg = &msgs[3];
73 assert_eq!(tool_msg["role"], "tool");
74 assert_eq!(tool_msg["tool_call_id"], "toolu_abc");
75 assert_eq!(tool_msg["content"], "src\nREADME.md");
76
77 assert_eq!(msgs[4]["role"], "user");
78 assert_eq!(msgs[4]["content"], "and now?");
79
80 let tools = body["tools"].as_array().unwrap();
81 assert_eq!(tools[0]["type"], "function");
82 assert_eq!(tools[0]["function"]["name"], "bash");
83}
84
85#[test]
86fn user_tool_results_become_separate_tool_messages() {
87 let mut req = ChatRequest::new("m");
88 req.messages = vec![Message {
89 role: Role::User,
90 content: vec![
91 ContentBlock::text("results below"),
92 ContentBlock::tool_result("a", "one"),
93 ContentBlock::tool_result("b", "two"),
94 ],
95 }];
96 let msgs = build_body(&OpenAiConfig::default(), &req).unwrap()["messages"]
97 .as_array()
98 .unwrap()
99 .clone();
100 assert_eq!(msgs.len(), 3);
101 assert_eq!(msgs[0]["role"], "tool");
102 assert_eq!(msgs[1]["role"], "tool");
103 assert_eq!(msgs[1]["tool_call_id"], "b");
104 assert_eq!(msgs[2]["role"], "user");
105}
106
107const FIXTURE_TOOL_STREAM: &str = "\
108data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"Checking\"},\"finish_reason\":null}]}\n\
109\n\
110data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\" now.\"},\"finish_reason\":null}]}\n\
111\n\
112data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_9\",\"type\":\"function\",\"function\":{\"name\":\"read\",\"arguments\":\"\"}}]},\"finish_reason\":null}]}\n\
113\n\
114data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"path\\\":\"}}]},\"finish_reason\":null}]}\n\
115\n\
116data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"\\\"a.txt\\\"}\"}}]},\"finish_reason\":null}]}\n\
117\n\
118data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\
119\n\
120data: {\"choices\":[],\"usage\":{\"prompt_tokens\":11,\"completion_tokens\":7}}\n\
121\n\
122data: [DONE]\n\
123\n";
124
125#[tokio::test]
126async fn full_stream_accumulates_text_and_tool_calls() {
127 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
128 api_key: "k".into(),
129 base_url: mock_url(FIXTURE_TOOL_STREAM).await,
130 ..Default::default()
131 })
132 .unwrap();
133
134 let mut es = provider
135 .stream(sample_request(), CancellationToken::new())
136 .await
137 .unwrap();
138 let mut text = String::new();
139 while let Some(ev) = futures::StreamExt::next(&mut es).await {
140 if let StreamEvent::TextDelta { delta, .. } = ev {
141 text.push_str(&delta);
142 }
143 }
144 let msg = es.result().await.unwrap();
145
146 assert_eq!(text, "Checking now.");
147 assert_eq!(msg.stop_reason, StopReason::ToolUse);
148 assert_eq!(msg.usage.input_tokens, 11);
149 assert_eq!(msg.usage.output_tokens, 7);
150
151 let calls: Vec<&ContentBlock> = msg
152 .content
153 .iter()
154 .filter(|b| matches!(b, ContentBlock::ToolUse { .. }))
155 .collect();
156 assert_eq!(calls.len(), 1);
157 if let ContentBlock::ToolUse { id, name, input } = calls[0] {
158 assert_eq!(id, "call_9");
159 assert_eq!(name, "read");
160 assert_eq!(input, &serde_json::json!({"path": "a.txt"}));
161 } else {
162 unreachable!()
163 }
164}
165
166// Realistic cache-bearing usage payload (docs/design/68-context-engine.md
167// §1): OpenAI's `prompt_tokens` is the whole prompt, cache hits included —
168// `prompt_tokens_details.cached_tokens` is a SUBSET of it, not an addition.
169const FIXTURE_CACHED_USAGE_STREAM: &str = "\
170data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":null}]}\n\
171\n\
172data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\
173\n\
174data: {\"choices\":[],\"usage\":{\"prompt_tokens\":5000,\"completion_tokens\":120,\"prompt_tokens_details\":{\"cached_tokens\":4800}}}\n\
175\n\
176data: [DONE]\n\
177\n";
178
179#[tokio::test]
180async fn cached_tokens_are_split_out_of_prompt_tokens_not_added_on_top() {
181 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
182 api_key: "k".into(),
183 base_url: mock_url(FIXTURE_CACHED_USAGE_STREAM).await,
184 ..Default::default()
185 })
186 .unwrap();
187 let mut es = provider
188 .stream(sample_request(), CancellationToken::new())
189 .await
190 .unwrap();
191 while futures::StreamExt::next(&mut es).await.is_some() {}
192 let msg = es.result().await.unwrap();
193
194 // 5000 total, 4800 of it served from cache: normalized input_tokens is
195 // only the 200 fresh tokens, never the full 5000.
196 assert_eq!(msg.usage.input_tokens, 200);
197 assert_eq!(msg.usage.cache_read_input_tokens, Some(4800));
198 assert_eq!(msg.usage.output_tokens, 120);
199 // The original provider total is reconstructible from the split.
200 assert_eq!(msg.usage.prompt_tokens(), 5000);
201}
202
203const FIXTURE_STOP_STREAM: &str = "\
204data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"done\"},\"finish_reason\":null}]}\n\
205\n\
206data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\
207\n\
208data: [DONE]\n\
209\n";
210
211#[tokio::test]
212async fn finish_stop_maps_to_end_turn() {
213 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
214 api_key: "k".into(),
215 base_url: mock_url(FIXTURE_STOP_STREAM).await,
216 ..Default::default()
217 })
218 .unwrap();
219 let mut req = ChatRequest::new("m");
220 req.messages = vec![Message::user_text("hi")];
221 let mut es = provider
222 .stream(req, CancellationToken::new())
223 .await
224 .unwrap();
225 while futures::StreamExt::next(&mut es).await.is_some() {}
226 let msg = es.result().await.unwrap();
227 assert_eq!(msg.stop_reason, StopReason::EndTurn);
228 assert_eq!(msg.text_content(), "done");
229}
230
231#[tokio::test]
232async fn http_error_maps_to_typed_value() {
233 let url = error_url(429, r#"{"error":{"message":"quota exceeded"}}"#).await;
234 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
235 api_key: "k".into(),
236 base_url: url,
237 ..Default::default()
238 })
239 .unwrap();
240 let err = provider
241 .stream(ChatRequest::new("m"), CancellationToken::new())
242 .await
243 .err()
244 .expect("expected error");
245 assert!(matches!(err, LlmError::RateLimit { .. }), "got {err:?}");
246}
247
248// Regression (live, OpenCode Zen): proxies may end the body after the last
249// content chunk with neither finish_reason nor [DONE]. Content-bearing clean
250// closes complete; empty ones still fail closed.
251const FIXTURE_TRUNCATED_WITH_CONTENT: &str = "\
252data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"partial answer\"},\"finish_reason\":null}]}\n\
253\n";
254
255const FIXTURE_TRUNCATED_EMPTY: &str = "\
256data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\
257\n";
258
259#[tokio::test]
260async fn clean_close_with_content_completes_as_end_turn() {
261 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
262 api_key: "k".into(),
263 base_url: mock_url(FIXTURE_TRUNCATED_WITH_CONTENT).await,
264 ..Default::default()
265 })
266 .unwrap();
267 let mut req = ChatRequest::new("m");
268 req.messages = vec![Message::user_text("hi")];
269 let mut es = provider
270 .stream(req, CancellationToken::new())
271 .await
272 .unwrap();
273 while futures::StreamExt::next(&mut es).await.is_some() {}
274 let msg = es.result().await.unwrap();
275 assert_eq!(msg.stop_reason, StopReason::EndTurn);
276 assert_eq!(msg.text_content(), "partial answer");
277}
278
279#[tokio::test]
280async fn clean_close_without_content_still_fails_closed() {
281 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
282 api_key: "k".into(),
283 base_url: mock_url(FIXTURE_TRUNCATED_EMPTY).await,
284 ..Default::default()
285 })
286 .unwrap();
287 let mut req = ChatRequest::new("m");
288 req.messages = vec![Message::user_text("hi")];
289 let mut es = provider
290 .stream(req, CancellationToken::new())
291 .await
292 .unwrap();
293 while futures::StreamExt::next(&mut es).await.is_some() {}
294 let err = es.result().await.expect_err("expected parse error");
295 assert!(
296 matches!(err, LlmError::Parse(ref m) if m.contains("finish_reason")),
297 "got {err:?}"
298 );
299}
300
301async fn spawn_server(body: Vec<u8>) -> String {
302 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
303 let addr = listener.local_addr().unwrap();
304 tokio::spawn(async move {
305 let (mut sock, _) = listener.accept().await.unwrap();
306 drain_headers(&mut sock).await;
307 use tokio::io::AsyncWriteExt;
308 let _ = sock.write_all(&body).await;
309 let _ = sock.flush().await;
310 let _ = sock.shutdown().await;
311 });
312 format!("http://{addr}")
313}
314
315async fn drain_headers<S>(sock: &mut S)
316where
317 S: tokio::io::AsyncRead + Unpin,
318{
319 use tokio::io::AsyncReadExt;
320 let mut buf = Vec::new();
321 let mut chunk = [0u8; 1024];
322 loop {
323 let n = sock.read(&mut chunk).await.unwrap_or(0);
324 if n == 0 {
325 return;
326 }
327 buf.extend_from_slice(&chunk[..n]);
328 if buf.windows(4).any(|w| w == b"\r\n\r\n") {
329 return;
330 }
331 }
332}
333
334async fn mock_url(sse_body: &str) -> String {
335 let owned = format!(
336 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\nconnection: close\r\n\r\n{sse_body}"
337 );
338 spawn_server(owned.into_bytes()).await
339}
340
341async fn error_url(status: u16, json: &str) -> String {
342 let owned = format!(
343 "HTTP/1.1 {status} Err\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n{json}"
344 );
345 spawn_server(owned.into_bytes()).await
346}
347
348// Some OpenAI-compatible endpoints (observed on opencode-zen's free tier)
349// close tool-call turns with a finish_reason outside the canonical set —
350// e.g. plain "stop". The accumulated tool_use blocks are ground truth: the
351// loop must see StopReason::ToolUse or a dangling call kills the run.
352const FIXTURE_TOOL_STREAM_BAD_FINISH: &str = "\
353data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\
354\n\
355data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"glob\",\"arguments\":\"{\\\"pattern\\\":\\\"**/*.md\\\"}\"}}]},\"finish_reason\":null}]}\n\
356\n\
357data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\
358\n\
359data: [DONE]\n\
360\n";
361
362#[tokio::test]
363async fn nonstandard_finish_reason_with_tool_use_still_yields_tooluse() {
364 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
365 api_key: "k".into(),
366 base_url: mock_url(FIXTURE_TOOL_STREAM_BAD_FINISH).await,
367 ..Default::default()
368 })
369 .unwrap();
370
371 let mut req = ChatRequest::new("m");
372 req.messages = vec![Message::user_text("list markdown files")];
373
374 let mut es = provider
375 .stream(req, CancellationToken::new())
376 .await
377 .unwrap();
378 let mut end = None;
379 while let Some(ev) = futures::StreamExt::next(&mut es).await {
380 if let StreamEvent::End { message } = ev {
381 end = Some(message);
382 }
383 }
384 let msg = end.expect("stream must terminate with End");
385 assert_eq!(msg.stop_reason, StopReason::ToolUse);
386 assert!(
387 msg.content
388 .iter()
389 .any(|b| matches!(b, ContentBlock::ToolUse { name, .. } if name == "glob")),
390 "tool_use block preserved"
391 );
392}
393
394// Same endpoint family, worse: body ends right after the tool-call deltas
395// with NO finish_reason and NO [DONE]. The clean-close path must also
396// promote accumulated tool_use to StopReason::ToolUse.
397const FIXTURE_TOOL_STREAM_EOF_NO_FINISH: &str = "\
398data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\
399\n\
400data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_2\",\"type\":\"function\",\"function\":{\"name\":\"glob\",\"arguments\":\"{\\\"pattern\\\":\\\"*\\\"}\"}}]},\"finish_reason\":null}]}\n\
401\n";
402
403#[tokio::test]
404async fn clean_close_after_tool_calls_yields_tooluse() {
405 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
406 api_key: "k".into(),
407 base_url: mock_url(FIXTURE_TOOL_STREAM_EOF_NO_FINISH).await,
408 ..Default::default()
409 })
410 .unwrap();
411
412 let mut req = ChatRequest::new("m");
413 req.messages = vec![Message::user_text("list files")];
414
415 let es = provider
416 .stream(req, CancellationToken::new())
417 .await
418 .unwrap();
419 // EOF-close delivers the message through the stream's terminal.
420 let msg = es.result().await.expect("clean close must complete");
421 assert_eq!(msg.stop_reason, StopReason::ToolUse);
422}
423
424// The live culprit (opencode-zen free tier): reasoning deltas, one
425// tool_calls delta, then bare `data: [DONE]` — no finish_reason frame ever.
426const FIXTURE_DONE_WITHOUT_FINISH: &str = "\
427data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"reasoning_content\":\"thinking\"},\"finish_reason\":null}]}\n\
428\n\
429data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_3\",\"type\":\"function\",\"function\":{\"name\":\"glob\",\"arguments\":\"{\\\"pattern\\\":\\\"*\\\"}\"}}]},\"finish_reason\":null}]}\n\
430\n\
431data: [DONE]\n\
432\n";
433
434#[tokio::test]
435async fn done_without_finish_reason_still_yields_tooluse() {
436 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
437 api_key: "k".into(),
438 base_url: mock_url(FIXTURE_DONE_WITHOUT_FINISH).await,
439 ..Default::default()
440 })
441 .unwrap();
442
443 let mut req = ChatRequest::new("m");
444 req.messages = vec![Message::user_text("list files")];
445
446 let es = provider
447 .stream(req, CancellationToken::new())
448 .await
449 .unwrap();
450 let msg = es.result().await.expect("[DONE] close must complete");
451 assert_eq!(msg.stop_reason, StopReason::ToolUse);
452}
453

1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
2
3use tokio_util::sync::CancellationToken;
4
5use vak_llm::Provider;
6use vak_llm::error::LlmError;
7use vak_llm::openai::{OpenAiCompletionsProvider, OpenAiConfig, build_body};
8use vak_llm::stream::StreamEvent;
9use vak_llm::types::{ChatRequest, ContentBlock, Message, Role, StopReason, ToolDefinition};
10
11fn sample_request() -> ChatRequest {
12 let mut req = ChatRequest::new("gpt-5.6");
13 req.system = Some("You are a coding agent.".into());
14 req.messages = vec![
15 Message::user_text("list the files"),
16 Message::assistant(vec![
17 ContentBlock::Thinking {
18 text: "hmm".into(),
19 signature: Some("sig".into()),
20 },
21 ContentBlock::text("Let me check."),
22 ContentBlock::ToolUse {
23 id: "toolu_abc".into(),
24 name: "bash".into(),
25 input: serde_json::json!({"command": "ls"}),
26 },
27 ]),
28 Message {
29 role: Role::User,
30 content: vec![ContentBlock::tool_result("toolu_abc", "src\nREADME.md")],
31 },
32 Message::user_text("and now?"),
33 ];
34 req.tools = vec![ToolDefinition::new(
35 "bash",
36 "Run a shell command",
37 serde_json::json!({"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}),
38 )];
39 req
40}
41
42#[test]
43fn body_maps_neutral_history_to_openai_shape() {
44 let body = build_body(&OpenAiConfig::default(), &sample_request()).unwrap();
45 assert_eq!(body["model"], "gpt-5.6");
46 assert_eq!(body["stream"], true);
47 assert_eq!(body["stream_options"]["include_usage"], true);
48
49 let msgs = body["messages"].as_array().unwrap();
50 assert_eq!(msgs[0]["role"], "system");
51 assert_eq!(msgs[1]["role"], "user");
52 assert_eq!(msgs[1]["content"], "list the files");
53
54 let assistant = &msgs[2];
55 assert_eq!(assistant["role"], "assistant");
56 assert!(
57 assistant.get("thinking").is_none(),
58 "thinking blocks must not leak into openai payloads"
59 );
60 let calls = assistant["tool_calls"].as_array().unwrap();
61 assert_eq!(
62 calls[0]["id"], "toolu_abc",
63 "tool ids survive provider switches"
64 );
65 assert_eq!(calls[0]["function"]["name"], "bash");
66 let args = calls[0]["function"]["arguments"].as_str().unwrap();
67 assert!(
68 serde_json::from_str::<serde_json::Value>(args).is_ok(),
69 "arguments must be a JSON string"
70 );
71
72 let tool_msg = &msgs[3];
73 assert_eq!(tool_msg["role"], "tool");
74 assert_eq!(tool_msg["tool_call_id"], "toolu_abc");
75 assert_eq!(tool_msg["content"], "src\nREADME.md");
76
77 assert_eq!(msgs[4]["role"], "user");
78 assert_eq!(msgs[4]["content"], "and now?");
79
80 let tools = body["tools"].as_array().unwrap();
81 assert_eq!(tools[0]["type"], "function");
82 assert_eq!(tools[0]["function"]["name"], "bash");
83}
84
85#[test]
86fn user_tool_results_become_separate_tool_messages() {
87 let mut req = ChatRequest::new("m");
88 req.messages = vec![Message {
89 role: Role::User,
90 content: vec![
91 ContentBlock::text("results below"),
92 ContentBlock::tool_result("a", "one"),
93 ContentBlock::tool_result("b", "two"),
94 ],
95 }];
96 let msgs = build_body(&OpenAiConfig::default(), &req).unwrap()["messages"]
97 .as_array()
98 .unwrap()
99 .clone();
100 assert_eq!(msgs.len(), 3);
101 assert_eq!(msgs[0]["role"], "tool");
102 assert_eq!(msgs[1]["role"], "tool");
103 assert_eq!(msgs[1]["tool_call_id"], "b");
104 assert_eq!(msgs[2]["role"], "user");
105}
106
107const FIXTURE_TOOL_STREAM: &str = "\
108data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"Checking\"},\"finish_reason\":null}]}\n\
109\n\
110data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\" now.\"},\"finish_reason\":null}]}\n\
111\n\
112data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_9\",\"type\":\"function\",\"function\":{\"name\":\"read\",\"arguments\":\"\"}}]},\"finish_reason\":null}]}\n\
113\n\
114data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"path\\\":\"}}]},\"finish_reason\":null}]}\n\
115\n\
116data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"\\\"a.txt\\\"}\"}}]},\"finish_reason\":null}]}\n\
117\n\
118data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\
119\n\
120data: {\"choices\":[],\"usage\":{\"prompt_tokens\":11,\"completion_tokens\":7}}\n\
121\n\
122data: [DONE]\n\
123\n";
124
125#[tokio::test]
126async fn full_stream_accumulates_text_and_tool_calls() {
127 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
128 api_key: "k".into(),
129 base_url: mock_url(FIXTURE_TOOL_STREAM).await,
130 ..Default::default()
131 })
132 .unwrap();
133
134 let mut es = provider
135 .stream(sample_request(), CancellationToken::new())
136 .await
137 .unwrap();
138 let mut text = String::new();
139 while let Some(ev) = futures::StreamExt::next(&mut es).await {
140 if let StreamEvent::TextDelta { delta, .. } = ev {
141 text.push_str(&delta);
142 }
143 }
144 let msg = es.result().await.unwrap();
145
146 assert_eq!(text, "Checking now.");
147 assert_eq!(msg.stop_reason, StopReason::ToolUse);
148 assert_eq!(msg.usage.input_tokens, 11);
149 assert_eq!(msg.usage.output_tokens, 7);
150
151 let calls: Vec<&ContentBlock> = msg
152 .content
153 .iter()
154 .filter(|b| matches!(b, ContentBlock::ToolUse { .. }))
155 .collect();
156 assert_eq!(calls.len(), 1);
157 if let ContentBlock::ToolUse { id, name, input } = calls[0] {
158 assert_eq!(id, "call_9");
159 assert_eq!(name, "read");
160 assert_eq!(input, &serde_json::json!({"path": "a.txt"}));
161 } else {
162 unreachable!()
163 }
164}
165
166// Realistic cache-bearing usage payload (docs/design/68-context-engine.md
167// §1): OpenAI's `prompt_tokens` is the whole prompt, cache hits included —
168// `prompt_tokens_details.cached_tokens` is a SUBSET of it, not an addition.
169const FIXTURE_CACHED_USAGE_STREAM: &str = "\
170data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":null}]}\n\
171\n\
172data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\
173\n\
174data: {\"choices\":[],\"usage\":{\"prompt_tokens\":5000,\"completion_tokens\":120,\"prompt_tokens_details\":{\"cached_tokens\":4800}}}\n\
175\n\
176data: [DONE]\n\
177\n";
178
179#[tokio::test]
180async fn cached_tokens_are_split_out_of_prompt_tokens_not_added_on_top() {
181 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
182 api_key: "k".into(),
183 base_url: mock_url(FIXTURE_CACHED_USAGE_STREAM).await,
184 ..Default::default()
185 })
186 .unwrap();
187 let mut es = provider
188 .stream(sample_request(), CancellationToken::new())
189 .await
190 .unwrap();
191 while futures::StreamExt::next(&mut es).await.is_some() {}
192 let msg = es.result().await.unwrap();
193
194 // 5000 total, 4800 of it served from cache: normalized input_tokens is
195 // only the 200 fresh tokens, never the full 5000.
196 assert_eq!(msg.usage.input_tokens, 200);
197 assert_eq!(msg.usage.cache_read_input_tokens, Some(4800));
198 assert_eq!(msg.usage.output_tokens, 120);
199 // The original provider total is reconstructible from the split.
200 assert_eq!(msg.usage.prompt_tokens(), 5000);
201}
202
203const FIXTURE_STOP_STREAM: &str = "\
204data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"done\"},\"finish_reason\":null}]}\n\
205\n\
206data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\
207\n\
208data: [DONE]\n\
209\n";
210
211#[tokio::test]
212async fn finish_stop_maps_to_end_turn() {
213 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
214 api_key: "k".into(),
215 base_url: mock_url(FIXTURE_STOP_STREAM).await,
216 ..Default::default()
217 })
218 .unwrap();
219 let mut req = ChatRequest::new("m");
220 req.messages = vec![Message::user_text("hi")];
221 let mut es = provider
222 .stream(req, CancellationToken::new())
223 .await
224 .unwrap();
225 while futures::StreamExt::next(&mut es).await.is_some() {}
226 let msg = es.result().await.unwrap();
227 assert_eq!(msg.stop_reason, StopReason::EndTurn);
228 assert_eq!(msg.text_content(), "done");
229}
230
231#[tokio::test]
232async fn http_error_maps_to_typed_value() {
233 let url = error_url(429, r#"{"error":{"message":"quota exceeded"}}"#).await;
234 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
235 api_key: "k".into(),
236 base_url: url,
237 ..Default::default()
238 })
239 .unwrap();
240 let err = provider
241 .stream(ChatRequest::new("m"), CancellationToken::new())
242 .await
243 .err()
244 .expect("expected error");
245 assert!(matches!(err, LlmError::RateLimit { .. }), "got {err:?}");
246}
247
248// Regression (live, OpenCode Zen): proxies may end the body after the last
249// content chunk with neither finish_reason nor [DONE]. Content-bearing clean
250// closes complete; empty ones still fail closed.
251const FIXTURE_TRUNCATED_WITH_CONTENT: &str = "\
252data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"partial answer\"},\"finish_reason\":null}]}\n\
253\n";
254
255const FIXTURE_TRUNCATED_EMPTY: &str = "\
256data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\
257\n";
258
259#[tokio::test]
260async fn clean_close_with_content_completes_as_end_turn() {
261 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
262 api_key: "k".into(),
263 base_url: mock_url(FIXTURE_TRUNCATED_WITH_CONTENT).await,
264 ..Default::default()
265 })
266 .unwrap();
267 let mut req = ChatRequest::new("m");
268 req.messages = vec![Message::user_text("hi")];
269 let mut es = provider
270 .stream(req, CancellationToken::new())
271 .await
272 .unwrap();
273 while futures::StreamExt::next(&mut es).await.is_some() {}
274 let msg = es.result().await.unwrap();
275 assert_eq!(msg.stop_reason, StopReason::EndTurn);
276 assert_eq!(msg.text_content(), "partial answer");
277}
278
279#[tokio::test]
280async fn clean_close_without_content_still_fails_closed() {
281 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
282 api_key: "k".into(),
283 base_url: mock_url(FIXTURE_TRUNCATED_EMPTY).await,
284 ..Default::default()
285 })
286 .unwrap();
287 let mut req = ChatRequest::new("m");
288 req.messages = vec![Message::user_text("hi")];
289 let mut es = provider
290 .stream(req, CancellationToken::new())
291 .await
292 .unwrap();
293 while futures::StreamExt::next(&mut es).await.is_some() {}
294 let err = es.result().await.expect_err("expected parse error");
295 assert!(
296 matches!(err, LlmError::Parse(ref m) if m.contains("finish_reason")),
297 "got {err:?}"
298 );
299}
300
301async fn spawn_server(body: Vec<u8>) -> String {
302 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
303 let addr = listener.local_addr().unwrap();
304 tokio::spawn(async move {
305 let (mut sock, _) = listener.accept().await.unwrap();
306 drain_headers(&mut sock).await;
307 use tokio::io::AsyncWriteExt;
308 let _ = sock.write_all(&body).await;
309 let _ = sock.flush().await;
310 let _ = sock.shutdown().await;
311 });
312 format!("http://{addr}")
313}
314
315async fn drain_headers<S>(sock: &mut S)
316where
317 S: tokio::io::AsyncRead + Unpin,
318{
319 use tokio::io::AsyncReadExt;
320 let mut buf = Vec::new();
321 let mut chunk = [0u8; 1024];
322 loop {
323 let n = sock.read(&mut chunk).await.unwrap_or(0);
324 if n == 0 {
325 return;
326 }
327 buf.extend_from_slice(&chunk[..n]);
328 if buf.windows(4).any(|w| w == b"\r\n\r\n") {
329 return;
330 }
331 }
332}
333
334async fn mock_url(sse_body: &str) -> String {
335 let owned = format!(
336 "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\nconnection: close\r\n\r\n{sse_body}"
337 );
338 spawn_server(owned.into_bytes()).await
339}
340
341async fn error_url(status: u16, json: &str) -> String {
342 let owned = format!(
343 "HTTP/1.1 {status} Err\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n{json}"
344 );
345 spawn_server(owned.into_bytes()).await
346}
347
348// Some OpenAI-compatible endpoints (observed on opencode-zen's free tier)
349// close tool-call turns with a finish_reason outside the canonical set —
350// e.g. plain "stop". The accumulated tool_use blocks are ground truth: the
351// loop must see StopReason::ToolUse or a dangling call kills the run.
352const FIXTURE_TOOL_STREAM_BAD_FINISH: &str = "\
353data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\
354\n\
355data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"glob\",\"arguments\":\"{\\\"pattern\\\":\\\"**/*.md\\\"}\"}}]},\"finish_reason\":null}]}\n\
356\n\
357data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\
358\n\
359data: [DONE]\n\
360\n";
361
362#[tokio::test]
363async fn nonstandard_finish_reason_with_tool_use_still_yields_tooluse() {
364 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
365 api_key: "k".into(),
366 base_url: mock_url(FIXTURE_TOOL_STREAM_BAD_FINISH).await,
367 ..Default::default()
368 })
369 .unwrap();
370
371 let mut req = ChatRequest::new("m");
372 req.messages = vec![Message::user_text("list markdown files")];
373
374 let mut es = provider
375 .stream(req, CancellationToken::new())
376 .await
377 .unwrap();
378 let mut end = None;
379 while let Some(ev) = futures::StreamExt::next(&mut es).await {
380 if let StreamEvent::End { message } = ev {
381 end = Some(message);
382 }
383 }
384 let msg = end.expect("stream must terminate with End");
385 assert_eq!(msg.stop_reason, StopReason::ToolUse);
386 assert!(
387 msg.content
388 .iter()
389 .any(|b| matches!(b, ContentBlock::ToolUse { name, .. } if name == "glob")),
390 "tool_use block preserved"
391 );
392}
393
394// Same endpoint family, worse: body ends right after the tool-call deltas
395// with NO finish_reason and NO [DONE]. The clean-close path must also
396// promote accumulated tool_use to StopReason::ToolUse.
397const FIXTURE_TOOL_STREAM_EOF_NO_FINISH: &str = "\
398data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\"},\"finish_reason\":null}]}\n\
399\n\
400data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_2\",\"type\":\"function\",\"function\":{\"name\":\"glob\",\"arguments\":\"{\\\"pattern\\\":\\\"*\\\"}\"}}]},\"finish_reason\":null}]}\n\
401\n";
402
403#[tokio::test]
404async fn clean_close_after_tool_calls_yields_tooluse() {
405 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
406 api_key: "k".into(),
407 base_url: mock_url(FIXTURE_TOOL_STREAM_EOF_NO_FINISH).await,
408 ..Default::default()
409 })
410 .unwrap();
411
412 let mut req = ChatRequest::new("m");
413 req.messages = vec![Message::user_text("list files")];
414
415 let es = provider
416 .stream(req, CancellationToken::new())
417 .await
418 .unwrap();
419 // EOF-close delivers the message through the stream's terminal.
420 let msg = es.result().await.expect("clean close must complete");
421 assert_eq!(msg.stop_reason, StopReason::ToolUse);
422}
423
424// The live culprit (opencode-zen free tier): reasoning deltas, one
425// tool_calls delta, then bare `data: [DONE]` — no finish_reason frame ever.
426const FIXTURE_DONE_WITHOUT_FINISH: &str = "\
427data: {\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"reasoning_content\":\"thinking\"},\"finish_reason\":null}]}\n\
428\n\
429data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_3\",\"type\":\"function\",\"function\":{\"name\":\"glob\",\"arguments\":\"{\\\"pattern\\\":\\\"*\\\"}\"}}]},\"finish_reason\":null}]}\n\
430\n\
431data: [DONE]\n\
432\n";
433
434#[tokio::test]
435async fn done_without_finish_reason_still_yields_tooluse() {
436 let provider = OpenAiCompletionsProvider::new(OpenAiConfig {
437 api_key: "k".into(),
438 base_url: mock_url(FIXTURE_DONE_WITHOUT_FINISH).await,
439 ..Default::default()
440 })
441 .unwrap();
442
443 let mut req = ChatRequest::new("m");
444 req.messages = vec![Message::user_text("list files")];
445
446 let es = provider
447 .stream(req, CancellationToken::new())
448 .await
449 .unwrap();
450 let msg = es.result().await.expect("[DONE] close must complete");
451 assert_eq!(msg.stop_reason, StopReason::ToolUse);
452}
453

Openai Stream

Indexing the workspace…

Openai Stream