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

Remove dead code, double-Arc, and back-compat shims

Browse source 
- Remove RunStoreHandle type alias (double-Arc); use SlateRunStore directly
- Remove RunState::list_node_ids (test-only, inline in tests)
- Remove SlateRunStore::list_assets and scan_prefix (test-only)
- Remove RuntimeState::ensure_runtime_dir/ensure_artifact_values_dir (test-only)
- Remove read_dot_source and its 4 tests (reads files no longer written)
- Remove back-compat modules transform/vars/stylesheet; update integration test
- Remove unused InvalidTransition/ParseRunStatusError re-exports
- Fix crate::vars import to crate::transforms::variable_expansion

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-03 16:24:44 -07:00
parent 0577de38f9
commit 0844294440
No known key found for this signature in database
39 changed files with 97 additions and 168 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -32,7 +32,6 @@ async fn create_from(
let storage_dir = base.parent().unwrap_or(base);
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 state = run_store.state().await?;
@ -112,8 +111,7 @@ async fn create_from(
&model,
true,
None,
run_store.as_ref(),
&run_dir,
&run_store,
None,
)
.await

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

@ -63,7 +63,7 @@ pub(crate) async fn attach_run(
.collect::<Result<Vec<_>>>()?;
return attach_run_store(
run_dir,
run_store.as_ref(),
&run_store,
verbose,
event_lines,
events.last().map_or(0, |event| event.seq),

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

@ -20,7 +20,7 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> {
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let patch = resolve_diff(&run.path, run_store.as_ref(), &args).await?;
let patch = resolve_diff(&run.path, &run_store, &args).await?;
if globals.json {
let mut value = serde_json::json!({

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

@ -64,7 +64,7 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
if args.follow {
follow_store_logs(
run_store.as_ref(),
&run_store,
&run.path,
if last_seq == 0 { 1 } else { last_seq + 1 },
pretty,

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

@ -103,7 +103,7 @@ pub(crate) async fn print_run_summary(
pr_url.as_deref(),
styles,
);
print_final_output(run_store.as_deref(), run_dir, styles).await;
print_final_output(run_store.as_ref(), run_dir, styles).await;
print_assets(run_dir, styles);
Ok(())
}

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

@ -134,7 +134,7 @@ async fn reset_rewound_run_state(
anyhow::anyhow!("failed to open durable store run for rewind reset: {err}")
})?;
append_workflow_event(
run_store.as_ref(),
&run_store,
run_id,
&WorkflowRunEvent::RunRewound {
target_checkpoint_ordinal: entry.ordinal,
@ -146,15 +146,11 @@ async fn reset_rewound_run_state(
)
.await
.map_err(|err| anyhow::anyhow!("failed to append run rewound event: {err}"))?;
append_workflow_event(&run_store, run_id, &restored_checkpoint_event(&checkpoint))
.await
.map_err(|err| anyhow::anyhow!("failed to append restored checkpoint event: {err}"))?;
append_workflow_event(
run_store.as_ref(),
run_id,
&restored_checkpoint_event(&checkpoint),
)
.await
.map_err(|err| anyhow::anyhow!("failed to append restored checkpoint event: {err}"))?;
append_workflow_event(
run_store.as_ref(),
&run_store,
run_id,
&WorkflowRunEvent::RunSubmitted { reason: None },
)

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

@ -29,7 +29,7 @@ pub(crate) async fn run(args: &InspectArgs, globals: &GlobalArgs) -> Result<()>
let store = store::build_store(&cli_settings.storage_dir())?;
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let output = inspect_run_store(&run.run_id, &run.path, run.status, run_store.as_ref()).await;
let output = inspect_run_store(&run.run_id, &run.path, run.status, &run_store).await;
let json = serde_json::to_string_pretty(&[output])?;
println!("{json}");
Ok(())

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

@ -126,7 +126,7 @@ async fn remove_run_dir_with_cleanup(store: &SlateStore, run: &RunInfo) -> Resul
};
if let Some(run_store) = run_store.as_ref() {
if let Err(err) = append_workflow_event(
run_store.as_ref(),
run_store,
&run.run_id,
&WorkflowRunEvent::RunRemoving { reason: None },
)
@ -140,7 +140,7 @@ async fn remove_run_dir_with_cleanup(store: &SlateStore, run: &RunInfo) -> Resul
}
}
if let Some(record) = load_sandbox_record(&run.path, run_store.as_deref()).await {
if let Some(record) = load_sandbox_record(&run.path, run_store.as_ref()).await {
if record.provider != "local" {
match reconnect_sandbox(&record).await {
Ok(sandbox) => {

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

@ -22,7 +22,7 @@ pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) ->
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let file_count = export_run(run_store.as_ref(), &args.output).await?;
let file_count = export_run(&run_store, &args.output).await?;
if globals.json {
print_json_pretty(&serde_json::json!({
"run_id": run.run_id,

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

@ -3,7 +3,7 @@ use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use fabro_store::{RunStoreHandle, SlateStore};
use fabro_store::{SlateRunStore, SlateStore};
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
@ -18,7 +18,7 @@ pub(crate) fn build_store(storage_dir: &Path) -> Result<Arc<SlateStore>> {
)))
}
pub(crate) async fn open_run_reader(storage_dir: &Path, run_id: &RunId) -> Result<RunStoreHandle> {
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

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

@ -79,7 +79,7 @@ fn pr_view_reads_pull_request_from_store_without_pull_request_json() {
let store = build_store(&context.storage_dir);
let run_store = store.open_run(&run_id).await.unwrap();
append_workflow_event(
run_store.as_ref(),
&run_store,
&run_id,
&WorkflowRunEvent::PullRequestCreated {
pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(),

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

@ -4,7 +4,7 @@ use std::process::Output;
use std::sync::Arc;
use std::time::{Duration, Instant};
use fabro_store::{EventEnvelope, RunState, RunStoreHandle, SlateStore};
use fabro_store::{EventEnvelope, RunState, SlateRunStore, SlateStore};
use fabro_test::TestContext;
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
@ -480,7 +480,7 @@ fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
.block_on(future)
}
fn run_store(run_dir: &Path) -> RunStoreHandle {
fn run_store(run_dir: &Path) -> SlateRunStore {
let runs_dir = run_dir.parent().expect("run dir should have parent");
let storage_dir = runs_dir.parent().expect("runs dir should have parent");
let run_id: RunId = infer_run_id(run_dir).parse().expect("run id should parse");

4
lib/crates/fabro-cli/tests/it/scenario/mod.rs
View file

@ -7,7 +7,7 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use fabro_store::{RunState, RunStoreHandle, SlateStore};
use fabro_store::{RunState, SlateRunStore, SlateStore};
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
pub(super) fn fixture(name: &str) -> PathBuf {
@ -24,7 +24,7 @@ fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
.block_on(future)
}
fn run_store(run_dir: &Path) -> RunStoreHandle {
fn run_store(run_dir: &Path) -> SlateRunStore {
let runs_dir = run_dir.parent().expect("run dir should have parent");
let storage_dir = runs_dir.parent().expect("runs dir should have parent");
let run_id: RunId = std::fs::read_to_string(run_dir.join("id.txt"))

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

@ -702,7 +702,7 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
return;
}
};
let persisted = match Persisted::load_from_store(run_store.as_ref(), &run_dir).await {
let persisted = match Persisted::load_from_store(&run_store, &run_dir).await {
Ok(persisted) => persisted,
Err(e) => {
tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run");

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

@ -19,7 +19,6 @@ use fabro_types::{Outcome, StageUsage};
pub type NodeOutcomeRecord = Outcome<Option<StageUsage>>;
pub type StoreHandle = Arc<SlateStore>;
pub type RunStoreHandle = Arc<SlateRunStore>;
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct ListRunsQuery {

20
lib/crates/fabro-store/src/runtime.rs
View file

@ -60,14 +60,6 @@ impl RuntimeState {
.join(node_slug)
.join(format!("retry_{attempt}"))
}
pub fn ensure_runtime_dir(&self) -> std::io::Result<()> {
std::fs::create_dir_all(self.runtime_dir())
}
pub fn ensure_artifact_values_dir(&self) -> std::io::Result<()> {
std::fs::create_dir_all(self.artifact_values_dir())
}
}
#[cfg(test)]
@ -119,16 +111,4 @@ mod tests {
.join("retry_2")
);
}
#[test]
fn ensure_methods_create_directories() {
let dir = tempfile::tempdir().unwrap();
let state = RuntimeState::new(dir.path());
state.ensure_runtime_dir().unwrap();
state.ensure_artifact_values_dir().unwrap();
assert!(state.runtime_dir().is_dir());
assert!(state.artifact_values_dir().is_dir());
}
}

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

@ -102,6 +102,7 @@ pub(crate) async fn read_catalog_path(
#[cfg(test)]
pub(super) mod test_support {
use super::*;
use std::collections::{HashMap, HashSet};
pub(crate) async fn repair_catalog(
store: Arc<dyn ObjectStore>,

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

@ -14,7 +14,7 @@ use slatedb::config::{DbReaderOptions, Settings};
use tokio::sync::Mutex;
use crate::keys;
use crate::{CatalogRecord, ListRunsQuery, Result, RunStoreHandle, RunSummary, StoreError};
use crate::{CatalogRecord, ListRunsQuery, Result, RunSummary, StoreError};
use fabro_types::RunId;
pub use run_store::SlateRunStore;
use run_store::SlateRunStoreInner;
@ -175,7 +175,7 @@ impl SlateStore {
run_id: &RunId,
created_at: DateTime<Utc>,
run_dir: Option<&str>,
) -> Result<RunStoreHandle> {
) -> Result<SlateRunStore> {
let locator =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
if let Some(active) = self.get_active_run(run_id).await {
@ -196,7 +196,7 @@ impl SlateStore {
run_dir,
)
.await?;
return Ok(Arc::new(active));
return Ok(active);
}
let db_prefix = match locator {
@ -228,10 +228,10 @@ impl SlateStore {
run_dir,
)
.await?;
Ok(Arc::new(run_store))
Ok(run_store)
}
pub async fn open_run(&self, run_id: &RunId) -> Result<RunStoreHandle> {
pub async fn open_run(&self, run_id: &RunId) -> Result<SlateRunStore> {
let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
@ -240,10 +240,10 @@ impl SlateStore {
.open_run_store(&locator)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
Ok(Arc::new(run_store))
Ok(run_store)
}
pub async fn open_run_reader(&self, run_id: &RunId) -> Result<RunStoreHandle> {
pub async fn open_run_reader(&self, run_id: &RunId) -> Result<SlateRunStore> {
let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
@ -252,7 +252,7 @@ impl SlateStore {
.open_run_reader_store(&locator)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
Ok(Arc::new(run_store))
Ok(run_store)
}
pub async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>> {
@ -1146,10 +1146,6 @@ mod tests {
run.get_asset(&node, "src/lib.rs").await.unwrap(),
Some(Bytes::from_static(b"fn main() {}"))
);
assert_eq!(
run.list_assets(&node).await.unwrap(),
vec!["src/lib.rs".to_string()]
);
}
#[tokio::test]
@ -1497,7 +1493,7 @@ mod tests {
Some("diff --git a/src/lib.rs b/src/lib.rs\n")
);
assert_eq!(state.pull_request, Some(pull_request.clone()));
assert_eq!(state.list_node_ids(), vec!["code".to_string()]);
assert!(state.nodes.keys().any(|(id, _)| id == "code"));
let node_state = state
.node(&node)
.expect("node state should exist for code:2");
@ -1517,10 +1513,6 @@ mod tests {
.and_then(|v| v.as_str()),
Some("openai")
);
assert_eq!(
run.list_assets(&node).await.unwrap(),
vec!["src/lib.rs".to_string()]
);
}
#[tokio::test]
@ -1873,7 +1865,7 @@ mod tests {
.map(|checkpoint| checkpoint.current_node.as_str()),
Some("plan")
);
assert!(state.list_node_ids().is_empty());
assert!(state.nodes.is_empty());
}
#[tokio::test]

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

@ -313,19 +313,6 @@ impl SlateRunDb {
put_bytes(self.writer()?, key, value).await
}
async fn scan_prefix<P>(
&self,
prefix: P,
) -> std::result::Result<slatedb::DbIterator, slatedb::Error>
where
P: AsRef<[u8]> + Send,
{
match self {
Self::Writer(db) => db.scan_prefix(prefix).await,
Self::Reader(db) => db.scan_prefix(prefix).await,
}
}
async fn list_events_from(&self, start_seq: u32) -> Result<Vec<EventEnvelope>> {
match self {
Self::Writer(db) => list_events_from(db, start_seq).await,

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

@ -3,7 +3,7 @@ use std::sync::atomic::{AtomicI64, Ordering};
use anyhow::{Context, Result};
use chrono::{SecondsFormat, Utc};
use fabro_store::{EventPayload, RunStoreHandle, SlateRunStore};
use fabro_store::{EventPayload, SlateRunStore};
use fabro_types::RunId;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
@ -1510,7 +1510,7 @@ pub struct StoreProgressLogger {
impl StoreProgressLogger {
#[must_use]
pub fn new(run_store: RunStoreHandle) -> Self {
pub fn new(run_store: SlateRunStore) -> Self {
let (tx, mut rx) = mpsc::unbounded_channel();
tokio::spawn(async move {

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

@ -13,7 +13,7 @@ use crate::event::EventEmitter;
use crate::outcome::{
FailureCategory, FailureDetail, Outcome, OutcomeExt, StageStatus, StageUsage,
};
use crate::vars::expand_vars;
use crate::transforms::variable_expansion::expand_vars;
use fabro_graphviz::graph::{Graph, Node};
use super::{EngineServices, Handler};
@ -360,7 +360,7 @@ mod tests {
use super::*;
use crate::event::EventEmitter;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{NodeVisitRef, RunStoreHandle, SlateStore};
use fabro_store::{NodeVisitRef, SlateRunStore, SlateStore};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use std::sync::Arc;
@ -381,7 +381,7 @@ mod tests {
async fn make_services_with_run_store() -> (
EngineServices,
RunStoreHandle,
SlateRunStore,
crate::event::StoreProgressLogger,
) {
let store = test_store();

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

@ -169,7 +169,7 @@ mod tests {
use super::*;
use crate::outcome::StageStatus;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{NodeVisitRef, RunStoreHandle, SlateStore};
use fabro_store::{NodeVisitRef, SlateRunStore, SlateStore};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use std::sync::Arc;
@ -189,7 +189,7 @@ mod tests {
async fn make_services_with_run_store() -> (
EngineServices,
RunStoreHandle,
SlateRunStore,
crate::event::StoreProgressLogger,
) {
let store = test_store();

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

@ -20,7 +20,7 @@ use std::time::Duration;
use async_trait::async_trait;
use fabro_agent::Sandbox;
use fabro_store::RunStoreHandle;
use fabro_store::SlateRunStore;
#[cfg(test)]
use fabro_store::SlateStore;
#[cfg(test)]
@ -40,7 +40,7 @@ pub struct EngineServices {
pub registry: Arc<HandlerRegistry>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
/// Git state for the current run. Set via `set_git_state` at the start of
/// `run_via_core` and read by parallel/fan-in handlers.
pub(crate) git_state: std::sync::RwLock<Option<Arc<GitState>>>,

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

@ -175,7 +175,7 @@ impl Handler for PromptHandler {
mod tests {
use super::*;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{NodeVisitRef, RunStoreHandle, SlateStore};
use fabro_store::{NodeVisitRef, SlateRunStore, SlateStore};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use std::sync::Arc;
@ -196,7 +196,7 @@ mod tests {
async fn make_services_with_run_store() -> (
EngineServices,
RunStoreHandle,
SlateRunStore,
crate::event::StoreProgressLogger,
) {
let store = test_store();

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

@ -141,15 +141,3 @@ pub mod sandbox_git;
pub mod test_support;
#[doc(hidden)]
pub mod transforms;
// Re-export aliases (back-compat with `fabro_workflow::transform::*` imports)
#[doc(hidden)]
pub mod transform {
pub use crate::transforms::*;
}
#[doc(hidden)]
pub mod vars {
pub use crate::transforms::variable_expansion::*;
}
#[doc(hidden)]
pub use transforms::stylesheet;

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

@ -2,7 +2,7 @@ use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use fabro_store::RunStoreHandle;
use fabro_store::SlateRunStore;
use fabro_types::RunId;
use fabro_core::error::Result as CoreResult;
@ -26,7 +26,7 @@ type WfNodeResult = NodeResult<Option<StageUsage>>;
pub(crate) struct DiskLifecycle {
pub run_dir: PathBuf,
pub run_id: RunId,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub graph: Arc<GvGraph>,
pub run_options: Arc<RunOptions>,
pub emitter: Arc<EventEmitter>,
@ -38,7 +38,7 @@ pub(crate) struct DiskLifecycle {
impl RunLifecycle<WorkflowGraph> for DiskLifecycle {
async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
if let Err(err) = append_workflow_event(
self.run_store.as_ref(),
&self.run_store,
&self.run_id,
&WorkflowRunEvent::RunRunning { reason: None },
)

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

@ -2,7 +2,7 @@ use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use fabro_store::RunStoreHandle;
use fabro_store::SlateRunStore;
use fabro_types::RunId;
use fabro_core::error::{CoreError, Result as CoreResult};
@ -39,7 +39,7 @@ pub(crate) struct GitLifecycle {
pub emitter: Arc<EventEmitter>,
pub run_dir: PathBuf,
pub run_id: RunId,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub run_options: Arc<RunOptions>,
pub start_node_id: Option<String>,
// Cross-lifecycle data (shared with EventLifecycle)

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

@ -14,8 +14,8 @@ use std::sync::{Arc, Mutex};
use std::time::Instant;
use async_trait::async_trait;
use fabro_store::RunStoreHandle;
use fabro_store::RuntimeState;
use fabro_store::SlateRunStore;
use fabro_types::RunId;
use fabro_core::error::Result as CoreResult;
@ -85,7 +85,7 @@ impl WorkflowLifecycle {
sandbox: &Arc<dyn Sandbox>,
graph: Arc<GvGraph>,
run_dir: &PathBuf,
run_store: RunStoreHandle,
run_store: SlateRunStore,
run_options: &Arc<RunOptions>,
is_resume: bool,
on_node: crate::OnNodeCallback,

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

@ -187,7 +187,7 @@ async fn persist_created_run(
.map(|_| ())
.map_err(store_error)?;
append_workflow_event(
run_store.as_ref(),
&run_store,
&record.run_id,
&WorkflowRunEvent::RunSubmitted { reason: None },
)

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

@ -6,7 +6,7 @@ use anyhow::{Context, Result, bail};
use fabro_checkpoint::branch::BranchStore;
use fabro_checkpoint::git::Store as GitStore;
use fabro_store::{
ListRunsQuery, NodeVisitRef, RunStoreHandle as DurableRunStore, SlateStore as DurableStore,
ListRunsQuery, NodeVisitRef, SlateRunStore as DurableRunStore, SlateStore as DurableStore,
};
use fabro_types::RunId;
use git2::{Repository, Signature};
@ -434,7 +434,7 @@ mod tests {
let run_store = store.create_run(&run_id, created_at(), None).await.unwrap();
let run_record = sample_run_record(run_id, host_repo_path);
append_workflow_event(
run_store.as_ref(),
&run_store,
&run_id,
&WorkflowRunEvent::RunCreated {
run_id,

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

@ -41,7 +41,7 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
cleanup_resume_artifacts(run_dir);
append_workflow_event(
services.run_store.as_ref(),
&services.run_store,
&services.run_id,
&WorkflowRunEvent::RunSubmitted { reason: None },
)

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

@ -9,7 +9,7 @@ use fabro_config::{project as project_config, run as run_config, sandbox as sand
use fabro_interview::{AutoApproveInterviewer, Interviewer};
use fabro_model::{Catalog, FallbackTarget, Provider};
use fabro_sandbox::{SandboxProvider, SandboxSpec};
use fabro_store::{RunStoreHandle, SlateRunStore};
use fabro_store::SlateRunStore;
use fabro_types::{RunId, Settings};
use crate::context::Context;
@ -47,7 +47,7 @@ struct RunSession {
sandbox_env: SandboxEnvSpec,
devcontainer: Option<DevcontainerSpec>,
seed_context: Option<Context>,
run_store: RunStoreHandle,
run_store: SlateRunStore,
git: Option<GitCheckpointOptions>,
github_app: Option<fabro_github::GitHubAppCredentials>,
worktree_mode: Option<WorktreeMode>,
@ -65,7 +65,7 @@ pub struct StartServices {
pub cancel_token: Option<Arc<AtomicBool>>,
pub emitter: Arc<EventEmitter>,
pub interviewer: Arc<dyn Interviewer>,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
pub on_node: crate::OnNodeCallback,
pub registry_override: Option<Arc<HandlerRegistry>>,
@ -115,7 +115,7 @@ pub(super) async fn execute_persisted_run(
let error = FabroError::engine(err.to_string());
let _ = persist_detached_failure(
run_id,
run_store.as_ref(),
&run_store,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -125,7 +125,7 @@ pub(super) async fn execute_persisted_run(
return Err(error);
}
if let Err(err) = append_workflow_event(
run_store.as_ref(),
&run_store,
&run_id,
&WorkflowRunEvent::RunStarting {
reason: Some(StatusReason::SandboxInitializing),
@ -136,7 +136,7 @@ pub(super) async fn execute_persisted_run(
let error = FabroError::engine(err.to_string());
let _ = persist_detached_failure(
run_id,
run_store.as_ref(),
&run_store,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -149,12 +149,12 @@ pub(super) async fn execute_persisted_run(
let mut bootstrap_guard =
DetachedRunBootstrapGuard::arm(run_id, run_dir, run_store.clone(), cancel_token.clone());
let persisted = match Persisted::load_from_store(services.run_store.as_ref(), run_dir).await {
let persisted = match Persisted::load_from_store(&services.run_store, run_dir).await {
Ok(persisted) => persisted,
Err(err) => {
let _ = persist_detached_failure(
run_id,
run_store.as_ref(),
&run_store,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -171,7 +171,7 @@ pub(super) async fn execute_persisted_run(
Err(err) => {
let _ = persist_detached_failure(
run_id,
run_store.as_ref(),
&run_store,
run_dir,
"bootstrap",
StatusReason::BootstrapFailed,
@ -195,14 +195,8 @@ pub(super) async fn execute_persisted_run(
Ok(started)
}
Err(err) => {
persist_terminal_engine_failure(
run_id,
run_store.as_ref(),
run_dir,
&err,
run_start.elapsed(),
)
.await;
persist_terminal_engine_failure(run_id, &run_store, run_dir, &err, run_start.elapsed())
.await;
completion_guard.defuse();
Err(err)
}
@ -590,7 +584,7 @@ impl RunSession {
struct DetachedRunBootstrapGuard {
run_id: RunId,
run_store: RunStoreHandle,
run_store: SlateRunStore,
cancel_token: Option<Arc<AtomicBool>>,
active: bool,
}
@ -599,7 +593,7 @@ impl DetachedRunBootstrapGuard {
fn arm(
run_id: RunId,
_run_dir: &Path,
run_store: RunStoreHandle,
run_store: SlateRunStore,
cancel_token: Option<Arc<AtomicBool>>,
) -> Self {
Self {
@ -632,7 +626,7 @@ impl Drop for DetachedRunBootstrapGuard {
if let Ok(handle) = Handle::try_current() {
handle.spawn(async move {
let _ = append_workflow_event(
run_store.as_ref(),
&run_store,
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
error: FabroError::engine(format!("{reason:?}")),
@ -652,18 +646,14 @@ const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization
const POSTRUN_CANCELLED_MESSAGE: &str = "Run cancelled before post-run finalization completed.";
struct DetachedRunCompletionGuard {
run_store: RunStoreHandle,
run_store: SlateRunStore,
run_id: RunId,
cancel_token: Option<Arc<AtomicBool>>,
active: bool,
}
impl DetachedRunCompletionGuard {
fn arm(
run_id: RunId,
run_store: RunStoreHandle,
cancel_token: Option<Arc<AtomicBool>>,
) -> Self {
fn arm(run_id: RunId, run_store: SlateRunStore, cancel_token: Option<Arc<AtomicBool>>) -> Self {
Self {
run_store,
run_id,
@ -730,7 +720,7 @@ impl Drop for DetachedRunCompletionGuard {
if let Ok(handle) = Handle::try_current() {
handle.spawn(async move {
let _ = append_workflow_event(
run_store.as_ref(),
&run_store,
&run_id,
&WorkflowRunEvent::WorkflowRunFailed {
error: FabroError::engine(message.to_string()),
@ -1033,7 +1023,7 @@ mod tests {
node_visits: HashMap::new(),
};
append_workflow_event(
services.run_store.as_ref(),
&services.run_store,
&services.run_id,
&WorkflowRunEvent::CheckpointCompleted {
node_id: checkpoint.current_node.clone(),

12
lib/crates/fabro-workflow/src/pipeline/persist.rs
View file

@ -55,7 +55,7 @@ mod tests {
use chrono::Utc;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_store::{RunStoreHandle, SlateStore, StoreHandle};
use fabro_store::{SlateRunStore, SlateStore, StoreHandle};
use fabro_types::{Settings, fixtures};
use object_store::memory::InMemory;
use std::sync::Arc;
@ -142,7 +142,7 @@ mod tests {
run_dir: &Path,
record: &RunRecord,
source: Option<&str>,
) -> RunStoreHandle {
) -> SlateRunStore {
let store = memory_store();
let run_store = store
.create_run(
@ -153,7 +153,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run_store.as_ref(),
&run_store,
&record.run_id,
&WorkflowRunEvent::RunCreated {
run_id: record.run_id,
@ -240,7 +240,7 @@ mod tests {
.unwrap();
let run_store = seeded_store(&run_dir, &expected, Some(&source)).await;
let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap();
let loaded = load_from_store(&run_store, &run_dir).await.unwrap();
let loaded_record = loaded.run_record();
assert_eq!(loaded_record.run_id, expected.run_id);
@ -292,7 +292,7 @@ mod tests {
record.graph = graph;
let run_store = seeded_store(&run_dir, &record, None).await;
let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap();
let loaded = load_from_store(&run_store, &run_dir).await.unwrap();
assert!(loaded.source().is_empty());
}
@ -308,7 +308,7 @@ mod tests {
record.graph = graph.clone();
let run_store = seeded_store(&run_dir, &record, Some(&source)).await;
let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap();
let loaded = load_from_store(&run_store, &run_dir).await.unwrap();
assert_eq!(
serde_json::to_value(loaded.graph()).unwrap(),

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

@ -82,7 +82,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
});
run_retro_agent(
&options.sandbox,
options.run_store.as_ref(),
&options.run_store,
&options.run_dir,
client,
options.provider,
@ -215,7 +215,7 @@ mod tests {
async fn test_run_store(
run_dir: &std::path::Path,
checkpoint: &Checkpoint,
) -> fabro_store::RunStoreHandle {
) -> fabro_store::SlateRunStore {
let created_at = Utc::now();
let inner = test_store()
.create_run(
@ -238,7 +238,7 @@ mod tests {
labels: std::collections::HashMap::new(),
};
append_workflow_event(
run_store.as_ref(),
&run_store,
&test_run_id(),
&WorkflowRunEvent::RunCreated {
run_id: test_run_id(),
@ -258,7 +258,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run_store.as_ref(),
&run_store,
&test_run_id(),
&WorkflowRunEvent::CheckpointCompleted {
node_id: checkpoint.current_node.clone(),

16
lib/crates/fabro-workflow/src/pipeline/types.rs
View file

@ -11,7 +11,7 @@ use fabro_llm::Provider;
use fabro_mcp::config::McpServerSettings;
use fabro_model::FallbackTarget;
use fabro_sandbox::SandboxSpec;
use fabro_store::{RunStoreHandle, SlateRunStore};
use fabro_store::SlateRunStore;
use fabro_types::RunId;
use fabro_validate::Diagnostic;
@ -227,7 +227,7 @@ pub struct DevcontainerSpec {
pub struct InitOptions {
pub run_id: RunId,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub dry_run: bool,
pub emitter: Arc<EventEmitter>,
pub sandbox: SandboxSpec,
@ -251,7 +251,7 @@ pub struct Initialized {
pub graph: Graph,
pub source: String,
pub run_options: RunOptions,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub(crate) checkpoint: Option<Checkpoint>,
pub(crate) seed_context: Option<Context>,
pub emitter: Arc<EventEmitter>,
@ -272,7 +272,7 @@ pub struct Executed {
pub graph: Graph,
pub outcome: Result<Outcome, FabroError>,
pub run_options: RunOptions,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
@ -289,7 +289,7 @@ pub struct Retroed {
pub graph: Graph,
pub outcome: Result<Outcome, FabroError>,
pub run_options: RunOptions,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
@ -328,7 +328,7 @@ pub struct TransformOptions {
/// Options for the RETRO phase.
pub struct RetroOptions {
pub run_id: RunId,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub workflow_name: String,
pub goal: String,
pub run_dir: PathBuf,
@ -346,7 +346,7 @@ pub struct RetroOptions {
pub struct FinalizeOptions {
pub run_dir: PathBuf,
pub run_id: RunId,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub workflow_name: String,
pub hook_runner: Option<Arc<HookRunner>>,
pub preserve_sandbox: bool,
@ -356,7 +356,7 @@ pub struct FinalizeOptions {
/// Options for the PULL_REQUEST phase.
pub struct PullRequestOptions {
pub run_dir: PathBuf,
pub run_store: RunStoreHandle,
pub run_store: SlateRunStore,
pub pr_config: Option<PullRequestSettings>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
pub origin_url: Option<String>,

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

@ -370,7 +370,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run_store.as_ref(),
&run_store,
&fixtures::RUN_1,
&WorkflowRunEvent::RunCreated {
run_id: fixtures::RUN_1,
@ -390,7 +390,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run_store.as_ref(),
&run_store,
&fixtures::RUN_1,
&WorkflowRunEvent::RunSubmitted { reason: None },
)

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

@ -1,3 +1 @@
pub use fabro_types::status::{
InvalidTransition, ParseRunStatusError, RunStatus, RunStatusRecord, StatusReason,
};
pub use fabro_types::status::{RunStatus, RunStatusRecord, StatusReason};

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

@ -56,7 +56,7 @@ async fn initialized(
.expect("failed to create slate-backed test run store");
let run_store = inner_store;
append_workflow_event(
run_store.as_ref(),
&run_store,
&run_options.run_id,
&WorkflowRunEvent::RunCreated {
run_id: run_options.run_id,

4
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -43,9 +43,9 @@ use fabro_workflow::handler::{Handler, HandlerRegistry};
use fabro_workflow::outcome::{Outcome, OutcomeExt, StageStatus};
use fabro_workflow::records::{Checkpoint, CheckpointExt};
use fabro_workflow::run_options::{GitCheckpointOptions, RunOptions};
use fabro_workflow::stylesheet::{apply_stylesheet, parse_stylesheet};
use fabro_workflow::test_support::{WorkflowRunner, run_graph_with_hooks};
use fabro_workflow::transform::{
use fabro_workflow::transforms::stylesheet::{apply_stylesheet, parse_stylesheet};
use fabro_workflow::transforms::{
StylesheetApplicationTransform, Transform, VariableExpansionTransform,
};
use ulid::Ulid;