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

feat: resolve default Runtime for Workdir creation

Browse Source
This commit is contained in:
Hare
2026-08-24 13:15:18 +09:00
parent d69c367285
commit 4f84dfd73f
7 changed files with 979 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workdir/src/workspace.rs
+2
View File
@@ -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<WorkingDirectoryDiagnostic>,
}
@@ -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(),
};
crates/worker/src/feature/builtin/manage_workdir.rs
+90 -27
View File
@@ -398,24 +398,35 @@ impl WorkspaceHttpWorkdirBackend {
workdir_output(format!("Listed {count} Workdir(s)"), &response)
}
fn create(&self, input: WorkdirCreateInput) -> Result<ToolOutput, ToolError> {
let runtime_id = validate_identity(&input.runtime_id, CREATE_TOOL, "runtime_id")?;
fn create(
&self,
input: WorkdirCreateInput,
operation_id: String,
) -> Result<ToolOutput, ToolError> {
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::<WorkdirDetailResponse>(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<ToolOutput, ToolError> {
match self.operation {
WorkdirOperation::List => {
let _input = parse_input::<WorkdirListInput>(input_json)?;
self.backend.list()
}
WorkdirOperation::Create => self
.backend
.create(parse_input::<WorkdirCreateInput>(input_json)?),
WorkdirOperation::Create => self.backend.create(
parse_input::<WorkdirCreateInput>(input_json)?,
ctx.call_id.to_string(),
),
WorkdirOperation::Attach => self
.backend
.attach(parse_input::<WorkdirAttachInput>(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<String>,
repository_id: String,
#[serde(default)]
selector: Option<String>,
@@ -685,10 +698,12 @@ struct WorkdirCreateInput {
#[derive(Debug, Serialize)]
struct WorkdirCreateRequest {
runtime_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
runtime_id: Option<String>,
repository_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
selector: Option<String>,
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());
crates/workspace-server/src/lib.rs
+2
View File
@@ -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;
crates/workspace-server/src/runtime_settings.rs
+187
View File
@@ -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> {
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<String>,
}
#[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<RuntimeConfigProjection> {
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<Option<String>> {
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()
);
}
}
crates/workspace-server/src/server.rs
+358 -77
View File
@@ -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<String>,
}
#[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<String>,
pub repository_id: String,
#[serde(default)]
pub selector: Option<String>,
#[serde(default)]
pub operation_id: Option<String>,
}
#[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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimePath>,
Json(mut request): Json<BrowserWorkingDirectoryCreateRequest>,
) -> ApiResult<Json<BrowserWorkingDirectoryDetailResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
request.runtime_id = path.runtime_id;
create_working_directory_for_runtime(api, request)
Json(request): Json<BrowserWorkingDirectoryCreateRequest>,
) -> ApiResult<(StatusCode, Json<BrowserWorkingDirectoryDetailResponse>)> {
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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
Json(_request): Json<BrowserWorkingDirectoryCreateRequest>,
) -> ApiResult<Json<BrowserWorkingDirectoryDetailResponse>> {
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<BrowserWorkingDirectoryCreateRequest>,
) -> ApiResult<(StatusCode, Json<BrowserWorkingDirectoryDetailResponse>)> {
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<Json<BrowserWorkingDirectoryDetailResponse>> {
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<BrowserWorkingDirectoryDetailResponse>)> {
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();
crates/workspace-server/src/store.rs
+99 -18
View File
@@ -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<TicketCoderAssignmentRecord>,
}
#[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<String>,
pub requested_runtime_id: Option<String>,
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<String>,
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<String> = 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<String> = 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(),
crates/workspace-server/src/workdir_create_operations.rs
+241
View File
@@ -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<WorkdirCreateOperationRecord> {
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<WorkdirCreateOperationRecord> {
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<Option<WorkdirCreateOperationRecord>> {
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<Option<WorkdirCreateOperationRecord>> {
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")
);
}
}