use router_protocol::{
AttemptId, CanonicalRequest, CatalogSourceKind, ContentBlock, InputItem, MessageRole,
PhaseKind, ReasoningIntent, RouteIdentity, RouteIdentityError, ToolChoice, TransportKind,
ollama_cloud_provider_neutral_model,
};
use serde_json::{Map, Value, json};
use std::collections::HashSet;
use std::fmt;
pub const DEFAULT_MAX_NDJSON_LINE_BYTES: usize = 256 * 1024;
pub const MAX_RETRY_AFTER_SECONDS: u32 = 86_400;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum HttpMethod {
Post,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct NetworkPolicy {
pub allow_proxy: bool,
pub allow_dns: bool,
pub follow_redirects: bool,
pub require_tls: bool,
}
#[derive(Clone, PartialEq, Eq)]
pub struct SecretReferenceMarker(String);
impl SecretReferenceMarker {
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for SecretReferenceMarker {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("SecretReferenceMarker([REDACTED])")
}
}
#[derive(Clone, PartialEq, Eq)]
pub enum OllamaAuthentication {
None,
BearerSecretReference(SecretReferenceMarker),
}
impl fmt::Debug for OllamaAuthentication {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::None => formatter.write_str("None"),
Self::BearerSecretReference(_) => {
formatter.write_str("BearerSecretReference([REDACTED])")
}
}
}
}
#[derive(Clone, PartialEq)]
pub struct OllamaRequestSpec {
pub method: HttpMethod,
pub origin: String,
pub path: String,
pub content_type: &'static str,
pub accept: &'static str,
pub authentication: OllamaAuthentication,
pub network: NetworkPolicy,
body: Value,
}
impl OllamaRequestSpec {
pub fn body(&self) -> &Value {
&self.body
}
pub fn body_bytes(&self) -> Result<Vec<u8>, serde_json::Error> {
serde_json::to_vec(&self.body)
}
}
impl fmt::Debug for OllamaRequestSpec {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("OllamaRequestSpec")
.field("method", &self.method)
.field("origin", &self.origin)
.field("path", &self.path)
.field("content_type", &self.content_type)
.field("accept", &self.accept)
.field("authentication", &self.authentication)
.field("network", &self.network)
.field("body", &"[REDACTED]")
.finish()
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OllamaRequestErrorKind {
InvalidRouteIdentity,
UnsupportedRoute,
UnverifiedCloudAlias,
InvalidCanonicalRequest,
UnsupportedInput,
UnsupportedToolChoice,
AmbiguousToolName,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OllamaRequestError {
pub kind: OllamaRequestErrorKind,
}
impl fmt::Display for OllamaRequestError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("the Ollama chat request could not be prepared")
}
}
impl std::error::Error for OllamaRequestError {}
impl From<RouteIdentityError> for OllamaRequestError {
fn from(_: RouteIdentityError) -> Self {
Self {
kind: OllamaRequestErrorKind::InvalidRouteIdentity,
}
}
}
pub fn build_request(
route: &RouteIdentity,
request: &CanonicalRequest,
) -> Result<OllamaRequestSpec, OllamaRequestError> {
route.validate()?;
request.validate().map_err(|_| OllamaRequestError {
kind: OllamaRequestErrorKind::InvalidCanonicalRequest,
})?;
let (authentication, network) = match (route.phase, route.transport) {
(PhaseKind::OllamaCloud, TransportKind::OllamaSignedInRelay) => {
require_cloud_alias(route)?;
(
OllamaAuthentication::None,
NetworkPolicy {
allow_proxy: false,
allow_dns: false,
follow_redirects: false,
require_tls: false,
},
)
}
(PhaseKind::OllamaCloud, TransportKind::DirectHttps) => {
require_cloud_alias(route)?;
let secret_ref = route.credential_ref.clone().ok_or(OllamaRequestError {
kind: OllamaRequestErrorKind::InvalidRouteIdentity,
})?;
(
OllamaAuthentication::BearerSecretReference(SecretReferenceMarker(secret_ref)),
NetworkPolicy {
allow_proxy: false,
allow_dns: true,
follow_redirects: false,
require_tls: true,
},
)
}
(PhaseKind::OllamaLocal, TransportKind::OllamaLocalDaemon) => (
OllamaAuthentication::None,
NetworkPolicy {
allow_proxy: false,
allow_dns: false,
follow_redirects: false,
require_tls: false,
},
),
_ => {
return Err(OllamaRequestError {
kind: OllamaRequestErrorKind::UnsupportedRoute,
});
}
};
let body = build_body(route, request)?;
Ok(OllamaRequestSpec {
method: HttpMethod::Post,
origin: route.endpoint.origin.clone(),
path: route.endpoint.path.clone(),
content_type: "application/json",
accept: "application/x-ndjson",
authentication,
network,
body,
})
}
fn require_cloud_alias(route: &RouteIdentity) -> Result<(), OllamaRequestError> {
let transport_model_matches =
ollama_cloud_provider_neutral_model(&route.model_id, route.transport).is_ok();
if route.catalog_source == CatalogSourceKind::OllamaCloudCatalog && transport_model_matches {
Ok(())
} else {
Err(OllamaRequestError {
kind: OllamaRequestErrorKind::UnverifiedCloudAlias,
})
}
}
fn build_body(
route: &RouteIdentity,
request: &CanonicalRequest,
) -> Result<Value, OllamaRequestError> {
let mut object = Map::new();
object.insert("model".to_owned(), Value::String(route.model_id.clone()));
object.insert(
"messages".to_owned(),
Value::Array(build_messages(request)?),
);
object.insert("stream".to_owned(), Value::Bool(true));
object.insert(
"think".to_owned(),
Value::Bool(
request.reasoning == ReasoningIntent::PublicSummary
&& route.reasoning_modes.iter().any(|mode| mode == "reasoning"),
),
);
if !request.tools.is_empty() && request.tool_choice != ToolChoice::None {
if !matches!(request.tool_choice, ToolChoice::Auto) {
return Err(OllamaRequestError {
kind: OllamaRequestErrorKind::UnsupportedToolChoice,
});
}
let mut names = HashSet::new();
let mut tools = Vec::with_capacity(request.tools.len());
for tool in &request.tools {
let name = provider_tool_name(tool.namespace.as_deref(), &tool.name);
if !names.insert(name.clone()) {
return Err(OllamaRequestError {
kind: OllamaRequestErrorKind::AmbiguousToolName,
});
}
tools.push(json!({
"type": "function",
"function": {
"name": name,
"description": tool.description,
"parameters": tool.parameters,
}
}));
}
object.insert("tools".to_owned(), Value::Array(tools));
}
if let Some(maximum) = request.max_output_tokens {
object.insert("options".to_owned(), json!({"num_predict": maximum}));
}
if let Some(format) = &request.structured_output {
object.insert("format".to_owned(), format.schema.clone());
}
Ok(Value::Object(object))
}
fn build_messages(request: &CanonicalRequest) -> Result<Vec<Value>, OllamaRequestError> {
let mut messages = Vec::with_capacity(request.input.len());
for item in &request.input {
match item {
InputItem::Message { role, content } => {
let mut text = String::new();
for block in content {
match block {
ContentBlock::Text { text: part } => text.push_str(part),
ContentBlock::ImageReference { .. } => {
return Err(OllamaRequestError {
kind: OllamaRequestErrorKind::UnsupportedInput,
});
}
}
}
messages.push(json!({"role": role_name(*role), "content": text}));
}
InputItem::FunctionCall {
call_id: _,
namespace,
name,
arguments,
} => messages.push(json!({
"role": "assistant",
"content": "",
"tool_calls": [{
"function": {
"name": provider_tool_name(namespace.as_deref(), name),
"arguments": arguments,
}
}]
})),
InputItem::FunctionCallOutput { call_id: _, output } => messages.push(json!({
"role": "tool",
"content": output.to_string(),
})),
}
}
Ok(messages)
}
const fn role_name(role: MessageRole) -> &'static str {
match role {
MessageRole::Developer | MessageRole::System => "system",
MessageRole::User => "user",
MessageRole::Assistant => "assistant",
}
}
fn provider_tool_name(namespace: Option<&str>, name: &str) -> String {
namespace.map_or_else(
|| name.to_owned(),
|namespace| format!("{namespace}__{name}"),
)
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct OllamaUsage {
pub prompt_tokens: Option<u64>,
pub completion_tokens: Option<u64>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct OllamaDurations {
pub total_ns: Option<u64>,
pub load_ns: Option<u64>,
pub prompt_eval_ns: Option<u64>,
pub eval_ns: Option<u64>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OllamaTerminal {
pub model: Option<String>,
pub done_reason: Option<String>,
pub usage: OllamaUsage,
pub durations: OllamaDurations,
}
#[derive(Clone, Debug, PartialEq)]
pub enum OllamaStreamEvent {
Content {
text: String,
},
Thinking {
text: String,
},
ToolCall {
id: String,
ordinal: u64,
name: String,
arguments: Value,
},
Completed(OllamaTerminal),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OllamaStreamErrorKind {
InvalidUtf8,
InvalidJson,
InvalidFrame,
LineTooLarge,
ProviderError,
MissingTerminal,
DataAfterTerminal,
ParserFailed,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OllamaStreamError {
pub kind: OllamaStreamErrorKind,
pub byte_offset: u64,
}
impl fmt::Display for OllamaStreamError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("the Ollama response stream was invalid")
}
}
impl std::error::Error for OllamaStreamError {}
pub struct OllamaNdjsonParser {
attempt_ref: String,
buffer: Vec<u8>,
max_line_bytes: usize,
byte_offset: u64,
next_tool_ordinal: u64,
terminal: bool,
failed: bool,
}
impl fmt::Debug for OllamaNdjsonParser {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("OllamaNdjsonParser")
.field("buffered_bytes", &self.buffer.len())
.field("max_line_bytes", &self.max_line_bytes)
.field("byte_offset", &self.byte_offset)
.field("next_tool_ordinal", &self.next_tool_ordinal)
.field("terminal", &self.terminal)
.field("failed", &self.failed)
.finish()
}
}
impl OllamaNdjsonParser {
pub fn new(attempt_id: &AttemptId) -> Self {
Self::with_max_line_bytes(attempt_id, DEFAULT_MAX_NDJSON_LINE_BYTES)
}
pub fn with_max_line_bytes(attempt_id: &AttemptId, max_line_bytes: usize) -> Self {
Self {
attempt_ref: attempt_id.as_str().to_owned(),
buffer: Vec::new(),
max_line_bytes: max_line_bytes.max(1),
byte_offset: 0,
next_tool_ordinal: 0,
terminal: false,
failed: false,
}
}
pub fn feed(&mut self, bytes: &[u8]) -> Result<Vec<OllamaStreamEvent>, OllamaStreamError> {
if self.failed {
return Err(self.error(OllamaStreamErrorKind::ParserFailed));
}
let mut events = Vec::new();
for &byte in bytes {
self.byte_offset = self.byte_offset.saturating_add(1);
if byte == b'\n' {
let mut line = std::mem::take(&mut self.buffer);
if line.last() == Some(&b'\r') {
line.pop();
}
if let Err(error) = self.parse_line(&line, &mut events) {
self.failed = true;
return Err(error);
}
} else {
self.buffer.push(byte);
if self.buffer.len() > self.max_line_bytes {
self.failed = true;
return Err(self.error(OllamaStreamErrorKind::LineTooLarge));
}
}
}
Ok(events)
}
pub fn finish(&mut self) -> Result<Vec<OllamaStreamEvent>, OllamaStreamError> {
if self.failed {
return Err(self.error(OllamaStreamErrorKind::ParserFailed));
}
let mut events = Vec::new();
if !self.buffer.is_empty() {
let mut line = std::mem::take(&mut self.buffer);
if line.last() == Some(&b'\r') {
line.pop();
}
if let Err(error) = self.parse_line(&line, &mut events) {
self.failed = true;
return Err(error);
}
}
if !self.terminal {
self.failed = true;
return Err(self.error(OllamaStreamErrorKind::MissingTerminal));
}
Ok(events)
}
fn parse_line(
&mut self,
line: &[u8],
events: &mut Vec<OllamaStreamEvent>,
) -> Result<(), OllamaStreamError> {
if line.is_empty() {
return Ok(());
}
if self.terminal {
return Err(self.error(OllamaStreamErrorKind::DataAfterTerminal));
}
let text = std::str::from_utf8(line)
.map_err(|_| self.error(OllamaStreamErrorKind::InvalidUtf8))?;
let value: Value = serde_json::from_str(text)
.map_err(|_| self.error(OllamaStreamErrorKind::InvalidJson))?;
let object = value
.as_object()
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
if object.contains_key("error") {
return Err(self.error(OllamaStreamErrorKind::ProviderError));
}
if let Some(message) = object.get("message") {
self.parse_message(message, events)?;
}
match object.get("done") {
Some(Value::Bool(true)) => {
let terminal = OllamaTerminal {
model: optional_string(object, "model")?,
done_reason: optional_string(object, "done_reason")?,
usage: OllamaUsage {
prompt_tokens: optional_u64(object, "prompt_eval_count")?,
completion_tokens: optional_u64(object, "eval_count")?,
},
durations: OllamaDurations {
total_ns: optional_u64(object, "total_duration")?,
load_ns: optional_u64(object, "load_duration")?,
prompt_eval_ns: optional_u64(object, "prompt_eval_duration")?,
eval_ns: optional_u64(object, "eval_duration")?,
},
};
events.push(OllamaStreamEvent::Completed(terminal));
self.terminal = true;
}
Some(Value::Bool(false)) | None => {}
Some(_) => return Err(self.error(OllamaStreamErrorKind::InvalidFrame)),
}
Ok(())
}
fn parse_message(
&mut self,
value: &Value,
events: &mut Vec<OllamaStreamEvent>,
) -> Result<(), OllamaStreamError> {
let message = value
.as_object()
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
if let Some(content) = message.get("content") {
let content = content
.as_str()
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
if !content.is_empty() {
events.push(OllamaStreamEvent::Content {
text: content.to_owned(),
});
}
}
if let Some(thinking) = message.get("thinking") {
let thinking = thinking
.as_str()
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
if !thinking.is_empty() {
events.push(OllamaStreamEvent::Thinking {
text: thinking.to_owned(),
});
}
}
if let Some(tool_calls) = message.get("tool_calls") {
let tool_calls = tool_calls
.as_array()
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
for tool_call in tool_calls {
let function = tool_call
.get("function")
.and_then(Value::as_object)
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
let name = function
.get("name")
.and_then(Value::as_str)
.filter(|name| !name.is_empty())
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
let arguments = function
.get("arguments")
.filter(|arguments| arguments.is_object())
.ok_or_else(|| self.error(OllamaStreamErrorKind::InvalidFrame))?;
let ordinal = self.next_tool_ordinal;
self.next_tool_ordinal = self.next_tool_ordinal.saturating_add(1);
events.push(OllamaStreamEvent::ToolCall {
id: synthesized_tool_id(&self.attempt_ref, ordinal),
ordinal,
name: name.to_owned(),
arguments: arguments.clone(),
});
}
}
Ok(())
}
fn error(&self, kind: OllamaStreamErrorKind) -> OllamaStreamError {
OllamaStreamError {
kind,
byte_offset: self.byte_offset,
}
}
}
fn optional_string(
object: &Map<String, Value>,
name: &str,
) -> Result<Option<String>, OllamaStreamError> {
match object.get(name) {
None | Some(Value::Null) => Ok(None),
Some(Value::String(value)) => Ok(Some(value.clone())),
Some(_) => Err(OllamaStreamError {
kind: OllamaStreamErrorKind::InvalidFrame,
byte_offset: 0,
}),
}
}
fn optional_u64(object: &Map<String, Value>, name: &str) -> Result<Option<u64>, OllamaStreamError> {
match object.get(name) {
None | Some(Value::Null) => Ok(None),
Some(value) => value.as_u64().map(Some).ok_or(OllamaStreamError {
kind: OllamaStreamErrorKind::InvalidFrame,
byte_offset: 0,
}),
}
}
fn synthesized_tool_id(attempt_ref: &str, ordinal: u64) -> String {
const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
let mut left = FNV_OFFSET;
let mut right = FNV_OFFSET ^ 0x9e37_79b9_7f4a_7c15;
for byte in attempt_ref
.bytes()
.chain(*b":")
.chain(ordinal.to_be_bytes())
{
left ^= u64::from(byte);
left = left.wrapping_mul(FNV_PRIME);
right ^= u64::from(byte.rotate_left(1));
right = right.wrapping_mul(FNV_PRIME);
}
format!("call_{left:016x}{right:016x}")
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReplayBoundary {
PublicContent,
VisibleToolCall,
ForwardedToolResult,
AmbiguousDispatch,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReplayDecision {
Allowed,
Denied(ReplayBoundary),
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct OllamaReplayGuard {
boundary: Option<ReplayBoundary>,
body_written: bool,
}
impl OllamaReplayGuard {
pub fn mark_body_written(&mut self) {
self.body_written = true;
}
pub fn observe(&mut self, event: &OllamaStreamEvent) {
match event {
OllamaStreamEvent::Content { text } if !text.is_empty() => {
self.cross(ReplayBoundary::PublicContent);
}
OllamaStreamEvent::ToolCall { .. } => self.cross(ReplayBoundary::VisibleToolCall),
OllamaStreamEvent::Content { .. }
| OllamaStreamEvent::Thinking { .. }
| OllamaStreamEvent::Completed(_) => {}
}
}
pub fn mark_forwarded_tool_result(&mut self) {
self.cross(ReplayBoundary::ForwardedToolResult);
}
pub fn mark_transport_outcome_unknown(&mut self) {
if self.body_written {
self.cross(ReplayBoundary::AmbiguousDispatch);
}
}
pub fn decision(&self) -> ReplayDecision {
self.boundary
.map_or(ReplayDecision::Allowed, ReplayDecision::Denied)
}
fn cross(&mut self, boundary: ReplayBoundary) {
if self.boundary.is_none() {
self.boundary = Some(boundary);
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OllamaFailureClass {
InvalidRequest,
Authentication,
PaymentRequired,
Authorization,
NotFound,
Timeout,
Unprocessable,
RateLimited,
UpstreamServer,
UnexpectedStatus,
MismatchedContentType,
RedirectRejected,
Cancelled,
TransportTimeout,
AmbiguousDispatch,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BoundedRetryAfter {
pub seconds: u32,
pub capped: bool,
}
impl BoundedRetryAfter {
pub fn parse(value: Option<&str>) -> Option<Self> {
let parsed = value?.trim().parse::<u64>().ok()?;
Some(Self {
seconds: parsed.min(u64::from(MAX_RETRY_AFTER_SECONDS)) as u32,
capped: parsed > u64::from(MAX_RETRY_AFTER_SECONDS),
})
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OllamaFailure {
pub class: OllamaFailureClass,
pub status: Option<u16>,
pub retry_after: Option<BoundedRetryAfter>,
}
pub fn normalize_http_response(
status: u16,
content_type: Option<&str>,
retry_after: Option<&str>,
redirected: bool,
) -> Result<(), OllamaFailure> {
if redirected || (300..400).contains(&status) {
return Err(failure(
OllamaFailureClass::RedirectRejected,
Some(status),
retry_after,
));
}
if status != 200 {
let class = match status {
400 => OllamaFailureClass::InvalidRequest,
401 => OllamaFailureClass::Authentication,
402 => OllamaFailureClass::PaymentRequired,
403 => OllamaFailureClass::Authorization,
404 => OllamaFailureClass::NotFound,
408 => OllamaFailureClass::Timeout,
422 => OllamaFailureClass::Unprocessable,
429 => OllamaFailureClass::RateLimited,
500..=599 => OllamaFailureClass::UpstreamServer,
_ => OllamaFailureClass::UnexpectedStatus,
};
return Err(failure(class, Some(status), retry_after));
}
let media_type = content_type
.and_then(|value| value.split(';').next())
.map(str::trim);
if !matches!(
media_type,
Some("application/x-ndjson" | "application/json")
) {
return Err(failure(
OllamaFailureClass::MismatchedContentType,
Some(status),
None,
));
}
Ok(())
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BodyWriteState {
NotWritten,
Written,
Unknown,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TransportFailureSignal {
Cancelled,
Timeout,
}
pub fn normalize_transport_failure(
signal: TransportFailureSignal,
body_write: BodyWriteState,
) -> OllamaFailure {
let class = match body_write {
BodyWriteState::Written | BodyWriteState::Unknown => OllamaFailureClass::AmbiguousDispatch,
BodyWriteState::NotWritten => match signal {
TransportFailureSignal::Cancelled => OllamaFailureClass::Cancelled,
TransportFailureSignal::Timeout => OllamaFailureClass::TransportTimeout,
},
};
failure(class, None, None)
}
fn failure(
class: OllamaFailureClass,
status: Option<u16>,
retry_after: Option<&str>,
) -> OllamaFailure {
OllamaFailure {
class,
status,
retry_after: BoundedRetryAfter::parse(retry_after),
}
}
#[cfg(test)]
mod tests {
use super::*;
use router_protocol::{
BillingAuthorityClass, CapabilityRequirements, CredentialSourceKind, EndpointIdentity,
ExecutionLocation, FunctionTool, HarnessInstanceId, M1_WIRE_VERSION, ModelNamespace,
PhasePlacement, ProviderService, SessionId, TurnId,
};
fn route(phase: PhaseKind, transport: TransportKind, model: &str) -> RouteIdentity {
let cloud = phase == PhaseKind::OllamaCloud;
RouteIdentity {
phase,
placement: PhasePlacement::Builtin,
provider: if cloud {
ProviderService::OllamaCloud
} else {
ProviderService::OllamaLocal
},
transport,
endpoint: EndpointIdentity {
origin: if transport == TransportKind::DirectHttps {
"https://ollama.com".to_owned()
} else {
"http://127.0.0.1:11434".to_owned()
},
path: "/api/chat".to_owned(),
},
credential_source: if transport == TransportKind::DirectHttps {
CredentialSourceKind::SecretRef
} else if cloud {
CredentialSourceKind::OllamaDaemonSession
} else {
CredentialSourceKind::None
},
credential_ref: (transport == TransportKind::DirectHttps)
.then(|| "keychain:ollama-cloud-canary".to_owned()),
credential_refs: if transport == TransportKind::DirectHttps {
vec!["keychain:ollama-cloud-canary".to_owned()]
} else {
Vec::new()
},
catalog_source: if cloud {
CatalogSourceKind::OllamaCloudCatalog
} else {
CatalogSourceKind::OllamaLocalInventory
},
catalog_ref: if cloud {
"catalog:ollama-cloud:verified".to_owned()
} else {
"inventory:local:approved".to_owned()
},
billing_authority: if cloud {
BillingAuthorityClass::OllamaCloud
} else {
BillingAuthorityClass::LocalCompute
},
model_namespace: if cloud {
ModelNamespace::OllamaCloud
} else {
ModelNamespace::OllamaLocal
},
model_id: model.to_owned(),
execution_location: if cloud {
ExecutionLocation::RemoteCloud
} else {
ExecutionLocation::LocalMachine
},
account_ref: if cloud {
"account:ollama-cloud".to_owned()
} else {
"account:local".to_owned()
},
capacity_domain_ref: if cloud {
"capacity:ollama-cloud".to_owned()
} else {
"capacity:local".to_owned()
},
local_execution_proof: !cloud,
approved_local_inventory: if cloud {
Vec::new()
} else {
vec![model.to_owned()]
},
reasoning_modes: if cloud || model.starts_with("qwen") {
vec!["reasoning".to_owned()]
} else {
Vec::new()
},
}
}
fn request(prompt: &str) -> CanonicalRequest {
CanonicalRequest {
wire_version: M1_WIRE_VERSION.to_owned(),
model: "router-auto".to_owned(),
stream: true,
harness_instance: HarnessInstanceId::parse("fixture-cli").expect("fixture"),
session_id: SessionId::parse("session-m31").expect("fixture"),
turn_id: TurnId::parse("turn-m31").expect("fixture"),
input: vec![InputItem::Message {
role: MessageRole::User,
content: vec![ContentBlock::Text {
text: prompt.to_owned(),
}],
}],
tools: vec![FunctionTool {
namespace: Some("weather".to_owned()),
name: "lookup".to_owned(),
description: "lookup weather".to_owned(),
parameters: json!({"type":"object"}),
strict: true,
}],
tool_choice: ToolChoice::Auto,
structured_output: None,
reasoning: ReasoningIntent::PublicSummary,
max_output_tokens: Some(64),
max_spend_micro_usd: None,
capabilities: CapabilityRequirements::default(),
}
}
#[test]
fn exact_relay_and_direct_request_specs_are_io_free_and_redacted() {
let relay = build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::OllamaSignedInRelay,
"glm-5.2:cloud",
),
&request("prompt-canary"),
)
.expect("relay request");
assert_eq!(relay.method, HttpMethod::Post);
assert_eq!(relay.origin, "http://127.0.0.1:11434");
assert_eq!(relay.path, "/api/chat");
assert_eq!(relay.authentication, OllamaAuthentication::None);
assert!(!relay.network.allow_proxy);
assert!(!relay.network.allow_dns);
assert!(!relay.network.follow_redirects);
assert_eq!(
relay.body(),
&json!({
"model":"glm-5.2:cloud",
"messages":[{"role":"user","content":"prompt-canary"}],
"stream":true,
"think":true,
"tools":[{"type":"function","function":{
"name":"weather__lookup",
"description":"lookup weather",
"parameters":{"type":"object"}
}}],
"options":{"num_predict":64}
})
);
let serialized = relay.body_bytes().expect("JSON body");
let wire = String::from_utf8(serialized).expect("JSON UTF-8");
for forbidden in ["/api/tags", "/api/pull", "/api/create", "/api/push"] {
assert!(!wire.contains(forbidden));
}
let direct = build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::DirectHttps,
"glm-5.2",
),
&request("prompt-canary"),
)
.expect("direct request");
assert_eq!(direct.origin, "https://ollama.com");
assert_eq!(direct.path, "/api/chat");
assert!(direct.network.require_tls);
assert!(!direct.network.follow_redirects);
assert_eq!(direct.body()["model"], "glm-5.2");
let debug = format!("{direct:?}");
assert!(!debug.contains("keychain:ollama-cloud-canary"));
assert!(!debug.contains("prompt-canary"));
let hyphen_relay = build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::OllamaSignedInRelay,
"qwen3-coder:480b-cloud",
),
&request("safe"),
)
.expect("canonical hyphen-cloud relay request");
assert_eq!(hyphen_relay.body()["model"], "qwen3-coder:480b-cloud");
}
#[test]
fn cloud_model_names_are_transport_specific_and_local_rejects_cloud_alias() {
let error = build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::DirectHttps,
"unverified-model:cloud",
),
&request("safe"),
)
.expect_err("relay-only suffix on direct API");
assert_eq!(error.kind, OllamaRequestErrorKind::InvalidRouteIdentity);
let error = build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::DirectHttps,
"qwen3-coder:480b-cloud",
),
&request("safe"),
)
.expect_err("hyphen relay-only suffix on direct API");
assert_eq!(error.kind, OllamaRequestErrorKind::InvalidRouteIdentity);
let error = build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::OllamaSignedInRelay,
"unverified-model",
),
&request("safe"),
)
.expect_err("missing relay cloud suffix");
assert_eq!(error.kind, OllamaRequestErrorKind::InvalidRouteIdentity);
let local = route(
PhaseKind::OllamaLocal,
TransportKind::OllamaLocalDaemon,
"glm-5.2:cloud",
);
assert_eq!(
build_request(&local, &request("safe"))
.expect_err("local cloud alias")
.kind,
OllamaRequestErrorKind::InvalidRouteIdentity
);
let local = route(
PhaseKind::OllamaLocal,
TransportKind::OllamaLocalDaemon,
"qwen3-coder:480b-cloud",
);
assert_eq!(
build_request(&local, &request("safe"))
.expect_err("local hyphen cloud alias")
.kind,
OllamaRequestErrorKind::InvalidRouteIdentity
);
for invalid in [
"model-cloud-cloud",
"model:cloud-cloud",
"model-cloud:cloud",
"model:cloud:cloud",
] {
assert_eq!(
build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::OllamaSignedInRelay,
invalid,
),
&request("safe"),
)
.expect_err("double or mixed cloud suffix")
.kind,
OllamaRequestErrorKind::InvalidRouteIdentity
);
}
let local = build_request(
&route(
PhaseKind::OllamaLocal,
TransportKind::OllamaLocalDaemon,
"qwen3:8b",
),
&request("safe"),
)
.expect("approved local model");
assert_eq!(local.origin, "http://127.0.0.1:11434");
assert_eq!(local.path, "/api/chat");
assert_eq!(local.authentication, OllamaAuthentication::None);
assert!(!local.network.allow_proxy);
assert!(!local.network.allow_dns);
assert!(!local.network.follow_redirects);
let mut direct_without_secret = route(
PhaseKind::OllamaCloud,
TransportKind::DirectHttps,
"glm-5.2",
);
direct_without_secret.credential_ref = None;
assert_eq!(
build_request(&direct_without_secret, &request("safe"))
.expect_err("direct secret reference")
.kind,
OllamaRequestErrorKind::InvalidRouteIdentity
);
}
#[test]
fn think_is_limited_to_models_that_declare_reasoning_support() {
let mistral = build_request(
&route(
PhaseKind::OllamaLocal,
TransportKind::OllamaLocalDaemon,
"mistral:latest",
),
&request("prompt-canary"),
)
.expect("approved local model");
assert_eq!(mistral.body()["think"], false);
let qwen = build_request(
&route(
PhaseKind::OllamaLocal,
TransportKind::OllamaLocalDaemon,
"qwen3:4b",
),
&request("prompt-canary"),
)
.expect("approved local model");
assert_eq!(qwen.body()["think"], true);
let cloud = build_request(
&route(
PhaseKind::OllamaCloud,
TransportKind::OllamaSignedInRelay,
"glm-5.2:cloud",
),
&request("prompt-canary"),
)
.expect("approved cloud relay");
assert_eq!(cloud.body()["think"], true);
}
#[test]
fn ndjson_is_independent_of_every_byte_boundary_and_preserves_terminal_facts() {
let transcript = concat!(
"\n",
"{\"model\":\"glm-5.2:cloud\",\"message\":{\"thinking\":\"plan\",\"content\":\"hé\"},\"done\":false}\r\n",
"{\"message\":{\"content\":\"\",\"tool_calls\":[{\"function\":{\"name\":\"weather__lookup\",\"arguments\":{\"city\":\"Paris\"}}}]},\"done\":false}\n",
"{\"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}"
);
let attempt = AttemptId::parse("attempt-fragmented").expect("attempt");
let mut parser = OllamaNdjsonParser::new(&attempt);
let mut events = Vec::new();
for byte in transcript.as_bytes() {
events.extend(parser.feed(std::slice::from_ref(byte)).expect("fragment"));
}
events.extend(parser.finish().expect("terminal without newline"));
assert_eq!(events.len(), 4);
assert_eq!(
events[0],
OllamaStreamEvent::Content {
text: "hé".to_owned()
}
);
assert_eq!(
events[1],
OllamaStreamEvent::Thinking {
text: "plan".to_owned()
}
);
let OllamaStreamEvent::ToolCall { id, ordinal, .. } = &events[2] else {
panic!("tool event");
};
assert!(id.starts_with("call_"));
assert_eq!(*ordinal, 0);
let OllamaStreamEvent::Completed(terminal) = &events[3] else {
panic!("terminal event");
};
assert_eq!(terminal.model.as_deref(), Some("glm-5.2:cloud"));
assert_eq!(terminal.usage.prompt_tokens, Some(11));
assert_eq!(terminal.usage.completion_tokens, Some(7));
assert_eq!(terminal.durations.total_ns, Some(101));
let mut second = OllamaNdjsonParser::new(&attempt);
let second_events = second.feed(transcript.as_bytes()).expect("whole chunk");
let mut all = second_events;
all.extend(second.finish().expect("terminal"));
assert_eq!(events, all);
}
#[test]
fn ndjson_failures_are_typed_bounded_and_redacted() {
let attempt = AttemptId::parse("attempt-errors").expect("attempt");
let cases: &[(&[u8], OllamaStreamErrorKind)] = &[
(b"{not-json}\n", OllamaStreamErrorKind::InvalidJson),
(b"\xff\n", OllamaStreamErrorKind::InvalidUtf8),
(
b"{\"error\":\"provider-secret-canary\"}\n",
OllamaStreamErrorKind::ProviderError,
),
];
for (input, expected) in cases {
let mut parser = OllamaNdjsonParser::new(&attempt);
let error = parser.feed(input).expect_err("invalid stream");
assert_eq!(error.kind, *expected);
assert!(!format!("{error:?}").contains("provider-secret-canary"));
}
let mut oversized = OllamaNdjsonParser::with_max_line_bytes(&attempt, 8);
assert_eq!(
oversized.feed(b"123456789").expect_err("bounded").kind,
OllamaStreamErrorKind::LineTooLarge
);
let mut incomplete = OllamaNdjsonParser::new(&attempt);
incomplete
.feed(b"{\"message\":{\"content\":\"partial\"},\"done\":false}\n")
.expect("partial frame");
assert_eq!(
incomplete.finish().expect_err("terminal required").kind,
OllamaStreamErrorKind::MissingTerminal
);
}
#[test]
fn replay_crosses_each_public_or_ambiguous_boundary_once() {
let mut thinking = OllamaReplayGuard::default();
thinking.observe(&OllamaStreamEvent::Thinking {
text: "private".to_owned(),
});
assert_eq!(thinking.decision(), ReplayDecision::Allowed);
thinking.observe(&OllamaStreamEvent::Content {
text: "public".to_owned(),
});
assert_eq!(
thinking.decision(),
ReplayDecision::Denied(ReplayBoundary::PublicContent)
);
let mut tool = OllamaReplayGuard::default();
tool.observe(&OllamaStreamEvent::ToolCall {
id: "call_fixture".to_owned(),
ordinal: 0,
name: "lookup".to_owned(),
arguments: json!({}),
});
assert_eq!(
tool.decision(),
ReplayDecision::Denied(ReplayBoundary::VisibleToolCall)
);
let mut result = OllamaReplayGuard::default();
result.mark_forwarded_tool_result();
assert_eq!(
result.decision(),
ReplayDecision::Denied(ReplayBoundary::ForwardedToolResult)
);
let mut ambiguous = OllamaReplayGuard::default();
ambiguous.mark_transport_outcome_unknown();
assert_eq!(ambiguous.decision(), ReplayDecision::Allowed);
ambiguous.mark_body_written();
ambiguous.mark_transport_outcome_unknown();
assert_eq!(
ambiguous.decision(),
ReplayDecision::Denied(ReplayBoundary::AmbiguousDispatch)
);
}
#[test]
fn failures_cover_status_content_type_redirect_timeout_and_ambiguity() {
let matrix = [
(400, OllamaFailureClass::InvalidRequest),
(401, OllamaFailureClass::Authentication),
(402, OllamaFailureClass::PaymentRequired),
(403, OllamaFailureClass::Authorization),
(404, OllamaFailureClass::NotFound),
(408, OllamaFailureClass::Timeout),
(422, OllamaFailureClass::Unprocessable),
(429, OllamaFailureClass::RateLimited),
(503, OllamaFailureClass::UpstreamServer),
];
for (status, class) in matrix {
let error = normalize_http_response(status, None, Some("999999"), false)
.expect_err("non-success");
assert_eq!(error.class, class);
assert_eq!(
error.retry_after,
Some(BoundedRetryAfter {
seconds: MAX_RETRY_AFTER_SECONDS,
capped: true,
})
);
}
assert!(
normalize_http_response(
200,
Some("application/x-ndjson; charset=utf-8"),
None,
false
)
.is_ok()
);
assert_eq!(
normalize_http_response(200, Some("text/html"), None, false)
.expect_err("content type")
.class,
OllamaFailureClass::MismatchedContentType
);
assert_eq!(
normalize_http_response(302, None, None, true)
.expect_err("redirect")
.class,
OllamaFailureClass::RedirectRejected
);
assert_eq!(
normalize_transport_failure(
TransportFailureSignal::Cancelled,
BodyWriteState::NotWritten,
)
.class,
OllamaFailureClass::Cancelled
);
assert_eq!(
normalize_transport_failure(TransportFailureSignal::Timeout, BodyWriteState::Written,)
.class,
OllamaFailureClass::AmbiguousDispatch
);
}
}