ryo_server/

lib.rs

//! RYO Server - tarpc-based RPC server
//!
//! This crate provides the server implementation for RYO RPC.
//! It wraps the Api and exposes it via tarpc over Unix Domain Socket.
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────────┐
//! │  RyoServer                                                          │
//! │  ├── api: Arc<Mutex<Api>>  (single mutex for all operations)       │
//! │  ├── shutdown_tx: Shutdown signal sender                           │
//! │  └── last_activity: Atomic timestamp for idle detection            │
//! └─────────────────────────────────────────────────────────────────────┘
//!                              │
//!                              │ UDS + tarpc (serde_transport)
//!                              ▼
//! ┌─────────────────────────────────────────────────────────────────────┐
//! │  Clients (ryo-cli)                                                  │
//! │  └── RyoServiceClient                                               │
//! └─────────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Idle Timeout
//!
//! Server automatically shuts down after configurable idle time (default 1 hour).
//! Configure via `~/.ryo/config.toml`:
//! ```toml
//! [server]
//! idle_timeout = 3600  # 1 hour (0 = never timeout)
//! ```
//!
//! # Parallel Initialization
//!
//! By default, server uses parallel initialization for faster startup.
//! Configure via `~/.ryo/config.toml`:
//! ```toml
//! [server]
//! parallel_init = true
//! ```
//!
//! # Locking Strategy
//!
//! All operations acquire a single Mutex. This is sufficient because:
//! - Write operations (run) are high-frequency but short-duration (~100μs)
//! - Read operations (discover, suggest) are low-frequency
//! - Lock contention is ~1% (100 writes/sec × 100μs = 10ms/sec)
//!
//! RWLock overhead is unnecessary for this workload.

pub mod watcher;

use ryo_analysis::AnalysisContext;
use ryo_app::api::{
    Api, BorrowAnalysisRequest, BorrowAnalysisResponse, CascadeRequest, CascadeResponse,
    ChainAnalysisRequest, ChainAnalysisResponse, DiscoverRequest, DiscoverResponse,
    FlowAnalysisRequest, FlowAnalysisResponse, GraphSummaryRequest, GraphSummaryResponse,
    LiteralSearchRequest, LiteralSearchResponse, LockAnalysisRequest, LockAnalysisResponse,
    OverviewRequest, OverviewResponse, PingResponse, QueryResponse, RunRequest, RunResponse,
    RyoqlRequest, SpecRequest, SpecResponse, StatusResponse, SuggestApplyRequest,
    SuggestApplyResponse, SuggestChoicesRequest, SuggestChoicesResponse, SuggestCompareRequest,
    SuggestCompareResponse, SuggestGenerateRequest, SuggestGenerateResponse, SuggestRequest,
    SuggestResponse, SuggestVerifyRequest, SuggestVerifyResponse, TypeAnalysisRequest,
    TypeAnalysisResponse,
};
use ryo_app::service::{RyoError, RyoService};
use ryo_app::{InMemoryStorage, Project};
use ryo_storage::GlobalConfig;
use ryo_symbol::write_with_parents;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::{oneshot, Mutex};

/// Default idle timeout: 1 hour (from config default)
pub const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(60 * 60);

/// Get current timestamp in seconds since UNIX_EPOCH
fn now_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

/// Server startup options
#[derive(Debug, Clone)]
pub struct ServerOptions {
    /// Use parallel initialization
    pub parallel_init: bool,
    /// Idle timeout (None = never timeout)
    pub idle_timeout: Option<Duration>,
    /// Enable file watching for automatic reload
    pub watch: bool,
    /// Debounce duration for file watching (ms)
    pub watch_debounce_ms: u64,
}

impl Default for ServerOptions {
    fn default() -> Self {
        Self {
            parallel_init: true,
            idle_timeout: Some(DEFAULT_IDLE_TIMEOUT),
            watch: false,
            watch_debounce_ms: 500,
        }
    }
}

impl ServerOptions {
    /// Load options from global config (~/.ryo/config.toml)
    pub fn from_config() -> Self {
        let config = GlobalConfig::load_global().unwrap_or_default();
        Self {
            parallel_init: config.server.parallel_init,
            idle_timeout: config.server.idle_timeout_duration(),
            watch: config.server.watch, // Respect config setting
            watch_debounce_ms: config.server.watch_debounce_ms,
        }
    }

