From bbfbf92264fcfa552f67f961587d56fa37ac1524 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 5 Oct 2026 18:10:01 +0900 Subject: [PATCH] feat: bind Workdir attachments to durable connection lifetimes Add caller-scoped bounded connection inventory and conditional detach on the existing ledger/session lifecycle. Preserve lifetime identity during attenuation and compensation, migrate persisted rows, and bind self operations to verified Runtime/Worker proof. Validation: workspace-root cargo check; Backend 561 unit + 11 CLI tests; server-api 62 default + 69 typescript tests. Canonical OpenAPI and generated TypeScript source digests regenerated; browser declarations unchanged. --- crates/server-api/src/lib.rs | 130 ++++ crates/workspace-server/src/server.rs | 627 +++++++++++++++++- crates/workspace-server/src/store.rs | 443 ++++++++++++- openapi/server-api.json | 2 +- .../src/lib/generated/companion-api.ts | 2 +- .../src/lib/generated/repository-api.ts | 2 +- .../src/lib/generated/runtime-api.ts | 2 +- web/workspace/src/lib/generated/ticket-api.ts | 2 +- .../src/lib/generated/workdir-api.ts | 2 +- .../src/lib/generated/worker-launch-api.ts | 2 +- 10 files changed, 1147 insertions(+), 67 deletions(-) diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index a100ca901..b52bddace 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -192,6 +192,10 @@ impl_openapi_schema!( WorkingDirectoryCreateResponse, CurrentWorkerWorkdirAttachRequest, CurrentWorkerWorkdirAttachmentResponse, + CurrentWorkerWorkdirAttachmentItem, + CurrentWorkerWorkdirAttachmentListQuery, + CurrentWorkerWorkdirAttachmentListResponse, + CurrentWorkerWorkdirDetachQuery, CurrentWorkerWorkdirOperationRequest, CurrentWorkerWorkdirOperationResponse, ExternalWorkdirGrantCreateRequest, @@ -2549,6 +2553,19 @@ pub trait ServerApi { #[path] working_directory_id: String, #[body] request: WorkingDirectoryRemovalRequest, ) -> Result; + #[get( + "/api/w/{workspace_id}/workers/self/workdir-attachments", + status = 200, + error_status = 400, + additional_error_statuses = [401, 403, 404, 500], + openapi = false, + )] + async fn current_worker_workdir_attachment_list( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[query] query: CurrentWorkerWorkdirAttachmentListQuery, + ) -> Result; #[post( "/api/w/{workspace_id}/workers/self/workdir-attachments", status = 200, @@ -2574,6 +2591,7 @@ pub trait ServerApi { #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[path] alias: String, + #[query] query: CurrentWorkerWorkdirDetachQuery, ) -> Result; #[post( "/api/w/{workspace_id}/workers/self/workdir-session/operations", @@ -6754,6 +6772,45 @@ pub struct CurrentWorkerWorkdirAttachRequest { pub working_directory_id: String, } +/// Bounded, alias-ordered paging over the current Worker's active attachments. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct CurrentWorkerWorkdirAttachmentListQuery { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub limit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub offset: Option, + /// Exact caller-scoped connection lookup, applied before paging; 1..=128 bytes. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub connection_id: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct CurrentWorkerWorkdirAttachmentItem { + pub alias: String, + pub working_directory_id: String, + pub capabilities: workdir::WorkdirSessionCapabilities, + /// Opaque durable identity of this attachment lifetime, not a Workdir ID. + pub connection_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct CurrentWorkerWorkdirAttachmentListResponse { + pub workspace_id: String, + pub items: Vec, + pub next_offset: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct CurrentWorkerWorkdirDetachQuery { + /// When omitted, retain the normal Tool's unconditional detach contract. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub expected_connection_id: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] #[serde(deny_unknown_fields)] pub struct CurrentWorkerWorkdirAttachmentResponse { @@ -11930,6 +11987,78 @@ mod tests { ); } + #[test] + fn current_worker_attachment_contract_keeps_normal_tool_response_closed() { + let lookup = CurrentWorkerWorkdirAttachmentListQuery { + limit: Some(1), + offset: None, + connection_id: Some("opaque-connection".to_string()), + }; + let value = serde_json::json!({"limit": 1, "connection_id": "opaque-connection"}); + assert_eq!(serde_json::to_value(&lookup).unwrap(), value); + assert_eq!( + serde_json::from_value::(value).unwrap(), + lookup + ); + assert_eq!( + serde_json::to_value(CurrentWorkerWorkdirAttachmentListQuery::default()).unwrap(), + serde_json::json!({}) + ); + assert!( + serde_json::from_value::( + serde_json::json!({"connection_id": 123}) + ) + .is_err() + ); + assert!( + serde_json::from_value::( + serde_json::json!({"expected_connection_id": "wrong-field"}) + ) + .is_err() + ); + let query: CurrentWorkerWorkdirDetachQuery = + serde_json::from_value(serde_json::json!({})).unwrap(); + assert_eq!(query.expected_connection_id, None); + assert_eq!(serde_json::to_value(query).unwrap(), serde_json::json!({})); + assert!( + serde_json::from_value::( + serde_json::json!({"connection_id":"wrong-field"}) + ) + .is_err() + ); + let response = CurrentWorkerWorkdirAttachmentResponse { + workspace_id: "workspace".to_string(), + alias: "checkout".to_string(), + working_directory_id: "workdir".to_string(), + capabilities: workdir::WorkdirSessionCapabilities::EMPTY, + attached: false, + }; + let value = serde_json::to_value(&response).unwrap(); + assert_eq!(value.as_object().unwrap().len(), 5); + assert!(value.get("connection_id").is_none()); + assert_eq!( + serde_json::from_value::(value).unwrap(), + response + ); + let list = CurrentWorkerWorkdirAttachmentListResponse { + workspace_id: "workspace".to_string(), + next_offset: Some(1), + items: vec![CurrentWorkerWorkdirAttachmentItem { + alias: "checkout".to_string(), + working_directory_id: "workdir".to_string(), + capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY, + connection_id: "opaque".to_string(), + }], + }; + assert_eq!( + serde_json::from_value::( + serde_json::to_value(&list).unwrap() + ) + .unwrap(), + list + ); + } + #[test] fn workspace_and_repository_response_shapes_round_trip() { let workspace = serde_json::json!({ @@ -12895,6 +13024,7 @@ mod openapi_artifact_tests { // identity, or a one-use Reviewer capability. They remain generated Rust client/Axum // operations, but must not appear as unauthenticated operations in the public OpenAPI. const SIGNED_INTERNAL: &[&str] = &[ + "current_worker_workdir_attachment_list", "current_worker_workdir_attach", "current_worker_workdir_detach", "current_worker_workdir_operation", diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 39e0163ba..269b0f248 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -4739,6 +4739,7 @@ impl WorkspaceApi { for (alias, workdir_id, reservation_id, capabilities) in &reserved_attachments { let attachment = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: self.config.workspace_id.clone(), worker: worker_ref.clone(), workdir_id: workdir_id.clone(), @@ -5555,6 +5556,49 @@ fn contract_request_headers( Ok(headers) } +/// Self APIs derive identity from authenticated middleware, never caller-supplied hints. +fn current_worker_contract_headers( + context: &server_api::ServerRequestContext, +) -> std::result::Result { + let source = context.runtime_source.as_ref().ok_or_else(|| { + ApiError::from(Error::WorkspacePermissionDenied( + "self Workdir requests require a verified Runtime Worker source".to_string(), + )) + .into_repository_api_error() + })?; + let worker_id = source.worker_id.as_deref().ok_or_else(|| { + ApiError::from(Error::WorkspacePermissionDenied( + "self Workdir requests require a Runtime-bound Worker identity".to_string(), + )) + .into_repository_api_error() + })?; + let mut headers = contract_request_headers(context)?; + for (name, expected) in [ + ("x-yoi-runtime-id", source.runtime_id.as_str()), + ("x-yoi-worker-id", worker_id), + ] { + if headers + .get(name) + .is_some_and(|value| value.as_bytes() != expected.as_bytes()) + { + return Err(ApiError::from(Error::WorkspacePermissionDenied( + "self Workdir request identity does not match its verified source".to_string(), + )) + .into_repository_api_error()); + } + headers.insert( + name, + HeaderValue::from_str(expected).map_err(|_| { + ApiError::from(Error::WorkerSourceIdentity( + "invalid verified Worker identity".to_string(), + )) + .into_repository_api_error() + })?, + ); + } + Ok(headers) +} + async fn attach_server_origin_context(mut request: Request, next: Next) -> Response { let actor = request.extensions().get::().cloned(); let origin = request_origin(request.headers()); @@ -5841,6 +5885,7 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou .merge(server_api::server_api_axum::runtime_workdir_cleanup( service.clone(), )) + .merge(server_api::server_api_axum::current_worker_workdir_attachment_list(service.clone())) .merge(server_api::server_api_axum::current_worker_workdir_attach( service.clone(), )) @@ -10372,6 +10417,31 @@ impl server_api::ServerApi for ServerApiContractService { .map_err(ApiError::into_repository_api_error) } + async fn current_worker_workdir_attachment_list( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + query: server_api::CurrentWorkerWorkdirAttachmentListQuery, + ) -> std::result::Result< + server_api::CurrentWorkerWorkdirAttachmentListResponse, + server_api::RepositoryApiError, + > { + let api = self.workspace_api()?; + validate_workspace_scope(api, &workspace_id) + .map_err(ApiError::into_repository_api_error)?; + let worker = current_worker_identity( + api, + &workspace_id, + ¤t_worker_contract_headers(&context)?, + ) + .map_err(ApiError::from) + .map_err(ApiError::into_repository_api_error)?; + let session_lock = current_worker_session_lock(api, &worker); + let _session_guard = session_lock.lock().await; + list_current_worker_workdir_attachments(api, &worker, query) + .map_err(ApiError::into_repository_api_error) + } + async fn current_worker_workdir_attach( &self, context: server_api::ServerRequestContext, @@ -10384,7 +10454,7 @@ impl server_api::ServerApi for ServerApiContractService { let Json(response) = scoped_attach_current_worker_workdir( State(self.workspace_api()?.clone()), AxumPath(ScopedWorkspacePath { workspace_id }), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, Json(AttachCurrentWorkerWorkdirRequest { alias: request.alias, working_directory_id: request.working_directory_id, @@ -10400,6 +10470,7 @@ impl server_api::ServerApi for ServerApiContractService { context: server_api::ServerRequestContext, workspace_id: String, alias: String, + query: server_api::CurrentWorkerWorkdirDetachQuery, ) -> std::result::Result< server_api::CurrentWorkerWorkdirAttachmentResponse, server_api::RepositoryApiError, @@ -10407,7 +10478,8 @@ impl server_api::ServerApi for ServerApiContractService { let Json(response) = scoped_detach_current_worker_workdir( State(self.workspace_api()?.clone()), AxumPath((workspace_id, alias)), - contract_request_headers(&context)?, + current_worker_contract_headers(&context)?, + query, ) .await .map_err(ApiError::into_repository_api_error)?; @@ -10430,7 +10502,7 @@ impl server_api::ServerApi for ServerApiContractService { .clone(), ), AxumPath(ScopedWorkspacePath { workspace_id }), - contract_request_headers(&context) + current_worker_contract_headers(&context) .map_err(server_api::WorkdirOperationApiError::api)?, Json(WorkspaceWorkdirSessionOperationRequest { target_workdir: request.target_workdir, @@ -16354,6 +16426,7 @@ async fn scoped_attach_current_worker_workdir( } close_current_worker_attachment_session_locked(&api, &worker).await?; let link = api.store.attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: api.config.workspace_id.clone(), worker: worker.clone(), workdir_id: workdir_id.to_string(), @@ -16413,56 +16486,121 @@ async fn scoped_attach_current_worker_workdir( })) } +fn list_current_worker_workdir_attachments( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + query: server_api::CurrentWorkerWorkdirAttachmentListQuery, +) -> ApiResult { + let limit = query.limit.unwrap_or(50); + let offset = query.offset.unwrap_or(0); + if limit == 0 || limit > 100 { + return Err( + Error::InvalidInput("attachment page limit must be 1..=100".to_string()).into(), + ); + } + let mut links = api.store.list_worker_workdir_links_page( + &api.config.workspace_id, + worker, + limit + 1, + offset, + query.connection_id.as_deref(), + )?; + let next_offset = + if links.len() > limit as usize { + Some(offset.checked_add(limit).ok_or_else(|| { + Error::InvalidInput("attachment page offset overflow".to_string()) + })?) + } else { + None + }; + links.truncate(limit as usize); + Ok(server_api::CurrentWorkerWorkdirAttachmentListResponse { + workspace_id: api.config.workspace_id.clone(), + items: links + .into_iter() + .map(|link| server_api::CurrentWorkerWorkdirAttachmentItem { + alias: link.alias, + working_directory_id: link.workdir_id, + capabilities: link.capabilities, + connection_id: link.connection_id, + }) + .collect(), + next_offset, + }) +} + async fn scoped_detach_current_worker_workdir( State(api): State, AxumPath((workspace_id, alias)): AxumPath<(String, String)>, headers: HeaderMap, + query: server_api::CurrentWorkerWorkdirDetachQuery, ) -> ApiResult> { validate_workspace_scope(&api, &workspace_id)?; let alias = workdir::WorkdirAttachmentAlias::new(alias) .map_err(|error| Error::InvalidInput(error.to_string()))?; let worker = current_worker_identity(&api, &workspace_id, &headers)?; - let session_lock = current_worker_session_lock(&api, &worker); + detach_current_worker_workdir( + &api, + &worker, + alias.as_str(), + query.expected_connection_id.as_deref(), + ) + .await + .map(Json) +} + +async fn detach_current_worker_workdir( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + alias: &str, + expected_connection_id: Option<&str>, +) -> ApiResult { + let session_lock = current_worker_session_lock(api, worker); let _session_guard = session_lock.lock().await; let link = api .store - .list_worker_workdir_links(&api.config.workspace_id, &worker)? + .list_worker_workdir_links(&api.config.workspace_id, worker)? .into_iter() - .find(|link| link.unlinked_at.is_none() && link.alias == alias.as_str()) + .find(|link| link.unlinked_at.is_none() && link.alias == alias) .ok_or_else(|| { Error::WorkdirAttachmentConflict(format!( "Worker {}:{} has no Workdir attachment `{alias}`", worker.runtime_id, worker.worker_id )) })?; - close_current_worker_attachment_session_locked(&api, &worker).await?; - close_current_worker_command_sessions_for_alias_locked(&api, &worker, alias.as_str()).await?; - api.store.detach_worker_workdir( + if expected_connection_id.is_some_and(|expected| expected != link.connection_id) { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir attachment `{alias}` connection changed during detach" + )) + .into()); + } + close_current_worker_attachment_session_locked(api, worker).await?; + close_current_worker_command_sessions_for_alias_locked(api, worker, alias).await?; + // Even unconditional HTTP requests condition the mutation on the lifetime just read + // under the session lock, so a concurrent ledger replacement cannot be removed. + api.store.detach_worker_workdir_connection( &api.config.workspace_id, - &worker, - Some(&link.workdir_id), + worker, + alias, + &link.connection_id, &Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), )?; - if let Err(error) = sync_runtime_worker_workdir_attachments(&api, &worker) { - let mut restored = link.clone(); - restored.unlinked_at = None; - api.store.attach_worker_workdir(&restored)?; + if let Err(error) = sync_runtime_worker_workdir_attachments(api, worker) { + api.store.restore_worker_workdir_connection(&link)?; return Err(error); } let remaining = api .store - .list_worker_workdir_links(&api.config.workspace_id, &worker)? + .list_worker_workdir_links(&api.config.workspace_id, worker)? .into_iter() .filter(|link| link.unlinked_at.is_none()) .collect::>(); if remaining.len() == 1 && let Err(error) = - open_current_worker_workdir_session_locked(&api, &worker, &remaining[0]).await + open_current_worker_workdir_session_locked(api, worker, &remaining[0]).await { - let mut restored = link.clone(); - restored.unlinked_at = None; - api.store.attach_worker_workdir(&restored)?; - let _ = sync_runtime_worker_workdir_attachments(&api, &worker); + api.store.restore_worker_workdir_connection(&link)?; + let _ = sync_runtime_worker_workdir_attachments(api, worker); return Err(error.into()); } if let Err(error) = api @@ -16470,20 +16608,18 @@ async fn scoped_detach_current_worker_workdir( .refresh(&worker) .map_err(ApiError::from) { - let mut restored = link.clone(); - restored.unlinked_at = None; - api.store.attach_worker_workdir(&restored)?; - sync_runtime_worker_workdir_attachments(&api, &worker)?; - refresh_current_worker_session_locked(&api, &worker).await?; + api.store.restore_worker_workdir_connection(&link)?; + sync_runtime_worker_workdir_attachments(api, worker)?; + refresh_current_worker_session_locked(api, worker).await?; return Err(error); } - Ok(Json(server_api::CurrentWorkerWorkdirAttachmentResponse { + Ok(server_api::CurrentWorkerWorkdirAttachmentResponse { workspace_id: api.config.workspace_id.clone(), alias: alias.to_string(), working_directory_id: link.workdir_id, capabilities: workdir::WorkdirSessionCapabilities::EMPTY, attached: false, - })) + }) } fn validated_current_worker_attachment( @@ -31079,6 +31215,7 @@ fn link_worker_to_workdir( ) -> ApiResult<()> { let timestamp = now_registry_timestamp(); let record = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: api.config.workspace_id.clone(), worker: worker_record.worker.clone(), workdir_id: workdir_id.to_string(), @@ -33218,6 +33355,7 @@ mod tests { .unwrap(); schedule_external_workdir_expiry(&api, &grant).unwrap(); let link = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: worker.clone(), workdir_id: grant.workdir_id.clone(), @@ -34186,6 +34324,7 @@ mod tests { }) .unwrap(); let repository_link = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: worker.clone(), workdir_id: "analysis-repository".to_string(), @@ -34195,6 +34334,7 @@ mod tests { unlinked_at: None, }; let external_link = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: worker.clone(), workdir_id: grant.working_directory_id.clone(), @@ -35524,7 +35664,7 @@ mod tests { updated_at: "2026-01-01T00:00:00Z".to_owned(), revoked_at: None, }, - false, + true, ) .unwrap(); } @@ -35939,6 +36079,7 @@ mod tests { }; api.store.upsert_workdir_registry(&workdir).unwrap(); let link = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: worker.worker.clone(), workdir_id: workdir.workdir_id.clone(), @@ -36043,6 +36184,7 @@ mod tests { .create_external_workdir_grant(&grant, &workdir) .unwrap(); let link = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: worker.worker.clone(), workdir_id: workdir.workdir_id.clone(), @@ -36943,6 +37085,7 @@ mod tests { .unwrap(); api.store .attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: worker.clone(), workdir_id: workdir_id.to_string(), @@ -41307,6 +41450,7 @@ mod tests { .unwrap(); api.store .attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, "7"), workdir_id: "managed".to_string(), @@ -46566,6 +46710,7 @@ mod tests { .unwrap(); api.store .attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: worker.clone(), workdir_id: workdir_id.clone(), @@ -50104,6 +50249,7 @@ mod tests { let runtime_worker_id = runtime_worker_id.parse::().unwrap(); api.store .attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: api.config.workspace_id.clone(), worker: RuntimeWorkerRef::new("runtime-test", runtime_worker_id.to_string()), workdir_id: workdir_id.to_string(), @@ -50546,7 +50692,426 @@ mod tests { ) .await .unwrap(); - assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn current_worker_attachment_list_and_stale_detach_use_signed_identity() { + let mut fixture = manual_coder_assignment_fixture().await; + let identity = + worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id) + .unwrap(); + configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id); + let api = &fixture.api; + let base = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments"); + let request = |method: &str, path: &str| { + runtime_source_request( + &identity, + Some(&fixture.worker.worker_id), + method, + path, + Vec::new(), + ) + }; + let response = build_router(api.clone()) + .oneshot(request("GET", &format!("{base}?limit=1"))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let page: server_api::CurrentWorkerWorkdirAttachmentListResponse = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(page.workspace_id, TEST_WORKSPACE_ID); + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].alias, "checkout"); + assert_eq!(page.next_offset, Some(1)); + let old = api + .store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker) + .unwrap() + .into_iter() + .find(|link| link.alias == "checkout") + .unwrap(); + assert_eq!(page.items[0].connection_id, old.connection_id); + let response = build_router(api.clone()) + .oneshot(request("GET", &format!("{base}?limit=1&offset=1"))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let page: serde_json::Value = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(page["items"][0]["alias"], "docs"); + assert!(page["next_offset"].is_null()); + assert_eq!( + page["items"][0].as_object().unwrap().len(), + 4, + "no paths, credentials or materializer data" + ); + for query in ["limit=0", "limit=101", "offset=-1"] { + let response = build_router(api.clone()) + .oneshot(request("GET", &format!("{base}?{query}"))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + let mut forged = request("GET", &base); + forged + .headers_mut() + .insert("x-yoi-worker-id", "foreign-worker".parse().unwrap()); + assert_eq!( + build_router(api.clone()) + .oneshot(forged) + .await + .unwrap() + .status(), + StatusCode::FORBIDDEN + ); + let no_worker = runtime_source_request(&identity, None, "GET", &base, Vec::new()); + assert_eq!( + build_router(api.clone()) + .oneshot(no_worker) + .await + .unwrap() + .status(), + StatusCode::FORBIDDEN + ); + + api.store + .detach_worker_workdir_connection( + TEST_WORKSPACE_ID, + &fixture.worker, + "checkout", + &old.connection_id, + "first-detach", + ) + .unwrap(); + let replacement = api.store.attach_worker_workdir(&old).unwrap(); + assert_ne!(replacement.connection_id, old.connection_id); + // Leave one attachment so successful detach does not need a fixture provider session. + api.store + .detach_worker_workdir( + TEST_WORKSPACE_ID, + &fixture.worker, + Some(&fixture.docs_workdir_id), + "remove-docs", + ) + .unwrap(); + let root = tempfile::tempdir().unwrap(); + let session: WorkdirSessionHandle = Arc::new(workdir::LocalWorkdirSession::new( + manifest::Scope::writable(root.path()).unwrap(), + root.path().to_path_buf(), + )); + let provider_handle = session + .start_command(workdir::CommandRequest { + command: "sleep 30".to_string(), + timeout_secs: 60, + output_limit: 4096, + cwd: workdir::WorkdirPath::root(), + spill_dir: None, + tool_call_id: None, + }) + .await + .unwrap(); + let external = api.workdir_sessions.lock().unwrap().register_command( + fixture.worker.clone(), + "checkout".to_string(), + session.clone(), + provider_handle.clone(), + ); + let stale_path = format!( + "{base}/checkout?expected_connection_id={}", + old.connection_id + ); + let response = build_router(api.clone()) + .oneshot(request("DELETE", &stale_path)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::CONFLICT); + assert!( + api.workdir_sessions + .lock() + .unwrap() + .command(&fixture.worker, "checkout", &external) + .is_some() + ); + assert_eq!( + session + .command_status(provider_handle.clone()) + .await + .unwrap(), + workdir::CommandStatus::Running + ); + assert_eq!( + api.store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker) + .unwrap(), + vec![replacement.clone()] + ); + + // The same lock excludes command/SubWorker session activity before checking identity. + let lock = current_worker_session_lock(api, &fixture.worker); + let guard = lock.lock().await; + let mut detach = Box::pin(detach_current_worker_workdir( + api, + &fixture.worker, + "checkout", + Some(&old.connection_id), + )); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(20), &mut detach) + .await + .is_err() + ); + assert_eq!( + session + .command_status(provider_handle.clone()) + .await + .unwrap(), + workdir::CommandStatus::Running + ); + drop(guard); + assert!(matches!( + detach.await.unwrap_err().error, + Error::WorkdirAttachmentConflict(_) + )); + + let current_path = format!( + "{base}/checkout?expected_connection_id={}", + replacement.connection_id + ); + let response = build_router(api.clone()) + .oneshot(request("DELETE", ¤t_path)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let response: server_api::CurrentWorkerWorkdirAttachmentResponse = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert!(!response.attached); + assert!( + session.command_status(provider_handle).await.is_err(), + "matching detach cancels and closes command authority" + ); + assert!( + api.workdir_sessions + .lock() + .unwrap() + .command(&fixture.worker, "checkout", &external) + .is_none() + ); + assert!( + api.store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker) + .unwrap() + .is_empty() + ); + let new_lifetime = api.store.attach_worker_workdir(&old).unwrap(); + assert_ne!(new_lifetime.connection_id, replacement.connection_id); + let response = build_router(api.clone()) + .oneshot(request("DELETE", &format!("{base}/checkout"))) + .await + .unwrap(); + assert_eq!( + response.status(), + StatusCode::OK, + "normal Tool's query-free detach stays unconditional" + ); + } + + #[tokio::test] + async fn current_worker_attachment_connection_lookup_precedes_paging_and_is_scoped() { + let mut fixture = manual_coder_assignment_fixture().await; + let identity = + worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id) + .unwrap(); + configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id); + let api = &fixture.api; + let template = api + .store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker) + .unwrap() + .remove(0); + let workdir = api + .store + .get_workdir_registry(TEST_WORKSPACE_ID, &template.workdir_id) + .unwrap() + .unwrap(); + let mut target = None; + for index in 0..50 { + let workdir = WorkdirRegistryRecord { + workdir_id: format!("lookup-workdir-{index:03}"), + ..workdir.clone() + }; + api.store.upsert_workdir_registry(&workdir).unwrap(); + target = Some( + api.store + .attach_worker_workdir(&WorkerWorkdirLinkRecord { + workdir_id: workdir.workdir_id, + alias: format!("zz-lookup-{index:03}"), + ..template.clone() + }) + .unwrap(), + ); + } + let target = target.unwrap(); + let base = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments"); + let request = |path: &str| { + runtime_source_request( + &identity, + Some(&fixture.worker.worker_id), + "GET", + path, + Vec::new(), + ) + }; + let response = build_router(api.clone()) + .oneshot(request(&base)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let first: server_api::CurrentWorkerWorkdirAttachmentListResponse = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(first.items.len(), 50); + assert_eq!(first.next_offset, Some(50)); + assert!( + !first + .items + .iter() + .any(|item| item.connection_id == target.connection_id) + ); + let lookup_path = format!("{base}?limit=1&connection_id={}", target.connection_id); + let response = build_router(api.clone()) + .oneshot(request(&lookup_path)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let page: server_api::CurrentWorkerWorkdirAttachmentListResponse = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].alias, target.alias); + assert_eq!(page.items[0].working_directory_id, target.workdir_id); + assert_eq!(page.items[0].connection_id, target.connection_id); + assert_eq!(page.next_offset, None); + + let foreign_worker_id = seed_cleanup_worker(api, 902, "normal"); + seed_cleanup_workdir(api, "foreign-lookup-workdir", "present", "clean"); + seed_cleanup_link(api, &foreign_worker_id, "foreign-lookup-workdir"); + let foreign = api + .store + .list_worker_workdir_links( + TEST_WORKSPACE_ID, + &RuntimeWorkerRef::new("runtime-test", &foreign_worker_id), + ) + .unwrap() + .remove(0); + for query in [ + format!("limit=1&connection_id={}&offset=1", target.connection_id), + format!("limit=1&connection_id={}", foreign.connection_id), + "limit=1&connection_id=nonexistent".to_string(), + format!("limit=1&connection_id={}", "x".repeat(128)), + ] { + let response = build_router(api.clone()) + .oneshot(request(&format!("{base}?{query}"))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let page: server_api::CurrentWorkerWorkdirAttachmentListResponse = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert!(page.items.is_empty()); + assert_eq!(page.next_offset, None); + } + let mut forged = request(&lookup_path); + forged + .headers_mut() + .insert("x-yoi-worker-id", foreign_worker_id.parse().unwrap()); + assert_eq!( + build_router(api.clone()) + .oneshot(forged) + .await + .unwrap() + .status(), + StatusCode::FORBIDDEN + ); + for invalid in [ + String::new(), + "%20".to_string(), + "%0A".to_string(), + "x".repeat(129), + ] { + let response = build_router(api.clone()) + .oneshot(request(&format!("{base}?limit=1&connection_id={invalid}"))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + api.store + .detach_worker_workdir_connection( + TEST_WORKSPACE_ID, + &fixture.worker, + &target.alias, + &target.connection_id, + "lookup-detach", + ) + .unwrap(); + let replacement = api.store.attach_worker_workdir(&target).unwrap(); + assert_ne!(replacement.connection_id, target.connection_id); + let response = build_router(api.clone()) + .oneshot(request(&lookup_path)) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let stale: server_api::CurrentWorkerWorkdirAttachmentListResponse = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert!(stale.items.is_empty()); + assert_eq!(stale.next_offset, None); + let response = build_router(api.clone()) + .oneshot(request(&format!( + "{base}?limit=1&connection_id={}", + replacement.connection_id, + ))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let current: server_api::CurrentWorkerWorkdirAttachmentListResponse = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(current.items.len(), 1); + assert_eq!(current.items[0].connection_id, replacement.connection_id); + assert_eq!(current.next_offset, None); + } + + #[tokio::test] + async fn current_worker_attachment_failed_detach_preserves_connection_identity() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + let worker_id = seed_cleanup_worker(&api, 901, "normal"); + seed_cleanup_workdir(&api, "compensation-workdir", "present", "clean"); + seed_cleanup_link(&api, &worker_id, "compensation-workdir"); + let worker = RuntimeWorkerRef::new("runtime-test", worker_id); + let link = api + .store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker) + .unwrap() + .remove(0); + // No live Runtime provider: runtime synchronization rejects the detach after mutation. + assert!( + detach_current_worker_workdir(&api, &worker, "attachment", Some(&link.connection_id)) + .await + .is_err() + ); + assert_eq!( + api.store + .list_worker_workdir_links(TEST_WORKSPACE_ID, &worker) + .unwrap(), + vec![link] + ); } #[tokio::test] @@ -54773,6 +55338,7 @@ mod tests { .unwrap(); api.store .attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: RuntimeWorkerRef::new("embedded-worker-runtime", &worker_id), workdir_id: workdir_id.to_string(), @@ -55168,6 +55734,7 @@ mod tests { .unwrap(); api.store .attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: TEST_WORKSPACE_ID.to_string(), worker: RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, &worker_id), workdir_id: workdir_id.to_string(), diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index ecfebdec8..522827f52 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -28,7 +28,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 77; +const LATEST_SCHEMA_VERSION: i64 = 78; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -77,6 +77,7 @@ const ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME: &str = "preserve explicit observe grants for committed Worker Session archives"; const BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME: &str = "durable terminal Backend Job Worker cleanup"; +const WORKDIR_CONNECTION_ID_MIGRATION_NAME: &str = "durable Workdir attachment connection identity"; const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7; const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline"; const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#" @@ -259,6 +260,11 @@ const MIGRATIONS: &[Migration] = &[ name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME, apply: migrate_backend_job_worker_cleanup_v76_to_v77, }, + Migration { + version: 78, + name: WORKDIR_CONNECTION_ID_MIGRATION_NAME, + apply: migrate_workdir_connection_id_v77_to_v78, + }, ]; #[derive(Clone, Copy)] @@ -963,6 +969,8 @@ pub struct ExternalWorkdirGrantRecord { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerWorkdirLinkRecord { + /// Assigned by the ledger on attach; preserved only by explicit compensation. + pub connection_id: String, pub workspace_id: String, pub worker: RuntimeWorkerRef, pub workdir_id: String, @@ -1983,6 +1991,23 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { expected_workdir_id: Option<&str>, unlinked_at: &str, ) -> Result>; + fn detach_worker_workdir_connection( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + alias: &str, + expected_connection_id: &str, + unlinked_at: &str, + ) -> Result; + fn restore_worker_workdir_connection(&self, record: &WorkerWorkdirLinkRecord) -> Result<()>; + fn list_worker_workdir_links_page( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + limit: u32, + offset: u32, + connection_id: Option<&str>, + ) -> Result>; fn worker_workdir_link_history_exists( &self, workspace_id: &str, @@ -9349,6 +9374,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore { "reserved attachment finalization requires an active attachment".to_string(), )); } + let mut record = record.clone(); + record.connection_id = uuid::Uuid::now_v7().to_string(); self.with_conn(|conn| { let tx = rusqlite::Transaction::new_unchecked( conn, @@ -9370,11 +9397,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } tx.execute( r#"INSERT INTO worker_workdir_links ( - workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL) + workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8) ON CONFLICT(workspace_id, runtime_id, worker_id, workdir_id, alias) DO UPDATE SET capabilities = excluded.capabilities, linked_at = excluded.linked_at, + connection_id = excluded.connection_id, unlinked_at = NULL"#, params![ record.workspace_id, @@ -9384,6 +9412,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { record.alias, encode_workdir_link_capabilities(record.capabilities)?, record.linked_at, + record.connection_id, ], ) .map_err(|error| { @@ -9421,6 +9450,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore { "a new attachment cannot already be unlinked".to_string(), )); } + let mut record = record.clone(); + record.connection_id = uuid::Uuid::now_v7().to_string(); self.with_conn(|conn| { let tx = rusqlite::Transaction::new_unchecked( conn, @@ -9456,7 +9487,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } let active_for_alias = tx .query_row( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id FROM worker_workdir_links WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND alias = ?4 AND unlinked_at IS NULL"#, @@ -9499,7 +9530,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } let active_for_workdir = tx .query_row( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id FROM worker_workdir_links WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL"#, params![record.workspace_id, record.workdir_id], @@ -9514,11 +9545,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } let write = tx.execute( r#"INSERT INTO worker_workdir_links ( - workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL) + workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8) ON CONFLICT(workspace_id, runtime_id, worker_id, workdir_id, alias) DO UPDATE SET capabilities = excluded.capabilities, linked_at = excluded.linked_at, + connection_id = excluded.connection_id, unlinked_at = NULL"#, params![ record.workspace_id, @@ -9528,6 +9560,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { record.alias, encode_workdir_link_capabilities(record.capabilities)?, record.linked_at, + record.connection_id, ], ); if let Err(error) = write { @@ -9556,7 +9589,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { rusqlite::TransactionBehavior::Immediate, )?; let mut stmt = tx.prepare( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id FROM worker_workdir_links WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#, @@ -9571,7 +9604,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { let mut persisted_identity = persisted .iter() - .map(|link| (link.alias.as_str(), link.workdir_id.as_str())) + .map(|link| (link.alias.as_str(), link.workdir_id.as_str(), link.connection_id.as_str())) .collect::>(); persisted_identity.sort_unstable(); let mut requested_identity = active_links @@ -9592,7 +9625,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { link.alias )) })?; - Ok((link.alias.as_str(), link.workdir_id.as_str())) + Ok((link.alias.as_str(), link.workdir_id.as_str(), link.connection_id.as_str())) }) .collect::>>()?; requested_identity.sort_unstable(); @@ -9644,7 +9677,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { )?; let active = tx .query_row( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id FROM worker_workdir_links WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND (?4 IS NULL OR workdir_id = ?4) AND unlinked_at IS NULL"#, @@ -9683,6 +9716,144 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn detach_worker_workdir_connection( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + alias: &str, + expected_connection_id: &str, + unlinked_at: &str, + ) -> Result { + self.with_conn(|conn| { + let tx = rusqlite::Transaction::new_unchecked( + conn, + rusqlite::TransactionBehavior::Immediate, + )?; + let active = tx + .query_row( + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, + linked_at, unlinked_at, connection_id + FROM worker_workdir_links + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 + AND alias = ?4 AND connection_id = ?5 AND unlinked_at IS NULL"#, + params![ + workspace_id, + worker.runtime_id, + worker.worker_id, + alias, + expected_connection_id + ], + read_worker_workdir_link_record, + ) + .optional()? + .ok_or_else(|| { + Error::WorkdirAttachmentConflict(format!( + "Workdir attachment `{alias}` connection changed during detach" + )) + })?; + let changed = tx.execute( + r#"UPDATE worker_workdir_links SET unlinked_at = ?6 + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 + AND alias = ?4 AND connection_id = ?5 AND unlinked_at IS NULL"#, + params![ + workspace_id, + worker.runtime_id, + worker.worker_id, + alias, + expected_connection_id, + unlinked_at + ], + )?; + if changed != 1 { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir attachment `{alias}` connection changed during detach" + ))); + } + tx.commit()?; + Ok(WorkerWorkdirLinkRecord { + unlinked_at: Some(unlinked_at.to_string()), + ..active + }) + }) + } + + fn restore_worker_workdir_connection(&self, record: &WorkerWorkdirLinkRecord) -> Result<()> { + self.with_conn(|conn| { + // Compensation reactivates only the same lifetime, never an intervening attach. + let changed = conn.execute( + r#"UPDATE worker_workdir_links SET unlinked_at = NULL + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 + AND workdir_id = ?4 AND alias = ?5 AND connection_id = ?6 + AND unlinked_at IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM worker_workdir_attachment_reservations + WHERE workspace_id = ?1 AND workdir_id = ?4 + )"#, + params![record.workspace_id, record.worker.runtime_id, record.worker.worker_id, + record.workdir_id, record.alias, record.connection_id], + ).map_err(|error| { + if matches!(error, rusqlite::Error::SqliteFailure(ref code, _) if code.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE) { + Error::WorkdirAttachmentConflict("Workdir acquired another active attachment during compensation".to_string()) + } else { + error.into() + } + })?; + if changed != 1 { + return Err(Error::WorkdirAttachmentConflict( + "Workdir attachment connection changed during compensation".to_string(), + )); + } + Ok(()) + }) + } + + fn list_worker_workdir_links_page( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + limit: u32, + offset: u32, + connection_id: Option<&str>, + ) -> Result> { + if limit == 0 || limit > 101 { + return Err(Error::InvalidInput( + "attachment page limit must be 1..=101".to_string(), + )); + } + if connection_id.is_some_and(|id| { + id.trim().is_empty() || id.len() > 128 || id.chars().any(char::is_control) + }) { + return Err(Error::InvalidInput( + "attachment connection_id must contain 1..=128 bytes without control characters" + .to_string(), + )); + } + self.with_conn(|conn| { + let mut stmt = conn.prepare( + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, + linked_at, unlinked_at, connection_id + FROM worker_workdir_links + WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 + AND unlinked_at IS NULL + AND (?6 IS NULL OR connection_id = ?6) + ORDER BY alias LIMIT ?4 OFFSET ?5"#, + )?; + let rows = stmt.query_map( + params![ + workspace_id, + worker.runtime_id, + worker.worker_id, + limit, + offset, + connection_id + ], + read_worker_workdir_link_record, + )?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } + fn worker_workdir_link_history_exists( &self, workspace_id: &str, @@ -9708,7 +9879,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) -> Result> { self.with_conn(|conn| { let mut stmt = conn.prepare( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id FROM worker_workdir_links WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL ORDER BY linked_at DESC"#, @@ -9729,7 +9900,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) -> Result> { self.with_conn(|conn| { let mut stmt = conn.prepare( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id FROM worker_workdir_links WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL ORDER BY linked_at DESC"#, @@ -9750,7 +9921,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) -> Result> { self.with_conn(|conn| { conn.query_row( - r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id FROM worker_workdir_links WHERE workspace_id = ?1 AND workdir_id = ?2 ORDER BY linked_at DESC, rowid DESC @@ -10497,6 +10668,7 @@ fn read_worker_workdir_link_record( capabilities, linked_at: row.get(6)?, unlinked_at: row.get(7)?, + connection_id: row.get(8)?, }) } @@ -14486,6 +14658,27 @@ fn migrate_backend_job_worker_cleanup_v76_to_v77(conn: &Connection) -> Result<() Ok(()) } +fn migrate_workdir_connection_id_v77_to_v78(conn: &Connection) -> Result<()> { + use rusqlite::TransactionBehavior; + let current = current_schema_version(conn)?; + if current != 77 { + return Err(Error::Store(format!( + "expected schema version 77 before {WORKDIR_CONNECTION_ID_MIGRATION_NAME} migration, found {current}" + ))); + } + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + tx.execute_batch( + "ALTER TABLE worker_workdir_links ADD COLUMN connection_id TEXT NOT NULL DEFAULT ''; + UPDATE worker_workdir_links SET connection_id = lower(hex(randomblob(16)));", + )?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![78_i64, WORKDIR_CONNECTION_ID_MIGRATION_NAME], + )?; + tx.commit()?; + Ok(()) +} + fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?; conn.execute_batch(include_str!("latest_schema.sql"))?; @@ -14546,6 +14739,7 @@ fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { ); CREATE INDEX typed_ticket_targets_workspace_repository ON typed_ticket_targets(workspace_id, repository_key, ticket_id); + ALTER TABLE worker_workdir_links ADD COLUMN connection_id TEXT NOT NULL DEFAULT ''; ALTER TABLE worker_workdir_links ADD COLUMN capabilities TEXT NOT NULL CHECK (capabilities IN ( @@ -15610,6 +15804,46 @@ mod tests { Ok(()) } + #[test] + fn workdir_connection_migration_backfills_distinct_durable_lifetimes() { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch( + "CREATE TABLE __yoi_schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL); + INSERT INTO __yoi_schema_migrations VALUES (77, 'workspace schema baseline'); + CREATE TABLE worker_workdir_links (alias TEXT, linked_at TEXT, unlinked_at TEXT); + INSERT INTO worker_workdir_links VALUES ('checkout', 'same-time', NULL), ('docs', 'same-time', 'removed');", + ).unwrap(); + migrate_workdir_connection_id_v77_to_v78(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 78); + let rows = || { + let mut stmt = conn.prepare("SELECT alias, linked_at, unlinked_at, connection_id FROM worker_workdir_links ORDER BY alias").unwrap(); + stmt.query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, Option>(2)?, + row.get::<_, String>(3)?, + )) + }) + .unwrap() + .collect::, _>>() + .unwrap() + }; + let migrated = rows(); + assert_eq!(migrated[0].0, "checkout"); + assert_eq!(migrated[0].1, "same-time"); + assert_eq!(migrated[0].2, None); + assert_eq!(migrated[1].2.as_deref(), Some("removed")); + assert!(!migrated[0].3.is_empty()); + assert_ne!(migrated[0].3, migrated[1].3); + assert!(migrate_workdir_connection_id_v77_to_v78(&conn).is_err()); + assert_eq!( + rows(), + migrated, + "a retry cannot replace an already durable connection ID" + ); + } + #[test] fn worker_workdir_link_capability_codec_round_trips_only_canonical_sets() { for (capabilities, encoded) in WORKDIR_LINK_CAPABILITY_ENCODINGS { @@ -17052,6 +17286,10 @@ mod tests { version: 77, name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 78, + name: WORKDIR_CONNECTION_ID_MIGRATION_NAME.to_string(), + }, ] ); @@ -17139,6 +17377,7 @@ mod tests { 77, BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(), ), + (78, WORKDIR_CONNECTION_ID_MIGRATION_NAME.to_string()), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -17490,7 +17729,7 @@ mod tests { .collect::>(), vec![ 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, - 73, 74, 75, 76, 77 + 73, 74, 75, 76, 77, 78 ] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); @@ -17499,7 +17738,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 28); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 29); assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap()); assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap()); assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap()); @@ -20855,6 +21094,7 @@ INSERT INTO worker_registry ( store.upsert_workdir_registry(&unmanaged_workdir).unwrap(); let link = WorkerWorkdirLinkRecord { + connection_id: String::new(), workspace_id: "local-dev".to_string(), worker: worker.worker.clone(), workdir_id: workdir.workdir_id.clone(), @@ -20883,12 +21123,10 @@ INSERT INTO worker_registry ( store.attach_worker_workdir(&link), Err(Error::WorkdirAttachmentConflict(_)) )); - assert_eq!( - store - .finalize_reserved_worker_workdir_attachment(&link, "spawn-1") - .unwrap(), - link - ); + let link = store + .finalize_reserved_worker_workdir_attachment(&link, "spawn-1") + .unwrap(); + assert!(!link.connection_id.is_empty()); assert_eq!(store.attach_worker_workdir(&link).unwrap(), link); let downgraded_link = WorkerWorkdirLinkRecord { capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY, @@ -20958,10 +21196,7 @@ INSERT INTO worker_registry ( store.attach_worker_workdir(&invalid_capabilities), Err(Error::InvalidInput(_)) )); - assert_eq!( - store.attach_worker_workdir(&second_attachment).unwrap(), - second_attachment - ); + let second_attachment = store.attach_worker_workdir(&second_attachment).unwrap(); assert_eq!( store .list_worker_workdir_links("local-dev", &worker.worker) @@ -20991,6 +21226,157 @@ INSERT INTO worker_registry ( store.detach_worker_workdir("local-dev", &worker.worker, Some("wrong-workdir"), "6",), Err(Error::WorkdirAttachmentConflict(_)) )); + let old_connection = downgraded_link.connection_id.clone(); + let removed = store + .detach_worker_workdir_connection( + "local-dev", + &worker.worker, + "checkout", + &old_connection, + "5", + ) + .unwrap(); + store.restore_worker_workdir_connection(&removed).unwrap(); + assert_eq!( + store + .list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None) + .unwrap()[0], + downgraded_link + ); + store + .detach_worker_workdir_connection( + "local-dev", + &worker.worker, + "checkout", + &old_connection, + "5", + ) + .unwrap(); + // Reusing the exact record, including alias, Workdir ID and timestamp, starts a fresh lifetime. + let replacement = store.attach_worker_workdir(&downgraded_link).unwrap(); + assert_ne!(replacement.connection_id, old_connection); + assert!(matches!( + store.detach_worker_workdir_connection( + "local-dev", + &worker.worker, + "checkout", + &old_connection, + "6", + ), + Err(Error::WorkdirAttachmentConflict(_)) + )); + assert!(matches!( + store.restore_worker_workdir_connection(&removed), + Err(Error::WorkdirAttachmentConflict(_)) + )); + assert!(matches!( + store.replace_worker_workdir_link_capabilities( + "local-dev", + &worker.worker, + &[downgraded_link.clone(), second_attachment.clone()], + ), + Err(Error::WorkdirAttachmentConflict(_)) + )); + assert_eq!(store.attach_worker_workdir(&link).unwrap(), replacement); + assert_eq!( + store + .list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None) + .unwrap(), + vec![replacement.clone()] + ); + assert_eq!( + store + .list_worker_workdir_links_page("local-dev", &worker.worker, 1, 1, None) + .unwrap(), + vec![second_attachment.clone()] + ); + assert!( + store + .list_worker_workdir_links_page("local-dev", &worker.worker, 1, 2, None) + .unwrap() + .is_empty() + ); + assert!( + store + .list_worker_workdir_links_page("foreign", &worker.worker, 1, 0, None) + .unwrap() + .is_empty() + ); + assert!( + store + .list_worker_workdir_links_page( + "local-dev", + &RuntimeWorkerRef::new("foreign", worker.worker.worker_id.clone()), + 1, + 0, + None + ) + .unwrap() + .is_empty() + ); + assert_eq!( + store + .list_worker_workdir_links_page( + "local-dev", + &worker.worker, + 1, + 0, + Some(&second_attachment.connection_id) + ) + .unwrap(), + vec![second_attachment.clone()], + "connection filtering must precede paging" + ); + for (workspace_id, identity, connection_id) in [ + ( + "foreign", + worker.worker.clone(), + second_attachment.connection_id.as_str(), + ), + ( + "local-dev", + second_worker.worker.clone(), + second_attachment.connection_id.as_str(), + ), + ( + "local-dev", + RuntimeWorkerRef::new("foreign", worker.worker.worker_id.clone()), + second_attachment.connection_id.as_str(), + ), + ("local-dev", worker.worker.clone(), old_connection.as_str()), + ("local-dev", worker.worker.clone(), "nonexistent"), + ] { + assert!( + store + .list_worker_workdir_links_page( + workspace_id, + &identity, + 1, + 0, + Some(connection_id) + ) + .unwrap() + .is_empty() + ); + } + drop(store); + let store = SqliteWorkspaceStore::open(&db).unwrap(); + assert_eq!( + store + .list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None) + .unwrap(), + vec![replacement] + ); + assert!(matches!( + store.detach_worker_workdir_connection( + "local-dev", + &worker.worker, + "checkout", + &old_connection, + "6", + ), + Err(Error::WorkdirAttachmentConflict(_)) + )); let detached = store .detach_worker_workdir("local-dev", &worker.worker, Some(&workdir.workdir_id), "6") .unwrap() @@ -21001,10 +21387,7 @@ INSERT INTO worker_registry ( .worker_workdir_link_history_exists("local-dev", &worker.worker) .unwrap() ); - assert_eq!( - store.attach_worker_workdir(&workdir_conflict).unwrap(), - workdir_conflict - ); + let workdir_conflict = store.attach_worker_workdir(&workdir_conflict).unwrap(); drop(store); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); diff --git a/openapi/server-api.json b/openapi/server-api.json index 9fa559d03..30be582b1 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -14794,7 +14794,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870" + "x-yoi-source-digest": "sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778" }, "openapi": "3.1.0", "paths": { diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index ddba41954..36d091d7d 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870) +// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 052d8f330..ce89d157a 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870) +// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index 2a3b02082..163b263a5 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870) +// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index 24d0a4a79..061fdc848 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870) +// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/workdir-api.ts b/web/workspace/src/lib/generated/workdir-api.ts index 0fa79bece..89612a759 100644 --- a/web/workspace/src/lib/generated/workdir-api.ts +++ b/web/workspace/src/lib/generated/workdir-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-workdir-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870) +// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778) // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/workdir-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index 42881f6ce..7f56fba96 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870) +// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check