fix(ci): satisfy ruff linting requirements

Author: haasonsaas

agent_pm/clients/pagerduty_client.py

@@ -17,7 +17,9 @@ def __init__(self) -> None:
     def enabled(self) -> bool:
         return bool(self.routing_key)
 
-    async def trigger_incident(self, summary: str, source: str = "agent-pm", severity: str = "error", **details: Any) -> dict[str, Any]:
+    async def trigger_incident(
+        self, summary: str, source: str = "agent-pm", severity: str = "error", **details: Any
+    ) -> dict[str, Any]:
         if not self.enabled:
             return {"dry_run": True, "summary": summary, "details": details}
 

agent_pm/connectors/email.py

@@ -5,7 +5,7 @@
 import asyncio
 import json
 import logging
-from datetime import UTC, datetime
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 

agent_pm/observability/dashboard.py

@@ -4,9 +4,9 @@
 
 from dataclasses import dataclass
 
+from agent_pm.settings import settings
 from agent_pm.storage.redis import count_dead_letters
 from agent_pm.storage.tasks import get_task_queue
-from agent_pm.settings import settings
 
 
 @dataclass
@@ -21,4 +21,8 @@ async def gather_queue_health() -> QueueHealth:
     client = getattr(queue, "_redis", None)
     dead_letters = await count_dead_letters(client) if client else 0
     auto_triage_enabled = bool(settings.task_queue_auto_requeue_errors)
-    return QueueHealth(queue_name=getattr(queue, "queue_name", "unknown"), dead_letters=dead_letters, auto_triage_enabled=auto_triage_enabled)
+    return QueueHealth(
+        queue_name=getattr(queue, "queue_name", "unknown"),
+        dead_letters=dead_letters,
+        auto_triage_enabled=auto_triage_enabled,
+    )

agent_pm/observability/metrics.py

@@ -7,7 +7,6 @@
 
 from prometheus_client import Counter, Gauge, Histogram, Summary, generate_latest
 
