orchestrata_ultimate.cxp.threads

Reference › orchestrata_ultimate › cxp › threads

ThreadManager groups the packets of a multi-turn exchange into conversation threads (Spec 004 US-008) and implements the IThreadManager contract from interfaces. It keeps the live threads and their ThreadMetadata (from types) in memory and saves every change to a thread_store. See cxp.

create_thread uses the first packet's thread id (or generates one), records sender and recipient as participants and the subject, and raises ThreadError if the id already exists. add_to_thread appends a packet, updates activity, message count and participants, and raises ThreadNotFoundError for an unknown thread and ThreadClosedError for a resolved or archived one. close_thread sets the status to resolved, and raises ThreadError for an archived thread; reopen_thread reverses it. archive_stale_threads archives threads whose last activity is older than auto_archive_days (30 by default) and returns the count. get_thread_packets, get_thread_metadata, list_active_threads, get_thread_count and thread_exists read state.

At a glance

Class diagram

Dependency diagram

API

Classes

ThreadManager

Decorators: traces(spec='004', story='US-008', reqs=['REQ-004-PERF-013', 'REQ-004-US-008-014']) · Line 41

Manages CXP conversation threads.

Source

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

"""
CXP Thread Manager — Conversation thread management.

Implements thread lifecycle per Spec 004 Section US-008:
  create, add, close, reopen, auto-archive (30 days).

Thread packets are retrievable in chronological order (< 2ms target).
Thread metadata (ThreadMetadata from types.py) is tracked per thread.

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

import logging
import uuid
from datetime import UTC, datetime, timedelta

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.cxp.exceptions import (
    ThreadClosedError,
    ThreadError,
    ThreadNotFoundError,
)
from orchestrata_ultimate.cxp.thread_store import ThreadStore
from orchestrata_ultimate.cxp.types import ThreadMetadata, ThreadStatus

logger = logging.getLogger(__name__)

__all__ = ["ThreadManager"]


@traces(
    spec="004",
    story="US-008",
    reqs=[
        "REQ-004-PERF-013",
        "REQ-004-US-008-014",
    ],
)
class ThreadManager:
    """Manages CXP conversation threads.

    Each thread links related CXP packets into a conversation with
    tracked metadata (participants, message count, status, timestamps).

    Threads follow a lifecycle:
        ACTIVE -> RESOLVED (close) -> ACTIVE (reopen)
        ACTIVE -> ARCHIVED (auto-archive after inactivity)
        RESOLVED -> ARCHIVED (auto-archive after inactivity)

    Args:
        store: Optional ThreadStore backend. Defaults to a new in-memory store.
        auto_archive_days: Days of inactivity before a thread is archived.
            Defaults to 30.
    """

    def __init__(
        self,
        store: ThreadStore | None = None,
        auto_archive_days: int = 30,
    ) -> None:
        self._store = store or ThreadStore()
        self._auto_archive_days = auto_archive_days
        self._threads: dict[str, list[CXPPacket]] = {}
        self._metadata: dict[str, ThreadMetadata] = {}

    @traces(
        spec="004",
        story="US-008",
        reqs=[
            "REQ-004-US-008-001",
            "REQ-004-US-008-002",
            "REQ-004-US-008-011",
            "REQ-004-US-008-012",
            "REQ-004-US-008-015",
            "REQ-004-US-008-016",
        ],
    )
    def create_thread(self, initial_packet: CXPPacket) -> str:
        """Create a new thread from an initial packet.

        Uses the packet's thread_id as the thread identifier. Creates
        metadata tracking participants, subject, and timestamps.

        Args:
            initial_packet: The first packet in the conversation.

        Returns:
            The thread_id string.

        Raises:
            ThreadError: If a thread with this ID already exists.
        """
        thread_id = initial_packet.thread_id
        if not thread_id:
            thread_id = str(uuid.uuid4())

        if thread_id in self._threads:
            raise ThreadError(
                message=f"Thread already exists: {thread_id}",
                thread_id=thread_id,
            )

        now = datetime.now(UTC)
        metadata = ThreadMetadata(
            thread_id=thread_id,
            created_at=now,
            last_activity=now,
            participants=[],
            message_count=1,
            status=ThreadStatus.ACTIVE,
            subject=initial_packet.subject,
        )

        # Track participants from the initial packet
        metadata.add_participant(initial_packet.from_agent)
        metadata.add_participant(initial_packet.to_agent)

        self._threads[thread_id] = [initial_packet]
        self._metadata[thread_id] = metadata

        # Persist to store
        self._store.save_thread(thread_id, self._threads[thread_id], metadata)

        return thread_id

    @traces(
        spec="004",
        story="US-008",
        reqs=["REQ-004-US-008-017"],
    )
    def add_to_thread(self, thread_id: str, packet: CXPPacket) -> None:
        """Add a packet to an existing thread.

        Updates metadata: last_activity, message_count, participants.

        Args:
            thread_id: The thread to add the packet to.
            packet: The packet to append.

        Raises:
            ThreadNotFoundError: If the thread does not exist.
            ThreadClosedError: If the thread status is RESOLVED or ARCHIVED.
        """
        self._require_thread(thread_id)
        metadata = self._metadata[thread_id]

        if metadata.status == ThreadStatus.RESOLVED:
            raise ThreadClosedError(thread_id)
        if metadata.status == ThreadStatus.ARCHIVED:
            raise ThreadClosedError(thread_id)

        self._threads[thread_id].append(packet)
        metadata.touch()
        metadata.add_participant(packet.from_agent)
        metadata.add_participant(packet.to_agent)

        # Persist updated state
        self._store.save_thread(thread_id, self._threads[thread_id], metadata)

    @traces(
        spec="004",
        story="US-008",
        reqs=[
            "REQ-004-PERF-009",
            "REQ-004-US-008-005",
            "REQ-004-US-008-006",
            "REQ-004-US-008-010",
        ],
    )
    def get_thread_packets(self, thread_id: str) -> list[CXPPacket]:
        """Retrieve all packets in a thread in chronological order.

        Performance target: < 2ms.

        Args:
            thread_id: The thread to retrieve packets from.

        Returns:
            List of CXPPacket objects sorted by timestamp.

        Raises:
            ThreadNotFoundError: If the thread does not exist.
        """
        self._require_thread(thread_id)
        packets = list(self._threads[thread_id])
        packets.sort(key=lambda p: p.timestamp)
        return packets

    def get_thread_metadata(self, thread_id: str) -> ThreadMetadata:
        """Retrieve metadata for a thread.

        Args:
            thread_id: The thread to get metadata for.

        Returns:
            ThreadMetadata instance.

        Raises:
            ThreadNotFoundError: If the thread does not exist.
        """
        self._require_thread(thread_id)
        return self._metadata[thread_id]

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-018"])
    def close_thread(self, thread_id: str) -> None:
        """Close a thread so no new packets are accepted.

        Sets thread status to RESOLVED.

        Args:
            thread_id: The thread to close.

        Raises:
            ThreadNotFoundError: If the thread does not exist.
            ThreadError: If the thread is already archived.
        """
        self._require_thread(thread_id)
        metadata = self._metadata[thread_id]

        if metadata.status == ThreadStatus.ARCHIVED:
            raise ThreadError(
                message=f"Cannot close archived thread: {thread_id}",
                thread_id=thread_id,
            )

        metadata.status = ThreadStatus.RESOLVED
        self._store.save_thread(thread_id, self._threads[thread_id], metadata)

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-019"])
    def reopen_thread(self, thread_id: str) -> None:
        """Reopen a previously closed thread.

        Sets thread status back to ACTIVE. Archived threads cannot
        be reopened.

        Args:
            thread_id: The thread to reopen.

        Raises:
            ThreadNotFoundError: If the thread does not exist.
            ThreadError: If the thread is archived (cannot reopen).
        """
        self._require_thread(thread_id)
        metadata = self._metadata[thread_id]

        if metadata.status == ThreadStatus.ARCHIVED:
            raise ThreadError(
                message=f"Cannot reopen archived thread: {thread_id}",
                thread_id=thread_id,
            )

        metadata.status = ThreadStatus.ACTIVE
        metadata.last_activity = datetime.now(UTC)
        self._store.save_thread(thread_id, self._threads[thread_id], metadata)

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-020"])
    def archive_stale_threads(self) -> int:
        """Archive threads that have been inactive longer than auto_archive_days.

        Only ACTIVE and RESOLVED threads are candidates for archival.
        Already-archived threads are skipped.

        Returns:
            Number of threads archived.
        """
        cutoff = datetime.now(UTC) - timedelta(days=self._auto_archive_days)
        archived_count = 0

        for thread_id, metadata in list(self._metadata.items()):
            if metadata.status == ThreadStatus.ARCHIVED:
                continue
            if metadata.last_activity < cutoff:
                metadata.status = ThreadStatus.ARCHIVED
                self._store.save_thread(thread_id, self._threads[thread_id], metadata)
                archived_count += 1

        return archived_count

    def list_active_threads(self) -> list[str]:
        """Return IDs of all threads with ACTIVE status.

        Returns:
            List of thread_id strings.
        """
        return [
            tid
            for tid, meta in self._metadata.items()
            if meta.status == ThreadStatus.ACTIVE
        ]

    def get_thread_count(self) -> int:
        """Return the total number of threads (all statuses).

        Returns:
            Integer count.
        """
        return len(self._threads)

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

        Args:
            thread_id: The thread to check.

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

    # -- Internal helpers --

    def _require_thread(self, thread_id: str) -> None:
        """Raise ThreadNotFoundError if thread_id is not tracked."""
        if thread_id not in self._threads:
            raise ThreadNotFoundError(thread_id)