fix(workflows): remove the catalog source shown at the listed index

docs/reference/workflows.md

@@ -451,9 +451,11 @@ Re-adding the same workflow or step catalog URL with the same name succeeds with
 
 ```bash
 specify workflow catalog remove <index>
+specify workflow step catalog remove <index>
 ```
 
-Removes a catalog by its index in the catalog list.
+Removes a project catalog by its index in the corresponding priority-ordered `catalog list`.
+Sources supplied by `SPECKIT_WORKFLOW_CATALOG_URL` or `SPECKIT_STEP_CATALOG_URL` cannot be removed this way; unset the variable to manage project sources.
 
 ### Catalog Resolution Order
 

src/specify_cli/workflows/catalog/_domain.py

@@ -349,6 +349,15 @@ def _load_catalog_config(
             raise WorkflowValidationError(
                 f"Failed to read catalog config {config_path}: {exc}"
             ) from exc
+        indexed_entries = self._parse_catalog_config(data, config_path)
+        if indexed_entries is None:
+            return None
+        return [entry for _, entry in indexed_entries]
+
+    def _parse_catalog_config(
+        self, data: Any, config_path: Path
+    ) -> list[tuple[int, WorkflowCatalogEntry]] | None:
+        """Validate and order sources, retaining their original YAML positions."""
         # An empty document (or explicit ``null``) parses to None -> this config
         # layer contributes nothing, so ``get_active_catalogs`` moves on to the
         # next layer (this loader serves both the project and user configs;
@@ -384,7 +393,7 @@ def _load_catalog_config(
             # is valid — fall back to built-in defaults.
             return None
 
-        entries: list[WorkflowCatalogEntry] = []
+        entries: list[tuple[int, WorkflowCatalogEntry]] = []
         for idx, item in enumerate(catalogs_data):
             if not isinstance(item, dict):
                 raise WorkflowValidationError(
@@ -424,15 +433,18 @@ def _load_catalog_config(
             else:
                 install_allowed = bool(raw_install)
             entries.append(
-                WorkflowCatalogEntry(
-                    url=url,
-                    name=str(item.get("name", f"catalog-{idx + 1}")),
-                    priority=priority,
-                    install_allowed=install_allowed,
-                    description=str(item.get("description", "")),
+                (
+                    idx,
+                    WorkflowCatalogEntry(
+                        url=url,
+                        name=str(item.get("name", f"catalog-{idx + 1}")),
+                        priority=priority,
+                        install_allowed=install_allowed,
+                        description=str(item.get("description", "")),
+                    ),
                 )
             )
-        entries.sort(key=lambda e: e.priority)
+        entries.sort(key=lambda item: item[1].priority)
         if not entries:
             raise WorkflowValidationError(
                 f"Catalog config {config_path} contains {len(catalogs_data)} "
@@ -821,7 +833,12 @@ def _coerce_priority(value: Any) -> int:
         return "added"
 
     def remove_catalog(self, index: int) -> str:
-        """Remove a catalog source by index (0-based). Returns the removed name."""
+        """Remove a project source by its index in the active catalog list."""
+        if os.environ.get("SPECKIT_WORKFLOW_CATALOG_URL", "").strip():
+            raise WorkflowValidationError(
+                "The active catalog comes from SPECKIT_WORKFLOW_CATALOG_URL. "
+                "Unset that variable to remove a project catalog source."
+            )
         config_path = self.project_root / ".specify" / "workflow-catalogs.yml"
         if not config_path.exists():
             raise WorkflowValidationError("No catalog config file found.")
@@ -832,25 +849,15 @@ def remove_catalog(self, index: int) -> str:
             raise WorkflowValidationError(
                 f"Catalog config file is unreadable or malformed: {exc}"
             ) from exc
-        if data is None:
-            data = {}
-        elif not isinstance(data, dict):
-            raise WorkflowValidationError(
-                "Catalog config file is corrupted (expected a mapping)."
-            )
-        catalogs = data.get("catalogs", [])
-        if not isinstance(catalogs, list):
-            raise WorkflowValidationError(
-                "Catalog config 'catalogs' must be a list."
-            )
+        entries = self._parse_catalog_config(data, config_path) or []
 
-        if index < 0 or index >= len(catalogs):
+        if index < 0 or index >= len(entries):
             raise WorkflowValidationError(
-                f"Catalog index {index} out of range (0-{len(catalogs) - 1})."
+                f"Catalog index {index} out of range (0-{len(entries) - 1})."
             )
 
-        removed = catalogs.pop(index)
-        data["catalogs"] = catalogs
+        source_index, entry = entries[index]
+        data["catalogs"].pop(source_index)
 
         try:
             with open(config_path, "w", encoding="utf-8") as f:
@@ -860,6 +867,4 @@ def remove_catalog(self, index: int) -> str:
                 f"Failed to write catalog config {config_path}: {exc}"
             ) from exc
 
-        if isinstance(removed, dict):
-            return removed.get("name", f"catalog-{index + 1}")
-        return f"catalog-{index + 1}"
+        return entry.name

src/specify_cli/workflows/step/catalog/_domain.py

@@ -324,6 +324,15 @@ def _load_catalog_config(
             raise StepValidationError(
                 f"Failed to read catalog config {config_path}: {exc}"
             ) from exc
+        indexed_entries = self._parse_catalog_config(data, config_path)
+        if indexed_entries is None:
+            return None
+        return [entry for _, entry in indexed_entries]
+
+    def _parse_catalog_config(
+        self, data: Any, config_path: Path
+    ) -> list[tuple[int, StepCatalogEntry]] | None:
+        """Validate and order sources, retaining their original YAML positions."""
         # Same two guards as WorkflowCatalog._load_catalog_config above, kept in
         # lockstep: this is the step-catalog twin of that loader and read the
         # same way. Dropping ``or {}`` stops a falsy non-mapping top level from
@@ -347,7 +356,7 @@ def _load_catalog_config(
         if not catalogs_data:
             return None
 
-        entries: list[StepCatalogEntry] = []
+        entries: list[tuple[int, StepCatalogEntry]] = []
         for idx, item in enumerate(catalogs_data):
             if not isinstance(item, dict):
                 raise StepValidationError(
@@ -386,15 +395,18 @@ def _load_catalog_config(
             else:
                 install_allowed = bool(raw_install)
             entries.append(
-                StepCatalogEntry(
-                    url=url,
-                    name=str(item.get("name", f"catalog-{idx + 1}")),
-                    priority=priority,
-                    install_allowed=install_allowed,
-                    description=str(item.get("description", "")),
+                (
+                    idx,
+                    StepCatalogEntry(
+                        url=url,
+                        name=str(item.get("name", f"catalog-{idx + 1}")),
+                        priority=priority,
+                        install_allowed=install_allowed,
+                        description=str(item.get("description", "")),
+                    ),
                 )
             )
-        entries.sort(key=lambda e: e.priority)
+        entries.sort(key=lambda item: item[1].priority)
         if not entries:
             raise StepValidationError(
                 f"Catalog config {config_path} contains {len(catalogs_data)} "
@@ -770,7 +782,12 @@ def _coerce_priority(value: Any) -> int:
         return "added"
 
     def remove_catalog(self, index: int) -> str:
-        """Remove a catalog source by index (0-based). Returns the removed name."""
+        """Remove a project source by its index in the active catalog list."""
+        if os.environ.get("SPECKIT_STEP_CATALOG_URL", "").strip():
+            raise StepValidationError(
+                "The active catalog comes from SPECKIT_STEP_CATALOG_URL. "
+                "Unset that variable to remove a project catalog source."
+            )
         config_path = self.project_root / ".specify" / "step-catalogs.yml"
         if not config_path.exists():
             raise StepValidationError("No step catalog config file found.")
@@ -781,25 +798,15 @@ def remove_catalog(self, index: int) -> str:
             raise StepValidationError(
                 f"Catalog config file is unreadable or malformed: {exc}"
             ) from exc
-        if data is None:
-            data = {}
-        elif not isinstance(data, dict):
-            raise StepValidationError(
-                "Catalog config file is corrupted (expected a mapping)."
-            )
-        catalogs = data.get("catalogs", [])
-        if not isinstance(catalogs, list):
-            raise StepValidationError(
-                "Catalog config 'catalogs' must be a list."
-            )
+        entries = self._parse_catalog_config(data, config_path) or []
 
-        if index < 0 or index >= len(catalogs):
+        if index < 0 or index >= len(entries):
             raise StepValidationError(
-                f"Catalog index {index} out of range (0-{len(catalogs) - 1})."
+                f"Catalog index {index} out of range (0-{len(entries) - 1})."
             )
 
-        removed = catalogs.pop(index)
-        data["catalogs"] = catalogs
+        source_index, entry = entries[index]
+        data["catalogs"].pop(source_index)
 
         try:
             with open(config_path, "w", encoding="utf-8") as f:
@@ -811,6 +818,4 @@ def remove_catalog(self, index: int) -> str:
                 f"Failed to write catalog config {config_path}: {exc}"
             ) from exc
 
-        if isinstance(removed, dict):
-            return removed.get("name", f"catalog-{index + 1}")
-        return f"catalog-{index + 1}"
+        return entry.name

tests/specify_cli/workflows/catalog/test_command_remove.py

@@ -1,6 +1,7 @@
 """Tests for ``specify workflow catalog remove``."""
 
 from typer.testing import CliRunner
+import yaml
 
 from specify_cli import app
 from specify_cli.workflows.catalog import WorkflowCatalog
@@ -17,3 +18,47 @@ def test_workflow_catalog_remove_deletes_selected_source(project_dir, monkeypatc
     assert all(
         config["name"] != "local" for config in catalog.get_catalog_configs()
     )
+
+
+def test_workflow_catalog_remove_uses_listed_index(project_dir, monkeypatch):
+    monkeypatch.chdir(project_dir)
+    monkeypatch.delenv("SPECKIT_WORKFLOW_CATALOG_URL", raising=False)
+    config = project_dir / ".specify" / "workflow-catalogs.yml"
+    data = {"notes": "keep", "catalogs": [
+        {"name": "team", "url": "https://example.com/team.json", "priority": 10},
+        {"name": "review", "url": "https://example.com/review.json", "priority": 1},
+    ]}
+    config.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
+    runner = CliRunner()
+    listed = runner.invoke(app, ["workflow", "catalog", "list"])
+    assert listed.exit_code == 0, listed.output
+    assert "[0] review" in listed.output
+
+    result = runner.invoke(app, ["workflow", "catalog", "remove", "0"])
+
+    assert result.exit_code == 0, result.output
+    assert "'review' removed" in result.output
+    data["catalogs"].pop(1)
+    assert yaml.safe_load(config.read_text(encoding="utf-8")) == data
+
+
+def test_workflow_catalog_remove_refuses_environment_source(project_dir, monkeypatch):
+    monkeypatch.chdir(project_dir)
+    config = project_dir / ".specify" / "workflow-catalogs.yml"
+    config.write_text(
+        "# keep\ncatalogs:\n  - name: project\n"
+        "    url: https://example.com/project.json\n", encoding="utf-8",
+    )
+    before = config.read_bytes()
+    monkeypatch.setenv("SPECKIT_WORKFLOW_CATALOG_URL", "https://example.com/override.json")
+    runner = CliRunner()
+    listed = runner.invoke(app, ["workflow", "catalog", "list"])
+    assert listed.exit_code == 0, listed.output
+    assert "[0] env-override" in listed.output
+
+    result = runner.invoke(app, ["workflow", "catalog", "remove", "0"])
+
+    assert result.exit_code == 1, result.output
+    assert "SPECKIT_WORKFLOW_CATALOG_URL" in result.output
+    assert "Unset" in result.output
+    assert config.read_bytes() == before

tests/specify_cli/workflows/step/catalog/test_command_remove.py

@@ -1,6 +1,7 @@
 """Tests for ``specify workflow step catalog remove``."""
 
 from typer.testing import CliRunner
+import yaml
 
 from specify_cli import app
 from specify_cli.workflows.step.catalog import StepCatalog
@@ -21,3 +22,47 @@ def test_workflow_step_catalog_remove_deletes_selected_source(
     assert all(
         config["name"] != "local" for config in catalog.get_catalog_configs()
     )
+
+
+def test_workflow_step_catalog_remove_uses_listed_index(project_dir, monkeypatch):
+    monkeypatch.chdir(project_dir)
+    monkeypatch.delenv("SPECKIT_STEP_CATALOG_URL", raising=False)
+    config = project_dir / ".specify" / "step-catalogs.yml"
+    data = {"notes": "keep", "catalogs": [
+        {"name": "team", "url": "https://example.com/team.json", "priority": 10},
+        {"name": "review", "url": "https://example.com/review.json", "priority": 1},
+    ]}
+    config.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8")
+    runner = CliRunner()
+    listed = runner.invoke(app, ["workflow", "step", "catalog", "list"])
+    assert listed.exit_code == 0, listed.output
+    assert "[0] review" in listed.output
+
+    result = runner.invoke(app, ["workflow", "step", "catalog", "remove", "0"])
+
+    assert result.exit_code == 0, result.output
+    assert "'review' removed" in result.output
+    data["catalogs"].pop(1)
+    assert yaml.safe_load(config.read_text(encoding="utf-8")) == data
+
+
+def test_workflow_step_catalog_remove_refuses_environment_source(project_dir, monkeypatch):
+    monkeypatch.chdir(project_dir)
+    config = project_dir / ".specify" / "step-catalogs.yml"
+    config.write_text(
+        "# keep\ncatalogs:\n  - name: project\n"
+        "    url: https://example.com/project.json\n", encoding="utf-8",
+    )
+    before = config.read_bytes()
+    monkeypatch.setenv("SPECKIT_STEP_CATALOG_URL", "https://example.com/override.json")
+    runner = CliRunner()
+    listed = runner.invoke(app, ["workflow", "step", "catalog", "list"])
+    assert listed.exit_code == 0, listed.output
+    assert "[0] env-override" in listed.output
+
+    result = runner.invoke(app, ["workflow", "step", "catalog", "remove", "0"])
+
+    assert result.exit_code == 1, result.output
+    assert "SPECKIT_STEP_CATALOG_URL" in result.output
+    assert "Unset" in result.output
+    assert config.read_bytes() == before