feat(queue): expose dead-letter detail and purge

Author: haasonsaas

agent_pm/storage/redis.py

@@ -105,6 +105,14 @@ async def clear_dead_letter(client: redis.Redis, task_id: str) -> None:
     await client.hdel(_dead_letter_key(), task_id)
 
 
+async def purge_dead_letters(client: redis.Redis) -> int:
+    count = await client.hlen(_dead_letter_key())
+    if count == 0:
+        return 0
+    await client.delete(_dead_letter_key())
+    return int(count)
+
+
 async def write_heartbeat(client: redis.Redis, worker_id: str, payload: dict[str, Any], ttl: int) -> None:
     await client.hset(_heartbeat_key(), worker_id, json.dumps(payload))
     await client.expire(_heartbeat_key(), ttl)

agent_pm/storage/tasks.py

@@ -27,6 +27,7 @@
     get_redis_client,
     list_heartbeats,
     pop_task,
+    purge_dead_letters,
     record_dead_letter,
     set_task_result,
     write_heartbeat,
@@ -210,6 +211,12 @@ async def worker_heartbeats(self) -> dict[str, Any]:
     async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
         return None
 
+    async def get_dead_letter(self, task_id: str) -> dict[str, Any] | None:
+        return None
+
+    async def purge_dead_letters(self) -> int:
+        return 0
+
 
 # Global task queue instance
 _task_queue: TaskQueue | None = None
@@ -356,6 +363,12 @@ async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
                     logger.info("Dead-letter task requeued: %s", task_id)
                     return payload
 
+                async def get_dead_letter(self, task_id: str) -> dict[str, Any] | None:
+                    return await get_dead_letter(self._redis, task_id)
+
+                async def purge_dead_letters(self) -> int:
+                    return await purge_dead_letters(self._redis)
+
             _task_queue = RedisTaskQueue(max_workers=settings.task_queue_workers)
         else:
             _task_queue = TaskQueue(max_workers=settings.task_queue_workers)

app.py

@@ -463,6 +463,16 @@ async def list_dead_letter(limit: int = 50, _admin_key: AdminKeyDep = None) -> d
     return {"dead_letter": items, "total": len(items)}
 
 
+@app.get("/tasks/dead-letter/{task_id}")
+async def get_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")
+    item = await _task_queue.get_dead_letter(task_id)
+    if not item:
+        raise HTTPException(status_code=404, detail="Dead-letter task not found")
+    return item
+
+
 @app.delete("/tasks/dead-letter/{task_id}")
 async def delete_dead_letter(task_id: str, _admin_key: AdminKeyDep = None) -> dict[str, Any]:
     if not _task_queue:
@@ -471,6 +481,14 @@ async def delete_dead_letter(task_id: str, _admin_key: AdminKeyDep = None) -> di
     return {"task_id": task_id, "status": "deleted"}
 
 
+@app.delete("/tasks/dead-letter")
+async def purge_dead_letters(_admin_key: AdminKeyDep = None) -> dict[str, int]:
+    if not _task_queue:
+        raise HTTPException(status_code=503, detail="Task queue not initialized")
+    deleted = await _task_queue.purge_dead_letters()
+    return {"deleted": deleted}
+
+
 @app.get("/tasks/workers")
 async def worker_status(_admin_key: AdminKeyDep = None) -> dict[str, Any]:
     if not _task_queue:

tests/storage/test_redis.py

@@ -47,9 +47,18 @@ async def hdel(self, key: str, field: str):
         if key.endswith("dead_letter"):
             self.dead.pop(field, None)
 
+    async def hlen(self, key: str) -> int:
+        if key.endswith("dead_letter"):
+            return len(self.dead)
+        return len(self.results)
+
     async def expire(self, key: str, ttl: int):
         return None
 
+    async def delete(self, key: str):
+        if key.endswith("dead_letter"):
+            self.dead.clear()
+
 
 @pytest.mark.asyncio
 async def test_enqueue_task_pushes_payload():
@@ -94,6 +103,11 @@ async def test_dead_letter_and_heartbeat_helpers():
     await redis.clear_dead_letter(client, "abc")
     assert await redis.fetch_dead_letters(client) == []
 
+    await redis.record_dead_letter(client, payload)
+    removed = await redis.purge_dead_letters(client)
+    assert removed == 1
+    assert await redis.fetch_dead_letters(client) == []
+
     await redis.write_heartbeat(client, "worker:1", {"status": "ok"}, ttl=60)
     beats = await redis.list_heartbeats(client)
     assert beats["worker:1"]["status"] == "ok"

tests/storage/test_redis_worker.py

@@ -37,6 +37,9 @@ async def hgetall(self, key: str):
     async def hdel(self, key: str, field: str):
         self.hashes.setdefault(key, {}).pop(field, None)
 
+    async def hlen(self, key: str) -> int:
+        return len(self.hashes.get(key, {}))
+
     async def expire(self, key: str, ttl: int):
         return None
 

tests/test_tasks_api.py

@@ -73,6 +73,19 @@ async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
         self.requeued = task_id
         return {"task_id": task_id}
 
+    async def get_dead_letter(self, task_id: str) -> dict[str, Any] | None:
+        if task_id == "dead-1":
+            return {
+                "task_id": "dead-1",
+                "name": "explode",
+                "retry_count": 3,
+                "last_error": "boom",
+            }
+        return None
+
+    async def purge_dead_letters(self) -> int:
+        return 1
+
 
 @pytest.mark.asyncio
 async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
@@ -97,6 +110,10 @@ async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
                 assert del_resp.status_code == 200
                 assert stub.deleted == "dead-1"
 
+                detail_resp = await client.get("/tasks/dead-letter/dead-1")
+                assert detail_resp.status_code == 200
+                assert detail_resp.json()["task_id"] == "dead-1"
+
                 worker_resp = await client.get("/tasks/workers")
                 assert worker_resp.status_code == 200
                 assert worker_resp.json() == {"workers": {"worker:1": {"status": "ok"}}}
@@ -105,6 +122,10 @@ async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
                 assert requeue_resp.status_code == 200
                 assert stub.requeued == "dead-1"
                 assert requeue_resp.json()["status"] == "requeued"
+
+                purge_resp = await client.delete("/tasks/dead-letter")
+                assert purge_resp.status_code == 200
+                assert purge_resp.json()["deleted"] == 1
             finally:
                 settings.task_queue_backend = original_backend
                 app_module._task_queue = original_queue