ilink-hub 0.2.8

//! Built-in `codebuddy-code` profile: wraps the CodeBuddy Code CLI with session continuity.
//!
//! Reads P0 env vars, calls `codebuddy -p --output-format stream-json [--resume <uuid>]`,
//! and streams text output to the parent bridge via `ILINK_PARTIAL:` stdout lines.
//!
//! CodeBuddy Code CLI is compatible with Claude Code's `stream-json` protocol:
//!   {"type":"system","session_id":"<uuid>", ...}
//!   {"type":"assistant","message":{"content":[{"type":"text","text":"..."}],...},...}
//!   {"type":"result","subtype":"success","result":"...","session_id":"<uuid>"}
//!
//! Each assistant text chunk is written immediately as:
//!
//!   ILINK_PARTIAL:<json-encoded-string>
//!
//! When the stream ends, the final P0 session line is written:
//!
//!   ILINK_SESSION:<new_session_id>
//!
//! The response body is left empty so the bridge does not send a duplicate final message.
//!
//! If `--resume` fails (session expired / not found), automatically retries as a
//! fresh session so the user gets a response rather than a bare error.
//!
//! Supported env vars:
//!   ILINK_MESSAGE          — inbound user message (P0)
//!   ILINK_SESSION_ID       — existing session UUID to resume (P0, empty = new)
//!   ILINK_CODEBUDDY_MODEL  — override the CodeBuddy model (e.g. claude-sonnet-4.6)

use anyhow::{Context, Result};
use tokio::io::AsyncBufReadExt;
use tokio::process::Command;

use super::common;

/// JSON event line from `codebuddy --output-format stream-json`.
/// CodeBuddy Code uses the same `stream-json` schema as Claude Code CLI:
/// `type == "assistant"` carries incremental text, `type == "result"` carries the
/// final `session_id` and `result` text.
type CodeBuddyStreamEvent = common::StreamJsonEvent;

pub async fn run() -> Result<()> {
    let (message, session_id) = common::read_message_and_session();

    let new_session_id = common::with_session_resume_fallback(
        "codebuddy-code",
        &message,
        &session_id,
        |m, s| async move { stream_codebuddy(&m, &s).await },
    )
    .await?;

    // P0 output: optional session line only.
    // All response text was already streamed via ILINK_PARTIAL during execution.
    common::emit_session_line(new_session_id.as_deref());

    Ok(())
}

