"""In-memory session store — a portable harness pattern.

Single-process, GIL-protected. Each session holds its conversation history

as a list of message dicts compatible with the typical OpenAI chat

completions format. Not safe for multi-process deployments; replace with

Redis or a database for persistence and true concurrency safety.

"""

from __future__ import annotations

from datetime import UTC, datetime

from typing import Any

from uuid import UUID, uuid4

from src.models.session import SessionInfo

class SessionStore:

"""In-memory session store."""

def __init__(self) -> None:

self._sessions: dict[str, dict[str, Any]] = {}

def create(self) -> SessionInfo:

"""Create a new session with an empty conversation history."""

session_id = str(uuid4())

self._sessions[session_id] = {

"session_id": session_id,

"created_at": datetime.now(tz=UTC),

"messages": [],

}

return SessionInfo(

session_id=UUID(session_id),

created_at=self._sessions[session_id]["created_at"],

message_count=0,

)

def get(self, session_id: str) -> SessionInfo | None:

"""Get session info, or None if not found."""

data = self._sessions.get(session_id)

if data is None:

return None

return SessionInfo(

session_id=UUID(data["session_id"]),

created_at=data["created_at"],

message_count=len(data["messages"]),

)

def get_messages(self, session_id: str) -> list[dict[str, Any]] | None:

"""Get conversation history for a session, or None if not found."""

data = self._sessions.get(session_id)

if data is None:

return None

return list(data["messages"])

def set_messages(self, session_id: str, messages: list[dict[str, Any]]) -> None:

"""Replace the conversation history for a session."""

if session_id in self._sessions:

self._sessions[session_id]["messages"] = messages

def exists(self, session_id: str) -> bool:

"""Check if a session exists."""

return session_id in self._sessions