Start direct API requests when called

src/api/btql.rs

@@ -1,5 +1,5 @@
 //! BTQL dataset readback, independent of SDK ingestion.
-use super::{Client, Error, Plane};
+use super::{Client, Error, Plane, Transport};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 
@@ -63,8 +63,23 @@ pub struct Btql {
     pub(crate) client: Client,
 }
 impl Btql {
+    /// Read one dataset page without pagination loops or visibility polling.
+    /// Submission is immediate; dropping the token does not cancel accepted work.
+    pub fn query_dataset_records(
+        &self,
+        query: BTQLQuery,
+    ) -> crate::Completion<BTQLResponse, Error> {
+        self.client
+            .submit(move |transport| async move { transport.query_dataset_records(query).await })
+    }
+}
+
+impl Transport {
     /// Execute one dataset page request; no visibility polling or pagination loop.
-    pub async fn query_dataset_records(&self, query: BTQLQuery) -> Result<BTQLResponse, Error> {
-        self.client.post_json(Plane::Data, "btql", &query).await
+    pub(crate) async fn query_dataset_records(
+        &self,
+        query: BTQLQuery,
+    ) -> Result<BTQLResponse, Error> {
+        self.post_json(Plane::Data, "btql", &query).await
     }
 }

src/api/client.rs

@@ -1,88 +1,48 @@
-use std::sync::Arc;
-
-use reqwest::{
-    header::{HeaderMap, HeaderValue, AUTHORIZATION},
-    RequestBuilder, Response,
+use std::{
+    future::Future,
+    sync::{Arc, Mutex},
+    time::Duration,
 };
-use serde::{de::DeserializeOwned, Serialize};
-use url::Url;
-
-use super::{Config, Error};
-use crate::http::{build_http_client_builder, CaBundleConfig};
 
-#[derive(Clone, Copy)]
-pub(crate) enum Plane {
-    Control,
-    Data,
-}
+use super::{Config, Error, Transport};
+use crate::internal::{
+    execution::ExecutionBinding,
+    shared_outcome::{SharedOutcome, SharedResponder},
+};
+use crate::{Completion, ExecutionError, ShutdownError};
 
-/// Immutable HTTP client shared by lightweight endpoint views.
+/// Eager one-attempt API operations sharing one execution binding and lifecycle.
+/// Clones and endpoint views share admission and shutdown state. Construction
+/// validates local configuration and starts the selected binding without login.
 #[derive(Clone)]
 pub struct Client {
-    inner: Arc<Inner>,
-}
-struct Inner {
-    http: reqwest::Client,
-    api_url: Url,
-    app_url: Url,
-    headers: HeaderMap,
+    transport: Transport,
+    execution: ExecutionBinding,
+    shutdown: Arc<Mutex<Option<SharedOutcome<(), ShutdownError>>>>,
 }
 impl Client {
     pub fn new(config: Config) -> Result<Self, Error> {
-        let api_url = base_url(&config.api_url)?;
-        let app_url = base_url(&config.app_url)?;
-        if config.api_key.trim().is_empty() {
-            return Err(Error::Config("api_key must not be empty".into()));
-        }
-        let headers = scoped_headers(&config.api_key, config.org_name.as_deref())?;
-        let ca = match (config.custom_ca_bundle_path, config.custom_ca_bundle_pem) {
-            (Some(_), Some(_)) => {
-                return Err(Error::Config(
-                    "choose a CA file or PEM data, not both".into(),
-                ))
-            }
-            (Some(path), None) => Some(CaBundleConfig::File(path)),
-            (None, Some(pem)) => Some(CaBundleConfig::from_pem(pem, "API configuration")),
-            (None, None) => CaBundleConfig::from_environment(),
-        };
-        let http =
-            build_http_client_builder(config.timeout, ca.as_ref(), config.user_agent.as_deref())
-                .map_err(|err| Error::Config(err.to_string()))?
-                // Redirects would be additional HTTP requests and can change routing.
-                .redirect(reqwest::redirect::Policy::none())
-                .build()
-                .map_err(|_| Error::Config("failed to build HTTP client".into()))?;
+        let runtime = config.runtime.clone();
+        let transport = Transport::new(config)?;
+        let execution = ExecutionBinding::new(runtime).map_err(|error| match error {
+            crate::BraintrustError::Execution(error) => Error::Execution(error),
+            error => Error::Config(error.to_string()),
+        })?;
         Ok(Self {
-            inner: Arc::new(Inner {
-                http,
-                api_url,
-                app_url,
-                headers,
-            }),
+            transport,
+            execution,
+            shutdown: Arc::new(Mutex::new(None)),
         })
     }
-    /// Reuse the SDK's configured transport without copying session state into
-    /// the API layer. Credentials and endpoint routes stay immutable.
-    pub(crate) fn for_scope(
-        http: reqwest::Client,
-        api_url: &Url,
-        app_url: &Url,
-        token: &str,
-        org_name: Option<&str>,
-    ) -> Result<Self, Error> {
-        let api_url = base_url(api_url.as_str())?;
-        let app_url = base_url(app_url.as_str())?;
-        let headers = scoped_headers(token, org_name)?;
-        Ok(Self {
-            inner: Arc::new(Inner {
-                http,
-                app_url,
-                api_url,
-                headers,
-            }),
-        })
+    pub(crate) fn submit<T, F, W>(&self, make: F) -> Completion<T, Error>
+    where
+        T: Send + 'static,
+        F: FnOnce(Transport) -> W,
+        W: Future<Output = Result<T, Error>> + Send + 'static,
+    {
+        self.execution
+            .response_try(|| Ok(make(self.transport.clone())))
     }
-
     pub fn logs3(&self) -> super::logs3::Logs3 {
         super::logs3::Logs3 {
             client: self.clone(),
@@ -108,160 +68,43 @@ impl Client {
             client: self.clone(),
         }
     }
-    /// One login request. Organization selection and client reconfiguration are explicit.
-    pub async fn login(&self) -> Result<super::login::LoginResponse, Error> {
-        let response = self
-            .send(
-                self.request(Plane::Control, "api/apikey/login", true)?
-                    .header("content-type", "application/json"),
-                "api/apikey/login",
-            )
-            .await?;
-        decode(response, "api/apikey/login").await
-    }
-    fn request(
-        &self,
-        plane: Plane,
-        path: &'static str,
-        post: bool,
-    ) -> Result<RequestBuilder, Error> {
-        let base = if matches!(plane, Plane::Control) {
-            &self.inner.app_url
-        } else {
-            &self.inner.api_url
-        };
-        let url = base
-            .join(path)
-            .map_err(|_| Error::Config("invalid endpoint path".into()))?;
-        let request = if post {
-            self.inner.http.post(url)
-        } else {
-            self.inner.http.get(url)
-        };
-        Ok(request.headers(self.inner.headers.clone()))
-    }
-    async fn send(
-        &self,
-        request: RequestBuilder,
-        operation: &'static str,
-    ) -> Result<Response, Error> {
-        let response = request
-            .send()
-            .await
-            .map_err(|source| transport(operation, source))?;
-        if response.status().is_success() {
-            return Ok(response);
+    /// Submit one login request without changing this client's immutable scope.
+    pub fn login(&self) -> Completion<super::login::LoginResponse, Error> {
+        self.submit(|transport| async move { transport.login().await })
+    }
+    /// Close admission immediately and wait for accepted requests under one
+    /// deadline. Repeated calls observe the first deadline and outcome. Request
+    /// failures remain on their own tokens and are not a shutdown failure log.
+    pub fn shutdown(&self, timeout: Duration) -> Completion<(), ShutdownError> {
+        let mut state = self
+            .shutdown
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if let Some(outcome) = &*state {
+            return outcome.observe();
         }
-        let status = response.status().as_u16();
-        let body = response
-            .text()
-            .await
-            .map_err(|source| transport(operation, source))?;
-        Err(Error::Http {
-            operation,
-            status,
-            body,
-        })
-    }
-    pub(crate) async fn get_json<T: DeserializeOwned>(
-        &self,
-        plane: Plane,
-        path: &'static str,
-        query: &[(&str, &str)],
-    ) -> Result<T, Error> {
-        let response = self
-            .send(self.request(plane, path, false)?.query(query), path)
-            .await?;
-        decode(response, path).await
+        let outcome = SharedOutcome::default();
+        let (sender, completion) = Completion::pair();
+        let responder = SharedResponder::new(sender, Some(outcome.clone()));
+        *state = Some(outcome);
+        self.execution.close_responses();
+        let execution = self.execution.clone();
+        let deadline = tokio::time::Instant::now().checked_add(timeout);
+        self.execution.spawn(async move {
+            let work = execution.responses_drained();
+            let result = match deadline {
+                Some(deadline) => tokio::time::timeout_at(deadline, work).await,
+                None => Ok(work.await),
+            };
+            let result = match result {
+                Ok(result) => result.map_err(ShutdownError::Execution),
+                Err(_) => {
+                    execution.cancel_responses(ExecutionError::TimedOut);
+                    Err(ShutdownError::Execution(ExecutionError::TimedOut))
+                }
+            };
+            let _ = responder.send(result);
+        });
+        completion
     }
-    pub(crate) async fn post_json<Q: Serialize + ?Sized, T: DeserializeOwned>(
-        &self,
-        plane: Plane,
-        path: &'static str,
-        body: &Q,
-    ) -> Result<T, Error> {
-        let response = self
-            .send(self.request(plane, path, true)?.json(body), path)
-            .await?;
-        decode(response, path).await
-    }
-    pub(crate) async fn post_empty<Q: Serialize + ?Sized>(
-        &self,
-        plane: Plane,
-        path: &'static str,
-        body: &Q,
-    ) -> Result<(), Error> {
-        self.send(self.request(plane, path, true)?.json(body), path)
-            .await?;
-        Ok(())
-    }
-    pub(crate) async fn post_bytes(
-        &self,
-        path: &'static str,
-        payload: Vec<u8>,
-    ) -> Result<(), Error> {
-        self.send(
-            self.request(Plane::Data, path, true)?
-                .header("content-type", "application/json")
-                .body(payload),
-            path,
-        )
-        .await?;
-        Ok(())
-    }
-}
-fn scoped_headers(token: &str, org_name: Option<&str>) -> Result<HeaderMap, Error> {
-    let mut headers = HeaderMap::new();
-    let mut token = HeaderValue::from_str(&format!("Bearer {token}"))
-        .map_err(|_| Error::Config("invalid api_key header".into()))?;
-    token.set_sensitive(true);
-    headers.insert(AUTHORIZATION, token);
-    if let Some(org) = org_name {
-        headers.insert(
-            "x-bt-org-name",
-            HeaderValue::from_str(org)
-                .map_err(|_| Error::Config("invalid organization header".into()))?,
-        );
-    }
-    Ok(headers)
-}
-
-fn base_url(value: &str) -> Result<Url, Error> {
-    let mut url = Url::parse(value).map_err(|_| Error::Config("invalid endpoint URL".into()))?;
-    if !matches!(url.scheme(), "http" | "https")
-        || url.host_str().is_none()
-        || !url.username().is_empty()
-        || url.password().is_some()
-        || url.query().is_some()
-        || url.fragment().is_some()
-    {
-        return Err(Error::Config(
-            "endpoint must be an HTTP(S) base URL without credentials, query or fragment".into(),
-        ));
-    }
-    if !url.path().ends_with('/') {
-        url.set_path(&format!("{}/", url.path()));
-    }
-    Ok(url)
-}
-fn transport(operation: &'static str, source: reqwest::Error) -> Error {
-    Error::Transport {
-        operation,
-        source: source.without_url(),
-    }
-}
-async fn decode<T: DeserializeOwned>(
-    response: Response,
-    operation: &'static str,
-) -> Result<T, Error> {
-    let status = response.status().as_u16();
-    let body = response
-        .bytes()
-        .await
-        .map_err(|source| transport(operation, source))?;
-    serde_json::from_slice(&body).map_err(|source| Error::Decode {
-        operation,
-        status,
-        source,
-    })
 }

src/api/config.rs

@@ -11,6 +11,10 @@ pub struct Config {
     pub app_url: String,
     pub org_name: Option<String>,
     pub timeout: Duration,
+    /// Explicit execution context. None owns a dedicated runtime; ambient Tokio
+    /// contexts are never selected automatically. A supplied runtime must remain
+    /// alive and driven until its accepted operations and shutdown complete.
+    pub runtime: Option<tokio::runtime::Handle>,
     pub user_agent: Option<String>,
     /// An explicit PEM file; mutually exclusive with `custom_ca_bundle_pem`.
     pub custom_ca_bundle_path: Option<PathBuf>,
@@ -25,6 +29,7 @@ impl Config {
             app_url: DEFAULT_APP_URL.into(),
             org_name: None,
             timeout: Duration::from_secs(30),
+            runtime: None,
             user_agent: None,
             custom_ca_bundle_path: None,
             custom_ca_bundle_pem: None,