zeph-acp 0.22.1

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

//! IDE-proxied shell executor via ACP `terminal/*` methods.
//!
//! When the IDE advertises `terminal` capability, the agent routes `bash` tool
//! calls through the IDE's integrated terminal instead of spawning a local process.
//! This keeps the terminal visible in the IDE UI and allows live output streaming.
//!
//! # Security
//!
//! All terminal commands require an [`AcpPermissionGate`] to request IDE confirmation.
//! Stdin writes are rate-limited and capped at 64 KiB (REQ-P23-1). Commands that
//! resolve to shell interpreters (`bash`, `sh`, `zsh`, etc.) trigger an explicit
//! warning in the permission prompt.
//!
//! # Terminal lifecycle
//!
//! ACP requires the terminal to remain alive until after the `tool_call_update`
//! notification containing `ToolCallContent::Terminal(terminal_id)` is emitted.
//! Call [`AcpShellExecutor::release_terminal`] only after that notification is sent.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use agent_client_protocol as acp;
use schemars::JsonSchema;
use serde::Deserialize;
use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken;
use zeph_tools::{
    ToolCall, ToolError, ToolOutput,
    executor::deserialize_params,
    registry::{InvocationHint, ToolDef},
};

use crate::{error::AcpError, permission::AcpPermissionGate};

const KILL_GRACE_TIMEOUT: Duration = Duration::from_secs(5);

/// Maximum stdin payload size (64 KiB). REQ-P23-1.
const MAX_STDIN_BYTES: usize = 65_536;

/// Bounded stdin channel capacity (back-pressure). MED-02.
const STDIN_CHANNEL_CAPACITY: usize = 16;

/// Bounded terminal message channel capacity.
///
/// Each concurrent bash/release/stdin tool call occupies one slot. 64 is
/// sufficient for any realistic IDE session; excess messages are dropped with
/// a warning rather than growing memory without bound.
const TERMINAL_CHANNEL_CAPACITY: usize = 64;

/// Stdin rate-limit interval — 100 msg/sec. MED-02.
const STDIN_RATE_INTERVAL: Duration = Duration::from_millis(10);

/// Shell interpreters that require explicit warning in permission prompt. REQ-P23-5.
const SHELL_INTERPRETERS: &[&str] = &["bash", "sh", "zsh", "fish", "dash"];

/// Transparent prefixes that wrap another command without changing its semantics.
const TRANSPARENT_PREFIXES: &[&str] = &["env", "command", "exec", "nice", "nohup", "time"];

/// Extract the effective command binary name from a shell command string.
///
/// Iteratively skips transparent prefixes (`env`, `command`, `exec`, etc.) and
/// env-var assignments (`FOO=bar`) to reach the real binary. Falls back to `"bash"`
/// if the command is empty.
fn extract_command_binary(command: &str) -> &str {
    // Split into tokens and skip leading env-var assignments and transparent prefixes.
    let mut tokens = command.split_whitespace().peekable();
    loop {
        match tokens.peek() {
            None => return "bash",
            Some(tok) => {
                // Skip env-var assignments.
                if tok.contains('=') {
                    tokens.next();
                    continue;
                }
                // Skip transparent prefix commands.
                let base = tok.rsplit('/').next().unwrap_or(tok);
                if TRANSPARENT_PREFIXES.contains(&base) {
                    tokens.next();
                    continue;
                }
                // First non-prefix, non-assignment token is the binary.
                let binary = tok.rsplit('/').next().unwrap_or(tok);
                return binary;
            }
        }
    }
}

struct ShellResult {
    output: String,
    exit_code: Option<u32>,
    terminal_id: String,
}

struct TerminalRequest {
    session_id: acp::schema::v1::SessionId,
    command: String,
    args: Vec<String>,
    cwd: Option<PathBuf>,
    timeout: Duration,
    reply: oneshot::Sender<Result<ShellResult, AcpError>>,
    /// When `Some`, intermediate terminal output chunks are sent as `ToolCallUpdate`
    /// notifications on this channel so the IDE can stream output live.
    /// The `tool_call_id` is the ACP tool call ID to update.
    stream_tx: Option<(mpsc::Sender<acp::schema::v1::SessionNotification>, String)>,
}

