ag_agent/channel/

cli.rs

//! CLI subprocess [`AgentChannel`] adapter.
//!
//! Spawns a provider CLI process per turn, streams stdout line-by-line as
//! [`TurnEvent`]s, and parses the final process output when the process exits.

use std::os::unix::process::ExitStatusExt as _;
use std::sync::{Arc, Mutex};

use ag_protocol::{
    AgentResponse, AgentResponseSummary, ProtocolRequestProfile, TurnPrompt,
    build_protocol_repair_prompt,
};
use tokio::io::AsyncBufReadExt as _;
use tokio::sync::mpsc;

use crate::agent::cli::{error, stdin};
use crate::agent::{self as agent, AgentBackend, BuildCommandRequest};
use crate::channel::{
    AgentChannel, AgentError, AgentFuture, SessionRef, StartSessionRequest, TurnEvent, TurnRequest,
    TurnResult,
};
use crate::model::agent::AgentKind;

/// [`AgentChannel`] adapter that spawns one CLI subprocess per agent turn.
///
/// Stdout lines are classified by
/// [`agent::parse_stream_output_line`] and transient loader updates are
/// forwarded as [`TurnEvent::ThoughtDelta`]. A kill signal transitions the
/// turn to a failed state with a `[Stopped]` banner. A spawn failure is
/// surfaced through [`AgentError`].
pub(crate) struct CliAgentChannel {
    /// Provider-specific command builder.
    backend: Arc<dyn AgentBackend>,
    /// Provider family used for stream and response parsing.
    kind: AgentKind,
}

impl CliAgentChannel {
    /// Creates a CLI channel backed by the given pre-built backend.
    ///
    /// Channel factories use this helper so transport selection can be done
    /// once before constructing the concrete channel. Tests also use it to
    /// inject a [`MockAgentBackend`] that controls command construction and
    /// process spawning without relying on a real provider binary.
    pub(crate) fn with_backend(backend: Arc<dyn agent::AgentBackend>, kind: AgentKind) -> Self {
        Self { backend, kind }
    }
}

/// Builds the provider backend command request for one CLI turn.
fn build_command_request<'a>(
    request: &'a TurnRequest,
    prompt_text: &'a str,
) -> BuildCommandRequest<'a> {
    BuildCommandRequest {
        attachments: &request.prompt.attachments,
        folder: &request.folder,
        main_checkout_root: request.main_checkout_root.as_deref(),
        replay_transcript: request.replay_transcript.as_deref(),
        model: &request.model,
        prompt: prompt_text,
        reasoning_level: request.reasoning_level,
        request_kind: &request.request_kind,
    }
}

impl AgentChannel for CliAgentChannel {
    /// Returns a [`SessionRef`] immediately; CLI turns are stateless.
    fn start_session(
        &self,
        req: StartSessionRequest,
    ) -> AgentFuture<Result<SessionRef, AgentError>> {
        let session_id = req.session_id;

        Box::pin(async move { Ok(SessionRef { session_id }) })
    }

