diff --git a/lib/crates/fabro-cli/tests/it/cmd/inspect.rs b/lib/crates/fabro-cli/tests/it/cmd/inspect.rs index fd1daf2b0..08ff97bf5 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/inspect.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/inspect.rs @@ -68,7 +68,7 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() { let run = setup_completed_dry_run(&context); let output = run_success(&context, &["inspect", &run.run_id]); - assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r###" + assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r#" [ { "run_id": "[ULID]", @@ -86,7 +86,7 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() { "conclusion": { "status": "success", "duration_ms": "[DURATION_MS]", - "stage_count": 3 + "stage_count": null }, "checkpoint": { "current_node": "report", @@ -102,7 +102,7 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() { } } ] - "###); + "#); } #[test] @@ -120,7 +120,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() { } let output = run_success(&context, &["inspect", &run.run_id]); - assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r###" + assert_snapshot!(serde_json::to_string_pretty(&compact_inspect(&output)).unwrap(), @r#" [ { "run_id": "[ULID]", @@ -138,7 +138,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() { "conclusion": { "status": "success", "duration_ms": "[DURATION_MS]", - "stage_count": 3 + "stage_count": null }, "checkpoint": { "current_node": "report", @@ -154,7 +154,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() { } } ] - "###); + "#); } #[test] @@ -165,7 +165,7 @@ fn inspect_git_backed_run_exposes_checkpoint_and_sandbox_state() { assert_snapshot!( serde_json::to_string_pretty(&compact_git_inspect(&output)).unwrap(), - @r###" + @r#" [ { "run_id": "[ULID]", @@ -186,7 +186,7 @@ fn inspect_git_backed_run_exposes_checkpoint_and_sandbox_state() { "status": "success", "duration_ms": "[DURATION_MS]", "final_git_commit_sha": "[SHA]", - "stage_count": 3 + "stage_count": null }, "checkpoint": { "current_node": "step_two", @@ -204,6 +204,6 @@ fn inspect_git_backed_run_exposes_checkpoint_and_sandbox_state() { } } ] - "### + "# ); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/logs.rs b/lib/crates/fabro-cli/tests/it/cmd/logs.rs index ef9fe6977..d9ef6a4e2 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/logs.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/logs.rs @@ -66,10 +66,10 @@ fn logs_completed_run_outputs_raw_ndjson() { {"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + {"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} - {"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} @@ -231,10 +231,10 @@ fn logs_follow_detached_run_streams_until_completion() { {"event":"run.submitted","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.starting","id":"[EVENT_ID]","properties":{"reason":"sandbox_initializing"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.initializing","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + {"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.ready","id":"[EVENT_ID]","properties":{"cpu":null,"duration_ms": [DURATION_MS],"memory":null,"name":null,"provider":"local","url":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"sandbox.initialized","id":"[EVENT_ID]","properties":{"provider":"local","working_directory":"[TEMP_DIR]"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"run.started","id":"[EVENT_ID]","properties":{"goal":"Run tests and report results","name":"Simple"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} - {"event":"run.running","id":"[EVENT_ID]","properties":{},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"stage.started","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"handler_type":"start","index":0,"max_attempts":1},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"stage.completed","id":"[EVENT_ID]","node_id":"start","node_label":"Start","properties":{"attempt":1,"context_values":{"current.preamble":"Goal: Run tests and report results/n","current_node":"start","graph.goal":"Run tests and report results","graph.rankdir":"LR","internal.fidelity":"compact","internal.node_visit_count":1,"internal.run_id":"[ULID]","internal.thread_id":null},"duration_ms": [DURATION_MS],"files_touched":[],"index":0,"max_attempts":1,"node_visits":{"start":1},"notes":"[Simulated] start","preferred_label":null,"status":"success","suggested_next_ids":[],"usage":null},"run_id":"[ULID]","ts":"[TIMESTAMP]"} {"event":"edge.selected","id":"[EVENT_ID]","properties":{"condition":null,"from_node":"start","is_jump":false,"label":null,"reason":"unconditional","stage_status":"success","to_node":"run_tests"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} diff --git a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs index 020998d1d..4afdda377 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs @@ -68,8 +68,7 @@ fn rewind_list_prints_timeline_for_completed_git_run() { ----- stdout ----- ----- stderr ----- @ Node Details - @1 step_one - @2 step_two + @1 step_one "); } @@ -94,8 +93,8 @@ fn rewind_target_updates_metadata_and_resume_hint() { exit_code: 0 ----- stdout ----- ----- stderr ----- - Rewound metadata branch to @1 (step_one) - Rewound run branch fabro/run/[ULID] to [SHA] + Rewound metadata branch to @1 (start) + Warning: checkpoint @1 has no git_commit_sha; run branch not moved To resume: fabro resume [RUN_PREFIX] "); diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index c7066fbae..f70032f53 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -395,6 +395,13 @@ fn json_run_implies_auto_approve_for_human_gates() { "run_id": "[ULID]", "ts": "[TIMESTAMP]" }, + { + "event": "run.running", + "id": "[EVENT_ID]", + "properties": {}, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" + }, { "event": "sandbox.ready", "id": "[EVENT_ID]", @@ -429,13 +436,6 @@ fn json_run_implies_auto_approve_for_human_gates() { "run_id": "[ULID]", "ts": "[TIMESTAMP]" }, - { - "event": "run.running", - "id": "[EVENT_ID]", - "properties": {}, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, { "event": "stage.started", "id": "[EVENT_ID]", diff --git a/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs b/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs index 043b540cb..53b5545ce 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs @@ -51,26 +51,29 @@ fn store_dump_exports_completed_run_snapshot() { success: true exit_code: 0 ----- stdout ----- - Exported 14 files for run [ULID] to [TEMP_DIR]/export + Exported 17 files for run [ULID] to [TEMP_DIR]/export ----- stderr ----- "); - assert_snapshot!(dump_file_summary(&output_dir), @r###" + assert_snapshot!(dump_file_summary(&output_dir), @" checkpoint.json - checkpoints/0001.json - checkpoints/0002.json - checkpoints/0003.json + checkpoints/0012.json + checkpoints/0016.json + checkpoints/0020.json conclusion.json events.jsonl graph.fabro + nodes/exit/visit-1/status.json + nodes/report/visit-1/response.md nodes/report/visit-1/status.json + nodes/run_tests/visit-1/response.md nodes/run_tests/visit-1/status.json nodes/start/visit-1/status.json run.json sandbox.json start.json status.json - "###); + "); } #[test] diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index 302daaa1c..0c8d4b6bf 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -1,8 +1,6 @@ use crate::NodeVisitRef; pub(crate) const INIT_KEY: &str = "_init.json"; -pub(crate) const RETRO_PROMPT_KEY: &str = "retro/prompt.md"; -pub(crate) const RETRO_RESPONSE_KEY: &str = "retro/response.md"; pub(crate) const EVENTS_PREFIX: &str = "events/"; pub(crate) const ARTIFACT_VALUES_PREFIX: &str = "artifacts/values/"; pub(crate) const ARTIFACT_NODES_PREFIX: &str = "artifacts/nodes/"; @@ -11,62 +9,6 @@ pub(crate) fn init() -> &'static str { INIT_KEY } -pub(crate) fn node_visit_prefix(node: &NodeVisitRef<'_>) -> String { - format!("nodes/{}/visit-{}", node.node_id, node.visit) -} - -pub(crate) fn node_prompt(node: &NodeVisitRef<'_>) -> String { - format!("{}/prompt.md", node_visit_prefix(node)) -} - -pub(crate) fn node_response(node: &NodeVisitRef<'_>) -> String { - format!("{}/response.md", node_visit_prefix(node)) -} - -pub(crate) fn node_status(node: &NodeVisitRef<'_>) -> String { - format!("{}/status.json", node_visit_prefix(node)) -} - -pub(crate) fn node_outcome(node: &NodeVisitRef<'_>) -> String { - format!("{}/outcome.json", node_visit_prefix(node)) -} - -pub(crate) fn node_provider_used(node: &NodeVisitRef<'_>) -> String { - format!("{}/provider_used.json", node_visit_prefix(node)) -} - -pub(crate) fn node_diff(node: &NodeVisitRef<'_>) -> String { - format!("{}/diff.patch", node_visit_prefix(node)) -} - -pub(crate) fn node_script_invocation(node: &NodeVisitRef<'_>) -> String { - format!("{}/script_invocation.json", node_visit_prefix(node)) -} - -pub(crate) fn node_script_timing(node: &NodeVisitRef<'_>) -> String { - format!("{}/script_timing.json", node_visit_prefix(node)) -} - -pub(crate) fn node_parallel_results(node: &NodeVisitRef<'_>) -> String { - format!("{}/parallel_results.json", node_visit_prefix(node)) -} - -pub(crate) fn node_stdout(node: &NodeVisitRef<'_>) -> String { - format!("{}/stdout.log", node_visit_prefix(node)) -} - -pub(crate) fn node_stderr(node: &NodeVisitRef<'_>) -> String { - format!("{}/stderr.log", node_visit_prefix(node)) -} - -pub(crate) fn retro_prompt() -> &'static str { - RETRO_PROMPT_KEY -} - -pub(crate) fn retro_response() -> &'static str { - RETRO_RESPONSE_KEY -} - pub(crate) fn event_key(seq: u32, epoch_ms: i64) -> String { format!("{EVENTS_PREFIX}{seq:06}-{epoch_ms}.json") } @@ -96,10 +38,6 @@ pub(crate) fn parse_artifact_value_id(key: &str) -> Option { .map(ToString::to_string) } -pub(crate) fn parse_node_key(key: &str) -> Option<(String, u32, String)> { - parse_visit_scoped_key(key, "nodes/") -} - pub(crate) fn parse_node_asset_key(key: &str) -> Option<(String, u32, String)> { parse_visit_scoped_key(key, ARTIFACT_NODES_PREFIX) } @@ -123,40 +61,6 @@ mod tests { fn top_level_keys_match_spec() { assert_eq!(init(), "_init.json"); assert_eq!(event_key(7, 123), "events/000007-123.json"); - assert_eq!(retro_prompt(), "retro/prompt.md"); - assert_eq!(retro_response(), "retro/response.md"); - } - - #[test] - fn node_keys_match_spec() { - let node = NodeVisitRef { - node_id: "plan", - visit: 3, - }; - assert_eq!(node_visit_prefix(&node), "nodes/plan/visit-3"); - assert_eq!(node_prompt(&node), "nodes/plan/visit-3/prompt.md"); - assert_eq!(node_response(&node), "nodes/plan/visit-3/response.md"); - assert_eq!(node_status(&node), "nodes/plan/visit-3/status.json"); - assert_eq!(node_outcome(&node), "nodes/plan/visit-3/outcome.json"); - assert_eq!( - node_provider_used(&node), - "nodes/plan/visit-3/provider_used.json" - ); - assert_eq!(node_diff(&node), "nodes/plan/visit-3/diff.patch"); - assert_eq!( - node_script_invocation(&node), - "nodes/plan/visit-3/script_invocation.json" - ); - assert_eq!( - node_script_timing(&node), - "nodes/plan/visit-3/script_timing.json" - ); - assert_eq!( - node_parallel_results(&node), - "nodes/plan/visit-3/parallel_results.json" - ); - assert_eq!(node_stdout(&node), "nodes/plan/visit-3/stdout.log"); - assert_eq!(node_stderr(&node), "nodes/plan/visit-3/stderr.log"); } #[test] @@ -184,10 +88,6 @@ mod tests { parse_artifact_value_id("artifacts/values/summary.json"), Some("summary".to_string()) ); - assert_eq!( - parse_node_key("nodes/plan/visit-3/status.json"), - Some(("plan".to_string(), 3, "status.json".to_string())) - ); assert_eq!( parse_node_asset_key("artifacts/nodes/code/visit-2/src/main.rs"), Some(("code".to_string(), 2, "src/main.rs".to_string())) @@ -201,7 +101,6 @@ mod tests { parse_artifact_value_id("artifacts/values/summary.txt"), None ); - assert_eq!(parse_node_key("nodes/plan/status.json"), None); assert_eq!( parse_node_asset_key("artifacts/nodes/code/status.json"), None diff --git a/lib/crates/fabro-store/src/memory.rs b/lib/crates/fabro-store/src/memory.rs index a720792e1..fe77181ad 100644 --- a/lib/crates/fabro-store/src/memory.rs +++ b/lib/crates/fabro-store/src/memory.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; @@ -14,10 +14,10 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; use crate::run_state::EventProjectionCache; use crate::{ - CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeOutcomeRecord, NodeSnapshot, - NodeVisitRef, Result, RunState, RunSummary, StoreError, + CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeVisitRef, Result, RunState, + RunSummary, StoreError, }; -use fabro_types::{NodeStatusRecord, RunId}; +use fabro_types::RunId; #[derive(Debug, Default)] pub struct InMemoryStore { @@ -79,24 +79,6 @@ impl InMemoryRunStore { .map_err(Into::into) } - async fn put_text(&self, key: String, value: &str) { - self.data - .lock() - .await - .insert(key, value.as_bytes().to_vec()); - } - - async fn get_text(&self, key: &str) -> Result> { - let bytes = self.data.lock().await.get(key).cloned(); - bytes - .map(|value| { - String::from_utf8(value).map_err(|err| { - StoreError::Other(format!("stored text is not valid UTF-8: {err}")) - }) - }) - .transpose() - } - async fn put_bytes(&self, key: String, value: &[u8]) { self.data.lock().await.insert(key, value.to_vec()); } @@ -109,44 +91,6 @@ impl InMemoryRunStore { self.data.lock().await.clone() } - #[allow(clippy::unused_self)] - fn build_node_snapshot_from_data( - &self, - data: &BTreeMap>, - node: &NodeVisitRef<'_>, - ) -> Result { - Ok(NodeSnapshot { - node_id: node.node_id.to_string(), - visit: node.visit, - prompt: read_text(data, &keys::node_prompt(node))?, - response: read_text(data, &keys::node_response(node))?, - status: read_json(data, &keys::node_status(node))?, - outcome: read_json(data, &keys::node_outcome(node))?, - provider_used: read_json(data, &keys::node_provider_used(node))?, - diff: read_text(data, &keys::node_diff(node))?, - script_invocation: read_json(data, &keys::node_script_invocation(node))?, - script_timing: read_json(data, &keys::node_script_timing(node))?, - parallel_results: read_json(data, &keys::node_parallel_results(node))?, - stdout: read_text(data, &keys::node_stdout(node))?, - stderr: read_text(data, &keys::node_stderr(node))?, - }) - } - - async fn list_node_ids_inner(&self) -> Vec { - let data = self.snapshot_data().await; - let mut node_ids = BTreeSet::new(); - for key in data.keys() { - if let Some((node_id, _, _)) = keys::parse_node_key(key) { - node_ids.insert(node_id); - continue; - } - if let Some((node_id, _, _)) = keys::parse_node_asset_key(key) { - node_ids.insert(node_id); - } - } - node_ids.into_iter().collect() - } - async fn list_events_from_inner(&self, seq: u32) -> Result> { let data = self.snapshot_data().await; let mut events = Vec::new(); @@ -283,111 +227,6 @@ impl InMemoryStore { } impl InMemoryRunStore { - pub async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> { - self.put_text(keys::node_prompt(node), prompt).await; - Ok(()) - } - - pub async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> { - self.put_text(keys::node_response(node), response).await; - Ok(()) - } - - pub async fn put_node_status( - &self, - node: &NodeVisitRef<'_>, - status: &NodeStatusRecord, - ) -> Result<()> { - self.put_json(keys::node_status(node), status).await - } - - pub async fn put_node_outcome( - &self, - node: &NodeVisitRef<'_>, - outcome: &NodeOutcomeRecord, - ) -> Result<()> { - self.put_json(keys::node_outcome(node), outcome).await - } - - pub async fn put_node_provider_used( - &self, - node: &NodeVisitRef<'_>, - provider_used: &serde_json::Value, - ) -> Result<()> { - self.put_json(keys::node_provider_used(node), provider_used) - .await - } - - pub async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> { - self.put_text(keys::node_diff(node), diff).await; - Ok(()) - } - - pub async fn put_node_script_invocation( - &self, - node: &NodeVisitRef<'_>, - invocation: &serde_json::Value, - ) -> Result<()> { - self.put_json(keys::node_script_invocation(node), invocation) - .await - } - - pub async fn put_node_script_timing( - &self, - node: &NodeVisitRef<'_>, - timing: &serde_json::Value, - ) -> Result<()> { - self.put_json(keys::node_script_timing(node), timing).await - } - - pub async fn put_node_parallel_results( - &self, - node: &NodeVisitRef<'_>, - results: &serde_json::Value, - ) -> Result<()> { - self.put_json(keys::node_parallel_results(node), results) - .await - } - - pub async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - self.put_text(keys::node_stdout(node), log).await; - Ok(()) - } - - pub async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - self.put_text(keys::node_stderr(node), log).await; - Ok(()) - } - - pub async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result { - let data = self.snapshot_data().await; - self.build_node_snapshot_from_data(&data, node) - } - - pub async fn list_node_visits(&self, node_id: &str) -> Result> { - let data = self.snapshot_data().await; - let mut visits = BTreeSet::new(); - for key in data.keys() { - let Some((current_node_id, visit, _)) = keys::parse_node_key(key) else { - continue; - }; - if current_node_id == node_id { - visits.insert(visit); - } - } - Ok(visits.into_iter().collect()) - } - - pub async fn list_node_ids(&self) -> Result> { - Ok(self.list_node_ids_inner().await) - } - - pub async fn reset_for_rewind(&self) -> Result<()> { - let mut data = self.data.lock().await; - data.retain(|key, _| key == keys::init() || key.starts_with(keys::EVENTS_PREFIX)); - Ok(()) - } - pub async fn append_event(&self, payload: &EventPayload) -> Result { payload.validate(&self.run_id)?; @@ -445,25 +284,6 @@ impl InMemoryRunStore { Ok(Box::pin(UnboundedReceiverStream::new(receiver).map(Ok))) } - pub async fn put_retro_prompt(&self, text: &str) -> Result<()> { - self.put_text(keys::retro_prompt().to_string(), text).await; - Ok(()) - } - - pub async fn get_retro_prompt(&self) -> Result> { - self.get_text(keys::retro_prompt()).await - } - - pub async fn put_retro_response(&self, text: &str) -> Result<()> { - self.put_text(keys::retro_response().to_string(), text) - .await; - Ok(()) - } - - pub async fn get_retro_response(&self) -> Result> { - self.get_text(keys::retro_response()).await - } - pub async fn put_artifact_value( &self, artifact_id: &str, @@ -538,25 +358,6 @@ fn matches_query(created_at: &DateTime, query: &ListRunsQuery) -> bool { true } -fn read_json( - data: &BTreeMap>, - key: &str, -) -> Result> { - data.get(key) - .map(|value| serde_json::from_slice(value)) - .transpose() - .map_err(Into::into) -} - -fn read_text(data: &BTreeMap>, key: &str) -> Result> { - data.get(key) - .map(|value| { - String::from_utf8(value.clone()) - .map_err(|err| StoreError::Other(format!("stored text is not valid UTF-8: {err}"))) - }) - .transpose() -} - #[cfg(test)] mod tests { use super::*; @@ -696,24 +497,6 @@ mod tests { } } - fn sample_node_status() -> NodeStatusRecord { - NodeStatusRecord { - status: StageStatus::PartialSuccess, - notes: Some("captured output".to_string()), - failure_reason: Some("minor lint".to_string()), - timestamp: dt("2026-03-27T12:12:00Z"), - } - } - - fn sample_node_outcome() -> NodeOutcomeRecord { - fabro_types::Outcome { - status: StageStatus::Success, - notes: Some("all good".to_string()), - files_touched: vec!["src/lib.rs".to_string()], - ..Default::default() - } - } - fn sample_pull_request() -> PullRequestRecord { PullRequestRecord { html_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), @@ -767,12 +550,6 @@ mod tests { node_id: "code", visit: 2, }; - let node_status = sample_node_status(); - let node_outcome = sample_node_outcome(); - let provider_used = serde_json::json!({"provider": "openai", "model": "gpt-5.4"}); - let script_invocation = serde_json::json!({"command": "cargo test"}); - let script_timing = serde_json::json!({"duration_ms": 3210}); - let parallel_results = serde_json::json!([{"node_id": "lint", "status": "success"}]); let pull_request = sample_pull_request(); run.append_event(&event_payload( @@ -852,27 +629,106 @@ mod tests { )) .await .unwrap(); - run.put_node_prompt(&node, "Plan the fix").await.unwrap(); - run.put_node_response(&node, "Implemented").await.unwrap(); - run.put_node_status(&node, &node_status).await.unwrap(); - run.put_node_outcome(&node, &node_outcome).await.unwrap(); - run.put_node_provider_used(&node, &provider_used) - .await - .unwrap(); - run.put_node_diff(&node, "diff --git a/src/lib.rs b/src/lib.rs") - .await - .unwrap(); - run.put_node_script_invocation(&node, &script_invocation) - .await - .unwrap(); - run.put_node_script_timing(&node, &script_timing) - .await - .unwrap(); - run.put_node_parallel_results(&node, ¶llel_results) - .await - .unwrap(); - run.put_node_stdout(&node, "ok").await.unwrap(); - run.put_node_stderr(&node, "").await.unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.1Z", + "stage.prompt", + Some("code"), + serde_json::json!({ + "visit": 2, + "text": "Plan the fix", + "mode": "prompt", + "provider": "openai", + "model": "gpt-5.4" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.2Z", + "command.started", + Some("code"), + serde_json::json!({ + "visit": 2, + "command": "cargo test" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.3Z", + "command.completed", + Some("code"), + serde_json::json!({ + "visit": 2, + "stdout": "ok", + "stderr": "", + "exit_code": 0 + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.4Z", + "checkpoint.completed", + Some("code"), + serde_json::json!({ + "status": "success", + "ordinal": 2, + "current_node": checkpoint.current_node, + "completed_nodes": checkpoint.completed_nodes, + "node_retries": checkpoint.node_retries, + "context_values": checkpoint.context_values, + "node_outcomes": checkpoint.node_outcomes, + "next_node_id": checkpoint.next_node_id, + "git_commit_sha": checkpoint.git_commit_sha, + "node_visits": checkpoint.node_visits, + "diff": "diff --git a/src/lib.rs b/src/lib.rs" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.5Z", + "parallel.completed", + Some("code"), + serde_json::json!({ + "visit": 2, + "results": [{"node_id": "lint", "status": "success"}] + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.6Z", + "stage.completed", + Some("code"), + serde_json::json!({ + "visit": 2, + "status": "success", + "notes": "all good", + "response": "Implemented", + "files_touched": ["src/lib.rs"] + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.7Z", + "retro.started", + None, + serde_json::json!({ + "prompt": "How did it go?" + }), + )) + .await + .unwrap(); run.append_event(&event_payload( "run-1", "2026-03-27T12:00:09Z", @@ -917,8 +773,6 @@ mod tests { )) .await .unwrap(); - run.put_retro_prompt("How did it go?").await.unwrap(); - run.put_retro_response("Smooth enough").await.unwrap(); run.put_artifact_value("summary", &serde_json::json!({"done": true})) .await .unwrap(); @@ -957,14 +811,8 @@ mod tests { let stored_sandbox = state.sandbox.as_ref().unwrap(); assert_eq!(stored_sandbox.provider, sandbox.provider); assert_eq!(stored_sandbox.working_directory, sandbox.working_directory); - assert_eq!( - run.get_retro_prompt().await.unwrap(), - Some("How did it go?".to_string()) - ); - assert_eq!( - run.get_retro_response().await.unwrap(), - Some("Smooth enough".to_string()) - ); + assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?")); + assert_eq!(state.retro_response.as_deref(), Some("Smooth enough")); assert_eq!( run.get_artifact_value("summary").await.unwrap(), Some(serde_json::json!({"done": true})) @@ -978,7 +826,26 @@ mod tests { Some("diff --git a/src/lib.rs b/src/lib.rs\n") ); assert_eq!(state.pull_request, Some(pull_request.clone())); - assert_eq!(run.list_node_ids().await.unwrap(), vec!["code".to_string()]); + assert_eq!(state.list_node_ids(), vec!["code".to_string()]); + let node_state = state + .node(&node) + .expect("node state should exist for code:2"); + assert_eq!(node_state.prompt.as_deref(), Some("Plan the fix")); + assert_eq!(node_state.response.as_deref(), Some("Implemented")); + assert_eq!(node_state.stdout.as_deref(), Some("ok")); + assert_eq!(node_state.stderr.as_deref(), Some("")); + assert_eq!( + node_state.diff.as_deref(), + Some("diff --git a/src/lib.rs b/src/lib.rs") + ); + assert_eq!( + node_state + .provider_used + .as_ref() + .and_then(|v| v.get("provider")) + .and_then(|v| v.as_str()), + Some("openai") + ); assert_eq!( run.list_assets(&node).await.unwrap(), vec!["src/lib.rs".to_string()] @@ -1358,9 +1225,6 @@ mod tests { node_id: "code", visit: 2, }; - run.put_node_prompt(&snapshot_node, "Plan the fix") - .await - .unwrap(); run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}") .await .unwrap(); @@ -1388,13 +1252,6 @@ mod tests { ("code".to_string(), 2, "src/lib.rs".to_string()) ] ); - assert_eq!( - run.list_node_ids().await.unwrap(), - vec!["artifact-only".to_string(), "code".to_string()] - ); - - let code_node = run.get_node(&snapshot_node).await.unwrap(); - assert_eq!(code_node.node_id, "code"); } #[tokio::test] @@ -1517,33 +1374,6 @@ mod tests { ); } - #[tokio::test] - async fn node_visit_storage_round_trips() { - let store = InMemoryStore::default(); - let run = store - .create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None) - .await - .unwrap(); - - let first = NodeVisitRef { - node_id: "code", - visit: 1, - }; - let second = NodeVisitRef { - node_id: "code", - visit: 2, - }; - run.put_node_prompt(&first, "first").await.unwrap(); - run.put_node_prompt(&second, "second").await.unwrap(); - run.put_node_status(&second, &sample_node_status()) - .await - .unwrap(); - - let node = run.get_node(&second).await.unwrap(); - assert_eq!(node.prompt, Some("second".to_string())); - assert_eq!(run.list_node_visits("code").await.unwrap(), vec![1, 2]); - } - #[tokio::test] async fn list_runs_filters_dates_and_tolerates_missing_status() { let store = InMemoryStore::default(); diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index cd6718b7c..962156946 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -383,8 +383,7 @@ mod tests { use bytes::Bytes; use fabro_types::{ - AttrValue, Graph, NodeStatusRecord, RunId, RunRecord, RunStatus, Settings, StageStatus, - StatusReason, fixtures, + AttrValue, Graph, RunId, RunRecord, RunStatus, Settings, StatusReason, fixtures, }; use object_store::memory::InMemory; use slatedb::config::Settings as SlateSettings; @@ -432,15 +431,6 @@ mod tests { } } - fn sample_node_status() -> NodeStatusRecord { - NodeStatusRecord { - status: StageStatus::Success, - notes: Some("done".to_string()), - failure_reason: None, - timestamp: dt("2026-03-27T12:12:00Z"), - } - } - fn event_payload( run_id: &str, ts: &str, diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index f7e34fc9e..884c76601 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -639,7 +639,7 @@ mod tests { .unwrap(); run.append_event(&event( "checkpoint.completed", - serde_json::json!({"diff": "diff --git a/story.txt b/story.txt", "ordinal": 1}), + serde_json::json!({"diff": "diff --git a/story.txt b/story.txt", "ordinal": 1, "current_node": "work", "node_visits": {"work": 2}}), )) .await .unwrap(); diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index f34f2d90a..ac511d751 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -719,14 +719,17 @@ mod tests { .unwrap(); logger.flush().await; - let snapshot = run_store - .get_node(&NodeVisitRef { + let state = run_store.state().await.unwrap(); + let node_state = state + .node(&NodeVisitRef { node_id: "step", visit: 1, }) - .await .unwrap(); - assert_eq!(snapshot.provider_used.unwrap()["provider"], "openai"); + assert_eq!( + node_state.provider_used.as_ref().unwrap()["provider"], + "openai" + ); } #[test] diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 98af0958c..4c4872584 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -4,7 +4,6 @@ use std::time::Instant; use async_trait::async_trait; use fabro_agent::{Sandbox, WorktreeOptions, WorktreeSandbox}; -use fabro_store::NodeVisitRef; use fabro_types::RunId; use tokio::sync::Semaphore; @@ -713,14 +712,14 @@ mod tests { .await .unwrap(); - let snapshot = run_store - .get_node(&NodeVisitRef { + let state = run_store.state().await.unwrap(); + let node_state = state + .node(&fabro_store::NodeVisitRef { node_id: "par", visit: 1, }) - .await .unwrap(); - let results = snapshot.parallel_results.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 8c78d6b03..6c5de318f 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -377,14 +377,14 @@ mod tests { .unwrap(); logger.flush().await; - let snapshot = run_store - .get_node(&NodeVisitRef { + let state = run_store.state().await.unwrap(); + let node_state = state + .node(&NodeVisitRef { node_id: "classify", visit: 1, }) - .await .unwrap(); - assert_eq!(snapshot.provider_used.unwrap()["mode"], "prompt"); + assert_eq!(node_state.provider_used.as_ref().unwrap()["mode"], "prompt"); } struct OneShotCapturingBackend {