[kimi] Add /api/matrix/agents endpoint for Matrix visualization (#673) (#735)
Some checks failed
Tests / lint (push) Has been cancelled
Tests / test (push) Has been cancelled

This commit was merged in pull request #735.
This commit is contained in:
2026-03-21 14:18:46 +00:00
parent 646eaefa3e
commit c9601ba32c
3 changed files with 292 additions and 0 deletions

View File

@@ -15,11 +15,15 @@ from dashboard.routes.world import (
_bark_and_broadcast,
_broadcast,
_build_commitment_context,
_build_matrix_agents_response,
_build_world_state,
_commitments,
_compute_circular_positions,
_conversation,
_extract_commitments,
_generate_bark,
_get_agent_color,
_get_agent_shape,
_handle_client_message,
_heartbeat,
_log_bark_failure,
@@ -718,3 +722,144 @@ async def test_heartbeat_exits_on_dead_connection():
with patch("dashboard.routes.world.asyncio.sleep", new_callable=AsyncMock):
await _heartbeat(ws) # should not raise
# ---------------------------------------------------------------------------
# Matrix Agent Registry (/api/matrix/agents)
# ---------------------------------------------------------------------------
class TestMatrixAgentRegistry:
"""Tests for the Matrix agent registry endpoint."""
def test_get_agent_color_known_agents(self):
"""Known agents return their assigned colors."""
assert _get_agent_color("timmy") == "#FFD700" # Gold
assert _get_agent_color("orchestrator") == "#FFD700" # Gold
assert _get_agent_color("kimi") == "#06B6D4" # Cyan
assert _get_agent_color("claude") == "#A855F7" # Purple
assert _get_agent_color("researcher") == "#10B981" # Emerald
assert _get_agent_color("coder") == "#EF4444" # Red
def test_get_agent_color_unknown_agent(self):
"""Unknown agents return the default gray color."""
assert _get_agent_color("unknown") == "#9CA3AF"
assert _get_agent_color("xyz") == "#9CA3AF"
def test_get_agent_color_case_insensitive(self):
"""Agent ID lookup is case insensitive."""
assert _get_agent_color("Timmy") == "#FFD700"
assert _get_agent_color("KIMI") == "#06B6D4"
def test_get_agent_shape_known_agents(self):
"""Known agents return their assigned shapes."""
assert _get_agent_shape("timmy") == "sphere"
assert _get_agent_shape("coder") == "cube"
assert _get_agent_shape("writer") == "cone"
def test_get_agent_shape_unknown_agent(self):
"""Unknown agents return the default sphere shape."""
assert _get_agent_shape("unknown") == "sphere"
def test_compute_circular_positions(self):
"""Agents are arranged in a circle on the XZ plane."""
positions = _compute_circular_positions(4, radius=3.0)
assert len(positions) == 4
# All positions should have y=0
for pos in positions:
assert pos["y"] == 0.0
assert "x" in pos
assert "z" in pos
# First position should be at angle 0 (x=radius, z=0)
assert positions[0]["x"] == 3.0
assert positions[0]["z"] == 0.0
def test_compute_circular_positions_empty(self):
"""Zero agents returns empty positions list."""
positions = _compute_circular_positions(0)
assert positions == []
def test_build_matrix_agents_response_structure(self):
"""Response contains all required fields for each agent."""
with patch("timmy.agents.loader.list_agents") as mock_list:
mock_list.return_value = [
{"id": "timmy", "name": "Timmy", "role": "orchestrator", "status": "available"},
{"id": "researcher", "name": "Seer", "role": "research", "status": "busy"},
]
result = _build_matrix_agents_response()
assert len(result) == 2
# Check first agent
assert result[0]["id"] == "timmy"
assert result[0]["display_name"] == "Timmy"
assert result[0]["role"] == "orchestrator"
assert result[0]["color"] == "#FFD700"
assert result[0]["shape"] == "sphere"
assert result[0]["status"] == "available"
assert "position" in result[0]
assert "x" in result[0]["position"]
assert "y" in result[0]["position"]
assert "z" in result[0]["position"]
def test_build_matrix_agents_response_empty(self):
"""Returns empty list when no agents configured."""
with patch("timmy.agents.loader.list_agents") as mock_list:
mock_list.return_value = []
result = _build_matrix_agents_response()
assert result == []
def test_build_matrix_agents_response_handles_errors(self):
"""Returns empty list when loader fails."""
with patch("timmy.agents.loader.list_agents") as mock_list:
mock_list.side_effect = RuntimeError("Loader failed")
result = _build_matrix_agents_response()
assert result == []
@pytest.fixture
def matrix_client():
"""TestClient with matrix router."""
from fastapi import FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
from dashboard.routes.world import matrix_router
app.include_router(matrix_router)
return TestClient(app)
def test_matrix_agents_endpoint_returns_json(matrix_client):
"""GET /api/matrix/agents returns JSON list."""
with patch("timmy.agents.loader.list_agents") as mock_list:
mock_list.return_value = [
{"id": "timmy", "name": "Timmy", "role": "orchestrator", "status": "available"},
]
resp = matrix_client.get("/api/matrix/agents")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data, list)
assert len(data) == 1
assert data[0]["id"] == "timmy"
assert resp.headers["cache-control"] == "no-cache, no-store"
def test_matrix_agents_endpoint_empty_list(matrix_client):
"""Endpoint returns 200 with empty list when no agents."""
with patch("timmy.agents.loader.list_agents") as mock_list:
mock_list.return_value = []
resp = matrix_client.get("/api/matrix/agents")
assert resp.status_code == 200
assert resp.json() == []
def test_matrix_agents_endpoint_graceful_degradation(matrix_client):
"""Endpoint returns empty list when loader fails."""
with patch("timmy.agents.loader.list_agents") as mock_list:
mock_list.side_effect = FileNotFoundError("agents.yaml not found")
resp = matrix_client.get("/api/matrix/agents")
assert resp.status_code == 200
assert resp.json() == []