diff --git a/crates/adaptive/src/response_cache/store.rs b/crates/adaptive/src/response_cache/store.rs
index 001074005..281a581fc 100644
--- a/crates/adaptive/src/response_cache/store.rs
+++ b/crates/adaptive/src/response_cache/store.rs
@@ -445,6 +445,18 @@ pub async fn check_backend_health(
     Ok(store.backend_kind().to_string())
 }
 
+/// Validates the configured backend target without attempting network access.
+///
+/// Intended for offline `nemo-relay doctor` checks: it rejects malformed
+/// backend targets while still skipping live reachability probes.
+pub fn validate_backend_target(config: &ResponseCacheConfig) -> std::result::Result<(), String> {
+    match config.backend.kind.as_str() {
+        "in_memory" => Ok(()),
+        "redis" => validate_redis_backend_target(config).map_err(|err| err.to_string()),
+        other => Err(format!("response_cache: unknown backend kind '{other}'")),
+    }
+}
+
 /// Builds the response-cache backend from config.
 ///
 /// Returns the boxed [`CacheStore`] used by the intercept and the `doctor`
@@ -465,16 +477,7 @@ pub(crate) async fn build_store(config: &ResponseCacheConfig) -> Result<Arc<dyn
 async fn build_redis_store(config: &ResponseCacheConfig) -> Result<Arc<dyn CacheStore>> {
     use crate::response_cache::store::RedisCacheStore;
 
-    let url = config
-        .backend
-        .config
-        .get("url")
-        .and_then(Json::as_str)
-        .ok_or_else(|| {
-            AdaptiveError::InvalidConfig(
-                "response_cache: redis backend requires backend.config.url".to_string(),
-            )
-        })?;
+    let url = redis_backend_url(config)?;
     let key_prefix = config
         .backend
         .config
@@ -496,6 +499,36 @@ async fn build_redis_store(_config: &ResponseCacheConfig) -> Result<Arc<dyn Cach
     ))
 }
 
+fn redis_backend_url(config: &ResponseCacheConfig) -> Result<&str> {
+    config
+        .backend
+        .config
+        .get("url")
+        .and_then(Json::as_str)
+        .ok_or_else(|| {
+            AdaptiveError::InvalidConfig(
+                "response_cache: redis backend requires backend.config.url".to_string(),
+            )
+        })
+}
+
+#[cfg(feature = "redis-backend")]
+fn validate_redis_backend_target(config: &ResponseCacheConfig) -> Result<()> {
+    let url = redis_backend_url(config)?;
+    redis::Client::open(url)
+        .map(|_| ())
+        .map_err(|err| AdaptiveError::Storage(format!("response_cache: redis client: {err}")))
+}
+
+#[cfg(not(feature = "redis-backend"))]
+fn validate_redis_backend_target(_config: &ResponseCacheConfig) -> Result<()> {
+    Err(AdaptiveError::InvalidConfig(
+        "response_cache: backend.kind = \"redis\" requires building with the 'redis-backend' \
+         feature"
+            .to_string(),
+    ))
+}
+
 #[cfg(test)]
 #[path = "../../tests/unit/response_cache/store_tests.rs"]
 mod tests;
diff --git a/crates/cli/src/commands/diagnostics.rs b/crates/cli/src/commands/diagnostics.rs
index 35034c671..ccf6cbe08 100644
--- a/crates/cli/src/commands/diagnostics.rs
+++ b/crates/cli/src/commands/diagnostics.rs
@@ -21,6 +21,11 @@ pub(crate) struct DoctorCommand {
     pub(crate) install_dir: Option<PathBuf>,
     #[arg(long)]
     pub(crate) json: bool,
+    #[arg(
+        long,
+        help = "Validate configuration and endpoint syntax without running live network probes"
+    )]
+    pub(crate) offline: bool,
 }
 
 #[derive(Debug, Clone, Args)]
