scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-04 02:33:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Derive run metadata from RunId

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-03 17:46:44 -07:00
parent 7bc0b5db3a
commit 0b7e284924
No known key found for this signature in database
40 changed files with 354 additions and 623 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -2028,6 +2028,7 @@ dependencies = [
"tokio",
"tokio-stream",
"tracing",
"ulid",
]
[[package]]

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,

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

@ -163,7 +163,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(),
@ -171,7 +171,6 @@ mod tests {
);
RunRecord {
run_id,
created_at,
settings: Settings::default(),
graph,
workflow_slug: Some("night-sky".to_string()),
@ -283,7 +282,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();
@ -298,7 +297,7 @@ mod tests {
visit: 2,
};
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::RunCreated {
run_id,
@ -318,7 +317,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::WorkflowRunStarted {
name: "night-sky".to_string(),
@ -333,7 +332,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::RunRunning {
reason: status_record.reason,
@ -343,7 +342,7 @@ mod tests {
.unwrap();
for checkpoint in [&first_checkpoint, &second_checkpoint] {
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::CheckpointCompleted {
node_id: checkpoint.current_node.clone(),
@ -375,7 +374,7 @@ mod tests {
.unwrap();
}
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::SandboxInitialized {
working_directory: sandbox.working_directory.clone(),
@ -388,7 +387,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::Prompt {
stage: "code".to_string(),
@ -402,7 +401,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::PromptCompleted {
node_id: "code".to_string(),
@ -415,7 +414,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::StageCompleted {
node_id: "code".to_string(),
@ -446,7 +445,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::CommandStarted {
node_id: "code".to_string(),
@ -458,7 +457,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::CommandCompleted {
node_id: "code".to_string(),
@ -472,7 +471,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::RetroStarted {
prompt: Some("How did it go?".to_string()),
@ -483,7 +482,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::RetroCompleted {
duration_ms: 50,
@ -494,7 +493,7 @@ mod tests {
.await
.unwrap();
append_workflow_event(
run.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::WorkflowRunCompleted {
duration_ms: conclusion.duration_ms,
@ -542,7 +541,7 @@ mod tests {
.unwrap();
let output = tempfile::tempdir().unwrap();
let file_count = export_run(run.as_ref(), output.path()).await.unwrap();
let file_count = export_run(&run, output.path()).await.unwrap();
assert_eq!(file_count, 22);
let exported_run: RunRecord = read_json(&output.path().join("run.json"));
@ -637,10 +636,10 @@ 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.as_ref(),
&run,
&run_id,
&WorkflowRunEvent::RunCreated {
run_id,
@ -672,7 +671,7 @@ mod tests {
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("dump");
let err = export_run(run.as_ref(), &output).await.unwrap_err();
let err = export_run(&run, &output).await.unwrap_err();
assert!(err.to_string().contains("asset filename"));
assert!(!output.exists());
}

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

@ -1,4 +1,3 @@
#[cfg(feature = "server")]
use fabro_test::{fabro_snapshot, test_context};
#[test]

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

@ -1,4 +1,3 @@
#[cfg(feature = "server")]
use fabro_test::{fabro_snapshot, test_context};
#[test]

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

@ -1,4 +1,3 @@
#[cfg(feature = "server")]
use fabro_test::{fabro_snapshot, test_context};
#[test]

1
lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs
View file

@ -1,4 +1,3 @@
#[cfg(feature = "server")]
use fabro_test::{fabro_snapshot, test_context};
#[test]

1
lib/crates/fabro-cli/tests/it/workflow/real_cli.rs
View file

@ -32,7 +32,6 @@ async fn run_real_cli_test(provider: Provider, model: &str) {
&context,
None,
&emitter,
workspace.path(),
&env,
None,
)

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

@ -94,7 +94,6 @@ struct ManagedRun {
dot_source: String,
status: RunStatus,
error: Option<String>,
created_at: chrono::DateTime<chrono::Utc>,
// Populated when running:
interviewer: Option<Arc<WebInterviewer>>,
event_tx: Option<broadcast::Sender<RunEventEnvelope>>,
@ -496,7 +495,7 @@ async fn list_runs(
message: msg.clone(),
}),
queue_position: queue_positions.get(id).copied(),
created_at: managed_run.created_at,
created_at: id.created_at(),
})
.collect();
let page: Vec<_> = all_items.into_iter().skip(offset).take(limit + 1).collect();
@ -517,7 +516,7 @@ fn compute_queue_positions(runs: &HashMap<RunId, ManagedRun>) -> HashMap<RunId,
.iter()
.filter(|(_, r)| r.status == RunStatus::Queued)
.collect();
queued.sort_by_key(|(_, r)| r.created_at);
queued.sort_by_key(|(run_id, _)| run_id.created_at());
queued
.into_iter()
.enumerate()
@ -545,7 +544,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(),
@ -557,7 +555,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,
@ -589,8 +586,7 @@ 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 mut runs = state.runs.lock().expect("runs lock poisoned");
@ -600,14 +596,13 @@ async fn start_run(
dot_source: req.dot_source,
status: RunStatus::Queued,
error: None,
created_at,
interviewer: None,
event_tx: None,
context: None,
checkpoint: None,
cancel_tx: None,
cancel_token: None,
run_dir: Some(run_dir),
run_dir: Some(created.run_dir),
},
);
}
@ -853,7 +848,7 @@ pub fn spawn_scheduler(state: Arc<AppState>) {
}
runs.iter()
.filter(|(_, r)| r.status == RunStatus::Queued)
.min_by_key(|(_, r)| r.created_at)
.min_by_key(|(run_id, _)| run_id.created_at())
.map(|(id, _)| *id)
};
match run_to_start {
@ -894,7 +889,7 @@ async fn get_run_status(
error: managed_run.error.as_ref().map(|msg| RunError {
message: msg.clone(),
}),
created_at: managed_run.created_at,
created_at: id.created_at(),
queue_position,
}),
)
@ -1132,7 +1127,6 @@ async fn cancel_run(
let _ = cancel_tx.send(());
}
managed_run.status = RunStatus::Cancelled;
let created_at = managed_run.created_at;
(
StatusCode::OK,
Json(RunStatusResponse {
@ -1140,7 +1134,7 @@ async fn cancel_run(
status: RunStatus::Cancelled,
error: None,
queue_position: None,
created_at,
created_at: id.created_at(),
}),
)
.into_response()
@ -1165,7 +1159,6 @@ async fn pause_run(
Some(managed_run) => match managed_run.status {
RunStatus::Running => {
managed_run.status = RunStatus::Paused;
let created_at = managed_run.created_at;
(
StatusCode::OK,
Json(RunStatusResponse {
@ -1173,7 +1166,7 @@ async fn pause_run(
status: RunStatus::Paused,
error: None,
queue_position: None,
created_at,
created_at: id.created_at(),
}),
)
.into_response()
@ -1198,7 +1191,6 @@ async fn unpause_run(
Some(managed_run) => match managed_run.status {
RunStatus::Paused => {
managed_run.status = RunStatus::Running;
let created_at = managed_run.created_at;
(
StatusCode::OK,
Json(RunStatusResponse {
@ -1206,7 +1198,7 @@ async fn unpause_run(
status: RunStatus::Running,
error: None,
queue_position: None,
created_at,
created_at: id.created_at(),
}),
)
.into_response()

1
lib/crates/fabro-store/Cargo.toml
View file

@ -29,3 +29,4 @@ futures.workspace = true
[dev-dependencies]
tokio = { workspace = true, features = ["test-util", "macros"] }
tempfile = "3"
ulid.workspace = true

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

@ -13,7 +13,7 @@ pub use error::{Result, StoreError};
pub use run_state::{NodeState, RunState};
pub use runtime::RuntimeState;
pub use slate::{SlateRunStore, SlateStore};
pub use types::{CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, RunSummary};
pub use types::{EventEnvelope, EventPayload, NodeVisitRef, RunSummary};
use fabro_types::{Outcome, StageUsage};

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

@ -6,9 +6,7 @@ use chrono::{DateTime, Utc};
use serde::de::DeserializeOwned;
use serde_json::Value;
use crate::{
CatalogRecord, EventEnvelope, NodeOutcomeRecord, NodeVisitRef, Result, RunSummary, StoreError,
};
use crate::{EventEnvelope, NodeOutcomeRecord, NodeVisitRef, Result, RunSummary, StoreError};
use fabro_types::{
Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro,
RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StartRecord,
@ -90,7 +88,6 @@ impl RunState {
.collect::<HashMap<_, _>>();
self.run = Some(RunRecord {
run_id,
created_at: ts,
settings,
graph,
workflow_slug: optional_string(&properties, "workflow_slug"),
@ -296,7 +293,7 @@ impl RunState {
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()
@ -309,10 +306,7 @@ impl RunState {
(!goal.is_empty()).then(|| goal.to_string())
});
RunSummary {
run_id: catalog.run_id,
created_at: catalog.created_at,
db_prefix: catalog.db_prefix.clone(),
run_dir: catalog.run_dir.clone(),
run_id: *run_id,
workflow_name,
workflow_slug: self.run.as_ref().and_then(|run| run.workflow_slug.clone()),
goal,

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

@ -1,79 +1,82 @@
use std::collections::HashSet;
use std::sync::Arc;
use chrono::{DateTime, Utc};
use bytes::Bytes;
use futures::TryStreamExt;
use object_store::ObjectStore;
use object_store::path::Path;
use crate::{CatalogRecord, ListRunsQuery, Result};
use crate::{ListRunsQuery, Result};
use fabro_types::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::new().into())
.await?;
store
.put(
&by_start_path(base_prefix, created_at, run_id),
bytes.into(),
)
.put(&by_start_path(base_prefix, run_id), 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();
let mut seen = HashSet::new();
for meta in metas {
let Some(record) = read_catalog_path(store.clone(), meta.location).await? else {
let Some(run_id) = parse_run_id_from_path(&meta.location) else {
continue;
};
if !seen.insert(run_id) {
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 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()
}
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")
)
}
@ -81,28 +84,16 @@ 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)]
pub(super) mod test_support {
use super::*;
use std::collections::{HashMap, HashSet};
pub(crate) async fn repair_catalog(
store: Arc<dyn ObjectStore>,
@ -115,17 +106,15 @@ pub(super) mod test_support {
.list(Some(&by_id_prefix))
.try_collect::<Vec<_>>()
.await?;
let mut canonical = HashMap::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);
}
}
let run_ids = by_id_metas
.iter()
.filter_map(|meta| parse_run_id_from_path(&meta.location))
.collect::<Vec<_>>();
for record in canonical.values() {
let path = by_start_path(base_prefix, record.created_at, &record.run_id);
for run_id in &run_ids {
let path = by_start_path(base_prefix, run_id);
if !object_exists(store.clone(), &path).await? {
store.put(&path, serde_json::to_vec(record)?.into()).await?;
store.put(&path, Bytes::new().into()).await?;
}
}
@ -133,33 +122,26 @@ pub(super) mod test_support {
.list(Some(&by_start_prefix))
.try_collect::<Vec<_>>()
.await?;
let canonical = run_ids.into_iter().collect::<HashSet<_>>();
let mut seen = HashSet::new();
for meta in by_start_metas {
let location = meta.location.clone();
let Some(record) = read_catalog_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(&record.run_id).map(|canonical_record| {
by_start_path(base_prefix, canonical_record.created_at, &record.run_id)
});
match expected {
Some(expected) if expected == location => {
seen.insert(record.run_id);
}
_ => {
delete_if_exists(store.clone(), &location).await?;
}
let expected = by_start_path(base_prefix, &run_id);
if canonical.contains(&run_id) && expected == location {
seen.insert(run_id);
continue;
}
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(record)?.into(),
)
.put(&by_start_path(base_prefix, &run_id), Bytes::new().into())
.await?;
}
}

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

@ -5,7 +5,6 @@ use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use chrono::{DateTime, Utc};
use futures::TryStreamExt;
use object_store::ObjectStore;
use object_store::path::Path;
@ -14,7 +13,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;
pub use run_store::SlateRunStore;
use run_store::SlateRunStoreInner;
@ -96,7 +95,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> {
@ -104,21 +103,25 @@ impl SlateStore {
weak.upgrade().map(SlateRunStore::from_inner)
}
async fn open_run_store(&self, record: &CatalogRecord) -> Result<Option<SlateRunStore>> {
if let Some(active) = self.get_active_run(&record.run_id).await {
if active.matches_record(record) {
async fn open_run_store(
&self,
run_id: &RunId,
db_prefix: &str,
) -> Result<Option<SlateRunStore>> {
if let Some(active) = self.get_active_run(run_id).await {
if active.matches_run(run_id, db_prefix) {
return Ok(Some(active));
}
return Err(StoreError::Other(format!(
"active run cache mismatch for run_id {:?}",
record.run_id
run_id
)));
}
if !self.db_prefix_has_objects(&record.db_prefix).await? {
if !self.db_prefix_has_objects(db_prefix).await? {
return Ok(None);
}
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;
@ -129,17 +132,21 @@ 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_prefix.to_string(), db).await?;
self.cache_active_run(&run_store).await;
Ok(Some(run_store))
}
async fn open_run_reader_store(&self, record: &CatalogRecord) -> Result<Option<SlateRunStore>> {
if !self.db_prefix_has_objects(&record.db_prefix).await? {
async fn open_run_reader_store(
&self,
run_id: &RunId,
db_prefix: &str,
) -> Result<Option<SlateRunStore>> {
if !self.db_prefix_has_objects(db_prefix).await? {
return Ok(None);
}
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;
@ -150,7 +157,7 @@ impl SlateStore {
let _ = reader.close().await;
return Ok(None);
}
SlateRunStore::open_reader(record.clone(), reader)
SlateRunStore::open_reader(*run_id, db_prefix.to_string(), reader)
.await
.map(Some)
}
@ -170,140 +177,111 @@ impl SlateStore {
}
impl SlateStore {
pub async fn create_run(
&self,
run_id: &RunId,
created_at: DateTime<Utc>,
run_dir: Option<&str>,
) -> Result<SlateRunStore> {
let locator =
pub async fn create_run(&self, run_id: &RunId) -> Result<SlateRunStore> {
let locator_exists =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
let db_prefix = catalog::db_prefix(&self.base_prefix, run_id);
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_exists && !active.matches_run(run_id, &db_prefix) {
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),
};
if locator_exists && self.db_prefix_has_objects(&db_prefix).await? {
return Err(StoreError::RunAlreadyExists(run_id.to_string()));
}
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_prefix.clone(), 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 exists =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
if !exists {
return Err(StoreError::RunNotFound(run_id.to_string()));
}
let db_prefix = catalog::db_prefix(&self.base_prefix, run_id);
let run_store = self
.open_run_store(&locator)
.open_run_store(run_id, &db_prefix)
.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 exists =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
if !exists {
return Err(StoreError::RunNotFound(run_id.to_string()));
}
let db_prefix = catalog::db_prefix(&self.base_prefix, run_id);
let run_store = self
.open_run_reader_store(&locator)
.open_run_reader_store(run_id, &db_prefix)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
Ok(run_store)
}
pub async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>> {
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.matches_record(&record) {
for run_id in run_ids {
let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id);
if let Some(active) = self.get_active_run(&run_id).await {
if !active.matches_run(&run_id, &db_prefix) {
return Err(StoreError::Other(format!(
"active run cache mismatch for run_id {:?}",
record.run_id
run_id
)));
}
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? {
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.created_at.cmp(&a.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(locator) =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?
{
let db_prefix = catalog::db_prefix(&self.base_prefix, run_id);
if catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? {
delete_path(
self.object_store.clone(),
&catalog::by_start_path(&self.base_prefix, locator.created_at, run_id),
&catalog::by_start_path(&self.base_prefix, run_id),
)
.await?;
self.delete_db_prefix(&locator.db_prefix).await?;
self.delete_db_prefix(&db_prefix).await?;
delete_path(
self.object_store.clone(),
&catalog::by_id_path(&self.base_prefix, run_id),
@ -312,13 +290,13 @@ impl SlateStore {
return Ok(());
}
if let Some(record) = active_record {
if active.is_some() {
delete_path(
self.object_store.clone(),
&catalog::by_start_path(&self.base_prefix, record.created_at, run_id),
&catalog::by_start_path(&self.base_prefix, run_id),
)
.await?;
self.delete_db_prefix(&record.db_prefix).await?;
self.delete_db_prefix(&db_prefix).await?;
delete_path(
self.object_store.clone(),
&catalog::by_id_path(&self.base_prefix, run_id),
@ -338,16 +316,9 @@ impl SlateStore {
if meta.location.filename() != Some(expected_name.as_str()) {
continue;
}
let Some(record) =
catalog::read_catalog_path(self.object_store.clone(), meta.location.clone())
.await?
else {
delete_path(self.object_store.clone(), &meta.location).await?;
continue;
};
self.delete_db_prefix(&record.db_prefix).await?;
delete_path(self.object_store.clone(), &meta.location).await?;
}
self.delete_db_prefix(&db_prefix).await?;
Ok(())
}
}
@ -379,11 +350,11 @@ mod tests {
use std::time::Duration;
use bytes::Bytes;
use chrono::Duration as ChronoDuration;
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use fabro_types::{
AttrValue, Checkpoint, Conclusion, Graph, PullRequestRecord, Retro, RunId, RunRecord,
RunStatus, RunStatusRecord, SandboxRecord, Settings, StageStatus, StartRecord,
StatusReason, fixtures,
StatusReason,
};
use object_store::memory::InMemory;
use slatedb::config::Settings as SlateSettings;
@ -392,6 +363,14 @@ mod tests {
use crate::{EventPayload, NodeVisitRef};
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
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()
@ -409,16 +388,17 @@ mod tests {
}
fn test_run_id(label: &str) -> RunId {
match label {
"run-1" => fixtures::RUN_1,
"other-run" => fixtures::RUN_2,
"run-early" => fixtures::RUN_2,
"run-late" => fixtures::RUN_3,
let (timestamp_ms, random) = match label {
"run-1" => (dt("2026-03-27T12:00:00Z").timestamp_millis() as u64, 1),
"other-run" => (dt("2026-03-27T12:00:00Z").timestamp_millis() as u64, 2),
"run-early" => (dt("2026-03-27T10:00:00Z").timestamp_millis() as u64, 3),
"run-late" => (dt("2026-03-27T12:00:00Z").timestamp_millis() as u64, 4),
_ => panic!("unknown test run id: {label}"),
}
};
RunId::from(ulid::Ulid::from_parts(timestamp_ms, random))
}
fn sample_run_record(run_id: &str, created_at: DateTime<Utc>) -> RunRecord {
fn sample_run_record(run_id: &str, _created_at: DateTime<Utc>) -> RunRecord {
let mut graph = Graph::new("night-sky");
graph.attrs.insert(
"goal".to_string(),
@ -426,7 +406,6 @@ mod tests {
);
RunRecord {
run_id: test_run_id(run_id),
created_at,
settings: Settings::default(),
graph,
workflow_slug: Some("night-sky".to_string()),
@ -587,7 +566,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();
}
@ -598,10 +577,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(
@ -650,7 +626,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);
@ -680,7 +656,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,
};
@ -731,7 +707,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
);
@ -741,10 +717,7 @@ mod tests {
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",
@ -804,7 +777,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,
};
@ -819,7 +792,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
@ -838,23 +811,11 @@ mod tests {
#[tokio::test]
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();
let _created_at = dt("2026-03-27T12:00:00Z");
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(_))));
}
@ -862,10 +823,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",
@ -916,11 +874,8 @@ mod tests {
#[tokio::test]
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 _created_at = dt("2026-03-27T12:00:00Z");
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(
@ -951,7 +906,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;
@ -979,7 +934,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
@ -993,11 +948,8 @@ mod tests {
#[tokio::test]
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 _created_at = dt("2026-03-27T12:00:00Z");
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
.await
.unwrap();
@ -1019,24 +971,20 @@ mod tests {
#[tokio::test]
async fn repair_catalog_removes_stale_wrong_time_prefixes() {
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 _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")).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"))
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::new().into(),
)
.await
.unwrap();
@ -1051,38 +999,33 @@ 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 _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 _created_at = dt("2026-03-27T12:00:00Z");
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)),
@ -1091,21 +1034,15 @@ mod tests {
.build()
.await
.unwrap();
let mismatched = CatalogRecord {
run_id: test_run_id("other-run"),
created_at,
db_prefix,
run_dir: None,
};
db.put(keys::init(), serde_json::to_vec(&mismatched).unwrap())
.await
.unwrap();
db.put(
keys::init(),
serde_json::to_vec(&test_run_id("other-run")).unwrap(),
)
.await
.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!(
@ -1117,11 +1054,8 @@ mod tests {
#[tokio::test]
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 _created_at = dt("2026-03-27T12:00:00Z");
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let node = NodeVisitRef {
node_id: "code",
visit: 2,
@ -1151,11 +1085,8 @@ mod tests {
#[tokio::test]
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 _created_at = dt("2026-03-27T12:00:00Z");
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
.await
.unwrap();
@ -1200,10 +1131,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);
@ -1450,7 +1378,10 @@ 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.run_id.created_at(),
run_record.run_id.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 {}"));
@ -1519,10 +1450,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");
@ -1749,10 +1677,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",
@ -1871,10 +1796,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: EventPayload = serde_json::from_value(serde_json::json!({
"run_id": "run-1"
@ -1897,10 +1819,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(
@ -1941,10 +1860,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(
@ -1965,10 +1881,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(

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

@ -3,7 +3,7 @@ use std::sync::{Arc, Weak};
use std::time::Duration;
use bytes::Bytes;
use chrono::{DateTime, Utc};
use chrono::Utc;
use futures::Stream;
use serde::Serialize;
use serde::de::DeserializeOwned;
@ -14,10 +14,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::run_state::EventProjectionCache;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, Result, RunState, RunSummary,
StoreError,
};
use crate::{EventEnvelope, EventPayload, NodeVisitRef, Result, RunState, RunSummary, StoreError};
use fabro_types::RunId;
#[derive(Clone)]
@ -29,18 +26,14 @@ impl std::fmt::Debug for SlateRunStore {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SlateRunStore")
.field("run_id", &self.inner.run_id)
.field("created_at", &self.inner.created_at)
.field("db_prefix", &self.inner.db_prefix)
.field("run_dir", &self.inner.run_dir)
.finish_non_exhaustive()
}
}
pub(crate) struct SlateRunStoreInner {
run_id: RunId,
created_at: DateTime<Utc>,
db_prefix: String,
run_dir: Option<String>,
db: SlateRunDb,
event_seq: AtomicU32,
close_lock: Mutex<()>,
@ -53,14 +46,16 @@ 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_prefix: String,
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 {
run_id: record.run_id,
created_at: record.created_at,
db_prefix: record.db_prefix,
run_dir: record.run_dir,
run_id,
db_prefix,
db: SlateRunDb::Writer(db),
event_seq: AtomicU32::new(event_seq),
close_lock: Mutex::new(()),
@ -69,14 +64,16 @@ impl SlateRunStore {
})
}
pub(crate) async fn open_reader(record: CatalogRecord, db: DbReader) -> Result<Self> {
pub(crate) async fn open_reader(
run_id: RunId,
db_prefix: String,
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 {
run_id: record.run_id,
created_at: record.created_at,
db_prefix: record.db_prefix,
run_dir: record.run_dir,
run_id,
db_prefix,
db: SlateRunDb::Reader(Box::new(db)),
event_seq: AtomicU32::new(event_seq),
close_lock: Mutex::new(()),
@ -93,24 +90,12 @@ impl SlateRunStore {
Arc::downgrade(&self.inner)
}
pub(crate) fn record(&self) -> CatalogRecord {
CatalogRecord {
run_id: self.inner.run_id,
created_at: self.inner.created_at,
db_prefix: self.inner.db_prefix.clone(),
run_dir: self.inner.run_dir.clone(),
}
pub(crate) fn run_id(&self) -> RunId {
self.inner.run_id
}
pub(crate) fn matches_record(&self, record: &CatalogRecord) -> bool {
self.inner.run_id == record.run_id
&& self.inner.created_at == record.created_at
&& self.inner.db_prefix == record.db_prefix
&& self.inner.run_dir == record.run_dir
}
pub(crate) fn created_at(&self) -> DateTime<Utc> {
self.inner.created_at
pub(crate) fn matches_run(&self, run_id: &RunId, db_prefix: &str) -> bool {
self.inner.run_id == *run_id && self.inner.db_prefix == db_prefix
}
pub(crate) async fn close(&self) -> Result<()> {
@ -129,26 +114,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 = RunState::apply_events(&events)?;
Ok(state.build_summary(catalog))
Ok(state.build_summary(run_id))
}
async fn projected_state(&self) -> Result<RunState> {

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

@ -12,20 +12,9 @@ pub struct NodeVisitRef<'a> {
pub visit: u32,
}
#[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 {
pub run_id: RunId,
pub created_at: DateTime<Utc>,
pub db_prefix: String,
pub run_dir: Option<String>,
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")]

28
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::{TimeZone, Utc};
use super::{RunId, fixtures};
#[test]
@ -172,4 +184,20 @@ mod tests {
assert_eq!(value, fixtures::RUN_42);
}
#[test]
fn exposes_created_at_from_ulid_timestamp() {
let dt = Utc.with_ymd_and_hms(2026, 3, 27, 12, 34, 56).unwrap();
let run_id = RunId::from_datetime(dt);
assert_eq!(run_id.created_at(), dt);
}
#[test]
fn creates_run_id_from_datetime() {
let dt = Utc.with_ymd_and_hms(2026, 3, 27, 12, 34, 56).unwrap();
let run_id = RunId::from_datetime(dt);
assert_eq!(run_id.created_at(), dt);
}
}

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

@ -1891,10 +1891,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

@ -385,10 +385,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 {
run_store: run_store.clone(),
..EngineServices::test_default()

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

@ -193,10 +193,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

@ -88,7 +88,7 @@ impl EngineServices {
)),
run_store: futures::executor::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

@ -673,10 +673,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 {
run_store: run_store.clone(),
..EngineServices::test_default()

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

@ -200,10 +200,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 {
run_store: run_store.clone(),
..EngineServices::test_default()

69
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,7 +42,6 @@ pub struct CreatedRun {
struct PersistCreateOptions {
settings: Settings,
run_dir: Option<PathBuf>,
run_id: Option<RunId>,
workflow_slug: Option<String>,
labels: HashMap<String, String>,
@ -70,7 +68,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 +76,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,7 +93,6 @@ 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),
workflow_slug: workflow_slug.or(resolved.workflow_slug.clone()),
labels: resolved.settings.labels.clone(),
@ -135,11 +125,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 +160,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,7 +266,6 @@ fn persist_validated(
) -> Result<Persisted, FabroError> {
let PersistCreateOptions {
settings,
run_dir,
run_id,
workflow_slug,
labels,
@ -292,12 +277,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 = default_run_dir(&run_id);
let run_record = RunRecord {
run_id,
created_at: Utc::now(),
settings,
graph: validated.graph().clone(),
workflow_slug,
@ -361,25 +344,19 @@ 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(crate) fn make_run_dir(runs_base: &Path, run_id: &RunId) -> PathBuf {
let local_dt = run_id.created_at().with_timezone(&Local);
runs_base.join(format!("{}-{run_id}", local_dt.format("%Y%m%d")))
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::{TimeZone, Utc};
use fabro_graphviz::graph::AttrValue;
use fabro_store::{SlateStore, StoreHandle};
use fabro_types::fixtures;
@ -447,6 +424,24 @@ mod tests {
assert_eq!(prompt, "Goal: Fix bugs");
}
#[test]
fn make_run_dir_uses_run_id_timestamp_in_local_time() {
let runs_base = Path::new("/tmp/runs");
let run_id = RunId::from(ulid::Ulid::from_datetime(
Utc.with_ymd_and_hms(2026, 3, 27, 12, 0, 0).unwrap().into(),
));
let expected_date = run_id
.created_at()
.with_timezone(&Local)
.format("%Y%m%d")
.to_string();
assert_eq!(
make_run_dir(runs_base, &run_id),
runs_base.join(format!("{expected_date}-{run_id}"))
);
}
#[test]
fn validate_applies_stylesheet() {
let dot = r#"digraph Test {
@ -600,7 +595,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 +639,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()),
@ -697,6 +690,7 @@ mod tests {
run_store.state().await.unwrap().status.unwrap().status,
crate::run_status::RunStatus::Submitted
);
assert_eq!(created.run_dir, default_run_dir(&fixtures::RUN_1));
assert!(!created.run_dir.join("id.txt").exists());
}
@ -721,7 +715,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,
@ -748,7 +741,6 @@ mod tests {
async fn create_hydrates_run_created_event_into_store() {
let dir = tempfile::tempdir().unwrap();
let storage_dir = dir.path().join("storage");
let run_dir = dir.path().join("run");
std::fs::create_dir_all(storage_dir.join("store")).unwrap();
let object_store =
Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).unwrap());
@ -771,7 +763,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,

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

@ -90,7 +90,6 @@ fn fork_from_entry(
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")?;

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

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

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

@ -369,7 +369,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 +430,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,
@ -821,10 +820,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,

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

@ -7,7 +7,6 @@ use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::time::Duration;
use async_trait::async_trait;
use chrono::Utc;
use fabro_agent::Sandbox;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_hooks::HookSettings;
@ -132,7 +131,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 +161,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 {

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

@ -320,7 +320,6 @@ mod tests {
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use fabro_graphviz::graph::Graph;
use fabro_store::SlateStore;
use fabro_types::{RunId, Settings, fixtures};
@ -364,14 +363,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());

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

@ -669,7 +669,6 @@ mod tests {
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_interview::AutoApproveInterviewer;
use fabro_sandbox::SandboxSpec;
@ -743,7 +742,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 +768,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 +830,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({

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

@ -53,7 +53,6 @@ mod tests {
use std::collections::HashMap;
use std::path::PathBuf;
use chrono::Utc;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_store::{SlateRunStore, SlateStore, StoreHandle};
use fabro_types::{Settings, fixtures};
@ -120,7 +119,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),
@ -144,14 +142,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,
@ -245,8 +236,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);

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

@ -1058,16 +1058,8 @@ mod tests {
async fn build_pr_body_uses_in_memory_conclusion() {
install_mock_llm();
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",
@ -1089,21 +1081,11 @@ mod tests {
async fn build_pr_body_uses_store_records_without_legacy_files() {
install_mock_llm();
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()),
@ -1122,7 +1104,7 @@ mod tests {
workflow_source: Some("digraph test { plan -> code }".to_string()),
workflow_config: None,
labels: run_record.labels.clone().into_iter().collect(),
run_dir: tmp.path().display().to_string(),
run_dir: run_record.working_directory.display().to_string(),
working_directory: run_record.working_directory.display().to_string(),
host_repo_path: run_record.host_repo_path.clone(),
base_branch: run_record.base_branch.clone(),
@ -1165,21 +1147,11 @@ mod tests {
async fn build_pr_body_uses_plan_text_from_store_without_response_md() {
install_mock_llm();
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()),
@ -1198,7 +1170,7 @@ mod tests {
workflow_source: Some("digraph test { plan -> code }".to_string()),
workflow_config: None,
labels: run_record.labels.clone().into_iter().collect(),
run_dir: tmp.path().display().to_string(),
run_dir: run_record.working_directory.display().to_string(),
working_directory: run_record.working_directory.display().to_string(),
host_repo_path: run_record.host_repo_path.clone(),
base_branch: run_record.base_branch.clone(),
@ -1368,16 +1340,8 @@ mod tests {
#[tokio::test]
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,18 +1368,9 @@ mod tests {
async fn load_pull_request_diff_uses_store_without_disk_patch() {
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,
@ -1434,7 +1389,7 @@ mod tests {
workflow_source: None,
workflow_config: None,
labels: run_record.labels.clone().into_iter().collect(),
run_dir: tmp.path().display().to_string(),
run_dir: run_record.working_directory.display().to_string(),
working_directory: tmp.path().display().to_string(),
host_repo_path: None,
base_branch: None,

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

@ -168,7 +168,6 @@ mod tests {
use std::sync::{Arc, Mutex};
use std::time::Duration;
use chrono::Utc;
use fabro_graphviz::graph::Graph;
use fabro_store::SlateStore;
use fabro_types::{RunId, Settings, fixtures};
@ -216,19 +215,10 @@ mod tests {
run_dir: &std::path::Path,
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,

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

@ -7,6 +7,7 @@ use fabro_store::{ListRunsQuery, SlateStore};
use fabro_types::RunId;
use serde::Serialize;
use crate::operations::make_run_dir;
use crate::run_status::{RunStatus, StatusReason};
#[derive(Debug, Clone, Serialize)]
@ -121,7 +122,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<R
if let Ok(store_runs) = store.list_runs(&ListRunsQuery::default()).await {
for summary in store_runs {
let Some(run_info) = run_info_from_summary(&summary) else {
let Some(run_info) = run_info_from_summary(base, &summary) else {
continue;
};
runs_by_id.insert(summary.run_id, run_info);
@ -144,16 +145,15 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<R
Ok(runs)
}
fn run_info_from_summary(summary: &fabro_store::RunSummary) -> Option<RunInfo> {
let run_dir = summary.run_dir.as_deref()?;
let path = PathBuf::from(run_dir);
fn run_info_from_summary(runs_base: &Path, summary: &fabro_store::RunSummary) -> 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.created_at;
.map(|name: &std::ffi::OsStr| name.to_string_lossy().to_string())?;
let start_time_dt = summary.run_id.created_at();
let start_time = summary.start_time.unwrap_or(start_time_dt);
let end_time = if summary.status.is_some_and(RunStatus::is_terminal) {
summary.duration_ms.and_then(|duration_ms| {
@ -319,7 +319,6 @@ mod tests {
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use fabro_graphviz::graph::Graph;
use fabro_store::{SlateStore, StoreHandle};
use fabro_types::{RunStatus, Settings, fixtures};
@ -327,6 +326,7 @@ mod tests {
use super::scan_runs_combined;
use crate::event::{WorkflowRunEvent, append_workflow_event};
use crate::operations::make_run_dir;
use crate::records::RunRecord;
fn memory_store() -> StoreHandle {
@ -340,7 +340,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()),
@ -354,21 +353,13 @@ mod tests {
#[tokio::test]
async fn scan_runs_combined_uses_store_status_without_status_json() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join(fixtures::RUN_1.to_string());
let run_dir = make_run_dir(temp.path(), &fixtures::RUN_1);
std::fs::create_dir_all(&run_dir).unwrap();
std::fs::write(run_dir.join("id.txt"), format!("{}\n", fixtures::RUN_1)).unwrap();
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,
@ -379,7 +370,7 @@ mod tests {
workflow_source: None,
workflow_config: None,
labels: run_record.labels.clone().into_iter().collect(),
run_dir: run_dir_string.clone(),
run_dir: run_dir.display().to_string(),
working_directory: run_record.working_directory.display().to_string(),
host_repo_path: run_record.host_repo_path.clone(),
base_branch: run_record.base_branch.clone(),

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;