diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index e82430c3add..92fed7200e1 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -38,6 +38,7 @@ REPORTS: Final = pytest.StashKey[list[pytest.TestReport]]()
 WORKER_INVENTORY: Final = TypeAdapter(tuple[str, ...])
 SHARD_PATH: Final[TypeAdapter[Path | None]] = TypeAdapter(Path | None)
 SHARD_FILES: Final = pytest.StashKey[frozenset[str] | None]()
+FAILED_COLLECTIONS: Final = pytest.StashKey[frozenset[str]]()
 GROUP_SUFFIX: Final = re.compile(r"@integration_fixture_[0-9a-f]{64}$")
 INTEGER_OPTION: Final = TypeAdapter(int)
 STRING_ENVIRONMENT: Final = TypeAdapter(dict[str, str])
@@ -104,6 +105,12 @@ class IntegrationReportPlugin:
     def pytest_runtest_logreport(self, report: pytest.TestReport) -> None:
         self.config.stash[REPORTS].append(report)
 
+    def pytest_collectreport(self, report: pytest.CollectReport) -> None:
+        if report.failed:
+            self.config.stash[FAILED_COLLECTIONS] = self.config.stash.get(FAILED_COLLECTIONS, frozenset()) | {
+                report.nodeid.split("::", 1)[0]
+            }
+
     @pytest.hookimpl(optionalhook=True)
     def pytest_xdist_node_collection_finished(self, node: object, ids: Sequence[str]) -> None:
         self.config.stash[COLLECTED] = tuple(_case_identity(nodeid) for nodeid in ids if _owned(nodeid))
@@ -224,8 +231,12 @@ def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item
         raise pytest.UsageError("Parallel integration runs require CircleCI's shard file selection")
     if files is not None:
         available: Final = frozenset(item.nodeid.split("::", 1)[0] for item in owned)
-        if files - available:
-            raise pytest.UsageError(f"Integration shard selected uncollected files: {sorted(files - available)}")
+        for file in sorted(files - available - config.stash.get(FAILED_COLLECTIONS, frozenset())):
+            config.hook.pytest_collectreport(
+                report=pytest.CollectReport(
+                    file, "failed", f"Integration shard selected {file}, which has no tests", []
+                )
+            )
         selected: Final = tuple(item for item in items if item not in owned or item.nodeid.split("::", 1)[0] in files)
         deselected: Final = tuple(item for item in owned if item.nodeid.split("::", 1)[0] not in files)
         config.hook.pytest_deselected(items=deselected)
diff --git a/tests/integration/database/test_harness_lifecycle.py b/tests/integration/database/test_harness_lifecycle.py
index 60bd7ae72a3..d600a92c5a2 100644
--- a/tests/integration/database/test_harness_lifecycle.py
+++ b/tests/integration/database/test_harness_lifecycle.py
@@ -784,9 +784,16 @@ def test_worker_crash_keeps_the_original_junit_identity_and_an_incomplete_execut
         assert cases[0].find("error") is not None
 
 
-@pytest.mark.parametrize("assignment", ("absent", "missing", "foreign"))
+@pytest.mark.parametrize(
+    ("assignment", "exit_code"),
+    (
+        ("absent", pytest.ExitCode.USAGE_ERROR),
+        ("missing", pytest.ExitCode.USAGE_ERROR),
+        ("foreign", pytest.ExitCode.INTERRUPTED),
+    ),
+)
 def test_invalid_native_selections_fail_before_any_case_runs(
-    tmp_path: Path, assignment: Literal["absent", "missing", "foreign"]
+    tmp_path: Path, assignment: Literal["absent", "missing", "foreign"], exit_code: pytest.ExitCode
 ) -> None:
     root: Final = Path(__file__).resolve().parents[3]
     events: Final = tmp_path / "executed.txt"
@@ -823,7 +830,7 @@ def test_invalid_native_selections_fail_before_any_case_runs(
             timeout=15,
             check=False,
         )
-        assert result.returncode == pytest.ExitCode.USAGE_ERROR, result.stdout + result.stderr
+        assert result.returncode == exit_code, result.stdout + result.stderr
         assert not events.exists()
 
 
diff --git a/tests/unit/integration_support/test_sharding.py b/tests/unit/integration_support/test_sharding.py
index 2317279f2b2..0ffd7f3572a 100644
--- a/tests/unit/integration_support/test_sharding.py
+++ b/tests/unit/integration_support/test_sharding.py
@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import os
+import subprocess
+import sys
 from collections.abc import Mapping
 from dataclasses import dataclass
 from pathlib import Path
@@ -13,6 +16,21 @@ from tests.integration.conftest import COLLECTED, IntegrationReportPlugin, _shar
 from tests.integration.conftest import INVENTORY as WORKER_INVENTORY
 
 INVENTORY: Final = ("test_owned.py::test_first", "test_owned.py::test_skipped", "test_owned.py::test_last")
