scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-02 02:13:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3
fabro/lib/foundation/fabro-client/src/client.rs
Bryan Helmkamp 18a3c4741e
Run agent stages, Ask Fabro, and fabro exec on pebble's CodingAgent 
Replace fabro's hand-written agent loop with pebble's `CodingAgent` and
delete the `fabro-agent` crate.

Workflow: `PebbleBackend` builds one agent per stage over `RunSandbox`,
binds the stage's hooks as tool middleware, the interviewer as the
human-input provider, and a durable `EventSink` that writes every agent
event through the run event log before the agent goes on. Full-fidelity
threads continue across stages through `export`/`resume_from_export`.
Model failover takes the session record after the failed prompt and
continues it on the next route with `ResumeMode::UseModel`, so no tool
effect repeats. The steering hub targets pebble's control handle, with
a steering lease holding completion open while a human is paired.

Events: `EventBody::Agent` carries pebble's `CodingAgentEvent` envelope;
the per-variant bodies, the transcript projection, and the fabro-only
context-window, tool-summary, and skill types are gone in favor of
pebble's. The OpenAPI schemas, generated Rust and TypeScript clients,
and web readers follow.

Ask Fabro: the session runs a `CodingAgent` under a read-only permission
policy and a system prompt transform. Its conversation lives in a new
`run_session_records` table and resumes on the recorded model with the
event cursor advanced past the run log.

`fabro exec` builds the same agent over a local sandbox with pebble's
permission middleware and an interactive approval service.

The catalog fills in `metadata.agent.profile` for operator providers
that declare none, so pebble's lookup is the one resolution path.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-11 14:19:15 -06:00

3302 lines
115 KiB
Rust
Raw Blame History

