[kimi] Add /api/matrix/agents endpoint for Matrix visualization (#673) #735

Merged
kimi merged 1 commits from kimi/issue-673 into main 2026-03-21 14:18:47 +00:00
3 changed files with 292 additions and 0 deletions

View File

@@ -49,6 +49,7 @@ from dashboard.routes.tools import router as tools_router
from dashboard.routes.tower import router as tower_router
from dashboard.routes.voice import router as voice_router
from dashboard.routes.work_orders import router as work_orders_router
from dashboard.routes.world import matrix_router
from dashboard.routes.world import router as world_router
from timmy.workshop_state import PRESENCE_FILE
@@ -589,6 +590,7 @@ app.include_router(system_router)
app.include_router(experiments_router)
app.include_router(db_explorer_router)
app.include_router(world_router)
app.include_router(matrix_router)
app.include_router(tower_router)

View File

@@ -17,10 +17,12 @@ or missing.
import asyncio
import json
import logging
import math
import re
import time
from collections import deque
from datetime import UTC, datetime
from typing import Any
from fastapi import APIRouter, WebSocket
from fastapi.responses import JSONResponse
@@ -28,6 +30,121 @@ from fastapi.responses import JSONResponse
from infrastructure.presence import serialize_presence
from timmy.workshop_state import PRESENCE_FILE
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/world", tags=["world"])
matrix_router = APIRouter(prefix="/api/matrix", tags=["matrix"])
# ---------------------------------------------------------------------------
# Matrix Agent Registry — serves agents to the Matrix visualization
# ---------------------------------------------------------------------------
# Agent color mapping — consistent with Matrix visual identity
_AGENT_COLORS: dict[str, str] = {
"timmy": "#FFD700", # Gold
"orchestrator": "#FFD700", # Gold
"perplexity": "#3B82F6", # Blue
"replit": "#F97316", # Orange
"kimi": "#06B6D4", # Cyan
"claude": "#A855F7", # Purple
"researcher": "#10B981", # Emerald
"coder": "#EF4444", # Red
"writer": "#EC4899", # Pink
"memory": "#8B5CF6", # Violet
"experimenter": "#14B8A6", # Teal
"forge": "#EF4444", # Red (coder alias)
"seer": "#10B981", # Emerald (researcher alias)
"quill": "#EC4899", # Pink (writer alias)
"echo": "#8B5CF6", # Violet (memory alias)
"lab": "#14B8A6", # Teal (experimenter alias)
}
# Agent shape mapping for 3D visualization
_AGENT_SHAPES: dict[str, str] = {
"timmy": "sphere",
"orchestrator": "sphere",
"perplexity": "cube",
"replit": "cylinder",
"kimi": "dodecahedron",
"claude": "octahedron",
"researcher": "icosahedron",
"coder": "cube",
"writer": "cone",
"memory": "torus",
"experimenter": "tetrahedron",
"forge": "cube",
"seer": "icosahedron",
"quill": "cone",
"echo": "torus",
"lab": "tetrahedron",
}
# Default fallback values
_DEFAULT_COLOR = "#9CA3AF" # Gray
_DEFAULT_SHAPE = "sphere"
_DEFAULT_STATUS = "available"
def _get_agent_color(agent_id: str) -> str:
"""Get the Matrix color for an agent."""
return _AGENT_COLORS.get(agent_id.lower(), _DEFAULT_COLOR)
def _get_agent_shape(agent_id: str) -> str:
"""Get the Matrix shape for an agent."""
return _AGENT_SHAPES.get(agent_id.lower(), _DEFAULT_SHAPE)
def _compute_circular_positions(count: int, radius: float = 3.0) -> list[dict[str, float]]:
"""Compute circular positions for agents in the Matrix.
Agents are arranged in a circle on the XZ plane at y=0.
"""
positions = []
for i in range(count):
angle = (2 * math.pi * i) / count
x = radius * math.cos(angle)
z = radius * math.sin(angle)
positions.append({"x": round(x, 2), "y": 0.0, "z": round(z, 2)})
return positions
def _build_matrix_agents_response() -> list[dict[str, Any]]:
"""Build the Matrix agent registry response.
Reads from agents.yaml and returns agents with Matrix-compatible
formatting including colors, shapes, and positions.
"""
try:
from timmy.agents.loader import list_agents
agents = list_agents()
if not agents:
return []
positions = _compute_circular_positions(len(agents))
result = []
for i, agent in enumerate(agents):
agent_id = agent.get("id", "")
result.append(
{
"id": agent_id,
"display_name": agent.get("name", agent_id.title()),
"role": agent.get("role", "general"),
"color": _get_agent_color(agent_id),
"position": positions[i],
"shape": _get_agent_shape(agent_id),
"status": agent.get("status", _DEFAULT_STATUS),
}
)
return result
except Exception as exc:
logger.warning("Failed to load agents for Matrix: %s", exc)
return []
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/world", tags=["world"])
@@ -370,3 +487,31 @@ async def _generate_bark(visitor_text: str) -> str:
except Exception as exc:
logger.warning("Bark generation failed: %s", exc)
return "Hmm, my thoughts are a bit tangled right now."
# ---------------------------------------------------------------------------
# Matrix Agent Registry Endpoint
# ---------------------------------------------------------------------------
@matrix_router.get("/agents")
async def get_matrix_agents() -> JSONResponse:
"""Return the agent registry for Matrix visualization.
Serves agents from agents.yaml with Matrix-compatible formatting:
- id: agent identifier
- display_name: human-readable name
- role: functional role
- color: hex color code for visualization
- position: {x, y, z} coordinates in 3D space
- shape: 3D shape type
- status: availability status
Agents are arranged in a circular layout by default.
Returns 200 with empty list if no agents configured.
"""
agents = _build_matrix_agents_response()
return JSONResponse(
content=agents,
headers={"Cache-Control": "no-cache, no-store"},
)

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() == []