| 1 | //! Streaming client for OpenAI-compatible `/chat/completions` endpoints (Onde Cloud by default, |
| 2 | //! or any other with the `OPENAI_*` variables): text and reasoning deltas, tool-call |
| 3 | //! accumulation, token usage, and `GET /models`. |
| 4 | //! |
| 5 | //! Superseded by `onde-backend` once it exists (Onde Agent Platform §5.2); the agent depends |
| 6 | //! only on this module's public types so that swap stays local. |
| 7 | |
| 8 | use anyhow::{Context, Result, bail}; |
| 9 | use futures::StreamExt; |
| 10 | use serde::Deserialize; |
| 11 | use serde_json::{Value, json}; |
| 12 | |
| 13 | /// Onde Cloud's OpenAI-compatible endpoint. |
| 14 | pub const DEFAULT_BASE_URL: &str = "https://cloud.ondeinference.com/v1"; |
| 15 | /// The model used when neither the profile nor the environment names one. |
| 16 | pub const DEFAULT_MODEL: &str = "onde-kkk"; |
| 17 | /// The credential every Onde agent reads: `app-id:app-secret`. |
| 18 | pub const API_KEY_VAR: &str = "ONDE_API_KEY"; |
| 19 | /// Base URL of a generic OpenAI API compatible endpoint when `OPENAI_BASE_URL` is unset. |
| 20 | pub const OPENAI_DEFAULT_BASE_URL: &str = "https://api.openai.com/v1"; |
| 21 | /// Model for a generic endpoint when neither `<PREFIX>_MODEL` nor `OPENAI_MODEL` is set. |
| 22 | pub const OPENAI_DEFAULT_MODEL: &str = "gpt-4o-mini"; |
| 23 | |
| 24 | /// The on-device model when `<PREFIX>_MODEL` is unset: Qwen3 1.7B on phones, where memory is |
| 25 | /// tight, Qwen3 4B Instruct 2507 elsewhere. Both are in onde's tool-calling catalog. |
| 26 | #[cfg(any(target_os = "ios", target_os = "android"))] |
| 27 | pub const LOCAL_DEFAULT_MODEL: &str = "bartowski/Qwen_Qwen3-1.7B-GGUF"; |
| 28 | /// The on-device model when `<PREFIX>_MODEL` is unset: Qwen3 1.7B on phones, where memory is |
| 29 | /// tight, Qwen3 4B Instruct 2507 elsewhere. Both are in onde's tool-calling catalog. |
| 30 | #[cfg(not(any(target_os = "ios", target_os = "android")))] |
| 31 | pub const LOCAL_DEFAULT_MODEL: &str = "bartowski/Qwen_Qwen3-4B-Instruct-2507-GGUF"; |
| 32 | |
| 33 | /// Where completions come from. Both speak the OpenAI chat completions API. |
| 34 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 35 | pub enum Provider { |
| 36 | /// Onde Cloud: `ONDE_API_KEY` (`app-id:app-secret`), `<PREFIX>_BASE_URL` or |
| 37 | /// `ONDE_BASE_URL` for a local or staging deployment. |
| 38 | Onde, |
| 39 | /// Any OpenAI API compatible endpoint: `OPENAI_BASE_URL`, `OPENAI_API_KEY`, `OPENAI_MODEL`. |
| 40 | OpenAi, |
| 41 | /// On-device inference with onde: no endpoint and no key. `<PREFIX>_MODEL` names a model |
| 42 | /// from onde's tool-calling catalog. Needs the `local` feature; without it every turn fails |
| 43 | /// with an error saying so. |
| 44 | Local, |
| 45 | } |
| 46 | |
| 47 | impl Provider { |
| 48 | /// `onde` or `openai`, as written to `<PREFIX>_PROVIDER`. |
| 49 | pub fn name(self) -> &'static str { |
| 50 | match self { |
| 51 | Self::Onde => "onde", |
| 52 | Self::OpenAi => "openai", |
| 53 | Self::Local => "local", |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | pub fn parse(name: &str) -> Option<Self> { |
| 58 | match name.to_ascii_lowercase().as_str() { |
| 59 | "onde" | "onde-cloud" | "ondeinference" => Some(Self::Onde), |
| 60 | "openai" => Some(Self::OpenAi), |
| 61 | "local" | "on-device" | "ondevice" => Some(Self::Local), |
| 62 | _ => None, |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | /// The variable holding this provider's key. Empty for [`Provider::Local`], which has none. |
| 67 | pub fn key_var(self) -> &'static str { |
| 68 | match self { |
| 69 | Self::Onde => API_KEY_VAR, |
| 70 | Self::OpenAi => "OPENAI_API_KEY", |
| 71 | Self::Local => "", |
| 72 | } |
| 73 | } |
| 74 | |
| 75 | /// Name shown to people, e.g. in setup and errors. |
| 76 | pub fn display_name(self) -> &'static str { |
| 77 | match self { |
| 78 | Self::Onde => "Onde Inference", |
| 79 | Self::OpenAi => "the OpenAI API compatible endpoint", |
| 80 | Self::Local => "on-device inference", |
| 81 | } |
| 82 | } |
| 83 | } |
| 84 | |
| 85 | /// Where an agent's settings come from: `<PREFIX>_BASE_URL`, `<PREFIX>_MODEL` and |
| 86 | /// `<PREFIX>_DATA_DIR` in the environment, then `<config dir>/<dir_name>/env`. |
| 87 | #[derive(Debug, Clone, Copy, PartialEq, Eq)] |
| 88 | pub struct LlmEnv { |
| 89 | /// Environment prefix, e.g. `SPLITFIRE`. |
| 90 | pub prefix: &'static str, |
| 91 | /// Directory name under the platform config dir, e.g. `splitfire-agent`. |
| 92 | pub dir_name: &'static str, |
| 93 | /// Model when `<PREFIX>_MODEL` is unset. |
| 94 | pub default_model: &'static str, |
| 95 | } |
| 96 | |
| 97 | impl LlmEnv { |
| 98 | pub fn var(&self, suffix: &str) -> String { |
| 99 | format!("{}_{suffix}", self.prefix) |
| 100 | } |
| 101 | |
| 102 | /// Directory for the stored `env` file: `~/.config/<dir_name>` on Linux, |
| 103 | /// `~/Library/Application Support/<dir_name>` on macOS, `%APPDATA%\\<dir_name>` on |
| 104 | /// Windows. `<PREFIX>_DATA_DIR` overrides it (tests). `None` without a home directory. |
| 105 | pub fn config_dir(&self) -> Option<std::path::PathBuf> { |
| 106 | if let Some(dir) = std::env::var_os(self.var("DATA_DIR")).filter(|v| !v.is_empty()) { |
| 107 | return Some(dir.into()); |
| 108 | } |
| 109 | directories::BaseDirs::new().map(|dirs| { |
| 110 | #[cfg(target_os = "linux")] |
| 111 | let base = dirs.config_dir(); |
| 112 | #[cfg(not(target_os = "linux"))] |
| 113 | let base = dirs.data_dir(); |
| 114 | base.join(self.dir_name) |
| 115 | }) |
| 116 | } |
| 117 | |
| 118 | /// The stored `env` file `--setup` writes and `logout` removes. |
| 119 | pub fn config_file(&self) -> Option<std::path::PathBuf> { |
| 120 | self.config_dir().map(|d| d.join("env")) |
| 121 | } |
| 122 | |
| 123 | /// Where the stored `env` file is looked for, most preferred first: [`Self::config_file`], |
| 124 | /// then `~/.config/<dir_name>/env`, where agents on macOS kept it before they moved to the |
| 125 | /// platform directory. Only the first is used when `<PREFIX>_DATA_DIR` is set. |
| 126 | pub fn config_file_candidates(&self) -> Vec<std::path::PathBuf> { |
| 127 | let mut paths: Vec<std::path::PathBuf> = self.config_file().into_iter().collect(); |
| 128 | let overridden = std::env::var_os(self.var("DATA_DIR")).is_some_and(|v| !v.is_empty()); |
| 129 | #[cfg(not(target_os = "windows"))] |
| 130 | if !overridden && let Some(dirs) = directories::BaseDirs::new() { |
| 131 | let legacy = dirs |
| 132 | .home_dir() |
| 133 | .join(".config") |
| 134 | .join(self.dir_name) |
| 135 | .join("env"); |
| 136 | if !paths.contains(&legacy) { |
| 137 | paths.push(legacy); |
| 138 | } |
| 139 | } |
| 140 | #[cfg(target_os = "windows")] |
| 141 | let _ = overridden; |
| 142 | paths |
| 143 | } |
| 144 | |
| 145 | /// Where durable state (sessions, audio scratch) lives. Same directory as the `env` file. |
| 146 | pub fn data_dir(&self) -> std::path::PathBuf { |
| 147 | self.config_dir() |
| 148 | .unwrap_or_else(|| std::env::temp_dir().join(self.dir_name)) |
| 149 | } |
| 150 | |
| 151 | fn load_file_vars(&self) -> std::collections::HashMap<String, String> { |
| 152 | let mut vars = std::collections::HashMap::new(); |
| 153 | let content = self |
| 154 | .config_file_candidates() |
| 155 | .iter() |
| 156 | .find_map(|p| std::fs::read_to_string(p).ok()); |
| 157 | let Some(content) = content else { |
| 158 | return vars; |
| 159 | }; |
| 160 | for line in content.lines() { |
| 161 | let line = line.trim(); |
| 162 | if line.is_empty() || line.starts_with('#') { |
| 163 | continue; |
| 164 | } |
| 165 | let line = line.strip_prefix("export ").unwrap_or(line).trim(); |
| 166 | if let Some((k, v)) = line.split_once('=') { |
| 167 | let mut v = v.trim(); |
| 168 | if v.len() >= 2 |
| 169 | && ((v.starts_with('"') && v.ends_with('"')) |
| 170 | || (v.starts_with('\'') && v.ends_with('\''))) |
| 171 | { |
| 172 | v = &v[1..v.len() - 1]; |
| 173 | } |
| 174 | vars.insert(k.trim().to_string(), v.to_string()); |
| 175 | } |
| 176 | } |
| 177 | vars |
| 178 | } |
| 179 | } |
| 180 | |
| 181 | #[derive(Debug, Clone, PartialEq, Eq)] |
| 182 | pub struct LlmConfig { |
| 183 | pub provider: Provider, |
| 184 | pub base_url: String, |
| 185 | /// Sent as the bearer token. For Onde Cloud, `app-id:app-secret`. |
| 186 | pub api_key: Option<String>, |
| 187 | pub model: String, |
| 188 | } |
| 189 | |
| 190 | impl LlmConfig { |
| 191 | /// Settings from the environment, falling back to the stored `env` file. |
| 192 | /// |
| 193 | /// The provider is `<PREFIX>_PROVIDER` (`onde` or `openai`) when set, else Onde Cloud |
| 194 | /// whenever `ONDE_API_KEY` is set, else a generic endpoint when `OPENAI_API_KEY` is, else |
| 195 | /// Onde Cloud. Onde Cloud reads only `ONDE_API_KEY`, and its own URL unless |
| 196 | /// `<PREFIX>_BASE_URL` or `ONDE_BASE_URL` names another deployment, so stray `OPENAI_*` |
| 197 | /// variables can't redirect it or send it the wrong key. A generic endpoint reads |
| 198 | /// `OPENAI_BASE_URL`, `OPENAI_API_KEY` and `OPENAI_MODEL`. `<PREFIX>_MODEL` picks the |
| 199 | /// model for either. |
| 200 | /// |
| 201 | /// Empty values are ignored: GUI launchers (e.g. Zed via launchd) often export |
| 202 | /// variables set to "", which must not mask the config file. |
| 203 | pub fn from_env(env: &LlmEnv) -> Self { |
| 204 | let file_vars = env.load_file_vars(); |
| 205 | Self::from_lookup(env, |name| { |
| 206 | std::env::var(name) |
| 207 | .ok() |
| 208 | .filter(|v| !v.is_empty()) |
| 209 | .or_else(|| file_vars.get(name).cloned()) |
| 210 | }) |
| 211 | } |
| 212 | |
| 213 | /// Build a config from an arbitrary variable lookup (tests, setup). |
| 214 | pub fn from_lookup(env: &LlmEnv, lookup: impl Fn(&str) -> Option<String>) -> Self { |
| 215 | let get = |name: &str| lookup(name).filter(|v| !v.is_empty()); |
| 216 | let provider = match get(&env.var("PROVIDER")) { |
| 217 | Some(name) => Provider::parse(&name).unwrap_or_else(|| { |
| 218 | tracing::warn!("unknown {} {name:?}; using onde", env.var("PROVIDER")); |
| 219 | Provider::Onde |
| 220 | }), |
| 221 | None if get(API_KEY_VAR).is_some() => Provider::Onde, |
| 222 | None if get("OPENAI_API_KEY").is_some() => Provider::OpenAi, |
| 223 | None => Provider::Onde, |
| 224 | }; |
| 225 | let (base_url, model) = match provider { |
| 226 | Provider::Onde => ( |
| 227 | get(&env.var("BASE_URL")) |
| 228 | .or_else(|| get("ONDE_BASE_URL")) |
| 229 | .unwrap_or_else(|| DEFAULT_BASE_URL.into()), |
| 230 | get(&env.var("MODEL")).unwrap_or_else(|| env.default_model.into()), |
| 231 | ), |
| 232 | Provider::OpenAi => ( |
| 233 | get("OPENAI_BASE_URL").unwrap_or_else(|| OPENAI_DEFAULT_BASE_URL.into()), |
| 234 | get(&env.var("MODEL")) |
| 235 | .or_else(|| get("OPENAI_MODEL")) |
| 236 | .unwrap_or_else(|| OPENAI_DEFAULT_MODEL.into()), |
| 237 | ), |
| 238 | Provider::Local => ( |
| 239 | String::new(), |
| 240 | get(&env.var("MODEL")).unwrap_or_else(|| LOCAL_DEFAULT_MODEL.into()), |
| 241 | ), |
| 242 | }; |
| 243 | Self { |
| 244 | provider, |
| 245 | base_url: base_url.trim_end_matches('/').to_string(), |
| 246 | // On-device inference has no key, and `get("")` must not find one in an env file. |
| 247 | api_key: match provider { |
| 248 | Provider::Local => None, |
| 249 | _ => get(provider.key_var()), |
| 250 | }, |
| 251 | model, |
| 252 | } |
| 253 | } |
| 254 | } |
| 255 | |
| 256 | #[cfg(test)] |
| 257 | mod tests { |
| 258 | use super::*; |
| 259 | |
| 260 | const ENV: LlmEnv = LlmEnv { |
| 261 | prefix: "SPLITFIRE", |
| 262 | dir_name: "splitfire-agent", |
| 263 | default_model: DEFAULT_MODEL, |
| 264 | }; |
| 265 | |
| 266 | fn config(vars: &[(&str, &str)]) -> LlmConfig { |
| 267 | LlmConfig::from_lookup(&ENV, |name| { |
| 268 | vars.iter() |
| 269 | .find(|(k, _)| *k == name) |
| 270 | .map(|(_, v)| v.to_string()) |
| 271 | }) |
| 272 | } |
| 273 | |
| 274 | #[test] |
| 275 | fn defaults_point_at_onde_cloud() { |
| 276 | let c = config(&[("ONDE_API_KEY", "app:secret")]); |
| 277 | assert_eq!(c.base_url, "https://cloud.ondeinference.com/v1"); |
| 278 | assert_eq!(c.api_key.as_deref(), Some("app:secret")); |
| 279 | assert_eq!(c.model, "onde-kkk"); |
| 280 | assert_eq!(config(&[]).api_key, None); |
| 281 | assert_eq!(config(&[("ONDE_API_KEY", "")]).api_key, None); |
| 282 | } |
| 283 | |
| 284 | #[test] |
| 285 | fn local_needs_no_key() { |
| 286 | let c = config(&[ |
| 287 | ("SPLITFIRE_PROVIDER", "local"), |
| 288 | ("ONDE_API_KEY", "app:secret"), |
| 289 | ]); |
| 290 | assert_eq!(c.provider, Provider::Local); |
| 291 | assert_eq!(c.model, LOCAL_DEFAULT_MODEL); |
| 292 | // A stored Onde key is not this provider's. |
| 293 | assert_eq!(c.api_key, None); |
| 294 | assert!(LlmClient::new(c).has_api_key()); |
| 295 | assert_eq!(Provider::parse("on-device"), Some(Provider::Local)); |
| 296 | let picked = config(&[ |
| 297 | ("SPLITFIRE_PROVIDER", "local"), |
| 298 | ("SPLITFIRE_MODEL", "bartowski/Qwen_Qwen3-8B-GGUF"), |
| 299 | ]); |
| 300 | assert_eq!(picked.model, "bartowski/Qwen_Qwen3-8B-GGUF"); |
| 301 | } |
| 302 | |
| 303 | #[test] |
| 304 | fn base_url_and_model_overrides_apply() { |
| 305 | let c = config(&[ |
| 306 | ("ONDE_API_KEY", "app:secret"), |
| 307 | ("SPLITFIRE_MODEL", "onde-prism"), |
| 308 | ("SPLITFIRE_BASE_URL", "http://x/v1/"), |
| 309 | ]); |
| 310 | assert_eq!( |
| 311 | (c.base_url.as_str(), c.model.as_str()), |
| 312 | ("http://x/v1", "onde-prism") |
| 313 | ); |
| 314 | } |
| 315 | |
| 316 | #[test] |
| 317 | fn onde_base_url_points_onde_at_another_deployment() { |
| 318 | let c = config(&[ |
| 319 | ("ONDE_API_KEY", "app:secret"), |
| 320 | ("ONDE_BASE_URL", "http://localhost:8090/v1/"), |
| 321 | ]); |
| 322 | assert_eq!(c.base_url, "http://localhost:8090/v1"); |
| 323 | let both = config(&[ |
| 324 | ("ONDE_BASE_URL", "http://a/v1"), |
| 325 | ("SPLITFIRE_BASE_URL", "http://b/v1"), |
| 326 | ]); |
| 327 | assert_eq!(both.base_url, "http://b/v1"); |
| 328 | } |
| 329 | |
| 330 | #[test] |
| 331 | fn openai_key_selects_a_generic_endpoint() { |
| 332 | let c = config(&[ |
| 333 | ("OPENAI_API_KEY", "sk"), |
| 334 | ("OPENAI_MODEL", "my-model"), |
| 335 | ("OPENAI_BASE_URL", "http://x/v1/"), |
| 336 | ("ONDE_BASE_URL", "http://onde/v1"), |
| 337 | ]); |
| 338 | assert_eq!( |
| 339 | ( |
| 340 | c.provider, |
| 341 | c.base_url.as_str(), |
| 342 | c.api_key.as_deref(), |
| 343 | c.model.as_str() |
| 344 | ), |
| 345 | (Provider::OpenAi, "http://x/v1", Some("sk"), "my-model") |
| 346 | ); |
| 347 | let plain = config(&[("OPENAI_API_KEY", "sk")]); |
| 348 | assert_eq!( |
| 349 | (plain.base_url.as_str(), plain.model.as_str()), |
| 350 | ("https://api.openai.com/v1", "gpt-4o-mini") |
| 351 | ); |
| 352 | assert_eq!(config(&[]).provider, Provider::Onde); |
| 353 | } |
| 354 | |
| 355 | #[test] |
| 356 | fn onde_wins_over_openai_variables() { |
| 357 | let c = config(&[ |
| 358 | ("ONDE_API_KEY", "app:secret"), |
| 359 | ("OPENAI_API_KEY", "sk-other"), |
| 360 | ("OPENAI_BASE_URL", "http://elsewhere/v1"), |
| 361 | ("OPENAI_MODEL", "gpt-x"), |
| 362 | ]); |
| 363 | assert_eq!(c.provider, Provider::Onde); |
| 364 | assert_eq!(c.api_key.as_deref(), Some("app:secret")); |
| 365 | assert_eq!(c.base_url, "https://cloud.ondeinference.com/v1"); |
| 366 | assert_eq!(c.model, "onde-kkk"); |
| 367 | } |
| 368 | |
| 369 | #[test] |
| 370 | fn explicit_provider_wins_over_detected_keys() { |
| 371 | let c = config(&[ |
| 372 | ("ONDE_API_KEY", "app:secret"), |
| 373 | ("OPENAI_API_KEY", "sk"), |
| 374 | ("SPLITFIRE_PROVIDER", "openai"), |
| 375 | ("SPLITFIRE_MODEL", "m"), |
| 376 | ]); |
| 377 | assert_eq!( |
| 378 | (c.provider, c.api_key.as_deref(), c.model.as_str()), |
| 379 | (Provider::OpenAi, Some("sk"), "m") |
| 380 | ); |
| 381 | let onde = config(&[("SPLITFIRE_PROVIDER", "onde"), ("OPENAI_API_KEY", "sk")]); |
| 382 | assert_eq!((onde.provider, onde.api_key), (Provider::Onde, None)); |
| 383 | } |
| 384 | |
| 385 | /// One test touches the process environment, so it can't race another. |
| 386 | #[test] |
| 387 | fn key_from_env_then_file_ignoring_empty_env() { |
| 388 | let dir = std::env::temp_dir().join(format!("sf-llm-{}", uuid::Uuid::new_v4())); |
| 389 | std::fs::create_dir_all(&dir).unwrap(); |
| 390 | std::fs::write( |
| 391 | dir.join("env"), |
| 392 | "# comment\nexport ONDE_API_KEY=\"file:key\"\nSPLITFIRE_MODEL=m1\n", |
| 393 | ) |
| 394 | .unwrap(); |
| 395 | // SAFETY: only this test mutates these variables. |
| 396 | unsafe { |
| 397 | std::env::set_var("SPLITFIRE_DATA_DIR", &dir); |
| 398 | std::env::set_var("ONDE_API_KEY", ""); |
| 399 | } |
| 400 | assert_eq!(ENV.config_file_candidates(), vec![dir.join("env")]); |
| 401 | let c = LlmConfig::from_env(&ENV); |
| 402 | assert_eq!(c.api_key.as_deref(), Some("file:key")); |
| 403 | assert_eq!(c.model, "m1"); |
| 404 | unsafe { std::env::set_var("ONDE_API_KEY", "env:key") }; |
| 405 | assert_eq!( |
| 406 | LlmConfig::from_env(&ENV).api_key.as_deref(), |
| 407 | Some("env:key") |
| 408 | ); |
| 409 | unsafe { |
| 410 | std::env::remove_var("ONDE_API_KEY"); |
| 411 | std::env::remove_var("SPLITFIRE_DATA_DIR"); |
| 412 | } |
| 413 | let _ = std::fs::remove_dir_all(dir); |
| 414 | } |
| 415 | |
| 416 | #[test] |
| 417 | fn usage_tolerates_missing_total() { |
| 418 | let chunk: Chunk = serde_json::from_str( |
| 419 | r#"{"choices":[],"usage":{"prompt_tokens":7,"completion_tokens":3}}"#, |
| 420 | ) |
| 421 | .unwrap(); |
| 422 | assert_eq!(chunk.usage.unwrap().used(), 10); |
| 423 | let full = Usage { |
| 424 | prompt_tokens: 1, |
| 425 | completion_tokens: 1, |
| 426 | total_tokens: 5, |
| 427 | }; |
| 428 | assert_eq!(full.used(), 5); |
| 429 | } |
| 430 | |
| 431 | #[test] |
| 432 | fn parses_models_response() { |
| 433 | let body = r#"{ |
| 434 | "object": "list", |
| 435 | "data": [ |
| 436 | {"id": "onde-balanced", "object": "model", "created": 1700000000, "owned_by": "onde"}, |
| 437 | {"id": "custom-model"} |
| 438 | ] |
| 439 | }"#; |
| 440 | let parsed: ModelsResponse = serde_json::from_str(body).unwrap(); |
| 441 | assert_eq!(parsed.data.len(), 2); |
| 442 | assert_eq!(parsed.data[0].owned_by.as_deref(), Some("onde")); |
| 443 | assert_eq!(parsed.data[1].owned_by, None); |
| 444 | } |
| 445 | } |
| 446 | |
| 447 | #[derive(Debug, Clone, Default)] |
| 448 | pub struct ToolCallRequest { |
| 449 | pub id: String, |
| 450 | pub name: String, |
| 451 | pub arguments: String, |
| 452 | /// Provider extras that must be echoed back (e.g. Gemini's `thought_signature`). |
| 453 | pub extra_content: Option<Value>, |
| 454 | } |
| 455 | |
| 456 | #[derive(Debug, Default)] |
| 457 | pub struct Completion { |
| 458 | pub content: String, |
| 459 | pub tool_calls: Vec<ToolCallRequest>, |
| 460 | pub finish_reason: Option<String>, |
| 461 | /// Token usage, when the endpoint reports it (`stream_options.include_usage`). |
| 462 | pub usage: Option<Usage>, |
| 463 | } |
| 464 | |
| 465 | #[derive(Debug, Clone, Copy, Default, Deserialize)] |
| 466 | pub struct Usage { |
| 467 | #[serde(default)] |
| 468 | pub prompt_tokens: u64, |
| 469 | #[serde(default)] |
| 470 | pub completion_tokens: u64, |
| 471 | #[serde(default)] |
| 472 | pub total_tokens: u64, |
| 473 | } |
| 474 | |
| 475 | impl Usage { |
| 476 | /// Tokens in the context after this completion. Some servers leave `total_tokens` at 0. |
| 477 | pub fn used(&self) -> u64 { |
| 478 | self.total_tokens |
| 479 | .max(self.prompt_tokens + self.completion_tokens) |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | impl Completion { |
| 484 | /// The assistant message to append to the conversation history. |
| 485 | pub fn to_message(&self) -> Value { |
| 486 | let mut msg = json!({ "role": "assistant", "content": self.content }); |
| 487 | if !self.tool_calls.is_empty() { |
| 488 | msg["tool_calls"] = self |
| 489 | .tool_calls |
| 490 | .iter() |
| 491 | .map(|tc| { |
| 492 | let mut call = json!({ |
| 493 | "id": tc.id, |
| 494 | "type": "function", |
| 495 | "function": { "name": tc.name, "arguments": tc.arguments }, |
| 496 | }); |
| 497 | if let Some(extra) = &tc.extra_content { |
| 498 | call["extra_content"] = extra.clone(); |
| 499 | } |
| 500 | call |
| 501 | }) |
| 502 | .collect(); |
| 503 | } |
| 504 | msg |
| 505 | } |
| 506 | } |
| 507 | |
| 508 | /// Streamed pieces surfaced to the caller as they arrive. |
| 509 | pub enum Delta<'a> { |
| 510 | Text(&'a str), |
| 511 | Reasoning(&'a str), |
| 512 | } |
| 513 | |
| 514 | #[derive(Deserialize)] |
| 515 | struct Chunk { |
| 516 | #[serde(default)] |
| 517 | choices: Vec<Choice>, |
| 518 | usage: Option<Usage>, |
| 519 | } |
| 520 | |
| 521 | #[derive(Deserialize)] |
| 522 | struct Choice { |
| 523 | #[serde(default)] |
| 524 | delta: ChunkDelta, |
| 525 | finish_reason: Option<String>, |
| 526 | } |
| 527 | |
| 528 | #[derive(Deserialize, Default)] |
| 529 | struct ChunkDelta { |
| 530 | content: Option<String>, |
| 531 | // Non-standard but common (DeepSeek, vLLM, OpenRouter, llama.cpp). |
| 532 | reasoning_content: Option<String>, |
| 533 | reasoning: Option<String>, |
| 534 | #[serde(default)] |
| 535 | tool_calls: Vec<ToolCallDelta>, |
| 536 | } |
| 537 | |
| 538 | #[derive(Deserialize)] |
| 539 | struct ToolCallDelta { |
| 540 | #[serde(default)] |
| 541 | index: usize, |
| 542 | id: Option<String>, |
| 543 | function: Option<FunctionDelta>, |
| 544 | extra_content: Option<Value>, |
| 545 | } |
| 546 | |
| 547 | #[derive(Deserialize)] |
| 548 | struct FunctionDelta { |
| 549 | name: Option<String>, |
| 550 | arguments: Option<String>, |
| 551 | } |
| 552 | |
| 553 | /// A model from `GET /v1/models`. Extra fields in the response are ignored. |
| 554 | #[derive(Debug, Clone, Deserialize)] |
| 555 | pub struct ModelInfo { |
| 556 | pub id: String, |
| 557 | #[serde(default)] |
| 558 | pub owned_by: Option<String>, |
| 559 | } |
| 560 | |
| 561 | #[derive(Deserialize)] |
| 562 | struct ModelsResponse { |
| 563 | data: Vec<ModelInfo>, |
| 564 | } |
| 565 | |
| 566 | #[derive(Debug, Clone)] |
| 567 | pub struct LlmClient { |
| 568 | http: reqwest::Client, |
| 569 | config: LlmConfig, |
| 570 | } |
| 571 | |
| 572 | impl LlmClient { |
| 573 | pub fn new(config: LlmConfig) -> Self { |
| 574 | Self { |
| 575 | http: reqwest::Client::new(), |
| 576 | config, |
| 577 | } |
| 578 | } |
| 579 | |
| 580 | pub fn model(&self) -> &str { |
| 581 | &self.config.model |
| 582 | } |
| 583 | |
| 584 | pub fn config(&self) -> &LlmConfig { |
| 585 | &self.config |
| 586 | } |
| 587 | |
| 588 | /// Whether the provider can be used: an API key is configured, or the provider needs none |
| 589 | /// ([`Provider::Local`]). The agent answers `AUTH_REQUIRED` when this is false. |
| 590 | pub fn has_api_key(&self) -> bool { |
| 591 | self.config.provider == Provider::Local || self.config.api_key.is_some() |
| 592 | } |
| 593 | |
| 594 | /// Verify the configured key with a minimal chat completion. |
| 595 | /// Returns `Err` when no key is set, the key is rejected, or the endpoint is unreachable. |
| 596 | pub async fn check_auth(&self) -> Result<()> { |
| 597 | let provider = self.config.provider; |
| 598 | if provider == Provider::Local { |
| 599 | return Ok(()); |
| 600 | } |
| 601 | let Some(key) = &self.config.api_key else { |
| 602 | bail!("no {} configured", provider.key_var()); |
| 603 | }; |
| 604 | let body = json!({ |
| 605 | "model": self.config.model, |
| 606 | "messages": [{ "role": "user", "content": "ping" }], |
| 607 | "max_tokens": 1, |
| 608 | "stream": false, |
| 609 | }); |
| 610 | let resp = self |
| 611 | .http |
| 612 | .post(format!("{}/chat/completions", self.config.base_url)) |
| 613 | .json(&body) |
| 614 | .bearer_auth(key) |
| 615 | .send() |
| 616 | .await |
| 617 | .with_context(|| format!("reaching {}", provider.display_name()))?; |
| 618 | let status = resp.status(); |
| 619 | if matches!(status.as_u16(), 401 | 403) { |
| 620 | bail!( |
| 621 | "{} rejected the API key ({status})", |
| 622 | provider.display_name() |
| 623 | ); |
| 624 | } |
| 625 | if !status.is_success() { |
| 626 | let text = resp.text().await.unwrap_or_default(); |
| 627 | bail!("{} returned {status}: {text}", provider.display_name()); |
| 628 | } |
| 629 | Ok(()) |
| 630 | } |
| 631 | |
| 632 | /// List the models the configured endpoint serves (`GET {base_url}/models`). |
| 633 | pub async fn models(&self) -> Result<Vec<ModelInfo>> { |
| 634 | if self.config.provider == Provider::Local { |
| 635 | return Ok(crate::local::catalog()); |
| 636 | } |
| 637 | let url = format!("{}/models", self.config.base_url); |
| 638 | let mut req = self.http.get(&url); |
| 639 | if let Some(key) = &self.config.api_key { |
| 640 | req = req.bearer_auth(key); |
| 641 | } |
| 642 | let mut resp = req.send().await.context("sending models request")?; |
| 643 | // Some endpoints (Onde Cloud) list models publicly but reject an invalid key, so a |
| 644 | // bad key shouldn't hide the catalog: retry without auth. |
| 645 | if self.config.api_key.is_some() && matches!(resp.status().as_u16(), 401 | 403) { |
| 646 | resp = self |
| 647 | .http |
| 648 | .get(&url) |
| 649 | .send() |
| 650 | .await |
| 651 | .context("sending models request")?; |
| 652 | } |
| 653 | if !resp.status().is_success() { |
| 654 | let status = resp.status(); |
| 655 | let text = resp.text().await.unwrap_or_default(); |
| 656 | bail!("models endpoint returned {status}: {text}"); |
| 657 | } |
| 658 | let mut models = resp |
| 659 | .json::<ModelsResponse>() |
| 660 | .await |
| 661 | .context("parsing models response")? |
| 662 | .data; |
| 663 | models.sort_by(|a, b| a.id.cmp(&b.id)); |
| 664 | Ok(models) |
| 665 | } |
| 666 | |
| 667 | /// Run one streaming chat completion, invoking `on_delta` for every text fragment. |
| 668 | pub async fn complete( |
| 669 | &self, |
| 670 | model: &str, |
| 671 | messages: &[Value], |
| 672 | tools: &Value, |
| 673 | mut on_delta: impl FnMut(Delta<'_>), |
| 674 | ) -> Result<Completion> { |
| 675 | if self.config.provider == Provider::Local { |
| 676 | return crate::local::complete(model, messages, tools, on_delta).await; |
| 677 | } |
| 678 | if self.config.api_key.is_none() { |
| 679 | bail!( |
| 680 | "No API key configured. Set {} in the environment or run the agent's `--setup`", |
| 681 | self.config.provider.key_var() |
| 682 | ); |
| 683 | } |
| 684 | let mut body = json!({ |
| 685 | "model": model, |
| 686 | "messages": messages, |
| 687 | "tools": tools, |
| 688 | "stream": true, |
| 689 | "stream_options": { "include_usage": true }, |
| 690 | }); |
| 691 | let send = |body: &Value| { |
| 692 | let mut req = self |
| 693 | .http |
| 694 | .post(format!("{}/chat/completions", self.config.base_url)) |
| 695 | .json(body); |
| 696 | if let Some(key) = &self.config.api_key { |
| 697 | req = req.bearer_auth(key); |
| 698 | } |
| 699 | req.send() |
| 700 | }; |
| 701 | let mut resp = send(&body) |
| 702 | .await |
| 703 | .context("sending chat completion request")?; |
| 704 | // Not every OpenAI-compatible server knows `stream_options`; retry without it. |
| 705 | if matches!(resp.status().as_u16(), 400 | 422) { |
| 706 | body.as_object_mut().unwrap().remove("stream_options"); |
| 707 | resp = send(&body) |
| 708 | .await |
| 709 | .context("sending chat completion request")?; |
| 710 | } |
| 711 | if !resp.status().is_success() { |
| 712 | let status = resp.status(); |
| 713 | let text = resp.text().await.unwrap_or_default(); |
| 714 | bail!("LLM endpoint returned {status}: {text}"); |
| 715 | } |
| 716 | |
| 717 | let mut out = Completion::default(); |
| 718 | let mut stream = resp.bytes_stream(); |
| 719 | let mut buf: Vec<u8> = Vec::new(); |
| 720 | while let Some(bytes) = stream.next().await { |
| 721 | buf.extend_from_slice(&bytes.context("reading response stream")?); |
| 722 | while let Some(pos) = buf.iter().position(|&b| b == b'\n') { |
| 723 | let line: Vec<u8> = buf.drain(..=pos).collect(); |
| 724 | let line = String::from_utf8_lossy(&line); |
| 725 | let Some(data) = line.trim().strip_prefix("data:") else { |
| 726 | continue; |
| 727 | }; |
| 728 | let data = data.trim(); |
| 729 | if data == "[DONE]" { |
| 730 | return Ok(out); |
| 731 | } |
| 732 | let chunk: Chunk = match serde_json::from_str(data) { |
| 733 | Ok(c) => c, |
| 734 | Err(e) => { |
| 735 | tracing::warn!("skipping unparseable chunk ({e}): {data}"); |
| 736 | continue; |
| 737 | } |
| 738 | }; |
| 739 | if chunk.usage.is_some() { |
| 740 | out.usage = chunk.usage; |
| 741 | } |
| 742 | for choice in chunk.choices { |
| 743 | let d = choice.delta; |
| 744 | if let Some(r) = d.reasoning_content.as_deref().or(d.reasoning.as_deref()) |
| 745 | && !r.is_empty() |
| 746 | { |
| 747 | on_delta(Delta::Reasoning(r)); |
| 748 | } |
| 749 | if let Some(t) = d.content.as_deref() |
| 750 | && !t.is_empty() |
| 751 | { |
| 752 | out.content.push_str(t); |
| 753 | on_delta(Delta::Text(t)); |
| 754 | } |
| 755 | for tc in d.tool_calls { |
| 756 | if out.tool_calls.len() <= tc.index { |
| 757 | out.tool_calls |
| 758 | .resize(tc.index + 1, ToolCallRequest::default()); |
| 759 | } |
| 760 | let slot = &mut out.tool_calls[tc.index]; |
| 761 | if let Some(id) = tc.id { |
| 762 | slot.id = id; |
| 763 | } |
| 764 | if tc.extra_content.is_some() { |
| 765 | slot.extra_content = tc.extra_content; |
| 766 | } |
| 767 | if let Some(f) = tc.function { |
| 768 | if let Some(n) = f.name { |
| 769 | slot.name.push_str(&n); |
| 770 | } |
| 771 | if let Some(a) = f.arguments { |
| 772 | slot.arguments.push_str(&a); |
| 773 | } |
| 774 | } |
| 775 | } |
| 776 | if choice.finish_reason.is_some() { |
| 777 | out.finish_reason = choice.finish_reason; |
| 778 | } |
| 779 | } |
| 780 | } |
| 781 | } |
| 782 | Ok(out) |
| 783 | } |
| 784 | } |