    /// Spawns a CLI process for the turn and streams its output as events.
    ///
    /// Stdout lines are parsed with the provider-specific stream parser and
    /// loader-oriented interim text is forwarded as
    /// [`TurnEvent::ThoughtDelta`]. After the process exits, usage
    /// statistics are extracted from the raw stdout/stderr and the final
    /// parsed response is returned in [`TurnResult`].
    ///
    /// # Errors
    /// Returns [`AgentError`] when command construction fails, the process
    /// cannot be spawned, or the process is killed by a signal.
    fn run_turn(
        &self,
        _session_id: String,
        req: TurnRequest,
        events: mpsc::UnboundedSender<TurnEvent>,
    ) -> AgentFuture<Result<TurnResult, AgentError>> {
        let kind = self.kind;
        let backend = Arc::clone(&self.backend);

        Box::pin(async move {
            let prompt_text = req.prompt.agent_text();
            let build_request = build_command_request(&req, &prompt_text);
            let build_result = backend.build_command(build_request);
            let stdin_payload_result = agent::build_command_stdin_payload(kind, build_request);
            let command = build_result.map_err(|error| {
                AgentError::Backend(format!("Failed to build command: {error}"))
            })?;
            let stdin_payload = stdin_payload_result.map_err(|error| {
                AgentError::Backend(format!("Failed to build command stdin payload: {error}"))
            })?;

            let mut tokio_cmd = tokio::process::Command::from(command);
            tokio_cmd.stdin(if stdin_payload.is_some() {
                std::process::Stdio::piped()
            } else {
                std::process::Stdio::null()
            });
            tokio_cmd.stdout(std::process::Stdio::piped());
            tokio_cmd.stderr(std::process::Stdio::piped());
            tokio_cmd.kill_on_drop(true);

            let mut child = tokio_cmd
                .spawn()
                .map_err(|error| AgentError::Io(format!("Failed to spawn process: {error}")))?;

            // Notify the consumer of the child PID so cancellation signals can
            // be sent while the process is running.
            // Fire-and-forget: receiver may be dropped during shutdown.
            let _ = events.send(TurnEvent::PidUpdate(child.id()));

            let raw_stdout = Arc::new(Mutex::new(String::new()));
            let raw_stderr = Arc::new(Mutex::new(String::new()));

            let stdout_task = {
                let stdout = child.stdout.take().ok_or_else(|| {
                    AgentError::Io("stdout pipe unavailable after spawn".to_string())
                })?;
                let raw_stdout = Arc::clone(&raw_stdout);
                let events = events.clone();

                tokio::spawn(stream_stdout(stdout, kind, events, raw_stdout))
            };

            let stderr_task = {
                let stderr = child.stderr.take().ok_or_else(|| {
                    AgentError::Io("stderr pipe unavailable after spawn".to_string())
                })?;
                let raw_stderr = Arc::clone(&raw_stderr);

                tokio::spawn(async move {
                    let mut reader = tokio::io::BufReader::new(stderr).lines();
                    while let Ok(Some(line)) = reader.next_line().await {
                        if let Ok(mut buf) = raw_stderr.lock() {
                            buf.push_str(&line);
                            buf.push('\n');
                        }
                    }
                })
            };
            let stdin_write_task = stdin::spawn_optional_stdin_write(
                child.stdin.take(),
                stdin_payload,
                "stdin pipe unavailable after spawn",
                AgentError::Io,
            );

            // Task join: panic in the spawned task is not recoverable here.
            let _ = stdout_task.await;
            // Task join: panic in the spawned task is not recoverable here.
            let _ = stderr_task.await;

            let exit_status = child.wait().await.ok();
            stdin::await_optional_stdin_write(
                stdin_write_task,
                "stdin write task failed",
                AgentError::Io,
            )
            .await?;

            // Clear the PID slot now that the child has exited.
            // Fire-and-forget: receiver may be dropped during shutdown.
            let _ = events.send(TurnEvent::PidUpdate(None));

            let killed_by_signal = exit_status
                .as_ref()
                .is_some_and(|status| status.signal().is_some());

            if killed_by_signal {
                return Err(AgentError::Backend(
                    "[Stopped] Agent interrupted by user.".to_string(),
                ));
            }

            let stdout_text = raw_stdout.lock().map(|buf| buf.clone()).unwrap_or_default();
            let stderr_text = raw_stderr.lock().map(|buf| buf.clone()).unwrap_or_default();
            if exit_status.as_ref().is_some_and(|status| !status.success()) {
                return Err(format_cli_turn_exit_error(
                    kind,
                    exit_status.and_then(|status| status.code()),
                    &stdout_text,
                    &stderr_text,
                ));
            }

            let parsed = agent::parse_response(kind, &stdout_text, &stderr_text);
            let assistant_message =
                parse_or_repair_cli_response(kind, &parsed.content, &req, &backend, &events)
                    .await?;

            Ok(TurnResult {
                assistant_message,
                context_reset: false,
                input_tokens: parsed.stats.input_tokens,
                output_tokens: parsed.stats.output_tokens,
                provider_conversation_id: None,
            })
        })
    }

