Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: add backend resource fetch api

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-08 20:18:38 +09:00
parent 1a2097b10d
commit 57e96d3be8
No known key found for this signature in database
16 changed files with 859 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -5936,6 +5936,7 @@ dependencies = [
"llm-engine",
"manifest",
"protocol",
"reqwest",
"serde",
"serde_json",
"session-store",

3
crates/worker-runtime/Cargo.toml
View File

@ -14,7 +14,7 @@ required-features = ["ws-server", "fs-store"]
[features]
default = []
fs-store = []
http-server = ["dep:axum", "dep:tower"]
http-server = ["dep:axum", "dep:tower", "dep:reqwest"]
ws-server = ["http-server", "axum/ws", "dep:futures", "tokio/sync"]
[dependencies]
@ -28,6 +28,7 @@ serde = { workspace = true, features = ["derive"] }
session-store.workspace = true
sha2.workspace = true
serde_json.workspace = true
reqwest = { version = "0.13", optional = true, default-features = false, features = ["json", "rustls"] }
tar.workspace = true
thiserror = { workspace = true }
tokio = { workspace = true, features = ["net", "rt", "sync", "time"] }

106
crates/worker-runtime/src/config_bundle.rs
View File

@ -4,6 +4,7 @@ use crate::profile_archive::{
ProfileArchiveError, ProfileSourceArchive, ProfileSourceArchiveRef,
VerifiedProfileSourceArchive,
};
use crate::resource::{BackendResourceHandle, validate_resource_handle_text};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
@ -24,6 +25,8 @@ pub struct ConfigBundle {
pub declarations: Vec<ConfigDeclaration>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile_source_archive: Option<ProfileSourceArchive>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile_source_archive_handle: Option<BackendResourceHandle>,
}
impl ConfigBundle {
@ -75,6 +78,27 @@ impl ConfigBundle {
lines.push(format!("profile_archive_entrypoint\0{selector}\0{path}"));
}
}
if let Some(handle) = &self.profile_source_archive_handle {
lines.push(format!(
"profile_archive_handle\0{}\0{}\0{}\0{}\0{}",
handle.workspace_id,
handle.resource_id,
handle.digest,
handle.revision,
handle.max_bytes
));
for (selector, path) in handle
.profile_source_graph
.as_ref()
.map(|graph| &graph.entrypoints)
.into_iter()
.flatten()
{
lines.push(format!(
"profile_archive_handle_entrypoint\0{selector}\0{path}"
));
}
}
lines.sort();
let mut hasher = Sha256::new();
@ -105,7 +129,12 @@ impl ConfigBundle {
profile_source_archive: self
.profile_source_archive
.as_ref()
.map(|archive| archive.reference.source_graph.clone()),
.map(|archive| archive.reference.source_graph.clone())
.or_else(|| {
self.profile_source_archive_handle
.as_ref()
.and_then(|handle| handle.profile_source_graph.clone())
}),
}
}
@ -253,6 +282,33 @@ pub(crate) fn validate_config_bundle(bundle: &ConfigBundle) -> Result<(), Runtim
RuntimeError::InvalidRequest(format!("invalid profile source archive: {err}"))
})?;
}
if let Some(handle) = &bundle.profile_source_archive_handle {
for (label, value) in [
("resource handle workspace id", handle.workspace_id.as_str()),
("resource handle resource id", handle.resource_id.as_str()),
("resource handle digest", handle.digest.as_str()),
("resource handle nonce", handle.nonce.as_str()),
("resource handle revision", handle.revision.as_str()),
("resource handle content type", handle.content_type.as_str()),
(
"resource handle audit correlation id",
handle.audit_correlation_id.as_str(),
),
] {
validate_resource_handle_text(label, value).map_err(RuntimeError::InvalidRequest)?;
}
if !handle.digest.starts_with("sha256:") {
return Err(RuntimeError::InvalidRequest(
"resource handle digest must use sha256:<hex>".to_string(),
));
}
if handle.profile_source_graph.is_none() {
return Err(RuntimeError::InvalidRequest(
"profile source archive resource handle must include source graph summary"
.to_string(),
));
}
}
Ok(())
}
@ -529,6 +585,7 @@ mod tests {
reference: reference.to_string(),
}],
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}
@ -560,6 +617,53 @@ mod tests {
validate_config_bundle(&bundle_with_declaration("vault:team.api-key")).unwrap();
}
#[test]
fn bundle_summary_redacts_runtime_internal_resource_handle() {
let mut bundle = bundle_with_declaration("secret:github-token");
let source_graph = crate::profile_archive::ProfileSourceGraphSummary {
entrypoints: std::collections::BTreeMap::from([(
"default".to_string(),
"profiles/default.dcdl".to_string(),
)]),
source_count: 1,
import_count: 0,
total_source_bytes: 16,
};
bundle.profile_source_archive_handle = Some(crate::resource::BackendResourceHandle {
kind: crate::resource::BackendResourceKind::ProfileSourceArchive,
workspace_id: "workspace-1".to_string(),
scope_id: Some("scope-1".to_string()),
runtime_id: Some("runtime-1".to_string()),
worker_id: Some("worker-1".to_string()),
resource_id: "profile-source-archive:test".to_string(),
digest: "sha256:0123456789abcdef".to_string(),
operation: crate::resource::BackendResourceOperation::FetchArchive,
expires_at_unix_seconds: 4_102_444_800,
nonce: "nonce-not-for-browser".to_string(),
revision: "rev-1".to_string(),
generation: Some(1),
max_bytes: 128,
content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
redaction: crate::resource::ResourceRedactionPolicy::RuntimeInternalOnly,
audit_correlation_id: "audit-1".to_string(),
profile_source_graph: Some(source_graph),
});
let rendered = serde_json::to_string(&bundle.summary()).unwrap();
assert!(rendered.contains("profile_source_archive"));
for forbidden in [
"nonce-not-for-browser",
"profile-source-archive:test",
"sha256:0123456789abcdef",
"runtime-1",
"worker-1",
] {
assert!(
!rendered.contains(forbidden),
"leaked {forbidden}: {rendered}"
);
}
}
#[test]
fn rejects_unsafe_bundle_ids_and_refs() {
for id in ["bundle/1", "bundle?x", "bundle&x", "bundle#x", " bundle"] {

2
crates/worker-runtime/src/http_server.rs
View File

@ -850,6 +850,7 @@ mod tests {
}],
declarations: Vec::new(),
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}
@ -1137,6 +1138,7 @@ mod ws_tests {
}],
declarations: Vec::new(),
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}

