From f0689ac12b4cfc63e4a8ded415f23682768af201 Mon Sep 17 00:00:00 2001 From: gramps Date: Thu, 9 Jul 2026 09:09:22 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20Roadmap=20N=20=E2=80=94=20AMQP=20cluste?= =?UTF-8?q?r=20nervous=20system=20complete?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 2 ++ TASKS.md | 4 +-- cluster.py | 10 +++++++ config.py | 2 +- templates/index.html | 44 +++++++++++++++++++++++++++++ tests/test_cluster_heartbeat.py | 49 +++++++++++++++++++++++++++++++++ 6 files changed, 108 insertions(+), 3 deletions(-) create mode 100644 tests/test_cluster_heartbeat.py diff --git a/AGENTS.md b/AGENTS.md index e7b9c05..a252dce 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,6 +28,8 @@ Every router has a dedicated test file: | `test_profile.py` | `routers/profile.py` — get, update, default, length validation | | `test_search_route.py` | `routers/search_route.py` — explicit search flow, no results, errors | | `test_search_url_sanitization.py` | `search.py` URL sanitizer | +| `test_cluster.py` | `cluster.py` — registration, deregistration, pong, events, coordinator query | +| `test_cluster_heartbeat.py` | `cluster.py` — heartbeat handler, known/unknown node | | `test_model_swap.py` | `cluster.py` + `triage.py` — request_model_swap, handle_model_ready/failed, select_node swap triggering | | `test_node_agent.py` | `node_agent/agent.py` — registration, ping/pong, model swap | | `test_triage.py` | `triage.py` — classify_query, select_node, get_inference_url | diff --git a/TASKS.md b/TASKS.md index 891be72..c2bbee4 100644 --- a/TASKS.md +++ b/TASKS.md @@ -868,7 +868,7 @@ Run full test suite. All existing tests must continue to pass. --- -## TASK 15 — Roadmap N7: Cluster Status UI +## ~~TASK 15 — Roadmap N7: Cluster Status UI [DONE]~~ Surface cluster awareness in the jC frontend (`templates/index.html`). @@ -899,7 +899,7 @@ Surface cluster awareness in the jC frontend (`templates/index.html`). Run full test suite. All 26+ existing tests must continue to pass. -Commit all changes introduced across Tasks 9–15 with message: `feat: Roadmap N — AMQP cluster nervous system complete` +~~Commit all changes introduced across Tasks 9–15 with message: `feat: Roadmap N — AMQP cluster nervous system complete`~~ --- diff --git a/cluster.py b/cluster.py index 70cc438..ea8fd9a 100644 --- a/cluster.py +++ b/cluster.py @@ -218,6 +218,15 @@ async def handle_model_ready(exchange: str, routing_key: str, payload: dict) -> _push_event("cluster", "info", node_name, f"Model swap complete: {model_filename}") +async def handle_heartbeat(exchange: str, routing_key: str, payload: dict) -> None: + node_name = payload.get("node_name", routing_key.split(".")[1] if "." in routing_key else "unknown") + if node_name not in CLUSTER_NODES: + log.warning("heartbeat from unknown node %s — ignore, full registration required", node_name) + return + now = datetime.now(timezone.utc).isoformat() + "Z" + CLUSTER_NODES[node_name]["last_seen"] = now + + async def handle_model_failed(exchange: str, routing_key: str, payload: dict) -> None: node_name = payload.get("node_name", routing_key.split(".")[1] if "." in routing_key else "unknown") error = payload.get("error", "unknown error") @@ -236,6 +245,7 @@ SUBSCRIBE_TABLE = [ (AMQP_EXCHANGE_ADMIN, ["node.*.pong"], handle_pong), (AMQP_EXCHANGE_SYSTEM, ["node.*.event"], handle_event), (AMQP_EXCHANGE_SYSTEM, ["cluster.coordinator.query"], handle_coordinator_query), + (AMQP_EXCHANGE_SYSTEM, ["node.*.heartbeat"], handle_heartbeat), (AMQP_EXCHANGE_SYSTEM, ["node.*.model_ready"], handle_model_ready), (AMQP_EXCHANGE_SYSTEM, ["node.*.model_failed"], handle_model_failed), ] diff --git a/config.py b/config.py index 7b1f6d7..3bf87cd 100644 --- a/config.py +++ b/config.py @@ -9,7 +9,7 @@ import logging log = logging.getLogger("caic") -VERSION = "v0.16.0" +VERSION = "v0.17.0" OLLAMA_BASE = os.environ.get("OLLAMA_BASE", "http://localhost:11434") LLAMA_SERVER_BASE = os.environ.get("LLAMA_SERVER_BASE", "http://192.168.50.108:8081") SEARXNG_BASE = "http://localhost:8888" diff --git a/templates/index.html b/templates/index.html index e68d3ef..650b6ef 100644 --- a/templates/index.html +++ b/templates/index.html @@ -59,6 +59,15 @@ body { font-family: var(--font-body); background: var(--bg-primary); color: var( .sidebar-footer { padding: 12px 16px; border-top: 1px solid var(--border); font-size: 11px; color: var(--text-muted); font-family: var(--font-mono); } .sidebar-footer .status-row { display: flex; align-items: center; gap: 8px; margin-bottom: 4px; } .stats-panel { margin-top: 10px; padding-top: 10px; border-top: 1px solid var(--border); } +.cluster-panel { font-size: 11px; font-family: var(--font-mono); } +.cluster-node-row { display: flex; align-items: center; gap: 6px; padding: 3px 0; } +.cluster-node-name { color: var(--text-primary); min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } +.cluster-node-model { color: var(--text-muted); font-size: 10px; min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } +.cluster-dot { width: 7px; height: 7px; border-radius: 50%; flex-shrink: 0; } +.cluster-dot.active { background: var(--success); } +.cluster-dot.swapping { background: var(--warning); animation: pulse 1s infinite; } +.cluster-dot.error { background: var(--danger); } +.cluster-dot.offline { background: var(--danger); } .stat-row { display: flex; align-items: center; gap: 6px; margin-bottom: 6px; } .stat-label { width: 36px; font-size: 10px; color: var(--text-muted); text-transform: uppercase; } .stat-bar { flex: 1; height: 8px; background: var(--bg-tertiary); border-radius: 4px; overflow: hidden; } @@ -290,6 +299,8 @@ body { font-family: var(--font-body); background: var(--bg-primary); color: var(
checking...
search: checking...
memory: -- entries
+
cluster: --
+
CPU
--%
MEM
--%
@@ -654,9 +665,11 @@ async function initializeMainApp() { await loadMemoryStats(); checkOllamaStatus(); checkSearchStatus(); + updateClusterStatus(); updateSystemStats(); setInterval(checkOllamaStatus, 30000); setInterval(checkSearchStatus, 60000); + setInterval(updateClusterStatus, 15000); setInterval(updateSystemStats, 2000); document.getElementById('userInput').addEventListener('input', updateTokenThermometer); updateTokenThermometer(); @@ -730,6 +743,37 @@ async function checkSearchStatus() { } } +async function updateClusterStatus() { + try { + const resp = await authFetch('/api/cluster'); + const data = await resp.json(); + const statusEl = document.getElementById('clusterStatus'); + const panel = document.getElementById('clusterPanel'); + const count = data.node_count || 0; + statusEl.innerHTML = ` cluster: ${count} node${count !== 1 ? 's' : ''}`; + if (count === 0) { + panel.style.display = 'none'; + return; + } + panel.style.display = 'block'; + const now = Date.now(); + let html = ''; + for (const [name, node] of Object.entries(data.nodes)) { + const lastSeen = node.last_seen ? new Date(node.last_seen).getTime() : 0; + const ageSec = (now - lastSeen) / 1000; + let dotClass = 'active'; + if (node.status === 'swapping') dotClass = 'swapping'; + else if (node.status === 'error') dotClass = 'error'; + else if (ageSec > 60) dotClass = 'offline'; + const modelName = node.active_model ? (node.active_model.name || 'unknown') : 'idle'; + html += `
${escapeHtml(name)}${escapeHtml(modelName)}
`; + } + panel.innerHTML = html; + } catch(e) { + document.getElementById('clusterStatus').innerHTML = ' cluster: error'; + } +} + async function loadModels() { try { const resp = await authFetch('/api/models'); diff --git a/tests/test_cluster_heartbeat.py b/tests/test_cluster_heartbeat.py new file mode 100644 index 0000000..4cf96cb --- /dev/null +++ b/tests/test_cluster_heartbeat.py @@ -0,0 +1,49 @@ +"""Tests for cluster.py heartbeat handler.""" +import asyncio + +import cluster +from config import AMQP_EXCHANGE_SYSTEM + + +def _reset(): + cluster.CLUSTER_NODES.clear() + cluster.CLUSTER_EVENTS.clear() + cluster.CLUSTER_COORDINATOR = None + cluster._pending_pings.clear() + + +# ---------- 1. handle_heartbeat() for known node updates last_seen ---------- + + +def test_heartbeat_updates_last_seen(monkeypatch): + _reset() + + cluster.CLUSTER_NODES["jarvis"] = { + "name": "jarvis", "type": "worker", "status": "active", + "last_seen": "2020-01-01T00:00:00Z", + } + original = cluster.CLUSTER_NODES["jarvis"]["last_seen"] + + asyncio.run(cluster.handle_heartbeat( + AMQP_EXCHANGE_SYSTEM, "node.jarvis.heartbeat", + {"node_name": "jarvis"}, + )) + + assert cluster.CLUSTER_NODES["jarvis"]["last_seen"] != original + assert cluster.CLUSTER_NODES["jarvis"]["status"] == "active" # unchanged + + +# ---------- 2. handle_heartbeat() for unknown node logs warning, no add ---------- + + +def test_heartbeat_unknown_node_logs_warning(caplog, monkeypatch): + _reset() + caplog.set_level("WARNING") + + asyncio.run(cluster.handle_heartbeat( + AMQP_EXCHANGE_SYSTEM, "node.stranger.heartbeat", + {"node_name": "stranger"}, + )) + + assert "stranger" not in cluster.CLUSTER_NODES + assert any("unknown node" in rec.message and "stranger" in rec.message for rec in caplog.records)