[kimi] refactor: extract presence state serializer from workshop heartbeat (#668) (#697)

src/dashboard/routes/world.py

@@ -25,6 +25,7 @@ from datetime import UTC, datetime
 from fastapi import APIRouter, WebSocket
 from fastapi.responses import JSONResponse
 
+from infrastructure.presence import serialize_presence
 from timmy.workshop_state import PRESENCE_FILE
 
 logger = logging.getLogger(__name__)
@@ -149,21 +150,7 @@ def _read_presence_file() -> dict | None:
 
 def _build_world_state(presence: dict) -> dict:
     """Transform presence dict into the world/state API response."""
-    return {
-        "timmyState": {
-            "mood": presence.get("mood", "calm"),
-            "activity": presence.get("current_focus", "idle"),
-            "energy": presence.get("energy", 0.5),
-            "confidence": presence.get("confidence", 0.7),
-        },
-        "familiar": presence.get("familiar"),
-        "activeThreads": presence.get("active_threads", []),
-        "recentEvents": presence.get("recent_events", []),
-        "concerns": presence.get("concerns", []),
-        "visitorPresent": False,
-        "updatedAt": presence.get("liveness", datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")),
-        "version": presence.get("version", 1),
-    }
+    return serialize_presence(presence)
 
 
 def _get_current_state() -> dict:

src/infrastructure/presence.py

@@ -0,0 +1,42 @@
+"""Presence state serializer — transforms ADR-023 presence dicts for consumers.
+
+Converts the raw presence schema (version, liveness, mood, energy, etc.)
+into the camelCase world-state payload consumed by the Workshop 3D renderer
+and WebSocket gateway.
+"""
+
+from datetime import UTC, datetime
+
+
+def serialize_presence(presence: dict) -> dict:
+    """Transform an ADR-023 presence dict into the world-state API shape.
+
+    Parameters
+    ----------
+    presence:
+        Raw presence dict as written by
+        :func:`~timmy.workshop_state.get_state_dict` or read from
+        ``~/.timmy/presence.json``.
+
+    Returns
+    -------
+    dict
+        CamelCase world-state payload with ``timmyState``, ``familiar``,
+        ``activeThreads``, ``recentEvents``, ``concerns``, ``visitorPresent``,
+        ``updatedAt``, and ``version`` keys.
+    """
+    return {
+        "timmyState": {
+            "mood": presence.get("mood", "calm"),
+            "activity": presence.get("current_focus", "idle"),
+            "energy": presence.get("energy", 0.5),
+            "confidence": presence.get("confidence", 0.7),
+        },
+        "familiar": presence.get("familiar"),
+        "activeThreads": presence.get("active_threads", []),
+        "recentEvents": presence.get("recent_events", []),
+        "concerns": presence.get("concerns", []),
+        "visitorPresent": False,
+        "updatedAt": presence.get("liveness", datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")),
+        "version": presence.get("version", 1),
+    }

tests/unit/test_presence.py

@@ -0,0 +1,80 @@
+"""Tests for infrastructure.presence — presence state serializer."""
+
+import pytest
+
+from infrastructure.presence import serialize_presence
+
+
+class TestSerializePresence:
+    """Round-trip and edge-case tests for serialize_presence()."""
+
+    @pytest.fixture()
+    def full_presence(self):
+        """A complete ADR-023 presence dict."""
+        return {
+            "version": 1,
+            "liveness": "2026-03-21T12:00:00Z",
+            "current_focus": "writing tests",
+            "mood": "focused",
+            "energy": 0.9,
+            "confidence": 0.85,
+            "active_threads": [
+                {"type": "thinking", "ref": "refactor presence", "status": "active"}
+            ],
+            "recent_events": ["committed code"],
+            "concerns": ["test coverage"],
+            "familiar": {"name": "Pip", "state": "alert"},
+        }
+
+    def test_full_round_trip(self, full_presence):
+        """All ADR-023 fields map to the expected camelCase keys."""
+        result = serialize_presence(full_presence)
+
+        assert result["timmyState"]["mood"] == "focused"
+        assert result["timmyState"]["activity"] == "writing tests"
+        assert result["timmyState"]["energy"] == 0.9
+        assert result["timmyState"]["confidence"] == 0.85
+        assert result["familiar"] == {"name": "Pip", "state": "alert"}
+        assert result["activeThreads"] == full_presence["active_threads"]
+        assert result["recentEvents"] == ["committed code"]
+        assert result["concerns"] == ["test coverage"]
+        assert result["visitorPresent"] is False
+        assert result["updatedAt"] == "2026-03-21T12:00:00Z"
+        assert result["version"] == 1
+
+    def test_defaults_on_empty_dict(self):
+        """Missing fields fall back to safe defaults."""
+        result = serialize_presence({})
+
+        assert result["timmyState"]["mood"] == "calm"
+        assert result["timmyState"]["activity"] == "idle"
+        assert result["timmyState"]["energy"] == 0.5
+        assert result["timmyState"]["confidence"] == 0.7
+        assert result["familiar"] is None
+        assert result["activeThreads"] == []
+        assert result["recentEvents"] == []
+        assert result["concerns"] == []
+        assert result["visitorPresent"] is False
+        assert result["version"] == 1
+        # updatedAt should be an ISO timestamp string
+        assert "T" in result["updatedAt"]
+
+    def test_partial_presence(self):
+        """Only some fields provided — others get defaults."""
+        result = serialize_presence({"mood": "excited", "energy": 0.3})
+
+        assert result["timmyState"]["mood"] == "excited"
+        assert result["timmyState"]["energy"] == 0.3
+        assert result["timmyState"]["confidence"] == 0.7  # default
+        assert result["activeThreads"] == []  # default
+
+    def test_return_type_is_dict(self, full_presence):
+        """serialize_presence always returns a plain dict."""
+        result = serialize_presence(full_presence)
+        assert isinstance(result, dict)
+        assert isinstance(result["timmyState"], dict)
+
+    def test_visitor_present_always_false(self, full_presence):
+        """visitorPresent is always False — set by the WS layer, not here."""
+        assert serialize_presence(full_presence)["visitorPresent"] is False
+        assert serialize_presence({})["visitorPresent"] is False