github--spec-kit/scripts/python/common.py

"""Shared helpers for Spec Kit Python scripts."""

from __future__ import annotations

import json
import os
import re
import sys
from dataclasses import dataclass
from pathlib import Path


def _trim_trailing_separators(value: Path) -> str:
    text = str(value)
    while len(text) > 1 and text.endswith((os.sep, "/")):
        text = text[:-1]
    return text


def find_specify_root(start_dir: Path | None = None) -> Path | None:
    current = (start_dir or Path.cwd()).resolve()
    while True:
        if (current / ".specify").is_dir():
            return current
        parent = current.parent
        if parent == current:
            return None
        current = parent


def resolve_specify_init_dir() -> Path:
    raw = os.environ.get("SPECIFY_INIT_DIR", "")
    candidate = Path(raw)
    if not candidate.is_absolute():
        candidate = Path.cwd() / candidate
    try:
        init_root = candidate.resolve(strict=True)
    except OSError:
        print(
            f"ERROR: SPECIFY_INIT_DIR does not point to an existing directory: {raw}",
            file=sys.stderr,
        )
        raise SystemExit(1)
    if not init_root.is_dir():
        print(
            f"ERROR: SPECIFY_INIT_DIR does not point to an existing directory: {raw}",
            file=sys.stderr,
        )
        raise SystemExit(1)
    if not (init_root / ".specify").is_dir():
        print(
            "ERROR: SPECIFY_INIT_DIR is not a Spec Kit project "
            f"(no .specify/ directory): {init_root}",
            file=sys.stderr,
        )
        raise SystemExit(1)
    return init_root


def get_repo_root(script_file: Path | None = None) -> Path:
    if os.environ.get("SPECIFY_INIT_DIR"):
        return resolve_specify_init_dir()

    specify_root = find_specify_root()
    if specify_root is not None:
        return specify_root

    if script_file is not None:
        script_root = find_specify_root(script_file.resolve().parent)
        if script_root is not None:
            return script_root

        # Installed scripts live at .specify/scripts/python/<script>.py.
        return script_file.resolve().parents[3]
    return Path.cwd().resolve()


def get_current_branch() -> str:
    return os.environ.get("SPECIFY_FEATURE", "")


def read_feature_json_feature_directory(repo_root: Path) -> str:
    feature_json = repo_root / ".specify" / "feature.json"
    if not feature_json.is_file():
        return ""
    try:
        data = json.loads(feature_json.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError):
        return ""
    value = data.get("feature_directory") if isinstance(data, dict) else None
    return value if isinstance(value, str) else ""


def _json_dump(data: dict[str, str]) -> str:
    return json.dumps(data, ensure_ascii=False, separators=(",", ":")) + "\n"


def persist_feature_json(repo_root: Path, feature_dir_value: str) -> None:
    # Strip the repo root prefix lexically (no resolve()) to mirror the
    # Bash/PowerShell helpers: with a symlinked <repo>/specs, resolve() would
    # escape the repo and persist a machine-specific absolute path instead of
    # the relative "specs/NNN-name" the other variants store.
    value = feature_dir_value
    relative = Path(value)
    if relative.is_absolute():
        try:
            value = relative.relative_to(repo_root).as_posix()
        except ValueError:
            value = str(relative)

    current = read_feature_json_feature_directory(repo_root)
    if current == value:
        return

    specify_dir = repo_root / ".specify"
    specify_dir.mkdir(parents=True, exist_ok=True)
    (specify_dir / "feature.json").write_bytes(
        _json_dump({"feature_directory": value}).encode("utf-8")
    )


@dataclass(frozen=True)
class FeaturePaths:
    repo_root: Path
    current_branch: str
    feature_dir: Path
    feature_spec: Path
    impl_plan: Path
    tasks: Path
    research: Path
    data_model: Path
    quickstart: Path
    contracts_dir: Path


