"""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']}")