    /// Enable file watching
    pub fn with_watch(mut self, enabled: bool) -> Self {
        self.watch = enabled;
        self
    }
}

/// Server wraps Api and implements RyoService
///
/// ## Locking Strategy: Single Mutex
/// - All operations (read/write) acquire the same Mutex
/// - Rationale: Write is high-frequency but short-duration (~100μs)
/// - Lock contention is ~1% (100 writes/sec × 100μs = 10ms/sec)
/// - RWLock overhead is unnecessary for this workload
#[derive(Clone)]
pub struct RyoServer {
    /// Single Mutex for all operations
    api: Arc<Mutex<Api>>,
    /// Shutdown signal sender (shared via Arc for clone)
    shutdown_tx: Arc<Mutex<Option<oneshot::Sender<()>>>>,
    /// Last activity timestamp (seconds since UNIX_EPOCH)
    last_activity: Arc<AtomicU64>,
}

impl RyoServer {
    /// Create a new RyoServer with the given Api and shutdown channel
    pub fn new(api: Api, shutdown_tx: oneshot::Sender<()>) -> Self {
        Self {
            api: Arc::new(Mutex::new(api)),
            shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))),
            last_activity: Arc::new(AtomicU64::new(now_secs())),
        }
    }

    /// Update last activity timestamp (called on each RPC)
    fn touch(&self) {
        self.last_activity.store(now_secs(), Ordering::Relaxed);
    }

    /// Get seconds since last activity
    pub fn idle_secs(&self) -> u64 {
        now_secs().saturating_sub(self.last_activity.load(Ordering::Relaxed))
    }

    /// Reload the project (rebuild AnalysisContext)
    ///
    /// Called when files change to update the internal state.
    /// Preserves the suggestion store across reloads.
    pub async fn reload(&self, project_path: &std::path::Path) -> anyhow::Result<()> {
        let start = Instant::now();
        tracing::info!("Reloading project due to file changes...");

        // Take suggestion store from old api before replacing
        let mut api = self.api.lock().await;
        let old_store = api.take_suggest_store();
        let store_count = old_store.len();

        // Build new context
        let project = Project::load(project_path)?;
        let context = AnalysisContext::from_workspace_root_parallel(project.workspace_root())
            .map_err(|e| anyhow::anyhow!("Context rebuild failed: {}", e))?;
        let new_api = Api::with_context(context, project, Box::new(InMemoryStorage::new()));

        // Restore suggestion store to new api
        new_api.restore_suggest_store(old_store);

        // Replace the api
        *api = new_api;

        let status = api.status();
        tracing::info!(
            "Reloaded: {} symbols, {} files, {} suggestions preserved in {:.2}s",
            status.symbols,
            status.files,
            store_count,
            start.elapsed().as_secs_f64()
        );

        Ok(())
    }
}

impl RyoService for RyoServer {
    async fn ping(self, _: tarpc::context::Context) -> PingResponse {
        self.touch();
        PingResponse {
            version: format!("{}-{}", env!("CARGO_PKG_VERSION"), env!("RYO_COMMIT_HASH")),
        }
    }

    async fn status(self, _: tarpc::context::Context) -> StatusResponse {
        self.touch();
        let api = self.api.lock().await;
        api.status()
    }

    async fn shutdown(self, _: tarpc::context::Context) {
        // Save UUID mappings before shutdown for persistence
        {
            let api = self.api.lock().await;
            if let Err(e) = api.save_uuid_mappings() {
                eprintln!("Warning: Failed to save UUID mappings: {}", e);
            }
        }

        // Graceful shutdown: signal the server loop to stop
        if let Some(tx) = self.shutdown_tx.lock().await.take() {
            let _ = tx.send(());
        }
    }

    async fn discover(
        self,
        _: tarpc::context::Context,
        req: DiscoverRequest,
    ) -> Result<DiscoverResponse, RyoError> {
        self.touch();
        let mut api = self.api.lock().await;
        api.discover(req).map_err(Into::into)
    }

