cleveragents-core/benchmarks/session_model_bench.py

"""ASV benchmarks for Session domain model validation and serialization.

Measures the performance of:
- Session model construction (Pydantic validation)
- SessionMessage construction
- Session.append_message() auto-sequencing
- Session.as_cli_dict() serialization
- Session.as_export_dict() serialization with checksum
- SessionTokenUsage construction
- Session.get_messages() with pagination
"""

from __future__ import annotations

import sys
from pathlib import Path

try:
    from cleveragents.domain.models.core.session import (
        MessageRole,
        Session,
        SessionMessage,
        SessionTokenUsage,
    )
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.domain.models.core.session import (
        MessageRole,
        Session,
        SessionMessage,
        SessionTokenUsage,
    )

from ulid import ULID


def _make_session(message_count: int = 0) -> Session:
    """Create a Session with optional messages for benchmarking."""
    session = Session(session_id=str(ULID()))
    for i in range(message_count):
        session.append_message(MessageRole.USER, f"Benchmark message {i}")
    return session


def _make_message() -> SessionMessage:
    """Create a fully-populated SessionMessage for benchmarking."""
    return SessionMessage(
        message_id=str(ULID()),
        role=MessageRole.USER,
        content="Benchmark message content",
        sequence=0,
        metadata={"key": "value"},
    )


class SessionValidationSuite:
    """Benchmark Session model construction (Pydantic validation)."""

    def time_session_construction(self) -> None:
        """Benchmark creating a minimal Session object."""
        Session(session_id=str(ULID()))

    def time_session_with_actor(self) -> None:
        """Benchmark creating a Session with actor_name."""
        Session(session_id=str(ULID()), actor_name="local/orchestrator")

    def time_message_construction(self) -> None:
        """Benchmark creating a SessionMessage."""
        _make_message()

    def time_tool_message_construction(self) -> None:
        """Benchmark creating a tool SessionMessage with tool_call_id."""
        SessionMessage(
            message_id=str(ULID()),
            role=MessageRole.TOOL,
            content="Tool result",
            sequence=0,
            tool_call_id="call_123",
        )

    def time_token_usage_construction(self) -> None:
        """Benchmark SessionTokenUsage construction."""
        SessionTokenUsage(
            input_tokens=1000,
            output_tokens=500,
            estimated_cost=0.05,
        )


class SessionAppendSuite:
    """Benchmark Session.append_message() auto-sequencing."""

    def setup(self) -> None:
        """Create a session for append benchmarks."""
        self.session = _make_session()

    def time_append_single_message(self) -> None:
        """Benchmark appending a single message."""
        self.session.append_message(MessageRole.USER, "Benchmark message")

    def time_append_10_messages(self) -> None:
        """Benchmark appending 10 messages sequentially."""
        session = _make_session()
        for i in range(10):
            session.append_message(MessageRole.USER, f"Message {i}")


class SessionSerializationSuite:
    """Benchmark Session serialization methods."""

    def setup(self) -> None:
        """Create a session with messages for serialization benchmarks."""
        self.session = _make_session(message_count=10)
        self.session.link_plan(str(ULID()))
        self.large_session = _make_session(message_count=100)

    def time_as_cli_dict(self) -> None:
        """Benchmark Session.as_cli_dict() ordered dict generation."""
        self.session.as_cli_dict()

    def time_as_export_dict(self) -> None:
        """Benchmark Session.as_export_dict() with checksum generation."""
        self.session.as_export_dict()

    def time_as_export_dict_large(self) -> None:
        """Benchmark export dict for a large session (100 messages)."""
        self.large_session.as_export_dict()

    def time_model_dump(self) -> None:
        """Benchmark Pydantic model_dump() serialization."""
        self.session.model_dump()

    def time_model_dump_json(self) -> None:
        """Benchmark Pydantic model_dump_json() JSON serialization."""
        self.session.model_dump_json()


class SessionPaginationSuite:
    """Benchmark Session.get_messages() with pagination."""

    def setup(self) -> None:
        """Create a session with many messages for pagination benchmarks."""
        self.session = _make_session(message_count=100)

    def time_get_all_messages(self) -> None:
        """Benchmark getting all messages (no limit)."""
        self.session.get_messages()

    def time_get_first_10(self) -> None:
        """Benchmark getting first 10 messages."""
        self.session.get_messages(limit=10)

    def time_get_with_offset(self) -> None:
        """Benchmark getting messages with offset."""
        self.session.get_messages(limit=10, offset=50)