def get_feature_paths(
    *, no_persist: bool = False, script_file: Path | None = None
) -> FeaturePaths:
    repo_root = get_repo_root(script_file)
    current_branch = get_current_branch()

    feature_dir_raw = os.environ.get("SPECIFY_FEATURE_DIRECTORY", "")
    if feature_dir_raw:
        feature_dir = Path(feature_dir_raw)
        if not feature_dir.is_absolute():
            feature_dir = repo_root / feature_dir
        if not no_persist:
            persist_feature_json(repo_root, feature_dir_raw)
    elif (repo_root / ".specify" / "feature.json").is_file():
        stored = read_feature_json_feature_directory(repo_root)
        if not stored:
            print(
                "ERROR: Feature directory not found. Set SPECIFY_FEATURE_DIRECTORY "
                "or ensure .specify/feature.json contains feature_directory.",
                file=sys.stderr,
            )
            raise SystemExit(1)
        feature_dir = Path(stored)
        if not feature_dir.is_absolute():
            feature_dir = repo_root / feature_dir
    else:
        print(
            "ERROR: Feature directory not found. Set SPECIFY_FEATURE_DIRECTORY "
            "or run the specify command to create .specify/feature.json.",
            file=sys.stderr,
        )
        raise SystemExit(1)

    if not current_branch:
        current_branch = Path(_trim_trailing_separators(feature_dir)).name

    return FeaturePaths(
        repo_root=repo_root,
        current_branch=current_branch,
        feature_dir=feature_dir,
        feature_spec=feature_dir / "spec.md",
        impl_plan=feature_dir / "plan.md",
        tasks=feature_dir / "tasks.md",
        research=feature_dir / "research.md",
        data_model=feature_dir / "data-model.md",
        quickstart=feature_dir / "quickstart.md",
        contracts_dir=feature_dir / "contracts",
    )


_SAFE_COMPONENT_PATTERN = re.compile(r"[a-z0-9-]+")


def _is_safe_component(value: object) -> bool:
    return (
        isinstance(value, str)
        and _SAFE_COMPONENT_PATTERN.fullmatch(value) is not None
    )


def _normalize_priority(value: object) -> int:
    if isinstance(value, bool):
        return 10
    try:
        priority = int(value)
    except (TypeError, ValueError, OverflowError):
        return 10
    return priority if priority >= 1 else 10


def _sorted_preset_ids(presets_dir: Path) -> list[str]:
    registry = presets_dir / ".registry"
    if registry.is_file():
        # Invalid JSON or registry shapes fall back to the directory scan below.
        try:
            data = json.loads(registry.read_text(encoding="utf-8"))
            presets = data.get("presets", {})
            return [
                pid
                for pid, meta in sorted(
                    presets.items(),
                    key=lambda kv: (
                        _normalize_priority(kv[1].get("priority"))
                        if isinstance(kv[1], dict)
                        else 10,
                        kv[0],
                    ),
                )
                if (
                    _is_safe_component(pid)
                    and isinstance(meta, dict)
                    and bool(meta.get("enabled", True))
                )
            ]
        except Exception:
            pass
    try:
        return sorted(
            p.name
            for p in presets_dir.iterdir()
            if p.is_dir() and _is_safe_component(p.name)
        )
    except OSError:
        return []


def _sorted_extension_ids(extensions_dir: Path) -> list[str]:
    registry = extensions_dir / ".registry"
    registered_ids: set[str] = set()
    extensions: dict[object, object] = {}
    if os.path.lexists(registry):
        if not registry.is_file():
            raise TemplateResolutionError(
                f"Invalid extension registry {registry}: not a regular file"
            )
        try:
            data = json.loads(registry.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, json.JSONDecodeError) as exc:
            raise TemplateResolutionError(
                f"Failed to parse extension registry {registry}: {exc}"
            ) from exc
        if not isinstance(data, dict):
            raise TemplateResolutionError(
                f"Invalid extension registry {registry}: root must be a mapping"
            )
        raw_extensions = data.get("extensions", {})
        if not isinstance(raw_extensions, dict):
            raise TemplateResolutionError(
                f"Invalid extension registry {registry}: "
                "'extensions' must be a mapping"
            )
        extensions = raw_extensions
        registered_ids = {
            ext_id for ext_id in extensions if isinstance(ext_id, str)
        }

    ranked: list[tuple[int, str]] = []
    for ext_id, metadata in extensions.items():
        if (
            _is_safe_component(ext_id)
            and isinstance(metadata, dict)
            and bool(metadata.get("enabled", True))
        ):
            ranked.append((_normalize_priority(metadata.get("priority")), ext_id))

    try:
        ranked.extend(
            (10, path.name)
            for path in extensions_dir.iterdir()
            if (
                path.is_dir()
                and _is_safe_component(path.name)
                and path.name not in registered_ids
            )
        )
    except OSError:
        pass
    return [ext_id for _, ext_id in sorted(ranked)]