    async fn overview(
        self,
        _: tarpc::context::Context,
        req: OverviewRequest,
    ) -> Result<OverviewResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.overview(req).map_err(Into::into)
    }

    async fn run(
        self,
        _: tarpc::context::Context,
        req: RunRequest,
    ) -> Result<RunResponse, RyoError> {
        tracing::info!(
            "RPC run: intents={}, dry_run={}",
            req.goal.intents.len(),
            req.dry_run
        );
        self.touch();
        let is_dry_run = req.dry_run;
        let mut api = self.api.lock().await;
        tracing::debug!("Acquired API lock, executing run...");
        let response = api.run(req).map_err(|e| {
            tracing::error!("Run failed: {:?}", e);
            RyoError::from(e)
        })?;
        tracing::info!(
            "Run completed: success={}, files_modified={}",
            response.success,
            response.files_modified
        );

        // Write modified files to disk (server's responsibility in server mode)
        // INVARIANT: Only write when actual mutations occurred.
        // total_changes == 0 means no AST was modified → no file should be touched.
        if response.success
            && !is_dry_run
            && response.total_changes > 0
            && !response.modified_files.is_empty()
        {
            for path in &response.modified_files {
                if let Some(file) = api.project().get_file(path) {
                    let source = match file.to_source() {
                        Ok(s) => s,
                        Err(e) => {
                            tracing::error!("Failed to generate source for {:?}: {}", path, e);
                            return Ok(RunResponse {
                                success: false,
                                error: Some(format!(
                                    "Failed to generate source for {:?}: {}",
                                    path, e
                                )),
                                ..response
                            });
                        }
                    };
                    // Use write_with_parents to create intermediate directories if needed
                    if let Err(e) = write_with_parents(path, &source) {
                        tracing::error!("Failed to write file {:?}: {}", path, e);
                        return Ok(RunResponse {
                            success: false,
                            error: Some(format!("Failed to write file {:?}: {}", path, e)),
                            ..response
                        });
                    }
                    tracing::debug!("Wrote {} bytes to {:?}", source.len(), path);
                }
            }
            tracing::info!("Wrote {} files to disk", response.modified_files.len());
        }

        Ok(response)
    }

    async fn cascade(
        self,
        _: tarpc::context::Context,
        req: CascadeRequest,
    ) -> Result<CascadeResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.graph_cascade(req).map_err(Into::into)
    }

    async fn graph_summary(
        self,
        _: tarpc::context::Context,
        req: GraphSummaryRequest,
    ) -> Result<GraphSummaryResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.graph_summary(req).map_err(Into::into)
    }

    async fn graph_type(
        self,
        _: tarpc::context::Context,
        req: TypeAnalysisRequest,
    ) -> Result<TypeAnalysisResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.graph_type(req).map_err(Into::into)
    }

    async fn graph_flow(
        self,
        _: tarpc::context::Context,
        req: FlowAnalysisRequest,
    ) -> Result<FlowAnalysisResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.graph_flow(req).map_err(Into::into)
    }

    async fn graph_borrow(
        self,
        _: tarpc::context::Context,
        req: BorrowAnalysisRequest,
    ) -> Result<BorrowAnalysisResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.graph_borrow(req).map_err(Into::into)
    }

    async fn graph_lock(
        self,
        _: tarpc::context::Context,
        req: LockAnalysisRequest,
    ) -> Result<LockAnalysisResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.graph_lock(req).map_err(Into::into)
    }

    async fn graph_chain(
        self,
        _: tarpc::context::Context,
        req: ChainAnalysisRequest,
    ) -> Result<ChainAnalysisResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.graph_chain(req).map_err(Into::into)
    }

    async fn suggest(
        self,
        _: tarpc::context::Context,
        req: SuggestRequest,
    ) -> Result<SuggestResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.suggest(req).map_err(Into::into)
    }

    async fn suggest_apply(
        self,
        _: tarpc::context::Context,
        req: SuggestApplyRequest,
    ) -> Result<SuggestApplyResponse, RyoError> {
        self.touch();
        let mut api = self.api.lock().await;
        api.suggest_apply(req).map_err(Into::into)
    }

    async fn suggest_choices(
        self,
        _: tarpc::context::Context,
        req: SuggestChoicesRequest,
    ) -> Result<SuggestChoicesResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.suggest_choices(req).map_err(Into::into)
    }

    async fn suggest_verify(
        self,
        _: tarpc::context::Context,
        req: SuggestVerifyRequest,
    ) -> Result<SuggestVerifyResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.suggest_verify(req).map_err(Into::into)
    }

    async fn suggest_compare(
        self,
        _: tarpc::context::Context,
        req: SuggestCompareRequest,
    ) -> Result<SuggestCompareResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.suggest_compare(req).map_err(Into::into)
    }

    async fn suggest_generate(
        self,
        _: tarpc::context::Context,
        req: SuggestGenerateRequest,
    ) -> Result<SuggestGenerateResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.suggest_generate(req).map_err(Into::into)
    }

    async fn spec(
        self,
        _: tarpc::context::Context,
        req: SpecRequest,
    ) -> Result<SpecResponse, RyoError> {
        self.touch();
        let mut api = self.api.lock().await;
        api.spec(req).map_err(Into::into)
    }

    async fn query_ryoql(
        self,
        _: tarpc::context::Context,
        req: RyoqlRequest,
    ) -> Result<QueryResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.query_ryoql(req).map_err(Into::into)
    }

    async fn search_literal(
        self,
        _: tarpc::context::Context,
        req: LiteralSearchRequest,
    ) -> Result<LiteralSearchResponse, RyoError> {
        self.touch();
        let api = self.api.lock().await;
        api.search_literal(req).map_err(Into::into)
    }
}

