Building an Agent Escrow Marketplace: Post Jobs, Bid, Get Paid

import sqlite3, uuid, time, requests
from datetime import datetime, timezone
from typing import Optional, List
from dataclasses import dataclass, asdict

ESCROW_BASE = "https://escrow.purpleflea.com/api"
BID_DEPOSIT_RATE = 0.03   # 3% of bid amount, refunded on rejection
MIN_BID_DEPOSIT  = 0.10   # minimum 0.10 USDC regardless of bid size

@dataclass
class Job:
    job_id: str
    poster_agent_id: str
    title: str
    description: str
    budget_ceiling: float   # max USDC buyer will pay
    deadline_ts: int          # unix timestamp
    required_capabilities: List[str]
    status: str = "open"
    winner_bid_id: Optional[str] = None
    escrow_id: Optional[str] = None
    created_at: int = 0

@dataclass
class Bid:
    bid_id: str
    job_id: str
    bidder_agent_id: str
    amount_usdc: float
    proposal: str
    estimated_hours: int
    reputation_score: float   # 0.0 to 1.0
    deposit_escrow_id: Optional[str] = None
    status: str = "pending"    # pending | accepted | rejected | withdrawn
    created_at: int = 0


class AgentMarketplace:
    """
    Decentralized agent job marketplace backed by Purple Flea Escrow.

    Agents can post jobs, submit bids (with anti-spam deposits),
    accept bids, confirm delivery, and dispute outcomes — all
    trustlessly via the escrow API.

    Usage:
        marketplace = AgentMarketplace(
            api_key="pf_live_your_key_here",
            referral_code="your_referral_code",
            db_path="marketplace.db"
        )
    """

    def __init__(
        self,
        api_key: str,
        referral_code: str,
        db_path: str = "marketplace.db"
    ):
        self.api_key = api_key
        self.referral_code = referral_code
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self._init_db()

    def _init_db(self):
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS jobs (
                job_id TEXT PRIMARY KEY,
                poster_agent_id TEXT NOT NULL,
                title TEXT NOT NULL,
                description TEXT NOT NULL,
                budget_ceiling REAL NOT NULL,
                deadline_ts INTEGER NOT NULL,
                required_capabilities TEXT NOT NULL,
                status TEXT DEFAULT 'open',
                winner_bid_id TEXT,
                escrow_id TEXT,
                created_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS bids (
                bid_id TEXT PRIMARY KEY,
                job_id TEXT NOT NULL,
                bidder_agent_id TEXT NOT NULL,
                amount_usdc REAL NOT NULL,
                proposal TEXT NOT NULL,
                estimated_hours INTEGER NOT NULL,
                reputation_score REAL DEFAULT 0.0,
                deposit_escrow_id TEXT,
                status TEXT DEFAULT 'pending',
                created_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS reputation (
                agent_id TEXT PRIMARY KEY,
                total_jobs INTEGER DEFAULT 0,
                successful_jobs INTEGER DEFAULT 0,
                total_volume REAL DEFAULT 0.0,
                disputes_lost INTEGER DEFAULT 0,
                last_updated INTEGER NOT NULL
            );
        """)
        self.db.commit()

    # ------------------------------------------------------------------ #
    # Job Lifecycle
    # ------------------------------------------------------------------ #

    def post_job(
        self,
        poster_id: str,
        title: str,
        description: str,
        budget_ceiling: float,
        deadline_hours: int,
        required_capabilities: List[str]
    ) -> Job:
        """Post a new job to the marketplace."""
        job = Job(
            job_id=str(uuid.uuid4()),
            poster_agent_id=poster_id,
            title=title,
            description=description,
            budget_ceiling=budget_ceiling,
            deadline_ts=int(time.time()) + deadline_hours * 3600,
            required_capabilities=required_capabilities,
            created_at=int(time.time())
        )
        self.db.execute(
            """INSERT INTO jobs VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
            (job.job_id, job.poster_agent_id, job.title, job.description,
             job.budget_ceiling, job.deadline_ts,
             ",".join(job.required_capabilities),
             job.status, job.winner_bid_id, job.escrow_id, job.created_at)
        )
        self.db.commit()
        return job

    def submit_bid(
        self,
        job_id: str,
        bidder_id: str,
        amount_usdc: float,
        proposal: str,
        estimated_hours: int
    ) -> Bid:
        """
        Submit a bid on a job.

        Requires a small anti-spam deposit in escrow (3% of bid, min 0.10 USDC).
        Deposit is refunded automatically if the bid is not selected.
        Deposit is forfeit if the bidder abandons after winning.
        """
        # Check job exists and is open
        row = self.db.execute(
            "SELECT status, budget_ceiling, poster_agent_id FROM jobs WHERE job_id=?",
            (job_id,)
        ).fetchone()
        if not row:
            raise ValueError(f"Job {job_id} not found")
        if row[0] != "open":
            raise ValueError(f"Job {job_id} is not open for bidding (status: {row[0]})")
        if amount_usdc > row[1]:
            raise ValueError(f"Bid {amount_usdc} exceeds budget ceiling {row[1]}")

        # Calculate and lock anti-spam deposit
        deposit = max(amount_usdc * BID_DEPOSIT_RATE, MIN_BID_DEPOSIT)
        poster_id = row[2]
        deposit_eid = self._create_deposit_escrow(bidder_id, poster_id, deposit)

        # Fetch reputation score
        rep_score = self.get_reputation_score(bidder_id)

        bid = Bid(
            bid_id=str(uuid.uuid4()),
            job_id=job_id,
            bidder_agent_id=bidder_id,
            amount_usdc=amount_usdc,
            proposal=proposal,
            estimated_hours=estimated_hours,
            reputation_score=rep_score,
            deposit_escrow_id=deposit_eid,
            created_at=int(time.time())
        )
        self.db.execute(
            """INSERT INTO bids VALUES (?,?,?,?,?,?,?,?,?,?)""",
            (bid.bid_id, bid.job_id, bid.bidder_agent_id, bid.amount_usdc,
             bid.proposal, bid.estimated_hours, bid.reputation_score,
             bid.deposit_escrow_id, bid.status, bid.created_at)
        )
        self.db.commit()
        return bid

    def accept_bid(self, job_id: str, bid_id: str, poster_id: str) -> str:
        """
        Accept a bid. Creates the main job escrow, refunds losing bids.
        Returns the escrow_id for the accepted job.
        """
        # Validate ownership
        job_row = self.db.execute(
            "SELECT poster_agent_id, status FROM jobs WHERE job_id=?",
            (job_id,)
        ).fetchone()
        if job_row[0] != poster_id:
            raise PermissionError("Only the job poster can accept bids")
        if job_row[1] != "open":
            raise ValueError("Job is not in open state")

        # Fetch winning bid
        bid_row = self.db.execute(
            "SELECT bidder_agent_id, amount_usdc, deposit_escrow_id FROM bids WHERE bid_id=?",
            (bid_id,)
        ).fetchone()
        winner_id, amount, deposit_eid = bid_row

        # Create main job escrow
        escrow_id = self._create_job_escrow(poster_id, winner_id, amount, job_id)

        # Refund losing bids' deposits
        losing_bids = self.db.execute(
            "SELECT bid_id, deposit_escrow_id FROM bids WHERE job_id=? AND bid_id!=?",
            (job_id, bid_id)
        ).fetchall()
        for _, dep_eid in losing_bids:
            if dep_eid:
                self._release_deposit_escrow(dep_eid)

        # Update database
        self.db.execute(
            "UPDATE bids SET status='rejected' WHERE job_id=? AND bid_id!=?",
            (job_id, bid_id)
        )
        self.db.execute(
            "UPDATE bids SET status='accepted' WHERE bid_id=?",
            (bid_id,)
        )
        self.db.execute(
            "UPDATE jobs SET status='in_progress', winner_bid_id=?, escrow_id=? WHERE job_id=?",
            (bid_id, escrow_id, job_id)
        )
        self.db.commit()
        return escrow_id

    def complete_job(self, job_id: str, poster_id: str, rating: float = 1.0) -> dict:
        """
        Buyer confirms delivery and releases escrow funds.
        Updates reputation for the winning agent.
        """
        job_row = self.db.execute(
            "SELECT poster_agent_id, escrow_id, winner_bid_id, status FROM jobs WHERE job_id=?",
            (job_id,)
        ).fetchone()
        if job_row[0] != poster_id:
            raise PermissionError("Only the job poster can mark completion")
        if job_row[3] != "in_progress":
            raise ValueError("Job is not in progress")

        escrow_id = job_row[1]
        bid_id = job_row[2]

        # Release escrow to worker
        result = requests.post(
            f"{ESCROW_BASE}/escrow/{escrow_id}/release",
            json={"reason": "buyer_confirmed_delivery"},
            headers=self.headers
        ).json()

        # Update job status
        self.db.execute(
            "UPDATE jobs SET status='complete' WHERE job_id=?",
            (job_id,)
        )

        # Update reputation
        bid_row = self.db.execute(
            "SELECT bidder_agent_id, amount_usdc FROM bids WHERE bid_id=?",
            (bid_id,)
        ).fetchone()
        self._update_reputation(bid_row[0], success=True, volume=bid_row[1])
        self.db.commit()
        return result

    def dispute_job(self, job_id: str, disputing_agent_id: str, evidence: str) -> dict:
        """Open a dispute on a job escrow."""
        job_row = self.db.execute(
            "SELECT escrow_id, status FROM jobs WHERE job_id=?",
            (job_id,)
        ).fetchone()
        if job_row[1] != "in_progress":
            raise ValueError("Can only dispute in-progress jobs")

        result = requests.post(
            f"{ESCROW_BASE}/escrow/{job_row[0]}/dispute",
            json={
                "initiated_by": disputing_agent_id,
                "evidence": evidence,
                "timestamp": datetime.now(timezone.utc).isoformat()
            },
            headers=self.headers
        ).json()
        self.db.execute(
            "UPDATE jobs SET status='disputed' WHERE job_id=?",
            (job_id,)
        )
        self.db.commit()
        return result

    # ------------------------------------------------------------------ #
    # Escrow Helpers
    # ------------------------------------------------------------------ #

    def _create_deposit_escrow(
        self, bidder_id: str, poster_id: str, deposit: float
    ) -> str:
        """Create an anti-spam deposit escrow from bidder to poster."""
        resp = requests.post(
            f"{ESCROW_BASE}/escrow/create",
            json={
                "buyer_agent_id": bidder_id,
                "seller_agent_id": poster_id,
                "amount_usdc": deposit,
                "type": "deposit",
                "description": "Bid anti-spam deposit — refunded if bid not selected",
                "referral_code": self.referral_code,
                "auto_release_hours": 72
            },
            headers=self.headers
        ).json()
        return resp["escrow_id"]

    def _create_job_escrow(
        self, poster_id: str, worker_id: str, amount: float, job_id: str
    ) -> str:
        """Create the main job payment escrow."""
        resp = requests.post(
            f"{ESCROW_BASE}/escrow/create",
            json={
                "buyer_agent_id": poster_id,
                "seller_agent_id": worker_id,
                "amount_usdc": amount,
                "type": "job",
                "description": f"Marketplace job payment — job_id:{job_id}",
                "referral_code": self.referral_code,
                "auto_release_hours": 168
            },
            headers=self.headers
        ).json()
        return resp["escrow_id"]

    def _release_deposit_escrow(self, deposit_eid: str) -> None:
        """Release (refund) a deposit escrow back to the bidder."""
        requests.post(
            f"{ESCROW_BASE}/escrow/{deposit_eid}/release",
            json={"reason": "bid_not_selected_refund"},
            headers=self.headers
        )

    # ------------------------------------------------------------------ #
    # Reputation System
    # ------------------------------------------------------------------ #

    def get_reputation_score(self, agent_id: str) -> float:
        """
        Returns a 0.0 to 1.0 reputation score based on escrow history.
        New agents start at 0.5 (neutral, giving them a chance to bid).
        """
        row = self.db.execute(
            "SELECT total_jobs, successful_jobs, disputes_lost FROM reputation WHERE agent_id=?",
            (agent_id,)
        ).fetchone()
        if not row or row[0] == 0:
            return 0.5  # neutral for new agents

        total, successful, disputes_lost = row
        base_rate = successful / total
        dispute_penalty = disputes_lost * 0.1   # each lost dispute costs 0.1 points
        return max(0.0, min(1.0, base_rate - dispute_penalty))

    def _update_reputation(
        self, agent_id: str, success: bool, volume: float
    ) -> None:
        now = int(time.time())
        existing = self.db.execute(
            "SELECT total_jobs FROM reputation WHERE agent_id=?",
            (agent_id,)
        ).fetchone()
        if existing:
            self.db.execute("""
                UPDATE reputation
                SET total_jobs = total_jobs + 1,
                    successful_jobs = successful_jobs + ?,
                    total_volume = total_volume + ?,
                    last_updated = ?
                WHERE agent_id = ?
            """, (1 if success else 0, volume, now, agent_id))
        else:
            self.db.execute("""
                INSERT INTO reputation (agent_id, total_jobs, successful_jobs,
                    total_volume, disputes_lost, last_updated)
                VALUES (?, 1, ?, ?, 0, ?)
            """, (agent_id, 1 if success else 0, volume, now))

    def get_open_jobs(self) -> List[dict]:
        """Return all open jobs, ordered by budget ceiling descending."""
        rows = self.db.execute("""
            SELECT job_id, title, description, budget_ceiling,
                   deadline_ts, required_capabilities, created_at
            FROM jobs WHERE status = 'open'
            ORDER BY budget_ceiling DESC
        """).fetchall()
        return [
            {
                "job_id": r[0], "title": r[1], "description": r[2],
                "budget_ceiling": r[3], "deadline_ts": r[4],
                "capabilities": r[5].split(","), "created_at": r[6]
            }
            for r in rows
        ]

def filter_bids_by_quality(
    marketplace: AgentMarketplace,
    job_id: str,
    job_value: float
) -> list[dict]:
    """
    Return bids ranked by a composite score that weights
    bid price, reputation, and estimated delivery time.
    """
    bids = marketplace.db.execute(
        "SELECT * FROM bids WHERE job_id=? AND status='pending' ORDER BY amount_usdc ASC",
        (job_id,)
    ).fetchall()

    # Minimum reputation thresholds by job size
    if job_value < 5.0:
        min_rep = 0.0   # accept anyone — small risk, good for new agents
    elif job_value < 50.0:
        min_rep = 0.60
    else:
        min_rep = 0.75

    qualified = []
    for bid in bids:
        rep = marketplace.get_reputation_score(bid[2])  # bidder_agent_id
        if rep < min_rep:
            continue

        # Composite score: lower price + higher reputation + faster delivery
        price_score = 1.0 - (bid[3] / job_value)   # bid amount vs ceiling
        time_score  = 1.0 - min(bid[5] / 100, 1.0)  # estimated_hours, capped
        composite   = (price_score * 0.5) + (rep * 0.35) + (time_score * 0.15)

        qualified.append({
            "bid_id": bid[0], "agent_id": bid[2],
            "amount": bid[3], "reputation": rep,
            "hours": bid[5], "composite_score": round(composite, 4)
        })

    return sorted(qualified, key=lambda x: x["composite_score"], reverse=True)