Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 12 additions & 26 deletions src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,13 +61,14 @@
)
from .integration_state import (
INTEGRATION_JSON,
INTEGRATION_STATE_SCHEMA,
IntegrationStateError as _IntegrationStateError,
IntegrationStateSchemaError as _IntegrationStateSchemaError,
dedupe_integration_keys as _dedupe_integration_keys,
default_integration_key as _default_integration_key,
installed_integration_keys as _installed_integration_keys,
integration_setting as _integration_setting,
integration_settings as _integration_settings,
normalize_integration_state as _normalize_integration_state,
read_integration_state as _read_integration_state,
write_integration_json as _write_integration_json_file,
)
from .shared_infra import (
Expand Down Expand Up @@ -1926,34 +1927,19 @@ def get_speckit_version() -> str:

def _read_integration_json(project_root: Path) -> dict[str, Any]:
"""Load ``.specify/integration.json``. Returns normalized state when present."""
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.")
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"[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}")
state = _read_integration_state(project_root)
except _IntegrationStateSchemaError as exc:
console.print(f"[red]Error:[/red] {exc}")
console.print("Please upgrade Spec Kit before modifying integrations.")
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__}.")
except _IntegrationStateError as exc:
console.print(f"[red]Error:[/red] {exc}")
console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
raise typer.Exit(1)
Comment on lines 1930 to 1939
schema = data.get("integration_state_schema")
if isinstance(schema, int) and not isinstance(schema, bool) and schema > INTEGRATION_STATE_SCHEMA:
console.print(
f"[red]Error:[/red] {path} uses integration state schema {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)
if state is None:
return {}
return state


def _write_integration_json(
Expand Down
109 changes: 109 additions & 0 deletions src/specify_cli/integration_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,15 @@

INTEGRATION_JSON = ".specify/integration.json"
INTEGRATION_STATE_SCHEMA = 1
SPECIFY_DIR = ".specify"


class IntegrationStateError(Exception):
"""Raised when integration.json is invalid or unreadable."""


class IntegrationStateSchemaError(IntegrationStateError):
"""Raised when integration.json uses an unsupported schema version."""


def clean_integration_key(key: Any) -> str | None:
Expand Down Expand Up @@ -159,3 +168,103 @@ def write_integration_json(
data["default_integration"] = integration_key

dest.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")


def read_integration_state(project_root: Path) -> dict[str, Any] | None:
"""Read and normalize ``.specify/integration.json``.

Returns None if the file does not exist.

Raises
------
IntegrationStateSchemaError
If the file declares a schema version newer than this CLI supports.
IntegrationStateError
If the file cannot be read, parsed, or is not a JSON object.
"""

path = project_root / INTEGRATION_JSON
if not path.exists():
return None
if not path.is_file():
raise IntegrationStateError(
f"{path} exists but is not a regular file."
)

try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError) as exc:
raise IntegrationStateError(f"Could not read {path}") from exc
except json.JSONDecodeError as exc:
raise IntegrationStateError(f"{path} contains invalid JSON") from exc
Comment on lines +197 to +199

if not isinstance(data, dict):
raise IntegrationStateError(
f"{path} must contain a JSON object, got {type(data).__name__}."
)

schema = data.get("integration_state_schema")
if (
isinstance(schema, int)
and not isinstance(schema, bool)
and schema > INTEGRATION_STATE_SCHEMA
):
raise IntegrationStateSchemaError(
f"{path} uses integration state schema {schema}, "
f"but this CLI only supports schema {INTEGRATION_STATE_SCHEMA}."
)

return normalize_integration_state(data)


def resolve_project_integration(project_root: Path) -> str:
"""Return the active integration key for a project.

Fallback chain:
- ``.specify/integration.json`` (normalized, schema-guarded)
- ``.specify/init-options.json`` (legacy keys: ``integration`` then ``ai``)
- ``"copilot"`` (hard-coded default)

Notes
-----
If ``integration.json`` exists but is unreadable/invalid or declares a future
schema version, this function raises an exception instead of silently falling
back. That keeps the engine and CLI consistent.
"""

state = read_integration_state(project_root) # raises on invalid files
if state is not None:
key = default_integration_key(state)
if key and key != "auto":
return key

init_opts_path = project_root / f"{SPECIFY_DIR}/init-options.json"
integration = _read_legacy_init_options(init_opts_path, "integration", "ai")
if integration is not None:
return integration

return "copilot"


def _read_legacy_init_options(path: Path, *keys: str) -> str | None:
"""Read a string value from a legacy init-options.json file."""

if not path.is_file():
return None

try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError):
return None

if not isinstance(data, dict):
return None

for key in keys:
value = data.get(key)
if isinstance(value, str):
cleaned = value.strip()
if cleaned and cleaned != "auto":
return cleaned

return None
Loading
Loading