[kimi] refactor: extract presence state serializer from workshop heartbeat (#668) #697
@@ -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:
|
||||
|
||||
42
src/infrastructure/presence.py
Normal file
42
src/infrastructure/presence.py
Normal file
@@ -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),
|
||||
}
|
||||
80
tests/unit/test_presence.py
Normal file
80
tests/unit/test_presence.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user