Delete auto-provisioned query keys with services

README.md

@@ -534,12 +534,12 @@ clickhousectl cloud service delete <service-id> --force
 
 `cloud service query` is the canonical way to run SQL against a cloud service — over HTTP, with no `clickhouse` binary and no service password required. It works with both credential modes:
 
-- **API key auth** (read + write SQL): the first time `cloud service query` runs against a service without a stored key, it provisions a Query API endpoint for that service and creates a dedicated API key bound to it. The key (`keyId`, `keySecret`, and `endpointId`) is stored in `.clickhouse/credentials.json` under `service_query_keys.<service-id>`, alongside any user-level API key. Subsequent queries use that key. It is scoped to a single service, so it can read and write (SELECT, INSERT, DDL) against that service but cannot reach any other service in the org. Pass `--no-auto-enable` to fail instead of provisioning.
+- **API key auth** (read + write SQL): the first time `cloud service query` runs against a service without a stored key, it provisions a Query API endpoint for that service and creates a dedicated API key bound to it. The query credentials, endpoint ID, exact management API key ID, and provisioning organization ID are stored in `.clickhouse/credentials.json` under `service_query_keys.<service-id>`, alongside any user-level API key. Subsequent queries use that key. It is scoped to a single service, so it can read and write (SELECT, INSERT, DDL) against that service but cannot reach any other service in the org. Pass `--no-auto-enable` to fail instead of provisioning.
 - **OAuth** (`cloud auth login`): the query runs as your own identity — the CLI sends your bearer token straight to the Query API, which grants **read-only** SQL access (SELECT and other read statements only; no INSERT, DDL, or other writes). No Query API key is provisioned or stored, and no query endpoint needs to be configured on the service. Use API key auth if you need to write. `--no-auto-enable` has no effect in this mode.
 
 Provisioning happens lazily (rather than at `service create` time) because the endpoint can only be bound once the service has finished provisioning, which can take several minutes — `service create` returns immediately instead of blocking on it.
 
-Per-service scoping is enforced at the query endpoint binding, which is created with role `sql_console_admin` (read + write inside the bound service only). The API key itself has no org-level roles, so the binding is the only thing that grants it any access. `cloud service delete` removes the stored key from `credentials.json`.
+Per-service scoping is enforced at the query endpoint binding, which is created with role `sql_console_admin` (read + write inside the bound service only). The API key itself has no org-level roles, so the binding is the only thing that grants it any access. After deleting a service, `cloud service delete` deletes an auto-provisioned key by its stored management and organization IDs, then removes the local record. Legacy records without that metadata remain readable, but service deletion will not guess at a cloud key by name; a partial record with a management ID is retained for manual recovery.
 
 Querying an **idled** service wakes it automatically in both auth modes — under OAuth the Query API first asks for a wake confirmation, which the CLI sends after printing a notice to stderr (the first query may take a minute while the service wakes). A **stopped** service is never woken: the query fails with a hint to run `cloud service start`.
 

crates/clickhousectl/src/cloud/client.rs

@@ -500,6 +500,18 @@ impl CloudClient {
         Self::unwrap_response(response)
     }
 
