sbx-mgr: build every test service from explicit test options

crates/libs/sbx/mgr/server/AGENTS.md

@@ -24,10 +24,13 @@ Follow "Privileged tests and the CI host" in `docs/rust/space-conventions.md`.
   (`pplx_space_node_mgr_client::connect()`, `connect_once()`, `lazy()`); give
   the code under test a test-served node-manager. Cancelling `shutdown_token`
   first protects only code paths that check it before dialing.
-- `SandboxManagerServiceOptions::default()` reads `SPACE_NODE_ID` or the
-  host's node-id file, uses the production host-control and
-  credential-manager sockets, and probes `fc.pid` under `SANDBOX_RUN_DIR`. Set
-  every field where a test reaches them; constructors read env regardless.
+- Build test services from `SandboxManagerServiceOptions::for_test()` or
+  `for_test_in(dir)` and override only the fields a test exercises.
+  `SandboxManagerServiceOptions::default()` is production wiring: it reads
+  `SPACE_NODE_ID` or the host's node-id file, uses the production host-control
+  and credential-manager sockets, probes `fc.pid` under `SANDBOX_RUN_DIR`, and
+  dials the host's node-manager and template-manager. Constructors read env
+  regardless.
 - Never call `db::open()`; use `db::test_conn()`.
 
 Module moves change log targets: see "Log targets" in

crates/libs/sbx/mgr/server/src/node_manager_transport/fake.rs

@@ -7,7 +7,7 @@
 
 use std::collections::HashMap;
 use std::marker::PhantomData;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 
 use futures_util::future::BoxFuture;
@@ -131,21 +131,68 @@ pub(crate) fn fixed_client(client: NodeManagerGrpcClient) -> NodeManagerConnecto
 }
 
 /// A connector whose clients all dial `socket`, a test-owned path nothing
-/// binds, so every node-manager call fails before reaching any server.
+/// binds, so every node-manager call fails before reaching any server. As on a
+/// node whose node-manager is down, `connect` and `connect_once` fail to dial
+/// and a lazy client's calls fail with `Unavailable`. `connect` dials once
+/// rather than waiting out the host connector's retry budget.
 pub(crate) fn unbound(socket: &Path) -> NodeManagerConnector {
-    fixed_client(lazy_client(socket))
+    NodeManagerConnector::new(Arc::new(Unbound(socket.to_owned())))
+}
+
+/// [`unbound`] on a socket in a fresh directory that nothing creates, for
+/// tests that build a subsystem context rather than a service.
+pub(crate) fn unbound_in_absent_dir() -> NodeManagerConnector {
+    unbound(
+        &std::env::temp_dir()
+            .join(format!("space-sbx-mgr-test-{}", uuid::Uuid::now_v7()))
+            .join("absent-node-manager.sock"),
+    )
 }
 
 fn lazy_client(socket: &Path) -> NodeManagerGrpcClient {
-    let channel = pplx_space_common_grpc::connect_uds_channel_lazy(
-        socket,
-        tonic::transport::Endpoint::from_static("http://[::]:50051"),
-    );
+    let channel = pplx_space_common_grpc::connect_uds_channel_lazy(socket, endpoint());
+    client(channel)
+}
+
+fn endpoint() -> tonic::transport::Endpoint {
+    tonic::transport::Endpoint::from_static("http://[::]:50051")
+}
+
+fn client(channel: tonic::transport::Channel) -> NodeManagerGrpcClient {
     NodeManagerGrpcClient::new(pplx_space_common_svc::bootstrap::wrap_grpc_client_channel(
         channel,
     ))
 }
 
+struct Unbound(PathBuf);
+
+impl Unbound {
+    async fn dial(&self) -> Result<NodeManagerGrpcClient, ConnectError> {
+        let channel = pplx_space_common_grpc::connect_uds_channel_with_endpoint(
+            "node-manager",
+            &self.0,
+            endpoint(),
+        )
+        .await?;
+        Ok(client(channel))
+    }
+}
+
+#[async_trait::async_trait]
+impl NodeManagerTransport for Unbound {
+    async fn connect(&self) -> Result<NodeManagerGrpcClient, ConnectError> {
+        self.dial().await
+    }
+
+    async fn connect_once(&self) -> Result<NodeManagerGrpcClient, ConnectError> {
+        self.dial().await
+    }
+
+    fn lazy(&self) -> NodeManagerGrpcClient {
+        lazy_client(&self.0)
+    }
+}
+
 struct InProcess(NodeManagerGrpcClient);
 
 #[async_trait::async_trait]

