fix: keep refresh and frozen IAM setup-owned

crates/alien-deployment/src/running.rs

@@ -61,13 +61,6 @@ pub async fn handle_running(
         )?;
     }
 
-    // TODO: Add mechanism to limit executor to only perform read-only health checks
-    // and prevent any mutable operations on cloud resources during refresh.
-    // This could be done by:
-    // 1. Adding a "read-only" mode to StackExecutor
-    // 2. Having controllers check this mode and skip any mutating operations
-    // 3. Or creating a separate HealthCheckExecutor that only calls read methods
-
     // Create executor with the target stack configuration
     // No lifecycle filter - check all resources during health checks
     let executor = StackExecutor::builder(&target_stack, client_config)
@@ -82,7 +75,7 @@ pub async fn handle_running(
     // The Ready handlers perform heartbeat checks (e.g., verify function still exists, bucket is accessible)
     let step_result =
         executor
-            .step(stack_state)
+            .refresh(stack_state)
             .await
             .context(ErrorData::StackExecutionFailed {
                 message: "Failed to execute health check step".to_string(),

crates/alien-deployment/src/updating.rs

@@ -263,18 +263,6 @@ pub async fn handle_updating(
                 message: "Pending prepared stack not found in runtime metadata".to_string(),
             })
         })?;
-    let setup_update_authorized = runtime_metadata
-        .setup_update_authorization
-        .as_ref()
-        .is_some_and(|authorization| {
-            authorization.target_frozen_digest == target_stack.frozen_resources_digest()
-                && current
-                    .target_release
-                    .as_ref()
-                    .and_then(|release| release.release_id.as_deref())
-                    == Some(authorization.release_id.as_str())
-        });
-
     // Inject environment variables into the prepared stack
     crate::helpers::inject_environment_variables(&mut target_stack, &config, current.platform)?;
 
@@ -305,19 +293,10 @@ pub async fn handle_updating(
         debug!("Secrets already synced, continuing with update");
     }
 
-    // Create executor for resources
-    // By default, only deploy live resources (frozen resources don't change)
-    // If allow_frozen_changes is true, also deploy frozen resources
-    let mut lifecycle_filter_vec = vec![ResourceLifecycle::Live];
-    if config.allow_frozen_changes || setup_update_authorized {
-        info!("Including frozen resources in authorized update");
-        lifecycle_filter_vec.push(ResourceLifecycle::Frozen);
-    }
-
     let executor = StackExecutor::builder(&target_stack, client_config)
         .deployment_config(&config)
         .running_resource_policy(RunningResourcePolicy::OptIn)
