feat(queue): requeue dead-letter tasks

Author: haasonsaas

agent_pm/storage/redis.py

@@ -74,6 +74,18 @@ async def record_dead_letter(client: redis.Redis, payload: dict[str, Any]) -> No
     await client.hset(_dead_letter_key(), task_id, json.dumps(payload))
 
 
+async def get_dead_letter(client: redis.Redis, task_id: str) -> dict[str, Any] | None:
+    item = await client.hget(_dead_letter_key(), task_id)
+    if not item:
+        return None
+    try:
+        data = json.loads(item)
+    except json.JSONDecodeError:
+        return None
+    data.setdefault("task_id", task_id)
+    return data
+
+
 async def fetch_dead_letters(client: redis.Redis, limit: int = 100) -> list[dict[str, Any]]:
     items = await client.hgetall(_dead_letter_key())
     tasks: list[dict[str, Any]] = []

agent_pm/storage/tasks.py

@@ -23,6 +23,7 @@
     clear_dead_letter,
     enqueue_task as redis_enqueue_task,
     fetch_dead_letters,
+    get_dead_letter,
     get_redis_client,
     list_heartbeats,
     pop_task,
@@ -206,6 +207,9 @@ async def delete_dead_letter(self, task_id: str) -> None:
     async def worker_heartbeats(self) -> dict[str, Any]:
         return {}
 
+    async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
+        return None
+
 
 # Global task queue instance
 _task_queue: TaskQueue | None = None
@@ -338,6 +342,20 @@ async def delete_dead_letter(self, task_id: str) -> None:
                 async def worker_heartbeats(self) -> dict[str, dict[str, Any]]:
                     return await list_heartbeats(self._redis)
 
+                async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
+                    payload = await get_dead_letter(self._redis, task_id)
+                    if payload is None:
+                        return None
+
+                    await clear_dead_letter(self._redis, task_id)
+                    payload.pop("last_error", None)
+                    payload["retry_count"] = 0
+                    payload["requeued_at"] = utc_now().isoformat()
+                    await redis_enqueue_task(self._redis, payload.get("name", "unknown"), payload)
+                    record_task_enqueued(self.queue_name)
+                    logger.info("Dead-letter task requeued: %s", task_id)
+                    return payload
+
             _task_queue = RedisTaskQueue(max_workers=settings.task_queue_workers)
         else:
             _task_queue = TaskQueue(max_workers=settings.task_queue_workers)

app.py

@@ -498,6 +498,16 @@ async def get_task(task_id: str, _admin_key: AdminKeyDep = None) -> dict[str, An
     }
 
 
