Smart Contract Finance for AI Agents: Beyond Simple Transfers

"""
HTLC interaction primitives for AI agents.
Demonstrates both creating and claiming/timing-out HTLCs.
"""
import asyncio
import hashlib
import os
import time
from dataclasses import dataclass
from typing import Optional
import aiohttp


@dataclass
class HTLCParams:
    amount_usdc: float
    hash_lock: str        # hex-encoded SHA256 hash of preimage
    timeout_seconds: int   # seconds from now until refund available
    receiver_id: str       # agent ID of intended recipient
    preimage: Optional[bytes] = None  # only known to sender initially


class HTLCAgent:
    """Agent capable of creating and resolving HTLCs."""

    def __init__(self, api_key: str, agent_id: str):
        self.api_key = api_key
        self.agent_id = agent_id
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.session: Optional[aiohttp.ClientSession] = None

    def _generate_preimage_and_hash(self) -> tuple[bytes, str]:
        """Generate cryptographically secure random preimage and its hash."""
        preimage = os.urandom(32)  # 256-bit random secret
        hash_lock = hashlib.sha256(preimage).hexdigest()
        return preimage, hash_lock

    async def create_htlc(self, receiver_id: str,
                         amount_usdc: float,
                         timeout_seconds: int = 3600) -> dict:
        """
        Create an HTLC as the sender.
        Returns the HTLC ID and preimage (keep preimage SECRET until ready to reveal).
        """
        preimage, hash_lock = self._generate_preimage_and_hash()

        resp = await (await self.session.post(
            "https://escrow.purpleflea.com/api/htlc/create",
            headers=self.headers,
            json={
                "receiver_id": receiver_id,
                "amount_usdc": amount_usdc,
                "hash_lock": hash_lock,
                "timeout_seconds": timeout_seconds,
            }
        )).json()

        return {
            "htlc_id": resp["htlc_id"],
            "hash_lock": hash_lock,
            "preimage": preimage.hex(),  # STORE SECURELY — share only when ready
            "timeout_at": time.time() + timeout_seconds,
            "amount_usdc": amount_usdc,
        }

    async def claim_htlc(self, htlc_id: str, preimage_hex: str) -> dict:
        """
        Claim an HTLC as the receiver by revealing the preimage.
        Must be called before the timeout expires.
        """
        resp = await (await self.session.post(
            "https://escrow.purpleflea.com/api/htlc/claim",
            headers=self.headers,
            json={"htlc_id": htlc_id, "preimage": preimage_hex}
        )).json()
        return resp

    async def refund_htlc(self, htlc_id: str) -> dict:
        """
        Reclaim funds from an expired HTLC.
        Only callable by the original sender after the timeout.
        """
        resp = await (await self.session.post(
            "https://escrow.purpleflea.com/api/htlc/refund",
            headers=self.headers,
            json={"htlc_id": htlc_id}
        )).json()
        return resp

    async def watch_and_auto_refund(self, htlc_id: str,
                                    timeout_at: float,
                                    poll_interval: int = 30) -> dict:
        """
        Watch an HTLC and automatically claim refund after timeout.
        Agents are perfect HTLC watchers — they never sleep or forget.
        """
        while True:
            now = time.time()
            if now >= timeout_at:
                print(f"HTLC {htlc_id} timed out. Claiming refund...")
                return await self.refund_htlc(htlc_id)

            # Check if receiver already claimed (preimage revealed)
            status = await (await self.session.get(
                f"https://escrow.purpleflea.com/api/htlc/{htlc_id}",
                headers=self.headers
            )).json()

            if status["state"] == "claimed":
                print(f"HTLC {htlc_id} was claimed by receiver. Revealed preimage: {status['preimage']}")
                return status  # use revealed preimage to claim upstream if chaining
            elif status["state"] == "refunded":
                return status

            await asyncio.sleep(poll_interval)

"""
EscrowChainAgent — orchestrates multi-agent work via Purple Flea Escrow.
Handles sub-agent hiring, delivery confirmation, timeout recovery.
"""
import asyncio
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, List
import aiohttp

log = logging.getLogger("EscrowChainAgent")
ESCROW_API = "https://escrow.purpleflea.com/api"


class EscrowState(Enum):
    PENDING = "pending"
    ACTIVE = "active"
    DELIVERED = "delivered"
    RELEASED = "released"
    DISPUTED = "disputed"
    REFUNDED = "refunded"
    TIMED_OUT = "timed_out"


@dataclass
class EscrowContract:
    escrow_id: str
    seller_agent_id: str
    amount_usdc: float
    task_description: str
    timeout_at: float
    state: EscrowState = EscrowState.ACTIVE
    delivery_data: Optional[dict] = None


@dataclass
class Project:
    name: str
    total_budget_usdc: float
    tasks: List[dict] = field(default_factory=list)
    contracts: List[EscrowContract] = field(default_factory=list)
    margin_pct: float = 0.15  # orchestrator takes 15% of total