struct TerminalReleaseRequest {
    session_id: acp::schema::v1::SessionId,
    terminal_id: String,
}

struct StdinWriteRequest {
    session_id: acp::schema::v1::SessionId,
    terminal_id: acp::schema::v1::TerminalId,
    data: Vec<u8>,
    reply: oneshot::Sender<Result<(), AcpError>>,
}

enum TerminalMessage {
    Execute(TerminalRequest),
    Release(TerminalReleaseRequest),
    WriteStdin(StdinWriteRequest),
}

/// IDE-proxied shell executor.
///
/// Routes `bash` tool calls to the IDE terminal via ACP `terminal/*` methods.
/// Only constructed when the IDE advertises `terminal` capability.
#[derive(Clone)]
pub struct AcpShellExecutor {
    session_id: acp::schema::v1::SessionId,
    request_tx: mpsc::Sender<TerminalMessage>,
    permission_gate: Option<AcpPermissionGate>,
    timeout: Duration,
}

impl AcpShellExecutor {
    /// Create the executor and its background handler future.
    ///
    /// Spawn the returned future with `tokio::spawn`; it drives terminal
    /// create/execute/release requests forwarded from the `bash` and
    /// `bash_stdin` tools.
    pub fn new(
        conn: Arc<acp::ConnectionTo<acp::Client>>,
        session_id: acp::schema::v1::SessionId,
        permission_gate: Option<AcpPermissionGate>,
        timeout_secs: u64,
    ) -> (Self, impl std::future::Future<Output = ()>) {
        Self::with_timeout(
            conn,
            session_id,
            permission_gate,
            Duration::from_secs(timeout_secs),
        )
    }

    /// Create the executor with a configurable command timeout.
    pub fn with_timeout(
        conn: Arc<acp::ConnectionTo<acp::Client>>,
        session_id: acp::schema::v1::SessionId,
        permission_gate: Option<AcpPermissionGate>,
        timeout: Duration,
    ) -> (Self, impl std::future::Future<Output = ()>) {
        let (tx, rx) = mpsc::channel::<TerminalMessage>(TERMINAL_CHANNEL_CAPACITY);
        let handler = async move { run_terminal_handler(conn, rx).await };
        (
            Self {
                session_id,
                request_tx: tx,
                permission_gate,
                timeout,
            },
            handler,
        )
    }

    /// Release a terminal by ID after the `tool_call_update` notification has been sent.
    ///
    /// This must be called after the ACP `tool_call_update` containing
    /// `ToolCallContent::Terminal(terminal_id)` is emitted so that the IDE can
    /// still display the terminal output when it processes the notification.
    pub fn release_terminal(&self, terminal_id: String) {
        if let Err(e) = self
            .request_tx
            .try_send(TerminalMessage::Release(TerminalReleaseRequest {
                session_id: self.session_id.clone(),
                terminal_id,
            }))
        {
            tracing::warn!(error = %e, "terminal release dropped: handler channel full or closed");
        }
    }

