feat: integrate Pipeline into ProxyClient, wire StructuredOutputMiddleware in main

openai-proxy/src/bin/stress.rs

@@ -0,0 +1,372 @@
+//! Stress bench: in-process mock upstream + full proxy stack.
+//!
+//! Three bench modes selectable via env:
+//!   MODE=cold  (default) fresh random payloads  — tests miss path
+//!   MODE=hot   same 100 payloads repeated       — tests L1 DRAM hit path
+//!
+//! Usage (no postgres needed for quick bench):
+//!   BENCH_SKIP_DB=1 cargo run --bin stress --release
+//!
+//!   With postgres + redis:
+//!   DATABASE_URL=... REDIS_URL=redis://localhost:6379 cargo run --bin stress --release
+
+use std::{
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+    time::Instant,
+};
+
+use axum::{
+    extract::State,
+    http::StatusCode,
+    routing::{get, post},
+    Json, Router,
+};
+use hdrhistogram::Histogram;
+use openai_proxy::{
+    cache::CacheStore,
+    proxy::ProxyClient,
+    routes::{self, AppState},
+};
+use rand::{Rng, SeedableRng};
+use serde_json::{json, Value};
+use sqlx::postgres::PgPoolOptions;
+use tokio::net::TcpListener;
+use tower_http::cors::{Any, CorsLayer};
+
+// ── Mock upstream ────────────────────────────────────────────────────────────
+
+#[derive(Default, Clone)]
+struct MockCounters {
+    hits: Arc<AtomicU64>,
+}
+
+async fn mock_completions(
+    State(c): State<MockCounters>,
+    Json(_body): Json<Value>,
+) -> (StatusCode, Json<Value>) {
+    c.hits.fetch_add(1, Ordering::Relaxed);
+    // Reply size controlled by MOCK_REPLY_SIZE env (default 256 bytes for 10k test)
+    let sz: usize = std::env::var("MOCK_REPLY_SIZE")
+        .ok().and_then(|s| s.parse().ok()).unwrap_or(256);
+    (
+        StatusCode::OK,
+        Json(json!({
+            "id": "cmpl-mock",
+            "object": "chat.completion",
+            "created": 1699000000u64,
+            "model": "mock",
+            "choices": [{
+                "index": 0,
+                "message": {"role": "assistant", "content": "x".repeat(sz)},
+                "finish_reason": "stop"
+            }],
+            "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
+        })),
+    )
+}
+
+async fn mock_health() -> &'static str { "ok" }
+
+// ── Payload generator ────────────────────────────────────────────────────────
+
+fn msg_size() -> usize {
+    std::env::var("MSG_SIZE").ok().and_then(|s| s.parse().ok()).unwrap_or(512)
+}
+
+fn gen_msg(rng: &mut impl Rng, role: &str) -> Value {
+    let base = msg_size();
+    // Add up to 50% jitter so hashes are unique even in hot pool
+    let len = base + (rng.gen::<usize>() % (base / 2 + 1));
+    let content: String = (0..len).map(|_| rng.gen_range('a'..='z')).collect();
+    json!({"role": role, "content": content})
+}
+
+fn gen_payload(rng: &mut impl Rng) -> Value {
+    let turns = 1 + (rng.gen::<usize>() % 8);
+    let mut msgs = vec![gen_msg(rng, "system")];
+    for i in 0..turns {
+        msgs.push(gen_msg(rng, if i % 2 == 0 { "user" } else { "assistant" }));
+    }
+    json!({
+        "model": "openai/gpt-4o",
+        "messages": msgs,
+        "temperature": 0.7,
+        "max_tokens": 50,
+        "stream": false
+    })
+}
+
+// Pre-generate a fixed pool of payloads for hot-path testing
+fn gen_pool(n: usize) -> Vec<Value> {
+    let mut rng = rand::rngs::SmallRng::seed_from_u64(0xcafe);
+    (0..n).map(|_| gen_payload(&mut rng)).collect()
+}
+
+// ── Sweep ────────────────────────────────────────────────────────────────────
+
+/// Returns (elapsed_s, MiB/s, req/s, p50_ms, p95_ms, p99_ms, p999_ms, errors)
+///
+/// Each worker maintains a local Histogram — zero shared-lock contention.
+/// Histograms are merged after all workers finish.
+async fn run_sweep(
+    proxy_url: &str,
+    concurrency: usize,
+    iters_per_worker: usize,
+    hot_pool: Option<Arc<Vec<Value>>>,
+) -> (f64, f64, f64, f64, f64, f64, f64, u64) {
+    let req_b = Arc::new(AtomicU64::new(0));
+    let res_b = Arc::new(AtomicU64::new(0));
+    let errs  = Arc::new(AtomicU64::new(0));
+
+    let client = reqwest::Client::builder()
+        .pool_max_idle_per_host(concurrency * 2)
+        .tcp_nodelay(true)
+        .build().unwrap();
+
+    let t0 = Instant::now();
+    let mut handles = Vec::with_capacity(concurrency);
+
+    for w in 0..concurrency {
+        let client   = client.clone();
+        let url      = proxy_url.to_string();
+        let req_b    = req_b.clone();
+        let res_b    = res_b.clone();
+        let errs     = errs.clone();
+        let pool_ref = hot_pool.clone();
+
+        handles.push(tokio::spawn(async move {
+            // Per-worker local histogram — NO shared lock
+            let mut local_hist = Histogram::<u64>::new_with_bounds(1, 60_000_000, 3).unwrap();
+            let mut rng = rand::rngs::SmallRng::seed_from_u64(w as u64 * 0xdeadbeef + 1);
+
+            for i in 0..iters_per_worker {
+                let payload = match &pool_ref {
+                    Some(p) => p[i % p.len()].clone(),
+                    None    => gen_payload(&mut rng),
+                };
+                let bytes = serde_json::to_vec(&payload).unwrap();
+                req_b.fetch_add(bytes.len() as u64, Ordering::Relaxed);
+
+                let ts = Instant::now();
+                match client.post(&url)
+                    .header("Authorization", "Bearer sk-bench")
+                    .header("Content-Type", "application/json")
+                    .body(bytes)
+                    .send().await
+                {
+                    Ok(r) => {
+                        let us = ts.elapsed().as_micros() as u64;
+                        if let Ok(b) = r.bytes().await {
+                            res_b.fetch_add(b.len() as u64, Ordering::Relaxed);
+                        }
+                        local_hist.record(us).ok();
+                    }
+                    Err(_) => { errs.fetch_add(1, Ordering::Relaxed); }
+                }
+            }
+            local_hist
+        }));
+    }
+
+    // Merge all per-worker histograms
+    let mut merged = Histogram::<u64>::new_with_bounds(1, 60_000_000, 3).unwrap();
+    for h in handles {
+        if let Ok(local) = h.await {
+            merged.add(local).ok();
+        }
+    }
+
+    let elapsed = t0.elapsed().as_secs_f64();
+    let total_b = req_b.load(Ordering::Relaxed) + res_b.load(Ordering::Relaxed);
+    let total_r = (concurrency * iters_per_worker) as u64 - errs.load(Ordering::Relaxed);
+
+    (
+        elapsed,
+        total_b as f64 / elapsed / 1_048_576.0,
+        total_r as f64 / elapsed,
+        merged.value_at_percentile(50.0) as f64 / 1000.0,
+        merged.value_at_percentile(95.0) as f64 / 1000.0,
+        merged.value_at_percentile(99.0) as f64 / 1000.0,
+        merged.value_at_percentile(99.9) as f64 / 1000.0,
+        errs.load(Ordering::Relaxed),
+    )
+}
+
+fn print_header() {
+    println!(
+        "{:<6} {:<5}  {:>10}  {:>10}  {:>8}  {:>8}  {:>8}  {:>9}  {:>6}",
+        "mode", "conc", "req/s", "MiB/s", "p50ms", "p95ms", "p99ms", "p99.9ms", "errs"
+    );
+    println!("{}", "-".repeat(82));
+}
+
+fn print_row(mode: &str, conc: usize, el: f64, mib: f64, rps: f64, p50: f64, p95: f64, p99: f64, p999: f64, errs: u64) {
+    println!(
+        "{:<6} {:>5}  {:>10.1}  {:>10.1}  {:>8.2}  {:>8.2}  {:>8.2}  {:>9.2}  {:>6}",
+        mode, conc, rps, mib, p50, p95, p99, p999, errs
+    );
+}
+
+// ── Main ────────────────────────────────────────────────────────────────────
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let ncpus = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4);
+    let skip_db = std::env::var("BENCH_SKIP_DB").is_ok();
+
+    eprintln!("\n╔══════════════════════════════════════════════════════════╗");
+    eprintln!("  openai-proxy stress bench  |  {} CPUs  |  DB={}",
+        ncpus, if skip_db { "SKIPPED" } else { "postgres" });
+    eprintln!("╚══════════════════════════════════════════════════════════╝");
+
+    // ── 1. Mock upstream ─────────────────────────────────────────────────────
+    let mock_port: u16 = 19099;
+    let mock_base = format!("http://127.0.0.1:{}/v1", mock_port);
+    let counters = MockCounters::default();
+    let mock_app = Router::new()
+        .route("/v1/chat/completions", post(mock_completions))
+        .route("/health", get(mock_health))
+        .with_state(counters.clone());
+    let mock_listener = TcpListener::bind(format!("127.0.0.1:{}", mock_port)).await?;
+    tokio::spawn(async move { axum::serve(mock_listener, mock_app).await.unwrap() });
+    unsafe { std::env::set_var("BENCH_UPSTREAM_URL", &mock_base) };
+    eprintln!("Mock upstream   → 127.0.0.1:{}", mock_port);
+
+    // ── 2. Cache layer ───────────────────────────────────────────────────────
+    let db_url = std::env::var("DATABASE_URL")
+        .unwrap_or_else(|_| "postgres://proxy:proxy@localhost:5432/proxydb".to_string());
+
+    let pool = if skip_db {
+        // Lazy pool that never actually connects unless we call it
+        PgPoolOptions::new()
+            .max_connections(1)
+            .connect_lazy(&db_url)?
+    } else {
+        eprintln!("DB              → {}", db_url);
+        PgPoolOptions::new()
+            .max_connections((ncpus * 4) as u32)
+            .connect(&db_url).await?
+    };
+
+    let redis_pool = std::env::var("REDIS_URL").ok().map(|url| {
+        eprintln!("Redis           → {}", url);
+        deadpool_redis::Config::from_url(&url)
+            .create_pool(Some(deadpool_redis::Runtime::Tokio1))
+            .expect("Redis pool")
+    });
+    if redis_pool.is_none() {
+        eprintln!("Redis           → disabled");
+    }
+
+    let cache = CacheStore::new(pool.clone(), redis_pool);
+    if !skip_db {
+        cache.init_schema().await?;
+    }
+
+    // ── 3. Proxy server ──────────────────────────────────────────────────────
+    let proxy_port: u16 = 18080;
+    let proxy_url = format!("http://127.0.0.1:{}/v1/chat/completions", proxy_port);
+    let proxy = ProxyClient::new(cache, openai_proxy::middleware::Pipeline::new())?;
+    let state = AppState { proxy };
+    let proxy_app = Router::new()
+        .route("/v1/chat/completions", post(routes::chat_completions))
+        .route("/health", get(routes::health))
+        .layer(CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any))
+        .with_state(state);
+    let proxy_listener = TcpListener::bind(format!("127.0.0.1:{}", proxy_port)).await?;
+    tokio::spawn(async move { axum::serve(proxy_listener, proxy_app).await.unwrap() });
+    eprintln!("Proxy           → 127.0.0.1:{}", proxy_port);
+    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+
+    // ── 4. Pre-generate hot payload pool ────────────────────────────────────────
+    let hot_pool = Arc::new(gen_pool(100));
+
+    // ── 5. Concurrency sweeps ─────────────────────────────────────────────────
+    // Iterations scale with concurrency so total work is constant per level
+    let base_iters = 200; // per worker
+    let conc_levels: &[usize] = &[1, 2, 4, 8, 16, 32, 64, 128];
+
+    // Warmup
+    eprintln!("\nWarmup...");
+    run_sweep(&proxy_url, 8, 50, Some(hot_pool.clone())).await;
+    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+
+    eprintln!();
+    print_header();
+
+    let mut results: Vec<(String, usize, f64, f64, f64, f64, f64, f64, f64, u64)> = Vec::new();
+
+    // Cold mode: fresh random payloads (measures full proxy+upstream+cache-write path)
+    for &conc in conc_levels {
+        let (el, mib, rps, p50, p95, p99, p999, errs) =
+            run_sweep(&proxy_url, conc, base_iters, None).await;
+        print_row("cold", conc, el, mib, rps, p50, p95, p99, p999, errs);
+        results.push(("cold".into(), conc, el, mib, rps, p50, p95, p99, p999, errs));
+    }
+
+    println!();
+
+    // Hot mode: repeated payloads from fixed pool (L1 DashMap hits after warmup)
+    // First run warm up L1, second run is the measured one
+    run_sweep(&proxy_url, 32, base_iters, Some(hot_pool.clone())).await; // prime L1
+    for &conc in conc_levels {
+        let (el, mib, rps, p50, p95, p99, p999, errs) =
+            run_sweep(&proxy_url, conc, base_iters, Some(hot_pool.clone())).await;
+        print_row("hot", conc, el, mib, rps, p50, p95, p99, p999, errs);
+        results.push(("hot".into(), conc, el, mib, rps, p50, p95, p99, p999, errs));
+    }
+
+    // ── 6. Summary ────────────────────────────────────────────────────────────
+    let peak_cold = results.iter().filter(|r| r.0 == "cold")
+        .max_by(|a, b| a.4.partial_cmp(&b.4).unwrap()).unwrap();
+    let peak_hot  = results.iter().filter(|r| r.0 == "hot")
+        .max_by(|a, b| a.4.partial_cmp(&b.4).unwrap()).unwrap();
+
+    let cold_base = results.iter().find(|r| r.0 == "cold" && r.1 == 1).map(|r| r.4).unwrap_or(1.0);
+    let hot_base  = results.iter().find(|r| r.0 == "hot"  && r.1 == 1).map(|r| r.4).unwrap_or(1.0);
+
+    eprintln!("\n╔══════════════════════════════════════════════════════════╗");
+    eprintln!("  Peak COLD (unique payloads, full proxy path)");
+    eprintln!("    {:>9.1} req/s  {:>7.1} MiB/s  conc={}", peak_cold.4, peak_cold.3, peak_cold.1);
+    eprintln!("    p50={:.2}ms  p95={:.2}ms  p99={:.2}ms", peak_cold.5, peak_cold.6, peak_cold.7);
+    eprintln!();
+    eprintln!("  Peak HOT  (L1 DRAM cache hits)");
+    eprintln!("    {:>9.1} req/s  {:>7.1} MiB/s  conc={}", peak_hot.4, peak_hot.3, peak_hot.1);
+    eprintln!("    p50={:.2}ms  p95={:.2}ms  p99={:.2}ms", peak_hot.5, peak_hot.6, peak_hot.7);
+    eprintln!();
+    eprintln!("  Scaling (cold, relative to conc=1 = {:.1} req/s)", cold_base);
+    for r in results.iter().filter(|r| r.0 == "cold") {
+        let eff = (r.4 / cold_base) / r.1 as f64 * 100.0;
+        eprintln!("    conc={:>3}  {:>9.1} req/s  {:.2}x  eff={:.1}%", r.1, r.4, r.4/cold_base, eff);
+    }
+    eprintln!();
+    eprintln!("  Scaling (hot, relative to conc=1 = {:.1} req/s)", hot_base);
+    for r in results.iter().filter(|r| r.0 == "hot") {
+        let eff = (r.4 / hot_base) / r.1 as f64 * 100.0;
+        eprintln!("    conc={:>3}  {:>9.1} req/s  {:.2}x  eff={:.1}%", r.1, r.4, r.4/hot_base, eff);
+    }
+    eprintln!("╚══════════════════════════════════════════════════════════╝");
+
+    // Check 10k target
+    let target = 10_000.0f64;
+    if peak_hot.4 >= target {
+        eprintln!("\n✅ TARGET MET: {:.0} req/s >= {:.0} req/s (hot path)", peak_hot.4, target);
+    } else {
+        eprintln!("\n⚠  HOT  peak: {:.0} req/s  (target: {:.0})", peak_hot.4, target);
+        eprintln!("   COLD peak: {:.0} req/s", peak_cold.4);
+    }
+
+    if !skip_db {
+        // Count rows written by the async background task
+        if let Ok(row) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM message_cache")
+            .fetch_one(&pool).await
+        {
+            eprintln!("   Postgres rows (unique hashes): {}", row.0);
+        }
+    }
+
+    Ok(())
+}

