mod backend;
mod checkpoint;
mod git_cache;
mod lifecycle;
pub mod move_session;
mod provisioning;
mod readiness;
mod recovery_scan;
mod resume;
mod reviewer;
#[cfg(test)]
mod test_support;
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 hel::hel_config::{
HelConfig, ProjectBundle, ProjectRepository, SshConnection, TargetTemplate, atomic_write,
container_size_host, data_dir, is_bare_project_target, mount_history_host,
};
use crate::hel_import::{
RepositoryIdentity, bundle_matches, configured_bundle_for_local, configured_bundle_for_origin,
setup_style_id,
};
use crate::hel_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 hel::hel_local_git::dirty_local_repositories;
use hel::hel_state::{
HelState, HostContainerSize, SessionRecord, SessionResourceAllocation, SessionState,
new_session_id, normalize_session_title,
};
use hel::hel_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, checkpoint_was_deferred,
reconcile_managed_checkpoint_archives,
};
pub use recovery_scan::{RecoveryCandidate, RecoveryScan};
pub use resume::{
ResumeRepositorySourceMismatch, ResumeRepositorySourcePreflight, ResumeRepositorySourceReceipt,
};
pub use worker_binary::{WorkerBinaryAvailability, worker_binary_prerequisite_for_arch};
pub use worker_restart::WorkerUpgradeOutcome;
pub use worktree::{ResumePlan, local_project_repository, resume_compatibility};
pub struct Controller {
pub config: HelConfig,
pub state: HelState,
}
#[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()
)
})
}
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 }))
}
pub fn start_database_writer(&self) -> Result<hel::hel_database::DatabaseWriterOwner> {
hel::hel_database::start_database_writer()
}
}
impl Drop for ControllerStoreGuard {
fn drop(&mut self) {
let _ = self.file.unlock();
}
}
#[derive(Debug)]
pub struct QuickBundleCreation {
pub config: HelConfig,
pub bundle_id: String,
}
#[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()),
}
}
}
pub fn create_quick_bundle(
source: &str,
) -> std::result::Result<QuickBundleCreation, QuickBundleFailure> {
let (config, bundle_id) = HelConfig::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 })
}
pub fn create_quick_bundle_in_config(config: &mut HelConfig, 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)
}
pub fn create_bundle_from_sources(
sources: &[String],
) -> std::result::Result<QuickBundleCreation, QuickBundleFailure> {
let (config, bundle_id) = HelConfig::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 })
}
pub fn create_bundle_from_sources_in_config(
config: &mut HelConfig,
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);
}
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::hel_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,
}
}
}
fn interpret_repository_source(source: &str) -> Result<InterpretedRepositorySource> {
let source = source.trim();
if source.is_empty() {
bail!("repository source cannot be empty");
}
let candidate = Path::new(source);
if candidate.exists() {
let root = hel::hel_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: &HelConfig,
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 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 = HelConfig::load()?;
let state = HelState::load()?;
state.validate_against_config(&config)?;
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) => hel::hel_database::save_lifecycle_session(session),
None => hel::hel_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,
&hel::hel_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,
hel::hel_database::save_lifecycle_session,
)
}
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:?}"))?;
match target {
TargetTemplate::LocalPodman { .. }
| TargetTemplate::LocalDocker { .. }
| TargetTemplate::AppleContainer { .. }
| TargetTemplate::AwsEc2 { .. } => Ok(hel_targets::local_directory_completions(prefix)),
TargetTemplate::SshPodman { ssh, .. } | TargetTemplate::SshDocker { ssh, .. } => {
hel_targets::ssh_directory_completions(&backend_ssh(ssh), prefix, executor)
}
TargetTemplate::LocalBare | TargetTemplate::SshBare { .. } => {
bail!("resource path completion is unsupported for bare targets")
}
}
}
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, .. } => {
hel_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))
}
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))
}
_ => return None,
};
let filesystem = hel_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 = hel_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 = format!("{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 {
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:?}"))?;
let template = self
.config
.targets
.get(target_id)
.with_context(|| format!("unknown target template {target_id:?}"))?;
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,
hel::hel_config::HarnessKind::Deepseek | hel::hel_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()
);
}
let dirty = bundle
.map(dirty_local_repositories)
.transpose()?
.unwrap_or_default();
if !allow_dirty_local && !dirty.is_empty() {
let repositories = dirty
.iter()
.map(|repository| format!("{} ({})", repository.path.display(), repository.summary))
.collect::<Vec<_>>()
.join(", ");
bail!(
"local repositories have uncommitted changes: {repositories}; explicit confirmation is required"
);
}
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");
}
hel_targets::validate_additional_mounts(&additional_mounts)?;
let id = new_session_id()?;
let now = now();
let record = SessionRecord {
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,
};
if let Some((host, size)) = selected_container_size.as_ref() {
hel::hel_database::save_session_with_container_size(&record, host, *size)?;
} else {
hel::hel_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) {
match hel::hel_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();
hel::hel_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<()> {
hel::hel_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 HelConfig::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);
if config.startup.profile.as_deref() == Some(old_id) {
config.startup.profile = Some(new_id.to_owned());
}
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;
hel::hel_test_hooks::reach_test_hook("config_replacement_before_reference_migration")?;
if let Err(error) = hel::hel_database::rename_profile_references(old_id, new_id) {
let restore = HelConfig::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);
if config.startup.profile.as_deref() == Some(new_id) {
config.startup.profile = Some(old_id.to_owned());
}
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<()> {
hel::hel_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 HelConfig::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);
if config.startup.target.as_deref() == Some(old_id) {
config.startup.target = Some(new_id.to_owned());
}
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;
hel::hel_test_hooks::reach_test_hook("config_replacement_before_reference_migration")?;
if let Err(error) = hel::hel_database::rename_target_references(old_id, new_id) {
let restore = HelConfig::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);
if config.startup.target.as_deref() == Some(new_id) {
config.startup.target = Some(old_id.to_owned());
}
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(())
}
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 => {
HelConfig::update(|config| {
finish_config_map_rename(
&mut config.profiles,
&journal.old_id,
&journal.new_id,
"profile",
)?;
Ok(())
})?;
hel::hel_database::rename_profile_references(&journal.old_id, &journal.new_id)?;
}
ConfigRenameKind::Target => {
HelConfig::update(|config| {
finish_config_map_rename(
&mut config.targets,
&journal.old_id,
&journal.new_id,
"target",
)?;
Ok(())
})?;
hel::hel_database::rename_target_references(&journal.old_id, &journal.new_id)?;
}
}
remove_config_rename_journal()?;
Ok(true)
}
pub fn update_session_container_settings(
&mut self,
session_id: &str,
cpus: Option<String>,
memory: Option<String>,
additional_mounts: Vec<hel_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();
hel::hel_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(hel::hel_config::mount_history_host)
{
let host = host.to_owned();
hel::hel_database::replace_mount_history(&host, &mount_history)?;
hel::hel_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: &hel_targets::TargetLocator) -> &'static str {
match locator {
hel_targets::TargetLocator::LocalBare { .. } => "local-bare",
hel_targets::TargetLocator::LocalPodman { .. } => "local-podman",
hel_targets::TargetLocator::LocalDocker { .. } => "local-docker",
hel_targets::TargetLocator::AppleContainer { .. } => "apple-container",
hel_targets::TargetLocator::AwsEc2 { .. } => "aws-ec2",
hel_targets::TargetLocator::SshBare { .. } => "ssh-bare",
hel_targets::TargetLocator::SshPodman { .. } => "ssh-podman",
hel_targets::TargetLocator::SshDocker { .. } => "ssh-docker",
}
}
fn target_profile_home(
locator: &hel_targets::TargetLocator,
session_id: &str,
profile: &hel::hel_config::HarnessProfile,
) -> String {
let home = match locator {
hel_targets::TargetLocator::LocalBare { .. } => profile.home.to_string_lossy().into_owned(),
hel_targets::TargetLocator::LocalPodman { .. }
| hel_targets::TargetLocator::LocalDocker { .. }
| hel_targets::TargetLocator::AppleContainer { .. }
| hel_targets::TargetLocator::SshPodman { .. }
| hel_targets::TargetLocator::SshDocker { .. } => {
format!("/var/lib/hel/profiles/{session_id}")
}
hel_targets::TargetLocator::AwsEc2 { .. } | hel_targets::TargetLocator::SshBare { .. } => {
format!(".local/share/hel/profiles/{session_id}")
}
};
if profile.kind == hel::hel_config::HarnessKind::Muse {
let root = if matches!(locator, hel_targets::TargetLocator::LocalBare { .. }) {
hel::hel_config::data_dir()
.join("profiles")
.join(session_id)
} else {
PathBuf::from(home)
};
root.join("muse").to_string_lossy().into_owned()
} else {
home
}
}
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(hel_targets::join_remote_command(&remote));
CommandSpec::new("ssh", command_args)
}
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));
CommandSpec::new("scp", args)
}
fn ssh_args_with_identity(args: &[String], identity: Option<&Path>) -> Vec<String> {
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 HelState,
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 HelState,
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,
)),
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::path::Path;
use hel::hel_config::{
ContainerTemplate as ConfigContainer, HarnessKind, HarnessProfile, HelConfig,
ProjectBundle, ProjectRepository, TargetTemplate,
};
use hel::hel_state::HelState;
use hel::hel_targets::ProcessExecutor;
use super::*;
fn registration_config() -> HelConfig {
let mut config = HelConfig::default();
config.profiles.insert(
"codex".into(),
HarnessProfile {
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 bundle_creation_combines_sources_with_first_primary_and_stable_collisions() {
let mut config = HelConfig::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 = hel::hel_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 = HelConfig::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 = HelConfig::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 = HelConfig::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_only_an_exact_unpinned_source_set() {
let mut config = HelConfig::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 unpinned = create_bundle_from_sources_in_config(&mut config, &exact_sources).unwrap();
assert_ne!(unpinned, "all");
assert!(
config.bundles[&unpinned]
.repositories
.iter()
.all(|repo| repo.git_ref.is_none())
);
}
fn launch_options(additional_mounts: Vec<AdditionalMount>) -> SessionLaunchOptions {
SessionLaunchOptions {
initial_prompt: None,
workspace_id: hel::hel_workspace::DEFAULT_WORKSPACE_ID.to_owned(),
additional_mounts,
allow_dirty_local: false,
resource_allocation: None,
project_directory: None,
session_title_override: None,
}
}
#[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(hel::hel_config::ProjectRepository {
id: "second".into(),
destination: "second".into(),
..second
});
let mut controller = Controller {
config,
state: HelState::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());
}
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!("hel_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 = hel::hel_database::install_isolated_test_writer();
let mut controller = Controller {
config: registration_config(),
state: HelState::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 = hel::hel_database::load_state().unwrap();
assert_eq!(saved.sessions[&id].draft_input, prompt);
assert_eq!(saved.sessions[&id].state, SessionState::Provisioning);
hel::hel_database::set_session_draft_input(&id, "a newer draft").unwrap();
hel::hel_database::clear_session_draft_input_if_matches(&id, &prompt).unwrap();
assert_eq!(
hel::hel_database::load_state().unwrap().sessions[&id].draft_input,
"a newer draft"
);
hel::hel_database::clear_session_draft_input_if_matches(&id, "a newer draft").unwrap();
assert!(
hel::hel_database::load_state().unwrap().sessions[&id]
.draft_input
.is_empty()
);
}
const UNPERSISTABLE_SESSION_CHILD: &str = "MJ_TEST_UNPERSISTABLE_SESSION_CHILD";
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;
}
let _writer = hel::hel_database::install_isolated_test_writer();
let mut controller = Controller {
config: registration_config(),
state: HelState::default(),
};
controller.config.startup.profile = Some("codex".into());
controller.config.startup.target = Some("podman".into());
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_eq!(
loaded.config.startup.profile.as_deref(),
Some("codex-renamed")
);
assert_eq!(
loaded.config.startup.target.as_deref(),
Some("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;
}
let _writer = hel::hel_database::install_isolated_test_writer();
let mut controller = Controller {
config: registration_config(),
state: HelState::default(),
};
controller
.register_session_with_resources(
"codex",
"project",
"podman",
"first",
launch_options(Vec::new()),
)
.expect("a healthy store registers a session");
rusqlite::Connection::open(hel::hel_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;
}
let _writer = hel::hel_database::install_isolated_test_writer();
let mut controller = Controller {
config: registration_config(),
state: HelState::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!(HelState::load().unwrap().container_sizes["local"], expected);
controller
.update_session_container_settings(
&id,
Some("2".into()),
Some("4g".into()),
Vec::new(),
Vec::new(),
)
.unwrap();
assert_eq!(HelState::load().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;
}
let _writer = hel::hel_database::install_isolated_test_writer();
let mut controller = Controller {
config: registration_config(),
state: HelState::default(),
};
controller
.register_session_with_resources(
"codex",
"project",
"podman",
"first",
launch_options(Vec::new()),
)
.expect("a healthy store registers a session");
let database = hel::hel_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",
"hel_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 = HelConfig::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: HelState::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")
);
}
}
}