-        .lifecycle_filter(lifecycle_filter_vec)
+        .lifecycle_filter(vec![ResourceLifecycle::Live])
         .service_provider(service_provider)
         .build()
         .context(ErrorData::StackExecutionFailed {

crates/alien-infra/src/core/executor.rs

@@ -620,7 +620,10 @@ impl StackExecutor {
     fn external_binding_state(
         &self,
         resource_id: &str,
-    ) -> Result<(Option<serde_json::Value>, Option<alien_core::ResourceOutputs>)> {
+    ) -> Result<(
+        Option<serde_json::Value>,
+        Option<alien_core::ResourceOutputs>,
+    )> {
         let Some(binding) = self.deployment_config.external_bindings.get(resource_id) else {
             return Ok((None, None));
         };
@@ -1139,36 +1142,55 @@ impl StackExecutor {
     /// The method is entirely *stateless* from the executor's perspective: all
     /// data required for the next iteration is carried within the returned
     /// `StackState`.
+    pub async fn step(&self, state: StackState) -> Result<StepResult> {
+        self.step_inner(state, true).await
+    }
+
+    /// Steps controllers that are already active without planning stack changes.
+    ///
+    /// Refresh deliberately does not compare the persisted state with the desired
+    /// stack, resume failed updates, or initiate create/update/delete transitions.
+    /// It is suitable for periodic observation of a Running deployment.
+    pub async fn refresh(&self, state: StackState) -> Result<StepResult> {
+        self.step_inner(state, false).await
+    }
+
     #[alien_event(AlienEvent::StackStep {
         next_state: state.clone(),
         suggested_delay_ms: None,
     })]
-    pub async fn step(&self, state: StackState) -> Result<StepResult> {
+    async fn step_inner(&self, state: StackState, apply_plan: bool) -> Result<StepResult> {
         validate_stack_controller_state_versions(&state)?;
 
         let mut state = state;
-        for (resource_id, resource_state) in &mut state.resources {
-            if resource_state.status != ResourceStatus::UpdateFailed {
-                continue;
-            }
-            let Some(desired) = self.resources.get(resource_id) else {
-                continue;
-            };
-            if resource_state.config != desired.resource {
-                continue;
-            }
-            if resource_state.retry_failed()? {
-                debug!(
-                    "Resumed unchanged failed update for '{}' before planning",
-                    resource_id
-                );
+        if apply_plan {
+            for (resource_id, resource_state) in &mut state.resources {
+                if resource_state.status != ResourceStatus::UpdateFailed {
+                    continue;
+                }
+                let Some(desired) = self.resources.get(resource_id) else {
+                    continue;
+                };
+                if resource_state.config != desired.resource {
+                    continue;
+                }
+                if resource_state.retry_failed()? {
+                    debug!(
+                        "Resumed unchanged failed update for '{}' before planning",
+                        resource_id
+                    );
+                }
             }
         }
 
         let mut next_state = state.clone(); // Clone the input state to modify
 
         // --- Planning Phase ---
-        let plan_result = self.plan(&state)?;
+        let plan_result = if apply_plan {
+            self.plan(&state)?
+        } else {
+            PlanResult::default()
+        };
         debug!(
             "Plan result: {} creates, {} updates, {} deletes",
             plan_result.creates.len(),
@@ -1663,11 +1685,12 @@ impl StackExecutor {
             let context_resource: Resource;
 
             // Use current desired config from the stack if it exists, otherwise use stored config for deletion
-            context_resource = if let Some(resource_config) = self.resources.get(&resource_id) {
-                // Resource is still in desired stack (create/update case)
-                resource_config.resource.clone()
+            context_resource = if apply_plan {
+                self.resources
+                    .get(&resource_id)
+                    .map(|resource_config| resource_config.resource.clone())
+                    .unwrap_or_else(|| current_resource_state.config.clone())
             } else {
-                // Resource is not in desired stack (deletion case) - use stored config from stack state
                 current_resource_state.config.clone()
             };
 
@@ -1890,8 +1913,10 @@ impl StackExecutor {
                 };
 
             // Automatically update config to match desired state (except during deletion)
-            let next_config = if current_resource_state.status == ResourceStatus::Deleting {
-                // During deletion, preserve the current config
+            let next_config = if !apply_plan
+                || current_resource_state.status == ResourceStatus::Deleting
+            {
+                // Refresh and deletion preserve the deployed config.
                 current_resource_state.config.clone()
             } else {
                 // For create/update operations, ensure config matches desired state

crates/alien-infra/src/core/executor_tests/update_tests.rs

@@ -38,6 +38,41 @@ async fn test_config_change_triggers_update() -> Result<()> {
     Ok(())
 }
 
+/// Refresh observes the persisted deployment; it must not turn desired release
+/// drift into an update.
+#[tokio::test]
+async fn test_refresh_does_not_plan_config_changes() -> Result<()> {
+    let func_v1 = test_function_with_image("func1", "image-v1");
+    let stack_v1 = Stack::new("refresh-test".to_owned())
+        .add(func_v1.clone(), ResourceLifecycle::Live)
+        .build();
+    let state = run_to_synced(&new_executor(&stack_v1)?, new_test_state()).await?;
+
+    let stack_v2 = Stack::new("refresh-test".to_owned())
+        .add(
+            test_function_with_image("func1", "image-v2"),
+            ResourceLifecycle::Live,
+        )
+        .add(test_storage("new-storage"), ResourceLifecycle::Frozen)
+        .build();
+    let refreshed = new_executor(&stack_v2)?.refresh(state).await?.next_state;
+
+    assert_eq!(
+        refreshed.resources.len(),
+        1,
+        "refresh must not create desired resources missing from persisted state"
+    );
+    let resource = refreshed.resources.get("func1").unwrap();
+    assert_eq!(resource.status, ResourceStatus::Running);
+    assert_eq!(
+        resource.config,
+        Resource::new(func_v1),
+        "refresh must retain the deployed config instead of applying desired drift"
+    );
+
+    Ok(())
+}
+
 /// Tests that config changes while a resource is still provisioning do not
 /// interrupt the in-flight create. The update should happen after create reaches
 /// a stable state.

crates/alien-infra/src/core/resource_permissions_helper.rs

@@ -1069,6 +1069,15 @@ impl ResourcePermissionsHelper {
         resource_name: &str,
         resource_type: &str,
     ) -> Result<()> {
+        if !Self::resource_is_setup_owned(ctx, resource_id) {
+            debug!(
+                resource_id = %resource_id,
+                resource_name = %resource_name,
+                "Skipping AWS resource-scoped data policy attachment for live resource; these policies are setup-owned"
+            );
+            return Ok(());
+        }
+
         let aws_config = ctx.get_aws_config()?;
 
         // Build permission context for this specific resource
@@ -1132,34 +1141,16 @@ impl ResourcePermissionsHelper {
             }
         }
 
-        if Self::resource_is_setup_owned(ctx, resource_id) {
-            // Setup-owned resources run while setup credentials are still active.
-            // Live resource controllers must not edit the management role after
-            // the deployment has moved to provisioning credentials.
-            Self::apply_aws_management_resource_permissions(
-                ctx,
-                resource_id,
-                resource_name,
-                resource_type,
-                &generator,
-                &permission_context,
-            )
-            .await?;
-        } else if ctx
-            .desired_stack
-            .management()
-            .profile()
-            .map(|profile| {
-                !Self::aws_management_resource_permission_refs(profile, resource_id).is_empty()
-            })
-            .unwrap_or(false)
-        {
-            debug!(
-                resource_id = %resource_id,
-                resource_name = %resource_name,
-                "Skipping AWS management resource-scoped permissions for live resource; setup must grant them"
-            );
-        }
+        // Setup-owned resources run while setup credentials are still active.
+        Self::apply_aws_management_resource_permissions(
+            ctx,
+            resource_id,
+            resource_name,
+            resource_type,
+            &generator,
+            &permission_context,
+        )
+        .await?;
 
         Ok(())
     }

crates/alien-manager/src/routes/stack.rs

@@ -571,9 +571,30 @@ fn merge_reimported_stack_state(
             })?;
         imported.resources.insert(resource.id.clone(), merged);
     }
+
+    // The setup payload is authoritative for setup-owned resources: omission
+    // means setup removed one. Runtime-owned resources are not present in that
+    // payload, so carry their current controller state across explicitly.
+    preserve_live_runtime_state(stack, existing, &mut imported);
+
     Ok(imported)
 }
 
+fn preserve_live_runtime_state(stack: &Stack, existing: &StackState, imported: &mut StackState) {
+    for (resource_id, entry) in stack.resources() {
+        if entry.lifecycle != alien_core::ResourceLifecycle::Live
+            || imported.resources.contains_key(resource_id)
+        {
+            continue;
+        }
+        if let Some(existing_resource) = existing.resources.get(resource_id) {
+            imported
+                .resources
+                .insert(resource_id.clone(), existing_resource.clone());
+        }
+    }
+}
+
 fn assert_supported_import_region(
     config: &crate::config::ManagerConfig,
     req: &StackImportRequest,
@@ -1408,4 +1429,47 @@ mod setup_update_authorization_tests {
         );
         assert!(metadata.registry_access_granted);
     }
+
+    #[test]
+    fn reimport_preserves_live_state_without_resurrecting_removed_frozen_state() {
+        let target = stack("live", "frozen-new");
+        let mut existing = StackState::new(Platform::Aws);
+        for resource_id in ["live", "frozen-old"] {
+            existing.resources.insert(
+                resource_id.to_string(),
+                StackResourceState::new_pending(
+                    if resource_id == "live" {
+                        Worker::RESOURCE_TYPE.to_string()
+                    } else {
+                        Storage::RESOURCE_TYPE.to_string()
+                    },
+                    target
+                        .resources
+                        .get(if resource_id == "live" {
+                            "live"
+                        } else {
+                            "frozen-new"
+                        })
+                        .unwrap()
+                        .config
+                        .clone(),
+                    Some(if resource_id == "live" {
+                        ResourceLifecycle::Live
+                    } else {
+                        ResourceLifecycle::Frozen
+                    }),
+                    vec![],
+                ),
+            );
+        }
+        let mut imported = StackState::new(Platform::Aws);
+
+        preserve_live_runtime_state(&target, &existing, &mut imported);
+
+        assert!(imported.resources.contains_key("live"));
+        assert!(
+            !imported.resources.contains_key("frozen-old"),
+            "an omitted setup-owned resource must remain removed"
+        );
+    }
 }

crates/alien-manager/src/stores/sqlite/deployment.rs

@@ -714,19 +714,11 @@ impl DeploymentStore for SqliteDeploymentStore {
             schedule_reconciliation,
             input_values,
         } = params;
-        let mut merged_stack_state = stack_state;
-        if let Some(existing) = self.get_deployment(caller, deployment_id).await? {
-            if let Some(mut existing_stack_state) = existing.stack_state {
-                for (resource_id, resource_state) in merged_stack_state.resources {
-                    existing_stack_state
-                        .resources
-                        .insert(resource_id, resource_state);
-                }
-                merged_stack_state = existing_stack_state;
-            }
-        }
-
-        let stack_state_json = serde_json::to_string(&merged_stack_state)
+        // The import route supplies a complete ownership-aware state: freshly
+        // imported setup resources plus preserved Live controller state.
+        // Re-merging here would resurrect setup-owned resources that the new
+        // setup payload intentionally omitted.
+        let stack_state_json = serde_json::to_string(&stack_state)
             .into_alien_error()
             .context(GenericError {
                 message: "Failed to serialize imported stack_state".to_string(),