Spaces:
Running
Running
| """snapshot — immutable point-in-time view of WorldState. | |
| The engine freezes exactly one WorldSnapshot per tick and hands the same | |
| object to every agent's parallel phase, so reads can never observe a | |
| mid-tick mutation. | |
| Architecture: consumed by perceive.py, the engine's observation builder, | |
| and the frontend snapshot projection. | |
| Design: frozen data structures; agents see the same world each tick. | |
| """ | |
| from __future__ import annotations | |
| from typing import Dict, List, Optional | |
| from pydantic import BaseModel, ConfigDict | |
| from src.core.log import get_logger | |
| from src.core.world_state import AgentState, WorldState, Position | |
| logger = get_logger(__name__) | |
| class WorldSnapshot(BaseModel): | |
| """ | |
| Frozen copy of WorldState taken at the start of a tick's decide phase. | |
| Note on depth of immutability: the top-level model is frozen (you can't | |
| reassign `snap.tick`, `snap.agents`, etc.), and every `AgentState` inside | |
| `agents` is a deep copy made at snapshot time -- so even if something | |
| reaches in and mutates `snap.agents["x"].position`, that mutation is | |
| inert: it's a throwaway copy, isolated from the real WorldState and from | |
| every other agent's snapshot reference. It cannot leak into the resolve | |
| phase. Don't rely on that as a substitute for writing correct code, but | |
| it does mean a stray mutation is a bug in one agent's own reasoning, not | |
| a simulation-wide correctness issue. | |
| """ | |
| model_config = ConfigDict(frozen=True) | |
| tick: int | |
| agents: Dict[str, AgentState] | |
| occupancy: Dict[str, Optional[str]] | |
| # Read-only queries | |
| def get_agent(self, agent_id: str) -> AgentState: | |
| if agent_id not in self.agents: | |
| raise KeyError(f"Unknown agent_id '{agent_id}' in snapshot") | |
| return self.agents[agent_id] | |
| def all_agent_ids(self) -> List[str]: | |
| return list(self.agents.keys()) | |
| def other_agents(self, agent_id: str) -> List[AgentState]: | |
| """Every agent except `agent_id`. Convenience for perceive.py.""" | |
| return [a for a in self.agents.values() if a.agent_id != agent_id] | |
| def agents_near(self, agent_id: str, radius: int) -> List[AgentState]: | |
| """ | |
| Chebyshev-distance ("square") radius query -- i.e. everyone within | |
| a (radius*2+1) x (radius*2+1) tile box centered on `agent_id`. | |
| perceive.py should not reimplement this math, just call this method. | |
| """ | |
| ### NOTE : Can swap the distance metric to squared euclidian distance. | |
| me = self.get_agent(agent_id) | |
| nearby = [] | |
| for other in self.other_agents(agent_id): | |
| dx = abs(other.position.x - me.position.x) | |
| dy = abs(other.position.y - me.position.y) | |
| if max(dx, dy) <= radius: | |
| nearby.append(other) | |
| return nearby | |
| def agents_within_px(self, agent_id: str, radius_px: float) -> List[AgentState]: | |
| """ | |
| Euclidean (circle) proximity query in **pixel** space -- everyone whose | |
| pixel position is within `radius_px` of `agent_id`, nearest first. | |
| This is the query behind the 50px proximity circle used for perception | |
| and proximity-based conversation triggering. Agent positions are pixel | |
| coordinates, so this is a true straight-line distance. | |
| """ | |
| me = self.get_agent(agent_id) | |
| r2 = float(radius_px) * float(radius_px) | |
| scored = [] | |
| for other in self.other_agents(agent_id): | |
| dx = other.position.x - me.position.x | |
| dy = other.position.y - me.position.y | |
| d2 = dx * dx + dy * dy | |
| if d2 <= r2: | |
| scored.append((d2, other)) | |
| scored.sort(key=lambda t: t[0]) | |
| return [o for _d2, o in scored] | |
| def distance_px(self, agent_a: str, agent_b: str) -> float: | |
| """Straight-line pixel distance between two agents.""" | |
| a = self.get_agent(agent_a) | |
| b = self.get_agent(agent_b) | |
| dx = a.position.x - b.position.x | |
| dy = a.position.y - b.position.y | |
| return (dx * dx + dy * dy) ** 0.5 | |
| def crowd_at(self, location_id: str) -> int: | |
| """How many agents currently share a named location (crowdedness).""" | |
| return len(self.agents_at_location(location_id)) | |
| def agents_at_location(self, location_id: str) -> List[AgentState]: | |
| """Everyone currently standing in a named semantic zone.""" | |
| return [a for a in self.agents.values() if a.position.location_id == location_id] | |
| def is_free(self, resource_id: str) -> bool: | |
| if resource_id not in self.occupancy: | |
| logger.warning("Resource '%s' was never registered.", resource_id) | |
| return self.occupancy.get(resource_id) is None | |
| def holder_of(self, resource_id: str) -> Optional[str]: | |
| return self.occupancy.get(resource_id) | |
| def free_resources(self) -> List[str]: | |
| return [rid for rid, holder in self.occupancy.items() if holder is None] | |
| def take_snapshot(world: WorldState) -> WorldSnapshot: | |
| """ | |
| Build an immutable snapshot from the live WorldState. | |
| Call this exactly once per tick, at the very start of the decide phase, | |
| before any agent tick graph runs -- then pass the *same* `WorldSnapshot` | |
| instance to every agent invoked via asyncio.gather() this tick. Do not | |
| call this once per agent; that would let a slow tick graph observe a | |
| world one tick "younger" than a fast one that started later in the same | |
| gather() batch, which reintroduces the exact ordering bug lockstep is | |
| meant to eliminate. | |
| """ | |
| agents_copy = { | |
| agent_id: agent_state.model_copy(deep=True) | |
| for agent_id, agent_state in world.agents.items() | |
| } | |
| occupancy_copy = dict(world.occupancy) # values are str | None, shallow copy is sufficient | |
| snapshot = WorldSnapshot( | |
| tick=world.tick, | |
| agents=agents_copy, | |
| occupancy=occupancy_copy, | |
| ) | |
| logger.debug( | |
| "Snapshot taken at tick=%d (%d agents, %d resources)", | |
| world.tick, len(agents_copy), len(occupancy_copy), | |
| ) | |
| return snapshot | |
| # Standalone sanity check -- matches the "Done when" check from the build plan: | |
| # instantiate WorldState with one dummy agent, take a snapshot, mutate the | |
| # original, confirm the snapshot didn't change. | |
| if __name__ == "__main__": | |
| from src.core.world_state import CurrentAction | |
| world = WorldState() | |
| world.register_agent("test_agent", Position(x=0, y=0, location_id="start")) | |
| world.register_resource("test_chair") | |
| snap = take_snapshot(world) | |
| assert snap.tick == 0 | |
| assert snap.get_agent("test_agent").position.x == 0 | |
| assert snap.is_free("test_chair") | |
| # mutate the REAL world after the snapshot was taken | |
| world.move_agent("test_agent", Position(x=5, y=5, location_id="cafeteria")) | |
| world.occupy("test_chair", "test_agent") | |
| world.advance_tick() | |
| # snapshot must be unaffected | |
| assert snap.tick == 0, "snapshot tick leaked a live update" | |
| assert snap.get_agent("test_agent").position.x == 0, "snapshot agent position leaked a live update" | |
| assert snap.is_free("test_chair"), "snapshot occupancy leaked a live update" | |
| # frozen model: reassignment must raise | |
| try: | |
| snap.tick = 999 | |
| raise AssertionError("expected snapshot mutation to raise, it didn't") | |
| except Exception as e: | |
| assert type(e).__name__ in ("ValidationError", "TypeError"), f"unexpected error type: {type(e)}" | |
| print("snapshot.py sanity check passed.") |