mj_controller/

controller.rs

//! Controller-side lifecycle transitions and canonical-to-backend conversion.

mod backend;
mod checkpoint;
mod git_cache;
mod lifecycle;
pub mod move_session;
mod network_git;
pub mod profile_config;
mod provisioning;
mod readiness;
mod recovery_scan;
mod resume;
mod reviewer;
mod subagents;
#[cfg(test)]
mod test_support;
pub mod update;
mod worker_binary;
mod worker_restart;
mod worktree;

use std::collections::{BTreeMap, BTreeSet};
use std::fs::{self, File, OpenOptions};
use std::path::{Path, PathBuf};

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

use mj_core::config::{
    Config, ProjectBundle, ProjectRepository, SshConnection, TargetTemplate, atomic_write,
    container_size_host, data_dir, is_bare_project_target, mount_history_host,
};

use crate::import::{
    RepositoryIdentity, bundle_matches, configured_bundle_for_local, configured_bundle_for_origin,
    setup_style_id,
};
use crate::setup::github_repository_from_origin;

const CONFIG_RENAME_JOURNAL: &str = "config-rename.json";

#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
enum ConfigRenameKind {
    Profile,
    Target,
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct ConfigRenameJournal {
    kind: ConfigRenameKind,
    old_id: String,
    new_id: String,
}
use mj_core::state::{
    HostContainerSize, SessionRecord, SessionResourceAllocation, SessionState, State,
    new_session_id, normalize_session_title,
};

use crate::targets::{
    self, AdditionalMount, CommandExecutor, CommandOutput, CommandSpec, SshTarget,
};

pub(crate) use backend::controller_github_token;
pub use backend::image_refresh_plan;
use backend::validate_resource_allocation;
use provisioning::apply_failed_new_session_rollback;
pub(crate) use worker_binary::refresh_remote_worker_binary_if_stale;
pub(crate) use worktree::path_exists_on_managed_target;

pub use checkpoint::{
    CheckpointArtifact, CheckpointDeferred, IdleWorkspaceLease, SessionExportLayout,
    checkpoint_was_deferred, reconcile_managed_checkpoint_archives,
};
pub use recovery_scan::{RecoveryCandidate, RecoveryScan};
pub use resume::{
    ResumeRepositorySourceMismatch, ResumeRepositorySourcePreflight, ResumeRepositorySourceReceipt,
    raw_conversion_preview_for,
};
pub use reviewer::reviewer_stager;
pub use subagents::RegisterSubagentRequest;
pub use worker_binary::{
    WorkerBinaryAvailability, pin_worker_binary_sources, worker_binary_prerequisite_for_arch,
};
pub use worker_restart::WorkerUpgradeOutcome;
pub use worktree::{ResumePlan, local_project_repository, resume_compatibility};

pub struct Controller {
    pub config: Config,
    pub state: State,
}

/// Machine-wide advisory lock for one controller data store. This prevents a
/// dashboard, server, or CLI lifecycle command from concurrently acting as a
/// second controller against the same SQLite state and relay sessions.
#[derive(Debug)]
pub struct ControllerStoreGuard {
    file: File,
}

impl ControllerStoreGuard {
    pub fn acquire() -> Result<Self> {
        let directory = data_dir();
        Self::acquire_at(&directory)
    }

    fn acquire_at(directory: &Path) -> Result<Self> {
        Self::try_acquire_at(directory)?.with_context(|| {
            format!(
                "another Mjolnir controller is already using {}; stop it before starting this command",
                directory.display()
            )
        })
    }

    /// Probe exclusivity without treating an owner that is still exiting as an error.
    pub fn try_acquire() -> Result<Option<Self>> {
        Self::try_acquire_at(&data_dir())
    }

    fn try_acquire_at(directory: &Path) -> Result<Option<Self>> {
        std::fs::create_dir_all(directory)
            .with_context(|| format!("create controller data directory {}", directory.display()))?;
        let path = directory.join("controller.lock");
        let mut options = OpenOptions::new();
        options.create(true).read(true).write(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        let file = options
            .open(&path)
            .with_context(|| format!("open controller lock {}", path.display()))?;
        match file.try_lock() {
            Ok(()) => {}
            Err(std::fs::TryLockError::WouldBlock) => return Ok(None),
            Err(std::fs::TryLockError::Error(error)) => {
                return Err(error)
                    .with_context(|| format!("lock controller store {}", directory.display()));
            }
        }
        Ok(Some(Self { file }))
    }

    /// Start the sole production SQLite writer after controller exclusivity
    /// has been established by this guard.
    pub fn start_database_writer(&self) -> Result<crate::database::DatabaseWriterOwner> {
        crate::database::start_database_writer()
    }
}

impl Drop for ControllerStoreGuard {
    fn drop(&mut self) {
        // Make release explicit. `File` also unlocks on close, but an explicit
        // unlock keeps same-process handoff deterministic across platforms.
        let _ = self.file.unlock();
    }
}

/// The durable result of creating a quick bundle. The returned config is the
/// same fresh config that was written, allowing a serving projection to publish
/// the new bundle before acknowledging the request that created it.
#[derive(Debug)]
pub struct QuickBundleCreation {
    pub config: Config,
    pub bundle_id: String,
}

/// Failure stages exposed to a viewer request without exposing the underlying
/// filesystem/configuration error. The detailed error remains available to
/// the caller for logs and terminal notices.
#[derive(Debug)]
pub enum QuickBundleFailure {
    InvalidSource(anyhow::Error),
    Persistence(anyhow::Error),
}

impl std::fmt::Display for QuickBundleFailure {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::InvalidSource(error) => write!(formatter, "invalid repository source: {error}"),
            Self::Persistence(error) => write!(formatter, "persist quick bundle: {error}"),
        }
    }
}

impl std::error::Error for QuickBundleFailure {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::InvalidSource(error) | Self::Persistence(error) => Some(error.root_cause()),
        }
    }
}

/// Create a quick bundle from a local repository or GitHub source and persist
/// it as one serialized fresh-config transaction. Identical sources reuse the
/// existing configured bundle, matching the terminal's behavior. The returned
/// config is the same fresh config that was written, allowing a serving
/// projection to publish the new bundle before acknowledging the request.
pub fn create_quick_bundle(
    source: &str,
) -> std::result::Result<QuickBundleCreation, QuickBundleFailure> {
    let (config, bundle_id) = Config::update(|config| {
        create_quick_bundle_in_config(config, source)
            .map_err(|error| anyhow::Error::new(QuickBundleFailure::InvalidSource(error)))
    })
    .map_err(|error| {
        error
            .downcast::<QuickBundleFailure>()
            .unwrap_or_else(QuickBundleFailure::Persistence)
    })?;
    Ok(QuickBundleCreation { config, bundle_id })
}

/// Add a quick bundle to an already-loaded config. The helper still performs
/// the local repository canonicalization/GitHub-source parsing, but callers
/// that persist a config should use [`create_quick_bundle`] so concurrent saves
/// cannot clobber one another.
pub fn create_quick_bundle_in_config(config: &mut Config, source: &str) -> Result<String> {
    let source = interpret_repository_source(source)?;
    let existing = match &source.kind {
        RepositorySourceKind::Local(root) => configured_bundle_for_local(config, root),
        RepositorySourceKind::Github(repository) => {
            configured_bundle_for_origin(config, repository)
        }
    };
    if let Some(existing) = existing {
        return Ok(existing);
    }
    let repository_id = setup_style_id(&source.name);
    let mut bundle_id = repository_id.clone();
    for suffix in 2_u32.. {
        if !config.bundles.contains_key(&bundle_id) {
            break;
        }
        bundle_id = format!("{repository_id}-{suffix}");
    }
    config.bundles.insert(
        bundle_id.clone(),
        ProjectBundle {
            primary_repo: repository_id.clone(),
            repositories: vec![source.into_project_repository(repository_id.clone())],
        },
    );
    config.validate()?;
    Ok(bundle_id)
}

/// Create one bundle from one or more local repositories or GitHub sources.
///
/// All sources are interpreted and checked before the config transaction can
/// write anything. An existing bundle is reused only when its repository set
/// and primary repository exactly match the request; this keeps selecting one
/// repository from a larger bundle from silently changing the wizard's choice.
pub fn create_bundle_from_sources(
    sources: &[String],
) -> std::result::Result<QuickBundleCreation, QuickBundleFailure> {
    let (config, bundle_id) = Config::update(|config| {
        create_bundle_from_sources_in_config(config, sources)
            .map_err(|error| anyhow::Error::new(QuickBundleFailure::InvalidSource(error)))
    })
    .map_err(|error| {
        error
            .downcast::<QuickBundleFailure>()
            .unwrap_or_else(QuickBundleFailure::Persistence)
    })?;
    Ok(QuickBundleCreation { config, bundle_id })
}

