openrouter-batch 0.1.2

//! The production transport: a real HTTP client against a configurable API
//! root. The root is configuration, not a constant (context/05): the default
//! is the public endpoint; e2e points it at the local fake batch server.

use super::facade::{Message, StructuredParams, structured_body};
use super::transport::{
    BatchObject, BatchTransport, EmbeddingRequest, EmbeddingTransport, RawChatResponse,
    RawEmbeddingResponse, TransportError,
};
use async_trait::async_trait;
use serde_json::Value;
use std::time::Duration;

/// Path under the API root where the Batch API lives.
const BATCHES_PATH: &str = "/beta/batches";
const EMBEDDINGS_PATH: &str = "/v1/embeddings";
/// The synchronous chat-completions endpoint (same path the Batch API wraps
/// per item; called directly here with `model` inline).
const CHAT_COMPLETIONS_PATH: &str = "/v1/chat/completions";

pub struct HttpTransport {
    http: reqwest::Client,
    api_root: String,
    api_key: String,
}

impl HttpTransport {
    /// Build the transport. `api_root` is the OpenRouter API root (for
    /// example `https://openrouter.ai/api`); `api_key` is the bearer token, read
    /// once at process start and never logged (AGENTS.md secrets rules).
    pub fn new(api_root: &str, api_key: &str) -> Result<Self, TransportError> {
        Self::with_timeout(api_root, api_key, Duration::from_secs(30))
    }

    /// Build the transport with an explicit total request timeout. The
    /// synchronous chat-completions route can run far longer than the default
    /// 30s (measured: a deepseek-v4-flash sync call at 44.4s completed
    /// successfully; under `new` the same call dies mid-body as a reqwest
    /// decode error). Batch calls stay on the short default.
    pub fn with_timeout(
        api_root: &str,
        api_key: &str,
        timeout: Duration,
    ) -> Result<Self, TransportError> {
        let http = reqwest::Client::builder()
            .connect_timeout(Duration::from_secs(5))
            .timeout(timeout)
            .build()?;
        Ok(Self {
            http,
            api_root: api_root.trim_end_matches('/').to_string(),
            api_key: api_key.to_string(),
        })
    }

    /// POST/GET helper: any 2xx with a parseable batch body is success; any
    /// other status becomes a typed `TransportError::Api`.
    async fn send(&self, request: reqwest::RequestBuilder) -> Result<BatchObject, TransportError> {
        let response = request.send().await?;
        let status = response.status();
        let provider_request_id = response
            .headers()
            .get("x-request-id")
            .and_then(|value| value.to_str().ok())
            .map(str::to_string);
        let body = response
            .text()
            .await
            .map_err(|source| TransportError::Http {
                source,
                provider_request_id: provider_request_id.clone(),
            })?;
        if !status.is_success() {
            return Err(TransportError::Api {
                status: status.as_u16(),
                message: error_message(&body),
                provider_request_id,
            });
        }
        let mut batch: BatchObject =
            serde_json::from_str(&body).map_err(|source| TransportError::Decode {
                source,
                provider_request_id: provider_request_id.clone(),
            })?;
        batch.provider_request_id = provider_request_id;
        Ok(batch)
    }

    /// One synchronous structured chat-completions call (eval harness and
    /// other read-only tooling; the production normalization path stays on
    /// the Batch API). The body is built by the shared `structured_body`, so
    /// the sync wire shape cannot drift from a batch item; `model` is inline.
    /// The request timeout is the transport's: build with `with_timeout` when
    /// generation time exceeds the 30s `new` default (measured 2026-09-01: a
    /// glm-5.3-flash structured call ran 242s; an earlier hardcoded 120s
    /// per-request override silently capped every sync call and surfaced the
    /// deadline hit mid-body as a reqwest decode error).
    pub async fn chat_completion(
        &self,
        model: &str,
        params: &StructuredParams,
        messages: Vec<Message>,
    ) -> Result<RawChatResponse, TransportError> {
        let mut body = match structured_body(params, messages)? {
            Value::Object(map) => map,
            _ => unreachable!("structured_body always returns a JSON object"),
        };
        body.insert("model".to_string(), model.to_string().into());
        let response = self
            .http
            .post(format!("{}{CHAT_COMPLETIONS_PATH}", self.api_root))
            .bearer_auth(&self.api_key)
            .json(&body)
            .send()
            .await?;
        let status = response.status();
        let provider_request_id = response
            .headers()
            .get("x-request-id")
            .and_then(|value| value.to_str().ok())
            .map(str::to_string);
        let raw_text = response
            .text()
            .await
            .map_err(|source| TransportError::Http {
                source,
                provider_request_id: provider_request_id.clone(),
            })?;
        if !status.is_success() {
            return Err(TransportError::Api {
                status: status.as_u16(),
                message: error_message(&raw_text),
                provider_request_id,
            });
        }
        // Preserve an unparseable body verbatim for evidence instead of
        // failing the decode: the caller scores schema validity downstream.
        let raw: Value = serde_json::from_str(&raw_text)
            .unwrap_or_else(|_| serde_json::json!({"unparseable_body": raw_text}));
        Ok(RawChatResponse {
            raw,
            provider_request_id,
        })
    }
}

