diff --git a/crates/workdir/src/workspace.rs b/crates/workdir/src/workspace.rs index 1ea2cbb1..d0646706 100644 --- a/crates/workdir/src/workspace.rs +++ b/crates/workdir/src/workspace.rs @@ -211,6 +211,7 @@ pub struct WorkingDirectoryListResponse { #[serde(deny_unknown_fields)] pub struct WorkingDirectoryDetailResponse { pub workspace_id: String, + pub runtime_id: String, pub item: WorkingDirectorySummary, pub diagnostics: Vec, } @@ -306,6 +307,7 @@ mod tests { let detail = WorkingDirectoryDetailResponse { workspace_id: decoded.workspace_id.clone(), + runtime_id: "arcadia".to_string(), item: decoded.items[0].clone(), diagnostics: decoded.diagnostics.clone(), }; diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index 03c3ce1e..ffc7cbdc 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -398,24 +398,35 @@ impl WorkspaceHttpWorkdirBackend { workdir_output(format!("Listed {count} Workdir(s)"), &response) } - fn create(&self, input: WorkdirCreateInput) -> Result { - let runtime_id = validate_identity(&input.runtime_id, CREATE_TOOL, "runtime_id")?; + fn create( + &self, + input: WorkdirCreateInput, + operation_id: String, + ) -> Result { + let runtime_id = input + .runtime_id + .as_deref() + .map(|value| validate_identity(value, CREATE_TOOL, "runtime_id")) + .transpose()?; let repository_id = validate_identity(&input.repository_id, CREATE_TOOL, "repository_id")?; let selector = validate_optional_selector(input.selector)?; let workspace_id = encode_path_segment(self.workspace_id()?); - let runtime_path = encode_path_segment(runtime_id); let request = WorkdirCreateRequest { - runtime_id: runtime_id.to_string(), + runtime_id: runtime_id.map(str::to_string), repository_id: repository_id.to_string(), selector, + operation_id, }; let response = self.execute_json::(WorkspaceRequest::json( WorkspaceRequestMethod::Post, - format!("/api/w/{workspace_id}/runtimes/{runtime_path}/working-directories"), + format!("/api/w/{workspace_id}/working-directories"), serde_json::to_string(&request).map_err(decode_error)?, ))?; workdir_output( - format!("Created Workdir {}", response.item.working_directory_id), + format!( + "Created Workdir {} on Runtime {}", + response.item.working_directory_id, response.runtime_id + ), &response, ) } @@ -518,16 +529,17 @@ impl Tool for WorkspaceHttpWorkdirTool { async fn execute( &self, input_json: &str, - _ctx: ToolExecutionContext, + ctx: ToolExecutionContext, ) -> Result { match self.operation { WorkdirOperation::List => { let _input = parse_input::(input_json)?; self.backend.list() } - WorkdirOperation::Create => self - .backend - .create(parse_input::(input_json)?), + WorkdirOperation::Create => self.backend.create( + parse_input::(input_json)?, + ctx.call_id.to_string(), + ), WorkdirOperation::Attach => self .backend .attach(parse_input::(input_json)?), @@ -635,9 +647,9 @@ fn create_schema() -> serde_json::Value { json!({ "type": "object", "additionalProperties": false, - "required": ["runtime_id", "repository_id"], + "required": ["repository_id"], "properties": { - "runtime_id": {"type": "string", "minLength": 1}, + "runtime_id": {"type": ["string", "null"], "minLength": 1}, "repository_id": {"type": "string", "minLength": 1}, "selector": {"type": ["string", "null"], "minLength": 1} } @@ -677,7 +689,8 @@ struct WorkdirListInput {} #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct WorkdirCreateInput { - runtime_id: String, + #[serde(default)] + runtime_id: Option, repository_id: String, #[serde(default)] selector: Option, @@ -685,10 +698,12 @@ struct WorkdirCreateInput { #[derive(Debug, Serialize)] struct WorkdirCreateRequest { - runtime_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + runtime_id: Option, repository_id: String, #[serde(skip_serializing_if = "Option::is_none")] selector: Option, + operation_id: String, } #[derive(Debug, Deserialize)] @@ -916,7 +931,11 @@ mod tests { #[test] fn schemas_expose_identities_without_paths_or_session_handles() { let create = create_schema(); - assert_eq!(create["required"], json!(["runtime_id", "repository_id"])); + assert_eq!(create["required"], json!(["repository_id"])); + assert_eq!( + create["properties"]["runtime_id"]["type"], + json!(["string", "null"]) + ); assert!(create["properties"].get("path").is_none()); assert!(create["properties"].get("session_id").is_none()); assert_eq!(attach_schema()["required"], json!(["workdir_id"])); @@ -946,6 +965,7 @@ mod tests { })), response(json!({ "workspace_id": "workspace/test", + "runtime_id": "runtime/one", "item": workdir_json("wd-created"), "diagnostics": [] })), @@ -961,6 +981,7 @@ mod tests { })), response(json!({ "workspace_id": "workspace/test", + "runtime_id": "runtime/one", "item": { "working_directory_id": "wd-created", "repository_id": "main", @@ -985,13 +1006,19 @@ mod tests { .is_none() ); let created = backend - .create(WorkdirCreateInput { - runtime_id: "runtime/one".to_string(), - repository_id: "main".to_string(), - selector: Some("refs/heads/topic".to_string()), - }) + .create( + WorkdirCreateInput { + runtime_id: Some("runtime/one".to_string()), + repository_id: "main".to_string(), + selector: Some("refs/heads/topic".to_string()), + }, + "call-create-1".to_string(), + ) .unwrap(); - assert_eq!(created.summary, "Created Workdir wd-created"); + assert_eq!( + created.summary, + "Created Workdir wd-created on Runtime runtime/one" + ); let created: serde_json::Value = serde_json::from_str(created.content.as_deref().unwrap()).unwrap(); assert_eq!(created["item"]["working_directory_id"], "wd-created"); @@ -1017,12 +1044,14 @@ mod tests { assert_eq!(requests[0].method, WorkspaceRequestMethod::Get); assert_eq!( requests[1].path, - "/api/w/workspace%2Ftest/runtimes/runtime%2Fone/working-directories" + "/api/w/workspace%2Ftest/working-directories" ); assert_eq!(requests[1].method, WorkspaceRequestMethod::Post); let body: serde_json::Value = serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap(); assert_eq!(body["repository_id"], "main"); + assert_eq!(body["runtime_id"], "runtime/one"); + assert_eq!(body["operation_id"], "call-create-1"); assert_eq!(body["selector"], "refs/heads/topic"); assert_eq!( requests[2].path, @@ -1216,16 +1245,50 @@ mod tests { assert_eq!(body["operation"]["request"]["path"], "file"); } + #[test] + fn create_omits_runtime_for_backend_default_resolution() { + let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({ + "workspace_id": "workspace/test", + "runtime_id": "arcadia", + "item": workdir_json("wd-default"), + "diagnostics": [] + }))])); + let backend = WorkspaceHttpWorkdirBackend::new(client.clone()); + + let created = backend + .create( + WorkdirCreateInput { + runtime_id: None, + repository_id: "main".to_string(), + selector: None, + }, + "call-default".to_string(), + ) + .unwrap(); + assert_eq!( + created.summary, + "Created Workdir wd-default on Runtime arcadia" + ); + let requests = client.requests(); + let body: serde_json::Value = + serde_json::from_str(requests[0].body.as_deref().unwrap()).unwrap(); + assert!(body.get("runtime_id").is_none()); + assert_eq!(body["operation_id"], "call-default"); + } + #[test] fn invalid_or_extra_inputs_are_rejected_before_workspace_request() { let client = Arc::new(RecordingWorkspaceClient::new(Vec::new())); let backend = WorkspaceHttpWorkdirBackend::new(client.clone()); let error = backend - .create(WorkdirCreateInput { - runtime_id: " ".to_string(), - repository_id: "main".to_string(), - selector: None, - }) + .create( + WorkdirCreateInput { + runtime_id: Some(" ".to_string()), + repository_id: "main".to_string(), + selector: None, + }, + "call-invalid".to_string(), + ) .unwrap_err(); assert!(matches!(error, ToolError::InvalidArgument(_))); assert!(client.requests().is_empty()); diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index d8ca871d..44b462c4 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -22,10 +22,12 @@ pub use records::ticket_api_typescript; pub mod repositories; pub mod resource_broker; pub mod retention; +pub mod runtime_settings; pub mod runtime_subscription; pub mod server; pub mod skills; pub mod store; +pub mod workdir_create_operations; pub mod worker_source; pub mod workspace_catalog; mod workspace_subscription; diff --git a/crates/workspace-server/src/runtime_settings.rs b/crates/workspace-server/src/runtime_settings.rs new file mode 100644 index 00000000..0167647d --- /dev/null +++ b/crates/workspace-server/src/runtime_settings.rs @@ -0,0 +1,187 @@ +use config_source::ConfigSchemaContribution; +use serde::Deserialize; + +use crate::config_source::{ + WorkspaceConfigSchemaProvider, WorkspaceConfigState, evaluate_workspace_config_state, +}; +use crate::{Error, Result}; + +const RUNTIME_SCHEMA_SOURCE: &str = r#"{ + runtime = { + default_runtime_id = String default ""; + }; +}"#; + +#[derive(Debug, Default)] +pub struct RuntimeConfigSchemaProvider; + +impl WorkspaceConfigSchemaProvider for RuntimeConfigSchemaProvider { + fn contribution(&self) -> Result { + ConfigSchemaContribution::new("builtin:runtime", "runtime", "1", RUNTIME_SCHEMA_SOURCE) + .map_err(|error| Error::Config(error.to_string())) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RuntimeConfigProjection { + pub config_revision: u64, + pub projection_digest: String, + pub default_runtime_id: Option, +} + +#[derive(Debug, Deserialize)] +struct VirtualRuntimeConfig { + runtime: VirtualRuntimeSection, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct VirtualRuntimeSection { + default_runtime_id: String, +} + +pub fn project_runtime_from_workspace_config( + workspace_id: &str, + state: &WorkspaceConfigState, +) -> Result { + let bundle = if state.contract.schema_bundle.contributions.is_empty() { + config_source::WorkspaceConfigSchemaBundle::compose([ + RuntimeConfigSchemaProvider.contribution()? + ]) + .map_err(|error| Error::Config(error.to_string()))? + } else { + state.contract.schema_bundle.clone() + }; + let evaluation = evaluate_workspace_config_state(state, bundle)?; + if evaluation.projection_digest != state.projection_digest + && state + .contract + .schema_bundle + .contributions + .iter() + .any(|entry| entry.provider_id == "builtin:runtime") + { + return Err(Error::RegistryInconsistency(format!( + "Runtime projection digest mismatch for Workspace {workspace_id}" + ))); + } + let projected = evaluation.projections.first().ok_or_else(|| { + Error::RegistryInconsistency("Workspace config has no active projection".to_string()) + })?; + let config: VirtualRuntimeConfig = serde_json::from_value(projected.data_json.clone()) + .map_err(|error| Error::RegistryInconsistency(error.to_string()))?; + let default_runtime_id = normalize_runtime_id(&config.runtime.default_runtime_id)?; + Ok(RuntimeConfigProjection { + config_revision: state.snapshot.revision, + projection_digest: evaluation.projection_digest, + default_runtime_id, + }) +} + +fn normalize_runtime_id(value: &str) -> Result> { + let value = value.trim(); + if value.is_empty() { + return Ok(None); + } + if value.chars().any(char::is_control) { + return Err(Error::InvalidRuntimeIdentifier { + kind: "runtime_id".to_string(), + value: "[redacted invalid value]".to_string(), + }); + } + Ok(Some(value.to_string())) +} + +#[cfg(test)] +mod tests { + use config_source::{ConfigContentType, ConfigEntry, ConfigTreeSnapshot, VirtualPath}; + + use super::*; + + fn state(source: &str) -> WorkspaceConfigState { + let bundle = + config_source::WorkspaceConfigSchemaBundle::compose([RuntimeConfigSchemaProvider + .contribution() + .unwrap()]) + .unwrap(); + let snapshot = ConfigTreeSnapshot::from_entries( + 7, + [ConfigEntry::new( + VirtualPath::parse("main.dcdl").unwrap(), + ConfigContentType::Decodal, + source, + ) + .unwrap()], + ) + .unwrap(); + let contract = config_source::ToolchainContract::with_schema_bundle( + config_source::DEFAULT_SCHEMA_VERSION, + vec![VirtualPath::parse("main.dcdl").unwrap()], + config_source::DEFAULT_IMPORT_POLICY_VERSION, + bundle, + ); + let projection_digest = config_source::SnapshotEnvironment::new(snapshot.clone()) + .evaluate_contract(&contract) + .unwrap() + .projection_digest; + WorkspaceConfigState { + snapshot, + contract, + projection_digest, + } + } + + #[test] + fn runtime_projection_reads_default_and_preserves_revision_evidence() { + let projection = project_runtime_from_workspace_config( + "workspace", + &state( + r#"{ runtime = { default_runtime_id = "arcadia"; }; } as WorkspaceConfigSchema"#, + ), + ) + .unwrap(); + assert_eq!(projection.default_runtime_id.as_deref(), Some("arcadia")); + assert_eq!(projection.config_revision, 7); + assert!(!projection.projection_digest.is_empty()); + } + + #[test] + fn runtime_projection_treats_missing_default_as_unconfigured() { + let projection = project_runtime_from_workspace_config( + "workspace", + &state("{} as WorkspaceConfigSchema"), + ) + .unwrap(); + assert_eq!(projection.default_runtime_id, None); + } + + #[test] + fn runtime_schema_rejects_non_string_default() { + let bundle = + config_source::WorkspaceConfigSchemaBundle::compose([RuntimeConfigSchemaProvider + .contribution() + .unwrap()]) + .unwrap(); + let snapshot = ConfigTreeSnapshot::from_entries( + 1, + [ConfigEntry::new( + VirtualPath::parse("main.dcdl").unwrap(), + ConfigContentType::Decodal, + "{ runtime = { default_runtime_id = 42; }; } as WorkspaceConfigSchema", + ) + .unwrap()], + ) + .unwrap(); + let contract = config_source::ToolchainContract::with_schema_bundle( + config_source::DEFAULT_SCHEMA_VERSION, + vec![VirtualPath::parse("main.dcdl").unwrap()], + config_source::DEFAULT_IMPORT_POLICY_VERSION, + bundle, + ); + assert!( + config_source::SnapshotEnvironment::new(snapshot) + .evaluate_contract(&contract) + .is_err() + ); + } +} diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index dc8d2b58..951108ce 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -110,14 +110,16 @@ use crate::repositories::{ RepositoryRegistryReader, RepositorySummary, }; use crate::resource_broker::BackendResourceBroker; +use crate::runtime_settings::RuntimeConfigSchemaProvider; use crate::runtime_subscription::RuntimeSubscriptionBroker; use crate::skills; use crate::store::{ AccountRecord, ApiTokenRecord, AuthChallengeRecord, BrowserSessionRecord, ControlPlaneStore, DeviceLoginFlowRecord, FlowSourceRecord, PasskeyCredentialRecord, RepositoryRecord, TicketAssignmentPrincipal, TicketAssignmentRole, TicketCoderAssignmentRecord, - TicketRoleAssignmentRecord, UserRecord, WorkdirRegistryRecord, WorkerControlGrantRecord, - WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord, WorkspaceResourceKind, + TicketRoleAssignmentRecord, UserRecord, WorkdirCreateOperationRecord, WorkdirRegistryRecord, + WorkerControlGrantRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord, + WorkspaceResourceKind, }; use crate::workspace_catalog::{WorkspaceCatalogService, WorkspaceCreateRequest}; use crate::{Error, Result}; @@ -134,10 +136,6 @@ use worker_runtime::identity::{RuntimeWorkerRef, WorkerId}; const EMBEDDED_WORKER_RUNTIME_ID: &str = "embedded-worker-runtime"; -fn embedded_runtime_id() -> String { - EMBEDDED_WORKER_RUNTIME_ID.to_string() -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub enum AuthConfig { /// Browser human auth uses Passkey ceremonies and HttpOnly cookie sessions; @@ -1099,6 +1097,7 @@ impl WorkspaceApi { crate::profile_settings::ProfileConfigSchemaProvider, )) .with_provider(Arc::new(crate::prompt_settings::PromptConfigSchemaProvider)) + .with_provider(Arc::new(RuntimeConfigSchemaProvider)) .with_provider(Arc::new(skills::SkillConfigSchemaProvider)); config_store.ensure_workspace_config_materialized_with_schema( &config.workspace_id, @@ -2548,14 +2547,16 @@ pub struct WorkingDirectoryRepositoryOption { pub default_selector: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Clone, Deserialize)] #[serde(deny_unknown_fields)] pub struct BrowserWorkingDirectoryCreateRequest { - #[serde(default = "embedded_runtime_id")] - pub runtime_id: String, + #[serde(default)] + pub runtime_id: Option, pub repository_id: String, #[serde(default)] pub selector: Option, + #[serde(default)] + pub operation_id: Option, } #[derive(Debug, Serialize, Deserialize)] @@ -7607,11 +7608,15 @@ async fn scoped_list_runtime_working_directories( async fn scoped_create_runtime_working_directory( State(api): State, AxumPath(path): AxumPath, - Json(mut request): Json, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - request.runtime_id = path.runtime_id; - create_working_directory_for_runtime(api, request) + Json(request): Json, +) -> ApiResult<(StatusCode, Json)> { + create_workspace_working_directory( + &api, + &path.workspace_id, + Some(path.runtime_id.as_str()), + request, + ) + .await } async fn scoped_runtime_working_directory_detail( @@ -7646,21 +7651,9 @@ async fn scoped_list_working_directories( async fn scoped_create_working_directory( State(api): State, AxumPath(path): AxumPath, - Json(_request): Json, -) -> ApiResult> { - validate_workspace_scope(&api, &path.workspace_id)?; - Err(ApiError::with_diagnostics( - Error::RuntimeOperationFailed { - runtime_id: "workspace-backend".to_string(), - code: "backend_workdir_create_unsupported".to_string(), - message: "Working directory creation must be scoped to a concrete Runtime".to_string(), - }, - vec![RuntimeDiagnostic { - code: "backend_workdir_create_unsupported".to_string(), - severity: DiagnosticSeverity::Error, - message: "Use runtime-scoped Worker creation or a runtime-scoped working-directory API; the backend does not own workdir lifecycle.".to_string(), - }], - )) + Json(request): Json, +) -> ApiResult<(StatusCode, Json)> { + create_workspace_working_directory(&api, &path.workspace_id, None, request).await } async fn scoped_working_directory_detail( @@ -7700,60 +7693,225 @@ fn registered_workdir_runtime_id( }) } -fn create_working_directory_for_runtime( - api: WorkspaceApi, +async fn create_workspace_working_directory( + api: &WorkspaceApi, + workspace_id: &str, + route_runtime_id: Option<&str>, request: BrowserWorkingDirectoryCreateRequest, -) -> ApiResult> { - let runtime_id = request.runtime_id.clone(); - let mut working_directory_request = working_directory_request_for_browser(&api, request)?; - let workdir_id = next_backend_workdir_id(&working_directory_request.repository.id); - working_directory_request.backend_workdir_id = Some(workdir_id.clone()); - let pending = WorkdirRegistryRecord { - workspace_id: api.config.workspace_id.clone(), - workdir_id: workdir_id.clone(), - runtime_id: runtime_id.clone(), - repository_id: working_directory_request.repository.id.clone(), - creation_selector: working_directory_request - .repository - .selector - .as_ref() - .map(|selector| selector.as_ref().to_string()), - creation_ref: None, - current_selector: None, - current_ref: None, - materialization_status: "pending".to_string(), - cleanliness: "unknown".to_string(), - created_at: now_registry_timestamp(), - updated_at: now_registry_timestamp(), +) -> ApiResult<(StatusCode, Json)> { + validate_workspace_scope(api, workspace_id)?; + if let (Some(route_runtime_id), Some(request_runtime_id)) = + (route_runtime_id, request.runtime_id.as_deref()) + && route_runtime_id != request_runtime_id + { + return Err(Error::InvalidInput( + "runtime_id does not match the runtime-scoped route".to_string(), + ) + .into()); + } + let requested_runtime_id = route_runtime_id + .map(str::to_string) + .or_else(|| request.runtime_id.clone()); + let mut working_directory_request = + working_directory_request_for_browser(api, request.clone())?; + let operation_id = request + .operation_id + .clone() + .unwrap_or_else(|| Uuid::now_v7().to_string()); + if operation_id.trim().is_empty() || operation_id.len() > 256 { + return Err(Error::InvalidInput( + "operation_id must be non-empty and at most 256 bytes".to_string(), + ) + .into()); + } + let selector = working_directory_request + .repository + .selector + .as_ref() + .map(|selector| selector.as_ref().to_string()); + let request_fingerprint = crate::workdir_create_operations::request_fingerprint( + &request.repository_id, + selector.as_deref(), + requested_runtime_id.as_deref(), + ); + let reserved = if let Some(existing) = api + .config_store + .load_workdir_create_operation(workspace_id, &operation_id)? + { + if existing.request_fingerprint != request_fingerprint { + return Err(Error::InvalidInput(format!( + "Workdir create operation `{operation_id}` was reused with different input" + )) + .into()); + } + existing + } else { + let config_state = api + .config_store + .load_workspace_config(workspace_id)? + .ok_or_else(|| { + Error::RegistryInconsistency(format!( + "Workspace {workspace_id} has no active configuration" + )) + })?; + let runtime_projection = crate::runtime_settings::project_runtime_from_workspace_config( + workspace_id, + &config_state, + )?; + let resolved_runtime_id = requested_runtime_id + .clone() + .or(runtime_projection.default_runtime_id.clone()) + .ok_or_else(|| { + Error::InvalidInput( + "runtime_id was omitted and Workspace configuration has no runtime.default_runtime_id" + .to_string(), + ) + })?; + let now = now_registry_timestamp(); + api.config_store + .reserve_workdir_create_operation(&WorkdirCreateOperationRecord { + workspace_id: workspace_id.to_string(), + operation_id: operation_id.clone(), + request_fingerprint: request_fingerprint.clone(), + repository_id: request.repository_id.clone(), + selector, + requested_runtime_id, + resolved_runtime_id, + config_revision: runtime_projection.config_revision, + config_projection_digest: runtime_projection.projection_digest, + working_directory_id: next_backend_workdir_id(&request.repository_id), + state: "pending".to_string(), + failure: None, + created_at: now.clone(), + updated_at: now, + })? }; - api.store.upsert_workdir_registry(&pending)?; - let result = api + + let runtime = match api .runtime - .create_working_directory(&runtime_id, working_directory_request) + .list_runtimes(usize::MAX) + .items + .into_iter() + .find(|runtime| runtime.runtime_id == reserved.resolved_runtime_id) + { + Some(runtime) => runtime, + None => { + api.config_store.finish_workdir_create_operation( + workspace_id, + &operation_id, + &request_fingerprint, + false, + Some("resolved Runtime is not registered"), + &now_registry_timestamp(), + )?; + return Err(Error::UnknownRuntime(reserved.resolved_runtime_id).into()); + } + }; + if runtime.kind == "remote_http" && runtime.status != "connected" { + api.config_store.finish_workdir_create_operation( + workspace_id, + &operation_id, + &request_fingerprint, + false, + Some("resolved Runtime is unavailable"), + &now_registry_timestamp(), + )?; + return Err(Error::RuntimeOperationFailed { + runtime_id: reserved.resolved_runtime_id, + code: "runtime_unavailable".to_string(), + message: "Selected Runtime is not connected".to_string(), + } + .into()); + } + + if reserved.state == "succeeded" { + return working_directory_detail_for_runtime( + api.clone(), + &reserved.resolved_runtime_id, + &reserved.working_directory_id, + ) + .map(|response| (StatusCode::OK, response)); + } + + working_directory_request.backend_workdir_id = Some(reserved.working_directory_id.clone()); + let existing = api + .runtime + .working_directory( + &reserved.resolved_runtime_id, + &reserved.working_directory_id, + ) .map_err(|err| err.into_error())?; + let result = if existing.working_directory.is_some() { + existing + } else { + match api + .runtime + .create_working_directory(&reserved.resolved_runtime_id, working_directory_request) + { + Ok(result) => result, + Err(error) => { + let error = error.into_error(); + let _ = api + .store + .delete_workdir_registry(workspace_id, &reserved.working_directory_id); + api.config_store.finish_workdir_create_operation( + workspace_id, + &operation_id, + &request_fingerprint, + false, + Some("runtime rejected Workdir creation"), + &now_registry_timestamp(), + )?; + return Err(error.into()); + } + } + }; let Some(working_directory) = result.working_directory else { - let mut failed = pending; - failed.materialization_status = "failed".to_string(); - failed.updated_at = now_registry_timestamp(); - api.store.upsert_workdir_registry(&failed)?; + let _ = api + .store + .delete_workdir_registry(workspace_id, &reserved.working_directory_id); + api.config_store.finish_workdir_create_operation( + workspace_id, + &operation_id, + &request_fingerprint, + false, + Some("runtime did not create working directory"), + &now_registry_timestamp(), + )?; return Err(ApiError::with_diagnostics( Error::RuntimeOperationFailed { - runtime_id, + runtime_id: reserved.resolved_runtime_id, code: "workspace_working_directory_create_failed".to_string(), message: "Runtime did not create working directory".to_string(), }, result.diagnostics, )); }; - let record = workdir_record_from_summary(&api, &runtime_id, &working_directory.summary); + let record = workdir_record_from_summary( + api, + &reserved.resolved_runtime_id, + &working_directory.summary, + ); api.store.upsert_workdir_registry(&record)?; + api.config_store.finish_workdir_create_operation( + workspace_id, + &operation_id, + &request_fingerprint, + true, + None, + &now_registry_timestamp(), + )?; let mut summary = working_directory.summary; - apply_workdir_occupancy_projection(&api, &mut summary)?; - Ok(Json(BrowserWorkingDirectoryDetailResponse { - workspace_id: api.config.workspace_id.clone(), - item: summary, - diagnostics: working_directory_diagnostics(result.diagnostics), - })) + apply_workdir_occupancy_projection(api, &mut summary)?; + Ok(( + StatusCode::CREATED, + Json(BrowserWorkingDirectoryDetailResponse { + workspace_id: workspace_id.to_string(), + runtime_id: reserved.resolved_runtime_id, + item: summary, + diagnostics: working_directory_diagnostics(result.diagnostics), + }), + )) } fn working_directory_detail_for_runtime( @@ -7772,6 +7930,7 @@ fn working_directory_detail_for_runtime( apply_workdir_occupancy_projection(&api, &mut summary)?; return Ok(Json(BrowserWorkingDirectoryDetailResponse { workspace_id: api.config.workspace_id.clone(), + runtime_id: runtime_id.to_string(), item: summary, diagnostics: working_directory_diagnostics(result.diagnostics), })); @@ -7782,6 +7941,7 @@ fn working_directory_detail_for_runtime( { return Ok(Json(BrowserWorkingDirectoryDetailResponse { workspace_id: api.config.workspace_id.clone(), + runtime_id: runtime_id.to_string(), item: projected_workdir_summary_from_record(&api, &record)?, diagnostics: working_directory_diagnostics(result.diagnostics), })); @@ -7841,6 +8001,7 @@ fn cleanup_working_directory_for_runtime( apply_workdir_occupancy_projection(&api, &mut summary)?; Ok(Json(BrowserWorkingDirectoryDetailResponse { workspace_id: api.config.workspace_id.clone(), + runtime_id: runtime_id.to_string(), item: summary, diagnostics: working_directory_diagnostics(result.diagnostics), })) @@ -17446,6 +17607,38 @@ mod tests { test_api_with_recording_backend(workspace_root).await.0 } + fn set_test_default_runtime(api: &WorkspaceApi, runtime_id: &str) { + let current = api + .config_store + .load_workspace_config(TEST_WORKSPACE_ID) + .unwrap() + .unwrap(); + let main_path = config_source::VirtualPath::parse("main.dcdl").unwrap(); + let request = crate::config_source::ConfigCommitRequest { + base_revision: current.snapshot.revision, + base_digest: current.snapshot.digest.clone(), + changes: vec![config_source::ConfigTreeChange::Update { + path: main_path.clone(), + expected_digest: current.snapshot.entries[&main_path].content_digest.clone(), + content: format!( + "{{ runtime = {{ default_runtime_id = {runtime_id:?}; }}; }} as WorkspaceConfigSchema" + ), + }], + entrypoints: current.contract.entrypoints.clone(), + }; + let candidate = api + .config_store + .evaluate_workspace_config_candidate_with_schema( + TEST_WORKSPACE_ID, + &request, + api.config_schema_registry.compose().unwrap(), + ) + .unwrap(); + api.config_store + .commit_evaluated_workspace_config(TEST_WORKSPACE_ID, &candidate) + .unwrap(); + } + fn assign_test_orchestrator(api: &WorkspaceApi, ticket_id: &str) { api.store .set_current_ticket_role_assignment( @@ -18709,7 +18902,7 @@ mod tests { } #[tokio::test] - async fn browser_working_directory_create_is_rejected_as_backend_lifecycle() { + async fn browser_workspace_workdir_create_requires_configured_default_runtime() { let dir = tempfile::tempdir().unwrap(); init_clean_git_workspace(dir.path()); let app = test_app(dir.path()).await; @@ -18726,18 +18919,106 @@ mod tests { StatusCode::BAD_REQUEST, ) .await; - assert!( - response["diagnostics"] - .as_array() - .unwrap() - .iter() - .any(|diagnostic| diagnostic["code"] == "backend_workdir_create_unsupported"), - "expected backend lifecycle diagnostic, got {response}" - ); + assert_eq!(response["error"], "Bad Request"); let projected = serde_json::to_string(&response).unwrap(); assert!(!projected.contains(dir.path().to_string_lossy().as_ref())); } + #[tokio::test] + async fn browser_workspace_workdir_create_does_not_fallback_from_explicit_runtime() { + let dir = tempfile::tempdir().unwrap(); + init_clean_git_workspace(dir.path()); + let api = test_api(dir.path()).await; + set_test_default_runtime(&api, EMBEDDED_WORKER_RUNTIME_ID); + let operation_id = "workdir-create-explicit-runtime"; + + let response = request_json( + build_router(api.clone()), + "POST", + &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), + Some(serde_json::json!({ + "runtime_id": "missing-runtime", + "repository_id": TEST_REPOSITORY_ID, + "selector": "HEAD", + "operation_id": operation_id, + })), + StatusCode::NOT_FOUND, + ) + .await; + assert_eq!(response["error"], "Not Found"); + let operation = api + .config_store + .load_workdir_create_operation(TEST_WORKSPACE_ID, operation_id) + .unwrap() + .unwrap(); + assert_eq!( + operation.requested_runtime_id.as_deref(), + Some("missing-runtime") + ); + assert_eq!(operation.resolved_runtime_id, "missing-runtime"); + assert_eq!(operation.state, "failed"); + assert!( + api.store + .get_workdir_registry(TEST_WORKSPACE_ID, &operation.working_directory_id) + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn browser_workspace_workdir_create_records_failed_default_resolution() { + let dir = tempfile::tempdir().unwrap(); + init_clean_git_workspace(dir.path()); + let api = test_api(dir.path()).await; + set_test_default_runtime(&api, EMBEDDED_WORKER_RUNTIME_ID); + let app = build_router(api.clone()); + let operation_id = "workdir-create-default-runtime"; + + let response = request_json( + app.clone(), + "POST", + &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), + Some(serde_json::json!({ + "repository_id": TEST_REPOSITORY_ID, + "selector": "HEAD", + "operation_id": operation_id, + })), + StatusCode::BAD_GATEWAY, + ) + .await; + assert_eq!(response["error"], "Bad Gateway"); + + set_test_default_runtime(&api, "not-a-registered-runtime"); + request_json( + app, + "POST", + &format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"), + Some(serde_json::json!({ + "repository_id": TEST_REPOSITORY_ID, + "selector": "HEAD", + "operation_id": operation_id, + })), + StatusCode::BAD_GATEWAY, + ) + .await; + + let operation = api + .config_store + .load_workdir_create_operation(TEST_WORKSPACE_ID, operation_id) + .unwrap() + .unwrap(); + assert_eq!(operation.resolved_runtime_id, EMBEDDED_WORKER_RUNTIME_ID); + assert_eq!(operation.state, "failed"); + assert_eq!(operation.config_revision, 2); + assert!(!operation.config_projection_digest.is_empty()); + assert!( + api.store + .get_workdir_registry(TEST_WORKSPACE_ID, &operation.working_directory_id) + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn scoped_worker_create_invalid_relative_cwd_returns_typed_diagnostic() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index c49960af..75e0d358 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -241,6 +241,11 @@ const MIGRATIONS: &[Migration] = &[ name: "generalize Ticket assignments to role principals", apply: generalize_ticket_role_assignments, }, + Migration { + version: 44, + name: "create Workdir create operations", + apply: create_workdir_create_operations, + }, ]; struct Migration { @@ -561,6 +566,24 @@ pub struct TicketWorkerAssignmentUpdate { pub previous: Option, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkdirCreateOperationRecord { + pub workspace_id: String, + pub operation_id: String, + pub request_fingerprint: String, + pub repository_id: String, + pub selector: Option, + pub requested_runtime_id: Option, + pub resolved_runtime_id: String, + pub config_revision: u64, + pub config_projection_digest: String, + pub working_directory_id: String, + pub state: String, + pub failure: Option, + pub created_at: String, + pub updated_at: String, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkdirRegistryRecord { pub workspace_id: String, @@ -6399,6 +6422,32 @@ CREATE UNIQUE INDEX ux_worker_workdir_attachment_reservation_id Ok(()) } +fn create_workdir_create_operations(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" + CREATE TABLE workdir_create_operations ( + workspace_id TEXT NOT NULL, + operation_id TEXT NOT NULL, + request_fingerprint TEXT NOT NULL, + repository_id TEXT NOT NULL, + selector TEXT, + requested_runtime_id TEXT, + resolved_runtime_id TEXT NOT NULL, + config_revision INTEGER NOT NULL, + config_projection_digest TEXT NOT NULL, + working_directory_id TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'succeeded', 'failed')), + failure TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, operation_id), + UNIQUE (workspace_id, working_directory_id) + ); + "#, + )?; + Ok(()) +} + fn create_workspace_catalog_operations(conn: &Connection) -> Result<()> { conn.execute_batch( r#" @@ -9278,7 +9327,7 @@ mod tests { let before = std::fs::read(&path).unwrap(); let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap(); assert_eq!(plan.current_schema_version, 36); - assert_eq!(plan.target_schema_version, 43); + assert_eq!(plan.target_schema_version, 44); assert!(plan.migration_required); assert_eq!(plan.worker_count, 1); assert_eq!(plan.mappings[0].legacy_worker_id, 7); @@ -9292,7 +9341,7 @@ mod tests { store .with_conn(|conn| { assert!(table_exists(conn, "worker_diagnostics_archives")?); - assert_eq!(current_schema_version(conn)?, 43); + assert_eq!(current_schema_version(conn)?, 44); Ok(()) }) .unwrap(); @@ -9371,7 +9420,7 @@ mod tests { ), ] ); - assert_eq!(current_schema_version(&conn).unwrap(), 43); + assert_eq!(current_schema_version(&conn).unwrap(), 44); let foreign_key_error: Option = conn .query_row("PRAGMA foreign_key_check", [], |row| row.get(0)) .optional() @@ -9500,7 +9549,7 @@ INSERT INTO worker_orphan_diagnostics ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 43); + assert_eq!(current_schema_version(&conn).unwrap(), 44); assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap()); let controller_worker_id: String = conn .query_row( @@ -9618,7 +9667,7 @@ INSERT INTO worker_orphan_diagnostics ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 43); + assert_eq!(current_schema_version(&conn).unwrap(), 44); assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap()); } @@ -9636,7 +9685,7 @@ INSERT INTO worker_orphan_diagnostics ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 43); + assert_eq!(current_schema_version(&conn).unwrap(), 44); let settings = conn .query_row( "SELECT settings_revision, language FROM workspace_memory_settings \ @@ -9677,7 +9726,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 43); + assert_eq!(current_schema_version(&conn).unwrap(), 44); assert!(table_exists(&conn, "flow_sources").unwrap()); assert!(table_exists(&conn, "flow_source_revisions").unwrap()); assert!(!table_exists(&conn, "flow_instances").unwrap()); @@ -9744,7 +9793,7 @@ INSERT INTO worker_workdir_attachment_reservations ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 43); + assert_eq!(current_schema_version(&conn).unwrap(), 44); let repositories_sql: String = conn .query_row( "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'", @@ -9922,7 +9971,7 @@ INSERT INTO workdir_registry ( let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 43); + assert_eq!(store.schema_version().await.unwrap(), 44); assert!( !store .with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) @@ -9939,7 +9988,7 @@ INSERT INTO workdir_registry ( store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 43); + assert_eq!(reopened.schema_version().await.unwrap(), 44); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -10693,7 +10742,7 @@ INSERT INTO worker_registry ( let migrated = SqliteWorkspaceStore::open(&db_path).unwrap(); migrated .with_conn(|conn| { - assert_eq!(current_schema_version(conn)?, 43); + assert_eq!(current_schema_version(conn)?, 44); assert_eq!( conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?, 1, @@ -10991,19 +11040,51 @@ INSERT INTO worker_registry ( ); } + #[test] + fn schema_v44_adds_workdir_create_operations_to_v43_database() { + let conn = Connection::open_in_memory().unwrap(); + configure_sqlite(&conn).unwrap(); + apply_migrations(&conn).unwrap(); + conn.execute_batch( + "DROP TABLE workdir_create_operations; + DELETE FROM __yoi_schema_migrations WHERE version = 44;", + ) + .unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 43); + + apply_migrations(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 44); + assert!(table_exists(&conn, "workdir_create_operations").unwrap()); + let columns = table_columns(&conn, "workdir_create_operations").unwrap(); + for required in [ + "operation_id", + "request_fingerprint", + "resolved_runtime_id", + "config_revision", + "config_projection_digest", + "working_directory_id", + "state", + ] { + assert!( + columns.iter().any(|column| column == required), + "missing column {required}" + ); + } + } + #[test] fn server_refuses_a_database_from_a_newer_schema_generation() { let conn = Connection::open_in_memory().unwrap(); configure_sqlite(&conn).unwrap(); apply_migrations(&conn).unwrap(); conn.execute( - "INSERT INTO __yoi_schema_migrations (version, name) VALUES (44, 'future')", + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (45, 'future')", [], ) .unwrap(); let error = apply_migrations(&conn).unwrap_err().to_string(); - assert!(error.contains("schema version 44 is newer"), "{error}"); + assert!(error.contains("schema version 45 is newer"), "{error}"); assert!(error.contains("refusing to serve"), "{error}"); } @@ -11224,7 +11305,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026- apply_migrations(&mut conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 43); + assert_eq!(current_schema_version(&conn).unwrap(), 44); let workspace_id: Option = conn .query_row( "SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'", @@ -11841,7 +11922,7 @@ WHERE workspace_id = 'workspace-a' .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 43); + assert_eq!(store.schema_version().await.unwrap(), 44); store .with_conn(|conn| { @@ -12030,7 +12111,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 43); + assert_eq!(store.schema_version().await.unwrap(), 44); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -12096,7 +12177,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 43); + assert_eq!(store.schema_version().await.unwrap(), 44); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -12498,7 +12579,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 43); + assert_eq!(store.schema_version().await.unwrap(), 44); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(), diff --git a/crates/workspace-server/src/workdir_create_operations.rs b/crates/workspace-server/src/workdir_create_operations.rs new file mode 100644 index 00000000..e1e6e737 --- /dev/null +++ b/crates/workspace-server/src/workdir_create_operations.rs @@ -0,0 +1,241 @@ +use rusqlite::{OptionalExtension, params}; +use sha2::{Digest, Sha256}; + +use crate::store::WorkdirCreateOperationRecord; +use crate::{Error, Result, SqliteWorkspaceStore}; + +pub fn request_fingerprint( + repository_id: &str, + selector: Option<&str>, + requested_runtime_id: Option<&str>, +) -> String { + let mut hasher = Sha256::new(); + for value in [Some(repository_id), selector, requested_runtime_id] { + match value { + Some(value) => { + hasher.update([1]); + hasher.update((value.len() as u64).to_be_bytes()); + hasher.update(value.as_bytes()); + } + None => hasher.update([0]), + } + } + let digest = hasher.finalize(); + let mut encoded = String::with_capacity(digest.len() * 2); + for byte in digest { + use std::fmt::Write as _; + write!(&mut encoded, "{byte:02x}").expect("writing to String cannot fail"); + } + format!("sha256:{encoded}") +} + +impl SqliteWorkspaceStore { + pub fn reserve_workdir_create_operation( + &self, + record: &WorkdirCreateOperationRecord, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction()?; + tx.execute( + r#"INSERT OR IGNORE INTO workdir_create_operations ( + workspace_id, operation_id, request_fingerprint, repository_id, selector, + requested_runtime_id, resolved_runtime_id, config_revision, + config_projection_digest, working_directory_id, state, failure, + created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)"#, + params![ + record.workspace_id, + record.operation_id, + record.request_fingerprint, + record.repository_id, + record.selector, + record.requested_runtime_id, + record.resolved_runtime_id, + record.config_revision as i64, + record.config_projection_digest, + record.working_directory_id, + record.state, + record.failure, + record.created_at, + record.updated_at, + ], + )?; + let persisted = + read_workdir_create_operation(&tx, &record.workspace_id, &record.operation_id)? + .ok_or_else(|| { + Error::RegistryInconsistency(format!( + "Workdir create operation `{}` was not persisted", + record.operation_id + )) + })?; + if persisted.request_fingerprint != record.request_fingerprint { + return Err(Error::InvalidInput(format!( + "Workdir create operation `{}` was reused with different input", + record.operation_id + ))); + } + tx.commit()?; + Ok(persisted) + }) + } + + pub fn finish_workdir_create_operation( + &self, + workspace_id: &str, + operation_id: &str, + request_fingerprint: &str, + succeeded: bool, + failure: Option<&str>, + updated_at: &str, + ) -> Result { + self.with_conn_mut(|conn| { + let changed = conn.execute( + r#"UPDATE workdir_create_operations + SET state = ?1, failure = ?2, updated_at = ?3 + WHERE workspace_id = ?4 AND operation_id = ?5 + AND request_fingerprint = ?6"#, + params![ + if succeeded { "succeeded" } else { "failed" }, + failure, + updated_at, + workspace_id, + operation_id, + request_fingerprint, + ], + )?; + if changed != 1 { + return Err(Error::RegistryInconsistency(format!( + "Workdir create operation `{operation_id}` could not be finalized" + ))); + } + read_workdir_create_operation(conn, workspace_id, operation_id)?.ok_or_else(|| { + Error::RegistryInconsistency(format!( + "Workdir create operation `{operation_id}` disappeared" + )) + }) + }) + } + + pub fn load_workdir_create_operation( + &self, + workspace_id: &str, + operation_id: &str, + ) -> Result> { + self.with_conn(|conn| read_workdir_create_operation(conn, workspace_id, operation_id)) + } +} + +fn read_workdir_create_operation( + conn: &rusqlite::Connection, + workspace_id: &str, + operation_id: &str, +) -> Result> { + conn.query_row( + r#"SELECT workspace_id, operation_id, request_fingerprint, repository_id, selector, + requested_runtime_id, resolved_runtime_id, config_revision, + config_projection_digest, working_directory_id, state, failure, + created_at, updated_at + FROM workdir_create_operations + WHERE workspace_id = ?1 AND operation_id = ?2"#, + params![workspace_id, operation_id], + |row| { + Ok(WorkdirCreateOperationRecord { + workspace_id: row.get(0)?, + operation_id: row.get(1)?, + request_fingerprint: row.get(2)?, + repository_id: row.get(3)?, + selector: row.get(4)?, + requested_runtime_id: row.get(5)?, + resolved_runtime_id: row.get(6)?, + config_revision: row.get::<_, i64>(7)? as u64, + config_projection_digest: row.get(8)?, + working_directory_id: row.get(9)?, + state: row.get(10)?, + failure: row.get(11)?, + created_at: row.get(12)?, + updated_at: row.get(13)?, + }) + }, + ) + .optional() + .map_err(Error::from) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::store::{ControlPlaneStore, RepositoryRecord, WorkspaceRecord}; + + #[test] + fn retry_keeps_resolved_config_evidence_and_rejects_changed_input() { + let store = SqliteWorkspaceStore::in_memory().unwrap(); + futures::executor::block_on(store.upsert_workspace(&WorkspaceRecord { + workspace_id: "workspace".to_string(), + owner_account_id: None, + display_name: "Workspace".to_string(), + state: "active".to_string(), + created_at: "2026-08-24T00:00:00Z".to_string(), + updated_at: "2026-08-24T00:00:00Z".to_string(), + })) + .unwrap(); + store + .upsert_repository(&RepositoryRecord { + workspace_id: "workspace".to_string(), + repository_id: "main".to_string(), + name: "main".to_string(), + kind: "git".to_string(), + provider: Some("git".to_string()), + uri: "/tmp/main".to_string(), + default_ref: Some("develop".to_string()), + auth_ref_kind: None, + auth_ref_key: None, + created_at: "2026-08-24T00:00:00Z".to_string(), + updated_at: "2026-08-24T00:00:00Z".to_string(), + }) + .unwrap(); + let record = WorkdirCreateOperationRecord { + workspace_id: "workspace".to_string(), + operation_id: "call-1".to_string(), + request_fingerprint: request_fingerprint("main", Some("develop"), None), + repository_id: "main".to_string(), + selector: Some("develop".to_string()), + requested_runtime_id: None, + resolved_runtime_id: "arcadia".to_string(), + config_revision: 7, + config_projection_digest: "sha256:projection".to_string(), + working_directory_id: "wd-1".to_string(), + state: "pending".to_string(), + failure: None, + created_at: "2026-08-24T00:00:00Z".to_string(), + updated_at: "2026-08-24T00:00:00Z".to_string(), + }; + assert_eq!( + store.reserve_workdir_create_operation(&record).unwrap(), + record + ); + let mut changed_resolution = record.clone(); + changed_resolution.resolved_runtime_id = "other".to_string(); + changed_resolution.config_revision = 8; + assert_eq!( + store + .reserve_workdir_create_operation(&changed_resolution) + .unwrap(), + record + ); + assert_eq!( + store + .load_workdir_create_operation("workspace", "call-1") + .unwrap(), + Some(record.clone()) + ); + let mut changed_input = record.clone(); + changed_input.request_fingerprint = request_fingerprint("main", Some("main"), None); + assert!( + store + .reserve_workdir_create_operation(&changed_input) + .unwrap_err() + .to_string() + .contains("reused with different input") + ); + } +}