diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 7837a9ef..55d8210c 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -117,6 +117,10 @@ pub struct WorkingDirectoryRequest { pub materializer: MaterializerKind, #[serde(default)] pub dirty_state_policy: DirtyStatePolicy, + /// Backend-assigned stable Workdir id. Runtimes use this when present so the + /// Backend can create canonical registry rows before materialization. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub backend_workdir_id: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -158,6 +162,10 @@ pub struct WorkingDirectorySummary { #[serde(default, skip_serializing_if = "Option::is_none")] pub cleanup_policy: Option, pub status: WorkingDirectoryStatusKind, + /// Backend projection metadata. Runtimes leave this absent; Workspace Browser + /// APIs fill it with `backend_managed` or `runtime_unmanaged`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub management_kind: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index e11a07b3..6743f3bc 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1036,6 +1036,7 @@ mod tests { }, materializer: MaterializerKind::LocalGitWorktree, dirty_state_policy: DirtyStatePolicy::CleanPointOnly, + backend_workdir_id: None, } } diff --git a/crates/worker-runtime/src/working_directory.rs b/crates/worker-runtime/src/working_directory.rs index fb4593e8..76f7c007 100644 --- a/crates/worker-runtime/src/working_directory.rs +++ b/crates/worker-runtime/src/working_directory.rs @@ -51,6 +51,7 @@ impl WorkingDirectory { cleanup_target: Some(self.cleanup_target.clone()), cleanup_policy: Some(self.cleanup_policy.clone()), status: self.status.clone(), + management_kind: None, } } } @@ -395,7 +396,10 @@ impl WorkingDirectoryMaterializer for LocalGitWorktreeMaterializer { &self, request: &WorkingDirectoryRequest, ) -> Result { - let working_directory_id = next_working_directory_id(&request.repository.id); + let working_directory_id = request + .backend_workdir_id + .clone() + .unwrap_or_else(|| next_working_directory_id(&request.repository.id)); self.materialize_with_working_directory_id( working_directory_id, request, @@ -719,6 +723,7 @@ mod tests { }, materializer: MaterializerKind::LocalGitWorktree, dirty_state_policy: DirtyStatePolicy::CleanPointOnly, + backend_workdir_id: None, } } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 47c87934..1e3988c7 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -24,10 +24,11 @@ use crate::config::{RemoteRuntimeConfigFile, WorkspaceBackendConfigFile, resolve use crate::hosts::{ ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EmbeddedWorkerRuntime, HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, RuntimeDiagnostic, RuntimeRegistry, - RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerInputRequest, WorkerInputResult, - WorkerLifecycleRequest, WorkerLifecycleResult, WorkerOperationState, - WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, - WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerTranscriptProjection, + RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary, + WorkerImplementationSummary, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, + WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement, + WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, + WorkerSummary, WorkerTranscriptProjection, WorkerWorkspaceSummary, }; use crate::identity::WorkspaceIdentity; use crate::observation::{ @@ -48,12 +49,16 @@ use crate::repositories::{ RepositoryRegistryReader, RepositorySummary, }; use crate::resource_broker::BackendResourceBroker; -use crate::store::{ControlPlaneStore, WorkspaceRecord}; +use crate::store::{ + ControlPlaneStore, WorkdirRegistryRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, + WorkspaceRecord, +}; use crate::{Error, Result}; use worker_runtime::catalog::{ ConfigBundleRef, DirtyStatePolicy, MaterializerKind, ProfileSelector, RepositorySelector as RuntimeRepositorySelector, WorkingDirectoryClaim, - WorkingDirectoryRepository, WorkingDirectoryRequest, WorkingDirectorySummary, + WorkingDirectoryRepository, WorkingDirectoryRequest, WorkingDirectoryStatusKind, + WorkingDirectorySummary, }; use worker_runtime::config_bundle::ConfigBundle; use worker_runtime::http_server::{ @@ -1182,18 +1187,11 @@ async fn scoped_list_runtime_working_directories( AxumPath(path): AxumPath, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let list = api - .runtime - .list_working_directories(&path.runtime_id) - .map_err(|err| err.into_error())?; + let (items, diagnostics) = runtime_working_directory_summaries(&api, &path.runtime_id)?; Ok(Json(BrowserWorkingDirectoryListResponse { workspace_id: api.config.workspace_id.clone(), - items: list - .items - .into_iter() - .map(|status| status.summary) - .collect(), - diagnostics: list.diagnostics, + items, + diagnostics, })) } @@ -1274,12 +1272,36 @@ fn create_working_directory_for_runtime( request: BrowserWorkingDirectoryCreateRequest, ) -> ApiResult> { let runtime_id = request.runtime_id.clone(); - let working_directory_request = working_directory_request_for_browser(&api, request)?; + 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(), + selector: working_directory_request + .repository + .selector + .as_ref() + .map(|selector| selector.as_ref().to_string()), + resolved_commit: None, + materialization_status: "pending".to_string(), + cleanliness: "clean".to_string(), + management_kind: "backend_managed".to_string(), + created_at: now_registry_timestamp(), + updated_at: now_registry_timestamp(), + }; + api.store.upsert_workdir_registry(&pending)?; let result = api .runtime .create_working_directory(&runtime_id, working_directory_request) .map_err(|err| err.into_error())?; 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)?; return Err(ApiError::with_diagnostics( Error::RuntimeOperationFailed { runtime_id, @@ -1289,6 +1311,13 @@ fn create_working_directory_for_runtime( result.diagnostics, )); }; + let record = workdir_record_from_summary( + &api, + &runtime_id, + &working_directory.summary, + "backend_managed", + ); + api.store.upsert_workdir_registry(&record)?; Ok(Json(BrowserWorkingDirectoryDetailResponse { workspace_id: api.config.workspace_id.clone(), item: working_directory.summary, @@ -1305,21 +1334,43 @@ fn working_directory_detail_for_runtime( .runtime .working_directory(runtime_id, working_directory_id) .map_err(|err| err.into_error())?; - let Some(working_directory) = result.working_directory else { - return Err(ApiError::with_diagnostics( - Error::RuntimeOperationFailed { - runtime_id: runtime_id.to_string(), - code: "workspace_working_directory_lookup_failed".to_string(), - message: "Runtime did not return working directory".to_string(), - }, - result.diagnostics, - )); - }; - Ok(Json(BrowserWorkingDirectoryDetailResponse { - workspace_id: api.config.workspace_id.clone(), - item: working_directory.summary, - diagnostics: result.diagnostics, - })) + if let Some(working_directory) = result.working_directory { + let management_kind = api + .store + .get_workdir_registry(&api.config.workspace_id, working_directory_id)? + .map(|record| record.management_kind) + .unwrap_or_else(|| "runtime_unmanaged".to_string()); + let record = workdir_record_from_summary( + &api, + runtime_id, + &working_directory.summary, + management_kind.as_str(), + ); + api.store.upsert_workdir_registry(&record)?; + return Ok(Json(BrowserWorkingDirectoryDetailResponse { + workspace_id: api.config.workspace_id.clone(), + item: working_directory.summary, + diagnostics: result.diagnostics, + })); + } + if let Some(record) = api + .store + .get_workdir_registry(&api.config.workspace_id, working_directory_id)? + { + return Ok(Json(BrowserWorkingDirectoryDetailResponse { + workspace_id: api.config.workspace_id.clone(), + item: workdir_summary_from_record(&record), + diagnostics: result.diagnostics, + })); + } + Err(ApiError::with_diagnostics( + Error::RuntimeOperationFailed { + runtime_id: runtime_id.to_string(), + code: "workspace_working_directory_lookup_failed".to_string(), + message: "Runtime did not return working directory".to_string(), + }, + result.diagnostics, + )) } fn cleanup_working_directory_for_runtime( @@ -1341,6 +1392,18 @@ fn cleanup_working_directory_for_runtime( result.diagnostics, )); }; + let management_kind = api + .store + .get_workdir_registry(&api.config.workspace_id, working_directory_id)? + .map(|record| record.management_kind) + .unwrap_or_else(|| "runtime_unmanaged".to_string()); + let record = workdir_record_from_summary( + &api, + runtime_id, + &working_directory.summary, + management_kind.as_str(), + ); + api.store.upsert_workdir_registry(&record)?; Ok(Json(BrowserWorkingDirectoryDetailResponse { workspace_id: api.config.workspace_id.clone(), item: working_directory.summary, @@ -1966,6 +2029,7 @@ fn working_directory_request_from_repository( }, materializer: MaterializerKind::LocalGitWorktree, dirty_state_policy: DirtyStatePolicy::CleanPointOnly, + backend_workdir_id: None, } } @@ -2026,6 +2090,10 @@ async fn create_workspace_worker( content: initial_text, }) }; + let selected_working_directory_id = request + .working_directory + .as_ref() + .map(|selection| selection.working_directory_id.clone()); let resolved_working_directory = request .working_directory @@ -2039,7 +2107,7 @@ async fn create_workspace_worker( .spawn_worker( &request.runtime_id, WorkerSpawnRequest { - requested_worker_name: Some(display_name), + requested_worker_name: Some(display_name.clone()), intent: WorkerSpawnIntent::WorkspaceCoding, acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { expected_segments: if initial_input.is_some() { 1 } else { 0 }, @@ -2064,6 +2132,37 @@ async fn create_workspace_worker( code: "workspace_worker_create_missing_summary".to_string(), message: "Runtime completed worker creation without returning a Worker summary".to_string(), })?; + let worker_record = sync_worker_observation(&api, &worker)?; + if let Some(workdir_id) = selected_working_directory_id.as_deref() { + if api + .store + .get_workdir_registry(&api.config.workspace_id, workdir_id)? + .is_none() + { + if let Ok(result) = api + .runtime + .working_directory(worker.runtime_id.as_str(), workdir_id) + .map_err(|err| err.into_error()) + { + if let Some(status) = result.working_directory { + let record = workdir_record_from_summary( + &api, + worker.runtime_id.as_str(), + &status.summary, + "runtime_unmanaged", + ); + api.store.upsert_workdir_registry(&record)?; + } + } + } + if api + .store + .get_workdir_registry(&api.config.workspace_id, workdir_id)? + .is_some() + { + link_worker_to_workdir(&api, &worker_record, workdir_id)?; + } + } let runtime_id = worker.runtime_id.clone(); let worker_id = worker.worker_id.clone(); let workspace_id = api.workspace_id().to_string(); @@ -2172,10 +2271,47 @@ async fn create_runtime_worker( configured_working_directory_request(&api.config, working_directory) }) .transpose()?; + let prepared_workdir_id = if let Some(working_directory_request) = + request.resolved_working_directory_request.as_mut() + { + Some(upsert_pending_backend_workdir( + &api, + &runtime_id, + working_directory_request, + )?) + } else { + request + .resolved_working_directory + .as_ref() + .map(|claim| claim.working_directory_id.clone()) + }; let result = api .runtime .spawn_worker(&runtime_id, request) .map_err(|err| err.into_error())?; + if let Some(worker) = result.worker.as_ref() { + let record = sync_worker_observation(&api, worker)?; + if worker.working_directory.is_none() { + if let Some(workdir_id) = prepared_workdir_id.as_deref() { + if api + .store + .get_workdir_registry(&api.config.workspace_id, workdir_id)? + .is_some() + { + link_worker_to_workdir(&api, &record, workdir_id)?; + } + } + } + } else if let Some(workdir_id) = prepared_workdir_id.as_deref() { + if let Some(mut record) = api + .store + .get_workdir_registry(&api.config.workspace_id, workdir_id)? + { + record.materialization_status = "failed".to_string(); + record.updated_at = now_registry_timestamp(); + api.store.upsert_workdir_registry(&record)?; + } + } Ok(Json(result)) } @@ -2230,6 +2366,16 @@ async fn stop_runtime_worker( .runtime .stop_worker(&runtime_id, &worker_id, request) .map_err(|err| err.into_error())?; + let backend_id = backend_worker_id(&runtime_id, &worker_id); + if let Some(mut record) = api + .store + .get_worker_registry(&api.config.workspace_id, backend_id.as_str())? + { + record.lifecycle_state = "stopped".to_string(); + record.updated_at = now_registry_timestamp(); + api.store.upsert_worker_registry(&record)?; + sync_linked_workdir_after_worker_stop(&api, &runtime_id, &record)?; + } Ok(Json(result)) } @@ -2423,11 +2569,37 @@ async fn list_host_workers( fn workers_response(api: WorkspaceApi) -> ApiResult> { let limit = api.config.max_records.min(200); let runtime_workers = api.runtime.list_workers(limit); + let mut observed = std::collections::BTreeMap::new(); + for worker in &runtime_workers.items { + let _ = sync_worker_observation(&api, worker); + observed.insert( + backend_worker_id(worker.runtime_id.as_str(), worker.worker_id.as_str()), + worker.clone(), + ); + } + let worker_records = api + .store + .list_worker_registry(&api.config.workspace_id, limit)?; + let workdir_records = api + .store + .list_workdir_registry(&api.config.workspace_id, 500)?; + let mut items = Vec::new(); + for record in worker_records { + let links = api + .store + .list_worker_workdir_links(&api.config.workspace_id, record.worker_id.as_str())?; + items.push(merge_worker_registry_projection( + observed.get(record.worker_id.as_str()), + &record, + links, + &workdir_records, + )); + } Ok(RuntimeListResponse { workspace_id: api.config.workspace_id, limit, - items: runtime_workers.items, - source: "worker_runtime_registry".to_string(), + items, + source: "backend_worker_registry".to_string(), diagnostics: runtime_workers.diagnostics, }) } @@ -3126,25 +3298,377 @@ fn working_directory_repository_options( } fn working_directory_summaries(api: &WorkspaceApi) -> ApiResult> { - let list = api - .runtime - .list_working_directories(EMBEDDED_WORKER_RUNTIME_ID) - .map_err(|err| err.into_error())?; - if !list.diagnostics.is_empty() { - return Err(ApiError::with_diagnostics( - Error::RuntimeOperationFailed { - runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), - code: "workspace_working_directory_list_failed".to_string(), - message: "Runtime did not list working directories".to_string(), - }, - list.diagnostics, - )); + let _ = sync_all_runtime_workdir_observations(api); + let records = api + .store + .list_managed_workdir_registry(&api.config.workspace_id, 200)?; + Ok(records + .iter() + .map(workdir_summary_from_record) + .collect::>()) +} + +fn runtime_working_directory_summaries( + api: &WorkspaceApi, + runtime_id: &str, +) -> ApiResult<(Vec, Vec)> { + let diagnostics = sync_runtime_workdir_observations(api, runtime_id)?; + let records = api + .store + .list_workdir_registry(&api.config.workspace_id, 200)?; + let items = records + .iter() + .filter(|record| record.runtime_id == runtime_id) + .map(workdir_summary_from_record) + .collect::>(); + Ok((items, diagnostics)) +} + +fn backend_worker_id(runtime_id: &str, runtime_worker_id: &str) -> String { + format!("{runtime_id}/{runtime_worker_id}") +} + +fn now_registry_timestamp() -> String { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_millis().to_string()) + .unwrap_or_else(|_| "0".to_string()) +} + +fn registry_safe_id_component(value: &str) -> String { + let sanitized: String = value + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') { + ch + } else { + '-' + } + }) + .collect(); + sanitized.trim_matches('-').chars().take(48).collect() +} + +fn next_backend_workdir_id(repository_id: &str) -> String { + let repository = registry_safe_id_component(repository_id); + format!( + "backend-{}-{}", + now_registry_timestamp(), + if repository.is_empty() { + "workdir" + } else { + &repository + } + ) +} + +fn record_worker_summary( + api: &WorkspaceApi, + worker: &WorkerSummary, + display_name: &str, + profile: Option, +) -> ApiResult { + let timestamp = now_registry_timestamp(); + let worker_id = backend_worker_id(worker.runtime_id.as_str(), worker.worker_id.as_str()); + let record = WorkerRegistryRecord { + workspace_id: api.config.workspace_id.clone(), + worker_id: worker_id.clone(), + runtime_id: worker.runtime_id.as_str().to_string(), + runtime_worker_id: worker.worker_id.as_str().to_string(), + display_name: display_name.to_string(), + profile, + lifecycle_state: worker.status.clone(), + retention_state: "normal".to_string(), + transcript_ref: Some(format!( + "runtime://{}/workers/{}/transcript", + worker.runtime_id.as_str(), + worker.worker_id.as_str() + )), + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: timestamp.clone(), + updated_at: timestamp, + }; + api.store.upsert_worker_registry(&record)?; + Ok(api + .store + .get_worker_registry(&api.config.workspace_id, worker_id.as_str())? + .unwrap_or(record)) +} + +fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary { + WorkerSummary { + worker_id: record.runtime_worker_id.clone(), + runtime_id: record.runtime_id.clone(), + host_id: "backend-registry".to_string(), + role: None, + label: record.display_name.clone(), + status: record.lifecycle_state.clone(), + state: record.lifecycle_state.clone(), + last_seen_at: Some(record.updated_at.clone()), + capabilities: WorkerCapabilitySummary { + can_accept_input: false, + can_stop: false, + can_spawn_followup: false, + }, + workspace: WorkerWorkspaceSummary { + visibility: "backend_registry".to_string(), + identity: record.workspace_id.clone(), + }, + profile: record.profile.clone(), + implementation: WorkerImplementationSummary { + kind: "backend_worker_registry".to_string(), + display_hint: "Archived Worker".to_string(), + }, + working_directory: None, + diagnostics: vec![RuntimeDiagnostic { + code: "backend_worker_registry_only".to_string(), + severity: DiagnosticSeverity::Info, + message: + "Worker is preserved in the Backend registry without a live Runtime observation" + .to_string(), + }], } - Ok(list - .items +} + +fn merge_worker_registry_projection( + live: Option<&WorkerSummary>, + record: &WorkerRegistryRecord, + links: Vec, + workdirs: &[WorkdirRegistryRecord], +) -> WorkerSummary { + let mut summary = live + .cloned() + .unwrap_or_else(|| worker_summary_from_registry(record)); + summary.label = record.display_name.clone(); + summary.status = record.lifecycle_state.clone(); + summary.state = record.lifecycle_state.clone(); + summary.profile = record.profile.clone(); + summary.working_directory = links.iter().find_map(|link| { + workdirs + .iter() + .find(|workdir| workdir.workdir_id == link.workdir_id) + .map(|workdir| workdir_summary_from_record(workdir)) + }); + summary +} + +fn sync_worker_observation( + api: &WorkspaceApi, + worker: &WorkerSummary, +) -> ApiResult { + let record = record_worker_summary(api, worker, worker.label.as_str(), worker.profile.clone())?; + if let Some(working_directory) = worker.working_directory.as_ref() { + let management_kind = api + .store + .get_workdir_registry( + &api.config.workspace_id, + &working_directory.working_directory_id, + )? + .map(|existing| existing.management_kind) + .unwrap_or_else(|| "runtime_unmanaged".to_string()); + let workdir_record = workdir_record_from_summary( + api, + worker.runtime_id.as_str(), + working_directory, + management_kind.as_str(), + ); + api.store.upsert_workdir_registry(&workdir_record)?; + link_worker_to_workdir(api, &record, &working_directory.working_directory_id)?; + } + Ok(record) +} + +fn upsert_pending_backend_workdir( + api: &WorkspaceApi, + runtime_id: &str, + request: &mut WorkingDirectoryRequest, +) -> ApiResult { + let workdir_id = request + .backend_workdir_id + .clone() + .unwrap_or_else(|| next_backend_workdir_id(&request.repository.id)); + request.backend_workdir_id = Some(workdir_id.clone()); + let timestamp = now_registry_timestamp(); + api.store.upsert_workdir_registry(&WorkdirRegistryRecord { + workspace_id: api.config.workspace_id.clone(), + workdir_id: workdir_id.clone(), + runtime_id: runtime_id.to_string(), + repository_id: request.repository.id.clone(), + selector: request + .repository + .selector + .as_ref() + .map(|selector| selector.as_ref().to_string()), + resolved_commit: None, + materialization_status: "pending".to_string(), + cleanliness: "clean".to_string(), + management_kind: "backend_managed".to_string(), + created_at: timestamp.clone(), + updated_at: timestamp, + })?; + Ok(workdir_id) +} + +fn sync_runtime_workdir_observations( + api: &WorkspaceApi, + runtime_id: &str, +) -> ApiResult> { + let response = api + .runtime + .list_working_directories(runtime_id) + .map_err(|err| err.into_error())?; + let mut observed = std::collections::BTreeSet::new(); + for status in &response.items { + observed.insert(status.summary.working_directory_id.clone()); + let management_kind = api + .store + .get_workdir_registry( + &api.config.workspace_id, + &status.summary.working_directory_id, + )? + .map(|existing| existing.management_kind) + .unwrap_or_else(|| "runtime_unmanaged".to_string()); + let record = + workdir_record_from_summary(api, runtime_id, &status.summary, management_kind.as_str()); + api.store.upsert_workdir_registry(&record)?; + } + for mut record in api + .store + .list_workdir_registry(&api.config.workspace_id, 500)? .into_iter() - .map(|status| status.summary) - .collect()) + .filter(|record| record.runtime_id == runtime_id && !observed.contains(&record.workdir_id)) + { + if record.materialization_status == "present" { + record.materialization_status = "missing".to_string(); + record.updated_at = now_registry_timestamp(); + api.store.upsert_workdir_registry(&record)?; + } + } + Ok(response.diagnostics) +} + +fn sync_all_runtime_workdir_observations(api: &WorkspaceApi) -> Vec { + let mut diagnostics = Vec::new(); + let runtimes = api.runtime.list_runtimes(api.config.max_records.min(200)); + for runtime in runtimes.items { + if runtime.capabilities.supports_worktrees { + match sync_runtime_workdir_observations(api, runtime.runtime_id.as_str()) { + Ok(mut runtime_diagnostics) => diagnostics.append(&mut runtime_diagnostics), + Err(err) => diagnostics.extend(err.diagnostics), + } + } + } + diagnostics +} + +fn sync_linked_workdir_after_worker_stop( + api: &WorkspaceApi, + runtime_id: &str, + worker_record: &WorkerRegistryRecord, +) -> ApiResult<()> { + let links = api + .store + .list_worker_workdir_links(&api.config.workspace_id, worker_record.worker_id.as_str())?; + for link in links { + let result = api + .runtime + .working_directory(runtime_id, link.workdir_id.as_str()) + .map_err(|err| err.into_error())?; + if let Some(status) = result.working_directory { + let management_kind = api + .store + .get_workdir_registry(&api.config.workspace_id, link.workdir_id.as_str())? + .map(|record| record.management_kind) + .unwrap_or_else(|| "runtime_unmanaged".to_string()); + let record = workdir_record_from_summary( + api, + runtime_id, + &status.summary, + management_kind.as_str(), + ); + api.store.upsert_workdir_registry(&record)?; + } else if let Some(mut record) = api + .store + .get_workdir_registry(&api.config.workspace_id, link.workdir_id.as_str())? + { + record.materialization_status = "missing".to_string(); + record.updated_at = now_registry_timestamp(); + api.store.upsert_workdir_registry(&record)?; + } + } + Ok(()) +} + +fn workdir_record_from_summary( + api: &WorkspaceApi, + runtime_id: &str, + summary: &WorkingDirectorySummary, + management_kind: &str, +) -> WorkdirRegistryRecord { + let timestamp = now_registry_timestamp(); + WorkdirRegistryRecord { + workspace_id: api.config.workspace_id.clone(), + workdir_id: summary.working_directory_id.clone(), + runtime_id: runtime_id.to_string(), + repository_id: summary.repository_id.clone(), + selector: summary.requested_selector.clone(), + resolved_commit: summary.resolved_commit.clone(), + materialization_status: match summary.status { + WorkingDirectoryStatusKind::Active => "present", + WorkingDirectoryStatusKind::Removed => "removed", + WorkingDirectoryStatusKind::CleanupPending => "pending", + } + .to_string(), + cleanliness: "clean".to_string(), + management_kind: management_kind.to_string(), + created_at: timestamp.clone(), + updated_at: timestamp, + } +} + +fn workdir_summary_from_record(record: &WorkdirRegistryRecord) -> WorkingDirectorySummary { + let status = match record.materialization_status.as_str() { + "present" => WorkingDirectoryStatusKind::Active, + "pending" => WorkingDirectoryStatusKind::CleanupPending, + _ => WorkingDirectoryStatusKind::Removed, + }; + WorkingDirectorySummary { + working_directory_id: record.workdir_id.clone(), + repository_id: record.repository_id.clone(), + requested_selector: record.selector.clone(), + materializer_kind: MaterializerKind::LocalGitWorktree, + dirty_state_policy: DirtyStatePolicy::CleanPointOnly, + resolved_commit: record.resolved_commit.clone(), + resolved_tree: None, + cleanup_target: Some(worker_runtime::catalog::WorkingDirectoryCleanupTarget { + kind: "local_git_worktree".to_string(), + working_directory_id: record.workdir_id.clone(), + repository_id: record.repository_id.clone(), + }), + cleanup_policy: Some("manual_or_worker_stop".to_string()), + status, + management_kind: Some(record.management_kind.clone()), + } +} + +fn link_worker_to_workdir( + api: &WorkspaceApi, + worker_record: &WorkerRegistryRecord, + workdir_id: &str, +) -> ApiResult<()> { + let timestamp = now_registry_timestamp(); + api.store + .upsert_worker_workdir_link(&WorkerWorkdirLinkRecord { + workspace_id: api.config.workspace_id.clone(), + worker_id: worker_record.worker_id.clone(), + workdir_id: workdir_id.to_string(), + role: "primary_cwd".to_string(), + linked_at: timestamp, + unlinked_at: None, + })?; + Ok(()) } fn validate_working_directory_claim_for_browser( @@ -3207,6 +3731,7 @@ fn working_directory_request_for_browser( }, materializer: MaterializerKind::LocalGitWorktree, dirty_state_policy: DirtyStatePolicy::CleanPointOnly, + backend_workdir_id: None, }) } @@ -3711,6 +4236,148 @@ mod tests { const TEST_REPOSITORY_ID: &str = "main"; const TEST_CREATED_AT: &str = "2026-06-23T06:43:28Z"; + #[test] + fn backend_worker_projection_preserves_archive_rows_links_and_redacts_paths() { + let worker = WorkerRegistryRecord { + workspace_id: "workspace-1".to_string(), + worker_id: "embedded/worker-1".to_string(), + runtime_id: "embedded".to_string(), + runtime_worker_id: "worker-1".to_string(), + display_name: "Archived Worker".to_string(), + profile: Some("builtin:coder".to_string()), + lifecycle_state: "stopped".to_string(), + retention_state: "pinned".to_string(), + transcript_ref: Some("runtime://embedded/workers/worker-1/transcript".to_string()), + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: "1".to_string(), + updated_at: "2".to_string(), + }; + let workdir = WorkdirRegistryRecord { + workspace_id: "workspace-1".to_string(), + workdir_id: "backend-1-repo".to_string(), + runtime_id: "embedded".to_string(), + repository_id: "repo".to_string(), + selector: Some("develop".to_string()), + resolved_commit: Some("abcdef".to_string()), + materialization_status: "missing".to_string(), + cleanliness: "clean".to_string(), + management_kind: "backend_managed".to_string(), + created_at: "1".to_string(), + updated_at: "3".to_string(), + }; + let link = WorkerWorkdirLinkRecord { + workspace_id: "workspace-1".to_string(), + worker_id: worker.worker_id.clone(), + workdir_id: workdir.workdir_id.clone(), + role: "primary_cwd".to_string(), + linked_at: "4".to_string(), + unlinked_at: None, + }; + + let projected = merge_worker_registry_projection(None, &worker, vec![link], &[workdir]); + + assert_eq!(projected.status, "stopped"); + assert_eq!( + projected.working_directory.as_ref().unwrap().status, + WorkingDirectoryStatusKind::Removed + ); + assert_eq!( + projected + .working_directory + .as_ref() + .unwrap() + .management_kind + .as_deref(), + Some("backend_managed") + ); + let serialized = serde_json::to_string(&projected).unwrap(); + assert!(!serialized.contains("/tmp/")); + assert!(!serialized.contains("materialized_path")); + } + + #[tokio::test] + async fn workspace_managed_workdir_summaries_exclude_runtime_unmanaged_rows() { + let dir = tempfile::tempdir().unwrap(); + let api = test_api(dir.path()).await; + api.store + .upsert_workdir_registry(&WorkdirRegistryRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + workdir_id: "managed".to_string(), + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + repository_id: "repo".to_string(), + selector: None, + resolved_commit: None, + materialization_status: "present".to_string(), + cleanliness: "clean".to_string(), + management_kind: "backend_managed".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }) + .unwrap(); + api.store + .upsert_workdir_registry(&WorkdirRegistryRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + workdir_id: "runtime-direct".to_string(), + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + repository_id: "repo".to_string(), + selector: None, + resolved_commit: None, + materialization_status: "present".to_string(), + cleanliness: "unknown".to_string(), + management_kind: "runtime_unmanaged".to_string(), + created_at: "1".to_string(), + updated_at: "2".to_string(), + }) + .unwrap(); + + let managed = working_directory_summaries(&api) + .unwrap_or_else(|err| panic!("working_directory_summaries failed: {}", err.error)); + assert_eq!(managed.len(), 1); + assert_eq!(managed[0].working_directory_id, "managed"); + assert_eq!( + managed[0].management_kind.as_deref(), + Some("backend_managed") + ); + + let (runtime_projection, _) = + runtime_working_directory_summaries(&api, EMBEDDED_WORKER_RUNTIME_ID).unwrap_or_else( + |err| panic!("runtime_working_directory_summaries failed: {}", err.error), + ); + assert!(runtime_projection.iter().any(|summary| { + summary.working_directory_id == "runtime-direct" + && summary.management_kind.as_deref() == Some("runtime_unmanaged") + })); + } + #[test] + fn unmanaged_runtime_workdir_projection_is_typed_and_diagnostic_safe() { + let workdir = WorkdirRegistryRecord { + workspace_id: "workspace-1".to_string(), + workdir_id: "runtime-direct".to_string(), + runtime_id: "embedded".to_string(), + repository_id: "repo".to_string(), + selector: None, + resolved_commit: None, + materialization_status: "present".to_string(), + cleanliness: "unknown".to_string(), + management_kind: "runtime_unmanaged".to_string(), + created_at: "1".to_string(), + updated_at: "2".to_string(), + }; + + let projected = workdir_summary_from_record(&workdir); + + assert_eq!(projected.status, WorkingDirectoryStatusKind::Active); + assert_eq!( + projected.management_kind.as_deref(), + Some("runtime_unmanaged") + ); + let serialized = serde_json::to_string(&projected).unwrap(); + assert!(!serialized.contains("/tmp/")); + assert!(!serialized.contains("materialized_path")); + } + #[test] fn worker_profile_candidates_are_backend_published_and_mapped() { let candidates = worker_profile_candidates(); diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index c9ec8b0f..ca830c23 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -27,6 +27,11 @@ const MIGRATIONS: &[Migration] = &[ name: "align legacy workspace bootstrap with schema v0", apply: align_legacy_bootstrap_schema, }, + Migration { + version: 3, + name: "backend worker workdir registry schema", + apply: create_worker_workdir_registry_tables, + }, ]; struct Migration { @@ -44,11 +49,99 @@ pub struct WorkspaceRecord { pub updated_at: String, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerRegistryRecord { + pub workspace_id: String, + /// Backend-owned archival Worker id. In v0 it is derived from runtime_id + runtime_worker_id. + pub worker_id: String, + pub runtime_id: String, + pub runtime_worker_id: String, + pub display_name: String, + pub profile: Option, + pub lifecycle_state: String, + /// Retention state is explicit so `pinned` can be represented before prune exists. + pub retention_state: String, + pub transcript_ref: Option, + pub session_ref: Option, + pub summary_ref: Option, + pub diagnostics_ref: Option, + pub created_at: String, + pub updated_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkdirRegistryRecord { + pub workspace_id: String, + pub workdir_id: String, + pub runtime_id: String, + pub repository_id: String, + pub selector: Option, + pub resolved_commit: Option, + pub materialization_status: String, + pub cleanliness: String, + /// `backend_managed` rows are authored by this Backend; `runtime_unmanaged` is for diagnostics only. + pub management_kind: String, + pub created_at: String, + pub updated_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerWorkdirLinkRecord { + pub workspace_id: String, + pub worker_id: String, + pub workdir_id: String, + pub role: String, + pub linked_at: String, + pub unlinked_at: Option, +} + #[async_trait] pub trait ControlPlaneStore: Send + Sync { async fn schema_version(&self) -> Result; async fn upsert_workspace(&self, record: &WorkspaceRecord) -> Result<()>; async fn get_workspace(&self, workspace_id: &str) -> Result>; + + fn upsert_worker_registry(&self, record: &WorkerRegistryRecord) -> Result<()>; + fn get_worker_registry( + &self, + workspace_id: &str, + worker_id: &str, + ) -> Result>; + fn get_worker_registry_by_runtime( + &self, + workspace_id: &str, + runtime_id: &str, + runtime_worker_id: &str, + ) -> Result>; + fn list_worker_registry( + &self, + workspace_id: &str, + limit: usize, + ) -> Result>; + + fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()>; + fn get_workdir_registry( + &self, + workspace_id: &str, + workdir_id: &str, + ) -> Result>; + fn list_workdir_registry( + &self, + workspace_id: &str, + limit: usize, + ) -> Result>; + fn list_managed_workdir_registry( + &self, + workspace_id: &str, + limit: usize, + ) -> Result>; + + fn upsert_worker_workdir_link(&self, record: &WorkerWorkdirLinkRecord) -> Result<()>; + fn list_worker_workdir_links( + &self, + workspace_id: &str, + worker_id: &str, + ) -> Result>; } #[derive(Clone)] @@ -131,6 +224,355 @@ impl ControlPlaneStore for SqliteWorkspaceStore { .map_err(Error::from) }) } + + fn upsert_worker_registry(&self, record: &WorkerRegistryRecord) -> Result<()> { + self.with_conn(|conn| { + conn.execute( + r#"INSERT INTO worker_registry ( + workspace_id, worker_id, runtime_id, runtime_worker_id, display_name, profile, + lifecycle_state, retention_state, transcript_ref, session_ref, summary_ref, + diagnostics_ref, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) + ON CONFLICT(workspace_id, worker_id) DO UPDATE SET + runtime_id = excluded.runtime_id, + runtime_worker_id = excluded.runtime_worker_id, + display_name = excluded.display_name, + profile = excluded.profile, + lifecycle_state = excluded.lifecycle_state, + retention_state = CASE + WHEN worker_registry.retention_state = 'pinned' AND excluded.retention_state = 'normal' + THEN worker_registry.retention_state + ELSE excluded.retention_state + END, + transcript_ref = excluded.transcript_ref, + session_ref = excluded.session_ref, + summary_ref = excluded.summary_ref, + diagnostics_ref = excluded.diagnostics_ref, + updated_at = excluded.updated_at"#, + params![ + record.workspace_id, + record.worker_id, + record.runtime_id, + record.runtime_worker_id, + record.display_name, + record.profile, + record.lifecycle_state, + record.retention_state, + record.transcript_ref, + record.session_ref, + record.summary_ref, + record.diagnostics_ref, + record.created_at, + record.updated_at, + ], + )?; + Ok(()) + }) + } + + fn get_worker_registry( + &self, + workspace_id: &str, + worker_id: &str, + ) -> Result> { + self.with_conn(|conn| { + conn.query_row( + worker_registry_select_sql("WHERE workspace_id = ?1 AND worker_id = ?2").as_str(), + params![workspace_id, worker_id], + read_worker_registry_record, + ) + .optional() + .map_err(Error::from) + }) + } + + fn get_worker_registry_by_runtime( + &self, + workspace_id: &str, + runtime_id: &str, + runtime_worker_id: &str, + ) -> Result> { + self.with_conn(|conn| { + conn.query_row( + worker_registry_select_sql( + "WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3", + ) + .as_str(), + params![workspace_id, runtime_id, runtime_worker_id], + read_worker_registry_record, + ) + .optional() + .map_err(Error::from) + }) + } + + fn list_worker_registry( + &self, + workspace_id: &str, + limit: usize, + ) -> Result> { + self.with_conn(|conn| { + let sql = worker_registry_select_sql( + "WHERE workspace_id = ?1 ORDER BY updated_at DESC LIMIT ?2", + ); + let mut stmt = conn.prepare(sql.as_str())?; + let rows = stmt.query_map( + params![workspace_id, limit as i64], + read_worker_registry_record, + )?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } + + fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()> { + self.with_conn(|conn| { + conn.execute( + r#"INSERT INTO workdir_registry ( + workspace_id, workdir_id, runtime_id, repository_id, selector, resolved_commit, + materialization_status, cleanliness, management_kind, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) + ON CONFLICT(workspace_id, workdir_id) DO UPDATE SET + runtime_id = excluded.runtime_id, + repository_id = excluded.repository_id, + selector = excluded.selector, + resolved_commit = excluded.resolved_commit, + materialization_status = excluded.materialization_status, + cleanliness = excluded.cleanliness, + management_kind = excluded.management_kind, + updated_at = excluded.updated_at"#, + params![ + record.workspace_id, + record.workdir_id, + record.runtime_id, + record.repository_id, + record.selector, + record.resolved_commit, + record.materialization_status, + record.cleanliness, + record.management_kind, + record.created_at, + record.updated_at, + ], + )?; + Ok(()) + }) + } + + fn get_workdir_registry( + &self, + workspace_id: &str, + workdir_id: &str, + ) -> Result> { + self.with_conn(|conn| { + conn.query_row( + workdir_registry_select_sql("WHERE workspace_id = ?1 AND workdir_id = ?2").as_str(), + params![workspace_id, workdir_id], + read_workdir_registry_record, + ) + .optional() + .map_err(Error::from) + }) + } + + fn list_workdir_registry( + &self, + workspace_id: &str, + limit: usize, + ) -> Result> { + self.with_conn(|conn| { + let sql = workdir_registry_select_sql( + "WHERE workspace_id = ?1 ORDER BY updated_at DESC LIMIT ?2", + ); + let mut stmt = conn.prepare(sql.as_str())?; + let rows = stmt.query_map( + params![workspace_id, limit as i64], + read_workdir_registry_record, + )?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } + + fn list_managed_workdir_registry( + &self, + workspace_id: &str, + limit: usize, + ) -> Result> { + self.with_conn(|conn| { + let sql = workdir_registry_select_sql( + "WHERE workspace_id = ?1 AND management_kind = 'backend_managed' ORDER BY updated_at DESC LIMIT ?2", + ); + let mut stmt = conn.prepare(sql.as_str())?; + let rows = stmt.query_map(params![workspace_id, limit as i64], read_workdir_registry_record)?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } + + fn upsert_worker_workdir_link(&self, record: &WorkerWorkdirLinkRecord) -> Result<()> { + self.with_conn(|conn| { + conn.execute( + r#"INSERT INTO worker_workdir_links ( + workspace_id, worker_id, workdir_id, role, linked_at, unlinked_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(workspace_id, worker_id, workdir_id, role) DO UPDATE SET + linked_at = excluded.linked_at, + unlinked_at = excluded.unlinked_at"#, + params![ + record.workspace_id, + record.worker_id, + record.workdir_id, + record.role, + record.linked_at, + record.unlinked_at, + ], + )?; + Ok(()) + }) + } + + fn list_worker_workdir_links( + &self, + workspace_id: &str, + worker_id: &str, + ) -> Result> { + self.with_conn(|conn| { + let mut stmt = conn.prepare( + r#"SELECT workspace_id, worker_id, workdir_id, role, linked_at, unlinked_at + FROM worker_workdir_links + WHERE workspace_id = ?1 AND worker_id = ?2 AND unlinked_at IS NULL + ORDER BY linked_at DESC"#, + )?; + let rows = stmt.query_map(params![workspace_id, worker_id], |row| { + Ok(WorkerWorkdirLinkRecord { + workspace_id: row.get(0)?, + worker_id: row.get(1)?, + workdir_id: row.get(2)?, + role: row.get(3)?, + linked_at: row.get(4)?, + unlinked_at: row.get(5)?, + }) + })?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } +} + +fn worker_registry_select_sql(where_clause: &str) -> String { + format!( + "SELECT workspace_id, worker_id, runtime_id, runtime_worker_id, display_name, profile, \ + lifecycle_state, retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, \ + created_at, updated_at FROM worker_registry {where_clause}" + ) +} + +fn read_worker_registry_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(WorkerRegistryRecord { + workspace_id: row.get(0)?, + worker_id: row.get(1)?, + runtime_id: row.get(2)?, + runtime_worker_id: row.get(3)?, + display_name: row.get(4)?, + profile: row.get(5)?, + lifecycle_state: row.get(6)?, + retention_state: row.get(7)?, + transcript_ref: row.get(8)?, + session_ref: row.get(9)?, + summary_ref: row.get(10)?, + diagnostics_ref: row.get(11)?, + created_at: row.get(12)?, + updated_at: row.get(13)?, + }) +} + +fn workdir_registry_select_sql(where_clause: &str) -> String { + format!( + "SELECT workspace_id, workdir_id, runtime_id, repository_id, selector, resolved_commit, \ + materialization_status, cleanliness, management_kind, created_at, updated_at \ + FROM workdir_registry {where_clause}" + ) +} + +fn read_workdir_registry_record( + row: &rusqlite::Row<'_>, +) -> rusqlite::Result { + Ok(WorkdirRegistryRecord { + workspace_id: row.get(0)?, + workdir_id: row.get(1)?, + runtime_id: row.get(2)?, + repository_id: row.get(3)?, + selector: row.get(4)?, + resolved_commit: row.get(5)?, + materialization_status: row.get(6)?, + cleanliness: row.get(7)?, + management_kind: row.get(8)?, + created_at: row.get(9)?, + updated_at: row.get(10)?, + }) +} + +fn create_worker_workdir_registry_tables(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" +CREATE TABLE IF NOT EXISTS worker_registry ( + workspace_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + runtime_worker_id TEXT NOT NULL, + display_name TEXT NOT NULL, + profile TEXT, + lifecycle_state TEXT NOT NULL, + retention_state TEXT NOT NULL CHECK (retention_state IN ('normal', 'pinned')), + transcript_ref TEXT, + session_ref TEXT, + summary_ref TEXT, + diagnostics_ref TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, worker_id), + UNIQUE (workspace_id, runtime_id, runtime_worker_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS workdir_registry ( + workspace_id TEXT NOT NULL, + workdir_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + repository_id TEXT NOT NULL, + selector TEXT, + resolved_commit TEXT, + materialization_status TEXT NOT NULL CHECK (materialization_status IN ('pending', 'present', 'missing', 'removed', 'failed')), + cleanliness TEXT NOT NULL CHECK (cleanliness IN ('clean', 'dirty', 'unknown')), + management_kind TEXT NOT NULL CHECK (management_kind IN ('backend_managed', 'runtime_unmanaged')), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, workdir_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS worker_workdir_links ( + workspace_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + workdir_id TEXT NOT NULL, + role TEXT NOT NULL, + linked_at TEXT NOT NULL, + unlinked_at TEXT, + PRIMARY KEY (workspace_id, worker_id, workdir_id, role), + FOREIGN KEY (workspace_id, worker_id) REFERENCES worker_registry(workspace_id, worker_id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, workdir_id) REFERENCES workdir_registry(workspace_id, workdir_id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_worker_registry_workspace_updated + ON worker_registry(workspace_id, updated_at DESC); +CREATE INDEX IF NOT EXISTS idx_workdir_registry_workspace_updated + ON workdir_registry(workspace_id, updated_at DESC); +CREATE INDEX IF NOT EXISTS idx_worker_workdir_links_worker + ON worker_workdir_links(workspace_id, worker_id, linked_at DESC); +"#, + )?; + Ok(()) } fn configure_sqlite(conn: &Connection) -> Result<()> { @@ -488,7 +930,7 @@ mod tests { let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 2); + assert_eq!(store.schema_version().await.unwrap(), 3); let record = WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -500,7 +942,7 @@ mod tests { store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 2); + assert_eq!(reopened.schema_version().await.unwrap(), 3); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -527,6 +969,9 @@ mod tests { "ticket_worker_links", "artifacts", "audit_events", + "worker_registry", + "workdir_registry", + "worker_workdir_links", ] { assert!( tables.contains(expected), @@ -688,7 +1133,7 @@ mod tests { .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 2); + assert_eq!(store.schema_version().await.unwrap(), 3); store .with_conn(|conn| { @@ -773,6 +1218,115 @@ mod tests { ); } + #[tokio::test] + async fn worker_workdir_registry_round_trips_and_preserves_pinned_retention() { + let temp = tempfile::tempdir().unwrap(); + let db = temp.path().join("workspace.db"); + let store = SqliteWorkspaceStore::open(&db).unwrap(); + let workspace = WorkspaceRecord { + workspace_id: "local-dev".to_string(), + display_name: "Local Dev".to_string(), + state: "active".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }; + store.upsert_workspace(&workspace).await.unwrap(); + + let worker = WorkerRegistryRecord { + workspace_id: "local-dev".to_string(), + worker_id: "embedded/browser-1".to_string(), + runtime_id: "embedded".to_string(), + runtime_worker_id: "browser-1".to_string(), + display_name: "Browser 1".to_string(), + profile: Some("builtin:companion".to_string()), + lifecycle_state: "idle".to_string(), + retention_state: "pinned".to_string(), + transcript_ref: Some("runtime://embedded/workers/browser-1/transcript".to_string()), + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: "2".to_string(), + updated_at: "2".to_string(), + }; + store.upsert_worker_registry(&worker).unwrap(); + let mut runtime_sync_worker = worker.clone(); + runtime_sync_worker.lifecycle_state = "running".to_string(); + runtime_sync_worker.retention_state = "normal".to_string(); + runtime_sync_worker.updated_at = "5".to_string(); + store.upsert_worker_registry(&runtime_sync_worker).unwrap(); + let mut expected_worker = worker.clone(); + expected_worker.lifecycle_state = "running".to_string(); + expected_worker.updated_at = "5".to_string(); + + let workdir = WorkdirRegistryRecord { + workspace_id: "local-dev".to_string(), + workdir_id: "backend-2-repo".to_string(), + runtime_id: "embedded".to_string(), + repository_id: "repo".to_string(), + selector: Some("develop".to_string()), + resolved_commit: Some("abcdef".to_string()), + materialization_status: "removed".to_string(), + cleanliness: "clean".to_string(), + management_kind: "backend_managed".to_string(), + created_at: "2".to_string(), + updated_at: "3".to_string(), + }; + store.upsert_workdir_registry(&workdir).unwrap(); + let unmanaged_workdir = WorkdirRegistryRecord { + workspace_id: "local-dev".to_string(), + workdir_id: "runtime-direct".to_string(), + runtime_id: "embedded".to_string(), + repository_id: "repo".to_string(), + selector: Some("feature".to_string()), + resolved_commit: Some("123456".to_string()), + materialization_status: "present".to_string(), + cleanliness: "unknown".to_string(), + management_kind: "runtime_unmanaged".to_string(), + created_at: "3".to_string(), + updated_at: "4".to_string(), + }; + store.upsert_workdir_registry(&unmanaged_workdir).unwrap(); + + let link = WorkerWorkdirLinkRecord { + workspace_id: "local-dev".to_string(), + worker_id: worker.worker_id.clone(), + workdir_id: workdir.workdir_id.clone(), + role: "primary_cwd".to_string(), + linked_at: "4".to_string(), + unlinked_at: None, + }; + store.upsert_worker_workdir_link(&link).unwrap(); + + assert_eq!( + store + .get_worker_registry_by_runtime("local-dev", "embedded", "browser-1") + .unwrap(), + Some(expected_worker.clone()) + ); + assert_eq!( + store + .get_workdir_registry("local-dev", "backend-2-repo") + .unwrap(), + Some(workdir.clone()) + ); + assert_eq!( + store.list_workdir_registry("local-dev", 10).unwrap(), + vec![unmanaged_workdir.clone(), workdir.clone()] + ); + assert_eq!( + store + .list_managed_workdir_registry("local-dev", 10) + .unwrap(), + vec![workdir] + ); + assert_eq!( + store + .list_worker_workdir_links("local-dev", "embedded/browser-1") + .unwrap(), + vec![link] + ); + } + fn table_names(conn: &Connection) -> BTreeSet { let mut stmt = conn .prepare( diff --git a/web/workspace/src/lib/workspace-sidebar/types.ts b/web/workspace/src/lib/workspace-sidebar/types.ts index e8379c87..9e926f52 100644 --- a/web/workspace/src/lib/workspace-sidebar/types.ts +++ b/web/workspace/src/lib/workspace-sidebar/types.ts @@ -124,6 +124,7 @@ export type WorkingDirectorySummary = { resolved_commit: string; resolved_tree?: string | null; status: string; + management_kind?: "backend_managed" | "runtime_unmanaged" | string | null; cleanup_policy: string; cleanup_target: { kind: string;