Spaces:
Running
Running
| """world_state — the canonical mutable world. | |
| Holds agent positions, occupancy, append-only history, the absolute tick | |
| clock, and conversation cooldowns; only the engine's resolve phase may | |
| mutate it. | |
| Architecture: the bottom layer under snapshot.py (read view), checkpoint | |
| manager, and the engine's sync step. | |
| Design: mutation is restricted by convention — the engine mirrors the | |
| registry into WorldState exactly twice per tick. | |
| """ | |
| from __future__ import annotations | |
| from datetime import datetime, timezone | |
| from enum import Enum | |
| from typing import Dict, List, Optional | |
| from pydantic import BaseModel, Field | |
| from src.core.log import get_logger | |
| # Logging | |
| logger = get_logger(__name__) | |
| # Enums / small value types | |
| class AgentStatus(str, Enum): | |
| IDLE = "idle" # no current action, needs a decision | |
| ACTING = "acting" # mid-action, action_end_tick in the future | |
| INTERRUPTED = "interrupted" # was acting, got interrupted by a perceived event | |
| class Position(BaseModel): | |
| """Tile coordinates plus an optional semantic zone name.""" | |
| x: int | |
| y: int | |
| location_id: Optional[str] = None # e.g. "cafeteria", "dorm_room_1" | |
| class CurrentAction(BaseModel): | |
| """An action an agent has committed to, spanning [start_tick, end_tick).""" | |
| description: str | |
| start_tick: int | |
| end_tick: int | |
| target_location_id: Optional[str] = None | |
| target_object_id: Optional[str] = None | |
| # set by the resolver if a conflict forced a different outcome than requested | |
| was_reprioritized: bool = False | |
| def is_finished(self, tick: int) -> bool: | |
| return tick >= self.end_tick | |
| class AgentState(BaseModel): | |
| agent_id: str | |
| position: Position | |
| status: AgentStatus = AgentStatus.IDLE | |
| current_action: Optional[CurrentAction] = None | |
| last_updated_tick: int = 0 | |
| class ActionLogEntry(BaseModel): | |
| """One row of the append-only history, used later for replay/debugging.""" | |
| tick: int | |
| agent_id: str | |
| action: str | |
| timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) | |
| # WorldState | |
| class WorldState(BaseModel): | |
| """ | |
| Mutable, canonical world state. | |
| `agents`: dict of : agent_id -> AgentState | |
| `occupancy`: resource_id (a location slot or object) -> agent_id holding | |
| it, or None if free. Register a resource once with `register_resource` | |
| before anyone can occupy it -- this catches typos in resource ids | |
| early instead of silently creating them on first use. | |
| `history`: append-only action log, source of truth for replay/debugging. | |
| """ | |
| tick: int = 0 | |
| agents: Dict[str, AgentState] = Field(default_factory=dict) | |
| occupancy: Dict[str, Optional[str]] = Field(default_factory=dict) | |
| history: List[ActionLogEntry] = Field(default_factory=list) | |
| # Conversation cooldown tracking | |
| agent_last_conversation: Dict[str, int] = Field(default_factory=dict) | |
| conversation_cooldown_ticks: int = 30 | |
| # Agent registry | |
| def register_agent(self, agent_id: str, position: Position) -> None: | |
| if agent_id in self.agents: | |
| logger.warning("Agent '%s' already registered; overwriting.", agent_id) | |
| self.agents[agent_id] = AgentState( | |
| agent_id=agent_id, position=position, last_updated_tick=self.tick | |
| ) | |
| logger.info( | |
| "Registered agent '%s' at (%d, %d) [tick=%d]", | |
| agent_id, position.x, position.y, self.tick, | |
| ) | |
| def remove_agent(self, agent_id: str) -> None: | |
| """Remove an agent and all live references during a stopped roster edit.""" | |
| self.release_all_for_agent(agent_id) | |
| self.agents.pop(agent_id, None) | |
| self.agent_last_conversation.pop(agent_id, None) | |
| self.history = [entry for entry in self.history if entry.agent_id != agent_id] | |
| def get_agent(self, agent_id: str) -> AgentState: | |
| if agent_id not in self.agents: | |
| raise KeyError(f"Unknown agent_id '{agent_id}'") | |
| return self.agents[agent_id] | |
| def all_agent_ids(self) -> List[str]: | |
| return list(self.agents.keys()) | |
| # Occupancy / resource contention with agents | |
| def register_resource(self, resource_id: str) -> None: | |
| """Idempotent -- safe to call every time a location/object is defined.""" | |
| self.occupancy.setdefault(resource_id, None) | |
| 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 occupy(self, resource_id: str, agent_id: str) -> bool: | |
| """ | |
| Attempt to claim a resource for an agent. | |
| Returns False (and does NOT mutate state) if it's already held by a | |
| *different* agent -- resolver.py is expected to check this return | |
| value and arbitrate, not assume success. | |
| """ | |
| if resource_id not in self.occupancy: | |
| logger.warning("Cannot occupy unregistered resource '%s'.", resource_id) | |
| return False | |
| holder = self.occupancy[resource_id] | |
| if holder is not None and holder != agent_id: | |
| return False | |
| self.occupancy[resource_id] = agent_id | |
| return True | |
| def release(self, resource_id: str) -> None: | |
| if resource_id in self.occupancy: | |
| self.occupancy[resource_id] = None | |
| def release_all_for_agent(self, agent_id: str) -> None: | |
| """Call this when clearing/interrupting an agent's action.""" | |
| for resource_id, holder in self.occupancy.items(): | |
| if holder == agent_id: | |
| self.occupancy[resource_id] = None | |
| # Mutations -- resolve phase only | |
| def set_agent_action(self, agent_id: str, action: CurrentAction) -> None: | |
| agent = self.get_agent(agent_id) | |
| # The engine mirrors its registry into WorldState twice per tick. That | |
| # mirror must be idempotent: unchanged actions are not new replay | |
| # events, and logging each mirror makes checkpoint history grow with | |
| # every tick rather than with meaningful action transitions. | |
| if agent.status == AgentStatus.ACTING and agent.current_action == action: | |
| return | |
| agent.current_action = action | |
| agent.status = AgentStatus.ACTING | |
| agent.last_updated_tick = self.tick | |
| self.history.append( | |
| ActionLogEntry(tick=self.tick, agent_id=agent_id, action=action.description) | |
| ) | |
| def move_agent(self, agent_id: str, position: Position) -> None: | |
| agent = self.get_agent(agent_id) | |
| agent.position = position | |
| agent.last_updated_tick = self.tick | |
| def interrupt_agent(self, agent_id: str) -> None: | |
| """Called by event_bus.py when a perceived event forces a replan.""" | |
| agent = self.get_agent(agent_id) | |
| agent.status = AgentStatus.INTERRUPTED | |
| self.release_all_for_agent(agent_id) | |
| # Conversation cooldown | |
| def record_conversation(self, agent_id: str, tick: Optional[int] = None) -> None: | |
| """Record that an agent finished a conversation at ``tick``.""" | |
| self.agent_last_conversation[agent_id] = self.tick if tick is None else tick | |
| def can_converse(self, agent_id: str, other_id: str) -> bool: | |
| """ | |
| Check if two agents can start a conversation based on cooldown. | |
| Returns False if either agent is still on cooldown. | |
| """ | |
| for aid in (agent_id, other_id): | |
| last = self.agent_last_conversation.get(aid) | |
| if last is not None and self.tick - last < self.conversation_cooldown_ticks: | |
| return False | |
| return True | |
| def clear_agent_action(self, agent_id: str) -> None: | |
| agent = self.get_agent(agent_id) | |
| agent.current_action = None | |
| agent.status = AgentStatus.IDLE | |
| # Scheduling helper (used by engine/scheduler.py) | |
| def agents_ready_for_decision(self) -> List[str]: | |
| """ | |
| Agents who need a decision this tick: no current action at all, | |
| or were flagged INTERRUPTED by the WorldEngine. | |
| Planned action transitions are handled by Actions.py's state | |
| machine -- the tick graph only runs for unplanned gaps or | |
| interruptions. | |
| """ | |
| ready = [] | |
| for agent_id, agent in self.agents.items(): | |
| if ( | |
| agent.current_action is None | |
| or agent.status == AgentStatus.INTERRUPTED | |
| ): | |
| ready.append(agent_id) | |
| return ready | |
| # Tick control, moving the ticks | |
| def advance_tick(self, minutes: int = 10) -> None: | |
| if minutes <= 0: | |
| raise ValueError("minutes must be a positive integer") | |
| self.tick += minutes | |
| logger.debug("World tick advanced to %d", self.tick) | |