diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index a100aa11d1..95e33573e5 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -4420,8 +4420,42 @@ impl ComputeRuntime { } let sandbox_name = current.object_name().to_string(); - self.validate_isolation_backend_binding(¤t) - .map_err(|error| error.to_string())?; + if let Err(error) = self.validate_isolation_backend_binding(¤t) { + // Only an explicit stop and start can relaunch the sandbox once the + // gateway selection matches again, so leave the restart scan. + let message = error.message().to_string(); + match self + .store + .update_message_cas::( + sandbox_id, + sandbox_resource_version(¤t), + |sandbox| { + let status = sandbox.status.get_or_insert_with(Default::default); + set_next_restart_at_ms(status, 0); + sandbox.set_phase(SandboxPhase::Error as i32); + upsert_ready_condition( + &mut sandbox.status, + SandboxCondition { + r#type: "Ready".to_string(), + status: "False".to_string(), + reason: "IsolationBackendMismatch".to_string(), + message: message.clone(), + transition_time: None, + }, + ); + }, + ) + .await + { + Ok(updated) => { + self.sandbox_index.update_from_sandbox(&updated); + self.sandbox_watch_bus.notify(sandbox_id); + } + Err(crate::persistence::PersistenceError::Conflict { .. }) => return Ok(()), + Err(err) => return Err(err.to_string()), + } + return Err(error.to_string()); + } let already_claimed = current .status .as_ref() @@ -10526,6 +10560,64 @@ mod tests { assert!(next_restart_at_ms(&stored.status.unwrap()) > openshell_core::time::now_ms()); } + #[tokio::test] + async fn restart_scan_reports_isolation_backend_mismatch_once() { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver.clone()).await; + let mut sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Starting); + sandbox + .metadata + .as_mut() + .unwrap() + .annotations + .insert(ISOLATION_BACKEND_ANNOTATION.into(), "other-backend".into()); + sandbox.status = Some(SandboxStatus { + phase: SandboxPhase::Starting as i32, + exit_code: Some(9), + restart_count: 1, + next_restart_time: restart_timestamp(openshell_core::time::now_ms() - 1), + ..Default::default() + }); + runtime.store.put_message(&sandbox).await.unwrap(); + + runtime.restart_due_sandboxes().await.unwrap(); + let failed = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + runtime.restart_due_sandboxes().await.unwrap(); + + assert_eq!(driver.stop_calls(), 0); + assert_eq!(driver.start_calls(), 0); + assert_eq!(failed.phase(), SandboxPhase::Error as i32); + assert!( + failed + .status + .as_ref() + .unwrap() + .conditions + .iter() + .any(|condition| { + condition.r#type == "Ready" + && condition.reason == "IsolationBackendMismatch" + && condition.message.contains("restore the gateway selection") + }) + ); + let rescanned = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + sandbox_resource_version(&rescanned), + sandbox_resource_version(&failed), + "the next scan must leave a mismatched sandbox alone" + ); + } + #[tokio::test] async fn concurrent_stop_transition_prevents_replacement_start() { let driver = ControlledDriver::new(); @@ -17277,6 +17369,17 @@ endpoint = { kind = "unix", socket_path = "/run/agent-substrate/isolation.sock" driver.calls().is_empty(), "another replica must not launch a differently bound resource" ); + assert_eq!( + runtime + .store + .get_message::("external-start") + .await + .unwrap() + .unwrap() + .phase(), + SandboxPhase::Starting as i32, + "another replica must not fail a launch it cannot run" + ); } let mut capabilities = driver .get_capabilities(Request::new(GetCapabilitiesRequest::default())) @@ -19436,6 +19539,79 @@ endpoint = { kind = "unix", socket_path = "/run/agent-substrate/isolation.sock" assert_eq!(driver.stop_calls(), 1); } + #[tokio::test] + async fn preparation_timeout_reports_isolation_backend_mismatch() { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver.clone()).await; + let mut sandbox = sandbox_record("sb-prepare", "prepare", SandboxPhase::Starting); + sandbox + .metadata + .as_mut() + .unwrap() + .annotations + .insert(ISOLATION_BACKEND_ANNOTATION.into(), "other-backend".into()); + sandbox.status.as_mut().unwrap().provisioning = + Some(provisioning_deadline::new_preparation_record(0, 1800)); + runtime.store.put_message(&sandbox).await.unwrap(); + let sandbox = runtime + .store + .get_message::("sb-prepare") + .await + .unwrap() + .unwrap(); + let gate = runtime.lifecycle_gates.lock_for("sb-prepare").await; + let _sandbox_guard = runtime.lock_sandbox_for_lifecycle(&gate).await; + let expired = runtime + .claim_provisioning_timeout(&sandbox, 1_800_000) + .await + .unwrap() + .unwrap(); + assert_eq!(expired.phase(), i32::from(SandboxPhase::Error)); + assert!( + expired + .status + .as_ref() + .unwrap() + .conditions + .iter() + .any(|condition| { + condition.reason == "IsolationBackendMismatch" + && condition.message.contains("restore the gateway selection") + }) + ); + + // A registered supervisor keeps the admission diagnostic. + let mut admitted = sandbox_record("sb-admission", "admission", SandboxPhase::Starting); + admitted + .metadata + .as_mut() + .unwrap() + .annotations + .insert(ISOLATION_BACKEND_ANNOTATION.into(), "other-backend".into()); + admitted.status.as_mut().unwrap().provisioning = Some(provisioning_deadline::new_record(0)); + runtime.store.put_message(&admitted).await.unwrap(); + let admitted = runtime + .store + .get_message::("sb-admission") + .await + .unwrap() + .unwrap(); + let expired = runtime + .claim_provisioning_timeout(&admitted, 300_000) + .await + .unwrap() + .unwrap(); + assert!( + expired + .status + .as_ref() + .unwrap() + .conditions + .iter() + .any(|condition| condition.reason == "ProvisioningTimedOut") + ); + } + #[tokio::test] async fn preparation_timeout_retains_reason_and_uses_existing_cleanup() { let driver = ControlledDriver::new(); diff --git a/crates/openshell-server/src/compute/provisioning_deadline.rs b/crates/openshell-server/src/compute/provisioning_deadline.rs index 7b7608d38d..9385e1c836 100644 --- a/crates/openshell-server/src/compute/provisioning_deadline.rs +++ b/crates/openshell-server/src/compute/provisioning_deadline.rs @@ -552,6 +552,12 @@ impl super::ComputeRuntime { }; let mut deadline = ProvisioningDeadline::from_record(record)?; let preparation_expired = deadline.is_preparing(); + // A supervisor that never registered may be waiting on a launch bound + // to another isolation backend; report that instead of a timeout. + let backend_mismatch = preparation_expired + .then(|| self.validate_isolation_backend_binding(current).err()) + .flatten() + .map(|error| error.message().to_string()); if !deadline.expire(&record.attempt_id, &record.configuration_change_id, now_ms) { return Ok(None); } @@ -573,22 +579,26 @@ impl super::ComputeRuntime { .configuration_admission .as_ref() .map_or("", |admission| admission.error.as_str()); - let message = if preparation_expired { - "Image preparation or initial supervisor startup exceeded its absolute deadline".to_string() - } else if diagnostic.is_empty() { - "Provisioning repair window expired after 300 seconds".to_string() - } else { - format!( - "Provisioning repair window expired after 300 seconds: {diagnostic}" - ) - }; + let message = backend_mismatch.clone().unwrap_or_else(|| { + if preparation_expired { + "Image preparation or initial supervisor startup exceeded its absolute deadline".to_string() + } else if diagnostic.is_empty() { + "Provisioning repair window expired after 300 seconds".to_string() + } else { + format!( + "Provisioning repair window expired after 300 seconds: {diagnostic}" + ) + } + }); status .conditions .retain(|condition| condition.r#type != "Ready"); status.conditions.push(SandboxCondition { r#type: "Ready".into(), status: "False".into(), - reason: if preparation_expired { + reason: if backend_mismatch.is_some() { + "IsolationBackendMismatch" + } else if preparation_expired { "ImagePreparationTimedOut" } else { "ProvisioningTimedOut" diff --git a/e2e/rust/src/harness/sandbox.rs b/e2e/rust/src/harness/sandbox.rs index 0d6ff621dd..8aa4ee176b 100644 --- a/e2e/rust/src/harness/sandbox.rs +++ b/e2e/rust/src/harness/sandbox.rs @@ -676,7 +676,6 @@ impl SandboxGuard { if self.cleaned_up { return; } - self.cleaned_up = true; // Kill the background child process if present. if let Some(ref mut child) = self.child { @@ -690,6 +689,8 @@ impl SandboxGuard { cmd.stdout(Stdio::null()).stderr(Stdio::null()); let _ = cmd.status().await; + // Cancellation leaves Drop responsible until the delete command returns. + self.cleaned_up = true; } } @@ -722,6 +723,79 @@ impl Drop for SandboxGuard { mod tests { use super::{has_explicit_sandbox_name, has_explicit_sandbox_workload}; + #[cfg(unix)] + #[test] + fn cancelled_cleanup_finishes_deletion_before_drop_returns() { + use std::os::unix::fs::PermissionsExt; + use std::time::Duration; + + const CHILD_DIRECTORY: &str = "OPENSHELL_E2E_TEST_CLEANUP_CANCEL_DIR"; + if let Some(directory) = std::env::var_os(CHILD_DIRECTORY) { + let directory = std::path::PathBuf::from(directory); + let mut guard = super::SandboxGuard::manage_existing("cleanup-cancellation".into()); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let mut cleanup = std::pin::pin!(guard.cleanup()); + tokio::time::timeout(Duration::from_secs(5), async { + while !directory.join("started").exists() { + tokio::select! { + () = &mut cleanup => panic!("delete must remain pending until cancellation"), + () = tokio::time::sleep(Duration::from_millis(10)) => {} + } + } + }) + .await + .expect("delete command did not start"); + }); + drop(guard); + assert!( + directory.join("completed").exists(), + "cancelled cleanup must complete deletion during Drop" + ); + return; + } + + // Isolate the binary override from tests running in the parent process. + let directory = tempfile::tempdir().unwrap(); + let binary = directory.path().join("openshell"); + std::fs::write( + &binary, + r#"#!/bin/sh +set -eu +test "$1" = sandbox +test "$2" = delete +test "$3" = cleanup-cancellation +if [ -f "$OPENSHELL_E2E_TEST_CLEANUP_CANCEL_DIR/started" ]; then + : > "$OPENSHELL_E2E_TEST_CLEANUP_CANCEL_DIR/completed" + exit 0 +fi +: > "$OPENSHELL_E2E_TEST_CLEANUP_CANCEL_DIR/started" +exec sleep 60 +"#, + ) + .unwrap(); + std::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o700)).unwrap(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "harness::sandbox::tests::cancelled_cleanup_finishes_deletion_before_drop_returns", + "--nocapture", + ]) + .env(CHILD_DIRECTORY, directory.path()) + .env("OPENSHELL_BIN", &binary) + .output() + .unwrap(); + assert!( + output.status.success() && directory.path().join("completed").is_file(), + "cleanup cancellation regression failed:\n{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + #[test] fn detects_explicit_sandbox_names() { assert!(has_explicit_sandbox_name(&["--name", "example"])); diff --git a/e2e/rust/tests/credential_gating.rs b/e2e/rust/tests/credential_gating.rs index bf0d6ae9b5..f210c780f9 100644 --- a/e2e/rust/tests/credential_gating.rs +++ b/e2e/rust/tests/credential_gating.rs @@ -6,12 +6,14 @@ //! E2E coverage for credentialed endpoint admission and REST body backstops. use std::io::Write; +use std::panic::{AssertUnwindSafe, resume_unwind}; use std::process::Stdio; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use base64::Engine as _; +use futures_util::FutureExt; use openshell_e2e::harness::binary::openshell_cmd; use openshell_e2e::harness::sandbox::SandboxGuard; use sha1::{Digest, Sha1}; @@ -254,7 +256,7 @@ fn write_policy( } }; let policy = format!( - r#"version: 1 + r"version: 1 filesystem_policy: include_workdir: true read_only: [/usr, /lib, /proc, /dev/urandom, /app, /etc, /var/log] @@ -274,7 +276,7 @@ network_policies: - path: /usr/bin/python* - path: /usr/local/bin/python* - path: /sandbox/.uv/python/*/bin/python* -"#, +", ); file.write_all(policy.as_bytes()) .map_err(|error| format!("write policy: {error}"))?; @@ -895,46 +897,62 @@ async fn credentialed_endpoint_gates_work_end_to_end() { .await .expect("install credentialed provider"); - let result = async { - assert_rest_body_preserves_placeholder(&server).await?; - assert_websocket_binary_denied(&websocket_server).await - } - .await; + eprintln!("credential gating: profile REST and WebSocket scenarios"); + let (rest_result, websocket_result) = tokio::join!( + AssertUnwindSafe(assert_rest_body_preserves_placeholder(&server)).catch_unwind(), + AssertUnwindSafe(assert_websocket_binary_denied(&websocket_server)).catch_unwind(), + ); + eprintln!("credential gating: profile provider cleanup"); cleanup_provider_resources().await; - result.expect("credential gating E2E"); + rest_result + .unwrap_or_else(|panic| resume_unwind(panic)) + .and_then(|()| websocket_result.unwrap_or_else(|panic| resume_unwind(panic))) + .expect("credential gating E2E"); install_endpointless_provider() .await .expect("install endpointless provider"); - let endpointless_result = async { - assert_gateway_admission(server.port, CredentialSource::PolicyBinding).await?; - let literal = run_body_sandbox( + eprintln!("credential gating: endpointless admission, REST bodies and live environment"); + let (admission_result, body_result, environment_result) = tokio::join!( + AssertUnwindSafe(assert_gateway_admission( server.port, - EndpointMode::RestBody { rewrite: false }, CredentialSource::PolicyBinding, - ) - .await?; - assert!(literal.contains("BODY_TEXT")); - let rewritten = run_body_sandbox( - server.port, - EndpointMode::RestBody { rewrite: true }, - CredentialSource::PolicyBinding, - ) - .await?; - assert!(rewritten.contains("BODY_REWRITTEN")); - assert!(!rewritten.contains(TEST_SECRET)); - assert!(!rewritten.contains(PLACEHOLDER_PREFIX)); - let observations = server.wait_for_observations(3).await; - assert_eq!(observations.len(), 3, "observations: {observations:?}"); - assert!(observations[1].saw_placeholder); - assert!(!observations[1].saw_secret); - assert!(!observations[2].saw_placeholder); - assert!(observations[2].saw_secret); - assert_endpointless_provider_env_live_update(server.port).await?; - Ok::<(), String>(()) - } - .await; + )) + .catch_unwind(), + AssertUnwindSafe(async { + let literal = run_body_sandbox( + server.port, + EndpointMode::RestBody { rewrite: false }, + CredentialSource::PolicyBinding, + ) + .await?; + assert!(literal.contains("BODY_TEXT")); + let rewritten = run_body_sandbox( + server.port, + EndpointMode::RestBody { rewrite: true }, + CredentialSource::PolicyBinding, + ) + .await?; + assert!(rewritten.contains("BODY_REWRITTEN")); + assert!(!rewritten.contains(TEST_SECRET)); + assert!(!rewritten.contains(PLACEHOLDER_PREFIX)); + let observations = server.wait_for_observations(3).await; + assert_eq!(observations.len(), 3, "observations: {observations:?}"); + assert!(observations[1].saw_placeholder); + assert!(!observations[1].saw_secret); + assert!(!observations[2].saw_placeholder); + assert!(observations[2].saw_secret); + Ok::<(), String>(()) + }) + .catch_unwind(), + AssertUnwindSafe(assert_endpointless_provider_env_live_update(server.port)).catch_unwind(), + ); + eprintln!("credential gating: endpointless provider cleanup"); cleanup_provider_resources().await; - endpointless_result.expect("endpointless provider credential gating E2E"); + admission_result + .unwrap_or_else(|panic| resume_unwind(panic)) + .and_then(|()| body_result.unwrap_or_else(|panic| resume_unwind(panic))) + .and_then(|()| environment_result.unwrap_or_else(|panic| resume_unwind(panic))) + .expect("endpointless provider credential gating E2E"); }