/// Add a bundle for all `sources` to an already-loaded config. The source
/// interpretation is shared with the persisted [`create_bundle_from_sources`]
/// entry point and the legacy quick-bundle helper.
pub fn create_bundle_from_sources_in_config(
    config: &mut Config,
    sources: &[String],
) -> Result<String> {
    let sources = sources
        .iter()
        .map(|source| interpret_repository_source(source))
        .collect::<Result<Vec<_>>>()?;
    if sources.is_empty() {
        bail!("at least one repository source is required");
    }

    let mut identities = BTreeSet::new();
    for source in &sources {
        if !identities.insert(source.identity()) {
            bail!("duplicate repository source {:?}", source.display_name);
        }
    }

    if let Some(existing) = exact_configured_bundle(config, &sources) {
        return Ok(existing);
    }

    // Build and validate a candidate before replacing the caller's config, so
    // a later validation error cannot leave an in-memory partial mutation.
    let mut updated = config.clone();
    let mut used_repository_ids = BTreeSet::new();
    let mut repositories = Vec::with_capacity(sources.len());
    for source in sources {
        let base = setup_style_id(&source.name);
        let repository_id = unique_id(&base, |candidate| used_repository_ids.contains(candidate));
        used_repository_ids.insert(repository_id.clone());
        repositories.push(source.into_project_repository(repository_id));
    }
    let primary_repo = repositories
        .first()
        .map(|repository| repository.id.clone())
        .context("at least one repository source is required")?;
    let bundle_id = unique_id(&primary_repo, |candidate| {
        updated.bundles.contains_key(candidate)
    });
    updated.bundles.insert(
        bundle_id.clone(),
        ProjectBundle {
            primary_repo,
            repositories,
        },
    );
    updated.validate()?;
    *config = updated;
    Ok(bundle_id)
}

#[derive(Debug, Clone)]
enum RepositorySourceKind {
    Github(crate::setup::GithubRepository),
    Local(PathBuf),
}

#[derive(Debug, Clone)]
struct InterpretedRepositorySource {
    display_name: String,
    name: String,
    kind: RepositorySourceKind,
}

impl InterpretedRepositorySource {
    fn identity(&self) -> RepositoryIdentity {
        match &self.kind {
            RepositorySourceKind::Github(repository) => RepositoryIdentity::Github(
                repository.owner.to_ascii_lowercase(),
                repository.repository.to_ascii_lowercase(),
            ),
            RepositorySourceKind::Local(root) => RepositoryIdentity::Local(root.clone()),
        }
    }

    fn into_project_repository(self, id: String) -> ProjectRepository {
        let (github, local) = match self.kind {
            RepositorySourceKind::Github(repository) => (
                Some(format!("{}/{}", repository.owner, repository.repository)),
                None,
            ),
            RepositorySourceKind::Local(root) => (None, Some(root)),
        };
        ProjectRepository {
            id: id.clone(),
            github,
            local,
            destination: PathBuf::from(id),
            git_ref: None,
        }
    }
}

/// Interpret a source once, including local Git canonicalization and GitHub
/// parsing, so all creation paths use exactly the same source semantics.
fn interpret_repository_source(source: &str) -> Result<InterpretedRepositorySource> {
    let source = source.trim();
    if source.is_empty() {
        bail!("repository source cannot be empty");
    }
    let expanded = mj_core::path_input::expand_local(Path::new(source))?;
    let candidate = expanded.as_path();
    if candidate.exists() {
        let root = mj_core::local_git::canonical_repository(candidate)?;
        let name = root
            .file_name()
            .and_then(|name| name.to_str())
            .context("local repository has no usable directory name")?
            .to_owned();
        return Ok(InterpretedRepositorySource {
            display_name: source.to_owned(),
            name,
            kind: RepositorySourceKind::Local(root),
        });
    }
    if candidate.is_absolute() || source.starts_with('.') || source.starts_with('~') {
        bail!("local repository path {source:?} does not exist");
    }
    let repository = github_repository_from_origin(source).context(format!(
        "{source:?} is not a GitHub owner/repository or URL"
    ))?;
    Ok(InterpretedRepositorySource {
        display_name: source.to_owned(),
        name: repository.repository.clone(),
        kind: RepositorySourceKind::Github(repository),
    })
}

fn exact_configured_bundle(
    config: &Config,
    requested: &[InterpretedRepositorySource],
) -> Option<String> {
    let requested_identities = requested
        .iter()
        .map(InterpretedRepositorySource::identity)
        .collect::<BTreeSet<_>>();
    let primary = requested.first()?.identity();
    config.bundles.iter().find_map(|(id, bundle)| {
        if bundle.repositories.len() != requested.len()
            || bundle
                .repositories
                .iter()
                .any(|repository| repository.git_ref.is_some())
        {
            return None;
        }
        bundle_matches(bundle, &requested_identities, &primary).then(|| id.clone())
    })
}

fn unique_id(base: &str, mut is_used: impl FnMut(&str) -> bool) -> String {
    if !is_used(base) {
        return base.to_owned();
    }
    for suffix in 2_u32.. {
        let suffix = format!("-{suffix}");
        let prefix_len = 64usize.saturating_sub(suffix.len());
        let prefix = base.chars().take(prefix_len).collect::<String>();
        let candidate = format!("{prefix}{suffix}");
        if !is_used(&candidate) {
            return candidate;
        }
    }
    unreachable!("u32 repository/bundle id suffixes exhausted")
}

pub struct SessionLaunchOptions {
    pub create_managed_worktree: Option<bool>,
    pub mjolnir_subagents: Option<bool>,
    pub initial_prompt: Option<String>,
    pub workspace_id: String,
    pub additional_mounts: Vec<AdditionalMount>,
    pub allow_dirty_local: bool,
    pub resource_allocation: Option<SessionResourceAllocation>,
    pub project_directory: Option<PathBuf>,
    pub session_title_override: Option<String>,
}

pub struct SessionResumeOptions {
    pub additional_mounts: Option<Vec<AdditionalMount>>,
    pub resource_allocation: Option<SessionResourceAllocation>,
    pub discard_queue: bool,
}

fn selected_host_container_size(
    template: &TargetTemplate,
    allocation: Option<&SessionResourceAllocation>,
) -> Option<(String, HostContainerSize)> {
    let host = container_size_host(template)?;
    let SessionResourceAllocation::Container { cpus, memory_bytes } = allocation? else {
        return None;
    };
    Some((
        host.to_owned(),
        HostContainerSize {
            cpus: *cpus,
            memory_bytes: *memory_bytes,
        },
    ))
}

impl Controller {
    pub fn load() -> Result<Self> {
        let config = Config::load()?;
        let state = crate::database::load_state_migrating()?;
        // Missing session dependencies must not lock users out of the tools
        // needed to repair them. Operations validate the session they act on.
        state.validate()?;
        for session in state.sessions.values() {
            if let Some(issue) = session.configuration_issue(&config) {
                tracing::warn!(session_id = %session.id, "{issue}");
            }
        }
        Ok(Self { config, state })
    }

    pub fn reload(&mut self) -> Result<()> {
        *self = Self::load()?;
        Ok(())
    }

    fn persist_session_state(&self, session_id: &str) -> Result<()> {
        match self.state.sessions.get(session_id) {
            Some(session) => crate::database::save_lifecycle_session(session),
            None => crate::database::delete_session(session_id),
        }
    }

    fn persist_session_transition_or_restore(
        &mut self,
        session_id: &str,
        previous: &SessionRecord,
        context: &'static str,
    ) -> Result<()> {
        persist_session_record_transition_or_restore(
            &mut self.state,
            session_id,
            previous,
            context,
            &crate::database::save_lifecycle_session,
        )
    }

    fn restore_prior_session_after_persistence_failure(
        &mut self,
        session_id: &str,
        previous: &SessionRecord,
        primary: anyhow::Error,
    ) -> anyhow::Error {
        restore_session_after_persistence_failure(
            &mut self.state,
            session_id,
            previous,
            primary,
            crate::database::save_lifecycle_session,
        )
    }

    /// Resolve an entered path on its owning host. Call only from background work.
    pub fn resolve_input_path(
        &self,
        target_id: &str,
        path: &Path,
        executor: &impl CommandExecutor,
    ) -> Result<PathBuf> {
        let target = self
            .config
            .targets
            .get(target_id)
            .context("Unknown path target")?;
        resolve_target_input_path(target, path, executor)
    }

