diff --git a/src/timmy/cognitive_state.py b/src/timmy/cognitive_state.py new file mode 100644 index 0000000..e513ac0 --- /dev/null +++ b/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() diff --git a/src/timmy/session.py b/src/timmy/session.py index 39577e1..45ff9e3 100644 --- a/src/timmy/session.py +++ b/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() diff --git a/tests/timmy/test_cognitive_state.py b/tests/timmy/test_cognitive_state.py new file mode 100644 index 0000000..409ecc1 --- /dev/null +++ b/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()