gui: add consumers admin page (view + delete JetStream consumers)

Adds GET /consumers (list all stream consumers grouped by stream) and
POST /consumers/{stream}/{consumer}/delete. Archive-* consumers are
protected in both the template (no delete button rendered) and the POST
handler (hard refuse before touching NATS). CSRF validated, audit
logged via CONSUMER_DELETE action, DB conn acquired same pattern as
api_keys_delete.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/central/gui/audit.py

@@ -14,6 +14,7 @@ STREAM_UPDATE = "stream.update"
 API_KEY_CREATE = "api_key.create"
 API_KEY_ROTATE = "api_key.rotate"
 API_KEY_DELETE = "api_key.delete"
+CONSUMER_DELETE = "consumer.delete"
 SYSTEM_UPDATE = "system.update"
 MONITORING_AREA_CREATE = "monitoring_area.create"
 MONITORING_AREA_UPDATE = "monitoring_area.update"

src/central/gui/routes.py

@@ -44,6 +44,7 @@ from central.gui.audit import (
     AUTH_LOGIN_FAILED,
     AUTH_LOGOUT,
     AUTH_PASSWORD_CHANGE,
+    CONSUMER_DELETE,
     MONITORING_AREA_CREATE,
     MONITORING_AREA_DELETE,
     MONITORING_AREA_UPDATE,
@@ -2194,6 +2195,120 @@ async def streams_update(
     return RedirectResponse(url="/streams", status_code=302)
 
 
+# =============================================================================
+# Consumers routes
+# =============================================================================
+
+
+@router.get("/consumers", response_class=HTMLResponse)
+async def consumers_list(request: Request) -> HTMLResponse:
+    """List all JetStream consumers across all registered streams."""
+    from central.gui.nats import get_js
+
+    templates = _get_templates()
+    operator = request.state.operator
+    js = get_js()
+
+    streams_data = []
+    for stream_entry in STREAM_REGISTRY:
+        stream_name = stream_entry.name
+        consumers = []
+        stream_error = None
+
+        if js is not None:
+            try:
+                async for ci in js.consumers_info(stream_name):
+                    consumers.append({
+                        "name": ci.name,
+                        "num_pending": ci.num_pending,
+                        "num_ack_pending": ci.num_ack_pending,
+                        "num_redelivered": ci.num_redelivered,
+                        "num_waiting": ci.num_waiting,
+                        "created": ci.created,
+                        "protected": ci.name.startswith("archive-"),
+                    })
+            except Exception as e:
+                logger.warning(
+                    "consumers_info failed",
+                    extra={"stream": stream_name, "err": type(e).__name__},
+                )
+                stream_error = f"unavailable: {type(e).__name__}"
+        else:
+            stream_error = "NATS unavailable"
+
+        streams_data.append({
+            "stream": stream_name,
+            "consumers": consumers,
+            "error": stream_error,
+        })
+
+    csrf_token = request.state.csrf_token
+    response = templates.TemplateResponse(
+        request=request,
+        name="consumers_list.html",
+        context={
+            "operator": operator,
+            "csrf_token": csrf_token,
+            "streams": streams_data,
+        },
+    )
+    return response
+
+
+@router.post("/consumers/{stream}/{consumer}/delete", response_class=HTMLResponse)
+async def consumers_delete(request: Request, stream: str, consumer: str) -> Response:
+    """Delete a JetStream consumer."""
+    pool = get_pool()
+    operator = request.state.operator
+
+    form = await request.form()
+    form_csrf = form.get("csrf_token", "")
+    if not form_csrf or form_csrf != request.state.csrf_token:
+        raise CsrfValidationError("Invalid CSRF token")
+
+    # Hard guard: never delete archive-* consumers even if path is forged
+    if consumer.startswith("archive-"):
+        return RedirectResponse("/consumers", status_code=302)
+
+    js = get_js()
+    if js is None:
+        return RedirectResponse("/consumers", status_code=302)
+
+    # Capture before state from NATS for audit log
+    try:
+        before_info = await js.consumer_info(stream, consumer)
+        before = {
+            "name": before_info.name,
+            "stream": stream,
+            "num_pending": before_info.num_pending,
+        }
+    except Exception:
+        before = {"name": consumer, "stream": stream}
+
+    # Delete the consumer
+    try:
+        await js.delete_consumer(stream, consumer)
+    except Exception:
+        logger.exception(
+            "delete_consumer failed",
+            extra={"stream": stream, "consumer": consumer},
+        )
+        return RedirectResponse("/consumers", status_code=302)
+
+    # Write audit log
+    async with pool.acquire() as conn:
+        await write_audit(
+            conn,
+            CONSUMER_DELETE,
+            operator_id=operator.id,
+            target=f"{stream}/{consumer}",
+            before=before,
+            after=None,
+        )
+
+    return RedirectResponse("/consumers", status_code=302)
+
+
 # =============================================================================
 # Enrichment config route
 # =============================================================================

src/central/gui/templates/base.html

@@ -18,6 +18,7 @@
             <a href="/events">Events</a>
             <a href="/telemetry">Telemetry</a>
             <a href="/streams">Streams</a>
+            <a href="/consumers">Consumers</a>
             <a href="/enrichment">Enrichment</a>
             <a href="/monitoring-area">Monitoring Area</a>
             <a href="/api-keys">API Keys</a>

src/central/gui/templates/consumers_list.html

@@ -0,0 +1,70 @@
+{% extends "base.html" %}
+
+{% block title %}Central — Consumers{% endblock %}
+
+{% block content %}
+<h1>Consumers</h1>
+<p class="muted">JetStream consumers across all registered streams. A consumer with high
+<em>Pending</em> and zero <em>Waiting</em> has accumulated unacknowledged messages and
+has no active subscriber — it is safe to delete if it is not a central-owned consumer.</p>
+
+<div class="cols">
+{% for stream in streams %}
+<article>
+    <header><strong>{{ stream.stream }}</strong></header>
+
+    {% if stream.error %}
+    <p class="error" style="margin-top: 0.5rem;">({{ stream.error }})</p>
+    {% elif stream.consumers %}
+    <table style="width: 100%; border-collapse: collapse; margin-top: 0.5rem;">
+        <thead>
+            <tr>
+                <th style="text-align: left; padding: 0.25rem 0.5rem;">Name</th>
+                <th style="text-align: right; padding: 0.25rem 0.5rem;">Pending</th>
+                <th style="text-align: right; padding: 0.25rem 0.5rem;">Ack Pending</th>
+                <th style="text-align: right; padding: 0.25rem 0.5rem;">Redelivered</th>
+                <th style="text-align: right; padding: 0.25rem 0.5rem;">Waiting</th>
+                <th style="text-align: left; padding: 0.25rem 0.5rem;">Created</th>
+                <th style="text-align: center; padding: 0.25rem 0.5rem;">Action</th>
+            </tr>
+        </thead>
+        <tbody>
+        {% for c in stream.consumers %}
+            <tr>
+                <td style="padding: 0.25rem 0.5rem;">{{ c.name }}</td>
+                <td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_pending }}</td>
+                <td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_ack_pending }}</td>
+                <td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_redelivered }}</td>
+                <td style="text-align: right; padding: 0.25rem 0.5rem;">{{ c.num_waiting }}</td>
+                <td style="padding: 0.25rem 0.5rem;">{{ c.created.isoformat() if c.created else '—' }}</td>
+                <td style="text-align: center; padding: 0.25rem 0.5rem;">
+                    {% if c.protected %}
+                    <span class="muted" style="font-size: 0.85em;">central-owned</span>
+                    {% else %}
+                    <form method="post" action="/consumers/{{ stream.stream }}/{{ c.name }}/delete" style="margin: 0;">
+                        <input type="hidden" name="csrf_token" value="{{ csrf_token }}">
+                        <button type="submit" class="btn-danger" style="height: 28px; padding: 0 10px; font-size: 0.85em;"
+                                onclick="return confirm('Delete consumer {{ c.name }} on {{ stream.stream }}? This cannot be undone.')">Delete</button>
+                    </form>
+                    {% endif %}
+                </td>
+            </tr>
+        {% endfor %}
+        </tbody>
+    </table>
+    {% else %}
+    <p class="muted" style="margin-top: 0.5rem;">(no consumers)</p>
+    {% endif %}
+</article>
+{% endfor %}
+</div>
+
+<p class="muted" style="margin-top: 1rem; font-size: 0.9em;">
+    <strong>Legend:</strong> <em>Pending</em> = messages not yet delivered to this consumer;
+    <em>Ack Pending</em> = delivered but not yet acknowledged;
+    <em>Waiting</em> = active pull requests from a live subscriber.
+    A consumer with high <em>Pending</em> and zero <em>Waiting</em> is abandoned — no subscriber
+    is pulling from it and messages are piling up.
+    Consumers marked <em>central-owned</em> (archive-*) are managed by central and cannot be deleted here.
+</p>
+{% endblock %}