#[async_trait]
impl EmbeddingTransport for HttpTransport {
    async fn embed(
        &self,
        model: &str,
        inputs: &[String],
        idempotency_key: &str,
    ) -> Result<RawEmbeddingResponse, TransportError> {
        let response = self
            .http
            .post(format!("{}{EMBEDDINGS_PATH}", self.api_root))
            .bearer_auth(&self.api_key)
            .header("Idempotency-Key", idempotency_key)
            .json(&EmbeddingRequest::new(model, inputs.to_vec()))
            .send()
            .await?;
        let status = response.status();
        let provider_request_id = response
            .headers()
            .get("x-request-id")
            .and_then(|value| value.to_str().ok())
            .map(str::to_string);
        let body = response
            .text()
            .await
            .map_err(|source| TransportError::Http {
                source,
                provider_request_id: provider_request_id.clone(),
            })?;
        if !status.is_success() {
            return Err(TransportError::Api {
                status: status.as_u16(),
                message: error_message(&body),
                provider_request_id,
            });
        }
        let raw = serde_json::from_str(&body)
            .unwrap_or_else(|_| serde_json::json!({"unparseable_body": body}));
        Ok(RawEmbeddingResponse {
            raw,
            provider_request_id,
        })
    }
}

#[async_trait]
impl BatchTransport for HttpTransport {
    async fn submit(
        &self,
        batch: &super::transport::SubmitBatch,
        idempotency_key: &str,
    ) -> Result<BatchObject, TransportError> {
        // `batch` serializes endpoint, then model, then requests (declaration
        // order, pinned by the serialization test); reqwest's `.json` uses
        // serde_json, which preserves it.
        self.send(
            self.http
                .post(format!("{}{BATCHES_PATH}", self.api_root))
                .bearer_auth(&self.api_key)
                .header("Idempotency-Key", idempotency_key)
                .json(batch),
        )
        .await
    }

    async fn get(&self, batch_id: &str) -> Result<BatchObject, TransportError> {
        self.send(
            self.http
                .get(format!("{}{BATCHES_PATH}/{batch_id}", self.api_root))
                .bearer_auth(&self.api_key),
        )
        .await
    }
}