+    pub async fn get_service_if_exists(
+        &self,
+        org_id: &str,
+        service_id: &str,
+    ) -> Result<Option<clickhouse_cloud_api::models::Service>> {
+        match self.api().instance_get(org_id, service_id).await {
+            Ok(response) => Self::unwrap_response(response).map(Some),
+            Err(clickhouse_cloud_api::Error::Api { status: 404, .. }) => Ok(None),
+            Err(error) => Err(self.convert_error_for_organization(error, org_id)),
+        }
+    }
+
     pub async fn create_service(
         &self,
         org_id: &str,
@@ -513,16 +525,24 @@ impl CloudClient {
         Self::unwrap_response(response)
     }
 
-    pub async fn delete_service(&self, org_id: &str, service_id: &str) -> Result<DeleteResponse> {
-        let response = self
-            .api()
-            .instance_delete(org_id, service_id)
-            .await
-            .map_err(|e| self.convert_error_for_organization(e, org_id))?;
-        Ok(DeleteResponse {
-            status: response.status,
-            request_id: response.request_id,
-        })
+    pub async fn delete_service_if_exists(
+        &self,
+        org_id: &str,
+        service_id: &str,
+    ) -> Result<Option<DeleteResponse>> {
+        match self.api().instance_delete(org_id, service_id).await {
+            Ok(response) => Ok(Some(DeleteResponse {
+                status: response.status,
+                request_id: response.request_id,
+            })),
+            Err(clickhouse_cloud_api::Error::Api { status: 404, .. }) => {
+                // The service and its organization share the same 404 shape.
+                // Confirm the request scope before declaring the service absent.
+                self.get_organization(org_id).await?;
+                Ok(None)
+            }
+            Err(error) => Err(self.convert_error_for_organization(error, org_id)),
+        }
     }
 
     pub async fn change_service_state(
@@ -916,6 +936,21 @@ impl CloudClient {
         })
     }
 
+    pub async fn delete_api_key_if_exists(
+        &self,
+        org_id: &str,
+        key_id: &str,
+    ) -> Result<Option<DeleteResponse>> {
+        match self.api().openapi_key_delete(org_id, key_id).await {
+            Ok(response) => Ok(Some(DeleteResponse {
+                status: response.status,
+                request_id: response.request_id,
+            })),
+            Err(clickhouse_cloud_api::Error::Api { status: 404, .. }) => Ok(None),
+            Err(error) => Err(self.convert_error_for_organization(error, org_id)),
+        }
+    }
+
     // Phase 6 - Activity endpoints
     pub async fn list_activities(
         &self,

crates/clickhousectl/src/cloud/commands.rs

@@ -993,6 +993,65 @@ fn service_delete_error(error: CloudError, force: bool, service_id: &str) -> Clo
     }
 }
 
