From 7efb66cd3ee1f97ca8bede3c17ac82605302e8a4 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 3 Apr 2026 18:00:29 -0700 Subject: [PATCH] refactor: introduce StageId --- lib/crates/fabro-cli/src/commands/run/diff.rs | 2 +- .../src/commands/run/run_progress/mod.rs | 6 + .../fabro-cli/src/commands/store/dump.rs | 25 +-- lib/crates/fabro-cli/tests/it/cmd/support.rs | 4 +- lib/crates/fabro-store/src/keys.rs | 25 ++- lib/crates/fabro-store/src/lib.rs | 3 +- lib/crates/fabro-store/src/run_state.rs | 35 ++-- lib/crates/fabro-store/src/slate/mod.rs | 41 +---- lib/crates/fabro-store/src/slate/run_store.rs | 26 +-- lib/crates/fabro-store/src/types.rs | 35 +--- lib/crates/fabro-types/src/lib.rs | 2 + lib/crates/fabro-types/src/stage_id.rs | 154 ++++++++++++++++++ .../fabro-workflow/src/handler/agent.rs | 20 +-- .../fabro-workflow/src/handler/command.rs | 7 +- .../fabro-workflow/src/handler/parallel.rs | 7 +- .../fabro-workflow/src/handler/prompt.rs | 9 +- .../src/operations/rebuild_meta.rs | 29 ++-- .../src/pipeline/execute/tests.rs | 12 +- .../src/pipeline/pull_request.rs | 29 +--- lib/crates/fabro-workflow/src/run_dump.rs | 34 ++-- 20 files changed, 262 insertions(+), 243 deletions(-) create mode 100644 lib/crates/fabro-types/src/stage_id.rs diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index abc7cbb5a..ef5965be9 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -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); diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index a65801357..98cb6ee73 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -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(), }, diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 363ebca66..c72fcf765 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -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"); diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 433db3a6d..a7382ac00 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -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 => { diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index bb145994b..25b81ff13 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -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 { .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 { 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() )) ); diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 7cc889c19..df10a6034 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -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; diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 6888f2e00..27ecc9511 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -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, pub final_patch: Option, pub pull_request: Option, - nodes: HashMap, + nodes: HashMap, } #[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, &NodeState)> { - self.nodes - .iter() - .map(|(node, state)| (node.as_ref(), state)) + pub fn iter_nodes(&self) -> impl Iterator { + 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 { 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::>(); 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 { 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() } diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 4a0ea42d3..bfa840751 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -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 { 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!( diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 7e2ea363b..259cea462 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -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> { + pub async fn get_asset(&self, node: &StageId, filename: &str) -> Result> { 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) diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index cb263d6c8..3718470bd 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -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) } } diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 098e6f25b..7741df0cd 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -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, diff --git a/lib/crates/fabro-types/src/stage_id.rs b/lib/crates/fabro-types/src/stage_id.rs new file mode 100644 index 000000000..747133c78 --- /dev/null +++ b/lib/crates/fabro-types/src/stage_id.rs @@ -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, 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 { + 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(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&self.to_string()) + } +} + +impl<'de> Deserialize<'de> for StageId { + fn deserialize(deserializer: D) -> Result + 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::().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::().unwrap_err(); + assert_eq!(err.to_string(), "stage id must contain '@'"); + } + + #[test] + fn parse_rejects_empty_suffix() { + let err = "code@".parse::().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::().unwrap_err(); + assert!(err.to_string().starts_with("invalid stage id visit:")); + } + + #[test] + fn parse_rejects_empty_node_id() { + let err = "@3".parse::().unwrap_err(); + assert_eq!(err.to_string(), "stage id node_id must not be empty"); + } +} diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 66ea4b667..5b521abc6 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -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" diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 2383f59ad..1ccfbef58 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -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(); diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 9854428bc..1a1be405b 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -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); diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 54f8e1e89..3b78b4a2d 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -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"); } diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index f5d057162..17024ade3 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -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) -> Result { 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, diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index 2e01e37f2..b4fbae7ce 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -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() diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 8be8fc69d..e9d69d8b0 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -208,8 +208,8 @@ fn read_plan_text(state: &RunProjection) -> Option { 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(), ); diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index 1ad570518..115623795 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -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(), ));