Building Agent Labor Markets: Architecture and Implementation

from typing import List, Dict
from dataclasses import dataclass
import asyncio
from queue import Queue

@dataclass
class Task:
    id: str
    requirements: Dict[str, any]
    priority: int
    deadline: float

@dataclass
class Agent:
    id: str
    capabilities: List[str]
    reputation: float
    availability: bool

class AgentCoordinator:
    """Hub-and-spoke coordinator for centralized agent orchestration"""

    def __init__(self, agents: List[Agent]):
        self.agents = agents
        self.task_queue = Queue()
        self.active_tasks = {}

    def match_agents(self, task: Task) -> List[Agent]:
        """Match task requirements to agent capabilities"""
        suitable = [
            agent for agent in self.agents
            if all(cap in agent.capabilities for cap in task.requirements.get('capabilities', []))
            and agent.availability
            and agent.reputation >= task.requirements.get('min_reputation', 0.0)
        ]
        # Sort by reputation descending
        return sorted(suitable, key=lambda a: a.reputation, reverse=True)

    async def dispatch(self, agents: List[Agent], task: Task) -> Dict:
        """Dispatch task to best available agent"""
        if not agents:
            return {"status": "failed", "reason": "no_suitable_agents"}

        # Select highest-reputation available agent
        selected_agent = agents[0]

        # Execute task (simplified - real implementation would use agent API)
        result = await self._execute_task(selected_agent, task)

        return {
            "status": "completed",
            "agent_id": selected_agent.id,
            "result": result
        }

    async def orchestrate(self, task: Task) -> Dict:
        """Main orchestration logic"""
        suitable_agents = self.match_agents(task)
        result = await self.dispatch(suitable_agents, task)

        # Update agent availability and reputation based on result
        if result["status"] == "completed":
            self._update_agent_metrics(result["agent_id"], task, result)

        return result

    async def _execute_task(self, agent: Agent, task: Task):
        """Execute task using selected agent"""
        # Simplified execution - in practice, this calls agent's API
        await asyncio.sleep(0.1)  # Simulate task execution
        return {"output": f"Task {task.id} completed by {agent.id}"}

    def _update_agent_metrics(self, agent_id: str, task: Task, result: Dict):
        """Update agent reputation and availability"""
        # Simplified reputation update
        pass

from typing import Set, Dict, List
import asyncio
from dataclasses import dataclass, field

@dataclass
class MeshAgent:
    """Autonomous agent in mesh network"""
    id: str
    capabilities: List[str]
    peers: Set[str] = field(default_factory=set)
    local_state: Dict = field(default_factory=dict)

    async def broadcast_task(self, task: Task) -> Dict:
        """Gossip-based task propagation to peers"""
        # Phase 1: Broadcast to immediate peers
        peer_responses = await asyncio.gather(*[
            self._send_to_peer(peer_id, task)
            for peer_id in self.peers
        ], return_exceptions=True)

        # Phase 2: Consensus on who executes task
        allocation = self._consensus_allocation(task, peer_responses)

        return allocation

    async def _send_to_peer(self, peer_id: str, task: Task) -> Dict:
        """Send task to peer and get capability response"""
        # Simplified - in practice, uses gRPC or WebSocket
        await asyncio.sleep(0.01)  # Network latency

        return {
            "peer_id": peer_id,
            "can_execute": self._can_peer_execute(peer_id, task),
            "bid": self._calculate_bid(task)
        }

    def _consensus_allocation(self, task: Task, responses: List[Dict]) -> Dict:
        """Determine task allocation via auction consensus"""
        valid_responses = [r for r in responses if isinstance(r, dict)]

        # Simple auction: lowest bid wins
        # (Real implementation uses Byzantine fault-tolerant consensus)
        if valid_responses:
            winner = min(valid_responses, key=lambda r: r.get('bid', float('inf')))
            return {
                "status": "allocated",
                "executor": winner['peer_id'],
                "bid": winner['bid']
            }

        return {"status": "no_allocation"}

    def _can_peer_execute(self, peer_id: str, task: Task) -> bool:
        """Check if peer can execute task (simplified)"""
        # In practice, queries peer's capability registry
        return True

    def _calculate_bid(self, task: Task) -> float:
        """Calculate bid for task based on local cost model"""
        # Simplified pricing - real implementation from Episode 2
        base_cost = 10.0
        complexity_multiplier = task.requirements.get('complexity', 1.0)
        return base_cost * complexity_multiplier

