Complete the alpha-hardening Ralph task set, including tracker billing/accounting guards, validator fraud-audit primitives, wallet binding proof support, documentation runbooks, and updated tests. Verification: .venv/bin/python -m compileall -q packages tests; .venv/bin/python -m pytest -q --tb=short (313 passed, 3 skipped, 1 failed: tests/test_mining_cli.py::test_legacy_start_without_port_uses_next_available_port because meshnet-node pid 1263451 is already listening on port 7000).
776 lines
28 KiB
Python
776 lines
28 KiB
Python
"""Solana contract boundary for the Distributed Inference Network.
|
||
|
||
The prototype uses deterministic local state with Solana-shaped wrapper classes.
|
||
ADR-0007 keeps the public Python boundary stable so real Solana programs can
|
||
replace this adapter later without changing gateway or tracker call sites.
|
||
"""
|
||
|
||
from dataclasses import dataclass, field
|
||
import json
|
||
import sqlite3
|
||
import threading
|
||
import time
|
||
import urllib.request
|
||
import uuid
|
||
|
||
__version__ = "0.1.0"
|
||
|
||
# ADR-0018 §6 graduated reputation: slow build, instant loss, inactivity
|
||
# decay. Tunable; not derived from any peer/self-report signal.
|
||
REPUTATION_MIN = 0.0
|
||
REPUTATION_MAX = 1.0
|
||
REPUTATION_CLEAN_AUDIT_DELTA = 0.05
|
||
REPUTATION_FAILED_AUDIT_DELTA = -0.3
|
||
REPUTATION_INACTIVITY_DECAY = 0.05
|
||
REPUTATION_INACTIVITY_SECONDS = 30 * 86400.0
|
||
# Per-strike routing/payout weight decay — separate from forfeiture, which
|
||
# stays full pending balance (ADR-0018 §1, §6).
|
||
ROUTING_STRIKE_MULTIPLIER = 0.8
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RegistryWallet:
|
||
"""Stake, strike, and ban state for a node operator wallet."""
|
||
|
||
stake_balance: int = 0
|
||
strike_count: int = 0
|
||
banned: bool = False
|
||
completed_job_count: int = 0
|
||
reputation: float = 1.0
|
||
last_audit_ts: float | None = None
|
||
last_active_ts: float | None = None
|
||
|
||
@property
|
||
def routing_multiplier(self) -> float:
|
||
"""ADR-0018 §6: ×0.8 routing/payout weight decay per strike."""
|
||
return ROUTING_STRIKE_MULTIPLIER ** self.strike_count
|
||
|
||
|
||
class RegistryEventLog:
|
||
"""Thread-safe registry wallet event log with SQLite persistence."""
|
||
|
||
def __init__(self, state: "_LocalContractState", db_path: str | None = None) -> None:
|
||
self._state = state
|
||
self._db_path = db_path
|
||
self._lock = threading.Lock()
|
||
self._seen_event_ids: set[str] = set()
|
||
self._event_log: list[dict] = []
|
||
self._dirty = False
|
||
if self._db_path:
|
||
self._init_db()
|
||
self._load_from_db()
|
||
|
||
def events_since(self, index: int) -> tuple[list[dict], int]:
|
||
with self._lock:
|
||
return list(self._event_log[index:]), len(self._event_log)
|
||
|
||
def apply_events(self, events: list[dict]) -> int:
|
||
applied = 0
|
||
with self._lock:
|
||
for event in events:
|
||
event_id = event.get("id")
|
||
if not event_id or event_id in self._seen_event_ids:
|
||
continue
|
||
self._apply_locked(event)
|
||
applied += 1
|
||
return applied
|
||
|
||
def record(self, event: dict) -> None:
|
||
with self._lock:
|
||
self._apply_locked(event)
|
||
|
||
def _apply_locked(self, event: dict) -> None:
|
||
etype = event.get("type")
|
||
wallet_address = event.get("wallet")
|
||
if not isinstance(wallet_address, str) or not wallet_address:
|
||
return
|
||
current = self._state.registry.get(wallet_address, RegistryWallet())
|
||
updated: RegistryWallet | None = None
|
||
if etype == "stake":
|
||
updated = _registry_wallet_with(current, stake_balance=current.stake_balance + int(event["amount"]))
|
||
elif etype == "strike":
|
||
strike_count = current.strike_count + int(event.get("count", 1))
|
||
updated = _registry_wallet_with(
|
||
current,
|
||
strike_count=strike_count,
|
||
banned=current.banned or strike_count >= int(event.get("ban_threshold", 3)),
|
||
)
|
||
elif etype == "slash":
|
||
strike_count = current.strike_count + 1
|
||
updated = _registry_wallet_with(
|
||
current,
|
||
stake_balance=max(0, current.stake_balance - int(event["slash_amount"])),
|
||
strike_count=strike_count,
|
||
banned=current.banned or strike_count >= int(event.get("strike_threshold", 3)),
|
||
)
|
||
elif etype == "ban":
|
||
updated = _registry_wallet_with(current, banned=True)
|
||
elif etype == "completed_job":
|
||
updated = _registry_wallet_with(
|
||
current,
|
||
completed_job_count=current.completed_job_count + int(event.get("count", 1)),
|
||
last_active_ts=float(event.get("ts", time.time())),
|
||
)
|
||
elif etype == "set_reputation":
|
||
updated = _registry_wallet_with(current, reputation=float(event["reputation"]))
|
||
elif etype == "audit":
|
||
updated = _registry_wallet_with(current, last_audit_ts=float(event["ts"]))
|
||
elif etype == "audit_outcome":
|
||
# ADR-0018 §6: reputation moves only from tracker-verified audit
|
||
# outcomes. Strike/ban stay on the existing slash/strike events —
|
||
# this event never touches strike_count to avoid double-counting
|
||
# a single failed audit that already went through submit_slash_proof.
|
||
delta = REPUTATION_CLEAN_AUDIT_DELTA if event["passed"] else REPUTATION_FAILED_AUDIT_DELTA
|
||
reputation = min(REPUTATION_MAX, max(REPUTATION_MIN, current.reputation + delta))
|
||
updated = _registry_wallet_with(
|
||
current,
|
||
reputation=reputation,
|
||
last_audit_ts=float(event.get("ts", time.time())),
|
||
)
|
||
elif etype == "inactivity_decay":
|
||
reputation = max(
|
||
REPUTATION_MIN,
|
||
current.reputation - float(event.get("amount", REPUTATION_INACTIVITY_DECAY)),
|
||
)
|
||
updated = _registry_wallet_with(
|
||
current,
|
||
reputation=reputation,
|
||
# Resets the idle clock so decay applies at most once per window.
|
||
last_active_ts=float(event.get("ts", time.time())),
|
||
)
|
||
else:
|
||
return
|
||
self._state.registry[wallet_address] = updated
|
||
self._seen_event_ids.add(event["id"])
|
||
self._event_log.append(event)
|
||
self._dirty = True
|
||
|
||
def _init_db(self) -> None:
|
||
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
||
con.execute(
|
||
"CREATE TABLE IF NOT EXISTS registry_events "
|
||
"(event_id TEXT PRIMARY KEY, payload TEXT NOT NULL, ts REAL NOT NULL)"
|
||
)
|
||
con.commit()
|
||
con.close()
|
||
|
||
def _load_from_db(self) -> None:
|
||
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
||
rows = con.execute(
|
||
"SELECT payload FROM registry_events ORDER BY ts, event_id"
|
||
).fetchall()
|
||
con.close()
|
||
with self._lock:
|
||
for (payload,) in rows:
|
||
try:
|
||
event = json.loads(payload)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if event.get("id") not in self._seen_event_ids:
|
||
self._apply_locked(event)
|
||
self._dirty = False
|
||
|
||
def save_to_db(self) -> None:
|
||
if not self._db_path:
|
||
return
|
||
with self._lock:
|
||
if not self._dirty:
|
||
return
|
||
events = list(self._event_log)
|
||
self._dirty = False
|
||
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
||
con.executemany(
|
||
"INSERT OR IGNORE INTO registry_events (event_id, payload, ts) VALUES (?, ?, ?)",
|
||
[(e["id"], json.dumps(e), float(e.get("ts", 0.0))) for e in events],
|
||
)
|
||
con.commit()
|
||
con.close()
|
||
|
||
|
||
def _registry_wallet_with(wallet: RegistryWallet, **changes: object) -> RegistryWallet:
|
||
values = {
|
||
"stake_balance": wallet.stake_balance,
|
||
"strike_count": wallet.strike_count,
|
||
"banned": wallet.banned,
|
||
"completed_job_count": wallet.completed_job_count,
|
||
"reputation": wallet.reputation,
|
||
"last_audit_ts": wallet.last_audit_ts,
|
||
"last_active_ts": wallet.last_active_ts,
|
||
}
|
||
values.update(changes)
|
||
return RegistryWallet(**values)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ApiKeyBalance:
|
||
"""Client API key payment account balance."""
|
||
|
||
lamports: int = 0
|
||
usdc_micro: int = 0
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ComputeAttribution:
|
||
"""On-chain work attribution recorded by the gateway after inference."""
|
||
|
||
session_id: str
|
||
api_key: str
|
||
node_wallet: str
|
||
layer_start: int
|
||
layer_end: int
|
||
tokens: int
|
||
speed_score: float = 1.0
|
||
settled_epoch: int | None = None
|
||
|
||
@property
|
||
def layer_count(self) -> int:
|
||
return self.layer_end - self.layer_start + 1
|
||
|
||
@property
|
||
def work_units(self) -> int:
|
||
return self.layer_count * self.tokens
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ValidationEvent:
|
||
"""Completed inference data consumed by fraud validators."""
|
||
|
||
index: int
|
||
session_id: str
|
||
model: str
|
||
messages: list[dict]
|
||
observed_output: str
|
||
route_nodes: list[dict]
|
||
ts: float = field(default_factory=time.time)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SlashReceipt:
|
||
"""Local slash transaction receipt."""
|
||
|
||
signature: str
|
||
cluster: str
|
||
wallet_address: str
|
||
slash_amount: int
|
||
stake_before: int
|
||
stake_after: int
|
||
strike_count: int
|
||
banned: bool
|
||
reason: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SettlementResult:
|
||
"""Summary returned by an epoch settlement transaction."""
|
||
|
||
epoch: int
|
||
client_debits: dict[str, int]
|
||
node_rewards: dict[str, int]
|
||
validator_rewards: dict[str, int]
|
||
|
||
|
||
@dataclass
|
||
class _LocalContractState:
|
||
registry: dict[str, RegistryWallet] = field(default_factory=dict)
|
||
api_keys: dict[str, ApiKeyBalance] = field(default_factory=dict)
|
||
attributions: list[ComputeAttribution] = field(default_factory=list)
|
||
validation_events: list[ValidationEvent] = field(default_factory=list)
|
||
slash_receipts: list[SlashReceipt] = field(default_factory=list)
|
||
token_balances: dict[str, int] = field(default_factory=dict)
|
||
probationary_job_count: int = 50
|
||
|
||
|
||
class RegistryContract:
|
||
"""Registry wrapper for node stake, strikes, and bans."""
|
||
|
||
def __init__(self, state: _LocalContractState, cluster: str, event_log: RegistryEventLog) -> None:
|
||
self._state = state
|
||
self._cluster = cluster
|
||
self._event_log = event_log
|
||
|
||
def submit_stake(self, wallet_address: str, amount: int) -> dict:
|
||
if amount <= 0:
|
||
raise ValueError("stake amount must be positive")
|
||
self._event_log.record({
|
||
"id": f"stake-{uuid.uuid4().hex}",
|
||
"type": "stake",
|
||
"wallet": wallet_address,
|
||
"amount": amount,
|
||
"ts": time.time(),
|
||
})
|
||
return {"signature": f"local-stake-{wallet_address}", "cluster": self._cluster}
|
||
|
||
def get_wallet(self, wallet_address: str) -> RegistryWallet:
|
||
return self._state.registry.get(wallet_address, RegistryWallet())
|
||
|
||
def record_strike(self, wallet_address: str, ban_threshold: int = 3) -> dict:
|
||
self._event_log.record({
|
||
"id": f"strike-{uuid.uuid4().hex}",
|
||
"type": "strike",
|
||
"wallet": wallet_address,
|
||
"ban_threshold": ban_threshold,
|
||
"ts": time.time(),
|
||
})
|
||
return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster}
|
||
|
||
def submit_slash_proof(
|
||
self,
|
||
*,
|
||
wallet_address: str,
|
||
slash_amount: int,
|
||
strike_threshold: int = 3,
|
||
reason: str = "fraudulent inference output",
|
||
webhook_url: str | None = None,
|
||
) -> SlashReceipt:
|
||
if not wallet_address:
|
||
raise ValueError("wallet_address is required")
|
||
if slash_amount <= 0:
|
||
raise ValueError("slash_amount must be positive")
|
||
if strike_threshold <= 0:
|
||
raise ValueError("strike_threshold must be positive")
|
||
|
||
current = self.get_wallet(wallet_address)
|
||
self._event_log.record({
|
||
"id": f"slash-{uuid.uuid4().hex}",
|
||
"type": "slash",
|
||
"wallet": wallet_address,
|
||
"slash_amount": slash_amount,
|
||
"strike_threshold": strike_threshold,
|
||
"ts": time.time(),
|
||
})
|
||
updated = self.get_wallet(wallet_address)
|
||
receipt = SlashReceipt(
|
||
signature=f"local-slash-{wallet_address}-{len(self._state.slash_receipts) + 1}",
|
||
cluster=self._cluster,
|
||
wallet_address=wallet_address,
|
||
slash_amount=slash_amount,
|
||
stake_before=current.stake_balance,
|
||
stake_after=updated.stake_balance,
|
||
strike_count=updated.strike_count,
|
||
banned=updated.banned,
|
||
reason=reason,
|
||
)
|
||
self._state.slash_receipts.append(receipt)
|
||
_notify_slash(receipt, webhook_url)
|
||
return receipt
|
||
|
||
def ban_wallet(self, wallet_address: str) -> dict:
|
||
self._event_log.record({
|
||
"id": f"ban-{uuid.uuid4().hex}",
|
||
"type": "ban",
|
||
"wallet": wallet_address,
|
||
"ts": time.time(),
|
||
})
|
||
return {"signature": f"local-ban-{wallet_address}", "cluster": self._cluster}
|
||
|
||
def has_minimum_stake(self, wallet_address: str | None, minimum_stake: int) -> bool:
|
||
if not wallet_address:
|
||
return False
|
||
wallet = self.get_wallet(wallet_address)
|
||
return wallet.stake_balance >= minimum_stake
|
||
|
||
def list_wallets(self) -> dict[str, RegistryWallet]:
|
||
"""Snapshot of all known wallets (dashboard / monitoring)."""
|
||
return dict(self._state.registry)
|
||
|
||
def record_completed_job(self, wallet_address: str) -> RegistryWallet:
|
||
self.record_completed_jobs(wallet_address, 1)
|
||
return self.get_wallet(wallet_address)
|
||
|
||
def record_completed_jobs(self, wallet_address: str, count: int, *, ts: float | None = None) -> RegistryWallet:
|
||
if count <= 0:
|
||
raise ValueError("completed job count must be positive")
|
||
self._event_log.record({
|
||
"id": f"job-{uuid.uuid4().hex}",
|
||
"type": "completed_job",
|
||
"wallet": wallet_address,
|
||
"count": count,
|
||
"ts": time.time() if ts is None else ts,
|
||
})
|
||
return self.get_wallet(wallet_address)
|
||
|
||
def set_reputation(self, wallet_address: str, reputation: float) -> RegistryWallet:
|
||
self._event_log.record({
|
||
"id": f"reputation-{uuid.uuid4().hex}",
|
||
"type": "set_reputation",
|
||
"wallet": wallet_address,
|
||
"reputation": reputation,
|
||
"ts": time.time(),
|
||
})
|
||
return self.get_wallet(wallet_address)
|
||
|
||
def record_audit(self, wallet_address: str, ts: float | None = None) -> RegistryWallet:
|
||
audit_ts = time.time() if ts is None else ts
|
||
self._event_log.record({
|
||
"id": f"audit-{uuid.uuid4().hex}",
|
||
"type": "audit",
|
||
"wallet": wallet_address,
|
||
"ts": audit_ts,
|
||
})
|
||
return self.get_wallet(wallet_address)
|
||
|
||
def probationary_jobs_remaining(self, wallet_address: str) -> int:
|
||
wallet = self.get_wallet(wallet_address)
|
||
return max(0, self._state.probationary_job_count - wallet.completed_job_count)
|
||
|
||
def record_audit_outcome(self, wallet_address: str, *, passed: bool, ts: float | None = None) -> RegistryWallet:
|
||
"""ADR-0018 §6: the only reputation signal — clean audits build score
|
||
slowly, a failed audit loses it instantly. Strikes/bans are recorded
|
||
separately (`record_strike` / `submit_slash_proof`) so a single
|
||
failed audit is never double-counted as two strikes."""
|
||
self._event_log.record({
|
||
"id": f"audit-outcome-{uuid.uuid4().hex}",
|
||
"type": "audit_outcome",
|
||
"wallet": wallet_address,
|
||
"passed": passed,
|
||
"ts": time.time() if ts is None else ts,
|
||
})
|
||
return self.get_wallet(wallet_address)
|
||
|
||
def routing_multiplier(self, wallet_address: str) -> float:
|
||
"""ADR-0018 §6: ×0.8 routing/payout weight per strike, separate from
|
||
the forfeiture penalty (which stays full pending balance)."""
|
||
return self.get_wallet(wallet_address).routing_multiplier
|
||
|
||
def apply_inactivity_decay(
|
||
self,
|
||
*,
|
||
idle_seconds: float = REPUTATION_INACTIVITY_SECONDS,
|
||
decay_amount: float = REPUTATION_INACTIVITY_DECAY,
|
||
now: float | None = None,
|
||
) -> dict[str, RegistryWallet]:
|
||
"""ADR-0018 §6: reputation decays for wallets with no completed job
|
||
in `idle_seconds`. Decaying resets the idle clock, so a wallet decays
|
||
at most once per idle window rather than every call while still idle."""
|
||
current_ts = time.time() if now is None else now
|
||
decayed: dict[str, RegistryWallet] = {}
|
||
for wallet_address, wallet in list(self._state.registry.items()):
|
||
if wallet.banned or wallet.last_active_ts is None:
|
||
continue
|
||
if current_ts - wallet.last_active_ts < idle_seconds:
|
||
continue
|
||
self._event_log.record({
|
||
"id": f"inactivity-decay-{uuid.uuid4().hex}",
|
||
"type": "inactivity_decay",
|
||
"wallet": wallet_address,
|
||
"amount": decay_amount,
|
||
"ts": current_ts,
|
||
})
|
||
decayed[wallet_address] = self.get_wallet(wallet_address)
|
||
return decayed
|
||
|
||
|
||
class ValidationContract:
|
||
"""Validation event log consumed by the optimistic fraud detector."""
|
||
|
||
def __init__(self, state: _LocalContractState) -> None:
|
||
self._state = state
|
||
|
||
def record_completed_inference(
|
||
self,
|
||
*,
|
||
session_id: str,
|
||
model: str,
|
||
messages: list[dict],
|
||
observed_output: str,
|
||
route_nodes: list[dict],
|
||
ts: float | None = None,
|
||
) -> ValidationEvent:
|
||
if not session_id:
|
||
raise ValueError("session_id is required")
|
||
if not model:
|
||
raise ValueError("model is required")
|
||
if not isinstance(messages, list):
|
||
raise ValueError("messages must be a list")
|
||
if not isinstance(observed_output, str):
|
||
raise ValueError("observed_output must be a string")
|
||
if not isinstance(route_nodes, list):
|
||
raise ValueError("route_nodes must be a list")
|
||
event = ValidationEvent(
|
||
index=len(self._state.validation_events),
|
||
session_id=session_id,
|
||
model=model,
|
||
messages=[dict(message) for message in messages],
|
||
observed_output=observed_output,
|
||
route_nodes=[dict(node) for node in route_nodes],
|
||
**({"ts": ts} if ts is not None else {}),
|
||
)
|
||
self._state.validation_events.append(event)
|
||
return event
|
||
|
||
def list_completed_inferences(self, *, after_index: int = -1) -> list[ValidationEvent]:
|
||
return [
|
||
event for event in self._state.validation_events
|
||
if event.index > after_index
|
||
]
|
||
|
||
|
||
class PaymentContract:
|
||
"""Payment wrapper for funded API keys and compute attribution."""
|
||
|
||
def __init__(self, state: _LocalContractState, cluster: str, starting_credit_lamports: int) -> None:
|
||
self._state = state
|
||
self._cluster = cluster
|
||
self._starting_credit_lamports = starting_credit_lamports
|
||
|
||
def fund_api_key(
|
||
self,
|
||
api_key: str,
|
||
*,
|
||
lamports: int = 0,
|
||
usdc_micro: int = 0,
|
||
) -> dict:
|
||
if lamports < 0 or usdc_micro < 0:
|
||
raise ValueError("funding amounts must be non-negative")
|
||
if lamports == 0 and usdc_micro == 0:
|
||
raise ValueError("funding amount must be positive")
|
||
current = self.get_balance(api_key)
|
||
self._state.api_keys[api_key] = ApiKeyBalance(
|
||
lamports=current.lamports + lamports,
|
||
usdc_micro=current.usdc_micro + usdc_micro,
|
||
)
|
||
return {"signature": f"local-fund-{api_key}", "cluster": self._cluster}
|
||
|
||
def get_balance(self, api_key: str) -> ApiKeyBalance:
|
||
return self._state.api_keys.get(
|
||
api_key,
|
||
ApiKeyBalance(lamports=self._starting_credit_lamports),
|
||
)
|
||
|
||
def record_attribution(
|
||
self,
|
||
*,
|
||
session_id: str,
|
||
api_key: str,
|
||
node_wallet: str,
|
||
layer_start: int,
|
||
layer_end: int,
|
||
tokens: int,
|
||
speed_score: float = 1.0,
|
||
) -> dict:
|
||
if not api_key:
|
||
raise ValueError("api_key is required")
|
||
if not node_wallet:
|
||
raise ValueError("node_wallet is required")
|
||
if layer_start < 0 or layer_end < layer_start:
|
||
raise ValueError("layer range must be non-negative and ordered")
|
||
if tokens <= 0:
|
||
raise ValueError("tokens must be positive")
|
||
if speed_score <= 0:
|
||
raise ValueError("speed_score must be positive")
|
||
self._state.attributions.append(ComputeAttribution(
|
||
session_id=session_id,
|
||
api_key=api_key,
|
||
node_wallet=node_wallet,
|
||
layer_start=layer_start,
|
||
layer_end=layer_end,
|
||
tokens=tokens,
|
||
speed_score=speed_score,
|
||
))
|
||
return {"signature": f"local-attribution-{session_id}", "cluster": self._cluster}
|
||
|
||
def list_attributions(self, *, include_settled: bool = True) -> list[ComputeAttribution]:
|
||
if include_settled:
|
||
return list(self._state.attributions)
|
||
return [a for a in self._state.attributions if a.settled_epoch is None]
|
||
|
||
|
||
class SettlementContract:
|
||
"""Settlement wrapper that debits clients and credits token rewards."""
|
||
|
||
def __init__(
|
||
self,
|
||
state: _LocalContractState,
|
||
cluster: str,
|
||
cost_per_layer_token_lamport: int,
|
||
registry: RegistryContract,
|
||
) -> None:
|
||
self._state = state
|
||
self._cluster = cluster
|
||
self._cost_per_layer_token_lamport = cost_per_layer_token_lamport
|
||
self._registry = registry
|
||
|
||
def settle_epoch(
|
||
self,
|
||
*,
|
||
epoch: int,
|
||
validator_wallet: str | None = None,
|
||
validator_reward_share_bps: int = 0,
|
||
) -> SettlementResult:
|
||
if epoch < 0:
|
||
raise ValueError("epoch must be non-negative")
|
||
if not 0 <= validator_reward_share_bps <= 10_000:
|
||
raise ValueError("validator_reward_share_bps must be between 0 and 10000")
|
||
unsettled = [a for a in self._state.attributions if a.settled_epoch is None]
|
||
|
||
client_debits: dict[str, int] = {}
|
||
eligible_work_by_node: dict[str, int] = {}
|
||
node_rewards: dict[str, int] = {}
|
||
eligible_raw_work = 0
|
||
completed_jobs_to_record: dict[str, int] = {}
|
||
for attribution in unsettled:
|
||
debit = attribution.work_units * self._cost_per_layer_token_lamport
|
||
client_debits[attribution.api_key] = client_debits.get(attribution.api_key, 0) + debit
|
||
wallet = self._state.registry.get(attribution.node_wallet, RegistryWallet())
|
||
completed_before_job = wallet.completed_job_count + completed_jobs_to_record.get(attribution.node_wallet, 0)
|
||
if completed_before_job >= self._state.probationary_job_count:
|
||
eligible_raw_work += attribution.work_units
|
||
weighted_work = int(attribution.work_units * attribution.speed_score)
|
||
eligible_work_by_node[attribution.node_wallet] = (
|
||
eligible_work_by_node.get(attribution.node_wallet, 0) + weighted_work
|
||
)
|
||
node_rewards.setdefault(attribution.node_wallet, 0)
|
||
completed_jobs_to_record[attribution.node_wallet] = completed_jobs_to_record.get(attribution.node_wallet, 0) + 1
|
||
|
||
for api_key, debit in client_debits.items():
|
||
current = self._state.api_keys.get(api_key, ApiKeyBalance())
|
||
if current.lamports < debit:
|
||
raise ValueError(f"insufficient SOL balance for API key: {api_key}")
|
||
self._state.api_keys[api_key] = ApiKeyBalance(
|
||
lamports=current.lamports - debit,
|
||
usdc_micro=current.usdc_micro,
|
||
)
|
||
|
||
for wallet_address, count in completed_jobs_to_record.items():
|
||
self._record_completed_jobs(wallet_address, count)
|
||
|
||
total_weighted_work = sum(eligible_work_by_node.values())
|
||
raw_total_work = sum(a.work_units for a in unsettled)
|
||
validator_rewards: dict[str, int] = {}
|
||
validator_reward_total = (raw_total_work * validator_reward_share_bps) // 10_000
|
||
if validator_wallet and validator_reward_total:
|
||
validator_rewards[validator_wallet] = validator_reward_total
|
||
|
||
node_reward_pool = max(0, eligible_raw_work - validator_reward_total)
|
||
for node_wallet, weighted_work_units in eligible_work_by_node.items():
|
||
reward = 0 if total_weighted_work == 0 else (node_reward_pool * weighted_work_units) // total_weighted_work
|
||
node_rewards[node_wallet] = reward
|
||
self._state.token_balances[node_wallet] = (
|
||
self._state.token_balances.get(node_wallet, 0) + reward
|
||
)
|
||
|
||
for wallet, reward in validator_rewards.items():
|
||
self._state.token_balances[wallet] = self._state.token_balances.get(wallet, 0) + reward
|
||
|
||
self._state.attributions = [
|
||
ComputeAttribution(
|
||
session_id=a.session_id,
|
||
api_key=a.api_key,
|
||
node_wallet=a.node_wallet,
|
||
layer_start=a.layer_start,
|
||
layer_end=a.layer_end,
|
||
tokens=a.tokens,
|
||
speed_score=a.speed_score,
|
||
settled_epoch=epoch,
|
||
)
|
||
if a.settled_epoch is None else a
|
||
for a in self._state.attributions
|
||
]
|
||
return SettlementResult(
|
||
epoch=epoch,
|
||
client_debits=client_debits,
|
||
node_rewards=node_rewards,
|
||
validator_rewards=validator_rewards,
|
||
)
|
||
|
||
def get_token_balance(self, wallet_address: str) -> int:
|
||
return self._state.token_balances.get(wallet_address, 0)
|
||
|
||
def _record_completed_jobs(self, wallet_address: str, count: int) -> None:
|
||
self._registry.record_completed_jobs(wallet_address, count)
|
||
|
||
def deployment_plan(self) -> dict:
|
||
"""Return the configured manual testnet deployment targets."""
|
||
return {
|
||
"cluster": self._cluster,
|
||
"programs": ["registry", "payment", "settlement"],
|
||
}
|
||
|
||
|
||
class LocalSolanaContracts:
|
||
"""Facade that exposes all three contract wrappers over local validator state."""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
cluster: str = "local-test-validator",
|
||
cost_per_layer_token_lamport: int = 1,
|
||
starting_credit_lamports: int = 1_000,
|
||
probationary_job_count: int = 50,
|
||
registry_db: str | None = None,
|
||
) -> None:
|
||
if cost_per_layer_token_lamport <= 0:
|
||
raise ValueError("cost_per_layer_token_lamport must be positive")
|
||
if starting_credit_lamports < 0:
|
||
raise ValueError("starting_credit_lamports must be non-negative")
|
||
if probationary_job_count < 0:
|
||
raise ValueError("probationary_job_count must be non-negative")
|
||
self.cluster = cluster
|
||
self._state = _LocalContractState()
|
||
self._state.probationary_job_count = probationary_job_count
|
||
self.registry_log = RegistryEventLog(self._state, registry_db)
|
||
self.registry = RegistryContract(self._state, cluster, self.registry_log)
|
||
self.validation = ValidationContract(self._state)
|
||
self.payment = PaymentContract(self._state, cluster, starting_credit_lamports)
|
||
self.settlement = SettlementContract(
|
||
self._state,
|
||
cluster,
|
||
cost_per_layer_token_lamport,
|
||
self.registry,
|
||
)
|
||
|
||
def save_to_db(self) -> None:
|
||
self.registry_log.save_to_db()
|
||
|
||
|
||
def _notify_slash(receipt: SlashReceipt, webhook_url: str | None) -> None:
|
||
message = (
|
||
f"WARNING: node {receipt.wallet_address} slashed "
|
||
f"by {receipt.slash_amount} lamports; strikes={receipt.strike_count}; "
|
||
f"banned={receipt.banned}; reason={receipt.reason}"
|
||
)
|
||
print(message, flush=True)
|
||
if not webhook_url:
|
||
return
|
||
payload = json.dumps({
|
||
"event": "node_slashed",
|
||
"wallet_address": receipt.wallet_address,
|
||
"slash_amount": receipt.slash_amount,
|
||
"strike_count": receipt.strike_count,
|
||
"banned": receipt.banned,
|
||
"reason": receipt.reason,
|
||
}).encode()
|
||
req = urllib.request.Request(
|
||
webhook_url,
|
||
data=payload,
|
||
headers={"Content-Type": "application/json"},
|
||
method="POST",
|
||
)
|
||
try:
|
||
urllib.request.urlopen(req, timeout=2).read()
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
__all__ = [
|
||
"ApiKeyBalance",
|
||
"ComputeAttribution",
|
||
"LocalSolanaContracts",
|
||
"PaymentContract",
|
||
"REPUTATION_CLEAN_AUDIT_DELTA",
|
||
"REPUTATION_FAILED_AUDIT_DELTA",
|
||
"REPUTATION_INACTIVITY_DECAY",
|
||
"REPUTATION_INACTIVITY_SECONDS",
|
||
"REPUTATION_MAX",
|
||
"REPUTATION_MIN",
|
||
"ROUTING_STRIKE_MULTIPLIER",
|
||
"RegistryContract",
|
||
"RegistryEventLog",
|
||
"RegistryWallet",
|
||
"SettlementContract",
|
||
"SettlementResult",
|
||
"SlashReceipt",
|
||
"ValidationContract",
|
||
"ValidationEvent",
|
||
]
|