Merge remote-tracking branch 'origin/main'

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
 

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.

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
 

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)

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"

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

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::<Event, std::convert::Infallible>)
-                }
-                Err(_) => None,
-            }
+        filtered_global_events(state.global_event_tx.subscribe(), run_filter).filter_map(|event| {
+            sse_event_from_store(&event).map(Ok::<Event, std::convert::Infallible>)
         });
 
     Sse::new(stream)
@@ -1553,6 +1545,16 @@ async fn attach_events(
         .into_response()
 }
 
+fn filtered_global_events(
+    event_rx: broadcast::Receiver<EventEnvelope>,
+    run_filter: Option<HashSet<RunId>>,
+) -> impl tokio_stream::Stream<Item = EventEnvelope> {
+    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<RunId>,
     rows:             Vec<PruneRunEntry>,
@@ -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::<Vec<_>>().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());

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

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<PathBuf> {
-    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"));