Upload ledger/ledger.py

ledger/ledger.py

@@ -1,13 +1,9 @@
 """
-Credit Ledger: non-transferable, decaying, capability-scoped credits with provenance.
+Credit Ledger - non-transferable, decaying credits with full provenance.
 """
-
-import json
-import math
 import time
-from dataclasses import asdict, dataclass, field
-from pathlib import Path
-from typing import Dict, List, Optional, Tuple
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
 
 
 @dataclass
@@ -15,50 +11,23 @@ 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"
+    earned_credit: float
+    spent_credit: float
+    decayed_credit: float
+    remaining_credit: float
+    reason: str
+    oracle_score: float
+    compute_cost: float
+    timestamp: float
+    capability_scope: str = "global"
     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
+    def __init__(self, decay_lambda: float = 0.05):
         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
-    # ------------------------------------------------------------------
+        self.agent_balances: Dict[str, Dict[str, Dict[str, float]]] = {}
+        self.decay_lambda = decay_lambda
 
     def earn(
         self,
@@ -69,20 +38,14 @@ class CreditLedger:
         oracle_score: float,
         compute_cost: float,
         reason: str,
-        capability_scope: str = "general",
+        capability_scope: str = "global",
         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")
+    ) -> None:
+        now = time.time()
+        self._apply_decay(agent_id, now, capability_scope, task_scope)
 
-        # 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]"
+        current = self._get_balance(agent_id, capability_scope, task_scope)
+        new_balance = current + amount
 
         entry = LedgerEntry(
             agent_id=agent_id,
@@ -91,18 +54,16 @@ class CreditLedger:
             earned_credit=amount,
             spent_credit=0.0,
             decayed_credit=0.0,
-            remaining_credit=amount,
+            remaining_credit=new_balance,
             reason=reason,
             oracle_score=oracle_score,
             compute_cost=compute_cost,
+            timestamp=now,
             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
+        self._set_balance(agent_id, capability_scope, task_scope, new_balance)
 
     def spend(
         self,
@@ -110,20 +71,18 @@ class CreditLedger:
         task_id: str,
         action_id: str,
         amount: float,
-        capability_scope: str = "general",
+        capability_scope: str = "global",
         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)
+        reason: str = "spend",
+    ) -> bool:
+        now = time.time()
+        self._apply_decay(agent_id, now, capability_scope, task_scope)
 
+        current = self._get_balance(agent_id, capability_scope, task_scope)
         if current < amount:
-            return False, None
+            return False
 
+        new_balance = current - amount
         entry = LedgerEntry(
             agent_id=agent_id,
             task_id=task_id,
@@ -131,212 +90,124 @@ class CreditLedger:
             earned_credit=0.0,
             spent_credit=amount,
             decayed_credit=0.0,
-            remaining_credit=current - amount,
+            remaining_credit=new_balance,
             reason=reason,
+            oracle_score=0.0,
+            compute_cost=0.0,
+            timestamp=now,
             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
+        self._set_balance(agent_id, capability_scope, task_scope, new_balance)
+        return True
 
     def transfer(
         self,
         from_agent: str,
         to_agent: str,
         amount: float,
-        capability_scope: str = "general",
+        capability_scope: str = "global",
         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)
+        # CREDITS ARE NON-TRANSFERABLE
         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",
+        capability_scope: str = "global",
         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
+        self._apply_decay(agent_id, now, capability_scope, task_scope)
+        return self._get_balance(agent_id, capability_scope, task_scope)
+
+    def _get_balance(self, agent_id: str, cap: str, task: str) -> float:
+        return self.agent_balances.get(agent_id, {}).get(cap, {}).get(task, 0.0)
+
+    def _set_balance(self, agent_id: str, cap: str, task: str, val: float) -> None:
+        if agent_id not in self.agent_balances:
+            self.agent_balances[agent_id] = {}
+        if cap not in self.agent_balances[agent_id]:
+            self.agent_balances[agent_id][cap] = {}
+        self.agent_balances[agent_id][cap][task] = val
+
+    def _apply_decay(self, agent_id: str, now: float, cap: str, task: str) -> None:
+        current = self._get_balance(agent_id, cap, task)
+        if current <= 0:
+            return
 
-        # 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:
+        # Exponential decay of idle credits
+        decayed = current * (1 - self.decay_lambda)
+        if decayed < current:
             entry = LedgerEntry(
                 agent_id=agent_id,
-                task_id="DECAY",
-                action_id="auto",
+                task_id="decay",
+                action_id="decay",
                 earned_credit=0.0,
                 spent_credit=0.0,
                 decayed_credit=current - decayed,
                 remaining_credit=decayed,
-                reason="automatic_decay",
+                reason="credit_decay",
+                oracle_score=0.0,
+                compute_cost=0.0,
+                timestamp=now,
+                capability_scope=cap,
+                task_scope=task,
+            )
+            self.entries.append(entry)
+            self._set_balance(agent_id, cap, task, decayed)
+
+    def revoke(
+        self,
+        agent_id: str,
+        task_id: str,
+        amount: float,
+        reason: str,
+        capability_scope: str = "global",
+        task_scope: str = "global",
+    ) -> bool:
+        now = time.time()
+        self._apply_decay(agent_id, now, capability_scope, task_scope)
+        current = self._get_balance(agent_id, capability_scope, task_scope)
+        revoke_amount = min(current, amount)
+        if revoke_amount > 0:
+            new_balance = current - revoke_amount
+            entry = LedgerEntry(
+                agent_id=agent_id,
+                task_id=task_id,
+                action_id="revoke",
+                earned_credit=0.0,
+                spent_credit=revoke_amount,
+                decayed_credit=0.0,
+                remaining_credit=new_balance,
+                reason=f"revoke: {reason}",
+                oracle_score=0.0,
+                compute_cost=0.0,
+                timestamp=now,
                 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
-    # ------------------------------------------------------------------
+            self._set_balance(agent_id, capability_scope, task_scope, new_balance)
+            return True
+        return False
 
-    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 provenance(self, agent_id: str) -> List[LedgerEntry]:
+        return [e for e in self.entries if e.agent_id == agent_id]
 
-    def _trim_history(self):
-        if len(self.entries) > self.max_history:
-            self.entries = self.entries[-self.max_history:]
+    def gaming_detected(
+        self,
+        agent_id: str,
+        task_id: str,
+        action_id: str,
+        reason: str,
+        capability_scope: str = "global",
+    ) -> None:
+        # Immediate revocation of credits on gaming detection
+        self.revoke(
+            agent_id, task_id, 999.0,
+            reason=f"gaming_detected: {reason}",
+            capability_scope=capability_scope,
+        )