amogus/docs/api.md

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

config/game_settings.yaml

num_impostors: 2
kill_cooldown: 25.0
vision_range: 10.0
impostor_vision_multiplier: 1.5
light_sabotage_vision_multiplier: 0.25
emergencies_per_player: 1
confirm_ejects: true

data/maps/skeld.json

{
  "rooms": [
    {"id": "cafeteria", "name": "Cafeteria", "tasks": [...], "vent": null},
    ...
  ],
  "edges": [
    {"id": "cafe_weapons", "room_a": "cafeteria", "room_b": "weapons", "distance": 5.0},
    ...
  ]
}