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

"""Approval repository for the MAF roundtable experiment.

Pending human-approval requests for high-risk external writes. Phase 1 ships
the storage + CRUD + decision transition; the actual pause/resume wiring
(checkpoint restore after a decision) is Phase 5.
"""

from __future__ import annotations

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

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

from deerflow.persistence.maf_roundtable.model import MafRoundtableApprovalRow


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


def _to_dict(row: MafRoundtableApprovalRow) -> dict[str, Any]:
    return {
        "id": row.id,
        "session_id": row.session_id,
        "run_id": row.run_id,
        "action_id": row.action_id,
        "requested_by_agent_id": row.requested_by_agent_id,
        "status": row.status,
        "summary": row.summary,
        "payload": row.payload_redacted_json,
        "decided_by": row.decided_by,
        "decided_at": row.decided_at.isoformat() if isinstance(row.decided_at, datetime) else row.decided_at,
        "expires_at": row.expires_at.isoformat() if isinstance(row.expires_at, datetime) else row.expires_at,
        "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,
    }


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

    async def create_approval(self, data: dict[str, Any]) -> dict[str, Any]:
        now = _now()
        row = MafRoundtableApprovalRow(
            id=data.get("id") or uuid4().hex,
            session_id=str(data["session_id"]),
            run_id=data.get("run_id"),
            action_id=data.get("action_id"),
            requested_by_agent_id=data.get("requested_by_agent_id"),
            status=str(data.get("status") or "pending"),
            summary=str(data.get("summary") or ""),
            payload_redacted_json=data.get("payload"),
            decided_by=data.get("decided_by"),
            decided_at=data.get("decided_at"),
            expires_at=data.get("expires_at"),
            created_at=now,
            updated_at=now,
        )
        async with self._sf() as session:
            session.add(row)
            await session.commit()
        return await self.get_approval(row.id)  # type: ignore[arg-type]

    async def get_approval(self, approval_id: str) -> dict[str, Any] | None:
        async with self._sf() as session:
            stmt = select(MafRoundtableApprovalRow).where(MafRoundtableApprovalRow.id == approval_id).limit(1)
            row = (await session.execute(stmt)).scalars().first()
            return _to_dict(row) if row is not None else None

    async def list_approvals(
        self, *, run_id: str | None = None, status: str | None = None
    ) -> list[dict[str, Any]]:
        stmt = select(MafRoundtableApprovalRow)
        if run_id is not None:
            stmt = stmt.where(MafRoundtableApprovalRow.run_id == run_id)
        if status is not None:
            stmt = stmt.where(MafRoundtableApprovalRow.status == status)
        stmt = stmt.order_by(MafRoundtableApprovalRow.created_at.asc())
        async with self._sf() as session:
            rows = (await session.execute(stmt)).scalars().all()
            return [_to_dict(r) for r in rows]

    async def decide(
        self,
        approval_id: str,
        *,
        decision: str,
        decided_by: str | None = None,
    ) -> dict[str, Any] | None:
        """Record a decision atomically. ``decision`` is ``approved`` /
        ``rejected`` / ``expired``.

        The transition is a single conditional UPDATE:
        ``UPDATE … SET status=:decision … WHERE id=:id AND status='pending'``.
        ``rowcount == 1`` means this caller won the race; ``rowcount == 0``
        means the approval is missing **or** was already decided by someone
        else, which the caller surfaces as a conflict. This guarantees a
        decision can never be overwritten by a concurrent (or later) request,
        and the winning approver + timestamp are the ones persisted.
        """
        if decision not in ("approved", "rejected", "expired"):
            raise ValueError(f"invalid approval decision: {decision}")
        now = _now()
        async with self._sf() as session:
            result = await session.execute(
                update(MafRoundtableApprovalRow)
                .where(
                    MafRoundtableApprovalRow.id == approval_id,
                    MafRoundtableApprovalRow.status == "pending",
                )
                .values(
                    status=decision,
                    decided_by=decided_by,
                    decided_at=now,
                    updated_at=now,
                )
            )
            await session.commit()
            if result.rowcount == 1:  # type: ignore[union-attr]
                # This caller won the race — the decision is persisted.
                return await self.get_approval(approval_id)
        # rowcount == 0: either the approval is missing, or it was already
        # decided by a concurrent request. Distinguish so the service layer can
        # map missing→404 and already-decided→409.
        existing = await self.get_approval(approval_id)
        if existing is None:
            return None
        raise ValueError(f"approval already decided: {existing['status']}")