feat(integrations): add a preflight_setup() hook that runs before shared writes

design/integration.md

@@ -384,6 +384,12 @@ partial and shared files are preserved by manifest-only uninstall.
 `teardown()` preserves modified tracked files by default; `--force` can remove
 them. Keep agent-specific settings and events consistent with that lifecycle.
 
+`integration install`, `switch` and `upgrade` call
+`preflight_setup(project_root, parsed_options, *, script_type)` before they
+write shared templates, scripts or integration files. Override it to reject a
+setup that would fail partway, raising `ValueError` or `OSError`; the CLI then
+exits with the message and changes nothing. The default accepts every setup.
+
 Event cleanup releases only the departing integration's dispatcher claim.
 It deletes `.specify/events.py` only when that integration's current or prior
 manifest records a produced copy and no other installed integration references

src/specify_cli/integrations/_helpers.py

@@ -246,6 +246,27 @@ def _parse_integration_options(integration: Any, raw_options: str) -> dict[str,
     return parsed or None
 
 
+def _preflight_setup_or_exit(
+    integration: Any,
+    project_root: Path,
+    parsed_options: dict[str, Any] | None,
+    script_type: str,
+    action: str,
+) -> None:
+    """Run ``integration.preflight_setup()`` and exit before any write on refusal."""
+    try:
+        integration.preflight_setup(
+            project_root, parsed_options, script_type=script_type
+        )
+    except (ValueError, OSError) as exc:
+        console.print(
+            f"[red]Error:[/red] Cannot {action} integration "
+            f"'{escape(integration.key)}': {escape(_cli_error_detail(exc))}"
+        )
+        console.print("No files were changed.")
+        raise typer.Exit(1) from exc
+
+
 def _resolve_integration_options(
     integration: Any,
     state: dict[str, Any],

src/specify_cli/integrations/base.py

@@ -998,6 +998,20 @@ def process_template(
 
         return content
 
+    def preflight_setup(
+        self,
+        project_root: Path,
+        parsed_options: dict[str, Any] | None = None,
+        *,
+        script_type: str | None = None,
+    ) -> None:
+        """Reject a planned ``setup()`` before the CLI writes shared files.
+
+        Install, switch and upgrade call this before touching shared
+        infrastructure. Raise ``ValueError`` or ``OSError`` to refuse.
+        The default accepts every setup.
+        """
+
     def setup(
         self,
         project_root: Path,

src/specify_cli/integrations/command_install.py

@@ -20,7 +20,7 @@
 )
 from ._commands import integration_app
 from ._lifecycle import external_lifecycle, lifecycle_owns_rollback, lifecycle_success
-from ._helpers import _cli_error_detail, _cli_phase_label, _get_speckit_version, _read_integration_json, _refresh_init_options_speckit_version, _remove_integration_json, _resolve_integration_options, _resolve_script_type, _update_init_options_for_integration, _write_integration_json
+from ._helpers import _cli_error_detail, _cli_phase_label, _get_speckit_version, _preflight_setup_or_exit, _read_integration_json, _refresh_init_options_speckit_version, _remove_integration_json, _resolve_integration_options, _resolve_script_type, _update_init_options_for_integration, _write_integration_json
 
 
 @integration_app.command("install")
@@ -99,6 +99,9 @@ def integration_install(
     raw_options, parsed_options = _resolve_integration_options(
         integration, current, key, integration_options
     )
+    _preflight_setup_or_exit(
+        integration, project_root, parsed_options, selected_script, "install"
+    )
 
     # Ensure shared infrastructure is present (safe to run unconditionally;
     # _install_shared_infra merges missing files without overwriting).

src/specify_cli/integrations/command_switch.py

@@ -16,7 +16,7 @@
 )
 from ._commands import integration_app
 from ._lifecycle import external_lifecycle, lifecycle_owns_rollback, lifecycle_success
-from ._helpers import _MANIFEST_READ_ERRORS, _SharedTemplateRefreshError, _clear_init_options_for_integration, _cli_error_detail, _cli_phase_label, _get_speckit_version, _read_integration_json, _register_extensions_for_agent, _register_presets_for_agent, _remove_integration_json, _resync_manifest_after_registration, _resolve_integration_options, _resolve_script_type, _set_default_integration, _set_default_integration_or_exit, _snapshot_manifest_before_registration, _unregister_extensions_for_agent, _unregister_presets_for_agent, _write_integration_json
+from ._helpers import _MANIFEST_READ_ERRORS, _SharedTemplateRefreshError, _clear_init_options_for_integration, _cli_error_detail, _cli_phase_label, _get_speckit_version, _preflight_setup_or_exit, _read_integration_json, _register_extensions_for_agent, _register_presets_for_agent, _remove_integration_json, _resync_manifest_after_registration, _resolve_integration_options, _resolve_script_type, _set_default_integration, _set_default_integration_or_exit, _snapshot_manifest_before_registration, _unregister_extensions_for_agent, _unregister_presets_for_agent, _write_integration_json
 
 
 def _missing_required_options(integration, parsed_options) -> list[str]:
@@ -160,6 +160,13 @@ def integration_switch(
         )
         console.print("No files were changed.")
         raise typer.Exit(1)
+    _preflight_setup_or_exit(
+        target_integration,
+        project_root,
+        target_parsed_options,
+        selected_script,
+        "switch to",
+    )
 
     # Phase 1: Uninstall current integration (if any)
     if installed_key:

src/specify_cli/integrations/command_upgrade.py

@@ -24,7 +24,7 @@
 )
 from ._commands import integration_app
 from ._lifecycle import external_lifecycle, lifecycle_success
-from ._helpers import _MANIFEST_READ_ERRORS, _SharedTemplateRefreshError, _cli_error_detail, _cli_phase_label, _get_speckit_version, _read_integration_json, _refresh_init_options_speckit_version, _register_extensions_for_agent, _register_presets_for_agent, _resolve_integration_options, _resolve_integration_script_type, _resync_manifest_after_registration, _unregister_enabled_extension_commands_for_agent, _update_init_options_for_integration, _write_integration_json
+from ._helpers import _MANIFEST_READ_ERRORS, _SharedTemplateRefreshError, _cli_error_detail, _cli_phase_label, _get_speckit_version, _preflight_setup_or_exit, _read_integration_json, _refresh_init_options_speckit_version, _register_extensions_for_agent, _register_presets_for_agent, _resolve_integration_options, _resolve_integration_script_type, _resync_manifest_after_registration, _unregister_enabled_extension_commands_for_agent, _update_init_options_for_integration, _write_integration_json
 
 
 @integration_app.command("upgrade")
@@ -94,6 +94,9 @@ def integration_upgrade(
     raw_options, parsed_options = _resolve_integration_options(
         integration, current, key, integration_options
     )
+    _preflight_setup_or_exit(
+        integration, project_root, parsed_options, selected_script, "upgrade"
+    )
 
     legacy_command_root_upgrade_pending = _legacy_command_root_upgrade_pending(
         integration,

src/specify_cli/shared_infra.py

@@ -464,6 +464,13 @@ def refresh_shared_templates(
             console.print(f"    {rel}")
 
 
+def script_variant_dirs(script_type: str) -> tuple[str, ...]:
+    """Return the ``scripts/`` variant directories installed for *script_type*."""
+    if script_type == "py":
+        return ("python", "powershell" if os.name == "nt" else "bash")
+    return ("bash" if script_type == "sh" else "powershell",)
+
+
 def install_shared_infra(
     project_path: Path,
     script_type: str,
@@ -515,12 +522,7 @@ def _is_managed(rel: str, dst: Path) -> bool:
     # manifest entries the core no longer ships (stale-script cleanup, #3076).
     seen_rels: set[str] = set()
     scanned_variant_dirs: set[str] = set()
-    shell_variant = "powershell" if os.name == "nt" else "bash"
-    variant_dirs = (
-        ("python", shell_variant)
-        if script_type == "py"
-        else ("bash" if script_type == "sh" else "powershell",)
-    )
+    variant_dirs = script_variant_dirs(script_type)
 
     def _decide_overwrite(rel: str, dst: Path) -> tuple[bool, str | None]:
         """Return (write, bucket) where bucket is 'skip', 'preserved', or None."""

tests/specify_cli/integrations/test_preflight_setup.py

@@ -0,0 +1,102 @@
+"""Tests for the ``IntegrationBase.preflight_setup()`` hook and its CLI wiring."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+
+import pytest
+
+from specify_cli.integrations.base import IntegrationBase
+from specify_cli.integrations.claude import ClaudeIntegration
+from specify_cli.integrations.codex import CodexIntegration
+from specify_cli.shared_infra import script_variant_dirs
+from tests.conftest import strip_ansi
+from tests.specify_cli.integrations._helpers import _init_project, _run_in_project
+
+
+def _snapshot(project):
+    """Map every file under *project* to its SHA-256, plus the directory set."""
+    files = {}
+    dirs = set()
+    for root, dirnames, filenames in os.walk(project):
+        for name in dirnames:
+            dirs.add(os.path.relpath(os.path.join(root, name), project))
+        for name in filenames:
+            path = os.path.join(root, name)
+            with open(path, "rb") as fh:
+                files[os.path.relpath(path, project)] = hashlib.sha256(fh.read()).hexdigest()
+    return files, dirs
+
+
+def _refuse(exc):
+    def preflight_setup(self, project_root, parsed_options=None, *, script_type=None):
+        raise exc
+
+    return preflight_setup
+
+
+# (command args, integration class whose hook runs, project's initial integration)
+_COMMANDS = {
+    "install": (["integration", "install", "codex", "--script", "sh", "--force"], CodexIntegration, "claude"),
+    "switch": (["integration", "switch", "codex", "--script", "sh"], CodexIntegration, "claude"),
+    "upgrade": (["integration", "upgrade", "claude", "--script", "sh", "--force"], ClaudeIntegration, "claude"),
+}
+
+
+def test_default_preflight_accepts(tmp_path):
+    assert CodexIntegration().preflight_setup(tmp_path, {}, script_type="sh") is None
+    assert IntegrationBase.preflight_setup is CodexIntegration.preflight_setup
+
+
+@pytest.mark.parametrize("command", sorted(_COMMANDS))
+@pytest.mark.parametrize("exc", [ValueError("bad config"), PermissionError("locked")])
+def test_refusal_exits_before_any_write(tmp_path, monkeypatch, command, exc):
+    args, cls, initial = _COMMANDS[command]
+    project = _init_project(tmp_path, initial)
+    # Make the shared infrastructure stale so a run past the hook would rewrite it.
+    template = next((project / ".specify" / "templates").glob("*.md"))
+    template.write_text("stale\n", encoding="utf-8")
+    before = _snapshot(project)
+
+    monkeypatch.setattr(cls, "preflight_setup", _refuse(exc))
+    result = _run_in_project(project, args)
+
+    assert result.exit_code == 1, result.output
+    output = " ".join(strip_ansi(result.output).split())
+    assert str(exc) in output
+    assert "No files were changed." in output
+    assert _snapshot(project) == before
+
+
+@pytest.mark.parametrize("command", sorted(_COMMANDS))
+def test_hook_receives_options_and_script_type(tmp_path, monkeypatch, command):
+    args, cls, initial = _COMMANDS[command]
+    project = _init_project(tmp_path, initial)
+    calls = []
+
+    def preflight_setup(self, project_root, parsed_options=None, *, script_type=None):
+        calls.append((self.key, project_root, parsed_options, script_type))
+
+    monkeypatch.setattr(cls, "preflight_setup", preflight_setup)
+    result = _run_in_project(project, [*args, "--integration-options=--events false"])
+
+    assert result.exit_code == 0, result.output
+    assert len(calls) == 1
+    key, root, parsed, script_type = calls[0]
+    assert key == cls.key
+    assert root.resolve() == project.resolve()
+    assert parsed == {"events": "false"}
+    assert script_type == "sh"
+
+
+@pytest.mark.parametrize(
+    ("script_type", "expected"),
+    [
+        ("sh", ("bash",)),
+        ("ps", ("powershell",)),
+        ("py", ("python", "powershell" if os.name == "nt" else "bash")),
+    ],
+)
+def test_script_variant_dirs(script_type, expected):
+    assert script_variant_dirs(script_type) == expected