use std::sync::{Mutex, PoisonError};
use anyhow::{Context, Result};
use serde_json::{Value, json};
use tokio_util::sync::CancellationToken;
use super::{VideoAnswer, VideoConfig, VideoRequest, VideoUnderstanding};
use crate::shared::api::openai::{ModelEnvelope, attributed};
const SUBJECT: &str = "OpenRouter video";
pub struct GatewayVideo {
http: reqwest::Client,
cfg: VideoConfig,
effort: Mutex<Option<Option<&'static str>>>,
}
impl GatewayVideo {
pub fn new(cfg: VideoConfig) -> Self {
Self {
http: reqwest::Client::new(),
cfg: VideoConfig {
base_url: cfg.base_url.trim_end_matches('/').to_string(),
model: cfg.model.trim().to_string(),
..cfg
},
effort: Mutex::new(None),
}
}
fn signed(&self, rb: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
let rb = rb.bearer_auth(&self.cfg.api_key);
if self.cfg.attribution {
attributed(rb)
} else {
rb
}
}
async fn lowest_effort(&self, cancel: &CancellationToken) -> Option<&'static str> {
if let Some(known) = *self.effort.lock().unwrap_or_else(PoisonError::into_inner) {
return known;
}
let url = format!("{}/model/{}", self.cfg.base_url, self.cfg.model);
let answer = tokio::select! {
biased;
_ = cancel.cancelled() => return None,
r = self.signed(self.http.get(&url)).send() => r,
};
let listed = match answer {
Ok(resp) if resp.status().is_success() => resp
.json::<ModelEnvelope>()
.await
.ok()
.and_then(|entry| entry.data.lowest_effort()),
Ok(resp)
if resp.status().is_server_error()
|| resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS =>
{
tracing::debug!(%url, status = %resp.status(), "the gateway's entry is unavailable for now");
return None;
}
Ok(_) => None,
Err(err) => {
tracing::debug!(%url, error = %err, "no answer about the video model");
return None;
}
};
*self.effort.lock().unwrap_or_else(PoisonError::into_inner) = Some(listed);
listed
}
pub(crate) fn body(&self, req: &VideoRequest, effort: Option<&str>) -> Value {
let mut body = json!({
"model": self.cfg.model,
"max_tokens": req.max_output_tokens,
"messages": [{
"role": "user",
"content": [
{ "type": "text", "text": req.prompt },
{ "type": "video_url", "video_url": { "url": req.url } },
],
}],
});
if let Some(effort) = effort {
body["reasoning"] = json!({ "effort": effort });
}
body
}
}
fn refusal_in(body: &Value) -> Option<String> {
let error = body.get("error")?;
let said = error.get("message").and_then(Value::as_str)?.trim();
let upstream = error
.pointer("/metadata/raw")
.and_then(Value::as_str)
.and_then(|raw| serde_json::from_str::<Value>(raw).ok())
.and_then(|raw| Some(raw.pointer("/error/message")?.as_str()?.trim().to_string()))
.filter(|m| !m.is_empty() && m != said);
Some(match upstream {
Some(upstream) => format!("{said}: {upstream}"),
None => said.to_string(),
})
}
fn answer_from(status: reqwest::StatusCode, text: &str) -> Result<VideoAnswer> {
let body: Value = match serde_json::from_str(text) {
Ok(body) => body,
Err(_) => {
let detail: String = text.trim().chars().take(500).collect();
anyhow::bail!("{SUBJECT}: status {status}: {detail}");
}
};
if let Some(said) = refusal_in(&body) {
anyhow::bail!("{SUBJECT}: status {status}: {said}");
}
anyhow::ensure!(status.is_success(), "{SUBJECT}: status {status}");
let usage = body.get("usage");
let tokens = |path: &str| usage.and_then(|u| u.pointer(path)).and_then(Value::as_u64);
if tokens("/prompt_tokens_details/video_tokens").unwrap_or(0) == 0 {
anyhow::bail!(
"{SUBJECT}: the gateway answered without reading the video — no video tokens \
among the {} prompt tokens it counted — so what it said is not about the video",
tokens("/prompt_tokens").unwrap_or(0)
);
}
let choice = body.pointer("/choices/0");
let finish = choice
.and_then(|c| c.get("finish_reason"))
.and_then(Value::as_str)
.unwrap_or_default();
let said = choice
.and_then(|c| c.pointer("/message/content"))
.and_then(Value::as_str)
.unwrap_or_default();
if said.trim().is_empty() {
if finish.is_empty() {
anyhow::bail!("{SUBJECT}: no description of the video came back");
}
anyhow::bail!("{SUBJECT}: no description of the video came back (finish reason: {finish})");
}
Ok(VideoAnswer {
text: said.to_string(),
truncated: finish == "length",
})
}
#[async_trait::async_trait]
impl VideoUnderstanding for GatewayVideo {
async fn describe(&self, req: VideoRequest, cancel: &CancellationToken) -> Result<VideoAnswer> {
let effort = self.lowest_effort(cancel).await;
let url = format!("{}/chat/completions", self.cfg.base_url);
let rb = self
.signed(self.http.post(&url))
.json(&self.body(&req, effort));
let resp = tokio::select! {
biased;
_ = cancel.cancelled() => anyhow::bail!("video request cancelled"),
r = rb.send() => r.with_context(|| format!("POST {url}"))?,
};
let status = resp.status();
let text = tokio::select! {
biased;
_ = cancel.cancelled() => anyhow::bail!("video request cancelled"),
t = resp.text() => t.context("reading the gateway's answer about the video")?,
};
let answer = answer_from(status, &text);
if let Err(err) = &answer {
tracing::warn!(%status, error = %err, "the gateway did not describe the video");
}
answer
}
fn reads_segments(&self) -> bool {
false
}
}