Session API

SessionManager stores conversations in SQLite. Agent.run() creates a session automatically and accepts a string session_id for later turns. For manual creation, use get_or_create_session(session_id, user_id), which returns the ID, not a session object. list_sessions() returns dictionaries.

The methods below follow the current source. They are storage operations, not an authorization boundary; applications must enforce user access and serialize operations on a shared session. See Sessions.

nagents.SessionManager

SQLite-based session and message history management.

Stores conversation history in a SQLite database, allowing for persistent sessions across restarts.

Example

session = SessionManager(Path("sessions.db")) await session.initialize()

await session.add_message("session-1", "user-1", Message(role="user", content="Hello")) history = await session.get_history("session-1")

Source code in src/nagents/session/manager.py

class SessionManager:
    """
    SQLite-based session and message history management.

    Stores conversation history in a SQLite database, allowing for
    persistent sessions across restarts.

    Example:
        session = SessionManager(Path("sessions.db"))
        await session.initialize()

        await session.add_message("session-1", "user-1", Message(role="user", content="Hello"))
        history = await session.get_history("session-1")
    """

    def __init__(self, db_path: Path):
        """
        Initialize the session manager.

        Args:
            db_path: Path to the SQLite database file
        """
        self.db_path = db_path
        self._initialized = False
        self._migration_manager: MigrationManager | None = None

    async def initialize(self) -> None:
        """Initialize database and run migrations."""
        if self._initialized:
            return

        # Ensure parent directory exists
        self.db_path.parent.mkdir(parents=True, exist_ok=True)

        # Create migration manager with session-specific migrations
        self._migration_manager = MigrationManager(
            self.db_path, db_name="sessions", migrations=list(session_migrations), atomic=True
        )
        await self._migration_manager.initialize()

        self._initialized = True
        logger.debug(f"Session database initialized at {self.db_path}")

    async def get_schema_version(self) -> int:
        """Get current database schema version.

        Returns:
            Current version number, or 0 if no migrations applied
        """
        await self.initialize()
        if self._migration_manager is None:
            return 0
        return await self._migration_manager.get_version()

    async def get_migration_history(self) -> list[dict[str, Any]]:
        """Get history of applied migrations.

        Returns:
            List of migration info dicts
        """
        await self.initialize()
        if self._migration_manager is None:
            return []
        return await self._migration_manager.get_applied_migrations()

    async def rollback_migration(self, steps: int = 1) -> None:
        """Rollback the last N migrations.

        Args:
            steps: Number of migrations to rollback (default: 1)

        Raises:
            ValueError: If not enough migrations to rollback
        """
        await self.initialize()
        if self._migration_manager is None:
            return
        await self._migration_manager.rollback(steps)

    async def migrate_to_version(self, version: int) -> None:
        """Migrate database to a specific version.

        Args:
            version: Target schema version

        Raises:
            ValueError: If target version is invalid
        """
        await self.initialize()
        if self._migration_manager is None:
            return
        await self._migration_manager.migrate_to(version)

    async def get_or_create_session(self, session_id: str, user_id: str) -> str:
        """
        Get or create a session.

        Args:
            session_id: Unique session identifier
            user_id: User identifier

        Returns:
            The session ID
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute("SELECT id FROM v2_sessions WHERE id = ?", (session_id,))
            if await cursor.fetchone() is None:
                await db.execute(
                    "INSERT INTO v2_sessions (id, user_id) VALUES (?, ?)",
                    (session_id, user_id),
                )
                await db.commit()
                logger.debug(f"Created new session: {session_id}")

        return session_id

    async def get_history(
        self,
        session_id: str,
        limit: int | None = None,
    ) -> list[Message]:
        """
        Get all messages for a session.

        Args:
            session_id: Session identifier
            limit: Optional limit on number of messages (most recent)

        Returns:
            List of Message objects in chronological order
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row

            # Check for compaction boundary
            boundary_cursor = await db.execute(
                "SELECT compacted_at_message_id FROM v2_sessions WHERE id = ?",
                (session_id,),
            )
            boundary_row = await boundary_cursor.fetchone()
            compaction_boundary: int | None = None
            if boundary_row and boundary_row["compacted_at_message_id"] is not None:
                compaction_boundary = boundary_row["compacted_at_message_id"]

            if limit:
                # Get most recent N messages (respecting compaction boundary)
                if compaction_boundary is not None:
                    cursor = await db.execute(
                        """SELECT * FROM (
                            SELECT * FROM v2_messages
                            WHERE session_id = ? AND id >= ?
                            ORDER BY id DESC
                            LIMIT ?
                        ) ORDER BY id ASC""",
                        (session_id, compaction_boundary, limit),
                    )
                else:
                    cursor = await db.execute(
                        """SELECT * FROM (
                            SELECT * FROM v2_messages
                            WHERE session_id = ?
                            ORDER BY id DESC
                            LIMIT ?
                        ) ORDER BY id ASC""",
                        (session_id, limit),
                    )
            else:
                if compaction_boundary is not None:
                    cursor = await db.execute(
                        "SELECT * FROM v2_messages WHERE session_id = ? AND id >= ? ORDER BY id",
                        (session_id, compaction_boundary),
                    )
                else:
                    cursor = await db.execute(
                        "SELECT * FROM v2_messages WHERE session_id = ? ORDER BY id",
                        (session_id,),
                    )

            rows = await cursor.fetchall()
            return [self._row_to_message(row) for row in rows]

    async def add_message(self, session_id: str, message: Message) -> int:
        """
        Add a message to session history.

        Args:
            session_id: Session identifier
            message: Message to add

        Returns:
            The database row ID of the inserted message
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute(
                """INSERT INTO v2_messages
                   (session_id, role, content, tool_calls, tool_call_id, name)
                   VALUES (?, ?, ?, ?, ?, ?)""",
                self._message_values(session_id, message),
            )
            message_id = cursor.lastrowid
            assert message_id is not None
            await db.execute(
                "UPDATE v2_sessions SET updated_at = CURRENT_TIMESTAMP WHERE id = ?",
                (session_id,),
            )
            await db.commit()

        return message_id

    @staticmethod
    def _message_values(session_id: str, message: Message) -> tuple[str | None, ...]:
        """Shared serialization for ordinary messages and context replacements."""
        return (
            session_id,
            message.role,
            json.dumps([asdict(part) for part in message.content])
            if isinstance(message.content, list)
            else message.content,
            json.dumps([asdict(call) for call in message.tool_calls]) if message.tool_calls else None,
            message.tool_call_id,
            message.name,
        )

    async def replace_context(self, session_id: str, messages: list[Message]) -> None:
        """Append a validated replacement and move the inclusive boundary atomically.

        Raw history is never deleted. Validation and serialization complete
        before any writes; failed inserts/boundary updates roll back together.
        """
        _validate_context(messages)
        values = [self._message_values(session_id, message) for message in messages]
        await self.initialize()
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("BEGIN IMMEDIATE")
            cursor = await db.execute("SELECT 1 FROM v2_sessions WHERE id = ?", (session_id,))
            if await cursor.fetchone() is None:
                raise ValueError(f"Session '{session_id}' not found")
            first_id: int | None = None
            for row in values:
                cursor = await db.execute(
                    """INSERT INTO v2_messages
                       (session_id, role, content, tool_calls, tool_call_id, name)
                       VALUES (?, ?, ?, ?, ?, ?)""",
                    row,
                )
                if first_id is None:
                    first_id = cursor.lastrowid
            assert first_id is not None
            await db.execute(
                """UPDATE v2_sessions SET compacted_at_message_id = ?,
                   updated_at = CURRENT_TIMESTAMP WHERE id = ?""",
                (first_id, session_id),
            )
            await db.commit()

    async def set_compaction_boundary(self, session_id: str, message_id: int) -> None:
        """
        Record a compaction boundary for a session.

        After compaction, get_history() will only return messages
        starting from this message_id (inclusive), filtering out
        all older pre-compaction messages.

        Args:
            session_id: Session identifier
            message_id: The message ID of the compaction summary
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            await db.execute(
                "UPDATE v2_sessions SET compacted_at_message_id = ? WHERE id = ?",
                (message_id, session_id),
            )
            await db.commit()
            logger.debug(f"Compaction boundary set: session={session_id}, message_id={message_id}")

    async def clear_session(self, session_id: str) -> None:
        """
        Clear all messages from a session.

        Args:
            session_id: Session identifier
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("BEGIN IMMEDIATE")
            for sql, parameters in delivery_cleanup_statements(session_id):
                await db.execute(sql, parameters)
            await db.execute("DELETE FROM ngn_web_uploads WHERE session_id = ?", (session_id,))
            await db.execute("DELETE FROM ngn_web_upload_scopes WHERE session_id = ?", (session_id,))
            await db.execute("DELETE FROM v2_messages WHERE session_id = ?", (session_id,))
            await db.execute("UPDATE v2_sessions SET compacted_at_message_id = NULL WHERE id = ?", (session_id,))
            await db.commit()
            logger.debug(f"Cleared session: {session_id}")

    async def delete_session(self, session_id: str) -> None:
        """
        Delete a session and all its messages.

        Args:
            session_id: Session identifier
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("BEGIN IMMEDIATE")
            for sql, parameters in delivery_cleanup_statements(session_id):
                await db.execute(sql, parameters)
            await db.execute("DELETE FROM ngn_web_uploads WHERE session_id = ?", (session_id,))
            await db.execute("DELETE FROM ngn_web_upload_scopes WHERE session_id = ?", (session_id,))
            await db.execute("DELETE FROM v2_messages WHERE session_id = ?", (session_id,))
            await db.execute("DELETE FROM v2_sessions WHERE id = ?", (session_id,))
            await db.commit()
            logger.debug(f"Deleted session: {session_id}")

    async def list_sessions(self, user_id: str | None = None) -> list[dict[str, Any]]:
        """
        List all sessions, optionally filtered by user.

        Args:
            user_id: Optional user ID to filter by

        Returns:
            List of session info dicts
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row

            if user_id:
                cursor = await db.execute(
                    "SELECT * FROM v2_sessions WHERE user_id = ? ORDER BY updated_at DESC",
                    (user_id,),
                )
            else:
                cursor = await db.execute("SELECT * FROM v2_sessions ORDER BY updated_at DESC")

            rows = await cursor.fetchall()
            return [dict(row) for row in rows]

    async def get_message_count(self, session_id: str) -> int:
        """
        Get the number of messages in a session.

        Args:
            session_id: Session identifier

        Returns:
            Number of messages
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute(
                "SELECT COUNT(*) FROM v2_messages WHERE session_id = ?",
                (session_id,),
            )
            row = await cursor.fetchone()
            return row[0] if row else 0

    async def session_exists(self, session_id: str) -> bool:
        """
        Check if a session exists.

        Args:
            session_id: Session identifier

        Returns:
            True if session exists, False otherwise
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            cursor = await db.execute(
                "SELECT 1 FROM v2_sessions WHERE id = ?",
                (session_id,),
            )
            return await cursor.fetchone() is not None

    async def get_session(self, session_id: str) -> dict[str, Any] | None:
        """
        Get session info by ID.

        Args:
            session_id: Session identifier

        Returns:
            Session info dict or None if not found
        """
        await self.initialize()

        async with aiosqlite.connect(self.db_path) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute(
                "SELECT * FROM v2_sessions WHERE id = ?",
                (session_id,),
            )
            row = await cursor.fetchone()
            return dict(row) if row else None

    def _row_to_message(self, row: aiosqlite.Row) -> Message:
        """Convert database row to Message."""
        tool_calls = []
        if row["tool_calls"]:
            try:
                tool_calls_data = json.loads(row["tool_calls"])
                tool_calls = [ToolCall(**tc) for tc in tool_calls_data]
            except (json.JSONDecodeError, TypeError):
                pass

        # Legacy rich content is a JSON array, but ordinary model replies can
        # also be JSON. Decode only complete, recognized content-part records.
        content: str | list[ContentPart] | None = row["content"]
        if isinstance(content, str) and content.startswith("["):
            try:
                parts_data = json.loads(content, object_pairs_hook=_content_object)
                if isinstance(parts_data, list) and parts_data:
                    content = [self._dict_to_content_part(p) for p in parts_data]
            except (ValueError, TypeError, RecursionError):
                pass  # Ordinary or malformed JSON remains the original text.

        return Message(
            role=row["role"],
            content=content,
            tool_calls=tool_calls,
            tool_call_id=row["tool_call_id"],
            name=row["name"],
        )

    @staticmethod
    def _dict_to_content_part(data: object) -> ContentPart:
        """Decode a recognized legacy content part without reinterpreting JSON text."""
        if not isinstance(data, dict):
            raise ValueError("Not a content-part record")
        part_type = data.get("type")
        fields = {
            "text": {"type", "text"},
            "image": {"type", "base64_data", "media_type", "detail"},
            "audio": {"type", "base64_data", "format"},
            "document": {"type", "base64_data", "media_type", "title"},
        }
        if not isinstance(part_type, str) or part_type not in fields or data.keys() - fields[part_type]:
            raise ValueError("Not a recognized content-part record")

        def text(name: str, default: object = None) -> str:
            value: object = data.get(name, default)
            if not isinstance(value, str):
                raise ValueError("Invalid content-part text")
            return value

        def optional_text(name: str) -> str | None:
            value: object = data.get(name)
            if value is not None and not isinstance(value, str):
                raise ValueError("Invalid optional content-part text")
            return value

        if part_type == "text":
            return TextContent(text=text("text"))
        elif part_type == "image":
            return ImageContent(
                base64_data=text("base64_data"),
                media_type=text("media_type", "image/jpeg"),
                detail=optional_text("detail"),
            )
        elif part_type == "audio":
            return AudioContent(
                base64_data=text("base64_data"),
                format=text("format", "wav"),
            )
        return DocumentContent(
            base64_data=text("base64_data"),
            media_type=text("media_type", "application/pdf"),
            title=optional_text("title"),
        )

__init__(db_path)

Initialize the session manager.

Parameters:

Name	Type	Description	Default
db_path	Path	Path to the SQLite database file	required

Source code in src/nagents/session/manager.py

def __init__(self, db_path: Path):
    """
    Initialize the session manager.

    Args:
        db_path: Path to the SQLite database file
    """
    self.db_path = db_path
    self._initialized = False
    self._migration_manager: MigrationManager | None = None

initialize() async

Initialize database and run migrations.

Source code in src/nagents/session/manager.py

async def initialize(self) -> None:
    """Initialize database and run migrations."""
    if self._initialized:
        return

    # Ensure parent directory exists
    self.db_path.parent.mkdir(parents=True, exist_ok=True)

    # Create migration manager with session-specific migrations
    self._migration_manager = MigrationManager(
        self.db_path, db_name="sessions", migrations=list(session_migrations), atomic=True
    )
    await self._migration_manager.initialize()

    self._initialized = True
    logger.debug(f"Session database initialized at {self.db_path}")

get_or_create_session(session_id, user_id) async

Get or create a session.

Parameters:

Name	Type	Description	Default
session_id	str	Unique session identifier	required
user_id	str	User identifier	required

Returns:

Type	Description
str	The session ID

Source code in src/nagents/session/manager.py

async def get_or_create_session(self, session_id: str, user_id: str) -> str:
    """
    Get or create a session.

    Args:
        session_id: Unique session identifier
        user_id: User identifier

    Returns:
        The session ID
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        cursor = await db.execute("SELECT id FROM v2_sessions WHERE id = ?", (session_id,))
        if await cursor.fetchone() is None:
            await db.execute(
                "INSERT INTO v2_sessions (id, user_id) VALUES (?, ?)",
                (session_id, user_id),
            )
            await db.commit()
            logger.debug(f"Created new session: {session_id}")

    return session_id

