"""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()