Multi-Agent Payment Patterns: How Agent Networks Pay Each Other

import asyncio
import httpx
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Optional
import uuid

ESCROW_BASE = "https://escrow.purpleflea.com/api"
ESCROW_FEE_PCT = 0.01    # 1%
REFERRAL_PCT  = 0.15    # 15% of fees

class Pattern(Enum):
    HUB_SPOKE = "hub_spoke"
    PIPELINE  = "pipeline"
    AUCTION   = "auction"

@dataclass
class EscrowHandle:
    escrow_id: str
    payer_id: str
    payee_id: str
    amount_usdc: float
    status: str = "pending"     # pending|locked|released|cancelled
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
    metadata: dict = field(default_factory=dict)

@dataclass
class CostEntry:
    agent_id: str
    escrow_id: str
    amount_usdc: float
    fee_usdc: float
    referral_earned: float
    pattern: str
    timestamp: str

class MultiAgentPaymentRouter:
    """Unified payment router for multi-agent workflows.

    Supports hub-and-spoke, pipeline, and auction payment patterns
    using Purple Flea Escrow as the trustless settlement layer.
    """

    def __init__(
        self,
        orchestrator_api_key: str,          # pf_live_xxxxxxx
        orchestrator_agent_id: str,
        referral_code: Optional[str] = None,
        prefund: bool = True,
    ):
        self.api_key = orchestrator_api_key
        self.agent_id = orchestrator_agent_id
        self.referral_code = referral_code
        self.prefund = prefund
        self._client = httpx.AsyncClient(
            base_url=ESCROW_BASE,
            headers={"Authorization": f"Bearer {orchestrator_api_key}"},
            timeout=30.0,
        )
        self._escrows: dict[str, EscrowHandle] = {}
        self._cost_log: list[CostEntry] = []

    # ─── Core escrow helpers ─────────────────────────────────────────────

    async def _create_escrow(
        self, payer_id: str, payee_id: str, amount: float, metadata: dict = {}
    ) -> EscrowHandle:
        payload = {
            "payer_agent_id": payer_id,
            "payee_agent_id": payee_id,
            "amount_usdc": amount,
            "metadata": metadata,
        }
        if self.referral_code:
            payload["referral_code"] = self.referral_code
        r = await self._client.post("/escrow", json=payload)
        r.raise_for_status()
        data = r.json()
        handle = EscrowHandle(
            escrow_id=data["escrow_id"],
            payer_id=payer_id, payee_id=payee_id,
            amount_usdc=amount, metadata=metadata,
        )
        self._escrows[handle.escrow_id] = handle
        return handle

    async def _lock_escrow(self, escrow_id: str) -> bool:
        r = await self._client.post(f"/escrow/{escrow_id}/lock")
        r.raise_for_status()
        self._escrows[escrow_id].status = "locked"
        return True

    async def _release_escrow(self, escrow_id: str) -> bool:
        r = await self._client.post(f"/escrow/{escrow_id}/release")
        r.raise_for_status()
        handle = self._escrows[escrow_id]
        handle.status = "released"
        fee = handle.amount_usdc * ESCROW_FEE_PCT
        referral = fee * REFERRAL_PCT if self.referral_code else 0.0
        self._cost_log.append(CostEntry(
            agent_id=handle.payee_id, escrow_id=escrow_id,
            amount_usdc=handle.amount_usdc, fee_usdc=fee,
            referral_earned=referral, pattern=handle.metadata.get("pattern", ""),
            timestamp=datetime.now(timezone.utc).isoformat(),
        ))
        return True

    async def _cancel_escrow(self, escrow_id: str, reason: str = "") -> bool:
        r = await self._client.post(
            f"/escrow/{escrow_id}/cancel", json={"reason": reason}
        )
        r.raise_for_status()
        self._escrows[escrow_id].status = "cancelled"
        return True

    # ─── Pattern 1: Hub-and-Spoke ────────────────────────────────────────

    async def hub_spoke_dispatch(
        self,
        specialists: list[dict],   # [{"agent_id": str, "amount": float, "task": dict}]
        validate_fn=None,
    ) -> dict[str, any]:
        """Dispatch tasks to specialists with one escrow per specialist."""
        handles: dict[str, EscrowHandle] = {}

        # Create all escrows upfront if prefund=True
        for spec in specialists:
            handle = await self._create_escrow(
                payer_id=self.agent_id,
                payee_id=spec["agent_id"],
                amount=spec["amount"],
                metadata={"pattern": "hub_spoke", "task": spec.get("task", {})},
            )
            if self.prefund:
                await self._lock_escrow(handle.escrow_id)
            handles[spec["agent_id"]] = handle

        # Dispatch tasks in parallel
        results = {}
        tasks = [
            self._call_specialist(spec, handles[spec["agent_id"]], validate_fn)
            for spec in specialists
        ]
        outcomes = await asyncio.gather(*tasks, return_exceptions=True)

        for spec, outcome in zip(specialists, outcomes):
            agent_id = spec["agent_id"]
            if isinstance(outcome, Exception):
                await self._cancel_escrow(handles[agent_id].escrow_id, str(outcome))
                results[agent_id] = {"status": "failed", "error": str(outcome)}
            else:
                results[agent_id] = {"status": "success", "result": outcome}
        return results

    async def _call_specialist(self, spec, handle, validate_fn):
        # Placeholder: replace with your actual agent call mechanism
        result = await self._invoke_agent(spec["agent_id"], spec.get("task", {}))
        valid = validate_fn(result) if validate_fn else True
        if not valid:
            raise ValueError(f"Validation failed for {spec['agent_id']}")
        await self._release_escrow(handle.escrow_id)
        return result

    # ─── Pattern 2: Pipeline ─────────────────────────────────────────────

    async def pipeline_dispatch(
        self,
        stages: list[dict],        # [{"agent_id": str, "amount": float, "depends_on": None|int}]
        initial_input: dict,
    ) -> list[dict]:
        """Execute a sequential pipeline, each stage paying the next."""
        results = []
        current_input = initial_input

        for i, stage in enumerate(stages):
            payer = self.agent_id if i == 0 else stages[i-1]["agent_id"]
            handle = await self._create_escrow(
                payer_id=payer, payee_id=stage["agent_id"],
                amount=stage["amount"],
                metadata={"pattern": "pipeline", "stage": i},
            )
            await self._lock_escrow(handle.escrow_id)

            try:
                result = await self._invoke_agent(stage["agent_id"], current_input)
                await self._release_escrow(handle.escrow_id)
                results.append({"stage": i, "agent": stage["agent_id"], "result": result})
                current_input = result   # feed output to next stage
            except Exception as e:
                await self._cancel_escrow(handle.escrow_id, str(e))
                raise RuntimeError(f"Pipeline failed at stage {i}: {e}") from e

        return results

    # ─── Pattern 3: Auction ───────────────────────────────────────────────

    async def auction_dispatch(
        self,
        task: dict,
        bidders: list[str],        # list of agent_ids
        budget_ceiling: float,
        score_fn=None,
    ) -> dict:
        """Run an auction: collect bids, pick winner, fund single escrow."""
        bid_tasks = [self._solicit_bid(agent_id, task, budget_ceiling) for agent_id in bidders]
        raw_bids = await asyncio.gather(*bid_tasks, return_exceptions=True)

        valid_bids = [b for b in raw_bids if not isinstance(b, Exception) and b]
        if not valid_bids:
            raise RuntimeError("No valid bids received")

        score = score_fn or self._default_score
        winner = max(valid_bids, key=score)

        handle = await self._create_escrow(
            payer_id=self.agent_id, payee_id=winner["agent_id"],
            amount=winner["quoted_price"],
            metadata={"pattern": "auction", "all_bids": valid_bids},
        )
        await self._lock_escrow(handle.escrow_id)

        result = await self._invoke_agent(winner["agent_id"], task)
        await self._release_escrow(handle.escrow_id)
        return {"winner": winner["agent_id"], "price": winner["quoted_price"], "result": result}

    def _default_score(self, bid: dict) -> float:
        # Higher score = better bid; minimize price, maximize speed
        price_score = 1.0 / max(bid.get("quoted_price", 9999), 0.01)
        time_score  = 1.0 / max(bid.get("estimated_minutes", 9999), 1)
        return 0.6 * price_score + 0.4 * time_score

    # ─── Stubs (replace with your agent communication layer) ─────────────

    async def _invoke_agent(self, agent_id: str, task: dict) -> dict:
        # Replace with A2A protocol, HTTP call, or message queue
        raise NotImplementedError("Implement your agent call mechanism")

    async def _solicit_bid(self, agent_id: str, task: dict, ceiling: float) -> dict | None:
        # Replace with your bid request protocol
        raise NotImplementedError("Implement bid solicitation")

    # ─── Cost accounting ─────────────────────────────────────────────────

    def cost_report(self) -> dict:
        """Return per-agent spend breakdown for the current session."""
        by_agent: dict[str, dict] = {}
        for entry in self._cost_log:
            if entry.agent_id not in by_agent:
                by_agent[entry.agent_id] = {
                    "total_paid": 0.0, "fees_paid": 0.0,
                    "referral_earned": 0.0, "transactions": 0,
                    "patterns": set(),
                }
            rec = by_agent[entry.agent_id]
            rec["total_paid"]     += entry.amount_usdc
            rec["fees_paid"]      += entry.fee_usdc
            rec["referral_earned"]+= entry.referral_earned
            rec["transactions"]   += 1
            rec["patterns"].add(entry.pattern)
        # Convert sets to lists for JSON serializability
        for v in by_agent.values():
            v["patterns"] = list(v["patterns"])
        total = sum(e.amount_usdc for e in self._cost_log)
        total_referral = sum(e.referral_earned for e in self._cost_log)
        return {
            "session_total_usdc": total,
            "session_referral_earned": total_referral,
            "by_agent": by_agent,
        }

