fix(bundler): only fall back to snapshot on transient catalog failures

The builtin catalog fallback previously caught every BundlerError from the
online fetch, silently masking malformed JSON, oversized responses, and
redirect/URL validation failures with the packaged snapshot. Classify only
transient failures (connection errors, timeouts, HTTP 5xx) as unavailable,
preserve validation failures, warn when a snapshot is used, and go quiet for
deliberate --offline use. Cover both builtin catalogs in unit and integration
tests, and assert the first-party catalog wheel mapping.

Assisted-by: opencode (model: deepseek-v4.1-flash, autonomous)

Author: Markus

pyproject.toml

@@ -45,7 +45,7 @@ packages = ["src/specify_cli"]
 "extensions/agent-context" = "specify_cli/core_pack/extensions/agent-context"
 "extensions/assess" = "specify_cli/core_pack/extensions/assess"
 "extensions/bug" = "specify_cli/core_pack/extensions/bug"
-# Bundled workflows (auto-installed during `specify init` or via first-party bundles)
+# Bundled workflows (`specify init` installs only `speckit`; `bugfix`/`assess` are opt-in via first-party bundles)
 "workflows/speckit" = "specify_cli/core_pack/workflows/speckit"
 "workflows/bugfix" = "specify_cli/core_pack/workflows/bugfix"
 "workflows/assess" = "specify_cli/core_pack/workflows/assess"

src/specify_cli/bundler/services/adapters.py

@@ -11,6 +11,8 @@
 from __future__ import annotations
 
 import re
+import urllib.error
+import warnings
 from pathlib import Path
 from urllib.parse import ParseResult, urlparse
 from urllib.request import url2pathname
@@ -46,6 +48,17 @@
 
 HTTP_TIMEOUT_SECONDS = 10
 
+
+class _CatalogUnavailable(BundlerError):
+    """A built-in catalog could not be reached (transport/availability failure).
+
+    Marks only transient fetch failures — connection/DNS errors, timeouts, and
+    HTTP 5xx — so the built-in catalog fallback does not swallow content or
+    security validation failures (malformed JSON, oversized or non-UTF-8
+    bodies, unsafe redirects, HTTP 4xx).
+    """
+
+
 # Windows absolute paths like ``C:\catalog.json`` parse with a single-letter
 # ``scheme`` under urlparse; treat them as local files rather than URLs.
 _WINDOWS_DRIVE_RE = re.compile(r"^[A-Za-z]:[\\/]")
@@ -142,18 +155,27 @@ def fetch(source: CatalogSource) -> dict:
             repository_url = _BUILTIN_REPOSITORY_URLS.get(url)
             if repository_url is None:
                 raise BundlerError(f"Unknown built-in catalog '{url}'.")
+            snapshot_name = _BUILTIN_PACKAGED_SNAPSHOTS[url]
             if allow_network:
                 try:
                     return _http_get_json(source.id, repository_url)
-                except BundlerError as exc:
+                except _CatalogUnavailable as exc:
                     # Built-in catalogs remain usable when the repository is
                     # temporarily unavailable; the packaged snapshot is the
-                    # authoritative offline fallback.
+                    # authoritative offline fallback. Only transient fetch
+                    # failures take this path -- content/security validation
+                    # errors propagate so a malformed response is never masked.
+                    warnings.warn(
+                        f"Built-in catalog '{url}' is unavailable ({exc}); "
+                        "using the packaged snapshot.",
+                        UserWarning,
+                        stacklevel=2,
+                    )
                     try:
-                        return _load_packaged_catalog(_BUILTIN_PACKAGED_SNAPSHOTS[url])
+                        return _load_packaged_catalog(snapshot_name)
                     except BundlerError as snapshot_exc:
                         raise snapshot_exc from exc
-            return _load_packaged_catalog(_BUILTIN_PACKAGED_SNAPSHOTS[url])
+            return _load_packaged_catalog(snapshot_name)
 
         if scheme == "file":
             path = _file_url_to_path(parsed)
@@ -212,7 +234,35 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                 label=f"bundle catalog '{source_id}'",
             ).decode("utf-8")
     except BundlerError:
+        # Size limits, redirect/URL validation: content or security failures,
+        # never transient -- must not be downgraded to availability.
         raise