    /// Complete a mount source on the container engine host, preserving ~/ while editing.
    pub fn complete_mount_source(
        &self,
        target_id: &str,
        prefix: &str,
        executor: &impl CommandExecutor,
    ) -> Result<Vec<String>> {
        let target = self
            .config
            .targets
            .get(target_id)
            .with_context(|| format!("unknown target template {target_id:?}"))?;
        let home = if mj_core::path_input::needs_home(Path::new(prefix))? {
            Some(resolve_target_input_path(target, Path::new("~"), executor)?)
        } else {
            None
        };
        let expanded = mj_core::path_input::expand_home(Path::new(prefix), home.as_deref())?;
        let mut lookup = expanded.to_string_lossy().into_owned();
        // Completion is a text protocol: a trailing separator requests children.
        if prefix.ends_with('/') && !lookup.ends_with('/') {
            lookup.push('/');
        }
        let candidates = match target {
            TargetTemplate::LocalPodman { .. }
            | TargetTemplate::LocalDocker { .. }
            | TargetTemplate::AppleContainer { .. }
            | TargetTemplate::AwsEc2 { .. } => targets::local_directory_completions(&lookup),
            TargetTemplate::SshPodman { ssh, .. } | TargetTemplate::SshDocker { ssh, .. } => {
                targets::ssh_directory_completions(&backend_ssh(ssh), &lookup, executor)?
            }
            TargetTemplate::LocalBare | TargetTemplate::SshBare { .. } => {
                bail!("resource path completion is unsupported for bare targets")
            }
        };
        candidates
            .into_iter()
            .map(|candidate| {
                let Some(home) = &home else {
                    return Ok(candidate);
                };
                let suffix = Path::new(&candidate)
                    .strip_prefix(home)
                    .context("Completed path is outside the requested home")?;
                let mut value = Path::new("~").join(suffix).to_string_lossy().into_owned();
                if candidate.ends_with('/') && !value.ends_with('/') {
                    value.push('/');
                }
                Ok(value)
            })
            .collect()
    }

    /// Verify a mount source on the host where Mjolnir will consume it, and report
    /// the filesystem reason it must be attached read-only, if there is one.
    ///
    /// The probe runs in the same round trip as the existence check so the
    /// editor learns both answers without a second wait. A probe that cannot
    /// answer reports no reason: provisioning decides that authoritatively.
    pub fn validate_mount_source(
        &self,
        target_id: &str,
        source: &Path,
        executor: &impl CommandExecutor,
    ) -> Result<Option<String>> {
        let target = self
            .config
            .targets
            .get(target_id)
            .with_context(|| format!("unknown target template {target_id:?}"))?;
        let exists = match target {
            TargetTemplate::LocalPodman { .. }
            | TargetTemplate::LocalDocker { .. }
            | TargetTemplate::AppleContainer { .. }
            | TargetTemplate::AwsEc2 { .. } => std::fs::metadata(source)
                .map(|metadata| metadata.is_dir())
                .or_else(|error| {
                    if error.kind() == std::io::ErrorKind::NotFound {
                        Ok(false)
                    } else {
                        Err(error)
                    }
                })
                .with_context(|| format!("inspect resource source {}", source.display()))?,
            TargetTemplate::SshPodman { ssh, .. } | TargetTemplate::SshDocker { ssh, .. } => {
                targets::ssh_directory_exists(&backend_ssh(ssh), source, executor)?
            }
            TargetTemplate::LocalBare | TargetTemplate::SshBare { .. } => {
                bail!("resource attachments are unsupported for bare targets")
            }
        };
        ensure!(
            exists,
            "source path {} does not exist or is not a directory",
            source.display()
        );
        Ok(self.forced_read_only_reason(target, source, executor))
    }

    /// The `filesystem (reason)` label for a source the runtime cannot overlay.
    fn forced_read_only_reason(
        &self,
        target: &TargetTemplate,
        source: &Path,
        executor: &impl CommandExecutor,
    ) -> Option<String> {
        let ssh = match target {
            TargetTemplate::LocalPodman { .. } | TargetTemplate::LocalDocker { .. } => None,
            TargetTemplate::SshPodman { ssh, .. } | TargetTemplate::SshDocker { ssh, .. } => {
                Some(backend_ssh(ssh))
            }
            // Apple Container already mounts read-only, and EC2 copies instead
            // of mounting, so neither has an overlay to lose.
            _ => return None,
        };
        let filesystem = targets::probe_filesystem_types(
            ssh.as_ref(),
            std::slice::from_ref(&source.to_path_buf()),
            executor,
        )
        .map_err(|error| {
            tracing::debug!(
                source = %source.display(),
                error = format!("{error:#}"),
                "could not probe the filesystem under a mount source"
            );
        })
        .ok()?
        .pop()?;
        let reason = targets::overlay_unsupported_filesystem(&filesystem)?;
        Some(format!("{filesystem} ({reason})"))
    }

    fn fail_new_session_with_cleanup(
        &mut self,
        session_id: &str,
        error: anyhow::Error,
        executor: &impl CommandExecutor,
    ) -> Result<anyhow::Error> {
        let original = provisioning::note_new_session_launch_failure(session_id, &error);
        let cleanup_error = self
            .cleanup_new_session_worktree_after_failure(session_id, executor)
            .err()
            .map(|cleanup_error| format!("{cleanup_error:#}"));
        if let Some(cleanup_error) = &cleanup_error {
            tracing::warn!(
                session_id,
                error = %cleanup_error,
                "new-session worktree rollback reported a cleanup failure"
            );
        }
        let failure = apply_failed_new_session_rollback(
            &mut self.state,
            session_id,
            &original,
            cleanup_error,
        );
        self.persist_session_state(session_id)?;
        Ok(failure)
    }

    pub fn register_session_with_resources(
        &mut self,
        profile_id: &str,
        bundle_id: &str,
        target_id: &str,
        title: impl Into<String>,
        options: SessionLaunchOptions,
    ) -> Result<String> {
        let SessionLaunchOptions {
            create_managed_worktree,
            mjolnir_subagents,
            initial_prompt,
            workspace_id,
            additional_mounts,
            allow_dirty_local: _,
            resource_allocation,
            project_directory,
            session_title_override,
        } = options;
        let session_title_override = match session_title_override {
            Some(title) => {
                Some(normalize_session_title(&title).context("session name cannot be empty")?)
            }
            None => None,
        };
        let profile = self
            .config
            .profiles
            .get(profile_id)
            .with_context(|| format!("unknown profile {profile_id:?}"))?;
        ensure!(profile.enabled, "profile {profile_id:?} is disabled");
        let template = self
            .config
            .targets
            .get(target_id)
            .with_context(|| format!("unknown target template {target_id:?}"))?;
        if create_managed_worktree == Some(true) && !is_bare_project_target(template) {
            bail!("managed worktree creation requires a bare Git project");
        }
        if project_directory.is_some() != is_bare_project_target(template) {
            bail!("raw project directories require a bare target, and bare targets require one");
        }
        if let Some(path) = &project_directory
            && (!path.is_absolute()
                || path
                    .components()
                    .any(|part| part == std::path::Component::ParentDir))
        {
            bail!("bare project directory must be an absolute safe path");
        }
        let bundle = project_directory
            .is_none()
            .then(|| self.config.bundles.get(bundle_id))
            .flatten();
        if project_directory.is_none() && bundle.is_none() {
            bail!("unknown bundle {bundle_id:?}");
        }
        if matches!(
            profile.kind,
            mj_core::config::HarnessKind::Deepseek | mj_core::config::HarnessKind::Muse
        ) && (!additional_mounts.is_empty()
            || bundle.is_some_and(|bundle| bundle.repositories.len() > 1))
        {
            bail!(
                "{} ACP supports one workspace root; use a single-repository bundle without attached directories",
                profile.kind.display_name()
            );
        }
        if let Some(bundle) = bundle {
            for repository in &bundle.repositories {
                mj_core::remote_git::resolve_repository(
                    repository,
                    &targets::CancellableProcessExecutor::with_timeout(
                        std::time::Duration::from_secs(15),
                    ),
                )
                .with_context(|| format!("repository {:?}", repository.id))?;
            }
        }
        validate_resource_allocation(template, resource_allocation.as_ref())?;
        let selected_container_size =
            selected_host_container_size(template, resource_allocation.as_ref());
        if !additional_mounts.is_empty() && mount_history_host(template).is_none() {
            bail!("attached resources are unsupported for this target");
        }
        targets::validate_additional_mounts(&additional_mounts)?;
        let id = new_session_id()?;
        let now = now();
        let record = SessionRecord {
            create_managed_worktree,
            mjolnir_subagents,
            archived: false,
            container_cpus: None,
            container_memory: None,
            id: id.clone(),
            workspace_id,
            title: title.into(),
            harness_kind: profile.kind,
            last_profile: profile_id.to_string(),
            bundle_id: bundle_id.to_string(),
            project_directory,
            managed_worktree: None,
            target_template_id: target_id.to_string(),
            resource_allocation,
            additional_mounts: additional_mounts.clone(),
            state: SessionState::Provisioning,
            target: None,
            native_session_id: None,
            acp_session_title: None,
            session_title_override,
            created_at: now.clone(),
            updated_at: now,
            viewed_through_event_ordinal: 0,
            draft_input: initial_prompt.unwrap_or_default(),
            last_error: None,
            last_checkpoint_error: None,
            checkpoint: None,
        };
        // Creation authors the whole record, so it writes the whole row. The
        // record reaches memory only once it is durable: a session this process
        // alone knows about is one the database can never resume or clean up.
        if let Some((host, size)) = selected_container_size.as_ref() {
            crate::database::save_session_with_container_size(&record, host, *size)?;
        } else {
            crate::database::save_session(&record)?;
        }
        self.state.sessions.insert(id.clone(), record);
        if let Some((host, size)) = selected_container_size {
            self.state.remember_container_size(&host, size);
        }
        if let Some(host) = mount_history_host(template) {
            // Mount history only seeds the attach dialog's suggestions. The
            // session row is already committed, so a failed suggestion write is
            // reported rather than turned into a failed registration.
            match crate::database::remember_mount_sources(host, &additional_mounts) {
                Ok(()) => self.state.remember_mount_sources(host, &additional_mounts),
                Err(error) => tracing::warn!(
                    session_id = id,
                    error = format!("{error:#}"),
                    "could not remember the attached resource directories for later suggestions"
                ),
            }
        }
        Ok(id)
    }

