openlatch-client 0.1.18

//! Config monitor main loop and hash pipeline.
//!
//! Hashes content, classifies severity, builds CloudEvents, and forwards
//! through the existing `cloud_tx` rail. The pipeline is fail-open: any
//! error (parse / canonicalize / forward) logs a warning under the
//! appropriate `OL-2xxx` code and the daemon continues.

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

use serde_json::Value;
use sha2::{Digest, Sha256};
use tokio::sync::mpsc;
use unicode_normalization::UnicodeNormalization;

use super::cache::{CacheEntry, ContentHashCache};
use super::manifest::{expand_path, AgentPath, ConfigScope, Manifest, WatchStrategy};
use super::watcher::{glob_expand, is_excluded};

/// Already-registered project roots — keyed on `path_compat::dedup_key` of the
/// canonical path, so one project reported in two casings on Windows stays one
/// entry. Used to short-circuit repeated `ProjectScopeRegister` requests for
/// the same project (each new Claude Code session fires another SessionStart).
type RegisteredProjects = HashSet<PathBuf>;
use crate::cloud::CloudEvent;
use crate::config::{Config, ContentForwardMode};
use crate::core::logging::EventLogger;
use crate::privacy::{filter_event_with, PrivacyFilter};

/// Inbound requests to the monitor loop. The watcher pumps `Fs*` events, and
/// timers / admin endpoints pump rescan + project-scope variants.
#[derive(Debug, Clone)]
pub enum ConfigChangeRequest {
    FsAdded(Vec<PathBuf>),
    FsModified(Vec<PathBuf>),
    FsRemoved(Vec<PathBuf>),
    /// Initial inventory walk — emits snapshots with `diffcounter == 0`.
    InitialInventory,
    /// Periodic 12 h rescan or `drain_notify` recovery rescan — emits
    /// snapshots with `diffcounter > 0`.
    PeriodicRescan,
    /// CLI-driven rescan (`openlatch inventory rescan`).
    ManualRescan {
        path_filter: Option<PathBuf>,
    },
    /// SessionStart cwd discovered a project root (P2 wires the trigger).
    ProjectScopeRegister {
        project_root: PathBuf,
    },
    Shutdown,
}

/// Severity bucket attached to outbound events as the `severityhint`
/// extension attribute. The classifier is hardcoded — manifest does NOT
/// declare severity.
#[derive(Debug, Clone, Copy)]
pub enum Severity {
    Critical,
    High,
    Medium,
    Low,
    Info,
}

impl Severity {
    pub fn as_str(&self) -> &'static str {
        match self {
            Severity::Critical => "critical",
            Severity::High => "high",
            Severity::Medium => "medium",
            Severity::Low => "low",
            Severity::Info => "info",
        }
    }
}

/// Source of an outbound event, surfaced as the `eventsource` extension
/// attribute. Cloud uses this to attribute backpressure, dedup on the
/// (source, content_hash) pair, and distinguish initial inventory from
/// a real change.
#[derive(Debug, Clone, Copy)]
pub enum EventSource {
    InitScan,
    Rescan,
    FsWatcher,
}

impl EventSource {
    pub fn as_str(&self) -> &'static str {
        match self {
            EventSource::InitScan => "init_scan",
            EventSource::Rescan => "rescan",
            EventSource::FsWatcher => "fs_watcher",
        }
    }
}

/// Concrete change category used in CloudEvent type strings + severity
/// classification.
#[derive(Debug, Clone, Copy)]
pub enum ChangeKind {
    Snapshot,
    Added,
    Modified,
    Removed,
}

impl ChangeKind {
    fn type_suffix(self) -> &'static str {
        match self {
            ChangeKind::Snapshot => "snapshot",
            ChangeKind::Added => "added",
            ChangeKind::Modified => "modified",
            ChangeKind::Removed => "removed",
        }
    }

    fn severity_key(self) -> &'static str {
        match self {
            ChangeKind::Snapshot | ChangeKind::Added => "added",
            ChangeKind::Modified => "modified",
            ChangeKind::Removed => "removed",
        }
    }
}

/// Run the monitor loop until the request channel closes (daemon shutdown
/// drops the sender) or a `Shutdown` request arrives.
#[allow(clippy::too_many_arguments)]
pub async fn run(
    manifest: Arc<Manifest>,
    cache: Arc<ContentHashCache>,
    privacy_filter: PrivacyFilter,
    cloud_tx: Option<mpsc::Sender<CloudEvent>>,
    event_logger: EventLogger,
    config: Arc<Config>,
    mut request_rx: mpsc::Receiver<ConfigChangeRequest>,
    request_tx: mpsc::Sender<ConfigChangeRequest>,
) {
    let mut diff_counters: HashMap<(String, String), u64> = HashMap::new();
    let mut registered_projects: RegisteredProjects = HashSet::new();

    // Kick off initial inventory in the background so the monitor loop is
    // ready to drain other requests immediately.
    let init_tx = request_tx.clone();
    tokio::spawn(async move {
        let _ = init_tx.send(ConfigChangeRequest::InitialInventory).await;
    });

    // Periodic rescan (every `periodic_rescan_interval_hours` hours).
    let rescan_tx = request_tx.clone();
    let rescan_interval = std::time::Duration::from_secs(
        config
            .inventory_monitor
            .periodic_rescan_interval_hours
            .saturating_mul(3600),
    );
    if !rescan_interval.is_zero() {
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(rescan_interval);
            interval.tick().await;
            loop {
                interval.tick().await;
                if rescan_tx
                    .send(ConfigChangeRequest::PeriodicRescan)
                    .await
                    .is_err()
                {
                    break;
                }
            }
        });
    }

    while let Some(req) = request_rx.recv().await {
        match req {
            ConfigChangeRequest::Shutdown => break,
            ConfigChangeRequest::InitialInventory => {
                let started = Instant::now();
                crate::telemetry::capture_global(
                    crate::telemetry::Event::config_initial_scan_started("claude-code"),
                );
                let emitted = run_full_walk(
                    &manifest,
                    &cache,
                    &privacy_filter,
                    cloud_tx.as_ref(),
                    &event_logger,
                    &config,
                    &mut diff_counters,
                    EventSource::InitScan,
                    None,
                )
                .await;
                let duration_ms = started.elapsed().as_millis() as u64;
                crate::telemetry::capture_global(
                    crate::telemetry::Event::config_initial_scan_completed(
                        "claude-code",
                        emitted,
                        duration_ms,
                    ),
                );
                tracing::info!(
                    items_emitted = emitted,
                    duration_ms,
                    "config_monitor: initial inventory walk complete"
                );
            }
            ConfigChangeRequest::PeriodicRescan => {
                let started = Instant::now();
                let (observed, changed) = run_rescan(
                    &manifest,
                    &cache,
                    &privacy_filter,
                    cloud_tx.as_ref(),
                    &event_logger,
                    &config,
                    &mut diff_counters,
                    None,
                )
                .await;
                let duration_ms = started.elapsed().as_millis() as u64;
                crate::telemetry::capture_global(
                    crate::telemetry::Event::config_periodic_rescan_completed(
                        "claude-code",
                        observed,
                        changed,
                        duration_ms,
                    ),
                );
                tracing::info!(
                    items_observed = observed,
                    items_changed = changed,
                    duration_ms,
                    "config_monitor: periodic rescan complete"
                );
            }
            ConfigChangeRequest::ManualRescan { path_filter } => {
                let started = Instant::now();
                let (observed, changed) = run_rescan(
                    &manifest,
                    &cache,
                    &privacy_filter,
                    cloud_tx.as_ref(),
                    &event_logger,
                    &config,
                    &mut diff_counters,
                    path_filter.as_deref(),
                )
                .await;
                let duration_ms = started.elapsed().as_millis() as u64;
                crate::telemetry::capture_global(
                    crate::telemetry::Event::config_periodic_rescan_completed(
                        "claude-code",
                        observed,
                        changed,
                        duration_ms,
                    ),
                );
            }
            ConfigChangeRequest::FsAdded(paths) => {
                for path in paths {
                    handle_fs_event(
                        ChangeKind::Added,
                        &path,
                        &manifest,
                        &cache,
                        &privacy_filter,
                        cloud_tx.as_ref(),
                        &event_logger,
                        &config,
                        &mut diff_counters,
                    )
                    .await;
                }
            }
            ConfigChangeRequest::FsModified(paths) => {
                for path in paths {
                    handle_fs_event(
                        ChangeKind::Modified,
                        &path,
                        &manifest,
                        &cache,
                        &privacy_filter,
                        cloud_tx.as_ref(),
                        &event_logger,
                        &config,
                        &mut diff_counters,
                    )
                    .await;
                }
            }
            ConfigChangeRequest::FsRemoved(paths) => {
                for path in paths {
                    handle_fs_event(
                        ChangeKind::Removed,
                        &path,
                        &manifest,
                        &cache,
                        &privacy_filter,
                        cloud_tx.as_ref(),
                        &event_logger,
                        &config,
                        &mut diff_counters,
                    )
                    .await;
                }
            }
            ConfigChangeRequest::ProjectScopeRegister { project_root } => {
                // Folded: two sessions in the same project can report its root
                // in different casings on Windows, and each spelling would
                // otherwise register — and rescan — the project again.
                if !registered_projects.insert(crate::path_compat::dedup_key(&project_root)) {
                    // Same project already scanned this session — skip.
                    continue;
                }
                let emitted = run_project_scope_scan(
                    &project_root,
                    &manifest,
                    &cache,
                    &privacy_filter,
                    cloud_tx.as_ref(),
                    &event_logger,
                    &config,
                    &mut diff_counters,
                )
                .await;
                tracing::info!(
                    project_root = %project_root.display(),
                    items_emitted = emitted,
                    "config_monitor: project-scope registered"
                );
            }
        }
    }
}

