fix(workflow): support integration: auto to follow project's initialized AI

src/specify_cli/__init__.py

@@ -68,6 +68,7 @@
     integration_setting as _integration_setting,
     integration_settings as _integration_settings,
     normalize_integration_state as _normalize_integration_state,
+    try_read_integration_json as _try_read_integration_json,
     write_integration_json as _write_integration_json_file,
 )
 from .shared_infra import (
@@ -1952,35 +1953,37 @@ def get_speckit_version() -> str:
 
 
 def _read_integration_json(project_root: Path) -> dict[str, Any]:
-    """Load ``.specify/integration.json``. Returns normalized state when present."""
+    """Load ``.specify/integration.json``. Returns normalized state when present.
+
+    Delegates the parse / schema-guard logic to the shared
+    :func:`_try_read_integration_json` helper so the CLI and workflow engine
+    cannot drift on validation rules. Each error variant is translated into
+    the existing loud-fail UX (console message + ``typer.Exit(1)``).
+    """
     path = project_root / INTEGRATION_JSON
-    if not path.exists():
-        return {}
-    try:
-        data = json.loads(path.read_text(encoding="utf-8"))
-    except json.JSONDecodeError as exc:
-        console.print(f"[red]Error:[/red] {path} contains invalid JSON.")
+    state, error = _try_read_integration_json(project_root)
+    if error is None:
+        return state or {}
+    if error.kind == "decode":
+        console.print(f"[red]Error:[/red] {path} contains invalid JSON or is not valid UTF-8.")
         console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
-        console.print(f"[dim]Details:[/dim] {exc}")
-        raise typer.Exit(1)
-    except OSError as exc:
+        console.print(f"[dim]Details:[/dim] {error.detail}")
+    elif error.kind == "os":
         console.print(f"[red]Error:[/red] Could not read {path}.")
         console.print(f"Please fix file permissions or delete {INTEGRATION_JSON} and retry.")
-        console.print(f"[dim]Details:[/dim] {exc}")
-        raise typer.Exit(1)
-    if not isinstance(data, dict):
-        console.print(f"[red]Error:[/red] {path} must contain a JSON object, got {type(data).__name__}.")
+        console.print(f"[dim]Details:[/dim] {error.detail}")
+    elif error.kind == "not_object":
+        console.print(
+            f"[red]Error:[/red] {path} must contain a JSON object, got {error.detail}."
+        )
         console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
-        raise typer.Exit(1)
-    schema = data.get("integration_state_schema")
-    if isinstance(schema, int) and not isinstance(schema, bool) and schema > INTEGRATION_STATE_SCHEMA:
+    elif error.kind == "schema_too_new":
         console.print(
-            f"[red]Error:[/red] {path} uses integration state schema {schema}, "
+            f"[red]Error:[/red] {path} uses integration state schema {error.schema}, "
             f"but this CLI only supports schema {INTEGRATION_STATE_SCHEMA}."
         )
         console.print("Please upgrade Spec Kit before modifying integrations.")
-        raise typer.Exit(1)
-    return _normalize_integration_state(data)
+    raise typer.Exit(1)
 
 
 def _write_integration_json(

src/specify_cli/integration_state.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -11,6 +12,56 @@
 INTEGRATION_STATE_SCHEMA = 1
 
 
+@dataclass(frozen=True)
+class IntegrationReadError:
+    """Structured failure from :func:`try_read_integration_json`.
+
+    Callers map ``kind`` to whatever surface they need (loud CLI error,
+    silent fallback, etc.) without re-implementing the parse/validation logic.
+    """
+
+    kind: str  # "decode", "os", "not_object", "schema_too_new"
+    detail: str = ""
+    schema: int | None = None
+
+
+def try_read_integration_json(
+    project_root: Path,
+) -> tuple[dict[str, Any] | None, IntegrationReadError | None]:
+    """Parse ``.specify/integration.json`` without raising.
+
+    Returns ``(normalized_state, None)`` on success, ``(None, None)`` when the
+    file does not exist, or ``(None, error)`` for any parse / validation
+    failure. This is the single low-level reader; both the CLI's loud
+    ``_read_integration_json`` and the workflow engine's silent
+    ``_load_project_integration`` consume it so the schema guard and parse
+    logic cannot drift between them.
+    """
+    path = project_root / INTEGRATION_JSON
+    if not path.is_file():
+        return None, None
+    try:
+        raw = path.read_text(encoding="utf-8")
+    except UnicodeDecodeError as exc:
+        return None, IntegrationReadError(kind="decode", detail=str(exc))
+    except OSError as exc:
+        return None, IntegrationReadError(kind="os", detail=str(exc))
+    try:
+        data = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        return None, IntegrationReadError(kind="decode", detail=str(exc))
+    if not isinstance(data, dict):
+        return None, IntegrationReadError(kind="not_object", detail=type(data).__name__)
+    schema = data.get("integration_state_schema")
+    if (
+        isinstance(schema, int)
+        and not isinstance(schema, bool)
+        and schema > INTEGRATION_STATE_SCHEMA
+    ):
+        return None, IntegrationReadError(kind="schema_too_new", schema=schema)
+    return normalize_integration_state(data), None
+
+
 def clean_integration_key(key: Any) -> str | None:
     """Return a stripped integration key, or None for empty/non-string values."""
     if not isinstance(key, str) or not key.strip():

src/specify_cli/workflows/engine.py

@@ -19,6 +19,10 @@
 
 import yaml
 
+from ..integration_state import (
+    default_integration_key,
+    try_read_integration_json,
+)
 from .base import RunStatus, StepContext, StepResult, StepStatus
 
 
@@ -143,6 +147,35 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
                     f"Must be 'string', 'number', or 'boolean'."
                 )
 
+            # Validate the default eagerly so authoring mistakes (e.g. a
+            # default not in the declared enum, or a non-numeric default for
+            # a number input) surface at install/validation time instead of
+            # at workflow-execution time. ``"auto"`` for the integration
+            # input is a runtime-resolved sentinel, so only the
+            # enum-membership check is exempted for that exact case — the
+            # declared type is still enforced (e.g. ``type: number`` paired
+            # with ``default: "auto"`` is still rejected).
+            if "default" in input_def:
+                default_value = input_def["default"]
+                is_auto_integration = (
+                    input_name == "integration" and default_value == "auto"
+                )
+                validation_input_def: dict[str, Any] = input_def
+                if is_auto_integration and "enum" in input_def:
+                    validation_input_def = {
+                        key: value
+                        for key, value in input_def.items()
+                        if key != "enum"
+                    }
+                try:
+                    WorkflowEngine._coerce_input(
+                        input_name, default_value, validation_input_def
+                    )
+                except ValueError as exc:
+                    errors.append(
+                        f"Input {input_name!r} has invalid default: {exc}"
+                    )
+
     # -- Steps ------------------------------------------------------------
     if not isinstance(definition.steps, list):
         errors.append("'steps' must be a list.")
@@ -715,12 +748,59 @@ def _resolve_inputs(
                     name, provided[name], input_def
                 )
             elif "default" in input_def:
-                resolved[name] = input_def["default"]
+                default_value = self._resolve_default(name, input_def["default"])
+                # If the integration default could not be resolved against
+                # project state and falls back to the literal ``"auto"``
+                # sentinel, exempt it from enum-membership coercion so a
+                # workflow that lists specific integrations in ``enum`` does
+                # not crash at runtime — declared type is still enforced.
+                coerce_input_def = input_def
+                if (
+                    name == "integration"
+                    and default_value == "auto"
+                    and "enum" in input_def
+                ):
+                    coerce_input_def = {
+                        key: value
+                        for key, value in input_def.items()
+                        if key != "enum"
+                    }
+                resolved[name] = self._coerce_input(
+                    name, default_value, coerce_input_def
+                )
             elif input_def.get("required", False):
                 msg = f"Required input {name!r} not provided."
                 raise ValueError(msg)
         return resolved
 
+    def _resolve_default(self, name: str, default: Any) -> Any:
+        """Resolve special default sentinels against project state.
+
+        For the ``integration`` input, ``"auto"`` resolves to the integration
+        recorded in ``.specify/integration.json`` so workflows dispatch to the
+        AI the project was actually initialized with, instead of a hardcoded
+        value baked into the workflow YAML.
+        """
+        if name == "integration" and default == "auto":
+            resolved = self._load_project_integration()
+            if resolved is not None:
+                return resolved
+        return default
+
+    def _load_project_integration(self) -> str | None:
+        """Read the default integration key from ``.specify/integration.json``.
+
+        Delegates parsing and schema validation to
+        :func:`try_read_integration_json` — the same low-level helper used by
+        the CLI — so the engine cannot drift from CLI behavior on the parse
+        path. Returns ``None`` when the file is missing, malformed, or
+        written by a newer CLI; callers fall back to the literal default.
+        """
+        state, error = try_read_integration_json(self.project_root)
+        if state is None or error is not None:
+            return None
+        return default_integration_key(state)
+
     @staticmethod
     def _coerce_input(
         name: str, value: Any, input_def: dict[str, Any]

tests/integrations/test_cli.py

@@ -1219,6 +1219,30 @@ def fail_search(self, **kwargs):
         assert "contains invalid JSON" in normalized_output
         assert "integration.json" in normalized_output
 
+    def test_search_rejects_non_utf8_integration_json_before_catalog_lookup(
+        self, tmp_path, monkeypatch
+    ):
+        """A non-UTF8 ``integration.json`` must surface a clear error and
+        avoid falling through to the catalog lookup, mirroring the malformed-JSON
+        case but for the ``UnicodeDecodeError`` branch in ``_read_integration_json``."""
+        project = self._make_project(tmp_path)
+        # 0xFF is invalid as the leading byte of any UTF-8 sequence, so
+        # ``Path.read_text(encoding="utf-8")`` raises ``UnicodeDecodeError``.
+        (project / ".specify" / "integration.json").write_bytes(b"\xff\xfe\x00\x00")
+
+        from specify_cli.integrations.catalog import IntegrationCatalog
+
+        def fail_search(self, **kwargs):
+            raise AssertionError("catalog search should not be called")
+
+        monkeypatch.setattr(IntegrationCatalog, "search", fail_search)
+
+        result = self._invoke(["integration", "search"], project)
+        normalized_output = _normalize_cli_output(result.output)
+        assert result.exit_code == 1
+        assert "not valid UTF-8" in normalized_output
+        assert "integration.json" in normalized_output
+
     def test_search_filters_by_tag(self, tmp_path, monkeypatch):
         project = self._make_project(tmp_path)
         self._patch_catalog(monkeypatch)