amogus/docs/api.md

7.0 KiB

Source Code Documentation

Module Overview

src/engine/

simulator.py

Discrete event simulator core.

class Simulator:
    def schedule_at(time: float, event_type: str, data: dict) -> Event
    def schedule_in(delay: float, event_type: str, data: dict) -> Event
    def step() -> Event | None  # Process next event
    def run_until(time: float)  # Run until game time
    def on(event_type: str, handler: Callable)  # Register handler

game.py

Game engine with full mechanics.

class GameEngine:
    def add_player(id, name, color, role) -> Player
    def queue_action(player_id, action_type, params) -> int
    def resolve_actions() -> list[dict]  # Priority-ordered resolution
    def check_win_condition() -> str | None  # "impostor", "crewmate", None
    def get_impostor_context(player_id) -> dict  # Fellow impostors

triggers.py

Event-driven trigger system.

class TriggerRegistry:
    def register_agent(agent_id)
    def subscribe(agent_id, trigger_type)
    def mute(agent_id, condition: TriggerCondition)
    def should_fire(agent_id, trigger_type, time) -> bool
    def get_agents_for_trigger(trigger_type, time) -> list[str]

discussion.py

Round-table discussion orchestrator.

class DiscussionOrchestrator:
    def calculate_priority(player_id, name, desire) -> int
    def select_speaker(bids: dict) -> str | None
    def add_message(player_id, name, message, target=None)
    def advance_round(all_desires_low: bool) -> bool
    def get_transcript() -> list[dict]

types.py

Core data structures.

@dataclass
class Player: id, name, color, role, position, speed, is_alive, ...
class Position: room_id, edge_id, progress
class Body: id, player_id, player_name, position, time_of_death
class Event: time, event_type, data
class Role: CREWMATE, IMPOSTOR, GHOST
class GamePhase: LOBBY, PLAYING, DISCUSSION, VOTING, ENDED

fog_of_war.py

Per-player knowledge tracking (fog-of-war).

class FogOfWarManager:
    def register_player(player_id)
    def update_vision(observer_id, visible_players, room_id, timestamp)
    def witness_vent(observer_id, venter_id, ...) -> None
    def witness_kill(observer_id, killer_id, victim_id, ...) -> None
    def announce_death(player_id, via)  # Broadcast to all
    def get_player_game_state(player_id, full_state) -> dict  # Filtered view

class PlayerKnowledge:
    known_dead: set[str]
    last_seen: dict[str, PlayerSighting]
    witnessed_events: list[WitnessedEvent]

available_actions.py

Dynamic action generator per tick.

class AvailableActionsGenerator:
    def get_available_actions(player_id) -> dict
    def to_prompt_context(player_id) -> dict  # Compact for LLM

Returns: movement, interactions, kills, sabotages based on role/location.

trigger_messages.py

JSON schemas for trigger reasons.

class TriggerMessageBuilder:
    @staticmethod player_enters_fov(...) -> TriggerMessage
    @staticmethod body_in_fov(...) -> TriggerMessage
    @staticmethod vent_witnessed(...) -> TriggerMessage
    @staticmethod kill_witnessed(...) -> TriggerMessage
    @staticmethod death(...) -> TriggerMessage
    # ... 15+ trigger types

meeting_flow.py

Full meeting lifecycle manager.

class MeetingFlowManager:
    def start_meeting(called_by, reason, body_location)
    def submit_interrupt_note(player_id, note)
    def submit_prep_thoughts(player_id, thoughts)
    def add_message(speaker_id, speaker_name, message, target)
    def submit_vote(player_id, vote)
    def tally_votes() -> (ejected, details)
    def end_meeting(ejected, was_impostor)

src/map/

graph.py

Graph-based map representation.

class GameMap:
    def add_room(room: Room)
    def add_edge(edge: Edge)
    def get_neighbors(room_id) -> list[tuple[edge_id, room_id]]
    def find_path(from_room, to_room) -> list[edge_id]
    def path_distance(path: list[edge_id]) -> float
    def load(path: str) -> GameMap  # From JSON
    def save(path: str)  # To JSON

@dataclass
class Room: id, name, tasks: list[Task], vent: Vent | None
class Edge: id, room_a, room_b, distance
class Task: id, name, duration
class Vent: id, connects_to: list[str]

src/agents/

agent.py

Stateless LLM agent wrapper.

class Agent:
    def get_action(game_context: dict, trigger: Trigger) -> dict
    def get_discussion_response(context: dict, transcript: list) -> dict
    def get_vote(context: dict, transcript: list) -> dict
    def reflect(game_summary: dict)  # Post-game learning

scratchpads.py

File-based persistent memory.

class ScratchpadManager:
    def read(name: str) -> dict
    def write(name: str, content: dict)
    def update(name: str, updates: dict)  # Merge
    def clear_game_specific()  # Keep only learned.json

prompt_assembler.py

System + user prompt builder.

class PromptAssembler:
    def build_system_prompt(phase, game_settings, map_name, learned) -> str
    def build_action_prompt(player_state, history, vision, actions, trigger) -> str
    def build_discussion_prompt(player_state, transcript, meeting_scratchpad) -> str
    def build_voting_prompt(player_state, transcript, vote_counts) -> str
    def build_meeting_interrupt_prompt(player_state, interrupted_action) -> str
    def build_consolidation_prompt(player_state, meeting_result, scratchpad) -> str

class PromptConfig:
    model_name, persona, strategy_level, meta_level, is_impostor, fellow_impostors

src/llm/

client.py

OpenRouter API wrapper.

class LLMClient:
    def chat(messages: list[dict], json_mode=True) -> dict
    def chat_stream(messages: list[dict]) -> Iterator[str]

Configuration

Use the CLI config manager for easy configuration:

python scripts/config_manager.py settings list    # List all settings
python scripts/config_manager.py settings get kill_cooldown
python scripts/config_manager.py settings set kill_cooldown 30
python scripts/config_manager.py validate         # Validate all configs

config/game_settings.json

{
    "map_name": "skeld",
    "num_impostors": 2,
    "player_speed": 100.0,
    "crewmate_vision": 1.0,
    "impostor_vision": 1.5,
    "kill_cooldown": 25.0,
    "kill_distance": "medium",
    "emergencies_per_player": 1,
    "discussion_time": 30.0,
    "voting_time": 120.0,
    "confirm_ejects": true,
    "visual_tasks": true,
    "taskbar_updates": "always"
}

All measurements are in pixels (map is 2000x1500). Speed is pixels/second.

data/maps/skeld.json

{
  "canvas": {"width": 2000, "height": 1500},
  "rooms": [
    {"id": "cafeteria", "name": "Cafeteria", "center": [1000, 350], "bounds": [[850, 200], [1150, 500]]},
    ...
  ],
  "edges": [
    {"id": "cafe_weapons", "room_a": "cafeteria", "room_b": "weapons", "waypoints": [[1150, 350], [1300, 350]]},
    ...
  ],
  "walls": [
    {"start": [850, 200], "end": [1150, 200]},
    ...
  ],
  "spawn_points": {"cafeteria": [1000, 400]}
}