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

refactor: introduce StageId

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-03 18:00:29 -07:00
parent a56a990ed4
commit 7efb66cd3e
No known key found for this signature in database
20 changed files with 262 additions and 243 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -59,7 +59,7 @@ async fn resolve_diff(
let state = run_store.state().await?;
if let Some(ref node_id) = args.node {
if let Some(visit) = state.list_node_visits(node_id).into_iter().max() {
if let Some(node) = state.node(&fabro_store::NodeVisitRef { node_id, visit }) {
if let Some(node) = state.node(&fabro_store::StageId::new(node_id, visit)) {
if let Some(patch) = node.diff.clone() {
debug!(node_id, visit, "Reading per-node diff from projected state");
return Ok(patch);

6
lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs
View file

@ -557,6 +557,8 @@ mod tests {
emit(
&mut ui,
WorkflowRunEvent::ParallelStarted {
node_id: "fork1".into(),
visit: 1,
branch_count: 2,
join_policy: "wait_all".into(),
},
@ -603,6 +605,8 @@ mod tests {
emit(
&mut ui,
WorkflowRunEvent::ParallelStarted {
node_id: "fork1".into(),
visit: 1,
branch_count: 1,
join_policy: "wait_all".into(),
},
@ -1110,6 +1114,8 @@ mod tests {
emit(
&mut ui,
WorkflowRunEvent::ParallelStarted {
node_id: "fork1".into(),
visit: 1,
branch_count: 1,
join_policy: "wait_all".into(),
},

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

@ -3,7 +3,7 @@ use std::path::Path;
use anyhow::{Context, Result};
#[cfg(test)]
use fabro_store::NodeVisitRef;
use fabro_store::StageId;
use fabro_store::{RunProjection, SlateRunStore};
use fabro_workflow::run_dump::RunDump;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
@ -294,10 +294,7 @@ mod tests {
let retro = sample_retro(run_id);
let sandbox = sample_sandbox();
let node = NodeVisitRef {
node_id: "code",
visit: 2,
};
let node = StageId::new("code", 2);
append_workflow_event(
&run,
&run_id,
@ -534,10 +531,7 @@ mod tests {
.await
.unwrap();
let asset_only_node = NodeVisitRef {
node_id: "artifact-only",
visit: 7,
};
let asset_only_node = StageId::new("artifact-only", 7);
run.put_asset(&asset_only_node, "logs/output.txt", b"hello")
.await
.unwrap();
@ -660,16 +654,9 @@ mod tests {
)
.await
.unwrap();
run.put_asset(
&NodeVisitRef {
node_id: "code",
visit: 1,
},
"../escape.txt",
b"boom",
)
.await
.unwrap();
run.put_asset(&StageId::new("code", 1), "../escape.txt", b"boom")
.await
.unwrap();
let temp = tempfile::tempdir().unwrap();
let output = temp.path().join("dump");

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

@ -785,12 +785,12 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git
assert!(
state
.iter_nodes()
.any(|(node, state)| node.node_id == "step_one" && state.diff.is_some())
.any(|(node, state)| node.node_id() == "step_one" && state.diff.is_some())
);
assert!(
state
.iter_nodes()
.any(|(node, state)| node.node_id == "step_two" && state.diff.is_some())
.any(|(node, state)| node.node_id() == "step_two" && state.diff.is_some())
);
}
GitWorkflowKind::Noop => {

25
lib/crates/fabro-store/src/keys.rs
View file

@ -1,4 +1,4 @@
use crate::NodeVisitRef;
use crate::StageId;
pub(crate) const INIT_KEY: &str = "_init.json";
pub(crate) const EVENTS_PREFIX: &str = "events#";
@ -17,14 +17,15 @@ pub(crate) fn artifact_value(artifact_id: &str) -> String {
format!("{ARTIFACT_VALUES_PREFIX}{artifact_id}.json")
}
pub(crate) fn node_asset_prefix(node: &NodeVisitRef<'_>) -> String {
pub(crate) fn node_asset_prefix(node: &StageId) -> String {
format!(
"{ARTIFACT_NODES_PREFIX}{}#visit-{}",
node.node_id, node.visit
node.node_id(),
node.visit()
)
}
pub(crate) fn node_asset(node: &NodeVisitRef<'_>, filename: &str) -> String {
pub(crate) fn node_asset(node: &StageId, filename: &str) -> String {
format!("{}#{filename}", node_asset_prefix(node))
}
@ -38,7 +39,7 @@ pub(crate) fn parse_artifact_value_id(key: &str) -> Option<String> {
.map(ToString::to_string)
}
pub(crate) fn parse_node_asset_key(key: &str) -> Option<(String, u32, String)> {
pub(crate) fn parse_node_asset_key(key: &str) -> Option<(StageId, String)> {
parse_visit_scoped_key(key, ARTIFACT_NODES_PREFIX)
}
@ -46,11 +47,11 @@ fn parse_seq(key: &str, prefix: &str) -> Option<u32> {
key.strip_prefix(prefix)?.split_once('-')?.0.parse().ok()
}
fn parse_visit_scoped_key(key: &str, prefix: &str) -> Option<(String, u32, String)> {
fn parse_visit_scoped_key(key: &str, prefix: &str) -> Option<(StageId, String)> {
let rest = key.strip_prefix(prefix)?;
let (node_id, rest) = rest.split_once("#visit-")?;
let (visit, file) = rest.split_once('#')?;
Some((node_id.to_string(), visit.parse().ok()?, file.to_string()))
Some((StageId::new(node_id, visit.parse().ok()?), file.to_string()))
}
#[cfg(test)]
@ -70,10 +71,7 @@ mod tests {
#[test]
fn artifact_keys_match_spec() {
let node = NodeVisitRef {
node_id: "code",
visit: 2,
};
let node = StageId::new("code", 2);
assert_eq!(artifact_value("summary"), "artifacts#values#summary.json");
assert_eq!(
node_asset(&node, "src/main.rs"),
@ -90,7 +88,7 @@ mod tests {
);
assert_eq!(
parse_node_asset_key("artifacts#nodes#code#visit-2#src/main.rs"),
Some(("code".to_string(), 2, "src/main.rs".to_string()))
Some((StageId::new("code", 2), "src/main.rs".to_string()))
);
}
@ -112,8 +110,7 @@ mod tests {
assert_eq!(
parse_node_asset_key("artifacts#nodes#build#visit-1#deep/nested/path/file.rs"),
Some((
"build".to_string(),
1,
StageId::new("build", 1),
"deep/nested/path/file.rs".to_string()
))
);

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

@ -10,10 +10,11 @@ mod slate;
mod types;
pub use error::{Result, StoreError};
pub use fabro_types::StageId;
pub use run_state::{NodeState, RunProjection};
pub use runtime::RuntimeState;
pub use slate::{NodeAsset, SlateRunStore, SlateStore};
pub use types::{CatalogRecord, EventEnvelope, EventPayload, NodeVisit, NodeVisitRef, RunSummary};
pub use types::{CatalogRecord, EventEnvelope, EventPayload, RunSummary};
pub type StoreHandle = Arc<SlateStore>;

35
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, NodeVisit, NodeVisitRef, Result, RunSummary, StoreError,
};
use crate::{CatalogRecord, EventEnvelope, Result, RunSummary, StageId, StoreError};
use fabro_types::{
Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro,
RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StageUsage,
@ -30,7 +28,7 @@ pub struct RunProjection {
pub sandbox: Option<SandboxRecord>,
pub final_patch: Option<String>,
pub pull_request: Option<PullRequestRecord>,
nodes: HashMap<NodeVisit, NodeState>,
nodes: HashMap<StageId, NodeState>,
}
#[derive(Debug, Clone, Default)]
@ -255,30 +253,28 @@ impl RunProjection {
Ok(())
}
pub fn node(&self, node: &NodeVisitRef<'_>) -> Option<&NodeState> {
self.nodes.get(&node.into_owned())
pub fn node(&self, node: &StageId) -> Option<&NodeState> {
self.nodes.get(node)
}
pub fn iter_nodes(&self) -> impl Iterator<Item = (NodeVisitRef<'_>, &NodeState)> {
self.nodes
.iter()
.map(|(node, state)| (node.as_ref(), state))
pub fn iter_nodes(&self) -> impl Iterator<Item = (&StageId, &NodeState)> {
self.nodes.iter()
}
pub fn is_empty(&self) -> bool {
self.nodes.is_empty()
}
pub fn set_node(&mut self, node: NodeVisitRef<'_>, state: NodeState) {
self.nodes.insert(node.into_owned(), state);
pub fn set_node(&mut self, node: StageId, state: NodeState) {
self.nodes.insert(node, state);
}
pub fn list_node_visits(&self, node_id: &str) -> Vec<u32> {
let mut visits = self
.nodes
.keys()
.filter(|node| node.node_id == node_id)
.map(|node| node.visit)
.filter(|node| node.node_id() == node_id)
.map(StageId::visit)
.collect::<Vec<_>>();
visits.sort_unstable();
visits.dedup();
@ -323,19 +319,14 @@ impl RunProjection {
}
fn node_mut(&mut self, node_id: &str, visit: u32) -> &mut NodeState {
self.nodes
.entry(NodeVisit {
node_id: node_id.to_string(),
visit,
})
.or_default()
self.nodes.entry(StageId::new(node_id, visit)).or_default()
}
fn current_visit_for(&self, node_id: &str) -> Option<u32> {
self.nodes
.keys()
.filter(|node| node.node_id == node_id)
.map(|node| node.visit)
.filter(|node| node.node_id() == node_id)
.map(StageId::visit)
.max()
}

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

@ -419,7 +419,7 @@ mod tests {
use slatedb::{CloseReason, ErrorKind};
use tokio::time::timeout;
use crate::{EventPayload, NodeVisitRef};
use crate::{EventPayload, StageId};
fn dt(rfc3339: &str) -> DateTime<Utc> {
DateTime::parse_from_rfc3339(rfc3339)
@ -1209,10 +1209,7 @@ mod tests {
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let node = NodeVisitRef {
node_id: "code",
visit: 2,
};
let node = StageId::new("code", 2);
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:01:00Z",
@ -1250,18 +1247,12 @@ mod tests {
.await
.unwrap();
let snapshot_node = NodeVisitRef {
node_id: "code",
visit: 2,
};
let snapshot_node = StageId::new("code", 2);
run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}")
.await
.unwrap();
let asset_only_node = NodeVisitRef {
node_id: "artifact-only",
visit: 7,
};
let asset_only_node = StageId::new("artifact-only", 7);
run.put_asset(&asset_only_node, "logs/output.txt", b"hello")
.await
.unwrap();
@ -1274,17 +1265,11 @@ mod tests {
run.list_all_assets().await.unwrap(),
vec![
crate::slate::NodeAsset {
node: crate::NodeVisit {
node_id: "artifact-only".to_string(),
visit: 7,
},
node: crate::StageId::new("artifact-only", 7),
filename: "logs/output.txt".to_string(),
},
crate::slate::NodeAsset {
node: crate::NodeVisit {
node_id: "code".to_string(),
visit: 2,
},
node: crate::StageId::new("code", 2),
filename: "src/lib.rs".to_string(),
}
]
@ -1308,10 +1293,7 @@ mod tests {
let conclusion = sample_conclusion();
let retro = sample_retro("run-1");
let sandbox = sample_sandbox();
let node = NodeVisitRef {
node_id: "code",
visit: 2,
};
let node = StageId::new("code", 2);
let pull_request = sample_pull_request();
run.append_event(&event_payload(
@ -1588,7 +1570,7 @@ mod tests {
Some("diff --git a/src/lib.rs b/src/lib.rs\n")
);
assert_eq!(state.pull_request, Some(pull_request.clone()));
assert!(state.iter_nodes().any(|(node, _)| node.node_id == "code"));
assert!(state.iter_nodes().any(|(node, _)| node.node_id() == "code"));
let node_state = state
.node(&node)
.expect("node state should exist for code:2");
@ -1820,12 +1802,7 @@ mod tests {
Some("local")
);
assert_eq!(state.list_node_visits("code"), vec![2]);
let node = state
.node(&NodeVisitRef {
node_id: "code",
visit: 2,
})
.unwrap();
let node = state.node(&StageId::new("code", 2)).unwrap();
assert_eq!(node.prompt.as_deref(), Some("Plan the fix"));
assert_eq!(node.response.as_deref(), Some("Implemented"));
assert_eq!(

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

@ -15,8 +15,8 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::run_state::EventProjectionCache;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, NodeVisit, NodeVisitRef, Result, RunProjection,
RunSummary, StoreError,
CatalogRecord, EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId,
StoreError,
};
#[derive(Clone)]
pub struct SlateRunStore {
@ -25,7 +25,7 @@ pub struct SlateRunStore {
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct NodeAsset {
pub node: NodeVisit,
pub node: StageId,
pub filename: String,
}
@ -230,23 +230,14 @@ impl SlateRunStore {
self.inner.db.list_artifact_values().await
}
pub async fn put_asset(
&self,
node: &NodeVisitRef<'_>,
filename: &str,
data: &[u8],
) -> Result<()> {
pub async fn put_asset(&self, node: &StageId, filename: &str, data: &[u8]) -> Result<()> {
self.inner
.db
.put_bytes(&keys::node_asset(node, filename), data)
.await
}
pub async fn get_asset(
&self,
node: &NodeVisitRef<'_>,
filename: &str,
) -> Result<Option<Bytes>> {
pub async fn get_asset(&self, node: &StageId, filename: &str) -> Result<Option<Bytes>> {
self.inner
.db
.get_bytes(&keys::node_asset(node, filename))
@ -414,13 +405,10 @@ where
let mut assets = Vec::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
let Some((node_id, visit, filename)) = keys::parse_node_asset_key(&key) else {
let Some((node, filename)) = keys::parse_node_asset_key(&key) else {
continue;
};
assets.push(NodeAsset {
node: NodeVisit { node_id, visit },
filename,
});
assets.push(NodeAsset { node, filename });
}
assets.sort();
Ok(assets)

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

@ -1,43 +1,12 @@
use std::collections::HashMap;
use chrono::{DateTime, Utc};
use serde::de::Error as _;
use serde::{Deserialize, Deserializer, 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,
}
impl<'a> NodeVisitRef<'a> {
#[must_use]
pub fn into_owned(self) -> NodeVisit {
NodeVisit {
node_id: self.node_id.to_string(),
visit: self.visit,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct NodeVisit {
pub node_id: String,
pub visit: u32,
}
impl NodeVisit {
#[must_use]
pub fn as_ref(&self) -> NodeVisitRef<'_> {
NodeVisitRef {
node_id: &self.node_id,
visit: self.visit,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CatalogRecord {
pub run_id: RunId,
@ -120,7 +89,7 @@ impl<'de> Deserialize<'de> for EventPayload {
D: Deserializer<'de>,
{
let payload = Self(serde_json::Value::deserialize(deserializer)?);
payload.validate_shape().map_err(serde::de::Error::custom)?;
payload.validate_shape().map_err(D::Error::custom)?;
Ok(payload)
}
}

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

@ -13,6 +13,7 @@ pub mod run;
pub mod run_id;
pub mod sandbox_record;
pub mod settings;
pub mod stage_id;
pub mod start;
pub mod status;
pub mod usage;
@ -33,6 +34,7 @@ pub use run_id::RunId;
pub use run_id::fixtures;
pub use sandbox_record::SandboxRecord;
pub use settings::Settings;
pub use stage_id::StageId;
pub use start::StartRecord;
pub use status::{
InvalidTransition, ParseRunStatusError, RunStatus, RunStatusRecord, StatusReason,

154
lib/crates/fabro-types/src/stage_id.rs Normal file
View file

@ -0,0 +1,154 @@
use std::fmt;
use std::str::FromStr;
use serde::de::Error as _;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct StageId {
node_id: String,
visit: u32,
}
impl StageId {
#[must_use]
pub fn new(node_id: impl Into<String>, visit: u32) -> Self {
Self {
node_id: node_id.into(),
visit,
}
}
#[must_use]
pub fn node_id(&self) -> &str {
&self.node_id
}
#[must_use]
pub fn visit(&self) -> u32 {
self.visit
}
}
impl fmt::Display for StageId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}@{}", self.node_id, self.visit)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParseStageIdError(String);
impl fmt::Display for ParseStageIdError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl std::error::Error for ParseStageIdError {}
impl FromStr for StageId {
type Err = ParseStageIdError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let (node_id, visit) = s
.rsplit_once('@')
.ok_or_else(|| ParseStageIdError("stage id must contain '@'".to_string()))?;
if node_id.is_empty() {
return Err(ParseStageIdError(
"stage id node_id must not be empty".to_string(),
));
}
if visit.is_empty() {
return Err(ParseStageIdError(
"stage id visit suffix must not be empty".to_string(),
));
}
let visit = visit
.parse()
.map_err(|err| ParseStageIdError(format!("invalid stage id visit: {err}")))?;
Ok(Self::new(node_id, visit))
}
}
impl Serialize for StageId {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(&self.to_string())
}
}
impl<'de> Deserialize<'de> for StageId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
value.parse().map_err(D::Error::custom)
}
}
#[cfg(test)]
mod tests {
use super::StageId;
#[test]
fn display_and_parse_round_trip() {
let stage = StageId::new("code", 2);
assert_eq!(stage.to_string(), "code@2");
assert_eq!("code@2".parse::<StageId>().unwrap(), stage);
}
#[test]
fn ordering_is_node_id_then_visit() {
let mut stages = vec![
StageId::new("code", 2),
StageId::new("build", 1),
StageId::new("code", 1),
];
stages.sort();
assert_eq!(
stages,
vec![
StageId::new("build", 1),
StageId::new("code", 1),
StageId::new("code", 2),
]
);
}
#[test]
fn serde_round_trip_uses_string_form() {
let stage = StageId::new("code", 2);
let value = serde_json::to_value(&stage).unwrap();
assert_eq!(value, serde_json::json!("code@2"));
let decoded: StageId = serde_json::from_value(value).unwrap();
assert_eq!(decoded, stage);
}
#[test]
fn parse_rejects_missing_at_sign() {
let err = "code".parse::<StageId>().unwrap_err();
assert_eq!(err.to_string(), "stage id must contain '@'");
}
#[test]
fn parse_rejects_empty_suffix() {
let err = "code@".parse::<StageId>().unwrap_err();
assert_eq!(err.to_string(), "stage id visit suffix must not be empty");
}
#[test]
fn parse_rejects_non_numeric_visit() {
let err = "code@two".parse::<StageId>().unwrap_err();
assert!(err.to_string().starts_with("invalid stage id visit:"));
}
#[test]
fn parse_rejects_empty_node_id() {
let err = "@3".parse::<StageId>().unwrap_err();
assert_eq!(err.to_string(), "stage id node_id must not be empty");
}
}

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

@ -5,8 +5,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use fabro_agent::Sandbox;
use fabro_model::Provider;
use fabro_store::NodeVisitRef;
use fabro_types::RunId;
use fabro_types::{RunId, StageId};
use tokio::fs;
use crate::context::keys;
@ -264,7 +263,7 @@ pub(crate) async fn write_provider_used_file(
.await
.ok()
.and_then(|state| {
let node = NodeVisitRef { node_id, visit };
let node = StageId::new(node_id, visit);
state
.node(&node)
.and_then(|node_state| node_state.provider_used.clone())
@ -467,8 +466,10 @@ impl Handler for AgentHandler {
"mode": if node.backend() == Some("cli") { "cli" } else { "agent" },
"provider": node
.provider()
.map(String::from)
.unwrap_or_else(|| Provider::default_from_env().as_str().to_string()),
.map_or_else(
|| Provider::default_from_env().as_str().to_string(),
String::from,
),
"model": node.model().map(String::from).unwrap_or_default(),
})),
)
@ -483,7 +484,7 @@ mod tests {
use super::*;
use crate::event::EventEmitter;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{NodeVisitRef, SlateRunStore, SlateStore};
use fabro_store::{SlateRunStore, SlateStore, StageId};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use std::sync::Arc;
@ -829,12 +830,7 @@ mod tests {
logger.flush().await;
let state = run_store.state().await.unwrap();
let node_state = state
.node(&NodeVisitRef {
node_id: "step",
visit: 1,
})
.unwrap();
let node_state = state.node(&StageId::new("step", 1)).unwrap();
assert_eq!(
node_state.provider_used.as_ref().unwrap()["provider"],
"openai"

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

@ -203,7 +203,7 @@ mod tests {
use super::*;
use crate::outcome::StageStatus;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{NodeVisitRef, SlateRunStore, SlateStore};
use fabro_store::{SlateRunStore, SlateStore, StageId};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use std::sync::Arc;
@ -600,10 +600,7 @@ mod tests {
let snapshot = run_store.state().await.unwrap();
let node = snapshot
.node(&NodeVisitRef {
node_id: "script_node",
visit: 1,
})
.node(&StageId::new("script_node", 1))
.cloned()
.unwrap();

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

@ -728,12 +728,7 @@ mod tests {
logger.flush().await;
let state = run_store.state().await.unwrap();
let node_state = state
.node(&fabro_store::NodeVisitRef {
node_id: "par",
visit: 1,
})
.unwrap();
let node_state = state.node(&fabro_store::StageId::new("par", 1)).unwrap();
let results = node_state.parallel_results.as_ref().unwrap();
assert!(results.is_array());
assert_eq!(results.as_array().unwrap().len(), 2);

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

@ -207,7 +207,7 @@ impl Handler for PromptHandler {
mod tests {
use super::*;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{NodeVisitRef, SlateRunStore, SlateStore};
use fabro_store::{SlateRunStore, SlateStore, StageId};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use std::sync::Arc;
@ -395,12 +395,7 @@ mod tests {
logger.flush().await;
let state = run_store.state().await.unwrap();
let node_state = state
.node(&NodeVisitRef {
node_id: "classify",
visit: 1,
})
.unwrap();
let node_state = state.node(&StageId::new("classify", 1)).unwrap();
assert_eq!(node_state.provider_used.as_ref().unwrap()["mode"], "prompt");
}

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

@ -5,8 +5,8 @@ use std::path::PathBuf;
use anyhow::{Context, Result, bail};
use fabro_checkpoint::branch::BranchStore;
use fabro_checkpoint::git::Store as GitStore;
use fabro_store::{NodeVisitRef, SlateRunStore as DurableRunStore, SlateStore as DurableStore};
use fabro_types::RunId;
use fabro_store::{SlateRunStore as DurableRunStore, SlateStore as DurableStore};
use fabro_types::{RunId, StageId};
use git2::{Repository, Signature};
use ulid::Ulid;
@ -68,7 +68,7 @@ pub async fn rebuild_metadata_branch(
for visit in 1..=max_visit {
let visit = u32::try_from(visit)
.with_context(|| format!("visit {visit} for node {node_id} exceeds u32"))?;
let Some(node) = state.node(&NodeVisitRef { node_id, visit }).cloned() else {
let Some(node) = state.node(&StageId::new(node_id, visit)).cloned() else {
continue;
};
@ -333,7 +333,7 @@ fn resolve_prefix_matches(prefix: &str, matches: Vec<RunId>) -> Result<RunId> {
mod tests {
use chrono::{TimeZone, Utc};
use fabro_graphviz::graph::Graph;
use fabro_store::{NodeVisitRef, SlateStore, StoreHandle};
use fabro_store::{SlateStore, StageId, StoreHandle};
use fabro_types::{RunId, RunRecord, SandboxRecord, Settings, StartRecord, fixtures};
use object_store::memory::InMemory;
use std::collections::HashMap;
@ -533,15 +533,15 @@ mod tests {
async fn append_prompt_event(
run_store: &DurableRunStore,
run_id: RunId,
node: &NodeVisitRef<'_>,
node: &StageId,
text: &str,
) {
append_workflow_event(
run_store,
&run_id,
&WorkflowRunEvent::Prompt {
stage: node.node_id.to_string(),
visit: node.visit,
stage: node.node_id().to_string(),
visit: node.visit(),
text: text.to_string(),
mode: None,
provider: None,
@ -637,16 +637,10 @@ mod tests {
let durable_store = memory_store();
let run_store = create_run_store(&durable_store, test_run_id(), None).await;
let build_v1 = NodeVisitRef {
node_id: "build",
visit: 1,
};
let build_v1 = StageId::new("build", 1);
append_prompt_event(&run_store, test_run_id(), &build_v1, "visit one").await;
let build_v2 = NodeVisitRef {
node_id: "build",
visit: 2,
};
let build_v2 = StageId::new("build", 2);
append_prompt_event(&run_store, test_run_id(), &build_v2, "visit two").await;
append_checkpoint_event(
@ -1007,10 +1001,7 @@ mod tests {
let run_store = create_run_store(&durable_store, test_run_id(), None).await;
let bad_node = "bad\0node";
let bad_visit = NodeVisitRef {
node_id: bad_node,
visit: 1,
};
let bad_visit = StageId::new(bad_node, 1);
append_prompt_event(&run_store, test_run_id(), &bad_visit, "prompt").await;
append_checkpoint_event(
&run_store,

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

@ -643,12 +643,7 @@ async fn execute_writes_start_json_and_node_status() {
);
assert_eq!(start.base_sha.as_deref(), Some("abc123"));
let node = state
.node(&fabro_store::NodeVisitRef {
node_id: "start",
visit: 1,
})
.unwrap();
let node = state.node(&fabro_store::StageId::new("start", 1)).unwrap();
assert_eq!(node.status.as_ref().unwrap().status, StageStatus::Success);
}
@ -700,10 +695,7 @@ async fn timeout_causes_fail_status_record() {
.await;
let state = executed.run_store.state().await.unwrap();
let status = state
.node(&fabro_store::NodeVisitRef {
node_id: "work",
visit: 1,
})
.node(&fabro_store::StageId::new("work", 1))
.unwrap()
.status
.as_ref()

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

@ -208,8 +208,8 @@ 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;
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()))
@ -908,10 +908,7 @@ mod tests {
fn read_plan_text_found() {
let mut state = RunProjection::default();
state.set_node(
fabro_store::NodeVisitRef {
node_id: "plan",
visit: 1,
},
fabro_store::StageId::new("plan", 1),
fabro_store::NodeState {
response: Some("This is the plan".to_string()),
..Default::default()
@ -926,10 +923,7 @@ mod tests {
fn read_plan_text_prefix_match() {
let mut state = RunProjection::default();
state.set_node(
fabro_store::NodeVisitRef {
node_id: "planning",
visit: 1,
},
fabro_store::StageId::new("planning", 1),
fabro_store::NodeState {
response: Some("Planning content".to_string()),
..Default::default()
@ -944,20 +938,14 @@ mod tests {
fn read_plan_text_prefers_alphabetically_first_plan_node() {
let mut state = RunProjection::default();
state.set_node(
fabro_store::NodeVisitRef {
node_id: "planning",
visit: 1,
},
fabro_store::StageId::new("planning", 1),
fabro_store::NodeState {
response: Some("Planning content".to_string()),
..Default::default()
},
);
state.set_node(
fabro_store::NodeVisitRef {
node_id: "plan",
visit: 1,
},
fabro_store::StageId::new("plan", 1),
fabro_store::NodeState {
response: Some("Plan content".to_string()),
..Default::default()
@ -972,10 +960,7 @@ mod tests {
fn read_plan_text_not_found() {
let mut state = RunProjection::default();
state.set_node(
fabro_store::NodeVisitRef {
node_id: "implement",
visit: 1,
},
fabro_store::StageId::new("implement", 1),
fabro_store::NodeState::default(),
);

34
lib/crates/fabro-workflow/src/run_dump.rs
View file

@ -43,18 +43,15 @@ impl RunDump {
#[must_use]
pub fn metadata_checkpoint(state: &RunProjection) -> Self {
let mut entries = Vec::new();
let mut keys: Vec<_> = state
.iter_nodes()
.map(|(node, _)| node.into_owned())
.collect();
let mut keys: Vec<_> = state.iter_nodes().map(|(node, _)| node.clone()).collect();
keys.sort();
for node_key in keys {
let Some(node) = state.node(&node_key.as_ref()) else {
let Some(node) = state.node(&node_key) else {
continue;
};
let node_id = node_key.node_id.as_str();
let visit = node_key.visit;
let node_id = node_key.node_id();
let visit = node_key.visit();
if let Some(prompt) = node.prompt.as_ref() {
entries.push(RunDumpEntry::text(
@ -147,19 +144,16 @@ impl RunDump {
push_json_entry(&mut entries, "sandbox.json", record);
}
let mut node_keys: Vec<_> = state
.iter_nodes()
.map(|(node, _)| node.into_owned())
.collect();
let mut node_keys: Vec<_> = state.iter_nodes().map(|(node, _)| node.clone()).collect();
node_keys.sort();
for node_key in &node_keys {
let node = state
.node(&node_key.as_ref())
.with_context(|| format!("missing node {:?} in projection", node_key))?;
let node_id_segment = validate_single_path_segment("node id", &node_key.node_id)?;
.node(node_key)
.with_context(|| format!("missing node {node_key:?} in projection"))?;
let node_id_segment = validate_single_path_segment("node id", node_key.node_id())?;
let base = PathBuf::from("nodes")
.join(node_id_segment)
.join(format!("visit-{}", node_key.visit));
.join(format!("visit-{}", node_key.visit()));
if let Some(prompt) = node.prompt.as_ref() {
entries.push(RunDumpEntry::text_path(
@ -229,22 +223,24 @@ impl RunDump {
}
for asset in run_store.list_all_assets().await? {
let node_id_segment = validate_single_path_segment("node id", &asset.node.node_id)?;
let node_id_segment = validate_single_path_segment("node id", asset.node.node_id())?;
let filename_path = validate_relative_path("asset filename", &asset.filename)?;
let data = run_store
.get_asset(&asset.node.as_ref(), &asset.filename)
.get_asset(&asset.node, &asset.filename)
.await?
.with_context(|| {
format!(
"asset {:?} for node {:?} visit {} is missing from the store",
asset.filename, asset.node.node_id, asset.node.visit
asset.filename,
asset.node.node_id(),
asset.node.visit()
)
})?;
entries.push(RunDumpEntry::bytes_path(
&PathBuf::from("artifacts")
.join("nodes")
.join(node_id_segment)
.join(format!("visit-{}", asset.node.visit))
.join(format!("visit-{}", asset.node.visit()))
.join(filename_path),
data.to_vec(),
));