+/// Return a query API key only when its exact resource and organization IDs
+/// were saved during provisioning. The boolean indicates that partial cleanup
+/// metadata must remain on disk because discarding it would lose the key ID.
+fn service_query_key_cleanup(
+    org_id: &str,
+    service_id: &str,
+) -> Result<(Option<String>, bool), Box<dyn std::error::Error>> {
+    let Some(key) = credentials::try_get_service_query_key(service_id)? else {
+        return Ok((None, false));
+    };
+    let Some(api_key_id) = key.api_key_id else {
+        eprintln!(
+            "Warning: the stored query key for service {service_id} predates exact management \
+             API key IDs; service deletion will continue without unsafe cloud key cleanup."
+        );
+        return Ok((None, false));
+    };
+    let Some(key_org_id) = key.organization_id else {
+        eprintln!(
+            "Warning: the stored query key for service {service_id} has a management API key ID \
+             but no provisioning organization; cloud key cleanup was skipped and the local \
+             record was retained."
+        );
+        return Ok((None, true));
+    };
+    if key_org_id != org_id {
+        return Err(format!(
+            "the stored query key for service {service_id} belongs to organization {key_org_id}, \
+             not {org_id}; refusing to delete either resource"
+        )
+        .into());
+    }
+    Ok((Some(api_key_id), false))
+}
+
+async fn cleanup_service_query_key(
+    client: &CloudClient,
+    org_id: &str,
+    service_id: &str,
+    api_key_id: Option<&str>,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let Some(api_key_id) = api_key_id else {
+        return Ok(());
+    };
+
+    client
+        .delete_api_key_if_exists(org_id, api_key_id)
+        .await
+        .map_err(|mut error| {
+            error.message = format!(
+                "failed to delete the auto-provisioned query API key for service \
+                 {service_id}: {}",
+                error.message
+            );
+            error
+        })?;
+    Ok(())
+}
+
 pub async fn service_delete(
     client: &CloudClient,
     service_id: &str,
@@ -1001,12 +1060,16 @@ pub async fn service_delete(
     json: bool,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let org_id = resolve_org_id(client, org_id).await?;
+    let (query_key_id, retain_query_key) = service_query_key_cleanup(&org_id, service_id)?;
 
     if force {
-        let svc = client.get_service(&org_id, service_id).await?;
+        let svc = client.get_service_if_exists(&org_id, service_id).await?;
         // An absent state matches nothing: skip the stop and let the delete
         // call decide, rather than guessing the service is running.
-        let state = or_absent(svc.state.as_ref());
+        let state = svc
+            .as_ref()
+            .map(|service| or_absent(service.state.as_ref()))
+            .unwrap_or_default();
         if matches!(state.as_str(), "running" | "idle" | "starting") {
             eprintln!("Stopping service {} before deletion...", service_id);
             client
@@ -1026,12 +1089,19 @@ pub async fn service_delete(
     }
 
     let response = client
-        .delete_service(&org_id, service_id)
+        .delete_service_if_exists(&org_id, service_id)
         .await
         .map_err(|error| service_delete_error(error, force, service_id))?;
-    let _ = credentials::remove_service_query_key(service_id);
+    // Delete the key only after the service is gone. If cleanup fails, retain
+    // its exact IDs locally so repeating service delete can retry safely.
+    cleanup_service_query_key(client, &org_id, service_id, query_key_id.as_deref()).await?;
+    if !retain_query_key {
+        credentials::remove_service_query_key(service_id)?;
+    }
     if json {
         println!("{}", serde_json::to_string_pretty(&response)?);
+    } else if response.is_none() {
+        println!("Service {} is already absent", service_id);
     } else {
         println!("Service {} deletion initiated", service_id);
     }

crates/clickhousectl/src/cloud/credentials.rs

@@ -16,6 +16,15 @@ pub struct Credentials {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ServiceQueryKey {
+    /// Organization in which the management API key was provisioned.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub organization_id: Option<String>,
+    /// Management API resource ID used to delete this exact key.
+    ///
+    /// Records written before the cleanup metadata was introduced remain
+    /// usable for queries, but cannot be safely cleaned up by service deletion.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub api_key_id: Option<String>,
     pub key_id: String,
     pub key_secret: String,
     /// The query endpoint the key is bound to, when the upsert echoed it.
@@ -67,6 +76,22 @@ pub fn get_service_query_key(service_id: &str) -> Option<ServiceQueryKey> {
     creds.service_query_keys.get(service_id).cloned()
 }
 
+pub fn try_get_service_query_key(
+    service_id: &str,
+) -> Result<Option<ServiceQueryKey>, Box<dyn std::error::Error>> {
+    let path = credentials_path();
+    let data = match std::fs::read_to_string(&path) {
+        Ok(data) => data,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => {
+            return Err(format!("failed to read {}: {error}", path.display()).into());
+        }
+    };
+    let creds: Credentials = serde_json::from_str(&data)
+        .map_err(|error| format!("failed to parse {}: {error}", path.display()))?;
+    Ok(creds.service_query_keys.get(service_id).cloned())
+}
+
 pub fn set_service_query_key(
     service_id: &str,
     key: ServiceQueryKey,
@@ -110,6 +135,8 @@ mod tests {
         creds.service_query_keys.insert(
             "svc-1".into(),
             ServiceQueryKey {
+                organization_id: Some("org-1".into()),
+                api_key_id: Some("api-key-uuid".into()),
                 key_id: "kid".into(),
                 key_secret: "sec".into(),
                 endpoint_id: Some("ep".into()),
@@ -121,8 +148,12 @@ mod tests {
         );
 
         let s = serde_json::to_string(&creds).unwrap();
+        assert!(s.contains("\"organization_id\":\"org-1\""));
+        assert!(s.contains("\"api_key_id\":\"api-key-uuid\""));
         let back: Credentials = serde_json::from_str(&s).unwrap();
         let key = back.service_query_keys.get("svc-1").unwrap();
+        assert_eq!(key.organization_id.as_deref(), Some("org-1"));
+        assert_eq!(key.api_key_id.as_deref(), Some("api-key-uuid"));
         assert_eq!(key.key_id, "kid");
         assert_eq!(key.key_secret, "sec");
         assert_eq!(key.endpoint_id.as_deref(), Some("ep"));
@@ -135,6 +166,8 @@ mod tests {
         creds.service_query_keys.insert(
             "svc-1".into(),
             ServiceQueryKey {
+                organization_id: Some("org-1".into()),
+                api_key_id: Some("api-key-uuid".into()),
                 key_id: "kid".into(),
                 key_secret: "sec".into(),
                 endpoint_id: None,
@@ -155,14 +188,20 @@ mod tests {
     }
 
     #[test]
-    fn existing_credentials_files_with_an_endpoint_id_still_deserialize() {
-        // Files written before `endpoint_id` became optional carry it as a
-        // bare string and must keep loading.
+    fn existing_query_keys_without_cleanup_metadata_still_deserialize() {
+        // Existing files contain query credentials and an endpoint ID, but
+        // not the ownership metadata added for exact cloud-side cleanup.
         let raw = r#"{"service_query_keys":{"svc-1":{"key_id":"kid","key_secret":"sec",
             "endpoint_id":"ep","service_name":"demo","created_at":"2026-05-11T12:00:00Z"}}}"#;
         let creds: Credentials = serde_json::from_str(raw).unwrap();
         let key = creds.service_query_keys.get("svc-1").unwrap();
+        assert_eq!(key.organization_id, None);
+        assert_eq!(key.api_key_id, None);
         assert_eq!(key.endpoint_id.as_deref(), Some("ep"));
         assert_eq!(key.key_id, "kid");
+
+        let written = serde_json::to_string(&creds).unwrap();
+        assert!(!written.contains("organization_id"));
+        assert!(!written.contains("api_key_id"));
     }
 }

crates/clickhousectl/src/cloud/service_query.rs

@@ -8,7 +8,7 @@
 
 use crate::cloud::client::CloudClient;
 use crate::cloud::credentials::{self, ServiceQueryKey};
-use chrono::Utc;
+use chrono::{DateTime, Utc};
 use clickhouse_cloud_api::models::{
     ApiKeyPostRequest, ApiKeyPostRequestState, ApiKeyPostResponse,
     InstanceServiceQueryApiEndpointsPostRequest, IpAccessListEntry,
@@ -41,6 +41,26 @@ fn require_credential_pair(
     Ok((key_id, key_secret))
 }
 
+fn build_service_query_key(
+    organization_id: &str,
+    api_key_id: String,
+    key_id: String,
+    key_secret: String,
+    endpoint_id: Option<String>,
+    service_name: &str,
+    created_at: DateTime<Utc>,
+) -> ServiceQueryKey {
+    ServiceQueryKey {
+        organization_id: Some(organization_id.to_string()),
+        api_key_id: Some(api_key_id),
+        key_id,
+        key_secret,
+        endpoint_id,
+        service_name: service_name.to_string(),
+        created_at,
+    }
+}
+
 /// Discard the API key created for a provisioning attempt that then failed,
 /// so a later retry doesn't leave an orphaned key behind per attempt. Best
 /// effort: the caller is already returning an error, and a key we couldn't
@@ -118,13 +138,15 @@ pub async fn ensure_service_query_setup(
     // `id` is diagnostic only, never an auth input: persist the record
     // without it rather than deleting a working credential and leaving a
     // dangling UUID in the endpoint's `openApiKeys`.
-    let stored = ServiceQueryKey {
+    let stored = build_service_query_key(
+        org_id,
+        api_key_uuid,
         key_id,
         key_secret,
-        endpoint_id: endpoint.id,
-        service_name: service_name.to_string(),
-        created_at: Utc::now(),
-    };
+        endpoint.id,
+        service_name,
+        Utc::now(),
+    );
     credentials::set_service_query_key(service_id, stored.clone())?;
 
     Ok(stored)
@@ -225,6 +247,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn stored_query_key_keeps_the_management_resource_ownership() {
+        let created_at = DateTime::parse_from_rfc3339("2026-05-11T12:00:00Z")
+            .unwrap()
+            .with_timezone(&Utc);
+        let key = build_service_query_key(
+            "org-1",
+            "api-key-uuid".into(),
+            "query-key-id".into(),
+            "query-key-secret".into(),
+            Some("endpoint-id".into()),
+            "demo",
+            created_at,
+        );
+
+        assert_eq!(key.organization_id.as_deref(), Some("org-1"));
+        assert_eq!(key.api_key_id.as_deref(), Some("api-key-uuid"));
+        assert_eq!(key.key_id, "query-key-id");
+        assert_eq!(key.key_secret, "query-key-secret");
+        assert_eq!(key.endpoint_id.as_deref(), Some("endpoint-id"));
+        assert_eq!(key.service_name, "demo");
+        assert_eq!(key.created_at, created_at);
+    }
+
     fn endpoint(
         open_api_keys: Option<Vec<&str>>,
     ) -> clickhouse_cloud_api::models::ServiceQueryAPIEndpoint {