scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-10 22:43:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(cli): restore green server-backed test checkpoint

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-05 03:41:10 -04:00
parent 443c9f735b
commit 2b6ba07f36
40 changed files with 362 additions and 288 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
lib/crates/fabro-cli/src/commands/pr/create.rs
View file

@ -4,13 +4,14 @@ use anyhow::{Context, Result, bail};
use fabro_model::Catalog;
use fabro_sandbox::daytona::detect_repo_info;
use fabro_workflow::outcome::StageStatus;
use fabro_workflow::pull_request::maybe_open_pull_request;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use tracing::info;
use crate::args::{GlobalArgs, PrCreateArgs};
use crate::commands::store::rebuild::rebuild_run_store;
use crate::shared::print_json_pretty;
use crate::server_client;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
pub(super) async fn create_command(
@ -104,7 +105,7 @@ async fn create_from(
.model
.unwrap_or_else(|| Catalog::builtin().default_from_env().id.clone());
let record = fabro_workflow::pull_request::maybe_open_pull_request(
let record = maybe_open_pull_request(
&creds,
&origin_url,
base_branch,

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

@ -8,8 +8,8 @@ use serde::Serialize;
use tracing::info;
use crate::args::{GlobalArgs, PrListArgs};
use crate::shared::print_json_pretty;
use crate::server_client;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
#[derive(Serialize)]

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

@ -12,8 +12,8 @@ use fabro_types::PullRequestRecord;
use fabro_workflow::run_lookup::resolve_run_from_summaries;
use crate::args::{GlobalArgs, PrCommand, PrNamespace};
use crate::shared::github::build_github_app_credentials;
use crate::server_client;
use crate::shared::github::build_github_app_credentials;
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn dispatch(ns: PrNamespace, globals: &GlobalArgs) -> Result<()> {

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

@ -10,7 +10,7 @@ use fabro_types::{EventBody, RunEvent, RunId};
use fabro_api::types;
use fabro_interview::{AnswerValue, ConsoleInterviewer, Question, QuestionOption, QuestionType};
use fabro_store::{EventEnvelope, RuntimeState};
use fabro_store::EventEnvelope;
use fabro_util::json::normalize_json_value;
use fabro_util::terminal::Styles;
use fabro_workflow::outcome::StageStatus;
@ -97,8 +97,6 @@ async fn attach_run_server(
engine_child: Option<std::process::Child>,
json_output: bool,
) -> Result<ExitCode> {
let runtime_state = RuntimeState::new(run_dir);
let mut engine_guard = engine_child.map(EngineChildGuard::new);
let is_tty = std::io::stderr().is_terminal();
@ -145,12 +143,17 @@ async fn attach_run_server(
}
// Wait briefly for a terminal status or conclusion
for _ in 0..20 {
if client.get_run_state(run_id).await.ok().is_some_and(|state| {
state.conclusion.is_some()
|| state
.status
.is_some_and(|record| record.status.is_terminal())
}) {
if client
.get_run_state(run_id)
.await
.ok()
.is_some_and(|state| {
state.conclusion.is_some()
|| state
.status
.is_some_and(|record| record.status.is_terminal())
})
{
break;
}
sleep(Duration::from_millis(100)).await;
@ -201,9 +204,11 @@ async fn attach_run_server(
hide_progress(&mut progress_ui, json_output);
let interviewer = ConsoleInterviewer::new(styles);
let answer =
fabro_interview::Interviewer::ask(&interviewer, api_question_to_question(&question))
.await;
let answer = fabro_interview::Interviewer::ask(
&interviewer,
api_question_to_question(&question),
)
.await;
show_progress(&mut progress_ui, json_output);
if answer_requires_reattach(&answer) {
@ -234,14 +239,26 @@ async fn attach_run_server(
if let Some(child_alive) = child_alive_via_handle {
if !child_alive && !saw_event {
flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output)
.await?;
flush_remaining_server_events(
client,
run_id,
next_seq,
&mut progress_ui,
json_output,
)
.await?;
break;
}
} else {
if terminal_status.is_some() && !saw_event {
flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output)
.await?;
flush_remaining_server_events(
client,
run_id,
next_seq,
&mut progress_ui,
json_output,
)
.await?;
break;
}
@ -261,8 +278,14 @@ async fn attach_run_server(
}
};
if !engine_alive {
flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output)
.await?;
flush_remaining_server_events(
client,
run_id,
next_seq,
&mut progress_ui,
json_output,
)
.await?;
break;
}
}
@ -318,7 +341,13 @@ async fn submit_server_interview_answer(
}
};
client
.submit_run_answer(run_id, qid, value, selected_option_key, selected_option_keys)
.submit_run_answer(
run_id,
qid,
value,
selected_option_key,
selected_option_keys,
)
.await?;
Ok(true)
}
@ -358,7 +387,8 @@ async fn flush_remaining_server_events(
}
fn is_run_not_found_error(err: &anyhow::Error) -> bool {
err.chain().any(|cause| cause.to_string() == "Run not found.")
err.chain()
.any(|cause| cause.to_string() == "Run not found.")
}
fn emit_progress_line(
@ -407,8 +437,7 @@ fn restore_empty_run_properties(value: &mut serde_json::Value) {
let Some(event_name) = object.get("event").and_then(serde_json::Value::as_str) else {
return;
};
if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties")
{
if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") {
let run_id = object.remove("run_id");
let ts = object.remove("ts");
object.insert("properties".to_string(), serde_json::json!({}));
@ -443,28 +472,12 @@ fn infer_run_id(run_dir: &Path) -> Option<RunId> {
})
}
#[allow(clippy::struct_field_names)]
#[derive(Debug, Clone, PartialEq, Eq)]
struct InterviewPaths {
claim_path: PathBuf,
request_path: PathBuf,
response_path: PathBuf,
}
impl InterviewPaths {
fn from_runtime_state(runtime_state: &RuntimeState) -> Self {
Self {
claim_path: runtime_state.interview_claim_path(),
request_path: runtime_state.interview_request_path(),
response_path: runtime_state.interview_response_path(),
}
}
}
#[cfg(test)]
struct InterviewClaimGuard {
claim_path: PathBuf,
}
#[cfg(test)]
impl InterviewClaimGuard {
fn acquire(claim_path: &Path) -> Option<Self> {
if try_claim_interview_request(claim_path) {
@ -477,6 +490,7 @@ impl InterviewClaimGuard {
}
}
#[cfg(test)]
impl Drop for InterviewClaimGuard {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.claim_path);
@ -516,6 +530,7 @@ impl Drop for EngineChildGuard {
}
}
#[cfg(test)]
fn try_claim_interview_request(claim_path: &Path) -> bool {
if let Some(parent) = claim_path.parent() {
if std::fs::create_dir_all(parent).is_err() {
@ -549,6 +564,7 @@ fn answer_requires_reattach(answer: &fabro_interview::Answer) -> bool {
matches!(answer.value, AnswerValue::Aborted | AnswerValue::Skipped)
}
#[cfg(test)]
fn write_interview_response_atomically(
response_path: &Path,
answer: &fabro_interview::Answer,
@ -611,13 +627,13 @@ fn process_alive(pid: u32) -> bool {
fn event_exit_code(event: &EventEnvelope) -> Option<ExitCode> {
let run_event = RunEvent::try_from(&event.payload).ok()?;
match run_event.body {
EventBody::RunCompleted(props) => Some(if props.status == "success"
|| props.status == "partial_success"
{
ExitCode::from(0)
} else {
ExitCode::from(1)
}),
EventBody::RunCompleted(props) => Some(
if props.status == "success" || props.status == "partial_success" {
ExitCode::from(0)
} else {
ExitCode::from(1)
},
),
EventBody::RunFailed(_) => Some(ExitCode::from(1)),
_ => None,
}

3
lib/crates/fabro-cli/src/commands/run/detached.rs
View file

@ -3,6 +3,7 @@ use std::time::Duration;
use anyhow::{Result, anyhow};
use fabro_types::{RunId, RunStatus};
use tokio::time::sleep;
use crate::server_client;
use crate::user_config::load_user_settings;
@ -44,7 +45,7 @@ pub(crate) async fn execute(
| RunStatus::Running
| RunStatus::Paused
| RunStatus::Removing => {
tokio::time::sleep(Duration::from_millis(100)).await;
sleep(Duration::from_millis(100)).await;
}
}
}

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

@ -8,8 +8,9 @@ use fabro_workflow::sandbox_git::GIT_REMOTE;
use tracing::{debug, info};
use crate::args::{DiffArgs, GlobalArgs};
use crate::shared::print_json_pretty;
use crate::server_client;
use crate::server_client::RunProjection;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> {
@ -52,11 +53,7 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> {
Ok(())
}
async fn resolve_diff(
_run_dir: &Path,
state: &crate::server_client::RunProjection,
args: &DiffArgs,
) -> Result<String> {
async fn resolve_diff(_run_dir: &Path, state: &RunProjection, args: &DiffArgs) -> Result<String> {
if let Some(ref node_id) = args.node {
if let Some(visit) = state.list_node_visits(node_id).into_iter().max() {
if let Some(node) = state.node(&fabro_store::StageId::new(node_id, visit)) {

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

@ -2,9 +2,7 @@ use anyhow::Context;
use anyhow::Result;
use fabro_checkpoint::git::Store;
use fabro_util::terminal::Styles;
use fabro_workflow::operations::{
ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork,
};
use fabro_workflow::operations::{ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork};
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use git2::Repository;

7
lib/crates/fabro-cli/src/commands/run/launcher.rs
View file

@ -1,5 +1,6 @@
use std::path::{Path, PathBuf};
#[cfg(test)]
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use fabro_types::RunId;
@ -26,10 +27,7 @@ pub(crate) fn launcher_record_path(storage_dir: &Path, run_id: &RunId) -> PathBu
launcher_dir(storage_dir).join(format!("{run_id}.json"))
}
pub(crate) fn launcher_log_path(storage_dir: &Path, run_id: &RunId) -> PathBuf {
launcher_dir(storage_dir).join(format!("{run_id}.log"))
}
#[cfg(test)]
pub(crate) fn write_launcher_record(path: &Path, record: &LauncherRecord) -> Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
@ -76,6 +74,7 @@ pub(crate) fn launcher_record_for_run(run_dir: &Path) -> Option<LauncherRecord>
None
}
#[cfg(test)]
pub(crate) fn active_launcher_record(storage_dir: &Path, run_id: &RunId) -> Option<LauncherRecord> {
let path = launcher_record_path(storage_dir, run_id);
let launcher = read_launcher_record(&path)?;

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

@ -154,7 +154,12 @@ async fn follow_store_logs(
let mut next_seq = seq;
loop {
match time::timeout(Duration::from_millis(200), client.list_run_events(run_id, Some(next_seq), None)).await {
match time::timeout(
Duration::from_millis(200),
client.list_run_events(run_id, Some(next_seq), None),
)
.await
{
Ok(Ok(events)) => {
let saw_terminal = events
.iter()
@ -172,16 +177,20 @@ async fn follow_store_logs(
next_seq = event.seq.saturating_add(1);
}
if saw_terminal {
flush_remaining_store_events(client, run_id, next_seq, pretty, styles, &mut out)
.await?;
flush_remaining_store_events(
client, run_id, next_seq, pretty, styles, &mut out,
)
.await?;
debug!("Observed terminal event while following logs, stopping follow");
break;
}
}
Err(_) => {
if run_concluded(client, run_id).await? {
flush_remaining_store_events(client, run_id, next_seq, pretty, styles, &mut out)
.await?;
flush_remaining_store_events(
client, run_id, next_seq, pretty, styles, &mut out,
)
.await?;
debug!("Run reached terminal status, stopping follow");
break;
}
@ -248,8 +257,7 @@ fn restore_empty_run_properties(value: &mut serde_json::Value) {
let Some(event_name) = object.get("event").and_then(serde_json::Value::as_str) else {
return;
};
if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties")
{
if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") {
let run_id = object.remove("run_id");
let ts = object.remove("ts");
object.insert("properties".to_string(), serde_json::json!({}));

23
lib/crates/fabro-cli/src/commands/run/output.rs
View file

@ -9,12 +9,12 @@ use fabro_util::terminal::Styles;
use fabro_util::text::strip_goal_decoration;
use fabro_workflow::artifact_snapshot::collect_artifact_paths;
use fabro_workflow::outcome::{StageStatus, format_cost};
use fabro_workflow::pipeline::{Persisted, Validated};
use fabro_workflow::pipeline::Validated;
use fabro_workflow::records::Conclusion;
use indicatif::HumanDuration;
use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path};
use crate::server_client;
use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path};
fn print_workflow_header(
graph: &Graph,
@ -56,21 +56,6 @@ pub(crate) fn print_workflow_report(
print_workflow_header(validated.graph(), validated.diagnostics(), dot_path, styles);
}
pub(crate) fn print_workflow_report_from_persisted(
persisted: &Persisted,
dot_path: Option<&Path>,
styles: &Styles,
) {
print_workflow_header(persisted.graph(), persisted.diagnostics(), dot_path, styles);
}
pub(crate) fn print_diagnostics_from_error(
diagnostics: &[fabro_validate::Diagnostic],
styles: &Styles,
) {
print_diagnostics(diagnostics, styles);
}
pub(crate) async fn print_run_summary(
storage_dir: &Path,
run_dir: &Path,
@ -104,7 +89,7 @@ pub(crate) async fn print_run_summary(
pr_url.as_deref(),
styles,
);
print_final_output(checkpoint.as_ref(), run_dir, styles).await;
print_final_output(checkpoint.as_ref(), run_dir, styles);
print_assets(run_dir, styles);
Ok(())
}
@ -198,7 +183,7 @@ pub(crate) fn print_run_conclusion(
}
}
pub(crate) async fn print_final_output(
pub(crate) fn print_final_output(
checkpoint: Option<&fabro_types::Checkpoint>,
_run_dir: &Path,
styles: &Styles,

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

@ -4,8 +4,8 @@ use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use tracing::info;
use crate::args::{GlobalArgs, PreviewArgs};
use crate::shared::{print_json_pretty, validate_daytona_provider};
use crate::server_client;
use crate::shared::{print_json_pretty, validate_daytona_provider};
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn run(args: PreviewArgs, globals: &GlobalArgs) -> Result<()> {

22
lib/crates/fabro-cli/src/commands/run/rewind.rs
View file

@ -3,15 +3,13 @@ use anyhow::Result;
use cli_table::format::{Border, Separator};
use cli_table::{Cell, CellStruct, Color, Style, Table};
use fabro_checkpoint::git::Store;
use fabro_types::run_event::{CheckpointCompletedProps, RunRewoundProps, RunStatusTransitionProps};
use fabro_types::{EventBody, RunEvent};
use fabro_util::terminal::Styles;
use fabro_types::run_event::{
CheckpointCompletedProps, RunRewoundProps, RunStatusTransitionProps,
};
use fabro_workflow::git::MetadataStore;
use fabro_workflow::operations::{
RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, rewind,
};
use fabro_types::{EventBody, RunEvent};
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use git2::Repository;
use serde::Serialize;
@ -19,6 +17,7 @@ use serde::Serialize;
use crate::args::{GlobalArgs, RewindArgs};
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_client;
use crate::server_client::ServerStoreClient;
use crate::shared::{color_if, print_json_pretty};
use crate::user_config::load_user_settings_with_globals;
@ -97,16 +96,15 @@ pub(crate) fn timeline_entries_json(timeline: &RunTimeline) -> Vec<TimelineEntry
}
async fn reset_rewound_run_state(
client: &crate::server_client::ServerStoreClient,
client: &ServerStoreClient,
git_store: &Store,
run_id: &fabro_types::RunId,
run_dir: &std::path::Path,
entry: &TimelineEntry,
) -> Result<()> {
let state = client
.get_run_state(run_id)
.await
.map_err(|err| anyhow::anyhow!("failed to load durable store state before rewind: {err}"))?;
let state = client.get_run_state(run_id).await.map_err(|err| {
anyhow::anyhow!("failed to load durable store state before rewind: {err}")
})?;
let _run_record = state
.run
@ -191,11 +189,7 @@ fn restored_checkpoint_event(
)
}
fn run_event(
run_id: fabro_types::RunId,
node_id: Option<String>,
body: EventBody,
) -> RunEvent {
fn run_event(run_id: fabro_types::RunId, node_id: Option<String>, body: EventBody) -> RunEvent {
RunEvent {
id: ulid::Ulid::new().to_string(),
ts: chrono::Utc::now(),

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

@ -9,8 +9,8 @@ use fabro_workflow::run_status::RunStatus;
use tracing::info;
use crate::args::{GlobalArgs, WaitArgs};
use crate::shared::format_duration_ms;
use crate::server_client;
use crate::shared::format_duration_ms;
use crate::user_config::load_user_settings_with_globals;
#[cfg(test)]

11
lib/crates/fabro-cli/src/commands/runs/inspect.rs
View file

@ -9,6 +9,7 @@ use fabro_workflow::run_status::RunStatus;
use crate::args::{GlobalArgs, InspectArgs};
use crate::server_client;
use crate::server_client::RunProjection;
use crate::user_config::load_user_settings_with_globals;
#[derive(Debug, Serialize)]
@ -41,14 +42,18 @@ fn inspect_run_state(
run_id: &RunId,
run_dir: &Path,
status: RunStatus,
state: crate::server_client::RunProjection,
state: RunProjection,
) -> InspectOutput {
InspectOutput {
run_id: run_id.to_string(),
run_dir: run_dir.to_path_buf(),
status: state.status.as_ref().map_or(status, |record| record.status),
run_record: state.run.and_then(|record| serde_json::to_value(record).ok()),
start_record: state.start.and_then(|record| serde_json::to_value(record).ok()),
run_record: state
.run
.and_then(|record| serde_json::to_value(record).ok()),
start_record: state
.start
.and_then(|record| serde_json::to_value(record).ok()),
conclusion: state
.conclusion
.and_then(|record| serde_json::to_value(record).ok()),

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

@ -11,8 +11,8 @@ use fabro_workflow::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs
use fabro_workflow::run_status::RunStatus;
use crate::args::{GlobalArgs, RunsListArgs};
use crate::shared::{color_if, format_duration_ms, tilde_path};
use crate::server_client;
use crate::shared::{color_if, format_duration_ms, tilde_path};
use crate::user_config::load_user_settings_with_globals;
use super::short_run_id;

24
lib/crates/fabro-cli/src/commands/runs/rm.rs
View file

@ -1,16 +1,17 @@
use std::path::Path;
use anyhow::{Context, Result, bail};
use tracing::warn;
use crate::args::{GlobalArgs, RunsRemoveArgs};
use crate::server_client;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
use fabro_sandbox::reconnect::reconnect as reconnect_sandbox;
use fabro_workflow::event::{Event, to_run_event};
use fabro_workflow::run_lookup::RunInfo;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use tracing::warn;
use crate::args::{GlobalArgs, RunsRemoveArgs};
use crate::server_client;
use crate::server_client::RunProjection;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
use super::short_run_id;
@ -134,10 +135,7 @@ async fn remove_run_dir_with_cleanup(
};
if run_state.is_some() {
let run_event = to_run_event(&run_id, &Event::RunRemoving { reason: None });
if let Err(err) = client
.append_run_event(&run_id, &run_event)
.await
{
if let Err(err) = client.append_run_event(&run_id, &run_event).await {
warn!(
run_id = %run_id,
error = %err,
@ -146,7 +144,7 @@ async fn remove_run_dir_with_cleanup(
}
}
if let Some(record) = load_sandbox_record(run_state.as_ref()).await {
if let Some(record) = load_sandbox_record(run_state.as_ref()) {
if record.provider != "local" {
match reconnect_sandbox(&record).await {
Ok(sandbox) => {
@ -175,9 +173,7 @@ async fn delete_run_store_state(
.with_context(|| format!("failed to delete store state for {}", run.run_id()))
}
async fn load_sandbox_record(
run_state: Option<&crate::server_client::RunProjection>,
) -> Option<fabro_sandbox::SandboxRecord> {
fn load_sandbox_record(run_state: Option<&RunProjection>) -> Option<fabro_sandbox::SandboxRecord> {
if let Some(run_state) = run_state {
return run_state.sandbox.clone();
}

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

@ -1,5 +1,3 @@
use std::io::ErrorKind;
use std::path::Path;
use anyhow::{Context, Result};
#[cfg(test)]
use fabro_store::StageId;
@ -8,6 +6,8 @@ use fabro_workflow::run_dump::RunDump;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
#[cfg(test)]
use serde::de::DeserializeOwned;
use std::io::ErrorKind;
use std::path::Path;
use crate::args::{GlobalArgs, StoreDumpArgs};
use crate::commands::store::rebuild::rebuild_run_store;

5
lib/crates/fabro-cli/src/commands/system/df.rs
View file

@ -10,8 +10,8 @@ use fabro_workflow::run_lookup::{logs_base, runs_base, scan_runs_with_summaries}
use fabro_workflow::run_status::RunStatus;
use crate::args::{DfArgs, GlobalArgs};
use crate::shared::{format_size, print_json_pretty};
use crate::server_client;
use crate::shared::{format_size, print_json_pretty};
use crate::user_config::load_user_settings_with_globals;
#[derive(Serialize)]
@ -59,11 +59,10 @@ pub(super) async fn df_command(args: &DfArgs, globals: &GlobalArgs) -> Result<()
&logs_base_dir,
globals,
)
.await
}
#[allow(clippy::print_stdout)]
async fn df_from(
fn df_from(
args: &DfArgs,
summaries: &[fabro_store::RunSummary],
data_dir: &Path,

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

@ -3,11 +3,10 @@
mod args;
mod commands;
mod logging;
mod server_client;
mod shared;
#[cfg(feature = "sleep_inhibitor")]
mod sleep_inhibitor;
mod server_client;
mod store;
mod user_config;
use anyhow::Result;

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

@ -5,11 +5,14 @@ use std::time::Duration;
use anyhow::{Context as _, Result, anyhow};
use fabro_api::types;
use fabro_server::bind::Bind;
use fabro_store::{EventEnvelope, RunSummary, StageId};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunEvent, RunId,
RunRecord, RunStatusRecord, SandboxRecord, Settings, StartRecord,
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunEvent, RunId, RunRecord,
RunStatusRecord, SandboxRecord, Settings, StartRecord,
};
use serde::de::DeserializeOwned;
use tokio::time::sleep;
use crate::commands::server::start;
@ -98,8 +101,8 @@ pub(crate) async fn connect_server(storage_dir: &Path) -> Result<ServerStoreClie
let bind = start::ensure_server_running(storage_dir)
.with_context(|| format!("Failed to start fabro server for {}", storage_dir.display()))?;
let socket_path = match bind {
fabro_server::bind::Bind::Unix(path) => path,
fabro_server::bind::Bind::Tcp(addr) => {
Bind::Unix(path) => path,
Bind::Tcp(addr) => {
return Err(anyhow!(
"Unsupported server bind for store client auto-connect: {addr}"
));
@ -133,7 +136,7 @@ async fn wait_for_server_ready(http_client: &reqwest::Client) -> Result<()> {
}
Err(err) => last_error = Some(anyhow!(err)),
}
tokio::time::sleep(Duration::from_millis(50)).await;
sleep(Duration::from_millis(50)).await;
}
Err(last_error.unwrap_or_else(|| anyhow!("server did not become ready in time")))
@ -327,7 +330,7 @@ where
fn convert_type<TInput, TOutput>(value: TInput) -> Result<TOutput>
where
TInput: serde::Serialize,
TOutput: serde::de::DeserializeOwned,
TOutput: DeserializeOwned,
{
serde_json::from_value(serde_json::to_value(value)?).map_err(Into::into)
}

26
lib/crates/fabro-cli/src/store.rs
View file

@ -1,26 +0,0 @@
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use fabro_store::{SlateRunStore, SlateStore};
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
pub(crate) fn build_store(storage_dir: &Path) -> Result<Arc<SlateStore>> {
let store_path = storage_dir.join("store");
std::fs::create_dir_all(&store_path)?;
let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path)?);
Ok(Arc::new(SlateStore::new(
object_store,
"",
Duration::from_millis(1),
)))
}
pub(crate) async fn open_run_reader(storage_dir: &Path, run_id: &RunId) -> Result<SlateRunStore> {
build_store(storage_dir)?
.open_run_reader(run_id)
.await
.map_err(Into::into)
}

4
lib/crates/fabro-cli/tests/it/cmd/logs.rs
View file

@ -63,7 +63,7 @@ fn logs_completed_run_outputs_raw_ndjson() {
success: true
exit_code: 0
----- stdout -----
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[STORAGE_DIR]/runs/20260404-[ULID]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"gpt-5.4","provider":"openai"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
@ -228,7 +228,7 @@ fn logs_follow_detached_run_streams_until_completion() {
success: true
exit_code: 0
----- stdout -----
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[STORAGE_DIR]/runs/20260404-[ULID]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"claude-sonnet-4-6","provider":"anthropic"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.created","id":"[EVENT_ID]","properties":{"graph":{"attrs":{"goal":{"String":"Run tests and report results"},"rankdir":{"String":"LR"}},"edges":[{"attrs":{},"from":"start","to":"run_tests"},{"attrs":{},"from":"run_tests","to":"report"},{"attrs":{},"from":"report","to":"exit"}],"name":"Simple","nodes":{"exit":{"attrs":{"label":{"String":"Exit"},"shape":{"String":"Msquare"}},"id":"exit"},"report":{"attrs":{"label":{"String":"Report"},"prompt":{"String":"Summarize the test results"}},"id":"report"},"run_tests":{"attrs":{"label":{"String":"Run Tests"},"prompt":{"String":"Run the test suite and report results"}},"id":"run_tests"},"start":{"attrs":{"label":{"String":"Start"},"shape":{"String":"Mdiamond"}},"id":"start"}}},"host_repo_path":"[TEMP_DIR]","labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"run_dir":"[RUN_DIR]","settings":{"auto_approve":true,"dry_run":true,"fabro":{"root":"fabro/"},"features":{"retros":false,"session_sandboxes":false},"goal":"Run tests and report results","hooks":[{"blocking":true,"command":"cargo fmt","event":"post_tool_use","matcher":"write_file|edit_file|apply_patch","name":"cargo-fmt","sandbox":null,"timeout_ms":null}],"labels":{"fabro_test_case":"[TEST_CASE]","fabro_test_run":"[TEST_RUN]"},"llm":{"fallbacks":null,"model":"gpt-5.4","provider":"openai"},"mode":"standalone","no_retro":true,"pull_request":{"auto_merge":false,"draft":false,"enabled":true,"merge_strategy":"squash"},"sandbox":{"daytona":{"auto_stop_interval":30,"labels":{"repo":"fabro-sh/fabro"},"network":null,"skip_clone":false,"snapshot":{"cpu":4,"disk":20,"dockerfile":"FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n","memory":8,"name":"fabro-v6"}},"devcontainer":null,"env":null,"local":null,"preserve":null,"provider":"local"},"storage_dir":"[STORAGE_DIR]","version":1},"workflow_slug":"simple","workflow_source":"digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}
{"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"}

12
lib/crates/fabro-cli/tests/it/cmd/ps.rs
View file

@ -69,11 +69,13 @@ fn ps_all_json_lists_created_and_completed_runs() {
"all runs should belong to the Simple workflow: {runs:#?}"
);
assert!(
runs.iter().all(|run| run["labels"]["fabro_test_case"] == context.test_case_id()),
runs.iter()
.all(|run| run["labels"]["fabro_test_case"] == context.test_case_id()),
"all runs should be scoped to the current test case: {runs:#?}"
);
assert!(
runs.iter().all(|run| run["labels"]["fabro_test_run"] == context.test_run_id()),
runs.iter()
.all(|run| run["labels"]["fabro_test_run"] == context.test_run_id()),
"all runs should be scoped to the current test session: {runs:#?}"
);
assert!(
@ -152,7 +154,11 @@ fn ps_filters_by_workflow_and_label() {
assert!(output.status.success(), "ps should succeed");
let runs: Vec<Value> = serde_json::from_slice(&output.stdout).expect("ps JSON should parse");
assert_eq!(runs.len(), 1, "workflow+label filter should isolate one run");
assert_eq!(
runs.len(),
1,
"workflow+label filter should isolate one run"
);
let run = &runs[0];
assert_eq!(run["workflow_name"], "Simple");
assert_eq!(run["status"], "succeeded");

4
lib/crates/fabro-cli/tests/it/cmd/rm.rs
View file

@ -1,7 +1,9 @@
use fabro_test::{fabro_snapshot, test_context};
use serde_json::Value;
use super::support::{setup_completed_fast_dry_run, setup_created_fast_dry_run, setup_local_sandbox_run};
use super::support::{
setup_completed_fast_dry_run, setup_created_fast_dry_run, setup_local_sandbox_run,
};
#[test]
fn help() {

6
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -68,7 +68,7 @@ fn dry_run_simple() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [STORAGE_DIR]/runs/20260404-[ULID]
Run: [RUN_DIR]
=== Output ===
[Simulated] Response for stage: report
@ -348,8 +348,8 @@ fn json_run_implies_auto_approve_for_human_gates() {
"goal": "Route through the default approval path",
"llm": {
"fallbacks": null,
"model": "claude-sonnet-4-6",
"provider": "anthropic"
"model": "gpt-5.4",
"provider": "openai"
},
"mode": "standalone",
"no_retro": true,

6
lib/crates/fabro-cli/tests/it/cmd/server_start.rs
View file

@ -100,7 +100,11 @@ fn start_already_running_exits_with_error() {
#[test]
fn concurrent_autostart_converges_on_one_shared_daemon_and_cleans_up() {
fn run_ps_json(home_dir: &std::path::Path, temp_dir: &std::path::Path, storage_dir: &std::path::Path) -> std::process::Output {
fn run_ps_json(
home_dir: &std::path::Path,
temp_dir: &std::path::Path,
storage_dir: &std::path::Path,
) -> std::process::Output {
std::process::Command::new(env!("CARGO_BIN_EXE_fabro"))
.current_dir(temp_dir)
.env("NO_COLOR", "1")

12
lib/crates/fabro-cli/tests/it/cmd/start.rs
View file

@ -53,7 +53,11 @@ fn start_by_run_id_starts_created_run() {
.assert()
.success();
context.command().args(["start", &run_id]).assert().success();
context
.command()
.args(["start", &run_id])
.assert()
.success();
context
.command()
.args(["wait", &run_id])
@ -102,7 +106,11 @@ fn start_by_run_id_starts_created_run_without_run_json_or_status_json() {
let run_dir = context.find_run_dir(&run_id);
let _ = std::fs::remove_file(run_dir.join("run.json"));
context.command().args(["start", &run_id]).assert().success();
context
.command()
.args(["start", &run_id])
.assert()
.success();
let output = context
.command()
.args(["wait", "--json", &run_id])

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

@ -6,8 +6,8 @@ use std::time::{Duration, Instant};
use fabro_store::EventEnvelope;
use fabro_test::TestContext;
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord,
RunStatusRecord, SandboxRecord, StageId, StartRecord,
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord, RunStatusRecord,
SandboxRecord, StageId, StartRecord,
};
use serde_json::Value;
use shlex::try_quote;
@ -185,7 +185,13 @@ fn run_completed_dry_run(context: &TestContext, workflow: &Path) -> RunSetup {
let mut cmd = context.run_cmd();
cmd.current_dir(&context.temp_dir);
cmd.timeout(COMMAND_TIMEOUT);
cmd.args(["--dry-run", "--auto-approve", "--no-retro", "--sandbox", "local"]);
cmd.args([
"--dry-run",
"--auto-approve",
"--no-retro",
"--sandbox",
"local",
]);
cmd.arg(workflow);
let output = cmd.output().expect("command should execute");
if !output.status.success() {
@ -213,7 +219,13 @@ fn run_created_dry_run(context: &TestContext, workflow: &Path) -> RunSetup {
let mut cmd = context.create_cmd();
cmd.current_dir(&context.temp_dir);
cmd.timeout(COMMAND_TIMEOUT);
cmd.args(["--dry-run", "--auto-approve", "--no-retro", "--sandbox", "local"]);
cmd.args([
"--dry-run",
"--auto-approve",
"--no-retro",
"--sandbox",
"local",
]);
cmd.arg(workflow);
let output = cmd.output().expect("command should execute");
if !output.status.success() {
@ -668,13 +680,18 @@ async fn get_server_json_for_storage<T: serde::de::DeserializeOwned>(
pub(crate) fn run_state(run_dir: &Path) -> RunProjection {
let run_id = infer_run_id(run_dir);
block_on(get_server_json(run_dir, &format!("/api/v1/runs/{run_id}/state")))
block_on(get_server_json(
run_dir,
&format!("/api/v1/runs/{run_id}/state"),
))
}
pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
let run_id = infer_run_id(run_dir);
let response: serde_json::Value =
block_on(get_server_json(run_dir, &format!("/api/v1/runs/{run_id}/events")));
let response: serde_json::Value = block_on(get_server_json(
run_dir,
&format!("/api/v1/runs/{run_id}/events"),
));
serde_json::from_value(response["data"].clone()).expect("event list should parse")
}

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

@ -58,7 +58,7 @@ fn system_prune_dry_run_lists_matching_runs_without_deleting() {
success: true
exit_code: 0
----- stdout -----
would delete: 20260404-[ULID] (Simple)
would delete: 20260405-[ULID] (Simple)
----- stderr -----
1 run(s) would be deleted ([SIZE] freed). Pass --yes to confirm.

6
lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs
View file

@ -119,7 +119,11 @@ fn dry_run_create_start_attach_works_with_default_run_lookup() {
.assert()
.success();
context.command().args(["start", &run_id]).assert().success();
context
.command()
.args(["start", &run_id])
.assert()
.success();
context
.command()
.args(["attach", &run_id])

2
lib/crates/fabro-cli/tests/it/support/mod.rs
View file

@ -1,7 +1,7 @@
use std::path::{Path, PathBuf};
use fabro_types::RunId;
use fabro_test::TestContext;
use fabro_types::RunId;
macro_rules! fabro_json_snapshot {
($context:expr, $value:expr, @$snapshot:literal) => {{
let mut filters = $context.filters();

10
lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs
View file

@ -30,7 +30,7 @@ fn dry_run_branching() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [STORAGE_DIR]/runs/20260404-[ULID]
Run: [RUN_DIR]
=== Output ===
[Simulated] Response for stage: validate
@ -62,7 +62,7 @@ fn dry_run_conditions() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [STORAGE_DIR]/runs/20260404-[ULID]
Run: [RUN_DIR]
=== Output ===
[Simulated] Response for stage: path_b
@ -95,7 +95,7 @@ fn dry_run_parallel() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [STORAGE_DIR]/runs/20260404-[ULID]
Run: [RUN_DIR]
=== Output ===
[Simulated] Response for stage: review
@ -128,7 +128,7 @@ fn dry_run_styled() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [STORAGE_DIR]/runs/20260404-[ULID]
Run: [RUN_DIR]
=== Output ===
[Simulated] Response for stage: critical_review
@ -159,6 +159,6 @@ fn dry_run_legacy_tool() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [STORAGE_DIR]/runs/20260404-[ULID]
Run: [RUN_DIR]
");
}

4
lib/crates/fabro-server/src/jwt_auth.rs
View file

@ -77,7 +77,9 @@ pub fn resolve_auth_mode(api_settings: &ApiSettings, allowed_usernames: &[String
if api_settings.authentication_strategies.is_empty()
&& std::env::var("FABRO_LOCAL_NO_AUTH").ok().as_deref() == Some("1")
{
warn!("No authentication strategies configured; allowing unauthenticated local daemon access");
warn!(
"No authentication strategies configured; allowing unauthenticated local daemon access"
);
return AuthMode::Disabled;
}

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

@ -51,6 +51,9 @@ use fabro_workflow::event::Emitter;
use fabro_workflow::operations::{self, CreateRunInput, WorkflowInput};
use fabro_workflow::pipeline::Persisted;
use fabro_workflow::records::Checkpoint;
use fabro_workflow::run_status::RunStatus as WorkflowRunStatus;
#[cfg(test)]
use fabro_workflow::run_status::StatusReason as WorkflowStatusReason;
use fabro_api::types::AggregateUsageTotals;
pub use fabro_api::types::{
@ -59,8 +62,8 @@ pub use fabro_api::types::{
CompletionResponse, CompletionToolChoiceMode, CompletionUsage, CreateCompletionRequest,
CreateRunRequest, EventEnvelope as ApiEventEnvelope, ModelReference, PaginatedEventList,
PaginatedRunList, PaginationMeta, QuestionType as ApiQuestionType, RunError,
RunEvent as ApiRunEvent, RunStatus, RunStatusResponse, StartRunRequest,
SubmitAnswerRequest, TokenUsage, UsageByModel, WriteBlobResponse,
RunEvent as ApiRunEvent, RunStatus, RunStatusResponse, StartRunRequest, SubmitAnswerRequest,
TokenUsage, UsageByModel, WriteBlobResponse,
};
pub fn default_page_limit() -> u32 {
@ -183,7 +186,6 @@ impl AppState {
}
}
/// Build the axum Router with all run endpoints and embedded static assets.
pub fn build_router(state: Arc<AppState>, auth_mode: AuthMode) -> Router {
let middleware_state = Arc::clone(&state);
@ -242,7 +244,10 @@ fn demo_routes() -> Router<Arc<AppState>> {
.route("/runs/{id}/questions", get(demo::get_questions_stub))
.route("/runs/{id}/questions/{qid}/answer", post(demo::answer_stub))
.route("/runs/{id}/state", get(not_implemented))
.route("/runs/{id}/events", get(not_implemented).post(not_implemented))
.route(
"/runs/{id}/events",
get(not_implemented).post(not_implemented),
)
.route("/runs/{id}/attach", get(demo::run_events_stub))
.route("/runs/{id}/blobs", post(not_implemented))
.route("/runs/{id}/blobs/{blobId}", get(not_implemented))
@ -331,7 +336,10 @@ fn real_routes() -> Router<Arc<AppState>> {
.route("/runs/{id}/questions", get(get_questions))
.route("/runs/{id}/questions/{qid}/answer", post(submit_answer))
.route("/runs/{id}/state", get(get_run_state))
.route("/runs/{id}/events", get(list_run_events).post(append_run_event))
.route(
"/runs/{id}/events",
get(list_run_events).post(append_run_event),
)
.route("/runs/{id}/attach", get(attach_run_events))
.route("/runs/{id}/blobs", post(write_run_blob))
.route("/runs/{id}/blobs/{blobId}", get(read_run_blob))
@ -569,17 +577,16 @@ async fn list_board_runs(
.into_response()
}
async fn list_runs(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
) -> Response {
async fn list_runs(_auth: AuthenticatedService, State(state): State<Arc<AppState>>) -> Response {
match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(runs) => (StatusCode::OK, Json(runs)).into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
@ -599,7 +606,9 @@ async fn delete_run(
match state.store.delete_run(&id).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
@ -714,7 +723,10 @@ async fn create_run(
};
info!(run_id = %run_id, "Run created");
let using_dot_source = req.dot_source.as_ref().is_some_and(|value| !value.is_empty());
let using_dot_source = req
.dot_source
.as_ref()
.is_some_and(|value| !value.is_empty());
let using_local_workflow = req
.workflow_path
.as_ref()
@ -807,7 +819,7 @@ async fn create_run(
created_at,
}),
)
.into_response()
.into_response()
}
async fn start_run(
@ -820,7 +832,7 @@ async fn start_run(
Ok(id) => id,
Err(response) => return response,
};
let resume = body.map(|Json(req)| req.resume).unwrap_or(false);
let resume = body.is_some_and(|Json(req)| req.resume);
{
let runs = state.runs.lock().expect("runs lock poisoned");
@ -842,9 +854,8 @@ async fn start_run(
}
}
let run_store = match state.store.open_run(&id).await {
Ok(run_store) => run_store,
Err(_) => return ApiError::not_found("Run not found.").into_response(),
let Ok(run_store) = state.store.open_run(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
let run_state = match run_store.state().await {
Ok(state) => state,
@ -863,19 +874,27 @@ async fn start_run(
.into_response();
}
} else if let Some(record) = run_state.status.as_ref() {
if !matches!(record.status, fabro_workflow::run_status::RunStatus::Submitted | fabro_workflow::run_status::RunStatus::Starting)
{
if !matches!(
record.status,
WorkflowRunStatus::Submitted | WorkflowRunStatus::Starting
) {
return ApiError::new(
StatusCode::CONFLICT,
format!("cannot start run: status is {:?}, expected submitted", record.status),
format!(
"cannot start run: status is {:?}, expected submitted",
record.status
),
)
.into_response();
}
}
let Some(run_record) = run_state.run.as_ref() else {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, "run record missing from store")
.into_response();
return ApiError::new(
StatusCode::INTERNAL_SERVER_ERROR,
"run record missing from store",
)
.into_response();
};
let run_dir = operations::make_run_dir(&run_record.settings.storage_dir().join("runs"), &id);
let dot_source = run_state.graph_source.unwrap_or_default();
@ -909,7 +928,7 @@ async fn start_run(
created_at: id.created_at(),
}),
)
.into_response()
.into_response()
}
/// Execute a single run: transitions queued → starting → running → completed/failed/cancelled.
@ -1174,7 +1193,9 @@ async fn get_run_status(
Some(run) => (StatusCode::OK, Json(run)).into_response(),
None => ApiError::not_found("Run not found.").into_response(),
},
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
}
@ -1305,8 +1326,9 @@ async fn get_run_state(
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => Json(run_state).into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -1324,7 +1346,9 @@ async fn append_run_event(
};
let event = match RunEvent::from_value(value.clone()) {
Ok(event) => event,
Err(err) => return ApiError::bad_request(format!("Invalid run event: {err}")).into_response(),
Err(err) => {
return ApiError::bad_request(format!("Invalid run event: {err}")).into_response();
}
};
if event.run_id != id {
return ApiError::bad_request("Event run_id does not match path run ID.").into_response();
@ -1336,9 +1360,13 @@ async fn append_run_event(
match state.store.open_run(&id).await {
Ok(run_store) => match run_store.append_event(&payload).await {
Ok(seq) => Json(AppendEventResponse { seq: i64::from(seq) }).into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Ok(seq) => Json(AppendEventResponse {
seq: i64::from(seq),
})
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -1357,7 +1385,10 @@ async fn list_run_events(
let since_seq = params.since_seq();
let limit = params.limit();
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.list_events_from_with_limit(since_seq, limit).await {
Ok(run_store) => match run_store
.list_events_from_with_limit(since_seq, limit)
.await
{
Ok(mut events) => {
let has_more = events.len() > limit;
events.truncate(limit);
@ -1375,8 +1406,9 @@ async fn list_run_events(
})
.into_response()
}
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -1427,18 +1459,16 @@ async fn attach_run_events(
.into_response();
}
};
let stream = stream.filter_map(|result| {
match result {
Ok(event) => {
let event = api_event_envelope_from_store(&event).ok()?;
let data = serde_json::to_string(&event).ok()?;
let data = redact_jsonl_line(&data);
Some(Ok::<Event, std::convert::Infallible>(
Event::default().data(data),
))
}
Err(_) => None,
let stream = stream.filter_map(|result| match result {
Ok(event) => {
let event = api_event_envelope_from_store(&event).ok()?;
let data = serde_json::to_string(&event).ok()?;
let data = redact_jsonl_line(&data);
Some(Ok::<Event, std::convert::Infallible>(
Event::default().data(data),
))
}
Err(_) => None,
});
Sse::new(stream).into_response()
@ -1498,8 +1528,9 @@ async fn write_run_blob(
id: blob_id.to_string(),
})
.into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -1522,8 +1553,9 @@ async fn read_run_blob(
Ok(run_store) => match run_store.read_blob(&blob_id).await {
Ok(Some(bytes)) => octet_stream_response(bytes),
Ok(None) => ApiError::not_found("Blob not found.").into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -1551,8 +1583,9 @@ async fn list_stage_artifacts(
.collect(),
})
.into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -1580,8 +1613,9 @@ async fn put_stage_artifact(
match state.store.open_run(&id).await {
Ok(run_store) => match run_store.put_artifact(&stage_id, &filename, &body).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -1609,8 +1643,9 @@ async fn get_stage_artifact(
Ok(run_store) => match run_store.get_artifact(&stage_id, &filename).await {
Ok(Some(bytes)) => octet_stream_response(bytes),
Ok(None) => ApiError::not_found("Artifact not found.").into_response(),
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(_) => ApiError::not_found("Run not found.").into_response(),
}
@ -3392,8 +3427,8 @@ mod tests {
let mut status_record = None;
for _ in 0..50 {
if let Some(record) = run_store.state().await.unwrap().status {
if record.status == fabro_workflow::run_status::RunStatus::Failed
&& record.reason == Some(fabro_workflow::run_status::StatusReason::Cancelled)
if record.status == WorkflowRunStatus::Failed
&& record.reason == Some(WorkflowStatusReason::Cancelled)
{
status_record = Some(record);
break;
@ -3403,14 +3438,8 @@ mod tests {
}
let status_record = status_record.expect("status record should be persisted");
assert_eq!(
status_record.status,
fabro_workflow::run_status::RunStatus::Failed
);
assert_eq!(
status_record.reason,
Some(fabro_workflow::run_status::StatusReason::Cancelled)
);
assert_eq!(status_record.status, WorkflowRunStatus::Failed);
assert_eq!(status_record.reason, Some(WorkflowStatusReason::Cancelled));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]

11
lib/crates/fabro-store/src/keys.rs
View file

@ -66,7 +66,11 @@ pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
}
pub(crate) fn parse_blob_id(key: &str) -> Option<RunBlobId> {
key.rsplit('/').next()?.strip_prefix(BLOBS_PREFIX)?.parse().ok()
key.rsplit('/')
.next()?
.strip_prefix(BLOBS_PREFIX)?
.parse()
.ok()
}
pub(crate) fn parse_node_artifact_key(key: &str) -> Option<(StageId, String)> {
@ -119,7 +123,10 @@ mod tests {
let node = StageId::new("code", 2);
let run_id = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
let blob_id = RunBlobId::new(&run_id, b"summary");
assert_eq!(blob_key(&run_id, &blob_id), format!("runs/{run_id}/blobs#{blob_id}"));
assert_eq!(
blob_key(&run_id, &blob_id),
format!("runs/{run_id}/blobs#{blob_id}")
);
assert_eq!(
node_artifact(&run_id, &node, "src/main.rs"),
"runs/01JT56VE4Z5NZ814GZN2JZD65A/artifacts#nodes#code#visit-2#src/main.rs"

5
lib/crates/fabro-store/src/slate/catalog.rs
View file

@ -25,8 +25,9 @@ pub(crate) async fn list_run_ids(db: &Db, query: &ListRunsQuery) -> Result<Vec<R
let mut iter = db.scan_prefix(keys::catalog_by_start_prefix()).await?;
let mut run_ids = Vec::new();
while let Some(entry) = iter.next().await? {
let key = String::from_utf8(entry.key.to_vec())
.map_err(|err| crate::StoreError::Other(format!("stored key is not valid UTF-8: {err}")))?;
let key = String::from_utf8(entry.key.to_vec()).map_err(|err| {
crate::StoreError::Other(format!("stored key is not valid UTF-8: {err}"))
})?;
let Some(run_id) = keys::parse_run_id_from_catalog_key(&key) else {
continue;
};

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

@ -108,7 +108,8 @@ impl SlateStore {
}
SlateRunStore::validate_init(&db, run_id).await?;
db.put(keys::init_key(run_id), serde_json::to_vec(run_id)?).await?;
db.put(keys::init_key(run_id), serde_json::to_vec(run_id)?)
.await?;
catalog::write_catalog(&db, run_id).await?;
let run_store = SlateRunStore::open_writer(*run_id, db).await?;
self.cache_active_run(&run_store).await;
@ -144,7 +145,7 @@ impl SlateStore {
"active run cache mismatch for run_id {run_id:?}"
)));
}
return Ok(active.into_read_only());
return Ok(active.read_only_clone());
}
if !catalog::read_locator(&db, run_id).await? {
return Err(StoreError::RunNotFound(run_id.to_string()));
@ -212,10 +213,10 @@ mod tests {
use super::*;
use chrono::{DateTime, Utc};
use futures::TryStreamExt;
use object_store::path::Path;
use object_store::memory::InMemory;
use fabro_types::{AttrValue, Graph, RunRecord, RunStatus, Settings, StatusReason};
use futures::TryStreamExt;
use object_store::memory::InMemory;
use object_store::path::Path;
use std::path::PathBuf;
use crate::EventPayload;

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

@ -1,6 +1,6 @@
use std::collections::VecDeque;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use bytes::Bytes;
use chrono::Utc;
@ -52,7 +52,8 @@ pub(crate) struct SlateRunStoreInner {
impl SlateRunStore {
pub(crate) async fn open_writer(run_id: RunId, db: Db) -> Result<Self> {
let event_seq = recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?;
let event_seq =
recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?;
let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16));
Ok(Self {
inner: Arc::new(SlateRunStoreInner {
@ -71,7 +72,8 @@ impl SlateRunStore {
}
pub(crate) async fn open_reader(run_id: RunId, db: Db) -> Result<Self> {
let event_seq = recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?;
let event_seq =
recover_next_seq(&db, &keys::events_prefix(&run_id), keys::parse_event_seq).await?;
let (event_tx, _) = broadcast::channel(DEFAULT_EVENT_TAIL_LIMIT.max(16));
Ok(Self {
inner: Arc::new(SlateRunStoreInner {
@ -96,7 +98,7 @@ impl SlateRunStore {
}
}
pub(crate) fn into_read_only(&self) -> Self {
pub(crate) fn read_only_clone(&self) -> Self {
Self {
inner: Arc::clone(&self.inner),
read_only: true,
@ -203,10 +205,13 @@ impl SlateRunStore {
seq,
payload: payload.clone(),
};
self.inner.db.put(
keys::event_key(&self.inner.run_id, seq, Utc::now().timestamp_millis()),
serde_json::to_vec(payload)?,
).await?;
self.inner
.db
.put(
keys::event_key(&self.inner.run_id, seq, Utc::now().timestamp_millis()),
serde_json::to_vec(payload)?,
)
.await?;
self.cache_event(&event).await?;
Ok(seq)
}
@ -293,7 +298,10 @@ impl SlateRunStore {
}
self.inner
.db
.put(keys::node_artifact(&self.inner.run_id, node, filename), data)
.put(
keys::node_artifact(&self.inner.run_id, node, filename),
data,
)
.await?;
Ok(())
}
@ -350,7 +358,9 @@ async fn list_events_from<R>(db: &R, run_id: &RunId, start_seq: u32) -> Result<V
where
R: DbRead + Sync,
{
let mut iter = db.scan_prefix(keys::events_prefix(run_id).as_bytes()).await?;
let mut iter = db
.scan_prefix(keys::events_prefix(run_id).as_bytes())
.await?;
let mut events = Vec::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
@ -387,7 +397,9 @@ async fn list_blobs<R>(db: &R, run_id: &RunId) -> Result<Vec<RunBlobId>>
where
R: DbRead + Sync,
{
let mut iter = db.scan_prefix(keys::blobs_prefix(run_id).as_bytes()).await?;
let mut iter = db
.scan_prefix(keys::blobs_prefix(run_id).as_bytes())
.await?;
let mut blob_ids = Vec::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
@ -404,9 +416,7 @@ async fn list_all_artifacts<R>(db: &R, run_id: &RunId) -> Result<Vec<NodeArtifac
where
R: DbRead + Sync,
{
let mut iter = db
.scan_prefix(keys::run_prefix(run_id).as_bytes())
.await?;
let mut iter = db.scan_prefix(keys::run_prefix(run_id).as_bytes()).await?;
let mut assets = Vec::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
@ -419,7 +429,11 @@ where
Ok(assets)
}
async fn list_artifacts_for_stage<R>(db: &R, run_id: &RunId, stage_id: &StageId) -> Result<Vec<String>>
async fn list_artifacts_for_stage<R>(
db: &R,
run_id: &RunId,
stage_id: &StageId,
) -> Result<Vec<String>>
where
R: DbRead + Sync,
{

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

@ -204,7 +204,7 @@ fn shorten_session_id(id: &str) -> String {
let trimmed = id.trim();
let shortened: String = trimmed
.chars()
.filter(|ch| ch.is_ascii_alphanumeric())
.filter(char::is_ascii_alphanumeric)
.take(12)
.collect();
if shortened.is_empty() {
@ -339,7 +339,10 @@ fn reap_stale_session_roots(fabro_bin: &Path, mode: SessionMode) {
if !root.is_dir() {
continue;
}
let file_name = root.file_name().and_then(|name| name.to_str()).unwrap_or("");
let file_name = root
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("");
let expected_prefix = match mode {
SessionMode::Nextest => "n-",
SessionMode::Process => "p-",
@ -494,10 +497,7 @@ impl TestContext {
cmd.env("HOME", &self.home_dir);
cmd.env("FABRO_NO_UPGRADE_CHECK", "true");
cmd.env("FABRO_STORAGE_DIR", &self.storage_dir);
cmd.env(
"FABRO_SERVER_MAX_CONCURRENT_RUNS",
"64",
);
cmd.env("FABRO_SERVER_MAX_CONCURRENT_RUNS", "64");
cmd
}
@ -1283,7 +1283,11 @@ mod tests {
let _guard = EnvGuard::set("NEXTEST_RUN_ID", None);
let (_, run_id, paths) = session_paths();
assert_eq!(run_id, format!("process-{}", current_pid()));
assert!(paths.root.ends_with(Path::new("fx").join(format!("p-{}", current_pid()))));
assert!(
paths
.root
.ends_with(Path::new("fx").join(format!("p-{}", current_pid())))
);
assert_eq!(paths.storage_dir, paths.root.join("storage"));
}

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

@ -199,7 +199,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<R
.list_runs(&fabro_store::ListRunsQuery::default())
.await
.unwrap_or_default();
Ok(scan_runs_with_summaries(&store_runs, base)?)
scan_runs_with_summaries(&store_runs, base)
}
pub fn scan_runs_with_summaries(summaries: &[RunSummary], base: &Path) -> Result<Vec<RunInfo>> {