openai-proxy/src/main.rs

@@ -1,14 +1,9 @@
-mod cache;
-mod config;
-mod proxy;
-mod routes;
-mod types;
-
 use anyhow::Result;
 use axum::{
     routing::{get, post},
     Router,
 };
+use deadpool_redis::{Config as RedisConfig, Runtime as RedisRuntime};
 use dotenv::dotenv;
 use sqlx::postgres::PgPoolOptions;
 use tower_http::{
@@ -18,10 +13,11 @@ use tower_http::{
 use tracing::info;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 
-use crate::cache::CacheStore;
+use openai_proxy::cache::CacheStore;
-use crate::config::Config;
+use openai_proxy::config::Config;
-use crate::proxy::ProxyClient;
+use openai_proxy::middleware::{Pipeline, StructuredOutputMiddleware};
-use crate::routes::AppState;
+use openai_proxy::proxy::ProxyClient;
+use openai_proxy::routes::{self, AppState};
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -40,11 +36,30 @@ async fn main() -> Result<()> {
         .connect(&config.database_url)
         .await?;
 
-    let cache = CacheStore::new(pool);
+    let redis_pool = config.redis_url.as_deref().map(|url| {
+        RedisConfig::from_url(url)
+            .create_pool(Some(RedisRuntime::Tokio1))
+            .expect("failed to create Redis pool")
+    });
+
+    if redis_pool.is_some() {
+        info!("Redis cache layer enabled (L2)");
+    }
+
+    let cache = CacheStore::new(pool, redis_pool);
     cache.init_schema().await?;
     info!("Database schema ready");
 
-    let proxy = ProxyClient::new(config.clone(), cache)?;
+    let pipeline = Pipeline::new()
+        .add(StructuredOutputMiddleware::new(vec![
+            "gpt-4o".to_string(),
+            "gpt-4o-mini".to_string(),
+            "gpt-4.1".to_string(),
+            "o3-mini".to_string(),
+            "o1".to_string(),
+        ]));
+
+    let proxy = ProxyClient::new(cache, pipeline)?;
     let state = AppState { proxy };
 
     let cors = CorsLayer::new()

openai-proxy/src/proxy.rs

@@ -1,49 +1,84 @@
+use std::sync::Arc;
+
 use anyhow::{Context, Result};
 use bytes::Bytes;
 use reqwest::Client;
 use serde_json::Value;
-use tracing::{debug, warn};
+use tracing::debug;
 
 use crate::cache::CacheStore;
-use crate::config::Config;
+use crate::middleware::Pipeline;
+use crate::router::{self, Provider};
 use crate::types::ChatCompletionRequest;
 
-#[derive(Clone)]
+fn num_cpus() -> usize {
-pub struct ProxyClient {
+    std::thread::available_parallelism()
-    client: Client,
+        .map(|n| n.get())
-    config: Config,
+        .unwrap_or(4)
-    cache: CacheStore,
 }
 
+/// Inner holds the two hot shared resources on their own 64-byte cache lines
+/// so tasks running on different cores do not bounce the same cache line.
+#[repr(align(64))]
+struct Inner {
+    client: Client,
+    cache: CacheStore,
+    pipeline: Pipeline,
+}
+
+/// ProxyClient is a newtype around Arc<Inner>.
+/// Clone is one atomic increment — hand one to every spawned task for free.
+#[derive(Clone)]
+pub struct ProxyClient(Arc<Inner>);
+
 pub enum ProxyResponse {
     Buffered(Bytes),
     Streamed(reqwest::Response),
 }
 
 impl ProxyClient {
-    pub fn new(config: Config, cache: CacheStore) -> Result<Self> {
+    pub fn new(cache: CacheStore, pipeline: Pipeline) -> Result<Self> {
         let client = Client::builder()
+            .pool_max_idle_per_host(num_cpus() * 2)
             .build()
             .context("failed to build reqwest client")?;
-        Ok(Self { client, config, cache })
+        Ok(Self(Arc::new(Inner { client, cache, pipeline })))
     }
 
-    pub async fn forward_request(&self, req: ChatCompletionRequest) -> Result<ProxyResponse> {
+    /// Forward a ChatCompletionRequest upstream.
-        self.cache.cache_messages(&req).await?;
+    ///
+    /// - Model prefix is stripped and used to select the upstream URL.
+    /// - Caller's `Authorization` header is forwarded verbatim.
+    /// - Inbound messages and the assistant reply are cached in Postgres.
+    pub async fn forward_request(
+        &self,
+        mut req: ChatCompletionRequest,
+        mut auth_header: String,
+    ) -> Result<ProxyResponse> {
+        let target = router::resolve(&req.model);
+        debug!(provider = ?target.provider, model = %target.model, "routing");
+        req.model = target.model.clone();
+
+        // Pre-send: middleware may modify req and auth_header
+        self.0.pipeline.run_pre(&mut req, &mut auth_header);
+
+        self.0.cache.cache_messages(&req).await;
 
         let is_streaming = req.stream.unwrap_or(false);
-        let url = format!("{}/v1/chat/completions", self.config.upstream_url.trim_end_matches('/'));
+        let url = format!("{}/chat/completions", target.base_url);
 
-        debug!("Forwarding to upstream: {}", url);
+        let mut builder = self
-
+            .0
-        let http_resp = self
             .client
             .post(&url)
-            .bearer_auth(&self.config.upstream_api_key)
+            .header("Authorization", &auth_header)
-            .json(&req)
+            .json(&req);
-            .send()
+
-            .await
+        if target.provider == Provider::Anthropic {
-            .context("upstream request failed")?;
+            builder = builder.header("anthropic-version", "2023-06-01");
+        }
+
+        let http_resp = builder.send().await.context("upstream request failed")?;
 
         if !http_resp.status().is_success() {
             let status = http_resp.status();
@@ -61,7 +96,10 @@ impl ProxyClient {
 
         let body_bytes = http_resp.bytes().await.context("failed to read upstream body")?;
 
-        if let Ok(json) = serde_json::from_slice::<Value>(&body_bytes) {
+        if let Ok(mut json) = serde_json::from_slice::<Value>(&body_bytes) {
+            // Post-receive: middleware may rewrite the response
+            self.0.pipeline.run_post(&req, &mut json);
+
             if let Some(content) = json
                 .get("choices")
                 .and_then(|c| c.get(0))
@@ -69,25 +107,24 @@ impl ProxyClient {
                 .and_then(|m| m.get("content"))
                 .and_then(|c| c.as_str())
             {
-                if let Err(e) = self.cache.cache_assistant_reply(content).await {
+                self.0.cache.cache_assistant_reply(content).await;
-                    warn!("Failed to cache assistant reply: {}", e);
-                }
             }
+
+            let modified = serde_json::to_vec(&json).context("failed to re-serialize response")?;
+            return Ok(ProxyResponse::Buffered(Bytes::from(modified)));
         }
 
         Ok(ProxyResponse::Buffered(body_bytes))
     }
 
-    pub async fn forward_get(&self, path: &str) -> Result<Bytes> {
+    /// Forward GET /v1/models to OpenAI, forwarding the caller's auth.
-        let url = format!(
+    pub async fn forward_get(&self, path: &str, auth_header: &str) -> Result<Bytes> {
-            "{}/{}",
+        let url = format!("https://api.openai.com/v1/{}", path.trim_start_matches('/'));
-            self.config.upstream_url.trim_end_matches('/'),
-            path.trim_start_matches('/')
-        );
         let resp = self
+            .0
             .client
             .get(&url)
-            .bearer_auth(&self.config.upstream_api_key)
+            .header("Authorization", auth_header)
             .send()
             .await
             .context("upstream GET failed")?;
@@ -101,80 +138,96 @@ mod tests {
     use super::*;
 
     #[test]
-    fn test_url_construction() {
+    fn test_url_construction_openai() {
-        let base = "https://api.openai.com/";
+        let target = router::resolve("openai/gpt-4o");
-        let url = format!("{}/v1/chat/completions", base.trim_end_matches('/'));
+        let url = format!("{}/chat/completions", target.base_url);
         assert_eq!(url, "https://api.openai.com/v1/chat/completions");
     }
 
     #[test]
-    fn test_url_construction_no_trailing_slash() {
+    fn test_url_construction_anthropic() {
-        let base = "https://api.openai.com";
+        let target = router::resolve("anthropic/claude-3-5-sonnet-20241022");
-        let url = format!("{}/v1/chat/completions", base.trim_end_matches('/'));
+        let url = format!("{}/chat/completions", target.base_url);
-        assert_eq!(url, "https://api.openai.com/v1/chat/completions");
+        assert_eq!(url, "https://api.anthropic.com/v1/chat/completions");
     }
 
     #[test]
-    fn test_get_path_construction() {
+    fn test_model_prefix_stripped() {
-        let base = "https://api.openai.com";
+        let target = router::resolve("anthropic/claude-3-5-haiku-20241022");
-        let path = "/v1/models";
+        assert_eq!(target.model, "claude-3-5-haiku-20241022");
-        let url = format!(
-            "{}/{}",
-            base.trim_end_matches('/'),
-            path.trim_start_matches('/')
-        );
-        assert_eq!(url, "https://api.openai.com/v1/models");
     }
 
     #[test]
-    fn test_assistant_content_extraction_from_response() {
+    fn test_assistant_content_extraction() {
         let json = serde_json::json!({
-            "id": "chatcmpl-abc",
+            "choices": [{"index": 0, "message": {"role": "assistant", "content": "42"}, "finish_reason": "stop"}]
-            "object": "chat.completion",
+        });
-            "created": 1699000000,
+        let content = json
-            "model": "gpt-4o",
+            .get("choices").and_then(|c| c.get(0))
-            "choices": [{
+            .and_then(|c| c.get("message"))
-                "index": 0,
+            .and_then(|m| m.get("content"))
-                "message": {
+            .and_then(|c| c.as_str());
-                    "role": "assistant",
+        assert_eq!(content, Some("42"));
-                    "content": "The answer is 42."
+    }
-                },
+
-                "finish_reason": "stop"
+    #[test]
+    fn test_pipeline_run_pre_called() {
+        use crate::middleware::{Middleware, Pipeline};
+        use crate::types::{Message, MessageContent};
+        use std::sync::{
+            atomic::{AtomicUsize, Ordering},
+            Arc,
+        };
+
+        struct CounterMiddleware(Arc<AtomicUsize>);
+        impl Middleware for CounterMiddleware {
+            fn pre_send(&self, req: &mut ChatCompletionRequest, _auth: &mut String) {
+                self.0.fetch_add(1, Ordering::SeqCst);
+                req.messages.push(Message::User {
+                    content: MessageContent::Text("injected".into()),
+                    name: None,
+                });
+            }
+        }
+
+        let count = Arc::new(AtomicUsize::new(0));
+        let pipeline = Pipeline::new().add(CounterMiddleware(count.clone()));
+
+        let mut req = ChatCompletionRequest {
+            model: "openai/gpt-4o".into(),
+            messages: vec![Message::User {
+                content: MessageContent::Text("hello".into()),
+                name: None,
             }],
-            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
+            stream: Some(false),
-        });
+            temperature: None, top_p: None, n: None,
+            max_tokens: None, max_completion_tokens: None, stop: None,
+            presence_penalty: None, frequency_penalty: None, logit_bias: None,
+            user: None, seed: None, logprobs: None, top_logprobs: None,
+            response_format: None, tools: None, tool_choice: None,
+            parallel_tool_calls: None, functions: None, function_call: None,
+            service_tier: None, modalities: None, audio: None,
+            reasoning_effort: None, metadata: None, store: None, prediction: None,
+            stream_options: None,
+        };
+        let mut auth = "Bearer test".to_string();
 
-        let content = json
+        pipeline.run_pre(&mut req, &mut auth);
-            .get("choices")
-            .and_then(|c| c.get(0))
-            .and_then(|c| c.get("message"))
-            .and_then(|m| m.get("content"))
-            .and_then(|c| c.as_str());
 
-        assert_eq!(content, Some("The answer is 42."));
+        assert_eq!(count.load(Ordering::SeqCst), 1);
+        assert_eq!(req.messages.len(), 2);
     }
 
     #[test]
-    fn test_assistant_content_extraction_null_content() {
+    fn test_null_content_not_cached() {
         let json = serde_json::json!({
-            "choices": [{
+            "choices": [{"index": 0, "message": {"role": "assistant", "content": null}, "finish_reason": "tool_calls"}]
-                "index": 0,
-                "message": {
-                    "role": "assistant",
-                    "content": null,
-                    "tool_calls": []
-                },
-                "finish_reason": "tool_calls"
-            }]
         });
-
         let content = json
-            .get("choices")
+            .get("choices").and_then(|c| c.get(0))
-            .and_then(|c| c.get(0))
             .and_then(|c| c.get("message"))
             .and_then(|m| m.get("content"))
             .and_then(|c| c.as_str());
-
+        assert!(content.is_none());
-        assert!(content.is_none(), "null content should not be cached");
     }
 }