Merge: B4.5 microvm executor — runtime_kind='microvm' now has a reader
This commit is contained in:
@@ -329,7 +329,7 @@ async fn run() -> Result<(), String> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
cm_api::phase_runner::spawn(pool.clone(), runtime.clone());
|
cm_api::phase_runner::spawn(pool.clone(), runtime.clone(), node_hub.clone());
|
||||||
// Per-mission runtime container sweeper (C3): tears down mission
|
// Per-mission runtime container sweeper (C3): tears down mission
|
||||||
// runtime containers 30 min after the mission reaches a terminal
|
// runtime containers 30 min after the mission reaches a terminal
|
||||||
// state so operators have a window to pull final artifacts.
|
// state so operators have a window to pull final artifacts.
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ pub mod harvest;
|
|||||||
pub mod library;
|
pub mod library;
|
||||||
pub mod mission_delivery;
|
pub mod mission_delivery;
|
||||||
pub mod microvm_client;
|
pub mod microvm_client;
|
||||||
|
pub mod microvm_executor;
|
||||||
pub mod mission_fs;
|
pub mod mission_fs;
|
||||||
pub mod papers;
|
pub mod papers;
|
||||||
pub mod phase_config;
|
pub mod phase_config;
|
||||||
|
|||||||
@@ -0,0 +1,295 @@
|
|||||||
|
//! Run a mission phase inside a Firecracker microVM.
|
||||||
|
//!
|
||||||
|
//! The step that makes the whole microVM track load-bearing. Before this,
|
||||||
|
//! `runtime_kind = 'microvm'` placed a mission on a KVM-capable node and then
|
||||||
|
//! nothing executed it — config accepted without a reader, which is one of the
|
||||||
|
//! four seams this project keeps closing.
|
||||||
|
//!
|
||||||
|
//! # The model
|
||||||
|
//!
|
||||||
|
//! **inject → run → collect → destroy**, the same shape copy mode already proved
|
||||||
|
//! for containers, with a VM boundary instead of a namespace boundary. Nothing is
|
||||||
|
//! shared: the checkout goes in as a tar, the work comes back as a tar, and the
|
||||||
|
//! guest's filesystem dies with it.
|
||||||
|
//!
|
||||||
|
//! # Why the agent does not push
|
||||||
|
//!
|
||||||
|
//! `session_executor` tells its agent to commit and push to a branch, and the
|
||||||
|
//! forge is on the VM's egress allow-list, so it could. It must not:
|
||||||
|
//!
|
||||||
|
//! - `mission_delivery::capture_phase_diff_at` is already host-side and diffs the
|
||||||
|
//! collected tree against the recorded clone point, covering committed, staged
|
||||||
|
//! and unstaged work in one pass. Pushing from the guest would add a second,
|
||||||
|
//! untested way for work to arrive.
|
||||||
|
//! - Pushing needs forge credentials in the VM. The point of collecting is that
|
||||||
|
//! the guest never holds them.
|
||||||
|
//!
|
||||||
|
//! So the prompt says explicitly not to push, and the work is collected over the
|
||||||
|
//! same host path the checkout came from — leaving the host directory a
|
||||||
|
//! server-owned staging area with exactly one writer.
|
||||||
|
//!
|
||||||
|
//! # One run row
|
||||||
|
//!
|
||||||
|
//! Like `launch_direct_session`, this creates exactly ONE `topology_runs` row
|
||||||
|
//! (tier `microvm`). The whole downstream lifecycle — `close_finished_phases`,
|
||||||
|
//! evaluation, capture, delivery — keys off those rows, and a second completion
|
||||||
|
//! path would mean two ways for a phase to finish with one of them untested.
|
||||||
|
|
||||||
|
use cm_domain::NodeId;
|
||||||
|
use uuid::Uuid;
|
||||||
|
|
||||||
|
use crate::fleet::NodeHub;
|
||||||
|
use crate::microvm_client::MicroVm;
|
||||||
|
|
||||||
|
/// Resources per phase VM. Generous enough to build: the mission toolchain in
|
||||||
|
/// `agent-claude` includes rustc, and a 512 MB guest OOMs partway through a
|
||||||
|
/// `cargo build` in a way that looks like an agent giving up.
|
||||||
|
const VCPUS: u32 = 4;
|
||||||
|
const MEM_MIB: u32 = 8192;
|
||||||
|
|
||||||
|
/// Budget for one agent turn inside the VM, matching the container path's.
|
||||||
|
const TURN_SECS: u64 = 3600;
|
||||||
|
|
||||||
|
/// Where the checkout lands in the guest. Same path as the container path uses,
|
||||||
|
/// so a prompt or a tool that hardcodes it behaves identically either way.
|
||||||
|
const GUEST_REPO: &str = "/mission/repo";
|
||||||
|
|
||||||
|
/// A vm id must be `[A-Za-z0-9_-]` — the node rejects anything else, since it
|
||||||
|
/// becomes a path component. Hyphenated uuids qualify; this also keeps the id
|
||||||
|
/// readable in `vm_list` output and in the node's egress log.
|
||||||
|
fn vm_id_for(phase_id: Uuid, iteration: i32) -> String {
|
||||||
|
format!("m-{}-{}", &phase_id.simple().to_string()[..12], iteration)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// What the agent is told. Deliberately not `session_executor::session_prompt`:
|
||||||
|
/// that one instructs a push, and on this path pushing is the host's job.
|
||||||
|
fn vm_prompt(task: &str) -> String {
|
||||||
|
format!(
|
||||||
|
"You are working in the git repository at {GUEST_REPO}.\n\
|
||||||
|
\n\
|
||||||
|
TASK\n\
|
||||||
|
{task}\n\
|
||||||
|
\n\
|
||||||
|
WHEN THE WORK IS DONE\n\
|
||||||
|
Leave it in the working tree. Do NOT push, and do not add a remote — \
|
||||||
|
this machine has no access to the forge. Committing locally is fine but \
|
||||||
|
not required: the whole tree is collected when you finish, and the work \
|
||||||
|
is recorded from the repository itself either way.\n\
|
||||||
|
\n\
|
||||||
|
If the task cannot be completed as written — a file it refers to does \
|
||||||
|
not exist, a premise is wrong, the tests cannot run — say so plainly. An \
|
||||||
|
honest report that the work could not be done is worth more than a tree \
|
||||||
|
that looks finished.\n"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Shell-quote for `sh -c`. The guest agent runs one command string, and a task
|
||||||
|
/// description contains apostrophes, quotes and newlines as a matter of course.
|
||||||
|
fn shell_quote(s: &str) -> String {
|
||||||
|
format!("'{}'", s.replace('\'', r"'\''"))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The command that runs the agent in the guest.
|
||||||
|
fn agent_command(prompt: &str) -> String {
|
||||||
|
// --permission-mode acceptEdits, matching the container path: the VM IS the
|
||||||
|
// boundary, so prompting for permission inside it would only mean a turn that
|
||||||
|
// waits for an answer nobody can give.
|
||||||
|
format!(
|
||||||
|
"cd {GUEST_REPO} && claude -p --allowedTools Read Edit Write Bash \
|
||||||
|
--permission-mode acceptEdits {}",
|
||||||
|
shell_quote(prompt)
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Outcome of one phase VM, as observed from outside it.
|
||||||
|
pub struct VmOutcome {
|
||||||
|
/// The agent's closing text. Diagnostic only — never evidence. Whether the
|
||||||
|
/// phase succeeded is decided downstream against the repository.
|
||||||
|
pub summary: String,
|
||||||
|
pub rc: i64,
|
||||||
|
/// Whether the work came back. A turn that ran and could not be collected is
|
||||||
|
/// a failure even if the agent was happy.
|
||||||
|
pub collected: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Boot a VM, run the phase in it, collect the result, and destroy it.
|
||||||
|
///
|
||||||
|
/// `destroy` runs on every exit path. A leaked VM holds an 8 GB sparse rootfs and
|
||||||
|
/// a firecracker process, and the node's orphan sweep is a backstop, not a plan.
|
||||||
|
pub async fn run_phase_in_vm(hub: &NodeHub, p: VmPhase<'_>) -> Result<VmOutcome, String> {
|
||||||
|
// Resolved BEFORE the VM boots: a missing subscription token must fail the
|
||||||
|
// phase, not boot a VM whose agent will sit there unauthenticated.
|
||||||
|
let env = crate::mission_runtime::microvm_provider_env(p.backend)?;
|
||||||
|
|
||||||
|
let vm = MicroVm::new(hub, p.node_id, vm_id_for(p.phase_id, p.iteration));
|
||||||
|
let created = vm.create(VCPUS, MEM_MIB, p.backend).await?;
|
||||||
|
|
||||||
|
// From here on every early return must still destroy the VM, so the work is
|
||||||
|
// one call whose result is held while teardown runs unconditionally.
|
||||||
|
let outcome = run_inside(&vm, &created, p.task, p.repo, &env).await;
|
||||||
|
|
||||||
|
if let Err(e) = vm.destroy().await {
|
||||||
|
// Not fatal to the phase — the work may already be collected — but loud,
|
||||||
|
// because the alternative is an 8 GB rootfs nobody is looking for.
|
||||||
|
eprintln!(
|
||||||
|
"microvm_executor: mission {} phase {}: destroy {} FAILED: {e}",
|
||||||
|
p.mission_id,
|
||||||
|
p.phase_id,
|
||||||
|
vm.vm_id()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
outcome
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One phase to run in one VM.
|
||||||
|
pub struct VmPhase<'a> {
|
||||||
|
pub node_id: NodeId,
|
||||||
|
pub mission_id: Uuid,
|
||||||
|
pub phase_id: Uuid,
|
||||||
|
/// Which pass. Part of the vm id, so two passes of the same phase cannot
|
||||||
|
/// collide on a node.
|
||||||
|
pub iteration: i32,
|
||||||
|
pub task: &'a str,
|
||||||
|
/// `missions.backend` — which rootfs image. `None` boots the node's default.
|
||||||
|
pub backend: Option<&'a str>,
|
||||||
|
/// The host checkout, injected as a tar and collected back over the same
|
||||||
|
/// path so `mission_delivery` needs no change.
|
||||||
|
pub repo: &'a std::path::Path,
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn run_inside(
|
||||||
|
vm: &MicroVm<'_>,
|
||||||
|
created: &serde_json::Value,
|
||||||
|
task: &str,
|
||||||
|
repo: &std::path::Path,
|
||||||
|
env: &[(String, String)],
|
||||||
|
) -> Result<VmOutcome, String> {
|
||||||
|
// An agent CLI cannot reach its API without the tunnel, and a turn without
|
||||||
|
// egress does not fail — it hangs, or reports a network error the operator
|
||||||
|
// then has to trace back through three layers. `create` measures both ends;
|
||||||
|
// this is the reader that makes those fields matter.
|
||||||
|
if created.get("egress").and_then(serde_json::Value::as_bool) != Some(true) {
|
||||||
|
return Err(format!(
|
||||||
|
"vm {} has no egress (host={}, guest={}), so the agent could not reach \
|
||||||
|
its API — refusing to run the phase in it",
|
||||||
|
vm.vm_id(),
|
||||||
|
created.get("egress_host").unwrap_or(&serde_json::Value::Null),
|
||||||
|
created.get("egress_guest").unwrap_or(&serde_json::Value::Null),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
// The checkout, as a tar. `pack_dir` names the entry `repo`, and the guest
|
||||||
|
// unpacks it under /mission, so it lands at /mission/repo.
|
||||||
|
let archive = crate::mission_fs::pack_dir(repo, "repo")?;
|
||||||
|
let injected = archive.len();
|
||||||
|
vm.inject("/mission", &archive).await?;
|
||||||
|
|
||||||
|
// Prove the guest actually has the checkout before spending an agent turn on
|
||||||
|
// it. An inject that reports success while landing nothing would otherwise
|
||||||
|
// become an agent reporting that the repository is empty.
|
||||||
|
let probe = vm
|
||||||
|
.exec(
|
||||||
|
&format!("test -d {GUEST_REPO}/.git && echo REPO-PRESENT"),
|
||||||
|
None,
|
||||||
|
60,
|
||||||
|
&[],
|
||||||
|
)
|
||||||
|
.await?;
|
||||||
|
if !probe.stdout.contains("REPO-PRESENT") {
|
||||||
|
return Err(format!(
|
||||||
|
"the checkout did not land in the guest ({injected} bytes injected, \
|
||||||
|
{GUEST_REPO}/.git is absent) — rc={} {}",
|
||||||
|
probe.rc, probe.stderr
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
let out = vm
|
||||||
|
.exec(&agent_command(&vm_prompt(task)), None, TURN_SECS, env)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
// Collect regardless of the agent's exit code. A turn that failed partway
|
||||||
|
// still wrote files, and throwing them away because the CLI exited non-zero
|
||||||
|
// would discard exactly the work a retry needs to see.
|
||||||
|
let tar = vm.collect(GUEST_REPO).await;
|
||||||
|
let collected = match tar {
|
||||||
|
Ok(bytes) => {
|
||||||
|
let parent = repo
|
||||||
|
.parent()
|
||||||
|
.ok_or_else(|| format!("{} has no parent", repo.display()))?;
|
||||||
|
// Unpacked over the SAME host path the checkout came from, so
|
||||||
|
// `capture_phase_diff_at` finds a normal checkout exactly where it
|
||||||
|
// always has and needs no change at all.
|
||||||
|
crate::mission_fs::unpack_into(&bytes, parent)?;
|
||||||
|
true
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
// Reported, not swallowed: an uncollected turn is a failed phase even
|
||||||
|
// when the agent said it finished.
|
||||||
|
eprintln!("microvm_executor: collect from {} failed: {e}", vm.vm_id());
|
||||||
|
false
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
Ok(VmOutcome {
|
||||||
|
summary: format!("{}{}", out.stdout, out.stderr),
|
||||||
|
rc: out.rc,
|
||||||
|
collected,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
/// The id becomes a path component on the node, which rejects anything
|
||||||
|
/// outside `[A-Za-z0-9_-]` rather than sanitising it.
|
||||||
|
#[test]
|
||||||
|
fn a_vm_id_is_acceptable_to_the_node() {
|
||||||
|
let id = vm_id_for(Uuid::now_v7(), 3);
|
||||||
|
assert!(
|
||||||
|
id.chars()
|
||||||
|
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_'),
|
||||||
|
"{id}"
|
||||||
|
);
|
||||||
|
assert!(id.len() <= 64, "{id}");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Two passes of the same phase must not collide: the second would fail with
|
||||||
|
/// "vm already exists" while the first is still running.
|
||||||
|
#[test]
|
||||||
|
fn each_iteration_gets_its_own_vm() {
|
||||||
|
let p = Uuid::now_v7();
|
||||||
|
assert_ne!(vm_id_for(p, 1), vm_id_for(p, 2));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A task description contains quotes and newlines as a matter of course,
|
||||||
|
/// and the whole command is handed to `sh -c` in the guest.
|
||||||
|
#[test]
|
||||||
|
fn a_quoted_task_cannot_break_out_of_the_command() {
|
||||||
|
let nasty = "it's a 'test'; rm -rf /\n$(whoami) `id` \"quoted\"";
|
||||||
|
let cmd = agent_command(&vm_prompt(nasty));
|
||||||
|
// Everything after the opening quote of the prompt is inside it: the only
|
||||||
|
// way out of a single-quoted string is a quote, and each one is escaped.
|
||||||
|
let body = cmd.split_once('\'').expect("a quoted argument").1;
|
||||||
|
for danger in ["; rm -rf /", "$(whoami)", "`id`"] {
|
||||||
|
let at = body.find(danger).expect("the text is still present");
|
||||||
|
let before = &body[..at];
|
||||||
|
// An odd number of unescaped quotes before it would mean it had
|
||||||
|
// escaped the quoting.
|
||||||
|
let unescaped = before.matches('\'').count() - before.matches(r"'\''").count() * 3;
|
||||||
|
assert_eq!(unescaped % 2, 0, "{danger} is not quoted in: {cmd}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The agent must not be told to push: delivery is host-side, and the guest
|
||||||
|
/// deliberately holds no forge credentials.
|
||||||
|
#[test]
|
||||||
|
fn the_prompt_does_not_ask_the_agent_to_push() {
|
||||||
|
let p = vm_prompt("do the thing");
|
||||||
|
assert!(p.contains("Do NOT push"), "{p}");
|
||||||
|
assert!(
|
||||||
|
!p.contains("push to a new branch"),
|
||||||
|
"the container path's push instruction must not leak in: {p}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -238,7 +238,13 @@ pub async fn on_launch(
|
|||||||
// `completed` with zero files, no commit error and no push error. A launch
|
// `completed` with zero files, no commit error and no push error. A launch
|
||||||
// that cannot bind its agents to the repo has no path to delivering work,
|
// that cannot bind its agents to the repo has no path to delivering work,
|
||||||
// so it must fail at launch where someone is still looking.
|
// so it must fail at launch where someone is still looking.
|
||||||
if !provisioned_claws.is_empty() && mission_gateway.is_some() {
|
//
|
||||||
|
// Not for a microVM mission: its agent is a `claude -p` inside a VM on a
|
||||||
|
// fleet node, not a ZeroClaw claw in a container here, so there is no
|
||||||
|
// workspace to pin. Leaving it would make a microVM launch FAIL on a
|
||||||
|
// container it was never going to use.
|
||||||
|
if !provisioned_claws.is_empty() && mission_gateway.is_some() && mission.runtime_kind != "microvm"
|
||||||
|
{
|
||||||
if let Some(mp) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() {
|
if let Some(mp) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() {
|
||||||
mp.pin_agent_workspaces(mission_id, &provisioned_claws, "/mission/repo")
|
mp.pin_agent_workspaces(mission_id, &provisioned_claws, "/mission/repo")
|
||||||
.await
|
.await
|
||||||
|
|||||||
@@ -34,23 +34,32 @@ use uuid::Uuid;
|
|||||||
const POLL_INTERVAL: Duration = Duration::from_secs(10);
|
const POLL_INTERVAL: Duration = Duration::from_secs(10);
|
||||||
|
|
||||||
/// `runtime` is needed only by the completion evaluator; phases without a
|
/// `runtime` is needed only by the completion evaluator; phases without a
|
||||||
/// `done_when` never touch it.
|
/// `done_when` never touch it. `hub` is needed only by microVM missions, which
|
||||||
pub fn spawn(pool: PgPool, runtime: cm_runtime::Runtime) {
|
/// execute on a fleet node rather than in a container here.
|
||||||
|
pub fn spawn(
|
||||||
|
pool: PgPool,
|
||||||
|
runtime: cm_runtime::Runtime,
|
||||||
|
hub: std::sync::Arc<crate::fleet::NodeHub>,
|
||||||
|
) {
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||||
let mut ticker = tokio::time::interval(POLL_INTERVAL);
|
let mut ticker = tokio::time::interval(POLL_INTERVAL);
|
||||||
ticker.tick().await;
|
ticker.tick().await;
|
||||||
loop {
|
loop {
|
||||||
ticker.tick().await;
|
ticker.tick().await;
|
||||||
if let Err(e) = sweep_once(&pool, &runtime).await {
|
if let Err(e) = sweep_once(&pool, &runtime, &hub).await {
|
||||||
eprintln!("phase_runner: sweep failed: {e}");
|
eprintln!("phase_runner: sweep failed: {e}");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn sweep_once(pool: &PgPool, runtime: &cm_runtime::Runtime) -> Result<(), String> {
|
async fn sweep_once(
|
||||||
start_pending_phases(pool).await?;
|
pool: &PgPool,
|
||||||
|
runtime: &cm_runtime::Runtime,
|
||||||
|
hub: &std::sync::Arc<crate::fleet::NodeHub>,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
start_pending_phases(pool, hub).await?;
|
||||||
close_finished_phases(pool).await?;
|
close_finished_phases(pool).await?;
|
||||||
// Between "all runs finished" and "phase done" sits the completion
|
// Between "all runs finished" and "phase done" sits the completion
|
||||||
// evaluation, for phases that declare a condition.
|
// evaluation, for phases that declare a condition.
|
||||||
@@ -89,7 +98,7 @@ const CAPTURE_BATCH: i64 = 5;
|
|||||||
/// with a `NOT EXISTS` guard covers both and is retryable by construction.
|
/// with a `NOT EXISTS` guard covers both and is retryable by construction.
|
||||||
async fn capture_finished_coding_phases(pool: &PgPool) -> Result<(), String> {
|
async fn capture_finished_coding_phases(pool: &PgPool) -> Result<(), String> {
|
||||||
let rows = sqlx::query(
|
let rows = sqlx::query(
|
||||||
"SELECT mp.id, mp.mission_id, mp.kind, mp.config
|
"SELECT mp.id, mp.mission_id, mp.kind, mp.config, m.runtime_kind
|
||||||
FROM mission_phases mp
|
FROM mission_phases mp
|
||||||
JOIN missions m ON m.id = mp.mission_id
|
JOIN missions m ON m.id = mp.mission_id
|
||||||
WHERE mp.status = 'completed'
|
WHERE mp.status = 'completed'
|
||||||
@@ -114,9 +123,16 @@ async fn capture_finished_coding_phases(pool: &PgPool) -> Result<(), String> {
|
|||||||
let mission_id: Uuid = row.get("mission_id");
|
let mission_id: Uuid = row.get("mission_id");
|
||||||
let kind: String = row.get("kind");
|
let kind: String = row.get("kind");
|
||||||
let config: serde_json::Value = row.get("config");
|
let config: serde_json::Value = row.get("config");
|
||||||
|
let runtime_kind: String = row.get("runtime_kind");
|
||||||
// Pull the agent's work back onto the host before capturing it.
|
// Pull the agent's work back onto the host before capturing it.
|
||||||
// Unpacks over the same checkout path, so capture below is unchanged.
|
// Unpacks over the same checkout path, so capture below is unchanged.
|
||||||
if crate::mission_fs::copy_mode() {
|
//
|
||||||
|
// NOT for a microVM mission: it has no container, so this would fail to
|
||||||
|
// connect and `continue` — skipping capture forever and delivering
|
||||||
|
// nothing, while the phase sat there marked completed. Its work was
|
||||||
|
// already collected out of the VM by `microvm_executor`, over the same
|
||||||
|
// host path, before the VM was destroyed.
|
||||||
|
if crate::mission_fs::copy_mode() && runtime_kind != "microvm" {
|
||||||
let container = crate::mission_runtime::container_name(mission_id);
|
let container = crate::mission_runtime::container_name(mission_id);
|
||||||
if let Err(e) = crate::mission_fs::sync_out(&container, mission_id).await {
|
if let Err(e) = crate::mission_fs::sync_out(&container, mission_id).await {
|
||||||
// Loud, and skip capture: capturing now would diff a stale
|
// Loud, and skip capture: capturing now would diff a stale
|
||||||
@@ -217,14 +233,21 @@ fn empty_delivery_is_a_failure(
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Enqueue topology_runs for every phase whose predecessors are done.
|
/// Enqueue topology_runs for every phase whose predecessors are done.
|
||||||
async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
|
async fn start_pending_phases(
|
||||||
|
pool: &PgPool,
|
||||||
|
hub: &std::sync::Arc<crate::fleet::NodeHub>,
|
||||||
|
) -> Result<(), String> {
|
||||||
// Eligible = pending phase, mission running, all lower-order phases
|
// Eligible = pending phase, mission running, all lower-order phases
|
||||||
// in this mission are 'completed'. `NOT EXISTS ... status <> completed`
|
// in this mission are 'completed'. `NOT EXISTS ... status <> completed`
|
||||||
// handles order 0 (no prior rows) + skipped phases naturally.
|
// handles order 0 (no prior rows) + skipped phases naturally.
|
||||||
let rows = sqlx::query(
|
let rows = sqlx::query(
|
||||||
"SELECT mp.id, mp.mission_id, mp.kind, mp.order_idx, mp.iteration,
|
"SELECT mp.id, mp.mission_id, mp.kind, mp.order_idx, mp.iteration,
|
||||||
mp.config->>'task' AS phase_task,
|
mp.config->>'task' AS phase_task,
|
||||||
m.workspace_id, m.title, m.description
|
m.workspace_id, m.title, m.description,
|
||||||
|
-- Where and how this mission executes. `runtime_kind` decides
|
||||||
|
-- which executor takes the phase; without it 'microvm' is a
|
||||||
|
-- value the placement code honours and nothing reads.
|
||||||
|
m.runtime_kind, m.backend, m.target_node_id
|
||||||
FROM mission_phases mp
|
FROM mission_phases mp
|
||||||
JOIN missions m ON m.id = mp.mission_id
|
JOIN missions m ON m.id = mp.mission_id
|
||||||
WHERE mp.status = 'pending'
|
WHERE mp.status = 'pending'
|
||||||
@@ -250,9 +273,13 @@ async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
|
|||||||
let description: Option<String> = row.get("description");
|
let description: Option<String> = row.get("description");
|
||||||
let phase_task: Option<String> = row.get("phase_task");
|
let phase_task: Option<String> = row.get("phase_task");
|
||||||
let iteration: i32 = row.get("iteration");
|
let iteration: i32 = row.get("iteration");
|
||||||
|
let runtime_kind: String = row.get("runtime_kind");
|
||||||
|
let backend: Option<String> = row.get("backend");
|
||||||
|
let target_node_id: Option<Uuid> = row.get("target_node_id");
|
||||||
|
|
||||||
if let Err(e) = launch_phase(
|
if let Err(e) = launch_phase(
|
||||||
pool,
|
pool,
|
||||||
|
hub,
|
||||||
PhaseLaunch {
|
PhaseLaunch {
|
||||||
phase_id,
|
phase_id,
|
||||||
mission_id,
|
mission_id,
|
||||||
@@ -262,6 +289,9 @@ async fn start_pending_phases(pool: &PgPool) -> Result<(), String> {
|
|||||||
description: description.as_deref(),
|
description: description.as_deref(),
|
||||||
phase_task: phase_task.as_deref(),
|
phase_task: phase_task.as_deref(),
|
||||||
iteration,
|
iteration,
|
||||||
|
runtime_kind: &runtime_kind,
|
||||||
|
backend: backend.as_deref(),
|
||||||
|
target_node_id,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
@@ -292,9 +322,20 @@ struct PhaseLaunch<'a> {
|
|||||||
/// Which pass this is, 0-based. Stamped onto the runs so the completion
|
/// Which pass this is, 0-based. Stamped onto the runs so the completion
|
||||||
/// check can tell this pass's work from the previous one's.
|
/// check can tell this pass's work from the previous one's.
|
||||||
iteration: i32,
|
iteration: i32,
|
||||||
|
/// `missions.runtime_kind`. Selects the executor.
|
||||||
|
runtime_kind: &'a str,
|
||||||
|
/// `missions.backend` — which per-CLI image, on the microVM path.
|
||||||
|
backend: Option<&'a str>,
|
||||||
|
/// Set by `mission_orchestrator` at launch. On the microVM path it is where
|
||||||
|
/// the VM boots, and it is not optional there.
|
||||||
|
target_node_id: Option<Uuid>,
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn launch_phase(pool: &PgPool, p: PhaseLaunch<'_>) -> Result<(), String> {
|
async fn launch_phase(
|
||||||
|
pool: &PgPool,
|
||||||
|
hub: &std::sync::Arc<crate::fleet::NodeHub>,
|
||||||
|
p: PhaseLaunch<'_>,
|
||||||
|
) -> Result<(), String> {
|
||||||
let PhaseLaunch {
|
let PhaseLaunch {
|
||||||
phase_id,
|
phase_id,
|
||||||
mission_id,
|
mission_id,
|
||||||
@@ -304,6 +345,11 @@ async fn launch_phase(pool: &PgPool, p: PhaseLaunch<'_>) -> Result<(), String> {
|
|||||||
description,
|
description,
|
||||||
phase_task,
|
phase_task,
|
||||||
iteration,
|
iteration,
|
||||||
|
// Destructured but read through `p` below, so the compiler keeps this
|
||||||
|
// pattern honest if a field is added.
|
||||||
|
runtime_kind: _,
|
||||||
|
backend: _,
|
||||||
|
target_node_id: _,
|
||||||
} = p;
|
} = p;
|
||||||
// Which team purposes should execute this phase.
|
// Which team purposes should execute this phase.
|
||||||
let purposes: &[&str] = match kind {
|
let purposes: &[&str] = match kind {
|
||||||
@@ -371,7 +417,15 @@ async fn launch_phase(pool: &PgPool, p: PhaseLaunch<'_>) -> Result<(), String> {
|
|||||||
// one-time pairing code even for existing containers. Old codes
|
// one-time pairing code even for existing containers. Old codes
|
||||||
// expire / are single-use, so a launch that reuses a container
|
// expire / are single-use, so a launch that reuses a container
|
||||||
// still needs a fresh code for the topology_worker's next /pair.
|
// still needs a fresh code for the topology_worker's next /pair.
|
||||||
if let Some(prov) = crate::mission_runtime::MissionRuntimeProvisioner::from_env() {
|
//
|
||||||
|
// Skipped for a microVM mission: its agent runs in a VM on a fleet node, so
|
||||||
|
// a container here would be provisioned, have the checkout copied into it,
|
||||||
|
// and then sit idle for the life of the mission — while the pairing code and
|
||||||
|
// runtime binding it writes describe a runtime nothing is using.
|
||||||
|
let needs_container = p.runtime_kind != "microvm";
|
||||||
|
if let Some(prov) = crate::mission_runtime::MissionRuntimeProvisioner::from_env()
|
||||||
|
.filter(|_| needs_container)
|
||||||
|
{
|
||||||
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);
|
||||||
@@ -446,6 +500,25 @@ async fn launch_phase(pool: &PgPool, p: PhaseLaunch<'_>) -> Result<(), String> {
|
|||||||
// downstream lifecycle — close_finished_phases, evaluation, capture,
|
// downstream lifecycle — close_finished_phases, evaluation, capture,
|
||||||
// delivery — keys off those rows, and inventing a second completion path
|
// delivery — keys off those rows, and inventing a second completion path
|
||||||
// would mean two ways for a phase to finish and one of them untested.
|
// would mean two ways for a phase to finish and one of them untested.
|
||||||
|
// microVM executor. Checked BEFORE `direct_mode` because it is a property of
|
||||||
|
// the mission, not of the deployment: `runtime_kind='microvm'` was chosen for
|
||||||
|
// this mission specifically, and an env var that happens to be set must not
|
||||||
|
// silently run it somewhere else.
|
||||||
|
if p.runtime_kind == "microvm" {
|
||||||
|
return launch_microvm_phase(
|
||||||
|
pool,
|
||||||
|
hub,
|
||||||
|
mission_id,
|
||||||
|
phase_id,
|
||||||
|
workspace_id,
|
||||||
|
iteration,
|
||||||
|
&task,
|
||||||
|
p.backend,
|
||||||
|
p.target_node_id,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
if crate::session_executor::direct_mode() {
|
if crate::session_executor::direct_mode() {
|
||||||
return launch_direct_session(pool, mission_id, phase_id, workspace_id, iteration, &task)
|
return launch_direct_session(pool, mission_id, phase_id, workspace_id, iteration, &task)
|
||||||
.await;
|
.await;
|
||||||
@@ -518,6 +591,142 @@ async fn launch_phase(pool: &PgPool, p: PhaseLaunch<'_>) -> Result<(), String> {
|
|||||||
/// Returns as soon as the session is spawned: `launch_phase` runs inside the
|
/// Returns as soon as the session is spawned: `launch_phase` runs inside the
|
||||||
/// sweep loop, and blocking it for the length of a coding session would stall
|
/// sweep loop, and blocking it for the length of a coding session would stall
|
||||||
/// every other mission.
|
/// every other mission.
|
||||||
|
/// Launch a phase inside a Firecracker microVM on the mission's placed node.
|
||||||
|
///
|
||||||
|
/// Mirrors [`launch_direct_session`] on purpose, down to creating exactly ONE
|
||||||
|
/// `topology_runs` row: `close_finished_phases`, evaluation, capture and delivery
|
||||||
|
/// all key off those rows, and a second completion path would be a second way for
|
||||||
|
/// a phase to finish with one of them untested.
|
||||||
|
///
|
||||||
|
/// The run row and the phase flip happen **before** any fallible VM work, so a
|
||||||
|
/// configuration error (no subscription token, a node that lost its capability)
|
||||||
|
/// surfaces as a failed run an operator can see — not as a phase that stays
|
||||||
|
/// `pending` and is retried every ten seconds forever.
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
|
async fn launch_microvm_phase(
|
||||||
|
pool: &PgPool,
|
||||||
|
hub: &std::sync::Arc<crate::fleet::NodeHub>,
|
||||||
|
mission_id: Uuid,
|
||||||
|
phase_id: Uuid,
|
||||||
|
workspace_id: Uuid,
|
||||||
|
iteration: i32,
|
||||||
|
task: &str,
|
||||||
|
backend: Option<&str>,
|
||||||
|
target_node_id: Option<Uuid>,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
sqlx::query(
|
||||||
|
"DELETE FROM topology_runs
|
||||||
|
WHERE mission_phase_id = $1 AND status IN ('failed', 'cancelled')",
|
||||||
|
)
|
||||||
|
.bind(phase_id)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("purge prior runs for phase {phase_id}: {e}"))?;
|
||||||
|
|
||||||
|
let run_id = Uuid::now_v7();
|
||||||
|
sqlx::query(
|
||||||
|
"INSERT INTO topology_runs
|
||||||
|
(id, workspace_id, task, kind, status, graph, tier,
|
||||||
|
mission_id, mission_phase_id, iteration)
|
||||||
|
VALUES ($1, $2, $3, 'run', 'running', $4, 'microvm', $5, $6, $7)",
|
||||||
|
)
|
||||||
|
.bind(run_id)
|
||||||
|
.bind(workspace_id)
|
||||||
|
.bind(task)
|
||||||
|
.bind(serde_json::json!({ "nodes": [], "edges": [], "executor": "microvm" }))
|
||||||
|
.bind(mission_id)
|
||||||
|
.bind(phase_id)
|
||||||
|
.bind(iteration)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("enqueue microvm run for phase {phase_id}: {e}"))?;
|
||||||
|
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE mission_phases
|
||||||
|
SET status = 'running', started_at = now()
|
||||||
|
WHERE id = $1 AND status = 'pending'",
|
||||||
|
)
|
||||||
|
.bind(phase_id)
|
||||||
|
.execute(pool)
|
||||||
|
.await
|
||||||
|
.map_err(|e| format!("mark phase {phase_id} running: {e}"))?;
|
||||||
|
|
||||||
|
let repo = crate::mission_workspace::checkout_path(mission_id);
|
||||||
|
let task = task.to_string();
|
||||||
|
let backend = backend.map(str::to_string);
|
||||||
|
let pool2 = pool.clone();
|
||||||
|
let hub = hub.clone();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
// Everything fallible lives in here, so every outcome closes the run.
|
||||||
|
let outcome = async {
|
||||||
|
let node = target_node_id.ok_or_else(|| {
|
||||||
|
"mission has no target_node_id — placement should have set one at \
|
||||||
|
launch; a microvm mission cannot run on the gateway, which has no \
|
||||||
|
/dev/kvm"
|
||||||
|
.to_string()
|
||||||
|
})?;
|
||||||
|
if !repo.is_dir() {
|
||||||
|
// A research-only mission has no checkout, and there is nothing
|
||||||
|
// for a VM to work on. Better to say so than to boot one and hand
|
||||||
|
// the agent an empty directory.
|
||||||
|
return Err(format!(
|
||||||
|
"mission has no checkout at {} — a microvm phase needs a repository",
|
||||||
|
repo.display()
|
||||||
|
));
|
||||||
|
}
|
||||||
|
crate::microvm_executor::run_phase_in_vm(
|
||||||
|
&hub,
|
||||||
|
crate::microvm_executor::VmPhase {
|
||||||
|
node_id: cm_domain::NodeId::from(node),
|
||||||
|
mission_id,
|
||||||
|
phase_id,
|
||||||
|
iteration,
|
||||||
|
task: &task,
|
||||||
|
backend: backend.as_deref(),
|
||||||
|
repo: &repo,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
.await;
|
||||||
|
|
||||||
|
// The agent's own account is diagnostic only. Whether the phase succeeded
|
||||||
|
// is decided downstream by capture + delivery against the repository.
|
||||||
|
let (status, note) = match outcome {
|
||||||
|
Ok(o) if o.rc == 0 && o.collected => ("completed", o.summary),
|
||||||
|
// A turn that ran and could not be collected is a failure even when
|
||||||
|
// the agent was satisfied: the work did not reach the host, so there
|
||||||
|
// is nothing for delivery to find.
|
||||||
|
Ok(o) if !o.collected => (
|
||||||
|
"failed",
|
||||||
|
format!("the agent's work could not be collected from the VM: {}", o.summary),
|
||||||
|
),
|
||||||
|
Ok(o) => ("failed", o.summary),
|
||||||
|
Err(e) => ("failed", e),
|
||||||
|
};
|
||||||
|
eprintln!(
|
||||||
|
"phase_runner: microvm phase {phase_id} of mission {mission_id} → {status} — {}",
|
||||||
|
note.chars().take(300).collect::<String>()
|
||||||
|
);
|
||||||
|
if let Err(e) = sqlx::query(
|
||||||
|
"UPDATE topology_runs SET status = $2, updated_at = now() WHERE id = $1",
|
||||||
|
)
|
||||||
|
.bind(run_id)
|
||||||
|
.bind(status)
|
||||||
|
.execute(&pool2)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
eprintln!("phase_runner: could not close microvm run {run_id}: {e}");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
eprintln!(
|
||||||
|
"phase_runner: mission {mission_id} phase {phase_id} launched in a MICROVM \
|
||||||
|
on node {target_node_id:?}"
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
async fn launch_direct_session(
|
async fn launch_direct_session(
|
||||||
pool: &PgPool,
|
pool: &PgPool,
|
||||||
mission_id: Uuid,
|
mission_id: Uuid,
|
||||||
|
|||||||
Reference in New Issue
Block a user