1use std::collections::HashMap;
4use std::fs::File;
5use std::io::{ErrorKind, Read, Write};
6use std::path::{Path, PathBuf};
7use std::sync::OnceLock;
8use std::time::Instant;
9
10use anyhow::{Context, Result, bail, ensure};
11use rayon::prelude::*;
12use sha2::{Digest, Sha256};
13
14use crate::session_manager::{
15 ProjectMemorySyncTarget, RemoteWorkerBinaryRefresh, WorkerBinaryRefresh,
16 WorkerBinaryRefreshPlan, WorkerLaunchRefreshPlan, WorkerRecoveryPlan, WorkerWorkspace,
17};
18use crate::targets::{
19 self, CommandExecutor, CommandPlan, CommandSpec, ProcessExecutor, ProvisionStage, SshTarget,
20};
21use mj_core::config::{
22 HarnessKind, HarnessProfile, ProjectBundle, ProjectRepository, atomic_write, data_dir,
23};
24use mj_core::harness_runtime::{
25 CLAUDE_ACP_VERSION, CODEX_ACP_PACKAGE, CODEX_ACP_VERSION, DEEPSEEK_DSH_VERSION,
26};
27use mj_core::project_memory::{ProjectMemoryIdentity, RepositoryMemoryIdentity};
28use mj_core::worker_launch::{
29 HarnessRuntimePolicy, ProjectMemoryLaunchConfig, ProjectMemoryMcpDelivery, WorkerLaunchConfig,
30 WorkerOwnership,
31};
32
33use super::backend::backend_locator;
34use super::readiness::WORKER_EXIT_RECORD_MARKER;
35use super::{Controller, execute_checked, scp_command_spec, ssh_command_spec, target_profile_home};
36
37impl Controller {
38 pub(super) fn worker_placement(
42 &self,
43 session_id: &str,
44 ) -> Result<(targets::TargetLocator, String)> {
45 let session = self
46 .state
47 .sessions
48 .get(session_id)
49 .with_context(|| format!("unknown session {session_id}"))?;
50 let locator = session
51 .target
52 .as_ref()
53 .context("session target is missing")?;
54 let backend = backend_locator(locator, session, &self.config)?;
55 let worker_root = targets::worker_root(&backend, session_id)?;
56 Ok((backend, worker_root))
57 }
58
59 pub(super) fn prepare_worker_files(
60 &self,
61 session_id: &str,
62 backend: &targets::TargetLocator,
63 worker_root: &str,
64 executor: &impl CommandExecutor,
65 ) -> Result<()> {
66 let session = self
67 .state
68 .sessions
69 .get(session_id)
70 .with_context(|| format!("unknown session {session_id}"))?;
71 session.validate_configuration(&self.config)?;
72 let profile = self
73 .config
74 .profiles
75 .get(&session.last_profile)
76 .context("session profile is missing")?;
77 let bundle = session
78 .project_directory
79 .is_none()
80 .then(|| self.config.bundles.get(&session.bundle_id))
81 .flatten();
82 let target = self
83 .config
84 .targets
85 .get(&session.target_template_id)
86 .context("session target template is missing")?;
87 let subagent = crate::database::load_subagent(session_id)?;
88 let workspace_session_id = subagent.as_ref().map_or_else(
89 || session_id.to_owned(),
90 |child| child.parent_session_id.clone(),
91 );
92 let (mut launch, project_memory, target_profile_home) = worker_launch_config(
93 session,
94 profile,
95 bundle,
96 backend,
97 session_id,
98 &workspace_session_id,
99 target,
100 )?;
101 launch.subagent_tools = self.config.subagents.enabled
102 && subagent.is_none()
103 && matches!(
104 session.harness_kind,
105 mj_core::config::HarnessKind::Claude | mj_core::config::HarnessKind::Codex
106 );
107 if let Some(subagent) = &subagent {
108 let parent = self
109 .state
110 .sessions
111 .get(&subagent.parent_session_id)
112 .context("sub-agent parent session is missing")?;
113 let parent_profile = self
114 .config
115 .profiles
116 .get(&parent.last_profile)
117 .context("sub-agent parent profile is missing")?;
118 let parent_target = self
119 .config
120 .targets
121 .get(&parent.target_template_id)
122 .context("sub-agent parent target template is missing")?;
123 let parent_locator = parent
124 .target
125 .as_ref()
126 .context("sub-agent parent has no live target")?;
127 let parent_backend = backend_locator(parent_locator, parent, &self.config)?;
128 let parent_bundle = parent
129 .project_directory
130 .is_none()
131 .then(|| self.config.bundles.get(&parent.bundle_id))
132 .flatten();
133 let (parent_launch, _, _) = worker_launch_config(
134 parent,
135 parent_profile,
136 parent_bundle,
137 &parent_backend,
138 &parent.id,
139 &parent.id,
140 parent_target,
141 )?;
142 launch.cwd = if subagent.working_directory.as_os_str().is_empty() {
143 parent_launch.cwd
144 } else {
145 parent_launch.cwd.join(&subagent.working_directory)
146 };
147 launch.additional_directories = parent_launch.additional_directories;
148 }
149
150 if session.native_session_id.is_some()
151 && profile.kind == mj_core::config::HarnessKind::Codex
152 {
153 launch.goal_resume_request = Some(mj_core::state::new_session_id()?);
154 }
155 let staging = tempfile::tempdir().context("create worker staging directory")?;
156 let launch_path = staging.path().join("launch.json");
157 launch.write(&launch_path)?;
158 let ownership_path = staging.path().join("ownership.json");
159 WorkerOwnership {
160 version: WorkerOwnership::VERSION,
161 workspace_id: session.workspace_id.clone(),
162 session_id: session_id.to_string(),
163 profile_id: session.last_profile.clone(),
164 bundle_id: session.bundle_id.clone(),
165 target_template_id: session.target_template_id.clone(),
166 }
167 .write(&ownership_path)?;
168 let profile_stage = staging.path().join("profile");
169 if !matches!(backend, targets::TargetLocator::LocalBare { .. })
170 || matches!(
171 profile.kind,
172 mj_core::config::HarnessKind::Claude | mj_core::config::HarnessKind::Muse
173 )
174 {
175 let started = Instant::now();
176 let result = stage_profile(profile, &profile_stage);
177 tracing::debug!(
178 session_id,
179 elapsed_ms = started.elapsed().as_millis(),
180 "profile staging completed"
181 );
182 result?;
183 append_hel_target_environment(profile.kind, &profile_stage, backend)?;
184 if launch.subagent_tools && profile.kind == mj_core::config::HarnessKind::Claude {
185 configure_claude_subagent_mcp(&profile_stage, worker_root)?;
186 }
187 stage_memory_replica(
188 &project_memory,
189 Path::new(&target_profile_home),
190 &profile_stage,
191 )?;
192 if project_memory.mcp_delivery == ProjectMemoryMcpDelivery::HarnessProfile {
193 configure_kimi_project_memory_mcp(&profile_stage, worker_root, &project_memory)?;
194 }
195 } else {
196 seed_local_memory_replica(&project_memory)?;
197 }
198 let worker_binary = worker_binary_for(backend, executor)?;
199
200 install_worker_files(
201 executor,
202 backend,
203 session_id,
204 worker_root,
205 &target_profile_home,
206 &worker_binary,
207 &launch_path,
208 &ownership_path,
209 &profile_stage,
210 )?;
211 prepare_installed_managed_harness(executor, backend, worker_root, &launch)
212 }
213
214 pub fn diagnose_worker(&self, session_id: &str) -> Option<String> {
218 self.diagnose_worker_controlled(session_id, &ProcessExecutor)
219 }
220
221 pub fn diagnose_worker_controlled(
222 &self,
223 session_id: &str,
224 executor: &impl CommandExecutor,
225 ) -> Option<String> {
226 let session = self.state.sessions.get(session_id)?;
227 let locator = session.target.as_ref()?;
228 let backend = match backend_locator(locator, session, &self.config) {
229 Ok(backend) => backend,
230 Err(error) => {
231 tracing::debug!(
232 session_id,
233 error = format!("{error:#}"),
234 "could not construct a worker diagnostic probe"
235 );
236 return None;
237 }
238 };
239 let worker_root = match targets::worker_root(&backend, session_id) {
240 Ok(root) => root,
241 Err(error) => {
242 tracing::debug!(
243 session_id,
244 error = format!("{error:#}"),
245 "could not derive the worker diagnostic root"
246 );
247 return None;
248 }
249 };
250 let binary_failure = worker_binary_probe_failure(executor, &backend, &worker_root);
251 let last_words = worker_last_words(executor, &backend, &worker_root);
252 match (binary_failure, last_words) {
253 (Some(binary_failure), Some(last_words)) => {
254 Some(format!("{binary_failure}; {last_words}"))
255 }
256 (Some(binary_failure), None) => Some(binary_failure),
257 (None, last_words) => last_words,
258 }
259 }
260
261 pub fn worker_recovery_plan(&self, session_id: &str) -> Result<WorkerRecoveryPlan> {
265 let (backend, worker_root) = self.worker_placement(session_id)?;
266 let launch = self.current_worker_launch_config(session_id, &backend)?;
267 let workspace = worker_workspace_for_recovery(&backend, &launch.cwd);
268 Ok(WorkerRecoveryPlan {
269 source_target: self.state.sessions[session_id]
270 .target
271 .clone()
272 .context("session target is missing")?,
273 target: targets::target_recovery_plan(&backend, session_id)?,
274 workspace,
275 liveness_probe: worker_liveness_command(&backend, &worker_root),
276 binary_refresh: worker_binary_refresh_plan(&backend, session_id)?,
277 launch_refresh: Some(worker_launch_refresh_plan(&backend, session_id, &launch)?),
278 restart: CommandPlan {
279 description: format!("restart Mjolnir worker for session {session_id}"),
280 commands: vec![
281 stop_worker_command(&backend, &worker_root),
282 start_worker_command(&backend, &worker_root),
283 ],
284 },
285 })
286 }
287
288 pub(super) fn current_worker_launch_config(
289 &self,
290 session_id: &str,
291 backend: &targets::TargetLocator,
292 ) -> Result<WorkerLaunchConfig> {
293 let session = self
294 .state
295 .sessions
296 .get(session_id)
297 .with_context(|| format!("unknown session {session_id}"))?;
298 session.validate_configuration(&self.config)?;
299 let profile = self
300 .config
301 .profiles
302 .get(&session.last_profile)
303 .context("session profile is missing")?;
304 let bundle = session
305 .project_directory
306 .is_none()
307 .then(|| self.config.bundles.get(&session.bundle_id))
308 .flatten();
309 let target = self
310 .config
311 .targets
312 .get(&session.target_template_id)
313 .context("session target template is missing")?;
314 let (mut launch, _, _) = worker_launch_config(
315 session, profile, bundle, backend, session_id, session_id, target,
316 )?;
317 if crate::database::load_move_operation(session_id)?.is_some_and(|operation| {
318 operation.source_checkpoint_only
319 && operation.destination_target.is_none()
320 && matches!(
321 operation.phase,
322 mj_core::state::MovePhase::Preparing
323 | mj_core::state::MovePhase::ClosingSource
324 | mj_core::state::MovePhase::Failed
325 | mj_core::state::MovePhase::Cancelled
326 )
327 && session.last_profile == operation.source_profile_id
328 && session.target == operation.source_target
329 && matches!(
330 session.state,
331 mj_core::state::SessionState::Running
332 | mj_core::state::SessionState::Disconnected
333 | mj_core::state::SessionState::Closing
334 )
335 }) {
336 launch.run_mode = mj_core::worker_launch::WorkerRunMode::CheckpointOnly;
337 }
338 Ok(launch)
339 }
340
341 pub fn project_memory_sync_target(&self, session_id: &str) -> Result<ProjectMemorySyncTarget> {
342 let session = self
343 .state
344 .sessions
345 .get(session_id)
346 .with_context(|| format!("unknown session {session_id}"))?;
347 session.validate_configuration(&self.config)?;
348 let locator = session
349 .target
350 .as_ref()
351 .context("session target is missing")?;
352 let backend = backend_locator(locator, session, &self.config)?;
353 let profile = self
354 .config
355 .profiles
356 .get(&session.last_profile)
357 .context("session profile is missing")?;
358 let bundle = session
359 .project_directory
360 .is_none()
361 .then(|| self.config.bundles.get(&session.bundle_id))
362 .flatten();
363 let workspace = if let Some(project_directory) = &session.project_directory {
364 (project_directory.to_string_lossy().into_owned(), Vec::new())
365 } else {
366 workspace_paths(
367 &backend,
368 bundle.context("session bundle is missing")?,
369 session_id,
370 )?
371 };
372 let target_home = target_profile_home(&backend, session_id, profile);
373 let launch = project_memory_launch(session, bundle, &workspace, &target_home)?;
374 Ok(ProjectMemorySyncTarget {
375 canonical_root: canonical_memory_root(&launch.project_key),
376 })
377 }
378}
379
380fn worker_workspace_for_recovery(
381 backend: &targets::TargetLocator,
382 directory: &Path,
383) -> Option<WorkerWorkspace> {
384 let target = match backend {
385 targets::TargetLocator::LocalBare { .. } => mj_core::state::ManagedWorktreeTarget::Local,
386 targets::TargetLocator::SshBare { ssh, .. } => mj_core::state::ManagedWorktreeTarget::Ssh {
387 destination: ssh.destination.clone(),
388 ssh_args: ssh.ssh_args.clone(),
389 },
390 targets::TargetLocator::LocalPodman { .. }
391 | targets::TargetLocator::LocalDocker { .. }
392 | targets::TargetLocator::AppleContainer { .. }
393 | targets::TargetLocator::AwsEc2 { .. }
394 | targets::TargetLocator::SshPodman { .. }
395 | targets::TargetLocator::SshDocker { .. } => return None,
396 };
397 Some(WorkerWorkspace {
398 target,
399 directory: directory.to_path_buf(),
400 })
401}
402
403fn worker_launch_config(
404 session: &mj_core::state::SessionRecord,
405 profile: &mj_core::config::HarnessProfile,
406 bundle: Option<&ProjectBundle>,
407 backend: &targets::TargetLocator,
408 session_id: &str,
409 workspace_session_id: &str,
410 target: &mj_core::config::TargetTemplate,
411) -> Result<(WorkerLaunchConfig, ProjectMemoryLaunchConfig, String)> {
412 let execution_policy = target.execution_policy();
413 let target_profile_home = target_profile_home(backend, session_id, profile);
414 let workspace = if let Some(project_directory) = &session.project_directory {
415 (project_directory.to_string_lossy().into_owned(), Vec::new())
416 } else {
417 workspace_paths(
418 backend,
419 bundle.context("session bundle is missing")?,
420 workspace_session_id,
421 )?
422 };
423 let mut additional_directories = workspace.1.iter().map(PathBuf::from).collect::<Vec<_>>();
424 additional_directories.extend(
425 session
426 .additional_mounts
427 .iter()
428 .map(|resource| resource.destination.clone()),
429 );
430 if matches!(
431 profile.kind,
432 mj_core::config::HarnessKind::Deepseek | mj_core::config::HarnessKind::Muse
433 ) && !additional_directories.is_empty()
434 {
435 bail!(
436 "{} ACP does not support multiple workspace roots; use a single-repository bundle",
437 profile.kind.display_name()
438 );
439 }
440 let (bridge_command, bridge_args) = bridge_launch(profile.kind, execution_policy);
441 use mj_core::config::TargetTemplate;
442 let target_environment = match target {
443 TargetTemplate::LocalPodman { container }
444 | TargetTemplate::LocalDocker { container }
445 | TargetTemplate::AppleContainer { container }
446 | TargetTemplate::SshPodman { container, .. }
447 | TargetTemplate::SshDocker { container, .. } => container.environment.clone(),
448 _ => Default::default(),
449 };
450 let mut environment = target_environment.clone();
451 environment.extend(profile.environment.clone());
452 profile
453 .kind
454 .configure_home_environment(Path::new(&target_profile_home), &mut environment);
455 profile
456 .kind
457 .configure_execution_environment(execution_policy, &mut environment)?;
458 environment.remove(mj_core::worker_launch::DISCOVER_LOGIN_PATH_ENV);
459 let mut project_memory =
460 project_memory_launch(session, bundle, &workspace, &target_profile_home)?;
461 project_memory.mcp_delivery = project_memory_mcp_delivery(profile.kind, backend);
462 if profile.kind == mj_core::config::HarnessKind::Claude {
463 environment.insert(
464 "CLAUDE_CODE_PROJECT_DIR_NAME".into(),
465 project_memory_replica_slug(&project_memory.project_key, session_id),
466 );
467 }
468 apply_claude_setup_token(
469 &mut environment,
470 profile.kind,
471 &mj_core::credentials::claude_oauth_token_path(&session.last_profile),
472 );
473 Ok((
474 WorkerLaunchConfig {
475 goal_resume_request: None,
476 target_environment,
477 run_mode: Default::default(),
478 session_id: session_id.to_string(),
479 subagent_tools: false,
480 harness: profile.kind,
481 bridge_command: PathBuf::from(bridge_command),
482 bridge_args,
483 harness_runtime: harness_runtime_policy(backend),
484 environment,
485 cwd: PathBuf::from(&workspace.0),
486 additional_directories,
487 native_session_id: session.native_session_id.clone(),
488 project_memory: profile
489 .kind
490 .supports_injected_mcp()
491 .then(|| project_memory.clone()),
492 execution_policy,
493 },
494 project_memory,
495 target_profile_home,
496 ))
497}
498
499fn harness_runtime_policy(backend: &targets::TargetLocator) -> HarnessRuntimePolicy {
500 match backend {
501 targets::TargetLocator::LocalBare { .. }
502 | targets::TargetLocator::AwsEc2 { .. }
503 | targets::TargetLocator::SshBare { .. } => HarnessRuntimePolicy::Managed,
504 _ => HarnessRuntimePolicy::Ambient,
505 }
506}
507
508pub(super) fn apply_claude_setup_token(
515 environment: &mut std::collections::BTreeMap<String, String>,
516 kind: mj_core::config::HarnessKind,
517 token_path: &Path,
518) {
519 use mj_core::credentials::CLAUDE_OAUTH_TOKEN_ENV;
520
521 if kind != mj_core::config::HarnessKind::Claude
522 || environment.contains_key(CLAUDE_OAUTH_TOKEN_ENV)
523 {
524 return;
525 }
526 match mj_core::credentials::read_claude_oauth_token(token_path) {
527 Ok(Some(token)) => {
528 environment.insert(CLAUDE_OAUTH_TOKEN_ENV.to_owned(), token);
529 }
530 Ok(None) => {}
531 Err(error) => tracing::warn!(
534 path = %token_path.display(),
535 %error,
536 "ignoring an unreadable Claude setup token"
537 ),
538 }
539}
540
541fn project_memory_launch(
542 session: &mj_core::state::SessionRecord,
543 bundle: Option<&ProjectBundle>,
544 workspace: &(String, Vec<String>),
545 target_profile_home: &str,
546) -> Result<ProjectMemoryLaunchConfig> {
547 let identity = if let Some(worktree) = &session.managed_worktree {
548 ProjectMemoryIdentity::Repository {
549 repository: RepositoryMemoryIdentity::Local {
550 canonical_root: std::fs::canonicalize(&worktree.source_repository)
551 .unwrap_or_else(|_| worktree.source_repository.clone()),
552 },
553 }
554 } else if let Some(bundle) = bundle {
555 let primary =
556 configured_memory_identity(bundle.primary().context("bundle primary is missing")?)?;
557 let members = bundle
558 .repositories
559 .iter()
560 .map(configured_memory_identity)
561 .collect::<Result<Vec<_>>>()?;
562 ProjectMemoryIdentity::bundle(primary, members)
563 } else {
564 let project = session
565 .project_directory
566 .as_ref()
567 .context("raw session project directory is missing")?;
568 let repository = match session.target.as_ref() {
569 Some(mj_core::state::TargetLocator::LocalBare { .. }) => {
570 RepositoryMemoryIdentity::Local {
571 canonical_root: std::fs::canonicalize(project)
572 .unwrap_or_else(|_| project.clone()),
573 }
574 }
575 _ => RepositoryMemoryIdentity::Remote {
576 target: session.target_template_id.clone(),
577 canonical_root: project.clone(),
578 },
579 };
580 ProjectMemoryIdentity::Repository { repository }
581 };
582 let project_key = identity.key()?;
583 let replica_slug = project_memory_replica_slug(&project_key, &session.id);
584 let project_root = PathBuf::from(target_profile_home)
585 .join("projects")
586 .join(replica_slug);
587 let root = project_root.join("memory");
588 let baseline_root = project_root.join(".hel-memory-baseline");
589 let mut repository_roots = std::collections::BTreeMap::new();
590 if let Some(bundle) = bundle {
591 let target_roots =
592 std::iter::once(workspace.0.as_str()).chain(workspace.1.iter().map(String::as_str));
593 let repositories = std::iter::once(bundle.primary().context("bundle primary is missing")?)
594 .chain(
595 bundle
596 .repositories
597 .iter()
598 .filter(|repository| repository.id != bundle.primary_repo),
599 );
600 repository_roots.extend(
601 repositories
602 .zip(target_roots)
603 .map(|(repository, root)| (repository.id.clone(), PathBuf::from(root))),
604 );
605 }
606 Ok(ProjectMemoryLaunchConfig {
607 project_key,
608 root,
609 baseline_root,
610 repository_roots,
611 mcp_delivery: ProjectMemoryMcpDelivery::Acp,
612 })
613}
614
615fn project_memory_replica_slug(project_key: &str, session_id: &str) -> String {
616 format!("hel-{}-{session_id}", &project_key[..16])
617}
618
619fn project_memory_mcp_delivery(
620 harness: mj_core::config::HarnessKind,
621 target: &targets::TargetLocator,
622) -> ProjectMemoryMcpDelivery {
623 if harness == mj_core::config::HarnessKind::Kimi
624 && !matches!(target, targets::TargetLocator::LocalBare { .. })
625 {
626 ProjectMemoryMcpDelivery::HarnessProfile
627 } else {
628 ProjectMemoryMcpDelivery::Acp
629 }
630}
631
632fn configured_memory_identity(repository: &ProjectRepository) -> Result<RepositoryMemoryIdentity> {
633 if let Some(source) = repository.github.as_deref() {
634 let github = crate::setup::github_repository_from_origin(source)
635 .with_context(|| format!("parse repository source {source:?} for project memory"))?;
636 return Ok(RepositoryMemoryIdentity::Github {
637 owner: github.owner.to_ascii_lowercase(),
638 repository: github.repository.to_ascii_lowercase(),
639 });
640 }
641 let root = repository
642 .local
643 .as_ref()
644 .context("project repository has no source for memory identity")?;
645 Ok(RepositoryMemoryIdentity::Local {
646 canonical_root: mj_core::local_git::main_worktree_root(root)
647 .or_else(|_| std::fs::canonicalize(root).map_err(anyhow::Error::from))
648 .unwrap_or_else(|_| root.clone()),
649 })
650}
651
652fn canonical_memory_root(project_key: &str) -> PathBuf {
653 data_dir().join("projects").join(project_key).join("memory")
654}
655
656fn stage_memory_replica(
657 memory: &ProjectMemoryLaunchConfig,
658 target_profile_home: &Path,
659 profile_stage: &Path,
660) -> Result<()> {
661 let canonical = canonical_memory_root(&memory.project_key);
662 std::fs::create_dir_all(&canonical)?;
663 let replica = memory.root.strip_prefix(target_profile_home)?;
664 let baseline = memory.baseline_root.strip_prefix(target_profile_home)?;
665 copy_profile_entry(&canonical, &profile_stage.join(replica))?;
666 copy_profile_entry(&canonical, &profile_stage.join(baseline))
667}
668
669fn seed_local_memory_replica(memory: &ProjectMemoryLaunchConfig) -> Result<()> {
670 let canonical = canonical_memory_root(&memory.project_key);
671 std::fs::create_dir_all(&canonical)?;
672 let canonical_has_files = directory_has_files(&canonical)?;
673 let replica_has_files = directory_has_files(&memory.root)?;
674 match (canonical_has_files, replica_has_files) {
675 (false, true) => copy_profile_entry(&memory.root, &canonical),
676 (true, false) => copy_profile_entry(&canonical, &memory.root),
677 _ => Ok(()),
678 }?;
679 copy_profile_entry(&canonical, &memory.baseline_root)
680}
681
682fn configure_kimi_project_memory_mcp(
686 profile_stage: &Path,
687 worker_root: &str,
688 memory: &ProjectMemoryLaunchConfig,
689) -> Result<()> {
690 let path = profile_stage.join("mcp.json");
691 let mut document = match std::fs::read(&path) {
692 Ok(body) => serde_json::from_slice::<serde_json::Value>(&body)
693 .with_context(|| format!("parse staged Kimi MCP configuration {}", path.display()))?,
694 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
695 serde_json::Value::Object(serde_json::Map::new())
696 }
697 Err(error) => {
698 return Err(error)
699 .with_context(|| format!("read staged Kimi MCP configuration {}", path.display()));
700 }
701 };
702 let root = document.as_object_mut().with_context(|| {
703 format!(
704 "staged Kimi MCP configuration {} must contain a JSON object",
705 path.display()
706 )
707 })?;
708 let servers = root
709 .entry("mcpServers")
710 .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()))
711 .as_object_mut()
712 .with_context(|| {
713 format!(
714 "mcpServers in staged Kimi MCP configuration {} must be a JSON object",
715 path.display()
716 )
717 })?;
718
719 let worker = Path::new(worker_root).join("hel");
720 let server = if worker.is_absolute() && memory.root.is_absolute() {
721 serde_json::json!({
722 "transport": "stdio",
723 "command": worker,
724 "args": ["worker", "memory-mcp", "--root", memory.root],
725 "runtime_id": "local"
726 })
727 } else {
728 let worker = worker.to_string_lossy();
729 let memory_root = memory.root.to_string_lossy();
730 serde_json::json!({
731 "transport": "stdio",
732 "command": "sh",
733 "args": [
734 "-c",
735 "exec \"$HOME/$1\" worker memory-mcp --root \"$HOME/$2\"",
736 "mj-project-memory",
737 worker,
738 memory_root
739 ],
740 "runtime_id": "local"
741 })
742 };
743 servers.insert("mj-project-memory".into(), server);
744 let mut body = serde_json::to_vec_pretty(&document)?;
745 body.push(b'\n');
746 atomic_write(&path, &body)
747 .with_context(|| format!("write staged Kimi MCP configuration {}", path.display()))
748}
749
750fn configure_claude_subagent_mcp(profile_stage: &Path, worker_root: &str) -> Result<()> {
754 let path = profile_stage.join(".claude.json");
755 let mut document = match std::fs::read(&path) {
756 Ok(body) => serde_json::from_slice::<serde_json::Value>(&body)
757 .with_context(|| format!("parse staged Claude configuration {}", path.display()))?,
758 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
759 serde_json::Value::Object(serde_json::Map::new())
760 }
761 Err(error) => {
762 return Err(error)
763 .with_context(|| format!("read staged Claude configuration {}", path.display()));
764 }
765 };
766 let root = document.as_object_mut().with_context(|| {
767 format!(
768 "staged Claude configuration {} must contain a JSON object",
769 path.display()
770 )
771 })?;
772 let servers = root
773 .entry("mcpServers")
774 .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()))
775 .as_object_mut()
776 .with_context(|| {
777 format!(
778 "mcpServers in staged Claude configuration {} must be a JSON object",
779 path.display()
780 )
781 })?;
782 servers.insert(
783 "mj-subagents".into(),
784 serde_json::json!({
785 "type":"stdio",
786 "command":Path::new(worker_root).join("hel"),
787 "args":[
788 "worker",
789 "subagent-mcp",
790 "--socket",
791 Path::new(worker_root).join(mj_worker_socket_name())
792 ]
793 }),
794 );
795 let mut body = serde_json::to_vec_pretty(&document)?;
796 body.push(b'\n');
797 atomic_write(&path, &body)
798 .with_context(|| format!("write staged Claude configuration {}", path.display()))
799}
800
801fn mj_worker_socket_name() -> &'static str {
802 "subagents.sock"
803}
804
805fn directory_has_files(path: &Path) -> Result<bool> {
806 let entries = match std::fs::read_dir(path) {
807 Ok(entries) => entries,
808 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
809 Err(error) => return Err(error.into()),
810 };
811 for entry in entries {
812 let entry = entry?;
813 let metadata = entry.metadata()?;
814 if metadata.is_file() || (metadata.is_dir() && directory_has_files(&entry.path())?) {
815 return Ok(true);
816 }
817 }
818 Ok(false)
819}
820
821#[derive(Debug, Clone, PartialEq, Eq)]
822pub enum WorkerBinaryAvailability {
823 Local {
824 path: PathBuf,
825 source: String,
826 },
827 Remote {
828 url: String,
829 sha256: String,
830 triple: String,
831 },
832}
833
834#[derive(Debug)]
840struct WorkerBinarySourceSnapshot {
841 entries: HashMap<
842 (String, WorkerBinaryRequirement),
843 std::result::Result<WorkerBinaryAvailability, String>,
844 >,
845}
846
847static PINNED_WORKER_BINARY_SOURCES: OnceLock<WorkerBinarySourceSnapshot> = OnceLock::new();
848
849fn packaged_worker_binary_path(directory: &Path, triple: &str) -> PathBuf {
850 directory.join(format!("mj-worker-{triple}"))
851}
852
853fn running_executable_file_name(controller: &Path) -> Option<std::ffi::OsString> {
858 let name = controller.file_name()?;
859 #[cfg(target_os = "linux")]
860 {
861 use std::os::unix::ffi::{OsStrExt, OsStringExt};
862
863 if let Some(name) = name.as_bytes().strip_suffix(b" (deleted)") {
864 return Some(std::ffi::OsString::from_vec(name.to_vec()));
865 }
866 }
867 Some(name.to_os_string())
868}
869
870fn worker_sibling_names(controller: &Path) -> Vec<std::ffi::OsString> {
875 use std::ffi::OsString;
876 let mut names = Vec::new();
877 if let Some(own) = running_executable_file_name(controller) {
878 names.push(own);
879 }
880 let legacy = OsString::from("hel");
881 if !names.contains(&legacy) {
882 names.push(legacy);
883 }
884 names
885}
886
887fn select_native_worker(
891 controller: &Path,
892 is_file: impl Fn(&Path) -> bool,
893) -> Option<(PathBuf, &'static str)> {
894 let directory = controller.parent()?;
895 if let (Some(profile), Some(target_dir)) = (directory.file_name(), directory.parent()) {
896 let development_worker = target_dir.join("worker").join(profile).join("mj-worker");
897 if is_file(&development_worker) {
898 return Some((development_worker, "isolated native development worker"));
899 }
900 }
901 let packaged_worker = directory.join("mj-worker");
902 is_file(&packaged_worker).then_some((packaged_worker, "native worker beside mj"))
903}
904
905fn select_sibling_worker(
912 controller: &Path,
913 triple: &str,
914 is_file: impl Fn(&Path) -> bool,
915) -> Option<(PathBuf, &'static str)> {
916 let directory = controller.parent()?;
917 let names = worker_sibling_names(controller);
918 let mut candidates: Vec<(PathBuf, &'static str)> = Vec::new();
919 candidates.push((
921 packaged_worker_binary_path(directory, triple),
922 "beside the mj binary",
923 ));
924 if let (Some(profile), Some(target_dir)) = (directory.file_name(), directory.parent()) {
930 candidates.push((
931 target_dir
932 .join("worker")
933 .join(triple)
934 .join(profile)
935 .join("mj-worker"),
936 "isolated development musl worker",
937 ));
938 candidates.push((
939 target_dir.join(triple).join(profile).join("mj-worker"),
940 "development musl worker",
941 ));
942 for name in &names {
943 candidates.push((
944 target_dir.join(triple).join(profile).join(name),
945 "development musl sibling",
946 ));
947 }
948 }
949 let controller_name = running_executable_file_name(controller);
954 for name in names
955 .iter()
956 .filter(|name| Some(name.as_os_str()) != controller_name.as_deref())
957 {
958 candidates.push((directory.join(name), "beside the running executable"));
959 }
960 candidates.into_iter().find(|(path, _)| is_file(path))
961}
962
963#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
964enum WorkerBinaryRequirement {
965 PortableLinux,
966 LocalHost,
967}
968
969impl WorkerBinarySourceSnapshot {
970 fn capture<F>(cache_root: &Path, resolve: F) -> Self
971 where
972 F: Fn(&str, WorkerBinaryRequirement) -> Result<WorkerBinaryAvailability>,
973 {
974 let mut entries = HashMap::new();
975 let mut local_cache = HashMap::<PathBuf, PathBuf>::new();
976 let architectures = [
977 (std::env::consts::ARCH, WorkerBinaryRequirement::LocalHost),
978 ("x86_64", WorkerBinaryRequirement::PortableLinux),
979 ("aarch64", WorkerBinaryRequirement::PortableLinux),
980 ];
981
982 for (arch, requirement) in architectures {
983 let pinned = match resolve(arch, requirement) {
984 Ok(WorkerBinaryAvailability::Local { path, source }) => {
985 match local_cache.get(&path).cloned().map(Ok).unwrap_or_else(|| {
986 copy_worker_source_to_cache(&path, cache_root).inspect(|cached| {
987 local_cache.insert(path.clone(), cached.clone());
988 })
989 }) {
990 Ok(cached) => Ok(WorkerBinaryAvailability::Local {
991 path: cached,
992 source,
993 }),
994 Err(error) => {
995 let error = format!(
996 "pin worker source {} for {arch} ({requirement:?}): {error:#}",
997 path.display()
998 );
999 tracing::warn!(arch, requirement = ?requirement, error = %error);
1000 Err(error)
1001 }
1002 }
1003 }
1004 Ok(WorkerBinaryAvailability::Remote {
1005 url,
1006 sha256,
1007 triple,
1008 }) => Ok(WorkerBinaryAvailability::Remote {
1009 url,
1010 sha256,
1011 triple,
1012 }),
1013 Err(error) => {
1014 let error = format!("{error:#}");
1015 tracing::debug!(
1016 arch,
1017 requirement = ?requirement,
1018 error = %error,
1019 "worker source was unavailable when the daemon started"
1020 );
1021 Err(error)
1022 }
1023 };
1024 entries.insert((arch.to_owned(), requirement), pinned);
1025 }
1026
1027 Self { entries }
1028 }
1029
1030 fn resolve(
1031 &self,
1032 arch: &str,
1033 requirement: WorkerBinaryRequirement,
1034 ) -> Result<WorkerBinaryAvailability> {
1035 let Some(source) = self.entries.get(&(arch.to_owned(), requirement)) else {
1036 bail!(
1037 "worker source for {arch} ({requirement:?}) was not captured when the daemon started"
1038 );
1039 };
1040 match source {
1041 Ok(availability) => Ok(availability.clone()),
1042 Err(error) => bail!(
1043 "worker source for {arch} ({requirement:?}) was unavailable when the daemon started; install it and restart the daemon to retry: {error}"
1044 ),
1045 }
1046 }
1047}
1048
1049pub fn pin_worker_binary_sources() -> Result<()> {
1053 if PINNED_WORKER_BINARY_SOURCES.get().is_some() {
1054 return Ok(());
1055 }
1056 let current = std::env::current_exe().context("resolve Mjolnir controller binary")?;
1057 let cache_root = data_dir().join("workers").join("pinned");
1058 let started = std::time::Instant::now();
1059 let snapshot = WorkerBinarySourceSnapshot::capture(&cache_root, |arch, requirement| {
1060 worker_binary_prerequisite_for_current(arch, requirement, ¤t, &|path| path.is_file())
1061 });
1062 tracing::info!(
1063 elapsed_ms = started.elapsed().as_millis(),
1064 "worker sources pinned"
1065 );
1066 let _ = PINNED_WORKER_BINARY_SOURCES.set(snapshot);
1069 Ok(())
1070}
1071
1072fn copy_worker_source_to_cache(source: &Path, cache_root: &Path) -> Result<PathBuf> {
1073 std::fs::create_dir_all(cache_root)
1074 .with_context(|| format!("create pinned worker cache {}", cache_root.display()))?;
1075 let mut input =
1076 File::open(source).with_context(|| format!("open worker source {}", source.display()))?;
1077 let metadata = input
1078 .metadata()
1079 .with_context(|| format!("stat worker source {}", source.display()))?;
1080 let mut temporary = tempfile::NamedTempFile::new_in(cache_root)
1081 .with_context(|| format!("create pinned worker staging file {}", cache_root.display()))?;
1082 let mut digest = Sha256::new();
1083 let mut buffer = [0_u8; 128 * 1024];
1084 loop {
1085 let count = input
1086 .read(&mut buffer)
1087 .with_context(|| format!("read worker source {}", source.display()))?;
1088 if count == 0 {
1089 break;
1090 }
1091 temporary
1092 .write_all(&buffer[..count])
1093 .with_context(|| format!("copy worker source {}", source.display()))?;
1094 digest.update(&buffer[..count]);
1095 }
1096 temporary
1097 .as_file_mut()
1098 .sync_all()
1099 .with_context(|| format!("flush pinned worker source {}", source.display()))?;
1100 std::fs::set_permissions(temporary.path(), metadata.permissions())
1101 .with_context(|| format!("preserve permissions for {}", source.display()))?;
1102 let digest = format!("{:x}", digest.finalize());
1103 publish_cached_worker(temporary, cache_root, &digest)
1104}
1105
1106fn publish_cached_worker(
1110 temporary: tempfile::NamedTempFile,
1111 cache_root: &Path,
1112 digest: &str,
1113) -> Result<PathBuf> {
1114 let directory = cache_root.join(digest);
1115 std::fs::create_dir_all(&directory)
1116 .with_context(|| format!("create pinned worker cache {}", directory.display()))?;
1117 let destination = directory.join("hel");
1118 if destination.is_file() {
1119 return Ok(destination);
1120 }
1121 match temporary.persist_noclobber(&destination) {
1122 Ok(_) => {
1123 #[cfg(unix)]
1124 File::open(&directory)
1125 .and_then(|directory| directory.sync_all())
1126 .with_context(|| format!("flush pinned worker cache {}", directory.display()))?;
1127 Ok(destination)
1128 }
1129 Err(error) if error.error.kind() == ErrorKind::AlreadyExists => {
1130 if destination.is_file() {
1131 Ok(destination)
1132 } else {
1133 Err(error.error).with_context(|| {
1134 format!("publish pinned worker artifact {}", destination.display())
1135 })
1136 }
1137 }
1138 Err(error) => Err(error.error)
1139 .with_context(|| format!("publish pinned worker artifact {}", destination.display())),
1140 }
1141}
1142
1143pub fn worker_binary_prerequisite_for_arch(arch: &str) -> Result<WorkerBinaryAvailability> {
1150 worker_binary_for_arch(arch, WorkerBinaryRequirement::PortableLinux)
1151}
1152
1153fn worker_binary_for_arch(
1154 arch: &str,
1155 requirement: WorkerBinaryRequirement,
1156) -> Result<WorkerBinaryAvailability> {
1157 if let Some(snapshot) = PINNED_WORKER_BINARY_SOURCES.get() {
1158 return snapshot.resolve(arch, requirement);
1159 }
1160 let current = std::env::current_exe().context("resolve Mjolnir controller binary")?;
1161 worker_binary_prerequisite_for_current(arch, requirement, ¤t, &|path| path.is_file())
1162}
1163
1164fn worker_binary_prerequisite_for_current(
1167 arch: &str,
1168 requirement: WorkerBinaryRequirement,
1169 current: &Path,
1170 is_file: &dyn Fn(&Path) -> bool,
1171) -> Result<WorkerBinaryAvailability> {
1172 let triple = format!("{arch}-unknown-linux-musl");
1173 if let Some(path) = mj_core::config::env_override_os("WORKER_BINARY").map(PathBuf::from) {
1174 if !is_file(&path) {
1175 bail!("MJ_WORKER_BINARY is not a file: {}", path.display());
1176 }
1177 return Ok(WorkerBinaryAvailability::Local {
1178 path,
1179 source: "MJ_WORKER_BINARY".into(),
1180 });
1181 }
1182 let controller_replaced = !is_file(current);
1186 let mut candidates = Vec::new();
1187 if let Some(directory) = mj_core::config::env_override_os("WORKER_DIR").map(PathBuf::from) {
1188 candidates.push((
1189 packaged_worker_binary_path(&directory, &triple),
1190 "MJ_WORKER_DIR",
1191 ));
1192 candidates.push((directory.join(&triple).join("hel"), "MJ_WORKER_DIR"));
1193 }
1194 if let Some((path, source)) = candidates.into_iter().find(|(path, _)| is_file(path)) {
1195 return Ok(WorkerBinaryAvailability::Local {
1196 path,
1197 source: source.into(),
1198 });
1199 }
1200 if requirement == WorkerBinaryRequirement::LocalHost
1201 && let Some((path, source)) = select_native_worker(current, is_file)
1202 {
1203 return Ok(WorkerBinaryAvailability::Local {
1204 path,
1205 source: source.into(),
1206 });
1207 }
1208 if !controller_replaced
1209 && let Some((path, source)) = select_sibling_worker(current, &triple, is_file)
1210 {
1211 return Ok(WorkerBinaryAvailability::Local {
1212 path,
1213 source: source.into(),
1214 });
1215 }
1216 if let Some(template) = mj_core::config::env_override("WORKER_URL") {
1217 let expected = mj_core::config::env_override("WORKER_SHA256")
1218 .context("MJ_WORKER_URL requires MJ_WORKER_SHA256")?;
1219 validate_worker_sha256(&expected)?;
1220 return Ok(WorkerBinaryAvailability::Remote {
1221 url: template.replace("{target}", &triple),
1222 sha256: expected,
1223 triple,
1224 });
1225 }
1226 ensure!(
1229 !controller_replaced,
1230 "the running mj binary was replaced or removed on disk ({}); restart the Mjolnir daemon so it runs the current build, then retry",
1231 display_path(current)
1232 );
1233 bail!(
1234 "no Linux worker for {triple}; install mj-worker-{triple} beside mj, set MJ_WORKER_DIR/MJ_WORKER_BINARY, or configure MJ_WORKER_URL and MJ_WORKER_SHA256"
1235 )
1236}
1237
1238fn display_path(path: &Path) -> String {
1241 let text = path.to_string_lossy();
1242 text.strip_suffix(" (deleted)").unwrap_or(&text).to_owned()
1243}
1244
1245fn template_architecture(template: &mj_core::config::TargetTemplate) -> Option<&'static str> {
1252 use mj_core::config::TargetTemplate as Template;
1253 let platform = match template {
1254 Template::LocalPodman { container }
1255 | Template::LocalDocker { container }
1256 | Template::AppleContainer { container }
1257 | Template::SshPodman { container, .. }
1258 | Template::SshDocker { container, .. } => container.platform.as_deref()?,
1259 Template::LocalBare | Template::SshBare { .. } | Template::AwsEc2 { .. } => return None,
1260 };
1261 platform.split('/').find_map(|part| match part.trim() {
1263 "x86_64" | "amd64" => Some("x86_64"),
1264 "aarch64" | "arm64" => Some("aarch64"),
1265 _ => None,
1266 })
1267}
1268
1269fn preflight_architectures(template: &mj_core::config::TargetTemplate) -> Vec<&'static str> {
1276 use mj_core::config::TargetTemplate as Template;
1277 if let Some(arch) = template_architecture(template) {
1278 return vec![arch];
1279 }
1280 match template {
1281 Template::LocalBare
1282 | Template::LocalPodman { .. }
1283 | Template::LocalDocker { .. }
1284 | Template::AppleContainer { .. } => vec![std::env::consts::ARCH],
1285 Template::SshBare { .. }
1286 | Template::SshPodman { .. }
1287 | Template::SshDocker { .. }
1288 | Template::AwsEc2 { .. } => {
1289 vec!["x86_64", "aarch64"]
1290 }
1291 }
1292}
1293
1294pub(super) fn preflight_worker_binary(template: &mj_core::config::TargetTemplate) -> Result<()> {
1303 let requirement = if matches!(template, mj_core::config::TargetTemplate::LocalBare) {
1306 WorkerBinaryRequirement::LocalHost
1307 } else {
1308 WorkerBinaryRequirement::PortableLinux
1309 };
1310 let mut failure = None;
1311 for arch in preflight_architectures(template) {
1312 match worker_binary_for_arch(arch, requirement) {
1313 Ok(_) => return Ok(()),
1314 Err(error) => failure = Some(error),
1315 }
1316 }
1317 match failure {
1318 Some(error) => Err(error).context("preflight the worker binary before resuming"),
1321 None => Ok(()),
1322 }
1323}
1324
1325pub(super) fn worker_binary_for(
1326 locator: &targets::TargetLocator,
1327 executor: &impl CommandExecutor,
1328) -> Result<PathBuf> {
1329 let arch = target_architecture(locator, executor)?;
1330 let requirement = if matches!(locator, targets::TargetLocator::LocalBare { .. }) {
1331 WorkerBinaryRequirement::LocalHost
1332 } else {
1333 WorkerBinaryRequirement::PortableLinux
1334 };
1335 match worker_binary_for_arch(arch, requirement)? {
1336 WorkerBinaryAvailability::Local { path, .. } => Ok(path),
1337 WorkerBinaryAvailability::Remote {
1338 url,
1339 sha256,
1340 triple,
1341 } => download_worker(&url, &sha256, &triple),
1342 }
1343}
1344
1345fn target_architecture(
1346 locator: &targets::TargetLocator,
1347 executor: &impl CommandExecutor,
1348) -> Result<&'static str> {
1349 let command = match locator {
1350 targets::TargetLocator::LocalBare { .. } => CommandSpec::new("uname", ["-m"]),
1351 targets::TargetLocator::LocalPodman { container_id, .. } => {
1352 CommandSpec::new("podman", ["exec", container_id, "uname", "-m"])
1353 }
1354 targets::TargetLocator::LocalDocker { container_id } => {
1355 CommandSpec::new("docker", ["exec", container_id, "uname", "-m"])
1356 }
1357 targets::TargetLocator::AppleContainer { container_id } => {
1358 CommandSpec::new("container", ["exec", container_id, "uname", "-m"])
1359 }
1360 targets::TargetLocator::AwsEc2 { ssh, .. }
1361 | targets::TargetLocator::SshBare { ssh, .. } => ssh_command_spec(ssh, ["uname", "-m"]),
1362 targets::TargetLocator::SshPodman {
1363 ssh, container_id, ..
1364 } => ssh_command_spec(ssh, ["podman", "exec", container_id, "uname", "-m"]),
1365 targets::TargetLocator::SshDocker { ssh, container_id } => {
1366 ssh_command_spec(ssh, ["docker", "exec", container_id, "uname", "-m"])
1367 }
1368 }
1369 .purpose("detect target architecture");
1370 let output = execute_checked(executor, command)?;
1371 match String::from_utf8(output.stdout)?.trim() {
1372 "x86_64" | "amd64" => Ok("x86_64"),
1373 "aarch64" | "arm64" => Ok("aarch64"),
1374 architecture => bail!("unsupported target architecture {architecture:?}"),
1375 }
1376}
1377
1378fn download_worker(url: &str, expected_sha256: &str, triple: &str) -> Result<PathBuf> {
1379 validate_worker_sha256(expected_sha256)?;
1380 let digest = expected_sha256.to_ascii_lowercase();
1381 let directory = data_dir().join("workers").join("pinned");
1382 let destination = directory.join(&digest).join("hel");
1383 std::fs::create_dir_all(destination.parent().unwrap_or(&directory))?;
1384 if destination.is_file() {
1385 let bytes = std::fs::read(&destination).with_context(|| {
1386 format!(
1387 "read cached worker for {triple} from {}",
1388 destination.display()
1389 )
1390 })?;
1391 if format!("{:x}", Sha256::digest(&bytes)).eq_ignore_ascii_case(expected_sha256) {
1392 return Ok(destination);
1393 }
1394 bail!(
1395 "content-addressed worker cache {} does not match {} checksum",
1396 destination.display(),
1397 expected_sha256
1398 );
1399 }
1400 let bytes = reqwest::blocking::Client::builder()
1401 .timeout(std::time::Duration::from_secs(120))
1402 .build()?
1403 .get(url)
1404 .send()?
1405 .error_for_status()?
1406 .bytes()?;
1407 let actual = format!("{:x}", Sha256::digest(&bytes));
1408 if !actual.eq_ignore_ascii_case(expected_sha256) {
1409 bail!("downloaded worker checksum mismatch: expected {expected_sha256}, got {actual}");
1410 }
1411 std::fs::create_dir_all(&directory)?;
1412 let mut temporary = tempfile::NamedTempFile::new_in(&directory)?;
1413 std::io::Write::write_all(&mut temporary, &bytes)?;
1414 temporary.as_file_mut().sync_all()?;
1415 #[cfg(unix)]
1416 {
1417 use std::os::unix::fs::PermissionsExt;
1418 std::fs::set_permissions(temporary.path(), std::fs::Permissions::from_mode(0o700))?;
1419 }
1420 publish_cached_worker(temporary, &directory, &digest)
1421}
1422
1423fn validate_worker_sha256(expected_sha256: &str) -> Result<()> {
1424 if expected_sha256.len() != 64 || !expected_sha256.bytes().all(|byte| byte.is_ascii_hexdigit())
1425 {
1426 bail!("MJ_WORKER_SHA256 must be a 64-character hexadecimal digest");
1427 }
1428 Ok(())
1429}
1430
1431fn workspace_paths(
1432 locator: &targets::TargetLocator,
1433 bundle: &ProjectBundle,
1434 session_id: &str,
1435) -> Result<(String, Vec<String>)> {
1436 let root = match locator {
1437 targets::TargetLocator::LocalBare { .. } => {
1438 bail!("local bare projects use their selected directory")
1439 }
1440 targets::TargetLocator::LocalPodman { .. }
1441 | targets::TargetLocator::LocalDocker { .. }
1442 | targets::TargetLocator::AppleContainer { .. }
1443 | targets::TargetLocator::SshPodman { .. }
1444 | targets::TargetLocator::SshDocker { .. } => "/workspace".to_string(),
1445 targets::TargetLocator::AwsEc2 { workspace, .. }
1446 | targets::TargetLocator::SshBare { workspace, .. } => workspace.clone(),
1447 };
1448 if matches!(locator, targets::TargetLocator::AwsEc2 { .. }) {
1449 let expected = format!(".local/share/hel/workspaces/{session_id}");
1450 if root != expected {
1451 bail!("AWS workspace does not match session")
1452 }
1453 }
1454 let primary = bundle.primary().context("bundle primary is missing")?;
1455 let primary_path = format!("{root}/{}", primary.destination.to_string_lossy());
1456 let additional = bundle
1457 .repositories
1458 .iter()
1459 .filter(|repository| repository.id != bundle.primary_repo)
1460 .map(|repository| format!("{root}/{}", repository.destination.to_string_lossy()))
1461 .collect();
1462 Ok((primary_path, additional))
1463}
1464
1465pub(super) fn bridge_readiness_stage(profile: &HarnessProfile) -> ProvisionStage {
1476 if matches!(
1477 profile.kind,
1478 HarnessKind::Codex
1479 | HarnessKind::Claude
1480 | HarnessKind::Kimi
1481 | HarnessKind::Grok
1482 | HarnessKind::Muse
1483 ) {
1484 ProvisionStage::Installing(profile.kind)
1485 } else {
1486 ProvisionStage::Starting
1487 }
1488}
1489
1490pub(super) fn bridge_launch(
1491 harness: mj_core::config::HarnessKind,
1492 policy: mj_core::config::ExecutionPolicy,
1493) -> (String, Vec<String>) {
1494 match harness {
1495 mj_core::config::HarnessKind::Muse => ("muse-acp".into(), Vec::new()),
1496 mj_core::config::HarnessKind::Codex => (
1497 "sh".into(),
1498 vec![
1499 "-c".into(),
1500 format!("if command -v codex-acp >/dev/null 2>&1 && [ \"$(codex-acp --version 2>/dev/null)\" = \"{CODEX_ACP_PACKAGE} {CODEX_ACP_VERSION}\" ]; then exec codex-acp; fi; {}; exec npx -y {CODEX_ACP_PACKAGE}@{CODEX_ACP_VERSION}", ensure_node_script()),
1501 ],
1502 ),
1503 mj_core::config::HarnessKind::Claude => (
1504 "sh".into(),
1505 vec![
1506 "-c".into(),
1507 format!("if command -v claude-agent-acp >/dev/null 2>&1; then exec claude-agent-acp; fi; {}; exec npx -y @agentclientprotocol/claude-agent-acp@{CLAUDE_ACP_VERSION}", ensure_node_script()),
1508 ],
1509 ),
1510 mj_core::config::HarnessKind::Kimi => (
1511 "sh".into(),
1512 vec![
1513 "-c".into(),
1514 "if command -v kimi >/dev/null 2>&1; then exec kimi acp; elif [ -x \"$HOME/.kimi-code/bin/kimi\" ]; then exec \"$HOME/.kimi-code/bin/kimi\" acp; elif command -v curl >/dev/null 2>&1; then curl -fsSL https://code.kimi.com/kimi-code/install.sh | bash && exec \"$HOME/.kimi-code/bin/kimi\" acp; else echo 'Mjolnir needs compatible Kimi Code or curl for its official installer; add the tool to PATH' >&2; exit 127; fi".into(),
1515 ],
1516 ),
1517 mj_core::config::HarnessKind::Grok => {
1518 let acp = mj_core::config::HarnessKind::Grok
1519 .bridge_args(policy)
1520 .join(" ");
1521 (
1522 "sh".into(),
1523 vec![
1524 "-c".into(),
1525 format!(
1526 "if command -v grok >/dev/null 2>&1; then exec grok {acp}; elif [ -x \"$GROK_HOME/bin/grok\" ]; then exec \"$GROK_HOME/bin/grok\" {acp}; elif [ -x \"$HOME/.grok/bin/grok\" ]; then exec \"$HOME/.grok/bin/grok\" {acp}; elif command -v curl >/dev/null 2>&1; then curl -fsSL https://x.ai/cli/install.sh | bash && exec \"$HOME/.grok/bin/grok\" {acp}; else echo 'Mjolnir needs compatible Grok Build or curl for its official installer; add the tool to PATH' >&2; exit 127; fi"
1527 ),
1528 ],
1529 )
1530 }
1531 mj_core::config::HarnessKind::Deepseek => {
1532 let acp = mj_core::config::HarnessKind::Deepseek
1533 .bridge_args(policy)
1534 .join(" ");
1535 (
1536 "sh".into(),
1537 vec![
1538 "-c".into(),
1539 format!(
1540 "{}; if command -v dsh >/dev/null 2>&1 && [ \"$(dsh --version 2>/dev/null)\" = \"{DEEPSEEK_DSH_VERSION}\" ]; then exec dsh {acp}; fi; echo 'Mjolnir needs @deepseek-ai/dsh@{DEEPSEEK_DSH_VERSION} installed on PATH' >&2; exit 127",
1541 ensure_node_22_script(),
1542 ),
1543 ],
1544 )
1545 }
1546 }
1547}
1548
1549pub(super) fn preflight_harness(
1550 template: &mj_core::config::TargetTemplate,
1551 profile: &HarnessProfile,
1552 executor: &impl CommandExecutor,
1553) -> Result<()> {
1554 use mj_core::config::TargetTemplate;
1555 if !matches!(
1556 profile.kind,
1557 HarnessKind::Codex | HarnessKind::Claude | HarnessKind::Deepseek
1558 ) {
1559 return Ok(());
1560 }
1561 if !matches!(
1562 template,
1563 TargetTemplate::LocalBare | TargetTemplate::SshBare { .. }
1564 ) {
1565 return Ok(());
1566 }
1567 let script = "if ! command -v node >/dev/null 2>&1; then echo 'Node.js is missing from PATH; install Node.js 22 or newer in the target environment' >&2; exit 127; fi; if ! node -e 'process.exit(Number(process.versions.node.split(\".\")[0]) >= 22 ? 0 : 1)'; then echo 'Node.js 22 or newer is required in the target environment' >&2; exit 1; fi; if ! command -v npm >/dev/null 2>&1 || ! npm --version >/dev/null; then echo 'npm is missing or unusable; install npm in the target environment' >&2; exit 127; fi";
1568 let mut args = if profile.environment.contains_key("PATH") {
1569 vec![
1570 "-c".to_owned(),
1571 format!("export PATH=\"$1\"; {script}"),
1572 "mj-node-preflight".into(),
1573 profile.environment["PATH"].clone(),
1574 ]
1575 } else {
1576 vec!["-lc".to_owned(), script.to_owned()]
1577 };
1578 let (command, destination) = match template {
1579 TargetTemplate::LocalBare => (CommandSpec::new("sh", args), "local host".to_owned()),
1580 TargetTemplate::SshBare { ssh, .. } => {
1581 let ssh = super::backend_ssh(ssh);
1582 args.insert(0, "sh".into());
1583 (ssh_command_spec(&ssh, args), ssh.destination)
1584 }
1585 _ => unreachable!(),
1586 };
1587 execute_checked(executor, command.purpose("preflight managed harness Node.js and npm"))
1588 .with_context(|| format!("{} launch preflight failed on {destination}; Node.js 22+ and npm must be available on the target PATH", profile.kind.display_name()))?;
1589 Ok(())
1590}
1591
1592fn ensure_node_script() -> &'static str {
1593 "if ! command -v node >/dev/null 2>&1 || ! command -v npm >/dev/null 2>&1 || ! command -v npx >/dev/null 2>&1; then echo 'Mjolnir needs Node.js, npm, and npx on PATH; install Node in the target environment' >&2; exit 127; fi"
1594}
1595
1596fn ensure_node_22_script() -> String {
1597 format!(
1598 "{}; if ! node -e 'process.exit(Number(process.versions.node.split(\".\")[0]) >= 22 ? 0 : 1)'; then echo 'DeepSeek Harness requires Node.js 22 or newer' >&2; exit 127; fi",
1599 ensure_node_script()
1600 )
1601}
1602
1603const MJ_CONTAINER_ENVIRONMENT: &str = "## Mjolnir disposable environment\n\nThis session runs in a disposable Mjolnir container. When the session closes, Mjolnir checkpoints everything in project workspace directories under `/workspace`, including committed work, staged and unstaged changes, and untracked files. Mjolnir then removes the container.\n\nEverything outside `/workspace`, including installed packages, `$HOME`, and `/tmp`, is ephemeral and will be lost. Keep durable results in the workspace or push them to a remote.\n\nNew workspaces start on their own session branch from the default network fetch remote’s default branch. Local unpublished commits and uncommitted files are not copied. Use normal git push to publish the current branch to the configured network push destination. Closing saves a checkpoint; it does not publish commits or update the original local checkout. Resumed sessions restore their saved work.\n";
1604
1605pub(super) fn stage_profile(
1606 profile: &mj_core::config::HarnessProfile,
1607 destination: &Path,
1608) -> Result<()> {
1609 let harness = profile.kind;
1610 let source = profile.home.as_path();
1611 std::fs::create_dir_all(destination)?;
1612 let allowlist: &[&str] = match harness {
1613 mj_core::config::HarnessKind::Muse => &[
1614 "auth.json",
1615 "settings.json",
1616 "trust.json",
1617 "AGENTS.md",
1618 "skills",
1619 "rules",
1620 ],
1621 mj_core::config::HarnessKind::Codex => &[
1622 "auth.json",
1623 "config.toml",
1624 "AGENTS.md",
1625 "instructions.md",
1626 "rules",
1627 "skills",
1628 ],
1629 mj_core::config::HarnessKind::Claude => &[
1630 ".claude.json",
1631 ".credentials.json",
1632 "settings.json",
1633 "CLAUDE.md",
1634 "skills",
1635 "plugins",
1636 ],
1637 mj_core::config::HarnessKind::Kimi => &[
1638 "credentials",
1639 "config.toml",
1640 "device_id",
1641 "AGENTS.md",
1642 "SYSTEM.md",
1643 "mcp.json",
1644 "skills",
1645 "agents",
1646 "plugins",
1647 ],
1648 mj_core::config::HarnessKind::Grok => &[
1649 "auth.json",
1650 "config.toml",
1651 "AGENTS.md",
1652 "agent_id",
1653 "skills",
1654 "plugins",
1655 ],
1656 mj_core::config::HarnessKind::Deepseek => &[
1657 ".credentials.yaml",
1658 "settings.yaml",
1659 "AGENTS.md",
1660 "skills",
1661 ".agent-presets",
1662 ],
1663 };
1664 allowlist.par_iter().try_for_each(|name| -> Result<()> {
1668 let from = source.join(name);
1669 if from.exists() {
1670 copy_profile_entry(&from, &destination.join(name))?;
1671 }
1672 Ok(())
1673 })?;
1674 Ok(())
1675}
1676
1677fn append_hel_target_environment(
1679 harness: mj_core::config::HarnessKind,
1680 destination: &Path,
1681 target: &targets::TargetLocator,
1682) -> Result<()> {
1683 let environment = match target {
1684 targets::TargetLocator::LocalPodman { .. }
1685 | targets::TargetLocator::LocalDocker { .. }
1686 | targets::TargetLocator::AppleContainer { .. }
1687 | targets::TargetLocator::SshPodman { .. }
1688 | targets::TargetLocator::SshDocker { .. } => MJ_CONTAINER_ENVIRONMENT.to_owned(),
1689 targets::TargetLocator::AwsEc2 { workspace, .. } => format!(
1690 "## Mjolnir disposable environment\n\nThis session runs on a disposable Mjolnir EC2 instance. When the session closes, Mjolnir checkpoints everything in project workspace directories under `$HOME/{workspace}`, including committed work, staged and unstaged changes, and untracked files. Mjolnir then terminates the instance.\n\nEverything outside `$HOME/{workspace}`, including installed packages, the rest of `$HOME`, and `/tmp`, is ephemeral and will be lost. Keep durable results in the workspace or push them to a remote.\n\nNew workspaces start on their own session branch from the default network fetch remote’s default branch. Local unpublished commits and uncommitted files are not copied. Use normal git push to publish the current branch to the configured network push destination. Closing saves a checkpoint; it does not publish commits or update the original local checkout. Resumed sessions restore their saved work.\n"
1691 ),
1692 targets::TargetLocator::LocalBare { .. } | targets::TargetLocator::SshBare { .. } => {
1693 return Ok(());
1694 }
1695 };
1696 let instructions = match harness {
1697 mj_core::config::HarnessKind::Codex => "AGENTS.md",
1698 mj_core::config::HarnessKind::Claude => "CLAUDE.md",
1699 mj_core::config::HarnessKind::Kimi => "AGENTS.md",
1700 mj_core::config::HarnessKind::Grok => "AGENTS.md",
1701 mj_core::config::HarnessKind::Deepseek => "AGENTS.md",
1702 mj_core::config::HarnessKind::Muse => "AGENTS.md",
1703 };
1704 let path = destination.join(instructions);
1705 let separator = match std::fs::read_to_string(&path) {
1706 Ok(contents) if !contents.is_empty() && !contents.ends_with('\n') => "\n\n",
1707 Ok(contents) if !contents.is_empty() => "\n",
1708 Ok(_) => "",
1709 Err(error) if error.kind() == std::io::ErrorKind::NotFound => "",
1710 Err(error) => return Err(error.into()),
1711 };
1712 use std::io::Write;
1713
1714 let mut file = std::fs::OpenOptions::new()
1715 .create(true)
1716 .append(true)
1717 .open(&path)
1718 .with_context(|| format!("open staged harness instructions {}", path.display()))?;
1719 file.write_all(separator.as_bytes())?;
1720 file.write_all(environment.as_bytes())?;
1721 Ok(())
1722}
1723
1724fn copy_profile_entry(source: &Path, destination: &Path) -> Result<()> {
1725 let metadata = std::fs::symlink_metadata(source)
1726 .with_context(|| format!("read staged profile entry metadata {}", source.display()))?;
1727 if metadata.file_type().is_symlink() {
1728 return Ok(());
1729 }
1730 if metadata.is_file() {
1731 if let Some(parent) = destination.parent() {
1732 std::fs::create_dir_all(parent)
1733 .with_context(|| format!("create staged profile directory {}", parent.display()))?;
1734 }
1735 std::fs::copy(source, destination).with_context(|| {
1736 format!(
1737 "copy staged profile file {} to {}",
1738 source.display(),
1739 destination.display()
1740 )
1741 })?;
1742 return Ok(());
1743 }
1744 if metadata.is_dir() {
1745 std::fs::create_dir_all(destination).with_context(|| {
1746 format!("create staged profile directory {}", destination.display())
1747 })?;
1748 let entries = std::fs::read_dir(source)
1749 .with_context(|| format!("list staged profile directory {}", source.display()))?
1750 .collect::<std::io::Result<Vec<_>>>()
1751 .with_context(|| {
1752 format!(
1753 "read staged profile directory entries in {}",
1754 source.display()
1755 )
1756 })?;
1757 entries.par_iter().try_for_each(|entry| {
1761 copy_profile_entry(&entry.path(), &destination.join(entry.file_name()))
1762 })?;
1763 std::fs::set_permissions(destination, metadata.permissions()).with_context(|| {
1764 format!(
1765 "set permissions for staged profile directory {}",
1766 destination.display()
1767 )
1768 })?;
1769 }
1770 Ok(())
1771}
1772
1773pub(super) fn container_upload_ownership_args(
1777 container_id: &str,
1778 worker_root: &str,
1779 paths: &[&str],
1780) -> Vec<String> {
1781 let mut args = vec![
1782 "exec".into(),
1783 "--user".into(),
1784 "0".into(),
1785 container_id.into(),
1786 "sh".into(),
1787 "-c".into(),
1788 r#"set -eu; owner=$(stat -c '%u:%g' -- "$1"); shift; chown -R "$owner" -- "$@""#.into(),
1790 "sh".into(),
1791 worker_root.into(),
1792 ];
1793 args.extend(paths.iter().map(|path| (*path).to_owned()));
1794 args
1795}
1796
1797#[allow(clippy::too_many_arguments)]
1798fn install_worker_files(
1799 executor: &impl CommandExecutor,
1800 locator: &targets::TargetLocator,
1801 session_id: &str,
1802 worker_root: &str,
1803 profile_home: &str,
1804 worker_binary: &Path,
1805 launch_config: &Path,
1806 ownership: &Path,
1807 profile_stage: &Path,
1808) -> Result<()> {
1809 match locator {
1810 targets::TargetLocator::LocalBare { .. } => {
1811 if profile_stage.is_dir() {
1812 std::fs::create_dir_all(profile_home).context("create isolated local profile")?;
1813 for entry in std::fs::read_dir(profile_stage)? {
1814 let entry = entry?;
1815 copy_profile_entry(
1816 &entry.path(),
1817 &Path::new(profile_home).join(entry.file_name()),
1818 )?;
1819 }
1820 }
1821 for command in [
1822 CommandSpec::new("mkdir", ["-p", worker_root])
1823 .purpose("create local bare worker directory"),
1824 CommandSpec::new(
1825 "cp",
1826 [
1827 worker_binary.to_string_lossy().into_owned(),
1828 format!("{worker_root}/hel"),
1829 ],
1830 )
1831 .purpose("install local Mjolnir worker"),
1832 CommandSpec::new(
1833 "cp",
1834 [
1835 launch_config.to_string_lossy().into_owned(),
1836 format!("{worker_root}/launch.json"),
1837 ],
1838 )
1839 .purpose("install local worker launch configuration"),
1840 CommandSpec::new(
1841 "cp",
1842 [
1843 ownership.to_string_lossy().into_owned(),
1844 format!("{worker_root}/ownership.json"),
1845 ],
1846 )
1847 .purpose("install local worker ownership marker"),
1848 CommandSpec::new("chmod", ["700", &format!("{worker_root}/hel")])
1849 .purpose("make local Mjolnir worker executable"),
1850 ] {
1851 execute_checked(executor, command)?;
1852 }
1853 }
1854 targets::TargetLocator::LocalPodman { container_id, .. }
1855 | targets::TargetLocator::LocalDocker { container_id }
1856 | targets::TargetLocator::AppleContainer { container_id } => {
1857 let engine = match locator {
1858 targets::TargetLocator::LocalPodman { .. } => "podman",
1859 targets::TargetLocator::LocalDocker { .. } => "docker",
1860 targets::TargetLocator::AppleContainer { .. } => "container",
1861 _ => unreachable!("matched local container target"),
1862 };
1863 for command in [
1864 CommandSpec::new(
1865 engine,
1866 [
1867 "exec".into(),
1868 container_id.clone(),
1869 "mkdir".into(),
1870 "-p".into(),
1871 worker_root.into(),
1872 profile_home.into(),
1873 ],
1874 )
1875 .purpose("create target worker directories"),
1876 CommandSpec::new(
1877 engine,
1878 [
1879 "cp".into(),
1880 worker_binary.to_string_lossy().into_owned(),
1881 format!("{container_id}:{worker_root}/hel"),
1882 ],
1883 )
1884 .purpose("upload Mjolnir worker"),
1885 CommandSpec::new(
1886 engine,
1887 [
1888 "cp".into(),
1889 launch_config.to_string_lossy().into_owned(),
1890 format!("{container_id}:{worker_root}/launch.json"),
1891 ],
1892 )
1893 .purpose("upload worker launch configuration"),
1894 CommandSpec::new(
1895 engine,
1896 [
1897 "cp".into(),
1898 ownership.to_string_lossy().into_owned(),
1899 format!("{container_id}:{worker_root}/ownership.json"),
1900 ],
1901 )
1902 .purpose("upload worker ownership marker"),
1903 CommandSpec::new(
1904 engine,
1905 [
1906 "cp".into(),
1907 format!("{}/.", profile_stage.display()),
1908 format!("{container_id}:{profile_home}"),
1909 ],
1910 )
1911 .purpose("upload harness profile allowlist"),
1912 CommandSpec::new(
1913 engine,
1914 container_upload_ownership_args(
1915 container_id,
1916 worker_root,
1917 &[
1918 &format!("{worker_root}/hel"),
1919 &format!("{worker_root}/launch.json"),
1920 &format!("{worker_root}/ownership.json"),
1921 profile_home,
1922 ],
1923 ),
1924 )
1925 .purpose("assign uploaded files to the worker user"),
1926 CommandSpec::new(
1927 engine,
1928 [
1929 "exec".into(),
1930 container_id.clone(),
1931 "chmod".into(),
1932 "700".into(),
1933 format!("{worker_root}/hel"),
1934 ],
1935 )
1936 .purpose("make Mjolnir worker executable"),
1937 CommandSpec::new(
1938 engine,
1939 [
1940 "exec".into(),
1941 container_id.clone(),
1942 "chmod".into(),
1943 "-R".into(),
1944 "go-rwx".into(),
1945 profile_home.into(),
1946 ],
1947 )
1948 .purpose("restrict harness profile permissions"),
1949 ] {
1950 execute_checked(executor, command)?;
1951 }
1952 }
1953 targets::TargetLocator::AwsEc2 { ssh, .. }
1954 | targets::TargetLocator::SshBare { ssh, .. } => {
1955 install_worker_over_ssh(
1956 executor,
1957 ssh,
1958 worker_root,
1959 profile_home,
1960 worker_binary,
1961 launch_config,
1962 ownership,
1963 profile_stage,
1964 )?;
1965 }
1966 targets::TargetLocator::SshPodman {
1967 ssh, container_id, ..
1968 }
1969 | targets::TargetLocator::SshDocker { ssh, container_id } => {
1970 let engine = match locator {
1971 targets::TargetLocator::SshPodman { .. } => "podman",
1972 targets::TargetLocator::SshDocker { .. } => "docker",
1973 _ => unreachable!("matched remote container target"),
1974 };
1975 let digest = mj_core::worker_launch::worker_executable_digest(worker_binary)?;
1979 let cache_dir = format!(".cache/mjolnir/workers/{digest}");
1985 let cached_worker = format!("{cache_dir}/hel");
1986 let cached = matches!(
1987 executor.execute(
1988 &ssh_command_spec(ssh, ["test", "-f", &cached_worker])
1989 .purpose("probe cached remote Mjolnir worker"),
1990 ),
1991 Ok(output) if output.status == 0
1992 );
1993 if !cached {
1994 execute_checked(
1995 executor,
1996 ssh_command_spec(ssh, ["mkdir", "-p", &cache_dir])
1997 .purpose("create remote worker cache"),
1998 )?;
1999 let partial = format!("{cache_dir}/hel.partial-{session_id}");
2000 execute_checked(
2001 executor,
2002 scp_command_spec(ssh, worker_binary, &partial, false)
2003 .purpose("upload remote container worker binary"),
2004 )?;
2005 execute_checked(
2008 executor,
2009 ssh_command_spec(ssh, ["mv", &partial, &cached_worker])
2010 .purpose("publish cached remote Mjolnir worker"),
2011 )?;
2012 }
2013 let upload = format!(".cache/mjolnir/uploads/{session_id}");
2014 execute_checked(
2015 executor,
2016 ssh_command_spec(ssh, ["mkdir", "-p", &upload])
2017 .purpose("create remote upload staging"),
2018 )?;
2019 for (source, name) in [
2020 (launch_config, "launch.json"),
2021 (ownership, "ownership.json"),
2022 ] {
2023 execute_checked(
2024 executor,
2025 scp_command_spec(ssh, source, &format!("{upload}/{name}"), false)
2026 .purpose("upload remote container worker file"),
2027 )?;
2028 }
2029 execute_checked(
2030 executor,
2031 scp_command_spec(ssh, profile_stage, &format!("{upload}/profile"), true)
2032 .purpose("upload remote container profile allowlist"),
2033 )?;
2034 let remote = [
2035 vec![
2036 engine.into(),
2037 "exec".into(),
2038 container_id.clone(),
2039 "mkdir".into(),
2040 "-p".into(),
2041 worker_root.into(),
2042 profile_home.into(),
2043 ],
2044 vec![
2045 engine.into(),
2046 "cp".into(),
2047 cached_worker.clone(),
2048 format!("{container_id}:{worker_root}/hel"),
2049 ],
2050 vec![
2051 engine.into(),
2052 "cp".into(),
2053 format!("{upload}/launch.json"),
2054 format!("{container_id}:{worker_root}/launch.json"),
2055 ],
2056 vec![
2057 engine.into(),
2058 "cp".into(),
2059 format!("{upload}/ownership.json"),
2060 format!("{container_id}:{worker_root}/ownership.json"),
2061 ],
2062 vec![
2063 engine.into(),
2064 "cp".into(),
2065 format!("{upload}/profile/."),
2066 format!("{container_id}:{profile_home}"),
2067 ],
2068 std::iter::once(engine.to_owned())
2069 .chain(container_upload_ownership_args(
2070 container_id,
2071 worker_root,
2072 &[
2073 &format!("{worker_root}/hel"),
2074 &format!("{worker_root}/launch.json"),
2075 &format!("{worker_root}/ownership.json"),
2076 profile_home,
2077 ],
2078 ))
2079 .collect(),
2080 vec![
2081 engine.into(),
2082 "exec".into(),
2083 container_id.clone(),
2084 "chmod".into(),
2085 "700".into(),
2086 format!("{worker_root}/hel"),
2087 ],
2088 vec![
2089 engine.into(),
2090 "exec".into(),
2091 container_id.clone(),
2092 "chmod".into(),
2093 "-R".into(),
2094 "go-rwx".into(),
2095 profile_home.into(),
2096 ],
2097 vec!["rm".into(), "-rf".into(), "--".into(), upload.clone()],
2098 ];
2099 for args in remote {
2100 execute_checked(
2101 executor,
2102 ssh_command_spec(ssh, args).purpose("install remote container worker"),
2103 )?;
2104 }
2105 }
2106 }
2107 Ok(())
2108}
2109
2110#[allow(clippy::too_many_arguments)]
2111fn install_worker_over_ssh(
2112 executor: &impl CommandExecutor,
2113 ssh: &SshTarget,
2114 worker_root: &str,
2115 profile_home: &str,
2116 worker_binary: &Path,
2117 launch_config: &Path,
2118 ownership: &Path,
2119 profile_stage: &Path,
2120) -> Result<()> {
2121 execute_checked(
2122 executor,
2123 ssh_command_spec(ssh, ["mkdir", "-p", worker_root, profile_home])
2124 .purpose("create SSH worker directories"),
2125 )?;
2126 for (source, remote, recursive) in [
2127 (worker_binary, format!("{worker_root}/hel"), false),
2128 (launch_config, format!("{worker_root}/launch.json"), false),
2129 (ownership, format!("{worker_root}/ownership.json"), false),
2130 ] {
2131 execute_checked(
2132 executor,
2133 scp_command_spec(ssh, source, &remote, recursive).purpose("upload SSH worker file"),
2134 )?;
2135 }
2136 let incoming_profile = format!("{profile_home}.incoming");
2137 execute_checked(
2138 executor,
2139 scp_command_spec(ssh, profile_stage, &incoming_profile, true)
2140 .purpose("upload SSH harness profile allowlist"),
2141 )?;
2142 execute_checked(
2143 executor,
2144 ssh_command_spec(
2145 ssh,
2146 ["cp", "-R", &format!("{incoming_profile}/."), profile_home],
2147 )
2148 .purpose("install SSH harness profile allowlist"),
2149 )?;
2150 execute_checked(
2151 executor,
2152 ssh_command_spec(ssh, ["rm", "-rf", "--", &incoming_profile])
2153 .purpose("remove SSH profile staging"),
2154 )?;
2155 execute_checked(
2156 executor,
2157 ssh_command_spec(ssh, ["chmod", "700", &format!("{worker_root}/hel")])
2158 .purpose("make SSH worker executable"),
2159 )?;
2160 execute_checked(
2161 executor,
2162 ssh_command_spec(ssh, ["chmod", "-R", "go-rwx", profile_home])
2163 .purpose("restrict SSH harness profile permissions"),
2164 )?;
2165 Ok(())
2166}
2167
2168pub(super) fn replace_installed_worker_binary(
2174 executor: &impl CommandExecutor,
2175 locator: &targets::TargetLocator,
2176 session_id: &str,
2177 worker_binary: &Path,
2178) -> Result<()> {
2179 let plan = installed_worker_binary_replacement_plan(locator, session_id, worker_binary)?;
2180 for command in plan.commands {
2181 execute_checked(executor, command)?;
2182 }
2183 Ok(())
2184}
2185
2186pub(super) fn replace_installed_worker_launch_config(
2187 executor: &impl CommandExecutor,
2188 locator: &targets::TargetLocator,
2189 session_id: &str,
2190 launch: &WorkerLaunchConfig,
2191) -> Result<()> {
2192 let plan = worker_launch_refresh_plan(locator, session_id, launch)?;
2193 for command in plan.replace.commands {
2194 execute_checked(executor, command)?;
2195 }
2196 Ok(())
2197}
2198
2199pub(super) fn prepare_managed_harness_for_upgrade(
2205 executor: &impl CommandExecutor,
2206 locator: &targets::TargetLocator,
2207 session_id: &str,
2208 worker_binary: &Path,
2209 launch: &WorkerLaunchConfig,
2210) -> Result<()> {
2211 if launch.harness_runtime != HarnessRuntimePolicy::Managed {
2212 return Ok(());
2213 }
2214 let worker_root = targets::worker_root(locator, session_id)?;
2215 let staging_root = format!("{worker_root}/harness-prepare");
2216 let staging_binary = format!("{staging_root}/hel");
2217 let staging_config = format!("{staging_root}/launch.json");
2218 let staging = tempfile::tempdir().context("create managed harness upgrade staging")?;
2219 let local_config = staging.path().join("launch.json");
2220 launch.write(&local_config)?;
2221
2222 if matches!(locator, targets::TargetLocator::LocalBare { .. }) {
2226 execute_checked(
2227 executor,
2228 CommandSpec::new(
2229 worker_binary.to_string_lossy().into_owned(),
2230 [
2231 "worker".to_owned(),
2232 "prepare-harness".to_owned(),
2233 "--config".to_owned(),
2234 local_config.to_string_lossy().into_owned(),
2235 ],
2236 )
2237 .purpose("prepare exact managed harness"),
2238 )?;
2239 return Ok(());
2240 }
2241
2242 let ssh = match locator {
2243 targets::TargetLocator::AwsEc2 { ssh, .. }
2244 | targets::TargetLocator::SshBare { ssh, .. } => ssh,
2245 _ => bail!("managed harness policy requires a local bare, SSH-bare, or EC2 target"),
2246 };
2247 let result = (|| {
2248 execute_checked(
2249 executor,
2250 ssh_command_spec(ssh, ["rm", "-rf", "--", &staging_root])
2251 .purpose("clear managed harness preparation staging"),
2252 )?;
2253 execute_checked(
2254 executor,
2255 ssh_command_spec(ssh, ["mkdir", "-p", &staging_root])
2256 .purpose("create managed harness preparation staging"),
2257 )?;
2258 execute_checked(
2259 executor,
2260 scp_command_spec(ssh, worker_binary, &staging_binary, false)
2261 .purpose("stage current worker for managed harness preparation"),
2262 )?;
2263 execute_checked(
2264 executor,
2265 scp_command_spec(ssh, &local_config, &staging_config, false)
2266 .purpose("stage managed harness launch configuration"),
2267 )?;
2268 execute_checked(
2269 executor,
2270 ssh_command_spec(ssh, ["chmod", "700", &staging_binary])
2271 .purpose("make managed harness preparation worker executable"),
2272 )?;
2273 execute_checked(
2274 executor,
2275 ssh_command_spec(
2276 ssh,
2277 [
2278 staging_binary.as_str(),
2279 "worker",
2280 "prepare-harness",
2281 "--config",
2282 staging_config.as_str(),
2283 ],
2284 )
2285 .purpose("prepare exact managed harness"),
2286 )?;
2287 Ok(())
2288 })();
2289 let cleanup = execute_checked(
2290 executor,
2291 ssh_command_spec(ssh, ["rm", "-rf", "--", &staging_root])
2292 .purpose("remove managed harness preparation staging"),
2293 );
2294 match (result, cleanup) {
2295 (Ok(()), Ok(_)) => Ok(()),
2296 (Ok(()), Err(error)) => Err(error).context("clean managed harness preparation staging"),
2297 (Err(error), Ok(_)) => Err(error),
2298 (Err(error), Err(cleanup)) => {
2299 tracing::warn!(%cleanup, path = %staging_root, "managed harness preparation staging cleanup failed");
2300 Err(error)
2301 }
2302 }
2303}
2304
2305fn prepare_installed_managed_harness(
2306 executor: &impl CommandExecutor,
2307 locator: &targets::TargetLocator,
2308 worker_root: &str,
2309 launch: &WorkerLaunchConfig,
2310) -> Result<()> {
2311 if launch.harness_runtime != HarnessRuntimePolicy::Managed {
2312 return Ok(());
2313 }
2314 let worker_binary = format!("{worker_root}/hel");
2315 let launch_config = format!("{worker_root}/launch.json");
2316 let command = match locator {
2317 targets::TargetLocator::LocalBare { .. } => CommandSpec::new(
2318 worker_binary.clone(),
2319 [
2320 "worker",
2321 "prepare-harness",
2322 "--config",
2323 launch_config.as_str(),
2324 ],
2325 ),
2326 targets::TargetLocator::AwsEc2 { ssh, .. }
2327 | targets::TargetLocator::SshBare { ssh, .. } => ssh_command_spec(
2328 ssh,
2329 [
2330 worker_binary.as_str(),
2331 "worker",
2332 "prepare-harness",
2333 "--config",
2334 launch_config.as_str(),
2335 ],
2336 ),
2337 _ => bail!("managed harness policy requires a local bare, SSH-bare, or EC2 target"),
2338 };
2339 execute_checked(
2340 executor,
2341 command.purpose("prepare exact managed harness before worker startup"),
2342 )?;
2343 Ok(())
2344}
2345
2346fn installed_worker_binary_replacement_plan(
2347 locator: &targets::TargetLocator,
2348 session_id: &str,
2349 worker_binary: &Path,
2350) -> Result<CommandPlan> {
2351 let worker_root = targets::worker_root(locator, session_id)?;
2352 let installed = format!("{worker_root}/hel");
2353 let staged = format!("{worker_root}/hel.next");
2354 let commands = match locator {
2355 targets::TargetLocator::LocalBare { .. } => vec![
2356 CommandSpec::new(
2357 "cp",
2358 [worker_binary.to_string_lossy().into_owned(), staged.clone()],
2359 )
2360 .purpose("stage replacement Mjolnir worker"),
2361 CommandSpec::new("mv", ["-f", &staged, &installed])
2362 .purpose("replace installed Mjolnir worker"),
2363 CommandSpec::new("chmod", ["700", &installed])
2364 .purpose("make replaced Mjolnir worker executable"),
2365 ],
2366 targets::TargetLocator::LocalPodman { container_id, .. }
2367 | targets::TargetLocator::LocalDocker { container_id }
2368 | targets::TargetLocator::AppleContainer { container_id } => {
2369 let engine = match locator {
2370 targets::TargetLocator::LocalPodman { .. } => "podman",
2371 targets::TargetLocator::LocalDocker { .. } => "docker",
2372 targets::TargetLocator::AppleContainer { .. } => "container",
2373 _ => unreachable!("matched local container target"),
2374 };
2375 vec![
2376 CommandSpec::new(
2377 engine,
2378 [
2379 "cp".into(),
2380 worker_binary.to_string_lossy().into_owned(),
2381 format!("{container_id}:{staged}"),
2382 ],
2383 )
2384 .purpose("stage replacement Mjolnir worker"),
2385 CommandSpec::new(
2386 engine,
2387 container_upload_ownership_args(container_id, &worker_root, &[&staged]),
2388 )
2389 .purpose("assign replacement worker to the worker user"),
2390 CommandSpec::new(
2391 engine,
2392 [
2393 "exec".into(),
2394 container_id.clone(),
2395 "mv".into(),
2396 "-f".into(),
2397 staged,
2398 installed.clone(),
2399 ],
2400 )
2401 .purpose("replace installed Mjolnir worker"),
2402 CommandSpec::new(
2403 engine,
2404 [
2405 "exec".into(),
2406 container_id.clone(),
2407 "chmod".into(),
2408 "700".into(),
2409 installed,
2410 ],
2411 )
2412 .purpose("make replaced Mjolnir worker executable"),
2413 ]
2414 }
2415 targets::TargetLocator::AwsEc2 { ssh, .. }
2416 | targets::TargetLocator::SshBare { ssh, .. } => vec![
2417 scp_command_spec(ssh, worker_binary, &staged, false)
2418 .purpose("stage replacement Mjolnir worker"),
2419 ssh_command_spec(ssh, ["mv", "-f", "--", &staged, &installed])
2420 .purpose("replace installed Mjolnir worker"),
2421 ssh_command_spec(ssh, ["chmod", "700", &installed])
2422 .purpose("make replaced Mjolnir worker executable"),
2423 ],
2424 targets::TargetLocator::SshPodman {
2425 ssh, container_id, ..
2426 }
2427 | targets::TargetLocator::SshDocker { ssh, container_id } => {
2428 let engine = match locator {
2429 targets::TargetLocator::SshPodman { .. } => "podman",
2430 targets::TargetLocator::SshDocker { .. } => "docker",
2431 _ => unreachable!("matched remote container target"),
2432 };
2433 let upload = format!(".cache/mjolnir/uploads/{session_id}-hel.next");
2434 vec![
2435 ssh_command_spec(ssh, ["mkdir", "-p", ".cache/mjolnir/uploads"])
2436 .purpose("create remote replacement worker staging"),
2437 scp_command_spec(ssh, worker_binary, &upload, false)
2438 .purpose("stage replacement Mjolnir worker"),
2439 ssh_command_spec(
2440 ssh,
2441 [engine, "cp", &upload, &format!("{container_id}:{staged}")],
2442 )
2443 .purpose("stage replacement Mjolnir worker"),
2444 ssh_command_spec(
2445 ssh,
2446 std::iter::once(engine.to_owned()).chain(container_upload_ownership_args(
2447 container_id,
2448 &worker_root,
2449 &[&staged],
2450 )),
2451 )
2452 .purpose("assign replacement worker to the worker user"),
2453 ssh_command_spec(
2454 ssh,
2455 [
2456 engine,
2457 "exec",
2458 container_id,
2459 "mv",
2460 "-f",
2461 "--",
2462 &staged,
2463 &installed,
2464 ],
2465 )
2466 .purpose("replace installed Mjolnir worker"),
2467 ssh_command_spec(
2468 ssh,
2469 [engine, "exec", container_id, "chmod", "700", &installed],
2470 )
2471 .purpose("make replaced Mjolnir worker executable"),
2472 ssh_command_spec(ssh, ["rm", "-f", "--", &upload])
2473 .purpose("remove remote replacement worker staging"),
2474 ]
2475 }
2476 };
2477 Ok(CommandPlan {
2478 description: format!("replace stale Mjolnir worker for session {session_id}"),
2479 commands,
2480 })
2481}
2482
2483fn installed_file_digest_command(
2484 locator: &targets::TargetLocator,
2485 path: &str,
2486 purpose: &str,
2487) -> CommandSpec {
2488 match locator {
2489 targets::TargetLocator::LocalBare { .. } => CommandSpec::new("sha256sum", [path]),
2490 targets::TargetLocator::LocalPodman { container_id, .. } => {
2491 CommandSpec::new("podman", ["exec", container_id, "sha256sum", path])
2492 }
2493 targets::TargetLocator::LocalDocker { container_id } => {
2494 CommandSpec::new("docker", ["exec", container_id, "sha256sum", path])
2495 }
2496 targets::TargetLocator::AppleContainer { container_id } => {
2497 CommandSpec::new("container", ["exec", container_id, "sha256sum", path])
2498 }
2499 targets::TargetLocator::AwsEc2 { ssh, .. }
2500 | targets::TargetLocator::SshBare { ssh, .. } => ssh_command_spec(ssh, ["sha256sum", path]),
2501 targets::TargetLocator::SshPodman {
2502 ssh, container_id, ..
2503 } => ssh_command_spec(ssh, ["podman", "exec", container_id, "sha256sum", path]),
2504 targets::TargetLocator::SshDocker { ssh, container_id } => {
2505 ssh_command_spec(ssh, ["docker", "exec", container_id, "sha256sum", path])
2506 }
2507 }
2508 .purpose(purpose)
2509}
2510
2511fn worker_launch_refresh_plan(
2512 locator: &targets::TargetLocator,
2513 session_id: &str,
2514 launch: &WorkerLaunchConfig,
2515) -> Result<WorkerLaunchRefreshPlan> {
2516 let worker_root = targets::worker_root(locator, session_id)?;
2517 let installed = format!("{worker_root}/launch.json");
2518 let staged = format!("{installed}.next");
2519 let staged_arg = targets::join_remote_command(std::slice::from_ref(&staged));
2520 let installed_arg = targets::join_remote_command(std::slice::from_ref(&installed));
2521 let script = format!("umask 077; cat > {staged_arg} && mv -f -- {staged_arg} {installed_arg}");
2522 let body = serde_json::to_vec_pretty(launch).context("serialize worker launch config")?;
2523 let expected_sha256 = format!("{:x}", Sha256::digest(&body));
2524 let replace = match locator {
2525 targets::TargetLocator::LocalBare { .. } => CommandSpec::new("sh", ["-c", &script]),
2526 targets::TargetLocator::LocalPodman { container_id, .. } => {
2527 CommandSpec::new("podman", ["exec", "-i", container_id, "sh", "-c", &script])
2528 }
2529 targets::TargetLocator::LocalDocker { container_id } => {
2530 CommandSpec::new("docker", ["exec", "-i", container_id, "sh", "-c", &script])
2531 }
2532 targets::TargetLocator::AppleContainer { container_id } => CommandSpec::new(
2533 "container",
2534 ["exec", "-i", container_id, "sh", "-c", &script],
2535 ),
2536 targets::TargetLocator::AwsEc2 { ssh, .. }
2537 | targets::TargetLocator::SshBare { ssh, .. } => {
2538 ssh_command_spec(ssh, ["sh", "-c", &script])
2539 }
2540 targets::TargetLocator::SshPodman {
2541 ssh, container_id, ..
2542 } => ssh_command_spec(
2543 ssh,
2544 ["podman", "exec", "-i", container_id, "sh", "-c", &script],
2545 ),
2546 targets::TargetLocator::SshDocker { ssh, container_id } => ssh_command_spec(
2547 ssh,
2548 ["docker", "exec", "-i", container_id, "sh", "-c", &script],
2549 ),
2550 }
2551 .purpose("replace stale Mjolnir worker launch config")
2552 .with_sensitive_stdin(body);
2553 Ok(WorkerLaunchRefreshPlan {
2554 expected_sha256,
2555 installed_digest: installed_file_digest_command(
2556 locator,
2557 &installed,
2558 "identify installed Mjolnir worker launch config",
2559 ),
2560 replace: CommandPlan {
2561 description: format!("replace stale Mjolnir launch config for session {session_id}"),
2562 commands: vec![replace],
2563 },
2564 })
2565}
2566
2567fn worker_binary_refresh_plan(
2570 locator: &targets::TargetLocator,
2571 session_id: &str,
2572) -> Result<Option<WorkerBinaryRefresh>> {
2573 let worker_root = targets::worker_root(locator, session_id)?;
2574 let installed = format!("{worker_root}/hel");
2575 if matches!(
2580 locator,
2581 targets::TargetLocator::AwsEc2 { .. }
2582 | targets::TargetLocator::SshBare { .. }
2583 | targets::TargetLocator::SshPodman { .. }
2584 | targets::TargetLocator::SshDocker { .. }
2585 ) {
2586 return Ok(Some(WorkerBinaryRefresh::Remote(
2587 RemoteWorkerBinaryRefresh {
2588 locator: locator.clone(),
2589 session_id: session_id.to_owned(),
2590 installed_digest: installed_file_digest_command(
2591 locator,
2592 &installed,
2593 "identify installed Mjolnir worker binary",
2594 ),
2595 },
2596 )));
2597 }
2598 if PINNED_WORKER_BINARY_SOURCES.get().is_none()
2603 && !std::env::current_exe().is_ok_and(|path| path.is_file())
2604 {
2605 return Ok(None);
2606 }
2607 let requirement = if matches!(locator, targets::TargetLocator::LocalBare { .. }) {
2608 WorkerBinaryRequirement::LocalHost
2609 } else {
2610 WorkerBinaryRequirement::PortableLinux
2611 };
2612 let source = match worker_binary_for_arch(std::env::consts::ARCH, requirement) {
2613 Ok(WorkerBinaryAvailability::Local { path, .. }) => path,
2614 Ok(WorkerBinaryAvailability::Remote { .. }) | Err(_) => return Ok(None),
2615 };
2616 Ok(Some(WorkerBinaryRefresh::Prepared(
2617 WorkerBinaryRefreshPlan {
2618 replace: installed_worker_binary_replacement_plan(locator, session_id, &source)?,
2619 source,
2620 installed_digest: installed_file_digest_command(
2621 locator,
2622 &installed,
2623 "identify installed Mjolnir worker binary",
2624 ),
2625 },
2626 )))
2627}
2628
2629pub(crate) fn refresh_remote_worker_binary_if_stale(
2638 executor: &impl CommandExecutor,
2639 refresh: &RemoteWorkerBinaryRefresh,
2640) -> Result<()> {
2641 let source = worker_binary_for(&refresh.locator, executor)
2642 .context("resolve the worker binary for the recovering target")?;
2643 replace_remote_worker_binary_if_stale(
2644 executor,
2645 &refresh.locator,
2646 &refresh.session_id,
2647 &refresh.installed_digest,
2648 &source,
2649 )
2650 .map(|_| ())
2651}
2652
2653fn replace_remote_worker_binary_if_stale(
2658 executor: &impl CommandExecutor,
2659 locator: &targets::TargetLocator,
2660 session_id: &str,
2661 installed_digest: &CommandSpec,
2662 source: &Path,
2663) -> Result<bool> {
2664 let expected = mj_core::worker_launch::worker_executable_digest(source)?;
2665 let installed = executor
2666 .execute(installed_digest)
2667 .context("read the installed remote worker digest")?;
2668 let matches = installed.status == 0
2669 && String::from_utf8_lossy(&installed.stdout)
2670 .split_whitespace()
2671 .next()
2672 .is_some_and(|digest| digest.eq_ignore_ascii_case(&expected));
2673 if matches {
2674 return Ok(false);
2675 }
2676 installed_worker_binary_replacement_plan(locator, session_id, source)?
2677 .execute(executor)
2678 .context("replace stale remote relay worker binary")?;
2679 Ok(true)
2680}
2681
2682pub(super) fn stop_worker(
2687 executor: &impl CommandExecutor,
2688 locator: &targets::TargetLocator,
2689 worker_root: &str,
2690) -> Result<()> {
2691 execute_checked(executor, stop_worker_command(locator, worker_root))?;
2692 Ok(())
2693}
2694
2695pub(super) fn stop_worker_after_target_recovery(
2698 executor: &impl CommandExecutor,
2699 locator: &targets::TargetLocator,
2700 session_id: &str,
2701 worker_root: &str,
2702) -> Result<()> {
2703 let target = targets::target_recovery_plan(locator, session_id)?;
2704 targets::ensure_recovery_target_running(executor, target.as_ref())
2705 .context("restore Mjolnir worker target")?;
2706 stop_worker(executor, locator, worker_root)
2707}
2708
2709fn stop_worker_command(locator: &targets::TargetLocator, worker_root: &str) -> CommandSpec {
2710 let script = targets::stop_worker_daemon_script(worker_root);
2711 match locator {
2712 targets::TargetLocator::LocalBare { .. } => CommandSpec::new("sh", ["-c", &script]),
2713 targets::TargetLocator::LocalPodman { container_id, .. } => {
2714 CommandSpec::new("podman", ["exec", container_id, "sh", "-c", &script])
2715 }
2716 targets::TargetLocator::LocalDocker { container_id } => {
2717 CommandSpec::new("docker", ["exec", container_id, "sh", "-c", &script])
2718 }
2719 targets::TargetLocator::AppleContainer { container_id } => {
2720 CommandSpec::new("container", ["exec", container_id, "sh", "-c", &script])
2721 }
2722 targets::TargetLocator::AwsEc2 { ssh, .. }
2723 | targets::TargetLocator::SshBare { ssh, .. } => {
2724 ssh_command_spec(ssh, ["sh", "-c", &script])
2725 }
2726 targets::TargetLocator::SshPodman {
2727 ssh, container_id, ..
2728 } => ssh_command_spec(ssh, ["podman", "exec", container_id, "sh", "-c", &script]),
2729 targets::TargetLocator::SshDocker { ssh, container_id } => {
2730 ssh_command_spec(ssh, ["docker", "exec", container_id, "sh", "-c", &script])
2731 }
2732 }
2733 .purpose("stop Mjolnir worker daemon")
2734}
2735
2736fn worker_liveness_command(locator: &targets::TargetLocator, worker_root: &str) -> CommandSpec {
2737 let script = targets::worker_daemon_liveness_script(worker_root);
2738 match locator {
2739 targets::TargetLocator::LocalBare { .. } => CommandSpec::new("sh", ["-c", &script]),
2740 targets::TargetLocator::LocalPodman { container_id, .. } => {
2741 CommandSpec::new("podman", ["exec", container_id, "sh", "-c", &script])
2742 }
2743 targets::TargetLocator::LocalDocker { container_id } => {
2744 CommandSpec::new("docker", ["exec", container_id, "sh", "-c", &script])
2745 }
2746 targets::TargetLocator::AppleContainer { container_id } => {
2747 CommandSpec::new("container", ["exec", container_id, "sh", "-c", &script])
2748 }
2749 targets::TargetLocator::AwsEc2 { ssh, .. }
2750 | targets::TargetLocator::SshBare { ssh, .. } => {
2751 ssh_command_spec(ssh, ["sh", "-c", &script])
2752 }
2753 targets::TargetLocator::SshPodman {
2754 ssh, container_id, ..
2755 } => ssh_command_spec(ssh, ["podman", "exec", container_id, "sh", "-c", &script]),
2756 targets::TargetLocator::SshDocker { ssh, container_id } => {
2757 ssh_command_spec(ssh, ["docker", "exec", container_id, "sh", "-c", &script])
2758 }
2759 }
2760 .purpose("probe Mjolnir worker daemon liveness")
2761}
2762
2763pub(super) fn start_worker(
2764 executor: &impl CommandExecutor,
2765 locator: &targets::TargetLocator,
2766 worker_root: &str,
2767) -> Result<()> {
2768 execute_checked(executor, start_worker_command(locator, worker_root))?;
2769 Ok(())
2770}
2771
2772fn start_worker_command(locator: &targets::TargetLocator, worker_root: &str) -> CommandSpec {
2773 let binary = format!("{worker_root}/hel");
2774 let config = format!("{worker_root}/launch.json");
2775 let clear_stale_runtime = format!(
2780 "rm -f {} {}; ",
2781 targets::join_remote_command(&[format!("{worker_root}/worker-exit.json")]),
2782 targets::join_remote_command(&[format!("{worker_root}/control.sock")]),
2783 );
2784 let detached_script = format!(
2785 "{clear_stale_runtime}nohup {} >{} 2>&1 </dev/null &",
2786 targets::join_remote_command(&[
2787 binary.clone(),
2788 "worker".into(),
2789 "run".into(),
2790 "--root".into(),
2791 worker_root.into(),
2792 "--config".into(),
2793 config.clone(),
2794 ]),
2795 targets::join_remote_command(&[format!("{worker_root}/worker.log")]),
2796 );
2797 let exec_script = format!(
2800 "{clear_stale_runtime}exec {} >{} 2>&1",
2801 targets::join_remote_command(&[
2802 binary.clone(),
2803 "worker".into(),
2804 "run".into(),
2805 "--root".into(),
2806 worker_root.into(),
2807 "--config".into(),
2808 config.clone(),
2809 ]),
2810 targets::join_remote_command(&[format!("{worker_root}/worker.log")]),
2811 );
2812 match locator {
2813 targets::TargetLocator::LocalBare { .. } => {
2814 CommandSpec::new("sh", ["-c", &detached_script])
2815 }
2816 targets::TargetLocator::LocalPodman { container_id, .. } => CommandSpec::new(
2817 "podman",
2818 ["exec", "--detach", container_id, "sh", "-c", &exec_script],
2819 ),
2820 targets::TargetLocator::LocalDocker { container_id } => CommandSpec::new(
2821 "docker",
2822 ["exec", "--detach", container_id, "sh", "-c", &exec_script],
2823 ),
2824 targets::TargetLocator::AppleContainer { container_id } => CommandSpec::new(
2825 "container",
2826 ["exec", "--detach", container_id, "sh", "-c", &exec_script],
2827 ),
2828 targets::TargetLocator::AwsEc2 { ssh, .. }
2829 | targets::TargetLocator::SshBare { ssh, .. } => {
2830 ssh_command_spec(ssh, ["sh", "-c", &detached_script])
2831 }
2832 targets::TargetLocator::SshPodman {
2833 ssh, container_id, ..
2834 } => ssh_command_spec(
2835 ssh,
2836 [
2837 "podman",
2838 "exec",
2839 "--detach",
2840 container_id,
2841 "sh",
2842 "-c",
2843 &exec_script,
2844 ],
2845 ),
2846 targets::TargetLocator::SshDocker { ssh, container_id } => ssh_command_spec(
2847 ssh,
2848 [
2849 "docker",
2850 "exec",
2851 "--detach",
2852 container_id,
2853 "sh",
2854 "-c",
2855 &exec_script,
2856 ],
2857 ),
2858 }
2859 .purpose("start detached Mjolnir worker")
2860 .stage(ProvisionStage::Starting)
2863}
2864
2865pub(super) fn worker_probe_diagnosis(
2870 executor: &impl CommandExecutor,
2871 locator: &targets::TargetLocator,
2872 worker_root: &str,
2873 error: anyhow::Error,
2874) -> anyhow::Error {
2875 let error = match worker_binary_probe_failure(executor, locator, worker_root) {
2876 Some(failure) => error.context(failure),
2877 None => error,
2878 };
2879 match worker_last_words(executor, locator, worker_root) {
2880 Some(last_words) => error.context(last_words),
2881 None => error,
2882 }
2883}
2884
2885fn worker_binary_probe_failure(
2886 executor: &impl CommandExecutor,
2887 locator: &targets::TargetLocator,
2888 worker_root: &str,
2889) -> Option<String> {
2890 let binary = format!("{worker_root}/hel");
2891 let command = match locator {
2892 targets::TargetLocator::LocalBare { .. } => CommandSpec::new(binary.clone(), ["--version"]),
2893 targets::TargetLocator::LocalPodman { container_id, .. } => {
2894 CommandSpec::new("podman", ["exec", container_id, &binary, "--version"])
2895 }
2896 targets::TargetLocator::LocalDocker { container_id } => {
2897 CommandSpec::new("docker", ["exec", container_id, &binary, "--version"])
2898 }
2899 targets::TargetLocator::AppleContainer { container_id } => {
2900 CommandSpec::new("container", ["exec", container_id, &binary, "--version"])
2901 }
2902 targets::TargetLocator::AwsEc2 { ssh, .. }
2903 | targets::TargetLocator::SshBare { ssh, .. } => {
2904 ssh_command_spec(ssh, [binary.as_str(), "--version"])
2905 }
2906 targets::TargetLocator::SshPodman {
2907 ssh, container_id, ..
2908 } => ssh_command_spec(
2909 ssh,
2910 ["podman", "exec", container_id, binary.as_str(), "--version"],
2911 ),
2912 targets::TargetLocator::SshDocker { ssh, container_id } => ssh_command_spec(
2913 ssh,
2914 ["docker", "exec", container_id, binary.as_str(), "--version"],
2915 ),
2916 }
2917 .purpose("probe installed worker binary");
2918 match executor.execute(&command) {
2919 Ok(output) if output.status == 0 => None,
2920 Ok(output) => {
2921 let stderr = String::from_utf8_lossy(&output.stderr);
2922 let stdout = String::from_utf8_lossy(&output.stdout);
2923 let detail = if !stderr.trim().is_empty() {
2924 stderr.trim()
2925 } else if !stdout.trim().is_empty() {
2926 stdout.trim()
2927 } else {
2928 "the process exited unsuccessfully without output"
2929 };
2930 Some(format!(
2931 "worker binary {binary} fails to run in the target: {detail}; \
2932 if this is a loader/glibc error, provide a musl worker \
2933 (cargo build --release --target <arch>-unknown-linux-musl \
2934 -p brokk-mj-worker --bin mj-worker, \
2935 or set MJ_WORKER_BINARY/MJ_WORKER_DIR)"
2936 ))
2937 }
2938 Err(probe_error) => Some(format!("worker probe failed: {probe_error:#}")),
2939 }
2940}
2941
2942pub(super) fn worker_last_words(
2945 executor: &impl CommandExecutor,
2946 locator: &targets::TargetLocator,
2947 worker_root: &str,
2948) -> Option<String> {
2949 let script = format!(
2950 "if [ -f {root}/worker-exit.json ]; then echo '{marker}'; cat {root}/worker-exit.json; fi; if [ -f {root}/worker.log ]; then echo '--- worker.log (tail) ---'; tail -n 20 {root}/worker.log; fi",
2951 root = worker_root,
2952 marker = WORKER_EXIT_RECORD_MARKER
2953 );
2954 let command = match locator {
2955 targets::TargetLocator::LocalBare { .. } => CommandSpec::new("sh", ["-c", &script]),
2956 targets::TargetLocator::LocalPodman { container_id, .. } => {
2957 CommandSpec::new("podman", ["exec", container_id, "sh", "-c", &script])
2958 }
2959 targets::TargetLocator::LocalDocker { container_id } => {
2960 CommandSpec::new("docker", ["exec", container_id, "sh", "-c", &script])
2961 }
2962 targets::TargetLocator::AppleContainer { container_id } => {
2963 CommandSpec::new("container", ["exec", container_id, "sh", "-c", &script])
2964 }
2965 targets::TargetLocator::AwsEc2 { ssh, .. }
2966 | targets::TargetLocator::SshBare { ssh, .. } => {
2967 ssh_command_spec(ssh, ["sh", "-c", &script])
2968 }
2969 targets::TargetLocator::SshPodman {
2970 ssh, container_id, ..
2971 } => ssh_command_spec(ssh, ["podman", "exec", container_id, "sh", "-c", &script]),
2972 targets::TargetLocator::SshDocker { ssh, container_id } => {
2973 ssh_command_spec(ssh, ["docker", "exec", container_id, "sh", "-c", &script])
2974 }
2975 }
2976 .purpose("collect worker last words");
2977 let output = match executor.execute(&command) {
2978 Ok(output) => output,
2979 Err(error) => {
2980 tracing::debug!(
2981 worker_root,
2982 %error,
2983 "could not collect worker diagnostics"
2984 );
2985 return None;
2986 }
2987 };
2988 if output.status != 0 {
2989 tracing::debug!(
2990 worker_root,
2991 status = output.status,
2992 "worker diagnostic probe returned a failure"
2993 );
2994 return None;
2995 }
2996 let text = String::from_utf8_lossy(&output.stdout).trim().to_string();
2997 (!text.is_empty()).then(|| format!("worker diagnostics:\n{text}"))
2998}
2999
3000#[cfg(test)]
3001mod tests {
3002 use super::*;
3003
3004 use anyhow::Result;
3005
3006 use crate::targets::{self, CommandExecutor, CommandOutput, CommandSpec, SshTarget};
3007 use mj_core::config::ExecutionPolicy;
3008
3009 use sha2::{Digest, Sha256};
3010 use std::cell::RefCell;
3011 use std::collections::BTreeMap;
3012
3013 use std::path::{Path, PathBuf};
3014
3015 #[cfg(unix)]
3016 #[test]
3017 fn node_preflight_checks_missing_old_and_supported_tools_on_profile_path() {
3018 use std::os::unix::fs::PermissionsExt;
3019 let directory = tempfile::tempdir().unwrap();
3020 let profile = HarnessProfile {
3021 enabled: true,
3022 kind: HarnessKind::Codex,
3023 home: directory.path().into(),
3024 environment: std::collections::BTreeMap::from([(
3025 "PATH".into(),
3026 directory.path().to_string_lossy().into_owned(),
3027 )]),
3028 context_window_bytes: None,
3029 };
3030 let check = || {
3031 preflight_harness(
3032 &mj_core::config::TargetTemplate::LocalBare,
3033 &profile,
3034 &ProcessExecutor,
3035 )
3036 };
3037 let write_tool = |name: &str, body: &str| {
3038 let path = directory.path().join(name);
3039 std::fs::write(&path, format!("#!/bin/sh\n{body}\n")).unwrap();
3040 std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).unwrap();
3041 };
3042 assert!(format!("{:#}", check().unwrap_err()).contains("Node.js is missing"));
3043 write_tool("node", "exit 1");
3044 assert!(format!("{:#}", check().unwrap_err()).contains("Node.js 22 or newer is required"));
3045 write_tool("node", "exit 0");
3046 assert!(format!("{:#}", check().unwrap_err()).contains("npm is missing or unusable"));
3047 write_tool("npm", "exit 0");
3048 check().unwrap();
3049 }
3050
3051 #[test]
3052 fn a_stored_setup_token_reaches_only_claude_workers_that_do_not_set_their_own() {
3053 use mj_core::config::HarnessKind;
3054 use mj_core::credentials::{CLAUDE_OAUTH_TOKEN_ENV, write_claude_oauth_token};
3055
3056 let directory = tempfile::tempdir().unwrap();
3057 let token_path = directory.path().join("profiles/claude/claude-oauth-token");
3058 let missing = directory.path().join("profiles/absent/claude-oauth-token");
3059 write_claude_oauth_token(&token_path, b"sk-ant-oat01-stored").unwrap();
3060
3061 let mut claude = BTreeMap::new();
3062 apply_claude_setup_token(&mut claude, HarnessKind::Claude, &token_path);
3063 assert_eq!(
3064 claude.get(CLAUDE_OAUTH_TOKEN_ENV).map(String::as_str),
3065 Some("sk-ant-oat01-stored")
3066 );
3067
3068 for kind in HarnessKind::ALL
3070 .into_iter()
3071 .filter(|kind| *kind != HarnessKind::Claude)
3072 {
3073 let mut environment = BTreeMap::new();
3074 apply_claude_setup_token(&mut environment, kind, &token_path);
3075 assert!(environment.is_empty(), "{kind:?} must not read the token");
3076 }
3077
3078 let mut overridden = BTreeMap::from([(
3080 CLAUDE_OAUTH_TOKEN_ENV.to_owned(),
3081 "profile-token".to_owned(),
3082 )]);
3083 apply_claude_setup_token(&mut overridden, HarnessKind::Claude, &token_path);
3084 assert_eq!(
3085 overridden.get(CLAUDE_OAUTH_TOKEN_ENV).map(String::as_str),
3086 Some("profile-token")
3087 );
3088
3089 let mut without = BTreeMap::new();
3091 apply_claude_setup_token(&mut without, HarnessKind::Claude, &missing);
3092 assert!(without.is_empty());
3093 }
3094
3095 #[test]
3096 fn packaged_worker_names_match_release_archives() {
3097 let directory = Path::new("/opt/hel/bin");
3098 assert_eq!(
3099 packaged_worker_binary_path(directory, "x86_64-unknown-linux-musl"),
3100 directory.join("mj-worker-x86_64-unknown-linux-musl")
3101 );
3102 assert_eq!(
3103 packaged_worker_binary_path(directory, "aarch64-unknown-linux-musl"),
3104 directory.join("mj-worker-aarch64-unknown-linux-musl")
3105 );
3106 }
3107
3108 #[test]
3109 fn pinned_snapshot_keeps_native_and_portable_sources_stable() {
3110 let directory = tempfile::tempdir().unwrap();
3111 let native = directory.path().join("native-worker");
3112 let x86 = directory.path().join("x86-worker");
3113 let arm = directory.path().join("arm-worker");
3114 std::fs::write(&native, b"native bytes").unwrap();
3115 std::fs::write(&x86, b"x86 bytes").unwrap();
3116 std::fs::write(&arm, b"arm bytes").unwrap();
3117 let cache = directory.path().join("cache");
3118 let snapshot = WorkerBinarySourceSnapshot::capture(&cache, |arch, requirement| {
3119 let path = match requirement {
3120 WorkerBinaryRequirement::LocalHost => &native,
3121 WorkerBinaryRequirement::PortableLinux if arch == "x86_64" => &x86,
3122 WorkerBinaryRequirement::PortableLinux => &arm,
3123 };
3124 Ok(WorkerBinaryAvailability::Local {
3125 path: path.clone(),
3126 source: format!("{arch}-{requirement:?}"),
3127 })
3128 });
3129
3130 let native = snapshot
3131 .resolve(std::env::consts::ARCH, WorkerBinaryRequirement::LocalHost)
3132 .unwrap();
3133 let x86 = snapshot
3134 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3135 .unwrap();
3136 let arm = snapshot
3137 .resolve("aarch64", WorkerBinaryRequirement::PortableLinux)
3138 .unwrap();
3139 let WorkerBinaryAvailability::Local { path: native, .. } = native else {
3140 panic!("native source should be local");
3141 };
3142 let WorkerBinaryAvailability::Local { path: x86, .. } = x86 else {
3143 panic!("x86 source should be local");
3144 };
3145 let WorkerBinaryAvailability::Local { path: arm, .. } = arm else {
3146 panic!("arm source should be local");
3147 };
3148 assert_eq!(std::fs::read(native).unwrap(), b"native bytes");
3149 assert_eq!(std::fs::read(x86).unwrap(), b"x86 bytes");
3150 assert_eq!(std::fs::read(arm).unwrap(), b"arm bytes");
3151 }
3152
3153 #[test]
3154 fn pinned_snapshot_survives_source_replacement_and_missing_candidate_install() {
3155 let directory = tempfile::tempdir().unwrap();
3156 let source = directory.path().join("worker");
3157 std::fs::write(&source, b"before").unwrap();
3158 let cache = directory.path().join("cache");
3159 let resolve_source = |_: &str, _: WorkerBinaryRequirement| {
3160 Ok(WorkerBinaryAvailability::Local {
3161 path: source.clone(),
3162 source: "test source".into(),
3163 })
3164 };
3165 let pinned = WorkerBinarySourceSnapshot::capture(&cache, resolve_source);
3166
3167 std::fs::write(&source, b"in-place mutation").unwrap();
3168 let WorkerBinaryAvailability::Local { path, .. } = pinned
3169 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3170 .unwrap()
3171 else {
3172 panic!("source should be local");
3173 };
3174 assert_eq!(std::fs::read(path).unwrap(), b"before");
3175
3176 let replacement = directory.path().join("replacement");
3177 std::fs::write(&replacement, b"after").unwrap();
3178 std::fs::rename(replacement, &source).unwrap();
3179 let WorkerBinaryAvailability::Local { path, .. } = pinned
3180 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3181 .unwrap()
3182 else {
3183 panic!("source should be local");
3184 };
3185 assert_eq!(std::fs::read(path).unwrap(), b"before");
3186 let fresh_replaced = WorkerBinarySourceSnapshot::capture(&cache, resolve_source);
3187 let WorkerBinaryAvailability::Local { path, .. } = fresh_replaced
3188 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3189 .unwrap()
3190 else {
3191 panic!("source should be local");
3192 };
3193 assert_eq!(std::fs::read(path).unwrap(), b"after");
3194
3195 let missing = directory.path().join("missing-worker");
3196 let missing_snapshot = WorkerBinarySourceSnapshot::capture(&cache, {
3197 let missing = missing.clone();
3198 move |_: &str, _: WorkerBinaryRequirement| {
3199 if missing.is_file() {
3200 Ok(WorkerBinaryAvailability::Local {
3201 path: missing.clone(),
3202 source: "new source".into(),
3203 })
3204 } else {
3205 Err(anyhow::anyhow!("candidate is unavailable"))
3206 }
3207 }
3208 });
3209 std::fs::write(&missing, b"now installed").unwrap();
3210 assert!(
3211 missing_snapshot
3212 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3213 .is_err()
3214 );
3215 let fresh_snapshot = WorkerBinarySourceSnapshot::capture(&cache, {
3216 let missing = missing.clone();
3217 move |_: &str, _: WorkerBinaryRequirement| {
3218 Ok(WorkerBinaryAvailability::Local {
3219 path: missing.clone(),
3220 source: "new source".into(),
3221 })
3222 }
3223 });
3224 assert!(
3225 fresh_snapshot
3226 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3227 .is_ok()
3228 );
3229
3230 let remote_url = std::cell::RefCell::new("https://old.example/{target}".to_owned());
3231 let remote_snapshot = WorkerBinarySourceSnapshot::capture(
3232 &directory.path().join("remote-cache"),
3233 |arch, _| {
3234 Ok(WorkerBinaryAvailability::Remote {
3235 url: remote_url.borrow().replace("{target}", arch),
3236 sha256: "a".repeat(64),
3237 triple: format!("{arch}-unknown-linux-musl"),
3238 })
3239 },
3240 );
3241 *remote_url.borrow_mut() = "https://new.example/{target}".into();
3242 let WorkerBinaryAvailability::Remote { url, .. } = remote_snapshot
3243 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3244 .unwrap()
3245 else {
3246 panic!("source should be remote");
3247 };
3248 assert_eq!(url, "https://old.example/x86_64");
3249
3250 let blocked_cache = directory.path().join("blocked-cache");
3251 std::fs::write(&blocked_cache, b"not a directory").unwrap();
3252 let failed_snapshot = WorkerBinarySourceSnapshot::capture(&blocked_cache, resolve_source);
3253 assert!(
3254 failed_snapshot
3255 .resolve("x86_64", WorkerBinaryRequirement::PortableLinux)
3256 .is_err()
3257 );
3258 }
3259
3260 #[test]
3261 fn dev_checkout_prefers_the_dedicated_musl_worker() {
3262 let controller = PathBuf::from("target/debug/mj");
3263 let musl = PathBuf::from("target/worker/x86_64-unknown-linux-musl/debug/mj-worker");
3264 let shared_target_worker =
3265 PathBuf::from("target/x86_64-unknown-linux-musl/debug/mj-worker");
3266 let legacy = PathBuf::from("target/x86_64-unknown-linux-musl/debug/mj");
3267 let present = [
3268 controller.clone(),
3269 musl.clone(),
3270 shared_target_worker,
3271 legacy,
3272 ];
3273 let selected = select_sibling_worker(&controller, "x86_64-unknown-linux-musl", |path| {
3274 present.iter().any(|p| p == path)
3275 });
3276 assert_eq!(
3277 selected,
3278 Some((musl, "isolated development musl worker")),
3279 "the dedicated worker must win over legacy artifacts"
3280 );
3281 }
3282
3283 #[test]
3284 fn local_bare_may_use_a_native_worker_beside_the_controller() {
3285 let controller = PathBuf::from("target/debug/mj");
3286 let worker = PathBuf::from("target/debug/mj-worker");
3287 let selected = worker_binary_prerequisite_for_current(
3288 std::env::consts::ARCH,
3289 WorkerBinaryRequirement::LocalHost,
3290 &controller,
3291 &|path| path == controller || path == worker,
3292 )
3293 .unwrap();
3294 assert_eq!(
3295 selected,
3296 WorkerBinaryAvailability::Local {
3297 path: worker,
3298 source: "native worker beside mj".into(),
3299 }
3300 );
3301 }
3302
3303 #[test]
3304 fn local_bare_prefers_the_isolated_native_development_worker() {
3305 let controller = PathBuf::from("target/debug/mj");
3306 let worker = PathBuf::from("target/worker/debug/mj-worker");
3307 let packaged = PathBuf::from("target/debug/mj-worker");
3308 let selected = worker_binary_prerequisite_for_current(
3309 std::env::consts::ARCH,
3310 WorkerBinaryRequirement::LocalHost,
3311 &controller,
3312 &|path| path == controller || path == worker || path == packaged,
3313 )
3314 .unwrap();
3315 assert_eq!(
3316 selected,
3317 WorkerBinaryAvailability::Local {
3318 path: worker,
3319 source: "isolated native development worker".into(),
3320 }
3321 );
3322 }
3323
3324 #[cfg(target_os = "linux")]
3325 #[test]
3326 fn replaced_dev_controller_still_finds_its_musl_sibling() {
3327 let controller = PathBuf::from("target/debug/mj (deleted)");
3328 let musl = PathBuf::from("target/x86_64-unknown-linux-musl/debug/mj");
3329 let selected = select_sibling_worker(&controller, "x86_64-unknown-linux-musl", |path| {
3330 path == musl
3331 });
3332
3333 assert_eq!(selected, Some((musl, "development musl sibling")));
3334 }
3335
3336 #[cfg(target_os = "linux")]
3337 #[test]
3338 fn replaced_dev_controller_never_selects_the_new_glibc_controller_as_its_worker() {
3339 let controller = PathBuf::from("target/debug/mj (deleted)");
3340 let replacement = PathBuf::from("target/debug/mj");
3341 let selected = select_sibling_worker(&controller, "x86_64-unknown-linux-musl", |path| {
3342 path == replacement
3343 });
3344
3345 assert_eq!(selected, None);
3346 }
3347
3348 fn container_template(platform: Option<&str>) -> mj_core::config::ContainerTemplate {
3351 mj_core::config::ContainerTemplate {
3352 image: "example.invalid/mj-test:latest".into(),
3353 pull_policy: Default::default(),
3354 platform: platform.map(str::to_owned),
3355 cpus: None,
3356 memory: None,
3357 environment: BTreeMap::new(),
3358 workspace_storage: Default::default(),
3359 }
3360 }
3361
3362 fn ssh_connection() -> mj_core::config::SshConnection {
3363 mj_core::config::SshConnection {
3364 host: "builder".into(),
3365 user: Some("dev".into()),
3366 identity_file: None,
3367 extra_args: Vec::new(),
3368 }
3369 }
3370
3371 #[test]
3372 fn recovery_workspace_uses_the_launch_directory_for_bare_targets_only() {
3373 let cwd = PathBuf::from("/workspace/session/project");
3374 let local = worker_workspace_for_recovery(
3375 &targets::TargetLocator::LocalBare {
3376 worker_root: "/workspace/session/worker".into(),
3377 },
3378 &cwd,
3379 )
3380 .expect("local bare targets need a workspace probe");
3381 assert_eq!(local.directory, cwd);
3382 assert_eq!(local.target, mj_core::state::ManagedWorktreeTarget::Local);
3383
3384 let remote = worker_workspace_for_recovery(
3385 &targets::TargetLocator::SshBare {
3386 worker_id: None,
3387 ssh: SshTarget {
3388 destination: "dev@builder".into(),
3389 ssh_args: vec!["-oBatchMode=yes".into()],
3390 },
3391 workspace: "/workspace/session".into(),
3392 },
3393 &cwd,
3394 )
3395 .expect("SSH bare targets need a workspace probe");
3396 assert_eq!(remote.directory, cwd);
3397 assert_eq!(
3398 remote.target,
3399 mj_core::state::ManagedWorktreeTarget::Ssh {
3400 destination: "dev@builder".into(),
3401 ssh_args: vec!["-oBatchMode=yes".into()],
3402 }
3403 );
3404
3405 assert!(
3406 worker_workspace_for_recovery(
3407 &targets::TargetLocator::LocalPodman {
3408 container_id: "container".into(),
3409 workspace_storage: Default::default(),
3410 },
3411 &cwd,
3412 )
3413 .is_none()
3414 );
3415 assert!(
3416 worker_workspace_for_recovery(
3417 &targets::TargetLocator::AwsEc2 {
3418 profile: "default".into(),
3419 region: "us-east-1".into(),
3420 instance_id: "i-test".into(),
3421 ssh: SshTarget {
3422 destination: "dev@builder".into(),
3423 ssh_args: Vec::new(),
3424 },
3425 workspace: "/workspace/session".into(),
3426 },
3427 &cwd,
3428 )
3429 .is_none()
3430 );
3431 }
3432
3433 #[test]
3434 fn preflight_reads_the_architecture_a_template_names() {
3435 use mj_core::config::TargetTemplate;
3436
3437 for (platform, expected) in [
3438 ("linux/arm64", "aarch64"),
3439 ("linux/arm64/v8", "aarch64"),
3440 ("linux/amd64", "x86_64"),
3441 ("aarch64", "aarch64"),
3442 ] {
3443 assert_eq!(
3444 preflight_architectures(&TargetTemplate::LocalPodman {
3445 container: container_template(Some(platform)),
3446 }),
3447 vec![expected],
3448 "platform {platform}"
3449 );
3450 }
3451 assert_eq!(
3454 preflight_architectures(&TargetTemplate::SshPodman {
3455 ssh: ssh_connection(),
3456 container: container_template(Some("linux/arm64")),
3457 }),
3458 vec!["aarch64"]
3459 );
3460 }
3461
3462 #[test]
3463 fn preflight_uses_the_host_architecture_for_a_local_target() {
3464 use mj_core::config::TargetTemplate;
3465
3466 for template in [
3467 TargetTemplate::LocalBare,
3468 TargetTemplate::LocalPodman {
3469 container: container_template(None),
3470 },
3471 TargetTemplate::LocalDocker {
3472 container: container_template(None),
3473 },
3474 TargetTemplate::AppleContainer {
3475 container: container_template(None),
3476 },
3477 ] {
3478 assert_eq!(
3479 preflight_architectures(&template),
3480 vec![std::env::consts::ARCH],
3481 "{template:?}"
3482 );
3483 }
3484 }
3485
3486 #[test]
3487 fn preflight_accepts_either_linux_architecture_for_a_remote_target() {
3488 use mj_core::config::TargetTemplate;
3489
3490 for template in [
3494 TargetTemplate::SshBare {
3495 ssh: ssh_connection(),
3496 permissions: mj_core::config::PermissionMode::Yolo,
3497 workspace_prefix: PathBuf::from(".local/share/hel/workspaces"),
3498 },
3499 TargetTemplate::SshPodman {
3500 ssh: ssh_connection(),
3501 container: container_template(None),
3502 },
3503 TargetTemplate::AwsEc2 {
3504 aws_profile: None,
3505 region: "us-east-1".into(),
3506 launch_template: "lt-mj".into(),
3507 launch_template_version: None,
3508 ssh_user: "dev".into(),
3509 address_source: Default::default(),
3510 identity_file: None,
3511 ssh_args: Vec::new(),
3512 },
3513 ] {
3514 assert_eq!(
3515 preflight_architectures(&template),
3516 vec!["x86_64", "aarch64"],
3517 "{template:?}"
3518 );
3519 }
3520 }
3521
3522 #[test]
3523 fn dev_checkout_still_finds_a_hel_named_sibling() {
3524 let controller = PathBuf::from("target/debug/hel");
3525 let musl = PathBuf::from("target/x86_64-unknown-linux-musl/debug/hel");
3526 let present = [controller.clone(), musl.clone()];
3527 let selected = select_sibling_worker(&controller, "x86_64-unknown-linux-musl", |path| {
3528 present.iter().any(|p| p == path)
3529 });
3530 assert_eq!(selected, Some((musl, "development musl sibling")));
3531 }
3532
3533 const FOREIGN_ARCH: &str = "riscv64";
3536
3537 #[test]
3541 fn a_replaced_controller_is_reported_instead_of_a_missing_worker() {
3542 let stale = PathBuf::from("/src/.backup-vHXvCs/target/debug/mj (deleted)");
3543 let probed = RefCell::new(Vec::new());
3544
3545 let error = worker_binary_prerequisite_for_current(
3546 FOREIGN_ARCH,
3547 WorkerBinaryRequirement::PortableLinux,
3548 &stale,
3549 &|path| {
3550 probed.borrow_mut().push(path.to_path_buf());
3551 false
3552 },
3553 )
3554 .unwrap_err();
3555
3556 let detail = format!("{error:#}");
3557 assert!(
3558 detail.contains("was replaced or removed on disk"),
3559 "{detail}"
3560 );
3561 assert!(detail.contains("restart the Mjolnir daemon"), "{detail}");
3562 assert!(
3564 detail.contains("/src/.backup-vHXvCs/target/debug/mj)"),
3565 "{detail}"
3566 );
3567 assert!(!detail.contains("(deleted)"), "{detail}");
3568 assert_eq!(
3569 probed.into_inner(),
3570 vec![stale],
3571 "nothing beside a path that no longer exists is worth probing"
3572 );
3573 }
3574
3575 #[test]
3581 fn a_present_controller_still_looks_beside_itself() {
3582 let controller = PathBuf::from("/opt/brokk/mj");
3583 let probed = RefCell::new(Vec::new());
3584
3585 let error = worker_binary_prerequisite_for_current(
3586 FOREIGN_ARCH,
3587 WorkerBinaryRequirement::PortableLinux,
3588 &controller,
3589 &|path| {
3590 probed.borrow_mut().push(path.to_path_buf());
3591 path == controller
3592 },
3593 )
3594 .unwrap_err();
3595
3596 let probed = probed.into_inner();
3597 assert!(
3598 probed
3599 .iter()
3600 .any(|path| path.ends_with("mj-worker-riscv64-unknown-linux-musl")),
3601 "the packaged worker name must still be probed: {probed:?}"
3602 );
3603 let detail = format!("{error:#}");
3604 assert!(
3605 detail.contains("no Linux worker for riscv64-unknown-linux-musl"),
3606 "{detail}"
3607 );
3608 assert!(!detail.contains("restart the Mjolnir daemon"), "{detail}");
3609
3610 let root = PathBuf::from("/");
3613 let error = worker_binary_prerequisite_for_current(
3614 FOREIGN_ARCH,
3615 WorkerBinaryRequirement::PortableLinux,
3616 &root,
3617 &|path| path == root,
3618 )
3619 .unwrap_err();
3620 let detail = format!("{error:#}");
3621 assert!(
3622 detail.contains("no Linux worker for riscv64-unknown-linux-musl"),
3623 "{detail}"
3624 );
3625 assert!(!detail.contains("restart the Mjolnir daemon"), "{detail}");
3626 }
3627
3628 const WORKER_BINARY_OVERRIDE_CHILD: &str = "MJ_WORKER_BINARY_OVERRIDE_CHILD";
3629
3630 #[test]
3633 fn a_replaced_controller_still_honors_the_worker_binary_override() {
3634 if std::env::var_os(WORKER_BINARY_OVERRIDE_CHILD).is_none() {
3637 let directory = tempfile::tempdir().unwrap();
3638 let worker = directory.path().join("mj-worker");
3639 std::fs::write(&worker, b"worker").unwrap();
3640 let test_name = format!(
3641 "{}::a_replaced_controller_still_honors_the_worker_binary_override",
3642 module_path!()
3643 .strip_prefix("mj_controller::")
3644 .unwrap_or(module_path!())
3645 );
3646 let output = std::process::Command::new(std::env::current_exe().unwrap())
3647 .args(["--exact", &test_name, "--nocapture"])
3648 .env(WORKER_BINARY_OVERRIDE_CHILD, "1")
3649 .env("MJ_WORKER_BINARY", &worker)
3650 .output()
3651 .unwrap();
3652 assert!(
3653 output.status.success(),
3654 "isolated worker override test failed\nstdout:\n{}\nstderr:\n{}",
3655 String::from_utf8_lossy(&output.stdout),
3656 String::from_utf8_lossy(&output.stderr)
3657 );
3658 return;
3659 }
3660
3661 let stale = PathBuf::from("/src/.backup-vHXvCs/target/debug/mj (deleted)");
3662 let availability = worker_binary_prerequisite_for_current(
3663 FOREIGN_ARCH,
3664 WorkerBinaryRequirement::PortableLinux,
3665 &stale,
3666 &|path| path.is_file(),
3667 )
3668 .unwrap();
3669
3670 match availability {
3671 WorkerBinaryAvailability::Local { source, .. } => {
3672 assert_eq!(source, "MJ_WORKER_BINARY");
3673 }
3674 other => panic!("expected the override to resolve, got {other:?}"),
3675 }
3676 }
3677
3678 #[test]
3679 fn sibling_lookup_falls_back_to_the_legacy_hel_name_beside_an_mj_controller() {
3680 let controller = PathBuf::from("/opt/brokk/mj");
3681 let legacy = PathBuf::from("/opt/brokk/hel");
3682 let selected = select_sibling_worker(&controller, "x86_64-unknown-linux-musl", |path| {
3683 path == legacy
3684 });
3685 assert_eq!(selected, Some((legacy, "beside the running executable")));
3686 }
3687
3688 #[test]
3689 fn worker_diagnosis_surfaces_a_loader_failure_from_the_installed_binary() {
3690 struct FailedProbe;
3691
3692 impl CommandExecutor for FailedProbe {
3693 fn execute(&self, _command: &CommandSpec) -> Result<CommandOutput> {
3694 Ok(CommandOutput {
3695 status: 1,
3696 stdout: Vec::new(),
3697 stderr: b"libc.so.6: version `GLIBC_2.39' not found\n".to_vec(),
3698 })
3699 }
3700 }
3701
3702 let failure = worker_binary_probe_failure(
3703 &FailedProbe,
3704 &targets::TargetLocator::LocalBare {
3705 worker_root: "/worker/root".into(),
3706 },
3707 "/worker/root",
3708 )
3709 .expect("an unsuccessful --version probe should explain the dead worker");
3710
3711 assert!(failure.contains("GLIBC_2.39"), "{failure}");
3712 assert!(failure.contains("provide a musl worker"), "{failure}");
3713 }
3714
3715 #[test]
3719 fn starting_a_worker_clears_stale_runtime_files_before_launching() {
3720 struct RecordingExecutor {
3721 commands: RefCell<Vec<CommandSpec>>,
3722 }
3723
3724 impl CommandExecutor for RecordingExecutor {
3725 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
3726 self.commands.borrow_mut().push(command.clone());
3727 Ok(CommandOutput {
3728 status: 0,
3729 stdout: Vec::new(),
3730 stderr: Vec::new(),
3731 })
3732 }
3733 }
3734
3735 for locator in [
3736 targets::TargetLocator::LocalBare {
3737 worker_root: "/worker/root".into(),
3738 },
3739 targets::TargetLocator::LocalPodman {
3740 container_id: "container-1".into(),
3741 workspace_storage: Default::default(),
3742 },
3743 ] {
3744 let executor = RecordingExecutor {
3745 commands: RefCell::new(Vec::new()),
3746 };
3747 start_worker(&executor, &locator, "/worker/root").unwrap();
3748
3749 let commands = executor.commands.borrow();
3750 let script = commands
3751 .iter()
3752 .flat_map(|command| command.args.iter())
3753 .find(|argument| argument.contains("worker-exit.json"))
3754 .unwrap_or_else(|| {
3755 panic!("no launch script cleared the exit record: {commands:?}")
3756 });
3757 let cleared = script.find("rm -f").expect("the exit record is removed");
3758 let launched = script.find("worker").expect("the daemon is launched");
3759 assert!(
3760 script.contains("control.sock"),
3761 "the stale relay endpoint must be cleared before startup: {script}"
3762 );
3763 assert!(
3764 cleared < launched,
3765 "stale runtime files must be cleared before the daemon starts: {script}"
3766 );
3767 }
3768 }
3769 #[test]
3770 fn stopping_a_worker_runs_the_daemon_stop_script() {
3771 struct RecordingExecutor {
3772 commands: RefCell<Vec<CommandSpec>>,
3773 }
3774
3775 impl CommandExecutor for RecordingExecutor {
3776 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
3777 self.commands.borrow_mut().push(command.clone());
3778 Ok(CommandOutput {
3779 status: 0,
3780 stdout: Vec::new(),
3781 stderr: Vec::new(),
3782 })
3783 }
3784 }
3785
3786 let locator = targets::TargetLocator::SshBare {
3787 worker_id: None,
3788 ssh: SshTarget {
3789 destination: "user@example.test".into(),
3790 ssh_args: Vec::new(),
3791 },
3792 workspace: "/workspace".into(),
3793 };
3794 let executor = RecordingExecutor {
3795 commands: RefCell::new(Vec::new()),
3796 };
3797 stop_worker(&executor, &locator, "/worker/root").unwrap();
3798
3799 let commands = executor.commands.borrow();
3800 assert_eq!(commands.len(), 1);
3801 assert_eq!(commands[0].purpose, "stop Mjolnir worker daemon");
3802 assert!(
3803 commands[0]
3804 .args
3805 .last()
3806 .is_some_and(|remote| remote.starts_with("'sh' '-c' ")),
3807 "raw SSH worker management must not source login profiles: {commands:?}"
3808 );
3809 let script = commands[0]
3810 .args
3811 .iter()
3812 .find(|argument| argument.contains("worker run --root"))
3813 .unwrap_or_else(|| panic!("stop script missing from {commands:?}"));
3814 assert!(
3815 script.contains("hel_match=\"hel worker run --root $hel_root\""),
3816 "stop must match only this session's worker: {script}"
3817 );
3818 assert!(
3819 script.contains("hel_match_home=\"hel worker run --root $HOME/$hel_root\""),
3820 "stop must also match a login-home-absolute --root: {script}"
3821 );
3822 assert!(
3823 !script.contains("grep -F"),
3824 "leftover detection must not grep the match string: {script}"
3825 );
3826 }
3827 #[test]
3828 fn checkpoint_worker_stop_restores_a_stopped_podman_target_first() {
3829 struct RecordingExecutor {
3830 commands: RefCell<Vec<CommandSpec>>,
3831 outputs: RefCell<Vec<CommandOutput>>,
3832 }
3833
3834 impl CommandExecutor for RecordingExecutor {
3835 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
3836 self.commands.borrow_mut().push(command.clone());
3837 Ok(self.outputs.borrow_mut().remove(0))
3838 }
3839 }
3840
3841 let session = "0123456789abcdef0123456789abcdef";
3842 let container_id = targets::resource_name(session).unwrap();
3843 let inspection = |status: &str| CommandOutput {
3844 status: 0,
3845 stdout: serde_json::to_vec(&serde_json::json!([{
3846 "Config": { "Labels": {
3847 (targets::MANAGED_LABEL): "true",
3848 (targets::SESSION_LABEL): session,
3849 }},
3850 "State": { "Status": status },
3851 }]))
3852 .unwrap(),
3853 stderr: Vec::new(),
3854 };
3855 let executor = RecordingExecutor {
3856 commands: RefCell::new(Vec::new()),
3857 outputs: RefCell::new(vec![
3858 CommandOutput {
3859 status: 0,
3860 stdout: Vec::new(),
3861 stderr: Vec::new(),
3862 },
3863 inspection("exited"),
3864 CommandOutput {
3865 status: 0,
3866 stdout: Vec::new(),
3867 stderr: Vec::new(),
3868 },
3869 inspection("running"),
3870 CommandOutput {
3871 status: 0,
3872 stdout: Vec::new(),
3873 stderr: Vec::new(),
3874 },
3875 ]),
3876 };
3877 let locator = targets::TargetLocator::LocalPodman {
3878 container_id,
3879 workspace_storage: Default::default(),
3880 };
3881
3882 stop_worker_after_target_recovery(&executor, &locator, session, "/worker/root").unwrap();
3883
3884 let commands = executor.commands.borrow();
3885 let purposes = commands
3886 .iter()
3887 .map(|command| command.purpose.as_str())
3888 .collect::<Vec<_>>();
3889 assert_eq!(
3890 purposes,
3891 [
3892 "check for Mjolnir session container",
3893 "inspect Mjolnir session container",
3894 "start stopped Mjolnir session container",
3895 "inspect Mjolnir session container",
3896 "stop Mjolnir worker daemon",
3897 ]
3898 );
3899 }
3900
3901 struct PodmanInstallExecutor {
3902 commands: RefCell<Vec<CommandSpec>>,
3903 worker_cached: bool,
3904 }
3905 impl CommandExecutor for PodmanInstallExecutor {
3906 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
3907 self.commands.borrow_mut().push(command.clone());
3908 let probing_cache = command
3909 .args
3910 .iter()
3911 .any(|argument| argument.contains("'test' '-f'"));
3912 let status = if probing_cache && !self.worker_cached {
3913 1
3914 } else {
3915 0
3916 };
3917 Ok(CommandOutput {
3918 status,
3919 stdout: Vec::new(),
3920 stderr: Vec::new(),
3921 })
3922 }
3923 }
3924 struct PodmanInstallFixture {
3925 _root: tempfile::TempDir,
3926 worker_binary: PathBuf,
3927 launch_config: PathBuf,
3928 ownership: PathBuf,
3929 profile_stage: PathBuf,
3930 locator: targets::TargetLocator,
3931 digest: String,
3932 }
3933 fn podman_install_fixture() -> PodmanInstallFixture {
3934 let root = tempfile::tempdir().unwrap();
3935 let worker_binary = root.path().join("hel");
3936 std::fs::write(&worker_binary, b"worker-binary-bytes").unwrap();
3937 let launch_config = root.path().join("launch.json");
3938 std::fs::write(&launch_config, b"{}").unwrap();
3939 let ownership = root.path().join("ownership.json");
3940 std::fs::write(&ownership, b"{}").unwrap();
3941 let profile_stage = root.path().join("profile");
3942 std::fs::create_dir_all(&profile_stage).unwrap();
3943 let digest = format!("{:x}", Sha256::digest(b"worker-binary-bytes"));
3944 PodmanInstallFixture {
3945 _root: root,
3946 worker_binary,
3947 launch_config,
3948 ownership,
3949 profile_stage,
3950 locator: targets::TargetLocator::SshPodman {
3951 ssh: SshTarget {
3952 destination: "user@example.test".into(),
3953 ssh_args: Vec::new(),
3954 },
3955 container_id: "container-1".into(),
3956 workspace_storage: Default::default(),
3957 },
3958 digest,
3959 }
3960 }
3961 fn run_podman_install(worker_cached: bool) -> (Vec<CommandSpec>, PodmanInstallFixture) {
3962 let fixture = podman_install_fixture();
3963 let executor = PodmanInstallExecutor {
3964 commands: RefCell::new(Vec::new()),
3965 worker_cached,
3966 };
3967 install_worker_files(
3968 &executor,
3969 &fixture.locator,
3970 "0123456789abcdef0123456789abcdef",
3971 "/workspace/.hel/worker",
3972 "/workspace/.hel/profile",
3973 &fixture.worker_binary,
3974 &fixture.launch_config,
3975 &fixture.ownership,
3976 &fixture.profile_stage,
3977 )
3978 .unwrap();
3979 let commands = executor.commands.borrow().clone();
3980 (commands, fixture)
3981 }
3982 fn rendered(commands: &[CommandSpec]) -> Vec<String> {
3983 commands
3984 .iter()
3985 .map(|command| format!("{} {}", command.program, command.args.join(" ")))
3986 .collect()
3987 }
3988 #[test]
3989 fn ssh_podman_install_caches_the_worker_binary_on_a_cache_miss() {
3990 let (commands, fixture) = run_podman_install(false);
3991 let lines = rendered(&commands);
3992 let digest = &fixture.digest;
3993 let cache_dir = format!(".cache/mjolnir/workers/{digest}");
3994 let session = "0123456789abcdef0123456789abcdef";
3995
3996 assert!(
3997 lines
3998 .iter()
3999 .any(|line| line.starts_with("ssh") && line.contains("'test' '-f'")),
4000 "expected a cache probe, got {lines:#?}"
4001 );
4002 assert!(
4003 !lines.iter().any(|line| line.contains('~')),
4004 "remote staging paths must be home-relative: ssh arguments are \
4005 single-quoted so a tilde stays literal in the remote shell while \
4006 scp expands it, got {lines:#?}"
4007 );
4008 assert!(
4009 lines.iter().any(|line| line.starts_with("ssh")
4010 && line.contains(&format!("'mkdir' '-p' '{cache_dir}'"))),
4011 "expected the cache directory to be created, got {lines:#?}"
4012 );
4013 let partial = format!("{cache_dir}/hel.partial-{session}");
4014 assert!(
4015 lines.iter().any(|line| line
4016 == &format!(
4017 "scp {} user@example.test:{partial}",
4018 fixture.worker_binary.display()
4019 )),
4020 "expected the worker to be uploaded to the partial cache path, got {lines:#?}"
4021 );
4022 assert!(
4023 lines.iter().any(|line| line.starts_with("ssh")
4024 && line.contains(&format!("'mv' '{partial}' '{cache_dir}/hel'"))),
4025 "expected an atomic rename into the cache, got {lines:#?}"
4026 );
4027 assert!(
4028 lines.iter().any(|line| line.contains("'podman' 'cp'")
4029 && line.contains(&format!("'{cache_dir}/hel'"))),
4030 "expected podman cp to read the cached worker, got {lines:#?}"
4031 );
4032 assert!(
4033 !lines.iter().any(|line| line.starts_with("scp")
4034 && line.ends_with(&format!(
4035 "user@example.test:.cache/mjolnir/uploads/{session}/hel"
4036 ))),
4037 "the worker must not be staged in the per-session upload directory, got {lines:#?}"
4038 );
4039 }
4040 #[test]
4041 fn ssh_podman_install_skips_the_worker_upload_on_a_cache_hit() {
4042 let (commands, fixture) = run_podman_install(true);
4043 let lines = rendered(&commands);
4044 let digest = &fixture.digest;
4045 let cache_dir = format!(".cache/mjolnir/workers/{digest}");
4046 let session = "0123456789abcdef0123456789abcdef";
4047
4048 assert!(
4049 !lines.iter().any(|line| line.starts_with("scp")
4050 && line.contains(&fixture.worker_binary.display().to_string())),
4051 "a cached worker must not be re-uploaded, got {lines:#?}"
4052 );
4053 assert!(
4054 !lines.iter().any(|line| line.contains("'mv'")),
4055 "a cache hit must not rename anything, got {lines:#?}"
4056 );
4057 assert!(
4058 lines.iter().any(|line| line.contains("'podman' 'cp'")
4059 && line.contains(&format!("'{cache_dir}/hel'"))),
4060 "expected podman cp to read the cached worker, got {lines:#?}"
4061 );
4062 for name in ["launch.json", "ownership.json"] {
4063 assert!(
4064 lines.iter().any(|line| line.starts_with("scp")
4065 && line.ends_with(&format!(
4066 "user@example.test:.cache/mjolnir/uploads/{session}/{name}"
4067 ))),
4068 "expected {name} to still be uploaded per session, got {lines:#?}"
4069 );
4070 }
4071 }
4072
4073 #[test]
4074 fn ssh_docker_install_uses_docker_for_remote_container_operations() {
4075 let mut fixture = podman_install_fixture();
4076 fixture.locator = targets::TargetLocator::SshDocker {
4077 ssh: SshTarget {
4078 destination: "user@example.test".into(),
4079 ssh_args: Vec::new(),
4080 },
4081 container_id: "container-1".into(),
4082 };
4083 let executor = PodmanInstallExecutor {
4084 commands: RefCell::new(Vec::new()),
4085 worker_cached: true,
4086 };
4087 install_worker_files(
4088 &executor,
4089 &fixture.locator,
4090 "0123456789abcdef0123456789abcdef",
4091 "/workspace/.hel/worker",
4092 "/workspace/.hel/profile",
4093 &fixture.worker_binary,
4094 &fixture.launch_config,
4095 &fixture.ownership,
4096 &fixture.profile_stage,
4097 )
4098 .unwrap();
4099
4100 let lines = rendered(&executor.commands.borrow());
4101 assert!(
4102 lines.iter().any(|line| line.contains("'docker' 'cp'")),
4103 "expected Docker to copy the cached worker, got {lines:#?}"
4104 );
4105 assert!(
4106 lines.iter().any(|line| line.contains("'docker' 'exec'")),
4107 "expected Docker to prepare the worker directories, got {lines:#?}"
4108 );
4109 assert!(
4110 !lines.iter().any(|line| line.contains("'podman'")),
4111 "Docker installation accidentally used Podman: {lines:#?}"
4112 );
4113 }
4114
4115 #[test]
4116 #[ignore = "requires Docker and the locally installed agent-dev image"]
4117 fn docker_uploads_and_replacements_are_usable_by_the_non_root_worker() {
4118 let fixture = podman_install_fixture();
4119 let session = mj_core::state::new_session_id().unwrap();
4120 let container_id = targets::resource_name(&session).unwrap();
4121 let locator = targets::TargetLocator::LocalDocker {
4122 container_id: container_id.clone(),
4123 };
4124 execute_checked(
4125 &ProcessExecutor,
4126 CommandSpec::new(
4127 "docker",
4128 [
4129 "run",
4130 "--pull=never",
4131 "-d",
4132 "--name",
4133 &container_id,
4134 "ghcr.io/brokkai/mjolnir/agent-dev:latest",
4135 "sleep",
4136 "infinity",
4137 ],
4138 ),
4139 )
4140 .unwrap();
4141 let result = (|| -> Result<()> {
4142 let root = targets::worker_root(&locator, &session)?;
4143 let profile = format!("{root}/profile");
4144 std::fs::write(fixture.profile_stage.join("credential"), "private")?;
4145 install_worker_files(
4146 &ProcessExecutor,
4147 &locator,
4148 &session,
4149 &root,
4150 &profile,
4151 &fixture.worker_binary,
4152 &fixture.launch_config,
4153 &fixture.ownership,
4154 &fixture.profile_stage,
4155 )?;
4156 replace_installed_worker_binary(
4157 &ProcessExecutor,
4158 &locator,
4159 &session,
4160 &fixture.worker_binary,
4161 )?;
4162 execute_checked(
4163 &ProcessExecutor,
4164 CommandSpec::new(
4165 "docker",
4166 [
4167 "exec",
4168 &container_id,
4169 "sh",
4170 "-c",
4171 "test \"$(id -u)\" != 0 && test -x \"$1/hel\" && test -r \"$1/launch.json\" && test -r \"$1/ownership.json\" && test -r \"$1/profile/credential\" && test -w \"$1/profile/credential\"",
4172 "sh",
4173 &root,
4174 ],
4175 ),
4176 )?;
4177 Ok(())
4178 })();
4179 let cleanup = execute_checked(
4180 &ProcessExecutor,
4181 CommandSpec::new("docker", ["rm", "-f", &container_id]),
4182 );
4183 result.unwrap();
4184 cleanup.unwrap();
4185 }
4186
4187 #[test]
4188 fn replacing_an_installed_podman_worker_writes_through_a_next_path() {
4189 struct RecordingExecutor {
4190 commands: RefCell<Vec<CommandSpec>>,
4191 }
4192 impl CommandExecutor for RecordingExecutor {
4193 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
4194 self.commands.borrow_mut().push(command.clone());
4195 Ok(CommandOutput {
4196 status: 0,
4197 stdout: Vec::new(),
4198 stderr: Vec::new(),
4199 })
4200 }
4201 }
4202
4203 let session = "0123456789abcdef0123456789abcdef";
4204 let container_id = targets::resource_name(session).unwrap();
4205 let locator = targets::TargetLocator::LocalPodman {
4206 container_id: container_id.clone(),
4207 workspace_storage: Default::default(),
4208 };
4209 let executor = RecordingExecutor {
4210 commands: RefCell::new(Vec::new()),
4211 };
4212 replace_installed_worker_binary(&executor, &locator, session, Path::new("/controller/hel"))
4213 .unwrap();
4214
4215 let mut lines = rendered(&executor.commands.borrow());
4216 let ownership = lines.remove(1);
4217 assert!(ownership.starts_with(&format!("podman exec --user 0 {container_id} sh -c")));
4218 assert!(ownership.contains("chown -R"));
4219 assert!(ownership.ends_with(&format!("/var/lib/hel/workers/{session}/hel.next")));
4220 assert_eq!(
4221 lines,
4222 vec![
4223 format!(
4224 "podman cp /controller/hel {container_id}:/var/lib/hel/workers/{session}/hel.next"
4225 ),
4226 format!(
4227 "podman exec {container_id} mv -f /var/lib/hel/workers/{session}/hel.next /var/lib/hel/workers/{session}/hel"
4228 ),
4229 format!("podman exec {container_id} chmod 700 /var/lib/hel/workers/{session}/hel"),
4230 ]
4231 );
4232 }
4233 #[test]
4234 fn default_bridges_pin_command_capable_adapter_versions() {
4235 let (codex_command, codex_arguments) = bridge_launch(
4236 mj_core::config::HarnessKind::Codex,
4237 ExecutionPolicy::Unconstrained,
4238 );
4239 assert_eq!(codex_command, "sh");
4240 assert_eq!(codex_arguments[0], "-c");
4241 assert!(codex_arguments[1].contains("@brokkai/codex-acp@1.11.3"));
4242 assert!(codex_arguments[1].contains("codex-acp --version"));
4243
4244 let (claude_command, claude_arguments) = bridge_launch(
4245 mj_core::config::HarnessKind::Claude,
4246 ExecutionPolicy::Unconstrained,
4247 );
4248 assert_eq!(claude_command, "sh");
4249 assert_eq!(claude_arguments[0], "-c");
4250 assert!(claude_arguments[1].contains("@agentclientprotocol/claude-agent-acp@0.73.0"));
4251
4252 let (deepseek_command, deepseek_arguments) = bridge_launch(
4253 mj_core::config::HarnessKind::Deepseek,
4254 ExecutionPolicy::Unconstrained,
4255 );
4256 assert_eq!(deepseek_command, "sh");
4257 assert_eq!(deepseek_arguments[0], "-c");
4258 assert!(deepseek_arguments[1].contains("@deepseek-ai/dsh@0.1.2-rc.1"));
4259 assert!(deepseek_arguments[1].contains("dsh --profile acp"));
4260 assert!(deepseek_arguments[1].contains("dsh --version"));
4261 assert!(!deepseek_arguments[1].contains("npx -y -p @deepseek-ai/dsh"));
4262 assert!(deepseek_arguments[1].contains("Mjolnir needs @deepseek-ai/dsh"));
4263 assert!(!deepseek_arguments[1].contains("Hel"));
4264 }
4265
4266 #[test]
4267 fn readiness_stage_names_only_install_capable_default_harnesses() {
4268 let profile = |kind| mj_core::config::HarnessProfile {
4269 enabled: true,
4270 kind,
4271 home: PathBuf::from("/profiles/test"),
4272 environment: BTreeMap::new(),
4273 context_window_bytes: None,
4274 };
4275
4276 for harness in [
4277 HarnessKind::Codex,
4278 HarnessKind::Claude,
4279 HarnessKind::Kimi,
4280 HarnessKind::Grok,
4281 ] {
4282 assert_eq!(
4283 bridge_readiness_stage(&profile(harness)),
4284 ProvisionStage::Installing(harness)
4285 );
4286 }
4287 assert_eq!(
4288 bridge_readiness_stage(&profile(HarnessKind::Deepseek)),
4289 ProvisionStage::Starting
4290 );
4291 }
4292 #[test]
4293 fn codex_execution_environment_follows_the_target_policy() {
4294 let mut podman_environment =
4295 BTreeMap::from([("INITIAL_AGENT_MODE".to_owned(), "read-only".to_owned())]);
4296 mj_core::config::HarnessKind::Codex
4297 .configure_execution_environment(
4298 ExecutionPolicy::Unconstrained,
4299 &mut podman_environment,
4300 )
4301 .unwrap();
4302 assert_eq!(
4303 podman_environment
4304 .get("INITIAL_AGENT_MODE")
4305 .map(String::as_str),
4306 Some("agent-full-access")
4307 );
4308
4309 let mut bare_environment =
4310 BTreeMap::from([("INITIAL_AGENT_MODE".to_owned(), "read-only".to_owned())]);
4311 mj_core::config::HarnessKind::Codex
4312 .configure_execution_environment(
4313 ExecutionPolicy::ConfiguredApprovals,
4314 &mut bare_environment,
4315 )
4316 .unwrap();
4317 assert_eq!(
4318 bare_environment
4319 .get("INITIAL_AGENT_MODE")
4320 .map(String::as_str),
4321 Some("agent"),
4322 "Codex uses guardian on raw localhost"
4323 );
4324 }
4325 #[test]
4326 fn bare_targets_use_managed_harnesses_but_containers_stay_ambient() {
4327 let ssh = SshTarget {
4328 destination: "user@example.test".into(),
4329 ssh_args: Vec::new(),
4330 };
4331 let targets = [
4332 (
4333 targets::TargetLocator::LocalBare {
4334 worker_root: "/worker".into(),
4335 },
4336 HarnessRuntimePolicy::Managed,
4337 ),
4338 (
4339 targets::TargetLocator::LocalPodman {
4340 container_id: "container".into(),
4341 workspace_storage: Default::default(),
4342 },
4343 HarnessRuntimePolicy::Ambient,
4344 ),
4345 (
4346 targets::TargetLocator::SshBare {
4347 worker_id: None,
4348 ssh: ssh.clone(),
4349 workspace: "/workspace/session".into(),
4350 },
4351 HarnessRuntimePolicy::Managed,
4352 ),
4353 (
4354 targets::TargetLocator::AwsEc2 {
4355 profile: "profile".into(),
4356 region: "us-east-1".into(),
4357 instance_id: "i-test".into(),
4358 ssh,
4359 workspace: "/workspace/session".into(),
4360 },
4361 HarnessRuntimePolicy::Managed,
4362 ),
4363 ];
4364
4365 for (target, expected) in targets {
4366 assert_eq!(harness_runtime_policy(&target), expected, "{target:?}");
4367 }
4368 }
4369 #[test]
4370 fn grok_sandbox_environment_follows_the_target_policy() {
4371 let mut isolated = BTreeMap::from([("GROK_SANDBOX".to_owned(), "strict".to_owned())]);
4372 mj_core::config::HarnessKind::Grok
4373 .configure_execution_environment(ExecutionPolicy::Unconstrained, &mut isolated)
4374 .unwrap();
4375 assert_eq!(
4376 isolated.get("GROK_SANDBOX").map(String::as_str),
4377 Some("off")
4378 );
4379
4380 let mut local = BTreeMap::from([("GROK_SANDBOX".to_owned(), "strict".to_owned())]);
4381 mj_core::config::HarnessKind::Grok
4382 .configure_execution_environment(ExecutionPolicy::ConfiguredApprovals, &mut local)
4383 .unwrap();
4384 assert_eq!(
4385 local.get("GROK_SANDBOX").map(String::as_str),
4386 Some("strict"),
4387 "raw localhost must preserve the profile's configured sandbox"
4388 );
4389 }
4390 #[test]
4391 fn bridge_fallback_pins_match_the_agent_dev_containerfile() {
4392 const CONTAINERFILE: &str = include_str!("../../../containers/Containerfile.agent-dev");
4393
4394 let codex = format!("codex-acp@{CODEX_ACP_VERSION}");
4395 assert!(
4396 CONTAINERFILE.contains(&codex),
4397 "containers/Containerfile.agent-dev must install {codex}. The image and the \
4398 bridge_launch() npx fallbacks have to stay in lockstep, otherwise a container \
4399 session and an npx session run different adapter versions."
4400 );
4401
4402 let claude = format!("claude-agent-acp@{CLAUDE_ACP_VERSION}");
4403 assert!(
4404 CONTAINERFILE.contains(&claude),
4405 "containers/Containerfile.agent-dev must install {claude}. The image and the \
4406 bridge_launch() npx fallbacks have to stay in lockstep, otherwise a container \
4407 session and an npx session run different adapter versions."
4408 );
4409
4410 let deepseek = format!("@deepseek-ai/dsh@{DEEPSEEK_DSH_VERSION}");
4411 assert!(
4412 CONTAINERFILE.contains(&deepseek),
4413 "containers/Containerfile.agent-dev must install {deepseek}"
4414 );
4415 assert!(!CONTAINERFILE.contains("dsh-acp-server"));
4416 }
4417 #[test]
4418 fn kimi_default_bridge_is_non_login_and_uses_bash_for_the_official_installer() {
4419 let (command, arguments) = bridge_launch(
4420 mj_core::config::HarnessKind::Kimi,
4421 ExecutionPolicy::Unconstrained,
4422 );
4423 assert_eq!(command, "sh");
4424 assert_eq!(arguments[0], "-c");
4425 assert!(arguments[1].contains("install.sh | bash &&"));
4426 assert!(arguments[1].contains("$HOME/.kimi-code/bin/kimi"));
4427 assert!(arguments[1].contains("Mjolnir needs compatible Kimi Code"));
4428 assert!(!arguments[1].contains("Hel"));
4429 }
4430 #[test]
4431 fn grok_default_bridge_is_non_login_and_uses_bash_for_the_official_installer() {
4432 let (command, arguments) = bridge_launch(
4433 mj_core::config::HarnessKind::Grok,
4434 ExecutionPolicy::ConfiguredApprovals,
4435 );
4436 assert_eq!(command, "sh");
4437 assert_eq!(arguments[0], "-c");
4438 let script = &arguments[1];
4439 assert!(script.contains("https://x.ai/cli/install.sh | bash &&"));
4440 assert!(script.contains("command -v grok"));
4441 assert!(script.contains("[ -x \"$GROK_HOME/bin/grok\" ]"));
4442 assert!(script.contains("[ -x \"$HOME/.grok/bin/grok\" ]"));
4443 assert!(script.contains("exit 127"));
4444 assert!(script.contains("exec grok agent stdio"));
4445 assert!(!script.contains("--always-approve"));
4446 assert!(script.contains("Mjolnir needs compatible Grok Build"));
4447 assert!(!script.contains("Hel"));
4448 }
4449 #[test]
4450 fn node_bootstrap_errors_name_mjolnir() {
4451 let script = ensure_node_script();
4452 assert!(script.contains("Mjolnir needs Node.js, npm, and npx"));
4453 assert!(!script.contains("sudo"));
4454 assert!(!script.contains("apt-get"));
4455 assert!(!script.contains("Hel"));
4456 }
4457 #[test]
4458 fn grok_default_bridge_adds_the_always_approve_flag_when_unrestricted() {
4459 let (_, arguments) = bridge_launch(
4460 mj_core::config::HarnessKind::Grok,
4461 ExecutionPolicy::Unconstrained,
4462 );
4463 let script = &arguments[1];
4464 assert!(script.contains("exec grok agent --always-approve stdio"));
4465 assert!(script.contains("exec \"$GROK_HOME/bin/grok\" agent --always-approve stdio"));
4466 assert!(script.contains("exec \"$HOME/.grok/bin/grok\" agent --always-approve stdio"));
4467 }
4468 #[test]
4469 fn kimi_uses_runtime_aware_memory_delivery_only_on_staged_targets() {
4470 let local = targets::TargetLocator::LocalBare {
4471 worker_root: "/worker".into(),
4472 };
4473 let podman = targets::TargetLocator::LocalPodman {
4474 container_id: "container".into(),
4475 workspace_storage: Default::default(),
4476 };
4477
4478 assert_eq!(
4479 project_memory_mcp_delivery(mj_core::config::HarnessKind::Kimi, &local),
4480 ProjectMemoryMcpDelivery::Acp
4481 );
4482 assert_eq!(
4483 project_memory_mcp_delivery(mj_core::config::HarnessKind::Kimi, &podman),
4484 ProjectMemoryMcpDelivery::HarnessProfile
4485 );
4486 assert_eq!(
4487 project_memory_mcp_delivery(mj_core::config::HarnessKind::Codex, &podman),
4488 ProjectMemoryMcpDelivery::Acp
4489 );
4490 }
4491 #[test]
4492 fn stage_grok_profile_copies_authentication_and_agent_identity() {
4493 let home = tempfile::tempdir().unwrap();
4494 std::fs::write(
4495 home.path().join("auth.json"),
4496 "{\"https://auth.x.ai::1\":{}}",
4497 )
4498 .unwrap();
4499 std::fs::write(home.path().join("agent_id"), "stable-agent-id").unwrap();
4500 std::fs::write(home.path().join("config.toml"), "model = \"grok-4.6\"\n").unwrap();
4501 std::fs::create_dir(home.path().join("sessions")).unwrap();
4503 std::fs::write(home.path().join("sessions/session_search.sqlite"), "x").unwrap();
4504 let staged = tempfile::tempdir().unwrap();
4505 let profile = mj_core::config::HarnessProfile {
4506 enabled: true,
4507 kind: mj_core::config::HarnessKind::Grok,
4508 home: home.path().to_path_buf(),
4509 environment: BTreeMap::new(),
4510 context_window_bytes: None,
4511 };
4512
4513 stage_profile(&profile, staged.path()).unwrap();
4514
4515 assert_eq!(
4516 std::fs::read_to_string(staged.path().join("agent_id")).unwrap(),
4517 "stable-agent-id"
4518 );
4519 assert!(staged.path().join("auth.json").is_file());
4520 assert!(staged.path().join("config.toml").is_file());
4521 assert!(!staged.path().join("sessions").exists());
4522 }
4523 #[test]
4524 fn stage_claude_profile_preserves_rollout_identity() {
4525 let home = tempfile::tempdir().unwrap();
4526 let identity = r#"{
4527 "machineID": "stable-machine",
4528 "userID": "stable-user",
4529 "cachedGrowthBookFeatures": {
4530 "tengu_velvet_mallet_fable_5": true
4531 }
4532 }"#;
4533 std::fs::write(home.path().join(".claude.json"), identity).unwrap();
4534 let staged = tempfile::tempdir().unwrap();
4535 let profile = mj_core::config::HarnessProfile {
4536 enabled: true,
4537 kind: mj_core::config::HarnessKind::Claude,
4538 home: home.path().to_path_buf(),
4539 environment: BTreeMap::new(),
4540 context_window_bytes: None,
4541 };
4542
4543 stage_profile(&profile, staged.path()).unwrap();
4544
4545 assert_eq!(
4546 std::fs::read_to_string(staged.path().join(".claude.json")).unwrap(),
4547 identity
4548 );
4549 }
4550 #[test]
4551 fn stage_kimi_profile_preserves_device_identity() {
4552 let home = tempfile::tempdir().unwrap();
4553 std::fs::write(home.path().join("config.toml"), "default_model = \"k3\"\n").unwrap();
4554 std::fs::write(home.path().join("device_id"), "stable-device-id").unwrap();
4555 std::fs::create_dir(home.path().join("credentials")).unwrap();
4556 std::fs::write(
4557 home.path().join("credentials/kimi-code.json"),
4558 "{\"access_token\":\"secret\"}",
4559 )
4560 .unwrap();
4561 let staged = tempfile::tempdir().unwrap();
4562 let profile = mj_core::config::HarnessProfile {
4563 enabled: true,
4564 kind: mj_core::config::HarnessKind::Kimi,
4565 home: home.path().to_path_buf(),
4566 environment: BTreeMap::new(),
4567 context_window_bytes: None,
4568 };
4569
4570 stage_profile(&profile, staged.path()).unwrap();
4571
4572 assert_eq!(
4573 std::fs::read_to_string(staged.path().join("device_id")).unwrap(),
4574 "stable-device-id"
4575 );
4576 assert!(staged.path().join("credentials/kimi-code.json").is_file());
4577 }
4578 #[test]
4579 fn staged_kimi_profile_binds_project_memory_to_the_target_runtime() {
4580 let home = tempfile::tempdir().unwrap();
4581 let original = serde_json::json!({
4582 "mcpServers": {
4583 "user-server": {
4584 "command": "user-mcp",
4585 "args": ["serve"]
4586 }
4587 },
4588 "userSetting": true
4589 });
4590 let original_body = serde_json::to_vec_pretty(&original).unwrap();
4591 std::fs::write(home.path().join("mcp.json"), &original_body).unwrap();
4592 let staged = tempfile::tempdir().unwrap();
4593 let profile = mj_core::config::HarnessProfile {
4594 enabled: true,
4595 kind: mj_core::config::HarnessKind::Kimi,
4596 home: home.path().to_path_buf(),
4597 environment: BTreeMap::new(),
4598 context_window_bytes: None,
4599 };
4600 stage_profile(&profile, staged.path()).unwrap();
4601 let memory = ProjectMemoryLaunchConfig {
4602 project_key: "project".into(),
4603 root: "/var/lib/hel/profiles/session/projects/project/memory".into(),
4604 baseline_root: PathBuf::new(),
4605 repository_roots: BTreeMap::new(),
4606 mcp_delivery: ProjectMemoryMcpDelivery::HarnessProfile,
4607 };
4608
4609 configure_kimi_project_memory_mcp(staged.path(), "/var/lib/hel/workers/session", &memory)
4610 .unwrap();
4611
4612 let configured: serde_json::Value =
4613 serde_json::from_slice(&std::fs::read(staged.path().join("mcp.json")).unwrap())
4614 .unwrap();
4615 assert_eq!(configured["userSetting"], true);
4616 assert_eq!(
4617 configured["mcpServers"]["user-server"]["command"],
4618 "user-mcp"
4619 );
4620 assert_eq!(
4621 configured["mcpServers"]["mj-project-memory"],
4622 serde_json::json!({
4623 "transport": "stdio",
4624 "command": "/var/lib/hel/workers/session/hel",
4625 "args": [
4626 "worker",
4627 "memory-mcp",
4628 "--root",
4629 "/var/lib/hel/profiles/session/projects/project/memory"
4630 ],
4631 "runtime_id": "local"
4632 })
4633 );
4634 assert_eq!(
4635 std::fs::read(home.path().join("mcp.json")).unwrap(),
4636 original_body,
4637 "the controller-side Kimi profile must remain unchanged"
4638 );
4639 }
4640
4641 #[test]
4642 fn staged_kimi_project_memory_resolves_ssh_paths_from_target_home() {
4643 let staged = tempfile::tempdir().unwrap();
4644 let memory = ProjectMemoryLaunchConfig {
4645 project_key: "project".into(),
4646 root: ".local/share/hel/profiles/session/projects/project/memory".into(),
4647 baseline_root: PathBuf::new(),
4648 repository_roots: BTreeMap::new(),
4649 mcp_delivery: ProjectMemoryMcpDelivery::HarnessProfile,
4650 };
4651
4652 configure_kimi_project_memory_mcp(
4653 staged.path(),
4654 ".local/share/hel/workers/session",
4655 &memory,
4656 )
4657 .unwrap();
4658
4659 let configured: serde_json::Value =
4660 serde_json::from_slice(&std::fs::read(staged.path().join("mcp.json")).unwrap())
4661 .unwrap();
4662 let server = &configured["mcpServers"]["mj-project-memory"];
4663 assert_eq!(server["command"], "sh");
4664 assert_eq!(server["runtime_id"], "local");
4665 assert_eq!(
4666 server["args"],
4667 serde_json::json!([
4668 "-c",
4669 "exec \"$HOME/$1\" worker memory-mcp --root \"$HOME/$2\"",
4670 "mj-project-memory",
4671 ".local/share/hel/workers/session/hel",
4672 ".local/share/hel/profiles/session/projects/project/memory"
4673 ])
4674 );
4675 }
4676 #[test]
4677 fn stage_deepseek_profile_copies_only_portable_configuration() {
4678 let home = tempfile::tempdir().unwrap();
4679 std::fs::write(
4680 home.path().join(".credentials.yaml"),
4681 "version: 1\nrefs: {}\n",
4682 )
4683 .unwrap();
4684 std::fs::write(home.path().join("settings.yaml"), "models: {}\n").unwrap();
4685 std::fs::create_dir(home.path().join("sessions")).unwrap();
4686 std::fs::write(home.path().join("sessions/native-session"), "private state").unwrap();
4687 std::fs::create_dir(home.path().join("profiles")).unwrap();
4688 let staged = tempfile::tempdir().unwrap();
4689 let profile = mj_core::config::HarnessProfile {
4690 enabled: true,
4691 kind: mj_core::config::HarnessKind::Deepseek,
4692 home: home.path().to_path_buf(),
4693 environment: BTreeMap::new(),
4694 context_window_bytes: None,
4695 };
4696
4697 stage_profile(&profile, staged.path()).unwrap();
4698
4699 assert!(staged.path().join(".credentials.yaml").is_file());
4700 assert!(staged.path().join("settings.yaml").is_file());
4701 assert!(!staged.path().join("sessions").exists());
4702 assert!(!staged.path().join("profiles").exists());
4703 }
4704 #[test]
4705 fn disposable_container_guidance_reaches_each_harness_without_touching_home() {
4706 let target = targets::TargetLocator::LocalPodman {
4707 container_id: "container".into(),
4708 workspace_storage: Default::default(),
4709 };
4710 for (kind, instructions) in [
4711 (mj_core::config::HarnessKind::Codex, "AGENTS.md"),
4712 (mj_core::config::HarnessKind::Claude, "CLAUDE.md"),
4713 (mj_core::config::HarnessKind::Kimi, "AGENTS.md"),
4714 (mj_core::config::HarnessKind::Grok, "AGENTS.md"),
4715 (mj_core::config::HarnessKind::Deepseek, "AGENTS.md"),
4716 (mj_core::config::HarnessKind::Muse, "AGENTS.md"),
4717 ] {
4718 let home = tempfile::tempdir().unwrap();
4719 let original = "# Controller instructions\n\nKeep this source unchanged.\n";
4720 let source_instructions = home.path().join(instructions);
4721 std::fs::write(&source_instructions, original).unwrap();
4722 let staged = tempfile::tempdir().unwrap();
4723 let profile = mj_core::config::HarnessProfile {
4724 enabled: true,
4725 kind,
4726 home: home.path().to_path_buf(),
4727 environment: std::collections::BTreeMap::new(),
4728 context_window_bytes: None,
4729 };
4730
4731 stage_profile(&profile, staged.path()).unwrap();
4732 append_hel_target_environment(kind, staged.path(), &target).unwrap();
4733
4734 let guidance = std::fs::read_to_string(staged.path().join(instructions)).unwrap();
4735 assert_eq!(
4736 guidance,
4737 format!("{original}\n{MJ_CONTAINER_ENVIRONMENT}"),
4738 "{instructions} receives the section in the staged profile"
4739 );
4740 assert!(guidance.contains("## Mjolnir disposable environment"));
4741 assert!(!guidance.contains("## Hel disposable environment"));
4742 assert_eq!(
4743 std::fs::read_to_string(source_instructions).unwrap(),
4744 original,
4745 "{instructions} in the controller-side home stays untouched"
4746 );
4747 }
4748 }
4749 #[test]
4750 fn kimi_guidance_uses_agents_md_without_mutating_the_system_override() {
4751 let home = tempfile::tempdir().unwrap();
4752 let system_override = "# Custom Kimi system prompt\n";
4753 std::fs::write(home.path().join("SYSTEM.md"), system_override).unwrap();
4754 let staged = tempfile::tempdir().unwrap();
4755 let profile = mj_core::config::HarnessProfile {
4756 enabled: true,
4757 kind: mj_core::config::HarnessKind::Kimi,
4758 home: home.path().to_path_buf(),
4759 environment: std::collections::BTreeMap::new(),
4760 context_window_bytes: None,
4761 };
4762
4763 stage_profile(&profile, staged.path()).unwrap();
4764 append_hel_target_environment(
4765 profile.kind,
4766 staged.path(),
4767 &targets::TargetLocator::LocalPodman {
4768 container_id: "container".into(),
4769 workspace_storage: Default::default(),
4770 },
4771 )
4772 .unwrap();
4773
4774 assert_eq!(
4775 std::fs::read_to_string(staged.path().join("AGENTS.md")).unwrap(),
4776 MJ_CONTAINER_ENVIRONMENT
4777 );
4778 assert_eq!(
4779 std::fs::read_to_string(staged.path().join("SYSTEM.md")).unwrap(),
4780 system_override
4781 );
4782 assert!(!home.path().join("AGENTS.md").exists());
4783 assert_eq!(
4784 std::fs::read_to_string(home.path().join("SYSTEM.md")).unwrap(),
4785 system_override
4786 );
4787 }
4788
4789 #[test]
4790 fn ec2_guidance_names_its_real_workspace_and_ssh_bare_gets_none() {
4791 let ec2 = tempfile::tempdir().unwrap();
4792 append_hel_target_environment(
4793 mj_core::config::HarnessKind::Codex,
4794 ec2.path(),
4795 &targets::TargetLocator::AwsEc2 {
4796 profile: "profile".into(),
4797 region: "region".into(),
4798 instance_id: "instance".into(),
4799 ssh: targets::SshTarget {
4800 destination: "host".into(),
4801 ssh_args: Vec::new(),
4802 },
4803 workspace: ".local/share/hel/workspaces/session".into(),
4804 },
4805 )
4806 .unwrap();
4807 let guidance = std::fs::read_to_string(ec2.path().join("AGENTS.md")).unwrap();
4808 assert_eq!(
4809 guidance,
4810 "## Mjolnir disposable environment\n\nThis session runs on a disposable Mjolnir EC2 instance. When the session closes, Mjolnir checkpoints everything in project workspace directories under `$HOME/.local/share/hel/workspaces/session`, including committed work, staged and unstaged changes, and untracked files. Mjolnir then terminates the instance.\n\nEverything outside `$HOME/.local/share/hel/workspaces/session`, including installed packages, the rest of `$HOME`, and `/tmp`, is ephemeral and will be lost. Keep durable results in the workspace or push them to a remote.\n\nNew workspaces start on their own session branch from the default network fetch remote’s default branch. Local unpublished commits and uncommitted files are not copied. Use normal git push to publish the current branch to the configured network push destination. Closing saves a checkpoint; it does not publish commits or update the original local checkout. Resumed sessions restore their saved work.\n"
4811 );
4812 assert!(!guidance.contains("## Hel disposable environment"));
4813
4814 let ssh_bare = tempfile::tempdir().unwrap();
4815 append_hel_target_environment(
4816 mj_core::config::HarnessKind::Codex,
4817 ssh_bare.path(),
4818 &targets::TargetLocator::SshBare {
4819 worker_id: None,
4820 ssh: targets::SshTarget {
4821 destination: "host".into(),
4822 ssh_args: Vec::new(),
4823 },
4824 workspace: ".local/share/hel/workspaces/session".into(),
4825 },
4826 )
4827 .unwrap();
4828 assert!(!ssh_bare.path().join("AGENTS.md").exists());
4829 }
4830
4831 #[test]
4832 fn project_memory_replicas_are_session_private() {
4833 let key = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
4834 assert_eq!(
4835 project_memory_replica_slug(key, "session-a"),
4836 "hel-0123456789abcdef-session-a"
4837 );
4838 assert_ne!(
4839 project_memory_replica_slug(key, "session-a"),
4840 project_memory_replica_slug(key, "session-b")
4841 );
4842 }
4843
4844 struct DigestExecutor {
4847 installed_line: String,
4848 commands: RefCell<Vec<CommandSpec>>,
4849 }
4850
4851 impl CommandExecutor for DigestExecutor {
4852 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
4853 self.commands.borrow_mut().push(command.clone());
4854 Ok(CommandOutput {
4855 status: 0,
4856 stdout: self.installed_line.clone().into_bytes(),
4857 stderr: Vec::new(),
4858 })
4859 }
4860 }
4861
4862 fn ssh_bare_locator(session_id: &str) -> targets::TargetLocator {
4865 targets::TargetLocator::SshBare {
4866 worker_id: None,
4867 ssh: SshTarget {
4868 destination: "user@host.test".into(),
4869 ssh_args: Vec::new(),
4870 },
4871 workspace: format!("/srv/mj/{session_id}"),
4872 }
4873 }
4874
4875 #[test]
4876 fn remote_upgrade_prepares_managed_harness_without_touching_running_worker() {
4877 let session = "session-remote";
4878 let executor = DigestExecutor {
4879 installed_line: String::new(),
4880 commands: RefCell::new(Vec::new()),
4881 };
4882 let launch = WorkerLaunchConfig {
4883 subagent_tools: false,
4884 goal_resume_request: Default::default(),
4885 target_environment: Default::default(),
4886 run_mode: Default::default(),
4887 session_id: session.into(),
4888 harness: HarnessKind::Codex,
4889 bridge_command: "ignored".into(),
4890 bridge_args: Vec::new(),
4891 harness_runtime: HarnessRuntimePolicy::Managed,
4892 environment: BTreeMap::new(),
4893 cwd: "/srv/mj/session-remote/project".into(),
4894 additional_directories: Vec::new(),
4895 native_session_id: None,
4896 project_memory: None,
4897 execution_policy: ExecutionPolicy::ConfiguredApprovals,
4898 };
4899
4900 prepare_managed_harness_for_upgrade(
4901 &executor,
4902 &ssh_bare_locator(session),
4903 session,
4904 Path::new("/controller/hel"),
4905 &launch,
4906 )
4907 .unwrap();
4908
4909 let commands = executor.commands.borrow();
4910 let purposes = commands
4911 .iter()
4912 .map(|command| command.purpose.as_str())
4913 .collect::<Vec<_>>();
4914 assert_eq!(
4915 purposes,
4916 vec![
4917 "clear managed harness preparation staging",
4918 "create managed harness preparation staging",
4919 "stage current worker for managed harness preparation",
4920 "stage managed harness launch configuration",
4921 "make managed harness preparation worker executable",
4922 "prepare exact managed harness",
4923 "remove managed harness preparation staging",
4924 ]
4925 );
4926 assert!(commands.iter().all(|command| {
4927 !command.purpose.contains("stop Mjolnir worker")
4928 && !command.purpose.contains("start Mjolnir worker")
4929 && !command
4930 .purpose
4931 .contains("install the current Mjolnir worker binary")
4932 }));
4933 let prepare = commands
4934 .iter()
4935 .find(|command| command.purpose == "prepare exact managed harness")
4936 .unwrap();
4937 let rendered = format!("{} {}", prepare.program, prepare.args.join(" "));
4938 assert!(rendered.contains("worker' 'prepare-harness' '--config'"));
4939 }
4940
4941 #[test]
4942 fn local_upgrade_preflight_uses_current_binary_and_preserves_launch_policy() {
4943 struct ConfigRecordingExecutor {
4944 command: RefCell<Option<CommandSpec>>,
4945 launch: RefCell<Option<WorkerLaunchConfig>>,
4946 }
4947
4948 impl CommandExecutor for ConfigRecordingExecutor {
4949 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
4950 let config_path = command
4951 .args
4952 .get(3)
4953 .context("local prepare command did not include its config path")?;
4954 *self.command.borrow_mut() = Some(command.clone());
4955 *self.launch.borrow_mut() = Some(WorkerLaunchConfig::read(Path::new(config_path))?);
4956 Ok(CommandOutput {
4957 status: 0,
4958 stdout: Vec::new(),
4959 stderr: Vec::new(),
4960 })
4961 }
4962 }
4963
4964 struct FailingExecutor {
4965 purposes: RefCell<Vec<String>>,
4966 }
4967
4968 impl CommandExecutor for FailingExecutor {
4969 fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
4970 self.purposes.borrow_mut().push(command.purpose.clone());
4971 Err(anyhow::anyhow!("managed harness installation failed"))
4972 }
4973 }
4974
4975 let executor = ConfigRecordingExecutor {
4976 command: RefCell::new(None),
4977 launch: RefCell::new(None),
4978 };
4979 let launch = WorkerLaunchConfig {
4980 subagent_tools: false,
4981 goal_resume_request: Default::default(),
4982 target_environment: Default::default(),
4983 run_mode: Default::default(),
4984 session_id: "session-local".into(),
4985 harness: HarnessKind::Codex,
4986 bridge_command: "ignored".into(),
4987 bridge_args: Vec::new(),
4988 harness_runtime: HarnessRuntimePolicy::Managed,
4989 environment: BTreeMap::from([("CODEX_HOME".into(), "/configured/profile/home".into())]),
4990 cwd: "/workspace/project".into(),
4991 additional_directories: Vec::new(),
4992 native_session_id: None,
4993 project_memory: None,
4994 execution_policy: ExecutionPolicy::ConfiguredApprovals,
4995 };
4996 let locator = targets::TargetLocator::LocalBare {
4997 worker_root: "/worker/session-local".into(),
4998 };
4999
5000 prepare_managed_harness_for_upgrade(
5001 &executor,
5002 &locator,
5003 "session-local",
5004 Path::new("/controller/hel"),
5005 &launch,
5006 )
5007 .unwrap();
5008
5009 {
5010 let command = executor.command.borrow();
5011 let command = command.as_ref().unwrap();
5012 assert_eq!(command.purpose, "prepare exact managed harness");
5013 assert_eq!(command.program, "/controller/hel");
5014 assert_eq!(
5015 &command.args[..3],
5016 ["worker", "prepare-harness", "--config"]
5017 );
5018 assert!(!command.args[3].contains("/worker/session-local"));
5019 }
5020
5021 let prepared = executor.launch.borrow();
5022 let prepared = prepared.as_ref().unwrap();
5023 assert_eq!(
5024 prepared.environment.get("CODEX_HOME").map(String::as_str),
5025 Some("/configured/profile/home")
5026 );
5027 assert_eq!(
5028 prepared.execution_policy,
5029 ExecutionPolicy::ConfiguredApprovals
5030 );
5031
5032 let failing = FailingExecutor {
5033 purposes: RefCell::new(Vec::new()),
5034 };
5035 let error = prepare_managed_harness_for_upgrade(
5036 &failing,
5037 &locator,
5038 "session-local",
5039 Path::new("/controller/hel"),
5040 &launch,
5041 )
5042 .unwrap_err();
5043 assert!(
5044 error
5045 .to_string()
5046 .contains("managed harness installation failed")
5047 );
5048 assert_eq!(
5049 failing.purposes.borrow().as_slice(),
5050 ["prepare exact managed harness"]
5051 );
5052 }
5053
5054 #[test]
5055 fn initial_bare_provision_prepares_the_harness_from_installed_files() {
5056 let session = "session-remote";
5057 let executor = DigestExecutor {
5058 installed_line: String::new(),
5059 commands: RefCell::new(Vec::new()),
5060 };
5061 let mut launch = WorkerLaunchConfig {
5062 subagent_tools: false,
5063 goal_resume_request: Default::default(),
5064 target_environment: Default::default(),
5065 run_mode: Default::default(),
5066 session_id: session.into(),
5067 harness: HarnessKind::Kimi,
5068 bridge_command: "ignored".into(),
5069 bridge_args: Vec::new(),
5070 harness_runtime: HarnessRuntimePolicy::Managed,
5071 environment: BTreeMap::new(),
5072 cwd: "/srv/mj/session-remote/project".into(),
5073 additional_directories: Vec::new(),
5074 native_session_id: None,
5075 project_memory: None,
5076 execution_policy: ExecutionPolicy::ConfiguredApprovals,
5077 };
5078
5079 let locator = ssh_bare_locator(session);
5080 prepare_installed_managed_harness(&executor, &locator, "/worker/root", &launch).unwrap();
5081 let commands = executor.commands.borrow();
5082 assert_eq!(commands.len(), 1);
5083 assert_eq!(
5084 commands[0].purpose,
5085 "prepare exact managed harness before worker startup"
5086 );
5087 let rendered = format!("{} {}", commands[0].program, commands[0].args.join(" "));
5088 assert!(rendered.contains("'/worker/root/hel' 'worker' 'prepare-harness'"));
5089 drop(commands);
5090
5091 let local = targets::TargetLocator::LocalBare {
5092 worker_root: "/worker/session-remote".into(),
5093 };
5094 prepare_installed_managed_harness(&executor, &local, "/worker/session-remote", &launch)
5095 .unwrap();
5096 let commands = executor.commands.borrow();
5097 assert_eq!(commands.len(), 2);
5098 assert_eq!(commands[1].program, "/worker/session-remote/hel");
5099 assert_eq!(
5100 commands[1].args,
5101 vec![
5102 "worker".to_owned(),
5103 "prepare-harness".to_owned(),
5104 "--config".to_owned(),
5105 "/worker/session-remote/launch.json".to_owned(),
5106 ]
5107 );
5108 drop(commands);
5109
5110 launch.harness_runtime = HarnessRuntimePolicy::Ambient;
5111 prepare_installed_managed_harness(&executor, &locator, "/worker/root", &launch).unwrap();
5112 assert_eq!(executor.commands.borrow().len(), 2);
5113 }
5114
5115 #[test]
5116 fn a_remote_worker_with_a_mismatched_binary_is_replaced_before_restart() {
5117 let directory = tempfile::tempdir().unwrap();
5118 let source = directory.path().join("worker");
5119 std::fs::write(&source, b"fresh musl worker").unwrap();
5120 let executor = DigestExecutor {
5121 installed_line: format!("{} /root/hel\n", "0".repeat(64)),
5122 commands: RefCell::new(Vec::new()),
5123 };
5124 let replaced = replace_remote_worker_binary_if_stale(
5125 &executor,
5126 &ssh_bare_locator("session-remote"),
5127 "session-remote",
5128 &CommandSpec::new("true", Vec::<String>::new()),
5129 &source,
5130 )
5131 .unwrap();
5132 assert!(replaced, "a stale remote binary must be replaced");
5133 assert!(
5134 executor.commands.borrow().len() > 1,
5135 "the digest probe must be followed by replacement commands"
5136 );
5137 }
5138
5139 #[test]
5140 fn a_remote_worker_already_current_is_restarted_without_recopying() {
5141 let directory = tempfile::tempdir().unwrap();
5142 let source = directory.path().join("worker");
5143 std::fs::write(&source, b"fresh musl worker").unwrap();
5144 let current = mj_core::worker_launch::worker_executable_digest(&source).unwrap();
5145 let executor = DigestExecutor {
5146 installed_line: format!("{current} /root/hel\n"),
5147 commands: RefCell::new(Vec::new()),
5148 };
5149 let replaced = replace_remote_worker_binary_if_stale(
5150 &executor,
5151 &ssh_bare_locator("session-remote"),
5152 "session-remote",
5153 &CommandSpec::new("true", Vec::<String>::new()),
5154 &source,
5155 )
5156 .unwrap();
5157 assert!(!replaced, "a current remote binary must not be recopied");
5158 assert_eq!(
5159 executor.commands.borrow().len(),
5160 1,
5161 "only the digest probe runs when the binary is already current"
5162 );
5163 }
5164
5165 #[test]
5166 fn a_remote_recovery_plan_defers_binary_refresh_to_the_recovery_task() {
5167 let locator = ssh_bare_locator("session-remote");
5168 let refresh = worker_binary_refresh_plan(&locator, "session-remote")
5169 .unwrap()
5170 .expect("a remote target now gets a binary refresh");
5171 match refresh {
5172 WorkerBinaryRefresh::Remote(remote) => {
5173 assert_eq!(remote.session_id, "session-remote");
5174 assert_eq!(remote.locator, locator);
5175 }
5176 WorkerBinaryRefresh::Prepared(_) => {
5177 panic!("a remote target must defer, not prepare, its binary refresh")
5178 }
5179 }
5180 }
5181}