Confirmation loops for high-stakes operations

# openai>=1.0.0, pydantic>=2.0, tenacity>=8.0
import os
import uuid
import time
import logging
from enum import Enum
from datetime import datetime, timezone
from pydantic import BaseModel
from tenacity import retry, stop_after_attempt, wait_exponential

log = logging.getLogger("agent.confirmation")
logging.basicConfig(level=logging.INFO, format='{"time":"%(asctime)s","level":"%(levelname)s","msg":%(message)s}')

class RiskTier(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class ConfirmationStatus(str, Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    TIMED_OUT = "timed_out"

class ActionProposal(BaseModel):
    action_id: str
    description: str
    reasoning: str
    affected_resources: list[str]
    reversible: bool
    risk_tier: RiskTier
    requested_at: str

class ConfirmationResult(BaseModel):
    action_id: str
    status: ConfirmationStatus
    approved_by: str | None = None
    resolved_at: str | None = None

# Approval store -- replace with Redis or a DB in real deployments
_pending_approvals: dict[str, ConfirmationResult] = {}

TIMEOUT_SECONDS = int(os.environ.get("CONFIRMATION_TIMEOUT_SECONDS", "300"))
REQUIRED_APPROVERS = {"high": 2, "medium": 1, "low": 0}

def classify_action(description: str, reversible: bool, affected_count: int) -> RiskTier:
    if not reversible and affected_count > 100:
        return RiskTier.HIGH
    if not reversible or affected_count > 10:
        return RiskTier.MEDIUM
    return RiskTier.LOW

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def _deliver_notification(proposal: ActionProposal) -> None:
    """Deliver via your real channel: Slack webhook, PagerDuty, etc."""
    webhook_url = os.environ.get("APPROVAL_WEBHOOK_URL")
    if not webhook_url:
        raise EnvironmentError("APPROVAL_WEBHOOK_URL not set")
    # import httpx; httpx.post(webhook_url, json=proposal.model_dump(), timeout=10)
    log.info(f'"event":"notification_sent","action_id":"{proposal.action_id}","tier":"{proposal.risk_tier}"')

def request_confirmation(proposal: ActionProposal) -> ConfirmationResult:
    result = ConfirmationResult(action_id=proposal.action_id, status=ConfirmationStatus.PENDING)
    _pending_approvals[proposal.action_id] = result

    if proposal.risk_tier == RiskTier.LOW:
        result.status = ConfirmationStatus.APPROVED
        result.approved_by = "auto"
        result.resolved_at = datetime.now(timezone.utc).isoformat()
        log.info(f'"event":"auto_approved","action_id":"{proposal.action_id}"')
        return result

    try:
        _deliver_notification(proposal)
    except Exception as exc:
        log.error(f'"event":"notification_failed","action_id":"{proposal.action_id}","error":"{exc}"')
        result.status = ConfirmationStatus.REJECTED
        result.resolved_at = datetime.now(timezone.utc).isoformat()
        return result

    deadline = time.monotonic() + TIMEOUT_SECONDS
    poll_interval = 5
    while time.monotonic() < deadline:
        current = _pending_approvals.get(proposal.action_id)
        if current and current.status != ConfirmationStatus.PENDING:
            log.info(f'"event":"resolved","action_id":"{proposal.action_id}","status":"{current.status}"')
            return current
        time.sleep(poll_interval)

    result.status = ConfirmationStatus.TIMED_OUT
    result.resolved_at = datetime.now(timezone.utc).isoformat()
    _pending_approvals[proposal.action_id] = result
    log.warning(f'"event":"timed_out","action_id":"{proposal.action_id}","timeout_seconds":{TIMEOUT_SECONDS}')
    return result

def execute_action_with_confirmation(
    description: str,
    reasoning: str,
    affected_resources: list[str],
    reversible: bool,
    action_fn,
):
    proposal = ActionProposal(
        action_id=str(uuid.uuid4()),
        description=description,
        reasoning=reasoning,
        affected_resources=affected_resources,
        reversible=reversible,
        risk_tier=classify_action(description, reversible, len(affected_resources)),
        requested_at=datetime.now(timezone.utc).isoformat(),
    )
    result = request_confirmation(proposal)
    if result.status == ConfirmationStatus.APPROVED:
        log.info(f'"event":"executing","action_id":"{proposal.action_id}"')
        return action_fn()
    log.warning(f'"event":"action_blocked","action_id":"{proposal.action_id}","reason":"{result.status}"')
    return None