hermes-agent/plugins/digital-pet/state.py

"""Deterministic state machine and persistence for the Hermes digital pet."""
from __future__ import annotations

import json
import os
from dataclasses import dataclass, asdict, fields
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any, Dict, Optional

try:
    from hermes_constants import get_hermes_home
except Exception:
    def get_hermes_home() -> Path:  # type: ignore
        return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))

MODES = {"hidden", "idle", "reacting", "prompting", "sleeping", "waking"}
MOODS = {"calm", "curious", "pleased", "tired"}
DEFAULT_NAME = "Signal"
MESSAGE_COOLDOWN_SECONDS = 120

CLICK_LINES = {
    "calm": "I am watching the station.",
    "curious": "Interesting. The recordings will tell us.",
    "pleased": "Good. Procedure completed.",
    "tired": "Quiet mode would help.",
}
PROMPT_LINES = [
    "Hydration check. Then back to the station.",
    "Long watch. Stretch your hands.",
    "No alarms. Good moment to breathe.",
    "The station is quiet. Proceed carefully.",
]


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


def iso(dt: datetime) -> str:
    return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")


def parse_iso(value: str | None) -> Optional[datetime]:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except Exception:
        return None


@dataclass
class PetState:
    enabled: bool = False
    mode: str = "hidden"
    last_mode: str = "hidden"
    name: str = DEFAULT_NAME
    mood: str = "calm"
    energy: int = 72
    last_interaction_at: Optional[str] = None
    prompt_cooldown_until: Optional[str] = None
    message: Optional[str] = None
    message_expires_at: Optional[str] = None
    revision: int = 0

    def normalised(self) -> "PetState":
        if self.mode not in MODES:
            self.mode = "hidden"
        if self.last_mode not in MODES:
            self.last_mode = self.mode
        if self.mood not in MOODS:
            self.mood = "calm"
        self.energy = max(0, min(100, int(self.energy)))
        self.enabled = bool(self.enabled)
        return self

    def public(self) -> Dict[str, Any]:
        data = asdict(self.normalised())
        data["modes"] = sorted(MODES)
        data["moods"] = sorted(MOODS)
        return data


def state_path() -> Path:
    override = os.environ.get("HERMES_DIGITAL_PET_STATE")
    if override:
        return Path(override).expanduser()
    return get_hermes_home() / "digital-pet" / "state.json"


def default_state() -> PetState:
    return PetState()


def load_state() -> PetState:
    path = state_path()
    if not path.exists():
        return default_state()
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return default_state()
    allowed = {field.name for field in fields(PetState)}
    return PetState(**{k: v for k, v in raw.items() if k in allowed}).normalised()


def save_state(state: PetState) -> PetState:
    state.normalised()
    path = state_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(state.public(), indent=2, sort_keys=True), encoding="utf-8")
    tmp.replace(path)
    return state


def _set_message(state: PetState, text: Optional[str], now: datetime, seconds: int = 5) -> None:
    state.message = text
    state.message_expires_at = iso(now + timedelta(seconds=seconds)) if text else None
    if text:
        state.prompt_cooldown_until = iso(now + timedelta(seconds=MESSAGE_COOLDOWN_SECONDS))


def _can_autospeak(state: PetState, now: datetime) -> bool:
    until = parse_iso(state.prompt_cooldown_until)
    return until is None or now >= until


def _recover_energy(state: PetState, amount: int) -> None:
    state.energy = max(0, min(100, state.energy + amount))
    if state.energy < 25:
        state.mood = "tired"
    elif state.mood == "tired" and state.energy >= 40:
        state.mood = "calm"


