zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: verification gate for N pipeline stage deployment

Browse source
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-07-25 11:42:29 +04:00
parent 1b8b5f9aff
commit f8beca8c5d
3 changed files with 162 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -4,6 +4,7 @@ fuzz/artifacts/**
corpus corpus
.loop/ .loop/
.model-cache/ .model-cache/
.deployment-notes/
# environment variables for deployment configurations # environment variables for deployment configurations
.config/ .config/

37
crates/mvp-system/src/orchestrator_app.rs
View file

@ -1405,11 +1405,6 @@ impl ConfigBuilder {
if self.pipeline_stages == 0 { if self.pipeline_stages == 0 {
return Err("--pipeline-stages must be greater than 0".to_owned()); return Err("--pipeline-stages must be greater than 0".to_owned());
} }
if provider == ProviderKind::VastAi && self.pipeline_stages > 2 {
return Err(
"provider=vastai currently supports at most 2 pipeline stages for activation-path smoke checks".to_owned(),
);
}
let mut cached_model_host_path = self.cached_model_host_path.clone(); let mut cached_model_host_path = self.cached_model_host_path.clone();
if matches!(provider, ProviderKind::Process | ProviderKind::Docker) if matches!(provider, ProviderKind::Process | ProviderKind::Docker)
&& self.pipeline_stages > 1 && self.pipeline_stages > 1
@ -6657,27 +6652,7 @@ kind = "docker"
} }
#[test] #[test]
fn vastai_rejects_pipeline_stages_count_above_two() { fn vastai_eight_stage_plan_uses_remote_gguf_and_no_mounts() {
let error = with_clean_env(&[], || {
match Config::from_layers_with_path_and_args(
None,
["--provider", "vastai", "--pipeline-stages", "3"]
.into_iter()
.map(str::to_owned),
) {
Ok(_) => panic!("VastAI smoke runs are capped at two pipeline stages"),
Err(error) => error,
}
});
assert!(
error.contains("provider=vastai currently supports at most 2 pipeline stages"),
"unexpected error: {error}"
);
}
#[test]
fn vastai_two_stage_plan_uses_remote_gguf_and_no_mounts() {
let config = with_clean_env(&[], || { let config = with_clean_env(&[], || {
Config::from_layers_with_path_and_args( Config::from_layers_with_path_and_args(
None, None,
@ -6685,7 +6660,7 @@ kind = "docker"
"--provider", "--provider",
"vastai", "vastai",
"--pipeline-stages", "--pipeline-stages",
"2", "8",
"--model-id", "--model-id",
"smollm2-135m-instruct-q4", "smollm2-135m-instruct-q4",
"--gguf-repo", "--gguf-repo",
@ -6702,11 +6677,11 @@ kind = "docker"
.into_iter() .into_iter()
.map(str::to_owned), .map(str::to_owned),
) )
.expect("VastAI two-stage pipeline config parses") .expect("VastAI eight-stage pipeline config parses")
}); });
let plan = config let plan = config
.build_run_plan() .build_run_plan()
.expect("VastAI two-stage run plan uses local metadata only"); .expect("VastAI eight-stage run plan uses local metadata only");
let coordinator = EndpointAddr::new(iroh::SecretKey::from_bytes(&[41; 32]).public()); let coordinator = EndpointAddr::new(iroh::SecretKey::from_bytes(&[41; 32]).public());
let orchestrator_actor = ActorAddress([42; 32]); let orchestrator_actor = ActorAddress([42; 32]);
@ -6718,7 +6693,7 @@ kind = "docker"
config.cached_model.is_none(), config.cached_model.is_none(),
"VastAI must not mount host caches" "VastAI must not mount host caches"
); );
assert_eq!(specs.len(), 2); assert_eq!(specs.len(), 8);
for (expected_stage_index, spec) in specs.iter().enumerate() { for (expected_stage_index, spec) in specs.iter().enumerate() {
let expected_stage_index = let expected_stage_index =
u32::try_from(expected_stage_index).expect("fixture stage index fits u32"); u32::try_from(expected_stage_index).expect("fixture stage index fits u32");
@ -6726,7 +6701,7 @@ kind = "docker"
assert_eq!(spec.node_id, expected_node_id); assert_eq!(spec.node_id, expected_node_id);
assert_eq!(spec.stage_index, Some(expected_stage_index)); assert_eq!(spec.stage_index, Some(expected_stage_index));
assert_eq!(env_value(&spec.env, "MVP_NODE_PROVIDER"), Some("vastai")); assert_eq!(env_value(&spec.env, "MVP_NODE_PROVIDER"), Some("vastai"));
assert_eq!(env_value(&spec.env, "MVP_PIPELINE_STAGES"), Some("2")); assert_eq!(env_value(&spec.env, "MVP_PIPELINE_STAGES"), Some("8"));
assert_eq!( assert_eq!(
env_value(&spec.env, "MVP_GGUF_REPO"), env_value(&spec.env, "MVP_GGUF_REPO"),
Some("QuantFactory/SmolLM2-135M-Instruct-GGUF") Some("QuantFactory/SmolLM2-135M-Instruct-GGUF")

201
xtask/src/main.rs
View file

@ -52,61 +52,93 @@ enum MvpChatCheckScenario {
VastAi, VastAi,
} }
impl MvpChatCheckScenario { #[derive(Clone, Debug, PartialEq, Eq)]
struct MvpChatCheckInvocation {
scenario: MvpChatCheckScenario,
pipeline_stages: Option<u32>,
}
impl MvpChatCheckInvocation {
fn parse_args(args: Vec<String>) -> Result<Self, String> { fn parse_args(args: Vec<String>) -> Result<Self, String> {
let mut scenario = Self::ProcessBaseline; let mut scenario = MvpChatCheckScenario::ProcessBaseline;
for arg in args { let mut pipeline_stages = None;
let selected = match arg.as_str() { let mut args = args.into_iter();
"--gpu" => Self::Gpu, while let Some(arg) = args.next() {
"--multinode" => Self::Multinode, if let Some(selected) = MvpChatCheckScenario::from_flag(&arg) {
"--multinode-docker" => Self::MultinodeDocker, if scenario != MvpChatCheckScenario::ProcessBaseline {
"--vastai" => Self::VastAi,
other => return Err(format!("unsupported mvp-chat-check argument {other:?}")),
};
if scenario != Self::ProcessBaseline {
return Err( return Err(
"mvp-chat-check accepts at most one scenario flag: --gpu, --multinode, --multinode-docker, or --vastai" "mvp-chat-check accepts at most one scenario flag: --gpu, --multinode, --multinode-docker, or --vastai"
.to_owned(), .to_owned(),
); );
} }
scenario = selected; scenario = selected;
} continue;
Ok(scenario)
} }
fn name(self) -> &'static str { match arg.as_str() {
match self { "--pipeline-stages" => {
Self::ProcessBaseline => "process", if pipeline_stages.is_some() {
Self::Gpu => "gpu", return Err(
Self::Multinode => "multinode", "mvp-chat-check accepts at most one --pipeline-stages value".to_owned()
Self::MultinodeDocker => "multinode-docker", );
Self::VastAi => "vastai",
} }
let value = args
.next()
.ok_or_else(|| "--pipeline-stages requires a value".to_owned())?;
let stages = value
.parse::<u32>()
.map_err(|error| format!("parse --pipeline-stages: {error}"))?;
if stages == 0 {
return Err("--pipeline-stages must be greater than 0".to_owned());
}
pipeline_stages = Some(stages);
}
other => return Err(format!("unsupported mvp-chat-check argument {other:?}")),
}
}
Ok(Self {
scenario,
pipeline_stages,
})
} }
fn mvp_chat_args(self, run_id: u64, dump_log: &Path) -> Vec<String> { fn scenario(&self) -> MvpChatCheckScenario {
self.scenario
}
fn name(&self) -> &'static str {
self.scenario.name()
}
fn mvp_chat_args(&self, run_id: u64, dump_log: &Path) -> Vec<String> {
let mut args = Vec::new(); let mut args = Vec::new();
match self { match self.scenario {
Self::ProcessBaseline | Self::Multinode => { MvpChatCheckScenario::ProcessBaseline | MvpChatCheckScenario::Multinode => {
args.push("--process".to_owned()); args.push("--process".to_owned());
} }
Self::Gpu => { MvpChatCheckScenario::Gpu => {
args.extend(["--process".to_owned(), "--gpu".to_owned()]); args.extend(["--process".to_owned(), "--gpu".to_owned()]);
} }
Self::MultinodeDocker => { MvpChatCheckScenario::MultinodeDocker => {
args.push("--docker".to_owned()); args.push("--docker".to_owned());
} }
Self::VastAi => { MvpChatCheckScenario::VastAi => {
args.push("--vastai".to_owned()); args.push("--vastai".to_owned());
} }
} }
if matches!(self, Self::Multinode | Self::MultinodeDocker) { if let Some(pipeline_stages) = self
args.extend(["--pipeline-stages".to_owned(), "2".to_owned()]); .pipeline_stages
.or_else(|| self.scenario.default_pipeline_stages())
{
args.extend(["--pipeline-stages".to_owned(), pipeline_stages.to_string()]);
} }
if !matches!(self, Self::Gpu | Self::VastAi) { if !matches!(
self.scenario,
MvpChatCheckScenario::Gpu | MvpChatCheckScenario::VastAi
) {
args.push("--cached-model".to_owned()); args.push("--cached-model".to_owned());
} }
if matches!(self, Self::VastAi) { if matches!(self.scenario, MvpChatCheckScenario::VastAi) {
args.extend([ args.extend([
"--yes".to_owned(), "--yes".to_owned(),
"--endpoint-addr-mask".to_owned(), "--endpoint-addr-mask".to_owned(),
@ -122,6 +154,39 @@ impl MvpChatCheckScenario {
args args
} }
fn env_overrides(&self) -> &'static [(&'static str, &'static str)] {
self.scenario.env_overrides()
}
}
impl MvpChatCheckScenario {
fn from_flag(flag: &str) -> Option<Self> {
match flag {
"--gpu" => Some(Self::Gpu),
"--multinode" => Some(Self::Multinode),
"--multinode-docker" => Some(Self::MultinodeDocker),
"--vastai" => Some(Self::VastAi),
_ => None,
}
}
fn name(self) -> &'static str {
match self {
Self::ProcessBaseline => "process",
Self::Gpu => "gpu",
Self::Multinode => "multinode",
Self::MultinodeDocker => "multinode-docker",
Self::VastAi => "vastai",
}
}
fn default_pipeline_stages(self) -> Option<u32> {
match self {
Self::Multinode | Self::MultinodeDocker => Some(2),
Self::ProcessBaseline | Self::Gpu | Self::VastAi => None,
}
}
fn env_overrides(self) -> &'static [(&'static str, &'static str)] { fn env_overrides(self) -> &'static [(&'static str, &'static str)] {
match self { match self {
Self::ProcessBaseline Self::ProcessBaseline
@ -189,7 +254,7 @@ USAGE: cargo xtask <command>
COMMANDS: COMMANDS:
mvp-chat [--gpu] [--process|--docker|--vastai] [--pipeline-stages n] [--cached-model] [-- args...] Run the human chat wrapper against the real orchestrator/worker bins. mvp-chat [--gpu] [--process|--docker|--vastai] [--pipeline-stages n] [--cached-model] [-- args...] Run the human chat wrapper against the real orchestrator/worker bins.
mvp-chat-check [--gpu|--multinode|--multinode-docker|--vastai] mvp-chat-check [--gpu|--multinode|--multinode-docker|--vastai] [--pipeline-stages n]
Run real cargo mvp-chat acceptance check and write benchmark artifacts. Run real cargo mvp-chat acceptance check and write benchmark artifacts.
mvp-chat-compare <baseline-summary.json> <candidate-summary.json> mvp-chat-compare <baseline-summary.json> <candidate-summary.json>
Compare two benchmark summaries and report comparable deltas. Compare two benchmark summaries and report comparable deltas.
@ -657,14 +722,15 @@ fn write_mvp_chat_check_paths(root: &Path) -> Result<MvpChatCheckPaths, String>
} }
fn run_mvp_chat_check(args: Vec<String>) -> ExitCode { fn run_mvp_chat_check(args: Vec<String>) -> ExitCode {
let scenario = match MvpChatCheckScenario::parse_args(args) { let invocation = match MvpChatCheckInvocation::parse_args(args) {
Ok(scenario) => scenario, Ok(invocation) => invocation,
Err(error) => { Err(error) => {
eprintln!("mvp-chat-check: failed: {error}"); eprintln!("mvp-chat-check: failed: {error}");
print_usage(); print_usage();
return ExitCode::from(1); return ExitCode::from(1);
} }
}; };
let scenario = invocation.scenario();
let workspace = workspace_root(); let workspace = workspace_root();
let temp_root = unique_temp_dir("mvp-chat-check"); let temp_root = unique_temp_dir("mvp-chat-check");
let paths = match write_mvp_chat_check_paths(&temp_root) { let paths = match write_mvp_chat_check_paths(&temp_root) {
@ -679,9 +745,9 @@ fn run_mvp_chat_check(args: Vec<String>) -> ExitCode {
} }
}; };
let run_id = mvp_chat_check_run_id(); let run_id = mvp_chat_check_run_id();
println!("mvp-chat-check: scenario {}", scenario.name()); println!("mvp-chat-check: scenario {}", invocation.name());
let output = match run_mvp_chat_check_process(&workspace, &paths, run_id, scenario) { let output = match run_mvp_chat_check_process(&workspace, &paths, run_id, &invocation) {
Ok(output) => output, Ok(output) => output,
Err(error) => return fail_mvp_chat_check(&error, &paths, "", "", None), Err(error) => return fail_mvp_chat_check(&error, &paths, "", "", None),
}; };
@ -790,14 +856,14 @@ fn run_mvp_chat_check_process(
workspace: &Path, workspace: &Path,
paths: &MvpChatCheckPaths, paths: &MvpChatCheckPaths,
run_id: u64, run_id: u64,
scenario: MvpChatCheckScenario, invocation: &MvpChatCheckInvocation,
) -> Result<MvpChatCheckOutput, String> { ) -> Result<MvpChatCheckOutput, String> {
let mut command = Command::new(cargo_bin()); let mut command = Command::new(cargo_bin());
command.current_dir(workspace).arg("mvp-chat").arg("--"); command.current_dir(workspace).arg("mvp-chat").arg("--");
for arg in scenario.mvp_chat_args(run_id, &paths.dump_log) { for arg in invocation.mvp_chat_args(run_id, &paths.dump_log) {
command.arg(arg); command.arg(arg);
} }
for &(key, value) in scenario.env_overrides() { for &(key, value) in invocation.env_overrides() {
command.env(key, value); command.env(key, value);
} }
command command
@ -2955,8 +3021,8 @@ mod tests {
let dump_log = Path::new("/tmp/mvp-chat-check.ndjson"); let dump_log = Path::new("/tmp/mvp-chat-check.ndjson");
let baseline = let baseline =
MvpChatCheckScenario::parse_args(Vec::new()).expect("default scenario parses"); MvpChatCheckInvocation::parse_args(Vec::new()).expect("default scenario parses");
assert_eq!(baseline, MvpChatCheckScenario::ProcessBaseline); assert_eq!(baseline.scenario(), MvpChatCheckScenario::ProcessBaseline);
assert_eq!( assert_eq!(
baseline.mvp_chat_args(42, dump_log), baseline.mvp_chat_args(42, dump_log),
strings(&[ strings(&[
@ -2968,8 +3034,8 @@ mod tests {
]) ])
); );
let gpu = MvpChatCheckScenario::parse_args(strings(&["--gpu"])).expect("gpu parses"); let gpu = MvpChatCheckInvocation::parse_args(strings(&["--gpu"])).expect("gpu parses");
assert_eq!(gpu, MvpChatCheckScenario::Gpu); assert_eq!(gpu.scenario(), MvpChatCheckScenario::Gpu);
assert!(gpu.env_overrides().is_empty()); assert!(gpu.env_overrides().is_empty());
assert_eq!( assert_eq!(
gpu.mvp_chat_args(42, dump_log), gpu.mvp_chat_args(42, dump_log),
@ -2982,9 +3048,9 @@ mod tests {
]) ])
); );
let multinode = let multinode = MvpChatCheckInvocation::parse_args(strings(&["--multinode"]))
MvpChatCheckScenario::parse_args(strings(&["--multinode"])).expect("multinode parses"); .expect("multinode parses");
assert_eq!(multinode, MvpChatCheckScenario::Multinode); assert_eq!(multinode.scenario(), MvpChatCheckScenario::Multinode);
assert_eq!( assert_eq!(
multinode.mvp_chat_args(42, dump_log), multinode.mvp_chat_args(42, dump_log),
strings(&[ strings(&[
@ -2998,9 +3064,12 @@ mod tests {
]) ])
); );
let multinode_docker = MvpChatCheckScenario::parse_args(strings(&["--multinode-docker"])) let multinode_docker = MvpChatCheckInvocation::parse_args(strings(&["--multinode-docker"]))
.expect("multinode docker parses"); .expect("multinode docker parses");
assert_eq!(multinode_docker, MvpChatCheckScenario::MultinodeDocker); assert_eq!(
multinode_docker.scenario(),
MvpChatCheckScenario::MultinodeDocker
);
assert_eq!( assert_eq!(
multinode_docker.mvp_chat_args(42, dump_log), multinode_docker.mvp_chat_args(42, dump_log),
strings(&[ strings(&[
@ -3015,8 +3084,8 @@ mod tests {
); );
let vastai = let vastai =
MvpChatCheckScenario::parse_args(strings(&["--vastai"])).expect("vastai parses"); MvpChatCheckInvocation::parse_args(strings(&["--vastai"])).expect("vastai parses");
assert_eq!(vastai, MvpChatCheckScenario::VastAi); assert_eq!(vastai.scenario(), MvpChatCheckScenario::VastAi);
assert_eq!( assert_eq!(
vastai.mvp_chat_args(42, dump_log), vastai.mvp_chat_args(42, dump_log),
strings(&[ strings(&[
@ -3030,20 +3099,50 @@ mod tests {
"--dump-logs=/tmp/mvp-chat-check.ndjson", "--dump-logs=/tmp/mvp-chat-check.ndjson",
]) ])
); );
let vastai_sweep =
MvpChatCheckInvocation::parse_args(strings(&["--vastai", "--pipeline-stages", "8"]))
.expect("vastai sweep parses");
assert_eq!(vastai_sweep.scenario(), MvpChatCheckScenario::VastAi);
assert_eq!(
vastai_sweep.mvp_chat_args(42, dump_log),
strings(&[
"--vastai",
"--pipeline-stages",
"8",
"--yes",
"--endpoint-addr-mask",
"relay-only",
"--skip-rebuild",
"--run-id",
"42",
"--dump-logs=/tmp/mvp-chat-check.ndjson",
])
);
} }
#[test] #[test]
fn scenario_flags_reject_unknown_or_ambiguous_invocations() { fn scenario_flags_reject_unknown_or_ambiguous_invocations() {
assert!( assert!(
MvpChatCheckScenario::parse_args(strings(&["--docker"])) MvpChatCheckInvocation::parse_args(strings(&["--docker"]))
.expect_err("unknown flag fails") .expect_err("unknown flag fails")
.contains("unsupported mvp-chat-check argument") .contains("unsupported mvp-chat-check argument")
); );
assert!( assert!(
MvpChatCheckScenario::parse_args(strings(&["--gpu", "--multinode"])) MvpChatCheckInvocation::parse_args(strings(&["--gpu", "--multinode"]))
.expect_err("multiple scenarios fail") .expect_err("multiple scenarios fail")
.contains("at most one scenario flag") .contains("at most one scenario flag")
); );
assert!(
MvpChatCheckInvocation::parse_args(strings(&["--vastai", "--pipeline-stages", "0"]))
.expect_err("zero pipeline stages fail")
.contains("--pipeline-stages must be greater than 0")
);
assert!(
MvpChatCheckInvocation::parse_args(strings(&["--vastai", "--pipeline-stages"]))
.expect_err("missing pipeline stages fail")
.contains("--pipeline-stages requires a value")
);
} }
fn temp_path(label: &str) -> PathBuf { fn temp_path(label: &str) -> PathBuf {