scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-07 03:00:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(run-projection): dedupe first_event_seq helper and tighten dump

Browse source 
Expose `fabro_types::first_event_seq` next to `StageProjection`, replacing
six identical `nonzero` test helpers and the private one in `run_state`.
Add `RunProjection::iter_stages_mut` so `SerializableProjection` can
clear bulky fields without the collect-then-lookup dance, and let the
`fabro-dump` loop iterate `(&StageId, &StageProjection)` borrows directly
to drop the per-stage `StageId::clone()` and redundant HashMap lookup.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-05-01 20:41:57 -04:00
parent cea1fa739d
commit 840dc42d3c
No known key found for this signature in database
9 changed files with 59 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
lib/crates/fabro-dump/src/lib.rs
View file

@ -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())

11
lib/crates/fabro-retro/src/retro_agent.rs
View file

@ -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,

24
lib/crates/fabro-store/src/run_state.rs
View file

@ -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"));

10
lib/crates/fabro-store/src/serializable_projection.rs
View file

@ -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;

11
lib/crates/fabro-store/tests/serializable_projection.rs
View file

@ -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()

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

@ -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};

11
lib/crates/fabro-types/src/run_projection.rs
View file

@ -63,6 +63,13 @@ pub struct StageProjection {
pub termination: Option<crate::CommandTermination>,
}
/// 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<Item = (&StageId, &mut StageProjection)> {
self.stages.iter_mut()
}
pub fn is_empty(&self) -> bool {
self.stages.is_empty()
}

17
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -593,7 +593,6 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result<Con
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::num::NonZeroU32;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
@ -604,7 +603,9 @@ mod tests {
use fabro_graphviz::graph::Graph;
use fabro_store::{Database, EventEnvelope, RunDatabase, RunProjection};
use fabro_types::run_event::{MetadataSnapshotFailureKind, MetadataSnapshotPhase};
use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, WorkflowSettings, fixtures};
use fabro_types::{
EventBody, RunBlobId, RunEvent, RunId, WorkflowSettings, first_event_seq, fixtures,
};
use object_store::memory::InMemory;
use super::*;
@ -715,10 +716,6 @@ mod tests {
events
}
fn nonzero(value: u32) -> NonZeroU32 {
NonZeroU32::new(value).expect("test sequence must be non-zero")
}
fn checkpoint_with(
completed_nodes: Vec<&str>,
node_outcomes: HashMap<String, Outcome>,
@ -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"],

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

@ -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);