1
crates/worker-runtime/src/lib.rs
View File

@ -20,6 +20,7 @@ pub mod interaction;
pub mod management;
pub mod observation;
pub mod profile_archive;
pub mod resource;
mod runtime;
pub mod worker_backend;
pub mod working_directory;

21
crates/worker-runtime/src/main.rs
View File

@ -78,6 +78,14 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
if let Some(runtime_base_dir) = config.worker_runtime_base_dir.clone() {
factory = factory.with_runtime_base_dir(runtime_base_dir);
}
if let Some(endpoint) = config.backend_resource_endpoint.clone() {
factory = factory.with_resource_client(Arc::new(
worker_runtime::resource::HttpBackendResourceClient::new(
endpoint,
config.backend_resource_token.clone(),
),
));
}
if let Some(profile) = config.profile.clone() {
factory = factory.with_profile(profile);
}
@ -173,6 +181,13 @@ where
config.worker_runtime_base_dir =
Some(PathBuf::from(take_value(&flag, inline_value, &mut args)?));
}
"--backend-resource-endpoint" => {
config.backend_resource_endpoint =
Some(take_value(&flag, inline_value, &mut args)?);
}
"--backend-resource-token" => {
config.backend_resource_token = Some(take_value(&flag, inline_value, &mut args)?);
}
"--profile" => {
config.profile = Some(take_value(&flag, inline_value, &mut args)?);
}
@ -317,6 +332,8 @@ struct ProcessConfig {
worker_store_dir: Option<PathBuf>,
worker_metadata_dir: Option<PathBuf>,
worker_runtime_base_dir: Option<PathBuf>,
backend_resource_endpoint: Option<String>,
backend_resource_token: Option<String>,
profile: Option<String>,
}
@ -330,6 +347,8 @@ impl ProcessConfig {
worker_store_dir: None,
worker_metadata_dir: None,
worker_runtime_base_dir: None,
backend_resource_endpoint: None,
backend_resource_token: None,
profile: None,
})
}
@ -405,6 +424,8 @@ Options:\n\
--worker-store-dir <PATH> Worker session store directory\n\
--worker-metadata-dir <PATH> Worker metadata directory\n\
--worker-runtime-base-dir <PATH> Worker controller runtime directory\n\
--backend-resource-endpoint <URL> Internal Backend resource fetch endpoint for resource handles\n\
--backend-resource-token <TOKEN> Optional bearer token for the Backend resource fetch endpoint\n\
--store <memory|fs> Runtime catalog store selection (default: memory)\n\
--fs-root <PATH> Runtime catalog filesystem store root\n\
--local-token <TOKEN> Minimal local bearer token placeholder\n\

2
crates/worker-runtime/src/profile_archive.rs
View File