-
 dead_letter_recorded_total = Counter(
     "task_dead_letter_recorded_total",
     "Dead-letter entries recorded",

agent_pm/settings.py

@@ -53,7 +53,9 @@ def _parse_csv_list(value: str | list[str] | None) -> list[str]:
             return [item.strip() for item in value.split(",") if item.strip()]
         return value
 
-    @field_validator("github_repositories", "slack_sync_channels", "gmail_label_filter", "notion_database_ids", mode="before")
+    @field_validator(
+        "github_repositories", "slack_sync_channels", "gmail_label_filter", "notion_database_ids", mode="before"
+    )
     @classmethod
     def _parse_string_lists(cls, value):
         return cls._parse_csv_list(value)

agent_pm/storage/syncs.py

@@ -45,11 +45,7 @@ async def record_sync(
 async def list_recent_syncs(limit: int = 50) -> list[dict[str, Any]]:
     session_factory = get_session_factory()
     async with session_factory() as session:
-        stmt = (
-            select(ConnectorSync)
-            .order_by(desc(ConnectorSync.started_at))
-            .limit(max(limit, 1))
-        )
+        stmt = select(ConnectorSync).order_by(desc(ConnectorSync.started_at)).limit(max(limit, 1))
         try:
             result = await session.execute(stmt)
         except Exception:

agent_pm/storage/tasks.py

@@ -13,6 +13,7 @@
 from enum import Enum
 from typing import Any
 
+from ..clients import slack_client
 from ..observability.metrics import (
     dead_letter_active_gauge,
     dead_letter_alert_total,
@@ -25,15 +26,12 @@
     record_task_latency,
 )
 from ..settings import settings
-from ..utils.datetime import utc_now
-from ..clients import pagerduty_client, slack_client
 from ..tasks.playbooks import run_playbook
+from ..utils.datetime import utc_now
 from .redis import (
-    append_dead_letter_audit,
     clear_dead_letter,
     count_dead_letters,
     delete_retry_policy,
-    enqueue_task as redis_enqueue_task,
     fetch_dead_letter_audit,
     fetch_dead_letters,
     get_dead_letter,
@@ -48,6 +46,9 @@
     set_task_result,
     write_heartbeat,
 )
+from .redis import (
+    enqueue_task as redis_enqueue_task,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -332,12 +333,22 @@ async def _worker(self, worker_id: int):
                         alert_channel = settings.task_queue_alert_channel or settings.slack_status_channel
                         cooldown = timedelta(minutes=settings.task_queue_alert_cooldown_minutes)
 
-                        def _should_auto_requeue(err_type: str | None) -> bool:
+                        def _should_auto_requeue(
+                            err_type: str | None,
+                            *,
+                            auto_errors: set[str] = auto_errors,
+                        ) -> bool:
                             if not err_type:
                                 return False
                             return err_type in auto_errors
 
-                        def _record_failure(err_type: str | None, task_identifier: str) -> bool:
+                        def _record_failure(
+                            err_type: str | None,
+                            task_identifier: str,
+                            *,
+                            alert_window: timedelta = alert_window,
+                            alert_threshold: int = alert_threshold,
+                        ) -> bool:
                             if not err_type:
                                 return False
                             now = utc_now()
@@ -348,13 +359,17 @@ def _record_failure(err_type: str | None, task_identifier: str) -> bool:
                             recent_failures[key] = [ts for ts in entries if ts >= cutoff]
                             return len(recent_failures[key]) >= alert_threshold
 
-                        async def _send_alert(error_type: str, payload: dict[str, Any]) -> None:
+                        async def _send_alert(
+                            error_type: str,
+                            payload: dict[str, Any],
+                            *,
+                            alert_channel: str | None = alert_channel,
+                            cooldown: timedelta = cooldown,
+                        ) -> None:
                             if not alert_channel:
                                 return
                             if settings.dry_run or not slack_client.enabled:
-                                logger.warning(
-                                    "Slack alert skipped (dry run): %s", error_type
-                                )
+                                logger.warning("Slack alert skipped (dry run): %s", error_type)
                                 return
                             now = utc_now()
                             last_sent = last_alert_sent.get(error_type)
@@ -373,19 +388,24 @@ 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:
+                        async def _apply_adaptive_policy(
+                            task_name: str,
+                            task_payload: dict[str, Any],
+                        ) -> 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("max_retries", task_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["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] = []
@@ -462,7 +482,9 @@ async def _apply_adaptive_policy(task_name: str) -> None:
                                 ).inc()
                                 record_task_completion(self.queue_name, TaskStatus.FAILED.value)
                                 record_task_latency(self.queue_name, (utc_now() - start).total_seconds())
-                                identifier = payload.get("metadata", {}).get("workflow_id") or payload.get("name", "unknown")
+                                identifier = payload.get("metadata", {}).get("workflow_id") or payload.get(
+                                    "name", "unknown"
+                                )
                                 playbook_name = settings.task_queue_playbooks.get(error_type)
                                 if playbook_name:
                                     await run_playbook(playbook_name, payload, self, error_type)
@@ -480,7 +502,7 @@ async def _apply_adaptive_policy(task_name: str) -> None:
                                             payload = auto_payload
                                 metrics = failure_metrics.setdefault(name, [])
                                 metrics.append(False)
-                                await _apply_adaptive_policy(name)
+                                await _apply_adaptive_policy(name, payload)
                                 if _record_failure(error_type, identifier):
                                     await _send_alert(error_type, payload)
                                 continue
@@ -492,13 +514,13 @@ async def _apply_adaptive_policy(task_name: str) -> None:
                             await asyncio.sleep(backoff)
                             metrics = failure_metrics.setdefault(name, [])
                             metrics.append(False)
-                            await _apply_adaptive_policy(name)
+                            await _apply_adaptive_policy(name, payload)
                             await redis_enqueue_task(self._redis, name, payload)
                             continue
 
                         metrics = failure_metrics.setdefault(name, [])
                         metrics.append(True)
-                        await _apply_adaptive_policy(name)
+                        await _apply_adaptive_policy(name, payload)
                         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())
@@ -588,7 +610,9 @@ async def purge_dead_letters(self, *, older_than: timedelta | None = None) -> in
                     if older_than is None:
                         deleted = await purge_dead_letters(self._redis)
                         dead_letter_purged_total.labels(queue=self.queue_name, mode="all").inc(deleted)
-                        dead_letter_active_gauge.labels(queue=self.queue_name).set(await count_dead_letters(self._redis))
+                        dead_letter_active_gauge.labels(queue=self.queue_name).set(
+                            await count_dead_letters(self._redis)
+                        )
                         return deleted
                     cutoff = utc_now() - older_than
                     deleted = await purge_dead_letters(self._redis, older_than=cutoff)

agent_pm/tasks/playbooks.py

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Awaitable, Callable
+from collections.abc import Awaitable, Callable
+from typing import Any
 
 import httpx
 
@@ -50,7 +51,8 @@ async def _invoke_webhook(payload: dict[str, Any], queue: Any, error_type: str)
 
 async def _log_only(payload: dict[str, Any], queue: Any, error_type: str) -> None:
     logger.warning(
-        "Remediation playbook invoked", extra={"queue": getattr(queue, "queue_name", "unknown"), "error": error_type, "task": payload.get("task_id")}
+        "Remediation playbook invoked",
+        extra={"queue": getattr(queue, "queue_name", "unknown"), "error": error_type, "task": payload.get("task_id")},
     )
 
 

agent_pm/tasks/sync.py

@@ -5,9 +5,9 @@
 import asyncio
 import logging
 import time
+from collections.abc import Iterable
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
-from typing import Iterable
 
 from agent_pm.connectors import (
     CalendarConnector,
@@ -136,7 +136,9 @@ async def _record_sync(
                 metadata={"started_at": started_at.isoformat()},
             )
         except Exception:  # pragma: no cover - defensive logging
-            logger.exception("Failed to persist connector sync record", extra={"connector": connector, "status": status})
+            logger.exception(
+                "Failed to persist connector sync record", extra={"connector": connector, "status": status}
+            )
 
 
 def create_default_sync_manager() -> PeriodicSyncManager:

app.py

@@ -5,11 +5,10 @@
 import asyncio
 import logging
 import uuid
+from contextlib import asynccontextmanager
 from datetime import datetime, timedelta
 from typing import Any
 
-from contextlib import asynccontextmanager
-
 from fastapi import Depends, FastAPI, HTTPException, WebSocket, WebSocketDisconnect
 from fastapi.responses import PlainTextResponse
 from pydantic import BaseModel, model_validator
@@ -73,6 +72,7 @@
 from agent_pm.tasks.sync import PeriodicSyncManager, create_default_sync_manager
 from agent_pm.tools import registry
 
+
 @asynccontextmanager
 async def lifespan(_app: FastAPI):
     global _task_queue, _sync_manager
@@ -543,9 +543,7 @@ async def delete_dead_letter(task_id: str, _admin_key: AdminKeyDep = None) -> di
 
 
 @app.delete("/tasks/dead-letter")
-async def purge_dead_letters(
-    older_than_minutes: int | None = None, _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]:
     task_queue = await get_task_queue()
     if older_than_minutes is None:
         deleted = await task_queue.purge_dead_letters()