brokk-mj-controller 2.5.0

//! Session close, force-stop, and permanent-destruction transitions.

use std::time::Duration;

use anyhow::{Context, Result, bail, ensure};

use crate::hel_session_manager::{SessionManagerControl, new_command_id};
use hel::hel_state::{CheckpointMetadata, SessionRecord, SessionState};
use hel::hel_targets::{
    self, CommandExecutor, ProcessExecutor, ProvisionStage, ProvisionStageGuard,
};
use hel::hel_worker::{RelayCommand, RelayExecutionState};

use super::backend::backend_locator;
use super::checkpoint::{
    CheckpointExportPolicy, LatchExclusivity, prune_replaced_checkpoint,
    release_projection_behind_checkpoint, verify_installed_checkpoint_gate, wait_for_relay_closed,
};
use super::worktree::{cleanup_managed_worktree, retire_managed_worktree};
use super::{Controller, now, persist_session_record_transition_or_restore};

impl Controller {
    /// Checkpoint, ask the harness to close, and only then tear down the exact
    /// provisioned target. Checkpoint failure is deliberately non-destructive.
    pub async fn close_session(&mut self, session_id: &str) -> Result<()> {
        self.close_session_controlled(session_id, &ProcessExecutor)
            .await
    }

    pub async fn close_session_controlled(
        &mut self,
        session_id: &str,
        executor: &(impl CommandExecutor + Sync),
    ) -> Result<()> {
        if self
            .close_session_controlled_with_manager(session_id, executor, None, None)
            .await?
        {
            self.cleanup_stopped_target(session_id, executor)?;
        }
        Ok(())
    }

    pub async fn close_session_managed_controlled(
        &mut self,
        session_id: &str,
        executor: &(impl CommandExecutor + Sync),
        manager: &SessionManagerControl,
    ) -> Result<bool> {
        self.close_session_controlled_with_manager(session_id, executor, Some(manager), None)
            .await
    }

    pub(super) async fn close_session_for_move(
        &mut self,
        session_id: &str,
        executor: &(impl CommandExecutor + Sync),
        manager: &SessionManagerControl,
        operation: &mut hel::hel_state::MoveOperation,
        preparation: Option<&hel::hel_state::MovePreparation>,
    ) -> Result<bool> {
        self.close_session_controlled_with_manager(
            session_id,
            executor,
            Some(manager),
            Some((operation, preparation)),
        )
        .await
    }

    async fn close_session_controlled_with_manager(
        &mut self,
        session_id: &str,
        executor: &(impl CommandExecutor + Sync),
        manager: Option<&SessionManagerControl>,
        move_intent: Option<(
            &mut hel::hel_state::MoveOperation,
            Option<&hel::hel_state::MovePreparation>,
        )>,
    ) -> Result<bool> {
        let previous = self
            .state
            .sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?
            .clone();
        let record = self.state.sessions.get_mut(session_id).unwrap();
        // Persist the close intent before beginning its checkpoint. A process
        // exit anywhere below must leave enough state for the next controller
        // to retry the close, even when no checkpoint has been installed yet.
        apply_close_checkpoint_started(record, now());
        self.persist_session_transition_or_restore(
            session_id,
            &previous,
            "persist closing state before checkpointing the session",
        )?;

        // Close seals the relay at the exact latched cursor, so this checkpoint
        // keeps its exclusive connection until the relay reports Closed.
        let mut latched = match self
            .checkpoint_session_latched(
                session_id,
                executor,
                manager,
                LatchExclusivity::HoldThroughClose,
                CheckpointExportPolicy::ReuseUnchangedArchive,
            )
            .await
        {
            Ok(latched) => latched,
            Err(error) => {
                let record = self.state.sessions.get_mut(session_id).unwrap();
                record.state = previous.state;
                record.updated_at = now();
                record.last_checkpoint_error = Some(format!("{error:#}"));
                return Err(
                    self.persist_failed_checkpoint_state_or_restore(session_id, &previous, error)
                );
            }
        };

        let artifact = latched.artifact.clone();
        let record = self.state.sessions.get_mut(session_id).unwrap();
        record.state = SessionState::Closing;
        record.native_session_id = Some(artifact.native_session_id.clone());
        record.checkpoint = Some(artifact.metadata.clone());
        record.updated_at = now();
        record.last_error = None;
        record.last_checkpoint_error = None;
        self.persist_checkpoint_transition_or_restore(
            session_id,
            &previous,
            "persist verified checkpoint and closing state before sealing the relay",
        )?;
        if let Some((operation, preparation)) = move_intent {
            // The source is still behind an unsealed barrier. A destination
            // preflight error must release it and leave its processes alive.
            if let Err(error) = self.validate_move_checkpoint(operation, preparation, executor) {
                let record = self.state.sessions.get_mut(session_id).unwrap();
                record.state = previous.state;
                record.last_error = Some(format!("{error:#}"));
                self.persist_session_transition_or_restore(
                    session_id,
                    &previous,
                    "restore source after move preflight failure",
                )?;
                return Err(error);
            }
            operation.checkpoint = Some(artifact.metadata.clone());
            operation.updated_at = now();
            hel::hel_database::save_move_operation(operation)?;
        }
        prune_replaced_checkpoint(previous.checkpoint.as_ref(), &artifact.metadata);
        // A stopping session will not checkpoint again, so this is its last
        // chance to release what its checkpoint now covers.
        release_projection_behind_checkpoint(session_id, &artifact.metadata);

        let close_command_id = new_command_id("close")?;
        let barrier_command_id = latched.barrier_command_id.clone();
        let close_result = {
            let _closing = ProvisionStageGuard::new(executor, ProvisionStage::Closing);
            latched
                .relay
                .connection_mut()
                .submit(
                    close_command_id,
                    RelayCommand::Close {
                        barrier_command_id: barrier_command_id.clone(),
                        expected: latched.cursor.clone(),
                    },
                )
                .await
        };
        if let Err(error) = close_result {
            self.record_interrupted_close(session_id, &error)?;
            return Err(error.context("seal verified checkpoint for close"));
        }
        let close_result = {
            let _closing = ProvisionStageGuard::new(executor, ProvisionStage::Closing);
            latched
                .relay
                .connection_mut()
                .submit(
                    new_command_id("checkpoint-complete")?,
                    RelayCommand::CompleteCheckpoint { barrier_command_id },
                )
                .await
        };
        if let Err(error) = close_result {
            self.record_interrupted_close(session_id, &error)?;
            return Err(error.context("release verified close checkpoint"));
        }
        let close_result = {
            let _closing = ProvisionStageGuard::new(executor, ProvisionStage::Closing);
            wait_for_relay_closed(latched.relay.connection_mut()).await
        };
        if let Err(error) = close_result {
            self.record_interrupted_close(session_id, &error)?;
            return Err(error);
        }
        latched.relay.release();

        match self.destroy_after_verified_checkpoint(session_id, &artifact.metadata, executor) {
            Ok(deferred) => Ok(deferred),
            Err(error) => {
                self.record_interrupted_close(session_id, &error)?;
                Err(error)
            }
        }
    }

