relay-knowledge 1.1.14

Graph-database-based knowledge graph project.
Documentation
use std::collections::BTreeSet;

use crate::{
    api::{
        FileIndexFreshnessCursor, FileIndexFreshnessDiagnostics, FileIndexFreshnessState,
        FileIndexLag,
    },
    domain::FreshnessPolicy,
    storage::{FileIndexDiagnostics, FileIndexRoot, FileIndexRootStatus},
};

pub(super) struct FileFreshnessContext<'a> {
    pub(super) file_index_enabled: bool,
    pub(super) configured_roots: &'a [FileIndexRoot],
    pub(super) diagnostics: &'a FileIndexDiagnostics,
    pub(super) freshness_policy: FreshnessPolicy,
    pub(super) source_scope: Option<String>,
    pub(super) root_id: Option<String>,
    pub(super) graph_version: u64,
    pub(super) query_degraded_reason: Option<String>,
    pub(super) returned_paths: &'a [String],
    pub(super) content_required: bool,
}

pub(super) fn file_freshness_diagnostics(
    context: FileFreshnessContext<'_>,
) -> FileIndexFreshnessDiagnostics {
    let selected = selected_root_statuses(
        context.configured_roots,
        context.diagnostics,
        context.source_scope.as_deref(),
        context.root_id.as_deref(),
    );
    let pending_roots = pending_configured_roots(
        context.configured_roots,
        context.diagnostics,
        context.source_scope.as_deref(),
        context.root_id.as_deref(),
    );
    let root_overflow = |status: &&FileIndexRootStatus| {
        status.truncated || (context.content_required && status.content_truncated)
    };
    let cursors = selected
        .iter()
        .map(|status| FileIndexFreshnessCursor {
            source_scope: status.scope_id.clone(),
            root_id: status.root_id.clone(),
            root_path: status.root_path.clone(),
            backend: "bounded_scan".to_owned(),
            scan_watermark_ms: status.last_indexed_at_ms,
            indexed_file_count: status.indexed_file_count,
            missing_file_count: status.missing_file_count,
            scan_error_count: status.scan_error_count,
            overflow: status.truncated || (context.content_required && status.content_truncated),
            last_error: status_last_error(status, context.content_required),
        })
        .collect::<Vec<_>>();
    let stale_root_count = selected
        .iter()
        .filter(|status| status.last_indexed_at_ms.is_none())
        .count()
        .saturating_add(pending_roots.len());
    let overflow_root_count = selected.iter().filter(root_overflow).count();
    let scan_error_count = selected
        .iter()
        .map(|status| status.scan_error_count)
        .sum::<usize>();
    let content_read_error_count = if context.content_required {
        selected
            .iter()
            .map(|status| status.content_read_error_count)
            .sum::<usize>()
    } else {
        0
    };
    let stale_content_cursor_count = if context.content_required {
        selected
            .iter()
            .map(|status| status.stale_content_cursor_count)
            .sum::<usize>()
    } else {
        0
    };
    let missing_file_count = selected
        .iter()
        .map(|status| status.missing_file_count)
        .sum::<usize>();
    let indexed_root_count = selected
        .iter()
        .filter(|status| status.last_indexed_at_ms.is_some())
        .count();
    let configured_root_count = selected.len().saturating_add(pending_roots.len());
    let degraded_reason = context.query_degraded_reason.or_else(|| {
        selected
            .iter()
            .find_map(|status| status_last_error(status, context.content_required))
    });
    let state = file_freshness_state(FileFreshnessStateInputs {
        enabled: context.file_index_enabled,
        configured_root_count,
        stale_root_count,
        stale_content_cursor_count,
        overflow_root_count,
        scan_error_count,
        content_read_error_count,
        degraded_reason: degraded_reason.as_ref(),
    });
    let stale_reason = stale_reason_for_state(
        state,
        stale_root_count,
        stale_content_cursor_count,
        overflow_root_count,
    );
    let direct_source_read_required = !matches!(
        state,
        FileIndexFreshnessState::Fresh | FileIndexFreshnessState::Paused
    );
    let bounded_rescan_required = matches!(
        state,
        FileIndexFreshnessState::Pending
            | FileIndexFreshnessState::Stale
            | FileIndexFreshnessState::Degraded
            | FileIndexFreshnessState::Overflow
    );
    let direct_source_read_paths = unique_paths(context.returned_paths);
    let content_read_model_cursors = content_cursors_for_selection(
        &context.diagnostics.content_cursors,
        context.source_scope.as_deref(),
        context.root_id.as_deref(),
        &direct_source_read_paths,
    );

    FileIndexFreshnessDiagnostics {
        state,
        freshness_policy: context.freshness_policy,
        graph_version: context.graph_version,
        source_scope: context.source_scope.clone(),
        root_id: context.root_id.clone(),
        stale_reason,
        degraded_reason,
        index_lag: FileIndexLag {
            configured_root_count,
            indexed_root_count,
            pending_root_count: pending_roots.len(),
            stale_root_count,
            overflow_root_count,
            missing_file_count,
            pending_task_count: 0,
        },
        cursors,
        direct_source_read_required,
        bounded_rescan_required,
        direct_source_read_paths,
        agent_instructions: agent_instructions(
            state,
            bounded_rescan_required,
            direct_source_read_required,
        ),
        content_read_model_cursors,
    }
}

