keyhog 0.5.73

use crate::args::ScanArgs;
use anyhow::Result;
use std::path::PathBuf;

/// Hard ceiling on the worker thread count. Requested values above this are
/// clamped by [`sanitise_thread_count`]; spawning thousands of threads thrashes
/// the OS scheduler without speeding the scan up.
pub(crate) const MAX_THREADS_CAP: usize = 256;

/// KeyHog workers use the platform-standard Rust stack reservation. Scanner
/// parsing and traversal are iterative; reserving 8 MiB per worker multiplied
/// address-space and allocator commitment without a corresponding call depth.
pub(crate) const KEYHOG_WORKER_STACK_BYTES: usize = 2 * 1024 * 1024;
/// Persistent daemons trade a small amount of single-request parallelism for a
/// bounded compile cache and worker-local scratch footprint.
pub(crate) const PERSISTENT_DAEMON_WORKER_CAP: usize = 8;

/// Documented conventional ML threshold value.
///
/// `ScanArgs::ml_threshold` is optional so the runtime can distinguish an
/// absent flag/config from an explicit `0.5`.
pub(crate) const ML_THRESHOLD_DEFAULT: f64 = 0.5;
pub(crate) const VERIFY_TIMEOUT_DEFAULT_SECS: u64 = 5;
pub(crate) const VERIFY_MAX_CONCURRENT_DEFAULT: usize = 5;
#[cfg(feature = "git")]
pub(crate) const MAX_COMMITS_DEFAULT: usize = 1000;

static CONFIGURED_RAYON_THREADS: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
static RAYON_CONFIGURATION_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

/// Maximum chunk count for one fused filesystem read+scan batch.
///
/// The byte ceiling remains authoritative. A 1 MiB window still dispatches
/// alone, while 4 KiB files coalesce into 256-chunk work units instead of
/// paying scheduler and phase-gate setup every 32 files.
pub(crate) const FUSED_BATCH_DEFAULT: usize = 1024;

/// Byte ceiling on one fused filesystem batch, applied alongside
/// [`FUSED_BATCH_DEFAULT`].
///
/// The fused consumer executes one batch at a time while the source-reader
/// boundary remains a rendezvous. This bounds payload residency to 1 MiB while
/// allowing tiny files to amortize one scanner dispatch.
///
/// Compile-time rather than configurable on purpose: it is hashed into the
/// autoroute identity, so a change here invalidates persisted calibration
/// instead of replaying decisions measured under different batching.
pub(crate) const FUSED_BATCH_BYTES: usize = 1024 * 1024;

/// Count representatives covering both edges of every logarithmic autoroute
/// bucket up to the fused-batch ceiling.
pub(crate) fn fused_batch_calibration_counts() -> Vec<usize> {
    let mut counts = vec![1];
    let mut lower = 2usize;
    while lower <= FUSED_BATCH_DEFAULT {
        counts.push(lower);
        let upper = lower
            .saturating_mul(2)
            .saturating_sub(1)
            .min(FUSED_BATCH_DEFAULT);
        if upper != lower {
            counts.push(upper);
        }
        let Some(next) = lower.checked_mul(2) else {
            break;
        };
        lower = next;
    }
    counts
}

/// Default rendezvous depth for fused filesystem batches. The producer may
/// finish one batch while the consumer scans the active batch, but no third
/// batch remains queued and resident.
pub(crate) fn fused_depth_default(_worker_threads: usize) -> usize {
    0
}

/// Bound explicit CPU/SIMD batch waves independently of the Rayon pool width.
/// A filesystem batch retains at most 1 MiB of payload, so four live scan
/// batches cap the consumer wave at 4 MiB while preserving parallel routing.
pub(crate) fn fused_cpu_wave_width(worker_threads: usize) -> usize {
    worker_threads.clamp(1, 4)
}

pub(crate) fn parse_backend_override(
    raw: Option<&str>,
) -> Result<Option<keyhog_scanner::ScanBackend>> {
    let Some(raw) = raw else {
        return Ok(None);
    };
    let trimmed = raw.trim();
    if trimmed.is_empty() || trimmed.eq_ignore_ascii_case("auto") {
        return Ok(None);
    }
    let operator_value = keyhog_scanner::hw_probe::BACKEND_OVERRIDE_VALUES
        .iter()
        .copied()
        .find(|value| !value.eq_ignore_ascii_case("auto") && value.eq_ignore_ascii_case(trimmed));
    operator_value
        .and_then(keyhog_scanner::hw_probe::parse_backend_str)
        .map(Some)
        .ok_or_else(|| {
            let supported = keyhog_scanner::hw_probe::BACKEND_OVERRIDE_VALUES.join(", ");
            anyhow::anyhow!(
                "invalid --backend value {:?}. Supported values: {supported}.",
                raw
            )
        })
}

