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

Fix clippy warnings: remove useless .into(), unnecessary async, large futures, and style lints

Browse source 
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-03 08:47:16 -07:00
parent 9c4923f338
commit 5737fcff66
No known key found for this signature in database
23 changed files with 241 additions and 562 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -21,7 +21,9 @@ pub(crate) async fn dispatch(ns: PrNamespace, globals: &GlobalArgs) -> Result<()
let github_app = build_github_app_credentials(cli_settings.app_id())?;
match ns.command {
PrCommand::Create(args) => create::create_command(args, github_app, globals).await,
PrCommand::Create(args) => {
Box::pin(create::create_command(args, github_app, globals)).await
}
PrCommand::List(args) => list::list_command(args, github_app, globals).await,
PrCommand::View(args) => view::view_command(args, github_app, globals).await,
PrCommand::Merge(args) => merge::merge_command(args, github_app, globals).await,

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

@ -137,9 +137,7 @@ async fn attach_run_store(
emit_progress_line(&mut progress_ui, line, json_output)?;
}
let mut stream = run_store
.watch_events_from(if last_seq == 0 { 1 } else { last_seq + 1 })
.await?;
let mut stream = run_store.watch_events_from(if last_seq == 0 { 1 } else { last_seq + 1 })?;
let mut next_seq = if last_seq == 0 { 1 } else { last_seq + 1 };
let mut cached_pid: Option<u32> = None;
let attach_started = Instant::now();
@ -745,29 +743,26 @@ fn determine_exit_code(conclusion_path: &Path, status_record: Option<RunStatusRe
async fn determine_exit_code_with_store(run_store: &SlateRunStore) -> ExitCode {
let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE;
loop {
match run_store.state().await {
Ok(state) => {
if let Some(conclusion) = state.conclusion {
let success = matches!(
conclusion.status,
StageStatus::Success | StageStatus::PartialSuccess
);
return if success {
ExitCode::from(0)
} else {
ExitCode::from(1)
};
}
match state.status {
Some(record) if matches!(record.status, RunStatus::Succeeded) => {
return ExitCode::from(0);
}
Some(record) if record.status.is_terminal() => return ExitCode::from(1),
Some(_) | None => {}
}
if let Ok(state) = run_store.state().await {
if let Some(conclusion) = state.conclusion {
let success = matches!(
conclusion.status,
StageStatus::Success | StageStatus::PartialSuccess
);
return if success {
ExitCode::from(0)
} else {
ExitCode::from(1)
};
}
match state.status {
Some(record) if matches!(record.status, RunStatus::Succeeded) => {
return ExitCode::from(0);
}
Some(record) if record.status.is_terminal() => return ExitCode::from(1),
Some(_) | None => {}
}
Err(_) => {}
}
if Instant::now() >= deadline {

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

@ -13,7 +13,7 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<(
let quiet = args.detach;
let prevent_idle_sleep = cli_settings.prevent_idle_sleep_enabled();
let (run_id, run_dir) = super::create::create_run(&args, cli, styles, quiet).await?;
let (run_id, run_dir) = Box::pin(super::create::create_run(&args, cli, styles, quiet)).await?;
#[cfg(feature = "sleep_inhibitor")]
let _sleep_guard = crate::sleep_inhibitor::guard(prevent_idle_sleep);

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

@ -39,7 +39,7 @@ pub(crate) async fn create_run(
let store = store::build_store(settings.storage_dir().as_path())?;
let created = match create(
let created = match Box::pin(create(
store.as_ref(),
CreateRunInput {
workflow: WorkflowInput::Path(workflow_path.clone()),
@ -51,7 +51,7 @@ pub(crate) async fn create_run(
base_branch: None,
host_repo_path: None,
},
)
))
.await
{
Ok(created) => created,

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

@ -147,7 +147,6 @@ async fn follow_store_logs(
) -> Result<()> {
let mut stream = run_store
.watch_events_from(seq)
.await
.context("Failed to watch store-backed run events")?;
let stdout = io::stdout();
let mut out = stdout.lock();

6
lib/crates/fabro-cli/src/commands/run/mod.rs
View file

@ -40,13 +40,13 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
match cmd {
RunCommands::Run(mut args) => {
apply_json_defaults(&mut args, globals);
command::execute(args, globals).await
Box::pin(command::execute(args, globals)).await
}
RunCommands::Create(mut args) => {
apply_json_defaults(&mut args, globals);
let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr()));
let cli = user_layer_with_globals(globals)?;
let (run_id, _run_dir) = create::create_run(&args, cli, styles, true).await?;
let (run_id, _run_dir) = Box::pin(create::create_run(&args, cli, styles, true)).await?;
if globals.json {
print_json_pretty(&serde_json::json!({ "run_id": run_id }))?;
} else {
@ -125,7 +125,7 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
}
RunCommands::Rewind(args) => {
let styles = Styles::detect_stderr();
rewind::run(&args, &styles, globals).await
Box::pin(rewind::run(&args, &styles, globals)).await
}
RunCommands::Fork(args) => {
let styles = Styles::detect_stderr();

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

@ -135,10 +135,6 @@ async fn reset_rewound_run_state(
let run_store = durable_store.open_run(run_id).await.map_err(|err| {
anyhow::anyhow!("failed to open durable store run for rewind reset: {err}")
})?;
run_store
.reset_for_rewind()
.await
.map_err(|err| anyhow::anyhow!("failed to clear rewound run state: {err}"))?;
append_workflow_event(
run_store.as_ref(),
run_id,

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

@ -224,7 +224,7 @@ async fn main_inner() -> (String, Result<()>) {
Commands::Install { web_url } => {
commands::install::run_install(&web_url, &globals).await?;
}
Commands::Pr(ns) => commands::pr::dispatch(ns, &globals).await?,
Commands::Pr(ns) => Box::pin(commands::pr::dispatch(ns, &globals)).await?,
Commands::Secret(ns) => commands::secret::dispatch(ns, &globals)?,
Commands::Settings(args) => commands::config::execute(&args, &globals)?,
Commands::Workflow(ns) => commands::workflow::dispatch(ns, &globals)?,

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

@ -207,7 +207,7 @@ pub async fn run_retro_agent(
let prompt = build_retro_prompt(RETRO_DATA_DIR);
write_retro_prompt(run_store, &retro_dir, &prompt).await?;
write_retro_prompt(run_store, &retro_dir, &prompt)?;
let process_result = session
.process_input(&prompt)
@ -250,7 +250,7 @@ pub async fn run_retro_agent(
};
// Write artifacts (on both success and failure)
write_retro_response(run_store, &retro_dir, &response_text).await?;
write_retro_response(run_store, &retro_dir, &response_text)?;
write_retro_artifacts(
&retro_dir,
provider.as_str(),
@ -285,7 +285,7 @@ pub fn dry_run_narrative() -> RetroNarrative {
}
}
async fn write_retro_prompt(
fn write_retro_prompt(
_run_store: &SlateRunStore,
retro_dir: &Path,
prompt: &str,
@ -294,7 +294,7 @@ async fn write_retro_prompt(
Ok(())
}
async fn write_retro_response(
fn write_retro_response(
_run_store: &SlateRunStore,
retro_dir: &Path,
response: &str,

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

@ -547,7 +547,7 @@ async fn start_run(
info!(run_id = %run_id, "Run queued");
let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4()));
let settings = state.settings.read().unwrap().clone();
let created = match operations::create(
let created = match Box::pin(operations::create(
state.store.as_ref(),
CreateRunInput {
workflow: WorkflowInput::DotSource {
@ -562,7 +562,7 @@ async fn start_run(
host_repo_path: None,
base_branch: None,
},
)
))
.await
{
Ok(created) => created,

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

@ -3,6 +3,7 @@ use std::path::PathBuf;
use std::str::FromStr;
use chrono::{DateTime, Utc};
use serde::de::DeserializeOwned;
use serde_json::Value;
use crate::{
@ -247,7 +248,7 @@ impl RunState {
};
let visit = required_u32(&properties, "visit")?;
self.node_mut(node_id, visit).provider_used =
provider_used_from_agent_event(event_name, &properties);
Some(provider_used_from_agent_event(event_name, &properties));
}
"command.started" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
@ -267,6 +268,13 @@ impl RunState {
node.stderr = optional_string(&properties, "stderr");
node.script_timing = Some(Value::Object(properties.clone()));
}
"parallel.completed" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = self.current_visit_for(node_id).unwrap_or(1);
self.node_mut(node_id, visit).parallel_results = properties.get("results").cloned();
}
_ => {}
}
@ -416,7 +424,7 @@ fn parse_ts(value: &Value) -> Result<DateTime<Utc>> {
.ok_or_else(|| StoreError::InvalidEvent("event payload missing ts".into()))?;
chrono::DateTime::parse_from_rfc3339(ts)
.map(|ts| ts.with_timezone(&Utc))
.map_err(|err| StoreError::InvalidEvent(format!("invalid event ts: {err}")).into())
.map_err(|err| StoreError::InvalidEvent(format!("invalid event ts: {err}")))
}
fn parse_run_id(value: &Value) -> Result<RunId> {
@ -426,7 +434,7 @@ fn parse_run_id(value: &Value) -> Result<RunId> {
.ok_or_else(|| StoreError::InvalidEvent("event payload missing run_id".into()))?;
run_id
.parse()
.map_err(|err| StoreError::InvalidEvent(format!("invalid run_id: {err}")).into())
.map_err(|err| StoreError::InvalidEvent(format!("invalid run_id: {err}")))
}
fn required_string(properties: &serde_json::Map<String, Value>, key: &str) -> Result<String> {
@ -434,9 +442,7 @@ fn required_string(properties: &serde_json::Map<String, Value>, key: &str) -> Re
.get(key)
.and_then(Value::as_str)
.map(ToString::to_string)
.ok_or_else(|| {
StoreError::InvalidEvent(format!("event missing string property {key}")).into()
})
.ok_or_else(|| StoreError::InvalidEvent(format!("event missing string property {key}")))
}
fn optional_string(properties: &serde_json::Map<String, Value>, key: &str) -> Option<String> {
@ -447,17 +453,18 @@ fn optional_string(properties: &serde_json::Map<String, Value>, key: &str) -> Op
}
fn required_u64(properties: &serde_json::Map<String, Value>, key: &str) -> Result<u64> {
properties.get(key).and_then(Value::as_u64).ok_or_else(|| {
StoreError::InvalidEvent(format!("event missing integer property {key}")).into()
})
properties
.get(key)
.and_then(Value::as_u64)
.ok_or_else(|| StoreError::InvalidEvent(format!("event missing integer property {key}")))
}
fn required_u32(properties: &serde_json::Map<String, Value>, key: &str) -> Result<u32> {
u32::try_from(required_u64(properties, key)?)
.map_err(|_| StoreError::InvalidEvent(format!("property {key} does not fit in u32")).into())
.map_err(|_| StoreError::InvalidEvent(format!("property {key} does not fit in u32")))
}
fn required_json<T: serde::de::DeserializeOwned>(
fn required_json<T: DeserializeOwned>(
properties: &serde_json::Map<String, Value>,
key: &str,
) -> Result<T> {
@ -466,10 +473,10 @@ fn required_json<T: serde::de::DeserializeOwned>(
.cloned()
.ok_or_else(|| StoreError::InvalidEvent(format!("event missing property {key}")))?;
serde_json::from_value(value)
.map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into())
.map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}")))
}
fn optional_json<T: serde::de::DeserializeOwned>(
fn optional_json<T: DeserializeOwned>(
properties: &serde_json::Map<String, Value>,
key: &str,
) -> Result<Option<T>> {
@ -478,9 +485,8 @@ fn optional_json<T: serde::de::DeserializeOwned>(
.filter(|value| !value.is_null())
.cloned()
.map(|value| {
serde_json::from_value(value).map_err(|err| {
StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into()
})
serde_json::from_value(value)
.map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}")))
})
.transpose()
}
@ -488,9 +494,8 @@ fn optional_json<T: serde::de::DeserializeOwned>(
fn parse_reason(properties: &serde_json::Map<String, Value>) -> Result<Option<StatusReason>> {
optional_string(properties, "reason")
.map(|reason| {
serde_json::from_value(Value::String(reason)).map_err(|err| {
StoreError::InvalidEvent(format!("invalid status reason: {err}")).into()
})
serde_json::from_value(Value::String(reason))
.map_err(|err| StoreError::InvalidEvent(format!("invalid status reason: {err}")))
})
.transpose()
}
@ -662,7 +667,7 @@ fn provider_used_from_prompt(properties: &serde_json::Map<String, Value>) -> Opt
fn provider_used_from_agent_event(
event_name: &str,
properties: &serde_json::Map<String, Value>,
) -> Option<Value> {
) -> Value {
let mut provider_used = serde_json::Map::new();
provider_used.insert(
"mode".to_string(),
@ -681,5 +686,5 @@ fn provider_used_from_agent_event(
if let Some(command) = optional_string(properties, "command") {
provider_used.insert("command".to_string(), Value::String(command));
}
Some(Value::Object(provider_used))
Value::Object(provider_used)
}

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

@ -824,7 +824,7 @@ mod tests {
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let mut stream = run.watch_events_from(1).await.unwrap();
let mut stream = run.watch_events_from(1).unwrap();
run.append_event(&event_payload(
"run-1",
@ -1018,7 +1018,7 @@ mod tests {
}
#[tokio::test]
async fn slate_run_store_round_trips_node_data_and_assets() {
async fn slate_run_store_round_trips_assets_and_projects_events() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
@ -1029,16 +1029,22 @@ mod tests {
node_id: "code",
visit: 2,
};
run.put_node_prompt(&node, "Plan").await.unwrap();
run.put_node_status(&node, &sample_node_status())
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:01:00Z",
"stage.prompt",
Some("code"),
serde_json::json!({"text": "Plan", "visit": 2}),
))
.await
.unwrap();
run.put_asset(&node, "src/lib.rs", b"fn main() {}")
.await
.unwrap();
let snapshot = run.get_node(&node).await.unwrap();
assert_eq!(snapshot.prompt, Some("Plan".to_string()));
let state = run.state().await.unwrap();
let node_state = state.node(&node).unwrap();
assert_eq!(node_state.prompt, Some("Plan".to_string()));
assert_eq!(
run.get_asset(&node, "src/lib.rs").await.unwrap(),
Some(Bytes::from_static(b"fn main() {}"))
@ -1050,7 +1056,7 @@ mod tests {
}
#[tokio::test]
async fn slate_run_store_lists_artifact_values_and_asset_only_visits() {
async fn slate_run_store_lists_artifact_values_and_assets() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
@ -1068,7 +1074,6 @@ mod tests {
node_id: "code",
visit: 2,
};
run.put_node_prompt(&snapshot_node, "Plan").await.unwrap();
run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}")
.await
.unwrap();
@ -1096,8 +1101,5 @@ mod tests {
("code".to_string(), 2, "src/lib.rs".to_string())
]
);
let code_node = run.get_node(&snapshot_node).await.unwrap();
assert_eq!(code_node.node_id, "code");
}
}

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

@ -1,4 +1,3 @@
use std::collections::BTreeSet;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::{Arc, Weak};
use std::time::Duration;
@ -16,10 +15,10 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::run_state::EventProjectionCache;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef,
Result, RunState, RunSummary, StoreError,
CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, Result, RunState, RunSummary,
StoreError,
};
use fabro_types::{NodeStatusRecord, RunId};
use fabro_types::RunId;
#[derive(Clone)]
pub struct SlateRunStore {
@ -152,40 +151,6 @@ impl SlateRunStore {
Ok(state.build_summary(catalog))
}
async fn build_node_snapshot(&self, node: &NodeVisitRef<'_>) -> Result<NodeSnapshot> {
Ok(NodeSnapshot {
node_id: node.node_id.to_string(),
visit: node.visit,
prompt: self.inner.db.get_text(&keys::node_prompt(node)).await?,
response: self.inner.db.get_text(&keys::node_response(node)).await?,
status: self.inner.db.get_json(&keys::node_status(node)).await?,
outcome: self.inner.db.get_json(&keys::node_outcome(node)).await?,
provider_used: self
.inner
.db
.get_json(&keys::node_provider_used(node))
.await?,
diff: self.inner.db.get_text(&keys::node_diff(node)).await?,
script_invocation: self
.inner
.db
.get_json(&keys::node_script_invocation(node))
.await?,
script_timing: self
.inner
.db
.get_json(&keys::node_script_timing(node))
.await?,
parallel_results: self
.inner
.db
.get_json(&keys::node_parallel_results(node))
.await?,
stdout: self.inner.db.get_text(&keys::node_stdout(node)).await?,
stderr: self.inner.db.get_text(&keys::node_stderr(node)).await?,
})
}
async fn projected_state(&self) -> Result<RunState> {
let next_seq = {
let cache = self.inner.projection_cache.lock().await;
@ -202,157 +167,6 @@ impl SlateRunStore {
}
impl SlateRunStore {
pub async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> {
self.inner
.db
.put_text(&keys::node_prompt(node), prompt)
.await
}
pub async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> {
self.inner
.db
.put_text(&keys::node_response(node), response)
.await
}
pub async fn put_node_status(
&self,
node: &NodeVisitRef<'_>,
status: &NodeStatusRecord,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_status(node), status)
.await
}
pub async fn put_node_outcome(
&self,
node: &NodeVisitRef<'_>,
outcome: &NodeOutcomeRecord,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_outcome(node), outcome)
.await
}
pub async fn put_node_provider_used(
&self,
node: &NodeVisitRef<'_>,
provider_used: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_provider_used(node), provider_used)
.await
}
pub async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> {
self.inner.db.put_text(&keys::node_diff(node), diff).await
}
pub async fn put_node_script_invocation(
&self,
node: &NodeVisitRef<'_>,
invocation: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_script_invocation(node), invocation)
.await
}
pub async fn put_node_script_timing(
&self,
node: &NodeVisitRef<'_>,
timing: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_script_timing(node), timing)
.await
}
pub async fn put_node_parallel_results(
&self,
node: &NodeVisitRef<'_>,
results: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_parallel_results(node), results)
.await
}
pub async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
self.inner.db.put_text(&keys::node_stdout(node), log).await
}
pub async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
self.inner.db.put_text(&keys::node_stderr(node), log).await
}
pub async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result<NodeSnapshot> {
self.build_node_snapshot(node).await
}
pub async fn list_node_visits(&self, node_id: &str) -> Result<Vec<u32>> {
let prefix = format!("nodes/{node_id}/visit-");
let mut iter = self.inner.db.scan_prefix(prefix.as_bytes()).await?;
let mut visits = BTreeSet::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
if let Some((current_node_id, visit, _)) = keys::parse_node_key(&key) {
if current_node_id == node_id {
visits.insert(visit);
}
}
}
Ok(visits.into_iter().collect())
}
pub async fn list_node_ids(&self) -> Result<Vec<String>> {
let mut iter = self.inner.db.scan_prefix(b"nodes/").await?;
let mut node_ids = BTreeSet::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
if let Some((node_id, _, _)) = keys::parse_node_key(&key) {
node_ids.insert(node_id);
}
}
let mut asset_iter = self
.inner
.db
.scan_prefix(keys::ARTIFACT_NODES_PREFIX.as_bytes())
.await?;
while let Some(entry) = asset_iter.next().await? {
let key = key_to_string(&entry.key)?;
if let Some((node_id, _, _)) = keys::parse_node_asset_key(&key) {
node_ids.insert(node_id);
}
}
Ok(node_ids.into_iter().collect())
}
pub async fn reset_for_rewind(&self) -> Result<()> {
let db = self.inner.db.writer()?;
for key in [keys::retro_prompt(), keys::retro_response()] {
db.delete(key).await?;
}
for prefix in [
b"nodes/".as_slice(),
keys::ARTIFACT_VALUES_PREFIX.as_bytes(),
keys::ARTIFACT_NODES_PREFIX.as_bytes(),
] {
delete_prefix(db, prefix).await?;
}
Ok(())
}
pub async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
payload.validate(&self.inner.run_id)?;
let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst);
@ -374,7 +188,7 @@ impl SlateRunStore {
self.inner.db.list_events_from(seq).await
}
pub async fn watch_events_from(
pub fn watch_events_from(
&self,
seq: u32,
) -> Result<std::pin::Pin<Box<dyn Stream<Item = Result<EventEnvelope>> + Send>>> {
@ -412,22 +226,6 @@ impl SlateRunStore {
Ok(Box::pin(UnboundedReceiverStream::new(receiver)))
}
pub async fn put_retro_prompt(&self, text: &str) -> Result<()> {
self.inner.db.put_text(keys::retro_prompt(), text).await
}
pub async fn get_retro_prompt(&self) -> Result<Option<String>> {
self.inner.db.get_text(keys::retro_prompt()).await
}
pub async fn put_retro_response(&self, text: &str) -> Result<()> {
self.inner.db.put_text(keys::retro_response(), text).await
}
pub async fn get_retro_response(&self) -> Result<Option<String>> {
self.inner.db.get_text(keys::retro_response()).await
}
pub async fn put_artifact_value(
&self,
artifact_id: &str,
@ -522,17 +320,6 @@ impl SlateRunDb {
put_json(self.writer()?, key, value).await
}
async fn get_text(&self, key: &str) -> Result<Option<String>> {
match self {
Self::Writer(db) => get_text(db, key).await,
Self::Reader(db) => get_text(db.as_ref(), key).await,
}
}
async fn put_text(&self, key: &str, value: &str) -> Result<()> {
put_text(self.writer()?, key, value).await
}
async fn get_bytes(&self, key: &str) -> Result<Option<Bytes>> {
match self {
Self::Writer(db) => get_bytes(db, key).await,
@ -596,41 +383,11 @@ where
.map_err(Into::into)
}
async fn put_text(db: &slatedb::Db, key: &str, value: &str) -> Result<()> {
db.put(key, value.as_bytes()).await?;
Ok(())
}
async fn get_text<R>(db: &R, key: &str) -> Result<Option<String>>
where
R: DbRead + Sync,
{
db.get(key)
.await?
.map(|value| {
String::from_utf8(value.to_vec())
.map_err(|err| StoreError::Other(format!("stored text is not valid UTF-8: {err}")))
})
.transpose()
}
async fn put_bytes(db: &slatedb::Db, key: &str, value: &[u8]) -> Result<()> {
db.put(key, value).await?;
Ok(())
}
async fn delete_prefix(db: &slatedb::Db, prefix: &[u8]) -> Result<()> {
let mut iter = db.scan_prefix(prefix).await?;
let mut keys = Vec::new();
while let Some(entry) = iter.next().await? {
keys.push(key_to_string(&entry.key)?);
}
for key in keys {
db.delete(key).await?;
}
Ok(())
}
async fn get_bytes(db: &slatedb::Db, key: &str) -> Result<Option<Bytes>> {
Ok(db.get(key).await?)
}

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

@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicI64, Ordering};
use anyhow::{Context, Result};
use chrono::{SecondsFormat, Utc};
use fabro_store::{EventPayload, NodeVisitRef, RunStoreHandle, SlateRunStore};
use fabro_store::{EventPayload, RunStoreHandle, SlateRunStore};
use fabro_types::RunId;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
@ -1572,15 +1572,6 @@ impl StoreProgressLogger {
if let Err(err) = run_store.append_event(&payload).await {
tracing::warn!(error = %err, "Failed to append event to run store");
}
if let Err(err) =
project_provider_used_from_event_payload(run_store.as_ref(), &payload)
.await
{
tracing::warn!(
error = %err,
"Failed to project provider metadata from event"
);
}
}
StoreProgressCommand::Flush(tx) => {
let _ = tx.send(());
@ -1626,80 +1617,6 @@ impl StoreProgressLogger {
}
}
async fn project_provider_used_from_event_payload(
run_store: &SlateRunStore,
payload: &EventPayload,
) -> Result<()> {
let value = payload.as_value();
let Some(event_name) = value.get("event").and_then(Value::as_str) else {
return Ok(());
};
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let Some(properties) = value.get("properties").and_then(Value::as_object) else {
return Ok(());
};
let Some(visit) = properties
.get("visit")
.and_then(Value::as_u64)
.and_then(|visit| u32::try_from(visit).ok())
else {
return Ok(());
};
let provider_used = match event_name {
"stage.prompt" => {
let mut provider_used = Map::new();
if let Some(mode) = properties.get("mode").and_then(Value::as_str) {
provider_used.insert("mode".to_string(), Value::String(mode.to_string()));
}
if let Some(provider) = properties.get("provider").and_then(Value::as_str) {
provider_used.insert("provider".to_string(), Value::String(provider.to_string()));
}
if let Some(model) = properties.get("model").and_then(Value::as_str) {
provider_used.insert("model".to_string(), Value::String(model.to_string()));
}
(!provider_used.is_empty()).then_some(Value::Object(provider_used))
}
"agent.session.started" => {
let mut provider_used = Map::new();
provider_used.insert("mode".to_string(), Value::String("agent".to_string()));
if let Some(provider) = properties.get("provider").and_then(Value::as_str) {
provider_used.insert("provider".to_string(), Value::String(provider.to_string()));
}
if let Some(model) = properties.get("model").and_then(Value::as_str) {
provider_used.insert("model".to_string(), Value::String(model.to_string()));
}
Some(Value::Object(provider_used))
}
"agent.cli.started" => {
let mut provider_used = Map::new();
provider_used.insert("mode".to_string(), Value::String("cli".to_string()));
if let Some(provider) = properties.get("provider").and_then(Value::as_str) {
provider_used.insert("provider".to_string(), Value::String(provider.to_string()));
}
if let Some(model) = properties.get("model").and_then(Value::as_str) {
provider_used.insert("model".to_string(), Value::String(model.to_string()));
}
if let Some(command) = properties.get("command").and_then(Value::as_str) {
provider_used.insert("command".to_string(), Value::String(command.to_string()));
}
Some(Value::Object(provider_used))
}
_ => None,
};
let Some(provider_used) = provider_used else {
return Ok(());
};
run_store
.put_node_provider_used(&NodeVisitRef { node_id, visit }, &provider_used)
.await
.map_err(anyhow::Error::from)
}
/// Current time as epoch milliseconds.
fn epoch_millis() -> i64 {
let millis = std::time::SystemTime::now()

201
lib/crates/fabro-workflow/src/git.rs
View file

@ -2,7 +2,7 @@ use std::path::Path;
use std::process::Command;
use fabro_checkpoint::git::Store;
use fabro_store::{NodeVisitRef, SlateRunStore};
use fabro_store::RunState;
use fabro_types::Settings;
use crate::error::{FabroError, Result};
@ -353,76 +353,64 @@ pub fn scan_node_files(run_dir: &Path) -> Vec<(String, Vec<u8>)> {
result
}
pub async fn scan_node_files_from_store(run_store: &SlateRunStore) -> Vec<(String, Vec<u8>)> {
pub fn scan_node_files_from_state(state: &RunState) -> Vec<(String, Vec<u8>)> {
let mut result = Vec::new();
let Ok(node_ids) = run_store.list_node_ids().await else {
return result;
};
let mut keys: Vec<_> = state.nodes.keys().collect();
keys.sort();
for node_id in node_ids {
let Ok(visits) = run_store.list_node_visits(&node_id).await else {
for (node_id, visit) in keys {
let Some(node) = state.nodes.get(&(node_id.clone(), *visit)) else {
continue;
};
for visit in visits {
let Ok(node) = run_store
.get_node(&NodeVisitRef {
node_id: &node_id,
visit,
})
.await
else {
continue;
};
if let Some(prompt) = node.prompt {
if let Some(ref prompt) = node.prompt {
result.push((
node_file_path(node_id, *visit, "prompt.md"),
prompt.as_bytes().to_vec(),
));
}
if let Some(ref response) = node.response {
result.push((
node_file_path(node_id, *visit, "response.md"),
response.as_bytes().to_vec(),
));
}
if let Some(ref status) = node.status {
if let Ok(bytes) = serde_json::to_vec_pretty(status) {
result.push((node_file_path(node_id, *visit, "status.json"), bytes));
}
}
if let Some(ref provider_used) = node.provider_used {
if let Ok(bytes) = serde_json::to_vec_pretty(provider_used) {
result.push((node_file_path(node_id, *visit, "provider_used.json"), bytes));
}
}
if let Some(ref diff) = node.diff {
result.push((
node_file_path(node_id, *visit, "diff.patch"),
diff.as_bytes().to_vec(),
));
}
if let Some(ref script_invocation) = node.script_invocation {
if let Ok(bytes) = serde_json::to_vec_pretty(script_invocation) {
result.push((
node_file_path(&node_id, visit, "prompt.md"),
prompt.into_bytes(),
node_file_path(node_id, *visit, "script_invocation.json"),
bytes,
));
}
if let Some(response) = node.response {
}
if let Some(ref script_timing) = node.script_timing {
if let Ok(bytes) = serde_json::to_vec_pretty(script_timing) {
result.push((node_file_path(node_id, *visit, "script_timing.json"), bytes));
}
}
if let Some(ref parallel_results) = node.parallel_results {
if let Ok(bytes) = serde_json::to_vec_pretty(parallel_results) {
result.push((
node_file_path(&node_id, visit, "response.md"),
response.into_bytes(),
node_file_path(node_id, *visit, "parallel_results.json"),
bytes,
));
}
if let Some(status) = node.status {
if let Ok(bytes) = serde_json::to_vec_pretty(&status) {
result.push((node_file_path(&node_id, visit, "status.json"), bytes));
}
}
if let Some(provider_used) = node.provider_used {
if let Ok(bytes) = serde_json::to_vec_pretty(&provider_used) {
result.push((node_file_path(&node_id, visit, "provider_used.json"), bytes));
}
}
if let Some(diff) = node.diff {
result.push((
node_file_path(&node_id, visit, "diff.patch"),
diff.into_bytes(),
));
}
if let Some(script_invocation) = node.script_invocation {
if let Ok(bytes) = serde_json::to_vec_pretty(&script_invocation) {
result.push((
node_file_path(&node_id, visit, "script_invocation.json"),
bytes,
));
}
}
if let Some(script_timing) = node.script_timing {
if let Ok(bytes) = serde_json::to_vec_pretty(&script_timing) {
result.push((node_file_path(&node_id, visit, "script_timing.json"), bytes));
}
}
if let Some(parallel_results) = node.parallel_results {
if let Ok(bytes) = serde_json::to_vec_pretty(&parallel_results) {
result.push((
node_file_path(&node_id, visit, "parallel_results.json"),
bytes,
));
}
}
}
}
@ -440,9 +428,8 @@ fn node_file_path(node_id: &str, visit: u32, filename: &str) -> String {
#[cfg(test)]
mod tests {
use super::*;
use chrono::Utc;
use fabro_store::SlateStore;
use fabro_types::{NodeStatusRecord, StageStatus, fixtures};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use std::fs;
use std::sync::Arc;
@ -592,52 +579,78 @@ mod tests {
}
#[tokio::test]
async fn scan_node_files_from_store_reconstructs_allowlisted_entries() {
async fn scan_node_files_from_state_reconstructs_allowlisted_entries() {
use fabro_store::EventPayload;
let store = test_store();
let run = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let node = NodeVisitRef {
node_id: "work",
visit: 2,
let run_id_str = fixtures::RUN_1.to_string();
let event = |event_name: &str, props: serde_json::Value| -> EventPayload {
let value = serde_json::json!({
"id": format!("evt-{event_name}"),
"ts": "2026-03-27T12:01:00Z",
"run_id": run_id_str,
"event": event_name,
"node_id": "work",
"properties": props,
});
EventPayload::new(value, &fixtures::RUN_1).unwrap()
};
run.put_node_prompt(&node, "hello").await.unwrap();
run.put_node_response(&node, "world").await.unwrap();
run.put_node_status(
&node,
&NodeStatusRecord {
status: StageStatus::Success,
notes: None,
failure_reason: None,
timestamp: Utc::now(),
},
)
run.append_event(&event(
"stage.prompt",
serde_json::json!({"text": "hello", "visit": 2, "mode": "prompt", "provider": "openai"}),
))
.await
.unwrap();
run.append_event(&event(
"prompt.completed",
serde_json::json!({"response": "world"}),
))
.await
.unwrap();
run.append_event(&event(
"stage.completed",
serde_json::json!({"response": "world", "status": "success", "visit": 2}),
))
.await
.unwrap();
run.append_event(&event(
"command.started",
serde_json::json!({"command": "echo hi"}),
))
.await
.unwrap();
run.append_event(&event(
"command.completed",
serde_json::json!({"stdout": "hi\n", "stderr": "", "exit_code": 0}),
))
.await
.unwrap();
run.append_event(&event(
"parallel.completed",
serde_json::json!({"results": [{"id": "a"}], "duration_ms": 100, "success_count": 1, "failure_count": 0}),
))
.await
.unwrap();
run.append_event(&event(
"checkpoint.completed",
serde_json::json!({"diff": "diff --git a/story.txt b/story.txt", "ordinal": 1}),
))
.await
.unwrap();
run.put_node_provider_used(&node, &serde_json::json!({"provider":"openai"}))
.await
.unwrap();
run.put_node_diff(&node, "diff --git a/story.txt b/story.txt")
.await
.unwrap();
run.put_node_script_invocation(&node, &serde_json::json!({"command":"echo hi"}))
.await
.unwrap();
run.put_node_script_timing(&node, &serde_json::json!({"exit_code":0}))
.await
.unwrap();
run.put_node_parallel_results(&node, &serde_json::json!([{"id":"a"}]))
.await
.unwrap();
let files = scan_node_files_from_store(run.as_ref()).await;
let state = run.state().await.unwrap();
let files = scan_node_files_from_state(&state);
let paths: Vec<&str> = files.iter().map(|(path, _)| path.as_str()).collect();
assert!(paths.contains(&"nodes/work-visit_2/prompt.md"));
assert!(paths.contains(&"nodes/work-visit_2/response.md"));
assert!(paths.contains(&"nodes/work-visit_2/status.json"));
assert!(paths.contains(&"nodes/work-visit_2/provider_used.json"));
assert!(paths.contains(&"nodes/work-visit_2/diff.patch"));
assert!(paths.contains(&"nodes/work-visit_2/script_invocation.json"));
assert!(paths.contains(&"nodes/work-visit_2/script_timing.json"));
assert!(paths.contains(&"nodes/work-visit_2/parallel_results.json"));

52
lib/crates/fabro-workflow/src/handler/fan_in.rs
View file

@ -1,17 +1,15 @@
use std::path::Path;
use std::sync::Arc;
use async_trait::async_trait;
use fabro_agent::Sandbox;
use fabro_store::{NodeVisitRef, RunStoreHandle};
use crate::context::Context;
use crate::context::keys;
use crate::error::FabroError;
use crate::event::EventEmitter;
use crate::event::{EventEmitter, WorkflowRunEvent};
use crate::outcome::{Outcome, OutcomeExt};
use crate::run_dir::{node_dir, visit_from_context};
use crate::sandbox_git::git_merge_ff_only;
use async_trait::async_trait;
use fabro_agent::Sandbox;
use fabro_graphviz::graph::{Graph, Node};
use tokio::fs;
@ -90,7 +88,6 @@ impl Handler for FanInHandler {
&node.id,
&services.emitter,
&services.sandbox,
services.run_store.clone(),
)
.await?
} else {
@ -226,7 +223,6 @@ async fn llm_evaluate(
node_id: &str,
emitter: &Arc<EventEmitter>,
sandbox: &Arc<dyn Sandbox>,
run_store: RunStoreHandle,
) -> Result<Candidate, FabroError> {
let results_text =
serde_json::to_string_pretty(results).unwrap_or_else(|_| results.to_string());
@ -236,20 +232,21 @@ async fn llm_evaluate(
Respond with the ID of the best candidate."
);
// Write prompt to logs
let visit = visit_from_context(context);
let visit_u32 = u32::try_from(visit).unwrap_or(u32::MAX);
let stage_dir = node_dir(run_dir, node_id, visit);
fs::create_dir_all(&stage_dir).await?;
let node_ref = NodeVisitRef {
node_id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
run_store
.put_node_prompt(&node_ref, &full_prompt)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
fs::write(stage_dir.join("prompt.md"), &full_prompt).await?;
emitter.emit(&WorkflowRunEvent::Prompt {
stage: node_id.to_string(),
visit: visit_u32,
text: full_prompt.clone(),
mode: Some("fan_in".to_string()),
provider: None,
model: None,
});
// Build a synthetic node for the backend call
let eval_node = Node::new("fan_in_eval");
@ -278,10 +275,13 @@ async fn llm_evaluate(
.unwrap_or_else(|| "unknown".to_string());
let response_text =
serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string());
run_store
.put_node_response(&node_ref, &response_text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
emitter.emit(&WorkflowRunEvent::PromptCompleted {
node_id: node_id.to_string(),
response: response_text.clone(),
model: String::new(),
provider: String::new(),
usage: None,
});
fs::write(stage_dir.join("response.md"), &response_text).await?;
Ok(Candidate {
id: best_id,
@ -290,11 +290,13 @@ async fn llm_evaluate(
})
}
Ok(CodergenResult::Text { text, .. }) => {
// Write response to logs
run_store
.put_node_response(&node_ref, &text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
emitter.emit(&WorkflowRunEvent::PromptCompleted {
node_id: node_id.to_string(),
response: text.clone(),
model: String::new(),
provider: String::new(),
usage: None,
});
fs::write(stage_dir.join("response.md"), &text).await?;
// The LLM responded with text; try to find a matching candidate ID

9
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

@ -481,15 +481,6 @@ impl Handler for ParallelHandler {
let visit = visit_from_context(context);
let node_dir = node_dir(run_dir, &node.id, visit);
let _ = fs::create_dir_all(&node_dir).await;
let node_ref = NodeVisitRef {
node_id: &node.id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
services
.run_store
.put_node_parallel_results(&node_ref, &serde_json::json!(results_json))
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
if let Ok(json) = serde_json::to_string_pretty(&results_json) {
let _ = fs::write(node_dir.join("parallel_results.json"), json).await;
}

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

@ -14,7 +14,7 @@ use fabro_core::state::RunState;
use crate::artifact::ArtifactStore;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::git::MetadataStore;
use crate::git::scan_node_files_from_store;
use crate::git::scan_node_files_from_state;
use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode;
use crate::outcome::{Outcome, StageStatus, StageUsage};
@ -149,7 +149,9 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
})
.collect()
};
extra_entries.extend(scan_node_files_from_store(self.run_store.as_ref()).await);
if let Ok(store_state) = self.run_store.state().await {
extra_entries.extend(scan_node_files_from_state(&store_state));
}
let extra_refs: Vec<(&str, &[u8])> = extra_entries
.iter()
.map(|(k, v)| (k.as_str(), v.as_slice()))

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

@ -148,7 +148,7 @@ async fn persist_created_run(
.open_run(&record.run_id)
.await
.map_err(|open_err| FabroError::engine(open_err.to_string()))
.or_else(|_| Err(FabroError::engine(err.to_string())))?,
.map_err(|_| FabroError::engine(err.to_string()))?,
};
let envelope = canonicalize_event_at(

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

@ -553,7 +553,7 @@ impl RunSession {
};
let retro_start = Instant::now();
let retroed = pipeline::retro(executed, &retro_opts).await;
let retroed = Box::pin(pipeline::retro(executed, &retro_opts)).await;
let retro_duration = retro_start.elapsed();
let finalize_opts = FinalizeOptions {

17
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -3,7 +3,7 @@ use std::sync::Arc;
use crate::error::FabroError;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::git::{MetadataStore, scan_node_files_from_store};
use crate::git::{MetadataStore, scan_node_files_from_state};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::records::{Checkpoint, Conclusion, StageSummary};
use crate::run_options::RunOptions;
@ -195,13 +195,14 @@ pub async fn write_finalize_commit(
let git_author = run_options.git_author();
let store = MetadataStore::new(repo_path, &git_author);
let mut entries = scan_node_files_from_store(run_store).await;
let retro_bytes = run_store
.state()
.await
.ok()
.and_then(|state| state.retro)
.and_then(|retro| serde_json::to_vec_pretty(&retro).ok());
let Ok(store_state) = run_store.state().await else {
return;
};
let mut entries = scan_node_files_from_state(&store_state);
let retro_bytes = store_state
.retro
.as_ref()
.and_then(|retro| serde_json::to_vec_pretty(retro).ok());
if let Some(bytes) = retro_bytes {
entries.push(("retro.json".to_string(), bytes));
}

19
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -23,18 +23,15 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
return None;
}
};
let cp = match state.checkpoint {
Some(cp) => cp,
None => {
tracing::warn!("Could not load checkpoint, skipping retro");
if let Some(ref emitter) = options.emitter {
emitter.emit(&WorkflowRunEvent::RetroFailed {
error: "checkpoint not found".to_string(),
duration_ms: 0,
});
}
return None;
let Some(cp) = state.checkpoint else {
tracing::warn!("Could not load checkpoint, skipping retro");
if let Some(ref emitter) = options.emitter {
emitter.emit(&WorkflowRunEvent::RetroFailed {
error: "checkpoint not found".to_string(),
duration_ms: 0,
});
}
return None;
};
let completed_stages = crate::build_completed_stages(&cp, options.failed);

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

@ -11,7 +11,7 @@ use object_store::memory::InMemory;
use crate::error::Result;
use crate::event::{EventEmitter, WorkflowRunEvent, append_workflow_event};
use crate::git::scan_node_files_from_store;
use crate::git::scan_node_files_from_state;
use crate::handler::HandlerRegistry;
use crate::outcome::Outcome;
use crate::pipeline;
@ -199,7 +199,7 @@ async fn persist_run_artifacts_for_tests(run_store: &SlateRunStore, run_dir: &st
let _ = std::fs::write(run_dir.join("final.patch"), final_patch);
}
for (relative_path, contents) in scan_node_files_from_store(run_store).await {
for (relative_path, contents) in scan_node_files_from_state(&state) {
let path = run_dir.join(relative_path);
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);