tests/test_consumers.py

@@ -0,0 +1,277 @@
+"""Tests for consumers admin routes (GET /consumers, POST /consumers/{s}/{c}/delete)."""
+
+import os
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+# Set required env vars before importing central modules
+os.environ.setdefault("CENTRAL_DB_DSN", "postgresql://test:test@localhost/test")
+os.environ.setdefault("CENTRAL_CSRF_SECRET", "testsecret12345678901234567890ab")
+os.environ.setdefault("CENTRAL_NATS_URL", "nats://localhost:4222")
+
+
+def _make_consumer_info(name: str, num_pending: int = 0, num_ack_pending: int = 0,
+                        num_redelivered: int = 0, num_waiting: int = 0):
+    ci = MagicMock()
+    ci.name = name
+    ci.num_pending = num_pending
+    ci.num_ack_pending = num_ack_pending
+    ci.num_redelivered = num_redelivered
+    ci.num_waiting = num_waiting
+    ci.created = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc)
+    return ci
+
+
+def _make_js_with_consumers(consumers_by_stream: dict):
+    """Build a mock JetStreamContext whose consumers_info is an async generator."""
+    mock_js = MagicMock()
+
+    async def consumers_info(stream_name):
+        for ci in consumers_by_stream.get(stream_name, []):
+            yield ci
+
+    mock_js.consumers_info = consumers_info
+    mock_js.consumer_info = AsyncMock()
+    mock_js.delete_consumer = AsyncMock()
+    return mock_js
+
+
+class TestConsumersListNatsUnavailable:
+    """GET /consumers when NATS is down shows per-stream error."""
+
+    @pytest.mark.asyncio
+    async def test_nats_unavailable_shows_error_per_stream(self):
+        from central.gui.routes import consumers_list
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1, username="testop")
+        mock_request.state.csrf_token = "test_csrf"
+
+        mock_templates = MagicMock()
+        mock_templates.TemplateResponse.return_value = MagicMock()
+
+        with patch("central.gui.routes._get_templates", return_value=mock_templates):
+            with patch("central.gui.nats.get_js", return_value=None):
+                await consumers_list(mock_request)
+
+        call_args = mock_templates.TemplateResponse.call_args
+        context = call_args.kwargs.get("context", call_args[1].get("context"))
+        streams = context["streams"]
+        # All streams should show the NATS unavailable error
+        assert all(s["error"] == "NATS unavailable" for s in streams)
+        # And no consumers listed
+        assert all(s["consumers"] == [] for s in streams)
+
+
+class TestConsumersListWithConsumers:
+    """GET /consumers with live NATS returns consumers per stream."""
+
+    @pytest.mark.asyncio
+    async def test_consumers_listed_with_protected_flag(self):
+        from central.gui.routes import consumers_list
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1, username="testop")
+        mock_request.state.csrf_token = "test_csrf"
+
+        mock_templates = MagicMock()
+        mock_templates.TemplateResponse.return_value = MagicMock()
+
+        consumers_by_stream = {
+            "CENTRAL_WX": [
+                _make_consumer_info("archive-CENTRAL_WX", num_pending=5, num_waiting=1),
+                _make_consumer_info("meshai-wx", num_pending=1000, num_waiting=0),
+            ],
+        }
+        mock_js = _make_js_with_consumers(consumers_by_stream)
+
+        with patch("central.gui.routes._get_templates", return_value=mock_templates):
+            with patch("central.gui.nats.get_js", return_value=mock_js):
+                await consumers_list(mock_request)
+
+        call_args = mock_templates.TemplateResponse.call_args
+        context = call_args.kwargs.get("context", call_args[1].get("context"))
+        streams = context["streams"]
+
+        wx = next(s for s in streams if s["stream"] == "CENTRAL_WX")
+        assert wx["error"] is None
+        assert len(wx["consumers"]) == 2
+
+        archive_c = next(c for c in wx["consumers"] if c["name"] == "archive-CENTRAL_WX")
+        assert archive_c["protected"] is True
+        assert archive_c["num_pending"] == 5
+
+        meshai_c = next(c for c in wx["consumers"] if c["name"] == "meshai-wx")
+        assert meshai_c["protected"] is False
+        assert meshai_c["num_pending"] == 1000
+        assert meshai_c["num_waiting"] == 0
+
+    @pytest.mark.asyncio
+    async def test_stream_with_no_consumers_shows_empty(self):
+        from central.gui.routes import consumers_list
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1, username="testop")
+        mock_request.state.csrf_token = "test_csrf"
+
+        mock_templates = MagicMock()
+        mock_templates.TemplateResponse.return_value = MagicMock()
+
+        mock_js = _make_js_with_consumers({})  # No consumers on any stream
+
+        with patch("central.gui.routes._get_templates", return_value=mock_templates):
+            with patch("central.gui.nats.get_js", return_value=mock_js):
+                await consumers_list(mock_request)
+
+        call_args = mock_templates.TemplateResponse.call_args
+        context = call_args.kwargs.get("context", call_args[1].get("context"))
+        streams = context["streams"]
+        assert all(s["consumers"] == [] for s in streams)
+        assert all(s["error"] is None for s in streams)
+
+    @pytest.mark.asyncio
+    async def test_one_stream_error_does_not_break_page(self):
+        from central.gui.routes import consumers_list
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1, username="testop")
+        mock_request.state.csrf_token = "test_csrf"
+
+        mock_templates = MagicMock()
+        mock_templates.TemplateResponse.return_value = MagicMock()
+
+        mock_js = MagicMock()
+
+        async def consumers_info_raising(stream_name):
+            if stream_name == "CENTRAL_FIRE":
+                raise RuntimeError("stream not found")
+            # other streams: empty
+            return
+            yield  # make it an async generator
+
+        mock_js.consumers_info = consumers_info_raising
+
+        with patch("central.gui.routes._get_templates", return_value=mock_templates):
+            with patch("central.gui.nats.get_js", return_value=mock_js):
+                await consumers_list(mock_request)
+
+        call_args = mock_templates.TemplateResponse.call_args
+        context = call_args.kwargs.get("context", call_args[1].get("context"))
+        streams = context["streams"]
+
+        fire = next(s for s in streams if s["stream"] == "CENTRAL_FIRE")
+        assert "unavailable" in fire["error"]
+        assert fire["consumers"] == []
+
+
+class TestConsumersDeleteArchiveGuard:
+    """POST /consumers/{stream}/archive-*/delete must be refused."""
+
+    @pytest.mark.asyncio
+    async def test_archive_consumer_refused_redirects(self):
+        from central.gui.routes import consumers_delete
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1)
+        mock_request.state.csrf_token = "tok"
+        form_data = MagicMock()
+        form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d)
+        mock_request.form = AsyncMock(return_value=form_data)
+
+        with patch("central.gui.nats.get_js", return_value=MagicMock()):
+            result = await consumers_delete(mock_request, "CENTRAL_WX", "archive-CENTRAL_WX")
+
+        assert result.status_code == 302
+        assert result.headers["location"] == "/consumers"
+
+
+class TestConsumersDeleteSuccess:
+    """POST /consumers/{stream}/{consumer}/delete happy path."""
+
+    @pytest.mark.asyncio
+    async def test_delete_non_protected_consumer_audits_and_redirects(self):
+        from central.gui.routes import consumers_delete
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1)
+        mock_request.state.csrf_token = "tok"
+        form_data = MagicMock()
+        form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d)
+        mock_request.form = AsyncMock(return_value=form_data)
+
+        before_ci = _make_consumer_info("meshai-wx", num_pending=500)
+        mock_js = MagicMock()
+        mock_js.consumer_info = AsyncMock(return_value=before_ci)
+        mock_js.delete_consumer = AsyncMock()
+
+        mock_conn = AsyncMock()
+        mock_pool = MagicMock()
+        mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn)
+        mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
+
+        captured_audit = {}
+
+        async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None):
+            captured_audit["action"] = action
+            captured_audit["operator_id"] = operator_id
+            captured_audit["target"] = target
+            captured_audit["before"] = before
+            captured_audit["after"] = after
+
+        with patch("central.gui.nats.get_js", return_value=mock_js):
+            with patch("central.gui.routes.get_pool", return_value=mock_pool):
+                with patch("central.gui.routes.write_audit", side_effect=capture_audit):
+                    result = await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx")
+
+        assert result.status_code == 302
+        assert result.headers["location"] == "/consumers"
+
+        mock_js.delete_consumer.assert_awaited_once_with("CENTRAL_WX", "meshai-wx")
+
+        assert captured_audit["action"] == "consumer.delete"
+        assert captured_audit["operator_id"] == 1
+        assert captured_audit["target"] == "CENTRAL_WX/meshai-wx"
+        assert captured_audit["before"]["name"] == "meshai-wx"
+        assert captured_audit["after"] is None
+
+
+class TestConsumersDeleteCsrfGuard:
+    """POST /consumers/{stream}/{consumer}/delete CSRF mismatch raises."""
+
+    @pytest.mark.asyncio
+    async def test_csrf_mismatch_raises(self):
+        from central.gui.routes import consumers_delete
+        from central.gui.auth import CsrfValidationError
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1)
+        mock_request.state.csrf_token = "real_token"
+        form_data = MagicMock()
+        form_data.get.side_effect = lambda k, d="": {"csrf_token": "wrong_token"}.get(k, d)
+        mock_request.form = AsyncMock(return_value=form_data)
+
+        with pytest.raises(CsrfValidationError):
+            await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx")
+
+
+class TestConsumersDeleteNatsUnavailable:
+    """POST /consumers/{stream}/{consumer}/delete when NATS is down redirects."""
+
+    @pytest.mark.asyncio
+    async def test_nats_unavailable_redirects(self):
+        from central.gui.routes import consumers_delete
+
+        mock_request = MagicMock()
+        mock_request.state.operator = MagicMock(id=1)
+        mock_request.state.csrf_token = "tok"
+        form_data = MagicMock()
+        form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d)
+        mock_request.form = AsyncMock(return_value=form_data)
+
+        with patch("central.gui.nats.get_js", return_value=None):
+            result = await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx")
+
+        assert result.status_code == 302
+        assert result.headers["location"] == "/consumers"