terraphim_orchestrator 1.20.2

//! Pure-function extractor for the assistant's final review text out of a
//! captured agent drain stream.
//!
//! The orchestrator writes a complete copy of the agent's stdout/stderr into
//! a drain `.log` file (`start_output_log_drain`, `lib.rs:1093-1138`). The
//! stream format depends on the `cli_tool`:
//!
//! - **pi-rust / glm / kimi** (cli_tool contains `pi-rust`): newline-delimited
//!   JSON events. The assistant text lives in `message.content[*].text` of
//!   `message_update` / `message_end` / `turn_end` events. The *last* such
//!   text wins (the streaming deltas plus the terminal cumulative text both
//!   carry it; we take the last non-empty one).
//! - **claude** (cli_tool contains `claude`): `--output-format stream-json`
//!   produces `{"type":"content_block_delta",...,"delta":{"type":"text_delta",
//!   "text":"..."}}` events whose `delta.text` accumulates. The last
//!   `text_delta` wins; we accumulate by concatenating in order.
//! - **opencode** (cli_tool contains `opencode`): `{"type":"text","part":
//!   {"text":"..."}}` events. We concatenate `part.text` in order.
//! - **non-JSON pass-through** (claude `--output-format text` or any free-form
//!   output): any non-empty line that is not valid JSON.
//!
//! Returns `None` if the stream contains no recognisable assistant text (e.g.
//! the agent only emitted thinking events, or the stream is empty).
//!
//! Refs terraphim/terraphim-ai#2301.

/// Extract the agent's final assistant text from a captured output stream.
///
/// `lines` should be the contents of the orchestrator-written drain `.log`,
/// in the order they were written. `cli_tool` is the resolved CLI executable
/// path or alias (e.g. `/home/alex/.local/bin/pi-rust` or
/// `/home/alex/.local/bin/claude` or `/home/alex/.bun/bin/opencode`).
///
/// Returns `None` if the stream contains no recognisable assistant text.
pub fn extract_final_assistant_text(lines: &[String], cli_tool: &str) -> Option<String> {
    let cli = cli_tool.to_ascii_lowercase();
    // Try the format-specific extractor first. If that returns None AND
    // the stream has any non-JSON text, fall through to pass-through — this
    // covers `claude --output-format text` (the format is non-JSON even
    // though the CLI is claude) and other free-form outputs.
    if cli.contains("pi-rust") || cli.contains("kimi") {
        if let Some(t) = extract_pi_rust(lines) {
            return Some(t);
        }
    } else if cli.contains("claude") {
        if let Some(t) = extract_claude_stream_json(lines) {
            return Some(t);
        }
    } else if cli.contains("opencode")
        && let Some(t) = extract_opencode(lines)
    {
        return Some(t);
    }
    extract_pass_through(lines)
}

fn extract_pi_rust(lines: &[String]) -> Option<String> {
    // pi-rust emits both streaming `message_update` deltas AND terminal
    // `message_end` / `turn_end` events whose `message.content[*].text`
    // carries the full text. The terminal event has the longest,
    // most-recently-accumulated text and is what the user reads as the
    // assistant's final turn. We return that as-is.
    let mut last_full_text: Option<String> = None;
    let mut last_delta_text: Option<String> = None;
    for line in lines {
        let line = line.trim();
        if line.is_empty() || line.starts_with("# ") {
            continue;
        }
        let Ok(value) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        let event_type = value.get("type").and_then(|v| v.as_str()).unwrap_or("");
        let msg = match value.get("message") {
            Some(m) if m.is_object() => m,
            _ => continue,
        };
        if msg.get("role").and_then(|v| v.as_str()) != Some("assistant") {
            continue;
        }
        let content = match msg.get("content") {
            Some(c) if c.is_array() => c.as_array().unwrap(),
            _ => continue,
        };
        if matches!(event_type, "message_end" | "turn_end" | "message_start") {
            if let Some(text) = join_text_parts(content)
                && !text.is_empty()
            {
                last_full_text = Some(text);
            }
        } else if event_type == "message_update" {
            // The streaming deltas carry partial text; the last non-empty
            // text in a message_update is the most-recent partial. We capture
            // it as a fallback in case the terminal event is missing.
            if let Some(text) = join_text_parts(content)
                && !text.is_empty()
            {
                last_delta_text = Some(text);
            }
        }
    }
    last_full_text.or(last_delta_text)
}

