codex_helper_core/
control_plane_client.rs

1use std::time::Duration;
2
3use anyhow::{Context, Result, anyhow};
4use reqwest::{Client, Url};
5use serde::Deserialize;
6use thiserror::Error;
7
8use crate::dashboard_core::{
9    ApiV1Capabilities, ApiV1OperatorSummary, ApiV1Snapshot, ProviderOption,
10};
11use crate::fleet::FleetSnapshot;
12use crate::proxy::{ADMIN_TOKEN_ENV_VAR, ADMIN_TOKEN_HEADER, RuntimeStatusResponse};
13use crate::request_chain::{RequestChainExport, RequestChainSelector};
14
15const ADMIN_DISCOVERY_PATH: &str = "/.well-known/codex-helper-admin";
16
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub struct ControlPlaneEndpoint {
19    pub admin_base_url: String,
20    pub admin_token_env: Option<String>,
21}
22
23#[derive(Debug, Clone)]
24pub struct ControlPlaneClient {
25    endpoint: ControlPlaneEndpoint,
26    client: Client,
27}
28
29#[derive(Debug, Error)]
30pub enum ControlPlaneError {
31    #[error("admin API not reachable at {base_url}: {source}")]
32    Transport {
33        base_url: String,
34        #[source]
35        source: reqwest::Error,
36    },
37    #[error("admin API returned {status}: {body}")]
38    HttpStatus { status: u16, body: String },
39    #[error("admin API response is not valid JSON: {source}")]
40    Decode {
41        #[source]
42        source: reqwest::Error,
43    },
44}
45
46#[derive(Debug, Clone, Deserialize)]
47pub struct AdminDiscoveryDocument {
48    pub api_version: u32,
49    pub service_name: String,
50    pub admin_base_url: String,
51}
52
53impl ControlPlaneEndpoint {
54    pub fn new(
55        admin_base_url: impl Into<String>,
56        admin_token_env: Option<impl Into<String>>,
57    ) -> Result<Self> {
58        let admin_base_url = normalize_base_url(&admin_base_url.into())
59            .ok_or_else(|| anyhow!("admin base URL is required"))?;
60        Ok(Self {
61            admin_base_url,
62            admin_token_env: admin_token_env.map(Into::into),
63        })
64    }
65}
66
67impl ControlPlaneClient {
68    pub fn new(endpoint: ControlPlaneEndpoint) -> Result<Self> {
69        let client = Client::builder()
70            .timeout(Duration::from_millis(1200))
71            .build()
72            .context("failed to build control-plane client")?;
73        Ok(Self { endpoint, client })
74    }
75
76    pub fn endpoint(&self) -> &ControlPlaneEndpoint {
77        &self.endpoint
78    }
79
80    pub async fn discover_admin_base_url(proxy_base_url: &str) -> Result<AdminDiscoveryDocument> {
81        let proxy_base_url = normalize_base_url(proxy_base_url)
82            .ok_or_else(|| anyhow!("proxy base URL is required"))?;
83        let url = format!("{proxy_base_url}{ADMIN_DISCOVERY_PATH}");
84        let response = Client::builder()
85            .timeout(Duration::from_millis(1200))
86            .build()
87            .context("failed to build discovery client")?
88            .get(url)
89            .send()
90            .await
91            .context("admin discovery request failed")?;
92        let status = response.status();
93        if !status.is_success() {
94            let body = response.text().await.unwrap_or_default();
95            anyhow::bail!("admin discovery returned {status}: {}", body.trim());
96        }
97        response
98            .json::<AdminDiscoveryDocument>()
99            .await
100            .context("admin discovery response is not valid JSON")
101    }
102
103    pub async fn fetch_json<T>(&self, path: &str) -> Result<T>
104    where
105        T: serde::de::DeserializeOwned,
106    {
107        self.fetch_json_classified(path).await.map_err(Into::into)
108    }
109
110    pub async fn fetch_json_classified<T>(&self, path: &str) -> Result<T, ControlPlaneError>
111    where
112        T: serde::de::DeserializeOwned,
113    {
114        let url = format!("{}{}", self.endpoint.admin_base_url, path);
115        let mut request = self.client.get(url);
116        if let Some(token) = self.admin_token() {
117            request = request.header(ADMIN_TOKEN_HEADER, token);
118        }
119
120        let response = request
121            .send()
122            .await
123            .map_err(|source| ControlPlaneError::Transport {
124                base_url: self.endpoint.admin_base_url.clone(),
125                source,
126            })?;
127        let status = response.status();
128        if !status.is_success() {
129            let body = response.text().await.unwrap_or_default();
130            return Err(ControlPlaneError::HttpStatus {
131                status: status.as_u16(),
132                body: body.trim().to_string(),
133            });
134        }
135        response
136            .json::<T>()
137            .await
138            .map_err(|source| ControlPlaneError::Decode { source })
139    }
140
141    pub async fn runtime_status(&self) -> Result<RuntimeStatusResponse> {
142        self.fetch_json("/__codex_helper/api/v1/runtime/status")
143            .await
144    }
145
146    pub async fn capabilities(&self) -> Result<ApiV1Capabilities> {
147        self.fetch_json("/__codex_helper/api/v1/capabilities").await
148    }
149
150    pub async fn operator_summary(&self) -> Result<ApiV1OperatorSummary> {
151        self.fetch_json("/__codex_helper/api/v1/operator/summary")
152            .await
153    }
154
155    pub async fn snapshot(&self, recent_limit: usize, stats_days: usize) -> Result<ApiV1Snapshot> {
156        self.fetch_json(&format!(
157            "/__codex_helper/api/v1/snapshot?recent_limit={recent_limit}&stats_days={stats_days}"
158        ))
159        .await
160    }
161
162    pub async fn fleet_snapshot(&self) -> Result<FleetSnapshot, ControlPlaneError> {
163        self.fetch_json_classified("/__codex_helper/api/v1/fleet/snapshot")
164            .await
165    }
166
167    pub async fn request_chain(
168        &self,
169        selector: RequestChainSelector,
170        limit: usize,
171    ) -> Result<RequestChainExport> {
172        self.fetch_json(&request_chain_path(selector, limit)).await
173    }
174
175    pub async fn providers(&self) -> Result<Vec<ProviderOption>> {
176        self.fetch_json("/__codex_helper/api/v1/providers").await
177    }
178
179    fn admin_token(&self) -> Option<String> {
180        let env_name = self
181            .endpoint
182            .admin_token_env
183            .as_deref()
184            .unwrap_or(ADMIN_TOKEN_ENV_VAR);
185        std::env::var(env_name)
186            .ok()
187            .map(|value| value.trim().to_string())
188            .filter(|value| !value.is_empty())
189    }
190}
191
192pub fn normalize_base_url(value: &str) -> Option<String> {
193    let value = value.trim().trim_end_matches('/');
194    if value.is_empty() {
195        return None;
196    }
197    let url = Url::parse(value).ok()?;
198    match url.scheme() {
199        "http" | "https" => Some(value.to_string()),
200        _ => None,
201    }
202}
203
204fn request_chain_path(selector: RequestChainSelector, limit: usize) -> String {
205    let selector = selector.normalized();
206    let mut url =
207        Url::parse("http://localhost/__codex_helper/api/v1/request-ledger/chain").expect("url");
208    {
209        let mut pairs = url.query_pairs_mut();
210        pairs.append_pair("limit", &limit.to_string());
211        if let Some(trace_id) = selector.trace_id {
212            pairs.append_pair("trace_id", &trace_id);
213        }
214        if let Some(request_id) = selector.request_id {
215            pairs.append_pair("request_id", &request_id.to_string());
216        }
217        if let Some(session_id) = selector.session_id {
218            pairs.append_pair("session", &session_id);
219        }
220    }
221    match url.query() {
222        Some(query) => format!("{}?{query}", url.path()),
223        None => url.path().to_string(),
224    }
225}
226
227#[cfg(test)]
228mod tests {
229    use super::*;
230
231    #[test]
232    fn control_plane_endpoint_normalizes_admin_base_url() {
233        let endpoint = ControlPlaneEndpoint::new(" http://nas.local:4211/ ", Some("TOKEN_ENV"))
234            .expect("endpoint");
235
236        assert_eq!(endpoint.admin_base_url, "http://nas.local:4211");
237        assert_eq!(endpoint.admin_token_env.as_deref(), Some("TOKEN_ENV"));
238    }
239
240    #[test]
241    fn control_plane_endpoint_rejects_non_http_url() {
242        let err = ControlPlaneEndpoint::new("file:///tmp/socket", None::<String>)
243            .expect_err("non-http admin url should fail");
244
245        assert!(err.to_string().contains("admin base URL"));
246    }
247
248    #[test]
249    fn request_chain_path_encodes_selector_query_values() {
250        let path = request_chain_path(
251            RequestChainSelector {
252                trace_id: Some("trace/with space".to_string()),
253                request_id: Some(42),
254                session_id: Some("session a".to_string()),
255            },
256            20,
257        );
258
259        assert_eq!(
260            path,
261            "/__codex_helper/api/v1/request-ledger/chain?limit=20&trace_id=trace%2Fwith+space&request_id=42&session=session+a"
262        );
263    }
264}
codex_helper_core/control_plane_client.rs

codex_helper_core/
control_plane_client.rs