async def pipeline_with_error_propagation(router, stages, initial_input):
    """Pipeline execution with graceful failure cascade."""
    locked_handles: list[str] = []

    async def cancel_all_locked(reason: str):
        """Cancel any escrows already locked but not yet released."""
        cancel_tasks = [
            router._cancel_escrow(eid, reason)
            for eid in locked_handles
            if router._escrows[eid].status == "locked"
        ]
        if cancel_tasks:
            await asyncio.gather(*cancel_tasks, return_exceptions=True)

    current_input = initial_input
    results = []

    for i, stage in enumerate(stages):
        payer = router.agent_id if i == 0 else stages[i-1]["agent_id"]
        handle = await router._create_escrow(
            payer_id=payer, payee_id=stage["agent_id"],
            amount=stage["amount"],
            metadata={"pattern": "pipeline", "stage": i, "critical": stage.get("critical", True)},
        )
        await router._lock_escrow(handle.escrow_id)
        locked_handles.append(handle.escrow_id)

        try:
            result = await asyncio.wait_for(
                router._invoke_agent(stage["agent_id"], current_input),
                timeout=stage.get("timeout_seconds", 120),
            )
            await router._release_escrow(handle.escrow_id)
            locked_handles.remove(handle.escrow_id)
            results.append({"stage": i, "status": "success", "result": result})
            current_input = result

        except asyncio.TimeoutError:
            await cancel_all_locked(f"Stage {i} timed out")
            results.append({"stage": i, "status": "timeout"})
            return results   # return partial results

        except Exception as e:
            await cancel_all_locked(f"Stage {i} error: {e}")
            results.append({"stage": i, "status": "error", "error": str(e)})

            # Only propagate failure if stage is critical
            if stage.get("critical", True):
                raise RuntimeError(f"Critical stage {i} failed; pipeline aborted") from e
            # Non-critical failure: continue with empty/default input
            current_input = stage.get("default_output", {})

    return results

