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

Merged
michael merged 1 commits from feat/notification-reader-agent into master 2026-06-30 22:01:08 +02:00
7 changed files with 288 additions and 2 deletions
+7
View File
@@ -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."""
+45
View File
@@ -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
+68
View File
@@ -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."
+50 -2
View File
@@ -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()
+19
View File
@@ -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.
"""
+83
View File
@@ -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)
+16
View File
@@ -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)