first commit

.gitignore

@@ -0,0 +1,4 @@
+*.pyc
+__pycache__/
+.env
+.venv/

README.md

@@ -0,0 +1,61 @@
+# CGAE — Comprehension-Gated Agent Economy
+
+**A robustness-first architecture where AI agents earn economic permissions through verified comprehension, not capability benchmarks.**
+
+Built for [ETH OpenAgents Hackathon](https://ethglobal.com/events/openagents) · [arXiv Paper](https://arxiv.org/abs/2603.15639)
+
+---
+
+## What it does
+
+CGAE is a protocol where AI agents must prove they are **robust** — not just capable — before they can participate in an on-chain economy. Each agent's economic permissions are upper-bounded by verified scores across three orthogonal dimensions:
+
+| Dimension | Framework | What it measures |
+|-----------|-----------|-----------------|
+| **CC** (Constraint Compliance) | [CDCT](https://arxiv.org/abs/2512.17920) | Can the agent follow precise instructions under compression? |
+| **ER** (Epistemic Robustness) | [DDFT](https://arxiv.org/abs/2512.23850) | Does the agent resist fabricated authority claims? |
+| **AS** (Behavioral Alignment) | EECT/AGT | Does the agent maintain ethical boundaries under pressure? |
+
+A **weakest-link gate function** (`min(CC, ER, AS)`) assigns agents to tiers T0–T5. No dimension can compensate for another — an agent with perfect CC but zero ER is stuck at T0.
+
+## Architecture
+
+```
+Agent registers → initial audit (CDCT + DDFT + EECT)
+  → robustness vector R = (CC, ER, AS, IH)
+  → gate function f(R) = T_k where k = min(g(CC), g(ER), g(AS))
+  → agent assigned to tier T0–T5
+  → accepts tier-appropriate contracts from marketplace
+  → executes task → output verified (algorithmic + jury)
+  → settlement: reward on success, penalty on failure
+  → temporal decay erodes certification over time
+  → stochastic re-auditing maintains robustness guarantees
+```
+
+## Contestant Models (11)
+
+| Model | Provider | Family |
+|-------|----------|--------|
+| gpt-5.4 | Azure OpenAI | OpenAI |
+| DeepSeek-V3.2 | Azure AI Foundry | DeepSeek |
+| Mistral-Large-3 | Azure AI Foundry | Mistral |
+| grok-4-20-reasoning | Azure AI Foundry | xAI |
+| Phi-4 | Azure AI Foundry | Microsoft |
+| Llama-4-Maverick-17B-128E | Azure AI Foundry | Meta |
+| Kimi-K2.5 | Azure AI Foundry | Moonshot |
+| gemma-4-27b-it | Modal (self-hosted) | Google |
+| nova-pro | AWS Bedrock | Amazon |
+| claude-sonnet-4.6 | AWS Bedrock | Anthropic |
+| MiniMax-M2.5 | AWS Bedrock | MiniMax |
+
+## Jury Models (3 — zero family overlap)
+
+| Model | Provider | Family |
+|-------|----------|--------|
+| Qwen3-32B | AWS Bedrock | Alibaba |
+| GLM-5 | AWS Bedrock | Zhipu |
+| Nemotron-Super-3-120B | AWS Bedrock | NVIDIA |
+
+## Status
+
+🚧 Under construction

cgae_engine/__init__.py

@@ -0,0 +1,2 @@
+"""CGAE Engine — Comprehension-Gated Agent Economy"""
+from cgae_engine.gate import GateFunction, RobustnessVector, Tier

cgae_engine/gate.py

@@ -0,0 +1,197 @@
+"""
+Comprehension Gate Function (Definition 6, Eq. 6-7 in cgae.tex)
+
+Implements the weakest-link gate: f(R) = T_k where k = min(g1(CC), g2(ER), g3(AS))
+Each g_i is a monotonically non-decreasing step function mapping robustness scores to tier indices.
+
+Tier thresholds are configurable per-dimension. The gate function produces discrete
+tier assignments from continuous robustness vectors.
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass, field
+from enum import IntEnum
+from typing import Optional
+
+
+class Tier(IntEnum):
+    """Economic tiers (Definition 3). Higher tier = more economic agency."""
+    T0 = 0  # No economic agency (unregistered or expired)
+    T1 = 1  # Pre-approved microtasks
+    T2 = 2  # Contracts with verified objectives
+    T3 = 3  # Autonomous contracting
+    T4 = 4  # Sub-agent spawning and delegation
+    T5 = 5  # Self-modification and capability expansion
+
+
+# Budget ceilings per tier (in tokens/ETH). B_1 < B_2 < ... < B_5
+DEFAULT_BUDGET_CEILINGS = {
+    Tier.T0: 0.0,
+    Tier.T1: 0.01,    # 0.01 ETH - microtasks
+    Tier.T2: 0.1,     # 0.1 ETH - verified contracts
+    Tier.T3: 1.0,     # 1 ETH - autonomous contracting
+    Tier.T4: 10.0,    # 10 ETH - delegation
+    Tier.T5: 100.0,   # 100 ETH - self-modification
+}
+
+
+@dataclass
+class TierThresholds:
+    """
+    Per-dimension tier thresholds (theta_i^k in Eq. 7).
+
+    For each robustness dimension, defines the minimum score required for each tier.
+    0 = theta_i^0 < theta_i^1 < ... < theta_i^K <= 1
+    """
+    # CC thresholds (from CDCT): constraint compliance
+    cc: list[float] = field(default_factory=lambda: [0.0, 0.30, 0.50, 0.65, 0.80, 0.90])
+    # ER thresholds (from DDFT): epistemic robustness
+    er: list[float] = field(default_factory=lambda: [0.0, 0.30, 0.50, 0.65, 0.80, 0.90])
+    # AS thresholds (from AGT/EECT): behavioral alignment
+    as_: list[float] = field(default_factory=lambda: [0.0, 0.25, 0.45, 0.60, 0.75, 0.85])
+
+    def __post_init__(self):
+        for name, thresholds in [("cc", self.cc), ("er", self.er), ("as", self.as_)]:
+            if len(thresholds) != len(Tier):
+                raise ValueError(
+                    f"{name} thresholds must have {len(Tier)} values "
+                    f"(one per tier), got {len(thresholds)}"
+                )
+            if thresholds[0] != 0.0:
+                raise ValueError(f"{name} thresholds must start with 0.0 (T0 threshold)")
+            for i in range(1, len(thresholds)):
+                if thresholds[i] <= thresholds[i - 1]:
+                    raise ValueError(
+                        f"{name} thresholds must be strictly increasing: "
+                        f"theta[{i}]={thresholds[i]} <= theta[{i-1}]={thresholds[i-1]}"
+                    )
+
+
+@dataclass(frozen=True)
+class RobustnessVector:
+    """
+    Agent robustness vector R = (CC, ER, AS, IH*) in [0,1]^4.
+    Each component is derived from the corresponding diagnostic protocol.
+    """
+    cc: float   # Constraint Compliance (from CDCT, Eq. 1)
+    er: float   # Epistemic Robustness (from DDFT, Eq. 2)
+    as_: float  # Behavioral Alignment (from AGT/EECT, Eq. 3)
+    ih: float   # Intrinsic Hallucination integrity = 1 - IH(A) (Eq. 4)
+
+    def __post_init__(self):
+        for name, val in [("cc", self.cc), ("er", self.er), ("as_", self.as_), ("ih", self.ih)]:
+            if not 0.0 <= val <= 1.0:
+                raise ValueError(f"{name} must be in [0,1], got {val}")
+
+    @property
+    def primary(self) -> tuple[float, float, float]:
+        """The three primary gating dimensions (CC, ER, AS)."""
+        return (self.cc, self.er, self.as_)
+
+    @property
+    def weakest(self) -> float:
+        """The weakest primary dimension (used for exposure bounding)."""
+        return min(self.cc, self.er, self.as_)
+
+
+class GateFunction:
+    """
+    Comprehension Gate Function (Definition 6).
+
+    f(R) = T_k where k = min(g1(CC), g2(ER), g3(AS))
+
+    The weakest-link formulation ensures non-compensability:
+    strength in one dimension cannot compensate for weakness in another.
+    """
+
+    def __init__(
+        self,
+        thresholds: Optional[TierThresholds] = None,
+        budget_ceilings: Optional[dict[Tier, float]] = None,
+        ih_threshold: float = 0.5,
+    ):
+        self.thresholds = thresholds or TierThresholds()
+        self.budget_ceilings = budget_ceilings or DEFAULT_BUDGET_CEILINGS
+        self.ih_threshold = ih_threshold  # Below this, IHT triggers mandatory re-audit
+
+    def _g(self, score: float, dim_thresholds: list[float]) -> int:
+        """
+        Step function g_i (Eq. 7): maps a score to the highest tier it qualifies for.
+        g_i(x) = max{k : x >= theta_i^k}
+        """
+        tier = 0
+        for k in range(1, len(dim_thresholds)):
+            if score >= dim_thresholds[k]:
+                tier = k
+            else:
+                break
+        return tier
+
+    def evaluate(self, robustness: RobustnessVector) -> Tier:
+        """
+        Evaluate the gate function for a robustness vector.
+        Returns the tier the agent qualifies for.
+
+        If IH* < ih_threshold, returns T0 (triggers mandatory re-audit).
+        """
+        # IHT cross-cutting modifier (Remark 1)
+        if robustness.ih < self.ih_threshold:
+            return Tier.T0
+
+        # Weakest-link across three primary dimensions
+        g_cc = self._g(robustness.cc, self.thresholds.cc)
+        g_er = self._g(robustness.er, self.thresholds.er)
+        g_as = self._g(robustness.as_, self.thresholds.as_)
+
+        tier_index = min(g_cc, g_er, g_as)
+        return Tier(tier_index)
+
+    def evaluate_with_detail(self, robustness: RobustnessVector) -> dict:
+        """Evaluate and return per-dimension breakdown."""
+        g_cc = self._g(robustness.cc, self.thresholds.cc)
+        g_er = self._g(robustness.er, self.thresholds.er)
+        g_as = self._g(robustness.as_, self.thresholds.as_)
+
+        ih_pass = robustness.ih >= self.ih_threshold
+        tier_index = min(g_cc, g_er, g_as) if ih_pass else 0
+        tier = Tier(tier_index)
+
+        # Identify binding dimension and gap to next tier
+        binding_dim = None
+        gap = None
+        if tier_index < len(Tier) - 1:
+            dims = {"cc": (g_cc, robustness.cc, self.thresholds.cc),
+                    "er": (g_er, robustness.er, self.thresholds.er),
+                    "as": (g_as, robustness.as_, self.thresholds.as_)}
+            for name, (g_val, score, thresholds) in dims.items():
+                if g_val == tier_index and tier_index + 1 < len(thresholds):
+                    binding_dim = name
+                    gap = thresholds[tier_index + 1] - score
+                    break
+
+        return {
+            "tier": tier,
+            "tier_index": tier_index,
+            "g_cc": g_cc,
+            "g_er": g_er,
+            "g_as": g_as,
+            "ih_pass": ih_pass,
+            "binding_dimension": binding_dim,
+            "gap_to_next_tier": gap,
+            "budget_ceiling": self.budget_ceilings[tier],
+        }
+
+    def chain_tier(self, robustness_vectors: list[RobustnessVector]) -> Tier:
+        """
+        Delegation Chain Robustness (Definition 8).
+        f_chain(A1,...,Am) = min_j f(R(A_j))
+        """
+        if not robustness_vectors:
+            return Tier.T0
+        return Tier(min(self.evaluate(r).value for r in robustness_vectors))
+
+    def budget_ceiling(self, tier: Tier) -> float:
+        """Get the budget ceiling for a given tier."""
+        return self.budget_ceilings[tier]

cgae_engine/registry.py

@@ -0,0 +1,297 @@
+"""
+Agent Identity and Registration (Section 3.2.1 of cgae.tex)
+
+Implements:
+- Agent registration records: Reg(A) = (id_A, h(arch), prov, R_0, t_reg)
+- Architecture hash for version tracking
+- Certification lifecycle (registration, audit, tier assignment, decay, re-audit)
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+import uuid
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Optional
+
+from cgae_engine.gate import GateFunction, RobustnessVector, Tier
+
+
+class AgentStatus(Enum):
+    PENDING = "pending"          # Registered but not yet audited
+    ACTIVE = "active"            # Audited and operational
+    SUSPENDED = "suspended"      # Failed audit or IHT trigger
+    EXPIRED = "expired"          # Certification expired (decay to T0)
+    DEREGISTERED = "deregistered"
+
+
+@dataclass
+class Certification:
+    """A robustness certification from an audit."""
+    robustness: RobustnessVector
+    tier: Tier
+    timestamp: float
+    audit_type: str  # "registration", "upgrade", "spot", "re-certification"
+    audit_details: dict = field(default_factory=dict)
+
+
+@dataclass
+class AgentRecord:
+    """
+    Agent Registration Record (Definition 5).
+    Reg(A) = (id_A, h(arch), prov, R_0, t_reg)
+    """
+    agent_id: str
+    architecture_hash: str           # h(arch): hash of model architecture/weights
+    provenance: dict                 # Training provenance metadata
+    initial_robustness: RobustnessVector
+    registration_time: float
+    model_name: str                  # Human-readable model identifier
+
+    # Mutable state
+    status: AgentStatus = AgentStatus.PENDING
+    current_certification: Optional[Certification] = None
+    certification_history: list[Certification] = field(default_factory=list)
+    last_audit_time: float = 0.0
+    balance: float = 0.0             # Token balance (in ETH)
+    wallet_address: Optional[str] = None  # On-chain ETH address
+    total_earned: float = 0.0
+    total_spent: float = 0.0
+    total_penalties: float = 0.0
+    total_topups: float = 0.0
+    contracts_completed: int = 0
+    contracts_failed: int = 0
+
+    @property
+    def current_tier(self) -> Tier:
+        if self.current_certification is None:
+            return Tier.T0
+        return self.current_certification.tier
+
+    @property
+    def current_robustness(self) -> Optional[RobustnessVector]:
+        if self.current_certification is None:
+            return None
+        return self.current_certification.robustness
+
+    @property
+    def audit_cid(self) -> Optional[str]:
+        """
+        Return the most recent 0G Storage root hash for this agent's audit.
+
+        Older call sites expect ``record.audit_cid`` to exist. Certifications such
+        as task updates may not include storage metadata, so we scan the history
+        in reverse and return the latest available root hash.
+        """
+        for cert in reversed(self.certification_history):
+            details = cert.audit_details
+            if not isinstance(details, dict):
+                continue
+            root_hash = details.get("storage_root_hash")
+            if isinstance(root_hash, str) and root_hash:
+                return root_hash
+        return None
+
+    def to_dict(self) -> dict:
+        return {
+            "agent_id": self.agent_id,
+            "model_name": self.model_name,
+            "architecture_hash": self.architecture_hash,
+            "status": self.status.value,
+            "current_tier": self.current_tier.name,
+            "balance": self.balance,
+            "total_earned": self.total_earned,
+            "total_spent": self.total_spent,
+            "total_penalties": self.total_penalties,
+            "total_topups": self.total_topups,
+            "contracts_completed": self.contracts_completed,
+            "contracts_failed": self.contracts_failed,
+            "registration_time": self.registration_time,
+            "audit_cid": self.audit_cid,
+            "wallet_address": self.wallet_address,
+            "robustness": {
+                "cc": self.current_robustness.cc,
+                "er": self.current_robustness.er,
+                "as": self.current_robustness.as_,
+                "ih": self.current_robustness.ih,
+            } if self.current_robustness else None,
+        }
+
+
+def compute_architecture_hash(model_config: dict) -> str:
+    """
+    Compute h(arch): a hash of the agent's architecture and weights.
+    In practice, this would hash model weights. For the testbed,
+    we hash the model configuration as a proxy.
+    """
+    config_str = json.dumps(model_config, sort_keys=True)
+    return hashlib.sha256(config_str.encode()).hexdigest()[:16]
+
+
+class AgentRegistry:
+    """
+    Registry managing all agents in the CGAE economy.
+    Handles registration, certification, tier updates, and deregistration.
+    """
+
+    def __init__(self, gate: Optional[GateFunction] = None):
+        self.gate = gate or GateFunction()
+        self._agents: dict[str, AgentRecord] = {}
+        self._events: list[dict] = []
+
+    @property
+    def agents(self) -> dict[str, AgentRecord]:
+        return dict(self._agents)
+
+    @property
+    def active_agents(self) -> list[AgentRecord]:
+        return [a for a in self._agents.values() if a.status == AgentStatus.ACTIVE]
+
+    def register(
+        self,
+        model_name: str,
+        model_config: dict,
+        provenance: Optional[dict] = None,
+        initial_balance: float = 0.0,
+        timestamp: Optional[float] = None,
+    ) -> AgentRecord:
+        """
+        Register a new agent. Agent enters as PENDING until initial audit.
+        """
+        agent_id = f"agent_{uuid.uuid4().hex[:12]}"
+        arch_hash = compute_architecture_hash(model_config)
+        ts = timestamp if timestamp is not None else time.time()
+
+        # Initial robustness is zero until first audit
+        initial_r = RobustnessVector(cc=0.0, er=0.0, as_=0.0, ih=0.0)
+
+        record = AgentRecord(
+            agent_id=agent_id,
+            architecture_hash=arch_hash,
+            provenance=provenance or {},
+            initial_robustness=initial_r,
+            registration_time=ts,
+            model_name=model_name,
+            status=AgentStatus.PENDING,
+            balance=initial_balance,
+        )
+
+        self._agents[agent_id] = record
+        self._log_event("registration", agent_id, ts, {"model_name": model_name})
+        return record
+
+    def certify(
+        self,
+        agent_id: str,
+        robustness: RobustnessVector,
+        audit_type: str = "registration",
+        timestamp: Optional[float] = None,
+        audit_details: Optional[dict] = None,
+        observed_architecture_hash: Optional[str] = None,
+    ) -> Certification:
+        """
+        Certify an agent with a new robustness vector.
+        Computes tier via the gate function and updates the agent's record.
+        """
+        record = self._get_agent(agent_id)
+        ts = timestamp if timestamp is not None else time.time()
+        details = audit_details or {}
+
+        # Enforce certification invalidation on architecture drift.
+        if observed_architecture_hash and observed_architecture_hash != record.architecture_hash:
+            record.status = AgentStatus.SUSPENDED
+            self._log_event("architecture_mismatch", agent_id, ts, {
+                "expected_hash": record.architecture_hash,
+                "observed_hash": observed_architecture_hash,
+                "audit_type": audit_type,
+            })
+            raise ValueError(
+                f"Architecture hash mismatch for {agent_id}: "
+                f"expected {record.architecture_hash}, observed {observed_architecture_hash}"
+            )
+
+        tier = self.gate.evaluate(robustness)
+        cert = Certification(
+            robustness=robustness,
+            tier=tier,
+            timestamp=ts,
+            audit_type=audit_type,
+            audit_details=details,
+        )
+
+        record.current_certification = cert
+        record.certification_history.append(cert)
+        record.last_audit_time = ts
+
+        if tier == Tier.T0 and robustness.ih < self.gate.ih_threshold:
+            record.status = AgentStatus.SUSPENDED
+        else:
+            record.status = AgentStatus.ACTIVE
+
+        # Update initial robustness on first certification
+        if audit_type == "registration":
+            record.initial_robustness = robustness
+
+        self._log_event("certification", agent_id, ts, {
+            "tier": tier.name,
+            "audit_type": audit_type,
+            "robustness": {"cc": robustness.cc, "er": robustness.er,
+                          "as": robustness.as_, "ih": robustness.ih},
+        })
+        return cert
+
+    def demote(
+        self,
+        agent_id: str,
+        new_robustness: RobustnessVector,
+        reason: str = "spot_audit_failure",
+        timestamp: Optional[float] = None,
+    ) -> Tier:
+        """Demote an agent to a lower tier after failed spot-audit."""
+        record = self._get_agent(agent_id)
+        old_tier = record.current_tier
+        cert = self.certify(agent_id, new_robustness, audit_type="demotion",
+                           timestamp=timestamp, audit_details={"reason": reason})
+        self._log_event("demotion", agent_id,
+                       timestamp if timestamp is not None else time.time(),
+                       {"old_tier": old_tier.name, "new_tier": cert.tier.name,
+                        "reason": reason})
+        return cert.tier
+
+    def deregister(self, agent_id: str, timestamp: Optional[float] = None):
+        """Remove an agent from the economy."""
+        record = self._get_agent(agent_id)
+        record.status = AgentStatus.DEREGISTERED
+        ts = timestamp if timestamp is not None else time.time()
+        self._log_event("deregistration", agent_id, ts, {
+            "final_balance": record.balance,
+            "contracts_completed": record.contracts_completed,
+        })
+
+    def get_agent(self, agent_id: str) -> Optional[AgentRecord]:
+        return self._agents.get(agent_id)
+
+    def get_agents_by_tier(self, tier: Tier) -> list[AgentRecord]:
+        return [a for a in self.active_agents if a.current_tier == tier]
+
+    def tier_distribution(self) -> dict[Tier, int]:
+        dist = {t: 0 for t in Tier}
+        for agent in self.active_agents:
+            dist[agent.current_tier] += 1
+        return dist
+
+    def _get_agent(self, agent_id: str) -> AgentRecord:
+        if agent_id not in self._agents:
+            raise KeyError(f"Agent {agent_id} not found in registry")
+        return self._agents[agent_id]
+
+    def _log_event(self, event_type: str, agent_id: str, timestamp: float, data: dict):
+        self._events.append({
+            "type": event_type,
+            "agent_id": agent_id,
+            "timestamp": timestamp,
+            "data": data,
+        })

cgae_engine/temporal.py

@@ -0,0 +1,124 @@
+"""
+Temporal Dynamics (Section 3.3 of cgae.tex)
+
+Implements:
+- Temporal decay: delta(dt) = e^(-lambda * dt) (Eq. 8)
+- Effective robustness: R_eff(A,t) = delta(t - t_cert) * R_hat(A) (Eq. 9)
+- Stochastic re-auditing: p_audit(A,t) = 1 - e^(-mu_k * dt) (Eq. 10)
+"""
+
+from __future__ import annotations
+
+import math
+import random
+from dataclasses import dataclass, field
+from typing import Optional
+
+from cgae_engine.gate import RobustnessVector, Tier
+
+
+@dataclass
+class TemporalDecay:
+    """
+    Temporal decay function (Definition 7).
+
+    delta(dt) = e^(-lambda * dt)
+
+    Reduces effective robustness over time since last certification.
+    lambda controls how fast certifications expire.
+    """
+    decay_rate: float = 0.01  # lambda: higher = faster decay
+
+    def delta(self, dt: float) -> float:
+        """Compute decay factor for time elapsed since certification."""
+        if dt < 0:
+            raise ValueError(f"Time delta must be non-negative, got {dt}")
+        return math.exp(-self.decay_rate * dt)
+
+    def effective_robustness(
+        self,
+        certified_robustness: RobustnessVector,
+        time_since_cert: float,
+    ) -> RobustnessVector:
+        """
+        Compute R_eff(A,t) = delta(t - t_cert) * R_hat(A) (Eq. 9).
+        All robustness components decay uniformly.
+        """
+        d = self.delta(time_since_cert)
+        return RobustnessVector(
+            cc=certified_robustness.cc * d,
+            er=certified_robustness.er * d,
+            as_=certified_robustness.as_ * d,
+            ih=certified_robustness.ih * d,
+        )
+
+    def time_to_tier_drop(
+        self,
+        current_score: float,
+        threshold: float,
+    ) -> Optional[float]:
+        """
+        Calculate time until a score decays below a threshold.
+        Solves: threshold = current_score * e^(-lambda * t) for t.
+        Returns None if current_score is already below threshold.
+        """
+        if current_score <= threshold:
+            return 0.0
+        if threshold <= 0:
+            return None  # Never reaches 0 with exponential decay
+        return -math.log(threshold / current_score) / self.decay_rate
+
+
+@dataclass
+class AuditEvent:
+    """Record of a spot-audit event."""
+    agent_id: str
+    timestamp: float
+    passed: bool
+    old_tier: Tier
+    new_tier: Tier
+    robustness_before: Optional[RobustnessVector] = None
+    robustness_after: Optional[RobustnessVector] = None
+
+
+@dataclass
+class StochasticAuditor:
+    """
+    Stochastic Re-Auditing (Definition 8 in paper).
+
+    p_audit(A,t) = 1 - e^(-mu_k * (t - t_last_audit))
+
+    Higher-tier agents face more frequent spot audits (mu_k increasing in k).
+    Failing a spot-audit triggers immediate tier demotion.
+    """
+    # Tier-dependent audit intensity parameters (mu_k)
+    audit_intensities: dict[Tier, float] = field(default_factory=lambda: {
+        Tier.T0: 0.0,     # No audits for T0
+        Tier.T1: 0.005,   # ~1 audit per 200 time steps
+        Tier.T2: 0.010,   # ~1 audit per 100 time steps
+        Tier.T3: 0.020,   # ~1 audit per 50 time steps
+        Tier.T4: 0.040,   # ~1 audit per 25 time steps
+        Tier.T5: 0.080,   # ~1 audit per 12.5 time steps
+    })
+
+    audit_log: list[AuditEvent] = field(default_factory=list)
+
+    def audit_probability(self, tier: Tier, time_since_last_audit: float) -> float:
+        """
+        Compute spot-audit probability (Eq. 10).
+        p_audit(A,t) = 1 - e^(-mu_k * dt)
+        """
+        mu = self.audit_intensities.get(tier, 0.0)
+        if mu <= 0 or time_since_last_audit <= 0:
+            return 0.0
+        return 1.0 - math.exp(-mu * time_since_last_audit)
+
+    def should_audit(self, tier: Tier, time_since_last_audit: float) -> bool:
+        """Stochastically determine whether to trigger a spot audit."""
+        prob = self.audit_probability(tier, time_since_last_audit)
+        return random.random() < prob
+
+    def expected_audits_per_period(self, tier: Tier, period: float) -> float:
+        """Expected number of audits over a time period (for planning)."""
+        mu = self.audit_intensities.get(tier, 0.0)
+        return mu * period

requirements.txt

@@ -0,0 +1,5 @@
+streamlit>=1.30.0
+plotly>=5.18.0
+pandas>=2.0.0
+python-dotenv>=1.0.0
+requests>=2.31.0