zeph-acp 0.22.4

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

//! stdio transport for the ACP server.
//!
//! The IDE spawns the agent binary and communicates over the process's stdin/stdout
//! pipes using newline-delimited JSON-RPC 2.0 frames (the ACP wire format).
//!
//! Agent session tasks are spawned via `tokio::task::spawn_local` because `Agent<LoopbackChannel>`
//! is `!Send` (async method bodies hold internal references across await points). `serve_stdio`
//! wraps `run_agent` in a `LocalSet` directly. `serve_connection` requires the caller to provide
//! an enclosing `LocalSet`; the HTTP transport satisfies this with a per-connection thread.
//!
//! # SECURITY(layer-2): Session binding limitation
//!
//! stdio transport has no cryptographic session binding. Any process with access
//! to the pipe can inject messages. For multi-tenant scenarios, use the HTTP/WS
//! transport which provides bearer-token session binding.

use std::sync::Arc;

use agent_client_protocol as acp;
use futures::{AsyncWrite, AsyncWriteExt as _};
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
use zeph_memory::store::SqliteStore;

use crate::agent::{AgentSpawner, ZephAcpAgentState, run_agent};
use crate::error::AcpError;
use crate::transport::{AcpServerConfig, ReadyNotification};

async fn write_ready_notification<W>(
    writer: &mut W,
    ready: &ReadyNotification,
) -> Result<(), AcpError>
where
    W: AsyncWrite + Unpin,
{
    let mut payload = serde_json::Map::new();
    payload.insert(
        "version".into(),
        serde_json::Value::String(ready.version.clone()),
    );
    payload.insert("pid".into(), serde_json::Value::from(ready.pid));
    if let Some(log_file) = &ready.log_file {
        payload.insert(
            "log_file".into(),
            serde_json::Value::String(log_file.clone()),
        );
    }

    let frame = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "zeph/ready",
        "params": payload,
    });
    let line = serde_json::to_string(&frame).map_err(|e| AcpError::Transport(e.to_string()))?;
    writer
        .write_all(line.as_bytes())
        .await
        .map_err(|e| AcpError::Transport(e.to_string()))?;
    writer
        .write_all(b"\n")
        .await
        .map_err(|e| AcpError::Transport(e.to_string()))?;
    writer
        .flush()
        .await
        .map_err(|e| AcpError::Transport(e.to_string()))
}

/// Build a [`ZephAcpAgentState`] from the provided configuration.
///
/// Shared by stdio and HTTP transports. `owner_key` (#5868) is the authenticated ACP
/// connection identity that scopes persisted session list/load/resume — `"acp-local"` for
/// stdio, the matched bearer-token client id for authenticated HTTP/WS.
pub(crate) async fn build_agent_state(
    spawner: AgentSpawner,
    server_config: AcpServerConfig,
    owner_key: String,
) -> Arc<ZephAcpAgentState> {
    let mut agent = ZephAcpAgentState::new(
        spawner,
        server_config.max_sessions,
        server_config.session_idle_timeout_secs,
        server_config.permission_file,
    )
    .with_agent_info(server_config.agent_name, server_config.agent_version)
    .with_title_max_chars(server_config.title_max_chars)
    .with_max_history(server_config.max_history)
    .with_owner_key(owner_key);

    if let Some(ref path) = server_config.sqlite_path {
        match SqliteStore::new(path).await {
            Ok(store) => agent = agent.with_store(store),
            Err(e) => tracing::warn!(error = %e, "failed to open ACP SQLite store"),
        }
    }
    if let Some(data_dir) = server_config.session_data_dir {
        agent = agent.with_session_data_dir(data_dir);
    }
    if let Some(factory) = server_config.provider_factory {
        agent = agent.with_provider_factory(factory, server_config.available_models);
    }
    #[cfg(feature = "unstable-llm-providers")]
    {
        agent = agent.with_provider_names(server_config.provider_names);
    }
    if let Some(manager) = server_config.mcp_manager {
        agent = agent.with_mcp_manager(manager);
    }
    if !server_config.project_rules.is_empty() {
        agent = agent.with_project_rules(server_config.project_rules);
    }
    if !server_config.additional_directories.is_empty() {
        agent = agent.with_additional_directories(server_config.additional_directories);
    }
    if !server_config.auth_methods.is_empty() {
        agent = agent.with_auth_methods(server_config.auth_methods);
    }
    agent = agent.with_timeouts(server_config.timeouts);
    agent = agent.with_model_config(server_config.model_config);

    let state = Arc::new(agent);
    state.start_idle_reaper();
    state
}