+    except urllib.error.HTTPError as exc:
+        # urllib raises HTTPError for any non-2xx status; only 5xx is a
+        # transient server-side availability failure. A 4xx (404, 403, ...)
+        # is a definitive response and must surface as a hard error.
+        if exc.code >= 500:
+            raise _CatalogUnavailable(
+                f"Failed to fetch catalog from {url}: HTTP {exc.code} {exc.reason}"
+            ) from exc
+        raise BundlerError(
+            f"Failed to fetch catalog from {url}: HTTP {exc.code} {exc.reason}"
+        ) from exc
+    except urllib.error.URLError as exc:
+        raise _CatalogUnavailable(
+            f"Failed to fetch catalog from {url}: {exc.reason}"
+        ) from exc
+    except (TimeoutError, OSError) as exc:
+        # socket.timeout is TimeoutError; OSError covers connection resets and
+        # other low-level transport failures not wrapped in URLError.
+        raise _CatalogUnavailable(
+            f"Failed to fetch catalog from {url}: {exc}"
+        ) from exc
+    except UnicodeDecodeError as exc:
+        # A non-UTF-8 body is a malformed response, not an availability issue.
+        raise BundlerError(
+            f"Failed to fetch catalog from {url}: response was not valid UTF-8 ({exc})"
+        ) from exc
     except Exception as exc:
         raise BundlerError(f"Failed to fetch catalog from {url}: {exc}") from exc
     return loads_json(raw, origin=final_url)

tests/contract/test_catalog_schema.py

@@ -229,6 +229,19 @@ def test_wheel_packages_community_bundle_catalog():
     )
 
 
+def test_wheel_packages_firstparty_bundle_catalog():
+    repo_root = Path(__file__).parents[2]
+    with (repo_root / "pyproject.toml").open("rb") as pyproject_file:
+        pyproject = tomllib.load(pyproject_file)
+
+    force_include = pyproject["tool"]["hatch"]["build"]["targets"]["wheel"][
+        "force-include"
+    ]
+    assert force_include["bundles/catalog.json"] == (
+        "specify_cli/core_pack/bundles/catalog.json"
+    )
+
+
 def test_catalog_entry_rejects_string_tags():
     from specify_cli.bundler.models.catalog import CatalogEntry
 

tests/integration/test_bundler_offline.py

@@ -41,9 +41,20 @@ def test_builtin_default_catalog_resolves_first_party_bundles_offline():
     assert resolved.install_allowed is True
 
 
-def test_builtin_catalog_failure_does_not_block_lower_priority_source(monkeypatch):
+@pytest.mark.parametrize(
+    "source_id, builtin_id, builtin_priority, project_priority",
+    [
+        pytest.param("default", "builtin://default", 1, 10, id="default"),
+        pytest.param("community", "builtin://community", 20, 30, id="community"),
+    ],
+)
+def test_builtin_catalog_failure_does_not_block_lower_priority_source(
+    monkeypatch, source_id, builtin_id, builtin_priority, project_priority
+):
+    from specify_cli.bundler.services import adapters
+
     def fail_http_get_json(source_id, url):
-        raise BundlerError("repository unavailable")
+        raise adapters._CatalogUnavailable("repository unavailable")
 
     monkeypatch.setattr(
         "specify_cli.bundler.services.adapters._http_get_json", fail_http_get_json
@@ -53,7 +64,9 @@ def fail_http_get_json(source_id, url):
         lambda filename: {"schema_version": "1.0", "bundles": {}},
     )
 
-    project = _src("project", "https://example.com/catalog.json", priority=10)
+    project = _src(
+        "project", "https://example.com/catalog.json", priority=project_priority
+    )
     fetcher = make_catalog_fetcher(allow_network=True)
 
     def fetch_project(source):
@@ -65,10 +78,13 @@ def fetch_project(source):
         return fetcher(source)
 
     stack = CatalogStack(
-        [_src("default", "builtin://default"), project], fetch_project
+        [_src(source_id, builtin_id, priority=builtin_priority), project],
+        fetch_project,
     )
 
-    assert stack.resolve("company").source.id == "project"
+    with pytest.warns(UserWarning, match="packaged snapshot"):
+        resolved = stack.resolve("company")
+    assert resolved.source.id == "project"
 
 
 def test_builtin_community_catalog_resolves_from_packaged_snapshot_offline():