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

Cut more CLI control-plane reads over to the store

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-01 21:19:54 -04:00
parent f73b28129c
commit 7f59d14245
No known key found for this signature in database
15 changed files with 233 additions and 164 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -6,9 +6,7 @@ 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::records::{
Conclusion, ConclusionExt, RunRecord, RunRecordExt, StartRecord, StartRecordExt,
};
use fabro_workflow::records::RunRecordExt;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use tracing::info;
@ -37,41 +35,24 @@ async fn create_from(
let store = store::build_store(storage_dir)?;
let run = resolve_run_combined(store.as_ref(), base, &args.run_id).await?;
let run_dir = run.path.clone();
let run_store = store::open_run_reader(storage_dir, &run.run_id).await?;
let run_store = store::open_run_reader(storage_dir, &run.run_id)
.await?
.context("Failed to open run store")?;
let record = match run_store.as_ref() {
Some(run_store) => run_store
.get_run()
.await
.ok()
.flatten()
.or_else(|| RunRecord::load(&run_dir).ok())
.context("Failed to load run.json")?,
None => RunRecord::load(&run_dir).context("Failed to load run.json")?,
};
let record = run_store
.get_run()
.await?
.context("Failed to load run record from store")?;
let start = match run_store.as_ref() {
Some(run_store) => run_store
.get_start()
.await
.ok()
.flatten()
.or_else(|| StartRecord::load(&run_dir).ok())
.context("Failed to load start.json")?,
None => StartRecord::load(&run_dir).context("Failed to load start.json")?,
};
let start = run_store
.get_start()
.await?
.context("Failed to load start record from store")?;
let conclusion = match run_store.as_ref() {
Some(run_store) => run_store
.get_conclusion()
.await
.ok()
.flatten()
.or_else(|| Conclusion::load(&run_dir.join("conclusion.json")).ok())
.context("Failed to load conclusion.json — is the run finished?")?,
None => Conclusion::load(&run_dir.join("conclusion.json"))
.context("Failed to load conclusion.json — is the run finished?")?,
};
let conclusion = run_store
.get_conclusion()
.await?
.context("Failed to load conclusion from store — is the run finished?")?;
match conclusion.status {
StageStatus::Success | StageStatus::PartialSuccess => {}
@ -83,17 +64,10 @@ async fn create_from(
.as_deref()
.context("Run has no run_branch — was it run with git push enabled?")?;
let diff = match run_store.as_ref() {
Some(run_store) => run_store
.get_final_patch()
.await
.ok()
.flatten()
.or_else(|| std::fs::read_to_string(run_dir.join("final.patch")).ok())
.context("Failed to read final.patch — no diff available")?,
None => std::fs::read_to_string(run_dir.join("final.patch"))
.context("Failed to read final.patch — no diff available")?,
};
let diff = run_store
.get_final_patch()
.await?
.context("Failed to load final patch from store — no diff available")?;
if diff.trim().is_empty() {
bail!("final.patch is empty — nothing to create a PR for");
}
@ -147,7 +121,7 @@ async fn create_from(
&model,
true,
None,
run_store.as_deref(),
Some(run_store.as_ref()),
&run_dir,
None,
)
@ -157,10 +131,8 @@ async fn create_from(
match record {
Some(record) => {
info!(pr_url = %record.html_url, "Pull request created");
if let Some(run_store) = run_store.as_ref() {
if let Err(err) = run_store.put_pull_request(&record).await {
tracing::warn!(error = %err, "Failed to persist pull request in run store");
}
if let Err(err) = run_store.put_pull_request(&record).await {
tracing::warn!(error = %err, "Failed to persist pull request in run store");
}
if let Err(err) = record.save(&run_dir.join("pull_request.json")) {
tracing::warn!(error = %err, "Failed to save pull_request.json");

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

@ -53,13 +53,6 @@ async fn list_from(
if let Ok(Some(run_store)) = store.open_run_reader(&run.run_id).await {
if let Ok(Some(record)) = run_store.get_pull_request().await {
entries.push((run.run_id.to_string(), record));
continue;
}
}
let pr_path = run.path.join("pull_request.json");
if let Ok(content) = std::fs::read_to_string(&pr_path) {
if let Ok(record) = serde_json::from_str::<PullRequestRecord>(&content) {
entries.push((run.run_id.to_string(), record));
}
}
}

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

@ -37,20 +37,11 @@ pub(crate) async fn load_pr_record(
let store = store::build_store(storage_dir)?;
let run = resolve_run_combined(store.as_ref(), base, run_id).await?;
let run_dir = run.path;
let run_store = store::open_run_reader(storage_dir, &run.run_id).await?;
if let Some(run_store) = run_store {
if let Some(record) = run_store.get_pull_request().await.ok().flatten() {
return Ok((record, run_dir));
}
}
let pr_path = run_dir.join("pull_request.json");
let content = std::fs::read_to_string(&pr_path).with_context(|| {
format!(
"No pull_request.json found in run directory. \
Create one first with: fabro pr create {run_id}"
)
let run_store = store::open_run_reader(storage_dir, &run.run_id)
.await?
.context("Failed to open run store")?;
let record = run_store.get_pull_request().await?.with_context(|| {
format!("No pull request found in store. Create one first with: fabro pr create {run_id}")
})?;
let record: PullRequestRecord =
serde_json::from_str(&content).context("Failed to parse pull_request.json")?;
Ok((record, run_dir))
}

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

@ -43,7 +43,13 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<(
)
.await?;
if !globals.json {
super::output::print_run_summary(&run_dir, run_id, styles);
super::output::print_run_summary(
cli_settings.storage_dir().as_path(),
&run_dir,
run_id,
styles,
)
.await?;
}
if exit_code != std::process::ExitCode::SUCCESS {
std::process::exit(1);

29
lib/crates/fabro-cli/src/commands/run/cp.rs
View file

@ -3,14 +3,14 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
use fabro_agent::sandbox::Sandbox;
use fabro_config::FabroSettingsExt;
use fabro_sandbox::SandboxRecordExt;
use fabro_sandbox::reconnect::reconnect;
use fabro_workflow::run_lookup::{resolve_run, runs_base};
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use tokio::fs;
use tracing::{debug, info};
use crate::args::{CpArgs, GlobalArgs};
use crate::shared::{print_json_pretty, split_run_path};
use crate::store;
use crate::user_config::load_user_settings_with_globals;
enum CopyDirection {
@ -37,7 +37,7 @@ pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()>
remote_path,
local_path,
} => {
let sandbox = load_sandbox(&base, &run_prefix).await?;
let sandbox = load_sandbox(&cli_settings.storage_dir(), &base, &run_prefix).await?;
let file_count = if args.recursive {
Some(download_recursive(&*sandbox, &remote_path, &local_path).await?)
@ -70,7 +70,7 @@ pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()>
run_prefix,
remote_path,
} => {
let sandbox = load_sandbox(&base, &run_prefix).await?;
let sandbox = load_sandbox(&cli_settings.storage_dir(), &base, &run_prefix).await?;
let file_count = if args.recursive {
Some(upload_recursive(&*sandbox, &local_path, &remote_path).await?)
@ -124,13 +124,20 @@ fn parse_direction(src: &str, dst: &str) -> Result<CopyDirection> {
}
}
async fn load_sandbox(base: &Path, run_prefix: &str) -> Result<Box<dyn Sandbox>> {
let run_dir = resolve_run(base, run_prefix)?.path;
let sandbox_json = run_dir.join("sandbox.json");
debug!(path = %sandbox_json.display(), "Loading sandbox record");
let record = fabro_sandbox::SandboxRecord::load(&sandbox_json).context(
"Failed to load sandbox.json — was this run started with a recent version of arc?",
)?;
async fn load_sandbox(
storage_dir: &Path,
base: &Path,
run_prefix: &str,
) -> Result<Box<dyn Sandbox>> {
let store = store::build_store(storage_dir)?;
let run = resolve_run_combined(store.as_ref(), base, run_prefix).await?;
let run_store = store::open_run_reader(storage_dir, &run.run_id)
.await?
.context("Failed to open run store")?;
let record = run_store
.get_sandbox()
.await?
.context("Failed to load sandbox record from store")?;
info!(run_id = %run_prefix, provider = %record.provider, "Connecting to sandbox");
reconnect(&record).await

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

@ -1,6 +1,7 @@
use std::path::Path;
use std::time::Duration;
use anyhow::Result;
use fabro_graphviz::graph::Graph;
use fabro_store::RuntimeState;
use fabro_types::PullRequestRecord;
@ -13,6 +14,7 @@ use fabro_workflow::records::{Checkpoint, CheckpointExt, Conclusion, ConclusionE
use indicatif::HumanDuration;
use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path};
use crate::store;
fn print_workflow_header(
graph: &Graph,
@ -69,20 +71,31 @@ pub(crate) fn print_diagnostics_from_error(
print_diagnostics(diagnostics, styles);
}
pub(crate) fn print_run_summary(run_dir: &Path, run_id: impl std::fmt::Display, styles: &Styles) {
pub(crate) async fn print_run_summary(
storage_dir: &Path,
run_dir: &Path,
run_id: impl std::fmt::Display,
styles: &Styles,
) -> Result<()> {
let run_id = run_id.to_string();
let conclusion_path = run_dir.join("conclusion.json");
let Ok(conclusion) = Conclusion::load(&conclusion_path) else {
return;
return Ok(());
};
let pr_url = std::fs::read_to_string(run_dir.join("pull_request.json"))
.ok()
.and_then(|content| {
serde_json::from_str::<PullRequestRecord>(&content)
.ok()
.map(|record| record.html_url)
});
let pr_url = match run_id.parse() {
Ok(parsed_run_id) => {
if let Some(run_store) = store::open_run_reader(storage_dir, &parsed_run_id).await? {
run_store
.get_pull_request()
.await?
.map(|record: PullRequestRecord| record.html_url)
} else {
None
}
}
Err(_) => None,
};
print_run_conclusion(
&conclusion,
@ -94,6 +107,7 @@ pub(crate) fn print_run_summary(run_dir: &Path, run_id: impl std::fmt::Display,
);
print_final_output(run_dir, styles);
print_assets(run_dir, styles);
Ok(())
}
pub(crate) fn print_run_conclusion(

12
lib/crates/fabro-cli/src/commands/run/resume.rs
View file

@ -24,10 +24,6 @@ pub(crate) async fn resume_command(
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_dir = run.path;
// find_run_by_prefix can match orphan directories (no run.json).
if !run_dir.join("run.json").exists() {
bail!("run directory exists but has no run.json — cannot resume");
}
let run_id = run.run_id;
if launcher_pid_alive(&run_dir) {
@ -55,7 +51,13 @@ pub(crate) async fn resume_command(
)
.await?;
if !globals.json {
super::output::print_run_summary(&run_dir, run_id, styles);
super::output::print_run_summary(
cli_settings.storage_dir().as_path(),
&run_dir,
run_id,
styles,
)
.await?;
}
if exit_code != std::process::ExitCode::SUCCESS {
std::process::exit(1);

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

@ -4,9 +4,9 @@ use anyhow::{Result, bail};
use fabro_config::FabroSettingsExt;
use fabro_types::RunId;
use fabro_util::terminal::Styles;
use fabro_workflow::records::{Conclusion, ConclusionExt};
use fabro_workflow::records::Conclusion;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use fabro_workflow::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt};
use fabro_workflow::run_status::RunStatus;
use tracing::info;
use crate::args::{GlobalArgs, WaitArgs};
@ -27,7 +27,6 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
info!(run_id = %run_info.run_id, "Waiting for run to complete");
let status_path = run_info.path.join("status.json");
let deadline = args
.timeout
.map(|secs| std::time::Instant::now() + std::time::Duration::from_secs(secs));
@ -35,16 +34,10 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
let started_waiting_at = std::time::Instant::now();
let final_status = loop {
let load_file_status = || RunStatusRecord::load(&status_path).ok().map(|r| r.status);
let run_store =
store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?;
let status = match run_store.as_ref() {
Some(run_store) => match run_store.get_status().await {
Ok(Some(record)) => Some(record.status),
Ok(None) | Err(_) => load_file_status(),
},
None => load_file_status(),
};
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id)
.await?
.ok_or_else(|| anyhow::anyhow!("Run {} not found in store", run_info.run_id))?;
let status = run_store.get_status().await?.map(|record| record.status);
let status = status.unwrap_or_else(|| {
if started_waiting_at.elapsed() < WAIT_STARTUP_GRACE {
RunStatus::Submitted
@ -72,17 +65,10 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
}
};
let conclusion_path = run_info.path.join("conclusion.json");
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?;
let conclusion = match run_store.as_ref() {
Some(run_store) => run_store
.get_conclusion()
.await
.ok()
.flatten()
.or_else(|| Conclusion::load(&conclusion_path).ok()),
None => Conclusion::load(&conclusion_path).ok(),
};
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id)
.await?
.ok_or_else(|| anyhow::anyhow!("Run {} not found in store", run_info.run_id))?;
let conclusion = run_store.get_conclusion().await?;
if globals.json {
let json_value = build_json_output(final_status, &run_info.run_id, conclusion.as_ref());

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

@ -1,13 +1,10 @@
use std::path::{Path, PathBuf};
use anyhow::Result;
use anyhow::{Result, anyhow};
use fabro_config::FabroSettingsExt;
use fabro_sandbox::SandboxRecordExt;
use fabro_types::RunId;
use fabro_workflow::records::{CheckpointExt, ConclusionExt, RunRecordExt, StartRecordExt};
use serde::Serialize;
use fabro_workflow::records::{Checkpoint, Conclusion, RunRecord, StartRecord};
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use fabro_workflow::run_status::RunStatus;
@ -32,12 +29,10 @@ pub(crate) async fn run(args: &InspectArgs, globals: &GlobalArgs) -> Result<()>
let base = runs_base(&cli_settings.storage_dir());
let store = store::build_store(&cli_settings.storage_dir())?;
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let output = match store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await? {
Some(run_store) => {
inspect_run_store(&run.run_id, &run.path, run.status, run_store.as_ref()).await
}
None => inspect_run_dir(&run.run_id, &run.path, run.status),
};
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id)
.await?
.ok_or_else(|| anyhow!("Run {} not found in store", run.run_id))?;
let output = inspect_run_store(&run.run_id, &run.path, run.status, run_store.as_ref()).await;
let json = serde_json::to_string_pretty(&[output])?;
println!("{json}");
Ok(())
@ -49,8 +44,8 @@ async fn inspect_run_store(
status: RunStatus,
run_store: &dyn fabro_store::RunStore,
) -> InspectOutput {
match run_store.get_snapshot().await {
Ok(Some(snapshot)) => InspectOutput {
if let Ok(Some(snapshot)) = run_store.get_snapshot().await {
return InspectOutput {
run_id: run_id.to_string(),
run_dir: run_dir.to_path_buf(),
status: snapshot
@ -70,36 +65,42 @@ async fn inspect_run_store(
sandbox: snapshot
.sandbox
.and_then(|record| serde_json::to_value(record).ok()),
},
_ => inspect_run_dir(run_id, run_dir, status),
};
}
}
fn inspect_run_dir(run_id: &RunId, run_dir: &Path, status: RunStatus) -> InspectOutput {
let run_record = RunRecord::load(run_dir)
.ok()
.and_then(|v| serde_json::to_value(v).ok());
let start_record = StartRecord::load(run_dir)
.ok()
.and_then(|v| serde_json::to_value(v).ok());
let conclusion = Conclusion::load(&run_dir.join("conclusion.json"))
.ok()
.and_then(|v| serde_json::to_value(v).ok());
let checkpoint = Checkpoint::load(&run_dir.join("checkpoint.json"))
.ok()
.and_then(|v| serde_json::to_value(v).ok());
let sandbox = fabro_sandbox::SandboxRecord::load(&run_dir.join("sandbox.json"))
.ok()
.and_then(|v| serde_json::to_value(v).ok());
InspectOutput {
run_id: run_id.to_string(),
run_dir: run_dir.to_path_buf(),
status,
run_record,
start_record,
conclusion,
checkpoint,
sandbox,
run_record: run_store
.get_run()
.await
.ok()
.flatten()
.and_then(|v| serde_json::to_value(v).ok()),
start_record: run_store
.get_start()
.await
.ok()
.flatten()
.and_then(|v| serde_json::to_value(v).ok()),
conclusion: run_store
.get_conclusion()
.await
.ok()
.flatten()
.and_then(|v| serde_json::to_value(v).ok()),
checkpoint: run_store
.get_checkpoint()
.await
.ok()
.flatten()
.and_then(|v| serde_json::to_value(v).ok()),
sandbox: run_store
.get_sandbox()
.await
.ok()
.flatten()
.and_then(|v| serde_json::to_value(v).ok()),
}
}

52
lib/crates/fabro-cli/tests/it/cmd/inspect.rs
View file

@ -105,6 +105,58 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() {
"###);
}
#[test]
fn inspect_completed_run_reads_store_without_disk_metadata_files() {
let context = test_context!();
let run = setup_completed_dry_run(&context);
for name in [
"run.json",
"start.json",
"conclusion.json",
"checkpoint.json",
"sandbox.json",
] {
std::fs::remove_file(run.run_dir.join(name)).unwrap();
}
let output = run_success(&context, &["inspect", &run.run_id]);
assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r###"
[
{
"run_id": "[ULID]",
"status": "succeeded",
"run_record": {
"goal": "Run tests and report results",
"workflow_name": "Simple",
"workflow_slug": "simple",
"sandbox_provider": "local",
"dry_run": true
},
"start_record": {
"has_start_time": true
},
"conclusion": {
"status": "success",
"duration_ms": "[DURATION_MS]",
"stage_count": 3
},
"checkpoint": {
"current_node": "report",
"completed_nodes": [
"start",
"run_tests",
"report"
],
"next_node_id": "exit"
},
"sandbox": {
"provider": "local"
}
}
]
"###);
}
#[test]
fn inspect_git_backed_run_exposes_checkpoint_and_sandbox_state() {
let context = test_context!();

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

@ -43,7 +43,7 @@ fn pr_create_unfinished_run_errors_before_network() {
exit_code: 1
----- stdout -----
----- stderr -----
error: Failed to load start.json
error: Failed to load start record from store
");
}
@ -68,6 +68,8 @@ fn pr_create_uses_store_run_record_without_run_json() {
let context = test_context!();
let run = setup_completed_dry_run(&context);
std::fs::remove_file(run.run_dir.join("run.json")).unwrap();
std::fs::remove_file(run.run_dir.join("start.json")).unwrap();
std::fs::remove_file(run.run_dir.join("conclusion.json")).unwrap();
let mut cmd = context.command();
cmd.args(["pr", "create", &run.run_id]);

3
lib/crates/fabro-cli/tests/it/cmd/pr_view.rs
View file

@ -63,8 +63,7 @@ fn pr_view_missing_pull_request_json_errors() {
exit_code: 1
----- stdout -----
----- stderr -----
error: No pull_request.json found in run directory. Create one first with: fabro pr create [ULID]
> No such file or directory (os error 2)
error: No pull request found in store. Create one first with: fabro pr create [ULID]
");
}

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

@ -71,6 +71,7 @@ fn resume_rewound_run_succeeds() {
&setup.repo_dir,
&["rev-parse", &format!("fabro/run/{}", setup.run.run_id)],
);
std::fs::remove_file(setup.run.run_dir.join("run.json")).unwrap();
let mut resume_cmd = context.command();
resume_cmd.current_dir(&setup.repo_dir);

24
lib/crates/fabro-cli/tests/it/cmd/sandbox_cp.rs
View file

@ -45,8 +45,7 @@ fn sandbox_cp_run_without_sandbox_json_errors_cleanly() {
exit_code: 1
----- stdout -----
----- stderr -----
error: Failed to load sandbox.json — was this run started with a recent version of arc?
> failed to read [DRY_RUN_DIR]/sandbox.json: No such file or directory (os error 2)
error: Failed to load sandbox record from store
");
}
@ -70,6 +69,27 @@ fn sandbox_cp_downloads_file_from_run() {
assert_eq!(read_text(&dest), "keep");
}
#[test]
fn sandbox_cp_downloads_file_from_store_without_sandbox_json() {
let context = test_context!();
let setup = setup_local_sandbox_run(&context);
std::fs::remove_file(setup.run.run_dir.join("sandbox.json")).unwrap();
let dest = context.temp_dir.join("downloaded-from-store.txt");
let mut cmd = context.cp();
cmd.args([
&format!("{}:sandbox_dir/download_me/root.txt", setup.run.run_id),
dest.to_str().unwrap(),
]);
fabro_snapshot!(context.filters(), cmd, @"
success: true
exit_code: 0
----- stdout -----
----- stderr -----
");
assert_eq!(read_text(&dest), "keep");
}
#[test]
fn sandbox_cp_uploads_file_to_run() {
let context = test_context!();

23
lib/crates/fabro-cli/tests/it/cmd/wait.rs
View file

@ -53,6 +53,29 @@ fn wait_completed_run_prints_success_summary() {
");
}
#[test]
fn wait_completed_run_reads_store_without_status_or_conclusion_files() {
let context = test_context!();
let run = setup_completed_dry_run(&context);
std::fs::remove_file(run.run_dir.join("status.json")).unwrap();
std::fs::remove_file(run.run_dir.join("conclusion.json")).unwrap();
let mut filters = context.filters();
filters.push((
r"\b\d+(\.\d+)?(ms|s)\b".to_string(),
"[DURATION]".to_string(),
));
let mut cmd = context.command();
cmd.args(["wait", &run.run_id]);
fabro_snapshot!(filters, cmd, @"
success: true
exit_code: 0
----- stdout -----
----- stderr -----
Succeeded [ULID] [DURATION]
");
}
#[test]
fn wait_completed_run_json_outputs_status_and_duration() {
let context = test_context!();