session_exists(session_id) async

Check if a session exists.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier	required

Returns:

Type	Description
bool	True if session exists, False otherwise

Source code in src/nagents/session/manager.py

async def session_exists(self, session_id: str) -> bool:
    """
    Check if a session exists.

    Args:
        session_id: Session identifier

    Returns:
        True if session exists, False otherwise
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        cursor = await db.execute(
            "SELECT 1 FROM v2_sessions WHERE id = ?",
            (session_id,),
        )
        return await cursor.fetchone() is not None

get_history(session_id, limit=None) async

Get all messages for a session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier	required
limit	int | None	Optional limit on number of messages (most recent)	None

Returns:

Type	Description
list[Message]	List of Message objects in chronological order

Source code in src/nagents/session/manager.py

async def get_history(
    self,
    session_id: str,
    limit: int | None = None,
) -> list[Message]:
    """
    Get all messages for a session.

    Args:
        session_id: Session identifier
        limit: Optional limit on number of messages (most recent)

    Returns:
        List of Message objects in chronological order
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        db.row_factory = aiosqlite.Row

        # Check for compaction boundary
        boundary_cursor = await db.execute(
            "SELECT compacted_at_message_id FROM v2_sessions WHERE id = ?",
            (session_id,),
        )
        boundary_row = await boundary_cursor.fetchone()
        compaction_boundary: int | None = None
        if boundary_row and boundary_row["compacted_at_message_id"] is not None:
            compaction_boundary = boundary_row["compacted_at_message_id"]

        if limit:
            # Get most recent N messages (respecting compaction boundary)
            if compaction_boundary is not None:
                cursor = await db.execute(
                    """SELECT * FROM (
                        SELECT * FROM v2_messages
                        WHERE session_id = ? AND id >= ?
                        ORDER BY id DESC
                        LIMIT ?
                    ) ORDER BY id ASC""",
                    (session_id, compaction_boundary, limit),
                )
            else:
                cursor = await db.execute(
                    """SELECT * FROM (
                        SELECT * FROM v2_messages
                        WHERE session_id = ?
                        ORDER BY id DESC
                        LIMIT ?
                    ) ORDER BY id ASC""",
                    (session_id, limit),
                )
        else:
            if compaction_boundary is not None:
                cursor = await db.execute(
                    "SELECT * FROM v2_messages WHERE session_id = ? AND id >= ? ORDER BY id",
                    (session_id, compaction_boundary),
                )
            else:
                cursor = await db.execute(
                    "SELECT * FROM v2_messages WHERE session_id = ? ORDER BY id",
                    (session_id,),
                )

        rows = await cursor.fetchall()
        return [self._row_to_message(row) for row in rows]