    async fn handle_bash_stdin(&self, call: &ToolCall) -> Result<Option<ToolOutput>, ToolError> {
        // REQ-P23-2: blocked if no permission gate
        let gate = self
            .permission_gate
            .as_ref()
            .ok_or_else(|| ToolError::Blocked {
                command: "bash_stdin: permission gate required".into(),
            })?;

        let params: BashStdinParams = deserialize_params(&call.params)?;

        if params.data.len() > MAX_STDIN_BYTES {
            return Err(ToolError::InvalidParams {
                message: AcpError::StdinTooLarge {
                    size: params.data.len(),
                }
                .to_string(),
            });
        }
        let data = params.data.as_bytes().to_vec();

        // REQ-P23-5: warn when writing to a shell interpreter terminal.
        // Terminal IDs are opaque strings, but common practice is to include
        // the command name. We always request permission explicitly for stdin writes.
        let is_shell = SHELL_INTERPRETERS
            .iter()
            .any(|s| params.terminal_id.contains(s));
        let title = if is_shell {
            "bash_stdin [WARNING: stdin to shell interpreter — data will be executed as commands]"
                .to_string()
        } else {
            "bash_stdin".to_owned()
        };
        let fields = acp::schema::v1::ToolCallUpdateFields::new()
            .title(title)
            .raw_input(serde_json::json!({
                "terminal_id": params.terminal_id,
                "data_length": params.data.len(),
            }));
        let tool_call = acp::schema::v1::ToolCallUpdate::new("bash_stdin".to_owned(), fields);
        let allowed = gate
            .check_permission(self.session_id.clone(), tool_call)
            .await
            .map_err(|e| ToolError::InvalidParams {
                message: e.to_string(),
            })?;
        if !allowed {
            return Err(ToolError::Blocked {
                command: "bash_stdin: permission denied".into(),
            });
        }

        let terminal_id: acp::schema::v1::TerminalId = params.terminal_id.clone().into();
        let (reply_tx, reply_rx) = oneshot::channel();
        self.request_tx
            .send(TerminalMessage::WriteStdin(StdinWriteRequest {
                session_id: self.session_id.clone(),
                terminal_id,
                data,
                reply: reply_tx,
            }))
            .await
            .map_err(|_| ToolError::InvalidParams {
                message: "terminal handler closed".into(),
            })?;
        reply_rx
            .await
            .map_err(|_| ToolError::InvalidParams {
                message: "terminal handler closed".into(),
            })?
            .map_err(|e| ToolError::InvalidParams {
                message: e.to_string(),
            })?;

        Ok(Some(ToolOutput {
            tool_name: zeph_tools::ToolName::new("bash_stdin"),
            summary: format!(
                "wrote {} bytes to stdin of {}",
                params.data.len(),
                params.terminal_id
            ),
            blocks_executed: 1,
            filter_stats: None,
            diff: None,
            streamed: false,
            terminal_id: Some(params.terminal_id),
            locations: None,
            raw_response: None,
            claim_source: Some(zeph_tools::ClaimSource::Shell),
            ..Default::default()
        }))
    }

    async fn execute_shell(
        &self,
        command: String,
        args: Vec<String>,
        cwd: Option<PathBuf>,
        stream_tx: Option<(mpsc::Sender<acp::schema::v1::SessionNotification>, String)>,
    ) -> Result<ShellResult, AcpError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.request_tx
            .send(TerminalMessage::Execute(TerminalRequest {
                session_id: self.session_id.clone(),
                command,
                args,
                cwd,
                timeout: self.timeout,
                reply: reply_tx,
                stream_tx,
            }))
            .await
            .map_err(|_| AcpError::ChannelClosed)?;
        reply_rx.await.map_err(|_| AcpError::ChannelClosed)?
    }
}

#[derive(Deserialize, JsonSchema)]
struct BashParams {
    command: String,
    #[serde(default)]
    args: Vec<String>,
    #[serde(default)]
    cwd: Option<String>,
}

#[derive(Deserialize, JsonSchema)]
struct BashStdinParams {
    terminal_id: String,
    data: String,
}

impl zeph_tools::ToolExecutor for AcpShellExecutor {
    async fn execute(&self, _response: &str) -> Result<Option<ToolOutput>, ToolError> {
        Ok(None)
    }