    /// No-op; CLI sessions are stateless and require no teardown.
    fn shutdown_session(&self, _session_id: String) -> AgentFuture<Result<(), AgentError>> {
        Box::pin(async { Ok(()) })
    }
}

/// Parses one CLI turn response strictly, falling back to a single
/// protocol-repair retry when the initial parse fails.
///
/// When repair is attempted, a concise [`TurnEvent::ThoughtDelta`] is emitted
/// so the user can see that schema repair is in progress without flooding the
/// session output with parser diagnostics unless the turn ultimately fails.
async fn parse_or_repair_cli_response(
    kind: AgentKind,
    content: &str,
    req: &TurnRequest,
    backend: &Arc<dyn AgentBackend>,
    events: &mpsc::UnboundedSender<TurnEvent>,
) -> Result<AgentResponse, AgentError> {
    let protocol_profile = req.request_kind.protocol_profile();

    let parse_error = match agent::parse_turn_response(kind, content, protocol_profile) {
        Ok(response) => return Ok(response),
        Err(error) => error,
    };

    let _ = events.send(TurnEvent::ThoughtDelta(format!(
        "Protocol parse error; retrying schema repair for {kind}."
    )));

    let repair_prompt = build_protocol_repair_prompt(&parse_error, content);

    let repair_content = execute_cli_repair_turn(
        backend.as_ref(),
        kind,
        &req.folder,
        &req.model,
        &req.request_kind,
        req.reasoning_level,
        &repair_prompt,
    )
    .await
    .map_err(|error| {
        AgentError::Backend(format!(
            "{parse_error}\nprotocol repair transport failed: {error}"
        ))
    })?;

    match agent::parse_turn_response(kind, &repair_content, protocol_profile) {
        Ok(response) => Ok(response),
        Err(repair_error) => {
            if let Some(response) =
                antigravity_plain_text_fallback(kind, content, &repair_content, protocol_profile)
            {
                return Ok(response);
            }

            Err(AgentError::Backend(format!(
                "{parse_error}\nprotocol repair retry also failed: \
                 {repair_error}\nrepair_response:\n{repair_content}"
            )))
        }
    }
}

/// Converts exhausted Antigravity protocol failures into plain answer text.
///
/// Antigravity print mode does not currently provide native response-schema
/// enforcement, and it can ignore the protocol repair prompt by returning
/// ordinary prose again. After strict parsing and the repair retry both fail,
/// this preserves the original useful response as `answer` so the session can
/// complete instead of surfacing an internal schema error to the user.
fn antigravity_plain_text_fallback(
    kind: AgentKind,
    original_content: &str,
    repair_content: &str,
    protocol_profile: ProtocolRequestProfile,
) -> Option<AgentResponse> {
    if kind != AgentKind::Antigravity {
        return None;
    }

    let fallback_content =
        non_empty_content(original_content).or_else(|| non_empty_content(repair_content))?;
    let mut response = AgentResponse::plain(fallback_content.to_string());
    if matches!(protocol_profile, ProtocolRequestProfile::SessionTurn) {
        response.summary = Some(AgentResponseSummary {
            session: String::new(),
            turn: String::new(),
        });
    }

    Some(response)
}

/// Returns one trimmed non-empty content string suitable for fallback output.
fn non_empty_content(content: &str) -> Option<&str> {
    let trimmed_content = content.trim();
    if trimmed_content.is_empty() {
        return None;
    }

    Some(trimmed_content)
}

