API Reference

MnesOS

MnesOS - MnesOS is fully Agentic RPG Game Engine..

CartridgeLoader

Load and validate a cartridge directory.

Usage::

loader = CartridgeLoader()
cartridge = loader.load("cartridges/generic-rpg")

# Pass results into GameState:
initial_state = {
    ...
    "yare_config": cartridge.yare_config,
    "prompt_directives": cartridge.prompt_directives,
    "lore_path": cartridge.lore_path,
    "bot_memory": cartridge.initial_state,
}

Source code in src/MnesOS/cartridge.py

class CartridgeLoader:
    """
    Load and validate a cartridge directory.

    Usage::

        loader = CartridgeLoader()
        cartridge = loader.load("cartridges/generic-rpg")

        # Pass results into GameState:
        initial_state = {
            ...
            "yare_config": cartridge.yare_config,
            "prompt_directives": cartridge.prompt_directives,
            "lore_path": cartridge.lore_path,
            "bot_memory": cartridge.initial_state,
        }
    """

    def load(self, cartridge_dir: str, persona: Optional[Any] = None) -> LoadedCartridge:
        base = Path(cartridge_dir)
        persona_tokens = _extract_persona_tokens(persona)

        # ── yare.yaml ─────────────────────────────────────────────────────
        yare_path = base / "yare.yaml"
        if not yare_path.exists():
            raise FileNotFoundError(f"yare.yaml not found in {cartridge_dir!r}")
        with yare_path.open() as f:
            yare_config: Dict[str, Any] = yaml.safe_load(f) or {}
        _validate_yare(yare_config)
        logger.info("yare.yaml validated for cartridge %r", cartridge_dir)

        # ── prompt_directives.yaml (optional) ─────────────────────────────
        directives_path = base / "prompt_directives.yaml"
        if directives_path.exists():
            with directives_path.open() as f:
                raw_directives = yaml.safe_load(f) or {}
            prompt_directives = _validate_prompt_directives(raw_directives)
            prompt_directives = {
                key: _compile_persona_macros(value, persona_tokens)
                for key, value in prompt_directives.items()
            }
            logger.info(
                "prompt_directives.yaml validated for cartridge %r "
                "(keys: %s)",
                cartridge_dir,
                list(prompt_directives.keys()),
            )
        else:
            prompt_directives = {}
            logger.info(
                "No prompt_directives.yaml found in %r — using empty directives.",
                cartridge_dir,
            )

        # ── bot_lore.md ───────────────────────────────────────────────────
        lore_path = base / "bot_lore.md"
        if not lore_path.exists():
            raise FileNotFoundError(f"bot_lore.md not found in {cartridge_dir!r}")
        lore_content = _compile_persona_macros(
            lore_path.read_text(encoding="utf-8"), persona_tokens
        )

        # ── first-message.md (optional) ───────────────────────────────────
        fm_path = base / "first-message.md"
        first_message = ""
        if fm_path.exists():
            first_message = _compile_persona_macros(
                fm_path.read_text(encoding="utf-8"), persona_tokens
            )

        # ── derive initial state from schema defaults ─────────────────────
        initial_state = _build_initial_state(yare_config.get("state_schema", {}))

        return LoadedCartridge(
            yare_config=yare_config,
            prompt_directives=prompt_directives,
            lore_path=str(lore_path),
            lore_content=lore_content,
            first_message=first_message,
            persona_context={
                "appearance": persona_tokens.get("appearance", ""),
                "background": persona_tokens.get("background", ""),
                "personality": persona_tokens.get("personality", ""),
            },
            initial_state=initial_state,
        )

    def load_from_version(self, version: Any, persona: Optional[Any] = None) -> LoadedCartridge:
        """Load a cartridge directly from a CartridgeVersion DB record."""
        persona_tokens = _extract_persona_tokens(persona)

        yare_config = version.yare_spec
        _validate_yare(yare_config)
        logger.info("yare_spec validated from CartridgeVersion %s", version.id)

        prompt_directives = {
            key: _compile_persona_macros(value, persona_tokens)
            for key, value in version.prompt_directives.items()
        }
        logger.info(
            "prompt_directives validated from CartridgeVersion %s (keys: %s)",
            version.id,
            list(prompt_directives.keys()),
        )

        lore_content = _compile_persona_macros(
            version.bot_lore, persona_tokens
        )

        first_message = _compile_persona_macros(
            getattr(version, "first_message", ""), persona_tokens
        )

        initial_state = _build_initial_state(yare_config.get("state_schema", {}))

        return LoadedCartridge(
            yare_config=yare_config,
            prompt_directives=prompt_directives,
            lore_path=f"db://{version.id}",
            lore_content=lore_content,
            first_message=first_message,
            persona_context={
                "appearance": persona_tokens.get("appearance", ""),
                "background": persona_tokens.get("background", ""),
                "personality": persona_tokens.get("personality", ""),
            },
            initial_state=initial_state,
        )