    /// Resume the durable closing state after a controller restart. If the
    /// relay had accepted Close, wait for it and destroy through the exact
    /// installed checkpoint gate. If it had not, take a fresh checkpoint;
    /// the previously installed archive may have become stale after EOF
    /// released its barrier.
    pub async fn recover_interrupted_close_managed(
        &mut self,
        session_id: &str,
        executor: &(impl CommandExecutor + Sync),
        manager: &SessionManagerControl,
    ) -> Result<bool> {
        let (state, verified) = {
            let session = self
                .state
                .sessions
                .get(session_id)
                .with_context(|| format!("unknown session {session_id}"))?;
            ensure!(
                matches!(
                    session.state,
                    SessionState::Closing | SessionState::Destroying
                ),
                "session {session_id} has no interrupted close to recover"
            );
            (session.state, session.checkpoint.clone())
        };
        if state == SessionState::Destroying {
            let verified = verified.context("destroying session has no verified checkpoint")?;
            return self.destroy_after_verified_checkpoint(session_id, &verified, executor);
        }
        ensure!(
            state == SessionState::Closing,
            "session {session_id} has no relay close to recover"
        );
        let handle = manager
            .wait_for_session(session_id, Duration::from_secs(5))
            .await?;
        let mut lease = handle.lease_connection().await?;
        let execution = lease.connection_mut().sync().await?.operational.execution;
        match execution {
            RelayExecutionState::Closed => {}
            RelayExecutionState::Closing => {
                let _closing = ProvisionStageGuard::new(executor, ProvisionStage::Closing);
                wait_for_relay_closed(lease.connection_mut()).await?;
            }
            RelayExecutionState::Idle | RelayExecutionState::Running => {
                lease.release();
                return self
                    .close_session_controlled_with_manager(
                        session_id,
                        executor,
                        Some(manager),
                        None,
                    )
                    .await;
            }
        }
        lease.release();
        let verified = verified.context("closed relay has no verified checkpoint")?;
        self.destroy_after_verified_checkpoint(session_id, &verified, executor)
    }

    fn record_interrupted_close(&mut self, session_id: &str, error: &anyhow::Error) -> Result<()> {
        let record = self.state.sessions.get_mut(session_id).unwrap();
        apply_interrupted_close_error(record, error, &now());
        self.persist_session_state(session_id)
    }

    /// Execute cleanup only after the close state machine has installed a
    /// verified checkpoint on the record.
    fn destroy_after_verified_checkpoint(
        &mut self,
        session_id: &str,
        verified: &CheckpointMetadata,
        executor: &impl CommandExecutor,
    ) -> Result<bool> {
        self.destroy_after_verified_checkpoint_with(
            session_id,
            verified,
            executor,
            hel::hel_database::save_lifecycle_session,
        )
    }

    fn destroy_after_verified_checkpoint_with(
        &mut self,
        session_id: &str,
        verified: &CheckpointMetadata,
        executor: &impl CommandExecutor,
        persist: impl Fn(&SessionRecord) -> Result<()>,
    ) -> Result<bool> {
        let session = self
            .state
            .sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?
            .clone();
        ensure!(
            matches!(
                session.state,
                SessionState::Closing | SessionState::Destroying
            ),
            "refusing to destroy session {session_id}: it is not closing or destroying"
        );
        ensure!(
            session.checkpoint.as_ref() == Some(verified),
            "refusing to destroy session {session_id}: verified checkpoint gate is stale"
        );
        if session.state == SessionState::Closing {
            let record = self.state.sessions.get_mut(session_id).unwrap();
            record.state = SessionState::Destroying;
            record.updated_at = now();
            record.last_error = None;
            persist_session_record_transition_or_restore(
                &mut self.state,
                session_id,
                &session,
                "persist destroying state before target cleanup",
                &persist,
            )?;
        }

        let destroying = self
            .state
            .sessions
            .get(session_id)
            .expect("destroying session disappeared")
            .clone();
        {
            let _verifying = ProvisionStageGuard::new(executor, ProvisionStage::Verifying);
            verify_installed_checkpoint_gate(session_id, verified)?;
        }
        // The reviewer's native session lives on the target that is about to
        // go. Recording that now, before the target is torn down, is what
        // stops a resumed session from trying to reload a conversation that no
        // longer exists; its transcript is kept for reference either way.
        if let Err(error) = hel::hel_database::lose_reviewer_continuity(session_id) {
            tracing::warn!(
                session_id,
                error = format!("{error:#}"),
                "could not record that the second-opinion conversation ends with this target"
            );
        }
        let locator = destroying
            .target
            .as_ref()
            .context("session has no target")?;
        let backend = backend_locator(locator, &destroying, &self.config)?;
        let deferred = if let Some(plan) = hel_targets::quiesce_plan(&backend, session_id)? {
            plan.execute(executor)?;
            true
        } else {
            execute_target_cleanup(&backend, session_id, executor)?;
            false
        };
        if let Some(worktree) = &destroying.managed_worktree {
            retire_managed_worktree(executor, worktree)
                .context("retire managed raw-session worktree after verified close")?;
        }
        let record = self.state.sessions.get_mut(session_id).unwrap();
        record.state = SessionState::Stopped;
        if !deferred {
            record.target = None;
        }
        record.updated_at = now();
        record.last_error = None;
        persist_session_record_transition_or_restore(
            &mut self.state,
            session_id,
            &destroying,
            "persist stopped state after target cleanup",
            &persist,
        )?;
        Ok(deferred)
    }