def _conventional_template(
    base_dir: Path, template_name: str
) -> Path | None:
    for candidate in (
        base_dir / "templates" / f"{template_name}.md",
        base_dir / f"{template_name}.md",
    ):
        if candidate.is_file():
            return candidate
    return None


def resolve_template(template_name: str, repo_root: Path) -> Path | None:
    """Resolve a template name to a file path using the priority stack.

    Order (mirrors resolve_template in scripts/bash/common.sh):
      1. .specify/templates/overrides/
      2. .specify/presets/<preset-id>/templates/ (sorted by .registry priority)
      3. .specify/extensions/<ext-id>/templates/ (hidden directories skipped)
      4. .specify/templates/ (core)
    """
    if not _is_safe_component(template_name):
        return None

    base = repo_root / ".specify" / "templates"

    override = base / "overrides" / f"{template_name}.md"
    if override.is_file():
        return override

    presets_dir = repo_root / ".specify" / "presets"
    if presets_dir.is_dir():
        for preset_id in _sorted_preset_ids(presets_dir):
            candidate = _conventional_template(
                presets_dir / preset_id, template_name
            )
            if candidate is not None:
                return candidate

    ext_dir = repo_root / ".specify" / "extensions"
    if ext_dir.is_dir():
        for extension_id in _sorted_extension_ids(ext_dir):
            ext = ext_dir / extension_id
            candidate = _conventional_template(ext, template_name)
            if candidate is not None:
                return candidate

    core = base / f"{template_name}.md"
    if core.is_file():
        return core
    return None


class TemplateResolutionError(RuntimeError):
    """Raised when template layers exist but cannot be composed safely."""


# Mirror the canonical PresetManifest contract (see src/specify_cli/presets)
# so runtime resolution rejects the same structurally malformed manifests.
_VALID_TEMPLATE_TYPES = ("template", "command", "script")
_VALID_TEMPLATE_STRATEGIES = ("replace", "prepend", "append", "wrap")
_VALID_SCRIPT_STRATEGIES = ("replace", "wrap")


def _validate_manifest_template_entry(entry: object) -> None:
    """Validate a single manifest template entry against the canonical rules."""
    if not isinstance(entry, dict):
        raise ValueError("manifest template entries must be mappings")
    if "type" not in entry or "name" not in entry or "file" not in entry:
        raise ValueError("manifest template entry missing type, name, or file")
    for field in ("type", "name", "file"):
        if not isinstance(entry[field], str):
            raise ValueError(f"manifest template {field} must be a string")
    if entry["type"] not in _VALID_TEMPLATE_TYPES:
        raise ValueError(f"invalid manifest template type '{entry['type']}'")
    strategy = entry.get("strategy", "replace")
    if not isinstance(strategy, str):
        raise ValueError("manifest template strategy must be a string")
    strategy = strategy.lower()
    if strategy not in _VALID_TEMPLATE_STRATEGIES:
        raise ValueError(f"invalid manifest template strategy '{strategy}'")
    if entry["type"] == "script" and strategy not in _VALID_SCRIPT_STRATEGIES:
        raise ValueError(
            f"invalid manifest script strategy '{strategy}'"
        )


def _preset_template_layer(
    preset_dir: Path, template_name: str
) -> tuple[Path, str] | None:
    """Return the preset template path and composition strategy."""
    manifest_path = preset_dir / "preset.yml"
    conventional = _conventional_template(preset_dir, template_name)

    try:
        import yaml
    except ImportError as exc:
        if manifest_path.is_file():
            raise TemplateResolutionError(
                "PyYAML is required to resolve preset template composition"
            ) from exc
        return (conventional, "replace") if conventional is not None else None

    if manifest_path.is_file():
        try:
            manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
            if not isinstance(manifest, dict):
                raise ValueError("manifest root must be a mapping")
            if "provides" not in manifest:
                raise ValueError("manifest missing provides section")
            provides = manifest["provides"]
            if not isinstance(provides, dict):
                raise ValueError("manifest provides must be a mapping")
            if "templates" not in provides:
                raise ValueError("manifest provides missing templates")
            templates = provides["templates"]
            if not isinstance(templates, list):
                raise ValueError("manifest templates must be a list")
            if not templates:
                raise ValueError("manifest must provide at least one template")
            for entry in templates:
                _validate_manifest_template_entry(entry)
            for entry in templates:
                if (
                    entry.get("name") != template_name
                    or entry.get("type", "template") != "template"
                ):
                    continue
                file_value = entry.get("file", "")
                strategy = entry.get("strategy", "replace")
                relative = Path(file_value)
                if (
                    not relative
                    or relative.is_absolute()
                    or ".." in relative.parts
                ):
                    return None
                candidate = preset_dir / relative
                if not candidate.is_file():
                    return None
                return candidate, strategy.lower()
        except (OSError, UnicodeError, ValueError, yaml.YAMLError) as exc:
            raise TemplateResolutionError(
                f"Failed to parse preset manifest {manifest_path}: {exc}"
            ) from exc

    return (conventional, "replace") if conventional is not None else None


