diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 17dba8b22..aec0124a8 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -7604,6 +7604,9 @@ pub struct CurrentWorkerWorkdirAttachmentResponse { #[serde(skip_serializing_if = "Option::is_none")] pub environment: Option, pub attached: bool, + /// The Workdir was already attached to this Worker; alias is its existing name. + #[serde(default)] + pub already_attached: bool, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] @@ -13384,9 +13387,11 @@ mod tests { capabilities: workdir::WorkdirSessionCapabilities::EMPTY, environment: None, attached: false, + already_attached: false, }; let value = serde_json::to_value(&response).unwrap(); - assert_eq!(value.as_object().unwrap().len(), 5); + assert_eq!(value.as_object().unwrap().len(), 6); + assert_eq!(value["already_attached"], false); assert!(value.get("connection_id").is_none()); assert_eq!( serde_json::from_value::(value).unwrap(), diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index a197867e2..66a9d5601 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -702,31 +702,46 @@ impl WorkspaceHttpWorkdirBackend { ATTACH_TOOL, "working_directory_id", )?; - let alias_key = workdir::WorkdirAttachmentAlias::new(alias) - .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; - if self.session_router.session(&alias_key).is_some() { - return Err(ToolError::InvalidArgument(format!( - "attachment alias `{alias}` already exists" - ))); - } let response = self.attach_response(alias, working_directory_id)?; - if let Err(error) = self.session_router.attach( - alias_key.clone(), + if !response.attachment.attached + || response.attachment.working_directory_id != working_directory_id + { + return Err(ToolError::ExecutionFailed( + "Workdir attachment response does not match the target".into(), + )); + } + // Backend ownership decides the name. A requested alias cannot rename + // an existing attachment or create a second route to the same Workdir. + let attached_alias = &response.attachment.alias; + let alias_key = workdir::WorkdirAttachmentAlias::new(attached_alias.clone()) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + if let Some(session) = self.session_router.session(&alias_key) { + if session.workdir().id().as_str() != working_directory_id { + return Err(ToolError::ExecutionFailed("Local Workdir attachment differs from Backend; refresh attachments before retrying".into())); + } + } else if let Err(error) = self.session_router.attach( + alias_key, WorkspaceAttachedWorkdirSession::handle_for_workdir_with_environment( self.client.clone(), - alias, + attached_alias, working_directory_id, response.attachment.capabilities, response.environment.clone(), ), ) { - let _ = self.detach(alias); + if !response.attachment.already_attached { + let _ = self.detach(attached_alias); + } return Err(ToolError::ExecutionFailed(format!( "attach Workdir session route: {error}" ))); } workdir_output( - format!("Attached Workdir {working_directory_id} as {alias}"), + if response.attachment.already_attached { + format!("Workdir {working_directory_id} is already attached as {attached_alias}") + } else { + format!("Attached Workdir {working_directory_id} as {attached_alias}") + }, &response, ) } @@ -737,15 +752,31 @@ impl WorkspaceHttpWorkdirBackend { working_directory_id: &str, ) -> Result { let workspace_id = encode_path_segment(self.workspace_id()?); - self.execute_json::(WorkspaceRequest::json( - WorkspaceRequestMethod::Post, - format!("/api/w/{workspace_id}/workers/self/workdir-attachments"), - serde_json::to_string(&WorkdirAttachRequest { - alias: alias.to_string(), - working_directory_id: working_directory_id.to_string(), - }) - .map_err(decode_error)?, - )) + let response = self + .client + .execute(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + format!("/api/w/{workspace_id}/workers/self/workdir-attachments"), + serde_json::to_string(&WorkdirAttachRequest { + alias: alias.to_string(), + working_directory_id: working_directory_id.to_string(), + }) + .map_err(decode_error)?, + )) + .map_err(|error| { + ToolError::ExecutionFailed(format!("Workspace Workdir request failed: {error}")) + })?; + if response.status == 409 + && let Ok(error) = + serde_json::from_str::(&response.body) + && error.error == "workdir_occupied" + { + return Err(ToolError::StructuredConflict { + code: "workdir_occupied".into(), + message: "Workdir is occupied by another Worker".into(), + }); + } + decode_response(response) } fn detach(&self, alias: &str) -> Result { @@ -1231,6 +1262,8 @@ struct WorkdirAttachmentResponse { working_directory_id: String, capabilities: WorkdirSessionCapabilities, attached: bool, + #[serde(default)] + already_attached: bool, } #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] @@ -1701,6 +1734,98 @@ mod tests { ); } + #[test] + fn attach_reports_other_worker_occupancy_without_exposing_backend_details() { + let client = Arc::new(RecordingWorkspaceClient::new(vec![WorkspaceResponse { + status: 409, + body: serde_json::to_string(&server_api::RepositoryApiError::new( + 409, + "workdir_occupied", + "private backend detail", + Vec::new(), + )) + .unwrap(), + }])); + let backend = WorkspaceHttpWorkdirBackend::new(client.clone()); + let error = backend + .attach(WorkdirAttachInput { + alias: "work".into(), + working_directory_id: "occupied".into(), + }) + .unwrap_err(); + assert!( + matches!(error, ToolError::StructuredConflict { code, message } if code == "workdir_occupied" && message == "Workdir is occupied by another Worker") + ); + assert_eq!(client.requests().len(), 1); + } + + #[test] + fn attach_reuses_backend_alias_without_duplicate_routes_or_detach() { + let client = Arc::new(RecordingWorkspaceClient::new((0..3).map(|_| response(json!({ + "workspace_id": "workspace/test", "alias": "existing", "working_directory_id": "wd-existing", + "capabilities": {"bits": 63}, "attached": true, "already_attached": true, + "environment": workdir::WorkdirSessionEnvironment::root(WorkdirSessionCapabilities::ALL) + }))).collect())); + let backend = WorkspaceHttpWorkdirBackend::new(client.clone()); + let occupied_alias = workdir::WorkdirAttachmentAlias::new("other").unwrap(); + backend + .session_router + .attach( + occupied_alias.clone(), + WorkspaceAttachedWorkdirSession::handle_for_workdir( + client.clone(), + "other", + "wd-other", + ), + ) + .unwrap(); + for alias in ["new-name", "existing", "other"] { + let output = backend + .attach(WorkdirAttachInput { + alias: alias.into(), + working_directory_id: "wd-existing".into(), + }) + .unwrap(); + assert_eq!( + output.summary, + "Workdir wd-existing is already attached as existing" + ); + let body: serde_json::Value = + serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); + assert_eq!(body["alias"], "existing"); + assert_eq!(body["already_attached"], true); + } + assert!( + backend + .session_router + .session(&workdir::WorkdirAttachmentAlias::new("new-name").unwrap()) + .is_none() + ); + assert_eq!( + backend + .session_router + .session(&occupied_alias) + .unwrap() + .workdir() + .id() + .as_str(), + "wd-other" + ); + assert!( + backend + .session_router + .session(&workdir::WorkdirAttachmentAlias::new("existing").unwrap()) + .is_some() + ); + assert_eq!(client.requests().len(), 3); + assert!( + client + .requests() + .iter() + .all(|request| request.method == WorkspaceRequestMethod::Post) + ); + } + #[tokio::test] async fn attached_session_proxies_operations_without_runtime_or_session_arguments() { let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({ diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index c4a95e8c0..c5e72d17d 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -134,6 +134,12 @@ pub enum Error { TicketAssignmentConflict(String), #[error("Workdir attachment conflict: {0}")] WorkdirAttachmentConflict(String), + #[error("Workdir {workdir_id} is already attached to Worker {runtime_id}:{worker_id}")] + WorkdirOccupied { + workdir_id: String, + runtime_id: String, + worker_id: String, + }, #[error("Workspace permission denied: {0}")] WorkspacePermissionDenied(String), #[error("Workspace config update conflict: {0}")] diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index dce299e60..1d55a3735 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -21001,15 +21001,17 @@ async fn scoped_attach_current_worker_workdir( run_worker_workdir_db_stage_locked(&api, &_session_guard, move |api| -> ApiResult<_> { let existing = api .store - .list_worker_workdir_links(&api.config.workspace_id, &stage_worker)? + .list_workdir_worker_links(&api.config.workspace_id, &stage_workdir_id)? .into_iter() - .find(|link| link.unlinked_at.is_none() && link.alias == stage_alias); + .next(); if let Some(existing) = existing { - if existing.workdir_id != stage_workdir_id { - return Err(Error::WorkdirAttachmentConflict(format!( - "Worker {}:{} attachment alias `{stage_alias}` is already bound to Workdir {}", - stage_worker.runtime_id, stage_worker.worker_id, existing.workdir_id - )).into()); + if existing.worker != stage_worker { + return Err(Error::WorkdirOccupied { + workdir_id: stage_workdir_id, + runtime_id: existing.worker.runtime_id, + worker_id: existing.worker.worker_id, + } + .into()); } let effective_capabilities = existing.capabilities.intersection(capabilities); if effective_capabilities != existing.capabilities { @@ -21019,6 +21021,21 @@ async fn scoped_attach_current_worker_workdir( } Ok(Some((existing, effective_capabilities))) } else { + if let Some(existing) = api + .store + .list_worker_workdir_links(&api.config.workspace_id, &stage_worker)? + .into_iter() + .find(|link| link.unlinked_at.is_none() && link.alias == stage_alias) + { + return Err(Error::WorkdirAttachmentConflict(format!( + "Worker {}:{} attachment alias '{}' is already bound to Workdir {}", + stage_worker.runtime_id, + stage_worker.worker_id, + stage_alias, + existing.workdir_id + )) + .into()); + } Ok(None) } }) @@ -21031,7 +21048,7 @@ async fn scoped_attach_current_worker_workdir( .map_err(ApiError::from)?; return Ok(Json(server_api::CurrentWorkerWorkdirAttachmentResponse { workspace_id: api.config.workspace_id.clone(), - alias: alias.to_string(), + alias: link.alias.clone(), working_directory_id: workdir_id.to_string(), capabilities: effective_capabilities, environment: Some( @@ -21046,6 +21063,7 @@ async fn scoped_attach_current_worker_workdir( .await?, ), attached: true, + already_attached: true, })); } close_current_worker_attachment_session_locked(&api, &worker).await?; @@ -21107,6 +21125,7 @@ async fn scoped_attach_current_worker_workdir( capabilities, environment: Some(environment), attached: true, + already_attached: false, })) } @@ -21350,6 +21369,7 @@ async fn detach_current_worker_workdir( capabilities: workdir::WorkdirSessionCapabilities::EMPTY, environment: None, attached: false, + already_attached: false, }) } @@ -37665,6 +37685,7 @@ impl ApiError { status.as_u16(), match self.error { Error::RestoreObservationConflict => "restore_observation_conflict", + Error::WorkdirOccupied { .. } => "workdir_occupied", Error::RuntimeBindingIdConflict { .. } => "stale_binding", Error::RuntimeBindingFingerprintConflict { .. } => "fingerprint_in_use", _ => status.canonical_reason().unwrap_or("error"), @@ -37688,6 +37709,7 @@ fn api_error_status(error: &Error) -> StatusCode { } Error::TicketAssignmentConflict(_) | Error::WorkdirAttachmentConflict(_) + | Error::WorkdirOccupied { .. } | Error::WorkspaceConfigConflict(_) | Error::RuntimeBindingConflict(_) | Error::RuntimeBindingIdConflict { .. } @@ -61386,6 +61408,130 @@ mod tests { ); } + #[tokio::test] + async fn current_worker_attach_uses_workdir_ownership_before_requested_alias() { + let mut fixture = manual_worker_assignment_fixture().await; + let identity = + worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id) + .unwrap(); + configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id); + let api = &fixture.api; + let before = api + .store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker) + .unwrap(); + let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments"); + for alias in ["docs", "different-name", "checkout"] { + let response = build_router(api.clone()) + .oneshot(runtime_source_request( + &identity, + Some(&fixture.worker.worker_id), + "POST", + &path, + serde_json::to_vec( + &json!({"alias": alias, "working_directory_id": fixture.docs_workdir_id}), + ) + .unwrap(), + )) + .await + .unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(status, StatusCode::OK, "{}", String::from_utf8_lossy(&body)); + let response: server_api::CurrentWorkerWorkdirAttachmentResponse = + serde_json::from_slice(&body).unwrap(); + assert!(response.attached && response.already_attached); + assert_eq!(response.alias, "docs"); + assert_eq!(response.working_directory_id, fixture.docs_workdir_id); + assert!(response.environment.is_some()); + assert_eq!( + api.store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker) + .unwrap(), + before + ); + } + api.store + .detach_worker_workdir( + TEST_WORKSPACE_ID, + &fixture.worker, + Some(&fixture.docs_workdir_id), + "test detach", + ) + .unwrap(); + sync_runtime_worker_workdir_attachments(api, &fixture.worker).unwrap(); + for (alias, expected) in [ + ("checkout", StatusCode::CONFLICT), + ("new-docs", StatusCode::OK), + ] { + let response = build_router(api.clone()) + .oneshot(runtime_source_request( + &identity, + Some(&fixture.worker.worker_id), + "POST", + &path, + serde_json::to_vec( + &json!({"alias": alias, "working_directory_id": fixture.docs_workdir_id}), + ) + .unwrap(), + )) + .await + .unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(status, expected, "{}", String::from_utf8_lossy(&body)); + if status == StatusCode::OK { + let response: server_api::CurrentWorkerWorkdirAttachmentResponse = + serde_json::from_slice(&body).unwrap(); + assert!(response.attached); + assert!(!response.already_attached); + assert_eq!(response.alias, "new-docs"); + } + } + let mut other = api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &fixture.worker) + .unwrap() + .unwrap(); + other.worker.worker_id = "another-owner".into(); + api.store.upsert_worker_registry(&other).unwrap(); + let mut occupied = api + .store + .list_workdir_worker_links(TEST_WORKSPACE_ID, &fixture.docs_workdir_id) + .unwrap() + .pop() + .unwrap(); + api.store + .detach_worker_workdir( + TEST_WORKSPACE_ID, + &fixture.worker, + Some(&fixture.docs_workdir_id), + "transfer fixture", + ) + .unwrap(); + sync_runtime_worker_workdir_attachments(api, &fixture.worker).unwrap(); + occupied.worker = other.worker; + api.store.attach_worker_workdir(&occupied).unwrap(); + let response = build_router(api.clone()) + .oneshot(runtime_source_request( + &identity, + Some(&fixture.worker.worker_id), + "POST", + &path, + serde_json::to_vec( + &json!({"alias": "checkout", "working_directory_id": fixture.docs_workdir_id}), + ) + .unwrap(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::CONFLICT); + let error: server_api::RepositoryApiError = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(error.error, "workdir_occupied"); + } + #[tokio::test] async fn current_worker_attachment_list_and_stale_detach_use_signed_identity() { let mut fixture = manual_worker_assignment_fixture().await; diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 85e1f24cd..68860e422 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -11119,49 +11119,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { record.workdir_id ))); } - let active_for_alias = tx - .query_row( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id - FROM worker_workdir_links - WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 - AND alias = ?4 AND unlinked_at IS NULL"#, - params![ - record.workspace_id, - record.worker.runtime_id, - record.worker.worker_id, - record.alias, - ], - read_worker_workdir_link_record, - ) - .optional()?; - if let Some(mut active) = active_for_alias { - if active.workdir_id == record.workdir_id { - let capabilities = active.capabilities.intersection(record.capabilities); - if capabilities != active.capabilities { - tx.execute( - r#"UPDATE worker_workdir_links - SET capabilities = ?1 - WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4 - AND workdir_id = ?5 AND alias = ?6 AND unlinked_at IS NULL"#, - params![ - encode_workdir_link_capabilities(capabilities)?, - record.workspace_id, - record.worker.runtime_id, - record.worker.worker_id, - record.workdir_id, - record.alias, - ], - )?; - active.capabilities = capabilities; - } - tx.commit()?; - return Ok(active); - } - return Err(Error::WorkdirAttachmentConflict(format!( - "Worker {}:{} attachment alias `{}` is already bound to Workdir {}", - record.worker.runtime_id, record.worker.worker_id, record.alias, active.workdir_id - ))); - } + // Ownership is keyed by Workdir, not by the requested local name. let active_for_workdir = tx .query_row( r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id @@ -11171,10 +11129,45 @@ impl ControlPlaneStore for SqliteWorkspaceStore { read_worker_workdir_link_record, ) .optional()?; - if let Some(active) = active_for_workdir { + if let Some(mut active) = active_for_workdir { + if active.worker != record.worker { + return Err(Error::WorkdirOccupied { + workdir_id: record.workdir_id.clone(), + runtime_id: active.worker.runtime_id, + worker_id: active.worker.worker_id, + }); + } + let capabilities = active.capabilities.intersection(record.capabilities); + if capabilities != active.capabilities { + tx.execute( + r#"UPDATE worker_workdir_links SET capabilities = ?1 + WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4 + AND workdir_id = ?5 AND alias = ?6 AND unlinked_at IS NULL"#, + params![ + encode_workdir_link_capabilities(capabilities)?, + active.workspace_id, active.worker.runtime_id, active.worker.worker_id, + active.workdir_id, active.alias, + ], + )?; + active.capabilities = capabilities; + } + tx.commit()?; + return Ok(active); + } + let active_for_alias = tx + .query_row( + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id + FROM worker_workdir_links + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 + AND alias = ?4 AND unlinked_at IS NULL"#, + params![record.workspace_id, record.worker.runtime_id, record.worker.worker_id, record.alias], + read_worker_workdir_link_record, + ) + .optional()?; + if let Some(active) = active_for_alias { return Err(Error::WorkdirAttachmentConflict(format!( - "Workdir {} is already attached to Worker {}:{}", - record.workdir_id, active.worker.runtime_id, active.worker.worker_id + "Worker {}:{} attachment alias `{}` is already bound to Workdir {}", + record.worker.runtime_id, record.worker.worker_id, record.alias, active.workdir_id ))); } // Direct attachments (including ordinary recovery/reconnection) @@ -25496,6 +25489,24 @@ INSERT INTO worker_registry ( 2 ); + for alias in ["another-name", "docs"] { + let requested = WorkerWorkdirLinkRecord { + alias: alias.into(), + ..link.clone() + }; + assert_eq!( + store.attach_worker_workdir(&requested).unwrap(), + downgraded_link + ); + assert_eq!( + store + .list_worker_workdir_links("local-dev", &worker.worker) + .unwrap() + .len(), + 2 + ); + } + let second_worker = WorkerRegistryRecord { worker: RuntimeWorkerRef::new("embedded", "2"), display_name: "Browser 2".to_string(), @@ -25511,7 +25522,7 @@ INSERT INTO worker_registry ( }; assert!(matches!( store.attach_worker_workdir(&workdir_conflict), - Err(Error::WorkdirAttachmentConflict(_)) + Err(Error::WorkdirOccupied { .. }) )); assert!(matches!( store.detach_worker_workdir("local-dev", &worker.worker, Some("wrong-workdir"), "6",),