Move UserToolSource container-shape check to a lint rule

Reviewer feedback on galaxyproject#22615: the container regex/prefix shape check is a
linter concern, not a tool source validation concern, and applies to any
ToolSource (not just YAML). Pull the shape check off
`UserToolSource._check_container_shape` and surface it as a new
`ContainerImageShape` linter under `galaxy.tool_util.linters.containers`,
discovered automatically by the existing lint framework
(`lint_tool_types = ["*"]`). The linter walks
`tool_source.parse_requirements()` and warns on identifiers that do not
match a recognized prefix or `<image>[:<tag>]` shape.

For Galaxy User Defined tools, enforcement is unconditional: the
unprivileged tools API and the `CustomToolAgent` both call a new
`lint_user_tool_source` helper after pydantic validation succeeds and
block creation if WARN+ messages are returned.

The not-empty/whitespace check on `container` stays on the pydantic
model — that is a basic field validity check rather than a "shape"
concern. Container-shape tests move from
`test_user_tool_source_validation.py` to a new
`test_container_shape_lint.py` covering both the linter and the helper.

Author: mvdbeek

lib/galaxy/agents/custom_tool.py

@@ -18,6 +18,7 @@
 )
 
 from galaxy.schema.agents import ConfidenceLevel
+from galaxy.tool_util.lint import lint_user_tool_source
 from galaxy.tool_util_models import (
     format_validation_errors,
     UserToolSource,
@@ -118,6 +119,23 @@ async def process(self, query: str, context: Optional[dict[str, Any]] = None) ->
                     error="invalid_structured_output",
                 )
 
+            lint_errors = lint_user_tool_source(tool)
+            if lint_errors:
+                log.debug("CustomToolAgent lint failure: %s", lint_errors)
+                bullet_text = "\n".join(f"- {issue}" for issue in lint_errors)
+                return self._build_response(
+                    content=(
+                        "The model produced a tool definition, but it has problems "
+                        "that need to be fixed before it can be saved:\n\n"
+                        f"{bullet_text}"
+                    ),
+                    confidence=ConfidenceLevel.LOW,
+                    method="lint_error",
+                    query=query,
+                    error="lint_failed",
+                    agent_data={"lint_errors": lint_errors},
+                )
+
             tool_dict = tool.model_dump(by_alias=True, exclude_none=True)
             tool_yaml = yaml.dump(tool_dict, default_flow_style=False, sort_keys=False)
 

lib/galaxy/managers/tools.py

@@ -28,6 +28,7 @@
     UserDynamicToolAssociation,
 )
 from galaxy.tool_util.cwl import tool_proxy