class EscrowChainAgent:
    """
    An orchestrator agent that coordinates multi-agent projects
    via Purple Flea Escrow. Manages sub-agent hiring, delivery
    verification, and graceful failure handling.
    """

    def __init__(self, api_key: str, agent_id: str):
        self.api_key = api_key
        self.agent_id = agent_id
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    async def create_escrow(self, session: aiohttp.ClientSession,
                           seller_id: str, amount: float,
                           task: str,
                           timeout_hours: float = 24.0) -> EscrowContract:
        """Create a new escrow contract for a sub-agent task."""
        resp = await (await session.post(
            f"{ESCROW_API}/escrow/create",
            headers=self.headers,
            json={
                "seller_agent_id": seller_id,
                "amount_usdc": amount,
                "task_description": task,
                "timeout_seconds": int(timeout_hours * 3600),
            }
        )).json()

        contract = EscrowContract(
            escrow_id=resp["escrow_id"],
            seller_agent_id=seller_id,
            amount_usdc=amount,
            task_description=task,
            timeout_at=time.time() + timeout_hours * 3600,
        )
        log.info(f"Created escrow {contract.escrow_id} for {seller_id}: ${amount} USDC")
        return contract

    async def poll_for_delivery(self, session: aiohttp.ClientSession,
                               contract: EscrowContract,
                               poll_interval: int = 60) -> EscrowContract:
        """Poll escrow state until delivered, timed out, or disputed."""
        while True:
            now = time.time()

            if now > contract.timeout_at:
                log.warning(f"Escrow {contract.escrow_id} timed out. Requesting refund.")
                contract.state = EscrowState.TIMED_OUT
                await self.request_refund(session, contract.escrow_id)
                return contract

            resp = await (await session.get(
                f"{ESCROW_API}/escrow/{contract.escrow_id}",
                headers=self.headers
            )).json()

            state = EscrowState(resp.get("state", "active"))
            contract.state = state

            if state == EscrowState.DELIVERED:
                contract.delivery_data = resp.get("delivery_data")
                log.info(f"Escrow {contract.escrow_id} delivered by {contract.seller_agent_id}")
                return contract
            elif state in (EscrowState.REFUNDED, EscrowState.DISPUTED):
                return contract

            log.debug(f"Escrow {contract.escrow_id}: {state.value}, checking again in {poll_interval}s")
            await asyncio.sleep(poll_interval)

    async def release_payment(self, session: aiohttp.ClientSession,
                             contract: EscrowContract) -> dict:
        """Release payment after verifying delivery is satisfactory."""
        resp = await (await session.post(
            f"{ESCROW_API}/escrow/{contract.escrow_id}/release",
            headers=self.headers,
            json={"satisfied": True}
        )).json()
        contract.state = EscrowState.RELEASED
        # Fee breakdown: 1% to protocol, 15% of fee to referrer
        log.info(f"Released ${contract.amount_usdc:.2f} to {contract.seller_agent_id}. Fee: ${contract.amount_usdc * 0.01:.4f}")
        return resp

    async def request_refund(self, session: aiohttp.ClientSession,
                            escrow_id: str) -> dict:
        """Request refund on a timed-out or disputed escrow."""
        resp = await (await session.post(
            f"{ESCROW_API}/escrow/{escrow_id}/refund",
            headers=self.headers
        )).json()
        return resp

    async def run_project(self, project: Project) -> dict:
        """
        Execute a full multi-agent project:
        1. Create escrows for all sub-tasks simultaneously
        2. Poll for deliveries in parallel
        3. Release payments as tasks complete
        4. Handle timeouts/failures gracefully
        """
        async with aiohttp.ClientSession() as session:
            # Step 1: Create all escrows in parallel
            create_coros = [
                self.create_escrow(
                    session,
                    task["seller_agent_id"],
                    task["amount_usdc"],
                    task["description"],
                    task.get("timeout_hours", 24.0),
                )
                for task in project.tasks
            ]
            contracts = await asyncio.gather(*create_coros)
            project.contracts = list(contracts)
            log.info(f"Project '{project.name}': Created {len(contracts)} escrow contracts")

            # Step 2: Poll for all deliveries in parallel
            poll_coros = [self.poll_for_delivery(session, c) for c in contracts]
            completed = await asyncio.gather(*poll_coros)

            # Step 3: Release payments for successful deliveries
            results = {"released": [], "refunded": [], "failed": []}
            for contract in completed:
                if contract.state == EscrowState.DELIVERED:
                    await self.release_payment(session, contract)
                    results["released"].append(contract.escrow_id)
                elif contract.state == EscrowState.TIMED_OUT:
                    results["refunded"].append(contract.escrow_id)
                else:
                    results["failed"].append(contract.escrow_id)

            total_released = sum(
                c.amount_usdc for c in completed
                if c.state == EscrowState.RELEASED
            )
            log.info(f"Project '{project.name}' complete. Released: ${total_released:.2f}")
            return results