/// Walk every manifest path and emit a snapshot for each existing file.
/// Returns the count of events emitted. The caller surrounds this with the
/// `config_initial_scan_*` telemetry pair.
#[allow(clippy::too_many_arguments)]
async fn run_full_walk(
    manifest: &Manifest,
    cache: &ContentHashCache,
    privacy_filter: &PrivacyFilter,
    cloud_tx: Option<&mpsc::Sender<CloudEvent>>,
    event_logger: &EventLogger,
    config: &Config,
    diff_counters: &mut HashMap<(String, String), u64>,
    source: EventSource,
    path_filter: Option<&Path>,
) -> usize {
    let mut emitted = 0usize;
    for agent in &manifest.agents {
        if agent.name != "claude-code" {
            continue;
        }
        for ap in &agent.paths {
            if ap.is_project_scoped() {
                continue;
            }
            for path in expand_paths_for_scan(ap) {
                if let Some(filter) = path_filter {
                    // The scan path comes from manifest expansion and the
                    // filter from a session's project root — two spellings of
                    // the same prefix on Windows, where an unfolded
                    // `starts_with` silently skips every path in the project.
                    if !crate::path_compat::dedup_key(&path)
                        .starts_with(crate::path_compat::dedup_key(filter))
                    {
                        continue;
                    }
                }
                if !path.exists() {
                    continue;
                }
                if is_excluded(&path, ap, manifest) {
                    continue;
                }
                let counter = diff_counters
                    .entry((agent.name.clone(), ap.kind.clone()))
                    .or_insert(0);
                let counter_value = match source {
                    EventSource::InitScan => 0,
                    _ => {
                        *counter = counter.saturating_add(1);
                        *counter
                    }
                };
                for event in scan_path_to_event(
                    &path,
                    ap,
                    &agent.name,
                    ChangeKind::Snapshot,
                    source,
                    counter_value,
                    privacy_filter,
                    cache,
                    config,
                ) {
                    log_and_forward(&event, cloud_tx, event_logger).await;
                    emitted += 1;
                }
            }
        }
    }
    emitted
}

/// Like `run_full_walk` but classifies snapshot vs change against the cache,
/// detects removals (cached entries whose paths no longer exist), and
/// returns `(observed, changed)`.
#[allow(clippy::too_many_arguments)]
async fn run_rescan(
    manifest: &Manifest,
    cache: &ContentHashCache,
    privacy_filter: &PrivacyFilter,
    cloud_tx: Option<&mpsc::Sender<CloudEvent>>,
    event_logger: &EventLogger,
    config: &Config,
    diff_counters: &mut HashMap<(String, String), u64>,
    path_filter: Option<&Path>,
) -> (usize, usize) {
    let mut observed = 0usize;
    let mut changed = 0usize;
    let mut seen: HashSet<PathBuf> = HashSet::new();

    for agent in &manifest.agents {
        if agent.name != "claude-code" {
            continue;
        }
        for ap in &agent.paths {
            if ap.is_project_scoped() {
                continue;
            }
            for path in expand_paths_for_scan(ap) {
                if let Some(filter) = path_filter {
                    // The scan path comes from manifest expansion and the
                    // filter from a session's project root — two spellings of
                    // the same prefix on Windows, where an unfolded
                    // `starts_with` silently skips every path in the project.
                    if !crate::path_compat::dedup_key(&path)
                        .starts_with(crate::path_compat::dedup_key(filter))
                    {
                        continue;
                    }
                }
                if !path.exists() {
                    continue;
                }
                if is_excluded(&path, ap, manifest) {
                    continue;
                }
                seen.insert(crate::path_compat::dedup_key(&path));
                observed += 1;
                let mut prev_hashes: HashMap<Option<String>, [u8; 32]> = HashMap::new();
                for entry in cache.entries_for_path(&path) {
                    prev_hashes.insert(entry.subpath.clone(), entry.content_hash);
                }
                let counter = diff_counters
                    .entry((agent.name.clone(), ap.kind.clone()))
                    .or_insert(0);
                *counter = counter.saturating_add(1);
                let counter_value = *counter;
                // Rescan emits Snapshot for both unchanged baselines (heartbeat)
                // and newly-observed paths; the cloud distinguishes via
                // diffcounter and the prev/new hash comparison below.
                let events = scan_path_to_event_with_hash(
                    &path,
                    ap,
                    &agent.name,
                    ChangeKind::Snapshot,
                    EventSource::Rescan,
                    counter_value,
                    privacy_filter,
                    cache,
                    config,
                );
                if events.is_empty() {
                    continue;
                }
                let mut emitted_subpaths: HashSet<Option<String>> = HashSet::new();
                for (event, new_hash, subpath) in events {
                    let prev = prev_hashes.get(&subpath).copied();
                    let is_change = prev.map(|p| p != new_hash).unwrap_or(true);
                    if is_change {
                        changed += 1;
                    }
                    emitted_subpaths.insert(subpath);
                    log_and_forward(&event, cloud_tx, event_logger).await;
                }
                // Detect server deletions inside a still-extant file: any
                // cached subpath the scan didn't re-emit is gone.
                for (sub, _) in prev_hashes
                    .iter()
                    .filter(|(s, _)| s.is_some() && !emitted_subpaths.contains(*s))
                {
                    if let Some(sub_str) = sub {
                        cache.remove_subpath(&path, Some(sub_str));
                        if let Some(removed_event) = build_removed_event_for_subpath(
                            &path,
                            Some(sub_str.as_str()),
                            &ap.kind,
                            &agent.name,
                            EventSource::Rescan,
                            *diff_counters
                                .entry((agent.name.clone(), ap.kind.clone()))
                                .and_modify(|c| *c = c.saturating_add(1))
                                .or_insert(1),
                            severity_for(&ap.kind, "removed"),
                            config,
                        ) {
                            log_and_forward(&removed_event, cloud_tx, event_logger).await;
                            changed += 1;
                        }
                    }
                }
            }
        }
    }

    // Detect removals: anything in the cache whose path is now gone.
    let mut to_remove: Vec<CacheEntry> = Vec::new();
    for entry in cache.snapshot() {
        if seen.contains(&crate::path_compat::dedup_key(&entry.path)) {
            continue;
        }
        if entry.path.exists() {
            continue;
        }
        to_remove.push(entry);
    }
    for entry in to_remove {
        cache.remove_subpath(&entry.path, entry.subpath.as_deref());
        if let Some(event) = build_removed_event_for_subpath(
            &entry.path,
            entry.subpath.as_deref(),
            &entry.kind,
            &entry.agent,
            EventSource::Rescan,
            diff_counters
                .entry((entry.agent.clone(), entry.kind.clone()))
                .and_modify(|c| *c = c.saturating_add(1))
                .or_insert(1)
                .to_owned(),
            severity_for(&entry.kind, "removed"),
            config,
        ) {
            log_and_forward(&event, cloud_tx, event_logger).await;
            changed += 1;
        }
    }

    (observed, changed)
}

/// Scan a project root: expand the manifest's project-scope `paths_relative`
/// and `paths_glob_relative` entries against `project_root`, hash each
/// existing file, emit `ai.openlatch.config.snapshot` events. Returns the
/// number of emitted events.
///
/// FS-watcher registration for project files is intentionally NOT done here
/// — project paths are observed only at SessionStart cwd time and via the
/// periodic 12 h rescan. Future enhancement: dynamic per-project watchers.
#[allow(clippy::too_many_arguments)]
async fn run_project_scope_scan(
    project_root: &Path,
    manifest: &Manifest,
    cache: &ContentHashCache,
    privacy_filter: &PrivacyFilter,
    cloud_tx: Option<&mpsc::Sender<CloudEvent>>,
    event_logger: &EventLogger,
    config: &Config,
    diff_counters: &mut HashMap<(String, String), u64>,
) -> usize {
    let mut emitted = 0usize;
    for agent in &manifest.agents {
        if agent.name != "claude-code" {
            continue;
        }
        for ap in &agent.paths {
            if !ap.is_project_scoped() {
                continue;
            }
            let mut candidates: Vec<PathBuf> = Vec::new();
            for rel in &ap.paths_relative {
                // `paths_relative` entries are bare relative paths — they
                // do NOT carry ${...} placeholders. Join directly under
                // the project root.
                candidates.push(project_root.join(rel));
            }
            for glob in &ap.paths_glob_relative {
                let pattern = project_root.join(glob);
                candidates.extend(glob_expand(&pattern));
            }
            candidates.sort();
            candidates.dedup();
            for path in candidates {
                if !path.exists() {
                    continue;
                }
                if is_excluded(&path, ap, manifest) {
                    continue;
                }
                let counter_value = {
                    let counter = diff_counters
                        .entry((agent.name.clone(), ap.kind.clone()))
                        .or_insert(0);
                    *counter = counter.saturating_add(1);
                    *counter
                };
                for event in scan_path_to_event(
                    &path,
                    ap,
                    &agent.name,
                    ChangeKind::Snapshot,
                    EventSource::InitScan,
                    counter_value,
                    privacy_filter,
                    cache,
                    config,
                ) {
                    log_and_forward(&event, cloud_tx, event_logger).await;
                    emitted += 1;
                }
            }
        }
    }
    emitted
}