from typing import List, Callable
from dataclasses import dataclass
import asyncio

@dataclass
class ComplexTask:
    """Multi-step task requiring decomposition"""
    id: str
    description: str
    subtask_count: int
    requirements: Dict

@dataclass
class WorkerAgent:
    """Specialized worker agent"""
    id: str
    specialization: str

    async def execute(self, subtask: Task) -> Dict:
        """Execute assigned subtask"""
        # Simplified execution
        await asyncio.sleep(0.05)  # Simulate work
        return {
            "subtask_id": subtask.id,
            "result": f"Completed by {self.id}",
            "quality_score": 0.95
        }

class ManagerAgent:
    """Manager agent that delegates to workers"""

    def __init__(self, workers: List[WorkerAgent]):
        self.workers = workers
        self.delegation_strategy = "round_robin"  # or "capability_match"

    def decompose(self, task: ComplexTask) -> List[Task]:
        """Break complex task into subtasks"""
        # Simplified decomposition logic
        subtasks = []
        for i in range(task.subtask_count):
            subtasks.append(Task(
                id=f"{task.id}_subtask_{i}",
                requirements={
                    "complexity": task.requirements.get("complexity", 1.0) / task.subtask_count
                },
                priority=task.requirements.get("priority", 5),
                deadline=task.requirements.get("deadline", float('inf'))
            ))
        return subtasks

    async def delegate_task(self, task: ComplexTask) -> Dict:
        """Decompose and delegate to workers"""
        subtasks = self.decompose(task)

        # Parallel execution across workers
        results = []
        for subtask, worker in zip(subtasks, self.workers):
            result = await worker.execute(subtask)
            results.append(result)

        # Synthesize results
        final_result = self.synthesize(results)
        return final_result

    def synthesize(self, results: List[Dict]) -> Dict:
        """Combine worker results into final output"""
        # Quality-weighted aggregation
        avg_quality = sum(r['quality_score'] for r in results) / len(results)

        return {
            "status": "completed",
            "subtask_results": results,
            "aggregate_quality": avg_quality,
            "synthesis": "Combined output from all workers"
        }

import json
from datetime import datetime
from typing import Dict, Optional
import redis.asyncio as redis

class RedisStreamMessaging:
    """Message queue pattern using Redis Streams"""

    def __init__(self, redis_url: str = "redis://localhost"):
        self.redis = redis.from_url(redis_url)
        self.stream_name = "agent_tasks"

    async def send_task(self, task: Task, sender_id: str) -> str:
        """Publish task to message queue"""
        message = {
            "task_id": task.id,
            "payload": {
                "requirements": task.requirements,
                "priority": task.priority,
                "deadline": task.deadline
            },
            "timestamp": datetime.now().isoformat(),
            "sender": sender_id
        }

        # Add to Redis Stream
        message_id = await self.redis.xadd(
            self.stream_name,
            {"data": json.dumps(message)},
            maxlen=10000  # Keep last 10K messages
        )

        return message_id

    async def consume_tasks(self, consumer_group: str, consumer_id: str):
        """Consume tasks from queue (consumer group pattern)"""
        # Create consumer group if not exists
        try:
            await self.redis.xgroup_create(
                self.stream_name,
                consumer_group,
                id='0',
                mkstream=True
            )
        except redis.ResponseError:
            pass  # Group already exists

        # Read messages
        while True:
            messages = await self.redis.xreadgroup(
                consumer_group,
                consumer_id,
                {self.stream_name: '>'},
                count=10,
                block=1000  # Block for 1 second
            )

            for stream_name, message_list in messages:
                for message_id, message_data in message_list:
                    await self._process_message(message_id, message_data)

    async def _process_message(self, message_id: str, data: Dict):
        """Process individual message"""
        task_data = json.loads(data[b'data'])
        # Process task...

        # Acknowledge message
        await self.redis.xack(self.stream_name, "agent_workers", message_id)