    pub fn rename_session(&mut self, session_id: &str, title: &str) -> Result<String> {
        let title = normalize_session_title(title).context("session name cannot be empty")?;
        ensure!(
            self.state.sessions.contains_key(session_id),
            "unknown session {session_id}"
        );
        let updated_at = now();
        crate::database::set_session_title_override(session_id, &title, &updated_at)?;
        let record = self
            .state
            .sessions
            .get_mut(session_id)
            .expect("session was checked before updating its title");
        record.session_title_override = Some(title.clone());
        record.updated_at = updated_at;
        Ok(title)
    }

    pub fn rename_profile_id(&mut self, old_id: &str, new_id: &str) -> Result<()> {
        mj_core::config::validate_id("profile", new_id)?;
        if old_id == new_id {
            ensure!(
                self.config.profiles.contains_key(old_id),
                "unknown profile {old_id:?}"
            );
            return Ok(());
        }
        let journal = ConfigRenameJournal {
            kind: ConfigRenameKind::Profile,
            old_id: old_id.to_owned(),
            new_id: new_id.to_owned(),
        };
        write_config_rename_journal(&journal)?;
        let (config, ()) = match Config::update(|config| {
            ensure!(
                config.profiles.contains_key(old_id),
                "unknown profile {old_id:?}"
            );
            ensure!(
                !config.profiles.contains_key(new_id),
                "profile {new_id:?} already exists"
            );
            let profile = config
                .profiles
                .remove(old_id)
                .expect("profile was checked in the transaction");
            config.profiles.insert(new_id.to_owned(), profile);
            Ok(())
        }) {
            Ok(result) => result,
            Err(error) => {
                remove_config_rename_journal()
                    .context("remove profile rename journal after config save failed")?;
                return Err(error).context("save renamed profile configuration");
            }
        };
        self.config = config;
        mj_core::test_hooks::reach_test_hook("config_replacement_before_reference_migration")?;
        if let Err(error) = crate::database::rename_profile_references(old_id, new_id) {
            let restore = Config::update(|config| {
                let profile = config
                    .profiles
                    .remove(new_id)
                    .with_context(|| format!("renamed profile {new_id:?} is missing"))?;
                ensure!(
                    !config.profiles.contains_key(old_id),
                    "cannot restore profile rename: both {old_id:?} and {new_id:?} exist"
                );
                config.profiles.insert(old_id.to_owned(), profile);
                Ok(())
            });
            let restored = match restore {
                Ok((config, ())) => config,
                Err(restore_error) => {
                    return Err(error).context(format!(
                        "rename profile references; additionally failed to restore config: {restore_error:#}"
                    ));
                }
            };
            self.config = restored;
            if let Err(restore_error) = remove_config_rename_journal() {
                return Err(error).context(format!(
                    "rename profile references; additionally failed to remove rename journal: {restore_error:#}"
                ));
            }
            return Err(error).context("rename profile references");
        }
        for session in self.state.sessions.values_mut() {
            if session.last_profile == old_id {
                session.last_profile = new_id.to_owned();
            }
        }
        remove_config_rename_journal()?;
        Ok(())
    }

    pub fn rename_target_id(&mut self, old_id: &str, new_id: &str) -> Result<()> {
        mj_core::config::validate_id("target template", new_id)?;
        if old_id == new_id {
            ensure!(
                self.config.targets.contains_key(old_id),
                "unknown target {old_id:?}"
            );
            return Ok(());
        }
        let journal = ConfigRenameJournal {
            kind: ConfigRenameKind::Target,
            old_id: old_id.to_owned(),
            new_id: new_id.to_owned(),
        };
        write_config_rename_journal(&journal)?;
        let (config, ()) = match Config::update(|config| {
            ensure!(
                config.targets.contains_key(old_id),
                "unknown target {old_id:?}"
            );
            ensure!(
                !config.targets.contains_key(new_id),
                "target {new_id:?} already exists"
            );
            let target = config
                .targets
                .remove(old_id)
                .expect("target was checked in the transaction");
            config.targets.insert(new_id.to_owned(), target);
            Ok(())
        }) {
            Ok(result) => result,
            Err(error) => {
                remove_config_rename_journal()
                    .context("remove target rename journal after config save failed")?;
                return Err(error).context("save renamed target configuration");
            }
        };
        self.config = config;
        mj_core::test_hooks::reach_test_hook("config_replacement_before_reference_migration")?;
        if let Err(error) = crate::database::rename_target_references(old_id, new_id) {
            let restore = Config::update(|config| {
                let target = config
                    .targets
                    .remove(new_id)
                    .with_context(|| format!("renamed target {new_id:?} is missing"))?;
                ensure!(
                    !config.targets.contains_key(old_id),
                    "cannot restore target rename: both {old_id:?} and {new_id:?} exist"
                );
                config.targets.insert(old_id.to_owned(), target);
                Ok(())
            });
            let restored = match restore {
                Ok((config, ())) => config,
                Err(restore_error) => {
                    return Err(error).context(format!(
                        "rename target references; additionally failed to restore config: {restore_error:#}"
                    ));
                }
            };
            self.config = restored;
            if let Err(restore_error) = remove_config_rename_journal() {
                return Err(error).context(format!(
                    "rename target references; additionally failed to remove rename journal: {restore_error:#}"
                ));
            }
            return Err(error).context("rename target references");
        }
        for session in self.state.sessions.values_mut() {
            if session.target_template_id == old_id {
                session.target_template_id = new_id.to_owned();
            }
        }
        remove_config_rename_journal()?;
        Ok(())
    }

    /// Finish a profile/target id rename interrupted between the atomic config
    /// replacement and SQLite transaction. Each step is idempotent, so a
    /// second crash leaves the same intent available for the next startup.
    pub fn recover_config_id_rename() -> Result<bool> {
        let path = config_rename_journal_path();
        let body = match fs::read(&path) {
            Ok(body) => body,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
            Err(error) => return Err(error).context(format!("read {}", path.display())),
        };
        let journal: ConfigRenameJournal =
            serde_json::from_slice(&body).with_context(|| format!("parse {}", path.display()))?;
        match journal.kind {
            ConfigRenameKind::Profile => {
                Config::update(|config| {
                    finish_config_map_rename(
                        &mut config.profiles,
                        &journal.old_id,
                        &journal.new_id,
                        "profile",
                    )?;
                    Ok(())
                })?;
                crate::database::rename_profile_references(&journal.old_id, &journal.new_id)?;
            }
            ConfigRenameKind::Target => {
                Config::update(|config| {
                    finish_config_map_rename(
                        &mut config.targets,
                        &journal.old_id,
                        &journal.new_id,
                        "target",
                    )?;
                    Ok(())
                })?;
                crate::database::rename_target_references(&journal.old_id, &journal.new_id)?;
            }
        }
        remove_config_rename_journal()?;
        Ok(true)
    }

    /// Record the per-session container size overrides and attached
    /// directories. Nothing is applied to a running container: the values are
    /// read the next time the session's container is created.
    pub fn update_session_container_settings(
        &mut self,
        session_id: &str,
        cpus: Option<String>,
        memory: Option<String>,
        additional_mounts: Vec<targets::AdditionalMount>,
        mount_history: Vec<std::path::PathBuf>,
    ) -> Result<()> {
        ensure!(
            self.state.sessions.contains_key(session_id),
            "unknown session {session_id}"
        );
        let cpus = cpus.filter(|value| !value.trim().is_empty());
        let memory = memory.filter(|value| !value.trim().is_empty());
        let updated_at = now();
        crate::database::set_session_container_settings(
            session_id,
            cpus.as_deref(),
            memory.as_deref(),
            &additional_mounts,
            &updated_at,
        )?;
        if let Some(host) = self
            .config
            .targets
            .get(
                &self.state.sessions[session_id]
                    .target_template_id
                    .to_owned(),
            )
            .and_then(mj_core::config::mount_history_host)
        {
            let host = host.to_owned();
            // The dialog owns the suggestion list, so forgetting a directory
            // there has to survive the mounts being remembered right after.
            crate::database::replace_mount_history(&host, &mount_history)?;
            crate::database::remember_mount_sources(&host, &additional_mounts)?;
            self.state.mount_history.insert(host.clone(), mount_history);
            self.state.remember_mount_sources(&host, &additional_mounts);
        }
        let record = self
            .state
            .sessions
            .get_mut(session_id)
            .expect("session was checked before updating its container settings");
        record.container_cpus = cpus;
        record.container_memory = memory;
        record.additional_mounts = additional_mounts;
        record.updated_at = updated_at;
        Ok(())
    }
}