/// Extract `error.message` from a provider error body; fall back to a
/// truncated raw body so operators always see something.
fn error_message(body: &str) -> String {
    let parsed: Option<Value> = serde_json::from_str(body).ok();
    match parsed.as_ref().and_then(|v| {
        v.get("error")
            .and_then(|error| error.get("message"))
            .and_then(Value::as_str)
            .map(str::to_string)
    }) {
        Some(message) => message,
        None => {
            let mut truncated: String = body.chars().take(200).collect();
            if body.chars().count() > 200 {
                truncated.push('…');
            }
            truncated
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::AsyncWriteExt;

    #[test]
    fn extracts_provider_error_message() {
        let body = r#"{"error":{"message":"Model 'x/y' does not have a :batch endpoint."}}"#;
        assert_eq!(
            error_message(body),
            "Model 'x/y' does not have a :batch endpoint."
        );
    }

    #[test]
    fn falls_back_to_truncated_body_on_non_json_errors() {
        let message = error_message("<html>bad gateway</html>");
        assert!(message.contains("bad gateway"));
    }

    #[tokio::test]
    async fn truncated_response_body_is_retryable() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-Request-Id: req-truncated\r\nContent-Length: 100\r\n\r\n{\"data\":[",
                )
                .await
                .unwrap();
        });
        let transport = HttpTransport::new(&format!("http://{address}"), "test-key").unwrap();

        let error = transport
            .embed("test-model", &["input".to_string()], "stable-key")
            .await
            .unwrap_err();

        server.await.unwrap();
        assert!(
            matches!(&error, TransportError::Http { source, .. } if source.is_body() || source.is_decode()),
            "unexpected transport error: {error:?}"
        );
        assert!(error.retryable());
        assert_eq!(error.provider_request_id(), Some("req-truncated"));
    }

    #[tokio::test]
    async fn sync_chat_completion_decodes_content_and_request_id() {
        let body = r#"{"id":"chatcmpl-1","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"{\"title\":\"X\"}"},"finish_reason":"stop"}]}"#;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let response = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-Request-Id: req-sync\r\nContent-Length: {}\r\n\r\n{}",
                body.len(),
                body
            );
            socket.write_all(response.as_bytes()).await.unwrap();
        });
        let transport = HttpTransport::new(&format!("http://{address}"), "test-key").unwrap();
        let params = StructuredParams {
            schema: serde_json::json!({"type": "object"}),
            schema_name: "normalized_job".to_string(),
            max_tokens: 300,
            disable_reasoning: false,
            require_parameters: false,
            reasoning_max_tokens: None,
        };

        let response = transport
            .chat_completion("test/model", &params, vec![Message::user("hi")])
            .await
            .unwrap();

        server.await.unwrap();
        assert_eq!(response.content().unwrap(), "{\"title\":\"X\"}");
        assert_eq!(response.provider_request_id.as_deref(), Some("req-sync"));
    }

    #[tokio::test]
    async fn sync_chat_completion_preserves_api_error_status() {
        let body = r#"{"error":{"message":"no such route"}}"#;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let response = format!(
                "HTTP/1.1 404 Not Found\r\nContent-Type: application/json\r\nX-Request-Id: req-404\r\nContent-Length: {}\r\n\r\n{}",
                body.len(),
                body
            );
            socket.write_all(response.as_bytes()).await.unwrap();
        });
        let transport = HttpTransport::new(&format!("http://{address}"), "test-key").unwrap();
        let params = StructuredParams {
            schema: serde_json::json!({"type": "object"}),
            schema_name: "normalized_job".to_string(),
            max_tokens: 300,
            disable_reasoning: false,
            require_parameters: false,
            reasoning_max_tokens: None,
        };
        let error = transport
            .chat_completion("test/model", &params, vec![Message::user("hi")])
            .await
            .unwrap_err();

        server.await.unwrap();
        assert!(
            matches!(&error, TransportError::Api { status: 404, .. }),
            "unexpected transport error: {error:?}"
        );
        assert_eq!(error.provider_request_id(), Some("req-404"));
    }

    #[tokio::test]
    async fn batch_decode_failure_preserves_provider_request_id() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-Request-Id: req-decode\r\nContent-Length: 8\r\n\r\nnot-json",
                )
                .await
                .unwrap();
        });
        let transport = HttpTransport::new(&format!("http://{address}"), "test-key").unwrap();

        let error = transport
            .get("batch-test")
            .await
            .expect_err("malformed batch response must fail");

        server.await.unwrap();
        assert!(matches!(&error, TransportError::Decode { .. }));
        assert_eq!(error.provider_request_id(), Some("req-decode"));
    }

    #[tokio::test]
    async fn sync_chat_completion_honors_transport_timeout() {
        // The constructor timeout must govern the sync route: the server
        // answers after 2s, the transport deadline is 1s, so the call must
        // fail with a retryable Http error instead of the deadline silently
        // extending to a hardcoded per-request override.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let address = listener.local_addr().unwrap();
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            tokio::time::sleep(Duration::from_secs(2)).await;
            socket
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n{}")
                .await
                .unwrap();
        });
        let transport = HttpTransport::with_timeout(
            &format!("http://{address}"),
            "test-key",
            Duration::from_secs(1),
        )
        .unwrap();
        let params = StructuredParams {
            schema: serde_json::json!({"type": "object"}),
            schema_name: "normalized_job".to_string(),
            max_tokens: 300,
            disable_reasoning: false,
            require_parameters: false,
            reasoning_max_tokens: None,
        };

        let error = transport
            .chat_completion("test/model", &params, vec![Message::user("hi")])
            .await
            .unwrap_err();

        server.await.unwrap();
        assert!(
            matches!(&error, TransportError::Http { source, .. } if source.is_timeout()),
            "unexpected transport error: {error:?}"
        );
        assert!(error.retryable());
    }
}