From 0cf80cdcfba0994a45855525577f8ecd2cbba4d8 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 11:36:22 -0400 Subject: [PATCH] refactor(store): share run projection with cli Make fabro-store::RunProjection the single projection type used by the server, CLI, and CLI test helpers. This removes the duplicated CLI-side mirrors and adds serde coverage for the store-owned projection. --- lib/crates/fabro-cli/src/server_client.rs | 83 +------------ lib/crates/fabro-cli/tests/it/cmd/support.rs | 74 +----------- lib/crates/fabro-store/src/run_state.rs | 118 ++++++++++++++++++- 3 files changed, 119 insertions(+), 156 deletions(-) diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index 36cb5e124..b1f24a593 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -1,4 +1,3 @@ -use std::collections::HashMap; use std::num::NonZeroU64; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -7,10 +6,7 @@ use anyhow::{Context as _, Result, anyhow, bail}; use fabro_api::types; use fabro_server::bind::Bind; use fabro_store::{EventEnvelope, RunSummary, StageId}; -use fabro_types::{ - Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunEvent, RunId, RunRecord, - RunStatusRecord, SandboxRecord, Settings, StartRecord, -}; +use fabro_types::{RunEvent, RunId, Settings}; use futures::StreamExt; use serde::de::DeserializeOwned; use tokio::time::sleep; @@ -30,82 +26,7 @@ struct LocalServerRuntime { storage_dir: PathBuf, } -#[allow(dead_code)] -#[derive(Debug, Clone, Default, serde::Deserialize)] -pub(crate) struct RunProjection { - #[serde(default)] - pub run: Option, - #[serde(default)] - pub graph_source: Option, - #[serde(default)] - pub start: Option, - #[serde(default)] - pub status: Option, - #[serde(default)] - pub checkpoint: Option, - #[serde(default)] - pub checkpoints: Vec<(u32, Checkpoint)>, - #[serde(default)] - pub conclusion: Option, - #[serde(default)] - pub retro: Option, - #[serde(default)] - pub retro_prompt: Option, - #[serde(default)] - pub retro_response: Option, - #[serde(default)] - pub sandbox: Option, - #[serde(default)] - pub final_patch: Option, - #[serde(default)] - pub pull_request: Option, - #[serde(default)] - nodes: HashMap, -} - -#[allow(dead_code)] -#[derive(Debug, Clone, Default, serde::Deserialize)] -pub(crate) struct NodeState { - #[serde(default)] - pub prompt: Option, - #[serde(default)] - pub response: Option, - #[serde(default)] - pub status: Option, - #[serde(default)] - pub provider_used: Option, - #[serde(default)] - pub diff: Option, - #[serde(default)] - pub script_invocation: Option, - #[serde(default)] - pub script_timing: Option, - #[serde(default)] - pub parallel_results: Option, - #[serde(default)] - pub stdout: Option, - #[serde(default)] - pub stderr: Option, -} - -impl RunProjection { - pub(crate) fn list_node_visits(&self, node_id: &str) -> Vec { - let mut visits = self - .nodes - .keys() - .filter_map(|key| key.parse::().ok()) - .filter(|stage_id| stage_id.node_id() == node_id) - .map(|stage_id| stage_id.visit()) - .collect::>(); - visits.sort_unstable(); - visits.dedup(); - visits - } - - pub(crate) fn node(&self, stage_id: &StageId) -> Option<&NodeState> { - self.nodes.get(&stage_id.to_string()) - } -} +pub(crate) use fabro_store::RunProjection; pub(crate) async fn connect_server(storage_dir: &Path) -> Result { Ok(ServerStoreClient { diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 73fef9765..f0ca6dd14 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -13,72 +13,12 @@ use fabro_config::Storage; use fabro_server::bind::Bind; use fabro_store::EventEnvelope; use fabro_test::TestContext; -use fabro_types::{ - Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord, RunStatusRecord, - SandboxRecord, StageId, StartRecord, -}; use serde_json::Value; use shlex::try_quote; const COMMAND_TIMEOUT: Duration = Duration::from_secs(30); -#[allow(dead_code)] -#[derive(Debug, Clone, Default, serde::Deserialize)] -pub(crate) struct RunProjection { - #[serde(default)] - pub run: Option, - #[serde(default)] - pub graph_source: Option, - #[serde(default)] - pub start: Option, - #[serde(default)] - pub status: Option, - #[serde(default)] - pub checkpoint: Option, - #[serde(default)] - pub checkpoints: Vec<(u32, Checkpoint)>, - #[serde(default)] - pub conclusion: Option, - #[serde(default)] - pub retro: Option, - #[serde(default)] - pub retro_prompt: Option, - #[serde(default)] - pub retro_response: Option, - #[serde(default)] - pub sandbox: Option, - #[serde(default)] - pub final_patch: Option, - #[serde(default)] - pub pull_request: Option, - #[serde(default)] - pub nodes: std::collections::HashMap, -} - -#[allow(dead_code)] -#[derive(Debug, Clone, Default, serde::Deserialize)] -pub(crate) struct NodeState { - #[serde(default)] - pub prompt: Option, - #[serde(default)] - pub response: Option, - #[serde(default)] - pub status: Option, - #[serde(default)] - pub provider_used: Option, - #[serde(default)] - pub diff: Option, - #[serde(default)] - pub script_invocation: Option, - #[serde(default)] - pub script_timing: Option, - #[serde(default)] - pub parallel_results: Option, - #[serde(default)] - pub stdout: Option, - #[serde(default)] - pub stderr: Option, -} +pub(crate) use fabro_store::RunProjection; #[derive(Debug, Clone, Default, serde::Deserialize)] struct RunSummaryRecord { @@ -87,18 +27,6 @@ struct RunSummaryRecord { labels: std::collections::HashMap, } -impl RunProjection { - pub(crate) fn iter_nodes(&self) -> impl Iterator { - self.nodes - .iter() - .filter_map(|(stage_id, state)| stage_id.parse::().ok().map(|id| (id, state))) - } - - pub(crate) fn is_empty(&self) -> bool { - self.nodes.is_empty() - } -} - pub(crate) struct RunSetup { pub(crate) run_id: String, pub(crate) run_dir: PathBuf, diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 8de4a4867..50b46d868 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -16,7 +16,8 @@ use fabro_types::{ RunStatusRecord, SandboxRecord, StageStatus, StageUsage, StartRecord, StatusReason, TokenUsage, }; -#[derive(Debug, Clone, Default, serde::Serialize)] +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] +#[serde(default)] pub struct RunProjection { pub run: Option, pub graph_source: Option, @@ -35,7 +36,7 @@ pub struct RunProjection { nodes: HashMap, } -#[derive(Debug, Clone, Default, serde::Serialize)] +#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] pub struct NodeState { pub prompt: Option, pub response: Option, @@ -578,3 +579,116 @@ fn run_usage_from_token_usage(usage: &TokenUsage) -> RunUsage { cost: None, } } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::{NodeState, RunProjection}; + use crate::StageId; + use fabro_types::{Checkpoint, RunControlAction}; + + #[test] + fn deserialize_projection_defaults_missing_nodes_and_checkpoints() { + let state: RunProjection = serde_json::from_value(serde_json::json!({ + "pending_control": "pause" + })) + .unwrap(); + + assert_eq!(state.pending_control, Some(RunControlAction::Pause)); + assert!(state.checkpoints.is_empty()); + assert!(state.is_empty()); + } + + #[test] + fn deserialize_and_round_trip_projection_preserves_stage_ids_and_pending_control() { + let state: RunProjection = serde_json::from_value(serde_json::json!({ + "pending_control": "cancel", + "checkpoints": [[ + 0, + { + "timestamp": "2026-04-07T12:00:00Z", + "current_node": "build", + "completed_nodes": ["build"], + "node_retries": {}, + "context_values": {}, + "node_outcomes": {}, + "loop_failure_signatures": {}, + "restart_failure_signatures": {}, + "node_visits": { "build": 2 } + } + ]], + "nodes": { + "build@2": { + "diff": "diff --git a/file b/file", + "stdout": "done" + } + } + })) + .unwrap(); + + let stage_id = StageId::new("build", 2); + let node = state.node(&stage_id).unwrap(); + assert_eq!(node.diff.as_deref(), Some("diff --git a/file b/file")); + assert_eq!(state.list_node_visits("build"), vec![2]); + assert_eq!(state.pending_control, Some(RunControlAction::Cancel)); + + let round_tripped: RunProjection = + serde_json::from_value(serde_json::to_value(&state).unwrap()).unwrap(); + let round_tripped_node = round_tripped.node(&stage_id).unwrap(); + assert_eq!(round_tripped_node.stdout.as_deref(), Some("done")); + assert_eq!(round_tripped.list_node_visits("build"), vec![2]); + assert_eq!( + round_tripped.pending_control, + Some(RunControlAction::Cancel) + ); + } + + #[test] + fn set_node_round_trips_through_json() { + let mut state = RunProjection { + pending_control: Some(RunControlAction::Unpause), + checkpoints: vec![( + 7, + Checkpoint { + timestamp: "2026-04-07T12:00:00Z".parse().unwrap(), + current_node: "build".to_string(), + completed_nodes: vec!["build".to_string()], + node_retries: HashMap::new(), + context_values: HashMap::new(), + node_outcomes: HashMap::new(), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: HashMap::new(), + restart_failure_signatures: HashMap::new(), + node_visits: HashMap::from([("build".to_string(), 2usize)]), + }, + )], + ..RunProjection::default() + }; + state.set_node( + StageId::new("build", 2), + NodeState { + stdout: Some("done".to_string()), + ..NodeState::default() + }, + ); + + let round_tripped: RunProjection = + serde_json::from_value(serde_json::to_value(&state).unwrap()).unwrap(); + + assert_eq!( + round_tripped + .node(&StageId::new("build", 2)) + .unwrap() + .stdout + .as_deref(), + Some("done") + ); + assert_eq!(round_tripped.list_node_visits("build"), vec![2]); + assert_eq!( + round_tripped.pending_control, + Some(RunControlAction::Unpause) + ); + } +}