def resolve_template_content(template_name: str, repo_root: Path) -> str | None:
    """Resolve and compose template content through the project layer stack."""
    if not _is_safe_component(template_name):
        return None

    layers: list[tuple[Path, str]] = []

    def compose_from_base() -> str:
        try:
            content = layers[-1][0].read_bytes().decode("utf-8")
            for path, strategy in reversed(layers[:-1]):
                layer_content = path.read_bytes().decode("utf-8")
                if strategy == "prepend":
                    content = f"{layer_content}\n\n{content}"
                elif strategy == "append":
                    content = f"{content}\n\n{layer_content}"
                elif strategy == "wrap":
                    placeholder = "{CORE_TEMPLATE}"
                    if placeholder not in layer_content:
                        raise TemplateResolutionError(
                            f"Wrap layer {path} is missing {placeholder}"
                        )
                    content = layer_content.replace(placeholder, content)
                else:
                    raise TemplateResolutionError(
                        f"Unknown template composition strategy '{strategy}' in {path}"
                    )
        except (OSError, UnicodeError) as exc:
            raise TemplateResolutionError(
                f"Failed to read template layer for '{template_name}': {exc}"
            ) from exc
        return content

    override = (
        repo_root
        / ".specify"
        / "templates"
        / "overrides"
        / f"{template_name}.md"
    )
    if override.is_file():
        layers.append((override, "replace"))
        return compose_from_base()

    presets_dir = repo_root / ".specify" / "presets"
    for preset_id in _sorted_preset_ids(presets_dir):
        layer = _preset_template_layer(presets_dir / preset_id, template_name)
        if layer is not None:
            layers.append(layer)
            if layer[1] == "replace":
                return compose_from_base()

    extensions_dir = repo_root / ".specify" / "extensions"
    for extension_id in _sorted_extension_ids(extensions_dir):
        extension_dir = extensions_dir / extension_id
        candidate = _conventional_template(extension_dir, template_name)
        if candidate is not None:
            layers.append((candidate, "replace"))
            return compose_from_base()

    core = repo_root / ".specify" / "templates" / f"{template_name}.md"
    if core.is_file():
        layers.append((core, "replace"))
        return compose_from_base()

    if not layers:
        return None

    raise TemplateResolutionError(
        f"Template '{template_name}' has composing layers but no replace base"
    )


def get_invoke_separator(repo_root: Path) -> str:
    integration_json = repo_root / ".specify" / "integration.json"
    if not integration_json.is_file():
        return "."
    # Split the parse out of the lookup and guard the top-level shape, matching
    # read_feature_json_feature_directory above and the bash/PowerShell twins,
    # which both fall back to "." for any unusable integration.json:
    #   * a non-mapping top level ([], "forge", 42, null) is valid JSON, so
    #     json.JSONDecodeError never fires and state.get(...) raised
    #     AttributeError;
    #   * a non-UTF-8 file raises UnicodeDecodeError, which is a ValueError --
    #     not an OSError -- so it escaped the except tuple. Realistic on
    #     Windows, where PowerShell 5.1's Out-File/`>` default to UTF-16.
    try:
        state = json.loads(integration_json.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError):
        return "."
    if not isinstance(state, dict):
        return "."
    key = state.get("default_integration") or state.get("integration") or ""
    settings = state.get("integration_settings")
    if isinstance(key, str) and isinstance(settings, dict):
        entry = settings.get(key)
        if isinstance(entry, dict) and entry.get("invoke_separator") in {".", "-"}:
            return entry["invoke_separator"]
    return "."


def format_speckit_command(command_name: str, repo_root: Path) -> str:
    separator = get_invoke_separator(repo_root)
    name = command_name.lstrip("/")
    if name.startswith("speckit."):
        name = name[len("speckit.") :]
    elif name.startswith("speckit-"):
        name = name[len("speckit-") :]
    name = name.replace(".", separator)
    return f"/speckit{separator}{name}"