add_message(session_id, message) async

Add a message to session history.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier	required
message	Message	Message to add	required

Returns:

Type	Description
int	The database row ID of the inserted message

Source code in src/nagents/session/manager.py

async def add_message(self, session_id: str, message: Message) -> int:
    """
    Add a message to session history.

    Args:
        session_id: Session identifier
        message: Message to add

    Returns:
        The database row ID of the inserted message
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        cursor = await db.execute(
            """INSERT INTO v2_messages
               (session_id, role, content, tool_calls, tool_call_id, name)
               VALUES (?, ?, ?, ?, ?, ?)""",
            self._message_values(session_id, message),
        )
        message_id = cursor.lastrowid
        assert message_id is not None
        await db.execute(
            "UPDATE v2_sessions SET updated_at = CURRENT_TIMESTAMP WHERE id = ?",
            (session_id,),
        )
        await db.commit()

    return message_id

replace_context(session_id, messages) async

Append a validated replacement and move the inclusive boundary atomically.

Raw history is never deleted. Validation and serialization complete before any writes; failed inserts/boundary updates roll back together.

Source code in src/nagents/session/manager.py

async def replace_context(self, session_id: str, messages: list[Message]) -> None:
    """Append a validated replacement and move the inclusive boundary atomically.

    Raw history is never deleted. Validation and serialization complete
    before any writes; failed inserts/boundary updates roll back together.
    """
    _validate_context(messages)
    values = [self._message_values(session_id, message) for message in messages]
    await self.initialize()
    async with aiosqlite.connect(self.db_path) as db:
        await db.execute("BEGIN IMMEDIATE")
        cursor = await db.execute("SELECT 1 FROM v2_sessions WHERE id = ?", (session_id,))
        if await cursor.fetchone() is None:
            raise ValueError(f"Session '{session_id}' not found")
        first_id: int | None = None
        for row in values:
            cursor = await db.execute(
                """INSERT INTO v2_messages
                   (session_id, role, content, tool_calls, tool_call_id, name)
                   VALUES (?, ?, ?, ?, ?, ?)""",
                row,
            )
            if first_id is None:
                first_id = cursor.lastrowid
        assert first_id is not None
        await db.execute(
            """UPDATE v2_sessions SET compacted_at_message_id = ?,
               updated_at = CURRENT_TIMESTAMP WHERE id = ?""",
            (first_id, session_id),
        )
        await db.commit()

list_sessions(user_id=None) async

List all sessions, optionally filtered by user.

Parameters:

Name	Type	Description	Default
user_id	str | None	Optional user ID to filter by	None

Returns:

Type	Description
list[dict[str, Any]]	List of session info dicts

Source code in src/nagents/session/manager.py

async def list_sessions(self, user_id: str | None = None) -> list[dict[str, Any]]:
    """
    List all sessions, optionally filtered by user.

    Args:
        user_id: Optional user ID to filter by

    Returns:
        List of session info dicts
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        db.row_factory = aiosqlite.Row

        if user_id:
            cursor = await db.execute(
                "SELECT * FROM v2_sessions WHERE user_id = ? ORDER BY updated_at DESC",
                (user_id,),
            )
        else:
            cursor = await db.execute("SELECT * FROM v2_sessions ORDER BY updated_at DESC")

        rows = await cursor.fetchall()
        return [dict(row) for row in rows]

