earl 0.5.2

AI-safe CLI for AI agents
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

use anyhow::Result;
use serde_json::{Map, Value};
use tokio::sync::mpsc;

use earl_core::schema::ResultTemplate;
use earl_core::{Redactor, StreamChunk, decode_response};

use crate::output::human::render_human_output;
use crate::protocol::extract::extract_result;

/// Consume streaming chunks and render each one to stdout.
///
/// Each [`StreamChunk`] is decoded, extracted, and printed independently,
/// giving the user real-time output as data arrives from the server.
///
/// When `json_mode` is true and `active_env` is `Some`, the `"environment"`
/// key is included in every emitted JSON line for parity with non-streaming
/// JSON output.
pub async fn render_streaming_output(
    mut rx: mpsc::Receiver<StreamChunk>,
    result_template: &ResultTemplate,
    args: &Map<String, Value>,
    redactor: &Redactor,
    json_mode: bool,
    active_env: Option<&str>,
) -> Result<()> {
    use std::io::Write;

    while let Some(chunk) = rx.recv().await {
        let decoded_body = match decode_response(
            result_template.decode,
            chunk.content_type.as_deref(),
            &chunk.data,
        ) {
            Ok(d) => d,
            Err(e) => {
                tracing::warn!("skipping stream chunk: failed to decode: {e}");
                continue;
            }
        };
        let extracted = match extract_result(result_template.extract.as_ref(), &decoded_body) {
            Ok(v) => v,
            Err(e) => {
                tracing::warn!("skipping stream chunk: failed to extract: {e}");
                continue;
            }
        };

        let stdout = std::io::stdout();
        let mut out = stdout.lock();
        if json_mode {
            let mut map = Map::from_iter([
                ("result".to_string(), extracted.clone()),
                ("decoded".to_string(), decoded_body.to_json_value()),
            ]);
            if let Some(env) = active_env {
                map.insert("environment".to_string(), Value::String(env.to_string()));
            }
            let redacted = redactor.redact_json(&Value::Object(map));
            writeln!(out, "{}", serde_json::to_string(&redacted)?)?;
        } else {
            let output = render_human_output(result_template, args, &extracted)?;
            writeln!(out, "{}", redactor.redact(&output))?;
        }
        out.flush()?;
    }

    Ok(())
}