Resolve #18 -- Add worker pool telemetry to inspector

AGENTS.md

@@ -18,7 +18,7 @@ uv run pytest -m "integration and benchmark"
 uv run pytest --benchmark-compare          # compare vs main baseline (run main first)
 uvx prek run --all-files
 uv run manage.py threadmill worker         # run the worker pool
-uv run manage.py threadmill inspector    # launch the textual TUI inspector
+uv run manage.py threadmill inspector      # launch the textual TUI inspector
 ```
 
 CI additionally pins Django per matrix step: `uv run --with django~=6.1a1 pytest -m "not benchmark"`.
@@ -28,8 +28,8 @@ Run a single test by node ID, e.g. `uv run pytest tests/test_command.py::TestCom
 ## Setup
 
 - Install pre-commit hooks before first commit: `uvx prek install` (not `pre-commit install`).
-- `DJANGO_SETTINGS_MODULE=tests.testapp.settings` is already set in `.env` and in `pyproject.toml`. Pytest auto-loads it.
-- CI's Linux job starts a Redis service and sets `REDIS_URL`; some integration tests may rely on it. Local runs of `-m integration` may need Redis if a backend test targets it.
+- `DJANGO_SETTINGS_MODULE=tests.testapp.settings` is set in `.env` and `pyproject.toml`; pytest auto-loads it.
+- CI's Linux job starts Redis and sets `REDIS_URL`; some integration tests rely on it. Local `-m integration` runs may need Redis.
 
 ## Code & style (repo-specific, beyond PEP 8)
 
@@ -50,15 +50,15 @@ Codecov requires 100% patch coverage on PRs (`pyproject.toml`, `.codecov.yml`).
 
 ## Architecture notes
 
-- Entry point for end users: `threadmill/management/commands/threadmill.py` (Django management command with `worker` and `inspector` subcommands).
+- Entry point: `threadmill/management/commands/threadmill.py` (management command with `worker` and `inspector` subcommands).
 - Core runtime: `threadmill/executor.py` (`TaskExecutor`) — process/thread pool, graceful shutdown, worker recycling, task timeout/backlog.
-- Integration point for queue authors: `threadmill/backends.py` (`AcknowledgeableTaskBackend`) — subclasses implement `acquire` (lock-without-remove) and `acknowledge` (remove + publish). Requires late-ack support from the underlying queue.
+- Queue author integration point: `threadmill/backends/base.py` (`ThreadmillTaskBackend`) — subclasses implement `acquire` (lock-without-remove) and `acknowledge` (remove + publish); requires late-ack support from the underlying queue.
 - Test app backend `tests/testapp/backends.py` (`GeneratingTaskBackend`) generates tasks in-process for benchmarks; reset between runs via `default_task_backend.reset()`.
 
 ## Pre-commit
 
-`.pre-commit-config.yaml` runs ruff (check + format), django-upgrade, pyupgrade, mdformat (excludes `.github/agents/`), yamlfmt, and `no-commit-to-branch` (protects `main`). Hooks auto-fix; ruff is configured `--exit-non-zero-on-fix`, so commit any fixes before pushing.
+`.pre-commit-config.yaml` runs ruff (check + format), django-upgrade, pyupgrade, mdformat (excludes `.github/agents/`), yamlfmt, and `no-commit-to-branch` (protects `main`). Hooks auto-fix; ruff uses `--exit-non-zero-on-fix`, so commit fixes before pushing.
 
 ## PR / release
 
-CI runs on `main` pushes and PRs. Releases are published to PyPI via `.github/workflows/release.yml` on GitHub release. Commits to `main` are blocked by `no-commit-to-branch`; work on a branch.
+CI runs on `main` pushes and PRs. Releases publish to PyPI via `.github/workflows/release.yml` on GitHub release. `no-commit-to-branch` blocks commits to `main`; work on a branch.

CONTRIBUTING.md

@@ -15,35 +15,21 @@ curl -sSL https://raw.githubusercontent.com/codingjoe/naming-things/refs/heads/m
 
 ## Testing
 
-We have unit tests, integration tests, and benchmarks. Avoid mocking if possible.
-
-To run the tests, use the following command:
-
-```bash
-uv run pytest
-```
-
-To run only integration tests:
-
-```bash
-uv run pytest -m integration
-```
-
-To run only integration benchmarks:
+The suite has unit tests, integration tests, and benchmarks. Avoid mocking where possible.
 
 ```bash
+uv run pytest                              # full suite
+uv run pytest -m integration               # integration tests only
 uv run pytest -m "integration and benchmark"
 ```
 
