orchestrata_ultimate.knowledge_graph.anomaly_detector

Reference › orchestrata_ultimate › knowledge_graph › anomaly_detector

Detects and corrects drift between the filesystem and the operational graph (orchkg on orchgraph); it implements Spec 008 FR-010 (REQ-008-025). AnomalyType names the drift kinds: stale file, missing file, hash mismatch, entity drift, orphan relationship and line-count drift. Anomaly records the type, path, description and whether it is auto_correctable; CorrectionResult records whether a fix was applied and what was done; AnomalyReport bundles anomalies and corrections.

AnomalyDetector is built from a graph_client GraphClient, anything matching SynchronizerProtocol from interfaces, a project id and an optional event loop. detect compares the project's File nodes with the files on disk (_check_single_file, _check_hash and related checks). correct dispatches through a table of handlers keyed by anomaly type and returns an uncorrected result with the action "no handler for anomaly type" when none exists. audit is the full cycle: with mode report it only detects, while auto and interactive also apply corrections and return the report. Corrections reuse the synchronizer's sync_file. The synchronizer is the only importer.

At a glance

Class diagram

Dependency diagram

API

Classes

AnomalyType (enum)

Bases: StrEnum · Decorators: traces(spec='008', story='FR-010', reqs=['REQ-008-025']) · Line 28

Types of graph-vs-filesystem anomalies.

Anomaly (dataclass)

Decorators: dataclass, traces(spec='008', story='FR-010', reqs=['REQ-008-025']) · Line 44

A single detected anomaly between graph and filesystem.

CorrectionResult (dataclass)

Decorators: dataclass, traces(spec='008', story='FR-010', reqs=['REQ-008-025']) · Line 56

Result of attempting to correct an anomaly.

AnomalyReport (dataclass)

Decorators: dataclass, traces(spec='008', story='FR-010', reqs=['REQ-008-025']) · Line 66

Full audit report with anomalies and corrections.

AnomalyDetector

Decorators: traces(spec='008', story='FR-010', reqs=['REQ-008-025']) · Line 78

Detects and corrects graph-vs-filesystem inconsistencies.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/anomaly_detector.py

# mypy: ignore-errors
"""Anomaly detection and correction for the knowledge graph.

Detects inconsistencies between the filesystem and the
knowledge graph, then applies corrections to restore consistency.
Implements Spec 008 FR-010 (Anomaly Correction).
"""

import asyncio
import inspect
import logging
from dataclasses import dataclass, field
from enum import StrEnum
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient
from orchestrata_ultimate.knowledge_graph.interfaces import SynchronizerProtocol

logger = logging.getLogger(__name__)


# ── Anomaly Types ────────────────────────────────────────────


@traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
class AnomalyType(StrEnum):
    """Types of graph-vs-filesystem anomalies."""

    STALE_FILE = "stale_file"
    MISSING_FILE = "missing_file"
    HASH_MISMATCH = "hash_mismatch"
    ENTITY_DRIFT = "entity_drift"
    ORPHAN_RELATIONSHIP = "orphan_relationship"
    LINE_COUNT_DRIFT = "line_count_drift"


# ── Data Models ──────────────────────────────────────────────


@dataclass
@traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
class Anomaly:
    """A single detected anomaly between graph and filesystem."""

    anomaly_type: AnomalyType
    file_path: str
    description: str
    auto_correctable: bool
    details: dict[str, Any] = field(default_factory=dict)


@dataclass
@traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
class CorrectionResult:
    """Result of attempting to correct an anomaly."""

    anomaly: Anomaly
    corrected: bool
    action_taken: str


@dataclass
@traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
class AnomalyReport:
    """Full audit report with anomalies and corrections."""

    anomalies: list[Anomaly] = field(default_factory=list)
    corrections: list[CorrectionResult] = field(default_factory=list)
    uncorrectable: list[Anomaly] = field(default_factory=list)


# ── Anomaly Detector ─────────────────────────────────────────


@traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
class AnomalyDetector:
    """Detects and corrects graph-vs-filesystem inconsistencies.

    Compares File nodes in the graph against actual filesystem state,
    re-parses files to detect entity drift and orphan relationships,
    and applies corrections when requested.
    """

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-059"])
    def __init__(
        self,
        graph_client: GraphClient,
        synchronizer: SynchronizerProtocol,
        project_id: str,
        event_loop: asyncio.AbstractEventLoop | None = None,
    ) -> None:
        self._client = graph_client
        self._synchronizer = synchronizer
        self._project_id = project_id
        self._parser = synchronizer._parser
        self._loop = event_loop

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-061"])
    async def connect(self) -> None:
        """Ensure the underlying graph client is connected."""
        if hasattr(self._client, "connect") and not self._client.connected:
            await self._client.connect()

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-070"])
    def _get_loop(self) -> asyncio.AbstractEventLoop:
        """Return the event loop for async operations.

        Uses the explicitly provided loop if available,
        otherwise falls back to creating a new loop.
        """
        if self._loop is not None and not self._loop.is_closed():
            return self._loop
        return _get_or_create_loop()

    @traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
    async def detect(self, project_path: str) -> list[Anomaly]:
        """Detect all anomalies for a project.

        Compares graph File nodes against the filesystem, checks
        hashes, line counts, and entity presence in the AST.

        Args:
            project_path: Root directory of the project.

        Returns:
            List of detected Anomaly objects.
        """
        graph_files = await self._client.get_project_files(self._project_id)
        anomalies: list[Anomaly] = []
        seen_paths: set[str] = set()
        for gf in graph_files:
            path = gf.get("path", "")
            seen_paths.add(path)
            anomalies.extend(await self._check_single_file(gf, project_path))
        anomalies.extend(self._find_missing_files(project_path, seen_paths))
        return anomalies

    @traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
    async def correct(self, anomaly: Anomaly) -> CorrectionResult:
        """Apply correction for a single anomaly.

        Dispatches to the appropriate correction strategy based
        on the anomaly type.

        Args:
            anomaly: The anomaly to correct.

        Returns:
            CorrectionResult describing what was done.
        """
        handler = _CORRECTION_MAP.get(anomaly.anomaly_type)
        if handler is None:
            return CorrectionResult(
                anomaly=anomaly,
                corrected=False,
                action_taken="no handler for anomaly type",
            )
        result = handler(self, anomaly)
        if inspect.isawaitable(result):
            return await result
        return result

    @traces(spec="008", story="FR-010", reqs=["REQ-008-025"])
    async def audit(self, project_path: str, mode: str = "report") -> AnomalyReport:
        """Full audit cycle: detect and optionally correct.

        Args:
            project_path: Root directory to audit.
            mode: 'report' (detect only), 'auto' (detect + fix),
                  'interactive' (detect + prompt per anomaly).

        Returns:
            AnomalyReport with anomalies and corrections.
        """
        anomalies = await self.detect(project_path)
        report = AnomalyReport(anomalies=anomalies)
        if mode == "report":
            return report
        return await self._apply_corrections(anomalies, report, mode)

    # ── Detection Helpers ────────────────────────────────────

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-076"])
    async def _check_single_file(
        self,
        graph_file: dict[str, Any],
        project_path: str,
    ) -> list[Anomaly]:
        """Check a single graph File node against filesystem.

        Args:
            graph_file: Dict with path, hash, line_count, id.
            project_path: Project root directory.

        Returns:
            List of anomalies found for this file.
        """
        anomalies: list[Anomaly] = []
        path = graph_file.get("path", "")
        if not Path(path).exists():
            anomalies.append(
                Anomaly(
                    anomaly_type=AnomalyType.STALE_FILE,
                    file_path=path,
                    description=f"File missing from disk: {path}",
                    auto_correctable=True,
                )
            )
            return anomalies
        anomalies.extend(self._check_hash(graph_file))
        anomalies.extend(self._check_line_count(graph_file))
        anomalies.extend(await self._check_entities(path))
        return anomalies

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-077"])
    def _check_hash(self, graph_file: dict[str, Any]) -> list[Anomaly]:
        """Check if file hash matches graph record."""
        path = graph_file.get("path", "")
        graph_hash = graph_file.get("hash", "")
        disk_hash = self._parser.get_file_hash(path)
        if graph_hash and disk_hash != graph_hash:
            return [
                Anomaly(
                    anomaly_type=AnomalyType.HASH_MISMATCH,
                    file_path=path,
                    description=(
                        f"Hash mismatch: graph={graph_hash[:12]}… "
                        f"disk={disk_hash[:12]}…"
                    ),
                    auto_correctable=True,
                    details={
                        "graph_hash": graph_hash,
                        "disk_hash": disk_hash,
                    },
                )
            ]
        return []

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-078"])
    def _check_line_count(self, graph_file: dict[str, Any]) -> list[Anomaly]:
        """Check if file line count matches graph record."""
        path = graph_file.get("path", "")
        graph_lines = graph_file.get("line_count", 0)
        try:
            content = Path(path).read_text(encoding="utf-8", errors="replace")
            disk_lines = content.count("\n") + 1
        except OSError:
            return []
        if graph_lines and disk_lines != graph_lines:
            return [
                Anomaly(
                    anomaly_type=AnomalyType.LINE_COUNT_DRIFT,
                    file_path=path,
                    description=(f"Line count: graph={graph_lines} disk={disk_lines}"),
                    auto_correctable=True,
                    details={
                        "graph_line_count": graph_lines,
                        "disk_line_count": disk_lines,
                    },
                )
            ]
        return []

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-079"])
    async def _check_entities(self, file_path: str) -> list[Anomaly]:
        """Check graph entities against file AST.

        Detects ENTITY_DRIFT (class/function in graph not in
        AST) and ORPHAN_RELATIONSHIP (DEFINES target absent).
        """
        graph_entities = await self._client.get_file_entities(
            file_path, self._project_id
        )
        analysis = self._parser.parse_file(file_path)
        if not analysis.success:
            return []
        ast_class_names = {c.name for c in analysis.classes}
        ast_func_names = {f.name for f in analysis.functions}
        anomalies: list[Anomaly] = []
        anomalies.extend(
            self._check_entity_kind(
                file_path,
                graph_entities.get("classes", []),
                ast_class_names,
                "class",
            )
        )
        anomalies.extend(
            self._check_entity_kind(
                file_path,
                graph_entities.get("functions", []),
                ast_func_names,
                "function",
            )
        )
        return anomalies

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-080"])
    def _check_entity_kind(
        self,
        file_path: str,
        graph_items: list[dict[str, Any]],
        ast_names: set[str],
        kind: str,
    ) -> list[Anomaly]:
        """Check one kind of entity for drift or orphans."""
        anomalies: list[Anomaly] = []
        for item in graph_items:
            name = item.get("name", "")
            if name not in ast_names:
                anomalies.append(
                    Anomaly(
                        anomaly_type=AnomalyType.ORPHAN_RELATIONSHIP,
                        file_path=file_path,
                        description=(f"Orphan {kind} '{name}' in graph but not in AST"),
                        auto_correctable=True,
                        details={
                            "entity_name": name,
                            "entity_id": item.get("id", ""),
                            "entity_kind": kind,
                        },
                    )
                )
        return anomalies

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-081"])
    def _find_missing_files(
        self, project_path: str, graph_paths: set[str]
    ) -> list[Anomaly]:
        """Find filesystem files not in the graph.

        Walks the project directory for supported extensions,
        filtering by SyncConfig ignore patterns.
        """
        anomalies: list[Anomaly] = []
        project = Path(project_path)
        if not project.is_dir():
            return anomalies
        config = self._synchronizer._config
        ignore = set(config.ignore_patterns)
        exts = {".py", ".pl"}
        for source_file in sorted(project.rglob("*")):
            if source_file.suffix not in exts:
                continue
            if any(part in ignore for part in source_file.parts):
                continue
            fpath = str(source_file)
            if fpath not in graph_paths:
                anomalies.append(
                    Anomaly(
                        anomaly_type=AnomalyType.MISSING_FILE,
                        file_path=fpath,
                        description=(f"File on disk not in graph: {fpath}"),
                        auto_correctable=True,
                    )
                )
        return anomalies

    # ── Correction Helpers ───────────────────────────────────

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-082"])
    async def _correct_stale_file(self, anomaly: Anomaly) -> CorrectionResult:
        """Correct a STALE_FILE by deleting from graph."""
        deleted = await self._client.delete_file_cascade(
            anomaly.file_path, self._project_id
        )
        return CorrectionResult(
            anomaly=anomaly,
            corrected=deleted > 0,
            action_taken=(f"Deleted {deleted} nodes for {anomaly.file_path}"),
        )

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-087"])
    def _correct_missing_file(self, anomaly: Anomaly) -> CorrectionResult:
        """Correct a MISSING_FILE by syncing to graph."""
        result = self._synchronizer.sync_file(anomaly.file_path)
        return CorrectionResult(
            anomaly=anomaly,
            corrected=result.status.value == "completed",
            action_taken=f"Synced {anomaly.file_path}",
        )

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-088"])
    def _correct_hash_mismatch(self, anomaly: Anomaly) -> CorrectionResult:
        """Correct a HASH_MISMATCH by re-syncing the file."""
        self._synchronizer._file_hashes.pop(anomaly.file_path, None)
        result = self._synchronizer.sync_file(anomaly.file_path)
        return CorrectionResult(
            anomaly=anomaly,
            corrected=result.status.value == "completed",
            action_taken=f"Re-synced {anomaly.file_path}",
        )

    # _check_entity_kind's "kind" is the human-readable word used in the
    # anomaly description ("class"/"function"), not the types_enums.NodeType
    # vocabulary schema.get_node_label expects (NodeType.CLASS ==
    # "class_def"). Translate only where the two diverge.
    _ENTITY_KIND_TO_NODE_TYPE = {"class": "class_def"}

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-089"])
    async def _correct_orphan_relationship(self, anomaly: Anomaly) -> CorrectionResult:
        """Correct an ORPHAN_RELATIONSHIP by deleting the node."""
        entity_id = anomaly.details.get("entity_id", "")
        if not entity_id:
            return CorrectionResult(
                anomaly=anomaly,
                corrected=False,
                action_taken="No entity_id in anomaly details",
            )
        entity_kind = anomaly.details.get("entity_kind")
        node_type: str | None = None
        if isinstance(entity_kind, str):
            node_type = self._ENTITY_KIND_TO_NODE_TYPE.get(entity_kind, entity_kind)
        deleted = await self._client.delete_node(entity_id, node_type=node_type)
        name = anomaly.details.get("entity_name", "unknown")
        return CorrectionResult(
            anomaly=anomaly,
            corrected=deleted,
            action_taken=f"Deleted orphan node '{name}'",
        )

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-090"])
    async def _correct_line_count_drift(self, anomaly: Anomaly) -> CorrectionResult:
        """Correct LINE_COUNT_DRIFT by updating the property."""
        disk_lines = anomaly.details.get("disk_line_count", 0)
        updated = await self._client.update_file_line_count(
            anomaly.file_path, disk_lines, self._project_id
        )
        return CorrectionResult(
            anomaly=anomaly,
            corrected=updated,
            action_taken=(
                f"Updated line_count to {disk_lines} for {anomaly.file_path}"
            ),
        )

    @traces(spec="008", story="US-008-001", reqs=["REQ-008-091"])
    async def _apply_corrections(
        self,
        anomalies: list[Anomaly],
        report: AnomalyReport,
        mode: str,
    ) -> AnomalyReport:
        """Apply corrections based on audit mode.

        Args:
            anomalies: Detected anomalies to process.
            report: The report being built.
            mode: 'auto' or 'interactive'.

        Returns:
            Updated AnomalyReport.
        """
        for anomaly in anomalies:
            if not anomaly.auto_correctable:
                report.uncorrectable.append(anomaly)
                continue
            if mode == "interactive":
                if not _prompt_user(anomaly):
                    continue
            result = await self.correct(anomaly)
            report.corrections.append(result)
            if not result.corrected:
                report.uncorrectable.append(anomaly)
        return report


