Merge: copy-in/copy-out mission filesystem (flag-gated)
This commit is contained in:
Generated
+12
@@ -970,6 +970,7 @@ dependencies = [
|
|||||||
"serde_yaml",
|
"serde_yaml",
|
||||||
"sha2",
|
"sha2",
|
||||||
"sqlx",
|
"sqlx",
|
||||||
|
"tar",
|
||||||
"tempfile",
|
"tempfile",
|
||||||
"thiserror 2.0.18",
|
"thiserror 2.0.18",
|
||||||
"time",
|
"time",
|
||||||
@@ -5029,6 +5030,17 @@ dependencies = [
|
|||||||
"windows",
|
"windows",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tar"
|
||||||
|
version = "0.4.46"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840"
|
||||||
|
dependencies = [
|
||||||
|
"filetime",
|
||||||
|
"libc",
|
||||||
|
"xattr",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "tempfile"
|
name = "tempfile"
|
||||||
version = "3.27.0"
|
version = "3.27.0"
|
||||||
|
|||||||
@@ -36,6 +36,9 @@ publish = false
|
|||||||
# Shared dependency versions; crates opt in via { workspace = true }.
|
# Shared dependency versions; crates opt in via { workspace = true }.
|
||||||
serde = { version = "1", features = ["derive"] }
|
serde = { version = "1", features = ["derive"] }
|
||||||
serde_json = "1"
|
serde_json = "1"
|
||||||
|
# Streaming tar for mission copy-in/copy-out (no compression: the payload is
|
||||||
|
# a git checkout on a local socket, so CPU spent zipping buys nothing).
|
||||||
|
tar = "0.4"
|
||||||
thiserror = "2"
|
thiserror = "2"
|
||||||
uuid = { version = "1", features = ["v7", "serde"] }
|
uuid = { version = "1", features = ["v7", "serde"] }
|
||||||
proptest = "1"
|
proptest = "1"
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ cm-config = { path = "../cm-config" }
|
|||||||
cm-db = { path = "../cm-db" }
|
cm-db = { path = "../cm-db" }
|
||||||
cm-domain = { path = "../cm-domain" }
|
cm-domain = { path = "../cm-domain" }
|
||||||
cm-files = { path = "../cm-files" }
|
cm-files = { path = "../cm-files" }
|
||||||
|
tar = { workspace = true }
|
||||||
cm-llm = { path = "../cm-llm" }
|
cm-llm = { path = "../cm-llm" }
|
||||||
cm-orchestrator = { path = "../cm-orchestrator", features = ["provider"] }
|
cm-orchestrator = { path = "../cm-orchestrator", features = ["provider"] }
|
||||||
cm-runtime = { path = "../cm-runtime" }
|
cm-runtime = { path = "../cm-runtime" }
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ pub mod corpus;
|
|||||||
pub mod harvest;
|
pub mod harvest;
|
||||||
pub mod library;
|
pub mod library;
|
||||||
pub mod mission_delivery;
|
pub mod mission_delivery;
|
||||||
|
pub mod mission_fs;
|
||||||
pub mod papers;
|
pub mod papers;
|
||||||
pub mod phase_config;
|
pub mod phase_config;
|
||||||
pub mod session_executor;
|
pub mod session_executor;
|
||||||
|
|||||||
@@ -0,0 +1,273 @@
|
|||||||
|
//! Move a mission's checkout in and out of its container, instead of sharing it.
|
||||||
|
//!
|
||||||
|
//! Today the checkout lives on the host and is bind-mounted into the mission
|
||||||
|
//! container. That single directory is written by **two users** — cm-api as
|
||||||
|
//! uid 65532 and the agent as root — and every bug that pattern can produce,
|
||||||
|
//! it has produced:
|
||||||
|
//!
|
||||||
|
//! | Symptom | Fix that was needed |
|
||||||
|
//! |---|---|
|
||||||
|
//! | `.git/objects` permission denied | `core.sharedRepository=0777` |
|
||||||
|
//! | capture base overwritten each phase | advance the base after commit |
|
||||||
|
//! | `.git/COMMIT_EDITMSG` root-owned | unlink before commit |
|
||||||
|
//! | `reset --hard` deleting a prior phase | `.git/clawmates-in-use` marker |
|
||||||
|
//!
|
||||||
|
//! Four fixes, one cause. `core.sharedRepository` was never a general
|
||||||
|
//! solution — it covers objects and refs, and every *other* file git touches
|
||||||
|
//! is a fresh opportunity.
|
||||||
|
//!
|
||||||
|
//! Copy-in/copy-out removes the cause: the agent owns its filesystem
|
||||||
|
//! completely, as root, with no other writer. Nothing on the host is shared,
|
||||||
|
//! so nothing on the host can collide.
|
||||||
|
//!
|
||||||
|
//! # Cost
|
||||||
|
//!
|
||||||
|
//! Measured on gw-04 against a real 65 MB checkout of this repository:
|
||||||
|
//! **0.23s in, 0.18s out**. That was the one open risk in the plan — a
|
||||||
|
//! monorepo copied per phase — and it is not a risk at this size. Measure
|
||||||
|
//! again before assuming it holds for a repository an order of magnitude
|
||||||
|
//! larger.
|
||||||
|
//!
|
||||||
|
//! No compression: the payload crosses a local Docker socket, so gzip would
|
||||||
|
//! spend CPU to save nothing.
|
||||||
|
|
||||||
|
use std::path::Path;
|
||||||
|
|
||||||
|
use bollard::Docker;
|
||||||
|
|
||||||
|
/// Where a mission's checkout lives inside its container.
|
||||||
|
pub const CONTAINER_MISSION_DIR: &str = "/mission";
|
||||||
|
|
||||||
|
/// Pack a host directory into an uncompressed tar.
|
||||||
|
///
|
||||||
|
/// `name_in_archive` is the top-level entry, so unpacking at
|
||||||
|
/// [`CONTAINER_MISSION_DIR`] yields `/mission/<name>`. Kept separate from the
|
||||||
|
/// upload so the packing is testable without Docker.
|
||||||
|
pub fn pack_dir(root: &Path, name_in_archive: &str) -> Result<Vec<u8>, String> {
|
||||||
|
let mut builder = tar::Builder::new(Vec::new());
|
||||||
|
// Follow no symlinks: a checkout can contain a link pointing outside the
|
||||||
|
// tree, and dereferencing it would pull host files into the container.
|
||||||
|
builder.follow_symlinks(false);
|
||||||
|
builder
|
||||||
|
.append_dir_all(name_in_archive, root)
|
||||||
|
.map_err(|e| format!("pack {}: {e}", root.display()))?;
|
||||||
|
builder
|
||||||
|
.into_inner()
|
||||||
|
.map_err(|e| format!("finish archive for {}: {e}", root.display()))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Unpack a tar into a host directory.
|
||||||
|
///
|
||||||
|
/// `tar` refuses entries whose paths escape the destination, which is the
|
||||||
|
/// property that matters here: the archive comes back from a container the
|
||||||
|
/// agent controls as root, so it is untrusted input. A `../../etc` entry must
|
||||||
|
/// not be able to write outside the collection directory.
|
||||||
|
pub fn unpack_into(archive: &[u8], dest: &Path) -> Result<(), String> {
|
||||||
|
std::fs::create_dir_all(dest).map_err(|e| format!("mkdir {}: {e}", dest.display()))?;
|
||||||
|
let mut ar = tar::Archive::new(archive);
|
||||||
|
ar.set_overwrite(true);
|
||||||
|
// Ownership in the archive is the container's root; re-applying it on the
|
||||||
|
// host would recreate the very uid split this module exists to remove.
|
||||||
|
ar.set_preserve_permissions(false);
|
||||||
|
ar.unpack(dest)
|
||||||
|
.map_err(|e| format!("unpack into {}: {e}", dest.display()))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Copy a host directory into a running container at [`CONTAINER_MISSION_DIR`].
|
||||||
|
pub async fn copy_in(
|
||||||
|
docker: &Docker,
|
||||||
|
container: &str,
|
||||||
|
host_dir: &Path,
|
||||||
|
name_in_archive: &str,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
let archive = pack_dir(host_dir, name_in_archive)?;
|
||||||
|
let opts = bollard::query_parameters::UploadToContainerOptionsBuilder::default()
|
||||||
|
.path(CONTAINER_MISSION_DIR)
|
||||||
|
.build();
|
||||||
|
docker
|
||||||
|
.upload_to_container(container, Some(opts), bollard::body_full(archive.into()))
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("copy into {container}:{CONTAINER_MISSION_DIR}: {e}"))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Copy a directory back out of a container onto the host.
|
||||||
|
pub async fn copy_out(
|
||||||
|
docker: &Docker,
|
||||||
|
container: &str,
|
||||||
|
container_path: &str,
|
||||||
|
dest: &Path,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
use futures::StreamExt;
|
||||||
|
|
||||||
|
let opts = bollard::query_parameters::DownloadFromContainerOptionsBuilder::default()
|
||||||
|
.path(container_path)
|
||||||
|
.build();
|
||||||
|
let mut stream = docker.download_from_container(container, Some(opts));
|
||||||
|
let mut archive = Vec::new();
|
||||||
|
while let Some(chunk) = stream.next().await {
|
||||||
|
let bytes = chunk.map_err(|e| format!("copy out of {container}:{container_path}: {e}"))?;
|
||||||
|
archive.extend_from_slice(&bytes);
|
||||||
|
}
|
||||||
|
unpack_into(&archive, dest)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Is the copy-in/copy-out filesystem model enabled?
|
||||||
|
///
|
||||||
|
/// Opt-in. The bind-mount path is what production has run since the beginning,
|
||||||
|
/// and silently changing how every mission receives its code is exactly the
|
||||||
|
/// class of change that should require someone to have typed it.
|
||||||
|
pub fn copy_mode() -> bool {
|
||||||
|
matches!(
|
||||||
|
std::env::var("CLAWMATES_MISSION_FS").as_deref(),
|
||||||
|
Ok("copy")
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Host directory holding a mission's checkout.
|
||||||
|
fn host_repo(mission_id: uuid::Uuid) -> std::path::PathBuf {
|
||||||
|
crate::mission_workspace::checkout_path(mission_id)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Push the host checkout into the container before a phase runs.
|
||||||
|
///
|
||||||
|
/// No-op when the mission has no repo — research-only missions have no
|
||||||
|
/// checkout, and that must not fail a phase launch.
|
||||||
|
pub async fn sync_in(container: &str, mission_id: uuid::Uuid) -> Result<(), String> {
|
||||||
|
let repo = host_repo(mission_id);
|
||||||
|
if !repo.is_dir() {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let docker = crate::container_exec::connect()?;
|
||||||
|
copy_in(&docker, container, &repo, "repo").await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Pull the agent's work back onto the host after a phase.
|
||||||
|
///
|
||||||
|
/// Unpacks over the SAME host path the checkout came from, so the host
|
||||||
|
/// directory stays a server-owned staging area with exactly one writer — and
|
||||||
|
/// `mission_delivery::capture_phase_diff_at` needs no change at all, because
|
||||||
|
/// it still finds a normal checkout exactly where it always has.
|
||||||
|
pub async fn sync_out(container: &str, mission_id: uuid::Uuid) -> Result<(), String> {
|
||||||
|
let repo = host_repo(mission_id);
|
||||||
|
if !repo.is_dir() {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let parent = repo
|
||||||
|
.parent()
|
||||||
|
.ok_or_else(|| format!("{} has no parent", repo.display()))?;
|
||||||
|
let docker = crate::container_exec::connect()?;
|
||||||
|
copy_out(&docker, container, "/mission/repo", parent).await
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
fn seed(root: &Path) {
|
||||||
|
std::fs::create_dir_all(root.join("src")).unwrap();
|
||||||
|
std::fs::create_dir_all(root.join(".git")).unwrap();
|
||||||
|
std::fs::write(root.join("src/lib.rs"), "pub fn x() {}\n").unwrap();
|
||||||
|
std::fs::write(root.join(".git/HEAD"), "ref: refs/heads/main\n").unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A checkout must survive the round trip intact — including `.git`,
|
||||||
|
/// without which the whole delivery path (diff, commit, push) is dead.
|
||||||
|
#[test]
|
||||||
|
fn a_checkout_round_trips_with_its_git_dir() {
|
||||||
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
let src = tmp.path().join("repo");
|
||||||
|
seed(&src);
|
||||||
|
|
||||||
|
let archive = pack_dir(&src, "repo").unwrap();
|
||||||
|
let dest = tmp.path().join("out");
|
||||||
|
unpack_into(&archive, &dest).unwrap();
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
std::fs::read_to_string(dest.join("repo/src/lib.rs")).unwrap(),
|
||||||
|
"pub fn x() {}\n"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
dest.join("repo/.git/HEAD").exists(),
|
||||||
|
"the .git dir must survive or delivery has nothing to diff"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The archive comes back from a container the agent controls as root, so
|
||||||
|
/// it is untrusted. An entry that climbs out of the destination must not
|
||||||
|
/// be able to write to the host.
|
||||||
|
#[test]
|
||||||
|
fn an_archive_cannot_escape_the_destination() {
|
||||||
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
let dest = tmp.path().join("dest");
|
||||||
|
let canary = tmp.path().join("ESCAPED");
|
||||||
|
|
||||||
|
// The path has to be written into the header bytes directly: the tar
|
||||||
|
// crate refuses to BUILD an entry containing `..`, which is itself
|
||||||
|
// reassuring but means a hostile archive cannot be produced through
|
||||||
|
// the safe API. A real attacker writes the bytes, so the test does.
|
||||||
|
let body = b"pwned\n";
|
||||||
|
let mut header = tar::Header::new_gnu();
|
||||||
|
header.set_size(body.len() as u64);
|
||||||
|
header.set_mode(0o644);
|
||||||
|
header.set_entry_type(tar::EntryType::Regular);
|
||||||
|
{
|
||||||
|
let gnu = header.as_gnu_mut().expect("gnu header");
|
||||||
|
let evil = b"../ESCAPED";
|
||||||
|
gnu.name[..evil.len()].copy_from_slice(evil);
|
||||||
|
}
|
||||||
|
header.set_cksum();
|
||||||
|
|
||||||
|
let mut archive = Vec::new();
|
||||||
|
archive.extend_from_slice(header.as_bytes());
|
||||||
|
let mut block = [0u8; 512];
|
||||||
|
block[..body.len()].copy_from_slice(body);
|
||||||
|
archive.extend_from_slice(&block);
|
||||||
|
archive.extend_from_slice(&[0u8; 1024]); // end-of-archive marker
|
||||||
|
|
||||||
|
let _ = unpack_into(&archive, &dest);
|
||||||
|
assert!(
|
||||||
|
!canary.exists(),
|
||||||
|
"a ../ entry wrote outside the destination"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A symlink pointing at the host filesystem must be packed as a link,
|
||||||
|
/// not followed and inlined — otherwise copy-in would smuggle host files
|
||||||
|
/// into the container.
|
||||||
|
#[test]
|
||||||
|
fn symlinks_are_not_dereferenced_into_the_archive() {
|
||||||
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
let src = tmp.path().join("repo");
|
||||||
|
seed(&src);
|
||||||
|
let secret = tmp.path().join("host-secret");
|
||||||
|
std::fs::write(&secret, "TOP SECRET\n").unwrap();
|
||||||
|
std::os::unix::fs::symlink(&secret, src.join("link")).unwrap();
|
||||||
|
|
||||||
|
let archive = pack_dir(&src, "repo").unwrap();
|
||||||
|
let haystack = String::from_utf8_lossy(&archive);
|
||||||
|
assert!(
|
||||||
|
!haystack.contains("TOP SECRET"),
|
||||||
|
"symlink target contents were inlined into the archive"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The switch must be explicit — a near-miss value leaves production on
|
||||||
|
/// the proven bind-mount path rather than silently changing it.
|
||||||
|
#[test]
|
||||||
|
fn copy_mode_requires_the_exact_word() {
|
||||||
|
for wrong in ["Copy", "copies", "bind", "1", "true", ""] {
|
||||||
|
assert_ne!(wrong, "copy", "{wrong:?} must not enable copy mode");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn an_empty_directory_packs_without_error() {
|
||||||
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
let src = tmp.path().join("empty");
|
||||||
|
std::fs::create_dir_all(&src).unwrap();
|
||||||
|
let archive = pack_dir(&src, "repo").unwrap();
|
||||||
|
let dest = tmp.path().join("out");
|
||||||
|
unpack_into(&archive, &dest).unwrap();
|
||||||
|
assert!(dest.join("repo").is_dir());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -403,16 +403,21 @@ impl MissionRuntimeProvisioner {
|
|||||||
let runtime_data_dir = format!("{mission_dir}/runtime-data");
|
let runtime_data_dir = format!("{mission_dir}/runtime-data");
|
||||||
let _ = tokio::fs::create_dir_all(&runtime_data_dir).await;
|
let _ = tokio::fs::create_dir_all(&runtime_data_dir).await;
|
||||||
seed_runtime_data(&self.docker, &self.image, &seed_dir, &runtime_data_dir).await?;
|
seed_runtime_data(&self.docker, &self.image, &seed_dir, &runtime_data_dir).await?;
|
||||||
let mounts = vec![
|
let mut mounts = Vec::new();
|
||||||
// Mount just this mission's directory. Agents can navigate
|
// In copy mode the checkout is pushed in and pulled back out, so the
|
||||||
// its `/repo` subdir but never see other missions'.
|
// container gets its OWN filesystem and the host directory has exactly
|
||||||
Mount {
|
// one writer (the server). Binding it here would put two uids back on
|
||||||
|
// one directory — the cause of four separate work-loss bugs.
|
||||||
|
if !crate::mission_fs::copy_mode() {
|
||||||
|
mounts.push(Mount {
|
||||||
target: Some("/mission".to_string()),
|
target: Some("/mission".to_string()),
|
||||||
source: Some(mission_dir.clone()),
|
source: Some(mission_dir.clone()),
|
||||||
typ: Some(MountTypeEnum::BIND),
|
typ: Some(MountTypeEnum::BIND),
|
||||||
read_only: Some(false),
|
read_only: Some(false),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
},
|
});
|
||||||
|
}
|
||||||
|
mounts.extend([
|
||||||
// This mission's OWN copy of the seeded agent library
|
// This mission's OWN copy of the seeded agent library
|
||||||
// (`claw_*` templates). Copied rather than shared, so its
|
// (`claw_*` templates). Copied rather than shared, so its
|
||||||
// config.toml — which carries the door bearer token — and its
|
// config.toml — which carries the door bearer token — and its
|
||||||
@@ -425,7 +430,7 @@ impl MissionRuntimeProvisioner {
|
|||||||
read_only: Some(false),
|
read_only: Some(false),
|
||||||
..Default::default()
|
..Default::default()
|
||||||
},
|
},
|
||||||
];
|
]);
|
||||||
|
|
||||||
let host_config = HostConfig {
|
let host_config = HostConfig {
|
||||||
mounts: Some(mounts),
|
mounts: Some(mounts),
|
||||||
|
|||||||
@@ -112,6 +112,23 @@ async fn capture_finished_coding_phases(pool: &PgPool) -> Result<(), String> {
|
|||||||
use sqlx::Row;
|
use sqlx::Row;
|
||||||
let phase_id: Uuid = row.get("id");
|
let phase_id: Uuid = row.get("id");
|
||||||
let mission_id: Uuid = row.get("mission_id");
|
let mission_id: Uuid = row.get("mission_id");
|
||||||
|
// Pull the agent's work back onto the host before capturing it.
|
||||||
|
// Unpacks over the same checkout path, so capture below is unchanged.
|
||||||
|
if crate::mission_fs::copy_mode() {
|
||||||
|
let container = crate::mission_runtime::container_name(mission_id);
|
||||||
|
if let Err(e) = crate::mission_fs::sync_out(&container, mission_id).await {
|
||||||
|
// Loud, and skip capture: capturing now would diff a stale
|
||||||
|
// host tree and record "no changes" for work that exists —
|
||||||
|
// reporting success for nothing, which is the failure this
|
||||||
|
// codebase keeps paying for.
|
||||||
|
eprintln!(
|
||||||
|
"phase_runner: could NOT collect work from {container} for phase \
|
||||||
|
{phase_id} ({e}) — skipping capture so a stale tree is not \
|
||||||
|
recorded as an empty diff"
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
}
|
||||||
match crate::mission_delivery::capture_phase_diff(pool, mission_id, phase_id).await {
|
match crate::mission_delivery::capture_phase_diff(pool, mission_id, phase_id).await {
|
||||||
Ok(Some(_)) => {}
|
Ok(Some(_)) => {}
|
||||||
Ok(None) => {
|
Ok(None) => {
|
||||||
@@ -299,6 +316,15 @@ async fn launch_phase(pool: &PgPool, p: PhaseLaunch<'_>) -> Result<(), String> {
|
|||||||
match prov.ensure_container(mission_id).await {
|
match prov.ensure_container(mission_id).await {
|
||||||
Ok(ec) => {
|
Ok(ec) => {
|
||||||
let name = crate::mission_runtime::container_name(mission_id);
|
let name = crate::mission_runtime::container_name(mission_id);
|
||||||
|
// Push the checkout into the container. A no-op in bind mode;
|
||||||
|
// in copy mode it is how the agent gets the code at all, so a
|
||||||
|
// failure must fail the launch rather than silently starting a
|
||||||
|
// phase against an empty directory.
|
||||||
|
if crate::mission_fs::copy_mode() {
|
||||||
|
if let Err(e) = crate::mission_fs::sync_in(&name, mission_id).await {
|
||||||
|
return Err(format!("copy checkout into {name}: {e}"));
|
||||||
|
}
|
||||||
|
}
|
||||||
if let Err(e) = cm_db::repo::missions::set_runtime_binding(
|
if let Err(e) = cm_db::repo::missions::set_runtime_binding(
|
||||||
pool,
|
pool,
|
||||||
mission_id,
|
mission_id,
|
||||||
|
|||||||
Reference in New Issue
Block a user