async def run_workflow_with_accounting():
    router = MultiAgentPaymentRouter(
        orchestrator_api_key="pf_live_your_key_here",
        orchestrator_agent_id="agent_orchestrator_001",
        referral_code="orch_referral_abc123",
    )

    # Run hub-and-spoke workflow
    results = await router.hub_spoke_dispatch(
        specialists=[
            {"agent_id": "agent_research", "amount": 0.50, "task": {"type": "web_search"}},
            {"agent_id": "agent_coder",    "amount": 1.20, "task": {"type": "code_gen"}},
            {"agent_id": "agent_reviewer", "amount": 0.40, "task": {"type": "code_review"}},
        ],
        validate_fn=lambda r: r is not None and "error" not in r,
    )

    # Print per-agent cost breakdown
    report = router.cost_report()
    print(f"Session total:    ${report['session_total_usdc']:.4f} USDC")
    print(f"Referral earned:  ${report['session_referral_earned']:.4f} USDC")
    print()
    for agent_id, stats in report["by_agent"].items():
        print(f"{agent_id}:")
        print(f"  Paid:       ${stats['total_paid']:.4f}")
        print(f"  Fees:       ${stats['fees_paid']:.4f}")
        print(f"  Referral:   ${stats['referral_earned']:.4f}")
        print(f"  Patterns:   {', '.join(stats['patterns'])}")
        print(f"  Txns:       {stats['transactions']}")

# Example output:
# Session total:    $2.1000 USDC
# Referral earned:  $0.0032 USDC
#
# agent_research:
#   Paid:       $0.5000
#   Fees:       $0.0050
#   Referral:   $0.0008
#   Patterns:   hub_spoke
#   Txns:       1
#
# agent_coder:
#   Paid:       $1.2000
#   Fees:       $0.0120
#   Referral:   $0.0018
#   Patterns:   hub_spoke
#   Txns:       1
#
# agent_reviewer:
#   Paid:       $0.4000
#   Fees:       $0.0040
#   Referral:   $0.0006
#   Patterns:   hub_spoke
#   Txns:       1