winx_code_agent/daemon/

control.rs

use std::collections::HashMap;
use std::io::ErrorKind;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};

use tokio::sync::Mutex;

use super::lifecycle::{GuardianLifecycle, GuardianLimits};
use super::protocol::{
    read_json_frame, write_json_frame, CancelActionParams, ConfigureSessionParams,
    DaemonProcessRole, HelloResult, JournalReadParams, PruneParams, RpcError, RpcRequest,
    RpcResponse, RunActionParams, SessionInfo, SessionParams,
    CANCELLABLE_ACTION_RESERVATIONS_CAPABILITY, COMPACT_ACTION_OUTPUT_CAPABILITY,
    GENERATION_BOUND_ACTIONS_CAPABILITY, PROTOCOL_MAJOR, TYPED_ACTION_RESULT_CAPABILITY,
};
use crate::errors::{Result, WinxError};
use crate::types::normalize_thread_id;

/// Stable control plane. Each logical session is owned by a separate guardian
/// process, so restarting this process does not close any PTY master.
pub struct ControlServer {
    listener: crate::daemon::transport::DaemonListener,
    socket_path: PathBuf,
    lifecycle: Arc<GuardianLifecycle>,
    guardian_capabilities: Arc<Mutex<HashMap<PathBuf, CachedGuardian>>>,
    guardian_negotiation_gates: Arc<Mutex<HashMap<PathBuf, GuardianNegotiationGate>>>,
    epoch: String,
}

#[derive(Clone)]
struct CachedGuardian {
    hello: HelloResult,
    last_seen: Instant,
    socket_identity: Option<SocketIdentity>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct SocketIdentity {
    device: u64,
    inode: u64,
    ctime_seconds: i64,
    ctime_nanoseconds: i64,
    mtime_seconds: i64,
    mtime_nanoseconds: i64,
}

struct GuardianNegotiationGate {
    gate: Arc<Mutex<()>>,
    last_seen: Instant,
}

const MAX_GUARDIAN_NEGOTIATIONS: usize = 256;
const GUARDIAN_NEGOTIATION_TTL: Duration = Duration::from_secs(5 * 60);

impl ControlServer {
    pub async fn bind(socket_path: impl AsRef<Path>) -> Result<Self> {
        let socket_path = socket_path.as_ref().to_path_buf();
        let parent = socket_path.parent().ok_or_else(|| {
            WinxError::ConfigurationError("daemon socket must have a parent directory".to_string())
        })?;
        tokio::fs::create_dir_all(parent).await?;
        crate::daemon::transport::restrict_to_owner(parent, 0o700).await?;
        let guardian_dir = parent.join("guardians");
        tokio::fs::create_dir_all(&guardian_dir).await?;
        crate::daemon::transport::restrict_to_owner(&guardian_dir, 0o700).await?;

        if tokio::fs::try_exists(&socket_path).await? {
            if crate::daemon::transport::connect(&socket_path).await.is_ok() {
                return Err(WinxError::ConfigurationError(format!(
                    "a winxd instance is already listening at {}",
                    socket_path.display()
                )));
            }
            tokio::fs::remove_file(&socket_path).await?;
        }
        let listener = crate::daemon::transport::DaemonListener::bind(&socket_path)?;
        crate::daemon::transport::restrict_to_owner(&socket_path, 0o600).await?;
        let lifecycle = Arc::new(GuardianLifecycle::new(
            guardian_dir,
            guardian_binary()?,
            GuardianLimits::from_env()?,
        ));

        Ok(Self {
            listener,
            socket_path,
            lifecycle,
            guardian_capabilities: Arc::new(Mutex::new(HashMap::new())),
            guardian_negotiation_gates: Arc::new(Mutex::new(HashMap::new())),
            epoch: format!("{:016x}", rand::random::<u64>()),
        })
    }

    pub async fn serve(mut self) -> Result<()> {
        self.lifecycle.clone().spawn_sweeper();
        loop {
            let stream = self.listener.accept().await?;
            if !crate::daemon::transport::peer_is_same_user(&stream)? {
                tracing::warn!("Rejected winxd control connection from a different uid");
                continue;
            }
            let lifecycle = self.lifecycle.clone();
            let guardian_capabilities = self.guardian_capabilities.clone();
            let guardian_negotiation_gates = self.guardian_negotiation_gates.clone();
            let epoch = self.epoch.clone();
            tokio::spawn(async move {
                if let Err(error) = serve_connection(
                    stream,
                    lifecycle,
                    guardian_capabilities,
                    guardian_negotiation_gates,
                    epoch,
                )
                .await
                {
                    tracing::debug!("winxd control client disconnected: {error}");
                }
            });
        }
    }
}

impl Drop for ControlServer {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.socket_path);
    }
}