    fn tool_definitions(&self) -> Vec<ToolDef> {
        let mut defs = vec![ToolDef {
            id: "bash".into(),
            description: "Execute a shell command in the IDE terminal.\n\nParameters: command (string, required) - shell command to run\nReturns: stdout/stderr combined with exit code\nErrors: Timeout; permission denied by IDE; command blocked by policy\nExample: {\"command\": \"cargo build\"}".into(),
            schema: schemars::schema_for!(BashParams),
            invocation: InvocationHint::ToolCall,
            output_schema: None,
            server_id: None,
        }];
        // REQ-P23-2: bash_stdin only available when a permission gate is present.
        if self.permission_gate.is_some() {
            defs.push(ToolDef {
                id: "bash_stdin".into(),
                description: "Write data to stdin of a running terminal process.\n\nParameters: terminal_id (string, required) - terminal to write to; data (string, required) - stdin data\nReturns: confirmation\nErrors: terminal not found; terminal process exited\nExample: {\"terminal_id\": \"term-1\", \"data\": \"yes\\n\"}".into(),
                schema: schemars::schema_for!(BashStdinParams),
                invocation: InvocationHint::ToolCall,
                output_schema: None,
                server_id: None,
            });
        }
        defs
    }

    async fn execute_tool_call(&self, call: &ToolCall) -> Result<Option<ToolOutput>, ToolError> {
        if call.tool_id == "bash_stdin" {
            return self.handle_bash_stdin(call).await;
        }
        if call.tool_id != "bash" {
            return Ok(None);
        }

        let params: BashParams = deserialize_params(&call.params)?;
        let cwd = params.cwd.map(PathBuf::from);

        let blocklist: Vec<String> = zeph_tools::DEFAULT_BLOCKED_COMMANDS
            .iter()
            .map(|s| (*s).to_owned())
            .collect();

        // Blocklist check — reject dangerous commands before hitting the permission gate.
        if let Some(pattern) = zeph_tools::check_blocklist(&params.command, &blocklist) {
            return Err(ToolError::Blocked { command: pattern });
        }
        // Also check args when the command is a shell interpreter (e.g. bash -c "rm -rf /").
        // This prevents args-field bypass: { command: "bash", args: ["-c", "blocked cmd"] }.
        if let Some(script) = zeph_tools::effective_shell_command(&params.command, &params.args)
            && let Some(pattern) = zeph_tools::check_blocklist(script, &blocklist)
        {
            return Err(ToolError::Blocked { command: pattern });
        }

        if self.permission_gate.is_none() {
            tracing::warn!(
                "AcpShellExecutor has no permission gate — only blocklist applies. \
                 Do not use in production without a permission gate."
            );
        }

        if let Some(gate) = &self.permission_gate {
            // Use the command binary as the cache key, not the tool_id ("bash").
            // This makes "Allow always" apply per binary (git, cargo, etc.).
            let cmd_binary = extract_command_binary(&params.command);
            let fields = acp::schema::v1::ToolCallUpdateFields::new()
                .title(cmd_binary.to_owned())
                .raw_input(serde_json::json!({ "command": params.command }));
            let tool_call = acp::schema::v1::ToolCallUpdate::new(cmd_binary.to_owned(), fields);
            let allowed = gate
                .check_permission(self.session_id.clone(), tool_call)
                .await
                .map_err(|e| ToolError::InvalidParams {
                    message: e.to_string(),
                })?;
            if !allowed {
                return Err(ToolError::Blocked {
                    command: params.command,
                });
            }
        }

        let result = self
            .execute_shell(params.command, params.args, cwd, None)
            .await
            .map_err(|e| ToolError::InvalidParams {
                message: e.to_string(),
            })?;

        let is_error = !matches!(result.exit_code, Some(0) | None);
        let summary = if is_error {
            format!(
                "[exit {}]\n{}",
                result.exit_code.unwrap_or(1),
                result.output
            )
        } else {
            result.output.clone()
        };
        let raw_response = Some(serde_json::json!({
            "stdout": result.output,
            "stderr": "",
            "interrupted": false,
            "isImage": false,
            "noOutputExpected": false
        }));

        Ok(Some(ToolOutput {
            tool_name: zeph_tools::ToolName::new("bash"),
            summary,
            blocks_executed: 1,
            filter_stats: None,
            diff: None,
            streamed: false,
            terminal_id: Some(result.terminal_id),
            locations: None,
            raw_response,
            claim_source: Some(zeph_tools::ClaimSource::Shell),
            ..Default::default()
        }))
    }

