diff --git a/docs/execution/checkpoints.mdx b/docs/execution/checkpoints.mdx index deed1b105..74771945b 100644 --- a/docs/execution/checkpoints.mdx +++ b/docs/execution/checkpoints.mdx @@ -50,7 +50,7 @@ After each node, the metadata branch is updated with: - **`run.json`** — Refreshed projection snapshot with the new current checkpoint - **`stages/{node_id}@{visit}/...`** — Per-stage execution trace files (prompts, responses, status, diffs, stdout/stderr, and tool metadata) -- **`retro/*.md`** — Retro prompt/response text when present +- **`stages/retro/*.md`** — Retro prompt/response text when present ## What's in a checkpoint diff --git a/docs/execution/retros.mdx b/docs/execution/retros.mdx index d0b5f6b63..879190803 100644 --- a/docs/execution/retros.mdx +++ b/docs/execution/retros.mdx @@ -143,4 +143,4 @@ Retros are also available via the REST API. See the [list retros](/api-reference ## Storage -Retros are stored in durable run state. If you need files on disk, `fabro store dump` materializes retro text under `retro/` alongside `run.json`, stage files, and the rest of the exported run data. +Retros are stored in durable run state. If you need files on disk, `fabro store dump` materializes retro text under `stages/retro/` alongside `run.json`, stage files, and the rest of the exported run data. diff --git a/docs/reference/run-directory.mdx b/docs/reference/run-directory.mdx index e88d11a91..d707afa35 100644 --- a/docs/reference/run-directory.mdx +++ b/docs/reference/run-directory.mdx @@ -38,14 +38,14 @@ Reconstructed metadata branches and `fabro store dump` exports now use the same - `run.json` for the current projection snapshot, including the current checkpoint - `graph.fabro` for workflow source -- `retro/*.md` for retro prompt/response text +- `stages/retro/*.md` for retro prompt/response text - `stages/{node_id}@{visit}/...` for per-stage prompt, response, status, diff, stdout, and stderr files `fabro store dump` adds export-only history surfaces on top of that shared layout: - `events.jsonl` for the durable event stream - `checkpoints/*.json` for checkpoint history snapshots -- `artifacts/nodes/{node_id}/visit-{n}/...` for exported artifact payloads +- `artifacts/{node_id}@{visit}/...` for exported artifact payloads ## Browsing runs diff --git a/lib/crates/fabro-checkpoint/src/metadata.rs b/lib/crates/fabro-checkpoint/src/metadata.rs index c75f6a0ae..c358863b3 100644 --- a/lib/crates/fabro-checkpoint/src/metadata.rs +++ b/lib/crates/fabro-checkpoint/src/metadata.rs @@ -390,12 +390,12 @@ mod tests { store .write_snapshot( &run_id, - &[("retro/prompt.md", b"how did it go?")], + &[("stages/retro/prompt.md", b"how did it go?")], "finalize run", ) .unwrap(); - let data = branch_entry(dir.path(), &run_id, "retro/prompt.md"); + let data = branch_entry(dir.path(), &run_id, "stages/retro/prompt.md"); assert_eq!(data, b"how did it go?"); let spec = MetadataStore::read_run_spec(dir.path(), &run_id) diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 3e27dbcde..700e94783 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -783,11 +783,11 @@ mod tests { ); assert_eq!( - std::fs::read_to_string(output.path().join("retro/prompt.md")).unwrap(), + std::fs::read_to_string(output.path().join("stages/retro/prompt.md")).unwrap(), "How did it go?" ); assert_eq!( - std::fs::read_to_string(output.path().join("retro/response.md")).unwrap(), + std::fs::read_to_string(output.path().join("stages/retro/response.md")).unwrap(), "Smooth enough" ); @@ -815,19 +815,14 @@ mod tests { assert!(!output.path().join("blobs").exists()); assert_eq!( - std::fs::read( - output - .path() - .join("artifacts/nodes/code/visit-2/src/lib.rs") - ) - .unwrap(), + std::fs::read(output.path().join("artifacts/code@2/src/lib.rs")).unwrap(), b"fn main() {}" ); assert_eq!( std::fs::read( output .path() - .join("artifacts/nodes/artifact-only/visit-7/logs/output.txt") + .join("artifacts/artifact-only@7/logs/output.txt") ) .unwrap(), b"hello" 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 6f01b2b50..1fedcf9e0 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs @@ -254,8 +254,7 @@ include = ["assets/**"] "run export should hydrate blob refs\n{run_json}" ); assert_eq!( - fs::read_to_string(output_dir.join("artifacts/nodes/big/visit-1/assets/shared/report.txt")) - .unwrap(), + fs::read_to_string(output_dir.join("artifacts/big@1/assets/shared/report.txt")).unwrap(), "exported" ); } diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index b0d776791..55d2f0120 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -1536,16 +1536,8 @@ async fn attach_events( }; let stream = - BroadcastStream::new(state.global_event_tx.subscribe()).filter_map(move |result| { - match result { - Ok(event) => { - if !event_matches_run_filter(&event, run_filter.as_ref()) { - return None; - } - sse_event_from_store(&event).map(Ok::) - } - Err(_) => None, - } + filtered_global_events(state.global_event_tx.subscribe(), run_filter).filter_map(|event| { + sse_event_from_store(&event).map(Ok::) }); Sse::new(stream) @@ -1553,6 +1545,16 @@ async fn attach_events( .into_response() } +fn filtered_global_events( + event_rx: broadcast::Receiver, + run_filter: Option>, +) -> impl tokio_stream::Stream { + BroadcastStream::new(event_rx).filter_map(move |result| match result { + Ok(event) if event_matches_run_filter(&event, run_filter.as_ref()) => Some(event), + Ok(_) | Err(_) => None, + }) +} + struct PrunePlan { run_ids: Vec, rows: Vec, @@ -7322,11 +7324,13 @@ mod tests { use axum::body::Body; use axum::http::{Request, header}; + use chrono::Utc; use fabro_interview::{AnswerValue, ControlInterviewer, Interviewer, Question, QuestionType}; use fabro_model::Provider; use fabro_types::settings::ServerAuthMethod; use fabro_types::{InterviewQuestionRecord, InterviewQuestionType, RunBlobId, RunId, fixtures}; use serde_json::json; + use tokio_stream::StreamExt as _; use tower::ServiceExt; use super::*; @@ -8116,6 +8120,27 @@ allowed_usernames = ["octocat"] run_store.append_event(&payload).await.unwrap(); } + fn test_event_envelope(seq: u32, run_id: RunId, body: EventBody) -> EventEnvelope { + EventEnvelope { + seq, + event: RunEvent { + id: format!("evt-{seq}"), + ts: Utc::now(), + run_id, + node_id: None, + node_label: None, + stage_id: None, + parallel_group_id: None, + parallel_branch_id: None, + session_id: None, + parent_session_id: None, + tool_call_id: None, + actor: None, + body, + }, + } + } + #[tokio::test] async fn test_model_unknown_returns_404() { let app = test_app_with(); @@ -11166,6 +11191,36 @@ timeout = "30s" assert!(matches!(sandbox_id, "sb-first" | "sb-second")); } + #[tokio::test] + async fn filtered_global_events_streams_only_matching_run_ids() { + let run_one = fixtures::RUN_1; + let run_two = fixtures::RUN_2; + let (event_tx, _) = broadcast::channel(8); + + let stream = filtered_global_events(event_tx.subscribe(), Some(HashSet::from([run_one]))); + + event_tx + .send(test_event_envelope( + 1, + run_two, + EventBody::RunQueued(Default::default()), + )) + .unwrap(); + event_tx + .send(test_event_envelope( + 2, + run_one, + EventBody::RunQueued(Default::default()), + )) + .unwrap(); + drop(event_tx); + + let events = stream.collect::>().await; + assert_eq!(events.len(), 1); + assert_eq!(events[0].seq, 2); + assert_eq!(events[0].event.run_id, run_one); + } + #[test] fn validate_github_slug_accepts_real_names() { assert!(super::validate_github_slug("owner", "anthropic", 39).is_ok()); diff --git a/lib/crates/fabro-server/tests/it/api/system.rs b/lib/crates/fabro-server/tests/it/api/system.rs index bcf066220..184eb0625 100644 --- a/lib/crates/fabro-server/tests/it/api/system.rs +++ b/lib/crates/fabro-server/tests/it/api/system.rs @@ -4,7 +4,6 @@ )] use std::path::PathBuf; -use std::time::Duration; use axum::body::Body; use axum::http::{Request, StatusCode}; @@ -13,9 +12,7 @@ use fabro_types::RunId; use fabro_types::settings::SettingsLayer; use fabro_types::settings::interp::InterpString; use fabro_types::settings::server::{ServerLayer, ServerStorageLayer}; -use http_body_util::BodyExt; use tempfile::tempdir; -use tokio::time::timeout; use tower::ServiceExt; use crate::helpers::{ @@ -263,22 +260,20 @@ async fn prune_runs_supports_dry_run_and_deletion() { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn attach_events_streams_only_matching_run_ids() { +async fn attach_events_returns_sse_stream() { let (_temp, settings, _storage_dir) = temp_storage_settings(); let app = test_app_with_scheduler(test_app_state_with_options(settings, 5)); - - let run_one = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await; - let run_two = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await; + let run_id = RunId::new(); let request = Request::builder() .method("GET") - .uri(api(&format!("/attach?run_id={run_one}"))) + .uri(api(&format!("/attach?run_id={run_id}"))) .body(Body::empty()) .unwrap(); let response = checked_response( app.clone().oneshot(request).await.unwrap(), StatusCode::OK, - format!("GET /api/v1/attach?run_id={run_one}"), + format!("GET /api/v1/attach?run_id={run_id}"), ) .await; let content_type = response @@ -288,27 +283,4 @@ async fn attach_events_streams_only_matching_run_ids() { .to_str() .unwrap(); assert!(content_type.contains("text/event-stream")); - - start_run(&app, &run_one).await; - start_run(&app, &run_two).await; - - let mut body = response.into_body(); - let mut sse_data = String::new(); - while let Ok(Some(Ok(frame))) = timeout(Duration::from_secs(2), body.frame()).await { - if let Some(data) = frame.data_ref() { - sse_data.push_str(&String::from_utf8_lossy(data)); - if sse_data.contains(&run_one) { - break; - } - } - } - - assert!( - sse_data.contains(&run_one), - "expected filtered stream data: {sse_data}" - ); - assert!( - !sse_data.contains(&run_two), - "filtered stream should exclude non-matching run ids: {sse_data}" - ); } diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index c9013bda7..573897c47 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -120,10 +120,13 @@ impl RunDump { } if let Some(prompt) = state.retro_prompt.as_ref() { - entries.push(RunDumpEntry::text("retro/prompt.md", prompt.clone())); + entries.push(RunDumpEntry::text("stages/retro/prompt.md", prompt.clone())); } if let Some(response) = state.retro_response.as_ref() { - entries.push(RunDumpEntry::text("retro/response.md", response.clone())); + entries.push(RunDumpEntry::text( + "stages/retro/response.md", + response.clone(), + )); } Self { entries } @@ -399,12 +402,10 @@ fn replace_blob_refs_in_value( } fn artifact_dump_path(stage_id: &StageId, filename: &str) -> Result { - let node_id_segment = validate_single_path_segment("node id", stage_id.node_id())?; + validate_single_path_segment("node id", stage_id.node_id())?; let filename_path = validate_relative_path("artifact filename", filename)?; Ok(PathBuf::from("artifacts") - .join("nodes") - .join(node_id_segment) - .join(format!("visit-{}", stage_id.visit())) + .join(stage_id.to_string()) .join(filename_path)) } @@ -540,8 +541,8 @@ mod tests { assert!(paths.contains(&"run.json")); assert!(paths.contains(&"graph.fabro")); - assert!(paths.contains(&"retro/prompt.md")); - assert!(paths.contains(&"retro/response.md")); + assert!(paths.contains(&"stages/retro/prompt.md")); + assert!(paths.contains(&"stages/retro/response.md")); assert!(paths.contains(&"stages/build@2/prompt.md")); assert!(paths.contains(&"stages/build@2/response.md")); assert!(paths.contains(&"stages/build@2/status.json"));