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

refactor: derive durable run paths from run IDs

Browse source 
Move durable run metadata and path derivation onto RunId, simplify the
Slate catalog/index format, and carry the storage-specific run directory
through workflow creation so detached and lookup flows stay aligned.

Also update affected CLI snapshots and test helpers to match the new
run discovery behavior.
This commit is contained in:
Bryan Helmkamp 2026-04-03 19:15:39 -07:00
parent f922a483c6
commit ebf5a0621d
No known key found for this signature in database
35 changed files with 274 additions and 566 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
lib/crates/fabro-checkpoint/src/metadata.rs
View file

@ -206,7 +206,6 @@ mod tests {
fn test_run_record(run_id: fabro_types::RunId) -> RunRecord {
RunRecord {
run_id,
created_at: Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).single().unwrap(),
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: None,

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

@ -46,7 +46,6 @@ pub(crate) async fn create_run(
settings,
cwd,
workflow_slug: None,
run_dir: None,
run_id,
base_branch: None,
host_repo_path: None,

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

@ -164,7 +164,7 @@ mod tests {
))
}
fn sample_run_record(run_id: RunId, created_at: DateTime<Utc>) -> RunRecord {
fn sample_run_record(run_id: RunId, _created_at: DateTime<Utc>) -> RunRecord {
let mut graph = Graph::new("night-sky");
graph.attrs.insert(
"goal".to_string(),
@ -172,7 +172,6 @@ mod tests {
);
RunRecord {
run_id,
created_at,
settings: Settings::default(),
graph,
workflow_slug: Some("night-sky".to_string()),
@ -284,7 +283,7 @@ mod tests {
let store = test_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run_id = test_run_id();
let run = store.create_run(&run_id, created_at, None).await.unwrap();
let run = store.create_run(&run_id).await.unwrap();
let run_record = sample_run_record(run_id, created_at);
let start_record = sample_start_record(run_id, created_at);
let status_record = sample_status();
@ -632,7 +631,7 @@ mod tests {
let store = test_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run_id = test_run_id();
let run = store.create_run(&run_id, created_at, None).await.unwrap();
let run = store.create_run(&run_id).await.unwrap();
let run_record = sample_run_record(run_id, created_at);
append_workflow_event(
&run,

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

@ -117,7 +117,7 @@ fn asset_cp_tree_preserves_structure() {
success: true
exit_code: 0
----- stdout -----
Copied 8 asset(s) to [TEMP_DIR]/asset-tree
Copied 6 asset(s) to [TEMP_DIR]/asset-tree
----- stderr -----
");
insta::assert_snapshot!(

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

@ -463,6 +463,13 @@ fn attach_json_errors_without_prompting_for_human_input() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.running",
"id": "[EVENT_ID]",
"properties": {},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.initialized",
"id": "[EVENT_ID]",
@ -483,13 +490,6 @@ fn attach_json_errors_without_prompting_for_human_input() {
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.running",
"id": "[EVENT_ID]",
"properties": {},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.started",
"id": "[EVENT_ID]",

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

@ -537,7 +537,6 @@ async fn start_run(
) -> Response {
let run_id = RunId::new();
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 Box::pin(operations::create(
state.store.as_ref(),
@ -549,7 +548,6 @@ async fn start_run(
settings,
cwd: std::env::current_dir().unwrap_or_else(|_| std::env::temp_dir()),
workflow_slug: None,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id),
host_repo_path: None,
base_branch: None,
@ -581,8 +579,8 @@ async fn start_run(
.into_response();
}
};
let persisted = created.persisted;
let created_at = persisted.run_record().created_at;
let created_at = run_id.created_at();
let run_dir = created.run_dir;
{
let mut runs = state.runs.lock().expect("runs lock poisoned");

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

@ -14,7 +14,7 @@ pub use fabro_types::StageId;
pub use run_state::{NodeState, RunProjection};
pub use runtime::RuntimeState;
pub use slate::{NodeAsset, SlateRunStore, SlateStore};
pub use types::{CatalogRecord, EventEnvelope, EventPayload, RunSummary};
pub use types::{EventEnvelope, EventPayload, RunSummary};
pub type StoreHandle = Arc<SlateStore>;

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

@ -6,7 +6,7 @@ use chrono::{DateTime, Utc};
use serde::de::DeserializeOwned;
use serde_json::Value;
use crate::{CatalogRecord, EventEnvelope, Result, RunSummary, StageId, StoreError};
use crate::{EventEnvelope, Result, RunSummary, StageId, StoreError};
use fabro_types::{
Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro,
RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StageUsage,
@ -86,7 +86,6 @@ impl RunProjection {
.collect::<HashMap<_, _>>();
self.run = Some(RunRecord {
run_id,
created_at: ts,
settings,
graph,
workflow_slug: optional_string(&properties, "workflow_slug"),
@ -281,7 +280,7 @@ impl RunProjection {
visits
}
pub(crate) fn build_summary(&self, catalog: &CatalogRecord) -> RunSummary {
pub(crate) fn build_summary(&self, run_id: &RunId) -> RunSummary {
let workflow_name = self.run.as_ref().map(|run| {
if run.graph.name.is_empty() {
"unnamed".to_string()
@ -294,7 +293,7 @@ impl RunProjection {
(!goal.is_empty()).then(|| goal.to_string())
});
RunSummary {
catalog: catalog.clone(),
run_id: *run_id,
workflow_name,
workflow_slug: self.run.as_ref().and_then(|run| run.workflow_slug.clone()),
goal,

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

@ -1,88 +1,76 @@
use std::sync::Arc;
use chrono::{DateTime, Utc};
use futures::TryStreamExt;
use object_store::ObjectStore;
use object_store::path::Path;
use serde::{Deserialize, Serialize};
use crate::{CatalogRecord, ListRunsQuery, Result};
use crate::{ListRunsQuery, Result};
use fabro_types::RunId;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct StartIndexRecord {
run_id: RunId,
}
pub(crate) async fn write_catalog(
store: Arc<dyn ObjectStore>,
base_prefix: &str,
run_id: &RunId,
created_at: DateTime<Utc>,
db_prefix: &str,
run_dir: Option<&str>,
) -> Result<CatalogRecord> {
let record = CatalogRecord {
run_id: *run_id,
created_at,
db_prefix: db_prefix.to_string(),
run_dir: run_dir.map(ToOwned::to_owned),
};
let bytes = serde_json::to_vec(&record)?;
) -> Result<()> {
store
.put(&by_id_path(base_prefix, run_id), bytes.clone().into())
.put(&by_id_path(base_prefix, run_id), bytes::Bytes::new().into())
.await?;
store
.put(
&by_start_path(base_prefix, created_at, run_id),
serde_json::to_vec(&StartIndexRecord { run_id: *run_id })?.into(),
&by_start_path(base_prefix, run_id),
bytes::Bytes::new().into(),
)
.await?;
Ok(record)
Ok(())
}
pub(crate) async fn read_locator(
store: Arc<dyn ObjectStore>,
base_prefix: &str,
run_id: &RunId,
) -> Result<Option<CatalogRecord>> {
read_catalog_path(store, by_id_path(base_prefix, run_id)).await
) -> Result<bool> {
match store.head(&by_id_path(base_prefix, run_id)).await {
Ok(_) => Ok(true),
Err(object_store::Error::NotFound { .. }) => Ok(false),
Err(err) => Err(err.into()),
}
}
pub(crate) async fn list_catalogs(
pub(crate) async fn list_run_ids(
store: Arc<dyn ObjectStore>,
base_prefix: &str,
query: &ListRunsQuery,
) -> Result<Vec<CatalogRecord>> {
) -> Result<Vec<RunId>> {
let prefix = Path::from(format!("{base_prefix}by-start"));
let metas = store.list(Some(&prefix)).try_collect::<Vec<_>>().await?;
let mut records = Vec::new();
let mut run_ids = Vec::new();
for meta in metas {
let Some(index) = read_start_index_path(store.clone(), meta.location).await? else {
let Some(run_id) = parse_run_id_from_path(&meta.location) else {
continue;
};
let Some(record) = read_locator(store.clone(), base_prefix, &index.run_id).await? else {
if !read_locator(store.clone(), base_prefix, &run_id).await? {
continue;
};
}
let created_at = run_id.created_at();
if let Some(start) = query.start {
if record.created_at < start {
if created_at < start {
continue;
}
}
if let Some(end) = query.end {
if record.created_at > end {
if created_at > end {
continue;
}
}
records.push(record);
run_ids.push(run_id);
}
Ok(records)
Ok(run_ids)
}
pub(crate) fn db_prefix(base_prefix: &str, created_at: DateTime<Utc>, run_id: &RunId) -> String {
pub(crate) fn db_prefix(base_prefix: &str, run_id: &RunId) -> String {
format!(
"{base_prefix}db/{}/{run_id}/",
created_at.format("%Y-%m-%d-%H-%M-%S-%3f")
run_id.created_at().format("%Y-%m-%d-%H-%M-%S-%3f")
)
}
@ -90,44 +78,23 @@ pub(crate) fn by_id_path(base_prefix: &str, run_id: &RunId) -> Path {
Path::from(format!("{base_prefix}by-id/{run_id}.json"))
}
pub(crate) fn by_start_path(base_prefix: &str, created_at: DateTime<Utc>, run_id: &RunId) -> Path {
pub(crate) fn by_start_path(base_prefix: &str, run_id: &RunId) -> Path {
Path::from(format!(
"{base_prefix}by-start/{}/{run_id}.json",
created_at.format("%Y-%m-%d-%H-%M")
run_id.created_at().format("%Y-%m-%d-%H-%M")
))
}
pub(crate) async fn read_catalog_path(
store: Arc<dyn ObjectStore>,
path: Path,
) -> Result<Option<CatalogRecord>> {
match store.get(&path).await {
Ok(result) => Ok(Some(serde_json::from_slice(&result.bytes().await?)?)),
Err(object_store::Error::NotFound { .. }) => Ok(None),
Err(err) => Err(err.into()),
}
}
#[cfg(test)]
fn by_start_record(run_id: &RunId) -> StartIndexRecord {
StartIndexRecord { run_id: *run_id }
}
async fn read_start_index_path(
store: Arc<dyn ObjectStore>,
path: Path,
) -> Result<Option<StartIndexRecord>> {
match store.get(&path).await {
Ok(result) => Ok(Some(serde_json::from_slice(&result.bytes().await?)?)),
Err(object_store::Error::NotFound { .. }) => Ok(None),
Err(err) => Err(err.into()),
}
pub(crate) fn parse_run_id_from_path(path: &Path) -> Option<RunId> {
let filename = path.filename()?;
let run_id = filename.strip_suffix(".json").unwrap_or(filename);
run_id.parse().ok()
}
#[cfg(test)]
pub(super) mod test_support {
use super::*;
use std::collections::{HashMap, HashSet};
use std::collections::HashSet;
pub(crate) async fn repair_catalog(
store: Arc<dyn ObjectStore>,
@ -140,22 +107,17 @@ pub(super) mod test_support {
.list(Some(&by_id_prefix))
.try_collect::<Vec<_>>()
.await?;
let mut canonical = HashMap::new();
let mut canonical = HashSet::new();
for meta in by_id_metas {
if let Some(record) = read_catalog_path(store.clone(), meta.location).await? {
canonical.insert(record.run_id, record);
if let Some(run_id) = parse_run_id_from_path(&meta.location) {
canonical.insert(run_id);
}
}
for record in canonical.values() {
let path = by_start_path(base_prefix, record.created_at, &record.run_id);
for run_id in &canonical {
let path = by_start_path(base_prefix, run_id);
if !object_exists(store.clone(), &path).await? {
store
.put(
&path,
serde_json::to_vec(&by_start_record(&record.run_id))?.into(),
)
.await?;
store.put(&path, bytes::Bytes::new().into()).await?;
}
}
@ -166,29 +128,28 @@ pub(super) mod test_support {
let mut seen = HashSet::new();
for meta in by_start_metas {
let location = meta.location.clone();
let Some(index) = read_start_index_path(store.clone(), location.clone()).await? else {
let Some(run_id) = parse_run_id_from_path(&location) else {
delete_if_exists(store.clone(), &location).await?;
continue;
};
let expected = canonical.get(&index.run_id).map(|canonical_record| {
by_start_path(base_prefix, canonical_record.created_at, &index.run_id)
});
match expected {
Some(expected) if expected == location => {
seen.insert(index.run_id);
}
_ => {
delete_if_exists(store.clone(), &location).await?;
}
if !canonical.contains(&run_id) {
delete_if_exists(store.clone(), &location).await?;
continue;
}
let expected = by_start_path(base_prefix, &run_id);
if expected == location {
seen.insert(run_id);
} else {
delete_if_exists(store.clone(), &location).await?;
}
}
for record in canonical.values() {
if !seen.contains(&record.run_id) {
for run_id in &canonical {
if !seen.contains(run_id) {
store
.put(
&by_start_path(base_prefix, record.created_at, &record.run_id),
serde_json::to_vec(&by_start_record(&record.run_id))?.into(),
&by_start_path(base_prefix, run_id),
bytes::Bytes::new().into(),
)
.await?;
}

312
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, RunSummary, StoreError};
use crate::{ListRunsQuery, Result, RunSummary, StoreError};
use fabro_types::RunId;
use run_store::SlateRunStoreInner;
pub use run_store::{NodeAsset, SlateRunStore};
@ -102,7 +102,7 @@ impl SlateStore {
self.active_runs
.lock()
.await
.insert(run_store.record().run_id, run_store.downgrade());
.insert(run_store.run_id(), run_store.downgrade());
}
async fn remove_active_run(&self, run_id: &RunId) -> Option<SlateRunStore> {
@ -112,27 +112,22 @@ impl SlateStore {
async fn open_existing_run(
&self,
record: &CatalogRecord,
run_id: RunId,
mode: RunOpenMode,
) -> Result<Option<SlateRunStore>> {
if matches!(mode, RunOpenMode::Writer) {
if let Some(active) = self.get_active_run(&record.run_id).await {
if active.record() == *record {
return Ok(Some(active));
}
return Err(StoreError::Other(format!(
"active run cache mismatch for run_id {:?}",
record.run_id
)));
if let Some(active) = self.get_active_run(&run_id).await {
return Ok(Some(active));
}
}
if !self.db_prefix_has_objects(&record.db_prefix).await? {
let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id);
if !self.db_prefix_has_objects(&db_prefix).await? {
return Ok(None);
}
match mode {
RunOpenMode::Writer => {
let db = self.open_db(&record.db_prefix).await?;
let has_init = match SlateRunStore::validate_init(&db, record).await {
let db = self.open_db(&db_prefix).await?;
let has_init = match SlateRunStore::validate_init(&db, &run_id).await {
Ok(has_init) => has_init,
Err(err) => {
let _ = db.close().await;
@ -143,13 +138,13 @@ impl SlateStore {
let _ = db.close().await;
return Ok(None);
}
let run_store = SlateRunStore::open_writer(record.clone(), db).await?;
let run_store = SlateRunStore::open_writer(run_id, db).await?;
self.cache_active_run(&run_store).await;
Ok(Some(run_store))
}
RunOpenMode::Reader => {
let reader = self.open_reader(&record.db_prefix).await?;
let has_init = match SlateRunStore::validate_init(&reader, record).await {
let reader = self.open_reader(&db_prefix).await?;
let has_init = match SlateRunStore::validate_init(&reader, &run_id).await {
Ok(has_init) => has_init,
Err(err) => {
let _ = reader.close().await;
@ -160,9 +155,7 @@ impl SlateStore {
let _ = reader.close().await;
return Ok(None);
}
SlateRunStore::open_reader(record.clone(), reader)
.await
.map(Some)
SlateRunStore::open_reader(run_id, reader).await.map(Some)
}
}
}
@ -182,86 +175,51 @@ impl SlateStore {
}
impl SlateStore {
pub async fn create_run(
&self,
run_id: &RunId,
created_at: DateTime<Utc>,
run_dir: Option<&str>,
) -> Result<SlateRunStore> {
pub async fn create_run(&self, run_id: &RunId) -> 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 {
if active.created_at() != created_at
|| locator
.as_ref()
.is_some_and(|existing| existing.created_at != created_at)
{
if locator {
return Err(StoreError::RunAlreadyExists(run_id.to_string()));
}
let record = active.record();
catalog::write_catalog(
self.object_store.clone(),
&self.base_prefix,
run_id,
created_at,
&record.db_prefix,
run_dir,
)
.await?;
catalog::write_catalog(self.object_store.clone(), &self.base_prefix, run_id).await?;
return Ok(active);
}
let db_prefix = match locator {
Some(existing) if existing.created_at != created_at => {
return Err(StoreError::RunAlreadyExists(run_id.to_string()));
}
Some(existing) => existing.db_prefix,
None => catalog::db_prefix(&self.base_prefix, created_at, run_id),
};
let record = CatalogRecord {
run_id: *run_id,
created_at,
db_prefix: db_prefix.clone(),
run_dir: run_dir.map(ToOwned::to_owned),
};
let db_prefix = catalog::db_prefix(&self.base_prefix, run_id);
let db = self.open_db(&db_prefix).await?;
SlateRunStore::validate_init(&db, &record).await?;
db.put(keys::init(), serde_json::to_vec(&record)?).await?;
let run_store = SlateRunStore::open_writer(record.clone(), db).await?;
SlateRunStore::validate_init(&db, run_id).await?;
db.put(keys::init(), serde_json::to_vec(run_id)?).await?;
let run_store = SlateRunStore::open_writer(*run_id, db).await?;
self.cache_active_run(&run_store).await;
catalog::write_catalog(
self.object_store.clone(),
&self.base_prefix,
run_id,
created_at,
&db_prefix,
run_dir,
)
.await?;
catalog::write_catalog(self.object_store.clone(), &self.base_prefix, run_id).await?;
Ok(run_store)
}
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()))?;
let locator =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
if !locator {
return Err(StoreError::RunNotFound(run_id.to_string()));
}
let run_store = self
.open_existing_run(&locator, RunOpenMode::Writer)
.open_existing_run(*run_id, RunOpenMode::Writer)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
Ok(run_store)
}
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()))?;
let locator =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
if !locator {
return Err(StoreError::RunNotFound(run_id.to_string()));
}
let run_store = self
.open_existing_run(&locator, RunOpenMode::Reader)
.open_existing_run(*run_id, RunOpenMode::Reader)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
Ok(run_store)
@ -277,51 +235,41 @@ impl SlateStore {
end: Option<DateTime<Utc>>,
) -> Result<Vec<RunSummary>> {
let query = ListRunsQuery { start, end };
let catalogs =
catalog::list_catalogs(self.object_store.clone(), &self.base_prefix, &query).await?;
let run_ids =
catalog::list_run_ids(self.object_store.clone(), &self.base_prefix, &query).await?;
let mut summaries = Vec::new();
for record in catalogs {
if let Some(active) = self.get_active_run(&record.run_id).await {
if active.record() != record {
return Err(StoreError::Other(format!(
"active run cache mismatch for run_id {:?}",
record.run_id
)));
}
for run_id in run_ids {
if let Some(active) = self.get_active_run(&run_id).await {
let snapshot = active.snapshot().await?;
summaries.push(SlateRunStore::build_summary(snapshot.as_ref(), &record).await?);
summaries.push(SlateRunStore::build_summary(snapshot.as_ref(), &run_id).await?);
continue;
}
if !self.db_prefix_has_objects(&record.db_prefix).await? {
let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id);
if !self.db_prefix_has_objects(&db_prefix).await? {
continue;
}
let reader = self.open_reader(&record.db_prefix).await?;
if !SlateRunStore::validate_init(&reader, &record).await? {
let reader = self.open_reader(&db_prefix).await?;
if !SlateRunStore::validate_init(&reader, &run_id).await? {
let _ = reader.close().await;
continue;
}
let summary = SlateRunStore::build_summary(&reader, &record).await;
let summary = SlateRunStore::build_summary(&reader, &run_id).await;
let _ = reader.close().await;
let summary = summary?;
summaries.push(summary);
}
summaries.sort_by(|a, b| b.catalog.created_at.cmp(&a.catalog.created_at));
summaries.sort_by(|a, b| b.run_id.created_at().cmp(&a.run_id.created_at()));
Ok(summaries)
}
pub async fn delete_run(&self, run_id: &RunId) -> Result<()> {
let active = self.remove_active_run(run_id).await;
let active_record = active.as_ref().map(SlateRunStore::record);
if let Some(active) = &active {
active.close().await?;
}
if let Some(record) =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id)
.await?
.or(active_record)
{
return self.delete_run_record(&record).await;
if catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? {
return self.delete_run_record(run_id).await;
}
self.repair_delete_run(run_id).await
@ -329,16 +277,17 @@ impl SlateStore {
}
impl SlateStore {
async fn delete_run_record(&self, record: &CatalogRecord) -> Result<()> {
async fn delete_run_record(&self, run_id: &RunId) -> Result<()> {
delete_path(
self.object_store.clone(),
&catalog::by_start_path(&self.base_prefix, record.created_at, &record.run_id),
&catalog::by_start_path(&self.base_prefix, run_id),
)
.await?;
self.delete_db_prefix(&record.db_prefix).await?;
self.delete_db_prefix(&catalog::db_prefix(&self.base_prefix, run_id))
.await?;
delete_path(
self.object_store.clone(),
&catalog::by_id_path(&self.base_prefix, &record.run_id),
&catalog::by_id_path(&self.base_prefix, run_id),
)
.await?;
Ok(())
@ -421,6 +370,14 @@ mod tests {
use crate::{EventPayload, StageId};
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
struct CatalogRecord {
run_id: RunId,
created_at: DateTime<Utc>,
db_prefix: String,
run_dir: Option<String>,
}
fn dt(rfc3339: &str) -> DateTime<Utc> {
DateTime::parse_from_rfc3339(rfc3339)
.unwrap()
@ -455,7 +412,6 @@ mod tests {
);
RunRecord {
run_id: test_run_id(run_id),
created_at,
settings: Settings::default(),
graph,
workflow_slug: Some("night-sky".to_string()),
@ -621,7 +577,7 @@ mod tests {
.await
.unwrap();
if include_init {
db.put(keys::init(), serde_json::to_vec(record).unwrap())
db.put(keys::init(), serde_json::to_vec(&record.run_id).unwrap())
.await
.unwrap();
}
@ -632,10 +588,7 @@ mod tests {
async fn create_open_list_and_delete_full_lifecycle() {
let (object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_record = sample_run_record("run-1", created_at);
run.append_event(&event_payload(
@ -684,7 +637,7 @@ mod tests {
.unwrap();
let by_id = catalog::by_id_path("runs/", &test_run_id("run-1"));
let by_start = catalog::by_start_path("runs/", created_at, &test_run_id("run-1"));
let by_start = catalog::by_start_path("runs/", &test_run_id("run-1"));
assert!(object_exists(object_store.clone(), &by_id).await);
assert!(object_exists(object_store.clone(), &by_start).await);
assert_eq!(
@ -696,7 +649,7 @@ mod tests {
let summary = store.list_runs().await.unwrap();
assert_eq!(summary.len(), 1);
assert_eq!(summary[0].catalog.run_id, test_run_id("run-1"));
assert_eq!(summary[0].run_id, test_run_id("run-1"));
assert_eq!(summary[0].workflow_name, Some("night-sky".to_string()));
assert_eq!(summary[0].goal, Some("map the constellations".to_string()));
assert_eq!(summary[0].status, Some(RunStatus::Succeeded));
@ -720,7 +673,7 @@ mod tests {
let record = CatalogRecord {
run_id: test_run_id("run-1"),
created_at,
db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")),
db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")),
run_dir: None,
};
@ -765,7 +718,7 @@ mod tests {
assert!(
object_exists(
object_store,
&catalog::by_start_path("runs/", created_at, &test_run_id("run-1"))
&catalog::by_start_path("runs/", &test_run_id("run-1"))
)
.await
);
@ -778,7 +731,7 @@ mod tests {
let record = CatalogRecord {
run_id: test_run_id("run-1"),
created_at,
db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")),
db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")),
run_dir: None,
};
@ -815,7 +768,7 @@ mod tests {
.unwrap();
object_store
.put(
&catalog::by_start_path("runs/", created_at, &test_run_id("run-1")),
&catalog::by_start_path("runs/", &test_run_id("run-1")),
serde_json::json!({
"run_id": test_run_id("run-1").to_string(),
})
@ -827,17 +780,14 @@ mod tests {
let listed = store.list_runs().await.unwrap();
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].catalog, record);
assert_eq!(listed[0].run_id, record.run_id);
}
#[tokio::test]
async fn reopen_recovers_event_sequences() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_record = sample_run_record("run-1", created_at);
run.append_event(&event_payload(
"run-1",
@ -897,7 +847,7 @@ mod tests {
let record = CatalogRecord {
run_id: test_run_id("run-1"),
created_at,
db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")),
db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")),
run_dir: None,
};
@ -912,7 +862,7 @@ mod tests {
.unwrap();
object_store
.put(
&catalog::by_start_path("runs/", created_at, &test_run_id("run-1")),
&catalog::by_start_path("runs/", &test_run_id("run-1")),
serde_json::to_vec(&record).unwrap().into(),
)
.await
@ -926,22 +876,10 @@ mod tests {
async fn create_run_allows_idempotent_retry_and_rejects_conflict() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
store.create_run(&test_run_id("run-1")).await.unwrap();
store.create_run(&test_run_id("run-1")).await.unwrap();
let conflict = store
.create_run(
&test_run_id("run-1"),
created_at + chrono::Duration::seconds(1),
None,
)
.await;
let conflict = store.create_run(&test_run_id("run-1")).await;
assert!(matches!(conflict, Err(StoreError::RunAlreadyExists(_))));
}
@ -949,10 +887,7 @@ mod tests {
async fn list_runs_and_open_run_reuse_active_handle_without_fencing() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_record = sample_run_record("run-1", created_at);
run.append_event(&event_payload(
"run-1",
@ -1004,10 +939,7 @@ mod tests {
async fn watch_events_from_polls_new_events() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let mut stream = run.watch_events_from(1).unwrap();
run.append_event(&event_payload(
@ -1038,7 +970,7 @@ mod tests {
let record = CatalogRecord {
run_id: test_run_id("run-1"),
created_at,
db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")),
db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")),
run_dir: None,
};
let db = seed_db(object_store.clone(), &record, true).await;
@ -1066,7 +998,7 @@ mod tests {
db.close().await.unwrap();
object_store
.put(
&catalog::by_start_path("runs/", created_at, &test_run_id("run-1")),
&catalog::by_start_path("runs/", &test_run_id("run-1")),
serde_json::to_vec(&record).unwrap().into(),
)
.await
@ -1081,10 +1013,7 @@ mod tests {
async fn delete_run_closes_active_handles() {
let (object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
.await
.unwrap();
@ -1108,22 +1037,18 @@ mod tests {
let (object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let wrong_time = dt("2026-03-27T11:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
.await
.unwrap();
let locator = catalog::read_locator(object_store.clone(), "runs/", &test_run_id("run-1"))
.await
.unwrap()
.unwrap();
object_store
.put(
&catalog::by_start_path("runs/", wrong_time, &test_run_id("run-1")),
serde_json::to_vec(&locator).unwrap().into(),
&catalog::by_start_path("runs/", &test_run_id("run-1")),
bytes::Bytes::new().into(),
)
.await
.unwrap();
@ -1138,38 +1063,32 @@ mod tests {
async fn create_run_uses_distinct_db_prefix_for_same_minute_orphan() {
let (object_store, store) = make_store();
let old_created_at = dt("2026-03-27T12:00:00Z");
let new_created_at = dt("2026-03-27T12:00:30Z");
let orphan = CatalogRecord {
run_id: test_run_id("run-1"),
created_at: old_created_at,
db_prefix: catalog::db_prefix("runs/", old_created_at, &test_run_id("run-1")),
db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")),
run_dir: None,
};
let new_prefix = catalog::db_prefix("runs/", new_created_at, &test_run_id("run-1"));
assert_ne!(orphan.db_prefix, new_prefix);
let new_prefix = catalog::db_prefix("runs/", &test_run_id("run-1"));
assert_eq!(orphan.db_prefix, new_prefix);
let db = seed_db(object_store.clone(), &orphan, true).await;
db.close().await.unwrap();
let run = store
.create_run(&test_run_id("run-1"), new_created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
assert!(run.state().await.unwrap().graph_source.is_none());
let locator = catalog::read_locator(object_store, "runs/", &test_run_id("run-1"))
.await
.unwrap()
.unwrap();
assert_eq!(locator.created_at, new_created_at);
assert_eq!(locator.db_prefix, new_prefix);
assert!(locator);
}
#[tokio::test]
async fn create_run_rejects_mismatched_init_for_existing_prefix() {
let (object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let db_prefix = catalog::db_prefix("runs/", created_at, &test_run_id("run-1"));
let db_prefix = catalog::db_prefix("runs/", &test_run_id("run-1"));
let db = slatedb::Db::builder(db_prefix.clone(), object_store)
.with_settings(SlateSettings {
flush_interval: Some(Duration::from_millis(1)),
@ -1189,10 +1108,7 @@ mod tests {
.unwrap();
db.close().await.unwrap();
let Err(err) = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
else {
let Err(err) = store.create_run(&test_run_id("run-1")).await else {
panic!("expected create_run to reject mismatched _init.json");
};
assert!(matches!(
@ -1205,10 +1121,7 @@ mod tests {
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
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let node = StageId::new("code", 2);
run.append_event(&event_payload(
"run-1",
@ -1236,10 +1149,7 @@ mod tests {
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
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
.await
.unwrap();
@ -1280,10 +1190,7 @@ mod tests {
async fn create_run_state_and_node_storage_round_trip() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_record = sample_run_record("run-1", created_at);
let start_record = sample_start_record("run-1", created_at);
@ -1527,7 +1434,6 @@ mod tests {
let state = run.state().await.unwrap();
let stored_run = state.run.as_ref().unwrap();
assert_eq!(stored_run.run_id, run_record.run_id);
assert_eq!(stored_run.created_at, run_record.created_at);
assert_eq!(stored_run.workflow_slug, run_record.workflow_slug);
assert_eq!(stored_run.graph.name, run_record.graph.name);
assert_eq!(state.graph_source.as_deref(), Some("digraph night_sky {}"));
@ -1596,10 +1502,7 @@ mod tests {
async fn state_projects_event_stream() {
let (_object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let run_record = sample_run_record("run-1", created_at);
let retro = sample_retro("run-1");
@ -1821,10 +1724,7 @@ mod tests {
#[tokio::test]
async fn state_rewind_keeps_active_projection_only() {
let (_object_store, store) = make_store();
let run = store
.create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
run.append_event(&event_payload(
"run-1",
@ -1943,10 +1843,7 @@ mod tests {
#[tokio::test]
async fn append_event_validates_payload_shape_and_run_id() {
let (_object_store, store) = make_store();
let run = store
.create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let invalid_missing_err = serde_json::from_value::<EventPayload>(serde_json::json!({
"run_id": "run-1"
@ -1972,10 +1869,7 @@ mod tests {
#[tokio::test]
async fn state_retains_checkpoint_history_by_event_sequence() {
let (_object_store, store) = make_store();
let run = store
.create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None)
.await
.unwrap();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let checkpoint = sample_checkpoint();
let seq = run
.append_event(&event_payload(
@ -2016,10 +1910,7 @@ mod tests {
let early = dt("2026-03-27T10:00:00Z");
let late = dt("2026-03-27T12:00:00Z");
let early_run = store
.create_run(&test_run_id("run-early"), early, None)
.await
.unwrap();
let early_run = store.create_run(&test_run_id("run-early")).await.unwrap();
let early_record = sample_run_record("run-early", early);
early_run
.append_event(&event_payload(
@ -2040,10 +1931,7 @@ mod tests {
.await
.unwrap();
let late_run = store
.create_run(&test_run_id("run-late"), late, None)
.await
.unwrap();
let late_run = store.create_run(&test_run_id("run-late")).await.unwrap();
let late_record = sample_run_record("run-late", late);
late_run
.append_event(&event_payload(
@ -2095,7 +1983,7 @@ mod tests {
let all = store.list_runs().await.unwrap();
assert_eq!(all.len(), 2);
assert_eq!(all[0].catalog.run_id, test_run_id("run-late"));
assert_eq!(all[0].run_id, test_run_id("run-late"));
assert_eq!(all[0].workflow_name, Some("night-sky".to_string()));
assert_eq!(all[0].goal, Some("map the constellations".to_string()));
assert_eq!(
@ -2115,6 +2003,6 @@ mod tests {
.await
.unwrap();
assert_eq!(filtered.len(), 1);
assert_eq!(filtered[0].catalog.run_id, test_run_id("run-late"));
assert_eq!(filtered[0].run_id, test_run_id("run-late"));
}
}

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

@ -14,10 +14,8 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::run_state::EventProjectionCache;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId,
StoreError,
};
use crate::{EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId, StoreError};
use fabro_types::RunId;
#[derive(Clone)]
pub struct SlateRunStore {
inner: Arc<SlateRunStoreInner>,
@ -32,13 +30,13 @@ pub struct NodeAsset {
impl std::fmt::Debug for SlateRunStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SlateRunStore")
.field("record", &self.inner.record)
.field("run_id", &self.inner.run_id)
.finish_non_exhaustive()
}
}
pub(crate) struct SlateRunStoreInner {
record: CatalogRecord,
run_id: RunId,
db: SlateRunDb,
event_seq: AtomicU32,
close_lock: Mutex<()>,
@ -51,11 +49,11 @@ enum SlateRunDb {
}
impl SlateRunStore {
pub(crate) async fn open_writer(record: CatalogRecord, db: slatedb::Db) -> Result<Self> {
pub(crate) async fn open_writer(run_id: RunId, db: slatedb::Db) -> Result<Self> {
let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?;
Ok(Self {
inner: Arc::new(SlateRunStoreInner {
record,
run_id,
db: SlateRunDb::Writer(db),
event_seq: AtomicU32::new(event_seq),
close_lock: Mutex::new(()),
@ -64,11 +62,11 @@ impl SlateRunStore {
})
}
pub(crate) async fn open_reader(record: CatalogRecord, db: DbReader) -> Result<Self> {
pub(crate) async fn open_reader(run_id: RunId, db: DbReader) -> Result<Self> {
let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?;
Ok(Self {
inner: Arc::new(SlateRunStoreInner {
record,
run_id,
db: SlateRunDb::Reader(Box::new(db)),
event_seq: AtomicU32::new(event_seq),
close_lock: Mutex::new(()),
@ -85,12 +83,12 @@ impl SlateRunStore {
Arc::downgrade(&self.inner)
}
pub(crate) fn record(&self) -> CatalogRecord {
self.inner.record.clone()
pub fn run_id(&self) -> RunId {
self.inner.run_id
}
pub fn created_at(&self) -> DateTime<Utc> {
self.inner.record.created_at
self.inner.run_id.created_at()
}
pub(crate) async fn close(&self) -> Result<()> {
@ -109,26 +107,26 @@ impl SlateRunStore {
}
}
pub(crate) async fn validate_init<R>(db: &R, expected: &CatalogRecord) -> Result<bool>
pub(crate) async fn validate_init<R>(db: &R, expected: &RunId) -> Result<bool>
where
R: DbRead + Sync,
{
match get_json::<R, CatalogRecord>(db, keys::init()).await? {
match get_json::<R, RunId>(db, keys::init()).await? {
Some(existing) if existing == *expected => Ok(true),
Some(existing) => Err(StoreError::Other(format!(
"existing _init.json {existing:?} does not match requested catalog {expected:?}"
"existing _init.json {existing:?} does not match requested run id {expected:?}"
))),
None => Ok(false),
}
}
pub(crate) async fn build_summary<R>(db: &R, catalog: &CatalogRecord) -> Result<RunSummary>
pub(crate) async fn build_summary<R>(db: &R, run_id: &RunId) -> Result<RunSummary>
where
R: DbRead + Sync,
{
let events = list_events_from(db, 1).await?;
let state = RunProjection::apply_events(&events)?;
Ok(state.build_summary(catalog))
Ok(state.build_summary(run_id))
}
async fn projected_state(&self) -> Result<RunProjection> {
@ -148,11 +146,11 @@ impl SlateRunStore {
impl SlateRunStore {
pub async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
if payload.run_id() != self.inner.record.run_id.to_string() {
if payload.run_id() != self.inner.run_id.to_string() {
return Err(StoreError::InvalidEvent(format!(
"payload run_id {:?} does not match store run_id {:?}",
payload.run_id(),
self.inner.record.run_id
self.inner.run_id
)));
}
let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst);

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

@ -7,18 +7,9 @@ use serde::{Deserialize, Deserializer, Serialize};
use crate::{Result, StoreError};
use fabro_types::{RunId, RunStatus, StatusReason};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogRecord {
pub run_id: RunId,
pub created_at: DateTime<Utc>,
pub db_prefix: String,
pub run_dir: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSummary {
#[serde(flatten)]
pub catalog: CatalogRecord,
pub run_id: RunId,
pub workflow_name: Option<String>,
pub workflow_slug: Option<String>,
pub goal: Option<String>,

2
lib/crates/fabro-types/src/run.rs
View file

@ -1,7 +1,6 @@
use std::collections::HashMap;
use std::path::PathBuf;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::graph::Graph;
@ -11,7 +10,6 @@ use crate::settings::Settings;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunRecord {
pub run_id: RunId,
pub created_at: DateTime<Utc>,
pub settings: Settings,
pub graph: Graph,
#[serde(default, skip_serializing_if = "Option::is_none")]

31
lib/crates/fabro-types/src/run_id.rs
View file

@ -1,6 +1,7 @@
use std::fmt;
use std::str::FromStr;
use chrono::{DateTime, Utc};
use serde::de::Error as _;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use ulid::Ulid;
@ -12,6 +13,15 @@ impl RunId {
pub fn new() -> Self {
Self(Ulid::new())
}
pub fn created_at(&self) -> DateTime<Utc> {
self.0.datetime().into()
}
#[cfg(test)]
pub fn from_datetime(dt: DateTime<Utc>) -> Self {
Self(Ulid::from_datetime(dt.into()))
}
}
impl Default for RunId {
@ -157,6 +167,8 @@ pub mod fixtures {
#[cfg(test)]
mod tests {
use chrono::{DateTime, Utc};
use super::{RunId, fixtures};
#[test]
@ -172,4 +184,23 @@ mod tests {
assert_eq!(value, fixtures::RUN_42);
}
#[test]
fn created_at_comes_from_ulid_timestamp() {
let expected = DateTime::parse_from_rfc3339("2026-03-27T12:00:00Z")
.unwrap()
.with_timezone(&Utc);
let run_id = RunId::from_datetime(expected);
assert_eq!(run_id.created_at(), expected);
}
#[test]
fn from_datetime_round_trips_timestamp() {
let expected = DateTime::parse_from_rfc3339("2026-03-27T12:00:00Z")
.unwrap()
.with_timezone(&Utc);
assert_eq!(RunId::from_datetime(expected).created_at(), expected);
}
}

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

@ -1899,10 +1899,7 @@ mod tests {
"",
std::time::Duration::from_millis(1),
);
let run_store = store
.create_run(&fixtures::RUN_7, Utc::now(), None)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_7).await.unwrap();
let envelope = canonicalize_event(
&fixtures::RUN_7,
&WorkflowRunEvent::RunNotice {

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

@ -511,10 +511,7 @@ mod tests {
use fabro_store::EventPayload;
let store = test_store();
let run = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let run = store.create_run(&fixtures::RUN_1).await.unwrap();
let run_id_str = fixtures::RUN_1.to_string();
let event = |event_name: &str, props: serde_json::Value| -> EventPayload {

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

@ -509,10 +509,7 @@ mod tests {
crate::event::StoreProgressLogger,
) {
let store = test_store();
let run_store = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let services = EngineServices {
emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)),
run_store: run_store.clone(),

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

@ -227,10 +227,7 @@ mod tests {
crate::event::StoreProgressLogger,
) {
let store = test_store();
let run_store = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let services = EngineServices {
emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)),
run_store: run_store.clone(),

7
lib/crates/fabro-workflow/src/handler/manager_loop.rs
View file

@ -15,7 +15,6 @@ use crate::pipeline::types::Initialized;
use crate::run_dir::visit_from_context;
use crate::run_options::RunOptions;
use async_trait::async_trait;
use chrono::Utc;
use fabro_graphviz::graph::{AttrValue, Graph, Node};
use fabro_store::SlateStore;
use fabro_types::Settings;
@ -199,11 +198,7 @@ impl Handler for SubWorkflowHandler {
Duration::from_millis(1),
));
let run_store = store
.create_run(
&child_run_options.run_id,
Utc::now(),
Some(child_run_options.run_dir.to_string_lossy().as_ref()),
)
.create_run(&child_run_options.run_id)
.await
.map_err(|err| FabroError::engine(err.to_string()))?;

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

@ -95,7 +95,7 @@ impl EngineServices {
.expect("test runtime should initialize")
.block_on(async {
store
.create_run(&fabro_types::RunId::new(), chrono::Utc::now(), None)
.create_run(&fabro_types::RunId::new())
.await
.expect("slate-backed test run store should initialize")
})

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

@ -692,10 +692,7 @@ mod tests {
#[tokio::test]
async fn parallel_handler_stores_results_in_run_store() {
let store = test_store();
let run_store = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let services = EngineServices {
emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)),
run_store: run_store.clone(),

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

@ -232,10 +232,7 @@ mod tests {
crate::event::StoreProgressLogger,
) {
let store = test_store();
let run_store = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let services = EngineServices {
emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)),
run_store: run_store.clone(),

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

@ -1,4 +1,4 @@
use chrono::{Local, Utc};
use chrono::Local;
use fabro_graphviz::graph::{AttrValue, Graph};
use fabro_model::{Catalog, Provider};
use fabro_sandbox::SandboxProvider;
@ -27,7 +27,6 @@ pub struct CreateRunInput {
pub settings: Settings,
pub cwd: PathBuf,
pub workflow_slug: Option<String>,
pub run_dir: Option<PathBuf>,
pub run_id: Option<RunId>,
pub host_repo_path: Option<String>,
pub base_branch: Option<String>,
@ -43,8 +42,8 @@ pub struct CreatedRun {
struct PersistCreateOptions {
settings: Settings,
run_dir: Option<PathBuf>,
run_id: Option<RunId>,
run_dir: Option<PathBuf>,
workflow_slug: Option<String>,
labels: HashMap<String, String>,
base_branch: Option<String>,
@ -70,7 +69,6 @@ pub async fn create(store: &SlateStore, request: CreateRunInput) -> Result<Creat
settings: _,
cwd: _,
workflow_slug,
run_dir,
run_id,
host_repo_path,
base_branch,
@ -79,13 +77,7 @@ pub async fn create(store: &SlateStore, request: CreateRunInput) -> Result<Creat
let settings = resolved.settings.clone();
let run_id = run_id.unwrap_or_else(RunId::new);
let storage_dir = settings.storage_dir();
let run_dir = run_dir.unwrap_or_else(|| {
make_run_dir(
&storage_dir.join("runs"),
&run_id.to_string(),
settings.dry_run_enabled(),
)
});
let run_dir = make_run_dir(&storage_dir.join("runs"), &run_id);
let working_directory = resolved.working_directory.clone();
let host_repo_path =
host_repo_path.or_else(|| Some(working_directory.to_string_lossy().to_string()));
@ -102,8 +94,8 @@ pub async fn create(store: &SlateStore, request: CreateRunInput) -> Result<Creat
&resolved.raw_source,
PersistCreateOptions {
settings,
run_dir: Some(run_dir.clone()),
run_id: Some(run_id),
run_dir: Some(run_dir.clone()),
workflow_slug: workflow_slug.or(resolved.workflow_slug.clone()),
labels: resolved.settings.labels.clone(),
base_branch,
@ -135,11 +127,7 @@ async fn persist_created_run(
workflow_config: Option<String>,
) -> Result<(), FabroError> {
let record = persisted.run_record();
let run_dir_string = persisted.run_dir().to_string_lossy().to_string();
let run_store = match store
.create_run(&record.run_id, record.created_at, Some(&run_dir_string))
.await
{
let run_store = match store.create_run(&record.run_id).await {
Ok(run_store) => run_store,
Err(err) => store
.open_run(&record.run_id)
@ -174,7 +162,7 @@ async fn persist_created_run(
workflow_slug: record.workflow_slug.clone(),
db_prefix: None,
},
record.created_at,
record.run_id.created_at(),
);
let payload = fabro_store::EventPayload::new(
serde_json::to_value(&envelope).map_err(|err| FabroError::engine(err.to_string()))?,
@ -280,8 +268,8 @@ fn persist_validated(
) -> Result<Persisted, FabroError> {
let PersistCreateOptions {
settings,
run_dir,
run_id,
run_dir,
workflow_slug,
labels,
base_branch,
@ -292,12 +280,10 @@ fn persist_validated(
let settings = resolve_run_settings(settings, validated.graph());
let run_id = run_id.unwrap_or_else(RunId::new);
let run_dir =
run_dir.unwrap_or_else(|| default_run_dir(&run_id.to_string(), settings.dry_run_enabled()));
let run_dir = run_dir.unwrap_or_else(|| default_run_dir(&run_id));
let run_record = RunRecord {
run_id,
created_at: Utc::now(),
settings,
graph: validated.graph().clone(),
workflow_slug,
@ -361,20 +347,13 @@ pub(crate) fn resolve_run_settings(mut settings: Settings, graph: &Graph) -> Set
settings
}
pub(crate) fn default_run_dir(run_id: &str, dry_run: bool) -> PathBuf {
make_run_dir(&default_runs_base(), run_id, dry_run)
pub(crate) fn default_run_dir(run_id: &RunId) -> PathBuf {
make_run_dir(&default_runs_base(), run_id)
}
pub(crate) fn make_run_dir(runs_base: &Path, run_id: &str, dry_run: bool) -> PathBuf {
if dry_run {
runs_base.join(format!(
"{}-dry-run-{}",
Local::now().format("%Y%m%d"),
run_id
))
} else {
runs_base.join(format!("{}-{}", Local::now().format("%Y%m%d"), run_id))
}
pub fn make_run_dir(runs_base: &Path, run_id: &RunId) -> PathBuf {
let local_created_at = run_id.created_at().with_timezone(&Local);
runs_base.join(format!("{}-{}", local_created_at.format("%Y%m%d"), run_id))
}
#[cfg(test)]
@ -600,7 +579,6 @@ mod tests {
settings: Settings::default(),
cwd: dir.path().to_path_buf(),
workflow_slug: None,
run_dir: Some(dir.path().join("run")),
run_id: None,
host_repo_path: None,
base_branch: None,
@ -645,7 +623,6 @@ mod tests {
},
cwd: dir.path().to_path_buf(),
workflow_slug: Some("slug".to_string()),
run_dir: Some(dir.path().join("run")),
run_id: Some(fixtures::RUN_1),
host_repo_path: Some(dir.path().display().to_string()),
base_branch: Some("main".to_string()),
@ -721,7 +698,6 @@ mod tests {
},
cwd: dir.path().to_path_buf(),
workflow_slug: None,
run_dir: Some(dir.path().join("run")),
run_id: Some(fixtures::RUN_2),
host_repo_path: None,
base_branch: None,
@ -771,7 +747,6 @@ mod tests {
},
cwd: dir.path().to_path_buf(),
workflow_slug: Some("slug".to_string()),
run_dir: Some(run_dir.clone()),
run_id: Some(fixtures::RUN_3),
host_repo_path: None,
base_branch: None,

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

@ -85,16 +85,14 @@ fn fork_from_entry(
let run_record_bytes =
run_record_bytes.ok_or_else(|| anyhow::anyhow!("source run has no run.json"))?;
let now = chrono::Utc::now();
let mut run_record: RunRecord =
serde_json::from_slice(&run_record_bytes).context("failed to parse source run.json")?;
run_record.run_id = new_run_id;
run_record.created_at = now;
let new_run_record_bytes =
serde_json::to_vec_pretty(&run_record).context("failed to serialize new run.json")?;
let new_start_record_bytes = if start_record_bytes.is_some() {
let now = new_run_id.created_at();
let start_record = StartRecord {
run_id: new_run_id,
start_time: now,

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

@ -10,7 +10,7 @@ mod test_support;
mod validate;
pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec};
pub use create::{CreateRunInput, CreatedRun, create};
pub use create::{CreateRunInput, CreatedRun, create, make_run_dir};
pub use fork::{ForkRunInput, fork};
pub use rebuild_meta::{
build_timeline_or_rebuild, find_run_id_by_prefix_or_store, rebuild_metadata_branch,

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

@ -183,9 +183,9 @@ pub async fn find_run_id_by_prefix_or_store(
let current_repo_root = canonical_repo_root(repo)?;
let mut matches = Vec::new();
for summary in fabro_store.list_runs().await? {
if summary.catalog.run_id.to_string() == prefix {
if summary.run_id.to_string() == prefix {
if summary.host_repo_path.is_none() {
return Ok(summary.catalog.run_id);
return Ok(summary.run_id);
}
let Some(host_repo_path) = summary.host_repo_path.as_deref() else {
@ -198,7 +198,7 @@ pub async fn find_run_id_by_prefix_or_store(
continue;
};
if host_repo_root == current_repo_root {
return Ok(summary.catalog.run_id);
return Ok(summary.run_id);
}
continue;
}
@ -212,10 +212,8 @@ pub async fn find_run_id_by_prefix_or_store(
let Ok(host_repo_root) = canonical_repo_root(&host_repo) else {
continue;
};
if host_repo_root == current_repo_root
&& summary.catalog.run_id.to_string().starts_with(prefix)
{
matches.push(summary.catalog.run_id);
if host_repo_root == current_repo_root && summary.run_id.to_string().starts_with(prefix) {
matches.push(summary.run_id);
}
}
@ -369,7 +367,6 @@ mod tests {
fn sample_run_record(run_id: RunId, host_repo_path: Option<&str>) -> RunRecord {
RunRecord {
run_id,
created_at: created_at(),
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: None,
@ -431,7 +428,7 @@ mod tests {
run_id: RunId,
host_repo_path: Option<&str>,
) -> DurableRunStore {
let run_store = store.create_run(&run_id, created_at(), None).await.unwrap();
let run_store = store.create_run(&run_id).await.unwrap();
let run_record = sample_run_record(run_id, host_repo_path);
append_workflow_event(
&run_store,
@ -815,10 +812,7 @@ mod tests {
async fn rebuild_metadata_branch_errors_when_run_record_is_missing() {
let (_dir, git_store) = temp_repo();
let durable_store = memory_store();
let run_store = durable_store
.create_run(&test_run_id(), created_at(), None)
.await
.unwrap();
let run_store = durable_store.create_run(&test_run_id()).await.unwrap();
let err = rebuild_metadata_branch(&git_store, &run_store, &test_run_id())
.await

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

@ -860,7 +860,6 @@ mod tests {
.unwrap_or_else(|| Path::new("."))
.to_path_buf(),
workflow_slug: Some("test".to_string()),
run_dir: Some(run_dir.to_path_buf()),
run_id: Some(fixtures::RUN_1),
host_repo_path: None,
base_branch: None,

6
lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -132,7 +132,6 @@ fn persisted_workflow(graph: Graph, source: String, run_dir: &Path, run_id: RunI
run_dir.to_path_buf(),
RunRecord {
run_id,
created_at: Utc::now(),
settings: Settings::default(),
graph,
workflow_slug: Some("test".to_string()),
@ -163,10 +162,7 @@ async fn test_run_store(run_id: &RunId) -> fabro_store::SlateRunStore {
"",
Duration::from_millis(1),
));
store
.create_run(run_id, chrono::Utc::now(), None)
.await
.unwrap()
store.create_run(run_id).await.unwrap()
}
async fn execute_test_run(run_dir: &Path, graph: Graph, run_id: &str) -> Executed {

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

@ -364,14 +364,7 @@ mod tests {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let inner_store = test_store()
.create_run(
&test_run_id(),
Utc::now(),
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
let inner_store = test_store().create_run(&test_run_id()).await.unwrap();
let run_store = inner_store;
let emitter = Arc::new(EventEmitter::new(test_run_id()));
let store_logger = StoreProgressLogger::new(run_store.clone());

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

@ -743,7 +743,6 @@ mod tests {
run_dir.to_path_buf(),
RunRecord {
run_id: test_run_id(),
created_at: Utc::now(),
settings: Settings::default(),
graph,
workflow_slug: Some("test".to_string()),
@ -770,14 +769,7 @@ mod tests {
run_id: test_run_id(),
run_store: {
let store = memory_store();
let inner = store
.create_run(
&test_run_id(),
chrono::Utc::now(),
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
let inner = store.create_run(&test_run_id()).await.unwrap();
inner
},
dry_run: false,
@ -839,14 +831,7 @@ mod tests {
let persisted = test_persisted(graph, source, &run_dir);
let emitter = Arc::new(crate::event::EventEmitter::new(test_run_id()));
let store = memory_store();
let run_store = store
.create_run(
&test_run_id(),
chrono::Utc::now(),
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
let run_store = store.create_run(&test_run_id()).await.unwrap();
let store_logger = StoreProgressLogger::new(run_store.clone());
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
emitter.on_event({

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

@ -33,10 +33,9 @@ pub(crate) async fn load_from_store(
.state()
.await
.map_err(|err| FabroError::engine(err.to_string()))?;
let mut run_record = state
let run_record = state
.run
.ok_or_else(|| FabroError::Precondition("run record missing from store".to_string()))?;
run_record.created_at = run_store.created_at();
let graph = run_record.graph.clone();
let source = state.graph_source.unwrap_or_default();
@ -121,7 +120,6 @@ mod tests {
fn sample_record(graph: Graph) -> RunRecord {
RunRecord {
run_id: fixtures::RUN_1,
created_at: Utc::now(),
settings: Settings {
dry_run: Some(true),
verbose: Some(true),
@ -145,14 +143,7 @@ mod tests {
source: Option<&str>,
) -> SlateRunStore {
let store = memory_store();
let run_store = store
.create_run(
&record.run_id,
record.created_at,
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
let run_store = store.create_run(&record.run_id).await.unwrap();
append_workflow_event(
&run_store,
&record.run_id,
@ -246,8 +237,9 @@ mod tests {
let loaded_record = loaded.run_record();
assert_eq!(loaded_record.run_id, expected.run_id);
assert!(
(loaded_record.created_at.timestamp_millis() - expected.created_at.timestamp_millis())
.abs()
(loaded_record.run_id.created_at().timestamp_millis()
- expected.run_id.created_at().timestamp_millis())
.abs()
<= 1
);
assert_eq!(loaded_record.settings, expected.settings);

48
lib/crates/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -1059,14 +1059,7 @@ mod tests {
let tmp = tempfile::tempdir().unwrap();
let store = test_store();
let run_store = store
.create_run(
&fixtures::RUN_1,
Utc::now(),
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let conclusion = make_test_conclusion();
let body = build_pr_body(
"diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n",
@ -1091,18 +1084,10 @@ mod tests {
let tmp = tempfile::tempdir().unwrap();
let store = test_store();
let created_at = Utc::now();
let run_store = store
.create_run(
&fixtures::RUN_1,
created_at,
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let run_record = RunRecord {
run_id: fixtures::RUN_1,
created_at,
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: Some("test".to_string()),
@ -1167,18 +1152,10 @@ mod tests {
let tmp = tempfile::tempdir().unwrap();
let store = test_store();
let created_at = Utc::now();
let run_store = store
.create_run(
&fixtures::RUN_1,
created_at,
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let run_record = RunRecord {
run_id: fixtures::RUN_1,
created_at,
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: Some("test".to_string()),
@ -1369,14 +1346,7 @@ mod tests {
async fn empty_diff_returns_none() {
let tmp = tempfile::tempdir().unwrap();
let store = test_store();
let run_store = store
.create_run(
&fixtures::RUN_1,
Utc::now(),
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let creds = GitHubAppCredentials {
app_id: "123".to_string(),
private_key_pem: "unused".to_string(),
@ -1404,17 +1374,9 @@ mod tests {
let tmp = tempfile::tempdir().unwrap();
let store = test_store();
let created_at = Utc::now();
let run_store = store
.create_run(
&fixtures::RUN_1,
created_at,
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
let run_record = RunRecord {
run_id: fixtures::RUN_1,
created_at,
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: None,

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

@ -217,18 +217,10 @@ mod tests {
checkpoint: &Checkpoint,
) -> fabro_store::SlateRunStore {
let created_at = Utc::now();
let inner = test_store()
.create_run(
&test_run_id(),
created_at,
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
let inner = test_store().create_run(&test_run_id()).await.unwrap();
let run_store = inner;
let run_record = RunRecord {
run_id: test_run_id(),
created_at,
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: None,

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

@ -7,6 +7,7 @@ use fabro_store::{RunSummary, SlateStore};
use fabro_types::RunId;
use serde::Serialize;
use crate::operations::make_run_dir;
use crate::run_status::{RunStatus, StatusReason};
#[derive(Debug, Clone)]
@ -48,7 +49,7 @@ impl RunInfo {
pub fn run_id(&self) -> RunId {
self.summary
.as_ref()
.map(|summary| summary.catalog.run_id)
.map(|summary| summary.run_id)
.or_else(|| parse_run_id(&self.dir_name))
.expect("RunInfo must have a run id")
}
@ -82,7 +83,7 @@ impl RunInfo {
pub fn start_time(&self) -> String {
self.summary
.as_ref()
.and_then(|summary| summary.start_time.or(Some(summary.catalog.created_at)))
.and_then(|summary| summary.start_time.or(Some(summary.run_id.created_at())))
.or(self.start_time_dt)
.map(|time| time.to_rfc3339())
.unwrap_or_default()
@ -194,7 +195,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<R
if let Ok(store_runs) = store.list_runs().await {
for summary in store_runs {
let Some(run_info) = run_info_from_summary(summary) else {
let Some(run_info) = run_info_from_summary(summary, base) else {
continue;
};
runs_by_id.insert(run_info.run_id(), run_info);
@ -217,16 +218,13 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<R
Ok(runs)
}
fn run_info_from_summary(summary: RunSummary) -> Option<RunInfo> {
let run_dir = summary.catalog.run_dir.as_deref()?;
let path = PathBuf::from(run_dir);
fn run_info_from_summary(summary: RunSummary, runs_base: &Path) -> Option<RunInfo> {
let path = make_run_dir(runs_base, &summary.run_id);
if !path.exists() {
return None;
}
let dir_name = path
.file_name()
.map(|name| name.to_string_lossy().to_string())?;
let start_time_dt = summary.catalog.created_at;
let dir_name = path.file_name()?.to_string_lossy().to_string();
let start_time_dt = summary.run_id.created_at();
let end_time = if summary.status.is_some_and(RunStatus::is_terminal) {
summary.duration_ms.and_then(|duration_ms| {
Some(start_time_dt + chrono::Duration::milliseconds(i64::try_from(duration_ms).ok()?))
@ -382,7 +380,6 @@ mod tests {
fn sample_run_record() -> RunRecord {
RunRecord {
run_id: fixtures::RUN_1,
created_at: Utc::now(),
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: Some("test".to_string()),
@ -403,14 +400,7 @@ mod tests {
let store = memory_store();
let run_dir_string = run_dir.to_string_lossy().to_string();
let run_record = sample_run_record();
let run_store = store
.create_run(
&fixtures::RUN_1,
run_record.created_at,
Some(&run_dir_string),
)
.await
.unwrap();
let run_store = store.create_run(&fixtures::RUN_1).await.unwrap();
append_workflow_event(
&run_store,
&fixtures::RUN_1,

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

@ -3,7 +3,6 @@ use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use fabro_agent::Sandbox;
use fabro_graphviz::graph::Graph as GvGraph;
use fabro_store::SlateStore;
@ -40,18 +39,13 @@ async fn initialized(
options: InitializedOptions,
) -> Initialized {
std::fs::create_dir_all(&run_options.run_dir).expect("failed to create run dir");
let created_at = Utc::now();
let store = Arc::new(SlateStore::new(
Arc::new(InMemory::new()),
"",
Duration::from_millis(1),
));
let inner_store = store
.create_run(
&run_options.run_id,
created_at,
Some(run_options.run_dir.to_string_lossy().as_ref()),
)
.create_run(&run_options.run_id)
.await
.expect("failed to create slate-backed test run store");
let run_store = inner_store;