feat: Pip the Familiar — behavioral state machine (#367)

Co-authored-by: Kimi Agent <kimi@timmy.local>
Co-committed-by: Kimi Agent <kimi@timmy.local>

src/timmy/familiar.py

@@ -0,0 +1,263 @@
+"""Pip the Familiar — a creature with its own small mind.
+
+Pip is a glowing sprite who lives in the Workshop independently of Timmy.
+He has a behavioral state machine that makes the room feel alive:
+
+    SLEEPING → WAKING → WANDERING → INVESTIGATING → BORED → SLEEPING
+
+Special states triggered by Timmy's cognitive signals:
+    ALERT    — confidence drops below 0.3
+    PLAYFUL  — Timmy is amused / energized
+    HIDING   — unknown visitor + Timmy uncertain
+
+The backend tracks Pip's *logical* state; the browser handles movement
+interpolation and particle rendering.
+"""
+
+import logging
+import random
+import time
+from dataclasses import asdict, dataclass, field
+from enum import StrEnum
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# States
+# ---------------------------------------------------------------------------
+
+
+class PipState(StrEnum):
+    """Pip's behavioral states."""
+
+    SLEEPING = "sleeping"
+    WAKING = "waking"
+    WANDERING = "wandering"
+    INVESTIGATING = "investigating"
+    BORED = "bored"
+    # Special states
+    ALERT = "alert"
+    PLAYFUL = "playful"
+    HIDING = "hiding"
+
+
+# States from which Pip can be interrupted by special triggers
+_INTERRUPTIBLE = frozenset(
+    {
+        PipState.SLEEPING,
+        PipState.WANDERING,
+        PipState.BORED,
+        PipState.WAKING,
+    }
+)
+
+# How long each state lasts before auto-transitioning (seconds)
+_STATE_DURATIONS: dict[PipState, tuple[float, float]] = {
+    PipState.SLEEPING: (120.0, 300.0),  # 2-5 min
+    PipState.WAKING: (1.5, 2.5),
+    PipState.WANDERING: (15.0, 45.0),
+    PipState.INVESTIGATING: (8.0, 12.0),
+    PipState.BORED: (20.0, 40.0),
+    PipState.ALERT: (10.0, 20.0),
+    PipState.PLAYFUL: (8.0, 15.0),
+    PipState.HIDING: (15.0, 30.0),
+}
+
+# Default position near the fireplace
+_FIREPLACE_POS = (2.1, 0.5, -1.3)
+
+
+# ---------------------------------------------------------------------------
+# Schema
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class PipSnapshot:
+    """Serialisable snapshot of Pip's current state."""
+
+    name: str = "Pip"
+    state: str = "sleeping"
+    position: tuple[float, float, float] = _FIREPLACE_POS
+    mood_mirror: str = "calm"
+    since: float = field(default_factory=time.monotonic)
+
+    def to_dict(self) -> dict:
+        """Public dict for API / WebSocket / state file consumers."""
+        d = asdict(self)
+        d["position"] = list(d["position"])
+        # Convert monotonic timestamp to duration
+        d["state_duration_s"] = round(time.monotonic() - d.pop("since"), 1)
+        return d
+
+
+# ---------------------------------------------------------------------------
+# Familiar
+# ---------------------------------------------------------------------------
+
+
+class Familiar:
+    """Pip's behavioral AI — a tiny state machine driven by events and time.
+
+    Usage::
+
+        pip_familiar.on_event("visitor_entered")
+        pip_familiar.on_mood_change("energized")
+        state = pip_familiar.tick()  # call periodically
+    """
+
+    def __init__(self) -> None:
+        self._state = PipState.SLEEPING
+        self._entered_at = time.monotonic()
+        self._duration = random.uniform(*_STATE_DURATIONS[PipState.SLEEPING])
+        self._mood_mirror = "calm"
+        self._pending_mood: str | None = None
+        self._mood_change_at: float = 0.0
+        self._position = _FIREPLACE_POS
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    @property
+    def state(self) -> PipState:
+        return self._state
+
+    @property
+    def mood_mirror(self) -> str:
+        return self._mood_mirror
+
+    def snapshot(self) -> PipSnapshot:
+        """Current state as a serialisable snapshot."""
+        return PipSnapshot(
+            state=self._state.value,
+            position=self._position,
+            mood_mirror=self._mood_mirror,
+            since=self._entered_at,
+        )
+
+    def tick(self, now: float | None = None) -> PipState:
+        """Advance the state machine. Call periodically (e.g. every second).
+
+        Returns the (possibly new) state.
+        """
+        now = now if now is not None else time.monotonic()
+
+        # Apply delayed mood mirror (3-second lag)
+        if self._pending_mood and now >= self._mood_change_at:
+            self._mood_mirror = self._pending_mood
+            self._pending_mood = None
+
+        # Check if current state has expired
+        elapsed = now - self._entered_at
+        if elapsed < self._duration:
+            return self._state
+
+        # Auto-transition
+        next_state = self._next_state()
+        self._transition(next_state, now)
+        return self._state
+
+    def on_event(self, event: str, now: float | None = None) -> PipState:
+        """React to a Workshop event.
+
+        Supported events:
+            visitor_entered, visitor_spoke, loud_event, scroll_knocked
+        """
+        now = now if now is not None else time.monotonic()
+
+        if event == "visitor_entered" and self._state in _INTERRUPTIBLE:
+            if self._state == PipState.SLEEPING:
+                self._transition(PipState.WAKING, now)
+            else:
+                self._transition(PipState.INVESTIGATING, now)
+
+        elif event == "visitor_spoke":
+            if self._state in (PipState.WANDERING, PipState.WAKING):
+                self._transition(PipState.INVESTIGATING, now)
+
+        elif event == "loud_event":
+            if self._state == PipState.SLEEPING:
+                self._transition(PipState.WAKING, now)
+
+        return self._state
+
+    def on_mood_change(
+        self,
+        timmy_mood: str,
+        confidence: float = 0.5,
+        now: float | None = None,
+    ) -> PipState:
+        """Mirror Timmy's mood with a 3-second delay.
+
+        Special states triggered by mood + confidence:
+            - confidence < 0.3 → ALERT (bristles, particles go red-gold)
+            - mood == "energized" → PLAYFUL (figure-8s around crystal ball)
+            - mood == "hesitant" + confidence < 0.4 → HIDING
+        """
+        now = now if now is not None else time.monotonic()
+
+        # Schedule mood mirror with 3s delay
+        self._pending_mood = timmy_mood
+        self._mood_change_at = now + 3.0
+
+        # Special state triggers (immediate)
+        if confidence < 0.3 and self._state in _INTERRUPTIBLE:
+            self._transition(PipState.ALERT, now)
+        elif timmy_mood == "energized" and self._state in _INTERRUPTIBLE:
+            self._transition(PipState.PLAYFUL, now)
+        elif timmy_mood == "hesitant" and confidence < 0.4 and self._state in _INTERRUPTIBLE:
+            self._transition(PipState.HIDING, now)
+
+        return self._state
+
+    # ------------------------------------------------------------------
+    # Internals
+    # ------------------------------------------------------------------
+
+    def _transition(self, new_state: PipState, now: float) -> None:
+        """Move to a new state."""
+        old = self._state
+        self._state = new_state
+        self._entered_at = now
+        self._duration = random.uniform(*_STATE_DURATIONS[new_state])
+        self._position = self._position_for(new_state)
+        logger.debug("Pip: %s → %s", old.value, new_state.value)
+
+    def _next_state(self) -> PipState:
+        """Determine the natural next state after the current one expires."""
+        transitions: dict[PipState, PipState] = {
+            PipState.SLEEPING: PipState.WAKING,
+            PipState.WAKING: PipState.WANDERING,
+            PipState.WANDERING: PipState.BORED,
+            PipState.INVESTIGATING: PipState.BORED,
+            PipState.BORED: PipState.SLEEPING,
+            # Special states return to wandering
+            PipState.ALERT: PipState.WANDERING,
+            PipState.PLAYFUL: PipState.WANDERING,
+            PipState.HIDING: PipState.WAKING,
+        }
+        return transitions.get(self._state, PipState.SLEEPING)
+
+    def _position_for(self, state: PipState) -> tuple[float, float, float]:
+        """Approximate position hint for a given state.
+
+        The browser interpolates smoothly; these are target anchors.
+        """
+        if state in (PipState.SLEEPING, PipState.BORED):
+            return _FIREPLACE_POS
+        if state == PipState.HIDING:
+            return (0.5, 0.3, -2.0)  # Behind the desk
+        if state == PipState.PLAYFUL:
+            return (1.0, 1.2, 0.0)  # Near the crystal ball
+        # Wandering / investigating / waking — random room position
+        return (
+            random.uniform(-1.0, 3.0),
+            random.uniform(0.5, 1.5),
+            random.uniform(-2.0, 1.0),
+        )
+
+
+# Module-level singleton
+pip_familiar = Familiar()

tests/timmy/test_familiar.py

@@ -0,0 +1,198 @@
+"""Tests for Pip the Familiar — behavioral state machine."""
+
+import time
+
+import pytest
+
+from timmy.familiar import _FIREPLACE_POS, Familiar, PipState
+
+
+@pytest.fixture
+def pip():
+    return Familiar()
+
+
+class TestInitialState:
+    def test_starts_sleeping(self, pip):
+        assert pip.state == PipState.SLEEPING
+
+    def test_starts_calm(self, pip):
+        assert pip.mood_mirror == "calm"
+
+    def test_snapshot_returns_dict(self, pip):
+        snap = pip.snapshot().to_dict()
+        assert snap["name"] == "Pip"
+        assert snap["state"] == "sleeping"
+        assert snap["position"] == list(_FIREPLACE_POS)
+        assert snap["mood_mirror"] == "calm"
+        assert "state_duration_s" in snap
+
+
+class TestAutoTransitions:
+    def test_sleeping_to_waking_after_duration(self, pip):
+        now = time.monotonic()
+        # Force a short duration
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        result = pip.tick(now=now)
+        assert result == PipState.WAKING
+
+    def test_waking_to_wandering(self, pip):
+        now = time.monotonic()
+        pip._state = PipState.WAKING
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        pip.tick(now=now)
+        assert pip.state == PipState.WANDERING
+
+    def test_wandering_to_bored(self, pip):
+        now = time.monotonic()
+        pip._state = PipState.WANDERING
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        pip.tick(now=now)
+        assert pip.state == PipState.BORED
+
+    def test_bored_to_sleeping(self, pip):
+        now = time.monotonic()
+        pip._state = PipState.BORED
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        pip.tick(now=now)
+        assert pip.state == PipState.SLEEPING
+
+    def test_full_cycle(self, pip):
+        """Pip cycles: SLEEPING → WAKING → WANDERING → BORED → SLEEPING."""
+        now = time.monotonic()
+        expected = [
+            PipState.WAKING,
+            PipState.WANDERING,
+            PipState.BORED,
+            PipState.SLEEPING,
+        ]
+        for expected_state in expected:
+            pip._duration = 0.1
+            pip._entered_at = now - 1.0
+            pip.tick(now=now)
+            assert pip.state == expected_state
+            now += 0.01
+
+    def test_no_transition_before_duration(self, pip):
+        now = time.monotonic()
+        pip._duration = 100.0
+        pip._entered_at = now
+        pip.tick(now=now + 1.0)
+        assert pip.state == PipState.SLEEPING
+
+
+class TestEventReactions:
+    def test_visitor_entered_wakes_pip(self, pip):
+        assert pip.state == PipState.SLEEPING
+        pip.on_event("visitor_entered")
+        assert pip.state == PipState.WAKING
+
+    def test_visitor_entered_while_wandering_investigates(self, pip):
+        pip._state = PipState.WANDERING
+        pip.on_event("visitor_entered")
+        assert pip.state == PipState.INVESTIGATING
+
+    def test_visitor_spoke_while_wandering_investigates(self, pip):
+        pip._state = PipState.WANDERING
+        pip.on_event("visitor_spoke")
+        assert pip.state == PipState.INVESTIGATING
+
+    def test_loud_event_wakes_sleeping_pip(self, pip):
+        pip.on_event("loud_event")
+        assert pip.state == PipState.WAKING
+
+    def test_unknown_event_no_change(self, pip):
+        pip.on_event("unknown_event")
+        assert pip.state == PipState.SLEEPING
+
+    def test_investigating_expires_to_bored(self, pip):
+        now = time.monotonic()
+        pip._state = PipState.INVESTIGATING
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        pip.tick(now=now)
+        assert pip.state == PipState.BORED
+
+
+class TestMoodMirroring:
+    def test_mood_mirrors_with_delay(self, pip):
+        now = time.monotonic()
+        pip.on_mood_change("curious", confidence=0.6, now=now)
+        # Before delay — still calm
+        pip.tick(now=now + 1.0)
+        assert pip.mood_mirror == "calm"
+        # After 3s delay — mirrors
+        pip.tick(now=now + 4.0)
+        assert pip.mood_mirror == "curious"
+
+    def test_low_confidence_triggers_alert(self, pip):
+        pip.on_mood_change("hesitant", confidence=0.2)
+        assert pip.state == PipState.ALERT
+
+    def test_energized_triggers_playful(self, pip):
+        pip.on_mood_change("energized", confidence=0.7)
+        assert pip.state == PipState.PLAYFUL
+
+    def test_hesitant_low_confidence_triggers_hiding(self, pip):
+        pip.on_mood_change("hesitant", confidence=0.35)
+        assert pip.state == PipState.HIDING
+
+    def test_special_state_not_from_non_interruptible(self, pip):
+        pip._state = PipState.INVESTIGATING
+        pip.on_mood_change("energized", confidence=0.7)
+        # INVESTIGATING is not interruptible
+        assert pip.state == PipState.INVESTIGATING
+
+
+class TestSpecialStateRecovery:
+    def test_alert_returns_to_wandering(self, pip):
+        now = time.monotonic()
+        pip._state = PipState.ALERT
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        pip.tick(now=now)
+        assert pip.state == PipState.WANDERING
+
+    def test_playful_returns_to_wandering(self, pip):
+        now = time.monotonic()
+        pip._state = PipState.PLAYFUL
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        pip.tick(now=now)
+        assert pip.state == PipState.WANDERING
+
+    def test_hiding_returns_to_waking(self, pip):
+        now = time.monotonic()
+        pip._state = PipState.HIDING
+        pip._duration = 1.0
+        pip._entered_at = now - 2.0
+        pip.tick(now=now)
+        assert pip.state == PipState.WAKING
+
+
+class TestPositionHints:
+    def test_sleeping_near_fireplace(self, pip):
+        snap = pip.snapshot()
+        assert snap.position == _FIREPLACE_POS
+
+    def test_hiding_behind_desk(self, pip):
+        pip.on_mood_change("hesitant", confidence=0.35)
+        assert pip.state == PipState.HIDING
+        snap = pip.snapshot()
+        assert snap.position == (0.5, 0.3, -2.0)
+
+    def test_playful_near_crystal_ball(self, pip):
+        pip.on_mood_change("energized", confidence=0.7)
+        snap = pip.snapshot()
+        assert snap.position == (1.0, 1.2, 0.0)
+
+
+class TestSingleton:
+    def test_module_singleton_exists(self):
+        from timmy.familiar import pip_familiar
+
+        assert isinstance(pip_familiar, Familiar)