async fn serve_connection(
    mut stream: crate::daemon::transport::DaemonStream,
    lifecycle: Arc<GuardianLifecycle>,
    guardian_capabilities: Arc<Mutex<HashMap<PathBuf, CachedGuardian>>>,
    guardian_negotiation_gates: Arc<Mutex<HashMap<PathBuf, GuardianNegotiationGate>>>,
    epoch: String,
) -> Result<()> {
    loop {
        let request: RpcRequest = match read_json_frame(&mut stream).await {
            Ok(request) => request,
            Err(error) if error.kind() == ErrorKind::UnexpectedEof => return Ok(()),
            Err(error) => return Err(error.into()),
        };
        let response = dispatch(
            request,
            &lifecycle,
            &guardian_capabilities,
            &guardian_negotiation_gates,
            &epoch,
        )
        .await;
        write_json_frame(&mut stream, &response).await?;
    }
}

#[allow(clippy::too_many_lines)]
async fn dispatch(
    mut request: RpcRequest,
    lifecycle: &GuardianLifecycle,
    guardian_capabilities: &Mutex<HashMap<PathBuf, CachedGuardian>>,
    guardian_negotiation_gates: &Mutex<HashMap<PathBuf, GuardianNegotiationGate>>,
    epoch: &str,
) -> RpcResponse {
    if request.jsonrpc != "2.0" {
        return rpc_error(request.id, -32600, "JSON-RPC version must be 2.0");
    }
    if request.method == "winx.hello" {
        return rpc_result(
            request.id,
            &HelloResult::current(
                DaemonProcessRole::Control,
                vec![
                    "per_session_guardians".to_string(),
                    "planned_control_restart".to_string(),
                    "guardian_quota".to_string(),
                    "guardian_idle_ttl".to_string(),
                    "guardian_activity_clock".to_string(),
                    "unused_session_ttl".to_string(),
                    "quota_reclaims_unused".to_string(),
                    "session.configure".to_string(),
                    "session.negotiate".to_string(),
                    "shell.run_action".to_string(),
                    TYPED_ACTION_RESULT_CAPABILITY.to_string(),
                    COMPACT_ACTION_OUTPUT_CAPABILITY.to_string(),
                    GENERATION_BOUND_ACTIONS_CAPABILITY.to_string(),
                    CANCELLABLE_ACTION_RESERVATIONS_CAPABILITY.to_string(),
                    "session.list".to_string(),
                    "session.read_output".to_string(),
                    "session.kill".to_string(),
                    "session.prune".to_string(),
                    "multi_consumer_cursors".to_string(),
                    "idempotency".to_string(),
                    crate::daemon::protocol::PROCESS_SHUTDOWN_CAPABILITY.to_string(),
                ],
                epoch,
            ),
        );
    }
    if request.method == "winx.shutdown" {
        // Planned control restart: guardians are detached processes and keep
        // their PTYs; only the control plane exits.
        crate::daemon::server::schedule_process_exit("winxd");
        return rpc_result(request.id, &serde_json::json!({ "shutting_down": true }));
    }

    if request.method == "session.list" {
        return match list_sessions(lifecycle.guardian_dir()).await {
            Ok(sessions) => rpc_result(request.id, &sessions),
            Err(error) => rpc_error(request.id, -32000, &error.to_string()),
        };
    }

    if request.method == "session.prune" {
        let params = match decode::<PruneParams>(request.params) {
            Ok(params) => params,
            Err(error) => return rpc_error(request.id, -32602, &error.to_string()),
        };
        return match lifecycle.prune(params.idle_seconds).await {
            Ok(result) => {
                let mut cache = guardian_capabilities.lock().await;
                for thread_id in &result.removed_thread_ids {
                    cache.remove(&lifecycle.socket_for(thread_id));
                }
                rpc_result(request.id, &result)
            }
            Err(error) => rpc_error(request.id, -32003, &error.to_string()),
        };
    }

    let thread_id = match request_thread_id(&request) {
        Ok(thread_id) if !thread_id.is_empty() => thread_id,
        Ok(_) => return rpc_error(request.id, -32602, "explicit thread_id is required"),
        Err(error) => return rpc_error(request.id, -32602, &error.to_string()),
    };
    let guardian_socket = lifecycle.socket_for(&thread_id);
    if request.method == "session.negotiate" {
        return match negotiated_guardian(
            lifecycle,
            guardian_capabilities,
            guardian_negotiation_gates,
            &thread_id,
            &guardian_socket,
            true,
        )
        .await
        {
            Ok(hello) => rpc_result(request.id, &hello),
            Err(error) => rpc_error(request.id, -32001, &error.to_string()),
        };
    }
    if request.method == "session.kill" {
        match tokio::fs::try_exists(&guardian_socket).await {
            Ok(true) => {}
            Ok(false) => return rpc_result(request.id, &false),
            Err(error) => return rpc_error(request.id, -32002, &error.to_string()),
        }
    }

    let create_if_missing =
        matches!(request.method.as_str(), "session.configure" | "shell.run_action");
    let mut hello = match negotiated_guardian(
        lifecycle,
        guardian_capabilities,
        guardian_negotiation_gates,
        &thread_id,
        &guardian_socket,
        create_if_missing,
    )
    .await
    {
        Ok(hello) => hello,
        Err(error) => return rpc_error(request.id, -32001, &error.to_string()),
    };
    let live_precondition = match request_requires_live_guardian(&request) {
        Ok(required) => required,
        Err(error) => return rpc_error(request.id, -32602, &error.to_string()),
    };
    if !live_precondition {
        if let Err(error) = normalize_guardian_request(&mut request, &hello) {
            return rpc_error(request.id, -32602, &error.to_string());
        }
    }
    let kill_after = request.method == "session.kill";
    let request_id = request.id;
    let response = if live_precondition {
        match relay(&guardian_socket, request.clone()).await {
            Ok((response, effective)) => {
                hello = effective.clone();
                cache_guardian(guardian_capabilities, guardian_socket.clone(), effective).await;
                response
            }
            Err(error) => {
                guardian_capabilities.lock().await.remove(&guardian_socket);
                return rpc_error(request_id, -32602, &error.to_string());
            }
        }
    } else {
        match relay_negotiated(&guardian_socket, request.clone(), &hello).await {
            Ok(result) => result,
            Err(first_error) => {
                guardian_capabilities.lock().await.remove(&guardian_socket);
                let refreshed = match negotiated_guardian(
                    lifecycle,
                    guardian_capabilities,
                    guardian_negotiation_gates,
                    &thread_id,
                    &guardian_socket,
                    create_if_missing,
                )
                .await
                {
                    Ok(hello) => hello,
                    Err(error) => {
                        return rpc_error(
                            request_id,
                            -32002,
                            &format!("{first_error}; guardian renegotiation failed: {error}"),
                        )
                    }
                };
                if request.method == "shell.run_action"
                    && refreshed.daemon_epoch != hello.daemon_epoch
                {
                    return rpc_error(
                    request_id,
                    -32002,
                    &format!(
                        "{first_error}; guardian epoch changed, so winxd will not retry an ambiguously delivered shell action"
                    ),
                );
                }
                if let Err(error) = normalize_guardian_request(&mut request, &refreshed) {
                    return rpc_error(request_id, -32602, &error.to_string());
                }
                match relay_negotiated(&guardian_socket, request, &refreshed).await {
                    Ok(result) => {
                        hello = refreshed;
                        result
                    }
                    Err(error) => return rpc_error(request_id, -32002, &error.to_string()),
                }
            }
        }
    };

    if kill_after && response.error.is_none() {
        guardian_capabilities.lock().await.remove(&guardian_socket);
        if let Err(error) = lifecycle.finish_kill(&guardian_socket, hello.daemon_pid).await {
            return rpc_error(
                request_id,
                -32003,
                &format!("session state was removed, but guardian cleanup failed: {error}"),
            );
        }
    } else {
        lifecycle.note_activity(&guardian_socket, &thread_id, hello.daemon_pid).await;
    }
    response
}

