+
Read model, validator and engine records from ledger.jsonl. Requires ADMIN_TOKEN.
+
+ """
+ )
+ with gr.Column(elem_classes="gs-card-inputs"):
+ with gr.Row(elem_classes="status-row"):
+ admin_token = gr.Textbox(
+ label="Admin token",
+ placeholder="ADMIN_TOKEN",
+ max_lines=1,
+ type="password",
+ scale=2,
+ )
+ admin_run = gr.Textbox(
+ label="Run id",
+ value="current",
+ max_lines=1,
+ scale=1,
+ )
+ with gr.Row(elem_classes="status-row"):
+ admin_phase = gr.Textbox(label="phase", max_lines=1)
+ admin_npc = gr.Textbox(label="npc_id", max_lines=1)
+ admin_tick_from = gr.Number(label="tick from", precision=0)
+ admin_tick_to = gr.Number(label="tick to", precision=0)
+ admin_limit = gr.Number(label="limit", value=100, precision=0)
+ admin_btn = gr.Button("Load ledger records")
+ admin_out = gr.JSON(label="Ledger records")
+
gr.HTML(_GROUND_FOOTER_HTML)
create_submit.click(
_make_create_handler(runtime),
- inputs=[name_box, icon_box],
+ inputs=[name_box, icon_box, country_box],
outputs=[create_status, token_group, token_box, mcp_url_box, config_box],
api_name="create_character",
)
@@ -149,12 +192,27 @@ def build_public_ui(runtime: GameRuntime, *, world_path: str = "/world") -> gr.B
api_name="check_status",
)
+ admin_btn.click(
+ _make_admin_logs_handler(runtime),
+ inputs=[
+ admin_token,
+ admin_run,
+ admin_phase,
+ admin_npc,
+ admin_tick_from,
+ admin_tick_to,
+ admin_limit,
+ ],
+ outputs=[admin_out],
+ api_name="admin_logs",
+ )
+
return blocks
def _make_create_handler(runtime: GameRuntime) -> Any:
- def create(name: str, icon: str, request: gr.Request) -> tuple[Any, ...]:
- status, payload = runtime.create_player(name=name, icon=icon)
+ def create(name: str, icon: str, country: str, request: gr.Request) -> tuple[Any, ...]:
+ status, payload = runtime.create_player(name=name, icon=icon, country_id=country)
if status != HTTPStatus.OK:
message = payload.get("message", "Could not create character.")
return (
@@ -203,10 +261,40 @@ def _make_status_handler(runtime: GameRuntime) -> Any:
return check
+def _make_admin_logs_handler(runtime: GameRuntime) -> Any:
+ def load(
+ token: str,
+ run_id: str,
+ phase: str,
+ npc_id: str,
+ tick_from: float | None,
+ tick_to: float | None,
+ limit: float | None,
+ ) -> dict[str, Any]:
+ if token != os.getenv("ADMIN_TOKEN", "dev-admin-token"):
+ return {"error": "admin_token_required"}
+ status, payload = runtime.admin_log_records(
+ run_id=(run_id or "current").strip() or "current",
+ phase=(phase or "").strip() or None,
+ npc_id=(npc_id or "").strip() or None,
+ tick_from=int(tick_from) if tick_from is not None else None,
+ tick_to=int(tick_to) if tick_to is not None else None,
+ limit=int(limit) if limit is not None else 100,
+ )
+ if status != HTTPStatus.OK:
+ return payload
+ runs = runtime.admin_logs_runs()
+ return {**payload, **runs}
+
+ return load
+
+
def _format_status(payload: dict[str, Any]) -> str:
icon = payload.get("icon", "")
name = payload.get("name", "?")
lines = [f"### {icon} {name}"]
+ if payload.get("country_id"):
+ lines.append(f"Country: `{payload['country_id']}`")
if not payload.get("spawned"):
lines.append("⏳ Not in the world yet — call `observe` or `act` over MCP to join.")
return "\n\n".join(lines)
@@ -226,7 +314,8 @@ def _format_status(payload: dict[str, Any]) -> str:
)
lines.append(
f"- Inventory — food: {inv.get('food', 0)}, herbs: {inv.get('herbs', 0)}, "
- f"wood: {inv.get('wood', 0)}, weapon: {inv.get('weapon', 0)}"
+ f"wood: {inv.get('wood', 0)}, weapon: {inv.get('weapon', 0)}, "
+ f"coins: {inv.get('coins', 0)}"
)
lines.append(f"- Doing: {me.get('intention', '—')} (tick {payload.get('tick')})")
return "\n".join(lines)
diff --git a/src/god_simulator/api/runtime.py b/src/god_simulator/api/runtime.py
index b70ec1d1f2f5549a3a908d1b569f134b5f87dc33..e3e3d367386e6f293fa0351878635130aa12d71b 100644
--- a/src/god_simulator/api/runtime.py
+++ b/src/god_simulator/api/runtime.py
@@ -1,9 +1,12 @@
from __future__ import annotations
+import json
+import os
import time
from copy import deepcopy
-from dataclasses import asdict, dataclass, field
+from dataclasses import dataclass
from http import HTTPStatus
+from pathlib import Path
from threading import Lock, Thread
from typing import Any, Protocol
from urllib.error import HTTPError, URLError
@@ -14,20 +17,13 @@ from god_simulator.api.players import (
PlayerError,
PlayerManager,
)
+from god_simulator.observability import RunLedger, append_record, set_active_tick_buffer
from god_simulator.config import ConnectorConfig, GameConfig
-from god_simulator.domain import WorldState
+from god_simulator.domain import WorldLogEvent, WorldState
from god_simulator.rendering.scene_contract import to_scene_snapshot
-from god_simulator.simulation.chaos import (
- CHAOS_COOLDOWNS,
- CHAOS_TOOL_LABELS,
- apply_chaos_action,
-)
from god_simulator.simulation.connectors.base import TickPlan, WorldSimulator
from god_simulator.simulation.connectors.factory import create_world_simulator
-from god_simulator.simulation.god_console import GodConsole, apply_god_command_plan
-from god_simulator.simulation.god_console import GodCommandPlan
from god_simulator.simulation.directives import expire_directives
-from god_simulator.simulation.overseer import OverseerController, create_overseer
from god_simulator.simulation.tick import apply_tick_plan, plan_world_tick
@@ -37,21 +33,19 @@ class GameRuntime:
*,
world: WorldState,
simulator: WorldSimulator,
- god_console: GodConsole | None = None,
modal_health_warmer: HealthWarmer | None = None,
- overseer: OverseerController | None = None,
players: PlayerManager | None = None,
+ ledger: RunLedger | None = None,
+ config: GameConfig | None = None,
) -> None:
self._world = world
self._simulator = simulator
- self._god_console = god_console
self._modal_health_warmer = modal_health_warmer
- self._overseer = overseer
self._players = players or PlayerManager()
+ self._ledger = ledger or RunLedger()
+ self._config = config
self._lock = Lock()
self._tick_status = TickStatus()
- self._god_status = GodCommandStatus()
- self._chaos_status = ChaosStatus()
self._pending_commands: list[PendingCommand] = []
@property
@@ -84,9 +78,6 @@ class GameRuntime:
tick_in_progress=self._tick_status.in_progress,
pending_tick=self._tick_status.pending_tick,
)
- payload["simulation"]["chaos_tools"] = chaos_tools_payload(
- self._chaos_status, self._world.tick
- )
if self._modal_health_warmer is not None:
payload["simulation"]["models"] = self._modal_health_warmer.statuses()
return payload
@@ -104,20 +95,13 @@ class GameRuntime:
npc_order=[npc.id for npc in planning_world.living_npcs()],
)
set_active_tick_buffer(buffer)
- prepared_commands: list[PreparedCommand] = []
try:
- prepared_commands = self._prepare_pending_commands(
- planning_world,
- pending_commands,
- next_tick=next_tick,
- )
expire_directives(planning_world, tick=next_tick)
planned_tick, plan = plan_world_tick(
planning_world,
self._simulator,
next_tick,
- overseer=self._overseer,
)
except Exception as exc:
with self._lock:
@@ -130,91 +114,29 @@ class GameRuntime:
"message": str(exc),
}
- with self._lock:
- plan = self._override_player_directives(planning_world, plan)
- apply_tick_plan(self._world, planned_tick, plan)
- append_record(
- {
- "tick": planned_tick,
- "phase": "engine_events",
- "events": [
- event
- for event in self._world.to_dict()["event_log"][before_events:]
- ],
- }
- )
- self._tick_status.in_progress = False
- self._tick_status.pending_tick = None
- payload = self._world.to_dict()
- payload["simulation"] = tick_status_payload(self._tick_status)
- self._ledger.flush_tick(buffer)
- set_active_tick_buffer(None)
- return HTTPStatus.OK, payload
-
- def god_command(self, command: str) -> tuple[HTTPStatus, dict[str, Any]]:
- if self._god_console is None:
- return HTTPStatus.BAD_REQUEST, {"error": "god_console_unavailable"}
- if not command.strip():
- return HTTPStatus.BAD_REQUEST, {
- "error": "invalid_request",
- "message": "Expected non-empty 'command'.",
- }
-
- with self._lock:
- self._pending_commands.append(
- PendingCommand(kind="god_console", raw_input=command.strip())
- )
- applies_on_tick = self._queued_command_tick()
- return HTTPStatus.OK, {
- "source": plan.source,
- "summary": plan.summary,
- "applied": applied,
- "snapshot": to_scene_snapshot(
- self._world,
- tick_in_progress=self._tick_status.in_progress,
- pending_tick=self._tick_status.pending_tick,
- ),
- }
-
- def chaos_action(self, action: str) -> tuple[HTTPStatus, dict[str, Any]]:
- if action not in CHAOS_COOLDOWNS:
- return HTTPStatus.BAD_REQUEST, {
- "error": "unknown_chaos_action",
- "message": f"Unknown chaos action: {action}",
- "actions": sorted(CHAOS_COOLDOWNS),
- }
-
- with self._lock:
- last_used = self._chaos_status.last_used.get(action)
- if last_used is not None:
- remaining = last_used + CHAOS_COOLDOWNS[action] - self._world.tick
- if remaining > 0:
- return HTTPStatus.CONFLICT, {
- "error": "chaos_on_cooldown",
- "message": (
- f"{CHAOS_TOOL_LABELS[action]} ready in {remaining} ticks."
- ),
- "remaining_ticks": remaining,
+ try:
+ with self._lock:
+ before_events = len(self._world.event_log)
+ plan = self._override_player_directives(planning_world, plan)
+ apply_tick_plan(self._world, planned_tick, plan)
+ append_record(
+ {
+ "tick": planned_tick,
+ "phase": "engine_events",
+ "events": [
+ event
+ for event in self._world.to_dict()["event_log"][before_events:]
+ ],
}
-
- applied = apply_chaos_action(self._world, action)
- if applied:
- self._chaos_status.last_used[action] = self._world.tick
- snapshot = to_scene_snapshot(
- self._world,
- tick_in_progress=self._tick_status.in_progress,
- pending_tick=self._tick_status.pending_tick,
- )
- snapshot["simulation"]["chaos_tools"] = chaos_tools_payload(
- self._chaos_status, self._world.tick
- )
- return HTTPStatus.OK, {
- "action": action,
- "queued": True,
- "applies_on_tick": applies_on_tick,
- "applied": [],
- "summary": f"{CHAOS_TOOL_LABELS[action]} queued for tick {applies_on_tick}.",
- }
+ )
+ self._tick_status.in_progress = False
+ self._tick_status.pending_tick = None
+ payload = self._world.to_dict()
+ payload["simulation"] = tick_status_payload(self._tick_status)
+ self._ledger.flush_tick(buffer)
+ return HTTPStatus.OK, payload
+ finally:
+ set_active_tick_buffer(None)
# -- player characters (external agents over MCP) ---------------------- #
def create_player(
@@ -222,12 +144,14 @@ class GameRuntime:
*,
name: str,
icon: str,
+ country_id: str = "nemotron",
) -> tuple[HTTPStatus, dict[str, Any]]:
with self._lock:
try:
character = self._players.create(
name=name,
icon=icon,
+ country_id=country_id,
created_tick=self._world.tick,
)
except PlayerError as exc:
@@ -272,6 +196,172 @@ class GameRuntime:
"observation": observation,
}
+ def player_vote_available(self, token: str) -> bool:
+ with self._lock:
+ try:
+ character = self._players.get(token)
+ except PlayerError:
+ return False
+ for election in self._world.elections:
+ if election.country_id != character.country_id or election.completed:
+ continue
+ if election.start_tick <= self._world.tick <= election.end_tick:
+ return True
+ return False
+
+ # -- admin ------------------------------------------------------------- #
+ def admin_logs_runs(self) -> dict[str, Any]:
+ root = self._ledger.run_dir.parent
+ runs: list[dict[str, Any]] = []
+ for run_dir in sorted(root.glob("run_*"), key=lambda item: item.name, reverse=True):
+ if not run_dir.is_dir():
+ continue
+ ledger_path = run_dir / "ledger.jsonl"
+ runs.append(
+ {
+ "run_id": run_dir.name,
+ "current": run_dir == self._ledger.run_dir,
+ "ledger_path": str(ledger_path),
+ "exists": ledger_path.is_file(),
+ "size_bytes": ledger_path.stat().st_size if ledger_path.is_file() else 0,
+ }
+ )
+ return {"runs": runs}
+
+ def admin_log_records(
+ self,
+ *,
+ run_id: str | None = None,
+ since_line: int = 0,
+ limit: int = 200,
+ phase: str | None = None,
+ npc_id: str | None = None,
+ tick_from: int | None = None,
+ tick_to: int | None = None,
+ ) -> tuple[HTTPStatus, dict[str, Any]]:
+ run_dir = self._ledger.run_dir if run_id in (None, "current") else self._run_dir(run_id)
+ if run_dir is None:
+ return HTTPStatus.NOT_FOUND, {"error": "run_not_found", "message": "Unknown ledger run."}
+ ledger_path = run_dir / "ledger.jsonl"
+ if not ledger_path.is_file():
+ return HTTPStatus.NOT_FOUND, {"error": "ledger_not_found", "message": "Run has no ledger.jsonl."}
+
+ records: list[dict[str, Any]] = []
+ next_line = max(0, since_line)
+ limit = max(1, min(limit, 1000))
+ with ledger_path.open("r", encoding="utf-8") as ledger_file:
+ for line_no, line in enumerate(ledger_file):
+ next_line = line_no + 1
+ if line_no < since_line:
+ continue
+ try:
+ record = json.loads(line)
+ except json.JSONDecodeError:
+ continue
+ if not _record_matches(
+ record,
+ phase=phase,
+ npc_id=npc_id,
+ tick_from=tick_from,
+ tick_to=tick_to,
+ ):
+ continue
+ records.append(_normalize_log_record(line_no, record))
+ if len(records) >= limit:
+ break
+ return HTTPStatus.OK, {
+ "run_id": run_dir.name,
+ "ledger_path": str(ledger_path),
+ "next_line": next_line,
+ "records": records,
+ }
+
+ def admin_models(self) -> dict[str, Any]:
+ return {
+ "profiles": self._model_profiles(),
+ "npcs": [
+ {
+ "id": npc.id,
+ "name": npc.name,
+ "country_id": npc.country_id,
+ "model_profile_id": npc.model_profile_id,
+ "connector_id": npc.connector_id,
+ }
+ for npc in self._world.npcs
+ ],
+ }
+
+ def admin_set_npc_model(self, npc_id: str, profile_id: str) -> tuple[HTTPStatus, dict[str, Any]]:
+ profiles = {profile["id"]: profile for profile in self._model_profiles()}
+ if profile_id not in profiles:
+ return HTTPStatus.BAD_REQUEST, {"error": "unknown_profile", "message": "Unknown model profile."}
+ profile = profiles[profile_id]
+ connector_id = profile.get("connector_id")
+ with self._lock:
+ npc = next((candidate for candidate in self._world.npcs if candidate.id == npc_id), None)
+ if npc is None:
+ return HTTPStatus.NOT_FOUND, {"error": "npc_not_found", "message": "Unknown NPC."}
+ npc.model_profile_id = profile_id
+ npc.connector_id = connector_id if isinstance(connector_id, str) and connector_id else None
+ self._world.event_log.append(
+ WorldLogEvent(
+ tick=self._world.tick,
+ type="model_profile_changed",
+ summary=f"{npc.name} model profile set to {profile_id}",
+ severity="neutral",
+ actor_id=npc.id,
+ )
+ )
+ self._ledger.write_records(
+ [
+ {
+ "tick": self._world.tick,
+ "phase": "admin_model_switch",
+ "npc_id": npc_id,
+ "model_profile_id": profile_id,
+ "connector_id": connector_id,
+ }
+ ]
+ )
+ return HTTPStatus.OK, {"npc_id": npc_id, "profile": profile}
+
+ def _run_dir(self, run_id: str | None) -> Path | None:
+ if not run_id or "/" in run_id or "\\" in run_id or not run_id.startswith("run_"):
+ return None
+ run_dir = self._ledger.run_dir.parent / run_id
+ if not run_dir.is_dir():
+ return None
+ return run_dir
+
+ def _model_profiles(self) -> list[dict[str, Any]]:
+ profiles: dict[str, dict[str, Any]] = {
+ "default": {
+ "id": "default",
+ "label": "Default NPC model",
+ "connector_id": None,
+ "model": self._config.connector.model if self._config else None,
+ }
+ }
+ if self._config is not None:
+ for connector_id, connector in self._config.secondary_connectors.items():
+ profiles[connector_id] = {
+ "id": connector_id,
+ "label": connector_id,
+ "connector_id": connector_id,
+ "model": connector.model,
+ "type": connector.type,
+ }
+ profiles_path = Path(os.getenv("MODEL_PROFILES_PATH", "config/model_profiles.json"))
+ if profiles_path.is_file():
+ try:
+ raw = json.loads(profiles_path.read_text(encoding="utf-8"))
+ except json.JSONDecodeError:
+ raw = {}
+ for item in raw.get("profiles", []) if isinstance(raw, dict) else []:
+ if isinstance(item, dict) and isinstance(item.get("id"), str):
+ profiles[item["id"]] = dict(item)
+ return list(profiles.values())
+
def _override_player_directives(self, world: WorldState, plan: TickPlan) -> TickPlan:
"""Replace planner directives for player-owned NPCs with their queued actions."""
player_ids = self._players.player_npc_ids
@@ -284,9 +374,16 @@ class GameRuntime:
directive = self._players.take_directive(character, world)
if directive is not None:
directives.append(directive)
- return TickPlan(source=plan.source, directives=directives, overseer=plan.overseer)
+ return TickPlan(
+ source=plan.source,
+ directives=directives,
+ overseer=plan.overseer,
+ ledger_entries=plan.ledger_entries,
+ )
- def _begin_tick(self) -> tuple[WorldState | None, int | None, dict[str, Any] | None]:
+ def _begin_tick(
+ self,
+ ) -> tuple[WorldState | None, int | None, list[PendingCommand] | None, dict[str, Any] | None]:
with self._lock:
if self._tick_status.in_progress:
return None, None, None, {
@@ -294,14 +391,6 @@ class GameRuntime:
"tick": self._world.tick,
**tick_status_payload(self._tick_status),
}
- if self._god_status.in_progress:
- return None, None, None, {
- "error": "god_command_in_progress",
- "message": "A god command is already modifying the world.",
- "tick": self._world.tick,
- **tick_status_payload(self._tick_status),
- }
-
next_tick = self._world.tick + 1
planning_world = deepcopy(self._world)
pending_commands = list(self._pending_commands)
@@ -310,48 +399,80 @@ class GameRuntime:
self._tick_status.pending_tick = next_tick
return planning_world, next_tick, pending_commands, None
- def _begin_god_command(self) -> tuple[WorldState | None, dict[str, Any] | None]:
- with self._lock:
- if self._tick_status.in_progress:
- return None, {
- "error": "tick_in_progress",
- "message": "God commands are disabled while a tick is running.",
- **tick_status_payload(self._tick_status),
- }
- if self._god_status.in_progress:
- return None, {
- "error": "god_command_in_progress",
- "message": "Another god command is already running.",
- **tick_status_payload(self._tick_status),
- }
-
- self._god_status.in_progress = True
- return deepcopy(self._world), None
-
def create_game_runtime(*, world: WorldState, config: GameConfig) -> GameRuntime:
simulator = create_world_simulator(config)
- overseer = create_overseer(config)
- god_console_config = config.god_console or config.connector
- god_console = (
- GodConsole(god_console_config)
- if god_console_config.type == "openai_compatible"
- else None
- )
return GameRuntime(
world=world,
simulator=simulator,
- god_console=god_console,
modal_health_warmer=ModalHealthWarmer(modal_health_targets(config)),
- overseer=overseer,
+ config=config,
)
+def _record_matches(
+ record: dict[str, Any],
+ *,
+ phase: str | None,
+ npc_id: str | None,
+ tick_from: int | None,
+ tick_to: int | None,
+) -> bool:
+ if phase and record.get("phase") != phase:
+ return False
+ if npc_id and record.get("npc_id") != npc_id:
+ return False
+ tick = record.get("tick")
+ if tick_from is not None and isinstance(tick, int) and tick < tick_from:
+ return False
+ if tick_to is not None and isinstance(tick, int) and tick > tick_to:
+ return False
+ return True
+
+
+def _normalize_log_record(line: int, record: dict[str, Any]) -> dict[str, Any]:
+ return {
+ "line": line,
+ "tick": record.get("tick"),
+ "phase": record.get("phase"),
+ "npc_id": record.get("npc_id"),
+ "model": _record_model(record),
+ "summary": _record_summary(record),
+ "record": record,
+ }
+
+
+def _record_model(record: dict[str, Any]) -> str | None:
+ model = record.get("model")
+ if isinstance(model, str):
+ return model
+ request = record.get("request")
+ if isinstance(request, dict) and isinstance(request.get("model"), str):
+ return request["model"]
+ return None
+
+
+def _record_summary(record: dict[str, Any]) -> str:
+ summary = record.get("summary")
+ if isinstance(summary, str) and summary:
+ return summary
+ phase = str(record.get("phase") or "record")
+ parsed = record.get("parsed_action") or record.get("action")
+ verdict = record.get("validator") or record.get("verdict") or record.get("reason")
+ parts = [phase]
+ if parsed:
+ parts.append(f"action={parsed}")
+ if verdict:
+ parts.append(str(verdict))
+ return "; ".join(parts)
+
+
def _character_summary(character: PlayerCharacter) -> dict[str, Any]:
return {
"token": character.token,
"name": character.name,
"icon": character.icon,
+ "country_id": character.country_id,
"npc_id": character.npc_id,
}
@@ -374,44 +495,12 @@ class TickStatus:
pending_tick: int | None = None
-@dataclass(slots=True)
-class GodCommandStatus:
- in_progress: bool = False
-
-
@dataclass(frozen=True, slots=True)
class PendingCommand:
kind: str
raw_input: str
-@dataclass(frozen=True, slots=True)
-class PreparedCommand:
- pending: PendingCommand
- god_plan: GodCommandPlan | None = None
-
-
-@dataclass(slots=True)
-class ChaosStatus:
- last_used: dict[str, int] = field(default_factory=dict)
-
-
-def chaos_tools_payload(chaos_status: ChaosStatus, tick: int) -> list[dict[str, Any]]:
- tools: list[dict[str, Any]] = []
- for action, cooldown in CHAOS_COOLDOWNS.items():
- last_used = chaos_status.last_used.get(action)
- remaining = 0 if last_used is None else max(0, last_used + cooldown - tick)
- tools.append(
- {
- "action": action,
- "label": CHAOS_TOOL_LABELS[action],
- "cooldown_ticks": cooldown,
- "remaining_ticks": remaining,
- }
- )
- return tools
-
-
def tick_status_payload(tick_status: TickStatus) -> dict[str, Any]:
return {
"tick_in_progress": tick_status.in_progress,
@@ -600,10 +689,7 @@ def modal_health_targets(config: GameConfig) -> list[ModalHealthTarget]:
targets: list[ModalHealthTarget] = []
connector_targets: list[tuple[str, str, ConnectorConfig | None]] = [
("npc_model", "NPC model", config.connector),
- ("god_console", "God Console", config.god_console),
]
- if config.overseer.mode != "off":
- connector_targets.append(("overseer", "Overseer", config.overseer.connector))
for target_id, label, connector in connector_targets:
if connector is None:
continue
diff --git a/src/god_simulator/api/server.py b/src/god_simulator/api/server.py
index 980b32784b17a33e5d24e9212bda3443695d2d56..bb70f7bef94cbf4cc3c933dcf05ad766ae98406d 100644
--- a/src/god_simulator/api/server.py
+++ b/src/god_simulator/api/server.py
@@ -31,8 +31,6 @@ from god_simulator.api.runtime import (
from god_simulator.config import GameConfig
from god_simulator.domain import WorldState
from god_simulator.simulation.connectors.base import WorldSimulator
-from god_simulator.simulation.god_console import GodConsole
-from god_simulator.simulation.overseer import OverseerController
__all__ = [
"_ModalHealthTarget",
@@ -55,7 +53,7 @@ def run_server(*, world: WorldState, config: GameConfig) -> None:
print(f"World simulator connector: {runtime.simulator_name}")
print(
"Available endpoints: GET /health, GET /state, GET /scene/state, "
- "POST /tick, POST /god-command, POST /chaos"
+ "POST /tick"
)
server.serve_forever()
@@ -63,17 +61,16 @@ def run_server(*, world: WorldState, config: GameConfig) -> None:
def _build_handler(
world: WorldState,
simulator: WorldSimulator,
- god_console: GodConsole | None = None,
+ god_console: object | None = None,
modal_health_warmer: _HealthWarmer | None = None,
- overseer: OverseerController | None = None,
+ overseer: object | None = None,
) -> type[BaseHTTPRequestHandler]:
+ _ = god_console, overseer
return build_handler(
GameRuntime(
world=world,
simulator=simulator,
- god_console=god_console,
modal_health_warmer=modal_health_warmer,
- overseer=overseer,
)
)
@@ -115,34 +112,6 @@ def build_handler(runtime: GameRuntime) -> type[BaseHTTPRequestHandler]:
self._send_json(payload, status=status)
return
- if self.path == "/god-command":
- try:
- command = self._read_json_field("command")
- except ValueError as exc:
- self._send_json(
- {"error": "invalid_request", "message": str(exc)},
- status=HTTPStatus.BAD_REQUEST,
- )
- return
-
- status, payload = runtime.god_command(command)
- self._send_json(payload, status=status)
- return
-
- if self.path == "/chaos":
- try:
- action = self._read_json_field("action")
- except ValueError as exc:
- self._send_json(
- {"error": "invalid_request", "message": str(exc)},
- status=HTTPStatus.BAD_REQUEST,
- )
- return
-
- status, payload = runtime.chaos_action(action)
- self._send_json(payload, status=status)
- return
-
self._send_json(
{"error": "not_found", "path": self.path},
status=HTTPStatus.NOT_FOUND,
diff --git a/src/god_simulator/domain.py b/src/god_simulator/domain.py
index a6e3faccbc053d06c8be63be234354464c54b215..f4cf7ad6838c0c1df7b22eec56d5186fabd585c2 100644
--- a/src/god_simulator/domain.py
+++ b/src/god_simulator/domain.py
@@ -139,6 +139,7 @@ class Npc:
inventory_herbs: int = 0
inventory_wood: int = 0
inventory_weapon: int = 0
+ inventory_coins: int = 0
help_target_id: str | None = None
# Whom the NPC is currently engaging (talk/attack/trade); drives facing in the renderer.
focus_target_id: str | None = None
@@ -153,6 +154,9 @@ class Npc:
# by the built-in deterministic/LLM planner.
is_player: bool = False
player_owner: str | None = None
+ country_id: str | None = None
+ special_status: str | None = None
+ model_profile_id: str | None = None
# When true, the engine does not role-gate this NPC's actions: it may take
# any action regardless of role. Player NPCs set this, but it can also be
# given to special non-player NPCs.
@@ -197,6 +201,47 @@ class House:
occupant_ids: list[str] = field(default_factory=list)
+@dataclass(slots=True)
+class TreasuryState:
+ id: str
+ position: Vec3
+ resources: dict[str, int] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class CannonState:
+ id: str
+ position: Vec3
+ operator_id: str | None = None
+ cooldown_until_tick: int = 0
+ damage: int = 35
+ radius: int = 5
+
+
+@dataclass(slots=True)
+class CountryState:
+ id: str
+ name: str
+ color: str
+ badge: str
+ citizen_ids: list[str] = field(default_factory=list)
+ ruler_id: str | None = None
+ policy: str = "Gather resources, protect citizens, and avoid reckless war."
+ treasury: TreasuryState | None = None
+ cannon: CannonState | None = None
+ next_election_tick: int = 50
+
+
+@dataclass(slots=True)
+class ElectionState:
+ country_id: str
+ start_tick: int
+ end_tick: int
+ candidate_ids: list[str]
+ votes: dict[str, str] = field(default_factory=dict)
+ completed: bool = False
+
+
@dataclass(slots=True)
class WorldLogEvent:
tick: int
@@ -239,6 +284,8 @@ class WorldState:
houses: list[House] = field(default_factory=list)
active_events: list[WorldEvent] = field(default_factory=list)
event_log: list[WorldLogEvent] = field(default_factory=list)
+ countries: list[CountryState] = field(default_factory=list)
+ elections: list[ElectionState] = field(default_factory=list)
game_status: GameStatus = "running"
population: int = 0
population_cap: int = 12
diff --git a/src/god_simulator/observability.py b/src/god_simulator/observability.py
index 80da431a575baaa28fd8ffb6c964446ace684993..826ba4a8fc96bcbb0cf7c2184db8d718cf2804bd 100644
--- a/src/god_simulator/observability.py
+++ b/src/god_simulator/observability.py
@@ -32,7 +32,9 @@ class RunLedger:
return TickLedgerBuffer(tick=tick, npc_order=npc_order)
def flush_tick(self, buffer: TickLedgerBuffer) -> None:
- records = buffer.sorted_records()
+ self.write_records(buffer.sorted_records())
+
+ def write_records(self, records: list[dict[str, Any]]) -> None:
if not records:
return
with self._lock:
diff --git a/src/god_simulator/rendering/scene_contract.py b/src/god_simulator/rendering/scene_contract.py
index 7842a8eeed8388d888cf42b08f85952f3a648ba2..81eeb84c455e15d16ac5bf61854e8368a3e422a9 100644
--- a/src/god_simulator/rendering/scene_contract.py
+++ b/src/god_simulator/rendering/scene_contract.py
@@ -23,10 +23,6 @@ def to_scene_snapshot(
"population": world.population,
"population_cap": world.population_cap,
"peak_population": world.peak_population,
- "scoreboard": {
- "overseer": world.overseer_score,
- "chaos": world.chaos_score,
- },
"stats": {
"total_births": world.total_births,
"deaths_by_cause": dict(world.deaths_by_cause),
@@ -34,15 +30,6 @@ def to_scene_snapshot(
"beasts_killed": world.beasts_killed,
},
"famine_until": world.famine_until,
- "overseer": {
- "mode": world.overseer_mode,
- "cycle_ticks": world.overseer_cycle_ticks,
- "last_tick": world.overseer_last_tick,
- "status": world.overseer_status,
- "thoughts": world.overseer_last_thoughts,
- "priority_alert": world.overseer_priority_alert,
- "directives": list(world.overseer_last_directives),
- },
},
"terrain": {
"kind": world.terrain.kind,
@@ -62,6 +49,8 @@ def to_scene_snapshot(
"kind": "npc",
"label": npc.name,
"role": npc.role,
+ "country_id": npc.country_id,
+ "special_status": npc.special_status,
"position": {
"x": npc.position.x,
"y": npc.position.y,
@@ -90,6 +79,7 @@ def to_scene_snapshot(
"herbs": npc.inventory_herbs,
"wood": npc.inventory_wood,
"weapon": npc.inventory_weapon,
+ "coins": npc.inventory_coins,
},
"memory_count": len(npc.memory),
"memory_summary": npc.memory_summary,
@@ -126,6 +116,55 @@ def to_scene_snapshot(
}
for npc in world.npcs
],
+ "countries": [
+ {
+ "id": country.id,
+ "name": country.name,
+ "color": country.color,
+ "badge": country.badge,
+ "citizen_ids": list(country.citizen_ids),
+ "ruler_id": country.ruler_id,
+ "policy": country.policy,
+ "next_election_tick": country.next_election_tick,
+ "treasury": None
+ if country.treasury is None
+ else {
+ "id": country.treasury.id,
+ "position": {
+ "x": country.treasury.position.x,
+ "y": country.treasury.position.y,
+ "z": country.treasury.position.z,
+ },
+ "resources": dict(country.treasury.resources),
+ },
+ "cannon": None
+ if country.cannon is None
+ else {
+ "id": country.cannon.id,
+ "position": {
+ "x": country.cannon.position.x,
+ "y": country.cannon.position.y,
+ "z": country.cannon.position.z,
+ },
+ "operator_id": country.cannon.operator_id,
+ "cooldown_until_tick": country.cannon.cooldown_until_tick,
+ "damage": country.cannon.damage,
+ "radius": country.cannon.radius,
+ },
+ }
+ for country in world.countries
+ ],
+ "elections": [
+ {
+ "country_id": election.country_id,
+ "start_tick": election.start_tick,
+ "end_tick": election.end_tick,
+ "candidate_ids": list(election.candidate_ids),
+ "votes": dict(election.votes),
+ "completed": election.completed,
+ }
+ for election in world.elections[-10:]
+ ],
"resource_nodes": [
{
"id": node.id,
diff --git a/src/god_simulator/simulation/actions/builtin.py b/src/god_simulator/simulation/actions/builtin.py
index 8dcdbf54539e855a4cea68fca26a7582e19eace2..3118ed81f325273cb79dcf54a72aec7adda80dd1 100644
--- a/src/god_simulator/simulation/actions/builtin.py
+++ b/src/god_simulator/simulation/actions/builtin.py
@@ -2,7 +2,7 @@ from __future__ import annotations
from god_simulator.simulation.actions.schemas import ActionDefinition
-# The whole action surface: six parameterized primitives. Behavioral flavor
+# The whole public action surface: five parameterized primitives. Behavioral flavor
# (fleeing, pleading, threatening, stealing, healing...) is expressed through
# parameters and free-text intent, not through dedicated action ids.
BUILTIN_ACTIONS: dict[str, ActionDefinition] = {
@@ -23,20 +23,19 @@ BUILTIN_ACTIONS: dict[str, ActionDefinition] = {
),
required_params=("message",),
),
- "strike": ActionDefinition(
- id="strike",
+ "attack": ActionDefinition(
+ id="attack",
category="combat",
description_for_model=(
- "Attack a nearby NPC or beast. The engine resolves movement, damage, and death."
+ "Attack a valid nearby threat or hostile target. The engine resolves approach, range, damage, and death."
),
- required_params=("target_id",),
),
"use": ActionDefinition(
id="use",
category="resource",
description_for_model=(
- "Use something: gather from a resource node (resource_id), eat food "
- "(item=food), or heal with herbs (item=herbs)."
+ "Use something: gather from a resource node, eat food, heal with herbs, "
+ "build/repair/craft an object, or operate a world object."
),
),
"transfer": ActionDefinition(
@@ -47,13 +46,6 @@ BUILTIN_ACTIONS: dict[str, ActionDefinition] = {
),
required_params=("target_id",),
),
- "idle": ActionDefinition(
- id="idle",
- category="passive",
- description_for_model=(
- "Do nothing this tick: wait, observe, or hold position defensively."
- ),
- ),
}
PRIMITIVE_ACTION_IDS: tuple[str, ...] = tuple(BUILTIN_ACTIONS)
diff --git a/src/god_simulator/simulation/actions/hints.py b/src/god_simulator/simulation/actions/hints.py
index d1529042433170877e24f9cf72f2a7281de7721c..f6bf954b13593a0c79b1f16bab9727b8a903aaa6 100644
--- a/src/god_simulator/simulation/actions/hints.py
+++ b/src/god_simulator/simulation/actions/hints.py
@@ -35,21 +35,21 @@ def build_action_hints(
match goal.goal_type:
case "survive" | "call_for_help":
- suggested: list[ActionKind] = ["move", "speak", "idle"]
+ suggested: list[ActionKind] = ["move", "speak"]
if _goal_target_in_range(world, npc, goal):
- suggested.insert(1, "strike")
+ suggested.insert(1, "attack")
return ActionHints(
suggested_actions=suggested,
reason=(
"A threat is nearby: moving away, calling for help, pleading, "
- "defending, or striking back are the most plausible responses"
+ "defending, or attacking back are the most plausible responses"
),
)
case "pursue_and_attack_target":
if _goal_target_in_range(world, npc, goal):
return ActionHints(
- suggested_actions=["strike", "speak"],
+ suggested_actions=["attack", "speak"],
reason="The hostile directive target is within striking range",
)
return ActionHints(
@@ -60,7 +60,7 @@ def build_action_hints(
case "retaliate":
if _goal_target_in_range(world, npc, goal):
return ActionHints(
- suggested_actions=["strike", "speak", "move"],
+ suggested_actions=["attack", "speak", "move"],
reason="The recent attacker is within reach",
)
return ActionHints(
@@ -69,9 +69,9 @@ def build_action_hints(
)
case "help_or_defend":
- helping: list[ActionKind] = ["speak", "idle", "move"]
+ helping: list[ActionKind] = ["speak", "move"]
if _goal_target_in_range(world, npc, goal):
- helping.insert(0, "strike")
+ helping.insert(0, "attack")
return ActionHints(
suggested_actions=helping,
reason="Someone nearby is being harmed; helping or defending is most plausible",
@@ -79,7 +79,7 @@ def build_action_hints(
case "respond_to_event" | "investigate":
return ActionHints(
- suggested_actions=["move", "idle", "speak"],
+ suggested_actions=["move", "speak"],
reason="Something suspicious happened; investigating or watching is most plausible",
)
@@ -91,13 +91,13 @@ def build_action_hints(
case "routine_life":
return ActionHints(
- suggested_actions=["move", "speak", "use", "transfer", "idle"],
+ suggested_actions=["move", "speak", "use", "transfer"],
reason="Nothing urgent is happening; everyday activity is most plausible",
)
return ActionHints(
- suggested_actions=["idle"],
- reason="Unknown goal; observing is the safest default",
+ suggested_actions=["move", "speak"],
+ reason="Unknown goal; cautious observation and communication are the safest defaults",
)
diff --git a/src/god_simulator/simulation/autopilot.py b/src/god_simulator/simulation/autopilot.py
index 798c0023475dce99f2c6efebe9cc928995027c04..43f98822a94834a08862cfa3d35f3fabbd9bf3d6 100644
--- a/src/god_simulator/simulation/autopilot.py
+++ b/src/god_simulator/simulation/autopilot.py
@@ -101,7 +101,7 @@ def strike_or_approach_target(
if distance_between(npc, target) <= ATTACK_RADIUS:
return NpcDirective(
npc_id=npc.id,
- action="strike",
+ action="attack",
target_npc_id=target.id,
message=directive.message,
memory=memory,
diff --git a/src/god_simulator/simulation/connectors/base.py b/src/god_simulator/simulation/connectors/base.py
index 9bef240aa4abd2859d8ed18730eb38cfb14e7066..6d15448fd5606e32030ac65a3847de35176a89e1 100644
--- a/src/god_simulator/simulation/connectors/base.py
+++ b/src/god_simulator/simulation/connectors/base.py
@@ -5,10 +5,10 @@ from typing import Any, Literal, Protocol, TypeAlias
from god_simulator.domain import Vec3, WorldState
-# The public action surface: every NPC decision is one of these six primitives.
+# The public action surface: every NPC decision is one of these primitives.
# Legacy action names (walk, talk, attack, flee, ...) are accepted as raw
# strings at parse/validation boundaries and normalized to primitives there.
-PrimitiveActionKind: TypeAlias = Literal["move", "speak", "strike", "use", "transfer", "idle"]
+PrimitiveActionKind: TypeAlias = Literal["move", "speak", "attack", "use", "transfer", "vote"]
# Survival worlds keep a richer role-gated action surface that resolves inside
# the survival engine (simulation/survival.py) instead of the primitive executor.
SurvivalActionKind: TypeAlias = Literal[
@@ -31,6 +31,9 @@ SurvivalActionKind: TypeAlias = Literal[
"flee",
"defend",
"attack",
+ "vote",
+ "strike",
+ "idle",
]
ActionKind: TypeAlias = PrimitiveActionKind | SurvivalActionKind
@@ -44,6 +47,8 @@ class CitizenAction:
target_entity_id: str | None = None
resource_id: str | None = None
resource_type: str | None = None
+ use_type: str | None = None
+ params: dict[str, Any] | None = None
amount: int | None = None
communication_intent: str | None = None
message: str | None = None
@@ -62,6 +67,8 @@ class NpcDirective:
target_entity_id: str | None = None
resource_id: str | None = None
resource_type: str | None = None
+ use_type: str | None = None
+ params: dict[str, Any] | None = None
amount: int | None = None
communication_intent: str | None = None
message: str | None = None
diff --git a/src/god_simulator/simulation/connectors/openai_compatible.py b/src/god_simulator/simulation/connectors/openai_compatible.py
index e003cff24a6646cca0cad414591238e81954c463..20bb5315e87f965ca8e78d56209afeb345b38d38 100644
--- a/src/god_simulator/simulation/connectors/openai_compatible.py
+++ b/src/god_simulator/simulation/connectors/openai_compatible.py
@@ -37,6 +37,7 @@ from god_simulator.simulation.memory import llm_compact_memory, recent_meaningfu
from god_simulator.simulation.perception import CitizenPerception
from god_simulator.simulation.roles import normalize_role
from god_simulator.simulation.survival import (
+ active_election_for_npc,
is_survival_world,
allowed_actions_for_npc,
suggested_actions_for_npc,
@@ -66,21 +67,13 @@ TOOL_DEFINITIONS: dict[str, dict[str, Any]] = {
"parameters": {
"type": "object",
"properties": {
- "target_id": {
- "type": "string",
- "description": "Optional visible NPC, beast, or resource node id.",
- },
- "target_x": {
+ "x": {
"type": "number",
- "description": "Optional desired x coordinate.",
+ "description": "Destination X coordinate.",
},
- "target_z": {
+ "z": {
"type": "number",
- "description": "Optional desired z coordinate.",
- },
- "away": {
- "type": "boolean",
- "description": "Move away from the target/threat instead of toward it.",
+ "description": "Destination Z coordinate.",
},
},
"required": [],
@@ -120,22 +113,24 @@ TOOL_DEFINITIONS: dict[str, dict[str, Any]] = {
},
},
},
- "strike": {
+ "attack": {
"type": "function",
"function": {
- "name": "strike",
+ "name": "attack",
"description": (
- "Attack a nearby NPC or beast. The engine resolves movement, damage, and death."
+ "Attack a valid nearby threat or hostile target. Give target_entity_id "
+ "to choose a specific target, or omit it to attack the nearest valid threat. "
+ "The engine resolves approach, range and damage."
),
"parameters": {
"type": "object",
"properties": {
- "target_id": {
+ "target_entity_id": {
"type": "string",
- "description": "Visible NPC or beast id to attack.",
+ "description": "Visible NPC, beast, or object id to attack.",
}
},
- "required": ["target_id"],
+ "required": [],
"additionalProperties": False,
},
},
@@ -155,10 +150,22 @@ TOOL_DEFINITIONS: dict[str, dict[str, Any]] = {
"type": "string",
"description": "Nearby resource node id to gather from.",
},
- "item": {
+ "target_entity_id": {
+ "type": "string",
+ "description": "World object/entity id to use.",
+ },
+ "resource_type": {
"type": "string",
- "enum": ["food", "herbs"],
- "description": "Inventory item to consume.",
+ "enum": ["food", "herbs", "wood", "coins"],
+ "description": "Optional resource type for consuming, crafting or treasury actions.",
+ },
+ "use_type": {
+ "type": "string",
+ "description": "Use mode, e.g. eat, heal, gather, build, repair, craft, fire.",
+ },
+ "params": {
+ "type": "object",
+ "description": "Additional structured parameters for the selected use_type.",
},
},
"required": [],
@@ -166,45 +173,6 @@ TOOL_DEFINITIONS: dict[str, dict[str, Any]] = {
},
},
},
- "go_home": {
- "type": "function",
- "function": {
- "name": "go_home",
- "description": "Move into the nearest completed house to regain safety.",
- "parameters": {
- "type": "object",
- "properties": {},
- "required": [],
- "additionalProperties": False,
- },
- },
- },
- "patrol": {
- "type": "function",
- "function": {
- "name": "patrol",
- "description": "Guard action: patrol between village houses and intercept threats.",
- "parameters": {
- "type": "object",
- "properties": {},
- "required": [],
- "additionalProperties": False,
- },
- },
- },
- "build": {
- "type": "function",
- "function": {
- "name": "build",
- "description": "Builder action: spend wood to start or continue building a house.",
- "parameters": {
- "type": "object",
- "properties": {},
- "required": [],
- "additionalProperties": False,
- },
- },
- },
"transfer": {
"type": "function",
"function": {
@@ -217,37 +185,48 @@ TOOL_DEFINITIONS: dict[str, dict[str, Any]] = {
"properties": {
"target_id": {
"type": "string",
- "description": "Nearby NPC id.",
+ "description": "Nearby NPC, treasury, storage, or object target id.",
+ },
+ "target_npc_id": {
+ "type": "string",
+ "description": "Nearby NPC id, if target is a character.",
+ },
+ "target_entity_id": {
+ "type": "string",
+ "description": "Treasury/storage/object target id.",
},
"resource_type": {
"type": "string",
- "enum": ["food", "herbs", "wood", "weapon"],
+ "enum": ["food", "herbs", "wood", "weapon", "coins"],
},
- "amount": {"type": "integer", "minimum": 1, "maximum": 5},
+ "amount": {"type": "integer", "minimum": 1, "maximum": 20},
"take": {
"type": "boolean",
- "description": "Steal from the target instead of giving.",
+ "description": "Steal/take instead of give.",
},
},
- "required": ["target_id"],
+ "required": ["resource_type", "amount"],
"additionalProperties": False,
},
},
},
- "idle": {
+ "vote": {
"type": "function",
"function": {
- "name": "idle",
- "description": "Do nothing this tick: wait, observe, or hold position defensively.",
+ "name": "vote",
+ "description": (
+ "Vote for a ruler candidate during an active election. Use candidate_id "
+ "from the current election candidate list."
+ ),
"parameters": {
"type": "object",
"properties": {
- "intent": {
+ "candidate_id": {
"type": "string",
- "description": "Optional flavor, e.g. observe or defend.",
+ "description": "NPC id of the candidate.",
}
},
- "required": [],
+ "required": ["candidate_id"],
"additionalProperties": False,
},
},
@@ -255,31 +234,50 @@ TOOL_DEFINITIONS: dict[str, dict[str, Any]] = {
}
-PRIMITIVE_TOOL_IDS = ("move", "speak", "strike", "use", "transfer", "idle")
+PRIMITIVE_TOOL_IDS = ("move", "speak", "attack", "use", "transfer")
def primitive_tools() -> list[dict[str, Any]]:
return [TOOL_DEFINITIONS[tool_id] for tool_id in PRIMITIVE_TOOL_IDS]
+def _public_action_suggestions(actions: Sequence[str]) -> list[str]:
+ mapped: list[str] = []
+ for action in actions:
+ if action in {"attack", "strike"}:
+ public = "attack"
+ elif action in {"gather", "consume", "eat", "heal", "build", "repair", "craft"}:
+ public = "use"
+ elif action in {
+ "move",
+ "move_to",
+ "move_to_resource",
+ "go_home",
+ "patrol",
+ "flee",
+ "wander",
+ "find_food",
+ "find_herbs",
+ }:
+ public = "move"
+ elif action in {"communicate", "talk", "speak", "request_trade", "trade"}:
+ public = "speak"
+ elif action in {"transfer", "steal"}:
+ public = "transfer"
+ else:
+ continue
+ if public not in mapped:
+ mapped.append(public)
+ return mapped or ["move", "speak", "use", "transfer"]
+
+
def survival_tools_for_role(role: str) -> list[dict[str, Any]]:
- """Primitives plus role-gated survival extras (GAME_DESIGN role whitelists).
+ """Return the unified public tool surface for every role.
- Builders never see strike, only guards see patrol, only builders see build;
- everyone can go_home. The engine still re-validates every request.
- Pass role="unrestricted" to expose every tool.
+ Roles now shape prompt context and suggestions only. They do not remove
+ base tools from the LLM-facing contract.
"""
- tool_ids = list(PRIMITIVE_TOOL_IDS) + ["go_home"]
- if role == "unrestricted":
- tool_ids += ["patrol", "build"]
- return [TOOL_DEFINITIONS[tool_id] for tool_id in dict.fromkeys(tool_ids)]
- normalized = normalize_role(role)
- if normalized == "guard":
- tool_ids.append("patrol")
- if normalized == "builder":
- tool_ids = [tool_id for tool_id in tool_ids if tool_id != "strike"]
- tool_ids.append("build")
- return [TOOL_DEFINITIONS[tool_id] for tool_id in tool_ids]
+ return primitive_tools()
_UNSET = object() # sentinel: handle all NPCs (backward-compatible default)
@@ -367,6 +365,7 @@ class OpenAICompatibleWorldSimulator:
assert self._config.model is not None
directives: list[NpcDirective] = []
+ ledger_entries: list[dict[str, Any]] = []
if self._connector_id_filter is _UNSET:
living_npcs = [npc for npc in world.npcs if is_alive(npc)]
else:
@@ -374,7 +373,10 @@ class OpenAICompatibleWorldSimulator:
npc for npc in world.npcs
if is_alive(npc) and npc.connector_id == self._connector_id_filter
]
- directives.extend(self._call_model_for_npcs(world, next_tick, living_npcs))
+ npc_by_id = {npc.id: npc for npc in living_npcs}
+ for result in self._call_model_for_npcs(world, next_tick, living_npcs):
+ directives.extend(result.directives)
+ ledger_entries.extend(result.ledger_entries)
missing_npcs = _missing_npcs(living_npcs, directives)
if missing_npcs:
@@ -387,6 +389,11 @@ class OpenAICompatibleWorldSimulator:
directives.extend(result.directives)
ledger_entries.extend(result.ledger_entries)
+ # Stamp model attribution on every entry (response/fallback as well as
+ # request) so the admin log can show which model drove each NPC.
+ for entry in ledger_entries:
+ _stamp_model_attribution(entry, npc_by_id.get(str(entry.get("npc_id"))), self._config.model)
+
return TickPlan(
source=self.name,
directives=_dedupe_directives(directives),
@@ -421,7 +428,7 @@ class OpenAICompatibleWorldSimulator:
world: WorldState,
next_tick: int,
npc: Npc,
- ) -> list[NpcDirective]:
+ ) -> NpcCallResult:
assert self._config.model is not None
llm_compact_memory(npc, self._chat_completer, self._config.model)
request = self._build_request(world, next_tick, npc)
@@ -553,10 +560,21 @@ class OpenAICompatibleWorldSimulator:
if is_survival_world(world):
survival_goal = select_survival_goal(npc, world)
- suggested = suggested_actions_for_npc(npc, survival_goal)
- messages = _build_survival_messages(world, next_tick, npc, survival_goal, suggested)
- role_key = "unrestricted" if npc.unrestricted_actions else npc.role
- tools = survival_tools_for_role(role_key)
+ allowed = list(PRIMITIVE_TOOL_IDS)
+ if active_election_for_npc(world, npc, tick=next_tick) is not None:
+ allowed.append("vote")
+ suggested = _public_action_suggestions(suggested_actions_for_npc(npc, survival_goal))
+ if "vote" in allowed and "vote" not in suggested:
+ suggested.insert(0, "vote")
+ messages = _build_survival_messages(
+ world,
+ next_tick,
+ npc,
+ survival_goal,
+ suggested,
+ allowed_actions=allowed,
+ )
+ tools = [TOOL_DEFINITIONS[tool_id] for tool_id in allowed]
else:
perception, goal = refresh_citizen_motivation(world, npc)
hints = build_action_hints(world, npc, perception, goal)
@@ -629,13 +647,13 @@ def _build_messages(
"do not invent npc ids",
f"move covers at most {MAX_WALK_DISTANCE:g} world units per tick",
f"speak requires target distance <= {TALK_RADIUS:g} world units",
- f"strike requires target distance <= {ATTACK_RADIUS:g} world units",
+ f"attack requires target distance <= {ATTACK_RADIUS:g} world units",
f"each NPC can see only visible_npcs within {VISIBLE_RADIUS:g} world units",
"visible_npcs is the full set of other NPCs you can currently observe",
"do not target dead NPCs",
"your current health and recent observations should affect your decision",
"treat current_goal as the strongest behavioral priority",
- "treat god_directive as strong input, already interpreted by current_goal",
+ "treat current policy/directive context as behavioral input only",
"your personality should shape action choice and tone",
"message must be direct speech, not narration about yourself",
"the simulation will report what happened after your action",
@@ -644,7 +662,7 @@ def _build_messages(
"the engine resolves damage and death"
),
(
- "out-of-range speak or strike is automatically converted "
+ "out-of-range speak or attack is automatically converted "
"into movement toward the target"
),
],
@@ -690,6 +708,7 @@ def _build_survival_messages(
"own_state": {
"name": npc.name,
"role": npc.role,
+ "position": {"x": npc.position.x, "z": npc.position.z},
"health": round(npc.health),
"hunger": round(npc.hunger),
"fear": round(npc.fear),
@@ -704,9 +723,12 @@ def _build_survival_messages(
"herbs": npc.inventory_herbs,
"wood": npc.inventory_wood,
"weapon": npc.inventory_weapon,
+ "coins": npc.inventory_coins,
},
},
"goal": goal,
+ "country": _country_payload(world, npc),
+ "world_countries": _world_countries_payload(world, npc),
"active_directive": active_directive_payload(npc, tick=next_tick),
"nearby_threats": _survival_nearby_threats(world, npc),
"nearby_allies": _survival_nearby_allies(world, npc),
@@ -724,6 +746,7 @@ def _build_survival_messages(
"content": (
f"You are {npc.name}, a {npc.role}. Call exactly one tool for the next tick. "
"Use only allowed_actions and visible ids from the payload. Do not invent ids. "
+ "Your role is context, not a tool restriction. "
"Do not write prose. The engine validates and applies effects."
),
},
@@ -738,19 +761,19 @@ def _role_rules_for_prompt(role: str) -> list[str]:
if role == "guard":
return [
"Protect living villagers and houses from live threats.",
- "Attack only visible beasts or hostile live NPCs.",
- "Patrol when no threat is reachable.",
- "Never gather resources.",
+ "You tend to attack visible beasts or hostile live NPCs.",
+ "When no threat is reachable, you tend to move through patrol routes.",
+ "You can still use, speak, move or transfer when the situation calls for it.",
]
if role == "builder":
return [
- "Build or gather wood when work is safe.",
- "Go home or flee from live threats.",
+ "You tend to build, repair, and gather wood when work is safe.",
+ "You tend to seek shelter or flee from live threats.",
"Share or ask for supplies only with live nearby allies.",
- "Never attack.",
+ "You can still use the same base tools as everyone else.",
]
return [
- "Gather reachable food, herbs, wood, or weapons when safe.",
+ "You tend to gather reachable food, herbs, wood, or weapons when safe.",
"Eat or heal before routine work when urgent.",
"Flee or call for help from live threats.",
"Transfer supplies only to live nearby allies.",
@@ -808,6 +831,7 @@ def _survival_nearby_allies(world: WorldState, npc: Npc) -> list[dict[str, Any]]
"id": other.id,
"name": other.name,
"role": other.role,
+ "country_id": other.country_id,
"distance": round(distance, 2),
"trust": round(npc.relationships.get(other.id, 0.0), 2),
"hp": round(other.health),
@@ -817,12 +841,81 @@ def _survival_nearby_allies(world: WorldState, npc: Npc) -> list[dict[str, Any]]
"herbs": other.inventory_herbs,
"wood": other.inventory_wood,
"weapon": other.inventory_weapon,
+ "coins": other.inventory_coins,
},
}
)
return sorted(allies, key=lambda item: (item["distance"], item["id"]))[:5]
+def _country_payload(world: WorldState, npc: Npc) -> dict[str, Any] | None:
+ if npc.country_id is None:
+ return None
+ country = next((item for item in world.countries if item.id == npc.country_id), None)
+ if country is None:
+ return None
+ election = active_election_for_npc(world, npc)
+ return {
+ "id": country.id,
+ "name": country.name,
+ "badge": country.badge,
+ "ruler_id": country.ruler_id,
+ "policy": country.policy,
+ "treasury": dict(country.treasury.resources) if country.treasury else {},
+ "treasury_id": country.treasury.id if country.treasury else None,
+ "treasury_position": None
+ if country.treasury is None
+ else {"x": country.treasury.position.x, "z": country.treasury.position.z},
+ "cannon": None
+ if country.cannon is None
+ else {
+ "id": country.cannon.id,
+ "position": {"x": country.cannon.position.x, "z": country.cannon.position.z},
+ "operator_id": country.cannon.operator_id,
+ "cooldown_until_tick": country.cannon.cooldown_until_tick,
+ },
+ "next_election_tick": country.next_election_tick,
+ "election": None
+ if election is None
+ else {
+ "start_tick": election.start_tick,
+ "end_tick": election.end_tick,
+ "candidate_ids": list(election.candidate_ids),
+ "already_voted": npc.id in election.votes,
+ },
+ }
+
+
+def _world_countries_payload(world: WorldState, npc: Npc) -> list[dict[str, Any]]:
+ """Public info on every country so an NPC knows the rival, its ruler/policy,
+ and where the rival treasury/cannon is (needed to steal or strike)."""
+ overview: list[dict[str, Any]] = []
+ for country in world.countries:
+ ruler = next((c for c in world.npcs if c.id == country.ruler_id), None)
+ treasury_pos = (
+ None
+ if country.treasury is None
+ else {"x": country.treasury.position.x, "z": country.treasury.position.z}
+ )
+ overview.append(
+ {
+ "id": country.id,
+ "name": country.name,
+ "is_own": country.id == npc.country_id,
+ "ruler_id": country.ruler_id,
+ "ruler_name": ruler.name if ruler else None,
+ "policy": country.policy,
+ "population": len(country.citizen_ids),
+ "treasury_id": country.treasury.id if country.treasury else None,
+ "treasury_position": treasury_pos,
+ # The rival's stored resources are public knowledge worth stealing.
+ "treasury": dict(country.treasury.resources) if country.treasury else {},
+ "cannon_id": country.cannon.id if country.cannon else None,
+ }
+ )
+ return overview
+
+
def _survival_reachable_resources(world: WorldState, npc: Npc) -> list[dict[str, Any]]:
resources: list[dict[str, Any]] = []
for node in world.resource_nodes:
@@ -929,6 +1022,20 @@ def _npc_context(
}
+def _stamp_model_attribution(
+ entry: dict[str, Any],
+ npc: Npc | None,
+ model: str | None,
+) -> None:
+ """Ensure a ledger entry says which model/profile is driving the NPC."""
+ if npc is None:
+ return
+ entry.setdefault("model", model or "")
+ entry.setdefault("model_profile", npc.model_profile_id or npc.connector_id or "default")
+ entry.setdefault("connector_id", npc.connector_id)
+ entry.setdefault("country_id", npc.country_id)
+
+
def _npc_request_entry(
*,
npc: Npc,
@@ -936,10 +1043,16 @@ def _npc_request_entry(
model: str,
allowed_actions: list[str],
) -> dict[str, Any]:
+ # `model_profile`/`connector_id` make it explicit, per NPC, which model is
+ # driving it (e.g. the Qwen country connector vs the default model). This is
+ # what the admin log surfaces so a viewer can see "npc-007 -> qwen".
return {
"phase": "npc_request",
"npc_id": npc.id,
"model": model,
+ "model_profile": npc.model_profile_id or npc.connector_id or "default",
+ "connector_id": npc.connector_id,
+ "country_id": npc.country_id,
"allowed_actions": allowed_actions,
"messages": request.get("messages"),
"request": request,
@@ -1029,6 +1142,8 @@ def _parse_tool_calls(
if not isinstance(function, dict):
continue
tool_name = function.get("name")
+ if tool_name == "strike":
+ tool_name = "attack"
if tool_name not in TOOL_DEFINITIONS:
continue
try:
@@ -1080,8 +1195,8 @@ def _parse_text_content(
payload: dict[str, Any] = {"npc_id": resolved_npc_id, "action": action}
if len(tokens) > 1 and action in {"use", "move"}:
payload["resource_id"] = tokens[1]
- if len(tokens) > 1 and action in {"transfer", "speak", "strike"}:
- payload["target_id"] = tokens[1]
+ if len(tokens) > 1 and action in {"transfer", "speak", "attack", "strike"}:
+ payload["target_id"] = tokens[1]
return _actions_to_plan([payload], world, source=source)
@@ -1104,7 +1219,12 @@ def _actions_to_plan(actions: Sequence[object], world: WorldState, *, source: st
continue
target = _target_from_action(action, half_width=half_width, half_depth=half_depth)
- raw_target_id = action.get("target_id") or action.get("target_npc_id")
+ raw_target_id = (
+ action.get("target_id")
+ or action.get("target_npc_id")
+ or action.get("target_entity_id")
+ or action.get("candidate_id")
+ )
target_npc_id = _optional_known_npc_id(raw_target_id, world)
target_entity_id = (
None if target_npc_id else _optional_known_entity_id(raw_target_id, world)
@@ -1115,6 +1235,8 @@ def _actions_to_plan(actions: Sequence[object], world: WorldState, *, source: st
resource_type = _optional_resource_type(
action.get("resource_type") or action.get("item")
)
+ use_type = _optional_short_text(action.get("use_type"), limit=40)
+ params = action.get("params") if isinstance(action.get("params"), dict) else None
amount = _optional_amount(action.get("amount"))
communication_intent = _optional_communication_intent(action.get("intent"))
message = _optional_short_text(action.get("message"), limit=140)
@@ -1132,6 +1254,8 @@ def _actions_to_plan(actions: Sequence[object], world: WorldState, *, source: st
target_entity_id=target_entity_id,
resource_id=resource_id,
resource_type=resource_type,
+ use_type=use_type,
+ params=params,
amount=amount,
communication_intent=communication_intent,
message=message,
@@ -1146,8 +1270,10 @@ def _actions_to_plan(actions: Sequence[object], world: WorldState, *, source: st
def _action_kind(raw: object) -> ActionKind | None:
- if raw in ("move", "speak", "strike", "use", "transfer", "idle"):
- return raw
+ if raw in ("move", "speak", "attack", "use", "transfer", "vote", "idle"):
+ return cast(ActionKind, raw)
+ if raw == "strike":
+ return "attack"
if raw in (
# Survival actions: resolved by the survival engine in survival worlds;
# the autopilot replaces them with a primitive in social worlds.
@@ -1186,8 +1312,17 @@ def _optional_known_npc_id(raw: object, world: WorldState) -> str | None:
def _optional_known_entity_id(raw: object, world: WorldState) -> str | None:
if not isinstance(raw, str):
return None
- beast_ids = {beast.id for beast in world.beasts}
- if raw in beast_ids:
+ if raw in {beast.id for beast in world.beasts}:
+ return raw
+ # Treasuries and cannons are engine objects an NPC can target with
+ # transfer/use (deposit, steal, fire). Without these the engine would null
+ # the target and silently degrade a treasury deposit into an NPC transfer.
+ for country in world.countries:
+ if country.treasury is not None and raw == country.treasury.id:
+ return raw
+ if country.cannon is not None and raw == country.cannon.id:
+ return raw
+ if raw in {house.id for house in world.houses}:
return raw
return None
@@ -1201,7 +1336,7 @@ def _optional_resource_id(raw: object, world: WorldState) -> str | None:
def _optional_resource_type(raw: object) -> str | None:
- if raw in {"food", "herbs", "wood", "weapon"}:
+ if raw in {"food", "herbs", "wood", "weapon", "coins"}:
return raw
return None
@@ -1209,7 +1344,7 @@ def _optional_resource_type(raw: object) -> str | None:
def _optional_amount(raw: object) -> int | None:
if isinstance(raw, bool) or not isinstance(raw, int):
return None
- return max(1, min(5, raw))
+ return max(1, min(20, raw))
def _optional_communication_intent(raw: object) -> str | None:
@@ -1256,7 +1391,10 @@ def _target_from_action(
half_width: float,
half_depth: float,
) -> Vec3 | None:
- if "target_x" in action or "target_z" in action:
+ if "x" in action or "z" in action:
+ x = action.get("x")
+ z = action.get("z")
+ elif "target_x" in action or "target_z" in action:
x = action.get("target_x")
z = action.get("target_z")
else:
diff --git a/src/god_simulator/simulation/connectors/routing.py b/src/god_simulator/simulation/connectors/routing.py
index 5d81a1e618ee15f293c410e4c382547882cd80aa..3aee0c39744cbdd508d19eb2ebbf801f1699883a 100644
--- a/src/god_simulator/simulation/connectors/routing.py
+++ b/src/god_simulator/simulation/connectors/routing.py
@@ -1,6 +1,7 @@
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Any
from god_simulator.domain import WorldState
from god_simulator.simulation.connectors.base import NpcDirective, TickPlan, WorldSimulator
@@ -28,6 +29,11 @@ class RoutingWorldSimulator:
routed_ids = {npc.id for npc in world.npcs if npc.connector_id in self._routes}
all_directives: list[NpcDirective] = []
+ # Ledger entries (npc_request/npc_response/npc_fallback) must survive the
+ # merge: the admin log and the goal's "every LLM call is recorded"
+ # requirement depend on them. They are filtered by the same ownership
+ # rule as the directives so each NPC is logged exactly once.
+ all_ledger_entries: list[dict[str, Any]] = []
source_parts: list[str] = []
# Run all route simulators and the default simulator in parallel.
@@ -44,12 +50,24 @@ class RoutingWorldSimulator:
for connector_id, sim in self._routes.items():
plan = results[connector_id]
filtered = [d for d in plan.directives if npc_connector.get(d.npc_id) == connector_id]
+ owned_entries = [
+ entry
+ for entry in plan.ledger_entries
+ if npc_connector.get(str(entry.get("npc_id"))) == connector_id
+ ]
+ all_ledger_entries.extend(owned_entries)
if filtered:
all_directives.extend(filtered)
source_parts.append(plan.source)
default_plan = results["__default__"]
filtered_default = [d for d in default_plan.directives if d.npc_id not in routed_ids]
+ default_entries = [
+ entry
+ for entry in default_plan.ledger_entries
+ if entry.get("npc_id") is None or entry.get("npc_id") not in routed_ids
+ ]
+ all_ledger_entries.extend(default_entries)
if filtered_default:
all_directives.extend(filtered_default)
source_parts.append(default_plan.source)
@@ -62,4 +80,4 @@ class RoutingWorldSimulator:
unique.append(d)
source = "+".join(dict.fromkeys(source_parts)) or self.name
- return TickPlan(source=source, directives=unique)
+ return TickPlan(source=source, directives=unique, ledger_entries=all_ledger_entries)
diff --git a/src/god_simulator/simulation/spawning.py b/src/god_simulator/simulation/spawning.py
index 4e5cb7fc3e6a237bae8435d717330324d6e1a0ea..d54ed8df1b5a171bf6d4de98c500c506ae592245 100644
--- a/src/god_simulator/simulation/spawning.py
+++ b/src/god_simulator/simulation/spawning.py
@@ -5,11 +5,13 @@ import random
from god_simulator.config import GameConfig
from god_simulator.domain import (
Beast,
+ CountryState,
House,
MemoryEntry,
Npc,
ResourceNode,
Terrain,
+ TreasuryState,
Vec3,
WorldState,
)
@@ -42,12 +44,6 @@ _NAMES = [
_ROLES = ["gatherer", "gatherer", "gatherer", "guard", "guard", "builder"]
_QWEN_NAMES = ["Kira", "Lena", "Mira", "Niko", "Oleg", "Pavel"]
-_BARB_NAMES = [
- "Borg", "Drak", "Hrag", "Krull", "Morg", "Ravn",
- "Skur", "Thax", "Vrex", "Wulf", "Xorn", "Yrag",
- "Zark", "Brox", "Crag", "Dusk",
-]
-
def create_world(config: GameConfig) -> WorldState:
rng = random.Random(config.world.seed)
@@ -56,19 +52,21 @@ def create_world(config: GameConfig) -> WorldState:
width=config.world.width,
depth=config.world.depth,
)
- village_npcs = [_spawn_npc(index=index, config=config, rng=rng) for index in range(config.npcs.count)]
- qwen_npcs = _spawn_qwen_npcs(config) if config.world.survival else []
- barbarians = _spawn_barbarians(config) if config.world.survival else []
+ village_npcs = [
+ _spawn_npc(index=index, config=config, rng=rng, country_id="nemotron")
+ for index in range(config.npcs.count)
+ ]
+ qwen_npcs = _spawn_qwen_npcs(config, count=config.npcs.count) if config.world.survival else []
houses = _spawn_houses(config)
if config.world.survival:
_seed_survival_relationships(village_npcs + qwen_npcs)
_assign_starting_homes(village_npcs + qwen_npcs, houses)
- _initialize_importance(village_npcs + qwen_npcs + barbarians)
+ _initialize_importance(village_npcs + qwen_npcs)
- all_npcs = village_npcs + qwen_npcs + barbarians
+ all_npcs = village_npcs + qwen_npcs
population = sum(1 for npc in all_npcs if npc.health > 0)
- return WorldState(
+ world = WorldState(
tick=0,
seed=config.world.seed,
terrain=terrain,
@@ -76,14 +74,18 @@ def create_world(config: GameConfig) -> WorldState:
resource_nodes=_spawn_resource_nodes(config),
beasts=_spawn_beasts(config),
houses=houses,
+ countries=_spawn_countries(config, all_npcs),
population=population,
+ population_cap=max(24, population + 8),
peak_population=population,
- overseer_mode=config.overseer.mode,
- overseer_cycle_ticks=config.overseer.cycle_ticks,
+ overseer_mode="off",
+ overseer_cycle_ticks=0,
)
+ _assign_initial_rulers(world)
+ return world
-def _spawn_npc(*, index: int, config: GameConfig, rng: random.Random) -> Npc:
+def _spawn_npc(*, index: int, config: GameConfig, rng: random.Random, country_id: str) -> Npc:
half_width = config.world.width / 2
half_depth = config.world.depth / 2
@@ -112,6 +114,7 @@ def _spawn_npc(*, index: int, config: GameConfig, rng: random.Random) -> Npc:
name=name,
role=role,
position=Vec3(x=x, y=0.0, z=z),
+ country_id=country_id,
attack_damage=rng.randint(5, 15),
memory=[MemoryEntry(tick=0, text=f"{name} arrived as a {role}.")],
age=60 if config.world.survival else 0,
@@ -194,16 +197,26 @@ def _spawn_houses(config: GameConfig) -> list[House]:
if not config.world.survival:
return []
village_x = -(config.world.width / 2 * 2 / 3)
+ qwen_x = config.world.width / 2 * 2 / 3
return [
House(
- id="house_001",
+ id="house_nemotron_001",
position=Vec3(x=village_x, y=0.0, z=0.0),
hp=60.0,
max_hp=60.0,
state="completed",
build_progress=10,
capacity=3,
- )
+ ),
+ House(
+ id="house_qwen_001",
+ position=Vec3(x=qwen_x, y=0.0, z=0.0),
+ hp=60.0,
+ max_hp=60.0,
+ state="completed",
+ build_progress=10,
+ capacity=3,
+ ),
]
@@ -218,10 +231,12 @@ def _seed_survival_relationships(npcs: list[Npc]) -> None:
def _assign_starting_homes(npcs: list[Npc], houses: list[House]) -> None:
if not houses:
return
- home = houses[0]
for npc in npcs:
+ home = min(houses, key=lambda house: abs(house.position.x - npc.position.x))
npc.home_house_id = home.id
- home.occupant_ids = [npc.id for npc in npcs[: home.capacity]]
+ for house in houses:
+ citizens = [npc.id for npc in npcs if npc.home_house_id == house.id]
+ house.occupant_ids = citizens[: house.capacity]
def _initialize_importance(npcs: list[Npc]) -> None:
@@ -233,10 +248,9 @@ def _initialize_importance(npcs: list[Npc]) -> None:
npc.importance = round(1.0 + 2.0 * (1 / role_count), 3)
-def _spawn_qwen_npcs(config: GameConfig) -> list[Npc]:
+def _spawn_qwen_npcs(config: GameConfig, *, count: int) -> list[Npc]:
half_width = config.world.width / 2
- village_x = -(half_width * 2 / 3)
- # Cluster around the house (which sits at village_x, z=0)
+ qwen_x = half_width * 2 / 3
offsets = [
(0.0, 0.0),
(BLOCK_SIZE, 0.0),
@@ -247,19 +261,21 @@ def _spawn_qwen_npcs(config: GameConfig) -> list[Npc]:
]
npcs = []
inv_rng = random.Random(f"{config.world.seed}:qwen_inventory")
- for i, (dx, dz) in enumerate(offsets):
- name = _QWEN_NAMES[i]
+ for i in range(count):
+ dx, dz = offsets[i % len(offsets)]
+ name = _QWEN_NAMES[i % len(_QWEN_NAMES)]
npc = Npc(
id=f"qwen-{i + 1:03d}",
name=name,
role="gatherer",
- position=Vec3(x=round(village_x + dx, 3), y=0.0, z=round(dz, 3)),
+ position=Vec3(x=round(qwen_x + dx, 3), y=0.0, z=round(dz, 3)),
attack_damage=inv_rng.randint(5, 12),
- memory=[MemoryEntry(tick=0, text=f"{name} joined the village as a free spirit.")],
+ memory=[MemoryEntry(tick=0, text=f"{name} joined Qwen as a free spirit.")],
age=25,
max_age=550,
unrestricted_actions=True,
connector_id="qwen",
+ country_id="qwen",
personality="curious and resourceful",
hunger=15.0,
safety=60.0,
@@ -270,36 +286,51 @@ def _spawn_qwen_npcs(config: GameConfig) -> list[Npc]:
return npcs
-def _spawn_barbarians(config: GameConfig) -> list[Npc]:
+def _spawn_countries(config: GameConfig, npcs: list[Npc]) -> list[CountryState]:
+ if not config.world.survival:
+ return []
half_width = config.world.width / 2
- half_depth = config.world.depth / 2
- barb_x = half_width * 2 / 3
- npcs = []
- inv_rng = random.Random(f"{config.world.seed}:barb_inventory")
- for i in range(16):
- col = i % 4
- row = i // 4
- dx = (col - 1.5) * BLOCK_SIZE
- dz = (row - 1.5) * (half_depth / 2)
- name = _BARB_NAMES[i]
- npc = Npc(
- id=f"barb-{i + 1:03d}",
- name=name,
- role="guard",
- position=Vec3(x=round(barb_x + dx, 3), y=0.0, z=round(dz, 3)),
- attack_damage=inv_rng.randint(12, 20),
- memory=[MemoryEntry(tick=0, text=f"{name} is a barbarian warrior on the warpath.")],
- age=30,
- max_age=350,
- connector_id=None,
- personality="aggressive barbarian",
- hunger=35.0,
- safety=0.0,
- inventory_food=inv_rng.randint(0, 1),
- inventory_weapon=1,
- )
- npcs.append(npc)
- return npcs
+ nemotron_x = -(half_width * 2 / 3)
+ qwen_x = half_width * 2 / 3
+ citizens_by_country = {
+ "nemotron": [npc.id for npc in npcs if npc.country_id == "nemotron"],
+ "qwen": [npc.id for npc in npcs if npc.country_id == "qwen"],
+ }
+ return [
+ CountryState(
+ id="nemotron",
+ name="Nemotron",
+ color="#4f8cff",
+ badge="N",
+ citizen_ids=citizens_by_country["nemotron"],
+ treasury=TreasuryState(
+ id="treasury_nemotron",
+ position=Vec3(x=nemotron_x, y=0.0, z=BLOCK_SIZE * 1.5),
+ resources={"food": 2, "herbs": 1, "wood": 20, "coins": 25},
+ ),
+ ),
+ CountryState(
+ id="qwen",
+ name="Qwen",
+ color="#d86bff",
+ badge="Q",
+ citizen_ids=citizens_by_country["qwen"],
+ treasury=TreasuryState(
+ id="treasury_qwen",
+ position=Vec3(x=qwen_x, y=0.0, z=BLOCK_SIZE * 1.5),
+ resources={"food": 2, "herbs": 1, "wood": 20, "coins": 25},
+ ),
+ ),
+ ]
+
+
+def _assign_initial_rulers(world: WorldState) -> None:
+ for country in world.countries:
+ ruler = next((npc for npc in world.living_npcs() if npc.id in country.citizen_ids), None)
+ if ruler is None:
+ continue
+ country.ruler_id = ruler.id
+ ruler.special_status = "ruler"
def _random_max_age(seed: int, index: int) -> int:
diff --git a/src/god_simulator/simulation/survival.py b/src/god_simulator/simulation/survival.py
index 30296dd31c9c069a225097d6176a7079a7e5561d..719351f6c663d1b52a1337b2adfd3b38d2035f9c 100644
--- a/src/god_simulator/simulation/survival.py
+++ b/src/god_simulator/simulation/survival.py
@@ -12,15 +12,20 @@ thresholds are scaled by ``TILE`` below.
from __future__ import annotations
import math
+import os
import random
from god_simulator.domain import (
Beast,
+ CannonState,
+ CountryState,
+ ElectionState,
House,
MemoryEntry,
MemoryEpisode,
Npc,
ResourceNode,
+ TreasuryState,
Vec3,
WorldEvent,
WorldLogEvent,
@@ -39,9 +44,7 @@ from god_simulator.simulation.mechanics import (
)
from god_simulator.simulation.memory import remember
from god_simulator.simulation.roles import (
- action_allowed_for_role,
canonical_action,
- filter_actions_for_role,
normalize_role,
)
@@ -114,6 +117,16 @@ SPREAD_NODE_CHOICES = 3
# Minimum surplus food before an NPC is willing to share with a needier neighbor.
SHARE_FOOD_SURPLUS = 2
+DEFAULT_ELECTION_INTERVAL_TICKS = 50
+DEFAULT_ELECTION_DURATION_TICKS = 10
+DEFAULT_CANNON_MIN_POPULATION = 6
+DEFAULT_CANNON_TREASURY_COST = 20
+DEFAULT_CANNON_WOOD_COST = 15
+DEFAULT_CANNON_HP_TAX_PER_CITIZEN = 5
+DEFAULT_CANNON_DAMAGE = 35
+DEFAULT_CANNON_RADIUS = 5
+DEFAULT_CANNON_COOLDOWN_TICKS = 20
+
# Goal -> allowed survival effect verbs (engine-side: the deterministic role
# planner chooses from these sets, and role whitelists filter them per NPC).
GOAL_ACTIONS: dict[str, list[str]] = {
@@ -132,27 +145,26 @@ GOAL_ACTIONS: dict[str, list[str]] = {
"dead": [],
}
-# Goal -> ranked LLM-facing suggestions, best first: the six primitives plus the
-# role-gated survival extras (go_home, patrol, build). These are hints; the
-# engine role-gates and repairs every request at validation time.
+# Goal -> ranked LLM-facing suggestions, best first. These are public tools;
+# internal survival verbs such as gather/build/go_home map through use/move.
GOAL_SUGGESTED_ACTIONS: dict[str, list[str]] = {
- "engage_threat": ["strike", "patrol", "speak", "move"],
- "survive_threat_fight": ["strike", "go_home", "speak", "move", "idle"],
- "survive_threat_flee": ["move", "go_home", "speak", "idle"],
- "help_ally": ["strike", "idle", "speak"],
+ "engage_threat": ["attack", "move", "speak"],
+ "survive_threat_fight": ["attack", "move", "speak"],
+ "survive_threat_flee": ["move", "speak"],
+ "help_ally": ["attack", "speak"],
"heal_self": ["use"],
"eat_food": ["use"],
"find_food": ["move", "use", "speak", "transfer"],
"find_herbs": ["move", "use", "speak"],
- "obey_directive": ["speak", "move", "strike", "go_home", "build", "patrol", "idle"],
- "settle": ["go_home", "use", "speak", "transfer"],
- "work": ["use", "move", "build", "patrol", "transfer", "speak"],
- "routine_life": ["use", "move", "transfer", "speak", "go_home", "build", "patrol"],
+ "obey_directive": ["speak", "move", "attack", "use", "transfer"],
+ "settle": ["move", "use", "speak", "transfer"],
+ "work": ["use", "move", "transfer", "speak"],
+ "routine_life": ["use", "move", "transfer", "speak"],
"dead": [],
}
PRIMITIVE_SURVIVAL_ACTIONS = frozenset(
- {"move", "speak", "strike", "use", "transfer", "idle"}
+ {"move", "speak", "attack", "use", "transfer", "vote"}
)
# Internal effect verbs the resolver knows how to execute. Primitives are
@@ -177,6 +189,10 @@ VALID_SURVIVAL_VERBS = frozenset(
"go_home",
"patrol",
"build",
+ "craft",
+ "assign_cannon_operator",
+ "fire",
+ "vote",
}
)
@@ -353,6 +369,42 @@ def build_relationships_summary(npc: Npc, world: WorldState | None = None) -> st
def build_perception(npc: Npc, world: WorldState) -> str:
lines: list[str] = []
+ country = country_for_npc(world, npc)
+ if country is not None:
+ ruler = _npc_by_id(world, country.ruler_id or "")
+ treasury = country.treasury.resources if country.treasury is not None else {}
+ lines.append(
+ f"- your country is {country.name} [{country.badge}], ruler="
+ f"{ruler.name if ruler else country.ruler_id or 'none'}"
+ )
+ lines.append(f"- country policy: {country.policy}")
+ lines.append(
+ "- treasury "
+ + " ".join(f"{key}={value}" for key, value in sorted(treasury.items()))
+ )
+ if country.cannon is not None:
+ operator = _npc_by_id(world, country.cannon.operator_id or "")
+ lines.append(
+ f"- cannon {country.cannon.id} operator="
+ f"{operator.name if operator else country.cannon.operator_id or 'none'}"
+ )
+ election = active_election_for_country(world, country.id)
+ if election is not None:
+ lines.append(
+ f"- election active until tick {election.end_tick}; candidates: "
+ + ", ".join(election.candidate_ids)
+ )
+ for rival in world.countries:
+ if rival.id == country.id:
+ continue
+ rival_ruler = _npc_by_id(world, rival.ruler_id or "")
+ rival_treasury = rival.treasury.resources if rival.treasury is not None else {}
+ lines.append(
+ f"- rival country {rival.name} [{rival.badge}] ruler="
+ f"{rival_ruler.name if rival_ruler else rival.ruler_id or 'none'}; "
+ f"treasury {rival.treasury.id if rival.treasury else 'none'} "
+ + " ".join(f"{key}={value}" for key, value in sorted(rival_treasury.items()))
+ )
for beast in sorted(world.beasts, key=lambda item: vec_distance(npc.position, item.position)):
if beast.state == "dead":
@@ -385,7 +437,8 @@ def build_perception(npc: Npc, world: WorldState) -> str:
for other in sorted(npcs_in_radius(world, npc.position, PERCEPTION_NPC_RADIUS, exclude_id=npc.id), key=lambda item: item.id):
trust = npc.relationships.get(other.id, 0.0)
label = _relationship_label(trust)
- inventory = [f"{label} trust={trust:.1f}"]
+ country_label = other.country_id or "no-country"
+ inventory = [f"{label} trust={trust:.1f}", f"country={country_label}"]
if other.inventory_food:
inventory.append(f"food={other.inventory_food}")
if other.inventory_weapon:
@@ -570,6 +623,44 @@ def nearest_resource_for_npc(
)
+def country_for_npc(world: WorldState, npc: Npc) -> CountryState | None:
+ if npc.country_id is None:
+ return None
+ return next((country for country in world.countries if country.id == npc.country_id), None)
+
+
+def active_election_for_country(
+ world: WorldState,
+ country_id: str,
+ *,
+ tick: int | None = None,
+) -> ElectionState | None:
+ current_tick = world.tick if tick is None else tick
+ for election in world.elections:
+ if election.country_id != country_id or election.completed:
+ continue
+ if election.start_tick <= current_tick <= election.end_tick:
+ return election
+ return None
+
+
+def active_election_for_npc(world: WorldState, npc: Npc, *, tick: int | None = None) -> ElectionState | None:
+ if npc.country_id is None:
+ return None
+ return active_election_for_country(world, npc.country_id, tick=tick)
+
+
+def _env_int(name: str, default: int) -> int:
+ raw = os.getenv(name)
+ if raw is None:
+ return default
+ try:
+ value = int(raw)
+ except ValueError:
+ return default
+ return value if value > 0 else default
+
+
def apply_world_systems(world: WorldState) -> None:
"""Run deterministic environment systems before NPC decisions."""
# Local import: chaos.py reuses survival helpers at module level.
@@ -823,15 +914,128 @@ def apply_survival_tick(world: WorldState) -> None:
def apply_post_survival_tick(world: WorldState, next_tick: int) -> None:
"""Run deterministic end-of-tick systems after NPC actions resolve."""
_refresh_house_occupants(world)
+ _refresh_country_citizens(world, next_tick)
+ _run_election_system(world, next_tick)
for npc in list(world.npcs):
if is_alive(npc):
_maybe_reproduce(world, npc, next_tick)
_refresh_house_occupants(world)
+ _refresh_country_citizens(world, next_tick)
if next_tick % 10 == 0:
_recompute_importance(world)
_update_population_and_status(world, next_tick)
+def _refresh_country_citizens(world: WorldState, next_tick: int) -> None:
+ living_ids = {npc.id for npc in world.living_npcs()}
+ for country in world.countries:
+ country.citizen_ids = [
+ npc.id
+ for npc in world.npcs
+ if npc.country_id == country.id and npc.id in living_ids
+ ]
+ if country.ruler_id not in living_ids or country.ruler_id not in country.citizen_ids:
+ country.ruler_id = country.citizen_ids[0] if country.citizen_ids else None
+ for npc in world.npcs:
+ if npc.country_id != country.id:
+ continue
+ npc.special_status = "ruler" if npc.id == country.ruler_id else (
+ "cannon_operator" if country.cannon and country.cannon.operator_id == npc.id else None
+ )
+
+def _run_election_system(world: WorldState, next_tick: int) -> None:
+ for country in world.countries:
+ active = active_election_for_country(world, country.id, tick=next_tick)
+ if active is None and country.citizen_ids and next_tick >= country.next_election_tick:
+ _start_election(world, country, next_tick)
+ active = active_election_for_country(world, country.id, tick=next_tick)
+ if active is not None and next_tick >= active.end_tick:
+ _finish_election(world, country, active, next_tick)
+
+
+def _start_election(world: WorldState, country: CountryState, tick: int) -> None:
+ duration = _env_int("ELECTION_DURATION_TICKS", DEFAULT_ELECTION_DURATION_TICKS)
+ candidate_ids = list(country.citizen_ids)
+ election = ElectionState(
+ country_id=country.id,
+ start_tick=tick,
+ end_tick=tick + duration,
+ candidate_ids=candidate_ids,
+ )
+ world.elections.append(election)
+ _log_event(
+ world,
+ "election_started",
+ f"{country.name} election started with {len(candidate_ids)} candidates",
+ severity="neutral",
+ object_id=country.id,
+ tick=tick,
+ )
+
+
+def _finish_election(
+ world: WorldState,
+ country: CountryState,
+ election: ElectionState,
+ tick: int,
+) -> None:
+ for npc_id in country.citizen_ids:
+ if npc_id in election.votes:
+ continue
+ election.votes[npc_id] = _fallback_vote(world, election, voter_id=npc_id)
+ counts: dict[str, int] = {}
+ for candidate_id in election.votes.values():
+ if candidate_id in election.candidate_ids:
+ counts[candidate_id] = counts.get(candidate_id, 0) + 1
+ if counts:
+ winner_id = min(counts, key=lambda candidate_id: (-counts[candidate_id], candidate_id))
+ country.ruler_id = winner_id
+ _set_country_policy(world, country, tick)
+ election.completed = True
+ country.next_election_tick = tick + _env_int(
+ "ELECTION_INTERVAL_TICKS",
+ DEFAULT_ELECTION_INTERVAL_TICKS,
+ )
+ _refresh_country_citizens(world, tick)
+ _log_event(
+ world,
+ "election_completed",
+ f"{country.name} elected {country.ruler_id} with {len(election.votes)} votes",
+ severity="good",
+ target_id=country.ruler_id,
+ object_id=country.id,
+ tick=tick,
+ )
+
+
+def _fallback_vote(world: WorldState, election: ElectionState, *, voter_id: str) -> str:
+ if not election.candidate_ids:
+ return voter_id
+ rng = random.Random(f"{world.seed}:{election.country_id}:{election.start_tick}:{voter_id}")
+ return rng.choice(election.candidate_ids)
+
+
+def _set_country_policy(world: WorldState, country: CountryState, tick: int) -> None:
+ treasury = country.treasury.resources if country.treasury else {}
+ wood = treasury.get("wood", 0)
+ coins = treasury.get("coins", 0)
+ if wood >= DEFAULT_CANNON_WOOD_COST and coins >= DEFAULT_CANNON_TREASURY_COST:
+ country.policy = "Craft a cannon, protect the treasury, and attack only clear threats."
+ elif coins < 8:
+ country.policy = "Gather coins and wood, defend citizens, and avoid reckless fights."
+ else:
+ country.policy = "Gather resources, help weak citizens, and watch the other country."
+ _log_event(
+ world,
+ "policy_updated",
+ f"{country.name} policy: {country.policy}",
+ severity="neutral",
+ actor_id=country.ruler_id,
+ object_id=country.id,
+ tick=tick,
+ )
+
+
def _apply_beast_system(world: WorldState) -> None:
half_width = world.terrain.width / 2
half_depth = world.terrain.depth / 2
@@ -1074,6 +1278,7 @@ def _drop_inventory(world: WorldState, npc: Npc) -> None:
("herbs", npc.inventory_herbs),
("wood", npc.inventory_wood),
("weapon", npc.inventory_weapon),
+ ("coins", npc.inventory_coins),
):
if amount <= 0:
continue
@@ -1084,13 +1289,17 @@ def _drop_inventory(world: WorldState, npc: Npc) -> None:
resource_type=resource_type,
position=npc.position,
amount=amount,
- max_amount=max(amount, {"food": 5, "herbs": 4, "wood": 6, "weapon": 2}[resource_type]),
+ max_amount=max(
+ amount,
+ {"food": 5, "herbs": 4, "wood": 6, "weapon": 2, "coins": 10}[resource_type],
+ ),
)
)
npc.inventory_food = 0
npc.inventory_herbs = 0
npc.inventory_wood = 0
npc.inventory_weapon = 0
+ npc.inventory_coins = 0
def _purge_dead_npc_references(world: WorldState, dead: Npc) -> None:
@@ -1328,7 +1537,7 @@ def _update_population_and_status(world: WorldState, next_tick: int) -> None:
return
if population <= 0:
world.game_status = "lose"
- elif population >= 12:
+ elif population >= world.population_cap:
world.game_status = "win_population"
elif next_tick >= 600:
world.game_status = "win_survival"
@@ -1413,7 +1622,7 @@ def allowed_actions_for_goal(goal: str) -> list[str]:
def allowed_actions_for_npc(world: WorldState, npc: Npc, goal: str) -> list[str]:
- allowed = filter_actions_for_role(allowed_actions_for_goal(goal), npc.role)
+ allowed = allowed_actions_for_goal(goal)
if goal == "survive_threat_flee":
allowed = [action for action in allowed if action != "transfer"]
if "gather" in allowed and not _has_gatherable_resource_for_goal(world, npc, goal):
@@ -1438,16 +1647,9 @@ def _has_gatherable_resource_for_goal(world: WorldState, npc: Npc, goal: str) ->
def suggested_actions_for_npc(npc: Npc, goal: str) -> list[str]:
- """Role-gated, ranked LLM suggestions: primitives plus survival extras."""
- role = normalize_role(npc.role)
+ """Ranked LLM suggestions. Roles influence planning context, not tool access."""
suggested = suggested_actions_for_goal(goal)
- if role != "builder":
- suggested = [action for action in suggested if action != "build"]
- if role != "guard":
- suggested = [action for action in suggested if action != "patrol"]
- if role == "builder":
- suggested = [action for action in suggested if action != "strike"]
- return suggested or ["idle"]
+ return suggested or ["move", "speak"]
def _beast_threatens_village(world: WorldState) -> bool:
@@ -1495,31 +1697,10 @@ def validate_survival_action(
if action not in VALID_SURVIVAL_VERBS:
return "idle"
resource_type = _resource_type_for_validation(world, npc, directive, action)
- self_defense = (
- action == "attack"
- and nearest_beast_distance(world, npc) <= NPC_MELEE_REACH
- and normalize_role(npc.role) == "gatherer"
- )
- # Unrestricted NPCs (e.g. external agents over MCP) are intentionally not
- # role-gated: they may request any action. Physical-possibility checks below
- # still apply, but role whitelists do not.
- if not npc.unrestricted_actions and not action_allowed_for_role(
- npc.role,
- action,
- resource_type=resource_type,
- self_defense=self_defense,
- ):
- return _role_fallback_action(world, npc)
if action == "gather":
node = _resource_from_directive(world, directive, npc, max_dist=GATHER_RADIUS)
if node is None or node.amount <= 0:
return "move_to_resource"
- if (
- not npc.unrestricted_actions
- and normalize_role(npc.role) == "builder"
- and node.resource_type != "wood"
- ):
- return "move_to_resource"
return "gather"
if action == "consume":
return "consume" if npc.inventory_food > 0 else "find_food"
@@ -1531,27 +1712,39 @@ def validate_survival_action(
return "defend"
return "attack" # apply_action_effects approaches when out of melee reach
if action == "build":
- if not npc.unrestricted_actions and normalize_role(npc.role) != "builder":
- return _role_fallback_action(world, npc)
if npc.build_target_house_id:
house = _house_by_id(world, npc.build_target_house_id)
if house is not None and house.state == "under_construction":
return "build"
return "build" if npc.inventory_wood >= 5 else "move_to_resource"
+ if action == "craft":
+ return "craft" if _can_craft_cannon(world, npc) else "idle"
+ if action == "assign_cannon_operator":
+ return "assign_cannon_operator" if _can_assign_cannon_operator(world, npc, directive) else "idle"
+ if action == "fire":
+ return "fire" if _can_fire_cannon(world, npc, directive) else "idle"
if action == "go_home":
return "go_home" if nearest_house(world, npc.position) is not None else "move_to"
if action == "patrol":
- return "patrol" if normalize_role(npc.role) == "guard" else _role_fallback_action(world, npc)
+ return "patrol"
if action == "steal":
partner = _target_npc_from_directive(world, npc, directive, max_dist=TRADE_RADIUS)
if partner is None:
partner = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS, require_food=True)
return "steal" if partner is not None else "move_to_resource"
if action == "transfer":
+ if _treasury_from_directive(world, directive) is not None:
+ return "transfer"
partner = _target_npc_from_directive(world, npc, directive, max_dist=TRADE_RADIUS)
if partner is None:
partner = nearest_alive_npc(world, npc, max_dist=TRADE_RADIUS)
return "transfer" if partner is not None else "communicate"
+ if action == "vote":
+ election = active_election_for_npc(world, npc)
+ candidate_id = directive.target_npc_id if directive is not None else None
+ if election is None or candidate_id not in election.candidate_ids:
+ return "idle"
+ return "vote"
return action
@@ -1580,10 +1773,25 @@ def _verb_for_action(
if action == "speak":
return "communicate"
- if action == "strike":
+ if action in {"attack", "strike"}:
return "attack"
if action == "use":
+ use_type = (directive.use_type or "").lower() if directive is not None else ""
+ if use_type in {"eat", "consume"}:
+ return "consume"
+ if use_type == "heal":
+ return "heal"
+ if use_type == "gather":
+ return "gather"
+ if use_type in {"build", "repair"}:
+ return "build"
+ if use_type == "craft":
+ return "craft"
+ if use_type == "assign_cannon_operator":
+ return "assign_cannon_operator"
+ if use_type == "fire":
+ return "fire"
if directive is not None and directive.resource_id:
return "gather"
resource_type = directive.resource_type if directive is not None else None
@@ -1599,10 +1807,18 @@ def _verb_for_action(
return "gather"
if action == "transfer":
+ # A transfer aimed at a treasury always resolves through the treasury
+ # handler, which honors `take` (steal from a rival) vs deposit. Routing
+ # take->NPC-steal here would bypass treasury raids entirely.
+ if _treasury_from_directive(world, directive) is not None:
+ return "transfer"
if directive is not None and directive.take:
return "steal"
return "transfer"
+ if action == "vote":
+ return "vote"
+
return "idle"
@@ -1704,8 +1920,6 @@ def apply_action_effects(
if action == "gather":
node = _resource_from_directive(world, directive, npc, max_dist=GATHER_RADIUS)
if node is not None and node.amount > 0:
- if normalize_role(npc.role) == "builder" and node.resource_type != "wood":
- return "searching for wood"
node.amount -= 1
_add_to_inventory(npc, node.resource_type)
add_episode(
@@ -1785,6 +1999,15 @@ def apply_action_effects(
if action == "build":
return _apply_build(npc, world)
+ if action == "craft":
+ return _apply_craft(npc, world, directive)
+
+ if action == "assign_cannon_operator":
+ return _apply_assign_cannon_operator(npc, world, directive)
+
+ if action == "fire":
+ return _apply_fire_cannon(npc, world, directive)
+
if action == "flee":
beast = _beast_from_directive(world, npc, directive)
if beast is not None:
@@ -1877,6 +2100,9 @@ def apply_action_effects(
return "charging the beast"
if action == "transfer":
+ treasury = _treasury_from_directive(world, directive)
+ if treasury is not None:
+ return _apply_treasury_transfer(npc, world, treasury, directive)
# Share food from whoever has a surplus with the neediest nearby neighbour,
# so food flows from haves to have-nots instead of people starving alone.
partner = _transfer_partner(world, npc, directive)
@@ -1985,9 +2211,31 @@ def apply_action_effects(
if action == "defend":
return "defending"
+ if action == "vote":
+ return _apply_vote(npc, world, directive)
+
return "idle"
+def _apply_vote(npc: Npc, world: WorldState, directive: NpcDirective | None) -> str:
+ election = active_election_for_npc(world, npc)
+ candidate_id = directive.target_npc_id if directive is not None else None
+ if election is None or candidate_id not in election.candidate_ids:
+ return "unable to vote"
+ election.votes[npc.id] = candidate_id
+ candidate = _npc_by_id(world, candidate_id)
+ candidate_name = candidate.name if candidate is not None else candidate_id
+ _log_event(
+ world,
+ "vote_cast",
+ f"{npc.name} voted for {candidate_name}",
+ actor_id=npc.id,
+ target_id=candidate_id,
+ object_id=election.country_id,
+ )
+ return f"voting for {candidate_name}"
+
+
def _apply_communicate(
npc: Npc,
world: WorldState,
@@ -2182,9 +2430,6 @@ def _overhear_speech(
def _apply_build(npc: Npc, world: WorldState) -> str:
- if not npc.unrestricted_actions and normalize_role(npc.role) != "builder":
- return "cannot build"
-
half_width = world.terrain.width / 2
half_depth = world.terrain.depth / 2
active_house = _house_by_id(world, npc.build_target_house_id or "")
@@ -2304,6 +2549,147 @@ def _apply_build(npc: Npc, world: WorldState) -> str:
return f"building {house.id}"
+def _can_craft_cannon(world: WorldState, npc: Npc) -> bool:
+ country = country_for_npc(world, npc)
+ if country is None or country.ruler_id != npc.id or country.cannon is not None:
+ return False
+ if len(country.citizen_ids) < _env_int("CANNON_MIN_POPULATION", DEFAULT_CANNON_MIN_POPULATION):
+ return False
+ treasury = country.treasury.resources if country.treasury else {}
+ return (
+ treasury.get("coins", 0) >= _env_int("CANNON_TREASURY_COST", DEFAULT_CANNON_TREASURY_COST)
+ and treasury.get("wood", 0) >= _env_int("CANNON_WOOD_COST", DEFAULT_CANNON_WOOD_COST)
+ )
+
+
+def _can_assign_cannon_operator(
+ world: WorldState,
+ npc: Npc,
+ directive: NpcDirective | None,
+) -> bool:
+ country = country_for_npc(world, npc)
+ target_id = _param_str(directive, "npc_id") or (directive.target_npc_id if directive else None)
+ target = _npc_by_id(world, target_id or "")
+ return (
+ country is not None
+ and country.ruler_id == npc.id
+ and country.cannon is not None
+ and target is not None
+ and target.country_id == country.id
+ )
+
+
+def _can_fire_cannon(world: WorldState, npc: Npc, directive: NpcDirective | None) -> bool:
+ country = country_for_npc(world, npc)
+ if country is None or country.cannon is None:
+ return False
+ return country.cannon.operator_id == npc.id and world.tick >= country.cannon.cooldown_until_tick
+
+
+def _apply_craft(npc: Npc, world: WorldState, directive: NpcDirective | None) -> str:
+ if _param_str(directive, "recipe_id") not in {None, "cannon"}:
+ return "unknown recipe"
+ if not _can_craft_cannon(world, npc):
+ return "cannot craft cannon yet"
+ country = country_for_npc(world, npc)
+ assert country is not None and country.treasury is not None
+ treasury = country.treasury.resources
+ coin_cost = _env_int("CANNON_TREASURY_COST", DEFAULT_CANNON_TREASURY_COST)
+ wood_cost = _env_int("CANNON_WOOD_COST", DEFAULT_CANNON_WOOD_COST)
+ hp_tax = _env_int("CANNON_HP_TAX_PER_CITIZEN", DEFAULT_CANNON_HP_TAX_PER_CITIZEN)
+ treasury["coins"] = treasury.get("coins", 0) - coin_cost
+ treasury["wood"] = treasury.get("wood", 0) - wood_cost
+ for citizen in world.living_npcs():
+ if citizen.country_id == country.id:
+ citizen.health = max(1, citizen.health - hp_tax)
+ country.cannon = CannonState(
+ id=f"cannon_{country.id}",
+ position=country.treasury.position,
+ damage=_env_int("CANNON_DAMAGE", DEFAULT_CANNON_DAMAGE),
+ radius=_env_int("CANNON_RADIUS", DEFAULT_CANNON_RADIUS),
+ )
+ _log_event(
+ world,
+ "cannon_crafted",
+ f"{country.name} crafted {country.cannon.id}",
+ severity="good",
+ actor_id=npc.id,
+ object_id=country.cannon.id,
+ )
+ return f"crafting {country.cannon.id}"
+
+
+def _apply_assign_cannon_operator(
+ npc: Npc,
+ world: WorldState,
+ directive: NpcDirective | None,
+) -> str:
+ if not _can_assign_cannon_operator(world, npc, directive):
+ return "cannot assign cannon operator"
+ country = country_for_npc(world, npc)
+ assert country is not None and country.cannon is not None
+ target_id = _param_str(directive, "npc_id") or directive.target_npc_id
+ country.cannon.operator_id = target_id
+ _refresh_country_citizens(world, world.tick)
+ _log_event(
+ world,
+ "cannon_operator_assigned",
+ f"{country.name} assigned {target_id} to {country.cannon.id}",
+ severity="good",
+ actor_id=npc.id,
+ target_id=target_id,
+ object_id=country.cannon.id,
+ )
+ return f"assigning cannon operator {target_id}"
+
+
+def _apply_fire_cannon(npc: Npc, world: WorldState, directive: NpcDirective | None) -> str:
+ if not _can_fire_cannon(world, npc, directive):
+ return "cannot fire cannon"
+ country = country_for_npc(world, npc)
+ assert country is not None and country.cannon is not None
+ x = _param_float(directive, "x")
+ z = _param_float(directive, "z")
+ if x is None or z is None:
+ return "missing cannon target"
+ target = Vec3(x=x, y=0.0, z=z)
+ damaged: list[str] = []
+ for target_npc in list(world.living_npcs()):
+ if vec_distance(target_npc.position, target) > country.cannon.radius:
+ continue
+ target_npc.health -= country.cannon.damage
+ damaged.append(target_npc.id)
+ if target_npc.health <= 0:
+ _kill_npc(world, target_npc, cause="cannon", actor_id=npc.id)
+ cooldown = _env_int("CANNON_COOLDOWN_TICKS", DEFAULT_CANNON_COOLDOWN_TICKS)
+ country.cannon.cooldown_until_tick = world.tick + cooldown
+ _log_event(
+ world,
+ "cannon_fired",
+ f"{npc.name} fired {country.cannon.id} at ({x:g}, {z:g}) hitting {len(damaged)} targets",
+ severity="danger" if damaged else "warning",
+ actor_id=npc.id,
+ object_id=country.cannon.id,
+ )
+ return f"firing cannon at {x:g}, {z:g}"
+
+
+def _param_str(directive: NpcDirective | None, key: str) -> str | None:
+ if directive is None or not isinstance(directive.params, dict):
+ return None
+ value = directive.params.get(key)
+ return value if isinstance(value, str) else None
+
+
+def _param_float(directive: NpcDirective | None, key: str) -> float | None:
+ if directive is None or not isinstance(directive.params, dict):
+ return None
+ value = directive.params.get(key)
+ if isinstance(value, int | float) and not isinstance(value, bool):
+ return float(value)
+ return None
+
+
def _patrol_target(world: WorldState, npc: Npc) -> Vec3 | None:
threat = _nearest_village_threat(world, npc)
if threat is not None:
@@ -2465,15 +2851,87 @@ def _transfer_partner(world: WorldState, npc: Npc, directive: NpcDirective | Non
return _neediest_food_neighbor(world, npc)
+def _treasury_from_directive(
+ world: WorldState,
+ directive: NpcDirective | None,
+) -> tuple[CountryState, TreasuryState] | None:
+ target_id = directive.target_entity_id if directive is not None else None
+ if target_id is None:
+ return None
+ for country in world.countries:
+ if country.treasury is not None and country.treasury.id == target_id:
+ return country, country.treasury
+ return None
+
+
+def _apply_treasury_transfer(
+ npc: Npc,
+ world: WorldState,
+ treasury_ref: tuple[CountryState, TreasuryState],
+ directive: NpcDirective | None,
+) -> str:
+ country, treasury = treasury_ref
+ resource_type = _resource_type_from_directive(directive, default="coins")
+ amount = _amount_from_directive(directive, default=1)
+ take = bool(directive and directive.take)
+ position = treasury.position
+ resources = treasury.resources
+ if vec_distance(npc.position, position) > TRADE_RADIUS:
+ npc.position = move_toward(
+ npc.position,
+ position,
+ NPC_MOVE_STEP,
+ half_width=world.terrain.width / 2,
+ half_depth=world.terrain.depth / 2,
+ )
+ return "approaching treasury"
+ if take:
+ if npc.country_id == country.id:
+ return "guarding own treasury"
+ available = resources.get(resource_type, 0)
+ if available <= 0:
+ return "found empty treasury"
+ moved = min(amount, available)
+ resources[resource_type] = available - moved
+ _add_to_inventory_amount(npc, resource_type, moved)
+ _log_event(
+ world,
+ "treasury_stolen",
+ f"{npc.name} stole {moved} {resource_type} from {country.name}",
+ severity="warning",
+ actor_id=npc.id,
+ object_id=treasury.id,
+ )
+ return f"stealing {resource_type} from {country.name}"
+ if npc.country_id != country.id:
+ return "refusing to fund foreign treasury"
+ available = _inventory_amount(npc, resource_type)
+ if available <= 0:
+ return f"no {resource_type} to deposit"
+ moved = min(amount, available)
+ _remove_from_inventory(npc, resource_type, moved)
+ resources[resource_type] = resources.get(resource_type, 0) + moved
+ npc.resources_transferred += moved
+ _log_event(
+ world,
+ "treasury_deposit",
+ f"{npc.name} deposited {moved} {resource_type} into {country.name} treasury",
+ severity="good",
+ actor_id=npc.id,
+ object_id=treasury.id,
+ )
+ return f"depositing {resource_type} into {country.name} treasury"
+
+
def _resource_type_from_directive(directive: NpcDirective | None, *, default: str) -> str:
- if directive is not None and directive.resource_type in {"food", "herbs", "wood", "weapon"}:
+ if directive is not None and directive.resource_type in {"food", "herbs", "wood", "weapon", "coins"}:
return directive.resource_type
return default
def _amount_from_directive(directive: NpcDirective | None, *, default: int) -> int:
if directive is not None and directive.amount is not None:
- return max(1, min(5, directive.amount))
+ return max(1, min(20, directive.amount))
return default
@@ -2522,6 +2980,8 @@ def _add_to_inventory_amount(npc: Npc, resource_type: str, amount: int) -> None:
npc.inventory_wood += amount
elif resource_type == "weapon":
npc.inventory_weapon += amount
+ elif resource_type == "coins":
+ npc.inventory_coins += amount
def _remove_from_inventory(npc: Npc, resource_type: str, amount: int) -> None:
@@ -2533,6 +2993,8 @@ def _remove_from_inventory(npc: Npc, resource_type: str, amount: int) -> None:
npc.inventory_wood = max(0, npc.inventory_wood - amount)
elif resource_type == "weapon":
npc.inventory_weapon = max(0, npc.inventory_weapon - amount)
+ elif resource_type == "coins":
+ npc.inventory_coins = max(0, npc.inventory_coins - amount)
def _inventory_amount(npc: Npc, resource_type: str) -> int:
@@ -2544,6 +3006,8 @@ def _inventory_amount(npc: Npc, resource_type: str) -> int:
return npc.inventory_wood
if resource_type == "weapon":
return npc.inventory_weapon
+ if resource_type == "coins":
+ return npc.inventory_coins
return 0
@@ -2795,28 +3259,113 @@ def _plan_directive(
*,
target: Vec3 | None = None,
target_npc_id: str | None = None,
+ target_entity_id: str | None = None,
resource_id: str | None = None,
resource_type: str | None = None,
amount: int | None = None,
communication_intent: str | None = None,
message: str | None = None,
+ params: dict[str, object] | None = None,
away: bool = False,
) -> NpcDirective:
+ public_action = action
+ public_use_type: str | None = None
+ public_take = False
+ if action in {"strike", "attack"}:
+ public_action = "attack"
+ elif action in {"gather", "consume", "eat", "heal", "build", "repair"}:
+ public_action = "use"
+ public_use_type = {
+ "consume": "eat",
+ "eat": "eat",
+ "heal": "heal",
+ "gather": "gather",
+ "build": "build",
+ "repair": "repair",
+ }[action]
+ if action in {"consume", "eat"} and resource_type is None:
+ resource_type = "food"
+ if action == "heal" and resource_type is None:
+ resource_type = "herbs"
+ elif action in {"craft", "assign_cannon_operator", "fire"}:
+ public_action = "use"
+ public_use_type = action
+ elif action in {"communicate", "trade", "request_trade"}:
+ public_action = "speak"
+ elif action == "steal":
+ public_action = "transfer"
+ public_take = True
+ elif action in {
+ "go_home",
+ "patrol",
+ "flee",
+ "move_to_resource",
+ "move_to",
+ "find_food",
+ "find_herbs",
+ "wander",
+ }:
+ public_action = "move"
+ if action == "go_home" and target is None:
+ home = nearest_house(world, npc.position)
+ target = home.position if home is not None else None
+ elif action == "patrol" and target is None:
+ target = _patrol_target(world, npc)
+ away = away or action == "flee"
return NpcDirective(
npc_id=npc.id,
- action=action, # type: ignore[arg-type]
+ action=public_action, # type: ignore[arg-type]
target=target,
target_npc_id=target_npc_id,
+ target_entity_id=target_entity_id,
resource_id=resource_id,
resource_type=resource_type,
+ use_type=public_use_type,
+ params=params,
amount=amount,
communication_intent=communication_intent,
message=message,
away=away,
+ take=public_take,
intent=f"survival:{goal}",
)
role = normalize_role(npc.role)
+ election = active_election_for_npc(world, npc, tick=next_tick)
+ if election is not None and npc.id not in election.votes and election.candidate_ids:
+ candidate_id = _fallback_vote(world, election, voter_id=npc.id)
+ return directive("vote", target_npc_id=candidate_id)
+ country = country_for_npc(world, npc)
+ if country is not None and country.ruler_id == npc.id:
+ if country.cannon is None and _can_craft_cannon(world, npc):
+ return directive("craft", params={"recipe_id": "cannon"})
+ if country.cannon is not None and country.cannon.operator_id is None:
+ operator = next(
+ (
+ citizen
+ for citizen in world.living_npcs()
+ if citizen.country_id == country.id and citizen.id != npc.id
+ ),
+ npc,
+ )
+ return directive("assign_cannon_operator", params={"npc_id": operator.id})
+ if (
+ country is not None
+ and country.cannon is not None
+ and country.cannon.operator_id == npc.id
+ and next_tick >= country.cannon.cooldown_until_tick
+ ):
+ target = next(
+ (other for other in world.living_npcs() if other.country_id != country.id),
+ None,
+ )
+ if target is not None:
+ return directive(
+ "fire",
+ target=country.cannon.position,
+ target_npc_id=target.id,
+ params={"x": target.position.x, "z": target.position.z},
+ )
if goal == "obey_directive":
text = (npc.god_directive or "").lower()
@@ -2951,6 +3500,24 @@ def _plan_directive(
# routine_life/work fallback: socialise, share, forage, or wander -- and spread out.
neighbor = nearest_alive_npc(world, npc, max_dist=TALK_RADIUS)
phase = (next_tick + index) % 4
+ country = country_for_npc(world, npc)
+ if country is not None and country.treasury is not None and phase == 2:
+ if npc.inventory_coins > 0:
+ return directive(
+ "transfer",
+ target=country.treasury.position,
+ target_entity_id=country.treasury.id,
+ resource_type="coins",
+ amount=1,
+ )
+ if npc.inventory_wood > 0:
+ return directive(
+ "transfer",
+ target=country.treasury.position,
+ target_entity_id=country.treasury.id,
+ resource_type="wood",
+ amount=1,
+ )
if (
neighbor is not None
and npc.inventory_food >= SHARE_FOOD_SURPLUS
diff --git a/src/god_simulator/simulation/tick.py b/src/god_simulator/simulation/tick.py
index 18eb47416e178fc565b97dcf00e3d0144ef7e821..c748ec91ee6c40d6a27afcfb439270f94ab5c3c0 100644
--- a/src/god_simulator/simulation/tick.py
+++ b/src/god_simulator/simulation/tick.py
@@ -154,7 +154,7 @@ def apply_tick_plan(world: WorldState, next_tick: int, plan: TickPlan) -> None:
)
case "speak":
resolution = _apply_speak(npc, directive, npcs_by_id, next_tick)
- case "strike":
+ case "attack" | "strike":
resolution = _apply_strike(npc, directive, npcs_by_id, world, next_tick)
case "idle":
resolution = _apply_idle(npc, directive)
@@ -241,26 +241,35 @@ def validate_directive(
resolved = idle_directive(directive, memory=None, intent="actor_dead")
return _validation_result(directive, resolved, valid=False, reason="actor dead")
+ original_directive = directive
+
+ if directive.action == "idle":
+ return _validation_result(directive, directive, valid=True, reason="idle accepted")
+
+ if directive.action == "strike":
+ directive = replace(directive, action="attack")
+
perception, goal = refresh_citizen_motivation(world, npc)
if directive.action not in PRIMITIVE_ACTION_IDS:
fallback = autopilot_directive(world, npc, directive, perception, goal)
return _validation_result(
- directive,
+ original_directive,
fallback,
valid=False,
reason=f"unknown action {directive.action!r} replaced by autopilot",
)
match directive.action:
- case "idle":
- return _validation_result(directive, directive, valid=True, reason="idle accepted")
case "move":
return _validate_move(world, npc, directive, perception, goal)
case "speak":
return _validate_speak(world, npc, directive)
- case "strike":
- return _validate_strike(world, npc, directive)
+ case "attack" | "strike":
+ validation = _validate_strike(world, npc, directive)
+ if original_directive.action != directive.action:
+ return replace(validation, original_action=str(original_directive.action))
+ return validation
case _:
# use/transfer have no effects outside survival worlds.
resolved = idle_directive(
@@ -859,10 +868,17 @@ def _publish_plan_ledger_entries(
for entry in entries:
record = dict(entry)
if record.get("phase") == "npc_response":
- npc_id = str(record.get("npc_id") or "")
- verdict = verdicts.get(npc_id)
- if verdict is not None:
- record["validator_verdict"] = verdict
+ # Only a genuine model response (verdict still "pending") is
+ # reconciled with the engine's accept/repair/reject verdict. A
+ # response whose verdict is already "rejected" (model call failed /
+ # output unusable / NPC omitted) keeps that explicit reason so a
+ # fallback is never masked as "accepted" by the deterministic fill.
+ current = record.get("validator_verdict")
+ current_status = current.get("status") if isinstance(current, dict) else None
+ if current_status in (None, "pending"):
+ verdict = verdicts.get(str(record.get("npc_id") or ""))
+ if verdict is not None:
+ record["validator_verdict"] = verdict
append_record(record)
diff --git a/src/god_simulator/ui/card_observe.html b/src/god_simulator/ui/card_observe.html
index be896067b1d42f8b88bba702dcdda1ac51afef3a..c763b961951692a900873c8046b29d08e3cfe2c7 100644
--- a/src/god_simulator/ui/card_observe.html
+++ b/src/god_simulator/ui/card_observe.html
@@ -1,12 +1,12 @@
-
Step into the live 3D view and watch the simulation unfold in real time — characters gather, build, fight and fall.
+
Watch two AI countries live in the same world: citizens gather, vote, build, raid treasuries and operate cannons.
- - 🌐 Live 3D map
- - ⚔️ Watch the Overseer vs. Chaos play out
- - 🗺️ No setup required
+ - Live 3D map
+ - Countries, rulers, policies and elections
+ - No setup required
diff --git a/tests/e2e/god-simulator-scene.spec.ts b/tests/e2e/god-simulator-scene.spec.ts
index 1aeefd461a7594151c520bb0c79f0b4590aee742..6abaf587cf0ca775e2b70aa129f8ca8ae16ed293 100644
--- a/tests/e2e/god-simulator-scene.spec.ts
+++ b/tests/e2e/god-simulator-scene.spec.ts
@@ -8,7 +8,7 @@ test("renders the terrain scene with visible canvas pixels", async ({ page }) =>
const canvas = page.locator("canvas").first();
await expect(canvas).toBeVisible();
await expect(page.getByLabel("Simulation status")).toBeVisible();
- await expect(page.getByLabel("God console")).toBeVisible();
+ await expect(page.getByLabel("Countries")).toBeVisible();
await expect(page.getByLabel("Ada memories")).toBeVisible();
await page.keyboard.down("w");
diff --git a/tests/test_action_pipeline.py b/tests/test_action_pipeline.py
index c620ffbe622e7bb429e4d5820b1985fe6972a63d..a7a515dc9a95c0c5dec10ebd3e195f46b8c78e86 100644
--- a/tests/test_action_pipeline.py
+++ b/tests/test_action_pipeline.py
@@ -242,7 +242,7 @@ def test_threatened_victim_gets_survival_hints() -> None:
assert cora.mode == "threatened"
assert boris_goal is not None and boris_goal.goal_type == "survive"
assert cora_goal is not None and cora_goal.goal_type == "survive"
- assert {"move", "speak", "strike"}.issubset(set(boris_hints.suggested_actions))
+ assert {"move", "speak", "attack"}.issubset(set(boris_hints.suggested_actions))
assert cora_hints.suggested_actions[0] == "move"
diff --git a/tests/test_api_server.py b/tests/test_api_server.py
index 75e5106a9a54167562ec52fa3d548962cde78c04..ad2d8bcd8527da580c9f24c43c612926650ad133 100644
--- a/tests/test_api_server.py
+++ b/tests/test_api_server.py
@@ -155,7 +155,6 @@ def test_modal_health_urls_use_only_modal_openai_connectors() -> None:
assert _modal_health_urls(config) == [
"https://workspace--npc-serve.modal.run/health",
- "https://workspace--god-serve.modal.run/health",
]
config = _config(
diff --git a/tests/test_country_systems.py b/tests/test_country_systems.py
new file mode 100644
index 0000000000000000000000000000000000000000..d2aeaa37e0bade5a3d8a2c846413672be3af804a
--- /dev/null
+++ b/tests/test_country_systems.py
@@ -0,0 +1,267 @@
+"""Regression tests for the two-country demo systems and the LLM ledger.
+
+These pin the fixes audited against goal.md:
+* demo configs spawn two populated countries (Nemotron + Qwen);
+* the routing simulator preserves per-NPC LLM ledger entries and records which
+ model/model_profile drives each NPC (Qwen vs default);
+* treasury deposit and cross-country steal both resolve through `transfer`;
+* perception exposes rival countries (so an agent can steal/strike);
+* `vote` appears only during an election window;
+* an unusable model output is logged as an explicit fallback.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from god_simulator.config import (
+ ConnectorConfig,
+ GameConfig,
+ NpcConfig,
+ ServerConfig,
+ SimulationConfig,
+ WorldConfig,
+ load_game_config,
+)
+from god_simulator.domain import ElectionState
+from god_simulator.simulation.connectors.base import NpcDirective
+from god_simulator.simulation.connectors.deterministic import DeterministicWorldSimulator
+from god_simulator.simulation.connectors.openai_compatible import (
+ OpenAICompatibleWorldSimulator,
+)
+from god_simulator.simulation.connectors.routing import RoutingWorldSimulator
+from god_simulator.simulation.spawning import create_world
+from god_simulator.simulation.survival import (
+ apply_action_effects,
+ validate_survival_action,
+)
+
+
+def _survival_config(*, npc_count: int = 6) -> GameConfig:
+ return GameConfig(
+ world=WorldConfig(width=240, depth=80, terrain="plain_green", seed=42, survival=True),
+ npcs=NpcConfig(count=npc_count),
+ simulation=SimulationConfig(tick_ms=500),
+ server=ServerConfig(host="127.0.0.1", port=8000),
+ )
+
+
+def _move_completer(request: dict[str, Any]) -> dict[str, Any]:
+ """A minimal valid model: always returns one `move` tool call."""
+ return {
+ "choices": [
+ {
+ "message": {
+ "role": "assistant",
+ "tool_calls": [
+ {
+ "type": "function",
+ "function": {"name": "move", "arguments": json.dumps({})},
+ }
+ ],
+ }
+ }
+ ]
+ }
+
+
+# --------------------------------------------------------------------------- #
+# Two populated countries
+# --------------------------------------------------------------------------- #
+def test_demo_config_spawns_two_populated_countries() -> None:
+ config = load_game_config(Path("config/game.modal.local.json"))
+ world = create_world(config)
+
+ assert world.population == 12
+ assert {c.id for c in world.countries} == {"nemotron", "qwen"}
+ by_id = {c.id: c for c in world.countries}
+ assert len(by_id["nemotron"].citizen_ids) == 6
+ assert len(by_id["qwen"].citizen_ids) == 6
+ assert by_id["nemotron"].ruler_id is not None
+ assert by_id["qwen"].ruler_id is not None
+ # The Qwen country routes its NPCs to the Qwen connector.
+ qwen_npcs = [n for n in world.npcs if n.country_id == "qwen"]
+ assert qwen_npcs and all(n.connector_id == "qwen" for n in qwen_npcs)
+ nemotron_npcs = [n for n in world.npcs if n.country_id == "nemotron"]
+ assert nemotron_npcs and all(n.connector_id is None for n in nemotron_npcs)
+
+
+# --------------------------------------------------------------------------- #
+# Routing preserves LLM ledger entries + records the model per NPC
+# --------------------------------------------------------------------------- #
+def test_routing_preserves_ledger_entries_with_model_per_npc() -> None:
+ world = create_world(_survival_config())
+ deterministic = DeterministicWorldSimulator()
+ default = OpenAICompatibleWorldSimulator(
+ ConnectorConfig(type="openai_compatible", base_url="http://nemo.local/v1", model="nemotron-test"),
+ fallback=deterministic,
+ connector_id_filter=None,
+ chat_completer=_move_completer,
+ )
+ qwen = OpenAICompatibleWorldSimulator(
+ ConnectorConfig(type="openai_compatible", base_url="http://qwen.local/v1", model="qwen-test"),
+ fallback=deterministic,
+ connector_id_filter="qwen",
+ chat_completer=_move_completer,
+ )
+ router = RoutingWorldSimulator(routes={"qwen": qwen}, default=default)
+
+ plan = router.propose_tick(world, 1)
+ requests = {
+ entry["npc_id"]: entry
+ for entry in plan.ledger_entries
+ if entry.get("phase") == "npc_request"
+ }
+ # Every living NPC of both countries produced a logged request.
+ assert {n.id for n in world.living_npcs()} <= set(requests)
+
+ nemo = requests["npc-001"]
+ assert nemo["model"] == "nemotron-test"
+ assert nemo["model_profile"] == "default"
+ assert nemo["connector_id"] is None
+
+ qwen_entry = requests["qwen-001"]
+ assert qwen_entry["model"] == "qwen-test"
+ assert qwen_entry["model_profile"] == "qwen"
+ assert qwen_entry["connector_id"] == "qwen"
+
+ # Each NPC also has a paired response carrying its validator verdict.
+ responses = [e for e in plan.ledger_entries if e.get("phase") == "npc_response"]
+ assert {e["npc_id"] for e in responses} >= {"npc-001", "qwen-001"}
+
+
+# --------------------------------------------------------------------------- #
+# Treasury deposit + cross-country steal through `transfer`
+# --------------------------------------------------------------------------- #
+def test_treasury_deposit_via_transfer() -> None:
+ world = create_world(_survival_config())
+ country = next(c for c in world.countries if c.id == "qwen")
+ assert country.treasury is not None
+ citizen = next(n for n in world.npcs if n.country_id == "qwen")
+ citizen.position = country.treasury.position
+ citizen.inventory_wood = 4
+ before = country.treasury.resources.get("wood", 0)
+
+ directive = NpcDirective(
+ npc_id=citizen.id, action="transfer", target_entity_id=country.treasury.id,
+ resource_type="wood", amount=2,
+ )
+ verb = validate_survival_action(citizen, directive.action, world, directive)
+ apply_action_effects(citizen, verb, world, directive=directive)
+
+ assert country.treasury.resources["wood"] == before + 2
+ assert citizen.inventory_wood == 2
+ assert any(e.type == "treasury_deposit" for e in world.event_log)
+
+
+def test_treasury_steal_via_transfer_take() -> None:
+ world = create_world(_survival_config())
+ rival = next(c for c in world.countries if c.id == "nemotron")
+ assert rival.treasury is not None
+ thief = next(n for n in world.npcs if n.country_id == "qwen")
+ thief.position = rival.treasury.position
+ before = rival.treasury.resources.get("coins", 0)
+ assert before > 0
+
+ directive = NpcDirective(
+ npc_id=thief.id, action="transfer", target_entity_id=rival.treasury.id,
+ resource_type="coins", amount=3, take=True,
+ )
+ verb = validate_survival_action(thief, directive.action, world, directive)
+ summary = apply_action_effects(thief, verb, world, directive=directive)
+
+ assert "stealing" in summary.lower()
+ assert rival.treasury.resources["coins"] == before - 3
+ assert thief.inventory_coins == 3
+ assert any(e.type == "treasury_stolen" for e in world.event_log)
+
+
+# --------------------------------------------------------------------------- #
+# Perception exposes rival countries + own position
+# --------------------------------------------------------------------------- #
+def test_survival_prompt_exposes_rival_countries() -> None:
+ captured: list[dict[str, Any]] = []
+
+ def completer(request: dict[str, Any]) -> dict[str, Any]:
+ captured.append(json.loads(request["messages"][1]["content"]))
+ return _move_completer(request)
+
+ world = create_world(_survival_config())
+ simulator = OpenAICompatibleWorldSimulator(
+ ConnectorConfig(type="openai_compatible", base_url="http://nemo.local/v1", model="t"),
+ fallback=DeterministicWorldSimulator(),
+ connector_id_filter=None,
+ chat_completer=completer,
+ )
+ simulator.propose_tick(world, 1)
+
+ assert captured
+ payload = captured[0]
+ assert "position" in payload["own_state"]
+ overview = payload["world_countries"]
+ assert {c["id"] for c in overview} == {"nemotron", "qwen"}
+ own = next(c for c in overview if c["is_own"])
+ rival = next(c for c in overview if not c["is_own"])
+ assert own["id"] == "nemotron"
+ assert rival["id"] == "qwen"
+ assert rival["treasury_id"] == "treasury_qwen"
+ assert "coins" in rival["treasury"]
+
+
+# --------------------------------------------------------------------------- #
+# vote appears only during an election window
+# --------------------------------------------------------------------------- #
+def test_vote_tool_only_during_election() -> None:
+ world = create_world(_survival_config())
+ simulator = OpenAICompatibleWorldSimulator(
+ ConnectorConfig(type="openai_compatible", base_url="http://nemo.local/v1", model="t"),
+ fallback=DeterministicWorldSimulator(),
+ connector_id_filter=None,
+ chat_completer=_move_completer,
+ )
+ npc = next(n for n in world.npcs if n.country_id == "nemotron")
+
+ request = simulator._build_request(world, 5, npc) # noqa: SLF001
+ assert "vote" not in [t["function"]["name"] for t in request["tools"]]
+
+ world.elections.append(
+ ElectionState(
+ country_id="nemotron",
+ start_tick=4,
+ end_tick=14,
+ candidate_ids=[n.id for n in world.npcs if n.country_id == "nemotron"],
+ )
+ )
+ request = simulator._build_request(world, 5, npc) # noqa: SLF001
+ assert "vote" in [t["function"]["name"] for t in request["tools"]]
+
+
+# --------------------------------------------------------------------------- #
+# Unusable model output is logged as an explicit fallback
+# --------------------------------------------------------------------------- #
+def test_unusable_model_output_is_logged_as_fallback() -> None:
+ def empty_completer(_: dict[str, Any]) -> dict[str, Any]:
+ # Neither tool calls nor JSON content -> connector must fall back.
+ return {"choices": [{"message": {"role": "assistant", "content": ""}}]}
+
+ world = create_world(_survival_config(npc_count=2))
+ simulator = OpenAICompatibleWorldSimulator(
+ ConnectorConfig(type="openai_compatible", base_url="http://nemo.local/v1", model="t"),
+ fallback=DeterministicWorldSimulator(),
+ connector_id_filter=None,
+ chat_completer=empty_completer,
+ )
+
+ plan = simulator.propose_tick(world, 1)
+ fallbacks = [e for e in plan.ledger_entries if e.get("phase") == "npc_fallback"]
+ assert fallbacks
+ assert all(e.get("reason") for e in fallbacks)
+ rejected = [
+ e for e in plan.ledger_entries
+ if e.get("phase") == "npc_response"
+ and isinstance(e.get("validator_verdict"), dict)
+ and e["validator_verdict"].get("status") == "rejected"
+ ]
+ assert rejected
diff --git a/tests/test_gradio_app.py b/tests/test_gradio_app.py
index 62db10be6fda590b730805842638c19997886df6..cc257677fe5452229f784ecbdf68487530e4f29e 100644
--- a/tests/test_gradio_app.py
+++ b/tests/test_gradio_app.py
@@ -61,8 +61,7 @@ def test_gradio_app_force_deterministic_disables_modal_runtime(
assert snapshot.json()["simulation"]["models"] == []
god_command = client.post("/god-command", json={"command": "make Ada happy"})
- assert god_command.status_code == 400
- assert god_command.json()["error"] == "god_console_unavailable"
+ assert god_command.status_code == 404
def _write_config(tmp_path: Path) -> Path:
diff --git a/tests/test_living_village_smoke.py b/tests/test_living_village_smoke.py
index 977d8abb4084ae4976c886b0c798d3505038c677..be364cc7bf097a7b60d8e0d49dff3bfa7d65b3a6 100644
--- a/tests/test_living_village_smoke.py
+++ b/tests/test_living_village_smoke.py
@@ -16,7 +16,7 @@ def test_living_village_smoke_50_ticks_with_survival_and_social() -> None:
)
start_positions = {npc.id: (npc.position.x, npc.position.z) for npc in world.npcs}
- assert len(world.npcs) == 8
+ assert len(world.npcs) == 16
assert world.resource_nodes
assert world.beasts
diff --git a/tests/test_observability_perception.py b/tests/test_observability_perception.py
index 9b4cb13a400eef5e451c777a4c07c198052bc4d2..5028fcccf33572d9de3fe412d6a62492ae3f619b 100644
--- a/tests/test_observability_perception.py
+++ b/tests/test_observability_perception.py
@@ -23,8 +23,6 @@ from god_simulator.simulation.directives import (
expire_directives,
set_active_directive,
)
-from god_simulator.simulation.god_console import GodConsole
-from god_simulator.simulation.overseer import build_overseer_context, scripted_overseer_controller
from god_simulator.simulation.spawning import create_world
from god_simulator.simulation.survival import _kill_npc
from god_simulator.simulation.tick import advance_world
@@ -38,7 +36,8 @@ def test_dead_npcs_never_appear_in_model_target_fields_over_scripted_deaths() ->
def fake_chat_completer(request: dict[str, Any]) -> dict[str, Any]:
payload = json.loads(request["messages"][1]["content"])
captured_payloads.append(payload)
- return _tool_response("idle", {"npc_id": payload["npc_id"]})
+ npc_id = payload.get("npc_id") or payload.get("npc", {}).get("id")
+ return _tool_response("speak", {"npc_id": npc_id, "message": "observing"})
simulator = OpenAICompatibleWorldSimulator(
ConnectorConfig(
@@ -69,11 +68,6 @@ def test_dead_npcs_never_appear_in_model_target_fields_over_scripted_deaths() ->
for payload in captured_payloads:
_assert_survival_payload_has_no_dead_targets(payload, world, pre_tick_dead_ids)
_assert_world_live_registries_have_no_dead_targets(world, dead_ids)
- _assert_overseer_context_has_no_dead_targets(
- build_overseer_context(world, world.tick + 1),
- dead_ids,
- )
-
def test_directive_ttl_expiry_and_replacement() -> None:
npc = _npc("npc-001", "Ada")
@@ -159,15 +153,9 @@ def test_runtime_ledger_contains_all_required_phase_types_after_20_ticks() -> No
runtime = GameRuntime(
world=world,
simulator=_ledger_test_simulator(),
- god_console=_ledger_test_god_console(),
- overseer=scripted_overseer_controller(mode="autopilot", cycle_ticks=1),
ledger=ledger,
)
- status, payload = runtime.god_command("Give the village a harmless omen.")
- assert status == HTTPStatus.OK
- assert payload["queued"] is True
-
for _ in range(20):
status, _ = runtime.tick()
assert status == HTTPStatus.OK
@@ -178,24 +166,23 @@ def test_runtime_ledger_contains_all_required_phase_types_after_20_ticks() -> No
]
phases = {record["phase"] for record in records}
- assert {
- "god_command",
- "overseer_request",
- "overseer_response",
- "npc_request",
- "npc_response",
- "npc_fallback",
- "engine_events",
- }.issubset(phases)
+ assert {"npc_request", "npc_response", "npc_fallback", "engine_events"}.issubset(phases)
+ assert "god_command" not in phases
+ assert "overseer_request" not in phases
+ assert "overseer_response" not in phases
+ status, payload = runtime.admin_log_records(run_id="current", limit=5)
+ assert status == HTTPStatus.OK
+ assert payload["records"]
assert ledger.pretty_path.read_text(encoding="utf-8").strip()
def _ledger_test_simulator() -> OpenAICompatibleWorldSimulator:
def fake_chat_completer(request: dict[str, Any]) -> dict[str, Any]:
payload = json.loads(request["messages"][1]["content"])
- if payload["npc_id"] == "npc-001":
+ npc_id = payload.get("npc_id") or payload.get("npc", {}).get("id")
+ if npc_id == "npc-001":
raise RuntimeError("scripted model outage")
- return _tool_response("idle", {"npc_id": payload["npc_id"]})
+ return _tool_response("speak", {"npc_id": npc_id, "message": "observing"})
return OpenAICompatibleWorldSimulator(
ConnectorConfig(
@@ -209,35 +196,13 @@ def _ledger_test_simulator() -> OpenAICompatibleWorldSimulator:
)
-def _ledger_test_god_console() -> GodConsole:
- def fake_chat_completer(_: dict[str, Any]) -> dict[str, Any]:
- return {
- "choices": [
- {
- "message": {
- "content": "The omen is recorded.",
- "tool_calls": [],
- }
- }
- ]
- }
-
- return GodConsole(
- ConnectorConfig(
- type="openai_compatible",
- base_url="http://llm.local/v1",
- model="god-test-model",
- ),
- chat_completer=fake_chat_completer,
- )
-
-
def _assert_survival_payload_has_no_dead_targets(
payload: dict[str, Any],
world: WorldState,
dead_ids: set[str],
) -> None:
- assert payload["npc_id"] not in dead_ids
+ payload_npc_id = payload.get("npc_id") or payload.get("npc", {}).get("id")
+ assert payload_npc_id not in dead_ids
dead_names = {
npc.name.lower()
for npc in world.npcs
@@ -278,20 +243,6 @@ def _assert_world_live_registries_have_no_dead_targets(
assert not any(name in lowered for name in dead_names)
-def _assert_overseer_context_has_no_dead_targets(
- context: dict[str, Any],
- dead_ids: set[str],
-) -> None:
- assert not ({npc["id"] for npc in context["npcs"]} & dead_ids)
- for house in context["houses"]:
- assert not (set(house["occupants"]) & dead_ids)
- for beast in context["beasts"]:
- assert beast["target_npc_id"] not in dead_ids
- for threat in context["current_threats"]:
- assert threat.get("npc_id") not in dead_ids
- assert threat.get("target_npc_id") not in dead_ids
-
-
def _survival_world(*, npc_count: int) -> WorldState:
return create_world(_config(npc_count=npc_count, survival=True))
diff --git a/tests/test_openai_connector.py b/tests/test_openai_connector.py
index 9f62fdb27b17f0128da0a20dca52ddad673e74e7..52035c40d3c9de9e02786be4d5fed1e226d3de54 100644
--- a/tests/test_openai_connector.py
+++ b/tests/test_openai_connector.py
@@ -78,10 +78,9 @@ def test_openai_connector_uses_tool_calls_for_npc_directives() -> None:
assert [tool["function"]["name"] for tool in first_request["tools"]] == [
"move",
"speak",
- "strike",
+ "attack",
"use",
"transfer",
- "idle",
]
payload = json.loads(first_request["messages"][1]["content"])
assert "npcs" not in payload
@@ -89,7 +88,7 @@ def test_openai_connector_uses_tool_calls_for_npc_directives() -> None:
assert payload["npc"]["health"] == 100
assert "attack_damage" in payload["npc"]
assert payload["current_goal"]["goal_type"] == "routine_life"
- assert payload["suggested_actions"] == ["move", "speak", "use", "transfer", "idle"]
+ assert payload["suggested_actions"] == ["move", "speak", "use", "transfer"]
assert payload["suggestion_reason"]
assert "forbidden_actions" not in payload
assert world.npcs[0].position.x == 6
@@ -432,8 +431,8 @@ def test_openai_payload_contains_goal_and_suggested_actions() -> None:
"target_id": payload["current_goal"]["target_npc_id"],
}
else:
- name = "idle"
- arguments = {"npc_id": npc_id, "intent": "observe"}
+ name = "speak"
+ arguments = {"npc_id": npc_id, "message": "observing"}
return {
"choices": [
{
@@ -478,7 +477,7 @@ def test_openai_payload_contains_goal_and_suggested_actions() -> None:
assert ada_payload["current_mode"] == "hostile_pursuit"
assert ada_payload["suggested_actions"] == ["move", "speak"]
assert "forbidden_actions" not in ada_payload
- assert tool_names == ["move", "speak", "strike", "use", "transfer", "idle"]
+ assert tool_names == ["move", "speak", "attack", "use", "transfer"]
assert ada.intention == "approaching Boris to attack"
diff --git a/tests/test_players_mcp.py b/tests/test_players_mcp.py
index 8e6c7c4dcebee26bd507490ae7c1661c10bd17ad..4c57b8f36489e68fdf7178123717ac31bdb8ea2d 100644
--- a/tests/test_players_mcp.py
+++ b/tests/test_players_mcp.py
@@ -88,7 +88,7 @@ def test_player_can_build_despite_default_role() -> None:
token, npc_id = created["token"], created["npc_id"]
runtime.player_observe(token)
- runtime.player_act(token, {"action": "build"})
+ runtime.player_act(token, {"action": "use", "use_type": "build"})
runtime.tick()
npc = next(npc for npc in runtime._world.npcs if npc.id == npc_id)
@@ -111,7 +111,7 @@ def test_dead_character_returns_clear_error() -> None:
assert payload["error"] == "character_dead"
assert "died" in payload["message"]
- act_status, act_payload = runtime.player_act(token, {"action": "gather"})
+ act_status, act_payload = runtime.player_act(token, {"action": "use", "use_type": "gather"})
assert act_status == HTTPStatus.GONE
assert act_payload["error"] == "character_dead"
@@ -190,22 +190,22 @@ def test_mcp_initialize_list_and_call(tmp_path: Path) -> None:
"attack",
"use",
"transfer",
- "build",
- "gather",
}
assert expected <= names
+ assert "build" not in names
+ assert "gather" not in names
- gather = client.post(
+ use = client.post(
f"/mcp/{token}",
json={
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
- "params": {"name": "gather", "arguments": {}},
+ "params": {"name": "use", "arguments": {"use_type": "gather"}},
},
headers=headers,
)
- result = gather.json()["result"]
+ result = use.json()["result"]
assert result["isError"] is False
assert result["structuredContent"]["observation"]["self"]["npc_id"].startswith("player-")
diff --git a/tests/test_survival_loop.py b/tests/test_survival_loop.py
index 7b559fabdcecae65bcc336c58a0cb4df0aa26eb5..6a329e2715133cab63ccf04990bae5f48808e246 100644
--- a/tests/test_survival_loop.py
+++ b/tests/test_survival_loop.py
@@ -375,25 +375,22 @@ def test_llm_connector_drives_survival_actions() -> None:
assert "openai_compatible" in world.last_tick_source
-def test_llm_always_sees_all_six_primitive_tools() -> None:
+def test_llm_always_sees_all_public_primitive_tools() -> None:
names = [tool["function"]["name"] for tool in primitive_tools()]
- assert names == ["move", "speak", "strike", "use", "transfer", "idle"]
- # Survival worlds add the role-gated extras on top of the primitives.
- assert set(TOOL_DEFINITIONS) == set(names) | {"go_home", "patrol", "build"}
+ assert names == ["move", "speak", "attack", "use", "transfer"]
+ assert set(names).issubset(set(TOOL_DEFINITIONS))
+ assert "vote" in TOOL_DEFINITIONS
# Every suggested action references a real tool.
for suggested in GOAL_SUGGESTED_ACTIONS.values():
assert set(suggested).issubset(set(TOOL_DEFINITIONS))
-def test_survival_tools_are_role_gated() -> None:
+def test_survival_tools_are_not_role_gated() -> None:
guard_names = {tool["function"]["name"] for tool in survival_tools_for_role("guard")}
builder_names = {tool["function"]["name"] for tool in survival_tools_for_role("builder")}
gatherer_names = {tool["function"]["name"] for tool in survival_tools_for_role("gatherer")}
- assert "patrol" in guard_names and "build" not in guard_names
- assert "build" in builder_names and "strike" not in builder_names
- assert "patrol" not in gatherer_names and "build" not in gatherer_names
- # Everyone keeps the shared survival surface.
+ assert guard_names == builder_names == gatherer_names
for names in (guard_names, builder_names, gatherer_names):
- assert {"move", "speak", "use", "transfer", "idle", "go_home"}.issubset(names)
+ assert names == {"move", "speak", "attack", "use", "transfer"}