mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-09-07 08:27:12 +00:00
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.
This commit is contained in:
Bryan Helmkamp
parent
a2672aef46
commit
0cf80cdcfb
3 changed files with 119 additions and 156 deletions
|
|
@ -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<RunRecord>,
|
||||
#[serde(default)]
|
||||
pub graph_source: Option<String>,
|
||||
#[serde(default)]
|
||||
pub start: Option<StartRecord>,
|
||||
#[serde(default)]
|
||||
pub status: Option<RunStatusRecord>,
|
||||
#[serde(default)]
|
||||
pub checkpoint: Option<Checkpoint>,
|
||||
#[serde(default)]
|
||||
pub checkpoints: Vec<(u32, Checkpoint)>,
|
||||
#[serde(default)]
|
||||
pub conclusion: Option<Conclusion>,
|
||||
#[serde(default)]
|
||||
pub retro: Option<Retro>,
|
||||
#[serde(default)]
|
||||
pub retro_prompt: Option<String>,
|
||||
#[serde(default)]
|
||||
pub retro_response: Option<String>,
|
||||
#[serde(default)]
|
||||
pub sandbox: Option<SandboxRecord>,
|
||||
#[serde(default)]
|
||||
pub final_patch: Option<String>,
|
||||
#[serde(default)]
|
||||
pub pull_request: Option<PullRequestRecord>,
|
||||
#[serde(default)]
|
||||
nodes: HashMap<String, NodeState>,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Default, serde::Deserialize)]
|
||||
pub(crate) struct NodeState {
|
||||
#[serde(default)]
|
||||
pub prompt: Option<String>,
|
||||
#[serde(default)]
|
||||
pub response: Option<String>,
|
||||
#[serde(default)]
|
||||
pub status: Option<NodeStatusRecord>,
|
||||
#[serde(default)]
|
||||
pub provider_used: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub diff: Option<String>,
|
||||
#[serde(default)]
|
||||
pub script_invocation: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub script_timing: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub parallel_results: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub stdout: Option<String>,
|
||||
#[serde(default)]
|
||||
pub stderr: Option<String>,
|
||||
}
|
||||
|
||||
impl RunProjection {
|
||||
pub(crate) fn list_node_visits(&self, node_id: &str) -> Vec<u32> {
|
||||
let mut visits = self
|
||||
.nodes
|
||||
.keys()
|
||||
.filter_map(|key| key.parse::<StageId>().ok())
|
||||
.filter(|stage_id| stage_id.node_id() == node_id)
|
||||
.map(|stage_id| stage_id.visit())
|
||||
.collect::<Vec<_>>();
|
||||
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<ServerStoreClient> {
|
||||
Ok(ServerStoreClient {
|
||||
|
|
|
|||
|
|
@ -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<RunRecord>,
|
||||
#[serde(default)]
|
||||
pub graph_source: Option<String>,
|
||||
#[serde(default)]
|
||||
pub start: Option<StartRecord>,
|
||||
#[serde(default)]
|
||||
pub status: Option<RunStatusRecord>,
|
||||
#[serde(default)]
|
||||
pub checkpoint: Option<Checkpoint>,
|
||||
#[serde(default)]
|
||||
pub checkpoints: Vec<(u32, Checkpoint)>,
|
||||
#[serde(default)]
|
||||
pub conclusion: Option<Conclusion>,
|
||||
#[serde(default)]
|
||||
pub retro: Option<Retro>,
|
||||
#[serde(default)]
|
||||
pub retro_prompt: Option<String>,
|
||||
#[serde(default)]
|
||||
pub retro_response: Option<String>,
|
||||
#[serde(default)]
|
||||
pub sandbox: Option<SandboxRecord>,
|
||||
#[serde(default)]
|
||||
pub final_patch: Option<String>,
|
||||
#[serde(default)]
|
||||
pub pull_request: Option<PullRequestRecord>,
|
||||
#[serde(default)]
|
||||
pub nodes: std::collections::HashMap<String, NodeState>,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Default, serde::Deserialize)]
|
||||
pub(crate) struct NodeState {
|
||||
#[serde(default)]
|
||||
pub prompt: Option<String>,
|
||||
#[serde(default)]
|
||||
pub response: Option<String>,
|
||||
#[serde(default)]
|
||||
pub status: Option<NodeStatusRecord>,
|
||||
#[serde(default)]
|
||||
pub provider_used: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub diff: Option<String>,
|
||||
#[serde(default)]
|
||||
pub script_invocation: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub script_timing: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub parallel_results: Option<serde_json::Value>,
|
||||
#[serde(default)]
|
||||
pub stdout: Option<String>,
|
||||
#[serde(default)]
|
||||
pub stderr: Option<String>,
|
||||
}
|
||||
pub(crate) use fabro_store::RunProjection;
|
||||
|
||||
#[derive(Debug, Clone, Default, serde::Deserialize)]
|
||||
struct RunSummaryRecord {
|
||||
|
|
@ -87,18 +27,6 @@ struct RunSummaryRecord {
|
|||
labels: std::collections::HashMap<String, String>,
|
||||
}
|
||||
|
||||
impl RunProjection {
|
||||
pub(crate) fn iter_nodes(&self) -> impl Iterator<Item = (StageId, &NodeState)> {
|
||||
self.nodes
|
||||
.iter()
|
||||
.filter_map(|(stage_id, state)| stage_id.parse::<StageId>().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,
|
||||
|
|
|
|||
|
|
@ -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<RunRecord>,
|
||||
pub graph_source: Option<String>,
|
||||
|
|
@ -35,7 +36,7 @@ pub struct RunProjection {
|
|||
nodes: HashMap<StageId, NodeState>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, serde::Serialize)]
|
||||
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
|
||||
pub struct NodeState {
|
||||
pub prompt: Option<String>,
|
||||
pub response: Option<String>,
|
||||
|
|
@ -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)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue