feat: Workshop state heartbeat for presence.json (#377)

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

tests/timmy/test_workshop_state.py

@@ -0,0 +1,179 @@
+"""Tests for Workshop presence heartbeat."""
+
+import json
+from unittest.mock import patch
+
+import pytest
+
+from timmy.workshop_state import (
+    WorkshopHeartbeat,
+    _state_hash,
+    get_state_dict,
+    write_state,
+)
+
+# ---------------------------------------------------------------------------
+# get_state_dict
+# ---------------------------------------------------------------------------
+
+
+def test_get_state_dict_returns_v1_schema():
+    state = get_state_dict()
+    assert state["version"] == 1
+    assert "liveness" in state
+    assert "current_focus" in state
+    assert "mood" in state
+    assert isinstance(state["active_threads"], list)
+    assert isinstance(state["recent_events"], list)
+    assert isinstance(state["concerns"], list)
+
+
+def test_get_state_dict_idle_mood():
+    """Idle engagement + settled mood → 'idle' presence mood."""
+    from timmy.cognitive_state import CognitiveState, CognitiveTracker
+
+    tracker = CognitiveTracker.__new__(CognitiveTracker)
+    tracker.state = CognitiveState(engagement="idle", mood="settled")
+    with patch("timmy.cognitive_state.cognitive_tracker", tracker):
+        state = get_state_dict()
+    assert state["mood"] == "idle"
+
+
+def test_get_state_dict_maps_mood():
+    """Cognitive moods map to presence moods."""
+    from timmy.cognitive_state import CognitiveState, CognitiveTracker
+
+    for cog_mood, expected in [
+        ("curious", "exploring"),
+        ("hesitant", "uncertain"),
+        ("energized", "excited"),
+    ]:
+        tracker = CognitiveTracker.__new__(CognitiveTracker)
+        tracker.state = CognitiveState(engagement="deep", mood=cog_mood)
+        with patch("timmy.cognitive_state.cognitive_tracker", tracker):
+            state = get_state_dict()
+        assert state["mood"] == expected, f"Expected {expected} for {cog_mood}"
+
+
+# ---------------------------------------------------------------------------
+# write_state
+# ---------------------------------------------------------------------------
+
+
+def test_write_state_creates_file(tmp_path):
+    target = tmp_path / "presence.json"
+    state = {"version": 1, "liveness": "2026-01-01T00:00:00Z", "current_focus": ""}
+    write_state(state, path=target)
+
+    assert target.exists()
+    data = json.loads(target.read_text())
+    assert data["version"] == 1
+
+
+def test_write_state_creates_parent_dirs(tmp_path):
+    target = tmp_path / "deep" / "nested" / "presence.json"
+    write_state({"version": 1}, path=target)
+    assert target.exists()
+
+
+# ---------------------------------------------------------------------------
+# _state_hash
+# ---------------------------------------------------------------------------
+
+
+def test_state_hash_ignores_liveness():
+    a = {"version": 1, "mood": "focused", "liveness": "2026-01-01T00:00:00Z"}
+    b = {"version": 1, "mood": "focused", "liveness": "2026-12-31T23:59:59Z"}
+    assert _state_hash(a) == _state_hash(b)
+
+
+def test_state_hash_detects_mood_change():
+    a = {"version": 1, "mood": "focused", "liveness": "t1"}
+    b = {"version": 1, "mood": "idle", "liveness": "t1"}
+    assert _state_hash(a) != _state_hash(b)
+
+
+# ---------------------------------------------------------------------------
+# WorkshopHeartbeat — _write_if_changed
+# ---------------------------------------------------------------------------
+
+
+def test_write_if_changed_writes_on_first_call(tmp_path):
+    target = tmp_path / "presence.json"
+    hb = WorkshopHeartbeat(interval=60, path=target)
+
+    with patch("timmy.workshop_state.get_state_dict") as mock_state:
+        mock_state.return_value = {
+            "version": 1,
+            "liveness": "t1",
+            "current_focus": "testing",
+            "mood": "focused",
+        }
+        hb._write_if_changed()
+
+    assert target.exists()
+    data = json.loads(target.read_text())
+    assert data["version"] == 1
+    assert data["current_focus"] == "testing"
+
+
+def test_write_if_changed_skips_when_unchanged(tmp_path):
+    target = tmp_path / "presence.json"
+    hb = WorkshopHeartbeat(interval=60, path=target)
+
+    fixed_state = {"version": 1, "liveness": "t1", "mood": "idle"}
+
+    with patch("timmy.workshop_state.get_state_dict", return_value=fixed_state):
+        hb._write_if_changed()  # First write
+        target.write_text("")  # Clear to detect if second write happens
+        hb._write_if_changed()  # Should skip — state unchanged
+
+    # File should still be empty (second write was skipped)
+    assert target.read_text() == ""
+
+
+def test_write_if_changed_writes_on_state_change(tmp_path):
+    target = tmp_path / "presence.json"
+    hb = WorkshopHeartbeat(interval=60, path=target)
+
+    state_a = {"version": 1, "liveness": "t1", "mood": "idle"}
+    state_b = {"version": 1, "liveness": "t2", "mood": "focused"}
+
+    with patch("timmy.workshop_state.get_state_dict", return_value=state_a):
+        hb._write_if_changed()
+
+    with patch("timmy.workshop_state.get_state_dict", return_value=state_b):
+        hb._write_if_changed()
+
+    data = json.loads(target.read_text())
+    assert data["mood"] == "focused"
+
+
+# ---------------------------------------------------------------------------
+# WorkshopHeartbeat — notify
+# ---------------------------------------------------------------------------
+
+
+def test_notify_sets_trigger():
+    hb = WorkshopHeartbeat(interval=60)
+    assert not hb._trigger.is_set()
+    hb.notify()
+    assert hb._trigger.is_set()
+
+
+# ---------------------------------------------------------------------------
+# WorkshopHeartbeat — start/stop lifecycle
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_heartbeat_start_stop_lifecycle(tmp_path):
+    target = tmp_path / "presence.json"
+    hb = WorkshopHeartbeat(interval=60, path=target)
+
+    with patch("timmy.workshop_state.get_state_dict", return_value={"version": 1}):
+        await hb.start()
+        assert hb._task is not None
+        assert not hb._task.done()
+        await hb.stop()
+        assert hb._task is None