fn config_rename_journal_path() -> PathBuf {
    data_dir().join(CONFIG_RENAME_JOURNAL)
}

fn write_config_rename_journal(journal: &ConfigRenameJournal) -> Result<()> {
    let path = config_rename_journal_path();
    let body = serde_json::to_vec(journal).context("serialize config rename journal")?;
    atomic_write(&path, &body).with_context(|| format!("write {}", path.display()))
}

fn remove_config_rename_journal() -> Result<()> {
    let path = config_rename_journal_path();
    match fs::remove_file(&path) {
        Ok(()) => Ok(()),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error).with_context(|| format!("remove {}", path.display())),
    }
}

fn finish_config_map_rename<T>(
    entries: &mut BTreeMap<String, T>,
    old_id: &str,
    new_id: &str,
    kind: &str,
) -> Result<()> {
    if let Some(entry) = entries.remove(old_id) {
        ensure!(
            !entries.contains_key(new_id),
            "cannot recover {kind} rename: both {old_id:?} and {new_id:?} exist"
        );
        entries.insert(new_id.to_owned(), entry);
    } else {
        ensure!(
            entries.contains_key(new_id),
            "cannot recover {kind} rename: neither {old_id:?} nor {new_id:?} exists"
        );
    }
    Ok(())
}

fn target_kind(locator: &targets::TargetLocator) -> &'static str {
    match locator {
        targets::TargetLocator::LocalBare { .. } => "local-bare",
        targets::TargetLocator::LocalPodman { .. } => "local-podman",
        targets::TargetLocator::LocalDocker { .. } => "local-docker",
        targets::TargetLocator::AppleContainer { .. } => "apple-container",
        targets::TargetLocator::AwsEc2 { .. } => "aws-ec2",
        targets::TargetLocator::SshBare { .. } => "ssh-bare",
        targets::TargetLocator::SshPodman { .. } => "ssh-podman",
        targets::TargetLocator::SshDocker { .. } => "ssh-docker",
    }
}

fn target_profile_home(
    locator: &targets::TargetLocator,
    session_id: &str,
    profile: &mj_core::config::HarnessProfile,
) -> String {
    let home = match locator {
        targets::TargetLocator::LocalBare { worker_root }
            if profile.kind == mj_core::config::HarnessKind::Claude =>
        {
            Path::new(worker_root)
                .join("profile")
                .to_string_lossy()
                .into_owned()
        }
        targets::TargetLocator::LocalBare { .. } => profile.home.to_string_lossy().into_owned(),
        targets::TargetLocator::LocalPodman { .. }
        | targets::TargetLocator::LocalDocker { .. }
        | targets::TargetLocator::AppleContainer { .. }
        | targets::TargetLocator::SshPodman { .. }
        | targets::TargetLocator::SshDocker { .. } => {
            format!("/var/lib/hel/profiles/{session_id}")
        }
        targets::TargetLocator::AwsEc2 { .. } | targets::TargetLocator::SshBare { .. } => {
            format!(".local/share/hel/profiles/{session_id}")
        }
    };
    if profile.kind == mj_core::config::HarnessKind::Muse {
        let root = if matches!(locator, targets::TargetLocator::LocalBare { .. }) {
            mj_core::config::data_dir()
                .join("profiles")
                .join(session_id)
        } else {
            PathBuf::from(home)
        };
        root.join("muse").to_string_lossy().into_owned()
    } else if profile.kind == mj_core::config::HarnessKind::Zcode
        && !matches!(locator, targets::TargetLocator::LocalBare { .. })
    {
        Path::new(&home)
            .join(".zcode")
            .to_string_lossy()
            .into_owned()
    } else {
        home
    }
}

