scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(client): extract fabro-client crate

Browse source 
Lift shared client DTOs into fabro-types, move auth/target/error/session
logic into fabro-client, and reduce fabro-cli to orchestration around the
builder-based client path.

This also lands the remaining plan cleanup for ApiError, ServerTarget
canonicalization, and the RunEventStream rename at the CLI boundary.
This commit is contained in:
Bryan Helmkamp 2026-04-20 20:42:20 -04:00
parent 90b911c927
commit f0f04abf44
No known key found for this signature in database
53 changed files with 2890 additions and 2454 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
Cargo.lock generated
View file

@ -1616,6 +1616,7 @@ dependencies = [
"fabro-api",
"fabro-auth",
"fabro-checkpoint",
"fabro-client",
"fabro-config",
"fabro-devcontainer",
"fabro-github",
@ -1682,6 +1683,32 @@ dependencies = [
"x509-parser",
]
[[package]]
name = "fabro-client"
version = "0.208.0-nightly.1"
dependencies = [
"anyhow",
"bytes",
"chrono",
"fabro-api",
"fabro-http",
"fabro-model",
"fabro-types",
"fabro-util",
"fs2",
"futures",
"libc",
"progenitor-client",
"rand 0.9.4",
"serde",
"serde_json",
"tempfile",
"thiserror 2.0.18",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "fabro-config"
version = "0.208.0-nightly.1"

25
docs/plans/2026-04-20-002-refactor-extract-fabro-client-crate-plan.md
View file

@ -41,6 +41,19 @@ These were all reasonable while the client had exactly one caller. They prevent
- R8. `fabro-cli` keeps CLI-owned orchestration: subprocess autostart, server-record lookup, dev-token-from-disk loading, `[cli.target]` TOML resolution, and the wrapper that stitches these together into a ready-to-use `fabro_client::Client`.
- R9. The full workspace build passes (`cargo build --workspace`), clippy is clean (`cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings`), rustfmt check passes, and all workspace tests continue to pass (`cargo nextest run --workspace`).
## Review Adjustments
- Remaining gaps against this plan in the current landed code:
- `HttpResponseFailure` → `ApiError` rename is still open.
- `ServerTarget` canonical-by-construction / lexical-only Unix-path canonicalization is still open; canonicalization still partly lives in `AuthStore`.
- `RunAttachEventStream` → `RunEventStream` rename is still open at the CLI boundary.
- Because the two rename cleanups above are still open, Unit 11's "no leaked old names" grep is currently expected to fail until that follow-up cleanup lands.
- Clarifications after implementation review:
- `RunProjection` ownership moved to `fabro-types`; external callers may continue to import it through `fabro_store`'s re-export. Verification should check type ownership and dependency boundaries, not the literal import spelling.
- `Client::from_http_client(...)` is the required public constructor. `Client::new_no_proxy(...)` may remain as a convenience wrapper, primarily for tests.
- `TransportConnector` is an acceptable internal helper inside `fabro-client` when needed to preserve caller-specific transport configuration across refresh/rebuild flows.
- `fabro-cli` may retain multiple thin `connect_*` convenience wrappers so long as they preserve CLI-only orchestration and funnel into `Client::builder()` under the hood.
## Scope Boundaries
**In scope:**
@ -96,7 +109,8 @@ None. Internal refactor; the patterns are all established locally.
- **`ServerTarget` canonical-by-construction, lexical only.** `ServerTarget::from_url` applies today's full HTTP canonicalization inline: lowercase scheme, lowercase host, strip default ports (`:443` on https, `:80` on http), strip trailing `/`, strip `/api/v1` suffix, rebuild authority as `{scheme}://{host}[:{port}]`. `ServerTarget::from_unix_path` applies lexical `.`/`..` resolution (no FS access, no symlink chasing). `PartialEq`/`Eq`/`Hash` operate on the canonical form directly. `ServerTargetKey` is deleted; `ServerTarget` itself serves as the `AuthStore` map key. In OOP style: construction *is* the canonicalizer — no separate `canonicalize()` method. Related helpers attach to `ServerTarget` as inherent methods (`target.loopback_classification()`, `target.build_public_http_client()`) rather than free functions.
- **`AuthStore` moves to `fabro-client`.** Default path remains `~/.fabro/auth.json` via `fabro_util::Home`. The public API (`get`/`put`/`remove`/`list`) is narrow enough that we keep it concrete — no `TokenRefresher` trait abstraction. Callers wanting alternative storage can pass an explicit path to `AuthStore::new`. If external-consumer flexibility becomes a real need later, we extract a trait then, not now.
- **Renames applied inline with moves.** We don't do a separate rename pass — the DTOs and internal types are renamed as they move. This keeps the compiler-driven find-all-callers loop honest: every broken import is both a move and a rename in one commit.
- **Connection API collapses to one builder.** Today's four `connect_*` functions in `server_client.rs` are all CLI-opinionated. `fabro-client` exposes a single `Client::builder().target(t).credential(c).oauth_session(s).connect().await?`. The CLI's `connect_server_with_settings` becomes a thin orchestrator: resolve target → autostart if needed → build `Credential` from dev-token/env/AuthStore → call `Client::builder()`.
- **Connection API centers on one builder.** Today's `connect_*` functions in `server_client.rs` are CLI-opinionated convenience wrappers. `fabro-client` exposes `Client::builder().target(t).credential(c).oauth_session(s).connect().await?` as the underlying connection API. `fabro-cli` may keep several thin `connect_*` wrappers, but they must remain orchestration-only and funnel into the builder instead of duplicating transport/session assembly logic.
- **Refresh rebuilds may use a transport connector hook.** If the CLI needs request-transport customization across OAuth refresh rebuilds (for example, preserving a CLI-specific user-agent), `fabro-client` may carry a `TransportConnector`-style helper as an internal implementation detail. This does not count as a second public connection API.
- **`OAuthSession` refresh fallback via `CredentialFallback` trait.** Today's refresh flow falls back to a dev-token-from-disk when the OAuth entry is missing, expired, or revoked. That fallback lookup reads CLI-owned sources (`FABRO_DEV_TOKEN` env, `~/.fabro/dev-token`, storage-dir dev-token file, fabro-server pidfile record) that don't belong in `fabro-client`. `OAuthSession` takes an optional `Box<dyn CredentialFallback>` at build time:
```text
@ -121,7 +135,7 @@ None. Internal refactor; the patterns are all established locally.
- **Does `EventEnvelope` need `EventPayload` to travel with it?** No. OpenAPI already defines the wire shape as `seq + RunEvent flattened`. `EventPayload` is a storage-internal validation helper and stays behind.
- **Does `AuthStore` need a trait-based abstraction?** No. Concrete type with a configurable file path is sufficient; we extract a trait the day a second implementation exists.
- **Does `ArtifactUpload` live in `fabro-types` or `fabro-client`?** `fabro-types`. Source is `fabro-workflow` (capture) → sink is `fabro-client` (upload). Placing the DTO in `fabro-types` prevents `fabro-workflow` from having to depend on `fabro-client`.
- **How do we handle the `Client::new_no_proxy(base_url)` constructor used by CLI tests today?** Expose `Client::from_http_client(base_url, http_client)` as a public `pub fn`; the CLI's `new_no_proxy` wrapper stays in `fabro-cli` test code.
- **How do we handle the `Client::new_no_proxy(base_url)` constructor used by CLI tests today?** Expose `Client::from_http_client(base_url, http_client)` as the stable public `pub fn`. `Client::new_no_proxy(base_url)` may remain as a small convenience wrapper (in `fabro-client` or CLI-local test code) if it continues to earn its keep.
- **`convert_type` serde round-trip helper — does it stay in the CLI or move with the client?** Moves with the client. It's how the client bridges `fabro_api::types::RunSummary` (wire) → `fabro_types::RunSummary` (domain) at response boundaries.
### Deferred to Implementation
@ -421,7 +435,7 @@ Write-path `EventPayload::new` sites (UNAFFECTED — stay in fabro-store as inte
**Verification:**
- `cargo build --workspace` succeeds.
- `cargo nextest run --workspace` passes.
- `grep -rn "fabro_store::RunProjection" lib/` returns only re-export lines and internal fabro-store uses; external callers use `fabro_types::RunProjection`.
- `grep -rn "struct RunProjection" lib/crates/fabro-store lib/crates/fabro-types` shows the concrete struct definition only in `fabro-types`; external callers may import either `fabro_types::RunProjection` or the `fabro-store` re-export.
---
@ -628,11 +642,12 @@ Write-path `EventPayload::new` sites (UNAFFECTED — stay in fabro-store as inte
}
```
The old `ClientBundle` name disappears; `ClientState` is private to the module.
- `Client::builder()`: new public API. Replaces today's four `connect_*` functions that blend CLI opinions with transport. The CLI's own `connect_server_with_settings` becomes a thin orchestrator over `Client::builder()` (handled in Unit 9).
- `Client::builder()`: new public API. It becomes the underlying connection API. The CLI may keep thin `connect_*` orchestration wrappers around it (handled in Unit 9), but transport/session assembly should live in the builder path rather than being duplicated across wrappers.
- `RunEventStream` rename: the struct, its `next_event`/`buffer_sse_events` methods, and the `VecDeque<EventEnvelope>` field. `EventEnvelope` is now `fabro_types::EventEnvelope`.
- Method bodies: the 40 wrappers move verbatim. They call `send_api(|client| ...)` — `client` is the `fabro_api::ApiClient` from `ClientState`. `convert_type::<_, fabro_types::RunSummary>(...)` continues to bridge wire → domain.
- `Client::from_http_client(base_url, http_client)` — public `pub fn` constructor for test use (replaces today's `new_no_proxy`). The CLI's test code can still build one via this with a `no_proxy()` builder.
- `Client::from_http_client(base_url, http_client)` — public `pub fn` constructor for test use and non-builder callers. `Client::new_no_proxy(base_url)` may remain as a small convenience wrapper built on top of it.
- Preserve `send_api`'s 401 → refresh → retry auto-logic. `OAuthSession` owns the refresh state it needs (`target`, `auth_store`, optional `fallback`); the actual refresh HTTP call uses a bespoke HTTP client built via `target.build_public_http_client()` (method on `ServerTarget`, not a free function — OOP style).
- If preserving caller-specific transport behavior across refresh rebuilds requires it, `Client` may carry an internal `TransportConnector` helper that can rebuild the transport with the same customization after credentials change.
- `CredentialFallback` trait lives in `fabro-client::credential`:
```text
// Directional — not implementation

1
lib/crates/fabro-cli/Cargo.toml
View file

@ -39,6 +39,7 @@ graphviz-sys.workspace = true
fabro-validate = { path = "../fabro-validate" }
fabro-workflow = { path = "../fabro-workflow" }
fabro-server = { path = "../fabro-server" }
fabro-client = { path = "../fabro-client" }
fabro-api = { path = "../fabro-api" }
fabro-telemetry = { path = "../fabro-telemetry" }
fabro-store = { path = "../fabro-store" }

24
lib/crates/fabro-cli/src/commands/auth/login.rs
View file

@ -3,6 +3,7 @@ use std::time::Duration;
use anyhow::{Context as _, Result, anyhow, bail};
use chrono::{DateTime, Utc};
use fabro_api::types;
use fabro_client::{AuthEntry, AuthStore, StoredSubject};
use fabro_http::header::CONTENT_TYPE;
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::CliLayer;
@ -11,9 +12,7 @@ use serde::Deserialize;
use tokio::time::timeout;
use crate::args::{AuthLoginArgs, require_no_json_override};
use crate::auth_store::{AuthEntry, AuthStore, ServerTargetKey, StoredSubject};
use crate::command_context::CommandContext;
use crate::loopback_target::{LoopbackClassification, is_loopback_or_unix_socket};
use crate::user_config;
use crate::user_config::ServerTarget;
@ -56,7 +55,6 @@ pub(super) async fn login_command(
{
let ctx = CommandContext::base(printer, cli.clone(), cli_layer)?;
let target = user_config::resolve_server_target(&args.server, ctx.machine_settings())?;
let server_key = ServerTargetKey::new(&target)?;
let config = fetch_cli_auth_config(&target).await?;
if !config.enabled {
bail!("{}", cli_auth_unavailable_message(config.reason.as_deref()));
@ -98,11 +96,11 @@ pub(super) async fn login_command(
}
};
match is_loopback_or_unix_socket(&target)? {
LoopbackClassification::Https
| LoopbackClassification::LoopbackHttp
| LoopbackClassification::UnixSocket => {}
LoopbackClassification::Rejected => {
match target.loopback_classification()? {
fabro_client::LoopbackClassification::Https
| fabro_client::LoopbackClassification::LoopbackHttp
| fabro_client::LoopbackClassification::UnixSocket => {}
fabro_client::LoopbackClassification::Rejected => {
bail!("{}", token_transport_error(&target));
}
}
@ -123,8 +121,8 @@ pub(super) async fn login_command(
logged_in_at: Utc::now(),
};
let summary = identity_summary(&entry.subject);
AuthStore::default().put(&server_key, entry)?;
fabro_util::printerr!(printer, "Logged in to {} as {}", server_key, summary);
AuthStore::default().put(&target, entry)?;
fabro_util::printerr!(printer, "Logged in to {} as {}", target, summary);
Ok(())
}
}
@ -270,11 +268,11 @@ struct OAuthErrorBody {
mod tests {
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use fabro_client::LoopbackClassification;
use insta::assert_snapshot;
use sha2::{Digest, Sha256};
use super::{build_browser_url, cli_auth_unavailable_message, login_failure_message};
use crate::loopback_target::{LoopbackClassification, is_loopback_or_unix_socket};
use crate::user_config::ServerTarget;
#[test]
@ -343,9 +341,9 @@ mod tests {
#[test]
fn token_transport_accepts_only_https_loopback_or_unix() {
let target = ServerTarget::HttpUrl("https://fabro.example.com".to_string());
let target = ServerTarget::http_url("https://fabro.example.com").unwrap();
assert_eq!(
is_loopback_or_unix_socket(&target).unwrap(),
target.loopback_classification().unwrap(),
LoopbackClassification::Https
);
}

74
lib/crates/fabro-cli/src/commands/auth/logout.rs
View file

@ -1,11 +1,11 @@
use anyhow::{Result, bail};
use fabro_client::{AuthEntry, AuthStore};
use fabro_http::header::AUTHORIZATION;
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::CliLayer;
use fabro_util::printer::Printer;
use crate::args::{AuthLogoutArgs, require_no_json_override};
use crate::auth_store::{AuthEntry, AuthStore, ServerTargetKey};
use crate::command_context::CommandContext;
use crate::user_config;
use crate::user_config::ServerTarget;
@ -29,13 +29,11 @@ pub(super) async fn logout_command(
}
let mut warnings = Vec::new();
for (key, entry) in entries {
if let Ok(target) = server_target_from_key(&key) {
if let Err(error) = revoke_remote_session(&target, &entry).await {
warnings.push(format_warning(&key, &error.to_string()));
}
for (target, entry) in entries {
if let Err(error) = revoke_remote_session(&target, &entry).await {
warnings.push(format_warning(&target, &error.to_string()));
}
store.remove(&key)?;
store.remove(&target)?;
}
for warning in warnings {
@ -46,17 +44,16 @@ pub(super) async fn logout_command(
}
let target = user_config::resolve_server_target(&args.server, ctx.machine_settings())?;
let key = ServerTargetKey::new(&target)?;
let Some(entry) = store.get(&key)? else {
fabro_util::printerr!(printer, "Not logged in to {}.", key);
let Some(entry) = store.get(&target)? else {
fabro_util::printerr!(printer, "Not logged in to {}.", target);
return Ok(());
};
if let Err(error) = revoke_remote_session(&target, &entry).await {
fabro_util::printerr!(printer, "{}", format_warning(&key, &error.to_string()));
fabro_util::printerr!(printer, "{}", format_warning(&target, &error.to_string()));
}
store.remove(&key)?;
fabro_util::printerr!(printer, "Logged out from {}.", key);
store.remove(&target)?;
fabro_util::printerr!(printer, "Logged out from {}.", target);
Ok(())
}
@ -79,62 +76,21 @@ async fn revoke_remote_session(target: &ServerTarget, entry: &AuthEntry) -> Resu
bail!("request failed with status {status}: {body}")
}
fn server_target_from_key(key: &ServerTargetKey) -> Result<ServerTarget> {
let value = key.to_string();
if let Some(path) = value.strip_prefix("unix://") {
return Ok(ServerTarget::UnixSocket(path.into()));
}
if value.starts_with("http://") || value.starts_with("https://") {
return Ok(ServerTarget::HttpUrl(value));
}
bail!("invalid auth store server key `{value}`")
}
fn format_warning(key: &ServerTargetKey, error: &str) -> String {
fn format_warning(target: &ServerTarget, error: &str) -> String {
format!(
"Warning: removed local session for {key}, but remote revocation failed: {error}. The refresh token may remain valid until it expires."
"Warning: removed local session for {target}, but remote revocation failed: {error}. The refresh token may remain valid until it expires."
)
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use super::{format_warning, server_target_from_key};
use crate::auth_store::ServerTargetKey;
use super::format_warning;
use crate::user_config::ServerTarget;
#[test]
fn rebuilds_server_target_from_http_key() {
let key = ServerTargetKey::new(&ServerTarget::HttpUrl(
"https://fabro.example.com/api/v1".to_string(),
))
.unwrap();
assert_eq!(
server_target_from_key(&key).unwrap(),
ServerTarget::HttpUrl("https://fabro.example.com".to_string())
);
}
#[test]
fn rebuilds_server_target_from_unix_key() {
let key = ServerTargetKey::new(&ServerTarget::UnixSocket(PathBuf::from("/tmp/fabro.sock")))
.unwrap();
assert_eq!(
server_target_from_key(&key).unwrap(),
ServerTarget::UnixSocket(PathBuf::from("/tmp/fabro.sock"))
);
}
#[test]
fn warning_mentions_local_removal_and_remote_failure() {
let key = ServerTargetKey::new(&ServerTarget::HttpUrl(
"https://fabro.example.com".to_string(),
))
.unwrap();
let warning = format_warning(&key, "request failed with status 500");
let target = ServerTarget::http_url("https://fabro.example.com").unwrap();
let warning = format_warning(&target, "request failed with status 500");
assert!(warning.contains("removed local session"));
assert!(warning.contains("remote revocation failed"));
}

15
lib/crates/fabro-cli/src/commands/auth/status.rs
View file

@ -1,5 +1,6 @@
use anyhow::Result;
use chrono::{DateTime, Utc};
use fabro_client::{AuthEntry, AuthStore};
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::CliLayer;
use fabro_util::dev_token::{read_dev_token_file, validate_dev_token_format};
@ -7,7 +8,6 @@ use fabro_util::printer::Printer;
use serde::Serialize;
use crate::args::AuthStatusArgs;
use crate::auth_store::{AuthEntry, AuthStore, ServerTargetKey};
use crate::command_context::CommandContext;
use crate::shared::print_json_pretty;
use crate::user_config;
@ -126,7 +126,7 @@ fn all_rows(store: &AuthStore, now: DateTime<Utc>) -> Result<Vec<StatusRow>> {
Ok(store
.list()?
.into_iter()
.map(|(key, entry)| status_row(&key, entry, now))
.map(|(target, entry)| status_row(&target, entry, now))
.collect())
}
@ -135,17 +135,16 @@ fn filter_rows(
target: &ServerTarget,
now: DateTime<Utc>,
) -> Result<Vec<StatusRow>> {
let key = ServerTargetKey::new(target)?;
Ok(store
.get(&key)?
.get(target)?
.into_iter()
.map(|entry| status_row(&key, entry, now))
.map(|entry| status_row(target, entry, now))
.collect())
}
fn status_row(key: &ServerTargetKey, entry: AuthEntry, now: DateTime<Utc>) -> StatusRow {
fn status_row(target: &ServerTarget, entry: AuthEntry, now: DateTime<Utc>) -> StatusRow {
StatusRow {
server: key.to_string(),
server: target.to_string(),
oauth_state: oauth_state(&entry, now),
access_token_expires_at: entry.access_token_expires_at,
refresh_token_expires_at: entry.refresh_token_expires_at,
@ -187,9 +186,9 @@ fn load_dev_token_if_available() -> bool {
#[cfg(test)]
mod tests {
use chrono::Duration;
use fabro_client::{AuthEntry, StoredSubject};
use super::{OAuthState, human_state, oauth_state};
use crate::auth_store::{AuthEntry, StoredSubject};
fn entry(access_offset_secs: i64, refresh_offset_secs: i64) -> AuthEntry {
let now = chrono::Utc::now();

2
lib/crates/fabro-cli/src/commands/doctor.rs
View file

@ -343,7 +343,7 @@ pub(crate) async fn run_doctor(
checks: vec![CheckResult {
name: "Location".to_string(),
status: CheckStatus::Pass,
summary: server.base_url().to_string(),
summary: server.base_url().clone(),
details: vec![],
remediation: None,
}],

4
lib/crates/fabro-cli/src/commands/exec.rs
View file

@ -115,7 +115,7 @@ struct AuthenticatedFabroServerAdapter {
impl AuthenticatedFabroServerAdapter {
fn new(client: server_client::Client, provider_name: impl Into<String>) -> Self {
let base_url = client.base_url().to_string();
let base_url = client.base_url().clone();
Self {
client,
base_url,
@ -212,7 +212,7 @@ fn transport_error(provider: &str, err: &anyhow::Error) -> LlmError {
}
}
fn map_response_failure(provider: &str, failure: &server_client::HttpResponseFailure) -> LlmError {
fn map_response_failure(provider: &str, failure: &fabro_client::ApiError) -> LlmError {
let retry_after = parse_retry_after(&failure.headers);
let (message, code, raw) = parse_server_error_body(&failure.body);
error_from_status_code(

14
lib/crates/fabro-cli/src/commands/run/attach.rs
View file

@ -21,7 +21,7 @@ use fabro_interview::{AnswerValue, ConsoleInterviewer, Question, QuestionOption,
use fabro_store::EventEnvelope;
use fabro_types::settings::cli::OutputVerbosity;
use fabro_types::settings::run::ApprovalMode;
use fabro_types::{EventBody, RunEvent, RunId};
use fabro_types::{EventBody, RunId};
use fabro_util::json::normalize_json_value;
use fabro_util::printer::Printer;
use fabro_util::terminal::Styles;
@ -155,7 +155,7 @@ async fn attach_live_run_with_client(
client: &server_client::Client,
run_id: &RunId,
existing_events: Vec<EventEnvelope>,
mut stream: server_client::RunAttachEventStream,
mut stream: server_client::RunEventStream,
styles: &'static Styles,
opts: AttachOptions,
printer: Printer,
@ -392,7 +392,7 @@ fn show_progress(progress_ui: &mut run_progress::ProgressUI, json_output: bool)
}
fn event_payload_line(event: &EventEnvelope) -> Result<String> {
let mut value = normalize_json_value(event.payload.as_value().clone());
let mut value = normalize_json_value(event.event.to_value()?);
restore_empty_run_properties(&mut value);
serde_json::to_string(&value).map_err(Into::into)
}
@ -462,8 +462,7 @@ fn state_exit_code(state: &server_client::RunProjection) -> Option<ExitCode> {
}
fn event_exit_code(event: &EventEnvelope) -> Option<ExitCode> {
let run_event = RunEvent::try_from(&event.payload).ok()?;
match run_event.body {
match &event.event.body {
EventBody::RunCompleted(props) => Some(
if props.status == "success" || props.status == "partial_success" {
ExitCode::from(0)
@ -477,10 +476,7 @@ fn event_exit_code(event: &EventEnvelope) -> Option<ExitCode> {
}
fn event_starts_interview(event: &EventEnvelope) -> bool {
let Ok(run_event) = RunEvent::try_from(&event.payload) else {
return false;
};
matches!(run_event.body, EventBody::InterviewStarted(_))
matches!(event.event.body, EventBody::InterviewStarted(_))
}
#[cfg(test)]

9
lib/crates/fabro-cli/src/commands/run/create.rs
View file

@ -73,14 +73,15 @@ pub(crate) async fn create_run(
}
let created_run_id = client.create_run_from_manifest(built.manifest).await?;
let local_run_dir = match &target {
ServerTarget::UnixSocket(_) => Some(
let local_run_dir = if target.is_unix_socket() {
Some(
Storage::new(user_config::storage_dir(ctx.machine_settings())?)
.run_scratch(&created_run_id)
.root()
.to_path_buf(),
),
ServerTarget::HttpUrl(_) => None,
)
} else {
None
};
Ok(CreatedRun {

8
lib/crates/fabro-cli/src/commands/run/logs.rs
View file

@ -87,8 +87,8 @@ pub(crate) async fn run(
Ok(())
}
fn event_name(event: &fabro_store::EventEnvelope) -> Option<&str> {
event.payload.as_value().get("event")?.as_str()
fn event_name(event: &fabro_store::EventEnvelope) -> &str {
event.event.event_name()
}
fn apply_filters(
@ -175,7 +175,7 @@ async fn follow_store_logs(
let had_events = !events.is_empty();
let saw_terminal = events
.iter()
.any(|event| matches!(event_name(event), Some("run.completed" | "run.failed")));
.any(|event| matches!(event_name(event), "run.completed" | "run.failed"));
for event in events {
let line = event_payload_line(&event)?;
if pretty {
@ -268,7 +268,7 @@ async fn flush_remaining_store_events(
}
fn event_payload_line(event: &fabro_store::EventEnvelope) -> Result<String> {
let mut value = normalize_json_value(event.payload.as_value().clone());
let mut value = normalize_json_value(event.event.to_value()?);
restore_empty_run_properties(&mut value);
let line = serde_json::to_string(&value)?;
Ok(redact_jsonl_line(&line))

15
lib/crates/fabro-cli/src/commands/run/runner.rs
View file

@ -14,12 +14,11 @@ use anyhow::{Context, Result, anyhow};
use async_trait::async_trait;
use fabro_config::Storage;
use fabro_interview::{ControlInterviewer, WorkerControlEnvelope, WorkerControlMessage};
use fabro_store::{EventEnvelope, EventPayload, RunProjection};
use fabro_store::{EventEnvelope, RunProjection, RunProjectionReducer};
use fabro_types::settings::run::RunMode;
use fabro_types::settings::{InterpString, SettingsLayer};
use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, StatusReason};
use fabro_types::{ArtifactUpload, EventBody, RunBlobId, RunEvent, RunId, StatusReason};
use fabro_vault::Vault;
use fabro_workflow::artifact_snapshot::CapturedArtifactInfo;
use fabro_workflow::artifact_upload::{ArtifactSink, StageArtifactUploader};
use fabro_workflow::event::{Emitter, RunEventSink};
use fabro_workflow::operations::{self, StartServices};
@ -264,7 +263,7 @@ impl StageArtifactUploader for HttpArtifactUploader {
&self,
stage_id: &fabro_types::StageId,
artifact_capture_dir: &Path,
artifacts: &[CapturedArtifactInfo],
artifacts: &[ArtifactUpload],
) -> Result<()> {
if artifacts.is_empty() {
return Ok(());
@ -306,7 +305,7 @@ impl StageArtifactUploader for MissingArtifactUploadTokenUploader {
&self,
_stage_id: &fabro_types::StageId,
_artifact_capture_dir: &Path,
_artifacts: &[CapturedArtifactInfo],
_artifacts: &[ArtifactUpload],
) -> Result<()> {
Err(anyhow!(
"run {} could not upload artifacts because the worker did not receive an artifact upload token",
@ -369,8 +368,10 @@ impl HttpRunStore {
}
async fn apply_acknowledged_event(&self, seq: u32, event: &RunEvent) -> Result<()> {
let payload = EventPayload::new(event.to_value()?, &self.run_id)?;
let envelope = EventEnvelope { seq, payload };
let envelope = EventEnvelope {
seq,
event: event.clone(),
};
{
let mut state = self.state.lock().await;

2
lib/crates/fabro-cli/src/commands/store/rebuild.rs
View file

@ -17,7 +17,7 @@ pub(crate) async fn rebuild_run_store(
));
let run_store = store.create_run(run_id).await?;
for event in events {
let payload = EventPayload::new(event.payload.as_value().clone(), run_id)?;
let payload = EventPayload::new(event.event.to_value()?, run_id)?;
run_store.append_event(&payload).await?;
}
Ok(run_store)

2
lib/crates/fabro-cli/src/commands/system/events.rs
View file

@ -1,4 +1,5 @@
use anyhow::Result;
use fabro_client::sse;
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::{CliLayer, OutputFormat};
use fabro_util::printer::Printer;
@ -6,7 +7,6 @@ use futures::StreamExt;
use crate::args::SystemEventsArgs;
use crate::command_context::CommandContext;
use crate::sse;
pub(super) async fn events_command(
args: &SystemEventsArgs,

10
lib/crates/fabro-cli/src/commands/version.rs
View file

@ -132,9 +132,13 @@ fn is_non_release_profile(profile: &str) -> bool {
}
fn format_server_target(target: &ServerTarget) -> String {
match target {
ServerTarget::HttpUrl(api_url) => api_url.clone(),
ServerTarget::UnixSocket(path) => path.display().to_string(),
if let Some(api_url) = target.as_http_url() {
api_url.to_string()
} else {
target
.as_unix_socket_path()
.map(|path| path.display().to_string())
.unwrap_or_default()
}
}

3
lib/crates/fabro-cli/src/main.rs
View file

@ -4,20 +4,17 @@
)]
mod args;
mod auth_store;
mod command_context;
mod commands;
mod gh;
mod landing;
mod logging;
mod loopback_target;
mod manifest_builder;
mod server_client;
mod server_runs;
mod shared;
#[cfg(feature = "sleep_inhibitor")]
mod sleep_inhibitor;
mod sse;
mod user_config;
#[cfg(test)]

1750
lib/crates/fabro-cli/src/server_client.rs
View file

File diff suppressed because it is too large Load diff

3
lib/crates/fabro-cli/src/server_runs.rs
View file

@ -3,8 +3,7 @@ use std::sync::Arc;
use anyhow::Result;
use chrono::{DateTime, Utc};
use fabro_store::RunSummary;
use fabro_types::{RunId, RunStatus, StatusReason};
use fabro_types::{RunId, RunStatus, RunSummary, StatusReason};
use crate::server_client::Client;

73
lib/crates/fabro-cli/src/user_config.rs
View file

@ -1,7 +1,8 @@
use std::fmt;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use anyhow::{Result, bail};
use anyhow::Result;
pub(crate) use fabro_client::ServerTarget;
pub(crate) use fabro_config::user::*;
use fabro_types::settings::cli::CliTargetSettings;
use fabro_types::settings::{CliSettings, SettingsLayer};
@ -60,50 +61,10 @@ pub(crate) fn apply_storage_dir_override(
layer
}
#[derive(Debug, Clone, PartialEq)]
pub(crate) enum ServerTarget {
HttpUrl(String),
UnixSocket(PathBuf),
}
impl fmt::Display for ServerTarget {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::HttpUrl(api_url) => f.write_str(api_url),
Self::UnixSocket(path) => write!(f, "unix://{}", path.display()),
}
}
}
pub(crate) fn normalized_http_base_url(api_url: &str) -> &str {
let trimmed = api_url.trim_end_matches('/');
trimmed.strip_suffix("/api/v1").unwrap_or(trimmed)
}
pub(crate) fn build_public_http_client(
target: &ServerTarget,
) -> Result<(fabro_http::HttpClient, String)> {
match target {
ServerTarget::HttpUrl(api_url) => {
let http_client = cli_http_client_builder().build()?;
Ok((http_client, normalized_http_base_url(api_url).to_string()))
}
ServerTarget::UnixSocket(path) => {
#[cfg(unix)]
{
let http_client = cli_http_client_builder()
.unix_socket(path)
.no_proxy()
.build()?;
Ok((http_client, "http://fabro".to_string()))
}
#[cfg(not(unix))]
{
let _ = path;
bail!("Unix-socket HTTP client is not supported on this platform")
}
}
}
target.build_public_http_client()
}
/// Pull the resolved CLI target configuration out of `[cli.target]`.
@ -125,7 +86,7 @@ fn configured_server_target(settings: &SettingsLayer) -> Result<Option<ServerTar
}
pub(crate) fn default_server_target() -> ServerTarget {
ServerTarget::UnixSocket(default_socket_path())
ServerTarget::unix_socket_path(default_socket_path()).expect("default socket path is absolute")
}
pub(crate) fn storage_dir(settings: &SettingsLayer) -> anyhow::Result<PathBuf> {
@ -153,16 +114,7 @@ pub(crate) fn storage_dir(settings: &SettingsLayer) -> anyhow::Result<PathBuf> {
}
fn parse_server_target(value: &str) -> Result<ServerTarget> {
if value.starts_with("http://") || value.starts_with("https://") {
return Ok(ServerTarget::HttpUrl(value.to_string()));
}
let path = Path::new(value);
if path.is_absolute() {
return Ok(ServerTarget::UnixSocket(path.to_path_buf()));
}
bail!("server target must be an http(s) URL or absolute Unix socket path")
ServerTarget::from_str(value)
}
fn explicit_server_target(args: &ServerTargetArgs) -> Result<Option<ServerTarget>> {
@ -219,7 +171,7 @@ mod tests {
fn exec_uses_cli_server_target() {
assert_eq!(
exec_server_target(&server_target_args(Some("https://cli.example.com"))).unwrap(),
Some(ServerTarget::HttpUrl("https://cli.example.com".to_string()))
Some(ServerTarget::http_url("https://cli.example.com").unwrap())
);
}
@ -227,7 +179,7 @@ mod tests {
fn exec_supports_explicit_unix_socket_target() {
assert_eq!(
exec_server_target(&server_target_args(Some("/tmp/fabro.sock"))).unwrap(),
Some(ServerTarget::UnixSocket(PathBuf::from("/tmp/fabro.sock")))
Some(ServerTarget::unix_socket_path("/tmp/fabro.sock").unwrap())
);
}
@ -249,7 +201,7 @@ url = "https://config.example.com"
);
assert_eq!(
resolve_server_target(&server_target_args(None), &settings).unwrap(),
ServerTarget::HttpUrl("https://config.example.com".to_string())
ServerTarget::http_url("https://config.example.com").unwrap()
);
}
@ -270,7 +222,7 @@ url = "https://config.example.com"
&settings
)
.unwrap(),
ServerTarget::HttpUrl("https://cli.example.com".to_string())
ServerTarget::http_url("https://cli.example.com").unwrap()
);
}
@ -279,7 +231,8 @@ url = "https://config.example.com"
let settings = SettingsLayer::default();
assert_eq!(
resolve_server_target(&server_target_args(None), &settings).unwrap(),
ServerTarget::UnixSocket(dirs::home_dir().unwrap().join(".fabro/fabro.sock"))
ServerTarget::unix_socket_path(dirs::home_dir().unwrap().join(".fabro/fabro.sock"))
.unwrap()
);
}
@ -300,7 +253,7 @@ url = "https://config.example.com"
&settings
)
.unwrap(),
ServerTarget::HttpUrl("https://cli.example.com".to_string())
ServerTarget::http_url("https://cli.example.com").unwrap()
);
}

1
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -635,6 +635,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
{
"event": "run.queued",
"id": "[EVENT_ID]",
"properties": {},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},

17
lib/crates/fabro-cli/tests/it/cmd/rewind.rs
View file

@ -129,7 +129,7 @@ fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() {
assert!(
before_events
.iter()
.any(|event| event.payload.as_value()["event"] == "run.completed"),
.any(|event| event.event.event_name() == "run.completed"),
"setup run should be completed before rewind"
);
@ -153,28 +153,31 @@ fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() {
assert_eq!(
after_events[..before_events.len()]
.iter()
.map(|event| event.payload.as_value()["event"].as_str().unwrap())
.map(|event| event.event.event_name())
.collect::<Vec<_>>(),
before_events
.iter()
.map(|event| event.payload.as_value()["event"].as_str().unwrap())
.map(|event| event.event.event_name())
.collect::<Vec<_>>(),
"rewind should preserve the prior event prefix"
);
assert_eq!(
after_events[before_events.len()].payload.as_value()["event"],
after_events[before_events.len()].event.event_name(),
"run.rewound"
);
assert_eq!(
after_events[before_events.len() + 1].payload.as_value()["event"],
after_events[before_events.len() + 1].event.event_name(),
"checkpoint.completed"
);
assert_eq!(
after_events[before_events.len() + 2].payload.as_value()["event"],
after_events[before_events.len() + 2].event.event_name(),
"run.submitted"
);
assert!(
after_events[before_events.len() + 2].payload.as_value()["properties"]["definition_blob"]
after_events[before_events.len() + 2]
.event
.properties()
.unwrap()["definition_blob"]
.is_string(),
"rewind should re-emit run.submitted with the definition_blob"
);

2
lib/crates/fabro-cli/tests/it/cmd/runner.rs
View file

@ -29,7 +29,7 @@ fn stored_worker_events(run_dir: &std::path::Path) -> Vec<RunEvent> {
}
fn run_event(event: &EventEnvelope) -> RunEvent {
RunEvent::try_from(&event.payload).expect("stored event should parse")
event.event.clone()
}
fn assert_worker_succeeded(run_dir: &std::path::Path, stdout: &[u8]) {

9
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -789,14 +789,7 @@ pub(crate) fn wait_for_event_names(run_dir: &Path, expected: &[&str]) {
loop {
let event_names = run_events(run_dir)
.into_iter()
.filter_map(|event| {
event
.payload
.as_value()
.get("event")
.and_then(serde_json::Value::as_str)
.map(ToString::to_string)
})
.map(|event| event.event.event_name().to_string())
.collect::<Vec<_>>();
if expected

11
lib/crates/fabro-cli/tests/it/workflow/mod.rs
View file

@ -54,14 +54,9 @@ pub(super) fn completed_nodes(run_dir: &Path) -> Vec<String> {
}
pub(super) fn has_event(run_dir: &Path, event_name: &str) -> bool {
run_events(run_dir).into_iter().any(|event| {
event
.payload
.as_value()
.get("event")
.and_then(Value::as_str)
== Some(event_name)
})
run_events(run_dir)
.into_iter()
.any(|event| event.event.event_name() == event_name)
}
pub(super) fn store_dump_export(context: &TestContext, run_id: &str) -> PathBuf {

37
lib/crates/fabro-client/Cargo.toml Normal file
View file

@ -0,0 +1,37 @@
[package]
name = "fabro-client"
edition.workspace = true
version.workspace = true
publish = false
license.workspace = true
description = "Typed HTTP client for the Fabro API"
[lib]
doctest = false
[lints]
workspace = true
[dependencies]
anyhow.workspace = true
bytes.workspace = true
chrono = { workspace = true, features = ["serde"] }
fabro-api = { path = "../fabro-api" }
fabro-http.workspace = true
fabro-model = { path = "../fabro-model" }
fabro-types = { path = "../fabro-types" }
fabro-util = { path = "../fabro-util" }
fs2.workspace = true
futures.workspace = true
libc = "0.2"
progenitor-client = "0.13"
rand.workspace = true
serde.workspace = true
serde_json.workspace = true
thiserror.workspace = true
tokio.workspace = true
tokio-util.workspace = true
tracing.workspace = true
[dev-dependencies]
tempfile = "3"

229
lib/crates/fabro-cli/src/auth_store.rs → lib/crates/fabro-client/src/auth_store.rs
View file

@ -8,9 +8,9 @@
)]
use std::collections::BTreeMap;
use std::fs;
use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::{fmt, fs};
use chrono::{DateTime, Utc};
use fs2::FileExt;
@ -18,68 +18,31 @@ use rand::Rng;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use crate::user_config::{ServerTarget, normalized_http_base_url};
use crate::target::ServerTarget;
const AUTH_FILE_ENV: &str = "FABRO_AUTH_FILE";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct StoredSubject {
pub(crate) idp_issuer: String,
pub(crate) idp_subject: String,
pub(crate) login: String,
pub(crate) name: String,
pub(crate) email: String,
pub struct StoredSubject {
pub idp_issuer: String,
pub idp_subject: String,
pub login: String,
pub name: String,
pub email: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct AuthEntry {
pub(crate) access_token: String,
pub(crate) access_token_expires_at: DateTime<Utc>,
pub(crate) refresh_token: String,
pub(crate) refresh_token_expires_at: DateTime<Utc>,
pub(crate) subject: StoredSubject,
pub(crate) logged_in_at: DateTime<Utc>,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub(crate) struct ServerTargetKey(String);
impl ServerTargetKey {
pub(crate) fn new(target: &ServerTarget) -> Result<Self, AuthStoreError> {
match target {
ServerTarget::HttpUrl(api_url) => canonical_http_target(api_url).map(Self),
ServerTarget::UnixSocket(path) => Ok(Self(format!(
"unix://{}",
canonical_socket_path(path)?.display()
))),
}
}
fn from_canonical(canonical: String) -> Self {
Self(canonical)
}
pub(crate) fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for ServerTargetKey {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl TryFrom<&ServerTarget> for ServerTargetKey {
type Error = AuthStoreError;
fn try_from(value: &ServerTarget) -> Result<Self, Self::Error> {
Self::new(value)
}
pub struct AuthEntry {
pub access_token: String,
pub access_token_expires_at: DateTime<Utc>,
pub refresh_token: String,
pub refresh_token_expires_at: DateTime<Utc>,
pub subject: StoredSubject,
pub logged_in_at: DateTime<Utc>,
}
#[derive(Debug, Error)]
pub(crate) enum AuthStoreError {
pub enum AuthStoreError {
#[allow(
dead_code,
reason = "This platform-gated variant is exercised on non-Unix targets."
@ -118,7 +81,7 @@ pub(crate) enum AuthStoreError {
}
#[derive(Debug, Error)]
pub(crate) enum LockError {
pub enum LockError {
#[error(
"the filesystem backing {path} does not support file locking; move the auth store to a local filesystem or set {AUTH_FILE_ENV} to a local path"
)]
@ -131,7 +94,7 @@ pub(crate) enum LockError {
}
#[derive(Debug, Clone)]
pub(crate) struct AuthStore {
pub struct AuthStore {
path: PathBuf,
}
@ -152,47 +115,45 @@ impl Default for AuthStore {
}
impl AuthStore {
pub(crate) fn new(path: PathBuf) -> Self {
pub fn new(path: PathBuf) -> Self {
Self { path }
}
pub(crate) fn get(&self, key: &ServerTargetKey) -> Result<Option<AuthEntry>, AuthStoreError> {
pub fn get(&self, target: &ServerTarget) -> Result<Option<AuthEntry>, AuthStoreError> {
if !self.path.exists() {
return Ok(None);
}
let key = key_for_target(target);
self.with_shared_lock(|| {
let file = self.read_auth_file()?;
Ok(file.servers.get(key.as_str()).cloned())
Ok(file.servers.get(&key).cloned())
})
}
pub(crate) fn put(
&self,
key: &ServerTargetKey,
entry: AuthEntry,
) -> Result<(), AuthStoreError> {
pub fn put(&self, target: &ServerTarget, entry: AuthEntry) -> Result<(), AuthStoreError> {
#[cfg(not(unix))]
{
let _ = (key, entry);
let _ = (target, entry);
Err(AuthStoreError::UnsupportedPlatform)
}
#[cfg(unix)]
{
let key = key_for_target(target);
self.ensure_parent_dir()?;
self.with_exclusive_lock(|| {
let mut file = self.read_auth_file_if_exists()?;
file.servers.insert(key.to_string(), entry);
file.servers.insert(key, entry);
self.write_auth_file(&file)
})
}
}
pub(crate) fn remove(&self, key: &ServerTargetKey) -> Result<bool, AuthStoreError> {
pub fn remove(&self, target: &ServerTarget) -> Result<bool, AuthStoreError> {
#[cfg(not(unix))]
{
let _ = key;
let _ = target;
Err(AuthStoreError::UnsupportedPlatform)
}
@ -202,28 +163,28 @@ impl AuthStore {
return Ok(false);
}
let key = key_for_target(target);
self.ensure_parent_dir()?;
self.with_exclusive_lock(|| {
let mut file = self.read_auth_file_if_exists()?;
let removed = file.servers.remove(key.as_str()).is_some();
let removed = file.servers.remove(&key).is_some();
self.write_auth_file(&file)?;
Ok(removed)
})
}
}
pub(crate) fn list(&self) -> Result<Vec<(ServerTargetKey, AuthEntry)>, AuthStoreError> {
pub fn list(&self) -> Result<Vec<(ServerTarget, AuthEntry)>, AuthStoreError> {
if !self.path.exists() {
return Ok(Vec::new());
}
self.with_shared_lock(|| {
let file = self.read_auth_file()?;
Ok(file
.servers
file.servers
.into_iter()
.map(|(key, entry)| (ServerTargetKey::from_canonical(key), entry))
.collect())
.map(|(key, entry)| Ok((parse_stored_target(&key)?, entry)))
.collect::<Result<Vec<_>, AuthStoreError>>()
})
}
@ -347,48 +308,26 @@ impl AuthStore {
}
}
fn canonical_http_target(api_url: &str) -> Result<String, AuthStoreError> {
let trimmed = api_url.trim();
let normalized = normalized_http_base_url(trimmed);
let url =
fabro_http::Url::parse(normalized).map_err(|_| AuthStoreError::InvalidServerTarget {
value: api_url.to_string(),
})?;
let scheme = url.scheme().to_ascii_lowercase();
let Some(host) = url.host_str() else {
return Err(AuthStoreError::InvalidServerTarget {
value: api_url.to_string(),
});
};
let host = host.to_ascii_lowercase();
let Some(port) = url.port_or_known_default() else {
return Err(AuthStoreError::InvalidServerTarget {
value: api_url.to_string(),
});
};
let default_port = match scheme.as_str() {
"http" => 80,
"https" => 443,
_ => {
return Err(AuthStoreError::InvalidServerTarget {
value: api_url.to_string(),
});
}
};
if port == default_port {
Ok(format!("{scheme}://{host}"))
} else {
Ok(format!("{scheme}://{host}:{port}"))
}
fn key_for_target(target: &ServerTarget) -> String {
target.to_string()
}
fn canonical_socket_path(path: &Path) -> Result<PathBuf, AuthStoreError> {
if !path.is_absolute() {
return Err(AuthStoreError::InvalidServerTarget {
value: path.display().to_string(),
fn parse_stored_target(value: &str) -> Result<ServerTarget, AuthStoreError> {
if let Some(path) = value.strip_prefix("unix://") {
return ServerTarget::unix_socket_path(path).map_err(|_| {
AuthStoreError::InvalidServerTarget {
value: value.to_string(),
}
});
}
Ok(fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()))
if value.starts_with("http://") || value.starts_with("https://") {
return ServerTarget::http_url(value).map_err(|_| AuthStoreError::InvalidServerTarget {
value: value.to_string(),
});
}
Err(AuthStoreError::InvalidServerTarget {
value: value.to_string(),
})
}
#[cfg(unix)]
@ -438,8 +377,8 @@ mod tests {
#[cfg(unix)]
use super::{AUTH_FILE_ENV, LockError, classify_lock_error};
use super::{AuthEntry, AuthStore, ServerTargetKey, StoredSubject};
use crate::user_config::ServerTarget;
use super::{AuthEntry, AuthStore, StoredSubject, key_for_target};
use crate::target::ServerTarget;
fn entry(login: &str) -> AuthEntry {
let now = chrono::Utc::now();
@ -460,7 +399,7 @@ mod tests {
}
fn https_target(value: &str) -> ServerTarget {
ServerTarget::HttpUrl(value.to_string())
ServerTarget::http_url(value).unwrap()
}
#[cfg(unix)]
@ -468,11 +407,11 @@ mod tests {
fn round_trips_https_entry() {
let temp = tempfile::tempdir().unwrap();
let store = AuthStore::new(temp.path().join("auth.json"));
let key = ServerTargetKey::new(&https_target("https://fabro.example.com")).unwrap();
let target = https_target("https://fabro.example.com");
store.put(&key, entry("octocat")).unwrap();
store.put(&target, entry("octocat")).unwrap();
let saved = store.get(&key).unwrap().unwrap();
let saved = store.get(&target).unwrap().unwrap();
assert_eq!(saved.subject.login, "octocat");
}
@ -481,11 +420,11 @@ mod tests {
fn round_trips_loopback_http_entry() {
let temp = tempfile::tempdir().unwrap();
let store = AuthStore::new(temp.path().join("auth.json"));
let key = ServerTargetKey::new(&https_target("http://127.0.0.1:3000")).unwrap();
let target = https_target("http://127.0.0.1:3000");
store.put(&key, entry("alice")).unwrap();
store.put(&target, entry("alice")).unwrap();
let saved = store.get(&key).unwrap().unwrap();
let saved = store.get(&target).unwrap().unwrap();
assert_eq!(saved.subject.login, "alice");
}
@ -496,19 +435,19 @@ mod tests {
let socket = temp.path().join("fabro.sock");
std::fs::write(&socket, "").unwrap();
let store = AuthStore::new(temp.path().join("auth.json"));
let key = ServerTargetKey::new(&ServerTarget::UnixSocket(socket)).unwrap();
let target = ServerTarget::unix_socket_path(socket).unwrap();
store.put(&key, entry("unix")).unwrap();
store.put(&target, entry("unix")).unwrap();
let saved = store.get(&key).unwrap().unwrap();
let saved = store.get(&target).unwrap().unwrap();
assert_eq!(saved.subject.login, "unix");
}
#[test]
fn https_normalization_collapses_equivalent_urls() {
let a = ServerTargetKey::new(&https_target("https://EXAMPLE.COM/")).unwrap();
let b = ServerTargetKey::new(&https_target("https://example.com:443")).unwrap();
let c = ServerTargetKey::new(&https_target("https://example.com")).unwrap();
let a = key_for_target(&https_target("https://EXAMPLE.COM/"));
let b = key_for_target(&https_target("https://example.com:443"));
let c = key_for_target(&https_target("https://example.com"));
assert_eq!(a, b);
assert_eq!(b, c);
@ -516,36 +455,34 @@ mod tests {
#[test]
fn distinct_unix_socket_paths_do_not_collide() {
let a =
ServerTargetKey::new(&ServerTarget::UnixSocket(PathBuf::from("/tmp/a.sock"))).unwrap();
let b =
ServerTargetKey::new(&ServerTarget::UnixSocket(PathBuf::from("/tmp/b.sock"))).unwrap();
let a = key_for_target(&ServerTarget::unix_socket_path("/tmp/a.sock").unwrap());
let b = key_for_target(&ServerTarget::unix_socket_path("/tmp/b.sock").unwrap());
assert_ne!(a, b);
}
#[cfg(unix)]
#[test]
fn canonicalizes_symlinked_socket_paths() {
fn preserves_distinct_symlinked_socket_paths() {
let temp = tempfile::tempdir().unwrap();
let socket = temp.path().join("fabro.sock");
let link = temp.path().join("fabro-link.sock");
std::fs::write(&socket, "").unwrap();
std::os::unix::fs::symlink(&socket, &link).unwrap();
let direct = ServerTargetKey::new(&ServerTarget::UnixSocket(socket)).unwrap();
let via_link = ServerTargetKey::new(&ServerTarget::UnixSocket(link)).unwrap();
let direct = key_for_target(&ServerTarget::unix_socket_path(socket).unwrap());
let via_link = key_for_target(&ServerTarget::unix_socket_path(link).unwrap());
assert_eq!(direct, via_link);
assert_ne!(direct, via_link);
}
#[test]
fn missing_file_returns_empty_results() {
let temp = tempfile::tempdir().unwrap();
let store = AuthStore::new(temp.path().join("auth.json"));
let key = ServerTargetKey::new(&https_target("https://fabro.example.com")).unwrap();
let target = https_target("https://fabro.example.com");
assert!(store.get(&key).unwrap().is_none());
assert!(store.get(&target).unwrap().is_none());
assert!(store.list().unwrap().is_empty());
}
@ -555,9 +492,9 @@ mod tests {
let path = temp.path().join("auth.json");
std::fs::write(&path, "{not-json").unwrap();
let store = AuthStore::new(path.clone());
let key = ServerTargetKey::new(&https_target("https://fabro.example.com")).unwrap();
let target = https_target("https://fabro.example.com");
let err = store.get(&key).unwrap_err();
let err = store.get(&target).unwrap_err();
assert!(err.to_string().contains(&path.display().to_string()));
}
@ -566,21 +503,21 @@ mod tests {
fn concurrent_puts_do_not_corrupt_file() {
let temp = tempfile::tempdir().unwrap();
let store = Arc::new(AuthStore::new(temp.path().join("auth.json")));
let key = ServerTargetKey::new(&https_target("https://fabro.example.com")).unwrap();
let target = https_target("https://fabro.example.com");
let mut tasks = Vec::new();
for login in ["alice", "bob"] {
let store = Arc::clone(&store);
let key = key.clone();
let target = target.clone();
tasks.push(thread::spawn(move || {
store.put(&key, entry(login)).unwrap();
store.put(&target, entry(login)).unwrap();
}));
}
for task in tasks {
task.join().unwrap();
}
let saved = store.get(&key).unwrap().unwrap();
let saved = store.get(&target).unwrap().unwrap();
assert!(matches!(saved.subject.login.as_str(), "alice" | "bob"));
}
@ -591,9 +528,9 @@ mod tests {
let temp = tempfile::tempdir().unwrap();
let store = AuthStore::new(temp.path().join("auth.json"));
let key = ServerTargetKey::new(&https_target("https://fabro.example.com")).unwrap();
let target = https_target("https://fabro.example.com");
store.put(&key, entry("octocat")).unwrap();
store.put(&target, entry("octocat")).unwrap();
let mode = std::fs::metadata(temp.path().join("auth.json"))
.unwrap()
@ -608,9 +545,9 @@ mod tests {
fn put_returns_unsupported_platform() {
let temp = tempfile::tempdir().unwrap();
let store = AuthStore::new(temp.path().join("auth.json"));
let key = ServerTargetKey::new(&https_target("https://fabro.example.com")).unwrap();
let target = https_target("https://fabro.example.com");
let err = store.put(&key, entry("octocat")).unwrap_err();
let err = store.put(&target, entry("octocat")).unwrap_err();
assert!(err.to_string().contains("not supported on this platform"));
}

1392
lib/crates/fabro-client/src/client.rs Normal file
View file

File diff suppressed because it is too large Load diff

40
lib/crates/fabro-client/src/credential.rs Normal file
View file

@ -0,0 +1,40 @@
use std::fmt;
use crate::AuthEntry;
#[derive(Clone)]
pub enum Credential {
DevToken(String),
OAuth(AuthEntry),
}
pub trait CredentialFallback: Send + Sync {
fn resolve(&self) -> Option<Credential>;
}
impl<F> CredentialFallback for F
where
F: Fn() -> Option<Credential> + Send + Sync,
{
fn resolve(&self) -> Option<Credential> {
self()
}
}
impl Credential {
pub fn bearer_token(&self) -> &str {
match self {
Self::DevToken(token) => token,
Self::OAuth(entry) => &entry.access_token,
}
}
}
impl fmt::Debug for Credential {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::DevToken(_) => f.write_str("Credential::DevToken(<redacted>)"),
Self::OAuth(_) => f.write_str("Credential::OAuth(<redacted>)"),
}
}
}

184
lib/crates/fabro-client/src/error.rs Normal file
View file

@ -0,0 +1,184 @@
use anyhow::{Result, anyhow};
use serde::de::DeserializeOwned;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApiFailure {
pub status: fabro_http::StatusCode,
pub code: Option<String>,
}
pub struct StructuredApiError {
pub error: anyhow::Error,
pub failure: Option<ApiFailure>,
}
pub struct ApiError {
pub status: fabro_http::StatusCode,
pub headers: fabro_http::HeaderMap,
pub body: String,
failure: ApiFailure,
}
impl ApiError {
pub fn api_failure(&self) -> &ApiFailure {
&self.failure
}
}
pub fn parse_error_response_value(value: &serde_json::Value) -> (Option<String>, Option<String>) {
let first = value
.get("errors")
.and_then(serde_json::Value::as_array)
.and_then(|errors| errors.first());
let detail = first
.and_then(|entry| entry.get("detail"))
.and_then(serde_json::Value::as_str)
.map(ToOwned::to_owned);
let code = first
.and_then(|entry| entry.get("code"))
.and_then(serde_json::Value::as_str)
.map(ToOwned::to_owned);
(detail, code)
}
pub async fn classify_api_error<E>(err: progenitor_client::Error<E>) -> StructuredApiError
where
E: serde::Serialize + std::fmt::Debug,
{
match err {
progenitor_client::Error::UnexpectedResponse(response) => {
let status = response.status();
let body = response.text().await.unwrap_or_default();
let mut code = None;
if let Ok(value) = serde_json::from_str::<serde_json::Value>(&body) {
let (detail, parsed_code) = parse_error_response_value(&value);
code = parsed_code;
if let Some(detail) = detail {
return StructuredApiError {
error: anyhow!("{detail}"),
failure: Some(ApiFailure { status, code }),
};
}
}
let error = if body.is_empty() {
anyhow!("request failed with status {status}")
} else {
anyhow!("request failed with status {status}: {body}")
};
StructuredApiError {
error,
failure: Some(ApiFailure { status, code }),
}
}
other => map_api_error_structured(other),
}
}
fn map_api_error_structured<E>(err: progenitor_client::Error<E>) -> StructuredApiError
where
E: serde::Serialize + std::fmt::Debug,
{
match err {
progenitor_client::Error::ErrorResponse(response) => {
let status = response.status();
let mut code = None;
if let Ok(value) = serde_json::to_value(response.into_inner()) {
let (detail, parsed_code) = parse_error_response_value(&value);
code = parsed_code;
if let Some(detail) = detail {
return StructuredApiError {
error: anyhow!("{detail}"),
failure: Some(ApiFailure { status, code }),
};
}
}
StructuredApiError {
error: anyhow!("request failed with status {status}"),
failure: Some(ApiFailure { status, code }),
}
}
progenitor_client::Error::UnexpectedResponse(response) => StructuredApiError {
error: anyhow!("request failed with status {}", response.status()),
failure: Some(ApiFailure {
status: response.status(),
code: None,
}),
},
other => StructuredApiError {
error: anyhow!("{other}"),
failure: None,
},
}
}
pub fn map_api_error<E>(err: progenitor_client::Error<E>) -> anyhow::Error
where
E: serde::Serialize + std::fmt::Debug,
{
map_api_error_structured(err).error
}
pub fn raw_response_failure_error(failure: &ApiError) -> anyhow::Error {
if let Ok(value) = serde_json::from_str::<serde_json::Value>(&failure.body) {
let (detail, _) = parse_error_response_value(&value);
if let Some(detail) = detail {
return anyhow!("{detail}");
}
}
if failure.body.is_empty() {
return anyhow!("request failed with status {}", failure.status);
}
anyhow!(
"request failed with status {}: {}",
failure.status,
failure.body
)
}
pub async fn classify_http_response(
response: fabro_http::Response,
) -> Result<std::result::Result<fabro_http::Response, ApiError>> {
if response.status().is_success() {
return Ok(Ok(response));
}
let status = response.status();
let headers = response.headers().clone();
let body = response.text().await.unwrap_or_default();
let mut code = None;
if let Ok(value) = serde_json::from_str::<serde_json::Value>(&body) {
let (_, parsed_code) = parse_error_response_value(&value);
code = parsed_code;
}
Ok(Err(ApiError {
status,
headers,
body,
failure: ApiFailure { status, code },
}))
}
pub fn is_not_found_error<E>(err: &progenitor_client::Error<E>) -> bool
where
E: serde::Serialize + std::fmt::Debug,
{
match err {
progenitor_client::Error::ErrorResponse(response) => {
response.status() == fabro_http::StatusCode::NOT_FOUND
}
progenitor_client::Error::UnexpectedResponse(response) => {
response.status() == fabro_http::StatusCode::NOT_FOUND
}
_ => false,
}
}
pub fn convert_type<TInput, TOutput>(value: TInput) -> Result<TOutput>
where
TInput: serde::Serialize,
TOutput: DeserializeOwned,
{
serde_json::from_value(serde_json::to_value(value)?).map_err(Into::into)
}

26
lib/crates/fabro-client/src/lib.rs Normal file
View file

@ -0,0 +1,26 @@
//! Typed HTTP client for the Fabro API.
//!
//! This crate hosts the reusable client and auth/session plumbing that was
//! previously embedded in `fabro-cli`.
pub mod auth_store;
pub mod client;
pub mod credential;
pub mod error;
pub mod loopback;
pub mod session;
pub mod sse;
pub mod target;
pub use auth_store::{AuthEntry, AuthStore, AuthStoreError, LockError, StoredSubject};
pub use client::{Client, RunEventStream, TransportConnector};
pub use credential::{Credential, CredentialFallback};
pub use error::{
ApiError, ApiFailure, StructuredApiError, classify_api_error, classify_http_response,
convert_type, is_not_found_error, map_api_error, parse_error_response_value,
raw_response_failure_error,
};
pub use fabro_api::types;
pub use loopback::{LoopbackClassification, TargetSchemeError};
pub use session::OAuthSession;
pub use target::ServerTarget;

64
lib/crates/fabro-cli/src/loopback_target.rs → lib/crates/fabro-client/src/loopback.rs
View file

@ -2,10 +2,10 @@ use std::net::IpAddr;
use thiserror::Error;
use crate::user_config::{self, ServerTarget};
use crate::target::ServerTarget;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum LoopbackClassification {
pub enum LoopbackClassification {
Https,
LoopbackHttp,
UnixSocket,
@ -13,7 +13,7 @@ pub(crate) enum LoopbackClassification {
}
#[derive(Debug, Error)]
pub(crate) enum TargetSchemeError {
pub enum TargetSchemeError {
#[error("invalid server URL `{value}`: {reason}")]
InvalidUrl { value: String, reason: String },
#[error("unsupported server URL scheme `{scheme}`")]
@ -22,22 +22,25 @@ pub(crate) enum TargetSchemeError {
MissingHost { value: String },
}
pub(crate) fn is_loopback_or_unix_socket(
pub(crate) fn classify_target(
target: &ServerTarget,
) -> Result<LoopbackClassification, TargetSchemeError> {
match target {
ServerTarget::UnixSocket(_) => Ok(LoopbackClassification::UnixSocket),
ServerTarget::HttpUrl(api_url) => classify_http_target(api_url),
if target.is_unix_socket() {
Ok(LoopbackClassification::UnixSocket)
} else if let Some(api_url) = target.as_http_url() {
classify_http_target(api_url)
} else {
Err(TargetSchemeError::MissingHost {
value: target.to_string(),
})
}
}
fn classify_http_target(api_url: &str) -> Result<LoopbackClassification, TargetSchemeError> {
let normalized = user_config::normalized_http_base_url(api_url);
let url =
fabro_http::Url::parse(normalized).map_err(|source| TargetSchemeError::InvalidUrl {
value: api_url.to_string(),
reason: source.to_string(),
})?;
let url = fabro_http::Url::parse(api_url).map_err(|source| TargetSchemeError::InvalidUrl {
value: api_url.to_string(),
reason: source.to_string(),
})?;
match url.scheme() {
"https" => Ok(LoopbackClassification::Https),
@ -50,7 +53,7 @@ fn classify_http_target(api_url: &str) -> Result<LoopbackClassification, TargetS
if !url.username().is_empty() || url.password().is_some() {
return Ok(LoopbackClassification::Rejected);
}
let Some(authority) = raw_authority(normalized) else {
let Some(authority) = raw_authority(api_url) else {
return Err(TargetSchemeError::MissingHost {
value: api_url.to_string(),
});
@ -121,36 +124,36 @@ fn ip_is_loopback(ip: &IpAddr) -> bool {
mod tests {
use std::path::PathBuf;
use super::{LoopbackClassification, is_loopback_or_unix_socket};
use crate::user_config::ServerTarget;
use super::LoopbackClassification;
use crate::target::ServerTarget;
#[test]
fn classifies_https_loopback_and_unix_targets() {
let cases = [
(
ServerTarget::HttpUrl("https://fabro.example.com".to_string()),
ServerTarget::http_url("https://fabro.example.com").unwrap(),
LoopbackClassification::Https,
),
(
ServerTarget::HttpUrl("http://127.0.0.1:3000".to_string()),
ServerTarget::http_url("http://127.0.0.1:3000").unwrap(),
LoopbackClassification::LoopbackHttp,
),
(
ServerTarget::HttpUrl("http://[::1]:3000".to_string()),
ServerTarget::http_url("http://[::1]:3000").unwrap(),
LoopbackClassification::LoopbackHttp,
),
(
ServerTarget::HttpUrl("http://[::ffff:127.0.0.1]:3000".to_string()),
ServerTarget::http_url("http://[::ffff:127.0.0.1]:3000").unwrap(),
LoopbackClassification::LoopbackHttp,
),
(
ServerTarget::UnixSocket(PathBuf::from("/tmp/fabro.sock")),
ServerTarget::unix_socket_path("/tmp/fabro.sock").unwrap(),
LoopbackClassification::UnixSocket,
),
];
for (target, expected) in cases {
assert_eq!(is_loopback_or_unix_socket(&target).unwrap(), expected);
assert_eq!(target.loopback_classification().unwrap(), expected);
}
}
@ -167,18 +170,23 @@ mod tests {
];
for api_url in cases {
let target = ServerTarget::HttpUrl(api_url.to_string());
let target = ServerTarget::http_url(api_url).unwrap();
assert_eq!(
is_loopback_or_unix_socket(&target).unwrap(),
target.loopback_classification().unwrap(),
LoopbackClassification::Rejected
);
}
}
#[test]
fn rejects_unsupported_schemes() {
let target = ServerTarget::HttpUrl("ftp://fabro.example.com".to_string());
let error = is_loopback_or_unix_socket(&target).unwrap_err();
assert!(error.to_string().contains("unsupported server URL scheme"));
fn rejects_non_http_server_targets_at_parse_time() {
let error = "ftp://fabro.example.com"
.parse::<ServerTarget>()
.unwrap_err();
assert!(
error
.to_string()
.contains("server target must be an http(s) URL or absolute Unix socket path")
);
}
}

48
lib/crates/fabro-client/src/session.rs Normal file
View file

@ -0,0 +1,48 @@
use std::fmt;
use std::sync::Arc;
use crate::{AuthStore, Credential, CredentialFallback, ServerTarget};
#[derive(Clone)]
pub struct OAuthSession {
pub target: ServerTarget,
pub auth_store: AuthStore,
pub fallback: Option<Arc<dyn CredentialFallback>>,
}
impl OAuthSession {
#[must_use]
pub fn new(target: ServerTarget, auth_store: AuthStore) -> Self {
Self {
target,
auth_store,
fallback: None,
}
}
#[must_use]
pub fn with_fallback(mut self, fallback: Arc<dyn CredentialFallback>) -> Self {
self.fallback = Some(fallback);
self
}
#[must_use]
pub fn resolve_fallback(&self) -> Option<Credential> {
self.fallback
.as_ref()
.and_then(|fallback| fallback.resolve())
}
}
impl fmt::Debug for OAuthSession {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("OAuthSession")
.field("target", &self.target)
.field("auth_store", &self.auth_store)
.field(
"fallback",
&self.fallback.as_ref().map(|_| "<credential fallback>"),
)
.finish()
}
}

2
lib/crates/fabro-cli/src/sse.rs → lib/crates/fabro-client/src/sse.rs
View file

@ -1,4 +1,4 @@
pub(crate) fn drain_sse_payloads(buffer: &mut Vec<u8>, finalize: bool) -> Vec<String> {
pub fn drain_sse_payloads(buffer: &mut Vec<u8>, finalize: bool) -> Vec<String> {
let mut payloads = Vec::new();
while let Some(pos) = buffer.iter().position(|byte| *byte == b'\n') {

220
lib/crates/fabro-client/src/target.rs Normal file
View file

@ -0,0 +1,220 @@
use std::fmt;
use std::path::{Component, Path, PathBuf};
use std::str::FromStr;
use anyhow::{Result, bail};
use crate::loopback::{LoopbackClassification, TargetSchemeError, classify_target};
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum ServerTarget {
HttpUrl(CanonicalHttpUrl),
UnixSocket(CanonicalUnixSocketPath),
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct CanonicalHttpUrl(String);
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct CanonicalUnixSocketPath(PathBuf);
impl ServerTarget {
pub fn http_url(value: impl AsRef<str>) -> Result<Self> {
Ok(Self::HttpUrl(CanonicalHttpUrl::new(value.as_ref())?))
}
pub fn unix_socket_path(path: impl AsRef<Path>) -> Result<Self> {
Ok(Self::UnixSocket(CanonicalUnixSocketPath::new(
path.as_ref(),
)?))
}
#[must_use]
pub fn as_http_url(&self) -> Option<&str> {
match self {
Self::HttpUrl(url) => Some(url.as_str()),
Self::UnixSocket(_) => None,
}
}
#[must_use]
pub fn as_unix_socket_path(&self) -> Option<&Path> {
match self {
Self::HttpUrl(_) => None,
Self::UnixSocket(path) => Some(path.as_path()),
}
}
#[must_use]
pub fn is_unix_socket(&self) -> bool {
matches!(self, Self::UnixSocket(_))
}
pub fn build_public_http_client(&self) -> Result<(fabro_http::HttpClient, String)> {
if let Some(api_url) = self.as_http_url() {
let http_client = fabro_http::HttpClientBuilder::new().build()?;
return Ok((http_client, api_url.to_string()));
}
let Some(path) = self.as_unix_socket_path() else {
bail!("server target must be an http(s) URL or absolute Unix socket path");
};
#[cfg(unix)]
{
let http_client = fabro_http::HttpClientBuilder::new()
.unix_socket(path)
.no_proxy()
.build()?;
Ok((http_client, "http://fabro".to_string()))
}
#[cfg(not(unix))]
{
let _ = path;
bail!("Unix-socket HTTP client is not supported on this platform")
}
}
pub fn loopback_classification(&self) -> Result<LoopbackClassification, TargetSchemeError> {
classify_target(self)
}
}
impl fmt::Display for ServerTarget {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
if let Some(api_url) = self.as_http_url() {
return f.write_str(api_url);
}
let Some(path) = self.as_unix_socket_path() else {
return Err(fmt::Error);
};
write!(f, "unix://{}", path.display())
}
}
impl FromStr for ServerTarget {
type Err = anyhow::Error;
fn from_str(value: &str) -> Result<Self, Self::Err> {
if value.starts_with("http://") || value.starts_with("https://") {
return Self::http_url(value);
}
let path = Path::new(value);
if path.is_absolute() {
return Self::unix_socket_path(path);
}
bail!("server target must be an http(s) URL or absolute Unix socket path")
}
}
impl CanonicalHttpUrl {
fn new(value: &str) -> Result<Self> {
Ok(Self(canonical_http_url(value)?))
}
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
}
impl CanonicalUnixSocketPath {
fn new(path: &Path) -> Result<Self> {
let normalized = lexical_normalize_absolute_path(path)?;
Ok(Self(normalized))
}
#[must_use]
pub fn as_path(&self) -> &Path {
&self.0
}
}
fn canonical_http_url(value: &str) -> Result<String> {
let trimmed = value.trim();
let normalized = trim_api_path_suffix(trimmed);
let url = fabro_http::Url::parse(normalized).map_err(|_| {
anyhow::anyhow!("server target must be an http(s) URL or absolute Unix socket path")
})?;
let scheme = url.scheme().to_ascii_lowercase();
let default_port = match scheme.as_str() {
"http" => 80,
"https" => 443,
_ => bail!("server target must be an http(s) URL or absolute Unix socket path"),
};
let Some(host) = url.host_str() else {
bail!("server target must be an http(s) URL or absolute Unix socket path");
};
let host = host.to_ascii_lowercase();
let Some(port) = url.port_or_known_default() else {
bail!("server target must be an http(s) URL or absolute Unix socket path");
};
if port == default_port {
Ok(format!("{scheme}://{host}"))
} else {
Ok(format!("{scheme}://{host}:{port}"))
}
}
fn trim_api_path_suffix(value: &str) -> &str {
let trimmed = value.trim_end_matches('/');
trimmed.strip_suffix("/api/v1").unwrap_or(trimmed)
}
fn lexical_normalize_absolute_path(path: &Path) -> Result<PathBuf> {
if !path.is_absolute() {
bail!("server target must be an http(s) URL or absolute Unix socket path");
}
let mut normalized = PathBuf::new();
for component in path.components() {
match component {
Component::Prefix(prefix) => normalized.push(prefix.as_os_str()),
Component::RootDir => normalized.push(component.as_os_str()),
Component::CurDir => {}
Component::ParentDir => {
let _ = normalized.pop();
}
Component::Normal(part) => normalized.push(part),
}
}
Ok(normalized)
}
#[cfg(test)]
mod tests {
use std::path::PathBuf;
use super::ServerTarget;
#[test]
fn canonicalizes_http_targets_at_construction() {
let target = ServerTarget::http_url("https://EXAMPLE.COM:443/api/v1/").unwrap();
assert_eq!(target.as_http_url(), Some("https://example.com"));
assert_eq!(target.to_string(), "https://example.com");
}
#[test]
fn canonicalizes_http_targets_by_rebuilding_authority() {
let target = ServerTarget::http_url("http://Example.com:3000/nested/path").unwrap();
assert_eq!(target.as_http_url(), Some("http://example.com:3000"));
}
#[test]
fn lexically_normalizes_unix_socket_paths_without_fs_access() {
let target = ServerTarget::unix_socket_path("/tmp/fabro/../fabro.sock").unwrap();
assert_eq!(
target.as_unix_socket_path(),
Some(PathBuf::from("/tmp/fabro.sock").as_path())
);
}
}

2
lib/crates/fabro-retro/src/retro_agent.rs
View file

@ -308,7 +308,7 @@ async fn upload_data_files(
let progress_content = {
let lines: Vec<String> = events
.iter()
.filter_map(|env| serde_json::to_string(env.payload.as_value()).ok())
.filter_map(|env| serde_json::to_string(&env.event).ok())
.collect();
if lines.is_empty() {
None

48
lib/crates/fabro-server/src/server.rs
View file

@ -885,9 +885,7 @@ fn start_optional_slack_service(state: &Arc<AppState>) {
loop {
match rx.recv().await {
Ok(envelope) => {
if let Ok(event) = RunEvent::try_from(&envelope.payload) {
event_service.handle_event(&event).await;
}
event_service.handle_event(&envelope.event).await;
}
Err(RecvError::Lagged(_)) => {}
Err(RecvError::Closed) => break,
@ -1553,7 +1551,7 @@ struct PrunePlan {
reason = "sync helper invoked from async handler via spawn_blocking (see callers at :1301 / :1341)"
)]
fn build_disk_usage_response(
summaries: &[fabro_store::RunSummary],
summaries: &[fabro_types::RunSummary],
storage_dir: &std::path::Path,
verbose: bool,
) -> anyhow::Result<DiskUsageResponse> {
@ -1626,7 +1624,7 @@ fn build_disk_usage_response(
fn build_prune_plan(
request: &PruneRunsRequest,
summaries: &[fabro_store::RunSummary],
summaries: &[fabro_types::RunSummary],
storage_dir: &std::path::Path,
) -> anyhow::Result<PrunePlan> {
let scratch_base_dir = scratch_base(storage_dir);
@ -1772,16 +1770,7 @@ fn event_matches_run_filter(event: &EventEnvelope, run_filter: Option<&HashSet<R
let Some(run_filter) = run_filter else {
return true;
};
let Some(run_id) = event
.payload
.as_value()
.get("run_id")
.and_then(serde_json::Value::as_str)
.and_then(|value| value.parse::<RunId>().ok())
else {
return false;
};
run_filter.contains(&run_id)
run_filter.contains(&event.event.run_id)
}
fn sse_event_from_store(event: &EventEnvelope) -> Option<Event> {
@ -1791,11 +1780,8 @@ fn sse_event_from_store(event: &EventEnvelope) -> Option<Event> {
}
fn attach_event_is_terminal(event: &EventEnvelope) -> bool {
let Ok(run_event) = RunEvent::try_from(&event.payload) else {
return false;
};
matches!(
run_event.body,
&event.event.body,
EventBody::RunCompleted(_) | EventBody::RunFailed(_)
)
}
@ -2771,7 +2757,7 @@ fn elapsed_secs(duration_ms: Option<u64>) -> Option<f64> {
duration_ms.map(|ms| ms as f64 / 1000.0)
}
fn summary_to_api_run_summary(summary: fabro_store::RunSummary) -> serde_json::Value {
fn summary_to_api_run_summary(summary: fabro_types::RunSummary) -> serde_json::Value {
let goal = summary.goal.unwrap_or_default();
let title = truncate_goal(&goal);
let repository = repository_name(summary.host_repo_path.as_deref());
@ -3302,9 +3288,13 @@ fn octet_stream_response(bytes: Bytes) -> Response {
reason = "Stored event conversion surfaces HTTP errors directly."
)]
fn api_event_envelope_from_store(event: &EventEnvelope) -> Result<ApiEventEnvelope, Response> {
// The payload is already a serde_json::Value; merge `seq` into it
// instead of serializing the whole envelope and re-parsing.
let mut obj = event.payload.as_value().clone();
let mut obj = event.event.to_value().map_err(|err| {
ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
format!("Failed to serialize stored event: {err}"),
)
.into_response()
})?;
if let serde_json::Value::Object(ref mut map) = obj {
map.insert("seq".into(), serde_json::Value::from(event.seq));
}
@ -3578,11 +3568,9 @@ async fn forward_run_events_to_global(
loop {
match run_events.recv().await {
Ok(event) => {
if let Ok(run_event) = RunEvent::try_from(&event.payload) {
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
reconcile_live_interview_state_for_event(managed_run, &run_event);
}
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
reconcile_live_interview_state_for_event(managed_run, &event.event);
}
let _ = state.global_event_tx.send(event);
}
@ -8767,8 +8755,8 @@ slug = "fabro"
let run_store = state.store.open_run_reader(&run_id).await.unwrap();
let events = run_store.list_events().await.unwrap();
let created = events[0].payload.as_value();
let submitted = events[1].payload.as_value();
let created = events[0].event.to_value().unwrap();
let submitted = events[1].event.to_value().unwrap();
let manifest_blob = created["properties"]["manifest_blob"]
.as_str()
.expect("run.created should carry manifest_blob")

8
lib/crates/fabro-store/src/lib.rs
View file

@ -9,13 +9,15 @@ mod types;
pub use artifact_store::{ArtifactStore, NodeArtifact};
pub use error::{Error, Result};
pub use fabro_types::{RunBlobId, StageId};
pub use run_state::{NodeState, PendingInterviewRecord, RunProjection};
pub use fabro_types::{
EventEnvelope, NodeState, PendingInterviewRecord, RunBlobId, RunProjection, StageId,
};
pub use run_state::RunProjectionReducer;
pub use slate::{
AuthCode, ConsumeOutcome, Database, RefreshToken, RunDatabase, Runs, SlateAuthCodeStore,
SlateAuthTokenStore,
};
pub use types::{EventEnvelope, EventPayload, RunSummary};
pub use types::EventPayload;
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct ListRunsQuery {

387
lib/crates/fabro-store/src/run_state.rs
View file

@ -9,56 +9,13 @@ use fabro_types::run_event::{
};
use fabro_types::{
BilledModelUsage, BlockedReason, Checkpoint, Conclusion, EventBody, FailureSignature,
InterviewQuestionRecord, InterviewQuestionType, NodeStatusRecord, Outcome, PullRequestRecord,
Retro, RunControlAction, RunEvent, RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord,
StageStatus, StartRecord, StatusReason,
InterviewQuestionRecord, InterviewQuestionType, NodeStatusRecord, Outcome,
PendingInterviewRecord, PullRequestRecord, RunControlAction, RunId, RunProjection, RunRecord,
RunStatus, RunStatusRecord, RunSummary, SandboxRecord, StageStatus, StartRecord, StatusReason,
};
use serde_json::Value;
use crate::{Error, EventEnvelope, Result, RunSummary, StageId};
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct RunProjection {
pub run: Option<RunRecord>,
pub graph_source: Option<String>,
pub start: Option<StartRecord>,
pub status: Option<RunStatusRecord>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub prior_status: Option<RunStatus>,
pub pending_control: Option<RunControlAction>,
pub checkpoint: Option<Checkpoint>,
pub checkpoints: Vec<(u32, Checkpoint)>,
pub conclusion: Option<Conclusion>,
pub retro: Option<Retro>,
pub retro_prompt: Option<String>,
pub retro_response: Option<String>,
pub sandbox: Option<SandboxRecord>,
pub final_patch: Option<String>,
pub pull_request: Option<PullRequestRecord>,
pub pending_interviews: BTreeMap<String, PendingInterviewRecord>,
nodes: HashMap<StageId, NodeState>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct PendingInterviewRecord {
pub question: InterviewQuestionRecord,
pub started_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct NodeState {
pub prompt: Option<String>,
pub response: Option<String>,
pub status: Option<NodeStatusRecord>,
pub provider_used: Option<serde_json::Value>,
pub diff: Option<String>,
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub stdout: Option<String>,
pub stderr: Option<String>,
}
use crate::{Error, EventEnvelope, Result};
#[derive(Debug, Clone, Default)]
pub(crate) struct EventProjectionCache {
@ -66,8 +23,16 @@ pub(crate) struct EventProjectionCache {
pub state: RunProjection,
}
impl RunProjection {
pub fn apply_events(events: &[EventEnvelope]) -> Result<Self> {
pub trait RunProjectionReducer {
fn apply_events(events: &[EventEnvelope]) -> Result<Self>
where
Self: Sized;
fn apply_event(&mut self, event: &EventEnvelope) -> Result<()>;
}
impl RunProjectionReducer for RunProjection {
fn apply_events(events: &[EventEnvelope]) -> Result<Self> {
let mut state = Self::default();
for event in events {
state.apply_event(event)?;
@ -75,9 +40,8 @@ impl RunProjection {
Ok(state)
}
pub fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> {
let stored = RunEvent::from_ref(event.payload.as_value())
.map_err(|err| Error::InvalidEvent(format!("invalid stored event: {err}")))?;
fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> {
let stored = &event.event;
let ts = stored.ts;
let run_id = stored.run_id;
@ -420,107 +384,57 @@ impl RunProjection {
Ok(())
}
}
pub fn node(&self, node: &StageId) -> Option<&NodeState> {
self.nodes.get(node)
}
pub fn iter_nodes(&self) -> impl Iterator<Item = (&StageId, &NodeState)> {
self.nodes.iter()
}
pub fn is_empty(&self) -> bool {
self.nodes.is_empty()
}
pub fn set_node(&mut self, node: StageId, state: NodeState) {
self.nodes.insert(node, state);
}
pub fn list_node_visits(&self, node_id: &str) -> Vec<u32> {
let mut visits = self
.nodes
.keys()
.filter(|node| node.node_id() == node_id)
.map(StageId::visit)
.collect::<Vec<_>>();
visits.sort_unstable();
visits.dedup();
visits
}
pub(crate) fn build_summary(&self, run_id: &RunId) -> RunSummary {
let workflow_name = self.run.as_ref().map(|run| {
if run.graph.name.is_empty() {
"unnamed".to_string()
} else {
run.graph.name.clone()
}
});
let goal = self.run.as_ref().and_then(|run| {
let goal = run.graph.goal();
(!goal.is_empty()).then(|| goal.to_string())
});
RunSummary {
run_id: *run_id,
workflow_name,
workflow_slug: self.run.as_ref().and_then(|run| run.workflow_slug.clone()),
goal,
labels: self
.run
.as_ref()
.map(|run| run.labels.clone())
.unwrap_or_default(),
host_repo_path: self.run.as_ref().and_then(|run| run.host_repo_path.clone()),
start_time: self.start.as_ref().map(|start| start.start_time),
status: self
.status
.as_ref()
.map_or(RunStatus::Submitted, |status| status.status),
status_reason: self.status.as_ref().and_then(|status| status.status_reason),
blocked_reason: self
.status
.as_ref()
.and_then(|status| status.blocked_reason),
pending_control: self.pending_control,
duration_ms: self
.conclusion
.as_ref()
.map(|conclusion| conclusion.duration_ms),
total_usd_micros: self
.conclusion
.as_ref()
.and_then(|conclusion| conclusion.billing.as_ref())
.and_then(|billing| billing.total_usd_micros),
pub(crate) fn build_summary(state: &RunProjection, run_id: &RunId) -> RunSummary {
let workflow_name = state.run.as_ref().map(|run| {
if run.graph.name.is_empty() {
"unnamed".to_string()
} else {
run.graph.name.clone()
}
}
fn node_mut(&mut self, node_id: &str, visit: u32) -> &mut NodeState {
self.nodes.entry(StageId::new(node_id, visit)).or_default()
}
fn current_visit_for(&self, node_id: &str) -> Option<u32> {
self.nodes
.keys()
.filter(|node| node.node_id() == node_id)
.map(StageId::visit)
.max()
}
fn reset_for_rewind(&mut self) {
self.status = None;
self.pending_control = None;
self.checkpoint = None;
self.checkpoints.clear();
self.conclusion = None;
self.retro = None;
self.retro_prompt = None;
self.retro_response = None;
self.sandbox = None;
self.final_patch = None;
self.pull_request = None;
self.pending_interviews.clear();
self.nodes.clear();
});
let goal = state.run.as_ref().and_then(|run| {
let goal = run.graph.goal();
(!goal.is_empty()).then(|| goal.to_string())
});
RunSummary {
run_id: *run_id,
workflow_name,
workflow_slug: state.run.as_ref().and_then(|run| run.workflow_slug.clone()),
goal,
labels: state
.run
.as_ref()
.map(|run| run.labels.clone())
.unwrap_or_default(),
host_repo_path: state
.run
.as_ref()
.and_then(|run| run.host_repo_path.clone()),
start_time: state.start.as_ref().map(|start| start.start_time),
status: state
.status
.as_ref()
.map_or(RunStatus::Submitted, |status| status.status),
status_reason: state
.status
.as_ref()
.and_then(|status| status.status_reason),
blocked_reason: state
.status
.as_ref()
.and_then(|status| status.blocked_reason),
pending_control: state.pending_control,
duration_ms: state
.conclusion
.as_ref()
.map(|conclusion| conclusion.duration_ms),
total_usd_micros: state
.conclusion
.as_ref()
.and_then(|conclusion| conclusion.billing.as_ref())
.and_then(|billing| billing.total_usd_micros),
}
}
@ -700,13 +614,13 @@ mod tests {
};
use fabro_types::settings::SettingsLayer;
use fabro_types::{
Checkpoint, EventBody, InterviewQuestionType, RunBlobId, RunControlAction, RunEvent,
fixtures,
Checkpoint, EventBody, InterviewQuestionType, NodeState, RunBlobId, RunControlAction,
RunEvent, fixtures,
};
use serde_json::json;
use super::{NodeState, RunProjection};
use crate::{EventEnvelope, EventPayload, StageId};
use super::{RunProjection, RunProjectionReducer, build_summary};
use crate::{EventEnvelope, StageId};
fn test_event(seq: u32, body: EventBody, node_id: Option<&str>) -> EventEnvelope {
let event = RunEvent {
@ -725,11 +639,7 @@ mod tests {
body,
};
EventEnvelope {
seq,
payload: EventPayload::new(serde_json::to_value(event).unwrap(), &fixtures::RUN_1)
.unwrap(),
}
EventEnvelope { seq, event }
}
fn test_raw_event(
@ -740,17 +650,14 @@ mod tests {
) -> EventEnvelope {
EventEnvelope {
seq,
payload: EventPayload::new(
json!({
"id": format!("evt-{seq}"),
"ts": Utc::now().to_rfc3339(),
"run_id": fixtures::RUN_1,
"event": event,
"node_id": node_id,
"properties": properties,
}),
&fixtures::RUN_1,
)
event: RunEvent::from_value(json!({
"id": format!("evt-{seq}"),
"ts": Utc::now().to_rfc3339(),
"run_id": fixtures::RUN_1,
"event": event,
"node_id": node_id,
"properties": properties,
}))
.unwrap(),
}
}
@ -813,23 +720,21 @@ mod tests {
#[test]
fn set_node_round_trips_through_json() {
let mut state = RunProjection {
pending_control: Some(RunControlAction::Unpause),
checkpoints: vec![(7, Checkpoint {
timestamp: "2026-04-07T12:00:00Z".parse().unwrap(),
current_node: "build".to_string(),
completed_nodes: vec!["build".to_string()],
node_retries: HashMap::new(),
context_values: HashMap::new(),
node_outcomes: HashMap::new(),
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::from([("build".to_string(), 2usize)]),
})],
..RunProjection::default()
};
let mut state = RunProjection::default();
state.pending_control = Some(RunControlAction::Unpause);
state.checkpoints = vec![(7, Checkpoint {
timestamp: "2026-04-07T12:00:00Z".parse().unwrap(),
current_node: "build".to_string(),
completed_nodes: vec!["build".to_string()],
node_retries: HashMap::new(),
context_values: HashMap::new(),
node_outcomes: HashMap::new(),
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::from([("build".to_string(), 2usize)]),
})];
state.set_node(StageId::new("build", 2), NodeState {
stdout: Some("done".to_string()),
..NodeState::default()
@ -955,7 +860,7 @@ mod tests {
assert_eq!(status_json["status_reason"], serde_json::Value::Null);
assert_eq!(status_json["blocked_reason"], "human_input_required");
let summary = state.build_summary(&fixtures::RUN_1);
let summary = build_summary(&state, &fixtures::RUN_1);
let summary_json = serde_json::to_value(summary).unwrap();
assert_eq!(summary_json["status"], "paused");
assert_eq!(summary_json["status_reason"], serde_json::Value::Null);
@ -1053,25 +958,23 @@ mod tests {
#[test]
fn summary_synthesizes_submitted_when_run_exists_without_status() {
let state = RunProjection {
run: Some(fabro_types::RunRecord {
run_id: fixtures::RUN_1,
settings: SettingsLayer::default(),
graph: fabro_types::Graph::new("test"),
workflow_slug: Some("test".to_string()),
working_directory: std::path::PathBuf::from("/tmp/run"),
host_repo_path: Some("/tmp/repo".to_string()),
repo_origin_url: None,
base_branch: None,
labels: HashMap::new(),
provenance: None,
manifest_blob: None,
definition_blob: None,
}),
..RunProjection::default()
};
let mut state = RunProjection::default();
state.run = Some(fabro_types::RunRecord {
run_id: fixtures::RUN_1,
settings: SettingsLayer::default(),
graph: fabro_types::Graph::new("test"),
workflow_slug: Some("test".to_string()),
working_directory: std::path::PathBuf::from("/tmp/run"),
host_repo_path: Some("/tmp/repo".to_string()),
repo_origin_url: None,
base_branch: None,
labels: HashMap::new(),
provenance: None,
manifest_blob: None,
definition_blob: None,
});
let summary_json = serde_json::to_value(state.build_summary(&fixtures::RUN_1)).unwrap();
let summary_json = serde_json::to_value(build_summary(&state, &fixtures::RUN_1)).unwrap();
assert_eq!(summary_json["status"], "submitted");
}
@ -1082,45 +985,39 @@ mod tests {
RunBlobId::new(br#"{"version":1,"workflow_path":"workflow.fabro"}"#).to_string();
let events = vec![
EventEnvelope {
seq: 1,
payload: EventPayload::new(
json!({
"id": "evt-run-created",
"ts": "2026-04-07T12:00:00Z",
"run_id": fixtures::RUN_1,
"event": "run.created",
"properties": {
"settings": SettingsLayer::default(),
"graph": {
"name": "test",
"nodes": {},
"edges": [],
"attrs": {}
},
"labels": {},
"run_dir": "/tmp/run",
"working_directory": "/tmp/run",
"manifest_blob": manifest_blob
}
}),
&fixtures::RUN_1,
)
seq: 1,
event: RunEvent::from_value(json!({
"id": "evt-run-created",
"ts": "2026-04-07T12:00:00Z",
"run_id": fixtures::RUN_1,
"event": "run.created",
"properties": {
"settings": SettingsLayer::default(),
"graph": {
"name": "test",
"nodes": {},
"edges": [],
"attrs": {}
},
"labels": {},
"run_dir": "/tmp/run",
"working_directory": "/tmp/run",
"manifest_blob": manifest_blob
}
}))
.unwrap(),
},
EventEnvelope {
seq: 2,
payload: EventPayload::new(
json!({
"id": "evt-run-submitted",
"ts": "2026-04-07T12:00:01Z",
"run_id": fixtures::RUN_1,
"event": "run.submitted",
"properties": {
"definition_blob": definition_blob
}
}),
&fixtures::RUN_1,
)
seq: 2,
event: RunEvent::from_value(json!({
"id": "evt-run-submitted",
"ts": "2026-04-07T12:00:01Z",
"run_id": fixtures::RUN_1,
"event": "run.submitted",
"properties": {
"definition_blob": definition_blob
}
}))
.unwrap(),
},
];
@ -1130,11 +1027,11 @@ mod tests {
assert_eq!(
value["run"]["manifest_blob"],
events[0].payload.as_value()["properties"]["manifest_blob"]
events[0].event.properties().unwrap()["manifest_blob"]
);
assert_eq!(
value["run"]["definition_blob"],
events[1].payload.as_value()["properties"]["definition_blob"]
events[1].event.properties().unwrap()["definition_blob"]
);
}

9
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -10,14 +10,15 @@ use std::time::Duration;
pub use auth_codes::{AuthCode, SlateAuthCodeStore};
pub use auth_tokens::{ConsumeOutcome, RefreshToken, SlateAuthTokenStore};
use fabro_types::RunId;
use fabro_types::{RunId, RunSummary};
use object_store::ObjectStore;
pub use run_store::RunDatabase;
use run_store::RunDatabaseInner;
use slatedb::config::{CompressionCodec, Settings};
use tokio::sync::{Mutex, OnceCell};
use crate::{Error, ListRunsQuery, Result, RunSummary, keys};
use crate::run_state::build_summary;
use crate::{Error, ListRunsQuery, Result, keys};
#[derive(Clone)]
pub struct Database {
@ -170,7 +171,7 @@ impl Database {
let mut summaries = Vec::new();
for run_id in run_ids {
if let Some(active) = self.get_active_run(&run_id).await {
summaries.push(active.state().await?.build_summary(&run_id));
summaries.push(build_summary(&active.state().await?, &run_id));
continue;
}
if !RunDatabase::has_any_events(&db, &run_id).await? {
@ -245,7 +246,7 @@ impl Runs {
pub async fn find(&self, run_id: &RunId) -> Result<Option<RunSummary>> {
match self.db.open_run_reader(run_id).await {
Ok(run_db) => Ok(Some(run_db.state().await?.build_summary(run_id))),
Ok(run_db) => Ok(Some(build_summary(&run_db.state().await?, run_id))),
Err(Error::RunNotFound(_)) => Ok(None),
Err(err) => Err(err),
}

12
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -4,14 +4,14 @@ use std::sync::atomic::{AtomicU32, Ordering};
use bytes::Bytes;
use chrono::Utc;
use fabro_types::{RunBlobId, RunId};
use fabro_types::{RunBlobId, RunEvent, RunId, RunSummary};
use futures::Stream;
use slatedb::{Db, DbRead};
use tokio::sync::{Mutex, broadcast, mpsc};
use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::run_state::EventProjectionCache;
use crate::{Error, EventEnvelope, EventPayload, Result, RunProjection, RunSummary, keys};
use crate::run_state::{EventProjectionCache, RunProjectionReducer, build_summary};
use crate::{Error, EventEnvelope, EventPayload, Result, RunProjection, keys};
const DEFAULT_EVENT_TAIL_LIMIT: usize = 1024;
#[derive(Clone)]
@ -131,7 +131,7 @@ impl RunDatabase {
{
let events = list_events_from(db, run_id, 1).await?;
let state = RunProjection::apply_events(&events)?;
Ok(state.build_summary(run_id))
Ok(build_summary(&state, run_id))
}
async fn projected_state(&self) -> Result<RunProjection> {
@ -193,7 +193,7 @@ impl RunDatabase {
let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst);
let event = EventEnvelope {
seq,
payload: payload.clone(),
event: RunEvent::try_from(payload)?,
};
self.inner
.db
@ -341,7 +341,7 @@ where
}
events.push(EventEnvelope {
seq,
payload: serde_json::from_slice(&entry.value)?,
event: serde_json::from_slice(&entry.value)?,
});
}
events.sort_by_key(|event| event.seq);

130
lib/crates/fabro-store/src/types.rs
View file

@ -1,28 +1,8 @@
use std::collections::HashMap;
use chrono::{DateTime, Utc};
use fabro_types::{BlockedReason, RunControlAction, RunEvent, RunId, RunStatus, StatusReason};
use fabro_types::{RunEvent, RunId};
use serde::{Deserialize, Serialize};
use crate::{Error, Result};
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSummary {
pub run_id: RunId,
pub workflow_name: Option<String>,
pub workflow_slug: Option<String>,
pub goal: Option<String>,
pub labels: HashMap<String, String>,
pub host_repo_path: Option<String>,
pub start_time: Option<DateTime<Utc>>,
pub status: RunStatus,
pub status_reason: Option<StatusReason>,
pub blocked_reason: Option<BlockedReason>,
pub pending_control: Option<RunControlAction>,
pub duration_ms: Option<u64>,
pub total_usd_micros: Option<i64>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct EventPayload(serde_json::Value);
@ -81,111 +61,3 @@ impl TryFrom<&EventPayload> for RunEvent {
.map_err(|err| Error::InvalidEvent(format!("invalid stored event: {err}")))
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EventEnvelope {
pub seq: u32,
#[serde(flatten)]
pub payload: EventPayload,
}
#[cfg(test)]
mod tests {
use chrono::{TimeZone, Utc};
use fabro_types::run_event::RunCompletedProps;
use fabro_types::{ActorRef, EventBody, ParallelBranchId, RunEvent, StageId, fixtures};
use super::{EventEnvelope, EventPayload};
#[test]
fn wire_event_envelope_round_trips() {
let event = RunEvent {
id: "evt_1".to_string(),
ts: Utc.with_ymd_and_hms(2026, 4, 9, 12, 0, 0).unwrap(),
run_id: fixtures::RUN_1,
node_id: Some("code".to_string()),
node_label: Some("Code".to_string()),
stage_id: Some(StageId::new("code", 1)),
parallel_group_id: None,
parallel_branch_id: None,
session_id: None,
parent_session_id: None,
tool_call_id: None,
actor: None,
body: EventBody::RunCompleted(RunCompletedProps {
duration_ms: 42,
artifact_count: 0,
status: "success".to_string(),
reason: None,
total_usd_micros: None,
final_git_commit_sha: None,
final_patch: None,
billing: None,
}),
};
let payload = EventPayload::new(event.to_value().unwrap(), &fixtures::RUN_1).unwrap();
let envelope = EventEnvelope { seq: 7, payload };
let wire = serde_json::to_value(&envelope).unwrap();
assert_eq!(wire["seq"], 7);
assert_eq!(wire["id"], "evt_1");
assert_eq!(wire["event"], "run.completed");
assert!(wire.get("payload").is_none(), "wire shape must be flat");
let parsed: EventEnvelope = serde_json::from_value(wire).unwrap();
assert_eq!(parsed, envelope);
}
#[test]
fn wire_event_envelope_round_trips_with_all_envelope_fields() {
let group = StageId::new("review", 2);
let branch = ParallelBranchId::new(group.clone(), 3);
let event = RunEvent {
id: "evt_2".to_string(),
ts: Utc.with_ymd_and_hms(2026, 4, 9, 13, 0, 0).unwrap(),
run_id: fixtures::RUN_1,
node_id: Some("review".to_string()),
node_label: Some("Review".to_string()),
stage_id: Some(StageId::new("review", 2)),
parallel_group_id: Some(group),
parallel_branch_id: Some(branch),
session_id: Some("ses_42".to_string()),
parent_session_id: Some("ses_root".to_string()),
tool_call_id: Some("tool_call_xyz".to_string()),
actor: Some(ActorRef::agent(
Some("ses_42".to_string()),
Some("claude-sonnet".to_string()),
)),
body: EventBody::RunCompleted(RunCompletedProps {
duration_ms: 100,
artifact_count: 1,
status: "success".to_string(),
reason: None,
total_usd_micros: None,
final_git_commit_sha: None,
final_patch: None,
billing: None,
}),
};
let payload = EventPayload::new(event.to_value().unwrap(), &fixtures::RUN_1).unwrap();
let envelope = EventEnvelope { seq: 99, payload };
let wire = serde_json::to_value(&envelope).unwrap();
assert_eq!(wire["seq"], 99);
assert_eq!(wire["id"], "evt_2");
assert_eq!(wire["stage_id"], "review@2");
assert_eq!(wire["parallel_group_id"], "review@2");
assert_eq!(wire["parallel_branch_id"], "review@2:3");
assert_eq!(wire["session_id"], "ses_42");
assert_eq!(wire["parent_session_id"], "ses_root");
assert_eq!(wire["tool_call_id"], "tool_call_xyz");
assert_eq!(wire["actor"]["kind"], "agent");
assert_eq!(wire["actor"]["id"], "ses_42");
assert_eq!(wire["actor"]["display"], "claude-sonnet");
assert_eq!(wire["event"], "run.completed");
assert!(wire.get("payload").is_none(), "wire shape must be flat");
let parsed: EventEnvelope = serde_json::from_value(wire).unwrap();
assert_eq!(parsed, envelope);
}
}

30
lib/crates/fabro-types/src/artifact.rs Normal file
View file

@ -0,0 +1,30 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ArtifactUpload {
pub path: String,
pub mime: String,
pub content_md5: String,
pub content_sha256: String,
pub bytes: u64,
}
#[cfg(test)]
mod tests {
use super::ArtifactUpload;
#[test]
fn round_trips_through_serde_json() {
let artifact = ArtifactUpload {
path: "artifacts/log.txt".to_string(),
mime: "text/plain".to_string(),
content_md5: "md5".to_string(),
content_sha256: "sha256".to_string(),
bytes: 42,
};
let value = serde_json::to_value(&artifact).unwrap();
let parsed: ArtifactUpload = serde_json::from_value(value).unwrap();
assert_eq!(parsed, artifact);
}
}

135
lib/crates/fabro-types/src/event_envelope.rs Normal file
View file

@ -0,0 +1,135 @@
use serde::{Deserialize, Serialize};
use crate::RunEvent;
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EventEnvelope {
pub seq: u32,
#[serde(flatten)]
pub event: RunEvent,
}
#[cfg(test)]
mod tests {
use chrono::{TimeZone, Utc};
use super::EventEnvelope;
use crate::run_event::RunCompletedProps;
use crate::{ActorRef, EventBody, ParallelBranchId, RunEvent, StageId, fixtures};
#[test]
fn wire_event_envelope_round_trips() {
let event = RunEvent {
id: "evt_1".to_string(),
ts: Utc.with_ymd_and_hms(2026, 4, 9, 12, 0, 0).unwrap(),
run_id: fixtures::RUN_1,
node_id: Some("code".to_string()),
node_label: Some("Code".to_string()),
stage_id: Some(StageId::new("code", 1)),
parallel_group_id: None,
parallel_branch_id: None,
session_id: None,
parent_session_id: None,
tool_call_id: None,
actor: None,
body: EventBody::RunCompleted(RunCompletedProps {
duration_ms: 42,
artifact_count: 0,
status: "success".to_string(),
reason: None,
total_usd_micros: None,
final_git_commit_sha: None,
final_patch: None,
billing: None,
}),
};
let envelope = EventEnvelope { seq: 7, event };
let wire = serde_json::to_value(&envelope).unwrap();
assert_eq!(wire["seq"], 7);
assert_eq!(wire["id"], "evt_1");
assert_eq!(wire["event"], "run.completed");
let parsed: EventEnvelope = serde_json::from_value(wire).unwrap();
assert_eq!(parsed, envelope);
}
#[test]
fn wire_event_envelope_round_trips_with_all_envelope_fields() {
let group = StageId::new("review", 2);
let branch = ParallelBranchId::new(group.clone(), 3);
let event = RunEvent {
id: "evt_2".to_string(),
ts: Utc.with_ymd_and_hms(2026, 4, 9, 13, 0, 0).unwrap(),
run_id: fixtures::RUN_1,
node_id: Some("review".to_string()),
node_label: Some("Review".to_string()),
stage_id: Some(StageId::new("review", 2)),
parallel_group_id: Some(group),
parallel_branch_id: Some(branch),
session_id: Some("ses_42".to_string()),
parent_session_id: Some("ses_root".to_string()),
tool_call_id: Some("tool_call_xyz".to_string()),
actor: Some(ActorRef::agent(
Some("ses_42".to_string()),
Some("claude-sonnet".to_string()),
)),
body: EventBody::RunCompleted(RunCompletedProps {
duration_ms: 100,
artifact_count: 1,
status: "success".to_string(),
reason: None,
total_usd_micros: None,
final_git_commit_sha: None,
final_patch: None,
billing: None,
}),
};
let envelope = EventEnvelope { seq: 99, event };
let wire = serde_json::to_value(&envelope).unwrap();
assert_eq!(wire["seq"], 99);
assert_eq!(wire["id"], "evt_2");
assert_eq!(wire["stage_id"], "review@2");
assert_eq!(wire["parallel_group_id"], "review@2");
assert_eq!(wire["parallel_branch_id"], "review@2:3");
assert_eq!(wire["session_id"], "ses_42");
assert_eq!(wire["parent_session_id"], "ses_root");
assert_eq!(wire["tool_call_id"], "tool_call_xyz");
assert_eq!(wire["actor"]["kind"], "agent");
assert_eq!(wire["actor"]["id"], "ses_42");
assert_eq!(wire["actor"]["display"], "claude-sonnet");
assert_eq!(wire["event"], "run.completed");
let parsed: EventEnvelope = serde_json::from_value(wire).unwrap();
assert_eq!(parsed, envelope);
}
#[test]
fn preserves_unknown_event_names_and_properties() {
let wire = serde_json::json!({
"seq": 7,
"id": "evt_unknown",
"ts": "2026-04-20T12:00:00.000Z",
"run_id": fixtures::RUN_1,
"event": "vendor.custom.event",
"properties": {
"answer": 42,
"nested": { "ok": true }
}
});
let parsed: EventEnvelope = serde_json::from_value(wire.clone()).unwrap();
let serialized = serde_json::to_value(&parsed).unwrap();
assert_eq!(serialized["seq"], wire["seq"]);
assert_eq!(serialized["id"], wire["id"]);
assert_eq!(serialized["run_id"], wire["run_id"]);
assert_eq!(serialized["event"], wire["event"]);
assert_eq!(serialized["properties"], wire["properties"]);
assert_eq!(
chrono::DateTime::parse_from_rfc3339(serialized["ts"].as_str().unwrap()).unwrap(),
chrono::DateTime::parse_from_rfc3339(wire["ts"].as_str().unwrap()).unwrap(),
);
}
}

8
lib/crates/fabro-types/src/lib.rs
View file

@ -1,10 +1,12 @@
extern crate self as fabro_types;
pub mod artifact;
pub mod auth;
pub mod billing;
pub mod blob_ref;
pub mod checkpoint;
pub mod conclusion;
pub mod event_envelope;
pub mod failure_signature;
pub mod graph;
pub mod interview;
@ -16,12 +18,15 @@ pub mod run;
pub mod run_blob_id;
pub mod run_event;
pub mod run_id;
pub mod run_projection;
pub mod run_summary;
pub mod sandbox_record;
pub mod settings;
pub mod stage_id;
pub mod start;
pub mod status;
pub use artifact::ArtifactUpload;
pub use auth::{IdpIdentity, IdpIdentityError};
pub use billing::{
AnthropicBillingFacts, AnthropicModelPricing, BilledModelUsage, BilledTokenCounts,
@ -34,6 +39,7 @@ pub use blob_ref::{
};
pub use checkpoint::Checkpoint;
pub use conclusion::{Conclusion, StageSummary};
pub use event_envelope::EventEnvelope;
pub use failure_signature::FailureSignature;
pub use graph::{AttrValue, Edge, Graph, Node, is_llm_handler_type, shape_to_handler_type};
pub use interview::{InterviewQuestionRecord, InterviewQuestionType};
@ -51,6 +57,8 @@ pub use run::{
pub use run_blob_id::RunBlobId;
pub use run_event::{ActorKind, ActorRef, EventBody, RunEvent, RunNoticeLevel};
pub use run_id::{RunId, fixtures};
pub use run_projection::{NodeState, PendingInterviewRecord, RunProjection};
pub use run_summary::RunSummary;
pub use sandbox_record::SandboxRecord;
pub use stage_id::{ParallelBranchId, StageId};
pub use start::StartRecord;

109
lib/crates/fabro-types/src/run_projection.rs Normal file
View file

@ -0,0 +1,109 @@
use std::collections::{BTreeMap, HashMap};
use chrono::{DateTime, Utc};
use crate::{
Checkpoint, Conclusion, InterviewQuestionRecord, NodeStatusRecord, PullRequestRecord, Retro,
RunControlAction, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageId, StartRecord,
};
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(default)]
pub struct RunProjection {
pub run: Option<RunRecord>,
pub graph_source: Option<String>,
pub start: Option<StartRecord>,
pub status: Option<RunStatusRecord>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub prior_status: Option<RunStatus>,
pub pending_control: Option<RunControlAction>,
pub checkpoint: Option<Checkpoint>,
pub checkpoints: Vec<(u32, Checkpoint)>,
pub conclusion: Option<Conclusion>,
pub retro: Option<Retro>,
pub retro_prompt: Option<String>,
pub retro_response: Option<String>,
pub sandbox: Option<SandboxRecord>,
pub final_patch: Option<String>,
pub pull_request: Option<PullRequestRecord>,
pub pending_interviews: BTreeMap<String, PendingInterviewRecord>,
nodes: HashMap<StageId, NodeState>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct PendingInterviewRecord {
pub question: InterviewQuestionRecord,
pub started_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct NodeState {
pub prompt: Option<String>,
pub response: Option<String>,
pub status: Option<NodeStatusRecord>,
pub provider_used: Option<serde_json::Value>,
pub diff: Option<String>,
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub stdout: Option<String>,
pub stderr: Option<String>,
}
impl RunProjection {
pub fn node(&self, node: &StageId) -> Option<&NodeState> {
self.nodes.get(node)
}
pub fn iter_nodes(&self) -> impl Iterator<Item = (&StageId, &NodeState)> {
self.nodes.iter()
}
pub fn is_empty(&self) -> bool {
self.nodes.is_empty()
}
pub fn set_node(&mut self, node: StageId, state: NodeState) {
self.nodes.insert(node, state);
}
pub fn list_node_visits(&self, node_id: &str) -> Vec<u32> {
let mut visits = self
.nodes
.keys()
.filter(|node| node.node_id() == node_id)
.map(StageId::visit)
.collect::<Vec<_>>();
visits.sort_unstable();
visits.dedup();
visits
}
pub fn node_mut(&mut self, node_id: &str, visit: u32) -> &mut NodeState {
self.nodes.entry(StageId::new(node_id, visit)).or_default()
}
pub fn current_visit_for(&self, node_id: &str) -> Option<u32> {
self.nodes
.keys()
.filter(|node| node.node_id() == node_id)
.map(StageId::visit)
.max()
}
pub fn reset_for_rewind(&mut self) {
self.status = None;
self.pending_control = None;
self.checkpoint = None;
self.checkpoints.clear();
self.conclusion = None;
self.retro = None;
self.retro_prompt = None;
self.retro_response = None;
self.sandbox = None;
self.final_patch = None;
self.pull_request = None;
self.pending_interviews.clear();
self.nodes.clear();
}
}

56
lib/crates/fabro-types/src/run_summary.rs Normal file
View file

@ -0,0 +1,56 @@
use std::collections::HashMap;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::{BlockedReason, RunControlAction, RunId, RunStatus, StatusReason};
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSummary {
pub run_id: RunId,
pub workflow_name: Option<String>,
pub workflow_slug: Option<String>,
pub goal: Option<String>,
pub labels: HashMap<String, String>,
pub host_repo_path: Option<String>,
pub start_time: Option<DateTime<Utc>>,
pub status: RunStatus,
pub status_reason: Option<StatusReason>,
pub blocked_reason: Option<BlockedReason>,
pub pending_control: Option<RunControlAction>,
pub duration_ms: Option<u64>,
pub total_usd_micros: Option<i64>,
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use chrono::{TimeZone, Utc};
use super::RunSummary;
use crate::{BlockedReason, RunControlAction, RunStatus, StatusReason, fixtures};
#[test]
fn round_trips_through_serde_json() {
let summary = RunSummary {
run_id: fixtures::RUN_1,
workflow_name: Some("workflow".to_string()),
workflow_slug: Some("workflow".to_string()),
goal: Some("ship it".to_string()),
labels: HashMap::from([("team".to_string(), "core".to_string())]),
host_repo_path: Some("/tmp/repo".to_string()),
start_time: Some(Utc.with_ymd_and_hms(2026, 4, 20, 12, 0, 0).unwrap()),
status: RunStatus::Blocked,
status_reason: Some(StatusReason::SandboxInitializing),
blocked_reason: Some(BlockedReason::HumanInputRequired),
pending_control: Some(RunControlAction::Pause),
duration_ms: Some(42),
total_usd_micros: Some(123),
};
let value = serde_json::to_value(&summary).unwrap();
let parsed: RunSummary = serde_json::from_value(value).unwrap();
assert_eq!(parsed, summary);
}
}

22
lib/crates/fabro-workflow/src/artifact_snapshot.rs
View file

@ -2,7 +2,7 @@ use std::path::Path;
use fabro_agent::Sandbox;
use fabro_sandbox::shell_quote;
use serde::{Deserialize, Serialize};
use fabro_types::ArtifactUpload;
use sha2::{Digest, Sha256};
use tokio::fs;
use tracing::{debug, warn};
@ -15,25 +15,15 @@ pub struct DiscoveredFile {
pub mtime_epoch_secs: f64,
}
/// Metadata for a single captured artifact file.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CapturedArtifactInfo {
pub path: String,
pub mime: String,
pub content_md5: String,
pub content_sha256: String,
pub bytes: u64,
}
/// Summary of an artifact collection run.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ArtifactCollectionSummary {
pub files_copied: usize,
pub total_bytes: u64,
pub files_skipped: usize,
pub download_errors: usize,
pub hash_errors: usize,
pub captured_assets: Vec<CapturedArtifactInfo>,
pub captured_assets: Vec<ArtifactUpload>,
}
/// Directories to exclude from the find search and checkpoint commits.
@ -261,7 +251,7 @@ fn normalize_paths(discovered: Vec<DiscoveredFile>, root: &str) -> Vec<Discovere
async fn compute_artifact_info(
relative_path: &str,
local_path: &Path,
) -> std::result::Result<CapturedArtifactInfo, String> {
) -> std::result::Result<ArtifactUpload, String> {
let mime = mime_guess::from_path(relative_path)
.first_or_octet_stream()
.to_string();
@ -271,7 +261,7 @@ async fn compute_artifact_info(
let bytes = u64::try_from(data.len()).unwrap_or(u64::MAX);
let content_md5 = format!("{:x}", md5::compute(&data));
let content_sha256 = hex::encode(Sha256::digest(&data));
Ok(CapturedArtifactInfo {
Ok(ArtifactUpload {
path: relative_path.to_string(),
mime,
content_md5,
@ -308,7 +298,7 @@ pub async fn collect_artifacts(
let mut total_bytes: u64 = 0;
let mut download_errors: usize = 0;
let mut hash_errors: usize = 0;
let mut captured_assets: Vec<CapturedArtifactInfo> = Vec::new();
let mut captured_assets: Vec<ArtifactUpload> = Vec::new();
for file in &to_collect {
let dest = artifact_capture_dir.join(&file.relative_path);

6
lib/crates/fabro-workflow/src/artifact_upload.rs
View file

@ -4,9 +4,7 @@ use std::sync::Arc;
use anyhow::Result;
use async_trait::async_trait;
use fabro_store::ArtifactStore;
use fabro_types::StageId;
use crate::artifact_snapshot::CapturedArtifactInfo;
use fabro_types::{ArtifactUpload, StageId};
#[async_trait]
pub trait StageArtifactUploader: Send + Sync {
@ -14,7 +12,7 @@ pub trait StageArtifactUploader: Send + Sync {
&self,
stage_id: &StageId,
artifact_capture_dir: &Path,
artifacts: &[CapturedArtifactInfo],
artifacts: &[ArtifactUpload],
) -> Result<()>;
}

2
lib/crates/fabro-workflow/src/event.rs
View file

@ -3166,7 +3166,7 @@ mod tests {
let line = events
.into_iter()
.next()
.map(|event| event.payload.as_value().clone())
.map(|event| event.event.to_value().unwrap())
.unwrap();
assert!(line.get("id").is_some());
assert_eq!(line["event"], "run.notice");

18
lib/crates/fabro-workflow/src/lib.rs
View file

@ -92,19 +92,19 @@ pub fn build_completed_stages(cp: &records::Checkpoint, run_failed: bool) -> Vec
pub fn extract_stage_durations_from_events(events: &[EventEnvelope]) -> HashMap<String, u64> {
let mut durations = HashMap::new();
for envelope in events {
let value = envelope.payload.as_value();
let event_name = value.get("event").and_then(serde_json::Value::as_str);
if event_name != Some("stage.completed") && event_name != Some("stage.failed") {
let event = &envelope.event;
let event_name = event.event_name();
if event_name != "stage.completed" && event_name != "stage.failed" {
continue;
}
let Some(node_id) = value.get("node_id").and_then(serde_json::Value::as_str) else {
let Some(node_id) = event.node_id.as_deref() else {
continue;
};
let Some(duration_ms) = value
.get("properties")
.and_then(serde_json::Value::as_object)
.and_then(|properties| properties.get("duration_ms"))
.and_then(serde_json::Value::as_u64)
let Some(duration_ms) = event
.properties()
.ok()
.and_then(|properties| properties.get("duration_ms").cloned())
.and_then(|duration| duration.as_u64())
else {
continue;
};

10
lib/crates/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -9,12 +9,12 @@ use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, NodeDecision,
use fabro_core::outcome::NodeResult;
use fabro_core::state::ExecutionState;
use fabro_store::ArtifactStore;
use fabro_types::{RunId, StageId};
use fabro_types::{ArtifactUpload, RunId, StageId};
use tokio::fs;
use tokio::time::sleep;
use crate::artifact::{normalize_durable_updates, offload_large_values, sync_artifacts_to_env};
use crate::artifact_snapshot::{CapturedArtifactInfo, collect_artifacts};
use crate::artifact_snapshot::collect_artifacts;
use crate::artifact_upload::ArtifactSink;
use crate::event::{Emitter, Event, RunNoticeLevel};
use crate::graph::{WorkflowGraph, WorkflowNode};
@ -209,7 +209,7 @@ impl ArtifactLifecycle {
&self,
stage_id: &StageId,
artifact_capture_dir: &std::path::Path,
artifacts: &[CapturedArtifactInfo],
artifacts: &[ArtifactUpload],
) -> Result<(), String> {
let Some(sink) = self.artifact_sink.as_ref() else {
return Err("artifact sink is not configured".to_string());
@ -238,7 +238,7 @@ impl ArtifactLifecycle {
sink: &ArtifactSink,
stage_id: &StageId,
artifact_capture_dir: &std::path::Path,
artifacts: &[CapturedArtifactInfo],
artifacts: &[ArtifactUpload],
) -> Result<(), String> {
match sink {
ArtifactSink::Store(store) => {
@ -257,7 +257,7 @@ impl ArtifactLifecycle {
store: &ArtifactStore,
stage_id: &StageId,
artifact_capture_dir: &std::path::Path,
artifacts: &[CapturedArtifactInfo],
artifacts: &[ArtifactUpload],
) -> Result<(), String> {
for artifact in artifacts {
let local_path = artifact_capture_dir.join(&artifact.path);

5
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -1022,10 +1022,7 @@ mod tests {
let run_store = store.open_run_reader(&created.run_id).await.unwrap();
let events = run_store.list_events().await.unwrap();
assert_eq!(
events.first().unwrap().payload.as_value()["event"],
"run.created"
);
assert_eq!(events.first().unwrap().event.event_name(), "run.created");
}
#[tokio::test]

4
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -11,8 +11,8 @@ use anyhow::{Context, Result, bail};
use chrono::{DateTime, Utc};
use fabro_config::Storage;
use fabro_config::user::default_storage_dir;
use fabro_store::{Database, RunSummary};
use fabro_types::RunId;
use fabro_store::Database;
use fabro_types::{RunId, RunSummary};
use serde::Serialize;
use crate::operations::make_run_dir;