1mod backend;
4mod cache_host;
5pub(crate) mod checkpoint;
6mod git_cache;
7mod github_app;
8mod lifecycle;
9pub mod local_profile_homes;
10pub(crate) mod mbx;
11pub mod move_session;
12mod network_git;
13mod new_session_preflight;
14mod tool_cache;
15pub use new_session_preflight::{NewSessionPreflight, NewSessionRepository};
16mod path_completion;
17pub mod profile_config;
18mod provisioning;
19pub(crate) mod publication;
20mod readiness;
21pub(crate) use network_git::workspace_root;
22pub(crate) use readiness::HarnessPreparationFailure;
23pub(crate) use readiness::NATIVE_SESSION_STARTUP_TIMEOUT;
24mod recovery_scan;
25mod resume;
26mod reviewer;
27mod subagent_park;
28pub use subagent_park::ParkOutcome;
29pub(crate) use subagent_park::{FAILED_STARTUP_CLEANUP_TIMEOUT, failed_launch_cleanup_executor};
30mod subagents;
31#[cfg(test)]
32pub(crate) mod test_support;
33pub mod update;
34mod worker_binary;
35mod worker_restart;
36mod worktree;
37pub(crate) use worktree::RemoteGitExecutor;
38
39use std::collections::{BTreeMap, BTreeSet};
40use std::fs::{self, File, OpenOptions};
41use std::path::{Path, PathBuf};
42
43use anyhow::{Context, Result, bail, ensure};
44use chrono::Utc;
45
46use mj_core::config::{
47 Config, ProjectBundle, ProjectRepository, TargetTemplate, atomic_write, container_size_host,
48 data_dir, is_bare_project_target, mount_history_host,
49};
50
51use crate::import::{RepositoryIdentity, setup_style_id};
52use crate::setup::github_repository_from_origin;
53
54const CONFIG_RENAME_JOURNAL: &str = "config-rename.json";
55
56#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
57#[serde(rename_all = "snake_case")]
58enum ConfigRenameKind {
59 Profile,
60 Target,
61}
62
63#[derive(Debug, serde::Serialize, serde::Deserialize)]
64#[serde(deny_unknown_fields)]
65struct ConfigRenameJournal {
66 kind: ConfigRenameKind,
67 old_id: String,
68 new_id: String,
69}
70use mj_core::state::{
71 HostContainerSize, SessionRecord, SessionResourceAllocation, SessionState, State,
72 new_session_id, normalize_session_title,
73};
74
75use crate::targets::{
76 self, AdditionalMount, CommandExecutor, CommandOutput, CommandSpec, SshTarget,
77};
78
79pub(crate) use backend::backend_locator;
80pub(crate) use backend::controller_github_token;
81pub use backend::image_refresh_plan;
82pub(crate) use backend::validate_resource_allocation;
83pub(crate) use backend::{LocalEngineReadiness, local_engine_readiness};
84pub use github_app::GithubBundleSelectionError;
85pub(crate) use github_app::{
86 GithubAppTokenProvider, github_app_token_for_session, github_token_for_session,
87};
88pub use mbx::install::{
89 MbxInstallKind, MbxInstallResult, install_kind as mbx_install_kind, install_mbx,
90};
91pub use mbx::preview_build_cache;
92pub(crate) use mbx::release::{FAILURE_REPORT_SECS, ReleaseFailure, recent_release_failures};
93pub(crate) use mbx::{DoctorHostMbxStatus, MBX_VERSION, doctor_host_mbx};
94use provisioning::apply_failed_new_session_rollback;
95pub(crate) use worker_binary::{
96 prepare_recovery_worker_binary, recorded_exit_reason, refresh_target_worker_binary_if_stale,
97 worker_source_problem,
98};
99pub(crate) use worktree::path_exists_on_managed_target;
100
101pub use checkpoint::{
102 CheckpointArtifact, CheckpointDeferred, IdleWorkspaceLease, SessionExportLayout,
103 checkpoint_was_deferred, reconcile_managed_checkpoint_archives,
104 sweep_local_checkpoint_leftovers,
105};
106pub use lifecycle::{
107 BeforeClose, BranchDisposition, CheckoutDisposition, has_nothing_to_checkpoint,
108};
109pub use recovery_scan::{RecoveryCandidate, RecoveryScan};
110pub(crate) use resume::InPlaceRestartError;
111pub use resume::{
112 ResumeRepositorySourceMismatch, ResumeRepositorySourcePreflight, ResumeRepositorySourceReceipt,
113 raw_conversion_preview_for,
114};
115pub use reviewer::reviewer_stager;
116pub use subagents::{RegisterSubagentRequest, stopped_subagent, subagent_has_handed_back};
117pub use worker_binary::{
118 WorkerBinaryAvailability, native_worker_binary_prerequisite, pin_worker_binary_sources,
119 ssh_worker_binary_prerequisite, warm_worker_binary_sources,
120 worker_binary_prerequisite_for_arch,
121};
122pub use worker_restart::WorkerUpgradeOutcome;
123pub use worktree::{
124 ResumePlan, local_project_repository, resume_compatibility, resume_compatibility_with_checkout,
125};
126
127pub struct Controller {
128 pub config: Config,
129 pub state: State,
130}
131
132#[derive(Debug)]
136pub struct ControllerStoreGuard {
137 file: File,
138}
139
140impl ControllerStoreGuard {
141 pub fn acquire() -> Result<Self> {
142 let directory = data_dir();
143 Self::acquire_at(&directory)
144 }
145
146 fn acquire_at(directory: &Path) -> Result<Self> {
147 Self::try_acquire_at(directory)?.with_context(|| {
148 format!(
149 "another Mjolnir controller is already using {}; stop it before starting this command",
150 directory.display()
151 )
152 })
153 }
154
155 pub fn try_acquire() -> Result<Option<Self>> {
157 Self::try_acquire_at(&data_dir())
158 }
159
160 fn try_acquire_at(directory: &Path) -> Result<Option<Self>> {
161 mj_core::config::ensure_may_control_store(directory, "own this data store")?;
163 std::fs::create_dir_all(directory)
164 .with_context(|| format!("create controller data directory {}", directory.display()))?;
165 let path = directory.join("controller.lock");
166 let mut options = OpenOptions::new();
167 options.create(true).read(true).write(true);
168 #[cfg(unix)]
169 {
170 use std::os::unix::fs::OpenOptionsExt;
171 options.mode(0o600);
172 }
173 let file = options
174 .open(&path)
175 .with_context(|| format!("open controller lock {}", path.display()))?;
176 match file.try_lock() {
177 Ok(()) => {}
178 Err(std::fs::TryLockError::WouldBlock) => return Ok(None),
179 Err(std::fs::TryLockError::Error(error)) => {
180 return Err(error)
181 .with_context(|| format!("lock controller store {}", directory.display()));
182 }
183 }
184 Ok(Some(Self { file }))
185 }
186
187 pub fn start_database_writer(&self) -> Result<crate::database::DatabaseWriterOwner> {
190 crate::database::start_database_writer()
191 }
192}
193
194impl Drop for ControllerStoreGuard {
195 fn drop(&mut self) {
196 let _ = self.file.unlock();
199 }
200}
201
202#[derive(Debug)]
206pub struct QuickBundleCreation {
207 pub config: Config,
208 pub bundle_id: String,
209}
210
211#[derive(Debug)]
215pub enum QuickBundleFailure {
216 InvalidSource(anyhow::Error),
217 Persistence(anyhow::Error),
218}
219
220impl std::fmt::Display for QuickBundleFailure {
221 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
222 match self {
223 Self::InvalidSource(error) => write!(formatter, "invalid repository source: {error}"),
224 Self::Persistence(error) => write!(formatter, "persist quick bundle: {error}"),
225 }
226 }
227}
228
229impl std::error::Error for QuickBundleFailure {
230 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
231 match self {
232 Self::InvalidSource(error) | Self::Persistence(error) => Some(error.root_cause()),
233 }
234 }
235}
236
237pub fn create_quick_bundle(
243 source: &str,
244) -> std::result::Result<QuickBundleCreation, QuickBundleFailure> {
245 create_bundle_from_sources(&[source.to_owned()])
246}
247
248pub fn create_quick_bundle_in_config(config: &mut Config, source: &str) -> Result<String> {
253 create_bundle_from_sources_in_config(config, &[source.to_owned()])
254}
255
256pub fn create_bundle_from_sources(
263 sources: &[String],
264) -> std::result::Result<QuickBundleCreation, QuickBundleFailure> {
265 let (config, bundle_id) = Config::update(|config| {
266 let previous = config.bundles.clone();
267 let id = create_bundle_from_sources_in_config(config, sources)
268 .map_err(|error| anyhow::Error::new(QuickBundleFailure::InvalidSource(error)))?;
269 if let Some(old) = previous.get(&id)
270 && config.bundles.get(&id) != Some(old)
271 {
272 let accepted = crate::project_catalog::snapshot(
273 old,
274 &targets::CancellableProcessExecutor::with_timeout(std::time::Duration::from_secs(
275 15,
276 )),
277 true,
278 )?;
279 crate::database::store_catalog_project(&id, &accepted, true)?;
280 }
281 let project = crate::project_catalog::snapshot(
282 config
283 .bundles
284 .get(&id)
285 .context("created project is missing")?,
286 &targets::CancellableProcessExecutor::with_timeout(std::time::Duration::from_secs(15)),
287 false,
288 )?;
289 let canonical = crate::database::store_catalog_project(&id, &project, true)?;
290 if canonical != id {
291 let mut saved = crate::database::read_project_catalog()?
292 .projects
293 .into_iter()
294 .find(|entry| entry.bundle_id == canonical)
295 .context("canonical project is missing")?
296 .project;
297 for repository in &mut saved.bundle.repositories {
298 let identity = saved
299 .identities
300 .get(&repository.id)
301 .context("canonical repository identity is missing")?;
302 if let Some(source) = project.bundle.repositories.iter().find(|source| {
303 project.identities.get(&source.id) == Some(identity)
304 && source.destination == repository.destination
305 }) {
306 repository.local = source.local.clone();
307 repository.github = source.github.clone();
308 }
309 }
310 config
311 .bundles
312 .insert(canonical.clone(), saved.bundle.clone());
313 config.bundles.remove(&id);
314 crate::database::store_catalog_project(&canonical, &saved, true)?;
315 }
316 Ok(canonical)
317 })
318 .map_err(|error| {
319 error
320 .downcast::<QuickBundleFailure>()
321 .unwrap_or_else(QuickBundleFailure::Persistence)
322 })?;
323 Ok(QuickBundleCreation { config, bundle_id })
324}
325
326pub fn remove_bundle(bundle_id: &str) -> Result<Config> {
331 let state = crate::database::load_state()?;
332 if let Some(refusal) = state.bundle_removal_refusal(bundle_id) {
333 bail!(refusal);
334 }
335 let (config, ()) = Config::update(|config| {
336 ensure!(
337 config.bundles.remove(bundle_id).is_some(),
338 "project {bundle_id:?} is no longer saved"
339 );
340 config.validate()
341 })?;
342 crate::database::hide_catalog_project(bundle_id)?;
343 Ok(config)
344}
345
346pub fn create_bundle_from_sources_in_config(
350 config: &mut Config,
351 sources: &[String],
352) -> Result<String> {
353 let sources = sources
354 .iter()
355 .map(|source| interpret_repository_source(source))
356 .collect::<Result<Vec<_>>>()?;
357 if sources.is_empty() {
358 bail!("at least one repository source is required");
359 }
360
361 let mut identities = BTreeSet::new();
362 for source in &sources {
363 if !identities.insert(source.identity()) {
364 bail!("duplicate repository source {:?}", source.display_name);
365 }
366 }
367
368 if let Some(existing) = exact_configured_bundle(config, &sources)? {
369 let bundle = config
370 .bundles
371 .get_mut(&existing)
372 .expect("matched project exists");
373 for repository in &mut bundle.repositories {
376 let Some(identity) =
377 crate::import::configured_repository_identity_for_lookup(&existing, repository)
378 else {
379 continue;
380 };
381 if let Some(source) = sources.iter().find(|source| {
382 source.identity == identity && matches!(source.kind, RepositorySourceKind::Local(_))
383 }) && let RepositorySourceKind::Local(root) = &source.kind
384 {
385 repository.local = Some(root.clone());
386 repository.github = None;
387 }
388 }
389 return Ok(existing);
390 }
391
392 let mut updated = config.clone();
395 let mut used_repository_ids = BTreeSet::new();
396 let mut repositories = Vec::with_capacity(sources.len());
397 for source in sources {
398 let base = setup_style_id(&source.name);
399 let repository_id = unique_id(&base, |candidate| used_repository_ids.contains(candidate));
400 used_repository_ids.insert(repository_id.clone());
401 repositories.push(source.into_project_repository(repository_id));
402 }
403 let primary_repo = repositories
404 .first()
405 .map(|repository| repository.id.clone())
406 .context("at least one repository source is required")?;
407 let bundle_id = unique_id(&primary_repo, |candidate| {
408 updated.bundles.contains_key(candidate)
409 });
410 updated.bundles.insert(
411 bundle_id.clone(),
412 ProjectBundle {
413 primary_repo,
414 repositories,
415 },
416 );
417 updated.validate()?;
418 *config = updated;
419 Ok(bundle_id)
420}
421
422#[derive(Debug, Clone)]
423enum RepositorySourceKind {
424 Github(crate::setup::GithubRepository),
425 Local(PathBuf),
426}
427
428#[derive(Debug, Clone)]
429struct InterpretedRepositorySource {
430 display_name: String,
431 name: String,
432 kind: RepositorySourceKind,
433 identity: RepositoryIdentity,
434}
435
436impl InterpretedRepositorySource {
437 fn identity(&self) -> RepositoryIdentity {
438 self.identity.clone()
439 }
440
441 fn into_project_repository(self, id: String) -> ProjectRepository {
442 let (github, local) = match self.kind {
443 RepositorySourceKind::Github(repository) => (
444 Some(format!("{}/{}", repository.owner, repository.repository)),
445 None,
446 ),
447 RepositorySourceKind::Local(root) => (None, Some(root)),
448 };
449 ProjectRepository {
450 id: id.clone(),
451 github,
452 local,
453 destination: PathBuf::from(id),
454 git_ref: None,
455 }
456 }
457}
458
459fn interpret_repository_source(source: &str) -> Result<InterpretedRepositorySource> {
462 let source = source.trim();
463 if source.is_empty() {
464 bail!("repository source cannot be empty");
465 }
466 let expanded = mj_core::path_input::expand_local(Path::new(source))?;
467 let candidate = expanded.as_path();
468 if candidate.exists() {
469 let resolved = mj_core::repository::resolve_directory(
470 candidate,
471 None,
472 &mj_core::targets::CancellableProcessExecutor::with_timeout(
473 std::time::Duration::from_secs(8),
474 ),
475 )?;
476 let identity = resolved.identity;
477 let name = identity.name();
478 return Ok(InterpretedRepositorySource {
479 display_name: source.to_owned(),
480 name,
481 identity,
482 kind: RepositorySourceKind::Local(resolved.checkout_root),
483 });
484 }
485 if candidate.is_absolute() || source.starts_with('.') || source.starts_with('~') {
486 bail!("local repository path {source:?} does not exist");
487 }
488 let repository = github_repository_from_origin(source).context(format!(
489 "{source:?} is not a GitHub owner/repository or URL"
490 ))?;
491 Ok(InterpretedRepositorySource {
492 display_name: source.to_owned(),
493 name: repository.repository.clone(),
494 identity: RepositoryIdentity::Github(
495 repository.owner.to_ascii_lowercase(),
496 repository.repository.to_ascii_lowercase(),
497 ),
498 kind: RepositorySourceKind::Github(repository),
499 })
500}
501
502fn exact_configured_bundle(
503 config: &Config,
504 requested: &[InterpretedRepositorySource],
505) -> Result<Option<String>> {
506 let primary = requested
507 .first()
508 .context("no repository sources")?
509 .identity();
510 let mut used = BTreeSet::new();
511 let requested_layout = requested
512 .iter()
513 .map(|source| {
514 let id = unique_id(&setup_style_id(&source.name), |id| used.contains(id));
515 used.insert(id.clone());
516 (source.identity(), PathBuf::from(id))
517 })
518 .collect::<BTreeSet<_>>();
519 for (id, bundle) in &config.bundles {
520 if bundle.repositories.len() != requested.len()
521 || bundle
522 .repositories
523 .iter()
524 .any(|repository| repository.git_ref.is_some())
525 {
526 continue;
527 }
528 let identity_results = bundle
529 .repositories
530 .iter()
531 .map(|repo| crate::import::configured_repository_identity_for_lookup(id, repo))
532 .collect::<Vec<_>>();
533 if identity_results.iter().any(Option::is_none) {
534 continue;
535 }
536 let identities = identity_results.into_iter().flatten().collect::<Vec<_>>();
537 let layout = bundle
538 .repositories
539 .iter()
540 .zip(&identities)
541 .map(|(repository, identity)| (identity.clone(), repository.destination.clone()))
542 .collect::<BTreeSet<_>>();
543 let Some(primary_index) = bundle
544 .repositories
545 .iter()
546 .position(|repository| repository.id == bundle.primary_repo)
547 else {
548 continue;
549 };
550 if layout == requested_layout && identities.get(primary_index) == Some(&primary) {
551 return Ok(Some(id.clone()));
552 }
553 }
554 Ok(None)
555}
556
557fn unique_id(base: &str, mut is_used: impl FnMut(&str) -> bool) -> String {
558 if !is_used(base) {
559 return base.to_owned();
560 }
561 for suffix in 2_u32.. {
562 let suffix = format!("-{suffix}");
563 let prefix_len = 64usize.saturating_sub(suffix.len());
564 let prefix = base.chars().take(prefix_len).collect::<String>();
565 let candidate = format!("{prefix}{suffix}");
566 if !is_used(&candidate) {
567 return candidate;
568 }
569 }
570 unreachable!("u32 repository/bundle id suffixes exhausted")
571}
572
573pub struct SessionLaunchOptions {
574 pub create_managed_worktree: Option<bool>,
575 pub at: Option<String>,
577 pub branch: Option<String>,
579 pub base: Option<String>,
581 pub subagents: Option<mj_core::subagent::SubagentPolicy>,
582 pub review: Option<mj_core::config::SessionReview>,
584 pub no_project_memory: bool,
586 pub no_mailbox: bool,
588 pub initial_prompt: Option<String>,
589 pub workspace_id: String,
590 pub additional_mounts: Vec<AdditionalMount>,
591 pub resource_allocation: Option<SessionResourceAllocation>,
592 pub project_directory: Option<PathBuf>,
593 pub session_title_override: Option<String>,
594}
595
596pub struct SessionResumeOptions {
597 pub additional_mounts: Option<Vec<AdditionalMount>>,
598 pub resource_allocation: Option<SessionResourceAllocation>,
599 pub discard_queue: bool,
600}
601
602fn selected_host_container_size(
603 template: &TargetTemplate,
604 allocation: Option<&SessionResourceAllocation>,
605) -> Option<(String, HostContainerSize)> {
606 let host = container_size_host(template)?;
607 let SessionResourceAllocation::Container { cpus, memory_bytes } = allocation? else {
608 return None;
609 };
610 Some((
611 host.to_owned(),
612 HostContainerSize {
613 cpus: *cpus,
614 memory_bytes: *memory_bytes,
615 },
616 ))
617}
618
619impl Controller {
620 pub fn load() -> Result<Self> {
621 let config = Config::load().map_err(|error| {
624 let refusal = mj_core::refusal::Refusal::precondition(format!("{error:#}"));
625 error.context(refusal)
626 })?;
627 let state = crate::database::load_state()?;
628 Ok(Self { config, state })
629 }
630
631 pub(crate) fn prepare_persisted_sessions(&mut self) -> Result<()> {
634 self.state.validate()?;
637 for session in self.state.sessions.values() {
638 if let Some(issue) = session.configuration_issue(&self.config) {
639 tracing::warn!(session_id = %session.id, "{issue}");
640 }
641 }
642 self.backfill_target_runtime()
643 }
644
645 pub fn reload(&mut self) -> Result<()> {
646 self.backfill_target_runtime()?;
649 *self = Self::load()?;
650 Ok(())
651 }
652
653 fn backfill_target_runtime(&mut self) -> Result<()> {
654 let missing: Vec<_> = self
655 .state
656 .sessions
657 .values()
658 .filter(|session| session.target_runtime.is_none())
659 .map(|session| session.id.clone())
660 .collect();
661 for id in missing {
662 let session = self.state.sessions.get_mut(&id).expect("collected session");
663 match session.target_runtime_settings(&self.config) {
664 Ok(runtime) => {
665 let runtime = runtime.into_owned();
666 session.target_runtime = crate::database::backfill_target_runtime(
667 &session.id,
668 &session.target_template_id,
669 &runtime,
670 )?;
671 }
672 Err(error) => {
673 tracing::warn!(session_id = %session.id, %error, "target access needs configuration repair")
674 }
675 }
676 }
677 Ok(())
678 }
679
680 fn persist_session_state(&self, session_id: &str) -> Result<()> {
681 match self.state.sessions.get(session_id) {
682 Some(session) => crate::database::save_lifecycle_session(session),
683 None => crate::database::delete_session(session_id),
684 }
685 }
686
687 fn persist_session_transition_or_restore(
688 &mut self,
689 session_id: &str,
690 previous: &SessionRecord,
691 context: &'static str,
692 ) -> Result<()> {
693 persist_session_record_transition_or_restore(
694 &mut self.state,
695 session_id,
696 previous,
697 context,
698 &crate::database::save_lifecycle_session,
699 )
700 }
701
702 fn restore_prior_session_after_persistence_failure(
703 &mut self,
704 session_id: &str,
705 previous: &SessionRecord,
706 primary: anyhow::Error,
707 ) -> anyhow::Error {
708 restore_session_after_persistence_failure(
709 &mut self.state,
710 session_id,
711 previous,
712 primary,
713 crate::database::save_lifecycle_session,
714 )
715 }
716
717 pub fn resolve_input_path(
719 &self,
720 target_id: &str,
721 path: &Path,
722 executor: &impl CommandExecutor,
723 ) -> Result<PathBuf> {
724 let target = self
725 .config
726 .targets
727 .get(target_id)
728 .context("Unknown path target")?;
729 resolve_target_input_path(target, path, executor)
730 }
731
732 pub fn validate_mount_source(
739 &self,
740 target_id: &str,
741 source: &Path,
742 executor: &impl CommandExecutor,
743 ) -> Result<Option<String>> {
744 let target = self
745 .config
746 .targets
747 .get(target_id)
748 .with_context(|| format!("unknown target template {target_id:?}"))?;
749 let exists = match target {
750 TargetTemplate::LocalPodman { .. }
751 | TargetTemplate::LocalDocker { .. }
752 | TargetTemplate::AppleContainer { .. }
753 | TargetTemplate::AwsEc2 { .. } => std::fs::metadata(source)
754 .map(|metadata| metadata.is_dir())
755 .or_else(|error| {
756 if error.kind() == std::io::ErrorKind::NotFound {
757 Ok(false)
758 } else {
759 Err(error)
760 }
761 })
762 .with_context(|| format!("inspect resource source {}", source.display()))?,
763 TargetTemplate::SshPodman { ssh, .. } | TargetTemplate::SshDocker { ssh, .. } => {
764 targets::ssh_directory_exists(&SshTarget::from(ssh), source, executor)?
765 }
766 TargetTemplate::LocalBare | TargetTemplate::SshBare { .. } => {
767 bail!("resource attachments are unsupported for bare targets")
768 }
769 };
770 ensure!(
771 exists,
772 "source path {} does not exist or is not a directory",
773 source.display()
774 );
775 Ok(self.forced_read_only_reason(target, source, executor))
776 }
777
778 fn forced_read_only_reason(
780 &self,
781 target: &TargetTemplate,
782 source: &Path,
783 executor: &impl CommandExecutor,
784 ) -> Option<String> {
785 let ssh = match target {
786 TargetTemplate::LocalPodman { .. } | TargetTemplate::LocalDocker { .. } => None,
787 TargetTemplate::SshPodman { ssh, .. } | TargetTemplate::SshDocker { ssh, .. } => {
788 Some(SshTarget::from(ssh))
789 }
790 _ => return None,
793 };
794 if matches!(target, TargetTemplate::LocalDocker { .. })
795 && let Ok(Some(_)) = targets::local_docker_vm_share(executor)
796 {
797 return Some("Docker VM file share (cannot back an overlay)".to_owned());
798 }
799 let filesystem = targets::probe_filesystem_types(
800 ssh.as_ref(),
801 std::slice::from_ref(&source.to_path_buf()),
802 executor,
803 )
804 .map_err(|error| {
805 tracing::debug!(
806 source = %source.display(),
807 error = format!("{error:#}"),
808 "could not probe the filesystem under a mount source"
809 );
810 })
811 .ok()?
812 .pop()?;
813 let reason = targets::overlay_unsupported_filesystem(&filesystem)?;
814 Some(format!("{filesystem} ({reason})"))
815 }
816
817 fn fail_new_session_with_cleanup(
818 &mut self,
819 session_id: &str,
820 error: anyhow::Error,
821 executor: &impl CommandExecutor,
822 ) -> Result<anyhow::Error> {
823 let original = provisioning::note_new_session_launch_failure(session_id, &error);
824 let cleanup_error = self
825 .cleanup_new_session_worktree_after_failure(session_id, executor)
826 .err()
827 .map(|cleanup_error| format!("{cleanup_error:#}"));
828 if let Some(cleanup_error) = &cleanup_error {
829 tracing::warn!(
830 session_id,
831 error = %cleanup_error,
832 "new-session worktree rollback reported a cleanup failure"
833 );
834 }
835 let failure = apply_failed_new_session_rollback(
836 &mut self.state,
837 session_id,
838 &original,
839 cleanup_error,
840 );
841 self.persist_session_state(session_id)?;
842 Ok(failure)
843 }
844
845 pub fn register_session_with_resources(
846 &mut self,
847 profile_id: &str,
848 bundle_id: &str,
849 target_id: &str,
850 title: impl Into<String>,
851 options: SessionLaunchOptions,
852 ) -> Result<String> {
853 self.register_session_with_executor(
854 profile_id,
855 bundle_id,
856 target_id,
857 title,
858 options,
859 &targets::CancellableProcessExecutor::with_timeout(std::time::Duration::from_secs(15)),
860 )
861 }
862
863 fn register_session_with_executor(
864 &mut self,
865 profile_id: &str,
866 bundle_id: &str,
867 target_id: &str,
868 title: impl Into<String>,
869 options: SessionLaunchOptions,
870 executor: &impl CommandExecutor,
871 ) -> Result<String> {
872 let SessionLaunchOptions {
873 create_managed_worktree,
874 at,
875 branch,
876 base,
877 subagents,
878 review,
879 no_project_memory,
880 no_mailbox,
881 initial_prompt,
882 workspace_id,
883 additional_mounts,
884 resource_allocation,
885 project_directory,
886 session_title_override,
887 } = options;
888 let launch_base = match base {
889 Some(base) => {
890 let base = base.trim();
891 if base.is_empty() {
892 bail!("`base` must not be empty");
893 }
894 Some(base.to_owned())
895 }
896 None => None,
897 };
898 let branch = match branch {
899 Some(branch) => {
900 let branch = branch.trim();
901 ensure!(!branch.is_empty(), "`branch` must not be empty");
902 Some(branch.to_owned())
903 }
904 None => None,
905 };
906 if launch_base.is_some() && create_managed_worktree == Some(false) {
907 bail!("`base` requires a managed worktree or a bundle session");
908 }
909 let session_title_override = match session_title_override {
910 Some(title) => {
911 Some(normalize_session_title(&title).context("session name cannot be empty")?)
912 }
913 None => None,
914 };
915 let profile = self
916 .config
917 .profiles
918 .get(profile_id)
919 .with_context(|| format!("unknown profile {profile_id:?}"))?;
920 ensure!(profile.enabled, "profile {profile_id:?} is disabled");
921 let template = self
922 .config
923 .targets
924 .get(target_id)
925 .with_context(|| format!("unknown target template {target_id:?}"))?;
926 if create_managed_worktree == Some(true) && !is_bare_project_target(template) {
927 bail!("managed worktree creation requires a bare Git project");
928 }
929 if project_directory.is_some() != is_bare_project_target(template) {
930 bail!("raw project directories require a bare target, and bare targets require one");
931 }
932 if let Some(path) = &project_directory
933 && (!path.is_absolute()
934 || path
935 .components()
936 .any(|part| part == std::path::Component::ParentDir))
937 {
938 bail!("bare project directory must be an absolute safe path");
939 }
940 let bundle = project_directory
941 .is_none()
942 .then(|| self.config.bundles.get(bundle_id))
943 .flatten();
944 if project_directory.is_none() && bundle.is_none() {
945 bail!("unknown bundle {bundle_id:?}");
946 }
947 let (checkout, launch_branch) = match at {
951 Some(at) => {
952 let bundle = bundle.context("`at` requires a bundle-backed session")?;
953 ensure!(
954 create_managed_worktree != Some(false),
955 "`at` requires an isolated workspace"
956 );
957 let mut checkout = mj_core::remote_git::ExactCheckout {
958 repository_id: bundle.primary_repo.clone(),
959 commit: at.trim().to_owned(),
960 branch,
961 };
962 checkout.validate()?;
963 checkout.commit.make_ascii_lowercase();
964 (Some(checkout), None)
965 }
966 None => (None, branch),
967 };
968 if profile.kind == mj_core::config::HarnessKind::Muse
969 && (!additional_mounts.is_empty()
970 || bundle.is_some_and(|bundle| bundle.repositories.len() > 1))
971 {
972 bail!(
973 "{} ACP supports one workspace root; use a single-repository bundle without attached directories",
974 profile.kind.display_name()
975 );
976 }
977 let (bundle_id, project) = if let Some(bundle) = bundle {
978 let project = crate::project_catalog::snapshot(bundle, executor, true)?;
979 let canonical = crate::database::store_catalog_project(bundle_id, &project, true)?;
980 (canonical, Some(project))
981 } else if let Some(directory) = project_directory.as_deref() {
982 let git = match template {
983 TargetTemplate::LocalBare => {
984 executor
985 .execute(
986 &targets::CommandSpec::new(
987 "git",
988 [
989 "-C",
990 &directory.to_string_lossy(),
991 "rev-parse",
992 "--is-inside-work-tree",
993 ],
994 )
995 .purpose("inspect raw project repository"),
996 )?
997 .status
998 == 0
999 }
1000 _ => true,
1001 };
1002 if git {
1003 let (id, project) =
1004 crate::project_catalog::accept_directory(template, directory, executor)?;
1005 (id, Some(project))
1006 } else {
1007 (bundle_id.to_owned(), None)
1008 }
1009 } else {
1010 (bundle_id.to_owned(), None)
1011 };
1012 validate_resource_allocation(template, resource_allocation.as_ref())?;
1013 let selected_container_size =
1014 selected_host_container_size(template, resource_allocation.as_ref());
1015 if !additional_mounts.is_empty() && mount_history_host(template).is_none() {
1016 bail!("attached resources are unsupported for this target");
1017 }
1018 targets::validate_additional_mounts(&additional_mounts)?;
1019 let id = new_session_id()?;
1020 let now = now();
1021 let record = SessionRecord {
1022 project,
1023 target_runtime: Some(template.into()),
1024 build_cache: None,
1025 create_managed_worktree,
1026 launch_base,
1027 launch_branch,
1028 checkout,
1029 publication: None,
1030 subagents: Some(subagents.unwrap_or_default()),
1031 archived: false,
1032 container_cpus: None,
1033 container_memory: None,
1034 container_workspace: Some(targets::new_container_workspace(&id)?),
1039 id: id.clone(),
1040 workspace_id,
1041 title: title.into(),
1042 harness_kind: profile.kind,
1043 last_profile: profile_id.to_string(),
1044 bundle_id,
1045 project_directory,
1046 managed_worktree: None,
1047 review,
1048 no_project_memory,
1049 no_mailbox,
1050 target_template_id: target_id.to_string(),
1051 resource_allocation,
1052 additional_mounts: additional_mounts.clone(),
1053 state: SessionState::Provisioning,
1054 target: None,
1055 native_session_id: None,
1056 acp_session_title: None,
1057 session_title_override,
1058 created_at: now.clone(),
1059 updated_at: now,
1060 viewed_through_event_ordinal: 0,
1061 draft_input: initial_prompt.unwrap_or_default(),
1062 last_error: None,
1063 last_checkpoint_error: None,
1064 checkpoint: None,
1065 };
1066 crate::database::save_new_session(&record, selected_container_size.clone())?;
1070 if record.harness_kind.supports_delegation_tools() {
1071 self.state.last_subagent_policy = record.subagents.clone().unwrap_or_default();
1072 }
1073 self.state.sessions.insert(id.clone(), record);
1074 if let Some((host, size)) = selected_container_size {
1075 self.state.remember_container_size(&host, size);
1076 }
1077 if let Some(host) = mount_history_host(template) {
1078 match crate::database::remember_mount_sources(host, &additional_mounts) {
1082 Ok(()) => self.state.remember_mount_sources(host, &additional_mounts),
1083 Err(error) => tracing::warn!(
1084 session_id = id,
1085 error = format!("{error:#}"),
1086 "could not remember the attached resource directories for later suggestions"
1087 ),
1088 }
1089 }
1090 Ok(id)
1091 }
1092
1093 pub fn rename_session(&mut self, session_id: &str, title: &str) -> Result<String> {
1094 let title = normalize_session_title(title).context("session name cannot be empty")?;
1095 ensure!(
1096 self.state.sessions.contains_key(session_id),
1097 "unknown session {session_id}"
1098 );
1099 let updated_at = now();
1100 crate::database::set_session_title_override(session_id, &title, &updated_at)?;
1101 let record = self
1102 .state
1103 .sessions
1104 .get_mut(session_id)
1105 .expect("session was checked before updating its title");
1106 record.session_title_override = Some(title.clone());
1107 record.updated_at = updated_at;
1108 Ok(title)
1109 }
1110
1111 pub fn set_session_workspace(&mut self, session_id: &str, workspace_id: &str) -> Result<usize> {
1114 ensure!(
1115 self.state.sessions.contains_key(session_id),
1116 "unknown session {session_id}"
1117 );
1118 let mut ids = vec![session_id.to_owned()];
1119 ids.extend(
1120 self.state
1121 .subagents
1122 .values()
1123 .filter(|child| child.parent_session_id == session_id)
1124 .map(|child| child.child_session_id.clone()),
1125 );
1126 crate::database::set_sessions_workspace(&ids, workspace_id)?;
1127 for id in &ids {
1128 if let Some(record) = self.state.sessions.get_mut(id) {
1129 record.workspace_id = workspace_id.to_owned();
1130 }
1131 }
1132 Ok(ids.len())
1133 }
1134
1135 pub fn rename_profile_id(&mut self, old_id: &str, new_id: &str) -> Result<()> {
1136 mj_core::config::validate_id("profile", new_id)?;
1137 if old_id == new_id {
1138 ensure!(
1139 self.config.profiles.contains_key(old_id),
1140 "unknown profile {old_id:?}"
1141 );
1142 return Ok(());
1143 }
1144 let journal = ConfigRenameJournal {
1145 kind: ConfigRenameKind::Profile,
1146 old_id: old_id.to_owned(),
1147 new_id: new_id.to_owned(),
1148 };
1149 write_config_rename_journal(&journal)?;
1150 let (config, ()) = match Config::update(|config| {
1151 ensure!(
1152 config.profiles.contains_key(old_id),
1153 "unknown profile {old_id:?}"
1154 );
1155 ensure!(
1156 !config.profiles.contains_key(new_id),
1157 "profile {new_id:?} already exists"
1158 );
1159 let profile = config
1160 .profiles
1161 .remove(old_id)
1162 .expect("profile was checked in the transaction");
1163 config.profiles.insert(new_id.to_owned(), profile);
1164 Ok(())
1165 }) {
1166 Ok(result) => result,
1167 Err(error) => {
1168 remove_config_rename_journal()
1169 .context("remove profile rename journal after config save failed")?;
1170 return Err(error).context("save renamed profile configuration");
1171 }
1172 };
1173 self.config = config;
1174 mj_core::test_hooks::reach_test_hook("config_replacement_before_reference_migration")?;
1175 if let Err(error) = crate::database::rename_profile_references(old_id, new_id) {
1176 let restore = Config::update(|config| {
1177 let profile = config
1178 .profiles
1179 .remove(new_id)
1180 .with_context(|| format!("renamed profile {new_id:?} is missing"))?;
1181 ensure!(
1182 !config.profiles.contains_key(old_id),
1183 "cannot restore profile rename: both {old_id:?} and {new_id:?} exist"
1184 );
1185 config.profiles.insert(old_id.to_owned(), profile);
1186 Ok(())
1187 });
1188 let restored = match restore {
1189 Ok((config, ())) => config,
1190 Err(restore_error) => {
1191 return Err(error).context(format!(
1192 "rename profile references; additionally failed to restore config: {restore_error:#}"
1193 ));
1194 }
1195 };
1196 self.config = restored;
1197 if let Err(restore_error) = remove_config_rename_journal() {
1198 return Err(error).context(format!(
1199 "rename profile references; additionally failed to remove rename journal: {restore_error:#}"
1200 ));
1201 }
1202 return Err(error).context("rename profile references");
1203 }
1204 let affected: Vec<_> = self
1205 .state
1206 .sessions
1207 .values()
1208 .filter(|session| session.last_profile == old_id)
1209 .map(|session| session.id.clone())
1210 .collect();
1211 for id in affected {
1212 self.state
1213 .sessions
1214 .get_mut(&id)
1215 .expect("collected session")
1216 .last_profile = new_id.to_owned();
1217 }
1218 remove_config_rename_journal()?;
1219 Ok(())
1220 }
1221
1222 pub fn rename_target_id(&mut self, old_id: &str, new_id: &str) -> Result<()> {
1223 mj_core::config::validate_id("target template", new_id)?;
1224 if old_id == new_id {
1225 ensure!(
1226 self.config.targets.contains_key(old_id),
1227 "unknown target {old_id:?}"
1228 );
1229 return Ok(());
1230 }
1231 let journal = ConfigRenameJournal {
1232 kind: ConfigRenameKind::Target,
1233 old_id: old_id.to_owned(),
1234 new_id: new_id.to_owned(),
1235 };
1236 write_config_rename_journal(&journal)?;
1237 let (config, ()) = match Config::update(|config| {
1238 ensure!(
1239 config.targets.contains_key(old_id),
1240 "unknown target {old_id:?}"
1241 );
1242 ensure!(
1243 !config.targets.contains_key(new_id),
1244 "target {new_id:?} already exists"
1245 );
1246 let target = config
1247 .targets
1248 .remove(old_id)
1249 .expect("target was checked in the transaction");
1250 config.targets.insert(new_id.to_owned(), target);
1251 Ok(())
1252 }) {
1253 Ok(result) => result,
1254 Err(error) => {
1255 remove_config_rename_journal()
1256 .context("remove target rename journal after config save failed")?;
1257 return Err(error).context("save renamed target configuration");
1258 }
1259 };
1260 self.config = config;
1261 mj_core::test_hooks::reach_test_hook("config_replacement_before_reference_migration")?;
1262 if let Err(error) = crate::database::rename_target_references(old_id, new_id) {
1263 let restore = Config::update(|config| {
1264 let target = config
1265 .targets
1266 .remove(new_id)
1267 .with_context(|| format!("renamed target {new_id:?} is missing"))?;
1268 ensure!(
1269 !config.targets.contains_key(old_id),
1270 "cannot restore target rename: both {old_id:?} and {new_id:?} exist"
1271 );
1272 config.targets.insert(old_id.to_owned(), target);
1273 Ok(())
1274 });
1275 let restored = match restore {
1276 Ok((config, ())) => config,
1277 Err(restore_error) => {
1278 return Err(error).context(format!(
1279 "rename target references; additionally failed to restore config: {restore_error:#}"
1280 ));
1281 }
1282 };
1283 self.config = restored;
1284 if let Err(restore_error) = remove_config_rename_journal() {
1285 return Err(error).context(format!(
1286 "rename target references; additionally failed to remove rename journal: {restore_error:#}"
1287 ));
1288 }
1289 return Err(error).context("rename target references");
1290 }
1291 let affected: Vec<_> = self
1292 .state
1293 .sessions
1294 .values()
1295 .filter(|session| session.target_template_id == old_id)
1296 .map(|session| session.id.clone())
1297 .collect();
1298 for id in affected {
1299 self.state
1300 .sessions
1301 .get_mut(&id)
1302 .expect("collected session")
1303 .target_template_id = new_id.to_owned();
1304 }
1305 remove_config_rename_journal()?;
1306 Ok(())
1307 }
1308
1309 pub fn recover_config_id_rename() -> Result<bool> {
1313 let path = config_rename_journal_path();
1314 let body = match fs::read(&path) {
1315 Ok(body) => body,
1316 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
1317 Err(error) => return Err(error).context(format!("read {}", path.display())),
1318 };
1319 let journal: ConfigRenameJournal =
1320 serde_json::from_slice(&body).with_context(|| format!("parse {}", path.display()))?;
1321 match journal.kind {
1322 ConfigRenameKind::Profile => {
1323 Config::update(|config| {
1324 finish_config_map_rename(
1325 &mut config.profiles,
1326 &journal.old_id,
1327 &journal.new_id,
1328 "profile",
1329 )?;
1330 Ok(())
1331 })?;
1332 crate::database::rename_profile_references(&journal.old_id, &journal.new_id)?;
1333 }
1334 ConfigRenameKind::Target => {
1335 Config::update_to(&mj_core::config::config_path(), |config| {
1340 finish_config_map_rename(
1341 &mut config.targets,
1342 &journal.old_id,
1343 &journal.new_id,
1344 "target",
1345 )?;
1346 Ok(())
1347 })?;
1348 crate::database::rename_target_references(&journal.old_id, &journal.new_id)?;
1349 }
1350 }
1351 remove_config_rename_journal()?;
1352 Ok(true)
1353 }
1354
1355 pub fn update_session_container_settings(
1359 &mut self,
1360 session_id: &str,
1361 cpus: Option<String>,
1362 memory: Option<String>,
1363 additional_mounts: Vec<targets::AdditionalMount>,
1364 mount_history: Vec<std::path::PathBuf>,
1365 executor: &impl CommandExecutor,
1366 ) -> Result<()> {
1367 let session = self
1368 .state
1369 .sessions
1370 .get(session_id)
1371 .with_context(|| format!("unknown session {session_id}"))?;
1372 for mount in &additional_mounts {
1379 if session
1380 .additional_mounts
1381 .iter()
1382 .any(|attached| attached.source == mount.source)
1383 {
1384 continue;
1385 }
1386 self.validate_mount_source(&session.target_template_id, &mount.source, executor)?;
1387 }
1388 let cpus = cpus.filter(|value| !value.trim().is_empty());
1389 let memory = memory.filter(|value| !value.trim().is_empty());
1390 let updated_at = now();
1391 crate::database::set_session_container_settings(
1392 session_id,
1393 cpus.as_deref(),
1394 memory.as_deref(),
1395 &additional_mounts,
1396 &updated_at,
1397 )?;
1398 if let Some(host) = self
1399 .config
1400 .targets
1401 .get(
1402 &self.state.sessions[session_id]
1403 .target_template_id
1404 .to_owned(),
1405 )
1406 .and_then(mj_core::config::mount_history_host)
1407 {
1408 let host = host.to_owned();
1409 crate::database::replace_mount_history(&host, &mount_history)?;
1412 crate::database::remember_mount_sources(&host, &additional_mounts)?;
1413 self.state.mount_history.insert(host.clone(), mount_history);
1414 self.state.remember_mount_sources(&host, &additional_mounts);
1415 }
1416 let record = self
1417 .state
1418 .sessions
1419 .get_mut(session_id)
1420 .expect("session was checked before updating its container settings");
1421 record.container_cpus = cpus;
1422 record.container_memory = memory;
1423 record.additional_mounts = additional_mounts;
1424 record.updated_at = updated_at;
1425 Ok(())
1426 }
1427}
1428
1429fn config_rename_journal_path() -> PathBuf {
1430 data_dir().join(CONFIG_RENAME_JOURNAL)
1431}
1432
1433fn write_config_rename_journal(journal: &ConfigRenameJournal) -> Result<()> {
1434 let path = config_rename_journal_path();
1435 let body = serde_json::to_vec(journal).context("serialize config rename journal")?;
1436 atomic_write(&path, &body).with_context(|| format!("write {}", path.display()))
1437}
1438
1439fn remove_config_rename_journal() -> Result<()> {
1440 let path = config_rename_journal_path();
1441 match fs::remove_file(&path) {
1442 Ok(()) => Ok(()),
1443 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
1444 Err(error) => Err(error).with_context(|| format!("remove {}", path.display())),
1445 }
1446}
1447
1448fn finish_config_map_rename<T>(
1449 entries: &mut BTreeMap<String, T>,
1450 old_id: &str,
1451 new_id: &str,
1452 kind: &str,
1453) -> Result<()> {
1454 if let Some(entry) = entries.remove(old_id) {
1455 ensure!(
1456 !entries.contains_key(new_id),
1457 "cannot recover {kind} rename: both {old_id:?} and {new_id:?} exist"
1458 );
1459 entries.insert(new_id.to_owned(), entry);
1460 } else {
1461 ensure!(
1462 entries.contains_key(new_id),
1463 "cannot recover {kind} rename: neither {old_id:?} nor {new_id:?} exists"
1464 );
1465 }
1466 Ok(())
1467}
1468
1469#[cfg(test)]
1477pub(crate) fn target_profile_home_for_test(
1478 locator: &targets::TargetLocator,
1479 session_id: &str,
1480 profile: &mj_core::config::HarnessProfile,
1481) -> String {
1482 target_profile_home(locator, session_id, profile)
1483}
1484
1485fn target_profile_home(
1486 locator: &targets::TargetLocator,
1487 session_id: &str,
1488 profile: &mj_core::config::HarnessProfile,
1489) -> String {
1490 let root = removable_profile_root(locator, session_id, profile);
1491 if profile.kind.nested_home() {
1492 PathBuf::from(root)
1493 .join(profile.kind.id())
1494 .to_string_lossy()
1495 .into_owned()
1496 } else {
1497 root
1498 }
1499}
1500
1501pub(super) fn removable_profile_root(
1509 locator: &targets::TargetLocator,
1510 session_id: &str,
1511 profile: &mj_core::config::HarnessProfile,
1512) -> String {
1513 match locator {
1514 targets::TargetLocator::LocalBare { .. } if profile.kind.nested_home() => {
1515 targets::local_nested_profile_root(session_id)
1516 .to_string_lossy()
1517 .into_owned()
1518 }
1519 targets::TargetLocator::LocalBare { worker_root } => Path::new(worker_root)
1520 .join("profile")
1521 .to_string_lossy()
1522 .into_owned(),
1523 targets::TargetLocator::LocalPodman { .. }
1524 | targets::TargetLocator::LocalDocker { .. }
1525 | targets::TargetLocator::AppleContainer { .. }
1526 | targets::TargetLocator::SshPodman { .. }
1527 | targets::TargetLocator::SshDocker { .. } => {
1528 format!("/var/lib/hel/profiles/{session_id}")
1529 }
1530 targets::TargetLocator::AwsEc2 { .. } | targets::TargetLocator::SshBare { .. } => {
1531 format!(".local/share/hel/profiles/{session_id}")
1532 }
1533 }
1534}
1535
1536pub fn resolve_target_input_path(
1538 target: &TargetTemplate,
1539 path: &Path,
1540 executor: &impl CommandExecutor,
1541) -> Result<PathBuf> {
1542 if !mj_core::path_input::needs_home(path)? {
1543 return Ok(path.to_path_buf());
1544 }
1545 let host = cache_host::CacheHost::for_path_target(target)?;
1546 mj_core::path_input::expand_home(path, Some(&host.home(executor)?))
1547}
1548
1549pub fn resolve_machine_input_path(
1552 machine: &mj_core::config::Machine,
1553 path: &Path,
1554 executor: &impl CommandExecutor,
1555) -> Result<PathBuf> {
1556 if !mj_core::path_input::needs_home(path)? {
1557 return Ok(path.to_path_buf());
1558 }
1559 let host = cache_host::CacheHost::for_path_machine(machine)?;
1562 mj_core::path_input::expand_home(path, Some(&host.home(executor)?))
1563}
1564
1565fn execute_checked(executor: &impl CommandExecutor, command: CommandSpec) -> Result<CommandOutput> {
1566 let output = executor.execute(&command)?;
1567 if output.status != 0 {
1568 let detail = command_error_detail(&output.stderr);
1569 if detail.is_empty() {
1570 bail!("{} failed with status {}", command.purpose, output.status);
1571 }
1572 bail!("{detail}");
1573 }
1574 Ok(output)
1575}
1576
1577fn command_error_detail(stderr: &[u8]) -> String {
1578 let reported = String::from_utf8_lossy(stderr);
1579 let reported = reported.trim();
1580 let detail = reported
1581 .rsplit_once("\nCaused by:\n")
1582 .map_or(reported, |(_, causes)| causes);
1583 let detail = detail.strip_prefix("Error: ").unwrap_or(detail);
1584 detail
1585 .lines()
1586 .map(|line| line.strip_prefix(" ").unwrap_or(line))
1587 .collect::<Vec<_>>()
1588 .join("\n")
1589 .trim()
1590 .to_owned()
1591}
1592
1593fn now() -> String {
1594 Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
1595}
1596
1597fn restore_session_after_persistence_failure(
1598 state: &mut State,
1599 session_id: &str,
1600 previous: &SessionRecord,
1601 primary: anyhow::Error,
1602 persist: impl FnOnce(&SessionRecord) -> Result<()>,
1603) -> anyhow::Error {
1604 state
1605 .sessions
1606 .insert(session_id.to_owned(), previous.clone());
1607 let restored = state
1608 .sessions
1609 .get(session_id)
1610 .expect("restored session record disappeared");
1611 match persist(restored) {
1612 Ok(()) => primary,
1613 Err(error) => primary.context(format!(
1614 "restored prior session state in memory, but failed to persist the rollback: {error:#}"
1615 )),
1616 }
1617}
1618
1619fn persist_session_record_transition_or_restore(
1620 state: &mut State,
1621 session_id: &str,
1622 previous: &SessionRecord,
1623 context: &'static str,
1624 persist: &impl Fn(&SessionRecord) -> Result<()>,
1625) -> Result<()> {
1626 let result = persist(
1627 state
1628 .sessions
1629 .get(session_id)
1630 .expect("checkpoint session disappeared before persistence"),
1631 );
1632 match result {
1633 Ok(()) => Ok(()),
1634 Err(error) => Err(restore_session_after_persistence_failure(
1635 state,
1636 session_id,
1637 previous,
1638 error.context(context),
1639 persist,
1640 )),
1641 }
1642}
1643
1644pub fn config_only_controller(config: Config) -> Controller {
1645 Controller {
1646 config,
1647 state: State::default(),
1648 }
1649}
1650
1651#[cfg(test)]
1652mod tests;