scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Remove InMemoryStore, RunSnapshot, and NodeSnapshot; use RunState directly

Browse source 
InMemoryStore duplicated SlateStore's interface and was unused in
production. RunSnapshot/NodeSnapshot were intermediate projections that
tests consumed — replaced with RunState to eliminate the indirection.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-04-03 14:59:24 -07:00
parent 9595cbb6c0
commit ffee4688eb
13 changed files with 1034 additions and 1705 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -6,7 +6,7 @@ use fabro_test::{fabro_snapshot, test_context};
use fabro_types::Settings;
use predicates::prelude::*;
use super::support::run_snapshot;
use super::support::run_state;
#[test]
fn help() {
@ -444,8 +444,9 @@ fn create_explicit_workflow_path_uses_project_config_relative_to_workflow() {
)
});
let snapshot = run_snapshot(&run_dir);
let run_record = serde_json::to_value(&snapshot.run).unwrap();
let state = run_state(&run_dir);
let run_record =
serde_json::to_value(state.run.as_ref().expect("run record should exist")).unwrap();
assert_eq!(run_record["settings"]["auto_approve"].as_bool(), Some(true));
assert_eq!(
run_record["settings"]["storage_dir"].as_str(),

54
lib/crates/fabro-cli/tests/it/cmd/create.rs
View file

@ -5,7 +5,7 @@ use fabro_test::{fabro_snapshot, test_context};
use crate::support::fabro_json_snapshot;
use super::support::{fixture, output_stdout, resolve_run, run_snapshot};
use super::support::{fixture, output_stdout, resolve_run, run_state};
#[test]
fn help() {
@ -78,13 +78,14 @@ digraph BarBaz {
.success();
let run_dir = context.find_run_dir(run_id);
let snapshot = run_snapshot(&run_dir);
let state = run_state(&run_dir);
let run = state.run.as_ref().expect("run record should exist");
fabro_json_snapshot!(
context,
serde_json::json!({
"workflow_slug": snapshot.run.workflow_slug,
"graph_name": snapshot.run.graph.name,
"cached_graph_lines": snapshot.graph.expect("graph should exist").lines().collect::<Vec<_>>(),
"workflow_slug": run.workflow_slug,
"graph_name": run.graph.name,
"cached_graph_lines": state.graph_source.as_ref().expect("graph should exist").lines().collect::<Vec<_>>(),
}),
@r#"
{
@ -133,13 +134,14 @@ digraph FooWorkflow {
.success();
let run_dir = context.find_run_dir(run_id);
let snapshot = run_snapshot(&run_dir);
let state = run_state(&run_dir);
let run = state.run.as_ref().expect("run record should exist");
fabro_json_snapshot!(
context,
serde_json::json!({
"workflow_slug": snapshot.run.workflow_slug,
"graph_name": snapshot.run.graph.name,
"cached_graph_lines": snapshot.graph.expect("graph should exist").lines().collect::<Vec<_>>(),
"workflow_slug": run.workflow_slug,
"graph_name": run.graph.name,
"cached_graph_lines": state.graph_source.as_ref().expect("graph should exist").lines().collect::<Vec<_>>(),
}),
@r#"
{
@ -199,26 +201,27 @@ fn create_persists_requested_overrides_into_store() {
.expect("create should print a run ID")
.to_string();
let run = resolve_run(&context, &run_id);
let snapshot = run_snapshot(&run.run_dir);
let state = run_state(&run.run_dir);
let run_record = state.run.as_ref().expect("run record should exist");
let labels = json!({
"env": snapshot.run.labels.get("env"),
"team": snapshot.run.labels.get("team"),
"env": run_record.labels.get("env"),
"team": run_record.labels.get("team"),
});
let compact = json!({
"workflow_slug": snapshot.run.workflow_slug,
"workflow_slug": run_record.workflow_slug,
"settings": {
"goal": snapshot.run.settings.goal,
"dry_run": snapshot.run.settings.dry_run,
"auto_approve": snapshot.run.settings.auto_approve,
"no_retro": snapshot.run.settings.no_retro,
"verbose": snapshot.run.settings.verbose,
"goal": run_record.settings.goal,
"dry_run": run_record.settings.dry_run,
"auto_approve": run_record.settings.auto_approve,
"no_retro": run_record.settings.no_retro,
"verbose": run_record.settings.verbose,
"llm": {
"model": snapshot.run.settings.llm.as_ref().and_then(|llm| llm.model.clone()),
"provider": snapshot.run.settings.llm.as_ref().and_then(|llm| llm.provider.clone()),
"model": run_record.settings.llm.as_ref().and_then(|llm| llm.model.clone()),
"provider": run_record.settings.llm.as_ref().and_then(|llm| llm.provider.clone()),
},
"sandbox": {
"provider": snapshot.run.settings.sandbox.as_ref().and_then(|sandbox| sandbox.provider.clone()),
"preserve": snapshot.run.settings.sandbox.as_ref().and_then(|sandbox| sandbox.preserve),
"provider": run_record.settings.sandbox.as_ref().and_then(|sandbox| sandbox.provider.clone()),
"preserve": run_record.settings.sandbox.as_ref().and_then(|sandbox| sandbox.preserve),
},
},
"labels": labels,
@ -275,7 +278,12 @@ fn create_json_implies_auto_approve() {
let run = resolve_run(&context, run_id);
assert_eq!(
run_snapshot(&run.run_dir).run.settings.auto_approve,
run_state(&run.run_dir)
.run
.as_ref()
.expect("run record should exist")
.settings
.auto_approve,
Some(true)
);
}

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

@ -1,6 +1,6 @@
use fabro_test::{fabro_snapshot, test_context};
use super::support::run_snapshot;
use super::support::run_state;
use crate::support::fabro_json_snapshot;
#[test]
@ -89,7 +89,7 @@ digraph CachedGraph {
.success();
let conclusion = serde_json::to_value(
run_snapshot(&run_dir)
run_state(&run_dir)
.conclusion
.expect("conclusion should exist"),
)
@ -147,11 +147,12 @@ digraph GitHubApp {
.success();
let run_dir = context.find_run_dir(run_id);
let snapshot = run_snapshot(&run_dir);
let state = run_state(&run_dir);
let run = state.run.as_ref().expect("run record should exist");
fabro_json_snapshot!(
context,
serde_json::json!({
"app_id": snapshot.run.settings.git.and_then(|git| git.app_id),
"app_id": run.settings.git.clone().and_then(|git| git.app_id),
}),
@r#"
{
@ -230,7 +231,7 @@ digraph DetachedStoreOnly {
.success();
let conclusion = serde_json::to_value(
run_snapshot(&run_dir)
run_state(&run_dir)
.conclusion
.expect("conclusion should exist"),
)

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

@ -4,7 +4,7 @@ use fabro_test::{fabro_snapshot, run_and_format, test_context};
use super::support::{
git_filters, git_stdout, output_stderr as support_stderr, run_branch_commits_since_base,
run_events, run_snapshot, setup_git_backed_changed_run,
run_events, run_state, setup_git_backed_changed_run,
};
#[test]
@ -176,25 +176,22 @@ fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() {
"run.submitted"
);
let snapshot = run_snapshot(&setup.run.run_dir);
let state = run_state(&setup.run.run_dir);
assert_eq!(
snapshot.status.as_ref().map(|status| &status.status),
state.status.as_ref().map(|status| &status.status),
Some(&fabro_types::RunStatus::Submitted)
);
assert!(state.conclusion.is_none(), "rewind should clear conclusion");
assert!(
snapshot.conclusion.is_none(),
"rewind should clear conclusion"
);
assert!(
snapshot.final_patch.is_none(),
state.final_patch.is_none(),
"rewind should clear final patch"
);
assert!(
snapshot.pull_request.is_none(),
state.pull_request.is_none(),
"rewind should clear pull request"
);
assert!(
snapshot.nodes.is_empty(),
"rewind should clear node snapshots that belonged to the prior execution"
state.nodes.is_empty(),
"rewind should clear node state that belonged to the prior execution"
);
}

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

@ -4,7 +4,7 @@ use std::process::Output;
use std::sync::Arc;
use std::time::{Duration, Instant};
use fabro_store::{EventEnvelope, RunSnapshot, RunStoreHandle, SlateStore};
use fabro_store::{EventEnvelope, RunState, RunStoreHandle, SlateStore};
use fabro_test::TestContext;
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
@ -492,12 +492,9 @@ fn run_store(run_dir: &Path) -> RunStoreHandle {
block_on(store.open_run_reader(&run_id)).expect("run store should exist")
}
pub(crate) fn run_snapshot(run_dir: &Path) -> RunSnapshot {
pub(crate) fn run_state(run_dir: &Path) -> RunState {
let store = run_store(run_dir);
block_on(store.state())
.ok()
.and_then(|state| state.to_snapshot())
.expect("run store snapshot should exist")
block_on(store.state()).expect("run store state should exist")
}
pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
@ -768,7 +765,7 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git
let run = only_run(context);
let start = serde_json::to_value(
run_snapshot(&run.run_dir)
run_state(&run.run_dir)
.start
.expect("start record should exist"),
)
@ -781,26 +778,26 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git
match workflow {
GitWorkflowKind::Changed => {
assert!(
run_snapshot(&run.run_dir).final_patch.is_some(),
run_state(&run.run_dir).final_patch.is_some(),
"changed git-backed run should persist final patch in store"
);
let snapshot = run_snapshot(&run.run_dir);
let state = run_state(&run.run_dir);
assert!(
snapshot
state
.nodes
.iter()
.any(|node| node.node_id == "step_one" && node.diff.is_some())
.any(|((node_id, _), node)| node_id == "step_one" && node.diff.is_some())
);
assert!(
snapshot
state
.nodes
.iter()
.any(|node| node.node_id == "step_two" && node.diff.is_some())
.any(|((node_id, _), node)| node_id == "step_two" && node.diff.is_some())
);
}
GitWorkflowKind::Noop => {
assert!(
run_snapshot(&run.run_dir).final_patch.is_none(),
run_state(&run.run_dir).final_patch.is_none(),
"no-op git-backed run should not persist final.patch"
);
}

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

@ -1,7 +1,7 @@
use fabro_test::test_context;
use serde_json::Value;
use super::{fixture, run_snapshot, timeout_for};
use super::{fixture, run_state, timeout_for};
use crate::support::{example_fixture, fabro_json_snapshot};
#[fabro_macros::e2e_test()]
@ -238,7 +238,9 @@ digraph BarBaz {
.assert()
.success();
let run_record = run_snapshot(&context.find_run_dir(run_id)).run;
let run_record = run_state(&context.find_run_dir(run_id))
.run
.expect("run record should exist");
fabro_json_snapshot!(
context,
serde_json::json!({
@ -297,7 +299,9 @@ digraph FooWorkflow {
.assert()
.success();
let run_record = run_snapshot(&context.find_run_dir(run_id)).run;
let run_record = run_state(&context.find_run_dir(run_id))
.run
.expect("run record should exist");
fabro_json_snapshot!(
context,
serde_json::json!({

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

@ -7,7 +7,7 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use fabro_store::{RunSnapshot, RunStoreHandle, SlateStore};
use fabro_store::{RunState, RunStoreHandle, SlateStore};
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
pub(super) fn fixture(name: &str) -> PathBuf {
@ -47,12 +47,9 @@ fn run_store(run_dir: &Path) -> RunStoreHandle {
block_on(store.open_run_reader(&run_id)).expect("run store should exist")
}
pub(super) fn run_snapshot(run_dir: &Path) -> RunSnapshot {
pub(super) fn run_state(run_dir: &Path) -> RunState {
let store = run_store(run_dir);
block_on(store.state())
.ok()
.and_then(|state| state.to_snapshot())
.expect("run store snapshot should exist")
block_on(store.state()).expect("run store state should exist")
}
pub(super) fn timeout_for(sandbox: &str) -> Duration {

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

@ -4,20 +4,16 @@ use chrono::{DateTime, Utc};
mod error;
mod keys;
mod memory;
mod run_state;
mod runtime;
mod slate;
mod types;
pub use error::{Result, StoreError};
pub use memory::{InMemoryRunStore, InMemoryStore};
pub use run_state::{NodeState, RunState};
pub use runtime::RuntimeState;
pub use slate::{SlateRunStore, SlateStore};
pub use types::{
CatalogRecord, EventEnvelope, EventPayload, NodeSnapshot, NodeVisitRef, RunSnapshot, RunSummary,
};
pub use types::{CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, RunSummary};
use fabro_types::{Outcome, StageUsage};

1528
lib/crates/fabro-store/src/memory.rs
View file

File diff suppressed because it is too large Load diff

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

@ -7,8 +7,7 @@ use serde::de::DeserializeOwned;
use serde_json::Value;
use crate::{
CatalogRecord, EventEnvelope, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, Result,
RunSnapshot, RunSummary, StoreError,
CatalogRecord, EventEnvelope, NodeOutcomeRecord, NodeVisitRef, Result, RunSummary, StoreError,
};
use fabro_types::{
Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro,
@ -308,48 +307,6 @@ impl RunState {
visits
}
pub fn to_snapshot(&self) -> Option<RunSnapshot> {
let run = self.run.clone()?;
let mut node_keys = self.nodes.keys().cloned().collect::<Vec<_>>();
node_keys.sort();
let nodes = node_keys
.into_iter()
.filter_map(|(node_id, visit)| {
self.nodes
.get(&(node_id.clone(), visit))
.map(|node| NodeSnapshot {
node_id,
visit,
prompt: node.prompt.clone(),
response: node.response.clone(),
status: node.status.clone(),
outcome: node.outcome.clone(),
provider_used: node.provider_used.clone(),
diff: node.diff.clone(),
script_invocation: node.script_invocation.clone(),
script_timing: node.script_timing.clone(),
parallel_results: node.parallel_results.clone(),
stdout: node.stdout.clone(),
stderr: node.stderr.clone(),
})
})
.collect();
Some(RunSnapshot {
run,
start: self.start.clone(),
status: self.status.clone(),
checkpoint: self.checkpoint.clone(),
conclusion: self.conclusion.clone(),
retro: self.retro.clone(),
graph: self.graph_source.clone(),
sandbox: self.sandbox.clone(),
final_patch: self.final_patch.clone(),
pull_request: self.pull_request.clone(),
nodes,
})
}
pub fn build_summary(&self, catalog: &CatalogRecord) -> RunSummary {
let workflow_name = self.run.as_ref().map(|run| {
if run.graph.name.is_empty() {

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

@ -378,12 +378,16 @@ pub(crate) fn normalize_base_prefix(prefix: String) -> String {
mod tests {
use super::*;
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::Duration;
use bytes::Bytes;
use chrono::Duration as ChronoDuration;
use fabro_types::{
AttrValue, Graph, RunId, RunRecord, RunStatus, Settings, StatusReason, fixtures,
AttrValue, Checkpoint, Conclusion, Graph, PullRequestRecord, Retro, RunId, RunRecord,
RunStatus, RunStatusRecord, SandboxRecord, Settings, StageStatus, StartRecord,
StatusReason, fixtures,
};
use object_store::memory::InMemory;
use slatedb::config::Settings as SlateSettings;
@ -408,6 +412,8 @@ mod tests {
match label {
"run-1" => fixtures::RUN_1,
"other-run" => fixtures::RUN_2,
"run-early" => fixtures::RUN_2,
"run-late" => fixtures::RUN_3,
_ => panic!("unknown test run id: {label}"),
}
}
@ -451,6 +457,107 @@ mod tests {
EventPayload::new(value, &test_run_id(run_id)).unwrap()
}
fn sample_start_record(run_id: &str, created_at: DateTime<Utc>) -> StartRecord {
StartRecord {
run_id: test_run_id(run_id),
start_time: created_at + ChronoDuration::seconds(5),
run_branch: Some("fabro/run/demo".to_string()),
base_sha: Some("abc123".to_string()),
}
}
fn sample_status(status: RunStatus, reason: Option<StatusReason>) -> RunStatusRecord {
RunStatusRecord {
status,
reason,
updated_at: dt("2026-03-27T12:05:00Z"),
}
}
fn sample_checkpoint() -> Checkpoint {
Checkpoint {
timestamp: dt("2026-03-27T12:10:00Z"),
current_node: "code".to_string(),
completed_nodes: vec!["plan".to_string()],
node_retries: HashMap::from([("code".to_string(), 1)]),
context_values: HashMap::from([(
"artifact".to_string(),
serde_json::json!({"kind": "summary"}),
)]),
node_outcomes: HashMap::new(),
next_node_id: Some("review".to_string()),
git_commit_sha: Some("def456".to_string()),
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::from([("code".to_string(), 2)]),
}
}
fn sample_conclusion() -> Conclusion {
Conclusion {
timestamp: dt("2026-03-27T12:15:00Z"),
status: StageStatus::Success,
duration_ms: 3210,
failure_reason: None,
final_git_commit_sha: Some("feedbeef".to_string()),
stages: Vec::new(),
total_cost: Some(1.25),
total_retries: 2,
total_input_tokens: 10,
total_output_tokens: 20,
total_cache_read_tokens: 30,
total_cache_write_tokens: 40,
total_reasoning_tokens: 50,
has_pricing: true,
}
}
fn sample_retro(run_id: &str) -> Retro {
Retro {
run_id: test_run_id(run_id),
workflow_name: "night-sky".to_string(),
goal: "map the constellations".to_string(),
timestamp: dt("2026-03-27T12:20:00Z"),
smoothness: None,
stages: Vec::new(),
stats: fabro_types::AggregateStats {
total_duration_ms: 3210,
total_cost: Some(1.25),
total_retries: 2,
files_touched: vec!["src/lib.rs".to_string()],
stages_completed: 3,
stages_failed: 0,
},
intent: Some("ship the fix".to_string()),
outcome: Some("done".to_string()),
learnings: None,
friction_points: None,
open_items: None,
}
}
fn sample_sandbox() -> SandboxRecord {
SandboxRecord {
provider: "local".to_string(),
working_directory: "/tmp/night-sky".to_string(),
identifier: Some("sandbox-1".to_string()),
host_working_directory: Some("/tmp/night-sky".to_string()),
container_mount_point: None,
}
}
fn sample_pull_request() -> PullRequestRecord {
PullRequestRecord {
html_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(),
number: 123,
owner: "fabro-sh".to_string(),
repo: "fabro".to_string(),
base_branch: "main".to_string(),
head_branch: "fabro/run/demo".to_string(),
title: "Map the constellations".to_string(),
}
}
async fn list_paths(store: Arc<dyn ObjectStore>, prefix: &str) -> Vec<String> {
let mut items = store
.list(Some(&Path::from(prefix.to_string())))
@ -1092,4 +1199,855 @@ mod tests {
]
);
}
#[tokio::test]
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_record = sample_run_record("run-1", created_at);
let start_record = sample_start_record("run-1", created_at);
let status_record =
sample_status(RunStatus::Running, Some(StatusReason::SandboxInitializing));
let checkpoint = sample_checkpoint();
let conclusion = sample_conclusion();
let retro = sample_retro("run-1");
let sandbox = sample_sandbox();
let node = NodeVisitRef {
node_id: "code",
visit: 2,
};
let pull_request = sample_pull_request();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": run_record.settings,
"graph": run_record.graph,
"workflow_source": "digraph night_sky {}",
"workflow_slug": run_record.workflow_slug,
"working_directory": run_record.working_directory,
"host_repo_path": run_record.host_repo_path,
"base_branch": run_record.base_branch,
"labels": run_record.labels,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"run.started",
None,
serde_json::json!({
"run_branch": start_record.run_branch,
"base_sha": start_record.base_sha,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:06Z",
"run.running",
None,
serde_json::json!({
"reason": status_record.reason,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:07Z",
"checkpoint.completed",
Some("code"),
serde_json::json!({
"status": "success",
"current_node": checkpoint.current_node,
"completed_nodes": checkpoint.completed_nodes,
"node_retries": checkpoint.node_retries,
"context_values": checkpoint.context_values,
"node_outcomes": checkpoint.node_outcomes,
"next_node_id": checkpoint.next_node_id,
"git_commit_sha": checkpoint.git_commit_sha,
"loop_failure_signatures": serde_json::json!({}),
"restart_failure_signatures": serde_json::json!({}),
"node_visits": checkpoint.node_visits,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08Z",
"sandbox.initialized",
None,
serde_json::json!({
"provider": sandbox.provider,
"working_directory": sandbox.working_directory,
"identifier": sandbox.identifier,
"host_working_directory": sandbox.host_working_directory,
"container_mount_point": sandbox.container_mount_point,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08.1Z",
"stage.prompt",
Some("code"),
serde_json::json!({
"visit": 2,
"text": "Plan the fix",
"mode": "prompt",
"provider": "openai",
"model": "gpt-5.4"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08.2Z",
"command.started",
Some("code"),
serde_json::json!({
"visit": 2,
"command": "cargo test"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08.3Z",
"command.completed",
Some("code"),
serde_json::json!({
"visit": 2,
"stdout": "ok",
"stderr": "",
"exit_code": 0
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08.4Z",
"checkpoint.completed",
Some("code"),
serde_json::json!({
"status": "success",
"ordinal": 2,
"current_node": checkpoint.current_node,
"completed_nodes": checkpoint.completed_nodes,
"node_retries": checkpoint.node_retries,
"context_values": checkpoint.context_values,
"node_outcomes": checkpoint.node_outcomes,
"next_node_id": checkpoint.next_node_id,
"git_commit_sha": checkpoint.git_commit_sha,
"node_visits": checkpoint.node_visits,
"diff": "diff --git a/src/lib.rs b/src/lib.rs"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08.5Z",
"parallel.completed",
Some("code"),
serde_json::json!({
"visit": 2,
"results": [{"node_id": "lint", "status": "success"}]
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08.6Z",
"stage.completed",
Some("code"),
serde_json::json!({
"visit": 2,
"status": "success",
"notes": "all good",
"response": "Implemented",
"files_touched": ["src/lib.rs"]
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08.7Z",
"retro.started",
None,
serde_json::json!({
"prompt": "How did it go?"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:09Z",
"retro.completed",
None,
serde_json::json!({
"response": "Smooth enough",
"retro": retro,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:10Z",
"run.completed",
None,
serde_json::json!({
"status": conclusion.status,
"duration_ms": conclusion.duration_ms,
"total_cost": conclusion.total_cost,
"final_git_commit_sha": conclusion.final_git_commit_sha,
"final_patch": "diff --git a/src/lib.rs b/src/lib.rs\n",
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:11Z",
"pull_request.created",
None,
serde_json::json!({
"pr_url": pull_request.html_url,
"pr_number": pull_request.number,
"owner": pull_request.owner,
"repo": pull_request.repo,
"base_branch": pull_request.base_branch,
"head_branch": pull_request.head_branch,
"title": pull_request.title,
}),
))
.await
.unwrap();
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
.await
.unwrap();
run.put_asset(&node, "src/lib.rs", b"fn main() {}")
.await
.unwrap();
let state = run.state().await.unwrap();
let stored_run = state.run.as_ref().unwrap();
assert_eq!(stored_run.run_id, run_record.run_id);
assert_eq!(stored_run.created_at, run_record.created_at);
assert_eq!(stored_run.workflow_slug, run_record.workflow_slug);
assert_eq!(stored_run.graph.name, run_record.graph.name);
assert_eq!(state.graph_source.as_deref(), Some("digraph night_sky {}"));
let stored_start = state.start.as_ref().unwrap();
assert_eq!(stored_start.run_id, start_record.run_id);
assert_eq!(stored_start.start_time, start_record.start_time);
let stored_status = state.status.as_ref().unwrap();
assert_eq!(stored_status.status, RunStatus::Succeeded);
assert_eq!(stored_status.reason, None);
let stored_checkpoint = state.checkpoint.as_ref().unwrap();
assert_eq!(stored_checkpoint.current_node, checkpoint.current_node);
assert_eq!(stored_checkpoint.next_node_id, checkpoint.next_node_id);
let stored_conclusion = state.conclusion.as_ref().unwrap();
assert_eq!(stored_conclusion.status, conclusion.status);
assert_eq!(stored_conclusion.duration_ms, conclusion.duration_ms);
assert_eq!(stored_conclusion.total_cost, conclusion.total_cost);
let stored_retro = state.retro.as_ref().unwrap();
assert_eq!(stored_retro.run_id, retro.run_id);
assert_eq!(stored_retro.intent, retro.intent);
let stored_sandbox = state.sandbox.as_ref().unwrap();
assert_eq!(stored_sandbox.provider, sandbox.provider);
assert_eq!(stored_sandbox.working_directory, sandbox.working_directory);
assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?"));
assert_eq!(state.retro_response.as_deref(), Some("Smooth enough"));
assert_eq!(
run.get_artifact_value("summary").await.unwrap(),
Some(serde_json::json!({"done": true}))
);
assert_eq!(
run.get_asset(&node, "src/lib.rs").await.unwrap(),
Some(Bytes::from_static(b"fn main() {}"))
);
assert_eq!(
state.final_patch.as_deref(),
Some("diff --git a/src/lib.rs b/src/lib.rs\n")
);
assert_eq!(state.pull_request, Some(pull_request.clone()));
assert_eq!(state.list_node_ids(), vec!["code".to_string()]);
let node_state = state
.node(&node)
.expect("node state should exist for code:2");
assert_eq!(node_state.prompt.as_deref(), Some("Plan the fix"));
assert_eq!(node_state.response.as_deref(), Some("Implemented"));
assert_eq!(node_state.stdout.as_deref(), Some("ok"));
assert_eq!(node_state.stderr.as_deref(), Some(""));
assert_eq!(
node_state.diff.as_deref(),
Some("diff --git a/src/lib.rs b/src/lib.rs")
);
assert_eq!(
node_state
.provider_used
.as_ref()
.and_then(|v| v.get("provider"))
.and_then(|v| v.as_str()),
Some("openai")
);
assert_eq!(
run.list_assets(&node).await.unwrap(),
vec!["src/lib.rs".to_string()]
);
}
#[tokio::test]
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_record = sample_run_record("run-1", created_at);
let retro = sample_retro("run-1");
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": run_record.settings,
"graph": run_record.graph,
"workflow_source": "digraph night_sky {}",
"workflow_slug": run_record.workflow_slug,
"working_directory": run_record.working_directory,
"host_repo_path": run_record.host_repo_path,
"base_branch": run_record.base_branch,
"labels": run_record.labels,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"run.started",
None,
serde_json::json!({
"run_branch": "fabro/run/demo",
"base_sha": "abc123"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:06Z",
"run.running",
None,
serde_json::json!({}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:07Z",
"stage.prompt",
Some("code"),
serde_json::json!({
"visit": 2,
"text": "Plan the fix",
"mode": "prompt",
"provider": "openai",
"model": "gpt-5.4"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08Z",
"stage.completed",
Some("code"),
serde_json::json!({
"status": "success",
"notes": "all good",
"response": "Implemented",
"files_touched": ["src/lib.rs"],
"node_visits": {"code": 2}
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:09Z",
"checkpoint.completed",
Some("code"),
serde_json::json!({
"status": "success",
"current_node": "code",
"completed_nodes": ["plan"],
"context_values": {"artifact": {"kind": "summary"}},
"next_node_id": "review",
"git_commit_sha": "def456",
"node_visits": {"code": 2},
"diff": "diff --git a/src/lib.rs b/src/lib.rs"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:10Z",
"sandbox.initialized",
None,
serde_json::json!({
"provider": "local",
"working_directory": "/tmp/night-sky",
"identifier": "sandbox-1",
"host_working_directory": "/tmp/night-sky"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:11Z",
"retro.started",
None,
serde_json::json!({
"prompt": "How did it go?"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:12Z",
"retro.completed",
None,
serde_json::json!({
"response": "Smooth enough",
"retro": retro
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:13Z",
"pull_request.created",
None,
serde_json::json!({
"pr_url": "https://github.com/fabro-sh/fabro/pull/123",
"pr_number": 123,
"owner": "fabro-sh",
"repo": "fabro",
"base_branch": "main",
"head_branch": "fabro/run/demo",
"title": "Map the constellations",
"draft": false
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:15Z",
"run.completed",
None,
serde_json::json!({
"duration_ms": 3210,
"artifact_count": 1,
"status": "success",
"total_cost": 1.25,
"final_git_commit_sha": "feedbeef",
"final_patch": "diff --git a/src/lib.rs b/src/lib.rs\n"
}),
))
.await
.unwrap();
let state = run.state().await.unwrap();
assert_eq!(
state.run.as_ref().map(|run| run.run_id),
Some(test_run_id("run-1"))
);
assert_eq!(state.graph_source.as_deref(), Some("digraph night_sky {}"));
assert_eq!(
state
.start
.as_ref()
.and_then(|start| start.run_branch.as_deref()),
Some("fabro/run/demo")
);
assert_eq!(
state.status.as_ref().map(|status| status.status),
Some(RunStatus::Succeeded)
);
assert_eq!(
state
.checkpoint
.as_ref()
.map(|checkpoint| checkpoint.current_node.as_str()),
Some("code")
);
assert_eq!(state.checkpoints.len(), 1);
assert_eq!(
state.final_patch.as_deref(),
Some("diff --git a/src/lib.rs b/src/lib.rs\n")
);
assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?"));
assert_eq!(state.retro_response.as_deref(), Some("Smooth enough"));
assert_eq!(state.pull_request.as_ref().map(|pr| pr.number), Some(123));
assert_eq!(
state
.sandbox
.as_ref()
.map(|sandbox| sandbox.provider.as_str()),
Some("local")
);
assert_eq!(state.list_node_visits("code"), vec![2]);
let node = state
.node(&NodeVisitRef {
node_id: "code",
visit: 2,
})
.unwrap();
assert_eq!(node.prompt.as_deref(), Some("Plan the fix"));
assert_eq!(node.response.as_deref(), Some("Implemented"));
assert_eq!(
node.diff.as_deref(),
Some("diff --git a/src/lib.rs b/src/lib.rs")
);
assert_eq!(
node.provider_used
.as_ref()
.and_then(|value| value.get("provider"))
.and_then(|value| value.as_str()),
Some("openai")
);
}
#[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();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": Settings::default(),
"graph": Graph::new("night-sky"),
"working_directory": "/tmp/night-sky",
"labels": {}
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:01Z",
"stage.prompt",
Some("code"),
serde_json::json!({
"visit": 1,
"text": "before rewind"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:02Z",
"pull_request.created",
None,
serde_json::json!({
"pr_url": "https://github.com/fabro-sh/fabro/pull/123",
"pr_number": 123,
"owner": "fabro-sh",
"repo": "fabro",
"base_branch": "main",
"head_branch": "fabro/run/demo",
"title": "Map the constellations",
"draft": false
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:03Z",
"run.completed",
None,
serde_json::json!({
"duration_ms": 10,
"artifact_count": 0,
"status": "success",
"final_patch": "old patch"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:04Z",
"run.rewound",
None,
serde_json::json!({
"target_checkpoint_ordinal": 1,
"target_node_id": "plan",
"target_visit": 1
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"checkpoint.completed",
Some("plan"),
serde_json::json!({
"status": "success",
"current_node": "plan",
"completed_nodes": [],
"node_visits": {"plan": 1}
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:06Z",
"run.submitted",
None,
serde_json::json!({}),
))
.await
.unwrap();
let state = run.state().await.unwrap();
assert_eq!(
state.status.as_ref().map(|status| status.status),
Some(RunStatus::Submitted)
);
assert!(state.conclusion.is_none());
assert!(state.final_patch.is_none());
assert!(state.pull_request.is_none());
assert_eq!(state.checkpoints.len(), 1);
assert_eq!(
state
.checkpoint
.as_ref()
.map(|checkpoint| checkpoint.current_node.as_str()),
Some("plan")
);
assert!(state.list_node_ids().is_empty());
}
#[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 invalid_missing: EventPayload = serde_json::from_value(serde_json::json!({
"run_id": "run-1"
}))
.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",
"ts": "2026-03-27T12:00:00Z",
"run_id": "other-run",
"event": "StageStarted"
}))
.unwrap();
let err = run.append_event(&invalid_run_id).await.unwrap_err();
assert!(matches!(err, StoreError::InvalidEvent(_)));
}
#[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 checkpoint = sample_checkpoint();
let seq = run
.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"checkpoint.completed",
Some(&checkpoint.current_node),
serde_json::json!({
"status": "success",
"current_node": checkpoint.current_node,
"completed_nodes": checkpoint.completed_nodes,
"node_retries": checkpoint.node_retries,
"context_values": checkpoint.context_values,
"node_outcomes": checkpoint.node_outcomes,
"next_node_id": checkpoint.next_node_id,
"git_commit_sha": checkpoint.git_commit_sha,
"loop_failure_signatures": serde_json::json!({}),
"restart_failure_signatures": serde_json::json!({}),
"node_visits": checkpoint.node_visits,
}),
))
.await
.unwrap();
let state = run.state().await.unwrap();
assert_eq!(seq, 1);
assert_eq!(state.checkpoints.len(), 1);
assert_eq!(state.checkpoints[0].0, 1);
assert_eq!(state.checkpoints[0].1.current_node, checkpoint.current_node);
assert_eq!(
state.checkpoint.as_ref().unwrap().current_node,
checkpoint.current_node
);
}
#[tokio::test]
async fn list_runs_filters_dates_and_tolerates_missing_status() {
let (_object_store, store) = make_store();
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_record = sample_run_record("run-early", early);
early_run
.append_event(&event_payload(
"run-early",
"2026-03-27T10:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": early_record.settings,
"graph": early_record.graph,
"workflow_slug": early_record.workflow_slug,
"working_directory": early_record.working_directory,
"host_repo_path": early_record.host_repo_path,
"base_branch": early_record.base_branch,
"labels": early_record.labels,
}),
))
.await
.unwrap();
let late_run = store
.create_run(&test_run_id("run-late"), late, None)
.await
.unwrap();
let late_record = sample_run_record("run-late", late);
late_run
.append_event(&event_payload(
"run-late",
"2026-03-27T12:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": late_record.settings,
"graph": late_record.graph,
"workflow_slug": late_record.workflow_slug,
"working_directory": late_record.working_directory,
"host_repo_path": late_record.host_repo_path,
"base_branch": late_record.base_branch,
"labels": late_record.labels,
}),
))
.await
.unwrap();
late_run
.append_event(&event_payload(
"run-late",
"2026-03-27T12:00:01Z",
"run.started",
None,
serde_json::json!({
"run_branch": "fabro/run/demo",
"base_sha": "abc123",
}),
))
.await
.unwrap();
late_run
.append_event(&event_payload(
"run-late",
"2026-03-27T12:00:02Z",
"run.completed",
None,
serde_json::json!({
"duration_ms": 3210,
"artifact_count": 1,
"status": "success",
"reason": "completed",
"total_cost": 1.25,
}),
))
.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()));
assert_eq!(all[0].goal, Some("map the constellations".to_string()));
assert_eq!(
all[0].host_repo_path,
Some("github.com/fabro-sh/fabro".to_string())
);
assert_eq!(all[0].duration_ms, Some(3210));
assert_eq!(all[0].total_cost, Some(1.25));
assert_eq!(all[0].status_reason, Some(StatusReason::Completed));
assert_eq!(all[1].status, None);
let filtered = store
.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);
assert_eq!(filtered[0].run_id, test_run_id("run-late"));
}
}

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

@ -3,11 +3,8 @@ use std::collections::HashMap;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::{NodeOutcomeRecord, Result, StoreError};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord,
RunStatus, RunStatusRecord, SandboxRecord, StartRecord, StatusReason,
};
use crate::{Result, StoreError};
use fabro_types::{RunId, RunStatus, StatusReason};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct NodeVisitRef<'a> {
@ -41,38 +38,6 @@ pub struct RunSummary {
pub total_cost: Option<f64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunSnapshot {
pub run: RunRecord,
pub start: Option<StartRecord>,
pub status: Option<RunStatusRecord>,
pub checkpoint: Option<Checkpoint>,
pub conclusion: Option<Conclusion>,
pub retro: Option<Retro>,
pub graph: Option<String>,
pub sandbox: Option<SandboxRecord>,
pub final_patch: Option<String>,
pub pull_request: Option<PullRequestRecord>,
pub nodes: Vec<NodeSnapshot>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeSnapshot {
pub node_id: String,
pub visit: u32,
pub prompt: Option<String>,
pub response: Option<String>,
pub status: Option<NodeStatusRecord>,
pub outcome: Option<NodeOutcomeRecord>,
pub provider_used: Option<serde_json::Value>,
pub diff: Option<String>,
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub stdout: Option<String>,
pub stderr: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct EventPayload(serde_json::Value);

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

@ -11,7 +11,6 @@ use object_store::memory::InMemory;
use crate::error::Result;
use crate::event::{EventEmitter, WorkflowRunEvent, append_workflow_event};
use crate::git::scan_node_files_from_state;
use crate::handler::HandlerRegistry;
use crate::outcome::Outcome;
use crate::pipeline;
@ -185,30 +184,7 @@ pub async fn run_graph_from_checkpoint(
executed.outcome
}
async fn persist_run_artifacts_for_tests(run_store: &SlateRunStore, run_dir: &std::path::Path) {
let state: fabro_store::RunState = match run_store.state().await {
Ok(state) => state,
Err(_) => return,
};
if let Some(checkpoint) = state.checkpoint.as_ref() {
if let Ok(json) = serde_json::to_string_pretty(checkpoint) {
let _ = std::fs::write(run_dir.join("checkpoint.json"), json);
}
}
if let Some(final_patch) = state.final_patch.as_ref() {
let _ = std::fs::write(run_dir.join("final.patch"), final_patch);
}
for (relative_path, contents) in scan_node_files_from_state(&state) {
let path = run_dir.join(relative_path);
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
let _ = std::fs::write(path, contents);
}
}
async fn persist_run_artifacts_for_tests(_run_store: &SlateRunStore, _run_dir: &std::path::Path) {}
pub struct WorkflowRunner {
registry: std::sync::Mutex<Option<HandlerRegistry>>,