get_message_count(session_id) async

Get the number of messages in a session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier	required

Returns:

Type	Description
int	Number of messages

Source code in src/nagents/session/manager.py

async def get_message_count(self, session_id: str) -> int:
    """
    Get the number of messages in a session.

    Args:
        session_id: Session identifier

    Returns:
        Number of messages
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        cursor = await db.execute(
            "SELECT COUNT(*) FROM v2_messages WHERE session_id = ?",
            (session_id,),
        )
        row = await cursor.fetchone()
        return row[0] if row else 0

clear_session(session_id) async

Clear all messages from a session.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier	required

Source code in src/nagents/session/manager.py

async def clear_session(self, session_id: str) -> None:
    """
    Clear all messages from a session.

    Args:
        session_id: Session identifier
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        await db.execute("BEGIN IMMEDIATE")
        for sql, parameters in delivery_cleanup_statements(session_id):
            await db.execute(sql, parameters)
        await db.execute("DELETE FROM ngn_web_uploads WHERE session_id = ?", (session_id,))
        await db.execute("DELETE FROM ngn_web_upload_scopes WHERE session_id = ?", (session_id,))
        await db.execute("DELETE FROM v2_messages WHERE session_id = ?", (session_id,))
        await db.execute("UPDATE v2_sessions SET compacted_at_message_id = NULL WHERE id = ?", (session_id,))
        await db.commit()
        logger.debug(f"Cleared session: {session_id}")

delete_session(session_id) async

Delete a session and all its messages.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier	required

Source code in src/nagents/session/manager.py

async def delete_session(self, session_id: str) -> None:
    """
    Delete a session and all its messages.

    Args:
        session_id: Session identifier
    """
    await self.initialize()

    async with aiosqlite.connect(self.db_path) as db:
        await db.execute("BEGIN IMMEDIATE")
        for sql, parameters in delivery_cleanup_statements(session_id):
            await db.execute(sql, parameters)
        await db.execute("DELETE FROM ngn_web_uploads WHERE session_id = ?", (session_id,))
        await db.execute("DELETE FROM ngn_web_upload_scopes WHERE session_id = ?", (session_id,))
        await db.execute("DELETE FROM v2_messages WHERE session_id = ?", (session_id,))
        await db.execute("DELETE FROM v2_sessions WHERE id = ?", (session_id,))
        await db.commit()
        logger.debug(f"Deleted session: {session_id}")