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

Compare commits

base: zacheryasc:1b8b5f9aff5cdf2f0ea219f22ce1343dac618cb3
Branches Tags
zacheryasc:master
zacheryasc:development
zacheryasc:backup/pre-rewrite-msg
..
compare: zacheryasc:18f06e34f9c16ee60ea012a8f6aba4ca85977340
Branches Tags
zacheryasc:development
zacheryasc:master
zacheryasc:backup/pre-rewrite-msg

No commits in common. "1b8b5f9aff5cdf2f0ea219f22ce1343dac618cb3" and "18f06e34f9c16ee60ea012a8f6aba4ca85977340" have entirely different histories.
1b8b5f9aff ... 18f06e34f9

15 changed files with 49 additions and 1625 deletions
Show stats Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -5736,7 +5736,6 @@ dependencies = [
name = "xtask" name = "xtask"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"blake3",
"libc", "libc",
"serde_json", "serde_json",
] ]

69
apps/mvp-node/tinygrad_worker.py
View file

@ -751,7 +751,6 @@ def decode_greedy_device_resident(
request_id: int | None, request_id: int | None,
model_id: str | None, model_id: str | None,
progress_every: int, progress_every: int,
decode_started_at: float,
) -> list[int]: ) -> list[int]:
if max_tokens <= 0: if max_tokens <= 0:
return [] return []
@ -798,7 +797,6 @@ def decode_greedy_device_resident(
model_id=model_id, model_id=model_id,
token_index=1, token_index=1,
prompt_tokens=len(prompt_tokens), prompt_tokens=len(prompt_tokens),
first_token_elapsed_ms=int((time.monotonic() - decode_started_at) * 1000),
) )
elif progress_every > 0 and tokens_generated % progress_every == 0: elif progress_every > 0 and tokens_generated % progress_every == 0:
control( control(
@ -838,7 +836,6 @@ def infer_prompt(cmd: dict[str, Any]) -> None:
prompt_chars=len(prompt), prompt_chars=len(prompt),
max_tokens=max_tokens, max_tokens=max_tokens,
) )
encode_started = time.monotonic()
model_prompt, prompt_template = model_prompt_text(prompt) model_prompt, prompt_template = model_prompt_text(prompt)
control( control(
type="PromptEncodeStarted", type="PromptEncodeStarted",
@ -855,10 +852,8 @@ def infer_prompt(cmd: dict[str, Any]) -> None:
model_prompt_bytes=len(model_prompt.encode("utf-8")), model_prompt_bytes=len(model_prompt.encode("utf-8")),
prompt_template=prompt_template, prompt_template=prompt_template,
prompt_tokens=len(prompt_tokens), prompt_tokens=len(prompt_tokens),
elapsed_ms=int((time.monotonic() - encode_started) * 1000),
) )
progress_every = int(os.environ.get("MVP_TOKEN_PROGRESS_EVERY", "16") or "16") progress_every = int(os.environ.get("MVP_TOKEN_PROGRESS_EVERY", "16") or "16")
decode_started = time.monotonic()
control( control(
type="DecodeStarted", type="DecodeStarted",
request_id=request_id, request_id=request_id,
@ -879,7 +874,6 @@ def infer_prompt(cmd: dict[str, Any]) -> None:
request_id=request_id, request_id=request_id,
model_id=loaded.get("model_id"), model_id=loaded.get("model_id"),
progress_every=progress_every, progress_every=progress_every,
decode_started_at=decode_started,
) )
finally: finally:
stop_cpu_line_sampler(cpu_sampler) stop_cpu_line_sampler(cpu_sampler)
@ -889,9 +883,7 @@ def infer_prompt(cmd: dict[str, Any]) -> None:
model_id=loaded.get("model_id"), model_id=loaded.get("model_id"),
prompt_tokens=len(prompt_tokens), prompt_tokens=len(prompt_tokens),
tokens_generated=len(generated), tokens_generated=len(generated),
elapsed_ms=int((time.monotonic() - decode_started) * 1000),
) )
text_decode_started = time.monotonic()
control(type="TextDecodeStarted", request_id=request_id, model_id=loaded.get("model_id"), tokens_generated=len(generated)) control(type="TextDecodeStarted", request_id=request_id, model_id=loaded.get("model_id"), tokens_generated=len(generated))
raw_text = tokenizer.decode(generated) if generated else "" raw_text = tokenizer.decode(generated) if generated else ""
text = strip_chat_stop_markers(raw_text) text = strip_chat_stop_markers(raw_text)
@ -901,7 +893,6 @@ def infer_prompt(cmd: dict[str, Any]) -> None:
model_id=loaded.get("model_id"), model_id=loaded.get("model_id"),
tokens_generated=len(generated), tokens_generated=len(generated),
text_bytes=len(text.encode("utf-8")), text_bytes=len(text.encode("utf-8")),
elapsed_ms=int((time.monotonic() - text_decode_started) * 1000),
) )
control( control(
type="PromptCompleted", type="PromptCompleted",
@ -939,11 +930,7 @@ def install_ring(cmd: dict[str, Any]) -> None:
type="RingInstalled", type="RingInstalled",
ring_id=ring_id, ring_id=ring_id,
edge_id=rings[ring_id]["edge_id"], edge_id=rings[ring_id]["edge_id"],
port=rings[ring_id]["port"],
direction=rings[ring_id]["direction"], direction=rings[ring_id]["direction"],
data_capacity=rings[ring_id]["data_capacity"],
max_extent=rings[ring_id]["max_extent"],
alignment=rings[ring_id]["alignment"],
) )
@ -1014,7 +1001,6 @@ def materialize_object(payload: bytes, sequence: int, flags: int) -> dict[str, A
return { return {
"kind": "tokens", "kind": "tokens",
"tokens": tokens, "tokens": tokens,
"token_count": token_count,
"tensor": TensorCls([tokens], dtype="int32").realize(), "tensor": TensorCls([tokens], dtype="int32").realize(),
"start_pos": object_start_pos(sequence, token_count, flags), "start_pos": object_start_pos(sequence, token_count, flags),
} }
@ -1025,7 +1011,6 @@ def materialize_object(payload: bytes, sequence: int, flags: int) -> dict[str, A
return { return {
"kind": "tokens", "kind": "tokens",
"tokens": tokens, "tokens": tokens,
"token_count": token_count,
"tensor": TensorCls([tokens], dtype="int32").realize(), "tensor": TensorCls([tokens], dtype="int32").realize(),
"start_pos": object_start_pos(sequence, token_count, flags), "start_pos": object_start_pos(sequence, token_count, flags),
} }
@ -1041,20 +1026,18 @@ def materialize_object(payload: bytes, sequence: int, flags: int) -> dict[str, A
array = np.frombuffer(payload, dtype=np.float16).copy().reshape(1, token_count, hidden_dim) array = np.frombuffer(payload, dtype=np.float16).copy().reshape(1, token_count, hidden_dim)
return { return {
"kind": "activation", "kind": "activation",
"token_count": token_count,
"hidden_dim": hidden_dim,
"tensor": TensorCls(array).realize(), "tensor": TensorCls(array).realize(),
"start_pos": object_start_pos(sequence, token_count, flags), "start_pos": object_start_pos(sequence, token_count, flags),
} }
def ring_readable(cmd: dict[str, Any]) -> None: def ring_readable(cmd: dict[str, Any]) -> None:
global next_handle global next_handle
ring_id = int(cmd["ring_id"]) ring_id = int(cmd["ring_id"])
ring = rings[ring_id] ring = rings[ring_id]
if ring["direction"] != "ingress": if ring["direction"] != "ingress":
fatal("WrongRingDirection", ring_id=ring_id, direction=ring["direction"]) fatal("WrongRingDirection", ring_id=ring_id, direction=ring["direction"])
started = time.monotonic()
object_id, sequence, extent, flags, payload = parse_record(ring) object_id, sequence, extent, flags, payload = parse_record(ring)
handle = next_handle handle = next_handle
next_handle += 1 next_handle += 1
@ -1064,6 +1047,7 @@ def ring_readable(cmd: dict[str, Any]) -> None:
sequence=sequence, sequence=sequence,
extent=extent, extent=extent,
flags=flags, flags=flags,
payload=payload,
) )
device_objects[handle] = materialized device_objects[handle] = materialized
control( control(
@ -1075,11 +1059,6 @@ def ring_readable(cmd: dict[str, Any]) -> None:
extent=extent, extent=extent,
handle_generation=WORKER_GENERATION, handle_generation=WORKER_GENERATION,
handle_id=handle, handle_id=handle,
kind=materialized.get("kind"),
token_count=materialized.get("token_count"),
hidden_dim=materialized.get("hidden_dim"),
start_pos=materialized.get("start_pos"),
elapsed_ms=int((time.monotonic() - started) * 1000),
) )
@ -1113,7 +1092,6 @@ def write_record(ring: dict[str, Any], object_id: int, sequence: int, payload: b
def execute_step(cmd: dict[str, Any]) -> None: def execute_step(cmd: dict[str, Any]) -> None:
if not role: if not role:
fatal("RoleNotConfigured") fatal("RoleNotConfigured")
step_started = time.monotonic()
handle = int(cmd["input_handle_id"]) handle = int(cmd["input_handle_id"])
obj = device_objects.get(handle) obj = device_objects.get(handle)
if obj is None: if obj is None:
@ -1125,41 +1103,35 @@ def execute_step(cmd: dict[str, Any]) -> None:
if ring["direction"] != "egress": if ring["direction"] != "egress":
fatal("WrongRingDirection", ring_id=output_ring_id, direction=ring["direction"]) fatal("WrongRingDirection", ring_id=output_ring_id, direction=ring["direction"])
final_stage = bool(cmd.get("final_stage")) final_stage = bool(cmd.get("final_stage"))
input_kind = obj.get("kind")
input_extent = int(obj.get("extent", 0))
execution_backend = "pipeline_stage" execution_backend = "pipeline_stage"
if not isinstance(model, PipelineStageTinygradModel): if not isinstance(model, PipelineStageTinygradModel):
execution_backend = "full_transformer" execution_backend = "full_transformer"
if not final_stage: if not final_stage:
fatal("FullTransformerNonFinalStageUnsupported", step_id=int(cmd["step_id"])) fatal("FullTransformerNonFinalStageUnsupported", step_id=int(cmd["step_id"]))
if input_kind != "tokens": if obj.get("kind") != "tokens":
fatal("FullTransformerInputUnsupported", step_id=int(cmd["step_id"]), kind=input_kind) fatal("FullTransformerInputUnsupported", step_id=int(cmd["step_id"]), kind=obj.get("kind"))
if int(obj.get("flags", 0)) & FLAG_BEGIN_SEQUENCE and hasattr(model, "forward_jit"): if int(obj.get("flags", 0)) & FLAG_BEGIN_SEQUENCE and hasattr(model, "forward_jit"):
model.forward_jit.reset() model.forward_jit.reset()
token_array = model(obj["tensor"], int(obj.get("start_pos", 0))).realize().numpy().reshape(-1) token_array = model(obj["tensor"], int(obj.get("start_pos", 0))).realize().numpy().reshape(-1)
token = int(token_array[0]) token = int(token_array[0])
payload = struct.pack("<I", token) payload = struct.pack("<I", token)
output_kind = "token"
flags = 1 if token == int(loaded.get("eos_token_id", 0)) else 0 flags = 1 if token == int(loaded.get("eos_token_id", 0)) else 0
else: else:
if final_stage != bool(getattr(model, "final_stage", False)): if final_stage != bool(getattr(model, "final_stage", False)):
fatal("FinalStageMismatch", command_final_stage=final_stage, model_final_stage=bool(getattr(model, "final_stage", False))) fatal("FinalStageMismatch", command_final_stage=final_stage, model_final_stage=bool(getattr(model, "final_stage", False)))
input_tensor = model.token_hidden(obj["tensor"]) if input_kind == "tokens" else obj["tensor"] input_tensor = model.token_hidden(obj["tensor"]) if obj.get("kind") == "tokens" else obj["tensor"]
hidden = model.forward_hidden(input_tensor, int(obj.get("start_pos", 0))) hidden = model.forward_hidden(input_tensor, int(obj.get("start_pos", 0)))
if final_stage: if final_stage:
token_array = model.next_token(hidden).realize().numpy().reshape(-1) token_array = model.next_token(hidden).realize().numpy().reshape(-1)
token = int(token_array[0]) token = int(token_array[0])
payload = struct.pack("<I", token) payload = struct.pack("<I", token)
output_kind = "token"
flags = 1 if token == int(loaded.get("eos_token_id", 0)) else 0 flags = 1 if token == int(loaded.get("eos_token_id", 0)) else 0
else: else:
import numpy as np import numpy as np
activation = hidden.realize().numpy().astype(np.float16, copy=False) activation = hidden.realize().numpy().astype(np.float16, copy=False)
payload = activation.tobytes() payload = activation.tobytes()
output_kind = "activation"
flags = 0 flags = 0
compute_ready = time.monotonic()
committed = write_record( committed = write_record(
ring, ring,
int(cmd["output_object_id"]), int(cmd["output_object_id"]),
@ -1167,7 +1139,6 @@ def execute_step(cmd: dict[str, Any]) -> None:
payload, payload,
flags, flags,
) )
write_ready = time.monotonic()
control( control(
type="StepExecuted", type="StepExecuted",
step_id=int(cmd["step_id"]), step_id=int(cmd["step_id"]),
@ -1176,15 +1147,6 @@ def execute_step(cmd: dict[str, Any]) -> None:
sequence=int(cmd["output_sequence"]), sequence=int(cmd["output_sequence"]),
committed_bytes=committed, committed_bytes=committed,
execution_backend=execution_backend, execution_backend=execution_backend,
final_stage=final_stage,
input_kind=input_kind,
input_extent=input_extent,
output_kind=output_kind,
payload_bytes=len(payload),
record_bytes=committed,
stage_execution_ms=int((compute_ready - step_started) * 1000),
record_write_ms=int((write_ready - compute_ready) * 1000),
elapsed_ms=int((write_ready - step_started) * 1000),
) )
@ -1195,39 +1157,22 @@ def release_device_object(cmd: dict[str, Any]) -> None:
def encode_prompt(cmd: dict[str, Any]) -> None: def encode_prompt(cmd: dict[str, Any]) -> None:
started = time.monotonic()
prompt = str(cmd.get("prompt", "")) prompt = str(cmd.get("prompt", ""))
if tokenizer is not None: if tokenizer is not None:
model_prompt, _ = model_prompt_text(prompt) model_prompt, _ = model_prompt_text(prompt)
tokens = [int(token) for token in tokenizer.encode(model_prompt)] tokens = [int(token) for token in tokenizer.encode(model_prompt)]
else: else:
model_prompt = prompt
tokens = [int(byte) for byte in prompt.encode("utf-8")] or [0] tokens = [int(byte) for byte in prompt.encode("utf-8")] or [0]
control( control(type="PromptEncoded", request_id=cmd.get("request_id"), tokens=tokens)
type="PromptEncoded",
request_id=cmd.get("request_id"),
tokens=tokens,
prompt_bytes=len(prompt.encode("utf-8")),
model_prompt_bytes=len(model_prompt.encode("utf-8")),
elapsed_ms=int((time.monotonic() - started) * 1000),
)
def decode_tokens(cmd: dict[str, Any]) -> None: def decode_tokens(cmd: dict[str, Any]) -> None:
started = time.monotonic()
tokens = [int(token) for token in cmd.get("tokens", [])] tokens = [int(token) for token in cmd.get("tokens", [])]
if tokenizer is not None: if tokenizer is not None:
text = strip_chat_stop_markers(tokenizer.decode(tokens)) text = strip_chat_stop_markers(tokenizer.decode(tokens))
else: else:
text = "".join(chr(token) if 32 <= token <= 126 else f"<tok:{token}>" for token in tokens) text = "".join(chr(token) if 32 <= token <= 126 else f"<tok:{token}>" for token in tokens)
control( control(type="TokensDecoded", request_id=cmd.get("request_id"), text=text)
type="TokensDecoded",
request_id=cmd.get("request_id"),
text=text,
tokens=len(tokens),
text_bytes=len(text.encode("utf-8")),
elapsed_ms=int((time.monotonic() - started) * 1000),
)
def shutdown_worker(_: dict[str, Any]) -> None: def shutdown_worker(_: dict[str, Any]) -> None:

167
crates/mvp-system/src/bin/mvp_chat.rs
View file

@ -61,7 +61,6 @@ const CHAT_LIFECYCLE_CHANNEL: &str = "mvp.chat.lifecycle";
const CHAT_RUNTIME_CHANNEL: &str = "mvp.chat.runtime"; const CHAT_RUNTIME_CHANNEL: &str = "mvp.chat.runtime";
const CHAT_PROMPT_CHANNEL: &str = "mvp.chat.prompt"; const CHAT_PROMPT_CHANNEL: &str = "mvp.chat.prompt";
const CHAT_COMPONENT_CHANNEL: &str = "mvp.chat.component"; const CHAT_COMPONENT_CHANNEL: &str = "mvp.chat.component";
const CHAT_BENCHMARK_CHANNEL: &str = "mvp.chat.benchmark";
#[derive(Debug)] #[derive(Debug)]
enum PromptInput { enum PromptInput {
@ -152,7 +151,6 @@ where
"gpu_run": config.gpu_run, "gpu_run": config.gpu_run,
}), }),
); );
progress.emit_benchmark_envelope(&config);
confirm_vastai_if_needed(&config)?; confirm_vastai_if_needed(&config)?;
progress.emit( progress.emit(
CHAT_RUNTIME_CHANNEL, CHAT_RUNTIME_CHANNEL,
@ -333,7 +331,6 @@ struct Config {
run_id: u64, run_id: u64,
vastai_yes: bool, vastai_yes: bool,
vastai: Option<ResolvedVastAiConfig>, vastai: Option<ResolvedVastAiConfig>,
model: ChatModelConfig,
pipeline_stages: u32, pipeline_stages: u32,
max_tokens: u32, max_tokens: u32,
skip_rebuild: bool, skip_rebuild: bool,
@ -382,7 +379,6 @@ impl ChatDatastream {
CHAT_RUNTIME_CHANNEL, CHAT_RUNTIME_CHANNEL,
CHAT_PROMPT_CHANNEL, CHAT_PROMPT_CHANNEL,
CHAT_COMPONENT_CHANNEL, CHAT_COMPONENT_CHANNEL,
CHAT_BENCHMARK_CHANNEL,
] { ] {
out.channel_by_name(name); out.channel_by_name(name);
} }
@ -419,68 +415,6 @@ impl ChatDatastream {
self.flush(); self.flush();
} }
fn emit_benchmark_envelope(&mut self, config: &Config) {
let id = self.channel_by_name(CHAT_BENCHMARK_CHANNEL);
let payload = serde_json::to_vec(&json!({
"type": "BenchmarkRunEnvelope",
"phase": "run_envelope",
"status": "ready",
"run_id": self.run_id,
"benchmark": benchmark_observability::stamp("mvp-chat"),
"detail": {
"scenario": "mvp-chat",
"detail_level": "benchmark_observability_v1",
"workload": {
"mode": "stdin_prompt_corpus",
"max_tokens": config.max_tokens,
"prompt_corpus": "external_or_stdin",
},
"model": {
"id": config.model.id.as_deref(),
"gguf_local_path": config.model.gguf_local_path.as_deref(),
"gguf_repo": config.model.gguf_repo.as_deref(),
"gguf_file": config.model.gguf_file.as_deref(),
"gguf_revision": config.model.gguf_revision.as_deref(),
"tokenizer_local_path": config.model.tokenizer_local_path.as_deref(),
"max_context": config.model.max_context,
},
"runtime": {
"provider": config.provider.as_str(),
"pipeline_stages": config.pipeline_stages,
"orchestrator_launch_mode": config.orchestrator_launch_mode(),
"gpu_run": config.gpu_run,
"relay_mode": config.relay_mode.as_deref(),
"relay_configured": config.relay_url.is_some(),
"endpoint_addr_mask": config.endpoint_addr_mask.as_str(),
},
"provider": {
"kind": config.provider.as_str(),
"node_image": &config.node_image,
"image_tag": config.image_tag.as_deref(),
"cached_model": config.cached_model.as_ref().map(|model| model.host_path.to_string_lossy().to_string()),
"vastai": config.vastai.as_ref().map(|vastai| json!({
"image": &vastai.image,
"relay_configured": !vastai.relay_url.is_empty(),
"bootstrap_command_configured": !vastai.bootstrap_command.is_empty(),
"gpu_name": vastai.gpu_name.as_deref(),
"min_gpu_ram_mb": vastai.min_gpu_ram_mb,
"min_down_mbps": vastai.min_down_mbps,
"min_up_mbps": vastai.min_up_mbps,
"max_dph_total": vastai.max_dph_total,
"min_reliability": vastai.min_reliability,
"require_verified": vastai.require_verified,
"disk_gb": vastai.disk_gb,
"has_onstart": vastai.onstart.is_some(),
"has_ssh_identity": vastai.ssh_identity.is_some(),
})),
},
},
}))
.expect("serialize mvp-chat benchmark envelope");
self.producer.submit_bytes(id, payload);
self.flush();
}
fn flush(&mut self) { fn flush(&mut self) {
let stream = self.stream.clone(); let stream = self.stream.clone();
for frame in self.endpoint.mux().drain() { for frame in self.endpoint.mux().drain() {
@ -587,7 +521,6 @@ struct ChatTomlConfig {
observability: ChatObservabilityConfig, observability: ChatObservabilityConfig,
image: ChatImageConfig, image: ChatImageConfig,
vastai: ChatVastAiConfig, vastai: ChatVastAiConfig,
model: ChatModelConfig,
relay: ChatRelayConfig, relay: ChatRelayConfig,
} }
@ -635,7 +568,6 @@ struct ChatVastAiConfig {
min_gpu_ram_mb: Option<u64>, min_gpu_ram_mb: Option<u64>,
min_down_mbps: Option<f64>, min_down_mbps: Option<f64>,
min_up_mbps: Option<f64>, min_up_mbps: Option<f64>,
max_dph_total: Option<f64>,
min_reliability: Option<f64>, min_reliability: Option<f64>,
require_verified: Option<bool>, require_verified: Option<bool>,
disk_gb: Option<u32>, disk_gb: Option<u32>,
@ -643,18 +575,6 @@ struct ChatVastAiConfig {
ssh_identity: Option<String>, ssh_identity: Option<String>,
} }
#[derive(Clone, Debug, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
struct ChatModelConfig {
id: Option<String>,
gguf_local_path: Option<String>,
gguf_repo: Option<String>,
gguf_file: Option<String>,
gguf_revision: Option<String>,
tokenizer_local_path: Option<String>,
max_context: Option<u32>,
}
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
struct LoadedChatTomlConfig { struct LoadedChatTomlConfig {
overlay: ChatTomlConfig, overlay: ChatTomlConfig,
@ -767,7 +687,6 @@ impl Config {
vastai_yes: args.vastai_yes, vastai_yes: args.vastai_yes,
pipeline_stages, pipeline_stages,
max_tokens, max_tokens,
model: toml.model,
vastai, vastai,
skip_rebuild: args.skip_rebuild, skip_rebuild: args.skip_rebuild,
gpu_run, gpu_run,
@ -795,27 +714,6 @@ impl Config {
self.pipeline_stages.to_string(), self.pipeline_stages.to_string(),
"--no-dashboard".to_owned(), "--no-dashboard".to_owned(),
]; ];
if let Some(model_id) = &self.model.id {
args.extend(["--model-id".to_owned(), model_id.clone()]);
}
if let Some(path) = &self.model.gguf_local_path {
args.extend(["--gguf-local-path".to_owned(), path.clone()]);
}
if let Some(repo) = &self.model.gguf_repo {
args.extend(["--gguf-repo".to_owned(), repo.clone()]);
}
if let Some(file) = &self.model.gguf_file {
args.extend(["--gguf-file".to_owned(), file.clone()]);
}
if let Some(revision) = &self.model.gguf_revision {
args.extend(["--gguf-revision".to_owned(), revision.clone()]);
}
if let Some(path) = &self.model.tokenizer_local_path {
args.extend(["--tokenizer-local-path".to_owned(), path.clone()]);
}
if let Some(max_context) = self.model.max_context {
args.extend(["--max-context".to_owned(), max_context.to_string()]);
}
if self.provider == ProviderKind::Process { if self.provider == ProviderKind::Process {
args.extend([ args.extend([
"--worker-bin".to_owned(), "--worker-bin".to_owned(),
@ -875,12 +773,6 @@ impl Config {
if let Some(min_up_mbps) = vastai.min_up_mbps { if let Some(min_up_mbps) = vastai.min_up_mbps {
args.extend(["--vastai-min-up-mbps".to_owned(), min_up_mbps.to_string()]); args.extend(["--vastai-min-up-mbps".to_owned(), min_up_mbps.to_string()]);
} }
if let Some(max_dph_total) = vastai.max_dph_total {
args.extend([
"--vastai-max-dph-total".to_owned(),
max_dph_total.to_string(),
]);
}
if let Some(min_reliability) = vastai.min_reliability { if let Some(min_reliability) = vastai.min_reliability {
args.extend([ args.extend([
"--vastai-min-reliability".to_owned(), "--vastai-min-reliability".to_owned(),
@ -1027,7 +919,6 @@ fn resolve_vastai_config(
min_gpu_ram_mb: file.min_gpu_ram_mb, min_gpu_ram_mb: file.min_gpu_ram_mb,
min_down_mbps: file.min_down_mbps, min_down_mbps: file.min_down_mbps,
min_up_mbps: file.min_up_mbps, min_up_mbps: file.min_up_mbps,
max_dph_total: file.max_dph_total,
min_reliability: file.min_reliability, min_reliability: file.min_reliability,
require_verified: file.require_verified, require_verified: file.require_verified,
onstart: first_non_empty([file.onstart.clone()]), onstart: first_non_empty([file.onstart.clone()]),
@ -1471,23 +1362,6 @@ fn prepare_runtime_with_progress(
} }
if config.skip_rebuild { if config.skip_rebuild {
if config.provider == ProviderKind::VastAi {
emit_chat_progress(
&mut progress,
CHAT_RUNTIME_CHANNEL,
"ensure_worker_binary",
"skipped",
json!({"mode": binary_mode, "reason": "vastai_remote_image"}),
);
emit_chat_progress(
&mut progress,
CHAT_RUNTIME_CHANNEL,
"prepare_node_image",
"skipped",
json!({"provider": config.provider.as_str(), "reason": "skip_rebuild"}),
);
return Ok(config.node_image.clone());
}
emit_chat_progress( emit_chat_progress(
&mut progress, &mut progress,
CHAT_RUNTIME_CHANNEL, CHAT_RUNTIME_CHANNEL,
@ -1727,10 +1601,6 @@ fn emit_chat_progress(
} }
} }
fn prompt_hash_hex(prompt: &str) -> String {
blake3::hash(prompt.as_bytes()).to_hex().to_string()
}
fn run_chat_session_with_output_and_progress<R, W, O>( fn run_chat_session_with_output_and_progress<R, W, O>(
writer: &mut W, writer: &mut W,
mut reader: R, mut reader: R,
@ -1746,7 +1616,6 @@ where
{ {
let mut progress = progress; let mut progress = progress;
let mut next_request_id = 1_u64; let mut next_request_id = 1_u64;
let mut next_prompt_index = 1_u64;
loop { loop {
if STOP_REQUESTED.load(Ordering::SeqCst) { if STOP_REQUESTED.load(Ordering::SeqCst) {
@ -1764,7 +1633,7 @@ where
CHAT_PROMPT_CHANNEL, CHAT_PROMPT_CHANNEL,
"waiting_for_prompt", "waiting_for_prompt",
"started", "started",
json!({"next_request_id": next_request_id, "next_prompt_index": next_prompt_index}), json!({"next_request_id": next_request_id}),
); );
write!(output, "prompt:> ").map_err(|e| format!("write prompt: {e}"))?; write!(output, "prompt:> ").map_err(|e| format!("write prompt: {e}"))?;
output.flush().map_err(|e| format!("flush prompt: {e}"))?; output.flush().map_err(|e| format!("flush prompt: {e}"))?;
@ -1797,15 +1666,12 @@ where
let request_id = next_request_id; let request_id = next_request_id;
next_request_id = next_request_id.wrapping_add(1).max(1); next_request_id = next_request_id.wrapping_add(1).max(1);
let prompt_index = next_prompt_index;
next_prompt_index = next_prompt_index.wrapping_add(1).max(1);
let prompt_hash = prompt_hash_hex(&prompt);
emit_chat_progress( emit_chat_progress(
&mut progress, &mut progress,
CHAT_PROMPT_CHANNEL, CHAT_PROMPT_CHANNEL,
"prompt_submitted", "prompt_submitted",
"ready", "ready",
json!({"request_id": request_id, "prompt_index": prompt_index, "prompt_hash": &prompt_hash, "prompt_bytes": prompt.len(), "max_tokens": max_tokens}), json!({"request_id": request_id, "prompt_bytes": prompt.len(), "max_tokens": max_tokens}),
); );
write_json_line( write_json_line(
writer, writer,
@ -1821,7 +1687,7 @@ where
CHAT_PROMPT_CHANNEL, CHAT_PROMPT_CHANNEL,
"decoding", "decoding",
"started", "started",
json!({"request_id": request_id, "prompt_index": prompt_index, "prompt_hash": &prompt_hash}), json!({"request_id": request_id}),
); );
let mut response_started = false; let mut response_started = false;
@ -1902,7 +1768,7 @@ where
CHAT_PROMPT_CHANNEL, CHAT_PROMPT_CHANNEL,
"response_text", "response_text",
"observed", "observed",
json!({"request_id": request_id, "prompt_index": prompt_index, "prompt_hash": &prompt_hash, "text_bytes": text.len()}), json!({"request_id": request_id, "text_bytes": text.len()}),
); );
} }
PromptEvent::Done { PromptEvent::Done {
@ -1924,8 +1790,6 @@ where
"ready", "ready",
json!({ json!({
"request_id": request_id, "request_id": request_id,
"prompt_index": prompt_index,
"prompt_hash": &prompt_hash,
"response_started": response_started, "response_started": response_started,
"tokens_generated": tokens_generated, "tokens_generated": tokens_generated,
"elapsed_ms": elapsed_ms, "elapsed_ms": elapsed_ms,
@ -1942,7 +1806,7 @@ where
CHAT_PROMPT_CHANNEL, CHAT_PROMPT_CHANNEL,
"request_faulted", "request_faulted",
"ready", "ready",
json!({"request_id": request_id, "prompt_index": prompt_index, "prompt_hash": &prompt_hash, "error": error}), json!({"request_id": request_id, "error": error}),
); );
break; break;
} }
@ -2321,7 +2185,6 @@ mod tests {
run_id: 1, run_id: 1,
vastai_yes: false, vastai_yes: false,
vastai: None, vastai: None,
model: ChatModelConfig::default(),
pipeline_stages: 1, pipeline_stages: 1,
max_tokens: DEFAULT_MAX_TOKENS, max_tokens: DEFAULT_MAX_TOKENS,
skip_rebuild: true, skip_rebuild: true,
@ -2343,7 +2206,6 @@ mod tests {
min_gpu_ram_mb: None, min_gpu_ram_mb: None,
min_down_mbps: None, min_down_mbps: None,
min_up_mbps: None, min_up_mbps: None,
max_dph_total: None,
min_reliability: None, min_reliability: None,
require_verified: None, require_verified: None,
onstart: None, onstart: None,
@ -3072,25 +2934,6 @@ relay_url = "https://relay.example"
assert_eq!(image_ref, "docker.io/acme/node:latest"); assert_eq!(image_ref, "docker.io/acme/node:latest");
} }
#[test]
fn vastai_skip_rebuild_uses_remote_image_without_worker_artifact() {
let temp = TempDir::new("vastai-skip-rebuild");
let orch_bin = temp.path().join("mvp-orchestrator");
let worker_bin = temp.path().join("mvp-worker-node");
let mut config = base_config(ProviderKind::VastAi);
config.skip_rebuild = true;
config.orch_bin = orch_bin.clone();
config.worker_bin = worker_bin;
config.node_image = "docker.io/acme/node:latest".to_owned();
assert!(prepare_runtime_with(&config, panic_prepare_node_image).is_err());
fs::write(&orch_bin, b"orch").expect("write orchestrator artifact");
let image_ref = prepare_runtime_with(&config, panic_prepare_node_image)
.expect("VastAI skip rebuild reuses remote image");
assert_eq!(image_ref, "docker.io/acme/node:latest");
}
#[test] #[test]
fn artifact_roots_use_current_directory() { fn artifact_roots_use_current_directory() {
let temp = TempDir::new("artifact-root"); let temp = TempDir::new("artifact-root");

59
crates/mvp-system/src/bin/worker_node.rs
View file

@ -592,14 +592,6 @@ impl WorkerEdgeRuntime {
edge_id: driver_model::EdgeId(edge_id), edge_id: driver_model::EdgeId(edge_id),
stream_id: driver_model::StreamId(stream_id), stream_id: driver_model::StreamId(stream_id),
}); });
emit_node_event(
datastream,
config,
NODE_STAGE_CHANNEL,
"iroh_edge_stream_arrived",
"observed",
json!({"edge_id":edge_id,"stream_id":stream_id}),
);
self.drive_edge_workflow( self.drive_edge_workflow(
stack, stack,
node_actor, node_actor,
@ -616,15 +608,6 @@ impl WorkerEdgeRuntime {
bytes, bytes,
.. ..
} => { } => {
let byte_count = bytes.len();
emit_node_event(
datastream,
config,
NODE_STAGE_CHANNEL,
"iroh_edge_bytes_read",
"observed",
json!({"edge_id":edge_id,"stream_id":stream_id,"bytes":byte_count}),
);
self.ingest_stream_bytes( self.ingest_stream_bytes(
edge_id, edge_id,
stream_id, stream_id,
@ -809,22 +792,6 @@ impl WorkerEdgeRuntime {
.read_arena(lease.layout.data_offset, committed_bytes) .read_arena(lease.layout.data_offset, committed_bytes)
.map_err(|e| format!("read egress ring: {e}"))? .map_err(|e| format!("read egress ring: {e}"))?
}; };
let record_bytes = record.len();
emit_node_event(
datastream,
config,
NODE_STAGE_CHANNEL,
"egress_ring_read",
"ready",
json!({
"edge_id":outbound.edge_id,
"edge_kind":format!("{:?}", outbound.kind),
"ring_id":output_ring_id,
"record_bytes":record_bytes,
"committed_bytes":committed_bytes,
"final_stage":final_stage,
}),
);
self.driver_model self.driver_model
.observe(driver_model::DriverEvent::EgressBytesCommitted { .observe(driver_model::DriverEvent::EgressBytesCommitted {
edge_id: driver_model::EdgeId(outbound.edge_id), edge_id: driver_model::EdgeId(outbound.edge_id),
@ -839,19 +806,6 @@ impl WorkerEdgeRuntime {
.as_ref() .as_ref()
.ok_or_else(|| "outbound edge sender missing".to_owned())?; .ok_or_else(|| "outbound edge sender missing".to_owned())?;
sender.send(record)?; sender.send(record)?;
emit_node_event(
datastream,
config,
NODE_STAGE_CHANNEL,
"iroh_edge_bytes_sent",
"ready",
json!({
"edge_id":outbound.edge_id,
"edge_kind":format!("{:?}", outbound.kind),
"bytes":record_bytes,
"record_bytes":record_bytes,
}),
);
stack stack
.runtime .runtime
.send_to(node_actor, NodeAgentMsg::StepCompleted { step_id }) .send_to(node_actor, NodeAgentMsg::StepCompleted { step_id })
@ -913,19 +867,6 @@ impl WorkerEdgeRuntime {
.write_arena(lease.layout.data_offset, &record) .write_arena(lease.layout.data_offset, &record)
.map_err(|e| format!("write ingress ring: {e}"))?; .map_err(|e| format!("write ingress ring: {e}"))?;
} }
emit_node_event(
datastream,
config,
NODE_STAGE_CHANNEL,
"ingress_ring_write",
"ready",
json!({
"edge_id":edge_id,
"edge_kind":format!("{:?}", inbound.kind),
"ring_id":ring_id,
"record_bytes":record.len(),
}),
);
let loaded = worker.ring_readable( let loaded = worker.ring_readable(
ring_id, ring_id,
edge_id, edge_id,

8
crates/mvp-system/src/config.rs
View file

@ -106,7 +106,6 @@ pub struct VastAiConfig {
pub min_gpu_ram_mb: Option<u64>, pub min_gpu_ram_mb: Option<u64>,
pub min_down_mbps: Option<f64>, pub min_down_mbps: Option<f64>,
pub min_up_mbps: Option<f64>, pub min_up_mbps: Option<f64>,
pub max_dph_total: Option<f64>,
pub min_reliability: Option<f64>, pub min_reliability: Option<f64>,
pub require_verified: Option<bool>, pub require_verified: Option<bool>,
pub poll_interval_secs: Option<u64>, pub poll_interval_secs: Option<u64>,
@ -131,7 +130,6 @@ pub struct ResolvedVastAiConfig {
pub min_gpu_ram_mb: Option<u64>, pub min_gpu_ram_mb: Option<u64>,
pub min_down_mbps: Option<f64>, pub min_down_mbps: Option<f64>,
pub min_up_mbps: Option<f64>, pub min_up_mbps: Option<f64>,
pub max_dph_total: Option<f64>,
pub min_reliability: Option<f64>, pub min_reliability: Option<f64>,
pub require_verified: Option<bool>, pub require_verified: Option<bool>,
pub onstart: Option<String>, pub onstart: Option<String>,
@ -206,9 +204,6 @@ impl ResolvedVastAiConfig {
if let Some(min_up_mbps) = self.min_up_mbps { if let Some(min_up_mbps) = self.min_up_mbps {
policy.min_up_mbps = Some(min_up_mbps); policy.min_up_mbps = Some(min_up_mbps);
} }
if let Some(max_dph_total) = self.max_dph_total {
policy.max_dph_total = Some(max_dph_total);
}
if let Some(min_reliability) = self.min_reliability { if let Some(min_reliability) = self.min_reliability {
policy.min_reliability = min_reliability; policy.min_reliability = min_reliability;
} }
@ -264,7 +259,6 @@ mod tests {
min_gpu_ram_mb: Some(16_000), min_gpu_ram_mb: Some(16_000),
min_down_mbps: Some(100.0), min_down_mbps: Some(100.0),
min_up_mbps: Some(25.0), min_up_mbps: Some(25.0),
max_dph_total: Some(0.10),
min_reliability: Some(0.98), min_reliability: Some(0.98),
require_verified: Some(true), require_verified: Some(true),
onstart: None, onstart: None,
@ -332,7 +326,6 @@ gpu_name = "RTX 4090"
min_gpu_ram_mb = 24000 min_gpu_ram_mb = 24000
min_down_mbps = 250.5 min_down_mbps = 250.5
min_up_mbps = 50.25 min_up_mbps = 50.25
max_dph_total = 0.10
min_reliability = 0.99 min_reliability = 0.99
require_verified = true require_verified = true
onstart = "echo preparing" onstart = "echo preparing"
@ -409,7 +402,6 @@ poll_interval_secs = 30
assert_eq!(config.vastai.min_gpu_ram_mb, Some(24_000)); assert_eq!(config.vastai.min_gpu_ram_mb, Some(24_000));
assert_eq!(config.vastai.min_down_mbps, Some(250.5)); assert_eq!(config.vastai.min_down_mbps, Some(250.5));
assert_eq!(config.vastai.min_up_mbps, Some(50.25)); assert_eq!(config.vastai.min_up_mbps, Some(50.25));
assert_eq!(config.vastai.max_dph_total, Some(0.10));
assert_eq!(config.vastai.min_reliability, Some(0.99)); assert_eq!(config.vastai.min_reliability, Some(0.99));
assert_eq!(config.vastai.require_verified, Some(true)); assert_eq!(config.vastai.require_verified, Some(true));
assert_eq!(config.vastai.poll_interval_secs, Some(30)); assert_eq!(config.vastai.poll_interval_secs, Some(30));

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

@ -636,15 +636,6 @@ impl VastAiRuntimeConfig {
if let Some(min_down_mbps) = min_down_mbps { if let Some(min_down_mbps) = min_down_mbps {
provisioning.selection.min_down_mbps = min_down_mbps; provisioning.selection.min_down_mbps = min_down_mbps;
} }
let max_dph_total = builder
.vastai_max_dph_total_raw
.as_ref()
.map(|value| ConfigBuilder::parse_value("MVP_VASTAI_MAX_DPH_TOTAL", value))
.transpose()?
.or(builder.vastai_max_dph_total);
if let Some(max_dph_total) = max_dph_total {
provisioning.selection.max_dph_total = Some(max_dph_total);
}
let min_up_mbps = builder let min_up_mbps = builder
.vastai_min_up_mbps_raw .vastai_min_up_mbps_raw
.as_ref() .as_ref()
@ -704,7 +695,6 @@ impl VastAiRuntimeConfig {
"min_gpu_ram_mb": self.provisioning.selection.min_gpu_ram_mb, "min_gpu_ram_mb": self.provisioning.selection.min_gpu_ram_mb,
"min_down_mbps": self.provisioning.selection.min_down_mbps, "min_down_mbps": self.provisioning.selection.min_down_mbps,
"min_up_mbps": self.provisioning.selection.min_up_mbps, "min_up_mbps": self.provisioning.selection.min_up_mbps,
"max_dph_total": self.provisioning.selection.max_dph_total,
"min_reliability": self.provisioning.selection.min_reliability, "min_reliability": self.provisioning.selection.min_reliability,
"require_verified": self.provisioning.selection.require_verified, "require_verified": self.provisioning.selection.require_verified,
"confirm_lease": self.provisioning.confirm_lease, "confirm_lease": self.provisioning.confirm_lease,
@ -920,8 +910,6 @@ struct ConfigBuilder {
vastai_min_down_mbps: Option<f64>, vastai_min_down_mbps: Option<f64>,
vastai_min_down_mbps_raw: Option<String>, vastai_min_down_mbps_raw: Option<String>,
vastai_min_up_mbps: Option<f64>, vastai_min_up_mbps: Option<f64>,
vastai_max_dph_total: Option<f64>,
vastai_max_dph_total_raw: Option<String>,
vastai_min_up_mbps_raw: Option<String>, vastai_min_up_mbps_raw: Option<String>,
vastai_min_reliability: Option<f64>, vastai_min_reliability: Option<f64>,
vastai_min_reliability_raw: Option<String>, vastai_min_reliability_raw: Option<String>,
@ -977,8 +965,6 @@ impl ConfigBuilder {
vastai_min_gpu_ram_mb_raw: None, vastai_min_gpu_ram_mb_raw: None,
vastai_min_down_mbps: None, vastai_min_down_mbps: None,
vastai_min_down_mbps_raw: None, vastai_min_down_mbps_raw: None,
vastai_max_dph_total: None,
vastai_max_dph_total_raw: None,
vastai_min_up_mbps: None, vastai_min_up_mbps: None,
vastai_min_up_mbps_raw: None, vastai_min_up_mbps_raw: None,
vastai_min_reliability: None, vastai_min_reliability: None,
@ -1097,9 +1083,6 @@ impl ConfigBuilder {
if let Some(min_down_mbps) = overlay.vastai.min_down_mbps { if let Some(min_down_mbps) = overlay.vastai.min_down_mbps {
self.vastai_min_down_mbps = Some(min_down_mbps); self.vastai_min_down_mbps = Some(min_down_mbps);
} }
if let Some(max_dph_total) = overlay.vastai.max_dph_total {
self.vastai_max_dph_total = Some(max_dph_total);
}
if let Some(min_up_mbps) = overlay.vastai.min_up_mbps { if let Some(min_up_mbps) = overlay.vastai.min_up_mbps {
self.vastai_min_up_mbps = Some(min_up_mbps); self.vastai_min_up_mbps = Some(min_up_mbps);
} }
@ -1232,9 +1215,6 @@ impl ConfigBuilder {
if let Some(min_down_mbps) = env_optional("MVP_VASTAI_MIN_DOWN_MBPS") { if let Some(min_down_mbps) = env_optional("MVP_VASTAI_MIN_DOWN_MBPS") {
self.vastai_min_down_mbps_raw = Some(min_down_mbps); self.vastai_min_down_mbps_raw = Some(min_down_mbps);
} }
if let Some(max_dph_total) = env_optional("MVP_VASTAI_MAX_DPH_TOTAL") {
self.vastai_max_dph_total_raw = Some(max_dph_total);
}
if let Some(min_up_mbps) = env_optional("MVP_VASTAI_MIN_UP_MBPS") { if let Some(min_up_mbps) = env_optional("MVP_VASTAI_MIN_UP_MBPS") {
self.vastai_min_up_mbps_raw = Some(min_up_mbps); self.vastai_min_up_mbps_raw = Some(min_up_mbps);
} }
@ -1359,11 +1339,6 @@ impl ConfigBuilder {
Some(parse_next(&mut args, "--vastai-min-down-mbps")?); Some(parse_next(&mut args, "--vastai-min-down-mbps")?);
self.vastai_min_down_mbps_raw = None; self.vastai_min_down_mbps_raw = None;
} }
"--vastai-max-dph-total" => {
self.vastai_max_dph_total =
Some(parse_next(&mut args, "--vastai-max-dph-total")?);
self.vastai_max_dph_total_raw = None;
}
"--vastai-min-up-mbps" => { "--vastai-min-up-mbps" => {
self.vastai_min_up_mbps = Some(parse_next(&mut args, "--vastai-min-up-mbps")?); self.vastai_min_up_mbps = Some(parse_next(&mut args, "--vastai-min-up-mbps")?);
self.vastai_min_up_mbps_raw = None; self.vastai_min_up_mbps_raw = None;
@ -1405,9 +1380,9 @@ 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 { if provider == ProviderKind::VastAi && self.pipeline_stages > 1 {
return Err( return Err(
"provider=vastai currently supports at most 2 pipeline stages for activation-path smoke checks".to_owned(), "pipeline stages greater than 1 are only supported with provider=docker; provider=vastai does not support pipelined provisioning yet".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();
@ -1673,20 +1648,6 @@ impl Config {
)) ))
} }
} }
GgufSource::HuggingFaceGguf { repo, file, .. }
if self.provider == ProviderKind::VastAi
&& gguf_source_matches_default_pipeline_cache(&self.gguf_source) =>
{
let host_path = default_pipeline_cached_model_path();
if host_path.is_file() {
Ok(host_path)
} else {
Err(format!(
"VastAI pipeline planning requires local GGUF metadata at {}; selected remote source is {repo}/{file}",
host_path.display()
))
}
}
GgufSource::HuggingFaceGguf { repo, file, .. } => Err(format!( GgufSource::HuggingFaceGguf { repo, file, .. } => Err(format!(
"local pipeline requires a locally inspectable GGUF before provisioning; selected source {repo}/{file} is remote, so use --cached-model-host-path" "local pipeline requires a locally inspectable GGUF before provisioning; selected source {repo}/{file} is remote, so use --cached-model-host-path"
)), )),
@ -6657,91 +6618,25 @@ kind = "docker"
} }
#[test] #[test]
fn vastai_rejects_pipeline_stages_count_above_two() { fn vastai_rejects_pipeline_stages_count_above_one() {
let error = with_clean_env(&[], || { let error = with_clean_env(&[], || {
match Config::from_layers_with_path_and_args( match Config::from_layers_with_path_and_args(
None, None,
["--provider", "vastai", "--pipeline-stages", "3"] ["--provider", "vastai", "--pipeline-stages", "2"]
.into_iter() .into_iter()
.map(str::to_owned), .map(str::to_owned),
) { ) {
Ok(_) => panic!("VastAI smoke runs are capped at two pipeline stages"), Ok(_) => panic!("VastAI cannot provision more than one pipeline stage yet"),
Err(error) => error, Err(error) => error,
} }
}); });
assert!( assert!(
error.contains("provider=vastai currently supports at most 2 pipeline stages"), error.contains("provider=vastai does not support pipelined provisioning yet"),
"unexpected error: {error}" "unexpected error: {error}"
); );
} }
#[test]
fn vastai_two_stage_plan_uses_remote_gguf_and_no_mounts() {
let config = with_clean_env(&[], || {
Config::from_layers_with_path_and_args(
None,
[
"--provider",
"vastai",
"--pipeline-stages",
"2",
"--model-id",
"smollm2-135m-instruct-q4",
"--gguf-repo",
"QuantFactory/SmolLM2-135M-Instruct-GGUF",
"--gguf-file",
DEFAULT_PIPELINE_CACHED_MODEL_FILE,
"--max-context",
"256",
"--relay-mode",
"default",
"--vastai-bootstrap-command",
"boot",
]
.into_iter()
.map(str::to_owned),
)
.expect("VastAI two-stage pipeline config parses")
});
let plan = config
.build_run_plan()
.expect("VastAI two-stage run plan uses local metadata only");
let coordinator = EndpointAddr::new(iroh::SecretKey::from_bytes(&[41; 32]).public());
let orchestrator_actor = ActorAddress([42; 32]);
let specs = stage_node_specs(&config, Some(&plan), coordinator, orchestrator_actor)
.expect("VastAI pipeline stage node specs build");
assert_eq!(config.provider, ProviderKind::VastAi);
assert!(
config.cached_model.is_none(),
"VastAI must not mount host caches"
);
assert_eq!(specs.len(), 2);
for (expected_stage_index, spec) in specs.iter().enumerate() {
let expected_stage_index =
u32::try_from(expected_stage_index).expect("fixture stage index fits u32");
let expected_node_id = config.node_id + 1 + u64::from(expected_stage_index);
assert_eq!(spec.node_id, expected_node_id);
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_PIPELINE_STAGES"), Some("2"));
assert_eq!(
env_value(&spec.env, "MVP_GGUF_REPO"),
Some("QuantFactory/SmolLM2-135M-Instruct-GGUF")
);
assert_eq!(
env_value(&spec.env, "MVP_GGUF_FILE"),
Some(DEFAULT_PIPELINE_CACHED_MODEL_FILE)
);
assert_eq!(env_value(&spec.env, "MVP_GGUF_LOCAL_PATH"), None);
assert_eq!(env_value(&spec.env, "MVP_MAX_CONTEXT"), Some("256"));
assert_eq!(spec.args, vec!["boot".to_owned()]);
assert!(spec.mounts.is_empty(), "VastAI stage specs must not mount");
}
}
#[test] #[test]
fn pipeline_cached_model_resolution_uses_toml_smollm2_gguf_file_with_cli_stage_count() { fn pipeline_cached_model_resolution_uses_toml_smollm2_gguf_file_with_cli_stage_count() {
let toml = TempTomlFile::new( let toml = TempTomlFile::new(

148
crates/mvp-system/src/vastai_provisioning.rs
View file

@ -797,24 +797,6 @@ where
instance.contract_id instance.contract_id
), ),
}); });
sink.observe(PluginObservation::ProviderLine {
run_id: spec.run_id,
node_id: spec.node_id,
line: serde_json::json!({
"type": "VastAiLeaseReady",
"run_id": spec.run_id,
"node_id": spec.node_id,
"label": &label,
"image": &spec.image,
"contract_id": instance.contract_id,
"offer_id": instance.offer_id,
"host_id": instance.host_id,
"gpu_name": &instance.gpu_name,
"gpu_ram": instance.gpu_ram,
"dph_total": instance.dph_total,
})
.to_string(),
});
let endpoint = match self.client.ssh_endpoint( let endpoint = match self.client.ssh_endpoint(
instance.contract_id, instance.contract_id,
@ -830,20 +812,6 @@ where
)); ));
} }
}; };
sink.observe(PluginObservation::ProviderLine {
run_id: spec.run_id,
node_id: spec.node_id,
line: serde_json::json!({
"type": "VastAiSshEndpointReady",
"run_id": spec.run_id,
"node_id": spec.node_id,
"contract_id": instance.contract_id,
"host": &endpoint.host,
"port": endpoint.port,
"user": &endpoint.user,
})
.to_string(),
});
let bootstrap = match self.bootstrap.start_bootstrap( let bootstrap = match self.bootstrap.start_bootstrap(
spec.clone(), spec.clone(),
@ -875,7 +843,14 @@ where
Ok(handle) Ok(handle)
} }
fn complete_bootstrap(&mut self, _handle: &PluginNodeHandle) -> Result<(), String> { fn complete_bootstrap(&mut self, handle: &PluginNodeHandle) -> Result<(), String> {
let Some(node) = self.nodes.get_mut(&handle.id) else {
return Ok(());
};
if let Some(mut bootstrap) = node.bootstrap.take() {
self.bootstrap
.stop_bootstrap(&mut bootstrap, BootstrapStopReason::RuntimeReady);
}
Ok(()) Ok(())
} }
@ -962,81 +937,6 @@ mod tests {
VastAiProvisioningPlugin::new(NoopLeaseClient, NoopBootstrapLauncher, config) VastAiProvisioningPlugin::new(NoopLeaseClient, NoopBootstrapLauncher, config)
} }
#[derive(Clone, Default)]
struct ObservationSink {
observations: Arc<Mutex<Vec<PluginObservation>>>,
}
impl crate::provisioning::PluginObservationSink for ObservationSink {
fn observe(&self, observation: PluginObservation) {
self.observations.lock().push(observation);
}
}
#[derive(Clone)]
struct RecordingLeaseClient {
destroyed_contracts: Arc<Mutex<Vec<u64>>>,
}
impl VastAiLeaseClient for RecordingLeaseClient {
fn provision_one(
&mut self,
_request: ProvisionRequest,
) -> Result<ProvisionedInstance, String> {
Ok(ProvisionedInstance {
index: 0,
contract_id: 42,
offer_id: 7,
host_id: Some(99),
gpu_name: "RTX 4060".to_owned(),
gpu_ram: Some(8_192.0),
dph_total: 0.064,
})
}
fn ssh_endpoint(
&mut self,
_contract_id: u64,
_label: &str,
_lifecycle: &LifecyclePolicy,
ssh_user: &str,
) -> Result<VastAiSshEndpoint, String> {
Ok(VastAiSshEndpoint {
host: "ssh5.vast.ai".to_owned(),
port: 22_017,
user: ssh_user.to_owned(),
})
}
fn destroy_contract(&mut self, contract_id: u64) -> Result<(), String> {
self.destroyed_contracts.lock().push(contract_id);
Ok(())
}
}
#[derive(Clone)]
struct RecordingBootstrapLauncher {
stop_reasons: Arc<Mutex<Vec<BootstrapStopReason>>>,
}
impl VastAiBootstrapLauncher for RecordingBootstrapLauncher {
type Handle = u64;
fn start_bootstrap(
&mut self,
spec: NodeProvisionSpec,
_endpoint: VastAiSshEndpoint,
_sink: PluginSink,
_producer: Option<DatastreamProducer>,
) -> Result<Self::Handle, String> {
Ok(spec.node_id)
}
fn stop_bootstrap(&mut self, _handle: &mut Self::Handle, reason: BootstrapStopReason) {
self.stop_reasons.lock().push(reason);
}
}
#[test] #[test]
fn vastai_provisioning_build_request_keeps_bootstrap_args_out_of_onstart() { fn vastai_provisioning_build_request_keeps_bootstrap_args_out_of_onstart() {
let plugin = plugin_with_onstart(None); let plugin = plugin_with_onstart(None);
@ -1057,38 +957,6 @@ mod tests {
assert_eq!(request.onstart.as_deref(), Some("echo explicit setup")); assert_eq!(request.onstart.as_deref(), Some("echo explicit setup"));
} }
#[test]
fn vastai_complete_bootstrap_keeps_log_tail_until_node_stop() {
let destroyed_contracts = Arc::new(Mutex::new(Vec::new()));
let stop_reasons = Arc::new(Mutex::new(Vec::new()));
let sink = PluginSink::new(Arc::new(ObservationSink::default()));
let mut plugin = VastAiProvisioningPlugin::new(
RecordingLeaseClient {
destroyed_contracts: destroyed_contracts.clone(),
},
RecordingBootstrapLauncher {
stop_reasons: stop_reasons.clone(),
},
VastAiProvisioningConfig::default(),
);
let handle = plugin
.start_node(node_spec_with_bootstrap_args(), sink)
.expect("VastAI node starts");
plugin
.complete_bootstrap(&handle)
.expect("runtime-ready bootstrap completion succeeds");
assert!(
stop_reasons.lock().is_empty(),
"VastAI bootstrap SSH tail must remain alive for post-ready worker logs"
);
plugin.stop_node(&handle).expect("VastAI node stops");
assert_eq!(*stop_reasons.lock(), vec![BootstrapStopReason::NodeStop]);
assert_eq!(*destroyed_contracts.lock(), vec![42]);
}
#[test] #[test]
fn vastai_provisioning_next_ssh_backoff_doubles_until_thirty_second_cap() { fn vastai_provisioning_next_ssh_backoff_doubles_until_thirty_second_cap() {
for (current, expected) in [ for (current, expected) in [

1
src/actor.rs
View file

@ -681,6 +681,7 @@ impl<'a> Ctx<'a> {
self.inner.extension() self.inner.extension()
} }
/// Return system-level information (worker count, actor count, uptime). /// Return system-level information (worker count, actor count, uptime).
pub fn system_info(&self) -> SystemInfo { pub fn system_info(&self) -> SystemInfo {
self.inner.system_info() self.inner.system_info()

1
src/runtime.rs
View file

@ -598,6 +598,7 @@ impl Runtime {
self.stats_hook = Some(hook); self.stats_hook = Some(hook);
} }
/// Set the sink for non-local (remote) message delivery. /// Set the sink for non-local (remote) message delivery.
/// ///
/// The sink owns all codec/transport concerns; core only knows how to hand /// The sink owns all codec/transport concerns; core only knows how to hand

2
tools/vastai/src/config.rs
View file

@ -9,7 +9,6 @@ pub const ENV_MIN_INET_UP_MBPS: &str = "PP_MIN_INET_UP_MBPS";
pub const ENV_MIN_RELIABILITY: &str = "PP_MIN_RELIABILITY"; pub const ENV_MIN_RELIABILITY: &str = "PP_MIN_RELIABILITY";
pub const ENV_REQUIRE_VERIFIED: &str = "PP_REQUIRE_VERIFIED"; pub const ENV_REQUIRE_VERIFIED: &str = "PP_REQUIRE_VERIFIED";
pub const ENV_DROP_CHEAP_FRAC: &str = "PP_DROP_CHEAP_FRAC"; pub const ENV_DROP_CHEAP_FRAC: &str = "PP_DROP_CHEAP_FRAC";
pub const ENV_MAX_DPH_TOTAL: &str = "PP_MAX_DPH_TOTAL";
pub const ENV_LEASE_PACE_MS: &str = "PP_LEASE_PACE_MS"; pub const ENV_LEASE_PACE_MS: &str = "PP_LEASE_PACE_MS";
pub const ENV_BLACKLIST_HOSTS: &str = "PP_BLACKLIST_HOSTS"; pub const ENV_BLACKLIST_HOSTS: &str = "PP_BLACKLIST_HOSTS";
pub const ENV_ASSUME_YES: &str = "PP_ASSUME_YES"; pub const ENV_ASSUME_YES: &str = "PP_ASSUME_YES";
@ -56,7 +55,6 @@ impl SelectionPolicy {
.unwrap_or(0.95); .unwrap_or(0.95);
policy.require_verified = truthy_env(ENV_REQUIRE_VERIFIED); policy.require_verified = truthy_env(ENV_REQUIRE_VERIFIED);
policy.min_up_mbps = env_optional_positive_f64(ENV_MIN_INET_UP_MBPS); policy.min_up_mbps = env_optional_positive_f64(ENV_MIN_INET_UP_MBPS);
policy.max_dph_total = env_optional_positive_f64(ENV_MAX_DPH_TOTAL);
policy.drop_cheap_frac = std::env::var(ENV_DROP_CHEAP_FRAC) policy.drop_cheap_frac = std::env::var(ENV_DROP_CHEAP_FRAC)
.ok() .ok()
.and_then(|s| s.trim().parse::<f64>().ok()) .and_then(|s| s.trim().parse::<f64>().ok())

33
tools/vastai/src/filters.rs
View file

@ -13,38 +13,5 @@ pub(crate) fn reachable_offers(offers: Vec<Offer>, policy: &SelectionPolicy) ->
}) })
.filter(|o| o.host_id.map_or(true, |h| !blacklist.contains(&h))) .filter(|o| o.host_id.map_or(true, |h| !blacklist.contains(&h)))
.filter(|o| o.verification.as_deref() != Some("deverified")) .filter(|o| o.verification.as_deref() != Some("deverified"))
.filter(|o| policy.max_dph_total.is_none_or(|max| o.dph_total <= max))
.collect() .collect()
} }
#[cfg(test)]
mod tests {
use super::*;
fn offer(id: u64, dph_total: f64) -> Offer {
Offer {
id,
gpu_name: "Tesla T4".to_owned(),
dph_total,
gpu_ram: Some(16_000.0),
geolocation: Some("US".to_owned()),
inet_down_cost_per_tb: 0.0,
inet_up_cost_per_tb: 0.0,
host_id: Some(id),
verification: Some("unverified".to_owned()),
}
}
#[test]
fn max_price_cap_keeps_only_affordable_reachable_offers() {
let policy = SelectionPolicy {
max_dph_total: Some(0.10),
..SelectionPolicy::default()
};
let reachable = reachable_offers(vec![offer(1, 0.09), offer(2, 0.11)], &policy);
assert_eq!(reachable.len(), 1);
assert_eq!(reachable[0].id, 1);
}
}

18
tools/vastai/src/search.rs
View file

@ -47,9 +47,6 @@ pub async fn select_offer_pool_with_policy(
if let Some(min_ram) = policy.min_gpu_ram_mb { if let Some(min_ram) = policy.min_gpu_ram_mb {
query["gpu_ram"] = serde_json::json!({"gte": min_ram}); query["gpu_ram"] = serde_json::json!({"gte": min_ram});
} }
if let Some(max_dph_total) = policy.max_dph_total {
query["dph_total"] = serde_json::json!({"lte": max_dph_total});
}
if let Some(gpu_name) = policy.gpu_name.as_deref().filter(|s| !s.is_empty()) { if let Some(gpu_name) = policy.gpu_name.as_deref().filter(|s| !s.is_empty()) {
query["gpu_name"] = serde_json::json!({"eq": gpu_name}); query["gpu_name"] = serde_json::json!({"eq": gpu_name});
} }
@ -81,19 +78,14 @@ pub async fn select_offer_pool_with_policy(
let pool = rank_survivors(reachable, &cost, policy.drop_cheap_frac); let pool = rank_survivors(reachable, &cost, policy.drop_cheap_frac);
if pool.is_empty() { if pool.is_empty() {
let cap = policy return Err(
.max_dph_total "no offers available (after quality/geo/host-blacklist filters and cheap-tail drop)"
.map_or_else(|| "uncapped".to_owned(), |max| format!("max ${max:.3}/hr")); .to_string(),
return Err(format!( );
"no offers available ({cap}, after quality/geo/host-blacklist filters and cheap-tail drop)"
));
} }
let cap = policy
.max_dph_total
.map_or_else(|| "uncapped".to_owned(), |max| format!("max ${max:.3}/hr"));
eprintln!( eprintln!(
"select_offer_pool: {} survivor(s) for {target_count} instance(s) after \ "select_offer_pool: {} survivor(s) for {target_count} instance(s) after \
per-model {:.0}% cheap-drop ({cap}, cheapest ${:.3}/hr eff)", per-model {:.0}% cheap-drop (cheapest ${:.3}/hr eff)",
pool.len(), pool.len(),
policy.drop_cheap_frac * 100.0, policy.drop_cheap_frac * 100.0,
cost.effective_price(&pool[0]), cost.effective_price(&pool[0]),

2
tools/vastai/src/types.rs
View file

@ -62,7 +62,6 @@ pub struct SelectionPolicy {
pub require_verified: bool, pub require_verified: bool,
pub min_down_mbps: f64, pub min_down_mbps: f64,
pub min_up_mbps: Option<f64>, pub min_up_mbps: Option<f64>,
pub max_dph_total: Option<f64>,
pub blacklist_hosts: Vec<u64>, pub blacklist_hosts: Vec<u64>,
pub drop_cheap_frac: f64, pub drop_cheap_frac: f64,
pub image_size_gb: Option<f64>, pub image_size_gb: Option<f64>,
@ -77,7 +76,6 @@ impl Default for SelectionPolicy {
require_verified: false, require_verified: false,
min_down_mbps: 100.0, min_down_mbps: 100.0,
min_up_mbps: None, min_up_mbps: None,
max_dph_total: None,
blacklist_hosts: vec![59017], blacklist_hosts: vec![59017],
drop_cheap_frac: 0.30, drop_cheap_frac: 0.30,
image_size_gb: None, image_size_gb: None,

1
xtask/Cargo.toml
View file

@ -5,7 +5,6 @@ edition = "2024"
[dependencies] [dependencies]
serde_json = "1" serde_json = "1"
blake3 = "1"
[target.'cfg(target_os = "linux")'.dependencies] [target.'cfg(target_os = "linux")'.dependencies]
libc = "0.2" libc = "0.2"

1047
xtask/src/main.rs
View file

File diff suppressed because it is too large Load diff