load_from_version(version, persona=None)

Load a cartridge directly from a CartridgeVersion DB record.

Source code in src/MnesOS/cartridge.py

def load_from_version(self, version: Any, persona: Optional[Any] = None) -> LoadedCartridge:
    """Load a cartridge directly from a CartridgeVersion DB record."""
    persona_tokens = _extract_persona_tokens(persona)

    yare_config = version.yare_spec
    _validate_yare(yare_config)
    logger.info("yare_spec validated from CartridgeVersion %s", version.id)

    prompt_directives = {
        key: _compile_persona_macros(value, persona_tokens)
        for key, value in version.prompt_directives.items()
    }
    logger.info(
        "prompt_directives validated from CartridgeVersion %s (keys: %s)",
        version.id,
        list(prompt_directives.keys()),
    )

    lore_content = _compile_persona_macros(
        version.bot_lore, persona_tokens
    )

    first_message = _compile_persona_macros(
        getattr(version, "first_message", ""), persona_tokens
    )

    initial_state = _build_initial_state(yare_config.get("state_schema", {}))

    return LoadedCartridge(
        yare_config=yare_config,
        prompt_directives=prompt_directives,
        lore_path=f"db://{version.id}",
        lore_content=lore_content,
        first_message=first_message,
        persona_context={
            "appearance": persona_tokens.get("appearance", ""),
            "background": persona_tokens.get("background", ""),
            "personality": persona_tokens.get("personality", ""),
        },
        initial_state=initial_state,
    )

ConfigMerger

Utility for building a :class:MnesOSRuntimeConfig from layered dicts.

Source code in src/MnesOS/config.py

class ConfigMerger:
    """Utility for building a :class:`MnesOSRuntimeConfig` from layered dicts."""

    @staticmethod
    def merge(
        cartridge_defaults: dict,
        player_settings: dict,
        request_overrides: dict,
    ) -> MnesOSRuntimeConfig:
        """Merge three config layers into a :class:`MnesOSRuntimeConfig`.

        Precedence (highest wins):

        1. *request_overrides*
        2. *player_settings*
        3. *cartridge_defaults*

        Each input dict may contain any subset of the top-level keys defined
        in :class:`MnesOSRuntimeConfig`.  LLM role keys (``director_llm``,
        ``narrator_llm``, ``npc_llm``, ``embedding_llm``) are merged
        recursively so that a player can override only ``temperature`` without
        having to specify the full role config.

        Parameters
        ----------
        cartridge_defaults:
            Typically built from a :class:`~MnesOS.cartridge.LoadedCartridge`
            (``yare_config``, ``prompt_directives``, and optional LLM hints).
        player_settings:
            Persisted player preferences (e.g. preferred provider/model).
        request_overrides:
            Per-request overrides supplied by the frontend for this turn.

        Returns
        -------
        MnesOSRuntimeConfig
        """
        merged: dict = {}
        for layer in (cartridge_defaults, player_settings, request_overrides):
            merged = _deep_update(merged, layer)

        # Build per-role LLMRoleConfig objects from the merged dicts
        role_configs: Dict[str, LLMRoleConfig] = {}
        for role in _LLM_ROLE_KEYS:
            role_data = merged.get(role)
            if isinstance(role_data, dict):
                role_configs[role] = LLMRoleConfig(**role_data)
            elif isinstance(role_data, LLMRoleConfig):
                role_configs[role] = role_data
            else:
                role_configs[role] = LLMRoleConfig()

        return MnesOSRuntimeConfig(
            **role_configs,
            yare_config=merged.get("yare_config", {}),
            prompt_directives=merged.get("prompt_directives", {}),
        )

merge(cartridge_defaults, player_settings, request_overrides) staticmethod

Merge three config layers into a :class:MnesOSRuntimeConfig.

Precedence (highest wins):

1. request_overrides
2. player_settings
3. cartridge_defaults

Each input dict may contain any subset of the top-level keys defined in :class:MnesOSRuntimeConfig. LLM role keys (director_llm, narrator_llm, npc_llm, embedding_llm) are merged recursively so that a player can override only temperature without having to specify the full role config.

Parameters

cartridge_defaults: Typically built from a :class:~MnesOS.cartridge.LoadedCartridge (yare_config, prompt_directives, and optional LLM hints). player_settings: Persisted player preferences (e.g. preferred provider/model). request_overrides: Per-request overrides supplied by the frontend for this turn.

Returns

MnesOSRuntimeConfig

Source code in src/MnesOS/config.py

