[claude] refactor: break up produce_system_status() (#1194) (#1196)

src/infrastructure/presence.py

@@ -242,6 +242,64 @@ def produce_agent_state(agent_id: str, presence: dict) -> dict:
     }
 
 
+def _get_agents_online() -> int:
+    """Return the count of agents with a non-offline status."""
+    try:
+        from timmy.agents.loader import list_agents
+
+        agents = list_agents()
+        return sum(1 for a in agents if a.get("status", "") not in ("offline", ""))
+    except Exception as exc:
+        logger.debug("Failed to count agents: %s", exc)
+        return 0
+
+
+def _get_visitors() -> int:
+    """Return the count of active WebSocket visitor clients."""
+    try:
+        from dashboard.routes.world import _ws_clients
+
+        return len(_ws_clients)
+    except Exception as exc:
+        logger.debug("Failed to count visitors: %s", exc)
+        return 0
+
+
+def _get_uptime_seconds() -> int:
+    """Return seconds elapsed since application start."""
+    try:
+        from config import APP_START_TIME
+
+        return int((datetime.now(UTC) - APP_START_TIME).total_seconds())
+    except Exception as exc:
+        logger.debug("Failed to calculate uptime: %s", exc)
+        return 0
+
+
+def _get_thinking_active() -> bool:
+    """Return True if the thinking engine is enabled and running."""
+    try:
+        from config import settings
+        from timmy.thinking import thinking_engine
+
+        return settings.thinking_enabled and thinking_engine is not None
+    except Exception as exc:
+        logger.debug("Failed to check thinking status: %s", exc)
+        return False
+
+
+def _get_memory_count() -> int:
+    """Return total entries in the vector memory store."""
+    try:
+        from timmy.memory_system import get_memory_stats
+
+        stats = get_memory_stats()
+        return stats.get("total_entries", 0)
+    except Exception as exc:
+        logger.debug("Failed to count memories: %s", exc)
+        return 0
+
+
 def produce_system_status() -> dict:
     """Generate a system_status message for the Matrix.
 
@@ -270,64 +328,14 @@ def produce_system_status() -> dict:
         "ts": 1742529600,
     }
     """
-    # Count agents with status != offline
-    agents_online = 0
-    try:
-        from timmy.agents.loader import list_agents
-
-        agents = list_agents()
-        agents_online = sum(1 for a in agents if a.get("status", "") not in ("offline", ""))
-    except Exception as exc:
-        logger.debug("Failed to count agents: %s", exc)
-
-    # Count visitors from WebSocket clients
-    visitors = 0
-    try:
-        from dashboard.routes.world import _ws_clients
-
-        visitors = len(_ws_clients)
-    except Exception as exc:
-        logger.debug("Failed to count visitors: %s", exc)
-
-    # Calculate uptime
-    uptime_seconds = 0
-    try:
-        from datetime import UTC
-
-        from config import APP_START_TIME
-
-        uptime_seconds = int((datetime.now(UTC) - APP_START_TIME).total_seconds())
-    except Exception as exc:
-        logger.debug("Failed to calculate uptime: %s", exc)
-
-    # Check thinking engine status
-    thinking_active = False
-    try:
-        from config import settings
-        from timmy.thinking import thinking_engine
-
-        thinking_active = settings.thinking_enabled and thinking_engine is not None
-    except Exception as exc:
-        logger.debug("Failed to check thinking status: %s", exc)
-
-    # Count memories in vector store
-    memory_count = 0
-    try:
-        from timmy.memory_system import get_memory_stats
-
-        stats = get_memory_stats()
-        memory_count = stats.get("total_entries", 0)
-    except Exception as exc:
-        logger.debug("Failed to count memories: %s", exc)
-
     return {
         "type": "system_status",
         "data": {
-            "agents_online": agents_online,
-            "visitors": visitors,
-            "uptime_seconds": uptime_seconds,
-            "thinking_active": thinking_active,
-            "memory_count": memory_count,
+            "agents_online": _get_agents_online(),
+            "visitors": _get_visitors(),
+            "uptime_seconds": _get_uptime_seconds(),
+            "thinking_active": _get_thinking_active(),
+            "memory_count": _get_memory_count(),
         },
         "ts": int(time.time()),
     }

tests/unit/test_presence.py

@@ -6,7 +6,12 @@ import pytest
 
 from infrastructure.presence import (
     DEFAULT_PIP_STATE,
+    _get_agents_online,
     _get_familiar_state,
+    _get_memory_count,
+    _get_thinking_active,
+    _get_uptime_seconds,
+    _get_visitors,
     produce_agent_state,
     produce_bark,
     produce_system_status,
@@ -500,3 +505,36 @@ class TestProduceSystemStatus:
         """produce_system_status always returns a plain dict."""
         result = produce_system_status()
         assert isinstance(result, dict)
+
+
+class TestSystemStatusHelpers:
+    """Tests for the helper functions extracted from produce_system_status()."""
+
+    def test_get_agents_online_returns_int(self):
+        """_get_agents_online returns a non-negative int."""
+        result = _get_agents_online()
+        assert isinstance(result, int)
+        assert result >= 0
+
+    def test_get_visitors_returns_int(self):
+        """_get_visitors returns a non-negative int."""
+        result = _get_visitors()
+        assert isinstance(result, int)
+        assert result >= 0
+
+    def test_get_uptime_seconds_returns_int(self):
+        """_get_uptime_seconds returns a non-negative int."""
+        result = _get_uptime_seconds()
+        assert isinstance(result, int)
+        assert result >= 0
+
+    def test_get_thinking_active_returns_bool(self):
+        """_get_thinking_active returns a bool."""
+        result = _get_thinking_active()
+        assert isinstance(result, bool)
+
+    def test_get_memory_count_returns_int(self):
+        """_get_memory_count returns a non-negative int."""
+        result = _get_memory_count()
+        assert isinstance(result, int)
+        assert result >= 0