From 2a4f2aa718294d31af177a3a81e24edf2b4c948a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 19 Sep 2026 14:00:56 -0400 Subject: [PATCH] Test the server's attach to the container Petri created A server scenario runs a command workflow on Docker through Petri's plugin (skipped without the plugin or a daemon), then reaches the container without Petri: the sandbox tab describes it under its `petri.run` label, Run Files writes, lists and reads a file in its workspace after starting the stopped container, a preview URL opens to a port in it, and an Ask Fabro turn runs against it through the OpenAI twin. A unit test attaches through the ownership seam with a scripted provider: the run's own container attaches, another run's and one that carries only Fabro's retired `sh.fabro.*` labels are refused as not owned. Co-Authored-By: Claude Fable 5.1 --- lib/apps/fabro-server/src/sandbox_access.rs | 79 ++++++- .../fabro-server/tests/it/scenario/petri.rs | 217 ++++++++++++++++++ 2 files changed, 288 insertions(+), 8 deletions(-) diff --git a/lib/apps/fabro-server/src/sandbox_access.rs b/lib/apps/fabro-server/src/sandbox_access.rs index 52798ad91..2fde91ee3 100644 --- a/lib/apps/fabro-server/src/sandbox_access.rs +++ b/lib/apps/fabro-server/src/sandbox_access.rs @@ -288,13 +288,19 @@ pub(crate) async fn run_provider( run_id: RunId, ) -> Result, ConnectError> { let provider = connect_provider(kind, access).await?; + Ok(scope_to_run(kind, provider, run_id)) +} + +/// `provider` narrowed to the sandboxes of `run_id`; see [`run_provider`]. +fn scope_to_run( + kind: &SandboxProviderKind, + provider: Arc, + run_id: RunId, +) -> Arc { if kind.bundled() == Some(BundledProvider::Local) { - return Ok(provider); + return provider; } - Ok(Arc::new(OwnedProvider::new( - provider, - run_ownership(run_id), - ))) + Arc::new(OwnedProvider::new(provider, run_ownership(run_id))) } /// Attaches to a run's sandbox from its record, through the record's @@ -309,12 +315,23 @@ pub(crate) async fn attach_run_sandbox( access: &ProviderAccess, record: &RunSandboxInstance, run_id: RunId, +) -> anyhow::Result> { + let provider = connect_provider(&record.provider, access) + .await + .with_context(|| format!("Failed to connect to the {} provider", record.provider))?; + attach_run_sandbox_on(provider, record, run_id).await +} + +/// [`attach_run_sandbox`] on an already connected, unscoped `provider` for +/// the record's kind: the run scope is applied here. +async fn attach_run_sandbox_on( + provider: Arc, + record: &RunSandboxInstance, + run_id: RunId, ) -> anyhow::Result> { let kind = &record.provider; let sandbox_id = &record.runtime.id; - let provider = run_provider(kind, access, run_id) - .await - .with_context(|| format!("Failed to connect to the {kind} provider"))?; + let provider = scope_to_run(kind, provider, run_id); let id = SandboxId::try_new(sandbox_id).with_context(|| format!("Invalid {kind} sandbox id"))?; match provider.attach(&id, None).await { @@ -880,6 +897,52 @@ mod tests { } } + #[tokio::test] + async fn attach_run_sandbox_keys_ownership_on_petris_run_label() { + let run_id = RunId::new(); + let other_run = RunId::new(); + // Petri's own sandbox for the run, another run's, and one that carries + // only Fabro's retired labels. + let fabro_labelled = Arc::new( + ScriptedSandbox::with_id_and_working_dir("fabro-era", "/workspace") + .state(SandboxState::Running) + .label("sh.fabro.managed", "true") + .label("sh.fabro.run_id", run_id.to_string()), + ); + let provider = scripted_provider("docker", vec![ + petri_scripted_sandbox("petri-container", &run_id.to_string()), + petri_scripted_sandbox("other-runs-container", &other_run.to_string()), + fabro_labelled, + ]); + + let attached = attach_run_sandbox_on( + Arc::clone(&provider), + &record(SandboxProviderKind::DOCKER, "petri-container"), + run_id, + ) + .await + .expect("the container Petri labelled with the run attaches"); + assert_eq!(attached.id().as_str(), "petri-container"); + + for foreign in ["other-runs-container", "fabro-era"] { + let error = attach_run_sandbox_on( + Arc::clone(&provider), + &record(SandboxProviderKind::DOCKER, foreign), + run_id, + ) + .await + .err() + .unwrap_or_else(|| panic!("{foreign} does not carry petri.run={run_id}")); + assert!( + error.chain().any(|cause| matches!( + cause.downcast_ref::(), + Some(DriverError::NotOwned { .. }) + )), + "{foreign}: {error:#}" + ); + } + } + #[tokio::test] async fn daytona_requires_explicit_credentials() { let error = connect_provider(&SandboxProviderKind::DAYTONA, &ProviderAccess::default()) diff --git a/lib/apps/fabro-server/tests/it/scenario/petri.rs b/lib/apps/fabro-server/tests/it/scenario/petri.rs index 2f93f82c3..651b885cc 100644 --- a/lib/apps/fabro-server/tests/it/scenario/petri.rs +++ b/lib/apps/fabro-server/tests/it/scenario/petri.rs @@ -870,6 +870,223 @@ async fn a_runs_projection_carries_its_docker_sandbox_instance() { } } +/// The server reaches the container Petri created, without Petri: the +/// sandbox tab describes it, Run Files lists and round-trips a file in +/// its workspace, a preview URL is opened to a port in it, and an Ask +/// Fabro session runs its turn against it. The container carries Petri's +/// `petri.run` label and none of Fabro's own, so an attach scoped to +/// Fabro's retired labels would refuse it; the server's ownership is the +/// run label. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn the_server_attaches_to_the_container_petri_created() { + if docker_plugin().is_none() { + return; + } + let twin = twin_openai().await; + let namespace = format!("{}::{}", module_path!(), line!()); + TwinScenarios::new(&namespace) + .scenario(TwinScenario::responses(OPENAI_MODEL).text("The workspace is /workspace.")) + .load(twin) + .await; + let settings = settings_from_toml("_version = 1\n\n[run.environment]\nid = \"docker\"\n"); + let state = TestAppStateBuilder::new() + .runtime_settings(settings.server_settings, settings.manifest_run_defaults) + .max_concurrent_runs(5) + .in_process_execution() + .llm_overlay(llm_overlay_with_provider_base_url( + "openai", + twin.base_url.clone(), + )) + .vault_entries([(EnvVars::OPENAI_API_KEY, namespace.clone())]) + .build(); + let app = test_app_with_scheduler(Arc::clone(&state)); + create_docker_environment(&app, "docker", CATALOG_IMAGE).await; + + let version_id = register_version(&app, &[ + ("workflow.fabro", COMMAND_DOT), + ("workflow.toml", PLAIN_SETTINGS), + ]) + .await; + let intent = serde_json::json!({ + "workflow_version_id": version_id, + "target": {"kind": "none"}, + "environment_id": "docker", + "args": {}, + }); + let run_id = create_and_start_run_from_intent(&app, intent).await; + let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + assert_eq!( + status, + "succeeded", + "run: {}", + run_json(&app, &run_id).await + ); + let projection = settled_state(&state, &app, &run_id).await; + let container = projection["sandbox"]["instance"]["runtime"]["id"] + .as_str() + .expect("the container id") + .to_string(); + + // The sandbox tab: the record and the daemon's status for the container. + let details = response_json( + app.clone() + .oneshot(get(&format!("/runs/{run_id}/sandbox"))) + .await + .expect("sandbox details route"), + StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/sandbox"), + ) + .await; + assert_eq!(details["sandbox"]["provider"], "docker", "{details}"); + assert_eq!(details["sandbox"]["runtime"]["id"], container, "{details}"); + assert_eq!(details["status"]["id"], container, "{details}"); + assert_eq!( + details["status"]["labels"]["petri.run"], run_id, + "the container carries Petri's run label: {details}" + ); + assert!( + details["status"]["labels"] + .as_object() + .is_some_and(|labels| !labels.contains_key("sh.fabro.managed")), + "Petri stamps no Fabro label: {details}" + ); + + // Run Files: a file written into the workspace is listed and read back, + // which starts the container Petri stopped at the run's end. + let put = Request::builder() + .method("PUT") + .uri(api(&format!( + "/runs/{run_id}/sandbox/file?path=/workspace/from-fabro.txt" + ))) + .header("content-type", "application/octet-stream") + .body(Body::from("written through the server")) + .expect("file upload request should build"); + crate::helpers::response_text( + app.clone().oneshot(put).await.expect("file upload routes"), + StatusCode::NO_CONTENT, + format!("PUT /api/v1/runs/{run_id}/sandbox/file"), + ) + .await; + let listing = response_json( + app.clone() + .oneshot(get(&format!( + "/runs/{run_id}/sandbox/files?path=/workspace" + ))) + .await + .expect("sandbox files route"), + StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/sandbox/files"), + ) + .await; + let names: Vec<&str> = listing["data"] + .as_array() + .expect("the listing's entries") + .iter() + .filter_map(|entry| entry["name"].as_str()) + .collect(); + assert!( + names.contains(&"from-fabro.txt"), + "the workspace lists the file: {names:?}" + ); + let content = crate::helpers::response_text( + app.clone() + .oneshot(get(&format!( + "/runs/{run_id}/sandbox/file?path=/workspace/from-fabro.txt" + ))) + .await + .expect("file download route"), + StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/sandbox/file"), + ) + .await; + assert_eq!(content, "written through the server"); + + // A preview URL to a port in the container, through the driver's + // forward. + let preview = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/preview"))) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ "port": 8080, "expires_in_secs": 60, "signed": false }).to_string(), + )) + .expect("preview request should build"); + let preview = response_json( + app.clone().oneshot(preview).await.expect("preview route"), + StatusCode::CREATED, + format!("POST /api/v1/runs/{run_id}/preview"), + ) + .await; + assert!( + preview["url"] + .as_str() + .is_some_and(|url| url.starts_with("http://")), + "{preview}" + ); + + // Ask Fabro: the session reconnects to the container for its turn and + // the turn completes on the model's answer. + let session = Request::builder() + .method("POST") + .uri(api(&format!("/runs/{run_id}/sessions"))) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ "title": "Ask Fabro", "model": OPENAI_MODEL }).to_string(), + )) + .expect("session request should build"); + let session = response_json( + app.clone().oneshot(session).await.expect("session route"), + StatusCode::CREATED, + format!("POST /api/v1/runs/{run_id}/sessions"), + ) + .await; + let session_id = session["id"].as_str().expect("the session id"); + let turn = Request::builder() + .method("POST") + .uri(api(&format!("/sessions/{session_id}/turns"))) + .header("content-type", "application/json") + .body(Body::from(r#"{"input":"Where is the workspace?"}"#)) + .expect("submit-turn request should build"); + let stream = crate::helpers::response_text( + app.clone().oneshot(turn).await.expect("turn route"), + StatusCode::OK, + format!("POST /api/v1/sessions/{session_id}/turns"), + ) + .await; + let events: Vec = stream + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .map(|data| serde_json::from_str(data).expect("session event data should be JSON")) + .collect(); + let failed = events + .iter() + .find(|event| event["event"] == "run.session.turn.failed"); + assert!( + failed.is_none(), + "the turn reached the container: {failed:?}" + ); + assert!( + events + .iter() + .any(|event| event["event"] == "run.session.turn.succeeded"), + "the turn completed: {events:?}" + ); + + let _ = Command::new("docker") + .args(["rm", "-f", &container]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); +} + +fn get(path: &str) -> Request { + Request::builder() + .method("GET") + .uri(api(path)) + .body(Body::empty()) + .expect("GET request should build") +} + /// The image the server's `docker-small` environment names in the tests /// below: a runner image with `git` for the checkpoint commit, and not the /// plugin's default, so the container proves the catalog's image reached it.