    /// Finish storage cleanup for a stopped Podman target retained by the
    /// quiescence transition. The locator stays durable until every command
    /// succeeds, making daemon restart and explicit retry idempotent.
    pub fn cleanup_stopped_target(
        &mut self,
        session_id: &str,
        executor: &impl CommandExecutor,
    ) -> Result<()> {
        self.cleanup_stopped_target_with(
            session_id,
            executor,
            hel::hel_database::save_lifecycle_session,
        )
    }

    fn cleanup_stopped_target_with(
        &mut self,
        session_id: &str,
        executor: &impl CommandExecutor,
        persist: impl Fn(&SessionRecord) -> Result<()>,
    ) -> Result<()> {
        let previous = self
            .state
            .sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?
            .clone();
        ensure!(
            previous.state == SessionState::Stopped,
            "refusing deferred cleanup for active session {session_id}"
        );
        let Some(locator) = previous.target.as_ref() else {
            return Ok(());
        };
        let backend = backend_locator(locator, &previous, &self.config)?;
        ensure!(
            hel_targets::quiesce_plan(&backend, session_id)?.is_some(),
            "session {session_id} retained a non-Podman target after stopping"
        );
        if let Err(error) = execute_target_cleanup(&backend, session_id, executor) {
            let record = self.state.sessions.get_mut(session_id).unwrap();
            record.updated_at = now();
            record.last_error = Some(format!("deferred target cleanup failed: {error:#}"));
            let persisted = persist_session_record_transition_or_restore(
                &mut self.state,
                session_id,
                &previous,
                "persist deferred target cleanup failure",
                &persist,
            );
            return match persisted {
                Ok(()) => Err(error),
                Err(persist_error) => Err(error.context(format!(
                    "also failed to persist deferred target cleanup failure: {persist_error:#}"
                ))),
            };
        }
        let record = self.state.sessions.get_mut(session_id).unwrap();
        record.target = None;
        record.updated_at = now();
        record.last_error = None;
        persist_session_record_transition_or_restore(
            &mut self.state,
            session_id,
            &previous,
            "persist completion of deferred Podman target cleanup",
            &persist,
        )
    }

    /// Tear down the current target without taking a fresh checkpoint, then
    /// leave the logical session resumable from its latest verified archive.
    pub fn force_stop(
        &mut self,
        session_id: &str,
        executor: &impl CommandExecutor,
    ) -> Result<bool> {
        self.force_stop_with(
            session_id,
            executor,
            hel::hel_database::save_lifecycle_session,
        )
    }

    fn force_stop_with(
        &mut self,
        session_id: &str,
        executor: &impl CommandExecutor,
        persist: impl Fn(&SessionRecord) -> Result<()>,
    ) -> Result<bool> {
        let session = self
            .state
            .sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?
            .clone();
        ensure!(
            session.state.is_active(),
            "session {session_id} is already inactive"
        );
        let checkpoint = session
            .checkpoint
            .as_ref()
            .context("force stop requires an existing recovery archive")?;
        // Force stop skips a new checkpoint, never the checksum gate on the
        // archive that makes the logical session resumable afterwards.
        {
            let _verifying = ProvisionStageGuard::new(executor, ProvisionStage::Verifying);
            verify_installed_checkpoint_gate(session_id, checkpoint)
                .context("verify the recovery archive before force stopping")?;
        }
        let mut deferred = false;
        if let Some(locator) = &session.target {
            let backend = backend_locator(locator, &session, &self.config)?;
            if let Some(plan) = hel_targets::quiesce_plan(&backend, session_id)? {
                plan.execute(executor)?;
                deferred = true;
            } else {
                execute_target_cleanup(&backend, session_id, executor)?;
            }
        }
        if let Some(worktree) = &session.managed_worktree {
            retire_managed_worktree(executor, worktree)
                .context("retire managed raw-session worktree after force stop")?;
        }
        let record = self.state.sessions.get_mut(session_id).unwrap();
        record.state = SessionState::Stopped;
        if !deferred {
            record.target = None;
        }
        record.updated_at = now();
        record.last_error = None;
        record.last_checkpoint_error = None;
        persist_session_record_transition_or_restore(
            &mut self.state,
            session_id,
            &session,
            "persist stopped state after force stopping the current target",
            &persist,
        )?;
        Ok(deferred)
    }