#[allow(clippy::too_many_arguments)]
async fn handle_fs_event(
    kind: ChangeKind,
    path: &Path,
    manifest: &Manifest,
    cache: &ContentHashCache,
    privacy_filter: &PrivacyFilter,
    cloud_tx: Option<&mpsc::Sender<CloudEvent>>,
    event_logger: &EventLogger,
    config: &Config,
    diff_counters: &mut HashMap<(String, String), u64>,
) {
    let Some((agent_name, ap)) = lookup_path_in_manifest(path, manifest) else {
        return;
    };
    if is_excluded(path, ap, manifest) {
        return;
    }

    // An atomic rewrite (write temp file, then rename over the target) surfaces
    // as Remove followed by Create even though the file never went away. Taking
    // the Remove at face value wipes every per-subpath cache entry for the path,
    // so the paired Create can no longer suppress anything by hash and replays
    // the entire file as `removed` + `added`. Re-check existence instead: if the
    // file is back by the time we handle the event, this was a modification.
    let kind = if matches!(kind, ChangeKind::Removed) && path.exists() {
        ChangeKind::Modified
    } else {
        kind
    };

    let agent_owned = agent_name.to_string();
    let counter = diff_counters
        .entry((agent_owned.clone(), ap.kind.clone()))
        .or_insert(0);
    *counter = counter.saturating_add(1);
    let counter_value = *counter;

    let events: Vec<CloudEvent> = match kind {
        ChangeKind::Removed => {
            let removed_entries = cache.remove_all_under_path(path);
            if removed_entries.is_empty() {
                build_removed_event_for_subpath(
                    path,
                    None,
                    &ap.kind,
                    &agent_owned,
                    EventSource::FsWatcher,
                    counter_value,
                    severity_for(&ap.kind, "removed"),
                    config,
                )
                .into_iter()
                .collect()
            } else {
                let mut out = Vec::with_capacity(removed_entries.len());
                for (i, entry) in removed_entries.into_iter().enumerate() {
                    let c = if i == 0 {
                        counter_value
                    } else {
                        let counter = diff_counters
                            .entry((agent_owned.clone(), ap.kind.clone()))
                            .or_insert(0);
                        *counter = counter.saturating_add(1);
                        *counter
                    };
                    if let Some(event) = build_removed_event_for_subpath(
                        path,
                        entry.subpath.as_deref(),
                        &ap.kind,
                        &agent_owned,
                        EventSource::FsWatcher,
                        c,
                        severity_for(&ap.kind, "removed"),
                        config,
                    ) {
                        out.push(event);
                    }
                }
                out
            }
        }
        ChangeKind::Added | ChangeKind::Modified => {
            // Snapshot prior per-subpath hashes before the scan overwrites the
            // cache, so a no-op FS modify (native hook just forwarded the same
            // content) can be suppressed per server.
            let mut prev_hashes: HashMap<Option<String>, [u8; 32]> = HashMap::new();
            for entry in cache.entries_for_path(path) {
                prev_hashes.insert(entry.subpath.clone(), entry.content_hash);
            }
            let scanned = scan_path_to_event_with_hash(
                path,
                ap,
                &agent_owned,
                kind,
                EventSource::FsWatcher,
                counter_value,
                privacy_filter,
                cache,
                config,
            );
            let mut emitted_subpaths: HashSet<Option<String>> = HashSet::new();
            let mut out: Vec<CloudEvent> = Vec::with_capacity(scanned.len());
            for (event, new_hash, subpath) in scanned {
                emitted_subpaths.insert(subpath.clone());
                let prev = prev_hashes.get(&subpath).copied();
                // Identical hash against a live cache entry means nothing about
                // this subpath changed — regardless of whether the FS reported
                // Modified or Added. The Added case is the second half of an
                // atomic rewrite: the entry is already known and its content is
                // byte-identical, so it is not a newly-appeared server. A real
                // re-add after a real deletion has no cache entry left, so
                // `prev` is None there and the event still fires.
                if prev == Some(new_hash) {
                    continue;
                }
                out.push(event);
            }
            for (sub, _) in prev_hashes
                .iter()
                .filter(|(s, _)| s.is_some() && !emitted_subpaths.contains(*s))
            {
                if let Some(sub_str) = sub {
                    cache.remove_subpath(path, Some(sub_str));
                    let counter = diff_counters
                        .entry((agent_owned.clone(), ap.kind.clone()))
                        .or_insert(0);
                    *counter = counter.saturating_add(1);
                    let c = *counter;
                    if let Some(removed_event) = build_removed_event_for_subpath(
                        path,
                        Some(sub_str.as_str()),
                        &ap.kind,
                        &agent_owned,
                        EventSource::FsWatcher,
                        c,
                        severity_for(&ap.kind, "removed"),
                        config,
                    ) {
                        out.push(removed_event);
                    }
                }
            }
            out
        }
        ChangeKind::Snapshot => Vec::new(),
    };

    if events.is_empty() {
        return;
    }

    crate::telemetry::capture_global(crate::telemetry::Event::config_change_detected(
        &ap.kind,
        severity_for(&ap.kind, kind.severity_key()).as_str(),
        &agent_owned,
        match kind {
            ChangeKind::Added => "added",
            ChangeKind::Modified => "modified",
            ChangeKind::Removed => "removed",
            ChangeKind::Snapshot => "snapshot",
        },
    ));

    for event in &events {
        log_and_forward(event, cloud_tx, event_logger).await;
    }
}

fn lookup_path_in_manifest<'a>(
    path: &Path,
    manifest: &'a Manifest,
) -> Option<(&'a str, &'a AgentPath)> {
    for agent in &manifest.agents {
        for ap in &agent.paths {
            if ap.is_project_scoped() {
                continue;
            }
            for p in &ap.paths {
                if let Ok(expanded) = expand_path(p, None) {
                    if expanded == path {
                        return Some((agent.name.as_str(), ap));
                    }
                }
            }
            if let Some(g) = &ap.paths_glob {
                if let Ok(expanded) = expand_path(g, None) {
                    if let Some(pattern_str) = expanded.to_str() {
                        if let Ok(pattern) = glob::Pattern::new(pattern_str) {
                            if let Some(path_str) = path.to_str() {
                                if pattern.matches(path_str) {
                                    return Some((agent.name.as_str(), ap));
                                }
                            }
                        }
                    }
                }
            }
            for slice in &ap.json_slice_paths {
                if let Ok(expanded) = expand_path(&slice.path, None) {
                    if expanded == path {
                        return Some((agent.name.as_str(), ap));
                    }
                }
            }
        }
    }
    None
}

fn expand_paths_for_scan(ap: &AgentPath) -> Vec<PathBuf> {
    let mut out = Vec::new();
    match ap.watch_strategy {
        WatchStrategy::ExactFile | WatchStrategy::ExactFileWithSlice => {
            for p in &ap.paths {
                if let Ok(expanded) = expand_path(p, None) {
                    out.push(expanded);
                }
            }
            for slice in &ap.json_slice_paths {
                if let Ok(expanded) = expand_path(&slice.path, None) {
                    out.push(expanded);
                }
            }
        }
        WatchStrategy::Glob => {
            if let Some(g) = &ap.paths_glob {
                if let Ok(expanded) = expand_path(g, None) {
                    out.extend(glob_expand(&expanded));
                }
            }
        }
        WatchStrategy::ExactFileAndGlob => {
            for p in &ap.paths {
                if let Ok(expanded) = expand_path(p, None) {
                    out.push(expanded);
                }
            }
            if let Some(g) = &ap.paths_glob {
                if let Ok(expanded) = expand_path(g, None) {
                    out.extend(glob_expand(&expanded));
                }
            }
        }
    }
    out.sort();
    out.dedup();
    out
}

#[allow(clippy::too_many_arguments)]
fn scan_path_to_event(
    path: &Path,
    ap: &AgentPath,
    agent_name: &str,
    kind: ChangeKind,
    source: EventSource,
    diffcounter: u64,
    privacy_filter: &PrivacyFilter,
    cache: &ContentHashCache,
    config: &Config,
) -> Vec<CloudEvent> {
    scan_path_to_event_with_hash(
        path,
        ap,
        agent_name,
        kind,
        source,
        diffcounter,
        privacy_filter,
        cache,
        config,
    )
    .into_iter()
    .map(|(event, _, _)| event)
    .collect()
}

