cleveragents-core/benchmarks/influence_dag_bench.py

"""Airspeed Velocity benchmarks for influence DAG traversal.

Measures BFS traversal performance over combined structural tree and
influence DAG edges at varying graph sizes, including cycle detection
overhead.
"""

from __future__ import annotations

from cleveragents.application.services.correction_service import CorrectionService
from cleveragents.domain.models.core.correction import CorrectionMode

# ---------------------------------------------------------------------------
# Graph generators
# ---------------------------------------------------------------------------

_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6EA00"


def _linear_tree(root: str, depth: int) -> dict[str, list[str]]:
    """Build a linear chain of *depth* nodes starting at *root*."""
    tree: dict[str, list[str]] = {}
    current = root
    for i in range(1, depth):
        child = f"{root}_c{i}"
        tree[current] = [child]
        current = child
    return tree


def _star_influence(root: str, fan_out: int) -> dict[str, list[str]]:
    """Build a star influence DAG: root influences *fan_out* nodes."""
    return {root: [f"{root}_inf{i}" for i in range(fan_out)]}


def _chain_influence(root: str, depth: int) -> dict[str, list[str]]:
    """Build a linear chain influence DAG of *depth* nodes."""
    dag: dict[str, list[str]] = {}
    current = root
    for i in range(1, depth):
        nxt = f"{root}_d{i}"
        dag[current] = [nxt]
        current = nxt
    return dag


def _mixed_graph(
    root: str,
    tree_depth: int,
    influence_fan_out: int,
) -> tuple[dict[str, list[str]], dict[str, list[str]]]:
    """Build a combined tree + influence DAG."""
    tree = _linear_tree(root, tree_depth)
    dag = _star_influence(root, influence_fan_out)
    return tree, dag


# ---------------------------------------------------------------------------
# Structural-only BFS benchmarks (baseline)
# ---------------------------------------------------------------------------


class StructuralOnlyBfsSuite:
    """Baseline: BFS over structural tree only (no influence edges)."""

    def setup(self) -> None:
        """Prepare trees and service instances."""
        self.svc = CorrectionService()
        self.small_tree = _linear_tree("D1", 10)
        self.medium_tree = _linear_tree("D1", 100)
        self.large_tree = _linear_tree("D1", 1000)

    def time_small_structural_only(self) -> None:
        """BFS 10-node linear tree, no influence edges."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.small_tree)

    def time_medium_structural_only(self) -> None:
        """BFS 100-node linear tree, no influence edges."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.medium_tree)

    def time_large_structural_only(self) -> None:
        """BFS 1000-node linear tree, no influence edges."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.large_tree)


# ---------------------------------------------------------------------------
# Influence-only BFS benchmarks
# ---------------------------------------------------------------------------


class InfluenceOnlyBfsSuite:
    """BFS over influence DAG only (no structural tree)."""

    def setup(self) -> None:
        """Prepare influence DAGs."""
        self.svc = CorrectionService()
        self.empty_tree: dict[str, list[str]] = {}
        self.small_dag = _chain_influence("D1", 10)
        self.medium_dag = _chain_influence("D1", 100)
        self.large_dag = _chain_influence("D1", 1000)
        self.star_dag = _star_influence("D1", 500)

    def time_small_influence_only(self) -> None:
        """BFS 10-node influence chain."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.empty_tree, self.small_dag)

    def time_medium_influence_only(self) -> None:
        """BFS 100-node influence chain."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.empty_tree, self.medium_dag)

    def time_large_influence_only(self) -> None:
        """BFS 1000-node influence chain."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.empty_tree, self.large_dag)

    def time_star_influence(self) -> None:
        """BFS star DAG with 500 fan-out."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.empty_tree, self.star_dag)


# ---------------------------------------------------------------------------
# Combined structural + influence BFS benchmarks
# ---------------------------------------------------------------------------


class CombinedBfsSuite:
    """BFS over combined structural tree + influence DAG."""

    def setup(self) -> None:
        """Prepare combined graphs."""
        self.svc = CorrectionService()
        self.small_tree, self.small_dag = _mixed_graph("D1", 10, 10)
        self.medium_tree, self.medium_dag = _mixed_graph("D1", 100, 50)
        self.large_tree, self.large_dag = _mixed_graph("D1", 500, 200)

    def time_small_combined(self) -> None:
        """BFS combined 10+10 graph."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.small_tree, self.small_dag)

    def time_medium_combined(self) -> None:
        """BFS combined 100+50 graph."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.medium_tree, self.medium_dag)

    def time_large_combined(self) -> None:
        """BFS combined 500+200 graph."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.large_tree, self.large_dag)


# ---------------------------------------------------------------------------
# Cycle detection overhead benchmarks
# ---------------------------------------------------------------------------


class CycleDetectionSuite:
    """Measure overhead of cycle detection in corrupt DAGs."""

    def setup(self) -> None:
        """Prepare DAGs with cycles."""
        self.svc = CorrectionService()
        self.empty_tree: dict[str, list[str]] = {}
        # Small cycle: D1 → D2 → D3 → D1
        self.small_cycle = {"D1": ["D2"], "D2": ["D3"], "D3": ["D1"]}
        # Medium: chain of 50 with cycle at end
        dag: dict[str, list[str]] = {}
        current = "D1"
        for i in range(1, 50):
            nxt = f"D1_cyc{i}"
            dag[current] = [nxt]
            current = nxt
        dag[current] = ["D1"]  # cycle back to start
        self.medium_cycle = dag

    def time_small_cycle(self) -> None:
        """BFS with 3-node cycle."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.empty_tree, self.small_cycle)

    def time_medium_cycle(self) -> None:
        """BFS with 50-node chain + cycle."""
        req = self.svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
        self.svc.analyze_impact(req.correction_id, self.empty_tree, self.medium_cycle)