    /// Permanently destroy an inactive session and every artifact Hel owns for it.
    /// External cleanup happens before the durable record is dropped so failures
    /// remain visible and retryable.
    pub fn destroy_session_controlled(
        &mut self,
        session_id: &str,
        executor: &impl CommandExecutor,
    ) -> Result<()> {
        let session = self
            .state
            .sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?
            .clone();
        if session.state.is_active() {
            bail!("refusing to destroy active session {session_id}");
        }
        if let Some(worktree) = &session.managed_worktree {
            cleanup_managed_worktree(executor, worktree)
                .context("remove managed raw-session worktree")?;
        }
        if let Some(checkpoint) = &session.checkpoint
            && let Err(error) = std::fs::remove_file(&checkpoint.archive_path)
            && error.kind() != std::io::ErrorKind::NotFound
        {
            return Err(error).with_context(|| {
                format!(
                    "remove session recovery archive {}",
                    checkpoint.archive_path.display()
                )
            });
        }
        hel::hel_attachment::AttachmentStore::controller(session_id)?
            .remove_session_data()
            .context("remove session image attachments")?;
        hel::hel_database::delete_session(session_id)
            .context("destroy stopped session in database")?;
        self.state.destroy_stopped_session(session_id)?;
        Ok(())
    }

    /// Permanently destroy a session from any state, without checkpointing
    /// and without requiring a recovery archive.
    ///
    /// Unlike [`Controller::destroy_session_controlled`], this accepts active
    /// states: it tears the live target down with the same close plan a
    /// verified close uses, so the owning process group dies before any files
    /// go. External cleanup happens before the durable record is dropped so
    /// failures stay visible and retryable; the recovery archive is removed,
    /// which is what makes the destruction irreversible.
    pub fn force_destroy_session(
        &mut self,
        session_id: &str,
        executor: &impl CommandExecutor,
    ) -> Result<()> {
        self.force_destroy_session_with(session_id, executor, hel::hel_database::delete_session)
    }

    fn force_destroy_session_with(
        &mut self,
        session_id: &str,
        executor: &impl CommandExecutor,
        delete: impl Fn(&str) -> Result<()>,
    ) -> Result<()> {
        let session = self
            .state
            .sessions
            .get(session_id)
            .with_context(|| format!("unknown session {session_id}"))?
            .clone();
        // A session destroyed for good keeps nothing, including a broker an
        // earlier failure left running; retiring it first also stops a live
        // writer from recreating files under the teardown below.
        if let Some(locator) = &session.target {
            let backend = backend_locator(locator, &session, &self.config)?;
            execute_target_cleanup(&backend, session_id, executor)?;
        }
        if let Some(worktree) = &session.managed_worktree {
            cleanup_managed_worktree(executor, worktree)
                .context("remove managed raw-session worktree")?;
        }
        if let Some(checkpoint) = &session.checkpoint
            && let Err(error) = std::fs::remove_file(&checkpoint.archive_path)
            && error.kind() != std::io::ErrorKind::NotFound
        {
            return Err(error).with_context(|| {
                format!(
                    "remove session recovery archive {}",
                    checkpoint.archive_path.display()
                )
            });
        }
        hel::hel_attachment::AttachmentStore::controller(session_id)?
            .remove_session_data()
            .context("remove session image attachments")?;
        delete(session_id).context("force destroy session in database")?;
        self.state.destroy_session_force(session_id)?;
        Ok(())
    }
}

fn execute_target_cleanup(
    backend: &hel_targets::TargetLocator,
    session_id: &str,
    executor: &impl CommandExecutor,
) -> Result<()> {
    if let Err(cleanup_error) = hel_targets::close_plan(backend, session_id)?.execute(executor) {
        match hel_targets::cleanup_target_is_confirmed_absent(backend, session_id, executor) {
            Ok(true) => {
                tracing::warn!(
                    session_id,
                    error = format!("{cleanup_error:#}"),
                    "target cleanup command failed, but the target was confirmed absent"
                );
            }
            Ok(false) => {
                tracing::error!(
                    session_id,
                    error = format!("{cleanup_error:#}"),
                    "target cleanup failed and the target is still present"
                );
                return Err(cleanup_error);
            }
            Err(probe_error) => {
                tracing::error!(
                    session_id,
                    cleanup_error = format!("{cleanup_error:#}"),
                    probe_error = format!("{probe_error:#}"),
                    "target cleanup failed and exact absence could not be confirmed"
                );
                return Err(cleanup_error.context(format!(
                    "target cleanup failed and exact absence could not be confirmed: {probe_error:#}"
                )));
            }
        }
    }
    Ok(())
}

fn apply_close_checkpoint_started(record: &mut SessionRecord, updated_at: String) {
    record.state = SessionState::Closing;
    record.updated_at = updated_at;
    record.last_checkpoint_error = None;
}

fn apply_interrupted_close_error(
    record: &mut SessionRecord,
    error: &anyhow::Error,
    updated_at: &str,
) {
    let destroying = record.state == SessionState::Destroying;
    if !destroying {
        record.state = SessionState::Closing;
    }
    record.updated_at = updated_at.to_owned();
    record.last_error = Some(if destroying {
        format!("target cleanup is safely retryable from its verified checkpoint: {error:#}")
    } else {
        format!("close is safely resumable from its verified checkpoint: {error:#}")
    });
}

#[cfg(test)]
mod tests {
    use std::cell::RefCell;
    use std::collections::BTreeMap;

    use anyhow::Result;

    use crate::hel_controller::Controller;
    use crate::hel_controller::test_support::{
        checkpoint_test_session, committed_repository, managed_worktree_session, test_git,
        write_checkpoint_gate_archive,
    };
    use hel::hel_config::{ContainerTemplate as ConfigContainer, HelConfig, TargetTemplate};
    use hel::hel_state::{HelState, SessionState, TargetLocator};
    use hel::hel_targets::{self, CommandExecutor, CommandOutput, CommandSpec, ProcessExecutor};

    use super::*;

