//! `houndd` — the Hound Antivirus daemon.
//!
//! A single-purpose service that exposes a pluggable scanning engine plus a
//! set of detection subsystems over a Unix-socket, line-delimited JSON-RPC
//! 2.0 API. Both the CLI and the GUI are thin clients of this socket, which
//! is what lets future suite tools (firewall, updater, …) share the same
//! engine without forking it.
//!
//! ## The subsystems
//!
//!   - **engine**    — signature-scan engine, pluggable behind
//!                     [`engine::ScanEngine`] (ClamAV today, native Rust
//!                     engine tomorrow). Wire API only reports `engine` name.
//!   - **realtime**  — inotify monitor: watches configured dirs, scans each
//!                     new/modified file, quarantines or alerts, and feeds a
//!                     rolling write-window into the **ransomware** heuristic.
//!   - **quarantine** — the vault: files that got caught are moved here.
//!   - **rootkit**   — userspace rootkit heuristics (hidden pids, hidden
//!                     files, writable system dirs, setuid anomalies).
//!   - **events**    — the alert log every subsystem writes into.
//!   - **settings**  — the user-tunable knobs, persisted to XDG config.
//!
//! ## RPC methods
//!   status            → engine health + realtime + quarantine counts
//!   scan              → recursive scan of a path, per-file findings
//!   update            → refresh the signature store
//!   settings.get      → current knobs
//!   settings.set      → replace the knobs (live-reload realtime)
//!   events.list       → the alert log (newest first)
//!   events.clear      → drop all alerts
//!   quarantine.list   → files held in quarantine
//!   quarantine.add    → move a file into the vault
//!   quarantine.restore→ put a file back
//!   quarantine.remove → delete a quarantined file
//!   rootkit.scan      → run the rootkit heuristics
//!   realtime.status   → what's watched, counters, ransomware state
//!   realtime.set_enabled → toggle the monitor
//!
//! The engine is selectable at startup for tests: `HOUNDD_ENGINE=fake`
//! swaps in [`engine::engine`]'s fake backend (a full daemon lifecycle
//! without ClamAV installed).

mod cache;
mod caps;
mod defs;
mod engine;
mod events;
mod fanotify;
mod native;
mod persistence;
mod quarantine;
mod realtime;
mod rootkit;
mod rules;
mod settings;
#[cfg(test)]
mod test_util;

use anyhow::{bail, Context, Result};
use hound_api::{Response, Settings};
use serde_json::Value;
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::PathBuf;

const DAEMON_VERSION: &str = env!("CARGO_PKG_VERSION");

/// Shared, daemon-lifetime state. Cheap to clone (everything is `Arc`-shared),
/// so the accept loop hands a copy to each connection thread and the realtime
/// monitor holds its own copy.
#[derive(Clone)]
struct DaemonState {
    settings: settings::SettingsStore,
    events: events::EventLog,
    quarantine: quarantine::Quarantine,
    realtime: realtime::RealtimeMonitor,
    defs: defs::DefsStore,
    /// The execution gate, when it came up. `None` covers both "switched
    /// off" and "could not be armed"; `gate_detail` says which.
    gate: Option<std::sync::Arc<fanotify::Gate>>,
    gate_detail: std::sync::Arc<String>,
    gate_paths: Vec<String>,
}

fn main() -> Result<()> {
    let sock = hound_api::default_socket_path();
    let sock_path = PathBuf::from(&sock);
    if let Some(parent) = sock_path.parent() {
        fs::create_dir_all(parent)
            .with_context(|| format!("creating socket dir {}", parent.display()))?;
    }
    // Reconnect-friendly startup: drop a stale socket from a dead daemon.
    let _ = fs::remove_file(&sock_path);
    let listener = UnixListener::bind(&sock_path).with_context(|| format!("binding {sock}"))?;

    let state = DaemonState::boot();

    eprintln!(
        "houndd {DAEMON_VERSION} listening on {sock} [engine: {}] (Ctrl-C to stop)",
        engine::engine().name()
    );

    for stream in listener.incoming() {
        let stream = match stream {
            Ok(s) => s,
            Err(e) => {
                eprintln!("accept error: {e}");
                continue;
            }
        };
        let state = state.clone();
        std::thread::spawn(move || {
            if let Err(e) = handle_conn(stream, state) {
                eprintln!("connection error: {e}");
            }
        });
    }
    Ok(())
}