/// Stack size for the dedicated stdio agent thread.
///
/// Agent futures are deeply nested (~512 KiB measured on overflow). The tokio
/// multi-thread runtime uses a 2 MiB worker-thread stack by default, which was
/// insufficient for complex sessions. 8 MiB provides comfortable headroom.
const ACP_AGENT_STACK_SIZE: usize = 8 * 1024 * 1024;

/// Run the ACP server over stdin/stdout until the connection closes.
///
/// Agent futures are `!Send` and deeply nested, so the dispatcher runs on a
/// dedicated OS thread with an 8 MiB stack and a `current_thread` Tokio runtime
/// rather than on the caller's multi-thread runtime worker (which uses the
/// default 2 MiB stack and would overflow for complex sessions).
///
/// # Errors
///
/// Returns `AcpError::Transport` if the underlying JSON-RPC I/O fails or if the
/// agent thread cannot be spawned.
pub async fn serve_stdio(
    spawner: AgentSpawner,
    server_config: AcpServerConfig,
) -> Result<(), AcpError> {
    let mut stdout = tokio::io::stdout().compat_write();

    if let Some(ready) = server_config.ready_notification.as_ref() {
        write_ready_notification(&mut stdout, ready).await?;
        tracing::info!(
            transport = "stdio",
            pid = ready.pid,
            version = %ready.version,
            log_file = ready.log_file.as_deref().unwrap_or("<disabled>"),
            "ACP server ready"
        );
    }

    let state = build_agent_state(
        spawner,
        server_config,
        crate::transport::OWNER_KEY_LOCAL.to_owned(),
    )
    .await;

    // Run the agent on a dedicated thread with a larger stack.
    // Agent session tasks use spawn_local (futures are !Send), so the thread
    // uses a current_thread runtime wrapped in a LocalSet.
    let (tx, rx) = tokio::sync::oneshot::channel::<Result<(), AcpError>>();
    std::thread::Builder::new()
        .name("acp-stdio".into())
        .stack_size(ACP_AGENT_STACK_SIZE)
        .spawn(move || {
            let rt = match tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
            {
                Ok(rt) => rt,
                Err(e) => {
                    let _ = tx.send(Err(AcpError::Transport(format!(
                        "failed to build tokio runtime: {e}"
                    ))));
                    return;
                }
            };
            let local = tokio::task::LocalSet::new();
            let result = rt
                .block_on(local.run_until(run_agent(
                    state,
                    acp::ByteStreams::new(stdout, tokio::io::stdin().compat()),
                )))
                .map_err(|e| AcpError::Transport(e.to_string()));
            let _ = tx.send(result);
        })
        .map_err(|e| AcpError::Transport(format!("failed to spawn stdio agent thread: {e}")))?;

    rx.await
        .map_err(|_| AcpError::Transport("stdio agent thread panicked".into()))?
}

/// Run the ACP server over arbitrary async byte streams.
///
/// Extracted from [`serve_stdio`] to allow integration tests to use
/// `tokio::io::duplex` or similar in-process transports. The caller must
/// ensure this future runs inside a `tokio::task::LocalSet` (or equivalent)
/// because agent session tasks are spawned via `spawn_local`.
///
/// The HTTP transport satisfies this requirement by running each connection
/// on a dedicated thread with a `current_thread` runtime and `LocalSet`.
///
/// `owner_key` (#5868) scopes this connection's persisted session list/load/resume — see
/// `build_agent_state`.
///
/// # Errors
///
/// Returns `AcpError::Transport` if the underlying JSON-RPC I/O fails.
pub async fn serve_connection<W, R>(
    spawner: AgentSpawner,
    server_config: AcpServerConfig,
    writer: W,
    reader: R,
    owner_key: String,
) -> Result<(), AcpError>
where
    W: futures::AsyncWrite + Unpin + Send + 'static,
    R: futures::AsyncRead + Unpin + Send + 'static,
{
    let state = build_agent_state(spawner, server_config, owner_key).await;
    tokio::task::LocalSet::new()
        .run_until(run_agent(state, acp::ByteStreams::new(writer, reader)))
        .await
        .map_err(|e| AcpError::Transport(e.to_string()))
}