fn request_thread_id(request: &RpcRequest) -> Result<String> {
    let thread_id = match request.method.as_str() {
        "session.configure" => {
            decode::<ConfigureSessionParams>(request.params.clone())?.snapshot.chat_id
        }
        "shell.run_action" => decode::<RunActionParams>(request.params.clone())?.command.thread_id,
        "shell.cancel_action" => decode::<CancelActionParams>(request.params.clone())?.thread_id,
        "session.read_output" => decode::<JournalReadParams>(request.params.clone())?.thread_id,
        "session.negotiate" | "session.info" | "session.kill" | "session.interrupt" => {
            decode::<SessionParams>(request.params.clone())?.thread_id
        }
        _ => {
            return Err(WinxError::InvalidInput(format!(
                "unsupported control method {}",
                request.method
            )))
        }
    };
    Ok(normalize_thread_id(&thread_id))
}

async fn negotiated_guardian(
    lifecycle: &GuardianLifecycle,
    cache: &Mutex<HashMap<PathBuf, CachedGuardian>>,
    gates: &Mutex<HashMap<PathBuf, GuardianNegotiationGate>>,
    thread_id: &str,
    socket: &Path,
    create_if_missing: bool,
) -> Result<HelloResult> {
    if let Some(hello) = cached_guardian(cache, socket).await {
        return Ok(hello);
    }
    let negotiation_gate = guardian_negotiation_gate(gates, socket).await?;
    let _single_flight = negotiation_gate.lock().await;
    if let Some(hello) = cached_guardian(cache, socket).await {
        return Ok(hello);
    }
    if !create_if_missing && !tokio::fs::try_exists(socket).await? {
        return Err(WinxError::ShellInitializationError(format!(
            "guardian at {} is unavailable",
            socket.display()
        )));
    }
    let hello = lifecycle.ensure_guardian(thread_id, socket).await?;
    cache_guardian(cache, socket.to_path_buf(), hello.clone()).await;
    Ok(hello)
}

