File Records
+
+
+
+
+
+
+ 0 item(s) are currently queued or being processed.
+ Files will appear here once processing completes.
+
+
+
+ View Queue
+
+
+
+
{% if error %}
Error: {{ error }}
@@ -1050,5 +1067,35 @@
_searchCurrentPage = 1;
}
+
+
+
{% endblock %}
diff --git a/frontend/templates/queue_dashboard.html b/frontend/templates/queue_dashboard.html
new file mode 100644
index 00000000..20f7fdbe
--- /dev/null
+++ b/frontend/templates/queue_dashboard.html
@@ -0,0 +1,319 @@
+{% extends "base.html" %}
+{% block title %}Queue Monitor — DocuElevate{% endblock %}
+
+{% block head_extra %}
+
+{% endblock %}
+
+{% block content %}
+
+
+
+
+
+ Queue Monitor
+
+
Real-time view of the document processing pipeline and Celery task queues.
+
+
+
+
+
+
Loading queue statistics…
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Redis Queues
+
+
+
+
+
+
+ | Queue |
+ Pending |
+
+
+
+ | No data |
+
+
+
+
+
+
+
+
+
+ Processing Pipeline
+
+
+
+
+
+
+ | State |
+ Files |
+
+
+
+ | No data |
+
+
+
+
+
+
+
+
+
+
+ Active Tasks
+ (0)
+
+
+
+
+
+
+ | Task |
+ Arguments |
+ Task ID |
+
+
+
+ | No active tasks |
+
+
+
+
+
+
+
+
+
+ Recently Processing Files
+
+
+
+
+
+
+ | File |
+ Current Step |
+ Actions |
+
+
+
+ | No files currently processing |
+
+
+
+
+
+
+
+
+ Auto-refreshes every 10 seconds —
+ Last updated: —
+
+
+
+
+
+{% endblock %}
diff --git a/tests/test_queue_monitoring.py b/tests/test_queue_monitoring.py
new file mode 100644
index 00000000..256948b1
--- /dev/null
+++ b/tests/test_queue_monitoring.py
@@ -0,0 +1,241 @@
+"""Tests for app/api/queue.py and app/views/queue.py modules."""
+
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+
+
+@pytest.mark.unit
+class TestGetRedisQueueLength:
+ """Tests for the _get_redis_queue_length helper."""
+
+ def test_returns_queue_length(self):
+ """Test returns queue length from Redis."""
+ from app.api.queue import _get_redis_queue_length
+
+ mock_redis = MagicMock()
+ mock_redis.llen.return_value = 42
+ assert _get_redis_queue_length(mock_redis, "document_processor") == 42
+ mock_redis.llen.assert_called_once_with("document_processor")
+
+ def test_returns_zero_on_error(self):
+ """Test returns 0 when Redis call fails."""
+ from app.api.queue import _get_redis_queue_length
+
+ mock_redis = MagicMock()
+ mock_redis.llen.side_effect = Exception("Connection refused")
+ assert _get_redis_queue_length(mock_redis, "default") == 0
+
+
+@pytest.mark.unit
+class TestGetCeleryInspectStats:
+ """Tests for the _get_celery_inspect_stats helper."""
+
+ @patch("app.celery_app.celery")
+ def test_returns_worker_stats(self, mock_celery_mod):
+ """Test returns active, reserved, scheduled tasks."""
+ from app.api.queue import _get_celery_inspect_stats
+
+ mock_inspector = MagicMock()
+ mock_inspector.active.return_value = {
+ "worker1": [
+ {"id": "task-1", "name": "app.tasks.process_document.process_document", "args": [1], "time_start": 123}
+ ]
+ }
+ mock_inspector.reserved.return_value = {
+ "worker1": [{"id": "task-2", "name": "app.tasks.upload_to_s3.upload_to_s3", "args": [2]}]
+ }
+ mock_inspector.scheduled.return_value = {
+ "worker1": [{"request": {"id": "task-3", "name": "app.tasks.check_credentials.check_credentials"}, "eta": "2026-01-01"}]
+ }
+ mock_celery_mod.control.inspect.return_value = mock_inspector
+
+ result = _get_celery_inspect_stats()
+
+ assert result["workers_online"] == 1
+ assert len(result["active"]) == 1
+ assert result["active"][0]["id"] == "task-1"
+ assert len(result["reserved"]) == 1
+ assert len(result["scheduled"]) == 1
+
+ @patch("app.celery_app.celery")
+ def test_handles_no_workers(self, mock_celery_mod):
+ """Test handles case where no workers are online."""
+ from app.api.queue import _get_celery_inspect_stats
+
+ mock_inspector = MagicMock()
+ mock_inspector.active.return_value = None
+ mock_inspector.reserved.return_value = None
+ mock_inspector.scheduled.return_value = None
+ mock_celery_mod.control.inspect.return_value = mock_inspector
+
+ result = _get_celery_inspect_stats()
+
+ assert result["workers_online"] == 0
+ assert result["active"] == []
+ assert result["reserved"] == []
+ assert result["scheduled"] == []
+
+ @patch("app.celery_app.celery")
+ def test_handles_inspect_exception(self, mock_celery_mod):
+ """Test handles exception during inspect."""
+ from app.api.queue import _get_celery_inspect_stats
+
+ mock_celery_mod.control.inspect.side_effect = Exception("Broker unreachable")
+
+ result = _get_celery_inspect_stats()
+
+ assert result["workers_online"] == 0
+ assert result["active"] == []
+
+
+@pytest.mark.unit
+class TestGetDbProcessingSummary:
+ """Tests for the _get_db_processing_summary helper."""
+
+ def test_returns_summary(self, db_session):
+ """Test returns processing summary from DB."""
+ from app.api.queue import _get_db_processing_summary
+ from app.models import FileProcessingStep, FileRecord
+
+ # Create some test files
+ file1 = FileRecord(filehash="abc1", local_filename="f1.pdf", file_size=100, is_duplicate=False)
+ file2 = FileRecord(filehash="abc2", local_filename="f2.pdf", file_size=200, is_duplicate=False)
+ db_session.add_all([file1, file2])
+ db_session.commit()
+
+ # Add steps: file1 completed, file2 in_progress
+ step1 = FileProcessingStep(file_id=file1.id, step_name="extract_text", status="success")
+ step2 = FileProcessingStep(file_id=file2.id, step_name="extract_text", status="in_progress")
+ db_session.add_all([step1, step2])
+ db_session.commit()
+
+ result = _get_db_processing_summary(db_session)
+
+ assert result["total_files"] == 2
+ assert result["processing"] == 1
+ assert isinstance(result["recent_processing"], list)
+
+ def test_returns_empty_on_error(self):
+ """Test returns empty summary on DB error."""
+ from app.api.queue import _get_db_processing_summary
+
+ mock_db = MagicMock()
+ mock_db.query.side_effect = Exception("DB error")
+
+ result = _get_db_processing_summary(mock_db)
+
+ assert result["total_files"] == 0
+ assert result["processing"] == 0
+ assert result["recent_processing"] == []
+
+
+@pytest.mark.integration
+class TestQueueStatsEndpoint:
+ """Tests for the GET /api/queue/stats endpoint."""
+
+ @patch("app.api.queue.redis.Redis")
+ @patch("app.celery_app.celery")
+ def test_queue_stats_returns_200(self, mock_celery_mod, mock_redis_cls, client):
+ """Test queue stats endpoint returns 200 with data."""
+ # Mock Redis
+ mock_redis_instance = MagicMock()
+ mock_redis_instance.llen.return_value = 5
+ mock_redis_cls.from_url.return_value = mock_redis_instance
+
+ # Mock Celery inspector
+ mock_inspector = MagicMock()
+ mock_inspector.active.return_value = {}
+ mock_inspector.reserved.return_value = {}
+ mock_inspector.scheduled.return_value = {}
+ mock_celery_mod.control.inspect.return_value = mock_inspector
+
+ response = client.get("/api/queue/stats")
+ assert response.status_code == 200
+
+ data = response.json()
+ assert "queues" in data
+ assert "total_queued" in data
+ assert "celery" in data
+ assert "db_summary" in data
+
+ @patch("app.api.queue.redis.Redis")
+ def test_queue_stats_handles_redis_error(self, mock_redis_cls, client):
+ """Test queue stats handles Redis connection failure."""
+ mock_redis_cls.from_url.side_effect = Exception("Redis unavailable")
+
+ response = client.get("/api/queue/stats")
+ assert response.status_code == 200
+ data = response.json()
+ assert data["total_queued"] == 0
+
+
+@pytest.mark.integration
+class TestPendingCountEndpoint:
+ """Tests for the GET /api/queue/pending-count endpoint."""
+
+ @patch("app.api.queue.redis.Redis")
+ def test_pending_count_returns_200(self, mock_redis_cls, client):
+ """Test pending count endpoint returns 200."""
+ mock_redis_instance = MagicMock()
+ mock_redis_instance.llen.return_value = 3
+ mock_redis_cls.from_url.return_value = mock_redis_instance
+
+ response = client.get("/api/queue/pending-count")
+ assert response.status_code == 200
+
+ data = response.json()
+ assert "total_pending" in data
+ assert data["total_pending"] >= 0
+
+ @patch("app.api.queue.redis.Redis")
+ def test_pending_count_redis_failure_still_works(self, mock_redis_cls, client):
+ """Test pending count still works when Redis is down."""
+ mock_redis_cls.from_url.side_effect = Exception("Connection refused")
+
+ response = client.get("/api/queue/pending-count")
+ assert response.status_code == 200
+ data = response.json()
+ assert "total_pending" in data
+
+
+@pytest.mark.integration
+class TestQueueDashboardView:
+ """Tests for the GET /admin/queue view."""
+
+ def test_queue_dashboard_requires_login(self, client):
+ """Test queue dashboard requires authentication."""
+ response = client.get("/admin/queue", follow_redirects=False)
+ assert response.status_code in [200, 302, 401]
+
+ def test_queue_dashboard_non_admin_redirect(self, client):
+ """Test queue dashboard redirects non-admin users."""
+ # Set non-admin session
+ with client:
+ client.cookies.set("session", "test")
+ response = client.get("/admin/queue", follow_redirects=False)
+ # Should redirect or deny since no admin session
+ assert response.status_code in [200, 302, 401]
+
+
+@pytest.mark.unit
+class TestQueueDashboardViewFunction:
+ """Tests for the queue_dashboard view function."""
+
+ @patch("app.views.queue.templates")
+ @pytest.mark.asyncio
+ async def test_queue_dashboard_returns_template(self, mock_templates):
+ """Test queue dashboard returns template response."""
+ from app.views.queue import queue_dashboard
+
+ mock_request = Mock()
+ mock_request.session = {"user": {"is_admin": True}}
+ mock_db = MagicMock()
+
+ result = await queue_dashboard(mock_request, db=mock_db)
+
+ mock_templates.TemplateResponse.assert_called_once()
+ call_args = mock_templates.TemplateResponse.call_args
+ assert call_args[0][0] == "queue_dashboard.html"
+ context = call_args[0][1]
+ assert "request" in context
From 4dd018f21011b16be66b22b35a877e6f97959eae Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Fri, 27 Feb 2026 00:47:58 +0000
Subject: [PATCH 3/3] =?UTF-8?q?refactor(queue):=20address=20code=20review?=
=?UTF-8?q?=20feedback=20=E2=80=94=20extract=20constants=20and=20sync=20re?=
=?UTF-8?q?fresh=20interval?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
---
app/api/__init__.py | 2 +-
app/api/queue.py | 10 +++--
docs/API.md | 51 +++++++++++++++++++++++++
docs/UserGuide.md | 17 +++++++++
frontend/templates/queue_dashboard.html | 1 +
tests/test_queue_monitoring.py | 7 +++-
6 files changed, 83 insertions(+), 5 deletions(-)
diff --git a/app/api/__init__.py b/app/api/__init__.py
index ec050431..97f13099 100644
--- a/app/api/__init__.py
+++ b/app/api/__init__.py
@@ -15,9 +15,9 @@ from app.api.logs import router as logs_router
from app.api.onedrive import router as onedrive_router
from app.api.openai import router as openai_router
from app.api.process import router as process_router
+from app.api.queue import router as queue_router
from app.api.search import router as search_router
from app.api.settings import router as settings_router
-from app.api.queue import router as queue_router
from app.api.url_upload import router as url_upload_router
# Import all the individual routers
diff --git a/app/api/queue.py b/app/api/queue.py
index a2677678..fdc851af 100644
--- a/app/api/queue.py
+++ b/app/api/queue.py
@@ -20,6 +20,10 @@ from app.models import FileProcessingStep, FileRecord
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/queue", tags=["queue"])
+# Constants
+CELERY_INSPECT_TIMEOUT = 2.0
+MAX_ARGS_DISPLAY_LENGTH = 200
+
def _get_redis_queue_length(redis_client: redis.Redis, queue_name: str) -> int:
"""Get the number of messages in a Redis-backed Celery queue.
@@ -54,7 +58,7 @@ def _get_celery_inspect_stats() -> dict[str, Any]:
}
try:
- inspector = celery.control.inspect(timeout=2.0)
+ inspector = celery.control.inspect(timeout=CELERY_INSPECT_TIMEOUT)
active = inspector.active() or {}
reserved = inspector.reserved() or {}
@@ -68,7 +72,7 @@ def _get_celery_inspect_stats() -> dict[str, Any]:
{
"id": task.get("id", ""),
"name": task.get("name", "unknown"),
- "args": str(task.get("args", []))[:200],
+ "args": str(task.get("args", []))[:MAX_ARGS_DISPLAY_LENGTH],
"started": task.get("time_start"),
}
)
@@ -79,7 +83,7 @@ def _get_celery_inspect_stats() -> dict[str, Any]:
{
"id": task.get("id", ""),
"name": task.get("name", "unknown"),
- "args": str(task.get("args", []))[:200],
+ "args": str(task.get("args", []))[:MAX_ARGS_DISPLAY_LENGTH],
}
)
diff --git a/docs/API.md b/docs/API.md
index 4255f02b..f0d6f87e 100644
--- a/docs/API.md
+++ b/docs/API.md
@@ -437,6 +437,57 @@ Errors follow standard HTTP status codes with descriptive messages:
}
```
+## Queue Monitoring
+
+### GET /api/queue/stats
+
+Get comprehensive queue and processing statistics, including Redis queue lengths, Celery worker inspection data, and database-level processing summaries.
+
+**Authentication:** Required
+
+**Response (200 OK):**
+```json
+{
+ "queues": {
+ "document_processor": 12,
+ "default": 0,
+ "celery": 0
+ },
+ "total_queued": 12,
+ "celery": {
+ "active": [
+ {"id": "abc123", "name": "process_document", "args": "[42]", "started": 1700000000}
+ ],
+ "reserved": [],
+ "scheduled": [],
+ "workers_online": 1
+ },
+ "db_summary": {
+ "total_files": 5000,
+ "processing": 3,
+ "failed": 1,
+ "completed": 4900,
+ "pending": 96,
+ "recent_processing": [
+ {"file_id": 42, "filename": "invoice.pdf", "current_step": "extract_metadata_with_gpt"}
+ ]
+ }
+}
+```
+
+### GET /api/queue/pending-count
+
+Lightweight endpoint returning the total number of queued + in-progress items. Designed for the files page banner indicator.
+
+**Authentication:** Required
+
+**Response (200 OK):**
+```json
+{
+ "total_pending": 15
+}
+```
+
## Rate Limiting
The API implements rate limiting to ensure system stability. If you exceed the limits, you'll receive a `429 Too Many Requests` response.
diff --git a/docs/UserGuide.md b/docs/UserGuide.md
index 8f1d5921..39c83cb4 100644
--- a/docs/UserGuide.md
+++ b/docs/UserGuide.md
@@ -92,6 +92,23 @@ The **Files** page provides access to all processed documents:
3. Click on any file to view its details
4. Sort the list by any column by clicking on the column header
+> **Tip:** When documents are being processed, a blue banner appears at the top of the Files page showing how many items are queued or currently processing. Files will appear in the list once their processing completes. Admins can click "View Queue" in the banner to open the Queue Monitor dashboard.
+
+## Queue Monitor (Admin)
+
+The **Queue Monitor** dashboard provides real-time visibility into the document processing pipeline. It is available to admin users under **Admin → Queue Monitor** in the navigation bar.
+
+The dashboard shows:
+- **Queued Tasks** — number of tasks waiting in Redis-backed Celery queues
+- **Active Tasks** — tasks currently being executed by Celery workers
+- **Files Processing** — files with at least one in-progress processing step
+- **Workers Online** — number of connected Celery worker processes
+- **Redis Queues** — per-queue breakdown of pending task counts
+- **Processing Pipeline** — database-level summary of file states (completed, processing, pending, failed)
+- **Recently Processing Files** — the most recent files being actively processed, with links to their detail pages
+
+The dashboard auto-refreshes every 10 seconds.
+
## Searching Documents
DocuElevate provides two ways to search your documents:
diff --git a/frontend/templates/queue_dashboard.html b/frontend/templates/queue_dashboard.html
index 20f7fdbe..65c9fdea 100644
--- a/frontend/templates/queue_dashboard.html
+++ b/frontend/templates/queue_dashboard.html
@@ -189,6 +189,7 @@