/// Returns `(event, content_hash, subpath)` tuples. `subpath` is `Some`
/// only on fan-out kinds (MCP with `slice_kind_subpath = true`).
#[allow(clippy::too_many_arguments)]
fn scan_path_to_event_with_hash(
    path: &Path,
    ap: &AgentPath,
    agent_name: &str,
    kind: ChangeKind,
    source: EventSource,
    diffcounter: u64,
    privacy_filter: &PrivacyFilter,
    cache: &ContentHashCache,
    config: &Config,
) -> Vec<(CloudEvent, [u8; 32], Option<String>)> {
    let raw = match std::fs::read_to_string(path) {
        Ok(s) => s,
        Err(e) => {
            tracing::debug!(
                path = %path.display(),
                error = %e,
                "config_monitor: read failed during scan"
            );
            return Vec::new();
        }
    };
    build_event_from_content(
        path,
        ap,
        agent_name,
        kind,
        source,
        diffcounter,
        &raw,
        privacy_filter,
        cache,
        config,
    )
}

#[allow(clippy::too_many_arguments)]
fn build_event_from_content(
    path: &Path,
    ap: &AgentPath,
    agent_name: &str,
    kind: ChangeKind,
    source: EventSource,
    diffcounter: u64,
    raw: &str,
    privacy_filter: &PrivacyFilter,
    cache: &ContentHashCache,
    config: &Config,
) -> Vec<(CloudEvent, [u8; 32], Option<String>)> {
    let slice_pointers: Vec<&str> =
        if matches!(ap.watch_strategy, WatchStrategy::ExactFileWithSlice) {
            ap.json_slice_paths
                .iter()
                .filter_map(|s| {
                    expand_path(&s.path, None)
                        .ok()
                        .and_then(|exp| (exp == path).then_some(s.json_pointer.as_str()))
                })
                .collect()
        } else {
            Vec::new()
        };

    if ap.kind == "mcp" && ap.slice_kind_subpath {
        return build_mcp_fanout_events(
            path,
            ap,
            agent_name,
            kind,
            source,
            diffcounter,
            raw,
            &slice_pointers,
            privacy_filter,
            cache,
            config,
        );
    }

    let content_hash = match hash_for_kind(&ap.kind, raw, &slice_pointers) {
        Ok(h) => h,
        Err(e) => {
            tracing::warn!(
                code = crate::error::ERR_INVENTORY_HASH_FAILED,
                path = %path.display(),
                error = ?e,
                "config_monitor: hash pipeline failed"
            );
            return Vec::new();
        }
    };
    let path_hash = sha256(path.to_string_lossy().as_bytes());

    cache.insert(CacheEntry {
        path: path.to_path_buf(),
        subpath: None,
        path_hash,
        content_hash,
        last_observed: Instant::now(),
        kind: ap.kind.clone(),
        agent: agent_name.to_string(),
    });

    let severity = severity_for(&ap.kind, kind.severity_key());

    let data = match config.inventory_monitor.content_forward {
        ContentForwardMode::HashOnly => Value::Null,
        ContentForwardMode::Filtered => {
            filtered_payload(raw, &ap.kind, ap.scope, path, privacy_filter, config)
        }
        ContentForwardMode::FullUnfiltered => {
            unfiltered_payload(raw, &ap.kind, ap.scope, path, config)
        }
    };

    let event = build_modified_event(
        path,
        &ap.kind,
        agent_name,
        &content_hash,
        &path_hash,
        kind,
        source,
        diffcounter,
        severity,
        data,
        config,
    );
    vec![(event, content_hash, None)]
}

/// Resolve the MCP server entries a fan-out should emit for one file.
///
/// When the manifest declares JSON Pointer slices for this path, ONLY those
/// slices are consulted:
///
/// - a pointer landing on an `mcpServers` map contributes its entries directly;
/// - a pointer landing on a container of per-project blocks (`/projects` in
///   `~/.claude.json`) contributes each block's own `mcpServers` map, with the
///   block key namespaced into the subpath so two projects declaring the same
///   server name stay distinct rows.
///
/// The whole-document fallback applies only to files with NO declared slices —
/// standalone `.mcp.json` documents whose top level *is* the server map.
/// Applying it to a sliced file was the bug: `~/.claude.json` has no
/// `/mcpServers` key, so all 65 of its top-level preference keys (`userID`,
/// `numStartups`, `tipsHistory`, …) were emitted as phantom MCP servers, and
/// every rewrite of that file — Claude Code rewrites it constantly — replayed
/// the whole set through the cloud and event-log channels.
fn resolve_mcp_servers(parsed: &Value, slice_pointers: &[&str]) -> serde_json::Map<String, Value> {
    if slice_pointers.is_empty() {
        return parsed
            .pointer("/mcpServers")
            .and_then(|v| v.as_object())
            .cloned()
            .or_else(|| parsed.as_object().cloned())
            .unwrap_or_default();
    }

    let mut out = serde_json::Map::new();
    for pointer in slice_pointers {
        let Some(sliced) = parsed.pointer(pointer).and_then(|v| v.as_object()) else {
            continue;
        };
        if pointer.rsplit('/').next() == Some("mcpServers") {
            for (name, cfg) in sliced {
                out.insert(name.clone(), cfg.clone());
            }
        } else {
            for (block, value) in sliced {
                let Some(nested) = value.pointer("/mcpServers").and_then(|v| v.as_object()) else {
                    continue;
                };
                for (name, cfg) in nested {
                    out.insert(format!("{block}::{name}"), cfg.clone());
                }
            }
        }
    }
    out
}

/// MCP fan-out emission: parse the file, then produce one envelope per
/// server entry with a per-server `configpathhash` / `configcontenthash`
/// so the platform tracks each as its own row. Empty result on parse
/// failure (degrade to silent — the next FS event retries).
#[allow(clippy::too_many_arguments)]
fn build_mcp_fanout_events(
    path: &Path,
    ap: &AgentPath,
    agent_name: &str,
    kind: ChangeKind,
    source: EventSource,
    diffcounter: u64,
    raw: &str,
    slice_pointers: &[&str],
    privacy_filter: &PrivacyFilter,
    cache: &ContentHashCache,
    config: &Config,
) -> Vec<(CloudEvent, [u8; 32], Option<String>)> {
    let parsed: Value = match serde_json::from_str(raw) {
        Ok(v) => v,
        Err(e) => {
            tracing::debug!(
                path = %path.display(),
                error = %e,
                "config_monitor: mcp parse failed; falling back to non-fanout event"
            );
            return Vec::new();
        }
    };
    let servers = resolve_mcp_servers(&parsed, slice_pointers);

    if servers.is_empty() {
        return Vec::new();
    }

    let severity = severity_for(&ap.kind, kind.severity_key());
    let mut out = Vec::with_capacity(servers.len());

    for (server_name, server_value) in servers {
        let content_hash = hash_mcp_server_entry(&server_name, &server_value);
        let path_hash = subpath_path_hash(path, Some(&server_name));

        cache.insert(CacheEntry {
            path: path.to_path_buf(),
            subpath: Some(server_name.clone()),
            path_hash,
            content_hash,
            last_observed: Instant::now(),
            kind: ap.kind.clone(),
            agent: agent_name.to_string(),
        });

        let data = match config.inventory_monitor.content_forward {
            ContentForwardMode::HashOnly => Value::Null,
            ContentForwardMode::Filtered => {
                let mut payload = serde_json::json!({
                    "server_name": server_name,
                    "tools": [],
                });
                filter_event_with(&mut payload, privacy_filter);
                attach_scope(&mut payload, &ap.kind, ap.scope);
                payload
            }
            ContentForwardMode::FullUnfiltered => {
                if std::env::var("OPENLATCH_TESTING").as_deref() == Ok("true") {
                    let mut payload = serde_json::json!({
                        "server_name": server_name,
                        "tools": [],
                        "raw": server_value,
                    });
                    attach_scope(&mut payload, &ap.kind, ap.scope);
                    payload
                } else {
                    let mut payload = serde_json::json!({
                        "server_name": server_name,
                        "tools": [],
                    });
                    filter_event_with(&mut payload, privacy_filter);
                    attach_scope(&mut payload, &ap.kind, ap.scope);
                    payload
                }
            }
        };

        let event = build_modified_event(
            path,
            &ap.kind,
            agent_name,
            &content_hash,
            &path_hash,
            kind,
            source,
            diffcounter,
            severity,
            data,
            config,
        );
        out.push((event, content_hash, Some(server_name)));
    }
    out
}

fn filtered_payload(
    raw: &str,
    kind: &str,
    scope: Option<ConfigScope>,
    path: &Path,
    privacy_filter: &PrivacyFilter,
    config: &Config,
) -> Value {
    let inline_cap = config.inventory_monitor.max_inline_content_bytes as usize;
    let mut payload = build_filtered_body(raw, kind, inline_cap, path, privacy_filter);
    attach_scope(&mut payload, kind, scope);
    payload
}

