feat(manager): represent restricted deployment teardown credentials

client-sdks/platform/typescript/src/models/deploymentrole.ts

@@ -11,6 +11,7 @@ import { ClosedEnum } from "../types/enums.js";
 export const DeploymentRole = {
   DeploymentViewer: "deployment.viewer",
   DeploymentManager: "deployment.manager",
+  DeploymentTeardown: "deployment.teardown",
   DeploymentDynamicContainerManager: "deployment.dynamic-container-manager",
   DeploymentTelemetryWriter: "deployment.telemetry-writer",
 } as const;

crates/alien-cli/src/commands/destroy.rs

@@ -482,7 +482,9 @@ async fn destroy_tracked_deployment(
         return Ok(());
     }
 
-    let pre_delete_deployment = manager_client
+    // Discovery uses the operator credential; the scoped cleanup token only
+    // authorizes the deletion sync transport.
+    let pre_delete_deployment = operator_client
         .get_deployment()
         .id(&tracked_deployment.deployment_id)
         .send()
@@ -493,13 +495,10 @@ async fn destroy_tracked_deployment(
         })?
         .into_inner();
 
-    if matches!(
-        pre_delete_deployment.status.as_str(),
-        "teardown-required" | "teardown-failed"
-    ) {
+    if pre_delete_deployment.status.as_str() == "teardown-required" {
         steps.activate(2, Some("Setup teardown required".to_string()));
     } else {
-        // Request deletion
+        // Request deletion, including an explicit retry of failed setup teardown.
         operator_client
             .delete_deployment()
             .id(&tracked_deployment.deployment_id)
@@ -622,7 +621,7 @@ async fn destroy_tracked_deployment(
 
     let preparation: Result<()> = async {
         // Re-fetch under lock
-        let deployment = manager_client
+        let deployment = operator_client
             .get_deployment()
             .id(&tracked_deployment.deployment_id)
             .send()
@@ -639,6 +638,7 @@ async fn destroy_tracked_deployment(
                 .context(ErrorData::ConfigurationError {
                     message: format!("Unknown deployment status: {}", deployment.status),
                 })?;
+        current.retry_requested = deployment.retry_requested;
         current.stack_state = deployment
             .stack_state
             .map(serde_json::from_value)
@@ -677,35 +677,51 @@ async fn destroy_tracked_deployment(
         delay_strategy: alien_deployment::runner::DelayStrategy::Inline,
     };
 
-    let runner_result = match alien_deployment::runner::run_step_loop(
+    let setup_result = alien_deployment::setup_teardown::run_setup_teardown_after_handoff(
         &mut current,
         &mut config,
         &client_config,
         &tracked_deployment.deployment_id,
         &policy,
         &transport,
         None,
-        None,
     )
-    .await
-    {
-        Ok(result)
-            if result.loop_result.outcome == LoopOutcome::Success
-                && current.status == DeploymentStatus::TeardownRequired =>
-        {
-            alien_deployment::setup_teardown::run_setup_teardown_after_handoff(
+    .await;
+    let runner_result = match setup_result {
+        Ok(Some(result)) => Ok(result),
+        Err(error) => Err(error),
+        Ok(None) => {
+            match alien_deployment::runner::run_step_loop(
                 &mut current,
                 &mut config,
                 &client_config,
                 &tracked_deployment.deployment_id,
                 &policy,
                 &transport,
                 None,
+                None,
             )
             .await
-            .map(|setup_result| setup_result.unwrap_or(result))
+            {
+                Ok(result)
+                    if result.loop_result.outcome == LoopOutcome::Success
+                        && current.status == DeploymentStatus::TeardownRequired =>
+                {
+                    alien_deployment::setup_teardown::run_setup_teardown_after_handoff(
+                        &mut current,
+                        &mut config,
+                        &client_config,
+                        &tracked_deployment.deployment_id,
+                        &policy,
+                        &transport,
+                        None,
+                    )
+                    .await
+                    .map(|setup_result| setup_result.unwrap_or(result))
+                }
+                other => other,
+            }
         }
-        other => other,
     };
     let semantic_failure_status = runner_result.as_ref().ok().and_then(|result| {
         (result.loop_result.outcome == LoopOutcome::Failure)

crates/alien-deployment/src/manager_api_transport.rs

@@ -495,11 +495,13 @@ fn setup_run_acquire_statuses() -> Vec<String> {
 /// Unlike the normal manager acquire path, this can acquire `teardown-required`
 /// so setup-authority callers can resume frozen-resource teardown.
 ///
-/// `deletion_lookup` confirms that a refused acquire means the record is gone.
-/// Removing the record also removes the deployment's own credential, so a
-/// manager that authenticates through the platform refuses `client` with 401
-/// rather than 404. Callers that requested the deletion with other credentials
-/// pass a client holding them; others pass `client`.
+/// `deletion_lookup` reads the deployment while waiting and confirms that a
+/// refused acquire means the record is gone. Removing the record also removes
+/// the deployment's own credential, so a manager that authenticates through the
+/// platform refuses `client` with 401 rather than 404, and a teardown-only
+/// credential cannot read deployments at all. Callers that requested the
+/// deletion with other credentials pass a client holding them; others pass
+/// `client`.
 pub async fn acquire_setup_delete_deployment(
     client: &ManagerClient,
     deletion_lookup: &ManagerClient,
@@ -575,7 +577,12 @@ pub async fn acquire_setup_delete_deployment(
 
         reject_non_retryable_acquire_reason(deployment_id, &response.not_acquired, true)?;
 
-        let status = match client.get_deployment().id(deployment_id).send().await {
+        let status = match deletion_lookup
+            .get_deployment()
+            .id(deployment_id)
+            .send()
+            .await
+        {
             Ok(resp) => resp.into_inner().status,
             Err(err) => {
                 let message = err.to_string();
@@ -893,6 +900,40 @@ mod tests {
         lookup.assert_async().await;
     }
 
+    /// A teardown-only credential cannot read deployments, so the wait loop reads
+    /// through the lookup client.
+    #[tokio::test]
+    async fn setup_delete_acquire_uses_discovery_when_runtime_has_not_handed_off() {
+        let server = MockServer::start_async().await;
+        let discovery_server = MockServer::start_async().await;
+        let acquire = server
+            .mock_async(|when, then| {
+                when.method(POST).path("/v1/sync/acquire");
+                then.status(200).json_body(serde_json::json!({
+                    "deployments": [], "notAcquired": []
+                }));
+            })
+            .await;
+        let lookup = discovery_server
+            .mock_async(|when, then| {
+                when.method(GET).path("/v1/deployments/deployment-1");
+                then.status(404).body("removed deployment");
+            })
+            .await;
+        let result = acquire_setup_delete_deployment(
+            &ManagerClient::new(&server.base_url()),
+            &ManagerClient::new(&discovery_server.base_url()),
+            "deployment-1",
+            "session-1",
+            DeploymentModel::Push,
+        )
+        .await
+        .expect("confirmed removed record is completed deletion");
+        assert!(matches!(result, SetupDeleteAcquireOutcome::AlreadyDeleted));
+        acquire.assert_async().await;
+        lookup.assert_async().await;
+    }
+
     #[tokio::test]
     async fn setup_delete_acquire_does_not_hide_a_failed_completion_lookup() {
         let server = MockServer::start_async().await;