+from galaxy.tool_util.lint import lint_user_tool_source
 from galaxy.tool_util.parser.yaml import YamlToolSource
 from galaxy.tool_util.toolbox import AbstractToolBox
 from galaxy.tool_util_models.dynamic_tool_models import (
@@ -194,6 +195,11 @@ def create_unprivileged_tool(
                 "Set 'enable_beta_tool_formats' in Galaxy config to create dynamic tools."
             )
         self.ensure_can_use_unprivileged_tool(user)
+        lint_errors = lint_user_tool_source(tool_payload.representation)
+        if lint_errors:
+            raise exceptions.RequestParameterInvalidException(
+                "Tool failed lint checks: " + "; ".join(lint_errors)
+            )
         dynamic_tool = self.create(
             tool_format=tool_payload.representation.class_,
             tool_id=tool_payload.representation.id,

lib/galaxy/tool_util/lint.py

@@ -62,13 +62,15 @@
 
 import galaxy.tool_util.linters
 from galaxy.tool_util.parser import get_tool_source
+from galaxy.tool_util.parser.yaml import YamlToolSource
 from galaxy.util import (
     Element,
     submodules,
 )
 
 if TYPE_CHECKING:
     from galaxy.tool_util.parser.interface import ToolSource
+    from galaxy.tool_util_models import UserToolSource
 
 
 class LintLevel(IntEnum):
@@ -308,6 +310,23 @@ def failed(self, fail_level: Union[LintLevel, str]) -> bool:
         return lint_fail
 
 
+def lint_user_tool_source(user_tool_source: "UserToolSource") -> List[str]:
+    """Run the lint pipeline against a ``UserToolSource`` pydantic value.
+
+    Returns a list of formatted ``"<linter>: <message>"`` bullets at WARN
+    level or above, suitable for surfacing through ``format_validation_errors``-
+    style consumers (the agent's bullet list, an API 4xx body).
+    """
+    root_dict = user_tool_source.model_dump(by_alias=True, exclude_none=True)
+    tool_source = YamlToolSource(root_dict)
+    lint_context = get_lint_context_for_tool_source(tool_source)
+    bullets: List[str] = []
+    for message in lint_context.error_messages + lint_context.warn_messages:
+        prefix = f"{message.linter}: " if message.linter else ""
+        bullets.append(f"{prefix}{message.message}")
+    return bullets
+
+
 def lint_tool_source(
     tool_source, level=LintLevel.ALL, fail_level=LintLevel.WARN, extra_modules=None, skip_types=None, name=None
 ) -> bool:

lib/galaxy/tool_util/linters/containers.py

@@ -0,0 +1,54 @@
+"""Linter rules covering container references on a tool source."""
+
+import re
+from typing import (
+    Iterator,
+    Tuple,
+    TYPE_CHECKING,
+)
+
+from galaxy.tool_util.lint import Linter
+
+if TYPE_CHECKING:
+    from galaxy.tool_util.lint import LintContext
+    from galaxy.tool_util.parser.interface import ToolSource
+
+
+lint_tool_types = ["*"]
+
+
+CONTAINER_PREFIXES: Tuple[str, ...] = ("quay.io/biocontainers/", "docker://", "oras://")
+DOCKER_IMAGE_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*(/[a-zA-Z0-9._-]+)*(:[\w][\w.-]*)?$")
+
+
+def _iter_container_identifiers(tool_source: "ToolSource") -> Iterator[str]:
+    try:
+        _, containers, _, _, _ = tool_source.parse_requirements()
+    except Exception:
+        return
+    for container in containers or ():
+        identifier = getattr(container, "identifier", None)
+        if identifier:
+            yield identifier
+
+
+class ContainerImageShape(Linter):
+    """Container identifiers should match a recognized shape.
+
+    Recognized: a `quay.io/biocontainers/...`, `docker://...`, or `oras://...`
+    prefix; or a Docker-Hub-style `<image>[:<tag>]` reference.
+    """
+
+    @classmethod
+    def lint(cls, tool_source: "ToolSource", lint_ctx: "LintContext"):
+        for identifier in _iter_container_identifiers(tool_source):
+            stripped = identifier.strip()
+            if stripped.startswith(CONTAINER_PREFIXES):
+                continue
+            if DOCKER_IMAGE_RE.match(stripped):
+                continue
+            lint_ctx.warn(
+                f"container '{identifier}' does not match a recognized shape "
+                "(quay.io/biocontainers/..., docker://..., oras://..., or <image>[:<tag>])",
+                linter=cls.name(),
+            )

lib/galaxy/tool_util_models/__init__.py

@@ -11,7 +11,6 @@
     List,
     Optional,
     Set,
-    Tuple,
     Union,
 )
 
@@ -73,12 +72,6 @@ def normalize_dict(values, keys: List[str]):
 _TOOL_ID_RE = re.compile(r"^[a-z][a-z0-9_-]*$")
 TOOL_ID_PATTERN = r"^[a-z][a-z0-9_-]*$"
 
-# Recognized container reference shapes. Kept module-level so future
-# admin-configurable code can extend them in place. Docker-Hub-style image
-# references allow optional registry path segments and an optional tag.
-CONTAINER_PREFIXES: Tuple[str, ...] = ("quay.io/biocontainers/", "docker://", "oras://")
-DOCKER_IMAGE_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*(/[a-zA-Z0-9._-]+)*(:[\w][\w.-]*)?$")
-
 # Templated ecmascript inside `shell_command` / `configfiles[*].content`.
 # Pull every $(<expr>) and extract the *leading* 'inputs.<name>' identifier.
 # Only the top-level name is checked, so nested references like
@@ -239,18 +232,10 @@ class UserToolSource(_DynamicToolSourceBase):
 
     @field_validator("container", mode="after")
     @classmethod
-    def _check_container_shape(cls, value: str) -> str:
+    def _reject_blank_container(cls, value: str) -> str:
         if not value or not value.strip():
             raise ValueError("container must not be empty")
-        stripped = value.strip()
-        if stripped.startswith(CONTAINER_PREFIXES):
-            return value
-        if DOCKER_IMAGE_RE.match(stripped):
-            return value
-        raise ValueError(
-            f"container '{value}' does not match a recognized shape "
-            "(quay.io/biocontainers/..., docker://..., oras://..., or <image>[:<tag>])"
-        )
+        return value
 
 
 class YamlToolSource(_DynamicToolSourceBase):

test/unit/tool_util/test_container_shape_lint.py

@@ -0,0 +1,78 @@
+"""Tests for the ContainerImageShape linter and the lint_user_tool_source helper."""
+
+from copy import deepcopy
+
+import pytest
+
+from galaxy.tool_util.lint import (
+    get_lint_context_for_tool_source,
+    lint_user_tool_source,
+)
+from galaxy.tool_util.linters.containers import ContainerImageShape
+from galaxy.tool_util.parser.yaml import YamlToolSource
+from galaxy.tool_util_models import UserToolSource
+
+VALID_TOOL = {
+    "class": "GalaxyUserTool",
+    "id": "my-cool-tool",
+    "name": "My Cool Tool",
+    "version": "0.1.0",
+    "description": "A cool tool.",
+    "container": "quay.io/biocontainers/python:3.13",
+    "shell_command": "head -n '$(inputs.n_lines)' '$(inputs.data_input.path)' > out.txt",
+    "inputs": [
+        {"type": "integer", "name": "n_lines"},
+        {"type": "data", "name": "data_input"},
+    ],
+    "outputs": [
+        {"type": "data", "name": "out", "from_work_dir": "out.txt"},
+    ],
+    "citations": [
+        {"type": "doi", "content": "10.1234/abc.def"},
+    ],
+}
+
+
+def _doc(**overrides):
+    base = deepcopy(VALID_TOOL)
+    base.update(overrides)
+    return base
+
+
+@pytest.mark.parametrize(
+    "container",
+    [
+        "quay.io/biocontainers/samtools:1.17",
+        "docker://my-registry/image:tag",
+        "oras://example.org/image",
+        "busybox",
+        "ubuntu:latest",
+        "library/python:3.11-slim",
+    ],
+)
+def test_valid_container_shapes_pass_lint(container):
+    tool_source = YamlToolSource(_doc(container=container))
+    ctx = get_lint_context_for_tool_source(tool_source)
+    container_warns = [m for m in ctx.warn_messages if m.linter == ContainerImageShape.name()]
+    assert container_warns == []
+
+
+@pytest.mark.parametrize("bad_container", ["definitely not a container", "foo bar baz"])
+def test_invalid_container_shapes_warn(bad_container):
+    tool_source = YamlToolSource(_doc(container=bad_container))
+    ctx = get_lint_context_for_tool_source(tool_source)
+    container_warns = [m for m in ctx.warn_messages if m.linter == ContainerImageShape.name()]
+    assert len(container_warns) == 1
+    assert "does not match a recognized shape" in container_warns[0].message
+
+
+def test_lint_user_tool_source_returns_empty_on_clean_tool():
+    user_tool = UserToolSource.model_validate(VALID_TOOL)
+    assert lint_user_tool_source(user_tool) == []
+
+
+def test_lint_user_tool_source_surfaces_container_shape_failure():
+    user_tool = UserToolSource.model_validate(_doc(container="totally bogus value"))
+    bullets = lint_user_tool_source(user_tool)
+    assert any("does not match a recognized shape" in b for b in bullets)
+    assert any(b.startswith(f"{ContainerImageShape.name()}:") for b in bullets)

test/unit/tool_util/test_tool_linters.py

@@ -2543,7 +2543,7 @@ def test_skip_by_module(lint_ctx):
 def test_list_linters():
     linter_names = Linter.list_listers()
     # make sure to add/remove a test for new/removed linters if this number changes
-    assert len(linter_names) == 145
+    assert len(linter_names) == 146
     assert "Linter" not in linter_names
     # make sure that linters from all modules are available
     for prefix in [

test/unit/tool_util/test_user_tool_source_validation.py

@@ -1,10 +1,13 @@
 """Direct tests for the semantic validators on ``UserToolSource``.
 
 These tests construct ``UserToolSource`` payloads in-memory and assert that
-each rule (id pattern, container shape, citation shape, undeclared
-``inputs.<name>`` references in ``shell_command`` / ``configfiles``,
-output discovery requirements, blank required fields) raises a
-``ValidationError`` whose distilled output identifies the offending field.
+each rule (id pattern, citation shape, undeclared ``inputs.<name>``
+references in ``shell_command`` / ``configfiles``, output discovery
+requirements, blank required fields) raises a ``ValidationError`` whose
+distilled output identifies the offending field.
+
+Container *shape* is enforced by the lint framework, not pydantic — see
+``test_container_shape_lint.py``.
 """
 
 from copy import deepcopy
@@ -67,26 +70,11 @@ def test_invalid_id_rejected(bad_id):
     _assert_error_contains(info.value, "String should match pattern")
 
 
-@pytest.mark.parametrize("bad_container", ["", "   ", "definitely not a container", "foo bar baz"])
-def test_invalid_container_rejected(bad_container):
+@pytest.mark.parametrize("bad_container", ["", "   "])
+def test_blank_container_rejected(bad_container):
     with pytest.raises(ValidationError) as info:
         UserToolSource.model_validate(_doc(container=bad_container))
-    _assert_error_contains(info.value, "container")
-
-
-@pytest.mark.parametrize(
-    "container",
-    [
-        "quay.io/biocontainers/samtools:1.17",
-        "docker://my-registry/image:tag",
-        "oras://example.org/image",
-        "busybox",
-        "ubuntu:latest",
-        "library/python:3.11-slim",
-    ],
-)
-def test_valid_container_shapes(container):
-    UserToolSource.model_validate(_doc(container=container))
+    _assert_error_contains(info.value, "container must not be empty")
 
 
 def test_blank_name_rejected():