feat: make catalog add idempotent across all catalog families (#4505)

src/specify_cli/bundler/commands_impl/catalog_config.py

@@ -139,7 +139,7 @@ def add_source(
     policy: str,
     priority: int,
     source_id: str | None = None,
-) -> CatalogSource:
+) -> tuple[CatalogSource, str]:
     url = url.strip()
     if not url:
         raise BundlerError("A catalog url is required.")
@@ -186,21 +186,33 @@ def add_source(
     resolved_id = (source_id or _derive_id(url)).strip()
 
     catalogs = _read(project_root)
+    requested_source = CatalogSource.from_dict(
+        {
+            "id": resolved_id,
+            "url": url,
+            "priority": priority,
+            "install_policy": install_policy.value,
+        },
+        Scope.PROJECT,
+    )
     for existing in catalogs:
-        if existing.get("id") == resolved_id or existing.get("url") == url:
+        existing_id = str(existing.get("id", "")).strip()
+        existing_url = str(existing.get("url", "")).strip()
+        if existing_id == requested_source.id or existing_url == requested_source.url:
+            existing_source = CatalogSource.from_dict(existing, Scope.PROJECT)
+            if (
+                existing_source.priority == requested_source.priority
+                and existing_source.install_policy is requested_source.install_policy
+            ):
+                return existing_source, "unchanged"
             raise BundlerError(
                 f"Catalog source '{resolved_id}' (or url) already exists in this project."
             )
 
-    entry = {
-        "id": resolved_id,
-        "url": url,
-        "priority": int(priority),
-        "install_policy": install_policy.value,
-    }
+    entry = requested_source.to_dict()
     catalogs.append(entry)
     _write(project_root, catalogs)
-    return CatalogSource.from_dict(entry, Scope.PROJECT)
+    return requested_source, "added"
 
 
 def remove_source(project_root: Path, id_or_url: str) -> str:

src/specify_cli/commands/bundle/__init__.py

@@ -672,15 +672,28 @@ def catalog_add(
         project_root = require_project_root()
         from ...bundler.commands_impl.catalog_config import add_source
 
-        source = add_source(project_root, url, policy=policy, priority=priority, source_id=source_id)
+        source, status = add_source(
+            project_root,
+            url,
+            policy=policy,
+            priority=priority,
+            source_id=source_id,
+        )
     except BundlerError as exc:
         _fail(str(exc))
         return
 
-    console.print(
-        f"[green]✓[/green] Added catalog '{_escape_markup(str(source.id))}' "
-        f"(priority {source.priority}, {source.install_policy.value})."
-    )
+    safe_id = _escape_markup(str(source.id))
+    if status == "unchanged":
+        console.print(
+            f"[green]✓[/green] Catalog '{safe_id}' already configured "
+            f"(priority {source.priority}, {source.install_policy.value})."
+        )
+    else:
+        console.print(
+            f"[green]✓[/green] Added catalog '{safe_id}' "
+            f"(priority {source.priority}, {source.install_policy.value})."
+        )
 
 
 @bundle_catalog_app.command("remove")

src/specify_cli/extensions/_commands.py

@@ -640,20 +640,24 @@ def catalog_add(
     safe_name = _escape_markup(name)
     safe_url = _escape_markup(url)
 
+    entry = {
+        "name": name,
+        "url": url,
+        "priority": priority,
+        "install_allowed": install_allowed,
+        "description": description,
+    }
+
     # Check for duplicate name
     for existing in catalogs:
         if isinstance(existing, dict) and existing.get("name") == name:
+            if existing == entry:
+                return
             console.print(f"[yellow]Warning:[/yellow] A catalog named '{safe_name}' already exists.")
             console.print("Use 'specify extension catalog remove' first, or choose a different name.")
             raise typer.Exit(1)
 
-    catalogs.append({
-        "name": name,
-        "url": url,
-        "priority": priority,
-        "install_allowed": install_allowed,
-        "description": description,
-    })
+    catalogs.append(entry)
 
     config["catalogs"] = catalogs
     config_path.write_text(yaml.safe_dump(config, default_flow_style=False, sort_keys=False, allow_unicode=True), encoding="utf-8")

src/specify_cli/integrations/_query_commands.py

@@ -543,14 +543,20 @@ def integration_catalog_add(
     normalized_url = url.strip()
 
     try:
-        catalog.add_catalog(normalized_url, name)
+        status = catalog.add_catalog(normalized_url, name)
     except IntegrationCatalogError as exc:
         # Covers both URL validation (base class) and config-file validation
         # (IntegrationValidationError subclass).
         console.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(1)
 
-    console.print(f"[green]✓[/green] Catalog source added: {normalized_url}")
+    safe_url = _rich_escape(normalized_url)
+    if status == "unchanged":
+        console.print(
+            f"[green]✓[/green] Catalog source already configured: {safe_url}"
+        )
+    else:
+        console.print(f"[green]✓[/green] Catalog source added: {safe_url}")
 
 
 @integration_catalog_app.command("remove")

src/specify_cli/integrations/catalog.py

@@ -390,14 +390,14 @@ def get_project_catalog_configs(self) -> Optional[List[Dict[str, Any]]]:
             for e in entries
         ]
 
-    def add_catalog(self, url: str, name: Optional[str] = None) -> None:
+    def add_catalog(self, url: str, name: str | None = None) -> str:
         """Add a catalog source to the project-level config file.
 
         The URL is normalized (whitespace stripped) and validated before being
-        written. Duplicate URLs are rejected, including near-duplicates that
-        differ only by surrounding whitespace. Priority is derived as
-        ``max(existing) + 1`` so the new entry sorts last in the resolution
-        order unless the user edits the file manually.
+        written. An existing URL is unchanged when no name is supplied or the
+        supplied name matches; a different explicit name is rejected. Priority
+        is derived as ``max(existing) + 1`` so the new entry sorts last in the
+        resolution order unless the user edits the file manually.
         """
         url = url.strip()
         if not url:
@@ -432,6 +432,7 @@ def add_catalog(self, url: str, name: Optional[str] = None) -> None:
         # Validate each existing entry before mutating anything. Fail fast so
         # we don't silently preserve a corrupt sibling entry or derive a new
         # priority from a bogus value.
+        normalized_name = str(name).strip() if name is not None else ""
         existing_priorities: List[int] = []
         valid_catalog_count = 0
         for idx, cat in enumerate(catalogs):
@@ -452,6 +453,11 @@ def add_catalog(self, url: str, name: Optional[str] = None) -> None:
                     f"Invalid catalog entry at index {idx} in {config_path}: {exc}"
                 ) from exc
             if existing_url == url:
+                generated_name = f"catalog-{valid_catalog_count + 1}"
+                existing_name = str(cat.get("name") or generated_name).strip()
+                if not normalized_name or existing_name == normalized_name:
+                    self._load_catalog_config(config_path)
+                    return "unchanged"
                 raise IntegrationValidationError(
                     f"Catalog URL already configured: {url}"
                 )
@@ -478,9 +484,7 @@ def add_catalog(self, url: str, name: Optional[str] = None) -> None:
                 # Match `_load_catalog_config()`'s defaulting rule so the new
                 # entry still sorts after implicit-priority siblings.
                 existing_priorities.append(idx + 1)
-
         max_priority = max(existing_priorities, default=0)
-        normalized_name = str(name).strip() if name is not None else ""
         generated_name = f"catalog-{valid_catalog_count + 1}"
         catalogs.append(
             {
@@ -502,6 +506,7 @@ def add_catalog(self, url: str, name: Optional[str] = None) -> None:
                 sort_keys=False,
                 allow_unicode=True,
             )
+        return "added"
 
     def remove_catalog(self, index: int) -> str:
         """Remove a catalog source by 0-based index.

src/specify_cli/presets/_commands.py

@@ -928,20 +928,24 @@ def preset_catalog_add(
     safe_name = _escape_markup(str(name))
     safe_url = _escape_markup(str(url))
 
+    entry = {
+        "name": name,
+        "url": url,
+        "priority": priority,
+        "install_allowed": install_allowed,
+        "description": description,
+    }
+
     # Check for duplicate name
     for existing in catalogs:
         if isinstance(existing, dict) and existing.get("name") == name:
+            if existing == entry:
+                return
             console.print(f"[yellow]Warning:[/yellow] A catalog named '{safe_name}' already exists.")
             console.print("Use 'specify preset catalog remove' first, or choose a different name.")
             raise typer.Exit(1)
 
-    catalogs.append({
-        "name": name,
-        "url": url,
-        "priority": priority,
-        "install_allowed": install_allowed,
-        "description": description,
-    })
+    catalogs.append(entry)
 
     config["catalogs"] = catalogs
     config_path.write_text(yaml.safe_dump(config, default_flow_style=False, sort_keys=False, allow_unicode=True), encoding="utf-8")

src/specify_cli/workflows/_commands.py

@@ -3025,12 +3025,18 @@ def workflow_catalog_add(
     project_root = _require_specify_project()
     catalog = WorkflowCatalog(project_root)
     try:
-        catalog.add_catalog(url, name)
+        status = catalog.add_catalog(url, name)
     except WorkflowValidationError as exc:
         console.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(1)
 
-    console.print(f"[green]✓[/green] Catalog source added: {url}")
+    safe_url = _escape_markup(url)
+    if status == "unchanged":
+        console.print(
+            f"[green]✓[/green] Catalog source already configured: {safe_url}"
+        )
+    else:
+        console.print(f"[green]✓[/green] Catalog source added: {safe_url}")
 
 
 @workflow_catalog_app.command("remove")
@@ -3771,12 +3777,18 @@ def workflow_step_catalog_add(
 
     catalog = StepCatalog(project_root)
     try:
-        catalog.add_catalog(url, name)
+        status = catalog.add_catalog(url, name)
     except StepValidationError as exc:
         console.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(1)
 
-    console.print(f"[green]✓[/green] Step catalog source added: {url}")
+    safe_url = _escape_markup(url)
+    if status == "unchanged":
+        console.print(
+            f"[green]✓[/green] Step catalog source already configured: {safe_url}"
+        )
+    else:
+        console.print(f"[green]✓[/green] Step catalog source added: {safe_url}")
 
 
 @workflow_step_catalog_app.command("remove")

src/specify_cli/workflows/catalog.py

@@ -705,10 +705,11 @@ def get_catalog_configs(self) -> list[dict[str, Any]]:
             for e in entries
         ]
 
-    def add_catalog(self, url: str, name: str | None = None) -> None:
+    def add_catalog(self, url: str, name: str | None = None) -> str:
         """Add a catalog source to the project-level config."""
         self._validate_catalog_url(url)
         config_path = self.project_root / ".specify" / "workflow-catalogs.yml"
+        normalized_name = str(name).strip() if name is not None else ""
 
         data: dict[str, Any] = {"catalogs": []}
         if config_path.exists():
@@ -732,8 +733,13 @@ def add_catalog(self, url: str, name: str | None = None) -> None:
                 "Catalog config 'catalogs' must be a list."
             )
         # Check for duplicate URL (guard against non-dict entries)
-        for cat in catalogs:
+        for idx, cat in enumerate(catalogs):
             if isinstance(cat, dict) and cat.get("url") == url:
+                generated_name = f"catalog-{idx + 1}"
+                existing_name = str(cat.get("name") or generated_name).strip()
+                if not normalized_name or existing_name == normalized_name:
+                    self._load_catalog_config(config_path)
+                    return "unchanged"
                 raise WorkflowValidationError(
                     f"Catalog URL already configured: {url}"
                 )
@@ -759,7 +765,7 @@ def _coerce_priority(value: Any) -> int:
         )
         catalogs.append(
             {
-                "name": name or f"catalog-{len(catalogs) + 1}",
+                "name": normalized_name or f"catalog-{len(catalogs) + 1}",
                 "url": url,
                 "priority": max_priority + 1,
                 "install_allowed": True,
@@ -776,6 +782,7 @@ def _coerce_priority(value: Any) -> int:
             raise WorkflowValidationError(
                 f"Failed to write catalog config {config_path}: {exc}"
             ) from exc
+        return "added"
 
     def remove_catalog(self, index: int) -> str:
         """Remove a catalog source by index (0-based). Returns the removed name."""
@@ -1388,10 +1395,11 @@ def get_catalog_configs(self) -> list[dict[str, Any]]:
             for e in entries
         ]
 
-    def add_catalog(self, url: str, name: str | None = None) -> None:
+    def add_catalog(self, url: str, name: str | None = None) -> str:
         """Add a catalog source to the project-level config."""
         self._validate_catalog_url(url)
         config_path = self.project_root / ".specify" / "step-catalogs.yml"
+        normalized_name = str(name).strip() if name is not None else ""
 
         data: dict[str, Any] = {"catalogs": []}
         if config_path.exists():
@@ -1414,8 +1422,13 @@ def add_catalog(self, url: str, name: str | None = None) -> None:
             raise StepValidationError(
                 "Catalog config 'catalogs' must be a list."
             )
-        for cat in catalogs:
+        for idx, cat in enumerate(catalogs):
             if isinstance(cat, dict) and cat.get("url") == url:
+                generated_name = f"catalog-{idx + 1}"
+                existing_name = str(cat.get("name") or generated_name).strip()
+                if not normalized_name or existing_name == normalized_name:
+                    self._load_catalog_config(config_path)
+                    return "unchanged"
                 raise StepValidationError(
                     f"Catalog URL already configured: {url}"
                 )
@@ -1440,7 +1453,7 @@ def _coerce_priority(value: Any) -> int:
         )
         catalogs.append(
             {
-                "name": name or f"catalog-{len(catalogs) + 1}",
+                "name": normalized_name or f"catalog-{len(catalogs) + 1}",
                 "url": url,
                 "priority": max_priority + 1,
                 "install_allowed": True,
@@ -1459,6 +1472,7 @@ def _coerce_priority(value: Any) -> int:
             raise StepValidationError(
                 f"Failed to write catalog config {config_path}: {exc}"
             ) from exc
+        return "added"
 
     def remove_catalog(self, index: int) -> str:
         """Remove a catalog source by index (0-based). Returns the removed name."""

tests/contract/test_bundle_cli.py

@@ -203,6 +203,11 @@ def test_catalog_add_and_remove(project: Path):
         app, ["bundle", "catalog", "add", str(catalog), "--id", "local"]
     )
     assert added.exit_code == 0, added.output
+    unchanged = runner.invoke(
+        app, ["bundle", "catalog", "add", str(catalog), "--id", "local"]
+    )
+    assert unchanged.exit_code == 0, unchanged.output
+    assert "already configured" in unchanged.output
 
     listed = runner.invoke(app, ["bundle", "catalog", "list"])
     assert "local" in listed.output