ledger/ledger.py

"""
Credit Ledger - non-transferable, decaying credits with full provenance.
"""
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional


@dataclass
class LedgerEntry:
    agent_id: str
    task_id: str
    action_id: str
    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"


class CreditLedger:
    def __init__(self, decay_lambda: float = 0.05):
        self.entries: List[LedgerEntry] = []
        self.agent_balances: Dict[str, Dict[str, Dict[str, float]]] = {}
        self.decay_lambda = decay_lambda

    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 = "global",
        task_scope: str = "global",
    ) -> None:
        now = time.time()
        self._apply_decay(agent_id, now, capability_scope, task_scope)

        current = self._get_balance(agent_id, capability_scope, task_scope)
        new_balance = current + amount

        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=new_balance,
            reason=reason,
            oracle_score=oracle_score,
            compute_cost=compute_cost,
            timestamp=now,
            capability_scope=capability_scope,
            task_scope=task_scope,
        )
        self.entries.append(entry)
        self._set_balance(agent_id, capability_scope, task_scope, new_balance)

    def spend(
        self,
        agent_id: str,
        task_id: str,
        action_id: str,
        amount: float,
        capability_scope: str = "global",
        task_scope: str = "global",
        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

        new_balance = current - amount
        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=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._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 = "global",
        task_scope: str = "global",
    ) -> bool:
        # CREDITS ARE NON-TRANSFERABLE
        return False

    def balance(
        self,
        agent_id: str,
        capability_scope: str = "global",
        task_scope: str = "global",
    ) -> float:
        now = time.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

        # 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="decay",
                earned_credit=0.0,
                spent_credit=0.0,
                decayed_credit=current - decayed,
                remaining_credit=decayed,
                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._set_balance(agent_id, capability_scope, task_scope, new_balance)
            return True
        return False

    def provenance(self, agent_id: str) -> List[LedgerEntry]:
        return [e for e in self.entries if e.agent_id == agent_id]

    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,
        )