@ -472,7 +472,7 @@ fn validate_archive_path(path: &str) -> Result<(), ProfileArchiveError> {
Ok(())
}
fn sha256_hex(data: &[u8]) -> String {
pub fn sha256_hex(data: &[u8]) -> String {
let digest = Sha256::digest(data);
let mut out = String::from("sha256:");
for byte in digest.as_slice() {

256
crates/worker-runtime/src/resource.rs Normal file
View File

@ -0,0 +1,256 @@
use crate::identity::{RuntimeId, WorkerId};
use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Mutex;
pub const PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE: &str =
"application/vnd.yoi.profile-source-archive+tar";
pub const DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES: u64 = 2 * 1024 * 1024;
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BackendResourceKind {
ProfileSourceArchive,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BackendResourceOperation {
FetchArchive,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendResourceHandle {
pub kind: BackendResourceKind,
pub workspace_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub scope_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub runtime_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_id: Option<String>,
pub resource_id: String,
pub digest: String,
pub operation: BackendResourceOperation,
pub expires_at_unix_seconds: i64,
pub nonce: String,
pub revision: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub generation: Option<u64>,
pub max_bytes: u64,
pub content_type: String,
pub redaction: ResourceRedactionPolicy,
pub audit_correlation_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub profile_source_graph: Option<crate::profile_archive::ProfileSourceGraphSummary>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ResourceRedactionPolicy {
RuntimeInternalOnly,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendResourceFetchRequest {
pub handle: BackendResourceHandle,
pub runtime_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_id: Option<String>,
pub audit_correlation_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendResourceFetchResponse {
pub kind: BackendResourceKind,
pub resource_id: String,
pub digest: String,
pub content_type: String,
pub bytes: Vec<u8>,
pub audit_correlation_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
#[serde(tag = "code", rename_all = "snake_case")]
pub enum BackendResourceError {
#[error("backend resource handle is expired")]
Expired,
#[error("backend resource handle is unauthorized: {message}")]
Unauthorized { message: String },
#[error("backend resource kind is unsupported")]
UnsupportedKind,
#[error("backend resource is missing")]
MissingResource,
#[error("backend resource digest mismatch: expected {expected}, got {actual}")]
DigestMismatch { expected: String, actual: String },
#[error("backend resource response is oversized: limit {max_bytes}, actual {actual_bytes}")]
Oversized { max_bytes: u64, actual_bytes: u64 },
#[error("backend resource content type mismatch: expected {expected}, got {actual}")]
ContentTypeMismatch { expected: String, actual: String },
#[error("backend resource transport failed: {message}")]
Transport { message: String },
#[error("backend resource response is invalid: {message}")]
InvalidResponse { message: String },
}
#[async_trait]
pub trait BackendResourceClient: Send + Sync + 'static {
async fn fetch_resource(
&self,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError>;
}
#[cfg(feature = "http-server")]
#[derive(Clone, Debug)]
pub struct HttpBackendResourceClient {
endpoint: String,
bearer_token: Option<String>,
client: reqwest::Client,
}
#[cfg(feature = "http-server")]
impl HttpBackendResourceClient {
pub fn new(endpoint: impl Into<String>, bearer_token: Option<String>) -> Self {
Self {
endpoint: endpoint.into(),
bearer_token,
client: reqwest::Client::new(),
}
}
}
#[cfg(feature = "http-server")]
#[async_trait]
impl BackendResourceClient for HttpBackendResourceClient {
async fn fetch_resource(
&self,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> {
let builder = self.client.post(&self.endpoint).json(&request);
let builder = if let Some(token) = self.bearer_token.as_deref() {
builder.bearer_auth(token)
} else {
builder
};
let response = builder
.send()
.await
.map_err(|err| BackendResourceError::Transport {
message: err.to_string(),
})?;
if response.status().is_success() {
response
.json::<BackendResourceFetchResponse>()
.await
.map_err(|err| BackendResourceError::InvalidResponse {
message: err.to_string(),
})
} else {
let status = response.status();
match response.json::<BackendResourceError>().await {
Ok(error) => Err(error),
Err(err) => Err(BackendResourceError::Transport {
message: format!("backend resource fetch failed with HTTP {status}: {err}"),
}),
}
}
}
}
#[derive(Default, Debug)]
pub struct ProfileSourceArchiveCache {
archives: Mutex<HashMap<String, ProfileSourceArchive>>,
}
impl ProfileSourceArchiveCache {
pub fn get(&self, digest: &str) -> Option<ProfileSourceArchive> {
self.archives.lock().ok()?.get(digest).cloned()
}
pub fn insert(&self, archive: ProfileSourceArchive) {
if let Ok(mut archives) = self.archives.lock() {
archives.insert(archive.reference.digest.clone(), archive);
}
}
}
pub fn build_profile_source_archive_fetch_request(
handle: BackendResourceHandle,
runtime_id: &RuntimeId,
worker_id: Option<&WorkerId>,
) -> BackendResourceFetchRequest {
let audit_correlation_id = handle.audit_correlation_id.clone();
BackendResourceFetchRequest {
handle,
runtime_id: runtime_id.as_str().to_string(),
worker_id: worker_id.map(|id| id.as_str().to_string()),
audit_correlation_id,
}
}
pub fn profile_source_archive_from_response(
handle: &BackendResourceHandle,
response: BackendResourceFetchResponse,
) -> Result<ProfileSourceArchive, BackendResourceError> {
if handle.kind != BackendResourceKind::ProfileSourceArchive
|| response.kind != BackendResourceKind::ProfileSourceArchive
{
return Err(BackendResourceError::UnsupportedKind);
}
if handle.operation != BackendResourceOperation::FetchArchive {
return Err(BackendResourceError::Unauthorized {
message: "resource handle operation is not fetch_archive".to_string(),
});
}
if response.content_type != handle.content_type {
return Err(BackendResourceError::ContentTypeMismatch {
expected: handle.content_type.clone(),
actual: response.content_type,
});
}
let actual_bytes = response.bytes.len() as u64;
if actual_bytes > handle.max_bytes {
return Err(BackendResourceError::Oversized {
max_bytes: handle.max_bytes,
actual_bytes,
});
}
let actual_digest = sha256_hex(&response.bytes);
if actual_digest != handle.digest || response.digest != handle.digest {
return Err(BackendResourceError::DigestMismatch {
expected: handle.digest.clone(),
actual: if response.digest != handle.digest {
response.digest
} else {
actual_digest
},
});
}
Ok(ProfileSourceArchive {
reference: ProfileSourceArchiveRef {
id: handle.resource_id.clone(),
digest: handle.digest.clone(),
size_bytes: actual_bytes,
source_graph: handle.profile_source_graph.clone().ok_or_else(|| {
BackendResourceError::InvalidResponse {
message: "profile source archive handle omitted source graph summary"
.to_string(),
}
})?,
},
content: response.bytes,
})
}
pub fn validate_resource_handle_text(label: &str, value: &str) -> Result<(), String> {
if value.trim().is_empty() {
return Err(format!("{label} must not be empty"));
}
if value.len() > 256 || value.contains('\0') || value.contains('\n') || value.contains('\r') {
return Err(format!("{label} contains unsupported boundary text"));
}
Ok(())
}

1
crates/worker-runtime/src/runtime.rs
View File

@ -1628,6 +1628,7 @@ mod tests {
reference: "capability:read".to_string(),
}],
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}

77
crates/worker-runtime/src/worker_backend.rs
View File

@ -20,6 +20,10 @@ use crate::execution::{
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::working_directory::{WorkingDirectoryBinding, WorkingDirectoryMaterializer};
use async_trait::async_trait;
use manifest::paths;
@ -46,7 +50,7 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static {
/// Production factory that resolves a normal Worker profile and spawns it under
/// `WorkerController`.
#[derive(Debug, Clone)]
#[derive(Clone)]
pub struct ProfileRuntimeWorkerFactory {
profile_base_dir: PathBuf,
cwd: PathBuf,
@ -54,6 +58,8 @@ pub struct ProfileRuntimeWorkerFactory {
worker_metadata_dir: Option<PathBuf>,
profile: Option<String>,
runtime_base_dir: Option<PathBuf>,
resource_client: Option<Arc<dyn BackendResourceClient>>,
profile_archive_cache: Arc<ProfileSourceArchiveCache>,
}
impl ProfileRuntimeWorkerFactory {
@ -66,6 +72,8 @@ impl ProfileRuntimeWorkerFactory {
worker_metadata_dir: None,
profile: None,
runtime_base_dir: None,
resource_client: None,
profile_archive_cache: Arc::new(ProfileSourceArchiveCache::default()),
}
}
@ -96,6 +104,11 @@ impl ProfileRuntimeWorkerFactory {
self
}
pub fn with_resource_client(mut self, resource_client: Arc<dyn BackendResourceClient>) -> Self {
self.resource_client = Some(resource_client);
self
}
fn store_dir(&self) -> Result<PathBuf, String> {
self.store_dir
.clone()
@ -152,6 +165,56 @@ impl ProfileRuntimeWorkerFactory {
}
Self::runtime_profile_value(&request.request.profile)
}
async fn resolve_profile_source_archive(
&self,
bundle: &crate::config_bundle::ConfigBundle,
request: &WorkerExecutionSpawnRequest,
) -> Result<crate::profile_archive::VerifiedProfileSourceArchive, String> {
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
)
})?;
if let Some(cached) = self.profile_archive_cache.get(&handle.digest) {
return cached
.verify()
.map_err(|err| format!("failed to verify cached profile source archive: {err}"));
}
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 archive = profile_source_archive_from_response(&handle, response)
.map_err(format_backend_resource_error)?;
self.profile_archive_cache.insert(archive.clone());
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}")
}
#[async_trait]
@ -174,14 +237,9 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
.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 = verified_profile_source_archive(bundle)
.map_err(|err| format!("failed to verify profile source archive: {err}"))?
.ok_or_else(|| {
format!(
"config bundle {} does not contain a ProfileSourceArchive",
bundle.metadata.id
)
})?;
let archive = self
.resolve_profile_source_archive(bundle, &request)
.await?;
let manifest = archive
.resolve_profile(selector, &worker_root, &worker_name)
.map_err(|err| format!("failed to resolve profile source archive: {err}"))?;
@ -791,6 +849,7 @@ mod tests {
}],
declarations: Vec::new(),
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}

1
crates/workspace-server/src/companion.rs
View File

@ -455,6 +455,7 @@ fn companion_config_bundle() -> ConfigBundle {
}],
declarations: Vec::new(),
profile_source_archive: Some(companion_profile_archive()),
profile_source_archive_handle: None,
}
.with_computed_digest()
}

102
crates/workspace-server/src/hosts.rs
View File

@ -1,4 +1,5 @@
use crate::Error;
use crate::resource_broker::BackendResourceBroker;
use chrono::Utc;
use reqwest::StatusCode;
use reqwest::blocking::{Client as BlockingHttpClient, RequestBuilder};
@ -1032,6 +1033,7 @@ pub struct EmbeddedWorkerRuntime {
host_id: String,
runtime: worker_runtime::Runtime,
execution_enabled: bool,
resource_broker: BackendResourceBroker,
}
fn embedded_runtime_options() -> EmbeddedRuntimeOptions {
@ -1082,6 +1084,11 @@ impl EmbeddedWorkerRuntime {
Ok(embedded)
}
pub fn with_resource_broker(mut self, resource_broker: BackendResourceBroker) -> Self {
self.resource_broker = resource_broker;
self
}
pub fn from_runtime(workspace_id: impl AsRef<str>, runtime: worker_runtime::Runtime) -> Self {
let runtime_id = runtime
.runtime_id()
@ -1093,6 +1100,7 @@ impl EmbeddedWorkerRuntime {
host_id: host_id_for_embedded_workspace(workspace_id.as_ref()),
runtime,
execution_enabled: false,
resource_broker: BackendResourceBroker::default(),
}
}
@ -1339,7 +1347,14 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
.profile
.clone()
.unwrap_or_else(|| embedded_profile_selector(&request.intent));
let config_bundle = match default_embedded_config_bundle(&profile).and_then(|bundle| {
let runtime_id = EmbeddedRuntimeId::new(self.runtime_id.clone());
let config_bundle = match 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())
@ -1717,6 +1732,7 @@ pub struct RemoteWorkerRuntime {
cached_capabilities: RuntimeCapabilitySummary,
cached_status: String,
host_id: String,
resource_broker: BackendResourceBroker,
http: BlockingHttpClient,
}
@ -1740,10 +1756,16 @@ impl RemoteWorkerRuntime {
bearer_token: config.bearer_token,
cached_capabilities: config.cached_capabilities,
cached_status: config.cached_status,
resource_broker: BackendResourceBroker::default(),
http,
})
}
pub fn with_resource_broker(mut self, resource_broker: BackendResourceBroker) -> Self {
self.resource_broker = resource_broker;
self
}
fn endpoint(&self, path: &str) -> String {
format!("{}{}", self.base_url, path)
}
@ -2025,7 +2047,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.profile
.clone()
.unwrap_or_else(|| embedded_profile_selector(&request.intent));
let sync = match default_embedded_config_bundle(&profile) {
let runtime_id = EmbeddedRuntimeId::new(self.runtime_id.clone());
let sync = match 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,
@ -2365,19 +2393,31 @@ fn embedded_worker_execution_status_label(
}
}
fn default_embedded_config_bundle(profile: &ProfileSelector) -> Result<ConfigBundle, String> {
fn default_embedded_config_bundle(
profile: &ProfileSelector,
workspace_id: &str,
runtime_id: Option<&EmbeddedRuntimeId>,
resource_broker: &BackendResourceBroker,
) -> Result<ConfigBundle, String> {
let id = format!(
"workspace-runtime-{}",
embedded_profile_label(profile)
.unwrap_or_else(|| "default".to_string())
.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,
);
Ok(ConfigBundle {
metadata: ConfigBundleMetadata {
id,
digest: String::new(),
revision: "workspace-runtime-v0".to_string(),
workspace_id: "workspace-server".to_string(),
workspace_id: workspace_id.to_string(),
created_at: "runtime-generated".to_string(),
provenance: ConfigBundleProvenance {
source: "workspace-server".to_string(),
@ -2389,7 +2429,8 @@ fn default_embedded_config_bundle(profile: &ProfileSelector) -> Result<ConfigBun
label: embedded_profile_label(profile),
}],
declarations: Vec::new(),
profile_source_archive: Some(default_profile_source_archive(profile)?),
profile_source_archive: None,
profile_source_archive_handle: Some(handle),
}
.with_computed_digest())
}
@ -2967,6 +3008,8 @@ mod tests {
#[test]
fn embedded_builtin_decodal_profiles_resolve_through_archive() {
let root = tempfile::tempdir().unwrap();
let broker = BackendResourceBroker::default();
let runtime_id = EmbeddedRuntimeId::new("runtime-test".to_string()).unwrap();
for selector in [
ProfileSelector::RuntimeDefault,
ProfileSelector::Builtin("builtin:companion".to_string()),
@ -2975,13 +3018,30 @@ mod tests {
ProfileSelector::Builtin("builtin:coder".to_string()),
ProfileSelector::Builtin("builtin:reviewer".to_string()),
] {
let bundle = default_embedded_config_bundle(&selector).unwrap();
let archive = bundle
.profile_source_archive
.as_ref()
.unwrap()
.verify()
let bundle = default_embedded_config_bundle(
&selector,
"workspace-test",
Some(&runtime_id),
&broker,
)
.unwrap();
let handle = bundle.profile_source_archive_handle.as_ref().unwrap();
assert!(bundle.profile_source_archive.is_none());
let response = broker
.fetch_profile_source_archive(
worker_runtime::resource::BackendResourceFetchRequest {
handle: handle.clone(),
runtime_id: runtime_id.as_str().to_string(),
worker_id: None,
audit_correlation_id: handle.audit_correlation_id.clone(),
},
)
.unwrap();
let archive =
worker_runtime::resource::profile_source_archive_from_response(handle, response)
.unwrap()
.verify()
.unwrap();
let selector_key = match &selector {
ProfileSelector::RuntimeDefault => "default".to_string(),
ProfileSelector::Builtin(name) => name.clone(),
@ -2996,14 +3056,25 @@ mod tests {
#[test]
fn embedded_archive_rejects_unknown_selectors() {
let broker = BackendResourceBroker::default();
let runtime_id = EmbeddedRuntimeId::new("runtime-test".to_string()).unwrap();
assert!(
default_embedded_config_bundle(&ProfileSelector::Builtin(
"builtin:missing".to_string()
))
default_embedded_config_bundle(
&ProfileSelector::Builtin("builtin:missing".to_string()),
"workspace-test",
Some(&runtime_id),
&broker,
)
.is_err()
);
assert!(
default_embedded_config_bundle(&ProfileSelector::Named("custom".to_string())).is_err()
default_embedded_config_bundle(
&ProfileSelector::Named("custom".to_string()),
"workspace-test",
Some(&runtime_id),
&broker,
)
.is_err()
);
}
@ -3030,6 +3101,7 @@ mod tests {
reference: "capability:read".to_string(),
}],
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}

1
crates/workspace-server/src/lib.rs
View File

@ -11,6 +11,7 @@ pub mod identity;
pub mod observation;
pub mod records;
pub mod repositories;
pub mod resource_broker;
pub mod server;
pub mod store;

231
crates/workspace-server/src/resource_broker.rs Normal file
View File

@ -0,0 +1,231 @@
use async_trait::async_trait;
use chrono::{Duration, Utc};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use uuid::Uuid;
use worker_runtime::identity::{RuntimeId, WorkerId};
use worker_runtime::profile_archive::ProfileSourceArchive;
use worker_runtime::resource::{
BackendResourceClient, BackendResourceError, BackendResourceFetchRequest,
BackendResourceFetchResponse, BackendResourceHandle, BackendResourceKind,
BackendResourceOperation, DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE, ResourceRedactionPolicy,
};
#[derive(Clone, Default)]
pub struct BackendResourceBroker {
resources: Arc<Mutex<HashMap<String, StoredResource>>>,
}
#[derive(Clone)]
struct StoredResource {
workspace_id: String,
runtime_id: Option<String>,
worker_id: Option<String>,
handle: BackendResourceHandle,
archive: ProfileSourceArchive,
}
impl BackendResourceBroker {
pub fn issue_profile_source_archive_handle(
&self,
workspace_id: impl Into<String>,
runtime_id: Option<&RuntimeId>,
worker_id: Option<&WorkerId>,
archive: ProfileSourceArchive,
) -> BackendResourceHandle {
let workspace_id = workspace_id.into();
let nonce = Uuid::now_v7().to_string();
let audit_correlation_id = format!("resource-fetch-{nonce}");
let expires_at = Utc::now() + Duration::minutes(15);
let handle = BackendResourceHandle {
kind: BackendResourceKind::ProfileSourceArchive,
workspace_id: workspace_id.clone(),
scope_id: Some("workspace-profile-source".to_string()),
runtime_id: runtime_id.map(|id| id.as_str().to_string()),
worker_id: worker_id.map(|id| id.as_str().to_string()),
resource_id: archive.reference.id.clone(),
digest: archive.reference.digest.clone(),
operation: BackendResourceOperation::FetchArchive,
expires_at_unix_seconds: expires_at.timestamp(),
nonce: nonce.clone(),
revision: archive.reference.digest.clone(),
generation: None,
max_bytes: DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
redaction: ResourceRedactionPolicy::RuntimeInternalOnly,
audit_correlation_id,
profile_source_graph: Some(archive.reference.source_graph.clone()),
};
let stored = StoredResource {
workspace_id,
runtime_id: runtime_id.map(|id| id.as_str().to_string()),
worker_id: worker_id.map(|id| id.as_str().to_string()),
handle: handle.clone(),
archive,
};
if let Ok(mut resources) = self.resources.lock() {
resources.insert(nonce, stored);
}
handle
}
pub fn fetch_profile_source_archive(
&self,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> {
verify_handle_shape(&request.handle)?;
if request.handle.expires_at_unix_seconds < Utc::now().timestamp() {
return Err(BackendResourceError::Expired);
}
let stored = self
.resources
.lock()
.map_err(|_| BackendResourceError::Transport {
message: "resource broker lock poisoned".to_string(),
})?
.get(&request.handle.nonce)
.cloned()
.ok_or(BackendResourceError::MissingResource)?;
if stored.workspace_id != request.handle.workspace_id
|| stored.runtime_id != request.handle.runtime_id
|| stored.worker_id != request.handle.worker_id
|| stored.handle.resource_id != request.handle.resource_id
|| stored.handle.digest != request.handle.digest
|| stored.handle.revision != request.handle.revision
{
return Err(BackendResourceError::Unauthorized {
message: "resource handle metadata does not match broker record".to_string(),
});
}
if let Some(expected_runtime_id) = stored.runtime_id.as_deref() {
if expected_runtime_id != request.runtime_id {
return Err(BackendResourceError::Unauthorized {
message: "runtime id does not match resource handle".to_string(),
});
}
}
if let Some(expected_worker_id) = stored.worker_id.as_deref() {
if Some(expected_worker_id) != request.worker_id.as_deref() {
return Err(BackendResourceError::Unauthorized {
message: "worker id does not match resource handle".to_string(),
});
}
}
let actual_bytes = stored.archive.content.len() as u64;
if actual_bytes > request.handle.max_bytes {
return Err(BackendResourceError::Oversized {
max_bytes: request.handle.max_bytes,
actual_bytes,
});
}
Ok(BackendResourceFetchResponse {
kind: BackendResourceKind::ProfileSourceArchive,
resource_id: stored.archive.reference.id,
digest: stored.archive.reference.digest,
content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
bytes: stored.archive.content,
audit_correlation_id: request.audit_correlation_id,
})
}
}
#[async_trait]
impl BackendResourceClient for BackendResourceBroker {
async fn fetch_resource(
&self,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> {
self.fetch_profile_source_archive(request)
}
}
fn verify_handle_shape(handle: &BackendResourceHandle) -> Result<(), BackendResourceError> {
if handle.kind != BackendResourceKind::ProfileSourceArchive {
return Err(BackendResourceError::UnsupportedKind);
}
if handle.operation != BackendResourceOperation::FetchArchive {
return Err(BackendResourceError::Unauthorized {
message: "resource handle operation is not fetch_archive".to_string(),
});
}
if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE {
return Err(BackendResourceError::ContentTypeMismatch {
expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
actual: handle.content_type.clone(),
});
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use worker_runtime::identity::RuntimeId;
use worker_runtime::profile_archive::{
ProfileSourceArchive, ProfileSourceArchiveRef, ProfileSourceGraphSummary, sha256_hex,
};
fn archive() -> ProfileSourceArchive {
let content = b"archive-content".to_vec();
let mut entrypoints = BTreeMap::new();
entrypoints.insert("default".to_string(), "profiles/default.dcdl".to_string());
ProfileSourceArchive {
reference: ProfileSourceArchiveRef {
id: "profile-source-archive:test".to_string(),
digest: sha256_hex(&content),
size_bytes: content.len() as u64,
source_graph: ProfileSourceGraphSummary {
entrypoints,
source_count: 1,
import_count: 0,
total_source_bytes: content.len() as u64,
},
},
content,
}
}
#[test]
fn broker_issues_and_verifies_profile_source_archive_handles() {
let broker = BackendResourceBroker::default();
let runtime_id = RuntimeId::new("runtime-test").unwrap();
let handle = broker.issue_profile_source_archive_handle(
"workspace-test",
Some(&runtime_id),
None,
archive(),
);
let response = broker
.fetch_profile_source_archive(BackendResourceFetchRequest {
handle: handle.clone(),
runtime_id: runtime_id.as_str().to_string(),
worker_id: None,
audit_correlation_id: handle.audit_correlation_id.clone(),
})
.expect("fetch succeeds");
assert_eq!(response.digest, handle.digest);
assert_eq!(response.content_type, PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE);
}
#[test]
fn broker_rejects_runtime_mismatch() {
let broker = BackendResourceBroker::default();
let handle = broker.issue_profile_source_archive_handle(
"workspace-test",
Some(&RuntimeId::new("runtime-a").unwrap()),
None,
archive(),
);
let err = broker
.fetch_profile_source_archive(BackendResourceFetchRequest {
handle: handle.clone(),
runtime_id: RuntimeId::new("runtime-b").unwrap().as_str().to_string(),
worker_id: None,
audit_correlation_id: handle.audit_correlation_id.clone(),
})
.unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
}
}

93
crates/workspace-server/src/server.rs
View File

@ -12,7 +12,8 @@ use chrono::{SecondsFormat, Utc};
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use worker_runtime::worker_backend::WorkerRuntimeExecutionBackend;
use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest};
use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend};
use worker_runtime::working_directory::{
LocalGitWorktreeMaterializer, WorkingDirectoryDiagnostic, WorkingDirectoryMaterializer,
};
@ -42,6 +43,7 @@ use crate::repositories::{
ConfiguredRepository, RepositoryListProjection, RepositoryLogRead, RepositoryLookupError,
RepositoryRegistryReader, RepositorySummary,
};
use crate::resource_broker::BackendResourceBroker;
use crate::store::{ControlPlaneStore, WorkspaceRecord};
use crate::{Error, Result};
use worker_runtime::catalog::{
@ -154,6 +156,7 @@ pub struct WorkspaceApi {
runtime: Arc<RuntimeRegistry>,
companion: Arc<CompanionConsole>,
observation_proxy: BackendObservationProxy,
resource_broker: BackendResourceBroker,
working_directory_materializer: Arc<LocalGitWorktreeMaterializer>,
}
@ -162,21 +165,46 @@ impl WorkspaceApi {
let materializer = Arc::new(LocalGitWorktreeMaterializer::new(
config.embedded_runtime_store_root.clone(),
));
let execution_backend =
WorkerRuntimeExecutionBackend::from_workspace(config.workspace_root.clone())
.map_err(|err| {
crate::Error::Store(format!(
"failed to initialize embedded Worker backend: {err}"
))
})?
.with_working_directory_materializer((*materializer).clone());
Self::new_with_execution_backend(config, store, Arc::new(execution_backend)).await
let resource_broker = BackendResourceBroker::default();
let execution_backend = WorkerRuntimeExecutionBackend::new(
ProfileRuntimeWorkerFactory::new(config.workspace_root.clone())
.with_resource_client(Arc::new(resource_broker.clone())),
)
.map_err(|err| {
crate::Error::Store(format!(
"failed to initialize embedded Worker backend: {err}"
))
})?
.with_working_directory_materializer((*materializer).clone());
Self::new_with_execution_backend_and_broker(
config,
store,
Arc::new(execution_backend),
resource_broker,
)
.await
}
#[cfg(test)]
async fn new_with_execution_backend(
config: ServerConfig,
store: Arc<dyn ControlPlaneStore>,
execution_backend: Arc<dyn worker_runtime::execution::WorkerExecutionBackend>,
) -> Result<Self> {
Self::new_with_execution_backend_and_broker(
config,
store,
execution_backend,
BackendResourceBroker::default(),
)
.await
}
async fn new_with_execution_backend_and_broker(
config: ServerConfig,
store: Arc<dyn ControlPlaneStore>,
execution_backend: Arc<dyn worker_runtime::execution::WorkerExecutionBackend>,
resource_broker: BackendResourceBroker,
) -> Result<Self> {
store
.upsert_workspace(&WorkspaceRecord {
@ -193,13 +221,17 @@ impl WorkspaceApi {
config.embedded_runtime_store_root.clone(),
execution_backend,
)
.map(|runtime| runtime.with_resource_broker(resource_broker.clone()))
.map_err(|err| {
crate::Error::Store(format!("invalid embedded Worker backend: {err}"))
})?,
);
for remote_config in config.remote_runtime_sources.iter().cloned() {
runtime
.register(RemoteWorkerRuntime::new(remote_config).map_err(|err| err.into_error())?);
runtime.register(
RemoteWorkerRuntime::new(remote_config)
.map(|host| host.with_resource_broker(resource_broker.clone()))
.map_err(|err| err.into_error())?,
);
}
let runtime = Arc::new(runtime);
let companion = Arc::new(CompanionConsole::new(runtime.clone()));
@ -214,6 +246,7 @@ impl WorkspaceApi {
runtime,
companion,
observation_proxy,
resource_broker,
working_directory_materializer,
})
}
@ -328,6 +361,10 @@ pub fn build_router(api: WorkspaceApi) -> Router {
"/api/w/{workspace_id}/settings/runtime-connections/remotes/{runtime_id}/test",
post(scoped_test_remote_runtime_connection),
)
.route(
"/internal/runtime/resources/fetch",
post(post_internal_runtime_resource_fetch),
)
.route("/api/companion/status", get(get_companion_status))
.route(
"/api/w/{workspace_id}/companion/status",
@ -1445,7 +1482,9 @@ async fn add_remote_runtime_connection(
vec![diagnostic],
)
})?;
let active_runtime = RemoteWorkerRuntime::new(active_config).map_err(|err| err.into_error())?;
let active_runtime = RemoteWorkerRuntime::new(active_config)
.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);
@ -1696,6 +1735,33 @@ async fn create_workspace_worker(
}))
}
async fn post_internal_runtime_resource_fetch(
State(api): State<WorkspaceApi>,
Json(request): Json<BackendResourceFetchRequest>,
) -> std::result::Result<
Json<worker_runtime::resource::BackendResourceFetchResponse>,
(StatusCode, Json<BackendResourceError>),
> {
api.resource_broker
.fetch_profile_source_archive(request)
.map(Json)
.map_err(|error| (backend_resource_error_status(&error), Json(error)))
}
fn backend_resource_error_status(error: &BackendResourceError) -> StatusCode {
match error {
BackendResourceError::Expired => StatusCode::GONE,
BackendResourceError::Unauthorized { .. } => StatusCode::UNAUTHORIZED,
BackendResourceError::MissingResource => StatusCode::NOT_FOUND,
BackendResourceError::UnsupportedKind
| BackendResourceError::DigestMismatch { .. }
| BackendResourceError::Oversized { .. }
| BackendResourceError::ContentTypeMismatch { .. }
| BackendResourceError::InvalidResponse { .. } => StatusCode::BAD_REQUEST,
BackendResourceError::Transport { .. } => StatusCode::BAD_GATEWAY,
}
}
async fn get_companion_status(
State(api): State<WorkspaceApi>,
) -> ApiResult<Json<CompanionStatusResponse>> {
@ -3406,6 +3472,7 @@ mod tests {
}],
declarations: Vec::new(),
profile_source_archive: None,
profile_source_archive_handle: None,
}
.with_computed_digest()
}

2
package.nix
View File

@ -43,7 +43,7 @@ rustPlatform.buildRustPackage rec {
filter = sourceFilter;
};
cargoHash = "sha256-g9Bh9qQ1fGZtzsjO2/vc4jZdrH5YoCR0rOgs//xUDrU=";
cargoHash = "sha256-Pk6pRrtQlMAw2Shl+nQyX81FHQsHlKYAsxtg0M4Ya8Y=";
depsExtraArgs = {
# Older fetchCargoVendor utilities used crates.io's API download endpoint,