/// Run the server on Unix Domain Socket with default options from config
///
/// # Arguments
/// * `socket_path` - Path to the Unix socket file
/// * `project_path` - Path to the project root
///
/// # Example
/// ```ignore
/// run_server("/tmp/ryo.sock".into(), "/path/to/project".into()).await?;
/// ```
pub async fn run_server(socket_path: PathBuf, project_path: PathBuf) -> anyhow::Result<()> {
    let opts = ServerOptions::from_config();
    run_server_with_options(socket_path, project_path, opts).await
}

/// Run the server on Unix Domain Socket with configurable idle timeout (legacy)
///
/// # Arguments
/// * `socket_path` - Path to the Unix socket file
/// * `project_path` - Path to the project root
/// * `idle_timeout` - Optional idle timeout (None = no timeout)
pub async fn run_server_with_timeout(
    socket_path: PathBuf,
    project_path: PathBuf,
    idle_timeout: Option<Duration>,
) -> anyhow::Result<()> {
    let config = GlobalConfig::load_global().unwrap_or_default();
    let opts = ServerOptions {
        parallel_init: config.server.parallel_init,
        idle_timeout,
        watch: false,
        watch_debounce_ms: 500,
    };
    run_server_with_options(socket_path, project_path, opts).await
}

/// Run the server on Unix Domain Socket with full options
///
/// # Arguments
/// * `socket_path` - Path to the Unix socket file
/// * `project_path` - Path to the project root
/// * `options` - Server options including parallel_init and idle_timeout
pub async fn run_server_with_options(
    socket_path: PathBuf,
    project_path: PathBuf,
    options: ServerOptions,
) -> anyhow::Result<()> {
    use futures::StreamExt;
    use tarpc::server::{self, Channel};
    use tokio::net::UnixListener;

    let start = Instant::now();

    tracing::info!("━━━ RYO Server Starting ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
    tracing::info!("   Project: {:?}", project_path);
    tracing::info!("   Socket:  {:?}", socket_path);
    tracing::info!("   Parallel init: {}", options.parallel_init);
    tracing::info!(
        "   Watch mode:    {}",
        if options.watch { "enabled" } else { "disabled" }
    );
    if let Some(timeout) = options.idle_timeout {
        tracing::info!("   Idle timeout:  {}s", timeout.as_secs());
    } else {
        tracing::info!("   Idle timeout:  disabled (daemon mode)");
    }

    // Initialize Api with parallel or sequential loading
    tracing::info!("   Loading project...");

    let api = if options.parallel_init {
        // Use parallel initialization for faster startup
        let project = Project::load(&project_path)?;
        let context = AnalysisContext::from_workspace_root_parallel(project.workspace_root())
            .map_err(|e| anyhow::anyhow!("Context build failed: {}", e))?;
        Api::with_context(context, project, Box::new(InMemoryStorage::new()))
    } else {
        // Sequential initialization (legacy)
        Api::from_path(&project_path)?
    };

    let status = api.status();
    let load_time = start.elapsed();
    tracing::info!(
        "   Loaded: {} symbols, {} files in {:.2}s",
        status.symbols,
        status.files,
        load_time.as_secs_f64()
    );

    // Setup graceful shutdown
    let (shutdown_tx, shutdown_rx) = oneshot::channel();
    let server = RyoServer::new(api, shutdown_tx);

    // Cleanup stale socket file if exists
    let _ = std::fs::remove_file(&socket_path);
    let listener = UnixListener::bind(&socket_path)?;

    tracing::info!("━━━ Server Ready ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
    tracing::info!("   Listening on {:?}", socket_path);

    // Setup file watcher if enabled
    let mut file_watcher = if options.watch {
        let config = watcher::WatcherConfig {
            debounce: Duration::from_millis(options.watch_debounce_ms),
            ..Default::default()
        };
        match watcher::FileWatcher::new(&project_path, config) {
            Ok(w) => {
                tracing::info!("   File watcher started");
                Some(w)
            }
            Err(e) => {
                tracing::warn!("   Failed to start file watcher: {}", e);
                None
            }
        }
    } else {
        None
    };

    // Idle timeout checker (runs every 60 seconds)
    let server_for_idle = server.clone();
    let idle_timeout = options.idle_timeout;
    let idle_check = async move {
        if let Some(timeout) = idle_timeout {
            let check_interval = Duration::from_secs(60);
            loop {
                tokio::time::sleep(check_interval).await;
                let idle = server_for_idle.idle_secs();
                tracing::debug!(
                    "Idle check: {} secs (timeout: {} secs)",
                    idle,
                    timeout.as_secs()
                );
                if idle >= timeout.as_secs() {
                    tracing::info!("Idle timeout reached ({} secs), shutting down", idle);
                    // Save UUID mappings before shutdown
                    {
                        let api = server_for_idle.api.lock().await;
                        if let Err(e) = api.save_uuid_mappings() {
                            tracing::warn!("Failed to save UUID mappings on idle timeout: {}", e);
                        }
                    }
                    // Trigger shutdown
                    if let Some(tx) = server_for_idle.shutdown_tx.lock().await.take() {
                        let _ = tx.send(());
                    }
                    break;
                }
            }
        } else {
            // No timeout - wait forever (daemon mode)
            tracing::info!("Daemon mode: no idle timeout");
            std::future::pending::<()>().await;
        }
    };

    // File watcher event handler
    let server_for_watch = server.clone();
    let project_path_for_watch = project_path.clone();
    let watch_handler = async move {
        if let Some(ref mut watcher) = file_watcher {
            loop {
                match watcher.recv().await {
                    Some(watcher::WatchEvent::FilesChanged(paths)) => {
                        tracing::info!("Files changed: {:?}", paths.len());
                        if let Err(e) = server_for_watch.reload(&project_path_for_watch).await {
                            tracing::error!("Reload failed: {}", e);
                        }
                    }
                    Some(watcher::WatchEvent::Error(e)) => {
                        tracing::warn!("Watch error: {}", e);
                    }
                    None => {
                        tracing::debug!("Watcher channel closed");
                        break;
                    }
                }
            }
        } else {
            // No watcher, wait forever
            std::future::pending::<()>().await;
        }
    };

    // Accept connections until shutdown signal or idle timeout
    tokio::select! {
        _ = async {
            loop {
                match listener.accept().await {
                    Ok((stream, _)) => {
                        tracing::debug!("Client connected");
                        // Use create_server_transport for proper MessagePack serialization
                        // (compatible with skip_serializing_if attributes)
                        let transport = ryo_app::codec::create_server_transport(stream);

                        let channel = server::BaseChannel::with_defaults(transport);
                        let server_clone = server.clone();
                        tokio::spawn(async move {
                            channel
                                .execute(server_clone.serve())
                                .for_each(|response| async move {
                                    tokio::spawn(response);
                                })
                                .await;
                        });
                    }
                    Err(e) => {
                        tracing::error!("accept error: {}", e);
                    }
                }
            }
        } => {}
        _ = idle_check => {}
        _ = watch_handler => {}
        _ = shutdown_rx => {
            tracing::info!("Shutdown signal received");
        }
    }

    // Cleanup
    let _ = std::fs::remove_file(&socket_path);
    tracing::info!("Server stopped");
    Ok(())
}

// ============================================================================
// Test Harness Module
// ============================================================================

/// Test harness for API integration testing.
///
/// Provides utilities for testing Request/Response serialization and
/// in-process tarpc server/client communication without UDS.
#[cfg(test)]
pub mod test_harness {
    use super::*;
    use ryo_app::api::{
        CascadeRequest, CascadeResponse, DiscoverRequest, DiscoverResponse, RunRequest,
        RunResponse, StatusResponse, SuggestRequest,
    };
    use ryo_app::service::RyoError;
    use ryo_app::{ConflictStrategy, Goal, IdentKind, Intent};

    /// Test helper: Create a minimal Api instance for testing
    pub fn create_test_api() -> Api {
        use ryo_app::InMemoryStorage;
        let storage = Box::new(InMemoryStorage::new());
        Api::new(storage)
    }

    /// Test helper: Create a RyoServer with test Api
    pub fn create_test_server() -> (RyoServer, oneshot::Receiver<()>) {
        let api = create_test_api();
        let (tx, rx) = oneshot::channel();
        (RyoServer::new(api, tx), rx)
    }

    // ========================================================================
    // MessagePack Serialization Tests (used by tarpc transport)
    // ========================================================================

    /// Roundtrip test helper for MessagePack serialization
    fn msgpack_roundtrip<T: serde::Serialize + serde::de::DeserializeOwned + std::fmt::Debug>(
        value: &T,
        type_name: &str,
    ) {
        // Use to_vec_named for map-based serialization (compatible with skip_serializing_if)
        let encoded = rmp_serde::to_vec_named(value)
            .unwrap_or_else(|e| panic!("Failed to serialize {}: {}", type_name, e));
        let _decoded: T = rmp_serde::from_slice(&encoded)
            .unwrap_or_else(|e| panic!("Failed to deserialize {}: {}", type_name, e));
    }

    #[test]
    fn test_msgpack_discover_request() {
        let req = DiscoverRequest {
            pattern: "*Config*".to_string(),
            kind: None,
            sort: None,
            limit: Some(10),
            view: None,
            is_async: None,
            is_unsafe: None,
            scope_path: None,
            ignore_case: false,
            ignore_word_separate: false,
            attr: None,
            is_id: false,
        };
        msgpack_roundtrip(&req, "DiscoverRequest");
    }

    #[test]
    fn test_msgpack_discover_response() {
        let resp = DiscoverResponse {
            status: "found".to_string(),
            symbols: vec![],
            total: 0,
            elapsed_ms: 42,
            hint: None,
        };
        msgpack_roundtrip(&resp, "DiscoverResponse");
    }

    #[test]
    fn test_msgpack_status_response() {
        let resp = StatusResponse {
            project: std::path::PathBuf::from("/test/project"),
            symbols: 100,
            files: 10,
        };
        msgpack_roundtrip(&resp, "StatusResponse");
    }

    #[test]
    fn test_msgpack_suggest_request() {
        let req = SuggestRequest {
            pattern_filter: Some("*Handler*".to_string()),
            high_impact: true,
            quick: false,
            scan: false,
            precheck: false,
            exclude_rules: vec![],
            enhanced: false,
            scope_filter: vec![],
        };
        msgpack_roundtrip(&req, "SuggestRequest");
    }

    #[test]
    fn test_msgpack_cascade_request() {
        let req = CascadeRequest {
            id: "1v1".to_string(),
            uuid: None,
            depth: Some(3),
        };
        msgpack_roundtrip(&req, "CascadeRequest");
    }

    #[test]
    fn test_msgpack_cascade_response() {
        let resp = CascadeResponse {
            display_name: "test_symbol".to_string(),
            callers: vec!["foo".to_string(), "bar".to_string()],
            users: vec!["baz".to_string()],
            match_functions: vec![],
            containing_types: vec![],
        };
        msgpack_roundtrip(&resp, "CascadeResponse");
    }

    #[test]
    fn test_msgpack_ryo_error() {
        let err = RyoError::NotFound {
            name: "test".to_string(),
        };
        msgpack_roundtrip(&err, "RyoError::NotFound");

        let err = RyoError::ParseError {
            message: "syntax error".to_string(),
        };
        msgpack_roundtrip(&err, "RyoError::ParseError");

        let err = RyoError::InvalidRequest {
            message: "bad input".to_string(),
        };
        msgpack_roundtrip(&err, "RyoError::InvalidRequest");

        let err = RyoError::Internal {
            message: "crash".to_string(),
        };
        msgpack_roundtrip(&err, "RyoError::Internal");
    }

    #[test]
    fn test_msgpack_goal() {
        let goal = Goal::new(
            "rename foo to bar",
            Intent::RenameIdent {
                symbol_id: None,
                symbol_path: None,
                target_ident: Some("foo".to_string()),
                to: "bar".to_string(),
                kind: IdentKind::Any,
            },
        );
        msgpack_roundtrip(&goal, "Goal");
    }

    #[test]
    fn test_msgpack_run_request() {
        let goal = Goal::new(
            "rename foo to bar",
            Intent::RenameIdent {
                symbol_id: None,
                symbol_path: None,
                target_ident: Some("foo".to_string()),
                to: "bar".to_string(),
                kind: IdentKind::Any,
            },
        );
        let req = RunRequest {
            goal,
            dry_run: true,
            check_syntax: false,
        };
        msgpack_roundtrip(&req, "RunRequest");
    }

    #[test]
    fn test_msgpack_run_response() {
        let resp = RunResponse {
            success: true,
            files_modified: 2,
            total_changes: 5,
            modified_files: vec![
                std::path::PathBuf::from("/test/file1.rs"),
                std::path::PathBuf::from("/test/file2.rs"),
            ],
            conflicts: vec![],
            syntax_errors: vec![],
            error: None,
        };
        msgpack_roundtrip(&resp, "RunResponse");
    }

    #[test]
    fn test_msgpack_intent_variants() {
        // Test various Intent variants with 3-field specification
        let intents = [
            Intent::RenameIdent {
                symbol_id: None,
                symbol_path: None,
                target_ident: Some("old".to_string()),
                to: "new".to_string(),
                kind: IdentKind::Fn,
            },
            Intent::RenameIdent {
                symbol_id: None,
                symbol_path: None,
                target_ident: Some("old_struct".to_string()),
                to: "new_struct".to_string(),
                kind: IdentKind::Type,
            },
            Intent::RenameIdent {
                symbol_id: None,
                symbol_path: Some("crate::config".to_string()),
                target_ident: None,
                to: "ConfigNew".to_string(),
                kind: IdentKind::Any,
            },
        ];

        for (i, intent) in intents.iter().enumerate() {
            msgpack_roundtrip(intent, &format!("Intent variant {}", i));
        }
    }

    #[test]
    fn test_msgpack_conflict_strategy() {
        let strategies = [
            ConflictStrategy::Fail,
            ConflictStrategy::IntentOrder,
            ConflictStrategy::ParallelOnly,
        ];

        for (i, strategy) in strategies.iter().enumerate() {
            msgpack_roundtrip(strategy, &format!("ConflictStrategy variant {}", i));
        }
    }

    // ========================================================================
    // In-Process tarpc Tests (no UDS)
    // ========================================================================

    #[tokio::test]
    async fn test_harness_status_via_trait() {
        use ryo_app::service::RyoService;

        let (server, _rx) = create_test_server();
        let ctx = tarpc::context::current();
        let status = server.status(ctx).await;

        let _ = status.symbols; // verify status fields are accessible
        let _ = status.files;
    }

    #[tokio::test]
    async fn test_harness_discover_via_trait() {
        use ryo_app::service::RyoService;

        let (server, _rx) = create_test_server();
        let ctx = tarpc::context::current();
        let req = DiscoverRequest {
            pattern: "*".to_string(),
            ..Default::default()
        };
        let result = server.discover(ctx, req).await;

        assert!(result.is_ok());
        let resp = result.unwrap();
        let _ = resp.elapsed_ms; // verify elapsed_ms field is accessible
    }

    #[tokio::test]
    async fn test_harness_cascade_via_trait() {
        use ryo_app::service::RyoService;

        let (server, _rx) = create_test_server();
        let ctx = tarpc::context::current();
        // Use a very large index that definitely doesn't exist
        let req = CascadeRequest {
            id: "9999999v1".to_string(),
            uuid: None,
            depth: Some(2),
        };
        let result = server.cascade(ctx, req).await;

        // Should return NotFound for nonexistent symbol
        assert!(result.is_err());
        match result.unwrap_err() {
            RyoError::NotFound { name } => assert_eq!(name, "'9999999v1'"),
            other => panic!("Expected NotFound, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn test_harness_suggest_via_trait() {
        use ryo_app::service::RyoService;

        let (server, _rx) = create_test_server();
        let ctx = tarpc::context::current();
        let req = SuggestRequest::default();
        let result = server.suggest(ctx, req).await;

        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_harness_ping_via_trait() {
        use ryo_app::service::RyoService;

        let (server, _rx) = create_test_server();
        let ctx = tarpc::context::current();
        // ping returns () so just verify no panic
        server.ping(ctx).await;
    }

    #[tokio::test]
    async fn test_harness_shutdown_via_trait() {
        use ryo_app::service::RyoService;

        let (server, rx) = create_test_server();
        let ctx = tarpc::context::current();

        // Shutdown should send signal through the channel
        server.shutdown(ctx).await;

        // The receiver should be notified
        // Use a timeout to avoid hanging
        let result = tokio::time::timeout(std::time::Duration::from_millis(100), rx).await;

        assert!(result.is_ok(), "Shutdown signal should be received");
    }
}

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

    #[tokio::test]
    async fn test_server_creation() {
        use ryo_app::InMemoryStorage;
        use tokio::sync::oneshot;

        let storage = Box::new(InMemoryStorage::new());
        let api = Api::new(storage);
        let (tx, _rx) = oneshot::channel();
        let _server = RyoServer::new(api, tx);
        // Just verify no panic
    }

    #[test]
    fn test_server_options_default() {
        let opts = ServerOptions::default();
        assert!(opts.parallel_init);
        assert!(opts.idle_timeout.is_some());
    }

    #[tokio::test]
    async fn test_status_returns_non_empty_project_path() {
        use ryo_app::service::RyoService;
        use ryo_app::InMemoryStorage;
        use tokio::sync::oneshot;

        // Api::new uses current directory as project root
        let storage = Box::new(InMemoryStorage::new());
        let api = Api::new(storage);
        let (tx, _rx) = oneshot::channel();
        let server = RyoServer::new(api, tx);

        // Call status and verify project path is set
        let ctx = tarpc::context::current();
        let status = server.status(ctx).await;

        // Project path should not be empty
        assert!(
            !status.project.as_os_str().is_empty(),
            "Project path should not be empty"
        );
        // Project path should exist (current directory)
        assert!(
            status.project.exists(),
            "Project path should exist: {:?}",
            status.project
        );
    }

    #[tokio::test]
    async fn test_ping_updates_last_activity() {
        use ryo_app::service::RyoService;
        use ryo_app::InMemoryStorage;
        use tokio::sync::oneshot;

        let storage = Box::new(InMemoryStorage::new());
        let api = Api::new(storage);
        let (tx, _rx) = oneshot::channel();
        let server = RyoServer::new(api, tx);

        // Record initial idle time
        let _initial_idle = server.idle_secs();

        // Wait a bit
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Ping should update last_activity
        let ctx = tarpc::context::current();
        server.clone().ping(ctx).await;

        // After ping, idle time should be reset (close to 0)
        let after_ping_idle = server.idle_secs();
        assert!(
            after_ping_idle <= 1,
            "Idle time after ping should be ~0, got {}",
            after_ping_idle
        );
    }

    #[tokio::test]
    async fn test_status_updates_last_activity() {
        use ryo_app::service::RyoService;
        use ryo_app::InMemoryStorage;
        use tokio::sync::oneshot;

        let storage = Box::new(InMemoryStorage::new());
        let api = Api::new(storage);
        let (tx, _rx) = oneshot::channel();
        let server = RyoServer::new(api, tx);

        // Wait a bit
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // Status should update last_activity
        let ctx = tarpc::context::current();
        let _ = server.clone().status(ctx).await;

        // After status, idle time should be reset (close to 0)
        let after_status_idle = server.idle_secs();
        assert!(
            after_status_idle <= 1,
            "Idle time after status should be ~0, got {}",
            after_status_idle
        );
    }
}