async fn cached_guardian(
    cache: &Mutex<HashMap<PathBuf, CachedGuardian>>,
    socket: &Path,
) -> Option<HelloResult> {
    let socket_identity = socket_identity(socket).await;
    let now = Instant::now();
    let mut cache = cache.lock().await;
    prune_guardian_cache(&mut cache, now);
    if let Some(entry) = cache
        .get_mut(socket)
        .filter(|entry| socket_identity.is_some() && entry.socket_identity == socket_identity)
    {
        entry.last_seen = now;
        return Some(entry.hello.clone());
    }
    cache.remove(socket);
    None
}

fn prune_guardian_cache(cache: &mut HashMap<PathBuf, CachedGuardian>, now: Instant) {
    cache.retain(|_, entry| now.duration_since(entry.last_seen) <= GUARDIAN_NEGOTIATION_TTL);
}

async fn guardian_negotiation_gate(
    gates: &Mutex<HashMap<PathBuf, GuardianNegotiationGate>>,
    socket: &Path,
) -> Result<Arc<Mutex<()>>> {
    let now = Instant::now();
    let mut gates = gates.lock().await;
    gates.retain(|_, entry| {
        now.duration_since(entry.last_seen) <= GUARDIAN_NEGOTIATION_TTL
            || Arc::strong_count(&entry.gate) > 1
    });
    if let Some(entry) = gates.get_mut(socket) {
        entry.last_seen = now;
        return Ok(Arc::clone(&entry.gate));
    }
    if gates.len() >= MAX_GUARDIAN_NEGOTIATIONS {
        let oldest = gates
            .iter()
            .filter(|(_, entry)| Arc::strong_count(&entry.gate) == 1)
            .min_by_key(|(_, entry)| entry.last_seen)
            .map(|(path, _)| path.clone());
        if let Some(oldest) = oldest {
            gates.remove(&oldest);
        }
    }
    if gates.len() >= MAX_GUARDIAN_NEGOTIATIONS {
        return Err(WinxError::ResourceAllocationError {
            message: format!(
                "all {MAX_GUARDIAN_NEGOTIATIONS} guardian negotiation slots are active"
            ),
        });
    }
    let gate = Arc::new(Mutex::new(()));
    gates.insert(
        socket.to_path_buf(),
        GuardianNegotiationGate { gate: Arc::clone(&gate), last_seen: now },
    );
    Ok(gate)
}

async fn cache_guardian(
    cache: &Mutex<HashMap<PathBuf, CachedGuardian>>,
    socket: PathBuf,
    hello: HelloResult,
) {
    // Read filesystem identity before taking the global cache mutex. Besides
    // avoiding I/O under the lock, this makes an unlinked/recreated guardian
    // socket invalidate the old epoch immediately instead of waiting for TTL.
    let socket_identity = socket_identity(&socket).await;
    let now = Instant::now();
    let mut cache = cache.lock().await;
    prune_guardian_cache(&mut cache, now);
    if !cache.contains_key(&socket) && cache.len() >= MAX_GUARDIAN_NEGOTIATIONS {
        if let Some(oldest) =
            cache.iter().min_by_key(|(_, entry)| entry.last_seen).map(|(path, _)| path.clone())
        {
            cache.remove(&oldest);
        }
    }
    cache.insert(socket, CachedGuardian { hello, last_seen: now, socket_identity });
}