/// Resolve the login home on the machine that owns an editable path.
pub fn resolve_target_input_path(
    target: &TargetTemplate,
    path: &Path,
    executor: &impl CommandExecutor,
) -> Result<PathBuf> {
    if !mj_core::path_input::needs_home(path)? {
        return Ok(path.to_path_buf());
    }
    match target {
        TargetTemplate::SshBare { ssh, .. }
        | TargetTemplate::SshPodman { ssh, .. }
        | TargetTemplate::SshDocker { ssh, .. } => {
            let mut ssh = ssh.clone();
            ssh.identity_file = ssh
                .identity_file
                .as_deref()
                .map(mj_core::path_input::expand_local)
                .transpose()?;
            let command =
                ssh_command_spec(&backend_ssh(&ssh), ["sh", "-c", "printf '%s' \"$HOME\""])
                    .purpose("resolve remote home directory");
            let output = executor.execute(&command)?;
            anyhow::ensure!(
                output.status == 0,
                "Could not resolve remote home: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
            let home =
                String::from_utf8(output.stdout).context("Remote home is not valid UTF-8")?;
            mj_core::path_input::expand_home(path, Some(Path::new(&home)))
        }
        _ => mj_core::path_input::expand_local(path),
    }
}

pub(crate) fn backend_ssh(ssh: &SshConnection) -> SshTarget {
    let destination = match &ssh.user {
        Some(user) => format!("{user}@{}", ssh.host),
        None => ssh.host.clone(),
    };
    SshTarget {
        destination,
        ssh_args: ssh_args_with_identity(&ssh.extra_args, ssh.identity_file.as_deref()),
    }
}

fn ssh_command_spec(
    ssh: &SshTarget,
    args: impl IntoIterator<Item = impl AsRef<str>>,
) -> CommandSpec {
    let remote = args
        .into_iter()
        .map(|arg| arg.as_ref().to_string())
        .collect::<Vec<_>>();
    let mut command_args = ssh.ssh_args.clone();
    command_args.push(ssh.destination.clone());
    command_args.push(targets::join_remote_command(&remote));
    CommandSpec::new("ssh", command_args).ssh_destination(ssh.destination.clone())
}

fn scp_command_spec(ssh: &SshTarget, source: &Path, remote: &str, recursive: bool) -> CommandSpec {
    let mut args = ssh.ssh_args.clone();
    if recursive {
        args.push("-r".into());
    }
    args.push(source.to_string_lossy().into_owned());
    args.push(format!("{}:{remote}", ssh.destination));
    // `scp` opens its own connection to the same host, so it competes for the
    // same pre-auth budget and is admitted and retried the same way.
    CommandSpec::new("scp", args).ssh_destination(ssh.destination.clone())
}

fn ssh_args_with_identity(args: &[String], identity: Option<&Path>) -> Vec<String> {
    // Mjolnir drives ssh non-interactively from a TUI; a host-key or password
    // prompt would steal the terminal and wedge provisioning. BatchMode fails
    // fast instead of prompting, and accept-new trusts a first-seen host key
    // (fresh EC2 instances are always first-seen) while still rejecting
    // changed keys. User-supplied ssh_args come last so they can override.
    let mut result = vec![
        "-o".into(),
        "BatchMode=yes".into(),
        "-o".into(),
        "StrictHostKeyChecking=accept-new".into(),
        "-o".into(),
        "ConnectTimeout=15".into(),
    ];
    result.extend(args.iter().cloned());
    if let Some(identity) = identity {
        result.push("-i".into());
        result.push(identity.to_string_lossy().into_owned());
    }
    result
}

fn execute_checked(executor: &impl CommandExecutor, command: CommandSpec) -> Result<CommandOutput> {
    let output = executor.execute(&command)?;
    if output.status != 0 {
        let detail = command_error_detail(&output.stderr);
        if detail.is_empty() {
            bail!("{} failed with status {}", command.purpose, output.status);
        }
        bail!("{detail}");
    }
    Ok(output)
}

fn command_error_detail(stderr: &[u8]) -> String {
    let reported = String::from_utf8_lossy(stderr);
    let reported = reported.trim();
    let detail = reported
        .rsplit_once("\nCaused by:\n")
        .map_or(reported, |(_, causes)| causes);
    let detail = detail.strip_prefix("Error: ").unwrap_or(detail);
    detail
        .lines()
        .map(|line| line.strip_prefix("    ").unwrap_or(line))
        .collect::<Vec<_>>()
        .join("\n")
        .trim()
        .to_owned()
}

fn now() -> String {
    Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
}

fn restore_session_after_persistence_failure(
    state: &mut State,
    session_id: &str,
    previous: &SessionRecord,
    primary: anyhow::Error,
    persist: impl FnOnce(&SessionRecord) -> Result<()>,
) -> anyhow::Error {
    state
        .sessions
        .insert(session_id.to_owned(), previous.clone());
    let restored = state
        .sessions
        .get(session_id)
        .expect("restored session record disappeared");
    match persist(restored) {
        Ok(()) => primary,
        Err(error) => primary.context(format!(
            "restored prior session state in memory, but failed to persist the rollback: {error:#}"
        )),
    }
}

fn persist_session_record_transition_or_restore(
    state: &mut State,
    session_id: &str,
    previous: &SessionRecord,
    context: &'static str,
    persist: &impl Fn(&SessionRecord) -> Result<()>,
) -> Result<()> {
    let result = persist(
        state
            .sessions
            .get(session_id)
            .expect("checkpoint session disappeared before persistence"),
    );
    match result {
        Ok(()) => Ok(()),
        Err(error) => Err(restore_session_after_persistence_failure(
            state,
            session_id,
            previous,
            error.context(context),
            persist,
        )),
    }
}

pub fn config_only_controller(config: Config) -> Controller {
    Controller {
        config,
        state: State::default(),
    }
}

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

    use crate::targets::ProcessExecutor;
    use mj_core::config::{
        Config, ContainerTemplate as ConfigContainer, HarnessKind, HarnessProfile, ProjectBundle,
        ProjectRepository, TargetTemplate,
    };
    use mj_core::state::State;

    use super::*;

    /// Every command that opens a connection to the host has to be admitted,
    /// `scp` included: it competes for the same pre-auth budget as `ssh`, and
    /// a connection the server drops carried no bytes, so it can be retried.
    #[test]
    fn ssh_and_scp_specs_are_both_tagged_with_the_connection_destination() {
        let ssh = SshTarget {
            destination: "build@10.0.0.1".into(),
            ssh_args: vec!["-o".into(), "BatchMode=yes".into()],
        };

        let uploaded = scp_command_spec(&ssh, Path::new("/tmp/local"), "remote/path", true);
        let ran = ssh_command_spec(&ssh, ["true"]);

        assert_eq!(uploaded.program, "scp");
        assert_eq!(uploaded.ssh_destination.as_deref(), Some("build@10.0.0.1"));
        assert_eq!(ran.ssh_destination.as_deref(), Some("build@10.0.0.1"));
    }

    /// One profile, one bundle with nothing checked out locally, and one
    /// container target, which is all `register_session_with_resources` reads.
    fn registration_config() -> Config {
        let mut config = Config::default();
        config.profiles.insert(
            "codex".into(),
            HarnessProfile {
                enabled: true,
                kind: HarnessKind::Codex,
                home: PathBuf::from("/home/dev/.codex"),
                environment: BTreeMap::new(),
                context_window_bytes: None,
            },
        );
        config.bundles.insert(
            "project".into(),
            ProjectBundle {
                primary_repo: "project".into(),
                repositories: vec![ProjectRepository {
                    id: "project".into(),
                    github: Some("owner/project".into()),
                    local: None,
                    destination: PathBuf::from("project"),
                    git_ref: None,
                }],
            },
        );
        config.targets.insert(
            "podman".into(),
            TargetTemplate::LocalPodman {
                container: ConfigContainer {
                    image: "example.invalid/hel-test:latest".into(),
                    pull_policy: Default::default(),
                    platform: None,
                    cpus: None,
                    memory: None,
                    environment: BTreeMap::new(),
                    workspace_storage: Default::default(),
                },
            },
        );
        config
    }

    #[test]
    fn remote_completion_preserves_home_shorthand_and_trailing_separator() {
        struct CompletionExecutor;
        impl CommandExecutor for CompletionExecutor {
            fn execute(&self, command: &CommandSpec) -> Result<targets::CommandOutput> {
                let script = command.args.last().unwrap();
                let stdout = if script.contains("$HOME") {
                    b"/remote".to_vec()
                } else {
                    assert!(script.contains("'/remote/cache/'"), "{script}");
                    b"/remote/cache/alpha/\n/remote/cache/alpine/\n".to_vec()
                };
                Ok(targets::CommandOutput {
                    status: 0,
                    stdout,
                    stderr: Vec::new(),
                })
            }
        }
        let target: TargetTemplate =
            serde_json::from_str(r#"{"kind":"ssh-podman","host":"builder","image":"test"}"#)
                .unwrap();
        let mut config = Config::default();
        config.targets.insert("remote".into(), target);
        let controller = Controller {
            config,
            state: State::default(),
        };
        let candidates = controller
            .complete_mount_source("remote", "~/cache/", &CompletionExecutor)
            .unwrap();
        assert_eq!(candidates, ["~/cache/alpha/", "~/cache/alpine/"]);
        assert_eq!(
            targets::path_completion("~/cache/", &candidates).as_deref(),
            Some("~/cache/alp")
        );
    }

    #[test]
    fn remote_path_resolution_uses_login_home_without_evaluating_suffix() {
        struct HomeExecutor {
            status: i32,
            home: &'static str,
        }
        impl CommandExecutor for HomeExecutor {
            fn execute(&self, command: &CommandSpec) -> Result<targets::CommandOutput> {
                assert_eq!(command.program, "ssh");
                let script = command.args.last().unwrap();
                assert!(!script.contains("touch"));
                assert!(script.contains("$HOME"));
                Ok(targets::CommandOutput {
                    status: self.status,
                    stdout: self.home.as_bytes().to_vec(),
                    stderr: b"home lookup failed".to_vec(),
                })
            }
        }
        let target: TargetTemplate = serde_json::from_str(
            r#"{"kind":"ssh-bare","host":"builder","permissions":"guardian"}"#,
        )
        .unwrap();
        let path = Path::new("~/資料/$(touch nope)");
        assert_eq!(
            resolve_target_input_path(
                &target,
                path,
                &HomeExecutor {
                    status: 0,
                    home: "/remote user"
                }
            )
            .unwrap(),
            Path::new("/remote user/資料/$(touch nope)")
        );
        assert!(
            resolve_target_input_path(
                &target,
                path,
                &HomeExecutor {
                    status: 1,
                    home: "/remote"
                }
            )
            .unwrap_err()
            .to_string()
            .contains("home lookup failed")
        );
        assert!(
            resolve_target_input_path(
                &target,
                path,
                &HomeExecutor {
                    status: 0,
                    home: ""
                }
            )
            .is_err()
        );
        assert!(
            resolve_target_input_path(
                &target,
                path,
                &HomeExecutor {
                    status: 0,
                    home: "relative"
                }
            )
            .is_err()
        );
    }

    #[test]
    fn bundle_creation_combines_sources_with_first_primary_and_stable_collisions() {
        let mut config = Config::default();
        let sources = vec!["example/app".into(), "other/app".into()];

        let bundle_id = create_bundle_from_sources_in_config(&mut config, &sources).unwrap();
        let bundle = &config.bundles[&bundle_id];
        assert_eq!(bundle_id, "app");
        assert_eq!(bundle.primary_repo, "app");
        assert_eq!(
            bundle
                .repositories
                .iter()
                .map(|repository| repository.id.as_str())
                .collect::<Vec<_>>(),
            ["app", "app-2"]
        );
        assert_eq!(
            bundle
                .repositories
                .iter()
                .map(|repository| repository.destination.to_string_lossy().into_owned())
                .collect::<Vec<_>>(),
            ["app".to_owned(), "app-2".to_owned()]
        );
        assert_eq!(
            bundle.repositories[0].github.as_deref(),
            Some("example/app")
        );
        assert_eq!(bundle.repositories[1].github.as_deref(), Some("other/app"));
    }

    #[test]
    fn bundle_creation_combines_local_and_github_sources_and_rejects_local_aliases() {
        let directory = tempfile::tempdir().unwrap();
        let root = directory.path().join("app");
        let output = mj_core::subprocess::run_capturing_stdout(
            std::process::Command::new("git").arg("init").arg(&root),
        )
        .unwrap();
        assert!(output.status.success(), "{output:?}");
        let nested = root.join("nested");
        fs::create_dir(&nested).unwrap();
        let mut config = Config::default();
        let sources = vec![root.to_str().unwrap().into(), "example/shared".into()];
        let id = create_bundle_from_sources_in_config(&mut config, &sources).unwrap();
        let bundle = &config.bundles[&id];
        assert_eq!(
            bundle.primary().unwrap().local,
            Some(root.canonicalize().unwrap())
        );
        assert_eq!(
            bundle.repositories[1].github.as_deref(),
            Some("example/shared")
        );
        let before = config.clone();
        let aliases = vec![
            root.to_str().unwrap().into(),
            nested.to_str().unwrap().into(),
        ];
        let error = create_bundle_from_sources_in_config(&mut config, &aliases).unwrap_err();
        assert!(
            error.to_string().contains("duplicate repository source"),
            "{error:#}"
        );
        assert_eq!(config, before);
    }

    #[test]
    fn bundle_creation_rejects_duplicate_normalized_sources_atomically() {
        let mut config = Config::default();
        let before = config.clone();
        let sources = vec![
            "example/app".into(),
            "https://github.com/EXAMPLE/APP.git".into(),
        ];

        let error = create_bundle_from_sources_in_config(&mut config, &sources).unwrap_err();
        assert!(error.to_string().contains("duplicate repository source"));
        assert_eq!(config, before);
    }

    #[test]
    fn bundle_creation_validates_every_source_before_mutating_config() {
        let mut config = Config::default();
        let before = config.clone();
        let invalid_directory = tempfile::tempdir().unwrap();
        let sources = vec![
            "example/app".into(),
            invalid_directory.path().to_string_lossy().into_owned(),
        ];

        let error = create_bundle_from_sources_in_config(&mut config, &sources).unwrap_err();
        assert!(error.to_string().contains("not a Git repository"));
        assert_eq!(config, before);
    }

    #[test]
    fn bundle_creation_reuses_an_exact_source_set_and_rejects_obsolete_pins() {
        let mut config = Config::default();
        config.bundles.insert(
            "all".into(),
            ProjectBundle {
                primary_repo: "app".into(),
                repositories: vec![
                    ProjectRepository {
                        id: "app".into(),
                        github: Some("example/app".into()),
                        local: None,
                        destination: "app".into(),
                        git_ref: None,
                    },
                    ProjectRepository {
                        id: "shared".into(),
                        github: Some("example/shared".into()),
                        local: None,
                        destination: "shared".into(),
                        git_ref: None,
                    },
                ],
            },
        );

        let one_source = vec!["example/app".into()];
        let created = create_bundle_from_sources_in_config(&mut config, &one_source).unwrap();
        assert_eq!(created, "app");
        assert_eq!(config.bundles[&created].repositories.len(), 1);
        assert_eq!(
            create_bundle_from_sources_in_config(&mut config, &one_source).unwrap(),
            "app"
        );

        let exact_sources = vec!["example/app".into(), "example/shared".into()];
        assert_eq!(
            create_bundle_from_sources_in_config(&mut config, &exact_sources).unwrap(),
            "all"
        );
        assert_eq!(config.bundles.len(), 2);
        config.bundles.get_mut("all").unwrap().repositories[0].git_ref = Some("release".into());
        let before = config.clone();
        let error = create_bundle_from_sources_in_config(&mut config, &exact_sources).unwrap_err();
        assert!(format!("{error:#}").contains("git_ref is no longer supported"));
        assert_eq!(config, before);
    }

    fn launch_options(additional_mounts: Vec<AdditionalMount>) -> SessionLaunchOptions {
        SessionLaunchOptions {
            mjolnir_subagents: None,
            create_managed_worktree: None,
            initial_prompt: None,
            workspace_id: mj_core::workspace::DEFAULT_WORKSPACE_ID.to_owned(),
            additional_mounts,
            allow_dirty_local: false,
            resource_allocation: None,
            project_directory: None,
            session_title_override: None,
        }
    }

    #[test]
    fn registration_rejects_a_disabled_profile_before_persisting() {
        let mut config = registration_config();
        config.profiles.get_mut("codex").unwrap().enabled = false;
        let mut controller = Controller {
            config,
            state: State::default(),
        };

        let error = controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "disabled",
                launch_options(Vec::new()),
            )
            .unwrap_err();

        assert!(error.to_string().contains("disabled"));
        assert!(controller.state.sessions.is_empty());
    }

    #[test]
    fn deepseek_registration_rejects_more_than_one_workspace_root_before_persisting() {
        let mut config = registration_config();
        config.profiles.get_mut("codex").unwrap().kind = HarnessKind::Deepseek;
        let second = config.bundles["project"].repositories[0].clone();
        config
            .bundles
            .get_mut("project")
            .unwrap()
            .repositories
            .push(mj_core::config::ProjectRepository {
                id: "second".into(),
                destination: "second".into(),
                ..second
            });
        let mut controller = Controller {
            config,
            state: State::default(),
        };

        let error = controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "unsupported",
                launch_options(Vec::new()),
            )
            .unwrap_err();

        assert!(error.to_string().contains("one workspace root"));
        assert!(controller.state.sessions.is_empty());
    }

    /// MJ_DATA_DIR is process-global, so every test that reaches the
    /// controller database runs in an exact child with its own data directory.
    fn run_registration_child(marker: &str, test: &str, data_directory: &Path) {
        let output = std::process::Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                &format!("controller::tests::{test}"),
                "--nocapture",
            ])
            .env(marker, "1")
            .env("MJ_DATA_DIR", data_directory)
            .env("MJ_CONFIG_DIR", data_directory)
            .output()
            .unwrap();
        assert!(
            output.status.success(),
            "isolated {test} failed\nstdout:\n{}\nstderr:\n{}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
    }

    #[test]
    fn registration_saves_the_initial_task_before_provisioning() {
        const MARKER: &str = "MJ_TEST_INITIAL_TASK_CHILD";
        if std::env::var_os(MARKER).is_none() {
            let directory = tempfile::tempdir().unwrap();
            run_registration_child(
                MARKER,
                "registration_saves_the_initial_task_before_provisioning",
                directory.path(),
            );
            return;
        }
        let _writer = crate::database::install_isolated_test_writer();
        let mut controller = Controller {
            config: registration_config(),
            state: State::default(),
        };
        let prompt = format!(
            "Initial task\n{}\n\tPreserve indentation and λ",
            "x".repeat(70_000)
        );
        let mut options = launch_options(Vec::new());
        options.initial_prompt = Some(prompt.clone());
        let id = controller
            .register_session_with_resources("codex", "project", "podman", "fresh task", options)
            .unwrap();
        let saved = crate::database::load_state().unwrap();
        assert_eq!(saved.sessions[&id].draft_input, prompt);
        assert_eq!(saved.sessions[&id].state, SessionState::Provisioning);
        crate::database::set_session_draft_input(&id, "a newer draft").unwrap();
        crate::database::clear_session_draft_input_if_matches(&id, &prompt).unwrap();
        assert_eq!(
            crate::database::load_state().unwrap().sessions[&id].draft_input,
            "a newer draft"
        );
        crate::database::clear_session_draft_input_if_matches(&id, "a newer draft").unwrap();
        assert!(
            crate::database::load_state().unwrap().sessions[&id]
                .draft_input
                .is_empty()
        );
    }

    const UNPERSISTABLE_SESSION_CHILD: &str = "MJ_TEST_UNPERSISTABLE_SESSION_CHILD";

    #[test]
    fn missing_bundle_does_not_block_controller_or_other_sessions() {
        const CHILD: &str = "MJ_TEST_MISSING_BUNDLE_CHILD";
        if std::env::var_os(CHILD).is_none() {
            let directory = tempfile::tempdir().unwrap();
            run_registration_child(
                CHILD,
                "missing_bundle_does_not_block_controller_or_other_sessions",
                directory.path(),
            );
            return;
        }
        let _writer = crate::database::install_isolated_test_writer();
        let mut controller = Controller {
            config: registration_config(),
            state: State::default(),
        };
        let bundle = controller.config.bundles["project"].clone();
        controller
            .config
            .bundles
            .insert("other".into(), bundle.clone());
        controller.config.save().unwrap();
        let affected = controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "affected",
                launch_options(Vec::new()),
            )
            .unwrap();
        let healthy = controller
            .register_session_with_resources(
                "codex",
                "other",
                "podman",
                "healthy",
                launch_options(Vec::new()),
            )
            .unwrap();
        controller.config.bundles.remove("project");
        controller.config.save().unwrap();
        let loaded = Controller::load().unwrap();
        assert_eq!(loaded.state.sessions.len(), 2);
        assert!(
            loaded.state.sessions[&healthy]
                .configuration_issue(&loaded.config)
                .is_none()
        );
        let issue = loaded.reconnect_command(&affected).unwrap_err().to_string();
        assert!(issue.contains("missing bundle"), "{issue}");
        assert!(issue.contains("config.toml"), "{issue}");
        // Loading must not turn a configuration problem into a persisted lifecycle failure.
        assert_eq!(
            loaded.state.sessions[&affected],
            controller.state.sessions[&affected]
        );
        Config::update(|config| {
            config.bundles.insert("project".into(), bundle);
            Ok(())
        })
        .unwrap();
        let repaired = Controller::load().unwrap();
        assert!(
            repaired.state.sessions[&affected]
                .configuration_issue(&repaired.config)
                .is_none()
        );
    }

    const CONFIG_ID_RENAME_CHILD: &str = "MJ_TEST_CONFIG_ID_RENAME_CHILD";

    #[test]
    fn configuration_id_rename_rewrites_durable_session_references() {
        if std::env::var_os(CONFIG_ID_RENAME_CHILD).is_none() {
            let directory = tempfile::tempdir().unwrap();
            run_registration_child(
                CONFIG_ID_RENAME_CHILD,
                "configuration_id_rename_rewrites_durable_session_references",
                directory.path(),
            );
            return;
        }
        // Alone in this child process, so it installs the one writer.
        let _writer = crate::database::install_isolated_test_writer();

        let mut controller = Controller {
            config: registration_config(),
            state: State::default(),
        };
        controller.config.save().unwrap();
        let session_id = controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "rename references",
                launch_options(Vec::new()),
            )
            .unwrap();

        controller
            .rename_profile_id("codex", "codex-renamed")
            .unwrap();
        controller
            .rename_target_id("podman", "podman-renamed")
            .unwrap();

        let loaded = Controller::load().unwrap();
        let session = &loaded.state.sessions[&session_id];
        assert_eq!(session.last_profile, "codex-renamed");
        assert_eq!(session.target_template_id, "podman-renamed");
        assert!(loaded.config.profiles.contains_key("codex-renamed"));
        assert!(loaded.config.targets.contains_key("podman-renamed"));
        assert!(!config_rename_journal_path().exists());
    }

    #[test]
    fn a_session_the_database_rejects_is_never_left_in_memory() {
        if std::env::var_os(UNPERSISTABLE_SESSION_CHILD).is_none() {
            let directory = tempfile::tempdir().unwrap();
            run_registration_child(
                UNPERSISTABLE_SESSION_CHILD,
                "a_session_the_database_rejects_is_never_left_in_memory",
                directory.path(),
            );
            return;
        }
        // Alone in this child process, so it installs the one writer.
        let _writer = crate::database::install_isolated_test_writer();

        let mut controller = Controller {
            config: registration_config(),
            state: State::default(),
        };
        // The store has to be healthy enough to open before it can reject a
        // write: this test is about a write the database refuses, not about a
        // store that cannot be opened at all, which now fails earlier and
        // louder when the writer is installed. The first registration builds
        // the schema the second one then loses.
        controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "first",
                launch_options(Vec::new()),
            )
            .expect("a healthy store registers a session");
        rusqlite::Connection::open(crate::database::database_path())
            .unwrap()
            .execute_batch("DROP TABLE sessions")
            .unwrap();

        let error = controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "unpersistable",
                launch_options(Vec::new()),
            )
            .expect_err("a store that rejects the write cannot register a session");
        assert!(
            format!("{error:#}").contains("sessions"),
            "unexpected error: {error:#}"
        );
        assert_eq!(
            controller.state.sessions.len(),
            1,
            "a session the database never accepted stayed in controller memory"
        );
        assert!(
            controller
                .state
                .sessions
                .values()
                .all(|session| session.title != "unpersistable"),
            "the rejected session is the one that stayed"
        );
    }

    const MOUNT_HISTORY_FAILURE_CHILD: &str = "MJ_TEST_MOUNT_HISTORY_FAILURE_CHILD";

    const CONTAINER_SIZE_HISTORY_CHILD: &str = "MJ_TEST_CONTAINER_SIZE_HISTORY_CHILD";

    #[test]
    fn registration_remembers_launch_size_but_session_overrides_do_not_replace_it() {
        if std::env::var_os(CONTAINER_SIZE_HISTORY_CHILD).is_none() {
            let directory = tempfile::tempdir().unwrap();
            run_registration_child(
                CONTAINER_SIZE_HISTORY_CHILD,
                "registration_remembers_launch_size_but_session_overrides_do_not_replace_it",
                directory.path(),
            );
            return;
        }
        // Alone in this child process, so it installs the one writer.
        let _writer = crate::database::install_isolated_test_writer();

        let mut controller = Controller {
            config: registration_config(),
            state: State::default(),
        };
        let mut options = launch_options(Vec::new());
        options.resource_allocation = Some(SessionResourceAllocation::Container {
            cpus: 12,
            memory_bytes: 48 * 1024 * 1024 * 1024,
        });
        let id = controller
            .register_session_with_resources("codex", "project", "podman", "sized", options)
            .unwrap();
        let expected = HostContainerSize {
            cpus: 12,
            memory_bytes: 48 * 1024 * 1024 * 1024,
        };
        assert_eq!(controller.state.container_sizes["local"], expected);
        assert_eq!(
            crate::database::load_state_migrating()
                .unwrap()
                .container_sizes["local"],
            expected
        );

        controller
            .update_session_container_settings(
                &id,
                Some("2".into()),
                Some("4g".into()),
                Vec::new(),
                Vec::new(),
            )
            .unwrap();
        assert_eq!(
            crate::database::load_state_migrating()
                .unwrap()
                .container_sizes["local"],
            expected
        );
    }

    #[test]
    fn a_failed_mount_history_write_does_not_fail_the_registered_session() {
        if std::env::var_os(MOUNT_HISTORY_FAILURE_CHILD).is_none() {
            let directory = tempfile::tempdir().unwrap();
            run_registration_child(
                MOUNT_HISTORY_FAILURE_CHILD,
                "a_failed_mount_history_write_does_not_fail_the_registered_session",
                directory.path(),
            );
            return;
        }
        // Alone in this child process, so it installs the one writer.
        let _writer = crate::database::install_isolated_test_writer();

        let mut controller = Controller {
            config: registration_config(),
            state: State::default(),
        };
        // The first registration builds the schema this test then breaks.
        controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "first",
                launch_options(Vec::new()),
            )
            .expect("a healthy store registers a session");
        let database = crate::database::database_path();
        rusqlite::Connection::open(&database)
            .unwrap()
            .execute_batch("DROP TABLE mount_history")
            .unwrap();

        let id = controller
            .register_session_with_resources(
                "codex",
                "project",
                "podman",
                "attached",
                launch_options(vec![AdditionalMount {
                    source: PathBuf::from("/host/models"),
                    destination: PathBuf::from("/mnt/models"),
                    read_only: false,
                }]),
            )
            .expect("a suggestion list that cannot be written must not fail a registration");

        let stored: i64 = rusqlite::Connection::open(&database)
            .unwrap()
            .query_row(
                "SELECT count(*) FROM sessions WHERE session_id = ?1",
                [&id],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(stored, 1, "the registered session was not committed");
        assert!(
            controller.state.mount_history.is_empty(),
            "controller memory remembered mount sources the database never stored"
        );
    }

    #[test]
    fn command_errors_report_the_root_cause_without_worker_wrappers() {
        let stderr = b"Error: restore target checkpoint failed with status 1: Error: restore repository \"bifrost\"\n\nCaused by:\n    checkpoint base b41dc78 is absent from configured source\n    repository may have moved\n";

        assert_eq!(
            command_error_detail(stderr),
            "checkpoint base b41dc78 is absent from configured source\nrepository may have moved"
        );
    }

    #[test]
    fn controller_store_lock_excludes_a_second_process_owner() {
        let directory = tempfile::tempdir().unwrap();
        let first = ControllerStoreGuard::acquire_at(directory.path()).unwrap();
        run_controller_lock_probe(directory.path(), true);
        drop(first);
        run_controller_lock_probe(directory.path(), false);
    }
    fn run_controller_lock_probe(directory: &Path, expect_locked: bool) {
        let output = std::process::Command::new(std::env::current_exe().unwrap())
            .args([
                "--exact",
                "controller::tests::controller_store_lock_subprocess_probe",
                "--nocapture",
            ])
            .env("MJ_CONTROLLER_LOCK_PROBE", directory)
            .env(
                "MJ_CONTROLLER_LOCK_EXPECTED",
                if expect_locked { "locked" } else { "available" },
            )
            .output()
            .unwrap();
        assert!(
            output.status.success(),
            "controller lock subprocess failed:\nstdout:\n{}\nstderr:\n{}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
    }
    #[test]
    fn controller_store_lock_subprocess_probe() {
        let Some(directory) = std::env::var_os("MJ_CONTROLLER_LOCK_PROBE") else {
            return;
        };
        let expected = std::env::var("MJ_CONTROLLER_LOCK_EXPECTED").unwrap();
        let acquired = ControllerStoreGuard::acquire_at(Path::new(&directory));
        match expected.as_str() {
            "locked" => {
                let error = acquired.expect_err("a second process acquired the controller store");
                assert!(error.to_string().contains("another Mjolnir controller"));
            }
            "available" => {
                acquired.expect("released controller store stayed locked");
            }
            value => panic!("unexpected lock probe expectation {value:?}"),
        }
    }
    #[test]
    fn local_mount_source_must_be_an_existing_directory() {
        let directory = tempfile::tempdir().unwrap();
        let file = directory.path().join("file");
        std::fs::write(&file, "not a directory").unwrap();
        let mut config = Config::default();
        config.targets.insert(
            "local".into(),
            TargetTemplate::LocalPodman {
                container: ConfigContainer {
                    image: "ubuntu:24.04".into(),
                    pull_policy: Default::default(),
                    platform: None,
                    cpus: None,
                    memory: None,
                    environment: BTreeMap::new(),
                    workspace_storage: Default::default(),
                },
            },
        );
        let controller = Controller {
            config,
            state: State::default(),
        };

        assert!(
            controller
                .validate_mount_source("local", directory.path(), &ProcessExecutor)
                .is_ok()
        );
        for invalid in [file, directory.path().join("missing")] {
            let error = controller
                .validate_mount_source("local", &invalid, &ProcessExecutor)
                .unwrap_err();
            assert!(
                error
                    .to_string()
                    .contains("does not exist or is not a directory")
            );
        }
    }
}