Fix polymorphic URL artifact input validation

CHANGELOG.md

@@ -16,6 +16,11 @@ using [PEP 440](https://packaging.python.org/en/latest/specifications/version-sp
 
 - Reads variable values from the bundled configuration rather than the pipeline signature; signatures written by older releases still supply their defaults.
 
+### Fixed
+
+- Uses the same framework type checking for URL and local artifact inputs,
+  avoiding parse failures on polymorphic type display labels.
+
 ## 0.1.2 - 2026-10-11
 
 ### Added

src/adagio/cli/task_exec.py

@@ -202,13 +202,6 @@ def resolve(name: str, source: str, *, metadata: bool = False) -> str:
             raise ValueError("URL inputs support QIIME artifacts and metadata, not raw imports.")
 
         from qiime2 import Artifact, Metadata
-        from qiime2.sdk.util import parse_type
-
-        expected = None
-        if not metadata and declared_type and declared_type != "Metadata":
-            expected = parse_type(declared_type)
-            if declared_type.startswith(("List[", "Collection[")):
-                expected = expected.fields[0]
 
         def validate(path: str) -> dict[str, str]:
             try:
@@ -232,10 +225,6 @@ def validate(path: str) -> dict[str, str]:
             raise ValueError("Remote inputs require a task result manifest in the run work directory.")
         store = RemoteArtifactStore(Path(manifest).parent / ".adagio-downloads")
         path, record = store.resolve(source, validate=validate)
-        if expected is not None and not parse_type(record["type"]) <= expected:
-            raise ValueError(
-                f"Remote input {name!r} has type {record['type']}; expected {declared_type}."
-            )
         entry = {"input": name, **record}
         downloads.append(entry)
         print("Input download: " + json.dumps(entry), flush=True)

tests/integration/remote_polymorphic_inputs.py

@@ -0,0 +1,91 @@
+"""Run with the taxa plugin installed and the patched adagio package on PYTHONPATH."""
+
+from io import BytesIO
+import json
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from biom import Table
+import numpy as np
+import pandas as pd
+from qiime2 import Artifact
+from qiime2.sdk import PluginManager
+
+from adagio.cli.task_exec import _run_task
+
+
+URL = "https://example.org/table.qza"
+
+
+class Response(BytesIO):
+    def __init__(self, payload):
+        super().__init__(payload)
+        self.headers = {"Content-Length": str(len(payload))}
+
+    def geturl(self):
+        return URL
+
+
+with tempfile.TemporaryDirectory() as temporary:
+    root = Path(temporary)
+    action = PluginManager().plugins["taxa"].actions["filter_table"]
+    display_type = str(action.signature.inputs["table"].qiime_type)
+    assert "¹" in display_type, display_type
+    table = Artifact.import_data(
+        "FeatureTable[Frequency]",
+        Table(np.array([[1, 2], [3, 4]]), ["f1", "f2"], ["s1", "s2"]),
+    )
+    taxonomy = Artifact.import_data(
+        "FeatureData[Taxonomy]",
+        pd.DataFrame(
+            {"Taxon": ["k__Bacteria; p__Firmicutes", "k__Bacteria; p__Proteobacteria"]},
+            index=pd.Index(["f1", "f2"], name="Feature ID"),
+        ),
+    )
+    table_path = table.save(str(root / "table.qza"))
+    taxonomy_path = taxonomy.save(str(root / "taxonomy.qza"))
+    payload = Path(table_path).read_bytes()
+    output_name = next(iter(action.signature.outputs))
+    calls = []
+
+    def open_response(request, timeout):
+        calls.append(request.full_url)
+        return Response(payload)
+
+    def run(name, source):
+        work = root / name
+        work.mkdir()
+        spec = {
+            "plugin": "taxa",
+            "action": "filter_table",
+            "archive_inputs": {"table": source, "taxonomy": taxonomy_path},
+            "remote_input_types": {"table": display_type},
+            "params": {"include": "Firmicutes"},
+            "outputs": {output_name: str(work / "filtered")},
+            "result_manifest": str(work / "results.json"),
+        }
+        _run_task(spec)
+        manifest = json.loads((work / "results.json").read_text())
+        return Artifact.load(manifest["outputs"][output_name]).view(Table)
+
+    with patch(
+        "adagio.remote_inputs.build_opener",
+        return_value=SimpleNamespace(open=open_response),
+    ):
+        local = run("local", table_path)
+        remote = run("remote", URL)
+        assert local == remote
+        assert list(remote.ids(axis="observation")) == ["f1"]
+        assert calls == [URL]
+        payload = Path(taxonomy_path).read_bytes()
+        for name, source in (("invalid-local", taxonomy_path), ("invalid-remote", URL)):
+            try:
+                run(name, source)
+            except TypeError as error:
+                assert "requires an argument of type FeatureTable" in str(error), str(error)
+            else:
+                raise AssertionError("Taxonomy was accepted as a feature table")
+
+    print("PASS: taxa.filter_table produces the same result for URL and local inputs; wrong types are rejected.")

tests/test_remote_inputs.py

@@ -348,12 +348,22 @@ def test_worker_rejects_invalid_remote_metadata(monkeypatch, tmp_path, fake_qiim
     assert not list((tmp_path / ".adagio-downloads").glob("*.tsv"))
 
 
-def test_worker_rejects_declared_type_mismatch(monkeypatch, tmp_path, fake_qiime):
+def test_worker_stages_polymorphic_input_without_parsing_display_type(
+    monkeypatch, tmp_path, fake_qiime
+):
     network(monkeypatch, iter([Response()]))
     spec = task_spec(tmp_path)
-    spec["remote_input_types"]["data"] = "FeatureData[Sequence]"
-    with pytest.raises(ValueError, match="expected FeatureData"):
-        _stage_remote_inputs(spec)
+    spec["remote_input_types"]["data"] = (
+        "FeatureTable[Frequency¹ | PresenceAbsence² | RelativeFrequency³]"
+    )
+
+    def unexpected_parse_type(text):
+        raise AssertionError("Staging must defer semantic type checking to the action")
+
+    monkeypatch.setattr(sys.modules["qiime2.sdk.util"], "parse_type", unexpected_parse_type)
+    staged, records = _stage_remote_inputs(spec)
+    assert Path(staged["archive_inputs"]["data"]).read_bytes() == PAYLOAD
+    assert records[0]["type"] == INFO["type"]
 
 
 def test_worker_accepts_collection_member_type(monkeypatch, tmp_path, fake_qiime):