/// Produce the inner `data` body for `filtered_payload`. Caller stamps the
/// optional `scope` tag afterwards.
///
/// Platform parser (`app/config_plane/parser.py::Parsed*Artifact`) uses
/// `extra="forbid"`, so any unknown field 400s the envelope. rules / skill
/// / command share the markdown-body shape produced below. mcp + hooks
/// each have their own structural shape — see the per-arm logic.
fn build_filtered_body(
    raw: &str,
    kind: &str,
    inline_cap: usize,
    path: &Path,
    privacy_filter: &PrivacyFilter,
) -> Value {
    match kind {
        "rules" => {
            let (frontmatter, body_only) = split_frontmatter(raw);
            let cap = per_kind_body_cap(kind, inline_cap);
            let body = truncate_body_in_place(&body_only, cap);
            let mut body_v = Value::String(body);
            filter_event_with(&mut body_v, privacy_filter);
            let mut payload = serde_json::Map::new();
            payload.insert("body".into(), body_v);
            payload.insert("frontmatter".into(), frontmatter);
            payload.insert(
                "path".into(),
                Value::String(truncate_path_for_platform(path)),
            );
            Value::Object(payload)
        }
        "skill" | "command" => {
            let (frontmatter, body_only) = split_frontmatter(raw);
            let cap = per_kind_body_cap(kind, inline_cap);
            let body = truncate_body_in_place(&body_only, cap);
            let mut body_v = Value::String(body);
            filter_event_with(&mut body_v, privacy_filter);
            let (name, description) = derive_name_and_description(&frontmatter, kind, path);
            let mut payload = serde_json::Map::new();
            payload.insert("body".into(), body_v);
            payload.insert("frontmatter".into(), frontmatter);
            payload.insert("name".into(), Value::String(name));
            payload.insert("description".into(), Value::String(description));
            Value::Object(payload)
        }
        "hooks" => build_hooks_body(raw, privacy_filter),
        "mcp" => {
            // Non-fan-out fallback. Fan-out callers use `build_mcp_fanout_events`
            // directly; reaching this arm means the agent_path is not
            // `slice_kind_subpath`-flagged. Emit a degraded shape carrying the
            // required `server_name` so the platform Pydantic validator accepts
            // the envelope.
            let server_name = derive_name_from_path(path);
            serde_json::json!({
                "server_name": server_name,
                "tools": [],
            })
        }
        _ => legacy_text_payload(kind, raw, privacy_filter),
    }
}

/// Hooks payload — `{hooks: {...}}` matching `ParsedHooksArtifact`. Lifts
/// the `/hooks` slice if present (settings.json shape) and otherwise treats
/// the whole file as the hooks dict (plugin-installed `hooks/hooks.json`
/// shape).
fn build_hooks_body(raw: &str, privacy_filter: &PrivacyFilter) -> Value {
    let parsed: Value = match jsonc_parser::parse_to_serde_value(raw, &Default::default()) {
        Ok(Some(v)) => v,
        Ok(None) | Err(_) => match serde_json::from_str::<Value>(raw) {
            Ok(v) => v,
            Err(_) => {
                tracing::debug!(
                    "config_monitor: hooks file parse failed; emitting empty hooks dict"
                );
                return serde_json::json!({"hooks": {}});
            }
        },
    };
    let hooks_obj = parsed
        .pointer("/hooks")
        .cloned()
        .filter(|v| v.is_object())
        .unwrap_or_else(|| {
            if parsed.is_object() {
                parsed
            } else {
                Value::Object(Default::default())
            }
        });
    let mut wrapped = serde_json::json!({"hooks": hooks_obj});
    filter_event_with(&mut wrapped, privacy_filter);
    wrapped
}

/// Separator between a file path and its subpath when deriving a unique
/// `configpathhash` for fan-out kinds (one MCP server inside a shared
/// file). Both the `modified` and `removed` envelope builders MUST use
/// the same separator or platform-side rows will desync.
const SUBPATH_HASH_SEP: char = '#';

fn subpath_path_hash(path: &Path, subpath: Option<&str>) -> [u8; 32] {
    let path_str = path.to_string_lossy();
    match subpath {
        Some(sub) => sha256(format!("{path_str}{SUBPATH_HASH_SEP}{sub}").as_bytes()),
        None => sha256(path_str.as_bytes()),
    }
}

/// Canonicalize a single MCP server entry for stable per-server hashing.
/// Returns the SHA-256 of the JCS canonical form of `{server_name: cfg}`.
fn hash_mcp_server_entry(server_name: &str, server_value: &Value) -> [u8; 32] {
    let nfc: String = server_name.nfc().collect();
    let mut wrap = serde_json::Map::new();
    wrap.insert(nfc, server_value.clone());
    let synth = Value::Object(wrap);
    let canonical = serde_json_canonicalizer::to_string(&synth)
        .unwrap_or_else(|_| serde_json::to_string(&synth).unwrap_or_default());
    sha256(canonical.as_bytes())
}

/// Fallback `{kind, text}` shape for configkinds whose proper extractor
/// isn't implemented yet. Ingest 400s these — they're sent so the audit
/// trail survives until the extractor lands.
fn legacy_text_payload(kind: &str, raw: &str, privacy_filter: &PrivacyFilter) -> Value {
    let mut as_string = Value::String(raw.to_string());
    filter_event_with(&mut as_string, privacy_filter);
    serde_json::json!({"kind": kind, "text": as_string})
}

/// Hard caps from `parser.py::Parsed*Artifact`. Anything over → 400.
const PLATFORM_RULES_BODY_MAX: usize = 512 * 1024;
const PLATFORM_SKILL_COMMAND_BODY_MAX: usize = 128 * 1024;
const PLATFORM_NAME_MAX: usize = 128;
const PLATFORM_PATH_MAX: usize = 1024;
/// Reserve in `truncate_body_in_place` for the truncation marker.
const TRUNCATION_MARKER_RESERVE: usize = 64;

fn per_kind_body_cap(kind: &str, inline_cap: usize) -> usize {
    let platform_max = match kind {
        "rules" => PLATFORM_RULES_BODY_MAX,
        "skill" | "command" => PLATFORM_SKILL_COMMAND_BODY_MAX,
        _ => usize::MAX,
    };
    std::cmp::min(inline_cap, platform_max)
}

/// Preserves the head + tail of the content (where injection markers tend
/// to cluster) so platform detectors still have signal on truncated files.
fn truncate_body_in_place(raw: &str, cap: usize) -> String {
    if raw.len() <= cap {
        return raw.to_string();
    }
    let half = cap.saturating_sub(TRUNCATION_MARKER_RESERVE) / 2;
    let head_end = raw.char_indices().nth(half).map(|(i, _)| i).unwrap_or(half);
    // Walk forward from raw.len()-half to land on a UTF-8 char boundary
    // in O(1) — avoids the O(n) reverse iteration the previous version did.
    let tail_target = raw.len().saturating_sub(half);
    let mut tail_start = tail_target;
    while tail_start < raw.len() && !raw.is_char_boundary(tail_start) {
        tail_start += 1;
    }
    let head = &raw[..head_end];
    let tail = &raw[tail_start..];
    let marker = format!(
        "\n\n…<truncated {} bytes>…\n\n",
        raw.len() - head.len() - tail.len()
    );
    format!("{head}{marker}{tail}")
}

fn derive_name_from_path(path: &Path) -> String {
    let stem = path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("unnamed");
    if stem.is_empty() {
        return "unnamed".into();
    }
    if stem.len() <= PLATFORM_NAME_MAX {
        return stem.to_string();
    }
    stem.chars().take(PLATFORM_NAME_MAX).collect()
}

/// Skills live at `<dir>/<skill-name>/SKILL.md` upstream — the file stem is
/// always the literal `SKILL`, so the useful name is the parent directory.
/// Commands live at `<dir>/<name>.md`, so the file stem is the name. Both
/// are overridden by frontmatter `name` when present.
fn derive_name_and_description(frontmatter: &Value, kind: &str, path: &Path) -> (String, String) {
    let mut name = frontmatter
        .get("name")
        .and_then(|v| v.as_str())
        .map(str::to_string)
        .unwrap_or_default();
    if name.is_empty() {
        name = match kind {
            "skill" => path
                .parent()
                .and_then(|p| p.file_name())
                .and_then(|s| s.to_str())
                .map(str::to_string)
                .unwrap_or_else(|| derive_name_from_path(path)),
            _ => derive_name_from_path(path),
        };
    }
    if name.is_empty() {
        name = "unnamed".to_string();
    }
    if name.chars().count() > PLATFORM_NAME_MAX {
        name = name.chars().take(PLATFORM_NAME_MAX).collect();
    }
    let description = frontmatter
        .get("description")
        .and_then(|v| v.as_str())
        .map(str::to_string)
        .unwrap_or_default();
    // ParsedSkillArtifact / ParsedCommandArtifact cap description at 4096.
    let description = if description.len() > 4096 {
        description.chars().take(4096).collect()
    } else {
        description
    };
    (name, description)
}

