From e21ad31efd4b31faa64eb2b91e92ecf280dd9d50 Mon Sep 17 00:00:00 2001
From: Jer Miller <jeremie.miller@gmail.com>
Date: Sun, 22 Feb 2026 13:30:11 -0700
Subject: [PATCH] Add callosum connected client count to supervisor status

Files changed: think/callosum.py, think/supervisor.py, think/health_cli.py, tests/test_health_cli.py
---
 tests/test_health_cli.py | 2 ++
 think/callosum.py        | 5 +++++
 think/health_cli.py      | 2 ++
 think/supervisor.py      | 3 +++
 4 files changed, 12 insertions(+)

diff --git a/tests/test_health_cli.py b/tests/test_health_cli.py
index daef6fd45..2b3340891 100644
--- a/tests/test_health_cli.py
+++ b/tests/test_health_cli.py
@@ -31,6 +31,7 @@ def test_health_check_prints_status(capsys):
         "tasks": [{"name": "dream", "duration_seconds": 12}],
         "queues": {"indexer": 3, "planner": 0},
         "stale_heartbeats": [],
+        "callosum_clients": 5,
     }
 
     print_status(status)
@@ -46,6 +47,7 @@ def test_health_check_prints_status(capsys):
     assert "dream" in output
     assert "queued indexer" in output
     assert "Heartbeat: ok" in output
+    assert "Callosum: 5 clients" in output
 
 
 def test_health_check_timeout(tmp_path, monkeypatch, capsys):
diff --git a/think/callosum.py b/think/callosum.py
index 6a809cd11..5b55c0823 100644
--- a/think/callosum.py
+++ b/think/callosum.py
@@ -43,6 +43,11 @@ class CallosumServer:
         self.broadcast_queue: queue.Queue = queue.Queue(maxsize=10000)
         self.writer_thread: threading.Thread | None = None
 
+    def client_count(self) -> int:
+        """Return the number of currently connected clients."""
+        with self.lock:
+            return len(self.clients)
+
     def start(self) -> None:
         """Start the broadcast server."""
         # Ensure health directory exists
diff --git a/think/health_cli.py b/think/health_cli.py
index 575405154..2a54486cd 100644
--- a/think/health_cli.py
+++ b/think/health_cli.py
@@ -88,6 +88,8 @@ def print_status(status: dict[str, Any]) -> None:
         print(f"Heartbeat: STALE ({', '.join(stale)})")
     else:
         print("Heartbeat: ok")
+    callosum_clients = status.get("callosum_clients", 0)
+    print(f"Callosum: {callosum_clients} clients")
 
 
 def health_check() -> int:
diff --git a/think/supervisor.py b/think/supervisor.py
index d7968074c..1cd93a3d8 100644
--- a/think/supervisor.py
+++ b/think/supervisor.py
@@ -779,6 +779,8 @@ def collect_status(procs: list[ManagedProcess]) -> dict:
 
     # Scheduled tasks
     schedules = scheduler.collect_status()
+    # Connected callosum clients
+    callosum_clients = _callosum_server.client_count() if _callosum_server else 0
 
     return {
         "services": services,
@@ -787,6 +789,7 @@ def collect_status(procs: list[ManagedProcess]) -> dict:
         "queues": queues,
         "stale_heartbeats": stale,
         "schedules": schedules,
+        "callosum_clients": callosum_clients,
     }
 
 
-- 
2.51.2