impl DaemonState {
    /// Build the shared state and start the realtime monitor.
    ///
    /// The monitor is started lazily: `start()` is idempotent and reads the
    /// watch dirs from settings at the moment it's called. We start it here
    /// so the daemon is protective from boot; a `settings.set` that changes
    /// the watch list calls `start()` again (a no-op if already running —
    /// the thread picks up new dirs via its own settings reads).
    fn boot() -> Self {
        let settings = settings::SettingsStore::load();
        let events = events::EventLog::new();
        let quarantine = quarantine::Quarantine::new();
        let defs = defs::DefsStore::load();
        {
            let d = defs.current();
            if d.indicators > 0 {
                eprintln!(
                    "defs: {} indicators from {} pack(s) [{}]",
                    d.indicators,
                    d.packs.len(),
                    d.version
                );
            } else if !d.detail.is_empty() {
                eprintln!("defs: {}", d.detail);
            }
        }
        let realtime =
            realtime::RealtimeMonitor::new(settings.clone(), quarantine.clone(), events.clone());

        let s = settings.get();
        let (gate, gate_detail, gate_paths) = Self::arm_gate(&s, &events, &quarantine);

        // The inotify monitor is now the FALLBACK, not the primary path.
        //
        // When the gate is armed it already sees every completed write on
        // the whole filesystem, with no watch-descriptor ceiling and no
        // blind spots outside the configured directories. Running both
        // would scan everything twice and quarantine the same file from
        // two threads. inotify survives only for the unprivileged case,
        // where fanotify is not available at all.
        if gate.is_some() {
            events.push(
                "realtime",
                "info",
                "inotify monitor idle — the execution gate covers writes".into(),
            );
        } else if s.realtime_enabled && !s.paused {
            match realtime.start() {
                Ok(()) => events.push("realtime", "info", "real-time monitor started".into()),
                Err(e) => events.push("realtime", "warn", format!("real-time monitor idle: {e}")),
            };
        }

        events.push("info", "info", "houndd daemon started".into());

        Self {
            settings,
            events,
            quarantine,
            realtime,
            defs,
            gate,
            gate_detail: std::sync::Arc::new(gate_detail),
            gate_paths,
        }
    }

