feat: publish Repository and Workdir WIP objects
Use Host root namespace ownership and disjoint Feature operation contributions. Separate read-only workdir_catalog from manage_workdir, encode all existing identities canonically, publish safe bounded inventory and lifetime-bound attachment Objects, and reuse normal lifecycle/session router/lock authority. Contextual interfaces resolve the current permission/capability intersection; object and interface validators reject stale cached access. Cover fresh workspace WIP-only discovery/create/attach/read/detach, readonly and unsupported targets, collision/revocation/alias reuse, active cleanup exclusion and terminal unknown outcomes without retries. Validation: WIP 40 tests; worker 730 unit + 96 integration tests serial; manifest 182 tests; workspace-root cargo check; cargo fmt --all -- --check; git diff --check HEAD.
This commit is contained in:
@@ -1504,7 +1504,13 @@ where
|
||||
),
|
||||
);
|
||||
}
|
||||
if feature_config.manage_workdir.enabled {
|
||||
let wip_mode = worker.manifest().worker.mode == manifest::WorkerMode::Wip;
|
||||
let mut workdir_wip_feature = None;
|
||||
if feature_config.manage_workdir.enabled
|
||||
|| (wip_mode
|
||||
&& feature_config.workdir_catalog.enabled
|
||||
&& worker.workspace_client_handle().is_available())
|
||||
{
|
||||
// Workdir lifecycle is Workspace control-plane authority. The Worker
|
||||
// receives only the injected WorkspaceClient and never Runtime URLs,
|
||||
// repository paths, materializer handles, or cleanup sessions.
|
||||
@@ -1520,7 +1526,7 @@ where
|
||||
}
|
||||
let shutdown_registry = spawned_registry.clone();
|
||||
let reopen_registry = spawned_registry.clone();
|
||||
feature_registry.add_module(
|
||||
let module =
|
||||
crate::feature::builtin::manage_workdir::ManageWorkdirFeature::with_child_lifecycle(
|
||||
workspace_client,
|
||||
worker.workdir_sessions(),
|
||||
@@ -1529,8 +1535,11 @@ where
|
||||
Box::pin(async move { child_registry.shutdown_internal().await })
|
||||
}),
|
||||
Arc::new(move || reopen_registry.reopen_internal()),
|
||||
),
|
||||
);
|
||||
);
|
||||
if feature_config.manage_workdir.enabled {
|
||||
feature_registry.add_module(module.clone());
|
||||
}
|
||||
workdir_wip_feature = Some(module);
|
||||
}
|
||||
if feature_config.workspace_worker_discovery.enabled {
|
||||
let workspace_client = worker.workspace_client_handle();
|
||||
@@ -1585,9 +1594,23 @@ where
|
||||
}
|
||||
|
||||
let host_worker_observation_provider = worker.worker_observation_provider();
|
||||
let wip_mode = worker.manifest().worker.mode == manifest::WorkerMode::Wip;
|
||||
let wip_permissions = worker.manifest().permissions.clone();
|
||||
let mut wip_mount_registry = crate::wip::WipMountRegistry::new();
|
||||
if wip_mode && let Some(feature) = &workdir_wip_feature {
|
||||
crate::feature::builtin::manage_workdir::wip::mount_workspace_workdir_wip(
|
||||
&mut wip_mount_registry,
|
||||
feature,
|
||||
feature_config.workdir_catalog.enabled,
|
||||
feature_config.manage_workdir.enabled,
|
||||
wip_permissions.clone(),
|
||||
)
|
||||
.map_err(|error| {
|
||||
std::io::Error::new(
|
||||
std::io::ErrorKind::InvalidInput,
|
||||
format!("mount Workdir WIP projection: {error}"),
|
||||
)
|
||||
})?;
|
||||
}
|
||||
{
|
||||
let workspace_client = worker.workspace_client_handle();
|
||||
let engine = worker.engine_mut();
|
||||
|
||||
@@ -5,6 +5,8 @@
|
||||
//! endpoints, credentials, materializer handles, and operation sessions stay
|
||||
//! behind [`WorkspaceClient`].
|
||||
|
||||
pub mod wip;
|
||||
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
@@ -67,6 +69,7 @@ pub struct ManageWorkdirFeature {
|
||||
session_router: Arc<workdir::WorkdirSessionRouter>,
|
||||
before_workdir_release: Option<BeforeWorkdirRelease>,
|
||||
after_workdir_attach: Option<AfterWorkdirAttach>,
|
||||
attachment_mutation_lock: Arc<tokio::sync::Mutex<()>>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for ManageWorkdirFeature {
|
||||
@@ -86,6 +89,7 @@ impl ManageWorkdirFeature {
|
||||
session_router: Arc::new(workdir::WorkdirSessionRouter::new()),
|
||||
before_workdir_release: None,
|
||||
after_workdir_attach: None,
|
||||
attachment_mutation_lock: Arc::new(tokio::sync::Mutex::new(())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -100,8 +104,19 @@ impl ManageWorkdirFeature {
|
||||
session_router,
|
||||
before_workdir_release: Some(before_workdir_release),
|
||||
after_workdir_attach: Some(after_workdir_attach),
|
||||
attachment_mutation_lock: Arc::new(tokio::sync::Mutex::new(())),
|
||||
}
|
||||
}
|
||||
fn backend(&self) -> WorkspaceHttpWorkdirBackend {
|
||||
let mut backend = WorkspaceHttpWorkdirBackend::new(self.client.clone())
|
||||
.with_session_router(self.session_router.clone())
|
||||
.with_child_lifecycle(
|
||||
self.before_workdir_release.clone(),
|
||||
self.after_workdir_attach.clone(),
|
||||
);
|
||||
backend.attachment_mutation_lock = self.attachment_mutation_lock.clone();
|
||||
backend
|
||||
}
|
||||
}
|
||||
|
||||
pub fn manage_workdir_feature(client: Arc<dyn WorkspaceClient>) -> ManageWorkdirFeature {
|
||||
@@ -120,12 +135,7 @@ impl FeatureModule for ManageWorkdirFeature {
|
||||
}
|
||||
|
||||
fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
|
||||
let backend = WorkspaceHttpWorkdirBackend::new(self.client.clone())
|
||||
.with_session_router(self.session_router.clone())
|
||||
.with_child_lifecycle(
|
||||
self.before_workdir_release.clone(),
|
||||
self.after_workdir_attach.clone(),
|
||||
);
|
||||
let backend = self.backend();
|
||||
for (name, definition) in [
|
||||
(
|
||||
LIST_TOOL,
|
||||
@@ -581,11 +591,24 @@ impl WorkspaceHttpWorkdirBackend {
|
||||
}
|
||||
|
||||
fn detach(&self, alias: &str) -> Result<ToolOutput, ToolError> {
|
||||
self.detach_expected(alias, None)
|
||||
}
|
||||
|
||||
fn detach_expected(
|
||||
&self,
|
||||
alias: &str,
|
||||
expected_connection_id: Option<&str>,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let workspace_id = encode_path_segment(self.workspace_id()?);
|
||||
let alias_path = encode_path_segment(alias);
|
||||
let suffix = expected_connection_id
|
||||
.map(|id| format!("?expected_connection_id={}", encode_path_segment(id)))
|
||||
.unwrap_or_default();
|
||||
let response = self.execute_json::<WorkdirAttachmentResponse>(WorkspaceRequest {
|
||||
method: WorkspaceRequestMethod::Delete,
|
||||
path: format!("/api/w/{workspace_id}/workers/self/workdir-attachments/{alias_path}"),
|
||||
path: format!(
|
||||
"/api/w/{workspace_id}/workers/self/workdir-attachments/{alias_path}{suffix}"
|
||||
),
|
||||
body: None,
|
||||
})?;
|
||||
workdir_output(
|
||||
@@ -622,6 +645,72 @@ impl WorkspaceHttpWorkdirBackend {
|
||||
)
|
||||
}
|
||||
|
||||
async fn detach_managed(
|
||||
&self,
|
||||
alias: &str,
|
||||
expected_connection_id: Option<&str>,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let _mutation_guard = self.attachment_mutation_lock.lock().await;
|
||||
let alias = validate_identity(alias, DETACH_TOOL, "alias")?;
|
||||
if let Some(expected) = expected_connection_id {
|
||||
// Under the shared local mutation lock, check lifetime before any
|
||||
// local route exclusion or child cleanup. Backend DELETE still performs
|
||||
// the authoritative conditional mutation under its lifecycle lock.
|
||||
let workspace = encode_path_segment(self.workspace_id()?);
|
||||
let page: serde_json::Value = self.execute_json(WorkspaceRequest::get(format!(
|
||||
"/api/w/{workspace}/workers/self/workdir-attachments?limit=1&connection_id={}",
|
||||
encode_path_segment(expected)
|
||||
)))?;
|
||||
let current = page
|
||||
.get("items")
|
||||
.and_then(serde_json::Value::as_array)
|
||||
.is_some_and(|items| {
|
||||
items.iter().any(|item| {
|
||||
item.get("connection_id")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
== Some(expected)
|
||||
&& item.get("alias").and_then(serde_json::Value::as_str) == Some(alias)
|
||||
})
|
||||
});
|
||||
if !current {
|
||||
return Err(ToolError::StructuredConflict {
|
||||
code: "workspace_http_409".into(),
|
||||
message: "attachment lifetime changed before local cleanup".into(),
|
||||
});
|
||||
}
|
||||
}
|
||||
let alias_key = workdir::WorkdirAttachmentAlias::new(alias)
|
||||
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
|
||||
self.session_router
|
||||
.begin_detach(&alias_key)
|
||||
.await
|
||||
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
|
||||
if let Some(before_release) = &self.before_workdir_release
|
||||
&& let Err(error) = before_release().await
|
||||
{
|
||||
self.session_router.cancel_detach(&alias_key);
|
||||
return Err(ToolError::ExecutionFailed(format!(
|
||||
"stop Internal SubWorkers before Workdir detach: {error}"
|
||||
)));
|
||||
}
|
||||
let result = match self.detach_expected(alias, expected_connection_id) {
|
||||
Ok(result) => result,
|
||||
Err(error) => {
|
||||
self.session_router.cancel_detach(&alias_key);
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
self.session_router
|
||||
.finish_detach(&alias_key)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
ToolError::ExecutionFailed(format!(
|
||||
"detach Workdir session route after Backend release: {error}"
|
||||
))
|
||||
})?;
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
fn execute_json<T: for<'de> Deserialize<'de>>(
|
||||
&self,
|
||||
request: WorkspaceRequest,
|
||||
@@ -697,42 +786,10 @@ impl Tool for WorkspaceHttpWorkdirTool {
|
||||
result
|
||||
}
|
||||
WorkdirOperation::Detach => {
|
||||
let _mutation_guard = self.backend.attachment_mutation_lock.lock().await;
|
||||
let input = parse_input::<WorkdirDetachInput>(input_json)?;
|
||||
let alias = validate_identity(&input.alias, DETACH_TOOL, "alias")?;
|
||||
let alias_key = workdir::WorkdirAttachmentAlias::new(alias)
|
||||
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
|
||||
self.backend
|
||||
.session_router
|
||||
.begin_detach(&alias_key)
|
||||
.await
|
||||
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
|
||||
if let Some(before_release) = &self.backend.before_workdir_release
|
||||
&& let Err(error) = before_release().await
|
||||
{
|
||||
self.backend.session_router.cancel_detach(&alias_key);
|
||||
return Err(ToolError::ExecutionFailed(format!(
|
||||
"stop Internal SubWorkers before Workdir detach: {error}"
|
||||
)));
|
||||
}
|
||||
let result = match self.backend.detach(alias) {
|
||||
Ok(result) => result,
|
||||
Err(error) => {
|
||||
self.backend.session_router.cancel_detach(&alias_key);
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
self.backend
|
||||
.session_router
|
||||
.finish_detach(&alias_key)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
ToolError::ExecutionFailed(format!(
|
||||
"detach Workdir session route after Backend release: {error}"
|
||||
))
|
||||
})?;
|
||||
Ok(result)
|
||||
self.backend.detach_managed(&input.alias, None).await
|
||||
}
|
||||
|
||||
WorkdirOperation::Delete => self
|
||||
.backend
|
||||
.delete(parse_input::<WorkdirDeleteInput>(input_json)?),
|
||||
@@ -747,6 +804,12 @@ fn parse_input<T: for<'de> Deserialize<'de>>(input: &str) -> Result<T, ToolError
|
||||
fn decode_response<T: for<'de> Deserialize<'de>>(
|
||||
response: WorkspaceResponse,
|
||||
) -> Result<T, ToolError> {
|
||||
if matches!(response.status, 401 | 403 | 404 | 409) {
|
||||
return Err(ToolError::StructuredConflict {
|
||||
code: format!("workspace_http_{}", response.status),
|
||||
message: "Workspace lifecycle authority rejected the request before execution".into(),
|
||||
});
|
||||
}
|
||||
if !response.is_success() {
|
||||
return Err(ToolError::ExecutionFailed(format!(
|
||||
"Workspace Workdir API returned HTTP {}: {}",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
+1260
-25
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user