fix(test): eliminate recovery, cancel, and label flakes

Stabilize the recovery scenario around rebuilt metadata timing and node
ordinals, make in-process run cancellation converge on a cancelled
reason, and keep the label assertion unit test out of the shared
TestContext session lifecycle.

lib/crates/fabro-cli/tests/it/scenario/recovery.rs

@@ -5,7 +5,7 @@ use fabro_checkpoint::branch::BranchStore;
 use fabro_checkpoint::git::Store as GitStore;
 use fabro_test::{fabro_snapshot, test_context};
 use fabro_types::Checkpoint;
-use fabro_workflow::operations::build_timeline;
+use fabro_workflow::operations::{RunTimeline, build_timeline};
 use git2::{Repository, Signature};
 
 use crate::support::unique_run_id;
@@ -58,16 +58,59 @@ fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint {
 }
 
 fn timeline_run_shas(repo_dir: &Path, run_id: &str) -> Vec<Option<String>> {
-    let repo = Repository::discover(repo_dir).unwrap();
-    let store = GitStore::new(repo);
-    build_timeline(&store, run_id)
-        .unwrap()
+    build_timeline_when_ready(repo_dir, run_id)
         .entries
         .into_iter()
         .map(|entry| entry.run_commit_sha)
         .collect()
 }
 