fn selected_root_statuses(
    configured_roots: &[FileIndexRoot],
    diagnostics: &FileIndexDiagnostics,
    source_scope: Option<&str>,
    root_id: Option<&str>,
) -> Vec<FileIndexRootStatus> {
    let configured = configured_roots
        .iter()
        .map(|root| (root.scope_id.clone(), root.root_id.clone()))
        .collect::<BTreeSet<_>>();

    diagnostics
        .roots
        .iter()
        .filter(|status| {
            configured.contains(&(status.scope_id.clone(), status.root_id.clone()))
                && source_scope.is_none_or(|scope| status.scope_id == scope)
                && root_id.is_none_or(|root| status.root_id == root)
        })
        .cloned()
        .collect()
}

fn pending_configured_roots(
    configured_roots: &[FileIndexRoot],
    diagnostics: &FileIndexDiagnostics,
    source_scope: Option<&str>,
    root_id: Option<&str>,
) -> Vec<FileIndexRoot> {
    let known = diagnostics
        .roots
        .iter()
        .map(|status| (status.scope_id.clone(), status.root_id.clone()))
        .collect::<BTreeSet<_>>();

    configured_roots
        .iter()
        .filter(|root| {
            source_scope.is_none_or(|scope| root.scope_id == scope)
                && root_id.is_none_or(|filter| root.root_id == filter)
                && !known.contains(&(root.scope_id.clone(), root.root_id.clone()))
        })
        .cloned()
        .collect()
}

struct FileFreshnessStateInputs<'a> {
    enabled: bool,
    configured_root_count: usize,
    stale_root_count: usize,
    stale_content_cursor_count: usize,
    overflow_root_count: usize,
    scan_error_count: usize,
    content_read_error_count: usize,
    degraded_reason: Option<&'a String>,
}

fn file_freshness_state(inputs: FileFreshnessStateInputs<'_>) -> FileIndexFreshnessState {
    if !inputs.enabled && inputs.configured_root_count == 0 {
        FileIndexFreshnessState::Paused
    } else if inputs.overflow_root_count > 0 {
        FileIndexFreshnessState::Overflow
    } else if inputs.scan_error_count > 0
        || inputs.content_read_error_count > 0
        || inputs.degraded_reason.is_some()
    {
        FileIndexFreshnessState::Degraded
    } else if inputs.stale_root_count > 0 {
        FileIndexFreshnessState::Pending
    } else if inputs.stale_content_cursor_count > 0 || inputs.configured_root_count == 0 {
        FileIndexFreshnessState::Stale
    } else {
        FileIndexFreshnessState::Fresh
    }
}

fn status_last_error(status: &FileIndexRootStatus, content_required: bool) -> Option<String> {
    if !content_required && content_only_incomplete(status) {
        None
    } else {
        status.last_error.clone()
    }
}

fn content_only_incomplete(status: &FileIndexRootStatus) -> bool {
    (status.content_truncated || status.content_read_error_count > 0)
        && !status.truncated
        && status.scan_error_count == 0
}

fn stale_reason_for_state(
    state: FileIndexFreshnessState,
    stale_root_count: usize,
    stale_content_cursor_count: usize,
    overflow_root_count: usize,
) -> Option<String> {
    match state {
        FileIndexFreshnessState::Pending => Some(format!(
            "{stale_root_count} configured file-index root(s) have not completed a scan"
        )),
        FileIndexFreshnessState::Stale if stale_content_cursor_count > 0 => Some(format!(
            "{stale_content_cursor_count} file-content read-model cursor(s) are stale"
        )),
        FileIndexFreshnessState::Stale => Some("no matching file-index root is fresh".to_owned()),
        FileIndexFreshnessState::Overflow => Some(format!(
            "{overflow_root_count} file-index root scan(s) overflowed the bounded scan budget"
        )),
        FileIndexFreshnessState::Degraded => {
            Some("file-index root scan or query is degraded".to_owned())
        }
        FileIndexFreshnessState::Paused | FileIndexFreshnessState::Fresh => None,
    }
}

fn unique_paths(paths: &[String]) -> Vec<String> {
    paths
        .iter()
        .cloned()
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect()
}

fn content_cursors_for_selection(
    cursors: &[crate::storage::FileContentReadModelCursor],
    source_scope: Option<&str>,
    root_id: Option<&str>,
    returned_paths: &[String],
) -> Vec<crate::storage::FileContentReadModelCursor> {
    if returned_paths.is_empty() {
        return Vec::new();
    }
    let returned_paths = returned_paths.iter().collect::<BTreeSet<_>>();
    cursors
        .iter()
        .filter(|cursor| {
            source_scope.is_none_or(|scope| cursor.source_scope == scope)
                && root_id.is_none_or(|filter| cursor.root_id == filter)
                && returned_paths.contains(&cursor.path)
        })
        .cloned()
        .collect()
}

fn agent_instructions(
    state: FileIndexFreshnessState,
    bounded_rescan_required: bool,
    direct_source_read_required: bool,
) -> Vec<String> {
    let mut instructions = Vec::new();
    if bounded_rescan_required {
        instructions.push(
            "Run a bounded file index scan before trusting local file-index freshness.".to_owned(),
        );
    }
    if direct_source_read_required {
        instructions.push(format!(
            "File-index state is {}; read direct source paths before editing or citing changed files.",
            state_label(state)
        ));
    }

    instructions
}

fn state_label(state: FileIndexFreshnessState) -> &'static str {
    match state {
        FileIndexFreshnessState::Fresh => "fresh",
        FileIndexFreshnessState::Pending => "pending",
        FileIndexFreshnessState::Paused => "paused",
        FileIndexFreshnessState::Stale => "stale",
        FileIndexFreshnessState::Degraded => "degraded",
        FileIndexFreshnessState::Overflow => "overflow",
    }
}

#[cfg(test)]
#[path = "mod_tests.rs"]
mod tests;