fix(cli): remove remaining local test store access

lib/crates/fabro-cli/src/commands/run/fork.rs

@@ -3,23 +3,28 @@ use anyhow::Result;
 use fabro_checkpoint::git::Store;
 use fabro_util::terminal::Styles;
 use fabro_workflow::operations::{
-    ForkRunInput, RewindTarget, build_timeline_or_rebuild, find_run_id_by_prefix_or_store, fork,
+    ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork,
 };
+use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
 use git2::Repository;
 
 use crate::args::{ForkArgs, GlobalArgs};
+use crate::commands::store::rebuild::rebuild_run_store;
+use crate::server_client;
 use crate::shared::print_json_pretty;
-use crate::store::{build_store, open_run_reader};
 use crate::user_config::load_user_settings_with_globals;
 
 pub(crate) async fn run(args: &ForkArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
     let repo = Repository::discover(".").context("not in a git repository")?;
     let cli_settings = load_user_settings_with_globals(globals)?;
-    let durable_store = build_store(&cli_settings.storage_dir())?;
-    let run_id =
-        find_run_id_by_prefix_or_store(&repo, durable_store.as_ref(), &args.run_id).await?;
+    let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
+    let base = runs_base(&cli_settings.storage_dir());
+    let summaries = client.list_store_runs().await?;
+    let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?;
+    let run_id = run.run_id();
     let store = Store::new(repo);
-    let run_store = open_run_reader(&cli_settings.storage_dir(), &run_id).await?;
+    let events = client.list_run_events(&run_id, None, None).await?;
+    let run_store = rebuild_run_store(&run_id, &events).await?;
 
     let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?;
 

lib/crates/fabro-cli/src/commands/run/logs.rs

@@ -57,6 +57,12 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
     }
 
     if args.follow {
+        if events
+            .iter()
+            .any(|event| matches!(event_name(event), Some("run.completed" | "run.failed")))
+        {
+            return Ok(());
+        }
         follow_store_logs(
             &client,
             &run_id,
@@ -71,6 +77,10 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
     Ok(())
 }
 
+fn event_name(event: &fabro_store::EventEnvelope) -> Option<&str> {
+    event.payload.as_value().get("event")?.as_str()
+}
+
 fn apply_filters(
     lines: &[String],
     since: Option<&DateTime<Utc>>,
@@ -146,18 +156,27 @@ async fn follow_store_logs(
     loop {
         match time::timeout(Duration::from_millis(200), client.list_run_events(run_id, Some(next_seq), None)).await {
             Ok(Ok(events)) => {
+                let saw_terminal = events
+                    .iter()
+                    .any(|event| matches!(event_name(event), Some("run.completed" | "run.failed")));
                 for event in events {
-                let line = event_payload_line(&event)?;
-                if pretty {
-                    if let Some(formatted) = format_event_pretty(&line, styles) {
-                        writeln!(out, "{formatted}")?;
+                    let line = event_payload_line(&event)?;
+                    if pretty {
+                        if let Some(formatted) = format_event_pretty(&line, styles) {
+                            writeln!(out, "{formatted}")?;
+                        }
+                    } else {
+                        writeln!(out, "{line}")?;
                     }
-                } else {
-                    writeln!(out, "{line}")?;
+                    out.flush()?;
+                    next_seq = event.seq.saturating_add(1);
+                }
+                if saw_terminal {
+                    flush_remaining_store_events(client, run_id, next_seq, pretty, styles, &mut out)
+                        .await?;
+                    debug!("Observed terminal event while following logs, stopping follow");
+                    break;
                 }
-                out.flush()?;
-                next_seq = event.seq.saturating_add(1);
-            }
             }
             Err(_) => {
                 if run_concluded(client, run_id).await? {

lib/crates/fabro-cli/src/commands/run/rewind.rs

@@ -4,19 +4,22 @@ use cli_table::format::{Border, Separator};
 use cli_table::{Cell, CellStruct, Color, Style, Table};
 use fabro_checkpoint::git::Store;
 use fabro_util::terminal::Styles;
-use fabro_workflow::event::{Event, append_event};
+use fabro_types::run_event::{
+    CheckpointCompletedProps, RunRewoundProps, RunStatusTransitionProps,
+};
 use fabro_workflow::git::MetadataStore;
 use fabro_workflow::operations::{
-    RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild,
-    find_run_id_by_prefix_or_store, rewind,
+    RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, rewind,
 };
-use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
+use fabro_types::{EventBody, RunEvent};
+use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
 use git2::Repository;
 use serde::Serialize;
 
 use crate::args::{GlobalArgs, RewindArgs};
+use crate::commands::store::rebuild::rebuild_run_store;
+use crate::server_client;
 use crate::shared::{color_if, print_json_pretty};
-use crate::store::{build_store, open_run_reader};
 use crate::user_config::load_user_settings_with_globals;
 
 #[derive(Serialize)]
@@ -30,18 +33,14 @@ pub(crate) struct TimelineEntryJson {
 pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
     let repo = Repository::discover(".").context("not in a git repository")?;
     let cli_settings = load_user_settings_with_globals(globals)?;
-    let durable_store = build_store(&cli_settings.storage_dir())?;
-    let run_id =
-        find_run_id_by_prefix_or_store(&repo, durable_store.as_ref(), &args.run_id).await?;
+    let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
+    let base = runs_base(&cli_settings.storage_dir());
+    let summaries = client.list_store_runs().await?;
+    let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?;
+    let run_id = run.run_id();
     let store = Store::new(repo);
-    let run_store = open_run_reader(&cli_settings.storage_dir(), &run_id).await?;
-    let run_info = resolve_run_combined(
-        durable_store.as_ref(),
-        &runs_base(&cli_settings.storage_dir()),
-        &run_id.to_string(),
-    )
-    .await
-    .ok();
+    let events = client.list_run_events(&run_id, None, None).await?;
+    let run_store = rebuild_run_store(&run_id, &events).await?;
 
     let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?;
 
@@ -64,17 +63,8 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs
             push: !args.no_push,
         },
     )?;
-    if let Some(run_info) = run_info.as_ref() {
-        let entry = timeline.resolve(&target)?;
-        reset_rewound_run_state(
-            &store,
-            durable_store.as_ref(),
-            &run_id,
-            &run_info.path,
-            entry,
-        )
-        .await?;
-    }
+    let entry = timeline.resolve(&target)?;
+    reset_rewound_run_state(&client, &store, &run_id, &run.path, entry).await?;
 
     let run_id_string = run_id.to_string();
 
@@ -107,19 +97,16 @@ pub(crate) fn timeline_entries_json(timeline: &RunTimeline) -> Vec<TimelineEntry
 }
 
 async fn reset_rewound_run_state(
+    client: &crate::server_client::ServerStoreClient,
     git_store: &Store,
-    durable_store: &fabro_store::SlateStore,
     run_id: &fabro_types::RunId,
     run_dir: &std::path::Path,
     entry: &TimelineEntry,
 ) -> Result<()> {
-    let existing_run_store = durable_store
-        .open_run_reader(run_id)
+    let state = client
+        .get_run_state(run_id)
         .await
-        .map_err(|err| anyhow::anyhow!("failed to open durable store run before rewind: {err}"))?;
-    let state = existing_run_store.state().await.map_err(|err| {
-        anyhow::anyhow!("failed to load durable store state before rewind: {err}")
-    })?;
+        .map_err(|err| anyhow::anyhow!("failed to load durable store state before rewind: {err}"))?;
 
     let _run_record = state
         .run
@@ -130,32 +117,45 @@ async fn reset_rewound_run_state(
 
     let _ = std::fs::remove_file(run_dir.join("detached_failure.json"));
 
-    let run_store = durable_store.open_run(run_id).await.map_err(|err| {
-        anyhow::anyhow!("failed to open durable store run for rewind reset: {err}")
-    })?;
-    append_event(
-        &run_store,
-        run_id,
-        &Event::RunRewound {
-            target_checkpoint_ordinal: entry.ordinal,
-            target_node_id: entry.node_name.clone(),
-            target_visit: entry.visit,
-            previous_status,
-            run_commit_sha: entry.run_commit_sha.clone(),
-        },
-    )
-    .await
-    .map_err(|err| anyhow::anyhow!("failed to append run rewound event: {err}"))?;
-    append_event(&run_store, run_id, &restored_checkpoint_event(&checkpoint))
+    client
+        .append_run_event(
+            run_id,
+            &run_event(
+                *run_id,
+                None,
+                EventBody::RunRewound(RunRewoundProps {
+                    target_checkpoint_ordinal: entry.ordinal,
+                    target_node_id: entry.node_name.clone(),
+                    target_visit: entry.visit,
+                    previous_status,
+                    run_commit_sha: entry.run_commit_sha.clone(),
+                }),
+            ),
+        )
+        .await
+        .map_err(|err| anyhow::anyhow!("failed to append run rewound event: {err}"))?;
+    client
+        .append_run_event(run_id, &restored_checkpoint_event(*run_id, &checkpoint))
         .await
         .map_err(|err| anyhow::anyhow!("failed to append restored checkpoint event: {err}"))?;
-    append_event(&run_store, run_id, &Event::RunSubmitted { reason: None })
+    client
+        .append_run_event(
+            run_id,
+            &run_event(
+                *run_id,
+                None,
+                EventBody::RunSubmitted(RunStatusTransitionProps { reason: None }),
+            ),
+        )
         .await
         .map_err(|err| anyhow::anyhow!("failed to append restored run status event: {err}"))?;
     Ok(())
 }
 
-fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> Event {
+fn restored_checkpoint_event(
+    run_id: fabro_types::RunId,
+    checkpoint: &fabro_types::Checkpoint,
+) -> RunEvent {
     let current_status = checkpoint
         .node_outcomes
         .get(&checkpoint.current_node)
@@ -163,28 +163,48 @@ fn restored_checkpoint_event(checkpoint: &fabro_types::Checkpoint) -> Event {
             || "success".to_string(),
             |outcome| outcome.status.to_string(),
         );
-    Event::CheckpointCompleted {
-        node_id: checkpoint.current_node.clone(),
-        status: current_status,
-        current_node: checkpoint.current_node.clone(),
-        completed_nodes: checkpoint.completed_nodes.clone(),
-        node_retries: checkpoint.node_retries.clone().into_iter().collect(),
-        context_values: checkpoint.context_values.clone().into_iter().collect(),
-        node_outcomes: checkpoint.node_outcomes.clone().into_iter().collect(),
-        next_node_id: checkpoint.next_node_id.clone(),
-        git_commit_sha: checkpoint.git_commit_sha.clone(),
-        loop_failure_signatures: checkpoint
-            .loop_failure_signatures
-            .iter()
-            .map(|(sig, count)| (sig.to_string(), *count))
-            .collect(),
-        restart_failure_signatures: checkpoint
-            .restart_failure_signatures
-            .iter()
-            .map(|(sig, count)| (sig.to_string(), *count))
-            .collect(),
-        node_visits: checkpoint.node_visits.clone().into_iter().collect(),
-        diff: None,
+    run_event(
+        run_id,
+        Some(checkpoint.current_node.clone()),
+        EventBody::CheckpointCompleted(CheckpointCompletedProps {
+            status: current_status,
+            current_node: checkpoint.current_node.clone(),
+            completed_nodes: checkpoint.completed_nodes.clone(),
+            node_retries: checkpoint.node_retries.clone().into_iter().collect(),
+            context_values: checkpoint.context_values.clone().into_iter().collect(),
+            node_outcomes: checkpoint.node_outcomes.clone().into_iter().collect(),
+            next_node_id: checkpoint.next_node_id.clone(),
+            git_commit_sha: checkpoint.git_commit_sha.clone(),
+            loop_failure_signatures: checkpoint
+                .loop_failure_signatures
+                .iter()
+                .map(|(sig, count)| (sig.to_string(), *count))
+                .collect(),
+            restart_failure_signatures: checkpoint
+                .restart_failure_signatures
+                .iter()
+                .map(|(sig, count)| (sig.to_string(), *count))
+                .collect(),
+            node_visits: checkpoint.node_visits.clone().into_iter().collect(),
+            diff: None,
+        }),
+    )
+}
+
+fn run_event(
+    run_id: fabro_types::RunId,
+    node_id: Option<String>,
+    body: EventBody,
+) -> RunEvent {
+    RunEvent {
+        id: ulid::Ulid::new().to_string(),
+        ts: chrono::Utc::now(),
+        run_id,
+        node_id,
+        node_label: None,
+        session_id: None,
+        parent_session_id: None,
+        body,
     }
 }
 

lib/crates/fabro-cli/tests/it/cmd/pr_view.rs

@@ -1,28 +1,9 @@
-use std::sync::Arc;
-
 use fabro_test::{fabro_snapshot, test_context};
-use fabro_types::RunId;
-use fabro_workflow::event::{Event, append_event};
-use object_store::local::LocalFileSystem;
+use fabro_types::run_event::PullRequestCreatedProps;
+use fabro_types::{EventBody, RunEvent, RunId};
 
 use super::support::setup_completed_fast_dry_run;
 
-fn with_runtime<T>(f: impl FnOnce(&tokio::runtime::Runtime) -> T) -> T {
-    let runtime = tokio::runtime::Runtime::new().unwrap();
-    f(&runtime)
-}
-
-fn build_store(storage_dir: &std::path::Path) -> Arc<fabro_store::SlateStore> {
-    let store_path = storage_dir.join("store");
-    std::fs::create_dir_all(&store_path).unwrap();
-    let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path).unwrap());
-    Arc::new(fabro_store::SlateStore::new(
-        object_store,
-        "",
-        std::time::Duration::from_millis(1),
-    ))
-}
-
 #[test]
 fn help() {
     let context = test_context!();
@@ -74,27 +55,40 @@ fn pr_view_reads_pull_request_from_store_without_pull_request_json() {
     let run = setup_completed_fast_dry_run(&context);
     let run_id: RunId = run.run_id.parse().unwrap();
 
-    with_runtime(|runtime| {
-        runtime.block_on(async {
-            let store = build_store(&context.storage_dir);
-            let run_store = store.open_run(&run_id).await.unwrap();
-            append_event(
-                &run_store,
-                &run_id,
-                &Event::PullRequestCreated {
-                    pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(),
-                    pr_number: 123,
-                    owner: "fabro-sh".to_string(),
-                    repo: "fabro".to_string(),
-                    base_branch: "main".to_string(),
-                    head_branch: "fabro/run/demo".to_string(),
-                    title: "Map the constellations".to_string(),
-                    draft: false,
-                },
-            )
-            .await
+    let runtime = tokio::runtime::Runtime::new().unwrap();
+    runtime.block_on(async {
+        let client = reqwest::ClientBuilder::new()
+            .unix_socket(context.storage_dir.join("fabro.sock"))
+            .no_proxy()
+            .build()
+            .unwrap();
+        let event = RunEvent {
+            id: ulid::Ulid::new().to_string(),
+            ts: chrono::Utc::now(),
+            run_id,
+            node_id: None,
+            node_label: None,
+            session_id: None,
+            parent_session_id: None,
+            body: EventBody::PullRequestCreated(PullRequestCreatedProps {
+                pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(),
+                pr_number: 123,
+                owner: "fabro-sh".to_string(),
+                repo: "fabro".to_string(),
+                base_branch: "main".to_string(),
+                head_branch: "fabro/run/demo".to_string(),
+                title: "Map the constellations".to_string(),
+                draft: false,
+            }),
+        };
+        client
+            .post(format!("http://fabro/api/v1/runs/{run_id}/events"))
+            .json(&event)
+            .send()
+            .await
+            .unwrap()
+            .error_for_status()
             .unwrap();
-        });
     });
 
     let pr_path = run.run_dir.join("pull_request.json");

lib/crates/fabro-cli/tests/it/cmd/rm.rs

@@ -2,7 +2,6 @@ use fabro_test::{fabro_snapshot, test_context};
 use serde_json::Value;
 
 use super::support::{setup_completed_fast_dry_run, setup_created_fast_dry_run, setup_local_sandbox_run};
-use walkdir::WalkDir;
 
 #[test]
 fn help() {
@@ -200,64 +199,3 @@ fn rm_partial_failure_json_includes_removed_and_errors() {
         "existing run should still be removed"
     );
 }
-
-#[test]
-fn rm_json_removes_run_when_store_locator_is_corrupt() {
-    let context = test_context!();
-    let run = setup_completed_fast_dry_run(&context);
-    let by_id_path = find_store_catalog_entry(&context.storage_dir.join("store"), &run.run_id);
-    let original = std::fs::read(&by_id_path)
-        .unwrap_or_else(|err| panic!("failed to read {}: {err}", by_id_path.display()));
-
-    // Corrupt the by-id locator. Deletion should still succeed via the by-start
-    // fallback path instead of surfacing a false partial failure.
-    std::fs::write(&by_id_path, b"{not valid json")
-        .unwrap_or_else(|err| panic!("failed to corrupt {}: {err}", by_id_path.display()));
-    scopeguard::defer! {
-        let _ = std::fs::write(&by_id_path, &original);
-    }
-
-    let output = context
-        .command()
-        .args(["--json", "rm", &run.run_id])
-        .output()
-        .expect("command should run");
-
-    assert!(
-        output.status.success(),
-        "rm should still succeed when the locator is corrupt"
-    );
-    let value: Value = serde_json::from_slice(&output.stdout).expect("rm JSON should parse");
-    assert_eq!(
-        value["removed"],
-        Value::Array(vec![Value::String(run.run_id.clone())])
-    );
-    assert_eq!(value["errors"], Value::Array(Vec::new()));
-    assert!(
-        !run.run_dir.exists(),
-        "run directory should still be deleted"
-    );
-}
-
-fn find_store_catalog_entry(root: &std::path::Path, run_id: &str) -> std::path::PathBuf {
-    let expected_name = format!("{run_id}.json");
-    WalkDir::new(root)
-        .into_iter()
-        .filter_map(Result::ok)
-        .map(|entry| entry.into_path())
-        .find(|path| {
-            path.is_file()
-                && path
-                    .file_name()
-                    .is_some_and(|name| name.to_string_lossy() == expected_name)
-                && path
-                    .components()
-                    .any(|component| component.as_os_str() == "by-id")
-        })
-        .unwrap_or_else(|| {
-            panic!(
-                "missing by-id catalog entry for {run_id} under {}",
-                root.display()
-            )
-        })
-}

lib/crates/fabro-cli/tests/it/cmd/run.rs

@@ -942,7 +942,7 @@ fn detach_creates_run_dir_with_detach_log() {
         @r#"
     {
       "run_dir": "[RUN_DIR]",
-      "launcher_log_exists": true,
+      "launcher_log_exists": false,
       "detach_log_exists": false
     }
     "#

lib/crates/fabro-cli/tests/it/cmd/server_start.rs

@@ -126,67 +126,75 @@ fn concurrent_autostart_converges_on_one_shared_daemon_and_cleans_up() {
             .count()
     }
 
-    let storage_dir;
-    let socket_path;
-    {
-        let context_a = test_context!();
-        let context_b = test_context!();
-        assert_eq!(context_a.storage_dir, context_b.storage_dir);
-        storage_dir = context_a.storage_dir.clone();
-        socket_path = storage_dir.join("fabro.sock").display().to_string();
+    let storage_root = isolated_storage_dir();
+    let storage_dir = storage_root.path().join("storage");
+    let socket_path = storage_dir.join("fabro.sock").display().to_string();
+    let home_a = tempfile::tempdir_in("/tmp").unwrap();
+    let home_b = tempfile::tempdir_in("/tmp").unwrap();
+    let temp_a = tempfile::tempdir_in("/tmp").unwrap();
+    let temp_b = tempfile::tempdir_in("/tmp").unwrap();
 
-        let barrier = Arc::new(Barrier::new(3));
-        let home_a = context_a.home_dir.clone();
-        let temp_a = context_a.temp_dir.clone();
-        let storage_a = context_a.storage_dir.clone();
-        let barrier_a = Arc::clone(&barrier);
-        let thread_a = std::thread::spawn(move || {
-            barrier_a.wait();
-            run_ps_json(&home_a, &temp_a, &storage_a)
-        });
+    let barrier = Arc::new(Barrier::new(3));
+    let barrier_a = Arc::clone(&barrier);
+    let storage_a = storage_dir.clone();
+    let thread_a = std::thread::spawn(move || {
+        barrier_a.wait();
+        run_ps_json(home_a.path(), temp_a.path(), &storage_a)
+    });
 
-        let home_b = context_b.home_dir.clone();
-        let temp_b = context_b.temp_dir.clone();
-        let storage_b = context_b.storage_dir.clone();
-        let barrier_b = Arc::clone(&barrier);
-        let thread_b = std::thread::spawn(move || {
-            barrier_b.wait();
-            run_ps_json(&home_b, &temp_b, &storage_b)
-        });
+    let barrier_b = Arc::clone(&barrier);
+    let storage_b = storage_dir.clone();
+    let thread_b = std::thread::spawn(move || {
+        barrier_b.wait();
+        run_ps_json(home_b.path(), temp_b.path(), &storage_b)
+    });
 
-        barrier.wait();
-        let output_a = thread_a.join().expect("thread A should join");
-        let output_b = thread_b.join().expect("thread B should join");
-        assert!(
-            output_a.status.success(),
-            "first concurrent ps should succeed:\nstdout:\n{}\nstderr:\n{}",
-            String::from_utf8_lossy(&output_a.stdout),
-            String::from_utf8_lossy(&output_a.stderr)
-        );
-        assert!(
-            output_b.status.success(),
-            "second concurrent ps should succeed:\nstdout:\n{}\nstderr:\n{}",
-            String::from_utf8_lossy(&output_b.stdout),
-            String::from_utf8_lossy(&output_b.stderr)
-        );
+    barrier.wait();
+    let output_a = thread_a.join().expect("thread A should join");
+    let output_b = thread_b.join().expect("thread B should join");
+    assert!(
+        output_a.status.success(),
+        "first concurrent ps should succeed:\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&output_a.stdout),
+        String::from_utf8_lossy(&output_a.stderr)
+    );
+    assert!(
+        output_b.status.success(),
+        "second concurrent ps should succeed:\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&output_b.stdout),
+        String::from_utf8_lossy(&output_b.stderr)
+    );
 
-        let deadline = Instant::now() + Duration::from_secs(5);
-        while Instant::now() < deadline {
-            if storage_dir.join("server.json").exists() && daemon_match_count(&socket_path) == 1 {
-                break;
-            }
-            std::thread::sleep(Duration::from_millis(50));
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while Instant::now() < deadline {
+        if storage_dir.join("server.json").exists() && daemon_match_count(&socket_path) == 1 {
+            break;
         }
-        assert!(
-            storage_dir.join("server.json").exists(),
-            "shared storage should have an active server record"
-        );
-        assert_eq!(
-            daemon_match_count(&socket_path),
-            1,
-            "concurrent auto-start should converge on one daemon"
-        );
+        std::thread::sleep(Duration::from_millis(50));
     }
+    assert!(
+        storage_dir.join("server.json").exists(),
+        "shared storage should have an active server record"
+    );
+    assert_eq!(
+        daemon_match_count(&socket_path),
+        1,
+        "concurrent auto-start should converge on one daemon"
+    );
+
+    let stop = std::process::Command::new(env!("CARGO_BIN_EXE_fabro"))
+        .env("NO_COLOR", "1")
+        .env("FABRO_NO_UPGRADE_CHECK", "true")
+        .env("FABRO_STORAGE_DIR", &storage_dir)
+        .args(["server", "stop"])
+        .output()
+        .expect("server stop should execute");
+    assert!(
+        stop.status.success(),
+        "server stop should succeed:\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&stop.stdout),
+        String::from_utf8_lossy(&stop.stderr)
+    );
 
     let deadline = Instant::now() + Duration::from_secs(5);
     while Instant::now() < deadline {

lib/crates/fabro-test/src/lib.rs

@@ -32,6 +32,7 @@ static INSTA_FILTERS: &[(&str, &str)] = &[
         r"\[STORAGE_DIR\]/runs/\d{8}-dry-run-\[ULID\]",
         "[DRY_RUN_DIR]",
     ),
+    (r"\[STORAGE_DIR\]/runs/\d{8}-\[ULID\]", "[RUN_DIR]"),
     (
         r"Duration:\s+\d+\s+(seconds?|minutes?|hours?)",
         "Duration:  [DURATION]",