refactor(server): split HTTP handlers into modules

This commit is contained in:
Bryan Helmkamp 2026-05-02 13:32:49 -04:00
parent 0723ca068c
commit 10555a292d
No known key found for this signature in database
14 changed files with 5003 additions and 4798 deletions

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,679 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, ArtifactEntry, ArtifactKey, ArtifactListResponse, AsyncWriteExt, Body,
Bytes, DefaultBodyLimit, Digest, HashMap, HashSet, HeaderMap, IntoResponse, Json, NodeArtifact,
Path, Query, RequireRunBlob, RequireRunScoped, RequireStageArtifact, RequiredUser, Response,
Router, RunArtifactEntry, RunArtifactListResponse, RunId, Sha256, StageArtifactEntry, StageId,
State, StatusCode, StreamExt, WriteBlobResponse, axum_extract, bad_request_response, get,
header, octet_stream_response, parse_run_id_path, parse_stage_id_path,
payload_too_large_response, post, reject_if_archived, required_query_param,
validate_relative_artifact_path,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs/{id}/checkpoint", get(get_checkpoint))
.route("/runs/{id}/blobs", post(write_run_blob))
.route("/runs/{id}/blobs/{blobId}", get(read_run_blob))
.route("/runs/{id}/artifacts", get(list_run_artifacts))
.route(
"/runs/{id}/stages/{stageId}/artifacts",
get(list_stage_artifacts)
.post(put_stage_artifact)
.layer(DefaultBodyLimit::disable()),
)
.route(
"/runs/{id}/stages/{stageId}/artifacts/download",
get(get_stage_artifact),
)
}
#[derive(serde::Deserialize)]
struct ArtifactFilenameParams {
#[serde(default)]
filename: Option<String>,
#[serde(default)]
retry: Option<u32>,
}
const MAX_SINGLE_ARTIFACT_BYTES: u64 = 10 * 1024 * 1024;
const MAX_MULTIPART_ARTIFACTS: usize = 100;
const MAX_MULTIPART_REQUEST_BYTES: u64 = 50 * 1024 * 1024;
const MAX_MULTIPART_MANIFEST_BYTES: usize = 256 * 1024;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct ArtifactBatchUploadManifest {
entries: Vec<ArtifactBatchUploadEntry>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct ArtifactBatchUploadEntry {
part: String,
path: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
sha256: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
expected_bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
content_type: Option<String>,
}
async fn get_checkpoint(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let live_checkpoint = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => managed_run.checkpoint.clone(),
None => return ApiError::not_found("Run not found.").into_response(),
}
};
if let Some(cp) = live_checkpoint {
return (StatusCode::OK, Json(cp)).into_response();
}
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => match run_state.checkpoint {
Some(cp) => (StatusCode::OK, Json(cp)).into_response(),
None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
},
Err(err) => {
tracing::warn!(run_id = %id, error = %err, "Failed to load checkpoint state from store");
(StatusCode::OK, Json(serde_json::json!(null))).into_response()
}
},
Err(err) => {
tracing::warn!(run_id = %id, error = %err, "Failed to open run store reader");
ApiError::not_found("Run not found.").into_response()
}
}
}
async fn write_run_blob(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
body: Bytes,
) -> Response {
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
match state.store.open_run(&id).await {
Ok(run_store) => match run_store.write_blob(&body).await {
Ok(blob_id) => Json(WriteBlobResponse {
id: blob_id.to_string(),
})
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
}
async fn read_run_blob(
RequireRunBlob(id, blob_id): RequireRunBlob,
State(state): State<Arc<AppState>>,
) -> Response {
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.read_blob(&blob_id).await {
Ok(Some(bytes)) => octet_stream_response(bytes),
Ok(None) => ApiError::not_found("Blob not found.").into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
}
async fn load_run_spec(state: &AppState, run_id: &RunId) -> Result<fabro_types::RunSpec, Response> {
let run_store = state
.store
.open_run_reader(run_id)
.await
.map_err(|_| ApiError::not_found("Run not found.").into_response())?;
let run_state = run_store.state().await.map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
})?;
run_state.spec.ok_or_else(|| {
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"run spec missing from store",
)
.into_response()
})
}
async fn list_run_artifacts(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Err(response) = load_run_spec(state.as_ref(), &id).await {
return response;
}
match state.artifact_store.list_for_run(&id).await {
Ok(entries) => Json(RunArtifactListResponse {
data: entries.into_iter().map(run_artifact_entry_from).collect(),
})
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
fn run_artifact_entry_from(entry: NodeArtifact) -> RunArtifactEntry {
RunArtifactEntry {
stage_id: entry.node.to_string(),
node_slug: entry.node.node_id().to_string(),
retry: entry.retry.cast_signed(),
relative_path: entry.filename,
size: entry.size.cast_signed(),
}
}
fn artifact_entry_from(entry: StageArtifactEntry) -> ArtifactEntry {
ArtifactEntry {
filename: entry.filename,
retry: entry.retry.cast_signed(),
size: entry.size.cast_signed(),
}
}
async fn list_stage_artifacts(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path((id, stage_id)): Path<(String, String)>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let stage_id = match parse_stage_id_path(&stage_id) {
Ok(stage_id) => stage_id,
Err(response) => return response,
};
if let Err(response) = load_run_spec(state.as_ref(), &id).await {
return response;
}
match state.artifact_store.list_for_node(&id, &stage_id).await {
Ok(entries) => Json(ArtifactListResponse {
data: entries.into_iter().map(artifact_entry_from).collect(),
})
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
enum ArtifactUploadContentType {
OctetStream,
Multipart { boundary: String },
}
struct ValidatedArtifactBatchEntry {
path: String,
sha256: Option<String>,
expected_bytes: Option<u64>,
}
#[allow(
clippy::result_large_err,
reason = "Upload content-type parsing returns HTTP client errors directly."
)]
fn artifact_upload_content_type(
headers: &HeaderMap,
) -> Result<ArtifactUploadContentType, Response> {
let value = headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.ok_or_else(|| {
ApiError::new(
StatusCode::UNSUPPORTED_MEDIA_TYPE,
"artifact uploads require a supported Content-Type",
)
.into_response()
})?;
let mime = value.split(';').next().unwrap_or(value).trim();
match mime {
"application/octet-stream" => Ok(ArtifactUploadContentType::OctetStream),
"multipart/form-data" => multer::parse_boundary(value)
.map(|boundary| ArtifactUploadContentType::Multipart { boundary })
.map_err(|err| bad_request_response(format!("invalid multipart boundary: {err}"))),
_ => Err(ApiError::new(
StatusCode::UNSUPPORTED_MEDIA_TYPE,
"artifact uploads only support application/octet-stream or multipart/form-data",
)
.into_response()),
}
}
#[allow(
clippy::result_large_err,
reason = "Content-Length parsing returns HTTP client errors directly."
)]
fn content_length_from_headers(headers: &HeaderMap) -> Result<Option<u64>, Response> {
headers
.get(header::CONTENT_LENGTH)
.map(|value| {
value
.to_str()
.map_err(|err| {
bad_request_response(format!("invalid content-length header: {err}"))
})
.and_then(|value| {
value.parse::<u64>().map_err(|err| {
bad_request_response(format!("invalid content-length header: {err}"))
})
})
})
.transpose()
}
#[allow(
clippy::result_large_err,
reason = "Multipart manifest parsing returns HTTP client errors directly."
)]
async fn read_multipart_manifest(
field: &mut multer::Field<'_>,
) -> Result<ArtifactBatchUploadManifest, Response> {
let mut manifest_bytes = Vec::new();
while let Some(chunk) = field
.chunk()
.await
.map_err(|err| bad_request_response(format!("invalid multipart body: {err}")))?
{
manifest_bytes.extend_from_slice(&chunk);
if manifest_bytes.len() > MAX_MULTIPART_MANIFEST_BYTES {
return Err(payload_too_large_response(
"multipart manifest exceeds the server limit",
));
}
}
serde_json::from_slice(&manifest_bytes)
.map_err(|err| bad_request_response(format!("invalid multipart manifest: {err}")))
}
#[allow(
clippy::result_large_err,
reason = "Artifact batch validation returns HTTP client errors directly."
)]
fn validate_artifact_batch_manifest(
manifest: ArtifactBatchUploadManifest,
) -> Result<HashMap<String, ValidatedArtifactBatchEntry>, Response> {
if manifest.entries.is_empty() {
return Err(bad_request_response(
"multipart manifest must include at least one artifact entry",
));
}
if manifest.entries.len() > MAX_MULTIPART_ARTIFACTS {
return Err(payload_too_large_response(format!(
"multipart upload exceeds the {MAX_MULTIPART_ARTIFACTS} artifact limit"
)));
}
let mut entries = HashMap::with_capacity(manifest.entries.len());
let mut seen_paths = HashSet::new();
let mut expected_total_bytes = 0_u64;
for entry in manifest.entries {
if entry.part.is_empty() {
return Err(bad_request_response(
"multipart manifest part names must not be empty",
));
}
if entry.part == "manifest" {
return Err(bad_request_response(
"multipart manifest part name 'manifest' is reserved",
));
}
let path = validate_relative_artifact_path("manifest path", &entry.path)?;
if !seen_paths.insert(path.clone()) {
return Err(bad_request_response(format!(
"duplicate artifact path in multipart manifest: {path}"
)));
}
if let Some(sha256) = entry.sha256.as_ref() {
if sha256.len() != 64 || !sha256.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(bad_request_response(format!(
"invalid sha256 for multipart part {}",
entry.part
)));
}
}
if let Some(expected_bytes) = entry.expected_bytes {
if expected_bytes > MAX_SINGLE_ARTIFACT_BYTES {
return Err(payload_too_large_response(format!(
"artifact {path} exceeds the {MAX_SINGLE_ARTIFACT_BYTES} byte limit"
)));
}
expected_total_bytes = expected_total_bytes.saturating_add(expected_bytes);
if expected_total_bytes > MAX_MULTIPART_REQUEST_BYTES {
return Err(payload_too_large_response(format!(
"multipart upload exceeds the {MAX_MULTIPART_REQUEST_BYTES} byte limit"
)));
}
}
if entries
.insert(entry.part.clone(), ValidatedArtifactBatchEntry {
path,
sha256: entry.sha256.map(|value| value.to_ascii_lowercase()),
expected_bytes: entry.expected_bytes,
})
.is_some()
{
return Err(bad_request_response(format!(
"duplicate multipart part name in manifest: {}",
entry.part
)));
}
}
Ok(entries)
}
async fn upload_stage_artifact_octet_stream(
state: &AppState,
run_id: &RunId,
stage_id: &StageId,
retry: u32,
filename: String,
body: Body,
content_length: Option<u64>,
) -> Response {
let relative_path = match validate_relative_artifact_path("filename", &filename) {
Ok(path) => path,
Err(response) => return response,
};
if content_length.is_some_and(|length| length > MAX_SINGLE_ARTIFACT_BYTES) {
return payload_too_large_response(format!(
"artifact exceeds the {MAX_SINGLE_ARTIFACT_BYTES} byte limit"
));
}
let mut writer = match state.artifact_store.writer(
run_id,
&ArtifactKey::new(stage_id.clone(), retry, relative_path),
) {
Ok(writer) => writer,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let mut bytes_written = 0_u64;
let mut data_stream = body.into_data_stream();
while let Some(chunk) = data_stream.next().await {
let chunk = match chunk
.map_err(|err| bad_request_response(format!("invalid request body: {err}")))
{
Ok(chunk) => chunk,
Err(response) => return response,
};
bytes_written =
bytes_written.saturating_add(u64::try_from(chunk.len()).unwrap_or(u64::MAX));
if bytes_written > MAX_SINGLE_ARTIFACT_BYTES {
return payload_too_large_response(format!(
"artifact exceeds the {MAX_SINGLE_ARTIFACT_BYTES} byte limit"
));
}
if let Err(err) = writer.write_all(&chunk).await {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
match writer.shutdown().await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
async fn upload_stage_artifact_multipart(
state: &AppState,
run_id: &RunId,
stage_id: &StageId,
retry: u32,
boundary: String,
body: Body,
) -> Response {
let mut multipart = multer::Multipart::new(body.into_data_stream(), boundary);
let Some(mut manifest_field) = (match multipart
.next_field()
.await
.map_err(|err| bad_request_response(format!("invalid multipart body: {err}")))
{
Ok(field) => field,
Err(response) => return response,
}) else {
return bad_request_response("multipart upload must begin with a manifest part");
};
if manifest_field.name() != Some("manifest") {
return bad_request_response("multipart upload must begin with a manifest part");
}
let manifest = match read_multipart_manifest(&mut manifest_field).await {
Ok(manifest) => manifest,
Err(response) => return response,
};
drop(manifest_field);
let mut expected_parts = match validate_artifact_batch_manifest(manifest) {
Ok(entries) => entries,
Err(response) => return response,
};
let mut total_bytes = 0_u64;
while let Some(mut field) = match multipart
.next_field()
.await
.map_err(|err| bad_request_response(format!("invalid multipart body: {err}")))
{
Ok(field) => field,
Err(response) => return response,
} {
let Some(part_name) = field.name().map(ToOwned::to_owned) else {
return bad_request_response("multipart file parts must be named");
};
let Some(entry) = expected_parts.remove(&part_name) else {
return bad_request_response(format!("unexpected multipart part: {part_name}"));
};
let mut writer = match state.artifact_store.writer(
run_id,
&ArtifactKey::new(stage_id.clone(), retry, entry.path.clone()),
) {
Ok(writer) => writer,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let mut bytes_written = 0_u64;
let mut sha256 = Sha256::new();
while let Some(chunk) = match field
.chunk()
.await
.map_err(|err| bad_request_response(format!("invalid multipart body: {err}")))
{
Ok(chunk) => chunk,
Err(response) => return response,
} {
let chunk_len = u64::try_from(chunk.len()).unwrap_or(u64::MAX);
bytes_written = bytes_written.saturating_add(chunk_len);
total_bytes = total_bytes.saturating_add(chunk_len);
if bytes_written > MAX_SINGLE_ARTIFACT_BYTES {
return payload_too_large_response(format!(
"artifact {} exceeds the {MAX_SINGLE_ARTIFACT_BYTES} byte limit",
entry.path
));
}
if total_bytes > MAX_MULTIPART_REQUEST_BYTES {
return payload_too_large_response(format!(
"multipart upload exceeds the {MAX_MULTIPART_REQUEST_BYTES} byte limit"
));
}
sha256.update(&chunk);
if let Err(err) = writer.write_all(&chunk).await {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
if let Some(expected_bytes) = entry.expected_bytes {
if bytes_written != expected_bytes {
return bad_request_response(format!(
"multipart part {part_name} expected {expected_bytes} bytes but received {bytes_written}"
));
}
}
if let Some(expected_sha256) = entry.sha256.as_ref() {
let actual_sha256 = hex::encode(sha256.finalize());
if actual_sha256 != *expected_sha256 {
return bad_request_response(format!(
"multipart part {part_name} sha256 did not match manifest"
));
}
}
if let Err(err) = writer.shutdown().await {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
if !expected_parts.is_empty() {
let mut missing = expected_parts.into_keys().collect::<Vec<_>>();
missing.sort();
return bad_request_response(format!(
"multipart upload is missing part(s): {}",
missing.join(", ")
));
}
StatusCode::NO_CONTENT.into_response()
}
async fn put_stage_artifact(
State(state): State<Arc<AppState>>,
RequireStageArtifact(id, stage_id): RequireStageArtifact,
Query(params): Query<ArtifactFilenameParams>,
request: axum_extract::Request,
) -> Response {
let (parts, body) = request.into_parts();
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
if let Err(response) = load_run_spec(state.as_ref(), &id).await.map(|_| ()) {
return response;
}
let retry = match required_query_param(params.retry.as_ref(), "retry") {
Ok(retry) => retry,
Err(response) => return response,
};
let content_length = match content_length_from_headers(&parts.headers) {
Ok(length) => length,
Err(response) => return response,
};
match artifact_upload_content_type(&parts.headers) {
Ok(ArtifactUploadContentType::OctetStream) => {
let filename = match required_query_param(params.filename.as_ref(), "filename") {
Ok(filename) => filename,
Err(response) => return response,
};
upload_stage_artifact_octet_stream(
state.as_ref(),
&id,
&stage_id,
retry,
filename,
body,
content_length,
)
.await
}
Ok(ArtifactUploadContentType::Multipart { boundary }) => {
if content_length.is_some_and(|length| length > MAX_MULTIPART_REQUEST_BYTES) {
return payload_too_large_response(format!(
"multipart upload exceeds the {MAX_MULTIPART_REQUEST_BYTES} byte limit"
));
}
upload_stage_artifact_multipart(state.as_ref(), &id, &stage_id, retry, boundary, body)
.await
}
Err(response) => response,
}
}
async fn get_stage_artifact(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path((id, stage_id)): Path<(String, String)>,
Query(params): Query<ArtifactFilenameParams>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let stage_id = match parse_stage_id_path(&stage_id) {
Ok(stage_id) => stage_id,
Err(response) => return response,
};
let filename = match required_query_param(params.filename.as_ref(), "filename") {
Ok(filename) => filename,
Err(response) => return response,
};
let retry = match required_query_param(params.retry.as_ref(), "retry") {
Ok(retry) => retry,
Err(response) => return response,
};
let relative_path = match validate_relative_artifact_path("filename", &filename) {
Ok(path) => path,
Err(response) => return response,
};
if let Err(response) = load_run_spec(state.as_ref(), &id).await {
return response;
}
match state
.artifact_store
.get(
&id,
&ArtifactKey::new(stage_id.clone(), retry, relative_path),
)
.await
{
Ok(Some(bytes)) => octet_stream_response(bytes),
Ok(None) => ApiError::not_found("Artifact not found.").into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}

View file

@ -0,0 +1,236 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, BilledTokenCounts, BillingByModel, BillingStageRef, EventEnvelope, HashMap,
IntoResponse, Json, ListResponse, ModelBillingTotals, ModelReference, PaginationParams, Path,
Query, RequiredUser, Response, Router, RunBilling, RunBillingStage, RunBillingTotals, RunId,
RunStage, RunStatus, StageState, State, StatusCode, accumulate_model_billing, get,
parse_run_id_path,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs/{id}/stages", get(list_run_stages))
.route("/runs/{id}/billing", get(get_run_billing))
}
fn active_stage_state_from_events(events: &[EventEnvelope], node_id: &str) -> StageState {
let latest = events.iter().rev().find(|envelope| {
envelope.event.node_id.as_deref() == Some(node_id)
&& matches!(
envelope.event.event_name(),
"stage.retrying" | "stage.started" | "stage.completed" | "stage.failed"
)
});
if latest.is_some_and(|e| e.event.event_name() == "stage.retrying") {
StageState::Retrying
} else {
StageState::Running
}
}
async fn list_run_stages(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Query(_pagination): Query<PaginationParams>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
// Try live run first.
let (checkpoint, run_is_active) = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => {
let active = !matches!(
managed_run.status,
RunStatus::Succeeded { .. } | RunStatus::Failed { .. } | RunStatus::Dead
);
(managed_run.checkpoint.clone(), active)
}
None => (None, false),
}
};
// Fall back to stored run.
let (checkpoint, run_is_active) = if checkpoint.is_some() {
(checkpoint, run_is_active)
} else {
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => {
let active = run_state.status.is_some_and(|status| !status.is_terminal());
(run_state.checkpoint, active)
}
Err(_) => (None, false),
},
Err(_) => return ApiError::not_found("Run not found.").into_response(),
}
};
let Some(checkpoint) = checkpoint else {
return (
StatusCode::OK,
Json(ListResponse::new(Vec::<RunStage>::new())),
)
.into_response();
};
let events = match state.store.open_run_reader(&id).await {
Ok(run_store) => run_store.list_events().await.unwrap_or_default(),
Err(_) => Vec::new(),
};
let stage_durations = fabro_workflow::extract_stage_durations_from_events(&events);
let mut stages = Vec::new();
for node_id in &checkpoint.completed_nodes {
let duration_ms = stage_durations.get(node_id).copied().unwrap_or(0);
let status = match checkpoint.node_outcomes.get(node_id) {
Some(outcome) => StageState::from(outcome.status),
None => StageState::Succeeded,
};
stages.push(RunStage {
id: node_id.clone(),
name: node_id.clone(),
status,
duration_secs: Some(duration_ms as f64 / 1000.0),
dot_id: Some(node_id.clone()),
});
}
// Add next node as running if the run is still active.
// The checkpoint's current_node is the last *completed* stage; next_node_id
// is the stage that is currently executing.
if let Some(next_id) = &checkpoint.next_node_id {
if run_is_active && next_id != "exit" && !checkpoint.completed_nodes.contains(next_id) {
stages.push(RunStage {
id: next_id.clone(),
name: next_id.clone(),
status: active_stage_state_from_events(&events, next_id),
duration_secs: None,
dot_id: Some(next_id.clone()),
});
}
}
(StatusCode::OK, Json(ListResponse::new(stages))).into_response()
}
async fn get_run_billing(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<RunId>,
) -> Response {
let run_store = match state.store.open_run_reader(&id).await {
Ok(run_store) => run_store,
Err(err) => {
return ApiError::new(StatusCode::NOT_FOUND, err.to_string()).into_response();
}
};
let checkpoint = match run_store.state().await {
Ok(state) => state.checkpoint,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let Some(checkpoint) = checkpoint else {
let empty = RunBilling {
by_model: Vec::new(),
stages: Vec::new(),
totals: RunBillingTotals {
cache_read_tokens: 0,
cache_write_tokens: 0,
input_tokens: 0,
output_tokens: 0,
reasoning_tokens: 0,
runtime_secs: 0.0,
total_tokens: 0,
total_usd_micros: None,
},
};
return (StatusCode::OK, Json(empty)).into_response();
};
let stage_durations = match run_store.list_events().await {
Ok(events) => fabro_workflow::extract_stage_durations_from_events(&events),
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let mut by_model_totals = HashMap::<String, ModelBillingTotals>::new();
let mut billed_usages = Vec::new();
let mut runtime_secs = 0.0_f64;
let mut stages = Vec::new();
for node_id in &checkpoint.completed_nodes {
let duration_ms = stage_durations.get(node_id).copied().unwrap_or(0);
runtime_secs += duration_ms as f64 / 1000.0;
let Some(usage) = checkpoint
.node_outcomes
.get(node_id)
.and_then(|outcome| outcome.usage.as_ref())
else {
continue;
};
billed_usages.push(usage.clone());
let tokens = usage.tokens();
let billing = BilledTokenCounts {
cache_read_tokens: tokens.cache_read_tokens,
cache_write_tokens: tokens.cache_write_tokens,
input_tokens: tokens.input_tokens,
output_tokens: tokens.output_tokens,
reasoning_tokens: tokens.reasoning_tokens,
total_tokens: tokens.total_tokens(),
total_usd_micros: usage.total_usd_micros,
};
let model_id = usage.model_id().to_string();
accumulate_model_billing(by_model_totals.entry(model_id.clone()).or_default(), usage);
stages.push(RunBillingStage {
billing,
model: ModelReference { id: model_id },
runtime_secs: duration_ms as f64 / 1000.0,
stage: BillingStageRef {
id: node_id.clone(),
name: node_id.clone(),
},
});
}
let totals = BilledTokenCounts::from_billed_usage(&billed_usages);
let by_model = by_model_totals
.into_iter()
.map(|(model, totals)| BillingByModel {
billing: totals.billing,
model: ModelReference { id: model },
stages: totals.stages,
})
.collect::<Vec<_>>();
let response = RunBilling {
by_model,
stages,
totals: RunBillingTotals {
cache_read_tokens: totals.cache_read_tokens,
cache_write_tokens: totals.cache_write_tokens,
input_tokens: totals.input_tokens,
output_tokens: totals.output_tokens,
reasoning_tokens: totals.reasoning_tokens,
runtime_secs,
total_tokens: totals.total_tokens,
total_usd_micros: totals.total_usd_micros,
},
};
(StatusCode::OK, Json(response)).into_response()
}

View file

@ -0,0 +1,269 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, CompletionContentPart, CompletionMessage, CompletionMessageRole,
CompletionResponse, CompletionToolChoiceMode, CompletionUsage, ContentPart,
CreateCompletionRequest, Duration, Event, FinishReason, GenerateParams, IntoResponse, Json,
KeepAlive, LlmMessage, LlmRequest, RequiredUser, Response, Role, Router, Sse, State,
StatusCode, ToolChoice, ToolDefinition, Ulid, error, generate_object, info, post, warn,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new().route("/completions", post(create_completion))
}
fn finish_reason_to_api_stop_reason(reason: &FinishReason) -> String {
match reason {
FinishReason::Stop => "end_turn".to_string(),
FinishReason::Length => "max_tokens".to_string(),
FinishReason::ToolCalls => "tool_calls".to_string(),
FinishReason::ContentFilter => "content_filter".to_string(),
FinishReason::Error => "error".to_string(),
FinishReason::Other(s) => s.clone(),
}
}
fn convert_api_message(msg: &CompletionMessage) -> LlmMessage {
let role = match msg.role {
CompletionMessageRole::System => Role::System,
CompletionMessageRole::User => Role::User,
CompletionMessageRole::Assistant => Role::Assistant,
CompletionMessageRole::Tool => Role::Tool,
CompletionMessageRole::Developer => Role::Developer,
};
let content: Vec<ContentPart> = msg
.content
.iter()
.filter_map(|part| {
let json = serde_json::to_value(part).ok()?;
serde_json::from_value(json).ok()
})
.collect();
LlmMessage {
role,
content,
name: msg.name.clone(),
tool_call_id: msg.tool_call_id.clone(),
}
}
fn convert_llm_message(msg: &LlmMessage) -> CompletionMessage {
let role = match msg.role {
Role::System => CompletionMessageRole::System,
Role::User => CompletionMessageRole::User,
Role::Assistant => CompletionMessageRole::Assistant,
Role::Tool => CompletionMessageRole::Tool,
Role::Developer => CompletionMessageRole::Developer,
};
let content: Vec<CompletionContentPart> = msg
.content
.iter()
.filter_map(|part| {
let json = serde_json::to_value(part).ok()?;
serde_json::from_value(json).ok()
})
.collect();
CompletionMessage {
role,
content,
name: msg.name.clone(),
tool_call_id: msg.tool_call_id.clone(),
}
}
async fn create_completion(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Json(req): Json<CreateCompletionRequest>,
) -> Response {
// Resolve model
let model_id = req.model.unwrap_or_else(|| {
fabro_model::Catalog::builtin()
.list(None)
.first()
.map_or_else(|| "claude-sonnet-4-5".to_string(), |m| m.id.clone())
});
let catalog_info = fabro_model::Catalog::builtin().get(&model_id);
// Resolve provider: explicit request > catalog > None
let provider_name = req
.provider
.or_else(|| catalog_info.map(|i| i.provider.to_string()));
info!(model = %model_id, provider = ?provider_name, "Completion request received");
// Build messages list
let mut messages: Vec<LlmMessage> = Vec::new();
if let Some(system) = req.system {
messages.push(LlmMessage::system(system));
}
for msg in &req.messages {
messages.push(convert_api_message(msg));
}
// Convert tools
let tools: Option<Vec<ToolDefinition>> = if req.tools.is_empty() {
None
} else {
Some(
req.tools
.into_iter()
.map(|t| ToolDefinition {
name: t.name,
description: t.description,
parameters: t.parameters,
})
.collect(),
)
};
// Convert tool_choice
let tool_choice: Option<ToolChoice> = req.tool_choice.map(|tc| match tc.mode {
CompletionToolChoiceMode::Auto => ToolChoice::Auto,
CompletionToolChoiceMode::None => ToolChoice::None,
CompletionToolChoiceMode::Required => ToolChoice::Required,
CompletionToolChoiceMode::Named => ToolChoice::named(tc.tool_name.unwrap_or_default()),
});
// Build the LLM request
let request = LlmRequest {
model: model_id.clone(),
messages,
provider: provider_name,
tools,
tool_choice,
response_format: None,
temperature: req.temperature,
top_p: req.top_p,
max_tokens: req.max_tokens,
stop_sequences: if req.stop_sequences.is_empty() {
None
} else {
Some(req.stop_sequences)
},
reasoning_effort: req.reasoning_effort.as_deref().and_then(|s| s.parse().ok()),
speed: None,
metadata: None,
provider_options: req.provider_options,
};
// Force non-streaming for structured output
let use_stream = req.stream && req.schema.is_none();
let llm_result = match state.resolve_llm_client().await {
Ok(result) => result,
Err(err) => {
error!(error = ?err, "Failed to create LLM client");
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to create LLM client: {err}"),
)
.into_response();
}
};
for (provider, issue) in &llm_result.auth_issues {
warn!(provider = %provider, error = %issue, "LLM provider unavailable due to auth issue");
}
let client = llm_result.client;
if use_stream {
// Streaming path: forward all StreamEvents as SSE
let stream_result = match client.stream(&request).await {
Ok(s) => s,
Err(e) => {
return ApiError::new(StatusCode::BAD_GATEWAY, format!("LLM error: {e}"))
.into_response();
}
};
let sse_stream = tokio_stream::StreamExt::filter_map(stream_result, |event| match event {
Ok(ref evt) => match serde_json::to_string(evt) {
Ok(json) => Some(Ok::<_, std::convert::Infallible>(
Event::default().event("stream_event").data(json),
)),
Err(e) => Some(Ok(Event::default().event("stream_event").data(
serde_json::json!({
"type": "error",
"error": {"Stream": {"message": format!("failed to serialize event: {e}")}},
"raw": null
})
.to_string(),
))),
},
Err(e) => Some(Ok(Event::default().event("stream_event").data(
serde_json::json!({
"type": "error",
"error": {"Stream": {"message": e.to_string()}},
"raw": null
})
.to_string(),
))),
});
Sse::new(sse_stream)
.keep_alive(
KeepAlive::new().interval(Duration::from_secs(15)).event(
Event::default()
.event("ping")
.data(serde_json::json!({"type": "ping"}).to_string()),
),
)
.into_response()
} else {
// Non-streaming path
let msg_id = Ulid::new().to_string();
if let Some(schema) = req.schema {
// Structured output uses generate_object for JSON parsing logic
let mut params =
GenerateParams::new(&request.model, std::sync::Arc::new(client.clone()))
.messages(request.messages);
if let Some(ref p) = request.provider {
params = params.provider(p);
}
if let Some(temp) = request.temperature {
params = params.temperature(temp);
}
if let Some(max_tokens) = request.max_tokens {
params = params.max_tokens(max_tokens);
}
if let Some(top_p) = request.top_p {
params = params.top_p(top_p);
}
match generate_object(params, schema).await {
Ok(result) => Json(CompletionResponse {
id: msg_id,
model: model_id,
message: convert_llm_message(&result.response.message),
stop_reason: finish_reason_to_api_stop_reason(&result.finish_reason),
usage: CompletionUsage {
input_tokens: result.usage.input_tokens,
output_tokens: result.usage.output_tokens,
},
output: result.output,
})
.into_response(),
Err(e) => ApiError::new(StatusCode::BAD_GATEWAY, format!("LLM error: {e}"))
.into_response(),
}
} else {
match client.complete(&request).await {
Ok(response) => Json(CompletionResponse {
id: response.id,
model: response.model,
message: convert_llm_message(&response.message),
stop_reason: finish_reason_to_api_stop_reason(&response.finish_reason),
usage: CompletionUsage {
input_tokens: response.usage.input_tokens,
output_tokens: response.usage.output_tokens,
},
output: None,
})
.into_response(),
Err(e) => ApiError::new(StatusCode::BAD_GATEWAY, format!("LLM error: {e}"))
.into_response(),
}
}
}
}

View file

@ -0,0 +1,344 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, AppendEventResponse, BroadcastStream, Event, EventBody, EventEnvelope,
EventPayload, HashSet, IntoResponse, Json, KeepAlive, PaginatedEventList, PaginationMeta, Path,
Query, RequireRunScoped, RequiredUser, Response, Router, RunEvent, RunId, RunStatus, Sse,
State, StatusCode, StreamExt, UnboundedReceiverStream, broadcast, get, mpsc, parse_run_id_path,
redact_jsonl_line, reject_if_archived, update_live_run_from_event,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/attach", get(attach_events))
.route(
"/runs/{id}/events",
get(list_run_events).post(append_run_event),
)
.route("/runs/{id}/attach", get(attach_run_events))
}
#[derive(serde::Deserialize)]
struct EventListParams {
#[serde(default)]
since_seq: Option<u32>,
#[serde(default)]
limit: Option<usize>,
}
impl EventListParams {
fn since_seq(&self) -> u32 {
self.since_seq.unwrap_or(1).max(1)
}
fn limit(&self) -> usize {
self.limit.unwrap_or(100).clamp(1, 1000)
}
}
#[derive(serde::Deserialize)]
struct AttachParams {
#[serde(default)]
since_seq: Option<u32>,
}
#[derive(serde::Deserialize)]
struct GlobalAttachParams {
#[serde(default)]
run_id: Option<String>,
}
async fn attach_events(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Query(params): Query<GlobalAttachParams>,
) -> Response {
let run_filter = match parse_global_run_filter(params.run_id.as_deref()) {
Ok(filter) => filter,
Err(err) => return ApiError::new(StatusCode::BAD_REQUEST, err).into_response(),
};
let stream =
filtered_global_events(state.global_event_tx.subscribe(), run_filter).filter_map(|event| {
sse_event_from_store(&event).map(Ok::<Event, std::convert::Infallible>)
});
Sse::new(stream)
.keep_alive(KeepAlive::default())
.into_response()
}
pub(in crate::server) fn filtered_global_events(
event_rx: broadcast::Receiver<EventEnvelope>,
run_filter: Option<HashSet<RunId>>,
) -> impl tokio_stream::Stream<Item = EventEnvelope> {
BroadcastStream::new(event_rx).filter_map(move |result| match result {
Ok(event) if event_matches_run_filter(&event, run_filter.as_ref()) => Some(event),
Ok(_) | Err(_) => None,
})
}
fn parse_global_run_filter(raw: Option<&str>) -> Result<Option<HashSet<RunId>>, String> {
let Some(raw) = raw else {
return Ok(None);
};
let mut run_ids = HashSet::new();
for part in raw
.split(',')
.map(str::trim)
.filter(|part| !part.is_empty())
{
let run_id = part
.parse::<RunId>()
.map_err(|err| format!("invalid run_id '{part}': {err}"))?;
run_ids.insert(run_id);
}
if run_ids.is_empty() {
Ok(None)
} else {
Ok(Some(run_ids))
}
}
fn event_matches_run_filter(event: &EventEnvelope, run_filter: Option<&HashSet<RunId>>) -> bool {
let Some(run_filter) = run_filter else {
return true;
};
run_filter.contains(&event.event.run_id)
}
fn sse_event_from_store(event: &EventEnvelope) -> Option<Event> {
let data = serde_json::to_string(event).ok()?;
let data = redact_jsonl_line(&data);
Some(Event::default().data(data))
}
fn attach_event_is_terminal(event: &EventEnvelope) -> bool {
matches!(
&event.event.body,
EventBody::RunCompleted(_) | EventBody::RunFailed(_)
)
}
fn run_projection_is_active(state: &fabro_store::RunProjection) -> bool {
state.status.is_some_and(RunStatus::is_active)
}
async fn append_run_event(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
Json(value): Json<serde_json::Value>,
) -> Response {
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let event = match RunEvent::from_value(value.clone()) {
Ok(event) => event,
Err(err) => {
return ApiError::bad_request(format!("Invalid run event: {err}")).into_response();
}
};
if event.run_id != id {
return ApiError::bad_request("Event run_id does not match path run ID.").into_response();
}
if let Some(denied) = denied_lifecycle_event_name(&event.body) {
return ApiError::bad_request(format!(
"{denied} is a lifecycle event; clients must call the corresponding operation endpoint instead of injecting it via append_run_event"
))
.into_response();
}
let payload = match EventPayload::new(value, &id) {
Ok(payload) => payload,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
match state.store.open_run(&id).await {
Ok(run_store) => match run_store.append_event(&payload).await {
Ok(seq) => {
update_live_run_from_event(&state, id, &event);
Json(AppendEventResponse {
seq: i64::from(seq),
})
.into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
}
async fn list_run_events(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
Query(params): Query<EventListParams>,
) -> Response {
let since_seq = params.since_seq();
let limit = params.limit();
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store
.list_events_from_with_limit(since_seq, limit)
.await
{
Ok(mut events) => {
let has_more = events.len() > limit;
events.truncate(limit);
Json(PaginatedEventList {
data: events,
meta: PaginationMeta { has_more },
})
.into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
}
async fn attach_run_events(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Query(params): Query<AttachParams>,
) -> Response {
const ATTACH_REPLAY_BATCH_LIMIT: usize = 256;
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let Ok(run_store) = state.store.open_run_reader(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let start_seq = match params.since_seq {
Some(seq) if seq >= 1 => seq,
Some(_) => 1,
None => match run_store.list_events().await {
Ok(events) => events.last().map_or(1, |event| event.seq.saturating_add(1)),
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
},
};
let (sender, receiver) = mpsc::unbounded_channel();
tokio::spawn(async move {
let mut next_seq = start_seq;
loop {
let Ok(replay_batch) = run_store
.list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT)
.await
else {
return;
};
let replay_has_more = replay_batch.len() > ATTACH_REPLAY_BATCH_LIMIT;
for event in replay_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) {
next_seq = event.seq.saturating_add(1);
let terminal = attach_event_is_terminal(&event);
if let Some(sse_event) = sse_event_from_store(&event) {
if sender
.send(Ok::<Event, std::convert::Infallible>(sse_event))
.is_err()
{
return;
}
}
if terminal {
return;
}
}
if replay_has_more {
continue;
}
let Ok(state) = run_store.state().await else {
return;
};
if run_projection_is_active(&state) {
break;
}
let Ok(tail_batch) = run_store
.list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT)
.await
else {
return;
};
let tail_has_more = tail_batch.len() > ATTACH_REPLAY_BATCH_LIMIT;
for event in tail_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) {
next_seq = event.seq.saturating_add(1);
let terminal = attach_event_is_terminal(&event);
if let Some(sse_event) = sse_event_from_store(&event) {
if sender
.send(Ok::<Event, std::convert::Infallible>(sse_event))
.is_err()
{
return;
}
}
if terminal {
return;
}
}
if tail_has_more {
continue;
}
return;
}
let Ok(mut live_stream) = run_store.watch_events_from(next_seq) else {
return;
};
while let Some(result) = live_stream.next().await {
let Ok(event) = result else {
return;
};
let terminal = attach_event_is_terminal(&event);
if let Some(sse_event) = sse_event_from_store(&event) {
if sender
.send(Ok::<Event, std::convert::Infallible>(sse_event))
.is_err()
{
return;
}
}
if terminal {
return;
}
}
});
Sse::new(UnboundedReceiverStream::new(receiver))
.keep_alive(KeepAlive::default())
.into_response()
}
/// Returns the wire event name if the given body has a dedicated operation
/// endpoint that clients must use instead of injecting via `append_run_event`.
/// These endpoints enforce authorization and status-transition preconditions
/// (e.g. "archive only from terminal") that a direct event append would
/// bypass. Other run-lifecycle events flow through this endpoint legitimately:
/// the worker subprocess emits state transitions during execution.
fn denied_lifecycle_event_name(body: &EventBody) -> Option<&'static str> {
match body {
EventBody::RunArchived(_) => Some("run.archived"),
EventBody::RunUnarchived(_) => Some("run.unarchived"),
EventBody::RunCancelRequested(_) => Some("run.cancel.requested"),
EventBody::RunPauseRequested(_) => Some("run.pause.requested"),
EventBody::RunUnpauseRequested(_) => Some("run.unpause.requested"),
_ => None,
}
}

View file

@ -0,0 +1,298 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, AsyncWriteExt, Command, EnvVars, IntoResponse, Json, LazyLock, Path,
PathBuf, Query, RenderWorkflowGraphDirection, RenderWorkflowGraphRequest, RequiredUser,
Response, Router, RunId, Semaphore, State, StatusCode, Stdio, apply_render_graph_env, get,
parse_run_id_path, post, run_manifest,
};
pub(super) fn manifest_routes() -> Router<Arc<AppState>> {
Router::new().route("/graph/render", post(render_graph_from_manifest))
}
pub(super) fn run_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs/{id}/graph", get(get_graph))
.route("/runs/{id}/graph/source", get(get_graph_source))
}
const RENDER_ERROR_PREFIX: &[u8] = b"RENDER_ERROR:";
const GRAPHVIZ_RENDER_CONCURRENCY_LIMIT: usize = 4;
static GRAPHVIZ_RENDER_SEMAPHORE: LazyLock<Semaphore> =
LazyLock::new(|| Semaphore::new(GRAPHVIZ_RENDER_CONCURRENCY_LIMIT));
#[derive(Debug, thiserror::Error)]
pub(in crate::server) enum RenderSubprocessError {
#[error("failed to spawn render subprocess: {0}")]
SpawnFailed(String),
#[error("render subprocess crashed: {0}")]
ChildCrashed(String),
#[error("render subprocess returned invalid output: {0}")]
ProtocolViolation(String),
#[error("{0}")]
RenderFailed(String),
}
async fn render_graph_from_manifest(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Json(req): Json<RenderWorkflowGraphRequest>,
) -> Response {
let manifest_run_defaults = state.manifest_run_defaults();
let prepared =
match run_manifest::prepare_manifest(manifest_run_defaults.as_ref(), &req.manifest) {
Ok(prepared) => prepared,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
let validated = match run_manifest::validate_prepared_manifest(&prepared) {
Ok(validated) => validated,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
if validated.has_errors() {
return ApiError::bad_request("Validation failed").into_response();
}
let direction = req.direction.as_ref().map(|direction| match direction {
RenderWorkflowGraphDirection::Lr => "LR",
RenderWorkflowGraphDirection::Tb => "TB",
});
let dot_source = run_manifest::graph_source(&prepared, direction);
render_graph_bytes(&dot_source).await
}
#[expect(
clippy::disallowed_methods,
reason = "Render-graph subprocess startup resolves Cargo's test binary env override when present."
)]
fn render_graph_subprocess_exe(
exe_override: Option<&std::path::Path>,
) -> Result<PathBuf, RenderSubprocessError> {
if let Some(path) = exe_override {
Ok(path.to_path_buf())
} else {
if let Some(path) = std::env::var_os(EnvVars::CARGO_BIN_EXE_FABRO).map(PathBuf::from) {
return Ok(path);
}
let current = std::env::current_exe()
.map_err(|err| RenderSubprocessError::SpawnFailed(err.to_string()))?;
let current_name = current.file_stem().and_then(|name| name.to_str());
if current_name == Some("fabro") {
return Ok(current);
}
let candidate = current
.parent()
.and_then(|parent| parent.parent())
.map(|parent| parent.join(if cfg!(windows) { "fabro.exe" } else { "fabro" }));
if let Some(candidate) = candidate.filter(|path| path.is_file()) {
return Ok(candidate);
}
Ok(current)
}
}
fn render_subprocess_failure(
status: std::process::ExitStatus,
stderr: &[u8],
) -> RenderSubprocessError {
#[cfg(unix)]
{
use std::os::unix::process::ExitStatusExt;
if let Some(signal) = status.signal() {
let stderr = String::from_utf8_lossy(stderr).trim().to_string();
let detail = if stderr.is_empty() {
format!("terminated by signal {signal}")
} else {
format!("terminated by signal {signal}: {stderr}")
};
return RenderSubprocessError::ChildCrashed(detail);
}
}
let stderr = String::from_utf8_lossy(stderr).trim().to_string();
let detail = match status.code() {
Some(code) if stderr.is_empty() => format!("exited with status {code}"),
Some(code) => format!("exited with status {code}: {stderr}"),
None if stderr.is_empty() => "child exited unsuccessfully".to_string(),
None => format!("child exited unsuccessfully: {stderr}"),
};
RenderSubprocessError::ChildCrashed(detail)
}
pub(in crate::server) async fn render_dot_subprocess(
styled_source: &str,
exe_override: Option<&std::path::Path>,
) -> Result<Vec<u8>, RenderSubprocessError> {
let _permit = GRAPHVIZ_RENDER_SEMAPHORE
.acquire()
.await
.map_err(|err| RenderSubprocessError::SpawnFailed(err.to_string()))?;
let exe = render_graph_subprocess_exe(exe_override)?;
let mut cmd = Command::new(exe);
apply_render_graph_env(&mut cmd);
cmd.arg("__render-graph")
.env(EnvVars::FABRO_TELEMETRY, "off")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
let mut child = cmd
.spawn()
.map_err(|err| RenderSubprocessError::SpawnFailed(err.to_string()))?;
let mut stdin = child.stdin.take().ok_or_else(|| {
RenderSubprocessError::SpawnFailed("render subprocess stdin was not piped".to_string())
})?;
if let Err(err) = stdin.write_all(styled_source.as_bytes()).await {
drop(stdin);
let output = child
.wait_with_output()
.await
.map_err(|wait_err| RenderSubprocessError::SpawnFailed(wait_err.to_string()))?;
return Err(RenderSubprocessError::ChildCrashed(format!(
"failed writing DOT to child stdin: {err}; {}",
render_subprocess_failure(output.status, &output.stderr)
)));
}
drop(stdin);
let output = child
.wait_with_output()
.await
.map_err(|err| RenderSubprocessError::SpawnFailed(err.to_string()))?;
if !output.status.success() {
return Err(render_subprocess_failure(output.status, &output.stderr));
}
if let Some(error) = output.stdout.strip_prefix(RENDER_ERROR_PREFIX) {
return Err(RenderSubprocessError::RenderFailed(
String::from_utf8_lossy(error).trim().to_string(),
));
}
if output.stdout.starts_with(b"<?xml") || output.stdout.starts_with(b"<svg") {
return Ok(output.stdout);
}
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
Err(RenderSubprocessError::ProtocolViolation(format!(
"stdout did not contain SVG or error protocol (stdout: {:?}, stderr: {:?})",
stdout.trim(),
stderr.trim()
)))
}
async fn render_graph_response(
dot_source: &str,
exe_override: Option<&std::path::Path>,
) -> Response {
use fabro_graphviz::render::{inject_dot_style_defaults, postprocess_svg};
let styled_source = inject_dot_style_defaults(dot_source);
match render_dot_subprocess(&styled_source, exe_override).await {
Ok(raw) => {
let bytes = postprocess_svg(raw);
(StatusCode::OK, [("content-type", "image/svg+xml")], bytes).into_response()
}
Err(RenderSubprocessError::RenderFailed(err)) => {
ApiError::new(StatusCode::BAD_REQUEST, err).into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
pub(crate) async fn render_graph_bytes(dot_source: &str) -> Response {
render_graph_response(dot_source, None).await
}
#[cfg(test)]
pub(in crate::server) async fn render_graph_bytes_with_exe_override(
dot_source: &str,
exe_override: Option<&std::path::Path>,
) -> Response {
render_graph_response(dot_source, exe_override).await
}
#[derive(serde::Deserialize)]
struct GraphParams {
#[serde(default)]
direction: Option<String>,
}
async fn load_run_dot_source(state: &AppState, id: &RunId) -> Result<String, Response> {
let live_dot_source = {
let runs = state.runs.lock().expect("runs lock poisoned");
runs.get(id)
.map(|managed_run| managed_run.dot_source.clone())
};
let dot_source = if let Some(dot) = live_dot_source.filter(|d| !d.is_empty()) {
Some(dot)
} else {
match state.store.open_run_reader(id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => run_state.graph_source,
Err(err) => {
return Err(
ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response()
);
}
},
Err(_) => return Err(ApiError::not_found("Run not found.").into_response()),
}
};
dot_source
.ok_or_else(|| ApiError::new(StatusCode::NOT_FOUND, "Graph not found.").into_response())
}
async fn get_graph(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Query(params): Query<GraphParams>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let dot = match load_run_dot_source(&state, &id).await {
Ok(dot) => dot,
Err(response) => return response,
};
let dot = match params.direction.as_deref() {
Some(dir @ ("LR" | "TB" | "BT" | "RL")) => {
use fabro_graphviz::render;
render::apply_direction(&dot, dir).into_owned()
}
_ => dot,
};
render_graph_bytes(&dot).await
}
async fn get_graph_source(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
match load_run_dot_source(&state, &id).await {
Ok(dot) => (StatusCode::OK, [("content-type", "text/vnd.graphviz")], dot).into_response(),
Err(response) => response,
}
}

View file

@ -0,0 +1,797 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, FailureReason, ForkRequest, ForkResponse, IntoResponse, Json, Ordering,
Path, Principal, RequiredUser, Response, RewindRequest, RewindResponse, Router,
RunAnswerTransport, RunControlAction, RunExecutionMode, RunId, RunStatus, RunStatusResponse,
StartRunRequest, State, StatusCode, Storage, TimelineEntryResponse, WORKER_CANCEL_GRACE,
WorkflowError, append_control_request, get, load_pending_control, managed_run, operations,
parse_run_id_path, persist_cancelled_run_status, post, reject_if_archived, sleep,
update_live_run_from_event, workflow_event,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs/{id}/cancel", post(cancel_run))
.route("/runs/{id}/start", post(start_run))
.route("/runs/{id}/pause", post(pause_run))
.route("/runs/{id}/unpause", post(unpause_run))
.route("/runs/{id}/archive", post(archive_run))
.route("/runs/{id}/rewind", post(rewind_run))
.route("/runs/{id}/fork", post(fork_run))
.route("/runs/{id}/timeline", get(run_timeline))
.route("/runs/{id}/unarchive", post(unarchive_run))
}
async fn start_run(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
body: Option<Json<StartRunRequest>>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let resume = body.is_some_and(|Json(req)| req.resume);
{
let runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get(&id) {
if matches!(
managed_run.status,
RunStatus::Queued
| RunStatus::Starting
| RunStatus::Running
| RunStatus::Blocked { .. }
| RunStatus::Paused { .. }
) {
return ApiError::new(
StatusCode::CONFLICT,
if resume {
"an engine process is still running for this run — cannot resume"
} else {
"an engine process is still running for this run — cannot start"
},
)
.into_response();
}
}
}
let Ok(run_store) = state.store.open_run(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let run_state = match run_store.state().await {
Ok(state) => state,
Err(err) => {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to load run state: {err}"),
)
.into_response();
}
};
if resume {
if run_state.checkpoint.is_none() {
return ApiError::new(StatusCode::CONFLICT, "no checkpoint to resume from")
.into_response();
}
} else if let Some(status) = run_state.status {
if !matches!(
status,
RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting
) {
return ApiError::new(
StatusCode::CONFLICT,
format!("cannot start run: status is {status}, expected submitted"),
)
.into_response();
}
}
if run_state.spec.is_none() {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"run spec missing from store",
)
.into_response();
}
let run_dir = Storage::new(state.server_storage_dir())
.run_scratch(&id)
.root()
.to_path_buf();
let dot_source = run_state.graph_source.unwrap_or_default();
if let Err(err) =
workflow_event::append_event(&run_store, &id, &workflow_event::Event::RunQueued).await
{
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response();
}
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
runs.insert(
id,
managed_run(
dot_source,
RunStatus::Queued,
id.created_at(),
run_dir,
if resume {
RunExecutionMode::Resume
} else {
RunExecutionMode::Start
},
),
);
}
state.scheduler_notify.notify_one();
(
StatusCode::OK,
Json(RunStatusResponse {
id: id.to_string(),
status: RunStatus::Queued,
error: None,
queue_position: None,
pending_control: None,
created_at: id.created_at(),
}),
)
.into_response()
}
fn schedule_worker_kill(state: Arc<AppState>, run_id: RunId, worker_pid: u32) {
tokio::spawn(async move {
sleep(WORKER_CANCEL_GRACE).await;
let current_pid = {
let runs = state.runs.lock().expect("runs lock poisoned");
runs.get(&run_id).and_then(|run| run.worker_pid)
};
if current_pid == Some(worker_pid) && fabro_proc::process_group_alive(worker_pid) {
#[cfg(unix)]
fabro_proc::sigkill_process_group(worker_pid);
}
});
}
async fn cancel_run(
subject: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let pending_control = match load_pending_control(state.as_ref(), id).await {
Ok(pending_control) => pending_control,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let (
created_at,
response_status,
persist_cancelled_status,
answer_transport,
cancel_token,
cancel_tx,
worker_pid,
) = {
let mut runs = state.runs.lock().expect("runs lock poisoned");
match runs.get_mut(&id) {
Some(managed_run) => match managed_run.status {
RunStatus::Submitted
| RunStatus::Queued
| RunStatus::Starting
| RunStatus::Running
| RunStatus::Blocked { .. }
| RunStatus::Paused { .. } => {
let use_cancel_signal = !matches!(
managed_run.answer_transport,
Some(RunAnswerTransport::InProcess { .. })
);
let persist_cancelled_status =
matches!(managed_run.status, RunStatus::Submitted | RunStatus::Queued);
let response_status = if persist_cancelled_status {
let cancelled = RunStatus::Failed {
reason: FailureReason::Cancelled,
};
managed_run.status = cancelled;
cancelled
} else {
managed_run.status
};
(
managed_run.created_at,
response_status,
persist_cancelled_status,
managed_run.answer_transport.clone(),
managed_run.cancel_token.clone(),
use_cancel_signal
.then(|| managed_run.cancel_tx.take())
.flatten(),
managed_run.worker_pid,
)
}
_ => {
return ApiError::new(StatusCode::CONFLICT, "Run is not cancellable.")
.into_response();
}
},
None => return ApiError::not_found("Run not found.").into_response(),
}
};
if pending_control != Some(RunControlAction::Cancel) {
if let Err(err) = append_control_request(
state.as_ref(),
id,
RunControlAction::Cancel,
Some(Principal::User(subject.0.clone())),
)
.await
{
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
if let Some(token) = &cancel_token {
token.store(true, Ordering::SeqCst);
}
let sent_cancel_signal = if let Some(cancel_tx) = cancel_tx {
let _ = cancel_tx.send(());
true
} else {
false
};
if let Some(answer_transport) = answer_transport {
if !(sent_cancel_signal && matches!(answer_transport, RunAnswerTransport::InProcess { .. }))
{
let _ = answer_transport.cancel_run().await;
}
}
if let Some(worker_pid) = worker_pid {
#[cfg(unix)]
fabro_proc::sigterm(worker_pid);
schedule_worker_kill(Arc::clone(&state), id, worker_pid);
}
if persist_cancelled_status {
if let Err(err) = persist_cancelled_run_status(state.as_ref(), id).await {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
let pending_control = match load_pending_control(state.as_ref(), id).await {
Ok(pending_control) => pending_control,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
(
StatusCode::OK,
Json(RunStatusResponse {
id: id.to_string(),
status: response_status,
error: None,
queue_position: None,
pending_control,
created_at,
}),
)
.into_response()
}
/// How `pause_run` should enact the transition, chosen from the current run
/// status.
enum PauseMode {
/// Worker is running; ask it to pause via SIGUSR1. Status flips to
/// `Paused` once the worker acknowledges.
Signal { worker_pid: u32 },
/// Worker is blocked on a human gate; flip to `Paused` directly by
/// appending `RunPaused` ourselves.
AppendEvent,
}
/// How `unpause_run` should enact the transition.
enum UnpauseMode {
/// No outstanding block; ask the worker to resume via SIGUSR2.
Signal { worker_pid: u32 },
/// Was paused while blocked; append `RunUnpaused` and let the reducer
/// restore the underlying blocked state from `Paused { prior_block }`.
AppendEvent,
}
async fn pause_run(
subject: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let pending_control = match load_pending_control(state.as_ref(), id).await {
Ok(pending_control) => pending_control,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let (created_at, mode) = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) if managed_run.status == RunStatus::Running => {
let Some(worker_pid) = managed_run.worker_pid else {
return ApiError::new(StatusCode::CONFLICT, "Run worker is not available.")
.into_response();
};
(managed_run.created_at, PauseMode::Signal { worker_pid })
}
Some(managed_run) if matches!(managed_run.status, RunStatus::Blocked { .. }) => {
(managed_run.created_at, PauseMode::AppendEvent)
}
Some(_) => {
return ApiError::new(StatusCode::CONFLICT, "Run is not pausable.").into_response();
}
None => return ApiError::not_found("Run not found.").into_response(),
}
};
if pending_control.is_some() {
return ApiError::new(
StatusCode::CONFLICT,
"Run control request is already pending.",
)
.into_response();
}
if let Err(err) = append_control_request(
state.as_ref(),
id,
RunControlAction::Pause,
Some(Principal::User(subject.0.clone())),
)
.await
{
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response();
}
let response_status = match mode {
PauseMode::Signal { worker_pid } => {
#[cfg(unix)]
fabro_proc::sigusr1(worker_pid);
#[cfg(not(unix))]
let _ = worker_pid;
RunStatus::Running
}
PauseMode::AppendEvent => {
if let Some(response) = synchronous_transition(state.as_ref(), id, |events| {
events.push(workflow_event::Event::RunPaused);
})
.await
{
return response;
}
state
.runs
.lock()
.expect("runs lock poisoned")
.get(&id)
.map_or(RunStatus::Paused { prior_block: None }, |run| run.status)
}
};
let pending_control = match load_pending_control(state.as_ref(), id).await {
Ok(pending_control) => pending_control,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
(
StatusCode::OK,
Json(RunStatusResponse {
id: id.to_string(),
status: response_status,
error: None,
queue_position: None,
pending_control,
created_at,
}),
)
.into_response()
}
async fn unpause_run(
subject: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let pending_control = match load_pending_control(state.as_ref(), id).await {
Ok(pending_control) => pending_control,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let (created_at, mode) = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => match managed_run.status {
RunStatus::Paused {
prior_block: Some(_),
} => (managed_run.created_at, UnpauseMode::AppendEvent),
RunStatus::Paused { prior_block: None } => {
let Some(worker_pid) = managed_run.worker_pid else {
return ApiError::new(StatusCode::CONFLICT, "Run worker is not available.")
.into_response();
};
(managed_run.created_at, UnpauseMode::Signal { worker_pid })
}
_ => {
return ApiError::new(StatusCode::CONFLICT, "Run is not paused.")
.into_response();
}
},
None => return ApiError::not_found("Run not found.").into_response(),
}
};
if pending_control.is_some() {
return ApiError::new(
StatusCode::CONFLICT,
"Run control request is already pending.",
)
.into_response();
}
if let Err(err) = append_control_request(
state.as_ref(),
id,
RunControlAction::Unpause,
Some(Principal::User(subject.0.clone())),
)
.await
{
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response();
}
let response_status = match mode {
UnpauseMode::Signal { worker_pid } => {
#[cfg(unix)]
fabro_proc::sigusr2(worker_pid);
#[cfg(not(unix))]
let _ = worker_pid;
RunStatus::Paused { prior_block: None }
}
UnpauseMode::AppendEvent => {
if let Some(response) = synchronous_transition(state.as_ref(), id, |events| {
events.push(workflow_event::Event::RunUnpaused);
})
.await
{
return response;
}
state
.runs
.lock()
.expect("runs lock poisoned")
.get(&id)
.map_or(RunStatus::Running, |run| run.status)
}
};
let pending_control = match load_pending_control(state.as_ref(), id).await {
Ok(pending_control) => pending_control,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
(
StatusCode::OK,
Json(RunStatusResponse {
id: id.to_string(),
status: response_status,
error: None,
queue_position: None,
pending_control,
created_at,
}),
)
.into_response()
}
async fn archive_run(
subject: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
run_archive_action(state, subject, id, ArchiveAction::Archive).await
}
async fn unarchive_run(
subject: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
run_archive_action(state, subject, id, ArchiveAction::Unarchive).await
}
async fn rewind_run(
subject: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
body: Option<Json<RewindRequest>>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let request = body.map(|Json(body)| body).unwrap_or_default();
let target = match parse_fork_target(request.target) {
Ok(target) => target,
Err(err) => return err.into_response(),
};
let input = operations::RewindInput { run_id: id, target };
match Box::pin(operations::rewind(
&state.store,
&input,
Some(Principal::User(subject.0.clone())),
))
.await
{
Ok(operations::RewindOutcome::Full {
source_run_id,
new_run_id,
target,
}) => (
StatusCode::OK,
Json(RewindResponse {
source_run_id: source_run_id.to_string(),
new_run_id: new_run_id.to_string(),
target: target.response_target(),
archived: true,
archive_error: None,
}),
)
.into_response(),
Ok(operations::RewindOutcome::Partial {
source_run_id,
new_run_id,
target,
archive_error,
}) => (
StatusCode::MULTI_STATUS,
Json(RewindResponse {
source_run_id: source_run_id.to_string(),
new_run_id: new_run_id.to_string(),
target: target.response_target(),
archived: false,
archive_error: Some(archive_error),
}),
)
.into_response(),
Err(err) => workflow_operation_error_response(err),
}
}
async fn fork_run(
_subject: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
body: Option<Json<ForkRequest>>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let request = body.map(|Json(body)| body).unwrap_or_default();
let target = match parse_fork_target(request.target) {
Ok(target) => target,
Err(err) => return err.into_response(),
};
let input = operations::ForkRunInput {
source_run_id: id,
target,
};
match Box::pin(operations::fork_run(&state.store, &input)).await {
Ok(outcome) => (
StatusCode::OK,
Json(ForkResponse {
source_run_id: outcome.source_run_id.to_string(),
new_run_id: outcome.new_run_id.to_string(),
target: outcome.target.response_target(),
}),
)
.into_response(),
Err(err) => workflow_operation_error_response(err),
}
}
async fn run_timeline(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
match operations::timeline(&state.store, &id).await {
Ok(entries) => Json(
entries
.into_iter()
.map(|entry| TimelineEntryResponse {
ordinal: std::num::NonZeroU64::new(entry.ordinal as u64)
.expect("timeline ordinals start at 1"),
node_name: entry.node_name,
visit: std::num::NonZeroU64::new(entry.visit as u64)
.expect("timeline visits start at 1"),
checkpoint_seq: std::num::NonZeroU64::new(u64::from(entry.checkpoint_seq))
.expect("checkpoint event sequence starts at 1"),
run_commit_sha: entry.run_commit_sha,
})
.collect::<Vec<_>>(),
)
.into_response(),
Err(err) => workflow_operation_error_response(err),
}
}
fn parse_fork_target(target: Option<String>) -> Result<Option<operations::ForkTarget>, ApiError> {
target
.map(|target| {
target
.parse::<operations::ForkTarget>()
.map_err(|err| ApiError::bad_request(err.to_string()))
})
.transpose()
}
fn workflow_operation_error_response(err: WorkflowError) -> Response {
match err {
WorkflowError::Parse(message) | WorkflowError::Validation(message) => {
ApiError::bad_request(message).into_response()
}
WorkflowError::ValidationFailed { .. } => {
ApiError::bad_request("Validation failed").into_response()
}
WorkflowError::Precondition(message) => {
ApiError::new(StatusCode::CONFLICT, message).into_response()
}
WorkflowError::RunNotFound(_) => ApiError::not_found("Run not found.").into_response(),
WorkflowError::Unsupported(message) => {
ApiError::new(StatusCode::NOT_IMPLEMENTED, message).into_response()
}
err => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
}
}
#[derive(Clone, Copy)]
enum ArchiveAction {
Archive,
Unarchive,
}
async fn run_archive_action(
state: Arc<AppState>,
subject: RequiredUser,
id: String,
action: ArchiveAction,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let actor = Some(Principal::User(subject.0.clone()));
let result = match action {
ArchiveAction::Archive => operations::archive(&state.store, &id, actor)
.await
.map(|_| ()),
ArchiveAction::Unarchive => operations::unarchive(&state.store, &id, actor)
.await
.map(|_| ()),
};
match result {
Ok(()) => archive_status_response(state.as_ref(), id).await,
Err(WorkflowError::Precondition(message)) => {
ApiError::new(StatusCode::CONFLICT, message).into_response()
}
Err(WorkflowError::RunNotFound(_)) => ApiError::not_found("Run not found.").into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
/// Build a `RunStatusResponse` reflecting the durable projection after an
/// archive/unarchive transition. The run is terminal in both directions, so no
/// live queue position or worker-only fields apply.
async fn archive_status_response(state: &AppState, id: RunId) -> Response {
let Ok(run_store) = state.store.open_run_reader(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let projection = match run_store.state().await {
Ok(projection) => projection,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let Some(status) = projection.status else {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"run has no status after archive/unarchive",
)
.into_response();
};
(
StatusCode::OK,
Json(RunStatusResponse {
id: id.to_string(),
status,
error: None,
queue_position: None,
pending_control: None,
created_at: id.created_at(),
}),
)
.into_response()
}
/// Persist a synchronous pause/unpause transition: append the caller-supplied
/// events to the run store and mirror the new status in the in-memory run map.
/// Returns `Some(Response)` on error, `None` on success.
async fn synchronous_transition(
state: &AppState,
id: RunId,
append_events: impl FnOnce(&mut Vec<workflow_event::Event>),
) -> Option<Response> {
let run_store = match state.store.open_run(&id).await {
Ok(run_store) => run_store,
Err(err) => {
return Some(
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
);
}
};
let mut events = Vec::new();
append_events(&mut events);
for event in events {
if let Err(err) = workflow_event::append_event(&run_store, &id, &event).await {
return Some(
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
);
}
let stored = workflow_event::to_run_event(&id, &event);
update_live_run_from_event(state, id, &stored);
}
None
}

View file

@ -0,0 +1,146 @@
use std::sync::Arc;
use axum::Router;
use axum::routing::{get, post};
use super::{ApiError, AppState, IntoResponse, Response, StatusCode, demo};
mod artifacts;
mod billing;
mod completions;
pub(in crate::server) mod events;
pub(in crate::server) mod graph;
mod lifecycle;
mod models;
mod pull_requests;
mod runs;
mod sandbox;
mod secrets;
pub(in crate::server) mod system;
pub(super) use system::{health, openapi_spec};
async fn not_implemented() -> Response {
ApiError::new(StatusCode::NOT_IMPLEMENTED, "Not implemented.").into_response()
}
pub(super) fn demo_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs", get(demo::list_runs).post(demo::create_run_stub))
.route("/runs/resolve", get(demo::resolve_run))
.route("/boards/runs", get(demo::list_board_runs))
.route("/attach", get(demo::attach_events_stub))
.route("/runs/{id}", get(demo::get_run_status))
.route("/runs/{id}/questions", get(demo::get_questions_stub))
.route("/runs/{id}/questions/{qid}/answer", post(demo::answer_stub))
.route("/runs/{id}/state", get(not_implemented))
.route("/runs/{id}/logs", get(not_implemented))
.route(
"/runs/{id}/events",
get(not_implemented).post(not_implemented),
)
.route("/runs/{id}/attach", get(demo::run_events_stub))
.route("/runs/{id}/blobs", post(not_implemented))
.route("/runs/{id}/blobs/{blobId}", get(not_implemented))
.route(
"/runs/{id}/stages/{stageId}/logs/{stream}",
get(not_implemented),
)
.route("/runs/{id}/checkpoint", get(demo::checkpoint_stub))
.route("/runs/{id}/cancel", post(demo::cancel_stub))
.route("/runs/{id}/start", post(demo::start_run_stub))
.route("/runs/{id}/pause", post(demo::pause_stub))
.route("/runs/{id}/unpause", post(demo::unpause_stub))
.route("/runs/{id}/graph", get(demo::get_run_graph))
.route("/runs/{id}/graph/source", get(demo::get_run_graph_source))
.route("/runs/{id}/stages", get(demo::get_run_stages))
.route("/runs/{id}/artifacts", get(demo::list_run_artifacts_stub))
.route("/runs/{id}/files", get(demo::list_run_files_stub))
.route(
"/runs/{id}/stages/{stageId}/turns",
get(demo::get_stage_turns),
)
.route(
"/runs/{id}/stages/{stageId}/artifacts",
get(not_implemented).post(not_implemented),
)
.route(
"/runs/{id}/stages/{stageId}/artifacts/download",
get(not_implemented),
)
.route("/runs/{id}/billing", get(demo::get_run_billing))
.route("/runs/{id}/settings", get(demo::get_run_settings))
.route("/runs/{id}/preview", post(demo::generate_preview_url_stub))
.route("/runs/{id}/ssh", post(demo::create_ssh_access_stub))
.route(
"/runs/{id}/sandbox/files",
get(demo::list_sandbox_files_stub),
)
.route(
"/runs/{id}/sandbox/file",
get(demo::get_sandbox_file_stub).put(demo::put_sandbox_file_stub),
)
.route(
"/insights/queries",
get(demo::list_saved_queries).post(demo::save_query_stub),
)
.route(
"/insights/queries/{id}",
get(demo::get_saved_query)
.put(demo::update_query_stub)
.delete(demo::delete_query_stub),
)
.route("/insights/execute", post(demo::execute_query_stub))
.route("/insights/history", get(demo::list_query_history))
.route(
"/secrets",
get(demo::list_secrets)
.post(demo::create_secret)
.delete(demo::delete_secret_by_name),
)
.route("/repos/github/{owner}/{name}", get(demo::get_github_repo))
.route("/health/diagnostics", post(demo::run_diagnostics))
.route("/settings", get(demo::get_server_settings))
.route("/system/info", get(demo::get_system_info))
.route("/system/df", get(demo::get_system_disk_usage))
.route("/system/prune/runs", post(demo::prune_runs))
.route("/billing", get(demo::get_aggregate_billing))
.merge(runs::manifest_routes())
.merge(graph::manifest_routes())
.merge(models::routes())
.merge(completions::routes())
}
pub(super) fn real_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs/{id}/stages/{stageId}/turns", get(not_implemented))
.route("/runs/{id}/steer", post(not_implemented))
.route("/workflows", get(not_implemented))
.route("/workflows/{name}", get(not_implemented))
.route("/workflows/{name}/runs", get(not_implemented))
.route(
"/insights/queries",
get(not_implemented).post(not_implemented),
)
.route(
"/insights/queries/{id}",
get(not_implemented)
.put(not_implemented)
.delete(not_implemented),
)
.route("/insights/execute", post(not_implemented))
.route("/insights/history", get(not_implemented))
.merge(runs::routes())
.merge(events::routes())
.merge(billing::routes())
.merge(pull_requests::routes())
.merge(artifacts::routes())
.merge(sandbox::routes())
.merge(lifecycle::routes())
.merge(graph::manifest_routes())
.merge(graph::run_routes())
.merge(models::routes())
.merge(secrets::routes())
.merge(system::routes())
.merge(completions::routes())
}

View file

@ -0,0 +1,154 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, FromStr, HashSet, IntoResponse, Json, MAX_PAGE_OFFSET, ModelTestMode, Path,
Provider, Query, RequiredUser, Response, Router, State, StatusCode, auth_issue_message,
default_page_limit, error, get, post, run_model_test,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/models", get(list_models))
.route("/models/{id}/test", post(test_model))
}
#[derive(serde::Deserialize)]
struct ModelListParams {
#[serde(rename = "page[limit]", default = "default_page_limit")]
limit: u32,
#[serde(rename = "page[offset]", default)]
offset: u32,
#[serde(default)]
provider: Option<String>,
#[serde(default)]
query: Option<String>,
}
#[derive(serde::Deserialize)]
struct ModelTestParams {
#[serde(default)]
mode: Option<String>,
}
async fn list_models(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Query(params): Query<ModelListParams>,
) -> Response {
let provider = match params.provider.as_deref() {
Some(value) => match Provider::from_str(value) {
Ok(provider) => Some(provider),
Err(_) => {
return ApiError::new(
StatusCode::BAD_REQUEST,
format!("unknown provider: {value}"),
)
.into_response();
}
},
None => None,
};
let query = params.query.as_ref().map(|value| value.to_lowercase());
let limit = params.limit.clamp(1, 100) as usize;
let offset = params.offset.min(MAX_PAGE_OFFSET) as usize;
let configured: HashSet<Provider> = state
.llm_source
.configured_providers()
.await
.into_iter()
.collect();
let mut models = fabro_model::Catalog::builtin()
.list(provider)
.into_iter()
.filter(|model| match &query {
Some(query) => {
model.id.to_lowercase().contains(query)
|| model.display_name.to_lowercase().contains(query)
|| model
.aliases
.iter()
.any(|alias| alias.to_lowercase().contains(query))
}
None => true,
})
.cloned()
.map(|mut model| {
model.configured = configured.contains(&model.provider);
model
})
.collect::<Vec<_>>();
let has_more = models.len() > offset.saturating_add(limit);
let data = models.drain(offset..models.len().min(offset.saturating_add(limit)));
(
StatusCode::OK,
Json(serde_json::json!({
"data": data.collect::<Vec<_>>(),
"meta": { "has_more": has_more }
})),
)
.into_response()
}
async fn test_model(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Query(params): Query<ModelTestParams>,
) -> Response {
let mode = match params.mode.as_deref() {
Some(value) => match ModelTestMode::from_str(value) {
Ok(mode) => mode,
Err(_) => {
return ApiError::new(
StatusCode::BAD_REQUEST,
format!("invalid model test mode: {value}"),
)
.into_response();
}
},
None => ModelTestMode::Basic,
};
let Some(info) = fabro_model::Catalog::builtin().get(&id) else {
return ApiError::not_found(format!("Model not found: {id}")).into_response();
};
let llm_result = match state.resolve_llm_client().await {
Ok(result) => result,
Err(err) => {
error!(error = ?err, "Failed to resolve LLM client");
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to resolve LLM client: {err}"),
)
.into_response();
}
};
if let Some((_, issue)) = llm_result
.auth_issues
.iter()
.find(|(provider, _)| *provider == info.provider)
{
return ApiError::bad_request(auth_issue_message(info.provider, issue)).into_response();
}
let provider_name = <&'static str>::from(info.provider);
if !llm_result.client.provider_names().contains(&provider_name) {
return Json(serde_json::json!({
"model_id": info.id,
"status": "skip",
}))
.into_response();
}
let client = Arc::new(llm_result.client);
let outcome = run_model_test(info, mode, client).await;
Json(serde_json::json!({
"model_id": info.id,
"status": <&'static str>::from(outcome.status),
"error_message": outcome.error_message,
}))
.into_response()
}

View file

@ -0,0 +1,399 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, Catalog, CloseRunPullRequestResponse, CreateRunPullRequestRequest,
IntoResponse, Json, MergeRunPullRequestRequest, MergeRunPullRequestResponse, PullRequestRecord,
RequireRunScoped, Response, Router, RunId, State, StatusCode, get, lock_pull_request_create,
post, pull_request, warn, workflow_event,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route(
"/runs/{id}/pull_request",
get(get_run_pull_request).post(create_run_pull_request),
)
.route(
"/runs/{id}/pull_request/merge",
post(merge_run_pull_request),
)
.route(
"/runs/{id}/pull_request/close",
post(close_run_pull_request),
)
}
#[expect(
clippy::disallowed_types,
reason = "Pull-request API validates public github.com URLs; these raw URLs are not credential-bearing log output."
)]
fn parse_github_owner_repo_from_url(url: &str, kind: &str) -> Result<(String, String), ApiError> {
let parsed = fabro_http::Url::parse(url)
.map_err(|err| ApiError::bad_request(format!("Invalid {kind}: {err}")))?;
match parsed.host_str() {
Some("github.com") => {}
Some(host) => {
return Err(ApiError::with_code(
StatusCode::BAD_REQUEST,
format!("Pull request operations support github.com only (got {host})."),
"unsupported_host",
));
}
None => {
return Err(ApiError::bad_request(format!(
"Invalid {kind}: missing host"
)));
}
}
fabro_github::parse_github_owner_repo(url).map_err(|err| ApiError::bad_request(err.to_string()))
}
fn load_server_github_credentials(
state: &AppState,
) -> Result<fabro_github::GitHubCredentials, ApiError> {
let settings = state.server_settings();
match state.github_credentials(&settings.server.integrations.github) {
Ok(Some(creds)) => Ok(creds),
Ok(None) => {
warn!("GitHub integration unavailable on server: credentials not configured");
Err(ApiError::with_code(
StatusCode::SERVICE_UNAVAILABLE,
"GitHub integration unavailable on server.",
"integration_unavailable",
))
}
Err(err) => {
warn!(error = %err, "GitHub integration unavailable on server");
Err(ApiError::with_code(
StatusCode::SERVICE_UNAVAILABLE,
"GitHub integration unavailable on server.",
"integration_unavailable",
))
}
}
}
fn server_github_context<'a>(
state: &'a AppState,
creds: &'a fabro_github::GitHubCredentials,
) -> Result<fabro_github::GitHubContext<'a>, ApiError> {
let http_client = state.http_client().map_err(|err| {
ApiError::with_code(
StatusCode::SERVICE_UNAVAILABLE,
format!("GitHub integration unavailable on server: {err}"),
"integration_unavailable",
)
})?;
Ok(fabro_github::GitHubContext::with_http_client(
creds,
state.github_api_base_url.as_str(),
http_client,
))
}
fn github_pull_request_not_found_error(record: &PullRequestRecord) -> ApiError {
ApiError::with_code(
StatusCode::BAD_GATEWAY,
format!("Pull request #{} was deleted on GitHub.", record.number),
"github_not_found",
)
}
struct PullRequestGithubContext {
record: PullRequestRecord,
creds: fabro_github::GitHubCredentials,
}
async fn load_pull_request_github_context(
state: &Arc<AppState>,
id: &RunId,
) -> Result<PullRequestGithubContext, ApiError> {
let run_store = state
.store
.open_run_reader(id)
.await
.map_err(|_| ApiError::not_found("Run not found."))?;
let run_state = run_store
.state()
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
let record = run_state.pull_request.ok_or_else(|| {
ApiError::with_code(
StatusCode::NOT_FOUND,
format!("No pull request found in store. Create one first with: fabro pr create {id}"),
"no_stored_record",
)
})?;
parse_github_owner_repo_from_url(&record.html_url, "pull request URL")?;
let creds = load_server_github_credentials(state.as_ref())?;
Ok(PullRequestGithubContext { record, creds })
}
struct RunPrInputs<'a> {
goal: &'a str,
base_branch: &'a str,
run_branch: &'a str,
diff: &'a str,
conclusion: &'a fabro_types::Conclusion,
normalized_origin: String,
}
impl<'a> RunPrInputs<'a> {
fn extract(run_state: &'a fabro_store::RunProjection, force: bool) -> Result<Self, ApiError> {
if let Some(record) = run_state.pull_request.as_ref() {
return Err(ApiError::with_code(
StatusCode::CONFLICT,
format!("Pull request already exists at {}", record.html_url),
"pull_request_exists",
));
}
let run_spec = run_state.spec.as_ref().ok_or_else(|| {
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"Run spec missing from store.",
)
})?;
let origin_url = run_spec.repo_origin_url().ok_or_else(|| {
ApiError::with_code(
StatusCode::BAD_REQUEST,
"Run has no repo origin URL — pull request creation requires git metadata.",
"missing_repo_origin",
)
})?;
let base_branch = run_spec.base_branch().ok_or_else(|| {
ApiError::with_code(
StatusCode::BAD_REQUEST,
"Run has no base branch — pull request creation requires git metadata.",
"missing_base_branch",
)
})?;
let run_branch = run_state
.start
.as_ref()
.and_then(|start| start.run_branch.as_deref())
.ok_or_else(|| {
ApiError::with_code(
StatusCode::BAD_REQUEST,
"Run has no run_branch — was it run with git push enabled?",
"missing_run_branch",
)
})?;
let diff = run_state
.final_patch
.as_deref()
.filter(|d| !d.trim().is_empty())
.ok_or_else(|| {
ApiError::with_code(
StatusCode::BAD_REQUEST,
"Stored diff is empty — nothing to create a PR for",
"empty_diff",
)
})?;
let conclusion = run_state.conclusion.as_ref().ok_or_else(|| {
ApiError::with_code(
StatusCode::BAD_REQUEST,
"Run is not finished yet.",
"run_not_finished",
)
})?;
if !force && !conclusion.status.is_successful() {
return Err(ApiError::with_code(
StatusCode::BAD_REQUEST,
format!(
"Run status is '{}', expected succeeded or partially_succeeded",
conclusion.status
),
"run_not_successful",
));
}
let normalized_origin = fabro_github::normalize_repo_origin_url(origin_url);
parse_github_owner_repo_from_url(&normalized_origin, "repo origin URL")?;
Ok(Self {
goal: run_spec.graph.goal(),
base_branch,
run_branch,
diff,
conclusion,
normalized_origin,
})
}
}
async fn create_run_pull_request(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
Json(body): Json<CreateRunPullRequestRequest>,
) -> Response {
let _create_guard = lock_pull_request_create(&state.pull_request_create_locks, &id).await;
let Ok(run_store) = state.store.open_run(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let run_state = match run_store.state().await {
Ok(run_state) => run_state,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let inputs = match RunPrInputs::extract(&run_state, body.force) {
Ok(inputs) => inputs,
Err(err) => return err.into_response(),
};
let creds = match load_server_github_credentials(state.as_ref()) {
Ok(creds) => creds,
Err(err) => return err.into_response(),
};
let github = match server_github_context(state.as_ref(), &creds) {
Ok(ctx) => ctx,
Err(err) => return err.into_response(),
};
let model = if let Some(model) = body.model {
model
} else {
let configured = state.llm_source.configured_providers().await;
Catalog::builtin()
.default_for_configured(&configured)
.id
.clone()
};
let run_store_handle = run_store.clone().into();
let request = pull_request::OpenPullRequestRequest {
github,
origin_url: &inputs.normalized_origin,
base_branch: inputs.base_branch,
head_branch: inputs.run_branch,
goal: inputs.goal,
diff: inputs.diff,
model: &model,
draft: true,
auto_merge: None,
run_store: &run_store_handle,
llm_source: state.llm_source.as_ref(),
conclusion: Some(inputs.conclusion),
run_state: Some(&run_state),
};
let pull_request = match pull_request::maybe_open_pull_request(request).await {
Ok(Some(record)) => record,
Ok(None) => {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"Pull request creation returned no record unexpectedly.",
)
.into_response();
}
Err(err) => return ApiError::new(StatusCode::BAD_GATEWAY, err).into_response(),
};
let event = workflow_event::Event::pull_request_created(&pull_request, true);
if let Err(err) = workflow_event::append_event(&run_store, &id, &event).await {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response();
}
Json(pull_request).into_response()
}
async fn get_run_pull_request(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
) -> Response {
let ctx = match load_pull_request_github_context(&state, &id).await {
Ok(ctx) => ctx,
Err(err) => return err.into_response(),
};
let github = match server_github_context(state.as_ref(), &ctx.creds) {
Ok(github) => github,
Err(err) => return err.into_response(),
};
match fabro_github::get_pull_request(
&github,
&ctx.record.owner,
&ctx.record.repo,
ctx.record.number,
)
.await
{
Ok(github) => Json(fabro_types::PullRequestDetail {
record: ctx.record,
github,
})
.into_response(),
Err(fabro_github::PullRequestApiError::NotFound { .. }) => {
github_pull_request_not_found_error(&ctx.record).into_response()
}
Err(err) => ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(),
}
}
async fn merge_run_pull_request(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
Json(body): Json<MergeRunPullRequestRequest>,
) -> Response {
let ctx = match load_pull_request_github_context(&state, &id).await {
Ok(ctx) => ctx,
Err(err) => return err.into_response(),
};
let github = match server_github_context(state.as_ref(), &ctx.creds) {
Ok(github) => github,
Err(err) => return err.into_response(),
};
match fabro_github::merge_pull_request(
&github,
&ctx.record.owner,
&ctx.record.repo,
ctx.record.number,
body.method,
)
.await
{
Ok(()) => Json(MergeRunPullRequestResponse {
number: i64::try_from(ctx.record.number)
.expect("stored pull request number should fit in i64"),
html_url: ctx.record.html_url,
method: body.method,
})
.into_response(),
Err(fabro_github::PullRequestApiError::NotFound { .. }) => {
github_pull_request_not_found_error(&ctx.record).into_response()
}
Err(err) => ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(),
}
}
async fn close_run_pull_request(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
) -> Response {
let ctx = match load_pull_request_github_context(&state, &id).await {
Ok(ctx) => ctx,
Err(err) => return err.into_response(),
};
let github = match server_github_context(state.as_ref(), &ctx.creds) {
Ok(github) => github,
Err(err) => return err.into_response(),
};
match fabro_github::close_pull_request(
&github,
&ctx.record.owner,
&ctx.record.repo,
ctx.record.number,
)
.await
{
Ok(()) => Json(CloseRunPullRequestResponse {
number: i64::try_from(ctx.record.number)
.expect("stored pull request number should fit in i64"),
html_url: ctx.record.html_url,
})
.into_response(),
Err(fabro_github::PullRequestApiError::NotFound { .. }) => {
github_pull_request_not_found_error(&ctx.record).into_response()
}
Err(err) => ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(),
}
}

View file

@ -0,0 +1,780 @@
use super::super::*;
pub(super) fn manifest_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/preflight", post(run_preflight))
.route("/validate", post(validate_run_manifest))
}
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs", get(list_runs).post(create_run))
.route("/runs/resolve", get(resolve_run))
.route("/boards/runs", get(list_board_runs))
.route("/runs/{id}", get(get_run_status).delete(delete_run))
.route("/runs/{id}/questions", get(get_questions))
.route("/runs/{id}/questions/{qid}/answer", post(submit_answer))
.route("/runs/{id}/state", get(get_run_state))
.route("/runs/{id}/logs", get(get_run_logs))
.route(
"/runs/{id}/stages/{stageId}/logs/{stream}",
get(get_run_stage_command_log),
)
.route("/runs/{id}/settings", get(get_run_settings))
.route("/runs/{id}/files", get(list_run_files))
.merge(manifest_routes())
}
#[derive(serde::Deserialize)]
struct ListRunsParams {
#[serde(rename = "page[limit]", default = "default_page_limit")]
limit: u32,
#[serde(rename = "page[offset]", default)]
offset: u32,
#[serde(default)]
include_archived: bool,
}
impl ListRunsParams {
fn pagination(&self) -> PaginationParams {
PaginationParams {
limit: self.limit,
offset: self.offset,
}
}
}
fn board_column(status: RunStatus) -> Option<&'static str> {
match status {
RunStatus::Submitted | RunStatus::Queued | RunStatus::Starting => Some("initializing"),
RunStatus::Running | RunStatus::Paused { .. } => Some("running"),
RunStatus::Blocked { .. } => Some("blocked"),
RunStatus::Succeeded { .. } => Some("succeeded"),
RunStatus::Failed { .. } | RunStatus::Dead => Some("failed"),
RunStatus::Removing | RunStatus::Archived { .. } => None,
}
}
pub(crate) fn board_columns() -> serde_json::Value {
serde_json::json!([
{"id": "initializing", "name": "Initializing"},
{"id": "running", "name": "Running"},
{"id": "blocked", "name": "Blocked"},
{"id": "succeeded", "name": "Succeeded"},
{"id": "failed", "name": "Failed"},
])
}
async fn board_run_metadata(
state: &AppState,
run_id: RunId,
) -> serde_json::Map<String, serde_json::Value> {
let mut metadata = serde_json::Map::new();
let Ok(run_store) = state.store.open_run_reader(&run_id).await else {
return metadata;
};
let Ok(run_state) = run_store.state().await else {
return metadata;
};
if let Some(pull_request) = run_state.pull_request {
metadata.insert(
"pull_request".to_string(),
serde_json::json!({
"number": pull_request.number,
}),
);
}
if let Some(sandbox) = run_state.sandbox {
let mut sandbox_metadata = serde_json::Map::new();
sandbox_metadata.insert(
"working_directory".to_string(),
serde_json::json!(sandbox.working_directory),
);
if let Some(identifier) = sandbox.identifier {
sandbox_metadata.insert("id".to_string(), serde_json::json!(identifier));
}
metadata.insert(
"sandbox".to_string(),
serde_json::Value::Object(sandbox_metadata),
);
}
if let Some((_, record)) =
run_state
.pending_interviews
.iter()
.min_by(|(left_id, left), (right_id, right)| {
left.started_at
.cmp(&right.started_at)
.then_with(|| left_id.cmp(right_id))
})
{
metadata.insert(
"question".to_string(),
serde_json::json!({
"text": record.question.text,
}),
);
}
metadata
}
fn paginate_items<T>(items: Vec<T>, pagination: &PaginationParams) -> (Vec<T>, bool) {
let limit = pagination.limit.clamp(1, 100) as usize;
let offset = pagination.offset.min(MAX_PAGE_OFFSET) as usize;
let mut data: Vec<_> = items.into_iter().skip(offset).take(limit + 1).collect();
let has_more = data.len() > limit;
data.truncate(limit);
(data, has_more)
}
async fn list_board_runs(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Query(pagination): Query<PaginationParams>,
) -> Response {
let summaries = match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(runs) => runs,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let board_summaries: Vec<_> = summaries
.into_iter()
.filter_map(|summary| {
let column = board_column(summary.status)?;
Some((summary, column))
})
.collect();
let (page_summaries, has_more) = paginate_items(board_summaries, &pagination);
let mut data = Vec::with_capacity(page_summaries.len());
for (summary, column) in page_summaries {
let run_id = summary.run_id;
let mut item =
serde_json::to_value(&summary).expect("RunSummary serialization is infallible");
item["column"] = serde_json::json!(column);
if let Some(object) = item.as_object_mut() {
object.extend(board_run_metadata(state.as_ref(), run_id).await);
}
data.push(item);
}
(
StatusCode::OK,
Json(serde_json::json!({
"columns": board_columns(),
"data": data,
"meta": { "has_more": has_more }
})),
)
.into_response()
}
async fn list_runs(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Query(params): Query<ListRunsParams>,
) -> Response {
match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(runs) => {
let include_archived = params.include_archived;
let items = runs
.into_iter()
.filter(|summary| {
include_archived || !matches!(summary.status, RunStatus::Archived { .. })
})
.collect::<Vec<_>>();
let (data, has_more) = paginate_items(items, &params.pagination());
(
StatusCode::OK,
Json(serde_json::json!({
"data": data,
"meta": { "has_more": has_more }
})),
)
.into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
#[derive(Debug, serde::Deserialize)]
struct ResolveRunQuery {
selector: String,
}
#[derive(Debug, Default, serde::Deserialize)]
struct DeleteRunQuery {
#[serde(default)]
force: bool,
}
fn default_command_log_limit() -> u64 {
65_536
}
#[derive(Debug, serde::Deserialize)]
struct CommandLogQuery {
#[serde(default)]
offset: u64,
#[serde(default = "default_command_log_limit")]
limit: u64,
}
#[derive(Debug, serde::Serialize)]
struct CommandLogResponseBody {
stream: CommandOutputStream,
offset: u64,
next_offset: u64,
total_bytes: u64,
bytes_base64: String,
eof: bool,
cas_ref: Option<String>,
live_streaming: bool,
}
async fn resolve_run(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Query(query): Query<ResolveRunQuery>,
) -> Response {
let runs = match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(runs) => runs,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
match resolve_run_by_selector(
&runs,
&query.selector,
|run| run.run_id.to_string(),
|run| run.workflow_slug.clone(),
|run| run.workflow_name.clone(),
|run| run.run_id.created_at(),
|run| run.run_id.created_at().to_rfc3339(),
|run| run.repo_origin_url.clone(),
) {
Ok(run) => (StatusCode::OK, Json(run.clone())).into_response(),
Err(err @ (ResolveRunError::InvalidSelector | ResolveRunError::AmbiguousPrefix { .. })) => {
ApiError::bad_request(err.to_string()).into_response()
}
Err(err @ ResolveRunError::NotFound { .. }) => {
ApiError::not_found(err.to_string()).into_response()
}
}
}
async fn delete_run(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Query(query): Query<DeleteRunQuery>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
match delete_run_internal(&state, id, query.force).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(response) => response,
}
}
async fn create_run(
RequestAuth(auth_slot): RequestAuth,
State(state): State<Arc<AppState>>,
headers: HeaderMap,
body: Bytes,
) -> Response {
let subject = match require_user(&auth_slot) {
Ok(subject) => subject,
Err(err) => return err.into_response(),
};
let req = match serde_json::from_slice::<RunManifest>(&body) {
Ok(req) => req,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
let manifest_run_defaults = state.manifest_run_defaults();
let prepared = match run_manifest::prepare_manifest(manifest_run_defaults.as_ref(), &req) {
Ok(prepared) => prepared,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
let run_id = prepared.run_id.unwrap_or_else(RunId::new);
info!(run_id = %run_id, "Run created");
let configured_providers = state.llm_source.configured_providers().await;
let mut create_input = run_manifest::create_run_input(prepared.clone(), configured_providers);
create_input.run_id = Some(run_id);
create_input.provenance = Some(run_provenance(&headers, &subject));
create_input.submitted_manifest_bytes = Some(body.to_vec());
let storage_root = match resolve_interp_string(&state.server_settings().server.storage.root) {
Ok(path) => PathBuf::from(path),
Err(err) => {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to resolve server storage root: {err}"),
)
.into_response();
}
};
let created = match Box::pin(operations::create(
state.store.as_ref(),
create_input,
storage_root,
))
.await
{
Ok(created) => created,
Err(WorkflowError::ValidationFailed { .. } | WorkflowError::Parse(_)) => {
return ApiError::bad_request("Validation failed").into_response();
}
Err(err) => {
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to persist run state: {err}"),
)
.into_response();
}
};
let created_at = created.run_id.created_at();
{
let mut runs = state.runs.lock().expect("runs lock poisoned");
runs.insert(
created.run_id,
managed_run(
created.persisted.source().to_string(),
RunStatus::Submitted,
created_at,
created.run_dir,
RunExecutionMode::Start,
),
);
}
(
StatusCode::CREATED,
Json(RunStatusResponse {
id: run_id.to_string(),
status: RunStatus::Submitted,
error: None,
queue_position: None,
pending_control: None,
created_at,
}),
)
.into_response()
}
fn run_provenance(headers: &HeaderMap, subject: &UserPrincipal) -> RunProvenance {
RunProvenance {
server: Some(RunServerProvenance {
version: FABRO_VERSION.to_string(),
}),
client: run_client_provenance(headers),
subject: Some(Principal::User(subject.clone())),
}
}
fn run_client_provenance(headers: &HeaderMap) -> Option<RunClientProvenance> {
let user_agent = headers
.get(header::USER_AGENT)
.and_then(|value| value.to_str().ok())
.map(str::to_string)?;
let (name, version) = parse_known_fabro_user_agent(&user_agent)
.map_or((None, None), |(name, version)| {
(Some(name.to_string()), Some(version.to_string()))
});
Some(RunClientProvenance {
user_agent: Some(user_agent),
name,
version,
})
}
fn parse_known_fabro_user_agent(user_agent: &str) -> Option<(&str, &str)> {
let token = user_agent.split_whitespace().next()?;
let (name, version) = token.split_once('/')?;
if version.is_empty() {
return None;
}
match name {
"fabro-cli" | "fabro-web" => Some((name, version)),
_ => None,
}
}
async fn run_preflight(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Json(req): Json<RunManifest>,
) -> Response {
let manifest_run_defaults = state.manifest_run_defaults();
let prepared = match run_manifest::prepare_manifest(manifest_run_defaults.as_ref(), &req) {
Ok(prepared) => prepared,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
let validated = match run_manifest::validate_prepared_manifest(&prepared) {
Ok(validated) => validated,
Err(WorkflowError::Parse(_)) => {
return ApiError::bad_request("Validation failed").into_response();
}
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
let response = match run_manifest::run_preflight(&state, &prepared, &validated).await {
Ok((response, _ok)) => response,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
(StatusCode::OK, Json(response)).into_response()
}
async fn validate_run_manifest(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Json(req): Json<RunManifest>,
) -> Response {
let manifest_run_defaults = state.manifest_run_defaults();
let prepared = match run_manifest::prepare_manifest(manifest_run_defaults.as_ref(), &req) {
Ok(prepared) => prepared,
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
let validated = match run_manifest::validate_prepared_manifest(&prepared) {
Ok(validated) => validated,
Err(WorkflowError::Parse(_)) => {
return ApiError::bad_request("Validation failed").into_response();
}
Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
};
(
StatusCode::OK,
Json(run_manifest::validate_response(&prepared, &validated)),
)
.into_response()
}
async fn get_run_status(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(runs) => match runs.into_iter().find(|run| run.run_id == id) {
Some(run) => (StatusCode::OK, Json(run)).into_response(),
None => ApiError::not_found("Run not found.").into_response(),
},
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
async fn get_run_settings(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let run_store = match state.store.open_run_reader(&id).await {
Ok(store) => store,
Err(fabro_store::Error::RunNotFound(_)) => {
return ApiError::not_found("Run not found.").into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let run_state = match run_store.state().await {
Ok(state) => state,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let Some(run_spec) = run_state.spec else {
return ApiError::not_found("Run not found.").into_response();
};
(StatusCode::OK, Json(run_spec.settings)).into_response()
}
async fn get_questions(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => {
let questions = run_state
.pending_interviews
.values()
.map(api_question_from_pending_interview)
.collect::<Vec<_>>();
(StatusCode::OK, Json(ListResponse::new(questions))).into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(fabro_store::Error::RunNotFound(_)) => {
ApiError::not_found("Run not found.").into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
async fn submit_answer(
auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path((id, qid)): Path<(String, String)>,
Json(req): Json<SubmitAnswerRequest>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let pending = match load_pending_interview(state.as_ref(), id, &qid).await {
Ok(pending) => pending,
Err(response) => return response,
};
let answer = match answer_from_request(req, &pending.question) {
Ok(answer) => answer,
Err(response) => return response,
};
let submission = AnswerSubmission::new(answer, Principal::User(auth.0));
match submit_pending_interview_answer(state.as_ref(), &pending, submission).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(response) => response,
}
}
async fn get_run_state(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
) -> Response {
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => Json(run_state).into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
}
async fn get_run_logs(
RequireRunScoped(id): RequireRunScoped,
State(state): State<Arc<AppState>>,
) -> Response {
if state.store.open_run_reader(&id).await.is_err() {
return ApiError::not_found("Run not found.").into_response();
}
let path = Storage::new(state.server_storage_dir())
.run_scratch(&id)
.runtime_dir()
.join("server.log");
match fs::read(&path).await {
Ok(bytes) => ([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], bytes).into_response(),
Err(err) if err.kind() == ErrorKind::NotFound => {
ApiError::not_found("Run log not available.").into_response()
}
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
async fn get_run_stage_command_log(
RequireCommandLog(id, stage_id, stream): RequireCommandLog,
State(state): State<Arc<AppState>>,
Query(query): Query<CommandLogQuery>,
) -> Response {
const MAX_COMMAND_LOG_LIMIT: u64 = 1_048_576;
if query.limit == 0 {
return ApiError::bad_request("limit must be greater than 0").into_response();
}
let limit = query.limit.min(MAX_COMMAND_LOG_LIMIT);
let Ok(run_store) = state.store.open_run_reader(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let run_state = match run_store.state().await {
Ok(run_state) => run_state,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let Some(node) = run_state.stage(&stage_id) else {
return ApiError::not_found("Stage not found.").into_response();
};
let stream_value = match stream {
CommandOutputStream::Stdout => node.stdout.as_deref(),
CommandOutputStream::Stderr => node.stderr.as_deref(),
};
let cas_ref = stream_value
.filter(|value| parse_blob_ref(value).is_some())
.map(str::to_string);
let live_streaming = node
.live_streaming
.unwrap_or_else(|| cas_ref.is_none() && node.completion.is_none());
let run_dir = Storage::new(state.server_storage_dir())
.run_scratch(&id)
.root()
.to_path_buf();
let scratch_path = command_log_path(&run_dir, &stage_id, stream);
match read_log_slice(&scratch_path, query.offset, limit).await {
Ok((bytes, total_bytes)) => {
return build_command_log_response(
stream,
query.offset,
limit,
LogSource::Sliced { bytes, total_bytes },
cas_ref.is_some(),
cas_ref,
live_streaming,
);
}
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
if let Some(cas_ref) = cas_ref {
let text = match read_json_string_blob(&run_store.clone().into(), &cas_ref).await {
Ok(Some(text)) => text,
Ok(None) => String::new(),
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
return build_command_log_response(
stream,
query.offset,
limit,
LogSource::Full(text.as_bytes()),
true,
Some(cas_ref),
live_streaming,
);
}
if let Some(inline_text) = stream_value {
return build_command_log_response(
stream,
query.offset,
limit,
LogSource::Full(inline_text.as_bytes()),
true,
None,
live_streaming,
);
}
build_command_log_response(
stream,
query.offset,
limit,
LogSource::Full(&[]),
node.completion.is_some(),
None,
live_streaming,
)
}
enum LogSource<'a> {
Sliced {
bytes: Vec<u8>,
total_bytes: u64,
},
Full(&'a [u8]),
}
fn build_command_log_response(
stream: CommandOutputStream,
requested_offset: u64,
limit: u64,
source: LogSource<'_>,
eof: bool,
cas_ref: Option<String>,
live_streaming: bool,
) -> Response {
let (body_bytes, total_bytes, offset) = match source {
LogSource::Sliced { bytes, total_bytes } => {
let offset = requested_offset.min(total_bytes);
(bytes, total_bytes, offset)
}
LogSource::Full(bytes) => {
let total_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
let offset = requested_offset.min(total_bytes);
let start = usize::try_from(offset).unwrap_or(bytes.len());
let end = start
.saturating_add(usize::try_from(limit).unwrap_or(usize::MAX))
.min(bytes.len());
(bytes[start..end].to_vec(), total_bytes, offset)
}
};
Json(CommandLogResponseBody {
stream,
offset,
next_offset: offset + u64::try_from(body_bytes.len()).unwrap_or(u64::MAX),
total_bytes,
bytes_base64: BASE64_STANDARD.encode(body_bytes),
eof,
cas_ref,
live_streaming,
})
.into_response()
}

View file

@ -0,0 +1,274 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, Bytes, DaytonaSandbox, EnvVars, IntoResponse, Json, NamedTempFile, Path,
PreviewUrlRequest, PreviewUrlResponse, Query, RequiredUser, Response, Router, RunId, Sandbox,
SandboxFileEntry, SandboxFileListResponse, SandboxProvider, SshAccessRequest,
SshAccessResponse, State, StatusCode, collect_causes, fs, get, octet_stream_response,
parse_run_id_path, post, reconnect, reject_if_archived, render_with_causes,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs/{id}/preview", post(generate_preview_url))
.route("/runs/{id}/ssh", post(create_ssh_access))
.route("/runs/{id}/sandbox/files", get(list_sandbox_files))
.route(
"/runs/{id}/sandbox/file",
get(get_sandbox_file).put(put_sandbox_file),
)
}
#[derive(serde::Deserialize)]
struct SandboxFilesParams {
path: String,
#[serde(default)]
depth: Option<usize>,
}
#[derive(serde::Deserialize)]
struct SandboxFileParams {
path: String,
}
async fn generate_preview_url(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Json(request): Json<PreviewUrlRequest>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let Ok(port) = u16::try_from(request.port) else {
return ApiError::bad_request("Port must fit in a u16.").into_response();
};
let Ok(expires_in_secs) = i32::try_from(request.expires_in_secs.get()) else {
return ApiError::bad_request("Preview expiry exceeds supported range.").into_response();
};
let sandbox = match reconnect_daytona_sandbox(&state, &id).await {
Ok(sandbox) => sandbox,
Err(response) => return response,
};
let response = if request.signed {
match sandbox
.get_signed_preview_url(port, Some(expires_in_secs))
.await
{
Ok(preview) => PreviewUrlResponse {
token: None,
url: preview.url,
},
Err(err) => {
return ApiError::new(StatusCode::CONFLICT, err.display_with_causes())
.into_response();
}
}
} else {
match sandbox.get_preview_link(port).await {
Ok(preview) => PreviewUrlResponse {
token: Some(preview.token),
url: preview.url,
},
Err(err) => {
return ApiError::new(StatusCode::CONFLICT, err.display_with_causes())
.into_response();
}
}
};
(StatusCode::CREATED, Json(response)).into_response()
}
async fn create_ssh_access(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Json(request): Json<SshAccessRequest>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let sandbox = match reconnect_daytona_sandbox(&state, &id).await {
Ok(sandbox) => sandbox,
Err(response) => return response,
};
match sandbox.create_ssh_access(Some(request.ttl_minutes)).await {
Ok(command) => (StatusCode::CREATED, Json(SshAccessResponse { command })).into_response(),
Err(err) => ApiError::new(StatusCode::CONFLICT, err.display_with_causes()).into_response(),
}
}
async fn list_sandbox_files(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Query(params): Query<SandboxFilesParams>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let sandbox = match reconnect_run_sandbox(&state, &id).await {
Ok(sandbox) => sandbox,
Err(response) => return response,
};
match sandbox.list_directory(&params.path, params.depth).await {
Ok(entries) => Json(SandboxFileListResponse {
data: entries
.into_iter()
.map(|entry| SandboxFileEntry {
is_dir: entry.is_dir,
name: entry.name,
size: entry.size.map(u64::cast_signed),
})
.collect(),
})
.into_response(),
Err(err) => ApiError::new(StatusCode::NOT_FOUND, err.display_with_causes()).into_response(),
}
}
async fn get_sandbox_file(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Query(params): Query<SandboxFileParams>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let sandbox = match reconnect_run_sandbox(&state, &id).await {
Ok(sandbox) => sandbox,
Err(response) => return response,
};
let temp = match NamedTempFile::new() {
Ok(temp) => temp,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
if let Err(err) = sandbox
.download_file_to_local(&params.path, temp.path())
.await
{
return ApiError::new(StatusCode::NOT_FOUND, err.display_with_causes()).into_response();
}
match fs::read(temp.path()).await {
Ok(bytes) => octet_stream_response(bytes.into()),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
async fn put_sandbox_file(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
Query(params): Query<SandboxFileParams>,
body: Bytes,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
if let Some(response) = reject_if_archived(state.as_ref(), &id).await {
return response;
}
let sandbox = match reconnect_run_sandbox(&state, &id).await {
Ok(sandbox) => sandbox,
Err(response) => return response,
};
let temp = match NamedTempFile::new() {
Ok(temp) => temp,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
if let Err(err) = fs::write(temp.path(), &body).await {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response();
}
match sandbox
.upload_file_from_local(temp.path(), &params.path)
.await
{
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.display_with_causes())
.into_response(),
}
}
async fn reconnect_run_sandbox(
state: &Arc<AppState>,
run_id: &RunId,
) -> Result<Box<dyn Sandbox>, Response> {
let record = load_run_sandbox_record(state, run_id).await?;
let daytona_api_key = state.vault_or_env(EnvVars::DAYTONA_API_KEY);
reconnect(&record, daytona_api_key).await.map_err(|err| {
let detail = render_with_causes(&err.to_string(), &collect_causes(err.as_ref()));
ApiError::new(StatusCode::CONFLICT, detail).into_response()
})
}
async fn reconnect_daytona_sandbox(
state: &Arc<AppState>,
run_id: &RunId,
) -> Result<DaytonaSandbox, Response> {
let record = load_run_sandbox_record(state, run_id).await?;
if record.provider != SandboxProvider::Daytona.to_string() {
return Err(ApiError::new(
StatusCode::CONFLICT,
"Sandbox provider does not support this capability.",
)
.into_response());
}
let Some(name) = record.identifier.as_deref() else {
return Err(ApiError::new(
StatusCode::CONFLICT,
"Sandbox record is missing the Daytona identifier.",
)
.into_response());
};
let Some(repo_cloned) = record.repo_cloned else {
return Err(ApiError::new(
StatusCode::CONFLICT,
"Sandbox record is missing clone metadata.",
)
.into_response());
};
let daytona_api_key = state.vault_or_env(EnvVars::DAYTONA_API_KEY);
DaytonaSandbox::reconnect(
name,
daytona_api_key,
repo_cloned,
record.clone_origin_url.clone(),
record.clone_branch.clone(),
)
.await
.map_err(|err| ApiError::new(StatusCode::CONFLICT, err.display_with_causes()).into_response())
}
async fn load_run_sandbox_record(
state: &Arc<AppState>,
run_id: &RunId,
) -> Result<fabro_types::SandboxRecord, Response> {
match state.store.open_run_reader(run_id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => run_state.sandbox.ok_or_else(|| {
ApiError::new(StatusCode::CONFLICT, "Run has no active sandbox.").into_response()
}),
Err(err) => Err(
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
),
},
Err(_) => Err(ApiError::not_found("Run not found.").into_response()),
}
}

View file

@ -0,0 +1,102 @@
use std::sync::Arc;
use super::super::{
ApiError, AppState, CreateSecretRequest, DeleteSecretRequest, IntoResponse, Json, RequiredUser,
Response, Router, SecretType, State, StatusCode, VaultError, get, parse_credential_secret,
spawn_blocking,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new().route(
"/secrets",
get(list_secrets)
.post(create_secret)
.delete(delete_secret_by_name),
)
}
async fn list_secrets(_auth: RequiredUser, State(state): State<Arc<AppState>>) -> Response {
let data = state.vault.read().await.list();
(StatusCode::OK, Json(serde_json::json!({ "data": data }))).into_response()
}
async fn create_secret(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Json(body): Json<CreateSecretRequest>,
) -> Response {
let secret_type = body.type_;
let name = body.name;
let value = body.value;
let description = body.description;
if secret_type == SecretType::Credential {
if let Err(err) = parse_credential_secret(&name, &value) {
return ApiError::bad_request(err).into_response();
}
}
let state_for_write = Arc::clone(&state);
let result = spawn_blocking(move || {
let mut vault = state_for_write.vault.blocking_write();
vault.set(&name, &value, secret_type, description.as_deref())
})
.await;
match result {
Ok(Ok(meta)) => (StatusCode::OK, Json(meta)).into_response(),
Ok(Err(VaultError::InvalidName(_))) => {
ApiError::bad_request("invalid secret name").into_response()
}
Ok(Err(VaultError::Io(err))) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
Ok(Err(VaultError::Serde(err))) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
Ok(Err(VaultError::NotFound(_))) => ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"secret unexpectedly missing",
)
.into_response(),
Err(err) => ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("secret write task failed: {err}"),
)
.into_response(),
}
}
async fn delete_secret_by_name(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Json(body): Json<DeleteSecretRequest>,
) -> Response {
let name = body.name;
let state_for_write = Arc::clone(&state);
let result = spawn_blocking(move || {
let mut vault = state_for_write.vault.blocking_write();
vault.remove(&name)
})
.await;
match result {
Ok(Ok(())) => StatusCode::NO_CONTENT.into_response(),
Ok(Err(VaultError::InvalidName(_))) => {
ApiError::bad_request("invalid secret name").into_response()
}
Ok(Err(VaultError::NotFound(name))) => {
ApiError::new(StatusCode::NOT_FOUND, format!("secret not found: {name}"))
.into_response()
}
Ok(Err(VaultError::Io(err))) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
Ok(Err(VaultError::Serde(err))) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
Err(err) => ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("secret delete task failed: {err}"),
)
.into_response(),
}
}

View file

@ -0,0 +1,506 @@
use std::sync::Arc;
use super::super::{
AggregateBilling, AggregateBillingTotals, ApiError, AppState, BilledTokenCounts,
BillingByModel, DfParams, FABRO_VERSION, GithubIntegrationStrategy, IntoResponse, Json,
ModelReference, Path, PruneRunsRequest, PruneRunsResponse, Query, RequiredUser, Response,
Router, RunStatus, State, StatusCode, SystemInfoResponse, SystemRunCounts,
build_disk_usage_response, build_prune_plan, delete_run_internal, diagnostics, get, post,
resolve_interp_string, spawn_blocking, system_features, system_sandbox_provider, to_i64,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/repos/github/{owner}/{name}", get(get_github_repo))
.route("/health/diagnostics", post(run_diagnostics))
.route("/settings", get(get_server_settings))
.route("/system/info", get(get_system_info))
.route("/system/df", get(get_system_df))
.route("/system/prune/runs", post(prune_runs))
.route("/billing", get(get_aggregate_billing))
}
pub(in crate::server) async fn health() -> Response {
Json(serde_json::json!({
"status": "ok",
}))
.into_response()
}
async fn get_server_settings(_auth: RequiredUser, State(state): State<Arc<AppState>>) -> Response {
(
StatusCode::OK,
Json(state.server_settings().as_ref().clone()),
)
.into_response()
}
async fn get_system_info(_auth: RequiredUser, State(state): State<Arc<AppState>>) -> Response {
let manifest_run_settings = state.manifest_run_settings();
let server_settings = state.server_settings();
let (total_runs, active_runs) = {
let runs = state.runs.lock().expect("runs lock poisoned");
let active = runs
.values()
.filter(|run| {
matches!(
run.status,
RunStatus::Queued
| RunStatus::Starting
| RunStatus::Running
| RunStatus::Blocked { .. }
| RunStatus::Paused { .. }
)
})
.count();
(runs.len(), active)
};
let response = SystemInfoResponse {
version: Some(FABRO_VERSION.to_string()),
server_url: Some(server_settings.server.web.url.as_source()),
git_sha: option_env!("FABRO_GIT_SHA").map(str::to_string),
build_date: option_env!("FABRO_BUILD_DATE").map(str::to_string),
profile: option_env!("FABRO_BUILD_PROFILE").map(str::to_string),
os: Some(std::env::consts::OS.to_string()),
arch: Some(std::env::consts::ARCH.to_string()),
storage_engine: Some("slatedb".to_string()),
storage_dir: Some(state.server_storage_dir().display().to_string()),
uptime_secs: Some(to_i64(state.started_at.elapsed().as_secs())),
runs: Some(SystemRunCounts {
total: Some(to_i64(total_runs)),
active: Some(to_i64(active_runs)),
}),
sandbox_provider: Some(system_sandbox_provider(&manifest_run_settings)),
features: Some(system_features(
server_settings.as_ref(),
&manifest_run_settings,
)),
};
(StatusCode::OK, Json(response)).into_response()
}
async fn get_system_df(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Query(params): Query<DfParams>,
) -> Response {
let storage_dir = state.server_storage_dir();
let summaries = match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(summaries) => summaries,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let response = match spawn_blocking(move || {
build_disk_usage_response(&summaries, &storage_dir, params.verbose)
})
.await
{
Ok(Ok(response)) => response,
Ok(Err(err)) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
(StatusCode::OK, Json(response)).into_response()
}
async fn prune_runs(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Json(body): Json<PruneRunsRequest>,
) -> Response {
let storage_dir = state.server_storage_dir();
let summaries = match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(summaries) => summaries,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let dry_run = body.dry_run;
let body_for_plan = body.clone();
let prune_plan =
match spawn_blocking(move || build_prune_plan(&body_for_plan, &summaries, &storage_dir))
.await
{
Ok(Ok(plan)) => plan,
Ok(Err(err)) => {
return ApiError::new(StatusCode::BAD_REQUEST, err.to_string()).into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
if dry_run {
return (
StatusCode::OK,
Json(PruneRunsResponse {
dry_run: Some(true),
runs: Some(prune_plan.rows),
total_count: Some(to_i64(prune_plan.run_ids.len())),
total_size_bytes: Some(to_i64(prune_plan.total_size_bytes)),
deleted_count: Some(0),
freed_bytes: Some(0),
}),
)
.into_response();
}
for run_id in &prune_plan.run_ids {
if let Err(response) = delete_run_internal(&state, *run_id, true).await {
return response;
}
}
(
StatusCode::OK,
Json(PruneRunsResponse {
dry_run: Some(false),
runs: None,
total_count: Some(to_i64(prune_plan.run_ids.len())),
total_size_bytes: Some(to_i64(prune_plan.total_size_bytes)),
deleted_count: Some(to_i64(prune_plan.run_ids.len())),
freed_bytes: Some(to_i64(prune_plan.total_size_bytes)),
}),
)
.into_response()
}
#[derive(serde::Deserialize)]
struct GitHubRepoResponse {
default_branch: String,
private: bool,
permissions: Option<serde_json::Value>,
}
/// Reject owner/repo path segments that could rewrite the GitHub API endpoint
/// via `..` traversal after URL normalization. Conservative compared to
/// GitHub's real rules, which is fine for server-side input validation.
#[allow(
clippy::result_large_err,
reason = "GitHub slug validation returns HTTP 400 responses directly."
)]
pub(in crate::server) fn validate_github_slug(
kind: &str,
value: &str,
max_len: usize,
) -> Result<(), Response> {
if value.is_empty() || value.len() > max_len || matches!(value, "." | "..") {
return Err(ApiError::bad_request(format!("invalid github {kind}")).into_response());
}
if !value
.bytes()
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.'))
{
return Err(ApiError::bad_request(format!("invalid github {kind}")).into_response());
}
Ok(())
}
async fn get_github_repo(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path((owner, name)): Path<(String, String)>,
) -> Response {
if let Err(response) = validate_github_slug("owner", &owner, 39) {
return response;
}
if let Err(response) = validate_github_slug("repo", &name, 100) {
return response;
}
let settings = state.server_settings();
let github_settings = &settings.server.integrations.github;
let base_url = fabro_github::github_api_base_url();
let mut client: Option<fabro_http::HttpClient> = None;
let token = match github_settings.strategy {
GithubIntegrationStrategy::App => {
let Some(app_id) = github_settings.app_id.as_ref() else {
return ApiError::new(
StatusCode::SERVICE_UNAVAILABLE,
"server.integrations.github.app_id is not configured",
)
.into_response();
};
if let Err(err) = resolve_interp_string(app_id) {
return ApiError::new(StatusCode::SERVICE_UNAVAILABLE, err.to_string())
.into_response();
}
let creds = match state.github_credentials(github_settings) {
Ok(Some(fabro_github::GitHubCredentials::App(creds))) => creds,
Ok(Some(_)) => unreachable!("app strategy should not return token credentials"),
Ok(None) => {
return ApiError::new(
StatusCode::SERVICE_UNAVAILABLE,
"GITHUB_APP_PRIVATE_KEY is not configured",
)
.into_response();
}
Err(err) => {
return ApiError::new(StatusCode::SERVICE_UNAVAILABLE, err).into_response();
}
};
let jwt = match fabro_github::sign_app_jwt(&creds.app_id, &creds.private_key_pem) {
Ok(jwt) => jwt,
Err(err) => {
tracing::error!(error = ?err, "failed to sign GitHub App JWT");
return ApiError::new(StatusCode::SERVICE_UNAVAILABLE, err.to_string())
.into_response();
}
};
let install_url = match github_settings.slug.as_ref() {
Some(slug) => match resolve_interp_string(slug) {
Ok(slug) => format!("https://github.com/apps/{slug}/installations/new"),
Err(err) => {
return ApiError::new(StatusCode::SERVICE_UNAVAILABLE, err.to_string())
.into_response();
}
},
None => format!("https://github.com/organizations/{owner}/settings/installations"),
};
if client.is_none() {
client = Some(match state.http_client() {
Ok(http) => http,
Err(err) => {
return ApiError::new(StatusCode::SERVICE_UNAVAILABLE, err.to_string())
.into_response();
}
});
}
let client_ref = client.as_ref().expect("client initialized above");
let installed =
match fabro_github::check_app_installed(client_ref, &jwt, &owner, &name, &base_url)
.await
{
Ok(installed) => installed,
Err(err) => {
tracing::error!(error = ?err, "failed to check GitHub App installation");
return ApiError::new(StatusCode::BAD_GATEWAY, err.to_string())
.into_response();
}
};
if !installed {
return (
StatusCode::OK,
Json(serde_json::json!({
"owner": owner,
"name": name,
"accessible": false,
"default_branch": null,
"private": null,
"permissions": null,
"install_url": install_url,
})),
)
.into_response();
}
match fabro_github::create_installation_access_token_with_permissions_and_install_url(
client_ref,
&jwt,
&owner,
&name,
&base_url,
serde_json::json!({ "contents": "write", "pull_requests": "write" }),
Some(&install_url),
)
.await
{
Ok(token) => token,
Err(err) => {
tracing::error!(
error = ?err,
"failed to create GitHub App installation token"
);
return ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response();
}
}
}
GithubIntegrationStrategy::Token => match state.github_credentials(github_settings) {
Ok(Some(fabro_github::GitHubCredentials::Token(token))) => token,
Ok(Some(_)) => unreachable!("token strategy should not return app credentials"),
Ok(None) => {
return ApiError::new(
StatusCode::SERVICE_UNAVAILABLE,
"GITHUB_TOKEN is not configured",
)
.into_response();
}
Err(err) => {
return ApiError::new(StatusCode::SERVICE_UNAVAILABLE, err).into_response();
}
},
};
let client = match client {
Some(client) => client,
None => match state.http_client() {
Ok(http) => http,
Err(err) => {
return ApiError::new(StatusCode::SERVICE_UNAVAILABLE, err.to_string())
.into_response();
}
},
};
let repo_response = match client
.get(format!("{base_url}/repos/{owner}/{name}"))
.header("Authorization", format!("Bearer {token}"))
.header("Accept", "application/vnd.github+json")
.header("User-Agent", "fabro-server")
.send()
.await
{
Ok(response) if response.status().is_success() => response,
Ok(response)
if github_settings.strategy == GithubIntegrationStrategy::Token
&& matches!(
response.status(),
fabro_http::StatusCode::FORBIDDEN | fabro_http::StatusCode::NOT_FOUND
) =>
{
return (
StatusCode::OK,
Json(serde_json::json!({
"owner": owner,
"name": name,
"accessible": false,
"default_branch": null,
"private": null,
"permissions": null,
"install_url": serde_json::Value::Null,
})),
)
.into_response();
}
Ok(response)
if github_settings.strategy == GithubIntegrationStrategy::Token
&& response.status() == fabro_http::StatusCode::UNAUTHORIZED =>
{
return ApiError::new(
StatusCode::SERVICE_UNAVAILABLE,
"Stored GitHub token is invalid — run fabro install or update GITHUB_TOKEN",
)
.into_response();
}
Ok(response) => {
return ApiError::new(
StatusCode::BAD_GATEWAY,
format!("GitHub repo lookup failed: {}", response.status()),
)
.into_response();
}
Err(err) => return ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(),
};
let repo = match repo_response.json::<GitHubRepoResponse>().await {
Ok(repo) => repo,
Err(err) => {
return ApiError::new(
StatusCode::BAD_GATEWAY,
format!("Failed to parse GitHub repo response: {err}"),
)
.into_response();
}
};
(
StatusCode::OK,
Json(serde_json::json!({
"owner": owner,
"name": name,
"accessible": true,
"default_branch": repo.default_branch,
"private": repo.private,
"permissions": repo.permissions,
"install_url": serde_json::Value::Null,
})),
)
.into_response()
}
async fn run_diagnostics(_auth: RequiredUser, State(state): State<Arc<AppState>>) -> Response {
(
StatusCode::OK,
Json(diagnostics::run_all(state.as_ref()).await),
)
.into_response()
}
pub(in crate::server) async fn openapi_spec() -> Response {
let yaml = include_str!("../../../../../../docs/public/api-reference/fabro-api.yaml");
let value: serde_json::Value =
serde_yaml::from_str(yaml).expect("embedded OpenAPI YAML is invalid");
Json(value).into_response()
}
async fn get_aggregate_billing(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
) -> Response {
let agg = state
.aggregate_billing
.lock()
.expect("aggregate_billing lock poisoned");
let by_model: Vec<BillingByModel> = agg
.by_model
.iter()
.map(|(model, totals)| BillingByModel {
billing: totals.billing.clone(),
model: ModelReference { id: model.clone() },
stages: totals.stages,
})
.collect();
let total_billing =
agg.by_model
.values()
.fold(BilledTokenCounts::default(), |mut acc, totals| {
let billing = &totals.billing;
acc.input_tokens += billing.input_tokens;
acc.output_tokens += billing.output_tokens;
acc.reasoning_tokens += billing.reasoning_tokens;
acc.cache_read_tokens += billing.cache_read_tokens;
acc.cache_write_tokens += billing.cache_write_tokens;
acc.total_tokens += billing.total_tokens;
if let Some(value) = billing.total_usd_micros {
*acc.total_usd_micros.get_or_insert(0) += value;
}
acc
});
let response = AggregateBilling {
totals: AggregateBillingTotals {
cache_read_tokens: total_billing.cache_read_tokens,
cache_write_tokens: total_billing.cache_write_tokens,
input_tokens: total_billing.input_tokens,
output_tokens: total_billing.output_tokens,
reasoning_tokens: total_billing.reasoning_tokens,
runs: agg.total_runs,
runtime_secs: agg.total_runtime_secs,
total_tokens: total_billing.total_tokens,
total_usd_micros: total_billing.total_usd_micros,
},
by_model,
};
(StatusCode::OK, Json(response)).into_response()
}