fix: escape find_experiment_in_project pattern as literal CI regex (#6)

Author: prassanna-ravishankar

src/clearml_mcp/clearml_mcp.py

@@ -26,6 +26,17 @@ def initialize_clearml_connection() -> None:
 _TASK_FIELDS = ("id", "name", "status", "type", "comment", "created", "project", "tags")
 
 
+def _literal_ci_regex(query: str) -> str:
+    """Build a case-insensitive regex that matches ``query`` as a literal substring.
+
+    ClearML matches ``task_name`` and ``_any_`` patterns as regular expressions
+    server-side, so user input must be escaped to match literally (e.g. ``a.b``
+    matches the literal string, ``[`` does not error the query) and prefixed with
+    ``(?i)`` to preserve the case-insensitive behaviour callers expect.
+    """
+    return f"(?i){re.escape(query)}"
+
+
 def _project_id_to_name(task_dicts: list[dict[str, Any]]) -> dict[str, str]:
     """Build a project-id -> project-name map for the projects referenced by tasks.
 
@@ -311,8 +322,11 @@ async def find_experiment_in_project(
 ) -> list[dict[str, Any]]:
     """Find experiments in a specific project by name pattern."""
     try:
-        # task_name matching happens server-side, so only matching tasks come back.
-        tasks = _query_task_dicts(project_name=project_name, task_name=experiment_pattern)
+        # Match the pattern as a literal, case-insensitive substring server-side
+        # (ClearML treats task_name as a regex), so only matching tasks come back.
+        tasks = _query_task_dicts(
+            project_name=project_name, task_name=_literal_ci_regex(experiment_pattern)
+        )
         return [
             {
                 "id": t["id"],
@@ -412,8 +426,7 @@ async def search_tasks(query: str, project_name: str | None = None) -> list[dict
     try:
         # Match the query as a literal substring (case-insensitive) against
         # name/comment/tags server-side, so we never hydrate non-matching tasks.
-        pattern = f"(?i){re.escape(query)}"
-        tasks = _query_task_dicts(project_name=project_name, any_pattern=pattern)
+        tasks = _query_task_dicts(project_name=project_name, any_pattern=_literal_ci_regex(query))
         return [
             {
                 "id": t["id"],

tests/test_clearml_mcp.py

@@ -721,10 +721,59 @@ async def test_find_experiment_in_project_returns_matching_experiments(self, moc
         assert result[0]["name"] == "Training Experiment"
         assert result[1]["name"] == "Validation Experiment"
         # The pattern is matched server-side via task_name, not by hydrating each task.
+        # It is built as a case-insensitive literal-substring regex.
         _, kwargs = mock_task.query_tasks.call_args
-        assert kwargs["task_name"] == "experiment"
+        assert kwargs["task_name"] == "(?i)experiment"
         mock_task.get_task.assert_not_called()
 
+    @pytest.mark.asyncio
+    @patch("clearml_mcp.clearml_mcp.Task")
+    async def test_find_experiment_escapes_regex_special_characters(self, mock_task):
+        """find_experiment_in_project matches the pattern literally, not as a regex.
+
+        ClearML treats task_name as a regex, so metacharacters must be escaped:
+        "a.b" should match only the literal "a.b" (not "axb"), and "exp[1]" must
+        not error the query.
+        """
+        mock_task.query_tasks.side_effect = _make_query_tasks([])
+        mock_task.get_projects.return_value = []
+
+        result = await clearml_mcp.find_experiment_in_project.fn("ML Project", "exp[1].a")
+
+        assert result == []
+        _, kwargs = mock_task.query_tasks.call_args
+        # Special characters are escaped (literal match) and case-insensitive.
+        assert kwargs["task_name"] == r"(?i)exp\[1\]\.a"
+        mock_task.get_task.assert_not_called()
+
+    @pytest.mark.asyncio
+    @patch("clearml_mcp.clearml_mcp.Task")
+    async def test_find_experiment_is_case_insensitive(self, mock_task):
+        """find_experiment_in_project preserves case-insensitive matching."""
+        mock_task.query_tasks.side_effect = _make_query_tasks(
+            [
+                {
+                    "id": "task_1",
+                    "name": "Training EXPERIMENT",
+                    "status": "completed",
+                    "type": "training",
+                    "comment": "",
+                    "created": "2024-01-01T00:00:00Z",
+                    "project": "proj_1",
+                    "tags": [],
+                }
+            ]
+        )
+        mock_task.get_projects.return_value = [_fake_project("proj_1", "ML Project")]
+
+        result = await clearml_mcp.find_experiment_in_project.fn("ML Project", "experiment")
+
+        # The (?i) prefix makes the server match regardless of case.
+        _, kwargs = mock_task.query_tasks.call_args
+        assert kwargs["task_name"].startswith("(?i)")
+        assert len(result) == 1
+        assert result[0]["name"] == "Training EXPERIMENT"
+
     @pytest.mark.asyncio
     @patch("clearml_mcp.clearml_mcp.Task")
     async def test_find_experiment_returns_error_on_query_failure(self, mock_task):