    /// Bring up the execution gate, if it is switched on and we can.
    ///
    /// Ordering is deliberate: `fanotify_init` needs CAP_SYS_ADMIN, so the
    /// capability reduction happens *after* the group is open. Every
    /// failure here degrades to a working daemon without a gate — never
    /// to a daemon that will not start. An antivirus that refuses to run
    /// protects nothing.
    fn arm_gate(
        s: &Settings,
        events: &events::EventLog,
        quarantine: &quarantine::Quarantine,
    ) -> (Option<std::sync::Arc<fanotify::Gate>>, String, Vec<String>) {
        if !s.exec_gate {
            eprintln!("gate: disabled in settings");
            return (None, "disabled".into(), Vec::new());
        }
        if !caps::is_root() {
            let why = "needs root (CAP_SYS_ADMIN)";
            events.push("gate", "warn", format!("execution gate off: {why}"));
            return (None, why.into(), Vec::new());
        }

        let gate = match fanotify::Gate::init() {
            Ok(g) => std::sync::Arc::new(g),
            Err(e) => {
                events.push("gate", "warn", format!("execution gate off: {e}"));
                return (None, e.to_string(), Vec::new());
            }
        };

        // Empty means the whole root filesystem, which is the production
        // shape. Anything listed is treated as a mount to cover.
        let paths: Vec<String> = if s.exec_gate_paths.is_empty() {
            vec!["/".to_string()]
        } else {
            s.exec_gate_paths.clone()
        };
        // ALWAYS mark the filesystem, never the mount — even when the
        // operator scoped the gate to one path.
        //
        // systemd gives this service a private mount namespace (any of
        // ProtectProc, ProtectKernelTunables or ProtectControlGroups is
        // enough to force one). FAN_MARK_MOUNT marks a vfsmount, and a
        // private namespace holds its own vfsmount for the same
        // filesystem — so the daemon marks its copy, every other process
        // on the machine uses the host's copy, and not one event is ever
        // delivered. The gate reports itself armed and silently protects
        // nothing, which is the worst way for a security feature to fail.
        //
        // FAN_MARK_FILESYSTEM marks the SUPERBLOCK, which is shared
        // across namespaces. Scoping still works, because a superblock is
        // exactly one filesystem: marking a dedicated mount covers that
        // mount and nothing else.
        let mut marked = Vec::new();
        for p in &paths {
            let path = std::path::Path::new(p);
            match gate.mark_filesystem(path) {
                Ok(()) => marked.push(p.clone()),
                Err(e) => {
                    events.push("gate", "warn", format!("could not watch {p}: {e}"));
                    eprintln!("gate: could not watch {p}: {e}");
                }
            }
        }
        if marked.is_empty() {
            return (None, "no mount could be watched".into(), Vec::new());
        }

        // Give up everything we do not need, and do it HERE — before a
        // single thread exists.
        //
        // Capabilities are per-thread. Dropping them after spawning the
        // reader and workers would reduce only this thread and leave the
        // workers holding full root, which is the opposite of the point.
        // Threads created after this inherit the reduced set. fanotify_init
        // and the marks are already done, and CAP_SYS_ADMIN is retained so
        // a settings change can still add one later.
        match caps::drop_to_gate_minimum() {
            Ok(true) => {
                events.push(
                    "gate",
                    "info",
                    "capabilities reduced to the four the gate needs".into(),
                );
                eprintln!("gate: capabilities reduced to 4 of 41 (CapEff 0x20000e)");
            }
            Ok(false) => {}
            Err(e) => {
                events.push("gate", "warn", format!("could not reduce capabilities: {e}"));
                eprintln!("gate: could not reduce capabilities: {e}");
            }
        }

        // The watchdog before the workers, always: it is what guarantees
        // no process is held past the deadline, including during startup.
        gate.start_watchdog();

        let mut excludes = s.exclude_paths.clone();
        // Never gate our own state. The vault holds live malware by
        // definition, and holding a process hostage over our own database
        // is a way to deadlock the daemon against itself.
        for own in ["/var/lib/hound", "/run/hound"] {
            if !excludes.iter().any(|e| e == own) {
                excludes.push(own.to_string());
            }
        }
        // The gate holds a process while it decides, so its budget is the
        // deadline, not the on-demand scan limit.
        let max_size = s
            .max_file_size_mb
            .saturating_mul(1024 * 1024)
            .min(fanotify::GATE_MAX_FILE_BYTES);
        let ev = events.clone();
        let quarantine_on_write = s.on_detect == "quarantine";
        let q = quarantine.clone();
        gate.serve(
            fanotify::GateConfig {
                workers: 4,
                max_size,
                excludes,
            },
            std::sync::Arc::new(|_path: &std::path::Path, bytes: &[u8]| {
                engine::engine().scan_bytes(bytes)
            }),
                        std::sync::Arc::new(
                move |path: &std::path::Path, name: &str, verdict: fanotify::Verdict| {
                    match verdict {
                        fanotify::Verdict::Blocked => {
                            ev.push(
                                "gate",
                                "critical",
                                format!("blocked execution of {} ({name})", path.display()),
                            );
                        }
                        fanotify::Verdict::Seen => {
                        // Only reachable if read events are ever requested
                        // again. Reported, never blocked.
                        ev.push(
                            "gate",
                            "warn",
                            format!("{} matched {name} while being read", path.display()),
                        );
                    }
                    fanotify::Verdict::Written => {
                            // Nothing was waiting on this one, so the file is
                            // already on disk. This is the path that replaces
                            // what inotify used to do, with whole-filesystem
                            // coverage and no watch-descriptor ceiling.
                            // Two gates before anything is moved: the
                            // operator's policy, and the RULE's own
                            // declaration that it is anchored enough to
                            // justify destroying a file. A content-only
                            // rule reports and leaves the file alone,
                            // however confident it looks — text matches
                            // turn up inside logs, transcripts, build
                            // caches and documentation about the very
                            // thing being detected.
                            if !quarantine_on_write
                                || !engine::engine().may_quarantine(name)
                            {
                                ev.push(
                                    "gate",
                                    "critical",
                                    format!(
                                        "threat detected in {} ({name}) — reported, not moved",
                                        path.display()
                                    ),
                                );
                                return;
                            }
                            match q.add(&path.to_string_lossy(), name) {
                                Ok(entry) => ev.push(
                                    "quarantine",
                                    "critical",
                                    format!(
                                        "quarantined {} ({name}) as {}",
                                        path.display(),
                                        entry.id
                                    ),
                                ),
                                Err(e) => ev.push(
                                    "quarantine",
                                    "warn",
                                    format!("could not quarantine {}: {e}", path.display()),
                                ),
                            };
                        }
                    }
                },
            ),
            // Fast path: fstat the descriptor the kernel already gave us
            // and ask the engine whether it has judged this exact file
            // version before. No read, no scan, no worker.
            std::sync::Arc::new(|event: &fanotify::Event| {
                let md = event.metadata()?;
                engine::engine()
                    .cached_verdict(&md)
                    .map(|verdict| verdict.is_none())
            }),
        );

        events.push(
            "gate",
            "info",
            format!("execution gate armed on {}", marked.join(", ")),
        );
        eprintln!("gate: armed on {} (4 workers, watchdog live)", marked.join(", "));
        (Some(gate), String::new(), marked)
    }
}