def apply_event(
    event_type: str,
    *,
    metadata: Optional[Dict[str, Any]] = None,
    state: Optional[PetState] = None,
    now: Optional[datetime] = None,
    persist: bool = False,
) -> PetState:
    now = now or utc_now()
    state = (state or load_state()).normalised()
    before = state.public().copy()
    metadata = metadata or {}

    if event_type == "summon":
        state.enabled = True
        state.last_mode = "idle"
        state.mode = "idle"
        state.last_interaction_at = iso(now)
        _set_message(state, metadata.get("message") or "Station companion online.", now)
    elif event_type == "hide":
        state.enabled = False
        state.last_mode = state.mode if state.mode != "hidden" else state.last_mode
        state.mode = "hidden"
        state.last_interaction_at = iso(now)
        _set_message(state, None, now)
    elif event_type == "hover_hold":
        if state.mode in {"idle", "reacting", "prompting", "waking"}:
            state.enabled = True
            state.last_mode = state.mode
            state.mode = "sleeping"
            state.last_interaction_at = iso(now)
            _recover_energy(state, 2)
            _set_message(state, "Going quiet.", now)
    elif event_type == "sleep":
        if state.mode != "hidden":
            state.enabled = True
            state.last_mode = state.mode
            state.mode = "sleeping"
            state.last_interaction_at = iso(now)
            _recover_energy(state, 2)
            _set_message(state, "Sleeping.", now)
    elif event_type == "wake":
        if state.mode == "sleeping":
            state.enabled = True
            state.last_mode = "sleeping"
            state.mode = "waking"
            state.last_interaction_at = iso(now)
            _recover_energy(state, 1)
            _set_message(state, "Back online.", now)
        elif state.mode == "hidden":
            state.enabled = True
            state.last_mode = "hidden"
            state.mode = "idle"
            state.last_interaction_at = iso(now)
            _set_message(state, "Quiet watch resumed.", now)
    elif event_type == "wake_complete":
        if state.mode == "waking":
            state.last_mode = "waking"
            state.mode = "idle"
            _set_message(state, None, now)
    elif event_type == "click":
        state.enabled = True
        state.last_interaction_at = iso(now)
        if state.mode == "sleeping":
            state.last_mode = "sleeping"
            state.mode = "waking"
            _recover_energy(state, 1)
            _set_message(state, "Back online.", now)
        elif state.mode == "hidden":
            state.last_mode = "hidden"
            state.mode = "idle"
            _set_message(state, "Station companion online.", now)
        else:
            state.last_mode = state.mode
            state.mode = "reacting"
            state.energy = max(0, state.energy - 4)
            if state.energy < 25:
                state.mood = "tired"
            elif state.energy > 65:
                state.mood = "pleased"
            else:
                state.mood = "curious"
            _set_message(state, CLICK_LINES[state.mood], now)
    elif event_type == "reaction_complete":
        if state.mode == "reacting":
            state.last_mode = "reacting"
            state.mode = "idle"
            _set_message(state, None, now)
    elif event_type == "idle_tick":
        if state.mode in {"sleeping", "hidden"}:
            _recover_energy(state, 1)
    elif event_type == "prompt_tick":
        if state.mode == "idle" and state.enabled and _can_autospeak(state, now):
            state.last_mode = "idle"
            state.mode = "prompting"
            _set_message(state, PROMPT_LINES[state.revision % len(PROMPT_LINES)], now, seconds=7)
    elif event_type == "prompt_complete":
        if state.mode == "prompting":
            state.last_mode = "prompting"
            state.mode = "idle"
            _set_message(state, None, now)
    elif event_type in {"tool_call_succeeded", "tool_call_failed", "session_started"}:
        if state.mode == "idle" and state.enabled and _can_autospeak(state, now):
            if (state.revision + 1) % 4 == 0:
                state.last_mode = "idle"
                state.mode = "reacting"
                if event_type == "tool_call_failed":
                    state.mood = "curious"
                    _set_message(state, "Interesting. The recordings will tell us.", now)
                elif event_type == "session_started":
                    state.mood = "calm"
                    _set_message(state, "Quiet watch resumed.", now)
                else:
                    state.mood = "pleased"
                    _set_message(state, "Good. Procedure completed.", now)

    if state.public() != before:
        state.revision += 1
    if persist:
        save_state(state)
    return state.normalised()