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

merge: resolve conflicts from origin/main

Browse source 
Merged origin/main incorporating:
- db_prefix threading in SlateRunStore for run isolation
- matches_run validation in active run cache
- NodeVisitRef type in fabro-store types
- ListRunsQuery parameter for list_runs API
- HashSet dedup in catalog listing
- Updated snapshot tests for new run directory format

Preserved from feature branch:
- NodeAsset struct and exports
- StageId-based node references in run state
- make_run_dir as pub for cross-crate access
- Thread-spawn approach in handler test_default for tokio safety
- parse_run_id handles YYYYMMDD-ULID directory format

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-03 20:46:09 -07:00
commit 2943508dd1
No known key found for this signature in database
27 changed files with 357 additions and 403 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

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

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

@ -4,7 +4,7 @@ use std::sync::LazyLock;
use base64::Engine as _;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use fabro_config::server::{ApiAuthStrategy, AuthProvider, load_server_settings};
use fabro_config::server::{self, ApiAuthStrategy, AuthProvider};
use fabro_config::user::{default_user_config_path, legacy_user_config_path};
use fabro_llm::client::Client as LlmClient;
use fabro_llm::types::{Message, Request};
@ -938,7 +938,7 @@ pub(crate) async fn run_doctor(
let daytona_configured = std::env::var("DAYTONA_API_KEY").is_ok();
let server_settings = load_server_settings(None).unwrap_or_default();
let server_settings = server::load_server_settings(None).unwrap_or_default();
let api_status = {
let api = server_settings.api.clone().unwrap_or_default();

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

@ -303,7 +303,7 @@ digraph Test {
let run_dir = before_summary["run_dir"].as_str().unwrap().to_string();
fabro_json_snapshot!(context, &before_summary, @r#"
{
"run_dir": "[DRY_RUN_DIR]",
"run_dir": "[RUN_DIR]",
"start_time": "[TIMESTAMP]",
"conclusion_timestamp": "[TIMESTAMP]",
"conclusion_status": "success"

19
lib/crates/fabro-cli/tests/it/cmd/ps.rs
View file

@ -66,25 +66,28 @@ fn ps_all_json_lists_created_and_completed_runs() {
let mut cmd = context.ps();
cmd.args(["-a", "--json"]);
fabro_snapshot!(filters, cmd, @r###"
fabro_snapshot!(filters, cmd, @r#"
success: true
exit_code: 0
----- stdout -----
[
{
"run_id": "[ULID]",
"dir_name": "[DATE]-dry-run-[ULID]",
"dir_name": "20260403-[ULID]",
"workflow_name": "Simple",
"workflow_slug": "simple",
"status": "submitted",
"status_reason": null,
"start_time": "[TIMESTAMP]",
"labels": {},
"duration_ms": null,
"total_cost": null,
"host_repo_path": "[TEMP_DIR]",
"goal": "Run tests and report results"
},
{
"run_id": "[ULID]",
"dir_name": "[DATE]-dry-run-[ULID]",
"dir_name": "20260403-[ULID]",
"workflow_name": "Simple",
"workflow_slug": "simple",
"status": "succeeded",
@ -92,12 +95,13 @@ fn ps_all_json_lists_created_and_completed_runs() {
"start_time": "[TIMESTAMP]",
"labels": {},
"duration_ms": [DURATION_MS],
"total_cost": null,
"host_repo_path": "[TEMP_DIR]",
"goal": "Run tests and report results"
}
]
----- stderr -----
"###);
"#);
}
#[test]
@ -169,14 +173,14 @@ fn ps_filters_by_workflow_and_label() {
"suite=alpha",
]);
fabro_snapshot!(filters, cmd, @r###"
fabro_snapshot!(filters, cmd, @r#"
success: true
exit_code: 0
----- stdout -----
[
{
"run_id": "[ULID]",
"dir_name": "[DATE]-dry-run-[ULID]",
"dir_name": "20260403-[ULID]",
"workflow_name": "Simple",
"workflow_slug": "simple",
"status": "succeeded",
@ -186,10 +190,11 @@ fn ps_filters_by_workflow_and_label() {
"suite": "alpha"
},
"duration_ms": [DURATION_MS],
"total_cost": null,
"host_repo_path": "[TEMP_DIR]",
"goal": "Run tests and report results"
}
]
----- stderr -----
"###);
"#);
}

7
lib/crates/fabro-cli/tests/it/cmd/rewind.rs
View file

@ -69,7 +69,8 @@ fn rewind_list_prints_timeline_for_completed_git_run() {
----- stdout -----
----- stderr -----
@ Node Details
@1 step_one
@1 step_one
@2 step_two
");
}
@ -94,8 +95,8 @@ fn rewind_target_updates_metadata_and_resume_hint() {
exit_code: 0
----- stdout -----
----- stderr -----
Rewound metadata branch to @1 (start)
Warning: checkpoint @1 has no git_commit_sha; run branch not moved
Rewound metadata branch to @1 (step_one)
Rewound run branch fabro/run/[ULID] to [SHA]
To resume: fabro resume [RUN_PREFIX]
");

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

@ -68,7 +68,7 @@ fn dry_run_simple() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [DRY_RUN_DIR]
Run: [STORAGE_DIR]/runs/20260403-[ULID]
=== Output ===
[Simulated] Response for stage: report
@ -966,11 +966,11 @@ fn detach_creates_run_dir_with_detach_log() {
"detach_log_exists": run_dir.join("detach.log").exists(),
}),
@r#"
{
"run_dir": "[DRY_RUN_DIR]",
"launcher_log_exists": true,
"detach_log_exists": false
}
"#
{
"run_dir": "[RUN_DIR]",
"launcher_log_exists": true,
"detach_log_exists": false
}
"#
);
}

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

@ -48,9 +48,9 @@ fn system_df_summarizes_runs_and_logs() {
success: true
exit_code: 0
----- stdout -----
TYPE COUNT ACTIVE SIZE RECLAIMABLE
Runs 1 0 [SIZE] [SIZE] (0%)
Logs 1 - [SIZE] [SIZE] (100%)
TYPE COUNT ACTIVE SIZE RECLAIMABLE
Runs 1 0 [SIZE] [SIZE] (100%)
Logs 1 - [SIZE] [SIZE] (100%)
Data directory: [STORAGE_DIR]
----- stderr -----
@ -79,13 +79,13 @@ fn system_df_verbose_lists_runs_with_reclaimable_marker() {
success: true
exit_code: 0
----- stdout -----
TYPE COUNT ACTIVE SIZE RECLAIMABLE
Runs 1 0 [SIZE] [SIZE] (0%)
Logs 0 - [SIZE] [SIZE] (0%)
TYPE COUNT ACTIVE SIZE RECLAIMABLE
Runs 1 0 [SIZE] [SIZE] (100%)
Logs 0 - [SIZE] [SIZE] (0%)
Data directory: [STORAGE_DIR]
RUN ID WORKFLOW STATUS AGE SIZE
RUN ID WORKFLOW STATUS AGE SIZE
[RUN_PREFIX] Simple succeeded [AGE] [SIZE] *
* = reclaimable

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

@ -51,7 +51,7 @@ fn system_prune_dry_run_lists_matching_runs_without_deleting() {
success: true
exit_code: 0
----- stdout -----
would delete: [DATE]-dry-run-[ULID] (Simple)
would delete: 20260403-[ULID] (Simple)
----- stderr -----
1 run(s) would be deleted ([SIZE] freed). Pass --yes to confirm.

10
lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs
View file

@ -175,11 +175,11 @@ fn dry_run_detach_attach_works_with_default_run_lookup() {
"conclusion_json_exists": run_dir.join("conclusion.json").exists(),
}),
@r#"
{
"run_dir": "[DRY_RUN_DIR]",
"conclusion_json_exists": false
}
"#
{
"run_dir": "[RUN_DIR]",
"conclusion_json_exists": false
}
"#
);
}

10
lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs
View file

@ -30,7 +30,7 @@ fn dry_run_branching() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [DRY_RUN_DIR]
Run: [STORAGE_DIR]/runs/20260403-[ULID]
=== Output ===
[Simulated] Response for stage: validate
@ -62,7 +62,7 @@ fn dry_run_conditions() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [DRY_RUN_DIR]
Run: [STORAGE_DIR]/runs/20260403-[ULID]
=== Output ===
[Simulated] Response for stage: path_b
@ -95,7 +95,7 @@ fn dry_run_parallel() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [DRY_RUN_DIR]
Run: [STORAGE_DIR]/runs/20260403-[ULID]
=== Output ===
[Simulated] Response for stage: review
@ -128,7 +128,7 @@ fn dry_run_styled() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [DRY_RUN_DIR]
Run: [STORAGE_DIR]/runs/20260403-[ULID]
=== Output ===
[Simulated] Response for stage: critical_review
@ -159,6 +159,6 @@ fn dry_run_legacy_tool() {
Run: [ULID]
Status: SUCCESS
Duration: [DURATION]
Run: [DRY_RUN_DIR]
Run: [STORAGE_DIR]/runs/20260403-[ULID]
");
}

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

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

@ -14,12 +14,12 @@ pub use fabro_types::StageId;
pub use run_state::{NodeState, RunProjection};
pub use runtime::RuntimeState;
pub use slate::{NodeAsset, SlateRunStore, SlateStore};
pub use types::{EventEnvelope, EventPayload, RunSummary};
pub use types::{EventEnvelope, EventPayload, NodeVisitRef, RunSummary};
pub type StoreHandle = Arc<SlateStore>;
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub(crate) struct ListRunsQuery {
pub struct ListRunsQuery {
pub start: Option<DateTime<Utc>>,
pub end: Option<DateTime<Utc>>,
}

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

@ -1,5 +1,7 @@
use std::collections::HashSet;
use std::sync::Arc;
use bytes::Bytes;
use futures::TryStreamExt;
use object_store::ObjectStore;
use object_store::path::Path;
@ -13,13 +15,10 @@ pub(crate) async fn write_catalog(
run_id: &RunId,
) -> Result<()> {
store
.put(&by_id_path(base_prefix, run_id), bytes::Bytes::new().into())
.put(&by_id_path(base_prefix, run_id), Bytes::new().into())
.await?;
store
.put(
&by_start_path(base_prefix, run_id),
bytes::Bytes::new().into(),
)
.put(&by_start_path(base_prefix, run_id), Bytes::new().into())
.await?;
Ok(())
}
@ -44,11 +43,12 @@ pub(crate) async fn list_run_ids(
let prefix = Path::from(format!("{base_prefix}by-start"));
let metas = store.list(Some(&prefix)).try_collect::<Vec<_>>().await?;
let mut run_ids = Vec::new();
let mut seen = HashSet::new();
for meta in metas {
let Some(run_id) = parse_run_id_from_path(&meta.location) else {
continue;
};
if !read_locator(store.clone(), base_prefix, &run_id).await? {
if !seen.insert(run_id) {
continue;
}
let created_at = run_id.created_at();
@ -67,6 +67,12 @@ pub(crate) async fn list_run_ids(
Ok(run_ids)
}
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}/",
@ -85,16 +91,9 @@ pub(crate) fn by_start_path(base_prefix: &str, run_id: &RunId) -> Path {
))
}
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::HashSet;
pub(crate) async fn repair_catalog(
store: Arc<dyn ObjectStore>,
@ -107,17 +106,15 @@ pub(super) mod test_support {
.list(Some(&by_id_prefix))
.try_collect::<Vec<_>>()
.await?;
let mut canonical = HashSet::new();
for meta in by_id_metas {
if let Some(run_id) = parse_run_id_from_path(&meta.location) {
canonical.insert(run_id);
}
}
let run_ids = by_id_metas
.iter()
.filter_map(|meta| parse_run_id_from_path(&meta.location))
.collect::<Vec<_>>();
for run_id in &canonical {
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, bytes::Bytes::new().into()).await?;
store.put(&path, Bytes::new().into()).await?;
}
}
@ -125,6 +122,7 @@ 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();
@ -132,25 +130,18 @@ pub(super) mod test_support {
delete_if_exists(store.clone(), &location).await?;
continue;
};
if !canonical.contains(&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;
}
let expected = by_start_path(base_prefix, &run_id);
if expected == location {
seen.insert(run_id);
} else {
delete_if_exists(store.clone(), &location).await?;
}
delete_if_exists(store.clone(), &location).await?;
}
for run_id in &canonical {
if !seen.contains(run_id) {
for run_id in canonical {
if !seen.contains(&run_id) {
store
.put(
&by_start_path(base_prefix, run_id),
bytes::Bytes::new().into(),
)
.put(&by_start_path(base_prefix, &run_id), Bytes::new().into())
.await?;
}
}

404
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;
@ -19,12 +18,6 @@ use fabro_types::RunId;
use run_store::SlateRunStoreInner;
pub use run_store::{NodeAsset, SlateRunStore};
#[derive(Clone, Copy)]
enum RunOpenMode {
Writer,
Reader,
}
#[derive(Clone)]
pub struct SlateStore {
object_store: Arc<dyn ObjectStore>,
@ -110,54 +103,62 @@ impl SlateStore {
weak.upgrade().map(SlateRunStore::from_inner)
}
async fn open_existing_run(
async fn open_run_store(
&self,
run_id: RunId,
mode: RunOpenMode,
run_id: &RunId,
db_prefix: &str,
) -> Result<Option<SlateRunStore>> {
if matches!(mode, RunOpenMode::Writer) {
if let Some(active) = self.get_active_run(&run_id).await {
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 {run_id:?}"
)));
}
let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id);
if !self.db_prefix_has_objects(&db_prefix).await? {
if !self.db_prefix_has_objects(db_prefix).await? {
return Ok(None);
}
match mode {
RunOpenMode::Writer => {
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;
return Err(err);
}
};
if !has_init {
let _ = db.close().await;
return Ok(None);
}
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(&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;
return Err(err);
}
};
if !has_init {
let _ = reader.close().await;
return Ok(None);
}
SlateRunStore::open_reader(run_id, reader).await.map(Some)
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;
return Err(err);
}
};
if !has_init {
let _ = db.close().await;
return Ok(None);
}
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,
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(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;
return Err(err);
}
};
if !has_init {
let _ = reader.close().await;
return Ok(None);
}
SlateRunStore::open_reader(*run_id, db_prefix.to_string(), reader)
.await
.map(Some)
}
async fn delete_db_prefix(&self, db_prefix: &str) -> Result<()> {
@ -176,75 +177,77 @@ impl SlateStore {
impl SlateStore {
pub async fn create_run(&self, run_id: &RunId) -> Result<SlateRunStore> {
let locator =
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 locator {
if locator_exists && !active.matches_run(run_id, &db_prefix) {
return Err(StoreError::RunAlreadyExists(run_id.to_string()));
}
catalog::write_catalog(self.object_store.clone(), &self.base_prefix, run_id).await?;
return Ok(active);
}
let db_prefix = catalog::db_prefix(&self.base_prefix, run_id);
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, run_id).await?;
db.put(keys::init(), serde_json::to_vec(run_id)?).await?;
let run_store = SlateRunStore::open_writer(*run_id, db).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).await?;
Ok(run_store)
}
pub async fn open_run(&self, run_id: &RunId) -> Result<SlateRunStore> {
let locator =
let exists =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
if !locator {
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_existing_run(*run_id, RunOpenMode::Writer)
.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 =
let exists =
catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?;
if !locator {
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_existing_run(*run_id, RunOpenMode::Reader)
.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) -> Result<Vec<RunSummary>> {
self.list_runs_in_range(None, None).await
}
pub async fn list_runs_in_range(
&self,
start: Option<DateTime<Utc>>,
end: Option<DateTime<Utc>>,
) -> Result<Vec<RunSummary>> {
let query = ListRunsQuery { start, end };
pub async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>> {
let run_ids =
catalog::list_run_ids(self.object_store.clone(), &self.base_prefix, &query).await?;
catalog::list_run_ids(self.object_store.clone(), &self.base_prefix, query).await?;
let mut summaries = Vec::new();
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 {run_id:?}"
)));
}
let snapshot = active.snapshot().await?;
summaries.push(SlateRunStore::build_summary(snapshot.as_ref(), &run_id).await?);
continue;
}
let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id);
if !self.db_prefix_has_objects(&db_prefix).await? {
continue;
}
@ -268,64 +271,52 @@ impl SlateStore {
active.close().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? {
return self.delete_run_record(run_id).await;
delete_path(
self.object_store.clone(),
&catalog::by_start_path(&self.base_prefix, run_id),
)
.await?;
self.delete_db_prefix(&db_prefix).await?;
delete_path(
self.object_store.clone(),
&catalog::by_id_path(&self.base_prefix, run_id),
)
.await?;
return Ok(());
}
self.repair_delete_run(run_id).await
}
}
impl SlateStore {
async fn delete_run_record(&self, run_id: &RunId) -> Result<()> {
delete_path(
self.object_store.clone(),
&catalog::by_start_path(&self.base_prefix, run_id),
)
.await?;
self.delete_db_prefix(&catalog::db_prefix(&self.base_prefix, run_id))
if active.is_some() {
delete_path(
self.object_store.clone(),
&catalog::by_start_path(&self.base_prefix, run_id),
)
.await?;
delete_path(
self.object_store.clone(),
&catalog::by_id_path(&self.base_prefix, run_id),
)
.await?;
Ok(())
}
self.delete_db_prefix(&db_prefix).await?;
delete_path(
self.object_store.clone(),
&catalog::by_id_path(&self.base_prefix, run_id),
)
.await?;
return Ok(());
}
async fn repair_delete_run(&self, run_id: &RunId) -> Result<()> {
let by_start_prefix = Path::from(format!("{}by-start", self.base_prefix));
let db_prefix = Path::from(format!("{}db", self.base_prefix));
let expected_name = format!("{run_id}.json");
for meta in self
let metas = self
.object_store
.list(Some(&by_start_prefix))
.try_collect::<Vec<_>>()
.await?
{
if meta.location.filename() == Some(expected_name.as_str()) {
delete_path(self.object_store.clone(), &meta.location).await?;
.await?;
let expected_name = format!("{run_id}.json");
for meta in metas {
if meta.location.filename() != Some(expected_name.as_str()) {
continue;
}
delete_path(self.object_store.clone(), &meta.location).await?;
}
let db_run_segment = format!("/{run_id}/");
for meta in self
.object_store
.list(Some(&db_prefix))
.try_collect::<Vec<_>>()
.await?
{
if meta.location.to_string().contains(&db_run_segment) {
delete_path(self.object_store.clone(), &meta.location).await?;
}
}
delete_path(
self.object_store.clone(),
&catalog::by_id_path(&self.base_prefix, run_id),
)
.await?;
self.delete_db_prefix(&db_prefix).await?;
Ok(())
}
}
@ -357,11 +348,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;
@ -370,7 +361,7 @@ mod tests {
use crate::{EventPayload, StageId};
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
struct CatalogRecord {
run_id: RunId,
created_at: DateTime<Utc>,
@ -395,16 +386,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(),
@ -558,11 +550,6 @@ mod tests {
store.head(path).await.is_ok()
}
async fn read_json_value(store: Arc<dyn ObjectStore>, path: &Path) -> serde_json::Value {
let result = store.get(path).await.unwrap();
serde_json::from_slice(&result.bytes().await.unwrap()).unwrap()
}
async fn seed_db(
object_store: Arc<dyn ObjectStore>,
record: &CatalogRecord,
@ -640,14 +627,8 @@ mod tests {
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!(
read_json_value(object_store.clone(), &by_start).await,
serde_json::json!({
"run_id": test_run_id("run-1").to_string(),
})
);
let summary = store.list_runs().await.unwrap();
let summary = store.list_runs(&ListRunsQuery::default()).await.unwrap();
assert_eq!(summary.len(), 1);
assert_eq!(summary[0].run_id, test_run_id("run-1"));
assert_eq!(summary[0].workflow_name, Some("night-sky".to_string()));
@ -710,10 +691,16 @@ mod tests {
.unwrap();
assert!(store.open_run(&test_run_id("run-1")).await.is_ok());
assert!(store.list_runs().await.unwrap().is_empty());
assert!(
store
.list_runs(&ListRunsQuery::default())
.await
.unwrap()
.is_empty()
);
repair_catalog_for_tests(&store).await.unwrap();
let listed = store.list_runs().await.unwrap();
let listed = store.list_runs(&ListRunsQuery::default()).await.unwrap();
assert_eq!(listed.len(), 1);
assert!(
object_exists(
@ -724,65 +711,6 @@ mod tests {
);
}
#[tokio::test]
async fn list_runs_uses_by_start_index_to_resolve_canonical_by_id_record() {
let (object_store, store) = make_store();
let created_at = dt("2026-03-27T12:00:00Z");
let record = CatalogRecord {
run_id: test_run_id("run-1"),
created_at,
db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")),
run_dir: None,
};
let db = seed_db(object_store.clone(), &record, true).await;
db.put(
keys::event_key(1, created_at.timestamp_millis()),
serde_json::to_vec(&event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": sample_run_record("run-1", created_at).settings,
"graph": sample_run_record("run-1", created_at).graph,
"workflow_slug": sample_run_record("run-1", created_at).workflow_slug,
"working_directory": sample_run_record("run-1", created_at).working_directory,
"host_repo_path": sample_run_record("run-1", created_at).host_repo_path,
"base_branch": sample_run_record("run-1", created_at).base_branch,
"labels": sample_run_record("run-1", created_at).labels,
}),
))
.unwrap(),
)
.await
.unwrap();
db.close().await.unwrap();
object_store
.put(
&catalog::by_id_path("runs/", &test_run_id("run-1")),
serde_json::to_vec(&record).unwrap().into(),
)
.await
.unwrap();
object_store
.put(
&catalog::by_start_path("runs/", &test_run_id("run-1")),
serde_json::json!({
"run_id": test_run_id("run-1").to_string(),
})
.to_string()
.into(),
)
.await
.unwrap();
let listed = store.list_runs().await.unwrap();
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].run_id, record.run_id);
}
#[tokio::test]
async fn reopen_recovers_event_sequences() {
let (_object_store, store) = make_store();
@ -869,16 +797,29 @@ mod tests {
.unwrap();
assert!(store.open_run(&test_run_id("run-1")).await.is_err());
assert!(store.list_runs().await.unwrap().is_empty());
assert!(
store
.list_runs(&ListRunsQuery::default())
.await
.unwrap()
.is_empty()
);
}
#[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")).await.unwrap();
let _created_at = dt("2026-03-27T12:00:00Z");
// Hold the first run store so the active cache Weak ref stays alive.
let _run = 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("other-run")).await;
// Different run_id should work fine
assert!(conflict.is_ok());
// Dropping and re-creating with locator already written should reject
drop(_run);
let conflict = store.create_run(&test_run_id("run-1")).await;
assert!(matches!(conflict, Err(StoreError::RunAlreadyExists(_))));
}
@ -907,7 +848,7 @@ mod tests {
.await
.unwrap();
let listed = store.list_runs().await.unwrap();
let listed = store.list_runs(&ListRunsQuery::default()).await.unwrap();
assert_eq!(listed.len(), 1);
let reopened = store.open_run(&test_run_id("run-1")).await.unwrap();
@ -938,7 +879,7 @@ 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 _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();
@ -1012,7 +953,7 @@ 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 _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
@ -1035,20 +976,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 _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();
object_store
.put(
&catalog::by_start_path("runs/", &test_run_id("run-1")),
bytes::Bytes::new().into(),
Bytes::new().into(),
)
.await
.unwrap();
@ -1063,6 +1004,7 @@ 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,
@ -1087,7 +1029,7 @@ mod tests {
#[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 _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 {
@ -1097,15 +1039,12 @@ 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")).await else {
@ -1120,7 +1059,7 @@ 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 _created_at = dt("2026-03-27T12:00:00Z");
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let node = StageId::new("code", 2);
run.append_event(&event_payload(
@ -1148,7 +1087,7 @@ 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 _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
@ -1434,6 +1373,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.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 {}"));
@ -1845,15 +1788,12 @@ mod tests {
let (_object_store, store) = make_store();
let run = store.create_run(&test_run_id("run-1")).await.unwrap();
let invalid_missing_err = serde_json::from_value::<EventPayload>(serde_json::json!({
let invalid_missing: EventPayload = serde_json::from_value(serde_json::json!({
"run_id": "run-1"
}))
.unwrap_err();
assert!(
invalid_missing_err
.to_string()
.contains("missing or non-string required field: id")
);
.unwrap();
let err = run.append_event(&invalid_missing).await.unwrap_err();
assert!(matches!(err, StoreError::InvalidEvent(_)));
let invalid_run_id: EventPayload = serde_json::from_value(serde_json::json!({
"id": "evt-invalid-run",
@ -1981,7 +1921,7 @@ mod tests {
.await
.unwrap();
let all = store.list_runs().await.unwrap();
let all = store.list_runs(&ListRunsQuery::default()).await.unwrap();
assert_eq!(all.len(), 2);
assert_eq!(all[0].run_id, test_run_id("run-late"));
assert_eq!(all[0].workflow_name, Some("night-sky".to_string()));
@ -1996,10 +1936,10 @@ mod tests {
assert_eq!(all[1].status, None);
let filtered = store
.list_runs_in_range(
Some(dt("2026-03-27T11:00:00Z")),
Some(dt("2026-03-27T13:00:00Z")),
)
.list_runs(&ListRunsQuery {
start: Some(dt("2026-03-27T11:00:00Z")),
end: Some(dt("2026-03-27T13:00:00Z")),
})
.await
.unwrap();
assert_eq!(filtered.len(), 1);

43
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;
@ -16,10 +16,6 @@ use crate::keys;
use crate::run_state::EventProjectionCache;
use crate::{EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId, StoreError};
use fabro_types::RunId;
#[derive(Clone)]
pub struct SlateRunStore {
inner: Arc<SlateRunStoreInner>,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct NodeAsset {
@ -27,16 +23,23 @@ pub struct NodeAsset {
pub filename: String,
}
#[derive(Clone)]
pub struct SlateRunStore {
inner: Arc<SlateRunStoreInner>,
}
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("db_prefix", &self.inner.db_prefix)
.finish_non_exhaustive()
}
}
pub(crate) struct SlateRunStoreInner {
run_id: RunId,
db_prefix: String,
db: SlateRunDb,
event_seq: AtomicU32,
close_lock: Mutex<()>,
@ -49,11 +52,16 @@ enum SlateRunDb {
}
impl SlateRunStore {
pub(crate) async fn open_writer(run_id: RunId, 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,
db_prefix,
db: SlateRunDb::Writer(db),
event_seq: AtomicU32::new(event_seq),
close_lock: Mutex::new(()),
@ -62,11 +70,16 @@ impl SlateRunStore {
})
}
pub(crate) async fn open_reader(run_id: RunId, 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,
db_prefix,
db: SlateRunDb::Reader(Box::new(db)),
event_seq: AtomicU32::new(event_seq),
close_lock: Mutex::new(()),
@ -83,12 +96,12 @@ impl SlateRunStore {
Arc::downgrade(&self.inner)
}
pub fn run_id(&self) -> RunId {
pub(crate) fn run_id(&self) -> RunId {
self.inner.run_id
}
pub fn created_at(&self) -> DateTime<Utc> {
self.inner.run_id.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<()> {
@ -114,7 +127,7 @@ impl SlateRunStore {
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 run id {expected:?}"
"existing _init.json {existing:?} does not match requested run_id {expected:?}"
))),
None => Ok(false),
}
@ -146,13 +159,7 @@ impl SlateRunStore {
impl SlateRunStore {
pub async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
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.run_id
)));
}
payload.validate(&self.inner.run_id)?;
let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst);
self.inner
.db

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

@ -1,12 +1,17 @@
use std::collections::HashMap;
use chrono::{DateTime, Utc};
use serde::de::Error as _;
use serde::{Deserialize, Deserializer, Serialize};
use serde::{Deserialize, Serialize};
use crate::{Result, StoreError};
use fabro_types::{RunId, RunStatus, StatusReason};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct NodeVisitRef<'a> {
pub node_id: &'a str,
pub visit: u32,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSummary {
pub run_id: RunId,
@ -22,7 +27,7 @@ pub struct RunSummary {
pub total_cost: Option<f64>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct EventPayload(serde_json::Value);
@ -34,17 +39,6 @@ impl EventPayload {
}
pub(crate) fn validate(&self, expected_run_id: &RunId) -> Result<()> {
self.validate_shape()?;
if self.run_id() == expected_run_id.to_string() {
return Ok(());
}
Err(StoreError::InvalidEvent(format!(
"payload run_id {:?} does not match store run_id {expected_run_id:?}",
self.run_id()
)))
}
fn validate_shape(&self) -> Result<()> {
let obj = self.0.as_object().ok_or_else(|| {
StoreError::InvalidEvent("event payload must be a JSON object".into())
})?;
@ -59,14 +53,22 @@ impl EventPayload {
}
}
}
Ok(())
match obj.get("run_id") {
Some(serde_json::Value::String(run_id)) if run_id == &expected_run_id.to_string() => {
Ok(())
}
Some(serde_json::Value::String(run_id)) => Err(StoreError::InvalidEvent(format!(
"payload run_id {run_id:?} does not match store run_id {expected_run_id:?}"
))),
_ => Err(StoreError::InvalidEvent(
"missing or non-string required field: run_id".into(),
)),
}
}
#[must_use]
pub fn run_id(&self) -> &str {
self.0["run_id"]
.as_str()
.expect("EventPayload::run_id called before validation")
pub fn into_inner(self) -> serde_json::Value {
self.0
}
pub fn as_value(&self) -> &serde_json::Value {
@ -74,17 +76,6 @@ impl EventPayload {
}
}
impl<'de> Deserialize<'de> for EventPayload {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let payload = Self(serde_json::Value::deserialize(deserializer)?);
payload.validate_shape().map_err(D::Error::custom)?;
Ok(payload)
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct EventEnvelope {
pub seq: u32,

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

@ -167,7 +167,7 @@ pub mod fixtures {
#[cfg(test)]
mod tests {
use chrono::{DateTime, Utc};
use chrono::{TimeZone, Utc};
use super::{RunId, fixtures};
@ -186,21 +186,18 @@ mod tests {
}
#[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);
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(), expected);
assert_eq!(run_id.created_at(), dt);
}
#[test]
fn from_datetime_round_trips_timestamp() {
let expected = DateTime::parse_from_rfc3339("2026-03-27T12:00:00Z")
.unwrap()
.with_timezone(&Utc);
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!(RunId::from_datetime(expected).created_at(), expected);
assert_eq!(run_id.created_at(), dt);
}
}

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

@ -352,13 +352,14 @@ pub(crate) fn default_run_dir(run_id: &RunId) -> PathBuf {
}
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))
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;
@ -426,6 +427,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 {
@ -674,6 +693,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());
}
@ -724,7 +744,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());

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

@ -182,7 +182,10 @@ 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? {
for summary in fabro_store
.list_runs(&fabro_store::ListRunsQuery::default())
.await?
{
if summary.run_id.to_string() == prefix {
if summary.host_repo_path.is_none() {
return Ok(summary.run_id);

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

@ -1092,6 +1092,7 @@ mod tests {
async fn resume_errors_when_run_already_finished_successfully() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
let registry = Arc::new(test_registry());

1
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;

1
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};

1
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;

1
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};

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

@ -207,12 +207,12 @@ fn parse_dot_summary(dot: &str) -> (String, usize, usize) {
fn read_plan_text(state: &RunProjection) -> Option<String> {
let mut plan_nodes = state
.iter_nodes()
.filter_map(|(node, node_state)| {
let node_id = node.node_id();
let visit = node.visit();
node_id
.starts_with("plan")
.then_some((node_id, visit, node_state.response.as_deref()))
.filter_map(|(stage_id, node)| {
stage_id.node_id().starts_with("plan").then_some((
stage_id.node_id(),
stage_id.visit(),
node.response.as_deref(),
))
})
.collect::<Vec<_>>();
plan_nodes.sort_by(|left, right| left.0.cmp(right.0).then(left.1.cmp(&right.1)));
@ -1057,7 +1057,6 @@ 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).await.unwrap();
let conclusion = make_test_conclusion();
@ -1081,9 +1080,7 @@ 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).await.unwrap();
let run_record = RunRecord {
@ -1106,7 +1103,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(),
@ -1149,9 +1146,7 @@ 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).await.unwrap();
let run_record = RunRecord {
@ -1174,7 +1169,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(),
@ -1344,7 +1339,6 @@ 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).await.unwrap();
let creds = GitHubAppCredentials {
@ -1373,7 +1367,6 @@ 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).await.unwrap();
let run_record = RunRecord {
run_id: fixtures::RUN_1,
@ -1395,7 +1388,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,

2
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,7 +215,6 @@ 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()).await.unwrap();
let run_store = inner;
let run_record = RunRecord {

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

@ -193,9 +193,12 @@ fn scan_orphan_runs(base: &Path) -> Result<Vec<RunInfo>> {
pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<RunInfo>> {
let mut runs_by_id: HashMap<RunId, RunInfo> = HashMap::new();
if let Ok(store_runs) = store.list_runs().await {
if let Ok(store_runs) = store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
for summary in store_runs {
let Some(run_info) = run_info_from_summary(summary, base) else {
let Some(run_info) = run_info_from_summary(&summary, base) else {
continue;
};
runs_by_id.insert(run_info.run_id(), run_info);
@ -218,7 +221,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<R
Ok(runs)
}
fn run_info_from_summary(summary: RunSummary, runs_base: &Path) -> Option<RunInfo> {
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;
@ -234,7 +237,7 @@ fn run_info_from_summary(summary: RunSummary, runs_base: &Path) -> Option<RunInf
};
Some(RunInfo::new(
Some(summary),
Some(summary.clone()),
RunLocalState {
dir_name,
start_time_dt: Some(start_time_dt),
@ -345,7 +348,15 @@ fn collapse_separators(s: &str) -> String {
fn parse_run_id(value: &str) -> Option<RunId> {
let value = value.trim();
(!value.is_empty()).then_some(value)?.parse().ok()
if value.is_empty() {
return None;
}
// Try direct ULID parse first, then try extracting ULID after date prefix (YYYYMMDD-ULID).
value.parse().ok().or_else(|| {
value
.split_once('-')
.and_then(|(_, ulid)| ulid.parse().ok())
})
}
fn run_id_matches(run_id: RunId, prefix: &str) -> bool {
@ -359,7 +370,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};
@ -367,6 +377,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 {
@ -393,12 +404,11 @@ 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).await.unwrap();
append_workflow_event(
@ -411,7 +421,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(),