use std::collections::VecDeque;
use std::future::Future;
use std::num::NonZeroU64;
use std::path::Path;
use std::pin::Pin;
use std::sync::{Arc, RwLock};
use anyhow::{Context as _, Result, anyhow, bail};
use bytes::Bytes;
use fabro_api::types;
use fabro_http::header::{ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
use fabro_http::multipart::{Form, Part};
use fabro_types::settings::run::MergeStrategy;
use fabro_types::{
ArtifactUpload, BlobHash, EventEnvelope, Model, ModelTestMode, PairId, PairMessageRecord,
PairMessageRequest, PairRecord, PairStartRequest, PairTranscriptResponse, Run, RunEvent,
RunEventDetailResponse, RunId, RunPairStatusResponse, RunProjection, RunSessionMetadata,
SessionId, StageId, WorkflowVersion, WorkflowVersionId,
};
use fabro_util::exit::{ErrorExt, ExitClass};
use futures::future::BoxFuture;
use futures::{Stream, StreamExt};
use lithos_llm::catalog::ProviderId;
use lithos_llm::types::ReasoningEffort;
use serde::{Deserialize, Serialize};
use tokio::fs::File;
use tokio::sync::Mutex;
use tokio::time;
use tokio_util::io::ReaderStream;
use crate::credential::Credential;
use crate::error::{
ApiError, ApiFailure, api_failure_for, classify_api_error, classify_http_response,
convert_type, is_not_found_error, raw_response_failure_error,
};
use crate::session::OAuthSession;
use crate::target::ServerTarget;
use crate::{AuthEntry, OAuthEntry, StoredSubject, sse};
const DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT: std::time::Duration =
std::time::Duration::from_secs(30);
const DEFAULT_HEALTH_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1);
/// Matches the `Retry-After` the server sends on the 202
/// (`PULL_REQUEST_CREATION_RETRY_AFTER` in `fabro-server`).
const PULL_REQUEST_CREATION_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
/// Overall polling deadline. The server abandons a creation attempt after 10
/// minutes, but a creation can also sit pending behind the server's worker
/// pool (or a dead server), so the client needs its own bound.
const PULL_REQUEST_CREATION_POLL_DEADLINE: std::time::Duration = std::time::Duration::from_mins(15);
type TransportFuture = BoxFuture<'static, Result<(fabro_http::HttpClient, String)>>;
pub struct RunEventStream {
stream: progenitor_client::ByteStream,
pending_bytes: Vec<u8>,
buffered_events: VecDeque<EventEnvelope>,
}
type HttpByteStream = Pin<Box<dyn Stream<Item = Result<Bytes>> + Send>>;
pub struct SessionEventStream {
stream: HttpByteStream,
pending_bytes: Vec<u8>,
buffered_events: VecDeque<EventEnvelope>,
}
pub struct RewindRunResult {
pub status: u16,
pub response: types::RewindResponse,
}
#[derive(Default)]
struct ListStoreRunsOptions {
parent_id: Option<RunId>,
}
#[derive(Clone)]
struct ClientState {
client: fabro_api::ApiClient,
http_client: fabro_http::HttpClient,
bearer_token: Option<String>,
base_url: String,
}
#[derive(Clone)]
pub struct Client {
state: Arc<RwLock<ClientState>>,
oauth_session: Option<OAuthSession>,
/// Serializes rotation between this client's own tasks. The cross-process
/// half lives in `AuthStore::acquire_refresh_lock`.
local_refresh_lock: Arc<Mutex<()>>,
transport_connector: Option<TransportConnector>,
request_timeout: Option<std::time::Duration>,
}
#[derive(Clone)]
pub struct TransportConnector {
connect: Arc<dyn Fn(Option<String>) -> TransportFuture + Send + Sync>,
}
#[derive(Default)]
pub struct ClientBuilder {
target: Option<ServerTarget>,
credential: Option<Credential>,
oauth_session: Option<OAuthSession>,
transport: Option<(String, fabro_http::HttpClient)>,
transport_connector: Option<TransportConnector>,
request_timeout: Option<std::time::Duration>,
}
#[derive(Debug, Deserialize)]
struct CliTokenResponse {
access_token: String,
access_token_expires_at: chrono::DateTime<chrono::Utc>,
refresh_token: String,
refresh_token_expires_at: chrono::DateTime<chrono::Utc>,
subject: CliTokenSubject,
}
#[derive(Debug, Deserialize)]
struct CliTokenSubject {
idp_issuer: String,
idp_subject: String,
login: String,
name: String,
email: String,
}
#[derive(Debug, Deserialize)]
struct OAuthErrorBody {
error: String,
#[serde(default)]
error_description: Option<String>,
}
#[derive(Debug, Serialize)]
struct ArtifactBatchUploadManifest {
entries: Vec<ArtifactBatchUploadEntry>,
}
#[derive(Debug, Serialize)]
struct ArtifactBatchUploadEntry {
part: String,
path: String,
#[serde(skip_serializing_if = "Option::is_none")]
sha256: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
expected_bytes: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
content_type: Option<String>,
}
impl RunEventStream {
#[must_use]
pub fn new(stream: progenitor_client::ByteStream) -> Self {
Self {
stream,
pending_bytes: Vec::new(),
buffered_events: VecDeque::new(),
}
}
pub async fn next_event(&mut self) -> Result<Option<EventEnvelope>> {
loop {
if let Some(event) = self.buffered_events.pop_front() {
return Ok(Some(event));
}
if let Some(chunk) = self.stream.next().await {
let chunk = chunk.map_err(anyhow::Error::new)?;
self.pending_bytes.extend_from_slice(&chunk);
self.buffer_sse_events(false)?;
} else {
self.buffer_sse_events(true)?;
return Ok(self.buffered_events.pop_front());
}
}
}
fn buffer_sse_events(&mut self, finalize: bool) -> Result<()> {
for payload in sse::drain_sse_payloads(&mut self.pending_bytes, finalize) {
self.buffered_events
.push_back(serde_json::from_str(&payload)?);
}
Ok(())
}
}
impl SessionEventStream {
#[must_use]
pub fn new(stream: HttpByteStream) -> Self {
Self {
stream,
pending_bytes: Vec::new(),
buffered_events: VecDeque::new(),
}
}
pub async fn next_event(&mut self) -> Result<Option<EventEnvelope>> {
loop {
if let Some(event) = self.buffered_events.pop_front() {
return Ok(Some(event));
}
if let Some(chunk) = self.stream.next().await {
let chunk = chunk?;
self.pending_bytes.extend_from_slice(&chunk);
self.buffer_sse_events(false)?;
} else {
self.buffer_sse_events(true)?;
return Ok(self.buffered_events.pop_front());
}
}
}
fn buffer_sse_events(&mut self, finalize: bool) -> Result<()> {
for payload in sse::drain_sse_payloads(&mut self.pending_bytes, finalize) {
self.buffered_events
.push_back(serde_json::from_str(&payload)?);
}
Ok(())
}
}
impl TransportConnector {
pub fn new<F, Fut>(connect: F) -> Self
where
F: Fn(Option<String>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<(fabro_http::HttpClient, String)>> + Send + 'static,
{
Self {
connect: Arc::new(move |bearer_token| Box::pin(connect(bearer_token))),
}
}
pub async fn connect(
&self,
bearer_token: Option<String>,
) -> Result<(fabro_http::HttpClient, String)> {
(self.connect)(bearer_token).await
}
}
impl ClientBuilder {
#[must_use]
pub fn target(mut self, target: ServerTarget) -> Self {
self.target = Some(target);
self
}
#[must_use]
pub fn credential(mut self, credential: Credential) -> Self {
self.credential = Some(credential);
self
}
#[must_use]
pub fn oauth_session(mut self, oauth_session: OAuthSession) -> Self {
self.oauth_session = Some(oauth_session);
self
}
#[must_use]
pub fn transport(
mut self,
base_url: impl Into<String>,
http_client: fabro_http::HttpClient,
) -> Self {
self.transport = Some((base_url.into(), http_client));
self
}
#[must_use]
pub fn transport_connector(mut self, transport_connector: TransportConnector) -> Self {
self.transport_connector = Some(transport_connector);
self
}
#[must_use]
pub fn request_timeout(mut self, timeout: std::time::Duration) -> Self {
self.request_timeout = Some(timeout);
self
}
pub async fn connect(self) -> Result<Client> {
let bearer_token = self
.credential
.as_ref()
.map(Credential::bearer_token)
.map(ToOwned::to_owned);
let target = self.target.clone().or_else(|| {
self.oauth_session
.as_ref()
.map(|session| session.target.clone())
});
let uses_default_target_transport =
self.transport.is_none() && self.transport_connector.is_none() && target.is_some();
let request_timeout = self.request_timeout.or_else(|| {
uses_default_target_transport.then_some(DEFAULT_CONTROL_PLANE_REQUEST_TIMEOUT)
});
let transport_connector = self
.transport_connector
.or_else(|| target.map(default_transport_connector));
let state = if let Some((base_url, http_client)) = self.transport {
client_state(base_url, http_client, bearer_token.clone())
} else {
let Some(transport_connector) = transport_connector.clone() else {
bail!("client builder requires a target, transport, or transport connector");
};
let (http_client, base_url) = transport_connector.connect(bearer_token.clone()).await?;
client_state(base_url, http_client, bearer_token.clone())
};
Ok(Client {
state: Arc::new(RwLock::new(state)),
oauth_session: self.oauth_session,
local_refresh_lock: Arc::new(Mutex::new(())),
transport_connector,
request_timeout,
})
}
}
impl Client {
#[must_use]
pub fn builder() -> ClientBuilder {
ClientBuilder::default()
}
#[must_use]
pub fn from_http_client(
base_url: impl Into<String>,
http_client: fabro_http::HttpClient,
) -> Self {
Self {
state: Arc::new(RwLock::new(client_state(
base_url.into(),
http_client,
None,
))),
oauth_session: None,
local_refresh_lock: Arc::new(Mutex::new(())),
transport_connector: None,
request_timeout: None,
}
}
pub fn new_no_proxy(base_url: &str) -> Result<Self> {
let http_client = fabro_http::HttpClientBuilder::new().no_proxy().build()?;
Ok(Self::from_http_client(base_url.to_string(), http_client))
}
#[must_use]
pub fn clone_for_reuse(&self) -> Self {
self.clone()
}
#[must_use]
pub fn api_client(&self) -> fabro_api::ApiClient {
self.current_state().client
}
#[must_use]
pub fn http_client(&self) -> fabro_http::HttpClient {
self.current_state().http_client
}
#[must_use]
pub fn base_url(&self) -> String {
self.current_state().base_url
}
fn current_state(&self) -> ClientState {
self.state
.read()
.expect("client state lock should not be poisoned")
.clone()
}
fn replace_state(&self, state: ClientState) {
*self
.state
.write()
.expect("client state lock should not be poisoned") = state;
}
async fn with_request_timeout<T>(&self, future: impl Future<Output = T>) -> Result<T> {
let Some(timeout) = self.request_timeout else {
return Ok(future.await);
};
match time::timeout(timeout, future).await {
Ok(value) => Ok(value),
Err(_) => bail!("server request timed out after {timeout:?}"),
}
}
async fn send_api<T, E, F, Fut>(
&self,
request: F,
) -> Result<progenitor_client::ResponseValue<T>>
where
F: FnOnce(fabro_api::ApiClient) -> Fut + Clone,
Fut: Future<
Output = std::result::Result<
progenitor_client::ResponseValue<T>,
progenitor_client::Error<E>,
>,
>,
E: serde::Serialize + std::fmt::Debug + Send + Sync + 'static,
{
let state = self.current_state();
match self
.with_request_timeout(Box::pin(request.clone()(state.client.clone())))
.await?
{
Ok(response) => Ok(response),
Err(err) => {
let mapped = classify_api_error(err).await;
if self.should_refresh(mapped.failure.as_ref()) {
if let Some(failed_token) = state.bearer_token.as_deref() {
self.refresh_access_token(failed_token).await?;
let state = self.current_state();
let retry_response = self
.with_request_timeout(Box::pin(request(state.client.clone())))
.await?;
return match retry_response {
Ok(response) => Ok(response),
Err(err) => Err(classify_api_error(err).await.error),
};
}
}
Err(mapped.error)
}
}
}
fn should_refresh(&self, failure: Option<&ApiFailure>) -> bool {
self.oauth_session.is_some()
&& failure.is_some_and(|failure| {
failure.status == fabro_http::StatusCode::UNAUTHORIZED
&& failure.code.as_deref() == Some("access_token_expired")
})
}
async fn refresh_access_token(&self, failed_access_token: &str) -> Result<()> {
fn session_expired() -> anyhow::Error {
anyhow!("CLI session has expired. Run `fabro auth login` again.")
.classify(ExitClass::AuthRequired)
}
let Some(oauth_session) = &self.oauth_session else {
return Err(session_expired());
};
let _guard = self.local_refresh_lock.lock().await;
let current_state = self.current_state();
if current_state.bearer_token.as_deref() != Some(failed_access_token) {
return Ok(());
}
// Refresh tokens are single-use, so rotation has to be serialized
// across processes too, not just across this client's tasks. The
// AuthStore calls below take the shorter auth-file lock inside this
// guard; nothing takes the two in the other order.
let _refresh_guard = oauth_session.auth_store.acquire_refresh_lock().await?;
let Some(entry) = oauth_session.auth_store.get(&oauth_session.target)? else {
self.rebuild_with_fallback(oauth_session).await?;
return Err(session_expired());
};
let oauth_entry = match entry {
AuthEntry::DevToken(entry) => {
self.rebuild_client(Some(entry.token)).await?;
return Ok(());
}
AuthEntry::OAuth(entry) => entry,
};
// Adopt a token another process already rotated, but only while it is
// still usable. The caller retries once and does not refresh again, so
// installing an expired token here would surface a 401. An expired one
// falls through and rotates with the refresh token just read.
if oauth_entry.access_token != failed_access_token
&& oauth_entry.access_token_expires_at > chrono::Utc::now()
{
self.rebuild_client(Some(oauth_entry.access_token)).await?;
return Ok(());
}
if oauth_entry.refresh_token_expires_at <= chrono::Utc::now() {
oauth_session.auth_store.remove(&oauth_session.target)?;
self.rebuild_with_fallback(oauth_session).await?;
return Err(session_expired());
}
let (http_client, base_url) = oauth_session.target.build_public_http_client()?;
let response = http_client
.post(format!("{base_url}/auth/cli/refresh"))
.header(
AUTHORIZATION,
format!("Bearer {}", oauth_entry.refresh_token),
)
.send()
.await?;
if response.status().is_success() {
let tokens = response
.json::<CliTokenResponse>()
.await
.context("failed to parse CLI auth refresh response")?;
let entry = OAuthEntry {
access_token: tokens.access_token.clone(),
access_token_expires_at: tokens.access_token_expires_at,
refresh_token: tokens.refresh_token.clone(),
refresh_token_expires_at: tokens.refresh_token_expires_at,
subject: StoredSubject {
idp_issuer: tokens.subject.idp_issuer,
idp_subject: tokens.subject.idp_subject,
login: tokens.subject.login,
name: tokens.subject.name,
email: tokens.subject.email,
},
logged_in_at: oauth_entry.logged_in_at,
};
oauth_session
.auth_store
.put(&oauth_session.target, AuthEntry::OAuth(entry.clone()))
.context("failed to persist refreshed CLI auth tokens")?;
self.rebuild_client(Some(entry.access_token)).await?;
return Ok(());
}
let status = response.status();
let body = response.text().await.unwrap_or_default();
let parsed_error = serde_json::from_str::<OAuthErrorBody>(&body).ok();
let auth_recoverable = parsed_error.as_ref().is_some_and(|error| {
matches!(
error.error.as_str(),
"refresh_token_expired" | "refresh_token_revoked"
)
});
if auth_recoverable {
oauth_session.auth_store.remove(&oauth_session.target)?;
self.rebuild_with_fallback(oauth_session).await?;
}
let err = if let Some(parsed_error) = parsed_error {
let message = parsed_error
.error_description
.filter(|value| !value.is_empty())
.unwrap_or_else(|| format!("request failed with status {status}"));
anyhow!("{message}")
} else if body.is_empty() {
anyhow!("request failed with status {status}")
} else {
anyhow!("request failed with status {status}: {body}")
};
Err(if auth_recoverable {
err.classify(ExitClass::AuthRequired)
} else {
err
})
}
async fn rebuild_with_fallback(&self, oauth_session: &OAuthSession) -> Result<()> {
let credential = oauth_session.resolve_fallback();
self.rebuild_client(
credential
.as_ref()
.map(Credential::bearer_token)
.map(ToOwned::to_owned),
)
.await
}
async fn rebuild_client(&self, bearer_token: Option<String>) -> Result<()> {
let Some(transport_connector) = &self.transport_connector else {
bail!("client transport cannot be rebuilt");
};
let (http_client, base_url) = transport_connector.connect(bearer_token.clone()).await?;
self.replace_state(client_state(base_url, http_client, bearer_token));
Ok(())
}
pub async fn send_http_response<T, F, Fut>(
&self,
request: F,
) -> Result<std::result::Result<fabro_http::Response, ApiError>>
where
F: FnOnce(fabro_http::HttpClient) -> Fut + Clone,
Fut: Future<Output = std::result::Result<fabro_http::Response, T>>,
T: Into<anyhow::Error>,
{
let state = self.current_state();
let response = self
.with_request_timeout(Box::pin(request.clone()(state.http_client.clone())))
.await?
.map_err(Into::into)?;
match classify_http_response(response).await? {
Ok(response) => Ok(Ok(response)),
Err(failure) => {
if self.should_refresh(Some(failure.api_failure())) {
if let Some(failed_token) = state.bearer_token.as_deref() {
self.refresh_access_token(failed_token).await?;
let state = self.current_state();
let response = self
.with_request_timeout(Box::pin(request(state.http_client.clone())))
.await?
.map_err(Into::into)?;
return classify_http_response(response).await;
}
}
Ok(Err(failure))
}
}
}
async fn send_http<T, F, Fut>(&self, request: F) -> Result<fabro_http::Response>
where
F: FnOnce(fabro_http::HttpClient) -> Fut + Clone,
Fut: Future<Output = std::result::Result<fabro_http::Response, T>>,
T: Into<anyhow::Error>,
{
match self.send_http_response(request).await? {
Ok(response) => Ok(response),
Err(failure) => Err(raw_response_failure_error(&failure)),
}
}
pub async fn retrieve_resolved_server_settings(&self) -> Result<types::ServerSettings> {
let url = format!("{}/api/v1/settings", self.base_url());
let response = self
.send_http(|http_client| async move { http_client.get(&url).send().await })
.await?;
response
.json::<types::ServerSettings>()
.await
.context("server returned invalid JSON for server settings")
}
pub async fn create_run_session(
&self,
run_id: RunId,
body: types::CreateRunSessionRequest,
) -> Result<RunSessionMetadata> {
let response = self
.send_api(|client| {
let body = body.clone();
async move {
client
.create_run_session()
.id(run_id.to_string())
.body(body)
.send()
.await
}
})
.await?;
Ok(response.into_inner())
}
#[expect(
clippy::disallowed_types,
reason = "Client builds raw server API request URLs for wire transit; logging redaction is handled at log boundaries."
)]
pub async fn submit_session_turn_stream(
&self,
session_id: SessionId,
input: impl Into<String>,
) -> Result<SessionEventStream> {
let base_url = self.base_url();
let mut url = fabro_http::Url::parse(&base_url)
.with_context(|| format!("invalid server base URL {base_url}"))?;
url.path_segments_mut()
.map_err(|()| anyhow!("server base URL cannot accept path segments"))?
.extend(["api", "v1", "sessions", &session_id.to_string(), "turns"]);
let body = types::SubmitTurnRequest {
input: input.into(),
turn_id: None,
};
let response = self
.send_http(|http_client| {
let url = url.clone();
let body = body.clone();
async move {
http_client
.post(url)
.header(ACCEPT, "text/event-stream")
.json(&body)
.send()
.await
}
})
.await?;
let stream = response
.bytes_stream()
.map(|chunk| chunk.map_err(anyhow::Error::new));
Ok(SessionEventStream::new(Box::pin(stream)))
}
pub async fn create_run_from_manifest(&self, manifest: types::RunManifest) -> Result<RunId> {
self.submit_create_run(manifest.into()).await
}
/// Retrieves one canonical server-managed environment by ID.
pub async fn retrieve_environment(&self, id: &str) -> Result<types::Environment> {
let response = self
.send_api(|client| {
let id = id.to_string();
async move { client.retrieve_environment().id(id).send().await }
})
.await?;
Ok(response.into_inner())
}
/// Lists the canonical server-managed environment catalog.
pub async fn list_environments(&self) -> Result<Vec<types::Environment>> {
let response = self
.send_api(|client| async move { client.list_environments().send().await })
.await?;
Ok(response.into_inner().data)
}
/// Registers one workflow version and verifies the server assigned the
/// content-derived id, so a mismatched response fails loudly here rather
/// than being trusted downstream.
pub async fn create_workflow_version(
&self,
version: &WorkflowVersion,
) -> Result<WorkflowVersionId> {
let expected_id = version.id()?;
let response = self
.send_api(|client| {
let version = version.clone();
async move { client.create_workflow_version().body(version).send().await }
})
.await?;
let returned_id = response.into_inner().workflow_version_id;
if returned_id != expected_id {
bail!(
"workflow version registration returned {returned_id} for expected {expected_id}"
);
}
Ok(returned_id)
}
/// Registers versions in iteration order, stopping at the first failure.
/// Callers must order dependencies before the versions that reference them.
pub async fn register_workflow_versions<'a>(
&self,
versions: impl IntoIterator<Item = &'a WorkflowVersion>,
) -> Result<()> {
for (index, version) in versions.into_iter().enumerate() {
self.create_workflow_version(version)
.await
.with_context(|| format!("failed to register workflow version at index {index}"))?;
}
Ok(())
}
pub async fn create_run_from_intent(&self, intent: types::RunIntent) -> Result<RunId> {
self.submit_create_run(intent.into()).await
}
async fn submit_create_run(&self, body: types::CreateRunRequest) -> Result<RunId> {
let response = self
.send_api(|client| async move { client.create_run().body(body.clone()).send().await })
.await?;
let status = response.into_inner();
Ok(status.id)
}
pub async fn list_secrets(&self) -> Result<Vec<types::SecretMetadata>> {
let response = self
.send_api(|client| async move { client.list_secrets().send().await })
.await?;
Ok(response.into_inner().data)
}
pub async fn create_secret(
&self,
body: types::CreateSecretRequest,
) -> Result<types::SecretMetadata> {
let response = self
.send_api(
|client| async move { client.create_secret().body(body.clone()).send().await },
)
.await?;
Ok(response.into_inner())
}
pub async fn delete_secret_by_name(&self, name: &str) -> Result<()> {
self.send_api(|client| async move {
client
.delete_secret_by_name()
.body(types::DeleteSecretRequest {
name: name.to_string(),
})
.send()
.await
})
.await?;
Ok(())
}
pub async fn list_variables(&self) -> Result<Vec<types::Variable>> {
let response = self
.send_api(|client| async move { client.list_variables().send().await })
.await?;
Ok(response.into_inner().data)
}
pub async fn get_variable(&self, name: &str) -> Result<types::Variable> {
let response = self
.send_api(
|client| async move { client.get_variable().name(name.to_string()).send().await },
)
.await?;
Ok(response.into_inner())
}
pub async fn create_variable(
&self,
body: types::CreateVariableRequest,
) -> Result<types::Variable> {
let response = self
.send_api(
|client| async move { client.create_variable().body(body.clone()).send().await },
)
.await?;
Ok(response.into_inner())
}
pub async fn update_variable(
&self,
name: &str,
body: types::UpdateVariableRequest,
) -> Result<types::Variable> {
let response = self
.send_api(|client| async move {
client
.update_variable()
.name(name.to_string())
.body(body.clone())
.send()
.await
})
.await?;
Ok(response.into_inner())
}
pub async fn delete_variable(&self, name: &str) -> Result<()> {
self.send_api(|client| async move {
client.delete_variable().name(name.to_string()).send().await
})
.await?;
Ok(())
}
pub async fn list_models(
&self,
provider: Option<&str>,
query: Option<&str>,
) -> Result<Vec<Model>> {
let provider = provider.map(ProviderId::new);
let mut offset = 0u64;
let mut models = Vec::new();
loop {
let provider = provider.clone();
let response = self
.send_api(|client| async move {
let mut request = client.list_models().page_limit(100u64).page_offset(offset);
if let Some(provider) = provider {
request = request.provider(provider);
}
if let Some(query) = query {
request = request.query(query.to_string());
}
request.send().await
})
.await?;
let parsed = response.into_inner();
let count = parsed.data.len() as u64;
models.extend(convert_type::<_, Vec<Model>>(parsed.data)?);
if !parsed.meta.has_more {
break;
}
offset += count;
}
Ok(models)
}
pub async fn list_providers(&self) -> Result<Vec<types::Provider>> {
let response = self
.send_api(|client| async move { client.list_providers().send().await })
.await?;
convert_type::<_, Vec<types::Provider>>(response.into_inner().data)
}
pub async fn test_provider_credentials(
&self,
provider: &ProviderId,
api_key: &str,
) -> Result<()> {
self.send_api(|client| async move {
client
.test_provider_credentials()
.provider(provider.to_string())
.body(types::ProviderCredentialTestRequest {
api_key: api_key.to_string(),
})
.send()
.await
})
.await?;
Ok(())
}
pub async fn test_model(
&self,
id: &str,
provider: Option<&ProviderId>,
mode: Option<ModelTestMode>,
reasoning_effort: Option<ReasoningEffort>,
) -> Result<types::ModelTestResult> {
let response = self
.send_api(|client| async move {
let mut request = client.test_model().id(id.to_string());
if let Some(provider) = provider {
request = request.provider(provider.clone());
}
if let Some(mode) = mode {
request = request.mode(mode);
}
if let Some(reasoning_effort) = reasoning_effort {
request = request.reasoning_effort(reasoning_effort);
}
request.send().await
})
.await?;
Ok(response.into_inner())
}
pub async fn attach_events(&self, run_ids: &[String]) -> Result<progenitor_client::ByteStream> {
let response = self
.send_api(|client| async move {
let mut request = client.attach_events();
if !run_ids.is_empty() {
request = request.run_id(run_ids.join(","));
}
request.send().await
})
.await?;
Ok(response.into_inner())
}
pub async fn get_system_info(&self) -> Result<types::SystemInfoResponse> {
let response = self
.send_api(|client| async move { client.get_system_info().send().await })
.await?;
Ok(response.into_inner())
}
pub async fn get_system_disk_usage(&self, verbose: bool) -> Result<types::DiskUsageResponse> {
let response = self
.send_api(|client| async move {
client.get_system_disk_usage().verbose(verbose).send().await
})
.await?;
Ok(response.into_inner())
}
pub async fn get_system_repair_runs(&self) -> Result<types::SystemRepairRunsResponse> {
let response = self
.send_api(|client| async move { client.get_system_repair_runs().send().await })
.await?;
Ok(response.into_inner())
}
pub async fn prune_runs(
&self,
body: types::PruneRunsRequest,
) -> Result<types::PruneRunsResponse> {
let response = self
.send_api(|client| async move { client.prune_runs().body(body.clone()).send().await })
.await?;
Ok(response.into_inner())
}
pub async fn get_health(&self) -> Result<()> {
match time::timeout(
DEFAULT_HEALTH_REQUEST_TIMEOUT,
self.send_api(|client| async move { client.get_health().send().await }),
)
.await
{
Ok(result) => {
result?;
}
Err(_) => bail!("server health check timed out"),
}
Ok(())
}
pub async fn run_diagnostics(&self) -> Result<types::DiagnosticsReport> {
let response = self
.send_api(|client| async move { client.run_diagnostics().send().await })
.await?;
Ok(response.into_inner())
}
pub async fn get_github_repo(
&self,
owner: &str,
name: &str,
) -> Result<types::RepoCheckResponse> {
let response = self
.send_api(|client| async move {
client
.get_github_repo()
.owner(owner.to_string())
.name(name.to_string())
.send()
.await
})
.await?;
Ok(response.into_inner())
}
pub async fn run_preflight(
&self,
manifest: types::RunManifest,
) -> Result<types::PreflightResponse> {
self.send_api(
|client| async move { client.run_preflight().body(manifest.clone()).send().await },
)
.await
.map(progenitor_client::ResponseValue::into_inner)
}
pub async fn validate_run_manifest(
&self,
manifest: types::RunManifest,
) -> Result<types::ValidateResponse> {
self.send_api(|client| async move {
client
.validate_run_manifest()
.body(manifest.clone())
.send()
.await
})
.await
.map(progenitor_client::ResponseValue::into_inner)
}
pub async fn render_workflow_graph(
&self,
request: types::RenderWorkflowGraphRequest,
) -> Result<Vec<u8>> {
let response = self
.send_api(|client| async move {
client
.render_workflow_graph()
.body(request.clone())
.send()
.await
})
.await?;
let mut stream = response.into_inner();
let mut bytes = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(anyhow::Error::new)?;
bytes.extend_from_slice(&chunk);
}
Ok(bytes)
}
pub async fn start_run(&self, run_id: &RunId, resume: bool) -> Result<Run> {
let response = self
.send_api(|client| async move {
client
.start_run()
.id(run_id.to_string())
.body(types::StartRunRequest { resume })
.send()
.await
})
.await?;
convert_type(response.into_inner())
}
pub async fn cancel_run(&self, run_id: &RunId) -> Result<Run> {
let response = self
.send_api(
|client| async move { client.cancel_run().id(run_id.to_string()).send().await },
)
.await?;
convert_type(response.into_inner())
}
pub async fn approve_run(&self, run_id: &RunId) -> Result<Run> {
let response = self
.send_api(
|client| async move { client.approve_run().id(run_id.to_string()).send().await },
)
.await?;
convert_type(response.into_inner())
}
pub async fn deny_run(&self, run_id: &RunId, reason: Option<String>) -> Result<Run> {
let body = types::DenyRunRequest { reason };
let response = self
.send_api(|client| {
let body = body.clone();
async move {
client
.deny_run()
.id(run_id.to_string())
.body(body)
.send()
.await
}
})
.await?;
convert_type(response.into_inner())
}
pub async fn interrupt_run(&self, run_id: &RunId) -> Result<()> {
self.send_api(|client| async move {
client.interrupt_run().id(run_id.to_string()).send().await
})
.await?;
Ok(())
}
pub async fn steer_run(&self, run_id: &RunId, text: String, interrupt: bool) -> Result<()> {
let body: types::SteerRunRequest = types::SteerRunRequest::builder()
.text(text)
.interrupt(interrupt)
.try_into()
.map_err(|e| anyhow!("failed to build SteerRunRequest: {e}"))?;
self.send_api(|client| {
let body = body.clone();
async move {
client
.steer_run()
.id(run_id.to_string())
.body(body)
.send()
.await
}
})
.await?;
Ok(())
}
pub async fn get_run_pair_status(&self, run_id: &RunId) -> Result<RunPairStatusResponse> {
let response = self
.send_api(|client| async move {
client
.get_run_pair_status()
.id(run_id.to_string())
.send()
.await
})
.await?;
convert_type(response.into_inner())
}
pub async fn start_run_pair(&self, run_id: &RunId, stage_id: StageId) -> Result<PairRecord> {
let body = PairStartRequest { stage_id };
let response = self
.send_api(|client| {
let body = body.clone();
async move {
client
.start_run_pair()
.id(run_id.to_string())
.body(body)
.send()
.await
}
})
.await?;
convert_type(response.into_inner())
}
pub async fn get_run_pair(&self, run_id: &RunId, pair_id: &PairId) -> Result<PairRecord> {
let response = self
.send_api(|client| async move {
client
.get_run_pair()
.id(run_id.to_string())
.pair_id(*pair_id)
.send()
.await
})
.await?;
convert_type(response.into_inner())
}
pub async fn end_run_pair(&self, run_id: &RunId, pair_id: &PairId) -> Result<PairRecord> {
let response = self
.send_api(|client| async move {
client
.end_run_pair()
.id(run_id.to_string())
.pair_id(*pair_id)
.send()
.await
})
.await?;
convert_type(response.into_inner())
}
pub async fn send_run_pair_message(
&self,
run_id: &RunId,
pair_id: &PairId,
request: PairMessageRequest,
) -> Result<PairMessageRecord> {
let body = request;
let response = self
.send_api(|client| {
let body = body.clone();
async move {
client
.send_run_pair_message()
.id(run_id.to_string())
.pair_id(*pair_id)
.body(body)
.send()
.await
}
})
.await?;
convert_type(response.into_inner())
}
pub async fn get_run_pair_transcript(
&self,
run_id: &RunId,
pair_id: &PairId,
since_seq: Option<u32>,
limit: Option<u32>,
) -> Result<PairTranscriptResponse> {
let response = self
.send_api(|client| async move {
let mut builder = client
.get_run_pair_transcript()
.id(run_id.to_string())
.pair_id(*pair_id);
if let Some(since_seq) = since_seq.and_then(non_zero_u64_from_u32) {
builder = builder.since_seq(since_seq);
}
if let Some(limit) = limit.and_then(non_zero_u64_from_u32) {
builder = builder.limit(limit);
}
builder.send().await
})
.await?;
convert_type(response.into_inner())
}
pub async fn get_run_event_detail(
&self,
run_id: &RunId,
seq: u32,
max_content_length: Option<u32>,
) -> Result<RunEventDetailResponse> {
let seq = non_zero_u64_from_u32(seq).context("event seq must be non-zero")?;
let max_content_length = max_content_length.and_then(non_zero_u64_from_u32);
let response = self
.send_api(|client| async move {
let mut builder = client
.get_run_event_detail()
.id(run_id.to_string())
.seq(seq);
if let Some(max_content_length) = max_content_length {
builder = builder.max_content_length(max_content_length);
}
builder.send().await
})
.await?;
convert_type(response.into_inner())
}
pub async fn archive_run(&self, run_id: &RunId) -> Result<Run> {
let response = self
.send_api(
|client| async move { client.archive_run().id(run_id.to_string()).send().await },
)
.await?;
convert_type(response.into_inner())
}
pub async fn unarchive_run(&self, run_id: &RunId) -> Result<Run> {
let response = self
.send_api(
|client| async move { client.unarchive_run().id(run_id.to_string()).send().await },
)
.await?;
convert_type(response.into_inner())
}
pub async fn rewind_run(
&self,
run_id: &RunId,
request: types::RewindRequest,
) -> Result<RewindRunResult> {
let response = self
.send_api(|client| async move {
client
.rewind_run()
.id(run_id.to_string())
.body(request)
.send()
.await
})
.await?;
let status = response.status().as_u16();
Ok(RewindRunResult {
status,
response: response.into_inner(),
})
}
pub async fn fork_run(
&self,
run_id: &RunId,
request: types::ForkRequest,
) -> Result<types::ForkResponse> {
let response = self
.send_api(|client| async move {
client
.fork_run()
.id(run_id.to_string())
.body(request)
.send()
.await
})
.await?;
Ok(response.into_inner())
}
pub async fn run_timeline(&self, run_id: &RunId) -> Result<Vec<types::TimelineEntryResponse>> {
let response = self
.send_api(|client| async move {
client
.get_run_timeline()
.id(run_id.to_string())
.send()
.await
})
.await?;
Ok(response.into_inner())
}
pub async fn list_store_runs(&self) -> Result<Vec<Run>> {
self.list_store_runs_with_options(ListStoreRunsOptions::default())
.await
}
pub async fn list_store_runs_by_parent(&self, parent_id: RunId) -> Result<Vec<Run>> {
self.list_store_runs_with_options(ListStoreRunsOptions {
parent_id: Some(parent_id),
})
.await
}
async fn list_store_runs_with_options(
&self,
options: ListStoreRunsOptions,
) -> Result<Vec<Run>> {
let mut all_runs = Vec::new();
let mut offset = 0_u64;
let limit = 100_u64;
let parent_id = options.parent_id.map(|run_id| run_id.to_string());
loop {
let response = self
.send_api(|client| {
let parent_id = parent_id.clone();
async move {
let mut request = client
.list_runs()
.page_limit(limit)
.page_offset(offset)
.include_archived(true);
if let Some(parent_id) = parent_id {
request = request.parent_id(parent_id);
}
request.send().await
}
})
.await?;
let parsed = response.into_inner();
let batch = parsed
.data
.into_iter()
.map(convert_type)
.collect::<Result<Vec<_>>>()?;
let batch_len = batch.len() as u64;
all_runs.extend(batch);
if !parsed.meta.has_more || batch_len == 0 {
break;
}
offset += batch_len;
}
Ok(all_runs)
}
pub async fn link_run_parent(&self, child_id: &RunId, parent_id: &RunId) -> Result<Run> {
let body = types::UpdateRunParentRequest {
parent_id: parent_id.to_string(),
};
let response = self
.send_api(|client| async move {
client
.link_run_parent()
.id(child_id.to_string())
.body(body.clone())
.send()
.await
})
.await?;
convert_type(response.into_inner())
}
pub async fn unlink_run_parent(&self, child_id: &RunId) -> Result<Run> {
let response = self
.send_api(|client| async move {
client
.unlink_run_parent()
.id(child_id.to_string())
.send()
.await
})
.await?;
convert_type(response.into_inner())
}
pub async fn retrieve_run(&self, run_id: &RunId) -> Result<Run> {
let response = self
.send_api(
|client| async move { client.retrieve_run().id(run_id.to_string()).send().await },
)
.await?;
convert_type(response.into_inner())
}
pub async fn resolve_run(&self, selector: &str) -> Result<Run> {
let response = self
.send_api(|client| async move {
client
.resolve_run()
.selector(selector.to_string())
.send()
.await
})
.await?;
convert_type(response.into_inner())
}
pub async fn get_run_state(&self, run_id: &RunId) -> Result<RunProjection> {
let response = self
.send_api(
|client| async move { client.get_run_state().id(run_id.to_string()).send().await },
)
.await?;
convert_type(response.into_inner())
}
pub async fn get_run_logs(&self, run_id: &RunId) -> Result<Option<Vec<u8>>> {
let response = self
.current_state()
.client
.get_run_logs()
.id(run_id.to_string())
.send()
.await;
match response {
Ok(response) => {
let mut stream = response.into_inner();
let mut bytes = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(anyhow::Error::new)?;
bytes.extend_from_slice(&chunk);
}
Ok(Some(bytes))
}
Err(err) => {
let err = classify_api_error(err).await.error;
if is_not_found_error(&err) {
Ok(None)
} else {
Err(err)
}
}
}
}
pub async fn create_run_pull_request(
&self,
run_id: &RunId,
force: bool,
model: Option<String>,
) -> Result<fabro_types::PullRequestLink> {
let mut creation = self
.request_run_pull_request_creation(run_id, force, model)
.await?;
let creation_id = creation.id;
let deadline = std::time::Instant::now() + PULL_REQUEST_CREATION_POLL_DEADLINE;
loop {
match creation.status {
fabro_types::PullRequestCreationStatus::Pending => {
if std::time::Instant::now() >= deadline {
bail!(
"Pull request creation {creation_id} is still pending after {} \
minutes. Check its status with: fabro pr create {run_id}",
PULL_REQUEST_CREATION_POLL_DEADLINE.as_secs() / 60
);
}
time::sleep(PULL_REQUEST_CREATION_POLL_INTERVAL).await;
creation = self.get_run_pull_request_creation(run_id).await?;
if creation.id != creation_id {
bail!(
"Pull request creation {creation_id} was superseded by {}",
creation.id
);
}
}
fabro_types::PullRequestCreationStatus::Succeeded => {
return creation.pull_request.ok_or_else(|| {
anyhow!(
"Pull request creation {} succeeded without a pull request record",
creation.id
)
});
}
fabro_types::PullRequestCreationStatus::Failed => {
bail!(
"Pull request creation failed: {}",
creation
.error
.as_deref()
.unwrap_or("the server did not provide an error")
);
}
}
}
}
pub async fn request_run_pull_request_creation(
&self,
run_id: &RunId,
force: bool,
model: Option<String>,
) -> Result<fabro_types::PullRequestCreation> {
let body = types::CreateRunPullRequestRequest { force, model };
let response = self
.send_api(|client| async move {
client
.create_run_pull_request()
.id(run_id.to_string())
.body(body.clone())
.send()
.await
})
.await
.map_err(add_pr_upgrade_hint)?;
Ok(response.into_inner())
}
pub async fn get_run_pull_request_creation(
&self,
run_id: &RunId,
) -> Result<fabro_types::PullRequestCreation> {
let response = self
.send_api(|client| async move {
client
.get_run_pull_request_creation()
.id(run_id.to_string())
.send()
.await
})
.await
.map_err(add_pr_upgrade_hint)?;
Ok(response.into_inner())
}
pub async fn get_run_pull_request(
&self,
run_id: &RunId,
) -> Result<fabro_types::PullRequestResponse> {
let response = self
.send_api(|client| async move {
client
.get_run_pull_request()
.id(run_id.to_string())
.send()
.await
})
.await
.map_err(add_pr_upgrade_hint)?;
Ok(response.into_inner())
}
pub async fn link_run_pull_request(
&self,
run_id: &RunId,
html_url: String,
) -> Result<fabro_types::PullRequestLink> {
let body = types::LinkRunPullRequestRequest { html_url };
let response = self
.send_api(|client| async move {
client
.link_run_pull_request()
.id(run_id.to_string())
.body(body.clone())
.send()
.await
})
.await
.map_err(add_pr_upgrade_hint)?;
convert_type(response.into_inner())
}
pub async fn unlink_run_pull_request(
&self,
run_id: &RunId,
) -> Result<fabro_types::PullRequestLink> {
let response = self
.send_api(|client| async move {
client
.unlink_run_pull_request()
.id(run_id.to_string())
.send()
.await
})
.await
.map_err(add_pr_upgrade_hint)?;
convert_type(response.into_inner())
}
pub async fn merge_run_pull_request(
&self,
run_id: &RunId,
method: MergeStrategy,
) -> Result<types::MergeRunPullRequestResponse> {
let body = types::MergeRunPullRequestRequest { method };
let response = self
.send_api(|client| async move {
client
.merge_run_pull_request()
.id(run_id.to_string())
.body(body.clone())
.send()
.await
})
.await
.map_err(add_pr_upgrade_hint)?;
convert_type(response.into_inner())
}
pub async fn close_run_pull_request(
&self,
run_id: &RunId,
) -> Result<types::CloseRunPullRequestResponse> {
let response = self
.send_api(|client| async move {
client
.close_run_pull_request()
.id(run_id.to_string())
.send()
.await
})
.await
.map_err(add_pr_upgrade_hint)?;
convert_type(response.into_inner())
}
pub async fn list_run_events(
&self,
run_id: &RunId,
since_seq: Option<u32>,
limit: Option<usize>,
) -> Result<Vec<EventEnvelope>> {
let mut next_since_seq = since_seq;
let mut all_events = Vec::new();
loop {
let page = EventPageCursor::Ascending {
since_seq: next_since_seq,
};
let (page_events, has_more) = self.fetch_run_events_page(run_id, page, limit).await?;
let next_page_since_seq = page_events.last().map(|event| event.seq.saturating_add(1));
all_events.extend(page_events);
if limit.is_some() || !has_more || next_page_since_seq.is_none() {
break;
}
next_since_seq = next_page_since_seq;
}
Ok(all_events)
}
/// Returns the newest `max_events` in ascending sequence order.
pub async fn list_run_events_tail(
&self,
run_id: &RunId,
max_events: usize,
) -> Result<Vec<EventEnvelope>> {
if max_events == 0 {
return Ok(Vec::new());
}
// Fetch two events when the caller asks for one so an older server
// that silently ignores the new order parameter can be detected.
let fetch_target = max_events.max(2);
let mut before_seq = None;
let mut descending_events: Vec<EventEnvelope> = Vec::new();
loop {
let remaining = fetch_target - descending_events.len();
let (page_events, has_more) = self
.fetch_run_events_page(
run_id,
EventPageCursor::Descending { before_seq },
Some(remaining),
)
.await?;
let keeps_descending = descending_events
.last()
.into_iter()
.chain(&page_events)
.is_sorted_by(|previous, next| previous.seq > next.seq);
if !keeps_descending {
// An older server ignored the order parameter and returned
// ascending history; fetch everything and slice the tail.
let mut events = self.list_run_events(run_id, None, None).await?;
let tail_start = events.len().saturating_sub(max_events);
return Ok(events.split_off(tail_start));
}
before_seq = page_events.last().map(|event| event.seq);
descending_events.extend(page_events);
if descending_events.len() >= fetch_target || !has_more || before_seq.is_none() {
break;
}
}
descending_events.reverse();
let tail_start = descending_events.len().saturating_sub(max_events);
Ok(descending_events.split_off(tail_start))
}
pub async fn list_run_events_until(
&self,
run_id: &RunId,
since_seq: Option<u32>,
max_events: usize,
) -> Result<Vec<EventEnvelope>> {
if max_events == 0 {
return Ok(Vec::new());
}
let mut next_since_seq = since_seq;
let mut all_events = Vec::new();
while all_events.len() < max_events {
let remaining = max_events - all_events.len();
let page = EventPageCursor::Ascending {
since_seq: next_since_seq,
};
let (page_events, has_more) = self
.fetch_run_events_page(run_id, page, Some(remaining))
.await?;
let next_page_since_seq = page_events.last().map(|event| event.seq.saturating_add(1));
all_events.extend(page_events);
if !has_more || next_page_since_seq.is_none() {
break;
}
next_since_seq = next_page_since_seq;
}
Ok(all_events)
}
async fn fetch_run_events_page(
&self,
run_id: &RunId,
cursor: EventPageCursor,
limit: Option<usize>,
) -> Result<(Vec<EventEnvelope>, bool)> {
let response = self
.send_api(|client| async move {
let mut request = client.list_run_events().id(run_id.to_string());
match cursor {
EventPageCursor::Ascending { since_seq } => {
if let Some(seq) = since_seq.and_then(non_zero_u64_from_u32) {
request = request.since_seq(seq);
}
}
EventPageCursor::Descending { before_seq } => {
request = request.order(types::ListRunEventsOrder::Desc);
if let Some(seq) = before_seq.and_then(non_zero_u64_from_u32) {
request = request.before_seq(seq);
}
}
}
let page_limit = limit.map(|limit| limit.min(1000));
if let Some(limit) = page_limit.and_then(non_zero_u64_from_usize) {
request = request.limit(limit);
}
request.send().await
})
.await?;
let parsed = response.into_inner();
let events = parsed
.data
.into_iter()
.map(convert_type::<_, EventEnvelope>)
.collect::<Result<Vec<EventEnvelope>>>()?;
Ok((events, parsed.meta.has_more))
}
pub async fn attach_run_events(
&self,
run_id: &RunId,
since_seq: Option<u32>,
) -> Result<RunEventStream> {
let response = self
.send_api(|client| async move {
let mut request = client.attach_run_events().id(run_id.to_string());
if let Some(seq) = since_seq.and_then(non_zero_u64_from_u32) {
request = request.since_seq(seq);
}
request.send().await
})
.await?;
Ok(RunEventStream::new(response.into_inner()))
}
pub async fn list_run_questions(&self, run_id: &RunId) -> Result<Vec<types::ApiQuestion>> {
let response = self
.send_api(|client| async move {
client
.list_run_questions()
.id(run_id.to_string())
.page_limit(100)
.page_offset(0)
.send()
.await
})
.await?;
Ok(response.into_inner().data)
}
pub async fn submit_run_answer(
&self,
run_id: &RunId,
qid: &str,
body: types::SubmitAnswerRequest,
) -> Result<()> {
self.send_api(|client| async move {
client
.submit_run_answer()
.id(run_id.to_string())
.qid(qid)
.body(body.clone())
.send()
.await
})
.await?;
Ok(())
}
pub async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result<u32> {
let body: types::RunEvent = convert_type(event)?;
let response = self
.send_api(|client| async move {
client
.append_run_event()
.id(run_id.to_string())
.body(body.clone())
.send()
.await
})
.await?;
u32::try_from(response.into_inner().seq).context("append_run_event returned invalid seq")
}
pub async fn write_run_blob(&self, run_id: &RunId, data: &[u8]) -> Result<BlobHash> {
let response = self
.send_api(|client| async move {
client
.write_run_blob()
.id(run_id.to_string())
.body(data.to_vec())
.send()
.await
})
.await?;
Ok(response.into_inner().hash)
}
pub async fn read_run_blob(
&self,
run_id: &RunId,
blob_hash: &BlobHash,
) -> Result<Option<Bytes>> {
let response = self
.current_state()
.client
.read_run_blob()
.id(run_id.to_string())
.blob_hash(*blob_hash)
.send()
.await;
match response {
Ok(response) => {
let mut stream = response.into_inner();
let mut bytes = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(anyhow::Error::new)?;
bytes.extend_from_slice(&chunk);
}
Ok(Some(Bytes::from(bytes)))
}
Err(err) => {
let err = classify_api_error(err).await.error;
if is_not_found_error(&err) {
Ok(None)
} else {
Err(err)
}
}
}
}
#[expect(
clippy::disallowed_types,
reason = "Client builds raw server API request URLs for wire transit; logging redaction is handled at log boundaries."
)]
pub async fn delete_store_run(&self, run_id: &RunId, force: bool) -> Result<()> {
let base_url = self.base_url();
let mut url = fabro_http::Url::parse(&base_url)
.with_context(|| format!("invalid server base URL {base_url}"))?;
url.path_segments_mut()
.map_err(|()| anyhow!("server base URL cannot accept path segments"))?
.extend(["api", "v1", "runs", &run_id.to_string()]);
if force {
url.query_pairs_mut().append_pair("force", "true");
}
self.send_http(|http_client| async move { http_client.delete(url.clone()).send().await })
.await?;
Ok(())
}
pub async fn list_run_artifacts(&self, run_id: &RunId) -> Result<Vec<types::RunArtifactEntry>> {
let response = self
.send_api(|client| async move {
client
.list_run_artifacts()
.id(run_id.to_string())
.send()
.await
})
.await?;
Ok(response.into_inner().data)
}
pub async fn download_stage_artifact(
&self,
run_id: &RunId,
stage_id: &StageId,
retry: u32,
filename: &str,
) -> Result<Vec<u8>> {
let response = self
.send_api(|client| async move {
client
.get_stage_artifact()
.id(run_id.to_string())
.stage_id(stage_id.to_string())
.retry(retry.cast_signed())
.filename(filename)
.send()
.await
})
.await?;
let mut stream = response.into_inner();
let mut bytes = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(anyhow::Error::new)?;
bytes.extend_from_slice(&chunk);
}
Ok(bytes)
}
#[expect(
clippy::disallowed_types,
reason = "Client builds raw server API request URLs for wire transit; logging redaction is handled at log boundaries."
)]
fn stage_artifacts_url(
&self,
run_id: &RunId,
stage_id: &StageId,
retry: u32,
) -> Result<fabro_http::Url> {
let base_url = self.base_url();
let mut url = fabro_http::Url::parse(&base_url)
.with_context(|| format!("invalid server base URL {base_url}"))?;
url.path_segments_mut()
.map_err(|()| anyhow!("server base URL cannot accept path segments"))?
.extend([
"api",
"v1",
"runs",
&run_id.to_string(),
"stages",
&stage_id.to_string(),
"artifacts",
]);
url.query_pairs_mut()
.append_pair("retry", &retry.to_string());
Ok(url)
}
pub async fn upload_stage_artifact_file(
&self,
run_id: &RunId,
stage_id: &StageId,
retry: u32,
filename: &str,
path: &Path,
bearer_token: &str,
) -> Result<()> {
let mut url = self.stage_artifacts_url(run_id, stage_id, retry)?;
url.query_pairs_mut().append_pair("filename", filename);
let file = File::open(path)
.await
.with_context(|| format!("failed to open artifact {}", path.display()))?;
let content_length = file
.metadata()
.await
.with_context(|| format!("failed to stat artifact {}", path.display()))?
.len();
let body = fabro_http::Body::wrap_stream(ReaderStream::new(file));
let response = self
.current_state()
.http_client
.post(url)
.bearer_auth(bearer_token)
.header(CONTENT_TYPE, "application/octet-stream")
.header(CONTENT_LENGTH, content_length.to_string())
.body(body)
.send()
.await
.with_context(|| format!("failed to upload artifact {}", path.display()))?;
classify_http_response(response)
.await?
.map(|_| ())
.map_err(|failure| raw_response_failure_error(&failure))
}
pub async fn upload_stage_artifact_batch(
&self,
run_id: &RunId,
stage_id: &StageId,
retry: u32,
artifact_capture_dir: &Path,
artifacts: &[ArtifactUpload],
bearer_token: &str,
) -> Result<()> {
let url = self.stage_artifacts_url(run_id, stage_id, retry)?;
let mut manifest_entries = Vec::with_capacity(artifacts.len());
let mut file_parts = Vec::with_capacity(artifacts.len());
for (index, artifact) in artifacts.iter().enumerate() {
let part_name = format!("file{}", index + 1);
let path = artifact_capture_dir.join(&artifact.path);
let file = File::open(&path)
.await
.with_context(|| format!("failed to open artifact {}", path.display()))?;
let content_length = file
.metadata()
.await
.with_context(|| format!("failed to stat artifact {}", path.display()))?
.len();
manifest_entries.push(ArtifactBatchUploadEntry {
part: part_name.clone(),
path: artifact.path.clone(),
sha256: Some(artifact.content_sha256.clone()),
expected_bytes: Some(artifact.bytes),
content_type: Some(artifact.mime.clone()),
});
file_parts.push((
part_name,
Part::stream_with_length(
fabro_http::Body::wrap_stream(ReaderStream::new(file)),
content_length,
)
.file_name(artifact.path.clone()),
));
}
let manifest = ArtifactBatchUploadManifest {
entries: manifest_entries,
};
let manifest_part =
Part::text(serde_json::to_string(&manifest)?).mime_str("application/json")?;
let mut form = Form::new().part("manifest", manifest_part);
for (part_name, part) in file_parts {
form = form.part(part_name, part);
}
let response = self
.current_state()
.http_client
.post(url)
.bearer_auth(bearer_token)
.multipart(form)
.send()
.await
.context("failed to upload artifact batch")?;
classify_http_response(response)
.await?
.map(|_| ())
.map_err(|failure| raw_response_failure_error(&failure))
}
pub async fn generate_preview_url(
&self,
run_id: &RunId,
port: u16,
expires_in_secs: u64,
signed: bool,
) -> Result<types::PreviewUrlResponse> {
let expires_in_secs = NonZeroU64::new(expires_in_secs)
.ok_or_else(|| anyhow!("preview expiry must be greater than zero"))?;
let response = self
.send_api(|client| async move {
client
.generate_preview_url()
.id(run_id.to_string())
.body(types::PreviewUrlRequest {
expires_in_secs,
port: i64::from(port),
signed,
})
.send()
.await
})
.await?;
Ok(response.into_inner())
}
pub async fn create_run_ssh_access(
&self,
run_id: &RunId,
ttl_minutes: f64,
) -> Result<types::SshAccessResponse> {
let response = self
.send_api(|client| async move {
client
.create_run_ssh_access()
.id(run_id.to_string())
.body(types::SshAccessRequest { ttl_minutes })
.send()
.await
})
.await?;
Ok(response.into_inner())
}
pub async fn list_sandbox_files(
&self,
run_id: &RunId,
path: &str,
depth: Option<u32>,
) -> Result<Vec<types::SandboxFileEntry>> {
let response = self
.send_api(|client| async move {
let mut request = client
.list_sandbox_files()
.id(run_id.to_string())
.path(path);
if let Some(depth) = depth.and_then(non_zero_u64_from_u32) {
request = request.depth(depth);
}
request.send().await
})
.await?;
Ok(response.into_inner().data)
}
pub async fn get_sandbox_file(&self, run_id: &RunId, path: &str) -> Result<Vec<u8>> {
let response = self
.send_api(|client| async move {
client
.get_sandbox_file()
.id(run_id.to_string())
.path(path)
.send()
.await
})
.await?;
let mut stream = response.into_inner();
let mut bytes = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(anyhow::Error::new)?;
bytes.extend_from_slice(&chunk);
}
Ok(bytes)
}
pub async fn put_sandbox_file(&self, run_id: &RunId, path: &str, bytes: Vec<u8>) -> Result<()> {
self.send_api(|client| async move {
client
.put_sandbox_file()
.id(run_id.to_string())
.path(path)
.body(bytes.clone())
.send()
.await
})
.await?;
Ok(())
}
}
fn client_state(
base_url: String,
http_client: fabro_http::HttpClient,
bearer_token: Option<String>,
) -> ClientState {
let client = fabro_api::ApiClient::new_with_client(&base_url, http_client.clone());
ClientState {
client,
http_client,
bearer_token,
base_url,
}
}
fn default_transport_connector(target: ServerTarget) -> TransportConnector {
TransportConnector::new(move |bearer_token| {
let target = target.clone();
async move { connect_target_transport(&target, bearer_token.as_deref()) }
})
}
fn connect_target_transport(
target: &ServerTarget,
bearer_token: Option<&str>,
) -> Result<(fabro_http::HttpClient, String)> {
if let Some(api_url) = target.as_http_url() {
let mut builder = fabro_http::HttpClientBuilder::new();
builder = match bearer_token {
Some(token) => apply_bearer_token_auth(builder, token)?,
None => builder,
};
let http_client = builder.build()?;
return Ok((http_client, api_url.to_string()));
}
let Some(path) = target.as_unix_socket_path() else {
bail!("server target must be an http(s) URL or absolute Unix socket path");
};
let mut builder = fabro_http::HttpClientBuilder::new()
.unix_socket(path)
.no_proxy();
builder = match bearer_token {
Some(token) => apply_bearer_token_auth(builder, token)?,
None => builder,
};
let http_client = builder.build()?;
Ok((http_client, "http://fabro".to_string()))
}
pub fn apply_bearer_token_auth(
builder: fabro_http::HttpClientBuilder,
token: &str,
) -> Result<fabro_http::HttpClientBuilder> {
let mut headers = fabro_http::HeaderMap::new();
headers.insert(
AUTHORIZATION,
fabro_http::HeaderValue::from_str(&format!("Bearer {token}"))
.context("invalid bearer token header value")?,
);
Ok(builder.default_headers(headers))
}
#[derive(Clone, Copy)]
enum EventPageCursor {
Ascending { since_seq: Option<u32> },
Descending { before_seq: Option<u32> },
}
fn non_zero_u64_from_u32(value: u32) -> Option<NonZeroU64> {
NonZeroU64::new(u64::from(value))
}
fn non_zero_u64_from_usize(value: usize) -> Option<NonZeroU64> {
u64::try_from(value).ok().and_then(NonZeroU64::new)
}
// A 404 without a structured error code means the server didn't know the
// route — PR commands moved server-side in a recent release. Point users at
// an upgrade rather than leaving them with an opaque message. A 404 with a
// code (e.g. no_stored_record) is a normal app-level response and passes
// through unchanged.
fn add_pr_upgrade_hint(err: anyhow::Error) -> anyhow::Error {
let is_missing_route = api_failure_for(&err).is_some_and(|failure| {
failure.status == fabro_http::StatusCode::NOT_FOUND && failure.code.is_none()
});
if is_missing_route {
anyhow!(
"{err}\n\n\
The fabro server may not support pull request endpoints — `fabro pr` commands \
moved server-side in a recent release. Upgrade the fabro server."
)
} else {
err
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use chrono::Duration as ChronoDuration;
use fabro_types::{SandboxProviderKind, WorkflowPath};
use fabro_util::exit;
use httpmock::Method::{GET, POST};
use httpmock::{HttpMockResponse, MockServer};
use serde_json::json;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use super::*;
use crate::error::tag_with_failure;
use crate::{AuthStore, DevTokenEntry};
fn oauth_entry(login: &str) -> OAuthEntry {
let now = chrono::Utc::now();
OAuthEntry {
access_token: format!("access-{login}"),
access_token_expires_at: now + ChronoDuration::minutes(10),
refresh_token: format!("refresh-{login}"),
refresh_token_expires_at: now + ChronoDuration::days(30),
subject: StoredSubject {
idp_issuer: "https://github.com".to_string(),
idp_subject: "12345".to_string(),
login: login.to_string(),
name: format!("Name {login}"),
email: format!("{login}@example.com"),
},
logged_in_at: now,
}
}
fn run_event_json(run_id: &RunId, seq: u32) -> serde_json::Value {
json!({
"seq": seq,
"event": "run.running",
"id": format!("evt-{seq}"),
"run_id": run_id,
"ts": "2026-07-24T12:00:00Z",
"properties": {},
})
}
fn test_workflow_version(
name: &str,
workflow_dependencies: BTreeMap<WorkflowPath, WorkflowVersionId>,
) -> WorkflowVersion {
let entrypoint = WorkflowPath::new("workflow.fabro").unwrap();
WorkflowVersion::new(
entrypoint.clone(),
BTreeMap::from([(entrypoint, format!("digraph {name} {{}}"))]),
workflow_dependencies,
)
.unwrap()
}
/// Mocks `POST /api/v1/workflow-versions` for exactly this version body.
async fn mock_create_workflow_version<'a>(
server: &'a MockServer,
version: &WorkflowVersion,
then: impl FnOnce(httpmock::Then) -> httpmock::Then,
) -> httpmock::Mock<'a> {
let body = serde_json::to_value(version).unwrap();
server
.mock_async(|when, respond| {
when.method(POST)
.path("/api/v1/workflow-versions")
.json_body(body);
then(respond);
})
.await
}
fn created_workflow_version(then: httpmock::Then, id: WorkflowVersionId) -> httpmock::Then {
then.status(201)
.header("content-type", "application/json")
.json_body(json!({ "workflow_version_id": id }))
}
fn workflow_version_response(id: WorkflowVersionId) -> HttpMockResponse {
HttpMockResponse::builder()
.status(201)
.header("content-type", "application/json")
.body(json!({ "workflow_version_id": id }).to_string())
.build()
}
fn environment_json(id: &str, provider: &str) -> serde_json::Value {
json!({
"id": id,
"revision": "0".repeat(64),
"provider": provider,
"image": { "docker": null, "dockerfile": null },
"resources": { "cpu": null, "memory": null, "disk": null },
"network": { "mode": "allow_all", "allow": [] },
"lifecycle": {
"preserve": false,
"stop_on_terminal": true,
"auto_stop": null
},
"labels": {},
"env": {}
})
}
#[tokio::test]
async fn retrieve_environment_returns_the_canonical_environment() {
let server = MockServer::start_async().await;
let mock = server
.mock_async(|when, then| {
when.method(GET).path("/api/v1/environments/local");
then.status(200)
.header("content-type", "application/json")
.json_body(environment_json("local", "local"));
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let environment = client.retrieve_environment("local").await.unwrap();
mock.assert_async().await;
assert_eq!(environment.id.as_str(), "local");
assert_eq!(environment.settings.provider, SandboxProviderKind::LOCAL);
}
#[tokio::test]
async fn list_environments_returns_the_canonical_catalog() {
let server = MockServer::start_async().await;
let mock = server
.mock_async(|when, then| {
when.method(GET).path("/api/v1/environments");
then.status(200)
.header("content-type", "application/json")
.json_body(json!({
"data": [environment_json("production", "daytona")],
"meta": { "total": 1 }
}));
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let environments = client.list_environments().await.unwrap();
mock.assert_async().await;
assert_eq!(environments.len(), 1);
assert_eq!(environments[0].id.as_str(), "production");
assert_eq!(
environments[0].settings.provider,
SandboxProviderKind::DAYTONA
);
}
#[tokio::test]
async fn retrieve_environment_preserves_api_failure_metadata() {
let server = MockServer::start_async().await;
let mock = server
.mock_async(|when, then| {
when.method(GET).path("/api/v1/environments/missing");
then.status(404)
.header("content-type", "application/json")
.json_body(json!({
"errors": [{
"status": "404",
"title": "Not Found",
"detail": "environment not found",
"code": "environment_not_found"
}]
}));
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let error = client.retrieve_environment("missing").await.unwrap_err();
mock.assert_async().await;
let failure = api_failure_for(&error).expect("API failure metadata should be preserved");
assert_eq!(failure.status, fabro_http::StatusCode::NOT_FOUND);
assert_eq!(failure.code.as_deref(), Some("environment_not_found"));
}
#[tokio::test]
async fn create_workflow_version_posts_exact_version_and_returns_server_id() {
let server = MockServer::start_async().await;
let version = test_workflow_version("ExactVersion", BTreeMap::new());
let expected_id = version.id().unwrap();
let mock = mock_create_workflow_version(&server, &version, |then| {
created_workflow_version(then, expected_id)
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let actual_id = client.create_workflow_version(&version).await.unwrap();
mock.assert_async().await;
assert_eq!(actual_id, expected_id);
}
#[tokio::test]
async fn create_workflow_version_rejects_returned_id_mismatch() {
let server = MockServer::start_async().await;
let version = test_workflow_version("Expected", BTreeMap::new());
let expected_id = version.id().unwrap();
let returned_id = test_workflow_version("Returned", BTreeMap::new())
.id()
.unwrap();
let mock = mock_create_workflow_version(&server, &version, |then| {
created_workflow_version(then, returned_id)
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let message = client
.create_workflow_version(&version)
.await
.unwrap_err()
.to_string();
mock.assert_async().await;
assert!(message.contains(&expected_id.to_string()));
assert!(message.contains(&returned_id.to_string()));
}
#[tokio::test]
async fn register_workflow_versions_preserves_dependency_first_order() {
let server = MockServer::start_async().await;
let child = test_workflow_version("Child", BTreeMap::new());
let child_id = child.id().unwrap();
let parent = test_workflow_version(
"Parent",
BTreeMap::from([(WorkflowPath::new("child").unwrap(), child_id)]),
);
let parent_id = parent.id().unwrap();
let child_response_completed = Arc::new(AtomicBool::new(false));
let child_response_completed_for_child = Arc::clone(&child_response_completed);
let child_mock = mock_create_workflow_version(&server, &child, move |then| {
then.respond_with(move |_| {
let response = workflow_version_response(child_id);
child_response_completed_for_child.store(true, Ordering::SeqCst);
response
})
})
.await;
let parent_mock = mock_create_workflow_version(&server, &parent, move |then| {
then.respond_with(move |_| {
assert!(
child_response_completed.load(Ordering::SeqCst),
"parent registration began before the child response completed"
);
workflow_version_response(parent_id)
})
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
client
.register_workflow_versions([&child, &parent])
.await
.unwrap();
child_mock.assert_async().await;
parent_mock.assert_async().await;
}
#[tokio::test]
async fn register_workflow_versions_stops_after_request_failure() {
let server = MockServer::start_async().await;
let first = test_workflow_version("First", BTreeMap::new());
let first_id = first.id().unwrap();
let failing = test_workflow_version("Failing", BTreeMap::new());
let later = test_workflow_version("NeverSent", BTreeMap::new());
let later_id = later.id().unwrap();
let first_mock = mock_create_workflow_version(&server, &first, |then| {
created_workflow_version(then, first_id)
})
.await;
let failing_mock = mock_create_workflow_version(&server, &failing, |then| {
then.status(422)
.header("content-type", "application/json")
.json_body(json!({
"errors": [{
"status": "422",
"title": "Unprocessable Entity",
"detail": "workflow dependency was not found",
"code": "workflow_version_dependency_not_found"
}]
}))
})
.await;
let later_mock = mock_create_workflow_version(&server, &later, |then| {
created_workflow_version(then, later_id)
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let error = client
.register_workflow_versions([&first, &failing, &later])
.await
.unwrap_err();
first_mock.assert_async().await;
failing_mock.assert_async().await;
later_mock.assert_calls_async(0).await;
assert!(error.to_string().contains("index 1"));
let failure = api_failure_for(&error).expect("API failure metadata should survive context");
assert_eq!(failure.status, fabro_http::StatusCode::UNPROCESSABLE_ENTITY);
assert_eq!(
failure.code.as_deref(),
Some("workflow_version_dependency_not_found")
);
}
#[tokio::test]
async fn register_workflow_versions_accepts_empty_input_without_requests() {
let server = MockServer::start_async().await;
let unexpected = server
.mock_async(|when, then| {
when.method(POST).path("/api/v1/workflow-versions");
then.status(500);
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let none: [&WorkflowVersion; 0] = [];
client.register_workflow_versions(none).await.unwrap();
unexpected.assert_calls_async(0).await;
}
#[tokio::test]
async fn create_workflow_version_accepts_repeated_canonical_content() {
let server = MockServer::start_async().await;
let version = test_workflow_version("Repeated", BTreeMap::new());
let expected_id = version.id().unwrap();
let mock = mock_create_workflow_version(&server, &version, |then| {
created_workflow_version(then, expected_id)
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let first = client.create_workflow_version(&version).await.unwrap();
let second = client.create_workflow_version(&version).await.unwrap();
mock.assert_calls_async(2).await;
assert_eq!(first, expected_id);
assert_eq!(second, expected_id);
}
#[cfg(unix)]
#[tokio::test]
async fn refresh_access_token_allows_plain_http_targets() {
let temp = tempfile::tempdir().unwrap();
let auth_store = AuthStore::new(temp.path().join("auth.json"));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let server = tokio::spawn(async move {
let (mut stream, _) = listener.accept().await.unwrap();
let mut request = vec![0_u8; 4096];
let read = stream.read(&mut request).await.unwrap();
let request = String::from_utf8_lossy(&request[..read]);
assert!(
request.starts_with("POST /auth/cli/refresh HTTP/1.1"),
"unexpected refresh request: {request}"
);
let body = serde_json::json!({
"access_token": "access-refreshed",
"access_token_expires_at": (chrono::Utc::now() + ChronoDuration::minutes(10)).to_rfc3339(),
"refresh_token": "refresh-refreshed",
"refresh_token_expires_at": (chrono::Utc::now() + ChronoDuration::days(30)).to_rfc3339(),
"subject": {
"idp_issuer": "https://github.com",
"idp_subject": "12345",
"login": "octocat",
"name": "Name octocat",
"email": "octocat@example.com"
}
})
.to_string();
let response = format!(
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
body.len(),
body
);
stream.write_all(response.as_bytes()).await.unwrap();
});
let target = ServerTarget::http_url(format!("http://localhost:{port}")).unwrap();
let entry = oauth_entry("octocat");
auth_store
.put(&target, AuthEntry::OAuth(entry.clone()))
.unwrap();
let client = Client::builder()
.target(target.clone())
.credential(Credential::OAuth(entry))
.oauth_session(OAuthSession::new(target.clone(), auth_store.clone()))
.transport("http://localhost", fabro_http::test_http_client().unwrap())
.connect()
.await
.unwrap();
client.refresh_access_token("access-octocat").await.unwrap();
let refreshed = auth_store.get(&target).unwrap().unwrap();
let AuthEntry::OAuth(refreshed) = refreshed else {
panic!("expected OAuth entry");
};
assert_eq!(refreshed.access_token, "access-refreshed");
assert_eq!(refreshed.refresh_token, "refresh-refreshed");
server.abort();
}
#[tokio::test]
async fn request_timeout_does_not_cap_stream_body_after_headers() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
let (mut stream, _) = listener.accept().await.unwrap();
let mut request = vec![0_u8; 4096];
let read = stream.read(&mut request).await.unwrap();
let request = String::from_utf8_lossy(&request[..read]);
assert!(
request.starts_with("GET /api/v1/attach HTTP/1.1"),
"unexpected attach request: {request}"
);
let body = b"data: hello\n\n";
let headers = format!(
"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",
body.len()
);
stream.write_all(headers.as_bytes()).await.unwrap();
time::sleep(Duration::from_millis(100)).await;
stream.write_all(body).await.unwrap();
});
let target = ServerTarget::http_url(format!("http://{addr}")).unwrap();
let client = Client::builder()
.target(target)
.request_timeout(Duration::from_millis(50))
.connect()
.await
.unwrap();
let mut stream = client.attach_events(&[]).await.unwrap();
let chunk = time::timeout(Duration::from_millis(500), stream.next())
.await
.unwrap()
.unwrap()
.unwrap();
assert_eq!(chunk, Bytes::from_static(b"data: hello\n\n"));
server.await.unwrap();
}
#[tokio::test]
async fn list_models_allows_custom_provider_filters() {
let server = MockServer::start_async().await;
let mock = server
.mock_async(|when, then| {
when.method(GET)
.path("/api/v1/models")
.query_param("provider", "bedrock");
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!({
"data": [],
"meta": { "has_more": false }
})
.to_string(),
);
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let models = client.list_models(Some("bedrock"), None).await.unwrap();
mock.assert_async().await;
assert!(models.is_empty());
}
#[tokio::test]
async fn list_run_events_tail_pages_backward_and_returns_ascending() {
let server = MockServer::start_async().await;
let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
let newest_page = server
.mock_async(|when, then| {
when.method(GET)
.path(format!("/api/v1/runs/{run_id}/events"))
.query_param("order", "desc")
.query_param("limit", "5");
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({
"data": [
run_event_json(&run_id, 6),
run_event_json(&run_id, 5),
run_event_json(&run_id, 4),
],
"meta": { "has_more": true },
}));
})
.await;
let older_page = server
.mock_async(|when, then| {
when.method(GET)
.path(format!("/api/v1/runs/{run_id}/events"))
.query_param("order", "desc")
.query_param("before_seq", "4")
.query_param("limit", "2");
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({
"data": [
run_event_json(&run_id, 3),
run_event_json(&run_id, 2),
],
"meta": { "has_more": true },
}));
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let events = client.list_run_events_tail(&run_id, 5).await.unwrap();
newest_page.assert_async().await;
older_page.assert_async().await;
let seqs = events
.into_iter()
.map(|event| event.seq)
.collect::<Vec<_>>();
assert_eq!(seqs, vec![2, 3, 4, 5, 6]);
}
#[tokio::test]
async fn list_run_events_tail_falls_back_when_server_ignores_descending_order() {
let server = MockServer::start_async().await;
let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
let unsupported_descending_page = server
.mock_async(|when, then| {
when.method(GET)
.path(format!("/api/v1/runs/{run_id}/events"))
.query_param("order", "desc")
.query_param("limit", "3");
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({
"data": [
run_event_json(&run_id, 1),
run_event_json(&run_id, 2),
run_event_json(&run_id, 3),
],
"meta": { "has_more": true },
}));
})
.await;
let full_history = server
.mock_async(|when, then| {
when.method(GET)
.path(format!("/api/v1/runs/{run_id}/events"))
.query_param_missing("order")
.query_param_missing("before_seq")
.query_param_missing("since_seq")
.query_param_missing("limit");
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({
"data": [
run_event_json(&run_id, 1),
run_event_json(&run_id, 2),
run_event_json(&run_id, 3),
run_event_json(&run_id, 4),
run_event_json(&run_id, 5),
],
"meta": { "has_more": false },
}));
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
let events = client.list_run_events_tail(&run_id, 3).await.unwrap();
unsupported_descending_page.assert_async().await;
full_history.assert_async().await;
let seqs = events
.into_iter()
.map(|event| event.seq)
.collect::<Vec<_>>();
assert_eq!(seqs, vec![3, 4, 5]);
}
#[tokio::test]
async fn test_provider_credentials_posts_api_key() {
let server = MockServer::start_async().await;
let mock = server
.mock_async(|when, then| {
when.method(POST)
.path("/api/v1/providers/openrouter/credentials/test")
.json_body(json!({ "api_key": "sk-test" }));
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({ "ok": true }));
})
.await;
let client = Client::new_no_proxy(&server.url("")).unwrap();
client
.test_provider_credentials(&ProviderId::new("openrouter"), "sk-test")
.await
.unwrap();
mock.assert_async().await;
}
async fn oauth_client(
server: &MockServer,
) -> (tempfile::TempDir, Client, AuthStore, ServerTarget) {
let temp = tempfile::tempdir().unwrap();
let auth_store = AuthStore::new(temp.path().join("auth.json"));
let target = ServerTarget::http_url(server.base_url()).unwrap();
let entry = oauth_entry("octocat");
auth_store
.put(&target, AuthEntry::OAuth(entry.clone()))
.unwrap();
let client = Client::builder()
.target(target.clone())
.credential(Credential::OAuth(entry))
.oauth_session(OAuthSession::new(target.clone(), auth_store.clone()))
.transport(
server.base_url(),
fabro_http::HttpClientBuilder::new()
.no_proxy()
.build()
.unwrap(),
)
.connect()
.await
.unwrap();
(temp, client, auth_store, target)
}
#[tokio::test]
async fn refresh_access_token_reinstalls_stored_dev_token_without_refresh_request() {
let server = MockServer::start();
let refresh_mock = server.mock(|when, then| {
when.method(POST).path("/auth/cli/refresh");
then.status(500);
});
let temp = tempfile::tempdir().unwrap();
let auth_store = AuthStore::new(temp.path().join("auth.json"));
let target = ServerTarget::http_url(server.base_url()).unwrap();
let old_token =
"fabro_dev_cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd";
let stored_token =
"fabro_dev_abababababababababababababababababababababababababababababababab";
auth_store
.put(
&target,
AuthEntry::DevToken(DevTokenEntry {
token: stored_token.to_string(),
logged_in_at: chrono::Utc::now(),
}),
)
.unwrap();
let seen_tokens = Arc::new(std::sync::Mutex::new(Vec::new()));
let seen_tokens_for_connector = Arc::clone(&seen_tokens);
let base_url = server.base_url();
let client = Client::builder()
.target(target.clone())
.credential(Credential::DevToken(old_token.to_string()))
.oauth_session(OAuthSession::new(target.clone(), auth_store.clone()))
.transport_connector(TransportConnector::new(move |bearer_token| {
let seen_tokens = Arc::clone(&seen_tokens_for_connector);
let base_url = base_url.clone();
async move {
seen_tokens.lock().unwrap().push(bearer_token);
Ok((fabro_http::test_http_client().unwrap(), base_url))
}
}))
.connect()
.await
.unwrap();
client.refresh_access_token(old_token).await.unwrap();
assert_eq!(refresh_mock.calls(), 0);
assert_eq!(*seen_tokens.lock().unwrap(), vec![
Some(old_token.to_string()),
Some(stored_token.to_string()),
]);
}
#[cfg(unix)]
#[tokio::test]
async fn concurrent_clients_refresh_a_rotating_token_once() {
let server = MockServer::start_async().await;
let refresh_mock = server
.mock_async(|when, then| {
when.method(POST)
.path("/auth/cli/refresh")
.header("authorization", "Bearer refresh-octocat");
then.status(200)
.delay(Duration::from_millis(100))
.header("Content-Type", "application/json")
.json_body(json!({
"access_token": "access-refreshed",
"access_token_expires_at": (chrono::Utc::now()
+ ChronoDuration::minutes(10))
.to_rfc3339(),
"refresh_token": "refresh-refreshed",
"refresh_token_expires_at": (chrono::Utc::now()
+ ChronoDuration::days(30))
.to_rfc3339(),
"subject": {
"idp_issuer": "https://github.com",
"idp_subject": "12345",
"login": "octocat",
"name": "Name octocat",
"email": "octocat@example.com"
}
}));
})
.await;
let temp = tempfile::tempdir().unwrap();
let auth_store = AuthStore::new(temp.path().join("auth.json"));
let target = ServerTarget::http_url(server.base_url()).unwrap();
let entry = oauth_entry("octocat");
auth_store
.put(&target, AuthEntry::OAuth(entry.clone()))
.unwrap();
// Two separately built clients hold separate in-process mutexes, so the
// only thing serializing them is the lock file. That works in one
// process because flock conflicts across distinct descriptors.
let no_proxy_connector = || {
let base_url = server.base_url();
TransportConnector::new(move |_bearer_token| {
let base_url = base_url.clone();
async move { Ok((fabro_http::test_http_client().unwrap(), base_url)) }
})
};
let first = Client::builder()
.target(target.clone())
.credential(Credential::OAuth(entry.clone()))
.oauth_session(OAuthSession::new(target.clone(), auth_store.clone()))
.transport_connector(no_proxy_connector())
.connect()
.await
.unwrap();
let second = Client::builder()
.target(target.clone())
.credential(Credential::OAuth(entry))
.oauth_session(OAuthSession::new(target.clone(), auth_store.clone()))
.transport_connector(no_proxy_connector())
.connect()
.await
.unwrap();
let (first_result, second_result) = tokio::join!(
first.refresh_access_token("access-octocat"),
second.refresh_access_token("access-octocat"),
);
first_result.unwrap();
second_result.unwrap();
refresh_mock.assert_calls_async(1).await;
assert_eq!(
first.current_state().bearer_token.as_deref(),
Some("access-refreshed")
);
assert_eq!(
second.current_state().bearer_token.as_deref(),
Some("access-refreshed")
);
// The rotated refresh token must be what landed in the store, or the
// next rotation would replay a spent one.
let stored = match auth_store.get(&target).unwrap().unwrap() {
AuthEntry::OAuth(stored) => stored,
AuthEntry::DevToken(_) => panic!("expected an OAuth entry"),
};
assert_eq!(stored.access_token, "access-refreshed");
assert_eq!(stored.refresh_token, "refresh-refreshed");
}
#[tokio::test]
async fn refresh_access_token_rotates_when_the_stored_token_is_also_expired() {
let server = MockServer::start_async().await;
let refresh_mock = server
.mock_async(|when, then| {
when.method(POST)
.path("/auth/cli/refresh")
.header("authorization", "Bearer refresh-octocat");
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({
"access_token": "access-refreshed",
"access_token_expires_at": (chrono::Utc::now()
+ ChronoDuration::minutes(10))
.to_rfc3339(),
"refresh_token": "refresh-refreshed",
"refresh_token_expires_at": (chrono::Utc::now()
+ ChronoDuration::days(30))
.to_rfc3339(),
"subject": {
"idp_issuer": "https://github.com",
"idp_subject": "12345",
"login": "octocat",
"name": "Name octocat",
"email": "octocat@example.com"
}
}));
})
.await;
let temp = tempfile::tempdir().unwrap();
let auth_store = AuthStore::new(temp.path().join("auth.json"));
let target = ServerTarget::http_url(server.base_url()).unwrap();
// A sibling process rotated the store a while ago, and that token has
// since expired too. Adopting it would 401 on the caller's single retry.
let mut stored = oauth_entry("octocat");
stored.access_token = "access-stale".to_string();
stored.access_token_expires_at = chrono::Utc::now() - ChronoDuration::minutes(1);
auth_store.put(&target, AuthEntry::OAuth(stored)).unwrap();
let base_url = server.base_url();
let client = Client::builder()
.target(target.clone())
.credential(Credential::OAuth(oauth_entry("octocat")))
.oauth_session(OAuthSession::new(target, auth_store))
.transport_connector(TransportConnector::new(move |_bearer_token| {
let base_url = base_url.clone();
async move { Ok((fabro_http::test_http_client().unwrap(), base_url)) }
}))
.connect()
.await
.unwrap();
client.refresh_access_token("access-octocat").await.unwrap();
refresh_mock.assert_calls_async(1).await;
assert_eq!(
client.current_state().bearer_token.as_deref(),
Some("access-refreshed")
);
}
#[tokio::test]
async fn refresh_access_token_classifies_expired_refresh_tokens() {
let server = MockServer::start();
server.mock(|when, then| {
when.method(POST)
.path("/auth/cli/refresh")
.header("authorization", "Bearer refresh-octocat");
then.status(401)
.header("Content-Type", "application/json")
.json_body(json!({
"error": "refresh_token_expired",
"error_description": "CLI session has expired. Run `fabro auth login` again."
}));
});
let (_temp, client, auth_store, target) = oauth_client(&server).await;
let err = client
.refresh_access_token("access-octocat")
.await
.unwrap_err();
assert_eq!(exit::exit_code_for(&err), 4);
assert!(auth_store.get(&target).unwrap().is_none());
}
#[tokio::test]
async fn refresh_access_token_keeps_server_errors_as_exit_1() {
let server = MockServer::start();
server.mock(|when, then| {
when.method(POST)
.path("/auth/cli/refresh")
.header("authorization", "Bearer refresh-octocat");
then.status(500)
.header("Content-Type", "application/json")
.json_body(json!({
"error": "server_error",
"error_description": "OAuth server exploded."
}));
});
let (_temp, client, auth_store, target) = oauth_client(&server).await;
let err = client
.refresh_access_token("access-octocat")
.await
.unwrap_err();
assert_eq!(exit::exit_code_for(&err), 1);
assert!(auth_store.get(&target).unwrap().is_some());
}
#[tokio::test]
async fn refresh_access_token_keeps_login_not_permitted_as_exit_1() {
let server = MockServer::start();
server.mock(|when, then| {
when.method(POST)
.path("/auth/cli/refresh")
.header("authorization", "Bearer refresh-octocat");
then.status(403)
.header("Content-Type", "application/json")
.json_body(json!({
"error": "unauthorized",
"error_description": "Login not permitted for this user."
}));
});
let (_temp, client, auth_store, target) = oauth_client(&server).await;
let err = client
.refresh_access_token("access-octocat")
.await
.unwrap_err();
assert_eq!(exit::exit_code_for(&err), 1);
assert!(auth_store.get(&target).unwrap().is_some());
}
#[test]
fn add_pr_upgrade_hint_appends_on_unstructured_404() {
let err = tag_with_failure(
anyhow!("request failed with status 404 Not Found"),
ApiFailure {
status: fabro_http::StatusCode::NOT_FOUND,
code: None,
},
);
let wrapped = super::add_pr_upgrade_hint(err);
let message = wrapped.to_string();
assert!(
message.contains("Upgrade the fabro server"),
"expected upgrade hint, got: {message}"
);
assert!(message.contains("status 404"), "original preserved");
}
#[test]
fn add_pr_upgrade_hint_does_not_touch_structured_404() {
let err = tag_with_failure(
anyhow!("No pull request found in store. Create one first with: fabro pr create abc"),
ApiFailure {
status: fabro_http::StatusCode::NOT_FOUND,
code: Some("no_stored_record".to_string()),
},
);
let wrapped = super::add_pr_upgrade_hint(err);
let message = wrapped.to_string();
assert!(
!message.contains("Upgrade the fabro server"),
"hint should only fire on unstructured 404s, got: {message}"
);
}
}
Reference in a new issue View git blame Copy permalink