/// Identity of the socket artifact, used to notice a guardian that was
/// recreated behind the same path. Unix reads the inode; Windows uses the
/// marker file's creation time and size, which change on every re-bind.
async fn socket_identity(socket: &Path) -> Option<SocketIdentity> {
    let metadata = tokio::fs::metadata(socket).await.ok()?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::MetadataExt as _;
        Some(SocketIdentity {
            device: metadata.dev(),
            inode: metadata.ino(),
            ctime_seconds: metadata.ctime(),
            ctime_nanoseconds: metadata.ctime_nsec(),
            mtime_seconds: metadata.mtime(),
            mtime_nanoseconds: metadata.mtime_nsec(),
        })
    }
    #[cfg(not(unix))]
    {
        use std::os::windows::fs::MetadataExt as _;
        let modified = metadata
            .modified()
            .ok()
            .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok());
        Some(SocketIdentity {
            device: 0,
            inode: metadata.file_size(),
            ctime_seconds: i64::try_from(metadata.creation_time()).unwrap_or(i64::MAX),
            ctime_nanoseconds: 0,
            mtime_seconds: modified
                .map_or(0, |duration| i64::try_from(duration.as_secs()).unwrap_or(i64::MAX)),
            mtime_nanoseconds: modified.map_or(0, |duration| i64::from(duration.subsec_nanos())),
        })
    }
}

fn normalize_guardian_request(request: &mut RpcRequest, hello: &HelloResult) -> Result<()> {
    let has = |capability: &str| hello.capabilities.iter().any(|candidate| candidate == capability);
    if request.method == "shell.run_action" {
        let mut params = decode::<RunActionParams>(request.params.clone())?;
        if !has(TYPED_ACTION_RESULT_CAPABILITY) {
            return Err(WinxError::ConfigurationError(format!(
                "guardian pid {} does not advertise {TYPED_ACTION_RESULT_CAPABILITY}; terminate this durable session and initialize it again before running another BashCommand",
                hello.daemon_pid
            )));
        }
        if params
            .options
            .expected_guardian_epoch
            .as_deref()
            .is_some_and(|expected| expected != hello.daemon_epoch)
            || params
                .options
                .expected_execution
                .as_ref()
                .is_some_and(|expected| expected.guardian_epoch != hello.daemon_epoch)
        {
            return Err(WinxError::InvalidInput(
                "shell action precondition refers to a different effective guardian epoch"
                    .to_string(),
            ));
        }
        if (params.options.expected_generation.is_some()
            || params.options.expected_execution.is_some()
            || params.options.require_generation_binding)
            && !has(GENERATION_BOUND_ACTIONS_CAPABILITY)
        {
            return Err(WinxError::InvalidInput(
                "the running guardian does not support generation-bound shell actions".to_string(),
            ));
        }
        if params.options.cancellation_key.is_some()
            && !has(CANCELLABLE_ACTION_RESERVATIONS_CAPABILITY)
        {
            return Err(WinxError::InvalidInput(
                "the running guardian does not support cancellable action reservations".to_string(),
            ));
        }
        if !has(COMPACT_ACTION_OUTPUT_CAPABILITY) {
            params.options.compact_output = false;
        }
        request.params = serde_json::to_value(params)
            .map_err(|error| WinxError::SerializationError(error.to_string()))?;
    } else if request.method == "session.interrupt" {
        let params = decode::<SessionParams>(request.params.clone())?;
        if params
            .expected_guardian_epoch
            .as_deref()
            .is_some_and(|expected| expected != hello.daemon_epoch)
            || params
                .expected_execution
                .as_ref()
                .is_some_and(|expected| expected.guardian_epoch != hello.daemon_epoch)
        {
            return Err(WinxError::InvalidInput(
                "interrupt precondition refers to a different effective guardian epoch".to_string(),
            ));
        }
        if (params.expected_generation.is_some() || params.expected_execution.is_some())
            && !has(GENERATION_BOUND_ACTIONS_CAPABILITY)
        {
            return Err(WinxError::InvalidInput(
                "the running guardian does not support generation-bound interruption".to_string(),
            ));
        }
    } else if request.method == "shell.cancel_action" {
        let params = decode::<CancelActionParams>(request.params.clone())?;
        if !has(CANCELLABLE_ACTION_RESERVATIONS_CAPABILITY) {
            return Err(WinxError::InvalidInput(
                "the running guardian does not support cancellable action reservations".to_string(),
            ));
        }
        if params
            .expected_guardian_epoch
            .as_deref()
            .is_some_and(|expected| expected != hello.daemon_epoch)
        {
            return Err(WinxError::InvalidInput(
                "cancel precondition refers to a different effective guardian epoch".to_string(),
            ));
        }
    }
    Ok(())
}

