//! HTTP client for clawdforge. use std::path::Path; use std::time::Duration; use reqwest::header::{HeaderMap, HeaderValue}; use reqwest::multipart::{Form, Part}; use reqwest::{Body, Method, Response, StatusCode}; use serde::de::DeserializeOwned; use tokio_util::io::ReaderStream; use url::Url; use crate::error::Error; use crate::session::{ Session, SessionCloseResponse, SessionCreateResponse, SessionList, SessionOptions, SessionState, TurnResult, }; use crate::types::{ AppToken, FileToken, Healthz, RunRequest, RunResult, TokenCreateRequest, TokenList, }; /// Default request timeout if neither the builder nor the per-call helper sets /// one. 120 s leaves headroom over the server's default 60 s `claude` timeout /// without making `/healthz` callers wait forever on a dead host. const DEFAULT_TIMEOUT: Duration = Duration::from_secs(120); /// Async client for the clawdforge HTTP API. /// /// Construct via [`Client::builder`]. `Client` is cheap to clone — internally /// it wraps an `Arc`-backed `reqwest::Client`. /// /// `Debug` is hand-written to redact bearer tokens — `format!("{:?}", client)` /// will never expose `app_token` or `admin_token` plaintext. #[derive(Clone)] pub struct Client { inner: reqwest::Client, base: Url, /// App-level bearer (used for `/run`, `/files`, `/healthz`). Optional so /// admin-only callers don't have to mint a worthless app token. app_token: Option, /// Admin bootstrap token (used for `/admin/*`). Optional. admin_token: Option, /// Optional cap on `upload_file` size in bytes — `None` = no cap. max_upload_bytes: Option, } impl std::fmt::Debug for Client { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("Client") .field("base_url", &self.base.as_str()) .field("app_token", &self.app_token.as_ref().map(|_| "")) .field( "admin_token", &self.admin_token.as_ref().map(|_| ""), ) .field("max_upload_bytes", &self.max_upload_bytes) .finish_non_exhaustive() } } /// Builder for [`Client`]. /// /// `Debug` is hand-written to redact bearer tokens. #[derive(Default)] pub struct ClientBuilder { base_url: Option, app_token: Option, admin_token: Option, timeout: Option, user_agent: Option, danger_accept_invalid_certs: bool, max_upload_bytes: Option, } impl std::fmt::Debug for ClientBuilder { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("ClientBuilder") .field("base_url", &self.base_url) .field("app_token", &self.app_token.as_ref().map(|_| "")) .field( "admin_token", &self.admin_token.as_ref().map(|_| ""), ) .field("timeout", &self.timeout) .field("user_agent", &self.user_agent) .field( "danger_accept_invalid_certs", &self.danger_accept_invalid_certs, ) .field("max_upload_bytes", &self.max_upload_bytes) .finish_non_exhaustive() } } impl Client { /// Start building a client. pub fn builder() -> ClientBuilder { ClientBuilder::default() } /// Base URL the client was configured with (trailing slash trimmed). pub fn base_url(&self) -> &str { self.base.as_str().trim_end_matches('/') } // ---- public API -------------------------------------------------------- /// `GET /healthz`. Does not require an app token, but the server still /// enforces the global IP allowlist. pub async fn healthz(&self) -> Result { let url = self.url("/healthz")?; let req = self.inner.request(Method::GET, url); // healthz works without a token, but if we have one, send it — the // server treats it as a no-op. let req = match &self.app_token { Some(t) => req.bearer_auth(t), None => req, }; let resp = req.send().await?; json_or_error(resp).await } /// `POST /run`. Returns the parsed [`RunResult`] on success. On HTTP 502 /// the body is surfaced as [`Error::Api`] with `status = 502` and the /// failure JSON in `body` — see [`crate::types::RunFailure`] for the /// structured form. pub async fn run(&self, body: RunRequest) -> Result { let token = self .app_token .as_deref() .ok_or_else(|| Error::Auth("no app token configured".into()))?; let url = self.url("/run")?; let resp = self .inner .post(url) .bearer_auth(token) .json(&body) .send() .await?; json_or_error(resp).await } /// `POST /files`. Streams the file from disk via `tokio::fs::File`; large /// uploads do not buffer fully in memory. /// /// `ttl_secs` defaults to the server's 3600 if `None`. Server clamps to /// `60..=86400`. /// /// If [`ClientBuilder::max_upload_bytes`] was set and the file's size on /// disk exceeds it, returns [`Error::Config`] before opening any network /// connection. pub async fn upload_file( &self, path: impl AsRef, ttl_secs: Option, ) -> Result { let token = self .app_token .as_deref() .ok_or_else(|| Error::Auth("no app token configured".into()))?; let path = path.as_ref(); let file_name = path .file_name() .and_then(|n| n.to_str()) .unwrap_or("upload") .to_string(); let file = tokio::fs::File::open(path).await?; let len = file.metadata().await?.len(); if let Some(max) = self.max_upload_bytes { if len > max { return Err(Error::Config(format!( "file size {len} bytes exceeds max_upload_bytes={max}" ))); } } let stream = ReaderStream::new(file); let body = Body::wrap_stream(stream); let part = Part::stream_with_length(body, len) .file_name(file_name) .mime_str("application/octet-stream") .map_err(|e| Error::Config(format!("invalid mime: {e}")))?; let mut form = Form::new().part("file", part); if let Some(t) = ttl_secs { form = form.text("ttl_secs", t.to_string()); } let url = self.url("/files")?; let resp = self .inner .post(url) .bearer_auth(token) .multipart(form) .send() .await?; json_or_error(resp).await } /// `POST /admin/tokens`. Requires an admin token on the client. pub async fn create_token(&self, body: TokenCreateRequest) -> Result { let token = self.require_admin()?; let url = self.url("/admin/tokens")?; let resp = self .inner .post(url) .bearer_auth(token) .json(&body) .send() .await?; json_or_error(resp).await } /// `GET /admin/tokens`. Requires an admin token on the client. pub async fn list_tokens(&self) -> Result { let token = self.require_admin()?; let url = self.url("/admin/tokens")?; let resp = self.inner.get(url).bearer_auth(token).send().await?; json_or_error(resp).await } /// `DELETE /admin/tokens/{name}`. Requires an admin token on the client. /// Returns `Ok(())` on success, [`Error::Api`] with status 404 if the /// token does not exist. /// /// `name` is validated client-side to match the server's /// `[a-z0-9][a-z0-9_-]{0,63}` constraint — anything containing `/`, `?`, /// `#`, `..`, or empty short-circuits with [`Error::Config`] before a /// request is sent. This is defense-in-depth against path traversal via /// `Url::join` (which honors RFC 3986 `..` resolution). pub async fn revoke_token(&self, name: &str) -> Result<(), Error> { if name.is_empty() || name.contains('/') || name.contains('?') || name.contains('#') || name.contains("..") { return Err(Error::Config(format!("invalid token name: {name:?}"))); } let token = self.require_admin()?; let url = self.url(&format!("/admin/tokens/{name}"))?; let resp = self.inner.delete(url).bearer_auth(token).send().await?; // 2xx is success regardless of body — RFC-correct DELETE may return // 204 No Content with no body. if resp.status().is_success() { return Ok(()); } // Non-2xx: route through json_or_error to get Auth/Api mapping. // Discard the (already-non-success) deserialization slot. let _: serde_json::Value = json_or_error(resp).await?; Ok(()) } // ---- v0.2 multi-turn / sessions --------------------------------------- /// `POST /sessions`. Create a new multi-turn session and return a /// [`Session`] handle bound to this client. /// /// The handle owns a clone of the client; dropping it without an explicit /// `Session::close().await?` triggers a best-effort async DELETE via /// `tokio::spawn`. See [`Session`] for the full lifecycle contract. pub async fn new_session(&self, opts: SessionOptions) -> Result { let token = self.require_app()?; let url = self.url("/sessions")?; let resp = self .inner .post(url) .bearer_auth(token) .json(&opts) .send() .await?; let created: SessionCreateResponse = json_or_error(resp).await?; Ok(Session { client: self.clone(), session_id: created.session_id, agent: created.agent, created_at: created.created_at, closed: std::sync::atomic::AtomicBool::new(false), }) } /// `GET /sessions`. List all sessions visible to the calling app token. pub async fn list_sessions(&self) -> Result { let token = self.require_app()?; let url = self.url("/sessions")?; let resp = self.inner.get(url).bearer_auth(token).send().await?; json_or_error(resp).await } /// `GET /sessions/{id}`. Fetch the current state of a session. /// /// `id` is validated client-side against the same path-traversal guard as /// [`Self::revoke_token`] — anything containing `/`, `?`, `#`, `..`, or /// empty short-circuits with [`Error::Config`]. pub async fn get_session(&self, id: &str) -> Result { validate_session_id(id)?; let token = self.require_app()?; let url = self.url(&format!("/sessions/{id}"))?; let resp = self.inner.get(url).bearer_auth(token).send().await?; json_or_error(resp).await } /// Internal helper used by both [`Session::close`] and [`Session`]'s /// `Drop` impl to issue `DELETE /sessions/{id}`. pub(crate) async fn close_session_internal(&self, id: &str) -> Result<(), Error> { validate_session_id(id)?; let token = self.require_app()?; let url = self.url(&format!("/sessions/{id}"))?; let resp = self.inner.delete(url).bearer_auth(token).send().await?; if resp.status().is_success() { // Body is informational — `{ ok, already_closed? }`. Drain and // ignore decode failure (server may legitimately 204). let bytes = resp.bytes().await?; if !bytes.is_empty() { let _ = serde_json::from_slice::(&bytes); } return Ok(()); } // Funnel non-2xx through json_or_error for Auth/Api mapping. let _: serde_json::Value = json_or_error(resp).await?; Ok(()) } /// Internal helper used by [`Session::turn`] / /// [`Session::turn_with_files`] to dispatch to /// `POST /sessions/{id}/turn`. pub(crate) async fn turn_internal( &self, id: &str, body: &B, ) -> Result { validate_session_id(id)?; let token = self.require_app()?; let url = self.url(&format!("/sessions/{id}/turn"))?; let resp = self .inner .post(url) .bearer_auth(token) .json(body) .send() .await?; json_or_error(resp).await } // ---- internal ---------------------------------------------------------- fn require_app(&self) -> Result<&str, Error> { self.app_token .as_deref() .ok_or_else(|| Error::Auth("no app token configured".into())) } fn require_admin(&self) -> Result<&str, Error> { self.admin_token .as_deref() .ok_or_else(|| Error::Auth("no admin token configured".into())) } fn url(&self, path: &str) -> Result { let trimmed = path.strip_prefix('/').unwrap_or(path); self.base .join(trimmed) .map_err(|e| Error::Config(format!("bad path {path:?}: {e}"))) } } impl ClientBuilder { /// Set the base URL (e.g. `http://localhost:8800`). Required. pub fn base_url(mut self, url: impl Into) -> Self { self.base_url = Some(url.into()); self } /// Set the app bearer token used for `/run`, `/files`, and `/healthz`. pub fn token(mut self, token: impl Into) -> Self { self.app_token = Some(token.into()); self } /// Set the admin bootstrap token used for `/admin/*`. May be set /// alongside [`Self::token`]. pub fn admin_token(mut self, token: impl Into) -> Self { self.admin_token = Some(token.into()); self } /// Per-request timeout for the underlying `reqwest::Client`. Defaults to /// 120 s. pub fn timeout(mut self, timeout: Duration) -> Self { self.timeout = Some(timeout); self } /// Override the `User-Agent` header. Defaults to /// `clawdforge-rs/`. pub fn user_agent(mut self, ua: impl Into) -> Self { self.user_agent = Some(ua.into()); self } /// Skip TLS certificate verification. Off by default. Only useful against /// self-signed local deployments. pub fn danger_accept_invalid_certs(mut self, enable: bool) -> Self { self.danger_accept_invalid_certs = enable; self } /// Maximum file size (in bytes) accepted by [`Client::upload_file`]. /// Files exceeding this cap fail with [`Error::Config`] before any /// network I/O. Default `None` = no client-side cap (the server's own /// limit still applies). pub fn max_upload_bytes(mut self, max: u64) -> Self { self.max_upload_bytes = Some(max); self } /// Finalize. Errors if `base_url` is missing or unparseable. pub fn build(self) -> Result { let base_raw = self .base_url .ok_or_else(|| Error::Config("base_url is required".into()))?; // Ensure trailing slash so `Url::join` treats the base as a directory. let base_str = if base_raw.ends_with('/') { base_raw } else { format!("{base_raw}/") }; let base = Url::parse(&base_str).map_err(|e| Error::Config(format!("invalid base_url: {e}")))?; if !matches!(base.scheme(), "http" | "https") { return Err(Error::Config(format!( "unsupported scheme: {}", base.scheme() ))); } let ua = self .user_agent .unwrap_or_else(|| format!("clawdforge-rs/{}", env!("CARGO_PKG_VERSION"))); let mut headers = HeaderMap::new(); headers.insert( reqwest::header::ACCEPT, HeaderValue::from_static("application/json"), ); // We don't preset Authorization here — per-call helpers do it because // the right token depends on which endpoint is being hit. let inner = reqwest::Client::builder() .timeout(self.timeout.unwrap_or(DEFAULT_TIMEOUT)) .user_agent(ua) .default_headers(headers) .danger_accept_invalid_certs(self.danger_accept_invalid_certs) .build()?; Ok(Client { inner, base, app_token: self.app_token, admin_token: self.admin_token, max_upload_bytes: self.max_upload_bytes, }) } } /// Decode `T` from a successful 2xx response, otherwise lift to [`Error`]. async fn json_or_error(resp: Response) -> Result { let status = resp.status(); if status.is_success() { let bytes = resp.bytes().await?; return Ok(serde_json::from_slice::(&bytes)?); } // Non-2xx: capture body lossily then translate. let body = resp.text().await.unwrap_or_default(); match status { StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => Err(Error::Auth(format!( "{}: {}", status.as_u16(), truncate(&body, 500) ))), _ => Err(Error::api(status.as_u16(), body)), } } /// Defense-in-depth path validator for session ids in `/sessions/{id}*`. Same /// shape as [`Client::revoke_token`]'s name guard — RFC 3986 dot-segment /// resolution inside `Url::join` is the threat model here. fn validate_session_id(id: &str) -> Result<(), Error> { if id.is_empty() || id.contains('/') || id.contains('?') || id.contains('#') || id.contains("..") { return Err(Error::Config(format!("invalid session id: {id:?}"))); } Ok(()) } /// Truncate `s` to at most `max` bytes, snapping down to the nearest UTF-8 /// codepoint boundary so we never panic on multibyte sequences. Appends `…` /// to the truncated form. `str::floor_char_boundary` (stable 1.80+) does the /// boundary math. fn truncate(s: &str, max: usize) -> String { if s.len() <= max { s.to_string() } else { let safe = s.floor_char_boundary(max); format!("{}…", &s[..safe]) } }