/// Read one request line, dispatch, write one response line.
fn handle_conn(stream: UnixStream, state: DaemonState) -> Result<()> {
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line)?;

    let req: hound_api::Request = serde_json::from_str(line.trim()).context("decoding request")?;

    let result = dispatch(&req, &state);
    let resp = match result {
        Ok(value) => Response {
            jsonrpc: "2.0".into(),
            id: req.id,
            result: Some(value),
            error: None,
        },
        Err(e) => Response {
            jsonrpc: "2.0".into(),
            id: req.id,
            result: None,
            error: Some(hound_api::ErrorObject {
                code: -32000,
                message: e.to_string(),
                data: None,
            }),
        },
    };
    let mut out = serde_json::to_string(&resp)?;
    out.push('\n');
    writer_flush(&mut reader, &out)?;
    Ok(())
}

fn writer_flush(reader: &mut BufReader<UnixStream>, bytes: &str) -> Result<()> {
    // The BufReader consumed the stream; get the stream back out to write.
    let stream = reader.get_mut();
    stream.write_all(bytes.as_bytes())?;
    stream.flush()?;
    Ok(())
}

fn dispatch(req: &hound_api::Request, st: &DaemonState) -> Result<Value> {
    match req.method.as_str() {
        "status" => Ok(serde_json::to_value(status(st)?)?),
        "update" => Ok(serde_json::to_value(update(st)?)?),
        "scan" => {
            let path = req
                .params
                .as_ref()
                .and_then(|p| p.get("path"))
                .and_then(Value::as_str)
                .context("scan requires params.path")?;
            let recursive = req
                .params
                .as_ref()
                .and_then(|p| p.get("recursive"))
                .and_then(Value::as_bool)
                .unwrap_or(true);
            Ok(serde_json::to_value(scan(path, recursive)?)?)
        }

        // ── settings ──
        "settings.get" => Ok(serde_json::to_value(st.settings.get())?),
        "settings.set" => {
            let mut incoming: Settings = serde_json::from_value(
                req.params
                    .clone()
                    .context("settings.set requires a params object")?,
            )?;
            // Clients are not trusted to send a theme we can render.
            incoming.normalise_appearance();
            st.settings
                .set(&incoming)
                .map_err(|e| anyhow::anyhow!("persisting settings: {e}"))?;

            // Live-reload the realtime monitor: if it's not running and the
            // new settings ask for it, start it; if the watch list changed,
            // the thread's per-event settings reads pick it up.
            let cur = st.settings.get();
            if cur.realtime_enabled && !cur.paused && !st.realtime.is_running() {
                let _ = st.realtime.start();
            }
            Ok(serde_json::to_value(cur)?)
        }

        // ── events / alerts ──
        "events.list" => {
            let limit = req
                .params
                .as_ref()
                .and_then(|p| p.get("limit"))
                .and_then(Value::as_u64)
                .unwrap_or(100) as u32;
            Ok(serde_json::to_value(st.events.list(limit))?)
        }
        "events.clear" => Ok(serde_json::to_value(st.events.clear())?),

        // ── quarantine ──
        "quarantine.list" => Ok(serde_json::to_value(st.quarantine.list())?),
        "quarantine.add" => {
            let path = req
                .params
                .as_ref()
                .and_then(|p| p.get("path"))
                .and_then(Value::as_str)
                .context("quarantine.add requires params.path")?;
            let virus = req
                .params
                .as_ref()
                .and_then(|p| p.get("virus"))
                .and_then(Value::as_str)
                .unwrap_or("manual");
            let entry = realtime::quarantine_and_log(&st.quarantine, &st.events, path, virus)
                .with_context(|| format!("quarantining {path}"))?;
            Ok(serde_json::to_value(entry)?)
        }
        "quarantine.restore" => {
            let id = req
                .params
                .as_ref()
                .and_then(|p| p.get("id"))
                .and_then(Value::as_str)
                .context("quarantine.restore requires params.id")?;
            let entry = st
                .quarantine
                .restore(id)
                .with_context(|| format!("restoring {id}"))?;
            st.events.push(
                "restore",
                "info",
                format!("restored {} from quarantine", entry.original_path),
            );
            Ok(serde_json::to_value(entry)?)
        }
        "quarantine.remove" => {
            let id = req
                .params
                .as_ref()
                .and_then(|p| p.get("id"))
                .and_then(Value::as_str)
                .context("quarantine.remove requires params.id")?;
            let freed = st.quarantine.remove(id)?;
            st.events.push(
                "quarantine",
                "info",
                format!("deleted quarantined file {id} ({freed} bytes)"),
            );
            Ok(serde_json::to_value(freed)?)
        }

        // ── rootkit ──
        "rootkit.scan" => {
            let dirs = st.settings.get().realtime_watch.clone();
            let scan = rootkit::run_scan(&dirs);
            let sev = if scan.critical > 0 {
                "critical"
            } else if scan.warn > 0 {
                "warn"
            } else {
                "info"
            };
            st.events
                .push("rootkit", sev, format!("rootkit scan: {}", scan.verdict));
            Ok(serde_json::to_value(scan)?)
        }

        // ── persistence ledger ──
        "persistence.scan" => {
            // Writing the baseline is an explicit act. A plain check must
            // not quietly record whatever is currently installed as normal
            // — that is how a compromise becomes the new baseline.
            let update = req
                .params
                .as_ref()
                .and_then(|p| p.get("update_baseline"))
                .and_then(Value::as_bool)
                .unwrap_or(false);
            let report = persistence::scan(update);
            let warns = report
                .changes
                .iter()
                .filter(|c| c.severity == "warn" || c.severity == "critical")
                .count();
            if !report.first_run && warns > 0 {
                st.events.push(
                    "persistence",
                    "warn",
                    format!(
                        "{warns} unexplained change(s) to startup configuration across {} item(s)",
                        report.total
                    ),
                );
            }
            Ok(serde_json::to_value(report)?)
        }

        // ── supply chain ──
        "supply.sweep" => {
            let path = req
                .params
                .as_ref()
                .and_then(|p| p.get("path"))
                .and_then(Value::as_str)
                .context("supply.sweep requires params.path")?;
            let root = std::fs::canonicalize(path)
                .with_context(|| format!("no such path: {path}"))?;
            let loaded = st.defs.current();
            let report = hound_supply::sweep::sweep_with(&root, Some(&loaded.index));

            let critical = report.count(hound_supply::Severity::Critical);
            let warnings = report.count(hound_supply::Severity::Warning);
            let sev = if critical > 0 {
                "critical"
            } else if warnings > 0 {
                "warn"
            } else {
                "info"
            };
            st.events.push(
                "supply",
                sev,
                format!(
                    "supply-chain sweep of {}: {critical} critical, {warnings} warning(s) across {} files",
                    root.display(),
                    report.examined
                ),
            );
            Ok(serde_json::to_value(report)?)
        }

        // ── realtime ──
        "realtime.status" => Ok(serde_json::to_value(st.realtime.status())?),
        "realtime.set_enabled" => {
            let enabled = req
                .params
                .as_ref()
                .and_then(|p| p.get("enabled"))
                .and_then(Value::as_bool)
                .context("realtime.set_enabled requires params.enabled")?;
            let mut s = st.settings.get();
            s.realtime_enabled = enabled;
            let updated = st.settings.mutate(|s2| *s2 = s);
            if enabled && !updated.paused {
                let _ = st.realtime.start();
            } else if !enabled {
                st.realtime.stop();
            }
            Ok(serde_json::to_value(st.realtime.status())?)
        }

        other => bail!("unknown method {other:?}"),
    }
}

