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