    zeph_tools::tool_executor_no_inner_defaults!();
}

async fn forward_stdin_via_ext(
    conn: &Arc<acp::ConnectionTo<acp::Client>>,
    session_id: &acp::schema::v1::SessionId,
    terminal_id: &acp::schema::v1::TerminalId,
    data: Vec<u8>,
) -> Result<(), AcpError> {
    use base64::Engine as _;
    let encoded = base64::engine::general_purpose::STANDARD.encode(&data);
    let params_json = serde_json::json!({
        "session_id": session_id.to_string(),
        "terminal_id": terminal_id.to_string(),
        "data": encoded,
    });
    let req = acp::UntypedMessage::new("terminal/write_stdin", params_json)
        .map_err(|e| AcpError::ClientError(e.to_string()))?;
    conn.send_request(req)
        .block_task()
        .await
        .map(|_| ())
        .map_err(|e| AcpError::ClientError(e.to_string()))
}

/// Background pump: drains bounded stdin channel at ≤100 msg/sec (MED-02).
///
/// REQ-P23-3: on any error from `ext_method`, cancels the token and exits.
async fn run_stdin_pump(
    conn: Arc<acp::ConnectionTo<acp::Client>>,
    session_id: acp::schema::v1::SessionId,
    terminal_id: acp::schema::v1::TerminalId,
    mut data_rx: mpsc::Receiver<Vec<u8>>,
    cancel: CancellationToken,
) {
    let mut interval = tokio::time::interval(STDIN_RATE_INTERVAL);
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    loop {
        let data = tokio::select! {
            () = cancel.cancelled() => break,
            msg = data_rx.recv() => match msg {
                Some(d) => d,
                None => break,
            },
        };
        // Rate-limit: wait for tick before forwarding. MED-02.
        tokio::select! {
            () = cancel.cancelled() => break,
            _ = interval.tick() => {}
        }
        if let Err(e) = forward_stdin_via_ext(&conn, &session_id, &terminal_id, data).await {
            // REQ-P23-3: no panics, log and cancel.
            tracing::warn!(%terminal_id, error = %e, "stdin pump error — cancelling");
            cancel.cancel();
            break;
        }
    }
}

async fn run_terminal_handler(
    conn: Arc<acp::ConnectionTo<acp::Client>>,
    mut rx: mpsc::Receiver<TerminalMessage>,
) {
    // Maps terminal_id -> (bounded stdin sender, CancellationToken). MED-02, REQ-P23-4.
    let mut stdin_pumps: std::collections::HashMap<
        String,
        (mpsc::Sender<Vec<u8>>, CancellationToken),
    > = std::collections::HashMap::new();

    while let Some(msg) = rx.recv().await {
        match msg {
            TerminalMessage::Execute(req) => {
                let result = execute_in_terminal(
                    &conn,
                    req.session_id,
                    req.command,
                    req.args,
                    req.cwd,
                    req.timeout,
                    req.stream_tx,
                )
                .await;
                // Cancel stdin pump when terminal completes. REQ-P23-4.
                if let Ok(ref shell_result) = result
                    && let Some((_, token)) = stdin_pumps.remove(&shell_result.terminal_id)
                {
                    token.cancel();
                }
                req.reply.send(result).ok();
            }
            TerminalMessage::Release(req) => {
                // Cancel stdin pump on release. REQ-P23-4.
                if let Some((_, token)) = stdin_pumps.remove(&req.terminal_id) {
                    token.cancel();
                }
                let tid = req.terminal_id.clone();
                let release_req =
                    acp::schema::v1::ReleaseTerminalRequest::new(req.session_id, req.terminal_id);
                if let Err(e) = conn.send_request(release_req).block_task().await {
                    tracing::warn!(
                        terminal_id = %tid,
                        error = %e,
                        "failed to release terminal"
                    );
                }
            }
            TerminalMessage::WriteStdin(req) => {
                let tid_str = req.terminal_id.to_string();

                // Lazily start a bounded pump task per terminal. MED-02.
                let (data_tx, cancel) = stdin_pumps.entry(tid_str).or_insert_with(|| {
                    let (tx, rx) = mpsc::channel::<Vec<u8>>(STDIN_CHANNEL_CAPACITY);
                    let token = CancellationToken::new();
                    // EXEMPT(#5144): per-terminal stdin pump with dedicated CancellationToken
                    // and map-based lifecycle (stdin_pumps); supervisor adds no value here.
                    tokio::spawn(run_stdin_pump(
                        conn.clone(),
                        req.session_id.clone(),
                        req.terminal_id.clone(),
                        rx,
                        token.clone(),
                    ));
                    (tx, token)
                });

                let result = if cancel.is_cancelled() {
                    Err(AcpError::BrokenPipe)
                } else {
                    // Bounded send — returns Err if channel is full (back-pressure).
                    data_tx.try_send(req.data).map_err(|_| AcpError::BrokenPipe)
                };

                req.reply.send(result).ok();
            }
        }
    }
}

