diff --git a/lib/crates/fabro-dump/src/lib.rs b/lib/crates/fabro-dump/src/lib.rs index 262669469..57c51cbf7 100644 --- a/lib/crates/fabro-dump/src/lib.rs +++ b/lib/crates/fabro-dump/src/lib.rs @@ -47,21 +47,14 @@ impl RunDump { entries.push(RunDumpEntry::text("graph.fabro", graph_source.clone())); } - let mut stage_ids: Vec<_> = state - .iter_stages() - .map(|(stage_id, stage)| (stage_id.clone(), stage.first_event_seq)) - .collect(); - stage_ids.sort_by(|(left_id, left_seq), (right_id, right_seq)| { - left_seq - .get() - .cmp(&right_seq.get()) + let mut stages: Vec<_> = state.iter_stages().collect(); + stages.sort_by(|(left_id, left), (right_id, right)| { + left.first_event_seq + .cmp(&right.first_event_seq) .then_with(|| left_id.cmp(right_id)) }); - for (index, (stage_id, _)) in stage_ids.into_iter().enumerate() { - let Some(stage) = state.stage(&stage_id) else { - continue; - }; + for (index, (stage_id, stage)) in stages.into_iter().enumerate() { let rank = index + 1; let base = PathBuf::from("stages").join(format!("{rank:03}-{stage_id}")); @@ -431,7 +424,6 @@ fn ensure_parent_dir(path: &Path) -> Result<()> { #[cfg(test)] mod tests { use std::collections::HashMap; - use std::num::NonZeroU32; use chrono::{TimeZone, Utc}; use fabro_store::{RunProjection, StageId}; @@ -439,16 +431,12 @@ mod tests { use fabro_types::run::RunSpec; use fabro_types::{ Checkpoint, Conclusion, RunStatus, SandboxRecord, StageCompletion, StageOutcome, - StartRecord, SuccessReason, WorkflowSettings, fixtures, + StartRecord, SuccessReason, WorkflowSettings, first_event_seq, fixtures, }; use futures::executor; use super::{RunDump, RunDumpContents, RunDumpEntry}; - fn nonzero(value: u32) -> NonZeroU32 { - NonZeroU32::new(value).expect("test sequence must be non-zero") - } - fn sample_run_spec() -> RunSpec { RunSpec { run_id: fixtures::RUN_1, @@ -533,7 +521,8 @@ mod tests { }); projection.retro_prompt = Some("retro prompt".to_string()); projection.retro_response = Some("retro response".to_string()); - let stage = projection.stage_entry(stage_id.node_id(), stage_id.visit(), nonzero(2)); + let stage = + projection.stage_entry(stage_id.node_id(), stage_id.visit(), first_event_seq(2)); stage.prompt = Some("plan".to_string()); stage.response = Some("done".to_string()); stage.completion = Some(StageCompletion { @@ -612,8 +601,12 @@ mod tests { #[test] fn from_projection_prefixes_stage_paths_but_not_artifact_paths() { let mut projection = RunProjection::default(); - projection.stage_entry("zebra", 1, nonzero(1)).prompt = Some("first".to_string()); - projection.stage_entry("apple", 1, nonzero(2)).prompt = Some("second".to_string()); + projection + .stage_entry("zebra", 1, first_event_seq(1)) + .prompt = Some("first".to_string()); + projection + .stage_entry("apple", 1, first_event_seq(2)) + .prompt = Some("second".to_string()); let mut dump = RunDump::from_projection(&projection).unwrap(); dump.add_artifact_bytes(&StageId::new("zebra", 1), "report.txt", b"z".to_vec()) diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index fb412f371..1046f60b8 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/lib/crates/fabro-retro/src/retro_agent.rs @@ -330,21 +330,16 @@ async fn upload_data_files( #[cfg(test)] mod tests { - use std::num::NonZeroU32; use std::sync::Arc; use chrono::{TimeZone, Utc}; use fabro_agent::LocalSandbox; use fabro_store::StageId; - use fabro_types::{StageCompletion, StageOutcome}; + use fabro_types::{StageCompletion, StageOutcome, first_event_seq}; use tokio::fs; use super::*; - fn nonzero(value: u32) -> NonZeroU32 { - NonZeroU32::new(value).expect("test sequence must be non-zero") - } - #[test] fn submit_retro_schema_is_valid_json() { let schema: serde_json::Value = serde_json::from_str(SUBMIT_RETRO_SCHEMA).unwrap(); @@ -415,7 +410,7 @@ mod tests { let stage_id = StageId::new("build", 2); let mut state = RunProjection::default(); state.graph_source = Some("digraph Ship {}".to_string()); - let stage = state.stage_entry(stage_id.node_id(), stage_id.visit(), nonzero(2)); + let stage = state.stage_entry(stage_id.node_id(), stage_id.visit(), first_event_seq(2)); stage.prompt = Some("plan".to_string()); stage.response = Some("done".to_string()); stage.completion = Some(StageCompletion { @@ -520,7 +515,7 @@ mod tests { let mut state = RunProjection::default(); let stdout_ref = fabro_types::format_blob_ref(&stdout_id); let stderr_ref = fabro_types::format_blob_ref(&stderr_id); - let stage = state.stage_entry(stage_id.node_id(), stage_id.visit(), nonzero(1)); + let stage = state.stage_entry(stage_id.node_id(), stage_id.visit(), first_event_seq(1)); stage.script_invocation = Some(serde_json::json!({ "command": "cargo test", "stdout": stdout_ref, diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 98309385e..8527ab9d0 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -1,5 +1,4 @@ use std::collections::{BTreeMap, HashMap}; -use std::num::NonZeroU32; use std::str::FromStr; use chrono::{DateTime, Utc}; @@ -11,7 +10,7 @@ use fabro_types::{ BilledModelUsage, Checkpoint, Conclusion, EventBody, FailureSignature, InterviewQuestionRecord, Outcome, PendingInterviewRecord, PullRequestRecord, RunControlAction, RunId, RunProjection, RunSpec, RunStatus, RunSummary, SandboxRecord, StageCompletion, StageOutcome, StartRecord, - TerminalStatus, + TerminalStatus, first_event_seq, }; use fabro_util::error::render_with_causes; use serde_json::Value; @@ -403,10 +402,6 @@ impl RunProjectionReducer for RunProjection { } } -fn first_event_seq(seq: u32) -> NonZeroU32 { - NonZeroU32::new(seq).expect("event seq starts at 1") -} - pub(crate) fn build_summary(state: &RunProjection, run_id: &RunId) -> RunSummary { let workflow_name = state.spec.as_ref().map(|spec| { if spec.graph.name.is_empty() { @@ -607,7 +602,6 @@ fn provider_used_from_agent_cli_started(props: &AgentCliStartedProps) -> Value { #[cfg(test)] mod tests { use std::collections::{BTreeMap, HashMap}; - use std::num::NonZeroU32; use chrono::Utc; use fabro_types::run_event::run::RunFailedProps; @@ -618,17 +612,13 @@ mod tests { use fabro_types::{ BlockedReason, Checkpoint, EventBody, FailureReason, Outcome, QuestionType, RunBlobId, RunControlAction, RunEvent, RunStatus, StageOutcome, SuccessReason, TerminalStatus, - WorkflowSettings, fixtures, + WorkflowSettings, first_event_seq, fixtures, }; use serde_json::json; use super::{RunProjection, RunProjectionReducer, build_summary}; use crate::{Error, EventEnvelope, StageId}; - fn nonzero(value: u32) -> NonZeroU32 { - NonZeroU32::new(value).expect("test sequence must be non-zero") - } - fn test_event(seq: u32, body: EventBody, node_id: Option<&str>) -> EventEnvelope { let event = RunEvent { id: format!("evt-{seq}"), @@ -751,7 +741,7 @@ mod tests { let stage_id = StageId::new("build", 2); let node = state.stage(&stage_id).unwrap(); - assert_eq!(node.first_event_seq, nonzero(1)); + assert_eq!(node.first_event_seq, first_event_seq(1)); 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)); @@ -787,7 +777,7 @@ mod tests { restart_failure_signatures: HashMap::new(), node_visits: HashMap::from([("build".to_string(), 2usize)]), })]; - state.stage_entry("build", 2, nonzero(7)).stdout = Some("done".to_string()); + state.stage_entry("build", 2, first_event_seq(7)).stdout = Some("done".to_string()); let round_tripped: RunProjection = serde_json::from_value(serde_json::to_value(&state).unwrap()).unwrap(); @@ -826,7 +816,7 @@ mod tests { .unwrap(); let stage = state.stage(&stage_id).unwrap(); - assert_eq!(stage.first_event_seq, nonzero(3)); + assert_eq!(stage.first_event_seq, first_event_seq(3)); } #[test] @@ -861,7 +851,7 @@ mod tests { .unwrap(); let stage = state.stage(&stage_id).unwrap(); - assert_eq!(stage.first_event_seq, nonzero(3)); + assert_eq!(stage.first_event_seq, first_event_seq(3)); assert_eq!(stage.prompt.as_deref(), Some("prompt")); } @@ -895,7 +885,7 @@ mod tests { .unwrap(); let stage = state.stage(&stage_id).unwrap(); - assert_eq!(stage.first_event_seq, nonzero(5)); + assert_eq!(stage.first_event_seq, first_event_seq(5)); let completion = stage.completion.as_ref().unwrap(); assert_eq!(completion.outcome, StageOutcome::Skipped); assert_eq!(completion.notes.as_deref(), Some("condition was false")); diff --git a/lib/crates/fabro-store/src/serializable_projection.rs b/lib/crates/fabro-store/src/serializable_projection.rs index 6bcf0c837..7d888b4a3 100644 --- a/lib/crates/fabro-store/src/serializable_projection.rs +++ b/lib/crates/fabro-store/src/serializable_projection.rs @@ -10,15 +10,7 @@ impl Serialize for SerializableProjection<'_> { S: Serializer, { let mut projection = self.0.clone(); - let stage_ids: Vec<_> = projection - .iter_stages() - .map(|(stage_id, _)| stage_id.clone()) - .collect(); - - for stage_id in stage_ids { - let Some(stage) = projection.stage_mut(&stage_id) else { - continue; - }; + for (_, stage) in projection.iter_stages_mut() { stage.prompt = None; stage.response = None; stage.diff = None; diff --git a/lib/crates/fabro-store/tests/serializable_projection.rs b/lib/crates/fabro-store/tests/serializable_projection.rs index 1560b5c7c..6d51065ba 100644 --- a/lib/crates/fabro-store/tests/serializable_projection.rs +++ b/lib/crates/fabro-store/tests/serializable_projection.rs @@ -1,5 +1,4 @@ use std::collections::{BTreeMap, HashMap}; -use std::num::NonZeroU32; use chrono::{TimeZone, Utc}; use fabro_store::{RunProjection, SerializableProjection, StageId}; @@ -7,14 +6,10 @@ use fabro_types::graph::Graph; use fabro_types::run::RunSpec; use fabro_types::{ Checkpoint, RunStatus, SandboxRecord, StageCompletion, StageOutcome, StartRecord, - TerminalStatus, WorkflowSettings, fixtures, + TerminalStatus, WorkflowSettings, first_event_seq, fixtures, }; use serde_json::json; -fn nonzero(value: u32) -> NonZeroU32 { - NonZeroU32::new(value).expect("test sequence must be non-zero") -} - fn sample_run_spec() -> RunSpec { RunSpec { run_id: fixtures::RUN_1, @@ -82,7 +77,7 @@ fn serializable_projection_round_trips_and_trims_bulky_node_fields() { clone_branch: None, }); projection.pending_interviews = BTreeMap::new(); - let stage = projection.stage_entry(stage_id.node_id(), stage_id.visit(), nonzero(2)); + let stage = projection.stage_entry(stage_id.node_id(), stage_id.visit(), first_event_seq(2)); stage.prompt = Some("plan the work".to_string()); stage.response = Some("done".to_string()); stage.completion = Some(StageCompletion { @@ -123,7 +118,7 @@ fn serializable_projection_round_trips_and_trims_bulky_node_fields() { assert_eq!(node.diff, None); assert_eq!(node.stdout, None); assert_eq!(node.stderr, None); - assert_eq!(node.first_event_seq, nonzero(2)); + assert_eq!(node.first_event_seq, first_event_seq(2)); assert_eq!( node.completion .as_ref() diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 6181ff184..01f8e08f9 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -70,7 +70,7 @@ pub use run_event::{ MetadataSnapshotPhase, RunEvent, RunNoticeLevel, }; pub use run_id::{RunId, fixtures}; -pub use run_projection::{PendingInterviewRecord, RunProjection, StageProjection}; +pub use run_projection::{PendingInterviewRecord, RunProjection, StageProjection, first_event_seq}; pub use run_summary::RunSummary; pub use sandbox_record::SandboxRecord; pub use secret::{SecretMetadata, SecretType}; diff --git a/lib/crates/fabro-types/src/run_projection.rs b/lib/crates/fabro-types/src/run_projection.rs index 971068a7e..de2e4a1a5 100644 --- a/lib/crates/fabro-types/src/run_projection.rs +++ b/lib/crates/fabro-types/src/run_projection.rs @@ -63,6 +63,13 @@ pub struct StageProjection { pub termination: Option, } +/// Convert a 1-based event sequence number into the `NonZeroU32` form used for +/// `StageProjection::first_event_seq`. Run event seqs always start at 1. +#[must_use] +pub fn first_event_seq(seq: u32) -> NonZeroU32 { + NonZeroU32::new(seq).expect("event seq starts at 1") +} + impl StageProjection { #[must_use] pub fn new(first_event_seq: NonZeroU32) -> Self { @@ -96,6 +103,10 @@ impl RunProjection { self.stages.iter() } + pub fn iter_stages_mut(&mut self) -> impl Iterator { + self.stages.iter_mut() + } + pub fn is_empty(&self) -> bool { self.stages.is_empty() } diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index f158bf5bc..c3fe3c215 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -593,7 +593,6 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result NonZeroU32 { - NonZeroU32::new(value).expect("test sequence must be non-zero") - } - fn checkpoint_with( completed_nodes: Vec<&str>, node_outcomes: HashMap, @@ -745,8 +742,8 @@ mod tests { #[test] fn conclusion_stage_order_follows_projection_first_event_order() { let mut projection = RunProjection::default(); - projection.stage_entry("zebra", 1, nonzero(1)); - projection.stage_entry("apple", 1, nonzero(2)); + projection.stage_entry("zebra", 1, first_event_seq(1)); + projection.stage_entry("apple", 1, first_event_seq(2)); let projection_order = stage_projection_order(&projection); let checkpoint = checkpoint_with( vec!["apple", "zebra"], @@ -777,8 +774,8 @@ mod tests { #[test] fn conclusion_includes_skipped_stage_from_projection_checkpoint_fallback() { let mut projection = RunProjection::default(); - projection.stage_entry("skipped", 1, nonzero(4)); - projection.stage_entry("finished", 1, nonzero(5)); + projection.stage_entry("skipped", 1, first_event_seq(4)); + projection.stage_entry("finished", 1, first_event_seq(5)); let projection_order = stage_projection_order(&projection); let checkpoint = checkpoint_with( vec!["finished"], diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 7067e5039..933f53052 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -625,7 +625,6 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> #[cfg(test)] mod tests { use std::collections::HashMap; - use std::num::NonZeroU32; use std::sync::Arc; use std::time::Duration; @@ -642,7 +641,7 @@ mod tests { AggregateStats, FrictionKind, FrictionPoint, OpenItem, OpenItemKind, StageRetro, }; use fabro_store::Database; - use fabro_types::{BilledTokenCounts, RunSpec, SuccessReason, fixtures}; + use fabro_types::{BilledTokenCounts, RunSpec, SuccessReason, first_event_seq, fixtures}; use fabro_vault::{SecretType, Vault}; use futures::stream; use httpmock::Method::POST; @@ -654,10 +653,6 @@ mod tests { use crate::event::{Event, append_event}; use crate::records::StageSummary; - fn nonzero(value: u32) -> NonZeroU32 { - NonZeroU32::new(value).expect("test sequence must be non-zero") - } - struct MockProvider { name: String, response_text: String, @@ -1002,7 +997,8 @@ mod tests { #[test] fn read_plan_text_found() { let mut state = RunProjection::default(); - state.stage_entry("plan", 1, nonzero(1)).response = Some("This is the plan".to_string()); + state.stage_entry("plan", 1, first_event_seq(1)).response = + Some("This is the plan".to_string()); let result = read_plan_text(&state); assert_eq!(result, Some("This is the plan".to_string())); @@ -1011,8 +1007,9 @@ mod tests { #[test] fn read_plan_text_prefix_match() { let mut state = RunProjection::default(); - state.stage_entry("planning", 1, nonzero(1)).response = - Some("Planning content".to_string()); + state + .stage_entry("planning", 1, first_event_seq(1)) + .response = Some("Planning content".to_string()); let result = read_plan_text(&state); assert_eq!(result, Some("Planning content".to_string())); @@ -1021,9 +1018,11 @@ mod tests { #[test] fn read_plan_text_prefers_alphabetically_first_plan_node() { let mut state = RunProjection::default(); - state.stage_entry("planning", 1, nonzero(1)).response = - Some("Planning content".to_string()); - state.stage_entry("plan", 1, nonzero(2)).response = Some("Plan content".to_string()); + state + .stage_entry("planning", 1, first_event_seq(1)) + .response = Some("Planning content".to_string()); + state.stage_entry("plan", 1, first_event_seq(2)).response = + Some("Plan content".to_string()); let result = read_plan_text(&state); assert_eq!(result, Some("Plan content".to_string())); @@ -1032,7 +1031,7 @@ mod tests { #[test] fn read_plan_text_not_found() { let mut state = RunProjection::default(); - state.stage_entry("implement", 1, nonzero(1)); + state.stage_entry("implement", 1, first_event_seq(1)); let result = read_plan_text(&state); assert_eq!(result, None);