/// Parse a leading YAML frontmatter block delimited by `---` lines. Returns
/// `(frontmatter_object, body_without_frontmatter)`. When no frontmatter is
/// present, returns `(Object({}), raw.to_string())`.
///
/// Scope is intentionally narrow: top-level `key: value` pairs with
/// optional single/double-quoted string values. Multi-line scalars and
/// nested mappings are not supported — Claude Code's skill/command spec
/// only uses flat string scalars, and the parser fails open (returns the
/// empty map + full body) on anything more complex.
fn split_frontmatter(raw: &str) -> (Value, String) {
    let stripped = raw.strip_prefix('\u{FEFF}').unwrap_or(raw);
    let after_optional_lf = stripped.strip_prefix('\n').unwrap_or(stripped);
    let body = after_optional_lf;
    let first_line_terminator = if body.starts_with("---\r\n") {
        Some(5)
    } else if body.starts_with("---\n") {
        Some(4)
    } else if body == "---" {
        Some(3)
    } else {
        None
    };
    let Some(start) = first_line_terminator else {
        return (Value::Object(Default::default()), raw.to_string());
    };
    let after_open = &body[start..];
    let mut close_offset = None;
    let mut cursor = 0;
    for line in after_open.split_inclusive('\n') {
        let stripped_line = line.trim_end_matches(['\n', '\r']);
        if stripped_line == "---" {
            close_offset = Some(cursor + line.len());
            break;
        }
        cursor += line.len();
    }
    let Some(close) = close_offset else {
        return (Value::Object(Default::default()), raw.to_string());
    };
    let frontmatter_block = &after_open[..cursor];
    let body_remainder = &after_open[close..];
    let body_remainder = body_remainder.strip_prefix('\n').unwrap_or(body_remainder);

    let mut map = serde_json::Map::new();
    for line in frontmatter_block.lines() {
        let trimmed = line.trim();
        if trimmed.is_empty() || trimmed.starts_with('#') {
            continue;
        }
        let Some((key, value)) = trimmed.split_once(':') else {
            continue;
        };
        let key = key.trim().to_string();
        if key.is_empty() {
            continue;
        }
        let mut value = value.trim().to_string();
        if value.len() >= 2 {
            let bytes = value.as_bytes();
            let first = bytes[0];
            let last = bytes[bytes.len() - 1];
            if (first == b'"' && last == b'"') || (first == b'\'' && last == b'\'') {
                value = value[1..value.len() - 1].to_string();
            }
        }
        map.insert(key, Value::String(value));
    }
    (Value::Object(map), body_remainder.to_string())
}

/// Truncate from the left (preserves the filename) when the path exceeds
/// the platform's char-bounded `ParsedRulesArtifact.path` field.
fn truncate_path_for_platform(path: &Path) -> String {
    let s = path.to_string_lossy();
    let count = s.chars().count();
    if count <= PLATFORM_PATH_MAX {
        return s.into_owned();
    }
    s.chars().skip(count - PLATFORM_PATH_MAX).collect()
}

fn unfiltered_payload(
    raw: &str,
    kind: &str,
    scope: Option<ConfigScope>,
    path: &Path,
    config: &Config,
) -> Value {
    if std::env::var("OPENLATCH_TESTING").as_deref() != Ok("true") {
        tracing::warn!(
            "config_monitor: full_unfiltered content_forward refused (OPENLATCH_TESTING != true)"
        );
        return filtered_payload(raw, kind, scope, path, &PrivacyFilter::new(&[]), config);
    }
    let mut payload = serde_json::json!({"kind": kind, "raw": raw});
    attach_scope(&mut payload, kind, scope);
    payload
}

/// Attach the MCP precedence-scope tag to the payload `data` object.
///
/// The cloud routing engine uses `data.scope` to resolve same-named MCP
/// servers discovered at multiple filesystem tiers (enterprise > personal >
/// project > local). The tag is omitted when the manifest entry has no
/// declared scope and on non-MCP kinds — platform treats absence as the
/// least-specific tier.
fn attach_scope(payload: &mut Value, kind: &str, scope: Option<ConfigScope>) {
    if kind != "mcp" {
        return;
    }
    if let (Some(scope), Some(obj)) = (scope, payload.as_object_mut()) {
        obj.insert(
            "scope".to_string(),
            Value::String(scope.as_str().to_string()),
        );
    }
}

#[allow(clippy::too_many_arguments)]
fn build_modified_event(
    path: &Path,
    kind: &str,
    agent_name: &str,
    content_hash: &[u8; 32],
    path_hash: &[u8; 32],
    change: ChangeKind,
    source: EventSource,
    diffcounter: u64,
    severity: Severity,
    data: Value,
    config: &Config,
) -> CloudEvent {
    let path_hash_hex = hex::encode(path_hash);
    let content_hash_hex = hex::encode(content_hash);

    let (resolved_path, is_symlink) = match path.symlink_metadata() {
        Ok(m) if m.file_type().is_symlink() => (std::fs::canonicalize(path).ok(), true),
        _ => (None, false),
    };

    let event_type = format!("ai.openlatch.config.{}", change.type_suffix());

    let mut envelope = serde_json::json!({
        "specversion": "1.0",
        "id": crate::envelope::new_event_id(),
        "source": agent_name,
        "type": event_type,
        "time": crate::envelope::current_timestamp(),
        "datacontenttype": "application/json",
        "configkind": kind,
        "configsource": agent_name,
        "configpathhash": path_hash_hex,
        "configcontenthash": content_hash_hex,
        "diffcounter": diffcounter,
        "severityhint": severity.as_str(),
        "eventsource": source.as_str(),
        "configpath": path.display().to_string(),
        "configissymlink": is_symlink,
        "data": data,
    });
    if let Some(rp) = resolved_path {
        // `canonicalize` hands back a `\\?\`-prefixed path on Windows; the
        // wire carries the form the user would recognise.
        envelope["configresolvedpath"] = serde_json::json!(crate::path_compat::display_path(&rp));
    }

    CloudEvent {
        envelope,
        agent_id: config.agent_id.clone().unwrap_or_default(),
    }
}

#[allow(clippy::too_many_arguments)]
fn build_removed_event_for_subpath(
    path: &Path,
    subpath: Option<&str>,
    kind: &str,
    agent_name: &str,
    source: EventSource,
    diffcounter: u64,
    severity: Severity,
    config: &Config,
) -> Option<CloudEvent> {
    let path_hash_hex = hex::encode(subpath_path_hash(path, subpath));
    let envelope = serde_json::json!({
        "specversion": "1.0",
        "id": crate::envelope::new_event_id(),
        "source": agent_name,
        "type": "ai.openlatch.config.removed",
        "time": crate::envelope::current_timestamp(),
        "datacontenttype": "application/json",
        "configkind": kind,
        "configsource": agent_name,
        "configpathhash": path_hash_hex,
        "diffcounter": diffcounter,
        "severityhint": severity.as_str(),
        "eventsource": source.as_str(),
        "configpath": path.display().to_string(),
        "data": Value::Null,
    });
    Some(CloudEvent {
        envelope,
        agent_id: config.agent_id.clone().unwrap_or_default(),
    })
}

async fn log_and_forward(
    event: &CloudEvent,
    cloud_tx: Option<&mpsc::Sender<CloudEvent>>,
    event_logger: &EventLogger,
) {
    // BACKPRESSURE, not fail-open. The verdict path uses `try_send` on both of
    // these rails because a hook call must never wait on a queue. The config
    // monitor is a background producer under no latency budget, and it emits in
    // bursts — the boot inventory walk pushes its whole result set through this
    // function in one loop that yields nowhere else. Dropping there is silent
    // audit loss reported as a WARN per event, which is how a burst turned into
    // thousands of "channel full" lines. Awaiting paces the walk to whichever
    // consumer is slower and loses nothing.
    if let Ok(line) = serde_json::to_string(&event.envelope) {
        event_logger.log_backpressured(line).await;
    }
    if let Some(tx) = cloud_tx {
        if tx.send(event.clone()).await.is_err() {
            tracing::warn!(
                code = crate::error::ERR_CLOUD_UNREACHABLE,
                "config_monitor: cloud channel closed — config event dropped"
            );
        }
    }
}

// ---------------------------------------------------------------------------
// Hash pipeline — NFC → JCS → SHA-256 (JSON) / NFC → line-normalize →
// SHA-256 (text) / strip-comments → NFC → JCS → SHA-256 (jsonc)
// ---------------------------------------------------------------------------