fn request_requires_live_guardian(request: &RpcRequest) -> Result<bool> {
    match request.method.as_str() {
        "shell.run_action" => {
            let params = decode::<RunActionParams>(request.params.clone())?;
            Ok(params.options.require_generation_binding
                || params.options.expected_generation.is_some()
                || params.options.expected_execution.is_some()
                || params.options.expected_guardian_epoch.is_some()
                || params.options.cancellation_key.is_some())
        }
        "session.interrupt" => {
            let params = decode::<SessionParams>(request.params.clone())?;
            Ok(params.expected_generation.is_some()
                || params.expected_execution.is_some()
                || params.expected_guardian_epoch.is_some())
        }
        "shell.cancel_action" => Ok(true),
        _ => Ok(false),
    }
}

async fn relay(socket: &Path, mut request: RpcRequest) -> Result<(RpcResponse, HelloResult)> {
    let mut stream = crate::daemon::transport::connect(socket).await.map_err(|error| {
        WinxError::ShellInitializationError(format!(
            "guardian at {} is unavailable: {error}",
            socket.display()
        ))
    })?;
    let hello_id = rand::random::<u64>();
    write_json_frame(
        &mut stream,
        &RpcRequest {
            jsonrpc: "2.0".to_string(),
            id: hello_id,
            method: "winx.hello".to_string(),
            params: serde_json::json!({ "protocol_major": PROTOCOL_MAJOR }),
        },
    )
    .await?;
    let hello_response: RpcResponse = read_json_frame(&mut stream).await?;
    let hello: HelloResult = decode(hello_response.result.ok_or_else(|| {
        WinxError::ConfigurationError("guardian handshake omitted result".to_string())
    })?)?;
    if hello.protocol_major != PROTOCOL_MAJOR {
        return Err(WinxError::ConfigurationError(format!(
            "guardian protocol major {} is incompatible with control {}",
            hello.protocol_major, PROTOCOL_MAJOR
        )));
    }
    if request.method == "shell.run_action"
        && !hello.capabilities.iter().any(|capability| capability == TYPED_ACTION_RESULT_CAPABILITY)
    {
        return Err(WinxError::ConfigurationError(format!(
            "guardian pid {} does not advertise {TYPED_ACTION_RESULT_CAPABILITY}; terminate this durable session and initialize it again before running another BashCommand",
            hello.daemon_pid
        )));
    }
    normalize_guardian_request(&mut request, &hello)?;
    write_json_frame(&mut stream, &request).await?;
    let response: RpcResponse = read_json_frame(&mut stream).await?;
    Ok((response, hello))
}

async fn relay_negotiated(
    socket: &Path,
    request: RpcRequest,
    hello: &HelloResult,
) -> Result<RpcResponse> {
    let mut stream = crate::daemon::transport::connect(socket).await.map_err(|error| {
        WinxError::ShellInitializationError(format!(
            "guardian at {} is unavailable: {error}",
            socket.display()
        ))
    })?;
    write_json_frame(&mut stream, &request).await?;
    let response: RpcResponse = read_json_frame(&mut stream).await?;
    if response.id != request.id || response.jsonrpc != "2.0" {
        return Err(WinxError::ParseError(format!(
            "guardian pid {} returned a mismatched JSON-RPC response",
            hello.daemon_pid
        )));
    }
    Ok(response)
}

async fn list_sessions(guardian_dir: &Path) -> Result<Vec<SessionInfo>> {
    let mut sessions = Vec::new();
    let mut entries = tokio::fs::read_dir(guardian_dir).await?;
    while let Some(entry) = entries.next_entry().await? {
        let path = entry.path();
        if path.extension().and_then(|extension| extension.to_str()) != Some("sock") {
            continue;
        }
        let request = RpcRequest {
            jsonrpc: "2.0".to_string(),
            id: rand::random::<u64>(),
            method: "session.list".to_string(),
            params: serde_json::json!({}),
        };
        let Ok((response, hello)) = relay(&path, request).await else { continue };
        let Some(result) = response.result else { continue };
        if let Ok(mut guardian_sessions) = serde_json::from_value::<Vec<SessionInfo>>(result) {
            for session in &mut guardian_sessions {
                session.runtime = Some(hello.clone());
            }
            sessions.append(&mut guardian_sessions);
        }
    }
    sessions.sort_by(|left, right| left.thread_id.cmp(&right.thread_id));
    Ok(sessions)
}

