cleveragents-core/benchmarks/temporal_data_model_bench.py

"""ASV benchmarks for Temporal Data Model (Revision-Aware RDF).

Measures the performance of:
- TemporalMetadata / TemporalNode / RevisionChain model creation
- InMemoryTemporalBackend revision operations
- Tier-aware temporal queries
- Revision chain traversal
"""

from __future__ import annotations

import importlib
import sys
from datetime import UTC, datetime, timedelta
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)

# Force-reload so ASV picks up the source tree version.
import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.domain.models.acms.temporal import (  # noqa: E402
    RevisionChain,
    TemporalMetadata,
    TemporalNode,
    TierRetentionConfig,
)
from cleveragents.domain.models.acms.temporal_stubs import (  # noqa: E402
    InMemoryTemporalBackend,
)
from cleveragents.domain.models.acms.tiers import ContextTier  # noqa: E402
from cleveragents.domain.models.core.project import (  # noqa: E402
    TemporalScope,
)

_NOW = datetime.now(tz=UTC)


# ---------------------------------------------------------------------------
# Model creation benchmarks
# ---------------------------------------------------------------------------


class TemporalModelSuite:
    """Benchmark temporal domain model creation overhead."""

    timeout = 60

    def time_create_temporal_metadata(self) -> None:
        TemporalMetadata(valid_from=_NOW)

    def time_create_temporal_metadata_full(self) -> None:
        TemporalMetadata(
            valid_from=_NOW,
            valid_until=_NOW + timedelta(days=1),
            is_current=False,
            is_revision_of="uko-py:class/Foo_v1",
        )

    def time_create_temporal_node(self) -> None:
        TemporalNode(
            node_uri="uko-py:class/Auth_v1",
            source_resource="RES001",
            source_path="src/auth.py",
            temporal=TemporalMetadata(valid_from=_NOW),
        )

    def time_create_revision_chain(self) -> None:
        RevisionChain(
            current_uri="uko-py:class/Auth_v3",
            predecessors=(
                "uko-py:class/Auth_v1",
                "uko-py:class/Auth_v2",
            ),
        )

    def time_create_tier_retention_config(self) -> None:
        TierRetentionConfig()


# ---------------------------------------------------------------------------
# Backend operation benchmarks
# ---------------------------------------------------------------------------


class TemporalBackendSuite:
    """Benchmark InMemoryTemporalBackend operations."""

    timeout = 60

    def setup(self) -> None:
        """Create a backend with 10 revision chains of 5 versions each."""
        self.backend = InMemoryTemporalBackend()
        self.retention = TierRetentionConfig()

        for chain_idx in range(10):
            base = f"uko-py:class/Chain{chain_idx}"
            prev_uri: str | None = None
            for ver in range(1, 6):
                uri = f"{base}_v{ver}"
                node = TemporalNode(
                    node_uri=uri,
                    source_resource=f"RES{chain_idx:03d}",
                    source_path=f"src/chain{chain_idx}.py",
                    temporal=TemporalMetadata(
                        valid_from=_NOW - timedelta(days=10 - ver),
                        is_current=(ver == 5),
                        is_revision_of=prev_uri,
                        valid_until=(
                            _NOW - timedelta(days=10 - ver - 1) if ver < 5 else None
                        ),
                    ),
                )
                self.backend.store_node(node)
                prev_uri = uri

    def time_get_current(self) -> None:
        self.backend.get_current("uko-py:class/Chain5")

    def time_get_history_all(self) -> None:
        self.backend.get_history(
            "uko-py:class/Chain3",
            TemporalScope.ALL,
        )

    def time_get_history_current(self) -> None:
        self.backend.get_history(
            "uko-py:class/Chain3",
            TemporalScope.CURRENT,
        )

    def time_get_revision_chain(self) -> None:
        self.backend.get_revision_chain("uko-py:class/Chain3_v5")

    def time_query_hot_tier(self) -> None:
        self.backend.query_by_tier(
            ContextTier.HOT,
            TemporalScope.CURRENT,
            self.retention,
        )

    def time_query_cold_tier(self) -> None:
        self.backend.query_by_tier(
            ContextTier.COLD,
            TemporalScope.ALL,
            self.retention,
        )

    def time_create_revision(self) -> None:
        """Benchmark a single revision creation (resets after)."""
        # Store a temporary node to revise
        tmp = TemporalNode(
            node_uri="uko-py:class/Tmp_v1",
            source_resource="RES999",
            source_path="src/tmp.py",
            temporal=TemporalMetadata(
                valid_from=_NOW - timedelta(days=1),
                is_current=True,
            ),
        )
        self.backend.store_node(tmp)

        new = TemporalNode(
            node_uri="uko-py:class/Tmp_v2",
            source_resource="RES999",
            source_path="src/tmp.py",
            temporal=TemporalMetadata(
                valid_from=_NOW,
                is_current=True,
                is_revision_of="uko-py:class/Tmp_v1",
            ),
        )
        self.backend.create_revision(
            "uko-py:class/Tmp_v1",
            new,
            _NOW,
        )

        # Clean up for next iteration
        del self.backend._nodes["uko-py:class/Tmp_v1"]
        del self.backend._nodes["uko-py:class/Tmp_v2"]

    def time_mark_historical(self) -> None:
        """Benchmark marking a node historical (resets after)."""
        tmp = TemporalNode(
            node_uri="uko-py:class/MarkTmp_v1",
            source_resource="RES999",
            source_path="src/tmp.py",
            temporal=TemporalMetadata(
                valid_from=_NOW,
                is_current=True,
            ),
        )
        self.backend.store_node(tmp)
        self.backend.mark_historical("uko-py:class/MarkTmp_v1", _NOW)
        del self.backend._nodes["uko-py:class/MarkTmp_v1"]