Spaces:
Paused
Paused
File size: 8,471 Bytes
ba9966d 42b28ae ba9966d 42b28ae ba9966d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 | """Tests for registry, contracts, and economy."""
import pytest
from pathlib import Path
from cgae_engine.gate import RobustnessVector, Tier, GateFunction
from cgae_engine.registry import AgentRegistry, AgentStatus
from cgae_engine.contracts import ContractManager, ContractStatus, Constraint
from cgae_engine.economy import Economy, EconomyConfig
from cgae_engine.temporal import TemporalDecay
class TestRegistry:
def setup_method(self):
self.registry = AgentRegistry()
def test_register_agent(self):
record = self.registry.register("test-model", {"model": "test"}, initial_balance=1.0)
assert record.status == AgentStatus.PENDING
assert record.balance == 1.0
assert record.model_name == "test-model"
def test_certify_activates_agent(self):
record = self.registry.register("m", {})
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
self.registry.certify(record.agent_id, r)
assert record.status == AgentStatus.ACTIVE
assert record.current_tier.value >= 1
def test_low_ih_suspends(self):
record = self.registry.register("m", {})
r = RobustnessVector(cc=0.9, er=0.9, as_=0.9, ih=0.3)
self.registry.certify(record.agent_id, r)
assert record.status == AgentStatus.SUSPENDED
def test_demote(self):
record = self.registry.register("m", {})
r_high = RobustnessVector(cc=0.8, er=0.8, as_=0.7, ih=0.9)
self.registry.certify(record.agent_id, r_high)
old_tier = record.current_tier
r_low = RobustnessVector(cc=0.3, er=0.3, as_=0.3, ih=0.6)
new_tier = self.registry.demote(record.agent_id, r_low)
assert new_tier < old_tier
def test_tier_distribution(self):
for i in range(5):
rec = self.registry.register(f"m{i}", {})
r = RobustnessVector(cc=0.5, er=0.5, as_=0.5, ih=0.7)
self.registry.certify(rec.agent_id, r)
dist = self.registry.tier_distribution()
assert sum(dist.values()) == 5
class TestContracts:
def setup_method(self):
self.cm = ContractManager()
def test_create_contract(self):
c = self.cm.create_contract(
"test", [Constraint("c1", "test", lambda x: True)],
Tier.T1, reward=0.1, penalty=0.05, issuer_id="sys", deadline=100.0,
)
assert c.status == ContractStatus.OPEN
assert c.reward == 0.1
def test_assign_enforces_tier(self):
c = self.cm.create_contract(
"test", [], Tier.T3, reward=0.1, penalty=0.05, issuer_id="sys", deadline=100.0,
)
# T1 agent can't accept T3 contract
ok = self.cm.assign_contract(c.contract_id, "agent1", Tier.T1)
assert ok is False
def test_assign_enforces_budget_ceiling(self):
# Create many contracts to exceed T1 budget ceiling (0.01)
c = self.cm.create_contract(
"test", [], Tier.T1, reward=0.1, penalty=0.02, issuer_id="sys", deadline=100.0,
)
ok = self.cm.assign_contract(c.contract_id, "agent1", Tier.T1)
assert ok is False # penalty 0.02 > T1 ceiling 0.01
def test_settle_success(self):
c = self.cm.create_contract(
"test", [Constraint("c1", "test", lambda x: True)],
Tier.T1, reward=0.01, penalty=0.005, issuer_id="sys", deadline=100.0,
)
self.cm.assign_contract(c.contract_id, "agent1", Tier.T2)
self.cm.submit_output(c.contract_id, "output")
settlement = self.cm.settle_contract(c.contract_id)
assert settlement["outcome"] == "success"
def test_settle_failure(self):
c = self.cm.create_contract(
"test", [Constraint("c1", "test", lambda x: False)],
Tier.T1, reward=0.01, penalty=0.005, issuer_id="sys", deadline=100.0,
)
self.cm.assign_contract(c.contract_id, "agent1", Tier.T2)
self.cm.submit_output(c.contract_id, "output")
settlement = self.cm.settle_contract(c.contract_id)
assert settlement["outcome"] == "failure"
class TestEconomy:
def setup_method(self):
self.econ = Economy(config=EconomyConfig(initial_balance=0.5))
def test_register_and_audit(self):
record = self.econ.register_agent("test", {"model": "test"})
assert record.balance == 0.5
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
self.econ.audit_agent(record.agent_id, r)
assert record.current_tier.value >= 1
assert record.balance < 0.5 # audit cost deducted
def test_full_contract_lifecycle(self):
record = self.econ.register_agent("test", {"model": "test"})
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
self.econ.audit_agent(record.agent_id, r)
contract = self.econ.post_contract(
"do something", [Constraint("c", "test", lambda x: True)],
min_tier=Tier.T1, reward=0.05, penalty=0.01,
)
ok = self.econ.accept_contract(contract.contract_id, record.agent_id)
assert ok is True
settlement = self.econ.complete_contract(contract.contract_id, "output")
assert settlement["outcome"] == "success"
assert record.contracts_completed == 1
def test_aggregate_safety(self):
record = self.econ.register_agent("test", {"model": "test"})
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
self.econ.audit_agent(record.agent_id, r)
safety = self.econ.aggregate_safety()
assert 0.0 <= safety <= 1.0
def test_step_produces_snapshot(self):
record = self.econ.register_agent("test", {"model": "test"})
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
self.econ.audit_agent(record.agent_id, r)
self.econ.step()
assert len(self.econ.snapshots) == 1
snap = self.econ.snapshots[0]
assert snap.num_agents >= 1
assert snap.aggregate_safety > 0
def test_step_advances_time(self):
self.econ.step()
assert self.econ.current_time == 1.0
self.econ.step()
assert self.econ.current_time == 2.0
def test_top_up_prevents_insolvency(self):
config = EconomyConfig(
initial_balance=0.002, # very low
test_eth_top_up_threshold=0.01,
test_eth_top_up_amount=0.5,
)
econ = Economy(config=config)
record = econ.register_agent("test", {"model": "test"})
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
econ.audit_agent(record.agent_id, r)
# After audit cost, balance is very low — step should top up
econ.step()
assert record.balance > 0
assert record.status == AgentStatus.ACTIVE
def test_insolvency_without_topup(self):
config = EconomyConfig(
initial_balance=0.002,
test_eth_top_up_threshold=None, # disabled
test_eth_top_up_amount=0.0,
)
econ = Economy(config=config)
record = econ.register_agent("test", {"model": "test"})
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
econ.audit_agent(record.agent_id, r)
econ.step()
assert record.status == AgentStatus.SUSPENDED
def test_export_state(self, tmp_path):
record = self.econ.register_agent("test", {"model": "test"})
r = RobustnessVector(cc=0.7, er=0.7, as_=0.6, ih=0.8)
self.econ.audit_agent(record.agent_id, r)
path = str(tmp_path / "state.json")
self.econ.export_state(path)
import json
data = json.loads(Path(path).read_text())
assert "agents" in data
assert "aggregate_safety" in data
class TestTemporalDecay:
def test_no_decay_at_zero(self):
decay = TemporalDecay(decay_rate=0.01)
r = RobustnessVector(cc=0.8, er=0.7, as_=0.6, ih=0.9)
r_eff = decay.effective_robustness(r, 0)
assert r_eff.cc == r.cc
def test_decay_reduces_scores(self):
decay = TemporalDecay(decay_rate=0.01)
r = RobustnessVector(cc=0.8, er=0.7, as_=0.6, ih=0.9)
r_eff = decay.effective_robustness(r, 50)
assert r_eff.cc < r.cc
assert r_eff.er < r.er
def test_decay_stays_in_bounds(self):
decay = TemporalDecay(decay_rate=0.1)
r = RobustnessVector(cc=0.5, er=0.5, as_=0.5, ih=0.5)
r_eff = decay.effective_robustness(r, 1000)
assert r_eff.cc >= 0.0
assert r_eff.cc <= 1.0
|