// agent.proto - Protocol Buffers schema
syntax = "proto3";

package agent;

service AgentCoordination {
  rpc AssignTask (TaskRequest) returns (TaskResponse);
  rpc GetAgentStatus (StatusRequest) returns (StatusResponse);
  rpc NegotiatePrice (PriceRequest) returns (PriceResponse);
}

message TaskRequest {
  string task_id = 1;
  map<string, string> requirements = 2;
  int32 priority = 3;
  double max_price = 4;
}

message TaskResponse {
  string task_id = 1;
  bool accepted = 2;
  double quoted_price = 3;
  string estimated_completion = 4;
}

import grpc
from concurrent import futures
import agent_pb2
import agent_pb2_grpc

class AgentCoordinationService(agent_pb2_grpc.AgentCoordinationServicer):
    """gRPC service implementation for agent coordination"""

    def AssignTask(self, request, context):
        """Handle synchronous task assignment request"""
        task_id = request.task_id
        max_price = request.max_price

        # Calculate if agent can accept task at this price
        quoted_price = self._calculate_price(request.requirements)
        accepted = quoted_price <= max_price

        return agent_pb2.TaskResponse(
            task_id=task_id,
            accepted=accepted,
            quoted_price=quoted_price,
            estimated_completion="2024-01-15T14:30:00Z"
        )

    def _calculate_price(self, requirements):
        # Pricing logic from Episode 2
        base_cost = 10.0
        complexity = float(requirements.get('complexity', '1.0'))
        return base_cost * complexity