/// Polling interval for terminal output streaming.
const STREAM_POLL_INTERVAL: Duration = Duration::from_millis(200);

/// Kill a terminal, then wait up to [`KILL_GRACE_TIMEOUT`] for it to exit.
async fn kill_terminal(
    conn: &Arc<acp::ConnectionTo<acp::Client>>,
    session_id: &acp::schema::v1::SessionId,
    terminal_id: &acp::schema::v1::TerminalId,
) -> Result<(), AcpError> {
    tracing::warn!(%terminal_id, "terminal command timed out — sending kill");
    let kill_req =
        acp::schema::v1::KillTerminalRequest::new(session_id.clone(), terminal_id.clone());
    conn.send_request(kill_req)
        .block_task()
        .await
        .map_err(|e| AcpError::ClientError(e.to_string()))?;
    let wait_again =
        acp::schema::v1::WaitForTerminalExitRequest::new(session_id.clone(), terminal_id.clone());
    let _ = tokio::time::timeout(
        KILL_GRACE_TIMEOUT,
        conn.send_request(wait_again).block_task(),
    )
    .await;
    Ok(())
}

/// Stream terminal output chunks to `notify_tx` while polling for process exit.
///
/// Returns the exit code once the process terminates or the timeout is reached.
async fn stream_until_exit(
    conn: &Arc<acp::ConnectionTo<acp::Client>>,
    session_id: &acp::schema::v1::SessionId,
    terminal_id: &acp::schema::v1::TerminalId,
    timeout: Duration,
    notify_tx: &mpsc::Sender<acp::schema::v1::SessionNotification>,
    tool_call_id: &str,
) -> Result<Option<u32>, AcpError> {
    let wait_req =
        acp::schema::v1::WaitForTerminalExitRequest::new(session_id.clone(), terminal_id.clone());
    let exit_future = conn.send_request(wait_req).block_task();
    tokio::pin!(exit_future);
    let deadline = tokio::time::Instant::now() + timeout;
    let mut last_output_len = 0usize;

    loop {
        tokio::select! {
            result = &mut exit_future => {
                return match result {
                    Ok(resp) => Ok(resp.exit_status.exit_code),
                    Err(e) => Err(AcpError::ClientError(e.to_string())),
                };
            }
            () = tokio::time::sleep(STREAM_POLL_INTERVAL) => {
                if tokio::time::Instant::now() >= deadline {
                    kill_terminal(conn, session_id, terminal_id).await?;
                    return Ok(Some(124u32));
                }
                let output_req =
                    acp::schema::v1::TerminalOutputRequest::new(session_id.clone(), terminal_id.clone());
                if let Ok(resp) = conn.send_request(output_req).block_task().await {
                    let new_data = resp.output.get(last_output_len..).unwrap_or("");
                    if !new_data.is_empty() {
                        last_output_len = resp.output.len();
                        let mut meta = serde_json::Map::new();
                        meta.insert(
                            "terminal_output".to_owned(),
                            serde_json::json!({
                                "terminal_id": terminal_id.to_string(),
                                "data": new_data,
                            }),
                        );
                        let update = acp::schema::v1::ToolCallUpdate::new(
                            tool_call_id.to_owned(),
                            acp::schema::v1::ToolCallUpdateFields::new(),
                        )
                        .meta(meta);
                        let notif = acp::schema::v1::SessionNotification::new(
                            session_id.clone(),
                            acp::schema::v1::SessionUpdate::ToolCallUpdate(update),
                        );
                        let _ = notify_tx.try_send(notif);
                    }
                }
            }
        }
    }
}

