vtcode_llm/providers/
merge_gateway_catalog.rs

1//! Merge Gateway authenticated model catalog discovery.
2//!
3//! This module fetches the native `/v1/models` catalog, paginates cursor-based
4//! results, and normalizes vendor-specific capability data into vtcode-core
5//! friendly snapshot structures.
6
7use anyhow::{Context, Result, bail, ensure};
8use reqwest::{
9    Client, StatusCode,
10    header::{ACCEPT, IF_NONE_MATCH},
11};
12use serde::{Deserialize, Serialize};
13use serde_json::Value;
14use std::collections::{BTreeMap, BTreeSet, HashSet};
15use std::time::Duration;
16use vtcode_commons::tool_types::CompactStr;
17use vtcode_config::TimeoutsConfig;
18
19use super::merge_gateway_contract::{
20    MergeAvailabilityStatus, MergeInputModality, MergeModelCatalogResponse, MergeModelRecord, MergeModelsListQuery,
21    MergeServiceTier, MergeVendorModelInfo,
22};
23
24const CATALOG_PAGE_LIMIT: u32 = 500;
25const MAX_CATALOG_PAGES: usize = 1_000;
26const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 30;
27const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 120;
28
29/// Normalized catalog availability for vtcode-core.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
31#[serde(rename_all = "snake_case")]
32pub enum MergeCatalogAvailability {
33    Available,
34    Deprecated,
35    Unknown,
36}
37
38impl From<MergeAvailabilityStatus> for MergeCatalogAvailability {
39    fn from(value: MergeAvailabilityStatus) -> Self {
40        match value {
41            MergeAvailabilityStatus::Available => Self::Available,
42            MergeAvailabilityStatus::Deprecated => Self::Deprecated,
43            MergeAvailabilityStatus::Unknown => Self::Unknown,
44        }
45    }
46}
47
48/// Normalized service tiers for the catalog snapshot.
49#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
50#[serde(rename_all = "snake_case")]
51pub enum MergeCatalogServiceTier {
52    Standard,
53    Flex,
54    Priority,
55    Unknown,
56}
57
58impl From<MergeServiceTier> for MergeCatalogServiceTier {
59    fn from(value: MergeServiceTier) -> Self {
60        match value {
61            MergeServiceTier::Standard => Self::Standard,
62            MergeServiceTier::Flex => Self::Flex,
63            MergeServiceTier::Priority => Self::Priority,
64            MergeServiceTier::Unknown => Self::Unknown,
65        }
66    }
67}
68
69/// Optional Merge catalog filters preserved across pagination.
70#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
71pub struct MergeCatalogFilters {
72    pub model: Option<String>,
73    pub provider: Option<String>,
74    pub vendor: Option<String>,
75}
76
77/// Normalized Merge catalog model metadata.
78#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
79pub struct MergeCatalogModel {
80    pub model: String,
81    pub provider: String,
82    pub display_name: Option<String>,
83    pub availability: MergeCatalogAvailability,
84    pub context_window: Option<u32>,
85    pub max_output_tokens: Option<u32>,
86    pub supports_tool_use: bool,
87    pub supports_streaming: bool,
88    pub supports_vision: bool,
89    pub supports_structured_output: bool,
90    pub service_tiers: Vec<MergeCatalogServiceTier>,
91    pub supports_reasoning: bool,
92    pub reasoning_disable_supported: bool,
93    pub reasoning_controls: Vec<String>,
94}
95
96/// Full normalized snapshot of the Merge catalog.
97#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
98pub struct MergeCatalogSnapshot {
99    pub models: Vec<MergeCatalogModel>,
100    pub etag: Option<String>,
101}
102
103/// Authenticated Merge Gateway `/v1/models` discovery client.
104#[derive(Clone)]
105pub struct MergeGatewayCatalogClient {
106    api_key: String,
107    catalog_base_url: String,
108    http_client: Client,
109}
110
111impl MergeGatewayCatalogClient {
112    /// Creates a client using an injected HTTP client.
113    pub fn try_with_client(
114        api_key: impl Into<String>,
115        base_url: impl Into<String>,
116        http_client: Client,
117    ) -> Result<Self> {
118        Self::try_from_parts(api_key.into(), base_url.into(), http_client)
119    }
120
121    /// Creates a client using a timeout-aware HTTP client derived from `TimeoutsConfig`.
122    pub fn try_with_timeouts(
123        api_key: impl Into<String>,
124        base_url: impl Into<String>,
125        timeouts: Option<TimeoutsConfig>,
126    ) -> Result<Self> {
127        let http_client = build_http_client(timeouts.as_ref())?;
128        Self::try_from_parts(api_key.into(), base_url.into(), http_client)
129    }
130
131    /// Fetches the catalog snapshot. Returns `Ok(None)` when Merge replies `304 Not Modified`
132    /// and a non-empty `etag` validator was supplied.
133    pub async fn fetch_snapshot(
134        &self,
135        filters: &MergeCatalogFilters,
136        etag: Option<&str>,
137    ) -> Result<Option<MergeCatalogSnapshot>> {
138        let etag = normalize_optional_value(etag);
139        let mut cursor: Option<String> = None;
140        let mut seen_cursors: HashSet<String> = HashSet::new();
141        let mut models = Vec::new();
142        let mut snapshot_etag: Option<String> = None;
143        let mut page_index: usize = 1;
144
145        loop {
146            ensure!(
147                page_index <= MAX_CATALOG_PAGES,
148                "Merge Gateway catalog exceeded the maximum of {MAX_CATALOG_PAGES} pages"
149            );
150            let query = build_wire_query(filters, cursor.as_deref());
151            let mut request = self
152                .http_client
153                .get(self.models_endpoint_url())
154                .bearer_auth(&self.api_key)
155                .header(ACCEPT, "application/json")
156                .query(&query);
157
158            if page_index == 1 {
159                if let Some(etag) = etag.as_deref() {
160                    request = request.header(IF_NONE_MATCH, etag);
161                }
162            }
163
164            let response = request.send().await.with_context(|| {
165                format!(
166                    "Merge Gateway catalog request failed on page {page_index} ({})",
167                    describe_filters(filters, cursor.as_deref())
168                )
169            })?;
170
171            if response.status() == StatusCode::NOT_MODIFIED {
172                if page_index == 1 && etag.is_some() {
173                    return Ok(None);
174                }
175
176                bail!(
177                    "Merge Gateway catalog returned 304 Not Modified on page {page_index} without a usable If-None-Match validator"
178                );
179            }
180
181            if !response.status().is_success() {
182                bail!(
183                    "Merge Gateway catalog request failed on page {page_index} with HTTP {} ({})",
184                    response.status(),
185                    describe_filters(filters, cursor.as_deref())
186                );
187            }
188
189            if snapshot_etag.is_none() {
190                snapshot_etag = header_value_to_string(response.headers().get(reqwest::header::ETAG));
191            }
192
193            let raw: Value = response
194                .json()
195                .await
196                .with_context(|| format!("Merge Gateway catalog response on page {page_index} was not valid JSON"))?;
197            validate_catalog_envelope(&raw, page_index)?;
198            let page: MergeModelCatalogResponse = serde_json::from_value(raw)
199                .with_context(|| format!("Merge Gateway catalog response on page {page_index} was malformed"))?;
200
201            models.extend(page.data.into_iter().map(normalize_catalog_model));
202
203            if page.has_more {
204                let next_cursor = page
205                    .next_cursor
206                    .as_deref()
207                    .map(str::trim)
208                    .filter(|value| !value.is_empty())
209                    .map(ToOwned::to_owned)
210                    .ok_or_else(|| {
211                        anyhow::anyhow!(
212                            "Merge Gateway catalog page {page_index} set has_more=true without a non-empty next_cursor"
213                        )
214                    })?;
215
216                if !seen_cursors.insert(next_cursor.clone()) {
217                    bail!("Merge Gateway catalog repeated pagination cursor `{next_cursor}` on page {page_index}");
218                }
219
220                cursor = Some(next_cursor);
221                page_index += 1;
222                continue;
223            }
224
225            return Ok(Some(MergeCatalogSnapshot { models, etag: snapshot_etag }));
226        }
227    }
228
229    fn try_from_parts(api_key: String, base_url: String, http_client: Client) -> Result<Self> {
230        let api_key = api_key.trim().to_owned();
231        ensure!(!api_key.is_empty(), "Merge Gateway API key cannot be empty");
232
233        let catalog_base_url = normalize_catalog_base_url(&base_url)?;
234        Ok(Self { api_key, catalog_base_url, http_client })
235    }
236
237    fn models_endpoint_url(&self) -> String {
238        format!("{}/models", self.catalog_base_url)
239    }
240}
241
242fn build_http_client(timeouts: Option<&TimeoutsConfig>) -> Result<Client> {
243    let request_timeout = timeouts
244        .and_then(|config| config.ceiling_duration(config.default_ceiling_seconds))
245        .unwrap_or_else(|| Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS));
246
247    Client::builder()
248        .timeout(request_timeout)
249        .connect_timeout(Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS))
250        .build()
251        .context("failed to build Merge Gateway catalog HTTP client")
252}
253
254fn build_wire_query(filters: &MergeCatalogFilters, cursor: Option<&str>) -> MergeModelsListQuery {
255    MergeModelsListQuery {
256        model: normalize_optional_value(filters.model.as_deref()),
257        provider: normalize_optional_value(filters.provider.as_deref()),
258        vendor: normalize_optional_value(filters.vendor.as_deref()),
259        cursor: normalize_optional_value(cursor),
260        limit: Some(CATALOG_PAGE_LIMIT),
261    }
262}
263
264fn normalize_optional_value(value: Option<&str>) -> Option<CompactStr> {
265    value.map(str::trim).filter(|value| !value.is_empty()).map(CompactStr::from)
266}
267
268fn normalize_catalog_base_url(base_url: &str) -> Result<String> {
269    let mut normalized = base_url.trim().trim_end_matches('/').to_owned();
270    ensure!(!normalized.is_empty(), "Merge Gateway base URL cannot be empty");
271
272    loop {
273        let mut changed = false;
274        for suffix in ["/chat/completions", "/responses", "/openai", "/models"] {
275            if let Some(stripped) = normalized.strip_suffix(suffix) {
276                normalized = stripped.trim_end_matches('/').to_owned();
277                changed = true;
278                break;
279            }
280        }
281
282        if !changed {
283            break;
284        }
285    }
286
287    if !normalized.ends_with("/v1") && !normalized.contains("/v1/") {
288        normalized.push_str("/v1");
289    }
290
291    Ok(normalized)
292}
293
294fn validate_catalog_envelope(raw: &Value, page_index: usize) -> Result<()> {
295    let object = raw.as_object().and_then(|map| map.get("object")).and_then(Value::as_str);
296    ensure!(
297        object == Some("list"),
298        "Merge Gateway catalog envelope on page {page_index} must set object=\"list\""
299    );
300    Ok(())
301}
302
303fn header_value_to_string(value: Option<&reqwest::header::HeaderValue>) -> Option<String> {
304    value
305        .and_then(|header| header.to_str().ok())
306        .map(str::trim)
307        .filter(|value| !value.is_empty())
308        .map(ToOwned::to_owned)
309}
310
311fn describe_filters(filters: &MergeCatalogFilters, cursor: Option<&str>) -> String {
312    format!(
313        "cursor={cursor:?}, model={:?}, provider={:?}, vendor={:?}",
314        filters.model.as_deref(),
315        filters.provider.as_deref(),
316        filters.vendor.as_deref(),
317    )
318}
319
320fn normalize_catalog_model(record: MergeModelRecord) -> MergeCatalogModel {
321    let context_window = aggregate_min_context_window(&record.vendors);
322    let max_output_tokens = aggregate_min_max_output_tokens(&record.vendors);
323    let supports_tool_use = all_vendors_support(&record.vendors, |vendor| vendor.capabilities.supports_tool_calling);
324    let supports_streaming = all_vendors_support(&record.vendors, |vendor| vendor.capabilities.streaming);
325    let supports_vision = all_vendors_support(&record.vendors, |vendor| {
326        vendor
327            .capabilities
328            .input
329            .iter()
330            .any(|modality| matches!(modality, MergeInputModality::Image))
331    });
332    let supports_structured_output =
333        all_vendors_support(&record.vendors, |vendor| vendor.capabilities.supports_structured_outputs);
334    let service_tiers = aggregate_service_tiers(&record.vendors);
335    let supports_reasoning = all_vendors_support(&record.vendors, |vendor| vendor.capabilities.supports_reasoning());
336    let reasoning_disable_supported = !record.vendors.is_empty()
337        && record
338            .vendors
339            .values()
340            .all(|vendor| vendor.capabilities.reasoning_disable_supported());
341    let reasoning_controls = aggregate_reasoning_controls(&record.vendors);
342
343    MergeCatalogModel {
344        model: record.model.to_string(),
345        provider: record.provider.to_string(),
346        display_name: record.display_name.map(|value| value.to_string()),
347        availability: record.availability_status.into(),
348        context_window,
349        max_output_tokens,
350        supports_tool_use,
351        supports_streaming,
352        supports_vision,
353        supports_structured_output,
354        service_tiers,
355        supports_reasoning,
356        reasoning_disable_supported,
357        reasoning_controls,
358    }
359}
360
361fn aggregate_min_context_window(vendors: &BTreeMap<CompactStr, MergeVendorModelInfo>) -> Option<u32> {
362    vendors.values().filter_map(|vendor| vendor.context_window).min()
363}
364
365fn aggregate_min_max_output_tokens(vendors: &BTreeMap<CompactStr, MergeVendorModelInfo>) -> Option<u32> {
366    vendors.values().filter_map(|vendor| vendor.max_output_tokens).min()
367}
368
369fn all_vendors_support(
370    vendors: &BTreeMap<CompactStr, MergeVendorModelInfo>,
371    predicate: impl Fn(&MergeVendorModelInfo) -> bool,
372) -> bool {
373    !vendors.is_empty() && vendors.values().all(predicate)
374}
375
376fn aggregate_service_tiers(vendors: &BTreeMap<CompactStr, MergeVendorModelInfo>) -> Vec<MergeCatalogServiceTier> {
377    let mut intersection: Option<BTreeSet<MergeCatalogServiceTier>> = None;
378
379    for vendor in vendors.values() {
380        let tiers: BTreeSet<MergeCatalogServiceTier> = vendor
381            .service_tiers
382            .iter()
383            .copied()
384            .map(MergeCatalogServiceTier::from)
385            .collect();
386
387        intersection = Some(match intersection {
388            None => tiers,
389            Some(current) => current.intersection(&tiers).copied().collect(),
390        });
391    }
392
393    intersection.map(|tiers| tiers.into_iter().collect()).unwrap_or_default()
394}
395
396fn aggregate_reasoning_controls(vendors: &BTreeMap<CompactStr, MergeVendorModelInfo>) -> Vec<String> {
397    if !all_vendors_support(vendors, |vendor| vendor.capabilities.supports_reasoning()) {
398        return Vec::new();
399    }
400
401    let mut controls = BTreeSet::new();
402    for vendor in vendors.values() {
403        controls.extend(
404            vendor
405                .capabilities
406                .reasoning_controls()
407                .into_iter()
408                .map(|control| control.to_string()),
409        );
410    }
411    controls.into_iter().collect()
412}
413
414#[cfg(test)]
415mod tests;
vtcode_llm/providers/merge_gateway_catalog.rs

vtcode_llm/providers/
merge_gateway_catalog.rs
