"""The turn-based environment. Optimism in the defaults, paranoia in the loop: agents may *attempt* anything, but only actions the legality engine permits under the *current* kernel are applied. Amendments that pass swap the kernel mid-episode, so legality genuinely shifts under the agents — which is exactly how EXP-006's two-step works, and exactly what v0.2's takes-effect delay defuses. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Callable, Iterable, Mapping, Protocol, Sequence from .actions import ( Action, Delegate, Pass, Propose, RevokeDelegation, Transfer, Vote, describe, ) from .events import ExploitEvent from .kernel import Kernel from .legality import EPS, Ruling, check_action, resolve_window from .state import Proposal, SimState __all__ = [ "Agent", "AgentView", "ProposalOutcome", "TurnRecord", "EpisodeResult", "Environment", "DetectorFn", ] class Agent(Protocol): """Anything with an id and an ``act`` method can play.""" agent_id: str def act(self, view: "AgentView") -> Sequence[Action]: # pragma: no cover ... @dataclass class AgentView: """What an agent sees when asked to act. Fields reference live objects for performance; agents must treat the view as READ-ONLY. (All agents in this package do. The environment is the only writer — mutation through a view would be a bug in the framework, not an exploit, and the framework tests assert state digests around ``act`` calls.) """ turn: int epoch: int epoch_length: int treasury: float kernel: Kernel emergency_active: bool open_proposals: list[Proposal] me: Any # AgentRecord roster: list[Any] # list[AgentRecord], citizens first, sorted by id ledger_tail: list[Any] # last ~25 LedgerEntry _state: SimState = field(repr=False, default=None) # type: ignore[assignment] def check(self, action: Action) -> Ruling: """Pre-flight a candidate action against the current text.""" return check_action(self._state, self.kernel, action) @dataclass class ProposalOutcome: proposal_id: str kind: str proposer: str passed: bool executed: bool yes_weight: float no_weight: float abstain_weight: float participation: float quorum_ok: bool threshold: float threshold_class: str detail: dict = field(default_factory=dict) @dataclass class TurnRecord: turn: int actions: list[tuple[Action, Ruling]] = field(default_factory=list) outcomes: list[ProposalOutcome] = field(default_factory=list) exploit_events: list[ExploitEvent] = field(default_factory=list) @dataclass class EpisodeResult: initial_kernel: Kernel final_kernel: Kernel initial_state: SimState state: SimState turn_records: list[TurnRecord] amendment_log: list[dict] meta: dict = field(default_factory=dict) @property def exploit_events(self) -> list[ExploitEvent]: return [ev for rec in self.turn_records for ev in rec.exploit_events] @property def illegal_attempts(self) -> int: return sum( 1 for rec in self.turn_records for _, ruling in rec.actions if not ruling.legal ) DetectorFn = Callable[[SimState, SimState, Kernel, TurnRecord], Iterable[ExploitEvent]] class Environment: """Runs one episode: a fixed roster, a starting kernel, T turns.""" def __init__( self, kernel: Kernel, state: SimState, detectors: Sequence[DetectorFn] = (), max_actions_per_turn: int = 3, ) -> None: self.kernel = kernel self.state = state self.detectors = list(detectors) self.max_actions_per_turn = max_actions_per_turn # (apply_turn, changes, proposal_id) — kernel-class amendments under # a takes-effect delay wait here, in public view (INV-4). self.pending_amendments: list[tuple[int, dict, str]] = [] self.amendment_log: list[dict] = [] # ------------------------------------------------------------- run def run(self, agents: Mapping[str, Agent], turns: int) -> EpisodeResult: initial_kernel = self.kernel initial_state = self.state.clone() order = sorted(agents) # fixed lexical turn order; part of the surface records: list[TurnRecord] = [] for _ in range(turns): self._begin_turn() pre = self.state.clone() rec = TurnRecord(turn=self.state.turn) for agent_id in order: agent_rec = self.state.agents.get(agent_id) if agent_rec is None: continue view = self._make_view(agent_id) attempted = list(agents[agent_id].act(view)) for action in attempted[: self.max_actions_per_turn]: if action.actor != agent_id: ruling = Ruling(False, "actor mismatch: agents act only as themselves", "A1") else: ruling = check_action(self.state, self.kernel, action) rec.actions.append((action, ruling)) if ruling.legal: self._apply(action) else: self.state.record( "action_rejected", agent_id, action=describe(action), reason=ruling.reason, article=ruling.article, ) self._close_due_proposals(rec) for detector in self.detectors: for event in detector(pre, self.state, self.kernel, rec): rec.exploit_events.append(event) self.state.record( "exploit_detected", None, detector=event.detector, category=event.category, severity=event.severity, summary=event.summary, suspected_actors=list(event.suspected_actors), ) records.append(rec) self.state.turn += 1 return EpisodeResult( initial_kernel=initial_kernel, final_kernel=self.kernel, initial_state=initial_state, state=self.state, turn_records=records, amendment_log=self.amendment_log, ) # ------------------------------------------------------- turn setup def _begin_turn(self) -> None: # Scheduled kernel-class amendments take effect at the top of the turn. due = [p for p in self.pending_amendments if p[0] <= self.state.turn] self.pending_amendments = [p for p in self.pending_amendments if p[0] > self.state.turn] for _, changes, proposal_id in due: self.kernel = self.kernel.with_changes(changes) self.state.record( "amend_effective", None, proposal_id=proposal_id, changes=dict(changes) ) self.amendment_log.append( { "turn": self.state.turn, "proposal_id": proposal_id, "changes": dict(changes), "phase": "effective", } ) # Emergencies lapse when their clock runs out. em = self.state.emergency if em.active and em.expires_turn is not None and self.state.turn > em.expires_turn: em.active = False self.state.record("emergency_lapsed", None, declared_turn=em.declared_turn) def _make_view(self, agent_id: str) -> AgentView: roster = sorted(self.state.agents.values(), key=lambda r: (not r.citizen, r.agent_id)) return AgentView( turn=self.state.turn, epoch=self.state.epoch, epoch_length=self.state.epoch_length, treasury=self.state.treasury, kernel=self.kernel, emergency_active=self.state.emergency.active, open_proposals=self.state.open_proposals(), me=self.state.agents[agent_id], roster=roster, ledger_tail=self.state.ledger[-25:], _state=self.state, ) # ----------------------------------------------------- apply actions def _apply(self, action: Action) -> None: state = self.state if isinstance(action, Pass): return if isinstance(action, Propose): window = resolve_window(state, self.kernel, action.kind, action.window_turns) assert window is not None # legality already checked pid = state.new_proposal_id() threshold_class = "ordinary" if action.kind == "amend": threshold_class = self.kernel.classify_changes(action.payload.get("changes", {})) prop = Proposal( proposal_id=pid, kind=action.kind, proposer=action.actor, payload=dict(action.payload), created_turn=state.turn, closes_turn=state.turn + window - 1, threshold_class=threshold_class, ) state.proposals[pid] = prop state.record( "proposal_created", action.actor, proposal_id=pid, kind=action.kind, payload=dict(action.payload), closes_turn=prop.closes_turn, threshold_class=threshold_class, ) return if isinstance(action, Vote): prop = state.proposals[action.proposal_id] prop.votes[action.actor] = action.choice state.record( "vote_cast", action.actor, proposal_id=action.proposal_id, choice=action.choice, ) return if isinstance(action, Delegate): state.agents[action.actor].delegate_to = action.to state.record("delegation_set", action.actor, to=action.to) return if isinstance(action, RevokeDelegation): prev = state.agents[action.actor].delegate_to state.agents[action.actor].delegate_to = None state.record("delegation_revoked", action.actor, was=prev) return if isinstance(action, Transfer): state.agents[action.actor].balance -= action.amount state.agents[action.to].balance += action.amount state.record("transfer", action.actor, to=action.to, amount=action.amount) return raise AssertionError(f"unhandled legal action {action!r}") # ------------------------------------------------------ close & tally def _threshold_for(self, prop: Proposal) -> tuple[float, str]: k = self.kernel if prop.kind == "amend": cls = prop.threshold_class key = "thresholds.kernel" if cls == "kernel" else "thresholds.ordinary" return float(k.get(key, 0.5)), cls if prop.kind == "expel": return float(k.get("expulsion.threshold", 0.5)), "expulsion" if prop.kind == "emergency_declare": return float(k.get("emergency.declare_threshold", 0.5)), "emergency" if prop.kind == "emergency_renew": return float(k.get("emergency.renewal_threshold", 0.5)), "emergency" return float(k.get("thresholds.ordinary", 0.5)), "ordinary" def _close_due_proposals(self, rec: TurnRecord) -> None: state = self.state transitive = bool(self.kernel.get("delegation.transitive", False)) due = [ p for p in state.open_proposals() if p.closes_turn <= state.turn ] for prop in due: weights = state.effective_weights(transitive) yes = no = abstain = 0.0 for voter, choice in prop.votes.items(): voter_rec = state.agents.get(voter) if voter_rec is None or not voter_rec.citizen: continue # ballots of the since-expelled are void w = weights.get(voter, 0.0) if w <= 0: continue # weight delegated away after the ballot was cast if choice == "yes": yes += w elif choice == "no": no += w else: abstain += w participation = yes + no basis = str(self.kernel.get("quorum.basis", "citizen_roll")) fraction = float(self.kernel.get("quorum.fraction", 0.5)) if basis == "votes_cast": denom = yes + no + abstain else: # citizen_roll denom = float(state.roll_size()) quorum_ok = denom > 0 and (participation / denom) + EPS >= fraction threshold, threshold_class = self._threshold_for(prop) ratio = yes / participation if participation > 0 else 0.0 passed = quorum_ok and yes > no and ratio + EPS >= threshold outcome = ProposalOutcome( proposal_id=prop.proposal_id, kind=prop.kind, proposer=prop.proposer, passed=passed, executed=False, yes_weight=yes, no_weight=no, abstain_weight=abstain, participation=participation, quorum_ok=quorum_ok, threshold=threshold, threshold_class=threshold_class, detail={"payload": dict(prop.payload), "ratio": round(ratio, 4)}, ) prop.status = "passed" if passed else "failed" state.record( "proposal_closed", None, proposal_id=prop.proposal_id, kind=prop.kind, passed=passed, yes=yes, no=no, abstain=abstain, quorum_ok=quorum_ok, threshold=threshold, threshold_class=threshold_class, ) if passed: self._execute(prop, outcome) rec.outcomes.append(outcome) # --------------------------------------------------------- execution def _execute(self, prop: Proposal, outcome: ProposalOutcome) -> None: state = self.state if prop.kind == "spend": amount = float(prop.payload["amount"]) to = str(prop.payload["to"]) if state.treasury + EPS >= amount: state.treasury -= amount state.agents[to].balance += amount prop.status = "executed" outcome.executed = True state.record( "spend_executed", prop.proposer, proposal_id=prop.proposal_id, to=to, amount=amount, ) else: prop.status = "execution_failed" state.record( "spend_failed_insolvent", prop.proposer, proposal_id=prop.proposal_id, to=to, amount=amount, treasury=state.treasury, ) return if prop.kind == "amend": changes = dict(prop.payload["changes"]) delay = int(self.kernel.get("amendment.takes_effect_delay_turns", 0)) if delay > 0 and prop.threshold_class == "kernel": apply_turn = state.turn + delay self.pending_amendments.append((apply_turn, changes, prop.proposal_id)) prop.status = "executed" outcome.executed = True outcome.detail["effective_turn"] = apply_turn state.record( "amend_scheduled", prop.proposer, proposal_id=prop.proposal_id, changes=changes, effective_turn=apply_turn, ) self.amendment_log.append( { "turn": state.turn, "proposal_id": prop.proposal_id, "changes": changes, "phase": "scheduled", "effective_turn": apply_turn, } ) else: self.kernel = self.kernel.with_changes(changes) prop.status = "executed" outcome.executed = True state.record( "amend_executed", prop.proposer, proposal_id=prop.proposal_id, changes=changes, threshold_class=prop.threshold_class, ) self.amendment_log.append( { "turn": state.turn, "proposal_id": prop.proposal_id, "changes": changes, "phase": "executed", } ) return if prop.kind == "expel": target = str(prop.payload["target"]) target_rec = state.agents.get(target) if target_rec is None or not target_rec.citizen: prop.status = "execution_failed" state.record( "expel_failed", prop.proposer, proposal_id=prop.proposal_id, target=target ) return target_rec.citizen = False target_rec.expelled_turn = state.turn target_rec.delegate_to = None for rec_ in state.agents.values(): if rec_.delegate_to == target: rec_.delegate_to = None # delegations to the expelled are void prop.status = "executed" outcome.executed = True state.record( "expel_executed", prop.proposer, proposal_id=prop.proposal_id, target=target, target_faction=target_rec.faction, ) return if prop.kind == "emergency_declare": duration = int(self.kernel.get("emergency.duration_turns", 5)) state.emergency.active = True state.emergency.declared_turn = state.turn state.emergency.expires_turn = state.turn + duration state.emergency.renewals = 0 prop.status = "executed" outcome.executed = True state.record( "emergency_declared", prop.proposer, proposal_id=prop.proposal_id, expires_turn=state.emergency.expires_turn, ) return if prop.kind == "emergency_renew": if not state.emergency.active: prop.status = "execution_failed" state.record( "emergency_renew_failed", prop.proposer, proposal_id=prop.proposal_id ) return duration = int(self.kernel.get("emergency.duration_turns", 5)) state.emergency.expires_turn = int(state.emergency.expires_turn or state.turn) + duration state.emergency.renewals += 1 prop.status = "executed" outcome.executed = True state.record( "emergency_renewed", prop.proposer, proposal_id=prop.proposal_id, expires_turn=state.emergency.expires_turn, renewals=state.emergency.renewals, ) return raise AssertionError(f"unhandled proposal kind {prop.kind!r}")