+@app.post("/tasks/dead-letter/{task_id}/requeue")
+async def requeue_dead_letter(task_id: str, _admin_key: AdminKeyDep = None) -> dict[str, Any]:
+    if not _task_queue:
+        raise HTTPException(status_code=503, detail="Task queue not initialized")
+    payload = await _task_queue.requeue_dead_letter(task_id)
+    if payload is None:
+        raise HTTPException(status_code=404, detail="Dead-letter task not found")
+    return {"task_id": payload.get("task_id", task_id), "status": "requeued"}
+
+
 @app.post("/prd/{plan_id}/versions")
 async def create_prd_version(
     plan_id: str,

tests/storage/test_redis.py

@@ -20,6 +20,9 @@ async def lpop(self, key: str):
             return None
         return self.items.pop(0)
 
+    async def llen(self, key: str) -> int:
+        return len(self.items)
+
     async def hset(self, key: str, field: str, value: str) -> None:
         if key.endswith("dead_letter"):
             self.dead[field] = value
@@ -85,6 +88,9 @@ async def test_dead_letter_and_heartbeat_helpers():
     records = await redis.fetch_dead_letters(client)
     assert records[0]["task_id"] == "abc"
 
+    stored = await redis.get_dead_letter(client, "abc")
+    assert stored["task_id"] == "abc"
+
     await redis.clear_dead_letter(client, "abc")
     assert await redis.fetch_dead_letters(client) == []
 

tests/storage/test_redis_worker.py

@@ -22,6 +22,9 @@ async def lpop(self, key: str):
             return None
         return self.items.popleft()
 
+    async def llen(self, key: str) -> int:
+        return len(self.items)
+
     async def hset(self, key: str, field: str, value: str) -> None:
         self.hashes.setdefault(key, {})[field] = value
 
@@ -113,3 +116,11 @@ async def fail_task() -> None:
     assert match is not None
     assert match["retry_count"] == 1
     assert match["last_error"] == "boom"
+
+    requeued = await queue.requeue_dead_letter(task_id)
+    assert requeued is not None
+    assert requeued["task_id"] == task_id
+    assert requeued["retry_count"] == 0
+
+    queued_len = await fake.llen("agent_pm:tasks")
+    assert queued_len >= 1

tests/test_tasks_api.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 import pytest
 from httpx import ASGITransport, AsyncClient
 
@@ -7,22 +9,17 @@
 from app import app
 
 
-class AppClient:
-    def __init__(self):
-        self.transport = ASGITransport(app=app)
-        self.client = AsyncClient(transport=self.transport, base_url="http://test")
-
-    async def __aenter__(self):
-        await app_module.startup_event()
-        return self.client
+async def _create_client():
+    transport = ASGITransport(app=app)
+    client = AsyncClient(transport=transport, base_url="http://test")
+    await app_module.startup_event()
 
-    async def __aexit__(self, exc_type, exc, tb):
-        await self.client.aclose()
+    async def cleanup():
+        await client.aclose()
         await app_module.shutdown_event()
 
-
-async def _create_client():
-    return AppClient()
+    client.cleanup = cleanup  # type: ignore[attr-defined]
+    return client
 
 
 @pytest.mark.asyncio
@@ -31,14 +28,17 @@ async def test_tasks_admin_endpoints_with_memory_backend(monkeypatch):
     app.dependency_overrides[AdminKeyDep] = lambda: "admin-test-key"
 
     try:
-        async with await _create_client() as client:
+        client = await _create_client()
+        try:
             response = await client.get("/tasks/dead-letter")
             assert response.status_code == 200
             assert response.json() == {"dead_letter": [], "total": 0}
 
             worker_resp = await client.get("/tasks/workers")
             assert worker_resp.status_code == 200
             assert worker_resp.json() == {"workers": {}}
+        finally:
+            await client.cleanup()
     finally:
         app.dependency_overrides.clear()
 
@@ -47,6 +47,7 @@ class StubQueue:
     def __init__(self):
         self.limit: int | None = None
         self.deleted: str | None = None
+        self.requeued: str | None = None
 
     async def list_dead_letters(self, limit: int = 100):
         self.limit = limit
@@ -68,14 +69,19 @@ async def worker_heartbeats(self) -> dict[str, dict[str, str]]:
     async def get_task(self, task_id: str):  # pragma: no cover - minimal stub for routing
         return None
 
+    async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
+        self.requeued = task_id
+        return {"task_id": task_id}
+
 
 @pytest.mark.asyncio
 async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
     monkeypatch.setattr(settings, "task_queue_backend", "memory")
     app.dependency_overrides[AdminKeyDep] = lambda: "admin-test-key"
 
     try:
-        async with await _create_client() as client:
+        client = await _create_client()
+        try:
             original_backend = settings.task_queue_backend
             original_queue = app_module._task_queue
             stub = StubQueue()
@@ -94,8 +100,15 @@ async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
                 worker_resp = await client.get("/tasks/workers")
                 assert worker_resp.status_code == 200
                 assert worker_resp.json() == {"workers": {"worker:1": {"status": "ok"}}}
+
+                requeue_resp = await client.post("/tasks/dead-letter/dead-1/requeue")
+                assert requeue_resp.status_code == 200
+                assert stub.requeued == "dead-1"
+                assert requeue_resp.json()["status"] == "requeued"
             finally:
                 settings.task_queue_backend = original_backend
                 app_module._task_queue = original_queue
+        finally:
+            await client.cleanup()
     finally:
         app.dependency_overrides.clear()