fix: add cognitive state as observable signal for Matrix avatar (#358)

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

src/timmy/cognitive_state.py

@@ -0,0 +1,248 @@
+"""Observable cognitive state for Timmy.
+
+Tracks Timmy's internal cognitive signals — focus, engagement, mood,
+and active commitments — so external systems (Matrix avatar, dashboard)
+can render observable behaviour.
+
+State persists to ``~/.tower/timmy-state.txt`` alongside the existing
+loop coordination fields.
+
+Schema (YAML-ish in the state file)::
+
+    FOCUS_TOPIC: three-phase loop architecture
+    ENGAGEMENT: deep
+    MOOD: curious
+    CONVERSATION_DEPTH: 42
+    LAST_INITIATIVE: proposed Unsplash API exploration
+    ACTIVE_COMMITMENTS: draft skeleton ticket
+"""
+
+import json
+import logging
+from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+
+from timmy.confidence import estimate_confidence
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Schema
+# ---------------------------------------------------------------------------
+
+ENGAGEMENT_LEVELS = ("idle", "surface", "deep")
+MOOD_VALUES = ("curious", "settled", "hesitant", "energized")
+
+STATE_FILE = Path.home() / ".tower" / "timmy-state.txt"
+
+
+@dataclass
+class CognitiveState:
+    """Observable snapshot of Timmy's cognitive state."""
+
+    focus_topic: str | None = None
+    engagement: str = "idle"  # idle | surface | deep
+    mood: str = "settled"  # curious | settled | hesitant | energized
+    conversation_depth: int = 0
+    last_initiative: str | None = None
+    active_commitments: list[str] = field(default_factory=list)
+
+    # Internal tracking (not written to state file)
+    _confidence_sum: float = field(default=0.0, repr=False)
+    _confidence_count: int = field(default=0, repr=False)
+
+    # ------------------------------------------------------------------
+    # Serialisation helpers
+    # ------------------------------------------------------------------
+
+    def to_dict(self) -> dict:
+        """Public fields only (exclude internal tracking)."""
+        d = asdict(self)
+        d.pop("_confidence_sum", None)
+        d.pop("_confidence_count", None)
+        return d
+
+    def to_state_lines(self) -> list[str]:
+        """Format for ``~/.tower/timmy-state.txt``."""
+        lines = [
+            f"LAST_UPDATED: {datetime.now(UTC).strftime('%Y-%m-%dT%H:%M:%SZ')}",
+            f"FOCUS_TOPIC: {self.focus_topic or 'none'}",
+            f"ENGAGEMENT: {self.engagement}",
+            f"MOOD: {self.mood}",
+            f"CONVERSATION_DEPTH: {self.conversation_depth}",
+        ]
+        if self.last_initiative:
+            lines.append(f"LAST_INITIATIVE: {self.last_initiative}")
+        if self.active_commitments:
+            lines.append(f"ACTIVE_COMMITMENTS: {'; '.join(self.active_commitments)}")
+        return lines
+
+
+# ---------------------------------------------------------------------------
+# Cognitive signal extraction
+# ---------------------------------------------------------------------------
+
+# Keywords that suggest deep engagement
+_DEEP_KEYWORDS = frozenset(
+    {
+        "architecture",
+        "design",
+        "implement",
+        "refactor",
+        "debug",
+        "analyze",
+        "investigate",
+        "deep dive",
+        "explain how",
+        "walk me through",
+        "step by step",
+    }
+)
+
+# Keywords that suggest initiative / commitment
+_COMMITMENT_KEYWORDS = frozenset(
+    {
+        "i will",
+        "i'll",
+        "let me",
+        "i'm going to",
+        "plan to",
+        "commit to",
+        "i propose",
+        "i suggest",
+    }
+)
+
+
+def _infer_engagement(message: str, response: str) -> str:
+    """Classify engagement level from the exchange."""
+    combined = (message + " " + response).lower()
+    if any(kw in combined for kw in _DEEP_KEYWORDS):
+        return "deep"
+    # Short exchanges are surface-level
+    if len(response.split()) < 15:
+        return "surface"
+    return "surface"
+
+
+def _infer_mood(response: str, confidence: float) -> str:
+    """Derive mood from response signals."""
+    lower = response.lower()
+    if confidence < 0.4:
+        return "hesitant"
+    if "!" in response and any(w in lower for w in ("great", "exciting", "love", "awesome")):
+        return "energized"
+    if "?" in response or any(w in lower for w in ("wonder", "interesting", "curious", "hmm")):
+        return "curious"
+    return "settled"
+
+
+def _extract_topic(message: str) -> str | None:
+    """Best-effort topic extraction from the user message.
+
+    Takes the first meaningful clause (up to 60 chars) as a topic label.
+    """
+    text = message.strip()
+    if not text:
+        return None
+    # Strip leading question words
+    for prefix in ("what is ", "how do ", "can you ", "please ", "hey timmy "):
+        if text.lower().startswith(prefix):
+            text = text[len(prefix) :]
+    # Truncate
+    if len(text) > 60:
+        text = text[:57] + "..."
+    return text.strip() or None
+
+
+def _extract_commitments(response: str) -> list[str]:
+    """Pull commitment phrases from Timmy's response."""
+    commitments: list[str] = []
+    lower = response.lower()
+    for kw in _COMMITMENT_KEYWORDS:
+        idx = lower.find(kw)
+        if idx == -1:
+            continue
+        # Grab the rest of the sentence (up to period/newline, max 80 chars)
+        start = idx
+        end = len(lower)
+        for sep in (".", "\n", "!"):
+            pos = lower.find(sep, start)
+            if pos != -1:
+                end = min(end, pos)
+        snippet = response[start : min(end, start + 80)].strip()
+        if snippet:
+            commitments.append(snippet)
+    return commitments[:3]  # Cap at 3
+
+
+# ---------------------------------------------------------------------------
+# Tracker singleton
+# ---------------------------------------------------------------------------
+
+
+class CognitiveTracker:
+    """Maintains and persists Timmy's cognitive state."""
+
+    def __init__(self, state_file: Path | None = None) -> None:
+        self.state = CognitiveState()
+        self._state_file = state_file or STATE_FILE
+
+    def update(self, user_message: str, response: str) -> CognitiveState:
+        """Update cognitive state from a chat exchange.
+
+        Called after each chat round-trip in ``session.py``.
+        """
+        confidence = estimate_confidence(response)
+
+        # Track running confidence average
+        self.state._confidence_sum += confidence
+        self.state._confidence_count += 1
+
+        self.state.conversation_depth += 1
+        self.state.focus_topic = _extract_topic(user_message) or self.state.focus_topic
+        self.state.engagement = _infer_engagement(user_message, response)
+        self.state.mood = _infer_mood(response, confidence)
+
+        # Extract commitments from response
+        new_commitments = _extract_commitments(response)
+        if new_commitments:
+            self.state.last_initiative = new_commitments[0]
+            # Merge, keeping last 5
+            seen = set(self.state.active_commitments)
+            for c in new_commitments:
+                if c not in seen:
+                    self.state.active_commitments.append(c)
+                    seen.add(c)
+            self.state.active_commitments = self.state.active_commitments[-5:]
+
+        # Persist to disk (best-effort)
+        self._write_state_file()
+
+        return self.state
+
+    def get_state(self) -> CognitiveState:
+        """Return current cognitive state."""
+        return self.state
+
+    def reset(self) -> None:
+        """Reset to idle state (e.g. on session reset)."""
+        self.state = CognitiveState()
+        self._write_state_file()
+
+    def _write_state_file(self) -> None:
+        """Persist state to ``~/.tower/timmy-state.txt``."""
+        try:
+            self._state_file.parent.mkdir(parents=True, exist_ok=True)
+            self._state_file.write_text("\n".join(self.state.to_state_lines()) + "\n")
+        except OSError as exc:
+            logger.warning("Failed to write cognitive state: %s", exc)
+
+    def to_json(self) -> str:
+        """Serialise current state as JSON (for API / WebSocket consumers)."""
+        return json.dumps(self.state.to_dict())
+
+
+# Module-level singleton
+cognitive_tracker = CognitiveTracker()

src/timmy/session.py

@@ -13,6 +13,7 @@ import re
 
 import httpx
 
+from timmy.cognitive_state import cognitive_tracker
 from timmy.confidence import estimate_confidence
 from timmy.session_logger import get_session_logger
 
@@ -119,6 +120,9 @@ async def chat(message: str, session_id: str | None = None) -> str:
     # Record Timmy response after getting it
     session_logger.record_message("timmy", response_text, confidence=confidence)
 
+    # Update cognitive state (observable signal for Matrix avatar)
+    cognitive_tracker.update(message, response_text)
+
     # Flush session logs to disk
     session_logger.flush()
 

tests/timmy/test_cognitive_state.py

@@ -0,0 +1,198 @@
+"""Tests for cognitive state tracking in src/timmy/cognitive_state.py."""
+
+from timmy.cognitive_state import (
+    ENGAGEMENT_LEVELS,
+    MOOD_VALUES,
+    CognitiveState,
+    CognitiveTracker,
+    _extract_commitments,
+    _extract_topic,
+    _infer_engagement,
+    _infer_mood,
+)
+
+
+class TestCognitiveState:
+    """Test the CognitiveState dataclass."""
+
+    def test_defaults(self):
+        state = CognitiveState()
+        assert state.focus_topic is None
+        assert state.engagement == "idle"
+        assert state.mood == "settled"
+        assert state.conversation_depth == 0
+        assert state.last_initiative is None
+        assert state.active_commitments == []
+
+    def test_to_dict_excludes_private_fields(self):
+        state = CognitiveState(focus_topic="testing")
+        d = state.to_dict()
+        assert "focus_topic" in d
+        assert "_confidence_sum" not in d
+        assert "_confidence_count" not in d
+
+    def test_to_state_lines_format(self):
+        state = CognitiveState(
+            focus_topic="loop architecture",
+            engagement="deep",
+            mood="curious",
+            conversation_depth=42,
+            last_initiative="proposed refactor",
+            active_commitments=["draft ticket", "review PR"],
+        )
+        lines = state.to_state_lines()
+        text = "\n".join(lines)
+        assert "FOCUS_TOPIC: loop architecture" in text
+        assert "ENGAGEMENT: deep" in text
+        assert "MOOD: curious" in text
+        assert "CONVERSATION_DEPTH: 42" in text
+        assert "LAST_INITIATIVE: proposed refactor" in text
+        assert "ACTIVE_COMMITMENTS: draft ticket; review PR" in text
+
+    def test_to_state_lines_none_topic(self):
+        state = CognitiveState()
+        lines = state.to_state_lines()
+        text = "\n".join(lines)
+        assert "FOCUS_TOPIC: none" in text
+
+    def test_to_state_lines_no_commitments(self):
+        state = CognitiveState()
+        lines = state.to_state_lines()
+        text = "\n".join(lines)
+        assert "ACTIVE_COMMITMENTS" not in text
+
+
+class TestInferEngagement:
+    """Test engagement level inference."""
+
+    def test_deep_keywords(self):
+        assert _infer_engagement("help me debug this", "looking at the stack trace") == "deep"
+
+    def test_architecture_is_deep(self):
+        assert (
+            _infer_engagement("explain the architecture", "the system has three layers") == "deep"
+        )
+
+    def test_short_response_is_surface(self):
+        assert _infer_engagement("hi", "hello there") == "surface"
+
+    def test_normal_conversation_is_surface(self):
+        result = _infer_engagement("what time is it", "It is 3pm right now.")
+        assert result == "surface"
+
+
+class TestInferMood:
+    """Test mood inference."""
+
+    def test_low_confidence_is_hesitant(self):
+        assert _infer_mood("I'm not really sure about this", 0.3) == "hesitant"
+
+    def test_exclamation_with_positive_words_is_energized(self):
+        assert _infer_mood("That's a great idea!", 0.8) == "energized"
+
+    def test_question_words_are_curious(self):
+        assert _infer_mood("I wonder if that would work", 0.6) == "curious"
+
+    def test_neutral_is_settled(self):
+        assert _infer_mood("The answer is 42.", 0.7) == "settled"
+
+    def test_valid_mood_values(self):
+        for mood in MOOD_VALUES:
+            assert isinstance(mood, str)
+
+
+class TestExtractTopic:
+    """Test topic extraction from messages."""
+
+    def test_simple_message(self):
+        assert _extract_topic("Python decorators") == "Python decorators"
+
+    def test_strips_question_prefix(self):
+        topic = _extract_topic("what is a monad")
+        assert topic == "a monad"
+
+    def test_truncates_long_messages(self):
+        long_msg = "a" * 100
+        topic = _extract_topic(long_msg)
+        assert len(topic) <= 60
+
+    def test_empty_returns_none(self):
+        assert _extract_topic("") is None
+        assert _extract_topic("   ") is None
+
+
+class TestExtractCommitments:
+    """Test commitment extraction from responses."""
+
+    def test_i_will_commitment(self):
+        result = _extract_commitments("I will draft the skeleton ticket for you.")
+        assert len(result) >= 1
+        assert "I will draft the skeleton ticket for you" in result[0]
+
+    def test_let_me_commitment(self):
+        result = _extract_commitments("Let me look into that for you.")
+        assert len(result) >= 1
+
+    def test_no_commitments(self):
+        result = _extract_commitments("The answer is 42.")
+        assert result == []
+
+    def test_caps_at_three(self):
+        text = "I will do A. I'll do B. Let me do C. I'm going to do D."
+        result = _extract_commitments(text)
+        assert len(result) <= 3
+
+
+class TestCognitiveTracker:
+    """Test the CognitiveTracker singleton behaviour."""
+
+    def test_update_increments_depth(self, tmp_path):
+        tracker = CognitiveTracker(state_file=tmp_path / "state.txt")
+        tracker.update("hello", "Hi there, how can I help?")
+        assert tracker.get_state().conversation_depth == 1
+        tracker.update("thanks", "You're welcome!")
+        assert tracker.get_state().conversation_depth == 2
+
+    def test_update_sets_focus_topic(self, tmp_path):
+        tracker = CognitiveTracker(state_file=tmp_path / "state.txt")
+        tracker.update(
+            "Python decorators", "Decorators are syntactic sugar for wrapping functions."
+        )
+        assert tracker.get_state().focus_topic == "Python decorators"
+
+    def test_update_persists_to_file(self, tmp_path):
+        state_file = tmp_path / "state.txt"
+        tracker = CognitiveTracker(state_file=state_file)
+        tracker.update("debug the loop", "Let me investigate the issue.")
+        assert state_file.exists()
+        content = state_file.read_text()
+        assert "ENGAGEMENT:" in content
+        assert "MOOD:" in content
+
+    def test_reset_clears_state(self, tmp_path):
+        tracker = CognitiveTracker(state_file=tmp_path / "state.txt")
+        tracker.update("hello", "world")
+        tracker.reset()
+        state = tracker.get_state()
+        assert state.conversation_depth == 0
+        assert state.focus_topic is None
+
+    def test_to_json(self, tmp_path):
+        import json
+
+        tracker = CognitiveTracker(state_file=tmp_path / "state.txt")
+        tracker.update("test", "response")
+        data = json.loads(tracker.to_json())
+        assert "focus_topic" in data
+        assert "engagement" in data
+        assert "mood" in data
+
+    def test_engagement_values_are_valid(self):
+        for level in ENGAGEMENT_LEVELS:
+            assert isinstance(level, str)
+
+    def test_creates_parent_directory(self, tmp_path):
+        state_file = tmp_path / "nested" / "dir" / "state.txt"
+        tracker = CognitiveTracker(state_file=state_file)
+        tracker.update("test", "response")
+        assert state_file.exists()