/// Reads stdout line-by-line, classifying each line and forwarding loader
/// updates.
///
/// Only non-response-content stream lines are forwarded as
/// [`TurnEvent::ThoughtDelta`]. Final assistant transcript output is parsed
/// from the accumulated raw stdout after the process exits. The raw bytes are
/// also accumulated in `raw_buffer` for final response parsing.
async fn stream_stdout(
    stdout: tokio::process::ChildStdout,
    kind: AgentKind,
    events: mpsc::UnboundedSender<TurnEvent>,
    raw_buffer: Arc<Mutex<String>>,
) {
    let mut reader = tokio::io::BufReader::new(stdout).lines();

    while let Ok(Some(line)) = reader.next_line().await {
        if let Ok(mut buf) = raw_buffer.lock() {
            buf.push_str(&line);
            buf.push('\n');
        }

        let Some((text, is_response_content)) = agent::parse_stream_output_line(kind, &line) else {
            continue;
        };

        if is_response_content {
            continue;
        }

        let trimmed_text = text.trim();
        if trimmed_text.is_empty() {
            continue;
        }

        // Fire-and-forget: receiver may be dropped during shutdown.
        let _ = events.send(TurnEvent::ThoughtDelta(trimmed_text.to_string()));
    }
}

/// Maximum wall-clock time for one protocol-repair CLI subprocess.
///
/// Repair turns ask the agent to re-emit a single JSON object, so they
/// should complete quickly. The timeout prevents a hung subprocess from
/// blocking the parent turn indefinitely. `kill_on_drop(true)` on the
/// child ensures the process is terminated when the future is dropped.
const REPAIR_TURN_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(1);

/// Spawns a fresh CLI process for one protocol-repair retry and returns
/// the parsed provider content string.
///
/// This helper strips down the full turn-execution pipeline to the minimum
/// needed for repair: command build, spawn, stdout/stderr collection, and
/// provider response parsing. No streaming, PID tracking, or signal
/// handling is performed because the repair is a transparent one-shot
/// correction, not a user-visible turn. A [`REPAIR_TURN_TIMEOUT`] guard
/// ensures a stuck process does not block the parent turn indefinitely.
async fn execute_cli_repair_turn(
    backend: &dyn AgentBackend,
    kind: AgentKind,
    folder: &std::path::Path,
    model: &str,
    request_kind: &crate::channel::AgentRequestKind,
    reasoning_level: crate::model::agent::ReasoningLevel,
    repair_prompt: &str,
) -> Result<String, String> {
    let prompt_payload = TurnPrompt::from_agent_data(repair_prompt.to_string());
    let build_request = BuildCommandRequest {
        attachments: &prompt_payload.attachments,
        folder,
        main_checkout_root: None,
        replay_transcript: None,
        model,
        prompt: repair_prompt,
        reasoning_level,
        request_kind,
    };
    let command = backend
        .build_command(build_request)
        .map_err(|error| format!("repair command build failed: {error}"))?;
    let repair_stdin_payload = agent::build_command_stdin_payload(kind, build_request)
        .map_err(|error| format!("repair stdin payload build failed: {error}"))?;

    let mut tokio_command = tokio::process::Command::from(command);
    tokio_command
        .stdin(if repair_stdin_payload.is_some() {
            std::process::Stdio::piped()
        } else {
            std::process::Stdio::null()
        })
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped())
        .kill_on_drop(true);

    let mut child = tokio_command
        .spawn()
        .map_err(|error| format!("repair process spawn failed: {error}"))?;

    let repair_stdin_write_task = stdin::spawn_optional_stdin_write(
        child.stdin.take(),
        repair_stdin_payload,
        "repair stdin pipe unavailable after spawn",
        std::convert::identity,
    );

    let output = tokio::time::timeout(REPAIR_TURN_TIMEOUT, child.wait_with_output())
        .await
        .map_err(|_| {
            format!(
                "repair process timed out after {}s",
                REPAIR_TURN_TIMEOUT.as_secs()
            )
        })?
        .map_err(|error| format!("repair process execution failed: {error}"))?;

    stdin::await_optional_stdin_write(
        repair_stdin_write_task,
        "repair stdin write task failed",
        std::convert::identity,
    )
    .await?;

    if !output.status.success() {
        return Err(format!(
            "repair process exited with status {}",
            output.status
        ));
    }

    let stdout_text = String::from_utf8_lossy(&output.stdout).into_owned();
    let stderr_text = String::from_utf8_lossy(&output.stderr).into_owned();
    let parsed = agent::parse_response(kind, &stdout_text, &stderr_text);

    Ok(parsed.content)
}

