provider_core/
ollama_chat.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3//! Transport-independent Ollama `/api/chat` request and NDJSON contracts.
4
5use router_protocol::{
6    AttemptId, CanonicalRequest, CatalogSourceKind, ContentBlock, InputItem, MessageRole,
7    PhaseKind, ReasoningIntent, RouteIdentity, RouteIdentityError, ToolChoice, TransportKind,
8    ollama_cloud_provider_neutral_model,
9};
10use serde_json::{Map, Value, json};
11use std::collections::HashSet;
12use std::fmt;
13
14pub const DEFAULT_MAX_NDJSON_LINE_BYTES: usize = 256 * 1024;
15pub const MAX_RETRY_AFTER_SECONDS: u32 = 86_400;
16
17#[derive(Clone, Copy, Debug, PartialEq, Eq)]
18pub enum HttpMethod {
19    Post,
20}
21
22#[derive(Clone, Copy, Debug, PartialEq, Eq)]
23pub struct NetworkPolicy {
24    pub allow_proxy: bool,
25    pub allow_dns: bool,
26    pub follow_redirects: bool,
27    pub require_tls: bool,
28}
29
30#[derive(Clone, PartialEq, Eq)]
31pub struct SecretReferenceMarker(String);
32
33impl SecretReferenceMarker {
34    pub fn as_str(&self) -> &str {
35        &self.0
36    }
37}
38
39impl fmt::Debug for SecretReferenceMarker {
40    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
41        formatter.write_str("SecretReferenceMarker([REDACTED])")
42    }
43}
44
45#[derive(Clone, PartialEq, Eq)]
46pub enum OllamaAuthentication {
47    None,
48    BearerSecretReference(SecretReferenceMarker),
49}
50
51impl fmt::Debug for OllamaAuthentication {
52    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
53        match self {
54            Self::None => formatter.write_str("None"),
55            Self::BearerSecretReference(_) => {
56                formatter.write_str("BearerSecretReference([REDACTED])")
57            }
58        }
59    }
60}
61
62#[derive(Clone, PartialEq)]
63pub struct OllamaRequestSpec {
64    pub method: HttpMethod,
65    pub origin: String,
66    pub path: String,
67    pub content_type: &'static str,
68    pub accept: &'static str,
69    pub authentication: OllamaAuthentication,
70    pub network: NetworkPolicy,
71    body: Value,
72}
73
74impl OllamaRequestSpec {
75    pub fn body(&self) -> &Value {
76        &self.body
77    }
78
79    pub fn body_bytes(&self) -> Result<Vec<u8>, serde_json::Error> {
80        serde_json::to_vec(&self.body)
81    }
82}
83
84impl fmt::Debug for OllamaRequestSpec {
85    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
86        formatter
87            .debug_struct("OllamaRequestSpec")
88            .field("method", &self.method)
89            .field("origin", &self.origin)
90            .field("path", &self.path)
91            .field("content_type", &self.content_type)
92            .field("accept", &self.accept)
93            .field("authentication", &self.authentication)
94            .field("network", &self.network)
95            .field("body", &"[REDACTED]")
96            .finish()
97    }
98}
99
100#[derive(Clone, Copy, Debug, PartialEq, Eq)]
101pub enum OllamaRequestErrorKind {
102    InvalidRouteIdentity,
103    UnsupportedRoute,
104    UnverifiedCloudAlias,
105    InvalidCanonicalRequest,
106    UnsupportedInput,
107    UnsupportedToolChoice,
108    AmbiguousToolName,
109}
110
111#[derive(Clone, Copy, Debug, PartialEq, Eq)]
112pub struct OllamaRequestError {
113    pub kind: OllamaRequestErrorKind,
114}
115
116impl fmt::Display for OllamaRequestError {
117    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
118        formatter.write_str("the Ollama chat request could not be prepared")
119    }
120}
121
122impl std::error::Error for OllamaRequestError {}
123
124impl From<RouteIdentityError> for OllamaRequestError {
125    fn from(_: RouteIdentityError) -> Self {
126        Self {
127            kind: OllamaRequestErrorKind::InvalidRouteIdentity,
128        }
129    }
130}
131
132pub fn build_request(
133    route: &RouteIdentity,
134    request: &CanonicalRequest,
135) -> Result<OllamaRequestSpec, OllamaRequestError> {
136    route.validate()?;
137    request.validate().map_err(|_| OllamaRequestError {
138        kind: OllamaRequestErrorKind::InvalidCanonicalRequest,
139    })?;
140
141    let (authentication, network) = match (route.phase, route.transport) {
142        (PhaseKind::OllamaCloud, TransportKind::OllamaSignedInRelay) => {
143            require_cloud_alias(route)?;
144            (
145                OllamaAuthentication::None,
146                NetworkPolicy {
147                    allow_proxy: false,
148                    allow_dns: false,
149                    follow_redirects: false,
150                    require_tls: false,
151                },
152            )
153        }
154        (PhaseKind::OllamaCloud, TransportKind::DirectHttps) => {
155            require_cloud_alias(route)?;
156            let secret_ref = route.credential_ref.clone().ok_or(OllamaRequestError {
157                kind: OllamaRequestErrorKind::InvalidRouteIdentity,
158            })?;
159            (
160                OllamaAuthentication::BearerSecretReference(SecretReferenceMarker(secret_ref)),
161                NetworkPolicy {
162                    allow_proxy: false,
163                    allow_dns: true,
164                    follow_redirects: false,
165                    require_tls: true,
166                },
167            )
168        }
169        (PhaseKind::OllamaLocal, TransportKind::OllamaLocalDaemon) => (
170            OllamaAuthentication::None,
171            NetworkPolicy {
172                allow_proxy: false,
173                allow_dns: false,
174                follow_redirects: false,
175                require_tls: false,
176            },
177        ),
178        _ => {
179            return Err(OllamaRequestError {
180                kind: OllamaRequestErrorKind::UnsupportedRoute,
181            });
182        }
183    };
184
185    let body = build_body(route, request)?;
186    Ok(OllamaRequestSpec {
187        method: HttpMethod::Post,
188        origin: route.endpoint.origin.clone(),
189        path: route.endpoint.path.clone(),
190        content_type: "application/json",
191        accept: "application/x-ndjson",
192        authentication,
193        network,
194        body,
195    })
196}
197
198fn require_cloud_alias(route: &RouteIdentity) -> Result<(), OllamaRequestError> {
199    let transport_model_matches =
200        ollama_cloud_provider_neutral_model(&route.model_id, route.transport).is_ok();
201    if route.catalog_source == CatalogSourceKind::OllamaCloudCatalog && transport_model_matches {
202        Ok(())
203    } else {
204        Err(OllamaRequestError {
205            kind: OllamaRequestErrorKind::UnverifiedCloudAlias,
206        })
207    }
208}
209
210fn build_body(
211    route: &RouteIdentity,
212    request: &CanonicalRequest,
213) -> Result<Value, OllamaRequestError> {
214    let mut object = Map::new();
215    object.insert("model".to_owned(), Value::String(route.model_id.clone()));
216    object.insert(
217        "messages".to_owned(),
218        Value::Array(build_messages(request)?),
219    );
220    object.insert("stream".to_owned(), Value::Bool(true));
221    object.insert(
222        "think".to_owned(),
223        Value::Bool(
224            request.reasoning == ReasoningIntent::PublicSummary
225                && route.reasoning_modes.iter().any(|mode| mode == "reasoning"),
226        ),
227    );
228
229    if !request.tools.is_empty() && request.tool_choice != ToolChoice::None {
230        if !matches!(request.tool_choice, ToolChoice::Auto) {
231            return Err(OllamaRequestError {
232                kind: OllamaRequestErrorKind::UnsupportedToolChoice,
233            });
234        }
235        let mut names = HashSet::new();
236        let mut tools = Vec::with_capacity(request.tools.len());
237        for tool in &request.tools {
238            let name = provider_tool_name(tool.namespace.as_deref(), &tool.name);
239            if !names.insert(name.clone()) {
240                return Err(OllamaRequestError {
241                    kind: OllamaRequestErrorKind::AmbiguousToolName,
242                });
243            }
244            tools.push(json!({
245                "type": "function",
246                "function": {
247                    "name": name,
248                    "description": tool.description,
249                    "parameters": tool.parameters,
250                }
251            }));
252        }
253        object.insert("tools".to_owned(), Value::Array(tools));
254    }
255    if let Some(maximum) = request.max_output_tokens {
256        object.insert("options".to_owned(), json!({"num_predict": maximum}));
257    }
258    if let Some(format) = &request.structured_output {
259        object.insert("format".to_owned(), format.schema.clone());
260    }
261    Ok(Value::Object(object))
262}
263
264fn build_messages(request: &CanonicalRequest) -> Result<Vec<Value>, OllamaRequestError> {
265    let mut messages = Vec::with_capacity(request.input.len());
266    for item in &request.input {
267        match item {
268            InputItem::Message { role, content } => {
269                let mut text = String::new();
270                for block in content {
271                    match block {
272                        ContentBlock::Text { text: part } => text.push_str(part),
273                        ContentBlock::ImageReference { .. } => {
274                            return Err(OllamaRequestError {
275                                kind: OllamaRequestErrorKind::UnsupportedInput,
276                            });
277                        }
278                    }
279                }
280                messages.push(json!({"role": role_name(*role), "content": text}));
281            }
282            InputItem::FunctionCall {
283                call_id: _,
284                namespace,
285                name,
286                arguments,
287            } => messages.push(json!({
288                "role": "assistant",
289                "content": "",
290                "tool_calls": [{
291                    "function": {
292                        "name": provider_tool_name(namespace.as_deref(), name),
293                        "arguments": arguments,
294                    }
295                }]
296            })),
297            InputItem::FunctionCallOutput { call_id: _, output } => messages.push(json!({
298                "role": "tool",
299                "content": output.to_string(),
300            })),
301        }
302    }
303    Ok(messages)
304}
305
306const fn role_name(role: MessageRole) -> &'static str {
307    match role {
308        MessageRole::Developer | MessageRole::System => "system",
309        MessageRole::User => "user",
310        MessageRole::Assistant => "assistant",
311    }
312}
313
314fn provider_tool_name(namespace: Option<&str>, name: &str) -> String {
315    namespace.map_or_else(
316        || name.to_owned(),
317        |namespace| format!("{namespace}__{name}"),
318    )
319}
320
321#[derive(Clone, Debug, Default, PartialEq, Eq)]
322pub struct OllamaUsage {
323    pub prompt_tokens: Option<u64>,
324    pub completion_tokens: Option<u64>,
325}
326
327#[derive(Clone, Debug, Default, PartialEq, Eq)]
328pub struct OllamaDurations {
329    pub total_ns: Option<u64>,
330    pub load_ns: Option<u64>,
331    pub prompt_eval_ns: Option<u64>,
332    pub eval_ns: Option<u64>,
333}
334
335#[derive(Clone, Debug, PartialEq, Eq)]
336pub struct OllamaTerminal {
337    pub model: Option<String>,
338    pub done_reason: Option<String>,
339    pub usage: OllamaUsage,
340    pub durations: OllamaDurations,
341}
342
343#[derive(Clone, Debug, PartialEq)]
344pub enum OllamaStreamEvent {
345    Content {
346        text: String,
347    },
348    Thinking {
349        text: String,
350    },
351    ToolCall {
352        id: String,
353        ordinal: u64,
354        name: String,
355        arguments: Value,
356    },
357    Completed(OllamaTerminal),
358}
359
360#[derive(Clone, Copy, Debug, PartialEq, Eq)]
361pub enum OllamaStreamErrorKind {
362    InvalidUtf8,
363    InvalidJson,
364    InvalidFrame,
365    LineTooLarge,
366    ProviderError,
367    MissingTerminal,
368    DataAfterTerminal,
369    ParserFailed,
370}
371
372#[derive(Clone, Copy, Debug, PartialEq, Eq)]
373pub struct OllamaStreamError {
374    pub kind: OllamaStreamErrorKind,
375    pub byte_offset: u64,
376}
377
378impl fmt::Display for OllamaStreamError {
379    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
380        formatter.write_str("the Ollama response stream was invalid")
381    }
382}
383
384impl std::error::Error for OllamaStreamError {}
385
386pub struct OllamaNdjsonParser {
387    attempt_ref: String,
388    buffer: Vec<u8>,
389    max_line_bytes: usize,
390    byte_offset: u64,
391    next_tool_ordinal: u64,
392    terminal: bool,
393    failed: bool,
394}
395
396impl fmt::Debug for OllamaNdjsonParser {
397    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
398        formatter
399            .debug_struct("OllamaNdjsonParser")
400            .field("buffered_bytes", &self.buffer.len())
401            .field("max_line_bytes", &self.max_line_bytes)
402            .field("byte_offset", &self.byte_offset)
403            .field("next_tool_ordinal", &self.next_tool_ordinal)
404            .field("terminal", &self.terminal)
405            .field("failed", &self.failed)
406            .finish()
407    }
408}
409
410impl OllamaNdjsonParser {
411    pub fn new(attempt_id: &AttemptId) -> Self {
412        Self::with_max_line_bytes(attempt_id, DEFAULT_MAX_NDJSON_LINE_BYTES)
413    }
414
415    pub fn with_max_line_bytes(attempt_id: &AttemptId, max_line_bytes: usize) -> Self {
416        Self {
417            attempt_ref: attempt_id.as_str().to_owned(),
418            buffer: Vec::new(),
419            max_line_bytes: max_line_bytes.max(1),
420            byte_offset: 0,
421            next_tool_ordinal: 0,
422            terminal: false,
423            failed: false,
424        }
425    }
426
427    pub fn feed(&mut self, bytes: &[u8]) -> Result<Vec<OllamaStreamEvent>, OllamaStreamError> {
428        if self.failed {
429            return Err(self.error(OllamaStreamErrorKind::ParserFailed));
430        }
431        let mut events = Vec::new();
432        for &byte in bytes {
433            self.byte_offset = self.byte_offset.saturating_add(1);
434            if byte == b'\n' {
435                let mut line = std::mem::take(&mut self.buffer);
436                if line.last() == Some(&b'\r') {
437                    line.pop();
438                }
439                if let Err(error) = self.parse_line(&line, &mut events) {
440                    self.failed = true;
441                    return Err(error);
442                }
443            } else {
444                self.buffer.push(byte);
445                if self.buffer.len() > self.max_line_bytes {
446                    self.failed = true;
447                    return Err(self.error(OllamaStreamErrorKind::LineTooLarge));
448                }
449            }
450        }
451        Ok(events)
452    }
453
454    pub fn finish(&mut self) -> Result<Vec<OllamaStreamEvent>, OllamaStreamError> {
455        if self.failed {
456            return Err(self.error(OllamaStreamErrorKind::ParserFailed));
457        }
458        let mut events = Vec::new();
459        if !self.buffer.is_empty() {
460            let mut line = std::mem::take(&mut self.buffer);
461            if line.last() == Some(&b'\r') {
462                line.pop();
463            }
464            if let Err(error) = self.parse_line(&line, &mut events) {
465                self.failed = true;
466                return Err(error);
467            }
468        }
469        if !self.terminal {
470            self.failed = true;
471            return Err(self.error(OllamaStreamErrorKind::MissingTerminal));
472        }
473        Ok(events)
474    }
475
476    fn parse_line(
477        &mut self,
478        line: &[u8],
479        events: &mut Vec<OllamaStreamEvent>,
480    ) -> Result<(), OllamaStreamError> {
481        if line.is_empty() {
482            return Ok(());
483        }
484        if self.terminal {
485            return Err(self.error(OllamaStreamErrorKind::DataAfterTerminal));
486        }
487        let text = std::str::from_utf8(line)
488            .map_err(|_| self.error(OllamaStreamErrorKind::InvalidUtf8))?;
489        let value: Value = serde_json::from_str(text)
490            .map_err(|_| self.error(OllamaStreamErrorKind::InvalidJson))?;
491        let object = value
492            .as_object()
493            .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
494        if object.contains_key("error") {
495            return Err(self.error(OllamaStreamErrorKind::ProviderError));
496        }
497        if let Some(message) = object.get("message") {
498            self.parse_message(message, events)?;
499        }
500        match object.get("done") {
501            Some(Value::Bool(true)) => {
502                let terminal = OllamaTerminal {
503                    model: optional_string(object, "model")?,
504                    done_reason: optional_string(object, "done_reason")?,
505                    usage: OllamaUsage {
506                        prompt_tokens: optional_u64(object, "prompt_eval_count")?,
507                        completion_tokens: optional_u64(object, "eval_count")?,
508                    },
509                    durations: OllamaDurations {
510                        total_ns: optional_u64(object, "total_duration")?,
511                        load_ns: optional_u64(object, "load_duration")?,
512                        prompt_eval_ns: optional_u64(object, "prompt_eval_duration")?,
513                        eval_ns: optional_u64(object, "eval_duration")?,
514                    },
515                };
516                events.push(OllamaStreamEvent::Completed(terminal));
517                self.terminal = true;
518            }
519            Some(Value::Bool(false)) | None => {}
520            Some(_) => return Err(self.error(OllamaStreamErrorKind::InvalidFrame)),
521        }
522        Ok(())
523    }
524
525    fn parse_message(
526        &mut self,
527        value: &Value,
528        events: &mut Vec<OllamaStreamEvent>,
529    ) -> Result<(), OllamaStreamError> {
530        let message = value
531            .as_object()
532            .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
533        if let Some(content) = message.get("content") {
534            let content = content
535                .as_str()
536                .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
537            if !content.is_empty() {
538                events.push(OllamaStreamEvent::Content {
539                    text: content.to_owned(),
540                });
541            }
542        }
543        if let Some(thinking) = message.get("thinking") {
544            let thinking = thinking
545                .as_str()
546                .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
547            if !thinking.is_empty() {
548                events.push(OllamaStreamEvent::Thinking {
549                    text: thinking.to_owned(),
550                });
551            }
552        }
553        if let Some(tool_calls) = message.get("tool_calls") {
554            let tool_calls = tool_calls
555                .as_array()
556                .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
557            for tool_call in tool_calls {
558                let function = tool_call
559                    .get("function")
560                    .and_then(Value::as_object)
561                    .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
562                let name = function
563                    .get("name")
564                    .and_then(Value::as_str)
565                    .filter(|name| !name.is_empty())
566                    .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
567                let arguments = function
568                    .get("arguments")
569                    .filter(|arguments| arguments.is_object())
570                    .ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
571                let ordinal = self.next_tool_ordinal;
572                self.next_tool_ordinal = self.next_tool_ordinal.saturating_add(1);
573                events.push(OllamaStreamEvent::ToolCall {
574                    id: synthesized_tool_id(&self.attempt_ref, ordinal),
575                    ordinal,
576                    name: name.to_owned(),
577                    arguments: arguments.clone(),
578                });
579            }
580        }
581        Ok(())
582    }
583
584    fn error(&self, kind: OllamaStreamErrorKind) -> OllamaStreamError {
585        OllamaStreamError {
586            kind,
587            byte_offset: self.byte_offset,
588        }
589    }
590}
591
592fn optional_string(
593    object: &Map<String, Value>,
594    name: &str,
595) -> Result<Option<String>, OllamaStreamError> {
596    match object.get(name) {
597        None | Some(Value::Null) => Ok(None),
598        Some(Value::String(value)) => Ok(Some(value.clone())),
599        Some(_) => Err(OllamaStreamError {
600            kind: OllamaStreamErrorKind::InvalidFrame,
601            byte_offset: 0,
602        }),
603    }
604}
605
606fn optional_u64(object: &Map<String, Value>, name: &str) -> Result<Option<u64>, OllamaStreamError> {
607    match object.get(name) {
608        None | Some(Value::Null) => Ok(None),
609        Some(value) => value.as_u64().map(Some).ok_or(OllamaStreamError {
610            kind: OllamaStreamErrorKind::InvalidFrame,
611            byte_offset: 0,
612        }),
613    }
614}
615
616fn synthesized_tool_id(attempt_ref: &str, ordinal: u64) -> String {
617    const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
618    const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
619    let mut left = FNV_OFFSET;
620    let mut right = FNV_OFFSET ^ 0x9e37_79b9_7f4a_7c15;
621    for byte in attempt_ref
622        .bytes()
623        .chain(*b":")
624        .chain(ordinal.to_be_bytes())
625    {
626        left ^= u64::from(byte);
627        left = left.wrapping_mul(FNV_PRIME);
628        right ^= u64::from(byte.rotate_left(1));
629        right = right.wrapping_mul(FNV_PRIME);
630    }
631    format!("call_{left:016x}{right:016x}")
632}
633
634#[derive(Clone, Copy, Debug, PartialEq, Eq)]
635pub enum ReplayBoundary {
636    PublicContent,
637    VisibleToolCall,
638    ForwardedToolResult,
639    AmbiguousDispatch,
640}
641
642#[derive(Clone, Copy, Debug, PartialEq, Eq)]
643pub enum ReplayDecision {
644    Allowed,
645    Denied(ReplayBoundary),
646}
647
648#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
649pub struct OllamaReplayGuard {
650    boundary: Option<ReplayBoundary>,
651    body_written: bool,
652}
653
654impl OllamaReplayGuard {
655    pub fn mark_body_written(&mut self) {
656        self.body_written = true;
657    }
658
659    pub fn observe(&mut self, event: &OllamaStreamEvent) {
660        match event {
661            OllamaStreamEvent::Content { text } if !text.is_empty() => {
662                self.cross(ReplayBoundary::PublicContent);
663            }
664            OllamaStreamEvent::ToolCall { .. } => self.cross(ReplayBoundary::VisibleToolCall),
665            OllamaStreamEvent::Content { .. }
666            | OllamaStreamEvent::Thinking { .. }
667            | OllamaStreamEvent::Completed(_) => {}
668        }
669    }
670
671    pub fn mark_forwarded_tool_result(&mut self) {
672        self.cross(ReplayBoundary::ForwardedToolResult);
673    }
674
675    pub fn mark_transport_outcome_unknown(&mut self) {
676        if self.body_written {
677            self.cross(ReplayBoundary::AmbiguousDispatch);
678        }
679    }
680
681    pub fn decision(&self) -> ReplayDecision {
682        self.boundary
683            .map_or(ReplayDecision::Allowed, ReplayDecision::Denied)
684    }
685
686    fn cross(&mut self, boundary: ReplayBoundary) {
687        if self.boundary.is_none() {
688            self.boundary = Some(boundary);
689        }
690    }
691}
692
693#[derive(Clone, Copy, Debug, PartialEq, Eq)]
694pub enum OllamaFailureClass {
695    InvalidRequest,
696    Authentication,
697    PaymentRequired,
698    Authorization,
699    NotFound,
700    Timeout,
701    Unprocessable,
702    RateLimited,
703    UpstreamServer,
704    UnexpectedStatus,
705    MismatchedContentType,
706    RedirectRejected,
707    Cancelled,
708    TransportTimeout,
709    AmbiguousDispatch,
710}
711
712#[derive(Clone, Copy, Debug, PartialEq, Eq)]
713pub struct BoundedRetryAfter {
714    pub seconds: u32,
715    pub capped: bool,
716}
717
718impl BoundedRetryAfter {
719    pub fn parse(value: Option<&str>) -> Option<Self> {
720        let parsed = value?.trim().parse::<u64>().ok()?;
721        Some(Self {
722            seconds: parsed.min(u64::from(MAX_RETRY_AFTER_SECONDS)) as u32,
723            capped: parsed > u64::from(MAX_RETRY_AFTER_SECONDS),
724        })
725    }
726}
727
728#[derive(Clone, Copy, Debug, PartialEq, Eq)]
729pub struct OllamaFailure {
730    pub class: OllamaFailureClass,
731    pub status: Option<u16>,
732    pub retry_after: Option<BoundedRetryAfter>,
733}
734
735pub fn normalize_http_response(
736    status: u16,
737    content_type: Option<&str>,
738    retry_after: Option<&str>,
739    redirected: bool,
740) -> Result<(), OllamaFailure> {
741    if redirected || (300..400).contains(&status) {
742        return Err(failure(
743            OllamaFailureClass::RedirectRejected,
744            Some(status),
745            retry_after,
746        ));
747    }
748    if status != 200 {
749        let class = match status {
750            400 => OllamaFailureClass::InvalidRequest,
751            401 => OllamaFailureClass::Authentication,
752            402 => OllamaFailureClass::PaymentRequired,
753            403 => OllamaFailureClass::Authorization,
754            404 => OllamaFailureClass::NotFound,
755            408 => OllamaFailureClass::Timeout,
756            422 => OllamaFailureClass::Unprocessable,
757            429 => OllamaFailureClass::RateLimited,
758            500..=599 => OllamaFailureClass::UpstreamServer,
759            _ => OllamaFailureClass::UnexpectedStatus,
760        };
761        return Err(failure(class, Some(status), retry_after));
762    }
763    let media_type = content_type
764        .and_then(|value| value.split(';').next())
765        .map(str::trim);
766    if !matches!(
767        media_type,
768        Some("application/x-ndjson" | "application/json")
769    ) {
770        return Err(failure(
771            OllamaFailureClass::MismatchedContentType,
772            Some(status),
773            None,
774        ));
775    }
776    Ok(())
777}
778
779#[derive(Clone, Copy, Debug, PartialEq, Eq)]
780pub enum BodyWriteState {
781    NotWritten,
782    Written,
783    Unknown,
784}
785
786#[derive(Clone, Copy, Debug, PartialEq, Eq)]
787pub enum TransportFailureSignal {
788    Cancelled,
789    Timeout,
790}
791
792pub fn normalize_transport_failure(
793    signal: TransportFailureSignal,
794    body_write: BodyWriteState,
795) -> OllamaFailure {
796    let class = match body_write {
797        BodyWriteState::Written | BodyWriteState::Unknown => OllamaFailureClass::AmbiguousDispatch,
798        BodyWriteState::NotWritten => match signal {
799            TransportFailureSignal::Cancelled => OllamaFailureClass::Cancelled,
800            TransportFailureSignal::Timeout => OllamaFailureClass::TransportTimeout,
801        },
802    };
803    failure(class, None, None)
804}
805
806fn failure(
807    class: OllamaFailureClass,
808    status: Option<u16>,
809    retry_after: Option<&str>,
810) -> OllamaFailure {
811    OllamaFailure {
812        class,
813        status,
814        retry_after: BoundedRetryAfter::parse(retry_after),
815    }
816}
817
818#[cfg(test)]
819mod tests {
820    use super::*;
821    use router_protocol::{
822        BillingAuthorityClass, CapabilityRequirements, CredentialSourceKind, EndpointIdentity,
823        ExecutionLocation, FunctionTool, HarnessInstanceId, M1_WIRE_VERSION, ModelNamespace,
824        PhasePlacement, ProviderService, SessionId, TurnId,
825    };
826
827    fn route(phase: PhaseKind, transport: TransportKind, model: &str) -> RouteIdentity {
828        let cloud = phase == PhaseKind::OllamaCloud;
829        RouteIdentity {
830            phase,
831            placement: PhasePlacement::Builtin,
832            provider: if cloud {
833                ProviderService::OllamaCloud
834            } else {
835                ProviderService::OllamaLocal
836            },
837            transport,
838            endpoint: EndpointIdentity {
839                origin: if transport == TransportKind::DirectHttps {
840                    "https://ollama.com".to_owned()
841                } else {
842                    "http://127.0.0.1:11434".to_owned()
843                },
844                path: "/api/chat".to_owned(),
845            },
846            credential_source: if transport == TransportKind::DirectHttps {
847                CredentialSourceKind::SecretRef
848            } else if cloud {
849                CredentialSourceKind::OllamaDaemonSession
850            } else {
851                CredentialSourceKind::None
852            },
853            credential_ref: (transport == TransportKind::DirectHttps)
854                .then(|| "keychain:ollama-cloud-canary".to_owned()),
855            credential_refs: if transport == TransportKind::DirectHttps {
856                vec!["keychain:ollama-cloud-canary".to_owned()]
857            } else {
858                Vec::new()
859            },
860            catalog_source: if cloud {
861                CatalogSourceKind::OllamaCloudCatalog
862            } else {
863                CatalogSourceKind::OllamaLocalInventory
864            },
865            catalog_ref: if cloud {
866                "catalog:ollama-cloud:verified".to_owned()
867            } else {
868                "inventory:local:approved".to_owned()
869            },
870            billing_authority: if cloud {
871                BillingAuthorityClass::OllamaCloud
872            } else {
873                BillingAuthorityClass::LocalCompute
874            },
875            model_namespace: if cloud {
876                ModelNamespace::OllamaCloud
877            } else {
878                ModelNamespace::OllamaLocal
879            },
880            model_id: model.to_owned(),
881            execution_location: if cloud {
882                ExecutionLocation::RemoteCloud
883            } else {
884                ExecutionLocation::LocalMachine
885            },
886            account_ref: if cloud {
887                "account:ollama-cloud".to_owned()
888            } else {
889                "account:local".to_owned()
890            },
891            capacity_domain_ref: if cloud {
892                "capacity:ollama-cloud".to_owned()
893            } else {
894                "capacity:local".to_owned()
895            },
896            local_execution_proof: !cloud,
897            approved_local_inventory: if cloud {
898                Vec::new()
899            } else {
900                vec![model.to_owned()]
901            },
902            reasoning_modes: if cloud || model.starts_with("qwen") {
903                vec!["reasoning".to_owned()]
904            } else {
905                Vec::new()
906            },
907        }
908    }
909
910    fn request(prompt: &str) -> CanonicalRequest {
911        CanonicalRequest {
912            wire_version: M1_WIRE_VERSION.to_owned(),
913            model: "router-auto".to_owned(),
914            stream: true,
915            harness_instance: HarnessInstanceId::parse("fixture-cli").expect("fixture"),
916            session_id: SessionId::parse("session-m31").expect("fixture"),
917            turn_id: TurnId::parse("turn-m31").expect("fixture"),
918            input: vec![InputItem::Message {
919                role: MessageRole::User,
920                content: vec![ContentBlock::Text {
921                    text: prompt.to_owned(),
922                }],
923            }],
924            tools: vec![FunctionTool {
925                namespace: Some("weather".to_owned()),
926                name: "lookup".to_owned(),
927                description: "lookup weather".to_owned(),
928                parameters: json!({"type":"object"}),
929                strict: true,
930            }],
931            tool_choice: ToolChoice::Auto,
932            structured_output: None,
933            reasoning: ReasoningIntent::PublicSummary,
934            max_output_tokens: Some(64),
935            max_spend_micro_usd: None,
936            capabilities: CapabilityRequirements::default(),
937        }
938    }
939
940    #[test]
941    fn exact_relay_and_direct_request_specs_are_io_free_and_redacted() {
942        let relay = build_request(
943            &route(
944                PhaseKind::OllamaCloud,
945                TransportKind::OllamaSignedInRelay,
946                "glm-5.2:cloud",
947            ),
948            &request("prompt-canary"),
949        )
950        .expect("relay request");
951        assert_eq!(relay.method, HttpMethod::Post);
952        assert_eq!(relay.origin, "http://127.0.0.1:11434");
953        assert_eq!(relay.path, "/api/chat");
954        assert_eq!(relay.authentication, OllamaAuthentication::None);
955        assert!(!relay.network.allow_proxy);
956        assert!(!relay.network.allow_dns);
957        assert!(!relay.network.follow_redirects);
958        assert_eq!(
959            relay.body(),
960            &json!({
961                "model":"glm-5.2:cloud",
962                "messages":[{"role":"user","content":"prompt-canary"}],
963                "stream":true,
964                "think":true,
965                "tools":[{"type":"function","function":{
966                    "name":"weather__lookup",
967                    "description":"lookup weather",
968                    "parameters":{"type":"object"}
969                }}],
970                "options":{"num_predict":64}
971            })
972        );
973        let serialized = relay.body_bytes().expect("JSON body");
974        let wire = String::from_utf8(serialized).expect("JSON UTF-8");
975        for forbidden in ["/api/tags", "/api/pull", "/api/create", "/api/push"] {
976            assert!(!wire.contains(forbidden));
977        }
978
979        let direct = build_request(
980            &route(
981                PhaseKind::OllamaCloud,
982                TransportKind::DirectHttps,
983                "glm-5.2",
984            ),
985            &request("prompt-canary"),
986        )
987        .expect("direct request");
988        assert_eq!(direct.origin, "https://ollama.com");
989        assert_eq!(direct.path, "/api/chat");
990        assert!(direct.network.require_tls);
991        assert!(!direct.network.follow_redirects);
992        assert_eq!(direct.body()["model"], "glm-5.2");
993        let debug = format!("{direct:?}");
994        assert!(!debug.contains("keychain:ollama-cloud-canary"));
995        assert!(!debug.contains("prompt-canary"));
996
997        let hyphen_relay = build_request(
998            &route(
999                PhaseKind::OllamaCloud,
1000                TransportKind::OllamaSignedInRelay,
1001                "qwen3-coder:480b-cloud",
1002            ),
1003            &request("safe"),
1004        )
1005        .expect("canonical hyphen-cloud relay request");
1006        assert_eq!(hyphen_relay.body()["model"], "qwen3-coder:480b-cloud");
1007    }
1008
1009    #[test]
1010    fn cloud_model_names_are_transport_specific_and_local_rejects_cloud_alias() {
1011        let error = build_request(
1012            &route(
1013                PhaseKind::OllamaCloud,
1014                TransportKind::DirectHttps,
1015                "unverified-model:cloud",
1016            ),
1017            &request("safe"),
1018        )
1019        .expect_err("relay-only suffix on direct API");
1020        assert_eq!(error.kind, OllamaRequestErrorKind::InvalidRouteIdentity);
1021
1022        let error = build_request(
1023            &route(
1024                PhaseKind::OllamaCloud,
1025                TransportKind::DirectHttps,
1026                "qwen3-coder:480b-cloud",
1027            ),
1028            &request("safe"),
1029        )
1030        .expect_err("hyphen relay-only suffix on direct API");
1031        assert_eq!(error.kind, OllamaRequestErrorKind::InvalidRouteIdentity);
1032
1033        let error = build_request(
1034            &route(
1035                PhaseKind::OllamaCloud,
1036                TransportKind::OllamaSignedInRelay,
1037                "unverified-model",
1038            ),
1039            &request("safe"),
1040        )
1041        .expect_err("missing relay cloud suffix");
1042        assert_eq!(error.kind, OllamaRequestErrorKind::InvalidRouteIdentity);
1043
1044        let local = route(
1045            PhaseKind::OllamaLocal,
1046            TransportKind::OllamaLocalDaemon,
1047            "glm-5.2:cloud",
1048        );
1049        assert_eq!(
1050            build_request(&local, &request("safe"))
1051                .expect_err("local cloud alias")
1052                .kind,
1053            OllamaRequestErrorKind::InvalidRouteIdentity
1054        );
1055        let local = route(
1056            PhaseKind::OllamaLocal,
1057            TransportKind::OllamaLocalDaemon,
1058            "qwen3-coder:480b-cloud",
1059        );
1060        assert_eq!(
1061            build_request(&local, &request("safe"))
1062                .expect_err("local hyphen cloud alias")
1063                .kind,
1064            OllamaRequestErrorKind::InvalidRouteIdentity
1065        );
1066
1067        for invalid in [
1068            "model-cloud-cloud",
1069            "model:cloud-cloud",
1070            "model-cloud:cloud",
1071            "model:cloud:cloud",
1072        ] {
1073            assert_eq!(
1074                build_request(
1075                    &route(
1076                        PhaseKind::OllamaCloud,
1077                        TransportKind::OllamaSignedInRelay,
1078                        invalid,
1079                    ),
1080                    &request("safe"),
1081                )
1082                .expect_err("double or mixed cloud suffix")
1083                .kind,
1084                OllamaRequestErrorKind::InvalidRouteIdentity
1085            );
1086        }
1087
1088        let local = build_request(
1089            &route(
1090                PhaseKind::OllamaLocal,
1091                TransportKind::OllamaLocalDaemon,
1092                "qwen3:8b",
1093            ),
1094            &request("safe"),
1095        )
1096        .expect("approved local model");
1097        assert_eq!(local.origin, "http://127.0.0.1:11434");
1098        assert_eq!(local.path, "/api/chat");
1099        assert_eq!(local.authentication, OllamaAuthentication::None);
1100        assert!(!local.network.allow_proxy);
1101        assert!(!local.network.allow_dns);
1102        assert!(!local.network.follow_redirects);
1103
1104        let mut direct_without_secret = route(
1105            PhaseKind::OllamaCloud,
1106            TransportKind::DirectHttps,
1107            "glm-5.2",
1108        );
1109        direct_without_secret.credential_ref = None;
1110        assert_eq!(
1111            build_request(&direct_without_secret, &request("safe"))
1112                .expect_err("direct secret reference")
1113                .kind,
1114            OllamaRequestErrorKind::InvalidRouteIdentity
1115        );
1116    }
1117
1118    #[test]
1119    fn think_is_limited_to_models_that_declare_reasoning_support() {
1120        let mistral = build_request(
1121            &route(
1122                PhaseKind::OllamaLocal,
1123                TransportKind::OllamaLocalDaemon,
1124                "mistral:latest",
1125            ),
1126            &request("prompt-canary"),
1127        )
1128        .expect("approved local model");
1129        assert_eq!(mistral.body()["think"], false);
1130        let qwen = build_request(
1131            &route(
1132                PhaseKind::OllamaLocal,
1133                TransportKind::OllamaLocalDaemon,
1134                "qwen3:4b",
1135            ),
1136            &request("prompt-canary"),
1137        )
1138        .expect("approved local model");
1139        assert_eq!(qwen.body()["think"], true);
1140        let cloud = build_request(
1141            &route(
1142                PhaseKind::OllamaCloud,
1143                TransportKind::OllamaSignedInRelay,
1144                "glm-5.2:cloud",
1145            ),
1146            &request("prompt-canary"),
1147        )
1148        .expect("approved cloud relay");
1149        assert_eq!(cloud.body()["think"], true);
1150    }
1151
1152    #[test]
1153    fn ndjson_is_independent_of_every_byte_boundary_and_preserves_terminal_facts() {
1154        let transcript = concat!(
1155            "\n",
1156            "{\"model\":\"glm-5.2:cloud\",\"message\":{\"thinking\":\"plan\",\"content\":\"hé\"},\"done\":false}\r\n",
1157            "{\"message\":{\"content\":\"\",\"tool_calls\":[{\"function\":{\"name\":\"weather__lookup\",\"arguments\":{\"city\":\"Paris\"}}}]},\"done\":false}\n",
1158            "{\"model\":\"glm-5.2:cloud\",\"done_reason\":\"stop\",\"done\":true,\"prompt_eval_count\":11,\"eval_count\":7,\"total_duration\":101,\"load_duration\":2,\"prompt_eval_duration\":3,\"eval_duration\":5}"
1159        );
1160        let attempt = AttemptId::parse("attempt-fragmented").expect("attempt");
1161        let mut parser = OllamaNdjsonParser::new(&attempt);
1162        let mut events = Vec::new();
1163        for byte in transcript.as_bytes() {
1164            events.extend(parser.feed(std::slice::from_ref(byte)).expect("fragment"));
1165        }
1166        events.extend(parser.finish().expect("terminal without newline"));
1167        assert_eq!(events.len(), 4);
1168        assert_eq!(
1169            events[0],
1170            OllamaStreamEvent::Content {
1171                text: "hé".to_owned()
1172            }
1173        );
1174        assert_eq!(
1175            events[1],
1176            OllamaStreamEvent::Thinking {
1177                text: "plan".to_owned()
1178            }
1179        );
1180        let OllamaStreamEvent::ToolCall { id, ordinal, .. } = &events[2] else {
1181            panic!("tool event");
1182        };
1183        assert!(id.starts_with("call_"));
1184        assert_eq!(*ordinal, 0);
1185        let OllamaStreamEvent::Completed(terminal) = &events[3] else {
1186            panic!("terminal event");
1187        };
1188        assert_eq!(terminal.model.as_deref(), Some("glm-5.2:cloud"));
1189        assert_eq!(terminal.usage.prompt_tokens, Some(11));
1190        assert_eq!(terminal.usage.completion_tokens, Some(7));
1191        assert_eq!(terminal.durations.total_ns, Some(101));
1192
1193        let mut second = OllamaNdjsonParser::new(&attempt);
1194        let second_events = second.feed(transcript.as_bytes()).expect("whole chunk");
1195        let mut all = second_events;
1196        all.extend(second.finish().expect("terminal"));
1197        assert_eq!(events, all);
1198    }
1199
1200    #[test]
1201    fn ndjson_failures_are_typed_bounded_and_redacted() {
1202        let attempt = AttemptId::parse("attempt-errors").expect("attempt");
1203        let cases: &[(&[u8], OllamaStreamErrorKind)] = &[
1204            (b"{not-json}\n", OllamaStreamErrorKind::InvalidJson),
1205            (b"\xff\n", OllamaStreamErrorKind::InvalidUtf8),
1206            (
1207                b"{\"error\":\"provider-secret-canary\"}\n",
1208                OllamaStreamErrorKind::ProviderError,
1209            ),
1210        ];
1211        for (input, expected) in cases {
1212            let mut parser = OllamaNdjsonParser::new(&attempt);
1213            let error = parser.feed(input).expect_err("invalid stream");
1214            assert_eq!(error.kind, *expected);
1215            assert!(!format!("{error:?}").contains("provider-secret-canary"));
1216        }
1217
1218        let mut oversized = OllamaNdjsonParser::with_max_line_bytes(&attempt, 8);
1219        assert_eq!(
1220            oversized.feed(b"123456789").expect_err("bounded").kind,
1221            OllamaStreamErrorKind::LineTooLarge
1222        );
1223        let mut incomplete = OllamaNdjsonParser::new(&attempt);
1224        incomplete
1225            .feed(b"{\"message\":{\"content\":\"partial\"},\"done\":false}\n")
1226            .expect("partial frame");
1227        assert_eq!(
1228            incomplete.finish().expect_err("terminal required").kind,
1229            OllamaStreamErrorKind::MissingTerminal
1230        );
1231    }
1232
1233    #[test]
1234    fn replay_crosses_each_public_or_ambiguous_boundary_once() {
1235        let mut thinking = OllamaReplayGuard::default();
1236        thinking.observe(&OllamaStreamEvent::Thinking {
1237            text: "private".to_owned(),
1238        });
1239        assert_eq!(thinking.decision(), ReplayDecision::Allowed);
1240        thinking.observe(&OllamaStreamEvent::Content {
1241            text: "public".to_owned(),
1242        });
1243        assert_eq!(
1244            thinking.decision(),
1245            ReplayDecision::Denied(ReplayBoundary::PublicContent)
1246        );
1247
1248        let mut tool = OllamaReplayGuard::default();
1249        tool.observe(&OllamaStreamEvent::ToolCall {
1250            id: "call_fixture".to_owned(),
1251            ordinal: 0,
1252            name: "lookup".to_owned(),
1253            arguments: json!({}),
1254        });
1255        assert_eq!(
1256            tool.decision(),
1257            ReplayDecision::Denied(ReplayBoundary::VisibleToolCall)
1258        );
1259
1260        let mut result = OllamaReplayGuard::default();
1261        result.mark_forwarded_tool_result();
1262        assert_eq!(
1263            result.decision(),
1264            ReplayDecision::Denied(ReplayBoundary::ForwardedToolResult)
1265        );
1266
1267        let mut ambiguous = OllamaReplayGuard::default();
1268        ambiguous.mark_transport_outcome_unknown();
1269        assert_eq!(ambiguous.decision(), ReplayDecision::Allowed);
1270        ambiguous.mark_body_written();
1271        ambiguous.mark_transport_outcome_unknown();
1272        assert_eq!(
1273            ambiguous.decision(),
1274            ReplayDecision::Denied(ReplayBoundary::AmbiguousDispatch)
1275        );
1276    }
1277
1278    #[test]
1279    fn failures_cover_status_content_type_redirect_timeout_and_ambiguity() {
1280        let matrix = [
1281            (400, OllamaFailureClass::InvalidRequest),
1282            (401, OllamaFailureClass::Authentication),
1283            (402, OllamaFailureClass::PaymentRequired),
1284            (403, OllamaFailureClass::Authorization),
1285            (404, OllamaFailureClass::NotFound),
1286            (408, OllamaFailureClass::Timeout),
1287            (422, OllamaFailureClass::Unprocessable),
1288            (429, OllamaFailureClass::RateLimited),
1289            (503, OllamaFailureClass::UpstreamServer),
1290        ];
1291        for (status, class) in matrix {
1292            let error = normalize_http_response(status, None, Some("999999"), false)
1293                .expect_err("non-success");
1294            assert_eq!(error.class, class);
1295            assert_eq!(
1296                error.retry_after,
1297                Some(BoundedRetryAfter {
1298                    seconds: MAX_RETRY_AFTER_SECONDS,
1299                    capped: true,
1300                })
1301            );
1302        }
1303        assert!(
1304            normalize_http_response(
1305                200,
1306                Some("application/x-ndjson; charset=utf-8"),
1307                None,
1308                false
1309            )
1310            .is_ok()
1311        );
1312        assert_eq!(
1313            normalize_http_response(200, Some("text/html"), None, false)
1314                .expect_err("content type")
1315                .class,
1316            OllamaFailureClass::MismatchedContentType
1317        );
1318        assert_eq!(
1319            normalize_http_response(302, None, None, true)
1320                .expect_err("redirect")
1321                .class,
1322            OllamaFailureClass::RedirectRejected
1323        );
1324        assert_eq!(
1325            normalize_transport_failure(
1326                TransportFailureSignal::Cancelled,
1327                BodyWriteState::NotWritten,
1328            )
1329            .class,
1330            OllamaFailureClass::Cancelled
1331        );
1332        assert_eq!(
1333            normalize_transport_failure(TransportFailureSignal::Timeout, BodyWriteState::Written,)
1334                .class,
1335            OllamaFailureClass::AmbiguousDispatch
1336        );
1337    }
1338}
provider_core/ollama_chat.rs

provider_core/
ollama_chat.rs