async fn execute_in_terminal(
    conn: &Arc<acp::ConnectionTo<acp::Client>>,
    session_id: acp::schema::v1::SessionId,
    command: String,
    args: Vec<String>,
    cwd: Option<PathBuf>,
    timeout: Duration,
    stream_tx: Option<(mpsc::Sender<acp::schema::v1::SessionNotification>, String)>,
) -> Result<ShellResult, AcpError> {
    // 1. Create terminal.
    let create_req = acp::schema::v1::CreateTerminalRequest::new(session_id.clone(), command)
        .args(args)
        .cwd(cwd);
    let create_resp = conn
        .send_request(create_req)
        .block_task()
        .await
        .map_err(|e| AcpError::ClientError(e.to_string()))?;
    let terminal_id = create_resp.terminal_id;

    // 2. Wait for exit with timeout; kill if exceeded.
    let exit_code = if let Some((ref notify_tx, ref tool_call_id)) = stream_tx {
        stream_until_exit(
            conn,
            &session_id,
            &terminal_id,
            timeout,
            notify_tx,
            tool_call_id,
        )
        .await?
    } else {
        let wait_req = acp::schema::v1::WaitForTerminalExitRequest::new(
            session_id.clone(),
            terminal_id.clone(),
        );
        match tokio::time::timeout(timeout, conn.send_request(wait_req).block_task()).await {
            Ok(Ok(resp)) => resp.exit_status.exit_code,
            Ok(Err(e)) => return Err(AcpError::ClientError(e.to_string())),
            Err(_) => {
                kill_terminal(conn, &session_id, &terminal_id).await?;
                Some(124u32)
            }
        }
    };

    // 3. Get final output. Terminal is NOT released here — the caller releases it
    //    after the ACP `tool_call_update` notification carrying `ToolCallContent::Terminal`
    //    has been sent, so the IDE can still display the terminal output.
    let output_req =
        acp::schema::v1::TerminalOutputRequest::new(session_id.clone(), terminal_id.clone());
    let output_resp = conn
        .send_request(output_req)
        .block_task()
        .await
        .map_err(|e| AcpError::ClientError(e.to_string()))?;

    // 4. Emit terminal_exit notification if streaming is active.
    if let Some((ref notify_tx, ref tool_call_id)) = stream_tx {
        let mut meta = serde_json::Map::new();
        meta.insert(
            "terminal_exit".to_owned(),
            serde_json::json!({ "terminal_id": terminal_id.to_string(), "exit_code": exit_code }),
        );
        let update = acp::schema::v1::ToolCallUpdate::new(
            tool_call_id.clone(),
            acp::schema::v1::ToolCallUpdateFields::new(),
        )
        .meta(meta);
        let notif = acp::schema::v1::SessionNotification::new(
            session_id.clone(),
            acp::schema::v1::SessionUpdate::ToolCallUpdate(update),
        );
        let _ = notify_tx.try_send(notif);
    }

    // Terminal release is handled by AcpShellExecutor::release_terminal via TerminalMessage::Release.
    Ok(ShellResult {
        output: output_resp.output,
        exit_code,
        terminal_id: terminal_id.to_string(),
    })
}