session.py

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeGuard, runtime_checkable

from typing_extensions import TypedDict

if TYPE_CHECKING:
    from ..items import TResponseInputItem
    from ..run_context import RunContextWrapper
    from .session_settings import SessionSettings


@runtime_checkable
class Session(Protocol):
    """Protocol for session implementations.

    Session stores conversation history for a specific session, allowing
    agents to maintain context without requiring explicit manual memory management.
    """

    session_id: str
    session_settings: SessionSettings | None = None

    async def get_items(
        self,
        limit: int | None = None,
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> list[TResponseInputItem]:
        """Retrieve the conversation history for this session.

        Args:
            limit: Maximum number of items to retrieve. If None, retrieves all items.
                   When specified, returns the latest N items in chronological order.
            wrapper: Optional run context wrapper providing context and usage info.

        Returns:
            List of input items representing the conversation history
        """
        ...

    async def add_items(
        self,
        items: list[TResponseInputItem],
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> None:
        """Add new items to the conversation history.

        Args:
            items: List of input items to add to the history
            wrapper: Optional run context wrapper providing context and usage info.
        """
        ...

    async def pop_item(
        self,
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> TResponseInputItem | None:
        """Remove and return the most recent item from the session.

        Args:
            wrapper: Optional run context wrapper providing context and usage info.

        Returns:
            The most recent item if it exists, None if the session is empty
        """
        ...

    async def clear_session(
        self,
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> None:
        """Clear all items for this session.

        Args:
            wrapper: Optional run context wrapper providing context and usage info.
        """
        ...


class SessionABC(ABC):
    """Abstract base class for session implementations.

    Session stores conversation history for a specific session, allowing
    agents to maintain context without requiring explicit manual memory management.

    This ABC is intended for internal use and as a base class for concrete implementations.
    Third-party libraries should implement the Session protocol instead.
    """

    session_id: str
    session_settings: SessionSettings | None = None

    @abstractmethod
    async def get_items(
        self,
        limit: int | None = None,
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> list[TResponseInputItem]:
        """Retrieve the conversation history for this session.

        Args:
            limit: Maximum number of items to retrieve. If None, retrieves all items.
                   When specified, returns the latest N items in chronological order.
            wrapper: Optional run context wrapper providing context and usage info.

        Returns:
            List of input items representing the conversation history
        """
        ...

    @abstractmethod
    async def add_items(
        self,
        items: list[TResponseInputItem],
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> None:
        """Add new items to the conversation history.

        Args:
            items: List of input items to add to the history
            wrapper: Optional run context wrapper providing context and usage info.
        """
        ...

    @abstractmethod
    async def pop_item(
        self,
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> TResponseInputItem | None:
        """Remove and return the most recent item from the session.

        Args:
            wrapper: Optional run context wrapper providing context and usage info.

        Returns:
            The most recent item if it exists, None if the session is empty
        """
        ...

    @abstractmethod
    async def clear_session(
        self,
        wrapper: RunContextWrapper[Any] | None = None,
    ) -> None:
        """Clear all items for this session.

        Args:
            wrapper: Optional run context wrapper providing context and usage info.
        """
        ...


class OpenAIResponsesCompactionArgs(TypedDict, total=False):
    """Arguments for the run_compaction method."""

    response_id: str
    """The ID of the last response to use for compaction."""

    compaction_mode: Literal["previous_response_id", "input", "auto"]
    """How to provide history for compaction.

    - "auto": Use input when the last response was not stored or no response ID is available.
    - "previous_response_id": Use server-managed response history.
    - "input": Send locally stored session items as input.
    """

    store: bool
    """Whether the last model response was stored on the server.

    When set to False, compaction should avoid "previous_response_id" unless explicitly requested.
    """

    force: bool
    """Whether to force compaction even if the threshold is not met."""


@runtime_checkable
class OpenAIResponsesCompactionAwareSession(Session, Protocol):
    """Protocol for session implementations that support responses compaction."""

    async def run_compaction(self, args: OpenAIResponsesCompactionArgs | None = None) -> None:
        """Run the compaction process for the session."""
        ...


def is_openai_responses_compaction_aware_session(
    session: Session | None,
) -> TypeGuard[OpenAIResponsesCompactionAwareSession]:
    """Check if a session supports responses compaction."""
    if session is None:
        return False
    try:
        run_compaction = getattr(session, "run_compaction", None)
    except Exception:
        return False
    return callable(run_compaction)