    #[test]
    fn starting_close_persists_its_intent_before_checkpointing() {
        let mut session = checkpoint_test_session("0123456789abcdef0123456789abcdef");
        session.state = SessionState::Running;
        session.last_checkpoint_error = Some("old failure".into());

        apply_close_checkpoint_started(&mut session, "2026-08-14T12:00:00Z".into());

        assert_eq!(session.state, SessionState::Closing);
        assert_eq!(session.updated_at, "2026-08-14T12:00:00Z");
        assert!(session.last_checkpoint_error.is_none());
    }

    struct DeferredCleanupExecutor {
        statuses: RefCell<Vec<i32>>,
    }

    impl CommandExecutor for DeferredCleanupExecutor {
        fn execute(&self, _command: &CommandSpec) -> Result<CommandOutput> {
            let status = self
                .statuses
                .borrow_mut()
                .pop()
                .expect("test cleanup command status");
            Ok(CommandOutput {
                status,
                stdout: Vec::new(),
                stderr: if status != 0 {
                    b"cleanup failed".to_vec()
                } else {
                    Vec::new()
                },
            })
        }
    }

    fn stopped_podman_cleanup_controller(session_id: &str) -> Controller {
        let container_id = hel_targets::resource_name(session_id).unwrap();
        let volume = format!("{container_id}-workspace");
        let mut session = checkpoint_test_session(session_id);
        session.target_template_id = "podman".into();
        session.state = SessionState::Stopped;
        session.target = Some(TargetLocator::LocalPodman {
            container_id,
            workspace_storage: hel::hel_state::PodmanWorkspaceLocator::Volume { name: volume },
        });
        let mut config = HelConfig::default();
        config.targets.insert(
            "podman".into(),
            TargetTemplate::LocalPodman {
                container: ConfigContainer {
                    image: "test:latest".into(),
                    pull_policy: Default::default(),
                    platform: None,
                    cpus: None,
                    memory: None,
                    environment: BTreeMap::new(),
                    workspace_storage: hel::hel_config::PodmanWorkspaceStorage::PodmanVolume,
                },
            },
        );
        Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        }
    }

    #[test]
    fn deferred_cleanup_failure_is_visible_and_successful_retry_clears_it() {
        let session_id = "0123456789abcdef0123456789abcdef";
        let mut controller = stopped_podman_cleanup_controller(session_id);
        let persisted = RefCell::new(Vec::new());
        let failure = controller
            .cleanup_stopped_target_with(
                session_id,
                &DeferredCleanupExecutor {
                    // The cleanup command fails, then the exact-absence probe
                    // confirms that the owned target is still present.
                    statuses: RefCell::new(vec![1, 1]),
                },
                |record| {
                    persisted
                        .borrow_mut()
                        .push((record.target.is_some(), record.last_error.clone()));
                    Ok(())
                },
            )
            .unwrap_err();
        assert!(format!("{failure:#}").contains("cleanup failed"));
        assert_eq!(persisted.borrow().len(), 1);
        assert!(persisted.borrow()[0].0);
        assert!(
            persisted.borrow()[0]
                .1
                .as_deref()
                .is_some_and(|error| error.contains("deferred target cleanup failed"))
        );
        assert!(controller.state.sessions[session_id].target.is_some());
        assert!(controller.state.sessions[session_id].last_error.is_some());

        let retry_persisted = RefCell::new(Vec::new());
        controller
            .cleanup_stopped_target_with(
                session_id,
                &DeferredCleanupExecutor {
                    statuses: RefCell::new(vec![0, 0, 0]),
                },
                |record| {
                    retry_persisted
                        .borrow_mut()
                        .push((record.target.is_some(), record.last_error.clone()));
                    Ok(())
                },
            )
            .unwrap();
        assert_eq!(retry_persisted.borrow().as_slice(), &[(false, None)]);
        assert!(controller.state.sessions[session_id].target.is_none());
        assert!(controller.state.sessions[session_id].last_error.is_none());
    }

    #[test]
    fn deferred_cleanup_persistence_failure_restores_the_stopped_record() {
        let session_id = "0123456789abcdef0123456789abcdef";
        let mut controller = stopped_podman_cleanup_controller(session_id);
        let previous = controller.state.sessions[session_id].clone();
        let failure = controller
            .cleanup_stopped_target_with(
                session_id,
                &DeferredCleanupExecutor {
                    statuses: RefCell::new(vec![1, 1]),
                },
                |_| Err(anyhow::anyhow!("database unavailable")),
            )
            .unwrap_err();
        let detail = format!("{failure:#}");
        assert!(detail.contains("cleanup failed"), "{detail}");
        assert!(detail.contains("database unavailable"), "{detail}");
        assert_eq!(controller.state.sessions[session_id], previous);
    }

    #[test]
    fn target_cleanup_persists_destroying_and_rechecks_the_installed_archive() {
        struct RecordingExecutor {
            commands: RefCell<Vec<CommandSpec>>,
        }

        impl CommandExecutor for RecordingExecutor {
            fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
                self.commands.borrow_mut().push(command.clone());
                Ok(CommandOutput {
                    status: 0,
                    stdout: Vec::new(),
                    stderr: Vec::new(),
                })
            }
        }

        let directory = tempfile::tempdir().unwrap();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(directory.path(), session_id, 7);
        let mut session = checkpoint_test_session(session_id);
        session.target_template_id = "local".into();
        session.state = SessionState::Closing;
        session.target = Some(TargetLocator::LocalBare {
            worker_root: directory.path().join(session_id),
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config
            .targets
            .insert("local".into(), TargetTemplate::LocalBare);
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let executor = RecordingExecutor {
            commands: RefCell::new(Vec::new()),
        };
        let persisted = RefCell::new(Vec::new());

        controller
            .destroy_after_verified_checkpoint_with(session_id, &checkpoint, &executor, |record| {
                persisted.borrow_mut().push(record.state);
                Ok(())
            })
            .unwrap();

        assert_eq!(
            persisted.into_inner(),
            vec![SessionState::Destroying, SessionState::Stopped]
        );
        assert_eq!(executor.commands.borrow().len(), 1);
        let stopped = &controller.state.sessions[session_id];
        assert_eq!(stopped.state, SessionState::Stopped);
        assert!(stopped.target.is_none());
    }

    #[test]
    fn podman_close_persists_stopped_before_deferred_storage_cleanup() {
        struct RecordingExecutor {
            commands: RefCell<Vec<CommandSpec>>,
        }

        impl CommandExecutor for RecordingExecutor {
            fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
                self.commands.borrow_mut().push(command.clone());
                Ok(CommandOutput {
                    status: 0,
                    stdout: Vec::new(),
                    stderr: Vec::new(),
                })
            }
        }

        let directory = tempfile::tempdir().unwrap();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(directory.path(), session_id, 7);
        let container_id = hel_targets::resource_name(session_id).unwrap();
        let volume = format!("{container_id}-workspace");
        let mut session = checkpoint_test_session(session_id);
        session.target_template_id = "podman".into();
        session.state = SessionState::Closing;
        session.target = Some(TargetLocator::LocalPodman {
            container_id,
            workspace_storage: hel::hel_state::PodmanWorkspaceLocator::Volume {
                name: volume.clone(),
            },
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config.targets.insert(
            "podman".into(),
            TargetTemplate::LocalPodman {
                container: ConfigContainer {
                    image: "test:latest".into(),
                    pull_policy: Default::default(),
                    platform: None,
                    cpus: None,
                    memory: None,
                    environment: BTreeMap::new(),
                    workspace_storage: hel::hel_config::PodmanWorkspaceStorage::PodmanVolume,
                },
            },
        );
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let executor = RecordingExecutor {
            commands: RefCell::new(Vec::new()),
        };
        let persisted = RefCell::new(Vec::new());

        let deferred = controller
            .destroy_after_verified_checkpoint_with(session_id, &checkpoint, &executor, |record| {
                persisted
                    .borrow_mut()
                    .push((record.state, record.target.is_some()));
                Ok(())
            })
            .unwrap();

        assert!(deferred);
        assert_eq!(
            persisted.borrow().as_slice(),
            &[
                (SessionState::Destroying, true),
                (SessionState::Stopped, true)
            ]
        );
        let commands = executor.commands.borrow();
        assert_eq!(commands.len(), 1);
        assert!(commands[0].args[1].contains("podman stop --time 0"));
        assert!(!commands[0].args[1].contains("podman rm"));
        drop(commands);

        controller
            .cleanup_stopped_target_with(session_id, &executor, |record| {
                assert_eq!(record.state, SessionState::Stopped);
                assert!(record.target.is_none());
                Ok(())
            })
            .unwrap();

        let commands = executor.commands.borrow();
        assert_eq!(commands.len(), 4);
        assert_eq!(
            commands[1].stage,
            Some(hel_targets::ProvisionStage::RemovingContainer)
        );
        assert_eq!(
            commands[2].stage,
            Some(hel_targets::ProvisionStage::RemovingStorage)
        );
        assert_eq!(
            commands[3].stage,
            Some(hel_targets::ProvisionStage::CleaningCache)
        );
        assert!(commands[2].args.contains(&volume));
        assert!(controller.state.sessions[session_id].target.is_none());
    }
    #[test]
    fn verified_close_retires_managed_checkout_but_keeps_archive_and_branch() {
        let archive_directory = tempfile::tempdir().unwrap();
        let repository = committed_repository();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(archive_directory.path(), session_id, 7);
        let mut session = managed_worktree_session(repository.path(), session_id);
        let worktree = session.managed_worktree.clone().unwrap();
        std::fs::write(worktree.worktree_root.join("dirty.txt"), "worktree state\n").unwrap();
        session.state = SessionState::Closing;
        session.target = Some(TargetLocator::LocalBare {
            worker_root: archive_directory.path().join(session_id),
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config
            .targets
            .insert("local-bare".into(), TargetTemplate::LocalBare);
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };

        controller
            .destroy_after_verified_checkpoint_with(
                session_id,
                &checkpoint,
                &ProcessExecutor,
                |_| Ok(()),
            )
            .unwrap();

        assert!(!worktree.worktree_root.exists());
        assert!(checkpoint.archive_path.is_file());
        assert_eq!(
            test_git(
                repository.path(),
                &[
                    "show-ref",
                    "--hash",
                    &format!("refs/heads/{}", worktree.branch),
                ],
            )
            .len(),
            40
        );
        assert_eq!(
            controller.state.sessions[session_id].state,
            SessionState::Stopped
        );
    }
    #[test]
    fn force_stop_reuses_verified_archive_and_leaves_session_resumable() {
        let archive_directory = tempfile::tempdir().unwrap();
        let repository = committed_repository();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(archive_directory.path(), session_id, 7);
        let mut session = managed_worktree_session(repository.path(), session_id);
        let worktree = session.managed_worktree.clone().unwrap();
        session.state = SessionState::Running;
        session.target = Some(TargetLocator::LocalBare {
            worker_root: archive_directory.path().join(session_id),
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config
            .targets
            .insert("local-bare".into(), TargetTemplate::LocalBare);
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };

        controller
            .force_stop_with(session_id, &ProcessExecutor, |_| Ok(()))
            .unwrap();

        let stopped = &controller.state.sessions[session_id];
        assert_eq!(stopped.state, SessionState::Stopped);
        assert!(stopped.target.is_none());
        assert_eq!(stopped.checkpoint.as_ref(), Some(&checkpoint));
        assert!(checkpoint.archive_path.is_file());
        assert!(!worktree.worktree_root.exists());
        assert!(
            !test_git(
                repository.path(),
                &[
                    "show-ref",
                    "--hash",
                    &format!("refs/heads/{}", worktree.branch),
                ],
            )
            .is_empty()
        );
    }
    #[test]
    fn force_stop_without_a_recovery_archive_does_not_touch_the_target() {
        struct RecordingExecutor {
            calls: RefCell<usize>,
        }

        impl CommandExecutor for RecordingExecutor {
            fn execute(&self, _command: &CommandSpec) -> Result<CommandOutput> {
                *self.calls.borrow_mut() += 1;
                Ok(CommandOutput {
                    status: 0,
                    stdout: Vec::new(),
                    stderr: Vec::new(),
                })
            }
        }

        let directory = tempfile::tempdir().unwrap();
        let session_id = "0123456789abcdef0123456789abcdef";
        let mut session = checkpoint_test_session(session_id);
        session.state = SessionState::Running;
        session.checkpoint = None;
        session.target_template_id = "local".into();
        session.target = Some(TargetLocator::LocalBare {
            worker_root: directory.path().join(session_id),
        });
        let mut config = HelConfig::default();
        config
            .targets
            .insert("local".into(), TargetTemplate::LocalBare);
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let executor = RecordingExecutor {
            calls: RefCell::new(0),
        };

        let error = controller
            .force_stop_with(session_id, &executor, |_| Ok(()))
            .unwrap_err();

        assert!(error.to_string().contains("existing recovery archive"));
        assert_eq!(*executor.calls.borrow(), 0);
        assert_eq!(
            controller.state.sessions[session_id].state,
            SessionState::Running
        );
        assert!(controller.state.sessions[session_id].target.is_some());
    }
    #[test]
    fn destroying_retry_blocks_cleanup_when_the_archive_gate_changed() {
        struct RecordingExecutor {
            calls: RefCell<usize>,
        }

        impl CommandExecutor for RecordingExecutor {
            fn execute(&self, _command: &CommandSpec) -> Result<CommandOutput> {
                *self.calls.borrow_mut() += 1;
                Ok(CommandOutput {
                    status: 0,
                    stdout: Vec::new(),
                    stderr: Vec::new(),
                })
            }
        }

        let directory = tempfile::tempdir().unwrap();
        let repository = committed_repository();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(directory.path(), session_id, 7);
        let mut session = managed_worktree_session(repository.path(), session_id);
        let worktree = session.managed_worktree.clone().unwrap();
        session.target_template_id = "local".into();
        session.state = SessionState::Destroying;
        session.target = Some(TargetLocator::LocalBare {
            worker_root: directory.path().join(session_id),
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config
            .targets
            .insert("local".into(), TargetTemplate::LocalBare);
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let executor = RecordingExecutor {
            calls: RefCell::new(0),
        };
        let persisted = RefCell::new(Vec::new());
        std::fs::write(&checkpoint.archive_path, b"changed after checkpoint").unwrap();

        let error = controller
            .destroy_after_verified_checkpoint_with(session_id, &checkpoint, &executor, |record| {
                persisted.borrow_mut().push(record.state);
                Ok(())
            })
            .unwrap_err();

        assert!(error.to_string().contains("checkpoint SHA changed"));
        assert_eq!(*executor.calls.borrow(), 0);
        assert!(persisted.into_inner().is_empty());
        assert!(worktree.worktree_root.is_dir());
        assert_eq!(
            controller.state.sessions[session_id].state,
            SessionState::Destroying
        );
    }
    #[test]
    fn destroying_retry_finalizes_when_apple_container_is_confirmed_absent() {
        struct AlreadyRemovedExecutor {
            commands: RefCell<Vec<CommandSpec>>,
        }

        impl CommandExecutor for AlreadyRemovedExecutor {
            fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
                self.commands.borrow_mut().push(command.clone());
                if command.program == "sh"
                    && command
                        .args
                        .get(1)
                        .is_some_and(|script| script.contains("container rm --force"))
                {
                    Ok(CommandOutput {
                        status: 1,
                        stdout: Vec::new(),
                        stderr: b"container not found".to_vec(),
                    })
                } else {
                    Ok(CommandOutput {
                        status: 0,
                        stdout: Vec::new(),
                        stderr: Vec::new(),
                    })
                }
            }
        }

        let directory = tempfile::tempdir().unwrap();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(directory.path(), session_id, 7);
        let mut session = checkpoint_test_session(session_id);
        session.target_template_id = "apple".into();
        session.state = SessionState::Destroying;
        session.target = Some(TargetLocator::AppleContainer {
            container_id: hel_targets::resource_name(session_id).unwrap(),
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config.targets.insert(
            "apple".into(),
            TargetTemplate::AppleContainer {
                container: ConfigContainer {
                    image: "test:latest".into(),
                    pull_policy: Default::default(),
                    platform: None,
                    cpus: None,
                    memory: None,
                    environment: BTreeMap::new(),
                    workspace_storage: Default::default(),
                },
            },
        );
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let executor = AlreadyRemovedExecutor {
            commands: RefCell::new(Vec::new()),
        };
        let persisted = RefCell::new(Vec::new());

        controller
            .destroy_after_verified_checkpoint_with(session_id, &checkpoint, &executor, |record| {
                persisted.borrow_mut().push(record.state);
                Ok(())
            })
            .unwrap();

        let commands = executor.commands.borrow();
        assert_eq!(commands.len(), 2);
        assert_eq!(commands[0].program, "sh");
        assert!(commands[0].args[1].contains("container rm --force"));
        assert!(commands[0].args[1].contains(".cache/mjolnir/git/sessions"));
        assert_eq!(commands[1].args, ["list", "--all", "--quiet"]);
        assert_eq!(persisted.into_inner(), vec![SessionState::Stopped]);
        assert_eq!(
            controller.state.sessions[session_id].state,
            SessionState::Stopped
        );
    }

    #[test]
    fn interrupted_close_error_preserves_destroying_phase() {
        let session_id = "0123456789abcdef0123456789abcdef";
        let mut session = checkpoint_test_session(session_id);
        session.state = SessionState::Destroying;

        apply_interrupted_close_error(
            &mut session,
            &anyhow::anyhow!("podman unavailable"),
            "2026-08-14T12:00:00Z",
        );

        assert_eq!(session.state, SessionState::Destroying);
        assert_eq!(session.updated_at, "2026-08-14T12:00:00Z");
        assert!(
            session
                .last_error
                .as_deref()
                .is_some_and(|error| error.contains("cleanup is safely retryable"))
        );
    }

    struct FailingExecutor;

    impl CommandExecutor for FailingExecutor {
        fn execute(&self, _command: &CommandSpec) -> Result<CommandOutput> {
            Ok(CommandOutput {
                status: 1,
                stdout: Vec::new(),
                stderr: b"teardown unavailable".to_vec(),
            })
        }
    }

    fn branch_exists(repository: &std::path::Path, branch: &str) -> bool {
        std::process::Command::new("git")
            .arg("-C")
            .arg(repository)
            .args(["show-ref", "--verify", "--quiet"])
            .arg(format!("refs/heads/{branch}"))
            .output()
            .unwrap()
            .status
            .success()
    }

    #[test]
    fn force_destroy_from_running_removes_target_worktree_branch_and_archive() {
        let directory = tempfile::tempdir().unwrap();
        let repository = committed_repository();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(directory.path(), session_id, 7);
        let worker_root = directory.path().join(session_id);
        std::fs::create_dir_all(&worker_root).unwrap();
        let mut session = managed_worktree_session(repository.path(), session_id);
        session.state = SessionState::Running;
        session.target_template_id = "local".into();
        session.target = Some(TargetLocator::LocalBare {
            worker_root: worker_root.clone(),
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config
            .targets
            .insert("local".into(), TargetTemplate::LocalBare);
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let deleted = RefCell::new(Vec::new());

        controller
            .force_destroy_session_with(session_id, &ProcessExecutor, |id| {
                deleted.borrow_mut().push(id.to_owned());
                Ok(())
            })
            .unwrap();

        assert!(!worker_root.exists(), "local target must be removed");
        let worktree_root = repository.path().join(".mj/worktrees").join(session_id);
        assert!(!worktree_root.exists(), "managed worktree must be removed");
        assert!(
            !branch_exists(repository.path(), &format!("mj/{session_id}")),
            "generated branch must be removed"
        );
        assert!(!checkpoint.archive_path.exists(), "archive must be removed");
        assert!(!controller.state.sessions.contains_key(session_id));
        assert_eq!(deleted.into_inner(), vec![session_id.to_owned()]);
    }

    #[test]
    fn force_destroy_without_a_target_or_archive_still_removes_the_record() {
        let session_id = "0123456789abcdef0123456789abcdef";
        let mut session = checkpoint_test_session(session_id);
        session.state = SessionState::Provisioning;
        session.target = None;
        session.checkpoint = None;
        let mut controller = Controller {
            config: HelConfig::default(),
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let deleted = RefCell::new(Vec::new());

        controller
            .force_destroy_session_with(session_id, &ProcessExecutor, |id| {
                deleted.borrow_mut().push(id.to_owned());
                Ok(())
            })
            .unwrap();

        assert!(!controller.state.sessions.contains_key(session_id));
        assert_eq!(deleted.into_inner(), vec![session_id.to_owned()]);
    }

    #[test]
    fn force_destroy_aborts_and_keeps_the_record_when_the_target_survives() {
        let directory = tempfile::tempdir().unwrap();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(directory.path(), session_id, 7);
        let mut session = checkpoint_test_session(session_id);
        session.target_template_id = "local".into();
        session.state = SessionState::Running;
        session.target = Some(TargetLocator::LocalBare {
            worker_root: directory.path().join(session_id),
        });
        session.checkpoint = Some(checkpoint.clone());
        let mut config = HelConfig::default();
        config
            .targets
            .insert("local".into(), TargetTemplate::LocalBare);
        let mut controller = Controller {
            config,
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };
        let deleted = RefCell::new(Vec::new());

        let error = controller
            .force_destroy_session_with(session_id, &FailingExecutor, |id| {
                deleted.borrow_mut().push(id.to_owned());
                Ok(())
            })
            .unwrap_err();

        assert!(
            error.to_string().contains("teardown unavailable"),
            "{error:#}"
        );
        assert!(
            controller.state.sessions.contains_key(session_id),
            "a surviving target must keep the record for a retry"
        );
        assert!(
            checkpoint.archive_path.exists(),
            "a surviving target must keep the recovery archive"
        );
        assert!(deleted.into_inner().is_empty());
    }

    #[test]
    fn force_destroy_tolerates_a_missing_archive() {
        let directory = tempfile::tempdir().unwrap();
        let session_id = "0123456789abcdef0123456789abcdef";
        let checkpoint = write_checkpoint_gate_archive(directory.path(), session_id, 7);
        std::fs::remove_file(&checkpoint.archive_path).unwrap();
        let mut session = checkpoint_test_session(session_id);
        session.state = SessionState::Error;
        session.checkpoint = Some(checkpoint);
        let mut controller = Controller {
            config: HelConfig::default(),
            state: HelState {
                sessions: BTreeMap::from([(session_id.into(), session)]),
                ..HelState::default()
            },
        };

        controller
            .force_destroy_session_with(session_id, &ProcessExecutor, |_| Ok(()))
            .unwrap();

        assert!(!controller.state.sessions.contains_key(session_id));
    }
}