← back to invincible-jha__agent-session-linker

Function bodies 227 total

ContextWindowManager.__init__ method · python · L44-L62 (19 LOC)

src/agent_session_linker/middleware/context_window.py

    def __init__(
        self,
        max_tokens: int = 4000,
        max_segments: int = 50,
        role_separator: str = ": ",
        segment_separator: str = "\n\n",
    ) -> None:
        if max_tokens < 1:
            raise ValueError(f"max_tokens must be >= 1, got {max_tokens!r}.")
        if max_segments < 1:
            raise ValueError(f"max_segments must be >= 1, got {max_segments!r}.")

        self.max_tokens = max_tokens
        self.max_segments = max_segments
        self.role_separator = role_separator
        self.segment_separator = segment_separator

        self._window: deque[ContextSegment] = deque()
        self._token_total: int = 0

ContextWindowManager.add method · python · L68-L97 (30 LOC)

src/agent_session_linker/middleware/context_window.py

    def add(self, segment: ContextSegment) -> None:
        """Append a segment to the window, evicting old segments if needed.

        Token count is taken from ``segment.token_count`` when non-zero,
        otherwise estimated from content length.

        Parameters
        ----------
        segment:
            The segment to append.
        """
        segment_tokens = segment.token_count or _estimate_tokens(segment.content)

        # If a single segment is larger than the entire budget, accept it
        # as a lone entry (the window holds at least one segment always).
        if segment_tokens > self.max_tokens and not self._window:
            self._window.append(segment)
            self._token_total += segment_tokens
            return

        # Evict from the front until there is room.
        while self._window and (
            self._token_total + segment_tokens > self.max_tokens
            or len(self._window) >= self.max_segments
        ):
            evicted = sel

ContextWindowManager.get_window method · python · L99-L119 (21 LOC)

src/agent_session_linker/middleware/context_window.py

    def get_window(self) -> str:
        """Render the current window as a formatted string.

        Each segment is rendered as ``"<ROLE>: <content>"`` with segments
        separated by ``self.segment_separator``.

        Returns
        -------
        str
            Formatted context window string.  Empty string when the window
            is empty.
        """
        if not self._window:
            return ""

        parts: list[str] = []
        for segment in self._window:
            role_label = segment.role.upper()
            parts.append(f"{role_label}{self.role_separator}{segment.content}")

        return self.segment_separator.join(parts)

ContextWindowManager.get_segments method · python · L121-L129 (9 LOC)

src/agent_session_linker/middleware/context_window.py

    def get_segments(self) -> list[ContextSegment]:
        """Return a copy of the current window as an ordered list.

        Returns
        -------
        list[ContextSegment]
            Segments from oldest to newest.
        """
        return list(self._window)

ContextWindowManager.token_count method · python · L131-L139 (9 LOC)

src/agent_session_linker/middleware/context_window.py

    def token_count(self) -> int:
        """Return the current cumulative token count in the window.

        Returns
        -------
        int
            Total estimated tokens for all segments in the window.
        """
        return self._token_total

ContextWindowManager.__repr__ method · python · L149-L154 (6 LOC)

src/agent_session_linker/middleware/context_window.py

    def __repr__(self) -> str:
        return (
            f"ContextWindowManager("
            f"segments={len(self._window)}, "
            f"tokens={self._token_total}/{self.max_tokens})"
        )

SessionMiddleware.__init__ method · python · L40-L47 (8 LOC)

src/agent_session_linker/middleware/session_middleware.py

    def __init__(
        self,
        manager: SessionManager,
        auto_create: bool = True,
    ) -> None:
        self._manager = manager
        self.auto_create = auto_create
        self._active_sessions: dict[str, SessionState] = {}

Repobility · code-quality intelligence · https://repobility.com

SessionMiddleware.before_request method · python · L53-L87 (35 LOC)

src/agent_session_linker/middleware/session_middleware.py

    def before_request(self, session_id: str) -> SessionState:
        """Load (or create) the session for ``session_id``.

        The loaded session is cached in-memory for the duration of the
        request and returned for immediate use.

        Parameters
        ----------
        session_id:
            The session identifier for the incoming request.

        Returns
        -------
        SessionState
            The existing or newly created session.

        Raises
        ------
        SessionNotFoundError
            If ``auto_create`` is False and no session exists for
            ``session_id``.
        """
        if self._manager.session_exists(session_id):
            session = self._manager.load_session(session_id)
            logger.debug("SessionMiddleware: loaded session %r", session_id)
        elif self.auto_create:
            session = self._manager.create_session()
            # Force the session ID to match the requested one.
            session.session_

SessionMiddleware.after_request method · python · L89-L137 (49 LOC)

src/agent_session_linker/middleware/session_middleware.py

    def after_request(
        self,
        session_id: str,
        new_context: list[ContextSegment] | str | None = None,
    ) -> str:
        """Persist the session and optionally append new context.

        If ``new_context`` is provided the segments (or single text string)
        are appended to the session before it is saved.

        Parameters
        ----------
        session_id:
            The session identifier used in the preceding ``before_request``
            call.
        new_context:
            Context to append before saving.  May be:
            - A list of ``ContextSegment`` objects appended directly.
            - A plain string added as a single ``"assistant"`` segment.
            - None to save without any new context.

        Returns
        -------
        str
            The ``session_id`` under which the session was saved.

        Raises
        ------
        KeyError
            If ``before_request`` was not called for ``session_id`` in this

SessionMiddleware.get_active method · python · L139-L152 (14 LOC)

src/agent_session_linker/middleware/session_middleware.py

    def get_active(self, session_id: str) -> SessionState | None:
        """Return the currently active (in-flight) session, if any.

        Parameters
        ----------
        session_id:
            Session identifier to look up.

        Returns
        -------
        SessionState | None
            The cached session state, or None if not currently active.
        """
        return self._active_sessions.get(session_id)

SessionMiddleware.clear_active method · python · L154-L165 (12 LOC)

src/agent_session_linker/middleware/session_middleware.py

    def clear_active(self, session_id: str) -> None:
        """Discard the cached active session without saving.

        Useful for rolling back a request cycle that encountered an error.

        Parameters
        ----------
        session_id:
            Session identifier to discard.
        """
        self._active_sessions.pop(session_id, None)
        logger.debug("SessionMiddleware: cleared active session %r", session_id)

PluginNotFoundError.__init__ method · python · L55-L62 (8 LOC)