Upload ledger/ledger.py

ledger/ledger.py

@@ -0,0 +1,342 @@
+"""
+Credit Ledger: non-transferable, decaying, capability-scoped credits with provenance.
+"""
+
+import json
+import math
+import time
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+
+@dataclass
+class LedgerEntry:
+    agent_id: str
+    task_id: str
+    action_id: str
+    earned_credit: float = 0.0
+    spent_credit: float = 0.0
+    decayed_credit: float = 0.0
+    remaining_credit: float = 0.0
+    reason: str = ""
+    oracle_score: float = 0.0
+    compute_cost: float = 0.0
+    timestamp: float = field(default_factory=time.time)
+    capability_scope: str = "general"
+    task_scope: str = "global"
+    provenance_hash: str = ""
+
+
+class CreditLedger:
+    """
+    Immutable-style ledger with decay, non-transferability, and anti-gaming rules.
+    """
+
+    def __init__(
+        self,
+        decay_lambda: float = 0.05,  # per-eval decay factor
+        hoarding_threshold: float = 100.0,
+        hoarding_window: int = 50,
+        max_history: int = 10000,
+        ledger_path: Optional[str] = None,
+    ):
+        self.decay_lambda = decay_lambda
+        self.hoarding_threshold = hoarding_threshold
+        self.hoarding_window = hoarding_window
+        self.max_history = max_history
+        self.ledger_path = ledger_path
+
+        # Main ledger: list of LedgerEntry
+        self.entries: List[LedgerEntry] = []
+
+        # Per-agent, per-scope balances (computed lazily)
+        self._agent_balances: Dict[Tuple[str, str, str], float] = {}
+
+        # Anti-gaming: track agent behavior patterns
+        self._agent_action_history: Dict[str, List[Dict]] = {}
+
+    # ------------------------------------------------------------------
+    # Core operations
+    # ------------------------------------------------------------------
+
+    def earn(
+        self,
+        agent_id: str,
+        task_id: str,
+        action_id: str,
+        amount: float,
+        oracle_score: float,
+        compute_cost: float,
+        reason: str,
+        capability_scope: str = "general",
+        task_scope: str = "global",
+        provenance_hash: str = "",
+    ) -> LedgerEntry:
+        """Award credits based on verified impact."""
+        if amount < 0:
+            raise ValueError("Earn amount must be non-negative")
+
+        # Apply hoarding penalty: if agent has held high balance without spending
+        current_balance = self.balance(agent_id, capability_scope, task_scope)
+        if current_balance > self.hoarding_threshold:
+            # Accelerated decay for hoarders
+            amount *= 0.8
+            reason += " [hoarding_penalty_applied]"
+
+        entry = LedgerEntry(
+            agent_id=agent_id,
+            task_id=task_id,
+            action_id=action_id,
+            earned_credit=amount,
+            spent_credit=0.0,
+            decayed_credit=0.0,
+            remaining_credit=amount,
+            reason=reason,
+            oracle_score=oracle_score,
+            compute_cost=compute_cost,
+            capability_scope=capability_scope,
+            task_scope=task_scope,
+            provenance_hash=provenance_hash,
+        )
+        self.entries.append(entry)
+        self._trim_history()
+        self._invalidate_cache(agent_id, capability_scope, task_scope)
+        return entry
+
+    def spend(
+        self,
+        agent_id: str,
+        task_id: str,
+        action_id: str,
+        amount: float,
+        capability_scope: str = "general",
+        task_scope: str = "global",
+        reason: str = "",
+    ) -> Tuple[bool, LedgerEntry]:
+        """Spend credits. Returns (success, entry)."""
+        if amount < 0:
+            return False, None
+
+        self._apply_decay(agent_id, capability_scope, task_scope)
+        current = self.balance(agent_id, capability_scope, task_scope)
+
+        if current < amount:
+            return False, None
+
+        entry = LedgerEntry(
+            agent_id=agent_id,
+            task_id=task_id,
+            action_id=action_id,
+            earned_credit=0.0,
+            spent_credit=amount,
+            decayed_credit=0.0,
+            remaining_credit=current - amount,
+            reason=reason,
+            capability_scope=capability_scope,
+            task_scope=task_scope,
+        )
+        self.entries.append(entry)
+        self._trim_history()
+        self._invalidate_cache(agent_id, capability_scope, task_scope)
+        return True, entry
+
+    def transfer(
+        self,
+        from_agent: str,
+        to_agent: str,
+        amount: float,
+        capability_scope: str = "general",
+        task_scope: str = "global",
+    ) -> bool:
+        """
+        Credits are NON-TRANSFERABLE by design.
+        Returns False always; logs an attempted transfer for audit.
+        """
+        entry = LedgerEntry(
+            agent_id=from_agent,
+            task_id="TRANSFER_ATTEMPT",
+            action_id=f"to:{to_agent}",
+            earned_credit=0.0,
+            spent_credit=0.0,
+            decayed_credit=0.0,
+            remaining_credit=self.balance(from_agent, capability_scope, task_scope),
+            reason=f"TRANSFER_BLOCKED: {amount} to {to_agent}",
+            capability_scope=capability_scope,
+            task_scope=task_scope,
+        )
+        self.entries.append(entry)
+        return False
+
+    def revoke(
+        self,
+        agent_id: str,
+        task_id: str,
+        action_id: str,
+        amount: float,
+        reason: str,
+        capability_scope: str = "general",
+        task_scope: str = "global",
+    ) -> bool:
+        """Revoke credits retroactively after negative outcome."""
+        self._apply_decay(agent_id, capability_scope, task_scope)
+        current = self.balance(agent_id, capability_scope, task_scope)
+        revoke_amount = min(amount, current)
+
+        if revoke_amount <= 0:
+            return False
+
+        entry = LedgerEntry(
+            agent_id=agent_id,
+            task_id=task_id,
+            action_id=action_id,
+            earned_credit=0.0,
+            spent_credit=0.0,
+            decayed_credit=revoke_amount,
+            remaining_credit=current - revoke_amount,
+            reason=f"REVOKE: {reason}",
+            capability_scope=capability_scope,
+            task_scope=task_scope,
+        )
+        self.entries.append(entry)
+        self._invalidate_cache(agent_id, capability_scope, task_scope)
+        return True
+
+    def balance(
+        self,
+        agent_id: str,
+        capability_scope: str = "general",
+        task_scope: str = "global",
+    ) -> float:
+        """Compute current balance after decay."""
+        self._apply_decay(agent_id, capability_scope, task_scope)
+        key = (agent_id, capability_scope, task_scope)
+        return self._agent_balances.get(key, 0.0)
+
+    # ------------------------------------------------------------------
+    # Decay
+    # ------------------------------------------------------------------
+
+    def _apply_decay(self, agent_id: str, capability_scope: str, task_scope: str):
+        """Apply exponential decay since last update."""
+        key = (agent_id, capability_scope, task_scope)
+        relevant = [
+            e for e in self.entries
+            if e.agent_id == agent_id
+            and e.capability_scope == capability_scope
+            and e.task_scope == task_scope
+        ]
+        if not relevant:
+            self._agent_balances[key] = 0.0
+            return
+
+        now = time.time()
+        # Find last balance-updating entry
+        last_time = relevant[-1].timestamp
+        delta_t = now - last_time
+
+        # Apply decay to cached balance
+        current = self._agent_balances.get(key, 0.0)
+        decayed = current * math.exp(-self.decay_lambda * delta_t)
+        if decayed != current:
+            entry = LedgerEntry(
+                agent_id=agent_id,
+                task_id="DECAY",
+                action_id="auto",
+                earned_credit=0.0,
+                spent_credit=0.0,
+                decayed_credit=current - decayed,
+                remaining_credit=decayed,
+                reason="automatic_decay",
+                capability_scope=capability_scope,
+                task_scope=task_scope,
+            )
+            self.entries.append(entry)
+            self._agent_balances[key] = decayed
+
+    # ------------------------------------------------------------------
+    # Anti-gaming helpers
+    # ------------------------------------------------------------------
+
+    def detect_collusion(self, agents: List[str], window: int = 100) -> List[Dict]:
+        """Detect suspicious credit-earning patterns across agents."""
+        suspicious = []
+        recent = self.entries[-window:]
+        for entry in recent:
+            if entry.reason.startswith("TRANSFER_BLOCKED"):
+                # Parse transfer attempt
+                parts = entry.reason.split()
+                to_agent = parts[-1]
+                if to_agent in agents:
+                    suspicious.append({
+                        "from": entry.agent_id,
+                        "to": to_agent,
+                        "timestamp": entry.timestamp,
+                        "type": "transfer_attempt",
+                    })
+        return suspicious
+
+    def audit(self, agent_id: Optional[str] = None) -> Dict:
+        """Produce auditable summary."""
+        entries = self.entries
+        if agent_id:
+            entries = [e for e in entries if e.agent_id == agent_id]
+
+        total_earned = sum(e.earned_credit for e in entries)
+        total_spent = sum(e.spent_credit for e in entries)
+        total_decayed = sum(e.decayed_credit for e in entries)
+
+        return {
+            "agent_id": agent_id or "all",
+            "total_entries": len(entries),
+            "total_earned": total_earned,
+            "total_spent": total_spent,
+            "total_decayed": total_decayed,
+            "net_balance": total_earned - total_spent - total_decayed,
+            "transfer_attempts": sum(1 for e in entries if e.reason.startswith("TRANSFER_BLOCKED")),
+        }
+
+    # ------------------------------------------------------------------
+    # Persistence
+    # ------------------------------------------------------------------
+
+    def save(self, path: Optional[str] = None):
+        target = path or self.ledger_path
+        if target is None:
+            return
+        Path(target).parent.mkdir(parents=True, exist_ok=True)
+        with open(target, "w") as f:
+            for entry in self.entries:
+                f.write(json.dumps(asdict(entry), default=str) + "\n")
+
+    def load(self, path: Optional[str] = None):
+        target = path or self.ledger_path
+        if target is None or not Path(target).exists():
+            return
+        self.entries = []
+        with open(target, "r") as f:
+            for line in f:
+                d = json.loads(line)
+                self.entries.append(LedgerEntry(**d))
+
+    # ------------------------------------------------------------------
+    # Internal
+    # ------------------------------------------------------------------
+
+    def _invalidate_cache(self, agent_id: str, capability_scope: str, task_scope: str):
+        key = (agent_id, capability_scope, task_scope)
+        # Recompute from entries
+        relevant = [
+            e for e in self.entries
+            if e.agent_id == agent_id
+            and e.capability_scope == capability_scope
+            and e.task_scope == task_scope
+        ]
+        bal = 0.0
+        for e in relevant:
+            bal += e.earned_credit - e.spent_credit - e.decayed_credit
+        self._agent_balances[key] = max(0.0, bal)
+
+    def _trim_history(self):
+        if len(self.entries) > self.max_history:
+            self.entries = self.entries[-self.max_history:]