# ── Correction dispatch map ──────────────────────────────────

_CORRECTION_MAP = {
    AnomalyType.STALE_FILE: (AnomalyDetector._correct_stale_file),
    AnomalyType.MISSING_FILE: (AnomalyDetector._correct_missing_file),
    AnomalyType.HASH_MISMATCH: (AnomalyDetector._correct_hash_mismatch),
    AnomalyType.ENTITY_DRIFT: (AnomalyDetector._correct_hash_mismatch),
    AnomalyType.ORPHAN_RELATIONSHIP: (AnomalyDetector._correct_orphan_relationship),
    AnomalyType.LINE_COUNT_DRIFT: (AnomalyDetector._correct_line_count_drift),
}


# ── Module-level helpers ─────────────────────────────────────


@traces(spec="008", story="US-008-001", reqs=["REQ-008-070"])
def _get_or_create_loop() -> asyncio.AbstractEventLoop:
    """Get or create a persistent event loop for sync calls."""
    try:
        loop = asyncio.get_event_loop()
        if loop.is_closed():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        return loop
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        return loop


@traces(spec="008", story="US-008-001", reqs=["REQ-008-091"])
def _prompt_user(anomaly: Anomaly) -> bool:
    """Prompt user for interactive correction approval.

    Args:
        anomaly: The anomaly to ask about.

    Returns:
        True if user approves correction.
    """
    print(f"\n[{anomaly.anomaly_type.value}] {anomaly.file_path}")
    print(f"  {anomaly.description}")
    response = input("  Correct? (y/n/q): ").strip().lower()
    if response == "q":
        raise KeyboardInterrupt("User quit interactive mode")
    return response == "y"