orchestrata_ultimate.cxp.thread_store

Reference › orchestrata_ultimate › cxp › thread_store

ThreadStore is the persistence layer that threads writes to, and it is an in-memory backend: packets and ThreadMetadata are held in two dictionaries keyed by thread id. save_thread overwrites the stored copy of a thread, load_thread returns the packets and metadata as a pair, and delete_thread removes a thread; both raise ThreadNotFoundError for an unknown id. list_threads and thread_exists report what is stored. Nothing is written to disk. See types, exceptions and cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

ThreadStore

Line 26

In-memory persistence backend for CXP threads.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/cxp/thread_store.py

"""
CXP Thread Store — Persistence layer for thread data.

Provides in-memory storage for CXP conversation threads.
Designed as a pluggable backend: swap this implementation
for Redis, SQLite, or any other store without changing ThreadManager.

Spec: 004-cxp, Section US-008
Author: Mohamed
"""

import logging

from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.cxp.exceptions import ThreadNotFoundError
from orchestrata_ultimate.cxp.types import ThreadMetadata

# Spec: 004 - CXP Protocol


logger = logging.getLogger(__name__)

__all__ = ["ThreadStore"]


class ThreadStore:
    """In-memory persistence backend for CXP threads.

    Stores packets and metadata per thread_id. All data is held
    in dictionaries keyed by thread_id.
    """

    def __init__(self) -> None:
        self._packets: dict[str, list[CXPPacket]] = {}
        self._metadata: dict[str, ThreadMetadata] = {}

    def save_thread(
        self,
        thread_id: str,
        packets: list[CXPPacket],
        metadata: ThreadMetadata,
    ) -> None:
        """Persist thread packets and metadata.

        Overwrites any existing data for the given thread_id.

        Args:
            thread_id: Unique thread identifier.
            packets: Ordered list of packets in the thread.
            metadata: Thread metadata to persist.
        """
        self._packets[thread_id] = list(packets)
        self._metadata[thread_id] = metadata

    def load_thread(
        self,
        thread_id: str,
    ) -> tuple[list[CXPPacket], ThreadMetadata]:
        """Load thread packets and metadata.

        Args:
            thread_id: Unique thread identifier.

        Returns:
            Tuple of (packets, metadata).

        Raises:
            ThreadNotFoundError: If the thread does not exist.
        """
        if thread_id not in self._packets:
            raise ThreadNotFoundError(thread_id)

        packets = list(self._packets[thread_id])
        metadata = self._metadata[thread_id]
        return packets, metadata

    def delete_thread(self, thread_id: str) -> None:
        """Remove a thread and all its data.

        Args:
            thread_id: Unique thread identifier.

        Raises:
            ThreadNotFoundError: If the thread does not exist.
        """
        if thread_id not in self._packets:
            raise ThreadNotFoundError(thread_id)

        del self._packets[thread_id]
        del self._metadata[thread_id]

    def list_threads(self) -> list[str]:
        """Return all stored thread IDs.

        Returns:
            List of thread_id strings.
        """
        return list(self._packets.keys())

    def thread_exists(self, thread_id: str) -> bool:
        """Check whether a thread exists in the store.

        Args:
            thread_id: Unique thread identifier.

        Returns:
            True if the thread exists, False otherwise.
        """
        return thread_id in self._packets