feat: auto-trigger commands when bot is assigned as reviewer on GitLab MR

GitLab provides built-in commands like /request_review and /assign_reviewer
which currently do not trigger the PR-Agent. This PR adds webhook handling
for reviewer assignment events to automatically trigger commands (default: /review).

Features:
- Adds handle_reviewer_assignment toggle under [gitlab] section (default: false)
- Supports configurable reviewer_commands array (default: ['/review'])
- Bot identifies itself dynamically using the GitLab API
- Safely skips draft MRs and invalid events
- Validates configurations and payload structures to avoid exceptions

Assisted-by: opencode:deepseek-v4-pro

Author: igoraj

pr_agent/servers/gitlab_webhook.py

@@ -1,4 +1,6 @@
+import asyncio
 import copy
+import hashlib
 import json
 import os
 import re
@@ -108,38 +110,70 @@ def is_draft_ready(data) -> bool:
         get_logger().error(f"Failed 'is_draft_ready' logic: {e}")
     return False
 
-_bot_user_id_cache = None
+_bot_user_id_cache = {}
 
-def _get_bot_user_id():
-    global _bot_user_id_cache
-    if _bot_user_id_cache is not None:
-        return _bot_user_id_cache
-    try:
+async def _get_bot_user_id():
+    gitlab_url = get_settings().get("GITLAB.URL", "https://gitlab.com")
+    gitlab_token = get_settings().get("GITLAB.PERSONAL_ACCESS_TOKEN", None)
+    if not gitlab_token:
+        get_logger().error("No GitLab token available for bot user ID resolution")
+        return None
+
+    cache_key = hashlib.sha256(f"{gitlab_url}:{gitlab_token}".encode()).hexdigest()
+
+    cached = _bot_user_id_cache.get(cache_key)
+    if cached is not None:
+        return cached if cached != -1 else None
+
+    def _resolve_sync():
         import gitlab
-        gl = gitlab.Gitlab(
-            get_settings().get("GITLAB.URL", "https://gitlab.com"),
-            private_token=get_settings().get("GITLAB.PERSONAL_ACCESS_TOKEN", None),
-        )
+
+        ssl_verify = get_settings().get("GITLAB.SSL_VERIFY", True)
+        if isinstance(ssl_verify, str):
+            ssl_verify = ssl_verify.lower() in ("true", "1", "yes")
+
+        auth_method = get_settings().get("GITLAB.AUTH_TYPE", "oauth_token")
+        if auth_method not in ("oauth_token", "private_token"):
+            auth_method = "oauth_token"
+
+        kwargs = {"url": gitlab_url, "ssl_verify": ssl_verify}
+        if auth_method == "oauth_token":
+            kwargs["oauth_token"] = gitlab_token
+        else:
+            kwargs["private_token"] = gitlab_token
+
+        gl = gitlab.Gitlab(**kwargs)
         gl.auth()
-        _bot_user_id_cache = gl.user.id
-        get_logger().info(f"Bot user ID resolved via API: {_bot_user_id_cache}")
-        return _bot_user_id_cache
+        return gl.user.id
+
+    try:
+        user_id = await asyncio.to_thread(_resolve_sync)
+        if len(_bot_user_id_cache) > 1000:
+            _bot_user_id_cache.clear()
+        _bot_user_id_cache[cache_key] = user_id
+        get_logger().info(f"Bot user ID resolved via API: {user_id}")
+        return user_id
     except Exception as e:
         get_logger().error(f"Failed to resolve bot user ID: {e}")
         return None
 
-def is_bot_assigned_as_reviewer(data) -> bool:
+async def is_bot_assigned_as_reviewer(data) -> bool:
     try:
-        if 'reviewers' not in data.get('changes', {}):
+        changes = data.get("changes")
+        if not isinstance(changes, dict):
             return False
-        reviewers_change = data['changes']['reviewers']
-        previous = reviewers_change.get('previous', [])
-        current = reviewers_change.get('current', [])
-        bot_user_id = _get_bot_user_id()
+        if "reviewers" not in changes:
+            return False
+        reviewers_change = changes["reviewers"]
+        if not isinstance(reviewers_change, dict):
+            return False
+        previous = reviewers_change.get("previous", [])
+        current = reviewers_change.get("current", [])
+        bot_user_id = await _get_bot_user_id()
         if bot_user_id is None:
             return False