+INTEGRATION_PACKAGE: Final = Path(__file__).resolve().parents[2] / "integration"
+HEALTHY: Final = "tests/integration/providers/test_healthy.py"
+BROKEN: Final = "tests/integration/providers/test_broken.py"
+RENAMED: Final = "tests/integration/providers/test_renamed.py"
+CLASS_ERROR: Final = "tests/integration/providers/test_class_error.py"
+MODULES: Final = MappingProxyType(
+    {
+        HEALTHY: "def test_one():\n    pass\n\n\ndef test_two():\n    pass\n",
+        BROKEN: "from tests.integration._support.process import _removed_helper\n\n\ndef test_never():\n    pass\n",
+        CLASS_ERROR: (
+            "import pytest\n\n\nclass TestBroken:\n"
+            "    @pytest.mark.parametrize('unused', [1])\n    def test_case(self):\n        pass\n"
+        ),
+    }
+)
 
 
 def _execution(
@@ -111,3 +129,88 @@ def test_worker_results_retain_the_full_inventory_and_reject_disagreement() -> N
     incomplete: Final = WorkerResult(MappingProxyType({"integration_inventory": INVENTORY[:1]}))
     with pytest.raises(pytest.UsageError, match="inventories disagree"):
         plugin.pytest_testnodedown(incomplete, None)
+
+
+XDIST: Final = ("-n", "2", "--dist=loadgroup")
+SERIAL: Final = ("-p", "no:xdist")
+
+
+def _run_shard(
+    root: Path, selected: tuple[str, ...], distribution: tuple[str, ...] = XDIST
+) -> subprocess.CompletedProcess[str]:
+    package: Final = root / "tests" / "integration"
+    (package / "providers").mkdir(parents=True)
+    (root / "pytest.ini").write_text("[pytest]\n")
+    (root / "tests" / "__init__.py").write_text("")
+    for name in ("__init__.py", "conftest.py", "run.py", "_support"):
+        (package / name).symlink_to(INTEGRATION_PACKAGE / name)
+    for file in selected:
+        if file in MODULES:
+            (root / file).write_text(MODULES[file])
+    (root / "shard.txt").write_text("".join(f"{file}\n" for file in selected))
+    return subprocess.run(
+        (
+            sys.executable,
+            "-P",
+            "-m",
+            "pytest",
+            "tests/integration/providers",
+            "-p",
+            "no:cacheprovider",
+            *distribution,
+            "--integration-shard-count=2",
+            "--integration-shard-index=0",
+            "--integration-shard-files",
+            "shard.txt",
+        ),
+        cwd=root,
+        env={
+            "PATH": os.environ.get("PATH", ""),
+            "HOME": str(root),
+            "PYTHONPATH": os.pathsep.join((str(root), str(root / "tests"))),
+        },
+        capture_output=True,
+        text=True,
+        timeout=180,
+        check=False,
+    )
+
+
+@pytest.mark.parametrize(
+    ("distribution", "exit_code", "outcome"),
+    (
+        (XDIST, pytest.ExitCode.TESTS_FAILED, "2 passed, 1 error"),
+        (SERIAL, pytest.ExitCode.INTERRUPTED, "Interrupted: 1 error during collection"),
+    ),
+    ids=("xdist", "serial"),
+)
+def test_a_broken_import_in_a_shard_is_a_collection_error_not_a_worker_crash(
+    tmp_path: Path, distribution: tuple[str, ...], exit_code: pytest.ExitCode, outcome: str
+) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY, BROKEN), distribution)
+    assert result.returncode == exit_code, result.stdout + result.stderr
+    assert outcome in result.stdout
+    assert result.stdout.count(f"ERROR collecting {BROKEN}") == 1
+    assert "cannot import name '_removed_helper'" in result.stdout
+    assert "INTERNALERROR" not in result.stdout + result.stderr
+
+
+def test_a_shard_file_that_collects_no_tests_still_fails_as_a_collection_error(tmp_path: Path) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY, RENAMED))
+    assert result.returncode == pytest.ExitCode.TESTS_FAILED, result.stdout + result.stderr
+    assert result.stdout.count(f"ERROR collecting {RENAMED}") == 1
+    assert "2 passed" in result.stdout
+    assert "INTERNALERROR" not in result.stdout + result.stderr
+
+
+def test_a_class_level_collection_error_is_not_also_reported_as_a_file_without_tests(tmp_path: Path) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY, CLASS_ERROR), SERIAL)
+    assert result.returncode == pytest.ExitCode.INTERRUPTED, result.stdout + result.stderr
+    assert f"ERROR {CLASS_ERROR}::TestBroken" in result.stdout
+    assert "which has no tests" not in result.stdout
+
+
+def test_a_healthy_shard_runs_every_selected_test(tmp_path: Path) -> None:
+    result: Final = _run_shard(tmp_path, (HEALTHY,))
+    assert result.returncode == pytest.ExitCode.OK, result.stdout + result.stderr
+    assert "2 passed" in result.stdout