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

chore: merge develop into hare/develop

Browse Source
This commit is contained in:
Hare
2026-09-24 01:01:35 +09:00
parent 593e521e12 0baec54ae9
commit 9b22dcc190
178 changed files with 61787 additions and 7023 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+6 -1
View File
@@ -209,7 +209,9 @@ pub struct LogicalWorkdirAttachment {
}
fn default_workdir_session_capabilities() -> workdir::WorkdirSessionCapabilities {
workdir::WorkdirSessionCapabilities::ALL
// Capability-less wire and persisted records fail closed instead of
// silently regaining write or command authority.
workdir::WorkdirSessionCapabilities::READ_ONLY
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -218,6 +220,9 @@ pub struct WorkingDirectoryAttachmentClaim {
pub working_directory_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub relative_cwd: Option<String>,
/// Backend-authoritative maximum capabilities for the attachment session.
#[serde(default = "default_workdir_session_capabilities")]
pub capabilities: workdir::WorkdirSessionCapabilities,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
crates/worker-runtime/src/config_bundle.rs
+180 -2
View File
@@ -1,10 +1,13 @@
use crate::catalog::{ConfigBundleRef, ProfileSelector};
use crate::error::RuntimeError;
use crate::profile_archive::{
ProfileArchiveError, ProfileSourceArchive, ProfileSourceArchiveRef,
ProfileArchiveError, ProfileSourceArchive, ProfileSourceArchiveRef, ProfileSourceGraphSummary,
VerifiedProfileSourceArchive,
};
use crate::resource::{BackendResourceHandle, validate_resource_handle_text};
use crate::resource::{
BackendResourceHandle, BackendResourceKind, BackendResourceOperation, ResourceRedactionPolicy,
validate_resource_handle_text,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
@@ -162,6 +165,181 @@ impl ConfigBundle {
}
}
impl From<ConfigBundle> for server_api::WorkspaceRuntimeConfigResponse {
fn from(bundle: ConfigBundle) -> Self {
let ConfigBundle {
metadata,
profiles,
declarations,
prompt_catalog,
profile_source_archive,
profile_source_archive_handle,
} = bundle;
Self {
metadata: metadata.into(),
profiles: profiles.into_iter().map(Into::into).collect(),
declarations: declarations.into_iter().map(Into::into).collect(),
prompt_catalog: prompt_catalog.map(|catalog| {
server_api::WorkspaceRuntimePromptCatalog {
templates: catalog.templates,
config_revision: catalog.config_revision,
source_digest: catalog.source_digest,
schema_fingerprint: catalog.schema_fingerprint,
toolchain_fingerprint: catalog.toolchain_fingerprint,
catalog_digest: catalog.catalog_digest,
}
}),
profile_source_archive: profile_source_archive.map(Into::into),
profile_source_archive_handle: profile_source_archive_handle.map(Into::into),
}
}
}
impl From<ConfigBundleMetadata> for server_api::WorkspaceRuntimeConfigMetadata {
fn from(metadata: ConfigBundleMetadata) -> Self {
Self {
id: metadata.id,
digest: metadata.digest,
revision: metadata.revision,
workspace_id: metadata.workspace_id,
created_at: metadata.created_at,
provenance: metadata.provenance.into(),
}
}
}
impl From<ConfigBundleProvenance> for server_api::WorkspaceRuntimeConfigProvenance {
fn from(provenance: ConfigBundleProvenance) -> Self {
Self {
source: provenance.source,
detail: provenance.detail,
}
}
}
impl From<ConfigProfileDescriptor> for server_api::WorkspaceRuntimeConfigProfile {
fn from(profile: ConfigProfileDescriptor) -> Self {
Self {
selector: profile.selector.into(),
label: profile.label,
}
}
}
impl From<ProfileSelector> for server_api::WorkspaceRuntimeProfileSelector {
fn from(selector: ProfileSelector) -> Self {
match selector {
ProfileSelector::Builtin(value) => Self::Builtin(value),
ProfileSelector::Named(value) => Self::Named(value),
}
}
}
impl From<ConfigDeclaration> for server_api::WorkspaceRuntimeConfigDeclaration {
fn from(declaration: ConfigDeclaration) -> Self {
Self {
kind: declaration.kind.into(),
name: declaration.name,
reference: declaration.reference,
}
}
}
impl From<ConfigDeclarationKind> for server_api::WorkspaceRuntimeConfigDeclarationKind {
fn from(kind: ConfigDeclarationKind) -> Self {
match kind {
ConfigDeclarationKind::SecretRef => Self::SecretRef,
ConfigDeclarationKind::MountGrant => Self::MountGrant,
ConfigDeclarationKind::NetworkPolicy => Self::NetworkPolicy,
ConfigDeclarationKind::ShellPolicy => Self::ShellPolicy,
ConfigDeclarationKind::GitPolicy => Self::GitPolicy,
ConfigDeclarationKind::CapabilityGrant => Self::CapabilityGrant,
ConfigDeclarationKind::Unsupported => Self::Unsupported,
}
}
}
impl From<ProfileSourceArchive> for server_api::WorkspaceRuntimeProfileSourceArchive {
fn from(archive: ProfileSourceArchive) -> Self {
Self {
reference: archive.reference.into(),
content: archive.content,
}
}
}
impl From<ProfileSourceArchiveRef> for server_api::WorkspaceRuntimeProfileSourceArchiveRef {
fn from(reference: ProfileSourceArchiveRef) -> Self {
Self {
id: reference.id,
digest: reference.digest,
size_bytes: reference.size_bytes,
source_graph: reference.source_graph.into(),
}
}
}
impl From<ProfileSourceGraphSummary> for server_api::WorkspaceRuntimeProfileSourceGraph {
fn from(summary: ProfileSourceGraphSummary) -> Self {
Self {
source_count: summary.source_count,
total_source_bytes: summary.total_source_bytes,
entrypoints: summary.entrypoints,
import_count: summary.import_count,
}
}
}
impl From<BackendResourceHandle> for server_api::WorkspaceRuntimeResourceHandle {
fn from(handle: BackendResourceHandle) -> Self {
Self {
kind: handle.kind.into(),
workspace_id: handle.workspace_id,
scope_id: handle.scope_id,
runtime_id: handle.runtime_id,
worker_id: handle.worker_id,
resource_id: handle.resource_id,
digest: handle.digest,
operation: handle.operation.into(),
expires_at_unix_seconds: handle.expires_at_unix_seconds,
nonce: handle.nonce,
revision: handle.revision,
generation: handle.generation,
max_bytes: handle.max_bytes,
content_type: handle.content_type,
redaction: handle.redaction.into(),
audit_correlation_id: handle.audit_correlation_id,
profile_source_graph: handle.profile_source_graph.map(Into::into),
}
}
}
impl From<BackendResourceKind> for server_api::WorkspaceRuntimeResourceKind {
fn from(kind: BackendResourceKind) -> Self {
match kind {
BackendResourceKind::ProfileSourceArchive => Self::ProfileSourceArchive,
BackendResourceKind::RepositorySshAccess => Self::RepositorySshAccess,
}
}
}
impl From<BackendResourceOperation> for server_api::WorkspaceRuntimeResourceOperation {
fn from(operation: BackendResourceOperation) -> Self {
match operation {
BackendResourceOperation::FetchArchive => Self::FetchArchive,
BackendResourceOperation::FetchOnce => Self::FetchOnce,
}
}
}
impl From<ResourceRedactionPolicy> for server_api::WorkspaceRuntimeResourceRedactionPolicy {
fn from(policy: ResourceRedactionPolicy) -> Self {
match policy {
ResourceRedactionPolicy::RuntimeInternalOnly => Self::RuntimeInternalOnly,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ConfigBundleMetadata {
pub id: String,
crates/worker-runtime/src/fs_store.rs
+37 -1
View File
@@ -1801,6 +1801,11 @@ impl<'de> Deserialize<'de> for WorkerExecutionRecord {
alias,
working_directory_id: claim.working_directory_id,
relative_cwd: claim.relative_cwd,
// The removed singular field represented a full local
// Workdir binding. Preserve that known legacy contract
// explicitly; missing capabilities on the current claim
// shape default to read-only instead.
capabilities: workdir::WorkdirSessionCapabilities::ALL,
});
}
}
@@ -2474,10 +2479,41 @@ mod tests {
assert_eq!(attachment.alias.as_str(), "workdir");
assert_eq!(attachment.working_directory_id, "wd-legacy");
assert_eq!(attachment.relative_cwd.as_deref(), Some("crates/yoi"));
assert_eq!(
attachment.capabilities,
workdir::WorkdirSessionCapabilities::ALL
);
let serialized = serde_json::to_value(execution).unwrap();
assert!(serialized["request"].get("working_directory").is_none());
}
#[test]
fn current_attachment_claim_without_capabilities_defaults_to_read_only() {
let worker_id = WorkerId::now_v7();
let mut request = schema_v7_worker_document(worker_id)["request"].clone();
request["workdir_attachments"] = serde_json::json!([{
"alias": "docs",
"working_directory_id": "wd-docs"
}]);
let execution: WorkerExecutionRecord = serde_json::from_value(serde_json::json!({
"schema_version": SCHEMA_VERSION,
"request": request,
"binding": {},
"restore_intent": "automatic"
}))
.unwrap();
assert_eq!(
execution.request.workdir_attachments[0].capabilities,
workdir::WorkdirSessionCapabilities::READ_ONLY
);
let serialized = serde_json::to_value(execution).unwrap();
assert_eq!(
serialized["request"]["workdir_attachments"][0]["capabilities"],
serde_json::to_value(workdir::WorkdirSessionCapabilities::READ_ONLY).unwrap()
);
}
#[test]
fn worker_identity_round_trips_logical_workdir_attachments() {
let worker_id = WorkerId::now_v7();
@@ -2512,7 +2548,7 @@ mod tests {
[LogicalWorkdirAttachment {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory_id: "remote-workdir".to_string(),
capabilities: workdir::WorkdirSessionCapabilities::ALL,
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
}]
);
}
crates/worker-runtime/src/resource.rs
+60 -50
View File
@@ -1,7 +1,7 @@
use crate::auth::BACKEND_RESOURCE_FETCH_PERMISSION;
use crate::identity::WorkerId;
use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex};
use crate::workspace_request::{RuntimeWorkspaceRequest, RuntimeWorkspaceRequestClient};
use crate::workspace_request::RuntimeWorkspaceRequestClient;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
@@ -219,11 +219,6 @@ impl BackendResourceClient for HttpBackendResourceClient {
&self,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> {
let body = serde_json::to_vec(&request).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
})?;
let endpoint = reqwest::Url::parse(&self.endpoint).map_err(|error| {
BackendResourceError::Transport {
message: error.to_string(),
@@ -248,17 +243,17 @@ impl BackendResourceClient for HttpBackendResourceClient {
.to_string(),
}
})?;
if !endpoint_suffix.starts_with('/') {
let expected_path = format!(
"/api/runtime/v1/workspaces/{}/resources/fetch",
client.workspace_id()
);
if endpoint_suffix != expected_path {
return Err(BackendResourceError::Unauthorized {
message: "Workspace resource endpoint does not match its request client"
.to_string(),
});
}
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
reqwest::header::CONTENT_TYPE,
reqwest::header::HeaderValue::from_static("application/json"),
);
if let Some(token) = self.bearer_token.as_deref() {
let value = reqwest::header::HeaderValue::from_str(&format!("Bearer {token}"))
.map_err(|error| BackendResourceError::Transport {
@@ -266,41 +261,57 @@ impl BackendResourceClient for HttpBackendResourceClient {
})?;
headers.insert(reqwest::header::AUTHORIZATION, value);
}
let response = client
.execute(RuntimeWorkspaceRequest {
method: reqwest::Method::POST,
path_and_query: endpoint_suffix.to_string(),
body,
let request: server_api::RuntimeResourceFetchRequest =
serde_json::from_value(serde_json::to_value(&request).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
})?)
.map_err(|error| BackendResourceError::InvalidResponse {
message: error.to_string(),
})?;
let generated = client
.server_api_client(
BACKEND_RESOURCE_FETCH_PERMISSION,
None,
headers,
permission: BACKEND_RESOURCE_FETCH_PERMISSION.to_string(),
worker_id: None,
timeout: Some(self.request_timeout),
max_response_bytes: 8 * 1024 * 1024,
})
self.request_timeout,
8 * 1024 * 1024,
)
.map_err(|error| BackendResourceError::Transport {
message: error.to_string(),
})?;
match generated
.runtime_resource_fetch(client.workspace_id().to_string(), request)
.await
.map_err(|error| {
if error.is_timeout() {
BackendResourceError::Timeout
} else {
BackendResourceError::Transport {
{
Ok(response) => {
serde_json::from_value(serde_json::to_value(&response).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
}
})?;
if response.status.is_success() {
serde_json::from_slice::<BackendResourceFetchResponse>(&response.body).map_err(|err| {
BackendResourceError::InvalidResponse {
message: err.to_string(),
}
})
} else {
let status = response.status;
match serde_json::from_slice::<BackendResourceError>(&response.body) {
Ok(error) => Err(error),
Err(err) => Err(BackendResourceError::Transport {
message: format!("backend resource fetch failed with HTTP {status}: {err}"),
}),
})?)
.map_err(|error| BackendResourceError::InvalidResponse {
message: error.to_string(),
})
}
Err(server_api::client_support::ClientError::Public { error, .. }) => {
serde_json::from_value(serde_json::to_value(error).map_err(|error| {
BackendResourceError::InvalidResponse {
message: error.to_string(),
}
})?)
.map_err(|error| BackendResourceError::InvalidResponse {
message: error.to_string(),
})
.and_then(Err)
}
Err(server_api::client_support::ClientError::Failure(
server_api::client_support::ClientFailure::Timeout,
)) => Err(BackendResourceError::Timeout),
Err(error) => Err(BackendResourceError::Transport {
message: error.to_string(),
}),
}
}
}
@@ -437,16 +448,15 @@ mod tests {
let base_url = format!("http://{address}");
let identity = RuntimeIdentityMaterial::generate("runtime-test").unwrap();
let handle = handle_for(b"archive-bytes");
let client = HttpBackendResourceClient::new(format!("{base_url}/fetch"), None)
.with_request_timeout(std::time::Duration::from_millis(25))
.with_workspace_request_client(
RuntimeWorkspaceRequestClient::new(
"workspace-test",
base_url.clone(),
"runtime-test",
)
let client = HttpBackendResourceClient::new(
format!("{base_url}/api/runtime/v1/workspaces/workspace-test/resources/fetch"),
None,
)
.with_request_timeout(std::time::Duration::from_millis(25))
.with_workspace_request_client(
RuntimeWorkspaceRequestClient::new("workspace-test", base_url.clone(), "runtime-test")
.with_runtime_request_source(&identity, base_url),
);
);
let error = client
.fetch_resource(BackendResourceFetchRequest {
crates/worker-runtime/src/runtime.rs
+55 -12
View File
@@ -2217,8 +2217,13 @@ impl Runtime {
let _ = worker.apply_worker_state(snapshot);
}
worker.restore_intent = restore_intent_for_status(worker.status);
let claims = worker
.request
.as_ref()
.map(|request| request.workdir_attachments.as_slice())
.unwrap_or_default();
worker.logical_workdir_attachments =
logical_workdir_attachments_from_statuses(&workdir_attachments);
logical_workdir_attachments_from_statuses(&workdir_attachments, claims);
worker.workdir_attachments = workdir_attachments;
worker.detail()
};
@@ -2869,8 +2874,13 @@ impl Runtime {
candidate.execution_bound = true;
candidate.status = WorkerStatus::Idle;
candidate.restore_intent = WorkerRestoreIntent::Explicit;
let claims = candidate
.request
.as_ref()
.map(|request| request.workdir_attachments.as_slice())
.unwrap_or_default();
candidate.logical_workdir_attachments =
logical_workdir_attachments_from_statuses(&workdir_attachments);
logical_workdir_attachments_from_statuses(&workdir_attachments, claims);
candidate.workdir_attachments = workdir_attachments;
state.workers.insert(worker_ref.worker_id, candidate);
state.publish_worker_upsert(worker_ref.worker_id)?;
@@ -3215,7 +3225,11 @@ impl RuntimeState {
}
};
let logical_workdir_attachments = if worker.logical_workdir_attachments.is_empty() {
logical_workdir_attachments_from_statuses(&worker.workdir_attachments)
let claims = request
.as_ref()
.map(|request| request.workdir_attachments.as_slice())
.unwrap_or_default();
logical_workdir_attachments_from_statuses(&worker.workdir_attachments, claims)
} else {
worker.logical_workdir_attachments
};
@@ -4285,17 +4299,29 @@ fn requested_workdir_ids(request: &CreateWorkerRequest) -> Vec<&str> {
fn logical_workdir_attachments_from_statuses(
attachments: &[crate::catalog::WorkingDirectoryAttachmentStatus],
claims: &[crate::catalog::WorkingDirectoryAttachmentClaim],
) -> Vec<LogicalWorkdirAttachment> {
attachments
.iter()
.map(|attachment| LogicalWorkdirAttachment {
alias: attachment.alias.clone(),
working_directory_id: attachment
.working_directory
.summary
.working_directory_id
.clone(),
capabilities: workdir::WorkdirSessionCapabilities::ALL,
.map(|attachment| {
let working_directory_id = &attachment.working_directory.summary.working_directory_id;
let capabilities = match claims.iter().find(|claim| claim.alias == attachment.alias) {
Some(claim) if claim.working_directory_id == *working_directory_id => {
claim.capabilities
}
// An alias/id mismatch is invalid at the operation boundary. If
// corrupted persisted state reaches this compatibility path,
// fail closed rather than silently restoring write authority.
Some(_) => workdir::WorkdirSessionCapabilities::READ_ONLY,
// Runtime-materialized attachment requests predate capability
// claims and retain their existing full local-session behavior.
None => workdir::WorkdirSessionCapabilities::ALL,
};
LogicalWorkdirAttachment {
alias: attachment.alias.clone(),
working_directory_id: working_directory_id.clone(),
capabilities,
}
})
.collect()
}
@@ -6586,6 +6612,7 @@ mod tests {
alias: workdir::WorkdirAttachmentAlias::new("workdir").unwrap(),
working_directory_id: "workdir-idempotent".to_string(),
relative_cwd: None,
capabilities: workdir::WorkdirSessionCapabilities::ALL,
}];
let first = runtime.create_worker(request.clone()).unwrap();
@@ -6976,7 +7003,13 @@ mod tests {
.unwrap(),
}],
});
let request = task_request("uncertain create cleanup");
let mut request = task_request("uncertain create cleanup");
request.workdir_attachments = vec![WorkingDirectoryAttachmentClaim {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory_id: "wd-uncertain".to_string(),
relative_cwd: None,
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
}];
let worker_ref = WorkerRef::new(request.worker_id);
let error = runtime.create_worker(request).unwrap_err();
@@ -6988,6 +7021,16 @@ mod tests {
let retained = runtime.worker_detail(&worker_ref).unwrap();
assert_eq!(retained.status, WorkerStatus::Idle);
assert!(!retained.execution_metadata_available);
assert_eq!(
runtime
.lock()
.unwrap()
.worker(&worker_ref)
.unwrap()
.logical_workdir_attachments[0]
.capabilities,
workdir::WorkdirSessionCapabilities::READ_ONLY
);
assert!(matches!(
runtime.delete_worker(&worker_ref),
Err(RuntimeError::InvalidRequest(_))
crates/worker-runtime/src/worker_backend.rs
+117 -3
View File
@@ -755,11 +755,25 @@ async fn fetch_workspace_config_http<T>(
Err("Workspace Config fetch requires the worker-runtime http-server feature".to_string())
}
fn initial_workdir_session_capabilities(
alias: &WorkdirAttachmentAlias,
claims: &[crate::catalog::WorkingDirectoryAttachmentClaim],
) -> WorkdirSessionCapabilities {
claims
.iter()
.find(|claim| &claim.alias == alias)
.map(|claim| claim.capabilities)
// New Runtime materialization requests have no claim and retain their
// established full local-session behavior.
.unwrap_or(WorkdirSessionCapabilities::ALL)
}
fn runtime_local_workdir_session(
workdir_id: &str,
root: &Path,
cwd: &Path,
scope: manifest::SharedScope,
capabilities: WorkdirSessionCapabilities,
command_environment: std::collections::BTreeMap<String, String>,
resources: Vec<Arc<dyn workdir::WorkdirSessionResource>>,
) -> WorkdirSessionHandle {
@@ -768,7 +782,7 @@ fn runtime_local_workdir_session(
root.to_path_buf(),
cwd.to_path_buf(),
scope,
WorkdirSessionCapabilities::ALL,
capabilities,
command_environment,
resources,
))
@@ -776,6 +790,7 @@ fn runtime_local_workdir_session(
fn runtime_local_workdir_router(
attachments: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
capabilities: &BTreeMap<WorkdirAttachmentAlias, WorkdirSessionCapabilities>,
scope: manifest::SharedScope,
) -> Result<Arc<WorkdirSessionRouter>, String> {
let router = Arc::new(WorkdirSessionRouter::new());
@@ -788,6 +803,10 @@ fn runtime_local_workdir_router(
binding.root(),
binding.cwd(),
scope.clone(),
capabilities
.get(alias)
.copied()
.unwrap_or(WorkdirSessionCapabilities::READ_ONLY),
binding.command_environment(),
binding.session_resources(),
),
@@ -800,10 +819,25 @@ fn runtime_local_workdir_router(
fn restored_workdir_router(
local_attachments: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
logical_attachments: &[crate::catalog::LogicalWorkdirAttachment],
claims: &[crate::catalog::WorkingDirectoryAttachmentClaim],
scope: manifest::SharedScope,
workspace_client: Arc<dyn WorkspaceClient>,
) -> Result<Arc<WorkdirSessionRouter>, String> {
let router = runtime_local_workdir_router(local_attachments, scope)?;
let mut local_capabilities = local_attachments
.keys()
.map(|alias| {
(
alias.clone(),
initial_workdir_session_capabilities(alias, claims),
)
})
.collect::<BTreeMap<_, _>>();
for attachment in logical_attachments {
if local_attachments.contains_key(&attachment.alias) {
local_capabilities.insert(attachment.alias.clone(), attachment.capabilities);
}
}
let router = runtime_local_workdir_router(local_attachments, &local_capabilities, scope)?;
for attachment in logical_attachments {
if let Some(local) = local_attachments.get(&attachment.alias) {
if local.working_directory.id != attachment.working_directory_id {
@@ -1068,8 +1102,22 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
})?;
let worker = prepared.worker_mut();
validate_worker_memory_settings(worker.manifest(), &request.request)?;
let workdir_capabilities = request
.workdir_attachments
.keys()
.map(|alias| {
(
alias.clone(),
initial_workdir_session_capabilities(
alias,
&request.request.workdir_attachments,
),
)
})
.collect::<BTreeMap<_, _>>();
worker.bind_workdir_sessions(runtime_local_workdir_router(
&request.workdir_attachments,
&workdir_capabilities,
worker.scope().clone(),
)?);
if let (Some(runtime_id), Some(workspace_id)) =
@@ -1300,6 +1348,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
let workdir_sessions = restored_workdir_router(
&request.workdir_attachments,
&request.logical_workdir_attachments,
&request.request.workdir_attachments,
worker.scope().clone(),
worker.workspace_client_handle(),
)?;
@@ -2177,6 +2226,7 @@ where
binding.root(),
binding.cwd(),
manifest::SharedScope::new(scope),
WorkdirSessionCapabilities::ALL,
binding.command_environment(),
binding.session_resources(),
))
@@ -2979,6 +3029,7 @@ mod tests {
working_directory_id: "remote-workdir".to_string(),
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
}],
&[],
scope,
client,
)
@@ -2992,6 +3043,37 @@ mod tests {
);
}
#[test]
fn restored_router_preserves_read_only_capabilities_for_local_attachment() {
let runtime_base = tempfile::tempdir().unwrap();
let repo = create_clean_repo();
let materializer = RuntimeGitMaterializer::new(runtime_base.path());
let worker_ref = WorkerRef::new(crate::identity::WorkerId::now_v7());
let binding = materializer
.materialize(&worker_ref, &working_directory_request(repo.path()))
.unwrap();
let alias = WorkdirAttachmentAlias::new("docs").unwrap();
let workdir_id = binding.working_directory.id.clone();
let scope = manifest::SharedScope::new(manifest::Scope::writable(binding.root()).unwrap());
let attachments = BTreeMap::from([(alias.clone(), binding)]);
let logical = [LogicalWorkdirAttachment {
alias: alias.clone(),
working_directory_id: workdir_id,
capabilities: WorkdirSessionCapabilities::READ_ONLY,
}];
let client = WorkerWorkspaceContext::local_filesystem(Some(
WorkspaceId::new("workspace-a").unwrap(),
))
.client_handle();
let router = restored_workdir_router(&attachments, &logical, &[], scope, client).unwrap();
assert_eq!(
router.session(&alias).unwrap().capabilities(),
WorkdirSessionCapabilities::READ_ONLY
);
}
#[test]
fn profile_factory_routes_workspace_requests_by_workspace_id() {
let profiles = tempfile::tempdir().unwrap();
@@ -3949,13 +4031,35 @@ mod tests {
}
#[test]
fn restore_opens_a_fresh_session_for_the_same_workdir_identity() {
fn initial_claim_capabilities_override_full_materialization_default() {
let checkout = WorkdirAttachmentAlias::new("checkout").unwrap();
let docs = WorkdirAttachmentAlias::new("docs").unwrap();
let claims = [WorkingDirectoryAttachmentClaim {
alias: docs.clone(),
working_directory_id: "workdir-docs".to_string(),
relative_cwd: None,
capabilities: WorkdirSessionCapabilities::READ_ONLY,
}];
assert_eq!(
initial_workdir_session_capabilities(&docs, &claims),
WorkdirSessionCapabilities::READ_ONLY
);
assert_eq!(
initial_workdir_session_capabilities(&checkout, &claims),
WorkdirSessionCapabilities::ALL
);
}
#[test]
fn restore_opens_a_fresh_session_with_the_same_read_only_capabilities() {
let root = tempfile::tempdir().unwrap();
let spawned = runtime_local_workdir_session(
"working-directory-42",
root.path(),
root.path(),
manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
WorkdirSessionCapabilities::READ_ONLY,
Default::default(),
Vec::new(),
);
@@ -3964,12 +4068,21 @@ mod tests {
root.path(),
root.path(),
manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
WorkdirSessionCapabilities::READ_ONLY,
Default::default(),
Vec::new(),
);
assert_eq!(spawned.workdir().id().as_str(), "working-directory-42");
assert_eq!(restored.workdir().id().as_str(), "working-directory-42");
assert_eq!(
spawned.capabilities(),
WorkdirSessionCapabilities::READ_ONLY
);
assert_eq!(
restored.capabilities(),
WorkdirSessionCapabilities::READ_ONLY
);
assert!(!Arc::ptr_eq(&spawned, &restored));
}
@@ -4961,6 +5074,7 @@ mod tests {
alias: workdir::WorkdirAttachmentAlias::new("workdir").unwrap(),
working_directory_id: workdir_id.clone(),
relative_cwd: None,
capabilities: workdir::WorkdirSessionCapabilities::ALL,
}];
let error = runtime.create_worker(request).unwrap_err();
crates/worker-runtime/src/worker_source.rs
+222 -79
View File
@@ -5,6 +5,7 @@ use crate::workspace_request::{RuntimeWorkspaceRequest, RuntimeWorkspaceRequestC
use worker::{
WorkspaceClient, WorkspaceClientError, WorkspacePromptCatalogResolution,
WorkspacePromptProjection, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceResponse,
WorkspaceServerOperation,
};
use crate::auth::{
@@ -254,45 +255,53 @@ fn execute_remote_worker_remove_http(
fn execute_remote_worker_remove_http_blocking(
request: RemoteWorkerRemoveHttpRequest,
) -> Result<WorkspaceResponse, RuntimeWorkerMutationForwardError> {
let path = format!(
"/api/w/{}/workers/remove",
request.request_client.workspace_id()
);
let body = serde_json::to_vec(&serde_json::json!({
"target_runtime_id": request.target_runtime_id,
"target_worker_id": request.target_worker_id,
"reason": request.reason,
}))
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
crate::auth::WORKER_MUTATION_SOURCE_PROOF_HEADER,
reqwest::header::HeaderValue::from_str(&request.token)
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?,
);
headers.insert(
reqwest::header::CONTENT_TYPE,
reqwest::header::HeaderValue::from_static("application/json"),
);
let response = request
let client = request
.request_client
.execute_blocking(RuntimeWorkspaceRequest {
method: reqwest::Method::POST,
path_and_query: path,
body,
.server_api_client(
WORKSPACE_REQUEST_PERMISSION,
Some(request.source_worker_id),
headers,
permission: WORKSPACE_REQUEST_PERMISSION.to_string(),
worker_id: Some(request.source_worker_id),
timeout: Some(Duration::from_secs(5)),
max_response_bytes: 8 * 1024 * 1024,
})
Duration::from_secs(5),
8 * 1024 * 1024,
)
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
let body = String::from_utf8(response.body)
let workspace_id = request.request_client.workspace_id().to_string();
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?;
Ok(WorkspaceResponse {
status: response.status.as_u16(),
body,
})
let result = runtime.block_on(client.workspace_worker_remove(
workspace_id,
server_api::WorkerRemoveRequest {
target_runtime_id: request.target_runtime_id,
target_worker_id: request.target_worker_id,
reason: request.reason,
},
));
match result {
Ok(response) => Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(&response)
.map_err(|error| RuntimeWorkerMutationForwardError::Transport(error.to_string()))?,
}),
Err(server_api::client_support::ClientError::Public { status, error }) => {
Ok(WorkspaceResponse {
status: status.as_u16(),
body: serde_json::to_string(&error).map_err(|error| {
RuntimeWorkerMutationForwardError::Transport(error.to_string())
})?,
})
}
Err(error) => Err(RuntimeWorkerMutationForwardError::Transport(
error.to_string(),
)),
}
}
#[derive(Clone)]
@@ -432,6 +441,26 @@ impl std::fmt::Debug for RuntimeOwnedWorkspaceClient {
}
}
fn generated_workspace_response<T: serde::Serialize>(
response: Result<T, server_api::client_support::ClientError<server_api::RepositoryApiError>>,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
match response {
Ok(response) => Ok(WorkspaceResponse {
status: 200,
body: serde_json::to_string(&response)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?,
}),
Err(server_api::client_support::ClientError::Public { status, error }) => {
Ok(WorkspaceResponse {
status: status.as_u16(),
body: serde_json::to_string(&error)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?,
})
}
Err(error) => Err(WorkspaceClientError::Request(error.to_string())),
}
}
impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
fn workspace_id(&self) -> Option<&str> {
Some(self.request_client.workspace_id())
@@ -452,36 +481,143 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
self.execute_with_permission(request, WORKSPACE_REQUEST_PERMISSION)
}
fn execute_server_operation(
&self,
operation: WorkspaceServerOperation,
) -> Result<WorkspaceResponse, WorkspaceClientError> {
let request_client = self.request_client.clone();
let workspace_id = self.workspace_id.clone();
let worker_id = self.worker_id.clone();
let timeout = self.request_timeout.unwrap_or(Duration::from_secs(5));
std::thread::Builder::new()
.name("yoi-workspace-server-api".to_string())
.spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
let client = request_client
.server_api_client(
WORKSPACE_REQUEST_PERMISSION,
Some(worker_id),
reqwest::header::HeaderMap::new(),
timeout,
8 * 1024 * 1024,
)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
runtime.block_on(async move {
match operation {
WorkspaceServerOperation::WorkerControlList => {
generated_workspace_response(
client.worker_control_list(workspace_id).await,
)
}
WorkspaceServerOperation::WorkerControlSpawn(request) => {
generated_workspace_response(
client.worker_control_spawn(workspace_id, request).await,
)
}
WorkspaceServerOperation::WorkerControlInput {
runtime_id,
worker_id,
request,
} => generated_workspace_response(
client
.worker_control_input(workspace_id, runtime_id, worker_id, request)
.await,
),
WorkspaceServerOperation::WorkerControlCancel {
runtime_id,
worker_id,
request,
} => generated_workspace_response(
client
.worker_control_cancel(workspace_id, runtime_id, worker_id, request)
.await,
),
WorkspaceServerOperation::WorkerControlStop {
runtime_id,
worker_id,
request,
} => generated_workspace_response(
client
.worker_control_stop(workspace_id, runtime_id, worker_id, request)
.await,
),
WorkspaceServerOperation::WorkerControlRestore {
runtime_id,
worker_id,
} => generated_workspace_response(
client
.worker_control_restore(workspace_id, runtime_id, worker_id)
.await,
),
WorkspaceServerOperation::WorkerObservationSessions => {
generated_workspace_response(
client.worker_observation_sessions(workspace_id).await,
)
}
WorkspaceServerOperation::WorkerObservationCapture(request) => {
generated_workspace_response(
client
.worker_observation_capture(workspace_id, request)
.await,
)
}
}
})
})
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?
.join()
.map_err(|_| {
WorkspaceClientError::Request(
"Workspace Server API request thread panicked".to_string(),
)
})?
}
fn list_workspace_workers(
&self,
request: worker::WorkspaceWorkerDiscoveryRequest,
) -> Result<server_api::WorkspaceWorkerDiscoveryPage, WorkspaceClientError> {
let mut path = format!(
"/api/w/{}/worker-discovery/workers?limit={}",
self.workspace_id, request.limit
);
if let Some(cursor) = request.cursor.as_deref() {
path.push_str("&cursor=");
path.push_str(&percent_encode_query(cursor));
}
if let Some(query) = request.query.as_deref() {
path.push_str("&query=");
path.push_str(&percent_encode_query(query));
}
let response = self.execute_with_permission(
WorkspaceRequest::get(path),
WORKSPACE_WORKER_DISCOVERY_PERMISSION,
)?;
if !(200..300).contains(&response.status) {
return Err(WorkspaceClientError::Request(format!(
"Workspace Worker discovery failed with HTTP {}: {}",
response.status, response.body
)));
}
serde_json::from_str(&response.body).map_err(|error| {
WorkspaceClientError::Request(format!(
"invalid Workspace Worker discovery response: {error}"
))
let request_client = self.request_client.clone();
let workspace_id = self.workspace_id.clone();
let worker_id = self.worker_id.clone();
let timeout = self.request_timeout.unwrap_or(Duration::from_secs(5));
let query = server_api::WorkspaceWorkerDiscoveryQuery {
cursor: request.cursor,
limit: Some(request.limit),
query: request.query,
};
let result = std::thread::Builder::new()
.name("yoi-worker-discovery-http".to_string())
.spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| error.to_string())?;
let client = request_client
.server_api_client(
WORKSPACE_WORKER_DISCOVERY_PERMISSION,
Some(worker_id),
reqwest::header::HeaderMap::new(),
timeout,
8 * 1024 * 1024,
)
.map_err(|error| error.to_string())?;
runtime
.block_on(client.workspace_worker_discovery(workspace_id, query))
.map_err(|error| error.to_string())
})
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?
.join()
.map_err(|_| {
WorkspaceClientError::Request(
"Workspace Worker discovery HTTP thread panicked".to_string(),
)
})?;
result.map_err(|error| {
WorkspaceClientError::Request(format!("Workspace Worker discovery failed: {error}"))
})
}
@@ -532,18 +668,29 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
response.status, response.body
)));
}
let projection: WorkspacePromptProjection =
let wire_projection: server_api::WorkspacePromptProjection =
serde_json::from_str(&response.body).map_err(|error| {
WorkspaceClientError::Request(format!(
"invalid active Workspace Prompt projection response: {error}"
))
})?;
if projection.workspace_id != self.workspace_id {
if wire_projection.workspace_id != self.workspace_id {
return Err(WorkspaceClientError::Request(format!(
"active Workspace Prompt projection scope mismatch: expected {}, got {}",
self.workspace_id, projection.workspace_id
self.workspace_id, wire_projection.workspace_id
)));
}
let projection: WorkspacePromptProjection =
serde_json::from_value(serde_json::to_value(wire_projection).map_err(|error| {
WorkspaceClientError::Request(format!(
"project active Workspace Prompt projection response: {error}"
))
})?)
.map_err(|error| {
WorkspaceClientError::Request(format!(
"convert active Workspace Prompt projection response: {error}"
))
})?;
let resolution = cache
.observe(projection)
.map_err(WorkspaceClientError::Request)?;
@@ -576,19 +723,6 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
}
}
fn percent_encode_query(value: &str) -> String {
let mut encoded = String::with_capacity(value.len());
for byte in value.bytes() {
if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
encoded.push(char::from(byte));
} else {
use std::fmt::Write as _;
let _ = write!(encoded, "%{byte:02X}");
}
}
encoded
}
#[derive(Debug, thiserror::Error)]
pub enum RuntimeWorkerMutationForwardError {
#[error(transparent)]
@@ -936,9 +1070,12 @@ mod tests {
server.join().unwrap();
let request = received.lock().unwrap().clone();
assert!(request.contains(
"GET /api/w/workspace-a/worker-discovery/workers?limit=1&cursor=v1%3A0&query=coder%20two "
));
assert!(
request.contains(
"GET /api/w/workspace-a/worker-discovery/workers?cursor=v1%3A0&limit=1&query=coder+two "
),
"unexpected discovery request: {request}"
);
let token = request
.lines()
.find_map(|line| {
@@ -952,7 +1089,7 @@ mod tests {
assert_eq!(claims.permission, WORKSPACE_WORKER_DISCOVERY_PERMISSION);
assert_eq!(
claims.path,
"/api/w/workspace-a/worker-discovery/workers?limit=1&cursor=v1%3A0&query=coder%20two"
"/api/w/workspace-a/worker-discovery/workers?cursor=v1%3A0&limit=1&query=coder+two"
);
}
@@ -1011,9 +1148,15 @@ mod tests {
let count = stream.read(&mut bytes).unwrap();
*received_for_server.lock().unwrap() =
String::from_utf8_lossy(&bytes[..count]).into_owned();
stream
.write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n")
.unwrap();
let body =
r#"{"removed":true,"runtime_id":"runtime-target","worker_id":"worker-target"}"#;
write!(
stream,
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}",
body.len(),
body
)
.unwrap();
});
let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap();
@@ -1032,7 +1175,7 @@ mod tests {
let response = forwarder
.execute_worker_remove("runtime-target", "worker-target", "retire obsolete Worker")
.unwrap();
assert_eq!(response.status, 204);
assert_eq!(response.status, 200);
server.join().unwrap();
let request = received.lock().unwrap().clone();
crates/worker-runtime/src/workspace_request.rs
+78 -2
View File
@@ -62,9 +62,57 @@ impl RuntimeWorkspaceRequestError {
timeout,
}
}
}
pub(crate) fn is_timeout(&self) -> bool {
matches!(self, Self::Transport { timeout: true, .. })
#[derive(Clone)]
pub(crate) struct RuntimeWorkspaceRequestAuthorizer {
workspace_id: String,
runtime_id: String,
request_source: Option<(RuntimeRequestSourceSigner, String)>,
permission: String,
worker_id: Option<String>,
extra_headers: HeaderMap,
}
impl server_api::client_support::RequestAuthorizer for RuntimeWorkspaceRequestAuthorizer {
fn authorize(
&self,
request: server_api::client_support::AuthorizerRequest<'_>,
) -> Result<HeaderMap, server_api::client_support::AuthorizationError> {
let mut headers = self.extra_headers.clone();
headers.insert(
RUNTIME_ID_HEADER,
reqwest::header::HeaderValue::from_str(&self.runtime_id)
.map_err(|_| server_api::client_support::AuthorizationError::new())?,
);
if let Some(worker_id) = self.worker_id.as_deref() {
headers.insert(
WORKER_ID_HEADER,
reqwest::header::HeaderValue::from_str(worker_id)
.map_err(|_| server_api::client_support::AuthorizationError::new())?,
);
}
if let Some((signer, audience)) = self.request_source.as_ref() {
let proof = signer
.issue(
audience,
&self.workspace_id,
self.worker_id.as_deref(),
&self.permission,
request.method.as_str(),
request.path_and_query,
request.body,
unix_now_seconds(),
DEFAULT_REQUEST_PROOF_TTL_SECONDS,
)
.map_err(|_| server_api::client_support::AuthorizationError::new())?;
headers.insert(
RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
reqwest::header::HeaderValue::from_str(&proof)
.map_err(|_| server_api::client_support::AuthorizationError::new())?,
);
}
Ok(headers)
}
}
@@ -117,6 +165,34 @@ impl RuntimeWorkspaceRequestClient {
&& self.base_url.trim_end_matches('/') == base_url.trim_end_matches('/')
}
pub(crate) fn server_api_client(
&self,
permission: impl Into<String>,
worker_id: Option<String>,
extra_headers: HeaderMap,
timeout: Duration,
response_body_limit: usize,
) -> Result<
server_api::ServerApiClient<RuntimeWorkspaceRequestAuthorizer>,
RuntimeWorkspaceRequestError,
> {
let authorizer = RuntimeWorkspaceRequestAuthorizer {
workspace_id: self.workspace_id.clone(),
runtime_id: self.runtime_id.clone(),
request_source: self.request_source.clone(),
permission: permission.into(),
worker_id,
extra_headers,
};
server_api::ServerApiClient::builder(&self.base_url)
.map_err(|error| RuntimeWorkspaceRequestError::InvalidRequest(error.to_string()))?
.authorizer(authorizer)
.request_timeout(timeout)
.response_body_limit(response_body_limit)
.build()
.map_err(|error| RuntimeWorkspaceRequestError::InvalidRequest(error.to_string()))
}
pub(crate) async fn execute(
&self,
request: RuntimeWorkspaceRequest,