@staticmethod
def merge(
    cartridge_defaults: dict,
    player_settings: dict,
    request_overrides: dict,
) -> MnesOSRuntimeConfig:
    """Merge three config layers into a :class:`MnesOSRuntimeConfig`.

    Precedence (highest wins):

    1. *request_overrides*
    2. *player_settings*
    3. *cartridge_defaults*

    Each input dict may contain any subset of the top-level keys defined
    in :class:`MnesOSRuntimeConfig`.  LLM role keys (``director_llm``,
    ``narrator_llm``, ``npc_llm``, ``embedding_llm``) are merged
    recursively so that a player can override only ``temperature`` without
    having to specify the full role config.

    Parameters
    ----------
    cartridge_defaults:
        Typically built from a :class:`~MnesOS.cartridge.LoadedCartridge`
        (``yare_config``, ``prompt_directives``, and optional LLM hints).
    player_settings:
        Persisted player preferences (e.g. preferred provider/model).
    request_overrides:
        Per-request overrides supplied by the frontend for this turn.

    Returns
    -------
    MnesOSRuntimeConfig
    """
    merged: dict = {}
    for layer in (cartridge_defaults, player_settings, request_overrides):
        merged = _deep_update(merged, layer)

    # Build per-role LLMRoleConfig objects from the merged dicts
    role_configs: Dict[str, LLMRoleConfig] = {}
    for role in _LLM_ROLE_KEYS:
        role_data = merged.get(role)
        if isinstance(role_data, dict):
            role_configs[role] = LLMRoleConfig(**role_data)
        elif isinstance(role_data, LLMRoleConfig):
            role_configs[role] = role_data
        else:
            role_configs[role] = LLMRoleConfig()

    return MnesOSRuntimeConfig(
        **role_configs,
        yare_config=merged.get("yare_config", {}),
        prompt_directives=merged.get("prompt_directives", {}),
    )

LLMRoleConfig

Bases: BaseModel

Configuration for a single LLM role (director, narrator, npc, embedding).

Source code in src/MnesOS/config.py

class LLMRoleConfig(BaseModel):
    """Configuration for a single LLM role (director, narrator, npc, embedding)."""

    provider: str = "openrouter"
    model_name: str = ""
    temperature: float = 0.7
    max_tokens: Optional[int] = None

LoadedCartridge dataclass

Fully validated, runtime-ready cartridge.

Source code in src/MnesOS/cartridge.py

@dataclass
class LoadedCartridge:
    """Fully validated, runtime-ready cartridge."""
    yare_config: Dict[str, Any]
    prompt_directives: Dict[str, str]
    lore_path: str
    lore_content: str
    first_message: str = ""
    persona_context: Dict[str, str] = field(default_factory=dict)
    initial_state: Dict[str, Any] = field(default_factory=dict)

MnesOSRuntimeConfig

Bases: BaseModel

The final, merged configuration used for a single process_turn request.

Source code in src/MnesOS/config.py

class MnesOSRuntimeConfig(BaseModel):
    """The final, merged configuration used for a single ``process_turn`` request."""

    director_llm: LLMRoleConfig = LLMRoleConfig()
    narrator_llm: LLMRoleConfig = LLMRoleConfig()
    npc_llm: LLMRoleConfig = LLMRoleConfig()
    embedding_llm: LLMRoleConfig = LLMRoleConfig()

    # Cartridge specifics mapped into the run
    yare_config: Dict[str, Any] = {}
    prompt_directives: Dict[str, str] = {}

Orchestrator

MVP Orchestrator for the MnesOS YARE engine.

Supports two operating modes:

Stateful mode (no storage): the orchestrator keeps self._state in memory across process_turn calls. This is the legacy CLI / notebook experience.

Stateless mode (storage provided): each process_turn call receives a parent_turn_id, hydrates state from the turn-log tree, invokes the graph, and returns the result dict. The orchestrator does NOT persist to the database — the API route handles that.

Usage (stateful)::

orch = Orchestrator(cartridge_dir="cartridges/generic-rpg")
response = orch.process_turn("I look around.")

Usage (stateless)::

orch = Orchestrator(
    cartridge_version=my_cartridge_version,
    persona=my_persona,
    storage=my_sqlite_store,
)
result = orch.process_turn(
    "I look around.",
    parent_turn_id="prev-turn-uuid",
)
# result == {"narrator_text": "...", "yare_delta": {...}}

Source code in src/MnesOS/orchestrator.py

