fix: validate tags on every tag mutation route

add_prompt_tag, add_tags_to_prompt, bulk_add_tags, merge_tags and
rename_tag accepted whatever the body carried; a non-string tag raised
AttributeError (500) and over-long or control-character tags were
stored. They now run utils.validators.validate_tags through a shared
tags_error helper and answer 400 with the validator's message. Blank
entries in add_tags_to_prompt are still skipped rather than rejected.

py/api/prompts.py

@@ -28,6 +28,15 @@ def bad_request(message):
     return web.json_response({"success": False, "error": message}, status=400)
 
 
+def tags_error(tags):
+    """A 400 response when *tags* fail ``validate_tags``, else None."""
+    try:
+        validate_tags(tags)
+    except ValueError as exc:
+        return bad_request(str(exc))
+    return None
+
+
 def safe_error_message(exc):
     """Describe *exc* for a client without leaking absolute server paths.
 
@@ -511,11 +520,15 @@ class PromptRoutesMixin:
                 return web.json_response(
                     {"success": False, "error": "Invalid JSON body"}, status=400
                 )
-            new_name = (body.get("new_name") or "").strip()
-            if not new_name:
+            raw_name = body.get("new_name")
+            if not raw_name:
                 return web.json_response(
                     {"success": False, "error": "New tag name required"}, status=400
                 )
+            error = tags_error([raw_name])
+            if error is not None:
+                return error
+            new_name = raw_name.strip()
 
             result = await self._run_in_executor(
                 self.db.rename_tag_all_prompts, tag_name, new_name
@@ -581,16 +594,22 @@ class PromptRoutesMixin:
                     {"success": False, "error": "Invalid JSON body"}, status=400
                 )
             source_tags = body.get("source_tags", [])
-            target_tag = (body.get("target_tag") or "").strip()
+            raw_target = body.get("target_tag")
 
             if not source_tags:
                 return web.json_response(
                     {"success": False, "error": "Source tags required"}, status=400
                 )
-            if not target_tag:
+            if not raw_target:
                 return web.json_response(
                     {"success": False, "error": "Target tag required"}, status=400
                 )
+            if not isinstance(source_tags, list):
+                return bad_request("Source tags must be a list")
+            error = tags_error([*source_tags, raw_target])
+            if error is not None:
+                return error
+            target_tag = raw_target.strip()
 
             result = await self._run_in_executor(
                 self.db.merge_tags, source_tags, target_tag
@@ -819,12 +838,16 @@ class PromptRoutesMixin:
         try:
             prompt_id = int(request.match_info["prompt_id"])
             data = await request.json()
-            new_tag = (data.get("tag") or "").strip()
+            raw_tag = data.get("tag")
 
-            if not new_tag:
+            if not raw_tag:
                 return web.json_response(
                     {"success": False, "error": "Tag cannot be empty"}, status=400
                 )
+            error = tags_error([raw_tag])
+            if error is not None:
+                return error
+            new_tag = raw_tag.strip()
 
             prompt = await self._run_in_executor(self.db.get_prompt_by_id, prompt_id)
             if not prompt:
@@ -877,6 +900,12 @@ class PromptRoutesMixin:
                     {"success": False, "error": "Tags must be a non-empty list"},
                     status=400,
                 )
+            # Blank entries are skipped below, so only the rest is validated.
+            error = tags_error(
+                [t for t in new_tags if not (isinstance(t, str) and not t.strip())]
+            )
+            if error is not None:
+                return error
 
             prompt = await self._run_in_executor(self.db.get_prompt_by_id, prompt_id)
             if not prompt:
@@ -1008,6 +1037,11 @@ class PromptRoutesMixin:
                     {"success": False, "error": "No prompt IDs or tags provided"},
                     status=400,
                 )
+            if not isinstance(new_tags, list):
+                return bad_request("Tags must be a list")
+            error = tags_error(new_tags)
+            if error is not None:
+                return error
 
             updated_count = await self._run_in_executor(
                 self.db.bulk_add_tags, prompt_ids, new_tags

tests/test_api_prompts.py

@@ -464,6 +464,46 @@ class TestTagNamesAreDecodedOnce(RouteCoverageCase):
         self.assertEqual(self.api.db.get_prompt_by_id(self.pid)["tags"], [])
 
 
+class TestTagInputValidation(RouteCoverageCase):
+    """Tag mutations run validate_tags: bad input is a 400, never a traceback."""
+
+    def _requests(self, tag):
+        pid = self._save_prompt("validate me", tags=["old"])
+        return (
+            ("POST", f"/prompt_manager/prompts/{pid}/tags", {"tag": tag}),
+            ("POST", "/prompt_manager/prompts/tags", {"prompt_id": pid, "tags": [tag]}),
+            ("POST", "/prompt_manager/bulk/tags", {"prompt_ids": [pid], "tags": [tag]}),
+            (
+                "POST",
+                "/prompt_manager/tags/merge",
+                {"source_tags": ["old"], "target_tag": tag},
+            ),
+            (
+                "POST",
+                "/prompt_manager/tags/merge",
+                {"source_tags": [tag], "target_tag": "t"},
+            ),
+            ("PUT", "/prompt_manager/tags/old", {"new_name": tag}),
+        )
+
+    async def _assert_all_400(self, tag):
+        for method, route, body in self._requests(tag):
+            status, data = await self._json(method, route, json=body)
+
+            self.assertEqual(status, 400, (route, body))
+            self.assertFalse(data["success"], route)
+            self.assertIn("error", data, route)
+
+    async def test_non_string_tags_are_400(self):
+        await self._assert_all_400(123)
+
+    async def test_overlong_tags_are_400(self):
+        await self._assert_all_400("x" * 51)
+
+    async def test_control_characters_are_400(self):
+        await self._assert_all_400("bad\x00tag")
+
+
 class TestTagMutationRoutes(RouteCoverageCase):
 
     async def test_rename_rejects_bad_json_and_empty_name(self):