fix: notify Jira after queue retries exhaust

src/forge/integrations/jira/client.py

@@ -697,10 +697,28 @@ async def get_comments(self, issue_key: str) -> list[JiraComment]:
             List of JiraComment objects.
         """
         client = await self._get_client()
-        response = await client.get(f"/issue/{issue_key}/comment")
-        response.raise_for_status()
-        data = response.json()
-        return [JiraComment.from_api_response(c) for c in data.get("comments", [])]
+        comments: list[JiraComment] = []
+        start_at = 0
+        max_results = 100
+
+        while True:
+            response = await client.get(
+                f"/issue/{issue_key}/comment",
+                params={"startAt": start_at, "maxResults": max_results},
+            )
+            response.raise_for_status()
+            data = response.json()
+            page = data.get("comments", [])
+            comments.extend(JiraComment.from_api_response(comment) for comment in page)
+
+            page_start = int(data.get("startAt", start_at))
+            total = int(data.get("total", page_start + len(page)))
+            next_start = page_start + len(page)
+            if not page or next_start >= total:
+                break
+            start_at = next_start
+
+        return comments
 
     async def get_labels(self, issue_key: str) -> list[str]:
         """Get labels for a Jira issue.

src/forge/orchestrator/worker.py

@@ -119,7 +119,10 @@ def __init__(
         """
         self.settings = get_settings()
         self.consumer_name = consumer_name or f"worker-{uuid.uuid4().hex[:8]}"
-        self.consumer = QueueConsumer(self.consumer_name)
+        self.consumer = QueueConsumer(
+            self.consumer_name,
+            terminal_failure_handler=self._handle_terminal_failure,
+        )
         self.router = router or create_default_router()
         self._shutdown_event = asyncio.Event()
         self._checkpointer = None
@@ -140,6 +143,37 @@ async def _get_forge_github_login(self) -> str:
             self._forge_github_login = login
         return login
 
+    async def _handle_terminal_failure(self, message: QueueMessage, error: str) -> None:
+        """Post one Jira comment after queue retries are exhausted."""
+        jira = JiraClient()
+        event_marker = f"Event/correlation ID: {message.event_id}"
+        try:
+            comments = await jira.get_comments(message.ticket_key)
+            if any(event_marker in comment.body for comment in comments):
+                logger.info(
+                    f"Terminal failure notification already exists for event {message.event_id}"
+                )
+                return
+
+            safe_error = redact_secrets(error)
+            if len(safe_error) > 500:
+                safe_error = f"{safe_error[:500]}..."
+            details = (
+                f"{safe_error}\n\n"
+                f"Ticket: {message.ticket_key}\n"
+                f"{event_marker}\n"
+                "Recovery: inspect the dead-letter entry, resolve the root cause, "
+                "then requeue the event."
+            )
+            await jira.add_error_comment(
+                issue_key=message.ticket_key,
+                error_message=details,
+                node_name="queue execution (retries exhausted)",
+            )
+            logger.info(f"Posted terminal queue failure notification to {message.ticket_key}")
+        finally:
+            await jira.close()
+
     async def _handle_jira_event(self, message: QueueMessage) -> None:
         """Handle a Jira webhook event.
 

src/forge/queue/consumer.py

@@ -3,7 +3,7 @@
 import asyncio
 import logging
 from collections import defaultdict
-from collections.abc import Callable, Coroutine
+from collections.abc import AsyncIterator, Callable, Coroutine
 from contextlib import asynccontextmanager, suppress
 from typing import Any
 
@@ -16,7 +16,7 @@
 from forge.orchestrator.checkpointer import get_redis_client
 from forge.queue.models import QueueMessage
 from forge.queue.producer import GITHUB_STREAM, JIRA_STREAM
-from forge.queue.retry import RetryQueue
+from forge.queue.retry import RETRY_CLAIM_RENEW_SECONDS, RetryEntry, RetryQueue
 
 logger = logging.getLogger(__name__)
 
@@ -35,12 +35,17 @@
 
 # Handler type for message processing
 MessageHandler = Callable[[QueueMessage], Coroutine[Any, Any, None]]
+TerminalFailureHandler = Callable[[QueueMessage, str], Coroutine[Any, Any, None]]
 
 
 class TicketLockLostError(RuntimeError):
     """Raised when a worker loses its distributed per-ticket lease."""
 
 
+class RetryLeaseLostError(RuntimeError):
+    """Raised when a worker loses ownership of a claimed retry entry."""
+
+
 class QueueConsumer:
     """Consumes webhook events from Redis Streams with FIFO ordering per ticket.
 
@@ -54,6 +59,7 @@ def __init__(
         redis_client: redis.Redis | None = None,
         jira_client: JiraClient | None = None,
         max_concurrent_tasks: int | None = None,
+        terminal_failure_handler: TerminalFailureHandler | None = None,
     ):
         """Initialize the queue consumer.
 
@@ -63,6 +69,8 @@ def __init__(
             jira_client: Optional Jira client for freshness checks.
             max_concurrent_tasks: Maximum concurrent in-flight tasks. Defaults
                 to ``settings.queue_max_concurrent_tasks`` when not provided.
+            terminal_failure_handler: Optional callback invoked after a message
+                is moved to the dead-letter queue.
         """
         self.consumer_name = consumer_name
         self._redis = redis_client
@@ -78,9 +86,10 @@ def __init__(
         self._semaphore = asyncio.Semaphore(concurrency)
         self._active_tasks: set[asyncio.Task[None]] = set()
         self._retry_queue = RetryQueue()
+        self._terminal_failure_handler = terminal_failure_handler
 
     @asynccontextmanager
-    async def _distributed_ticket_lock(self, ticket_key: str):
+    async def _distributed_ticket_lock(self, ticket_key: str) -> AsyncIterator[None]:
         """Hold a renewable, deployment-wide lease for one ticket."""
         redis_client = await self._get_redis()
         lock = redis_client.lock(
@@ -199,6 +208,16 @@ async def _ack(self, stream: str, message_id: str) -> None:
         redis_client = await self._get_redis()
         await redis_client.xack(stream, CONSUMER_GROUP, message_id)
 
+    async def _ack_terminal_message(self, message: QueueMessage, stream: str) -> None:
+        """Best-effort acknowledge a message after it has reached the DLQ."""
+        try:
+            await self._ack(stream, message.message_id)
+        except Exception as ack_error:
+            logger.warning(
+                f"xack failed for terminal event {message.event_id}; "
+                f"PEL entry may linger: {ack_error}"
+            )
+
     async def _process_message(
         self,
         message: QueueMessage,
@@ -251,7 +270,7 @@ async def _process_message(
             try:
                 moved_to_dlq = not await self._retry_queue.enqueue_for_retry(message, str(exc))
                 if moved_to_dlq:
-                    await self._ack(stream, message.message_id)
+                    await self._ack_terminal_message(message, stream)
             except Exception as retry_err:
                 logger.error(
                     f"Failed to enqueue {message.event_id} for retry after ticket lock loss: "
@@ -290,7 +309,7 @@ async def _handle_locked_message(
             try:
                 moved_to_dlq = not await self._retry_queue.enqueue_for_retry(message, str(e))
                 if moved_to_dlq:
-                    await self._ack(stream, message.message_id)
+                    await self._ack_terminal_message(message, stream)
             except Exception as retry_err:
                 logger.error(
                     f"Failed to enqueue {message.event_id} for retry: {retry_err}. "
@@ -337,21 +356,70 @@ async def _consume_stream(self, stream: str, _source: EventSource) -> None:
         if self._active_tasks:
             await asyncio.gather(*self._active_tasks, return_exceptions=True)
 
+    @asynccontextmanager
+    async def _renew_retry_claim(self, entry: RetryEntry) -> AsyncIterator[None]:
+        """Renew a retry lease and abort processing if ownership is lost."""
+        if entry.lease_until is None:
+            yield
+            return
+
+        owner_task = asyncio.current_task()
+        lease_lost = asyncio.Event()
+
+        async def renew() -> None:
+            while True:
+                await asyncio.sleep(RETRY_CLAIM_RENEW_SECONDS)
+                try:
+                    if not await self._retry_queue.renew_retry_claim(entry):
+                        lease_lost.set()
+                        if owner_task is not None:
+                            owner_task.cancel()
+                        return
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    logger.exception("Failed to renew retry lease for %s", entry.message.event_id)
+                    lease_lost.set()
+                    if owner_task is not None:
+                        owner_task.cancel()
+                    return
+
+        renewal_task = asyncio.create_task(renew(), name=f"renew-retry-{entry.message.event_id}")
+        try:
+            yield
+        except asyncio.CancelledError:
+            if lease_lost.is_set():
+                raise RetryLeaseLostError(
+                    f"Lost retry lease for {entry.message.event_id}"
+                ) from None
+            raise
+        finally:
+            renewal_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await renewal_task
+
     async def _process_due_retries_once(self) -> None:
         """Dispatch one batch of due retries.
 
         Kept separate from the polling loop so recovery and terminal DLQ
         behavior can be exercised deterministically in integration tests.
         """
-        entries = await self._retry_queue.get_due_messages()
+        entries = await self._retry_queue.claim_due_messages()
         for entry in entries:
             retry_stream = (
                 JIRA_STREAM if entry.message.source == EventSource.JIRA else GITHUB_STREAM
             )
             try:
-                await self._process_message(
-                    entry.message, retry_stream, raise_on_error=True, skip_ack=True
+                async with self._renew_retry_claim(entry):
+                    await self._process_message(
+                        entry.message, retry_stream, raise_on_error=True, skip_ack=True
+                    )
+            except RetryLeaseLostError:
+                logger.error(
+                    "Stopped retry processing after losing ownership of %s",
+                    entry.message.event_id,
                 )
+                continue
             except Exception as e:
                 logger.warning(
                     f"Retry attempt {entry.attempt} failed for "
@@ -364,7 +432,7 @@ async def _process_due_retries_once(self) -> None:
                 if not queued:
                     # Terminal failure: the DLQ now owns the message, so clear
                     # its original stream PEL entry.
-                    await self._ack(retry_stream, entry.message.message_id)
+                    await self._ack_terminal_message(entry.message, retry_stream)
                 continue
 
             # Success: clear retry state and acknowledge the original entry.
@@ -381,11 +449,30 @@ async def _process_due_retries_once(self) -> None:
                     f"retry (PEL entry may linger): {xack_err}"
                 )
 
+    async def _process_terminal_notifications_once(self) -> None:
+        """Deliver one leased batch from the terminal-notification outbox."""
+        if self._terminal_failure_handler is None:
+            return
+
+        entries = await self._retry_queue.claim_due_terminal_notifications()
+        for entry in entries:
+            try:
+                await self._terminal_failure_handler(entry.message, entry.error)
+            except Exception as exc:
+                logger.error(
+                    "Terminal failure callback failed for %s: %s",
+                    entry.message.event_id,
+                    exc,
+                )
+                continue
+            await self._retry_queue.remove_terminal_notification(entry)
+
     async def _process_retry_queue(self) -> None:
         """Poll the retry queue and re-dispatch due messages."""
         while self._running:
             try:
                 await self._process_due_retries_once()
+                await self._process_terminal_notifications_once()
             except asyncio.CancelledError:
                 break
             except Exception as e: