diff --git a/crates/ployz/src/cloud_enroll.rs b/crates/ployz/src/cloud_enroll.rs
index f049df9472..1218a1af3b 100644
--- a/crates/ployz/src/cloud_enroll.rs
+++ b/crates/ployz/src/cloud_enroll.rs
@@ -697,6 +697,11 @@ mod tests {
             let server = tokio::spawn(async move {
                 let (mut held, _) = listener.accept().await.unwrap();
                 read_http(&mut held).await;
+                // The request has reached the server, so its timeout is armed.
+                // Advance only that wait; use real time again for the retry's I/O.
+                tokio::time::pause();
+                tokio::time::advance(REQUEST_TIMEOUT).await;
+                tokio::time::resume();
                 // Keep the first response open until the retry has succeeded.
                 let (mut retry, _) = listener.accept().await.unwrap();
                 read_http(&mut retry).await;
diff --git a/crates/ployz/src/connect.rs b/crates/ployz/src/connect.rs
index 9d12d3f06e..5e09ff509d 100644
--- a/crates/ployz/src/connect.rs
+++ b/crates/ployz/src/connect.rs
@@ -837,3 +837,6 @@ fn rpc_error_code(code: tonic::Code) -> RpcErrorCode {
 #[cfg(test)]
 #[path = "connect/tests.rs"]
 mod tests;
+
+#[cfg(test)]
+pub(crate) mod test_support;
diff --git a/crates/ployz/src/connect/test_support.rs b/crates/ployz/src/connect/test_support.rs
new file mode 100644
index 0000000000..99606d1cde
--- /dev/null
+++ b/crates/ployz/src/connect/test_support.rs
@@ -0,0 +1,79 @@
+//! In-memory RPC transport for tests that advance Tokio time.
+
+use std::{
+    convert::Infallible,
+    future::Future,
+    sync::{Arc, Mutex},
+};
+
+use futures_util::StreamExt;
+use ployz_core::OpaquePayload;
+use tonic::{
+    Request, Response, Status,
+    codec::ProstCodec,
+    transport::{Channel, Server},
+};
+
+use super::{BoxProxyStream, Client, ConnectError, Connector};
+use crate::context::{Connection, ConnectionSource};
+
+struct Connected(Channel);
+
+#[tonic::async_trait]
+impl Connector for Connected {
+    async fn connect(&self, _: &Connection) -> Result<Channel, ConnectError> {
+        Ok(self.0.clone())
+    }
+    async fn dial_proxy(
+        &self,
+        _: &Connection,
+        _: &str,
+        _: &str,
+    ) -> Result<BoxProxyStream, ConnectError> {
+        unreachable!("this RPC fixture does not open a proxy")
+    }
+}
+
+pub(crate) async fn rpc_client<F, Fut>(
+    rpc: F,
+) -> (
+    Client,
+    tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
+)
+where
+    F: Fn(Request<OpaquePayload>) -> Fut + Clone + Send + Sync + 'static,
+    Fut: Future<Output = Result<Response<OpaquePayload>, Status>> + Send + 'static,
+{
+    let rpc = tower::service_fn(rpc);
+    let service = tower::service_fn(move |request: http::Request<tonic::body::Body>| {
+        let rpc = rpc.clone();
+        async move {
+            Ok::<_, Infallible>(
+                tonic::server::Grpc::new(ProstCodec::default())
+                    .unary(rpc, request)
+                    .await,
+            )
+        }
+    });
+    // Both peers run inside Tokio; clock advancement cannot outrun socket I/O.
+    let (client_io, server_io) = tokio::io::duplex(64 * 1024);
+    let incoming =
+        tokio_stream::once(Ok::<_, std::io::Error>(server_io)).chain(tokio_stream::pending());
+    let server = tokio::spawn(Server::builder().serve_with_incoming(service, incoming));
+    let client_io = Arc::new(Mutex::new(Some(client_io)));
+    let channel = Channel::from_static("http://memory.invalid")
+        .connect_with_connector(tower::service_fn(move |_| {
+            std::future::ready(Ok::<_, std::io::Error>(hyper_util::rt::TokioIo::new(
+                client_io.lock().unwrap().take().unwrap(),
+            )))
+        }))
+        .await
+        .unwrap();
+    let client = Client::new(
+        channel.clone(),
+        Connection::tcp("127.0.0.1:1".parse().unwrap()),
+        ConnectionSource::Direct,
+        Arc::new(Connected(channel)),
+    );
+    (client, server)
+}
diff --git a/crates/ployz/src/global_catch_up_tests.rs b/crates/ployz/src/global_catch_up_tests.rs
index 16a8c915f8..8d6f5a1fa4 100644
--- a/crates/ployz/src/global_catch_up_tests.rs
+++ b/crates/ployz/src/global_catch_up_tests.rs
@@ -786,3 +786,106 @@ fn container_on(
     })
     .unwrap()
 }
+
+#[tokio::test(start_paused = true)]
+async fn real_catch_up_client_retries_readiness_and_placement_to_their_budget() {
+    use ployz_core::{
+        ContainerCreated, InspectTelemetry, LocalMachinePhase, MachineDetails, OpaquePayload,
+        RpcRequestBody, RpcResponse,
+    };
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
+    use std::time::Duration;
+    use tonic::{Request, Response, Status};
+
+    for placement in [false, true] {
+        for failures in [1, 4] {
+            let target = machine('1', "joiner");
+            let observed = target.clone();
+            let calls = Arc::new(AtomicUsize::new(0));
+            let attempts = calls.clone();
+            let request = EnsureGlobalSlotRequest {
+                project_name: ProjectName::parse("app").unwrap(),
+                resolved_spec: requested(ServiceMode::Global)
+                    .to_resolved(service_id('a'), ResolvedUpdateConfig::default())
+                    .unwrap(),
+            };
+            let expected = request.clone();
+            let (mut client, server) =
+                crate::connect::test_support::rpc_client(move |rpc: Request<OpaquePayload>| {
+                    let target = observed.clone();
+                    let expected = expected.clone();
+                    let attempt = attempts.fetch_add(1, Ordering::SeqCst);
+                    async move {
+                        assert_eq!(
+                            rpc.metadata().get(ployz_core::ONE_TARGET_HEADER).unwrap(),
+                            target.id.as_str()
+                        );
+                        #[expect(
+                            clippy::wildcard_enum_match_arm,
+                            reason = "fixture accepts only the two catch-up RPCs under test"
+                        )]
+                        let response = match rpc.into_inner().decode_request().unwrap().body {
+                            RpcRequestBody::Inspect(inspect) => {
+                                assert!(!placement);
+                                assert_eq!(inspect.telemetry, InspectTelemetry::BridgeCapacity);
+                                RpcResponse::from(MachineDetails {
+                                    id: target.id,
+                                    phase: LocalMachinePhase::Participating,
+                                    public_key: target.public_key,
+                                    advertised_endpoints: Vec::new(),
+                                    machine: Some(target),
+                                    store_version: Default::default(),
+                                    rtts: Vec::new(),
+                                    cloud_paired: false,
+                                    telemetry: None,
+                                    storage: None,
+                                })
+                            }
+                            RpcRequestBody::EnsureGlobalSlot(ensure) => {
+                                assert!(placement);
+                                assert_eq!(ensure, expected);
+                                RpcResponse::from(ContainerCreated {
+                                    container_id: container_id('a'),
+                                    display_name: "api".into(),
+                                })
+                            }
+                            other => panic!("unexpected catch-up RPC: {other:?}"),
+                        };
+                        if attempt < failures {
+                            Err(Status::unavailable("transient catch-up failure"))
+                        } else {
+                            Ok(Response::new(response.encode().unwrap()))
+                        }
+                    }
+                })
+                .await;
+            let started = tokio::time::Instant::now();
+            let succeeded = if placement {
+                CatchUpClient::ensure_global_slot(&mut client, &target.id, request)
+                    .await
+                    .is_ok()
+            } else {
+                CatchUpClient::bridge_capacity(&mut client, &target.id)
+                    .await
+                    .is_ok()
+            };
+            assert_eq!(succeeded, failures == 1);
+            assert_eq!(
+                calls.load(Ordering::SeqCst),
+                if failures == 1 { 2 } else { 4 }
+            );
+            assert_eq!(
+                started.elapsed(),
+                if failures == 1 {
+                    Duration::from_millis(500)
+                } else {
+                    Duration::from_secs(6)
+                }
+            );
+            server.abort();
+        }
+    }
+}
diff --git a/crates/ployz/src/handlers/machine/helpers.rs b/crates/ployz/src/handlers/machine/helpers.rs
index cd1eda8636..9de92b49ec 100644
--- a/crates/ployz/src/handlers/machine/helpers.rs
+++ b/crates/ployz/src/handlers/machine/helpers.rs
@@ -334,3 +334,7 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+#[path = "setup_tests.rs"]
+mod setup_tests;
diff --git a/crates/ployz/src/handlers/machine/setup_tests.rs b/crates/ployz/src/handlers/machine/setup_tests.rs
new file mode 100644
index 0000000000..c428b455d5
--- /dev/null
+++ b/crates/ployz/src/handlers/machine/setup_tests.rs
@@ -0,0 +1,129 @@
+//! Real setup RPC deadlines and initialization recovery under virtual time.
+
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+
+use crate::connect::Client;
+use ployz_core::{
+    InitializeRequest, InspectRequest, LocalMachinePhase, Machine, MachineDetails, MachineId,
+    MachineName, OpaquePayload, RpcRequestBody, RpcResponse, WireGuardPublicKey, op,
+};
+use tokio::time::Instant;
+use tonic::{Request, Response, Status};
+
+async fn starting_machine() -> (
+    Client,
+    Machine,
+    Arc<AtomicUsize>,
+    tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
+) {
+    let machine = Machine {
+        id: MachineId::random(),
+        name: MachineName::parse("founder").unwrap(),
+        subnet: "10.210.0.0/24".parse().unwrap(),
+        public_key: WireGuardPublicKey([2; 32]),
+        public_ip: None,
+        advertised_endpoints: Vec::new(),
+        runtime: Default::default(),
+    };
+    let initialized = Arc::new(AtomicUsize::new(0));
+    let calls = initialized.clone();
+    let observed = machine.clone();
+    let ready = Instant::now() + Duration::from_secs(70);
+    let rpc = move |request: Request<OpaquePayload>| {
+        let calls = calls.clone();
+        let machine = observed.clone();
+        async move {
+            #[expect(
+                clippy::wildcard_enum_match_arm,
+                reason = "this fixture rejects every RPC except Initialize and Inspect"
+            )]
+            match request.into_inner().decode_request().unwrap().body {
+                RpcRequestBody::Initialize(_) => {
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    std::future::pending::<Result<Response<OpaquePayload>, Status>>().await
+                }
+                RpcRequestBody::Inspect(_) => {
+                    let started = calls.load(Ordering::SeqCst) > 0;
+                    if started {
+                        tokio::time::sleep_until(ready).await;
+                    }
+                    let details = MachineDetails {
+                        id: machine.id,
+                        phase: if started {
+                            LocalMachinePhase::Participating
+                        } else {
+                            LocalMachinePhase::Uninitialized
+                        },
+                        public_key: machine.public_key,
+                        advertised_endpoints: Vec::new(),
+                        machine: started.then_some(machine),
+                        store_version: Default::default(),
+                        rtts: Vec::new(),
+                        cloud_paired: false,
+                        telemetry: None,
+                        storage: None,
+                    };
+                    Ok(Response::new(RpcResponse::from(details).encode().unwrap()))
+                }
+                request => panic!("unexpected setup request: {request:?}"),
+            }
+        }
+    };
+    let (client, server) = crate::connect::test_support::rpc_client(rpc).await;
+    (client, machine, initialized, server)
+}
+
+fn initialize_request(machine: &Machine) -> InitializeRequest {
+    InitializeRequest {
+        name: machine.name.clone(),
+        cluster_network: "10.210.0.0/16".parse().unwrap(),
+        public_ip: None,
+        advertised_endpoints: Vec::new(),
+        wireguard_mtu: None,
+        cloud_pairing: None,
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn initialize_recovers_beyond_the_read_budget_without_replaying_the_mutation() {
+    let started = Instant::now();
+    let (mut client, machine, calls, server) = starting_machine().await;
+    let result = super::initialize(&mut client, initialize_request(&machine))
+        .await
+        .unwrap();
+    assert_eq!(result.machine, machine);
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    assert_eq!(Instant::now() - started, Duration::from_secs(70));
+    server.abort();
+}
+
+#[tokio::test(start_paused = true)]
+async fn held_setup_rpcs_keep_the_mutation_and_read_deadlines() {
+    let (mut client, machine, calls, server) = starting_machine().await;
+    let started = Instant::now();
+    let error = client
+        .call_unretried::<op::Initialize>(initialize_request(&machine), None)
+        .await
+        .unwrap_err();
+    assert_eq!(Instant::now() - started, Duration::from_secs(5));
+    assert!(
+        error
+            .to_string()
+            .contains("Machine setup mutation reply timed out")
+    );
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    let started = Instant::now();
+    let error = client
+        .call_repeatable::<op::Inspect>(InspectRequest::default(), None)
+        .await
+        .unwrap_err();
+    assert_eq!(Instant::now() - started, Duration::from_secs(60));
+    assert!(error.to_string().contains("Machine setup read timed out"));
+    server.abort();
+}
diff --git a/crates/ployz/tests/build.rs b/crates/ployz/tests/build.rs
index bf3f091df3..c84cc52027 100644
--- a/crates/ployz/tests/build.rs
+++ b/crates/ployz/tests/build.rs
@@ -147,6 +147,9 @@ services:
     reason = "Fixed test fixtures use indexing; missing entries must fail the test."
 )]
 fn captured_build_preserves_sources_configuration_and_builder_flags() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(&root).unwrap();
@@ -342,6 +345,9 @@ fn check_with_direct_push_stops_after_validation() {
 
 #[test]
 fn built_images_bind_to_exact_content_after_tag_reuse() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-binding-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
@@ -393,6 +399,9 @@ fn built_images_bind_to_exact_content_after_tag_reuse() {
 
 #[test]
 fn an_image_the_store_does_not_hold_is_refused_as_a_result() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-content-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
@@ -443,6 +452,9 @@ fn several_requested_build_platforms_are_refused_with_the_service_named() {
 
 #[test]
 fn content_holding_several_platforms_is_refused_however_it_was_requested() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-index-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
@@ -521,6 +533,33 @@ fn settings_upstream_would_drop_are_named_before_execution() {
     }
 }
 
+/// Fake Docker has no shared builder; give its production lock a private HOME too.
+/// Re-exec avoids changing process environment while other tests are running.
+fn isolated_build_test() -> bool {
+    const CHILD: &str = "PLOYZ_ISOLATED_BUILD_TEST";
+    let thread = std::thread::current();
+    let name = thread.name().expect("libtest names its test threads");
+    if std::env::var(CHILD).as_deref() == Ok(name) {
+        return true;
+    }
+    let home = std::env::temp_dir().join(format!("ployz-build-home-{}", uuid::Uuid::new_v4()));
+    fs::create_dir(&home).unwrap();
+    let output = Command::new(std::env::current_exe().unwrap())
+        .args(["--exact", name, "--nocapture"])
+        .env(CHILD, name)
+        .env("HOME", &home)
+        .output()
+        .unwrap();
+    fs::remove_dir_all(home).unwrap();
+    assert!(
+        output.status.success() && String::from_utf8_lossy(&output.stdout).contains("1 passed"),
+        "{name}:\n{}\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr),
+    );
+    false
+}
+
 fn one_built(mut built: Vec<BuiltService>) -> BuiltService {
     assert_eq!(built.len(), 1, "expected one built Service, got {built:?}");
     built.remove(0)
@@ -611,6 +650,9 @@ exit 1
     reason = "Fixed test fixtures use indexing; missing entries must fail the test."
 )]
 fn build_and_runtime_share_one_captured_secret_resolution() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-secret-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
diff --git a/crates/ployz/tests/build/capture.rs b/crates/ployz/tests/build/capture.rs
index 326b2b14bb..899f088148 100644
--- a/crates/ployz/tests/build/capture.rs
+++ b/crates/ployz/tests/build/capture.rs
@@ -3,6 +3,9 @@ use super::*;
 
 #[test]
 fn capture_rejects_links_outside_the_captured_source() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-links-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
@@ -110,6 +113,9 @@ fn remote_contexts_require_valid_immutable_references() {
 #[test]
 #[expect(clippy::indexing_slicing, reason = "Fixed test fixture")]
 fn resolved_file_credentials_stay_private_across_captures() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-private-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
@@ -157,6 +163,9 @@ fn resolved_file_credentials_stay_private_across_captures() {
 
 #[test]
 fn builds_use_captured_explicit_registry_credentials_and_proxies() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-auth-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("auth")).unwrap();
@@ -208,6 +217,9 @@ fn builds_use_captured_explicit_registry_credentials_and_proxies() {
 #[test]
 #[expect(clippy::indexing_slicing, reason = "Fixed capture fixture")]
 fn capture_keeps_recipe_exclusions_negations_and_inline_contexts_separate() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-rules-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src/cache")).unwrap();
@@ -314,6 +326,9 @@ secrets:
 #[test]
 #[expect(clippy::indexing_slicing, reason = "Fixed capture fixture")]
 fn authored_configuration_follows_dockerignore() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-envfiles-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(&root).unwrap();
@@ -423,6 +438,9 @@ secrets:
 #[test]
 #[expect(clippy::indexing_slicing, reason = "Fixed capture fixture")]
 fn default_platform_is_captured_and_verified_unless_compose_overrides_it() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-platform-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
@@ -470,6 +488,9 @@ fn default_platform_is_captured_and_verified_unless_compose_overrides_it() {
 
 #[test]
 fn ssh_docker_hosts_are_refused_but_git_contexts_keep_the_captured_agent_socket() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-build-agent-{}", std::process::id()));
     let _ = fs::remove_dir_all(&root);
     fs::create_dir_all(root.join("src")).unwrap();
diff --git a/crates/ployz/tests/build/railpack.rs b/crates/ployz/tests/build/railpack.rs
index 79b632359b..96680e0a5f 100644
--- a/crates/ployz/tests/build/railpack.rs
+++ b/crates/ployz/tests/build/railpack.rs
@@ -57,6 +57,9 @@ fn railpack_selection_keeps_explicit_dockerfiles_and_refuses_check() {
 #[test]
 #[expect(clippy::indexing_slicing, reason = "Fixed capture fixture")]
 fn railpack_capture_filters_before_transfer_and_keeps_effective_variables_private() {
+    if !isolated_build_test() {
+        return;
+    }
     let root = std::env::temp_dir().join(format!("ployz-railpack-capture-{}", std::process::id()));
     fs::create_dir_all(root.join("src/config")).unwrap();
     fs::write(
diff --git a/crates/ployz/tests/cloud_init/catch_up.rs b/crates/ployz/tests/cloud_init/catch_up.rs
index 4c75c71abb..bbc3473226 100644
--- a/crates/ployz/tests/cloud_init/catch_up.rs
+++ b/crates/ployz/tests/cloud_init/catch_up.rs
@@ -1,4 +1,6 @@
 //! Shared post-join Global catch-up behavior for Cloud enrollment and Machine add.
+//! Retry exhaustion and exact budgets run against the real Client under virtual
+//! time in global_catch_up_tests; these CLI cases cover recovery and reporting.
 
 use std::{collections::BTreeMap, fs, process::Output};
 
@@ -13,68 +15,89 @@ use ployz_core::{
 };
 use serde_json::json;
 
+#[derive(Clone, Copy)]
+enum Fault {
+    Healthy,
+    Transient,
+    Permanent,
+}
+
+fn with_faults(daemon: JoinDaemon, target: Fault, ensure: Fault) -> JoinDaemon {
+    let daemon = match target {
+        Fault::Healthy => daemon,
+        Fault::Transient => daemon.transient_target_inspect_failures(1),
+        Fault::Permanent => daemon.fail_target_inspect(),
+    };
+    match ensure {
+        Fault::Healthy => daemon,
+        Fault::Transient => daemon.transient_ensure_failures(1),
+        Fault::Permanent => daemon.fail_ensure(),
+    }
+}
+
 #[tokio::test]
-async fn cloud_join_retries_target_readiness_and_reports_exhaustion() {
-    let (recovered, daemon) = cloud_join(1, 0).await;
+async fn cloud_join_retries_target_readiness_and_reports_failure() {
+    let (recovered, daemon) = cloud_join(Fault::Transient, Fault::Healthy).await;
     assert!(recovered.status.success());
     assert_eq!(daemon.target_inspect_attempts(), 2);
 
-    let (exhausted, daemon) = cloud_join(4, 0).await;
-    assert!(!exhausted.status.success());
-    assert_eq!(daemon.target_inspect_attempts(), 4);
+    let (failed, daemon) = cloud_join(Fault::Permanent, Fault::Healthy).await;
+    assert!(!failed.status.success());
+    assert_eq!(daemon.target_inspect_attempts(), 1);
     assert_eq!(daemon.ensure_attempts(), 0);
-    assert_joined_with_incomplete_catch_up(&exhausted);
+    assert_joined_with_incomplete_catch_up(&failed);
     daemon.join_request();
 }
 
 #[tokio::test]
-async fn machine_add_retries_target_readiness_and_reports_exhaustion() {
-    let (recovered, entry, target) = machine_add(1, 0).await;
+async fn machine_add_retries_target_readiness_and_reports_failure() {
+    let (recovered, entry, target) = machine_add(Fault::Transient, Fault::Healthy).await;
     assert!(recovered.status.success());
     assert_eq!(entry.target_inspect_attempts(), 2);
     target.join_request();
 
-    let (exhausted, entry, target) = machine_add(4, 0).await;
-    assert!(!exhausted.status.success());
-    assert_eq!(entry.target_inspect_attempts(), 4);
+    let (failed, entry, target) = machine_add(Fault::Permanent, Fault::Healthy).await;
+    assert!(!failed.status.success());
+    assert_eq!(entry.target_inspect_attempts(), 1);
     assert_eq!(entry.ensure_attempts(), 0);
-    assert!(String::from_utf8_lossy(&exhausted.stdout).contains("Added Machine joiner"));
-    assert_joined_with_incomplete_catch_up(&exhausted);
+    assert!(String::from_utf8_lossy(&failed.stdout).contains("Added Machine joiner"));
+    assert_joined_with_incomplete_catch_up(&failed);
     target.join_request();
 }
 
 #[tokio::test]
-async fn cloud_join_retries_catch_up_and_reports_exhaustion() {
-    let (recovered, daemon) = cloud_join(0, 1).await;
+async fn cloud_join_retries_catch_up_and_reports_failure() {
+    let (recovered, daemon) = cloud_join(Fault::Healthy, Fault::Transient).await;
     assert!(recovered.status.success());
     assert_eq!(daemon.ensure_attempts(), 3);
 
-    let (exhausted, daemon) = cloud_join(0, 8).await;
-    assert!(!exhausted.status.success());
-    assert_eq!(daemon.ensure_attempts(), 8);
-    assert_joined_with_incomplete_catch_up(&exhausted);
+    let (failed, daemon) = cloud_join(Fault::Healthy, Fault::Permanent).await;
+    assert!(!failed.status.success());
+    assert_eq!(daemon.ensure_attempts(), 2);
+    assert_joined_with_incomplete_catch_up(&failed);
     daemon.join_request();
 }
 
 #[tokio::test]
-async fn machine_add_retries_catch_up_and_reports_exhaustion() {
-    let (recovered, entry, target) = machine_add(0, 1).await;
+async fn machine_add_retries_catch_up_and_reports_failure() {
+    let (recovered, entry, target) = machine_add(Fault::Healthy, Fault::Transient).await;
     assert!(recovered.status.success());
     assert_eq!(entry.ensure_attempts(), 3);
     target.join_request();
 
-    let (exhausted, entry, target) = machine_add(0, 8).await;
-    assert!(!exhausted.status.success());
-    assert_eq!(entry.ensure_attempts(), 8);
-    assert!(String::from_utf8_lossy(&exhausted.stdout).contains("Added Machine joiner"));
-    assert_joined_with_incomplete_catch_up(&exhausted);
+    let (failed, entry, target) = machine_add(Fault::Healthy, Fault::Permanent).await;
+    assert!(!failed.status.success());
+    assert_eq!(entry.ensure_attempts(), 2);
+    assert!(String::from_utf8_lossy(&failed.stdout).contains("Added Machine joiner"));
+    assert_joined_with_incomplete_catch_up(&failed);
     target.join_request();
 }
 
 #[tokio::test]
 async fn machine_add_reports_omitted_targets_before_catch_up() {
     for membership in [MembershipObservation::Down, MembershipObservation::Unknown] {
-        let (output, entry, target) = machine_add_with_membership(0, 0, membership).await;
+        let (output, entry, target) =
+            machine_add_with_membership(Fault::Healthy, Fault::Healthy, membership).await;
         assert!(!output.status.success());
         let stderr = String::from_utf8_lossy(&output.stderr);
         assert!(stderr.contains("partial Service observations"), "{stderr}");
@@ -86,7 +109,7 @@ async fn machine_add_reports_omitted_targets_before_catch_up() {
     }
 }
 
-async fn cloud_join(target_failures: usize, ensure_failures: usize) -> (Output, JoinDaemon) {
+async fn cloud_join(target_failures: Fault, ensure_failures: Fault) -> (Output, JoinDaemon) {
     let founder = founder_machine();
     let mut registration = registration();
     registration.visible_peers = vec![founder.clone()];
@@ -100,10 +123,11 @@ async fn cloud_join(target_failures: usize, ensure_failures: usize) -> (Output,
         "registration": registration,
     }))
     .await;
-    let daemon = JoinDaemon::new(registration)
-        .with_containers(globals_on(&founder))
-        .transient_target_inspect_failures(target_failures)
-        .transient_ensure_failures(ensure_failures);
+    let daemon = with_faults(
+        JoinDaemon::new(registration).with_containers(globals_on(&founder)),
+        target_failures,
+        ensure_failures,
+    );
     let address = serve_machine(daemon.clone()).await;
     let output = tokio::process::Command::new(env!("CARGO_BIN_EXE_ployz"))
         .args([
@@ -125,25 +149,27 @@ async fn cloud_join(target_failures: usize, ensure_failures: usize) -> (Output,
 }
 
 async fn machine_add(
-    target_failures: usize,
-    ensure_failures: usize,
+    target_failures: Fault,
+    ensure_failures: Fault,
 ) -> (Output, JoinDaemon, JoinDaemon) {
     machine_add_with_membership(target_failures, ensure_failures, MembershipObservation::Up).await
 }
 
 async fn machine_add_with_membership(
-    target_failures: usize,
-    ensure_failures: usize,
+    target_failures: Fault,
+    ensure_failures: Fault,
     membership: MembershipObservation,
 ) -> (Output, JoinDaemon, JoinDaemon) {
     let founder = founder_machine();
     let mut registration = registration();
     registration.visible_peers = vec![founder.clone()];
-    let entry = JoinDaemon::new(registration.clone())
-        .with_containers(globals_on(&founder))
-        .with_membership(membership)
-        .transient_target_inspect_failures(target_failures)
-        .transient_ensure_failures(ensure_failures);
+    let entry = with_faults(
+        JoinDaemon::new(registration.clone())
+            .with_containers(globals_on(&founder))
+            .with_membership(membership),
+        target_failures,
+        ensure_failures,
+    );
     let target = JoinDaemon::new(registration);
     let entry_address = serve_machine(entry.clone()).await;
     let target_address = serve_machine(target.clone()).await;
diff --git a/crates/ployz/tests/cloud_init/founder_resumption.rs b/crates/ployz/tests/cloud_init/founder_resumption.rs
index 6b9f39f7e8..5faa0c25b2 100644
--- a/crates/ployz/tests/cloud_init/founder_resumption.rs
+++ b/crates/ployz/tests/cloud_init/founder_resumption.rs
@@ -324,8 +324,7 @@ async fn founder_tail_recovers_lost_replies_without_replaying_mutations() {
     .with_events(events.clone())
     .transient_founder_tail_failures(1);
     let machine_addr = serve_machine(daemon.clone()).await;
-    let (probe, probe_port) =
-        serve_ingress_probe(machine_id, std::time::Duration::from_secs(6)).await;
+    let (probe, probe_port) = serve_ingress_probe(machine_id).await;
 
     let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
     let proxy = format!("http://{}", closed.local_addr().unwrap());
@@ -347,6 +346,8 @@ async fn founder_tail_recovers_lost_replies_without_replaying_mutations() {
                 "caddy:2.10.0",
                 "--yes",
             ])
+            // Health-monitor timing is covered under virtual time in deploy tests.
+            .env("PLOYZ_HEALTH_MONITOR_PERIOD", "0s")
             .env("PLOYZ_INGRESS_VERIFY_PORT", probe_port.to_string())
             .env("HTTPS_PROXY", &proxy)
             .env("https_proxy", &proxy)
@@ -359,7 +360,9 @@ async fn founder_tail_recovers_lost_replies_without_replaying_mutations() {
     assert!(String::from_utf8_lossy(&first.stderr).contains(
         "rerun the same ployz cloud enroll command without --reset (keep all other options)"
     ));
-    assert!(String::from_utf8_lossy(&first.stderr).contains("Machine setup read timed out"));
+    assert!(
+        String::from_utf8_lossy(&first.stderr).contains("lost Ingress container creation reply")
+    );
     assert_eq!(daemon.founder_tail_attempts(), [1, 1, 0, 0]);
     assert_eq!(
         daemon.initialize_requests().len(),
diff --git a/crates/ployz/tests/cloud_init/harness.rs b/crates/ployz/tests/cloud_init/harness.rs
index bc2ebddb69..0cc7ac667a 100644
--- a/crates/ployz/tests/cloud_init/harness.rs
+++ b/crates/ployz/tests/cloud_init/harness.rs
@@ -61,8 +61,7 @@ struct JoinInner {
     lose_lifecycle_reply: AtomicBool,
     initialize_requests: Mutex<Vec<InitializeRequest>>,
     lose_initialize_reply: AtomicBool,
-    hold_inspect_once: AtomicBool,
-    startup_ready_at: Mutex<Option<std::time::Instant>>,
+    startup_inspect_failures: AtomicUsize,
     replace_identity_on_initialize: AtomicBool,
     fail_reservation: AtomicBool,
     reserve_request: Mutex<Option<ReserveDomainRequest>>,
@@ -86,6 +85,7 @@ struct JoinInner {
     target_inspect_attempts: AtomicUsize,
     transient_target_inspect_failures: AtomicUsize,
     fail_ensure: AtomicBool,
+    fail_target_inspect: AtomicBool,
     fail_list_on: Mutex<Option<MachineId>>,
     assigned_membership: Mutex<MembershipObservation>,
     _register: Mutex<Option<JoinHandle<()>>>,
@@ -104,8 +104,7 @@ impl JoinDaemon {
                 lose_lifecycle_reply: AtomicBool::new(false),
                 initialize_requests: Mutex::new(Vec::new()),
                 lose_initialize_reply: AtomicBool::new(false),
-                hold_inspect_once: AtomicBool::new(false),
-                startup_ready_at: Mutex::new(None),
+                startup_inspect_failures: AtomicUsize::new(0),
                 replace_identity_on_initialize: AtomicBool::new(false),
                 fail_reservation: AtomicBool::new(false),
                 reserve_request: Mutex::new(None),
@@ -129,6 +128,7 @@ impl JoinDaemon {
                 target_inspect_attempts: AtomicUsize::new(0),
                 transient_target_inspect_failures: AtomicUsize::new(0),
                 fail_ensure: AtomicBool::new(false),
+                fail_target_inspect: AtomicBool::new(false),
                 fail_list_on: Mutex::new(None),
                 assigned_membership: Mutex::new(MembershipObservation::Up),
                 _register: Mutex::new(None),
@@ -250,6 +250,11 @@ impl JoinDaemon {
         self.inner.events.lock().unwrap().record(event);
     }
 
+    pub fn fail_target_inspect(self) -> Self {
+        self.inner.fail_target_inspect.store(true, Ordering::SeqCst);
+        self
+    }
+
     pub fn fail_ensure(self) -> Self {
         self.inner.fail_ensure.store(true, Ordering::SeqCst);
         self
@@ -346,16 +351,7 @@ impl MachineRpc for JoinDaemon {
         &self,
         request: Request<OpaquePayload>,
     ) -> Result<Response<OpaquePayload>, Status> {
-        if self.inner.hold_inspect_once.swap(false, Ordering::SeqCst) {
-            tokio::time::sleep(std::time::Duration::from_secs(6)).await;
-        }
-        if self
-            .inner
-            .startup_ready_at
-            .lock()
-            .unwrap()
-            .is_some_and(|ready| std::time::Instant::now() < ready)
-        {
+        if consume_transient_failure(&self.inner.startup_inspect_failures) {
             return Err(Status::unavailable(
                 "first startup is still pulling Corrosion",
             ));
@@ -367,6 +363,11 @@ impl MachineRpc for JoinDaemon {
             self.inner
                 .target_inspect_attempts
                 .fetch_add(1, Ordering::SeqCst);
+            if self.inner.fail_target_inspect.load(Ordering::SeqCst) {
+                return Err(Status::failed_precondition(
+                    "target Machine cannot provide capacity",
+                ));
+            }
             if consume_transient_failure(&self.inner.transient_target_inspect_failures) {
                 return Err(Status::unavailable("target Machine is not ready"));
             }
@@ -458,7 +459,7 @@ impl MachineRpc for JoinDaemon {
             .lose_lifecycle_reply
             .swap(false, Ordering::SeqCst)
         {
-            return std::future::pending().await; // Applied, but the response stays open.
+            return Err(Status::unavailable("lost lifecycle reply"));
         }
         rpc_ok(JoinAccepted {})
     }
@@ -552,10 +553,10 @@ impl MachineRpc for JoinDaemon {
             .lose_initialize_reply
             .swap(false, Ordering::SeqCst)
         {
-            self.inner.hold_inspect_once.store(true, Ordering::SeqCst);
-            *self.inner.startup_ready_at.lock().unwrap() =
-                Some(std::time::Instant::now() + std::time::Duration::from_secs(70));
-            return std::future::pending().await; // Applied, but the response stays open.
+            self.inner
+                .startup_inspect_failures
+                .store(1, Ordering::SeqCst);
+            return Err(Status::unavailable("lost Initialize reply"));
         }
         rpc_ok(Initialized { machine })
     }
@@ -949,7 +950,7 @@ impl MachineRpc for JoinDaemon {
             .lose_lifecycle_reply
             .swap(false, Ordering::SeqCst)
         {
-            return std::future::pending().await; // Applied, but the response stays open.
+            return Err(Status::unavailable("lost lifecycle reply"));
         }
         rpc_ok(ResetAccepted {})
     }
diff --git a/crates/ployz/tests/cloud_init/main.rs b/crates/ployz/tests/cloud_init/main.rs
index 86f46f4951..7db8e58053 100644
--- a/crates/ployz/tests/cloud_init/main.rs
+++ b/crates/ployz/tests/cloud_init/main.rs
@@ -250,9 +250,34 @@ async fn caddy_lookup_failure_happens_before_initialize() {
         target_versions: Default::default(),
     });
     let machine_addr = serve_machine(daemon.clone()).await;
-    let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let proxy = format!("http://{}", closed.local_addr().unwrap());
-    drop(closed);
+    // Reject the proxy tunnel immediately. Retry exhaustion is covered with a
+    // paused clock in setup_retry; this test checks failure before Initialize.
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let proxy = format!("http://{}", listener.local_addr().unwrap());
+    let rejected = tokio::spawn(async move {
+        use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+        loop {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut stream = BufReader::new(stream);
+            let mut request = String::new();
+            stream.read_line(&mut request).await.unwrap();
+            assert!(request.starts_with("CONNECT "), "{request}");
+            loop {
+                request.clear();
+                assert!(stream.read_line(&mut request).await.unwrap() > 0);
+                if request == "\r\n" {
+                    break;
+                }
+            }
+            stream
+                .get_mut()
+                .write_all(
+                    b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
+                )
+                .await
+                .unwrap();
+        }
+    });
 
     let output = tokio::process::Command::new(env!("CARGO_BIN_EXE_ployz"))
         .args([
@@ -276,9 +301,11 @@ async fn caddy_lookup_failure_happens_before_initialize() {
         .await
         .unwrap();
 
+    rejected.abort();
+    assert!(rejected.await.unwrap_err().is_cancelled());
     assert!(!output.status.success());
     assert!(
-        String::from_utf8_lossy(&output.stderr).contains("Discovering Caddy image at Docker Hub"),
+        String::from_utf8_lossy(&output.stderr).contains("list Docker Hub Caddy tags"),
         "{}",
         String::from_utf8_lossy(&output.stderr)
     );
diff --git a/crates/ployz/tests/cloud_init/servers.rs b/crates/ployz/tests/cloud_init/servers.rs
index aa934a85fd..b51bf7c161 100644
--- a/crates/ployz/tests/cloud_init/servers.rs
+++ b/crates/ployz/tests/cloud_init/servers.rs
@@ -32,10 +32,7 @@ pub async fn serve_machine(daemon: JoinDaemon) -> SocketAddr {
     address
 }
 
-pub async fn serve_ingress_probe(
-    machine_id: MachineId,
-    mut delay_first: std::time::Duration,
-) -> (JoinHandle<()>, u16) {
+pub async fn serve_ingress_probe(machine_id: MachineId) -> (JoinHandle<()>, u16) {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
     let port = listener.local_addr().unwrap().port();
     let server = tokio::spawn(async move {
@@ -43,8 +40,8 @@ pub async fn serve_ingress_probe(
             let (mut stream, _) = listener.accept().await.unwrap();
             let mut request = [0; 1024];
             let _ = stream.read(&mut request).await.unwrap();
-            tokio::time::sleep(delay_first).await;
-            delay_first = std::time::Duration::ZERO;
+            // Probe retry coverage lives in dns::probe::tests; this fixture
+            // verifies that founder enrollment reaches the real HTTP endpoint.
             let body = machine_id.as_str();
             let response = format!(
                 "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
diff --git a/crates/ployz/tests/connect/main.rs b/crates/ployz/tests/connect/main.rs
index 712e74cdd9..0beb5a4448 100644
--- a/crates/ployz/tests/connect/main.rs
+++ b/crates/ployz/tests/connect/main.rs
@@ -879,7 +879,7 @@ async fn volume_listing_omits_down_and_unknown_and_probes_suspect() {
     server.abort();
 }
 
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn fanout_reads_retry_failed_legs_without_rerunning_successes() {
     let mut service = DiscoveryService::new(test_description());
     service.machines = vec![machine('a', "recovers"), machine('b', "fails")];
@@ -1043,7 +1043,7 @@ async fn machine_discovery_uses_the_same_rpc_over_tcp_and_unix() {
     std::fs::remove_dir_all(root).unwrap();
 }
 
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn unary_call_retries_unavailable_after_redial() {
     let description = test_description();
     let service = DiscoveryService::new(description.clone());
@@ -1068,7 +1068,7 @@ async fn unary_call_retries_unavailable_after_redial() {
     server.abort();
 }
 
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn unary_call_does_not_retry_remote_or_not_found() {
     let not_found = DiscoveryService::new(test_description());
     let (mut client, server, connects) = connected_client(not_found.clone()).await;
@@ -1117,7 +1117,7 @@ async fn unary_call_does_not_retry_remote_or_not_found() {
     server.abort();
 }
 
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn unary_call_gives_up_after_four_unavailable_attempts() {
     let service = DiscoveryService::new(test_description());
     let (mut client, server, connects) = connected_client(service.clone()).await;
diff --git a/crates/ployz/tests/connect/relay.rs b/crates/ployz/tests/connect/relay.rs
index f0c8f1be84..8f64bfe7a0 100644
--- a/crates/ployz/tests/connect/relay.rs
+++ b/crates/ployz/tests/connect/relay.rs
@@ -117,7 +117,8 @@ async fn sdk_script_temporary_files_are_removed_after_exit_and_timeout() {
             const path = require('node:path');
             const dir = fs.mkdtempSync(path.join(require('node:os').tmpdir(), 'ployz-sdk-check-'));
             fs.writeFileSync(path.join(dir, 'ployz-sdk.node'), 'test addon');
-            fs.writeFileSync(process.argv[1], dir);
+            fs.writeFileSync(process.argv[1] + '.tmp', dir);
+            fs.renameSync(process.argv[1] + '.tmp', process.argv[1]);
             fs.writeSync(1, 'o'.repeat(128 * 1024));
             fs.writeSync(2, 'e'.repeat(128 * 1024));
             if (process.argv[2] === 'timeout') setInterval(() => {}, 1000);
@@ -126,8 +127,27 @@ async fn sdk_script_temporary_files_are_removed_after_exit_and_timeout() {
             ])
             .arg(&report)
             .arg(mode);
-        let result = sdk_script_output(&mut command, Duration::from_secs(2)).await;
-        let dir = std::fs::read_to_string(&report).expect("Node created the SDK fixture");
+        let deadline = Duration::from_secs(60);
+        let output = sdk_script_output(&mut command, deadline);
+        tokio::pin!(output);
+        let result = tokio::select! {
+            result = &mut output => result,
+            () = async {
+                while !report.try_exists().unwrap() {
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+            }, if mode == "timeout" => {
+                // Node has created the fixture. Expire the script deadline only now,
+                // then restore real time for process termination and pipe draining.
+                tokio::time::pause();
+                tokio::time::advance(deadline).await;
+                tokio::time::resume();
+                output.await
+            }
+        };
+        let dir = std::fs::read_to_string(&report).unwrap_or_else(|error| {
+            panic!("{mode}: Node fixture report missing: {error}; subprocess result: {result:?}")
+        });
         std::fs::remove_file(report).unwrap();
         let leaked = std::path::Path::new(&dir).exists();
         if leaked {
diff --git a/crates/ployz/tests/connect/sdk_volumes.rs b/crates/ployz/tests/connect/sdk_volumes.rs
index 9fae8d1ef4..16a207ced1 100644
--- a/crates/ployz/tests/connect/sdk_volumes.rs
+++ b/crates/ployz/tests/connect/sdk_volumes.rs
@@ -209,9 +209,12 @@ async fn remove_volumes_omits_machines_that_do_not_invite_rpc() {
     );
 }
 
-#[tokio::test]
+#[tokio::test(start_paused = true)]
 async fn timed_out_removal_retains_identity_and_unknown_completion() {
-    let (client, _session, _machine) = volume_session().await;
+    // The Client owns this deadline; other tests cover the Session/Relay adapter.
+    let mut service = DiscoveryService::new(advertised_description());
+    service.machines = vec![machine('a', "one"), machine('b', "two")];
+    let (mut client, server, _) = super::support::connected_client(service).await;
     let outcomes = timeout(
         Duration::from_secs(15),
         client.remove_volumes(remove([volume('a', "slow"), volume('a', "data")], false)),
@@ -235,6 +238,7 @@ async fn timed_out_removal_retains_identity_and_unknown_completion() {
             outcome: VolumeRemovalOutcome::Removed
         }
     );
+    server.abort();
 }
 
 #[tokio::test]
diff --git a/crates/ployz/tests/connect/sdk_watch.rs b/crates/ployz/tests/connect/sdk_watch.rs
index 3f9c178b79..eaa9caa625 100644
--- a/crates/ployz/tests/connect/sdk_watch.rs
+++ b/crates/ployz/tests/connect/sdk_watch.rs
@@ -170,18 +170,20 @@ async fn cancel_interrupts_storage_enrichment() {
         .await;
     let client = connect(&session.url, description.machine_id.as_str()).await;
     let watch = client.watch().await.unwrap();
+    let (received, held) = tokio::sync::oneshot::channel();
     service
         .describe_outcomes
         .lock()
         .unwrap()
-        .push_back(DescribeOutcome::Hang);
+        .push_back(DescribeOutcome::Hang(received));
     let waiting = watch.next();
     tokio::pin!(waiting);
-    assert!(
-        timeout(Duration::from_millis(50), &mut waiting)
-            .await
-            .is_err()
-    );
+    tokio::select! {
+        result = &mut waiting => panic!("held enrichment completed: {result:?}"),
+        result = timeout(Duration::from_secs(2), held) => {
+            result.expect("enrichment request reached the server").unwrap();
+        }
+    }
 
     watch.cancel();
 
@@ -205,17 +207,25 @@ async fn storage_enrichment_has_a_short_overall_budget() {
         .await;
     let client = connect(&session.url, description.machine_id.as_str()).await;
     let watch = client.watch().await.unwrap();
+    let (received, held) = tokio::sync::oneshot::channel();
     service
         .describe_outcomes
         .lock()
         .unwrap()
-        .push_back(DescribeOutcome::Hang);
-
-    let frame = timeout(Duration::from_secs(4), watch.next())
-        .await
-        .expect("storage enrichment must have a short overall budget")
-        .unwrap()
-        .unwrap();
+        .push_back(DescribeOutcome::Hang(received));
+
+    let (frame, ()) = timeout(Duration::from_secs(4), async {
+        tokio::join!(watch.next(), async {
+            held.await.expect("enrichment request reached the server");
+            // Advance only once the held request has armed the overall budget.
+            tokio::time::pause();
+            tokio::time::advance(Duration::from_secs(3)).await;
+            tokio::time::resume();
+        })
+    })
+    .await
+    .expect("storage enrichment must have a short overall budget");
+    let frame = frame.unwrap().unwrap();
 
     assert_eq!(
         frame.machines.first().and_then(|machine| machine.storage),
@@ -424,9 +434,12 @@ async fn node_watch_decodes_frames_above_tonics_default() {
 #[tokio::test]
 async fn frame_above_ceiling_errors_without_closing_session() {
     let (client, service, _session, _machine) = watching_session().await;
-    let mut frame = frozen_frame();
-    frame.observed_at = "x".repeat(RUNTIME_WATCH_MESSAGE_SIZE_LIMIT);
-    service.push_watch_payload(OpaquePayload::from_json(&frame).unwrap());
+    let mut payload = OpaquePayload::from_json(&frozen_frame()).unwrap();
+    // Trailing whitespace keeps valid frame JSON without serializing a 64 MiB string.
+    payload
+        .json
+        .resize(RUNTIME_WATCH_MESSAGE_SIZE_LIMIT + 1, b' ');
+    service.push_watch_payload(payload);
     let watch = client.watch().await.unwrap();
 
     let error = timeout(Duration::from_secs(10), watch.next())
diff --git a/crates/ployz/tests/connect/support.rs b/crates/ployz/tests/connect/support.rs
index 59be2b65bf..511f367bf1 100644
--- a/crates/ployz/tests/connect/support.rs
+++ b/crates/ployz/tests/connect/support.rs
@@ -62,7 +62,7 @@ pub(super) async fn serve_discovery(
 pub(super) enum DescribeOutcome {
     Status(Status),
     Remote(RpcError),
-    Hang,
+    Hang(tokio::sync::oneshot::Sender<()>),
 }
 
 #[derive(Clone)]
@@ -372,7 +372,10 @@ impl MachineRpc for DiscoveryService {
             Some(DescribeOutcome::Remote(error)) => {
                 return Ok(Response::new(RpcResponse::from(error).encode().unwrap()));
             }
-            Some(DescribeOutcome::Hang) => return std::future::pending().await,
+            Some(DescribeOutcome::Hang(received)) => {
+                received.send(()).unwrap();
+                return std::future::pending().await;
+            }
             None => {}
         }
         let request = request
@@ -1072,14 +1075,54 @@ pub(super) async fn connected_client(
     tokio::task::JoinHandle<Result<(), tonic::transport::Error>>,
     Arc<AtomicUsize>,
 ) {
-    let (address, server) = serve_discovery(service).await;
+    // Keep semantic RPC tests inside Tokio so paused time never outruns OS I/O.
+    // Transport tests use serve_discovery / SystemConnector directly.
+    struct MemoryConnector {
+        incoming: mpsc::UnboundedSender<Result<tokio::io::DuplexStream, std::io::Error>>,
+        connects: Arc<AtomicUsize>,
+    }
+    #[tonic::async_trait]
+    impl Connector for MemoryConnector {
+        async fn connect(&self, _: &Connection) -> Result<Channel, ConnectError> {
+            self.connects.fetch_add(1, Ordering::SeqCst);
+            let incoming = self.incoming.clone();
+            Channel::from_static("http://memory.invalid")
+                .connect_with_connector(tower::service_fn(move |_| {
+                    let (client, server) = tokio::io::duplex(64 * 1024);
+                    incoming.send(Ok(server)).unwrap();
+                    async move { Ok::<_, std::io::Error>(hyper_util::rt::TokioIo::new(client)) }
+                }))
+                .await
+                .map_err(Into::into)
+        }
+
+        async fn dial_proxy(
+            &self,
+            _: &Connection,
+            _: &str,
+            _: &str,
+        ) -> Result<BoxProxyStream, ConnectError> {
+            Err(ConnectError::Attempt("unused".into()))
+        }
+    }
+    let (incoming, receiver) = mpsc::unbounded_channel();
+    let server = tokio::spawn(
+        Server::builder()
+            .add_service(MachineRpcServer::new(service))
+            .serve_with_incoming(tokio_stream::wrappers::UnboundedReceiverStream::new(
+                receiver,
+            )),
+    );
     let connects = Arc::new(AtomicUsize::new(0));
     let client = connect_selected_with(
         SelectedConnections {
             source: ConnectionSource::Direct,
-            connections: vec![Connection::tcp(address)],
+            connections: vec![Connection::tcp("127.0.0.1:1".parse().unwrap())],
         },
-        Arc::new(CountingConnector::new(connects.clone())),
+        Arc::new(MemoryConnector {
+            incoming,
+            connects: connects.clone(),
+        }),
     )
     .await
     .unwrap();