-        previous_ids = {r.get('id') for r in previous} if isinstance(previous, list) else set()
-        current_ids = {r.get('id') for r in current} if isinstance(current, list) else set()
+        previous_ids = {r.get("id") for r in previous if isinstance(r, dict)}
+        current_ids = {r.get("id") for r in current if isinstance(r, dict)}
         return bot_user_id in current_ids and bot_user_id not in previous_ids
     except Exception as e:
         get_logger().error(f"Failed 'is_bot_assigned_as_reviewer' logic: {e}")
@@ -254,7 +288,9 @@ async def inner(data: dict):
             # ignore MRs based on title, labels, source and target branches
             if not should_process_pr_logic(data):
                 return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
-            object_attributes = data.get('object_attributes', {})
+            object_attributes = data.get('object_attributes')
+            if not isinstance(object_attributes, dict):
+                object_attributes = {}
             if object_attributes.get('action') in ['open', 'reopen']:
                 url = object_attributes.get('url')
                 get_logger().info(f"New merge request: {url}")
@@ -284,7 +320,7 @@ async def inner(data: dict):
 
                 get_logger().debug(f'A push event has been received: {url}')
                 await _perform_commands_gitlab("push_commands", PRAgent(), url, log_context, data)
-                
+
             # for draft to ready triggered merge requests
             elif object_attributes.get('action') == 'update' and is_draft_ready(data):
                 url = object_attributes.get('url')
@@ -294,10 +330,40 @@ async def inner(data: dict):
                 await _perform_commands_gitlab("pr_commands", PRAgent(), url, log_context, data)
 
             # for reviewer assignment triggered merge requests
-            elif object_attributes.get('action') == 'update' and is_bot_assigned_as_reviewer(data):
+            elif object_attributes.get('action') == 'update' and not object_attributes.get('oldrev'):
                 url = object_attributes.get('url')
+                if not url:
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+
+                # Fast early-exit: no reviewer changes means nothing to do
+                changes = data.get("changes")
+                if not isinstance(changes, dict) or "reviewers" not in changes:
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+
                 apply_repo_settings(url)
-                if get_settings().gitlab.get('handle_reviewer_assignment', False):
+                handle_assignment = get_settings().gitlab.get("handle_reviewer_assignment", False)
+                if isinstance(handle_assignment, str):
+                    handle_assignment = handle_assignment.lower() in ("true", "1", "yes")
+                if not handle_assignment:
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+
+                # Check PR logic after applying repo settings
+                if not should_process_pr_logic(data):
+                    return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder({"message": "success"}))
+
+                if is_draft(data):
+                    get_logger().info(f"Skipping draft MR reviewer assignment: {url}")
+                    return JSONResponse(status_code=status.HTTP_200_OK,
+                                        content=jsonable_encoder({"message": "success"}))
+                if await is_bot_assigned_as_reviewer(data):
+                    reviewer_commands = get_settings().gitlab.get("reviewer_commands", [])
+                    if not isinstance(reviewer_commands, list) or not all(isinstance(c, str) for c in reviewer_commands):
+                        get_logger().warning("gitlab.reviewer_commands is not a list of strings, skipping")
+                        return JSONResponse(status_code=status.HTTP_200_OK,
+                                            content=jsonable_encoder({"message": "success"}))
                     get_logger().info(f"Bot was assigned as reviewer on MR: {url}")
                     await _perform_commands_gitlab("reviewer_commands", PRAgent(), url, log_context, data)
 

tests/unittest/test_gitlab_reviewer_assignment.py