fn extract_claude_stream_json(lines: &[String]) -> Option<String> {
    // claude --output-format stream-json emits `content_block_delta` events
    // whose `delta.text` is a text fragment. We accumulate them all and
    // return the concatenation.
    let mut accumulated = String::new();
    let mut any = false;
    for line in lines {
        let line = line.trim();
        if line.is_empty() || line.starts_with("# ") {
            continue;
        }
        let Ok(value) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        let event_type = value.get("type").and_then(|v| v.as_str()).unwrap_or("");
        if event_type != "content_block_delta" {
            continue;
        }
        if let Some(delta) = value.get("delta")
            && delta.get("type").and_then(|v| v.as_str()) == Some("text_delta")
            && let Some(text) = delta.get("text").and_then(|v| v.as_str())
            && !text.is_empty()
        {
            accumulated.push_str(text);
            any = true;
        }
    }
    if any { Some(accumulated) } else { None }
}

fn extract_opencode(lines: &[String]) -> Option<String> {
    // opencode emits `{"type":"text","part":{"text":"..."}}` events. We
    // concatenate `part.text` in order.
    let mut accumulated = String::new();
    let mut any = false;
    for line in lines {
        let line = line.trim();
        if line.is_empty() || line.starts_with("# ") {
            continue;
        }
        let Ok(value) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        if value.get("type").and_then(|v| v.as_str()) != Some("text") {
            continue;
        }
        if let Some(text) = value
            .get("part")
            .and_then(|p| p.get("text"))
            .and_then(|t| t.as_str())
            && !text.is_empty()
        {
            accumulated.push_str(text);
            any = true;
        }
    }
    if any { Some(accumulated) } else { None }
}

fn extract_pass_through(lines: &[String]) -> Option<String> {
    // Last resort: any non-empty line that isn't valid JSON. We concatenate
    // them in order with a newline. Used for claude --output-format text or
    // any other CLI that emits free-form output.
    //
    // We skip drain-header comment lines (`# agent: pr-reviewer`, `# exit_code:`,
    // etc.) — the convention is a single `#` followed by a space. Markdown
    // headings like `### Summary` have NO space after the first `#` and so
    // are NOT skipped.
    let mut accumulated = String::new();
    let mut any = false;
    for line in lines {
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        if line.starts_with("# ") {
            continue;
        }
        if serde_json::from_str::<serde_json::Value>(line).is_ok() {
            continue;
        }
        if !accumulated.is_empty() {
            accumulated.push('\n');
        }
        accumulated.push_str(line);
        any = true;
    }
    if any { Some(accumulated) } else { None }
}

