diff --git a/backend/package/yuxi/channel/session/__init__.py b/backend/package/yuxi/channel/session/__init__.py new file mode 100644 index 00000000..3f65f21f --- /dev/null +++ b/backend/package/yuxi/channel/session/__init__.py @@ -0,0 +1,6 @@ +from yuxi.channel.session.recorder import SessionOpResult, SessionRecorder + +__all__ = [ + "SessionOpResult", + "SessionRecorder", +] diff --git a/backend/package/yuxi/channel/session/recorder.py b/backend/package/yuxi/channel/session/recorder.py new file mode 100644 index 00000000..67c26c1f --- /dev/null +++ b/backend/package/yuxi/channel/session/recorder.py @@ -0,0 +1,293 @@ +import logging +from collections.abc import Callable +from dataclasses import dataclass + +from yuxi.channel.message.models import DispatchResult, UnifiedMessage +from yuxi.repositories.conversation_repository import ConversationRepository +from yuxi.utils.datetime_utils import utc_isoformat + +logger = logging.getLogger(__name__) + +_CONVERSATION_TERMINAL_STATUSES = frozenset({"closed", "archived", "deleted"}) + + +def is_terminal_status(status: str) -> bool: + return status in _CONVERSATION_TERMINAL_STATUSES + + +_METADATA_SESSION_KEY = "session_key" +_METADATA_SOURCE = "source" +_METADATA_SOURCE_CHANNEL = "channel" +_METADATA_CHANNEL_TYPE = "channel_type" +_METADATA_ACCOUNT_ID = "account_id" +_METADATA_PEER_ID = "peer_id" +_METADATA_PEER_KIND = "peer_kind" +_METADATA_LAST_MSG_ID = "last_msg_id" +_METADATA_FIRST_INTERACTION_AT = "first_interaction_at" +_METADATA_LAST_INTERACTION_AT = "last_interaction_at" +_METADATA_RECORDED_AT = "recorded_at" +_METADATA_CLOSED_AT = "closed_at" +_METADATA_CLOSE_REASON = "close_reason" + + +def _normalize_session_key(session_key: str) -> str: + return session_key.strip().lower() + + +def _build_base_metadata(msg: UnifiedMessage, session_key: str) -> dict: + return { + _METADATA_SOURCE: _METADATA_SOURCE_CHANNEL, + _METADATA_CHANNEL_TYPE: msg.channel_type, + _METADATA_ACCOUNT_ID: msg.account_id, + _METADATA_PEER_ID: msg.sender.id, + _METADATA_PEER_KIND: msg.sender.kind.value, + _METADATA_SESSION_KEY: _normalize_session_key(session_key), + } + + +def _build_create_metadata(msg: UnifiedMessage, session_key: str) -> dict: + now = utc_isoformat() + meta = _build_base_metadata(msg, session_key) + meta[_METADATA_LAST_MSG_ID] = msg.msg_id + meta[_METADATA_FIRST_INTERACTION_AT] = now + meta[_METADATA_LAST_INTERACTION_AT] = now + meta[_METADATA_RECORDED_AT] = now + return meta + + +@dataclass +class SessionOpResult: + success: bool + thread_id: str + error: str | None = None + found: bool = True + + +class SessionRecorder: + def __init__(self, conv_repo: ConversationRepository): + self._conv_repo = conv_repo + + async def record( + self, + msg: UnifiedMessage, + dispatch_result: DispatchResult, + session_key: str, + user_id: str = "channel_user", + title: str | None = None, + on_error: Callable[[Exception], None] | None = None, + track_task: Callable[[object], None] | None = None, + ) -> None: + task = self._create_conversation_async( + msg, + dispatch_result, + session_key, + user_id=user_id, + title=title, + on_error=on_error, + ) + if track_task is not None: + track_task(task) + + async def record_or_update( + self, + msg: UnifiedMessage, + dispatch_result: DispatchResult, + session_key: str, + user_id: str = "channel_user", + title: str | None = None, + on_error: Callable[[Exception], None] | None = None, + ) -> bool: + try: + conv = await self._conv_repo.get_conversation_by_thread_id( + dispatch_result.thread_id + ) + if conv is not None and is_terminal_status(conv.status): + logger.debug( + "Session record_or_update skipped: thread=%s status=%s", + dispatch_result.thread_id, + conv.status, + ) + return False + + base_meta = _build_create_metadata(msg, session_key) + update_meta = { + _METADATA_LAST_MSG_ID: msg.msg_id, + _METADATA_LAST_INTERACTION_AT: utc_isoformat(), + } + + await self._conv_repo.upsert_conversation( + thread_id=dispatch_result.thread_id, + user_id=user_id, + agent_id=str(dispatch_result.agent_config_id), + title=title, + metadata=base_meta, + update_metadata=update_meta, + ) + logger.info( + "Session upserted: msg=%s channel=%s thread=%s", + msg.msg_id, + msg.channel_type, + dispatch_result.thread_id, + ) + return True + except Exception as exc: + if on_error is not None: + on_error(exc) + logger.exception( + "Session record_or_update failed: msg=%s thread=%s", + msg.msg_id, + dispatch_result.thread_id, + ) + return False + + async def close( + self, + thread_id: str, + reason: str | None = None, + on_error: Callable[[Exception], None] | None = None, + ) -> SessionOpResult: + try: + meta = {_METADATA_CLOSED_AT: utc_isoformat()} + if reason: + meta[_METADATA_CLOSE_REASON] = reason + + updated = await self._conv_repo.update_conversation( + thread_id, + status="closed", + metadata=meta, + ) + if updated is None: + return SessionOpResult( + success=False, thread_id=thread_id, error="not found", found=False + ) + logger.info("Session closed: thread=%s", thread_id) + return SessionOpResult(success=True, thread_id=thread_id) + except Exception as exc: + if on_error is not None: + on_error(exc) + logger.exception("Session close failed: thread=%s", thread_id) + return SessionOpResult(success=False, thread_id=thread_id, error=str(exc)) + + async def reopen( + self, + thread_id: str, + on_error: Callable[[Exception], None] | None = None, + ) -> SessionOpResult: + try: + updated = await self._conv_repo.update_conversation( + thread_id, + status="active", + metadata={_METADATA_LAST_INTERACTION_AT: utc_isoformat()}, + ) + if updated is None: + return SessionOpResult( + success=False, thread_id=thread_id, error="not found", found=False + ) + logger.info("Session reopened: thread=%s", thread_id) + return SessionOpResult(success=True, thread_id=thread_id) + except Exception as exc: + if on_error is not None: + on_error(exc) + logger.exception("Session reopen failed: thread=%s", thread_id) + return SessionOpResult(success=False, thread_id=thread_id, error=str(exc)) + + async def touch( + self, + msg: UnifiedMessage, + thread_id: str, + on_error: Callable[[Exception], None] | None = None, + ) -> bool: + try: + conv = await self._conv_repo.get_conversation_by_thread_id(thread_id) + if conv is None: + return False + if is_terminal_status(conv.status): + logger.debug( + "Session touch skipped: thread=%s status=%s", + thread_id, + conv.status, + ) + return False + await self._conv_repo.update_conversation( + thread_id, + metadata={ + _METADATA_LAST_MSG_ID: msg.msg_id, + _METADATA_LAST_INTERACTION_AT: utc_isoformat(), + }, + ) + return True + except Exception as exc: + if on_error is not None: + on_error(exc) + logger.exception("Session touch failed: thread=%s", thread_id) + return False + + async def find_by_session_key( + self, + session_key: str, + on_error: Callable[[Exception], None] | None = None, + ) -> list: + try: + normalized = _normalize_session_key(session_key) + return await self._conv_repo.get_conversations_by_metadata_key( + _METADATA_SESSION_KEY, normalized + ) + except Exception as exc: + if on_error is not None: + on_error(exc) + logger.exception("Session find_by_session_key failed: session_key=%s", session_key) + return [] + + async def list_by_channel( + self, + channel_type: str, + account_id: str | None = None, + on_error: Callable[[Exception], None] | None = None, + ) -> list: + try: + return await self._conv_repo.get_conversations_by_channel( + channel_type, account_id + ) + except Exception as exc: + if on_error is not None: + on_error(exc) + logger.exception( + "Session list_by_channel failed: channel=%s account=%s", + channel_type, + account_id, + ) + return [] + + async def _create_conversation_async( + self, + msg: UnifiedMessage, + dispatch_result: DispatchResult, + session_key: str, + user_id: str, + title: str | None, + on_error: Callable[[Exception], None] | None, + ) -> None: + try: + metadata = _build_create_metadata(msg, session_key) + + await self._conv_repo.upsert_conversation( + thread_id=dispatch_result.thread_id, + user_id=user_id, + agent_id=str(dispatch_result.agent_config_id), + title=title, + metadata=metadata, + ) + logger.info( + "Session recorded: msg=%s channel=%s thread=%s", + msg.msg_id, + msg.channel_type, + dispatch_result.thread_id, + ) + except Exception as exc: + if on_error is not None: + on_error(exc) + logger.exception( + "Failed to record session: msg=%s thread=%s", + msg.msg_id, + dispatch_result.thread_id, + )