// ── RPC handlers (engine-agnostic) ──────────────────────────────────────────

fn status(st: &DaemonState) -> Result<hound_api::Status> {
    let (present, db_summary, db) = engine::engine().probe();
    let os = std::fs::read_to_string("/etc/os-release")
        .ok()
        .and_then(|c| {
            c.lines().find(|l| l.starts_with("PRETTY_NAME=")).map(|l| {
                l.trim_start_matches("PRETTY_NAME=")
                    .trim_matches('"')
                    .to_string()
            })
        })
        .unwrap_or_else(|| "unknown".into());
    Ok(hound_api::Status {
        daemon_version: DAEMON_VERSION.to_string(),
        engine: engine::engine().name().to_string(),
        engine_present: present,
        db_summary,
        os,
        db,
        realtime: st.realtime.status(),
        quarantined: st.quarantine.count(),
        gate: gate_status(st),
    })
}

/// Snapshot the execution gate for the wire.
fn gate_status(st: &DaemonState) -> hound_api::GateStatus {
    match &st.gate {
        Some(gate) => {
            let (allowed, denied, timed_out) = gate.responder().counters();
            hound_api::GateStatus {
                active: true,
                detail: String::new(),
                paths: st.gate_paths.clone(),
                allowed,
                denied,
                timed_out,
            }
        }
        None => hound_api::GateStatus {
            active: false,
            detail: st.gate_detail.as_ref().clone(),
            ..Default::default()
        },
    }
}