#[derive(thiserror::Error, Debug)]
pub enum HashError {
    #[error("JSON parse failed: {0}")]
    JsonParse(#[from] serde_json::Error),
    #[error("JCS canonicalization failed: {0}")]
    JcsCanonicalize(String),
    #[error("JSONC parse failed: {0}")]
    JsoncParse(String),
}

fn hash_for_kind(kind: &str, raw: &str, slice_pointers: &[&str]) -> Result<[u8; 32], HashError> {
    match kind {
        "mcp" => {
            if slice_pointers.is_empty() {
                content_hash_json(raw)
            } else {
                content_hash_json_slices(raw, slice_pointers)
            }
        }
        "hooks" => {
            if slice_pointers.is_empty() {
                content_hash_jsonc(raw)
            } else {
                content_hash_jsonc_slices(raw, slice_pointers)
            }
        }
        _ => Ok(content_hash_text(raw)),
    }
}

/// Production hash for JSON content. NFC → JCS → SHA-256.
pub fn content_hash_json(text: &str) -> Result<[u8; 32], HashError> {
    let nfc: String = text.nfc().collect();
    let value: Value = serde_json::from_str(&nfc)?;
    let canonical = serde_json_canonicalizer::to_string(&value)
        .map_err(|e| HashError::JcsCanonicalize(e.to_string()))?;
    Ok(sha256(canonical.as_bytes()))
}

/// Slice variant of `content_hash_json`: parse, extract each JSON Pointer
/// subtree, combine into a synthetic array (pointer order), canonicalize,
/// SHA-256. Missing pointers slot in as `Value::Null` — deterministic and
/// signals "no slice present" rather than crashing.
pub fn content_hash_json_slices(text: &str, pointers: &[&str]) -> Result<[u8; 32], HashError> {
    let nfc: String = text.nfc().collect();
    let value: Value = serde_json::from_str(&nfc)?;
    hash_slices(&value, pointers)
}

/// JSONC variant: strip comments via jsonc-parser, then NFC + JCS.
pub fn content_hash_jsonc(text: &str) -> Result<[u8; 32], HashError> {
    let nfc: String = text.nfc().collect();
    let parsed: Value = jsonc_parser::parse_to_serde_value(&nfc, &Default::default())
        .map_err(|e| HashError::JsoncParse(e.to_string()))?;
    let canonical = serde_json_canonicalizer::to_string(&parsed)
        .map_err(|e| HashError::JcsCanonicalize(e.to_string()))?;
    Ok(sha256(canonical.as_bytes()))
}

/// Slice variant of `content_hash_jsonc`: strip comments, extract pointer
/// subtrees, combine into a synthetic array (pointer order), canonicalize,
/// SHA-256.
pub fn content_hash_jsonc_slices(text: &str, pointers: &[&str]) -> Result<[u8; 32], HashError> {
    let nfc: String = text.nfc().collect();
    let parsed: Value = jsonc_parser::parse_to_serde_value(&nfc, &Default::default())
        .map_err(|e| HashError::JsoncParse(e.to_string()))?;
    hash_slices(&parsed, pointers)
}

fn hash_slices(root: &Value, pointers: &[&str]) -> Result<[u8; 32], HashError> {
    let subtrees: Vec<Value> = pointers
        .iter()
        .map(|ptr| root.pointer(ptr).cloned().unwrap_or(Value::Null))
        .collect();
    let synthetic = Value::Array(subtrees);
    let canonical = serde_json_canonicalizer::to_string(&synthetic)
        .map_err(|e| HashError::JcsCanonicalize(e.to_string()))?;
    Ok(sha256(canonical.as_bytes()))
}

/// Markdown / plain-text variant. NFC → CRLF/CR → LF → strip trailing
/// whitespace per line → SHA-256.
pub fn content_hash_text(text: &str) -> [u8; 32] {
    let nfc: String = text.nfc().collect();
    let normalized_eol = nfc.replace("\r\n", "\n").replace('\r', "\n");
    let lines: Vec<&str> = normalized_eol.split('\n').map(|l| l.trim_end()).collect();
    let normalized = lines.join("\n");
    sha256(normalized.as_bytes())
}

fn sha256(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

// ---------------------------------------------------------------------------
// Severity classifier (hardcoded; manifest does NOT declare severity)
// ---------------------------------------------------------------------------

pub fn severity_for(kind: &str, change_type: &str) -> Severity {
    match (kind, change_type) {
        ("mcp", "added") => Severity::Critical,
        ("mcp", "modified") => Severity::High,
        ("mcp", "removed") => Severity::Medium,
        ("skill", "added") => Severity::High,
        ("skill", "modified") => Severity::High,
        ("skill", "removed") => Severity::Low,
        ("hooks", "added") => Severity::Critical,
        ("hooks", "modified") => Severity::Critical,
        ("hooks", "removed") => Severity::High,
        ("command", "added") => Severity::High,
        ("command", "modified") => Severity::Medium,
        ("command", "removed") => Severity::Low,
        ("rules", "added") => Severity::Medium,
        ("rules", "modified") => Severity::Medium,
        ("rules", "removed") => Severity::Info,
        _ => Severity::Low,
    }
}

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

    #[test]
    fn content_hash_json_canonicalizes_keys() {
        let a = r#"{"b":1,"a":2}"#;
        let b = r#"{"a":2,"b":1}"#;
        assert_eq!(content_hash_json(a).unwrap(), content_hash_json(b).unwrap());
    }

    #[test]
    fn content_hash_text_normalizes_eol() {
        let a = "line1\nline2\n";
        let b = "line1\r\nline2\r\n";
        let c = "line1\rline2\r";
        assert_eq!(content_hash_text(a), content_hash_text(b));
        assert_eq!(content_hash_text(a), content_hash_text(c));
    }

    #[test]
    fn content_hash_text_strips_trailing_whitespace_per_line() {
        let a = "line1\nline2";
        let b = "line1   \nline2\t";
        assert_eq!(content_hash_text(a), content_hash_text(b));
    }

    #[test]
    fn content_hash_jsonc_strips_comments() {
        let with_comments = r#"{
            // comment
            "a": 1
        }"#;
        let plain = r#"{"a":1}"#;
        assert_eq!(
            content_hash_jsonc(with_comments).unwrap(),
            content_hash_json(plain).unwrap()
        );
    }

    #[test]
    fn content_hash_json_slices_ignores_surrounding_keys() {
        let with_theme = r#"{"theme":"dark","hooks":{"PreToolUse":[]}}"#;
        let without_theme = r#"{"hooks":{"PreToolUse":[]}}"#;
        let pointers = ["/hooks"];
        assert_eq!(
            content_hash_json_slices(with_theme, &pointers).unwrap(),
            content_hash_json_slices(without_theme, &pointers).unwrap(),
        );
    }

    #[test]
    fn content_hash_json_slices_detects_slice_edits() {
        let before = r#"{"hooks":{"PreToolUse":[]}}"#;
        let after = r#"{"hooks":{"PreToolUse":[{"matcher":"*"}]}}"#;
        let pointers = ["/hooks"];
        assert_ne!(
            content_hash_json_slices(before, &pointers).unwrap(),
            content_hash_json_slices(after, &pointers).unwrap(),
        );
    }

