diff --git a/src/specify_cli/events/__init__.py b/src/specify_cli/events/__init__.py index b78e5da841..5fe6c14bc7 100644 --- a/src/specify_cli/events/__init__.py +++ b/src/specify_cli/events/__init__.py @@ -1228,17 +1228,6 @@ def _shell_quote(value: str, target_os: str) -> str: return shlex.quote(value) -def _vibe_target_os() -> str: - """Quoting target for Vibe hook commands. - - Vibe launches hooks with ``asyncio.create_subprocess_shell`` — the host's - native shell: POSIX ``sh`` on Unix, ``cmd.exe`` (%COMSPEC%) on Windows, - where POSIX single-quoting is not quoting at all and an interpreter or - dispatcher path containing spaces would split. - """ - return "cmd" if os.name == "nt" else "host" - - def _dispatcher_command( integration: IntegrationBase, project_root: Path, @@ -1502,49 +1491,26 @@ def install_integration_events( created.append(config_path) elif fmt == "toml-vibe": - # Vibe hooks.toml custom merge. Flat [[hooks]] array; Vibe's - # HookConfig schema is name/type/command/match/timeout, with type - # limited to "pre_tool" | "post_tool" | "post_agent". Hook names must - # be unique (Vibe silently drops duplicates by name), so a per-file - # counter suffix disambiguates handlers whose commands share a final - # segment (e.g. speckit.a.validate vs speckit.b.validate). - lines: list[str] = [] - used_names: set[str] = set() - for ev, handlers in filtered.items(): - native = canonical_to_native[ev] - for cfg in handlers: - command = cfg.get("command", "") - dispatcher_cmd = _dispatcher_command( - integration, project_root, command, ev, - target_os=_vibe_target_os(), - timeout_seconds=cfg.get("timeout", 60), - ) - command_stem = command.split('.')[-1] if command else "unknown" - command_stem = re.sub(r'[^A-Za-z0-9_-]+', '-', command_stem) or "unknown" - base_name = f"speckit-{native}-{command_stem}" - hook_name = base_name - suffix = 2 - while hook_name in used_names: - hook_name = f"{base_name}-{suffix}" - suffix += 1 - used_names.add(hook_name) - lines.append("[[hooks]]") - lines.append(f'name = {_toml_quote(hook_name)}') - lines.append(f'type = {_toml_quote(native)}') - # Vibe's field is `match` (fnmatch glob, or `re:`-prefixed - # regex, case-insensitive) and it is only valid on tool - # hooks — HookConfig rejects `match` on post_agent. Canonical - # matchers are Claude-style regexes ("Edit|Write"), so - # non-wildcard matchers are emitted as `re:` patterns. - matcher = cfg.get("matcher", "*") - if matcher and matcher != "*" and native in ("pre_tool", "post_tool"): - lines.append(f'match = {_toml_quote("re:" + matcher)}') - lines.append(f'command = {_toml_quote(dispatcher_cmd)}') - lines.append(f'timeout = {_native_timeout(integration, cfg.get("timeout", 60) + EVENT_TIMEOUT_BUFFER)}') - lines.append('speckit_marker = true') - lines.append('') - # S5: only track when the merge wrote (skips on unreadable file). - if _merge_vibe_toml_fragment(config_path, "\n".join(lines)): + # Vibe owns its native TOML schema and its managed-entry boundaries. + # Shared events deliberately supplies only dispatcher construction and + # manifest handling, keeping this a narrow integration-specific seam. + merge_vibe_hooks = getattr(integration, "merge_vibe_event_hooks", None) + if not callable(merge_vibe_hooks): + raise TypeError("toml-vibe integrations must implement merge_vibe_event_hooks") + if merge_vibe_hooks( + project_root, + filtered, + build_dispatcher_command=lambda command, event, target_os, timeout: _dispatcher_command( + integration, + project_root, + command, + event, + target_os=target_os, + timeout_seconds=timeout, + ), + native_timeout=lambda seconds: _native_timeout(integration, seconds), + ensure_safe_destination=_ensure_safe_destination, + ): rel = str(config_path.relative_to(project_root)) if rel not in manifest.files: manifest.record_existing(rel) @@ -1673,7 +1639,13 @@ def _remove_native_event_hooks( elif fmt == "toml": _remove_toml_entries(config_path) elif fmt == "toml-vibe": - _remove_vibe_toml_entries(config_path) + remove_vibe_hooks = getattr(integration, "remove_vibe_event_hooks", None) + if not callable(remove_vibe_hooks): + raise TypeError("toml-vibe integrations must implement remove_vibe_event_hooks") + remove_vibe_hooks( + project_root, + ensure_safe_destination=_ensure_safe_destination, + ) elif fmt in ("json-nested", "json-flat"): _remove_json_entries(config_path) elif fmt == "json-root-nested": @@ -2173,42 +2145,6 @@ def _merge_toml_fragment(dst: Path, fragment: str) -> bool: return True -def _merge_vibe_toml_fragment(dst: Path, fragment: str) -> bool: - """Merge Specify-owned Vibe TOML hook entries into *dst*, regenerating the file. - - Vibe uses a flat [[hooks]] array with type/matcher/command fields. - This removes any existing Specify-marked hooks and appends the new fragment. - An unreadable or undecodable pre-existing file aborts the merge instead - of discarding the user's bytes, mirroring ``_load_user_json`` (#22). - Returns False when skipped so callers avoid tracking the untouched file - (S5). - """ - _ensure_safe_destination(dst) - existing = "" - if dst.exists(): - try: - existing = dst.read_text(encoding="utf-8") - except (OSError, UnicodeDecodeError) as exc: - logger.warning( - "Could not read %s (it may be unreadable or not UTF-8); " - "skipping event-config merge to preserve user content.", - dst, - ) - logger.debug("Read error detail: %s", exc) - return False - # Remove existing Specify-marked [[hooks]] blocks - # Match [[hooks]] ... speckit_marker = true (with any content in between) - existing = re.sub( - r'\[\[hooks\]\]\n(?:(?!\[\[hooks\]\]).)*?speckit_marker = true\n*', - "", - existing, - flags=re.DOTALL, - ) - dst.parent.mkdir(parents=True, exist_ok=True) - dst.write_text(existing.rstrip() + "\n\n" + fragment + "\n", encoding="utf-8") - return True - - def _remove_toml_entries(dst: Path) -> bool: """Remove Specify-marked TOML entries; delete the file if now empty (#14). @@ -2260,43 +2196,6 @@ def _remove_toml_entries(dst: Path) -> bool: return False -def _remove_vibe_toml_entries(dst: Path) -> bool: - """Remove Specify-marked Vibe TOML hook entries; delete the file if now empty. - - Returns True if the file was deleted (no user content remained). - """ - if not dst.exists(): - return False - _ensure_safe_destination(dst) - try: - existing = dst.read_text(encoding="utf-8") - except (OSError, UnicodeDecodeError) as exc: - logger.warning( - "Could not read %s (it may be unreadable or not UTF-8); " - "skipping event-config cleanup to preserve user content.", - dst, - ) - logger.debug("Read error detail: %s", exc) - return False - # Remove Specify-marked [[hooks]] blocks - cleaned = re.sub( - r'\[\[hooks\]\]\n(?:(?!\[\[hooks\]\]).)*?speckit_marker = true\n*', - "", - existing, - flags=re.DOTALL, - ) - # If only whitespace/comments remain, the file had no user content - stripped = "\n".join( - line for line in cleaned.splitlines() - if line.strip() and not line.strip().startswith("#") - ) - if not stripped: - dst.unlink(missing_ok=True) - return True - dst.write_text(cleaned, encoding="utf-8") - return False - - def _merge_copilot_json(dst: Path, new_hooks: dict[str, list]) -> bool: """Merge Specify-owned hooks into Copilot's dedicated hooks JSON (#8). diff --git a/src/specify_cli/integrations/vibe/__init__.py b/src/specify_cli/integrations/vibe/__init__.py index 44ae6f96df..6c0e5bf8c1 100644 --- a/src/specify_cli/integrations/vibe/__init__.py +++ b/src/specify_cli/integrations/vibe/__init__.py @@ -6,8 +6,12 @@ from __future__ import annotations +import json +import logging +import os +import re from pathlib import Path -from typing import Any +from typing import Any, Callable from ..base import IntegrationOption, SkillsIntegration from ..manifest import IntegrationManifest @@ -26,6 +30,13 @@ # place so a future command can be added here when that holds true. FORK_CONTEXT_COMMANDS: dict[str, dict[str, str]] = {} +# Keep the native Vibe timeout slightly longer than the dispatcher's inner +# timeout so Vibe does not terminate the dispatcher before it can reap its +# child process. +VIBE_EVENT_TIMEOUT_BUFFER = 5 + +logger = logging.getLogger(__name__) + class VibeIntegration(SkillsIntegration): """Integration for Mistral Vibe skills.""" @@ -200,3 +211,144 @@ def setup( ) return super().setup(project_root, manifest, parsed_options=parsed_options, **opts) + + @staticmethod + def _hook_target_os() -> str: + """Return the shell Vibe uses for hook commands on this host.""" + return "cmd" if os.name == "nt" else "host" + + @staticmethod + def _toml_quote(value: str) -> str: + """Render a TOML basic string without exposing Vibe syntax to events.""" + return json.dumps(value) + + @staticmethod + def _managed_hooks_pattern() -> re.Pattern[str]: + """Match one Vibe ``[[hooks]]`` block carrying our ownership marker.""" + return re.compile( + r"\[\[hooks\]\]\n(?:(?!\[\[hooks\]\]).)*?speckit_marker = true\n*", + re.DOTALL, + ) + + def merge_vibe_event_hooks( + self, + project_root: Path, + events: dict[str, list[dict[str, Any]]], + *, + build_dispatcher_command: Callable[[str, str, str, Any], str], + native_timeout: Callable[[Any], int], + ensure_safe_destination: Callable[[Path], None], + ) -> bool: + """Render and merge managed Vibe hooks without disturbing user content. + + This intentionally lives on Vibe rather than in shared events: Vibe's + flat TOML schema, supported fields, unique-name rule, native shell + quoting, and ownership-marker cleanup are all Vibe-specific. + """ + lines: list[str] = [] + used_names: set[str] = set() + for event, handlers in events.items(): + native = self.CANONICAL_TO_NATIVE[event] + for config in handlers: + command = config.get("command", "") + dispatcher_command = build_dispatcher_command( + command, + event, + self._hook_target_os(), + config.get("timeout", 60), + ) + command_stem = command.split(".")[-1] if command else "unknown" + command_stem = re.sub(r"[^A-Za-z0-9_-]+", "-", command_stem) or "unknown" + base_name = f"speckit-{native}-{command_stem}" + hook_name = base_name + suffix = 2 + while hook_name in used_names: + hook_name = f"{base_name}-{suffix}" + suffix += 1 + used_names.add(hook_name) + + lines.extend( + [ + "[[hooks]]", + f"name = {self._toml_quote(hook_name)}", + f"type = {self._toml_quote(native)}", + ] + ) + matcher = config.get("matcher", "*") + if matcher and matcher != "*" and native in ("pre_tool", "post_tool"): + lines.append(f"match = {self._toml_quote('re:' + matcher)}") + lines.extend( + [ + f"command = {self._toml_quote(dispatcher_command)}", + f"timeout = {native_timeout(config.get('timeout', 60) + VIBE_EVENT_TIMEOUT_BUFFER)}", + "speckit_marker = true", + "", + ] + ) + return self._merge_managed_hooks( + project_root / self.events_config_file, + "\n".join(lines), + ensure_safe_destination=ensure_safe_destination, + ) + + def remove_vibe_event_hooks( + self, + project_root: Path, + *, + ensure_safe_destination: Callable[[Path], None], + ) -> bool: + """Remove only Specify-owned Vibe hooks and delete an owned-only file.""" + path = project_root / self.events_config_file + if not path.exists(): + return False + ensure_safe_destination(path) + try: + existing = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + logger.warning( + "Could not read %s (it may be unreadable or not UTF-8); " + "skipping event-config cleanup to preserve user content.", + path, + ) + logger.debug("Read error detail: %s", exc) + return False + + cleaned = self._managed_hooks_pattern().sub("", existing) + if cleaned == existing: + return False + non_comment_content = "\n".join( + line + for line in cleaned.splitlines() + if line.strip() and not line.strip().startswith("#") + ) + if not non_comment_content: + path.unlink(missing_ok=True) + return True + path.write_text(cleaned, encoding="utf-8") + return False + + def _merge_managed_hooks( + self, + path: Path, + fragment: str, + *, + ensure_safe_destination: Callable[[Path], None], + ) -> bool: + """Replace managed entries while retaining every unowned byte sequence.""" + ensure_safe_destination(path) + existing = "" + if path.exists(): + try: + existing = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + logger.warning( + "Could not read %s (it may be unreadable or not UTF-8); " + "skipping event-config merge to preserve user content.", + path, + ) + logger.debug("Read error detail: %s", exc) + return False + cleaned = self._managed_hooks_pattern().sub("", existing) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(cleaned.rstrip() + "\n\n" + fragment + "\n", encoding="utf-8") + return True diff --git a/tests/integrations/test_integration_vibe.py b/tests/integrations/test_integration_vibe.py index 8bd26fe75f..020ece178f 100644 --- a/tests/integrations/test_integration_vibe.py +++ b/tests/integrations/test_integration_vibe.py @@ -1,5 +1,6 @@ """Tests for VibeIntegration.""" +import os from unittest.mock import MagicMock import yaml @@ -8,6 +9,7 @@ from specify_cli.integrations import get_integration from specify_cli.integrations.base import IntegrationBase from specify_cli.integrations.manifest import IntegrationManifest +from specify_cli.integrations.vibe import VibeIntegration from .test_integration_base_skills import SkillsIntegrationTests @@ -245,7 +247,9 @@ def test_windows_host_uses_cmd_quoting(self, tmp_path, monkeypatch): spaces must be double-quoted, never shlex-quoted.""" import specify_cli.events as events_mod - monkeypatch.setattr(events_mod, "_vibe_target_os", lambda: "cmd") + monkeypatch.setattr( + VibeIntegration, "_hook_target_os", staticmethod(lambda: "cmd") + ) monkeypatch.setattr( events_mod, "_resolve_interpreter", lambda root: r"C:\Program Files\Python\python.exe", @@ -258,9 +262,11 @@ def test_windows_host_uses_cmd_quoting(self, tmp_path, monkeypatch): def test_posix_host_keeps_shlex_quoting(self, tmp_path, monkeypatch): import specify_cli.events as events_mod - # Pin the target: on a Windows CI runner _vibe_target_os() would + # Pin the target: on a Windows CI runner _hook_target_os() would # return "cmd" and this test asserts the POSIX-host quoting path. - monkeypatch.setattr(events_mod, "_vibe_target_os", lambda: "host") + monkeypatch.setattr( + VibeIntegration, "_hook_target_os", staticmethod(lambda: "host") + ) monkeypatch.setattr( events_mod, "_resolve_interpreter", lambda root: "/opt/my venv/bin/python3", @@ -300,6 +306,63 @@ def test_teardown_deletes_file_without_user_content(self, tmp_path): assert not (tmp_path / ".vibe" / "hooks.toml").exists() +class TestVibeTomlNoOpRemoval: + """Unowned Vibe hooks.toml files survive no-op cleanup byte-for-byte.""" + + _FIXED_MTIME_NS = 1_700_000_000_123_456_789 + + def _user_hooks_file(self, tmp_path, content): + path = tmp_path / ".vibe" / "hooks.toml" + path.parent.mkdir(parents=True) + path.write_bytes(content) + os.utime(path, ns=(self._FIXED_MTIME_NS, self._FIXED_MTIME_NS)) + return path, path.stat().st_mtime_ns + + def _assert_untouched(self, path, original, original_mtime_ns): + assert path.exists() + assert path.read_bytes() == original + assert path.stat().st_mtime_ns == original_mtime_ns + + def test_empty_events_leave_user_crlf_file_untracked_and_untouched(self, tmp_path): + integration = get_integration("vibe") + manifest = _vibe_manifest() + original = b'user_option = "keep"\r\nsecond_option = true' + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + install_integration_events(integration, tmp_path, manifest, {}) + + self._assert_untouched(path, original, original_mtime_ns) + manifest.record_existing.assert_not_called() + + def test_empty_events_preserve_comments_only_file(self, tmp_path): + original = b"# maintained by the user\n# no hooks yet\n" + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + install_integration_events(get_integration("vibe"), tmp_path, _vibe_manifest(), {}) + + self._assert_untouched(path, original, original_mtime_ns) + + def test_empty_events_preserve_whitespace_only_file(self, tmp_path): + original = b"\r\n \t\r\n" + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + install_integration_events(get_integration("vibe"), tmp_path, _vibe_manifest(), {}) + + self._assert_untouched(path, original, original_mtime_ns) + + def test_forced_teardown_preserves_unowned_file_with_manifest_claim(self, tmp_path): + integration = get_integration("vibe") + original = b'user_option = "keep"\r\n' + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + manifest = IntegrationManifest(integration.key, tmp_path, version="test") + manifest.record_existing(".vibe/hooks.toml") + manifest.save() + + integration.teardown(tmp_path, manifest, force=True) + + self._assert_untouched(path, original, original_mtime_ns) + + class TestVibeUserInvocable: def test_all_skills_have_user_invocable(self, tmp_path): i = get_integration("vibe") diff --git a/tests/specify_cli/events/test_events.py b/tests/specify_cli/events/test_events.py index 9a193c762c..99338587b5 100644 --- a/tests/specify_cli/events/test_events.py +++ b/tests/specify_cli/events/test_events.py @@ -2106,6 +2106,54 @@ def _claude_manifest(tmp_path): return manifest +class TestNativeIntegrationDelegation: + """Shared events owns orchestration and manifest claims around native hooks.""" + + def test_vibe_merge_is_delegated_and_claimed_by_shared_events( + self, tmp_path, monkeypatch + ): + from specify_cli.integrations import get_integration + + integration = get_integration("vibe") + manifest = _claude_manifest(tmp_path) + merge = MagicMock(return_value=True) + monkeypatch.setattr(integration, "merge_vibe_event_hooks", merge) + + install_integration_events( + integration, + tmp_path, + manifest, + {"pre_tool_use": [{"command": "speckit.tdd.validate"}]}, + ) + + merge.assert_called_once() + assert merge.call_args.args[0] == tmp_path + assert {"build_dispatcher_command", "native_timeout", "ensure_safe_destination"} == set( + merge.call_args.kwargs + ) + manifest.record_existing.assert_called_once_with(".vibe/hooks.toml") + + def test_vibe_cleanup_is_delegated_and_manifest_claim_is_removed( + self, tmp_path, monkeypatch + ): + from specify_cli.integrations import get_integration + + integration = get_integration("vibe") + manifest = _claude_manifest(tmp_path) + config_path = tmp_path / ".vibe" / "hooks.toml" + config_path.parent.mkdir(parents=True) + config_path.write_text('user_option = "keep"\n', encoding="utf-8") + remove = MagicMock(return_value=False) + monkeypatch.setattr(integration, "remove_vibe_event_hooks", remove) + + remove_integration_events(integration, tmp_path, manifest) + + remove.assert_called_once() + assert remove.call_args.args == (tmp_path,) + assert set(remove.call_args.kwargs) == {"ensure_safe_destination"} + manifest.remove.assert_called_once_with(".vibe/hooks.toml") + + class TestMergeIdempotency: """#9/#11: marker recursion and full-clean-before-add."""