/// Call `codebuddy -p --output-format stream-json`, emit every assistant text chunk as an
/// `ILINK_PARTIAL:` stdout line, and return the session ID from the result event.
///
/// When the model uses tools between turns, the final assistant reply may only appear
/// in `result.result` (with no preceding `assistant` event); we emit it as an extra
/// ILINK_PARTIAL in that case.
async fn stream_codebuddy(message: &str, session_id: &str) -> Result<Option<String>> {
    let mut args: Vec<String> = vec![
        "--print".into(),
        "--output-format".into(),
        "stream-json".into(),
        "--dangerously-skip-permissions".into(),
        // In -p (non-interactive) mode stdin is closed, so AskUserQuestion would
        // block the process forever with no visible prompt to the user.
        "--disallowedTools".into(),
        "AskUserQuestion".into(),
    ];

    if let Ok(model) = std::env::var("ILINK_CODEBUDDY_MODEL") {
        if !model.trim().is_empty() {
            args.push("--model".into());
            args.push(model.trim().to_string());
        }
    }

    if !session_id.is_empty() {
        args.push("--resume".into());
        args.push(session_id.to_string());
    }

    // The prompt argument comes last (positional).
    args.push(message.to_string());

    let mut cmd = Command::new("codebuddy");
    cmd.args(&args);
    cmd.stdin(std::process::Stdio::null());
    cmd.stdout(std::process::Stdio::piped());
    cmd.stderr(std::process::Stdio::piped());
    cmd.kill_on_drop(true);

    let mut child = cmd.spawn().context(
        "failed to spawn `codebuddy`; ensure CodeBuddy Code CLI is installed and in PATH",
    )?;

    let child_stdout = child.stdout.take().context("stdout pipe missing")?;
    let child_stderr = child.stderr.take().context("stderr pipe missing")?;

    // Drain stderr in background to prevent pipe buffer deadlock.
    let stderr_task = common::spawn_capped_drain(child_stderr);

    let mut reader = tokio::io::BufReader::new(child_stdout);
    let mut line = String::new();
    let mut found_session_id: Option<String> = None;
    // Track the last partial text sent so we can detect when result.result differs.
    let mut last_partial: Option<String> = None;

    loop {
        line.clear();
        let n = reader
            .read_line(&mut line)
            .await
            .context("read codebuddy stdout")?;
        if n == 0 {
            break;
        }

        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }

        let Ok(event) = serde_json::from_str::<CodeBuddyStreamEvent>(trimmed) else {
            continue;
        };

        match event.event_type.as_deref() {
            Some("assistant") => {
                if let Some(msg) = &event.message {
                    let text = msg.text();
                    // Guard with trim() so that whitespace-only text blocks (e.g. a bare
                    // "\n" emitted between tool calls) do not produce an empty-looking
                    // ILINK_PARTIAL that the user sees as a blank message.
                    if !text.trim().is_empty() {
                        common::emit_partial(&text)?;
                        last_partial = Some(text);
                    }
                }
            }
            Some("result") => {
                found_session_id = event.session_id;
                // When the model uses tools between turns, the final assistant reply text
                // may only appear in result.result and have no corresponding assistant event.
                // Send it as a partial if it differs from the last streamed chunk.
                if let Some(result_text) = event.result.filter(|t| !t.trim().is_empty()) {
                    let already_sent = last_partial.as_deref() == Some(result_text.as_str());
                    if !already_sent {
                        common::emit_partial(&result_text)?;
                    }
                }
            }
            _ => {}
        }
    }

    let status = child.wait().await.context("wait for codebuddy")?;
    let stderr = stderr_task.await.unwrap_or_default();

    common::ensure_success("codebuddy", status, &stderr, found_session_id.is_some())?;

    Ok(found_session_id)
}

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

    #[test]
    fn deserialize_system_event() {
        let json = r#"{"type":"system","subtype":"init","session_id":"72f18735-deb2-42e1-be58-c07489548e02","uuid":"72f18735-deb2-42e1-be58-c07489548e02"}"#;
        let event: StreamJsonEvent = serde_json::from_str(json).unwrap();
        assert_eq!(event.event_type.as_deref(), Some("system"));
        assert_eq!(
            event.session_id.as_deref(),
            Some("72f18735-deb2-42e1-be58-c07489548e02")
        );
    }

    #[test]
    fn deserialize_assistant_event_with_text_block() {
        let json = r#"{"type":"assistant","uuid":"abc","session_id":"sess-1","message":{"id":"abc","content":[{"type":"text","text":"Hi!"}],"model":"claude-sonnet-4.6","role":"assistant","stop_reason":null,"stop_sequence":null,"type":"message","usage":{}},"parent_tool_use_id":null}"#;
        let event: StreamJsonEvent = serde_json::from_str(json).unwrap();
        assert_eq!(event.event_type.as_deref(), Some("assistant"));
        let text = event.message.unwrap().text();
        assert_eq!(text, "Hi!");
    }

    #[test]
    fn deserialize_assistant_event_skips_thinking_block() {
        let json = r#"{"type":"assistant","message":{"content":[{"type":"thinking","thinking":"internal thought","signature":""},{"type":"text","text":"Hi there!"}]}}"#;
        let event: StreamJsonEvent = serde_json::from_str(json).unwrap();
        let text = event.message.unwrap().text();
        // thinking blocks must be skipped, only text blocks contribute
        assert_eq!(text, "Hi there!");
    }

    #[test]
    fn deserialize_result_event() {
        let json = r#"{"type":"result","subtype":"success","is_error":false,"result":"Hi!","session_id":"72f18735-deb2-42e1-be58-c07489548e02"}"#;
        let event: StreamJsonEvent = serde_json::from_str(json).unwrap();
        assert_eq!(event.event_type.as_deref(), Some("result"));
        assert_eq!(event.result.as_deref(), Some("Hi!"));
        assert_eq!(
            event.session_id.as_deref(),
            Some("72f18735-deb2-42e1-be58-c07489548e02")
        );
    }

    #[test]
    fn result_with_no_prior_partial_always_emits() {
        let last_partial: Option<String> = None;
        let result_text = "The answer is 42.";
        let already_sent = last_partial.as_deref() == Some(result_text);
        assert!(
            !already_sent,
            "when no prior partial exists, result must NOT be skipped"
        );
    }

    #[test]
    fn result_matching_last_partial_is_not_resent() {
        let last_partial: Option<String> = Some("Hi!".to_string());
        let result_text = "Hi!";
        let already_sent = last_partial.as_deref() == Some(result_text);
        assert!(
            already_sent,
            "result matching last_partial must be skipped to avoid duplicate"
        );
    }

    #[test]
    fn whitespace_only_text_block_is_skipped() {
        for ws in ["\n", "  ", "\t", "\r\n"] {
            let json = format!(
                r#"{{"type":"assistant","message":{{"content":[{{"type":"text","text":"{ws}"}}]}}}}"#,
                ws = ws
                    .replace('\n', "\\n")
                    .replace('\t', "\\t")
                    .replace('\r', "\\r")
            );
            let event: StreamJsonEvent = serde_json::from_str(&json).unwrap();
            let text = event.message.unwrap().text();
            assert!(
                text.trim().is_empty(),
                "trimmed '{ws:?}' must be empty → bridge must skip it"
            );
        }
    }

    #[test]
    fn real_world_full_stream_parses_correctly() {
        // Simulate the actual output sequence from a real codebuddy invocation.
        let lines = [
            r#"{"type":"system","subtype":"init","uuid":"72f18735-deb2-42e1-be58-c07489548e02","session_id":"72f18735-deb2-42e1-be58-c07489548e02","apiKeySource":"copilot.tencent.com","model":"hy3-preview-ioa","permissionMode":"bypassPermissions"}"#,
            r#"{"type":"assistant","uuid":"799fc30a","session_id":"72f18735-deb2-42e1-be58-c07489548e02","message":{"id":"799fc30a","content":[{"type":"thinking","thinking":"The user is just saying hello","signature":""}],"model":"hy3-preview-ioa","role":"assistant","stop_reason":null,"stop_sequence":null,"type":"message","usage":{}},"parent_tool_use_id":null}"#,
            r#"{"type":"assistant","uuid":"1f990805","session_id":"72f18735-deb2-42e1-be58-c07489548e02","message":{"id":"1f990805","content":[{"type":"text","text":"Hi!"}],"model":"hy3-preview-ioa","role":"assistant","stop_reason":null,"stop_sequence":null,"type":"message","usage":{}},"parent_tool_use_id":null}"#,
            r#"{"type":"result","subtype":"success","is_error":false,"result":"Hi!","uuid":"83c578cc","session_id":"72f18735-deb2-42e1-be58-c07489548e02","duration_ms":3708,"num_turns":3,"total_cost_usd":0}"#,
        ];

        let mut found_session_id: Option<String> = None;
        let mut last_partial: Option<String> = None;
        let mut emitted: Vec<String> = Vec::new();

        for line in &lines {
            let event: StreamJsonEvent = serde_json::from_str(line).unwrap();
            match event.event_type.as_deref() {
                Some("assistant") => {
                    if let Some(msg) = &event.message {
                        let text = msg.text();
                        if !text.trim().is_empty() {
                            emitted.push(text.clone());
                            last_partial = Some(text);
                        }
                    }
                }
                Some("result") => {
                    found_session_id = event.session_id;
                    if let Some(rt) = event.result.filter(|t| !t.trim().is_empty()) {
                        if last_partial.as_deref() != Some(rt.as_str()) {
                            emitted.push(rt);
                        }
                    }
                }
                _ => {}
            }
        }

        // Only "Hi!" should be emitted (thinking block skipped, result deduped)
        assert_eq!(
            emitted,
            vec!["Hi!"],
            "only the text block should be emitted"
        );
        assert_eq!(
            found_session_id.as_deref(),
            Some("72f18735-deb2-42e1-be58-c07489548e02")
        );
    }
}