cleveragents-core/benchmarks/output_rendering_bench.py

"""ASV benchmarks for the output rendering framework.

Measures materialiser throughput, progress handle overhead, and session
lifecycle costs across all six output formats.
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path

# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.cli.output.handles import DiffLine  # noqa: E402
from cleveragents.cli.output.materializers import (  # noqa: E402
    ColorMaterializer,
    JsonMaterializer,
    PlainMaterializer,
    RichMaterializer,
    TableMaterializer,
    YamlMaterializer,
)
from cleveragents.cli.output.session import OutputSession  # noqa: E402

_STRATEGY_MAP: dict[str, type] = {
    "rich": RichMaterializer,
    "color": ColorMaterializer,
    "table": TableMaterializer,
    "plain": PlainMaterializer,
    "json": JsonMaterializer,
    "yaml": YamlMaterializer,
}


class MaterializerThroughputSuite:
    """Benchmark materialiser throughput for panel + table rendering."""

    params: list[str] = ["rich", "color", "table", "plain", "json", "yaml"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_panel_render(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            panel = session.panel("Benchmark Panel")
            for i in range(50):
                panel.set_entry(f"key-{i}", f"value-{i}")
            panel.close()
        strategy.get_output()

    def time_table_render(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            table = session.table(columns=["Name", "Status", "Count"])
            for i in range(100):
                table.add_row(
                    {"Name": f"item-{i}", "Status": "active", "Count": str(i)}
                )
            table.close()
        strategy.get_output()


class ProgressHandleSuite:
    """Benchmark progress handle update overhead."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_progress_ticks(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            prog = session.progress("Ticking", total=1000)
            for _ in range(1000):
                prog.tick()
            prog.close()

    def time_progress_set(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            prog = session.progress("Stepping", total=500)
            for i in range(500):
                prog.set_progress(i, 500)
            prog.close()


class SessionLifecycleSuite:
    """Benchmark session open/close overhead."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_empty_session(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy):
            pass

    def time_session_with_handles(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            p = session.panel("P")
            p.set_entry("k", "v")
            p.close()
            t = session.table(columns=["A"])
            t.add_row({"A": "1"})
            t.close()
            s = session.status("ok")
            s.close()


class TreeThroughputSuite:
    """Benchmark tree-building throughput with deep/wide trees."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_wide_tree(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            tree = session.tree("Root")
            for i in range(100):
                tree.add_child("Root", f"child-{i}")
            tree.close()
        strategy.get_output()

    def time_deep_tree(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            tree = session.tree("Root")
            path = "Root"
            for i in range(50):
                label = f"level-{i}"
                path = tree.add_child(path, label)
            tree.close()
        strategy.get_output()


class DiffThroughputSuite:
    """Benchmark diff rendering with multi-hunk diffs."""

    params: list[str] = ["plain", "color", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_multi_hunk_diff(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            diff = session.diff(file_a="old.py", file_b="new.py")
            for h in range(20):
                lines = []
                for j in range(10):
                    lines.append(DiffLine(type="context", content=f"ctx-{j}"))
                    lines.append(DiffLine(type="add", content=f"add-{j}"))
                    lines.append(DiffLine(type="remove", content=f"rem-{j}"))
                diff.add_hunk(f"@@ -{h * 30},{30} +{h * 30},{30} @@", lines)
            diff.set_stats(200, 200)
            diff.close()
        strategy.get_output()


class MixedElementSuite:
    """Benchmark session throughput with all 10 element types."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_all_element_types(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            p = session.panel("P")
            p.set_entry("k", "v")
            p.close()
            t = session.table(columns=["A"])
            t.add_row({"A": "1"})
            t.close()
            s = session.status("ok")
            s.close()
            pr = session.progress("go", total=10)
            pr.set_progress(5)
            pr.close()
            tr = session.tree("Root")
            tr.add_child("Root", "leaf")
            tr.close()
            tx = session.text("hello world")
            tx.close()
            co = session.code("x = 1", language="python")
            co.close()
            d = session.diff(file_a="a", file_b="b")
            d.add_hunk("@@", [DiffLine(type="add", content="new")])
            d.close()
            session.separator("line")
            session.action_hint(["cmd"])
        strategy.get_output()


class TextAppendSuite:
    """Benchmark TextHandle.append() throughput (N10: detect O(N²) regressions)."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_text_append_1000(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            tx = session.text("")
            for i in range(1000):
                tx.append(f"Line {i}\n")
            tx.close()
        strategy.get_output()

    def time_text_append_5000(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            tx = session.text("")
            for i in range(5000):
                tx.append(f"Line {i}\n")
            tx.close()
        strategy.get_output()


class LargeTableSuite:
    """Benchmark table rendering at 10K+ rows (N10)."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_table_10k_rows(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            table = session.table(columns=["ID", "Name", "Value"])
            for i in range(10_000):
                table.add_row({"ID": str(i), "Name": f"item-{i}", "Value": str(i * 10)})
            table.close()
        strategy.get_output()


class SnapshotCostSuite:
    """Benchmark snapshot() cost (N10)."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_snapshot_with_many_elements(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            for i in range(50):
                p = session.panel(f"Panel-{i}")
                p.set_entry("k", "v")
                p.close()
            for _ in range(20):
                session.snapshot()

    def time_snapshot_with_large_table(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            table = session.table(columns=["A", "B"])
            for i in range(5000):
                table.add_row({"A": str(i), "B": str(i)})
            table.close()
            for _ in range(10):
                session.snapshot()


class DeepTreeLookupSuite:
    """Benchmark deep tree path lookups (N10)."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_deep_tree_add_and_lookup(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            tree = session.tree("Root")
            path = "Root"
            for i in range(50):
                label = f"level-{i}"
                path = tree.add_child(path, label)
            # Now do lookups at various depths
            tree.add_child("Root", "wide-child-1")
            tree.add_child("Root", "wide-child-2")
            tree.close()
        strategy.get_output()


class ElementUpdatedOverheadSuite:
    """Benchmark ElementUpdated event dispatch overhead (N10)."""

    params: list[str] = ["plain", "json"]  # noqa: RUF012
    param_names: list[str] = ["format"]  # noqa: RUF012

    def time_panel_many_updates(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            panel = session.panel("Update Panel")
            for i in range(500):
                panel.set_entry(f"key-{i % 50}", f"value-{i}")
            panel.close()

    def time_progress_many_ticks(self, fmt: str) -> None:
        strategy = _STRATEGY_MAP[fmt]()
        with OutputSession(format=fmt, command="bench", strategy=strategy) as session:
            prog = session.progress("Ticking", total=5000)
            for _ in range(5000):
                prog.tick()
            prog.close()