pub(crate) fn backend_override_label(backend: Option<keyhog_scanner::ScanBackend>) -> &'static str {
    backend.map_or("auto", keyhog_scanner::ScanBackend::label)
}

/// Canonical value accepted by the public `--backend` parser for a resolved
/// backend. The engine's diagnostic label for the scalar CPU implementation is
/// `cpu-fallback`, while the stable operator spelling is `cpu`.
pub(crate) fn backend_override_cli_value(backend: keyhog_scanner::ScanBackend) -> &'static str {
    keyhog_scanner::execution_pack::ExecutionPackBackend::from_scan_backend(backend)
        .map_or_else(|| backend.label(), |backend| backend.lowercase_name())
}

pub(crate) fn gpu_runtime_policy_from_args(
    args: &ScanArgs,
) -> keyhog_scanner::gpu::GpuRuntimePolicy {
    if args.require_gpu || explicit_gpu_backend(args) {
        keyhog_scanner::gpu::GpuRuntimePolicy::Required
    } else if args.no_gpu || explicit_cpu_backend(args) {
        keyhog_scanner::gpu::GpuRuntimePolicy::Disabled
    } else {
        keyhog_scanner::gpu::GpuRuntimePolicy::Auto
    }
}

pub(crate) fn gpu_runtime_policy_for_backend_override(
    backend: Option<keyhog_scanner::ScanBackend>,
) -> Result<keyhog_scanner::gpu::GpuRuntimePolicy> {
    let policy = match backend {
        Some(
            keyhog_scanner::ScanBackend::GpuCuda
            | keyhog_scanner::ScanBackend::GpuMetal
            | keyhog_scanner::ScanBackend::GpuWgpu,
        ) => keyhog_scanner::gpu::GpuRuntimePolicy::Required,
        Some(keyhog_scanner::ScanBackend::SimdCpu | keyhog_scanner::ScanBackend::CpuFallback) => {
            keyhog_scanner::gpu::GpuRuntimePolicy::Disabled
        }
        None => keyhog_scanner::gpu::GpuRuntimePolicy::Auto,
        Some(backend) => anyhow::bail!(
            "daemon GPU runtime policy is undefined for backend {}; update the daemon policy mapping",
            backend.label()
        ),
    };
    Ok(policy)
}

/// True when the operator explicitly selected a CPU-only backend
/// (`--backend cpu`/`--backend simd`). Such a scan never acquires the GPU, so
/// the resolved policy is `Disabled`: this keeps `gpu_probe()` from creating a
/// wgpu/Vulkan instance the scan would never use. Beyond skipping a pointless
/// (and slow) Vulkan init on the CPU path (Law 7), it prevents a real crash
/// the probe spawns a mesa driver worker thread that SIGSEGVs during teardown
/// if the process exits fast on an early error (expired `.keyhogignore`,
/// missing scan path) before the driver finishes initialising, turning a clean
/// fail-closed `exit(2)` into a signal death (exit 139). `auto` (no explicit
/// `--backend`) is intentionally NOT treated as CPU-only: autoroute legitimately
/// probes to choose a backend.
fn explicit_cpu_backend(args: &ScanArgs) -> bool {
    args.backend
        .as_deref()
        .and_then(keyhog_scanner::hw_probe::parse_backend_str)
        .is_some_and(|backend| !backend.is_gpu())
}

fn explicit_gpu_backend(args: &ScanArgs) -> bool {
    args.backend
        .as_deref()
        .and_then(keyhog_scanner::hw_probe::parse_backend_str)
        .is_some_and(keyhog_scanner::ScanBackend::is_gpu)
}

#[derive(Debug, Clone)]
pub(crate) struct ScanRuntimeInput {
    pub(crate) cache_dir: Option<PathBuf>,
    pub(crate) autoroute_cache: Option<String>,
    pub(crate) matcher_cache: Option<String>,
    pub(crate) calibration_cache: Option<PathBuf>,
    pub(crate) backend: Option<String>,
    pub(crate) batch_pipeline: bool,
    pub(crate) threads: Option<usize>,
    pub(crate) reader_threads: Option<usize>,
    pub(crate) fused_batch: usize,
    pub(crate) fused_depth: Option<usize>,
    pub(crate) gpu_runtime_policy: keyhog_scanner::gpu::GpuRuntimePolicy,
    pub(crate) autoroute_gpu: bool,
    pub(crate) autoroute_calibration: bool,
    pub(crate) regex_dfa_limit: Option<usize>,
    pub(crate) gpu_batch_input_limit: Option<usize>,
    pub(crate) max_file_size: Option<usize>,
    #[cfg(feature = "git")]
    pub(crate) max_commits: usize,
    pub(crate) no_default_excludes: bool,
    pub(crate) exclude_paths: Vec<String>,
    pub(crate) incremental: bool,
    pub(crate) incremental_cache_path: Option<PathBuf>,
    pub(crate) source_limits: keyhog_sources::SourceLimits,
}