-Benchmarking snapshots are created automatically.
-To compare your feature branch against the main branch,
-run the test suite on main, followed by:
+Benchmark snapshots are created automatically. To compare a feature branch against main, run the suite on main first, then:
 
 ```
 uv run pytest --benchmark-compare
 ```
 
-Before your first commit, ensure that the pre-commit hooks are installed by running:
+Install pre-commit hooks before your first commit:
 
 ```bash
 uvx prek install

pyproject.toml

@@ -40,6 +40,10 @@ dependencies = ["django>=6.0"]
 redis = ["redis>=5.0"]
 inspector = [
     "textual>=8.2.7",
+    "textual-plotext>=1.0.1",
+]
+worker = [
+    "psutil>=7.2.2",
 ]
 
 [project.urls]

tests/backends/test_redis.py

@@ -1,22 +1,27 @@
 from __future__ import annotations
 
+import asyncio
 import dataclasses
 import datetime
 import logging
 import time
 from dataclasses import replace
 from unittest.mock import patch
 
+import pytest
 from django.tasks import default_task_backend
 from django.tasks.base import TaskResultStatus
 from django.utils import timezone
 
 from tests.testapp.tasks import boom, compute_workload, echo
 from threadmill.backends.base import (
     BackendTelemetry,
+    NodeTelemetry,
     QueueCounts,
     QueueRates,
     QueueStats,
+    ThreadmillTaskBackend,
+    WorkerTelemetry,
 )
 from threadmill.backends.redis import RedisBroker, RedisTaskBackend  # noqa: E402
 
@@ -687,3 +692,121 @@ def test_peek__empty_history_returns_nothing(self):
         )
         assert successful == []
         assert failed == []
+
+
+class TestWorkerTelemetrySerialization:
+    """Tests for worker telemetry JSON serialization/deserialization."""
+
+    def test_serialize_deserialize_roundtrip(self):
+        """A WorkerTelemetry snapshot survives a serialize→deserialize roundtrip."""
+        from threadmill.backends.base import (
+            NodeTelemetry,
+            WorkerTelemetry,
+        )
+        from threadmill.backends.redis import (
+            _deserialize_worker_telemetry,
+            _serialize_worker_telemetry,
+        )
+
+        sampled_at = datetime.datetime.now(tz=datetime.UTC)
+        node = NodeTelemetry(
+            hostname="host",
+            pid=100,
+            queues=("default", "high"),
+            process_count=1,
+            thread_count=4,
+            cpu_percent=45.0,
+            memory_bytes=8_000_000_000,
+            memory_total=16_000_000_000,
+            tasks_per_minute=12.5,
+            sampled_at=sampled_at,
+        )
+        original = WorkerTelemetry(
+            nodes={"host": node},
+            queues={"default": ("host",), "high": ("host",)},
+            sampled_at=sampled_at,
+        )
+        payload = _serialize_worker_telemetry(original)
+        restored = _deserialize_worker_telemetry(payload)
+        assert restored.sampled_at == original.sampled_at
+        assert set(restored.nodes) == {"host"}
+        assert set(restored.queues) == {"default", "high"}
+        r_node = restored.nodes["host"]
+        assert r_node.hostname == "host"
+        assert r_node.pid == 100
+        assert r_node.process_count == 1
+        assert r_node.thread_count == 4
+        assert r_node.cpu_percent == 45.0
+        assert r_node.memory_bytes == 8_000_000_000
+        assert r_node.memory_total == 16_000_000_000
+        assert r_node.queues == ("default", "high")
+
+
+@pytest.mark.integration
+class TestWorkerTelemetryRedis:
+    """Integration tests for worker telemetry storage over real Redis."""
+
+    async def test_publish_and_read_roundtrip(self):
+        """publish_worker_telemetry publishes, subscribe_worker_telemetry reads it back."""
+        backend = RedisTaskBackend(
+            "worker_telemetry_store_test",
+            {
+                "QUEUES": ["default"],
+                "REDIS_URL": "redis://localhost:6379/0",
+                "OPTIONS": {
+                    "result_ttl": datetime.timedelta(seconds=60),
+                },
+            },
+        )
+        try:
+            sampled_at = datetime.datetime.now(tz=datetime.UTC)
+            node = NodeTelemetry(
+                hostname="testhost",
+                pid=200,
+                queues=("default",),
+                process_count=1,
+                thread_count=2,
+                cpu_percent=30.0,
+                memory_bytes=4_000_000_000,
+                memory_total=8_000_000_000,
+                tasks_per_minute=10.0,
+                sampled_at=sampled_at,
+            )
+            snapshot = WorkerTelemetry(
+                nodes={"testhost": node},
+                queues={"default": ("testhost",)},
+                sampled_at=sampled_at,
+            )
+
+            async def _collect():
+                async for received in backend.subscribe_worker_telemetry():
+                    return received
+
+            task = asyncio.ensure_future(_collect())
+            # Give the pub/sub subscription a moment to establish.
+            await asyncio.sleep(0.1)
+            backend.publish_worker_telemetry(snapshot)
+
+            received = await asyncio.wait_for(task, timeout=5.0)
+            assert "testhost" in received.nodes
+            r_node = received.nodes["testhost"]
+            assert r_node.hostname == "testhost"
+            assert r_node.pid == 200
+            assert r_node.cpu_percent == 30.0
+            assert r_node.process_count == 1
+            assert r_node.thread_count == 2
+            assert "default" in received.queues
+        finally:
+            backend.close()
+
+    async def test_worker_telemetry_empty_when_no_workers(self):
+        """The default subscribe_worker_telemetry yields an empty snapshot."""
+
+        class _StubBackend(ThreadmillTaskBackend):
+            def enqueue(self, *args, **kwargs):  # pragma: no cover
+                raise NotImplementedError
+
+        backend = _StubBackend(alias="test", params={})
+        received = await backend.subscribe_worker_telemetry().__anext__()
+        assert received.nodes == {}
+        assert received.queues == {}

tests/test_executor.py

@@ -55,12 +55,13 @@ def _task_result(task, *args, **kwargs) -> TaskResult:
 
 
 def _make_worker(*, max_tasks: int | None = None) -> WorkerProcess:
-    """Build an unstarted `WorkerProcess`."""
+    """Build an unstarted `WorkerProcess` with a fresh shared task counter."""
     return WorkerProcess(
         thread_count=1,
         max_tasks=max_tasks,
         backend_alias="default",
         queues=("default",),
+        total_counter=multiprocessing.Value("q", 0),
     )
 
 
@@ -190,6 +191,16 @@ def test_shutdown__shuts_down_worker_processes(self):
         executor.shutdown()
         assert not worker.is_alive()
 
+    def test_shutdown__stops_telemetry_sampler(self):
+        """Shutdown stops the telemetry sampler when one is running."""
+        from unittest.mock import MagicMock
+
+        sampler = MagicMock()
+        executor = TaskExecutor(backend=default_task_backend, queues=("default",))
+        executor.telemetry_sampler = sampler
+        executor.shutdown()
+        sampler.stop.assert_called_once()
+
     def test_maintain_worker_pool__restarts_dead_workers(self):
         """maintain_worker_pool replaces dead workers with new ones."""
         executor = TaskExecutor(
@@ -217,12 +228,13 @@ class TestWorkerProcess:
     """Tests for the WorkerProcess class."""
 
     def test_record_task__increments_count(self):
-        """record_task increments task_count."""
+        """record_task increments task_count and the shared counter."""
         worker = _make_worker(max_tasks=5)
         worker.lock = threading.Lock()
         worker.expired = threading.Event()
         worker.record_task()
         assert worker.task_count == 1
+        assert worker.total_counter.value == 1
 
     def test_record_task__sets_expired_when_max_reached(self):
         """record_task sets expired event when max_tasks is reached."""
@@ -233,19 +245,21 @@ def test_record_task__sets_expired_when_max_reached(self):
         assert worker.expired.is_set()
 
     def test_record_task__noop_when_max_tasks_is_none(self):
-        """record_task is a no-op when max_tasks is None."""
+        """record_task still counts throughput when max_tasks is None."""
         worker = _make_worker(max_tasks=None)
         worker.lock = threading.Lock()
         worker.expired = threading.Event()
         worker.record_task()
         assert worker.task_count == 0
         assert not worker.expired.is_set()
+        assert worker.total_counter.value == 1
 
     def test_record_task__noop_before_run_sets_lock_and_expired(self):
-        """record_task is a safe no-op before run() initializes lock/expired."""
+        """record_task counts but skips recycling before run() sets lock/expired."""
         worker = _make_worker(max_tasks=5)
         worker.record_task()
         assert worker.task_count == 0
+        assert worker.total_counter.value == 1
 
     def test_shutdown_requested__is_settable(self):
         """shutdown_requested event can be set on an unstarted worker."""