feat: implement NotificationReaderAgent to pre-screen Gitea notifications (#2)

core/factory.py

@@ -11,6 +11,7 @@ from core.coding_agent import CodingAgent
 from core.agent import CavemanAgent
 from core.coordinator_agent import CoordinatorAgent
 from core.planning_agent import PlanningAgent
+from core.notification_agent import NotificationReaderAgent
 from gitea.workspace import WorkspaceManager
 
 logger: logging.Logger = logging.getLogger("core-factory")
@@ -77,6 +78,12 @@ class AgentFactory:
         logger.info(f"Factory creating PlanningAgent with model: {model_name}")
         return PlanningAgent(model_name)
 
+    @staticmethod
+    def create_notification_reader_agent(model_name: str) -> NotificationReaderAgent:
+        logger.info(f"Factory creating NotificationReaderAgent with model: {model_name}")
+        return NotificationReaderAgent(model_name)
+
+
 
 class WorkspaceFactory:
     """Factory for creating workspace manager instances."""

core/notification_agent.py

@@ -0,0 +1,45 @@
+import logging
+from typing import Any, Callable
+from core.agent import BaseAgent
+from core.prompts import NOTIFICATION_READER_SYSTEM_PROMPT
+from core.notification_tools import NotificationTools
+
+logger: logging.Logger = logging.getLogger("agent-notification-reader")
+
+
+class NotificationNoToolCalledError(Exception):
+    """Raised when the Notification Reader Agent completes execution without calling any routing tool."""
+    pass
+
+
+class NotificationReaderAgent(BaseAgent):
+    """AI agent that reviews Gitea notifications and decides how to route them."""
+
+    def __init__(self, model_name: str) -> None:
+        super().__init__(model_name)
+        self.system_prompt = NOTIFICATION_READER_SYSTEM_PROMPT
+
+    async def decide_notification(
+        self,
+        mission: str,
+        inspection_tools: list[Callable[..., Any]],
+        notification_tools: NotificationTools,
+    ) -> str:
+        """Run the Notification Reader Agent and ensure a decision tool is called."""
+        decision_tools: list[Callable[..., Any]] = [
+            notification_tools.process_issue,
+            notification_tools.process_pr,
+            notification_tools.skip_notification,
+        ]
+        combined_tools: list[Callable[..., Any]] = inspection_tools + decision_tools
+
+        logger.info("Running NotificationReaderAgent to decide action...")
+        response_text: str = await self.run_with_tools(mission, combined_tools)
+
+        if not notification_tools.tool_called:
+            logger.warning("NotificationReaderAgent did not call any tools!")
+            raise NotificationNoToolCalledError(
+                "NotificationReaderAgent failed to call a routing tool during execution."
+            )
+
+        return response_text

core/notification_tools.py

@@ -0,0 +1,68 @@
+import logging
+from typing import Any
+
+logger: logging.Logger = logging.getLogger("notification-tools")
+
+
+class NotificationTools:
+    """Tools exposed to the Notification Reader Agent for routing decisions."""
+
+    def __init__(self) -> None:
+        self.tool_called: bool = False
+        self.action: str = "NO_ACTION"
+        self.arguments: dict[str, Any] = {}
+
+    def process_issue(self, owner: str, repo: str, issue_number: int, reason: str) -> str:
+        """Process the notification as an issue.
+        Use this when a notification refers to an issue that requires agent action.
+
+        Args:
+            owner: The repository owner (organization).
+            repo: The repository name.
+            issue_number: The Gitea issue number.
+            reason: Why this notification should be processed.
+        """
+        logger.info(f"Notification tool 'process_issue' called for {owner}/{repo}#{issue_number}: {reason}")
+        self.tool_called = True
+        self.action = "PROCESS_ISSUE"
+        self.arguments = {
+            "owner": owner,
+            "repo": repo,
+            "issue_number": issue_number,
+            "reason": reason,
+        }
+        return "Issue notification marked for processing."
+
+    def process_pr(self, owner: str, repo: str, pr_number: int, reason: str) -> str:
+        """Process the notification as a pull request.
+        Use this when a notification refers to a PR that requires agent action.
+
+        Args:
+            owner: The repository owner (organization).
+            repo: The repository name.
+            pr_number: The Gitea PR number.
+            reason: Why this notification should be processed.
+        """
+        logger.info(f"Notification tool 'process_pr' called for {owner}/{repo}#{pr_number}: {reason}")
+        self.tool_called = True
+        self.action = "PROCESS_PR"
+        self.arguments = {
+            "owner": owner,
+            "repo": repo,
+            "pr_number": pr_number,
+            "reason": reason,
+        }
+        return "PR notification marked for processing."
+
+    def skip_notification(self, reason: str) -> str:
+        """Skip the notification and take no action.
+        Use this if the notification is irrelevant, already handled, or does not require agent intervention.
+
+        Args:
+            reason: The reason for skipping this notification.
+        """
+        logger.info(f"Notification tool 'skip_notification' called: {reason}")
+        self.tool_called = True
+        self.action = "SKIP"
+        self.arguments = {"reason": reason}
+        return "Notification marked to be skipped."

core/orchestrator.py

@@ -14,6 +14,9 @@ from gitea.models import IssueModel, PullRequestModel, RepositoryModel
 from gitea.tools.gitea_tools import GiteaTools
 from gitea.config import AGENT_MODEL_ID, AGENT_MAX_RETRIES
 from gitea.workspace import WorkspaceManager
+from core.factory import AgentFactory
+from core.notification_tools import NotificationTools
+from core.notification_agent import NotificationNoToolCalledError
 
 logger: logging.Logger = logging.getLogger("agent-orchestrator")
 
@@ -33,6 +36,9 @@ class AgentOrchestrator:
         self._model_name = model_name
         self._work_queue = WorkQueue()
         self._dispatcher = AgentDispatcher(client, tools, model_name, max_retries)
+        self._notification_reader = AgentFactory.create_notification_reader_agent(model_name)
+        self._max_retries = max_retries
+
 
     def _get_state_file_path(self) -> Path:
         """Get the path to the persistent state file."""
@@ -74,6 +80,13 @@ class AgentOrchestrator:
 
         # Filter and enqueue tasks from notifications
         latest_timestamp = last_checked
+        inspection_tools = [
+            self._tools.get_issue,
+            self._tools.get_pull_request,
+            self._tools.get_issue_comments,
+            self._tools.get_pull_request_comments,
+        ]
+
         for n in notifications:
             notification_id = n.get("id")
             subject = n.get("subject") or {}
@@ -97,7 +110,41 @@ class AgentOrchestrator:
                 logger.warning(f"Could not parse task number from subject URL: {subj_url}")
                 continue
 
-            if subj_type == "issue":
+            # Run NotificationReaderAgent to pre-screen the notification
+            mission = (
+                f"Analyze Gitea notification ID {notification_id}.\n"
+                f"Repository: {repo_full_name}\n"
+                f"Subject Type: {subj_type}\n"
+                f"Task Number: {task_number}\n"
+                f"Please decide if we should process or skip this notification."
+            )
+            notification_tools = NotificationTools()
+            attempt = 0
+            attempt_limit = self._max_retries
+            success = False
+            while attempt < attempt_limit:
+                attempt += 1
+                try:
+                    await self._notification_reader.decide_notification(
+                        mission,
+                        inspection_tools,
+                        notification_tools
+                    )
+                    success = True
+                    break
+                except NotificationNoToolCalledError as e:
+                    logger.error(f"Notification reader error on notification {notification_id} (attempt {attempt}/{attempt_limit}): {e}")
+
+            if not success or notification_tools.action == "SKIP":
+                reason = notification_tools.arguments.get("reason", "Failed to call routing tool / default skip")
+                logger.info(f"Skipping notification {notification_id} for {repo_full_name}#{task_number}. Reason: {reason}")
+                if notification_id is not None:
+                    self._client.mark_notification_as_read(notification_id)
+                    logger.info(f"Marked skipped Gitea notification thread {notification_id} as read.")
+                continue
+
+            # Route based on decided action
+            if notification_tools.action == "PROCESS_ISSUE":
                 try:
                     issue = self._client.get_issue(owner, repo_name, task_number)
                     if issue.repository is None:
@@ -114,7 +161,7 @@ class AgentOrchestrator:
                     self._work_queue.enqueue(item)
                 except Exception as e:
                     logger.error(f"Failed to fetch issue #{task_number} for notification: {e}")
-            elif subj_type in ("pull", "pullrequest"):
+            elif notification_tools.action == "PROCESS_PR":
                 try:
                     pr = self._client.get_pull_request(owner, repo_name, task_number)
                     if pr.repository is None:
@@ -132,6 +179,7 @@ class AgentOrchestrator:
                 except Exception as e:
                     logger.error(f"Failed to fetch PR #{task_number} for notification: {e}")
 
+
         # Process enqueued work
         if not self._work_queue.is_empty:
             await self._process_work()

core/prompts.py

@@ -29,3 +29,22 @@ CRITICAL INSTRUCTIONS:
 - You must call EXACTLY one tool. Do not guess, and do not output raw text instead of calling a tool.
 - The `plan`, `answer`, or `comment` argument you pass to the tool will be posted directly to Gitea. DO NOT include your thought process, reasoning, or internal details in those arguments. Keep them concise and professional.
 """
+
+NOTIFICATION_READER_SYSTEM_PROMPT: str = """
+You are a Gitea Notification Reader Agent. Your job is to analyze incoming Gitea notifications and determine how they should be routed.
+
+Based on the notification subject, details, and conversation comments (if retrieved), you must choose and call exactly one of the following tools:
+
+1. `process_issue`: Choose this if the notification refers to a Gitea issue that requires active intervention, planning, implementation, or answering a question by the AI agent.
+2. `process_pr`: Choose this if the notification refers to a Gitea Pull Request that requires active intervention, code reviews, updates, or merging by the AI agent.
+3. `skip_notification`: Choose this if:
+   - The notification is irrelevant or does not require AI agent intervention.
+   - It is a notification about an action taken by the AI agent itself (e.g. self-assigned, self-commented, self-opened).
+   - The discussion is closed or resolved, or the notification is just informational (e.g. a simple status update that needs no reply).
+   - You are unsure or think it does not fit the agent's scope. You must provide a clear reason for skipping.
+
+CRITICAL INSTRUCTIONS:
+- You must call EXACTLY one tool. Do not guess, and do not output raw text instead of calling a tool.
+- You can use the provided inspection tools (like get_issue, get_pull_request, get_issue_comments, get_pull_request_comments) to gather more details if the basic notification metadata is insufficient to make a decision.
+"""
+

tests/test_notification_agent.py

@@ -0,0 +1,83 @@
+import pytest
+from unittest.mock import MagicMock, AsyncMock, patch
+from core.notification_tools import NotificationTools
+from core.notification_agent import NotificationReaderAgent, NotificationNoToolCalledError
+
+pytestmark = pytest.mark.anyio
+
+
+def test_notification_tools() -> None:
+    tools = NotificationTools()
+    assert tools.tool_called is False
+    assert tools.action == "NO_ACTION"
+
+    res = tools.process_issue("meeks", "repo", 42, "fix bug")
+    assert tools.tool_called is True
+    assert tools.action == "PROCESS_ISSUE"
+    assert tools.arguments == {
+        "owner": "meeks",
+        "repo": "repo",
+        "issue_number": 42,
+        "reason": "fix bug",
+    }
+    assert "marked for processing" in res
+
+    tools = NotificationTools()
+    res = tools.process_pr("meeks", "repo", 10, "review change")
+    assert tools.tool_called is True
+    assert tools.action == "PROCESS_PR"
+    assert tools.arguments == {
+        "owner": "meeks",
+        "repo": "repo",
+        "pr_number": 10,
+        "reason": "review change",
+    }
+    assert "marked for processing" in res
+
+    tools = NotificationTools()
+    res = tools.skip_notification("unrelated comments")
+    assert tools.tool_called is True
+    assert tools.action == "SKIP"
+    assert tools.arguments == {"reason": "unrelated comments"}
+    assert "marked to be skipped" in res
+
+
+@patch("core.notification_agent.NotificationReaderAgent.initialize")
+@patch("core.notification_agent.NotificationReaderAgent.run_with_tools")
+async def test_notification_reader_agent_success(
+    mock_run_with_tools: MagicMock,
+    mock_initialize: MagicMock
+) -> None:
+    mock_initialize.return_value = None
+    agent = NotificationReaderAgent("dummy-model")
+
+    # Mock tool call inside run_with_tools
+    async def mock_run(mission: str, tools: list) -> str:
+        # Simulate calling a tool
+        for t in tools:
+            if getattr(t, "__name__", "") == "process_issue":
+                t("meeks", "repo", 42, "reason")
+        return "response"
+
+    mock_run_with_tools.side_effect = mock_run
+
+    tools = NotificationTools()
+    res = await agent.decide_notification("mission", [], tools)
+    assert res == "response"
+    assert tools.tool_called is True
+    assert tools.action == "PROCESS_ISSUE"
+
+
+@patch("core.notification_agent.NotificationReaderAgent.initialize")
+@patch("core.notification_agent.NotificationReaderAgent.run_with_tools")
+async def test_notification_reader_agent_no_tool_error(
+    mock_run_with_tools: MagicMock,
+    mock_initialize: MagicMock
+) -> None:
+    mock_initialize.return_value = None
+    agent = NotificationReaderAgent("dummy-model")
+    mock_run_with_tools.return_value = "no tool called"
+
+    tools = NotificationTools()
+    with pytest.raises(NotificationNoToolCalledError):
+        await agent.decide_notification("mission", [], tools)

tests/test_orchestrator.py

@@ -21,7 +21,9 @@ def temp_state_file(tmp_path: Path) -> Path:
 @patch("core.orchestrator.AgentOrchestrator._get_state_file_path")
 @patch("core.orchestrator.AgentDispatcher")
 @patch("core.orchestrator.WorkspaceManager")
+@patch("core.orchestrator.AgentFactory")
 async def test_poll_and_dispatch_no_notifications(
+    mock_factory: MagicMock,
     mock_workspace_class: MagicMock,
     mock_dispatcher_class: MagicMock,
     mock_get_path: MagicMock,
@@ -44,13 +46,27 @@ async def test_poll_and_dispatch_no_notifications(
 @patch("core.orchestrator.AgentOrchestrator._get_state_file_path")
 @patch("core.orchestrator.AgentDispatcher")
 @patch("core.orchestrator.WorkspaceManager")
+@patch("core.orchestrator.AgentFactory")
 async def test_poll_and_dispatch_with_notifications(
+    mock_factory: MagicMock,
     mock_workspace_class: MagicMock,
     mock_dispatcher_class: MagicMock,
     mock_get_path: MagicMock,
     temp_state_file: Path
 ) -> None:
     mock_get_path.return_value = temp_state_file
+    
+    mock_reader = MagicMock()
+    async def mock_decide_notification(mission: str, inspection_tools: list, notification_tools) -> str:
+        if "issue" in mission or "42" in mission:
+            notification_tools.process_issue("meeks", "repo1", 42, "Needs processing")
+        elif "pull" in mission or "10" in mission:
+            notification_tools.process_pr("meeks", "repo1", 10, "Needs processing")
+        else:
+            notification_tools.skip_notification("Unrelated")
+        return "Decided"
+    mock_reader.decide_notification = AsyncMock(side_effect=mock_decide_notification)
+    mock_factory.create_notification_reader_agent.return_value = mock_reader
     mock_client = MagicMock(spec=GiteaClient)
     mock_tools = MagicMock(spec=GiteaTools)