fn decode<T: serde::de::DeserializeOwned>(value: serde_json::Value) -> Result<T> {
    serde_json::from_value(value)
        .map_err(|error| WinxError::DeserializationError(error.to_string()))
}

fn guardian_binary() -> Result<PathBuf> {
    if let Some(path) = std::env::var_os("WINX_GUARDIAN_BIN") {
        return Ok(PathBuf::from(path));
    }
    let executable = std::env::current_exe()?;
    let sibling = executable.with_file_name(if cfg!(windows) {
        "winx-guardian.exe"
    } else {
        "winx-guardian"
    });
    if sibling.is_file() {
        Ok(sibling)
    } else {
        Err(WinxError::ConfigurationError(format!(
            "winx-guardian not found beside {} (set WINX_GUARDIAN_BIN)",
            executable.display()
        )))
    }
}

fn rpc_result<T: serde::Serialize>(id: u64, result: &T) -> RpcResponse {
    match serde_json::to_value(result) {
        Ok(result) => {
            RpcResponse { jsonrpc: "2.0".to_string(), id, result: Some(result), error: None }
        }
        Err(error) => rpc_error(id, -32603, &error.to_string()),
    }
}

fn rpc_error(id: u64, code: i32, message: &str) -> RpcResponse {
    RpcResponse {
        jsonrpc: "2.0".to_string(),
        id,
        result: None,
        error: Some(RpcError { code, message: message.to_string() }),
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used)]

    use super::*;
    use crate::runtime::ShellActionOptions;
    use crate::state::bash_state::BashState;
    use crate::types::BashCommand;
    #[cfg(unix)]
    use tokio::net::UnixListener;

    fn guardian_1_4() -> HelloResult {
        HelloResult {
            protocol_major: PROTOCOL_MAJOR,
            protocol_minor: 4,
            capabilities: vec![TYPED_ACTION_RESULT_CAPABILITY.to_string()],
            max_frame_bytes: crate::daemon::protocol::MAX_FRAME_BYTES,
            daemon_epoch: "guardian-14".to_string(),
            daemon_pid: 14,
            process_role: None,
            build: None,
        }
    }

    #[test]
    fn guardian_1_4_normalization_preserves_adapter_wait_policy_semantics() {
        let command: BashCommand = serde_json::from_value(serde_json::json!({
            "command": "printf compatible",
            "wait_policy": "until_complete",
            "thread_id": "guardian14"
        }))
        .expect("adapter command");
        let mut request = RpcRequest {
            jsonrpc: "2.0".to_string(),
            id: 14,
            method: "shell.run_action".to_string(),
            params: serde_json::to_value(RunActionParams {
                snapshot: BashState::new().snapshot(),
                command,
                request_key: "request14".to_string(),
                consumer_id: "consumer14".to_string(),
                options: ShellActionOptions {
                    compact_output: true,
                    ..ShellActionOptions::default()
                },
            })
            .expect("run params"),
        };

        normalize_guardian_request(&mut request, &guardian_1_4()).expect("normalization");
        let wire = request.params.as_object().expect("params object");
        assert!(wire.get("options").is_none());
        assert!(wire["command"].get("wait_policy").is_none());
    }

    #[test]
    fn guardian_1_4_rejects_generation_bound_action_before_relay() {
        let command: BashCommand = serde_json::from_value(serde_json::json!({
            "command": "printf compatible",
            "thread_id": "guardian14"
        }))
        .expect("adapter command");
        let mut request = RpcRequest {
            jsonrpc: "2.0".to_string(),
            id: 15,
            method: "shell.run_action".to_string(),
            params: serde_json::to_value(RunActionParams {
                snapshot: BashState::new().snapshot(),
                command,
                request_key: "request15".to_string(),
                consumer_id: "consumer15".to_string(),
                options: ShellActionOptions {
                    expected_generation: Some(7),
                    ..ShellActionOptions::default()
                },
            })
            .expect("run params"),
        };
        assert!(matches!(
            normalize_guardian_request(&mut request, &guardian_1_4()),
            Err(WinxError::InvalidInput(_))
        ));
    }

    #[cfg(unix)] // mocks a guardian with a raw UnixListener
    #[tokio::test]
    async fn stale_1_5_precondition_sends_zero_action_frames_to_1_4_guardian() {
        let temp = tempfile::tempdir().expect("temporary directory");
        let socket = temp.path().join("guardian.sock");
        let listener = UnixListener::bind(&socket).expect("mock guardian");
        let guardian = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.expect("control connection");
            let hello: RpcRequest = read_json_frame(&mut stream).await.expect("hello frame");
            assert_eq!(hello.method, "winx.hello");
            write_json_frame(&mut stream, &rpc_result(hello.id, &guardian_1_4()))
                .await
                .expect("hello response");
            tokio::time::timeout(
                Duration::from_millis(100),
                read_json_frame::<_, RpcRequest>(&mut stream),
            )
            .await
            .is_ok_and(|frame| frame.is_ok())
        });
        let command: BashCommand = serde_json::from_value(serde_json::json!({
            "command": "touch must-not-run",
            "thread_id": "stale-guardian"
        }))
        .expect("command");
        let request = RpcRequest {
            jsonrpc: "2.0".to_string(),
            id: 16,
            method: "shell.run_action".to_string(),
            params: serde_json::to_value(RunActionParams {
                snapshot: BashState::new().snapshot(),
                command,
                request_key: "stale-request".to_string(),
                consumer_id: "stale-consumer".to_string(),
                options: ShellActionOptions {
                    require_generation_binding: true,
                    expected_guardian_epoch: Some("guardian-15".to_string()),
                    ..ShellActionOptions::default()
                },
            })
            .expect("params"),
        };

        assert!(relay(&socket, request).await.is_err());
        assert!(!guardian.await.expect("guardian task"));
    }

    #[tokio::test]
    async fn guardian_negotiation_cache_is_bounded_under_churn() {
        let cache = Mutex::new(HashMap::new());
        for index in 0..(MAX_GUARDIAN_NEGOTIATIONS + 50) {
            let mut hello = guardian_1_4();
            hello.daemon_epoch = format!("epoch-{index}");
            cache_guardian(&cache, PathBuf::from(format!("guardian-{index}.sock")), hello).await;
        }
        assert_eq!(cache.lock().await.len(), MAX_GUARDIAN_NEGOTIATIONS);
    }

    #[tokio::test]
    async fn concurrent_cold_guardian_negotiation_uses_one_gate() {
        let gates = Mutex::new(HashMap::new());
        let socket = PathBuf::from("same-guardian.sock");
        let (first, second) = tokio::join!(
            guardian_negotiation_gate(&gates, &socket),
            guardian_negotiation_gate(&gates, &socket)
        );
        assert!(Arc::ptr_eq(
            &first.expect("first negotiation gate"),
            &second.expect("concurrent negotiation gate")
        ));
        assert_eq!(gates.lock().await.len(), 1);
    }

    #[tokio::test]
    async fn guardian_negotiation_gates_apply_backpressure_without_evicting_active_slots() {
        let gates = Mutex::new(HashMap::new());
        let mut active = Vec::with_capacity(MAX_GUARDIAN_NEGOTIATIONS);
        for index in 0..MAX_GUARDIAN_NEGOTIATIONS {
            active.push(
                guardian_negotiation_gate(
                    &gates,
                    &PathBuf::from(format!("active-guardian-{index}.sock")),
                )
                .await
                .expect("active gate"),
            );
        }

        let overflow =
            guardian_negotiation_gate(&gates, &PathBuf::from("overflow-guardian.sock")).await;
        assert!(matches!(overflow, Err(WinxError::ResourceAllocationError { .. })));
        assert_eq!(gates.lock().await.len(), MAX_GUARDIAN_NEGOTIATIONS);

        drop(active.pop());
        guardian_negotiation_gate(&gates, &PathBuf::from("replacement-guardian.sock"))
            .await
            .expect("inactive slot can be replaced");
        assert_eq!(gates.lock().await.len(), MAX_GUARDIAN_NEGOTIATIONS);
    }

    #[cfg(unix)] // mocks a guardian with a raw UnixListener
    #[tokio::test]
    async fn recreated_guardian_socket_invalidates_negotiation_cache() {
        let temp = tempfile::tempdir().expect("temporary directory");
        let socket = temp.path().join("guardian.sock");
        let first_listener = UnixListener::bind(&socket).expect("first guardian socket");
        let cache = Mutex::new(HashMap::new());
        let hello = guardian_1_4();
        cache_guardian(&cache, socket.clone(), hello.clone()).await;

        assert_eq!(
            cached_guardian(&cache, &socket).await.expect("cached first guardian").daemon_epoch,
            hello.daemon_epoch
        );

        drop(first_listener);
        tokio::fs::remove_file(&socket).await.expect("remove first socket");
        let _second_listener = UnixListener::bind(&socket).expect("recreated guardian socket");

        assert!(cached_guardian(&cache, &socket).await.is_none());
        assert!(cache.lock().await.is_empty());
    }
}