/// Formats one failed CLI turn into a user-facing error.
fn format_cli_turn_exit_error(
    kind: AgentKind,
    exit_code: Option<i32>,
    stdout: &str,
    stderr: &str,
) -> AgentError {
    AgentError::Backend(error::format_agent_cli_exit_error(
        kind,
        "Agent command",
        exit_code,
        stdout,
        stderr,
    ))
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;
    use std::sync::Arc;
    use std::time::Duration;

    use ag_protocol::{TurnPromptAttachment, TurnPromptTextSource};
    use tempfile::tempdir;
    use tokio::sync::mpsc;

    use super::*;
    use crate::MockAgentBackend;
    use crate::channel::AgentRequestKind;
    use crate::model::agent::{AgentKind, AgentModel, ReasoningLevel};

    fn make_turn_request(folder: PathBuf) -> TurnRequest {
        TurnRequest {
            folder,
            live_transcript: None,
            main_checkout_root: None,
            model: "claude-sonnet-5".to_string(),
            request_kind: AgentRequestKind::SessionStart,
            replay_transcript: None,
            prompt: "Write a test".into(),
            provider_conversation_id: None,
            persisted_instruction_conversation_id: None,
            reasoning_level: ReasoningLevel::default(),
        }
    }

    fn stdin_capture_command(capture_path: &std::path::Path) -> std::process::Command {
        let mut command = std::process::Command::new("sh");
        command.arg("-c").arg(
            "cat > \"$CLI_CAPTURE_PATH\"; printf '%s' \
             '{\"answer\":\"ok\",\"questions\":[],\"summary\":null}'",
        );
        command.env("CLI_CAPTURE_PATH", capture_path);

        command
    }

    /// Drains all currently buffered turn events from a test receiver.
    fn drain_events(receiver: &mut mpsc::UnboundedReceiver<TurnEvent>) -> Vec<TurnEvent> {
        let mut events = Vec::new();
        while let Ok(event) = receiver.try_recv() {
            events.push(event);
        }

        events
    }

    #[test]
    fn test_build_command_request_uses_agent_facing_prompt_text() {
        // Arrange
        let request = TurnRequest {
            folder: PathBuf::from("/tmp/session"),
            live_transcript: None,
            main_checkout_root: Some(PathBuf::from("/tmp/main")),
            model: "claude-sonnet-5".to_string(),
            request_kind: AgentRequestKind::SessionStart,
            replay_transcript: None,
            prompt: TurnPrompt::from("Review @src/main.rs"),
            provider_conversation_id: None,
            persisted_instruction_conversation_id: None,
            reasoning_level: ReasoningLevel::default(),
        };
        let prompt_text = request.prompt.agent_text();

        // Act
        let build_request = build_command_request(&request, &prompt_text);

        // Assert
        assert_eq!(build_request.prompt, "Review \"src/main.rs\"");
        assert_eq!(
            build_request.main_checkout_root,
            Some(std::path::Path::new("/tmp/main"))
        );
    }

    #[tokio::test]
    /// Verifies spawn failure returns `Err` with a descriptive message and
    /// does not emit any turn events when the process never starts.
    async fn test_run_turn_spawn_failure_returns_err_without_delta() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend
            .expect_build_command()
            .returning(|_| Ok(std::process::Command::new("/no-such-binary-agentty-test")));
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let result = channel.run_turn("sess-1".to_string(), req, events_tx).await;

        // Assert
        let error_message = result
            .expect_err("expected Err for spawn failure")
            .to_string();
        assert!(
            error_message.contains("Failed to spawn process"),
            "error was: {error_message}"
        );
        assert!(
            events_rx.try_recv().is_err(),
            "no events should be emitted when the process never spawned"
        );
    }

    #[tokio::test]
    /// Verifies kill-by-signal returns `Err` with a `[Stopped]` message and
    /// does not emit any loader updates.
    async fn test_run_turn_kill_signal_returns_err_without_stopped_delta() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning(|_| {
            let mut cmd = std::process::Command::new("sh");
            cmd.arg("-c").arg("kill -9 $$");

            Ok(cmd)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let result = channel.run_turn("sess-1".to_string(), req, events_tx).await;

        // Assert
        let error_message = result
            .expect_err("expected Err for kill-by-signal")
            .to_string();
        assert!(
            error_message.contains("[Stopped]"),
            "error was: {error_message}"
        );

        // Drain `PidUpdate` events and verify no loader update was emitted.
        while let Ok(event) = events_rx.try_recv() {
            assert!(
                matches!(event, TurnEvent::PidUpdate(_)),
                "only PidUpdate events expected, got: {event:?}"
            );
        }
    }

    #[tokio::test]
    /// Verifies that a clean process exit returns `Ok(TurnResult)` with no
    /// context reset (CLI turns never reset context).
    async fn test_run_turn_clean_exit_returns_ok_result_without_context_reset() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning(|_| {
            let mut command = std::process::Command::new("sh");
            command
                .arg("-c")
                .arg("printf '{\"answer\":\"ok\",\"questions\":[],\"summary\":null}'");

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let result = channel.run_turn("sess-1".to_string(), req, events_tx).await;

        // Assert
        let turn_result = result.expect("expected Ok for clean exit");
        assert!(!turn_result.context_reset);
    }

    #[tokio::test]
    /// Verifies strict turn parsing recovers one trailing protocol payload
    /// when Claude prepends extra prose before the final JSON object.
    async fn test_run_turn_recovers_wrapped_structured_output_for_claude() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning(|_| {
            let mut command = std::process::Command::new("sh");
            command.arg("-c").arg(concat!(
                "printf '%s\\n' 'Now I have the full context.';",
                "printf '%s' '{\"answer\":\"ok\",\"questions\":[],\"summary\":null}'",
            ));

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let result = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect("turn should succeed");

        // Assert
        assert_eq!(result.assistant_message.to_answer_display_text(), "ok");
    }

    #[tokio::test]
    /// Verifies session-turn Claude responses synthesize an empty summary when
    /// the provider returns `summary: null`.
    async fn test_run_turn_fills_missing_summary_for_session_turn() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning(|_| {
            let mut command = std::process::Command::new("sh");
            command
                .arg("-c")
                .arg("printf '{\"answer\":\"ok\",\"questions\":[],\"summary\":null}'");

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let result = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect("turn should succeed");

        // Assert
        assert_eq!(
            result.assistant_message.summary,
            Some(AgentResponseSummary {
                turn: String::new(),
                session: String::new(),
            })
        );
    }

    #[tokio::test]
    /// Verifies Claude CLI turns avoid deadlock when the child emits stderr
    /// before it starts reading a large stdin prompt.
    async fn test_run_turn_writes_large_stdin_concurrently_for_claude() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning(|_| {
            let mut command = std::process::Command::new("sh");
            command.arg("-c").arg(
                "printf 'warming up\\n' >&2; sleep 0.1; cat >/dev/null; printf '%s' \
                 '{\"answer\":\"ok\",\"questions\":[],\"summary\":null}'",
            );

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let mut req = make_turn_request(dir.path().to_path_buf());
        req.prompt = "x".repeat(512 * 1024).into();

        // Act
        let result = tokio::time::timeout(
            Duration::from_secs(5),
            channel.run_turn("sess-1".to_string(), req, events_tx),
        )
        .await
        .expect("turn should not deadlock")
        .expect("turn should succeed");

        // Assert
        assert_eq!(result.assistant_message.to_display_text(), "ok");
    }

    #[tokio::test]
    /// Verifies Claude CLI turns stream image-aware prompt text through stdin
    /// so large multimodal session prompts do not rely on argv transport.
    async fn test_run_turn_writes_prompt_to_stdin_for_claude() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let capture_path = dir.path().join("stdin.txt");
        let image_path = dir.path().join("pasted-image.png");
        std::fs::write(&image_path, b"image-bytes").expect("image should be written");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning({
            let capture_path = capture_path.clone();

            move |_| Ok(stdin_capture_command(&capture_path))
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let mut req = make_turn_request(dir.path().to_path_buf());
        req.prompt = TurnPrompt {
            attachments: vec![TurnPromptAttachment {
                placeholder: "[Image #1]".to_string(),
                local_image_path: image_path.clone(),
            }],
            text: "Review [Image #1]".to_string(),
            text_source: TurnPromptTextSource::UserPrompt,
        };

        // Act
        let result = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect("turn should succeed");
        let captured_prompt =
            std::fs::read_to_string(&capture_path).expect("captured stdin payload should exist");

        // Assert
        assert_eq!(result.assistant_message.to_display_text(), "ok");
        assert!(captured_prompt.contains("Structured response protocol:"));
        assert!(captured_prompt.contains(image_path.to_string_lossy().as_ref()));
        assert!(!captured_prompt.contains("[Image #1]"));
    }

    #[tokio::test]
    /// Verifies a broken stdin pipe does not hide the backend stderr or exit
    /// status when the CLI exits before consuming the full prompt.
    async fn test_run_turn_preserves_child_error_after_broken_pipe() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning(|_| {
            let mut command = std::process::Command::new("sh");
            command.arg("-c").arg("printf 'auth failed' >&2; exit 9");

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let mut req = make_turn_request(dir.path().to_path_buf());
        req.prompt = "x".repeat(512 * 1024).into();

        // Act
        let error = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect_err("turn should surface the child exit");

        // Assert
        let error_message = error.to_string();
        assert!(
            error_message.contains("auth failed"),
            "error was: {error_message}"
        );
        assert!(
            !error_message.contains("stdin payload"),
            "stdin write error should not mask child failure: {error_message}"
        );
    }

    #[tokio::test]
    /// Verifies Claude turns surface invalid structured output after both the
    /// original parse and the protocol-repair retry fail.
    async fn test_run_turn_returns_error_for_invalid_structured_output_for_claude() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend
            .expect_build_command()
            .times(2)
            .returning(|request| {
                assert!(matches!(
                    request.request_kind,
                    AgentRequestKind::SessionStart
                ));

                let mut command = std::process::Command::new("sh");
                command.arg("-c").arg("printf 'plain non-json response'");

                Ok(command)
            });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let error = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect_err("invalid structured output should fail");

        // Assert
        let error_message = error.to_string();
        assert!(error_message.contains("did not match the required JSON schema"));
        assert!(error_message.contains("response:\nplain non-json response"));
    }

    #[tokio::test]
    /// Verifies Claude turns recover valid output when the initial parse fails
    /// but the protocol-repair retry returns valid protocol JSON.
    async fn test_run_turn_recovers_valid_output_via_protocol_repair_for_claude() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let call_counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().times(2).returning({
            let counter = Arc::clone(&call_counter);

            move |_| {
                let call_number = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let mut command = std::process::Command::new("sh");

                if call_number == 0 {
                    command.arg("-c").arg("printf 'plain non-json response'");
                } else {
                    command.arg("-c").arg(
                        r#"printf '{"answer":"Repaired response","questions":[],"summary":null}'"#,
                    );
                }

                Ok(command)
            }
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let result = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect("repair retry should succeed");

        // Assert
        assert_eq!(
            result.assistant_message.to_display_text(),
            "Repaired response"
        );
    }

    #[tokio::test]
    /// Verifies Antigravity preserves useful prose when both strict parsing
    /// and the protocol-repair retry produce non-JSON text.
    async fn test_run_turn_falls_back_to_plain_text_for_antigravity_after_repair_failure() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let call_counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().times(2).returning({
            let counter = Arc::clone(&call_counter);

            move |_| {
                let call_number = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let mut command = std::process::Command::new("sh");

                if call_number == 0 {
                    command
                        .arg("-c")
                        .arg("cat >/dev/null; printf 'Plain Antigravity response'");
                } else {
                    command
                        .arg("-c")
                        .arg("cat >/dev/null; printf 'Still not JSON'");
                }

                Ok(command)
            }
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Antigravity,
        };
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let mut req = make_turn_request(dir.path().to_path_buf());
        req.model = AgentModel::Gemini31ProPreview
            .provider_model_str()
            .to_string();

        // Act
        let result = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect("Antigravity prose should fall back to a plain answer");

        // Assert
        assert_eq!(
            result.assistant_message.to_display_text(),
            "Plain Antigravity response"
        );
        assert_eq!(
            result.assistant_message.summary,
            Some(AgentResponseSummary {
                session: String::new(),
                turn: String::new(),
            })
        );

        let events = drain_events(&mut events_rx);
        assert!(events.iter().any(|event| {
            matches!(
                event,
                TurnEvent::ThoughtDelta(text)
                    if text == "Protocol parse error; retrying schema repair for antigravity."
            )
        }));
        assert!(events.iter().all(|event| {
            !matches!(
                event,
                TurnEvent::ThoughtDelta(text)
                    if text.contains("debug_details") || text.contains("response:")
            )
        }));
    }

    #[tokio::test]
    /// Verifies non-zero CLI turn exits surface actionable Claude
    /// re-authentication guidance instead of protocol schema errors.
    async fn test_run_turn_returns_claude_auth_guidance_for_expired_token() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().times(1).returning(|_| {
            let mut command = std::process::Command::new("sh");
            command.arg("-c").arg(
                "printf '%s' \
                 '{\"type\":\"error\",\"error\":{\"type\":\"authentication_error\",\"message\":\"\
                 OAuth token has expired. Please obtain a new token or refresh your existing \
                 token.\"}}'; exit 1",
            );

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let error_message = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect_err("expired Claude auth should fail")
            .to_string();

        // Assert
        assert!(
            error_message.contains("Agent command failed because Claude authentication expired")
        );
        assert!(error_message.contains("`claude auth login`"));
        assert!(error_message.contains("`claude auth status`"));
    }

    #[tokio::test]
    /// Verifies non-zero CLI turn exits preserve generic stderr details for
    /// non-authentication failures.
    async fn test_run_turn_returns_exit_error_for_non_zero_status() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().times(1).returning(|_| {
            let mut command = std::process::Command::new("sh");
            command
                .arg("-c")
                .arg("printf '%s' 'assist failed' >&2; exit 7");

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, _events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let error_message = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect_err("non-zero exit should fail")
            .to_string();

        // Assert
        assert!(error_message.contains("Agent command failed with exit code 7"));
        assert!(error_message.contains("assist failed"));
    }

    #[tokio::test]
    /// Verifies CLI channels surface only transient loader text while the
    /// final assistant response is returned at turn completion.
    async fn test_run_turn_surfaces_only_loader_updates_for_strict_protocol_provider() {
        // Arrange
        let dir = tempdir().expect("failed to create temp dir");
        let mut mock_backend = MockAgentBackend::new();
        mock_backend.expect_build_command().returning(|_| {
            let mut command = std::process::Command::new("sh");
            command.arg("-c").arg(concat!(
                r#"echo '{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","name":"Bash"}]}}';"#,
                r#"echo '{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"streamed fragment"}]}}';"#,
                r#"echo '{"result":"{\"answer\":\"final answer\",\"questions\":[],\"summary\":null}","usage":{"input_tokens":5,"output_tokens":3}}'"#,
            ));

            Ok(command)
        });
        let channel = CliAgentChannel {
            backend: Arc::new(mock_backend),
            kind: AgentKind::Claude,
        };
        let (events_tx, mut events_rx) = mpsc::unbounded_channel();
        let req = make_turn_request(dir.path().to_path_buf());

        // Act
        let result = channel
            .run_turn("sess-1".to_string(), req, events_tx)
            .await
            .expect("turn should succeed");

        // Assert
        let mut saw_loader_update = false;
        while let Ok(event) = events_rx.try_recv() {
            if matches!(event, TurnEvent::ThoughtDelta(_)) {
                saw_loader_update = true;
            }
        }
        assert!(saw_loader_update, "loader updates should be streamed live");
        assert_eq!(result.assistant_message.to_display_text(), "final answer");
    }
}