@@ -0,0 +1,179 @@
+import asyncio
+from unittest import mock
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def setup_env(monkeypatch):
+    monkeypatch.setenv("GITLAB__URL", "https://gitlab.example.com")
+
+
+class TestIsBotAssignedAsReviewer:
+    BOT_ID = 516
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_detects_new_assignment(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [],
+                    "current": [{"id": self.BOT_ID, "username": "k2so-bot"}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is True
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_ignores_already_assigned(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [{"id": self.BOT_ID, "username": "k2so-bot"}],
+                    "current": [{"id": self.BOT_ID, "username": "k2so-bot"}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_no_reviewers_key(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {"changes": {"updated_at": {"previous": "old", "current": "new"}}}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    def test_changes_not_dict(self):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        data = {"changes": "not-a-dict"}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_reviewers_not_dict(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {"changes": {"reviewers": "not-a-dict"}}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    def test_no_changes_key(self):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        data = {"object_kind": "merge_request"}
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_bot_id_unresolvable(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = None
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [],
+                    "current": [{"id": self.BOT_ID}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is False
+
+    @mock.patch("pr_agent.servers.gitlab_webhook._get_bot_user_id", new_callable=mock.AsyncMock)
+    def test_previous_with_non_dict_entries(self, mock_bot_id):
+        from pr_agent.servers.gitlab_webhook import is_bot_assigned_as_reviewer
+        mock_bot_id.return_value = self.BOT_ID
+        data = {
+            "changes": {
+                "reviewers": {
+                    "previous": [{"id": 100}, "not-a-dict"],
+                    "current": [{"id": self.BOT_ID}],
+                }
+            }
+        }
+        assert asyncio.run(is_bot_assigned_as_reviewer(data)) is True
+
+
+class TestGetBotUserId:
+    @staticmethod
+    def _make_fake_gitlab(user_id):
+        fake = mock.MagicMock()
+        fake.Gitlab.return_value.auth.return_value = None
+        fake.Gitlab.return_value.user.id = user_id
+        return fake
+
+    @staticmethod
+    def _make_settings(url, token):
+        s = mock.MagicMock()
+        s.get.side_effect = lambda k, d=None: {
+            "GITLAB.URL": url,
+            "GITLAB.PERSONAL_ACCESS_TOKEN": token,
+            "GITLAB.SSL_VERIFY": True,
+            "GITLAB.AUTH_TYPE": "oauth_token",
+        }.get(k, d)
+        return s
+
+    def test_caches_by_credential(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings",
+                        return_value=self._make_settings("https://a.example.com", "token-a")):
+            with mock.patch.dict("sys.modules", {"gitlab": self._make_fake_gitlab(111)}):
+                assert asyncio.run(_get_bot_user_id()) == 111
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings",
+                        return_value=self._make_settings("https://a.example.com", "token-b")):
+            with mock.patch.dict("sys.modules", {"gitlab": self._make_fake_gitlab(222)}):
+                assert asyncio.run(_get_bot_user_id()) == 222
+
+        assert len(_bot_user_id_cache) >= 2
+
+    def test_no_cache_on_failure(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        fake = self._make_fake_gitlab(0)
+        fake.Gitlab.side_effect = RuntimeError("auth failed")
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings",
+                        return_value=self._make_settings("https://x.example.com", "fail-token")):
+            with mock.patch.dict("sys.modules", {"gitlab": fake}):
+                assert asyncio.run(_get_bot_user_id()) is None
+
+        assert len(_bot_user_id_cache) == 0
+
+    def test_respects_auth_type_private_token(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        s = mock.MagicMock()
+        s.get.side_effect = lambda k, d=None: {
+            "GITLAB.URL": "https://x.example.com",
+            "GITLAB.PERSONAL_ACCESS_TOKEN": "tok",
+            "GITLAB.SSL_VERIFY": True,
+            "GITLAB.AUTH_TYPE": "private_token",
+        }.get(k, d)
+
+        fake_gitlab = self._make_fake_gitlab(99)
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings", return_value=s):
+            with mock.patch.dict("sys.modules", {"gitlab": fake_gitlab}):
+                assert asyncio.run(_get_bot_user_id()) == 99
+
+        call_kwargs = fake_gitlab.Gitlab.call_args.kwargs
+        assert "private_token" in call_kwargs
+        assert call_kwargs["private_token"] == "tok"
+
+    def test_no_token_returns_none(self):
+        from pr_agent.servers.gitlab_webhook import _bot_user_id_cache, _get_bot_user_id
+        _bot_user_id_cache.clear()
+
+        s = mock.MagicMock()
+        s.get.side_effect = lambda k, d=None: {
+            "GITLAB.URL": "https://x.example.com",
+            "GITLAB.PERSONAL_ACCESS_TOKEN": None,
+            "GITLAB.SSL_VERIFY": True,
+            "GITLAB.AUTH_TYPE": "oauth_token",
+        }.get(k, d)
+
+        with mock.patch("pr_agent.servers.gitlab_webhook.get_settings", return_value=s):
+            assert asyncio.run(_get_bot_user_id()) is None