use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::process::ExitCode;
use clap::{Args, Parser, Subcommand};
use lm_provision_driver::acquisition::{self as record, AcquisitionRow};
use lm_provision_driver::credentials;
use lm_provision_driver::infra;
use lm_provision_driver::inventory;
use lm_provision_driver::provisioner;
use lm_provision_driver::session::{self, InvokeMode, StepPlan};
use lm_provision_driver::ssh::{SshTransport, DEFAULT_REMOTE_DIR, DEFAULT_SSH_USER};
#[derive(Parser)]
#[command(
name = "lm-provision",
version,
about = "Provision a pod from a profile (apply / check), run the fleet it needs (machine list / acquire / release / sweep), and serve the same over MCP (mcp)"
)]
struct Cli {
#[command(subcommand)]
command: Command,
}
#[derive(Subcommand)]
enum Command {
Apply(ApplyArgs),
Check(CheckArgs),
Machine {
#[command(subcommand)]
command: MachineCommand,
},
Mcp,
}
#[derive(Subcommand)]
enum MachineCommand {
List(ListArgs),
Acquire(AcquireArgs),
Release(ReleaseArgs),
Sweep(SweepArgs),
}
#[derive(Args)]
struct ListArgs {
#[arg(long = "provider", required = true, num_args = 1..)]
providers: Vec<String>,
}
#[derive(Args)]
struct CheckArgs {
#[arg(long = "profile")]
profile: PathBuf,
#[arg(long = "inspected")]
inspected: PathBuf,
#[arg(long = "provider", default_value = "runpod")]
provider: String,
}
#[derive(Args)]
struct AcquireArgs {
#[arg(long = "profile")]
profile: PathBuf,
#[arg(long = "dry-run", default_value_t = true, action = clap::ArgAction::Set)]
dry_run: bool,
#[arg(long = "provider", default_value = "runpod")]
provider: String,
#[arg(long = "ttl-hours", default_value_t = 24)]
ttl_hours: u64,
#[arg(long = "acquisitions")]
acquisitions: Option<PathBuf>,
}
#[derive(Args)]
struct SweepArgs {
#[arg(long = "provider")]
providers: Vec<String>,
#[arg(long = "acquisitions")]
acquisitions: Option<PathBuf>,
#[arg(long = "ledger")]
ledger: Option<PathBuf>,
#[arg(long = "dry-run", default_value_t = true, action = clap::ArgAction::Set)]
dry_run: bool,
}
#[derive(Args)]
struct ReleaseArgs {
#[arg(long = "id")]
id: String,
#[arg(long = "provider", default_value = "runpod")]
provider: String,
#[arg(long = "profile")]
profile: PathBuf,
#[arg(long = "ledger")]
ledger: Option<PathBuf>,
#[arg(long = "force")]
force: bool,
#[arg(long = "acquisitions")]
acquisitions: Option<PathBuf>,
}
#[derive(Args)]
struct ApplyArgs {
#[arg(long = "ssh", help = ssh_help())]
ssh: String,
#[arg(long = "key")]
key: PathBuf,
#[arg(long = "profile")]
profile: PathBuf,
#[arg(long = "provisioner-path", alias = "artifact")]
provisioner_path: Option<PathBuf>,
#[arg(
long = "provisioner-version",
alias = "artifact-version",
default_value = provisioner::default_version(),
conflicts_with = "provisioner_path"
)]
provisioner_version: String,
#[arg(long = "remote-dir", default_value = DEFAULT_REMOTE_DIR)]
remote_dir: PathBuf,
#[arg(long = "pod-id")]
pod_id: Option<String>,
#[arg(long = "skip-install")]
skip_install: bool,
#[arg(long = "skip-verify")]
skip_verify: bool,
#[arg(long = "dry-run", conflicts_with = "validate_only")]
dry_run: bool,
#[arg(long = "validate-only")]
validate_only: bool,
#[arg(long = "no-ledger")]
no_ledger: bool,
#[arg(long = "ledger")]
ledger: Option<PathBuf>,
#[arg(long = "artifacts-dir", default_value = "artifacts")]
artifacts_dir: PathBuf,
#[arg(long = "no-artifacts")]
no_artifacts: bool,
}
fn main() -> ExitCode {
credentials::load();
let cli = Cli::parse();
match cli.command {
Command::Apply(args) => run_apply(args),
Command::Check(args) => run_check(args),
Command::Machine { command } => match command {
MachineCommand::List(args) => run_machine_list(args),
MachineCommand::Acquire(args) => run_acquire(args),
MachineCommand::Release(args) => run_release(args),
MachineCommand::Sweep(args) => run_sweep(args),
},
Command::Mcp => run_mcp(),
}
}
fn run_mcp() -> ExitCode {
use rmcp::transport::io::stdio;
use rmcp::ServiceExt as _;
tracing_subscriber::fmt()
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
.with_writer(std::io::stderr)
.init();
let served = tokio::runtime::Runtime::new()
.map_err(anyhow::Error::from)
.and_then(|runtime| {
runtime.block_on(async {
let config = lm_provision_mcp::config::Config::from_env()?;
let service = lm_provision_mcp::server::LmProvisionServer::new(config)
.serve(stdio())
.await?;
service.waiting().await?;
Ok(())
})
});
match served {
Ok(()) => ExitCode::SUCCESS,
Err(err) => {
eprintln!("error: {err:#}");
ExitCode::FAILURE
}
}
}
fn run_machine_list(args: ListArgs) -> ExitCode {
let providers: Vec<String> = args
.providers
.iter()
.collect::<BTreeSet<_>>()
.into_iter()
.cloned()
.collect();
let listing = inventory::list(&providers);
for (program, said) in &listing.said {
relay(program, said);
}
for (provider, reason) in &listing.failed {
eprintln!("error: could not list {provider}: {reason}");
}
println!("{}", listing.artifact());
if listing.complete() {
ExitCode::SUCCESS
} else {
ExitCode::FAILURE
}
}
fn run_check(args: CheckArgs) -> ExitCode {
let ProfileFacts { required, .. } = match requirements_of(&args.profile) {
Ok(facts) => facts,
Err(message) => {
eprintln!("error: {message}");
return ExitCode::from(2);
}
};
let text = match std::fs::read_to_string(&args.inspected) {
Ok(text) => text,
Err(err) => {
eprintln!("error: reading {}: {err}", args.inspected.display());
return ExitCode::from(2);
}
};
let inspected: serde_json::Value = match serde_json::from_str(&text) {
Ok(value) => value,
Err(err) => {
eprintln!("error: {} is not JSON: {err}", args.inspected.display());
return ExitCode::from(2);
}
};
let adapter = match infra::adapter_named(&args.provider) {
Ok(adapter) => adapter,
Err(message) => {
eprintln!("error: {message}");
return ExitCode::from(2);
}
};
let state = adapter.read_state(&inspected);
let findings = lm_provision::machine::observe(&required, &state);
let verdict = lm_provision::machine::verdict(&findings);
println!(
"{}",
serde_json::json!({
"verdict": format!("{verdict:?}"),
"findings": findings
.iter()
.map(|it| serde_json::json!({
"requirement": it.requirement,
"outcome": format!("{:?}", it.outcome),
}))
.collect::<Vec<_>>(),
})
);
match verdict {
lm_provision::machine::Outcome::Satisfied => ExitCode::SUCCESS,
_ => ExitCode::FAILURE,
}
}
struct ProfileFacts {
required: lm_provision::machine::Requirements,
provider: BTreeMap<String, String>,
hash: String,
}
fn requirements_of(profile: &std::path::Path) -> Result<ProfileFacts, String> {
let root = lm_provision::frontend::load_profile(profile).map_err(|err| err.to_string())?;
let root = lm_provision::resolve::resolve(root, profile).map_err(|err| err.to_string())?;
lm_provision::validate::validate(&root).map_err(|err| err.to_string())?;
let hash = lm_provision::canonical::hash(&root);
let lm_provision::profile_ast::ProfileNode::Spec {
requires_ports,
requires_gpu,
requires_disk,
provider,
..
} = &root
else {
return Err("the profile's root is not a Spec".to_string());
};
let required = lm_provision::machine::Requirements::from_slots(
requires_ports,
requires_gpu,
requires_disk,
)
.map_err(|err| err.to_string())?;
Ok(ProfileFacts {
required,
provider: provider.clone(),
hash,
})
}
fn run_acquire(args: AcquireArgs) -> ExitCode {
let ProfileFacts {
required,
provider,
hash: profile_hash,
} = match requirements_of(&args.profile) {
Ok(facts) => facts,
Err(message) => {
eprintln!("error: {message}");
return ExitCode::from(2);
}
};
let ttl = match i64::try_from(args.ttl_hours)
.map_err(|err| err.to_string())
.and_then(|hours| {
jiff::Span::new()
.try_hours(hours)
.map_err(|err| err.to_string())
}) {
Ok(ttl) => ttl,
Err(message) => {
eprintln!(
"error: --ttl-hours {} is not a lease: {message}",
args.ttl_hours
);
return ExitCode::from(2);
}
};
let adapter = match infra::adapter_named(&args.provider) {
Ok(adapter) => adapter,
Err(message) => {
eprintln!("error: {message}");
return ExitCode::from(2);
}
};
if let Err(refusal) = lm_provision::machine::admit(&required, &adapter.capability()) {
eprintln!("error: {refusal}");
return ExitCode::from(3);
}
let acquired_at = jiff::Timestamp::now();
let expires_at = match acquired_at.checked_add(ttl) {
Ok(expires_at) => expires_at,
Err(err) => {
eprintln!(
"warning: could not stamp an expiry {} hours out: {err}",
args.ttl_hours
);
acquired_at
}
};
let acquisition = match adapter.acquisition(&required, &provider, Some(expires_at)) {
Ok(acquisition) => acquisition,
Err(err) => {
eprintln!("error: {err}");
return ExitCode::from(2);
}
};
if args.dry_run {
println!("{}", dry_run_artifact(&acquisition));
return ExitCode::SUCCESS;
}
let release_template = acquisition.release.clone();
if let Some(image) = adapter.image_key().and_then(|key| provider.get(key)) {
match lm_provision_driver::image::manifest_check(image) {
lm_provision_driver::image::Manifest::Present => {}
lm_provision_driver::image::Manifest::Absent { registry } => {
eprintln!(
"error: image {image} is not in {registry} (manifest unknown); a machine \
created for it would retry the pull forever while billing"
);
return ExitCode::from(3);
}
lm_provision_driver::image::Manifest::Undetermined { reason } => {
eprintln!("note: could not preflight image {image}: {reason}; proceeding");
}
}
}
if let Err(missing) = credentials::require(adapter.provider_namespace(), adapter.credentials())
{
eprintln!("error: {missing}");
return ExitCode::from(4);
}
let mut acquired = match infra::acquire(acquisition) {
Ok(acquired) => acquired,
Err(err) => {
eprintln!("error: {err}");
return ExitCode::FAILURE;
}
};
eprintln!("acquired {}", acquired.id);
let acquisitions_path = args
.acquisitions
.clone()
.unwrap_or_else(default_acquisitions_path);
let row = AcquisitionRow {
id: acquired.id.clone(),
provider: args.provider.clone(),
acquired_at: acquired_at.to_string(),
expires_at: expires_at.to_string(),
profile_hash,
release: release_template,
released_at: None,
};
if let Err(err) = record_acquisition(&acquisitions_path, &row) {
eprintln!(
"error: could not record {} in {}: {err}",
acquired.id,
acquisitions_path.display()
);
eprintln!(
"note: {} is running and unrecorded — no sweep will find it; \
release it with `lm-provision machine release --id {} --provider {} --profile {}`",
acquired.id,
acquired.id,
args.provider,
args.profile.display()
);
}
if let Err(err) = acquired.inspect() {
eprintln!(
"warning: created {} but could not inspect it yet; retrying while waiting: {err}",
acquired.id
);
}
let started = std::time::Instant::now();
let deadline = started + ACQUIRE_REACHABILITY_TIMEOUT;
let cap = started + ACQUIRE_MATERIALIZING_CAP;
let mut extended = false;
let mut connection = adapter.connection(&acquired.inspected);
while !connection_covers(&required.ports, &connection) {
let now = std::time::Instant::now();
if now >= deadline {
if now < cap && adapter.still_materializing(&acquired.inspected) {
if !extended {
extended = true;
eprintln!(
"note: {} says it is still materializing after {}s; \
waiting up to {}s for it",
acquired.id,
ACQUIRE_REACHABILITY_TIMEOUT.as_secs(),
ACQUIRE_MATERIALIZING_CAP.as_secs()
);
}
} else {
eprintln!(
"warning: {} still has unanswered ports after {}s; reporting what is known",
acquired.id,
now.duration_since(started).as_secs()
);
break;
}
}
std::thread::sleep(ACQUIRE_REACHABILITY_POLL);
if let Err(err) = acquired.inspect() {
eprintln!(
"warning: {} answered inspection with an error; retrying: {err}",
acquired.id
);
continue;
}
connection = adapter.connection(&acquired.inspected);
}
let state = adapter.read_state(&acquired.inspected);
let findings = lm_provision::machine::observe(&required, &state);
let verdict = lm_provision::machine::verdict(&findings);
println!(
"{}",
serde_json::json!({
"id": acquired.id,
"verdict": format!("{verdict:?}"),
"findings": findings
.iter()
.map(|it| serde_json::json!({
"requirement": it.requirement,
"outcome": format!("{:?}", it.outcome),
}))
.collect::<Vec<_>>(),
"connection": connection,
"release": acquired.id,
})
);
match verdict {
lm_provision::machine::Outcome::Satisfied => ExitCode::SUCCESS,
_ => ExitCode::FAILURE,
}
}
const ACQUIRE_REACHABILITY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);
const ACQUIRE_REACHABILITY_POLL: std::time::Duration = std::time::Duration::from_secs(5);
const ACQUIRE_MATERIALIZING_CAP: std::time::Duration = std::time::Duration::from_secs(1200);
fn dry_run_artifact(acquisition: &infra::Acquisition) -> serde_json::Value {
serde_json::json!({
"dry_run": true,
"discover": acquisition.discover,
"create": acquisition.create,
"body": acquisition.body,
"release": acquisition.release,
})
}
fn connection_covers(
required: &[lm_provision::machine::PortRequirement],
connection: &infra::Connection,
) -> bool {
required
.iter()
.all(|it| connection.endpoints.contains_key(&it.port))
}
fn run_release(args: ReleaseArgs) -> ExitCode {
let ledger_path = args.ledger.clone().unwrap_or_else(default_ledger_path);
match uncollected_artifacts(&ledger_path, &args.id) {
Ok(uncollected) if !uncollected.is_empty() => {
for path in &uncollected {
eprintln!("error: artifact not collected: {path}");
}
if args.force {
eprintln!(
"warning: releasing {} anyway (--force); the artifacts above are deleted \
with it",
args.id
);
} else {
eprintln!(
"error: refusing to release {}: the newest apply recorded in {} left the \
artifacts above on the machine (re-run apply to collect them, or pass \
--force to delete them with it)",
args.id,
ledger_path.display()
);
return ExitCode::from(3);
}
}
Ok(_) => {}
Err(err) => {
eprintln!(
"error: release gate could not read {}: {err}",
ledger_path.display()
);
if !args.force {
eprintln!("note: pass --force to release without the gate");
return ExitCode::from(3);
}
}
}
let ProfileFacts {
required,
provider,
hash: profile_hash,
} = match requirements_of(&args.profile) {
Ok(facts) => facts,
Err(message) => {
eprintln!("error: {message}");
return ExitCode::from(2);
}
};
let adapter = match infra::adapter_named(&args.provider) {
Ok(adapter) => adapter,
Err(message) => {
eprintln!("error: {message}");
return ExitCode::from(2);
}
};
let acquisition = match adapter.acquisition(&required, &provider, None) {
Ok(acquisition) => acquisition,
Err(err) => {
eprintln!("error: {err}");
return ExitCode::from(2);
}
};
if let Err(missing) = credentials::require(adapter.provider_namespace(), adapter.credentials())
{
eprintln!("error: {missing}");
eprintln!("note: {} is still running", args.id);
return ExitCode::from(4);
}
let argv = substitute(&acquisition.release, &args.id);
match std::process::Command::new(&argv[0])
.args(&argv[1..])
.output()
{
Ok(output) if output.status.success() => {
relay(&argv[0], &output.stdout);
relay(&argv[0], &output.stderr);
let acquisitions_path = args
.acquisitions
.clone()
.unwrap_or_else(default_acquisitions_path);
let correction = correction_row(
&acquisitions_path,
&args.id,
&args.provider,
&acquisition.release,
&profile_hash,
);
if let Err(err) = record_acquisition(&acquisitions_path, &correction) {
eprintln!(
"error: released {} but could not record it in {}: {err}",
args.id,
acquisitions_path.display()
);
}
println!("{}", serde_json::json!({ "released": args.id }));
ExitCode::SUCCESS
}
Ok(output) => {
relay(&argv[0], &output.stdout);
relay(&argv[0], &output.stderr);
eprintln!("error: release exited with {}", output.status);
eprintln!("note: {} may still be running", args.id);
ExitCode::FAILURE
}
Err(err) => {
eprintln!("error: could not run the release: {err}");
eprintln!("note: {} is still running", args.id);
ExitCode::FAILURE
}
}
}
#[derive(Debug, PartialEq, Eq)]
enum Due {
Live,
Expired,
Refused(String),
Failed(String),
}
fn due(row: &AcquisitionRow, now: jiff::Timestamp, ledger_path: &std::path::Path) -> Due {
let expires_at = match row.expires_at.parse::<jiff::Timestamp>() {
Ok(expires_at) => expires_at,
Err(err) => {
return Due::Failed(format!(
"expires_at {:?} is not a timestamp: {err}",
row.expires_at
))
}
};
if expires_at > now {
return Due::Live;
}
match gate(ledger_path, &row.id) {
Gate::Clear => Due::Expired,
Gate::Holding(reason) => Due::Refused(reason),
Gate::Unreadable(reason) => Due::Failed(reason),
}
}
#[derive(Debug, PartialEq, Eq)]
enum Gate {
Clear,
Holding(String),
Unreadable(String),
}
fn gate(ledger_path: &std::path::Path, id: &str) -> Gate {
match uncollected_artifacts(ledger_path, id) {
Ok(uncollected) if !uncollected.is_empty() => Gate::Holding(format!(
"the newest apply left {} on the machine (collect them, or release --force)",
uncollected.join(", ")
)),
Ok(_) => Gate::Clear,
Err(err) => Gate::Unreadable(format!(
"the release gate could not read {}: {err}",
ledger_path.display()
)),
}
}
#[derive(Debug, PartialEq, Eq)]
enum Standing {
Live,
Expired,
Unknown,
}
fn standing(machine: &infra::Machine, now: jiff::Timestamp) -> Standing {
match machine.name.as_deref().and_then(infra::expiry_of) {
None => Standing::Unknown,
Some(expires_at) if expires_at > now => Standing::Live,
Some(_) => Standing::Expired,
}
}
#[derive(Debug, PartialEq, Eq)]
enum Bookkeeping {
Handled,
Gone,
Judge,
}
fn bookkeeping(
row: &AcquisitionRow,
handled: &BTreeSet<String>,
listed: &BTreeMap<String, BTreeSet<String>>,
) -> Bookkeeping {
if handled.contains(&row.id) {
return Bookkeeping::Handled;
}
match listed.get(&row.provider) {
Some(present) if !present.contains(&row.id) => Bookkeeping::Gone,
_ => Bookkeeping::Judge,
}
}
#[derive(Debug, Default)]
struct SweepOutcome {
dry_run: bool,
expired: usize,
released: Vec<String>,
refused: Vec<(String, String)>,
failed: Vec<(String, String)>,
unknown: Vec<(String, String)>,
}
fn sweep_artifact(outcome: &SweepOutcome) -> serde_json::Value {
let pairs = |entries: &[(String, String)]| {
entries
.iter()
.map(|(id, reason)| serde_json::json!({ "id": id, "reason": reason }))
.collect::<Vec<_>>()
};
serde_json::json!({
"dry_run": outcome.dry_run,
"expired": outcome.expired,
"released": outcome.released,
"refused": pairs(&outcome.refused),
"failed": pairs(&outcome.failed),
"unknown": outcome
.unknown
.iter()
.map(|(id, name)| serde_json::json!({ "id": id, "name_or_label": name }))
.collect::<Vec<_>>(),
})
}
fn sweep_exit(outcome: &SweepOutcome) -> u8 {
if outcome.failed.is_empty() {
0
} else {
1
}
}
fn run_sweep(args: SweepArgs) -> ExitCode {
let acquisitions_path = args.acquisitions.unwrap_or_else(default_acquisitions_path);
let ledger_path = args.ledger.unwrap_or_else(default_ledger_path);
let outstanding = match record::outstanding(&acquisitions_path) {
Ok(rows) => rows,
Err(err) => {
eprintln!(
"error: could not read the acquisitions record {}: {err}",
acquisitions_path.display()
);
return ExitCode::FAILURE;
}
};
let now = jiff::Timestamp::now();
let mut outcome = SweepOutcome {
dry_run: args.dry_run,
..SweepOutcome::default()
};
let mut handled = BTreeSet::new();
let mut listed: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
for name in args.providers.iter().collect::<BTreeSet<_>>() {
let (fleet, machines) = match listing(name) {
Ok(listing) => listing,
Err(reason) => {
outcome.failed.push((name.clone(), reason));
continue;
}
};
let mut present = BTreeSet::new();
for machine in machines {
present.insert(machine.id.clone());
match standing(&machine, now) {
Standing::Live => {
handled.insert(machine.id);
}
Standing::Unknown => {
outcome
.unknown
.push((machine.id, machine.name.unwrap_or_default()));
}
Standing::Expired => {
outcome.expired += 1;
handled.insert(machine.id.clone());
match gate(&ledger_path, &machine.id) {
Gate::Clear => {}
Gate::Holding(reason) => {
outcome.refused.push((machine.id, reason));
continue;
}
Gate::Unreadable(reason) => {
outcome.failed.push((machine.id, reason));
continue;
}
}
if args.dry_run {
outcome.released.push(machine.id);
continue;
}
match run_release_argv(&substitute(&fleet.release, &machine.id)) {
Ok(()) => {
if let Some(row) = outstanding.iter().find(|it| it.id == machine.id) {
retire(row, &acquisitions_path);
}
outcome.released.push(machine.id);
}
Err(reason) => outcome.failed.push((machine.id, reason)),
}
}
}
}
listed.insert(name.clone(), present);
}
for row in &outstanding {
match bookkeeping(row, &handled, &listed) {
Bookkeeping::Handled => {}
Bookkeeping::Gone => {
eprintln!(
"note: {} is not in {}'s list of running machines; it is gone{}",
row.id,
row.provider,
if args.dry_run {
", and the record would be corrected to say so"
} else {
", and the record is being corrected to say so"
}
);
if !args.dry_run {
retire(row, &acquisitions_path);
}
}
Bookkeeping::Judge => match due(row, now, &ledger_path) {
Due::Live => {}
Due::Failed(reason) => outcome.failed.push((row.id.clone(), reason)),
Due::Refused(reason) => {
outcome.expired += 1;
outcome.refused.push((row.id.clone(), reason));
}
Due::Expired => {
outcome.expired += 1;
if args.dry_run {
outcome.released.push(row.id.clone());
continue;
}
match release_recorded(row, &acquisitions_path) {
Ok(()) => outcome.released.push(row.id.clone()),
Err(reason) => outcome.failed.push((row.id.clone(), reason)),
}
}
},
}
}
println!("{}", sweep_artifact(&outcome));
ExitCode::from(sweep_exit(&outcome))
}
fn listing(name: &str) -> Result<(infra::Fleet, Vec<infra::Machine>), String> {
let fetched = inventory::fetch(name)?;
let (program, said) = &fetched.said;
relay(program, said);
Ok((fetched.fleet, fetched.machines))
}
fn release_recorded(
row: &AcquisitionRow,
acquisitions_path: &std::path::Path,
) -> Result<(), String> {
let adapter = infra::adapter_named(&row.provider)?;
credentials::require(adapter.provider_namespace(), adapter.credentials())
.map_err(|missing| missing.to_string())?;
run_release_argv(&substitute(&row.release, &row.id))?;
retire(row, acquisitions_path);
Ok(())
}
fn run_release_argv(argv: &[String]) -> Result<(), String> {
let Some(program) = argv.first() else {
return Err("the release names no command to run".to_string());
};
let output = std::process::Command::new(program)
.args(&argv[1..])
.output()
.map_err(|err| format!("could not run `{program}`: {err}"))?;
relay(program, &output.stdout);
relay(program, &output.stderr);
if !output.status.success() {
return Err(format!("`{program}` exited with {}", output.status));
}
Ok(())
}
fn substitute(argv: &[String], id: &str) -> Vec<String> {
argv.iter().map(|it| it.replace("{id}", id)).collect()
}
fn retire(row: &AcquisitionRow, acquisitions_path: &std::path::Path) {
let correction = AcquisitionRow {
released_at: Some(jiff::Timestamp::now().to_string()),
..row.clone()
};
if let Err(err) = record_acquisition(acquisitions_path, &correction) {
eprintln!(
"error: {} is no longer running but could not be recorded in {}: {err}",
row.id,
acquisitions_path.display()
);
}
}
fn relay(program: &str, bytes: &[u8]) {
for line in attributed(program, bytes) {
eprintln!("{line}");
}
}
fn attributed(program: &str, bytes: &[u8]) -> Vec<String> {
let text = String::from_utf8_lossy(bytes);
let text = text.trim();
if text.is_empty() || text == "\"\"" {
return Vec::new();
}
text.lines()
.map(|line| format!("{program}: {line}"))
.collect()
}
fn run_apply(args: ApplyArgs) -> ExitCode {
let (user, host, port) = match parse_ssh_target(&args.ssh) {
Ok(parts) => parts,
Err(message) => {
eprintln!("error: {message}");
return ExitCode::from(2);
}
};
let mode = if args.validate_only {
InvokeMode::ValidateOnly
} else if args.dry_run {
InvokeMode::DryRun
} else {
InvokeMode::Apply
};
let ledger = if args.no_ledger {
None
} else {
Some(args.ledger.unwrap_or_else(default_ledger_path))
};
let plan = StepPlan {
skip_install: args.skip_install,
skip_verify: args.skip_verify,
mode,
artifacts_dir: if args.no_artifacts {
None
} else {
Some(args.artifacts_dir)
},
ledger,
};
let artifact = match (args.provisioner_path, args.skip_install) {
(Some(path), _) => path,
(None, true) => PathBuf::from(provisioner::BINARY_NAME),
(None, false) => match provisioner::resolve(&args.provisioner_version) {
Ok(resolved) => {
match &resolved.source {
provisioner::Source::Fetched { url } => {
eprintln!("provisioner: fetched and verified {url}")
}
provisioner::Source::Cached => eprintln!(
"provisioner: {} (cached, version {})",
resolved.path.display(),
args.provisioner_version
),
provisioner::Source::Override => {}
}
resolved.path
}
Err(error) => {
eprintln!("error: {error}");
return ExitCode::from(1);
}
},
};
let pod_id = args.pod_id.unwrap_or_else(|| host.clone());
let transport = SshTransport::new(host, port, user, args.key, args.remote_dir);
match session::run(&transport, &plan, &artifact, &args.profile, &pod_id) {
Ok(output) => {
eprint!("{}", output.collected.stderr);
println!("{}", output.collected.report);
if let Some(warning) = &output.ledger_warning {
eprintln!("error: ledger append failed: {warning}");
}
let uncollected = output.artifacts.iter().filter(|it| !it.collected).count();
for it in output.artifacts.iter().filter(|it| !it.collected) {
eprintln!(
"error: artifact not collected: {}: {}",
it.path,
it.error.as_deref().unwrap_or("unknown")
);
}
let ok = output.collected.report["ok"] == serde_json::Value::Bool(true);
ExitCode::from(exit_status(
ok,
output.ledger_warning.as_deref(),
uncollected,
))
}
Err(err) => {
eprintln!("error: {err}");
ExitCode::from(1)
}
}
}
fn exit_status(report_ok: bool, ledger_warning: Option<&str>, uncollected_artifacts: usize) -> u8 {
if report_ok && ledger_warning.is_none() && uncollected_artifacts == 0 {
0
} else {
1
}
}
fn ssh_help() -> String {
format!("SSH target as [user@]host:port (user defaults to {DEFAULT_SSH_USER})")
}
fn parse_ssh_target(target: &str) -> Result<(String, String, u16), String> {
let (user, rest) = match target.split_once('@') {
Some((user, rest)) => (user.to_string(), rest),
None => (DEFAULT_SSH_USER.to_string(), target),
};
let (host, port) = rest
.split_once(':')
.ok_or_else(|| format!("--ssh target {target:?} must be [user@]host:port"))?;
if host.is_empty() {
return Err(format!("--ssh target {target:?} has an empty host"));
}
let port: u16 = port
.parse()
.map_err(|_| format!("--ssh target {target:?} has a non-numeric port"))?;
Ok((user, host.to_string(), port))
}
fn uncollected_artifacts(
ledger_path: &std::path::Path,
pod_id: &str,
) -> Result<Vec<String>, lm_provision_driver::ledger::LedgerError> {
let newest_real_apply = lm_provision_driver::ledger::list(ledger_path)?
.into_iter()
.find(|row| row.pod_id == pod_id && row.report["dry_run"] != serde_json::Value::Bool(true));
Ok(newest_real_apply
.map(|row| {
row.artifacts
.into_iter()
.filter(|it| !it.collected)
.map(|it| it.path)
.collect()
})
.unwrap_or_default())
}
fn default_ledger_path() -> PathBuf {
match std::env::var_os("HOME") {
Some(home) => PathBuf::from(home)
.join(".lm-provision")
.join("ledger.jsonl"),
None => PathBuf::from("lm-provision-ledger.jsonl"),
}
}
fn default_acquisitions_path() -> PathBuf {
match std::env::var_os("HOME") {
Some(home) => PathBuf::from(home)
.join(".lm-provision")
.join("acquisitions.jsonl"),
None => PathBuf::from("lm-provision-acquisitions.jsonl"),
}
}
fn record_acquisition(
path: &std::path::Path,
row: &AcquisitionRow,
) -> Result<(), record::AcquisitionError> {
if let Some(parent) = path.parent().filter(|it| !it.as_os_str().is_empty()) {
std::fs::create_dir_all(parent)?;
}
record::append(path, row)
}
fn correction_row(
path: &std::path::Path,
id: &str,
provider: &str,
release: &[String],
profile_hash: &str,
) -> AcquisitionRow {
let released_at = jiff::Timestamp::now().to_string();
let recorded = match record::list(path) {
Ok(rows) => rows.into_iter().find(|row| row.id == id),
Err(err) => {
eprintln!(
"warning: could not read {} for {id}'s acquisition; the correction will carry \
only what this release knows: {err}",
path.display()
);
None
}
};
match recorded {
Some(row) => AcquisitionRow {
released_at: Some(released_at),
..row
},
None => AcquisitionRow {
id: id.to_string(),
provider: provider.to_string(),
acquired_at: released_at.clone(),
expires_at: released_at.clone(),
profile_hash: profile_hash.to_string(),
release: release.to_vec(),
released_at: Some(released_at),
},
}
}
#[cfg(test)]
mod tests {
use super::{
attributed, exit_status, parse_ssh_target, record, ssh_help, AcquisitionRow, Cli, Command,
MachineCommand, PathBuf,
};
use clap::Parser as _;
use lm_provision_driver::ssh::{DEFAULT_REMOTE_DIR, DEFAULT_SSH_USER};
#[test]
fn only_what_the_service_actually_said_is_relayed() {
assert!(attributed("runpod-cli", b"").is_empty());
assert!(attributed("runpod-cli", b" \n").is_empty());
assert!(
attributed("runpod-cli", b"\"\"\n").is_empty(),
"an empty JSON string is a body with nothing in it"
);
assert_eq!(
attributed("runpod-cli", b"warning: pod was already gone\n"),
vec!["runpod-cli: warning: pod was already gone"],
"the GNU form: the program that said it, a colon, the message"
);
assert_eq!(
attributed("runpod-cli", b"first\nsecond\n"),
vec!["runpod-cli: first", "runpod-cli: second"],
"every line carries the attribution, not just the first"
);
assert_eq!(
attributed("vastai", b"destroyed\n"),
vec!["vastai: destroyed"],
"the prefix is whichever service spoke — a sweep releases \
machines from two platforms in one run"
);
}
#[test]
fn cli_defaults_are_the_shared_ssh_constants() {
let cli = Cli::parse_from([
"lm-provision",
"apply",
"--ssh",
"1.2.3.4:22",
"--key",
"/k",
"--profile",
"profile.json",
"--skip-install",
]);
let Command::Apply(args) = cli.command else {
panic!("the parsed subcommand is `apply`");
};
assert_eq!(args.remote_dir, PathBuf::from(DEFAULT_REMOTE_DIR));
let (user, _, _) = parse_ssh_target("1.2.3.4:22").expect("host:port parses");
assert_eq!(user, DEFAULT_SSH_USER);
assert!(
ssh_help().contains(DEFAULT_SSH_USER),
"--help must document the same default it applies"
);
}
#[test]
fn parse_ssh_target_accepts_user_host_port_and_defaults_root() {
assert_eq!(
parse_ssh_target("root@1.2.3.4:2222").unwrap(),
("root".to_string(), "1.2.3.4".to_string(), 2222)
);
assert_eq!(
parse_ssh_target("1.2.3.4:22").unwrap(),
("root".to_string(), "1.2.3.4".to_string(), 22)
);
}
#[test]
fn parse_ssh_target_rejects_missing_or_bad_port_and_empty_host() {
assert!(parse_ssh_target("1.2.3.4").is_err());
assert!(parse_ssh_target("1.2.3.4:abc").is_err());
assert!(parse_ssh_target("root@:22").is_err());
}
#[test]
fn an_ok_report_with_a_failed_ledger_append_still_exits_nonzero() {
assert_eq!(exit_status(true, None, 0), 0);
assert_eq!(
exit_status(true, Some("ledger i/o error: no such file"), 0),
1
);
assert_eq!(exit_status(false, None, 0), 1);
assert_eq!(
exit_status(false, Some("ledger i/o error: no such file"), 0),
1
);
}
#[test]
fn an_ok_report_with_an_uncollected_artifact_still_exits_nonzero() {
assert_eq!(exit_status(true, None, 1), 1);
assert_eq!(exit_status(true, None, 0), 0);
}
#[test]
fn acquire_waits_on_exactly_the_declared_ports() {
use lm_provision::machine::{Exposure, PortRequirement};
use lm_provision_driver::infra::Connection;
let declared = [
PortRequirement {
port: 22,
exposure: Exposure::RawTcp,
},
PortRequirement {
port: 8188,
exposure: Exposure::PublicHttp,
},
];
let mut connection = Connection::default();
assert!(!super::connection_covers(&declared, &connection));
connection
.endpoints
.insert(22, "203.0.113.10:22016".to_string());
assert!(!super::connection_covers(&declared, &connection));
connection
.endpoints
.insert(8188, "203.0.113.10:80".to_string());
assert!(super::connection_covers(&declared, &connection));
assert!(
super::connection_covers(&[], &Connection::default()),
"no declared ports leaves nothing to wait for"
);
}
#[test]
fn the_release_gate_reads_the_newest_real_apply_row() {
use lm_provision_driver::ledger::{self, ArtifactRow, LedgerRow};
let path = std::env::temp_dir().join(format!(
"lm-provision-cli-release-gate-test-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::SystemTime::UNIX_EPOCH)
.expect("system time")
.as_nanos()
));
let row = |pod_id: &str, dry_run: bool, artifacts: Vec<ArtifactRow>| LedgerRow {
pod_id: pod_id.to_string(),
profile_hash: "h".repeat(64),
report: serde_json::json!({ "ok": true, "dry_run": dry_run }),
collected_at: "2026-08-30T00:00:00Z".to_string(),
artifacts,
};
let uncollected_row = ArtifactRow {
path: "/workspace/out".to_string(),
collected: false,
dest: None,
error: Some("scp failed".to_string()),
};
ledger::append(&path, &row("pod-a", false, vec![uncollected_row.clone()]))
.expect("append 1");
ledger::append(&path, &row("pod-b", false, Vec::new())).expect("append 2");
ledger::append(&path, &row("pod-a", true, Vec::new())).expect("append 3");
assert_eq!(
super::uncollected_artifacts(&path, "pod-a").expect("gate reads the ledger"),
vec!["/workspace/out".to_string()],
"the dry-run row must not mask the real apply's debt"
);
assert!(super::uncollected_artifacts(&path, "pod-b")
.expect("gate reads the ledger")
.is_empty());
assert!(super::uncollected_artifacts(&path, "pod-never-applied")
.expect("gate reads the ledger")
.is_empty());
ledger::append(
&path,
&row(
"pod-a",
false,
vec![ArtifactRow {
collected: true,
dest: Some("artifacts/pod-a/workspace/out".to_string()),
error: None,
..uncollected_row
}],
),
)
.expect("append 4");
assert!(super::uncollected_artifacts(&path, "pod-a")
.expect("gate reads the ledger")
.is_empty());
std::fs::remove_file(&path).ok();
}
fn scratch(name: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"lm-provision-cli-{name}-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::SystemTime::UNIX_EPOCH)
.expect("system time")
.as_nanos()
))
}
#[test]
fn requirements_survive_a_fragment_import() {
let dir = scratch("requirements-import-test");
std::fs::create_dir_all(&dir).expect("create fixture dir");
std::fs::write(
dir.join("fragment.json"),
serde_json::json!({
"type": "Fragment",
"name": "requirements-fragment",
"capabilities": ["sh.exec"],
"phases": [{ "type": "ShExec", "argv": ["echo", "from-fragment"] }]
})
.to_string(),
)
.expect("write fragment");
let profile = dir.join("profile.json");
std::fs::write(
&profile,
serde_json::json!({
"type": "Spec",
"name": "importing-requirements",
"requires_gpu": { "count": "1", "min_vram_gb": "24" },
"provider": { "runpod.imageName": "example/image:tag" },
"phases": [{ "type": "Import", "src": "./fragment.json" }]
})
.to_string(),
)
.expect("write profile");
let super::ProfileFacts {
required, provider, ..
} = super::requirements_of(&profile).expect("an importing profile has requirements too");
let gpu = required.gpu.expect("the consumer declared a GPU");
assert_eq!(gpu.count, 1);
assert_eq!(gpu.min_vram_gb, Some(24));
assert_eq!(
provider.get("runpod.imageName").map(String::as_str),
Some("example/image:tag")
);
std::fs::remove_dir_all(&dir).ok();
}
fn recorded(id: &str, expires_at: &str) -> AcquisitionRow {
AcquisitionRow {
id: id.to_string(),
provider: "runpod".to_string(),
acquired_at: "2026-09-01T00:00:00Z".to_string(),
expires_at: expires_at.to_string(),
profile_hash: "h".repeat(64),
release: vec![
"runpod-cli".to_string(),
"pods".to_string(),
"delete-pod".to_string(),
"{id}".to_string(),
],
released_at: None,
}
}
#[test]
fn a_sweep_is_due_at_the_lease_and_refused_by_the_release_gate() {
use lm_provision_driver::ledger::{self, ArtifactRow, LedgerRow};
let ledger_path = scratch("sweep-gate");
let now: jiff::Timestamp = "2026-09-02T00:00:00Z".parse().expect("a fixed clock");
assert_eq!(
super::due(
&recorded("pod-live", "2026-09-02T00:00:01Z"),
now,
&ledger_path
),
super::Due::Live,
"a lease with a second left is a machine the sweep leaves alone"
);
assert_eq!(
super::due(
&recorded("pod-due", "2026-09-02T00:00:00Z"),
now,
&ledger_path
),
super::Due::Expired,
"expiry is reached, not merely passed — and no recorded apply is no debt"
);
ledger::append(
&ledger_path,
&LedgerRow {
pod_id: "pod-owing".to_string(),
profile_hash: "h".repeat(64),
report: serde_json::json!({ "ok": true, "dry_run": false }),
collected_at: "2026-09-01T12:00:00Z".to_string(),
artifacts: vec![ArtifactRow {
path: "/workspace/out".to_string(),
collected: false,
dest: None,
error: Some("scp failed".to_string()),
}],
},
)
.expect("seed the ledger");
let super::Due::Refused(reason) = super::due(
&recorded("pod-owing", "2026-09-01T00:00:00Z"),
now,
&ledger_path,
) else {
panic!("an expired machine still holding work is refused, not released");
};
assert!(
reason.contains("/workspace/out"),
"the refusal names what is still on it: {reason}"
);
assert!(matches!(
super::due(&recorded("pod-unreadable", "whenever"), now, &ledger_path),
super::Due::Failed(_)
));
std::fs::remove_file(&ledger_path).ok();
}
#[test]
fn a_corrupt_ledger_fails_the_machine_rather_than_refusing_it() {
let ledger_path = scratch("sweep-corrupt-ledger");
std::fs::write(&ledger_path, "not a ledger row\n").expect("seed the corruption");
let now: jiff::Timestamp = "2026-09-02T00:00:00Z".parse().expect("a fixed clock");
let super::Due::Failed(reason) = super::due(
&recorded("pod-due", "2026-09-01T00:00:00Z"),
now,
&ledger_path,
) else {
panic!("an unreadable ledger is a failure the exit code must carry");
};
assert!(
reason.contains("could not read"),
"the reason names the gate's problem, not the machine's: {reason}"
);
std::fs::remove_file(&ledger_path).ok();
}
#[test]
fn only_a_machine_left_running_costs_the_sweep_its_zero_exit() {
let refused = super::SweepOutcome {
dry_run: false,
expired: 1,
released: Vec::new(),
refused: vec![("pod-owing".to_string(), "artifacts uncollected".to_string())],
failed: Vec::new(),
unknown: Vec::new(),
};
assert_eq!(super::sweep_exit(&refused), 0);
assert_eq!(
super::sweep_artifact(&refused)["refused"][0]["id"],
serde_json::json!("pod-owing"),
"the refusal is in the artifact even though the exit is zero"
);
let failed = super::SweepOutcome {
failed: vec![("pod-stuck".to_string(), "credential missing".to_string())],
..refused
};
assert_eq!(super::sweep_exit(&failed), 1);
let nothing = super::SweepOutcome::default();
assert_eq!(super::sweep_exit(¬hing), 0);
assert_eq!(
super::sweep_artifact(¬hing),
serde_json::json!({
"dry_run": false,
"expired": 0,
"released": [],
"refused": [],
"failed": [],
"unknown": [],
}),
"an empty sweep still emits the one artifact, with every field present"
);
}
#[test]
fn a_correction_row_carries_the_recorded_lease_and_retires_the_id() {
let path = scratch("correction");
super::record_acquisition(&path, &recorded("pod-1", "2026-09-02T00:00:00Z"))
.expect("seed an acquisition");
let correction = super::correction_row(
&path,
"pod-1",
"vast",
&["vastai".to_string(), "destroy".to_string()],
&"z".repeat(64),
);
assert_eq!(correction.provider, "runpod");
assert_eq!(correction.profile_hash, "h".repeat(64));
assert_eq!(correction.acquired_at, "2026-09-01T00:00:00Z");
assert_eq!(correction.expires_at, "2026-09-02T00:00:00Z");
assert!(correction.released_at.is_some());
super::record_acquisition(&path, &correction).expect("append the correction");
assert!(
record::outstanding(&path)
.expect("the record reads back")
.is_empty(),
"nothing is believed to be running once the correction lands"
);
assert_eq!(
record::list(&path).expect("the record reads back").len(),
2,
"the acquisition row is still there: corrections are new rows"
);
std::fs::remove_file(&path).ok();
}
#[test]
fn a_correction_for_an_unrecorded_machine_stands_on_what_the_release_knows() {
let path = scratch("correction-orphan");
let correction = super::correction_row(
&path,
"pod-elsewhere",
"vast",
&[
"vastai".to_string(),
"destroy".to_string(),
"{id}".to_string(),
],
&"z".repeat(64),
);
assert_eq!(correction.provider, "vast");
assert_eq!(correction.profile_hash, "z".repeat(64));
assert_eq!(
Some(&correction.acquired_at),
correction.released_at.as_ref(),
"the moment of release stands in for a lease nobody recorded"
);
assert_eq!(correction.acquired_at, correction.expires_at);
assert!(!path.exists(), "reading a missing record creates nothing");
}
#[test]
fn sweep_and_acquire_default_to_doing_nothing_and_to_a_recorded_lease() {
let cli = Cli::parse_from(["lm-provision", "machine", "sweep"]);
let Command::Machine {
command: MachineCommand::Sweep(args),
} = cli.command
else {
panic!("the parsed subcommand is `machine sweep`");
};
assert!(
args.dry_run,
"a sweep that was not asked to release, does not"
);
let cli = Cli::parse_from([
"lm-provision",
"machine",
"acquire",
"--profile",
"profile.json",
]);
let Command::Machine {
command: MachineCommand::Acquire(args),
} = cli.command
else {
panic!("the parsed subcommand is `machine acquire`");
};
assert!(args.dry_run);
assert_eq!(args.ttl_hours, 24, "a day, the fleet's ephemeral default");
}
#[test]
fn the_fleet_subcommands_live_under_machine_and_the_profile_ones_do_not() {
let cli = Cli::parse_from(["lm-provision", "machine", "list", "--provider", "runpod"]);
let Command::Machine {
command: MachineCommand::List(args),
} = cli.command
else {
panic!("the parsed subcommand is `machine list`");
};
assert_eq!(args.providers, vec!["runpod".to_string()]);
let cli = Cli::parse_from([
"lm-provision",
"machine",
"list",
"--provider",
"runpod",
"--provider",
"vast",
]);
let Command::Machine {
command: MachineCommand::List(args),
} = cli.command
else {
panic!("the parsed subcommand is `machine list`");
};
assert_eq!(
args.providers,
vec!["runpod".to_string(), "vast".to_string()],
"--provider is repeatable, as it is on sweep"
);
assert!(
Cli::try_parse_from(["lm-provision", "sweep"]).is_err(),
"the old top-level spelling is gone, not silently accepted"
);
assert!(matches!(
Cli::parse_from(["lm-provision", "mcp"]).command,
Command::Mcp
));
}
#[test]
fn machine_list_demands_a_platform_rather_than_reporting_an_empty_account() {
let Err(err) = Cli::try_parse_from(["lm-provision", "machine", "list"]) else {
panic!("a listing of nothing is not a listing");
};
assert_eq!(
err.kind(),
clap::error::ErrorKind::MissingRequiredArgument,
"{err}"
);
assert!(
Cli::try_parse_from(["lm-provision", "machine", "list", "--provider"]).is_err(),
"and the flag needs a value: --provider with nothing after it names no platform"
);
}
#[test]
fn a_listed_machine_is_judged_by_the_stamp_it_carries() {
use lm_provision_driver::infra::{self, Infra as _, RunPodAdapter, VastAdapter};
let now: jiff::Timestamp = "2026-09-02T00:00:00Z".parse().expect("a fixed clock");
let pods = RunPodAdapter.fleet().expect("this target can be asked");
let listed = serde_json::json!({
"pods": [
{ "id": "pod-over", "name": "lmp-exp-20260901T235959Z" },
{ "id": "pod-due", "name": "lmp-exp-20260902T000000Z" },
{ "id": "pod-live", "name": "lmp-exp-20260903T000000Z" },
{ "id": "pod-someone-elses", "name": "jupyter-scratch" },
{ "id": "pod-nameless" },
]
});
let judged: Vec<(String, super::Standing)> = infra::machines(&listed, &pods)
.expect("rows carrying the id key are the fleet")
.into_iter()
.map(|it| (it.id.clone(), super::standing(&it, now)))
.collect();
assert_eq!(
judged,
vec![
("pod-over".to_string(), super::Standing::Expired),
("pod-due".to_string(), super::Standing::Expired),
("pod-live".to_string(), super::Standing::Live),
("pod-someone-elses".to_string(), super::Standing::Unknown),
("pod-nameless".to_string(), super::Standing::Unknown),
]
);
let instances = VastAdapter.fleet().expect("this target can be asked");
let listed = serde_json::json!([
{ "id": 49227715, "label": "lmp-exp-20260901T120000Z" },
{ "id": 49228600, "label": "lmp-exp-20260930T120000Z" },
{ "id": 49229000, "label": null },
]);
let judged: Vec<(String, super::Standing)> = infra::machines(&listed, &instances)
.expect("a bare array is the rows")
.into_iter()
.map(|it| (it.id.clone(), super::standing(&it, now)))
.collect();
assert_eq!(
judged,
vec![
("49227715".to_string(), super::Standing::Expired),
("49228600".to_string(), super::Standing::Live),
("49229000".to_string(), super::Standing::Unknown),
]
);
}
#[test]
fn a_machine_both_halves_see_is_released_once() {
let handled: std::collections::BTreeSet<String> =
["pod-1".to_string()].into_iter().collect();
let listed: super::BTreeMap<String, std::collections::BTreeSet<String>> = [(
"runpod".to_string(),
["pod-1".to_string()].into_iter().collect(),
)]
.into_iter()
.collect();
assert_eq!(
super::bookkeeping(
&recorded("pod-1", "2026-09-02T00:00:00Z"),
&handled,
&listed
),
super::Bookkeeping::Handled
);
let unstamped_but_listed: super::BTreeMap<String, std::collections::BTreeSet<String>> = [(
"runpod".to_string(),
["pod-old".to_string()].into_iter().collect(),
)]
.into_iter()
.collect();
assert_eq!(
super::bookkeeping(
&recorded("pod-old", "2026-09-02T00:00:00Z"),
&std::collections::BTreeSet::new(),
&unstamped_but_listed,
),
super::Bookkeeping::Judge
);
assert_eq!(
super::bookkeeping(
&recorded("pod-2", "2026-09-02T00:00:00Z"),
&std::collections::BTreeSet::new(),
&super::BTreeMap::new(),
),
super::Bookkeeping::Judge
);
}
#[test]
fn an_outstanding_row_absent_from_its_platforms_list_is_retired() {
let listed: super::BTreeMap<String, std::collections::BTreeSet<String>> = [(
"runpod".to_string(),
["pod-other".to_string()].into_iter().collect(),
)]
.into_iter()
.collect();
let row = recorded("pod-gone", "2036-09-02T00:00:00Z");
assert_eq!(
super::bookkeeping(&row, &std::collections::BTreeSet::new(), &listed),
super::Bookkeeping::Gone,
"a lease with ten years left does not keep a machine that is not there"
);
let path = scratch("gone");
super::record_acquisition(&path, &row).expect("seed an acquisition");
super::retire(&row, &path);
assert!(
record::outstanding(&path)
.expect("the record reads back")
.is_empty(),
"nothing is believed to be running once the correction lands"
);
assert_eq!(
record::list(&path).expect("the record reads back").len(),
2,
"corrections are new rows here as everywhere else"
);
std::fs::remove_file(&path).ok();
}
#[test]
fn a_dry_run_shows_the_lease_the_machine_would_carry() {
use lm_provision_driver::infra::{self, Infra as _, RunPodAdapter, VastAdapter};
let expires_at: jiff::Timestamp = "2026-09-02T06:30:00Z".parse().expect("a fixed lease");
let stamp = infra::expiry_stamp(expires_at);
let ports = [("22".to_string(), "raw_tcp".to_string())]
.into_iter()
.collect();
let gpu = [("count".to_string(), "1".to_string())]
.into_iter()
.collect();
let required =
lm_provision::machine::Requirements::from_slots(&ports, &gpu, &super::BTreeMap::new())
.expect("well-formed fixture");
let pod_provider: super::BTreeMap<String, String> =
[("runpod.imageName".to_string(), "some/image:1".to_string())]
.into_iter()
.collect();
let artifact = super::dry_run_artifact(
&RunPodAdapter
.acquisition(&required, &pod_provider, Some(expires_at))
.expect("an image was declared"),
);
assert_eq!(artifact["dry_run"], serde_json::json!(true));
assert!(
artifact["body"]
.as_str()
.is_some_and(|it| it.contains(&stamp)),
"the request an operator is shown carries the lease: {artifact}"
);
let instance_provider: super::BTreeMap<String, String> =
[("vast.image".to_string(), "some/image:1".to_string())]
.into_iter()
.collect();
let artifact = super::dry_run_artifact(
&VastAdapter
.acquisition(&required, &instance_provider, Some(expires_at))
.expect("an image was declared"),
);
assert!(
artifact["create"]
.as_array()
.is_some_and(|argv| argv.iter().any(|it| it == &serde_json::json!(stamp))),
"and so does the argv on the target that takes one: {artifact}"
);
}
#[test]
fn the_record_defaults_beside_the_ledger() {
let acquisitions = super::default_acquisitions_path();
let ledger = super::default_ledger_path();
assert_eq!(acquisitions.parent(), ledger.parent());
assert!(acquisitions.file_name().is_some_and(
|it| it == "acquisitions.jsonl" || it == "lm-provision-acquisitions.jsonl"
));
}
}