deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/persistence/maf_roundtable/conversations.py

"""Conversation repository for the MAF roundtable V2 (user-visible aggregation).

A conversation is the **only** entity the frontend knows about. Owner-scoped
by ``owner_user_id``: every read / update / delete checks ownership. Admins
bypass the owner filter when ``admin_diagnostic_access`` is true (handled in
the service / router layer, not here).
"""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Any
from uuid import uuid4

from sqlalchemy import delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow


def _now() -> datetime:
    return datetime.now(UTC)


def conversation_internal_to_dict(row: MafRoundtableConversationRow) -> dict[str, Any]:
    """Internal serialisation — includes all columns. Never expose to the browser."""
    return {
        "id": row.id,
        "owner_user_id": row.owner_user_id,
        "title": row.title,
        "task_context": row.task_context_json,
        "safe_status": row.safe_status,
        "active_run_id": row.active_run_id,
        "internal_session_id": row.internal_session_id,
        "last_stream_seq": row.last_stream_seq,
        "created_at": row.created_at.isoformat() if isinstance(row.created_at, datetime) else row.created_at,
        "updated_at": row.updated_at.isoformat() if isinstance(row.updated_at, datetime) else row.updated_at,
    }


_TASK_CONTEXT_SAFE_KEYS = frozenset({"business_code", "task_id", "task_name", "chain_id", "chain_name"})


def conversation_public_to_dict(row: MafRoundtableConversationRow) -> dict[str, Any]:
    """Public DTO — only fields safe for the browser. No owner, session, run, or thread ids."""
    raw_ctx = row.task_context_json
    safe_ctx: dict[str, Any] | None = None
    if isinstance(raw_ctx, dict):
        safe_ctx = {k: v for k, v in raw_ctx.items() if k in _TASK_CONTEXT_SAFE_KEYS}
    return {
        "id": row.id,
        "title": row.title,
        "task_context": safe_ctx,
        "safe_status": row.safe_status,
        "created_at": row.created_at.isoformat() if isinstance(row.created_at, datetime) else row.created_at,
        "updated_at": row.updated_at.isoformat() if isinstance(row.updated_at, datetime) else row.updated_at,
        "last_stream_seq": row.last_stream_seq,
    }


class MafConversationRepository:
    """Owner-scoped CRUD for MAF roundtable conversations."""

    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
        self._sf = session_factory

    async def create_conversation(self, data: dict[str, Any]) -> dict[str, Any]:
        """Create a new conversation. ``data`` must include ``owner_user_id``."""
        now = _now()
        row = MafRoundtableConversationRow(
            id=data.get("id") or uuid4().hex,
            owner_user_id=str(data["owner_user_id"]),
            title=str(data.get("title") or ""),
            task_context_json=data.get("task_context"),
            safe_status="idle",
            active_run_id=None,
            internal_session_id=None,
            last_stream_seq=0,
            created_at=now,
            updated_at=now,
        )
        async with self._sf() as session:
            session.add(row)
            await session.commit()
        return conversation_internal_to_dict(row)

    async def get_conversation(
        self, conversation_id: str, *, user_id: str | None = "REQUIRED"
    ) -> dict[str, Any] | None:
        """Read one conversation.

        ``user_id`` scoping:
        - ``"REQUIRED"`` (default): enforce owner scope.
        - ``None``: bypass owner scope (admin / internal path).
        """
        async with self._sf() as session:
            stmt = select(MafRoundtableConversationRow).where(
                MafRoundtableConversationRow.id == conversation_id
            ).limit(1)
            row = (await session.execute(stmt)).scalars().first()
            if row is None:
                return None
            if user_id is not None and row.owner_user_id != user_id:
                return None
            return conversation_internal_to_dict(row)

    async def list_conversations(
        self,
        *,
        user_id: str | None = None,
        safe_status: str | None = None,
        limit: int = 50,
        offset: int = 0,
    ) -> list[dict[str, Any]]:
        """List conversations, newest-first.

        ``user_id=None`` lists across all users (admin / diagnostic).
        """
        stmt = select(MafRoundtableConversationRow)
        if user_id is not None:
            stmt = stmt.where(MafRoundtableConversationRow.owner_user_id == user_id)
        if safe_status is not None:
            stmt = stmt.where(MafRoundtableConversationRow.safe_status == safe_status)
        stmt = stmt.order_by(MafRoundtableConversationRow.updated_at.desc()).limit(limit).offset(offset)
        async with self._sf() as session:
            rows = (await session.execute(stmt)).scalars().all()
            return [conversation_internal_to_dict(r) for r in rows]

    async def count_conversations(
        self, *, user_id: str | None = None, safe_status: str | None = None
    ) -> int:
        stmt = select(func.count()).select_from(MafRoundtableConversationRow)
        if user_id is not None:
            stmt = stmt.where(MafRoundtableConversationRow.owner_user_id == user_id)
        if safe_status is not None:
            stmt = stmt.where(MafRoundtableConversationRow.safe_status == safe_status)
        async with self._sf() as session:
            return int((await session.execute(stmt)).scalar_one())

    async def update_conversation(
        self, conversation_id: str, *, user_id: str | None, fields: dict[str, Any]
    ) -> dict[str, Any] | None:
        """Partial update. Returns ``None`` if missing or wrong owner."""
        async with self._sf() as session:
            stmt = select(MafRoundtableConversationRow).where(
                MafRoundtableConversationRow.id == conversation_id
            ).limit(1)
            row = (await session.execute(stmt)).scalars().first()
            if row is None:
                return None
            if user_id is not None and row.owner_user_id != user_id:
                return None
            col_map = {"task_context": "task_context_json"}
            for key in ("title", "task_context", "safe_status", "active_run_id",
                        "internal_session_id", "last_stream_seq"):
                if key in fields:
                    col = col_map.get(key, key)
                    setattr(row, col, fields[key])
            row.updated_at = _now()
            await session.commit()
            return conversation_internal_to_dict(row)

    async def delete_conversation(self, conversation_id: str, *, user_id: str | None) -> bool:
        """Delete a conversation (children via FK CASCADE)."""
        async with self._sf() as session:
            stmt = select(MafRoundtableConversationRow).where(
                MafRoundtableConversationRow.id == conversation_id
            ).limit(1)
            row = (await session.execute(stmt)).scalars().first()
            if row is None:
                return False
            if user_id is not None and row.owner_user_id != user_id:
                return False
            await session.execute(
                delete(MafRoundtableConversationRow).where(
                    MafRoundtableConversationRow.id == conversation_id
                )
            )
            await session.commit()
            return True

    async def set_status(self, conversation_id: str, safe_status: str) -> None:
        """Internal: update only the safe_status column."""
        async with self._sf() as session:
            await session.execute(
                update(MafRoundtableConversationRow)
                .where(MafRoundtableConversationRow.id == conversation_id)
                .values(safe_status=safe_status, updated_at=_now())
            )
            await session.commit()

    async def set_active_run(self, conversation_id: str, run_id: str | None, safe_status: str) -> None:
        """Internal: update active_run_id + safe_status together."""
        async with self._sf() as session:
            await session.execute(
                update(MafRoundtableConversationRow)
                .where(MafRoundtableConversationRow.id == conversation_id)
                .values(active_run_id=run_id, safe_status=safe_status, updated_at=_now())
            )
            await session.commit()