mod codex;
use reqwest::Client;
use serde::Deserialize;
use tracing::{debug, instrument};
use crate::key_pool::{KeyLease, KeyPool};
use crate::protocol::ProtocolHandler;
use crate::schema::{ApiProtocol, ModelSchema, ModelSource, ProprietaryAuth};
use crate::tasks::ContentBlock;
use crate::tls_client::Degradation;
use crate::{InferenceError, InferenceRetryProgress};
use std::collections::HashMap;
use std::future::Future;
use std::sync::{Mutex, OnceLock};
#[cfg(test)]
pub(crate) const PARSLEE_ACCESS_TOKEN_ENV: &str = "PARSLEE_ACCESS_TOKEN";
const REMOTE_MAX_ATTEMPTS: u32 = 3;
const REMOTE_BACKOFF_BASE_SECS: u64 = 3;
const REMOTE_PHASE_DEFAULT_SECS: u64 = 300;
const REMOTE_PHASE_DECODE_HEADROOM: f64 = 2.0;
const REMOTE_RETRY_FLOOR_SECS: u64 = 5;
const REMOTE_WATCHDOG_TICK_MS: u64 = 200;
const REMOTE_ABSOLUTE_CEILING_SECS: u64 = 2 * 60 * 60;
enum PhaseCut {
Deadline { applied_ms: u64 },
Default,
}
async fn phase_bounded<T, F>(
fut: F,
phase_start: std::time::Instant,
default: std::time::Duration,
) -> Result<T, PhaseCut>
where
F: std::future::Future<Output = T>,
{
tokio::pin!(fut);
loop {
let tick = tokio::time::sleep(std::time::Duration::from_millis(REMOTE_WATCHDOG_TICK_MS));
tokio::select! {
out = &mut fut => return Ok(out),
_ = tick => {
match crate::offload::current_remote_deadline() {
Some(d) => {
if d.remaining().is_zero() {
return Err(PhaseCut::Deadline { applied_ms: d.applied_ms });
}
}
None => {
if phase_start.elapsed() >= default {
return Err(PhaseCut::Default);
}
}
}
}
}
}
}
fn unarmed_phase_default(schema: &ModelSchema, max_tokens: usize) -> std::time::Duration {
let floor = REMOTE_PHASE_DEFAULT_SECS as f64;
let needed = schema
.performance
.tokens_per_second
.filter(|tps| tps.is_finite() && *tps > 0.0)
.map(|tps| max_tokens as f64 / tps * REMOTE_PHASE_DECODE_HEADROOM)
.unwrap_or(0.0);
let ceiling = (REMOTE_ABSOLUTE_CEILING_SECS / 2) as f64;
std::time::Duration::from_secs_f64(needed.clamp(floor, ceiling))
}
fn deadline_permits_retry(backoff: std::time::Duration) -> Result<(), InferenceError> {
let Some(d) = crate::offload::current_remote_deadline() else {
return Ok(());
};
let need = backoff + std::time::Duration::from_secs(REMOTE_RETRY_FLOOR_SECS);
if d.remaining() > need {
return Ok(());
}
Err(InferenceError::DeadlineExceeded {
applied_ms: d.applied_ms,
elapsed_ms: d
.applied_ms
.saturating_sub(d.remaining().as_millis() as u64),
last_error: "remaining deadline cannot fit another retry attempt".to_string(),
})
}
fn phase_cut_error(
cut: PhaseCut,
call_start: std::time::Instant,
phase: &str,
phase_default: std::time::Duration,
) -> Result<TransportAttemptError, InferenceError> {
match cut {
PhaseCut::Deadline { applied_ms } => Err(InferenceError::DeadlineExceeded {
applied_ms,
elapsed_ms: call_start.elapsed().as_millis() as u64,
last_error: format!("{phase} was still in flight when the deadline elapsed"),
}),
PhaseCut::Default => Ok(TransportAttemptError::timeout(&format!(
"{phase} timed out after {}s (no caller deadline armed)",
phase_default.as_secs()
))),
}
}
fn is_transient_http_status(code: u16) -> bool {
matches!(code, 429 | 500 | 502 | 503 | 529)
}
const MAX_ERROR_CHAIN_LEN: usize = 512;
const MAX_ERROR_CHAIN_DEPTH: usize = 8;
const SENSITIVE_QUERY_PARAMS: &[&str] = &[
"key",
"api_key",
"apikey",
"access_token",
"token",
"password",
"secret",
"signature",
"sig",
];
fn redact_sensitive(text: &str) -> String {
let lower = text.to_ascii_lowercase();
let bytes = text.as_bytes();
let mut out = String::with_capacity(text.len());
let mut i = 0usize;
while i < text.len() {
let userinfo_end = if i >= 3 && &bytes[i - 3..i] == b"://" {
let authority_end = text[i..]
.find(|c: char| c.is_whitespace() || matches!(c, '/' | '?' | '#' | ')'))
.map_or(text.len(), |offset| i + offset);
text[i..authority_end]
.rfind('@')
.map(|offset| i + offset + 1)
} else {
None
};
if let Some(end) = userinfo_end {
out.push_str("REDACTED@");
i = end;
continue;
}
let value_start = if i == 0 || matches!(bytes[i - 1], b'?' | b'&') {
SENSITIVE_QUERY_PARAMS
.iter()
.find(|name| {
let end = i + name.len();
end < text.len() && lower[i..].starts_with(**name) && bytes[end] == b'='
})
.map(|name| i + name.len() + 1)
} else {
None
};
match value_start {
Some(start) => {
out.push_str(&text[i..start]);
out.push_str("REDACTED");
i = text[start..]
.find(|c: char| c == '&' || c == ')' || c.is_whitespace())
.map_or(text.len(), |off| start + off);
}
None => {
let step = text[i..].chars().next().map_or(1, char::len_utf8);
out.push_str(&text[i..i + step]);
i += step;
}
}
}
out
}
fn error_chain(err: &dyn std::error::Error) -> String {
let mut rendered = err.to_string();
let mut source = err.source();
let mut depth = 0usize;
while let Some(cause) = source {
if depth >= MAX_ERROR_CHAIN_DEPTH {
rendered.push_str(": ...");
break;
}
let text = cause.to_string();
if !rendered.ends_with(&text) {
rendered.push_str(": ");
rendered.push_str(&text);
}
source = cause.source();
depth += 1;
}
let redacted = redact_sensitive(&rendered);
if redacted.len() <= MAX_ERROR_CHAIN_LEN {
return redacted;
}
let mut end = MAX_ERROR_CHAIN_LEN;
while end > 0 && !redacted.is_char_boundary(end) {
end -= 1;
}
format!("{}... (truncated)", &redacted[..end])
}
fn reqwest_error_is_transient(e: &reqwest::Error) -> bool {
if e.is_builder() {
return false;
}
if e.is_redirect() || e.is_status() {
return false;
}
true
}
struct TransportAttemptError {
error: InferenceError,
retryable: bool,
connect: bool,
unbilled: bool,
}
impl TransportAttemptError {
#[cfg(test)]
fn from_reqwest(context: &str, e: &reqwest::Error) -> Self {
Self::from_reqwest_error(
InferenceError::InferenceFailed(format!("{context}: {}", error_chain(e))),
e,
)
}
fn from_reqwest_error(error: InferenceError, e: &reqwest::Error) -> Self {
Self {
error,
retryable: reqwest_error_is_transient(e),
connect: e.is_connect(),
unbilled: e.is_connect(),
}
}
fn timeout(message: &str) -> Self {
Self {
error: InferenceError::InferenceFailed(message.to_string()),
retryable: true,
connect: false,
unbilled: false,
}
}
}
fn unread_success_body(
status: reqwest::StatusCode,
failure: TransportAttemptError,
) -> TransportAttemptError {
if !status.is_success() {
return TransportAttemptError {
unbilled: true,
..failure
};
}
TransportAttemptError {
error: InferenceError::Transient {
status: Some(status.as_u16()),
message: format!(
"the provider answered {status} but its response body could not be read: {}",
failure.error
),
},
retryable: false,
connect: false,
unbilled: false,
}
}
fn exhausted_transport_error(
failure: TransportAttemptError,
url: &str,
attempts: u32,
elapsed: std::time::Duration,
) -> InferenceError {
if !failure.retryable {
return failure.error;
}
let detail = failure.error.to_string();
let message = match reqwest::Url::parse(url)
.ok()
.and_then(|url| url.host_str().map(str::to_owned))
{
Some(host) if failure.connect => format!(
"could not reach {host} for {}s ({attempts} attempts): {detail}",
elapsed.as_secs()
),
_ => detail,
};
InferenceError::Transient {
status: None,
message,
}
}
fn openrouter_model_is_unavailable(status: reqwest::StatusCode, body: &str) -> bool {
if status == reqwest::StatusCode::NOT_FOUND {
return true;
}
if status != reqwest::StatusCode::BAD_REQUEST {
return false;
}
let message = serde_json::from_str::<serde_json::Value>(body)
.ok()
.and_then(|value| {
value
.pointer("/error/message")
.and_then(serde_json::Value::as_str)
.map(str::to_lowercase)
})
.unwrap_or_default();
[
"no endpoints found",
"model not found",
"invalid model",
"does not exist",
"deprecated",
]
.iter()
.any(|needle| message.contains(needle))
}
pub(crate) fn is_account_http_status(status: u16) -> bool {
matches!(status, 401..=403)
}
const EXPIRY_WARNING_HORIZON_SECS: u64 = 15 * 60;
const EXPIRY_PROBE_INTERVAL_SECS: u64 = 5 * 60;
fn warn_if_token_expires_soon(expires_at: u64) {
static LAST_PROBE: OnceLock<Mutex<Option<std::time::Instant>>> = OnceLock::new();
static WARNED_FOR: OnceLock<Mutex<Option<u64>>> = OnceLock::new();
if expires_at == 0 {
return;
}
{
let probe_gate = LAST_PROBE.get_or_init(|| Mutex::new(None));
let mut last = match probe_gate.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if last.is_some_and(|at| {
at.elapsed() < std::time::Duration::from_secs(EXPIRY_PROBE_INTERVAL_SECS)
}) {
return;
}
*last = Some(std::time::Instant::now());
}
let remaining = expires_at.saturating_sub(crate::remote::epoch_seconds());
if remaining > EXPIRY_WARNING_HORIZON_SECS {
return;
}
{
let warned_gate = WARNED_FOR.get_or_init(|| Mutex::new(None));
let mut warned = match warned_gate.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
};
if warned.is_some_and(|seen| seen.abs_diff(expires_at) <= EXPIRY_PROBE_INTERVAL_SECS) {
return;
}
*warned = Some(expires_at);
}
tracing::warn!(
target: "car_inference::auth",
remaining_secs = remaining,
expires_at_unix = expires_at,
"Parslee token expires in {}m — an operation longer than that will fail \
partway unless it can resume. Re-authenticate with `car auth login` first \
if you are starting a long run.",
remaining / 60
);
}
fn epoch_seconds() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
use crate::stream::content_refusal_tags;
fn gateway_unconfigured_detail(status: reqwest::StatusCode, body: &str) -> Option<String> {
if !status.is_server_error() && status != reqwest::StatusCode::NOT_IMPLEMENTED {
return None;
}
let json: serde_json::Value = serde_json::from_str(body).ok()?;
let error = json.get("error")?;
let code = error.get("code").and_then(|c| c.as_str())?;
if !code.ends_with("_not_configured") {
return None;
}
Some(
error
.get("message")
.and_then(|m| m.as_str())
.map(str::trim)
.filter(|m| !m.is_empty())
.unwrap_or(code)
.to_string(),
)
}
fn gateway_unconfigured_error(
status: reqwest::StatusCode,
message: String,
model: &str,
remember_global: bool,
) -> InferenceError {
let namespace = if crate::openrouter::is_curated_managed_gateway_alias(model) {
if remember_global {
crate::openrouter::note_gateway_unconfigured();
}
"parslee/openrouter/"
} else {
"parslee/"
};
InferenceError::GatewayUnconfigured {
provider: "parslee".to_string(),
namespace: namespace.to_string(),
status: status.as_u16(),
message,
}
}
fn account_error_message(protocol: ApiProtocol, status: u16) -> String {
match (protocol, status) {
(ApiProtocol::OpenRouter, 402) => {
"OpenRouter account is out of credits — add credits in OpenRouter and retry".to_string()
}
(ApiProtocol::OpenRouter, _) => {
"OpenRouter key rejected — re-add it with `car keys set openrouter`".to_string()
}
(_, 402) => "provider account is out of credits or over quota".to_string(),
_ => "provider rejected the API key — check the configured credential".to_string(),
}
}
fn openrouter_http_error(status: reqwest::StatusCode, body: &str, model: &str) -> InferenceError {
if openrouter_model_is_unavailable(status, body) {
return InferenceError::InferenceFailed(format!(
"OpenRouter model '{model}' is no longer available on OpenRouter"
));
}
InferenceError::InferenceFailed(format!("OpenRouter request failed with HTTP {status}"))
}
fn google_finish_outcome(reason: &str) -> Result<(), &'static str> {
match reason {
"STOP" | "MAX_TOKENS" => Ok(()),
"SAFETY" => Err("Google blocked the response for safety"),
"RECITATION" => Err("Google blocked the response for recitation"),
"BLOCKLIST" => Err("Google blocked the response because of a blocklist match"),
"PROHIBITED_CONTENT" => Err("Google blocked prohibited response content"),
"SPII" => Err("Google blocked the response because it may contain sensitive information"),
"MALFORMED_FUNCTION_CALL" => Err("Google returned a malformed function call"),
_ => Err("Google inference ended with an abnormal finish reason"),
}
}
fn refreshed_parslee_authority(
result: car_auth::ReactiveRefreshResult,
model: &str,
rejected: InferenceError,
) -> Result<(String, String), InferenceError> {
match result {
car_auth::ReactiveRefreshResult::Refreshed(credential) => {
Ok((credential.api_base, credential.access_token))
}
car_auth::ReactiveRefreshResult::ReauthenticationRequired { expires_at } => {
Err(InferenceError::CredentialUnavailable {
provider: "parslee".to_string(),
model: model.to_string(),
reason: crate::CredentialFailure::Expired { expires_at },
detail: "Your Parslee sign-in has expired. Sign in again to continue.".to_string(),
})
}
car_auth::ReactiveRefreshResult::SignedOut => {
Err(parslee_signed_out_error_for_model(model, "parslee"))
}
car_auth::ReactiveRefreshResult::TransientOrBackoff if is_bearer_refusal(&rejected) => {
Err(InferenceError::Transient {
status: None,
message: car_auth::TOKEN_RENEWAL_UNAVAILABLE.to_string(),
})
}
car_auth::ReactiveRefreshResult::TransientOrBackoff => Err(rejected),
car_auth::ReactiveRefreshResult::EnvOverrideRejected => {
Err(parslee_env_override_rejected_error())
}
car_auth::ReactiveRefreshResult::StoreError(read_error) => Err(
parslee_credential_read_error_for_model(model, "parslee", read_error),
),
}
}
fn is_bearer_refusal(e: &InferenceError) -> bool {
match e {
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::Expired { .. },
..
} => true,
InferenceError::ProviderAccount { status, .. } => *status == 401,
_ => first_http_status(&e.to_string()) == Some(401),
}
}
fn first_http_status(message: &str) -> Option<u16> {
message.match_indices("HTTP ").find_map(|(index, marker)| {
let digits = message.get(index + marker.len()..index + marker.len() + 3)?;
digits
.bytes()
.all(|byte| byte.is_ascii_digit())
.then(|| digits.parse().ok())
.flatten()
})
}
fn is_auth_rejection(e: &InferenceError) -> bool {
if matches!(
e,
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::Expired { .. },
..
}
) {
return true;
}
let m = e.to_string();
m.contains("HTTP 401") || m.contains("HTTP 403")
}
fn find_sse_separator(buf: &[u8]) -> Option<(usize, usize)> {
let lf = buf.windows(2).position(|w| w == b"\n\n").map(|p| (p, 2));
let crlf = buf
.windows(4)
.position(|w| w == b"\r\n\r\n")
.map(|p| (p, 4));
match (lf, crlf) {
(Some(a), Some(b)) => Some(if a.0 <= b.0 { a } else { b }),
(Some(a), None) => Some(a),
(None, Some(b)) => Some(b),
(None, None) => None,
}
}
type ParsleeIdentityCache = HashMap<(String, String), (String, String)>;
static PARSLEE_IDENTITY: OnceLock<tokio::sync::Mutex<ParsleeIdentityCache>> = OnceLock::new();
pub struct RemoteBackend {
pub(crate) client: Client,
pub(crate) tls_degradation: Option<Degradation>,
pub key_pool: std::sync::Arc<KeyPool>,
work_context: Option<car_auth::context::CredentialContext>,
context_client: std::sync::Arc<OnceLock<(Client, Option<Degradation>)>>,
codex_client: std::sync::Arc<OnceLock<(Client, Option<Degradation>)>>,
spend_gate: std::sync::Arc<std::sync::RwLock<Option<crate::spend::InstalledGate>>>,
}
#[derive(Clone, Copy)]
enum RedirectRule {
Follow,
Refuse,
SameOrigin,
}
impl RedirectRule {
fn policy(self) -> Option<reqwest::redirect::Policy> {
match self {
Self::Follow => None,
Self::Refuse => Some(reqwest::redirect::Policy::none()),
Self::SameOrigin => Some(crate::http_redirect::same_origin(
crate::http_redirect::CrossOrigin::Fail,
)),
}
}
}
pub fn estimate_tokens(text: &str) -> usize {
std::cmp::max(1, text.len() / 4)
}
fn request_input_estimate(req: &crate::protocol::ApiRequest) -> u64 {
req.messages
.iter()
.filter_map(|m| m.get("content").and_then(|c| c.as_str()))
.map(|s| s.len() / 4)
.sum::<usize>() as u64
}
fn key_pool_token_counts(
input_estimate: u64,
usage: Option<&crate::TokenUsage>,
text: &str,
tool_calls: &[crate::ToolCall],
) -> (u64, u64) {
let input = usage.map_or(input_estimate, |u| {
u.prompt_tokens + u.cache_read_input_tokens + u.cache_creation_input_tokens
});
let output = crate::booked_output_tokens(usage, text, tool_calls) as u64;
(input, output)
}
struct KeySettlement {
pool: std::sync::Arc<KeyPool>,
endpoint: String,
env_var: String,
settled: bool,
}
impl KeySettlement {
fn new(pool: std::sync::Arc<KeyPool>, endpoint: &str, env_var: &str) -> Self {
Self {
pool,
endpoint: endpoint.to_string(),
env_var: env_var.to_string(),
settled: env_var.is_empty(),
}
}
async fn succeed(mut self, latency_ms: u64, input_tokens: u64, output_tokens: u64) {
if !self.settled {
self.settled = true;
self.pool
.report_success(
&self.endpoint,
&self.env_var,
latency_ms,
input_tokens,
output_tokens,
)
.await;
}
}
async fn fail(mut self, is_rate_limit: bool) {
if !self.settled {
self.settled = true;
self.pool
.report_failure(&self.endpoint, &self.env_var, is_rate_limit)
.await;
}
}
}
impl Drop for KeySettlement {
fn drop(&mut self) {
if self.settled {
return;
}
if let Ok(runtime) = tokio::runtime::Handle::try_current() {
let pool = self.pool.clone();
let endpoint = std::mem::take(&mut self.endpoint);
let env_var = std::mem::take(&mut self.env_var);
runtime.spawn(async move { pool.release(&endpoint, &env_var).await });
}
}
}
struct BookingSender {
tx: tokio::sync::mpsc::Sender<crate::stream::StreamEvent>,
seen: crate::stream::StreamAccumulator,
completed: bool,
errored: bool,
cancelled: bool,
}
impl BookingSender {
fn new(tx: tokio::sync::mpsc::Sender<crate::stream::StreamEvent>) -> Self {
Self {
tx,
seen: crate::stream::StreamAccumulator::default(),
completed: false,
errored: false,
cancelled: false,
}
}
async fn send(
&mut self,
event: crate::stream::StreamEvent,
) -> Result<(), tokio::sync::mpsc::error::SendError<crate::stream::StreamEvent>> {
let (completed, errored, cancelled) = match &event {
crate::stream::StreamEvent::Done { .. } => (true, false, false),
crate::stream::StreamEvent::Error(_) => (false, true, false),
crate::stream::StreamEvent::Attribution(
crate::outcome::UnattributedReason::Cancelled,
) => (false, false, true),
_ => (false, false, false),
};
self.seen.push(&event);
self.tx.send(event).await?;
self.completed |= completed;
self.errored |= errored;
self.cancelled |= cancelled;
Ok(())
}
async fn settle(self, settlement: KeySettlement, latency_ms: u64, input_estimate: u64) {
let caller_gone = self.tx.is_closed();
if self.cancelled || (self.completed && caller_gone) {
drop(settlement);
} else if self.errored {
settlement.fail(false).await;
} else if self.completed {
let (text, tool_calls, usage, _) = self.seen.finish_with_usage();
let (input, output) =
key_pool_token_counts(input_estimate, usage.as_ref(), &text, &tool_calls);
settlement.succeed(latency_ms, input, output).await;
}
}
}
fn parse_parslee_responses_sse(raw: &str) -> Vec<crate::stream::StreamEvent> {
let mut events = Vec::new();
let normalized = if raw.contains("\r\n") {
std::borrow::Cow::Owned(raw.replace("\r\n", "\n"))
} else {
std::borrow::Cow::Borrowed(raw)
};
for block in normalized.split("\n\n") {
let block = block.trim();
if block.is_empty() {
continue;
}
let mut event_type = "";
let mut data = String::new();
for line in block.lines() {
if let Some(t) = line.strip_prefix("event:") {
event_type = t.trim();
} else if let Some(d) = line.strip_prefix("data:") {
if !data.is_empty() {
data.push('\n');
}
data.push_str(d.trim());
}
}
if event_type.is_empty() || data.is_empty() {
continue;
}
events.extend(crate::stream::parse_openai_responses_sse_line(
event_type, &data,
));
}
events
}
fn pump_responses_sse(
resp: reqwest::Response,
spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
label: &'static str,
) -> tokio::sync::mpsc::Receiver<crate::stream::StreamEvent> {
let (tx, rx) = tokio::sync::mpsc::channel(64);
tokio::spawn(async move {
use futures::StreamExt;
let mut stream = resp.bytes_stream();
let mut buffer = Vec::<u8>::new();
let mut spend_guard = spend_guard;
let mut saw_completed = false;
while let Some(chunk) = stream.next().await {
match chunk {
Ok(bytes) => buffer.extend_from_slice(&bytes),
Err(error) => {
let message = format!("{label} stream transport error");
tracing::warn!(%error, "{message}");
let _ = tx.send(crate::stream::StreamEvent::Error(message)).await;
return;
}
}
while let Some((position, separator_len)) = find_sse_separator(&buffer) {
let block = String::from_utf8_lossy(&buffer[..position]).into_owned();
buffer.drain(..position + separator_len);
for event in parse_parslee_responses_sse(&block) {
if let Some(guard) = spend_guard.as_mut() {
if let crate::stream::StreamEvent::Usage { input_tokens, .. } = &event {
guard.note_prompt_tokens(*input_tokens);
}
let delta = match &event {
crate::stream::StreamEvent::TextDelta(text) => Some(text.as_str()),
crate::stream::StreamEvent::ToolCallDelta {
arguments_delta, ..
} => Some(arguments_delta.as_str()),
_ => None,
};
if let Some(delta) = delta {
let tokens =
crate::routing_ext::MidStreamSpendGuard::estimate_tokens(delta);
if let Some(exceeded) = guard.record_tokens(tokens) {
let _ = tx.send(event).await;
let _ = tx
.send(crate::stream::StreamEvent::StopReason(format!(
"spend_limit: {exceeded}"
)))
.await;
let _ = tx
.send(crate::stream::StreamEvent::Attribution(
crate::outcome::UnattributedReason::Cancelled,
))
.await;
let _ = tx
.send(crate::stream::StreamEvent::Error(
"stream cancelled because the spend limit was reached"
.to_string(),
))
.await;
return;
}
}
}
if matches!(event, crate::stream::StreamEvent::Done { .. }) {
saw_completed = true;
}
let terminal = matches!(event, crate::stream::StreamEvent::Error(_));
if tx.send(event).await.is_err() || terminal {
return;
}
}
}
}
if !buffer.is_empty() || !saw_completed {
let _ = tx
.send(crate::stream::StreamEvent::Error(
"managed inference stream ended without response.completed".to_string(),
))
.await;
}
});
rx
}
fn parslee_request_body(
user_id: &str,
input: Vec<serde_json::Value>,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
tool_choice: Option<&str>,
model: Option<&str>,
) -> serde_json::Value {
let mut body = serde_json::json!({
"userId": user_id,
"input": input,
"max_output_tokens": max_tokens,
"store": false,
"include": ["reasoning.encrypted_content"],
"usage_details": true,
});
if let Some(tools) = tools {
if !tools.is_empty() {
body["tools"] = serde_json::json!(tools);
if let Some(choice) = tool_choice.map(str::trim).filter(|c| !c.is_empty()) {
body["tool_choice"] = match choice {
"auto" => serde_json::json!("auto"),
"required" | "any" => serde_json::json!("required"),
"none" => serde_json::json!("none"),
name => serde_json::json!({ "type": "function", "name": name }),
};
}
}
}
if let Some(model) = model.filter(|model| model.starts_with("parslee/openrouter/")) {
body["model"] = serde_json::json!(model);
}
body
}
fn parslee_input_items(
system: Option<&str>,
messages: &[serde_json::Value],
) -> Vec<serde_json::Value> {
responses_input_items(system, messages, UserParts::Drop)
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum UserParts {
Drop,
Keep,
}
fn responses_input_items(
system: Option<&str>,
messages: &[serde_json::Value],
user_parts: UserParts,
) -> Vec<serde_json::Value> {
let mut input: Vec<serde_json::Value> = Vec::new();
if let Some(sys) = system {
if !sys.trim().is_empty() {
input.push(serde_json::json!({ "role": "system", "content": sys }));
}
}
for m in messages {
if m.get("role").is_none() && m.get("type").is_some() {
input.push(m.clone());
continue;
}
let role = m.get("role").and_then(|v| v.as_str()).unwrap_or("user");
match role {
"assistant" => {
if let Some(tcs) = m.get("tool_calls").and_then(|v| v.as_array()) {
for tc in tcs {
let call_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or("");
let func = tc.get("function");
let name = func
.and_then(|f| f.get("name"))
.and_then(|v| v.as_str())
.unwrap_or("");
let args = func.and_then(|f| f.get("arguments")).map_or_else(
|| "{}".to_string(),
|a| {
a.as_str()
.map(str::to_string)
.unwrap_or_else(|| a.to_string())
},
);
input.push(serde_json::json!({
"type": "function_call",
"call_id": call_id,
"name": name,
"arguments": args,
}));
}
}
let text = m.get("content").and_then(|v| v.as_str()).unwrap_or("");
if !text.is_empty() {
input.push(serde_json::json!({ "role": "assistant", "content": text }));
}
}
"tool" => {
let call_id = m.get("tool_call_id").and_then(|v| v.as_str()).unwrap_or("");
let output = m.get("content").and_then(|v| v.as_str()).unwrap_or("");
input.push(serde_json::json!({
"type": "function_call_output",
"call_id": call_id,
"output": output,
}));
}
_ => {
if user_parts == UserParts::Keep {
if let Some(parts) = m.get("content").and_then(|v| v.as_array()) {
if !parts.is_empty() {
input.push(serde_json::json!({ "role": role, "content": parts }));
}
continue;
}
}
let content = m.get("content").and_then(|v| v.as_str()).unwrap_or("");
if !content.is_empty() {
input.push(serde_json::json!({ "role": role, "content": content }));
}
}
}
}
if tracing::enabled!(tracing::Level::DEBUG) {
let fc = input
.iter()
.filter(|i| i["type"] == "function_call")
.count();
let fco = input
.iter()
.filter(|i| i["type"] == "function_call_output")
.count();
tracing::debug!(
target = "car::parslee_input",
messages_in = messages.len(),
items_out = input.len(),
function_calls = fc,
function_call_outputs = fco,
"built parslee Responses input"
);
}
input
}
fn truncate_prompt_to_fit(
prompt: &str,
context: Option<&str>,
tools_json: Option<&[serde_json::Value]>,
max_tokens: usize,
media_tokens: usize,
context_window: usize,
) -> String {
let context_tokens = context.map(estimate_tokens).unwrap_or(0);
let tools_tokens = tools_json
.map(|t| estimate_tokens(&serde_json::to_string(t).unwrap_or_default()))
.unwrap_or(0);
let overhead = 100; let reserved = context_tokens + tools_tokens + media_tokens + max_tokens + overhead;
let available = context_window.saturating_sub(reserved);
let prompt_tokens = estimate_tokens(prompt);
if prompt_tokens <= available {
return prompt.to_string();
}
if prompt.trim().is_empty() {
let request_tokens =
crate::media_tokens::request_prompt_tokens(prompt, context, tools_json, None, None)
+ media_tokens;
tracing::warn!(
request_tokens = request_tokens,
history_and_media_tokens = media_tokens,
tools_tokens = tools_tokens,
context_tokens = context_tokens,
max_tokens = max_tokens,
context_window = context_window,
"message history exceeds the context window; nothing to truncate on this path \
— the caller's history compaction must trim it"
);
return prompt.to_string();
}
tracing::warn!(
prompt_tokens = prompt_tokens,
available_tokens = available,
context_window = context_window,
"truncating prompt to fit context window (prefer compaction via car-memgine)"
);
let chars_to_keep = available * 4;
if chars_to_keep >= prompt.len() {
return prompt.to_string();
}
let start = prompt.len().saturating_sub(chars_to_keep);
let safe_start = prompt.ceil_char_boundary(start);
let truncated = &prompt[safe_start..];
let break_point = truncated
.find('\n')
.or_else(|| truncated.find(' '))
.unwrap_or(0);
format!(
"[...truncated...]\n{}",
truncated[break_point..].trim_start()
)
}
impl RemoteBackend {
pub fn new() -> Self {
let (client, tls_degradation) = Self::build_http_client(RedirectRule::Follow);
Self {
client,
tls_degradation,
key_pool: std::sync::Arc::new(KeyPool::new()),
work_context: None,
context_client: std::sync::Arc::new(OnceLock::new()),
codex_client: std::sync::Arc::new(OnceLock::new()),
spend_gate: std::sync::Arc::new(std::sync::RwLock::new(None)),
}
}
pub(crate) fn with_spend_gate(
mut self,
gate: std::sync::Arc<std::sync::RwLock<Option<crate::spend::InstalledGate>>>,
) -> Self {
self.spend_gate = gate;
self
}
fn build_http_client(redirects: RedirectRule) -> (Client, Option<Degradation>) {
crate::tls_client::build_client_with_degradation(&crate::tls_client::REMOTE_BACKEND, || {
let builder = Client::builder();
let builder = match redirects.policy() {
Some(policy) => builder.redirect(policy),
None => builder,
};
builder
.timeout(std::time::Duration::from_secs(300))
.connect_timeout(std::time::Duration::from_secs(10))
})
}
pub fn with_work_context(&self, context: Option<car_auth::context::CredentialContext>) -> Self {
let (client, tls_degradation) = if context.is_some() {
let (client, degradation) = self
.context_client
.get_or_init(|| Self::build_http_client(RedirectRule::Refuse));
(client.clone(), degradation.clone())
} else {
(self.client.clone(), self.tls_degradation.clone())
};
Self {
client,
tls_degradation,
key_pool: self.key_pool.clone(),
work_context: context,
context_client: self.context_client.clone(),
codex_client: self.codex_client.clone(),
spend_gate: self.spend_gate.clone(),
}
}
pub(super) fn codex_client(&self) -> Client {
self.codex_client
.get_or_init(|| Self::build_http_client(RedirectRule::SameOrigin))
.0
.clone()
}
async fn refresh_bound_credentials(&self, rejected: &str) -> car_auth::ReactiveRefreshResult {
let Some(context) = &self.work_context else {
return car_auth::refresh_credential_after_rejection(rejected).await;
};
use car_auth::context::ContextCredentialError as E;
match car_auth::context::refresh_context_credential(context, rejected).await {
Ok(credential) => car_auth::ReactiveRefreshResult::Refreshed(credential),
Err(E::AccountUnavailable) => car_auth::ReactiveRefreshResult::SignedOut,
Err(E::ReauthenticationRequired) => {
car_auth::ReactiveRefreshResult::ReauthenticationRequired { expires_at: 0 }
}
Err(error) => {
car_auth::ReactiveRefreshResult::StoreError(car_auth::CredentialReadError {
kind: if error == E::OrganizationDenied {
car_auth::CredentialReadFailureKind::Denied
} else {
car_auth::CredentialReadFailureKind::Unreadable
},
message: error.to_string(),
})
}
}
}
fn refreshed_bound_authority(
&self,
result: car_auth::ReactiveRefreshResult,
model: &str,
rejected: InferenceError,
) -> Result<(String, String), InferenceError> {
match result {
car_auth::ReactiveRefreshResult::StoreError(error) if self.work_context.is_some() => {
Err(InferenceError::InferenceFailed(format!(
"Original work-context credential renewal failed: {}",
error.message
)))
}
result => refreshed_parslee_authority(result, model, rejected),
}
}
fn verify_bound_authority(&self, endpoint: &str) -> Result<(), InferenceError> {
if self
.work_context
.as_ref()
.is_some_and(|context| context.api_base != endpoint.trim_end_matches('/'))
{
return Err(InferenceError::InferenceFailed(
"Parslee authority differs from the immutable work context".into(),
));
}
Ok(())
}
fn verify_bound_org(&self, endpoint: &str, org: &str) -> Result<(), InferenceError> {
self.verify_bound_authority(endpoint)?;
if let Some(context) = &self.work_context {
if context.organization_id.as_deref() != Some(org) {
return Err(InferenceError::InferenceFailed(
"Parslee organization lookup differs from the immutable work context".into(),
));
}
}
Ok(())
}
pub(crate) fn request_error(&self, context: &str, err: &reqwest::Error) -> InferenceError {
InferenceError::InferenceFailed(format!(
"{context}: {}{}",
error_chain(err),
self.tls_degradation_note(err)
))
}
fn transport_attempt_error(
&self,
context: &str,
err: &reqwest::Error,
) -> TransportAttemptError {
TransportAttemptError::from_reqwest_error(self.request_error(context, err), err)
}
pub(crate) async fn send_retrying_connect<F, Fut>(
&self,
model: &str,
url: &str,
context: &str,
attempt_bound: Option<(std::time::Duration, &str)>,
attempt: &mut u32,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
send: F,
) -> Result<reqwest::Response, InferenceError>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<reqwest::Response, reqwest::Error>>,
{
self.send_retrying_connect_noting(
model,
url,
context,
attempt_bound,
attempt,
retry_observer,
send,
)
.await
.map_err(|(error, _sent)| error)
}
pub(crate) async fn send_retrying_connect_noting<F, Fut>(
&self,
model: &str,
url: &str,
context: &str,
attempt_bound: Option<(std::time::Duration, &str)>,
attempt: &mut u32,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
mut send: F,
) -> Result<reqwest::Response, (InferenceError, bool)>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<reqwest::Response, reqwest::Error>>,
{
let call_start = std::time::Instant::now();
loop {
let sent = match attempt_bound {
Some((bound, message)) => tokio::time::timeout(bound, send())
.await
.map_err(|_| (InferenceError::InferenceFailed(message.into()), true))?,
None => send().await,
};
let failure = match sent {
Ok(response) => return Ok(response),
Err(error) => self.transport_attempt_error(context, &error),
};
if !failure.connect {
let written = failure.retryable;
return Err((failure.error, written));
}
if *attempt >= REMOTE_MAX_ATTEMPTS {
return Err((
exhausted_transport_error(failure, url, *attempt, call_start.elapsed()),
false,
));
}
let backoff =
std::time::Duration::from_secs(REMOTE_BACKOFF_BASE_SECS * u64::from(*attempt));
deadline_permits_retry(backoff).map_err(|error| (error, false))?;
retry_observer(InferenceRetryProgress {
model: model.to_owned(),
attempt: *attempt + 1,
reason: "transport",
backoff_ms: backoff.as_millis() as u64,
});
tokio::time::sleep(backoff).await;
*attempt += 1;
}
}
fn tls_degradation_note(&self, err: &reqwest::Error) -> String {
match &self.tls_degradation {
Some(degradation) if is_trust_related(err.is_connect(), err.is_timeout()) => {
format!(" [{degradation}]")
}
_ => String::new(),
}
}
pub async fn register_model_keys(&self, schema: &ModelSchema) {
if let ModelSource::RemoteApi { ref endpoint, .. } = schema.source {
let env_vars = schema.all_api_key_envs();
if !env_vars.is_empty() {
self.key_pool.register_endpoint(endpoint, env_vars).await;
}
}
}
#[cfg(test)]
async fn execute_request(
&self,
schema: &ModelSchema,
req: crate::protocol::ApiRequest,
) -> Result<crate::protocol::ApiResponse, InferenceError> {
let mut ignore_retry = |_: InferenceRetryProgress| {};
self.execute_request_observed(schema, req, &mut ignore_retry)
.await
}
#[instrument(
name = "inference.remote_call",
skip_all,
fields(
model = %schema.name,
provider = %schema.provider,
gen_ai.operation.name = "chat",
gen_ai.request.model = %schema.name,
gen_ai.provider.name = tracing::field::Empty,
gen_ai.system = tracing::field::Empty,
),
)]
async fn execute_request_observed(
&self,
schema: &ModelSchema,
req: crate::protocol::ApiRequest,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
) -> Result<crate::protocol::ApiResponse, InferenceError> {
let (endpoint, protocol, lease) = self.resolve_request_authority(schema).await?;
let settlement = KeySettlement::new(self.key_pool.clone(), &endpoint, &lease.env_var);
{
let span = tracing::Span::current();
let provider_name = protocol.genai_provider_name();
span.record("gen_ai.provider.name", provider_name);
span.record("gen_ai.system", provider_name);
}
if codex::is_codex_subscription(schema) {
if req.response_format.is_some() {
return Err(InferenceError::UnsupportedMode {
mode: "structured-output",
backend: "openai-codex",
reason: "the ChatGPT Codex backend request path does not send a response_format yet; route structured-output requests to another model",
});
}
return self
.codex_request(schema, &endpoint, &req, retry_observer)
.await;
}
if let ModelSource::Proprietary { protocol, .. } = &schema.source {
if protocol.wire == crate::schema::ProprietaryWire::SystemOne {
return Err(InferenceError::UnsupportedMode {
mode: "generate",
backend: "parslee-system-one",
reason: "a System One model returns typed decisions, not text; use classify",
});
}
}
if matches!(schema.source, ModelSource::Proprietary { .. }) {
if req.response_format.is_some() {
return Err(InferenceError::UnsupportedMode {
mode: "structured-output",
backend: "parslee-inference",
reason: "the Parslee inference endpoint does not accept a \
response_format / JSON schema yet — route structured-output \
requests to a model that supports it",
});
}
return self
.execute_parslee_request_with_refresh(
&endpoint,
&lease.api_key,
&req,
retry_observer,
|rejected_bearer| async move {
self.refresh_bound_credentials(&rejected_bearer).await
},
)
.await;
}
if matches!(protocol, ApiProtocol::Bedrock) {
return self
.bedrock_converse_request(&endpoint, &req, retry_observer)
.await;
}
let handler = crate::protocol::handler_for(protocol);
let start = std::time::Instant::now();
let api_version = match &schema.source {
ModelSource::RemoteApi { api_version, .. } => api_version.clone(),
_ => None,
};
let url = if matches!(protocol, ApiProtocol::Google) {
crate::protocol::google_url(&endpoint, &req.model, &lease.api_key)
} else if matches!(protocol, ApiProtocol::VertexAi) {
crate::protocol::vertex_url(&endpoint, &req.model)
} else if matches!(protocol, ApiProtocol::AzureOpenAi) {
let version = api_version.as_deref().unwrap_or("2024-10-21");
format!(
"{}/openai/deployments/{}/chat/completions?api-version={}",
endpoint.trim_end_matches('/'),
req.model,
version
)
} else {
format_endpoint(&endpoint, chat_path_for(schema, handler.as_ref()))
};
let headers = handler.auth_headers(&lease.api_key);
let body = handler.build_request_body(&req);
crate::worker_cache::enforce_cached_prefix(
crate::worker_cache::cached_parts(protocol),
&body,
)?;
debug!(url = %url, model = %req.model, "protocol handler request");
let mut attempt = 0u32;
let call_start = std::time::Instant::now();
let phase_default = unarmed_phase_default(schema, req.max_tokens);
let resp_text = loop {
attempt += 1;
let send_outcome: Result<(reqwest::StatusCode, String), TransportAttemptError> =
async {
let mut builder = self
.client
.post(&url)
.timeout(std::time::Duration::from_secs(REMOTE_ABSOLUTE_CEILING_SECS));
for (name, value) in &headers {
builder = builder.header(name.as_str(), value.as_str());
}
let phase_start = std::time::Instant::now();
let resp =
match phase_bounded(builder.json(&body).send(), phase_start, phase_default)
.await
{
Ok(r) => {
r.map_err(|e| self.transport_attempt_error("HTTP error", &e))?
}
Err(cut) => {
return Err(
match phase_cut_error(
cut,
call_start,
"provider request",
phase_default,
) {
Ok(t) => t,
Err(deadline) => TransportAttemptError {
error: deadline,
retryable: false,
connect: false,
unbilled: false,
},
},
)
}
};
let status = resp.status();
let phase_start = std::time::Instant::now();
let txt = match phase_bounded(resp.text(), phase_start, phase_default).await {
Ok(r) => r.map_err(|e| {
unread_success_body(
status,
self.transport_attempt_error("read body", &e),
)
})?,
Err(cut) => {
return Err(
match phase_cut_error(
cut,
call_start,
"response body read",
phase_default,
) {
Ok(t) => unread_success_body(status, t),
Err(deadline) => TransportAttemptError {
error: deadline,
retryable: false,
connect: false,
unbilled: false,
},
},
)
}
};
Ok((status, txt))
}
.await;
match send_outcome {
Ok((status, txt)) if status.is_success() => break txt,
Ok((status, txt)) => {
let is_rl = txt.contains("429") || txt.contains("RESOURCE_EXHAUSTED");
let err_msg = format!("API returned {status}: {txt}");
let transient = is_transient_http_status(status.as_u16());
if attempt < REMOTE_MAX_ATTEMPTS && transient {
self.key_pool
.record_failed_attempt(&endpoint, &lease.env_var, is_rl)
.await;
let backoff = std::time::Duration::from_secs(
REMOTE_BACKOFF_BASE_SECS * attempt as u64,
);
deadline_permits_retry(backoff)?;
retry_observer(InferenceRetryProgress {
model: schema.id.clone(),
attempt: attempt + 1,
reason: "http_status",
backoff_ms: backoff.as_millis() as u64,
});
tokio::time::sleep(backoff).await;
continue;
}
settlement.fail(is_rl).await;
if is_account_http_status(status.as_u16()) {
return Err(InferenceError::ProviderAccount {
provider: schema.provider.clone(),
status: status.as_u16(),
message: account_error_message(protocol, status.as_u16()),
});
}
if matches!(protocol, ApiProtocol::OpenRouter) {
return Err(if transient {
InferenceError::Transient {
status: Some(status.as_u16()),
message: format!(
"OpenRouter upstream failed after {REMOTE_MAX_ATTEMPTS} attempts"
),
}
} else {
openrouter_http_error(status, &txt, &req.model)
});
}
return Err(if transient {
InferenceError::Transient {
status: Some(status.as_u16()),
message: err_msg,
}
} else {
InferenceError::InferenceFailed(err_msg)
});
}
Err(e) => {
let transient = e.retryable;
if attempt < REMOTE_MAX_ATTEMPTS && transient && e.unbilled {
let backoff = std::time::Duration::from_secs(
REMOTE_BACKOFF_BASE_SECS * attempt as u64,
);
deadline_permits_retry(backoff)?;
retry_observer(InferenceRetryProgress {
model: schema.id.clone(),
attempt: attempt + 1,
reason: "transport",
backoff_ms: backoff.as_millis() as u64,
});
tokio::time::sleep(backoff).await;
continue;
}
if transient && !e.unbilled {
crate::spend::note_sent_unread();
}
return Err(exhausted_transport_error(
e,
&url,
attempt,
call_start.elapsed(),
));
}
}
};
let latency_ms = start.elapsed().as_millis() as u64;
let parsed = if req.stream && matches!(protocol, ApiProtocol::OpenAiCompat) {
crate::stream::fold_openai_chat_sse(&resp_text)
.and_then(|folded| handler.parse_response(&folded))
} else {
handler.parse_response(&resp_text)
};
match &parsed {
Ok(response) => {
let (input_tokens, output_tokens) = key_pool_token_counts(
request_input_estimate(&req),
response.usage.as_ref(),
&response.text,
&response.tool_calls,
);
settlement
.succeed(latency_ms, input_tokens, output_tokens)
.await;
}
Err(InferenceError::ContentRefused { .. }) => {
let (input_tokens, _) =
key_pool_token_counts(request_input_estimate(&req), None, "", &[]);
settlement.succeed(latency_ms, input_tokens, 0).await;
}
Err(_) => settlement.fail(false).await,
}
if matches!(protocol, ApiProtocol::OpenRouter) {
crate::late_cost::note(
crate::late_cost::current_probe().as_ref(),
openrouter_generation_id(&resp_text).and_then(|generation| {
crate::late_cost::LateCost::openrouter(
self.client.clone(),
&endpoint,
&generation,
headers.clone(),
&schema.id,
&schema.name,
)
}),
);
}
let mut response = parsed?;
if let Some(ref mut usage) = response.usage {
usage.context_window = schema.context_length as u64;
}
Ok(response)
}
async fn bedrock_converse_request(
&self,
region: &str,
req: &crate::protocol::ApiRequest,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
) -> Result<crate::protocol::ApiResponse, InferenceError> {
let creds = crate::aws_sigv4::AwsCredentials::from_env().ok_or_else(|| {
InferenceError::InferenceFailed(
"Bedrock requires AWS credentials: set AWS_ACCESS_KEY_ID and \
AWS_SECRET_ACCESS_KEY (plus AWS_SESSION_TOKEN for temporary creds)"
.to_string(),
)
})?;
let handler = crate::protocol::BedrockHandler;
let body = handler.build_request_body(req);
crate::worker_cache::enforce_cached_prefix(
crate::worker_cache::cached_parts(ApiProtocol::Bedrock),
&body,
)?;
let body_bytes = serde_json::to_vec(&body)
.map_err(|e| InferenceError::InferenceFailed(format!("serialize Bedrock body: {e}")))?;
let BedrockEndpoint {
base_url,
host,
path_prefix,
} = bedrock_runtime_endpoint(region)?;
let encoded_model = crate::aws_sigv4::uri_encode_segment(&req.model);
let request_path = format!("/model/{encoded_model}/converse");
let url = format!("{base_url}{request_path}");
let canonical_path = format!("{path_prefix}{request_path}");
let base_headers = vec![
("host".to_string(), host.clone()),
("content-type".to_string(), "application/json".to_string()),
];
let signed_request = || {
let now = chrono::Utc::now();
let amz_date = now.format("%Y%m%dT%H%M%SZ").to_string();
let date_stamp = now.format("%Y%m%d").to_string();
let signed = crate::aws_sigv4::signed_headers(
&creds,
region,
"bedrock",
"POST",
&canonical_path,
"",
&base_headers,
&body_bytes,
&amz_date,
&date_stamp,
);
let mut builder = self.client.post(&url);
for (k, v) in &signed {
if k != "host" {
builder = builder.header(k, v);
}
}
builder.body(body_bytes.clone()).send()
};
let resp = self
.send_retrying_connect_noting(
&req.model,
&url,
"HTTP error",
Some((
std::time::Duration::from_secs(300),
"Bedrock request timed out after 300s",
)),
&mut 1,
retry_observer,
signed_request,
)
.await
.map_err(|(error, written)| {
if written {
crate::spend::note_sent_unread();
}
error
})?;
let status = resp.status();
let txt = resp.text().await.map_err(|e| {
if status.is_success() {
crate::spend::note_sent_unread();
}
InferenceError::InferenceFailed(format!("read Bedrock body: {}", error_chain(&e)))
})?;
if !status.is_success() {
return Err(InferenceError::InferenceFailed(format!(
"Bedrock returned {status}: {txt}"
)));
}
handler.parse_response(&txt)
}
async fn parslee_identity(
&self,
endpoint: &str,
bearer: &str,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
) -> Result<(String, String), InferenceError> {
self.verify_bound_authority(endpoint)?;
let cache = PARSLEE_IDENTITY.get_or_init(|| tokio::sync::Mutex::new(HashMap::new()));
let base = endpoint.trim_end_matches('/');
let cache_key = (base.to_string(), bearer.to_string());
if let Some(v) = cache.lock().await.get(&cache_key) {
self.verify_bound_org(endpoint, &v.0)?;
return Ok(v.clone());
}
let org_url = format!("{base}/api/v1/organizations/me");
let org_resp = self
.send_retrying_connect(
"parslee",
&org_url,
"Parslee org lookup",
None,
&mut 1,
retry_observer,
|| self.client.get(&org_url).bearer_auth(bearer).send(),
)
.await?;
if !org_resp.status().is_success() {
let s = org_resp.status();
if (s == reqwest::StatusCode::UNAUTHORIZED || s == reqwest::StatusCode::FORBIDDEN)
&& self.work_context.is_none()
{
crate::parslee_credential::note_credential_rejected();
}
let b = org_resp.text().await.unwrap_or_default();
if s == reqwest::StatusCode::UNAUTHORIZED {
return Err(InferenceError::CredentialUnavailable {
provider: "parslee".to_string(),
model: String::new(),
reason: crate::CredentialFailure::Expired { expires_at: 0 },
detail: format!(
"Parslee org lookup failed: HTTP {s}. Your Parslee sign-in is no longer accepted — run `car auth login` to sign in again. {b}"
)
.trim_end()
.to_string(),
});
}
if s == reqwest::StatusCode::FORBIDDEN {
return Err(InferenceError::InferenceFailed(
format!(
"Parslee org lookup failed: HTTP {s}. Parslee denied this account access — \
signing in again does not change access; ask an organization admin. {b}"
)
.trim_end()
.to_string(),
));
}
return Err(InferenceError::InferenceFailed(format!(
"Parslee org lookup failed: HTTP {s}: {b}"
)));
}
if self.work_context.is_none() {
crate::parslee_credential::clear_credential_rejected();
}
let org_json: serde_json::Value = org_resp.json().await.map_err(|e| {
InferenceError::InferenceFailed(format!("parse Parslee org: {}", error_chain(&e)))
})?;
let org_id = org_json
.get("organizationId")
.or_else(|| org_json.get("OrganizationId"))
.and_then(|v| v.as_str())
.ok_or_else(|| {
InferenceError::WorkspaceRequired {
provider: "Parslee".to_string(),
detail: "finish setting up at https://parslee.ai, then try again".to_string(),
}
})?
.to_string();
self.verify_bound_org(endpoint, &org_id)?;
let sess_url = format!("{base}/connect/session");
let session = self
.send_retrying_connect(
"parslee",
&sess_url,
"Parslee session lookup",
None,
&mut 1,
retry_observer,
|| self.client.get(&sess_url).bearer_auth(bearer).send(),
)
.await;
let answered = match session {
Ok(r) if r.status().is_success() => r.json::<serde_json::Value>().await.ok().map(|j| {
j.get("account")
.and_then(|a| a.get("email"))
.and_then(|v| v.as_str())
.map(String::from)
}),
Ok(_) => Some(None),
Err(_) => None,
};
let user_id = answered
.clone()
.flatten()
.unwrap_or_else(|| "car".to_string());
let pair = (org_id, user_id);
if answered.is_some() {
cache.lock().await.insert(cache_key, pair.clone());
}
Ok(pair)
}
pub async fn system_one(
&self,
schema: &ModelSchema,
body: &str,
) -> Result<serde_json::Value, InferenceError> {
if !crate::typesafe::is_direct(schema) {
return self.parslee_system_one(schema, body).await;
}
let bearer = crate::typesafe::stored_key(schema)?;
let ModelSource::Proprietary {
endpoint, protocol, ..
} = &schema.source
else {
unreachable!()
};
if let Some(foreign) = crate::credential_hosts::foreign_reserved_credential(schema) {
return Err(InferenceError::InferenceFailed(format!(
"{}: stored credential {} cannot be sent to this host",
schema.id, foreign.credential
)));
}
let base = reqwest::Url::parse(endpoint).map_err(|_| {
InferenceError::InferenceFailed(
"TypeSafe direct endpoint must be an absolute HTTP(S) URL".into(),
)
})?;
if !matches!(base.scheme(), "http" | "https")
|| base.host_str().is_none()
|| !base.username().is_empty()
|| base.password().is_some()
|| base.query().is_some()
|| base.fragment().is_some()
|| !protocol.chat_path.starts_with('/')
|| protocol.chat_path.contains(['?', '#', '{', '}'])
{
return Err(InferenceError::InferenceFailed("TypeSafe direct needs an HTTP(S) endpoint and an absolute API path without query, fragment or org placeholder".into()));
}
let url = format!("{}{}", endpoint.trim_end_matches('/'), protocol.chat_path);
let mut headers = reqwest::header::HeaderMap::new();
for (name, value) in &protocol.extra_headers {
let header =
reqwest::header::HeaderName::from_bytes(name.as_bytes()).map_err(|_| {
InferenceError::InferenceFailed(
"TypeSafe direct has an invalid extra header".into(),
)
})?;
if header == reqwest::header::HOST || header == reqwest::header::AUTHORIZATION {
return Err(InferenceError::InferenceFailed(
"TypeSafe direct cannot override Host or Authorization".into(),
));
}
let value = reqwest::header::HeaderValue::from_str(value).map_err(|_| {
InferenceError::InferenceFailed(
"TypeSafe direct has an invalid extra header value".into(),
)
})?;
headers.insert(header, value);
}
tracing::info!(provider = "typesafe", provider_row = %schema.id, key_id = %crate::typesafe::key_name(schema)?, "direct System One dispatch");
let (client, _) = self
.context_client
.get_or_init(|| Self::build_http_client(RedirectRule::Refuse));
let request = client
.post(&url)
.headers(headers)
.bearer_auth(&bearer)
.header("content-type", &protocol.content_type)
.body(body.to_owned());
let started = std::time::Instant::now();
let budget = std::time::Duration::from_secs(REMOTE_PHASE_DEFAULT_SECS);
let phase_error = |cut| match cut {
PhaseCut::Deadline { applied_ms } => InferenceError::DeadlineExceeded {
applied_ms,
elapsed_ms: started.elapsed().as_millis() as u64,
last_error: "TypeSafe direct exceeded the caller deadline".into(),
},
PhaseCut::Default => {
InferenceError::InferenceFailed("TypeSafe direct timed out".into())
}
};
let response = phase_bounded(request.send(), started, budget)
.await
.map_err(phase_error)?
.map_err(|_| {
InferenceError::InferenceFailed("TypeSafe direct transport failed".into())
})?;
let status = response.status();
if !status.is_success() {
return Err(InferenceError::InferenceFailed(format!(
"TypeSafe direct {} failed: HTTP {status}",
schema.id
)));
}
phase_bounded(response.json(), std::time::Instant::now(), budget)
.await
.map_err(phase_error)?
.map_err(|_| {
InferenceError::InferenceFailed("TypeSafe direct returned invalid JSON".into())
})
}
pub async fn parslee_system_one(
&self,
schema: &ModelSchema,
body: &str,
) -> Result<serde_json::Value, InferenceError> {
let (endpoint, _, lease) = self.resolve_request_authority(schema).await?;
let key = format!("{:032x}", rand::random::<u128>());
match self
.system_one_request(&endpoint, &lease.api_key, schema, body, &key)
.await
{
Ok(value) => Ok(value),
Err(error) if is_auth_rejection(&error) => {
let (api_base, token) = self.refreshed_bound_authority(
self.refresh_bound_credentials(&lease.api_key).await,
&schema.name,
error,
)?;
self.system_one_request(&api_base, &token, schema, body, &key)
.await
}
Err(error) => Err(error),
}
}
async fn system_one_request(
&self,
endpoint: &str,
bearer: &str,
schema: &ModelSchema,
body: &str,
key: &str,
) -> Result<serde_json::Value, InferenceError> {
let (org_id, _) = self.parslee_identity(endpoint, bearer, &mut |_| {}).await?;
let path = match &schema.source {
ModelSource::Proprietary { protocol, .. } if protocol.chat_path.contains("{orgId}") => {
protocol.chat_path.replace("{orgId}", &org_id)
}
_ => {
return Err(InferenceError::InferenceFailed(format!(
"{}: a System One row needs a protocol path containing {{orgId}}",
schema.id
)))
}
};
let url = format!("{}{}", endpoint.trim_end_matches('/'), path);
let resp = self
.send_retrying_connect(
&schema.id,
&url,
"Parslee System One HTTP error",
Some((
std::time::Duration::from_secs(150),
"Parslee System One request timed out (150s)",
)),
&mut 1,
&mut |_| {},
|| {
self.client
.post(&url)
.bearer_auth(bearer)
.header("content-type", "application/json")
.header("Idempotency-Key", key)
.body(body.to_string())
.send()
},
)
.await?;
let status = resp.status();
let raw = resp.text().await.map_err(|e| {
InferenceError::InferenceFailed(format!(
"Parslee System One read body: {}",
error_chain(&e)
))
})?;
if !status.is_success() {
if let Some(detail) = gateway_unconfigured_detail(status, &raw) {
return Err(gateway_unconfigured_error(
status,
detail,
&schema.name,
self.work_context.is_none(),
));
}
let shown: String = raw.chars().take(400).collect();
if matches!(status.as_u16(), 429 | 502 | 503 | 504) {
return Err(InferenceError::Transient {
status: Some(status.as_u16()),
message: format!("Parslee System One: {shown}"),
});
}
return Err(InferenceError::InferenceFailed(format!(
"Parslee System One failed: HTTP {status}: {shown}"
)));
}
serde_json::from_str(&raw).map_err(|e| {
InferenceError::InferenceFailed(format!(
"Parslee System One returned invalid JSON: {e}"
))
})
}
async fn parslee_assistant_request(
&self,
endpoint: &str,
bearer: &str,
req: &crate::protocol::ApiRequest,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
) -> Result<crate::protocol::ApiResponse, InferenceError> {
let (org_id, user_id) = self
.parslee_identity(endpoint, bearer, retry_observer)
.await?;
let input = parslee_input_items(req.system.as_deref(), &req.messages);
if input.is_empty() {
return Err(InferenceError::InferenceFailed(
"Parslee inference: empty prompt".to_string(),
));
}
let url = format!(
"{}/api/v1/orgs/{}/inference/responses",
endpoint.trim_end_matches('/'),
org_id
);
let _ = req.temperature;
let body = parslee_request_body(
&user_id,
input,
req.max_tokens,
req.tools.as_deref(),
req.tool_choice.as_deref(),
Some(&req.model),
);
crate::worker_cache::enforce_cached_prefix(
crate::worker_cache::PARSLEE_CACHED_PARTS,
&body,
)?;
let resp = self
.send_retrying_connect(
&req.model,
&url,
"Parslee chat HTTP error",
Some((
std::time::Duration::from_secs(300),
"Parslee chat request timed out after 300s",
)),
&mut 1,
retry_observer,
|| {
self.client
.post(&url)
.bearer_auth(bearer)
.header("content-type", "application/json")
.header("accept", "text/event-stream")
.json(&body)
.send()
},
)
.await?;
let status = resp.status();
let text_fut = resp.text();
let raw = tokio::time::timeout(std::time::Duration::from_secs(300), text_fut)
.await
.map_err(|_| {
InferenceError::InferenceFailed(
"Parslee chat body read timed out after 300s".to_string(),
)
})?
.map_err(|e| {
InferenceError::InferenceFailed(format!(
"Parslee chat read body: {}",
error_chain(&e)
))
})?;
if !status.is_success() {
if let Some(detail) = gateway_unconfigured_detail(status, &raw) {
return Err(gateway_unconfigured_error(
status,
detail,
&req.model,
self.work_context.is_none(),
));
}
return Err(InferenceError::InferenceFailed(format!(
"Parslee chat failed: HTTP {status}: {raw}"
)));
}
let mut acc = crate::stream::StreamAccumulator::default();
let mut stream_error = None;
let mut saw_completed = false;
for event in parse_parslee_responses_sse(&raw) {
if let crate::stream::StreamEvent::Error(message) = &event {
stream_error = Some(message.clone());
}
if matches!(event, crate::stream::StreamEvent::Done { .. }) {
saw_completed = true;
}
acc.push(&event);
}
let (text, tool_calls, usage, stop_reason, provider_output_items) =
acc.finish_with_provider_output_items();
if let Some(message) = stream_error {
if let Some((kind, code)) = content_refusal_tags(&message) {
return Err(InferenceError::ContentRefused {
provider: "parslee".to_string(),
kind,
code,
message,
});
}
return Err(InferenceError::InferenceFailed(message));
}
if !saw_completed {
return Err(InferenceError::InferenceFailed(
"managed inference stream ended without response.completed".to_string(),
));
}
if text.is_empty() && tool_calls.is_empty() {
let snippet: String = raw.trim().chars().take(600).collect();
return Err(InferenceError::InferenceFailed(format!(
"Parslee inference returned no content (HTTP {status}); raw response: {snippet}"
)));
}
Ok(crate::protocol::ApiResponse {
text,
tool_calls,
provider_output_items,
thinking: Vec::new(),
usage,
stop_reason,
})
}
async fn parslee_assistant_stream_request(
&self,
endpoint: &str,
bearer: &str,
req: &crate::protocol::ApiRequest,
spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
) -> Result<tokio::sync::mpsc::Receiver<crate::stream::StreamEvent>, InferenceError> {
let (org_id, user_id) = self.parslee_identity(endpoint, bearer, &mut |_| {}).await?;
let input = parslee_input_items(req.system.as_deref(), &req.messages);
if input.is_empty() {
return Err(InferenceError::InferenceFailed(
"Parslee inference: empty prompt".to_string(),
));
}
let url = format!(
"{}/api/v1/orgs/{}/inference/responses",
endpoint.trim_end_matches('/'),
org_id
);
let body = parslee_request_body(
&user_id,
input,
req.max_tokens,
req.tools.as_deref(),
req.tool_choice.as_deref(),
Some(&req.model),
);
crate::worker_cache::enforce_cached_prefix(
crate::worker_cache::PARSLEE_CACHED_PARTS,
&body,
)?;
crate::outcome::note_dispatch();
let resp = self
.send_retrying_connect(
&req.model,
&url,
"Parslee stream HTTP error",
None,
&mut 1,
&mut |_| {},
|| {
self.client
.post(&url)
.bearer_auth(bearer)
.header("content-type", "application/json")
.header("accept", "text/event-stream")
.json(&body)
.send()
},
)
.await?;
let status = resp.status();
if !status.is_success() {
let body = resp.text().await.unwrap_or_default();
if let Some(detail) = gateway_unconfigured_detail(status, &body) {
return Err(gateway_unconfigured_error(
status,
detail,
&req.model,
self.work_context.is_none(),
));
}
return Err(InferenceError::InferenceFailed(format!(
"Parslee stream failed: HTTP {status}: {body}"
)));
}
Ok(pump_responses_sse(resp, spend_guard, "Parslee"))
}
pub async fn generate(
&self,
schema: &ModelSchema,
prompt: &str,
context: Option<&str>,
temperature: f64,
max_tokens: usize,
images: Option<&[ContentBlock]>,
) -> Result<String, InferenceError> {
let resp = self
.generate_with_tools_multi(
schema,
prompt,
context,
temperature,
max_tokens,
None,
images,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
)
.await?;
Ok(resp.0)
}
pub async fn generate_with_tools(
&self,
schema: &ModelSchema,
prompt: &str,
context: Option<&str>,
temperature: f64,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
images: Option<&[ContentBlock]>,
) -> Result<(String, Vec<crate::tasks::generate::ToolCall>), InferenceError> {
let (text, calls, _thinking, _provider_items, _usage, _stop) = self
.generate_with_tools_multi(
schema,
prompt,
context,
temperature,
max_tokens,
tools,
images,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
)
.await?;
Ok((text, calls))
}
pub async fn generate_with_tools_multi(
&self,
schema: &ModelSchema,
prompt: &str,
context: Option<&str>,
temperature: f64,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
images: Option<&[ContentBlock]>,
messages: Option<&[crate::tasks::generate::Message]>,
tool_choice: Option<&str>,
parallel_tool_calls: Option<bool>,
budget_tokens: usize,
cache_control: bool,
cache_ttl: crate::tasks::generate::CacheTtl,
system_stable_prefix: Option<&str>,
response_format: Option<&crate::tasks::generate::ResponseFormat>,
) -> Result<
(
String,
Vec<crate::tasks::generate::ToolCall>,
Vec<crate::tasks::generate::ThinkingBlock>,
Vec<serde_json::Value>,
Option<crate::TokenUsage>,
Option<String>,
),
InferenceError,
> {
let mut ignore_retry = |_: InferenceRetryProgress| {};
self.generate_with_tools_multi_observed(
schema,
prompt,
context,
temperature,
max_tokens,
tools,
images,
messages,
tool_choice,
parallel_tool_calls,
budget_tokens,
cache_control,
cache_ttl,
system_stable_prefix,
response_format,
&crate::protocol::OutputControls::default(),
&mut ignore_retry,
)
.await
}
async fn continue_vllm_thinking(
&self,
schema: &ModelSchema,
split: &crate::vllm_thinking::ThinkingSplit,
response: &mut crate::protocol::ApiResponse,
tools: Option<&[serde_json::Value]>,
stop: &[String],
temperature: f64,
) -> Result<(), InferenceError> {
let first = response.usage.clone().unwrap_or_default();
let Some(next) = split.continuation(
&response.text,
&response.thinking,
&response.tool_calls,
first.completion_tokens,
) else {
return Ok(());
};
let (base, _) = extract_remote_endpoint(schema)?;
let url = format_endpoint(&base, "/v1/completions");
let mut body = serde_json::json!({
"model": request_model_name(schema),
"prompt": next.prompt,
"max_tokens": next.max_tokens,
});
if temperature >= 0.0 {
body["temperature"] = serde_json::json!(temperature);
}
if !stop.is_empty() {
body["stop"] = serde_json::json!(stop);
}
let call_start = std::time::Instant::now();
let phase_default = unarmed_phase_default(schema, next.max_tokens);
let send = self
.client
.post(&url)
.timeout(std::time::Duration::from_secs(REMOTE_ABSOLUTE_CEILING_SECS))
.json(&body)
.send();
let resp = match phase_bounded(send, std::time::Instant::now(), phase_default).await {
Ok(r) => r.map_err(|e| self.transport_attempt_error("HTTP error", &e).error)?,
Err(cut) => {
return Err(
match phase_cut_error(cut, call_start, "thinking continuation", phase_default) {
Ok(timeout) => timeout.error,
Err(deadline) => deadline,
},
)
}
};
let status = resp.status();
let text = match phase_bounded(resp.text(), std::time::Instant::now(), phase_default).await
{
Ok(r) => r.map_err(|e| self.transport_attempt_error("read body", &e).error)?,
Err(cut) => {
return Err(
match phase_cut_error(
cut,
call_start,
"thinking continuation body",
phase_default,
) {
Ok(timeout) => timeout.error,
Err(deadline) => deadline,
},
)
}
};
if !status.is_success() {
return Err(InferenceError::InferenceFailed(format!(
"vllm-mlx thinking continuation returned {status}: {text}"
)));
}
let parsed: serde_json::Value = serde_json::from_str(&text).map_err(|e| {
InferenceError::InferenceFailed(format!("parse thinking continuation: {e}"))
})?;
let choice = parsed.pointer("/choices/0").ok_or_else(|| {
InferenceError::InferenceFailed("thinking continuation returned no choice".into())
})?;
let continued = choice.get("text").and_then(|t| t.as_str()).unwrap_or("");
let count = |key: &str| {
parsed
.pointer(&format!("/usage/{key}"))
.and_then(serde_json::Value::as_u64)
.unwrap_or(0)
};
let (prompt_tokens, completion_tokens) =
(count("prompt_tokens"), count("completion_tokens"));
let expected =
first.prompt_tokens + first.cache_read_input_tokens + first.completion_tokens;
if prompt_tokens.abs_diff(expected) > 64 {
tracing::warn!(
model = %schema.id,
prompt_tokens,
expected,
"vllm-mlx thinking continuation prompt differs from the first request's; \
the chat template render may not match the server's",
);
}
let answer = format!("{}{}", next.answer_so_far, continued);
let (clean, calls) = crate::tasks::generate::parse_tool_calls_typed(&answer, tools);
response.text = clean;
response.tool_calls = calls;
if let Some(thought) = next.thought {
response.thinking = vec![crate::tasks::generate::ThinkingBlock {
text: thought,
signature: None,
redacted_data: None,
}];
}
let mut usage = first;
usage.prompt_tokens = prompt_tokens;
usage.cache_read_input_tokens = 0;
usage.cache_creation_input_tokens = 0;
usage.completion_tokens += completion_tokens;
usage.total_tokens = usage.prompt_tokens + usage.completion_tokens;
response.usage = Some(usage);
response.stop_reason = choice
.get("finish_reason")
.and_then(|v| v.as_str())
.map(str::to_string);
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub async fn generate_with_tools_multi_observed(
&self,
schema: &ModelSchema,
prompt: &str,
context: Option<&str>,
temperature: f64,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
images: Option<&[ContentBlock]>,
messages: Option<&[crate::tasks::generate::Message]>,
tool_choice: Option<&str>,
parallel_tool_calls: Option<bool>,
budget_tokens: usize,
cache_control: bool,
cache_ttl: crate::tasks::generate::CacheTtl,
system_stable_prefix: Option<&str>,
response_format: Option<&crate::tasks::generate::ResponseFormat>,
controls: &crate::protocol::OutputControls,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
) -> Result<
(
String,
Vec<crate::tasks::generate::ToolCall>,
Vec<crate::tasks::generate::ThinkingBlock>,
Vec<serde_json::Value>,
Option<crate::TokenUsage>,
Option<String>,
),
InferenceError,
> {
if crate::typesafe::is_direct(schema) {
if context.is_some()
|| tools.is_some()
|| images.is_some()
|| messages.is_some()
|| response_format.is_some()
|| !controls.stop.is_empty()
{
return Err(InferenceError::UnsupportedMode { mode: "chat", backend: "typesafe-system-one", reason: "TypeSafe accepts a System One JSON prompt with state and questions; chat context, tools, messages, stop and response formats are unsupported" });
}
let payload: serde_json::Value = serde_json::from_str(prompt).map_err(|_| {
InferenceError::InferenceFailed("TypeSafe inference prompt must be a JSON object containing state and questions".into())
})?;
let object = payload
.as_object()
.filter(|object| {
object.contains_key("state")
&& object
.get("questions")
.is_some_and(|q| q.as_object().is_some_and(|q| !q.is_empty()))
})
.ok_or_else(|| {
InferenceError::InferenceFailed(
"TypeSafe inference prompt needs state and non-empty questions".into(),
)
})?;
if object.contains_key("model") {
return Err(InferenceError::InferenceFailed(
"TypeSafe prompt must omit model; the provider row selects it".into(),
));
}
let trimmed = prompt.trim();
let body = format!(
"{},\"model\":{}}}",
&trimmed[..trimmed.len() - 1],
serde_json::to_string(&schema.name).unwrap()
);
let response = self.system_one(schema, &body).await?;
let count = |name: &str| {
response
.pointer(&format!("/usage/{name}"))
.and_then(|v| v.as_u64())
.unwrap_or(0)
};
let (input, output) = (count("input_tokens"), count("output_tokens"));
return Ok((
response.to_string(),
vec![],
vec![],
vec![],
Some(crate::TokenUsage {
prompt_tokens: input,
completion_tokens: output,
total_tokens: input.saturating_add(output),
..Default::default()
}),
None,
));
}
let (_, protocol) = extract_remote_endpoint(schema)?;
let handler = crate::protocol::handler_for(protocol);
if !handler.supports_video() {
let has_video_in_images =
images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_video));
let has_video_in_messages = messages.is_some_and(|msgs| {
msgs.iter().any(|msg| match msg {
crate::tasks::generate::Message::UserMultimodal { content } => {
content.iter().any(ContentBlock::is_video)
}
_ => false,
})
});
if has_video_in_images || has_video_in_messages {
return Err(InferenceError::UnsupportedMode {
mode: "video-content-block",
backend: handler.protocol_name(),
reason: "this remote protocol has no native video input path; route to \
a provider that implements ProtocolHandler::supports_video() (Gemini)",
});
}
}
if !handler.supports_audio() {
let has_audio_in_images =
images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_audio));
let has_audio_in_messages = messages.is_some_and(|msgs| {
msgs.iter().any(|msg| match msg {
crate::tasks::generate::Message::UserMultimodal { content } => {
content.iter().any(ContentBlock::is_audio)
}
_ => false,
})
});
if has_audio_in_images || has_audio_in_messages {
return Err(InferenceError::UnsupportedMode {
mode: "audio-content-block",
backend: handler.protocol_name(),
reason: "this remote protocol has no native audio input path; route to \
a provider that implements ProtocolHandler::supports_audio() (Gemini)",
});
}
}
if let Some(err) = response_format_rejection(handler.as_ref(), response_format) {
return Err(err);
}
let prompt = if schema.context_length > 0 {
let media_tokens =
crate::media_tokens::request_media_and_history_tokens(images, messages);
truncate_prompt_to_fit(
prompt,
context,
tools,
max_tokens,
media_tokens,
schema.context_length,
)
} else {
prompt.to_string()
};
let (api_messages, system) = if matches!(schema.source, ModelSource::Proprietary { .. }) {
crate::protocol::OpenAiResponsesHandler.build_messages(
messages.unwrap_or(&[]),
&prompt,
context,
images,
)
} else {
handler.build_messages(messages.unwrap_or(&[]), &prompt, context, images)
};
let api_tools = build_request_tools(schema, handler.as_ref(), tools);
let mut req = crate::protocol::ApiRequest {
model: request_model_name(schema),
messages: api_messages,
system,
system_stable_prefix: system_stable_prefix.map(str::to_string),
temperature,
max_tokens,
tools: api_tools,
tool_choice: tool_choice.map(str::to_string),
parallel_tool_calls,
stream: false,
budget_tokens,
cache_control,
cache_ttl,
response_format: response_format.cloned(),
stop: controls.stop.clone(),
reasoning_effort: controls.reasoning_effort,
enable_thinking: template_thinking(schema, controls),
template_reasoning_effort: template_effort(schema, controls),
};
let thinking_split = match &schema.source {
ModelSource::VllmMlx { model_name, .. }
if crate::vllm_pool::splits_think_tags(&schema.family) =>
{
crate::vllm_thinking::ThinkingSplit::plan(
std::path::Path::new(model_name),
&handler.build_request_body(&req),
max_tokens,
budget_tokens,
controls.thinking,
response_format.is_some(),
)
}
_ => None,
};
if let Some(split) = &thinking_split {
req.max_tokens = split.budget;
}
if matches!(schema.source, ModelSource::VllmMlx { .. })
&& crate::vllm_pool::folds_reasoning_from_stream(&schema.family)
{
req.stream = true;
}
let mut response = self
.execute_request_observed(schema, req, retry_observer)
.await?;
if let Some(split) = &thinking_split {
self.continue_vllm_thinking(
schema,
split,
&mut response,
tools,
&controls.stop,
temperature,
)
.await?;
}
let mut calls = response.tool_calls;
let thinking = response.thinking;
let provider_output_items = response.provider_output_items;
let usage = response.usage;
let (text, stop_reason) = crate::tasks::generate::apply_stop_backstop(
response.text,
response.stop_reason,
&controls.stop,
calls.len(),
);
if cache_control {
if let Some(u) = usage.as_ref() {
if u.cache_read_input_tokens == 0 && u.cache_creation_input_tokens == 0 {
tracing::debug!(
model = %request_model_name(schema),
prompt_tokens = u.prompt_tokens,
"prompt caching requested but nothing was cached (read=0, write=0) — \
prompt likely below the model's minimum cacheable length, or the \
cached prefix changed between requests",
);
}
}
}
if !text.is_empty() {
for call in &mut calls {
if call.name == "done" {
let result_val = call
.arguments
.get("result")
.and_then(|v| v.as_str())
.unwrap_or("");
if result_val.len() < 50 && text.len() > result_val.len() {
call.arguments.insert(
"result".to_string(),
serde_json::Value::String(text.clone()),
);
}
}
}
}
Ok((
text,
calls,
thinking,
provider_output_items,
usage,
stop_reason,
))
}
pub async fn embed(
&self,
schema: &ModelSchema,
texts: &[String],
) -> Result<Vec<Vec<f32>>, InferenceError> {
let (endpoint, protocol, lease) = self.resolve_request_authority(schema).await?;
let est_tokens: u64 = texts
.iter()
.map(|t| t.split_whitespace().count() as u64)
.sum();
let mut spend_ticket = self.spend_admit_embed(schema, est_tokens)?;
let start = std::time::Instant::now();
let result = match protocol {
ApiProtocol::OpenAiCompat => {
self.embed_openai(&endpoint, &lease.api_key, &schema.name, texts)
.await
}
_ => Err(InferenceError::InferenceFailed(format!(
"embedding not supported for {:?} protocol",
protocol
))),
};
if let (Some(ticket), Err(error)) = (spend_ticket.as_mut(), &result) {
ticket.ended_with(error);
}
let latency_ms = start.elapsed().as_millis() as u64;
match &result {
Ok(_) => {
self.key_pool
.report_success(&endpoint, &lease.env_var, latency_ms, est_tokens, 0)
.await;
}
Err(e) => {
let is_rl =
e.to_string().contains("429") || e.to_string().contains("RESOURCE_EXHAUSTED");
self.key_pool
.report_failure(&endpoint, &lease.env_var, is_rl)
.await;
}
}
result
}
async fn resolve_request_authority(
&self,
schema: &ModelSchema,
) -> Result<(String, ApiProtocol, KeyLease), InferenceError> {
let (endpoint, protocol) = extract_remote_endpoint(schema)?;
if let ModelSource::Proprietary {
provider,
auth: ProprietaryAuth::ChatGptSubscription {},
..
} = &schema.source
{
if !provider.eq_ignore_ascii_case(crate::schema::OPENAI_CODEX_PROVIDER)
|| !schema
.provider
.eq_ignore_ascii_case(crate::schema::OPENAI_CODEX_PROVIDER)
{
return Err(InferenceError::InferenceFailed(format!(
"ChatGPT subscription credentials are supported only for provider 'openai-codex'; refusing model {} from provider '{provider}'",
schema.id
)));
}
let requested = endpoint.trim().trim_end_matches('/');
if !requested.is_empty() && requested != codex::CODEX_BASE_URL {
return Err(InferenceError::InferenceFailed(format!(
"ChatGPT subscription credentials are sent only to {}; refusing model {} with endpoint '{endpoint}'",
codex::CODEX_BASE_URL,
schema.id
)));
}
return Ok((
codex::CODEX_BASE_URL.to_owned(),
protocol,
KeyLease {
api_key: String::new(),
env_var: String::new(),
},
));
}
if let ModelSource::Proprietary {
provider,
auth: ProprietaryAuth::OAuth2Pkce { .. },
..
} = &schema.source
{
if !provider.eq_ignore_ascii_case("parslee")
|| !schema.provider.eq_ignore_ascii_case("parslee")
{
return Err(InferenceError::InferenceFailed(format!(
"OAuth2 PKCE credentials are supported only for provider \
'parslee'; refusing model {} from proprietary provider \
'{provider}'",
schema.id
)));
}
let credential = if let Some(context) = &self.work_context {
if context.organization_id.is_none() {
return Err(InferenceError::InferenceFailed(
"Parslee managed inference requires an explicit organization context"
.into(),
));
}
car_auth::context::resolve_context_credential(context)
.await
.map_err(|error| work_context_credential_error(&schema.id, error))?
} else {
car_auth::resolve_credential(car_auth::CredentialReadMode::Use)
.await
.map_err(|error| parslee_credential_read_error(schema, provider, error))?
.ok_or_else(|| parslee_signed_out_error(schema, provider))?
};
warn_if_token_expires_soon(credential.expires_at);
return Ok((
credential.api_base,
protocol,
KeyLease {
api_key: credential.access_token,
env_var: String::new(),
},
));
}
let lease = self.lease_key(schema, &endpoint).await?;
Ok((endpoint, protocol, lease))
}
async fn execute_parslee_request_with_refresh<Refresh, RefreshFuture>(
&self,
endpoint: &str,
bearer: &str,
request: &crate::protocol::ApiRequest,
retry_observer: &mut (dyn FnMut(InferenceRetryProgress) + Send),
refresh: Refresh,
) -> Result<crate::protocol::ApiResponse, InferenceError>
where
Refresh: FnOnce(String) -> RefreshFuture,
RefreshFuture: Future<Output = car_auth::ReactiveRefreshResult>,
{
match self
.parslee_assistant_request(endpoint, bearer, request, retry_observer)
.await
{
Ok(response) => Ok(response),
Err(error) if is_auth_rejection(&error) => {
let (api_base, token) = self.refreshed_bound_authority(
refresh(bearer.to_string()).await,
&request.model,
error,
)?;
self.parslee_assistant_request(&api_base, &token, request, retry_observer)
.await
}
Err(error) => Err(error),
}
}
async fn execute_parslee_stream_with_refresh<Refresh, RefreshFuture>(
&self,
endpoint: &str,
bearer: &str,
request: &crate::protocol::ApiRequest,
spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
refresh: Refresh,
) -> Result<tokio::sync::mpsc::Receiver<crate::stream::StreamEvent>, InferenceError>
where
Refresh: FnOnce(String) -> RefreshFuture,
RefreshFuture: Future<Output = car_auth::ReactiveRefreshResult>,
{
match self
.parslee_assistant_stream_request(endpoint, bearer, request, spend_guard.clone())
.await
{
Ok(receiver) => Ok(receiver),
Err(error) if is_auth_rejection(&error) => {
let (api_base, token) = self.refreshed_bound_authority(
refresh(bearer.to_string()).await,
&request.model,
error,
)?;
self.parslee_assistant_stream_request(&api_base, &token, request, spend_guard)
.await
}
Err(error) => Err(error),
}
}
async fn lease_key(
&self,
schema: &ModelSchema,
endpoint: &str,
) -> Result<KeyLease, InferenceError> {
let fallback_env = match &schema.source {
ModelSource::RemoteApi {
protocol: ApiProtocol::OpenRouter,
..
} => {
let (api_key, source) = crate::openrouter::resolve_credential().ok_or_else(|| {
InferenceError::ProviderKeyMissing {
provider: schema.provider.clone(),
model: schema.id.clone(),
env_vars: vec![crate::openrouter::API_KEY_ENV.to_string()],
message: "OpenRouter requires a key — run `car keys set openrouter` or connect your OpenRouter account in CarHost"
.to_string(),
}
})?;
return Ok(KeyLease {
api_key,
env_var: format!("{}:{}", crate::openrouter::API_KEY_ENV, source.as_str()),
});
}
ModelSource::RemoteApi { api_key_env, .. } => api_key_env.as_str(),
ModelSource::Ollama { .. } | ModelSource::VllmMlx { .. } => {
return Ok(KeyLease {
api_key: String::new(),
env_var: String::new(),
})
}
ModelSource::Proprietary {
ref auth,
ref provider,
..
} => {
let (token, source_label) = match auth {
ProprietaryAuth::OAuth2Pkce { .. } => {
if !provider.eq_ignore_ascii_case("parslee")
|| !schema.provider.eq_ignore_ascii_case("parslee")
{
return Err(InferenceError::InferenceFailed(format!(
"OAuth2 PKCE credentials are supported only for provider \
'parslee'; refusing model {} from proprietary provider \
'{provider}'",
schema.id
)));
}
return Err(InferenceError::InferenceFailed(
"internal Parslee OAuth resolution must include its API base"
.to_string(),
));
}
ProprietaryAuth::ChatGptSubscription {} => {
return Err(InferenceError::InferenceFailed(
"internal: openai-codex credentials are resolved with the request authority"
.to_string(),
));
}
ProprietaryAuth::BearerTokenEnv { env_var }
| ProprietaryAuth::ApiKeyEnv { env_var } => (
car_secrets::resolve_env_or_keychain(env_var),
env_var.as_str(),
),
};
let token = match token {
Some(t) => t,
None => {
let (reason, detail) = (
crate::CredentialFailure::EnvVarMissing {
env_var: source_label.to_string(),
},
format!("set ${source_label} for provider '{provider}'"),
);
return Err(InferenceError::CredentialUnavailable {
provider: provider.to_string(),
model: schema.id.clone(),
reason,
detail,
});
}
};
return Ok(KeyLease {
api_key: token,
env_var: String::new(),
});
}
_ => {
return Err(InferenceError::InferenceFailed(format!(
"model {} is not remote",
schema.id
)))
}
};
self.register_model_keys(schema).await;
let env_vars = schema.all_api_key_envs();
self.key_pool
.lease_or_env(endpoint, fallback_env)
.await
.ok_or_else(|| {
let message = format!(
"no API keys available for endpoint {endpoint} (checked env vars: {env_vars:?}); \
{provider} has no key in the environment or the secret store: \
run `car keys set {key}`",
provider = schema.provider,
key = env_vars.first().map_or("<provider>", String::as_str),
);
InferenceError::ProviderKeyMissing {
provider: schema.provider.clone(),
model: schema.id.clone(),
env_vars,
message,
}
})
}
fn spend_admit_embed(
&self,
schema: &ModelSchema,
input_estimate: u64,
) -> Result<Option<crate::spend::SpendTicket>, InferenceError> {
let installed = self
.spend_gate
.read()
.unwrap_or_else(|p| p.into_inner())
.clone();
let Some(installed) = installed else {
return Ok(None);
};
let registry =
std::sync::Arc::new(crate::registry::UnifiedRegistry::single(schema.clone()));
let facts = crate::spend::facts_for(
®istry,
Some(schema),
&schema.id,
crate::billing::CallKind::Tokens,
input_estimate,
0,
);
installed.admit(®istry, facts, 0)
}
async fn embed_openai(
&self,
endpoint: &str,
api_key: &str,
model: &str,
texts: &[String],
) -> Result<Vec<Vec<f32>>, InferenceError> {
let url = format_endpoint(endpoint, "/v1/embeddings");
let body = serde_json::json!({
"model": model,
"input": texts,
});
let resp = self
.send_retrying_connect(model, &url, "HTTP error", None, &mut 1, &mut |_| {}, || {
self.client
.post(&url)
.header("Authorization", format!("Bearer {api_key}"))
.header("Content-Type", "application/json")
.json(&body)
.send()
})
.await?;
let status = resp.status();
let text = resp.text().await.map_err(|e| {
InferenceError::InferenceFailed(format!("read body: {}", error_chain(&e)))
})?;
if !status.is_success() {
return Err(InferenceError::InferenceFailed(format!(
"API returned {status}: {text}"
)));
}
let parsed: OpenAiEmbedResponse = serde_json::from_str(&text)
.map_err(|e| InferenceError::InferenceFailed(format!("parse response: {e}")))?;
Ok(order_embeddings(parsed.data))
}
#[allow(clippy::too_many_arguments)]
pub async fn generate_stream(
&self,
schema: &ModelSchema,
prompt: &str,
messages: Option<&[crate::tasks::generate::Message]>,
context: Option<&str>,
temperature: f64,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
images: Option<&[ContentBlock]>,
tool_choice: Option<&str>,
parallel_tool_calls: Option<bool>,
response_format: Option<&crate::tasks::generate::ResponseFormat>,
spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
) -> Result<tokio::sync::mpsc::Receiver<crate::stream::StreamEvent>, InferenceError> {
self.generate_stream_with_controls(
schema,
prompt,
messages,
context,
temperature,
max_tokens,
tools,
images,
tool_choice,
parallel_tool_calls,
response_format,
spend_guard,
&crate::protocol::OutputControls::default(),
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn generate_stream_with_controls(
&self,
schema: &ModelSchema,
prompt: &str,
messages: Option<&[crate::tasks::generate::Message]>,
context: Option<&str>,
temperature: f64,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
images: Option<&[ContentBlock]>,
tool_choice: Option<&str>,
parallel_tool_calls: Option<bool>,
response_format: Option<&crate::tasks::generate::ResponseFormat>,
spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
controls: &crate::protocol::OutputControls,
) -> Result<tokio::sync::mpsc::Receiver<crate::stream::StreamEvent>, InferenceError> {
let receiver = self
.open_stream(
schema,
prompt,
messages,
context,
temperature,
max_tokens,
tools,
images,
tool_choice,
parallel_tool_calls,
response_format,
spend_guard,
controls,
)
.await?;
Ok(crate::stream::enforce_stop_sequences(
receiver,
controls.stop.clone(),
))
}
#[allow(clippy::too_many_arguments)]
async fn open_stream(
&self,
schema: &ModelSchema,
prompt: &str,
messages: Option<&[crate::tasks::generate::Message]>,
context: Option<&str>,
temperature: f64,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
images: Option<&[ContentBlock]>,
tool_choice: Option<&str>,
parallel_tool_calls: Option<bool>,
response_format: Option<&crate::tasks::generate::ResponseFormat>,
spend_guard: Option<crate::routing_ext::MidStreamSpendGuard>,
controls: &crate::protocol::OutputControls,
) -> Result<tokio::sync::mpsc::Receiver<crate::stream::StreamEvent>, InferenceError> {
let (_, protocol) = extract_remote_endpoint(schema)?;
let handler = crate::protocol::handler_for(protocol);
if !handler.supports_streaming() {
return Err(InferenceError::UnsupportedMode {
mode: "streaming",
backend: handler.protocol_name(),
reason: "this remote protocol is wired for non-streaming calls only; use \
non-streaming infer/generate",
});
}
if !handler.supports_video()
&& (images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_video))
|| messages.is_some_and(|messages| {
messages.iter().any(|message| match message {
crate::tasks::generate::Message::UserMultimodal { content } => {
content.iter().any(ContentBlock::is_video)
}
_ => false,
})
}))
{
return Err(InferenceError::UnsupportedMode {
mode: "video-content-block",
backend: handler.protocol_name(),
reason: "this remote protocol has no native video input path; route to \
a provider that implements ProtocolHandler::supports_video() (Gemini)",
});
}
if !handler.supports_audio()
&& (images.is_some_and(|blocks| blocks.iter().any(ContentBlock::is_audio))
|| messages.is_some_and(|messages| {
messages.iter().any(|message| match message {
crate::tasks::generate::Message::UserMultimodal { content } => {
content.iter().any(ContentBlock::is_audio)
}
_ => false,
})
}))
{
return Err(InferenceError::UnsupportedMode {
mode: "audio-content-block",
backend: handler.protocol_name(),
reason: "this remote protocol has no native audio input path; route to \
a provider that implements ProtocolHandler::supports_audio() (Gemini)",
});
}
if let Some(err) = response_format_rejection(handler.as_ref(), response_format) {
return Err(err);
}
if crate::typesafe::is_direct(schema) {
return Err(InferenceError::UnsupportedMode {
mode: "stream",
backend: "typesafe-system-one",
reason: "TypeSafe System One is non-streaming; use infer or classify",
});
}
let (endpoint, resolved_protocol, lease) = self.resolve_request_authority(schema).await?;
debug_assert_eq!(protocol, resolved_protocol);
let started = std::time::Instant::now();
let settlement = KeySettlement::new(self.key_pool.clone(), &endpoint, &lease.env_var);
let api_key = lease.api_key;
let model = request_model_name(schema);
let (messages, system) = if matches!(schema.source, ModelSource::Proprietary { .. }) {
crate::protocol::OpenAiResponsesHandler.build_messages(
messages.unwrap_or(&[]),
prompt,
context,
images,
)
} else {
handler.build_messages(messages.unwrap_or(&[]), prompt, context, images)
};
let built_tools = build_request_tools(schema, handler.as_ref(), tools);
let req = crate::protocol::ApiRequest {
model: model.clone(),
messages,
system,
system_stable_prefix: None,
temperature,
max_tokens,
tools: built_tools,
tool_choice: tool_choice.map(str::to_string),
parallel_tool_calls,
stream: true,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: response_format.cloned(),
stop: controls.stop.clone(),
reasoning_effort: controls.reasoning_effort,
enable_thinking: template_thinking(schema, controls),
template_reasoning_effort: template_effort(schema, controls),
};
if codex::is_codex_subscription(schema) {
return self
.codex_stream_request(schema, &endpoint, &req, spend_guard)
.await;
}
if let ModelSource::Proprietary { protocol, .. } = &schema.source {
if protocol.wire == crate::schema::ProprietaryWire::SystemOne {
return Err(InferenceError::UnsupportedMode {
mode: "generate",
backend: "parslee-system-one",
reason: "a System One model returns typed decisions, not text; use classify",
});
}
return self
.execute_parslee_stream_with_refresh(
&endpoint,
&api_key,
&req,
spend_guard,
|rejected_bearer| async move {
self.refresh_bound_credentials(&rejected_bearer).await
},
)
.await;
}
let body = handler.build_request_body(&req);
crate::worker_cache::enforce_cached_prefix(
crate::worker_cache::cached_parts(protocol),
&body,
)?;
let input_estimate = request_input_estimate(&req);
let url = if matches!(protocol, ApiProtocol::Google) {
crate::protocol::google_stream_url(&endpoint, &model, &api_key)
} else if matches!(protocol, ApiProtocol::VertexAi) {
crate::protocol::vertex_stream_url(&endpoint, &model)
} else if matches!(protocol, ApiProtocol::AzureOpenAi) {
let api_version = match &schema.source {
ModelSource::RemoteApi { api_version, .. } => api_version.clone(),
_ => None,
};
let version = api_version.as_deref().unwrap_or("2024-10-21");
format!(
"{}/openai/deployments/{}/chat/completions?api-version={}",
endpoint.trim_end_matches('/'),
model,
version
)
} else {
format_endpoint(&endpoint, chat_path_for(schema, handler.as_ref()))
};
let mut headers = reqwest::header::HeaderMap::new();
for (name, value) in handler.auth_headers(&api_key) {
headers.insert(
reqwest::header::HeaderName::from_bytes(name.as_bytes()).map_err(|e| {
InferenceError::InferenceFailed(format!("auth header name: {e}"))
})?,
value.parse().map_err(|e| {
InferenceError::InferenceFailed(format!("auth header value: {e}"))
})?,
);
}
crate::outcome::note_dispatch();
let resp = self
.send_retrying_connect(
&schema.id,
&url,
"HTTP error",
Some((
std::time::Duration::from_secs(300),
"stream request timed out after 300s (tokio safety timeout)",
)),
&mut 1,
&mut |_| {},
|| {
self.client
.post(&url)
.headers(headers.clone())
.json(&body)
.send()
},
)
.await?;
let status = resp.status();
if !status.is_success() {
let err_text = resp.text().await.unwrap_or_default();
let is_rl = status.as_u16() == 429
|| err_text.contains("429")
|| err_text.contains("RESOURCE_EXHAUSTED");
settlement.fail(is_rl).await;
if is_account_http_status(status.as_u16()) {
return Err(InferenceError::ProviderAccount {
provider: schema.provider.clone(),
status: status.as_u16(),
message: account_error_message(protocol, status.as_u16()),
});
}
if matches!(protocol, ApiProtocol::OpenRouter) {
return Err(openrouter_http_error(status, &err_text, &model));
}
return Err(InferenceError::InferenceFailed(format!(
"API returned {status}: {err_text}"
)));
}
let (tx, rx) = tokio::sync::mpsc::channel::<crate::stream::StreamEvent>(64);
let mut late_lookup = if matches!(protocol, ApiProtocol::OpenRouter) {
crate::late_cost::current_probe().map(|probe| {
(
probe,
self.client.clone(),
endpoint.clone(),
handler.auth_headers(&api_key),
schema.id.clone(),
schema.name.clone(),
)
})
} else {
None
};
tokio::spawn(async move {
use futures::StreamExt;
let mut tx = BookingSender::new(tx);
async {
let mut byte_stream = resp.bytes_stream();
let mut buffer: Vec<u8> = Vec::new();
let mut spend_guard = spend_guard;
let mut saw_positive_completion = false;
let mut saw_chat_finish_reason = false;
let mut saw_google_finish_reason = false;
while let Some(chunk_result) = byte_stream.next().await {
match chunk_result {
Ok(bytes) => buffer.extend_from_slice(&bytes),
Err(e) => {
tracing::warn!(error = %e, "remote stream transport error mid-response");
let _ = tx
.send(crate::stream::StreamEvent::Error(
"remote stream transport error".to_string(),
))
.await;
return;
}
};
while let Some((pos, sep_len)) = find_sse_separator(&buffer) {
let event_block = String::from_utf8_lossy(&buffer[..pos]).into_owned();
buffer.drain(..pos + sep_len);
let sse_events = crate::stream::parse_sse_lines(&event_block);
for (event_type, data) in sse_events {
if data == "[DONE]" {
if matches!(
protocol,
ApiProtocol::OpenRouter
| ApiProtocol::OpenAiCompat
| ApiProtocol::AzureOpenAi
) && saw_chat_finish_reason
&& !saw_positive_completion
{
saw_positive_completion = true;
if tx
.send(crate::stream::StreamEvent::Done {
text: String::new(),
tool_calls: Vec::new(),
})
.await
.is_err()
{
return;
}
}
continue;
}
if late_lookup.is_some() {
if let Some(generation) = openrouter_generation_id(&data) {
if let Some((probe, client, base, headers, id, name)) =
late_lookup.take()
{
crate::late_cost::note(
Some(&probe),
crate::late_cost::LateCost::openrouter(
client,
&base,
&generation,
headers,
&id,
&name,
),
);
}
}
}
let stream_events = handler.parse_stream_event(&event_type, &data);
for evt in stream_events {
if let Some(guard) = spend_guard.as_mut() {
if let crate::stream::StreamEvent::Usage { input_tokens, .. } = &evt
{
guard.note_prompt_tokens(*input_tokens);
}
let delta_len = match &evt {
crate::stream::StreamEvent::TextDelta(t) => Some(t.as_str()),
crate::stream::StreamEvent::ToolCallDelta {
arguments_delta,
..
} => Some(arguments_delta.as_str()),
_ => None,
};
if let Some(delta) = delta_len {
let n =
crate::routing_ext::MidStreamSpendGuard::estimate_tokens(
delta,
);
if let Some(exceeded) = guard.record_tokens(n) {
tracing::warn!(
limit_usd = exceeded.limit_usd,
current_usd = exceeded.current_usd,
"mid-stream spend limit tripped; cancelling stream"
);
let _ = tx.send(evt).await;
let _ = tx
.send(crate::stream::StreamEvent::StopReason(format!(
"spend_limit: {exceeded}"
)))
.await;
let _ = tx
.send(crate::stream::StreamEvent::Attribution(
crate::outcome::UnattributedReason::Cancelled,
))
.await;
let _ = tx
.send(crate::stream::StreamEvent::Error(
"stream cancelled because the spend limit was reached"
.to_string(),
))
.await;
return; }
}
}
if let crate::stream::StreamEvent::StopReason(reason) = &evt {
if matches!(protocol, ApiProtocol::Anthropic) && reason == "refusal" {
let _ = tx
.send(crate::stream::StreamEvent::Attribution(
crate::outcome::UnattributedReason::ContentRefused,
))
.await;
let _ = tx
.send(crate::stream::StreamEvent::Error(
"anthropic declined to answer (type=refusal)"
.to_string(),
))
.await;
return;
}
if matches!(protocol, ApiProtocol::Google | ApiProtocol::VertexAi) {
match google_finish_outcome(reason) {
Ok(()) => saw_google_finish_reason = true,
Err(message) => {
let mut message = message.to_string();
if crate::protocol::google_reason_is_content_ruling(reason) {
let _ = tx
.send(crate::stream::StreamEvent::Attribution(
crate::outcome::UnattributedReason::ContentRefused,
))
.await;
message.push_str(&format!(
" (type=content_filter, code={reason})"
));
}
let _ = tx
.send(crate::stream::StreamEvent::Error(message))
.await;
return;
}
}
} else if matches!(
protocol,
ApiProtocol::OpenRouter
| ApiProtocol::OpenAiCompat
| ApiProtocol::AzureOpenAi
) {
saw_chat_finish_reason = true;
}
}
if matches!(evt, crate::stream::StreamEvent::Done { .. }) {
saw_positive_completion = true;
}
let terminal_error =
matches!(&evt, crate::stream::StreamEvent::Error(_));
if tx.send(evt).await.is_err() {
return; }
if terminal_error {
return; }
}
if matches!(protocol, ApiProtocol::Google | ApiProtocol::VertexAi)
&& saw_google_finish_reason
&& !saw_positive_completion
{
saw_positive_completion = true;
if tx
.send(crate::stream::StreamEvent::Done {
text: String::new(),
tool_calls: Vec::new(),
})
.await
.is_err()
{
return;
}
}
if protocol == ApiProtocol::Anthropic && event_type == "message_stop" {
saw_positive_completion = true;
if tx
.send(crate::stream::StreamEvent::Done {
text: String::new(),
tool_calls: Vec::new(),
})
.await
.is_err()
{
return;
}
}
}
}
}
if !buffer.is_empty() || !saw_positive_completion {
let _ = tx
.send(crate::stream::StreamEvent::Error(
"remote inference stream ended before provider completion".to_string(),
))
.await;
}
}
.await;
let latency_ms = started.elapsed().as_millis() as u64;
tx.settle(settlement, latency_ms, input_estimate).await;
});
Ok(rx)
}
}
impl Default for RemoteBackend {
fn default() -> Self {
Self::new()
}
}
pub(crate) fn is_trust_related(is_connect: bool, is_timeout: bool) -> bool {
is_connect && !is_timeout
}
fn extract_remote_endpoint(schema: &ModelSchema) -> Result<(String, ApiProtocol), InferenceError> {
match &schema.source {
ModelSource::RemoteApi {
endpoint, protocol, ..
} => {
let endpoint = if matches!(protocol, ApiProtocol::OpenRouter)
&& std::env::var("CAR_OPENROUTER_TEST_MODE").as_deref() == Ok("1")
{
std::env::var("OPENROUTER_API_BASE").unwrap_or_else(|_| endpoint.clone())
} else {
endpoint.clone()
};
Ok((endpoint, *protocol))
}
ModelSource::Ollama { host, .. } => Ok((host.clone(), ApiProtocol::OpenAiCompat)),
ModelSource::VllmMlx { endpoint, .. } => Ok((endpoint.clone(), ApiProtocol::OpenAiCompat)),
ModelSource::Proprietary {
endpoint,
auth: ProprietaryAuth::ChatGptSubscription {},
..
} => Ok((endpoint.clone(), ApiProtocol::OpenAiResponses)),
ModelSource::Proprietary { endpoint, .. } => {
Ok((endpoint.clone(), ApiProtocol::OpenAiCompat))
}
_ => Err(InferenceError::InferenceFailed(format!(
"model {} is not remote",
schema.id
))),
}
}
pub fn validate_bedrock_region(region: &str) -> Result<(), String> {
let shaped = (1..=63).contains(®ion.len())
&& region
.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
&& !region.starts_with('-')
&& !region.ends_with('-');
if shaped {
Ok(())
} else {
Err(format!(
"Bedrock model endpoint must be an AWS region such as \"us-east-1\" \
(lowercase letters, digits and hyphens); got {region:?}"
))
}
}
#[derive(Debug, PartialEq, Eq)]
struct BedrockEndpoint {
base_url: String,
host: String,
path_prefix: String,
}
fn bedrock_runtime_endpoint(region: &str) -> Result<BedrockEndpoint, InferenceError> {
validate_bedrock_region(region).map_err(InferenceError::InferenceFailed)?;
let override_url = ["AWS_ENDPOINT_URL_BEDROCK_RUNTIME", "AWS_ENDPOINT_URL"]
.iter()
.find_map(|name| std::env::var(name).ok().filter(|v| !v.trim().is_empty()));
let Some(raw) = override_url else {
let host = format!("bedrock-runtime.{region}.amazonaws.com");
return Ok(BedrockEndpoint {
base_url: format!("https://{host}"),
host,
path_prefix: String::new(),
});
};
let parsed = reqwest::Url::parse(raw.trim()).map_err(|e| {
InferenceError::InferenceFailed(format!("invalid AWS endpoint override: {e}"))
})?;
let hostname = parsed.host_str().ok_or_else(|| {
InferenceError::InferenceFailed("AWS endpoint override has no host".to_string())
})?;
let loopback = hostname.eq_ignore_ascii_case("localhost")
|| hostname
.trim_matches(|c| c == '[' || c == ']')
.parse::<std::net::IpAddr>()
.is_ok_and(|ip| ip.is_loopback());
match parsed.scheme() {
"https" => {}
"http" if loopback => {}
scheme => {
return Err(InferenceError::InferenceFailed(format!(
"AWS endpoint override must use https (plain {scheme} is accepted only for \
localhost, 127.0.0.0/8 and ::1)"
)))
}
}
let host = match parsed.port() {
Some(port) => format!("{hostname}:{port}"),
None => hostname.to_string(),
};
let path_prefix = parsed.path().trim_end_matches('/').to_string();
Ok(BedrockEndpoint {
base_url: format!("{}://{host}{path_prefix}", parsed.scheme()),
host,
path_prefix,
})
}
fn parslee_env_override_rejected_error() -> InferenceError {
InferenceError::InferenceFailed(
"the PARSLEE_ACCESS_TOKEN override was rejected; replace or remove the override before retrying"
.to_string(),
)
}
fn parslee_signed_out_error(schema: &ModelSchema, provider: &str) -> InferenceError {
parslee_signed_out_error_for_model(&schema.id, provider)
}
fn parslee_signed_out_error_for_model(model: &str, provider: &str) -> InferenceError {
InferenceError::CredentialUnavailable {
provider: provider.to_string(),
model: model.to_string(),
reason: crate::CredentialFailure::SignedOut,
detail: "no account is signed in. Run `car auth login`".to_string(),
}
}
fn parslee_credential_read_error(
schema: &ModelSchema,
provider: &str,
error: car_auth::CredentialReadError,
) -> InferenceError {
parslee_credential_read_error_for_model(&schema.id, provider, error)
}
fn work_context_credential_error(
model: &str,
error: car_auth::context::ContextCredentialError,
) -> InferenceError {
use car_auth::context::ContextCredentialError as E;
let reason = match &error {
E::AccountUnavailable => crate::CredentialFailure::SignedOut,
E::ReauthenticationRequired => crate::CredentialFailure::Expired { expires_at: 0 },
E::Conflict => crate::CredentialFailure::RaceRetryable,
E::Store(_) => crate::CredentialFailure::StoreUnreadable,
_ => {
return InferenceError::InferenceFailed(format!(
"Original work-context credentials unavailable: {error}"
))
}
};
InferenceError::CredentialUnavailable {
provider: "parslee".into(),
model: model.into(),
reason,
detail: format!("Original work-context credentials unavailable: {error}"),
}
}
fn parslee_credential_read_error_for_model(
model: &str,
provider: &str,
error: car_auth::CredentialReadError,
) -> InferenceError {
let recovery = match error.kind {
car_auth::CredentialReadFailureKind::Denied => {
"Keychain access was denied. Explicitly retry Keychain access when you are ready"
}
car_auth::CredentialReadFailureKind::Cancelled => {
"the Keychain prompt was cancelled. Explicitly retry Keychain access when you are ready"
}
car_auth::CredentialReadFailureKind::TimedOut => {
"the bounded Keychain helper timed out. Unlock the login keychain, then explicitly retry Keychain access"
}
car_auth::CredentialReadFailureKind::Unreadable => {
"the credential store could not be read. Unlock the login keychain, then explicitly retry Keychain access"
}
car_auth::CredentialReadFailureKind::Cooldown => {
"credential access is cooling down after a previous failure. Explicitly retry Keychain access; ordinary use will not prompt again"
}
};
InferenceError::CredentialUnavailable {
provider: provider.to_string(),
model: model.to_string(),
reason: crate::CredentialFailure::StoreUnreadable,
detail: format!("{recovery} ({error})"),
}
}
fn response_format_rejection(
handler: &dyn crate::protocol::ProtocolHandler,
response_format: Option<&crate::tasks::generate::ResponseFormat>,
) -> Option<InferenceError> {
match response_format {
Some(rf) if !handler.supports_response_format(rf) => {
Some(InferenceError::UnsupportedMode {
mode: match rf {
crate::tasks::generate::ResponseFormat::JsonSchema { .. } => {
"structured-output-json-schema"
}
crate::tasks::generate::ResponseFormat::JsonObject => "structured-output-json",
},
backend: handler.protocol_name(),
reason: "this provider is not wired for a provider-enforced \
response_format under CAR's pinned API version; supply a \
tool whose input schema is your schema plus a forcing \
tool_choice",
})
}
_ => None,
}
}
fn build_request_tools(
schema: &ModelSchema,
handler: &dyn crate::protocol::ProtocolHandler,
tools: Option<&[serde_json::Value]>,
) -> Option<Vec<serde_json::Value>> {
tools.map(|tools| {
if matches!(schema.source, ModelSource::Proprietary { .. }) {
crate::protocol::OpenAiResponsesHandler.build_tools(tools)
} else {
handler.build_tools(tools)
}
})
}
fn chat_path_for<'a>(
schema: &'a ModelSchema,
handler: &'a dyn crate::protocol::ProtocolHandler,
) -> &'a str {
match &schema.source {
ModelSource::Proprietary { protocol, .. } => protocol.chat_path.as_str(),
_ => handler.endpoint_path(),
}
}
fn request_model_name(schema: &ModelSchema) -> String {
if let Some(canonical_id) = crate::openrouter::canonical_managed_gateway_selector(schema) {
return canonical_id.to_string();
}
match &schema.source {
ModelSource::VllmMlx { model_name, .. } => model_name.clone(),
_ => schema.name.clone(),
}
}
fn openrouter_generation_id(body: &str) -> Option<String> {
serde_json::from_str::<serde_json::Value>(body)
.ok()?
.get("id")?
.as_str()
.filter(|id| !id.is_empty())
.map(str::to_string)
}
fn template_thinking(
schema: &ModelSchema,
controls: &crate::protocol::OutputControls,
) -> Option<bool> {
match schema.source {
ModelSource::VllmMlx { .. }
if crate::vllm_pool::reasoning_cannot_be_disabled(&schema.family) =>
{
None
}
ModelSource::VllmMlx { .. } => controls.enable_thinking(),
_ => None,
}
}
fn template_effort(
schema: &ModelSchema,
controls: &crate::protocol::OutputControls,
) -> Option<crate::tasks::generate::ReasoningEffort> {
match schema.source {
ModelSource::VllmMlx { .. }
if crate::vllm_pool::reasoning_cannot_be_disabled(&schema.family) =>
{
controls.reasoning_effort
}
_ => None,
}
}
fn format_endpoint(base: &str, path: &str) -> String {
let base = base.trim_end_matches('/');
if base.ends_with(path.trim_start_matches('/')) {
base.to_string()
} else {
format!("{}{}", base, path)
}
}
#[derive(Debug, Deserialize)]
struct OpenAiEmbedResponse {
data: Vec<OpenAiEmbedData>,
}
#[derive(Debug, Deserialize)]
struct OpenAiEmbedData {
embedding: Vec<f32>,
#[serde(default)]
index: usize,
}
fn order_embeddings(mut data: Vec<OpenAiEmbedData>) -> Vec<Vec<f32>> {
data.sort_by_key(|d| d.index);
data.into_iter().map(|d| d.embedding).collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::schema::ProprietaryProtocol;
use std::ffi::OsString;
use std::time::Duration;
#[test]
fn work_context_backends_reject_foreign_organization_and_authority() {
let backend = RemoteBackend::new();
let context = |org: &str| car_auth::context::CredentialContext {
api_base: "https://authority.example".into(),
account_id: "account-a".into(),
organization_id: Some(org.into()),
};
let parslee = backend.with_work_context(Some(context("parslee")));
let flyexclusive = backend.with_work_context(Some(context("flyexclusive")));
assert!(parslee
.verify_bound_org("https://authority.example/", "parslee")
.is_ok());
assert!(flyexclusive
.verify_bound_org("https://authority.example", "flyexclusive")
.is_ok());
assert!(parslee
.verify_bound_org("https://authority.example", "flyexclusive")
.is_err());
assert!(flyexclusive
.verify_bound_org("https://other.example", "flyexclusive")
.is_err());
assert!(backend.work_context.is_none());
assert!(std::sync::Arc::ptr_eq(
&parslee.key_pool,
&flyexclusive.key_pool
));
}
#[test]
fn a_renewal_outage_after_a_rejection_is_transient_not_a_sign_in_failure() {
let unauthorized = || {
InferenceError::InferenceFailed(
"Parslee org lookup failed: HTTP 401 Unauthorized".into(),
)
};
let error = refreshed_parslee_authority(
car_auth::ReactiveRefreshResult::TransientOrBackoff,
"m",
unauthorized(),
)
.unwrap_err();
let rendered = error.to_string();
assert!(
matches!(&error, InferenceError::Transient { status: None, message }
if message == car_auth::TOKEN_RENEWAL_UNAVAILABLE),
"{error:?}"
);
assert!(!crate::is_auth_failure_message(&rendered), "{rendered}");
assert_ne!(
crate::classify_fallback_reason(&error),
crate::FallbackReason::CredentialRejected
);
let aged_out = refreshed_parslee_authority(
car_auth::ReactiveRefreshResult::TransientOrBackoff,
"m",
InferenceError::CredentialUnavailable {
provider: "parslee".into(),
model: "m".into(),
reason: crate::CredentialFailure::Expired { expires_at: 1 },
detail: "expired".into(),
},
)
.unwrap_err();
assert!(
matches!(&aged_out, InferenceError::Transient { .. }),
"{aged_out:?}"
);
let forbidden = || InferenceError::ProviderAccount {
provider: "parslee".into(),
status: 403,
message: "no access to this organization".into(),
};
let kept = refreshed_parslee_authority(
car_auth::ReactiveRefreshResult::TransientOrBackoff,
"m",
forbidden(),
)
.unwrap_err();
assert!(
matches!(&kept, InferenceError::ProviderAccount { status: 403, .. }),
"{kept:?}"
);
let quoted_401 = refreshed_parslee_authority(
car_auth::ReactiveRefreshResult::TransientOrBackoff,
"m",
InferenceError::InferenceFailed(
"Parslee org lookup failed: HTTP 403 Forbidden: retried after HTTP 401".into(),
),
)
.unwrap_err();
assert!(
matches!("ed_401, InferenceError::InferenceFailed(message) if message.contains("HTTP 403")),
"{quoted_401:?}"
);
let quoted_403 = refreshed_parslee_authority(
car_auth::ReactiveRefreshResult::TransientOrBackoff,
"m",
InferenceError::InferenceFailed(
"Parslee org lookup failed: HTTP 401 Unauthorized: not HTTP 403".into(),
),
)
.unwrap_err();
assert!(
matches!("ed_403, InferenceError::Transient { .. }),
"{quoted_403:?}"
);
let expired = refreshed_parslee_authority(
car_auth::ReactiveRefreshResult::ReauthenticationRequired { expires_at: 1 },
"m",
unauthorized(),
)
.unwrap_err();
assert_eq!(
crate::classify_fallback_reason(&expired),
crate::FallbackReason::CredentialRejected
);
}
#[test]
fn work_context_failures_preserve_recovery_and_membership_advice() {
use car_auth::context::ContextCredentialError as E;
assert!(matches!(
work_context_credential_error("model", E::ReauthenticationRequired),
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::Expired { .. },
..
}
));
assert!(matches!(
work_context_credential_error("model", E::AccountUnavailable),
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::SignedOut,
..
}
));
assert!(matches!(
work_context_credential_error("model", E::Conflict),
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::RaceRetryable,
..
}
));
let backend =
RemoteBackend::new().with_work_context(Some(car_auth::context::CredentialContext {
api_base: "https://authority.example".into(),
account_id: "account".into(),
organization_id: Some("parslee".into()),
}));
let denied = backend
.refreshed_bound_authority(
car_auth::ReactiveRefreshResult::StoreError(car_auth::CredentialReadError {
kind: car_auth::CredentialReadFailureKind::Denied,
message: E::OrganizationDenied.to_string(),
}),
"model",
InferenceError::InferenceFailed("HTTP 403".into()),
)
.unwrap_err()
.to_string();
assert!(denied.contains("membership"));
assert!(denied.contains("other contexts remain available"));
assert!(!denied.contains("Keychain"));
}
#[tokio::test]
async fn work_context_http_retry_preserves_org_for_buffered_and_streamed_calls() {
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
let servers = [MockServer::start().await, MockServer::start().await];
for (streaming, server) in [false, true].into_iter().zip(&servers) {
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", "Bearer expired"))
.respond_with(ResponseTemplate::new(401))
.expect(1)
.mount(server)
.await;
for org in ["parslee", "flyexclusive"] {
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {org}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": org})),
)
.expect(1)
.mount(server)
.await;
Mock::given(method("POST"))
.and(path(format!("/api/v1/orgs/{org}/inference/responses")))
.and(header("authorization", format!("Bearer {org}")))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"event: response.output_text.delta\n",
"data: {\"delta\":\"original organization\"}\n\n",
"event: response.completed\n",
"data: {\"response\":{\"usage\":{\"input_tokens\":3,\"output_tokens\":2}}}\n\n"
), "text/event-stream"))
.expect(1)
.mount(server).await;
}
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(ResponseTemplate::new(200).set_body_json(
serde_json::json!({"account":{"email":"same-user@example.test"}}),
))
.expect(2)
.mount(server)
.await;
let context = |org: &str| car_auth::context::CredentialContext {
api_base: server.uri(),
account_id: "same-account".into(),
organization_id: Some(org.into()),
};
let backend = RemoteBackend::new();
let parslee = backend.with_work_context(Some(context("parslee")));
let flyexclusive = backend.with_work_context(Some(context("flyexclusive")));
let request = crate::protocol::ApiRequest {
model: "parslee/test".into(),
messages: vec![serde_json::json!({"role":"user","content":"hello"})],
system: None,
system_stable_prefix: None,
temperature: 0.0,
max_tokens: 32,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
stream: streaming,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: None,
stop: Vec::new(),
reasoning_effort: None,
enable_thinking: None,
template_reasoning_effort: None,
};
let renewed = car_auth::ResolvedParsleeCredential {
access_token: "parslee".into(),
api_base: server.uri(),
expires_at: 9_999_999_999,
};
if streaming {
let mut receiver = parslee
.execute_parslee_stream_with_refresh(
&server.uri(),
"expired",
&request,
None,
|rejected| async move {
assert_eq!(rejected, "expired");
car_auth::ReactiveRefreshResult::Refreshed(renewed)
},
)
.await
.expect("original stream recovers");
let mut text = String::new();
let mut completed = false;
while let Some(event) = receiver.recv().await {
match event {
crate::stream::StreamEvent::TextDelta(delta) => text.push_str(&delta),
crate::stream::StreamEvent::Done { .. } => completed = true,
crate::stream::StreamEvent::Error(error) => panic!("{error}"),
_ => {}
}
}
assert_eq!(text, "original organization");
assert!(completed);
} else {
let response = parslee
.execute_parslee_request_with_refresh(
&server.uri(),
"expired",
&request,
&mut |_| {},
|rejected| async move {
assert_eq!(rejected, "expired");
car_auth::ReactiveRefreshResult::Refreshed(renewed)
},
)
.await
.expect("original buffered call recovers");
assert_eq!(response.text, "original organization");
}
assert!(!crate::parslee_credential::credential_rejected());
flyexclusive
.parslee_assistant_request(&server.uri(), "flyexclusive", &request, &mut |_| {})
.await
.expect("other organization stays usable");
let error = parslee
.parslee_assistant_request(&server.uri(), "flyexclusive", &request, &mut |_| {})
.await
.expect_err("cached foreign organization is refused");
assert!(error.to_string().contains("immutable work context"));
let requests = server.received_requests().await.unwrap();
assert_eq!(
requests
.iter()
.filter(|r| r.method.as_str() == "POST")
.count(),
2
);
}
}
#[tokio::test]
async fn work_context_http_refuses_foreign_authority_before_sending_bearer() {
let server = wiremock::MockServer::start().await;
let backend =
RemoteBackend::new().with_work_context(Some(car_auth::context::CredentialContext {
api_base: "https://original.example".into(),
account_id: "original-account".into(),
organization_id: Some("parslee".into()),
}));
let error = backend
.parslee_identity(
&server.uri(),
"must-not-leave-original-authority",
&mut |_| {},
)
.await
.expect_err("foreign authority must be refused before HTTP");
assert!(error.to_string().contains("immutable work context"));
assert!(server.received_requests().await.unwrap().is_empty());
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let original = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(ResponseTemplate::new(302).insert_header("location", server.uri()))
.expect(1)
.mount(&original)
.await;
let bound = backend.with_work_context(Some(car_auth::context::CredentialContext {
api_base: original.uri(),
account_id: "original-account".into(),
organization_id: Some("parslee".into()),
}));
bound
.parslee_identity(&original.uri(), "original-bearer", &mut |_| {})
.await
.expect_err("context-bound lookup must not follow redirects");
assert!(server.received_requests().await.unwrap().is_empty());
}
struct TestEnvRestore(Vec<(&'static str, Option<OsString>)>);
impl TestEnvRestore {
fn capture(keys: &[&'static str]) -> Self {
Self(
keys.iter()
.map(|key| (*key, std::env::var_os(key)))
.collect(),
)
}
}
impl Drop for TestEnvRestore {
fn drop(&mut self) {
for (key, value) in self.0.drain(..) {
unsafe {
match value {
Some(value) => std::env::set_var(key, value),
None => std::env::remove_var(key),
}
}
}
}
}
use crate::tls_client::test_seam::trust_store_independent_client;
fn newline_in_api_key_error() -> reqwest::Error {
trust_store_independent_client()
.post("https://example.invalid/v1/chat/completions")
.header("authorization", "Bearer sk-test-key\n")
.build()
.expect_err("a header value containing a newline must fail to build")
}
#[test]
fn builder_error_surfaces_its_underlying_cause() {
let err = newline_in_api_key_error();
let old_rendering = format!("{err}");
assert!(
!old_rendering.contains("failed to parse header value"),
"reqwest's Display is expected to omit the cause; if this now \
includes it, the whole premise of error_chain has changed. \
got: {old_rendering}"
);
let chain = error_chain(&err);
assert!(
chain.contains("failed to parse header value"),
"the source chain must name the real cause, got: {chain}"
);
}
#[test]
fn builder_error_is_classified_permanent() {
let err = newline_in_api_key_error();
assert!(
err.is_builder(),
"precondition: this is a builder-kind error"
);
assert!(
!reqwest_error_is_transient(&err),
"a request that was never constructed must not be retried"
);
let attempt = TransportAttemptError::from_reqwest("HTTP error", &err);
assert!(!attempt.retryable);
assert!(
attempt
.error
.to_string()
.contains("failed to parse header value"),
"got: {}",
attempt.error
);
}
#[tokio::test]
async fn genuine_transport_error_is_classified_transient() {
let err = trust_store_independent_client()
.get("http://127.0.0.1:1/v1/chat/completions")
.send()
.await
.expect_err("nothing listens on loopback port 1");
assert!(
!err.is_builder(),
"precondition: this reached the transport"
);
assert!(
reqwest_error_is_transient(&err),
"a connect failure is worth the retry budget, got: {err}"
);
assert!(TransportAttemptError::from_reqwest("HTTP error", &err).retryable);
}
#[tokio::test]
async fn unarmed_phase_default_still_cuts_a_silent_future() {
let start = std::time::Instant::now();
let out: Result<(), PhaseCut> = phase_bounded(
std::future::pending::<()>(),
start,
Duration::from_millis(300),
)
.await;
assert!(
matches!(out, Err(PhaseCut::Default)),
"a silent future with no deadline must hit the default cut"
);
assert!(
start.elapsed() < Duration::from_secs(15),
"the default cut must fire near the default, took {:?}",
start.elapsed()
);
}
#[test]
fn unarmed_phase_default_scales_with_the_answer_a_slow_model_owes() {
let model = |tps: Option<f64>| {
let mut m = openrouter_schema("http://127.0.0.1:1");
m.performance.tokens_per_second = tps;
m
};
let slow = unarmed_phase_default(&model(Some(13.0)), 6000);
assert!(
slow > Duration::from_secs(900),
"a 13 tok/s model owes ~462 s for 6,000 tokens; got {slow:?}"
);
assert_eq!(
unarmed_phase_default(&model(Some(100.0)), 6000),
Duration::from_secs(REMOTE_PHASE_DEFAULT_SECS),
);
assert_eq!(
unarmed_phase_default(&model(None), 6000),
Duration::from_secs(REMOTE_PHASE_DEFAULT_SECS),
);
assert_eq!(
unarmed_phase_default(&model(Some(0.0)), 6000),
Duration::from_secs(REMOTE_PHASE_DEFAULT_SECS),
);
assert!(
unarmed_phase_default(&model(Some(0.01)), 1_000_000)
< Duration::from_secs(REMOTE_ABSOLUTE_CEILING_SECS),
"the watchdog must stay below the reqwest failsafe"
);
}
#[tokio::test]
async fn armed_deadline_overrides_the_phase_default() {
crate::offload::scope_inference_control_id("eyj-mech-a".to_string(), async {
crate::offload::set_remote_deadline("eyj-mech-a", Duration::from_secs(20));
let start = std::time::Instant::now();
let out = phase_bounded(
tokio::time::sleep(Duration::from_millis(900)),
start,
Duration::from_millis(300),
)
.await;
crate::offload::clear_remote_deadline("eyj-mech-a");
assert!(
out.is_ok(),
"an armed deadline must own the budget; the default must not cut the phase"
);
})
.await;
}
#[tokio::test]
async fn retry_that_cannot_fit_names_the_applied_deadline() {
crate::offload::scope_inference_control_id("eyj-mech-b".to_string(), async {
crate::offload::set_remote_deadline("eyj-mech-b", Duration::from_secs(3));
let err = deadline_permits_retry(Duration::from_secs(6))
.expect_err("6s backoff cannot fit in a 3s deadline");
crate::offload::clear_remote_deadline("eyj-mech-b");
let msg = err.to_string();
assert!(
msg.contains("3000 ms infer deadline"),
"the error must name the applied deadline, got: {msg}"
);
})
.await;
}
#[tokio::test]
async fn mock_provider_deadline_fires_at_the_deadline_and_names_it() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
std::env::set_var("CAR_EYJ_DEADLINE_TEST_KEY", "test-key");
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_secs(120))
.set_body_json(serde_json::json!({"never": "arrives"})),
)
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::OpenAiCompat,
"gpt-eyj-test",
"CAR_EYJ_DEADLINE_TEST_KEY",
);
let backend = RemoteBackend::new();
let start = std::time::Instant::now();
let err = crate::offload::scope_inference_control_id("eyj-e2e-a".to_string(), async {
crate::offload::set_remote_deadline("eyj-e2e-a", Duration::from_millis(900));
let r = backend
.generate(&schema, "hello", None, 0.0, 32, None)
.await;
crate::offload::clear_remote_deadline("eyj-e2e-a");
r
})
.await
.expect_err("a provider slower than the deadline must terminate at the deadline");
let elapsed = start.elapsed();
let msg = err.to_string();
assert!(
msg.contains("deadline exceeded") && msg.contains("900 ms infer deadline"),
"the terminal error must name the applied deadline, got: {msg}"
);
assert!(
elapsed >= Duration::from_millis(850) && elapsed < Duration::from_secs(30),
"termination must land at the deadline, not a transport ceiling; took {elapsed:?}"
);
}
#[tokio::test]
async fn mock_provider_completes_under_an_armed_deadline() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
std::env::set_var("CAR_EYJ_DEADLINE_TEST_KEY2", "test-key");
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_millis(700))
.set_body_json(serde_json::json!({
"choices": [{
"message": {"role": "assistant", "content": "deadline-ok"},
"finish_reason": "stop"
}],
"usage": {"prompt_tokens": 2, "completion_tokens": 3}
})),
)
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::OpenAiCompat,
"gpt-eyj-test2",
"CAR_EYJ_DEADLINE_TEST_KEY2",
);
let backend = RemoteBackend::new();
let out = crate::offload::scope_inference_control_id("eyj-e2e-b".to_string(), async {
crate::offload::set_remote_deadline("eyj-e2e-b", crate::LOADED_WAIT);
let r = backend
.generate(&schema, "hello", None, 0.0, 32, None)
.await;
crate::offload::clear_remote_deadline("eyj-e2e-b");
r
})
.await
.expect("a provider inside the deadline must complete normally");
assert_eq!(out, "deadline-ok");
}
#[tokio::test]
async fn a_key_kept_only_in_the_secret_store_authenticates_a_request() {
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
const NAME: &str = "CAR_STORE_ONLY_PROVIDER_TEST_KEY";
const FAKE: &str = "sk-store-only-fake-0123456789";
let _provider_env = crate::openrouter::test_environment_scope_async().await;
std::env::remove_var(NAME);
car_secrets::SecretStore::new()
.put(&car_secrets::SecretRef::with_default_service(NAME), FAKE)
.unwrap();
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(header("authorization", format!("Bearer {FAKE}").as_str()))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{
"message": {"role": "assistant", "content": "authenticated"},
"finish_reason": "stop"
}],
"usage": {"prompt_tokens": 2, "completion_tokens": 3}
})))
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::OpenAiCompat,
"store-only-key",
NAME,
);
let out = RemoteBackend::new()
.generate(&schema, "hello", None, 0.0, 32, None)
.await;
let exported = std::env::var_os(NAME);
let _ = car_secrets::SecretStore::new()
.delete(&car_secrets::SecretRef::with_default_service(NAME));
assert_eq!(out.expect("store-only key authenticates"), "authenticated");
assert!(exported.is_none(), "the call exported the key");
}
#[tokio::test]
async fn transport_error_redacts_url_userinfo() {
let mut err = trust_store_independent_client()
.get("http://127.0.0.1:1/v1/chat/completions")
.send()
.await
.expect_err("nothing listens on loopback port 1");
let url = err
.url_mut()
.expect("transport errors retain the request URL");
url.set_username("leaked-user")
.expect("the test username is valid URL userinfo");
url.set_password(Some("leaked-password"))
.expect("the test password is valid URL userinfo");
assert_eq!(url.username(), "leaked-user");
assert_eq!(url.password(), Some("leaked-password"));
let chain = error_chain(&err);
assert!(
!chain.contains("leaked-user"),
"the username must not survive redaction, got: {chain}"
);
assert!(
!chain.contains("leaked-password"),
"the password must not survive redaction, got: {chain}"
);
}
#[test]
fn timeout_attempt_is_retryable() {
assert!(TransportAttemptError::timeout("request timed out after 300s").retryable);
}
#[test]
fn error_chain_redacts_credentials_in_urls() {
let raw = "error sending request for url \
(https://generativelanguage.googleapis.com/v1beta/models/x:generateContent\
?alt=sse&key=AIzaSyREAL_SECRET_VALUE&pretty=1)";
let scrubbed = redact_sensitive(raw);
assert!(
!scrubbed.contains("AIzaSyREAL_SECRET_VALUE"),
"the API key must not survive redaction, got: {scrubbed}"
);
assert!(scrubbed.contains("key=REDACTED"), "got: {scrubbed}");
assert!(scrubbed.contains("alt=sse"), "got: {scrubbed}");
assert!(scrubbed.contains("pretty=1"), "got: {scrubbed}");
assert_eq!(redact_sensitive("?monkey=banana"), "?monkey=banana");
}
#[derive(Debug)]
struct Link {
message: String,
source: Option<Box<Link>>,
}
impl std::fmt::Display for Link {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.message)
}
}
impl std::error::Error for Link {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
self.source
.as_deref()
.map(|s| s as &(dyn std::error::Error + 'static))
}
}
fn distinct_chain(depth: usize, width: usize) -> Link {
let mut err = Link {
message: format!("cause-{depth}-{}", "z".repeat(width)),
source: None,
};
for i in (0..depth).rev() {
err = Link {
message: format!("cause-{i}-{}", "y".repeat(width)),
source: Some(Box::new(err)),
};
}
err
}
#[test]
fn error_chain_is_length_bounded() {
let chain = error_chain(&distinct_chain(40, 400));
assert!(
chain.len() <= MAX_ERROR_CHAIN_LEN + "... (truncated)".len(),
"chain must stay bounded, got {} bytes",
chain.len()
);
assert!(chain.ends_with("... (truncated)"), "got: {chain}");
}
#[test]
fn error_chain_is_depth_bounded() {
let chain = error_chain(&distinct_chain(40, 0));
assert!(chain.ends_with(": ..."), "got: {chain}");
assert!(
chain.contains(&format!("cause-{}", MAX_ERROR_CHAIN_DEPTH - 1)),
"the last link inside the cap must be rendered, got: {chain}"
);
assert!(
!chain.contains(&format!("cause-{}", MAX_ERROR_CHAIN_DEPTH + 1)),
"links past the cap must be dropped, got: {chain}"
);
}
#[test]
fn error_chain_does_not_repeat_a_restated_cause() {
let err = Link {
message: "inference failed: connection reset".to_string(),
source: Some(Box::new(Link {
message: "connection reset".to_string(),
source: None,
})),
};
assert_eq!(error_chain(&err), "inference failed: connection reset");
}
pub(super) fn openrouter_schema(endpoint: &str) -> ModelSchema {
let mut schema = crate::openrouter::curated_schemas()
.into_iter()
.find(|schema| schema.id == "openrouter/deepseek/deepseek-v3.2")
.expect("curated personal OpenRouter schema");
let ModelSource::RemoteApi {
endpoint: schema_endpoint,
..
} = &mut schema.source
else {
panic!("personal OpenRouter schema must be remote");
};
*schema_endpoint = endpoint.to_string();
schema
}
#[tokio::test]
async fn a_worker_request_without_its_cache_value_first_is_never_sent() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
const PARTITION: &str = "car_label_0123456789abcdef";
std::env::set_var("CAR_WORKER_CACHE_TEST_KEY", "test-key");
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{
"message": {"role": "assistant", "content": "ok"},
"finish_reason": "stop"
}],
"usage": {"prompt_tokens": 2, "completion_tokens": 3}
})))
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::OpenAiCompat,
"gpt-worker-cache-test",
"CAR_WORKER_CACHE_TEST_KEY",
);
let backend = RemoteBackend::new();
let system = "WORKER_SYSTEM".to_string();
let refused = crate::worker_cache::scope_cache_partition(Some(PARTITION.into()), async {
backend
.generate(&schema, "hello", Some(system.as_str()), 0.0, 32, None)
.await
})
.await;
assert!(
matches!(refused, Err(InferenceError::UnsupportedMode { .. })),
"{refused:?}"
);
let streamed = crate::worker_cache::scope_cache_partition(Some(PARTITION.into()), async {
backend
.generate_stream(
&schema,
"hello",
None,
Some(system.as_str()),
0.0,
32,
None,
None,
None,
None,
None,
None,
)
.await
.map(|_| ())
})
.await;
assert!(
matches!(streamed, Err(InferenceError::UnsupportedMode { .. })),
"{streamed:?}"
);
assert!(
server.received_requests().await.unwrap().is_empty(),
"a refused worker request reached the provider"
);
let leading = format!("{PARTITION}\nWORKER_SYSTEM");
let sent = crate::worker_cache::scope_cache_partition(Some(PARTITION.into()), async {
backend
.generate(&schema, "hello", Some(&leading), 0.0, 32, None)
.await
})
.await;
assert!(sent.is_ok(), "{sent:?}");
assert_eq!(server.received_requests().await.unwrap().len(), 1);
}
#[tokio::test]
async fn empty_tools_and_a_json_object_format_are_sent() {
use wiremock::matchers::method;
use wiremock::{Mock, MockServer, ResponseTemplate};
const PARTITION: &str = "car_label_0123456789abcdef";
std::env::set_var("CAR_WORKER_CACHE_TEST_KEY2", "test-key");
let system = format!("{PARTITION}\nWORKER_SYSTEM");
let json_object = crate::tasks::generate::ResponseFormat::JsonObject;
for (protocol, format) in [
(ApiProtocol::OpenAiCompat, Some(&json_object)),
(ApiProtocol::OpenAiResponses, Some(&json_object)),
(ApiProtocol::Anthropic, None),
] {
let server = MockServer::start().await;
Mock::given(method("POST"))
.respond_with(ResponseTemplate::new(200).set_body_raw("", "text/event-stream"))
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
protocol,
"worker-cache-shapes",
"CAR_WORKER_CACHE_TEST_KEY2",
);
let opened =
crate::worker_cache::scope_cache_partition(Some(PARTITION.into()), async {
RemoteBackend::new()
.generate_stream(
&schema,
"hello",
None,
Some(system.as_str()),
0.0,
32,
Some(&[]),
None,
None,
None,
format,
None,
)
.await
.map(|_| ())
})
.await;
assert!(opened.is_ok(), "{protocol:?}: {opened:?}");
let requests = server.received_requests().await.unwrap();
assert_eq!(requests.len(), 1, "{protocol:?}");
}
}
fn remote_stream_schema(
endpoint: &str,
protocol: ApiProtocol,
model: &str,
api_key_env: &str,
) -> ModelSchema {
ModelSchema {
id: format!("test/{model}"),
name: model.to_string(),
provider: "test".into(),
family: "test".into(),
version: "1".into(),
capabilities: vec![crate::schema::ModelCapability::Generate],
context_length: 128_000,
max_output_tokens: Some(8_192),
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::RemoteApi {
endpoint: endpoint.to_string(),
api_key_env: api_key_env.to_string(),
api_key_envs: vec![],
api_version: None,
protocol,
},
tags: vec!["test".into()],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Community,
deprecated: false,
available: true,
weights_ready: true,
}
}
#[test]
fn a_content_refusal_is_classified_from_the_tags_the_gateway_sent() {
let events = crate::stream::parse_openai_responses_sse_line(
"error",
r#"{"error":{"message":"content refused","type":"invalid_request_error","code":"content_policy_violation"}}"#,
);
let crate::stream::StreamEvent::Error(detail) = events.first().expect("an error event")
else {
panic!("expected an Error event");
};
let (kind, code) = content_refusal_tags(detail).expect("must classify as a refusal");
assert_eq!(code.as_deref(), Some("content_policy_violation"));
assert_eq!(kind.as_deref(), Some("invalid_request_error"));
}
#[test]
fn ordinary_failures_are_not_mistaken_for_refusals() {
for detail in [
"managed inference failed",
"boom (type=server_error, code=internal_error)",
"slow down (type=rate_limit_error, code=rate_limit_exceeded)",
"the safety content policy doc explains this",
"failed (type=",
] {
assert!(
content_refusal_tags(detail).is_none(),
"must not classify as a refusal: {detail}"
);
}
}
#[test]
fn either_tag_alone_is_enough() {
assert!(content_refusal_tags("nope (code=content_filter)").is_some());
assert!(content_refusal_tags("nope (type=content_policy_violation)").is_some());
assert!(content_refusal_tags("nope (code=moderation_blocked)").is_some());
assert!(content_refusal_tags("nope (code=safety_block)").is_some());
assert!(content_refusal_tags("anthropic declined to answer (type=refusal)").is_some());
assert!(content_refusal_tags("boom (type=refusal_service_unavailable)").is_none());
}
#[test]
fn gateway_not_configured_body_is_classified() {
let body = r#"{"error":{"code":"openrouter_not_configured","type":"gateway_error",
"message":"OpenRouter inference is not configured on this Parslee environment."}}"#;
let detail = gateway_unconfigured_detail(reqwest::StatusCode::SERVICE_UNAVAILABLE, body)
.expect("the reported 503 body must classify");
assert!(
detail.contains("not configured on this Parslee environment"),
"the operator-facing message must survive for the caller to surface: {detail}"
);
}
#[test]
fn gateway_not_configured_matches_the_code_not_the_prose_or_status() {
let other_upstream = r#"{"error":{"code":"bedrock_not_configured","type":"gateway_error","message":"nope"}}"#;
assert!(
gateway_unconfigured_detail(reqwest::StatusCode::BAD_GATEWAY, other_upstream).is_some(),
"a different upstream, and a different 5xx, is the same condition"
);
let bare = r#"{"error":{"code":"openrouter_not_configured"}}"#;
assert_eq!(
gateway_unconfigured_detail(reqwest::StatusCode::SERVICE_UNAVAILABLE, bare).as_deref(),
Some("openrouter_not_configured")
);
}
#[test]
fn unrelated_gateway_failures_do_not_classify() {
assert!(gateway_unconfigured_detail(
reqwest::StatusCode::INTERNAL_SERVER_ERROR,
r#"{"error":{"code":"internal_error","type":"gateway_error","message":"boom"}}"#
)
.is_none());
assert!(gateway_unconfigured_detail(
reqwest::StatusCode::FORBIDDEN,
r#"{"error":{"code":"openrouter_not_configured"}}"#
)
.is_none());
assert!(gateway_unconfigured_detail(
reqwest::StatusCode::SERVICE_UNAVAILABLE,
r#"{"error":{"code":"overloaded","message":"openrouter is not configured yet"}}"#
)
.is_none());
assert!(gateway_unconfigured_detail(
reqwest::StatusCode::SERVICE_UNAVAILABLE,
"<html>502</html>"
)
.is_none());
}
#[test]
fn parslee_body_uses_max_output_tokens_not_maxtokens() {
let input = vec![serde_json::json!({"role": "user", "content": "hi"})];
let body = parslee_request_body("u1", input, 42, None, None, None);
assert_eq!(body["max_output_tokens"], 42);
assert!(
body.get("maxTokens").is_none(),
"maxTokens makes the gateway fail inference and CAR fall back to local"
);
assert_eq!(body["userId"], "u1");
assert!(body.get("temperature").is_none());
assert!(body.get("tools").is_none());
let tools = vec![serde_json::json!({"type": "function", "name": "x"})];
let body2 = parslee_request_body(
"u1",
vec![serde_json::json!({"role": "user", "content": "hi"})],
8,
Some(&tools),
None,
None,
);
assert!(body2["tools"].is_array());
let gateway_model = "parslee/openrouter/open-reasoning";
let gateway_body = parslee_request_body(
"u1",
vec![serde_json::json!({"role": "user", "content": "hi"})],
8,
None,
None,
Some(gateway_model),
);
assert_eq!(gateway_body["model"], gateway_model);
let legacy_body = parslee_request_body(
"u1",
vec![serde_json::json!({"role": "user", "content": "hi"})],
8,
None,
None,
Some("parslee/gpt-5.5"),
);
assert!(legacy_body.get("model").is_none());
for schema in crate::openrouter::builtin_schemas()
.into_iter()
.filter(|schema| schema.id.starts_with("parslee/openrouter/"))
{
assert_eq!(
schema.supported_params,
vec![crate::schema::GenerateParam::MaxTokens],
"{} advertises a request control outside this body builder",
schema.id
);
assert_eq!(
gateway_body["max_output_tokens"], 8,
"the advertised MaxTokens control must reach the managed request"
);
}
}
fn parslee_body_with_choice(tool_choice: Option<&str>) -> serde_json::Value {
let tools = vec![serde_json::json!({"type": "function", "name": "extract"})];
parslee_request_body(
"u1",
vec![serde_json::json!({"role": "user", "content": "hi"})],
8,
Some(&tools),
tool_choice,
None,
)
}
#[test]
fn parslee_body_maps_tool_choice_to_the_responses_spelling() {
assert_eq!(
parslee_body_with_choice(Some("required"))["tool_choice"],
"required"
);
assert_eq!(
parslee_body_with_choice(Some("auto"))["tool_choice"],
"auto"
);
assert_eq!(
parslee_body_with_choice(Some("none"))["tool_choice"],
"none"
);
assert_eq!(
parslee_body_with_choice(Some("any"))["tool_choice"],
"required"
);
let named = parslee_body_with_choice(Some("extract"));
assert_eq!(named["tool_choice"]["type"], "function");
assert_eq!(named["tool_choice"]["name"], "extract");
}
#[test]
fn parslee_body_omits_tool_choice_unless_the_caller_set_one() {
assert!(parslee_body_with_choice(None).get("tool_choice").is_none());
assert!(parslee_body_with_choice(Some(" "))
.get("tool_choice")
.is_none());
let no_tools = parslee_request_body(
"u1",
vec![serde_json::json!({"role": "user", "content": "hi"})],
8,
None,
Some("required"),
None,
);
assert!(no_tools.get("tool_choice").is_none());
let empty_tools = parslee_request_body(
"u1",
vec![serde_json::json!({"role": "user", "content": "hi"})],
8,
Some(&[]),
Some("required"),
None,
);
assert!(empty_tools.get("tool_choice").is_none());
}
#[test]
fn parslee_input_preserves_multiturn_tool_history() {
let messages = vec![
serde_json::json!({ "role": "user", "content": "fix the bug" }),
serde_json::json!({
"role": "assistant",
"tool_calls": [{
"id": "call_1",
"type": "function",
"function": { "name": "read_file", "arguments": "{\"path\":\"src.py\"}" }
}]
}),
serde_json::json!({
"role": "tool",
"tool_call_id": "call_1",
"content": "def add(a,b): return a-b"
}),
];
let input = parslee_input_items(Some("you are a coder"), &messages);
assert_eq!(input.len(), 4, "no tool-history item may be dropped");
assert_eq!(input[0]["role"], "system");
assert_eq!(input[1]["role"], "user");
assert_eq!(input[2]["type"], "function_call");
assert_eq!(input[2]["call_id"], "call_1");
assert_eq!(input[2]["name"], "read_file");
assert_eq!(input[2]["arguments"], "{\"path\":\"src.py\"}");
assert_eq!(input[3]["type"], "function_call_output");
assert_eq!(input[3]["call_id"], "call_1");
assert_eq!(input[3]["output"], "def add(a,b): return a-b");
assert!(input[3].get("role").is_none());
}
#[test]
fn managed_incomplete_followed_by_done_remains_terminal_error() {
let events = parse_parslee_responses_sse(
concat!(
"event: response.output_text.delta\n",
"data: {\"delta\":\"partial\"}\n\n",
"event: response.incomplete\n",
"data: {\"response\":{\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"},\"usage\":{\"input_tokens\":4,\"output_tokens\":2}}}\n\n",
"data: [DONE]\n\n"
),
);
assert!(events.iter().any(
|event| matches!(event, crate::stream::StreamEvent::StopReason(reason) if reason == "max_output_tokens")
));
assert!(events.iter().any(
|event| matches!(event, crate::stream::StreamEvent::Error(message) if message == "managed inference incomplete: max_output_tokens")
));
assert!(
!events
.iter()
.any(|event| matches!(event, crate::stream::StreamEvent::Done { .. })),
"[DONE] after response.incomplete must not upgrade the turn to success"
);
}
#[tokio::test(flavor = "current_thread")]
async fn managed_buffered_response_rejects_partial_text_followed_by_clean_eof() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
let bearer = "buffered-partial-bearer";
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-partial"})),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200).set_body_json(
serde_json::json!({"account": {"email": "partial@example.test"}}),
),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-partial/inference/responses"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
"event: response.output_text.delta\ndata: {\"delta\":\"must not succeed\"}\n\n",
"text/event-stream",
))
.mount(&server)
.await;
let backend = RemoteBackend::new();
let request = crate::protocol::ApiRequest {
model: "managed-alias".into(),
messages: vec![serde_json::json!({"role": "user", "content": "hello"})],
system: None,
system_stable_prefix: None,
temperature: 0.0,
max_tokens: 64,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
stream: false,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: None,
stop: Vec::new(),
reasoning_effort: None,
enable_thinking: None,
template_reasoning_effort: None,
};
let error = backend
.parslee_assistant_request(&server.uri(), bearer, &request, &mut |_| {})
.await
.expect_err("clean EOF without response.completed must fail closed");
assert!(
error.to_string().contains("response.completed"),
"terminal error must tell callers the completion event was missing: {error}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn managed_request_retries_a_refused_connection_until_the_gateway_answers() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
async fn gateway_after(
addr: std::net::SocketAddr,
refused: std::sync::Arc<tokio::sync::Notify>,
) -> MockServer {
refused.notified().await;
let listener = std::net::TcpListener::bind(addr).expect("rebind the gateway port");
let server = MockServer::builder().listener(listener).start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-retry"})),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200).set_body_json(
serde_json::json!({"account": {"email": "retry@example.test"}}),
),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-retry/inference/responses"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
"event: response.output_text.delta\ndata: {\"delta\":\"reached\"}\n\n\
event: response.completed\ndata: {\"response\":{}}\n\n",
"text/event-stream",
))
.mount(&server)
.await;
server
}
let reserved = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a port");
let addr = reserved.local_addr().expect("reserved address");
drop(reserved);
let endpoint = format!("http://{addr}");
let bearer = "connect-retry-bearer";
let request = crate::protocol::ApiRequest {
model: "managed-alias".into(),
messages: vec![serde_json::json!({"role": "user", "content": "hello"})],
system: None,
system_stable_prefix: None,
temperature: 0.0,
max_tokens: 64,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
stream: false,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: None,
stop: Vec::new(),
reasoning_effort: None,
enable_thinking: None,
template_reasoning_effort: None,
};
let backend = RemoteBackend::new();
let refused = std::sync::Arc::new(tokio::sync::Notify::new());
let gateway = tokio::spawn(gateway_after(addr, std::sync::Arc::clone(&refused)));
let response = backend
.parslee_assistant_request(&endpoint, bearer, &request, &mut |_| refused.notify_one())
.await
.expect("the org lookup is retried until the gateway answers");
assert_eq!(response.text, "reached");
drop(gateway.await.expect("gateway task"));
let refused = std::sync::Arc::new(tokio::sync::Notify::new());
let gateway = tokio::spawn(gateway_after(addr, std::sync::Arc::clone(&refused)));
let mut retries = Vec::new();
let response = RemoteBackend::new()
.parslee_assistant_request(&endpoint, bearer, &request, &mut |progress| {
retries.push((progress.attempt, progress.reason));
refused.notify_one();
})
.await
.expect("the inference request is retried until the gateway answers");
assert_eq!(response.text, "reached");
assert_eq!(retries, vec![(2, "transport")]);
drop(gateway.await.expect("gateway task"));
}
#[tokio::test]
async fn exhausted_connect_failure_names_the_unreachable_host() {
let reserved = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a port");
let url = format!("http://{}/codex/responses", reserved.local_addr().unwrap());
drop(reserved);
let error = trust_store_independent_client()
.post(&url)
.send()
.await
.expect_err("nothing listens on that port");
assert!(
error.is_connect(),
"a refused connection is a connect error"
);
let failure = TransportAttemptError::from_reqwest("openai-codex HTTP error", &error);
match exhausted_transport_error(failure, &url, 3, Duration::from_secs(41)) {
InferenceError::Transient {
status: None,
message,
} => {
assert!(
message.starts_with("could not reach 127.0.0.1 for 41s (3 attempts): "),
"{message}"
);
assert!(
message.contains("openai-codex HTTP error: error sending request"),
"the underlying chain stays available for diagnosis: {message}"
);
}
other => panic!("expected Transient, got {other:?}"),
}
let reached = TransportAttemptError::timeout("response body read timed out");
match exhausted_transport_error(reached, &url, 3, Duration::from_secs(41)) {
InferenceError::Transient { message, .. } => {
assert_eq!(message, "inference failed: response body read timed out")
}
other => panic!("expected Transient, got {other:?}"),
}
}
#[tokio::test(flavor = "current_thread")]
async fn managed_stream_response_emits_error_after_partial_text_and_clean_eof() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
let bearer = "stream-partial-bearer";
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-stream-partial"})),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200).set_body_json(
serde_json::json!({"account": {"email": "partial@example.test"}}),
),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-stream-partial/inference/responses"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
"event: response.output_text.delta\ndata: {\"delta\":\"partial\"}\n\n",
"text/event-stream",
))
.mount(&server)
.await;
let backend = RemoteBackend::new();
let request = crate::protocol::ApiRequest {
model: "managed-alias".into(),
messages: vec![serde_json::json!({"role": "user", "content": "hello"})],
system: None,
system_stable_prefix: None,
temperature: 0.0,
max_tokens: 64,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
stream: true,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: None,
stop: Vec::new(),
reasoning_effort: None,
enable_thinking: None,
template_reasoning_effort: None,
};
let mut receiver = backend
.parslee_assistant_stream_request(&server.uri(), bearer, &request, None)
.await
.expect("HTTP stream starts");
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
assert!(
matches!(events.first(), Some(crate::stream::StreamEvent::TextDelta(text)) if text == "partial")
);
assert!(
matches!(events.last(), Some(crate::stream::StreamEvent::Error(message)) if message.contains("response.completed")),
"clean EOF without response.completed must append one terminal error: {events:?}"
);
assert!(
!events
.iter()
.any(|event| matches!(event, crate::stream::StreamEvent::Done { .. })),
"partial managed output must never be upgraded to success"
);
}
#[tokio::test(flavor = "current_thread")]
async fn openrouter_wire_success_stream_errors_concurrency_and_live_key_removal() {
use crate::stream::StreamEvent;
use wiremock::matchers::{body_partial_json, header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
crate::openrouter::set_test_credential(None);
let missing_schema = openrouter_schema("http://127.0.0.1:9");
let missing = RemoteBackend::new()
.generate(&missing_schema, "hello", None, 0.0, 32, None)
.await
.expect_err("missing OpenRouter key must fail before HTTP");
assert!(missing.to_string().contains("car keys set openrouter"));
crate::openrouter::set_test_credential(Some("test-openrouter-key"));
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(header("authorization", "Bearer test-openrouter-key"))
.and(header("x-openrouter-title", "CAR"))
.and(body_partial_json(serde_json::json!({
"model": "deepseek/deepseek-v3.2"
})))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_millis(100))
.set_body_json(serde_json::json!({
"choices": [{
"message": {"role": "assistant", "content": "openrouter-ok"},
"finish_reason": "stop"
}],
"usage": {"prompt_tokens": 2, "completion_tokens": 3}
})),
)
.mount(&server)
.await;
let schema = openrouter_schema(&server.uri());
let backend = RemoteBackend::new();
let (first, second) = tokio::join!(
backend.generate(&schema, "one", None, 0.0, 32, None),
backend.generate(&schema, "two", None, 0.0, 32, None),
);
assert_eq!(first.unwrap(), "openrouter-ok");
assert_eq!(second.unwrap(), "openrouter-ok");
let in_flight = backend.generate(&schema, "in flight", None, 0.0, 32, None);
let remove_key = async {
let deadline = std::time::Instant::now() + crate::LOADED_WAIT;
while server
.received_requests()
.await
.map_or(0, |seen| seen.len())
< 3
{
assert!(
std::time::Instant::now() < deadline,
"the in-flight request never reached the provider"
);
tokio::time::sleep(Duration::from_millis(5)).await;
}
crate::openrouter::set_test_credential(None);
};
let (in_flight_result, ()) = tokio::join!(in_flight, remove_key);
assert_eq!(in_flight_result.unwrap(), "openrouter-ok");
let next = backend
.generate(&schema, "next", None, 0.0, 32, None)
.await
.expect_err("the next request must observe key removal");
assert!(next.to_string().contains("car keys set openrouter"));
crate::openrouter::set_test_credential(Some("test-openrouter-key"));
let stream_server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"open\"}}]}\n\n",
"data: {\"choices\":[{\"delta\":{\"content\":\"router\"},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n"
),
"text/event-stream",
))
.mount(&stream_server)
.await;
let stream_schema = openrouter_schema(&stream_server.uri());
let mut receiver = backend
.generate_stream(
&stream_schema,
"stream",
None,
None,
0.0,
32,
None,
None,
None,
None,
None,
None,
)
.await
.unwrap();
let mut streamed = String::new();
let mut clean_stop = false;
let mut completed = false;
while let Some(event) = receiver.recv().await {
match event {
StreamEvent::TextDelta(delta) => streamed.push_str(&delta),
StreamEvent::StopReason(reason) if reason == "stop" => clean_stop = true,
StreamEvent::Done { .. } => completed = true,
_ => {}
}
}
assert_eq!(streamed, "openrouter");
assert!(clean_stop, "stream must expose the provider's clean stop");
assert!(
completed,
"a finish_reason followed by [DONE] is positive completion proof"
);
let stream_error_server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n",
"data: {\"error\":{\"code\":402,\"message\":\"private balance details\"}}\n\n",
"data: {\"choices\":[{\"delta\":{\"content\":\"must-not-appear\"}}]}\n\n"
),
"text/event-stream",
))
.mount(&stream_error_server)
.await;
let stream_error_schema = openrouter_schema(&stream_error_server.uri());
let mut receiver = backend
.generate_stream(
&stream_error_schema,
"stream error",
None,
None,
0.0,
32,
None,
None,
None,
None,
None,
None,
)
.await
.unwrap();
let mut errors = Vec::new();
let mut text = String::new();
while let Some(event) = receiver.recv().await {
match event {
StreamEvent::Error(message) => errors.push(message),
StreamEvent::TextDelta(delta) => text.push_str(&delta),
_ => {}
}
}
assert_eq!(errors, ["OpenRouter account is out of credits"]);
assert_eq!(text, "partial");
assert!(!errors[0].contains("private balance"));
for (status, body, expected) in [
(
401,
"sensitive upstream diagnostic must not escape",
"key rejected",
),
(
402,
"sensitive upstream diagnostic must not escape",
"out of credits",
),
(
404,
"sensitive upstream diagnostic must not escape",
"no longer available on OpenRouter",
),
(
400,
r#"{"error":{"message":"No endpoints found for this deprecated model; sensitive upstream diagnostic must not escape"}}"#,
"no longer available on OpenRouter",
),
] {
let error_server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(status).set_body_string(body))
.mount(&error_server)
.await;
let error_schema = openrouter_schema(&error_server.uri());
let error = backend
.generate(&error_schema, "fail", None, 0.0, 32, None)
.await
.expect_err("OpenRouter error should be translated");
let message = error.to_string();
assert!(message.contains(expected), "{status}: {message}");
assert!(!message.contains("sensitive upstream diagnostic"));
if status == 404 || status == 400 {
assert!(message.contains("deepseek/deepseek-v3.2"));
}
}
}
#[tokio::test]
async fn openrouter_cost_reaches_token_usage_one_shot_and_streamed() {
use crate::late_cost::{scope_late_cost, LateCostProbe, LateFetch};
use crate::stream::StreamAccumulator;
use std::sync::Arc;
use wiremock::matchers::{body_partial_json, header, method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
crate::openrouter::set_test_credential(Some("test-openrouter-key"));
async fn provider(cost: Option<f64>) -> MockServer {
let server = MockServer::start().await;
let figure = |usage: serde_json::Value| match cost {
Some(cost) => {
let mut usage = usage;
usage["cost"] = serde_json::json!(cost);
usage
}
None => usage,
};
let usage_chunk = serde_json::json!({
"id": "gen-stream", "choices": [],
"usage": figure(serde_json::json!({"prompt_tokens": 4, "completion_tokens": 1})),
});
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(body_partial_json(serde_json::json!({
"usage": {"include": true}, "stream": true
})))
.respond_with(ResponseTemplate::new(200).set_body_raw(
format!(
"data: {}\n\ndata: {}\n\ndata: {usage_chunk}\n\ndata: [DONE]\n\n",
r#"{"id":"gen-stream","choices":[{"delta":{"content":"hi"}}]}"#,
r#"{"id":"gen-stream","choices":[{"delta":{},"finish_reason":"stop"}]}"#,
),
"text/event-stream",
))
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(body_partial_json(
serde_json::json!({"usage": {"include": true}}),
))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"id": "gen-oneshot",
"choices": [{
"message": {"role": "assistant", "content": "hi"},
"finish_reason": "stop"
}],
"usage": figure(serde_json::json!({"prompt_tokens": 4, "completion_tokens": 1})),
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/v1/generation"))
.and(query_param("id", "gen-stream"))
.and(header("authorization", "Bearer test-openrouter-key"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"data": {"total_cost": 0.0031}})),
)
.mount(&server)
.await;
server
}
let backend = RemoteBackend::new();
for (cost, expected_one_shot, expected_stream) in [
(Some(0.0042), Some(0.0042), Some(0.0042)),
(None, None, None),
] {
let server = provider(cost).await;
let schema = openrouter_schema(&server.uri());
let probe = Arc::new(LateCostProbe::default());
let one_shot = scope_late_cost(
probe.clone(),
backend.generate_with_tools_multi(
&schema,
"q",
None,
0.0,
32,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
),
)
.await
.unwrap();
let usage = one_shot.4.expect("one-shot usage");
assert_eq!(usage.reported_cost_usd, expected_one_shot, "{cost:?}");
assert_eq!(usage.prompt_tokens, 4);
let late = probe.take().expect("the one-shot call's lookup");
assert!(late.serves(&schema.id));
let probe = Arc::new(LateCostProbe::default());
let mut receiver = scope_late_cost(
probe.clone(),
backend.generate_stream(
&schema, "q", None, None, 0.0, 32, None, None, None, None, None, None,
),
)
.await
.unwrap();
let mut accumulator = StreamAccumulator::default();
while let Some(event) = receiver.recv().await {
accumulator.push(&event);
}
let (_, _, usage, _) = accumulator.finish_with_usage();
let usage = usage.expect("streamed usage");
assert_eq!(usage.reported_cost_usd, expected_stream, "{cost:?}");
assert_eq!(usage.prompt_tokens, 4);
let late = probe.take().expect("the stream's lookup");
assert_eq!(late.fetch().await, LateFetch::Ready(0.0031));
}
}
#[tokio::test]
async fn an_unparseable_openrouter_answer_still_leaves_its_late_lookup() {
use crate::late_cost::{scope_late_cost, LateCostProbe};
use std::sync::Arc;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
crate::openrouter::set_test_credential(Some("test-openrouter-key"));
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(
serde_json::json!({"id": "gen-garbled", "object": "chat.completion"}),
))
.mount(&server)
.await;
let schema = openrouter_schema(&server.uri());
let probe = Arc::new(LateCostProbe::default());
let answer = scope_late_cost(
probe.clone(),
RemoteBackend::new().generate_with_tools_multi(
&schema,
"q",
None,
0.0,
32,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
),
)
.await;
assert!(answer.is_err(), "an answer with no choices parsed");
let late = probe
.take()
.expect("the unparseable call's late lookup was not left");
assert!(late.serves(&schema.id));
}
#[tokio::test]
async fn a_200_whose_body_cannot_be_read_is_not_sent_again() {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
crate::openrouter::set_test_credential(Some("test-openrouter-key"));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let requests = Arc::new(AtomicUsize::new(0));
let counted = requests.clone();
tokio::spawn(async move {
while let Ok((mut socket, _)) = listener.accept().await {
counted.fetch_add(1, Ordering::SeqCst);
tokio::spawn(async move {
let mut buf = [0u8; 8192];
let _ = socket.read(&mut buf).await;
let _ = socket
.write_all(
b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\
content-length: 1000\r\nconnection: close\r\n\r\n{\"id\":",
)
.await;
});
}
});
let schema = openrouter_schema(&format!("http://{address}"));
let answer = RemoteBackend::new()
.generate_with_tools_multi(
&schema,
"q",
None,
0.0,
32,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
)
.await;
let error = answer.expect_err("the body never arrived");
assert!(
matches!(
error,
InferenceError::Transient {
status: Some(200),
..
}
),
"{error:?}"
);
assert_eq!(requests.load(Ordering::SeqCst), 1, "sent once, never again");
}
async fn answering_with(
reply: &'static [u8],
hold: bool,
) -> (String, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let requests = Arc::new(AtomicUsize::new(0));
let counted = requests.clone();
tokio::spawn(async move {
while let Ok((mut socket, _)) = listener.accept().await {
let counted = counted.clone();
tokio::spawn(async move {
let mut buf = [0u8; 16384];
if matches!(socket.read(&mut buf).await, Ok(n) if n > 0) {
counted.fetch_add(1, Ordering::SeqCst);
}
let _ = socket.write_all(reply).await;
if hold {
std::future::pending::<()>().await;
}
});
}
});
(format!("http://{address}"), requests)
}
#[tokio::test]
async fn a_dropped_stream_only_frees_its_key_slot() {
const ENDPOINT: &str = "https://booking.test";
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let mut env = TestEnv::default();
env.set("CAR_BOOKING_DROPPED_KEY", Some("sk-booking"));
let pool = std::sync::Arc::new(KeyPool::new());
pool.register_endpoint(ENDPOINT, vec!["CAR_BOOKING_DROPPED_KEY".into()])
.await;
let lease = pool.lease(ENDPOINT).await.expect("a leased key");
let settlement = KeySettlement::new(pool.clone(), ENDPOINT, &lease.env_var);
let (tx, rx) = tokio::sync::mpsc::channel(4);
drop(rx);
let mut booking = BookingSender::new(tx);
let sent = booking
.send(crate::stream::StreamEvent::Done {
text: "never read".into(),
tool_calls: vec![],
})
.await;
assert!(sent.is_err(), "the receiver is gone");
booking.settle(settlement, 10, 5).await;
let mut stats = pool.endpoint_stats(ENDPOINT).await;
for _ in 0..crate::LOADED_WAIT.as_millis() / 5 {
if stats.first().is_some_and(|s| s.active_requests == 0) {
break;
}
tokio::time::sleep(Duration::from_millis(5)).await;
stats = pool.endpoint_stats(ENDPOINT).await;
}
assert_eq!(stats.len(), 1);
assert_eq!(
stats[0].successes, 0,
"an abandoned stream is not a success"
);
assert_eq!(stats[0].failures, 0, "nor a failure");
assert_eq!(stats[0].total_output_tokens, 0);
assert_eq!(stats[0].active_requests, 0, "the slot is freed");
}
#[tokio::test]
async fn a_queued_but_unread_done_only_frees_its_key_slot() {
const ENDPOINT: &str = "https://booking-unread.test";
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let mut env = TestEnv::default();
env.set("CAR_BOOKING_UNREAD_KEY", Some("sk-unread"));
let pool = std::sync::Arc::new(KeyPool::new());
pool.register_endpoint(ENDPOINT, vec!["CAR_BOOKING_UNREAD_KEY".into()])
.await;
let done = || crate::stream::StreamEvent::Done {
text: "the answer".into(),
tool_calls: vec![],
};
let settle_after = |read_it: bool| {
let pool = pool.clone();
async move {
let lease = pool.lease(ENDPOINT).await.expect("a leased key");
let settlement = KeySettlement::new(pool.clone(), ENDPOINT, &lease.env_var);
let (tx, mut rx) = tokio::sync::mpsc::channel(4);
let mut booking = BookingSender::new(tx);
booking.send(done()).await.expect("queued");
if read_it {
assert!(rx.recv().await.is_some());
booking.settle(settlement, 10, 5).await;
drop(rx);
} else {
drop(rx);
booking.settle(settlement, 10, 5).await;
}
}
};
settle_after(false).await;
let mut stats = pool.endpoint_stats(ENDPOINT).await;
for _ in 0..crate::LOADED_WAIT.as_millis() / 5 {
if stats.first().is_some_and(|s| s.active_requests == 0) {
break;
}
tokio::time::sleep(Duration::from_millis(5)).await;
stats = pool.endpoint_stats(ENDPOINT).await;
}
assert_eq!(stats[0].successes, 0, "an unread Done is not a success");
assert_eq!(stats[0].active_requests, 0, "the slot is freed");
settle_after(true).await;
let stats = pool.endpoint_stats(ENDPOINT).await;
assert_eq!(stats[0].successes, 1, "a Done the caller read is a success");
}
#[tokio::test]
async fn an_unparseable_200_is_booked_as_a_failure_with_no_tokens() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let mut env = TestEnv::default();
env.set("CAR_BOOKING_UNPARSEABLE_KEY", Some("sk-unparseable"));
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_raw("not json at all", "text/plain"))
.expect(1)
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::OpenAiCompat,
"gpt-unparseable-test",
"CAR_BOOKING_UNPARSEABLE_KEY",
);
let backend = RemoteBackend::new();
backend
.generate(&schema, "hello", None, 0.0, 32, None)
.await
.expect_err("an unparseable body reaches the caller as an error");
let stats: Vec<crate::key_pool::KeyStats> = backend
.key_pool
.all_stats()
.await
.into_values()
.flatten()
.filter(|s| s.env_var == "CAR_BOOKING_UNPARSEABLE_KEY")
.collect();
assert_eq!(stats.len(), 1, "{stats:?}");
assert_eq!(stats[0].successes, 0, "{stats:?}");
assert_eq!(stats[0].failures, 1, "{stats:?}");
assert_eq!(stats[0].total_input_tokens, 0, "{stats:?}");
assert_eq!(stats[0].total_output_tokens, 0, "{stats:?}");
}
#[tokio::test]
async fn a_request_reset_after_it_was_sent_is_not_sent_again() {
use std::sync::atomic::Ordering;
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
crate::openrouter::set_test_credential(Some("test-openrouter-key"));
let (endpoint, requests) = answering_with(b"", false).await;
let schema = openrouter_schema(&endpoint);
let (answer, noted) =
crate::spend::noting_sent_unread(RemoteBackend::new().generate_with_tools_multi(
&schema,
"q",
None,
0.0,
32,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
))
.await;
answer.expect_err("no answer came");
assert_eq!(requests.load(Ordering::SeqCst), 1, "sent once, never again");
assert!(noted, "the call is noted sent and unread");
}
fn bedrock_at(env: &mut TestEnv, endpoint: &str) -> ModelSchema {
env.set("AWS_ACCESS_KEY_ID", Some("AKIDEXAMPLEDUMMY"));
env.set(
"AWS_SECRET_ACCESS_KEY",
Some("dummy-secret-for-standin-only"),
);
env.set("AWS_SESSION_TOKEN", None);
env.set("AWS_ENDPOINT_URL", None);
env.set("AWS_ENDPOINT_URL_BEDROCK_RUNTIME", Some(endpoint));
env.set("CAR_BEDROCK_STANDIN_KEY", Some("unused-by-sigv4"));
remote_stream_schema(
"us-east-1",
ApiProtocol::Bedrock,
"anthropic.claude-test-v1:0",
"CAR_BEDROCK_STANDIN_KEY",
)
}
#[tokio::test(flavor = "current_thread")]
async fn a_bedrock_200_whose_body_cannot_be_read_ends_cut_off() {
use std::sync::atomic::Ordering;
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let _state_root = crate::openrouter::StateRootScope::new();
let (endpoint, requests) = answering_with(
b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\
content-length: 1000\r\nconnection: close\r\n\r\n{\"output\":",
false,
)
.await;
let mut env = TestEnv::default();
let schema = bedrock_at(&mut env, &endpoint);
let (answer, noted) = crate::spend::noting_sent_unread(
RemoteBackend::new().generate(&schema, "say hi", None, 0.0, 32, None),
)
.await;
answer.expect_err("the body never arrived");
assert_eq!(requests.load(Ordering::SeqCst), 1, "sent once, never again");
assert!(noted, "the call is noted sent and unread");
}
#[tokio::test(flavor = "current_thread")]
async fn a_bedrock_call_past_its_attempt_bound_ends_cut_off() {
use std::sync::atomic::Ordering;
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let _state_root = crate::openrouter::StateRootScope::new();
let (endpoint, requests) = answering_with(b"", true).await;
let mut env = TestEnv::default();
let schema = bedrock_at(&mut env, &endpoint);
let arrived = async {
for _ in 0..crate::LOADED_WAIT.as_millis() / 10 {
if requests.load(Ordering::SeqCst) > 0 {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
tokio::time::pause();
};
let backend = RemoteBackend::new();
let ((answer, noted), ()) = tokio::join!(
crate::spend::noting_sent_unread(
backend.generate(&schema, "say hi", None, 0.0, 32, None),
),
arrived,
);
answer.expect_err("no answer came");
assert_eq!(requests.load(Ordering::SeqCst), 1, "sent once, never again");
assert!(noted, "the call is noted sent and unread");
}
#[tokio::test(flavor = "current_thread")]
async fn personal_openrouter_partial_or_unframed_eof_never_emits_done() {
use crate::stream::StreamEvent;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _credential_scope = crate::openrouter::test_credential_scope();
crate::openrouter::set_test_credential(Some("terminal-key"));
for body in [
"data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n",
"data: {\"choices\":[{\"delta\":{\"content\":\"partial\"},\"finish_reason\":\"stop\"}]}\n\n",
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n",
"data: [DONE]\n\n"
),
"data: {\"choices\":[{\"delta\":{\"content\":\"trailing\"}}]}",
] {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(
ResponseTemplate::new(200).set_body_raw(body, "text/event-stream"),
)
.mount(&server)
.await;
let schema = openrouter_schema(&server.uri());
let mut receiver = RemoteBackend::new()
.generate_stream(
&schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
)
.await
.unwrap();
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
assert!(
!events
.iter()
.any(|event| matches!(event, StreamEvent::Done { .. })),
"partial sequence must not complete: {events:?}"
);
assert!(
matches!(events.last(), Some(StreamEvent::Error(message)) if message.contains("before provider completion")),
"partial sequence needs one sanitized terminal error: {events:?}"
);
}
}
#[tokio::test(flavor = "current_thread")]
async fn google_and_vertex_stream_only_deliberate_finish_reasons_complete() {
use crate::stream::StreamEvent;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _provider_env = crate::openrouter::test_environment_scope_async().await;
unsafe { std::env::set_var("CAR_GOOGLE_STREAM_MATRIX_KEY", "matrix-key") };
for protocol in [ApiProtocol::Google, ApiProtocol::VertexAi] {
for (reason, should_complete, refused) in [
("STOP", true, false),
("MAX_TOKENS", true, false),
("SAFETY", false, true),
("RECITATION", false, false),
("BLOCKLIST", false, true),
("PROHIBITED_CONTENT", false, true),
("SPII", false, true),
("IMAGE_SAFETY", false, true),
("MALFORMED_FUNCTION_CALL", false, false),
("UNKNOWN_PROVIDER_REASON", false, false),
] {
let server = MockServer::start().await;
let expected_path = match protocol {
ApiProtocol::Google => "/v1beta/models/gemini-test:streamGenerateContent",
ApiProtocol::VertexAi => {
"/publishers/google/models/gemini-test:streamGenerateContent"
}
_ => unreachable!(),
};
Mock::given(method("POST"))
.and(path(expected_path))
.respond_with(ResponseTemplate::new(200).set_body_raw(
format!(
"data: {{\"candidates\":[{{\"content\":{{\"parts\":[{{\"text\":\"matrix\"}}]}},\"finishReason\":\"{reason}\"}}]}}\n\n"
),
"text/event-stream",
))
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
protocol,
"gemini-test",
"CAR_GOOGLE_STREAM_MATRIX_KEY",
);
let mut receiver = RemoteBackend::new()
.generate_stream(
&schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
)
.await
.expect("Google and Vertex must both reach their streaming transports");
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
assert_eq!(
events
.iter()
.any(|event| matches!(event, StreamEvent::Done { .. })),
should_complete,
"{protocol:?}/{reason}: {events:?}"
);
assert_eq!(
events.iter().any(|event| matches!(
event,
StreamEvent::Attribution(crate::outcome::UnattributedReason::ContentRefused)
)),
refused,
"{protocol:?}/{reason}: a content ruling must not be charged to the model: {events:?}"
);
assert_eq!(
events.iter().any(|event| matches!(
event,
StreamEvent::Error(message) if content_refusal_tags(message).is_some()
)),
refused,
"{protocol:?}/{reason}: the daemon reads the verdict from the tags: {events:?}"
);
assert_eq!(
events
.iter()
.any(|event| matches!(event, StreamEvent::Error(_))),
!should_complete,
"{protocol:?}/{reason}: {events:?}"
);
}
}
unsafe { std::env::remove_var("CAR_GOOGLE_STREAM_MATRIX_KEY") };
}
#[tokio::test(flavor = "current_thread")]
async fn anthropic_stream_refusal_ends_as_a_typed_refusal() {
use crate::stream::StreamEvent;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _provider_env = crate::openrouter::test_environment_scope_async().await;
unsafe { std::env::set_var("CAR_ANTHROPIC_STREAM_REFUSAL_KEY", "refusal-key") };
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/messages"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"event: message_delta\n",
"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"refusal\"},\"usage\":{\"output_tokens\":0}}\n\n",
"event: message_stop\n",
"data: {\"type\":\"message_stop\"}\n\n"
),
"text/event-stream",
))
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::Anthropic,
"claude-test",
"CAR_ANTHROPIC_STREAM_REFUSAL_KEY",
);
let mut receiver = RemoteBackend::new()
.generate_stream(
&schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
)
.await
.unwrap();
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
assert!(
!events.iter().any(|e| matches!(e, StreamEvent::Done { .. })),
"a refusal must not complete: {events:?}"
);
assert!(
events.iter().any(|e| matches!(
e,
StreamEvent::Attribution(crate::outcome::UnattributedReason::ContentRefused)
)),
"{events:?}"
);
assert!(
matches!(events.last(), Some(StreamEvent::Error(m)) if content_refusal_tags(m).is_some()),
"the terminal error must classify as a refusal: {events:?}"
);
unsafe { std::env::remove_var("CAR_ANTHROPIC_STREAM_REFUSAL_KEY") };
}
#[tokio::test(flavor = "current_thread")]
async fn spend_cancelled_stream_returns_its_lease_without_a_key_failure() {
use crate::stream::StreamEvent;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _provider_env = crate::openrouter::test_environment_scope_async().await;
let api_key_env = "CAR_SPEND_CANCEL_LEASE_TEST_KEY";
unsafe { std::env::set_var(api_key_env, "fixture") };
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"a long and costly answer\"}}]}\n\n",
"data: [DONE]\n\n",
),
"text/event-stream",
))
.mount(&server)
.await;
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::OpenAiCompat,
"spend-test",
api_key_env,
);
let guard = crate::routing_ext::MidStreamSpendGuard::new(Some(0.01), 0.0099, 0.0, 1.0);
let backend = RemoteBackend::new();
let mut receiver = backend
.generate_stream(
&schema,
"stream",
None,
None,
0.0,
32,
None,
None,
None,
None,
None,
Some(guard),
)
.await
.unwrap();
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
unsafe { std::env::remove_var(api_key_env) };
assert!(
events.iter().any(|event| matches!(
event,
StreamEvent::Attribution(crate::outcome::UnattributedReason::Cancelled)
)),
"control: the spend guard cancelled the stream: {events:?}"
);
let keys: Vec<_> = backend
.key_pool
.all_stats()
.await
.into_values()
.flatten()
.collect();
assert_eq!(keys.len(), 1, "control: one pooled key");
assert_eq!(keys[0].total_requests, 1, "control: the key was leased");
assert_eq!(
keys[0].active_requests, 0,
"a cancelled stream leaves no request in flight on its key"
);
assert_eq!(
keys[0].failures, 0,
"CAR's own spend cancel is not booked as the key failing"
);
}
#[tokio::test(flavor = "current_thread")]
async fn azure_personal_openrouter_and_anthropic_terminal_matrix_stays_green() {
use crate::stream::StreamEvent;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
crate::openrouter::set_test_credential(Some("matrix-openrouter-key"));
unsafe {
std::env::set_var("CAR_AZURE_STREAM_MATRIX_KEY", "matrix-azure-key");
std::env::set_var("CAR_ANTHROPIC_STREAM_MATRIX_KEY", "matrix-anthropic-key");
}
for (protocol, model, expected_path, body, key_env) in [
(
ApiProtocol::AzureOpenAi,
"azure-deployment",
"/openai/deployments/azure-deployment/chat/completions",
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"azure\"},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n"
),
"CAR_AZURE_STREAM_MATRIX_KEY",
),
(
ApiProtocol::OpenRouter,
"vendor/personal-model",
"/v1/chat/completions",
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"personal\"},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n"
),
"IGNORED_FOR_OPENROUTER",
),
(
ApiProtocol::Anthropic,
"claude-test",
"/v1/messages",
concat!(
"event: content_block_delta\n",
"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"anthropic\"}}\n\n",
"event: message_delta\n",
"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":1}}\n\n",
"event: message_stop\n",
"data: {\"type\":\"message_stop\"}\n\n"
),
"CAR_ANTHROPIC_STREAM_MATRIX_KEY",
),
] {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path(expected_path))
.respond_with(
ResponseTemplate::new(200).set_body_raw(body, "text/event-stream"),
)
.mount(&server)
.await;
let schema = remote_stream_schema(&server.uri(), protocol, model, key_env);
let mut receiver = RemoteBackend::new()
.generate_stream(
&schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
)
.await
.unwrap();
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
assert!(
events
.iter()
.any(|event| matches!(event, StreamEvent::Done { .. })),
"{protocol:?}: {events:?}"
);
assert!(
!events
.iter()
.any(|event| matches!(event, StreamEvent::Error(_))),
"{protocol:?}: {events:?}"
);
}
unsafe {
std::env::remove_var("CAR_AZURE_STREAM_MATRIX_KEY");
std::env::remove_var("CAR_ANTHROPIC_STREAM_MATRIX_KEY");
}
}
#[tokio::test(flavor = "current_thread")]
async fn unary_call_is_served_through_each_direct_provider_format() {
use wiremock::matchers::{header, method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let mut env = TestEnv::default();
env.set("CAR_UNARY_MATRIX_KEY", Some("unary-matrix-key"));
crate::openrouter::set_test_credential(Some("unary-matrix-key"));
struct Case {
protocol: ApiProtocol,
model: &'static str,
endpoint_suffix: &'static str,
route: &'static str,
auth: Option<(&'static str, &'static str)>,
key_query: bool,
body: serde_json::Value,
expected: &'static str,
}
let cases = vec![
Case {
protocol: ApiProtocol::OpenAiResponses,
model: "gpt-responses-test",
endpoint_suffix: "",
route: "/v1/responses",
auth: Some(("authorization", "Bearer unary-matrix-key")),
key_query: false,
body: serde_json::json!({
"output": [
{"type": "reasoning", "summary": []},
{"type": "message", "content": [
{"type": "output_text", "text": "responses says hi"}
]}
],
"usage": {"input_tokens": 3, "output_tokens": 4}
}),
expected: "responses says hi",
},
Case {
protocol: ApiProtocol::Anthropic,
model: "claude-unary-test",
endpoint_suffix: "",
route: "/v1/messages",
auth: Some(("x-api-key", "unary-matrix-key")),
key_query: false,
body: serde_json::json!({
"content": [{"type": "text", "text": "anthropic says hi"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 3, "output_tokens": 4}
}),
expected: "anthropic says hi",
},
Case {
protocol: ApiProtocol::AzureOpenAi,
model: "azure-deployment",
endpoint_suffix: "",
route: "/openai/deployments/azure-deployment/chat/completions",
auth: None,
key_query: false,
body: serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "azure says hi"},
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 3, "completion_tokens": 4}
}),
expected: "azure says hi",
},
Case {
protocol: ApiProtocol::Google,
model: "gemini-unary-test",
endpoint_suffix: "",
route: "/v1beta/models/gemini-unary-test:generateContent",
auth: None,
key_query: true,
body: serde_json::json!({
"candidates": [{"content": {"parts": [{"text": "google says hi"}]},
"finishReason": "STOP"}]
}),
expected: "google says hi",
},
Case {
protocol: ApiProtocol::OpenAiCompat,
model: "gpt-compat-test",
endpoint_suffix: "",
route: "/v1/chat/completions",
auth: Some(("authorization", "Bearer unary-matrix-key")),
key_query: false,
body: serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "compat says hi"},
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 3, "completion_tokens": 4}
}),
expected: "compat says hi",
},
Case {
protocol: ApiProtocol::OpenRouter,
model: "vendor/openrouter-test",
endpoint_suffix: "",
route: "/v1/chat/completions",
auth: Some(("authorization", "Bearer unary-matrix-key")),
key_query: false,
body: serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "openrouter says hi"},
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 3, "completion_tokens": 4}
}),
expected: "openrouter says hi",
},
Case {
protocol: ApiProtocol::VertexAi,
model: "gemini-vertex-test",
endpoint_suffix: "/v1/projects/p/locations/l",
route: "/v1/projects/p/locations/l/publishers/google/models/gemini-vertex-test:generateContent",
auth: Some(("authorization", "Bearer unary-matrix-key")),
key_query: false,
body: serde_json::json!({
"candidates": [{"content": {"parts": [{"text": "vertex says hi"}]},
"finishReason": "STOP"}]
}),
expected: "vertex says hi",
},
];
for case in cases {
let server = MockServer::start().await;
let mut mock = Mock::given(method("POST")).and(path(case.route));
if let Some((name, value)) = case.auth {
mock = mock.and(header(name, value));
}
if case.key_query {
mock = mock.and(query_param("key", "unary-matrix-key"));
}
mock.respond_with(ResponseTemplate::new(200).set_body_json(case.body.clone()))
.expect(1)
.mount(&server)
.await;
let endpoint = format!("{}{}", server.uri(), case.endpoint_suffix);
let schema =
remote_stream_schema(&endpoint, case.protocol, case.model, "CAR_UNARY_MATRIX_KEY");
let text = RemoteBackend::new()
.generate(&schema, "say hi", None, 0.0, 32, None)
.await
.unwrap_or_else(|e| panic!("{:?}: {e}", case.protocol));
assert_eq!(text, case.expected, "{:?}", case.protocol);
drop(server);
}
}
#[derive(Default)]
struct TestEnv(Vec<(&'static str, Option<std::ffi::OsString>)>);
impl TestEnv {
fn set(&mut self, key: &'static str, value: Option<&str>) {
if !self.0.iter().any(|(k, _)| *k == key) {
self.0.push((key, std::env::var_os(key)));
}
unsafe {
match value {
Some(v) => std::env::set_var(key, v),
None => std::env::remove_var(key),
}
}
}
}
impl Drop for TestEnv {
fn drop(&mut self) {
for (key, previous) in self.0.drain(..).rev() {
unsafe {
match previous {
Some(v) => std::env::set_var(key, v),
None => std::env::remove_var(key),
}
}
}
}
}
struct SigV4VerifyingBedrock {
creds: crate::aws_sigv4::AwsCredentials,
reply: serde_json::Value,
}
impl wiremock::Respond for SigV4VerifyingBedrock {
fn respond(&self, request: &wiremock::Request) -> wiremock::ResponseTemplate {
let header = |name: &str| {
request
.headers
.get(name)
.and_then(|v| v.to_str().ok())
.unwrap_or_default()
.to_string()
};
let amz_date = header("x-amz-date");
let host = header("host");
let expected = crate::aws_sigv4::signed_headers(
&self.creds,
"us-east-1",
"bedrock",
"POST",
request.url.path(),
"",
&[
("host".to_string(), host),
("content-type".to_string(), header("content-type")),
],
&request.body,
&amz_date,
amz_date.get(..8).unwrap_or_default(),
);
let expected_auth = expected
.iter()
.find(|(k, _)| k == "authorization")
.map(|(_, v)| v.as_str())
.unwrap_or_default();
if header("authorization") == expected_auth {
wiremock::ResponseTemplate::new(200).set_body_json(self.reply.clone())
} else {
wiremock::ResponseTemplate::new(403).set_body_string("SignatureDoesNotMatch")
}
}
}
#[tokio::test(flavor = "current_thread")]
async fn bedrock_converse_call_is_served_through_the_aws_endpoint_override() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer};
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let _state_root = crate::openrouter::StateRootScope::new();
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/prefix/model/anthropic.claude-test-v1%3A0/converse"))
.respond_with(SigV4VerifyingBedrock {
creds: crate::aws_sigv4::AwsCredentials {
access_key_id: "AKIDEXAMPLEDUMMY".into(),
secret_access_key: "dummy-secret-for-standin-only".into(),
session_token: None,
},
reply: serde_json::json!({
"output": {"message": {"role": "assistant",
"content": [{"text": "bedrock says hi"}]}},
"stopReason": "end_turn",
"usage": {"inputTokens": 3, "outputTokens": 4, "totalTokens": 7}
}),
})
.expect(1)
.mount(&server)
.await;
let mut env = TestEnv::default();
env.set("AWS_ACCESS_KEY_ID", Some("AKIDEXAMPLEDUMMY"));
env.set(
"AWS_SECRET_ACCESS_KEY",
Some("dummy-secret-for-standin-only"),
);
env.set("AWS_SESSION_TOKEN", None);
env.set("AWS_ENDPOINT_URL", None);
env.set(
"AWS_ENDPOINT_URL_BEDROCK_RUNTIME",
Some(&format!("{}/prefix", server.uri())),
);
env.set("CAR_BEDROCK_STANDIN_KEY", Some("unused-by-sigv4"));
let schema = remote_stream_schema(
"us-east-1",
ApiProtocol::Bedrock,
"anthropic.claude-test-v1:0",
"CAR_BEDROCK_STANDIN_KEY",
);
let result = RemoteBackend::new()
.generate(&schema, "say hi", None, 0.0, 32, None)
.await;
assert_eq!(result.expect("bedrock converse call"), "bedrock says hi");
}
#[test]
fn bedrock_runtime_endpoint_defaults_to_the_real_aws_host_and_honors_overrides() {
let _guard = crate::openrouter::test_environment_scope();
let mut env = TestEnv::default();
env.set("AWS_ENDPOINT_URL_BEDROCK_RUNTIME", None);
env.set("AWS_ENDPOINT_URL", None);
let endpoint = |base_url: &str, host: &str, path_prefix: &str| BedrockEndpoint {
base_url: base_url.to_string(),
host: host.to_string(),
path_prefix: path_prefix.to_string(),
};
assert_eq!(
bedrock_runtime_endpoint("eu-west-1").unwrap(),
endpoint(
"https://bedrock-runtime.eu-west-1.amazonaws.com",
"bedrock-runtime.eu-west-1.amazonaws.com",
""
)
);
env.set("AWS_ENDPOINT_URL", Some("http://127.0.0.1:9000/"));
assert_eq!(
bedrock_runtime_endpoint("eu-west-1").unwrap(),
endpoint("http://127.0.0.1:9000", "127.0.0.1:9000", "")
);
env.set("AWS_ENDPOINT_URL", Some("http://[::1]:9000/"));
assert_eq!(
bedrock_runtime_endpoint("eu-west-1").unwrap(),
endpoint("http://[::1]:9000", "[::1]:9000", "")
);
let ipv6 = bedrock_runtime_endpoint("eu-west-1").unwrap();
assert!(reqwest::Url::parse(&format!("{}/model/x/converse", ipv6.base_url)).is_ok());
env.set(
"AWS_ENDPOINT_URL_BEDROCK_RUNTIME",
Some("https://vpce.example/prefix"),
);
assert_eq!(
bedrock_runtime_endpoint("eu-west-1").unwrap(),
endpoint("https://vpce.example/prefix", "vpce.example", "/prefix")
);
}
#[test]
fn bedrock_region_must_be_shaped_like_an_aws_region() {
let _guard = crate::openrouter::test_environment_scope();
let mut env = TestEnv::default();
env.set("AWS_ENDPOINT_URL_BEDROCK_RUNTIME", None);
env.set("AWS_ENDPOINT_URL", None);
for region in [
"us-east-1",
"us-gov-east-1",
"us-gov-west-1",
"eu-central-2",
"ap-southeast-5",
"il-central-1",
] {
let endpoint = bedrock_runtime_endpoint(region)
.unwrap_or_else(|e| panic!("{region:?} refused: {e}"));
assert_eq!(
endpoint.host,
format!("bedrock-runtime.{region}.amazonaws.com")
);
}
for region in [
"cn-north-1",
"cn-northwest-1",
"us-iso-east-1",
"us-isob-east-1",
"eusc-de-east-1",
] {
assert!(validate_bedrock_region(region).is_ok(), "{region:?}");
}
let too_long = "a".repeat(64);
let refused = [
"",
"x.evil.example/",
"us-east-1.evil.example",
"evil.example/x",
"us-east-1:443",
"user@evil.example",
"us-east-1/",
"us-east-1?",
"us-east-1#",
"us-east-1\\",
"us-east-1%2e",
"[::1]",
"US-EAST-1",
" us-east-1",
"us-east-1\n",
"us_east_1",
"-us-east-1",
"us-east-1-",
too_long.as_str(),
];
for region in refused {
let err = bedrock_runtime_endpoint(region).unwrap_err();
assert!(
err.to_string().contains("must be an AWS region"),
"{region:?}: {err}"
);
}
env.set("AWS_ENDPOINT_URL", Some("http://127.0.0.1:9000"));
for region in refused {
assert!(bedrock_runtime_endpoint(region).is_err(), "{region:?}");
}
}
#[tokio::test(flavor = "current_thread")]
async fn bedrock_call_with_a_crafted_region_sends_nothing() {
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let _state_root = crate::openrouter::StateRootScope::new();
let server = wiremock::MockServer::start().await;
let mut env = TestEnv::default();
env.set("AWS_ACCESS_KEY_ID", Some("AKIDEXAMPLEDUMMY"));
env.set(
"AWS_SECRET_ACCESS_KEY",
Some("dummy-secret-for-standin-only"),
);
env.set("AWS_SESSION_TOKEN", None);
env.set("AWS_ENDPOINT_URL", None);
env.set("AWS_ENDPOINT_URL_BEDROCK_RUNTIME", Some(&server.uri()));
env.set("CAR_BEDROCK_STANDIN_KEY", Some("unused-by-sigv4"));
let schema = remote_stream_schema(
"x.evil.example/",
ApiProtocol::Bedrock,
"anthropic.claude-test-v1:0",
"CAR_BEDROCK_STANDIN_KEY",
);
let err = RemoteBackend::new()
.generate(&schema, "say hi", None, 0.0, 32, None)
.await
.unwrap_err();
assert!(err.to_string().contains("must be an AWS region"), "{err}");
let received = server.received_requests().await.unwrap_or_default();
assert!(received.is_empty(), "sent {} request(s)", received.len());
}
#[tokio::test(flavor = "current_thread")]
async fn bedrock_stream_call_is_refused_and_sends_nothing() {
let (_credential_scope, _provider_env) =
crate::openrouter::test_credential_and_environment_scope_async().await;
let _state_root = crate::openrouter::StateRootScope::new();
let server = wiremock::MockServer::start().await;
let mut env = TestEnv::default();
env.set("CAR_BEDROCK_STANDIN_KEY", Some("unused-by-sigv4"));
let schema = remote_stream_schema(
&server.uri(),
ApiProtocol::Bedrock,
"anthropic.claude-test-v1:0",
"CAR_BEDROCK_STANDIN_KEY",
);
let result = RemoteBackend::new()
.generate_stream(
&schema, "say hi", None, None, 0.0, 32, None, None, None, None, None, None,
)
.await;
assert!(
matches!(
result,
Err(InferenceError::UnsupportedMode {
mode: "streaming",
backend: "bedrock",
..
})
),
"{:?}",
result.err()
);
let received = server.received_requests().await.unwrap_or_default();
assert!(received.is_empty(), "sent {} request(s)", received.len());
}
#[test]
fn bedrock_override_allows_plain_http_only_for_loopback() {
let _guard = crate::openrouter::test_environment_scope();
let mut env = TestEnv::default();
env.set("AWS_ENDPOINT_URL", None);
for ok in [
"http://localhost:8080",
"http://127.0.0.1:8080",
"http://127.9.9.9",
"http://[::1]:8080",
] {
env.set("AWS_ENDPOINT_URL_BEDROCK_RUNTIME", Some(ok));
assert!(bedrock_runtime_endpoint("us-east-1").is_ok(), "{ok}");
}
for refused in [
"http://bedrock.example",
"http://10.0.0.5:8080",
"http://localhost.evil.example",
"ftp://127.0.0.1",
] {
env.set("AWS_ENDPOINT_URL_BEDROCK_RUNTIME", Some(refused));
let err = bedrock_runtime_endpoint("us-east-1").unwrap_err();
assert!(
err.to_string().contains("must use https"),
"{refused}: {err}"
);
}
}
#[tokio::test(flavor = "current_thread")]
async fn personal_openrouter_spend_limit_is_error_not_success() {
use crate::stream::StreamEvent;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _credential_scope = crate::openrouter::test_credential_scope();
crate::openrouter::set_test_credential(Some("spend-key"));
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"cross limit\"}}]}\n\n",
"data: {\"choices\":[{\"delta\":{\"content\":\"must not arrive\"},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n"
),
"text/event-stream",
))
.mount(&server)
.await;
let schema = openrouter_schema(&server.uri());
let guard = crate::routing_ext::MidStreamSpendGuard::new(Some(0.0), 0.0, 0.0, 0.001);
let mut receiver = RemoteBackend::new()
.generate_stream(
&schema,
"stream",
None,
None,
0.0,
32,
None,
None,
None,
None,
None,
Some(guard),
)
.await
.unwrap();
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
assert!(events
.iter()
.any(|event| matches!(event, StreamEvent::StopReason(reason) if reason.starts_with("spend_limit:"))));
assert!(
matches!(events.last(), Some(StreamEvent::Error(message)) if message == "stream cancelled because the spend limit was reached"),
"spend cancellation must surface as a sanitized terminal error: {events:?}"
);
assert!(!events
.iter()
.any(|event| matches!(event, StreamEvent::Done { .. })));
}
#[tokio::test(flavor = "current_thread")]
async fn managed_openrouter_spend_limit_is_error_not_success() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
use crate::routing_ext::MidStreamSpendGuard;
use crate::stream::StreamEvent;
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
let bearer = "managed-spend-limit-bearer";
unsafe {
std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, bearer);
std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, server.uri());
}
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-spend"})),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"account": {"email": "user@example.test"}})),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-spend/inference/responses"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"event: response.output_text.delta\n",
"data: {\"delta\":\"costly managed delta\"}\n\n",
"event: response.completed\n",
"data: {\"response\":{\"usage\":{\"input_tokens\":4,\"output_tokens\":4}}}\n\n"
),
"text/event-stream",
))
.mount(&server)
.await;
let schema = crate::openrouter::curated_schemas()
.into_iter()
.find(|schema| schema.id == "parslee/openrouter/frontier-general")
.unwrap();
let guard = MidStreamSpendGuard::new(Some(0.0), 0.0, 0.0, 1.0);
let mut rx = RemoteBackend::new()
.generate_stream(
&schema,
"hello",
None,
None,
0.0,
32,
None,
None,
None,
None,
None,
Some(guard),
)
.await
.unwrap();
let mut events = Vec::new();
while let Some(event) = rx.recv().await {
events.push(event);
}
assert!(events.iter().any(
|event| matches!(event, StreamEvent::StopReason(reason) if reason.starts_with("spend_limit:"))
));
assert!(matches!(
events.last(),
Some(StreamEvent::Error(message))
if message == "stream cancelled because the spend limit was reached"
));
assert!(
matches!(
events.get(events.len() - 2),
Some(StreamEvent::Attribution(
crate::outcome::UnattributedReason::Cancelled
))
),
"the spend limit is the caller's policy, not the model: {events:?}"
);
assert!(!events
.iter()
.any(|event| matches!(event, StreamEvent::Done { .. })));
unsafe {
std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
}
}
#[test]
fn sse_separator_handles_lf_crlf_and_partials() {
assert_eq!(find_sse_separator(b"data: a\n\nrest"), Some((7, 2)));
assert_eq!(find_sse_separator(b"data: a\r\n\r\nrest"), Some((7, 4)));
assert_eq!(find_sse_separator(b"data: a\n"), None);
let buf = b"a\n\nb\r\n\r\nc";
assert_eq!(find_sse_separator(buf), Some((1, 2)));
}
#[test]
fn response_format_guard_rejects_only_unsupported() {
use crate::tasks::generate::ResponseFormat;
let schema = ResponseFormat::JsonSchema {
schema: serde_json::json!({"type": "object"}),
strict: true,
name: None,
};
let obj = ResponseFormat::JsonObject;
let anthropic = crate::protocol::handler_for(crate::schema::ApiProtocol::Anthropic);
for (format, expected_mode) in [
(&schema, "structured-output-json-schema"),
(&obj, "structured-output-json"),
] {
let rej = response_format_rejection(anthropic.as_ref(), Some(format))
.expect("Anthropic response format must be rejected");
match rej {
InferenceError::UnsupportedMode { mode, backend, .. } => {
assert_eq!(mode, expected_mode);
assert_eq!(backend, "anthropic");
}
other => panic!("expected UnsupportedMode, got {other:?}"),
}
}
assert!(
response_format_rejection(anthropic.as_ref(), None).is_none(),
"no response_format is never rejected"
);
let openai = crate::protocol::handler_for(crate::schema::ApiProtocol::OpenAiCompat);
assert!(response_format_rejection(openai.as_ref(), Some(&schema)).is_none());
assert!(response_format_rejection(openai.as_ref(), Some(&obj)).is_none());
}
#[test]
fn auth_rejection_is_anchored_on_http_status() {
use crate::InferenceError;
assert!(is_auth_rejection(&InferenceError::InferenceFailed(
"Parslee chat failed: HTTP 401: {\"error\":\"expired\"}".into()
)));
assert!(is_auth_rejection(&InferenceError::InferenceFailed(
"Parslee org lookup failed: HTTP 403: forbidden".into()
)));
assert!(is_auth_rejection(&InferenceError::CredentialUnavailable {
provider: "parslee".into(),
model: String::new(),
reason: crate::CredentialFailure::Expired { expires_at: 0 },
detail: "wording that mentions no status code at all".into(),
}));
assert!(!is_auth_rejection(&InferenceError::InferenceFailed(
"API returned 400: your last request 401'd upstream".into()
)));
assert!(!is_auth_rejection(&InferenceError::InferenceFailed(
"HTTP 500: server error".into()
)));
}
#[test]
fn embeddings_resorted_to_input_order() {
let data = vec![
OpenAiEmbedData {
embedding: vec![2.0],
index: 2,
},
OpenAiEmbedData {
embedding: vec![0.0],
index: 0,
},
OpenAiEmbedData {
embedding: vec![1.0],
index: 1,
},
];
assert_eq!(
order_embeddings(data),
vec![vec![0.0], vec![1.0], vec![2.0]]
);
}
#[test]
fn format_endpoint_no_dup() {
assert_eq!(
format_endpoint("https://api.openai.com", "/v1/chat/completions"),
"https://api.openai.com/v1/chat/completions"
);
assert_eq!(
format_endpoint(
"https://api.openai.com/v1/chat/completions",
"/v1/chat/completions"
),
"https://api.openai.com/v1/chat/completions"
);
assert_eq!(
format_endpoint("https://api.openai.com/", "/v1/chat/completions"),
"https://api.openai.com/v1/chat/completions"
);
}
#[test]
fn extract_endpoint_from_remote() {
let schema = ModelSchema {
id: "test/model:v1".into(),
name: "Test".into(),
provider: "test".into(),
family: "test".into(),
version: "1".into(),
capabilities: vec![],
context_length: 4096,
max_output_tokens: None,
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::RemoteApi {
endpoint: "https://api.test.com".into(),
api_key_env: "NONEXISTENT_TEST_KEY_12345".into(),
api_key_envs: vec![],
api_version: None,
protocol: ApiProtocol::OpenAiCompat,
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Curated,
deprecated: false,
available: false,
weights_ready: false,
};
let (endpoint, protocol) = extract_remote_endpoint(&schema).unwrap();
assert_eq!(endpoint, "https://api.test.com");
assert_eq!(protocol, ApiProtocol::OpenAiCompat);
}
#[test]
fn extract_endpoint_non_remote_fails() {
let schema = ModelSchema {
id: "local/model:v1".into(),
name: "Local".into(),
provider: "test".into(),
family: "test".into(),
version: "1".into(),
capabilities: vec![],
context_length: 4096,
max_output_tokens: None,
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::Local {
hf_repo: "test".into(),
hf_filename: "test".into(),
tokenizer_repo: "test".into(),
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Curated,
deprecated: false,
available: false,
weights_ready: false,
};
assert!(extract_remote_endpoint(&schema).is_err());
}
#[test]
fn proprietary_endpoint_is_openai_compat() {
let schema = ModelSchema {
id: "parslee/advisor".into(),
name: "Parslee Advisor".into(),
provider: "parslee".into(),
family: "parslee".into(),
version: "1".into(),
capabilities: vec![],
context_length: 128_000,
max_output_tokens: None,
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::Proprietary {
provider: "parslee".into(),
endpoint: "https://api.parslee.ai".into(),
auth: ProprietaryAuth::OAuth2Pkce {
authority: "https://api.parslee.ai".into(),
client_id: "parslee-car".into(),
scopes: vec!["inference:invoke".into()],
},
protocol: ProprietaryProtocol {
chat_path: "/api/v1/orgs/{orgId}/inference/responses".into(),
content_type: "application/json".into(),
streaming: false,
extra_headers: Default::default(),
wire: Default::default(),
},
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Curated,
deprecated: false,
available: true,
weights_ready: true,
};
let (endpoint, protocol) = extract_remote_endpoint(&schema).unwrap();
assert_eq!(endpoint, "https://api.parslee.ai");
assert_eq!(protocol, ApiProtocol::OpenAiCompat);
}
#[test]
fn rejected_parslee_env_override_names_the_override_and_remedy() {
let message = parslee_env_override_rejected_error().to_string();
assert!(message.contains("PARSLEE_ACCESS_TOKEN"), "{message}");
assert!(message.contains("replace or remove"), "{message}");
assert!(!message.contains("expired"), "{message}");
}
#[tokio::test]
async fn managed_parslee_authority_uses_one_resolved_credential() {
let _environment = crate::openrouter::test_environment_scope_async().await;
unsafe {
std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, "request-time-bearer");
std::env::set_var(
car_auth::PARSLEE_API_BASE_KEY,
"https://staging-api.parslee.ai/",
);
}
let schema = ModelSchema {
id: "parslee/openrouter/open-fast".into(),
name: "Parslee OpenRouter Fast".into(),
provider: "parslee".into(),
family: "openrouter".into(),
version: "1".into(),
capabilities: vec![],
context_length: 128_000,
max_output_tokens: None,
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::Proprietary {
provider: "parslee".into(),
endpoint: car_auth::DEFAULT_API_BASE.into(),
auth: ProprietaryAuth::OAuth2Pkce {
authority: car_auth::DEFAULT_API_BASE.into(),
client_id: "parslee-car".into(),
scopes: vec!["inference:invoke".into()],
},
protocol: ProprietaryProtocol {
chat_path: "/api/v1/orgs/{orgId}/inference/responses".into(),
content_type: "application/json".into(),
streaming: true,
extra_headers: Default::default(),
wire: Default::default(),
},
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Curated,
deprecated: false,
available: true,
weights_ready: true,
};
let (endpoint, protocol, lease) = RemoteBackend::new()
.resolve_request_authority(&schema)
.await
.unwrap();
assert_eq!(endpoint, "https://staging-api.parslee.ai");
assert_eq!(protocol, ApiProtocol::OpenAiCompat);
assert_eq!(lease.api_key, "request-time-bearer");
unsafe {
std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
}
}
#[tokio::test]
async fn reactive_retry_uses_api_base_from_cas_conflict_winner() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let rejected = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", "Bearer rejected-access"))
.respond_with(ResponseTemplate::new(401).set_body_string("expired"))
.expect(1)
.mount(&rejected)
.await;
let winner = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", "Bearer winning-access"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "winning-org"})),
)
.expect(1)
.mount(&winner)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.and(header("authorization", "Bearer winning-access"))
.respond_with(
ResponseTemplate::new(200).set_body_json(
serde_json::json!({"account": {"email": "winner@example.test"}}),
),
)
.expect(1)
.mount(&winner)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/winning-org/inference/responses"))
.and(header("authorization", "Bearer winning-access"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"event: response.output_text.delta\n",
"data: {\"delta\":\"winner endpoint\"}\n\n",
"event: response.completed\n",
"data: {\"response\":{\"usage\":{\"input_tokens\":3,\"output_tokens\":2}}}\n\n"
),
"text/event-stream",
))
.expect(1)
.mount(&winner)
.await;
let request = crate::protocol::ApiRequest {
model: "parslee/test".into(),
messages: vec![serde_json::json!({"role": "user", "content": "hello"})],
system: None,
system_stable_prefix: None,
temperature: 0.0,
max_tokens: 32,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
stream: false,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: None,
stop: Vec::new(),
reasoning_effort: None,
enable_thinking: None,
template_reasoning_effort: None,
};
let rejected_endpoint = rejected.uri();
let winning_credential = car_auth::ResolvedParsleeCredential {
access_token: "winning-access".into(),
api_base: winner.uri(),
expires_at: 9_999_999_999,
};
let response = RemoteBackend::new()
.execute_parslee_request_with_refresh(
&rejected_endpoint,
"rejected-access",
&request,
&mut |_| {},
|_| async move { car_auth::ReactiveRefreshResult::Refreshed(winning_credential) },
)
.await
.unwrap();
assert_eq!(response.text, "winner endpoint");
}
#[tokio::test]
async fn rejected_refresh_token_is_expired_once_then_backoff_stays_transient() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let home = crate::openrouter::StateRootScope::new();
let _restore = TestEnvRestore::capture(&[
"CAR_SECRETS_FILE_DIR",
PARSLEE_ACCESS_TOKEN_ENV,
car_auth::PARSLEE_API_BASE_KEY,
]);
let secrets = home.path().join("file-secrets");
unsafe {
std::env::set_var("CAR_SECRETS_FILE_DIR", &secrets);
std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
}
car_auth::invalidate_access_token_cache();
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
let bearer = "production-rejected-bearer";
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-backoff"})),
)
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200).set_body_json(
serde_json::json!({"account": {"email": "backoff@example.test"}}),
),
)
.expect(1)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-backoff/inference/responses"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(ResponseTemplate::new(401).set_body_string("expired"))
.expect(10)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/connect/token"))
.respond_with(
ResponseTemplate::new(401)
.set_body_json(serde_json::json!({"error": "invalid_grant"})),
)
.expect(1)
.mount(&server)
.await;
let active = serde_json::json!({
"account_id": "backoff-account",
"access_token": bearer,
"refresh_token": "invalid-refresh",
"expires_at": 9_999_999_999_u64,
"api_base": server.uri()
});
car_secrets::SecretStore::new()
.publish(
&car_secrets::SecretRef::with_default_service(
car_secrets::PARSLEE_AUTH_STATE_V2_KEY,
),
&serde_json::json!({
"schema": 2,
"revision": 1,
"generation": 1,
"active": active.clone(),
"accounts": [active]
})
.to_string(),
)
.unwrap();
let schema = crate::openrouter::curated_schemas()
.into_iter()
.find(|schema| schema.id == "parslee/openrouter/frontier-general")
.unwrap();
let request = crate::protocol::ApiRequest {
model: schema.id.clone(),
messages: vec![serde_json::json!({"role": "user", "content": "hello"})],
system: None,
system_stable_prefix: None,
temperature: 0.0,
max_tokens: 32,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
stream: false,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: None,
stop: Vec::new(),
reasoning_effort: None,
enable_thinking: None,
template_reasoning_effort: None,
};
let backend = RemoteBackend::new();
for attempt in 0..10 {
let error = backend
.execute_request(&schema, request.clone())
.await
.unwrap_err();
if attempt == 0 {
assert!(
matches!(
&error,
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::Expired { .. },
..
}
),
"the rejected refresh token must produce typed expiry: {error}"
);
assert!(
error
.to_string()
.contains("Your Parslee sign-in has expired. Sign in again to continue."),
"{error}"
);
} else {
assert!(
is_auth_rejection(&error),
"refresh cooldown must preserve the original retryable 401: {error}"
);
}
}
let requests = server.received_requests().await.unwrap();
assert_eq!(
requests
.iter()
.filter(|request| request.url.path() == "/connect/token")
.count(),
1,
"ten real inference 401 paths inside the interval must make one token-endpoint attempt"
);
car_auth::invalidate_access_token_cache();
}
#[test]
fn builtin_parslee_aliases_describe_shipped_responses_route() {
let catalog = crate::registry::builtin_catalog();
let aliases = ["parslee/fast", "parslee/reasoning", "parslee/advisor"];
for id in aliases {
let schema = catalog
.iter()
.find(|s| s.id == id)
.unwrap_or_else(|| panic!("builtin catalog is missing managed alias {id}"));
match &schema.source {
ModelSource::Proprietary { protocol, .. } => {
assert_eq!(
protocol.chat_path, "/api/v1/orgs/{orgId}/inference/responses",
"{id} must point at the shipped inference/responses route, \
not the retired /chat/stream stopgap"
);
}
other => panic!("{id} should be a Proprietary managed provider, got {other:?}"),
}
}
}
#[tokio::test(flavor = "current_thread")]
async fn managed_openrouter_stream_replays_provider_items_and_tool_history_on_second_turn() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
use crate::stream::StreamEvent;
use wiremock::matchers::{body_partial_json, header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
let bearer = "managed-stream-bearer";
unsafe {
std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, bearer);
std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, server.uri());
}
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-test"})),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"account": {"email": "user@example.test"}})),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-test/inference/responses"))
.and(body_partial_json(serde_json::json!({
"model": "parslee/openrouter/frontier-general",
"store": false,
"include": ["reasoning.encrypted_content"]
})))
.respond_with(ResponseTemplate::new(200).set_body_raw(
include_str!("../tests/fixtures/parslee-openrouter-reasoning-roundtrip.sse"),
"text/event-stream",
))
.expect(2)
.mount(&server)
.await;
let schema = crate::openrouter::curated_schemas()
.into_iter()
.find(|schema| schema.id == "parslee/openrouter/frontier-general")
.unwrap();
let mut rx = RemoteBackend::new()
.generate_stream(
&schema, "hello", None, None, 0.0, 32, None, None, None, None, None, None,
)
.await
.expect("managed stream should start");
let mut text = String::new();
let mut usage = None;
let mut provider_items = Vec::new();
while let Some(event) = rx.recv().await {
match event {
StreamEvent::TextDelta(delta) => text.push_str(&delta),
StreamEvent::ProviderOutputItem(item) => provider_items.push(item),
StreamEvent::Usage {
input_tokens,
output_tokens,
..
} => usage = Some((input_tokens, output_tokens)),
_ => {}
}
}
assert_eq!(text, "first answer");
assert_eq!(usage, Some((17, 9)));
let expected_reasoning = serde_json::json!({
"type": "reasoning",
"id": "rs_car_roundtrip",
"status": "completed",
"summary": [{"type": "summary_text", "text": "safe summary"}],
"encrypted_content": "opaque-encrypted-reasoning",
});
assert_eq!(provider_items, vec![expected_reasoning.clone()]);
let history = vec![
crate::tasks::generate::Message::User {
content: "first".into(),
},
crate::tasks::generate::Message::ProviderOutputItems {
protocol: crate::protocol::OPENAI_RESPONSES_PROTOCOL.into(),
items: provider_items,
},
crate::tasks::generate::Message::Assistant {
content: text,
tool_calls: vec![crate::tasks::generate::ToolCall {
id: Some("call-stream-history".into()),
name: "lookup".into(),
arguments: HashMap::from([(
"query".into(),
serde_json::Value::String("answer".into()),
)]),
}],
thinking: Vec::new(),
model_id: None,
local_last_resort: false,
},
crate::tasks::generate::Message::ToolResult {
tool_use_id: "call-stream-history".into(),
content: "tool output".into(),
provenance: Default::default(),
ok: None,
},
crate::tasks::generate::Message::User {
content: "continue".into(),
},
];
let mut second = RemoteBackend::new()
.generate_stream(
&schema,
"",
Some(&history),
None,
0.0,
32,
None,
None,
None,
None,
None,
None,
)
.await
.expect("managed second streaming turn should start");
while second.recv().await.is_some() {}
let requests = server.received_requests().await.unwrap();
let posts: Vec<serde_json::Value> = requests
.iter()
.filter(|request| {
request.method.as_str() == "POST"
&& request.url.path() == "/api/v1/orgs/org-test/inference/responses"
})
.map(|request| serde_json::from_slice(&request.body).unwrap())
.collect();
assert_eq!(posts.len(), 2);
let second_input = posts[1]["input"].as_array().unwrap();
assert!(second_input.contains(&expected_reasoning));
assert!(second_input.iter().any(|item| {
item["type"] == "function_call"
&& item["call_id"] == "call-stream-history"
&& item["name"] == "lookup"
}));
assert!(second_input.iter().any(|item| {
item["type"] == "function_call_output"
&& item["call_id"] == "call-stream-history"
&& item["output"] == "tool output"
}));
assert!(second_input
.iter()
.any(|item| item["role"] == "user" && item["content"] == "continue"));
unsafe {
std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
}
}
#[tokio::test(flavor = "current_thread")]
async fn managed_forced_tool_choice_reaches_the_wire_and_yields_a_tool_call() {
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
use wiremock::matchers::{body_partial_json, header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
let bearer = "managed-forced-tool-choice-bearer";
unsafe {
std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, bearer);
std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, server.uri());
}
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-test"})),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"account": {"email": "user@example.test"}})),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-test/inference/responses"))
.and(body_partial_json(
serde_json::json!({"tool_choice": "required"}),
))
.respond_with(ResponseTemplate::new(200).set_body_raw(
"event: response.output_item.added\n\
data: {\"output_index\":0,\"item\":{\"type\":\"function_call\",\"name\":\"extract\",\"call_id\":\"call_forced_1\"}}\n\n\
event: response.function_call_arguments.delta\n\
data: {\"output_index\":0,\"delta\":\"{\\\"city\\\":\\\"NYC\\\"}\"}\n\n\
event: response.completed\n\
data: {\"response\":{\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n",
"text/event-stream",
))
.expect(1)
.mount(&server)
.await;
let schema = crate::openrouter::curated_schemas()
.into_iter()
.find(|schema| schema.id == "parslee/openrouter/frontier-general")
.expect("managed frontier alias");
let tools = vec![serde_json::json!({
"name": "extract",
"description": "extract a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
})];
let (_text, calls, _thinking, _items, _usage, _stop) = RemoteBackend::new()
.generate_with_tools_multi(
&schema,
"which city",
None,
0.0,
64,
Some(&tools),
None,
None,
Some("required"),
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
)
.await
.expect("the forced managed request must reach the gateway and return a tool call");
assert_eq!(calls.len(), 1, "forced request must yield exactly one call");
assert_eq!(calls[0].name, "extract");
assert_eq!(
calls[0].arguments.get("city").and_then(|v| v.as_str()),
Some("NYC")
);
let requests = server.received_requests().await.unwrap();
let post = requests
.iter()
.find(|request| {
request.method.as_str() == "POST"
&& request.url.path() == "/api/v1/orgs/org-test/inference/responses"
})
.expect("the managed POST must have been made");
let body: serde_json::Value = serde_json::from_slice(&post.body).unwrap();
assert_eq!(body["tool_choice"], "required");
assert!(body["tools"].is_array());
unsafe {
std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY);
}
}
#[test]
fn build_request_tools_keeps_the_personal_chat_lane_nested() {
let schema = openrouter_schema("https://openrouter.test");
let handler = crate::protocol::handler_for(ApiProtocol::OpenRouter);
let tools = vec![serde_json::json!({
"name": "pick_number",
"description": "pick one",
"parameters": {"type": "object", "properties": {}},
})];
let built = build_request_tools(&schema, handler.as_ref(), Some(&tools))
.expect("tools were supplied");
assert_eq!(built[0]["type"], "function");
assert_eq!(
built[0]["function"]["name"], "pick_number",
"the chat lane must keep the nested shape: {}",
built[0]
);
assert!(
built[0].get("name").is_none(),
"a flat name on the chat lane would be the #1010 defect mirrored"
);
assert!(build_request_tools(&schema, handler.as_ref(), None).is_none());
}
#[tokio::test(flavor = "current_thread")]
async fn managed_lane_posts_flat_responses_tools_from_both_call_sites() {
use wiremock::matchers::{body_partial_json, header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
let _provider_env = crate::openrouter::test_environment_scope_async().await;
let _state_root = crate::openrouter::StateRootScope::new();
let bearer = "managed-flat-tools-bearer";
let _env = ManagedLaneEnvScope::set(bearer, &server.uri());
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.and(header("authorization", format!("Bearer {bearer}")))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"organizationId": "org-test"})),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({"account": {"email": "user@example.test"}})),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/api/v1/orgs/org-test/inference/responses"))
.and(body_partial_json(serde_json::json!({
"tools": [{"type": "function", "name": "pick_number"}]
})))
.respond_with(ResponseTemplate::new(200).set_body_raw(
"event: response.output_text.delta\n\
data: {\"delta\":\"ok\"}\n\n\
event: response.completed\n\
data: {\"response\":{\"usage\":{\"input_tokens\":5,\"output_tokens\":1}}}\n\n",
"text/event-stream",
))
.expect(2)
.mount(&server)
.await;
let schema = crate::openrouter::curated_schemas()
.into_iter()
.find(|schema| schema.id == "parslee/openrouter/frontier-general")
.expect("managed frontier alias");
let tools = vec![
serde_json::json!({
"type": "function",
"function": {
"name": "pick_number",
"description": "pick a number",
"parameters": {
"type": "object",
"properties": {"n": {"type": "integer"}},
"required": ["n"],
},
},
}),
serde_json::json!({
"name": "echo_text",
"description": "echo it back",
"parameters": {
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
}),
];
let (text, _calls, _thinking, _items, _usage, _stop) = RemoteBackend::new()
.generate_with_tools_multi(
&schema,
"pick one",
None,
0.0,
64,
Some(&tools),
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
)
.await
.expect("the managed non-streaming request must match the flat-tools mock");
assert_eq!(text, "ok");
let mut rx = RemoteBackend::new()
.generate_stream(
&schema,
"pick one",
None,
None,
0.0,
64,
Some(&tools),
None,
None,
None,
None,
None,
)
.await
.expect("the managed streaming request must match the flat-tools mock");
let mut streamed = String::new();
while let Some(event) = rx.recv().await {
if let crate::stream::StreamEvent::TextDelta(delta) = event {
streamed.push_str(&delta);
}
}
assert_eq!(streamed, "ok");
let requests = server.received_requests().await.unwrap();
let posts: Vec<serde_json::Value> = requests
.iter()
.filter(|request| {
request.method.as_str() == "POST"
&& request.url.path() == "/api/v1/orgs/org-test/inference/responses"
})
.map(|request| serde_json::from_slice(&request.body).unwrap())
.collect();
assert_eq!(posts.len(), 2, "both managed call sites must have posted");
for (index, body) in posts.iter().enumerate() {
let nested = &body["tools"][0];
assert_eq!(nested["type"], "function", "post {index}: {nested}");
assert_eq!(nested["name"], "pick_number", "post {index}: {nested}");
assert_eq!(
nested["description"], "pick a number",
"post {index}: {nested}"
);
assert_eq!(
nested["parameters"]["properties"]["n"]["type"], "integer",
"post {index}: {nested}"
);
assert!(
nested.get("function").is_none(),
"post {index} still carries the Chat-Completions nesting: {nested}"
);
let bare = &body["tools"][1];
assert_eq!(bare["type"], "function", "post {index}: {bare}");
assert_eq!(bare["name"], "echo_text", "post {index}: {bare}");
assert_eq!(bare["description"], "echo it back", "post {index}: {bare}");
assert_eq!(
bare["parameters"]["properties"]["text"]["type"], "string",
"post {index}: {bare}"
);
assert!(
bare.get("function").is_none(),
"post {index} wrapped a bare tool back into the nesting: {bare}"
);
}
}
struct ManagedLaneEnvScope {
previous_token: Option<std::ffi::OsString>,
previous_base: Option<std::ffi::OsString>,
}
impl ManagedLaneEnvScope {
fn set(token: &str, api_base: &str) -> Self {
let scope = Self {
previous_token: std::env::var_os(PARSLEE_ACCESS_TOKEN_ENV),
previous_base: std::env::var_os(car_auth::PARSLEE_API_BASE_KEY),
};
unsafe {
std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, token);
std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, api_base);
}
scope
}
}
impl Drop for ManagedLaneEnvScope {
fn drop(&mut self) {
unsafe {
match self.previous_token.take() {
Some(value) => std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, value),
None => std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV),
}
match self.previous_base.take() {
Some(value) => std::env::set_var(car_auth::PARSLEE_API_BASE_KEY, value),
None => std::env::remove_var(car_auth::PARSLEE_API_BASE_KEY),
}
}
}
}
#[tokio::test]
async fn resolved_authority_proprietary_oauth2_uses_env_override() {
let _provider_env = crate::openrouter::test_environment_scope_async().await;
std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, "test-bearer-abc123");
let schema = ModelSchema {
id: "parslee/advisor".into(),
name: "Parslee Advisor".into(),
provider: "parslee".into(),
family: "parslee".into(),
version: "1".into(),
capabilities: vec![],
context_length: 128_000,
max_output_tokens: None,
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::Proprietary {
provider: "parslee".into(),
endpoint: "https://api.parslee.ai".into(),
auth: ProprietaryAuth::OAuth2Pkce {
authority: "https://api.parslee.ai".into(),
client_id: "parslee-car".into(),
scopes: vec!["inference:invoke".into()],
},
protocol: ProprietaryProtocol {
chat_path: "/api/v1/orgs/{orgId}/inference/responses".into(),
content_type: "application/json".into(),
streaming: false,
extra_headers: Default::default(),
wire: Default::default(),
},
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Curated,
deprecated: false,
available: true,
weights_ready: true,
};
let (endpoint, protocol, lease) = RemoteBackend::new()
.resolve_request_authority(&schema)
.await
.expect("proprietary OAuth2 authority should resolve the env-override token");
assert_eq!(endpoint, "https://api.parslee.ai");
assert_eq!(protocol, ApiProtocol::OpenAiCompat);
assert_eq!(lease.api_key, "test-bearer-abc123");
std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
}
#[tokio::test]
async fn non_parslee_oauth_schema_cannot_lease_or_send_parslee_bearer() {
let _provider_env = crate::openrouter::test_environment_scope_async().await;
let server = wiremock::MockServer::start().await;
unsafe {
std::env::set_var(PARSLEE_ACCESS_TOKEN_ENV, "must-not-leave-process");
}
let schema = ModelSchema {
id: "community/custom-oauth".into(),
name: "Custom OAuth".into(),
provider: "community".into(),
family: "community".into(),
version: "1".into(),
capabilities: vec![crate::schema::ModelCapability::Generate],
context_length: 8_192,
max_output_tokens: None,
param_count: "api".into(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::Proprietary {
provider: "community".into(),
endpoint: server.uri(),
auth: ProprietaryAuth::OAuth2Pkce {
authority: "https://untrusted.example/authorize".into(),
client_id: "untrusted-client".into(),
scopes: vec!["inference".into()],
},
protocol: ProprietaryProtocol {
chat_path: "/custom/chat".into(),
content_type: "application/json".into(),
streaming: false,
extra_headers: Default::default(),
wire: Default::default(),
},
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Community,
deprecated: false,
available: true,
weights_ready: true,
};
let lease_error = RemoteBackend::new()
.lease_key(&schema, &server.uri())
.await
.expect_err("non-Parslee OAuth must be rejected before credential resolution");
assert!(
lease_error
.to_string()
.contains("OAuth2 PKCE credentials are supported only for provider 'parslee'"),
"{lease_error}"
);
let inference_error = RemoteBackend::new()
.generate(&schema, "must not send", None, 0.0, 16, None)
.await
.expect_err("explicit non-Parslee OAuth inference must fail closed");
assert!(
inference_error
.to_string()
.contains("OAuth2 PKCE credentials are supported only for provider 'parslee'"),
"{inference_error}"
);
assert!(
server.received_requests().await.unwrap().is_empty(),
"the rejected schema must not send the Parslee bearer or make any network request"
);
unsafe {
std::env::remove_var(PARSLEE_ACCESS_TOKEN_ENV);
}
}
#[tokio::test]
async fn chatgpt_subscription_gate_refuses_a_non_codex_provider() {
let schema: ModelSchema = serde_json::from_value(serde_json::json!({
"id": "openai/not-codex",
"name": "Not Codex",
"provider": "openai",
"family": "gpt",
"version": "1",
"capabilities": ["generate"],
"context_length": 8192,
"source": {
"type": "proprietary",
"provider": "openai",
"endpoint": "https://example.invalid",
"auth": {"type": "chatgpt_subscription"},
"protocol": {
"chat_path": "/responses",
"content_type": "application/json",
"streaming": true,
"extra_headers": {}
}
}
}))
.expect("deserialize non-Codex ChatGPT subscription schema");
let error = RemoteBackend::new()
.resolve_request_authority(&schema)
.await
.expect_err("ChatGPT subscription auth must reject non-Codex providers");
assert!(
error
.to_string()
.contains("supported only for provider 'openai-codex'"),
"{error}"
);
}
fn codex_schema_with_endpoint(endpoint: &str) -> ModelSchema {
serde_json::from_value(serde_json::json!({
"id": "openai-codex/gpt-5.6",
"name": "gpt-5.6",
"provider": "openai-codex",
"family": "gpt",
"version": "1",
"capabilities": ["generate"],
"context_length": 8192,
"source": {
"type": "proprietary",
"provider": "openai-codex",
"endpoint": endpoint,
"auth": {"type": "chatgpt_subscription"},
"protocol": {
"chat_path": "/codex/responses",
"content_type": "application/json",
"streaming": true,
"extra_headers": {}
}
}
}))
.expect("deserialize openai-codex schema")
}
#[tokio::test]
async fn chatgpt_subscription_gate_refuses_an_endpoint_off_the_codex_backend() {
for endpoint in [
"https://attacker.example/backend-api",
"http://chatgpt.com/backend-api",
"https://chatgpt.com.attacker.example/backend-api",
"https://chatgpt.com:8443/backend-api",
"http://127.0.0.1:8080",
] {
let error = RemoteBackend::new()
.resolve_request_authority(&codex_schema_with_endpoint(endpoint))
.await
.expect_err("ChatGPT credentials must not go to a custom endpoint");
assert!(
error.to_string().contains(codex::CODEX_BASE_URL),
"{endpoint}: {error}"
);
}
}
#[tokio::test]
async fn chatgpt_subscription_gate_resolves_the_codex_backend() {
for endpoint in [
"",
"https://chatgpt.com/backend-api",
"https://chatgpt.com/backend-api/",
] {
let (resolved, _, _) = RemoteBackend::new()
.resolve_request_authority(&codex_schema_with_endpoint(endpoint))
.await
.expect("the ChatGPT Codex backend is the credential's own destination");
assert_eq!(resolved, codex::CODEX_BASE_URL, "{endpoint:?}");
}
}
#[test]
fn parse_openai_embed_response() {
let json = r#"{"data":[{"embedding":[0.1,0.2,0.3]}]}"#;
let resp: OpenAiEmbedResponse = serde_json::from_str(json).unwrap();
assert_eq!(resp.data[0].embedding, vec![0.1, 0.2, 0.3]);
}
#[test]
fn request_model_name_uses_vllm_server_model() {
let schema = ModelSchema {
id: "vllm-mlx/test".into(),
name: "Display Name".into(),
provider: "test".into(),
family: "test".into(),
version: "1".into(),
capabilities: vec![],
context_length: 4096,
max_output_tokens: None,
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::VllmMlx {
endpoint: "http://localhost:8000".into(),
model_name: "mlx-community/Actual-Model".into(),
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Curated,
deprecated: false,
available: true,
weights_ready: true,
};
assert_eq!(request_model_name(&schema), "mlx-community/Actual-Model");
}
#[test]
fn managed_gateway_request_body_uses_canonical_id_when_display_name_drifts() {
let mut schema = crate::openrouter::curated_schemas()
.into_iter()
.find(|schema| schema.id == "parslee/openrouter/frontier-general")
.expect("managed frontier alias");
schema.name = "attacker-controlled-upstream-selector".into();
let selector = request_model_name(&schema);
let body = parslee_request_body(
"u1",
vec![serde_json::json!({"role": "user", "content": "hi"})],
8,
None,
None,
Some(&selector),
);
assert_eq!(selector, "parslee/openrouter/frontier-general");
assert_eq!(body["model"], "parslee/openrouter/frontier-general");
}
#[test]
fn exact_frontier_builtin_request_bodies_use_provider_native_selectors() {
fn request(model: String) -> crate::protocol::ApiRequest {
crate::protocol::ApiRequest {
model,
messages: vec![serde_json::json!({"role": "user", "content": "hi"})],
system: None,
system_stable_prefix: None,
temperature: 0.0,
max_tokens: 8,
tools: None,
tool_choice: None,
parallel_tool_calls: None,
stream: false,
budget_tokens: 0,
cache_control: false,
cache_ttl: crate::tasks::generate::CacheTtl::default(),
response_format: None,
stop: Vec::new(),
reasoning_effort: None,
enable_thinking: None,
template_reasoning_effort: None,
}
}
let catalog = crate::registry::builtin_catalog();
let openai = catalog
.iter()
.find(|schema| schema.id == "openai/gpt-5.5-2026-04-23")
.expect("exact OpenAI production row");
let openai_selector = request_model_name(openai);
let openai_body = crate::protocol::OpenAiResponsesHandler
.build_request_body(&request(openai_selector.clone()));
assert_eq!(openai_selector, "gpt-5.5-2026-04-23");
assert_eq!(openai_body["model"], "gpt-5.5-2026-04-23");
let anthropic = catalog
.iter()
.find(|schema| schema.id == "anthropic/claude-opus-4-8")
.expect("exact Anthropic production row");
let anthropic_selector = request_model_name(anthropic);
let anthropic_body = crate::protocol::AnthropicHandler
.build_request_body(&request(anthropic_selector.clone()));
assert_eq!(anthropic_selector, "claude-opus-4-8");
assert_eq!(anthropic_body["model"], "claude-opus-4-8");
}
#[test]
fn truncate_prompt_fits_returns_unchanged() {
let prompt = "short prompt";
let result = truncate_prompt_to_fit(prompt, None, None, 16, 0, 256);
assert_eq!(result, prompt);
}
#[test]
fn truncate_prompt_cjk_mid_codepoint_does_not_panic() {
let prompt: String = std::iter::repeat_n('\u{4E16}', 200).collect();
let result = truncate_prompt_to_fit(&prompt, None, None, 20, 0, 209);
assert!(result.starts_with("[...truncated...]"));
let kept = result.strip_prefix("[...truncated...]\n").unwrap();
assert!(!kept.is_empty());
}
#[test]
fn truncate_prompt_accounts_for_context_and_tools() {
let prompt = "line one\nline two\nline three\n".repeat(50);
let tools = vec![serde_json::json!({"name": "demo_tool"})];
let result = truncate_prompt_to_fit(&prompt, Some("ctx"), Some(&tools), 20, 0, 240);
assert!(result.starts_with("[...truncated...]"));
}
#[test]
fn truncate_prompt_reserves_media_tokens() {
let prompt = "word ".repeat(80);
let overflowing_history = 10_000;
assert_eq!(
truncate_prompt_to_fit("", None, None, 20, overflowing_history, 500),
""
);
let untouched = truncate_prompt_to_fit(&prompt, None, None, 20, 0, 500);
assert_eq!(untouched, prompt);
let truncated = truncate_prompt_to_fit(&prompt, None, None, 20, 300, 500);
assert!(truncated.starts_with("[...truncated...]"));
}
#[test]
fn parslee_responses_sse_concatenates_text_and_usage() {
let raw = "event: response.output_text.delta\ndata: {\"delta\":\"Hello\"}\n\n\
event: response.output_text.delta\ndata: {\"delta\":\", world\"}\n\n\
event: response.completed\ndata: {\"response\":{\"usage\":{\"input_tokens\":10,\"output_tokens\":3}}}\n\n";
let mut acc = crate::stream::StreamAccumulator::default();
for ev in parse_parslee_responses_sse(raw) {
acc.push(&ev);
}
let (text, tool_calls, usage, _stop) = acc.finish_with_usage();
assert_eq!(text, "Hello, world");
assert!(tool_calls.is_empty());
let usage = usage.expect("usage reported");
assert_eq!(usage.prompt_tokens, 10);
assert_eq!(usage.completion_tokens, 3);
}
#[test]
fn parslee_responses_sse_parses_crlf_event_blocks() {
let raw = concat!(
"event: response.output_text.delta\r\n",
"data: {\"delta\":\r\n",
"data: \"Hello\"}\r\n\r\n",
"event: response.completed\r\n",
"data: {\"response\":{\"usage\":{\"input_tokens\":4,\"output_tokens\":1}}}\r\n\r\n",
);
let mut accumulator = crate::stream::StreamAccumulator::default();
for event in parse_parslee_responses_sse(raw) {
accumulator.push(&event);
}
let (text, _, usage, _) = accumulator.finish_with_usage();
assert_eq!(text, "Hello");
assert_eq!(usage.expect("usage reported").completion_tokens, 1);
}
#[test]
fn parslee_responses_sse_parses_tool_call() {
let raw = "event: response.output_item.added\n\
data: {\"output_index\":0,\"item\":{\"type\":\"function_call\",\"name\":\"get_weather\",\"call_id\":\"call_1\"}}\n\n\
event: response.function_call_arguments.delta\n\
data: {\"output_index\":0,\"delta\":\"{\\\"city\\\":\\\"NYC\\\"}\"}\n\n\
event: response.completed\ndata: {\"response\":{\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n\n";
let mut acc = crate::stream::StreamAccumulator::default();
for ev in parse_parslee_responses_sse(raw) {
acc.push(&ev);
}
let (_text, tool_calls) = acc.finish();
assert_eq!(tool_calls.len(), 1);
assert_eq!(tool_calls[0].name, "get_weather");
assert_eq!(tool_calls[0].id.as_deref(), Some("call_1"));
assert_eq!(
tool_calls[0].arguments.get("city").and_then(|v| v.as_str()),
Some("NYC")
);
}
#[test]
fn parslee_responses_sse_uses_completed_tool_arguments_without_deltas() {
let raw = concat!(
"event: response.output_item.added\n",
"data: {\"sequence_number\":1,\"output_index\":0,\"item\":{\"id\":\"rs_1\",\"type\":\"reasoning\",\"status\":\"in_progress\",\"summary\":[]}}\n\n",
"event: response.output_item.done\n",
"data: {\"sequence_number\":2,\"output_index\":0,\"item\":{\"id\":\"rs_1\",\"type\":\"reasoning\",\"status\":\"completed\",\"summary\":[],\"encrypted_content\":\"opaque\"}}\n\n",
"event: response.output_item.added\n",
"data: {\"sequence_number\":3,\"output_index\":1,\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"status\":\"in_progress\",\"name\":\"read_file\",\"call_id\":\"call_1\",\"arguments\":\"\"}}\n\n",
"event: response.function_call_arguments.done\n",
"data: {\"sequence_number\":4,\"output_index\":1,\"item_id\":\"fc_1\",\"arguments\":\"{\\\"path\\\":\\\"src/lib.rs\\\"}\"}\n\n",
"event: response.output_item.added\n",
"data: {\"sequence_number\":5,\"output_index\":2,\"item\":{\"id\":\"fc_2\",\"type\":\"function_call\",\"status\":\"completed\",\"name\":\"grep_files\",\"call_id\":\"call_2\",\"arguments\":\"{\\\"query\\\":\\\"TODO\\\",\\\"path\\\":\\\"src\\\"}\"}}\n\n",
"event: response.output_item.done\n",
"data: {\"sequence_number\":6,\"output_index\":2,\"item\":{\"id\":\"fc_2\",\"type\":\"function_call\",\"status\":\"completed\",\"name\":\"grep_files\",\"call_id\":\"call_2\",\"arguments\":\"{\\\"query\\\":\\\"TODO\\\",\\\"path\\\":\\\"src\\\"}\"}}\n\n",
"event: response.output_item.added\n",
"data: {\"sequence_number\":7,\"output_index\":3,\"item\":{\"id\":\"fc_3\",\"type\":\"function_call\",\"status\":\"completed\",\"name\":\"shell\",\"call_id\":\"call_3\",\"arguments\":\"{\\\"command\\\":\\\"cargo test\\\"}\"}}\n\n",
"event: response.completed\n",
"data: {\"sequence_number\":8,\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":20,\"output_tokens\":8}}}\n\n",
);
let mut accumulator = crate::stream::StreamAccumulator::default();
for event in parse_parslee_responses_sse(raw) {
accumulator.push(&event);
}
let (_, tool_calls) = accumulator.finish();
assert_eq!(tool_calls.len(), 3);
assert_eq!(tool_calls[0].id.as_deref(), Some("call_1"));
assert_eq!(tool_calls[0].arguments["path"], "src/lib.rs");
assert_eq!(tool_calls[1].id.as_deref(), Some("call_2"));
assert_eq!(tool_calls[1].arguments["query"], "TODO");
assert_eq!(tool_calls[1].arguments["path"], "src");
assert_eq!(tool_calls[2].id.as_deref(), Some("call_3"));
assert_eq!(tool_calls[2].arguments["command"], "cargo test");
}
#[test]
fn parslee_responses_sse_routes_interleaved_parallel_tool_arguments() {
let raw = concat!(
"event: response.output_item.added\n",
"data: {\"sequence_number\":1,\"output_index\":1,\"item\":{\"id\":\"fc_read\",\"type\":\"function_call\",\"status\":\"in_progress\",\"name\":\"read_file\",\"call_id\":\"call_read\",\"arguments\":\"\"}}\n\n",
"event: response.output_item.added\n",
"data: {\"sequence_number\":2,\"output_index\":2,\"item\":{\"id\":\"fc_grep\",\"type\":\"function_call\",\"status\":\"in_progress\",\"name\":\"grep_files\",\"call_id\":\"call_grep\",\"arguments\":\"\"}}\n\n",
"event: response.function_call_arguments.delta\n",
"data: {\"sequence_number\":3,\"output_index\":1,\"item_id\":\"fc_read\",\"delta\":\"{\\\"path\\\":\\\"wrong\"}\"}\n\n",
"event: response.function_call_arguments.delta\n",
"data: {\"sequence_number\":4,\"output_index\":2,\"item_id\":\"fc_grep\",\"delta\":\"{\\\"query\\\":\\\"OLD\\\",\"}\n\n",
"event: response.function_call_arguments.delta\n",
"data: {\"sequence_number\":5,\"output_index\":2,\"item_id\":\"fc_grep\",\"delta\":\"\\\"path\\\":\\\"wrong\\\"}\"}\n\n",
"event: response.function_call_arguments.done\n",
"data: {\"sequence_number\":6,\"output_index\":2,\"item_id\":\"fc_grep\",\"arguments\":\"{\\\"query\\\":\\\"TODO\\\",\\\"path\\\":\\\"src\\\"}\"}\n\n",
"event: response.function_call_arguments.done\n",
"data: {\"sequence_number\":7,\"output_index\":1,\"item_id\":\"fc_read\",\"arguments\":\"{\\\"path\\\":\\\"src/lib.rs\\\"}\"}\n\n",
"event: response.completed\n",
"data: {\"sequence_number\":8,\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":20,\"output_tokens\":8}}}\n\n",
);
let mut accumulator = crate::stream::StreamAccumulator::default();
for event in parse_parslee_responses_sse(raw) {
accumulator.push(&event);
}
let (_, tool_calls) = accumulator.finish();
assert_eq!(tool_calls.len(), 2);
assert_eq!(tool_calls[0].id.as_deref(), Some("call_read"));
assert_eq!(tool_calls[0].arguments["path"], "src/lib.rs");
assert_eq!(tool_calls[1].id.as_deref(), Some("call_grep"));
assert_eq!(tool_calls[1].arguments["query"], "TODO");
assert_eq!(tool_calls[1].arguments["path"], "src");
}
#[test]
fn parslee_responses_sse_ignores_unparseable_frames() {
let raw = "event: response.output_text.delta\ndata: not-json\n\n\
event: response.output_text.delta\ndata: {\"delta\":\"ok\"}\n\n";
let mut acc = crate::stream::StreamAccumulator::default();
for ev in parse_parslee_responses_sse(raw) {
acc.push(&ev);
}
assert_eq!(acc.finish().0, "ok");
}
#[tokio::test]
async fn parslee_identity_resolves_org_and_user() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({ "organizationId": "org_test" })),
)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200).set_body_json(
serde_json::json!({ "account": { "email": "user@example.com" } }),
),
)
.mount(&server)
.await;
let backend = RemoteBackend::new();
let (org, user) = backend
.parslee_identity(&server.uri(), "ident-ok-bearer", &mut |_| {})
.await
.expect("identity should resolve");
assert_eq!(org, "org_test");
assert_eq!(user, "user@example.com");
}
#[tokio::test]
async fn a_403_org_lookup_is_access_denied_not_a_sign_in_failure() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
crate::parslee_credential::clear_credential_rejected();
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(ResponseTemplate::new(403).set_body_string("no access"))
.mount(&server)
.await;
let err = RemoteBackend::new()
.parslee_identity(&server.uri(), "ident-403-bearer", &mut |_| {})
.await
.expect_err("a 403 org lookup must fail");
let msg = err.to_string();
assert!(msg.contains("denied this account access"), "{msg}");
assert!(!crate::is_auth_failure_message(&msg), "{msg}");
assert!(
crate::parslee_credential::credential_rejected(),
"a 403 still makes the lane unusable for routing"
);
crate::parslee_credential::clear_credential_rejected();
}
#[tokio::test]
async fn a_401_org_lookup_records_the_credential_as_rejected() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
crate::parslee_credential::clear_credential_rejected();
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(ResponseTemplate::new(401).set_body_string("Authentication required"))
.mount(&server)
.await;
let backend = RemoteBackend::new();
let err = backend
.parslee_identity(&server.uri(), "ident-401-bearer", &mut |_| {})
.await
.expect_err("a 401 org lookup must fail");
assert!(
matches!(
err,
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::Expired { .. },
..
}
),
"a 401 org lookup must be a TYPED credential failure, not an anonymous \
inference error — the assistant loop's `auth_required_reason` only maps \
`CredentialUnavailable`, so an untyped error is why an expired sign-in \
surfaced as a generic model failure (car#1839). Got: {err:?}"
);
assert!(
crate::parslee_credential::credential_rejected(),
"a 401 from the org lookup must record the credential as rejected — \
without it the registry keeps advertising a lane that 401s on every \
call (#887)"
);
crate::parslee_credential::clear_credential_rejected();
let flaky = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(ResponseTemplate::new(503).set_body_string("upstream down"))
.mount(&flaky)
.await;
let _ = backend
.parslee_identity(&flaky.uri(), "ident-503-bearer", &mut |_| {})
.await
.expect_err("a 503 org lookup must still fail");
assert!(
!crate::parslee_credential::credential_rejected(),
"a 5xx says the server is unwell, not that this credential is dead"
);
crate::parslee_credential::note_credential_rejected();
let good = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({ "organizationId": "org_recovered" })),
)
.mount(&good)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({ "account": { "email": "u@example.com" } })),
)
.mount(&good)
.await;
let (org, _) = backend
.parslee_identity(&good.uri(), "ident-recovered-bearer", &mut |_| {})
.await
.expect("identity should resolve");
assert_eq!(org, "org_recovered");
assert!(
!crate::parslee_credential::credential_rejected(),
"a successful org lookup proves the credential works and must clear \
the verdict"
);
crate::parslee_credential::clear_credential_rejected();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn parslee_identity_does_not_cache_a_placeholder_after_a_session_transport_failure() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind stand-in gateway");
let endpoint = format!("http://{}", listener.local_addr().expect("gateway address"));
let server = tokio::spawn(async move {
let mut sessions = 0;
loop {
let (mut socket, _) = listener.accept().await.expect("accept request");
let mut head = Vec::new();
while !head.windows(4).any(|window| window == b"\r\n\r\n") {
let mut chunk = [0_u8; 1_024];
let read = socket.read(&mut chunk).await.expect("read request");
assert!(read > 0, "request ended before its head");
head.extend_from_slice(&chunk[..read]);
}
let head = String::from_utf8_lossy(&head);
let body = if head.starts_with("GET /api/v1/organizations/me ") {
r#"{"organizationId":"org-session"}"#
} else if head.starts_with("GET /connect/session ") {
sessions += 1;
if sessions == 1 {
continue;
}
r#"{"account":{"email":"session@example.test"}}"#
} else {
panic!("unexpected request: {head}");
};
let response = format!(
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len()
);
socket
.write_all(response.as_bytes())
.await
.expect("write response");
let _ = socket.shutdown().await;
}
});
let backend = RemoteBackend::new();
let bearer = "session-blip-bearer";
let first = backend
.parslee_identity(&endpoint, bearer, &mut |_| {})
.await
.expect("a session blip still resolves the org");
assert_eq!(first, ("org-session".to_string(), "car".to_string()));
let second = backend
.parslee_identity(&endpoint, bearer, &mut |_| {})
.await
.expect("the second lookup resolves");
server.abort();
assert_eq!(second.1, "session@example.test");
}
#[tokio::test]
async fn parslee_identity_cache_is_isolated_by_api_base() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
let first = MockServer::start().await;
let second = MockServer::start().await;
for (server, org) in [(&first, "org-first"), (&second, "org-second")] {
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({ "organizationId": org })),
)
.mount(server)
.await;
Mock::given(method("GET"))
.and(path("/connect/session"))
.respond_with(ResponseTemplate::new(200).set_body_json(
serde_json::json!({ "account": { "email": "same@example.test" } }),
))
.mount(server)
.await;
}
let backend = RemoteBackend::new();
let first_identity = backend
.parslee_identity(&first.uri(), "same-bearer-across-environments", &mut |_| {})
.await
.unwrap();
let second_identity = backend
.parslee_identity(
&second.uri(),
"same-bearer-across-environments",
&mut |_| {},
)
.await
.unwrap();
assert_eq!(first_identity.0, "org-first");
assert_eq!(second_identity.0, "org-second");
}
#[tokio::test]
async fn parslee_identity_surfaces_org_lookup_401() {
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _environment = crate::openrouter::test_environment_scope_async().await;
let _home = crate::openrouter::StateRootScope::new();
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/api/v1/organizations/me"))
.respond_with(ResponseTemplate::new(401).set_body_string("Authentication required"))
.mount(&server)
.await;
let backend = RemoteBackend::new();
let err = backend
.parslee_identity(&server.uri(), "ident-401-bearer", &mut |_| {})
.await
.expect_err("a 401 org lookup must error");
let msg = err.to_string();
assert!(msg.contains("org lookup failed"), "got: {msg}");
assert!(msg.contains("HTTP 401"), "got: {msg}");
assert!(
is_auth_rejection(&err),
"401 must be classified as auth rejection"
);
assert!(
matches!(
err,
InferenceError::CredentialUnavailable {
reason: crate::CredentialFailure::Expired { .. },
..
}
),
"a 401 org lookup must be a TYPED credential failure: {err:?}"
);
}
use crate::tls_client::test_seam::{
mixed_cert_pem, TrustStoreScope, UNPARSEABLE_CERT_PEM, VALID_CERT_PEM,
};
use crate::tls_client::{TrustFallback, HUGGINGFACE_PROBE, REMOTE_BACKEND, VLLM_HEALTH_CHECK};
fn closed_loopback_url() -> String {
format!("{}/", closed_loopback_endpoint())
}
fn closed_loopback_endpoint() -> String {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
let port = listener.local_addr().expect("local addr").port();
drop(listener);
format!("https://127.0.0.1:{port}")
}
#[tokio::test]
async fn empty_and_mixed_trust_stores_build_without_degrading() {
let mut scope = TrustStoreScope::acquire_async("").await;
crate::tls_client::reset_sites_for_test();
let backend = RemoteBackend::new();
assert!(
backend.tls_degradation.is_none(),
"an empty certificate store must not degrade anything"
);
scope.repoint(&mixed_cert_pem());
crate::tls_client::reset_sites_for_test();
let backend = RemoteBackend::new();
assert!(
backend.tls_degradation.is_none(),
"one readable certificate alongside an unreadable one must not degrade"
);
assert!(
crate::tls_client::last_degradation(&REMOTE_BACKEND).is_none(),
"nothing degraded, so nothing should have been recorded or logged"
);
}
#[tokio::test]
async fn remote_backend_degrades_to_public_cas_instead_of_panicking() {
let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
scope.assert_breaks_client_construction();
crate::tls_client::reset_sites_for_test();
let backend = RemoteBackend::new();
let degradation = backend
.tls_degradation
.clone()
.expect("a broken trust store must be recorded, not swallowed");
assert_eq!(degradation.fallback, TrustFallback::PublicCaOnly);
assert!(
!degradation.source.is_empty(),
"the original reqwest builder error must be kept"
);
let err = backend
.client
.get(closed_loopback_url())
.send()
.await
.expect_err("nothing listens on that port");
assert!(
err.is_connect(),
"expected a connect-class error, got: {err}"
);
let surfaced = backend
.request_error("remote call failed", &err)
.to_string();
assert!(
surfaced.contains("the OS certificate store failed to load"),
"got: {surfaced}"
);
assert!(
surfaced.contains("restart the daemon"),
"the operator must be told how to recover; got: {surfaced}"
);
}
#[tokio::test]
async fn degraded_backend_carries_its_reason_into_real_inference_requests() {
const KEY_ENV: &str = "CAR_TLS_DEGRADATION_TEST_KEY";
let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
scope.assert_breaks_client_construction();
crate::tls_client::reset_sites_for_test();
std::env::set_var(KEY_ENV, "test-key");
let backend = RemoteBackend::new();
assert!(
backend.tls_degradation.is_some(),
"setup: must have degraded"
);
let schema = remote_stream_schema(
&closed_loopback_endpoint(),
ApiProtocol::OpenAiCompat,
"test-model",
KEY_ENV,
);
let mut parslee_schema = remote_stream_schema(
&closed_loopback_endpoint(),
ApiProtocol::OpenAiCompat,
"parslee/test-model",
KEY_ENV,
);
parslee_schema.provider = "parslee".into();
parslee_schema.source = ModelSource::Proprietary {
provider: "parslee".into(),
endpoint: closed_loopback_endpoint(),
auth: ProprietaryAuth::BearerTokenEnv {
env_var: KEY_ENV.into(),
},
protocol: ProprietaryProtocol::default(),
};
let (streamed, completed, managed) = tokio::join!(
backend.generate_stream(
&schema, "hello", None, None, 0.0, 32, None, None, None, None, None, None,
),
backend.generate(&schema, "hello", None, 0.0, 32, None),
backend.generate(&parslee_schema, "hello", None, 0.0, 32, None),
);
let streamed = streamed
.expect_err("nothing listens on that port")
.to_string();
assert!(
streamed.contains("the OS certificate store failed to load"),
"the streaming inference path dropped the degradation reason: {streamed}"
);
assert!(
streamed.contains("restart the daemon"),
"the operator must be told how inference recovers: {streamed}"
);
let completed = completed
.expect_err("nothing listens on that port")
.to_string();
assert!(
completed.contains("the OS certificate store failed to load"),
"the completion inference path dropped the degradation reason: {completed}"
);
let managed = managed
.expect_err("the managed identity lookup cannot connect")
.to_string();
assert!(
managed.contains("the OS certificate store failed to load"),
"the managed inference path dropped the degradation reason: {managed}"
);
std::env::remove_var(KEY_ENV);
}
#[tokio::test]
async fn remote_backend_default_matches_new() {
let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
scope.assert_breaks_client_construction();
crate::tls_client::reset_sites_for_test();
let explicit = RemoteBackend::new();
let defaulted = RemoteBackend::default();
assert_eq!(
explicit.tls_degradation.map(|d| d.fallback),
defaulted.tls_degradation.map(|d| d.fallback),
"`Default` delegates to `new`, so the two must agree on degradation"
);
}
#[tokio::test]
async fn tls_floor_rung_builds_under_a_broken_trust_store() {
let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
scope.assert_breaks_client_construction();
assert!(
reqwest::Client::builder()
.tls_built_in_native_certs(false)
.build()
.is_ok(),
"rung 2 (built-in public CAs only) must build"
);
assert!(
reqwest::Client::builder()
.tls_built_in_root_certs(false)
.build()
.is_ok(),
"rung 3 (the floor) must build — it is what keeps construction panic-free"
);
}
#[tokio::test]
async fn plain_transport_failures_are_not_reported_as_certificate_problems() {
assert!(is_trust_related(true, false));
assert!(
!is_trust_related(true, true),
"a connect *timeout* is not a trust problem"
);
assert!(!is_trust_related(false, true));
assert!(!is_trust_related(false, false));
let scope = TrustStoreScope::acquire_async(VALID_CERT_PEM).await;
let healthy = RemoteBackend::new();
assert!(healthy.tls_degradation.is_none());
let err = healthy
.client
.get(closed_loopback_url())
.send()
.await
.expect_err("nothing listens on that port");
let surfaced = healthy
.request_error("remote call failed", &err)
.to_string();
assert!(
!surfaced.contains("certificate store"),
"a healthy backend must not blame certificates; got: {surfaced}"
);
drop(scope);
let scope = TrustStoreScope::acquire_async(UNPARSEABLE_CERT_PEM).await;
scope.assert_breaks_client_construction();
crate::tls_client::reset_sites_for_test();
let degraded = RemoteBackend::new();
assert!(degraded.tls_degradation.is_some());
let server = wiremock::MockServer::start().await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/ok"))
.respond_with(wiremock::ResponseTemplate::new(200).set_body_string("fine"))
.mount(&server)
.await;
wiremock::Mock::given(wiremock::matchers::method("GET"))
.and(wiremock::matchers::path("/denied"))
.respond_with(wiremock::ResponseTemplate::new(401))
.mount(&server)
.await;
let ok = degraded
.client
.get(format!("{}/ok", server.uri()))
.send()
.await
.expect("a degraded client still works against reachable endpoints");
assert!(ok.status().is_success());
assert_eq!(
ok.text().await.expect("body"),
"fine",
"a successful request carries no degradation text"
);
let denied = degraded
.client
.get(format!("{}/denied", server.uri()))
.send()
.await
.expect("a 401 is a response, not a transport failure");
assert_eq!(denied.status().as_u16(), 401);
}
#[test]
fn trust_store_scope_restores_the_previous_override_even_on_panic() {
const FILE_VAR: &str = "SSL_CERT_FILE";
const DIR_VAR: &str = "SSL_CERT_DIR";
let lock = crate::openrouter::test_environment_scope();
let before_file = std::env::var_os(FILE_VAR);
let before_dir = std::env::var_os(DIR_VAR);
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _scope = TrustStoreScope::with_lock(lock, UNPARSEABLE_CERT_PEM);
assert!(
std::env::var_os(FILE_VAR).is_some(),
"the scope must install the override"
);
assert!(
std::env::var_os(DIR_VAR).is_none(),
"the scope must suppress SSL_CERT_DIR so host roots cannot make \
the failure fixture pass"
);
panic!("simulated mid-test failure");
}));
assert!(outcome.is_err(), "the simulated failure must have unwound");
let _relock = crate::openrouter::test_environment_scope();
assert_eq!(
std::env::var_os(FILE_VAR),
before_file,
"the guard must restore the previous SSL_CERT_FILE value"
);
assert_eq!(
std::env::var_os(DIR_VAR),
before_dir,
"the guard must restore the previous SSL_CERT_DIR value so a \
panicking test cannot poison the rest of the suite"
);
}
#[test]
fn every_site_warns_once_and_keeps_its_own_record() {
let scope = TrustStoreScope::acquire_blocking(UNPARSEABLE_CERT_PEM);
scope.assert_breaks_client_construction();
static RESTART_SITE: crate::tls_client::Site = crate::tls_client::Site::new(
"test site built once per daemon",
"Inference does not re-read the store while the daemon is running: repair the \
certificate store and restart the daemon to recover.",
);
static SELF_HEALING_A: crate::tls_client::Site = crate::tls_client::Site::new(
"test site rebuilt per probe",
"This client is rebuilt on every probe, so repairing the certificate store restores \
it with no restart.",
);
static SELF_HEALING_B: crate::tls_client::Site = crate::tls_client::Site::new(
"test site rebuilt per health check",
"This client is rebuilt on every health check, so repairing the certificate store \
restores it with no restart.",
);
let sites: [(&'static crate::tls_client::Site, &str); 3] = [
(&RESTART_SITE, "restart the daemon to recover"),
(&SELF_HEALING_A, "with no restart"),
(&SELF_HEALING_B, "with no restart"),
];
for (index, (site, expected_recovery)) in sites.iter().enumerate() {
for attempt in 1..=2 {
let (_client, degradation) = crate::tls_client::build_client_with_degradation(
site,
reqwest::ClientBuilder::new,
);
let degradation = degradation.unwrap_or_else(|| {
panic!(
"{} must degrade on attempt {attempt}, not build cleanly against the \
broken store",
site.name()
)
});
assert_eq!(
degradation.fallback,
TrustFallback::PublicCaOnly,
"{} must land on rung 2, not the floor",
site.name()
);
assert!(
!degradation.source.is_empty(),
"{} must carry the underlying reqwest error",
site.name()
);
let rendered = degradation.to_string();
assert!(
rendered.contains("the OS certificate store failed to load"),
"every degradation must name the underlying error; got: {rendered}"
);
assert!(
rendered.contains(expected_recovery),
"{} must carry its own recovery advice; got: {rendered}",
site.name()
);
if *expected_recovery == "with no restart" {
assert!(
!rendered.contains("restart the daemon"),
"{} self-heals — telling the operator to restart is wrong advice: \
{rendered}",
site.name()
);
}
}
assert_eq!(
site.warn_count(),
1,
"{} degraded twice and must have warned exactly once",
site.name()
);
for (later, _) in sites.iter().skip(index + 1) {
assert_eq!(
later.warn_count(),
0,
"{} warned because {} did — the limiter must be per site, not per process",
later.name(),
site.name()
);
}
}
assert!(
REMOTE_BACKEND
.recovery()
.contains("restart the daemon to recover"),
"inference recovers only on a restart and must say so: {}",
REMOTE_BACKEND.recovery()
);
for site in [&HUGGINGFACE_PROBE, &VLLM_HEALTH_CHECK] {
let recovery = site.recovery();
assert!(
recovery.contains("with no restart"),
"{} rebuilds its client per call, so it must not demand a restart: {recovery}",
site.name()
);
assert!(
!recovery.contains("restart the daemon"),
"{} self-heals — telling the operator to restart is wrong advice: {recovery}",
site.name()
);
}
}
#[tokio::test(flavor = "current_thread")]
async fn remote_text_is_cut_at_the_callers_stop_sequence() {
use wiremock::matchers::{body_partial_json, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _credential_scope = crate::openrouter::test_credential_scope();
crate::openrouter::set_test_credential(Some("stop-key"));
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(body_partial_json(serde_json::json!({"stop": ["STOPHERE"]})))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{
"message": {"role": "assistant", "content": "Paris STOPHERE indeed"},
"finish_reason": "length",
}],
"usage": {"prompt_tokens": 9, "completion_tokens": 4, "total_tokens": 13},
})))
.expect(1)
.mount(&server)
.await;
let schema = openrouter_schema(&server.uri());
let controls = crate::protocol::OutputControls {
stop: vec!["STOPHERE".into()],
reasoning_effort: None,
..Default::default()
};
let (text, _calls, _thinking, _items, usage, stop_reason) = RemoteBackend::new()
.generate_with_tools_multi_observed(
&schema,
"Capital of France?",
None,
0.0,
64,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
&controls,
&mut |_| {},
)
.await
.expect("the stop-carrying request reaches the provider");
assert_eq!(text, "Paris ");
assert_eq!(stop_reason.as_deref(), Some("stop_sequence"));
assert_eq!(usage.map(|usage| usage.completion_tokens), Some(4));
}
#[tokio::test(flavor = "current_thread")]
async fn streamed_text_is_cut_at_a_stop_sequence_split_across_deltas() {
use crate::stream::StreamEvent;
use wiremock::matchers::{body_partial_json, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let _credential_scope = crate::openrouter::test_credential_scope();
crate::openrouter::set_test_credential(Some("stream-stop-key"));
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(body_partial_json(serde_json::json!({"stop": ["STOPHERE"]})))
.respond_with(ResponseTemplate::new(200).set_body_raw(
concat!(
"data: {\"choices\":[{\"delta\":{\"content\":\"Paris ST\"}}]}\n\n",
"data: {\"choices\":[{\"delta\":{\"content\":\"OPHERE indeed\"}}]}\n\n",
"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
),
"text/event-stream",
))
.expect(1)
.mount(&server)
.await;
let schema = openrouter_schema(&server.uri());
let controls = crate::protocol::OutputControls {
stop: vec!["STOPHERE".into()],
reasoning_effort: None,
..Default::default()
};
let mut receiver = RemoteBackend::new()
.generate_stream_with_controls(
&schema, "stream", None, None, 0.0, 32, None, None, None, None, None, None,
&controls,
)
.await
.expect("stream opens");
let mut events = Vec::new();
while let Some(event) = receiver.recv().await {
events.push(event);
}
let streamed: String = events
.iter()
.filter_map(|event| match event {
StreamEvent::TextDelta(delta) => Some(delta.as_str()),
_ => None,
})
.collect();
assert_eq!(streamed, "Paris ", "{events:?}");
assert!(
events.iter().any(|event| matches!(
event,
StreamEvent::Done { text, .. } if !text.contains("STOPHERE")
)),
"the stream completes, and its Done never carries the stop sequence: {events:?}"
);
assert!(
events.iter().any(|event| matches!(
event,
StreamEvent::StopReason(reason) if reason == "stop_sequence"
)),
"the cut stream must report stop_sequence: {events:?}"
);
}
}
#[cfg(test)]
mod vllm_thinking_continuation_tests {
use super::*;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
fn vllm_schema(endpoint: &str, model_dir: &std::path::Path) -> ModelSchema {
vllm_schema_of_family(endpoint, model_dir, "qwen3.8")
}
fn vllm_schema_of_family(
endpoint: &str,
model_dir: &std::path::Path,
family: &str,
) -> ModelSchema {
ModelSchema {
id: "vllm-mlx/qwen3.8-27b:4bit".into(),
name: "Qwen3.8 27B".into(),
provider: "qwen".into(),
family: family.into(),
version: "1".into(),
capabilities: vec![],
context_length: 32768,
max_output_tokens: None,
param_count: String::new(),
quantization: None,
performance: Default::default(),
cost: Default::default(),
source: ModelSource::VllmMlx {
endpoint: endpoint.into(),
model_name: model_dir.display().to_string(),
},
tags: vec![],
supported_params: vec![],
public_benchmarks: vec![],
trust_tier: crate::schema::TrustTier::Curated,
deprecated: false,
available: true,
weights_ready: true,
}
}
fn model_dir() -> tempfile::TempDir {
let dir = tempfile::tempdir().unwrap();
std::fs::write(
dir.path().join("chat_template.jinja"),
crate::vllm_thinking::THINKING_TEMPLATE,
)
.unwrap();
dir
}
fn write_file_tool() -> Vec<serde_json::Value> {
vec![serde_json::json!({"type": "function", "function": {
"name": "write_file",
"parameters": {"type": "object", "properties": {
"path": {"type": "string"}, "mode": {"type": "integer"}
}}
}})]
}
async fn generate(
server: &MockServer,
dir: &std::path::Path,
max_tokens: usize,
tools: Option<&[serde_json::Value]>,
) -> (
String,
Vec<crate::tasks::generate::ToolCall>,
Vec<crate::tasks::generate::ThinkingBlock>,
Option<crate::TokenUsage>,
Option<String>,
) {
let (text, calls, thinking, _items, usage, stop) = RemoteBackend::new()
.generate_with_tools_multi_observed(
&vllm_schema(&server.uri(), dir),
"Write the answer to answer.txt",
None,
0.0,
max_tokens,
tools,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
&crate::protocol::OutputControls::default(),
&mut |_| {},
)
.await
.expect("the request reaches the stand-in server");
(text, calls, thinking, usage, stop)
}
fn request_bodies(requests: &[wiremock::Request], at: &str) -> Vec<serde_json::Value> {
requests
.iter()
.filter(|r| r.url.path() == at)
.map(|r| serde_json::from_slice(&r.body).unwrap())
.collect()
}
fn sse(chunks: &[serde_json::Value]) -> String {
let mut body: String = chunks.iter().map(|c| format!("data: {c}\n\n")).collect();
body.push_str("data: [DONE]\n\n");
body
}
#[tokio::test]
async fn gpt_oss_on_vllm_mlx_is_streamed_so_its_reasoning_is_not_the_answer() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(wiremock::matchers::body_partial_json(serde_json::json!({"stream": true})))
.respond_with(ResponseTemplate::new(200).set_body_raw(
sse(&[
serde_json::json!({"choices": [{"index": 0, "delta": {"role": "assistant"}}]}),
serde_json::json!({"choices": [{"index": 0, "delta": {"reasoning_content": "We need to count"}}]}),
serde_json::json!({"choices": [{"index": 0, "delta": {"reasoning_content": " spanning trees"}, "finish_reason": "length"}]}),
serde_json::json!({"choices": [], "usage": {"prompt_tokens": 89, "completion_tokens": 200, "total_tokens": 289}}),
]),
"text/event-stream",
))
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
let schema = vllm_schema_of_family(&server.uri(), dir.path(), "gpt-oss");
let (text, _calls, thinking, _items, usage, stop) = RemoteBackend::new()
.generate_with_tools_multi_observed(
&schema,
"Count the spanning trees of K_{3,4}.",
None,
0.0,
200,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
&crate::protocol::OutputControls::default(),
&mut |_| {},
)
.await
.expect("a streamed gpt-oss turn folds into a response");
assert_eq!(text, "", "the analysis channel is not the answer");
assert_eq!(
thinking.iter().map(|t| t.text.as_str()).collect::<String>(),
"We need to count spanning trees"
);
assert_eq!(stop.as_deref(), Some("length"));
assert_eq!(usage.map(|u| u.completion_tokens), Some(200));
}
#[tokio::test]
async fn a_folded_gpt_oss_stream_keeps_answer_reasoning_and_tool_calls() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(wiremock::matchers::body_partial_json(serde_json::json!({"stream": true})))
.respond_with(ResponseTemplate::new(200).set_body_raw(
sse(&[
serde_json::json!({"choices": [{"index": 0, "delta": {"reasoning_content": "Write it."}}]}),
serde_json::json!({"choices": [{"index": 0, "delta": {"content": "Writing."}}]}),
serde_json::json!({"choices": [{"index": 0, "delta": {"tool_calls": [{"index": 0, "id": "call_1", "function": {"name": "write_file", "arguments": "{\"path\":"}}]}}]}),
serde_json::json!({"choices": [{"index": 0, "delta": {"tool_calls": [{"index": 0, "function": {"arguments": "\"a.txt\"}"}}]}, "finish_reason": "tool_calls"}]}),
]),
"text/event-stream",
))
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
let schema = vllm_schema_of_family(&server.uri(), dir.path(), "gpt-oss");
let tools = write_file_tool();
let (text, calls, thinking, _items, _usage, stop) = RemoteBackend::new()
.generate_with_tools_multi_observed(
&schema,
"Write a.txt",
None,
0.0,
200,
Some(&tools),
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
&crate::protocol::OutputControls::default(),
&mut |_| {},
)
.await
.expect("a streamed gpt-oss turn folds into a response");
assert_eq!(text, "Writing.");
assert_eq!(thinking.first().map(|t| t.text.as_str()), Some("Write it."));
assert_eq!(stop.as_deref(), Some("tool_calls"));
let [call] = calls.as_slice() else {
panic!("one tool call: {calls:?}")
};
assert_eq!(call.name, "write_file");
assert_eq!(
call.arguments.get("path").and_then(|v| v.as_str()),
Some("a.txt")
);
}
#[tokio::test]
async fn an_explicit_effort_reaches_the_gpt_oss_template() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_raw(
sse(&[
serde_json::json!({"choices": [{"index": 0, "delta": {"content": "ok"}, "finish_reason": "stop"}]}),
]),
"text/event-stream",
))
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
let send = |family: &'static str,
effort: Option<crate::tasks::generate::ReasoningEffort>| {
let schema = vllm_schema_of_family(&server.uri(), dir.path(), family);
async move {
let controls = crate::protocol::OutputControls {
reasoning_effort: effort,
..Default::default()
};
let _ = RemoteBackend::new()
.generate_with_tools_multi_observed(
&schema,
"hi",
None,
0.0,
64,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
&controls,
&mut |_| {},
)
.await;
}
};
use crate::tasks::generate::ReasoningEffort::{Low, Xhigh};
send("gpt-oss", Some(Low)).await;
send("gpt-oss", Some(Xhigh)).await;
send("gpt-oss", None).await;
send("qwen3.8", Some(Low)).await;
let efforts: Vec<Option<String>> = server
.received_requests()
.await
.unwrap()
.iter()
.map(|r| {
let body: serde_json::Value = serde_json::from_slice(&r.body).unwrap();
body.pointer("/chat_template_kwargs/reasoning_effort")
.and_then(|v| v.as_str())
.map(str::to_string)
})
.collect();
assert_eq!(
efforts,
vec![
Some("low".to_string()),
Some("high".to_string()),
None,
None
]
);
}
#[test]
fn only_gpt_oss_is_folded_from_a_stream() {
for family in ["gpt-oss", "gpt_oss"] {
assert!(
crate::vllm_pool::folds_reasoning_from_stream(family),
"{family}"
);
}
for family in [
"qwen3.8",
"gemma4_unified",
"glm4.7",
"deepseek_v3",
"llama",
"gpt-4",
] {
assert!(
!crate::vllm_pool::folds_reasoning_from_stream(family),
"{family}"
);
}
}
#[tokio::test]
async fn a_thought_cut_off_at_the_budget_is_closed_and_answered() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "We need answer user: write 4 to a file"},
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 40, "completion_tokens": 514, "total_tokens": 554},
})))
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/v1/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{"text": "Done.\n\n<tool_call>\n<function=write_file>\n<parameter=path>\nanswer.txt\n</parameter>\n<parameter=mode>\n420\n</parameter>\n</function>\n</tool_call>",
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 580, "completion_tokens": 40, "total_tokens": 620},
})))
.mount(&server)
.await;
let dir = model_dir();
let tools = write_file_tool();
let (text, calls, thinking, usage, stop) =
generate(&server, dir.path(), 1024, Some(&tools)).await;
assert_eq!(text, "Done.");
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "write_file");
assert_eq!(calls[0].arguments["path"], serde_json::json!("answer.txt"));
assert_eq!(calls[0].arguments["mode"], serde_json::json!(420));
assert_eq!(thinking.len(), 1);
assert_eq!(thinking[0].text, "We need answer user: write 4 to a file");
let usage = usage.unwrap();
assert_eq!((usage.prompt_tokens, usage.completion_tokens), (580, 554));
assert_eq!(stop.as_deref(), Some("stop"));
let requests = server.received_requests().await.unwrap();
let chat = request_bodies(&requests, "/v1/chat/completions");
assert_eq!(
chat[0]["max_tokens"],
serde_json::json!(512),
"the first request gets the budget"
);
let completions = request_bodies(&requests, "/v1/completions");
assert_eq!(completions.len(), 1);
assert_eq!(completions[0]["max_tokens"], serde_json::json!(1024 - 514));
let prompt = completions[0]["prompt"].as_str().unwrap();
assert!(prompt.contains("<|im_start|>user\nWrite the answer to answer.txt<|im_end|>"));
assert!(prompt.contains(
"<think>\nWe need answer user: write 4 to a file\n\nConsidering the limited time"
));
assert!(prompt.ends_with("</think>\n\n"));
}
#[tokio::test]
async fn an_answer_inside_its_budget_is_returned_with_no_second_request() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "391",
"reasoning_content": "17 * 23 = 391"},
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 30, "completion_tokens": 54, "total_tokens": 84},
})))
.mount(&server)
.await;
let dir = model_dir();
let (text, calls, thinking, usage, _) = generate(&server, dir.path(), 1024, None).await;
assert_eq!(text, "391");
assert!(calls.is_empty());
assert_eq!(thinking[0].text, "17 * 23 = 391");
assert_eq!(usage.unwrap().completion_tokens, 54);
let requests = server.received_requests().await.unwrap();
assert!(request_bodies(&requests, "/v1/completions").is_empty());
}
async fn body_sent_for(
schema: ModelSchema,
server: &MockServer,
thinking: crate::tasks::generate::ThinkingMode,
) -> serde_json::Value {
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6},
})))
.mount(server)
.await;
let controls = crate::protocol::OutputControls {
thinking,
..Default::default()
};
RemoteBackend::new()
.generate_with_tools_multi_observed(
&schema,
"hi",
None,
0.0,
64,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
&controls,
&mut |_| {},
)
.await
.unwrap();
let requests = server.received_requests().await.unwrap();
request_bodies(&requests, "/v1/chat/completions").remove(0)
}
#[tokio::test]
async fn the_thinking_mode_reaches_vllm_mlx_as_the_template_switch() {
use crate::tasks::generate::ThinkingMode;
let dir = model_dir();
for (mode, expected) in [
(
ThinkingMode::Off,
Some(serde_json::json!({"enable_thinking": false})),
),
(
ThinkingMode::On,
Some(serde_json::json!({"enable_thinking": true})),
),
(ThinkingMode::Auto, None),
] {
let server = MockServer::start().await;
let body = body_sent_for(vllm_schema(&server.uri(), dir.path()), &server, mode).await;
assert_eq!(
body.get("chat_template_kwargs").cloned(),
expected,
"{mode:?}"
);
}
}
#[tokio::test]
async fn a_hosted_provider_never_gets_the_template_switch() {
let _credential_scope = crate::openrouter::test_credential_scope();
crate::openrouter::set_test_credential(Some("k"));
let server = MockServer::start().await;
let body = body_sent_for(
super::tests::openrouter_schema(&server.uri()),
&server,
crate::tasks::generate::ThinkingMode::Off,
)
.await;
assert!(body.get("chat_template_kwargs").is_none());
}
#[tokio::test]
async fn a_family_served_without_a_think_parser_keeps_its_whole_budget() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "answer"},
"finish_reason": "stop"}],
"usage": {"prompt_tokens": 30, "completion_tokens": 40, "total_tokens": 70},
})))
.mount(&server)
.await;
let dir = model_dir();
RemoteBackend::new()
.generate_with_tools_multi_observed(
&vllm_schema_of_family(&server.uri(), dir.path(), "qwq"),
"hi",
None,
0.0,
1024,
None,
None,
None,
None,
None,
0,
false,
crate::tasks::generate::CacheTtl::default(),
None,
None,
&crate::protocol::OutputControls::default(),
&mut |_| {},
)
.await
.unwrap();
let requests = server.received_requests().await.unwrap();
assert_eq!(
request_bodies(&requests, "/v1/chat/completions")[0]["max_tokens"],
serde_json::json!(1024)
);
}
#[tokio::test]
async fn a_model_whose_prompt_opens_no_thinking_keeps_its_whole_budget() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"choices": [{"message": {"role": "assistant", "content": "partial"},
"finish_reason": "length"}],
"usage": {"prompt_tokens": 30, "completion_tokens": 1024, "total_tokens": 1054},
})))
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
std::fs::write(
dir.path().join("chat_template.jinja"),
r#"{%- for m in messages %}{{ m.content }}{%- endfor %}{{- '<|im_start|>assistant\n' }}"#,
)
.unwrap();
let (text, ..) = generate(&server, dir.path(), 1024, None).await;
assert_eq!(text, "partial");
let requests = server.received_requests().await.unwrap();
assert_eq!(
request_bodies(&requests, "/v1/chat/completions")[0]["max_tokens"],
serde_json::json!(1024)
);
assert!(request_bodies(&requests, "/v1/completions").is_empty());
}
}