From d3a67d021fc465bcad8df531dad856d1a5536fb1 Mon Sep 17 00:00:00 2001
From: Zachery Aaron Shores-Chmielewski <zacheryasc@gmail.com>
Date: Sat, 25 Jul 2026 20:04:57 +0400
Subject: [PATCH] feat: successful 8 stage pipeline parallel run, more metrics

---
 apps/mvp-node/tinygrad_worker.py              |   38 +
 crates/mvp-system/src/bin/mvp_chat.rs         |   17 +-
 crates/mvp-system/src/bin/worker_node.rs      |   81 +-
 crates/mvp-system/src/orchestrator_app.rs     |   70 +-
 .../tests/vastai_provisioning_guarantees.rs   |   46 +-
 crates/mvp-system/src/vastai_offer_preview.rs |    1 +
 crates/mvp-system/src/vastai_provisioning.rs  |   49 +-
 tools/vastai/Cargo.toml                       |    2 +-
 tools/vastai/src/config.rs                    |    8 +
 tools/vastai/src/filters.rs                   |   19 +
 tools/vastai/src/monitor.rs                   |   54 +
 tools/vastai/src/search.rs                    |    3 +
 tools/vastai/src/teardown.rs                  |    5 +-
 tools/vastai/src/types.rs                     |    9 +
 xtask/src/main.rs                             | 1043 ++++++++++++++++-
 15 files changed, 1366 insertions(+), 79 deletions(-)

diff --git a/apps/mvp-node/tinygrad_worker.py b/apps/mvp-node/tinygrad_worker.py
index 89298ec..28a5810 100755
--- a/apps/mvp-node/tinygrad_worker.py
+++ b/apps/mvp-node/tinygrad_worker.py
@@ -1061,6 +1061,7 @@ def ring_readable(cmd: dict[str, Any]) -> None:
     materialized = materialize_object(payload, sequence, flags)
     materialized.update(
         object_id=object_id,
+        edge_id=ring["edge_id"],
         sequence=sequence,
         extent=extent,
         flags=flags,
@@ -1127,6 +1128,11 @@ def execute_step(cmd: dict[str, Any]) -> None:
     final_stage = bool(cmd.get("final_stage"))
     input_kind = obj.get("kind")
     input_extent = int(obj.get("extent", 0))
+    validation_ready = time.monotonic()
+    input_prepare_ms = 0
+    forward_ms = 0
+    realize_ms = 0
+    payload_pack_ms = 0
     execution_backend = "pipeline_stage"
     if not isinstance(model, PipelineStageTinygradModel):
         execution_backend = "full_transformer"
@@ -1136,29 +1142,50 @@ def execute_step(cmd: dict[str, Any]) -> None:
             fatal("FullTransformerInputUnsupported", step_id=int(cmd["step_id"]), kind=input_kind)
         if int(obj.get("flags", 0)) & FLAG_BEGIN_SEQUENCE and hasattr(model, "forward_jit"):
             model.forward_jit.reset()
+        forward_started = time.monotonic()
         token_array = model(obj["tensor"], int(obj.get("start_pos", 0))).realize().numpy().reshape(-1)
+        forward_ready = time.monotonic()
+        forward_ms = int((forward_ready - forward_started) * 1000)
         token = int(token_array[0])
+        payload_started = time.monotonic()
         payload = struct.pack("<I", token)
         output_kind = "token"
         flags = 1 if token == int(loaded.get("eos_token_id", 0)) else 0
+        payload_pack_ms = int((time.monotonic() - payload_started) * 1000)
     else:
         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)))
+        input_started = time.monotonic()
         input_tensor = model.token_hidden(obj["tensor"]) if input_kind == "tokens" else obj["tensor"]
+        input_ready = time.monotonic()
+        input_prepare_ms = int((input_ready - input_started) * 1000)
+        forward_started = time.monotonic()
         hidden = model.forward_hidden(input_tensor, int(obj.get("start_pos", 0)))
+        forward_ready = time.monotonic()
+        forward_ms = int((forward_ready - forward_started) * 1000)
         if final_stage:
+            realize_started = time.monotonic()
             token_array = model.next_token(hidden).realize().numpy().reshape(-1)
+            realize_ready = time.monotonic()
+            realize_ms = int((realize_ready - realize_started) * 1000)
             token = int(token_array[0])
+            payload_started = time.monotonic()
             payload = struct.pack("<I", token)
             output_kind = "token"
             flags = 1 if token == int(loaded.get("eos_token_id", 0)) else 0
+            payload_pack_ms = int((time.monotonic() - payload_started) * 1000)
         else:
             import numpy as np
 
+            realize_started = time.monotonic()
             activation = hidden.realize().numpy().astype(np.float16, copy=False)
+            realize_ready = time.monotonic()
+            realize_ms = int((realize_ready - realize_started) * 1000)
+            payload_started = time.monotonic()
             payload = activation.tobytes()
             output_kind = "activation"
             flags = 0
+            payload_pack_ms = int((time.monotonic() - payload_started) * 1000)
     compute_ready = time.monotonic()
     committed = write_record(
         ring,
@@ -1172,6 +1199,8 @@ def execute_step(cmd: dict[str, Any]) -> None:
         type="StepExecuted",
         step_id=int(cmd["step_id"]),
         ring_id=output_ring_id,
+        role_id=int(cmd["role_id"]),
+        stage_index=max(0, int(cmd["role_id"]) - 1),
         object_id=int(cmd["output_object_id"]),
         sequence=int(cmd["output_sequence"]),
         committed_bytes=committed,
@@ -1182,6 +1211,15 @@ def execute_step(cmd: dict[str, Any]) -> None:
         output_kind=output_kind,
         payload_bytes=len(payload),
         record_bytes=committed,
+        input_handle_id=handle,
+        input_object_id=int(cmd["input_object_id"]),
+        input_sequence=int(cmd["input_sequence"]),
+        input_edge_id=obj.get("edge_id"),
+        input_prepare_ms=input_prepare_ms,
+        model_forward_ms=forward_ms,
+        output_realize_ms=realize_ms,
+        payload_pack_ms=payload_pack_ms,
+        validation_ms=int((validation_ready - step_started) * 1000),
         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),