/// Walk an assistant `content` array and return the concatenation of any
/// `{"type":"text","text":"..."}` parts. Returns `None` if there are no
/// such parts, or the concatenation is empty.
fn join_text_parts(content: &[serde_json::Value]) -> Option<String> {
    let mut s = String::new();
    let mut any = false;
    for c in content {
        if c.get("type").and_then(|v| v.as_str()) == Some("text")
            && let Some(text) = c.get("text").and_then(|v| v.as_str())
        {
            s.push_str(text);
            any = true;
        }
    }
    if any { Some(s) } else { None }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn read_fixture(name: &str) -> Vec<String> {
        let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("tests/fixtures/pr_review")
            .join(name);
        let raw = std::fs::read_to_string(&path)
            .unwrap_or_else(|e| panic!("read fixture {}: {e}", path.display()));
        raw.lines().map(|s| s.to_string()).collect()
    }

    #[test]
    fn extractor_returns_last_assistant_text_on_pi_rust_stream() {
        // The real fragment from PR #23 (glm-5.1, 240.8s). The agent did not
        // emit a verdict block (this is the bug being fixed); we just assert
        // the extractor returns the last assistant text it did emit.
        let lines = read_fixture("drain_pi_rust_glm_review.log");
        let out = extract_final_assistant_text(&lines, "/home/alex/.local/bin/pi-rust")
            .expect("extractor should return Some on a real pi-rust stream");
        assert!(
            !out.is_empty(),
            "extracted text should not be empty on a real stream"
        );
        assert!(
            out.contains("Handoff envelope") || out.len() > 50,
            "extracted text should be the agent's final summary (got len={})",
            out.len()
        );
    }

    #[test]
    fn extractor_returns_verdict_block_on_pi_rust_with_verdict() {
        // The synthetic pi-rust stream ends with a complete verdict block in
        // the terminal `message_end` event.
        let lines = read_fixture("drain_pi_rust_glm_with_verdict.log");
        let out = extract_final_assistant_text(&lines, "/home/alex/.local/bin/pi-rust")
            .expect("extractor should return Some");
        assert!(out.contains("### Summary"));
        assert!(out.contains("### Confidence Score: 5/5"));
        assert!(out.contains("### Inline Findings"));
        assert!(out.contains("<sub>Last reviewed commit: 2a3b4c5 | Reviews (1)</sub>"));
    }

    #[test]
    fn extractor_accumulates_claude_stream_json_text_deltas() {
        let lines = read_fixture("drain_claude_stream_json_assistant_only.log");
        let out = extract_final_assistant_text(&lines, "/home/alex/.local/bin/claude")
            .expect("extractor should return Some");
        // The fixture has two message rounds. The extractor accumulates ALL
        // deltas in order, so the final text is the concatenation of both.
        assert!(out.contains("SHA-bind"));
        assert!(out.contains("Confidence Score: 5/5"));
        // The second round's confidence is 4/5 and its Reviews (2).
        assert!(out.contains("Reviews (2)"));
    }

    #[test]
    fn extractor_concatenates_opencode_text_parts() {
        let lines = read_fixture("drain_opencode_text.log");
        let out = extract_final_assistant_text(&lines, "/home/alex/.bun/bin/opencode")
            .expect("extractor should return Some");
        assert!(out.contains("Removes the four openai/*"));
        assert!(out.contains("Confidence Score: 5/5"));
        assert!(out.contains("Last reviewed commit: c6d8567"));
    }

    #[test]
    fn extractor_passes_through_non_json_lines() {
        let lines = read_fixture("drain_pass_through.log");
        let out = extract_final_assistant_text(
            &lines,
            "/home/alex/.local/bin/claude --output-format text",
        )
        .expect("extractor should return Some on free-form output");
        assert!(out.contains("### Summary"));
        assert!(out.contains("Confidence Score: 5/5"));
        assert!(out.contains("received"));
    }

    #[test]
    fn extractor_returns_none_on_empty_stream() {
        let lines = read_fixture("drain_empty.log");
        let out = extract_final_assistant_text(&lines, "/home/alex/.local/bin/pi-rust");
        assert_eq!(out, None);
    }

    #[test]
    fn extractor_returns_none_on_header_only_stream() {
        // A stream with only the `# agent: pr-reviewer` header (the 2s empty
        // exit) and no JSON content.
        let lines: Vec<String> = vec![
            "# agent: pr-reviewer".to_string(),
            "# exit_code: Some(0)".to_string(),
            "# exit_class: empty_success".to_string(),
        ];
        let out = extract_final_assistant_text(&lines, "/home/alex/.local/bin/pi-rust");
        assert_eq!(out, None);
    }

    #[test]
    fn extractor_routes_by_cli_tool_substring() {
        // All three CLI tool path variants should dispatch to the right
        // extractor; the cli_tool check is case-insensitive substring.
        let lines = read_fixture("drain_pi_rust_glm_with_verdict.log");
        assert!(extract_final_assistant_text(&lines, "/home/alex/.local/bin/PI-RUST").is_some());
        assert!(extract_final_assistant_text(&lines, "pi-rust-extra-arg").is_some());
        assert!(extract_final_assistant_text(&lines, "kimi-via-pi-rust").is_some());
        // Wrong cli_tool falls through to pass-through, which returns None
        // here because the lines are all valid JSON.
        let out = extract_final_assistant_text(&lines, "/bin/cat");
        assert_eq!(out, None);
    }
}