impl ScanRuntimeInput {
    pub(crate) fn from_scan_args(args: &ScanArgs) -> Self {
        Self {
            cache_dir: args.cache_dir.clone(),
            autoroute_cache: args.autoroute_cache.clone(),
            matcher_cache: args.matcher_cache.clone(),
            calibration_cache: args.calibration_cache.clone(),
            backend: args.backend.clone(),
            batch_pipeline: args.batch_pipeline && !args.no_batch_pipeline,
            threads: args.threads,
            reader_threads: args.reader_threads,
            fused_batch: args.fused_batch.unwrap_or(FUSED_BATCH_DEFAULT), // LAW10: absent fused-batch config => documented compiled throughput default; no recall path changes and the value is printed/hashes into autoroute identity
            fused_depth: args.fused_depth,
            gpu_runtime_policy: gpu_runtime_policy_from_args(args),
            autoroute_gpu: args.autoroute_gpu && !args.no_autoroute_gpu,
            autoroute_calibration: args.autoroute_calibrate,
            regex_dfa_limit: args.regex_dfa_limit,
            gpu_batch_input_limit: args.gpu_batch_input_limit,
            max_file_size: args.max_file_size,
            #[cfg(feature = "git")]
            max_commits: args.max_commits.unwrap_or(MAX_COMMITS_DEFAULT), // LAW10: absent max-commits => documented compiled git traversal cap; effective config prints the concrete value and source construction consumes this resolved field
            no_default_excludes: args.no_default_excludes,
            exclude_paths: match &args.exclude_paths {
                Some(paths) => paths.clone(),
                None => Vec::new(),
            },
            incremental: args.incremental,
            incremental_cache_path: args.incremental_cache.clone(),
            source_limits: args.limits.to_source_limits(),
        }
    }
}

/// The scan worker width KeyHog would configure, WITHOUT creating Rayon's
/// global registry.
///
/// Daemon clients compute this value only to bind their expected runtime
/// identity. Using `rayon::current_num_threads()` there would create Rayon's
/// default logical-core pool before KeyHog can install its bounded
/// physical-core pool. That both changes the identity and leaves twice as many
/// long-lived worker stacks on SMT hosts.
///
/// An already configured KeyHog pool remains authoritative. Otherwise this
/// returns the bounded persistent-daemon width used by daemon startup.
pub(crate) fn keyhog_worker_threads() -> usize {
    if let Some(configured) = CONFIGURED_RAYON_THREADS.get().copied() {
        return configured;
    }
    persistent_daemon_worker_width(keyhog_scanner::hw_probe::probe_host_hardware().physical_cores)
}

fn persistent_daemon_worker_width(physical_cores: usize) -> usize {
    physical_cores.clamp(1, PERSISTENT_DAEMON_WORKER_CAP)
}

pub(crate) fn configure_threads(threads: Option<usize>, physical_cores: usize) -> Result<usize> {
    // Resolution order: --threads / [scan].threads > physical core count.
    // Physical cores are the right default for CPU-bound regex: SMT siblings
    // share execution units, so doubling threads mostly doubles cache pressure.
    let (n, source) = if let Some(t) = threads {
        (
            sanitise_thread_count(t, physical_cores, "cli-arg"),
            "cli-arg",
        )
    } else {
        (physical_cores.max(1), "physical-cores")
    };
    configure_resolved_threads(n, source, physical_cores)
}

pub(crate) fn configure_persistent_daemon_threads(physical_cores: usize) -> Result<usize> {
    configure_resolved_threads(
        persistent_daemon_worker_width(physical_cores),
        "persistent-daemon",
        physical_cores,
    )
}

fn configure_resolved_threads(
    n: usize,
    source: &'static str,
    physical_cores: usize,
) -> Result<usize> {
    let _configuration_guard = RAYON_CONFIGURATION_LOCK
        .lock()
        .map_err(|error| anyhow::anyhow!("Rayon configuration lock was poisoned: {error}"))?;
    if !thread_pool_needs_initialization(CONFIGURED_RAYON_THREADS.get().copied(), n, source)? {
        tracing::debug!(
            threads = n,
            source,
            "rayon thread pool already has the requested width"
        );
        return Ok(n);
    }

    let builder = rayon::ThreadPoolBuilder::new()
        .num_threads(n)
        .stack_size(KEYHOG_WORKER_STACK_BYTES)
        .thread_name(|i| format!("keyhog-worker-{i}"));

    require_keyhog_owned_rayon_pool(
        builder.build_global(),
        n,
        source,
        rayon::current_num_threads,
    )?;
    CONFIGURED_RAYON_THREADS.set(n).map_err(|_| {
        anyhow::anyhow!(
            "Rayon worker pool configured with {n} threads, but its KeyHog initialization state changed concurrently"
        )
    })?;
    tracing::info!(
        threads = n,
        source,
        physical_cores,
        "rayon thread pool configured"
    );
    Ok(n)
}

