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

chore: refresh T-642 onto current develop

Browse Source
This commit is contained in:
Hare
2026-09-21 22:59:43 +09:00
parent d7240470d1 769a3398fa
commit e085814502
6 changed files with 851 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/worker_backend.rs
+3 -3
View File
@@ -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
crates/worker/src/feature/builtin/manage_worker.rs
+79 -66
View File
@@ -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<CreateWorkspaceWorkerRequest, WorkspaceClientError> {
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<WorkspaceResponse, WorkspaceClientError> {
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<String>,
}
#[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<WorkerSpawnTicketAssignmentRequest>,
initial_submit: Vec<Segment>,
working_directory: WorkerWorkingDirectorySelection,
}
#[derive(Debug, Serialize)]
struct WorkerWorkingDirectorySelection {
working_directory_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
relative_cwd: Option<String>,
}
#[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::<CreateWorkspaceWorkerRequest>(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"],
crates/worker/src/feature/builtin/orchestration.rs
+4 -1
View File
@@ -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![
crates/workspace-server/src/hosts.rs
+71 -28
View File
@@ -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<ConfigBundleRef>,
) -> 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<ConfigBundleRef> {
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::<runtime_api::CreateWorkerRequest>(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(
crates/workspace-server/src/server.rs
+694 -6
View File
@@ -19569,14 +19569,55 @@ mod tests {
}
}
#[derive(Default)]
#[derive(Clone, Default)]
struct WorkdirlessFixtureRuntime {
workers: Mutex<Vec<WorkerSummary>>,
workers: Arc<Mutex<Vec<WorkerSummary>>>,
spawn_requests: Arc<Mutex<Vec<WorkerSpawnRequest>>>,
reject_next_spawn: Arc<Mutex<bool>>,
workdir_repositories: Arc<Mutex<HashMap<String, String>>>,
}
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<WorkerSpawnRequest> {
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<Mutex<Vec<Value>>>,
}
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<StrictRemoteSpawnState>,
headers: HeaderMap,
AxumPath(working_directory_id): AxumPath<String>,
) -> Json<Value> {
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<StrictRemoteSpawnState>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Json<Value> {
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<StrictRemoteSpawnState>,
headers: HeaderMap,
AxumPath(worker_id): AxumPath<String>,
Json(workspace_api): Json<runtime_api::WorkerWorkspaceApiRequest>,
) -> Json<Value> {
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,
crates/workspace-server/src/workspace_subscription.rs
-3
View File
@@ -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 {