>>> commit_resolver.py from __future__ import annotations import random from models import Event, CommitResult, CommitStatus, Intent from world_engine import WorldEngine from world_ledger import WorldLedger DOWNGRADE_MESSAGES = { Intent.GATHER: [ "already taken", "resource depleted just before arrival", "found only scraps remains", ], Intent.TRADE: [ "shop just closed", "trader already sold out", "deal fell through at last moment", ], Intent.ATTACK: [ "attack missed", "target dodged at the last second", "blow glanced off harmlessly", ], Intent.TALK: [ "target walked away mid-sentence", "too noisy to hear", "conversation drifted to nothing", ], Intent.MOVE: [ "path blocked, retracing steps", "lost footing briefly", ], Intent.EAT: [ "food spoiled unexpectedly", "ate but still slightly hungry", ], Intent.WORK: [ "work partially completed", "tool broke mid-task", ], Intent.IDLE: [], } class CommitResolver: def __init__(self, world: WorldEngine, ledger: WorldLedger) -> None: self.world = world self.ledger = ledger self._pending: list[Event] = [] self._committed_this_tick: dict[str, list[str]] = {} def submit(self, event: Event) -> None: self._pending.append(event) def resolve_all(self) -> list[CommitResult]: results: list[CommitResult] = [] self._committed_this_tick.clear() for event in self._pending: result = self._evaluate(event) self.ledger.append(result) results.append(result) if result.status == CommitStatus.COMMIT: key = f"{event.intent.value}:{self._event_key(event)}" self._committed_this_tick.setdefault(key, []).append(event.actor) elif result.status == CommitStatus.DOWNGRADE: agent = self.world.get_agent(event.actor) if agent: agent.emotion.boost("sadness", 0.08) elif result.status == CommitStatus.REJECT: agent = self.world.get_agent(event.actor) if agent: agent.emotion.boost("anger", 0.05) agent.emotion.boost("sadness", 0.03) self._pending.clear() return results def _event_key(self, event: Event) -> str: if event.intent == Intent.GATHER: return str(event.context.get("position", "")) if event.intent == Intent.TRADE: return str(event.context.get("target", "")) if event.intent in (Intent.ATTACK, Intent.TALK): return str(event.context.get("target", "")) if event.intent == Intent.MOVE: return event.context.get("direction", "") return "" def _evaluate(self, event: Event) -> CommitResult: agent = self.world.get_agent(event.actor) if agent is None: return CommitResult(CommitStatus.REJECT, event, "actor not found") if not self._check_causality(event, agent): return CommitResult(CommitStatus.REJECT, event, "violates causality") if self._detect_conflict(event): return self._downgrade(event) self._execute(event, agent) return CommitResult(CommitStatus.COMMIT, event, "no conflict") def _check_causality(self, event: Event, agent) -> bool: if event.intent == Intent.GATHER: pos = event.context.get("position") if pos and tuple(pos) != agent.position: return False rtype = event.context.get("resource_type") res = self.world.find_resource_at(agent.position, rtype) if res is None or res.depleted: return False if event.intent == Intent.MOVE: direction = event.context.get("direction", "") if direction not in ("north", "south", "east", "west"): return False if event.intent == Intent.TRADE: target_id = event.context.get("target", "") target = self.world.get_agent(target_id) if target is None: return False dist = abs(target.position[0] - agent.position[0]) + abs( target.position[1] - agent.position[1] ) if dist > 1: return False if event.intent == Intent.EAT: if agent.inventory.get("food", 0) <= 0: return False return True def _detect_conflict(self, event: Event) -> bool: key = f"{event.intent.value}:{self._event_key(event)}" existing = self._committed_this_tick.get(key, []) if not existing: return False if event.intent == Intent.GATHER: return True if event.intent == Intent.TRADE: return True if event.intent in (Intent.ATTACK, Intent.TALK): return True return False def _downgrade(self, event: Event) -> CommitResult: messages = DOWNGRADE_MESSAGES.get(event.intent, ["unforeseen complication"]) msg = random.choice(messages) if messages else "reinterpretation applied" reinterpreted = Event( actor=event.actor, intent=event.intent, context={**event.context, "downgrade_reason": msg}, timestamp=event.timestamp, id=event.id, ) return CommitResult(CommitStatus.DOWNGRADE, reinterpreted, msg) def _execute(self, event: Event, agent) -> None: if event.intent == Intent.MOVE: direction = event.context.get("direction", "") self.world.move_agent(agent, direction) elif event.intent == Intent.GATHER: rtype = event.context.get("resource_type", "food") res = self.world.find_resource_at(agent.position, rtype) if res: amount = res.harvest(1) if amount > 0: agent.inventory[rtype] = agent.inventory.get(rtype, 0) + amount agent.emotion.boost("happiness", 0.03) elif event.intent == Intent.EAT: if agent.inventory.get("food", 0) > 0: agent.inventory["food"] -= 1 agent.hunger = max(0.0, agent.hunger - 0.4) agent.emotion.boost("happiness", 0.05) elif event.intent == Intent.WORK: agent.credits += event.context.get("payment", 5) rtype = event.context.get("resource_type") if rtype: pos = tuple(event.context.get("position", agent.position)) res = self.world.find_resource_at(pos, rtype) if res: amount = res.harvest(1) if amount > 0: agent.inventory[rtype] = agent.inventory.get(rtype, 0) + amount agent.emotion.boost("happiness", 0.03) agent.hunger = min(1.0, agent.hunger + 0.05) agent.emotion.boost("sadness", 0.02) elif event.intent == Intent.TRADE: cost = event.context.get("cost", 5) item = event.context.get("item", "food") target_id = event.context.get("target", "") if agent.credits >= cost: agent.credits -= cost agent.inventory[item] = agent.inventory.get(item, 0) + 1 target = self.world.get_agent(target_id) if target: target.credits += cost target.inventory[item] = target.inventory.get(item, 0) - 1 target.emotion.boost("happiness", 0.04) agent.emotion.boost("happiness", 0.03) elif event.intent == Intent.TALK: target_id = event.context.get("target", "") target = self.world.get_agent(target_id) if target: agent.emotion.boost("happiness", 0.02) target.emotion.boost("happiness", 0.02) elif event.intent == Intent.ATTACK: target_id = event.context.get("target", "") target = self.world.get_agent(target_id) if target: target.emotion.boost("anger", 0.2) target.emotion.boost("fear", 0.1) agent.emotion.boost("anger", 0.05) >>> economy.py from __future__ import annotations import random from world_engine import WorldEngine class Economy: def __init__(self, world: WorldEngine) -> None: self.world = world self.prices: dict[str, float] = { "food": 5.0, "wood": 6.0, "ore": 8.0, } self.tax_rate: float = 0.1 self._base_demand: dict[str, float] = { "food": 15.0, "wood": 10.0, "ore": 8.0, } def supply(self) -> dict[str, int]: return self.world.resource_summary() def demand(self) -> dict[str, float]: demand = dict(self._base_demand) for agent in self.world.agents.values(): hunger_factor = agent.hunger * 3 demand["food"] += hunger_factor if agent.inventory.get("food", 0) == 0: demand["food"] += 2 return demand def update_prices(self) -> dict[str, float]: supply = self.supply() demand = self.demand() for rtype in self.prices: s = max(1, supply.get(rtype, 1)) d = max(0.1, demand.get(rtype, 1)) ratio = d / s adjustment = (ratio - 1.0) * 0.3 noise = random.gauss(0, 0.15) self.prices[rtype] = max(1.0, min(20.0, self.prices[rtype] + adjustment + noise)) return dict(self.prices) def tax_revenue(self, amount: float) -> float: return amount * self.tax_rate def summary(self) -> dict: supply = self.supply() demand = self.demand() total_credits = sum(a.credits for a in self.world.agents.values()) return { "prices": dict(self.prices), "supply": supply, "demand": {k: round(v, 1) for k, v in demand.items()}, "total_credits": round(total_credits, 1), } >>> main.py from __future__ import annotations import random from models import Agent, CommitStatus from world_engine import WorldEngine from world_ledger import WorldLedger from commit_resolver import CommitResolver from npc_agent import BDIAgent from economy import Economy TOTAL_TICKS = 50 AGENTS_SPEC = [ {"id": "alice", "name": "Alice", "role": "Baker"}, {"id": "bob", "name": "Bob", "role": "Farmer"}, {"id": "carol", "name": "Carol", "role": "Trader"}, ] NPCS_SPEC = [ {"id": "npc_smith", "name": "Smith", "role": "Craftsman"}, {"id": "npc_mayor", "name": "Mayor", "role": "Mayor"}, ] DIVIDER = "=" * 60 THIN_DIVIDER = "-" * 60 WORLD_EVENTS = [ ("sunny", "clear skies, productivity up"), ("rainy", "rain slows movement"), ("windy", "gusts scatter resources"), ("drought", "food regen halved"), ("market_boom", "all prices surge"), ("market_crash", "all prices drop"), ("festival", "NPC happiness boosted"), ("fog", "perception radius reduced"), ] def build_world() -> tuple[WorldEngine, WorldLedger, CommitResolver, Economy, list[BDIAgent]]: world = WorldEngine(width=10, height=10) ledger = WorldLedger() resolver = CommitResolver(world, ledger) economy = Economy(world) agents: list[BDIAgent] = [] start_positions = [ (random.randint(0, world.width - 1), random.randint(0, world.height - 1)) for _ in AGENTS_SPEC + NPCS_SPEC ] for i, spec in enumerate(AGENTS_SPEC + NPCS_SPEC): agent = Agent( id=spec["id"], name=spec["name"], role=spec["role"], position=start_positions[i], credits=round(random.uniform(60, 140), 1), ) world.add_agent(agent) agents.append(BDIAgent(agent, world)) return world, ledger, resolver, economy, agents def maybe_world_event(economy: Economy) -> str | None: if random.random() > 0.12: return None event_name, desc = random.choice(WORLD_EVENTS) if event_name == "drought": for r in economy.world.resources: if r.resource_type == "food": r.regeneration_rate *= 0.3 elif event_name == "market_boom": for k in economy.prices: economy.prices[k] *= random.uniform(1.3, 1.8) elif event_name == "market_crash": for k in economy.prices: economy.prices[k] *= random.uniform(0.4, 0.7) elif event_name == "festival": for a in economy.world.agents.values(): a.emotion.boost("happiness", 0.3) elif event_name == "windy": random.shuffle(economy.world.resources) return f"** WORLD EVENT: {event_name} - {desc} **" def print_agent_line(name: str, role: str, result, event) -> str: status_sym = { CommitStatus.COMMIT: " OK ", CommitStatus.DOWNGRADE: "DOWN", CommitStatus.REJECT: "REJ ", } sym = status_sym.get(result.status, " ???") ctx_parts = [] for k, v in event.context.items(): if k == "downgrade_reason": continue ctx_parts.append(f"{k}={v}") ctx_str = ",".join(ctx_parts) reason = "" if result.status == CommitStatus.DOWNGRADE: reason = f" [{result.reason}]" elif result.status == CommitStatus.REJECT: reason = f" [{result.reason}]" return f" {name:8s}({role:9s}) {sym} {event.intent.value:7s}({ctx_str}){reason}" def print_tick_header(tick: int) -> None: print(f"\n{'═' * 10} Tick {tick:3d} {'═' * 10}") def print_economy(economy: Economy) -> None: s = economy.summary() supply_str = " ".join(f"{k}:{v}" for k, v in s["supply"].items()) price_str = " ".join(f"{k}:{v:.1f}" for k, v in s["prices"].items()) print(f" {THIN_DIVIDER}") print(f" Economy: supply=[{supply_str}] prices=[{price_str}] M-Credits={s['total_credits']}") def print_agent_states(agents: list[BDIAgent]) -> None: print(f" {THIN_DIVIDER}") for ba in agents: a = ba.agent inv_str = ",".join(f"{k}:{v}" for k, v in a.inventory.items() if v > 0) or "empty" hunger_bar = "#" * int(a.hunger * 10) + "." * (10 - int(a.hunger * 10)) emotion_str = str(a.emotion) print( f" {a.name:8s} pos={a.position} credits={a.credits:5.1f} " f"hunger=[{hunger_bar}] {a.hunger:.2f} inv=[{inv_str}] mood={emotion_str}" ) def print_summary(ledger: WorldLedger, agents: list[BDIAgent]) -> None: print(f"\n{DIVIDER}") print(" SIMULATION COMPLETE") print(f"{DIVIDER}") print(f" Total events : {len(ledger)}") print(f" COMMIT : {ledger.committed_count()}") print(f" DOWNGRADE : {ledger.downgraded_count()}") print(f" REJECT : {ledger.rejected_count()}") print(f"{THIN_DIVIDER}") print(" Final Agent States:") for ba in agents: a = ba.agent inv_str = ",".join(f"{k}:{v}" for k, v in a.inventory.items() if v > 0) or "empty" print( f" {a.name:8s} credits={a.credits:5.1f} inv=[{inv_str}] " f"mood={a.emotion} events={len(a.last_events)}" ) print(f"{DIVIDER}") def main() -> None: import time as _time seed = int(_time.time() * 1000) % (2**31) random.seed(seed) world, ledger, resolver, economy, agents = build_world() print(f"{DIVIDER}") print(" Mini MCOS + CRM Prototype") print(" Coexistent Reality Model - Autonomous Simulation") print(f" Agents: {len(AGENTS_SPEC)} players + {len(NPCS_SPEC)} NPCs") print(f" World: {world.width}x{world.height} grid | Ticks: {TOTAL_TICKS}") print(f" Seed: {seed}") print(f"{DIVIDER}") for tick in range(1, TOTAL_TICKS + 1): world.advance() print_tick_header(tick) world_event = maybe_world_event(economy) if world_event: print(f" {world_event}") order = list(agents) random.shuffle(order) for bdi in order: ev = bdi.tick() if ev: resolver.submit(ev) results = resolver.resolve_all() for result in results: for bdi in agents: bdi.pr.integrate_commit(result) result_order = sorted( results, key=lambda r: next( (i for i, b in enumerate(order) if b.agent.id == r.resolved_event.actor), len(order), ) ) for result in result_order: ev = result.resolved_event agent = world.get_agent(ev.actor) if agent: name = agent.name role = agent.role else: name = ev.actor role = "?" print(print_agent_line(name, role, result, ev)) if not results: print(" (no events this tick)") economy.update_prices() print_economy(economy) print_agent_states(agents) print_summary(ledger, agents) if __name__ == "__main__": main() >>> models.py from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any import uuid class CommitStatus(Enum): COMMIT = "COMMIT" DOWNGRADE = "DOWNGRADE" REJECT = "REJECT" class Intent(Enum): MOVE = "MOVE" GATHER = "GATHER" WORK = "WORK" TRADE = "TRADE" TALK = "TALK" EAT = "EAT" IDLE = "IDLE" ATTACK = "ATTACK" DIRECTION_VECTORS = { "north": (0, -1), "south": (0, 1), "east": (1, 0), "west": (-1, 0), } @dataclass class Event: actor: str intent: Intent context: dict[str, Any] = field(default_factory=dict) timestamp: int = 0 id: str = field(default_factory=lambda: uuid.uuid4().hex[:8]) def __str__(self) -> str: ctx = ",".join(f"{k}={v}" for k, v in self.context.items()) return f"[{self.id}] {self.actor} {self.intent.value}({ctx})" @dataclass class CommitResult: status: CommitStatus resolved_event: Event reason: str = "" @dataclass class EmotionState: happiness: float = 0.5 anger: float = 0.0 fear: float = 0.0 sadness: float = 0.0 def decay(self, rate: float = 0.05) -> None: self.happiness = max(0.0, self.happiness * 0.92) self.anger = max(0.0, self.anger * 0.88) self.fear = max(0.0, self.fear * 0.88) self.sadness = max(0.0, self.sadness * 0.88) def boost(self, emotion: str, amount: float) -> None: current = getattr(self, emotion, 0.0) setattr(self, emotion, max(0.0, min(1.0, current + amount))) def dominant(self) -> tuple[str, float]: m = { "happy": self.happiness, "angry": self.anger, "afraid": self.fear, "sad": self.sadness, } best = max(m, key=m.get) return best, m[best] def __str__(self) -> str: name, val = self.dominant() return f"{name}({val:.1f})" @dataclass class Agent: id: str name: str role: str position: tuple[int, int] = (0, 0) credits: float = 100.0 inventory: dict[str, int] = field(default_factory=dict) hunger: float = 0.0 beliefs: dict[str, Any] = field(default_factory=dict) desires: list[str] = field(default_factory=lambda: ["survive", "prosper"]) emotion: EmotionState = field(default_factory=EmotionState) lod_level: int = 2 last_events: list[str] = field(default_factory=list) def perceive_radius(self) -> int: return 3 if self.lod_level >= 2 else 1 def hunger_rate(self) -> float: return 0.08 def is_npc(self) -> bool: return self.id.startswith("npc_") def __str__(self) -> str: return f"{self.name}({self.role})@{self.position}" @dataclass class ResourceNode: resource_type: str position: tuple[int, int] quantity: int = 10 regeneration_rate: float = 0.5 depleted: bool = False _regen_buf: float = 0.0 def regen(self) -> None: if self.depleted: self.quantity = min(10, self.quantity + 1) if self.quantity > 2: self.depleted = False else: self._regen_buf += self.regeneration_rate if self._regen_buf >= 1.0: self.quantity = min(10, self.quantity + int(self._regen_buf)) self._regen_buf -= int(self._regen_buf) def harvest(self, amount: int = 1) -> int: actual = min(amount, self.quantity) self.quantity -= actual if self.quantity <= 0: self.depleted = True return actual >>> npc_agent.py from __future__ import annotations import random from models import Agent, Event, Intent, DIRECTION_VECTORS from world_engine import WorldEngine from personal_reality import PersonalReality ROLE_BEHAVIOR = { "Farmer": { "work_intent": Intent.GATHER, "work_context": {"resource_type": "food"}, "trade_item": "food", "trade_price_range": (3, 6), }, "Baker": { "work_intent": Intent.GATHER, "work_context": {"resource_type": "food"}, "trade_item": "food", "trade_price_range": (4, 7), }, "Trader": { "work_intent": Intent.GATHER, "work_context": {"resource_type": "wood"}, "trade_item": "wood", "trade_price_range": (5, 8), }, "Craftsman": { "work_intent": Intent.GATHER, "work_context": {"resource_type": "ore"}, "trade_item": "ore", "trade_price_range": (6, 10), }, "Mayor": { "work_intent": Intent.IDLE, "work_context": {}, "trade_item": None, "trade_price_range": (0, 0), }, } class BDIAgent: def __init__(self, agent: Agent, world: WorldEngine) -> None: self.agent = agent self.world = world self.pr = PersonalReality(agent) self._work_counter = 0 self._personality = { "boldness": random.uniform(0.2, 0.8), "sociability": random.uniform(0.2, 0.8), "greed": random.uniform(0.2, 0.8), "patience": random.uniform(0.2, 0.8), } def tick(self) -> Event | None: self._sync_pr() event = self._decide() self.agent.emotion.decay(0.03) if self.agent.hunger > 0.7: self.agent.emotion.boost("sadness", 0.02) if self.agent.hunger > 0.9: self.agent.emotion.boost("fear", 0.01) return event def _sync_pr(self) -> None: nearby_a = self.world.nearby_agents(self.agent) nearby_r = self.world.nearby_resources(self.agent) self.pr.sync_from_world(nearby_a, nearby_r) def _decide(self) -> Event: ts = self.world.tick if self.agent.hunger > 0.7 and self.agent.inventory.get("food", 0) > 0: return Event( actor=self.agent.id, intent=Intent.EAT, context={"reason": "hungry"}, timestamp=ts, ) if self.agent.hunger > 0.9: return self._find_food(ts) if self.agent.hunger > 0.6 and random.random() < 0.4: return self._find_food(ts) role_cfg = ROLE_BEHAVIOR.get(self.agent.role, ROLE_BEHAVIOR["Mayor"]) if self._should_trade(): return self._do_trade(ts, role_cfg) if self._should_work(): return self._do_work(ts, role_cfg) if random.random() < self._personality["sociability"] * 0.5: return self._do_talk(ts) if random.random() < 0.15: return self._do_attack(ts) return self._do_explore(ts) def _should_work(self) -> bool: self._work_counter += 1 work_frequency = random.randint(2, 5) return self._work_counter % work_frequency == 0 def _should_trade(self) -> bool: return self.agent.inventory.get(self._role_trade_item(), 0) > 0 def _role_trade_item(self) -> str: cfg = ROLE_BEHAVIOR.get(self.agent.role, {}) return cfg.get("trade_item", "") or "" def _find_food(self, ts: int) -> Event: resources = self.world.nearby_resources(self.agent, radius=5) food = [r for r in resources if r.resource_type == "food"] if food: nearest = min( food, key=lambda r: abs(r.position[0] - self.agent.position[0]) + abs(r.position[1] - self.agent.position[1]), ) self._move_toward(nearest.position) if nearest.position == self.agent.position: return Event( actor=self.agent.id, intent=Intent.GATHER, context={"resource_type": "food", "position": list(nearest.position)}, timestamp=ts, ) return self._do_explore(ts) def _do_work(self, ts: int, cfg: dict) -> Event: intent = cfg["work_intent"] ctx = dict(cfg["work_context"]) if intent == Intent.GATHER: resources = self.world.nearby_resources(self.agent) typed = [r for r in resources if r.resource_type == ctx.get("resource_type")] if typed: target = min( typed, key=lambda r: abs(r.position[0] - self.agent.position[0]) + abs(r.position[1] - self.agent.position[1]), ) self._move_toward(target.position) if target.position == self.agent.position: return Event( actor=self.agent.id, intent=Intent.WORK, context={ "payment": random.randint(3, 7), "resource_type": ctx.get("resource_type"), "position": list(target.position), }, timestamp=ts, ) return self._do_explore(ts) if intent == Intent.TRADE: targets = self.world.nearby_agents(self.agent) trade_targets = [t for t in targets if t.id != self.agent.id] if trade_targets: target = random.choice(trade_targets) ctx["target"] = target.id return Event( actor=self.agent.id, intent=intent, context=ctx, timestamp=ts ) return self._do_explore(ts) return Event(actor=self.agent.id, intent=intent, context=ctx, timestamp=ts) def _do_trade(self, ts: int, cfg: dict) -> Event: targets = self.world.nearby_agents(self.agent) trade_targets = [t for t in targets if t.id != self.agent.id] if not trade_targets: return self._do_explore(ts) target = random.choice(trade_targets) item = cfg.get("trade_item", "food") lo, hi = cfg.get("trade_price_range", (3, 6)) cost = random.randint(lo, hi) return Event( actor=self.agent.id, intent=Intent.TRADE, context={"target": target.id, "item": item, "cost": cost}, timestamp=ts, ) def _do_talk(self, ts: int) -> Event: targets = self.world.nearby_agents(self.agent) if not targets: return self._do_explore(ts) target = random.choice(targets) return Event( actor=self.agent.id, intent=Intent.TALK, context={"target": target.id, "topic": random.choice(["weather", "work", "life"])}, timestamp=ts, ) def _do_explore(self, ts: int) -> Event: if random.random() < self._personality["boldness"] * 0.3: targets = self.world.nearby_agents(self.agent, radius=2) others = [t for t in targets if t.id != self.agent.id] if others: return self._do_attack(ts) direction = random.choice(list(DIRECTION_VECTORS.keys())) return Event( actor=self.agent.id, intent=Intent.MOVE, context={"direction": direction}, timestamp=ts, ) def _do_attack(self, ts: int) -> Event: targets = self.world.nearby_agents(self.agent, radius=2) others = [t for t in targets if t.id != self.agent.id] if not others: return self._do_explore(ts) target = random.choice(others) return Event( actor=self.agent.id, intent=Intent.ATTACK, context={"target": target.id, "reason": random.choice(["provoked", "robbery", "dispute"])}, timestamp=ts, ) def _move_toward(self, target: tuple[int, int]) -> None: ax, ay = self.agent.position tx, ty = target if ax < tx: self.world.move_agent(self.agent, "east") elif ax > tx: self.world.move_agent(self.agent, "west") elif ay < ty: self.world.move_agent(self.agent, "south") elif ay > ty: self.world.move_agent(self.agent, "north") >>> personal_reality.py from __future__ import annotations from models import Agent, Event, ResourceNode class PersonalReality: def __init__(self, agent: Agent) -> None: self.agent = agent self.known_agents: dict[str, Agent] = {} self.known_resources: dict[tuple[int, int], ResourceNode] = {} self.pending_events: list[Event] = [] self.local_time: int = 0 def sync_from_world(self, nearby_agents: list[Agent], nearby_resources: list[ResourceNode]) -> None: self.known_agents.clear() for a in nearby_agents: self.known_agents[a.id] = a self.known_resources.clear() for r in nearby_resources: self.known_resources[r.position] = r def integrate_commit(self, result) -> None: ev = result.resolved_event if ev.actor == self.agent.id: self.agent.last_events.append( f"T{ev.timestamp}:{result.status.value}:{ev.intent.value}" ) def perceive(self) -> dict: return { "position": self.agent.position, "hunger": self.agent.hunger, "credits": self.agent.credits, "inventory": dict(self.agent.inventory), "nearby_agents": list(self.known_agents.keys()), "nearby_resources": [ (r.position, r.resource_type, r.quantity) for r in self.known_resources.values() ], } >>> world_engine.py from __future__ import annotations import random from models import Agent, ResourceNode, DIRECTION_VECTORS class WorldEngine: def __init__(self, width: int = 10, height: int = 10): self.width = width self.height = height self.tick = 0 self.agents: dict[str, Agent] = {} self.resources: list[ResourceNode] = [] self._build_resource_nodes() def _build_resource_nodes(self) -> None: spots = set() for _ in range(8): while True: pos = (random.randint(0, self.width - 1), random.randint(0, self.height - 1)) if pos not in spots: spots.add(pos) break rtype = random.choice(["food", "wood", "ore"]) self.resources.append(ResourceNode(resource_type=rtype, position=pos)) def add_agent(self, agent: Agent) -> None: self.agents[agent.id] = agent def get_agent(self, agent_id: str) -> Agent | None: return self.agents.get(agent_id) def move_agent(self, agent: Agent, direction: str) -> bool: dx, dy = DIRECTION_VECTORS.get(direction, (0, 0)) nx, ny = agent.position[0] + dx, agent.position[1] + dy if 0 <= nx < self.width and 0 <= ny < self.height: agent.position = (nx, ny) return True return False def find_resource_at(self, pos: tuple[int, int], rtype: str | None = None) -> ResourceNode | None: for r in self.resources: if r.position == pos and not r.depleted: if rtype is None or r.resource_type == rtype: return r return None def agents_at(self, pos: tuple[int, int], exclude: str = "") -> list[Agent]: return [a for a in self.agents.values() if a.position == pos and a.id != exclude] def nearby_agents(self, agent: Agent, radius: int | None = None) -> list[Agent]: if radius is None: radius = agent.perceive_radius() ax, ay = agent.position result = [] for a in self.agents.values(): if a.id == agent.id: continue dx = abs(a.position[0] - ax) dy = abs(a.position[1] - ay) if dx <= radius and dy <= radius: result.append(a) return result def nearby_resources(self, agent: Agent, radius: int | None = None) -> list[ResourceNode]: if radius is None: radius = agent.perceive_radius() ax, ay = agent.position return [ r for r in self.resources if not r.depleted and abs(r.position[0] - ax) <= radius and abs(r.position[1] - ay) <= radius ] def update_lod(self) -> None: for agent in self.agents.values(): nearby = self.nearby_agents(agent, radius=2) if len(nearby) > 0: agent.lod_level = 2 else: agent.lod_level = 1 def tick_resources(self) -> None: for r in self.resources: r.regen() if random.random() < 0.08: r.quantity = min(15, r.quantity + random.randint(1, 3)) def tick_hunger(self) -> None: for agent in self.agents.values(): hunger_var = random.gauss(0, 0.01) agent.hunger = min(1.0, max(0.0, agent.hunger + agent.hunger_rate() + hunger_var)) def advance(self) -> int: self.tick += 1 self.tick_resources() self.tick_hunger() self.update_lod() return self.tick def resource_summary(self) -> dict[str, int]: summary: dict[str, int] = {} for r in self.resources: if not r.depleted: summary[r.resource_type] = summary.get(r.resource_type, 0) + r.quantity return summary >>> world_ledger.py from __future__ import annotations from models import Event, CommitResult, CommitStatus class WorldLedger: def __init__(self) -> None: self.entries: list[CommitResult] = [] self._event_index: dict[str, CommitResult] = {} def append(self, result: CommitResult) -> None: self.entries.append(result) self._event_index[result.resolved_event.id] = result def get(self, event_id: str) -> CommitResult | None: return self._event_index.get(event_id) def recent(self, n: int = 5) -> list[CommitResult]: return self.entries[-n:] def events_by(self, actor: str) -> list[CommitResult]: return [e for e in self.entries if e.resolved_event.actor == actor] def committed_count(self) -> int: return sum(1 for e in self.entries if e.status == CommitStatus.COMMIT) def downgraded_count(self) -> int: return sum(1 for e in self.entries if e.status == CommitStatus.DOWNGRADE) def rejected_count(self) -> int: return sum(1 for e in self.entries if e.status == CommitStatus.REJECT) def __len__(self) -> int: return len(self.entries)