orchestrata_ultimate.core.state.state_manager

Reference › orchestrata_ultimate › core › state › state_manager

The persistent state manager, implementing IStateManager (StateManagerProtocol) for volks, threads and system configuration. StateManager keeps state in namespaces. A key is split at its first dot, so session.mode addresses key mode in namespace session, and a key with no dot goes to the namespace default. get_state, set_state, update_state (with an updater), delete_state, create_namespace and list_namespaces implement the protocol, and initialize, start, stop, get_health and get_metadata implement the component lifecycle (metadata id orchestrata-state-manager, capabilities state persistence, namespaces and transactions).

Persistence is one JSON file per namespace inside a storage directory taken from the storage_path config key (default a directory named orchestrata_state under the system temporary directory, created at construction). set_state updates memory at once and schedules _persist_namespace as a background task; a failure is logged as a warning and does not reach the caller. _load_state reads the *.json files back, logging and skipping any file it cannot parse. The kernel's CLI main module imports it through the state package.

At a glance

Class diagram

Dependency diagram

API

Classes

StateManager

Bases: IStateManager · Line 25

Persistent state management.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/state/state_manager.py

"""
State Manager - Persistent state management

Manages state for Volks, threads, and system configuration
with support for multiple storage backends.
"""

import asyncio
import json
import logging
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.protocols import (
    ComponentMetadata,
    ComponentType,
    HealthStatus,
    IStateManager,
)

logger = logging.getLogger(__name__)


class StateManager(IStateManager):
    """
    Persistent state management.

    Manages state for Volks, threads, and system configuration
    with support for multiple storage backends.
    """

    def __init__(self, config: dict[str, Any] | None = None) -> None:
        self._config = config or {}
        self._state: dict[str, Any] = {}
        self._namespaces: dict[str, dict[str, Any]] = {}
        self._running = False

        self._storage_path = self._config.get(
            "storage_path", Path("/tmp/orchestrata_state")
        )
        Path(self._storage_path).mkdir(parents=True, exist_ok=True)

        self._metadata = ComponentMetadata(
            id="orchestrata-state-manager",
            name="StateManager",
            version="0.1.0",
            component_type=ComponentType.STATE_MANAGER,
            description="Persistent state management",
            capabilities=["state_persistence", "namespaces", "transactions"],
        )

        logger.info("StateManager initialized")

    async def initialize(self, config: dict[str, Any]) -> None:
        self._config.update(config)
        self._running = True
        await self._load_state()
        logger.info("StateManager started")

    async def start(self) -> None:
        self._running = True
        logger.info("StateManager operational")

    async def stop(self) -> None:
        await self._save_state()
        self._running = False
        logger.info("StateManager stopped")

    async def get_health(self) -> HealthStatus:
        return HealthStatus.HEALTHY if self._running else HealthStatus.STOPPED

    def get_metadata(self) -> ComponentMetadata:
        return self._metadata

    async def get_state(self, key: str, default: Any = None) -> Any:
        namespace, actual_key = self._parse_key(key)
        if namespace in self._namespaces:
            return self._namespaces[namespace].get(actual_key, default)
        return default

    async def set_state(self, key: str, value: Any) -> None:
        namespace, actual_key = self._parse_key(key)
        if namespace not in self._namespaces:
            self._namespaces[namespace] = {}
        self._namespaces[namespace][actual_key] = value
        task = asyncio.create_task(self._persist_namespace(namespace))
        task.add_done_callback(
            lambda t: (
                logger.warning(
                    "Namespace persistence failed: %s", t.exception(), exc_info=True
                )
                if t.exception() is not None
                else None
            )
        )

    async def update_state(self, key: str, updater: Any) -> Any:
        current = await self.get_state(key, None)
        new_value = updater(current) if current is not None else updater(None)
        await self.set_state(key, new_value)
        return new_value

    async def delete_state(self, key: str) -> None:
        namespace, actual_key = self._parse_key(key)
        if namespace in self._namespaces and actual_key in self._namespaces[namespace]:
            del self._namespaces[namespace][actual_key]
            task = asyncio.create_task(self._persist_namespace(namespace))
            task.add_done_callback(
                lambda t: (
                    logger.warning(
                        "Namespace persistence failed: %s", t.exception(), exc_info=True
                    )
                    if t.exception() is not None
                    else None
                )
            )

    async def create_namespace(self, name: str) -> None:
        if name not in self._namespaces:
            self._namespaces[name] = {}
            logger.debug(f"Created namespace: {name}")

    async def list_namespaces(self) -> list[str]:
        return list(self._namespaces.keys())

    def _parse_key(self, key: str) -> tuple[str, str]:
        if "." in key:
            parts = key.split(".", 1)
            return parts[0], parts[1]
        return "default", key

    async def _persist_namespace(self, namespace: str) -> None:
        if namespace not in self._namespaces:
            return
        try:
            filepath = Path(self._storage_path) / f"{namespace}.json"
            data = {
                "namespace": namespace,
                "updated_at": datetime.now(UTC).isoformat(),
                "data": self._namespaces[namespace],
            }
            with open(filepath, "w") as f:
                json.dump(data, f, indent=2, default=str)
            logger.debug(f"Persisted namespace: {namespace}")
        except Exception as e:
            logger.error(f"Failed to persist namespace {namespace}: {e}")

    async def _load_state(self) -> None:
        try:
            storage_path = Path(self._storage_path)
            if not storage_path.exists():
                return
            for filepath in storage_path.glob("*.json"):
                try:
                    with open(filepath) as f:
                        data = json.load(f)
                    namespace = data.get("namespace", filepath.stem)
                    self._namespaces[namespace] = data.get("data", {})
                    logger.debug(f"Loaded namespace: {namespace}")
                except Exception as e:
                    logger.error(f"Failed to load {filepath}: {e}")
        except Exception as e:
            logger.error(f"Failed to load state: {e}")

    async def _save_state(self) -> None:
        for namespace in self._namespaces:
            await self._persist_namespace(namespace)