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
15 changes: 13 additions & 2 deletions src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1894,11 +1894,17 @@ def get_speckit_version() -> str:
integration_app.add_typer(integration_catalog_app, name="catalog")


INTEGRATION_JSON = ".specify/integration.json"
from ._paths import INTEGRATION_JSON # noqa: E402 (re-export for backward compat)


def _read_integration_json(project_root: Path) -> dict[str, Any]:
"""Load ``.specify/integration.json``. Returns ``{}`` when missing."""
"""Load ``.specify/integration.json``. Returns ``{}`` when missing.

Shares its low-level parsing surface with
:func:`specify_cli._paths.try_read_integration_json` (used by the
workflow engine) but keeps loud, per-cause diagnostics so CLI users
get actionable messages instead of a silent fallback.
"""
path = project_root / INTEGRATION_JSON
if not path.exists():
return {}
Expand All @@ -1909,6 +1915,11 @@ def _read_integration_json(project_root: Path) -> dict[str, Any]:
console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
console.print(f"[dim]Details:[/dim] {exc}")
raise typer.Exit(1)
except UnicodeDecodeError as exc:
console.print(f"[red]Error:[/red] {path} 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"[red]Error:[/red] Could not read {path}.")
console.print(f"Please fix file permissions or delete {INTEGRATION_JSON} and retry.")
Expand Down
42 changes: 42 additions & 0 deletions src/specify_cli/_paths.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
"""Shared filesystem paths and low-level state helpers used across the
spec-kit package.

This module exists so the CLI entrypoint and the workflows engine can share
canonical paths and parsing logic without duplicating string literals (which
drift) and without importing each other (which would create a circular
import on the CLI module).
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any


# Project-relative path to the file that records which AI integration the
# project was initialized with. Read by the CLI to dispatch commands and by
# the workflow engine to resolve the ``"auto"`` integration default.
INTEGRATION_JSON = ".specify/integration.json"


def try_read_integration_json(project_root: Path) -> dict[str, Any] | None:
"""Best-effort read of ``.specify/integration.json``.

Returns the parsed mapping on success, or ``None`` when the file is
missing, unreadable, malformed JSON, non-UTF8, or not a JSON object.
Callers that want loud, user-facing error messages (e.g. the CLI's
:func:`specify_cli._read_integration_json`, which exits the process with
a diagnostic) should layer their own diagnostics on top instead of using
this helper directly.
"""
path = project_root / INTEGRATION_JSON
if not path.is_file():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError, UnicodeDecodeError):
return None
if not isinstance(data, dict):
return None
return data
81 changes: 80 additions & 1 deletion src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import yaml

from .._paths import try_read_integration_json
from .base import RunStatus, StepContext, StepResult, StepStatus


Expand Down Expand Up @@ -143,6 +144,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.")
Expand Down Expand Up @@ -715,12 +745,61 @@ 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):
Comment on lines 747 to 768
Copy link

Copilot AI May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_resolve_inputs now coerces/validates default values via _coerce_input, which is a user-visible behavior change (invalid defaults can now raise at runtime). Since validate_workflow() doesn’t currently validate defaults against declared types/enums, consider adding that validation there so workflows fail fast at install/validation time rather than during execution.

Copilot uses AI. Check for mistakes.
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 active integration key from ``.specify/integration.json``.

Returns None when the file is missing or malformed; callers are
expected to fall back to a literal default. The low-level read is
delegated to :func:`specify_cli._paths.try_read_integration_json`
so the parsing rules stay aligned with the CLI's loader.
"""
data = try_read_integration_json(self.project_root)
if data is None:
return None
value = data.get("integration")
if isinstance(value, str) and value:
return value
return None
Comment on lines +787 to +801
Copy link

Copilot AI May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_load_project_integration() duplicates the integration.json parsing logic that already exists in specify_cli.__init__._read_integration_json, but with different error handling semantics (CLI exits loudly vs. workflow engine silently falls back). To avoid drift and keep behavior consistent, consider extracting a shared low-level helper (e.g., “try-read integration.json → dict|None”) into a common module and layering the CLI/engine-specific error handling on top.

Copilot uses AI. Check for mistakes.

@staticmethod
def _coerce_input(
name: str, value: Any, input_def: dict[str, Any]
Expand Down
Loading
Loading