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

feat: add browser execution workspaces

Browse Source
This commit is contained in:
Hare
2026-07-07 23:21:20 +09:00
parent ecf10c72ab
commit 684b19e87c
12 changed files with 993 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+11
View File
@@ -89,6 +89,13 @@ pub struct ExecutionWorkspaceRequest {
pub dirty_state_policy: DirtyStatePolicy,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExecutionWorkspaceAllocationClaim {
pub allocation_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub relative_cwd: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExecutionWorkspaceStatusKind {
@@ -108,6 +115,8 @@ pub struct ExecutionWorkspaceCleanupTarget {
pub struct ExecutionWorkspaceSummary {
pub allocation_id: String,
pub repository_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub requested_selector: Option<String>,
pub materializer_kind: MaterializerKind,
pub dirty_state_policy: DirtyStatePolicy,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -144,6 +153,8 @@ pub struct CreateWorkerRequest {
pub initial_input: Option<WorkerInput>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub execution_workspace: Option<ExecutionWorkspaceRequest>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub execution_workspace_allocation: Option<ExecutionWorkspaceAllocationClaim>,
}
/// Worker lifecycle status for the in-memory embedded runtime.
crates/worker-runtime/src/execution_workspace.rs
+334 -14
View File
@@ -6,16 +6,20 @@ use crate::catalog::{
use crate::identity::WorkerRef;
use serde::{Deserialize, Serialize};
use std::fs;
use std::path::{Path, PathBuf};
use std::path::{Component, Path, PathBuf};
use std::process::Command;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
const EXECUTION_WORKSPACES_DIR: &str = "execution-workspaces";
const MATERIALIZATION_RECORD: &str = "materialization.json";
static NEXT_ALLOCATION_SEQUENCE: AtomicU64 = AtomicU64::new(0);
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExecutionWorkspaceEvidence {
pub repository_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub requested_selector: Option<String>,
pub resolved_commit: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resolved_tree: Option<String>,
@@ -40,6 +44,7 @@ impl ExecutionWorkspaceAllocation {
ExecutionWorkspaceSummary {
allocation_id: self.id.clone(),
repository_id: self.repository_id.clone(),
requested_selector: self.evidence.requested_selector.clone(),
materializer_kind: self.materializer_kind.clone(),
dirty_state_policy: self.dirty_state_policy.clone(),
resolved_commit: Some(self.evidence.resolved_commit.clone()),
@@ -114,6 +119,31 @@ pub trait ExecutionWorkspaceMaterializer: Send + Sync + 'static {
request: &ExecutionWorkspaceRequest,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic>;
fn preallocate(
&self,
request: &ExecutionWorkspaceRequest,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic>;
fn bind_allocation(
&self,
allocation_id: &str,
relative_cwd: Option<&str>,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic>;
fn list_allocations(
&self,
) -> Result<Vec<ExecutionWorkspaceStatus>, ExecutionWorkspaceDiagnostic>;
fn allocation_status(
&self,
allocation_id: &str,
) -> Result<ExecutionWorkspaceStatus, ExecutionWorkspaceDiagnostic>;
fn cleanup_allocation(
&self,
allocation_id: &str,
) -> Result<ExecutionWorkspaceStatus, ExecutionWorkspaceDiagnostic>;
fn cleanup(
&self,
binding: &ExecutionWorkspaceBinding,
@@ -174,14 +204,41 @@ impl LocalGitWorktreeMaterializer {
)
})
}
}
impl ExecutionWorkspaceMaterializer for LocalGitWorktreeMaterializer {
fn materialize(
fn read_binding(
&self,
worker_ref: &WorkerRef,
request: &ExecutionWorkspaceRequest,
allocation_id: &str,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic> {
let allocation_root = self.allocation_root(allocation_id);
let path = allocation_root.join(MATERIALIZATION_RECORD);
let raw = fs::read(&path).map_err(|_| {
ExecutionWorkspaceDiagnostic::new(
"execution_workspace_allocation_not_found",
"execution workspace allocation was not found",
)
})?;
let record: ExecutionWorkspaceMaterializationRecord = serde_json::from_slice(&raw).map_err(|_| {
ExecutionWorkspaceDiagnostic::new(
"execution_workspace_record_invalid",
"execution workspace allocation record is invalid; backend-private path details were omitted",
)
})?;
Ok(ExecutionWorkspaceBinding {
allocation: record.allocation,
workspace_root: record.workspace_root.clone(),
cwd: record.workspace_root,
allocation_root,
source_repository_path: record.source_repository_path,
})
}
fn materialize_with_allocation_id(
&self,
allocation_id: String,
request: &ExecutionWorkspaceRequest,
cleanup_policy: &str,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic> {
validate_allocation_id(&allocation_id)?;
if request.materializer != MaterializerKind::LocalGitWorktree {
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_materializer_unsupported",
@@ -248,7 +305,6 @@ impl ExecutionWorkspaceMaterializer for LocalGitWorktreeMaterializer {
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty());
let allocation_id = Self::allocation_id(worker_ref, &request.repository.id);
let allocation_root = self.allocation_root(&allocation_id);
let workspace_root = allocation_root
.join("root")
@@ -256,7 +312,7 @@ impl ExecutionWorkspaceMaterializer for LocalGitWorktreeMaterializer {
if workspace_root.exists() {
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_allocation_exists",
"execution workspace allocation target already exists; cleanup or choose a new Worker allocation",
"execution workspace allocation target already exists; cleanup or choose a new allocation",
));
}
fs::create_dir_all(workspace_root.parent().ok_or_else(|| {
@@ -285,12 +341,17 @@ impl ExecutionWorkspaceMaterializer for LocalGitWorktreeMaterializer {
)?;
let allocation = ExecutionWorkspaceAllocation {
id: allocation_id,
id: allocation_id.clone(),
repository_id: request.repository.id.clone(),
materializer_kind: MaterializerKind::LocalGitWorktree,
dirty_state_policy: DirtyStatePolicy::CleanPointOnly,
evidence: ExecutionWorkspaceEvidence {
repository_id: request.repository.id.clone(),
requested_selector: request
.repository
.selector
.as_ref()
.map(|selector| selector.as_ref().to_string()),
resolved_commit,
resolved_tree,
materializer_kind: MaterializerKind::LocalGitWorktree,
@@ -298,10 +359,10 @@ impl ExecutionWorkspaceMaterializer for LocalGitWorktreeMaterializer {
},
cleanup_target: ExecutionWorkspaceCleanupTarget {
kind: "git_worktree".to_string(),
allocation_id: Self::allocation_id(worker_ref, &request.repository.id),
allocation_id,
repository_id: request.repository.id.clone(),
},
cleanup_policy: "remove_on_worker_stop".to_string(),
cleanup_policy: cleanup_policy.to_string(),
status: ExecutionWorkspaceStatusKind::Active,
};
let binding = ExecutionWorkspaceBinding {
@@ -314,20 +375,128 @@ impl ExecutionWorkspaceMaterializer for LocalGitWorktreeMaterializer {
self.write_record(&binding)?;
Ok(binding)
}
}
impl ExecutionWorkspaceMaterializer for LocalGitWorktreeMaterializer {
fn materialize(
&self,
worker_ref: &WorkerRef,
request: &ExecutionWorkspaceRequest,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic> {
let allocation_id = Self::allocation_id(worker_ref, &request.repository.id);
self.materialize_with_allocation_id(allocation_id, request, "remove_on_worker_stop")
}
fn preallocate(
&self,
request: &ExecutionWorkspaceRequest,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic> {
let allocation_id = next_allocation_id(&request.repository.id);
self.materialize_with_allocation_id(allocation_id, request, "manual_or_worker_stop")
}
fn bind_allocation(
&self,
allocation_id: &str,
relative_cwd: Option<&str>,
) -> Result<ExecutionWorkspaceBinding, ExecutionWorkspaceDiagnostic> {
validate_allocation_id(allocation_id)?;
let binding = self.read_binding(allocation_id)?;
if binding.allocation.status != ExecutionWorkspaceStatusKind::Active {
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_not_active",
"execution workspace allocation is not active",
));
}
let cwd = validate_relative_cwd(binding.workspace_root(), relative_cwd)?;
Ok(ExecutionWorkspaceBinding { cwd, ..binding })
}
fn list_allocations(
&self,
) -> Result<Vec<ExecutionWorkspaceStatus>, ExecutionWorkspaceDiagnostic> {
let root = self.runtime_root.join(EXECUTION_WORKSPACES_DIR);
let entries = match fs::read_dir(&root) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(_) => {
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_list_failed",
"failed to list execution workspace allocations; backend-private path details were omitted",
));
}
};
let mut statuses = Vec::new();
for entry in entries.flatten() {
let file_type = match entry.file_type() {
Ok(file_type) => file_type,
Err(_) => continue,
};
if !file_type.is_dir() {
continue;
}
let allocation_id = entry.file_name().to_string_lossy().to_string();
if validate_allocation_id(&allocation_id).is_err() {
continue;
}
if let Ok(status) = self.allocation_status(&allocation_id) {
statuses.push(status);
}
}
statuses
.sort_by(|left, right| left.summary.allocation_id.cmp(&right.summary.allocation_id));
Ok(statuses)
}
fn allocation_status(
&self,
allocation_id: &str,
) -> Result<ExecutionWorkspaceStatus, ExecutionWorkspaceDiagnostic> {
validate_allocation_id(allocation_id)?;
Ok(self.read_binding(allocation_id)?.status())
}
fn cleanup_allocation(
&self,
allocation_id: &str,
) -> Result<ExecutionWorkspaceStatus, ExecutionWorkspaceDiagnostic> {
validate_allocation_id(allocation_id)?;
let binding = self.read_binding(allocation_id)?;
self.cleanup(&binding)?;
self.allocation_status(allocation_id)
}
fn cleanup(
&self,
binding: &ExecutionWorkspaceBinding,
) -> Result<(), ExecutionWorkspaceDiagnostic> {
let mut allocation = binding.allocation.clone();
let workspace_root_arg = path_str(binding.workspace_root())?;
let allocation_root = binding.allocation_root.canonicalize().map_err(|_| {
ExecutionWorkspaceDiagnostic::new(
"execution_workspace_cleanup_target_invalid",
"execution workspace allocation root is unavailable; backend-private path details were omitted",
)
})?;
let workspace_root = binding.workspace_root.canonicalize().map_err(|_| {
ExecutionWorkspaceDiagnostic::new(
"execution_workspace_cleanup_target_invalid",
"execution workspace root is unavailable; backend-private path details were omitted",
)
})?;
if !workspace_root.starts_with(&allocation_root) {
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_cleanup_escape_rejected",
"execution workspace cleanup target is outside the allocation root",
));
}
let workspace_root_arg = path_str(&workspace_root)?;
let remove_result = git_status(
binding.source_repository_path(),
["worktree", "remove", "--force", workspace_root_arg.as_str()],
)
.or_else(|_| {
if binding.workspace_root.exists() {
fs::remove_dir_all(binding.workspace_root()).map_err(|_| {
if workspace_root.exists() {
fs::remove_dir_all(&workspace_root).map_err(|_| {
ExecutionWorkspaceDiagnostic::new(
"execution_workspace_cleanup_failed",
"failed to remove execution workspace; backend-private path details were omitted",
@@ -431,6 +600,80 @@ fn sanitize_path_component(value: &str) -> String {
}
}
fn next_allocation_id(repository_id: &str) -> String {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis())
.unwrap_or_default();
let sequence = NEXT_ALLOCATION_SEQUENCE.fetch_add(1, Ordering::Relaxed);
format!(
"alloc-{now}-{sequence}-{}",
sanitize_path_component(repository_id)
)
}
fn validate_allocation_id(allocation_id: &str) -> Result<(), ExecutionWorkspaceDiagnostic> {
let sanitized = sanitize_path_component(allocation_id);
if allocation_id.is_empty()
|| allocation_id != sanitized
|| allocation_id.contains(std::path::MAIN_SEPARATOR)
|| allocation_id.contains('/')
|| allocation_id.contains('\\')
{
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_allocation_id_invalid",
"execution workspace allocation id is invalid",
));
}
Ok(())
}
fn validate_relative_cwd(
workspace_root: &Path,
relative_cwd: Option<&str>,
) -> Result<PathBuf, ExecutionWorkspaceDiagnostic> {
let workspace_root = workspace_root.canonicalize().map_err(|_| {
ExecutionWorkspaceDiagnostic::new(
"execution_workspace_root_unavailable",
"execution workspace root is unavailable; backend-private path details were omitted",
)
})?;
let relative = relative_cwd.unwrap_or(".").trim();
if relative.is_empty() || relative == "." {
return Ok(workspace_root);
}
let relative_path = Path::new(relative);
if relative_path.is_absolute()
|| relative_path.components().any(|component| {
matches!(
component,
Component::ParentDir | Component::RootDir | Component::Prefix(_)
)
})
{
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_relative_cwd_invalid",
"execution workspace relative_cwd must be a relative path inside the allocation root",
));
}
let target = workspace_root
.join(relative_path)
.canonicalize()
.map_err(|_| {
ExecutionWorkspaceDiagnostic::new(
"execution_workspace_relative_cwd_unavailable",
"execution workspace relative_cwd does not identify an existing directory",
)
})?;
if !target.starts_with(&workspace_root) || !target.is_dir() {
return Err(ExecutionWorkspaceDiagnostic::new(
"execution_workspace_relative_cwd_escape_rejected",
"execution workspace relative_cwd must resolve to a directory inside the allocation root",
));
}
Ok(target)
}
#[cfg(test)]
mod tests {
use super::*;
@@ -575,6 +818,83 @@ mod tests {
);
}
#[test]
fn preallocated_allocation_binds_safe_relative_cwd_and_lists_without_paths() {
let repo = create_clean_repo();
fs::create_dir_all(repo.path().join("crates/yoi")).unwrap();
fs::write(repo.path().join("crates/yoi/lib.rs"), "// ok\n").unwrap();
git(repo.path(), &["add", "crates/yoi/lib.rs"]);
git(repo.path(), &["commit", "-m", "add crate"]);
let runtime_root = tempfile::tempdir().unwrap();
let materializer = LocalGitWorktreeMaterializer::new(runtime_root.path());
let allocation = materializer.preallocate(&request(repo.path())).unwrap();
let bound = materializer
.bind_allocation(&allocation.allocation.id, Some("crates/yoi"))
.unwrap();
assert_eq!(bound.cwd.file_name().unwrap(), "yoi");
let listed = materializer.list_allocations().unwrap();
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].summary.allocation_id, allocation.allocation.id);
assert_eq!(
listed[0].summary.requested_selector.as_deref(),
Some("HEAD")
);
}
#[test]
fn relative_cwd_rejects_absolute_parent_nonexistent_file_and_symlink_escape() {
let repo = create_clean_repo();
fs::create_dir_all(repo.path().join("inside")).unwrap();
fs::write(repo.path().join("inside/file.txt"), "file\n").unwrap();
git(repo.path(), &["add", "inside/file.txt"]);
git(repo.path(), &["commit", "-m", "add inside"]);
let runtime_root = tempfile::tempdir().unwrap();
let materializer = LocalGitWorktreeMaterializer::new(runtime_root.path());
let allocation = materializer.preallocate(&request(repo.path())).unwrap();
assert_eq!(
materializer
.bind_allocation(&allocation.allocation.id, Some("/tmp"))
.unwrap_err()
.code,
"execution_workspace_relative_cwd_invalid"
);
assert_eq!(
materializer
.bind_allocation(&allocation.allocation.id, Some("../outside"))
.unwrap_err()
.code,
"execution_workspace_relative_cwd_invalid"
);
assert_eq!(
materializer
.bind_allocation(&allocation.allocation.id, Some("missing"))
.unwrap_err()
.code,
"execution_workspace_relative_cwd_unavailable"
);
assert_eq!(
materializer
.bind_allocation(&allocation.allocation.id, Some("inside/file.txt"))
.unwrap_err()
.code,
"execution_workspace_relative_cwd_escape_rejected"
);
#[cfg(unix)]
{
std::os::unix::fs::symlink("/tmp", allocation.workspace_root().join("escape")).unwrap();
assert_eq!(
materializer
.bind_allocation(&allocation.allocation.id, Some("escape"))
.unwrap_err()
.code,
"execution_workspace_relative_cwd_escape_rejected"
);
}
}
#[test]
fn cleanup_removes_worktree_and_updates_record() {
let repo = create_clean_repo();
crates/worker-runtime/src/http_server.rs
+2
View File
@@ -864,6 +864,7 @@ mod tests {
},
initial_input: None,
execution_workspace: None,
execution_workspace_allocation: None,
}
}
@@ -1148,6 +1149,7 @@ mod ws_tests {
},
initial_input: None,
execution_workspace: None,
execution_workspace_allocation: None,
}
}
crates/worker-runtime/src/runtime.rs
+1
View File
@@ -1594,6 +1594,7 @@ mod tests {
},
initial_input: None,
execution_workspace: None,
execution_workspace_allocation: None,
}
}
crates/worker-runtime/src/worker_backend.rs
+38 -3
View File
@@ -374,8 +374,17 @@ where
}
let mut request = request;
let execution_workspace = match request.request.execution_workspace.as_ref() {
Some(workspace_request) => {
let execution_workspace = match (
request.request.execution_workspace.as_ref(),
request.request.execution_workspace_allocation.as_ref(),
) {
(Some(_), Some(_)) => {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
"worker spawn cannot specify both execution_workspace and execution_workspace_allocation",
));
}
(Some(workspace_request), None) => {
let Some(materializer) = self.execution_workspace_materializer.as_ref() else {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
@@ -397,7 +406,32 @@ where
}
}
}
None => None,
(None, Some(allocation)) => {
let Some(materializer) = self.execution_workspace_materializer.as_ref() else {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
"execution workspace allocation requested, but no materializer is configured for this runtime backend",
));
};
match materializer.bind_allocation(
&allocation.allocation_id,
allocation.relative_cwd.as_deref(),
) {
Ok(binding) => {
request.execution_workspace = Some(binding.clone());
Some(binding)
}
Err(error) => {
return WorkerExecutionSpawnResult::Rejected(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
error.to_string(),
),
);
}
}
}
(None, None) => None,
};
let factory = self.factory.clone();
@@ -747,6 +781,7 @@ mod tests {
},
initial_input: None,
execution_workspace: None,
execution_workspace_allocation: None,
}
}
crates/workspace-server/src/companion.rs
+1
View File
@@ -389,6 +389,7 @@ fn spawn_companion_worker(runtime: &RuntimeRegistry) -> CompanionWorkerState {
initial_input: None,
execution_workspace: None,
resolved_execution_workspace: None,
resolved_execution_workspace_allocation: None,
},
);
crates/workspace-server/src/hosts.rs
+26 -1
View File
@@ -12,7 +12,8 @@ use std::{
time::Duration,
};
use worker_runtime::catalog::{
ConfigBundleRef, CreateWorkerRequest, ExecutionWorkspaceRequest, ProfileSelector,
ConfigBundleRef, CreateWorkerRequest, ExecutionWorkspaceAllocationClaim,
ExecutionWorkspaceRequest, ExecutionWorkspaceSummary, ProfileSelector,
WorkerDetail as EmbeddedWorkerDetail, WorkerStatus as EmbeddedWorkerStatus,
};
use worker_runtime::config_bundle::{
@@ -238,6 +239,8 @@ pub struct WorkerSummary {
pub last_seen_at: Option<String>,
pub implementation: WorkerImplementationSummary,
pub capabilities: WorkerCapabilitySummary,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub execution_workspace: Option<ExecutionWorkspaceSummary>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
@@ -299,6 +302,8 @@ pub struct WorkerSpawnRequest {
pub execution_workspace: Option<WorkerSpawnExecutionWorkspaceRequest>,
#[serde(skip, default)]
pub resolved_execution_workspace: Option<ExecutionWorkspaceRequest>,
#[serde(skip, default)]
pub resolved_execution_workspace_allocation: Option<ExecutionWorkspaceAllocationClaim>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -1138,6 +1143,11 @@ impl EmbeddedWorkerRuntime {
can_stop: self.can_stop_embedded_worker(summary.status, &summary.execution),
can_spawn_followup: false,
},
execution_workspace: summary
.execution
.execution_workspace
.clone()
.map(|status| status.summary),
diagnostics: embedded_worker_projection_diagnostics(&summary.execution),
}
}
@@ -1167,6 +1177,11 @@ impl EmbeddedWorkerRuntime {
can_stop: self.can_stop_embedded_worker(detail.status, &detail.execution),
can_spawn_followup: false,
},
execution_workspace: detail
.execution
.execution_workspace
.clone()
.map(|status| status.summary),
diagnostics: embedded_worker_projection_diagnostics(&detail.execution),
}
}
@@ -1342,6 +1357,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
config_bundle,
initial_input: request.initial_input.clone(),
execution_workspace: request.resolved_execution_workspace.clone(),
execution_workspace_allocation: request.resolved_execution_workspace_allocation.clone(),
};
match self.runtime.create_worker(create_request) {
Ok(detail) => {
@@ -1817,6 +1833,7 @@ impl RemoteWorkerRuntime {
can_stop: runtime_worker_can_stop(true, summary.status, &summary.execution),
can_spawn_followup: false,
},
execution_workspace: summary.execution.execution_workspace.clone().map(|status| status.summary),
diagnostics: vec![diagnostic(
"remote_runtime_projection",
DiagnosticSeverity::Info,
@@ -1850,6 +1867,7 @@ impl RemoteWorkerRuntime {
can_stop: runtime_worker_can_stop(true, detail.status, &detail.execution),
can_spawn_followup: false,
},
execution_workspace: detail.execution.execution_workspace.clone().map(|status| status.summary),
diagnostics: vec![diagnostic(
"remote_runtime_projection",
DiagnosticSeverity::Info,
@@ -2015,6 +2033,7 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
config_bundle,
initial_input: request.initial_input.clone(),
execution_workspace: request.resolved_execution_workspace.clone(),
execution_workspace_allocation: request.resolved_execution_workspace_allocation.clone(),
};
match self.post_json::<_, RuntimeHttpWorkerResponse>("/v1/workers", &create) {
Ok(response) => WorkerSpawnResult {
@@ -2825,6 +2844,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> WorkerSummary {
can_stop: false,
can_spawn_followup: false,
},
execution_workspace: None,
diagnostics: vec![diagnostic(
"runtime_capability_unsupported",
DiagnosticSeverity::Info,
@@ -3020,6 +3040,7 @@ mod tests {
can_stop: false,
can_spawn_followup: false,
},
execution_workspace: None,
diagnostics: Vec::new(),
}],
}
@@ -3172,6 +3193,7 @@ mod tests {
initial_input: None,
execution_workspace: None,
resolved_execution_workspace: None,
resolved_execution_workspace_allocation: None,
}
}
@@ -3303,6 +3325,7 @@ mod tests {
initial_input: None,
execution_workspace: None,
resolved_execution_workspace: None,
resolved_execution_workspace_allocation: None,
},
)
.unwrap();
@@ -3404,6 +3427,7 @@ mod tests {
initial_input: None,
execution_workspace: None,
resolved_execution_workspace: None,
resolved_execution_workspace_allocation: None,
},
)
.unwrap();
@@ -3434,6 +3458,7 @@ mod tests {
initial_input: None,
execution_workspace: None,
resolved_execution_workspace: None,
resolved_execution_workspace_allocation: None,
},
)
.unwrap();
crates/workspace-server/src/server.rs
+333 -38
View File
@@ -12,7 +12,9 @@ use chrono::{SecondsFormat, Utc};
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
use worker_runtime::execution_workspace::LocalGitWorktreeMaterializer;
use worker_runtime::execution_workspace::{
ExecutionWorkspaceMaterializer, LocalGitWorktreeMaterializer,
};
use worker_runtime::worker_backend::WorkerRuntimeExecutionBackend;
use crate::companion::{
@@ -43,7 +45,8 @@ use crate::repositories::{
use crate::store::{ControlPlaneStore, WorkspaceRecord};
use crate::{Error, Result};
use worker_runtime::catalog::{
ConfigBundleRef, DirtyStatePolicy, ExecutionWorkspaceRepository, ExecutionWorkspaceRequest,
ConfigBundleRef, DirtyStatePolicy, ExecutionWorkspaceAllocationClaim,
ExecutionWorkspaceRepository, ExecutionWorkspaceRequest, ExecutionWorkspaceSummary,
MaterializerKind, ProfileSelector, RepositorySelector as RuntimeRepositorySelector,
};
use worker_runtime::config_bundle::ConfigBundle;
@@ -151,10 +154,14 @@ pub struct WorkspaceApi {
runtime: Arc<RuntimeRegistry>,
companion: Arc<CompanionConsole>,
observation_proxy: BackendObservationProxy,
execution_workspace_materializer: Arc<LocalGitWorktreeMaterializer>,
}
impl WorkspaceApi {
pub async fn new(config: ServerConfig, store: Arc<dyn ControlPlaneStore>) -> Result<Self> {
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| {
@@ -162,9 +169,7 @@ impl WorkspaceApi {
"failed to initialize embedded Worker backend: {err}"
))
})?
.with_execution_workspace_materializer(LocalGitWorktreeMaterializer::new(
config.embedded_runtime_store_root.clone(),
));
.with_execution_workspace_materializer((*materializer).clone());
Self::new_with_execution_backend(config, store, Arc::new(execution_backend)).await
}
@@ -199,6 +204,9 @@ impl WorkspaceApi {
let runtime = Arc::new(runtime);
let companion = Arc::new(CompanionConsole::new(runtime.clone()));
let observation_proxy = BackendObservationProxy::new(config.runtime_event_sources.clone());
let execution_workspace_materializer = Arc::new(LocalGitWorktreeMaterializer::new(
config.embedded_runtime_store_root.clone(),
));
Ok(Self {
records: LocalProjectRecordReader::new(config.workspace_root.clone()),
config,
@@ -206,6 +214,7 @@ impl WorkspaceApi {
runtime,
companion,
observation_proxy,
execution_workspace_materializer,
})
}
@@ -261,6 +270,14 @@ 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}/execution-workspaces",
get(scoped_list_execution_workspaces).post(scoped_create_execution_workspace),
)
.route(
"/api/w/{workspace_id}/execution-workspaces/{allocation_id}",
get(scoped_execution_workspace_detail).delete(scoped_cleanup_execution_workspace),
)
.route("/api/runtimes", get(list_runtimes))
.route("/api/w/{workspace_id}/runtimes", get(scoped_list_runtimes))
.route(
@@ -530,6 +547,8 @@ pub struct WorkerLaunchOptionsResponse {
pub workspace_id: String,
pub runtimes: Vec<WorkerLaunchRuntimeOption>,
pub profiles: Vec<WorkerLaunchProfileCandidate>,
pub repositories: Vec<ExecutionWorkspaceRepositoryOption>,
pub execution_workspaces: Vec<ExecutionWorkspaceSummary>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
@@ -550,6 +569,68 @@ pub struct WorkerLaunchProfileCandidate {
pub description: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionWorkspaceRepositoryOption {
pub id: String,
pub display_name: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub default_selector: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct BrowserExecutionWorkspaceCreatePolicy {
#[serde(default)]
pub dirty_state: BrowserExecutionWorkspaceDirtyStatePolicy,
#[serde(default)]
pub cleanup: BrowserExecutionWorkspaceCleanupPolicy,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum BrowserExecutionWorkspaceDirtyStatePolicy {
#[default]
CleanPointOnly,
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum BrowserExecutionWorkspaceCleanupPolicy {
#[default]
ManualOrWorkerStop,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BrowserExecutionWorkspaceCreateRequest {
pub repository_id: String,
#[serde(default)]
pub selector: Option<String>,
#[serde(default)]
pub policy: BrowserExecutionWorkspaceCreatePolicy,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct BrowserExecutionWorkspaceListResponse {
pub workspace_id: String,
pub items: Vec<ExecutionWorkspaceSummary>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct BrowserExecutionWorkspaceDetailResponse {
pub workspace_id: String,
pub item: ExecutionWorkspaceSummary,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BrowserWorkerExecutionWorkspaceSelection {
pub allocation_id: String,
#[serde(default)]
pub relative_cwd: Option<String>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BrowserCreateWorkerRequest {
@@ -558,7 +639,7 @@ pub struct BrowserCreateWorkerRequest {
pub profile: String,
pub initial_text: String,
#[serde(default)]
pub repository_id: Option<String>,
pub execution_workspace: Option<BrowserWorkerExecutionWorkspaceSelection>,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -660,6 +741,12 @@ struct ScopedRuntimePath {
runtime_id: String,
}
#[derive(Debug, Deserialize)]
struct ScopedExecutionWorkspacePath {
workspace_id: String,
allocation_id: String,
}
#[derive(Debug, Deserialize)]
struct ScopedConfigBundlePath {
workspace_id: String,
@@ -797,6 +884,87 @@ async fn scoped_get_worker_launch_options(
get_worker_launch_options(State(api)).await
}
async fn scoped_list_execution_workspaces(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
) -> ApiResult<Json<BrowserExecutionWorkspaceListResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let items = execution_workspace_summaries(&api)?;
Ok(Json(BrowserExecutionWorkspaceListResponse {
workspace_id: api.config.workspace_id.clone(),
items,
diagnostics: Vec::new(),
}))
}
async fn scoped_create_execution_workspace(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
Json(request): Json<BrowserExecutionWorkspaceCreateRequest>,
) -> ApiResult<Json<BrowserExecutionWorkspaceDetailResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let workspace_request = execution_workspace_request_for_browser(&api, request)?;
let binding = api
.execution_workspace_materializer
.preallocate(&workspace_request)
.map_err(|diagnostic| {
ApiError::from(Error::RuntimeOperationFailed {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
code: diagnostic.code,
message: diagnostic.message,
})
})?;
Ok(Json(BrowserExecutionWorkspaceDetailResponse {
workspace_id: api.config.workspace_id.clone(),
item: binding.status().summary,
diagnostics: Vec::new(),
}))
}
async fn scoped_execution_workspace_detail(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedExecutionWorkspacePath>,
) -> ApiResult<Json<BrowserExecutionWorkspaceDetailResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let status = api
.execution_workspace_materializer
.allocation_status(&path.allocation_id)
.map_err(|diagnostic| {
ApiError::from(Error::RuntimeOperationFailed {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
code: diagnostic.code,
message: diagnostic.message,
})
})?;
Ok(Json(BrowserExecutionWorkspaceDetailResponse {
workspace_id: api.config.workspace_id.clone(),
item: status.summary,
diagnostics: Vec::new(),
}))
}
async fn scoped_cleanup_execution_workspace(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedExecutionWorkspacePath>,
) -> ApiResult<Json<BrowserExecutionWorkspaceDetailResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let status = api
.execution_workspace_materializer
.cleanup_allocation(&path.allocation_id)
.map_err(|diagnostic| {
ApiError::from(Error::RuntimeOperationFailed {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
code: diagnostic.code,
message: diagnostic.message,
})
})?;
Ok(Json(BrowserExecutionWorkspaceDetailResponse {
workspace_id: api.config.workspace_id.clone(),
item: status.summary,
diagnostics: Vec::new(),
}))
}
async fn scoped_get_runtime_connection_settings(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
@@ -1428,35 +1596,6 @@ fn configured_execution_workspace_request(
))
}
fn default_execution_workspace_request(
config: &ServerConfig,
repository_id: Option<&str>,
) -> Result<Option<ExecutionWorkspaceRequest>> {
let repository = match repository_id {
Some(id) => config
.repositories
.iter()
.find(|repository| repository.id == id)
.ok_or_else(|| {
Error::Config(format!(
"unknown repository id `{id}` for Worker execution workspace"
))
})?,
None => match config.repositories.as_slice() {
[] => return Ok(None),
[repository] => repository,
repositories => repositories
.iter()
.find(|repository| repository.id == "main")
.unwrap_or(&repositories[0]),
},
};
Ok(Some(execution_workspace_request_from_repository(
repository, None,
)))
}
async fn create_workspace_worker(
State(api): State<WorkspaceApi>,
Json(request): Json<BrowserCreateWorkerRequest>,
@@ -1482,8 +1621,23 @@ async fn create_workspace_worker(
content: initial_text,
})
};
let execution_workspace =
default_execution_workspace_request(&api.config, request.repository_id.as_deref())?;
let resolved_execution_workspace_allocation =
request
.execution_workspace
.map(|selection| ExecutionWorkspaceAllocationClaim {
allocation_id: selection.allocation_id,
relative_cwd: selection.relative_cwd,
});
if resolved_execution_workspace_allocation.is_some()
&& request.runtime_id != EMBEDDED_WORKER_RUNTIME_ID
{
return Err(Error::RuntimeOperationFailed {
runtime_id: request.runtime_id.clone(),
code: "execution_workspace_runtime_unsupported".to_string(),
message: "preallocated execution workspaces are only supported by the embedded local runtime in v0".to_string(),
}
.into());
}
let result = api
.runtime
.spawn_worker(
@@ -1497,7 +1651,8 @@ async fn create_workspace_worker(
profile: Some(profile_selector),
initial_input,
execution_workspace: None,
resolved_execution_workspace: execution_workspace,
resolved_execution_workspace: None,
resolved_execution_workspace_allocation,
},
)
.map_err(|err| err.into_error())?;
@@ -2523,10 +2678,75 @@ fn worker_launch_options_response(api: &WorkspaceApi) -> WorkerLaunchOptionsResp
workspace_id: api.config.workspace_id.clone(),
runtimes,
profiles: worker_profile_candidates(),
repositories: execution_workspace_repository_options(api),
execution_workspaces: execution_workspace_summaries(api).unwrap_or_default(),
diagnostics: Vec::new(),
}
}
fn execution_workspace_repository_options(
api: &WorkspaceApi,
) -> Vec<ExecutionWorkspaceRepositoryOption> {
api.config
.repositories
.iter()
.map(|repository| ExecutionWorkspaceRepositoryOption {
id: repository.id.clone(),
display_name: repository
.display_name
.clone()
.unwrap_or_else(|| repository.id.clone()),
default_selector: repository.default_selector.clone(),
})
.collect()
}
fn execution_workspace_summaries(api: &WorkspaceApi) -> ApiResult<Vec<ExecutionWorkspaceSummary>> {
api.execution_workspace_materializer
.list_allocations()
.map(|items| items.into_iter().map(|status| status.summary).collect())
.map_err(|diagnostic| {
ApiError::from(Error::RuntimeOperationFailed {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
code: diagnostic.code,
message: diagnostic.message,
})
})
}
fn execution_workspace_request_for_browser(
api: &WorkspaceApi,
request: BrowserExecutionWorkspaceCreateRequest,
) -> ApiResult<ExecutionWorkspaceRequest> {
let repository = api
.config
.repositories
.iter()
.find(|repository| repository.id == request.repository_id)
.ok_or_else(|| Error::UnknownRepository(request.repository_id.clone()))?;
let selector = request
.selector
.or_else(|| repository.default_selector.clone())
.filter(|selector| !selector.trim().is_empty());
match request.policy.dirty_state {
BrowserExecutionWorkspaceDirtyStatePolicy::CleanPointOnly => {}
}
match request.policy.cleanup {
BrowserExecutionWorkspaceCleanupPolicy::ManualOrWorkerStop => {}
}
Ok(ExecutionWorkspaceRequest {
repository: ExecutionWorkspaceRepository {
id: repository.id.clone(),
provider: "git".to_string(),
uri: repository.path.to_string_lossy().to_string(),
local_path: Some(repository.path.clone()),
selector: selector.map(RuntimeRepositorySelector),
},
materializer: MaterializerKind::LocalGitWorktree,
dirty_state_policy: DirtyStatePolicy::CleanPointOnly,
})
}
fn worker_profile_candidates() -> Vec<WorkerLaunchProfileCandidate> {
vec![
WorkerLaunchProfileCandidate {
@@ -3092,6 +3312,40 @@ mod tests {
config
}
fn init_clean_git_workspace(path: &std::path::Path) {
for args in [
vec!["init"],
vec!["config", "user.email", "test@example.invalid"],
vec!["config", "user.name", "Yoi Test"],
] {
let status = std::process::Command::new("git")
.arg("-C")
.arg(path)
.args(args)
.status()
.unwrap();
assert!(status.success());
}
std::fs::write(path.join("README.md"), "clean\n").unwrap();
std::fs::write(
path.join(".gitignore"),
".yoi/\n.test-embedded-runtime-store/\n",
)
.unwrap();
for args in [
vec!["add", "README.md", ".gitignore"],
vec!["commit", "-m", "init"],
] {
let status = std::process::Command::new("git")
.arg("-C")
.arg(path)
.args(args)
.status()
.unwrap();
assert!(status.success());
}
}
async fn test_app(workspace_root: impl Into<PathBuf>) -> Router {
let store = SqliteWorkspaceStore::in_memory().unwrap();
let api = WorkspaceApi::new_with_execution_backend(
@@ -3136,6 +3390,7 @@ mod tests {
},
initial_input: None,
execution_workspace: None,
execution_workspace_allocation: None,
}
}
@@ -3150,6 +3405,45 @@ mod tests {
(runtime, worker.worker_ref)
}
#[tokio::test]
async fn browser_execution_workspace_preallocate_list_detail_and_cleanup_are_path_safe() {
let dir = tempfile::tempdir().unwrap();
init_clean_git_workspace(dir.path());
let app = test_app(dir.path()).await;
let workspace_path = format!("/api/w/{TEST_WORKSPACE_ID}/execution-workspaces");
let created = post_json(
app.clone(),
&workspace_path,
serde_json::json!({
"repository_id": TEST_REPOSITORY_ID,
"selector": "HEAD",
"policy": { "dirty_state": "clean_point_only", "cleanup": "manual_or_worker_stop" }
}),
)
.await;
let allocation_id = created["item"]["allocation_id"]
.as_str()
.unwrap()
.to_string();
assert_eq!(created["item"]["repository_id"], TEST_REPOSITORY_ID);
assert_eq!(created["item"]["requested_selector"], "HEAD");
assert_eq!(created["item"]["status"], "active");
let projected = serde_json::to_string(&created).unwrap();
assert!(!projected.contains(dir.path().to_string_lossy().as_ref()));
let list = get_json(app.clone(), &workspace_path).await;
assert_eq!(list["items"].as_array().unwrap().len(), 1);
assert_eq!(list["items"][0]["allocation_id"], allocation_id);
let detail_path = format!("{workspace_path}/{allocation_id}");
let detail = get_json(app.clone(), &detail_path).await;
assert_eq!(detail["item"]["allocation_id"], allocation_id);
let removed = request_json(app, "DELETE", &detail_path, None, StatusCode::OK).await;
assert_eq!(removed["item"]["status"], "removed");
}
#[tokio::test]
async fn runtime_connection_settings_add_delete_apply_live_registry() {
let dir = tempfile::tempdir().unwrap();
@@ -4005,6 +4299,7 @@ mod tests {
initial_input: None,
execution_workspace: None,
resolved_execution_workspace: None,
resolved_execution_workspace_allocation: None,
},
)
.expect("spawn worker");