diff --git a/crates/mvp-system/src/bin/mvp_chat.rs b/crates/mvp-system/src/bin/mvp_chat.rs
index 68ad19a..26b7ce0 100644
--- a/crates/mvp-system/src/bin/mvp_chat.rs
+++ b/crates/mvp-system/src/bin/mvp_chat.rs
@@ -848,8 +848,6 @@ impl Config {
         }
         if let Some(vastai) = &self.vastai {
             args.extend([
-                "--vastai-api-key".to_owned(),
-                vastai.api_key.clone(),
                 "--vastai-bootstrap-command".to_owned(),
                 vastai.bootstrap_command.clone(),
                 "--no-vastai-confirm-lease".to_owned(),
@@ -1255,6 +1253,9 @@ impl OrchChild {
             .stdin(Stdio::null())
             .stdout(Stdio::null())
             .stderr(Stdio::null());
+        if let Some(vastai) = &config.vastai {
+            command.env("VAST_API_KEY", &vastai.api_key);
+        }
         #[cfg(all(target_os = "linux", not(test)))]
         unsafe {
             command.pre_exec(|| {
@@ -2838,6 +2839,18 @@ bootstrap_command = "boot"
                 assert_eq!(vastai.relay_url, "https://relay.example");
                 assert_eq!(vastai.bootstrap_command, "boot");
                 assert_eq!(vastai.image, "docker.io/acme/node:latest");
+                let args = config.orchestrator_cli_args("docker.io/acme/node:latest");
+                assert!(
+                    !args
+                        .iter()
+                        .any(|arg| arg == "--vastai-api-key" || arg == "secret"),
+                    "Vast.ai API key must not be exposed in orchestrator argv: {args:?}"
+                );
+                assert!(
+                    args.windows(2)
+                        .any(|pair| pair == ["--vastai-bootstrap-command", "boot"]),
+                    "non-secret Vast.ai config should still be forwarded"
+                );
             },
         );
     }
diff --git a/crates/mvp-system/src/bin/worker_node.rs b/crates/mvp-system/src/bin/worker_node.rs
index 055a500..828bca0 100644
--- a/crates/mvp-system/src/bin/worker_node.rs
+++ b/crates/mvp-system/src/bin/worker_node.rs
@@ -784,6 +784,7 @@ impl WorkerEdgeRuntime {
             outbound.kind,
             mvp_system::actors::node_agent::StageEdgeKindWire::TokenOut
         );
+        let step_started = Instant::now();
         let mut pump = || pump_network(driver, stack);
         let committed_bytes = worker.execute_step(
             u64::from(config.stage_index) + 1,
@@ -800,6 +801,8 @@ impl WorkerEdgeRuntime {
             datastream,
             &mut pump,
         )?;
+        let helper_execute_ms = duration_ms_u64(step_started.elapsed());
+        let egress_read_started = Instant::now();
         let record = {
             let arena = arena_manager.lock();
             let lease = arena
@@ -809,6 +812,7 @@ impl WorkerEdgeRuntime {
                 .read_arena(lease.layout.data_offset, committed_bytes)
                 .map_err(|e| format!("read egress ring: {e}"))?
         };
+        let egress_read_ms = duration_ms_u64(egress_read_started.elapsed());
         let record_bytes = record.len();
         emit_node_event(
             datastream,
@@ -820,9 +824,17 @@ impl WorkerEdgeRuntime {
                 "edge_id":outbound.edge_id,
                 "edge_kind":format!("{:?}", outbound.kind),
                 "ring_id":output_ring_id,
+                "step_id":step_id,
+                "input_object_id":object_id,
+                "input_edge_id":input_edge_id,
+                "sequence":sequence,
+                "output_object_id":output_object_id,
+                "output_sequence":sequence,
                 "record_bytes":record_bytes,
                 "committed_bytes":committed_bytes,
                 "final_stage":final_stage,
+                "helper_execute_ms":helper_execute_ms,
+                "egress_ring_read_ms":egress_read_ms,
             }),
         );
         self.driver_model
@@ -838,7 +850,9 @@ impl WorkerEdgeRuntime {
             .outbound_sender
             .as_ref()
             .ok_or_else(|| "outbound edge sender missing".to_owned())?;
+        let edge_send_started = Instant::now();
         sender.send(record)?;
+        let edge_send_ms = duration_ms_u64(edge_send_started.elapsed());
         emit_node_event(
             datastream,
             config,
@@ -848,8 +862,12 @@ impl WorkerEdgeRuntime {
             json!({
                 "edge_id":outbound.edge_id,
                 "edge_kind":format!("{:?}", outbound.kind),
+                "step_id":step_id,
+                "object_id":output_object_id,
+                "sequence":sequence,
                 "bytes":record_bytes,
                 "record_bytes":record_bytes,
+                "send_ms":edge_send_ms,
             }),
         );
         stack
@@ -891,28 +909,31 @@ impl WorkerEdgeRuntime {
         if inbound.edge_id != edge_id {
             return Ok(());
         }
-        let records = {
+        let (records, buffered_bytes) = {
             let buffer = self.ingress_streams.entry(stream_id).or_default();
             buffer.extend_from_slice(&bytes);
+            let buffered_bytes = buffer.len();
             let mut records = Vec::new();
             while let Some(record) = take_complete_ingress_record(buffer, inbound.object_spec)? {
                 records.push(record);
             }
-            records
+            (records, buffered_bytes)
         };
         for record in records {
             let ring_id = self
                 .inbound_ring_id
                 .ok_or_else(|| "inbound ring missing".to_owned())?;
+            let ring_write_started = Instant::now();
             {
                 let arena = arena_manager.lock();
                 let lease = arena
                     .lookup_lease(arena::RingId(ring_id))
                     .ok_or_else(|| format!("inbound ring {ring_id} lease missing"))?;
                 arena
-                    .write_arena(lease.layout.data_offset, &record)
+                    .write_arena(lease.layout.data_offset, &record.bytes)
                     .map_err(|e| format!("write ingress ring: {e}"))?;
             }
+            let ingress_ring_write_ms = duration_ms_u64(ring_write_started.elapsed());
             emit_node_event(
                 datastream,
                 config,
@@ -923,9 +944,18 @@ impl WorkerEdgeRuntime {
                     "edge_id":edge_id,
                     "edge_kind":format!("{:?}", inbound.kind),
                     "ring_id":ring_id,
-                    "record_bytes":record.len(),
+                    "stream_id":stream_id,
+                    "object_id":record.object_id,
+                    "sequence":record.sequence,
+                    "extent":record.extent,
+                    "begin_sequence":record.begin_sequence,
+                    "end_of_sequence":record.end_of_sequence,
+                    "record_bytes":record.bytes.len(),
+                    "ingress_buffer_bytes":buffered_bytes,
+                    "ingress_ring_write_ms":ingress_ring_write_ms,
                 }),
             );
+            let object_load_started = Instant::now();
             let loaded = worker.ring_readable(
                 ring_id,
                 edge_id,
@@ -934,11 +964,30 @@ impl WorkerEdgeRuntime {
                 datastream,
                 &mut || {},
             )?;
+            let object_load_ms = duration_ms_u64(object_load_started.elapsed());
             let key = ObjectKey {
                 edge_id,
                 object_id: loaded.object_id,
             };
             self.object_handles.insert(key, loaded.clone());
+            emit_node_event(
+                datastream,
+                config,
+                NODE_STAGE_CHANNEL,
+                "object_loaded",
+                "ready",
+                json!({
+                    "edge_id":edge_id,
+                    "edge_kind":format!("{:?}", inbound.kind),
+                    "ring_id":ring_id,
+                    "stream_id":stream_id,
+                    "object_id":loaded.object_id,
+                    "sequence":loaded.sequence,
+                    "handle_generation":loaded.handle_generation,
+                    "handle_id":loaded.handle_id,
+                    "object_load_ms":object_load_ms,
+                }),
+            );
             stack
                 .runtime
                 .send_to(
@@ -1270,6 +1319,9 @@ impl WorkerEdgeRuntime {
     }
 }
 
+fn duration_ms_u64(duration: Duration) -> u64 {
+    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
+}
 fn edge_object_spec(spec: StageObjectSpecWire) -> edge::ObjectSpec {
     edge::ObjectSpec {
         kind: edge::ObjectKind::Activation,
@@ -1294,17 +1346,34 @@ fn ingress_object_spec(spec: StageObjectSpecWire) -> ingress::ObjectSpec {
     }
 }
 
+struct IngressRecordBytes {
+    bytes: Vec<u8>,
+    object_id: u64,
+    sequence: u64,
+    extent: u64,
+    begin_sequence: bool,
+    end_of_sequence: bool,
+}
+
 fn take_complete_ingress_record(
     buffer: &mut Vec<u8>,
     spec: StageObjectSpecWire,
-) -> Result<Option<Vec<u8>>, String> {
+) -> Result<Option<IngressRecordBytes>, String> {
     let record = match ingress::read_object_record(buffer, ingress_object_spec(spec), false)
         .map_err(|reason| format!("invalid object record: {reason:?}"))?
     {
         ingress::ObjectRecordRead::Incomplete => return Ok(None),
         ingress::ObjectRecordRead::Complete(record) => record,
     };
-    Ok(Some(buffer.drain(..record.total_len).collect()))
+    let bytes = buffer.drain(..record.total_len).collect();
+    Ok(Some(IngressRecordBytes {
+        bytes,
+        object_id: record.object_id.0,
+        sequence: record.sequence,
+        extent: record.extent,
+        begin_sequence: record.flags.begin_sequence,
+        end_of_sequence: record.flags.end_of_sequence,
+    }))
 }
 
 fn value_u64(value: &Value, field: &str) -> Result<u64, String> {
diff --git a/crates/mvp-system/src/orchestrator_app.rs b/crates/mvp-system/src/orchestrator_app.rs
index ec0aeba..64632b6 100644
--- a/crates/mvp-system/src/orchestrator_app.rs
+++ b/crates/mvp-system/src/orchestrator_app.rs
@@ -702,11 +702,13 @@ impl VastAiRuntimeConfig {
             "ssh_user": &self.provisioning.ssh_user,
             "gpu_name": &self.provisioning.selection.gpu_name,
             "min_gpu_ram_mb": self.provisioning.selection.min_gpu_ram_mb,
+            "min_compute_cap": self.provisioning.selection.min_compute_cap,
             "min_down_mbps": self.provisioning.selection.min_down_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,
             "require_verified": self.provisioning.selection.require_verified,
+            "state_timeout_secs": self.provisioning.lifecycle.state_timeout.as_secs(),
             "confirm_lease": self.provisioning.confirm_lease,
             "has_api_key": self.api_key.is_some(),
             "has_onstart": self.provisioning.onstart.is_some(),
@@ -2975,6 +2977,7 @@ fn wait_for_weights_loaded_count(
     let mut last_resend = Instant::now();
     let mut active_stage = None::<u32>;
     let mut resend_attempt = 0_u64;
+    let mut stage_resend_counts = BTreeMap::<u32, u64>::new();
     loop {
         pump(driver, stack, frame_tx);
         emit_swim_transitions(orch_datastream, dashboard, run_id, node_id, stack);
@@ -3002,6 +3005,12 @@ fn wait_for_weights_loaded_count(
                 ));
             };
             let stage_node_id = stage.node_id.0;
+            let stage_send_count = {
+                let count = stage_resend_counts.entry(stage.stage_index).or_default();
+                *count += 1;
+                *count
+            };
+            let emit_wait_headline = stage_send_count == 1 || stage_send_count % 15 == 0;
             let ready = readies.get(&stage_node_id).ok_or_else(|| {
                 format!("missing runtime-ready node for stage {}", stage.stage_index)
             })?;
@@ -3015,6 +3024,7 @@ fn wait_for_weights_loaded_count(
                     "attempt":resend_attempt,
                     "stage_count":pipeline_plan.stages.len(),
                     "stage_index":stage.stage_index,
+                    "stage_send_count":stage_send_count,
                     "loaded_stage_count":loaded_stages.len()
                 }),
             );
@@ -3040,6 +3050,29 @@ fn wait_for_weights_loaded_count(
                     "route_matches_ready":route_owner == Some(ready.swim_node_id),
                 }),
             );
+            if emit_wait_headline {
+                orch_datastream.emit_bootstrap(
+                    dashboard,
+                    run_id,
+                    node_id,
+                    "stage_provision_wait",
+                    "observed",
+                    json!({
+                        "attempt":resend_attempt,
+                        "stage_count":pipeline_plan.stages.len(),
+                        "stage_index":stage.stage_index,
+                        "stage_node_id":stage_node_id,
+                        "stage_send_count":stage_send_count,
+                        "loaded_stage_count":loaded_stages.len(),
+                        "message":format!(
+                            "loaded {} of {}; waiting on stage {}",
+                            loaded_stages.len(),
+                            pipeline_plan.stages.len(),
+                            stage.stage_index
+                        )
+                    }),
+                );
+            }
             provision_stage_from_plan(
                 stack,
                 ready.node_actor,
@@ -3120,7 +3153,7 @@ fn next_pipeline_weight_load_stage<'a>(
         .stages
         .iter()
         .filter(|stage| !loaded_stages.contains(&stage.stage_index))
-        .min_by_key(|stage| stage.stage_index)
+        .max_by_key(|stage| stage.stage_index)
 }
 
 struct FailedProvisionPlugin;
@@ -4115,7 +4148,7 @@ impl PipelinePromptRuntime {
             request_id,
             "pipeline_token_in",
             "started",
-            json!({"edge_id":self.token_in_edge_id,"sequence":sequence,"tokens":tokens.len(),"begin_sequence":true}),
+            json!({"edge_id":self.token_in_edge_id,"sequence":sequence,"tokens":tokens.len(),"begin_sequence":true,"token_count":tokens.len(),"token_ids":&tokens}),
         );
         self.send_token_in(sequence, &tokens, true)?;
         orch_datastream.emit_prompt(
@@ -4125,7 +4158,7 @@ impl PipelinePromptRuntime {
             request_id,
             "pipeline_token_in",
             "ready",
-            json!({"edge_id":self.token_in_edge_id,"sequence":sequence,"begin_sequence":true}),
+            json!({"edge_id":self.token_in_edge_id,"sequence":sequence,"begin_sequence":true,"token_count":tokens.len()}),
         );
         Ok(())
     }
@@ -4200,7 +4233,26 @@ impl PipelinePromptRuntime {
             self.active = None;
             return Ok(());
         }
-        self.send_token_in(self.next_sequence, &[pending.token_id], false)?;
+        let sequence = self.next_sequence;
+        orch_datastream.emit_prompt(
+            dashboard,
+            run_id,
+            node_id,
+            request_id,
+            "pipeline_token_in",
+            "started",
+            json!({"edge_id":self.token_in_edge_id,"sequence":sequence,"tokens":1,"begin_sequence":false,"token_count":1,"token_id":pending.token_id}),
+        );
+        self.send_token_in(sequence, &[pending.token_id], false)?;
+        orch_datastream.emit_prompt(
+            dashboard,
+            run_id,
+            node_id,
+            request_id,
+            "pipeline_token_in",
+            "ready",
+            json!({"edge_id":self.token_in_edge_id,"sequence":sequence,"begin_sequence":false,"token_count":1,"token_id":pending.token_id}),
+        );
         Ok(())
     }
     fn send_token_in(
@@ -4317,7 +4369,7 @@ impl PipelinePromptRuntime {
                     request_id,
                     "pipeline_token_out",
                     "observed",
-                    json!({"edge_id":self.token_out_edge_id,"object_id":record.object_id,"sequence":record.sequence,"token_id":record.token_id,"eos":record.eos}),
+                    json!({"edge_id":self.token_out_edge_id,"object_id":record.object_id,"sequence":record.sequence,"token_id":record.token_id,"eos":record.eos,"generated_index":self.generated_tokens.len() + 1}),
                 );
                 self.generated_tokens.push(record.token_id);
                 let reached_limit = self.generated_tokens.len() as u32 >= active.request.max_tokens;
@@ -4328,7 +4380,7 @@ impl PipelinePromptRuntime {
                     request_id,
                     "pipeline_tokenizer_decode",
                     "started",
-                    json!({"node_actor":self.tokenizer_decode_actor,"reply_to":self.tokenizer_reply_to,"token_id":record.token_id}),
+                    json!({"node_actor":self.tokenizer_decode_actor,"reply_to":self.tokenizer_reply_to,"token_id":record.token_id,"sequence":record.sequence,"generated_index":self.generated_tokens.len()}),
                 );
                 self.request_decode(
                     runtime,
@@ -5325,13 +5377,13 @@ mod tests {
 
         let first =
             next_pipeline_weight_load_stage(&plan, &loaded, None).expect("first stage selected");
-        assert_eq!(first.stage_index, 0);
+        assert_eq!(first.stage_index, 6);
 
         let resent = next_pipeline_weight_load_stage(&plan, &loaded, Some(first.stage_index))
             .expect("active stage is resent before it loads");
-        assert_eq!(resent.stage_index, 0);
+        assert_eq!(resent.stage_index, 6);
 
-        for expected_stage in 0..7 {
+        for expected_stage in (0..7).rev() {
             let active = next_pipeline_weight_load_stage(&plan, &loaded, None)
                 .expect("next unloaded stage selected");
             assert_eq!(active.stage_index, expected_stage);
diff --git a/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs b/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs
index af3a56d..9ce0f6e 100644
--- a/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs
+++ b/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs
@@ -36,6 +36,7 @@ struct FakeLeaseClient {
     destroyed: Vec<u64>,
     destroy_result: Option<Result<(), String>>,
     next_contract_id: u64,
+    host_ids: VecDeque<Option<u64>>,
 }
 
 impl FakeLeaseClient {
@@ -50,11 +51,12 @@ impl VastAiLeaseClient for FakeLeaseClient {
         self.requests.push(request);
         let contract_id = self.next_contract_id;
         self.next_contract_id = self.next_contract_id.wrapping_add(1).max(1);
+        let host_id = self.host_ids.pop_front().unwrap_or(Some(77));
         Ok(ProvisionedInstance {
             index: 0,
             contract_id,
             offer_id: 55,
-            host_id: Some(77),
+            host_id,
             gpu_name: "RTX 4090".to_owned(),
             gpu_ram: Some(24_000.0),
             dph_total: 0.42,
@@ -238,6 +240,38 @@ fn vastai_plugin_omits_ssh_public_key_when_unconfigured() {
     assert_eq!(request.env.get("SSH_PUBLIC_KEY"), None);
 }
 
+#[test]
+fn pipeline_starts_blacklist_hosts_already_leased_in_run() {
+    let mut client = FakeLeaseClient::default().with_contract(100);
+    client.host_ids.extend([Some(77), Some(88)]);
+    let mut plugin = VastAiProvisioningPlugin::new(client, FakeBootstrap::default(), config());
+    let first = spec();
+    let mut second = spec();
+    second.node_id = 12;
+    second.stage_index = Some(3);
+
+    let first_handle = plugin.start_node(first, sink()).unwrap();
+    let second_handle = plugin.start_node(second, sink()).unwrap();
+
+    let requests = &plugin.client().requests;
+    assert_eq!(requests.len(), 2);
+    assert!(
+        !requests[0].selection.blacklist_hosts.contains(&77),
+        "first node should not preemptively blacklist the host it has not leased"
+    );
+    assert!(
+        requests[1].selection.blacklist_hosts.contains(&77),
+        "second node should avoid the first node's Vast.ai host"
+    );
+    assert!(
+        requests[1].selection.blacklist_hosts.contains(&59017),
+        "existing operator blacklist must be preserved"
+    );
+
+    plugin.stop_node(&second_handle).unwrap();
+    plugin.stop_node(&first_handle).unwrap();
+}
+
 #[test]
 fn stop_destroys_known_vastai_contract_exactly_once() {
     let mut plugin = VastAiProvisioningPlugin::new(
@@ -259,7 +293,7 @@ fn stop_destroys_known_vastai_contract_exactly_once() {
 }
 
 #[test]
-fn vastai_complete_bootstrap_stops_bootstrap_without_destroying_contract() {
+fn vastai_complete_bootstrap_keeps_log_tail_until_node_stop() {
     let mut plugin = VastAiProvisioningPlugin::new(
         FakeLeaseClient::default().with_contract(100),
         FakeBootstrap::default(),
@@ -269,9 +303,9 @@ fn vastai_complete_bootstrap_stops_bootstrap_without_destroying_contract() {
 
     plugin.complete_bootstrap(&handle).unwrap();
 
-    assert_eq!(
-        plugin.bootstrap().stops,
-        vec![(1, BootstrapStopReason::RuntimeReady)]
+    assert!(
+        plugin.bootstrap().stops.is_empty(),
+        "runtime-ready completion should keep the SSH log tail alive"
     );
     assert_eq!(plugin.client().destroyed, Vec::<u64>::new());
     assert_eq!(plugin.active_contract_count(), 1);
@@ -281,7 +315,7 @@ fn vastai_complete_bootstrap_stops_bootstrap_without_destroying_contract() {
     assert_eq!(plugin.client().destroyed, vec![100]);
     assert_eq!(
         plugin.bootstrap().stops,
-        vec![(1, BootstrapStopReason::RuntimeReady)]
+        vec![(1, BootstrapStopReason::NodeStop)]
     );
     assert_eq!(plugin.active_contract_count(), 0);
 }
diff --git a/crates/mvp-system/src/vastai_offer_preview.rs b/crates/mvp-system/src/vastai_offer_preview.rs
index 6e449be..a5abd57 100644
--- a/crates/mvp-system/src/vastai_offer_preview.rs
+++ b/crates/mvp-system/src/vastai_offer_preview.rs
@@ -57,6 +57,7 @@ mod tests {
             gpu_name: "RTX 4090".to_owned(),
             dph_total: 0.375,
             gpu_ram: Some(23.5),
+            compute_cap: 890,
             geolocation: Some("US".to_owned()),
             inet_down_cost_per_tb: 0.0,
             inet_up_cost_per_tb: 0.0,
diff --git a/crates/mvp-system/src/vastai_provisioning.rs b/crates/mvp-system/src/vastai_provisioning.rs
index 673c80f..a5279f9 100644
--- a/crates/mvp-system/src/vastai_provisioning.rs
+++ b/crates/mvp-system/src/vastai_provisioning.rs
@@ -1,5 +1,5 @@
 use parking_lot::Mutex;
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
 use std::sync::{
@@ -551,15 +551,14 @@ fn spawn_retrying_ssh_bootstrap(
 
                         match wait_result {
                             Some(Ok(status)) => {
-                                let line = if status.success() {
-                                    format!(
-                                        "VastAI SSH bootstrap exited before runtime ready: {status}; retrying"
-                                    )
+                                let readiness = if status.success() {
+                                    "exited before runtime ready"
                                 } else {
-                                    format!(
-                                        "VastAI SSH bootstrap failed before runtime ready: {status}; retrying"
-                                    )
+                                    "not ready before runtime ready"
                                 };
+                                let line = format!(
+                                    "VastAI SSH bootstrap {readiness} (attempt {attempt}, {status}); retrying"
+                                );
                                 sink.observe(PluginObservation::ProviderLine {
                                     run_id,
                                     node_id,
@@ -571,7 +570,9 @@ fn spawn_retrying_ssh_bootstrap(
                                 sink.observe(PluginObservation::ProviderLine {
                                     run_id,
                                     node_id,
-                                    line: format!("wait VastAI SSH bootstrap: {error}; retrying"),
+                                    line: format!(
+                                        "wait VastAI SSH bootstrap attempt {attempt}: {error}; retrying"
+                                    ),
                                 });
                                 break;
                             }
@@ -583,7 +584,9 @@ fn spawn_retrying_ssh_bootstrap(
                     sink.observe(PluginObservation::ProviderLine {
                         run_id,
                         node_id,
-                        line: format!("spawn VastAI SSH bootstrap failed: {error}; retrying"),
+                        line: format!(
+                            "spawn VastAI SSH bootstrap attempt {attempt} failed: {error}; retrying"
+                        ),
                     });
                 }
             }
@@ -594,7 +597,10 @@ fn spawn_retrying_ssh_bootstrap(
             sink.observe(PluginObservation::ProviderLine {
                 run_id,
                 node_id,
-                line: format!("VastAI SSH bootstrap retrying in {}s", backoff.as_secs()),
+                line: format!(
+                    "VastAI SSH bootstrap retrying in {}s after attempt {attempt}",
+                    backoff.as_secs()
+                ),
             });
             std::thread::sleep(backoff);
             backoff = next_ssh_backoff(backoff);
@@ -670,6 +676,7 @@ where
     bootstrap: B,
     config: VastAiProvisioningConfig,
     bootstrap_producer: Option<DatastreamProducer>,
+    leased_host_ids: BTreeSet<u64>,
     next_handle_id: u64,
     nodes: BTreeMap<u64, VastAiNode<B::Handle>>,
 }
@@ -677,6 +684,7 @@ where
 struct VastAiNode<H> {
     contract_id: u64,
     bootstrap: Option<H>,
+    host_id: Option<u64>,
 }
 
 impl<C, B> VastAiProvisioningPlugin<C, B>
@@ -691,6 +699,7 @@ where
             config,
             bootstrap_producer: None,
             next_handle_id: 1,
+            leased_host_ids: BTreeSet::new(),
             nodes: BTreeMap::new(),
         }
     }
@@ -741,6 +750,13 @@ where
         {
             env.insert("SSH_PUBLIC_KEY".to_owned(), key.to_owned());
         }
+        let mut selection = self.config.selection.clone();
+        for host_id in &self.leased_host_ids {
+            if !selection.blacklist_hosts.contains(host_id) {
+                selection.blacklist_hosts.push(*host_id);
+            }
+        }
+
         ProvisionRequest {
             count: 1,
             image: spec.image.clone(),
@@ -749,7 +765,7 @@ where
             env,
             per_instance_env: vec![BTreeMap::new()],
             onstart: self.config.onstart.clone(),
-            selection: self.config.selection.clone(),
+            selection,
             lifecycle: self.config.lifecycle.clone(),
             confirm_lease: self.config.confirm_lease,
         }
@@ -860,6 +876,11 @@ where
             }
         };
 
+        let host_id = instance.host_id;
+        if let Some(host_id) = host_id {
+            self.leased_host_ids.insert(host_id);
+        }
+
         let handle = PluginNodeHandle {
             id: self.next_handle_id,
             provider_process_id: None,
@@ -870,6 +891,7 @@ where
             VastAiNode {
                 contract_id: instance.contract_id,
                 bootstrap: Some(bootstrap),
+                host_id,
             },
         );
         Ok(handle)
@@ -883,6 +905,9 @@ where
         let Some(mut node) = self.nodes.remove(&handle.id) else {
             return Ok(());
         };
+        if let Some(host_id) = node.host_id {
+            self.leased_host_ids.remove(&host_id);
+        }
         if let Some(mut bootstrap) = node.bootstrap.take() {
             self.bootstrap
                 .stop_bootstrap(&mut bootstrap, BootstrapStopReason::NodeStop);
diff --git a/tools/vastai/Cargo.toml b/tools/vastai/Cargo.toml
index 49e6bf9..7ef7ab2 100644
--- a/tools/vastai/Cargo.toml
+++ b/tools/vastai/Cargo.toml
@@ -8,7 +8,7 @@ publish = false
 reqwest = { version = "0.12", features = ["json"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-tokio = { version = "1", features = ["time"] }
+tokio = { version = "1", features = ["macros", "rt", "time"] }
 urlencoding = "2"
 
 [dev-dependencies]
diff --git a/tools/vastai/src/config.rs b/tools/vastai/src/config.rs
index f149ec4..8a56a88 100644
--- a/tools/vastai/src/config.rs
+++ b/tools/vastai/src/config.rs
@@ -4,6 +4,7 @@ use crate::types::{LifecyclePolicy, SelectionPolicy};
 
 pub const ENV_IMAGE_SIZE_GB: &str = "PP_IMAGE_SIZE_GB";
 pub const ENV_GPU_MIN_RAM_MB: &str = "PP_GPU_MIN_RAM_MB";
+pub const ENV_MIN_COMPUTE_CAP: &str = "PP_MIN_COMPUTE_CAP";
 pub const ENV_MIN_INET_DOWN_MBPS: &str = "PP_MIN_INET_DOWN_MBPS";
 pub const ENV_MIN_INET_UP_MBPS: &str = "PP_MIN_INET_UP_MBPS";
 pub const ENV_MIN_RELIABILITY: &str = "PP_MIN_RELIABILITY";
@@ -11,6 +12,7 @@ pub const ENV_REQUIRE_VERIFIED: &str = "PP_REQUIRE_VERIFIED";
 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_STATE_TIMEOUT_SECS: &str = "PP_STATE_TIMEOUT_SECS";
 pub const ENV_BLACKLIST_HOSTS: &str = "PP_BLACKLIST_HOSTS";
 pub const ENV_ASSUME_YES: &str = "PP_ASSUME_YES";
 
@@ -48,6 +50,9 @@ impl SelectionPolicy {
     pub fn from_env() -> Self {
         let mut policy = Self::default();
         policy.min_gpu_ram_mb = env_positive_u64(ENV_GPU_MIN_RAM_MB);
+        if let Some(min_compute_cap) = env_positive_u64(ENV_MIN_COMPUTE_CAP) {
+            policy.min_compute_cap = Some(min_compute_cap);
+        }
         policy.min_down_mbps = env_nonnegative_f64(ENV_MIN_INET_DOWN_MBPS, 100.0);
         policy.min_reliability = std::env::var(ENV_MIN_RELIABILITY)
             .ok()
@@ -84,6 +89,9 @@ impl LifecyclePolicy {
                 .unwrap_or(600),
         );
         policy.poll_interval = poll_interval;
+        if let Some(state_timeout_secs) = env_positive_u64(ENV_STATE_TIMEOUT_SECS) {
+            policy.state_timeout = Duration::from_secs(state_timeout_secs);
+        }
         policy
     }
 }
diff --git a/tools/vastai/src/filters.rs b/tools/vastai/src/filters.rs
index 415ff80..a6639ad 100644
--- a/tools/vastai/src/filters.rs
+++ b/tools/vastai/src/filters.rs
@@ -14,6 +14,11 @@ 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.verification.as_deref() != Some("deverified"))
         .filter(|o| policy.max_dph_total.is_none_or(|max| o.dph_total <= max))
+        .filter(|o| {
+            policy
+                .min_compute_cap
+                .is_none_or(|min_compute_cap| o.compute_cap >= min_compute_cap)
+        })
         .collect()
 }
 
@@ -27,6 +32,7 @@ mod tests {
             gpu_name: "Tesla T4".to_owned(),
             dph_total,
             gpu_ram: Some(16_000.0),
+            compute_cap: 750,
             geolocation: Some("US".to_owned()),
             inet_down_cost_per_tb: 0.0,
             inet_up_cost_per_tb: 0.0,
@@ -47,4 +53,17 @@ mod tests {
         assert_eq!(reachable.len(), 1);
         assert_eq!(reachable[0].id, 1);
     }
+
+    #[test]
+    fn minimum_compute_cap_keeps_only_modern_cuda_offers() {
+        let mut legacy = offer(1, 0.03);
+        legacy.compute_cap = 520;
+        let mut modern = offer(2, 0.04);
+        modern.compute_cap = 750;
+
+        let reachable = reachable_offers(vec![legacy, modern], &SelectionPolicy::default());
+
+        assert_eq!(reachable.len(), 1);
+        assert_eq!(reachable[0].id, 2);
+    }
 }
diff --git a/tools/vastai/src/monitor.rs b/tools/vastai/src/monitor.rs
index a06343a..5c634cd 100644
--- a/tools/vastai/src/monitor.rs
+++ b/tools/vastai/src/monitor.rs
@@ -92,6 +92,12 @@ pub async fn wait_for_running_with_policy(
                 msg.unwrap_or_default()
             ));
         }
+        if !policy.state_timeout.is_zero() && state_since.elapsed() >= policy.state_timeout {
+            return Err(format!(
+                "instance {contract_id} stuck in status {actual} for {}s",
+                policy.state_timeout.as_secs()
+            ));
+        }
 
         match actual {
             "running" => {
@@ -112,3 +118,51 @@ pub async fn wait_for_running_with_policy(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use serde_json::json;
+    use wiremock::matchers::{method, path};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn stuck_loading_state_returns_error_instead_of_polling_forever() {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .and(path("/api/v0/instances/123/"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "instances": {
+                    "actual_status": "loading",
+                    "intended_status": "running",
+                    "status_msg": "afad30e59d72: Already exists"
+                }
+            })))
+            .mount(&server)
+            .await;
+
+        let policy = LifecyclePolicy {
+            poll_interval: Duration::from_millis(1),
+            state_timeout: Duration::from_millis(5),
+            ..LifecyclePolicy::default()
+        };
+
+        let error = wait_for_running_with_policy(
+            &reqwest::Client::new(),
+            &server.uri(),
+            "secret",
+            123,
+            &policy,
+        )
+        .await
+        .expect_err("stuck loading should be replaceable");
+
+        assert!(
+            error.contains("stuck in status loading"),
+            "error should name stuck provider state: {error}"
+        );
+    }
+}
diff --git a/tools/vastai/src/search.rs b/tools/vastai/src/search.rs
index 77ffcc8..a62f502 100644
--- a/tools/vastai/src/search.rs
+++ b/tools/vastai/src/search.rs
@@ -47,6 +47,9 @@ pub async fn select_offer_pool_with_policy(
     if let Some(min_ram) = policy.min_gpu_ram_mb {
         query["gpu_ram"] = serde_json::json!({"gte": min_ram});
     }
+    if let Some(min_compute_cap) = policy.min_compute_cap {
+        query["compute_cap"] = serde_json::json!({"gte": min_compute_cap});
+    }
     if let Some(max_dph_total) = policy.max_dph_total {
         query["dph_total"] = serde_json::json!({"lte": max_dph_total});
     }
diff --git a/tools/vastai/src/teardown.rs b/tools/vastai/src/teardown.rs
index dd5258e..1e40140 100644
--- a/tools/vastai/src/teardown.rs
+++ b/tools/vastai/src/teardown.rs
@@ -9,7 +9,10 @@ pub async fn destroy_instance(
     api_key: &str,
     contract_id: u64,
 ) -> Result<(), String> {
-    let url = format!("{base_url}/api/v0/instances/{contract_id}/");
+    let url = format!(
+        "{base_url}/api/v0/instances/{contract_id}/?api_key={}",
+        urlencoding::encode(api_key)
+    );
     let resp = client
         .delete(&url)
         .header("Authorization", format!("Bearer {api_key}"))
diff --git a/tools/vastai/src/types.rs b/tools/vastai/src/types.rs
index ee76230..41190ba 100644
--- a/tools/vastai/src/types.rs
+++ b/tools/vastai/src/types.rs
@@ -17,6 +17,9 @@ pub struct Offer {
     pub dph_total: f64,
     #[serde(default)]
     pub gpu_ram: Option<f64>,
+    /// CUDA compute capability encoded as vast.ai reports it (`890` = 8.9).
+    #[serde(default)]
+    pub compute_cap: u64,
     #[serde(default)]
     pub geolocation: Option<String>,
     /// Inbound bandwidth price ($/TB). vast.ai bills Docker image pulls here.
@@ -58,6 +61,8 @@ pub struct LabeledInstance {
 pub struct SelectionPolicy {
     pub gpu_name: Option<String>,
     pub min_gpu_ram_mb: Option<u64>,
+    /// Minimum CUDA compute capability encoded as vast.ai reports it (`700` = 7.0).
+    pub min_compute_cap: Option<u64>,
     pub min_reliability: f64,
     pub require_verified: bool,
     pub min_down_mbps: f64,
@@ -73,6 +78,7 @@ impl Default for SelectionPolicy {
         Self {
             gpu_name: None,
             min_gpu_ram_mb: None,
+            min_compute_cap: Some(700),
             min_reliability: 0.95,
             require_verified: false,
             min_down_mbps: 100.0,
@@ -90,6 +96,8 @@ impl Default for SelectionPolicy {
 pub struct LifecyclePolicy {
     pub lease_pace: Duration,
     pub poll_interval: Duration,
+    /// Maximum time to stay in one non-running provider state before replacing the lease.
+    pub state_timeout: Duration,
 }
 
 impl Default for LifecyclePolicy {
@@ -97,6 +105,7 @@ impl Default for LifecyclePolicy {
         Self {
             lease_pace: Duration::from_millis(600),
             poll_interval: Duration::from_secs(10),
+            state_timeout: Duration::from_secs(300),
         }
     }
 }
diff --git a/xtask/src/main.rs b/xtask/src/main.rs
index cb05f76..f236c5b 100644
--- a/xtask/src/main.rs
+++ b/xtask/src/main.rs
@@ -18,9 +18,9 @@ struct TestStep {
     args: &'static [&'static str],
 }
 
-const MVP_CHAT_CHECK_TIMEOUT_SECS: u64 = 900;
+const MVP_CHAT_CHECK_TIMEOUT_SECS: u64 = 1_800;
 const MVP_CHAT_CHECK_POLL_MS: u64 = 100;
-const MVP_CHAT_CHECK_TERM_GRACE_MS: u64 = 2_000;
+const MVP_CHAT_CHECK_TERM_GRACE_MS: u64 = 30_000;
 const MVP_CHAT_CHECK_PROMPTS: &[u8] = b"ping\nsecond prompt\n";
 const DATA_PATH_MIN_PAYLOAD_BYTES: u64 = 512;
 
@@ -143,7 +143,6 @@ impl MvpChatCheckInvocation {
                 "--yes".to_owned(),
                 "--endpoint-addr-mask".to_owned(),
                 "relay-only".to_owned(),
-                "--skip-rebuild".to_owned(),
             ]);
         }
         args.extend([
@@ -817,19 +816,26 @@ fn run_mvp_chat_check(args: Vec<String>) -> ExitCode {
             );
         }
     };
-    let summary =
-        match build_benchmark_summary(&events, output.child_elapsed_ms, run_id, scenario, &paths) {
-            Ok(summary) => summary,
-            Err(error) => {
-                return fail_mvp_chat_check(
-                    &error,
-                    &paths,
-                    &output.stdout,
-                    &output.stderr,
-                    Some(&output.status),
-                );
-            }
-        };
+    let summary = match build_benchmark_summary(
+        &events,
+        output.child_elapsed_ms,
+        run_id,
+        scenario,
+        &paths,
+        u64::try_from(output.stdout.len()).unwrap_or(u64::MAX),
+        u64::try_from(output.stderr.len()).unwrap_or(u64::MAX),
+    ) {
+        Ok(summary) => summary,
+        Err(error) => {
+            return fail_mvp_chat_check(
+                &error,
+                &paths,
+                &output.stdout,
+                &output.stderr,
+                Some(&output.status),
+            );
+        }
+    };
     if let Err(error) = write_benchmark_artifacts(&paths, run_id, scenario, &output, &summary) {
         return fail_mvp_chat_check(
             &error,
@@ -1005,6 +1011,9 @@ fn fail_mvp_chat_check(
     if let Some(status) = status {
         eprintln!("mvp-chat-check: child exit status: {status}");
     }
+    if let Err(error) = write_failure_artifacts(paths, reason, stdout, stderr, status) {
+        eprintln!("mvp-chat-check: warning: could not write failure artifacts: {error}");
+    }
     eprintln!(
         "mvp-chat-check: temp directory kept at {}",
         paths.root.display()
@@ -1030,6 +1039,66 @@ fn fail_mvp_chat_check(
     ExitCode::from(1)
 }
 
+fn write_failure_artifacts(
+    paths: &MvpChatCheckPaths,
+    reason: &str,
+    stdout: &str,
+    stderr: &str,
+    status: Option<&ExitStatus>,
+) -> Result<(), String> {
+    fs::write(&paths.stdout, stdout).map_err(|e| {
+        format!(
+            "mvp-chat-check: write stdout artifact {}: {e}",
+            paths.stdout.display()
+        )
+    })?;
+    fs::write(&paths.stderr, stderr).map_err(|e| {
+        format!(
+            "mvp-chat-check: write stderr artifact {}: {e}",
+            paths.stderr.display()
+        )
+    })?;
+    fs::write(&paths.prompts, MVP_CHAT_CHECK_PROMPTS).map_err(|e| {
+        format!(
+            "mvp-chat-check: write prompt corpus {}: {e}",
+            paths.prompts.display()
+        )
+    })?;
+    let summary = json!({
+        "schema": "swactor.mvp_chat_check.failure.v1",
+        "status": "failed",
+        "reason": reason,
+        "child_status": status.map(|status| status.to_string()),
+        "created_unix_ms": unix_ms_now(),
+        "artifacts": {
+            "root": paths.root.display().to_string(),
+            "datastream": {
+                "path": paths.dump_log.display().to_string(),
+                "exists": paths.dump_log.is_file(),
+                "bytes": file_len(&paths.dump_log),
+            },
+            "stdout": {
+                "path": paths.stdout.display().to_string(),
+                "bytes": stdout.len(),
+            },
+            "stderr": {
+                "path": paths.stderr.display().to_string(),
+                "bytes": stderr.len(),
+            },
+            "prompts": {
+                "path": paths.prompts.display().to_string(),
+                "bytes": MVP_CHAT_CHECK_PROMPTS.len(),
+                "blake3": bytes_blake3_hex(MVP_CHAT_CHECK_PROMPTS),
+            },
+        },
+    });
+    write_json_file(&paths.summary, &summary)
+}
+
+fn file_len(path: &Path) -> Option<u64> {
+    fs::metadata(path).ok().map(|metadata| metadata.len())
+}
+
 fn assert_stdout_contains_two_prompt_cycles(stdout: &str) -> Result<Vec<String>, String> {
     let decoding_count = stdout.matches("decoding...").count();
     if decoding_count < 2 {
@@ -1818,12 +1887,20 @@ fn build_benchmark_summary(
     run_id: u64,
     scenario: MvpChatCheckScenario,
     paths: &MvpChatCheckPaths,
+    stdout_bytes: u64,
+    stderr_bytes: u64,
 ) -> Result<Value, String> {
     let facts = BenchmarkFacts::from_events(events, run_id);
     let mut dump_facts = DumpLogFacts::default();
     for record in events {
         record_dump_log_event(&record.channel, &record.event, &mut dump_facts)?;
     }
+    let datastream_bytes = file_size(&paths.dump_log)?;
+    let prompt_bytes = u64::try_from(MVP_CHAT_CHECK_PROMPTS.len()).unwrap_or(u64::MAX);
+    let known_artifact_bytes = datastream_bytes
+        .saturating_add(stdout_bytes)
+        .saturating_add(stderr_bytes)
+        .saturating_add(prompt_bytes);
     let run_envelope = benchmark_run_envelope(events, run_id);
     let channel_counts = benchmark_channel_counts(events);
     let event_counts = benchmark_event_counts(events);
@@ -1848,18 +1925,23 @@ fn build_benchmark_summary(
         },
         "artifacts": {
             "root": paths.root.display().to_string(),
+            "known_total_bytes": known_artifact_bytes,
             "datastream": {
                 "path": paths.dump_log.display().to_string(),
+                "bytes": datastream_bytes,
                 "blake3": file_blake3_hex(&paths.dump_log)?,
             },
             "stdout": {
                 "path": paths.stdout.display().to_string(),
+                "bytes": stdout_bytes,
             },
             "stderr": {
                 "path": paths.stderr.display().to_string(),
+                "bytes": stderr_bytes,
             },
             "prompts": {
                 "path": paths.prompts.display().to_string(),
+                "bytes": prompt_bytes,
                 "blake3": bytes_blake3_hex(MVP_CHAT_CHECK_PROMPTS),
             },
             "redacted_config": {
@@ -1891,6 +1973,7 @@ fn build_benchmark_summary(
         },
         "pipeline": pipeline_summary_json(events, &dump_facts),
         "gpu": gpu_summary_json(events, &dump_facts),
+        "vastai": vastai_summary_json(events),
         "invariants": benchmark_invariants_json(&dump_facts, scenario),
         "legacy_tolerance": legacy_tolerance_summary(events),
     });
@@ -2092,6 +2175,7 @@ fn duration_summary_json(duration: DurationRender) -> Value {
 fn pipeline_summary_json(events: &[DumpLogEvent], facts: &DumpLogFacts) -> Value {
     let mut worker_steps = BenchmarkAggregate::default();
     let mut object_loads = BenchmarkAggregate::default();
+    let mut worker_steps_by_stage = BTreeMap::<u64, BenchmarkAggregate>::new();
     let mut ring_installs = BTreeMap::<String, u64>::new();
     for record in events {
         match (
@@ -2099,20 +2183,16 @@ fn pipeline_summary_json(events: &[DumpLogEvent], facts: &DumpLogFacts) -> Value
             record.event.get("type").and_then(Value::as_str),
         ) {
             ("mvp.worker.step", Some("StepExecuted")) => {
-                worker_steps.count += 1;
-                worker_steps.elapsed_ms += event_u64(&record.event, "elapsed_ms").unwrap_or(0);
-                worker_steps.stage_execution_ms +=
-                    event_u64(&record.event, "stage_execution_ms").unwrap_or(0);
-                worker_steps.record_write_ms +=
-                    event_u64(&record.event, "record_write_ms").unwrap_or(0);
-                worker_steps.payload_bytes +=
-                    event_u64(&record.event, "payload_bytes").unwrap_or(0);
-                worker_steps.record_bytes += event_u64(&record.event, "record_bytes").unwrap_or(0);
+                worker_steps.observe_step_event(&record.event);
+                if let Some(stage_index) = pipeline_stage_index(&record.event) {
+                    worker_steps_by_stage
+                        .entry(stage_index)
+                        .or_default()
+                        .observe_step_event(&record.event);
+                }
             }
             ("mvp.worker.ingress", Some("ObjectLoaded")) => {
-                object_loads.count += 1;
-                object_loads.elapsed_ms += event_u64(&record.event, "elapsed_ms").unwrap_or(0);
-                object_loads.record_bytes += event_u64(&record.event, "extent").unwrap_or(0);
+                object_loads.observe_object_load_event(&record.event);
             }
             ("mvp.worker.ring", Some("RingInstalled")) => {
                 let direction = record
@@ -2126,10 +2206,18 @@ fn pipeline_summary_json(events: &[DumpLogEvent], facts: &DumpLogFacts) -> Value
             _ => {}
         }
     }
+    let worker_steps_by_stage = worker_steps_by_stage
+        .into_iter()
+        .map(|(stage_index, stats)| stage_worker_summary_json(stage_index, &stats))
+        .collect::<Vec<_>>();
     json!({
         "worker_steps": worker_steps.to_json(),
+        "worker_steps_by_stage": worker_steps_by_stage,
         "object_loads": object_loads.to_json(),
         "ring_installs": ring_installs,
+        "prompt_critical_paths": prompt_pipeline_critical_summary_json(events),
+        "edge_handoffs": pipeline_edge_handoff_summary_json(events),
+        "provisioning": stage_provisioning_summary_json(events),
         "data_path": {
             "activation_object_loaded": facts.activation_object_loaded,
             "activation_step_executed": facts.activation_step_executed,
@@ -2147,6 +2235,381 @@ fn pipeline_summary_json(events: &[DumpLogEvent], facts: &DumpLogFacts) -> Value
     })
 }
 
+fn stage_worker_summary_json(stage_index: u64, stats: &BenchmarkAggregate) -> Value {
+    json!({
+        "stage_index": stage_index,
+        "stats": stats.to_json(),
+    })
+}
+
+#[derive(Default)]
+struct PromptPipelineCriticalStats {
+    request_id: u64,
+    token_in_started: u64,
+    token_in_ready: u64,
+    token_out_observed: u64,
+    tokenizer_decode_ready: u64,
+    first_sequence: Option<u64>,
+    last_sequence: Option<u64>,
+    last_token_out_arrival_ms: Option<u64>,
+    token_out_interval_ms: MetricStats,
+    token_in_to_out_ms: MetricStats,
+    token_in_started_by_sequence: BTreeMap<u64, u64>,
+}
+
+impl PromptPipelineCriticalStats {
+    fn new(request_id: u64) -> Self {
+        Self {
+            request_id,
+            ..Self::default()
+        }
+    }
+
+    fn observe_sequence(&mut self, sequence: u64) {
+        self.first_sequence = Some(
+            self.first_sequence
+                .map_or(sequence, |value| value.min(sequence)),
+        );
+        self.last_sequence = Some(
+            self.last_sequence
+                .map_or(sequence, |value| value.max(sequence)),
+        );
+    }
+
+    fn observe_token_in_started(&mut self, sequence: u64, arrival_unix_ms: Option<u64>) {
+        self.token_in_started += 1;
+        self.observe_sequence(sequence);
+        if let Some(arrival_unix_ms) = arrival_unix_ms {
+            self.token_in_started_by_sequence
+                .entry(sequence)
+                .or_insert(arrival_unix_ms);
+        }
+    }
+
+    fn observe_token_in_ready(&mut self, sequence: u64) {
+        self.token_in_ready += 1;
+        self.observe_sequence(sequence);
+    }
+
+    fn observe_token_out(&mut self, sequence: u64, arrival_unix_ms: Option<u64>) {
+        self.token_out_observed += 1;
+        self.observe_sequence(sequence);
+        if let Some(arrival_unix_ms) = arrival_unix_ms {
+            if let Some(last) = self.last_token_out_arrival_ms {
+                self.token_out_interval_ms
+                    .observe(arrival_unix_ms.saturating_sub(last));
+            }
+            self.last_token_out_arrival_ms = Some(arrival_unix_ms);
+            if let Some(started) = self.token_in_started_by_sequence.get(&sequence) {
+                self.token_in_to_out_ms
+                    .observe(arrival_unix_ms.saturating_sub(*started));
+            }
+        }
+    }
+
+    fn observe_tokenizer_decode_ready(&mut self) {
+        self.tokenizer_decode_ready += 1;
+    }
+
+    fn to_json(&self) -> Value {
+        json!({
+            "request_id": self.request_id,
+            "token_in_started": self.token_in_started,
+            "token_in_ready": self.token_in_ready,
+            "token_out_observed": self.token_out_observed,
+            "tokenizer_decode_ready": self.tokenizer_decode_ready,
+            "first_sequence": self.first_sequence,
+            "last_sequence": self.last_sequence,
+            "token_out_interval_ms": self.token_out_interval_ms.to_json(),
+            "token_in_to_out_ms": self.token_in_to_out_ms.to_json(),
+        })
+    }
+}
+
+fn prompt_pipeline_critical_summary_json(events: &[DumpLogEvent]) -> Vec<Value> {
+    let mut prompts = BTreeMap::<u64, PromptPipelineCriticalStats>::new();
+    for record in events {
+        if record.channel != "mvp.orch.prompt"
+            || record.event.get("type").and_then(Value::as_str) != Some("OrchPromptEvent")
+        {
+            continue;
+        }
+        let Some(request_id) = benchmark_request_id(&record.event) else {
+            continue;
+        };
+        let Some(phase) = record.event.get("phase").and_then(Value::as_str) else {
+            continue;
+        };
+        let status = record.event.get("status").and_then(Value::as_str);
+        let sequence = detail_u64(&record.event, "sequence");
+        let prompt = prompts
+            .entry(request_id)
+            .or_insert_with(|| PromptPipelineCriticalStats::new(request_id));
+        match (phase, status, sequence) {
+            ("pipeline_token_in", Some("started"), Some(sequence)) => {
+                prompt.observe_token_in_started(sequence, record.arrival_unix_ms);
+            }
+            ("pipeline_token_in", Some("ready"), Some(sequence)) => {
+                prompt.observe_token_in_ready(sequence);
+            }
+            ("pipeline_token_out", Some("observed"), Some(sequence)) => {
+                prompt.observe_token_out(sequence, record.arrival_unix_ms);
+            }
+            ("pipeline_tokenizer_decode", Some("ready"), _) => {
+                prompt.observe_tokenizer_decode_ready();
+            }
+            _ => {}
+        }
+    }
+    prompts
+        .values()
+        .map(PromptPipelineCriticalStats::to_json)
+        .collect()
+}
+
+#[derive(Default)]
+struct PipelineEdgeHandoffStats {
+    edge_id: u64,
+    producer_stages: BTreeSet<u64>,
+    consumer_stages: BTreeSet<u64>,
+    producer_ring_reads: u64,
+    producer_sends: u64,
+    consumer_stream_reads: u64,
+    consumer_ring_writes: u64,
+    consumer_object_loads: u64,
+    record_bytes: MetricStats,
+    network_read_bytes: MetricStats,
+    helper_execute_ms: MetricStats,
+    egress_ring_read_ms: MetricStats,
+    send_ms: MetricStats,
+    ingress_ring_write_ms: MetricStats,
+    object_load_ms: MetricStats,
+}
+
+impl PipelineEdgeHandoffStats {
+    fn observe_node_stage_event(&mut self, event: &Value) {
+        let detail = event.get("detail").unwrap_or(&Value::Null);
+        let phase = event.get("phase").and_then(Value::as_str);
+        if let Some(stage_index) = event.get("stage_index").and_then(Value::as_u64) {
+            match phase {
+                Some("egress_ring_read") | Some("iroh_edge_bytes_sent") => {
+                    self.producer_stages.insert(stage_index);
+                }
+                Some("iroh_edge_bytes_read")
+                | Some("ingress_ring_write")
+                | Some("object_loaded") => {
+                    self.consumer_stages.insert(stage_index);
+                }
+                _ => {}
+            }
+        }
+        match phase {
+            Some("egress_ring_read") => {
+                self.producer_ring_reads += 1;
+                self.record_bytes
+                    .observe_optional(detail.get("record_bytes").and_then(Value::as_u64));
+                self.helper_execute_ms
+                    .observe_optional(detail.get("helper_execute_ms").and_then(Value::as_u64));
+                self.egress_ring_read_ms
+                    .observe_optional(detail.get("egress_ring_read_ms").and_then(Value::as_u64));
+            }
+            Some("iroh_edge_bytes_sent") => {
+                self.producer_sends += 1;
+                self.record_bytes
+                    .observe_optional(detail.get("record_bytes").and_then(Value::as_u64));
+                self.send_ms
+                    .observe_optional(detail.get("send_ms").and_then(Value::as_u64));
+            }
+            Some("iroh_edge_bytes_read") => {
+                self.consumer_stream_reads += 1;
+                self.network_read_bytes
+                    .observe_optional(detail.get("bytes").and_then(Value::as_u64));
+            }
+            Some("ingress_ring_write") => {
+                self.consumer_ring_writes += 1;
+                self.record_bytes
+                    .observe_optional(detail.get("record_bytes").and_then(Value::as_u64));
+                self.ingress_ring_write_ms
+                    .observe_optional(detail.get("ingress_ring_write_ms").and_then(Value::as_u64));
+            }
+            Some("object_loaded") => {
+                self.consumer_object_loads += 1;
+                self.object_load_ms
+                    .observe_optional(detail.get("object_load_ms").and_then(Value::as_u64));
+            }
+            _ => {}
+        }
+    }
+
+    fn to_json(&self) -> Value {
+        json!({
+            "edge_id": self.edge_id,
+            "producer_stages": self.producer_stages.iter().copied().collect::<Vec<_>>(),
+            "consumer_stages": self.consumer_stages.iter().copied().collect::<Vec<_>>(),
+            "producer_ring_reads": self.producer_ring_reads,
+            "producer_sends": self.producer_sends,
+            "consumer_stream_reads": self.consumer_stream_reads,
+            "consumer_ring_writes": self.consumer_ring_writes,
+            "consumer_object_loads": self.consumer_object_loads,
+            "record_bytes": self.record_bytes.to_json(),
+            "network_read_bytes": self.network_read_bytes.to_json(),
+            "helper_execute_ms": self.helper_execute_ms.to_json(),
+            "egress_ring_read_ms": self.egress_ring_read_ms.to_json(),
+            "send_ms": self.send_ms.to_json(),
+            "ingress_ring_write_ms": self.ingress_ring_write_ms.to_json(),
+            "object_load_ms": self.object_load_ms.to_json(),
+        })
+    }
+}
+
+fn pipeline_edge_handoff_summary_json(events: &[DumpLogEvent]) -> Vec<Value> {
+    let mut edges = BTreeMap::<u64, PipelineEdgeHandoffStats>::new();
+    for record in events {
+        if record.channel != "mvp.node.stage"
+            || record.event.get("type").and_then(Value::as_str) != Some("NodeEvent")
+        {
+            continue;
+        }
+        let Some(edge_id) = detail_u64(&record.event, "edge_id") else {
+            continue;
+        };
+        let edge = edges
+            .entry(edge_id)
+            .or_insert_with(|| PipelineEdgeHandoffStats {
+                edge_id,
+                ..PipelineEdgeHandoffStats::default()
+            });
+        edge.observe_node_stage_event(&record.event);
+    }
+    edges
+        .values()
+        .map(PipelineEdgeHandoffStats::to_json)
+        .collect()
+}
+
+#[derive(Default)]
+struct StageProvisionStats {
+    send_events: u64,
+    first_attempt: Option<u64>,
+    last_attempt: Option<u64>,
+    min_loaded_stage_count: Option<u64>,
+    max_loaded_stage_count: Option<u64>,
+    stage_count: Option<u64>,
+}
+
+impl StageProvisionStats {
+    fn observe(&mut self, detail: &Value) {
+        self.send_events += 1;
+        if let Some(attempt) = detail.get("attempt").and_then(Value::as_u64) {
+            if self.first_attempt.is_none() {
+                self.first_attempt = Some(attempt);
+            }
+            self.last_attempt = Some(attempt);
+        }
+        if let Some(loaded) = detail.get("loaded_stage_count").and_then(Value::as_u64) {
+            self.min_loaded_stage_count = Some(
+                self.min_loaded_stage_count
+                    .map_or(loaded, |min| min.min(loaded)),
+            );
+            self.max_loaded_stage_count = Some(
+                self.max_loaded_stage_count
+                    .map_or(loaded, |max| max.max(loaded)),
+            );
+        }
+        if self.stage_count.is_none() {
+            self.stage_count = detail.get("stage_count").and_then(Value::as_u64);
+        }
+    }
+
+    fn to_json(&self, stage_index: u64) -> Value {
+        json!({
+            "stage_index": stage_index,
+            "send_events": self.send_events,
+            "first_attempt": self.first_attempt,
+            "last_attempt": self.last_attempt,
+            "min_loaded_stage_count": self.min_loaded_stage_count,
+            "max_loaded_stage_count": self.max_loaded_stage_count,
+            "stage_count": self.stage_count,
+        })
+    }
+}
+
+fn stage_provisioning_summary_json(events: &[DumpLogEvent]) -> Value {
+    let mut send_events = 0_u64;
+    let mut stages = BTreeMap::<u64, StageProvisionStats>::new();
+    let mut latest_wait = None;
+    let mut provider_event_counts = BTreeMap::<String, u64>::new();
+    let mut ssh_retry_counts = BTreeMap::<String, u64>::new();
+    let mut ssh_failure_counts = BTreeMap::<String, u64>::new();
+    for record in events {
+        if record.channel == "mvp.provisioning.events"
+            && let Some(kind) = record
+                .event
+                .get("event")
+                .and_then(|event| event.get("kind"))
+                .and_then(Value::as_str)
+        {
+            *provider_event_counts.entry(kind.to_owned()).or_default() += 1;
+        }
+        if record.channel.starts_with("mvp.provisioning.logs.node.") {
+            let node = record
+                .event
+                .get("node_id")
+                .and_then(Value::as_u64)
+                .map(|node_id| node_id.to_string())
+                .unwrap_or_else(|| "unknown".to_owned());
+            if let Some(line) = record.event.get("line").and_then(Value::as_str) {
+                if line.contains("VastAI SSH bootstrap retrying") {
+                    *ssh_retry_counts.entry(node.clone()).or_default() += 1;
+                }
+                if line.contains("VastAI SSH bootstrap failed before runtime ready") {
+                    *ssh_failure_counts.entry(node).or_default() += 1;
+                }
+            }
+        }
+        if record.channel != "mvp.orch.bootstrap"
+            || record.event.get("type").and_then(Value::as_str) != Some("OrchBootstrap")
+            || record.event.get("phase").and_then(Value::as_str) != Some("stage_provision_send")
+            || record.event.get("status").and_then(Value::as_str) != Some("sent")
+        {
+            continue;
+        }
+        let detail = record.event.get("detail").unwrap_or(&Value::Null);
+        let Some(stage_index) = detail.get("stage_index").and_then(Value::as_u64) else {
+            continue;
+        };
+        send_events += 1;
+        stages.entry(stage_index).or_default().observe(detail);
+        latest_wait = Some(json!({
+            "attempt": detail.get("attempt").and_then(Value::as_u64),
+            "waiting_stage_index": stage_index,
+            "loaded_stage_count": detail.get("loaded_stage_count").and_then(Value::as_u64),
+            "stage_count": detail.get("stage_count").and_then(Value::as_u64),
+            "stage_send_count": detail.get("stage_send_count").and_then(Value::as_u64),
+        }));
+    }
+    let max_send_events_for_stage = stages
+        .values()
+        .map(|stage| stage.send_events)
+        .max()
+        .unwrap_or(0);
+    let stages = stages
+        .into_iter()
+        .map(|(stage_index, stats)| stats.to_json(stage_index))
+        .collect::<Vec<_>>();
+    json!({
+        "send_events": send_events,
+        "max_send_events_for_stage": max_send_events_for_stage,
+        "latest_wait": latest_wait.unwrap_or(Value::Null),
+        "stages": stages,
+        "provider_event_counts": provider_event_counts,
+        "ssh_bootstrap": {
+            "retry_counts_by_node": ssh_retry_counts,
+            "failure_counts_by_node": ssh_failure_counts,
+        },
+    })
+}
+
 fn gpu_summary_json(events: &[DumpLogEvent], facts: &DumpLogFacts) -> Value {
     let mut cpu_profile_summaries = Vec::new();
     for record in events {
@@ -2164,34 +2627,244 @@ fn gpu_summary_json(events: &[DumpLogEvent], facts: &DumpLogFacts) -> Value {
         "first_token_request_ids": facts.gpu_first_token_ready,
         "decode_ready_request_ids": facts.gpu_decode_ready,
         "prompt_completed_request_ids": facts.gpu_prompt_completed,
+        "pipeline_prompt_encoded_request_ids": facts.gpu_pipeline_prompt_encoded,
+        "pipeline_prompt_begin_request_ids": facts.gpu_pipeline_prompt_begin,
+        "pipeline_token_in_request_ids": facts.gpu_pipeline_token_in,
+        "pipeline_token_out_request_ids": facts.gpu_pipeline_token_out,
+        "pipeline_tokenizer_decode_ready_request_ids": facts.gpu_pipeline_tokenizer_decode_ready,
+        "pipeline_tokens_decoded_request_ids": facts.gpu_pipeline_tokens_decoded,
         "pipeline_real_worker_step_seen": facts.gpu_pipeline_real_worker_step_seen,
         "cpu_profile_summaries": cpu_profile_summaries,
+        "host_gpu_samples": host_gpu_sample_summary_json(events),
     })
 }
 
-#[derive(Default)]
+fn host_gpu_sample_summary_json(events: &[DumpLogEvent]) -> Value {
+    let mut sample_count = 0_u64;
+    let mut error_count = 0_u64;
+    let mut device_sample_count = 0_u64;
+    let mut max_utilization_gpu_percent = MetricStats::default();
+    let mut max_memory_used_mib = MetricStats::default();
+    let mut max_memory_total_mib = MetricStats::default();
+    for record in events {
+        if record.channel != "host.gpu" {
+            continue;
+        }
+        sample_count += 1;
+        if record.event.get("error").and_then(Value::as_str).is_some() {
+            error_count += 1;
+        }
+        let Some(gpus) = record.event.get("gpus").and_then(Value::as_array) else {
+            continue;
+        };
+        device_sample_count = device_sample_count.saturating_add(gpus.len() as u64);
+        for gpu in gpus {
+            max_utilization_gpu_percent
+                .observe_optional(gpu.get("utilization_gpu_percent").and_then(Value::as_u64));
+            max_memory_used_mib
+                .observe_optional(gpu.get("memory_used_mib").and_then(Value::as_u64));
+            max_memory_total_mib
+                .observe_optional(gpu.get("memory_total_mib").and_then(Value::as_u64));
+        }
+    }
+    json!({
+        "sample_count": sample_count,
+        "error_count": error_count,
+        "device_sample_count": device_sample_count,
+        "utilization_gpu_percent": max_utilization_gpu_percent.to_json(),
+        "memory_used_mib": max_memory_used_mib.to_json(),
+        "memory_total_mib": max_memory_total_mib.to_json(),
+    })
+}
+
+struct VastAiLeaseSelection {
+    node_id: Option<u64>,
+    index: u64,
+    offer_id: u64,
+    gpu_name: String,
+    gpu_ram_mb: Option<u64>,
+    dph_total: f64,
+    geolocation: Option<String>,
+    host_id: Option<u64>,
+    effective_dph_total: f64,
+}
+
+impl VastAiLeaseSelection {
+    fn to_json(&self) -> Value {
+        json!({
+            "node_id": self.node_id,
+            "index": self.index,
+            "offer_id": self.offer_id,
+            "gpu_name": self.gpu_name,
+            "gpu_ram_mb": self.gpu_ram_mb,
+            "dph_total": self.dph_total,
+            "geolocation": self.geolocation,
+            "host_id": self.host_id,
+            "effective_dph_total": self.effective_dph_total,
+        })
+    }
+}
+
+fn vastai_summary_json(events: &[DumpLogEvent]) -> Value {
+    let selections = events
+        .iter()
+        .filter_map(parse_vastai_lease_selection)
+        .collect::<Vec<_>>();
+    let mut max_dph_total: Option<f64> = None;
+    for selection in &selections {
+        max_dph_total =
+            Some(max_dph_total.map_or(selection.dph_total, |max| max.max(selection.dph_total)));
+    }
+    json!({
+        "selected_lease_count": selections.len(),
+        "max_dph_total": max_dph_total,
+        "selected_leases": selections
+            .iter()
+            .map(VastAiLeaseSelection::to_json)
+            .collect::<Vec<_>>(),
+    })
+}
+
+fn parse_vastai_lease_selection(record: &DumpLogEvent) -> Option<VastAiLeaseSelection> {
+    let line = record.event.get("line").and_then(Value::as_str)?;
+    let rest = line.strip_prefix("lease_chain: index ")?;
+    let (index, rest) = rest.split_once(" → offer ")?;
+    let (offer_id, rest) = rest.split_once(" — ")?;
+    let (gpu_and_ram, rest) = rest.split_once(" @ $")?;
+    let (gpu_name, gpu_ram) = gpu_and_ram.rsplit_once(' ')?;
+    let (dph_total, rest) = rest.split_once("/hr [")?;
+    let (geolocation, rest) = rest.split_once("] host ")?;
+    let (host_id, effective_dph_total) = rest.split_once(" eff $")?;
+    let effective_dph_total = effective_dph_total.strip_suffix("/hr")?;
+    Some(VastAiLeaseSelection {
+        node_id: record.event.get("node_id").and_then(Value::as_u64),
+        index: index.parse().ok()?,
+        offer_id: offer_id.parse().ok()?,
+        gpu_name: gpu_name.to_owned(),
+        gpu_ram_mb: gpu_ram
+            .strip_suffix("MB")
+            .and_then(|value| value.parse().ok()),
+        dph_total: dph_total.parse().ok()?,
+        geolocation: (!geolocation.is_empty()).then(|| geolocation.to_owned()),
+        host_id: host_id.parse().ok(),
+        effective_dph_total: effective_dph_total.parse().ok()?,
+    })
+}
+
+#[derive(Clone, Default)]
+struct MetricStats {
+    count: u64,
+    sum: u64,
+    min: Option<u64>,
+    max: Option<u64>,
+}
+
+impl MetricStats {
+    fn observe(&mut self, value: u64) {
+        self.count += 1;
+        self.sum = self.sum.saturating_add(value);
+        self.min = Some(self.min.map_or(value, |min| min.min(value)));
+        self.max = Some(self.max.map_or(value, |max| max.max(value)));
+    }
+
+    fn observe_optional(&mut self, value: Option<u64>) {
+        if let Some(value) = value {
+            self.observe(value);
+        }
+    }
+
+    fn avg(&self) -> Option<f64> {
+        (self.count != 0).then(|| self.sum as f64 / self.count as f64)
+    }
+
+    fn to_json(&self) -> Value {
+        json!({
+            "count": self.count,
+            "sum": self.sum,
+            "min": self.min,
+            "max": self.max,
+            "avg": self.avg(),
+        })
+    }
+}
+
+#[derive(Clone, Default)]
 struct BenchmarkAggregate {
     count: u64,
     elapsed_ms: u64,
     stage_execution_ms: u64,
     record_write_ms: u64,
+    input_prepare_ms: u64,
+    model_forward_ms: u64,
+    output_realize_ms: u64,
+    payload_pack_ms: u64,
     payload_bytes: u64,
     record_bytes: u64,
 }
 
 impl BenchmarkAggregate {
+    fn observe_step_event(&mut self, event: &Value) {
+        self.count += 1;
+        self.elapsed_ms = self
+            .elapsed_ms
+            .saturating_add(event_u64(event, "elapsed_ms").unwrap_or(0));
+        self.stage_execution_ms = self
+            .stage_execution_ms
+            .saturating_add(event_u64(event, "stage_execution_ms").unwrap_or(0));
+        self.record_write_ms = self
+            .record_write_ms
+            .saturating_add(event_u64(event, "record_write_ms").unwrap_or(0));
+        self.input_prepare_ms = self
+            .input_prepare_ms
+            .saturating_add(event_u64(event, "input_prepare_ms").unwrap_or(0));
+        self.model_forward_ms = self
+            .model_forward_ms
+            .saturating_add(event_u64(event, "model_forward_ms").unwrap_or(0));
+        self.output_realize_ms = self
+            .output_realize_ms
+            .saturating_add(event_u64(event, "output_realize_ms").unwrap_or(0));
+        self.payload_pack_ms = self
+            .payload_pack_ms
+            .saturating_add(event_u64(event, "payload_pack_ms").unwrap_or(0));
+        self.payload_bytes = self
+            .payload_bytes
+            .saturating_add(event_u64(event, "payload_bytes").unwrap_or(0));
+        self.record_bytes = self
+            .record_bytes
+            .saturating_add(event_u64(event, "record_bytes").unwrap_or(0));
+    }
+
+    fn observe_object_load_event(&mut self, event: &Value) {
+        self.count += 1;
+        self.elapsed_ms = self
+            .elapsed_ms
+            .saturating_add(event_u64(event, "elapsed_ms").unwrap_or(0));
+        self.record_bytes = self
+            .record_bytes
+            .saturating_add(event_u64(event, "extent").unwrap_or(0));
+    }
+
     fn to_json(&self) -> Value {
         json!({
             "count": self.count,
             "elapsed_ms_sum": self.elapsed_ms,
             "stage_execution_ms_sum": self.stage_execution_ms,
             "record_write_ms_sum": self.record_write_ms,
+            "input_prepare_ms_sum": self.input_prepare_ms,
+            "model_forward_ms_sum": self.model_forward_ms,
+            "output_realize_ms_sum": self.output_realize_ms,
+            "payload_pack_ms_sum": self.payload_pack_ms,
             "payload_bytes_sum": self.payload_bytes,
             "record_bytes_sum": self.record_bytes,
         })
     }
 }
 
+fn pipeline_stage_index(event: &Value) -> Option<u64> {
+    event_u64(event, "stage_index")
+        .or_else(|| event_u64(event, "role_id").and_then(|role| role.checked_sub(1)))
+}
+
 fn benchmark_invariants_json(facts: &DumpLogFacts, scenario: MvpChatCheckScenario) -> Vec<Value> {
     let mut invariants = vec![
         invariant_json("chat_config_ready", facts.chat_config_ready),
@@ -2234,14 +2907,6 @@ fn benchmark_invariants_json(facts: &DumpLogFacts, scenario: MvpChatCheckScenari
                 "activation_large_object_step_executed",
                 facts.activation_step_executed,
             ),
-            invariant_json(
-                "activation_large_object_iroh_sent",
-                facts.activation_iroh_edge_sent,
-            ),
-            invariant_json(
-                "activation_large_object_iroh_read",
-                facts.activation_iroh_edge_read,
-            ),
         ]);
     }
     invariants
@@ -2329,6 +2994,12 @@ fn file_blake3_hex(path: &Path) -> Result<String, String> {
     Ok(bytes_blake3_hex(&bytes))
 }
 
+fn file_size(path: &Path) -> Result<u64, String> {
+    fs::metadata(path)
+        .map(|metadata| metadata.len())
+        .map_err(|e| format!("read artifact metadata {}: {e}", path.display()))
+}
+
 fn bytes_blake3_hex(bytes: &[u8]) -> String {
     blake3::hash(bytes).to_hex().to_string()
 }
@@ -2832,8 +3503,7 @@ fn require_gpu_dump_log_facts(facts: &DumpLogFacts) -> Result<(), String> {
             && facts.gpu_pipeline_token_out.contains(&request_id)
             && facts
                 .gpu_pipeline_tokenizer_decode_ready
-                .contains(&request_id)
-            && facts.gpu_pipeline_tokens_decoded.contains(&request_id);
+                .contains(&request_id);
         require_dump_log_fact(
             direct_decode || pipeline_decode,
             &format!("GPU decode/token evidence request_id={request_id}"),
@@ -3093,7 +3763,6 @@ mod tests {
                 "--yes",
                 "--endpoint-addr-mask",
                 "relay-only",
-                "--skip-rebuild",
                 "--run-id",
                 "42",
                 "--dump-logs=/tmp/mvp-chat-check.ndjson",
@@ -3113,7 +3782,6 @@ mod tests {
                 "--yes",
                 "--endpoint-addr-mask",
                 "relay-only",
-                "--skip-rebuild",
                 "--run-id",
                 "42",
                 "--dump-logs=/tmp/mvp-chat-check.ndjson",
@@ -3845,7 +4513,6 @@ mod tests {
             facts.gpu_pipeline_token_in.insert(request_id);
             facts.gpu_pipeline_token_out.insert(request_id);
             facts.gpu_pipeline_tokenizer_decode_ready.insert(request_id);
-            facts.gpu_pipeline_tokens_decoded.insert(request_id);
             if prompt_begin_markers {
                 facts.gpu_pipeline_prompt_begin.insert(request_id);
             }
@@ -4128,6 +4795,8 @@ mod tests {
             9,
             MvpChatCheckScenario::ProcessBaseline,
             &paths,
+            123,
+            45,
         )
         .expect("summary builds");
         let _ = fs::remove_file(path);
@@ -4160,6 +4829,296 @@ mod tests {
                 .and_then(Value::as_str)
                 .is_some()
         );
+        assert_eq!(
+            summary
+                .pointer("/artifacts/stdout/bytes")
+                .and_then(Value::as_u64),
+            Some(123)
+        );
+        assert_eq!(
+            summary
+                .pointer("/artifacts/stderr/bytes")
+                .and_then(Value::as_u64),
+            Some(45)
+        );
+    }
+
+    #[test]
+    fn benchmark_summary_exposes_vastai_pipeline_operator_summaries() {
+        let mut event_pairs = dump_log_fact_events(true, false);
+        event_pairs.extend([
+            (
+                "mvp.provisioning.logs.node.2.stderr",
+                json!({
+                    "run_id": 9,
+                    "node_id": 2,
+                    "stream": "Stderr",
+                    "line": "lease_chain: index 0 → offer 42528153 — RTX 2060 12288MB @ $0.036/hr [India, IN] host 581196 eff $0.036/hr",
+                }),
+            ),
+            (
+                "mvp.provisioning.logs.node.8.stderr",
+                json!({
+                    "run_id": 9,
+                    "node_id": 8,
+                    "stream": "Stderr",
+                    "line": "VastAI SSH bootstrap retrying in 3s after failed attempt",
+                }),
+            ),
+            (
+                "host.gpu",
+                json!({
+                    "schema":"host.gpu.v1",
+                    "seq":1,
+                    "sample_unix_ms":1_700,
+                    "query_elapsed_ms":4,
+                    "gpus":[{
+                        "index":0,
+                        "uuid":"GPU-unit",
+                        "name":"RTX 2060",
+                        "memory_used_mib":8120,
+                        "memory_total_mib":12288,
+                        "utilization_gpu_percent":73,
+                        "utilization_memory_percent":41,
+                        "temperature_c":61,
+                        "power_draw_w":120.0
+                    }],
+                    "processes":[],
+                    "error":null
+                }),
+            ),
+            (
+                "mvp.orch.bootstrap",
+                stamped(
+                    json!({"type":"OrchBootstrap","phase":"stage_provision_send","status":"sent","run_id":9,"node_id":1,"detail":{"attempt":1,"stage_count":8,"stage_index":7,"loaded_stage_count":0,"stage_send_count":1}}),
+                    "mvp-orchestrator",
+                    1_500,
+                    500,
+                ),
+            ),
+            (
+                "mvp.orch.bootstrap",
+                stamped(
+                    json!({"type":"OrchBootstrap","phase":"stage_provision_send","status":"sent","run_id":9,"node_id":1,"detail":{"attempt":2,"stage_count":8,"stage_index":7,"loaded_stage_count":0,"stage_send_count":2}}),
+                    "mvp-orchestrator",
+                    1_501,
+                    501,
+                ),
+            ),
+        ]);
+        for request_id in 1..=2 {
+            let base = 1_600 + request_id * 100;
+            event_pairs.extend([
+                (
+                    "mvp.worker.tokenizer",
+                    stamped(
+                        json!({"type":"PromptEncoded","run_id":9,"node_id":2,"stage_index":0,"request_id":request_id,"tokens":[1,2,3]}),
+                        "tinygrad-worker",
+                        base,
+                        base,
+                    ),
+                ),
+                (
+                    "mvp.orch.prompt",
+                    pipeline_prompt_event(
+                        "pipeline_token_in",
+                        "started",
+                        request_id,
+                        base,
+                        base,
+                        json!({"begin_sequence":true,"edge_id":1,"sequence":request_id}),
+                    ),
+                ),
+                (
+                    "mvp.orch.prompt",
+                    pipeline_prompt_event(
+                        "pipeline_token_in",
+                        "ready",
+                        request_id,
+                        base + 1,
+                        base + 1,
+                        json!({"begin_sequence":true,"edge_id":1,"sequence":request_id}),
+                    ),
+                ),
+                (
+                    "mvp.orch.prompt",
+                    pipeline_prompt_event(
+                        "pipeline_token_out",
+                        "observed",
+                        request_id,
+                        base + 2,
+                        base + 2,
+                        json!({"edge_id":9,"sequence":request_id,"token_id":7}),
+                    ),
+                ),
+                (
+                    "mvp.orch.prompt",
+                    pipeline_prompt_event(
+                        "pipeline_tokenizer_decode",
+                        "ready",
+                        request_id,
+                        base + 3,
+                        base + 3,
+                        json!({"text_bytes":1}),
+                    ),
+                ),
+            ]);
+        }
+        event_pairs.push((
+            "mvp.worker.step",
+            stamped(
+                json!({"type":"StepExecuted","run_id":9,"node_id":2,"stage_index":0,"execution_backend":"pipeline_stage","committed_bytes":4096}),
+                "tinygrad-worker",
+                1_900,
+                900,
+            ),
+        ));
+        event_pairs.extend([
+            (
+                "mvp.node.stage",
+                stamped(
+                    json!({"type":"NodeEvent","phase":"egress_ring_read","status":"ready","run_id":9,"node_id":2,"stage_index":0,"detail":{"edge_id":77,"edge_kind":"Activation","ring_id":11,"step_id":5,"sequence":3,"object_id":44,"record_bytes":4096,"helper_execute_ms":9,"egress_ring_read_ms":2}}),
+                    "mvp-worker-node",
+                    1_901,
+                    901,
+                ),
+            ),
+            (
+                "mvp.node.stage",
+                stamped(
+                    json!({"type":"NodeEvent","phase":"iroh_edge_bytes_sent","status":"ready","run_id":9,"node_id":2,"stage_index":0,"detail":{"edge_id":77,"edge_kind":"Activation","step_id":5,"sequence":3,"object_id":44,"record_bytes":4096,"send_ms":1}}),
+                    "mvp-worker-node",
+                    1_902,
+                    902,
+                ),
+            ),
+            (
+                "mvp.node.stage",
+                stamped(
+                    json!({"type":"NodeEvent","phase":"iroh_edge_bytes_read","status":"observed","run_id":9,"node_id":3,"stage_index":1,"detail":{"edge_id":77,"stream_id":1,"bytes":4096}}),
+                    "mvp-worker-node",
+                    1_903,
+                    903,
+                ),
+            ),
+            (
+                "mvp.node.stage",
+                stamped(
+                    json!({"type":"NodeEvent","phase":"ingress_ring_write","status":"ready","run_id":9,"node_id":3,"stage_index":1,"detail":{"edge_id":77,"edge_kind":"Activation","ring_id":12,"stream_id":1,"object_id":44,"sequence":3,"record_bytes":4096,"ingress_ring_write_ms":3}}),
+                    "mvp-worker-node",
+                    1_904,
+                    904,
+                ),
+            ),
+            (
+                "mvp.node.stage",
+                stamped(
+                    json!({"type":"NodeEvent","phase":"object_loaded","status":"ready","run_id":9,"node_id":3,"stage_index":1,"detail":{"edge_id":77,"edge_kind":"Activation","ring_id":12,"stream_id":1,"object_id":44,"sequence":3,"handle_id":99,"object_load_ms":4}}),
+                    "mvp-worker-node",
+                    1_905,
+                    905,
+                ),
+            ),
+        ]);
+        let path = write_synthetic_event_dump("vastai-summary-operator", event_pairs);
+        let events = parse_dump_log_events(&path).expect("parse summary events");
+        let paths = MvpChatCheckPaths {
+            root: std::env::temp_dir(),
+            dump_log: path.clone(),
+            stdout: temp_path("vastai-summary-stdout"),
+            stderr: temp_path("vastai-summary-stderr"),
+            prompts: temp_path("vastai-summary-prompts"),
+            redacted_config: temp_path("vastai-summary-config"),
+            summary: temp_path("vastai-summary-json"),
+        };
+
+        let summary =
+            build_benchmark_summary(&events, 80, 9, MvpChatCheckScenario::VastAi, &paths, 12, 34)
+                .expect("summary builds");
+        let _ = fs::remove_file(path);
+
+        let invariants = summary
+            .pointer("/invariants")
+            .and_then(Value::as_array)
+            .expect("invariants array");
+        assert!(!invariants.iter().any(|invariant| {
+            matches!(
+                invariant.get("name").and_then(Value::as_str),
+                Some("activation_large_object_iroh_sent" | "activation_large_object_iroh_read")
+            )
+        }));
+        assert_eq!(
+            summary
+                .pointer("/gpu/pipeline_token_out_request_ids/0")
+                .and_then(Value::as_u64),
+            Some(1)
+        );
+        assert_eq!(
+            summary
+                .pointer("/gpu/pipeline_tokenizer_decode_ready_request_ids/1")
+                .and_then(Value::as_u64),
+            Some(2)
+        );
+        assert_eq!(
+            summary
+                .pointer("/vastai/selected_lease_count")
+                .and_then(Value::as_u64),
+            Some(1)
+        );
+        assert_eq!(
+            summary
+                .pointer("/vastai/selected_leases/0/gpu_name")
+                .and_then(Value::as_str),
+            Some("RTX 2060")
+        );
+        assert_eq!(
+            summary
+                .pointer("/pipeline/provisioning/max_send_events_for_stage")
+                .and_then(Value::as_u64),
+            Some(2)
+        );
+        assert_eq!(
+            summary
+                .pointer("/pipeline/provisioning/latest_wait/waiting_stage_index")
+                .and_then(Value::as_u64),
+            Some(7)
+        );
+        assert_eq!(
+            summary
+                .pointer("/pipeline/provisioning/ssh_bootstrap/retry_counts_by_node/8")
+                .and_then(Value::as_u64),
+            Some(1)
+        );
+        assert_eq!(
+            summary
+                .pointer("/gpu/host_gpu_samples/sample_count")
+                .and_then(Value::as_u64),
+            Some(1)
+        );
+        assert_eq!(
+            summary
+                .pointer("/gpu/host_gpu_samples/utilization_gpu_percent/max")
+                .and_then(Value::as_u64),
+            Some(73)
+        );
+        assert_eq!(
+            summary
+                .pointer("/pipeline/prompt_critical_paths/0/token_in_to_out_ms/count")
+                .and_then(Value::as_u64),
+            Some(1)
+        );
+        assert_eq!(
+            summary
+                .pointer("/pipeline/edge_handoffs/0/producer_ring_reads")
+                .and_then(Value::as_u64),
+            Some(1)
+        );
+        assert_eq!(
+            summary
+                .pointer("/pipeline/edge_handoffs/0/ingress_ring_write_ms/sum")
+                .and_then(Value::as_u64),
+            Some(3)
+        );
     }
 }