feat(queue): add adaptive retry tuning

Author: haasonsaas

agent_pm/settings.py

@@ -86,6 +86,8 @@ def _parse_google_scopes(cls, value):
     pagerduty_routing_key: str | None = Field(None, alias="PAGERDUTY_ROUTING_KEY")
     task_queue_playbooks: dict[str, str] = Field(default_factory=dict, alias="TASK_QUEUE_PLAYBOOKS")
     pagerduty_service_name: str | None = Field(None, alias="PAGERDUTY_SERVICE_NAME")
+    task_queue_adaptive_failure_threshold: float = Field(0.6, alias="TASK_QUEUE_ADAPTIVE_FAILURE_THRESHOLD")
+    task_queue_adaptive_min_samples: int = Field(10, alias="TASK_QUEUE_ADAPTIVE_MIN_SAMPLES")
     database_url: str | None = Field("sqlite+aiosqlite:///./data/agent_pm.db", alias="DATABASE_URL")
     database_echo: bool = Field(False, alias="DATABASE_ECHO")
     redis_url: str = Field("redis://localhost:6379", alias="REDIS_URL")

agent_pm/storage/tasks.py

@@ -323,6 +323,7 @@ async def _worker(self, worker_id: int):
                     recent_failures: dict[str, list[datetime]] = {}
                     auto_requeue_counts: dict[str, int] = {}
                     last_alert_sent: dict[str, datetime] = {}
+                    failure_metrics: dict[str, list[bool]] = {}
 
                     while self.running:
                         auto_errors = set(settings.task_queue_auto_requeue_errors)
@@ -372,6 +373,23 @@ async def _send_alert(error_type: str, payload: dict[str, Any]) -> None:
                             except Exception as exc:  # pragma: no cover - logging
                                 logger.error("Failed to send Slack alert: %s", exc)
 
+                        async def _apply_adaptive_policy(task_name: str) -> None:
+                            samples = failure_metrics.get(task_name, [])
+                            if len(samples) < settings.task_queue_adaptive_min_samples:
+                                return
+                            failure_rate = 1 - (sum(1 for success in samples if success) / len(samples))
+                            if failure_rate < settings.task_queue_adaptive_failure_threshold:
+                                return
+                            policy = await get_retry_policy(self._redis, task_name) or {}
+                            policy.setdefault("max_retries", payload.get("max_retries", 3))
+                            policy.setdefault("timeout", self._task_timeout)
+                            policy.setdefault("backoff_base", self._backoff_base)
+                            policy.setdefault("backoff_max", self._backoff_max)
+                            policy["max_retries"] = min(int(policy["max_retries"]) + 1, settings.task_queue_max_auto_requeues)
+                            policy["timeout"] = float(policy.get("timeout", self._task_timeout)) + 5.0
+                            await set_retry_policy(self._redis, task_name, policy)
+                            failure_metrics[task_name] = []
+
                         payload = await self.pop()
                         if not payload:
                             await asyncio.sleep(self._poll_interval)
@@ -460,6 +478,9 @@ async def _send_alert(error_type: str, payload: dict[str, Any]) -> None:
                                         auto_requeue_counts[key] = count + 1
                                         if auto_payload:
                                             payload = auto_payload
+                                metrics = failure_metrics.setdefault(name, [])
+                                metrics.append(False)
+                                await _apply_adaptive_policy(name)
                                 if _record_failure(error_type, identifier):
                                     await _send_alert(error_type, payload)
                                 continue
@@ -469,9 +490,15 @@ async def _send_alert(error_type: str, payload: dict[str, Any]) -> None:
                                 float(policy.get("backoff_max", self._backoff_max)),
                             )
                             await asyncio.sleep(backoff)
+                            metrics = failure_metrics.setdefault(name, [])
+                            metrics.append(False)
+                            await _apply_adaptive_policy(name)
                             await redis_enqueue_task(self._redis, name, payload)
                             continue
 
+                        metrics = failure_metrics.setdefault(name, [])
+                        metrics.append(True)
+                        await _apply_adaptive_policy(name)
                         await set_task_result(self._redis, task_id, {"status": "completed", "result": result})
                         record_task_completion(self.queue_name, TaskStatus.COMPLETED.value)
                         record_task_latency(self.queue_name, (utc_now() - start).total_seconds())

tests/storage/test_redis_worker.py

@@ -1,4 +1,5 @@
 import asyncio
+from typing import Any
 from collections import deque
 
 import pytest
@@ -204,3 +205,46 @@ async def flaky_task() -> str:
     assert auto_metric == 1
     assert alert_metric == 1
     assert calls
+
+
+@pytest.mark.asyncio
+async def test_adaptive_retry_policy_updates(redis_queue, monkeypatch):
+    queue, fake = redis_queue
+
+    monkeypatch.setattr(tasks_module.settings, "task_queue_adaptive_min_samples", 2)
+    monkeypatch.setattr(tasks_module.settings, "task_queue_adaptive_failure_threshold", 0.5)
+    monkeypatch.setattr(tasks_module.settings, "task_queue_max_auto_requeues", 10)
+
+    policies: dict[str, dict[str, Any]] = {}
+
+    async def fake_get_policy(client, task_name: str):
+        return policies.get(task_name)
+
+    async def fake_set_policy(client, task_name: str, policy: dict[str, Any]):
+        policies[task_name] = policy
+
+    monkeypatch.setattr(tasks_module, "get_retry_policy", fake_get_policy)
+    monkeypatch.setattr(tasks_module, "set_retry_policy", fake_set_policy)
+
+    attempts = {"count": 0}
+
+    async def unreliable_task() -> str:
+        attempts["count"] += 1
+        if attempts["count"] <= 2:
+            raise RuntimeError("flaky")
+        return "ok"
+
+    task_id = await queue.enqueue("unstable", unreliable_task, max_retries=5)
+
+    result = None
+    for _ in range(100):
+        result = await redis_helpers.get_task_result(fake, task_id)
+        if result:
+            break
+        await asyncio.sleep(0.05)
+
+    assert result is not None
+    assert result["status"] == "completed"
+    assert "unstable" in policies
+    assert policies["unstable"]["max_retries"] == 6
+    assert policies["unstable"]["timeout"] == tasks_module.settings.task_queue_task_timeout + 5.0