Improve CLI streaming output, logging flags, and schema/help ergonomics

crates-cli/yaak-cli/README.md

@@ -46,7 +46,8 @@ Global options:
 
 - `--data-dir <path>`: use a custom data directory
 - `-e, --environment <id>`: environment to use during request rendering/sending
-- `-v, --verbose`: verbose logging and send output
+- `-v, --verbose`: verbose send output (events and streamed response body)
+- `--log [level]`: enable CLI logging; optional level is `error|warn|info|debug|trace`
 
 Notes:
 

crates-cli/yaak-cli/src/cli.rs

@@ -21,10 +21,14 @@ pub struct Cli {
     #[arg(long, short, global = true)]
     pub environment: Option<String>,
 
-    /// Enable verbose logging
+    /// Enable verbose send output (events and streamed response body)
     #[arg(long, short, global = true)]
     pub verbose: bool,
 
+    /// Enable CLI logging; optionally set level (error|warn|info|debug|trace)
+    #[arg(long, global = true, value_name = "LEVEL", num_args = 0..=1, ignore_case = true)]
+    pub log: Option<Option<LogLevel>>,
+
     #[command(subcommand)]
     pub command: Commands,
 }
@@ -227,6 +231,27 @@ pub enum RequestSchemaType {
     Websocket,
 }
 