@@ -41,6 +46,7 @@ pub(super) async fn execute(
     crate::diagnostics::run_doctor(
         command.agent.map(Into::into),
         command.json,
+        crate::diagnostics::DoctorProbeMode::from_offline_flag(command.offline),
         &gateway_overrides,
         logging_fallback_error,
     )
diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs
index 79f9f19d4..9771ba937 100644
--- a/crates/cli/src/commands/mod.rs
+++ b/crates/cli/src/commands/mod.rs
@@ -233,7 +233,14 @@ async fn run_default(
         .await?;
         Ok(ExitCode::SUCCESS)
     } else if runtime_configuration::any_config_file_exists() {
-        runtime_diagnostics::run_doctor(None, false, &runtime_args, None).await
+        runtime_diagnostics::run_doctor(
+            None,
+            false,
+            runtime_diagnostics::DoctorProbeMode::Live,
+            &runtime_args,
+            None,
+        )
+        .await
     } else {
         configure::run(None, None).await?;
         Ok(ExitCode::SUCCESS)
diff --git a/crates/cli/src/commands/root.rs b/crates/cli/src/commands/root.rs
index 9cce1033a..6b8c4f48f 100644
--- a/crates/cli/src/commands/root.rs
+++ b/crates/cli/src/commands/root.rs
@@ -112,7 +112,7 @@ pub(crate) enum Command {
     Uninstall(UninstallCommand),
     /// Validate and configure model pricing catalogs.
     ModelPricing(PricingCommand),
-    /// Diagnose env, agents, config, observability (optionally scoped to one agent)
+    /// Diagnose env, agents, config, observability (use --offline to skip live network probes)
     Doctor(DoctorCommand),
     /// List supported and locally-detected agents (use `--json` for machine output)
     Agents(AgentsCommand),
diff --git a/crates/cli/src/diagnostics/mod.rs b/crates/cli/src/diagnostics/mod.rs
index 1a7dc94fc..6a325ca0d 100644
--- a/crates/cli/src/diagnostics/mod.rs
+++ b/crates/cli/src/diagnostics/mod.rs
@@ -46,6 +46,22 @@ use crate::server::{GatewayOverrides, register_and_validate_plugin_components};
 const NETWORK_TIMEOUT: Duration = Duration::from_secs(2);
 const PRICING_PLUGIN_KIND: &str = "pricing";
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum DoctorProbeMode {
+    Live,
+    Offline,
+}
+
+impl DoctorProbeMode {
+    pub(crate) fn from_offline_flag(offline: bool) -> Self {
+        if offline { Self::Offline } else { Self::Live }
+    }
+
+    fn is_offline(self) -> bool {
+        matches!(self, Self::Offline)
+    }
+}
+
 struct PluginConfigurationDiagnostics {
     sources: Vec<PathBuf>,
     error: Option<String>,
@@ -57,6 +73,7 @@ struct PluginConfigurationDiagnostics {
 /// the first missing directory.
 pub(crate) async fn collect_report(
     target_agent: Option<CodingAgent>,
+    probe_mode: DoctorProbeMode,
     gateway_overrides: &GatewayOverrides,
 ) -> Result<DoctorReport, CliError> {
     let (resolved, resolution) = match resolve_server_config(gateway_overrides) {
@@ -131,7 +148,7 @@ pub(crate) async fn collect_report(
         ),
         agents: collect_agents(target_agent, &resolved).await,
         host_plugins: crate::agents::collect_default_integration_readiness(),
-        observability: collect_observability(&resolved.gateway).await,
+        observability: collect_observability(&resolved.gateway, probe_mode).await,
         completions: collect_completions(home.as_deref()),
     })
 }
@@ -569,7 +586,7 @@ async fn probe_version(argv: &[String]) -> Option<String> {
     }
 }
 
-async fn collect_observability(gateway: &GatewayConfig) -> Vec<Check> {
+async fn collect_observability(gateway: &GatewayConfig, probe_mode: DoctorProbeMode) -> Vec<Check> {
     let mut checks = Vec::new();
 
     let Some(plugin_value) = &gateway.plugin_config else {
@@ -630,7 +647,7 @@ async fn collect_observability(gateway: &GatewayConfig) -> Vec<Check> {
     }
 
     if let Some(config) = observability_component_config(plugin_value) {
-        collect_observability_component_checks(&mut checks, config).await;
+        collect_observability_component_checks(&mut checks, config, probe_mode).await;
     } else {
         checks.push(Check {
             name: "Observability plugin",
@@ -639,8 +656,13 @@ async fn collect_observability(gateway: &GatewayConfig) -> Vec<Check> {
         });
     }
     collect_pricing_component_checks(&mut checks, &plugin_config);
-    collect_response_cache_component_checks(&mut checks, &plugin_config, response_cache_invalid)
-        .await;
+    collect_response_cache_component_checks(
+        &mut checks,
+        &plugin_config,
+        response_cache_invalid,
+        probe_mode,
+    )
+    .await;
 
     checks
 }
@@ -649,6 +671,7 @@ async fn collect_response_cache_component_checks(
     checks: &mut Vec<Check>,
     plugin_config: &PluginConfig,
     config_invalid: bool,
+    probe_mode: DoctorProbeMode,
 ) {
     // The response cache is a section of the adaptive component, not its own
     // plugin kind: find the adaptive component and look for `response_cache`.
@@ -705,6 +728,27 @@ async fn collect_response_cache_component_checks(
             return;
         }
     };
+    if probe_mode.is_offline()
+        && let Err(error) = response_cache::store::validate_backend_target(&config)
+    {
+        checks.push(Check {
+            name: "Response cache",
+            status: Status::Fail,
+            details: format!("invalid backend target: {error}"),
+        });
+        return;
+    }
+    if probe_mode.is_offline() && config.backend.kind != "in_memory" {
+        checks.push(Check {
+            name: "Response cache",
+            status: Status::Info,
+            details: format!(
+                "configured; live {} backend probe skipped (--offline)",
+                config.backend.kind
+            ),
+        });
+        return;
+    }
     checks.push(response_cache_backend_check(response_cache::check_backend_health(&config)).await);
 }
 
@@ -733,15 +777,19 @@ async fn response_cache_backend_check(
     }
 }
 
-async fn collect_observability_component_checks(checks: &mut Vec<Check>, config: &Value) {
+async fn collect_observability_component_checks(
+    checks: &mut Vec<Check>,
+    config: &Value,
+    probe_mode: DoctorProbeMode,
+) {
     checks.extend(observability_atof_file_checks(config));
     if let Some(check) = observability_file_exporter_check(config, "atif") {
         checks.push(check);
     }
-    checks.extend(observability_http_exporter_checks(config).await);
+    checks.extend(observability_http_exporter_checks(config, probe_mode).await);
     if section_enabled(config, "atof") && !atof_stream_sinks(config).is_empty() {
         if atof_streaming_supported() {
-            checks.extend(observability_atof_stream_checks(config).await);
+            checks.extend(observability_atof_stream_checks(config, probe_mode).await);
         } else {
             checks.push(Check {
                 name: "ATOF stream sink",
@@ -810,7 +858,10 @@ fn observability_file_exporter_check(config: &Value, section: &str) -> Option<Ch
     })
 }
 
-async fn observability_http_exporter_checks(config: &Value) -> Vec<Check> {
+async fn observability_http_exporter_checks(
+    config: &Value,
+    probe_mode: DoctorProbeMode,
+) -> Vec<Check> {
     if !section_enabled(config, "opentelemetry") {
         return Vec::new();
     }
@@ -838,13 +889,53 @@ async fn observability_http_exporter_checks(config: &Value) -> Vec<Check> {
                 match endpoint.get("endpoint").and_then(Value::as_str) {
                     Some(url) => {
                         let mut check = if transport == "grpc" {
-                            probe_tcp_named(label, url).await
+                            if probe_mode.is_offline() {
+                                match validate_grpc_endpoint(url) {
+                                    Ok(_) => Check {
+                                        name: label,
+                                        status: Status::Info,
+                                        details: format!(
+                                            "endpoints[{index}] ({endpoint_type}): live network probe skipped (--offline)"
+                                        ),
+                                    },
+                                    Err(details) => Check {
+                                        name: label,
+                                        status: Status::Fail,
+                                        details: format!(
+                                            "endpoints[{index}] ({endpoint_type}): {details}"
+                                        ),
+                                    },
+                                }
+                            } else {
+                                probe_tcp_named(label, url).await
+                            }
                         } else {
                             let effective_url = resolve_http_trace_endpoint(url);
-                            probe_otlp_http_named(label, effective_url.as_ref()).await
+                            if probe_mode.is_offline() {
+                                match validate_otlp_http_endpoint(effective_url.as_ref()) {
+                                    Ok(()) => Check {
+                                        name: label,
+                                        status: Status::Info,
+                                        details: format!(
+                                            "endpoints[{index}] ({endpoint_type}): live network probe skipped (--offline)"
+                                        ),
+                                    },
+                                    Err(details) => Check {
+                                        name: label,
+                                        status: Status::Fail,
+                                        details: format!(
+                                            "endpoints[{index}] ({endpoint_type}): {details}"
+                                        ),
+                                    },
+                                }
+                            } else {
+                                probe_otlp_http_named(label, effective_url.as_ref()).await
+                            }
                         };
-                        check.details =
-                            format!("endpoints[{index}] ({endpoint_type}): {}", check.details);
+                        if !probe_mode.is_offline() {
+                            check.details =
+                                format!("endpoints[{index}] ({endpoint_type}): {}", check.details);
+                        }
                         check
                     }
                     None => Check {
@@ -993,16 +1084,23 @@ fn atof_streaming_supported() -> bool {
     cfg!(feature = "atof-streaming")
 }
 
-async fn observability_atof_stream_checks(config: &Value) -> Vec<Check> {
+async fn observability_atof_stream_checks(
+    config: &Value,
+    probe_mode: DoctorProbeMode,
+) -> Vec<Check> {
     let streams = atof_stream_sinks(config);
     let mut checks = Vec::with_capacity(streams.len());
     for (index, sink) in streams {
-        checks.push(probe_atof_stream_sink(index, sink).await);
+        checks.push(probe_atof_stream_sink(index, sink, probe_mode).await);
     }
     checks
 }
 
-async fn probe_atof_stream_sink(index: usize, endpoint: &Value) -> Check {
+async fn probe_atof_stream_sink(
+    index: usize,
+    endpoint: &Value,
+    probe_mode: DoctorProbeMode,
+) -> Check {
     let name = "ATOF stream sink";
     let Some(url) = endpoint.get("url").and_then(Value::as_str) else {
         return Check {
@@ -1036,6 +1134,22 @@ async fn probe_atof_stream_sink(index: usize, endpoint: &Value) -> Check {
             };
         }
     };
+    if let Err(details) = validate_atof_stream_probe_target(index, transport, url) {
+        return Check {
+            name,
+            status: Status::Fail,
+            details,
+        };
+    }
+    if probe_mode.is_offline() {
+        return Check {
+            name,
+            status: Status::Info,
+            details: format!(
+                "sinks[{index}] {transport} {url}: live network probe skipped (--offline)"
+            ),
+        };
+    }
     let payload = match doctor_atof_probe_payload() {
         Ok(payload) => payload,
         Err(err) => {
@@ -1061,7 +1175,40 @@ async fn probe_atof_stream_sink(index: usize, endpoint: &Value) -> Check {
 
 #[cfg(test)]
 async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check {
-    probe_atof_stream_sink(index, endpoint).await
+    probe_atof_stream_sink(index, endpoint, DoctorProbeMode::Live).await
+}
+
+fn validate_atof_stream_probe_target(
+    index: usize,
+    transport: &str,
+    url: &str,
+) -> Result<(), String> {
+    let parsed = reqwest::Url::parse(url)
+        .map_err(|error| format!("sinks[{index}] {transport} {url}: {error}"))?;
+    if parsed.host_str().is_none() {
+        return Err(format!("sinks[{index}] {transport} {url}: missing host"));
+    }
+    let valid_scheme = match transport {
+        "http_post" | "ndjson" => matches!(parsed.scheme(), "http" | "https"),
+        "websocket" => matches!(parsed.scheme(), "ws" | "wss"),
+        _ => {
+            return Err(format!(
+                "sinks[{index}] {transport} {url}: unsupported transport"
+            ));
+        }
+    };
+    if valid_scheme {
+        Ok(())
+    } else {
+        let expected = match transport {
+            "http_post" | "ndjson" => "http or https",
+            "websocket" => "ws or wss",
+            _ => unreachable!("unsupported transports return earlier"),
+        };
+        Err(format!(
+            "sinks[{index}] {transport} {url}: invalid scheme (must be {expected})"
+        ))
+    }
 }
 
 fn endpoint_headers(endpoint: &Value) -> Result<Vec<(String, String)>, String> {
@@ -1077,7 +1224,14 @@ fn endpoint_headers(endpoint: &Value) -> Result<Vec<(String, String)>, String> {
             let Some(value) = value.as_str() else {
                 return Err(format!("headers.{key} must be a string"));
             };
-            names.insert(name);
+            if value.trim().is_empty() {
+                return Err(format!("headers.{key} must not be blank"));
+            }
+            reqwest::header::HeaderValue::from_bytes(value.as_bytes())
+                .map_err(|error| format!("headers.{key} invalid: {error}"))?;
+            if !names.insert(name) {
+                return Err(format!("header {key:?} appears more than once"));
+            }
             out.push((key.clone(), value.to_string()));
         }
     }
@@ -1101,6 +1255,8 @@ fn endpoint_headers(endpoint: &Value) -> Result<Vec<(String, String)>, String> {
             if value.trim().is_empty() {
                 return Err(format!("environment variable {variable:?} is blank"));
             }
+            reqwest::header::HeaderValue::from_bytes(value.as_bytes())
+                .map_err(|error| format!("header_env.{key} invalid: {error}"))?;
             names.insert(name);
             out.push((key.clone(), value));
         }
@@ -1374,10 +1530,11 @@ pub(crate) fn format_agents_json(agents: &[AgentInfo]) -> Result<String, CliErro
 pub(crate) async fn run_doctor(
     target_agent: Option<CodingAgent>,
     json: bool,
+    probe_mode: DoctorProbeMode,
     gateway_overrides: &GatewayOverrides,
     logging_fallback_error: Option<&CliError>,
 ) -> Result<std::process::ExitCode, CliError> {
-    let mut report = collect_report(target_agent, gateway_overrides).await?;
+    let mut report = collect_report(target_agent, probe_mode, gateway_overrides).await?;
     if let Some(error) = logging_fallback_error {
         let logging_details = format!(
             "could not resolve logging configuration: {error}; repair or recreate the named logging configuration file"
diff --git a/crates/cli/src/diagnostics/probes.rs b/crates/cli/src/diagnostics/probes.rs
index 2625e0c56..4ba55df76 100644
--- a/crates/cli/src/diagnostics/probes.rs
+++ b/crates/cli/src/diagnostics/probes.rs
@@ -66,34 +66,31 @@ pub(super) async fn probe_otlp_http_named(name: &'static str, url: &str) -> Chec
             } else {
                 Status::Warn
             },
-            details: format!("{} (HTTP {})", url, response.status().as_u16()),
+            details: format!(
+                "{} (live HTTP reachability probe returned HTTP {})",
+                url,
+                response.status().as_u16()
+            ),
         },
         Err(error) => Check {
             name,
             status: Status::Fail,
-            details: format!("{url}: {error}"),
+            details: format!("{url}: live HTTP reachability probe failed: {error}"),
         },
     }
 }
 
 pub(super) async fn probe_tcp_named(name: &'static str, endpoint: &str) -> Check {
-    let parsed = match reqwest::Url::parse(endpoint) {
-        Ok(parsed) => parsed,
-        Err(error) => {
+    let (parsed, host) = match validate_grpc_endpoint(endpoint) {
+        Ok((parsed, host)) => (parsed, host),
+        Err(details) => {
             return Check {
                 name,
                 status: Status::Fail,
-                details: format!("{endpoint}: invalid gRPC endpoint: {error}"),
+                details,
             };
         }
     };
-    let Some(host) = parsed.host_str() else {
-        return Check {
-            name,
-            status: Status::Fail,
-            details: format!("{endpoint}: gRPC endpoint has no host"),
-        };
-    };
     let port = grpc_endpoint_port(&parsed);
     match tokio::time::timeout(
         NETWORK_TIMEOUT,
@@ -104,21 +101,52 @@ pub(super) async fn probe_tcp_named(name: &'static str, endpoint: &str) -> Check
         Ok(Ok(_)) => Check {
             name,
             status: Status::Pass,
-            details: format!("{endpoint} (gRPC TCP connection succeeded)"),
+            details: format!(
+                "{endpoint} (live gRPC reachability probe connected to the TCP port; OTLP handshake not verified)"
+            ),
         },
         Ok(Err(error)) => Check {
             name,
             status: Status::Fail,
-            details: format!("{endpoint}: gRPC TCP connection failed: {error}"),
+            details: format!("{endpoint}: live gRPC reachability probe failed: {error}"),
         },
         Err(_) => Check {
             name,
             status: Status::Fail,
-            details: format!("{endpoint}: gRPC TCP connection timed out"),
+            details: format!("{endpoint}: live gRPC reachability probe timed out"),
         },
     }
 }
 
+pub(super) fn validate_grpc_endpoint(endpoint: &str) -> Result<(reqwest::Url, String), String> {
+    let parsed = reqwest::Url::parse(endpoint)
+        .map_err(|error| format!("{endpoint}: invalid gRPC endpoint: {error}"))?;
+    if !matches!(parsed.scheme(), "http" | "https") {
+        return Err(format!(
+            "{endpoint}: gRPC endpoint must use http:// or https://"
+        ));
+    }
+    let host = parsed
+        .host_str()
+        .map(str::to_owned)
+        .ok_or_else(|| format!("{endpoint}: gRPC endpoint has no host"))?;
+    Ok((parsed, host))
+}
+
+pub(super) fn validate_otlp_http_endpoint(endpoint: &str) -> Result<(), String> {
+    let parsed = reqwest::Url::parse(endpoint)
+        .map_err(|error| format!("{endpoint}: invalid OTLP HTTP endpoint: {error}"))?;
+    if !matches!(parsed.scheme(), "http" | "https") {
+        return Err(format!(
+            "{endpoint}: OTLP HTTP endpoint must use http:// or https://"
+        ));
+    }
+    if parsed.host_str().is_none() {
+        return Err(format!("{endpoint}: OTLP HTTP endpoint has no host"));
+    }
+    Ok(())
+}
+
 fn grpc_endpoint_port(endpoint: &reqwest::Url) -> u16 {
     endpoint.port().unwrap_or_else(|| {
         if endpoint.scheme() == "https" {
diff --git a/crates/cli/tests/cli_tests.rs b/crates/cli/tests/cli_tests.rs
index ebe379418..4fce1eea1 100644
--- a/crates/cli/tests/cli_tests.rs
+++ b/crates/cli/tests/cli_tests.rs
@@ -4077,8 +4077,9 @@ try:
     read_until("RELAY_SHELL> ")
     assert os.tcgetpgrp(master) == pid, (os.tcgetpgrp(master), pid, buffer)
 
-    os.write(master, b"bg\n")
-    read_until("RELAY_SHELL> ")
+    # Continue Relay's stopped shell job directly. Shell `bg` bookkeeping varies across the
+    # macOS runner shell, but Relay only observes the process-group SIGCONT.
+    os.killpg(relay_group, signal.SIGCONT)
     time.sleep(0.1)
     assert os.tcgetpgrp(master) == pid, (os.tcgetpgrp(master), pid, buffer)
     os.write(master, b"echo BG_SHELL_OK\n")
@@ -4099,8 +4100,7 @@ try:
     read_until("AGENT_DELAY_ARMED")
     os.write(master, b"\x1a")
     read_until("RELAY_SHELL> ")
-    os.write(master, b"bg\n")
-    read_until("RELAY_SHELL> ")
+    os.killpg(relay_group, signal.SIGCONT)
     wait_until_present("AGENT_BG_DELAY")
     assert os.tcgetpgrp(master) == pid, (os.tcgetpgrp(master), pid, buffer)
     os.write(master, b"fg\n")
diff --git a/crates/cli/tests/coverage/commands/main_tests.rs b/crates/cli/tests/coverage/commands/main_tests.rs
index 2dfec86c5..2967a50ad 100644
--- a/crates/cli/tests/coverage/commands/main_tests.rs
+++ b/crates/cli/tests/coverage/commands/main_tests.rs
@@ -353,6 +353,15 @@ fn doctor_rejects_conflicting_agent_and_plugin_targets() {
     assert!(error.to_string().contains("cannot be used with"));
 }
 
+#[test]
+fn doctor_accepts_offline_flag() {
+    let cli = Cli::try_parse_from(["nemo-relay", "doctor", "--offline"]).unwrap();
+    match cli.command {
+        Some(Command::Doctor(command)) => assert!(command.offline),
+        other => panic!("expected doctor command, got {other:?}"),
+    }
+}
+
 #[test]
 fn multi_agent_operations_attempt_every_target_before_reporting_errors() {
     let visited = std::cell::RefCell::new(Vec::new());
diff --git a/crates/cli/tests/coverage/shared/doctor_tests.rs b/crates/cli/tests/coverage/shared/doctor_tests.rs
index 2e3d9d054..3c3da1391 100644
--- a/crates/cli/tests/coverage/shared/doctor_tests.rs
+++ b/crates/cli/tests/coverage/shared/doctor_tests.rs
@@ -101,6 +101,27 @@ fn empty_report() -> DoctorReport {
     }
 }
 
+async fn collect_live_observability(gateway: &GatewayConfig) -> Vec<Check> {
+    collect_observability(gateway, DoctorProbeMode::Live).await
+}
+
+async fn collect_offline_observability(gateway: &GatewayConfig) -> Vec<Check> {
+    collect_observability(gateway, DoctorProbeMode::Offline).await
+}
+
+async fn live_observability_http_exporter_checks(config: &serde_json::Value) -> Vec<Check> {
+    observability_http_exporter_checks(config, DoctorProbeMode::Live).await
+}
+
+async fn offline_observability_http_exporter_checks(config: &serde_json::Value) -> Vec<Check> {
+    observability_http_exporter_checks(config, DoctorProbeMode::Offline).await
+}
+
+#[cfg(test)]
+async fn offline_atof_endpoint(index: usize, endpoint: &serde_json::Value) -> Check {
+    probe_atof_stream_sink(index, endpoint, DoctorProbeMode::Offline).await
+}
+
 #[test]
 fn exit_code_passes_when_no_failures() {
     let report = empty_report();
@@ -1066,11 +1087,16 @@ async fn opentelemetry_doctor_uses_tcp_probe_for_grpc_endpoints() {
         }
     });
 
-    let checks = observability_http_exporter_checks(&config).await;
+    let checks = live_observability_http_exporter_checks(&config).await;
 
     assert_eq!(checks.len(), 1);
     assert_eq!(checks[0].status, Status::Pass);
-    assert!(checks[0].details.contains("gRPC TCP connection succeeded"));
+    assert!(
+        checks[0]
+            .details
+            .contains("live gRPC reachability probe connected to the TCP port")
+    );
+    assert!(checks[0].details.contains("OTLP handshake not verified"));
     assert!(checks[0].details.contains("endpoints[0] (gen_ai)"));
     accept.join().unwrap();
 }
@@ -1078,14 +1104,14 @@ async fn opentelemetry_doctor_uses_tcp_probe_for_grpc_endpoints() {
 #[tokio::test]
 async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_routes() {
     assert!(
-        observability_http_exporter_checks(&serde_json::json!({
+        live_observability_http_exporter_checks(&serde_json::json!({
             "opentelemetry": {"enabled": true, "endpoints": "not-a-list"}
         }))
         .await
         .is_empty()
     );
 
-    let missing = observability_http_exporter_checks(&serde_json::json!({
+    let missing = live_observability_http_exporter_checks(&serde_json::json!({
         "opentelemetry": {
             "enabled": true,
             "endpoints": [{"type": "openinference"}]
@@ -1105,7 +1131,7 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_
             .write_all(b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n")
             .unwrap();
     });
-    let checks = observability_http_exporter_checks(&serde_json::json!({
+    let checks = live_observability_http_exporter_checks(&serde_json::json!({
         "opentelemetry": {
             "enabled": true,
             "endpoints": [{"type": "full", "endpoint": endpoint}]
@@ -1114,7 +1140,11 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_
     .await;
     assert_eq!(checks[0].status, Status::Pass);
     assert!(checks[0].details.contains("endpoints[0] (full)"));
-    assert!(checks[0].details.contains("/v1/traces (HTTP 405)"));
+    assert!(
+        checks[0]
+            .details
+            .contains("/v1/traces (live HTTP reachability probe returned HTTP 405)")
+    );
     accept.join().unwrap();
 
     let listener = TcpListener::bind("127.0.0.1:0").unwrap();
@@ -1127,7 +1157,7 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_
             .unwrap();
         request
     });
-    let checks = observability_http_exporter_checks(&serde_json::json!({
+    let checks = live_observability_http_exporter_checks(&serde_json::json!({
         "opentelemetry": {
             "enabled": true,
             "endpoints": [{"type": "full", "endpoint": endpoint}]
@@ -1135,7 +1165,11 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_
     }))
     .await;
     assert_eq!(checks[0].status, Status::Pass);
-    assert!(checks[0].details.contains("/ (HTTP 405)"));
+    assert!(
+        checks[0]
+            .details
+            .contains("/ (live HTTP reachability probe returned HTTP 405)")
+    );
     let request = accept.join().unwrap();
     assert!(request.starts_with("GET / HTTP/1.1"));
 
@@ -1148,7 +1182,7 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_
             .write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n")
             .unwrap();
     });
-    let checks = observability_http_exporter_checks(&serde_json::json!({
+    let checks = live_observability_http_exporter_checks(&serde_json::json!({
         "opentelemetry": {
             "enabled": true,
             "endpoints": [{"type": "full", "endpoint": endpoint}]
@@ -1156,10 +1190,104 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_
     }))
     .await;
     assert_eq!(checks[0].status, Status::Warn);
-    assert!(checks[0].details.contains("/wrong (HTTP 404)"));
+    assert!(
+        checks[0]
+            .details
+            .contains("/wrong (live HTTP reachability probe returned HTTP 404)")
+    );
     accept.join().unwrap();
 }
 
+#[tokio::test]
+async fn opentelemetry_doctor_skips_live_network_probes_offline() {
+    let checks = offline_observability_http_exporter_checks(&serde_json::json!({
+        "opentelemetry": {
+            "enabled": true,
+            "endpoints": [
+                {
+                    "type": "gen_ai",
+                    "transport": "grpc",
+                    "endpoint": "http://127.0.0.1:4317"
+                },
+                {
+                    "type": "openinference",
+                    "endpoint": "http://127.0.0.1:4318"
+                }
+            ]
+        }
+    }))
+    .await;
+
+    assert_eq!(checks.len(), 2);
+    assert!(checks.iter().all(|check| check.status == Status::Info));
+    assert!(checks.iter().all(|check| {
+        check
+            .details
+            .contains("live network probe skipped (--offline)")
+    }));
+}
+
+#[tokio::test]
+async fn opentelemetry_doctor_offline_still_rejects_malformed_endpoints() {
+    let checks = offline_observability_http_exporter_checks(&serde_json::json!({
+        "opentelemetry": {
+            "enabled": true,
+            "endpoints": [
+                {
+                    "type": "gen_ai",
+                    "transport": "grpc",
+                    "endpoint": "http://:4317"
+                },
+                {
+                    "type": "full",
+                    "endpoint": "http://:4318"
+                }
+            ]
+        }
+    }))
+    .await;
+
+    assert_eq!(checks.len(), 2);
+    assert!(checks[0].details.contains("invalid gRPC endpoint"));
+    assert!(checks[1].details.contains("invalid OTLP HTTP endpoint"));
+    assert!(checks.iter().all(|check| check.status == Status::Fail));
+}
+
+#[tokio::test]
+async fn opentelemetry_doctor_offline_rejects_unsupported_endpoint_schemes() {
+    let checks = offline_observability_http_exporter_checks(&serde_json::json!({
+        "opentelemetry": {
+            "enabled": true,
+            "endpoints": [
+                {
+                    "type": "gen_ai",
+                    "transport": "grpc",
+                    "endpoint": "ftp://collector.example:4317"
+                },
+                {
+                    "type": "full",
+                    "endpoint": "file:///tmp/collector"
+                }
+            ]
+        }
+    }))
+    .await;
+
+    assert_eq!(checks.len(), 2);
+    assert_eq!(checks[0].status, Status::Fail);
+    assert_eq!(checks[1].status, Status::Fail);
+    assert!(
+        checks[0]
+            .details
+            .contains("gRPC endpoint must use http:// or https://")
+    );
+    assert!(
+        checks[1]
+            .details
+            .contains("OTLP HTTP endpoint must use http:// or https://")
+    );
+}
+
 #[test]
 fn atof_file_checks_preserve_configured_sink_indices() {
     let config = serde_json::json!({
@@ -1214,7 +1342,7 @@ async fn collect_observability_warns_for_missing_atif_dir_without_creating_it()
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let atif_check = checks
         .iter()
@@ -1251,7 +1379,7 @@ async fn collect_observability_registers_adaptive_before_validation() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     assert!(
         !checks.iter().any(|check| check
@@ -1285,7 +1413,7 @@ async fn collect_observability_reports_response_cache_on_when_configured() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let cache = checks
         .iter()
@@ -1299,6 +1427,98 @@ async fn collect_observability_reports_response_cache_on_when_configured() {
     );
 }
 
+#[tokio::test]
+async fn collect_observability_skips_redis_response_cache_probe_offline() {
+    let gateway = GatewayConfig {
+        plugin_config: Some(serde_json::json!({
+            "version": 1,
+            "components": [
+                {
+                    "kind": "adaptive",
+                    "enabled": true,
+                    "config": {
+                        "response_cache": {
+                            "ttl_seconds": 3600,
+                            "namespace": "doctor-test",
+                            "backend": {
+                                "kind": "redis",
+                                "config": {
+                                    "url": "redis://127.0.0.1:6379",
+                                    "key_prefix": "doctor-test:"
+                                }
+                            }
+                        }
+                    }
+                }
+            ]
+        })),
+        ..GatewayConfig::default()
+    };
+
+    let checks = collect_offline_observability(&gateway).await;
+
+    let cache = checks
+        .iter()
+        .find(|check| check.name == "Response cache")
+        .expect("a Response cache check should be present");
+    assert_eq!(cache.status, Status::Info, "checks: {checks:?}");
+    assert_eq!(
+        cache.details,
+        "configured; live redis backend probe skipped (--offline)"
+    );
+}
+
+#[tokio::test]
+async fn collect_observability_fails_invalid_redis_response_cache_target_offline() {
+    let gateway = GatewayConfig {
+        plugin_config: Some(serde_json::json!({
+            "version": 1,
+            "components": [
+                {
+                    "kind": "adaptive",
+                    "enabled": true,
+                    "config": {
+                        "response_cache": {
+                            "ttl_seconds": 3600,
+                            "namespace": "doctor-test",
+                            "backend": {
+                                "kind": "redis",
+                                "config": {
+                                    "url": "not-a-redis-url",
+                                    "key_prefix": "doctor-test:"
+                                }
+                            }
+                        }
+                    }
+                }
+            ]
+        })),
+        ..GatewayConfig::default()
+    };
+
+    let checks = collect_offline_observability(&gateway).await;
+
+    let cache = checks
+        .iter()
+        .find(|check| check.name == "Response cache")
+        .expect("a Response cache check should be present");
+    assert_eq!(cache.status, Status::Fail, "checks: {checks:?}");
+    assert!(
+        cache.details.contains("invalid backend target"),
+        "details: {}",
+        cache.details
+    );
+    assert!(
+        cache.details.contains("redis client"),
+        "details: {}",
+        cache.details
+    );
+    assert!(
+        !cache.details.contains("probe skipped"),
+        "must not report skipped for an invalid backend target"
+    );
+}
+
 #[tokio::test(start_paused = true)]
 async fn response_cache_backend_check_reports_timeout() {
     let check = response_cache_backend_check(std::future::pending()).await;
@@ -1333,7 +1553,7 @@ async fn collect_observability_reports_response_cache_not_configured_without_sec
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let cache = checks
         .iter()
@@ -1365,7 +1585,7 @@ async fn collect_observability_reports_response_cache_fail_when_config_invalid()
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let cache = checks
         .iter()
@@ -1415,7 +1635,7 @@ async fn collect_observability_registers_pii_redaction_before_validation() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     assert!(
         !checks.iter().any(|check| check
@@ -1447,7 +1667,7 @@ async fn collect_observability_reports_invalid_pii_redaction_config() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let diagnostic = checks
         .iter()
@@ -1483,7 +1703,7 @@ async fn collect_observability_probes_atof_streaming_endpoint() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
     let body = tokio::time::timeout(std::time::Duration::from_secs(2), async {
         loop {
             let captured = body.lock().unwrap().clone();
@@ -1510,11 +1730,11 @@ async fn collect_observability_probes_atof_streaming_endpoint() {
 
 #[tokio::test]
 async fn collect_observability_covers_absent_invalid_and_componentless_configs() {
-    let absent = collect_observability(&GatewayConfig::default()).await;
+    let absent = collect_live_observability(&GatewayConfig::default()).await;
     assert_eq!(absent[0].status, Status::Info);
     assert!(absent[0].details.contains("not configured"));
 
-    let invalid = collect_observability(&GatewayConfig {
+    let invalid = collect_live_observability(&GatewayConfig {
         plugin_config: Some(serde_json::json!({"version": "bad"})),
         ..GatewayConfig::default()
     })
@@ -1522,7 +1742,7 @@ async fn collect_observability_covers_absent_invalid_and_componentless_configs()
     assert_eq!(invalid[0].status, Status::Fail);
     assert!(invalid[0].details.contains("invalid plugin config"));
 
-    let no_observability = collect_observability(&GatewayConfig {
+    let no_observability = collect_live_observability(&GatewayConfig {
         plugin_config: Some(serde_json::json!({
             "version": 1,
             "components": []
@@ -1567,7 +1787,7 @@ async fn collect_observability_rejects_websocket_endpoint_http_scheme() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let endpoint = checks
         .iter()
@@ -1621,8 +1841,42 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp
             .contains("headers.x-test must be a string")
     );
 
-    let mixed_case_duplicate = probe_atof_endpoint(
+    let blank_static_header = probe_atof_endpoint(
         4,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "headers": {"x-test": "   "}
+        }),
+    )
+    .await;
+    assert_eq!(blank_static_header.status, Status::Fail);
+    assert!(
+        blank_static_header
+            .details
+            .contains("headers.x-test must not be blank")
+    );
+
+    let _env = EnvScope::set(&[(
+        "NEMO_RELAY_INVALID_ATOF_HEADER",
+        Some(std::ffi::OsStr::new("bad\nvalue")),
+    )]);
+    let invalid_header_env = probe_atof_endpoint(
+        5,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "header_env": {"x-test": "NEMO_RELAY_INVALID_ATOF_HEADER"}
+        }),
+    )
+    .await;
+    assert_eq!(invalid_header_env.status, Status::Fail);
+    assert!(
+        invalid_header_env
+            .details
+            .contains("header_env.x-test invalid")
+    );
+
+    let mixed_case_duplicate = probe_atof_endpoint(
+        6,
         &serde_json::json!({
             "url": "http://127.0.0.1:1/events",
             "headers": {"Authorization": "Bearer literal"},
@@ -1637,8 +1891,26 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp
             .contains("cannot appear in both headers and header_env")
     );
 
+    let duplicate_static_header = probe_atof_endpoint(
+        7,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "headers": {
+                "Authorization": "Bearer a",
+                "authorization": "Bearer b"
+            }
+        }),
+    )
+    .await;
+    assert_eq!(duplicate_static_header.status, Status::Fail);
+    assert!(
+        duplicate_static_header
+            .details
+            .contains("appears more than once")
+    );
+
     let unsupported = probe_atof_endpoint(
-        5,
+        8,
         &serde_json::json!({
             "url": "http://127.0.0.1:1/events",
             "transport": "grpc"
@@ -1649,6 +1921,90 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp
     assert!(unsupported.details.contains("unsupported transport"));
 }
 
+#[tokio::test]
+async fn atof_endpoint_offline_skips_live_network_probe_after_validation() {
+    let skipped = offline_atof_endpoint(
+        0,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "transport": "http_post"
+        }),
+    )
+    .await;
+
+    assert_eq!(skipped.status, Status::Info);
+    assert_eq!(
+        skipped.details,
+        "sinks[0] http_post http://127.0.0.1:1/events: live network probe skipped (--offline)"
+    );
+}
+
+#[tokio::test]
+async fn atof_endpoint_offline_still_rejects_invalid_transport_and_scheme() {
+    let invalid_websocket = offline_atof_endpoint(
+        0,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "transport": "websocket"
+        }),
+    )
+    .await;
+    assert_eq!(invalid_websocket.status, Status::Fail);
+    assert!(
+        invalid_websocket
+            .details
+            .contains("invalid scheme (must be ws or wss)")
+    );
+
+    let unsupported_transport = offline_atof_endpoint(
+        1,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "transport": "udp"
+        }),
+    )
+    .await;
+    assert_eq!(unsupported_transport.status, Status::Fail);
+    assert!(
+        unsupported_transport
+            .details
+            .contains("unsupported transport")
+    );
+
+    let blank_static_header = offline_atof_endpoint(
+        2,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "headers": {"x-test": "   "}
+        }),
+    )
+    .await;
+    assert_eq!(blank_static_header.status, Status::Fail);
+    assert!(
+        blank_static_header
+            .details
+            .contains("headers.x-test must not be blank")
+    );
+
+    let duplicate_static_header = offline_atof_endpoint(
+        3,
+        &serde_json::json!({
+            "url": "http://127.0.0.1:1/events",
+            "headers": {
+                "Authorization": "Bearer a",
+                "authorization": "Bearer b"
+            }
+        }),
+    )
+    .await;
+    assert_eq!(duplicate_static_header.status, Status::Fail);
+    assert!(
+        duplicate_static_header
+            .details
+            .contains("appears more than once")
+    );
+}
+
 #[tokio::test]
 async fn atof_http_and_websocket_probes_report_failure_branches() {
     let listener = TcpListener::bind("127.0.0.1:0").unwrap();
@@ -1869,7 +2225,7 @@ async fn collect_observability_validates_pricing_file_source() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let pricing = checks
         .iter()
@@ -1901,7 +2257,7 @@ async fn collect_observability_fails_for_missing_pricing_file_source() {
         ..GatewayConfig::default()
     };
 
-    let checks = collect_observability(&gateway).await;
+    let checks = collect_live_observability(&gateway).await;
 
     let pricing = checks
         .iter()
diff --git a/crates/cli/tests/coverage/shared/probes_tests.rs b/crates/cli/tests/coverage/shared/probes_tests.rs
index 17335fa38..bb2958a25 100644
--- a/crates/cli/tests/coverage/shared/probes_tests.rs
+++ b/crates/cli/tests/coverage/shared/probes_tests.rs
@@ -9,7 +9,12 @@ async fn grpc_probe_uses_tcp_connectivity() {
     let endpoint = format!("http://{}", listener.local_addr().unwrap());
     let check = probe_tcp_named("OpenTelemetry endpoint", &endpoint).await;
     assert_eq!(check.status, Status::Pass);
-    assert!(check.details.contains("gRPC TCP connection succeeded"));
+    assert!(
+        check
+            .details
+            .contains("live gRPC reachability probe connected to the TCP port")
+    );
+    assert!(check.details.contains("OTLP handshake not verified"));
 }
 
 #[tokio::test]
@@ -20,7 +25,11 @@ async fn grpc_probe_reports_invalid_hostless_and_refused_endpoints() {
 
     let hostless = probe_tcp_named("OpenTelemetry endpoint", "file:///tmp/collector").await;
     assert_eq!(hostless.status, Status::Fail);
-    assert!(hostless.details.contains("has no host"));
+    assert!(
+        hostless
+            .details
+            .contains("gRPC endpoint must use http:// or https://")
+    );
 
     let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
     let endpoint = format!("http://{}", listener.local_addr().unwrap());
@@ -28,8 +37,12 @@ async fn grpc_probe_reports_invalid_hostless_and_refused_endpoints() {
     let refused = probe_tcp_named("OpenTelemetry endpoint", &endpoint).await;
     assert_eq!(refused.status, Status::Fail);
     assert!(
-        refused.details.contains("connection failed")
-            || refused.details.contains("connection timed out"),
+        refused
+            .details
+            .contains("live gRPC reachability probe failed")
+            || refused
+                .details
+                .contains("live gRPC reachability probe timed out"),
         "{}",
         refused.details
     );
diff --git a/crates/core/tests/fixtures/worker_plugin/Cargo.lock b/crates/core/tests/fixtures/worker_plugin/Cargo.lock
index feb1c1172..548bd3517 100644
--- a/crates/core/tests/fixtures/worker_plugin/Cargo.lock
+++ b/crates/core/tests/fixtures/worker_plugin/Cargo.lock
@@ -4,22 +4,13 @@ version = 4
 
 [[package]]
 name = "aho-corasick"
-version = "1.1.4"
+version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
+checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
 dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "android_system_properties"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
-dependencies = [
- "libc",
-]
-
 [[package]]
 name = "anyhow"
 version = "1.0.104"
@@ -119,16 +110,6 @@ version = "1.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
 
-[[package]]
-name = "cc"
-version = "1.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9"
-dependencies = [
- "find-msvc-tools",
- "shlex",
-]
-
 [[package]]
 name = "cfg-if"
 version = "1.0.4"
@@ -141,20 +122,10 @@ version = "0.4.45"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
 dependencies = [
- "iana-time-zone",
- "js-sys",
  "num-traits",
  "serde",
- "wasm-bindgen",
- "windows-link",
 ]
 
-[[package]]
-name = "core-foundation-sys"
-version = "0.8.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
-
 [[package]]
 name = "either"
 version = "1.17.0"
@@ -183,12 +154,6 @@ version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223"
 
-[[package]]
-name = "find-msvc-tools"
-version = "0.1.9"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
-
 [[package]]
 name = "fixedbitset"
 version = "0.5.7"
@@ -421,30 +386,6 @@ dependencies = [
  "tracing",
 ]
 
-[[package]]
-name = "iana-time-zone"
-version = "0.1.65"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470"
-dependencies = [
- "android_system_properties",
- "core-foundation-sys",
- "iana-time-zone-haiku",
- "js-sys",
- "log",
- "wasm-bindgen",
- "windows-core",
-]
-
-[[package]]
-name = "iana-time-zone-haiku"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f"
-dependencies = [
- "cc",
-]
-
 [[package]]
 name = "indexmap"
 version = "2.14.0"
@@ -913,12 +854,6 @@ dependencies = [
  "zmij",
 ]
 
-[[package]]
-name = "shlex"
-version = "2.0.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
-
 [[package]]
 name = "slab"
 version = "0.4.12"
@@ -1303,65 +1238,12 @@ dependencies = [
  "unicode-ident",
 ]
 
-[[package]]
-name = "windows-core"
-version = "0.62.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
-dependencies = [
- "windows-implement",
- "windows-interface",
- "windows-link",
- "windows-result",
- "windows-strings",
-]
-
-[[package]]
-name = "windows-implement"
-version = "0.60.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
-]
-
-[[package]]
-name = "windows-interface"
-version = "0.59.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
-]
-
 [[package]]
 name = "windows-link"
 version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
 
-[[package]]
-name = "windows-result"
-version = "0.4.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
-dependencies = [
- "windows-link",
-]
-
-[[package]]
-name = "windows-strings"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
-dependencies = [
- "windows-link",
-]
-
 [[package]]
 name = "windows-sys"
 version = "0.61.2"
diff --git a/crates/python/src/py_api/mod.rs b/crates/python/src/py_api/mod.rs
index b2f055dd4..d09933a0f 100644
--- a/crates/python/src/py_api/mod.rs
+++ b/crates/python/src/py_api/mod.rs
@@ -108,6 +108,16 @@ impl SafeFutureCompleter {
     }
 }
 
+fn panic_message(panic: &(dyn std::any::Any + Send)) -> &str {
+    if let Some(message) = panic.downcast_ref::<&str>() {
+        message
+    } else if let Some(message) = panic.downcast_ref::<String>() {
+        message.as_str()
+    } else {
+        "unknown error"
+    }
+}
+
 fn safe_future_into_py<'py, F>(py: Python<'py>, future: F) -> PyResult<Bound<'py, PyAny>>
 where
     F: Future<Output = PyResult<Py<PyAny>>> + Send + 'static,
@@ -125,9 +135,22 @@ where
     )?;
     let completion_future = python_future.clone_ref(py);
     pyo3_async_runtimes::tokio::get_runtime().spawn(async move {
+        let mut task = tokio::spawn(pyo3_async_runtimes::tokio::scope(locals, future));
         let result = tokio::select! {
-            result = pyo3_async_runtimes::tokio::scope(locals, future) => Some(result),
-            _ = &mut cancel_receiver => None,
+            result = &mut task => Some(
+                match result {
+                    Ok(result) => result,
+                    Err(error) if error.is_panic() => Err(pyo3_async_runtimes::err::RustPanic::new_err(
+                        format!("rust future panicked: {}", panic_message(error.into_panic().as_ref())),
+                    )),
+                    Err(error) => Err(pyo3::exceptions::PyRuntimeError::new_err(error.to_string())),
+                },
+            ),
+            _ = &mut cancel_receiver => {
+                task.abort();
+                let _ = task.await;
+                None
+            },
         };
         let Some(result) = result else { return };
         Python::attach(|py| {
diff --git a/crates/python/tests/coverage/py_api_coverage_tests.rs b/crates/python/tests/coverage/py_api_coverage_tests.rs
index 5a449b960..cf87cbea1 100644
--- a/crates/python/tests/coverage/py_api_coverage_tests.rs
+++ b/crates/python/tests/coverage/py_api_coverage_tests.rs
@@ -6,6 +6,8 @@
 use super::*;
 
 use std::ffi::CString;
+use std::sync::mpsc;
+use std::time::Duration;
 
 use pyo3::types::PyModule;
 use serde_json::json;
@@ -38,6 +40,142 @@ fn with_event_loop<T>(py: Python<'_>, f: impl FnOnce(Bound<'_, PyAny>) -> T) ->
     result
 }
 
+fn test_loop(py: Python<'_>, closed: bool) -> Bound<'_, PyAny> {
+    let module = load_module(
+        py,
+        r#"
+import threading
+
+class Future:
+    def __init__(self):
+        self._callbacks = []
+        self._cancelled = False
+
+    def add_done_callback(self, callback):
+        self._callbacks.append(callback)
+
+    def cancelled(self):
+        return self._cancelled
+
+    def cancel(self):
+        self._cancelled = True
+        for callback in self._callbacks:
+            callback(self)
+
+class Loop:
+    def __init__(self, closed):
+        self.closed = closed
+        self.closed_checked = threading.Event()
+        self.completion_scheduled = False
+
+    def create_future(self):
+        return Future()
+
+    def is_closed(self):
+        self.closed_checked.set()
+        return self.closed
+
+    def call_soon_threadsafe(self, callback):
+        self.completion_scheduled = True
+        callback()
+"#,
+    );
+    module.getattr("Loop").unwrap().call1((closed,)).unwrap()
+}
+
+struct CancellationSignal(mpsc::Sender<()>);
+
+impl Drop for CancellationSignal {
+    fn drop(&mut self) {
+        let _ = self.0.send(());
+    }
+}
+
+#[test]
+fn safe_future_into_py_settles_rust_panics() {
+    let _python = crate::test_support::init_python_test();
+    Python::attach(|py| {
+        with_event_loop(py, |event_loop| {
+            let locals = pyo3_async_runtimes::TaskLocals::new(event_loop.clone());
+            let future = pyo3_async_runtimes::tokio::get_runtime()
+                .block_on(pyo3_async_runtimes::tokio::scope(locals, async move {
+                    Python::attach(|py| {
+                        safe_future_into_py(py, async move { panic!("expected test panic") })
+                            .map(Bound::unbind)
+                    })
+                }))
+                .unwrap();
+
+            let error = event_loop
+                .call_method1("run_until_complete", (future,))
+                .unwrap_err();
+            assert!(error.is_instance_of::<pyo3_async_runtimes::err::RustPanic>(py));
+            assert!(error.to_string().contains("expected test panic"));
+        });
+    });
+}
+
+#[test]
+fn safe_future_into_py_cancels_rust_work() {
+    let _python = crate::test_support::init_python_test();
+    Python::attach(|py| {
+        let event_loop = test_loop(py, false);
+        let locals = pyo3_async_runtimes::TaskLocals::new(event_loop.clone());
+        let (started_tx, started_rx) = mpsc::sync_channel(1);
+        let (dropped_tx, dropped_rx) = mpsc::channel();
+        let future = pyo3_async_runtimes::tokio::get_runtime()
+            .block_on(pyo3_async_runtimes::tokio::scope(locals, async move {
+                Python::attach(|py| {
+                    safe_future_into_py(py, async move {
+                        started_tx.send(()).unwrap();
+                        let _cancellation_signal = CancellationSignal(dropped_tx);
+                        std::future::pending::<PyResult<Py<PyAny>>>().await
+                    })
+                    .map(Bound::unbind)
+                })
+            }))
+            .unwrap();
+
+        assert!(started_rx.recv_timeout(Duration::from_secs(1)).is_ok());
+        future.bind(py).call_method0("cancel").unwrap();
+        assert!(dropped_rx.recv_timeout(Duration::from_secs(1)).is_ok());
+    });
+}
+
+#[test]
+fn safe_future_into_py_skips_completion_on_closed_loop() {
+    let _python = crate::test_support::init_python_test();
+    Python::attach(|py| {
+        let event_loop = test_loop(py, true);
+        let locals = pyo3_async_runtimes::TaskLocals::new(event_loop.clone());
+        let _future = pyo3_async_runtimes::tokio::get_runtime()
+            .block_on(pyo3_async_runtimes::tokio::scope(locals, async move {
+                Python::attach(|py| {
+                    safe_future_into_py(py, async move { Python::attach(|py| Ok(py.None())) })
+                        .map(Bound::unbind)
+                })
+            }))
+            .unwrap();
+
+        assert!(
+            event_loop
+                .getattr("closed_checked")
+                .unwrap()
+                .call_method1("wait", (1.0,))
+                .unwrap()
+                .is_truthy()
+                .unwrap()
+        );
+        assert!(
+            !event_loop
+                .getattr("completion_scheduled")
+                .unwrap()
+                .is_truthy()
+                .unwrap()
+        );
+    });
+}
+
 #[test]
 fn py_api_helpers_and_scope_lifecycle_round_trip() {
     let _python = crate::test_support::init_python_test();
diff --git a/crates/types/Cargo.toml b/crates/types/Cargo.toml
index b967cf972..c3de72a49 100644
--- a/crates/types/Cargo.toml
+++ b/crates/types/Cargo.toml
@@ -18,7 +18,7 @@ schema = ["dep:schemars"]
 
 [dependencies]
 bitflags = { version = "2", features = ["serde"] }
-chrono = { version = "0.4", features = ["serde"] }
+chrono = { version = "0.4", default-features = false, features = ["std", "serde", "now"] }
 schemars = { version = "0.8", optional = true }
 serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"