feat: enable redis task queue in docker setup

Author: haasonsaas

.gitignore

@@ -1,3 +1,4 @@
+.coverage
 .env
 __pycache__/
 *.pyc

agent_pm/storage/tasks.py

@@ -61,7 +61,7 @@ def __init__(self, max_workers: int = 5):
         self.running = False
         self._lock = asyncio.Lock()
 
-    def start(self):
+    async def start(self):
         """Start background workers."""
         if self.running:
             return
@@ -203,5 +203,5 @@ async def enqueue(  # type: ignore[override]
             _task_queue = RedisTaskQueue(max_workers=settings.task_queue_workers)
         else:
             _task_queue = TaskQueue(max_workers=settings.task_queue_workers)
-            _task_queue.start()
+            await _task_queue.start()
     return _task_queue

app.py

@@ -59,7 +59,7 @@
     get_correlation_id,
     set_correlation_id,
 )
-from agent_pm.storage.tasks import TaskQueue, TaskStatus, get_task_queue
+from agent_pm.storage.tasks import TaskStatus, get_task_queue
 from agent_pm.tools import registry
 from agent_pm.observability.export import schedule_trace_export
 from agent_pm.observability.traces import list_traces as list_trace_files
@@ -73,7 +73,7 @@
 logger = logging.getLogger(__name__)
 app = FastAPI(title="Agent PM", version="0.1.0")
 _jira_lock = asyncio.Lock()
-_task_queue: TaskQueue | None = None
+_task_queue = None
 
 plugin_registry.attach_app(app)
 
@@ -86,18 +86,14 @@ class FollowupUpdate(BaseModel):
 async def startup_event():
     global _task_queue
     _task_queue = await get_task_queue()
-    if settings.task_queue_backend == "memory" and isinstance(_task_queue, TaskQueue):
-        _task_queue.start()
+    if settings.task_queue_backend == "memory":
+        await _task_queue.start()
     logger.info("Agent PM service started")
 
 
 @app.on_event("shutdown")
 async def shutdown_event():
-    if (
-        _task_queue
-        and settings.task_queue_backend == "memory"
-        and hasattr(_task_queue, "stop")
-    ):
+    if settings.task_queue_backend == "memory" and _task_queue:
         await _task_queue.stop()
     logger.info("Agent PM service stopped")
 

docker-compose.yml

@@ -23,6 +23,7 @@ services:
       - "6379:6379"
     volumes:
       - redis_data:/data
+    command: redis-server --save 60 1 --loglevel warning
     healthcheck:
       test: ["CMD", "redis-cli", "ping"]
       interval: 10s
@@ -36,7 +37,8 @@ services:
     environment:
       - OPENAI_API_KEY=${OPENAI_API_KEY}
       - DATABASE_URL=postgresql+asyncpg://agent_pm:agent_pm_dev@postgres:5432/agent_pm
-      - REDIS_URL=redis://redis:6379
+      - REDIS_URL=redis://redis:6379/0
+      - TASK_QUEUE_BACKEND=redis
       - DRY_RUN=true
       - LOG_FORMAT=json
       - ENABLE_OPENTELEMETRY=false
@@ -45,20 +47,7 @@ services:
         condition: service_healthy
       redis:
         condition: service_healthy
-    command: uvicorn app:app --host 0.0.0.0 --port 8000 --reload
-
-  arq_worker:
-    build: .
-    environment:
-      - OPENAI_API_KEY=${OPENAI_API_KEY}
-      - DATABASE_URL=postgresql+asyncpg://agent_pm:agent_pm_dev@postgres:5432/agent_pm
-      - REDIS_URL=redis://redis:6379
-    depends_on:
-      postgres:
-        condition: service_healthy
-      redis:
-        condition: service_healthy
-    command: arq agent_pm.redis_queue.WorkerSettings
+    command: uv run uvicorn app:app --host 0.0.0.0 --port 8000 --reload
 
 volumes:
   postgres_data:

tests/storage/test_task_queue.py

@@ -20,7 +20,7 @@ async def failing_task(counter: dict[str, int]) -> None:
 @pytest.mark.asyncio
 async def test_task_queue_executes_successfully():
     queue = TaskQueue(max_workers=1)
-    queue.start()
+    await queue.start()
 
     task_id = await queue.enqueue("double", sample_task, 3)
 
@@ -43,7 +43,7 @@ async def test_task_queue_executes_successfully():
 @pytest.mark.asyncio
 async def test_task_queue_retries_then_fails(monkeypatch):
     queue = TaskQueue(max_workers=1)
-    queue.start()
+    await queue.start()
 
     counter: dict[str, int] = {}
 

tests/test_task_queue.py

@@ -19,7 +19,7 @@ async def failing_task() -> None:
 @pytest.mark.asyncio
 async def test_task_queue_enqueue_and_execute():
     queue = TaskQueue(max_workers=2)
-    queue.start()
+    await queue.start()
     try:
         task_id = await queue.enqueue("sample_task", sample_task, 5)
         assert task_id
@@ -37,7 +37,7 @@ async def test_task_queue_enqueue_and_execute():
 @pytest.mark.asyncio
 async def test_task_queue_retry_logic():
     queue = TaskQueue(max_workers=1)
-    queue.start()
+    await queue.start()
     try:
         task_id = await queue.enqueue("failing_task", failing_task, max_retries=2)
         assert task_id
@@ -56,7 +56,7 @@ async def test_task_queue_retry_logic():
 @pytest.mark.asyncio
 async def test_task_queue_list_tasks():
     queue = TaskQueue(max_workers=2)
-    queue.start()
+    await queue.start()
     try:
         task_id1 = await queue.enqueue("task1", sample_task, 1)
         task_id2 = await queue.enqueue("task2", sample_task, 2)