+#[derive(Clone, Copy, Debug, ValueEnum)]
+pub enum LogLevel {
+    Error,
+    Warn,
+    Info,
+    Debug,
+    Trace,
+}
+
+impl LogLevel {
+    pub fn as_filter(self) -> log::LevelFilter {
+        match self {
+            LogLevel::Error => log::LevelFilter::Error,
+            LogLevel::Warn => log::LevelFilter::Warn,
+            LogLevel::Info => log::LevelFilter::Info,
+            LogLevel::Debug => log::LevelFilter::Debug,
+            LogLevel::Trace => log::LevelFilter::Trace,
+        }
+    }
+}
+
 #[derive(Args)]
 #[command(disable_help_subcommand = true)]
 pub struct FolderArgs {

crates-cli/yaak-cli/src/commands/request.rs

@@ -9,7 +9,9 @@ use crate::utils::schema::append_agent_hints;
 use schemars::schema_for;
 use serde_json::{Map, Value, json};
 use std::collections::HashMap;
+use std::io::Write;
 use tokio::sync::mpsc;
+use yaak_http::sender::HttpResponseEvent as SenderHttpResponseEvent;
 use yaak::send::{SendHttpRequestByIdWithPluginsParams, send_http_request_by_id_with_plugins};
 use yaak_models::models::{GrpcRequest, HttpRequest, WebsocketRequest};
 use yaak_models::queries::any_request::AnyRequest;
@@ -470,14 +472,24 @@ async fn send_http_request_by_id(
 ) -> Result<(), String> {
     let plugin_context = PluginContext::new(None, Some(workspace_id.to_string()));
 
-    let (event_tx, mut event_rx) = mpsc::channel(100);
+    let (event_tx, mut event_rx) = mpsc::channel::<SenderHttpResponseEvent>(100);
+    let (body_chunk_tx, mut body_chunk_rx) = mpsc::unbounded_channel::<Vec<u8>>();
     let event_handle = tokio::spawn(async move {
         while let Some(event) = event_rx.recv().await {
-            if verbose {
+            if verbose && !matches!(event, SenderHttpResponseEvent::ChunkReceived { .. }) {
                 println!("{}", event);
             }
         }
     });
+    let body_handle = tokio::task::spawn_blocking(move || {
+        let mut stdout = std::io::stdout();
+        while let Some(chunk) = body_chunk_rx.blocking_recv() {
+            if stdout.write_all(&chunk).is_err() {
+                break;
+            }
+            let _ = stdout.flush();
+        }
+    });
     let response_dir = ctx.data_dir().join("responses");
 
     let result = send_http_request_by_id_with_plugins(SendHttpRequestByIdWithPluginsParams {
@@ -489,6 +501,7 @@ async fn send_http_request_by_id(
         cookie_jar_id: None,
         response_dir: &response_dir,
         emit_events_to: Some(event_tx),
+        emit_response_body_chunks_to: Some(body_chunk_tx),
         plugin_manager: ctx.plugin_manager(),
         encryption_manager: ctx.encryption_manager.clone(),
         plugin_context: &plugin_context,
@@ -498,24 +511,7 @@ async fn send_http_request_by_id(
     .await;
 
     let _ = event_handle.await;
-    let result = result.map_err(|e| e.to_string())?;
+    let _ = body_handle.await;
-
+    result.map_err(|e| e.to_string())?;
-    if verbose {
-        println!();
-    }
-    println!(
-        "HTTP {} {}",
-        result.response.status,
-        result.response.status_reason.as_deref().unwrap_or("")
-    );
-    if verbose {
-        for header in &result.response.headers {
-            println!("{}: {}", header.name, header.value);
-        }
-        println!();
-    }
-    let body = String::from_utf8(result.response_body)
-        .map_err(|e| format!("Failed to read response body: {e}"))?;
-    println!("{}", body);
     Ok(())
 }

crates-cli/yaak-cli/src/main.rs

@@ -12,10 +12,18 @@ use context::CliContext;
 
 #[tokio::main]
 async fn main() {
-    let Cli { data_dir, environment, verbose, command } = Cli::parse();
+    let Cli { data_dir, environment, verbose, log, command } = Cli::parse();
 
-    if verbose {
+    if let Some(log_level) = log {
-        env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
+        match log_level {
+            Some(level) => {
+                env_logger::Builder::new().filter_level(level.as_filter()).init();
+            }
+            None => {
+                env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info"))
+                    .init();
+            }
+        }
     }
 
     let app_id = if cfg!(debug_assertions) { "app.yaak.desktop.dev" } else { "app.yaak.desktop" };

crates-cli/yaak-cli/tests/request_commands.rs

@@ -156,7 +156,6 @@ fn request_send_persists_response_body_and_events() {
         .args(["request", "send", &request_id])
         .assert()
         .success()
-        .stdout(contains("HTTP 200 OK"))
         .stdout(contains("hello from integration test"));
 
     let qm = query_manager(data_dir);

crates-cli/yaak-cli/tests/send_commands.rs

@@ -31,7 +31,6 @@ fn top_level_send_workspace_sends_http_requests_and_prints_summary() {
         .args(["send", "wk_test"])
         .assert()
         .success()
-        .stdout(contains("HTTP 200 OK"))
         .stdout(contains("workspace bulk send"))
         .stdout(contains("Send summary: 1 succeeded, 0 failed"));
 }
@@ -62,7 +61,6 @@ fn top_level_send_folder_sends_http_requests_and_prints_summary() {
         .args(["send", "fl_test"])
         .assert()
         .success()
-        .stdout(contains("HTTP 200 OK"))
         .stdout(contains("folder bulk send"))
         .stdout(contains("Send summary: 1 succeeded, 0 failed"));
 }

crates-tauri/yaak-app/src/http_request.rs

@@ -154,6 +154,7 @@ async fn send_http_request_inner<R: Runtime>(
         cookie_jar_id,
         response_dir: &response_dir,
         emit_events_to: None,
+        emit_response_body_chunks_to: None,
         existing_response: Some(response_ctx.response().clone()),
         plugin_manager,
         encryption_manager,

crates/yaak/src/send.rs

@@ -239,6 +239,7 @@ pub struct SendHttpRequestByIdParams<'a, T: TemplateCallback> {
     pub cookie_jar_id: Option<String>,
     pub response_dir: &'a Path,
     pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub emit_response_body_chunks_to: Option<mpsc::UnboundedSender<Vec<u8>>>,
     pub cancelled_rx: Option<watch::Receiver<bool>>,
     pub prepare_sendable_request: Option<&'a dyn PrepareSendableRequest>,
     pub executor: Option<&'a dyn SendRequestExecutor>,
@@ -255,6 +256,7 @@ pub struct SendHttpRequestParams<'a, T: TemplateCallback> {
     pub cookie_jar_id: Option<String>,
     pub response_dir: &'a Path,
     pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub emit_response_body_chunks_to: Option<mpsc::UnboundedSender<Vec<u8>>>,
     pub cancelled_rx: Option<watch::Receiver<bool>>,
     pub auth_context_id: Option<String>,
     pub existing_response: Option<HttpResponse>,
@@ -271,6 +273,7 @@ pub struct SendHttpRequestWithPluginsParams<'a> {
     pub cookie_jar_id: Option<String>,
     pub response_dir: &'a Path,
     pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub emit_response_body_chunks_to: Option<mpsc::UnboundedSender<Vec<u8>>>,
     pub existing_response: Option<HttpResponse>,
     pub plugin_manager: Arc<PluginManager>,
     pub encryption_manager: Arc<EncryptionManager>,
@@ -288,6 +291,7 @@ pub struct SendHttpRequestByIdWithPluginsParams<'a> {
     pub cookie_jar_id: Option<String>,
     pub response_dir: &'a Path,
     pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
+    pub emit_response_body_chunks_to: Option<mpsc::UnboundedSender<Vec<u8>>>,
     pub plugin_manager: Arc<PluginManager>,
     pub encryption_manager: Arc<EncryptionManager>,
     pub plugin_context: &'a PluginContext,
@@ -353,6 +357,7 @@ pub async fn send_http_request_by_id_with_plugins(
         cookie_jar_id: params.cookie_jar_id,
         response_dir: params.response_dir,
         emit_events_to: params.emit_events_to,
+        emit_response_body_chunks_to: params.emit_response_body_chunks_to,
         existing_response: None,
         plugin_manager: params.plugin_manager,
         encryption_manager: params.encryption_manager,
@@ -397,6 +402,7 @@ pub async fn send_http_request_with_plugins(
         cookie_jar_id: params.cookie_jar_id,
         response_dir: params.response_dir,
         emit_events_to: params.emit_events_to,
+        emit_response_body_chunks_to: params.emit_response_body_chunks_to,
         cancelled_rx: params.cancelled_rx,
         auth_context_id: None,
         existing_response: params.existing_response,
@@ -427,6 +433,7 @@ pub async fn send_http_request_by_id<T: TemplateCallback>(
         cookie_jar_id: params.cookie_jar_id,
         response_dir: params.response_dir,
         emit_events_to: params.emit_events_to,
+        emit_response_body_chunks_to: params.emit_response_body_chunks_to,
         cancelled_rx: params.cancelled_rx,
         existing_response: None,
         prepare_sendable_request: params.prepare_sendable_request,
@@ -687,13 +694,17 @@ pub async fn send_http_request<T: TemplateCallback>(
             Ok(n) => {
                 written_bytes += n;
                 let start_idx = response_body.len() - n;
-                file.write_all(&response_body[start_idx..]).await.map_err(|source| {
+                let chunk = &response_body[start_idx..];
+                file.write_all(chunk).await.map_err(|source| {
                     SendHttpRequestError::WriteResponseBody { path: body_path.clone(), source }
                 })?;
                 file.flush().await.map_err(|source| SendHttpRequestError::WriteResponseBody {
                     path: body_path.clone(),
                     source,
                 })?;
+                if let Some(tx) = params.emit_response_body_chunks_to.as_ref() {
+                    let _ = tx.send(chunk.to_vec());
+                }
 
                 let now = Instant::now();
                 let should_update = now.duration_since(last_progress_update).as_millis()