    #[test]
    fn content_hash_jsonc_slices_strips_comments_and_slices() {
        let jsonc_with_comments = r#"{
            // ignore
            "theme": "dark",
            "hooks": { "PreToolUse": [] }
        }"#;
        let plain = r#"{"hooks":{"PreToolUse":[]}}"#;
        let pointers = ["/hooks"];
        assert_eq!(
            content_hash_jsonc_slices(jsonc_with_comments, &pointers).unwrap(),
            content_hash_json_slices(plain, &pointers).unwrap(),
        );
    }

    #[test]
    fn content_hash_json_slices_missing_pointer_is_deterministic() {
        let a = r#"{"other":1}"#;
        let b = r#"{"other":2}"#;
        let pointers = ["/hooks"];
        assert_eq!(
            content_hash_json_slices(a, &pointers).unwrap(),
            content_hash_json_slices(b, &pointers).unwrap(),
        );
    }

    #[test]
    fn content_hash_json_slices_multi_pointer_order_matters() {
        let raw = r#"{"a":1,"b":2,"c":3}"#;
        let forward = ["/a", "/b"];
        let reversed = ["/b", "/a"];
        assert_ne!(
            content_hash_json_slices(raw, &forward).unwrap(),
            content_hash_json_slices(raw, &reversed).unwrap(),
        );
    }

    #[test]
    fn severity_for_covers_known_pairs() {
        assert!(matches!(severity_for("mcp", "added"), Severity::Critical));
        assert!(matches!(severity_for("mcp", "modified"), Severity::High));
        assert!(matches!(severity_for("rules", "removed"), Severity::Info));
        assert!(matches!(severity_for("unknown", "added"), Severity::Low));
    }

    #[test]
    fn change_kind_type_suffix_matches_namespace() {
        assert_eq!(ChangeKind::Snapshot.type_suffix(), "snapshot");
        assert_eq!(ChangeKind::Added.type_suffix(), "added");
        assert_eq!(ChangeKind::Modified.type_suffix(), "modified");
        assert_eq!(ChangeKind::Removed.type_suffix(), "removed");
    }

    #[test]
    fn mcp_payload_omits_json_blob_to_match_platform_shape() {
        // The legacy MCP shape (`{kind: "mcp", json: <whole file>}`) is gone
        // — fan-out callers emit per-server envelopes shaped after
        // `ParsedMCPArtifact`. The non-fan-out fallback emits a minimal
        // degraded shape with `server_name` derived from the filename so
        // ingest never 400s on a missing field.
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        let raw = r#"{"mcpServers":{"a":{"command":"x"}}}"#;
        let path = Path::new("/etc/claude/mcp.json");
        let payload =
            filtered_payload(raw, "mcp", Some(ConfigScope::Personal), path, &filter, &cfg);
        assert!(
            payload.get("kind").is_none(),
            "platform forbids extra `kind` field"
        );
        assert!(payload.get("json").is_none());
        assert!(payload.get("server_name").is_some());
        assert_eq!(payload["scope"], "personal");
    }

    #[test]
    fn hash_mcp_server_entry_distinguishes_servers_by_name() {
        let raw = r#"{"context7":{"command":"npx"},"github":{"command":"docker"}}"#;
        let parsed: Value = serde_json::from_str(raw).unwrap();
        let servers = parsed.as_object().unwrap();
        let h_ctx7 = hash_mcp_server_entry("context7", &servers["context7"]);
        let h_gh = hash_mcp_server_entry("github", &servers["github"]);
        assert_ne!(h_ctx7, h_gh);
    }

    #[test]
    fn mcp_fanout_emits_one_envelope_per_server_with_distinct_hashes() {
        let cache = ContentHashCache::new(64);
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        let raw = r#"{"context7":{"command":"npx"},"github":{"command":"docker"}}"#;
        let path = Path::new("/home/u/.claude/plugins/cache/x/y/z/.mcp.json");
        let ap = AgentPath {
            kind: "mcp".to_string(),
            scope: Some(ConfigScope::Personal),
            paths: Vec::new(),
            paths_relative: Vec::new(),
            paths_glob: None,
            paths_glob_relative: Vec::new(),
            json_slice_paths: Vec::new(),
            watch_strategy: WatchStrategy::Glob,
            slice_kind_subpath: true,
        };
        let events = build_event_from_content(
            path,
            &ap,
            "claude-code",
            ChangeKind::Snapshot,
            EventSource::InitScan,
            0,
            raw,
            &filter,
            &cache,
            &cfg,
        );
        assert_eq!(events.len(), 2);
        let mut subpaths: Vec<String> = events
            .iter()
            .map(|(_, _, s)| s.clone().expect("fan-out subpath set"))
            .collect();
        subpaths.sort();
        assert_eq!(subpaths, vec!["context7", "github"]);
        let path_hashes: HashSet<_> = events
            .iter()
            .map(|(e, _, _)| e.envelope["configpathhash"].as_str().unwrap().to_string())
            .collect();
        assert_eq!(path_hashes.len(), 2, "each server gets a unique path_hash");
        let cached = cache.entries_for_path(path);
        assert_eq!(cached.len(), 2);
        for (event, _, sub) in &events {
            let data = &event.envelope["data"];
            assert_eq!(data["server_name"].as_str(), sub.as_deref());
            assert_eq!(data["tools"].as_array().unwrap().len(), 0);
            assert_eq!(data["scope"], "personal");
        }
    }

    /// `~/.claude.json` is declared with `/mcpServers` + `/projects` slices and
    /// has no top-level `mcpServers` key. The whole-document fallback used to
    /// win there and turned all of Claude Code's preference keys into phantom
    /// MCP servers, replayed on every rewrite of that file.
    #[test]
    fn mcp_fanout_ignores_non_slice_keys_when_manifest_declares_pointers() {
        let raw = r#"{
            "numStartups": 42,
            "userID": "abc",
            "tipsHistory": {"x": 1},
            "projects": {
                "/repo/a": {"mcpServers": {"github": {"command": "docker"}}},
                "/repo/b": {"mcpServers": {}}
            }
        }"#;
        let parsed: Value = serde_json::from_str(raw).unwrap();

        let sliced = resolve_mcp_servers(&parsed, &["/mcpServers", "/projects"]);
        let mut names: Vec<&String> = sliced.keys().collect();
        names.sort();
        assert_eq!(names, vec!["/repo/a::github"]);

        let unsliced = resolve_mcp_servers(&parsed, &[]);
        assert_eq!(
            unsliced.len(),
            4,
            "with no declared slices the whole document is still the server map"
        );
    }

    #[test]
    fn mcp_fanout_reads_top_level_mcpservers_slice() {
        let raw = r#"{"mcpServers":{"context7":{"command":"npx"}},"userID":"abc"}"#;
        let parsed: Value = serde_json::from_str(raw).unwrap();
        let servers = resolve_mcp_servers(&parsed, &["/mcpServers", "/projects"]);
        assert_eq!(servers.keys().collect::<Vec<_>>(), vec!["context7"]);
    }

    #[test]
    fn rules_payload_matches_platform_shape() {
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        let raw = "# CLAUDE.md\n\nBe terse.";
        let path = Path::new("/repo/CLAUDE.md");
        let payload = filtered_payload(
            raw,
            "rules",
            Some(ConfigScope::Project),
            path,
            &filter,
            &cfg,
        );
        // Platform contract: ParsedRulesArtifact has path/body/frontmatter only.
        assert_eq!(payload["path"], "/repo/CLAUDE.md");
        assert_eq!(payload["body"], raw);
        assert!(payload["frontmatter"].is_object());
        assert!(
            payload.get("kind").is_none(),
            "extra `kind` field would 400 the envelope"
        );
        assert!(payload.get("text").is_none());
        assert!(payload.get("scope").is_none(), "scope only attaches to mcp");
    }

    #[test]
    fn skill_payload_derives_name_from_parent_dir_for_skill_md() {
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        let raw = "Skill body content.";
        // Upstream Claude Code spec — every skill lives at
        // `<root>/<skill-name>/SKILL.md`. The file stem is the literal
        // `SKILL`; the parent directory carries the actual slug.
        let path = Path::new("/home/u/.claude/skills/code-review/SKILL.md");
        let payload = filtered_payload(raw, "skill", None, path, &filter, &cfg);
        assert_eq!(payload["name"], "code-review");
        assert_eq!(payload["body"], raw);
        assert_eq!(payload["description"], "");
        assert!(payload["frontmatter"].is_object());
    }

    #[test]
    fn skill_payload_pulls_name_and_description_from_frontmatter() {
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        let raw =
            "---\nname: playwright-cli\ndescription: \"Browser automation skill\"\n---\n\n# Body";
        let path = Path::new("/home/u/.claude/skills/SKILL.md");
        let payload = filtered_payload(raw, "skill", None, path, &filter, &cfg);
        assert_eq!(payload["name"], "playwright-cli");
        assert_eq!(payload["description"], "Browser automation skill");
        let fm = payload["frontmatter"].as_object().unwrap();
        assert_eq!(fm["name"], "playwright-cli");
        assert_eq!(fm["description"], "Browser automation skill");
        let body = payload["body"].as_str().unwrap();
        assert!(body.starts_with("# Body"));
        assert!(!body.contains("---"));
    }

    #[test]
    fn command_payload_derives_name_from_file_stem() {
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        let raw = "Run the command.";
        let path = Path::new("/home/u/.claude/commands/security-review.md");
        let payload = filtered_payload(raw, "command", None, path, &filter, &cfg);
        assert_eq!(payload["name"], "security-review");
        assert_eq!(payload["body"], raw);
    }

    #[test]
    fn hooks_payload_lifts_hooks_slice_from_settings_json() {
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        let raw = r#"{
            "theme": "dark",
            "hooks": {"PreToolUse": [{"matcher": "*"}]}
        }"#;
        let path = Path::new("/home/u/.claude/settings.json");
        let payload = filtered_payload(raw, "hooks", None, path, &filter, &cfg);
        assert!(payload.get("kind").is_none());
        assert!(payload.get("text").is_none());
        let hooks = payload["hooks"].as_object().unwrap();
        assert!(hooks.contains_key("PreToolUse"));
        assert!(!hooks.contains_key("theme"));
    }

    #[test]
    fn hooks_payload_uses_root_when_no_hooks_slice() {
        let cfg = Config::defaults();
        let filter = PrivacyFilter::new(&[]);
        // Plugin-installed hooks.json variant — top-level dict has the
        // lifecycle event names directly (no /hooks wrapper).
        let raw = r#"{"SessionStart": [{"matcher": "*"}]}"#;
        let path = Path::new("/home/u/.claude/plugins/cache/x/y/z/hooks/hooks.json");
        let payload = filtered_payload(raw, "hooks", None, path, &filter, &cfg);
        let hooks = payload["hooks"].as_object().unwrap();
        assert!(hooks.contains_key("SessionStart"));
    }

    #[test]
    fn rules_truncated_payload_keeps_shape() {
        let mut cfg = Config::defaults();
        cfg.inventory_monitor.max_inline_content_bytes = 256;
        let filter = PrivacyFilter::new(&[]);
        let raw = "x".repeat(10_000);
        let path = Path::new("/repo/CLAUDE.md");
        let payload = filtered_payload(&raw, "rules", None, path, &filter, &cfg);
        // Shape must still be path/body/frontmatter — no kind/truncated/etc.
        assert!(payload.get("kind").is_none());
        assert!(payload.get("truncated").is_none());
        let body = payload["body"].as_str().expect("body is string");
        assert!(
            body.len() <= 256 + 128,
            "body should be capped near inline_cap"
        );
        assert!(
            body.contains("…<truncated"),
            "marker should appear in truncated body"
        );
    }

    #[test]
    fn build_modified_event_uses_agent_name_as_source() {
        let cfg = Config::defaults();
        let path = Path::new("/etc/claude/mcp.json");
        let content_hash = [0u8; 32];
        let path_hash = [0u8; 32];
        let event = build_modified_event(
            path,
            "mcp",
            "claude-code",
            &content_hash,
            &path_hash,
            ChangeKind::Modified,
            EventSource::FsWatcher,
            1,
            Severity::High,
            Value::Null,
            &cfg,
        );
        assert_eq!(event.envelope["source"].as_str(), Some("claude-code"));
        assert_eq!(event.envelope["configsource"].as_str(), Some("claude-code"));
    }

    #[test]
    fn build_removed_event_uses_agent_name_as_source() {
        let cfg = Config::defaults();
        let path = Path::new("/home/u/.cursor/mcp.json");
        let event = build_removed_event_for_subpath(
            path,
            None,
            "mcp",
            "cursor",
            EventSource::FsWatcher,
            2,
            Severity::Critical,
            &cfg,
        )
        .expect("removed event must be built");
        assert_eq!(event.envelope["source"].as_str(), Some("cursor"));
        assert_eq!(event.envelope["configsource"].as_str(), Some("cursor"));
    }
}