From c3afcc74913ea3311f18b032234dd2cd80b69454 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 10 Jul 2026 01:32:26 +0900 Subject: [PATCH] runtime: fetch profile archives for remote workers --- crates/worker-runtime/src/catalog.rs | 57 +++- crates/worker-runtime/src/http_server.rs | 42 ++- crates/worker-runtime/src/runtime.rs | 120 ++++---- crates/worker-runtime/src/worker_backend.rs | 211 ++++++++------ crates/workspace-server/src/hosts.rs | 275 ++++++++++++------ .../workspace-server/src/resource_broker.rs | 12 + crates/workspace-server/src/server.rs | 87 +++++- 7 files changed, 524 insertions(+), 280 deletions(-) diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 1026eaa1..7837a9ef 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -1,10 +1,11 @@ use crate::execution::WorkerExecutionStatus; use crate::identity::{RuntimeId, WorkerId, WorkerRef}; use crate::interaction::WorkerInput; +use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef}; use serde::{Deserialize, Serialize}; use std::path::PathBuf; -/// Profile selector boundary. This is a selector, not a resolved config bundle. +/// Profile selector boundary. This is a selector, not a resolved runtime config. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "kind", content = "value", rename_all = "snake_case")] pub enum ProfileSelector { @@ -19,7 +20,36 @@ impl Default for ProfileSelector { } } -/// Backend-synced config bundle reference used during Worker creation. +/// Runtime fetch/caching metadata for a Backend-authored Decodal profile source archive. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProfileSourceArchiveHttpRef { + pub url: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub etag: Option, + pub archive: ProfileSourceArchiveRef, +} + +/// Profile source material available to a Runtime during Worker creation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ProfileSourceArchiveSource { + /// Backend-internal embedded runtimes may receive already-built archive bytes. + Embedded { archive: ProfileSourceArchive }, + /// Standalone runtimes fetch/cache the tar archive over HTTP. + Http { + location: ProfileSourceArchiveHttpRef, + }, +} + +impl ProfileSourceArchiveSource { + pub fn reference(&self) -> ProfileSourceArchiveRef { + match self { + Self::Embedded { archive } => archive.reference.clone(), + Self::Http { location } => location.archive.clone(), + } + } +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ConfigBundleRef { pub id: String, @@ -139,16 +169,17 @@ pub struct WorkingDirectoryStatus { /// /// Browser/product launch semantics are resolved by a backend before this /// request is built. The request contains only durable Runtime identity inputs: -/// a backend-decided profile selector, a previously synced ConfigBundle identity, -/// optional initial user input that is committed in the same transaction as -/// Worker catalog/transcript persistence, and an optional working directory -/// request that preserves RepositoryPoint-style semantics for runtime-side -/// materialization. Browser-facing status for materialized working directories is -/// summarized without exposing raw host paths. +/// a backend-decided profile selector, the Decodal profile source archive source +/// used to resolve that selector, optional initial user input committed with the +/// Worker catalog/transcript persistence, and an optional Runtime-owned working +/// directory binding. Browser-facing status for materialized working directories +/// is summarized without exposing raw host paths. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct CreateWorkerRequest { pub profile: ProfileSelector, - pub config_bundle: ConfigBundleRef, + pub profile_source: ProfileSourceArchiveSource, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_bundle: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub initial_input: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -181,7 +212,9 @@ pub struct WorkerSummary { pub status: WorkerStatus, pub execution: WorkerExecutionStatus, pub profile: ProfileSelector, - pub config_bundle: ConfigBundleRef, + pub profile_source: ProfileSourceArchiveRef, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_bundle: Option, pub transcript_len: usize, pub last_event_id: u64, } @@ -195,7 +228,9 @@ pub struct WorkerDetail { pub status: WorkerStatus, pub execution: WorkerExecutionStatus, pub profile: ProfileSelector, - pub config_bundle: ConfigBundleRef, + pub profile_source: ProfileSourceArchiveRef, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub config_bundle: Option, pub transcript_len: usize, pub last_event_id: u64, } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index ad746596..d03b1a44 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -933,10 +933,27 @@ mod tests { let bundle = test_bundle(profile.clone()); CreateWorkerRequest { profile, - config_bundle: ConfigBundleRef { + profile_source: crate::catalog::ProfileSourceArchiveSource::Http { + location: crate::catalog::ProfileSourceArchiveHttpRef { + url: "http://127.0.0.1/profile-source.tar".to_string(), + etag: None, + archive: crate::profile_archive::ProfileSourceArchiveRef { + id: "test-profile-source".to_string(), + digest: "test-digest".to_string(), + size_bytes: 0, + source_graph: crate::profile_archive::ProfileSourceGraphSummary { + source_count: 0, + total_source_bytes: 0, + entrypoints: std::collections::BTreeMap::new(), + import_count: 0, + }, + }, + }, + }, + config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id, digest: bundle.metadata.digest, - }, + }), initial_input: None, working_directory_request: None, working_directory: None, @@ -1220,10 +1237,27 @@ mod ws_tests { let bundle = ws_test_bundle(ProfileSelector::RuntimeDefault); CreateWorkerRequest { profile: ProfileSelector::RuntimeDefault, - config_bundle: ConfigBundleRef { + profile_source: crate::catalog::ProfileSourceArchiveSource::Http { + location: crate::catalog::ProfileSourceArchiveHttpRef { + url: "http://127.0.0.1/profile-source.tar".to_string(), + etag: None, + archive: crate::profile_archive::ProfileSourceArchiveRef { + id: "test-profile-source".to_string(), + digest: "test-digest".to_string(), + size_bytes: 0, + source_graph: crate::profile_archive::ProfileSourceGraphSummary { + source_count: 0, + total_source_bytes: 0, + entrypoints: std::collections::BTreeMap::new(), + import_count: 0, + }, + }, + }, + }, + config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id, digest: bundle.metadata.digest, - }, + }), initial_input: None, working_directory_request: None, working_directory: None, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 5875ba9e..7e63998c 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1,6 +1,6 @@ use crate::catalog::{ - ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck, - WorkerStatus, WorkerSummary, WorkingDirectoryRequest, + ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus, + WorkerSummary, WorkingDirectoryRequest, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, }; use crate::config_bundle::{ @@ -308,7 +308,7 @@ impl Runtime { .map_err(|diagnostic| RuntimeError::InvalidRequest(diagnostic.to_string())) } - /// Create a Worker through the canonical ConfigBundle + execution backend path. + /// Create a Worker through the canonical profile-source + execution backend path. pub fn create_worker( &self, request: CreateWorkerRequest, @@ -318,13 +318,6 @@ impl Runtime { state.ensure_running()?; validate_create_worker_request(&request)?; state.validate_worker_config_boundary(&request)?; - let config_bundle = state - .config_bundles - .get(&request.config_bundle.id) - .cloned() - .ok_or_else(|| RuntimeError::ConfigBundleMissing { - bundle_id: request.config_bundle.id.clone(), - })?; let backend = state.execution_backend.clone().ok_or_else(|| { RuntimeError::ExecutionBackendUnavailable { message: "worker creation requires an execution backend".to_string(), @@ -370,7 +363,7 @@ impl Runtime { request, context: self.execution_context(worker_ref.clone()), working_directory: None, - config_bundle: Some(config_bundle), + config_bundle: None, }; (backend, worker_ref, spawn_request) }; @@ -1317,23 +1310,8 @@ impl RuntimeState { fn validate_worker_config_boundary( &self, - request: &CreateWorkerRequest, + _request: &CreateWorkerRequest, ) -> Result<(), RuntimeError> { - let reference = &request.config_bundle; - let availability = self.check_config_bundle_ref(reference)?; - let bundle = self - .config_bundles - .get(&availability.reference.id) - .ok_or_else(|| RuntimeError::ConfigBundleMissing { - bundle_id: availability.reference.id.clone(), - })?; - if !bundle.contains_profile(&request.profile) { - return Err(RuntimeError::InvalidProfileSelector { - profile: profile_label(&request.profile), - bundle_id: Some(reference.id.clone()), - message: "profile selector is not declared by synced config bundle".to_string(), - }); - } Ok(()) } @@ -1571,6 +1549,7 @@ impl WorkerRecord { status: self.status, execution: self.execution.clone(), profile: self.request.profile.clone(), + profile_source: self.request.profile_source.reference(), config_bundle: self.request.config_bundle.clone(), transcript_len: self.transcript.len(), last_event_id: self.last_event_id, @@ -1585,6 +1564,7 @@ impl WorkerRecord { status: self.status, execution: self.execution.clone(), profile: self.request.profile.clone(), + profile_source: self.request.profile_source.reference(), config_bundle: self.request.config_bundle.clone(), transcript_len: self.transcript.len(), last_event_id: self.last_event_id, @@ -1606,24 +1586,25 @@ impl WorkerRecord { } } -fn profile_label(selector: &ProfileSelector) -> String { - match selector { - ProfileSelector::RuntimeDefault => "runtime_default".to_string(), - ProfileSelector::Builtin(value) => value.clone(), - ProfileSelector::Named(value) => value.clone(), - } -} - fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> { - if request.config_bundle.id.trim().is_empty() { - return Err(RuntimeError::InvalidRequest( - "config_bundle.id must not be empty".to_string(), - )); - } - if request.config_bundle.digest.trim().is_empty() { - return Err(RuntimeError::InvalidRequest( - "config_bundle.digest must not be empty".to_string(), - )); + match &request.profile_source { + crate::catalog::ProfileSourceArchiveSource::Embedded { archive } => { + archive.verify().map_err(|err| { + RuntimeError::InvalidRequest(format!("profile_source archive is invalid: {err}")) + })?; + } + crate::catalog::ProfileSourceArchiveSource::Http { location } => { + if location.url.trim().is_empty() { + return Err(RuntimeError::InvalidRequest( + "profile_source.location.url must not be empty".to_string(), + )); + } + if location.archive.digest.trim().is_empty() { + return Err(RuntimeError::InvalidRequest( + "profile_source.location.archive.digest must not be empty".to_string(), + )); + } + } } if let Some(input) = &request.initial_input { if input.kind != WorkerInputKind::User { @@ -1669,10 +1650,27 @@ mod tests { let bundle = test_bundle_for_profile(profile.clone()); CreateWorkerRequest { profile, - config_bundle: ConfigBundleRef { + profile_source: crate::catalog::ProfileSourceArchiveSource::Http { + location: crate::catalog::ProfileSourceArchiveHttpRef { + url: "http://127.0.0.1/profile-source.tar".to_string(), + etag: None, + archive: crate::profile_archive::ProfileSourceArchiveRef { + id: "test-profile-source".to_string(), + digest: "test-digest".to_string(), + size_bytes: 0, + source_graph: crate::profile_archive::ProfileSourceGraphSummary { + source_count: 0, + total_source_bytes: 0, + entrypoints: BTreeMap::new(), + import_count: 0, + }, + }, + }, + }, + config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id, digest: bundle.metadata.digest, - }, + }), initial_input: None, working_directory_request: None, working_directory: None, @@ -1806,10 +1804,10 @@ mod tests { fn bundled_task_request(objective: &str, bundle: &ConfigBundle) -> CreateWorkerRequest { let mut request = task_request(objective); - request.config_bundle = ConfigBundleRef { + request.config_bundle = Some(ConfigBundleRef { id: bundle.metadata.id.clone(), digest: bundle.metadata.digest.clone(), - }; + }); request } @@ -1820,7 +1818,7 @@ mod tests { assert_eq!(detail.worker_ref.runtime_id, runtime.runtime_id().unwrap()); assert_eq!(detail.status, WorkerStatus::Running); - assert_eq!(detail.config_bundle.id, "bundle-1"); + assert_eq!(detail.config_bundle.as_ref().unwrap().id, "bundle-1"); let list = runtime.list_workers().unwrap(); assert_eq!(list.len(), 1); @@ -1852,7 +1850,7 @@ mod tests { let detail = runtime .create_worker(bundled_task_request("synced", &bundle)) .unwrap(); - assert_eq!(detail.config_bundle, availability.reference); + assert_eq!(detail.config_bundle, Some(availability.reference)); } #[test] @@ -1860,11 +1858,6 @@ mod tests { let runtime = Runtime::new_memory(); let bundle = test_bundle(); - let missing = runtime - .create_worker(bundled_task_request("missing", &bundle)) - .unwrap_err(); - assert!(matches!(missing, RuntimeError::ConfigBundleMissing { .. })); - runtime.store_config_bundle(bundle.clone()).unwrap(); let mismatch = runtime .check_config_bundle(&ConfigBundleRef { @@ -1877,14 +1870,6 @@ mod tests { RuntimeError::ConfigBundleDigestMismatch { .. } )); - let mut bad_profile = bundled_task_request("bad profile", &bundle); - bad_profile.profile = ProfileSelector::Builtin("builtin:reviewer".to_string()); - let invalid_profile = runtime.create_worker(bad_profile).unwrap_err(); - assert!(matches!( - invalid_profile, - RuntimeError::InvalidProfileSelector { .. } - )); - let mut unsupported = test_bundle(); unsupported.declarations.push(ConfigDeclaration { kind: ConfigDeclarationKind::Unsupported, @@ -1947,12 +1932,15 @@ mod tests { } #[test] - fn create_worker_missing_config_bundle_is_rejected_before_backend() { + fn create_worker_without_execution_backend_is_rejected_before_persisting_worker() { let runtime = Runtime::new_memory(); let error = runtime - .create_worker(task_request("missing bundle")) + .create_worker(task_request("missing backend")) .unwrap_err(); - assert!(matches!(error, RuntimeError::ConfigBundleMissing { .. })); + assert!(matches!( + error, + RuntimeError::ExecutionBackendUnavailable { .. } + )); assert!(runtime.list_workers().unwrap().is_empty()); } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 62f693ef..b2372515 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -14,17 +14,16 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; -use crate::catalog::{WorkingDirectoryRequest, WorkingDirectoryStatus}; -use crate::config_bundle::verified_profile_source_archive; +use crate::catalog::{ + ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource, WorkingDirectoryRequest, + WorkingDirectoryStatus, +}; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; use crate::interaction::{WorkerInput, WorkerInputKind}; -use crate::resource::{ - BackendResourceClient, BackendResourceError, ProfileSourceArchiveCache, - build_profile_source_archive_fetch_request, profile_source_archive_from_response, -}; +use crate::resource::{BackendResourceClient, ProfileSourceArchiveCache}; use crate::working_directory::{ WorkingDirectoryBinding, WorkingDirectoryDiagnostic, WorkingDirectoryMaterializer, }; @@ -170,54 +169,103 @@ impl ProfileRuntimeWorkerFactory { } async fn resolve_profile_source_archive( &self, - bundle: &crate::config_bundle::ConfigBundle, - request: &WorkerExecutionSpawnRequest, + source: &ProfileSourceArchiveSource, ) -> Result { - if let Some(archive) = verified_profile_source_archive(bundle) - .map_err(|err| format!("failed to verify profile source archive: {err}"))? - { - return Ok(archive); - } - let handle = bundle - .profile_source_archive_handle - .clone() - .ok_or_else(|| { - format!( - "config bundle {} does not contain a ProfileSourceArchive resource handle", - bundle.metadata.id - ) - })?; - let client = self.resource_client.as_ref().ok_or_else(|| { - format!( - "config bundle {} requires a Backend resource client for profile source archive fetch", - bundle.metadata.id - ) - })?; - let fetch_request = build_profile_source_archive_fetch_request( - handle.clone(), - &request.worker_ref.runtime_id, - Some(&request.worker_ref.worker_id), - ); - let response = client - .fetch_resource(fetch_request) - .await - .map_err(format_backend_resource_error)?; - let fetched_archive = profile_source_archive_from_response(&handle, response) - .map_err(format_backend_resource_error)?; - if let Some(cached) = self.profile_archive_cache.get(&handle.digest) { - return cached + match source { + ProfileSourceArchiveSource::Embedded { archive } => archive .verify() - .map_err(|err| format!("failed to verify cached profile source archive: {err}")); + .map_err(|err| format!("failed to verify embedded profile source archive: {err}")), + ProfileSourceArchiveSource::Http { location } => { + self.fetch_profile_source_archive(location).await + } + } + } + + async fn fetch_profile_source_archive( + &self, + location: &ProfileSourceArchiveHttpRef, + ) -> Result { + if let Some(cached) = self.profile_archive_cache.get(&location.archive.digest) { + let response = + fetch_profile_source_archive_http(location, Some(&location.archive.digest)).await?; + if let Some(fetched) = response { + self.profile_archive_cache.insert(fetched.clone()); + fetched.verify().map_err(|err| { + format!("failed to verify fetched profile source archive: {err}") + }) + } else { + cached + .verify() + .map_err(|err| format!("failed to verify cached profile source archive: {err}")) + } + } else { + let archive = fetch_profile_source_archive_http(location, None) + .await? + .ok_or_else(|| { + "profile source archive HTTP revalidation returned 304 without a cached archive" + .to_string() + })?; + self.profile_archive_cache.insert(archive.clone()); + archive + .verify() + .map_err(|err| format!("failed to verify fetched profile source archive: {err}")) } - self.profile_archive_cache.insert(fetched_archive.clone()); - fetched_archive - .verify() - .map_err(|err| format!("failed to verify fetched profile source archive: {err}")) } } -fn format_backend_resource_error(error: BackendResourceError) -> String { - format!("backend resource fetch failed: {error}") +#[cfg(feature = "http-server")] +async fn fetch_profile_source_archive_http( + location: &ProfileSourceArchiveHttpRef, + cached_digest: Option<&str>, +) -> Result, String> { + let client = reqwest::Client::new(); + let mut request = client.get(&location.url); + if cached_digest == Some(location.archive.digest.as_str()) { + if let Some(etag) = location.etag.as_deref() { + request = request.header(reqwest::header::IF_NONE_MATCH, etag); + } + } + let response = request + .send() + .await + .map_err(|err| format!("failed to fetch profile source archive: {err}"))?; + if response.status() == reqwest::StatusCode::NOT_MODIFIED { + return Ok(None); + } + if !response.status().is_success() { + let status = response.status(); + return Err(format!( + "profile source archive fetch failed with HTTP {status}" + )); + } + let bytes = response + .bytes() + .await + .map_err(|err| format!("failed to read profile source archive response: {err}"))? + .to_vec(); + let archive = crate::profile_archive::ProfileSourceArchive { + reference: location.archive.clone(), + content: bytes, + }; + if archive.content.len() as u64 != archive.reference.size_bytes { + return Err(format!( + "profile source archive size mismatch: expected {}, got {}", + archive.reference.size_bytes, + archive.content.len() + )); + } + Ok(Some(archive)) +} + +#[cfg(not(feature = "http-server"))] +async fn fetch_profile_source_archive_http( + _location: &ProfileSourceArchiveHttpRef, + _cached_digest: Option<&str>, +) -> Result, String> { + Err( + "HTTP profile source archive fetch requires the worker-runtime http-server feature" + .to_string(), + ) } #[async_trait] @@ -238,11 +286,11 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .as_ref() .map(|binding| binding.cwd().to_path_buf()) .unwrap_or_else(|| self.cwd.clone()); - let (mut manifest, loader) = if let Some(bundle) = request.config_bundle.as_ref() { - let selector = profile.as_deref().unwrap_or("builtin:default"); - let archive = self - .resolve_profile_source_archive(bundle, &request) - .await?; + let selector = profile.as_deref().unwrap_or("builtin:default"); + let archive = self + .resolve_profile_source_archive(&request.request.profile_source) + .await?; + let (mut manifest, loader) = { let manifest = archive .resolve_profile(selector, &worker_root, &worker_name) .map_err(|err| format!("failed to resolve profile source archive: {err}"))?; @@ -251,15 +299,6 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { &worker_root, &worker_name, )? - } else { - // Compatibility/debug fallback for direct CLI tests. Normal Browser/Backend launch - // supplies a ProfileSourceArchive inside the Runtime config bundle and must not use - // Runtime-local filesystem profile discovery. - worker::entrypoint::resolve_runtime_profile_manifest( - profile.as_deref(), - &self.profile_base_dir, - &worker_name, - )? }; manifest.worker.name = worker_name; @@ -899,7 +938,7 @@ mod tests { label: Some("adapter-test".to_string()), }], declarations: Vec::new(), - profile_source_archive: None, + profile_source_archive: Some(sample_profile_archive()), profile_source_archive_handle: None, } .with_computed_digest() @@ -1010,10 +1049,13 @@ mod tests { worker_ref: worker_ref.clone(), request: CreateWorkerRequest { profile: ProfileSelector::RuntimeDefault, - config_bundle: ConfigBundleRef { + profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { + archive: bundle.profile_source_archive.clone().unwrap(), + }, + config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id.clone(), digest: bundle.metadata.digest.clone(), - }, + }), initial_input: None, working_directory_request: None, working_directory: None, @@ -1028,10 +1070,13 @@ mod tests { let bundle = test_bundle(); CreateWorkerRequest { profile: ProfileSelector::RuntimeDefault, - config_bundle: ConfigBundleRef { + profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { + archive: bundle.profile_source_archive.clone().unwrap(), + }, + config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id, digest: bundle.metadata.digest, - }, + }), initial_input: None, working_directory_request: None, working_directory: None, @@ -1077,34 +1122,16 @@ mod tests { } #[tokio::test] - async fn cached_profile_archive_still_requires_backend_authorization() { - let archive = sample_profile_archive(); - let handle = handle_for_archive(&archive); - let call_count = Arc::new(AtomicUsize::new(0)); - let client = SequencedResourceClient { - responses: Arc::new(Mutex::new(VecDeque::from([ - Ok(response_for_archive(&handle, &archive)), - Err(crate::resource::BackendResourceError::Expired), - ]))), - call_count: call_count.clone(), + async fn embedded_profile_source_archive_does_not_require_backend_resource_fetch() { + let factory = ProfileRuntimeWorkerFactory::new(tempfile::tempdir().unwrap().path()); + let bundle = test_bundle(); + let source = crate::catalog::ProfileSourceArchiveSource::Embedded { + archive: bundle.profile_source_archive.clone().unwrap(), }; - let factory = ProfileRuntimeWorkerFactory::new(tempfile::tempdir().unwrap().path()) - .with_resource_client(Arc::new(client)); - let mut bundle = test_bundle(); - bundle.profile_source_archive_handle = Some(handle); - bundle = bundle.with_computed_digest(); - factory - .resolve_profile_source_archive(&bundle, &spawn_request_with_bundle(bundle.clone())) + .resolve_profile_source_archive(&source) .await - .expect("first fetch should authorize and cache archive"); - let err = factory - .resolve_profile_source_archive(&bundle, &spawn_request_with_bundle(bundle.clone())) - .await - .unwrap_err(); - - assert!(err.contains("expired"), "unexpected error: {err}"); - assert_eq!(call_count.load(Ordering::SeqCst), 2); + .expect("embedded archive should resolve without Backend resource client"); } #[test] diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 56d491d7..6cde1b1e 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -14,7 +14,8 @@ use std::{ time::Duration, }; use worker_runtime::catalog::{ - ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail as EmbeddedWorkerDetail, + ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef, + ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail, WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectorySummary, }; @@ -1507,25 +1508,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { .profile .clone() .unwrap_or_else(|| embedded_profile_selector(&request.intent)); - let runtime_id = EmbeddedRuntimeId::new(self.runtime_id.clone()); - let config_bundle = match request - .resolved_config_bundle - .clone() - .map(Ok) - .unwrap_or_else(|| { - default_embedded_config_bundle( - &profile, - &self.host_id, - runtime_id.as_ref(), - &self.resource_broker, - ) - }) - .and_then(|bundle| { - self.runtime - .store_config_bundle(bundle) - .map_err(|err| err.to_string()) - }) { - Ok(availability) => availability.reference, + let profile_source = match default_profile_source_archive_source(&profile) { + Ok(source) => source, Err(error) => { diagnostics.push(diagnostic( "embedded_profile_source_archive_invalid", @@ -1542,7 +1526,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { }; let create_request = CreateWorkerRequest { profile, - config_bundle, + config_bundle: None, + profile_source, initial_input: request.initial_input.clone(), working_directory_request: request.resolved_working_directory_request.clone(), working_directory: request.resolved_working_directory.clone(), @@ -1894,6 +1879,7 @@ pub struct RemoteWorkerRuntime { runtime_id: String, display_name: String, base_url: String, + backend_base_url: String, bearer_token: Option, cached_capabilities: RuntimeCapabilitySummary, cached_status: String, @@ -1903,7 +1889,10 @@ pub struct RemoteWorkerRuntime { } impl RemoteWorkerRuntime { - pub fn new(config: RemoteRuntimeConfig) -> Result { + pub fn new( + config: RemoteRuntimeConfig, + backend_base_url: String, + ) -> Result { validate_backend_identifier("runtime_id", &config.runtime_id)?; let base_url = config.base_url.trim_end_matches('/').to_string(); let timeout = config.timeout; @@ -1919,6 +1908,7 @@ impl RemoteWorkerRuntime { runtime_id: config.runtime_id, display_name: config.display_name, base_url, + backend_base_url: backend_base_url.trim_end_matches('/').to_string(), bearer_token: config.bearer_token, cached_capabilities: config.cached_capabilities, cached_status: config.cached_status, @@ -2286,41 +2276,31 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { .clone() .unwrap_or_else(|| embedded_profile_selector(&request.intent)); let runtime_id = EmbeddedRuntimeId::new(self.runtime_id.clone()); - let sync = match request - .resolved_config_bundle - .clone() - .map(Ok) - .unwrap_or_else(|| { - default_embedded_config_bundle( - &profile, - &self.host_id, - runtime_id.as_ref(), - &self.resource_broker, - ) - }) { - Ok(bundle) => self.sync_config_bundle(bundle), - Err(error) => ConfigBundleSyncResult { - state: WorkerOperationState::Rejected, - availability: None, - diagnostics: vec![diagnostic( - "remote_profile_source_archive_invalid", - DiagnosticSeverity::Error, - error, - )], - }, - }; - let Some(config_bundle) = sync.availability.map(|availability| availability.reference) - else { - return WorkerSpawnResult { - state: WorkerOperationState::Rejected, - worker: None, - acceptance_evidence: Vec::new(), - diagnostics: sync.diagnostics, - }; + let profile_source = match default_profile_source_archive_http_source( + &profile, + &self.host_id, + runtime_id.as_ref(), + &self.resource_broker, + &self.backend_base_url, + ) { + Ok(source) => source, + Err(error) => { + return WorkerSpawnResult { + state: WorkerOperationState::Rejected, + worker: None, + acceptance_evidence: Vec::new(), + diagnostics: vec![diagnostic( + "remote_profile_source_archive_invalid", + DiagnosticSeverity::Error, + error, + )], + }; + } }; let create = CreateWorkerRequest { profile, - config_bundle, + config_bundle: None, + profile_source, initial_input: request.initial_input.clone(), working_directory_request: request.resolved_working_directory_request.clone(), working_directory: request.resolved_working_directory.clone(), @@ -2637,11 +2617,56 @@ fn embedded_worker_execution_status_label( } } +fn default_profile_source_archive_source( + profile: &ProfileSelector, +) -> Result { + Ok(ProfileSourceArchiveSource::Embedded { + archive: default_profile_source_archive(profile)?, + }) +} + +fn default_profile_source_archive_http_source( + profile: &ProfileSelector, + workspace_id: &str, + runtime_id: Option<&EmbeddedRuntimeId>, + resource_broker: &BackendResourceBroker, + backend_base_url: &str, +) -> Result { + let archive = default_profile_source_archive(profile)?; + let _handle = resource_broker.issue_profile_source_archive_handle( + workspace_id.to_string(), + runtime_id, + None, + archive.clone(), + ); + let etag = format!("\"profile-source:{}\"", archive.reference.digest); + let url = format!( + "{}/api/w/{}/profile-source-archives/{}", + backend_base_url.trim_end_matches('/'), + workspace_id, + archive.reference.digest + ); + Ok(ProfileSourceArchiveSource::Http { + location: ProfileSourceArchiveHttpRef { + url, + etag: Some(etag), + archive: archive.reference.clone(), + }, + }) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum ProfileSourceArchiveTransport { + Inline, + BackendResourceHandle, +} + fn default_embedded_config_bundle( profile: &ProfileSelector, workspace_id: &str, runtime_id: Option<&EmbeddedRuntimeId>, resource_broker: &BackendResourceBroker, + archive_transport: ProfileSourceArchiveTransport, ) -> Result { let id = format!( "workspace-runtime-{}", @@ -2650,12 +2675,18 @@ fn default_embedded_config_bundle( .replace([':', '/', ' '], "-") ); let archive = default_profile_source_archive(profile)?; - let handle = resource_broker.issue_profile_source_archive_handle( - workspace_id.to_string(), - runtime_id, - None, - archive, - ); + let (profile_source_archive, profile_source_archive_handle) = match archive_transport { + ProfileSourceArchiveTransport::Inline => (Some(archive), None), + ProfileSourceArchiveTransport::BackendResourceHandle => { + let handle = resource_broker.issue_profile_source_archive_handle( + workspace_id.to_string(), + runtime_id, + None, + archive, + ); + (None, Some(handle)) + } + }; Ok(ConfigBundle { metadata: ConfigBundleMetadata { id, @@ -2673,8 +2704,8 @@ fn default_embedded_config_bundle( label: embedded_profile_label(profile), }], declarations: Vec::new(), - profile_source_archive: None, - profile_source_archive_handle: Some(handle), + profile_source_archive, + profile_source_archive_handle, } .with_computed_digest()) } @@ -3277,6 +3308,7 @@ mod tests { "workspace-test", Some(&runtime_id), &broker, + ProfileSourceArchiveTransport::BackendResourceHandle, ) .unwrap(); let handle = bundle.profile_source_archive_handle.as_ref().unwrap(); @@ -3308,6 +3340,32 @@ mod tests { } } + #[test] + fn remote_default_bundle_inlines_profile_archive_for_standalone_runtime() { + let root = tempfile::tempdir().unwrap(); + let broker = BackendResourceBroker::default(); + let runtime_id = EmbeddedRuntimeId::new("remote:test".to_string()).unwrap(); + let bundle = default_embedded_config_bundle( + &ProfileSelector::Builtin("builtin:coder".to_string()), + "workspace-test", + Some(&runtime_id), + &broker, + ProfileSourceArchiveTransport::Inline, + ) + .unwrap(); + assert!(bundle.profile_source_archive_handle.is_none()); + let archive = bundle + .profile_source_archive + .as_ref() + .expect("remote default bundle carries inline profile archive") + .verify() + .unwrap(); + let manifest = archive + .resolve_profile("builtin:coder", root.path(), "remote-test-worker") + .unwrap(); + assert_eq!(manifest.worker.name, "remote-test-worker"); + } + #[test] fn embedded_archive_rejects_unknown_selectors() { let broker = BackendResourceBroker::default(); @@ -3318,6 +3376,7 @@ mod tests { "workspace-test", Some(&runtime_id), &broker, + ProfileSourceArchiveTransport::BackendResourceHandle, ) .is_err() ); @@ -3327,6 +3386,7 @@ mod tests { "workspace-test", Some(&runtime_id), &broker, + ProfileSourceArchiveTransport::BackendResourceHandle, ) .is_err() ); @@ -3936,12 +3996,15 @@ mod tests { #[tokio::test] async fn remote_runtime_client_can_initialize_inside_tokio_context() { - let runtime = RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( - "remote:async-init", - "Remote Async Init", - "http://127.0.0.1:9", - None, - )) + let runtime = RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "remote:async-init", + "Remote Async Init", + "http://127.0.0.1:9", + None, + ), + "http://127.0.0.1:8787".to_string(), + ) .unwrap(); assert_eq!(runtime.runtime_id(), "remote:async-init"); @@ -3984,12 +4047,15 @@ mod tests { let secret = "secret-token-do-not-leak".to_string(); let registry = RuntimeRegistry::new(Vec::new()); registry.register( - RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( - "remote:primary", - "Remote Primary", - base_url.clone(), - Some(secret.clone()), - )) + RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "remote:primary", + "Remote Primary", + base_url.clone(), + Some(secret.clone()), + ), + "http://127.0.0.1:8787".to_string(), + ) .unwrap(), ); @@ -4062,28 +4128,28 @@ mod tests { json!({ "workers": [ worker_json_with_execution( - "embedded-worker-runtime", + "remote:primary", "worker-stale", "stale", "unconnected", None, ), worker_json_with_execution( - "embedded-worker-runtime", + "remote:primary", "worker-unconnected", "unconnected", "unconnected", None, ), worker_json_with_execution( - "embedded-worker-runtime", + "remote:primary", "worker-rejected", "connected", "rejected", Some("rejected"), ), worker_json_with_execution( - "embedded-worker-runtime", + "remote:primary", "worker-errored", "connected", "errored", @@ -4100,7 +4166,7 @@ mod tests { 200, json!({ "worker": worker_json_with_execution( - "embedded-worker-runtime", + "remote:primary", "worker-stale", "stale", "unconnected", @@ -4110,12 +4176,15 @@ mod tests { ), ]); let registry = RuntimeRegistry::new(vec![Arc::new( - RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( - "remote:primary", - "Remote Primary", - base_url, - Some("secret-token-do-not-leak".to_string()), - )) + RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "remote:primary", + "Remote Primary", + base_url, + Some("secret-token-do-not-leak".to_string()), + ), + "http://127.0.0.1:8787".to_string(), + ) .unwrap(), )]); @@ -4181,12 +4250,15 @@ mod tests { ]); let registry = RuntimeRegistry::new(Vec::new()); registry.register( - RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( - "remote:primary", - "Remote Primary", - base_url, - Some("secret-token".to_string()), - )) + RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "remote:primary", + "Remote Primary", + base_url, + Some("secret-token".to_string()), + ), + "http://127.0.0.1:8787".to_string(), + ) .unwrap(), ); @@ -4227,12 +4299,15 @@ mod tests { )]); let registry = RuntimeRegistry::new(Vec::new()); registry.register( - RemoteWorkerRuntime::new(RemoteRuntimeConfig::new( - "remote:primary", - "Remote Primary", - base_url, - Some("secret-token".to_string()), - )) + RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "remote:primary", + "Remote Primary", + base_url, + Some("secret-token".to_string()), + ), + "http://127.0.0.1:8787".to_string(), + ) .unwrap(), ); @@ -4340,6 +4415,12 @@ mod tests { "execution": { "backend": backend, "run_state": run_state, "last_result": last_result }, "intent": { "kind": "role", "role": "coder", "purpose": "remote test" }, "profile": { "kind": "builtin", "value": "coder" }, + "profile_source": { + "id": "remote-profile-source", + "digest": "remote-profile-digest", + "size_bytes": 0, + "source_graph": { "source_count": 0, "total_source_bytes": 0, "entrypoints": {}, "import_count": 0 } + }, "config_bundle": { "id": "remote-bundle", "digest": "remote-digest" }, "transcript_len": 0, "last_event_id": 0 diff --git a/crates/workspace-server/src/resource_broker.rs b/crates/workspace-server/src/resource_broker.rs index a332d293..ade16bbb 100644 --- a/crates/workspace-server/src/resource_broker.rs +++ b/crates/workspace-server/src/resource_broker.rs @@ -68,6 +68,18 @@ impl BackendResourceBroker { handle } + pub fn profile_source_archive( + &self, + digest: &str, + ) -> Option { + self.resources + .lock() + .ok()? + .values() + .find(|resource| resource.handle.digest == digest) + .map(|resource| resource.archive.clone()) + } + pub fn fetch_profile_source_archive( &self, request: BackendResourceFetchRequest, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index dd4a9a00..1bfb173e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -3,8 +3,8 @@ use std::sync::Arc; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::extract::{Path as AxumPath, Query, State}; -use axum::http::header::{CONTENT_TYPE, LOCATION}; -use axum::http::{StatusCode, Uri}; +use axum::http::header::{CONTENT_TYPE, ETAG, IF_NONE_MATCH, LOCATION}; +use axum::http::{HeaderMap, StatusCode, Uri}; use axum::response::{IntoResponse, Response}; use axum::routing::{delete, get, post, put}; use axum::{Json, Router}; @@ -91,6 +91,7 @@ pub struct ServerConfig { pub repositories: Vec, pub runtime_event_sources: Vec, pub remote_runtime_sources: Vec, + pub backend_base_url: Option, } impl ServerConfig { @@ -113,6 +114,7 @@ impl ServerConfig { repositories: Vec::new(), runtime_event_sources: Vec::new(), remote_runtime_sources: Vec::new(), + backend_base_url: None, } } @@ -234,9 +236,15 @@ impl WorkspaceApi { ); for remote_config in config.remote_runtime_sources.iter().cloned() { runtime.register( - RemoteWorkerRuntime::new(remote_config) - .map(|host| host.with_resource_broker(resource_broker.clone())) - .map_err(|err| err.into_error())?, + RemoteWorkerRuntime::new( + remote_config, + config + .backend_base_url + .clone() + .unwrap_or_else(|| "http://127.0.0.1:8787".to_string()), + ) + .map(|host| host.with_resource_broker(resource_broker.clone())) + .map_err(|err| err.into_error())?, ); } let runtime = Arc::new(runtime); @@ -333,6 +341,10 @@ pub fn build_router(api: WorkspaceApi) -> Router { ) .route("/api/hosts", get(list_hosts)) .route("/api/w/{workspace_id}/hosts", get(scoped_list_hosts)) + .route( + "/api/w/{workspace_id}/profile-source-archives/{digest}", + get(scoped_get_profile_source_archive), + ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/working-directories", get(scoped_list_runtime_working_directories).post(scoped_create_runtime_working_directory), @@ -806,6 +818,12 @@ struct ScopedRepositoryPath { repository_id: String, } +#[derive(Debug, Deserialize)] +struct ScopedProfileArchivePath { + workspace_id: String, + digest: String, +} + #[derive(Debug, Deserialize)] struct ScopedHostPath { workspace_id: String, @@ -1099,6 +1117,32 @@ async fn scoped_list_hosts( list_hosts(State(api)).await } +async fn scoped_get_profile_source_archive( + State(api): State, + AxumPath(path): AxumPath, + headers: HeaderMap, +) -> ApiResult<(StatusCode, HeaderMap, Vec)> { + validate_workspace_scope(&api, &path.workspace_id)?; + let archive = api + .resource_broker + .profile_source_archive(&path.digest) + .ok_or_else(|| Error::Store("profile source archive not found".to_string()))?; + let etag = format!("\"profile-source:{}\"", archive.reference.digest); + if headers + .get(IF_NONE_MATCH) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.split(',').any(|candidate| candidate.trim() == etag)) + { + let mut response_headers = HeaderMap::new(); + response_headers.insert(ETAG, etag.parse().unwrap()); + return Ok((StatusCode::NOT_MODIFIED, response_headers, Vec::new())); + } + let mut response_headers = HeaderMap::new(); + response_headers.insert(ETAG, etag.parse().unwrap()); + response_headers.insert(CONTENT_TYPE, "application/x-tar".parse().unwrap()); + Ok((StatusCode::OK, response_headers, archive.content)) +} + async fn scoped_list_runtimes( State(api): State, AxumPath(path): AxumPath, @@ -1783,9 +1827,15 @@ async fn add_remote_runtime_connection( vec![diagnostic], ) })?; - let active_runtime = RemoteWorkerRuntime::new(active_config) - .map(|host| host.with_resource_broker(api.resource_broker.clone())) - .map_err(|err| err.into_error())?; + let active_runtime = RemoteWorkerRuntime::new( + active_config, + api.config + .backend_base_url + .clone() + .unwrap_or_else(|| "http://127.0.0.1:8787".to_string()), + ) + .map(|host| host.with_resource_broker(api.resource_broker.clone())) + .map_err(|err| err.into_error())?; local_config.runtimes.remote.push(remote_config); write_workspace_backend_config_for_settings(&api, &local_config)?; api.runtime.register_or_replace(active_runtime); @@ -4278,10 +4328,27 @@ mod tests { let bundle = runtime_test_bundle(); worker_runtime::catalog::CreateWorkerRequest { profile: worker_runtime::catalog::ProfileSelector::RuntimeDefault, - config_bundle: worker_runtime::catalog::ConfigBundleRef { + profile_source: worker_runtime::catalog::ProfileSourceArchiveSource::Http { + location: worker_runtime::catalog::ProfileSourceArchiveHttpRef { + url: "http://127.0.0.1/profile-source.tar".to_string(), + etag: None, + archive: worker_runtime::profile_archive::ProfileSourceArchiveRef { + id: "test-profile-source".to_string(), + digest: "test-digest".to_string(), + size_bytes: 0, + source_graph: worker_runtime::profile_archive::ProfileSourceGraphSummary { + source_count: 0, + total_source_bytes: 0, + entrypoints: std::collections::BTreeMap::new(), + import_count: 0, + }, + }, + }, + }, + config_bundle: Some(worker_runtime::catalog::ConfigBundleRef { id: bundle.metadata.id, digest: bundle.metadata.digest, - }, + }), initial_input: None, working_directory_request: None, working_directory: None,