refactor: extract helpers from produce_system_status() in presence.py
Some checks failed
Tests / lint (pull_request) Failing after 13s
Tests / test (pull_request) Has been skipped

Break the 88-line produce_system_status() into five focused helpers:
_get_agents_online(), _get_visitors(), _get_uptime_seconds(),
_get_thinking_active(), and _get_memory_count(). Main function becomes
a thin coordinator. Adds unit tests for each helper.

Fixes #1194
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Alexander Whitestone
2026-03-23 17:55:24 -04:00
parent 8f8061e224
commit 96724faed9
2 changed files with 101 additions and 55 deletions

View File

@@ -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()),
}

View File

@@ -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