use mj_core::config::HarnessKind;
use serde_json::json;
use super::*;
use crate::controller::test_support::test_git as git;
use mj_checkpoint::archive::*;
use mj_worker::checkpoint::*;
use std::cell::RefCell;
use mj_checkpoint::archive::{
CanonicalExecutionState, CanonicalQueuedCommandKind, CanonicalQueuedPrompt,
CanonicalSessionState, CanonicalTranscriptItem,
};
use mj_core::targets::CommandOutput;
const SESSION: &str = "018f9dd2-a3b4-7c8d-9000-123456789abc";
const NATIVE: &str = "0190aabb-ccdd-7eef-9000-abcdef012345";
fn ssh() -> crate::targets::SshTarget {
crate::targets::SshTarget {
destination: "dev@example.test".into(),
ssh_args: vec!["-p".into(), "2222".into()],
}
}
fn locators() -> Vec<TargetLocator> {
let name = mj_core::targets::resource_name(SESSION).unwrap();
vec![
TargetLocator::LocalBare {
worker_root: format!("/var/lib/hel/workers/{SESSION}"),
},
TargetLocator::LocalPodman {
borrowed_from: None,
container_id: name.clone(),
workspace_storage: Default::default(),
},
TargetLocator::AppleContainer {
borrowed_from: None,
container_id: name.clone(),
},
TargetLocator::AwsEc2 {
profile: "default".into(),
region: "us-east-1".into(),
instance_id: "i-0123456789abcdef0".into(),
ssh: ssh(),
workspace: format!("~/hel/{SESSION}"),
},
TargetLocator::SshBare {
worker_id: None,
ssh: ssh(),
workspace: format!("~/hel/{SESSION}"),
},
TargetLocator::SshPodman {
borrowed_from: None,
ssh: ssh(),
container_id: name,
workspace_storage: Default::default(),
},
]
}
#[test]
fn transfer_plans_cover_all_target_boundaries() {
let locators = locators();
let plans = locators
.iter()
.map(|locator| {
transfer_plan(
locator,
SESSION,
"/var/lib/hel/workers/checkpoint.hel.zip",
Path::new("/var/tmp/checkpoint.zip"),
&remote_staging_path(SESSION, "test-transfer").unwrap(),
)
.unwrap()
})
.collect::<Vec<_>>();
assert_eq!(plans[0].commands[0].program, "cp");
assert_eq!(plans[1].commands[0].program, "podman");
assert_eq!(plans[2].commands[0].program, "container");
assert_eq!(plans[3].commands[0].program, "scp");
assert_eq!(plans[4].commands[0].program, "scp");
assert_eq!(plans[5].commands.len(), 3);
assert!(
plans[5].commands[1]
.args
.last()
.unwrap()
.contains("'podman' 'cp'")
);
assert!(
!plans[5]
.commands
.iter()
.flat_map(|command| &command.args)
.any(|arg| arg == "--remote")
);
assert!(plans[3].commands[0].args.contains(&"-P".into()));
}
fn fixture(temp: &Path) -> (CheckpointExportSpec, PathBuf) {
let worker_root = temp.join("worker");
fs::create_dir_all(&worker_root).unwrap();
let harness_home = temp.join("codex");
let native = harness_home.join("sessions/2026/08/09");
fs::create_dir_all(&native).unwrap();
fs::write(native.join(format!("rollout-{NATIVE}.jsonl")), b"native").unwrap();
let workspace = temp.join("workspace");
let repository = workspace.join("app");
fs::create_dir_all(&repository).unwrap();
git(&repository, &["init"]);
git(&repository, &["config", "user.email", "hel@example.test"]);
git(&repository, &["config", "user.name", "Hel Test"]);
fs::write(repository.join("README.md"), b"hello").unwrap();
git(&repository, &["add", "."]);
git(&repository, &["commit", "-m", "base"]);
git(
&repository,
&[
"remote",
"add",
"origin",
"https://github.com/example/app.git",
],
);
let base = git(&repository, &["rev-parse", "HEAD"]);
let output = worker_root.join("source.hel.zip");
(
CheckpointExportSpec {
protocol_version: CHECKPOINT_EXPORT_PROTOCOL_VERSION,
session: SessionManifest {
id: SESSION.into(),
title: "test".into(),
harness_kind: HarnessKind::Codex,
profile_id: "codex-1".into(),
native_session_id: NATIVE.into(),
created_at: "2026-08-09T00:00:00Z".into(),
checkpointed_at: "2026-08-09T00:01:00Z".into(),
hel_version: "0.1.0".into(),
relay_version: "0.1.0".into(),
adapter_version: "test".into(),
},
target: TargetManifest {
template_id: "local".into(),
target_kind: "podman".into(),
details: Default::default(),
},
bundle: BundleManifest {
id: "bundle".into(),
primary_repository: "app".into(),
},
relay_root: worker_root,
harness_home,
workspace_root: workspace,
repositories: vec![CheckpointRepositorySpec {
id: "app".into(),
relative_destination: "app".into(),
capture: CheckpointRepositoryCapture::DeltaFrom { base_commit: base },
origin_override: None,
}],
canonical_session: CanonicalSessionSnapshot {
event_frontier: 1,
event_frontier_digest: "a".repeat(64),
session: CanonicalSessionState {
execution: CanonicalExecutionState::Idle,
last_activity_at_ms: Some(1),
session_title: Some("test".into()),
configuration: Default::default(),
},
transcript: vec![CanonicalTranscriptItem {
stable_id: "user-1".into(),
position: 1,
latest_content_event_ordinal: None,
created_at_ms: 1,
last_changed_at_ms: 1,
body: CanonicalTranscriptBody::User {
content: vec![json!({"type": "text", "text": "hello"})],
},
}],
queued_prompts: vec![CanonicalQueuedPrompt {
command_id: "queued-1".into(),
kind: CanonicalQueuedCommandKind::Prompt,
content: vec![json!({"type": "text", "text": "next"})],
queued_at_ms: 2,
}],
},
output_path: output.clone(),
},
output,
)
}
struct CopyExecutor {
source: PathBuf,
calls: RefCell<usize>,
}
impl CommandExecutor for CopyExecutor {
fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
*self.calls.borrow_mut() += 1;
fs::copy(
&self.source,
command.args.last().context("missing destination")?,
)?;
Ok(CommandOutput {
status: 0,
stdout: vec![],
stderr: vec![],
})
}
}
struct SshDockerTransferExecutor {
archive: Vec<u8>,
commands: RefCell<Vec<CommandSpec>>,
fail_download: bool,
fail_staging_cleanup: bool,
}
impl SshDockerTransferExecutor {
fn new(archive: Vec<u8>) -> Self {
Self {
archive,
commands: RefCell::new(Vec::new()),
fail_download: false,
fail_staging_cleanup: false,
}
}
}
impl CommandExecutor for SshDockerTransferExecutor {
fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
self.commands.borrow_mut().push(command.clone());
let remote = command.args.last().map(String::as_str).unwrap_or_default();
if command.program == "scp" {
if self.fail_download {
return Ok(CommandOutput {
status: 23,
stdout: Vec::new(),
stderr: b"scp unavailable".to_vec(),
});
}
fs::write(
command
.args
.last()
.context("missing local checkpoint path")?,
&self.archive,
)?;
}
if command.program == "ssh"
&& remote.contains("'rm' '-f' '--' '.local/share/hel/transfers/")
&& self.fail_staging_cleanup
{
return Ok(CommandOutput {
status: 19,
stdout: Vec::new(),
stderr: b"staging cleanup unavailable".to_vec(),
});
}
Ok(CommandOutput {
status: 0,
stdout: Vec::new(),
stderr: Vec::new(),
})
}
}
fn ssh_docker_locator() -> TargetLocator {
TargetLocator::SshDocker {
borrowed_from: None,
ssh: ssh(),
container_id: mj_core::targets::resource_name(SESSION).unwrap(),
}
}
#[test]
fn export_and_transfer_only_gate_after_local_verification() {
let temp = tempfile::tempdir().unwrap();
let (spec, source) = fixture(temp.path());
let target = export_checkpoint(&spec).unwrap();
assert_eq!(target.event_frontier, 1);
assert_eq!(
target.event_frontier_digest,
spec.canonical_session.event_frontier_digest
);
let destination = temp.path().join("controller/session.hel.zip");
let locator = &locators()[0];
let gate = CheckpointTransfer {
locator,
session_id: SESSION,
operation_id: "test-transfer",
remote_archive: "/var/lib/hel/workers/source.hel.zip",
destination: &destination,
expected_sha256: &target.sha256,
expected_event_frontier: 1,
expected_event_frontier_digest: &spec.canonical_session.event_frontier_digest,
}
.execute(&CopyExecutor {
source,
calls: RefCell::new(0),
})
.unwrap();
assert!(gate.teardown_allowed());
assert_eq!(gate.event_frontier(), 1);
assert_eq!(
gate.event_frontier_digest(),
spec.canonical_session.event_frontier_digest
);
assert_eq!(
read_archive_verified(&destination).unwrap().archive_sha256,
gate.sha256()
);
}
#[test]
fn a_streamed_spec_exports_the_same_archive_as_a_spec_file() {
let temp = tempfile::tempdir().unwrap();
let (mut spec, _) = fixture(temp.path());
let from_file = export_from_spec_file(&spec.output_path.with_extension("spec.json"))
.err()
.map(|error| format!("{error:#}"));
assert!(
from_file.is_some_and(|error| error.contains("read checkpoint export spec")),
"a missing spec file must still be reported as a read failure"
);
let spec_path = temp.path().join("checkpoint-spec.json");
spec.write(&spec_path).unwrap();
let from_file = export_from_spec_file(&spec_path).unwrap();
let file_archive = fs::read(&spec.output_path).unwrap();
spec.output_path = temp.path().join("worker/streamed.hel.zip");
let body = serde_json::to_vec(&spec).unwrap();
let streamed = export_from_spec_reader(&mut body.as_slice()).unwrap();
let streamed_archive = fs::read(&spec.output_path).unwrap();
assert_eq!(streamed.sha256, from_file.sha256);
assert_eq!(streamed.event_frontier, from_file.event_frontier);
assert_eq!(
streamed.event_frontier_digest,
from_file.event_frontier_digest
);
assert_eq!(streamed_archive, file_archive);
assert_eq!(
read_archive_verified(&spec.output_path)
.unwrap()
.archive_sha256,
streamed.sha256
);
}
#[test]
fn transfer_rejects_a_target_checksum_mismatch() {
let temp = tempfile::tempdir().unwrap();
let (spec, source) = fixture(temp.path());
export_checkpoint(&spec).unwrap();
let destination = temp.path().join("controller/session.hel.zip");
let unexpected_sha256 = "b".repeat(64);
let error = CheckpointTransfer {
locator: &locators()[0],
session_id: SESSION,
operation_id: "test-transfer",
remote_archive: "/var/lib/hel/workers/source.hel.zip",
destination: &destination,
expected_sha256: &unexpected_sha256,
expected_event_frontier: 1,
expected_event_frontier_digest: &spec.canonical_session.event_frontier_digest,
}
.execute(&CopyExecutor {
source,
calls: RefCell::new(0),
})
.unwrap_err();
assert!(format!("{error:#}").contains("checkpoint checksums differ"));
assert!(!destination.exists());
}
#[test]
fn ssh_docker_failed_hash_cleans_host_staging_but_preserves_container_archive() {
let temp = tempfile::tempdir().unwrap();
let destination = temp.path().join("controller/session.hel.zip");
let executor = SshDockerTransferExecutor::new(vec![b'x'; 128 * 1024]);
let error = CheckpointTransfer {
locator: &ssh_docker_locator(),
session_id: SESSION,
operation_id: "test-transfer",
remote_archive: "/var/lib/hel/workers/source.hel.zip",
destination: &destination,
expected_sha256: &"0".repeat(64),
expected_event_frontier: 1,
expected_event_frontier_digest: &"a".repeat(64),
}
.execute(&executor)
.unwrap_err();
assert!(format!("{error:#}").contains("checkpoint checksums differ"));
assert!(!destination.exists());
let commands = executor.commands.borrow();
assert!(commands.iter().any(|command| {
command.program == "ssh"
&& command.args.last().is_some_and(|remote| {
remote.contains("'rm' '-f' '--' '.local/share/hel/transfers/")
})
}));
assert!(!commands.iter().any(|command| {
command
.args
.last()
.is_some_and(|remote| remote.contains("'docker' 'exec'"))
}));
}
#[test]
fn ssh_docker_download_and_staging_cleanup_errors_keep_the_original_failure() {
let temp = tempfile::tempdir().unwrap();
let destination = temp.path().join("controller/session.hel.zip");
let mut executor = SshDockerTransferExecutor::new(vec![b'x'; 128 * 1024]);
executor.fail_download = true;
executor.fail_staging_cleanup = true;
let error = CheckpointTransfer {
locator: &ssh_docker_locator(),
session_id: SESSION,
operation_id: "test-transfer",
remote_archive: "/var/lib/hel/workers/source.hel.zip",
destination: &destination,
expected_sha256: &"0".repeat(64),
expected_event_frontier: 1,
expected_event_frontier_digest: &"a".repeat(64),
}
.execute(&executor)
.unwrap_err();
let text = format!("{error:#}");
assert!(text.contains("download target checkpoint"), "{text}");
assert!(text.contains("scp unavailable"), "{text}");
assert!(
text.contains("clean target checkpoint host staging also failed")
&& text.contains("staging cleanup unavailable"),
"{text}"
);
let commands = executor.commands.borrow();
assert!(commands.iter().any(|command| {
command.program == "ssh"
&& command.args.last().is_some_and(|remote| {
remote.contains("'rm' '-f' '--' '.local/share/hel/transfers/")
})
}));
assert!(!commands.iter().any(|command| {
command
.args
.last()
.is_some_and(|remote| remote.contains("'docker' 'exec'"))
}));
}
#[test]
fn overlapping_remote_transfers_keep_their_own_bytes_and_cleanup() {
use std::collections::BTreeMap;
use std::sync::{Condvar, Mutex};
use std::time::Duration;
#[derive(Default)]
struct Staging {
files: BTreeMap<String, Vec<u8>>,
copies: usize,
first_cleaned: bool,
}
struct InterleavedExecutor<'a> {
staging: &'a (Mutex<Staging>, Condvar),
archive: &'a [u8],
first: bool,
}
impl CommandExecutor for InterleavedExecutor<'_> {
fn execute(&self, command: &CommandSpec) -> Result<CommandOutput> {
let (mutex, changed) = self.staging;
let remote = command.args.last().unwrap();
let remote_path = || {
remote
.rsplit(' ')
.next()
.unwrap()
.trim_matches('\'')
.to_owned()
};
match command.purpose.as_str() {
"stage remote container checkpoint" => {
let mut state = mutex.lock().unwrap();
state.files.insert(remote_path(), self.archive.to_vec());
state.copies += 1;
changed.notify_all();
}
"download remote container checkpoint over SSH" => {
let (state, timeout) = changed
.wait_timeout_while(
mutex.lock().unwrap(),
Duration::from_secs(5),
|state| state.copies < 2 || (!self.first && !state.first_cleaned),
)
.unwrap();
ensure!(!timeout.timed_out(), "interleaved transfer stalled");
let source = command.args[command.args.len() - 2]
.split_once(':')
.unwrap()
.1;
let bytes = state
.files
.get(source)
.context("other transfer removed staging")?;
fs::write(remote, bytes)?;
}
"remove remote checkpoint staging" => {
let mut state = mutex.lock().unwrap();
state.files.remove(&remote_path());
if self.first {
state.first_cleaned = true;
changed.notify_all();
}
}
"create remote checkpoint staging directory" => {}
purpose => bail!("unexpected transfer command: {purpose}"),
}
Ok(CommandOutput {
status: 0,
stdout: Vec::new(),
stderr: Vec::new(),
})
}
}
for locator in [locators().pop().unwrap(), ssh_docker_locator()] {
let directory = tempfile::tempdir().unwrap();
let staging = (Mutex::new(Staging::default()), Condvar::new());
let first = vec![b'a'; 192 * 1024];
let second = vec![b'b'; 256 * 1024];
std::thread::scope(|scope| {
let run = |operation, bytes: &[u8], is_first| {
let source = directory.path().join(format!("{operation}-source.zip"));
fs::write(&source, bytes).unwrap();
let destination = directory.path().join(format!("{operation}-verified.zip"));
let digest = checkpoint_sha256(&source).unwrap();
let gate = CheckpointTransfer {
locator: &locator,
session_id: SESSION,
operation_id: operation,
remote_archive: &format!("/workers/{operation}.zip"),
destination: &destination,
expected_sha256: &digest,
expected_event_frontier: 1,
expected_event_frontier_digest: &"a".repeat(64),
}
.execute(&InterleavedExecutor {
staging: &staging,
archive: bytes,
first: is_first,
})
.unwrap();
assert_eq!(fs::read(gate.archive_path()).unwrap(), bytes);
assert_eq!(gate.sha256(), digest);
};
let first_run = scope.spawn(move || run("first", &first, true));
let second_run = scope.spawn(move || run("second", &second, false));
first_run.join().unwrap();
second_run.join().unwrap();
});
assert!(staging.0.lock().unwrap().files.is_empty());
}
}
#[test]
fn corrupt_or_truncated_transfer_preserves_previous_checkpoint_and_reports_evidence() {
let temp = tempfile::tempdir().unwrap();
let source = temp.path().join("source.zip");
let original = vec![b'a'; 192 * 1024];
fs::write(&source, &original).unwrap();
let expected_sha256 = checkpoint_sha256(&source).unwrap();
let mut corrupt = original.clone();
corrupt[80 * 1024] = b'b';
for bytes in [corrupt, original[..1024].to_vec()] {
let downloaded = temp.path().join("download.zip");
fs::write(&downloaded, &bytes).unwrap();
let actual_sha256 = checkpoint_sha256(&downloaded).unwrap();
let destination = temp.path().join("previous.zip");
fs::write(&destination, b"previous verified checkpoint").unwrap();
let error = CheckpointTransfer {
locator: &ssh_docker_locator(),
session_id: SESSION,
operation_id: "failed-export",
remote_archive: "/workers/failed-export.zip",
destination: &destination,
expected_sha256: &expected_sha256,
expected_event_frontier: 1,
expected_event_frontier_digest: &"a".repeat(64),
}
.execute(&SshDockerTransferExecutor::new(bytes.clone()))
.unwrap_err();
let detail = format!("{error:#}");
for expected in [
"complete checkpoint archive",
SESSION,
"failed-export",
&expected_sha256,
&actual_sha256,
&format!("downloaded_bytes={}", bytes.len()),
"target archive retained at /workers/failed-export.zip",
"retry a fresh export",
] {
assert!(
detail.contains(expected),
"missing {expected:?} from {detail}"
);
}
assert_eq!(
fs::read(&destination).unwrap(),
b"previous verified checkpoint"
);
assert_eq!(fs::read(&source).unwrap(), original);
}
}