crates/libs/sbx/mgr/server/src/node_state_sync.rs

@@ -15,7 +15,7 @@ use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 use tonic::Request;
 
-const SKIP_NODE_PUBLISH_METADATA: &str = "x-space-skip-cluster-sync";
+pub(crate) const SKIP_NODE_PUBLISH_METADATA: &str = "x-space-skip-cluster-sync";
 const DIRTY_MARK_INITIAL_RETRY: std::time::Duration = std::time::Duration::from_secs(1);
 const DIRTY_MARK_MAX_RETRY: std::time::Duration = std::time::Duration::from_secs(300);
 const DIRTY_OUTBOX_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);

crates/libs/sbx/mgr/server/src/run/tests.rs

@@ -145,7 +145,7 @@ async fn recovery_wait_does_not_hold_the_dependency_work_gate() {
     for path in [&capability, &admission, &work] {
         std::fs::File::create(path).unwrap();
     }
-    let service = SandboxManagerService::new(crate::db::test_conn());
+    let service = SandboxManagerService::for_test(crate::db::test_conn());
     service
         .enable_dependency_restart_drain(&capability, &admission, &work)
         .unwrap();
@@ -774,7 +774,7 @@ async fn grpc_writer_during_pre_handoff_drain_returns_retry_on_successor() {
     let listener =
         pplx_space_common_svc::bind_unix_socket(&socket_path, UnixSocketOptions::new(0o600))
             .unwrap();
-    let service = SandboxManagerService::new(rusqlite::Connection::open_in_memory().unwrap());
+    let service = SandboxManagerService::for_test(rusqlite::Connection::open_in_memory().unwrap());
     let rpc_drain_token = CancellationToken::new();
     let shutdown_token = CancellationToken::new();
     let server_shutdown = shutdown_token.clone();
@@ -821,7 +821,7 @@ async fn grpc_in_flight_request_interrupted_at_handoff_is_not_marked_replayable(
     let listener =
         pplx_space_common_svc::bind_unix_socket(&socket_path, UnixSocketOptions::new(0o600))
             .unwrap();
-    let service = SandboxManagerService::new(rusqlite::Connection::open_in_memory().unwrap());
+    let service = SandboxManagerService::for_test(rusqlite::Connection::open_in_memory().unwrap());
     let rpc_drain_token = CancellationToken::new();
     let shutdown_token = CancellationToken::new();
     let server_shutdown = shutdown_token.clone();

crates/libs/sbx/mgr/server/src/sandbox_eviction/tests.rs

@@ -841,7 +841,7 @@ async fn evict_tick_below_engage_returns_before_candidate_scan() {
         [],
     )
     .unwrap();
-    let service = SandboxManagerService::new(conn);
+    let service = SandboxManagerService::for_test(conn);
     let cfg = EvictionConfig {
         enabled: true,
         engage_fraction: 0.50,
@@ -892,7 +892,7 @@ fn resident_rows_count_every_state() {
 async fn evict_tick_without_disk_usage_fails_without_scanning() {
     let conn = open_test_db();
     insert_sandbox(&conn, "sbx-cold", "suspended", 1_000);
-    let service = SandboxManagerService::new(conn);
+    let service = SandboxManagerService::for_test(conn);
     let cfg = EvictionConfig {
         enabled: true,
         engage_fraction: 0.0,
@@ -924,7 +924,7 @@ async fn evict_tick_does_not_connect_when_all_candidates_are_backed_off() {
     let conn = open_test_db();
     insert_sandbox(&conn, "sbx-backed-off", "suspended", 1_000);
     insert_pinned_artifact(&conn, "sbx-backed-off", &v7(1_000).to_string(), Some("123"));
-    let service = SandboxManagerService::new(conn);
+    let service = SandboxManagerService::for_test(conn);
     let cfg = EvictionConfig {
         enabled: true,
         engage_fraction: 0.50,
@@ -1308,7 +1308,7 @@ async fn legacy_reupload_refusals_classify_as_state_skips_or_retryable_failures(
     use crate::state::SandboxState;
     use pplx_space_common_core::types::SandboxId;
 
-    let service = SandboxManagerService::new(open_test_db());
+    let service = SandboxManagerService::for_test(open_test_db());
     let unknown = service
         .reupload_legacy_sandbox_for_eviction(&SandboxId::new().to_string())
         .await
@@ -1329,7 +1329,7 @@ async fn legacy_reupload_refusals_classify_as_state_skips_or_retryable_failures(
         let conn = open_test_db();
         let id = SandboxId::new().to_string();
         insert_sandbox(&conn, &id, state.as_str(), 0);
-        let service = SandboxManagerService::new(conn);
+        let service = SandboxManagerService::for_test(conn);
 
         let refused = service
             .reupload_legacy_sandbox_for_eviction(&id)
@@ -1350,7 +1350,7 @@ async fn legacy_reupload_refusals_classify_as_state_skips_or_retryable_failures(
     insert_sandbox(&conn, &id, SandboxState::Restoring.as_str(), 0);
     let service = SandboxManagerService::new_with_config(
         conn,
-        Default::default(),
+        crate::service::SandboxManagerServiceOptions::for_test(),
         crate::config::SandboxConfig {
             hugepages: true,
             ..Default::default()
@@ -1370,7 +1370,7 @@ async fn legacy_reupload_refusals_classify_as_state_skips_or_retryable_failures(
 /// Empty cluster-api wiring is operator-fixable, so eviction retries the
 /// reupload instead of skipping the sandbox.
 async fn assert_missing_wiring_reupload_is_retried() {
-    let service = SandboxManagerService::new(open_test_db());
+    let service = SandboxManagerService::for_test(open_test_db());
     let refused = service
         .reupload_legacy_sandbox_for_eviction(
             &pplx_space_common_core::types::SandboxId::new().to_string(),

crates/libs/sbx/mgr/server/src/service/activation_support/tests.rs

@@ -590,7 +590,7 @@ async fn restore_finalizer_retains_cleanup_until_state_persistence_succeeds() {
                 SandboxCleanupEpisodeId::new(),
             )
         };
-        let service = SandboxManagerService::new(conn);
+        let service = SandboxManagerService::for_test(conn);
         let temp = tempfile::tempdir().unwrap();
         let snapshot_dir = temp.path().join("snapshot");
         std::fs::create_dir(&snapshot_dir).unwrap();
@@ -631,7 +631,7 @@ async fn restore_finalizer_retains_cleanup_until_state_persistence_succeeds() {
 
 #[tokio::test]
 async fn cold_finalizer_retains_cleanup_until_state_persistence_succeeds() {
-    let service = SandboxManagerService::new(crate::db::test_conn());
+    let service = SandboxManagerService::for_test(crate::db::test_conn());
     let (cleanup_tx, cleanup_rx) = tokio::sync::oneshot::channel();
     let cleanup_guard = ActivationCleanupGuard::with_interruption_reason(
         SandboxRuntimeLabel::parse("runtime1").unwrap(),
@@ -655,7 +655,7 @@ async fn cold_finalizer_retains_cleanup_until_state_persistence_succeeds() {
 
 #[tokio::test]
 async fn cold_finalizer_disarms_cleanup_after_state_persistence_succeeds() {
-    let service = SandboxManagerService::new(crate::db::test_conn());
+    let service = SandboxManagerService::for_test(crate::db::test_conn());
     let (cleanup_tx, cleanup_rx) = tokio::sync::oneshot::channel();
     let cleanup_guard = ActivationCleanupGuard::with_interruption_reason(
         SandboxRuntimeLabel::parse("runtime1").unwrap(),

crates/libs/sbx/mgr/server/src/service/activation_support/tests/interception_recovery.rs

@@ -200,7 +200,7 @@ async fn exercise_recovery(restore: bool) {
         SandboxManagerServiceOptions {
             credential_manager_socket_path: node_socket,
             shutdown_token: shutdown.clone(),
-            ..Default::default()
+            ..SandboxManagerServiceOptions::for_test()
         },
     );
     let _gateway = super::super::super::gateway_identity::tests::GatewayNode::attach(&mut service);

crates/libs/sbx/mgr/server/src/service/adapter/tests.rs

@@ -4,7 +4,7 @@ use crate::service::activation_support::internal_request;
 
 fn test_service() -> SandboxManagerService {
     let connection = crate::db::test_conn();
-    SandboxManagerService::new(connection)
+    SandboxManagerService::for_test(connection)
 }
 
 fn expect_authorization_error<T>(result: Result<T, Status>) -> Status {

crates/libs/sbx/mgr/server/src/service/auth_lookup/tests.rs

@@ -21,7 +21,7 @@ async fn gateway_identity_mapping_preserves_exact_pid_and_lifecycle_authority()
     crate::db::update_netgtw_pid(&conn, &id, owner, 4242).unwrap();
     let service = SandboxManagerService::new_with_options(
         conn,
-        super::super::SandboxManagerServiceOptions::default(),
+        super::super::SandboxManagerServiceOptions::for_test(),
     );
     for (pid, uid, allowed) in [
         (4242, owner, true),

crates/libs/sbx/mgr/server/src/service/auth_lookup/wait_ready_policy_tests.rs

@@ -7,7 +7,7 @@ use pplx_space_common_svc::space_api_route::{
 };
 
 fn service() -> SandboxManagerService {
-    SandboxManagerService::new(rusqlite::Connection::open_in_memory().unwrap())
+    SandboxManagerService::for_test(rusqlite::Connection::open_in_memory().unwrap())
 }
 
 fn node_api_peer() -> AuthorizedUdsPeer {

crates/libs/sbx/mgr/server/src/service/auth_lookup_tests.rs

@@ -73,7 +73,7 @@ fn fixture_with_claims(
         rusqlite::Connection::open_in_memory().unwrap(),
         SandboxManagerServiceOptions {
             route_ed25519_verification_bundle: Some(Arc::new(bundle)),
-            ..Default::default()
+            ..SandboxManagerServiceOptions::for_test()
         },
     );
     let mut request = tonic::Request::new(());

crates/libs/sbx/mgr/server/src/service/cache_cleanup/periodic/tests.rs

@@ -16,7 +16,7 @@ async fn periodic_inventory_progresses_in_pages_without_treating_other_pages_as_
     }
     let stale = SandboxId::new();
     insert_test_sandbox(&conn, &stale.to_string(), SandboxState::Stopped);
-    let service = SandboxManagerService::new(conn);
+    let service = SandboxManagerService::for_test(conn);
     let mut scan = PeriodicInventoryScan::default();
     let shutdown = CancellationToken::new();
     let paths = SandboxInventoryPaths::new(cache.path(), runtime.path());
@@ -104,7 +104,7 @@ async fn periodic_inventory_names_only_rowless_cache_directories_for_inspection(
         std::fs::create_dir(cache.path().join(id.to_string())).unwrap();
     }
     insert_test_sandbox(&conn, &owned.to_string(), SandboxState::Suspended);
-    let service = SandboxManagerService::new(conn);
+    let service = SandboxManagerService::for_test(conn);
     let batch = service
         .discover_periodic_inventory_in(
             &mut PeriodicInventoryScan::default(),
@@ -190,7 +190,7 @@ async fn orphan_quarantine_survives_inventory_once_its_runtime_is_gone() {
         Some(SandboxQuarantineReasonCode::OrphanedSandboxRowMissing.as_str())
     );
     std::fs::remove_dir_all(&sandbox_run_dir).unwrap();
-    let service = crate::service::tests::service_tests::service_with_runtime_liveness(conn, false);
+    let service = crate::service::tests::service_tests::service_with_runtime_probe(conn, false);
     let before = database_rows(&service.db_conn().unwrap());
 
     reconcile_one_periodic_page(&service, paths);
@@ -267,8 +267,7 @@ async fn startup_and_periodic_inventory_remove_only_rows_whose_state_requires_lo
             .collect::<BTreeSet<_>>();
         assert!(!expected_removed.is_empty());
 
-        let service =
-            crate::service::tests::service_tests::service_with_runtime_liveness(conn, false);
+        let service = crate::service::tests::service_tests::service_with_runtime_probe(conn, false);
         let removed_or_published = if periodic {
             reconcile_one_periodic_page(&service, paths);
             crate::db::list_node_dirty_refs(&service.db_conn().unwrap(), 1024)

crates/libs/sbx/mgr/server/src/service/cache_cleanup/tests.rs

@@ -398,7 +398,7 @@ async fn row_cleanup_retires_policy_only_after_identity_is_deleted() {
             .unwrap();
             episode.id
         });
-        let service = SandboxManagerService::new(conn);
+        let service = SandboxManagerService::for_test(conn);
         let queue = &service.network_policy_reconcile_queue;
         queue.enqueue(id.clone());
         let old_attempt = queue.take_pending(&Default::default(), 1).pop().unwrap();
@@ -455,7 +455,7 @@ async fn inventory_retires_registries_atomically_and_retries_busy_registrations(
     let conn = crate::db::test_conn();
     let id = SandboxId::new().to_string();
     insert_test_sandbox(&conn, &id, SandboxState::Stopped);
-    let service = SandboxManagerService::new(conn);
+    let service = SandboxManagerService::for_test(conn);
     let cache = tempfile::tempdir().unwrap();
     let run = tempfile::tempdir().unwrap();
     service.network_policy_reconcile_queue.enqueue(id.clone());

crates/libs/sbx/mgr/server/src/service/cleanup_executor.rs

@@ -42,8 +42,6 @@ pub(super) struct CleanupExecutor {
     #[cfg(test)]
     waiter_read_barrier: Mutex<Option<Arc<TestAdmissionBarrier>>>,
     #[cfg(test)]
-    pub(super) run_dir: Mutex<Option<PathBuf>>,
-    #[cfg(test)]
     pub(super) cache_dir: Mutex<Option<PathBuf>>,
     wake: tokio::sync::Notify,
     pub(super) active: dashmap::DashMap<String, bool>,
@@ -69,8 +67,6 @@ impl Default for CleanupExecutor {
             #[cfg(test)]
             waiter_read_barrier: Mutex::new(None),
             #[cfg(test)]
-            run_dir: Mutex::new(None),
-            #[cfg(test)]
             cache_dir: Mutex::new(None),
             wake: tokio::sync::Notify::new(),
             active: dashmap::DashMap::new(),

crates/libs/sbx/mgr/server/src/service/cleanup_executor/tests.rs

@@ -50,11 +50,9 @@ async fn blocking_stop_follows_a_retryable_attempt_to_durable_completion() {
     let sandbox_id = SandboxId::new();
     let subject = sandbox_id.to_string();
     insert_test_sandbox(&conn, &subject, SandboxState::Stopping);
-    let sockets = tempfile::tempdir().unwrap();
     let service = crate::service::tests::node_manager_transport_tests::service_with_node_manager(
         conn,
         node_manager.connector(),
-        sockets.path(),
     );
     let execution_fence =
         crate::db::node_execution_fence(&service.db.lock().unwrap(), &subject).unwrap();
@@ -565,14 +563,12 @@ async fn lifecycle_and_job_completion_commit_together_and_recover(
     );
     // Virtual time from here on. A case that registered an activation episode
     // also registered the sandbox with the snapshot scheduler, whose first tick
-    // opens a node-manager connection; that connect does not watch shutdown, so
-    // the drain below has to wait out its whole `tokio::time` retry budget
-    // (~9.5 s against an absent socket) before the task can observe the
-    // cancellation. Pausing collapses the backoff without changing what the
-    // drain observes: the connects still run and still fail, and every
-    // assertion after this point reads state the drain is required to have
-    // settled. Only the wait is virtual, so everything above runs on the real
-    // clock exactly as the executor's own timers need.
+    // opens a node-manager connection; that connect does not watch shutdown.
+    // The test service's node-manager is an absent socket dialed once, so the
+    // connect fails at once, and pausing only keeps any later scheduler timer
+    // from holding the drain on the real clock. Every assertion after this
+    // point reads state the drain is required to have settled. Everything
+    // above runs on the real clock exactly as the executor's own timers need.
     tokio::time::pause();
     service.shutdown_token.cancel();
     service.task_tracker.close();

crates/libs/sbx/mgr/server/src/service/cleanup_targets.rs

@@ -206,10 +206,7 @@ impl SandboxManagerService {
             .map_err(|_| Status::unavailable("cleanup storage is shutting down"))?;
         #[cfg(test)]
         let run_dir = self
-            .cleanup_executor
-            .run_dir
-            .lock()
-            .unwrap()
+            .cleanup_run_dir
             .clone()
             .unwrap_or_else(|| PathBuf::from(RUN_DIR));
         let cache_dir = self.sandbox_cache_root();