fn require_keyhog_owned_rayon_pool<E: std::fmt::Display>(
    build_result: std::result::Result<(), E>,
    requested: usize,
    source: &'static str,
    actual_threads: impl FnOnce() -> usize,
) -> Result<()> {
    build_result.map_err(|error| {
        let actual = actual_threads();
        anyhow::anyhow!(
            "Rayon worker pool was initialized outside KeyHog with {actual} threads, but this scan requires a KeyHog-owned pool with {requested} threads ({source}) and 2 MiB worker stacks ({error}). Fix: configure KeyHog before any library initializes Rayon's global pool or start this scan in a separate process"
        )
    })
}

fn thread_pool_needs_initialization(
    configured: Option<usize>,
    requested: usize,
    source: &'static str,
) -> Result<bool> {
    match configured {
        None => Ok(true),
        Some(actual) if actual == requested => Ok(false),
        Some(actual) => anyhow::bail!(
            "Rayon worker pool already has {actual} threads, but this scan requested {requested} ({source}); the requested width cannot take effect in this process. Fix: use one thread width for every in-process scan policy or start the incompatible policy in a separate process"
        ),
    }
}

pub(crate) fn configure_hyperscan_cache_dir(cache_dir: Option<PathBuf>) -> Result<()> {
    if let Some(path) = cache_dir.as_ref() {
        if !path.is_absolute() {
            anyhow::bail!(
                "Hyperscan cache dir '{}' must be absolute. Fix: pass an absolute path under \
                 your home directory or the per-user keyhog temp cache root.",
                path.display()
            );
        }
    }

    #[cfg(feature = "simd")]
    {
        if let Some(path) = cache_dir.as_ref() {
            keyhog_scanner::validate_hyperscan_cache_dir(path).map_err(|error| {
                anyhow::anyhow!("{error}. Configure with --cache-dir or [system].cache_dir")
            })?;
        }
        keyhog_scanner::set_hyperscan_cache_dir(cache_dir);
    }

    #[cfg(not(feature = "simd"))]
    {
        if cache_dir.is_some() {
            anyhow::bail!(
                "--cache-dir / [system].cache_dir requires a keyhog build with the simd \
                 feature; this binary has no Hyperscan cache to configure"
            );
        }
    }

    Ok(())
}

pub(crate) fn configure_matcher_artifact_cache_dir(cache_dir: Option<PathBuf>) -> Result<()> {
    if let Some(path) = cache_dir.as_ref() {
        keyhog_scanner::validate_matcher_artifact_cache_dir(path).map_err(|error| {
            anyhow::anyhow!("{error}. Configure with --matcher-cache or [system].matcher_cache")
        })?;
    }
    keyhog_scanner::set_matcher_artifact_cache_dir(cache_dir);
    Ok(())
}

/// Clamp a user-supplied thread count to a sane range. Logs a warning when the
/// value was outside the accepted bounds so an operator sees what was used.
fn sanitise_thread_count(requested: usize, physical_cores: usize, source: &'static str) -> usize {
    let safe_default = physical_cores.max(1);
    if requested == 0 {
        eprintln!(
            "keyhog: invalid {source} thread count 0; expected an integer >= 1; using {safe_default}"
        );
        tracing::warn!(
            source,
            requested = 0,
            using = safe_default,
            "thread count of 0 is not meaningful; falling back to physical-cores"
        );
        return safe_default;
    }
    if requested > MAX_THREADS_CAP {
        eprintln!(
            "keyhog: {source} thread count {requested} exceeds cap {MAX_THREADS_CAP}; using {MAX_THREADS_CAP}"
        );
        tracing::warn!(
            source,
            requested,
            cap = MAX_THREADS_CAP,
            "requested thread count exceeds cap; clamping"
        );
        return MAX_THREADS_CAP;
    }
    requested
}

#[doc(hidden)]
pub(crate) mod testing {
    pub(crate) fn sanitise_thread_count(
        requested: usize,
        physical_cores: usize,
        source: &'static str,
    ) -> usize {
        super::sanitise_thread_count(requested, physical_cores, source)
    }
}

// Sibling file (orchestrator_config/runtime_tests.rs), not runtime/ subdir.
#[path = "runtime_tests.rs"]
mod runtime_tests;