diff --git a/Cargo.lock b/Cargo.lock index f8ea8b75d..f510e36e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1445,6 +1445,7 @@ dependencies = [ "axum", "base64 0.22.1", "chrono", + "ctor", "dotenvy", "freshell-agent-runtime", "freshell-api", diff --git a/crates/freshell-server/Cargo.toml b/crates/freshell-server/Cargo.toml index 0359c1d96..979a377de 100644 --- a/crates/freshell-server/Cargo.toml +++ b/crates/freshell-server/Cargo.toml @@ -193,3 +193,8 @@ reqwest = { version = "0.13", default-features = false, features = ["json", "rus # `crates/freshell-opencode/Cargo.toml`'s dev-dependency comment). No new # download — tokio is already a direct dependency above. tokio = { version = "1", features = ["test-util"] } +# `test_trace_capture.rs` installs the test binary's one process-wide tracing +# subscriber before `main`, so no test thread can fire a callsite while it is +# being installed (tracing-core would cache that callsite as uninterested for +# the rest of the process). Already in the workspace lock via tauri-utils. +ctor = "0.8" diff --git a/crates/freshell-server/src/checkpoints.rs b/crates/freshell-server/src/checkpoints.rs index 943b3a954..4cb09fa7a 100644 --- a/crates/freshell-server/src/checkpoints.rs +++ b/crates/freshell-server/src/checkpoints.rs @@ -493,6 +493,9 @@ pub struct CheckpointService { test_capture_calls: std::sync::atomic::AtomicUsize, #[cfg(test)] test_capture_entered: tokio::sync::Notify, + /// When the latest automatic attempt finished resolving its workspace. + #[cfg(test)] + test_workspace_resolved_at: StdMutex>, } tokio::task_local! { @@ -899,6 +902,8 @@ impl CheckpointService { test_capture_calls: std::sync::atomic::AtomicUsize::new(0), #[cfg(test)] test_capture_entered: tokio::sync::Notify::new(), + #[cfg(test)] + test_workspace_resolved_at: StdMutex::new(None), } } @@ -2352,6 +2357,13 @@ impl CheckpointService { return record; } }; + #[cfg(test)] + { + *self + .test_workspace_resolved_at + .lock() + .expect("workspace resolution timestamp mutex poisoned") = Some(Instant::now()); + } if let Some(remaining) = self.active_cooldown(&workspace.path) { log_automatic_checkpoint_failure( Some(request_id), @@ -9046,16 +9058,32 @@ mod tests { .is_empty()); } + /// During a timeout cooldown an automatic attempt answers + /// `capture_timeout_cooldown` as soon as its workspace is known, and the + /// message is forwarded: the attempt neither waits for the workspace's + /// capture lock nor starts a capture, so it never repeats the capture + /// wait. The test holds that lock throughout, so an attempt that waited + /// for it (or captured under it) would settle as `workspace_lock_timeout` + /// instead. The answer's latency is measured from the moment workspace + /// resolution returns: every automatic attempt resolves its workspace + /// (which runs Git) before it can look up the cooldown, so that part + /// follows machine load, not the cooldown policy. #[tokio::test] async fn cooldown_sends_immediately_without_capture() { let home = tempfile::tempdir().unwrap(); - let workspace = tempfile::tempdir().unwrap(); + let workspace_dir = tempfile::tempdir().unwrap(); + // The cooldown and the capture lock are keyed by the resolved path. + let workspace = std::fs::canonicalize(workspace_dir.path()).unwrap(); let service = checkpoint_test_service(home.path()); - service.establish_cooldown(workspace.path()); + service.establish_cooldown(&workspace); + let _capture_lock = service + .lock_workspace(&workspace, Duration::from_secs(1)) + .await + .expect("the test holds the workspace's capture lock"); - let (outcome, dispatches, elapsed) = run_checkpoint_policy_send( + let (outcome, dispatches, _) = run_checkpoint_policy_send( Arc::clone(&service), - checkpoint_policy_send(workspace.path(), "cooldown-send", "cooldown-attempt"), + checkpoint_policy_send(&workspace, "cooldown-send", "cooldown-attempt"), ) .await; @@ -9063,9 +9091,32 @@ mod tests { assert_eq!(outcome["code"], "capture_timeout_cooldown"); assert_eq!(dispatches, 1); assert_eq!(service.test_capture_calls.load(Ordering::SeqCst), 0); + + // Time the cooldown answer itself, from the resolved workspace on. + *service.test_workspace_resolved_at.lock().unwrap() = None; + let send = checkpoint_policy_send(&workspace, "cooldown-timed", "cooldown-timed-attempt"); + let record = service + .capture_automatic(&send, "cooldown-timed", "cooldown-timed-attempt") + .await; + let answered_at = Instant::now(); + let resolved_at = service + .test_workspace_resolved_at + .lock() + .unwrap() + .expect("the attempt resolved its workspace"); + assert_eq!( + record.outcome.code.as_deref(), + Some("capture_timeout_cooldown") + ); + assert!( + record.capture_settled, + "a cooldown answer forwards the send" + ); + assert_eq!(service.test_capture_calls.load(Ordering::SeqCst), 0); + let answer = answered_at.duration_since(resolved_at); assert!( - elapsed < Duration::from_secs(1), - "cooldown should not repeat the capture wait: {elapsed:?}" + answer < Duration::from_secs(1), + "cooldown should answer without waiting once the workspace is known: {answer:?}" ); } diff --git a/crates/freshell-server/src/extensions.rs b/crates/freshell-server/src/extensions.rs index c0c9f0e02..7f67ecb11 100644 --- a/crates/freshell-server/src/extensions.rs +++ b/crates/freshell-server/src/extensions.rs @@ -611,7 +611,9 @@ mod tests { // A shared process-global collector avoids competing test modules // installing subscribers with disconnected event buffers. Thread-local // set_default proved nondeterministic under parallel cargo test because - // of callsite interest caching; this test filters by its unique path. + // of callsite interest caching, and so did installing the collector + // lazily (see `test_trace_capture`: it is installed before `main`); + // this test filters by its unique path. let events = crate::test_trace_capture::captured_trace_events(); let root = tmp(); diff --git a/crates/freshell-server/src/files.rs b/crates/freshell-server/src/files.rs index 345ab2189..300901f26 100644 --- a/crates/freshell-server/src/files.rs +++ b/crates/freshell-server/src/files.rs @@ -1115,51 +1115,17 @@ fn internal_error(msg: &str) -> Response { mod tests { use super::*; - // `std::env::set_var` mutates whole-process state, and this bin test - // binary ALREADY serializes env-mutating tests (session_directory.rs - // provider_home tests, main.rs resolve-wiring tests) on the crate-wide - // `HOME_ENV_TEST_LOCK` (session_directory.rs:463-464). Reuse THAT lock — - // a files-local mutex would not serialize against those 11 tests. - use crate::session_directory::HOME_ENV_TEST_LOCK as ENV_LOCK; - - /// Poison-tolerant acquisition (same pattern as `main.rs:2460`). - fn env_lock() -> std::sync::MutexGuard<'static, ()> { - ENV_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - } - - /// RAII: set/remove a group of env vars, restoring prior values on drop. - struct EnvGuard { - saved: Vec<(&'static str, Option)>, - } - - impl EnvGuard { - /// `Some(v)` sets the var; `None` removes it. Prior state is restored - /// (set-or-removed) on drop, even on panic. - fn set(pairs: &[(&'static str, Option<&str>)]) -> Self { - let saved = pairs - .iter() - .map(|(key, value)| { - let prior = std::env::var_os(key); - match value { - Some(v) => std::env::set_var(key, v), - None => std::env::remove_var(key), - } - (*key, prior) - }) - .collect(); - EnvGuard { saved } - } - } - - impl Drop for EnvGuard { - fn drop(&mut self) { - for (key, prior) in self.saved.drain(..) { - match prior { - Some(v) => std::env::set_var(key, v), - None => std::env::remove_var(key), - } + // `std::env::set_var` mutates whole-process state: every test here that + // changes or reads process env owns it through the crate-wide + // `TestEnv` (`crate::test_env_lock`), which restores what it changed. + use crate::test_env_lock::TestEnv; + + /// `Some(v)` sets the var; `None` removes it — for the rest of the test. + fn apply_env(env: &mut TestEnv, pairs: &[(&str, Option<&str>)]) { + for (key, value) in pairs { + match value { + Some(value) => env.set(key, value), + None => env.remove(key), } } } @@ -1173,12 +1139,11 @@ mod tests { /// `^(.*)/[a-zA-Z]/Windows/System32$` shape or the mount prefix silently /// falls back to /mnt. struct WslMountFixture { - _env: EnvGuard, root: tempfile::TempDir, } impl WslMountFixture { - fn new() -> Self { + fn new(env: &mut TestEnv) -> Self { let root = tempfile::tempdir().unwrap(); let sys32 = root.path().join("c/Windows/System32"); std::fs::create_dir_all(&sys32).unwrap(); @@ -1186,13 +1151,16 @@ mod tests { std::fs::write(root.path().join("c/Users/notes.txt"), b"x").unwrap(); std::fs::create_dir_all(root.path().join("d/proj")).unwrap(); let sys32_str = sys32.to_string_lossy().into_owned(); - let env = EnvGuard::set(&[ - ("WSL_DISTRO_NAME", Some("Ubuntu")), - ("WSL_INTEROP", None), - ("WSLENV", None), - ("WSL_WINDOWS_SYS32", Some(sys32_str.as_str())), - ]); - WslMountFixture { _env: env, root } + apply_env( + env, + &[ + ("WSL_DISTRO_NAME", Some("Ubuntu")), + ("WSL_INTEROP", None), + ("WSLENV", None), + ("WSL_WINDOWS_SYS32", Some(sys32_str.as_str())), + ], + ); + WslMountFixture { root } } /// The native directory a `X:\` drive maps to, e.g. `mount("c")`. @@ -1202,13 +1170,16 @@ mod tests { } /// Env pinned to a plain (non-WSL) Linux host. - fn non_wsl_env() -> EnvGuard { - EnvGuard::set(&[ - ("WSL_DISTRO_NAME", None), - ("WSL_INTEROP", None), - ("WSLENV", None), - ("WSL_WINDOWS_SYS32", None), - ]) + fn pin_non_wsl_env(env: &mut TestEnv) { + apply_env( + env, + &[ + ("WSL_DISTRO_NAME", None), + ("WSL_INTEROP", None), + ("WSLENV", None), + ("WSL_WINDOWS_SYS32", None), + ], + ); } fn test_state() -> FilesState { @@ -1236,8 +1207,8 @@ mod tests { #[test] fn resolve_user_path_windows_drive_on_wsl_maps_to_mount() { - let _guard = env_lock(); - let fixture = WslMountFixture::new(); + let mut env = TestEnv::lock_blocking(); + let fixture = WslMountFixture::new(&mut env); let mount_c = fixture.mount("c").to_string_lossy().into_owned(); let r = resolve_user_path("C:\\"); @@ -1257,8 +1228,8 @@ mod tests { #[test] fn resolve_user_path_windows_off_wsl_is_unaddressable() { - let _guard = env_lock(); - let _env = non_wsl_env(); + let mut env = TestEnv::lock_blocking(); + pin_non_wsl_env(&mut env); let r = resolve_user_path("C:\\Users"); assert_eq!(r.display, "C:\\Users"); assert_eq!(r.fs_path, None); @@ -1266,8 +1237,8 @@ mod tests { #[test] fn resolve_user_path_windows_unresolvable_forms_are_unaddressable() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock_blocking(); + let _fixture = WslMountFixture::new(&mut env); // Bare drive, rooted, and generic (non-WSL) UNC inputs have no native // address even on WSL. (NOTE: drive-relative `C:foo` is NOT Windows // flavor — Node's WINDOWS_DRIVE_PREFIX_RE requires a separator or @@ -1281,8 +1252,8 @@ mod tests { #[test] fn resolve_user_path_posix_and_tilde_unchanged() { - let _guard = env_lock(); - let _env = EnvGuard::set(&[("HOME", Some("/home/tester"))]); + let mut env = TestEnv::lock_blocking(); + env.set("HOME", "/home/tester"); // POSIX: exact normalize_user_path behavior — display == fs path. let r = resolve_user_path("/tmp/x///"); assert_eq!(r.display, "/tmp/x"); @@ -1301,8 +1272,8 @@ mod tests { /// name that no endpoint could address (cluster tilde_expand regression). #[test] fn test_tilde_expand_backslash_form_expands_to_home() { - let _guard = env_lock(); - let _env = EnvGuard::set(&[("HOME", Some("/home/tester"))]); + let mut env = TestEnv::lock_blocking(); + env.set("HOME", "/home/tester"); let r = resolve_user_path("~\\proj"); assert_eq!(r.display, "/home/tester/proj"); assert_eq!(r.fs_path, Some("/home/tester/proj".to_string())); @@ -1329,15 +1300,10 @@ mod tests { // ---- validate_dir (R-WIN2: Windows input on WSL resolves via the mount) ---- - // Intentional: `_guard` is held across every `.await` BY DESIGN — the - // whole test mutates/reads process env and must stay serialized on the - // crate-wide HOME_ENV_TEST_LOCK end-to-end. Holding the env guard across - // .await is the point: it serializes env-mutating tests process-wide. - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn validate_dir_accepts_windows_drive_on_wsl() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); let resp = validate_dir( State(test_state()), auth_headers(), @@ -1351,12 +1317,10 @@ mod tests { assert_eq!(v["resolvedPath"], "C:\\"); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn validate_dir_windows_deep_path_and_missing_dir() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); // D:\proj exists in the fixture. let resp = validate_dir( State(test_state()), @@ -1382,12 +1346,10 @@ mod tests { assert_eq!(v["resolvedPath"], "C:\\Nope"); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn validate_dir_windows_input_invalid_off_wsl() { - let _guard = env_lock(); - let _env = non_wsl_env(); + let mut env = TestEnv::lock().await; + pin_non_wsl_env(&mut env); let resp = validate_dir( State(test_state()), auth_headers(), @@ -1401,11 +1363,9 @@ mod tests { assert_eq!(v["resolvedPath"], "C:\\"); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn validate_dir_posix_regression() { - let _guard = env_lock(); + let _env = TestEnv::lock().await; let tmp = tempfile::tempdir().unwrap(); let dir = tmp.path().to_string_lossy().into_owned(); let resp = validate_dir( @@ -1432,19 +1392,21 @@ mod tests { assert_eq!(v["checkpointWorkspaceEligible"], false); } - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn validate_dir_keeps_account_home_as_launch_directory_only() { - let _guard = env_lock(); + let mut env = TestEnv::lock().await; let home = tempfile::tempdir().unwrap(); let home_path = home.path().to_string_lossy().into_owned(); let project = home.path().join("project"); std::fs::create_dir(&project).unwrap(); - let _env = EnvGuard::set(&[ - ("HOME", Some(home_path.as_str())), - ("FRESHELL_HOME", None), - ("USERPROFILE", None), - ]); + apply_env( + &mut env, + &[ + ("HOME", Some(home_path.as_str())), + ("FRESHELL_HOME", None), + ("USERPROFILE", None), + ], + ); let response = validate_dir( State(test_state()), auth_headers(), @@ -1485,20 +1447,22 @@ mod tests { } #[cfg(unix)] - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn validate_dir_excludes_passwd_home_when_home_env_is_redirected() { - let _guard = env_lock(); + let mut env = TestEnv::lock().await; let redirected_home = tempfile::tempdir().unwrap(); let redirected_home_path = redirected_home.path().to_string_lossy().into_owned(); let account_home = std::fs::canonicalize(crate::session_directory::passwd_entry_home()).unwrap(); let account_home_path = account_home.to_string_lossy().into_owned(); - let _env = EnvGuard::set(&[ - ("HOME", Some(redirected_home_path.as_str())), - ("FRESHELL_HOME", None), - ("USERPROFILE", None), - ]); + apply_env( + &mut env, + &[ + ("HOME", Some(redirected_home_path.as_str())), + ("FRESHELL_HOME", None), + ("USERPROFILE", None), + ], + ); let response = validate_dir( State(test_state()), @@ -1543,11 +1507,9 @@ mod tests { /// ENOENT the raw stat on the missing `nope` intermediate and echo the /// un-collapsed path; both the stat and the display string must use the /// collapsed form. - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn test_dotseg_collapse_display() { - let _guard = env_lock(); + let _env = TestEnv::lock().await; let tmp = tempfile::tempdir().unwrap(); std::fs::create_dir_all(tmp.path().join("real")).unwrap(); let spelled = format!("{}/nope/../real", tmp.path().to_string_lossy()); @@ -1590,14 +1552,10 @@ mod tests { .collect() } - // Intentional: `_guard` is held across every `.await` BY DESIGN — these - // tests mutate/read process env and must stay serialized on the - // crate-wide HOME_ENV_TEST_LOCK end-to-end. - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn complete_windows_drive_root_lists_in_input_flavor() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); // C:\ is a directory (the fixture's /c) -> list all children, // display paths joined win32-style in the input's flavor. // Children of /c: Users/ and Windows/ (dirs sort before files; @@ -1606,12 +1564,10 @@ mod tests { assert_eq!(paths, vec!["C:\\Users", "C:\\Windows"]); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn complete_windows_partial_leaf_filters_and_preserves_flavor() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); // Partial leaf: split parent/leaf on the display path, filter by leaf. let paths = complete_paths("C:\\Us", None, None).await; assert_eq!(paths, vec!["C:\\Users"]); @@ -1623,12 +1579,10 @@ mod tests { assert_eq!(paths, vec!["c:\\Users"]); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn complete_windows_dirs_only_filters_files() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); // C:\Users contains dan/ (dir) and notes.txt (file). let all = complete_paths("C:\\Users\\", None, None).await; assert_eq!(all, vec!["C:\\Users\\dan", "C:\\Users\\notes.txt"]); @@ -1636,27 +1590,23 @@ mod tests { assert_eq!(dirs, vec!["C:\\Users\\dan"]); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn complete_windows_missing_parent_and_off_wsl_return_empty() { - let _guard = env_lock(); + let mut env = TestEnv::lock().await; { - let _fixture = WslMountFixture::new(); + let _fixture = WslMountFixture::new(&mut env); // Parent C:\Nope doesn't exist -> readdir NotFound -> 200 { suggestions: [] }. assert!(complete_paths("C:\\Nope\\x", None, None).await.is_empty()); } - let _env = non_wsl_env(); + pin_non_wsl_env(&mut env); // Windows input on a non-WSL host is unaddressable -> empty suggestions. assert!(complete_paths("C:\\", None, None).await.is_empty()); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn complete_windows_root_anchoring_composes() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); // Relative prefix under a windows-flavor root: resolve_completion_input // joins with `/`, and win32_resolve then normalizes the mixed // separators into a windows display path. @@ -1664,12 +1614,10 @@ mod tests { assert_eq!(paths, vec!["C:\\Users\\dan"]); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn complete_wsl_unc_partial_leaf_round_trips() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); // WSL-UNC inputs convert to native root-relative paths // (`\\wsl.localhost\Ubuntu\

` -> `/

`; distro matched // case-insensitively against WSL_DISTRO_NAME, path.rs:384-419) — so a @@ -1690,11 +1638,9 @@ mod tests { ); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn complete_posix_regression_unchanged() { - let _guard = env_lock(); + let _env = TestEnv::lock().await; let tmp = tempfile::tempdir().unwrap(); std::fs::create_dir_all(tmp.path().join("subdir")).unwrap(); std::fs::create_dir_all(tmp.path().join("subzero")).unwrap(); @@ -1711,8 +1657,8 @@ mod tests { #[test] fn sandbox_target_uses_converted_native_path() { - let _guard = env_lock(); - let fixture = WslMountFixture::new(); + let mut env = TestEnv::lock_blocking(); + let fixture = WslMountFixture::new(&mut env); let mount_c = fixture.mount("c").to_string_lossy().into_owned(); // R-WIN4: sandbox comparisons use the CONVERTED native path (Node's // validatePath resolves through toFilesystemPath before isPathAllowed). @@ -1744,12 +1690,10 @@ mod tests { // ---- read/stat/write (R-WIN4: sandbox + fs access via the converted path) ---- - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn read_stat_write_follow_windows_conversion_on_wsl() { - let _guard = env_lock(); - let fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let fixture = WslMountFixture::new(&mut env); // stat: the fixture file C:\Users\notes.txt exists via the mount. let resp = stat_file( State(test_state()), @@ -1793,12 +1737,10 @@ mod tests { assert!(!std::path::Path::new("C:\\Users\\dan\\note2.txt").exists()); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn stat_windows_path_off_wsl_reports_not_exists() { - let _guard = env_lock(); - let _env = non_wsl_env(); + let mut env = TestEnv::lock().await; + pin_non_wsl_env(&mut env); // Node parity: the literal `C:\…` string is handed to fs and the stat // fails naturally -> { exists: false } with HTTP 200. let resp = stat_file( @@ -1819,12 +1761,10 @@ mod tests { // applies validatePath to validate-dir; 403 pinned by Node's // test/unit/server/files-router.test.ts:451-460) ---- - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn validate_dir_denies_path_outside_allowed_roots() { - let _guard = env_lock(); - let fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let fixture = WslMountFixture::new(&mut env); let state = test_state(); // Configure a sandbox root via the store's public patch API (persist // no-ops for home: None stores — settings_store.rs:420-422 — so this @@ -1864,8 +1804,8 @@ mod tests { #[test] fn windows_flavor_allowlist_roots_convert_like_node() { - let _guard = env_lock(); - let fixture = WslMountFixture::new(); + let mut env = TestEnv::lock_blocking(); + let fixture = WslMountFixture::new(&mut env); let mount_c = fixture.mount("c").to_string_lossy().into_owned(); // Drive-root allowlist entry `C:\` matches the mount root and // anything under it (Node test lines 245-247). @@ -1906,12 +1846,10 @@ mod tests { .into_response() } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn mkdir_windows_path_creates_under_mount() { - let _guard = env_lock(); - let fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let fixture = WslMountFixture::new(&mut env); let resp = mkdir_resp("C:\\Users\\dan\\newproj").await; assert_eq!(resp.status(), StatusCode::OK); let v = body_json(resp).await; @@ -1925,12 +1863,10 @@ mod tests { assert!(!std::path::Path::new("C:\\Users\\dan\\newproj").exists()); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn mkdir_rejects_unaddressable_windows_input_off_wsl() { - let _guard = env_lock(); - let _env = non_wsl_env(); + let mut env = TestEnv::lock().await; + pin_non_wsl_env(&mut env); let resp = mkdir_resp("C:\\").await; assert_eq!(resp.status(), StatusCode::BAD_REQUEST); let v = body_json(resp).await; @@ -1940,12 +1876,10 @@ mod tests { assert!(!std::path::Path::new("C:\\").exists()); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn mkdir_rejects_unresolvable_windows_forms_even_on_wsl() { - let _guard = env_lock(); - let _fixture = WslMountFixture::new(); + let mut env = TestEnv::lock().await; + let _fixture = WslMountFixture::new(&mut env); // Bare drive / rooted inputs are Windows flavor but have no absolute // native address, even on WSL. (`C:foo` is `native` flavor in both // servers — see resolve_user_path_windows_unresolvable_forms — so it @@ -1956,11 +1890,9 @@ mod tests { } } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn mkdir_posix_regression_unchanged() { - let _guard = env_lock(); + let _env = TestEnv::lock().await; let tmp = tempfile::tempdir().unwrap(); let target = format!("{}/fresh-sub", tmp.path().display()); let resp = mkdir_resp(&target).await; @@ -1972,11 +1904,9 @@ mod tests { assert!(std::path::Path::new(&target).is_dir()); } - // Intentional: env lock held across `.await` BY DESIGN (see above). - #[allow(clippy::await_holding_lock)] #[tokio::test] async fn test_mkdir_enotdir_is_409_not_500() { - let _guard = env_lock(); + let _env = TestEnv::lock().await; let tmp = tempfile::tempdir().unwrap(); // An INTERMEDIATE component that is a FILE: recursive mkdir fails // ENOTDIR, which Node maps to 409 "Path exists but is not a directory" @@ -2035,8 +1965,8 @@ mod tests { #[test] fn expand_tilde_uses_home() { - let _guard = env_lock(); - let _env = EnvGuard::set(&[("HOME", Some("/home/tester"))]); + let mut env = TestEnv::lock_blocking(); + env.set("HOME", "/home/tester"); assert_eq!(expand_tilde("~"), "/home/tester"); assert_eq!(expand_tilde("~/proj"), "/home/tester/proj"); assert_eq!(expand_tilde("/abs"), "/abs"); @@ -2050,8 +1980,8 @@ mod tests { /// POSIX path was stat'd/written verbatim. #[test] fn test_posix_input_trims_and_unquotes() { - let _guard = env_lock(); - let _env = EnvGuard::set(&[("HOME", Some("/home/tester"))]); + let mut env = TestEnv::lock_blocking(); + env.set("HOME", "/home/tester"); // Leading/trailing whitespace padding is trimmed. assert_eq!(normalize_user_path("/tmp/x "), "/tmp/x"); assert_eq!(normalize_user_path(" /tmp/x"), "/tmp/x"); @@ -2118,8 +2048,8 @@ mod tests { #[test] fn resolve_completion_input_honors_root_and_absolute() { - let _guard = env_lock(); - let _env = EnvGuard::set(&[("HOME", Some("/home/tester"))]); + let mut env = TestEnv::lock_blocking(); + env.set("HOME", "/home/tester"); // No root \u2192 prefix unchanged. assert_eq!(resolve_completion_input("a", None), "a"); // Absolute prefix ignores root. diff --git a/crates/freshell-server/src/fresh_agent_proxy_tests.rs b/crates/freshell-server/src/fresh_agent_proxy_tests.rs index 40c903142..6c19b68c6 100644 --- a/crates/freshell-server/src/fresh_agent_proxy_tests.rs +++ b/crates/freshell-server/src/fresh_agent_proxy_tests.rs @@ -920,44 +920,15 @@ async fn rollback_commands_keep_direction_target_and_request_fences() { } } -static SNAPSHOT_OUTAGE_ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); - -struct SnapshotTestEnv(Vec<(&'static str, Option)>); -impl SnapshotTestEnv { - fn isolate(root: &Path) -> Self { - let keys = [ - "HOME", - "CODEX_HOME", - "CLAUDE_CONFIG_DIR", - "CLAUDE_HOME", - "XDG_DATA_HOME", - "CODEX_CMD", - "FAKE_CODEX_APP_SERVER_BEHAVIOR", - "FAKE_CODEX_APP_SERVER_ALLOW_DURABLE_WRITES", - ]; - let saved = Self( - keys.iter() - .map(|key| (*key, std::env::var_os(key))) - .collect(), - ); - std::env::set_var("HOME", root.join("empty-home")); - std::env::set_var("CODEX_HOME", root.join("configured-codex")); - std::env::set_var("CLAUDE_CONFIG_DIR", root.join("configured-claude")); - std::env::remove_var("CLAUDE_HOME"); - std::env::set_var("XDG_DATA_HOME", root.join("configured-xdg")); - std::env::set_var("CODEX_CMD", "/never-start-a-provider-for-saved-history"); - saved - } -} -impl Drop for SnapshotTestEnv { - fn drop(&mut self) { - for (key, value) in &self.0 { - match value { - Some(value) => std::env::set_var(key, value), - None => std::env::remove_var(key), - } - } - } +/// Point every provider home at `root` (and keep a real provider from +/// starting) for the rest of the test, through the crate-wide `TestEnv`. +fn isolate_snapshot_env(env: &mut crate::test_env_lock::TestEnv, root: &Path) { + env.set("HOME", root.join("empty-home")); + env.set("CODEX_HOME", root.join("configured-codex")); + env.set("CLAUDE_CONFIG_DIR", root.join("configured-claude")); + env.remove("CLAUDE_HOME"); + env.set("XDG_DATA_HOME", root.join("configured-xdg")); + env.set("CODEX_CMD", "/never-start-a-provider-for-saved-history"); } fn snapshot_outage_proxy(socket: &Path) -> Arc { @@ -1084,9 +1055,9 @@ async fn real_gateway_pre_native_opencode_projects_only_current_empty_owned_regi #[tokio::test] async fn real_gateway_managed_snapshot_failure_reads_owned_history_and_refuses_wrong_local_store() { - let _lock = SNAPSHOT_OUTAGE_ENV_LOCK.lock().await; + let mut env = crate::test_env_lock::TestEnv::lock().await; let dir = tempfile::tempdir().unwrap(); - let _env = SnapshotTestEnv::isolate(dir.path()); + isolate_snapshot_env(&mut env, dir.path()); let local_path = dir .path() .join("configured-codex/sessions/rollout-session-activity.jsonl"); @@ -1264,9 +1235,9 @@ async fn real_gateway_managed_snapshot_failure_reads_owned_history_and_refuses_w #[tokio::test] async fn real_gateway_outage_preserves_exact_cold_native_history_without_live_authority() { - let _lock = SNAPSHOT_OUTAGE_ENV_LOCK.lock().await; + let mut env = crate::test_env_lock::TestEnv::lock().await; let dir = tempfile::tempdir().unwrap(); - let _env = SnapshotTestEnv::isolate(dir.path()); + isolate_snapshot_env(&mut env, dir.path()); let socket = dir.path().join("owned-supervisor.sock"); let listener = UnixListener::bind(&socket).unwrap(); drop(listener); @@ -1431,21 +1402,21 @@ async fn real_gateway_outage_preserves_exact_cold_native_history_without_live_au #[tokio::test] async fn real_gateway_outage_keeps_a_registered_local_provider_snapshot_live() { - let _lock = SNAPSHOT_OUTAGE_ENV_LOCK.lock().await; + let mut env = crate::test_env_lock::TestEnv::lock().await; let dir = tempfile::tempdir().unwrap(); - let _env = SnapshotTestEnv::isolate(dir.path()); - std::env::set_var( + isolate_snapshot_env(&mut env, dir.path()); + env.set( "CODEX_CMD", format!( "node {}/../../test/fixtures/coding-cli/codex-app-server/fake-app-server.mjs", env!("CARGO_MANIFEST_DIR") ), ); - std::env::set_var( + env.set( "FAKE_CODEX_APP_SERVER_BEHAVIOR", r#"{"threadStartThreadId":"local-owned-thread"}"#, ); - std::env::set_var("FAKE_CODEX_APP_SERVER_ALLOW_DURABLE_WRITES", "1"); + env.set("FAKE_CODEX_APP_SERVER_ALLOW_DURABLE_WRITES", "1"); let (tx, mut rx) = broadcast::channel(64); let broadcast = Arc::new(tx); let ownership = Arc::new(freshell_ownership::RuntimeOwnershipRegistry::new()); diff --git a/crates/freshell-server/src/main.rs b/crates/freshell-server/src/main.rs index fc0e6ccb8..23ecef2ed 100644 --- a/crates/freshell-server/src/main.rs +++ b/crates/freshell-server/src/main.rs @@ -86,6 +86,8 @@ pub(crate) mod test_clock_gate; mod test_clock_router; #[cfg(test)] pub(crate) mod test_env_lock; +#[cfg(all(test, unix))] +mod test_parked_child; #[cfg(test)] mod test_trace_capture; mod updater; @@ -4800,37 +4802,6 @@ mod tests { assert_eq!(second_epoch, first_epoch + 1); } - /// Save-and-restore guard for one env var (tests below mutate real - /// process env; the shared `HOME_ENV_TEST_LOCK` serializes them - /// crate-wide). Restores on drop, panic included. - struct EnvVarGuard { - name: &'static str, - saved: Option, - } - - impl EnvVarGuard { - fn unset(name: &'static str) -> Self { - let saved = std::env::var_os(name); - std::env::remove_var(name); - Self { name, saved } - } - - fn set(name: &'static str, value: &str) -> Self { - let saved = std::env::var_os(name); - std::env::set_var(name, value); - Self { name, saved } - } - } - - impl Drop for EnvVarGuard { - fn drop(&mut self) { - match self.saved.take() { - Some(v) => std::env::set_var(self.name, v), - None => std::env::remove_var(self.name), - } - } - } - /// TERM-09 boot wiring (responsive-terminal-restore Workstream 3): the /// boot resolution must fail fast on a configuration that would restore /// the spill≥disconnect inversion, naming the offending env vars — and @@ -4839,9 +4810,10 @@ mod tests { /// TERMINAL_* vars). #[test] fn term09_boot_resolution_fails_fast_on_inverted_env() { - let _queue = EnvVarGuard::unset("TERMINAL_CLIENT_QUEUE_MAX_BYTES"); - let _catastrophic = EnvVarGuard::unset("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES"); - let _stall = EnvVarGuard::unset("TERMINAL_WS_CATASTROPHIC_STALL_MS"); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + env.remove("TERMINAL_CLIENT_QUEUE_MAX_BYTES"); + env.remove("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES"); + env.remove("TERMINAL_WS_CATASTROPHIC_STALL_MS"); // Defaults must boot and satisfy the strict ordering. let defaults = resolve_term09_config().expect("defaults must resolve"); @@ -4849,7 +4821,7 @@ mod tests { // An inverted override (disconnect below spill — the production // incident's shape) refuses to boot, naming both env vars. - let _inverted = EnvVarGuard::set("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES", "1048576"); + env.set("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES", "1048576"); let err = resolve_term09_config().expect_err("inverted env must refuse boot"); let message = err.to_string(); assert!( @@ -4857,19 +4829,18 @@ mod tests { && message.contains("TERMINAL_CLIENT_QUEUE_MAX_BYTES"), "the boot error must name the offending env vars: {message}" ); - drop(_inverted); + env.remove("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES"); // An equalized pair (disconnect == spill) also refuses. - let _equalized_queue = EnvVarGuard::set("TERMINAL_CLIENT_QUEUE_MAX_BYTES", "268435456"); - let _equalized_cat = - EnvVarGuard::set("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES", "268435456"); + env.set("TERMINAL_CLIENT_QUEUE_MAX_BYTES", "268435456"); + env.set("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES", "268435456"); resolve_term09_config().expect_err("equalized env must refuse boot"); - drop(_equalized_queue); - drop(_equalized_cat); + env.remove("TERMINAL_CLIENT_QUEUE_MAX_BYTES"); + env.remove("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES"); // A correctly-ordered override boots. - let _ordered_queue = EnvVarGuard::set("TERMINAL_CLIENT_QUEUE_MAX_BYTES", "2097152"); - let _ordered_cat = EnvVarGuard::set("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES", "8388608"); + env.set("TERMINAL_CLIENT_QUEUE_MAX_BYTES", "2097152"); + env.set("TERMINAL_WS_CATASTROPHIC_BUFFERED_BYTES", "8388608"); let tuned = resolve_term09_config().expect("ordered env must boot"); assert_eq!(tuned.queue_max_bytes, 2 * 1024 * 1024); assert_eq!(tuned.catastrophic_buffered_bytes, 8 * 1024 * 1024); @@ -4948,11 +4919,9 @@ mod tests { #[cfg(unix)] #[test] fn wire_home_dir_unix_treats_empty_home_as_unset_using_passwd_entry() { - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let _home = EnvVarGuard::set("HOME", ""); - let _userprofile = EnvVarGuard::set("USERPROFILE", "/Users/win-fixture-wire"); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + env.set("HOME", ""); + env.set("USERPROFILE", "/Users/win-fixture-wire"); let resolved = resolve_wire_home_dir().map(|h| h.as_str().to_string()); assert_ne!( resolved.as_deref(), @@ -4973,11 +4942,9 @@ mod tests { #[cfg(windows)] #[test] fn wire_home_dir_windows_uses_userprofile_never_home() { - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let _home = EnvVarGuard::set("HOME", "C:\\never-consulted"); - let _userprofile = EnvVarGuard::set("USERPROFILE", "C:\\Users\\win-fixture-wire"); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + env.set("HOME", "C:\\never-consulted"); + env.set("USERPROFILE", "C:\\Users\\win-fixture-wire"); assert_eq!( resolve_wire_home_dir().map(|h| h.as_str().to_string()), Some("C:\\Users\\win-fixture-wire".to_string()), @@ -4993,15 +4960,10 @@ mod tests { // cwd-less/subagent transcript omitted from the index produces a // healthy-looking ready-empty where Node (os.homedir()) finds it. const SESSION_ID: &str = "AB2AFDA6-A340-443E-BA60-024A1B3554B4"; - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - // Mutates `CLAUDE_HOME` (unset) — hold the crate-wide CLAUDE env - // lock too, AFTER the HOME lock (`crate::test_env_lock`'s order). - let _claude_env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); let home = env_test_temp_dir("claude-fallback"); - let _home = EnvVarGuard::set("HOME", home.to_str().unwrap()); - let _claude_home = EnvVarGuard::unset("CLAUDE_HOME"); + env.set("HOME", home.to_str().unwrap()); + env.remove("CLAUDE_HOME"); let project = home.join(".claude").join("projects").join("-repo-alpha"); std::fs::create_dir_all(&project).expect("mkdir claude project dir"); @@ -5028,16 +4990,11 @@ mod tests { // transcript living ONLY under `USERPROFILE/.claude` must NOT be // surfaced (the pre-fix HOME||USERPROFILE approximation found it). const SESSION_ID: &str = "0C7B39D1-52E4-4F0F-9E7C-4E2B7A11D9AA"; - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - // Mutates `CLAUDE_HOME` (unset) — hold the crate-wide CLAUDE env - // lock too, AFTER the HOME lock (`crate::test_env_lock`'s order). - let _claude_env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); let profile = env_test_temp_dir("claude-fallback-userprofile"); - let _home = EnvVarGuard::unset("HOME"); - let _claude_home = EnvVarGuard::unset("CLAUDE_HOME"); - let _userprofile = EnvVarGuard::set("USERPROFILE", profile.to_str().unwrap()); + env.remove("HOME"); + env.remove("CLAUDE_HOME"); + env.set("USERPROFILE", profile.to_str().unwrap()); let project = profile.join(".claude").join("projects").join("-repo-beta"); std::fs::create_dir_all(&project).expect("mkdir claude project dir"); @@ -5065,15 +5022,10 @@ mod tests { // USERPROFILE-backed on Windows). The exact-id fallback itself — // not just `provider_home()` — must find the transcript there. const SESSION_ID: &str = "AB2AFDA6-A340-443E-BA60-024A1B3554B4"; - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - // Mutates `CLAUDE_HOME` (unset) — hold the crate-wide CLAUDE env - // lock too, AFTER the HOME lock (`crate::test_env_lock`'s order). - let _claude_env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); let home = env_test_temp_dir("claude-fallback"); - let _claude_home = EnvVarGuard::unset("CLAUDE_HOME"); - let _userprofile = EnvVarGuard::set("USERPROFILE", home.to_str().unwrap()); + env.remove("CLAUDE_HOME"); + env.set("USERPROFILE", home.to_str().unwrap()); let project = home.join(".claude").join("projects").join("-repo-alpha"); std::fs::create_dir_all(&project).expect("mkdir claude project dir"); @@ -5134,7 +5086,7 @@ mod tests { // Env-first reader (`claude_home` may resolve a FOREIGN root while a // same-binary mutator holds `CLAUDE_HOME`) — take the crate-wide // lock (see `crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let _env = crate::test_env_lock::TestEnv::lock_blocking(); let home = tempfile::tempdir().expect("tempdir"); let proj = home.path().join(".claude").join("projects").join("-p"); std::fs::create_dir_all(&proj).expect("mkdir projects/-p"); @@ -5148,7 +5100,7 @@ mod tests { #[test] fn claude_empty_projects_tree_is_definitively_absent() { // Env-first reader — take the crate-wide lock (`crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let _env = crate::test_env_lock::TestEnv::lock_blocking(); let home = tempfile::tempdir().expect("tempdir"); std::fs::create_dir_all(home.path().join(".claude").join("projects")) .expect("mkdir empty projects"); @@ -5167,7 +5119,7 @@ mod tests { // concurrently running discovery test's CLAUDE_HOME (readable tree, // no sess-1.jsonl) flipped the gate to "definitively absent" and // failed the must-DEFER assert. - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let _env = crate::test_env_lock::TestEnv::lock_blocking(); let home = tempfile::tempdir().expect("tempdir"); let projects = home.path().join(".claude").join("projects"); std::fs::create_dir_all(&projects).expect("mkdir projects"); @@ -5189,7 +5141,7 @@ mod tests { use std::os::unix::fs::PermissionsExt; // Env-first reader — take the crate-wide lock (`crate::test_env_lock`): // the gate-flake twin of the projects-root case above. - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let _env = crate::test_env_lock::TestEnv::lock_blocking(); let home = tempfile::tempdir().expect("tempdir"); let projects = home.path().join(".claude").join("projects"); let proj = projects.join("-p"); @@ -5212,7 +5164,7 @@ mod tests { // Env-first reader — take the crate-wide lock (`crate::test_env_lock`): // a concurrent mutator's foreign CLAUDE_HOME HAS a projects root, so // the no-root defer branch would flip to "definitively absent". - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let _env = crate::test_env_lock::TestEnv::lock_blocking(); let home = tempfile::tempdir().expect("tempdir"); assert!( !transcript_definitively_absent(home.path(), "claude", "sess-1"), @@ -5263,17 +5215,15 @@ mod tests { } /// `featureFlags.kilroy` under a given `KILROY_ENABLED` value (`None` = unset). - /// Serializes on the crate-wide `HOME_ENV_TEST_LOCK`; [`EnvVarGuard`] restores - /// the prior value on drop (panic included), so a real-world `KILROY_ENABLED` - /// never leaks across tests in either direction. + /// Owns the process env through the crate-wide `TestEnv`, which restores + /// the prior value on drop (panic included), so a real-world + /// `KILROY_ENABLED` never leaks across tests in either direction. fn kilroy_flag_for_env(value: Option<&str>) -> bool { - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let _kilroy = match value { - Some(v) => EnvVarGuard::set("KILROY_ENABLED", v), - None => EnvVarGuard::unset("KILROY_ENABLED"), - }; + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + match value { + Some(v) => env.set("KILROY_ENABLED", v), + None => env.remove("KILROY_ENABLED"), + } build_platform_payload(serde_json::json!({}), false)["featureFlags"]["kilroy"] .as_bool() .expect("featureFlags.kilroy is always a bool") @@ -5349,10 +5299,8 @@ mod tests { // mirroring Node's `process.platform !== 'win32'`). `KILROY_ENABLED` is // pinned off under the shared lock so the assertion is // environment-independent. - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let _kilroy = EnvVarGuard::unset("KILROY_ENABLED"); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + env.remove("KILROY_ENABLED"); let cell = ai_title::AiKeyCell::init(Some("sk-live-abc123".into()), None); let payload = build_platform_payload(serde_json::json!({}), cell.enabled()); assert_eq!( @@ -5363,10 +5311,8 @@ mod tests { #[test] fn platform_payload_ai_enabled_false_without_key() { - let _lock = crate::session_directory::HOME_ENV_TEST_LOCK - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let _kilroy = EnvVarGuard::unset("KILROY_ENABLED"); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + env.remove("KILROY_ENABLED"); let cell = ai_title::AiKeyCell::init(None, None); let payload = build_platform_payload(serde_json::json!({}), cell.enabled()); assert_eq!( @@ -5390,51 +5336,48 @@ mod tests { } // `load_dotenv_from` (legacy parity: `import 'dotenv/config'`, - // `server/index.ts:2-3`). Each test uses its own temp dir + a uniquely-named - // sentinel var, so parallel test execution can't collide. + // `server/index.ts:2-3`). Each test writes its `.env` into a private temp + // dir (a fixed path under the shared temp root collided with another + // concurrently running copy of this binary deleting it) and owns the + // process env through `TestEnv`, which also undoes the variable that + // `load_dotenv_from` sets. #[test] fn load_dotenv_from_sets_var_absent_from_process_env() { - let dir = std::env::temp_dir().join("freshell-dotenv-test-unset"); - std::fs::create_dir_all(&dir).unwrap(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + let dir = tempfile::tempdir().unwrap(); std::fs::write( - dir.join(".env"), + dir.path().join(".env"), "FRESHELL_TASK7_TEST_VAR_UNSET=from-dotenv\n", ) .unwrap(); - std::env::remove_var("FRESHELL_TASK7_TEST_VAR_UNSET"); + env.remove("FRESHELL_TASK7_TEST_VAR_UNSET"); - load_dotenv_from(&dir); + load_dotenv_from(dir.path()); assert_eq!( std::env::var("FRESHELL_TASK7_TEST_VAR_UNSET").as_deref(), Ok("from-dotenv") ); - - std::env::remove_var("FRESHELL_TASK7_TEST_VAR_UNSET"); - let _ = std::fs::remove_dir_all(&dir); } #[test] fn load_dotenv_from_never_overrides_existing_process_env_var() { - let dir = std::env::temp_dir().join("freshell-dotenv-test-set"); - std::fs::create_dir_all(&dir).unwrap(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); + let dir = tempfile::tempdir().unwrap(); std::fs::write( - dir.join(".env"), + dir.path().join(".env"), "FRESHELL_TASK7_TEST_VAR_SET=from-dotenv\n", ) .unwrap(); - std::env::set_var("FRESHELL_TASK7_TEST_VAR_SET", "already-set"); + env.set("FRESHELL_TASK7_TEST_VAR_SET", "already-set"); - load_dotenv_from(&dir); + load_dotenv_from(dir.path()); assert_eq!( std::env::var("FRESHELL_TASK7_TEST_VAR_SET").as_deref(), Ok("already-set") ); - - std::env::remove_var("FRESHELL_TASK7_TEST_VAR_SET"); - let _ = std::fs::remove_dir_all(&dir); } // SAFE-01: startup token hardening (`server/auth.ts#validateStartupSecurity`). diff --git a/crates/freshell-server/src/managed_provider_bootstrap.rs b/crates/freshell-server/src/managed_provider_bootstrap.rs index 8faa62f74..deffad5ec 100644 --- a/crates/freshell-server/src/managed_provider_bootstrap.rs +++ b/crates/freshell-server/src/managed_provider_bootstrap.rs @@ -425,22 +425,9 @@ mod tests { use super::*; use std::fs; - struct RestoreOnecliEnv(Vec<(&'static str, Option)>); - - impl Drop for RestoreOnecliEnv { - fn drop(&mut self) { - for (name, value) in self.0.drain(..) { - match value { - Some(value) => std::env::set_var(name, value), - None => std::env::remove_var(name), - } - } - } - } - #[test] fn named_opencode_onecli_references_include_private_environment_auth_and_ca_files() { - let _env_lock = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); let workspace = tempfile::tempdir().unwrap(); let environment = workspace.path().join("opencode.env"); let auth = workspace.path().join("opencode-auth.json"); @@ -455,9 +442,8 @@ mod tests { "FRESHELL_MANAGED_OPENCODE_ONECLI_AUTH_FILE", "FRESHELL_MANAGED_OPENCODE_ONECLI_CA_FILE", ]; - let _restore = RestoreOnecliEnv(keys.map(|key| (key, std::env::var_os(key))).into()); for (key, file) in keys.into_iter().zip([&environment, &auth, &ca]) { - std::env::set_var(key, file); + env.set(key, file); } let references = named_provider_onecli_references("opencode").unwrap(); diff --git a/crates/freshell-server/src/managed_provider_launch_tests.rs b/crates/freshell-server/src/managed_provider_launch_tests.rs index 175ba25ea..dd9395afa 100644 --- a/crates/freshell-server/src/managed_provider_launch_tests.rs +++ b/crates/freshell-server/src/managed_provider_launch_tests.rs @@ -711,115 +711,105 @@ impl RuntimeBackend for BrokerAuthorizedBackend { } } +/// Collect the failure classes of fresh-agent launch errors from every +/// thread, through the test binary's process-wide capture subscriber (one is +/// installed before `main`, so this observes rather than installs). fn capture_fresh_launch_failure_classes() -> std::sync::Arc>> { - use tracing_subscriber::prelude::*; - #[derive(Clone)] - struct FailureLayer(std::sync::Arc>>); - impl tracing_subscriber::Layer for FailureLayer { - fn on_event( - &self, - event: &tracing::Event<'_>, - _: tracing_subscriber::layer::Context<'_, S>, - ) { - if !event.metadata().target().contains("fresh_agent_proxy") - && !event.metadata().target().contains("managed_mcp_capability") - { - return; - } - struct Classes(Vec<&'static str>); - impl tracing::field::Visit for Classes { - fn record_debug( - &mut self, - field: &tracing::field::Field, - value: &dyn std::fmt::Debug, - ) { - if field.name() != "error" { - return; - } - let text = format!("{value:?}"); - for label in [ - "Claude native source selection changed", - "HostUnreachable", - "InvalidRequest", - "ProviderInputsChanged", - "prepare hosted provider config", - "prepare hosted fresh-agent provider state", - "prepare hosted OneCLI provider files", - "open hosted", - "OneCLI", - "profile", - "credential", - "unsafe or mismatched", - "MCP", - "sidecar", - "source", - "manifest", - "copy", - "Permission", - "unavailable", - "initialization", - "timed out", - "Claude", - "kilroy", - "environment", - "grant", - "provider", - "configuration", - "refresh", - "input bind", - "trusted uid/group", - "provider home", - "resource", - "destination", - "projection", - "stage", - "metadata", - "path", - "readable", - "preparation", - "version", - "changed", - "rootless", - "directory", - "filesystem", - "temporary", - "failed to start", - "failed with status", - "original", - "origins", - "No such file", - "resolve", - "root", - "budget", - "bound", - "pending", - "committed", - "publication", - "source evidence", - "source selection", - "source provenance", - ] { - if text.contains(label) { - self.0.push(label); - } + let classes = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let sink = classes.clone(); + crate::test_trace_capture::observe_all_events(move |event| { + if !event.metadata().target().contains("fresh_agent_proxy") + && !event.metadata().target().contains("managed_mcp_capability") + { + return; + } + struct Classes(Vec<&'static str>); + impl tracing::field::Visit for Classes { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + if field.name() != "error" { + return; + } + let text = format!("{value:?}"); + for label in [ + "Claude native source selection changed", + "HostUnreachable", + "InvalidRequest", + "ProviderInputsChanged", + "prepare hosted provider config", + "prepare hosted fresh-agent provider state", + "prepare hosted OneCLI provider files", + "open hosted", + "OneCLI", + "profile", + "credential", + "unsafe or mismatched", + "MCP", + "sidecar", + "source", + "manifest", + "copy", + "Permission", + "unavailable", + "initialization", + "timed out", + "Claude", + "kilroy", + "environment", + "grant", + "provider", + "configuration", + "refresh", + "input bind", + "trusted uid/group", + "provider home", + "resource", + "destination", + "projection", + "stage", + "metadata", + "path", + "readable", + "preparation", + "version", + "changed", + "rootless", + "directory", + "filesystem", + "temporary", + "failed to start", + "failed with status", + "original", + "origins", + "No such file", + "resolve", + "root", + "budget", + "bound", + "pending", + "committed", + "publication", + "source evidence", + "source selection", + "source provenance", + ] { + if text.contains(label) { + self.0.push(label); } } } - let mut classes = Classes(Vec::new()); - event.record(&mut classes); - self.0.lock().unwrap().extend(classes.0); } - } - let classes = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); - tracing::subscriber::set_global_default( - tracing_subscriber::registry().with(FailureLayer(classes.clone())), - ) - .unwrap(); + let mut found = Classes(Vec::new()); + event.record(&mut found); + sink.lock().unwrap().extend(found.0); + }); classes } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn managed_provider_amplifier_terminal_native_startup_consumes_prepared_inputs() { + // Live qualification mutates provider env for its whole run: own it + // through the crate-wide `TestEnv`, which restores everything at the end. + let mut env = crate::test_env_lock::TestEnv::lock().await; use freshell_supervisor::{ admission::AdmissionPolicy, registry::Registry, @@ -1047,7 +1037,7 @@ async def on_session_ready(coordinator): ) .unwrap(); std::fs::set_permissions(&env_file, std::fs::Permissions::from_mode(0o600)).unwrap(); - std::env::set_var("FRESHELL_MANAGED_AMPLIFIER_ONECLI_ENV_FILE", &env_file); + env.set("FRESHELL_MANAGED_AMPLIFIER_ONECLI_ENV_FILE", &env_file); let registry = Registry::open( fixture.join("registry"), Some( @@ -1056,7 +1046,7 @@ async def on_session_ready(coordinator): ), ) .unwrap(); - std::env::set_var("FRESHELL_MANAGED_PROVIDER_QUALIFICATION", "amplifier"); + env.set("FRESHELL_MANAGED_PROVIDER_QUALIFICATION", "amplifier"); let socket = control.join("control.sock"); let secret_file = control.join("secret"); std::fs::write(&secret_file, "closed-synthetic-control-secret").unwrap(); @@ -1126,11 +1116,11 @@ async def on_session_ready(coordinator): "FRESHELL_MANAGED_RUNTIME_V1", "FRESHELL_MANAGED_FRESH_AGENT_V1", ] { - std::env::set_var(name, "1"); + env.set(name, "1"); } - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); - std::env::set_var("HOME", &home); + env.set("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); + env.set("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); + env.set("HOME", &home); let controller = crate::managed_runtime::ServerManagedRuntimeController::from_env() .await .unwrap() @@ -1143,7 +1133,7 @@ async def on_session_ready(coordinator): "bundle:\n active: ./missing-ambient-decoy\n", ) .unwrap(); - std::env::set_var("HOME", &ambient_home); + env.set("HOME", &ambient_home); let terminal = controller .launch(ManagedTerminalLaunch { spec: freshell_platform::spawn::SpawnSpec { @@ -1842,6 +1832,9 @@ async def on_session_ready(coordinator): #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn managed_provider_claude_terminal_and_fresh_native_startup_consumes_prepared_inputs() { + // Live qualification mutates provider env for its whole run: own it + // through the crate-wide `TestEnv`, which restores everything at the end. + let mut env = crate::test_env_lock::TestEnv::lock().await; let failure_classes = capture_fresh_launch_failure_classes(); use freshell_supervisor::{ admission::AdmissionPolicy, @@ -2059,8 +2052,8 @@ let raw='';process.stdin.on('data',b=>raw+=b);process.stdin.on('end',()=>{ ) .unwrap(); std::fs::set_permissions(&env_file, std::fs::Permissions::from_mode(0o600)).unwrap(); - std::env::set_var("FRESHELL_MANAGED_CLAUDE_ONECLI_ENV_FILE", &env_file); - std::env::set_var("CLAUDE_CONFIG_DIR", &selected_config); + env.set("FRESHELL_MANAGED_CLAUDE_ONECLI_ENV_FILE", &env_file); + env.set("CLAUDE_CONFIG_DIR", &selected_config); let registry = Registry::open( fixture.join("registry"), Some( @@ -2069,7 +2062,7 @@ let raw='';process.stdin.on('data',b=>raw+=b);process.stdin.on('end',()=>{ ), ) .unwrap(); - std::env::set_var("FRESHELL_MANAGED_PROVIDER_QUALIFICATION", "claude"); + env.set("FRESHELL_MANAGED_PROVIDER_QUALIFICATION", "claude"); let socket = control.join("control.sock"); let secret_file = control.join("secret"); std::fs::write(&secret_file, "closed-synthetic-control-secret").unwrap(); @@ -2141,11 +2134,11 @@ let raw='';process.stdin.on('data',b=>raw+=b);process.stdin.on('end',()=>{ "FRESHELL_MANAGED_RUNTIME_V1", "FRESHELL_MANAGED_FRESH_AGENT_V1", ] { - std::env::set_var(name, "1"); + env.set(name, "1"); } - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); - std::env::set_var("HOME", &home); + env.set("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); + env.set("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); + env.set("HOME", &home); let controller = crate::managed_runtime::ServerManagedRuntimeController::from_env() .await .unwrap() @@ -2282,7 +2275,7 @@ let raw='';process.stdin.on('data',b=>raw+=b);process.stdin.on('end',()=>{ "@missing-ambient-decoy.md\n", ) .unwrap(); - std::env::set_var("HOME", &ambient_home); + env.set("HOME", &ambient_home); async fn consumed(path: &Path) -> Option { tokio::time::timeout(Duration::from_secs(45), async { loop { @@ -2442,7 +2435,7 @@ let raw='';process.stdin.on('data',b=>raw+=b);process.stdin.on('end',()=>{ std::fs::remove_file(&disabled_project_settings).unwrap(); let (broadcast, _) = tokio::sync::broadcast::channel(256); let mut events = broadcast.subscribe(); - std::env::set_var("HOME", &home); + env.set("HOME", &home); let proxy = crate::fresh_agent_proxy::HostedFreshAgentProxy::from_opt_in( Some(client.clone()), Arc::new(broadcast), @@ -2797,7 +2790,7 @@ let raw='';process.stdin.on('data',b=>raw+=b);process.stdin.on('end',()=>{ .unwrap(); let (broadcast, _) = tokio::sync::broadcast::channel(256); let mut events = broadcast.subscribe(); - std::env::set_var("HOME", &home); + env.set("HOME", &home); let proxy = crate::fresh_agent_proxy::HostedFreshAgentProxy::from_opt_in( Some(client.clone()), Arc::new(broadcast), @@ -3126,6 +3119,9 @@ try { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn managed_provider_codex_terminal_and_fresh_native_startup_consumes_prepared_inputs() { + // Live qualification mutates provider env for its whole run: own it + // through the crate-wide `TestEnv`, which restores everything at the end. + let mut env = crate::test_env_lock::TestEnv::lock().await; use freshell_supervisor::{ admission::AdmissionPolicy, registry::Registry, @@ -3267,7 +3263,7 @@ require('node:readline').createInterface({input:process.stdin}).on('line',line=> ), ) .unwrap(); - std::env::set_var("FRESHELL_MANAGED_PROVIDER_QUALIFICATION", "codex"); + env.set("FRESHELL_MANAGED_PROVIDER_QUALIFICATION", "codex"); let socket = control.join("control.sock"); let secret_file = control.join("secret"); std::fs::write(&secret_file, "closed-synthetic-control-secret").unwrap(); @@ -3340,11 +3336,11 @@ require('node:readline').createInterface({input:process.stdin}).on('line',line=> ("FRESHELL_MANAGED_RUNTIME_V1", "1"), ("FRESHELL_MANAGED_FRESH_AGENT_V1", "1"), ] { - std::env::set_var(name, value); + env.set(name, value); } - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); - std::env::set_var("HOME", &home); + env.set("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); + env.set("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); + env.set("HOME", &home); let controller = crate::managed_runtime::ServerManagedRuntimeController::from_env() .await .unwrap() @@ -3511,6 +3507,9 @@ require('node:readline').createInterface({input:process.stdin}).on('line',line=> #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn managed_provider_opencode_terminal_and_fresh_native_startup_consumes_prepared_inputs() { + // Live qualification mutates provider env for its whole run: own it + // through the crate-wide `TestEnv`, which restores everything at the end. + let mut env = crate::test_env_lock::TestEnv::lock().await; use freshell_supervisor::{ admission::AdmissionPolicy, registry::Registry, @@ -3622,10 +3621,10 @@ async fn managed_provider_opencode_terminal_and_fresh_native_startup_consumes_pr crate::managed_mcp_capability::spawn_callback_server() .await .unwrap(); - std::env::set_var("FRESHELL_MANAGED_RUNTIME_V1", "1"); - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); - std::env::set_var("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); - std::env::set_var("FRESHELL_MANAGED_FRESH_AGENT_V1", "1"); + env.set("FRESHELL_MANAGED_RUNTIME_V1", "1"); + env.set("FRESHELL_RUNTIME_CONTROL_SOCKET", &socket); + env.set("FRESHELL_RUNTIME_CONTROL_SECRET_FILE", &secret_file); + env.set("FRESHELL_MANAGED_FRESH_AGENT_V1", "1"); let controller = crate::managed_runtime::ServerManagedRuntimeController::from_env() .await .unwrap() @@ -4020,8 +4019,8 @@ export default {{id:'closed-directory-discovery',async server(){{ // cwd. Native project discovery must stay in the workspace while provider // HOME remains its independent physical volume. let terminal_home = fixture.join("workspace"); - std::env::set_var("XDG_CONFIG_HOME", fixture.join("source-home/.config")); - std::env::set_var("HOME", &terminal_home); + env.set("XDG_CONFIG_HOME", fixture.join("source-home/.config")); + env.set("HOME", &terminal_home); let request: ManagedTerminalLaunch = ManagedTerminalLaunch { spec: freshell_platform::spawn::SpawnSpec { program: "opencode".into(), @@ -4348,12 +4347,12 @@ export default {{id:'closed-directory-discovery',async server(){{ std::fs::write(fixture.join("grants/auth.json"), serde_json::to_vec(&json!({well_known_url: { "type":"wellknown", "key":"OPENAI_API_KEY", "token":"AUTH_PREFIX{file:~/home rule with spaces.txt}AUTH_SUFFIX" }, inline_auth_url: {"type":"wellknown", "key":"INLINE_VALUE", "token":"AUTH_INLINE_SHADOW_NO_FILE"}})).unwrap()).unwrap(); - std::env::set_var("HOME", &fixture); - std::env::set_var("OPENCODE_CONFIG", &explicit_config); - std::env::set_var("OPENCODE_CONFIG_DIR", &explicit_directory); - std::env::set_var("OPENCODE_TUI_CONFIG", &selected_tui); - std::env::set_var("OPENCODE_CONFIG_CONTENT", inline_source); - std::env::set_var("INLINE_VALUE", inline_value); + env.set("HOME", &fixture); + env.set("OPENCODE_CONFIG", &explicit_config); + env.set("OPENCODE_CONFIG_DIR", &explicit_directory); + env.set("OPENCODE_TUI_CONFIG", &selected_tui); + env.set("OPENCODE_CONFIG_CONTENT", inline_source); + env.set("INLINE_VALUE", inline_value); let mut fresh_expected = expected.clone(); fresh_expected["snapshotDisabled"] = json!(true); let (broadcast, _) = tokio::sync::broadcast::channel(256); diff --git a/crates/freshell-server/src/net_bind.rs b/crates/freshell-server/src/net_bind.rs index 3a5b6f32b..f7f35cfc2 100644 --- a/crates/freshell-server/src/net_bind.rs +++ b/crates/freshell-server/src/net_bind.rs @@ -11,9 +11,10 @@ //! controller owns its own accept loop per listener. Retiring a listener uses //! `Notify::notify_one()` (permit-storing: the wakeup cannot be lost, unlike //! `notify_waiters`) and then AWAITS the old accept-loop `JoinHandle`, which -//! first hands off queued connections and then drops its listener — a -//! deterministic "old socket closed" barrier, so callers may respond/probe -//! immediately after `serve_on` returns. In-flight connections (incl. +//! first hands off queued connections, then shuts the listening socket down +//! (so, on Linux, a duplicate held by a concurrently forked child process +//! cannot keep it listening) and drops it — a deterministic "old socket closed" barrier, +//! so callers may respond/probe immediately after `serve_on` returns. In-flight connections (incl. //! WebSockets) drain in their own spawned tasks — no mass 4009 on rebind. //! //! Trade-off: SO_REUSEPORT lets another process of the same effective UID bind @@ -65,6 +66,28 @@ pub fn bind_reusable(addr: SocketAddr, reuse_port: bool) -> std::io::Result {} + // Platforms that do not shut down a listening socket report it as + // not connected; closing the descriptors is all they offer. + Err(err) if err.kind() == std::io::ErrorKind::NotConnected => {} + Err(err) => tracing::warn!( + error = %err, + "retired listener shutdown failed; it stops listening once every inherited descriptor closes" + ), + } +} + /// One live listener: its shutdown signal and accept-loop task handle. The /// accept loop drops its listener before exiting, so awaiting the handle is /// a true "old socket closed" barrier. The bound address lives on the @@ -178,6 +201,9 @@ impl RebindController { } } } + // Dropping the handles below would not close the + // socket while a forked child still holds a copy. + stop_listening(&drain_listener); break; }, res = listener.accept() => { @@ -196,8 +222,9 @@ impl RebindController { } } } - // `listener` is dropped HERE, before the task completes: awaiting - // this JoinHandle is a true "old listener closed" barrier. + // `listener` is stopped above and dropped HERE, before the task + // completes: awaiting this JoinHandle is a true "old listener + // closed" barrier, even against copies held by forked children. }); let mut cur = self.current.lock().await; *self.current_addr.lock().expect("current_addr lock") = Some(bound_addr); @@ -239,12 +266,32 @@ mod tests { use std::net::{IpAddr, Ipv4Addr, SocketAddr}; // Keep this listener until the controller's initial bind succeeds, then - // drop it before sending traffic so only the controller accepts requests. - fn reusable_loopback_listener() -> StdTcpListener { - bind_reusable(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), true) + // retire it before sending traffic so only the controller accepts requests. + // + // It binds the WILDCARD address: these tests later rebind the controller + // to 0.0.0.0, which needs the port free on every address. A loopback-only + // `bind(127.0.0.1:0)` lets the kernel hand out a port that another socket + // on this host is already using on a different address (a busy machine's + // outgoing connection, say), and the later wildcard rebind then fails + // with AddrInUse. A wildcard `bind(0.0.0.0:0)` only picks a port that is + // free everywhere, and from then on some socket of this test holds it. + fn reusable_port_owner() -> StdTcpListener { + bind_reusable(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)), true) .expect("own an OS-assigned reusable listener") } + // The port owner shares the controller's SO_REUSEPORT port. Dropping it + // is not enough: a child process forked meanwhile by another test holds + // a duplicate until it execs, and while it does the owner keeps + // listening — the kernel can then hand this test's connection to the + // owner's queue, which nobody serves, and reset it once the child's copy + // closes. Stop it listening first, as the controller does with a + // retired listener. + fn retire_port_owner(port_owner: StdTcpListener) { + stop_listening(&port_owner); + drop(port_owner); + } + #[test] fn reuse_port_kill_switch_reads_env() { assert!(parse_reuse_port(None)); @@ -257,7 +304,7 @@ mod tests { #[test] fn two_reuseport_binds_on_same_addr_both_succeed() { - let a = reusable_loopback_listener(); + let a = reusable_port_owner(); let addr = a.local_addr().expect("first listener address"); let b = bind_reusable(addr, true).expect("second reuseport bind must also succeed"); drop((a, b)); @@ -279,7 +326,7 @@ mod tests { #[tokio::test] async fn serve_on_proves_bind_before_swapping_and_serves_traffic() { use axum::{routing::get, Router}; - let port_owner = reusable_loopback_listener(); + let port_owner = reusable_port_owner(); let port = port_owner .local_addr() .expect("initial listener address") @@ -295,7 +342,7 @@ mod tests { ctl.serve_on(IpAddr::V4(Ipv4Addr::LOCALHOST)) .await .expect("initial serve"); - drop(port_owner); + retire_port_owner(port_owner); let body = reqwest::get(format!("http://127.0.0.1:{port}/ping")) .await .unwrap() @@ -322,7 +369,7 @@ mod tests { // reports/V1.md): with notify_waiters and no barrier, 42-99/100 of // these iterations fail. Do NOT weaken this test. use axum::{routing::get, Router}; - let port_owner = reusable_loopback_listener(); + let port_owner = reusable_port_owner(); let port = port_owner .local_addr() .expect("initial listener address") @@ -333,7 +380,7 @@ mod tests { let localhost = IpAddr::V4(Ipv4Addr::LOCALHOST); let wildcard = IpAddr::V4(Ipv4Addr::UNSPECIFIED); ctl.serve_on(localhost).await.expect("initial serve"); - drop(port_owner); + retire_port_owner(port_owner); for i in 0..100 { let target = if i % 2 == 0 { wildcard } else { localhost }; ctl.serve_on(target).await.expect("swap"); @@ -360,7 +407,7 @@ mod tests { use std::io::Write; use tokio::io::AsyncReadExt; - let port_owner = reusable_loopback_listener(); + let port_owner = reusable_port_owner(); let port = port_owner .local_addr() .expect("initial listener address") @@ -371,7 +418,7 @@ mod tests { ctl.serve_on(IpAddr::V4(Ipv4Addr::LOCALHOST)) .await .expect("initial serve"); - drop(port_owner); + retire_port_owner(port_owner); // A current-thread runtime has not polled the accept loop yet. The // handshake and request reach the old socket before its shutdown. @@ -400,6 +447,56 @@ mod tests { ctl.shutdown_all().await; } + /// A retired listener stops listening even while a child process forked + /// before the rebind still holds a duplicate of it (every PTY, sidecar or + /// Git spawn on another thread opens that window until its exec). Merely + /// closing the server's descriptors would leave it LISTENING: a new + /// connection to its specific address would land in a queue nobody + /// serves — ahead of the new wildcard listener — and be reset when the + /// child's copy finally closes. + /// + /// Linux only: the fix shuts the listening socket down, and only Linux + /// stops a listening socket that way. macOS and the BSDs answer such a + /// shutdown with "not connected" and leave the socket listening, which + /// `stop_listening` accepts because closing the descriptors is all those + /// platforms offer. + #[cfg(target_os = "linux")] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_retired_listener_held_by_a_forked_child_stops_accepting() { + use axum::{routing::get, Router}; + let port_owner = reusable_port_owner(); + let port = port_owner + .local_addr() + .expect("initial listener address") + .port(); + let app = Router::new().route("/ping", get(|| async { "pong" })); + let ctl = RebindController::new(port, true); + ctl.set_app(app); + ctl.serve_on(IpAddr::V4(Ipv4Addr::LOCALHOST)) + .await + .expect("initial serve"); + retire_port_owner(port_owner); + + // Fork while the loopback listener is open, then retire it. + let child = crate::test_parked_child::ParkedChild::fork_now(); + ctl.serve_on(IpAddr::V4(Ipv4Addr::UNSPECIFIED)) + .await + .expect("rebind serve"); + let probe = tokio::time::timeout(std::time::Duration::from_secs(5), async { + reqwest::get(format!("http://127.0.0.1:{port}/ping")) + .await? + .text() + .await + }) + .await; + child.release(); + let body = probe + .expect("the new listener answers; the retired one must not swallow the probe") + .expect("the probe is not reset"); + assert_eq!(body, "pong"); + ctl.shutdown_all().await; + } + /// Pins the DEV-0013 drain property: a connection accepted by the OLD /// listener keeps working across a rebind (drains gracefully in its /// detached per-connection task) while the NEW listener serves new @@ -415,7 +512,7 @@ mod tests { use tokio::sync::{mpsc, watch}; use tokio::time::timeout; - let port_owner = reusable_loopback_listener(); + let port_owner = reusable_port_owner(); let port = port_owner .local_addr() .expect("initial listener address") @@ -446,7 +543,7 @@ mod tests { ctl.set_app(app); let localhost = IpAddr::V4(Ipv4Addr::LOCALHOST); ctl.serve_on(localhost).await.expect("initial serve"); - drop(port_owner); + retire_port_owner(port_owner); // Start a request that will still be in flight when we swap. Only the // OLD listener exists at this point, so it owns the connection. diff --git a/crates/freshell-server/src/session_directory.rs b/crates/freshell-server/src/session_directory.rs index d68d906ed..2373b38be 100644 --- a/crates/freshell-server/src/session_directory.rs +++ b/crates/freshell-server/src/session_directory.rs @@ -1040,16 +1040,6 @@ pub(crate) fn provider_home() -> Option { std::env::home_dir() } -/// Serializes tests (crate-wide) that mutate the process-global -/// `HOME`/`USERPROFILE`/`CLAUDE_HOME`/`FRESHELL_HOME` env vars: cargo runs -/// tests in parallel THREADS within one process, so two tests racing to -/// mutate the SAME vars would otherwise flake (one test's assertion -/// observing the OTHER test's in-flight env state). Shared `pub(crate)` so -/// `main.rs`'s resolve-wiring tests serialize with this module's -/// `provider_home()` tests. -#[cfg(test)] -pub(crate) static HOME_ENV_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); - /// Test-only oracle for the effective user's passwd-entry home directory /// (`getpwuid_r`) — the Node `os.homedir()` POSIX fallback when `HOME` is /// unset or empty. `pub(crate)` so `main.rs`'s resolve-wiring tests assert @@ -2410,48 +2400,38 @@ mod tests { home } + /// The Claude root of a synthetic test home, pinned explicitly. Tests + /// never resolve it through [`claude_home`]: that reader lets a + /// process-wide `CLAUDE_HOME` win, and other tests in this binary set + /// `CLAUDE_HOME` to their own homes while these run. + fn fixture_claude_root(home: &Path) -> PathBuf { + home.join(".claude") + } + fn default_query() -> DirQuery { DirQuery::default() } // `provider_home()` (FRESHELL_HOME root-alignment fix): coding-CLI provider // session sources must resolve against the REAL `HOME`, never the - // `FRESHELL_HOME`-overridden config root. Each test saves + restores both - // vars around itself since they're real process env (no injected `Env` - // plumbing exists at this call site), matching the existing convention in - // `files.rs` (`expand_tilde_uses_home` et al.) -- but ALSO serializes on - // `PROVIDER_HOME_ENV_LOCK` because cargo runs tests in parallel THREADS - // within one process: two tests racing to mutate the SAME process-global - // `HOME`/`FRESHELL_HOME` vars would otherwise flake (one test's assertion - // observing the OTHER test's in-flight env state). The lock itself is the - // crate-wide `HOME_ENV_TEST_LOCK` (module level, above) so `main.rs`'s - // resolve-wiring tests serialize with these. - use super::HOME_ENV_TEST_LOCK as PROVIDER_HOME_ENV_LOCK; + // `FRESHELL_HOME`-overridden config root. `provider_home()` reads real + // process env (no injected `Env` plumbing exists at this call site), so + // each test owns the environment through the crate-wide + // `crate::test_env_lock::TestEnv`, which restores what it changed. #[cfg(unix)] #[test] fn provider_home_ignores_freshell_home_uses_real_home() { - let _guard = PROVIDER_HOME_ENV_LOCK.lock().unwrap(); - let saved_freshell_home = std::env::var("FRESHELL_HOME").ok(); - let saved_home = std::env::var("HOME").ok(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); - std::env::set_var("FRESHELL_HOME", "/tmp/freshell-isolated-config-root"); - std::env::set_var("HOME", "/home/real-user-fixture"); + env.set("FRESHELL_HOME", "/tmp/freshell-isolated-config-root"); + env.set("HOME", "/home/real-user-fixture"); assert_eq!( provider_home(), Some(PathBuf::from("/home/real-user-fixture")), "provider_home() must resolve the real HOME, ignoring FRESHELL_HOME" ); - - match saved_freshell_home { - Some(v) => std::env::set_var("FRESHELL_HOME", v), - None => std::env::remove_var("FRESHELL_HOME"), - } - match saved_home { - Some(v) => std::env::set_var("HOME", v), - None => std::env::remove_var("HOME"), - } } // Node `os.homedir()` platform semantics (libuv `uv_os_homedir`): on @@ -2463,14 +2443,11 @@ mod tests { #[cfg(unix)] #[test] fn provider_home_unix_uses_passwd_entry_when_home_and_userprofile_unset() { - let _guard = PROVIDER_HOME_ENV_LOCK.lock().unwrap(); - let saved_freshell_home = std::env::var("FRESHELL_HOME").ok(); - let saved_home = std::env::var("HOME").ok(); - let saved_userprofile = std::env::var("USERPROFILE").ok(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); - std::env::set_var("FRESHELL_HOME", "/tmp/freshell-isolated-config-root-2"); - std::env::remove_var("HOME"); - std::env::remove_var("USERPROFILE"); + env.set("FRESHELL_HOME", "/tmp/freshell-isolated-config-root-2"); + env.remove("HOME"); + env.remove("USERPROFILE"); let resolved = provider_home(); assert!( @@ -2482,30 +2459,15 @@ mod tests { Some(super::passwd_entry_home()), "an unset HOME must fall back to the passwd-entry home (Node os.homedir() POSIX semantics)" ); - - match saved_freshell_home { - Some(v) => std::env::set_var("FRESHELL_HOME", v), - None => std::env::remove_var("FRESHELL_HOME"), - } - match saved_home { - Some(v) => std::env::set_var("HOME", v), - None => std::env::remove_var("HOME"), - } - match saved_userprofile { - Some(v) => std::env::set_var("USERPROFILE", v), - None => std::env::remove_var("USERPROFILE"), - } } #[cfg(unix)] #[test] fn provider_home_unix_ignores_userprofile_when_home_unset() { - let _guard = PROVIDER_HOME_ENV_LOCK.lock().unwrap(); - let saved_home = std::env::var("HOME").ok(); - let saved_userprofile = std::env::var("USERPROFILE").ok(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); - std::env::remove_var("HOME"); - std::env::set_var("USERPROFILE", "/Users/win-fixture"); + env.remove("HOME"); + env.set("USERPROFILE", "/Users/win-fixture"); let resolved = provider_home(); assert_ne!( @@ -2518,15 +2480,6 @@ mod tests { Some(super::passwd_entry_home()), "with HOME unset, POSIX must resolve the passwd-entry home" ); - - match saved_home { - Some(v) => std::env::set_var("HOME", v), - None => std::env::remove_var("HOME"), - } - match saved_userprofile { - Some(v) => std::env::set_var("USERPROFILE", v), - None => std::env::remove_var("USERPROFILE"), - } } #[cfg(unix)] @@ -2536,53 +2489,31 @@ mod tests { // an unset HOME — Node's `os.homedir()` never returns an empty // string; on POSIX it falls back to the passwd-entry home, never // USERPROFILE. - let _guard = PROVIDER_HOME_ENV_LOCK.lock().unwrap(); - let saved_home = std::env::var("HOME").ok(); - let saved_userprofile = std::env::var("USERPROFILE").ok(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); - std::env::set_var("HOME", ""); - std::env::set_var("USERPROFILE", "/Users/win-fixture-empty"); + env.set("HOME", ""); + env.set("USERPROFILE", "/Users/win-fixture-empty"); assert_eq!( provider_home(), Some(super::passwd_entry_home()), "an EMPTY HOME must behave like unset HOME: passwd-entry fallback, never USERPROFILE" ); - - match saved_home { - Some(v) => std::env::set_var("HOME", v), - None => std::env::remove_var("HOME"), - } - match saved_userprofile { - Some(v) => std::env::set_var("USERPROFILE", v), - None => std::env::remove_var("USERPROFILE"), - } } #[cfg(unix)] #[test] fn provider_home_unix_prefers_home_and_never_consults_userprofile() { - let _guard = PROVIDER_HOME_ENV_LOCK.lock().unwrap(); - let saved_home = std::env::var("HOME").ok(); - let saved_userprofile = std::env::var("USERPROFILE").ok(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); - std::env::set_var("HOME", "/home/real-user-fixture"); - std::env::set_var("USERPROFILE", "/Users/win-fixture"); + env.set("HOME", "/home/real-user-fixture"); + env.set("USERPROFILE", "/Users/win-fixture"); assert_eq!( provider_home(), Some(PathBuf::from("/home/real-user-fixture")), "a set, non-empty HOME must win on POSIX (USERPROFILE is never consulted)" ); - - match saved_home { - Some(v) => std::env::set_var("HOME", v), - None => std::env::remove_var("HOME"), - } - match saved_userprofile { - Some(v) => std::env::set_var("USERPROFILE", v), - None => std::env::remove_var("USERPROFILE"), - } } // Native Windows/Tauri parity: `os.homedir()` reads USERPROFILE on @@ -2591,27 +2522,16 @@ mod tests { #[cfg(windows)] #[test] fn provider_home_windows_uses_userprofile_never_home() { - let _guard = PROVIDER_HOME_ENV_LOCK.lock().unwrap(); - let saved_home = std::env::var("HOME").ok(); - let saved_userprofile = std::env::var("USERPROFILE").ok(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); - std::env::set_var("HOME", "C:\\never-consulted"); - std::env::set_var("USERPROFILE", "C:\\Users\\win-fixture"); + env.set("HOME", "C:\\never-consulted"); + env.set("USERPROFILE", "C:\\Users\\win-fixture"); assert_eq!( provider_home(), Some(PathBuf::from("C:\\Users\\win-fixture")), "Windows must read USERPROFILE and never consult HOME (Node os.homedir() parity)" ); - - match saved_home { - Some(v) => std::env::set_var("HOME", v), - None => std::env::remove_var("HOME"), - } - match saved_userprofile { - Some(v) => std::env::set_var("USERPROFILE", v), - None => std::env::remove_var("USERPROFILE"), - } } #[test] @@ -2671,16 +2591,14 @@ mod tests { #[test] fn invalid_utf8_transcript_is_indexed_lossily_like_node() { - // Env-first reader (`claude_home(&home)` may resolve a FOREIGN root - // while a same-binary mutator holds `CLAUDE_HOME`) — take the - // crate-wide lock (see `crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); // Regression (bug #7 class, found by the 007 seeded-home differential): // Node reads transcripts with `fs.readFile(file,'utf8')` -> invalid // bytes become U+FFFD and the session IS indexed; `read_to_string` // silently dropped the whole file (and skewed page `revision`). let home = claude_home_with(&[]); - let project = claude_home(&home).join("projects").join("-home-dan-proj"); + let project = fixture_claude_root(&home) + .join("projects") + .join("-home-dan-proj"); std::fs::create_dir_all(&project).unwrap(); let mut bytes: Vec = Vec::new(); bytes.extend_from_slice(br#"{"parentUuid":null,"cwd":"/home/dan/proj","sessionId":"cccc1111-2222-4333-8444-555566667777","type":"user","message":{"role":"user","content":"bad "#); @@ -2693,7 +2611,7 @@ mod tests { ) .unwrap(); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); assert_eq!( items.len(), 1, @@ -2713,16 +2631,12 @@ mod tests { #[test] fn default_query_hides_non_interactive_fixtures() { - // Env-first reader (`claude_home(&home)` may resolve a FOREIGN root - // while a same-binary mutator holds `CLAUDE_HOME`) — take the - // crate-wide lock (see `crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); // `real-corrupted.jsonl` has a `cwd` and parses as non-interactive → the // default History browse (no includeNonInteractive) hides it → empty // page. `healthy.jsonl` has NO `cwd` anywhere → excluded entirely at // discovery (R10b), never reaching the item list at all. let home = claude_home_with(&["real-corrupted.jsonl", "healthy.jsonl"]); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); assert_eq!( items.len(), 1, @@ -2738,12 +2652,8 @@ mod tests { #[test] fn include_non_interactive_surfaces_titled_session() { - // Env-first reader (`claude_home(&home)` may resolve a FOREIGN root - // while a same-binary mutator holds `CLAUDE_HOME`) — take the - // crate-wide lock (see `crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); let home = claude_home_with(&["real-corrupted.jsonl", "healthy.jsonl"]); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); let q = DirQuery { include_non_interactive: true, ..DirQuery::default() @@ -2760,17 +2670,13 @@ mod tests { #[test] fn include_empty_surfaces_untitled_sessions_sorted_desc() { - // Env-first reader (`claude_home(&home)` may resolve a FOREIGN root - // while a same-binary mutator holds `CLAUDE_HOME`) — take the - // crate-wide lock (see `crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); // `healthy.jsonl` has no `cwd` → excluded at discovery (R10b) even with // every include flag set; only the cwd-bearing `real-corrupted.jsonl` // (itself untitled-if-you-squint but DOES have a title) surfaces here. // (See `r10b_cwdless_repair_fixture_never_surfaces_under_any_flags` // below for the dedicated pin of the never-surfaces behavior.) let home = claude_home_with(&["real-corrupted.jsonl", "healthy.jsonl"]); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); let q = DirQuery { include_non_interactive: true, include_empty: true, @@ -2988,10 +2894,6 @@ mod tests { #[test] fn r10b_cwdless_repair_fixture_never_surfaces_under_any_flags() { - // Env-first reader (`claude_home(&home)` may resolve a FOREIGN root - // while a same-binary mutator holds `CLAUDE_HOME`) — take the - // crate-wide lock (see `crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); // Byte-matched against a live probe of the ORIGINAL: seeding // `healthy.jsonl` (renamed to a canonical UUID filename, exactly as // `port/oracle/rest-parity/sweep.mjs#seedClaudeSessions` does) and @@ -3012,7 +2914,7 @@ mod tests { ) .unwrap(); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); assert!(items.is_empty(), "a cwd-less session must never be indexed"); let q = DirQuery { @@ -3029,12 +2931,8 @@ mod tests { #[test] fn title_search_matches_and_annotates() { - // Env-first reader (`claude_home(&home)` may resolve a FOREIGN root - // while a same-binary mutator holds `CLAUDE_HOME`) — take the - // crate-wide lock (see `crate::test_env_lock`). - let _env = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.blocking_lock(); let home = claude_home_with(&["real-corrupted.jsonl"]); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); let q = DirQuery { include_non_interactive: true, query: Some("session 1".into()), @@ -3047,7 +2945,7 @@ mod tests { assert_eq!(arr[0]["snippet"], json!("Test Session 1")); // A non-matching query → empty. - let items2 = list_claude_sessions(&claude_home(&home)); + let items2 = list_claude_sessions(&fixture_claude_root(&home)); let q2 = DirQuery { include_non_interactive: true, query: Some("zzz-not-present".into()), @@ -5152,11 +5050,11 @@ mod tests { #[test] fn b_t1_claude_source_matches_list_claude_sessions_reference_scan() { let home = claude_home_with(&["real-corrupted.jsonl", "healthy.jsonl"]); - let mut reference: Vec = list_claude_sessions(&claude_home(&home)) + let mut reference: Vec = list_claude_sessions(&fixture_claude_root(&home)) .iter() .map(Comparable::from) .collect(); - let mut production: Vec = ClaudeSource::new(claude_home(&home)) + let mut production: Vec = ClaudeSource::new(fixture_claude_root(&home)) .scan() .iter() .map(Comparable::from) @@ -5190,7 +5088,7 @@ mod tests { let auth_token: std::sync::Arc = std::sync::Arc::new("tok".into()); let session_index = std::sync::Arc::new(test_session_index(vec![ - std::sync::Arc::new(ClaudeSource::new(claude_home(&home))) + std::sync::Arc::new(ClaudeSource::new(fixture_claude_root(&home))) as std::sync::Arc, ])); let state = SessionDirectoryState { @@ -5923,7 +5821,7 @@ mod tests { let auth_token: std::sync::Arc = std::sync::Arc::new("tok".into()); let session_index = std::sync::Arc::new(test_session_index(vec![ - std::sync::Arc::new(ClaudeSource::new(claude_home(&home))) + std::sync::Arc::new(ClaudeSource::new(fixture_claude_root(&home))) as std::sync::Arc, ])); let state = SessionDirectoryState { @@ -6004,7 +5902,7 @@ mod tests { let auth_token: std::sync::Arc = std::sync::Arc::new("tok".into()); let session_index = std::sync::Arc::new(test_session_index(vec![ - std::sync::Arc::new(ClaudeSource::new(claude_home(&home))) + std::sync::Arc::new(ClaudeSource::new(fixture_claude_root(&home))) as std::sync::Arc, ])); let state = SessionDirectoryState { @@ -6065,7 +5963,7 @@ mod tests { let auth_token: std::sync::Arc = std::sync::Arc::new("tok".into()); let session_index = std::sync::Arc::new(test_session_index(vec![ - std::sync::Arc::new(ClaudeSource::new(claude_home(&home))) + std::sync::Arc::new(ClaudeSource::new(fixture_claude_root(&home))) as std::sync::Arc, ])); let state = SessionDirectoryState { @@ -6188,7 +6086,7 @@ mod tests { let discover_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let parse_calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); let source = CountingClaudeSource { - inner: ClaudeSource::new(claude_home(&home)), + inner: ClaudeSource::new(fixture_claude_root(&home)), discover_calls: std::sync::Arc::clone(&discover_calls), parse_calls: std::sync::Arc::clone(&parse_calls), }; @@ -6786,7 +6684,7 @@ mod tests { "unique-search-term-alpha", "unique-search-term-alpha-assistant-only", ); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); assert_eq!(items.len(), 1); assert!( !items[0].source_files.is_empty(), @@ -6829,7 +6727,7 @@ mod tests { "hello there", "unique-fulltext-only-phrase", ); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); let q = DirQuery { include_non_interactive: true, tier: Tier::FullText, @@ -6850,7 +6748,7 @@ mod tests { "MixedCase NeedleValue Here", "irrelevant", ); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); let q = DirQuery { include_non_interactive: true, tier: Tier::UserMessages, @@ -6869,7 +6767,7 @@ mod tests { "nothing relevant here at all", "still nothing relevant", ); - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); let q = DirQuery { include_non_interactive: true, tier: Tier::FullText, @@ -6922,7 +6820,7 @@ mod tests { let mut seen: Vec = Vec::new(); let mut cursor: Option = None; for _ in 0..3 { - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); let q = DirQuery { include_non_interactive: true, tier: Tier::UserMessages, @@ -6971,7 +6869,7 @@ mod tests { std::fs::write(project.join(format!("{sid}.jsonl")), content).unwrap(); } - let items = list_claude_sessions(&claude_home(&home)); + let items = list_claude_sessions(&fixture_claude_root(&home)); assert_eq!(items.len(), 11); let q = DirQuery { include_non_interactive: true, diff --git a/crates/freshell-server/src/session_name_native_tests.rs b/crates/freshell-server/src/session_name_native_tests.rs index 04b3b3342..045de64a8 100644 --- a/crates/freshell-server/src/session_name_native_tests.rs +++ b/crates/freshell-server/src/session_name_native_tests.rs @@ -364,11 +364,11 @@ async fn armed_pending(store: &Arc, handle: &str, root: &str) -> S // `locate_transcript_selected` reads process-global env (the ordered // candidate roots). Both discovery tests below MUTATE `CLAUDE_HOME` / -// `CLAUDE_CONFIG_DIR` for their whole set…restore window, so they hold the -// crate-wide `CLAUDE_ENV_TEST_LOCK` — shared with every other same-binary -// mutator of those vars and with the `claude_home`-resolving readers (the -// final-suite gate's unreadable-projects deferral flake; see -// `crate::test_env_lock` for the full discipline). +// `CLAUDE_CONFIG_DIR` for their whole run, so they own the environment +// through the crate-wide `TestEnv` — the one lock every same-binary env +// mutator and env-resolving reader holds (the final-suite gate's +// unreadable-projects deferral flake; see `crate::test_env_lock` for the +// full discipline). /// Unified agent names (Task 8 acceptance): a session adopted ONLY through /// the index (the auto-title sweep's hydration — the sidebar/history rename @@ -379,7 +379,7 @@ async fn armed_pending(store: &Arc, handle: &str, root: &str) -> S /// attach the acquisition so the series runs to `synced`. #[tokio::test] async fn an_index_adopted_claude_series_discovers_its_route_on_demand() { - let _guard = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.lock().await; + let mut env = crate::test_env_lock::TestEnv::lock().await; let home = tempfile::tempdir().expect("claude home tempdir"); let session_id = "4a1b2c3d-4e5f-4a6b-8c7d-9e0f1a2b3c4d"; let project_cwd = home.path().join("proj"); @@ -403,12 +403,10 @@ async fn an_index_adopted_claude_series_discovers_its_route_on_demand() { ) .expect("transcript"); - let saved_home = std::env::var_os("CLAUDE_HOME"); - let saved_config = std::env::var_os("CLAUDE_CONFIG_DIR"); - std::env::set_var("CLAUDE_HOME", home.path()); - std::env::remove_var("CLAUDE_CONFIG_DIR"); + env.set("CLAUDE_HOME", home.path()); + env.remove("CLAUDE_CONFIG_DIR"); - let result = async { + async { let dir = temp_data_dir(); let store = open_store(dir.path()); let target = session( @@ -464,16 +462,6 @@ async fn an_index_adopted_claude_series_discovers_its_route_on_demand() { assert_eq!(sync.status, NativeSyncStatus::Synced); } .await; - - match saved_home { - Some(value) => std::env::set_var("CLAUDE_HOME", value), - None => std::env::remove_var("CLAUDE_HOME"), - } - match saved_config { - Some(value) => std::env::set_var("CLAUDE_CONFIG_DIR", value), - None => std::env::remove_var("CLAUDE_CONFIG_DIR"), - } - result } /// Task-008 review M-3: a FAILED on-demand route discovery must back the @@ -485,7 +473,7 @@ async fn an_index_adopted_claude_series_discovers_its_route_on_demand() { /// and a successful discovery clears the entry and unpauses the series. #[tokio::test] async fn a_failed_route_discovery_backs_off_instead_of_rescanning_every_pass() { - let _guard = crate::test_env_lock::CLAUDE_ENV_TEST_LOCK.lock().await; + let mut env = crate::test_env_lock::TestEnv::lock().await; let home = tempfile::tempdir().expect("claude home tempdir"); let session_id = "6c3d4e5f-6a7b-4c8d-9e0f-1a2b3c4d5e6f"; let project_cwd = home.path().join("proj"); @@ -498,12 +486,10 @@ async fn a_failed_route_discovery_backs_off_instead_of_rescanning_every_pass() { let transcript_dir = home.path().join("projects").join(&mangled); std::fs::create_dir_all(&transcript_dir).expect("mangled project dir"); - let saved_home = std::env::var_os("CLAUDE_HOME"); - let saved_config = std::env::var_os("CLAUDE_CONFIG_DIR"); - std::env::set_var("CLAUDE_HOME", home.path()); - std::env::remove_var("CLAUDE_CONFIG_DIR"); + env.set("CLAUDE_HOME", home.path()); + env.remove("CLAUDE_CONFIG_DIR"); - let result = async { + async { let dir = temp_data_dir(); let store = open_store(dir.path()); let target = session( @@ -579,16 +565,6 @@ async fn a_failed_route_discovery_backs_off_instead_of_rescanning_every_pass() { ); } .await; - - match saved_home { - Some(value) => std::env::set_var("CLAUDE_HOME", value), - None => std::env::remove_var("CLAUDE_HOME"), - } - match saved_config { - Some(value) => std::env::set_var("CLAUDE_CONFIG_DIR", value), - None => std::env::remove_var("CLAUDE_CONFIG_DIR"), - } - result } // --------------------------------------------------------------------------- @@ -2597,32 +2573,20 @@ async fn a_capability_paused_head_does_not_starve_the_ready_series_behind_it() { // T3-I4: the packaged desktop app's staged helper path derivation. // --------------------------------------------------------------------------- -/// `from_env` reads process-global env — serialize every from_env probe and -/// restore the prior values (the repo's ENV_LOCK convention). -static FROM_ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); - +/// `from_env` reads process-global env — own it through the crate-wide +/// `TestEnv` (`crate::test_env_lock`), which restores the prior values. fn probe_from_env( vars: &[(&str, Option<&str>)], probe: impl FnOnce(&super::ClaudeNativeNameAdapter), ) { - let _guard = FROM_ENV_LOCK.lock().unwrap(); - let saved: Vec<(&str, Option)> = vars - .iter() - .map(|(key, _)| (*key, std::env::var_os(key))) - .collect(); + let mut env = crate::test_env_lock::TestEnv::lock_blocking(); for (key, value) in vars { match value { - Some(value) => std::env::set_var(key, value), - None => std::env::remove_var(key), + Some(value) => env.set(key, value), + None => env.remove(key), } } probe(&super::ClaudeNativeNameAdapter::from_env()); - for (key, value) in saved { - match value { - Some(value) => std::env::set_var(key, value), - None => std::env::remove_var(key), - } - } } /// T3-I4: the packaged app spawns the server with `FRESHELL_CLAUDE_SIDECAR` diff --git a/crates/freshell-server/src/session_names.rs b/crates/freshell-server/src/session_names.rs index 9ac1619df..86bde496f 100644 --- a/crates/freshell-server/src/session_names.rs +++ b/crates/freshell-server/src/session_names.rs @@ -32,7 +32,9 @@ //! document fences every operation instead of silently reinitializing over //! accepted names. The executing blocking-IO owner retains its guard until //! completion even if an outer caller stops waiting (transactions run on -//! detached `spawn_blocking` tasks). +//! detached `spawn_blocking` tasks), and releasing the guard unlocks the +//! sidecar explicitly, so a child process forked mid-transaction (which +//! holds an inherited descriptor until it execs) never keeps it locked. //! //! `session.name.updated` publishes only after successful commit/adoption, //! in local generation order (publication happens while the lock is held). @@ -571,7 +573,43 @@ pub struct SessionNames { // Tasks 3–4 seam: no caller until the background worker lands. #[allow(dead_code)] pub(crate) struct BackgroundGuard { - _file: std::fs::File, + _lock: HeldSidecarLock, +} + +/// A held `flock` on one of the store's sidecar lock files. +/// +/// The lock belongs to the open file description, and every descriptor that +/// shares that description keeps it alive. A child process forked by ANY +/// thread of this server (PTYs, provider sidecars, Git) while the lock is +/// held inherits such a descriptor and keeps it until the child execs, so +/// closing only this process's descriptor would leave the lock held after +/// the transaction (or worker hold) has ended — another participant, or +/// this store's own next transaction, would find it busy. Dropping the +/// guard therefore unlocks the description explicitly, which releases it +/// for every descriptor that shares it, and only then closes the handle. +struct HeldSidecarLock { + file: std::fs::File, + path: PathBuf, +} + +impl Drop for HeldSidecarLock { + fn drop(&mut self) { + if let Err(error) = self.file.unlock() { + // Closing the handle below still releases the lock once every + // inherited descriptor is gone; report the slower path. + tracing::warn!( + target: "freshell_server::session_names", + op = "unlock_sidecar", + name_ref = "-", + revision = 0, + class = "sidecar_unlock_failed", + lock_path = %self.path.display(), + error = %error, + "session_names.sidecar_unlock_failed: explicit unlock failed; the lock \ + releases when the last inherited descriptor closes" + ); + } + } } impl std::fmt::Debug for BackgroundGuard { @@ -640,7 +678,12 @@ impl SessionNames { pub(crate) fn try_background_guard(&self) -> Result, NameError> { let file = open_lock_file(&self.core.worker_lock_path)?; match file.try_lock() { - Ok(()) => Ok(Some(BackgroundGuard { _file: file })), + Ok(()) => Ok(Some(BackgroundGuard { + _lock: HeldSidecarLock { + file, + path: self.core.worker_lock_path.clone(), + }, + })), Err(TryLockError::WouldBlock) => Ok(None), Err(TryLockError::Error(e)) => Err(NameError::LockUnavailable(format!( "worker lock error: {e}" @@ -1778,14 +1821,21 @@ fn open_lock_file(path: &Path) -> Result { /// retrying contention for at most one monotonic second. Callers hold their /// store's [`LocalTurn`], so contention seen here comes from another /// participant (another process, or another store opened on the same -/// directory), never from this store's own queue. The sidecar is never -/// replaced or unlinked — its inode lives as long as the data dir. -fn acquire_document_lock(path: &Path) -> Result { +/// directory), never from this store's own queue — nor from a child process +/// forked during an earlier transaction, because every release unlocks +/// explicitly ([`HeldSidecarLock`]). The sidecar is never replaced or +/// unlinked — its inode lives as long as the data dir. +fn acquire_document_lock(path: &Path) -> Result { let file = open_lock_file(path)?; let deadline = Instant::now() + LOCK_RETRY_BUDGET; loop { match file.try_lock() { - Ok(()) => return Ok(file), + Ok(()) => { + return Ok(HeldSidecarLock { + file, + path: path.to_path_buf(), + }) + } Err(TryLockError::WouldBlock) => { if Instant::now() >= deadline { return Err(NameError::LockUnavailable( diff --git a/crates/freshell-server/src/session_names_tests.rs b/crates/freshell-server/src/session_names_tests.rs index ae3828631..5b55e0a11 100644 --- a/crates/freshell-server/src/session_names_tests.rs +++ b/crates/freshell-server/src/session_names_tests.rs @@ -1194,6 +1194,76 @@ async fn cancelled_wait_retains_executing_lock() { assert_eq!(third.expect("third").record.name, "Third writer"); } +/// A transaction's commit frees the document lock even when a child process +/// forked mid-transaction still holds an inherited duplicate of the locked +/// file description: the store unlocks that description at release instead +/// of only closing its own descriptor (which leaves the child's copy holding +/// the lock until the child execs — another participant, or this store's +/// next transaction, would then find the lock busy after the transaction +/// already completed). +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_child_forked_mid_transaction_does_not_keep_the_document_lock() { + let dir = temp_data_dir(); + let store = open_store(dir.path()); + let target = pending("h-forked"); + ensure(&store, "h-forked", NamedProvider::Claude, Some("/w/forked")) + .await + .expect("ensure pending"); + + set_test_hooks(dir.path(), vec![TestHook::HoldLockUntilReleased]); + let rename = tokio::spawn({ + let store = Arc::clone(&store); + let target = target.clone(); + async move { rename_user(&store, target, "Renamed").await } + }); + wait_for_lock_held_sentinel(dir.path()).await; + + // Fork while the transaction holds the document lock. + let child = crate::test_parked_child::ParkedChild::fork_now(); + release_held_lock(dir.path()); + let renamed = rename + .await + .expect("rename task") + .expect("the held transaction commits"); + assert_eq!(renamed.record.name, "Renamed"); + + let probe = open_lock_file(&dir.path().join(LOCK_FILE_NAME)).expect("open sidecar probe"); + let free = probe.try_lock().is_ok(); + drop(probe); + child.release(); + assert!( + free, + "a committed transaction leaves the document lock held by a child forked mid-transaction" + ); +} + +/// The background worker lock has the same release contract: dropping the +/// guard frees the lock even while a child forked during the hold still +/// holds an inherited duplicate, so the next worker cycle (or another +/// cooperating process) can take it. +#[cfg(unix)] +#[tokio::test] +async fn a_child_forked_while_the_worker_lock_is_held_does_not_keep_it() { + let dir = temp_data_dir(); + let store = open_store(dir.path()); + let guard = store + .try_background_guard() + .expect("worker lock opens") + .expect("worker lock is free"); + + let child = crate::test_parked_child::ParkedChild::fork_now(); + drop(guard); + let reacquired = store + .try_background_guard() + .expect("worker lock opens again"); + child.release(); + assert!( + reacquired.is_some(), + "a dropped worker guard leaves the worker lock held by a child forked during the hold" + ); +} + /// Local serialization (plan Task 1, item 5): the bounded one-second retry /// budget is for contention with ANOTHER process. A transaction of this same /// process that holds the document lock longer than that (a durable write diff --git a/crates/freshell-server/src/test_env_lock.rs b/crates/freshell-server/src/test_env_lock.rs index ea79b139f..b25ecf7cc 100644 --- a/crates/freshell-server/src/test_env_lock.rs +++ b/crates/freshell-server/src/test_env_lock.rs @@ -1,46 +1,161 @@ -//! Test-only helper: the crate-wide lock for tests that mutate or read the -//! process-global `CLAUDE_HOME` / `CLAUDE_CONFIG_DIR` environment in the +//! Test-only: the ONE owner of the process environment in the //! `--bin freshell-server` test binary. //! -//! Why the lock exists: `session_directory::claude_home` (and the transcript -//! locator's candidate roots) let a process-global `CLAUDE_HOME` WIN over a -//! test's explicit home — the production override parity. So when one test -//! holds `CLAUDE_HOME` pointed at its own temp home, every OTHER test whose -//! production-code path resolves through that resolver inspects a FOREIGN -//! temp home. The final-suite gate observed exactly this race: -//! `claude_unreadable_projects_root_defers` / -//! `claude_unreadable_project_subdir_defers` (`main.rs`) chmod their own -//! `~/.claude/projects` to 000 and assert the tombstone gate DEFERs, but a -//! concurrently running discovery test's `CLAUDE_HOME` (whose projects tree -//! is readable and holds no `sess-1.jsonl`) made the gate answer -//! "definitively absent" and fail the must-DEFER assert. Both fail -//! deterministically solo; the pristine base (6ee5cf4b) was clean only -//! because the mutators — the native route-discovery tests — landed with the -//! unified-agent-names branch, and the ~150 added tests widened the parallel -//! window until the race fired. +//! Cargo runs a binary's tests as parallel threads of one process, and the +//! environment is process-global. Production code resolves provider homes +//! and feature switches from it (`HOME`/`USERPROFILE`, `CLAUDE_HOME`, +//! `CLAUDE_CONFIG_DIR`, `CODEX_HOME`, `XDG_DATA_HOME`, `KILROY_ENABLED`, the +//! `FRESHELL_*` knobs, ...), and some of those resolvers let a process-global +//! value WIN over a test's explicit home — production override parity. So a +//! test that points `HOME` or `CLAUDE_CONFIG_DIR` at its own temp home +//! changes what every concurrently running test resolves, and two tests +//! mutating overlapping variables under DIFFERENT locks (the old +//! `HOME_ENV_TEST_LOCK` / `CLAUDE_ENV_TEST_LOCK` / per-module locks) still +//! interleave. Observed: `claude_unreadable_projects_root_defers` flipping to +//! "definitively absent" under a concurrent discovery test's `CLAUDE_HOME`, +//! and the managed-runtime snapshot-outage test (`HOME`, `CODEX_HOME`, +//! `CLAUDE_CONFIG_DIR`, `XDG_DATA_HOME` under a module-local lock) colliding +//! with the `HOME` and `CLAUDE_*` mutators. //! -//! The discipline (mirroring `sessions_tests.rs`'s pin-roots note and the -//! crate-wide `HOME_ENV_TEST_LOCK` convention): +//! The discipline: //! -//! * every test that MUTATES `CLAUDE_HOME`/`CLAUDE_CONFIG_DIR` -//! process-globally holds this lock for the whole set…restore window (the -//! restore happens under the guard, before it releases); -//! * every test READER that cannot pin its root directly — one whose -//! production-code path resolves through `claude_home` (the -//! `transcript_definitively_absent` family, the `list_claude_sessions` -//! fixtures) — MUST also hold it: async tests acquire with -//! `lock().await`, plain `#[test]` fns with `blocking_lock()` (never call -//! `blocking_lock` from inside a runtime — it panics); -//! * readers that pin their root DIRECTLY — constructing provider sources -//! from explicit paths instead of `claude_home(&home)`, like -//! `sessions_tests` — don't need it. +//! * every test that changes the environment does so ONLY through a +//! [`TestEnv`], which holds the single crate-wide lock for its whole life +//! and restores every variable it touched when dropped — also while a +//! failing assertion unwinds; +//! * every test whose production path READS a variable another test may +//! change (and that cannot pin its inputs explicitly) holds a [`TestEnv`] +//! too, without changing anything; +//! * async tests acquire with [`TestEnv::lock`], plain `#[test]` fns with +//! [`TestEnv::lock_blocking`] (never from inside a runtime — it panics); +//! * a test takes ONE `TestEnv` and passes it down; helpers never lock again +//! (the lock is not re-entrant). //! -//! Lock order: tests that need BOTH this lock and `HOME_ENV_TEST_LOCK` (the -//! `claude_exact_id_fallback` family mutates `HOME`/`USERPROFILE` too) take -//! `HOME_ENV_TEST_LOCK` FIRST, then this one — never the reverse. - -/// The crate-wide `CLAUDE_HOME`/`CLAUDE_CONFIG_DIR` test-env lock. A -/// `tokio::sync::Mutex` because the discovery mutators are async and hold it -/// across `.await` points (a `std` guard there would trip clippy's -/// `await_holding_lock`); sync readers use `blocking_lock()`. -pub(crate) static CLAUDE_ENV_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +//! Tests that need a different environment for long-lived child state run +//! the scenario in a child process instead (for example the managed-runtime +//! re-plan tests), which isolates it without any lock. + +use std::ffi::{OsStr, OsString}; + +/// The crate-wide process-environment lock. A `tokio::sync::Mutex` so async +/// tests can hold it across `.await` points and sync tests can take it with +/// `blocking_lock`; it has no poisoning, so a failed test never wedges the +/// others (its [`TestEnv`] restores the variables while unwinding). +static PROCESS_ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +/// Exclusive use of the process environment for one test. Every variable +/// changed through it is restored to its value from before the first change +/// when the guard drops (in reverse order), and only then is the lock +/// released. +#[must_use = "the environment is only owned while the TestEnv is alive"] +pub(crate) struct TestEnv { + saved: Vec<(OsString, Option)>, + _lock: tokio::sync::MutexGuard<'static, ()>, +} + +impl TestEnv { + /// Take the environment from an async test. + pub(crate) async fn lock() -> Self { + Self::holding(PROCESS_ENV_LOCK.lock().await) + } + + /// Take the environment from a plain `#[test]` fn (never from inside a + /// tokio runtime). + pub(crate) fn lock_blocking() -> Self { + Self::holding(PROCESS_ENV_LOCK.blocking_lock()) + } + + fn holding(lock: tokio::sync::MutexGuard<'static, ()>) -> Self { + Self { + saved: Vec::new(), + _lock: lock, + } + } + + /// Set `key` for the rest of this test. + pub(crate) fn set(&mut self, key: impl AsRef, value: impl AsRef) { + let key = key.as_ref(); + self.remember(key); + std::env::set_var(key, value); + } + + /// Unset `key` for the rest of this test. + pub(crate) fn remove(&mut self, key: impl AsRef) { + let key = key.as_ref(); + self.remember(key); + std::env::remove_var(key); + } + + fn remember(&mut self, key: &OsStr) { + if !self.saved.iter().any(|(saved, _)| saved == key) { + self.saved.push((key.to_os_string(), std::env::var_os(key))); + } + } +} + +impl Drop for TestEnv { + fn drop(&mut self) { + for (key, value) in self.saved.drain(..).rev() { + match value { + Some(value) => std::env::set_var(&key, value), + None => std::env::remove_var(&key), + } + } + } +} + +#[cfg(test)] +mod tests { + use super::TestEnv; + + // One variable per test: the tests run concurrently, and each spans + // several guard lifetimes. + const RESTORE_PROBE: &str = "FRESHELL_TEST_ENV_RESTORE_PROBE"; + const UNWIND_PROBE: &str = "FRESHELL_TEST_ENV_UNWIND_PROBE"; + + #[test] + fn a_test_env_restores_the_value_from_before_its_first_change() { + // An ambient value the next guard finds already present (set + // directly, so no guard owns restoring it). + let ambient = TestEnv::lock_blocking(); + std::env::set_var(RESTORE_PROBE, "ambient"); + drop(ambient); + + let mut env = TestEnv::lock_blocking(); + env.set(RESTORE_PROBE, "first"); + env.set(RESTORE_PROBE, "second"); + env.remove(RESTORE_PROBE); + assert_eq!(std::env::var_os(RESTORE_PROBE), None); + drop(env); + + let check = TestEnv::lock_blocking(); + let restored = std::env::var(RESTORE_PROBE); + std::env::remove_var(RESTORE_PROBE); + drop(check); + assert_eq!( + restored.as_deref(), + Ok("ambient"), + "every change is undone back to the value from before the first one" + ); + } + + #[test] + fn a_failing_test_still_restores_its_environment() { + let before = { + let _env = TestEnv::lock_blocking(); + std::env::var_os(UNWIND_PROBE) + }; + let result = std::panic::catch_unwind(|| { + let mut env = TestEnv::lock_blocking(); + env.set(UNWIND_PROBE, "set by a test that then fails"); + panic!("the test fails while it owns the environment"); + }); + assert!(result.is_err()); + let _env = TestEnv::lock_blocking(); + assert_eq!( + std::env::var_os(UNWIND_PROBE), + before, + "the unwinding guard restored the variable" + ); + } +} diff --git a/crates/freshell-server/src/test_parked_child.rs b/crates/freshell-server/src/test_parked_child.rs new file mode 100644 index 000000000..0128b6455 --- /dev/null +++ b/crates/freshell-server/src/test_parked_child.rs @@ -0,0 +1,90 @@ +//! Test-only: a child process parked between fork and exec. +//! +//! Every descriptor of a process is duplicated into a child at fork and +//! closed (close-on-exec) only when the child execs. A lock or socket that +//! a test thread "releases" by closing its descriptor therefore stays alive +//! while ANY concurrently forked child sits in that window — which is how +//! PTY, sidecar and Git spawns on other threads kept the session-names +//! sidecar lock held and a retired listener listening. Tests use a parked +//! child to hold that window open deterministically. + +/// A child process forked by ANY thread of this process, parked between fork +/// and exec. Until it execs it holds a duplicate of every descriptor this +/// process had open at the fork — a sidecar lock a naming transaction held, +/// a listening socket about to be retired — exactly the window every PTY, +/// sidecar or Git spawn opens on another thread. +pub(crate) struct ParkedChild { + release: std::os::unix::net::UnixStream, + spawner: Option>>, +} + +impl ParkedChild { + /// Fork a child now and return once it is parked before its exec. + pub(crate) fn fork_now() -> Self { + use std::io::Read; + use std::os::fd::AsRawFd; + use std::os::unix::process::CommandExt; + + let (mut parent_end, child_end) = + std::os::unix::net::UnixStream::pair().expect("parked-child socket pair"); + let child_fd = child_end.as_raw_fd(); + let mut command = std::process::Command::new("true"); + // SAFETY: the closure runs in the forked child before exec and only + // makes write(2)/read(2) calls on an inherited socket — both + // async-signal-safe, with no allocation and no locks. + unsafe { + command.pre_exec(move || { + let parked = [b'p']; + libc::write(child_fd, parked.as_ptr().cast(), 1); + let mut go = [0u8]; + libc::read(child_fd, go.as_mut_ptr().cast(), 1); + Ok(()) + }); + } + // `spawn` returns only once the child has exec'd, so it runs on its + // own thread while the test holds the child parked. + let spawner = std::thread::spawn(move || { + let status = command.spawn()?.wait(); + drop(child_end); + status + }); + let mut parked = [0u8]; + parent_end + .read_exact(&mut parked) + .expect("the forked child reports that it is parked before exec"); + Self { + release: parent_end, + spawner: Some(spawner), + } + } + + /// Let the child exec (closing its inherited descriptors) and exit. + pub(crate) fn release(mut self) { + self.unpark(); + let status = self + .spawner + .take() + .expect("a parked child is released once") + .join() + .expect("parked-child spawner thread") + .expect("the parked child runs to completion"); + assert!(status.success(), "the parked child exits cleanly"); + } + + fn unpark(&mut self) { + use std::io::Write; + // The child holds its own copy of this end, so it would never see + // EOF: it is released only by this byte. + let _ = self.release.write_all(b"g"); + } +} + +impl Drop for ParkedChild { + /// A test that fails before `release` still lets its child exec and exit + /// (the spawner thread reaps it) instead of leaving it parked forever. + fn drop(&mut self) { + if self.spawner.is_some() { + self.unpark(); + } + } +} diff --git a/crates/freshell-server/src/test_trace_capture.rs b/crates/freshell-server/src/test_trace_capture.rs index 0e51d27f4..29e284d39 100644 --- a/crates/freshell-server/src/test_trace_capture.rs +++ b/crates/freshell-server/src/test_trace_capture.rs @@ -1,3 +1,26 @@ +//! The test binary's one process-wide tracing subscriber. +//! +//! tracing-core caches each callsite's `Interest` process-wide the first time +//! the callsite fires, computed from the dispatchers registered at that +//! moment, and two windows cache `Interest::never` for the rest of the +//! process (every later event from that callsite is dropped before any +//! subscriber sees it): +//! +//! * a callsite first fired by any thread while the global subscriber is +//! being installed — registered (so the level filter already admits the +//! event) but not yet the default the interest is computed from. A capture +//! installed lazily by the first test that asks for it can therefore lose a +//! callsite that another test's thread happens to fire at that moment. +//! * a callsite first fired by a thread outside a test's scoped +//! (`with_subscriber` / `set_default`) subscriber while that scoped +//! subscriber is the only registered dispatcher: the interest is computed +//! from the firing thread's own default, which is no subscriber at all. +//! +//! Installing the capture subscriber before `main` — before any test thread +//! exists — closes both windows: every callsite's interest is computed with +//! this subscriber present (it is interested in every event), and a scoped +//! subscriber is never the only registered dispatcher. + use std::collections::BTreeMap; use std::sync::{Arc, Mutex, OnceLock}; @@ -17,24 +40,56 @@ impl CapturedTraceEvents { } } -/// One process-wide collector is shared by tests that need to inspect tracing -/// events. Keeping the subscriber and buffer here prevents modules from racing -/// to install separate global subscribers with disconnected event buffers. +/// Install the process-wide capture subscriber at program load, before the +/// test harness starts any test thread (see the module docs for why it must +/// precede every callsite's first event). +// SAFETY: runs before `main` on the loading thread, with no other thread in +// existence. It only allocates and installs the tracing global default; it +// touches no stdio, thread-locals or runtime state. +#[ctor::ctor] +unsafe fn install_process_trace_capture() { + process_capture(); +} + +/// The shared collector of the process-wide capture subscriber. Tests that +/// inspect tracing events read it and filter by values unique to their own +/// fixture, because every test in the binary feeds the same buffer. pub(crate) fn captured_trace_events() -> CapturedTraceEvents { + process_capture().clone() +} + +fn process_capture() -> &'static CapturedTraceEvents { use tracing_subscriber::prelude::*; static GLOBAL_EVENTS: OnceLock = OnceLock::new(); - GLOBAL_EVENTS - .get_or_init(|| { - let events = CapturedTraceEvents::default(); - let subscriber = tracing_subscriber::registry().with(CaptureTraceLayer { - events: events.clone(), - }); - tracing::subscriber::set_global_default(subscriber) - .expect("test trace capture must install the sole test subscriber"); - events - }) - .clone() + GLOBAL_EVENTS.get_or_init(|| { + let events = CapturedTraceEvents::default(); + let subscriber = tracing_subscriber::registry().with(CaptureTraceLayer { + events: events.clone(), + }); + tracing::subscriber::set_global_default(subscriber) + .expect("test trace capture must install the sole test subscriber"); + events + }) +} + +/// A test that must see events this collector does not keep registers an +/// observer instead of installing a second global subscriber (which the +/// already-installed capture makes impossible). Observers run for every +/// event, from whichever thread emits it, and must not emit events +/// themselves. +#[cfg(feature = "managed-provider-qualification")] +pub(crate) fn observe_all_events(observer: impl Fn(&tracing::Event<'_>) + Send + Sync + 'static) { + event_observers().lock().unwrap().push(Box::new(observer)); +} + +#[cfg(feature = "managed-provider-qualification")] +type EventObserver = Box) + Send + Sync>; + +#[cfg(feature = "managed-provider-qualification")] +fn event_observers() -> &'static Mutex> { + static OBSERVERS: OnceLock>> = OnceLock::new(); + OBSERVERS.get_or_init(|| Mutex::new(Vec::new())) } struct CaptureTraceLayer { @@ -65,27 +120,81 @@ impl tracing::field::Visit for TraceFieldVisitor { } } +/// The events this collector keeps. +fn is_captured(metadata: &tracing::Metadata<'_>) -> bool { + let level = *metadata.level(); + match metadata.target() { + "freshell_server::extensions" => level == tracing::Level::WARN, + "freshell_server::session_directory" => level == tracing::Level::ERROR, + "freshell_server::session_names" => { + level == tracing::Level::WARN || level == tracing::Level::INFO + } + _ => false, + } +} + +/// Every event an observer may need to see (`observe_all_events`). +#[cfg(feature = "managed-provider-qualification")] +fn is_observed(_metadata: &tracing::Metadata<'_>) -> bool { + true +} + +#[cfg(not(feature = "managed-provider-qualification"))] +fn is_observed(metadata: &tracing::Metadata<'_>) -> bool { + is_captured(metadata) +} + impl tracing_subscriber::Layer for CaptureTraceLayer where S: tracing::Subscriber, { + /// Never `never`: a callsite this subscriber does not keep answers + /// `sometimes`, so the decision is made per event against whichever + /// subscriber is current — a test's scoped subscriber still receives + /// the events it is waiting for, whichever thread fired the callsite + /// first — while this subscriber rejects them cheaply in `enabled` + /// instead of every event in the process being built and dispatched. + fn register_callsite( + &self, + metadata: &'static tracing::Metadata<'static>, + ) -> tracing::subscriber::Interest { + if is_observed(metadata) { + tracing::subscriber::Interest::always() + } else { + tracing::subscriber::Interest::sometimes() + } + } + + fn enabled( + &self, + metadata: &tracing::Metadata<'_>, + _context: tracing_subscriber::layer::Context<'_, S>, + ) -> bool { + is_observed(metadata) + } + + /// The kept events are INFO and above, so while no scoped subscriber + /// asks for more, `debug!`/`trace!` callsites stay off entirely. + #[cfg(not(feature = "managed-provider-qualification"))] + fn max_level_hint(&self) -> Option { + Some(tracing::level_filters::LevelFilter::INFO) + } + fn on_event( &self, event: &tracing::Event<'_>, _context: tracing_subscriber::layer::Context<'_, S>, ) { + #[cfg(feature = "managed-provider-qualification")] + for observer in event_observers().lock().unwrap().iter() { + observer(event); + } + let metadata = event.metadata(); - let level = *metadata.level(); - let target = metadata.target(); - let is_extension_warning = - target == "freshell_server::extensions" && level == tracing::Level::WARN; - let is_session_directory_error = - target == "freshell_server::session_directory" && level == tracing::Level::ERROR; - let is_session_names_event = target == "freshell_server::session_names" - && (level == tracing::Level::WARN || level == tracing::Level::INFO); - if !is_extension_warning && !is_session_directory_error && !is_session_names_event { + if !is_captured(metadata) { return; } + let is_session_directory_error = metadata.target() == "freshell_server::session_directory"; let mut visitor = TraceFieldVisitor::default(); event.record(&mut visitor); @@ -98,9 +207,61 @@ where } self.events.0.lock().unwrap().push(CapturedTraceEvent { - level, + level: *metadata.level(), message: visitor.message, fields: visitor.fields, }); } } + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + const PROBE_TARGET: &str = "freshell_server::test_trace_capture::scoped_probe"; + + /// The callsite under test: one `warn!` that two threads fire. + fn fire_probe() { + tracing::warn!(target: PROBE_TARGET, "scoped capture probe"); + } + + struct CountProbe(Arc); + + impl tracing_subscriber::Layer for CountProbe { + fn on_event( + &self, + event: &tracing::Event<'_>, + _context: tracing_subscriber::layer::Context<'_, S>, + ) { + if event.metadata().target() == PROBE_TARGET { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + } + + /// A test's scoped subscriber receives a callsite's event even though + /// another thread, outside any scoped subscriber, fired that callsite + /// first while the scoped subscriber was registered. Without a + /// process-wide subscriber in place, that first firing caches the + /// callsite as uninterested for the whole process and the scoped + /// subscriber never sees it (`review_r6_automatic_checkpoint_cooldown_is_logged` + /// captured nothing this way). + #[test] + fn a_scoped_subscriber_sees_a_callsite_another_thread_fired_first() { + use tracing_subscriber::prelude::*; + + let seen = Arc::new(AtomicUsize::new(0)); + let scoped = + tracing::Dispatch::new(tracing_subscriber::registry().with(CountProbe(seen.clone()))); + std::thread::spawn(fire_probe) + .join() + .expect("the unscoped thread fires the callsite first"); + tracing::dispatcher::with_default(&scoped, fire_probe); + assert_eq!( + seen.load(Ordering::SeqCst), + 1, + "the scoped subscriber receives the event fired under it" + ); + } +} diff --git a/crates/freshell-ws/tests/term09_output_queue.rs b/crates/freshell-ws/tests/term09_output_queue.rs index 08b20daec..f2419fb8c 100644 --- a/crates/freshell-ws/tests/term09_output_queue.rs +++ b/crates/freshell-ws/tests/term09_output_queue.rs @@ -570,14 +570,18 @@ struct ThrottledDrain { /// consumer (the incident's shape: a real browser on a slow path that keeps /// draining, never a dead socket). After every `chunk_bytes` of output /// payload received, pause `chunk_pause`, so the sustained drain rate is -/// ~`chunk_bytes / chunk_pause`. Stops at `marker`, connection end, or the -/// deadline (whichever first). +/// ~`chunk_bytes / chunk_pause`. While `trickle_until` (when given) has not +/// yet reported `true`, the consumer is slower still: it pauses after EVERY +/// frame, so it reads one frame (at most 16 KiB) per `chunk_pause` — never +/// stuck, but too slow to keep a flood's backlog down. Stops at `marker`, +/// connection end, or the deadline (whichever first). async fn drain_throttled_until_marker( ws: &mut TestWs, marker: &str, chunk_bytes: usize, chunk_pause: Duration, deadline: tokio::time::Instant, + trickle_until: Option<&dyn Fn() -> bool>, ) -> ThrottledDrain { let mut report = ThrottledDrain { text: String::new(), @@ -587,12 +591,14 @@ async fn drain_throttled_until_marker( closed: false, }; let mut received_since_pause = 0usize; + let mut trickling = trickle_until.is_some(); // Rolling tail for marker detection: a marker can only arrive in fresh // data (possibly straddling a frame boundary), so scanning the whole // accumulated text per message would be quadratic — at incident-scale // flood sizes that alone throttles the drain and masquerades as server - // slowness. The tail spans several marker lengths; a few hundred bytes - // bounds the scan at O(1) per message. + // slowness. Each scan covers the fresh data plus the previous data's + // last `marker.len() - 1` bytes, which is exactly what a straddling + // marker needs. let mut tail = String::new(); while tokio::time::Instant::now() < deadline { let remaining = deadline.saturating_duration_since(tokio::time::Instant::now()); @@ -619,7 +625,10 @@ async fn drain_throttled_until_marker( } report.output_bytes += data_len; received_since_pause += data_len; - if received_since_pause >= chunk_bytes { + if trickling { + trickling = !trickle_until.is_some_and(|done| done()); + } + if trickling || received_since_pause >= chunk_bytes { received_since_pause = 0; tokio::time::sleep(chunk_pause).await; } @@ -627,13 +636,17 @@ async fn drain_throttled_until_marker( Some("terminal.output.gap") => report.gaps.push(value), _ => {} } - if tail.len() > 4 * marker.len() { - let keep = tail.len() - 2 * marker.len(); - tail.drain(..keep); - } + // Scan before trimming: the frame that carries the marker + // usually carries more output after it (the shell prompt), + // and trimming first would cut the marker away. if tail.contains(marker) { break; } + let mut carried_from = tail.len().saturating_sub(marker.len() - 1); + while !tail.is_char_boundary(carried_from) { + carried_from -= 1; + } + tail.drain(..carried_from); } Ok(Some(Ok(WsMessage::Close(_)))) | Ok(None) | Ok(Some(Err(_))) => { report.closed = true; @@ -657,6 +670,12 @@ async fn drain_throttled_until_marker( #[tokio::test] async fn incident_backlog_spills_instead_of_disconnecting() { let _flood_guard = flood_test_serial().await; + // Install the process-global capture BEFORE the server spawns: the + // server's own spill event (below) is the signal that the backlog has + // crossed the spill bound, and an event emitted before the subscriber + // exists is lost for good. + let events = global_capture(); + let captured_from = events.lock().expect("capture lock").len(); // The real default configuration — this test pins the shipped defaults, // not an injected variant. let term09 = Term09Config::default(); @@ -665,9 +684,15 @@ async fn incident_backlog_spills_instead_of_disconnecting() { let mut creator = connect_and_complete_handshake(&url).await; let terminal_id = create_shell_terminal(&mut creator, "create-incident").await; - // The victim: a normal client that reads continuously but slowly - // (~1 MiB/s), like a browser on a congested path. It is never stuck — - // its socket keeps accepting writes the whole time. + // The victim: a normal client that reads continuously but slowly, like a + // browser on a congested path. It is never stuck — its socket keeps + // accepting writes the whole time. Until the server reports its first + // spill for this terminal it reads one frame per pause (at most + // ~64 KiB/s); from then on ~1 MiB/s. A fixed ~1 MiB/s from the start + // kept pace with the flood whenever a loaded machine slowed the PTY + // producer down to that rate, so the backlog never reached the spill + // bound this test is about; the trickle cannot drain more than a few MiB + // of a ~40 MB flood before the deadline, so the backlog provably does. let mut victim = connect_and_complete_handshake(&url).await; attach(&mut victim, &terminal_id, "attach-incident").await; tokio::time::sleep(Duration::from_millis(200)).await; @@ -690,6 +715,17 @@ async fn incident_backlog_spills_instead_of_disconnecting() { .await .expect("send flood input"); + let spilled = || { + events + .lock() + .expect("capture lock") + .iter() + .skip(captured_from) + .any(|event| { + event.message == "ws.terminal_stream.queue_overflow_spill" + && event.fields.get("terminal_id") == Some(&terminal_id) + }) + }; let deadline = started + Duration::from_secs(60); let report = drain_throttled_until_marker( &mut victim, @@ -697,6 +733,7 @@ async fn incident_backlog_spills_instead_of_disconnecting() { 256 * 1024, Duration::from_millis(250), deadline, + Some(&spilled), ) .await; let elapsed = started.elapsed(); @@ -784,9 +821,22 @@ async fn incident_backlog_spills_instead_of_disconnecting() { /// and the stall window is widened to 5 s so a loaded runner's scheduling /// gaps cannot masquerade as a genuine >2 s send silence. Every assertion /// is unchanged; only the reachability margins were re-derived. +/// +/// Even the minimal inversion needs production to outpace the drain by +/// 2 MiB, and a loaded machine still slowed the PTY producer to the +/// ~1 MiB/s drain rate often enough that the queue never overflowed (no gap, +/// 1 run in 4 at load ~100). So, like the incident test, the client reads +/// one frame per pause until the server reports its first spill for this +/// terminal, and only then drains at ~1 MiB/s. It keeps reading the whole +/// time, so every pause still ends in a successful send. #[tokio::test] async fn slow_but_progressing_client_survives_over_threshold_backlog() { let _flood_guard = flood_test_serial().await; + // Install the process-global capture BEFORE the server spawns: the + // server's own spill event is the signal that the queue has overflowed, + // and an event emitted before the subscriber exists is lost for good. + let events = global_capture(); + let captured_from = events.lock().expect("capture lock").len(); // Inverted ON PURPOSE (see doc comment): 2 MiB queue > 1 MiB threshold. // The 5 s stall window keeps the decision observable while a loaded // runner cannot fake a >2 s send silence; the harness's 30 s ping @@ -825,9 +875,22 @@ async fn slow_but_progressing_client_survives_over_threshold_backlog() { .await .expect("send flood input"); + let spilled = || { + events + .lock() + .expect("capture lock") + .iter() + .skip(captured_from) + .any(|event| { + event.message == "ws.terminal_stream.queue_overflow_spill" + && event.fields.get("terminal_id") == Some(&terminal_id) + }) + }; // Deadline sized for a ~630 KB/s effective drain under CI load (the // per-frame JSON parsing contends with everything else on the runner): - // ~27 MB then needs ~45 s, so 75 s keeps >1.5x headroom. + // ~27 MB then needs ~45 s, so 75 s keeps >1.5x headroom. The trickle + // before the first spill only shortens what is left to drain, since the + // spill evicts what the client did not read. let deadline = tokio::time::Instant::now() + Duration::from_secs(75); let report = drain_throttled_until_marker( &mut victim, @@ -835,6 +898,7 @@ async fn slow_but_progressing_client_survives_over_threshold_backlog() { 128 * 1024, Duration::from_millis(125), deadline, + Some(&spilled), ) .await;