feat(queue): add dead-letter retention controls

Author: haasonsaas

agent_pm/storage/redis.py

@@ -5,6 +5,7 @@
 import json
 import logging
 import uuid
+from datetime import datetime, timezone
 from typing import Any
 
 import redis.asyncio as redis
@@ -71,6 +72,8 @@ async def get_task_result(client: redis.Redis, task_id: str) -> dict[str, Any] |
 
 async def record_dead_letter(client: redis.Redis, payload: dict[str, Any]) -> None:
     task_id = payload.get("task_id", uuid.uuid4().hex)
+    if "recorded_at" not in payload:
+        payload["recorded_at"] = datetime.now(timezone.utc).isoformat()
     await client.hset(_dead_letter_key(), task_id, json.dumps(payload))
 
 
@@ -105,12 +108,29 @@ 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 purge_dead_letters(client: redis.Redis, *, older_than: datetime | None = None) -> int:
+    if older_than is None:
+        count = await client.hlen(_dead_letter_key())
+        if count == 0:
+            return 0
+        await client.delete(_dead_letter_key())
+        return int(count)
+
+    items = await client.hgetall(_dead_letter_key())
+    removed = 0
+    for task_id, value in items.items():
+        try:
+            data = json.loads(value)
+            recorded = data.get("recorded_at")
+            if not recorded:
+                continue
+            recorded_dt = datetime.fromisoformat(recorded)
+        except (json.JSONDecodeError, ValueError):
+            continue
+        if recorded_dt <= older_than:
+            await client.hdel(_dead_letter_key(), task_id)
+            removed += 1
+    return removed
 
 
 async def write_heartbeat(client: redis.Redis, worker_id: str, payload: dict[str, Any], ttl: int) -> None:

agent_pm/storage/tasks.py

@@ -8,7 +8,7 @@
 from collections import deque
 from collections.abc import Callable, Coroutine
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timedelta
 from enum import Enum
 from typing import Any
 
@@ -217,6 +217,9 @@ async def get_dead_letter(self, task_id: str) -> dict[str, Any] | None:
     async def purge_dead_letters(self) -> int:
         return 0
 
+    async def purge_dead_letters_older_than(self, age: timedelta) -> int:
+        return 0
+
 
 # Global task queue instance
 _task_queue: TaskQueue | None = None
@@ -366,8 +369,11 @@ async def requeue_dead_letter(self, task_id: str) -> dict[str, Any] | None:
                 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)
+                async def purge_dead_letters(self, *, older_than: timedelta | None = None) -> int:
+                    if older_than is None:
+                        return await purge_dead_letters(self._redis)
+                    cutoff = utc_now() - older_than
+                    return await purge_dead_letters(self._redis, older_than=cutoff)
 
             _task_queue = RedisTaskQueue(max_workers=settings.task_queue_workers)
         else:

app.py

@@ -5,7 +5,7 @@
 import asyncio
 import logging
 import uuid
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any
 
 from fastapi import Depends, FastAPI, HTTPException, WebSocket, WebSocketDisconnect
@@ -482,10 +482,15 @@ async def delete_dead_letter(task_id: str, _admin_key: AdminKeyDep = None) -> di
 
 
 @app.delete("/tasks/dead-letter")
-async def purge_dead_letters(_admin_key: AdminKeyDep = None) -> dict[str, int]:
+async def purge_dead_letters(
+    older_than_minutes: int | None = None, _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()
+    if older_than_minutes is None:
+        deleted = await _task_queue.purge_dead_letters()
+    else:
+        deleted = await _task_queue.purge_dead_letters_older_than(timedelta(minutes=older_than_minutes))
     return {"deleted": deleted}
 
 

tests/storage/test_redis.py

@@ -1,4 +1,5 @@
 import json
+from datetime import datetime, timedelta, timezone
 
 import pytest
 
@@ -103,11 +104,17 @@ 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)
+    old_payload = {"task_id": "old", "name": "job", "args": [], "kwargs": {}, "recorded_at": "2000-01-01T00:00:00+00:00"}
+    await redis.record_dead_letter(client, old_payload)
     removed = await redis.purge_dead_letters(client)
     assert removed == 1
     assert await redis.fetch_dead_letters(client) == []
 
+    # ensure age-based purging skips fresh entries
+    await redis.record_dead_letter(client, payload)
+    removed = await redis.purge_dead_letters(client, older_than=datetime.now(timezone.utc) - timedelta(minutes=1))
+    assert removed == 0
+
     await redis.write_heartbeat(client, "worker:1", {"status": "ok"}, ttl=60)
     beats = await redis.list_heartbeats(client)
     assert beats["worker:1"]["status"] == "ok"

tests/test_tasks_api.py

@@ -1,3 +1,4 @@
+from datetime import timedelta
 from typing import Any
 
 import pytest
@@ -86,6 +87,9 @@ async def get_dead_letter(self, task_id: str) -> dict[str, Any] | None:
     async def purge_dead_letters(self) -> int:
         return 1
 
+    async def purge_dead_letters_older_than(self, age):  # pragma: no cover - stubbed for API test
+        return 0
+
 
 @pytest.mark.asyncio
 async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
@@ -126,6 +130,9 @@ async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
                 purge_resp = await client.delete("/tasks/dead-letter")
                 assert purge_resp.status_code == 200
                 assert purge_resp.json()["deleted"] == 1
+
+                purge_resp_age = await client.delete("/tasks/dead-letter", params={"older_than_minutes": 10})
+                assert purge_resp_age.status_code == 200
             finally:
                 settings.task_queue_backend = original_backend
                 app_module._task_queue = original_queue