diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index edfc77409..441e13e0a 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -4472,10 +4472,10 @@ mod tests { )] ); let names = captured_tool_names(&client, 0); - for forbidden in core_filesystem_tool_names() { + for expected in core_filesystem_tool_names() { assert!( - !names.contains(forbidden), - "no-workdir Worker unexpectedly exposed {forbidden}; tools={names:?}" + names.contains(expected), + "no-workdir Worker did not expose stable routed tool {expected}; tools={names:?}" ); } let observations = runtime diff --git a/crates/worker/src/feature/builtin/manage_worker.rs b/crates/worker/src/feature/builtin/manage_worker.rs index 1ea1950de..1f9ced49b 100644 --- a/crates/worker/src/feature/builtin/manage_worker.rs +++ b/crates/worker/src/feature/builtin/manage_worker.rs @@ -5,10 +5,14 @@ use std::sync::Arc; use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; use async_trait::async_trait; use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use uuid::Uuid; use protocol::Segment; +use server_api::{ + BrowserWorkerWorkingDirectorySelection, CreateWorkspaceWorkerRequest, + CreateWorkspaceWorkerTicketAssignmentRequest, +}; use crate::feature::{ FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, @@ -27,6 +31,7 @@ const FEATURE_DESCRIPTION: &str = pub const WORKER_LIFECYCLE_SERVICE_ID: &str = "worker.lifecycle"; pub const WORKER_CONTROL_SERVICE_ID: &str = "worker.control"; const WORKER_LIFECYCLE_SERVICE_VERSION: &str = "1"; +const INITIAL_WORKDIR_ALIAS: &str = "workdir"; #[derive(Debug, Clone, PartialEq, Eq)] pub enum WorkerControlSubject { @@ -457,40 +462,49 @@ struct WorkspaceWorkerLifecycleService { workspace_id: String, } +fn workspace_worker_create_request( + request: WorkerLifecycleSpawnRequest, +) -> Result { + let ticket_assignment = match (request.ticket_id, request.operation_id) { + (Some(ticket_id), Some(operation_id)) => { + Some(CreateWorkspaceWorkerTicketAssignmentRequest { + ticket_id, + operation_id, + }) + } + (None, None) => None, + _ => { + return Err(WorkspaceClientError::Request( + "ticket_id and operation_id must be provided together".to_string(), + )); + } + }; + let control_operation_id = ticket_assignment + .as_ref() + .map(|assignment| assignment.operation_id.clone()) + .unwrap_or_else(|| format!("worker-spawn-{}", Uuid::now_v7())); + Ok(CreateWorkspaceWorkerRequest { + runtime_id: request.runtime_id, + display_name: request.display_name, + profile: Some(request.profile), + ticket_assignment, + initial_submit: request.initial_submit, + workdir_attachments: vec![BrowserWorkerWorkingDirectorySelection { + alias: INITIAL_WORKDIR_ALIAS.to_string(), + working_directory_id: request.working_directory_id, + relative_cwd: request.relative_cwd, + }], + control_operation_id: Some(control_operation_id), + }) +} + #[async_trait] impl WorkerLifecycleService for WorkspaceWorkerLifecycleService { async fn spawn( &self, request: WorkerLifecycleSpawnRequest, ) -> Result { - let ticket_assignment = match (request.ticket_id, request.operation_id) { - (Some(ticket_id), Some(operation_id)) => Some(WorkerSpawnTicketAssignmentRequest { - ticket_id, - operation_id, - }), - (None, None) => None, - _ => { - return Err(WorkspaceClientError::Request( - "ticket_id and operation_id must be provided together".to_string(), - )); - } - }; - let control_operation_id = ticket_assignment - .as_ref() - .map(|assignment| assignment.operation_id.clone()) - .unwrap_or_else(|| format!("worker-spawn-{}", Uuid::now_v7())); - let body = WorkerSpawnRequest { - runtime_id: request.runtime_id, - display_name: request.display_name, - profile: request.profile, - control_operation_id, - ticket_assignment, - initial_submit: request.initial_submit, - working_directory: WorkerWorkingDirectorySelection { - working_directory_id: request.working_directory_id, - relative_cwd: request.relative_cwd, - }, - }; + let body = workspace_worker_create_request(request)?; self.client.execute(WorkspaceRequest::json( WorkspaceRequestMethod::Post, format!("/api/w/{}/worker-control/workers", self.workspace_id), @@ -694,31 +708,6 @@ struct WorkerSpawnInput { relative_cwd: Option, } -#[derive(Debug, Serialize)] -struct WorkerSpawnTicketAssignmentRequest { - ticket_id: String, - operation_id: String, -} - -#[derive(Debug, Serialize)] -struct WorkerSpawnRequest { - runtime_id: String, - display_name: String, - profile: String, - control_operation_id: String, - #[serde(skip_serializing_if = "Option::is_none")] - ticket_assignment: Option, - initial_submit: Vec, - working_directory: WorkerWorkingDirectorySelection, -} - -#[derive(Debug, Serialize)] -struct WorkerWorkingDirectorySelection { - working_directory_id: String, - #[serde(skip_serializing_if = "Option::is_none")] - relative_cwd: Option, -} - #[derive(Debug, Clone, Deserialize, JsonSchema)] #[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] enum WorkerSubjectInput { @@ -1445,7 +1434,23 @@ mod tests { "operation_id": "worker-spawn:00001KZ9E0DBS:call-1" }) ); + assert_eq!(body["workdir_attachments"].as_array().unwrap().len(), 1); + assert_eq!( + body["workdir_attachments"][0]["alias"], + INITIAL_WORKDIR_ALIAS + ); + assert_eq!( + body["workdir_attachments"][0]["working_directory_id"], + "workdir-1" + ); + assert_eq!( + body["workdir_attachments"][0]["relative_cwd"], + serde_json::Value::Null + ); + assert!(body.get("working_directory").is_none()); assert!(body.get("initial_text").is_none()); + serde_json::from_value::(body) + .expect("guarded spawn body must satisfy the strict Workspace create DTO"); } #[tokio::test] @@ -1714,30 +1719,38 @@ mod tests { } #[test] - fn worker_spawn_request_uses_authority_ids_without_runtime_paths() { - let request = WorkerSpawnRequest { + fn worker_spawn_request_uses_canonical_initial_attachment_contract() { + let request = workspace_worker_create_request(WorkerLifecycleSpawnRequest { runtime_id: "runtime-1".to_string(), - display_name: "Coder".to_string(), + working_directory_id: "wd-1".to_string(), + relative_cwd: Some("repo".to_string()), profile: "builtin:coder".to_string(), - control_operation_id: "spawn-operation-1".to_string(), - ticket_assignment: None, + ticket_id: None, + operation_id: None, + display_name: "Coder".to_string(), initial_submit: vec![ Segment::Flow { selector: "builtin:coder-review".to_string(), }, Segment::text("Implement the Ticket"), ], - working_directory: WorkerWorkingDirectorySelection { - working_directory_id: "wd-1".to_string(), - relative_cwd: Some("repo".to_string()), - }, - }; + }) + .unwrap(); let value = serde_json::to_value(request).unwrap(); assert_eq!(value["runtime_id"], "runtime-1"); - assert_eq!(value["working_directory"]["working_directory_id"], "wd-1"); + assert_eq!(value["workdir_attachments"].as_array().unwrap().len(), 1); + assert_eq!( + value["workdir_attachments"][0]["alias"], + INITIAL_WORKDIR_ALIAS + ); + assert_eq!( + value["workdir_attachments"][0]["working_directory_id"], + "wd-1" + ); + assert_eq!(value["workdir_attachments"][0]["relative_cwd"], "repo"); + assert!(value.get("working_directory").is_none()); assert!(value.get("cwd").is_none()); assert!(value.get("runtime_url").is_none()); - assert!(value["working_directory"].get("mode").is_none()); assert_eq!(value["initial_submit"][0]["kind"], "flow"); assert_eq!( value["initial_submit"][0]["selector"], diff --git a/crates/worker/src/feature/builtin/orchestration.rs b/crates/worker/src/feature/builtin/orchestration.rs index 0ad3cdb99..a16858705 100644 --- a/crates/worker/src/feature/builtin/orchestration.rs +++ b/crates/worker/src/feature/builtin/orchestration.rs @@ -260,7 +260,8 @@ mod tests { &serde_json::json!({ "ticket_id": "T-482", "runtime_id": "runtime-1", - "working_directory_id": "workdir-1" + "working_directory_id": "workdir-1", + "relative_cwd": "crates/yoi" }) .to_string(), ToolExecutionContext::new("call-7", "batch-1", 0), @@ -277,6 +278,8 @@ mod tests { Some("spawn-ticket-coder:00001KZXN51C7:call-7") ); assert_eq!(request.display_name, "Coder · T-482"); + assert_eq!(request.working_directory_id, "workdir-1"); + assert_eq!(request.relative_cwd.as_deref(), Some("crates/yoi")); assert_eq!( request.initial_submit, vec![ diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 7dba7d0b3..4f4dc7cc9 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -2635,21 +2635,13 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { }; let workspace_id = workspace_api.workspace_id.clone(); let config_bundle = spawn_config_bundle_ref(&request); - let create_request = CreateWorkerRequest { - worker_id: binding.worker_id, - create_fingerprint: binding.create_fingerprint, - profile, - display_name: request.requested_worker_name.clone(), - config_bundle, + let create_request = runtime_create_worker_request( + binding, + &request, profile_source, - initial_input: initial_worker_input(&request.initial_submit), - workdir_attachment_requests: request.resolved_workdir_attachment_requests.clone(), - workdir_attachments: request.resolved_workdir_attachments.clone(), - worker_observation_enabled: request.resolved_worker_observation_enabled, - worker_observation_grants: request.resolved_worker_observation_grants.clone(), - workspace_api: Some(workspace_api), - memory_settings: request.resolved_memory_settings.clone(), - }; + workspace_api, + config_bundle, + ); let workspace_scope = RuntimeWorkspaceScope::new(workspace_id, "embedded-backend"); match self .runtime @@ -4674,21 +4666,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } }; let config_bundle = spawn_config_bundle_ref(&request); - let create = CreateWorkerRequest { - worker_id: binding.worker_id, - create_fingerprint: binding.create_fingerprint, - profile, - display_name: request.requested_worker_name.clone(), - config_bundle, + let create = runtime_create_worker_request( + binding, + &request, profile_source, - initial_input: initial_worker_input(&request.initial_submit), - workdir_attachment_requests: request.resolved_workdir_attachment_requests.clone(), - workdir_attachments: request.resolved_workdir_attachments.clone(), - worker_observation_enabled: request.resolved_worker_observation_enabled, - worker_observation_grants: request.resolved_worker_observation_grants.clone(), - workspace_api: Some(workspace_api), - memory_settings: request.resolved_memory_settings.clone(), - }; + workspace_api, + config_bundle, + ); let create = match runtime_contract_convert(create) { Ok(create) => create, Err(diagnostic) => { @@ -5116,6 +5100,30 @@ fn remote_worker_projection_diagnostics( diagnostics } +fn runtime_create_worker_request( + binding: WorkerCreateBinding, + request: &WorkerSpawnRequest, + profile_source: ProfileSourceArchiveSource, + workspace_api: WorkspaceApiRef, + config_bundle: Option, +) -> CreateWorkerRequest { + CreateWorkerRequest { + worker_id: binding.worker_id, + create_fingerprint: binding.create_fingerprint, + profile: request.profile.clone(), + display_name: request.requested_worker_name.clone(), + config_bundle, + profile_source, + initial_input: initial_worker_input(&request.initial_submit), + workdir_attachment_requests: request.resolved_workdir_attachment_requests.clone(), + workdir_attachments: request.resolved_workdir_attachments.clone(), + worker_observation_enabled: request.resolved_worker_observation_enabled, + worker_observation_grants: request.resolved_worker_observation_grants.clone(), + workspace_api: Some(workspace_api), + memory_settings: request.resolved_memory_settings.clone(), + } +} + fn spawn_config_bundle_ref(request: &WorkerSpawnRequest) -> Option { request .resolved_config_bundle @@ -6813,6 +6821,41 @@ mod tests { assert_eq!(bundle_ref.digest, expected_digest); } + #[test] + fn runtime_create_request_serializes_only_canonical_workdir_attachments() { + let mut request = embedded_spawn_request(); + request.resolved_workdir_attachments = vec![WorkingDirectoryAttachmentClaim { + alias: workdir::WorkdirAttachmentAlias::new("workdir").unwrap(), + working_directory_id: "workdir-1".to_string(), + relative_cwd: Some("crates/yoi".to_string()), + }]; + let profile_source = + profile_source_archive_source(&request, &request.profile).expect("profile source"); + let create = runtime_create_worker_request( + test_create_binding(), + &request, + profile_source, + test_workspace_api(), + spawn_config_bundle_ref(&request), + ); + + let value = serde_json::to_value(create).unwrap(); + assert!(value.get("working_directory").is_none()); + assert!(value.get("working_directory_request").is_none()); + assert_eq!(value["workdir_attachments"].as_array().unwrap().len(), 1); + assert_eq!(value["workdir_attachments"][0]["alias"], "workdir"); + assert_eq!( + value["workdir_attachments"][0]["working_directory_id"], + "workdir-1" + ); + assert_eq!( + value["workdir_attachments"][0]["relative_cwd"], + "crates/yoi" + ); + serde_json::from_value::(value) + .expect("remote Runtime create payload must satisfy the strict DTO"); + } + #[test] fn registry_syncs_bundle_before_embedded_spawn() { let runtime = EmbeddedWorkerRuntime::new_memory_with_execution_backend( diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 7c996b441..c03a6843f 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -19569,14 +19569,55 @@ mod tests { } } - #[derive(Default)] + #[derive(Clone, Default)] struct WorkdirlessFixtureRuntime { - workers: Mutex>, + workers: Arc>>, + spawn_requests: Arc>>, + reject_next_spawn: Arc>, + workdir_repositories: Arc>>, } impl WorkdirlessFixtureRuntime { const RUNTIME_ID: &'static str = "workdirless-runtime"; + fn reject_next_spawn(&self) { + *self + .reject_next_spawn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = true; + } + + fn spawn_requests(&self) -> Vec { + self.spawn_requests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } + + fn set_workdir_repository(&self, working_directory_id: &str, repository_id: &str) { + self.workdir_repositories + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(working_directory_id.to_string(), repository_id.to_string()); + } + + fn configured_workdir_status( + &self, + working_directory_id: &str, + ) -> worker_runtime::catalog::WorkingDirectoryStatus { + let mut status = Self::workdir_status(working_directory_id); + if let Some(repository_id) = self + .workdir_repositories + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(working_directory_id) + .cloned() + { + status.summary.repository_id = repository_id; + } + status + } + fn workdir_status( working_directory_id: &str, ) -> worker_runtime::catalog::WorkingDirectoryStatus { @@ -19708,7 +19749,7 @@ mod tests { ) -> crate::hosts::RuntimeWorkingDirectoryResult { crate::hosts::RuntimeWorkingDirectoryResult { state: WorkerOperationState::Accepted, - working_directory: Some(Self::workdir_status(working_directory_id)), + working_directory: Some(self.configured_workdir_status(working_directory_id)), diagnostics: Vec::new(), } } @@ -19729,10 +19770,39 @@ mod tests { binding: WorkerCreateBinding, request: WorkerSpawnRequest, ) -> WorkerSpawnResult { + self.spawn_requests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(request.clone()); + if std::mem::take( + &mut *self + .reject_next_spawn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ) { + return WorkerSpawnResult { + state: WorkerOperationState::Rejected, + worker: None, + acceptance_evidence: Vec::new(), + diagnostics: vec![RuntimeDiagnostic { + code: "fixture_runtime_rejected".to_string(), + severity: DiagnosticSeverity::Error, + message: "fixture rejected Worker create".to_string(), + }], + }; + } assert!(request.workdir_attachment_requests.is_empty()); assert!(request.resolved_workdir_attachment_requests.is_empty()); - assert!(request.resolved_workdir_attachments.is_empty()); - let worker = Self::worker_summary(binding, &request); + let mut worker = Self::worker_summary(binding, &request); + worker.workdir_attachments = request + .resolved_workdir_attachments + .iter() + .map(|attachment| WorkingDirectoryAttachmentStatus { + alias: attachment.alias.clone(), + working_directory: self + .configured_workdir_status(&attachment.working_directory_id), + }) + .collect(); self.workers .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) @@ -21830,6 +21900,624 @@ mod tests { assert!(retried.online); } + struct GuardedSpawnFixture { + _workspace: tempfile::TempDir, + api: WorkspaceApi, + runtime: WorkdirlessFixtureRuntime, + controller: RuntimeWorkerRef, + ticket_id: String, + workdir_id: String, + } + + async fn guarded_spawn_fixture() -> GuardedSpawnFixture { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await; + let runtime = WorkdirlessFixtureRuntime::default(); + let workdir_id = "guarded-existing-workdir".to_string(); + let repository_id = test_repository_id(&api); + runtime.set_workdir_repository(&workdir_id, &repository_id); + api.runtime.register_or_replace(runtime.clone()); + let now = now_registry_timestamp(); + api.store + .upsert_workdir_registry(&WorkdirRegistryRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + workdir_id: workdir_id.clone(), + display_name: None, + runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.to_string(), + repository_id: repository_id.clone(), + creation_selector: Some("develop".to_string()), + creation_ref: None, + creation_tree: None, + current_selector: Some("develop".to_string()), + current_ref: None, + current_tree: None, + observed_at_epoch_seconds: None, + materialization_status: "present".to_string(), + cleanliness: "clean".to_string(), + created_at: now.clone(), + updated_at: now, + }) + .unwrap(); + let Json(controller) = create_workspace_worker( + State(api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + display_name: "Guarded spawn controller".to_string(), + profile: Some("builtin:orchestrator".to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachments: Vec::new(), + control_operation_id: None, + }), + ) + .await + .unwrap(); + let controller = RuntimeWorkerRef::new(controller.runtime_id, controller.worker_id); + let mut input = ticket::NewTicket::new("Guarded attachment spawn"); + input.workflow_state = Some(TicketWorkflowState::Queued); + input.repository_id = Some(repository_id); + input.ref_selector = Some("develop".to_string()); + let ticket_id = browser_ticket_backend(&api) + .unwrap() + .create(input) + .unwrap() + .id; + assign_test_orchestrator(&api, &ticket_id); + GuardedSpawnFixture { + _workspace: workspace, + api, + runtime, + controller, + ticket_id, + workdir_id, + } + } + + fn guarded_spawn_payload(fixture: &GuardedSpawnFixture, operation_id: &str) -> Value { + json!({ + "runtime_id": WorkdirlessFixtureRuntime::RUNTIME_ID, + "display_name": "Coder · guarded attachment", + "profile": "builtin:coder", + "ticket_assignment": { + "ticket_id": fixture.ticket_id, + "operation_id": operation_id, + }, + "initial_submit": [ + { "kind": "flow", "selector": "builtin:coder-review" }, + { "kind": "text", "content": "Implement guarded attachment Ticket." }, + ], + "workdir_attachments": [{ + "alias": "workdir", + "working_directory_id": fixture.workdir_id, + "relative_cwd": "crates/yoi", + }], + "control_operation_id": operation_id, + }) + } + + const STRICT_REMOTE_RUNTIME_ID: &str = "strict-remote-runtime"; + + #[derive(Clone)] + struct StrictRemoteSpawnState { + workdir_status: worker_runtime::catalog::WorkingDirectoryStatus, + create_requests: Arc>>, + } + + fn assert_strict_remote_request_headers(headers: &HeaderMap) { + assert_eq!( + headers + .get(worker_runtime::http_server::RUNTIME_WORKSPACE_SCOPE_HEADER) + .and_then(|value| value.to_str().ok()), + Some(TEST_WORKSPACE_ID) + ); + assert_eq!( + headers + .get(axum::http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()), + Some("Bearer workspace-server-test-runtime-token") + ); + } + + async fn strict_remote_workdir( + State(state): State, + headers: HeaderMap, + AxumPath(working_directory_id): AxumPath, + ) -> Json { + assert_strict_remote_request_headers(&headers); + assert_eq!( + state.workdir_status.summary.working_directory_id, + working_directory_id + ); + Json(json!({ "working_directory": state.workdir_status })) + } + + fn strict_remote_worker_response( + state: &StrictRemoteSpawnState, + request: &runtime_api::CreateWorkerRequest, + ) -> Value { + let attachment = &request.workdir_attachments[0]; + let profile_source = match &request.profile_source { + runtime_api::ProfileSourceArchiveSource::Embedded { archive } => { + archive.reference.clone() + } + runtime_api::ProfileSourceArchiveSource::WorkspaceConfig { archive } => archive.clone(), + }; + json!({ + "worker": { + "worker_ref": { "worker_id": request.worker_id }, + "worker_id": request.worker_id, + "status": "idle", + "execution_metadata_available": true, + "workspace_id": TEST_WORKSPACE_ID, + "workdir_attachments": [{ + "alias": attachment.alias, + "working_directory": state.workdir_status, + }], + "profile": request.profile, + "display_name": request.display_name, + "profile_source": profile_source, + "config_bundle": request.config_bundle, + } + }) + } + + async fn strict_remote_create_worker( + State(state): State, + headers: HeaderMap, + body: axum::body::Bytes, + ) -> Json { + assert_strict_remote_request_headers(&headers); + let raw: Value = serde_json::from_slice(&body).unwrap(); + assert!( + raw.get("working_directory").is_none(), + "legacy singular working_directory reached Runtime HTTP: {raw}" + ); + let request: runtime_api::CreateWorkerRequest = + serde_json::from_value(raw.clone()).unwrap(); + assert!(request.workdir_attachment_requests.is_empty()); + assert_eq!(request.workdir_attachments.len(), 1); + let attachment = &request.workdir_attachments[0]; + assert_eq!(attachment.alias.as_str(), "workdir"); + assert_eq!( + attachment.working_directory_id, + state.workdir_status.summary.working_directory_id + ); + assert_eq!(attachment.relative_cwd.as_deref(), Some("crates/yoi")); + state + .create_requests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(raw); + + Json(strict_remote_worker_response(&state, &request)) + } + + async fn strict_remote_replace_workspace_api( + State(state): State, + headers: HeaderMap, + AxumPath(worker_id): AxumPath, + Json(workspace_api): Json, + ) -> Json { + assert_strict_remote_request_headers(&headers); + assert_eq!(workspace_api.workspace_api.workspace_id, TEST_WORKSPACE_ID); + let raw = state + .create_requests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .last() + .cloned() + .expect("workspace API replacement follows Runtime create"); + let request: runtime_api::CreateWorkerRequest = serde_json::from_value(raw).unwrap(); + assert_eq!(request.worker_id.to_string(), worker_id); + Json(strict_remote_worker_response(&state, &request)) + } + + async fn post_guarded_spawn(fixture: &GuardedSpawnFixture, body: Value) -> Response { + build_inner_router(fixture.api.clone()) + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers")) + .header("content-type", "application/json") + .header("x-yoi-runtime-id", &fixture.controller.runtime_id) + .header("x-yoi-worker-id", &fixture.controller.worker_id) + .body(Body::from(body.to_string())) + .unwrap(), + ) + .await + .unwrap() + } + + fn guarded_spawn_artifact_counts(api: &WorkspaceApi, workdir_id: &str) -> (i64, i64, i64) { + api.config_store + .with_conn(|conn| { + Ok(( + conn.query_row( + "SELECT COUNT(*) FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2", + rusqlite::params![TEST_WORKSPACE_ID, WorkdirlessFixtureRuntime::RUNTIME_ID], + |row| row.get(0), + )?, + conn.query_row( + "SELECT COUNT(*) FROM worker_workdir_attachment_reservations WHERE workspace_id = ?1 AND workdir_id = ?2", + rusqlite::params![TEST_WORKSPACE_ID, workdir_id], + |row| row.get(0), + )?, + conn.query_row( + "SELECT COUNT(*) FROM worker_create_reservations WHERE workspace_id = ?1 AND runtime_id = ?2 AND state = 'reserved'", + rusqlite::params![TEST_WORKSPACE_ID, WorkdirlessFixtureRuntime::RUNTIME_ID], + |row| row.get(0), + )?, + )) + }) + .unwrap() + } + + fn assert_guarded_spawn_rolled_back(fixture: &GuardedSpawnFixture) { + assert_eq!( + guarded_spawn_artifact_counts(&fixture.api, &fixture.workdir_id), + (0, 0, 0) + ); + assert!( + fixture + .api + .store + .list_workdir_worker_links(TEST_WORKSPACE_ID, &fixture.workdir_id) + .unwrap() + .is_empty() + ); + assert!( + fixture + .api + .store + .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id) + .unwrap() + .is_none() + ); + assert!( + fixture + .runtime + .workers + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_empty() + ); + assert_eq!( + browser_ticket_backend(&fixture.api) + .unwrap() + .show(fixture.ticket_id.clone().into()) + .unwrap() + .meta + .workflow_state, + TicketWorkflowState::Queued + ); + } + + #[tokio::test] + async fn guarded_spawn_http_boundary_preserves_canonical_initial_attachment() { + let fixture = guarded_spawn_fixture().await; + let payload = guarded_spawn_payload(&fixture, "guarded-success"); + assert!(payload.get("working_directory").is_none()); + + let response = post_guarded_spawn(&fixture, payload).await; + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!( + status, + StatusCode::OK, + "unexpected guarded spawn response: {}", + String::from_utf8_lossy(&bytes) + ); + let body: BrowserCreateWorkerResponse = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body.worker.workdir_attachments.len(), 1); + assert_eq!(body.worker.workdir_attachments[0].alias, "workdir"); + assert_eq!( + body.worker.workdir_attachments[0] + .working_directory + .working_directory_id, + fixture.workdir_id + ); + let requests = fixture.runtime.spawn_requests(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].resolved_workdir_attachments.len(), 1); + let attachment = &requests[0].resolved_workdir_attachments[0]; + assert_eq!(attachment.alias.as_str(), "workdir"); + assert_eq!(attachment.working_directory_id, fixture.workdir_id); + assert_eq!(attachment.relative_cwd.as_deref(), Some("crates/yoi")); + assert!( + fixture + .api + .store + .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id) + .unwrap() + .is_some() + ); + assert_eq!( + fixture + .api + .store + .list_workdir_worker_links(TEST_WORKSPACE_ID, &fixture.workdir_id) + .unwrap() + .len(), + 1 + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn guarded_spawn_crosses_strict_remote_runtime_http_boundary() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + register_test_runtime(&api, STRICT_REMOTE_RUNTIME_ID).await; + + let workdir_id = "guarded-strict-remote-workdir".to_string(); + let repository_id = test_repository_id(&api); + let mut workdir_status = WorkdirlessFixtureRuntime::workdir_status(&workdir_id); + workdir_status.summary.repository_id = repository_id.clone(); + let strict_state = StrictRemoteSpawnState { + workdir_status, + create_requests: Arc::new(Mutex::new(Vec::new())), + }; + let runtime_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let runtime_endpoint = format!("http://{}", runtime_listener.local_addr().unwrap()); + let runtime_app = Router::new() + .route( + "/v1/working-directories/{working_directory_id}", + get(strict_remote_workdir), + ) + .route("/v1/workers", post(strict_remote_create_worker)) + .route( + "/v1/workers/{worker_id}/workspace-api", + post(strict_remote_replace_workspace_api), + ) + .with_state(strict_state.clone()); + let runtime_server = + tokio::spawn(async move { axum::serve(runtime_listener, runtime_app).await.unwrap() }); + api.runtime.register_or_replace( + RemoteWorkerRuntime::new( + RemoteRuntimeConfig { + runtime_id: STRICT_REMOTE_RUNTIME_ID.to_string(), + workspace_id: Some(TEST_WORKSPACE_ID.to_string()), + display_name: "Strict remote Runtime".to_string(), + base_url: runtime_endpoint, + bearer_token: Some(TEST_RUNTIME_HTTP_TOKEN.to_string()), + workspace_authorization: None, + strict_public_egress: false, + cached_worker_creation_available: true, + cached_os: "linux".to_string(), + cached_arch: "x86_64".to_string(), + cached_status: "active".to_string(), + timeout: std::time::Duration::from_secs(2), + }, + TEST_WORKSPACE_ID.to_string(), + "http://127.0.0.1:1".to_string(), + ) + .unwrap(), + ); + let now = now_registry_timestamp(); + api.store + .upsert_workdir_registry(&WorkdirRegistryRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + workdir_id: workdir_id.clone(), + display_name: None, + runtime_id: STRICT_REMOTE_RUNTIME_ID.to_string(), + repository_id: repository_id.clone(), + creation_selector: Some("develop".to_string()), + creation_ref: None, + creation_tree: None, + current_selector: Some("develop".to_string()), + current_ref: None, + current_tree: None, + observed_at_epoch_seconds: None, + materialization_status: "present".to_string(), + cleanliness: "clean".to_string(), + created_at: now.clone(), + updated_at: now, + }) + .unwrap(); + let Json(controller) = create_workspace_worker( + State(api.clone()), + HeaderMap::new(), + Json(CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + display_name: "Strict remote spawn controller".to_string(), + profile: Some("builtin:orchestrator".to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachments: Vec::new(), + control_operation_id: None, + }), + ) + .await + .unwrap(); + let controller = RuntimeWorkerRef::new(controller.runtime_id, controller.worker_id); + let mut ticket = ticket::NewTicket::new("Strict remote guarded attachment spawn"); + ticket.workflow_state = Some(TicketWorkflowState::Queued); + ticket.repository_id = Some(repository_id); + ticket.ref_selector = Some("develop".to_string()); + let ticket_id = browser_ticket_backend(&api) + .unwrap() + .create(ticket) + .unwrap() + .id; + assign_test_orchestrator(&api, &ticket_id); + let payload = json!({ + "runtime_id": STRICT_REMOTE_RUNTIME_ID, + "display_name": "Coder · strict remote attachment", + "profile": "builtin:coder", + "ticket_assignment": { + "ticket_id": ticket_id, + "operation_id": "guarded-strict-remote-success", + }, + "initial_submit": [ + { "kind": "flow", "selector": "builtin:coder-review" }, + { "kind": "text", "content": "Implement strict remote attachment Ticket." }, + ], + "workdir_attachments": [{ + "alias": "workdir", + "working_directory_id": workdir_id, + "relative_cwd": "crates/yoi", + }], + "control_operation_id": "guarded-strict-remote-success", + }); + assert!(payload.get("working_directory").is_none()); + + let response = build_inner_router(api.clone()) + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers")) + .header("content-type", "application/json") + .header("x-yoi-runtime-id", &controller.runtime_id) + .header("x-yoi-worker-id", &controller.worker_id) + .body(Body::from(payload.to_string())) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!( + status, + StatusCode::OK, + "unexpected guarded remote spawn response: {}", + String::from_utf8_lossy(&bytes) + ); + let body: BrowserCreateWorkerResponse = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body.worker.runtime_id, STRICT_REMOTE_RUNTIME_ID); + assert_eq!(body.worker.workdir_attachments.len(), 1); + assert_eq!(body.worker.workdir_attachments[0].alias.as_str(), "workdir"); + assert_eq!( + body.worker.workdir_attachments[0] + .working_directory + .working_directory_id, + workdir_id + ); + let requests = strict_state + .create_requests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + assert_eq!(requests.len(), 1); + assert!(requests[0].get("working_directory").is_none()); + assert_eq!( + requests[0]["workdir_attachments"].as_array().unwrap().len(), + 1 + ); + drop(requests); + assert!( + api.store + .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &ticket_id) + .unwrap() + .is_some() + ); + assert_eq!( + api.store + .list_workdir_worker_links(TEST_WORKSPACE_ID, &workdir_id) + .unwrap() + .len(), + 1 + ); + runtime_server.abort(); + } + + #[tokio::test] + async fn guarded_attachment_spawn_rolls_back_runtime_rejection() { + let fixture = guarded_spawn_fixture().await; + fixture.runtime.reject_next_spawn(); + + let response = post_guarded_spawn( + &fixture, + guarded_spawn_payload(&fixture, "guarded-runtime-rejection"), + ) + .await; + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(fixture.runtime.spawn_requests().len(), 1); + assert_guarded_spawn_rolled_back(&fixture); + } + + #[tokio::test] + async fn guarded_attachment_spawn_rolls_back_post_create_assignment_failure() { + let fixture = guarded_spawn_fixture().await; + rusqlite::Connection::open(&fixture.api.config.database_path) + .unwrap() + .execute_batch( + r#" + CREATE TRIGGER fail_guarded_coder_assignment + BEFORE INSERT ON ticket_current_worker_assignments + WHEN NEW.role = 'coder' + BEGIN + SELECT RAISE(ABORT, 'injected guarded Coder assignment failure'); + END; + "#, + ) + .unwrap(); + + let response = post_guarded_spawn( + &fixture, + guarded_spawn_payload(&fixture, "guarded-post-create-failure"), + ) + .await; + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(fixture.runtime.spawn_requests().len(), 1); + assert_guarded_spawn_rolled_back(&fixture); + } + + #[tokio::test] + async fn guarded_attachment_spawn_assignment_conflict_has_no_side_effects() { + let fixture = guarded_spawn_fixture().await; + fixture + .api + .store + .set_current_ticket_role_assignment( + &TicketRoleAssignmentRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + ticket_id: fixture.ticket_id.clone(), + assignment_id: "existing-coder-assignment".to_string(), + role: TicketAssignmentRole::Coder, + principal: TicketAssignmentPrincipal::Worker { + runtime_id: fixture.controller.runtime_id.clone(), + worker_id: fixture.controller.worker_id.clone(), + }, + assigned_by: "test".to_string(), + assigned_at: TEST_CREATED_AT.to_string(), + }, + None, + "existing-coder-event", + "existing-coder-operation", + false, + ) + .unwrap(); + + let response = post_guarded_spawn( + &fixture, + guarded_spawn_payload(&fixture, "guarded-assignment-conflict"), + ) + .await; + assert_eq!(response.status(), StatusCode::CONFLICT); + assert!(fixture.runtime.spawn_requests().is_empty()); + assert_eq!( + guarded_spawn_artifact_counts(&fixture.api, &fixture.workdir_id), + (0, 0, 0) + ); + assert!( + fixture + .api + .store + .list_workdir_worker_links(TEST_WORKSPACE_ID, &fixture.workdir_id) + .unwrap() + .is_empty() + ); + let current = fixture + .api + .store + .get_current_ticket_coder_assignment(TEST_WORKSPACE_ID, &fixture.ticket_id) + .unwrap() + .unwrap(); + assert_eq!(current.assignment_id, "existing-coder-assignment"); + } + #[tokio::test] async fn worker_control_spawn_retry_converges_on_one_worker_and_one_grant() { let workspace = tempfile::tempdir().unwrap(); @@ -33193,7 +33881,7 @@ mod tests { workspace_id: TEST_WORKSPACE_ID.to_string(), workdir_id: workdir_id.to_string(), display_name: None, - runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + runtime_id: "arcadia".to_string(), repository_id: test_repository_id(&api), creation_selector: Some("HEAD".to_string()), creation_ref: None, diff --git a/crates/workspace-server/src/workspace_subscription.rs b/crates/workspace-server/src/workspace_subscription.rs index 05acdfe0a..dac5df2a2 100644 --- a/crates/workspace-server/src/workspace_subscription.rs +++ b/crates/workspace-server/src/workspace_subscription.rs @@ -447,9 +447,6 @@ fn project_working_directory( else { return false; }; - if workdir.runtime_id != worker_ref.runtime_id { - return false; - } let Ok(working_directory_id) = protocol::subscription::SubscriptionWorkdirId::new(workdir.workdir_id) else {