+fn timeline_node_names(repo_dir: &Path, run_id: &str) -> Vec<String> {
+    build_timeline_when_ready(repo_dir, run_id)
+        .entries
+        .into_iter()
+        .map(|entry| entry.node_name)
+        .collect()
+}
+
+fn build_timeline_when_ready(repo_dir: &Path, run_id: &str) -> RunTimeline {
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        let repo = Repository::discover(repo_dir).unwrap();
+        let store = GitStore::new(repo);
+        match build_timeline(&store, run_id) {
+            Ok(timeline) => return timeline,
+            Err(err) => {
+                assert!(
+                    std::time::Instant::now() < deadline,
+                    "timeline for {run_id} never became readable: {err}"
+                );
+                std::thread::sleep(std::time::Duration::from_millis(50));
+            }
+        }
+    }
+}
+
+fn delete_metadata_branch_when_ready(repo_dir: &Path, run_id: &str) {
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        let repo = Repository::discover(repo_dir).unwrap();
+        let mut reference = repo
+            .find_reference(&format!("refs/heads/fabro/meta/{run_id}"))
+            .unwrap();
+        match reference.delete() {
+            Ok(()) => return,
+            Err(err) => {
+                assert!(
+                    std::time::Instant::now() < deadline,
+                    "metadata branch for {run_id} never became writable: {err}"
+                );
+                std::thread::sleep(std::time::Duration::from_millis(50));
+            }
+        }
+    }
+}
+
 fn init_repo_with_workflow(repo_dir: &Path) {
     std::fs::write(repo_dir.join("README.md"), "recovery test\n").unwrap();
     std::fs::write(
@@ -142,12 +185,7 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
     filters.push((r"\b[0-9a-f]{7,40}\b".to_string(), "[SHA]".to_string()));
     filters.extend(context.filters());
 
-    Repository::discover(repo_dir.path())
-        .unwrap()
-        .find_reference(&format!("refs/heads/fabro/meta/{source_run_id}"))
-        .unwrap()
-        .delete()
-        .unwrap();
+    delete_metadata_branch_when_ready(repo_dir.path(), &source_run_id);
 
     assert!(
         list_metadata_run_ids(repo_dir.path()).is_empty(),
@@ -158,15 +196,14 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
     rewind_list.current_dir(repo_dir.path());
     rewind_list.args(["rewind", &source_run_id, "--list"]);
     rewind_list.timeout(std::time::Duration::from_secs(15));
-    fabro_snapshot!(filters.clone(), rewind_list, @"
-    success: true
-    exit_code: 0
-    ----- stdout -----
-    ----- stderr -----
-    @   Node   Details 
-     @1  plan           
-     @2  build
-    ");
+    rewind_list.assert().success();
+
+    let rebuilt_nodes = timeline_node_names(repo_dir.path(), &source_run_id);
+    assert_eq!(rebuilt_nodes.last().map(String::as_str), Some("build"));
+    assert!(
+        rebuilt_nodes.ends_with(&["plan".to_string(), "build".to_string()]),
+        "expected rebuilt timeline to end with plan -> build, got {rebuilt_nodes:?}"
+    );
 
     let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), &source_run_id);
     assert_eq!(
@@ -202,17 +239,18 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
         regex::escape(&source_run_id[..8]),
         "[RUN_PREFIX]".to_string(),
     ));
+    rewind_filters.push((r"@\d+".to_string(), "@[ORDINAL]".to_string()));
 
     let mut source_rewind = context.command();
     source_rewind.current_dir(repo_dir.path());
-    source_rewind.args(["rewind", &source_run_id, "@2", "--no-push"]);
+    source_rewind.args(["rewind", &source_run_id, "build", "--no-push"]);
     source_rewind.timeout(std::time::Duration::from_secs(15));
     fabro_snapshot!(rewind_filters, source_rewind, @"
     success: true
     exit_code: 0
     ----- stdout -----
     ----- stderr -----
-    Rewound metadata branch to @2 (build)
+    Rewound metadata branch to @[ORDINAL] (build)
     Rewound run branch fabro/run/[ULID] to [SHA]
 
     To resume: fabro resume [RUN_PREFIX]

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

@@ -5182,11 +5182,17 @@ async fn cancel_run(
     if let Some(token) = &cancel_token {
         token.store(true, Ordering::SeqCst);
     }
-    if let Some(cancel_tx) = cancel_tx {
+    let sent_cancel_signal = if let Some(cancel_tx) = cancel_tx {
         let _ = cancel_tx.send(());
-    }
+        true
+    } else {
+        false
+    };
     if let Some(answer_transport) = answer_transport {
-        let _ = answer_transport.cancel_run().await;
+        if !(sent_cancel_signal && matches!(answer_transport, RunAnswerTransport::InProcess { .. }))
+        {
+            let _ = answer_transport.cancel_run().await;
+        }
     }
     if let Some(worker_pid) = worker_pid {
         #[cfg(unix)]

lib/crates/fabro-server/tests/it/scenario/lifecycle.rs

@@ -69,6 +69,24 @@ async fn wait_for_question(app: &axum::Router, run_id: &str) -> serde_json::Valu
     panic!("question should have appeared");
 }
 
+async fn wait_for_run_state(
+    app: &axum::Router,
+    run_id: &str,
+    expected_status: &str,
+    expected_reason: &str,
+) -> serde_json::Value {
+    for _ in 0..POLL_ATTEMPTS {
+        let body = run_json(app, run_id).await;
+        if body["status"].as_str() == Some(expected_status)
+            && body["status_reason"].as_str() == Some(expected_reason)
+        {
+            return body;
+        }
+        sleep(POLL_INTERVAL).await;
+    }
+    panic!("run {run_id} did not reach status={expected_status} reason={expected_reason}");
+}
+
 const GATE_DOT: &str = r#"digraph GateTest {
     graph [goal="Test gate"]
     start [shape=Mdiamond]
@@ -197,8 +215,6 @@ async fn full_http_lifecycle_cancel() {
     assert_eq!(body["pending_control"], "cancel");
 
     // Verify the durable store view converges to cancelled failure.
-    let status = wait_for_run_status(&app, &run_id, &["failed"]).await;
-    assert_eq!(status, "failed");
-    let body = run_json(&app, &run_id).await;
+    let body = wait_for_run_state(&app, &run_id, "failed", "cancelled").await;
     assert_eq!(body["status_reason"], "cancelled");
 }

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

@@ -1735,12 +1735,21 @@ mod tests {
 
     #[test]
     fn run_and_create_commands_include_test_labels() {
-        let _lock = env_lock().lock().expect("env lock poisoned");
-        let _guard = EnvGuard::set("NEXTEST_RUN_ID", Some("run-cmd-labels"));
-        let missing_bin_root = tempfile::tempdir().expect("failed to create temp dir");
-        // Keep the binary path missing so this unit test never bootstraps a
-        // shared server based on ambient filesystem state at /tmp/fabro.
-        let context = TestContext::new(missing_bin_root.path().join("fabro"));
+        let context_root = tempfile::tempdir().expect("failed to create temp dir");
+        let context = TestContext {
+            temp_dir: context_root.path().join("temp"),
+            home_dir: context_root.path().join("home"),
+            storage_dir: context_root.path().join("storage"),
+            test_case_id: "case-123".to_string(),
+            test_run_id: "run-cmd-labels".to_string(),
+            session_root: context_root.path().join("session"),
+            fabro_bin: context_root.path().join("fabro"),
+            filters: Vec::new(),
+            active_socket_path: context_root.path().join("fabro.sock"),
+            isolated_server: None,
+            managed_storage_dirs: Vec::new(),
+            _context_root: context_root,
+        };
 
         let run_args = context
             .run_cmd()