class Orchestrator:
    """
    MVP Orchestrator for the MnesOS YARE engine.

    Supports two operating modes:

    **Stateful mode** (no ``storage``): the orchestrator keeps ``self._state``
    in memory across ``process_turn`` calls.  This is the legacy CLI /
    notebook experience.

    **Stateless mode** (``storage`` provided): each ``process_turn`` call
    receives a ``parent_turn_id``, hydrates state from the turn-log tree,
    invokes the graph, and returns the result dict.  The orchestrator does
    NOT persist to the database — the API route handles that.

    Usage (stateful)::

        orch = Orchestrator(cartridge_dir="cartridges/generic-rpg")
        response = orch.process_turn("I look around.")

    Usage (stateless)::

        orch = Orchestrator(
            cartridge_version=my_cartridge_version,
            persona=my_persona,
            storage=my_sqlite_store,
        )
        result = orch.process_turn(
            "I look around.",
            parent_turn_id="prev-turn-uuid",
        )
        # result == {"narrator_text": "...", "yare_delta": {...}}
    """

    def __init__(
        self,
        storage: AbstractStorageComponent,
        cartridge_dir: Optional[str] = None,
        cartridge_version: Optional[Any] = None,
        persona: Any = None,
        llm_director=None,
        llm_npc=None,
        llm_narrator=None,
    ) -> None:
        loader = CartridgeLoader()
        if cartridge_version:
            self._cartridge: LoadedCartridge = loader.load_from_version(cartridge_version, persona=persona)
            logger.info("Cartridge loaded from DB version %s", cartridge_version.id)
        elif cartridge_dir:
            self._cartridge: LoadedCartridge = loader.load(cartridge_dir, persona=persona)
            logger.info("Cartridge loaded from %r", cartridge_dir)
        else:
            raise ValueError("Must provide either cartridge_dir or cartridge_version.")

        if self._cartridge.yare_config.get("separate_npc", False):
            raise NotImplementedError(
                "separate_npc=True is not yet implemented. "
                "This feature is planned for a future release. "
                "See docs/feature_roadmap.md for details. "
                "To use the orchestrator, set separate_npc=False or omit it."
            )

        if storage is None:
            raise ValueError("A storage backend is required for Orchestrator.")
        self._storage = storage
        self._app = self._compile_graph(llm_director, llm_npc, llm_narrator)
        logger.info("Graph compiled. Nodes: %s", list(self._app.get_graph().nodes.keys()))

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------



    @property
    def cartridge(self) -> LoadedCartridge:
        """The loaded cartridge metadata."""
        return self._cartridge

    def process_turn(
        self,
        user_input: str,
        *,
        interaction: Optional[Dict[str, Any]] = None,
        parent_turn_id: Optional[str] = None,
        llm_clients: Optional[Dict[str, Any]] = None,
        player_settings: Optional[Dict[str, Any]] = None,
        request_overrides: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """
        Execute one game turn.

        Hydrates state from the turn-log tree, invokes the graph, and returns a result dict.
        The Orchestrator does NOT save the turn to the DB; the API route handles that.

        Parameters
        ----------
        user_input : str
            The player's raw text input.
        parent_turn_id : str, optional
            ID of the previous turn. If None, assumes a new game starting at Turn 0.
        llm_clients : dict, optional
            Per-request LLM instances for BYOK. Keys: ``"director"``,
            ``"narrator"``, ``"npc"``.
        player_settings : dict, optional
            Persisted player preferences (e.g. preferred LLM provider/model).
            Merged above cartridge defaults.
        request_overrides : dict, optional
            Per-request config overrides sent by the frontend for this turn.
            Merged above player settings (highest precedence).

        Returns
        -------
        dict
            ``{"narrator_text": str, "yare_delta": dict}``.
        """
        if self._storage is None:
            raise RuntimeError("process_turn requires a storage backend.")

        # 1. Hydrate state from lineage
        if parent_turn_id is not None:
            lineage = self._storage.get_turn_lineage(parent_turn_id)
        else:
            lineage = []

        state = StateHydrator.hydrate_state(
            lineage, 
            self._cartridge.initial_state, 
            self._cartridge.first_message
        )
        state["incoming_interaction"] = interaction
        state["client_messages"].append({"role": "user", "content": user_input})

        # 2. Merge hierarchical config: cartridge < player < request
        cartridge_defaults = {
            "yare_config": self._cartridge.yare_config,
            "prompt_directives": self._cartridge.prompt_directives,
        }
        runtime_config = ConfigMerger.merge(
            cartridge_defaults,
            player_settings or {},
            request_overrides or {},
        )

        # 3. Invoke graph with merged config + BYOK LLMs via RunnableConfig
        config = self._build_runnable_config(runtime_config=runtime_config, llm_clients=llm_clients)

        # Log a filtered version of the state to avoid message noise
        log_state = {k: v for k, v in state.items() if k not in ["client_messages", "agent_messages"]}
        logger.debug("INVOKING GRAPH with hydrated state (filtered): %s", json.dumps(log_state, indent=2, default=str))

        new_state = self._app.invoke(state, config=config)

        # Log a filtered version of the result
        log_new_state = {k: v for k, v in new_state.items() if k not in ["client_messages", "agent_messages"]}
        logger.debug("GRAPH RESULT (filtered): %s", json.dumps(log_new_state, indent=2, default=str))

        # 4. Extract yare_delta from bot_memory changes
        yare_delta = self._extract_delta(
            self._cartridge.initial_state, lineage, new_state
        )

        # 5. Extract narrator response
        narrator_text = self._extract_narrator_response(new_state)

        return {
            "narrator_text": narrator_text,
            "yare_delta": yare_delta,
        }

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------



    def _build_runnable_config(
        self,
        runtime_config: Optional[MnesOSRuntimeConfig] = None,
        llm_clients: Optional[Dict[str, Any]] = None,
    ) -> dict:
        """Build the ``RunnableConfig`` dict carrying static cartridge data and
        the merged :class:`~MnesOS.config.MnesOSRuntimeConfig`.

        ``runtime_config`` LLM role settings are injected under their
        respective keys (``director_llm``, ``narrator_llm``, ``npc_llm``,
        ``embedding_llm``) so that graph nodes can retrieve them from
        ``config["configurable"]``.

        If *llm_clients* is provided the dict is included under
        ``configurable["llm_clients"]`` so graph nodes can pick them up
        for BYOK invocations (per 0005 §4.2).
        """
        # Use runtime_config fields when available; fall back to cartridge defaults.
        # Both MnesOSRuntimeConfig and LoadedCartridge expose yare_config and
        # prompt_directives, so the same attribute access works for both.
        cfg_src = runtime_config if runtime_config is not None else self._cartridge
        configurable: Dict[str, Any] = {
            "yare_config": cfg_src.yare_config,
            "prompt_directives": cfg_src.prompt_directives,
            "lore_path": self._cartridge.lore_path,
            "lore_content": self._cartridge.lore_content,
            "persona_context": self._cartridge.persona_context,
        }
        if runtime_config is not None:
            configurable["director_llm"] = runtime_config.director_llm.model_dump()
            configurable["narrator_llm"] = runtime_config.narrator_llm.model_dump()
            configurable["npc_llm"] = runtime_config.npc_llm.model_dump()
            configurable["embedding_llm"] = runtime_config.embedding_llm.model_dump()
        if llm_clients:
            configurable["llm_clients"] = llm_clients
        return {"configurable": configurable}

    def _compile_graph(self, llm_director, llm_npc, llm_narrator):
        """Delegate graph compilation to the build_graph factory in graph.py."""
        from .graph.tools.lore_batch import VectorLoreSearchService

        lore_content = self._cartridge.lore_content or ""
        if not lore_content and self._cartridge.lore_path:
            try:
                with open(self._cartridge.lore_path, "r", encoding="utf-8") as fh:
                    lore_content = fh.read()
            except (FileNotFoundError, OSError) as exc:
                logger.warning(
                    "Could not read lore file %r: %s. Lore retrieval will be unavailable.",
                    self._cartridge.lore_path,
                    exc,
                )

        lore_service = VectorLoreSearchService(lore_content)

        return build_graph(
            yare_config=self._cartridge.yare_config,
            llm_director=llm_director,
            llm_npc=llm_npc,
            llm_narrator=llm_narrator,
            prompt_directives=self._cartridge.prompt_directives,
            lore_service=lore_service,
        )

    @staticmethod
    def _extract_narrator_response(state: dict) -> str:
        """Return the most recent assistant message from client_messages."""
        for msg in reversed(state.get("client_messages", [])):
            if msg.get("role") == "assistant":
                return msg.get("content", "")
        return ""

    @staticmethod
    def _extract_delta(
        initial_state: dict, lineage: list, new_state: dict
    ) -> dict:
        """Compute the incremental delta produced by this turn.

        The delta is the difference between the hydrated bot_memory
        *before* the turn and the bot_memory *after* the graph ran.
        We store only top-level keys that changed.
        """
        from .storage.hydrator import _deep_merge

        # Reconstruct pre-turn bot_memory
        pre = copy.deepcopy(initial_state)
        for turn in lineage:
            delta = turn.yare_delta
            if isinstance(delta, dict) and delta:
                pre = _deep_merge(pre, delta)

        post = new_state.get("bot_memory", {})

        # Diff: only include keys whose values actually changed
        diff: Dict[str, Any] = {}
        for key in post:
            if key not in pre or pre[key] != post[key]:
                diff[key] = copy.deepcopy(post[key])
        for key in pre:
            if key not in post:
                diff[key] = None
        return diff

cartridge property

The loaded cartridge metadata.

process_turn(user_input, *, interaction=None, parent_turn_id=None, llm_clients=None, player_settings=None, request_overrides=None)

Execute one game turn.

Hydrates state from the turn-log tree, invokes the graph, and returns a result dict. The Orchestrator does NOT save the turn to the DB; the API route handles that.

Parameters

user_input : str The player's raw text input. parent_turn_id : str, optional ID of the previous turn. If None, assumes a new game starting at Turn 0. llm_clients : dict, optional Per-request LLM instances for BYOK. Keys: "director", "narrator", "npc". player_settings : dict, optional Persisted player preferences (e.g. preferred LLM provider/model). Merged above cartridge defaults. request_overrides : dict, optional Per-request config overrides sent by the frontend for this turn. Merged above player settings (highest precedence).

Returns

dict {"narrator_text": str, "yare_delta": dict}.

Source code in src/MnesOS/orchestrator.py

def process_turn(
    self,
    user_input: str,
    *,
    interaction: Optional[Dict[str, Any]] = None,
    parent_turn_id: Optional[str] = None,
    llm_clients: Optional[Dict[str, Any]] = None,
    player_settings: Optional[Dict[str, Any]] = None,
    request_overrides: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """
    Execute one game turn.

    Hydrates state from the turn-log tree, invokes the graph, and returns a result dict.
    The Orchestrator does NOT save the turn to the DB; the API route handles that.

    Parameters
    ----------
    user_input : str
        The player's raw text input.
    parent_turn_id : str, optional
        ID of the previous turn. If None, assumes a new game starting at Turn 0.
    llm_clients : dict, optional
        Per-request LLM instances for BYOK. Keys: ``"director"``,
        ``"narrator"``, ``"npc"``.
    player_settings : dict, optional
        Persisted player preferences (e.g. preferred LLM provider/model).
        Merged above cartridge defaults.
    request_overrides : dict, optional
        Per-request config overrides sent by the frontend for this turn.
        Merged above player settings (highest precedence).

    Returns
    -------
    dict
        ``{"narrator_text": str, "yare_delta": dict}``.
    """
    if self._storage is None:
        raise RuntimeError("process_turn requires a storage backend.")

    # 1. Hydrate state from lineage
    if parent_turn_id is not None:
        lineage = self._storage.get_turn_lineage(parent_turn_id)
    else:
        lineage = []

    state = StateHydrator.hydrate_state(
        lineage, 
        self._cartridge.initial_state, 
        self._cartridge.first_message
    )
    state["incoming_interaction"] = interaction
    state["client_messages"].append({"role": "user", "content": user_input})

    # 2. Merge hierarchical config: cartridge < player < request
    cartridge_defaults = {
        "yare_config": self._cartridge.yare_config,
        "prompt_directives": self._cartridge.prompt_directives,
    }
    runtime_config = ConfigMerger.merge(
        cartridge_defaults,
        player_settings or {},
        request_overrides or {},
    )

    # 3. Invoke graph with merged config + BYOK LLMs via RunnableConfig
    config = self._build_runnable_config(runtime_config=runtime_config, llm_clients=llm_clients)

    # Log a filtered version of the state to avoid message noise
    log_state = {k: v for k, v in state.items() if k not in ["client_messages", "agent_messages"]}
    logger.debug("INVOKING GRAPH with hydrated state (filtered): %s", json.dumps(log_state, indent=2, default=str))

    new_state = self._app.invoke(state, config=config)

    # Log a filtered version of the result
    log_new_state = {k: v for k, v in new_state.items() if k not in ["client_messages", "agent_messages"]}
    logger.debug("GRAPH RESULT (filtered): %s", json.dumps(log_new_state, indent=2, default=str))

    # 4. Extract yare_delta from bot_memory changes
    yare_delta = self._extract_delta(
        self._cartridge.initial_state, lineage, new_state
    )

    # 5. Extract narrator response
    narrator_text = self._extract_narrator_response(new_state)

    return {
        "narrator_text": narrator_text,
        "yare_delta": yare_delta,
    }

VectorLoreStore

A Vector RAG system for lore retrieval. Chunks Markdown content and uses a local vector similarity search (TF-IDF based for sandbox purity).

Source code in src/MnesOS/context.py

class VectorLoreStore:
    """
    A Vector RAG system for lore retrieval.
    Chunks Markdown content and uses a local vector similarity search (TF-IDF based for sandbox purity).
    """

    def __init__(self, lore_content: str):
        self.chunks: List[str] = self._chunk_lore(lore_content)
        self.vocab: Dict[str, int] = {}
        self.vectors: List[List[float]] = []
        self._build_index()

    def _chunk_lore(self, content: str) -> List[str]:
        """Segments lore into meaningful chunks based on headers."""
        pattern = r'(^#{1,3}\s+.*$)'
        parts = re.split(pattern, content, flags=re.MULTILINE)

        chunks = []
        for i in range(1, len(parts), 2):
            header = parts[i].strip()
            body = parts[i+1].strip() if i+1 < len(parts) else ""
            chunks.append(f"{header}\n{body}")
        return chunks

    def _tokenize(self, text: str) -> List[str]:
        return re.findall(r'\w+', text.lower())

    def _get_vector(self, tokens: List[str]) -> List[float]:
        """Simple frequency-based vectorization."""
        vec = [0.0] * len(self.vocab)
        for t in tokens:
            if t in self.vocab:
                vec[self.vocab[t]] += 1.0

        # Normalize
        magnitude = math.sqrt(sum(v*v for v in vec))
        if magnitude > 0:
            vec = [v / magnitude for v in vec]
        return vec

    def _build_index(self):
        """Creates the vocabulary and chunk vectors."""
        # Build Vocab
        word_idx = 0
        for chunk in self.chunks:
            for token in self._tokenize(chunk):
                if token not in self.vocab:
                    self.vocab[token] = word_idx
                    word_idx += 1

        # Build vectors
        for chunk in self.chunks:
            self.vectors.append(self._get_vector(self._tokenize(chunk)))

    def query(self, query_text: str, top_k: int = 2) -> str:
        """Retrieves top_k relevant lore chunks using Cosine Similarity."""
        query_vec = self._get_vector(self._tokenize(query_text))

        scores: List[Tuple[float, int]] = []
        for i, doc_vec in enumerate(self.vectors):
            # Cosine Similarity (dot product since vectors are normalized)
            score = sum(q * d for q, d in zip(query_vec, doc_vec))
            scores.append((score, i))

        # Sort by score descending
        scores.sort(key=lambda x: x[0], reverse=True)

        results = [self.chunks[idx] for score, idx in scores[:top_k] if score > 0]
        return "\n\n---\n\n".join(results)

    @classmethod
    def from_file(cls, filepath: str):
        with open(filepath, 'r') as f:
            return cls(f.read())

query(query_text, top_k=2)

Retrieves top_k relevant lore chunks using Cosine Similarity.

Source code in src/MnesOS/context.py

def query(self, query_text: str, top_k: int = 2) -> str:
    """Retrieves top_k relevant lore chunks using Cosine Similarity."""
    query_vec = self._get_vector(self._tokenize(query_text))

    scores: List[Tuple[float, int]] = []
    for i, doc_vec in enumerate(self.vectors):
        # Cosine Similarity (dot product since vectors are normalized)
        score = sum(q * d for q, d in zip(query_vec, doc_vec))
        scores.append((score, i))

    # Sort by score descending
    scores.sort(key=lambda x: x[0], reverse=True)

    results = [self.chunks[idx] for score, idx in scores[:top_k] if score > 0]
    return "\n\n---\n\n".join(results)

YAREInterpreter

A secure, non-cyclic interpreter for the YAML Agentic Rules Engine (YARE). Handles deterministic state mutations and logic evaluation.

Modularized into store, evaluator, and actions.

Source code in src/MnesOS/interpreter/__init__.py

class YAREInterpreter:
    """
    A secure, non-cyclic interpreter for the YAML Agentic Rules Engine (YARE).
    Handles deterministic state mutations and logic evaluation.

    Modularized into store, evaluator, and actions.
    """

    def __init__(self, config: Dict[str, Any], state: Dict[str, Any]):
        self._config = config
        self.state = state
        self.notes: List[str] = []
        self.temp: Dict[str, Any] = {}
        self.call_depth = 0
        self.max_call_depth = 10

        # Internal modules
        self.store = InterpreterStore(config, state)
        self.store.temp = self.temp  # Link temp storage
        self.evaluator = YAREEvaluator(self.store)
        self.actions = InterpreterActions(self)


    def evaluate(self, expr: Any, context: Dict[str, Any] = None) -> Any:
        return self.evaluator.evaluate(expr, context)

    def run_event(self, event_name: str, inputs: Dict[str, Any] = None):
        if self.call_depth > self.max_call_depth:
            self.notes.append("SYSTEM: Max event call depth reached. Halting.")
            return

        event = self.config.get("events", {}).get(event_name)
        if not event: return

        self.call_depth += 1

        coerced = dict(inputs or {})
        input_schema = event.get("inputs", {})
        if isinstance(input_schema, dict):
            for key, spec in input_schema.items():
                if key in coerced:
                    coerce_fn = self.store._TYPE_COERCIONS.get(spec.get("type", ""))
                    if coerce_fn is not None:
                        try:
                            coerced[key] = coerce_fn(coerced[key])
                        except (ValueError, TypeError):
                            pass
                    # Enum enforcement
                    if "enum" in spec:
                        allowed = [str(v).lower() for v in spec["enum"]]
                        val = coerced[key]
                        if isinstance(val, str): val = val.lower()
                        if val in allowed:
                            coerced[key] = val
                        elif "default" in spec:
                            coerced[key] = spec["default"]
                elif "default" in spec:
                    coerced[key] = spec["default"]

        context = {
            "state": self.state,
            "temp": self.temp,
            "inputs": coerced,
            "macros": self.config.get("macros", {}),
        }

        for step in event.get("steps", []):
            self.actions.execute_step(step, context)
        self.call_depth -= 1

    def _execute_step(self, step: Dict[str, Any], context: Dict[str, Any] = None):
        """Internal bridge for backward compatibility with tests/internal calls."""
        return self.actions.execute_step(step, context or {})

    def _to_numeric(self, value: Any) -> Any:
        return self.store.to_numeric(value)

    def _get_path(self, path: str) -> Any:
        return self.store.get_path(path)

    def _set_path(self, path: str, value: Any):
        return self.store.set_path(path, value)

    def _dict_depth(self, d: Any) -> int:
        return self.store.dict_depth(d)

    def _eval_node(self, node: Any, context: Dict[str, Any] = None) -> Any:
        return self.evaluator._eval_node(node, context or {})

    @property
    def config(self):
        return self._config

    @config.setter
    def config(self, value):
        self._config = value
        if hasattr(self, 'store'):
            self.store.config = value

build_graph(yare_config, llm_director=None, llm_npc=None, llm_narrator=None, prompt_directives=None, lore_service=None)

Build and compile a LangGraph for the given YARE config and LLM instances.

Static cartridge data that was formerly carried in GameState is now closed over at build time (for tools) or passed at invoke time via RunnableConfig["configurable"] (for nodes).

When lore_service is provided the multi_lore_lookup batch RAG tool is injected into the Director's dynamic tool list so the Director can actively retrieve lore on demand. The old Lore pre-node is omitted.

Source code in src/MnesOS/graph/factory.py

def build_graph(
    yare_config: Dict[str, Any],
    llm_director=None,
    llm_npc=None,
    llm_narrator=None,
    prompt_directives: Dict[str, str] | None = None,
    lore_service: Optional[LoreSearchService] = None,
):
    """
    Build and compile a LangGraph for the given YARE config and LLM instances.

    Static cartridge data that was formerly carried in ``GameState`` is now
    closed over at build time (for tools) or passed at invoke time via
    ``RunnableConfig["configurable"]`` (for nodes).

    When *lore_service* is provided the ``multi_lore_lookup`` batch RAG tool
    is injected into the Director's dynamic tool list so the Director can
    actively retrieve lore on demand.  The old ``Lore`` pre-node is omitted.
    """
    prompt_directives = prompt_directives or {}
    dynamic_tools = build_yare_event_tools(yare_config)
    dynamic_tools.append(advance_game_time)

    if llm_npc is not None:
        dynamic_tools = dynamic_tools + [
            build_npc_intent_tool(llm_npc, yare_config=yare_config, prompt_directives=prompt_directives)
        ]

    if lore_service is not None:
        dynamic_tools = dynamic_tools + [build_multi_lore_lookup_tool(lore_service)]

    graph = StateGraph(GameState)

    graph.add_node("ResetAgentMessages", reset_agent_messages_node)
    graph.add_node("CycleTick", cycle_tick_node)
    graph.add_node("MinigameInput", minigame_input_node)
    graph.add_node("Director", functools.partial(director_node, llm=llm_director, tools=dynamic_tools))
    graph.add_node("MinigameOutput", functools.partial(minigame_output_node, llm=llm_director))
    graph.add_node("PreTools", pre_tools_node)

    if dynamic_tools:
        graph.add_node("Tools", ToolNode(dynamic_tools, messages_key="agent_messages"))
    else:
        graph.add_node("Tools", lambda state: state)

    graph.add_node("PostTools", post_tools_node)
    graph.add_node("Narrator", functools.partial(narrator_node, llm=llm_narrator))
    graph.add_node("CleanupAgentMessages", cleanup_agent_messages_node)

    graph.set_entry_point("ResetAgentMessages")
    graph.add_edge("ResetAgentMessages", "CycleTick")
    graph.add_edge("CycleTick", "MinigameInput")
    graph.add_edge("MinigameInput", "Director")

    graph.add_conditional_edges(
        "Director", route_director, {"PreTools": "PreTools", "Narrator": "Narrator"}
    )
    graph.add_edge("PreTools", "Tools")
    graph.add_edge("Tools", "PostTools")
    graph.add_conditional_edges(
        "PostTools", route_rules, {"Director": "Director", "MinigameOutput": "MinigameOutput"}
    )

    graph.add_edge("MinigameOutput", "Narrator")

    graph.add_edge("Narrator", "CleanupAgentMessages")
    graph.add_edge("CleanupAgentMessages", END)

    return graph.compile()