/// Refresh the signature store via the engine, then re-probe so the
/// client can refresh its UI/tray from a single round-trip.
fn update(st: &DaemonState) -> Result<hound_api::UpdateResult> {
    use hound_api::UpdateResult;

    let (ok, command, combined) = engine::engine().update()?;
    st.events.push(
        "update",
        if ok { "info" } else { "warn" },
        format!(
            "signature update via {command}: {}",
            if ok { "ok" } else { "failed" }
        ),
    );
    Ok(UpdateResult {
        ok,
        command,
        output: cap_tail(&combined, 2048),
        status: status(st)?,
    })
}

/// Keep the tail of `s` within `max` bytes, back-stepping to a char
/// boundary and prefixing an ellipsis marker.
fn cap_tail(s: &str, max: usize) -> String {
    if s.len() <= max {
        return s.to_string();
    }
    let cut = s.len() - max;
    let cut = s
        .char_indices()
        .take_while(|(i, _)| *i < cut)
        .last()
        .map(|(i, c)| i + c.len_utf8())
        .unwrap_or(cut);
    format!("…\n{}", &s[cut..])
}

fn scan(path: &str, recursive: bool) -> Result<hound_api::ScanResult> {
    engine::engine().scan(path, recursive)
}

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

    #[test]
    fn client_type_is_constructible() {
        let _c: hound_api::Client = hound_api::Client::new("/tmp/does-not-matter.sock".into());
    }

    #[test]
    fn cap_tail_keeps_char_boundaries() {
        let s = "é".repeat(3000);
        let t = cap_tail(&s, 100);
        assert!(t.len() <= 100 + 4); // "…\n" prefix is 4 bytes
        assert!(t.starts_with('…'));
    }

    /// Full daemon lifecycle with the fake engine: boot, query status,
    /// round-trip settings, quarantine a file, run the rootkit scan, and
    /// confirm the realtime status shape — all over a real Unix socket.
    #[test]
    fn e2e_fake_engine_over_socket() {
        use serde_json::json;
        use std::io::{BufRead, Write as _};
        use std::os::unix::net::UnixStream;

        let dir = std::env::temp_dir().join(format!("hound-e2e-{}", std::process::id()));
        let _ = std::fs::create_dir_all(&dir);
        let sock = dir.join("houndd.sock");
        let cfg = dir.join("cfg");
        let data = dir.join("data");

        // Serialize against any other test that mutates process env.
        let _env_guard = crate::test_util::locked();

        // Isolate every path the daemon touches.
        std::env::set_var("HOUNDD_ENGINE", "fake");
        std::env::set_var("HOUNDD_SOCK", &sock);
        std::env::set_var("XDG_CONFIG_HOME", &cfg);
        std::env::set_var("XDG_DATA_HOME", &data);

        // Boot the listener on a thread (the monitor also starts from here).
        let listener = UnixListener::bind(&sock).unwrap();
        let state = DaemonState::boot();
        let shutdown_state = state.clone();
        let handle = std::thread::spawn(move || {
            for stream in listener.incoming() {
                if let Ok(s) = stream {
                    let state = state.clone();
                    std::thread::spawn(move || {
                        let _ = handle_conn(s, state);
                    });
                }
            }
        });

        // status — engine present, realtime active, quarantine 0.
        {
            let mut c = std::io::BufReader::new(UnixStream::connect(&sock).unwrap());
            let req = json!({"jsonrpc":"2.0","id":1,"method":"status"});
            c.get_mut()
                .write_all((serde_json::to_string(&req).unwrap() + "\n").as_bytes())
                .unwrap();
            let mut line = String::new();
            c.read_line(&mut line).unwrap();
            let v: Value = serde_json::from_str(line.trim()).unwrap();
            let r = v["result"].clone();
            assert_eq!(r["engine"].as_str().unwrap(), "fake");
            assert!(r["engine_present"].as_bool().unwrap());
            assert!(r["realtime"].is_object());
            assert_eq!(r["quarantined"].as_u64().unwrap(), 0);
        }

        // settings.get then settings.set (round-trip a changed knob).
        {
            let client = hound_api::Client::new(sock.to_str().unwrap().to_string());
            let mut s = client.settings().unwrap();
            assert!(s.recursive_default);
            s.max_file_size_mb = 42;
            let back = client.set_settings(&s).unwrap();
            assert_eq!(back.max_file_size_mb, 42);
            assert_eq!(client.settings().unwrap().max_file_size_mb, 42);
        }

        // quarantine.add + list + restore.
        {
            let client = hound_api::Client::new(sock.to_str().unwrap().to_string());
            let f = dir.join("victim.bin");
            std::fs::write(&f, b"evil").unwrap();
            let entry = client
                .quarantine_add(f.to_str().unwrap(), "EICAR-Test")
                .unwrap();
            assert!(!entry.restored);
            assert!(!f.exists());
            assert_eq!(client.quarantine_list().unwrap().len(), 1);
            let restored = client.quarantine_restore(&entry.id).unwrap();
            assert!(restored.restored);
            assert!(f.exists());
        }

        // rootkit.scan returns a well-formed shape.
        {
            let client = hound_api::Client::new(sock.to_str().unwrap().to_string());
            let scan = client.rootkit_scan().unwrap();
            assert!(scan.verdict.contains("clean") || scan.critical + scan.warn + scan.info > 0);
            let total = scan.findings.len() as u32;
            assert_eq!(scan.critical + scan.warn + scan.info, total);
        }

        // events reflect what happened (at least the boot + scan events).
        {
            let client = hound_api::Client::new(sock.to_str().unwrap().to_string());
            let evts = client.events(100).unwrap();
            assert!(!evts.is_empty());
            // Most-recent-first: the rootkit scan is one of the recent events.
            assert!(evts.iter().any(|e| e.kind == "rootkit"));
        }

        // realtime.status shape is intact.
        {
            let client = hound_api::Client::new(sock.to_str().unwrap().to_string());
            let rt = client.realtime_status().unwrap();
            assert!(rt.uptime_secs < 60);
        }

        // Clean shutdown: stop the monitor (drops the inotify loop) then
        // drop the listener so the accept thread can exit.
        shutdown_state.realtime.stop();
        drop(shutdown_state);
        // The listener thread blocks on incoming(); sending a signal isn't
        // portable, so just detach it (daemon is in a test process).
        let _ = handle;

        std::env::remove_var("HOUNDD_ENGINE");
        std::env::remove_var("HOUNDD_SOCK");
        std::env::remove_var("XDG_CONFIG_HOME");
        std::env::remove_var("XDG_DATA_HOME");
        let _ = std::fs::remove_dir_all(&dir);
    }
}