# Server setup
def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    agent_pb2_grpc.add_AgentCoordinationServicer_to_server(
        AgentCoordinationService(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;

contract AgentReputation {
    struct Review {
        address reviewer;
        uint8 rating;          // 1-5 stars
        uint256 timestamp;
        bytes32 taskHash;      // Proof of work completion
        string feedback;
    }

    struct ReputationScore {
        uint256 totalReviews;
        uint256 sumRatings;
        uint256 lastUpdated;
    }

    mapping(address => Review[]) public reviews;
    mapping(address => ReputationScore) public scores;

    event ReviewSubmitted(
        address indexed agent,
        address indexed reviewer,
        uint8 rating,
        bytes32 taskHash
    );

    function submitReview(
        address agent,
        uint8 rating,
        bytes32 taskHash,
        string memory feedback
    ) external {
        require(rating >= 1 && rating <= 5, "Invalid rating");
        require(agent != msg.sender, "Cannot review self");

        // Store review
        reviews[agent].push(Review({
            reviewer: msg.sender,
            rating: rating,
            timestamp: block.timestamp,
            taskHash: taskHash,
            feedback: feedback
        }));

        // Update aggregate score
        ReputationScore storage score = scores[agent];
        score.totalReviews++;
        score.sumRatings += rating;
        score.lastUpdated = block.timestamp;

        emit ReviewSubmitted(agent, msg.sender, rating, taskHash);
    }

    function getReputation(address agent)
        external
        view
        returns (uint256 avgRating, uint256 totalReviews)
    {
        ReputationScore memory score = scores[agent];

        if (score.totalReviews == 0) {
            return (0, 0);
        }

        // Average rating (scaled by 100 for precision)
        avgRating = (score.sumRatings * 100) / score.totalReviews;
        totalReviews = score.totalReviews;
    }

    function getReviews(address agent)
        external
        view
        returns (Review[] memory)
    {
        return reviews[agent];
    }
}

from typing import List, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import math

@dataclass
class Review:
    reviewer_id: str
    rating: int  # 1-5
    timestamp: datetime
    task_id: str
    verified: bool

class ReputationEngine:
    """Advanced off-chain reputation calculation

    Based on eBay reputation system research (Resnick et al., 2006)
    and Stack Overflow reputation mechanics
    """

    def __init__(self, db_connection):
        self.db = db_connection

    def calculate_score(self, agent_id: str) -> float:
        """Calculate reputation score using Bayesian average + time decay"""
        reviews = self._get_reviews(agent_id)

        if not reviews:
            return 0.0

        # Bayesian average to handle new agents (cold-start problem)
        prior_weight = 10  # Equivalent to 10 neutral reviews
        prior_mean = 3.5   # Assume 3.5-star prior

        total_reviews = len(reviews)
        sum_ratings = sum(r.rating for r in reviews)

        # Bayesian average
        bayesian_avg = (
            (sum_ratings + prior_weight * prior_mean) /
            (total_reviews + prior_weight)
        )

        # Time decay factor (recent reviews matter more)
        recency_weight = self._calculate_recency_weight(reviews)

        # Verified review bonus
        verified_bonus = self._calculate_verified_bonus(reviews)

        # Final score (0-5 scale)
        final_score = bayesian_avg * recency_weight * verified_bonus

        return min(5.0, max(0.0, final_score))

    def _calculate_recency_weight(self, reviews: List[Review]) -> float:
        """
        Apply exponential decay to older reviews
        Half-life: 90 days (reviews lose half weight after 3 months)
        """
        now = datetime.now()
        half_life_days = 90
        decay_constant = math.log(2) / half_life_days

        total_weight = 0.0
        weighted_sum = 0.0

        for review in reviews:
            days_old = (now - review.timestamp).days
            weight = math.exp(-decay_constant * days_old)

            total_weight += weight
            weighted_sum += review.rating * weight

        return weighted_sum / total_weight if total_weight > 0 else 1.0

    def _calculate_verified_bonus(self, reviews: List[Review]) -> float:
        """Boost score for high percentage of verified reviews"""
        if not reviews:
            return 1.0

        verified_ratio = sum(1 for r in reviews if r.verified) / len(reviews)

        # 10% bonus for 100% verified reviews
        return 1.0 + (0.1 * verified_ratio)

    def _get_reviews(self, agent_id: str) -> List[Review]:
        """Fetch reviews from database"""
        # Simplified - in practice, uses SQL query with indexes
        return self.db.query_reviews(agent_id)

    def detect_collusion(self, agent_id: str) -> Dict[str, any]:
        """
        Detect suspicious review patterns (gaming resistance)

        Based on social graph analysis and anomaly detection
        """
        reviews = self._get_reviews(agent_id)

        # Pattern 1: Same reviewers repeatedly (Sybil attack)
        reviewer_frequency = {}
        for review in reviews:
            reviewer_frequency[review.reviewer_id] = \
                reviewer_frequency.get(review.reviewer_id, 0) + 1

        max_repeat = max(reviewer_frequency.values()) if reviewer_frequency else 0
        sybil_risk = max_repeat / len(reviews) if reviews else 0

        # Pattern 2: Sudden rating changes (review bombing)
        recent_reviews = [r for r in reviews
                         if (datetime.now() - r.timestamp) < timedelta(days=7)]
        old_reviews = [r for r in reviews
                      if (datetime.now() - r.timestamp) >= timedelta(days=7)]

        if recent_reviews and old_reviews:
            recent_avg = sum(r.rating for r in recent_reviews) / len(recent_reviews)
            old_avg = sum(r.rating for r in old_reviews) / len(old_reviews)
            rating_shift = abs(recent_avg - old_avg)
        else:
            rating_shift = 0

        return {
            "sybil_risk": sybil_risk,
            "rating_shift": rating_shift,
            "suspicious": sybil_risk > 0.3 or rating_shift > 1.5
        }

from typing import List
from dataclasses import dataclass
import hashlib

@dataclass
class MerkleNode:
    hash: str
    left: Optional['MerkleNode'] = None
    right: Optional['MerkleNode'] = None

class HybridReputationSystem:
    """Combine on-chain immutability with off-chain performance"""

    def __init__(self, db, blockchain_client):
        self.offchain_db = db
        self.blockchain = blockchain_client
        self.merkle_tree = None
        self.pending_reviews = []

    async def submit_review(self, review: Review):
        """Fast off-chain write"""
        await self.offchain_db.insert(review)

        # Add to pending Merkle tree
        self.pending_reviews.append(review)

    async def anchor_daily(self):
        """Once per day, commit Merkle root to blockchain"""
        # Build Merkle tree from pending reviews
        self.merkle_tree = self._build_merkle_tree(self.pending_reviews)

        # Get root hash
        root_hash = self.merkle_tree.hash

        # Commit to blockchain
        tx_hash = await self.blockchain.commit_root(root_hash)

        # Clear pending reviews
        self.pending_reviews = []

        return tx_hash

    def _build_merkle_tree(self, reviews: List[Review]) -> MerkleNode:
        """Build Merkle tree from review list"""
        if not reviews:
            return None

        # Create leaf nodes
        leaves = [MerkleNode(hash=self._hash_review(r)) for r in reviews]

        # Build tree bottom-up
        while len(leaves) > 1:
            new_level = []
            for i in range(0, len(leaves), 2):
                left = leaves[i]
                right = leaves[i + 1] if i + 1 < len(leaves) else left

                parent_hash = hashlib.sha256(
                    (left.hash + right.hash).encode()
                ).hexdigest()

                new_level.append(MerkleNode(
                    hash=parent_hash,
                    left=left,
                    right=right
                ))

            leaves = new_level

        return leaves[0]

    def _hash_review(self, review: Review) -> str:
        """Hash review data"""
        data = f"{review.reviewer_id}{review.rating}{review.timestamp}{review.task_id}"
        return hashlib.sha256(data.encode()).hexdigest()

    def generate_proof(self, review: Review) -> List[str]:
        """Generate Merkle proof for review inclusion"""
        # Walk tree from leaf to root, collecting sibling hashes
        # Simplified implementation
        proof = []
        # ... proof generation logic ...
        return proof

    def verify_proof(self, review: Review, proof: List[str], root_hash: str) -> bool:
        """Verify review was included in committed tree"""
        current_hash = self._hash_review(review)

        for sibling_hash in proof:
            current_hash = hashlib.sha256(
                (current_hash + sibling_hash).encode()
            ).hexdigest()

        return current_hash == root_hash

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;

contract TaskEscrow {
    enum State {
        Created,
        Funded,
        Completed,
        Disputed,
        Released,
        Refunded
    }

    struct Task {
        address client;
        address agent;
        uint256 amount;
        State state;
        uint256 deadline;
        address arbitrator;
        bytes32 completionProof;
    }

    mapping(bytes32 => Task) public tasks;

    event TaskCreated(bytes32 indexed taskId, address client, address agent);
    event TaskFunded(bytes32 indexed taskId, uint256 amount);
    event TaskCompleted(bytes32 indexed taskId, bytes32 proof);
    event TaskReleased(bytes32 indexed taskId, address agent);
    event TaskRefunded(bytes32 indexed taskId, address client);
    event DisputeRaised(bytes32 indexed taskId, address disputant);

    function createTask(
        bytes32 taskId,
        address agent,
        uint256 deadline,
        address arbitrator
    ) external payable {
        require(msg.value > 0, "Must fund task");
        require(tasks[taskId].state == State.Created ||
                tasks[taskId].client == address(0), "Task exists");

        tasks[taskId] = Task({
            client: msg.sender,
            agent: agent,
            amount: msg.value,
            state: State.Funded,
            deadline: deadline,
            arbitrator: arbitrator,
            completionProof: bytes32(0)
        });

        emit TaskCreated(taskId, msg.sender, agent);
        emit TaskFunded(taskId, msg.value);
    }

    function completeTask(bytes32 taskId, bytes32 proof) external {
        Task storage task = tasks[taskId];
        require(msg.sender == task.agent, "Only agent can complete");
        require(task.state == State.Funded, "Invalid state");

        task.state = State.Completed;
        task.completionProof = proof;

        emit TaskCompleted(taskId, proof);
    }

    function releasePayment(bytes32 taskId) external {
        Task storage task = tasks[taskId];
        require(msg.sender == task.client, "Only client can release");
        require(task.state == State.Completed, "Task not completed");

        task.state = State.Released;
        payable(task.agent).transfer(task.amount);

        emit TaskReleased(taskId, task.agent);
    }

    function raiseDispute(bytes32 taskId) external {
        Task storage task = tasks[taskId];
        require(msg.sender == task.client || msg.sender == task.agent,
                "Not authorized");
        require(task.state == State.Completed || task.state == State.Funded,
                "Invalid state");

        task.state = State.Disputed;

        emit DisputeRaised(taskId, msg.sender);
    }

    function resolveDispute(bytes32 taskId, bool favorAgent) external {
        Task storage task = tasks[taskId];
        require(msg.sender == task.arbitrator, "Only arbitrator");
        require(task.state == State.Disputed, "Not in dispute");

        if (favorAgent) {
            task.state = State.Released;
            payable(task.agent).transfer(task.amount);
            emit TaskReleased(taskId, task.agent);
        } else {
            task.state = State.Refunded;
            payable(task.client).transfer(task.amount);
            emit TaskRefunded(taskId, task.client);
        }
    }

    function refundIfExpired(bytes32 taskId) external {
        Task storage task = tasks[taskId];
        require(block.timestamp > task.deadline, "Not expired");
        require(task.state == State.Funded, "Invalid state");

        task.state = State.Refunded;
        payable(task.client).transfer(task.amount);

        emit TaskRefunded(taskId, task.client);
    }
}

from typing import Callable, Tuple
from dataclasses import dataclass

@dataclass
class Proof:
    """Zero-knowledge proof of computation"""
    proof_data: bytes
    public_inputs: bytes
    claimed_output: bytes

class VerifiableComputation:
    """
    Conceptual framework for verifiable computation using zk-SNARKs

    Real implementation requires zk-SNARK libraries (snarkjs, circom)
    and significant cryptographic expertise.
    """

    def __init__(self, circuit_definition: str):
        self.circuit = self._compile_circuit(circuit_definition)

    def create_computation_proof(
        self,
        input_data: bytes,
        computation: Callable,
    ) -> Proof:
        """
        Perform computation and generate zero-knowledge proof

        The proof demonstrates:
        1. Computation was performed correctly
        2. Input and output are consistent with circuit
        3. No knowledge of private inputs revealed
        """
        # Perform computation
        output = computation(input_data)

        # Generate witness (intermediate computation steps)
        witness = self.circuit.generate_witness(input_data, output)

        # Create zero-knowledge proof
        proof_data = self.circuit.prove(witness)

        return Proof(
            proof_data=proof_data,
            public_inputs=input_data,
            claimed_output=output
        )

    def verify_computation(
        self,
        proof: Proof
    ) -> bool:
        """
        Verify proof without re-running computation

        Verification is fast (~10ms) even if computation took hours
        """
        return self.circuit.verify(
            proof.public_inputs,
            proof.claimed_output,
            proof.proof_data
        )

    def _compile_circuit(self, definition: str):
        """Compile circuit definition to constraint system"""
        # Simplified - real implementation uses Circom or similar
        pass

Hot Context (Redis, <50ms): Last 10 exchanges, always in memory
Warm Context (Pinecone, <100ms): Semantically relevant history, fetched on-demand
Cold Context (S3, <500ms): Full archive, rarely accessed