Vakyartha documentation

1//! Native Ollama adapter (`POST /api/chat`, streaming NDJSON) — replaces the
2//! OpenAI-compatible route for Ollama so `keep_alive` and `options.num_ctx`
3//! reach the server: the compat endpoint silently ignores both
4//! (docs/design/68-context-engine.md §8).
5
6use futures::StreamExt;
7use serde_json::Value;
8use tokio_util::sync::CancellationToken;
9
10use crate::Provider;
11use crate::error::LlmError;
12use crate::gate::ProviderGate;
13use crate::stream::{EventStream, StreamEvent, channel};
14use crate::types::{
15 AssistantMessage, ChatRequest, ContentBlock, Message, Role, StopReason, ToolDefinition, Usage,
16};
17
18pub const OLLAMA_DEFAULT_BASE_URL: &str = "http://localhost:11434";
19
20#[derive(Debug, Clone)]
21pub struct OllamaConfig {
22 /// Empty for a local, unauthenticated server; set when Ollama sits
23 /// behind an auth-terminating proxy.
24 pub api_key: String,
25 pub base_url: String,
26 /// Sent on every request so the runner does not evict the model between
27 /// turns under the default 5-minute idle unload.
28 pub keep_alive: String,
29 /// `options.num_ctx`; omitted from the request body when `None` so the
30 /// server's own modelfile default applies.
31 pub num_ctx: Option<u64>,
32}
33
34impl Default for OllamaConfig {
35 fn default() -> Self {
36 OllamaConfig {
37 api_key: String::new(),
38 base_url: OLLAMA_DEFAULT_BASE_URL.to_string(),
39 keep_alive: "30m".to_string(),
40 num_ctx: None,
41 }
42 }
43}
44
45#[derive(Clone)]
46pub struct OllamaProvider {
47 http: reqwest::Client,
48 config: OllamaConfig,
49 gate: ProviderGate,
50}
51
52impl OllamaProvider {
53 pub fn new(config: OllamaConfig) -> Result<Self, LlmError> {
54 let http = reqwest::Client::builder()
55 .connect_timeout(std::time::Duration::from_secs(30))
56 .build()
57 .map_err(|e| LlmError::Network(e.to_string()))?;
58 Ok(OllamaProvider {
59 gate: ProviderGate::new(&config.base_url, &config.api_key),
60 http,
61 config,
62 })
63 }
64}
65
66pub fn build_body(config: &OllamaConfig, request: &ChatRequest) -> Result<Value, LlmError> {
67 let mut messages: Vec<Value> = Vec::with_capacity(request.messages.len() + 1);
68 if let Some(system) = &request.system {
69 messages.push(serde_json::json!({"role": "system", "content": system}));
70 }
71 for m in &request.messages {
72 append_message(&mut messages, m)?;
73 }
74
75 let mut options = serde_json::Map::new();
76 if let Some(num_ctx) = config.num_ctx {
77 options.insert("num_ctx".to_string(), serde_json::json!(num_ctx));
78 }
79 options.insert(
80 "num_predict".to_string(),
81 serde_json::json!(request.max_tokens),
82 );
83
84 let mut body = serde_json::json!({
85 "model": request.model,
86 "messages": messages,
87 "stream": true,
88 "keep_alive": config.keep_alive,
89 "options": options,
90 });
91 if let Some(think) = request.think {
92 body["think"] = Value::Bool(think);
93 }
94 if !request.tools.is_empty() {
95 let tools: Vec<Value> = request
96 .tools
97 .iter()
98 .map(|t: &ToolDefinition| {
99 serde_json::json!({
100 "type": "function",
101 "function": {
102 "name": t.name,
103 "description": t.description,
104 "parameters": t.parameters,
105 },
106 })
107 })
108 .collect();
109 body["tools"] = Value::Array(tools);
110 }
111 Ok(body)
112}
113
114fn append_message(out: &mut Vec<Value>, m: &Message) -> Result<(), LlmError> {
115 match m.role {
116 Role::User => {
117 let mut text_parts: Vec<&str> = Vec::new();
118 let mut images: Vec<&str> = Vec::new();
119 let mut tool_results: Vec<&ContentBlock> = Vec::new();
120 for b in &m.content {
121 match b {
122 ContentBlock::Text { text } => text_parts.push(text),
123 // Ollama's native wire wants raw base64, unlike the
124 // compat endpoint's data-URL form.
125 ContentBlock::Image { source } => images.push(source.data.as_str()),
126 ContentBlock::ToolResult { .. } => tool_results.push(b),
127 ContentBlock::ToolUse { .. } => {
128 return Err(LlmError::InvalidRequest(
129 "tool_use blocks must appear in assistant messages".into(),
130 ));
131 }
132 // No provider needs thinking replayed across the wire
133 // back to it; Ollama is no exception (docs/design/68 §10).
134 ContentBlock::Thinking { .. } => {}
135 // Only the Anthropic adapter understands server-side
136 // tool search; Ollama skips this opaque block entirely
137 // (docs/design/68 §5/§12).
138 ContentBlock::Provider { .. } => {}
139 }
140 }
141 for r in tool_results {
142 let ContentBlock::ToolResult { content, .. } = r else {
143 unreachable!()
144 };
145 out.push(serde_json::json!({"role": "tool", "content": content}));
146 }
147 if !text_parts.is_empty() || !images.is_empty() {
148 let mut msg = serde_json::json!({
149 "role": "user",
150 "content": text_parts.join("\n"),
151 });
152 if !images.is_empty() {
153 msg["images"] = serde_json::json!(images);
154 }
155 out.push(msg);
156 }
157 }
158 Role::Assistant => {
159 let mut text = String::new();
160 let mut tool_calls: Vec<Value> = Vec::new();
161 for b in &m.content {
162 match b {
163 ContentBlock::Text { text: t } => {
164 if !text.is_empty() {
165 text.push('\n');
166 }
167 text.push_str(t);
168 }
169 ContentBlock::ToolUse { id, name, input } => {
170 tool_calls.push(serde_json::json!({
171 "id": id,
172 "function": {"name": name, "arguments": input},
173 }));
174 }
175 ContentBlock::Thinking { .. }
176 | ContentBlock::ToolResult { .. }
177 | ContentBlock::Image { .. }
178 | ContentBlock::Provider { .. } => {}
179 }
180 }
181 let mut msg = serde_json::json!({"role": "assistant", "content": text});
182 if !tool_calls.is_empty() {
183 msg["tool_calls"] = Value::Array(tool_calls);
184 }
185 out.push(msg);
186 }
187 }
188 Ok(())
189}
190
191fn map_status_error(status: u16, body: &str) -> LlmError {
192 let message = serde_json::from_str::<Value>(body)
193 .ok()
194 .and_then(|v| v.get("error").and_then(|m| m.as_str().map(String::from)))
195 .unwrap_or_else(|| body.chars().take(500).collect());
196 match status {
197 401 | 403 => LlmError::Auth(message),
198 400 => LlmError::classify_400(message),
199 404 | 413 | 422 => LlmError::InvalidRequest(message),
200 429 => LlmError::RateLimit {
201 message,
202 retry_after_secs: None,
203 },
204 503 | 529 => LlmError::Overloaded(message),
205 _ => LlmError::Api { status, message },
206 }
207}
208
209/// Splits a byte stream into newline-delimited JSON records — Ollama's
210/// `/api/chat` wire format has no `data:`/event framing, just one JSON
211/// object per line.
212#[derive(Default)]
213struct NdjsonDecoder {
214 buf: Vec<u8>,
215 cursor: usize,
216}
217
218impl NdjsonDecoder {
219 fn push(&mut self, chunk: &[u8]) {
220 self.buf.extend_from_slice(chunk);
221 }
222
223 fn next_line(&mut self) -> Option<String> {
224 loop {
225 let Some(nl) = self.buf[self.cursor..].iter().position(|&b| b == b'\n') else {
226 self.compact();
227 return None;
228 };
229 let start = self.cursor;
230 let end = self.cursor + nl;
231 let mut line = &self.buf[start..end];
232 self.cursor = end + 1;
233 if line.last() == Some(&b'\r') {
234 line = &line[..line.len() - 1];
235 }
236 if line.is_empty() {
237 continue;
238 }
239 let text = String::from_utf8_lossy(line).into_owned();
240 self.compact();
241 return Some(text);
242 }
243 }
244
245 fn compact(&mut self) {
246 if self.cursor > 0 {
247 self.buf.drain(..self.cursor);
248 self.cursor = 0;
249 }
250 }
251}
252
253struct Accumulator {
254 message: AssistantMessage,
255 /// Count of tool-call blocks materialized so far, used to synthesize
256 /// `call_<n>` ids when Ollama omits them.
257 tool_calls_seen: usize,
258}
259
260impl Accumulator {
261 fn new(model: &str) -> Self {
262 Accumulator {
263 message: AssistantMessage::empty(model),
264 tool_calls_seen: 0,
265 }
266 }
267
268 fn convert(&mut self, line: &str) -> Result<Option<StreamEvent>, LlmError> {
269 let v: Value = serde_json::from_str(line)
270 .map_err(|e| LlmError::Parse(format!("bad ndjson line: {e}")))?;
271
272 if let Some(err) = v.get("error").and_then(|e| e.as_str()) {
273 return Err(LlmError::classify_400(err.to_string()));
274 }
275
276 let mut event = None;
277 if let Some(message) = v.get("message") {
278 if let Some(text) = message
279 .get("content")
280 .and_then(|c| c.as_str())
281 .filter(|t| !t.is_empty())
282 {
283 append_text_block(&mut self.message.content, text);
284 event = Some(StreamEvent::TextDelta {
285 delta: text.to_string(),
286 partial: self.message.clone(),
287 });
288 }
289 if let Some(thinking) = message
290 .get("thinking")
291 .and_then(|c| c.as_str())
292 .filter(|t| !t.is_empty())
293 {
294 append_thinking_block(&mut self.message.content, thinking);
295 event = Some(StreamEvent::ThinkingDelta {
296 delta: thinking.to_string(),
297 partial: self.message.clone(),
298 });
299 }
300 if let Some(calls) = message.get("tool_calls").and_then(|c| c.as_array()) {
301 for call in calls {
302 let name = call
303 .pointer("/function/name")
304 .and_then(|n| n.as_str())
305 .unwrap_or_default()
306 .to_string();
307 let input = call
308 .pointer("/function/arguments")
309 .cloned()
310 .unwrap_or(Value::Object(Default::default()));
311 let id = call
312 .get("id")
313 .and_then(|i| i.as_str())
314 .map(str::to_string)
315 .unwrap_or_else(|| format!("call_{}", self.tool_calls_seen));
316 self.tool_calls_seen += 1;
317 let pos = self.message.content.len();
318 self.message.content.push(ContentBlock::ToolUse {
319 id: id.clone(),
320 name: name.clone(),
321 input,
322 });
323 self.message.stop_reason = StopReason::ToolUse;
324 event = Some(StreamEvent::ToolUseStart {
325 index: pos,
326 id,
327 name,
328 partial: self.message.clone(),
329 });
330 }
331 }
332 }
333
334 if v.get("done").and_then(|d| d.as_bool()).unwrap_or(false) {
335 self.message.usage = Usage {
336 input_tokens: v
337 .get("prompt_eval_count")
338 .and_then(|x| x.as_u64())
339 .unwrap_or(0),
340 output_tokens: v.get("eval_count").and_then(|x| x.as_u64()).unwrap_or(0),
341 cache_read_input_tokens: None,
342 cache_creation_input_tokens: None,
343 prefill_ms: v
344 .get("prompt_eval_duration")
345 .and_then(|x| x.as_u64())
346 .map(|ns| ns / 1_000_000),
347 load_ms: v
348 .get("load_duration")
349 .and_then(|x| x.as_u64())
350 .map(|ns| ns / 1_000_000),
351 };
352 if self.message.stop_reason != StopReason::ToolUse {
353 self.message.stop_reason = match v.get("done_reason").and_then(|r| r.as_str()) {
354 Some("length") => StopReason::MaxTokens,
355 _ => StopReason::EndTurn,
356 };
357 }
358 return Ok(Some(StreamEvent::End {
359 message: self.message.clone(),
360 }));
361 }
362
363 Ok(event)
364 }
365}
366
367fn append_text_block(content: &mut Vec<ContentBlock>, text: &str) {
368 if let Some(ContentBlock::Text { text: last }) = content.last_mut() {
369 last.push_str(text);
370 return;
371 }
372 content.push(ContentBlock::text(text));
373}
374
375fn append_thinking_block(content: &mut Vec<ContentBlock>, text: &str) {
376 if let Some(ContentBlock::Thinking { text: last, .. }) = content.last_mut() {
377 last.push_str(text);
378 return;
379 }
380 content.push(ContentBlock::Thinking {
381 text: text.to_string(),
382 signature: None,
383 });
384}
385
386#[async_trait::async_trait]
387impl Provider for OllamaProvider {
388 fn name(&self) -> &str {
389 "ollama"
390 }
391
392 fn circuit_key(&self) -> String {
393 crate::gate::route_identity(self.name(), &self.config.base_url, &self.config.api_key)
394 }
395
396 async fn stream(
397 &self,
398 request: ChatRequest,
399 cancel: CancellationToken,
400 ) -> Result<EventStream, LlmError> {
401 let provider_permit = self.gate.acquire(&cancel).await?;
402 let url = format!("{}/api/chat", self.config.base_url.trim_end_matches('/'));
403 let body = build_body(&self.config, &request)?;
404 let mut req = self.http.post(&url).json(&body);
405 if !self.config.api_key.is_empty() {
406 req = req.bearer_auth(&self.config.api_key);
407 }
408 let send_fut = req.send();
409 let response = tokio::select! {
410 _ = cancel.cancelled() => return Err(LlmError::Aborted { partial: None }),
411 r = send_fut => match r {
412 Ok(r) => r,
413 Err(e) => return Err(LlmError::Network(e.to_string())),
414 },
415 };
416
417 let status = response.status();
418 if !status.is_success() {
419 let text = response.text().await.unwrap_or_default();
420 return Err(map_status_error(status.as_u16(), &text));
421 }
422
423 let model = request.model.clone();
424 let (mut sink, stream_rx) = channel(256);
425 let mut byte_stream = response.bytes_stream();
426 let mut decoder = NdjsonDecoder::default();
427 let mut acc = Accumulator::new(&model);
428
429 tokio::spawn(async move {
430 let mut saw_end = false;
431 loop {
432 tokio::select! {
433 _ = cancel.cancelled() => {
434 let partial = (!acc.message.content.is_empty()).then(|| Box::new(acc.message.clone()));
435 sink.close_error(LlmError::Aborted { partial }).await;
436 return;
437 }
438 chunk = byte_stream.next() => {
439 match chunk {
440 Some(Ok(bytes)) => {
441 decoder.push(&bytes);
442 while let Some(line) = decoder.next_line() {
443 match acc.convert(&line) {
444 Ok(Some(event)) => {
445 if matches!(event, StreamEvent::End { .. }) {
446 saw_end = true;
447 }
448 sink.push(event);
449 }
450 Ok(None) => {}
451 Err(e) => {
452 sink.close_error(e).await;
453 return;
454 }
455 }
456 }
457 }
458 Some(Err(e)) => {
459 sink.close_error(LlmError::Network(e.to_string())).await;
460 return;
461 }
462 None => {
463 if saw_end {
464 sink.close_message(acc.message.clone()).await;
465 } else {
466 sink.close_error(LlmError::Parse(
467 "stream closed before done:true".into(),
468 )).await;
469 }
470 return;
471 }
472 }
473 }
474 }
475 }
476 });
477
478 Ok(stream_rx.with_guard(provider_permit))
479 }
480}
481
482#[cfg(test)]
483mod tests {
484 #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
485 use super::*;
486
487 #[test]
488 fn body_omits_num_ctx_by_default_and_sets_num_predict() {
489 let mut req = ChatRequest::new("gemma3:e2b");
490 req.max_tokens = 512;
491 req.messages = vec![Message::user_text("hi")];
492 let body = build_body(&OllamaConfig::default(), &req).unwrap();
493 assert_eq!(body["keep_alive"], "30m");
494 assert!(body["options"].get("num_ctx").is_none());
495 assert_eq!(body["options"]["num_predict"], 512);
496 assert_eq!(body["stream"], true);
497 }
498
499 #[test]
500 fn body_includes_num_ctx_when_configured() {
501 let mut req = ChatRequest::new("gemma3:e2b");
502 req.messages = vec![Message::user_text("hi")];
503 let config = OllamaConfig {
504 num_ctx: Some(8192),
505 keep_alive: "10m".into(),
506 ..Default::default()
507 };
508 let body = build_body(&config, &req).unwrap();
509 assert_eq!(body["options"]["num_ctx"], 8192);
510 assert_eq!(body["keep_alive"], "10m");
511 }
512
513 #[test]
514 fn tool_use_round_trips_with_object_arguments() {
515 let mut req = ChatRequest::new("gemma3:e2b");
516 req.messages = vec![
517 Message::user_text("search"),
518 Message::assistant(vec![ContentBlock::ToolUse {
519 id: "call_0".into(),
520 name: "search".into(),
521 input: serde_json::json!({"q": "vak"}),
522 }]),
523 Message {
524 role: Role::User,
525 content: vec![ContentBlock::tool_result("call_0", "result text")],
526 },
527 ];
528 let body = build_body(&OllamaConfig::default(), &req).unwrap();
529 let msgs = body["messages"].as_array().unwrap();
530 let call = &msgs[1]["tool_calls"][0];
531 // Ollama's native wire wants a JSON object, not an escaped string.
532 assert_eq!(
533 call["function"]["arguments"],
534 serde_json::json!({"q": "vak"})
535 );
536 assert_eq!(msgs[2]["role"], "tool");
537 assert_eq!(msgs[2]["content"], "result text");
538 }
539
540 #[test]
541 fn thinking_is_never_sent_back_to_ollama() {
542 let mut req = ChatRequest::new("gemma3:e2b");
543 req.messages = vec![Message::assistant(vec![
544 ContentBlock::Thinking {
545 text: "reasoning".into(),
546 signature: None,
547 },
548 ContentBlock::text("answer"),
549 ])];
550 let body = build_body(&OllamaConfig::default(), &req).unwrap();
551 assert_eq!(body["messages"][0]["content"], "answer");
552 }
553
554 #[test]
555 fn over_length_400_becomes_context_error() {
556 let error = map_status_error(
557 400,
558 r#"{"error":"POST predict: this model's max context length (4096) exceeds the model's maximum context length"}"#,
559 );
560 assert!(matches!(error, LlmError::Context(_)));
561 }
562
563 #[test]
564 fn unrelated_400_stays_invalid_request() {
565 let error = map_status_error(400, r#"{"error":"model 'x' not found"}"#);
566 assert!(matches!(error, LlmError::InvalidRequest(_)));
567 }
568
569 const FIXTURE_LINES: &[&str] = &[
570 r#"{"model":"gemma3:e2b","message":{"role":"assistant","thinking":"let me "},"done":false}"#,
571 r#"{"model":"gemma3:e2b","message":{"role":"assistant","thinking":"think"},"done":false}"#,
572 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":"The "},"done":false}"#,
573 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":"answer is 4."},"done":false}"#,
574 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":""},"done":true,"done_reason":"stop","total_duration":1000000,"load_duration":50000000,"prompt_eval_count":21,"prompt_eval_duration":120000000,"eval_count":9,"eval_duration":300000000}"#,
575 ];
576
577 #[test]
578 fn stream_fixture_accumulates_text_thinking_and_usage() {
579 let mut acc = Accumulator::new("gemma3:e2b");
580 let mut end = None;
581 for line in FIXTURE_LINES {
582 if let Some(event) = acc.convert(line).unwrap()
583 && let StreamEvent::End { message } = event
584 {
585 end = Some(message);
586 }
587 }
588 let msg = end.expect("fixture must end with done:true");
589 assert_eq!(msg.text_content(), "The answer is 4.");
590 let thinking = msg
591 .content
592 .iter()
593 .find_map(|b| match b {
594 ContentBlock::Thinking { text, .. } => Some(text.clone()),
595 _ => None,
596 })
597 .unwrap();
598 assert_eq!(thinking, "let me think");
599 assert_eq!(msg.usage.input_tokens, 21);
600 assert_eq!(msg.usage.output_tokens, 9);
601 assert_eq!(msg.usage.prefill_ms, Some(120));
602 assert_eq!(msg.usage.load_ms, Some(50));
603 assert_eq!(msg.stop_reason, StopReason::EndTurn);
604 }
605
606 const TOOL_CALL_LINES: &[&str] = &[
607 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":"","tool_calls":[{"function":{"name":"search","arguments":{"q":"vak"}}}]},"done":false}"#,
608 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":""},"done":true,"done_reason":"stop","prompt_eval_count":10,"eval_count":5}"#,
609 ];
610
611 #[test]
612 fn stream_fixture_generates_call_id_when_ollama_omits_one() {
613 let mut acc = Accumulator::new("gemma3:e2b");
614 let mut end = None;
615 for line in TOOL_CALL_LINES {
616 if let Some(event) = acc.convert(line).unwrap()
617 && let StreamEvent::End { message } = event
618 {
619 end = Some(message);
620 }
621 }
622 let msg = end.unwrap();
623 assert_eq!(msg.stop_reason, StopReason::ToolUse);
624 let ContentBlock::ToolUse { id, name, input } = &msg.content[0] else {
625 panic!("expected a tool_use block");
626 };
627 assert_eq!(id, "call_0");
628 assert_eq!(name, "search");
629 assert_eq!(input, &serde_json::json!({"q": "vak"}));
630 }
631
632 #[test]
633 fn ndjson_decoder_splits_on_newlines_across_pushes() {
634 let mut decoder = NdjsonDecoder::default();
635 decoder.push(b"{\"a\":1}\n{\"b\":");
636 assert_eq!(decoder.next_line().as_deref(), Some(r#"{"a":1}"#));
637 assert_eq!(decoder.next_line(), None);
638 decoder.push(b"2}\n");
639 assert_eq!(decoder.next_line().as_deref(), Some(r#"{"b":2}"#));
640 }
641}
642

1//! Native Ollama adapter (`POST /api/chat`, streaming NDJSON) — replaces the
2//! OpenAI-compatible route for Ollama so `keep_alive` and `options.num_ctx`
3//! reach the server: the compat endpoint silently ignores both
4//! (docs/design/68-context-engine.md §8).
5
6use futures::StreamExt;
7use serde_json::Value;
8use tokio_util::sync::CancellationToken;
9
10use crate::Provider;
11use crate::error::LlmError;
12use crate::gate::ProviderGate;
13use crate::stream::{EventStream, StreamEvent, channel};
14use crate::types::{
15 AssistantMessage, ChatRequest, ContentBlock, Message, Role, StopReason, ToolDefinition, Usage,
16};
17
18pub const OLLAMA_DEFAULT_BASE_URL: &str = "http://localhost:11434";
19
20#[derive(Debug, Clone)]
21pub struct OllamaConfig {
22 /// Empty for a local, unauthenticated server; set when Ollama sits
23 /// behind an auth-terminating proxy.
24 pub api_key: String,
25 pub base_url: String,
26 /// Sent on every request so the runner does not evict the model between
27 /// turns under the default 5-minute idle unload.
28 pub keep_alive: String,
29 /// `options.num_ctx`; omitted from the request body when `None` so the
30 /// server's own modelfile default applies.
31 pub num_ctx: Option<u64>,
32}
33
34impl Default for OllamaConfig {
35 fn default() -> Self {
36 OllamaConfig {
37 api_key: String::new(),
38 base_url: OLLAMA_DEFAULT_BASE_URL.to_string(),
39 keep_alive: "30m".to_string(),
40 num_ctx: None,
41 }
42 }
43}
44
45#[derive(Clone)]
46pub struct OllamaProvider {
47 http: reqwest::Client,
48 config: OllamaConfig,
49 gate: ProviderGate,
50}
51
52impl OllamaProvider {
53 pub fn new(config: OllamaConfig) -> Result<Self, LlmError> {
54 let http = reqwest::Client::builder()
55 .connect_timeout(std::time::Duration::from_secs(30))
56 .build()
57 .map_err(|e| LlmError::Network(e.to_string()))?;
58 Ok(OllamaProvider {
59 gate: ProviderGate::new(&config.base_url, &config.api_key),
60 http,
61 config,
62 })
63 }
64}
65
66pub fn build_body(config: &OllamaConfig, request: &ChatRequest) -> Result<Value, LlmError> {
67 let mut messages: Vec<Value> = Vec::with_capacity(request.messages.len() + 1);
68 if let Some(system) = &request.system {
69 messages.push(serde_json::json!({"role": "system", "content": system}));
70 }
71 for m in &request.messages {
72 append_message(&mut messages, m)?;
73 }
74
75 let mut options = serde_json::Map::new();
76 if let Some(num_ctx) = config.num_ctx {
77 options.insert("num_ctx".to_string(), serde_json::json!(num_ctx));
78 }
79 options.insert(
80 "num_predict".to_string(),
81 serde_json::json!(request.max_tokens),
82 );
83
84 let mut body = serde_json::json!({
85 "model": request.model,
86 "messages": messages,
87 "stream": true,
88 "keep_alive": config.keep_alive,
89 "options": options,
90 });
91 if let Some(think) = request.think {
92 body["think"] = Value::Bool(think);
93 }
94 if !request.tools.is_empty() {
95 let tools: Vec<Value> = request
96 .tools
97 .iter()
98 .map(|t: &ToolDefinition| {
99 serde_json::json!({
100 "type": "function",
101 "function": {
102 "name": t.name,
103 "description": t.description,
104 "parameters": t.parameters,
105 },
106 })
107 })
108 .collect();
109 body["tools"] = Value::Array(tools);
110 }
111 Ok(body)
112}
113
114fn append_message(out: &mut Vec<Value>, m: &Message) -> Result<(), LlmError> {
115 match m.role {
116 Role::User => {
117 let mut text_parts: Vec<&str> = Vec::new();
118 let mut images: Vec<&str> = Vec::new();
119 let mut tool_results: Vec<&ContentBlock> = Vec::new();
120 for b in &m.content {
121 match b {
122 ContentBlock::Text { text } => text_parts.push(text),
123 // Ollama's native wire wants raw base64, unlike the
124 // compat endpoint's data-URL form.
125 ContentBlock::Image { source } => images.push(source.data.as_str()),
126 ContentBlock::ToolResult { .. } => tool_results.push(b),
127 ContentBlock::ToolUse { .. } => {
128 return Err(LlmError::InvalidRequest(
129 "tool_use blocks must appear in assistant messages".into(),
130 ));
131 }
132 // No provider needs thinking replayed across the wire
133 // back to it; Ollama is no exception (docs/design/68 §10).
134 ContentBlock::Thinking { .. } => {}
135 // Only the Anthropic adapter understands server-side
136 // tool search; Ollama skips this opaque block entirely
137 // (docs/design/68 §5/§12).
138 ContentBlock::Provider { .. } => {}
139 }
140 }
141 for r in tool_results {
142 let ContentBlock::ToolResult { content, .. } = r else {
143 unreachable!()
144 };
145 out.push(serde_json::json!({"role": "tool", "content": content}));
146 }
147 if !text_parts.is_empty() || !images.is_empty() {
148 let mut msg = serde_json::json!({
149 "role": "user",
150 "content": text_parts.join("\n"),
151 });
152 if !images.is_empty() {
153 msg["images"] = serde_json::json!(images);
154 }
155 out.push(msg);
156 }
157 }
158 Role::Assistant => {
159 let mut text = String::new();
160 let mut tool_calls: Vec<Value> = Vec::new();
161 for b in &m.content {
162 match b {
163 ContentBlock::Text { text: t } => {
164 if !text.is_empty() {
165 text.push('\n');
166 }
167 text.push_str(t);
168 }
169 ContentBlock::ToolUse { id, name, input } => {
170 tool_calls.push(serde_json::json!({
171 "id": id,
172 "function": {"name": name, "arguments": input},
173 }));
174 }
175 ContentBlock::Thinking { .. }
176 | ContentBlock::ToolResult { .. }
177 | ContentBlock::Image { .. }
178 | ContentBlock::Provider { .. } => {}
179 }
180 }
181 let mut msg = serde_json::json!({"role": "assistant", "content": text});
182 if !tool_calls.is_empty() {
183 msg["tool_calls"] = Value::Array(tool_calls);
184 }
185 out.push(msg);
186 }
187 }
188 Ok(())
189}
190
191fn map_status_error(status: u16, body: &str) -> LlmError {
192 let message = serde_json::from_str::<Value>(body)
193 .ok()
194 .and_then(|v| v.get("error").and_then(|m| m.as_str().map(String::from)))
195 .unwrap_or_else(|| body.chars().take(500).collect());
196 match status {
197 401 | 403 => LlmError::Auth(message),
198 400 => LlmError::classify_400(message),
199 404 | 413 | 422 => LlmError::InvalidRequest(message),
200 429 => LlmError::RateLimit {
201 message,
202 retry_after_secs: None,
203 },
204 503 | 529 => LlmError::Overloaded(message),
205 _ => LlmError::Api { status, message },
206 }
207}
208
209/// Splits a byte stream into newline-delimited JSON records — Ollama's
210/// `/api/chat` wire format has no `data:`/event framing, just one JSON
211/// object per line.
212#[derive(Default)]
213struct NdjsonDecoder {
214 buf: Vec<u8>,
215 cursor: usize,
216}
217
218impl NdjsonDecoder {
219 fn push(&mut self, chunk: &[u8]) {
220 self.buf.extend_from_slice(chunk);
221 }
222
223 fn next_line(&mut self) -> Option<String> {
224 loop {
225 let Some(nl) = self.buf[self.cursor..].iter().position(|&b| b == b'\n') else {
226 self.compact();
227 return None;
228 };
229 let start = self.cursor;
230 let end = self.cursor + nl;
231 let mut line = &self.buf[start..end];
232 self.cursor = end + 1;
233 if line.last() == Some(&b'\r') {
234 line = &line[..line.len() - 1];
235 }
236 if line.is_empty() {
237 continue;
238 }
239 let text = String::from_utf8_lossy(line).into_owned();
240 self.compact();
241 return Some(text);
242 }
243 }
244
245 fn compact(&mut self) {
246 if self.cursor > 0 {
247 self.buf.drain(..self.cursor);
248 self.cursor = 0;
249 }
250 }
251}
252
253struct Accumulator {
254 message: AssistantMessage,
255 /// Count of tool-call blocks materialized so far, used to synthesize
256 /// `call_<n>` ids when Ollama omits them.
257 tool_calls_seen: usize,
258}
259
260impl Accumulator {
261 fn new(model: &str) -> Self {
262 Accumulator {
263 message: AssistantMessage::empty(model),
264 tool_calls_seen: 0,
265 }
266 }
267
268 fn convert(&mut self, line: &str) -> Result<Option<StreamEvent>, LlmError> {
269 let v: Value = serde_json::from_str(line)
270 .map_err(|e| LlmError::Parse(format!("bad ndjson line: {e}")))?;
271
272 if let Some(err) = v.get("error").and_then(|e| e.as_str()) {
273 return Err(LlmError::classify_400(err.to_string()));
274 }
275
276 let mut event = None;
277 if let Some(message) = v.get("message") {
278 if let Some(text) = message
279 .get("content")
280 .and_then(|c| c.as_str())
281 .filter(|t| !t.is_empty())
282 {
283 append_text_block(&mut self.message.content, text);
284 event = Some(StreamEvent::TextDelta {
285 delta: text.to_string(),
286 partial: self.message.clone(),
287 });
288 }
289 if let Some(thinking) = message
290 .get("thinking")
291 .and_then(|c| c.as_str())
292 .filter(|t| !t.is_empty())
293 {
294 append_thinking_block(&mut self.message.content, thinking);
295 event = Some(StreamEvent::ThinkingDelta {
296 delta: thinking.to_string(),
297 partial: self.message.clone(),
298 });
299 }
300 if let Some(calls) = message.get("tool_calls").and_then(|c| c.as_array()) {
301 for call in calls {
302 let name = call
303 .pointer("/function/name")
304 .and_then(|n| n.as_str())
305 .unwrap_or_default()
306 .to_string();
307 let input = call
308 .pointer("/function/arguments")
309 .cloned()
310 .unwrap_or(Value::Object(Default::default()));
311 let id = call
312 .get("id")
313 .and_then(|i| i.as_str())
314 .map(str::to_string)
315 .unwrap_or_else(|| format!("call_{}", self.tool_calls_seen));
316 self.tool_calls_seen += 1;
317 let pos = self.message.content.len();
318 self.message.content.push(ContentBlock::ToolUse {
319 id: id.clone(),
320 name: name.clone(),
321 input,
322 });
323 self.message.stop_reason = StopReason::ToolUse;
324 event = Some(StreamEvent::ToolUseStart {
325 index: pos,
326 id,
327 name,
328 partial: self.message.clone(),
329 });
330 }
331 }
332 }
333
334 if v.get("done").and_then(|d| d.as_bool()).unwrap_or(false) {
335 self.message.usage = Usage {
336 input_tokens: v
337 .get("prompt_eval_count")
338 .and_then(|x| x.as_u64())
339 .unwrap_or(0),
340 output_tokens: v.get("eval_count").and_then(|x| x.as_u64()).unwrap_or(0),
341 cache_read_input_tokens: None,
342 cache_creation_input_tokens: None,
343 prefill_ms: v
344 .get("prompt_eval_duration")
345 .and_then(|x| x.as_u64())
346 .map(|ns| ns / 1_000_000),
347 load_ms: v
348 .get("load_duration")
349 .and_then(|x| x.as_u64())
350 .map(|ns| ns / 1_000_000),
351 };
352 if self.message.stop_reason != StopReason::ToolUse {
353 self.message.stop_reason = match v.get("done_reason").and_then(|r| r.as_str()) {
354 Some("length") => StopReason::MaxTokens,
355 _ => StopReason::EndTurn,
356 };
357 }
358 return Ok(Some(StreamEvent::End {
359 message: self.message.clone(),
360 }));
361 }
362
363 Ok(event)
364 }
365}
366
367fn append_text_block(content: &mut Vec<ContentBlock>, text: &str) {
368 if let Some(ContentBlock::Text { text: last }) = content.last_mut() {
369 last.push_str(text);
370 return;
371 }
372 content.push(ContentBlock::text(text));
373}
374
375fn append_thinking_block(content: &mut Vec<ContentBlock>, text: &str) {
376 if let Some(ContentBlock::Thinking { text: last, .. }) = content.last_mut() {
377 last.push_str(text);
378 return;
379 }
380 content.push(ContentBlock::Thinking {
381 text: text.to_string(),
382 signature: None,
383 });
384}
385
386#[async_trait::async_trait]
387impl Provider for OllamaProvider {
388 fn name(&self) -> &str {
389 "ollama"
390 }
391
392 fn circuit_key(&self) -> String {
393 crate::gate::route_identity(self.name(), &self.config.base_url, &self.config.api_key)
394 }
395
396 async fn stream(
397 &self,
398 request: ChatRequest,
399 cancel: CancellationToken,
400 ) -> Result<EventStream, LlmError> {
401 let provider_permit = self.gate.acquire(&cancel).await?;
402 let url = format!("{}/api/chat", self.config.base_url.trim_end_matches('/'));
403 let body = build_body(&self.config, &request)?;
404 let mut req = self.http.post(&url).json(&body);
405 if !self.config.api_key.is_empty() {
406 req = req.bearer_auth(&self.config.api_key);
407 }
408 let send_fut = req.send();
409 let response = tokio::select! {
410 _ = cancel.cancelled() => return Err(LlmError::Aborted { partial: None }),
411 r = send_fut => match r {
412 Ok(r) => r,
413 Err(e) => return Err(LlmError::Network(e.to_string())),
414 },
415 };
416
417 let status = response.status();
418 if !status.is_success() {
419 let text = response.text().await.unwrap_or_default();
420 return Err(map_status_error(status.as_u16(), &text));
421 }
422
423 let model = request.model.clone();
424 let (mut sink, stream_rx) = channel(256);
425 let mut byte_stream = response.bytes_stream();
426 let mut decoder = NdjsonDecoder::default();
427 let mut acc = Accumulator::new(&model);
428
429 tokio::spawn(async move {
430 let mut saw_end = false;
431 loop {
432 tokio::select! {
433 _ = cancel.cancelled() => {
434 let partial = (!acc.message.content.is_empty()).then(|| Box::new(acc.message.clone()));
435 sink.close_error(LlmError::Aborted { partial }).await;
436 return;
437 }
438 chunk = byte_stream.next() => {
439 match chunk {
440 Some(Ok(bytes)) => {
441 decoder.push(&bytes);
442 while let Some(line) = decoder.next_line() {
443 match acc.convert(&line) {
444 Ok(Some(event)) => {
445 if matches!(event, StreamEvent::End { .. }) {
446 saw_end = true;
447 }
448 sink.push(event);
449 }
450 Ok(None) => {}
451 Err(e) => {
452 sink.close_error(e).await;
453 return;
454 }
455 }
456 }
457 }
458 Some(Err(e)) => {
459 sink.close_error(LlmError::Network(e.to_string())).await;
460 return;
461 }
462 None => {
463 if saw_end {
464 sink.close_message(acc.message.clone()).await;
465 } else {
466 sink.close_error(LlmError::Parse(
467 "stream closed before done:true".into(),
468 )).await;
469 }
470 return;
471 }
472 }
473 }
474 }
475 }
476 });
477
478 Ok(stream_rx.with_guard(provider_permit))
479 }
480}
481
482#[cfg(test)]
483mod tests {
484 #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
485 use super::*;
486
487 #[test]
488 fn body_omits_num_ctx_by_default_and_sets_num_predict() {
489 let mut req = ChatRequest::new("gemma3:e2b");
490 req.max_tokens = 512;
491 req.messages = vec![Message::user_text("hi")];
492 let body = build_body(&OllamaConfig::default(), &req).unwrap();
493 assert_eq!(body["keep_alive"], "30m");
494 assert!(body["options"].get("num_ctx").is_none());
495 assert_eq!(body["options"]["num_predict"], 512);
496 assert_eq!(body["stream"], true);
497 }
498
499 #[test]
500 fn body_includes_num_ctx_when_configured() {
501 let mut req = ChatRequest::new("gemma3:e2b");
502 req.messages = vec![Message::user_text("hi")];
503 let config = OllamaConfig {
504 num_ctx: Some(8192),
505 keep_alive: "10m".into(),
506 ..Default::default()
507 };
508 let body = build_body(&config, &req).unwrap();
509 assert_eq!(body["options"]["num_ctx"], 8192);
510 assert_eq!(body["keep_alive"], "10m");
511 }
512
513 #[test]
514 fn tool_use_round_trips_with_object_arguments() {
515 let mut req = ChatRequest::new("gemma3:e2b");
516 req.messages = vec![
517 Message::user_text("search"),
518 Message::assistant(vec![ContentBlock::ToolUse {
519 id: "call_0".into(),
520 name: "search".into(),
521 input: serde_json::json!({"q": "vak"}),
522 }]),
523 Message {
524 role: Role::User,
525 content: vec![ContentBlock::tool_result("call_0", "result text")],
526 },
527 ];
528 let body = build_body(&OllamaConfig::default(), &req).unwrap();
529 let msgs = body["messages"].as_array().unwrap();
530 let call = &msgs[1]["tool_calls"][0];
531 // Ollama's native wire wants a JSON object, not an escaped string.
532 assert_eq!(
533 call["function"]["arguments"],
534 serde_json::json!({"q": "vak"})
535 );
536 assert_eq!(msgs[2]["role"], "tool");
537 assert_eq!(msgs[2]["content"], "result text");
538 }
539
540 #[test]
541 fn thinking_is_never_sent_back_to_ollama() {
542 let mut req = ChatRequest::new("gemma3:e2b");
543 req.messages = vec![Message::assistant(vec![
544 ContentBlock::Thinking {
545 text: "reasoning".into(),
546 signature: None,
547 },
548 ContentBlock::text("answer"),
549 ])];
550 let body = build_body(&OllamaConfig::default(), &req).unwrap();
551 assert_eq!(body["messages"][0]["content"], "answer");
552 }
553
554 #[test]
555 fn over_length_400_becomes_context_error() {
556 let error = map_status_error(
557 400,
558 r#"{"error":"POST predict: this model's max context length (4096) exceeds the model's maximum context length"}"#,
559 );
560 assert!(matches!(error, LlmError::Context(_)));
561 }
562
563 #[test]
564 fn unrelated_400_stays_invalid_request() {
565 let error = map_status_error(400, r#"{"error":"model 'x' not found"}"#);
566 assert!(matches!(error, LlmError::InvalidRequest(_)));
567 }
568
569 const FIXTURE_LINES: &[&str] = &[
570 r#"{"model":"gemma3:e2b","message":{"role":"assistant","thinking":"let me "},"done":false}"#,
571 r#"{"model":"gemma3:e2b","message":{"role":"assistant","thinking":"think"},"done":false}"#,
572 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":"The "},"done":false}"#,
573 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":"answer is 4."},"done":false}"#,
574 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":""},"done":true,"done_reason":"stop","total_duration":1000000,"load_duration":50000000,"prompt_eval_count":21,"prompt_eval_duration":120000000,"eval_count":9,"eval_duration":300000000}"#,
575 ];
576
577 #[test]
578 fn stream_fixture_accumulates_text_thinking_and_usage() {
579 let mut acc = Accumulator::new("gemma3:e2b");
580 let mut end = None;
581 for line in FIXTURE_LINES {
582 if let Some(event) = acc.convert(line).unwrap()
583 && let StreamEvent::End { message } = event
584 {
585 end = Some(message);
586 }
587 }
588 let msg = end.expect("fixture must end with done:true");
589 assert_eq!(msg.text_content(), "The answer is 4.");
590 let thinking = msg
591 .content
592 .iter()
593 .find_map(|b| match b {
594 ContentBlock::Thinking { text, .. } => Some(text.clone()),
595 _ => None,
596 })
597 .unwrap();
598 assert_eq!(thinking, "let me think");
599 assert_eq!(msg.usage.input_tokens, 21);
600 assert_eq!(msg.usage.output_tokens, 9);
601 assert_eq!(msg.usage.prefill_ms, Some(120));
602 assert_eq!(msg.usage.load_ms, Some(50));
603 assert_eq!(msg.stop_reason, StopReason::EndTurn);
604 }
605
606 const TOOL_CALL_LINES: &[&str] = &[
607 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":"","tool_calls":[{"function":{"name":"search","arguments":{"q":"vak"}}}]},"done":false}"#,
608 r#"{"model":"gemma3:e2b","message":{"role":"assistant","content":""},"done":true,"done_reason":"stop","prompt_eval_count":10,"eval_count":5}"#,
609 ];
610
611 #[test]
612 fn stream_fixture_generates_call_id_when_ollama_omits_one() {
613 let mut acc = Accumulator::new("gemma3:e2b");
614 let mut end = None;
615 for line in TOOL_CALL_LINES {
616 if let Some(event) = acc.convert(line).unwrap()
617 && let StreamEvent::End { message } = event
618 {
619 end = Some(message);
620 }
621 }
622 let msg = end.unwrap();
623 assert_eq!(msg.stop_reason, StopReason::ToolUse);
624 let ContentBlock::ToolUse { id, name, input } = &msg.content[0] else {
625 panic!("expected a tool_use block");
626 };
627 assert_eq!(id, "call_0");
628 assert_eq!(name, "search");
629 assert_eq!(input, &serde_json::json!({"q": "vak"}));
630 }
631
632 #[test]
633 fn ndjson_decoder_splits_on_newlines_across_pushes() {
634 let mut decoder = NdjsonDecoder::default();
635 decoder.push(b"{\"a\":1}\n{\"b\":");
636 assert_eq!(decoder.next_line().as_deref(), Some(r#"{"a":1}"#));
637 assert_eq!(decoder.next_line(), None);
638 decoder.push(b"2}\n");
639 assert_eq!(decoder.next_line().as_deref(), Some(r#"{"b":2}"#));
640 }
641}
642

Indexing the workspace…