diff --git a/src/specify_cli/events/__init__.py b/src/specify_cli/events/__init__.py index 0d1aa3a0a1..bc415d8b2f 100644 --- a/src/specify_cli/events/__init__.py +++ b/src/specify_cli/events/__init__.py @@ -1230,17 +1230,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, @@ -1504,51 +1493,29 @@ 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)): - rel = str(config_path.relative_to(project_root)) - manifest.record_existing(rel, ownership="partial") + # 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 = config_path.relative_to(project_root).as_posix() + manifest.record_existing(rel) + manifest.set_ownership(rel, "partial") created.append(config_path) elif fmt == "json-flat": @@ -1671,7 +1638,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": @@ -2204,42 +2177,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) - _write_text(dst, 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). @@ -2291,43 +2228,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: - _unlink_file(dst, missing_ok=True) - return True - _write_text(dst, 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 e0d0957d9a..692e303da6 100644 --- a/src/specify_cli/integrations/vibe/__init__.py +++ b/src/specify_cli/integrations/vibe/__init__.py @@ -6,12 +6,18 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence +import logging +import os +import re +import tomllib +from collections.abc import Callable, Iterator, Mapping, Sequence from pathlib import Path from typing import Any from ..base import IntegrationOption, SkillsIntegration +from .._file_changes import unlink as _unlink_file, write_bytes as _write_bytes from ..manifest import IntegrationManifest +from ..._toml_string import escape_toml_basic from ..._utils import dump_frontmatter # Per-command frontmatter overrides for skills that should run in a forked @@ -27,6 +33,28 @@ # 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__) + + +def _toml_physical_lines(source: str) -> Iterator[str]: + """Yield TOML physical lines, preserving their original terminators. + + TOML recognizes LF and CRLF only. In particular, Python's ``splitlines`` + would incorrectly treat bare CR and Unicode line-separator characters as + physical line boundaries. + """ + start = 0 + while (newline := source.find("\n", start)) >= 0: + yield source[start:newline + 1] + start = newline + 1 + if start < len(source): + yield source[start:] + class VibeIntegration(SkillsIntegration): """Integration for Mistral Vibe skills.""" @@ -228,3 +256,368 @@ 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 escape_toml_basic(value) + + @staticmethod + def _parse_toml_header(code: str) -> tuple[bool, tuple[str | None, ...] | None]: + """Return a TOML header's kind and key path, or ``None`` if it is not one. + + This deliberately recognizes only header syntax, not full TOML. Its + caller has already excluded comments and unterminated strings. A + ``None`` key component represents a quoted basic key with an escape: + it is still a real header, but cannot be the literal ``hooks`` key we + manage without decoding TOML escapes. + """ + text = code.strip(" \t") + is_array = text.startswith("[[") + if is_array: + if not text.endswith("]]") or len(text) < 5: + return False, None + body = text[2:-2] + else: + if not text.startswith("[") or not text.endswith("]") or len(text) < 3: + return False, None + body = text[1:-1] + + parts: list[str | None] = [] + index = 0 + length = len(body) + while True: + while index < length and body[index] in " \t": + index += 1 + if index == length: + return False, None + + if body[index] in "\"'": + quote = body[index] + index += 1 + start = index + escaped = False + while index < length: + char = body[index] + if quote == "\"" and escaped: + escaped = False + elif quote == "\"" and char == "\\": + escaped = True + elif char == quote: + raw_part = body[start:index] + parts.append(raw_part if "\\\\" not in raw_part else None) + index += 1 + break + index += 1 + else: + return False, None + else: + start = index + while index < length and (body[index].isalnum() or body[index] in "_-"): + # TOML bare keys are ASCII-only; non-ASCII alphanumerics + # make this header ambiguous, so fail closed. + if not body[index].isascii(): + return False, None + index += 1 + if start == index: + return False, None + parts.append(body[start:index]) + + while index < length and body[index] in " \t": + index += 1 + if index == length: + return is_array, tuple(parts) + if body[index] != ".": + return False, None + index += 1 + + @staticmethod + def _managed_hook_spans(source: str) -> list[tuple[int, int]] | None: + """Find Specify-owned hook blocks in valid TOML without interpreting values. + + The scanner only needs TOML's lexical boundaries: comments, ordinary + strings, multiline strings, multiline arrays, and standalone table + headers. A parser validity gate and the scanner both return ``None`` + for malformed or ambiguous input so callers can leave the source + byte-for-byte untouched. + """ + try: + tomllib.loads(source) + except tomllib.TOMLDecodeError: + return None + + spans: list[tuple[int, int]] = [] + current_start: int | None = None + current_is_hooks = False + current_has_marker = False + multiline: str | None = None + array_depth = 0 + offset = 0 + + for line in _toml_physical_lines(source): + content = line.rstrip("\r\n") + index = 0 + code_end = len(content) + line_starts_in_multiline_string = multiline is not None + line_has_multiline_string = line_starts_in_multiline_string + line_starts_in_array = array_depth > 0 + structural = list(content) + while index < len(content): + char = content[index] + if multiline is not None: + if multiline == '"""' and char == "\\": + # A basic multiline string may escape its next + # character (including a newline continuation). + structural[index:min(index + 2, len(content))] = " " * min( + 2, len(content) - index + ) + index += 2 + continue + if content.startswith(multiline, index): + run_end = index + while run_end < len(content) and content[run_end] == multiline[0]: + run_end += 1 + if run_end - index > 5: + return None + if run_end - index >= 3: + structural[index:run_end] = " " * (run_end - index) + multiline = None + index = run_end + continue + structural[index] = " " + index += 1 + continue + + if char == "#": + code_end = index + break + if content.startswith('"""', index) or content.startswith("'''", index): + multiline = content[index:index + 3] + line_has_multiline_string = True + structural[index:index + 3] = " " + index += 3 + continue + if char in "\"'": + quote = char + start = index + index += 1 + while index < len(content): + if quote == "\"" and content[index] == "\\": + if index + 1 == len(content): + return None + index += 2 + continue + if content[index] == quote: + index += 1 + structural[start:index] = " " * (index - start) + break + index += 1 + else: + return None + continue + index += 1 + + if multiline is not None and not line.endswith(("\n", "\r")): + return None + + # A multiline string can contain arbitrary text that resembles a + # header or marker. A header/assignment cannot validly share its + # physical line, so ignore that line for ownership purposes. + code = "" if line_has_multiline_string else content[:code_end] + is_array, header_path = ( + VibeIntegration._parse_toml_header(code) + if not line_starts_in_array and not line_has_multiline_string + else (False, None) + ) + if header_path is not None: + if current_start is not None and current_has_marker: + spans.append((current_start, offset)) + current_start = offset + current_is_hooks = is_array and header_path == ("hooks",) + current_has_marker = False + elif ( + not line_starts_in_array + and not line_starts_in_multiline_string + and content.lstrip(" \t").startswith("[") + ): + # A line that starts like a header but cannot be classified is + # ambiguous (for example, a malformed header); never remove + # content from such a file. Inspect the original line here: + # triple-quoted tokens are illegal in a header, but would + # otherwise blank ``code`` before this guard sees it. + return None + elif current_is_hooks and re.fullmatch( + r"[ \t]*speckit_marker[ \t]*=[ \t]*true[ \t]*", code + ): + current_has_marker = True + + if header_path is None: + for char in structural[:code_end]: + if char == "[": + array_depth += 1 + elif char == "]": + array_depth -= 1 + if array_depth < 0: + return None + + offset += len(line) + + if multiline is not None or array_depth: + return None + if current_start is not None and current_has_marker: + spans.append((current_start, len(source))) + return spans + + @classmethod + def _without_managed_hooks(cls, source: str) -> str | None: + """Remove managed spans, preserving every other source slice verbatim.""" + spans = cls._managed_hook_spans(source) + if spans is None: + return None + if not spans: + return source + slices: list[str] = [] + previous_end = 0 + for start, end in spans: + slices.append(source[previous_end:start]) + previous_end = end + slices.append(source[previous_end:]) + return "".join(slices) + + 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: + with path.open(encoding="utf-8", newline="") as file: + existing = file.read() + 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._without_managed_hooks(existing) + if cleaned is None: + return False + 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: + _unlink_file(path, missing_ok=True) + return True + _write_bytes(path, cleaned.encode("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 unowned content. + + Active merges normalize trailing separators before appending generated + hooks, so they do not promise byte-for-byte preservation. + """ + ensure_safe_destination(path) + existing = "" + if path.exists(): + try: + with path.open(encoding="utf-8", newline="") as file: + existing = file.read() + 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._without_managed_hooks(existing) + if cleaned is None: + return False + path.parent.mkdir(parents=True, exist_ok=True) + _write_bytes(path, (cleaned.rstrip() + "\n\n" + fragment + "\n").encode("utf-8")) + return True diff --git a/tests/integrations/test_integration_vibe.py b/tests/integrations/test_integration_vibe.py index c2189a4b84..349c15fb17 100644 --- a/tests/integrations/test_integration_vibe.py +++ b/tests/integrations/test_integration_vibe.py @@ -1,13 +1,16 @@ """Tests for VibeIntegration.""" +import os from unittest.mock import MagicMock +import pytest import yaml from specify_cli.events import install_integration_events, remove_integration_events 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 @@ -217,6 +220,41 @@ def test_wildcard_matcher_omitted(self, tmp_path): (hook,) = self._parse(tmp_path)["hooks"] assert "match" not in hook + def test_non_bmp_matcher_round_trips_through_toml(self, tmp_path): + """Vibe hook strings retain non-BMP Unicode rather than TOML-invalid surrogates.""" + matcher = "Edit|😀" + self._install(tmp_path, { + "pre_tool_use": [{"command": "speckit.tdd.validate", "matcher": matcher}], + }) + (hook,) = self._parse(tmp_path)["hooks"] + assert hook["match"] == f"re:{matcher}" + + @pytest.mark.parametrize("separator", ("\u0085", "\u2028", "\u2029")) + def test_generated_matcher_unicode_separator_lifecycle(self, tmp_path, separator): + """Unicode separators in TOML strings do not split managed hook spans.""" + matcher = f"Edit{separator}Write" + integration, _ = self._install(tmp_path, { + "pre_tool_use": [{"command": "speckit.tdd.validate", "matcher": matcher}], + }) + assert self._parse(tmp_path)["hooks"][0]["match"] == f"re:{matcher}" + + self._install(tmp_path, { + "pre_tool_use": [{"command": "speckit.review.verify", "matcher": matcher}], + }) + (hook,) = self._parse(tmp_path)["hooks"] + assert hook["name"] == "speckit-pre_tool-verify" + assert hook["match"] == f"re:{matcher}" + + assert integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) is True + assert not (tmp_path / ".vibe" / "hooks.toml").exists() + + def test_toml_quote_round_trips_special_characters(self): + value = 'quote: " backslash: \\ newline: \n carriage: \r tab: \t nul: \x00 unit: \x1f del: \x7f' + quoted = VibeIntegration._toml_quote(value) + assert tomllib.loads(f"value = {quoted}\n")["value"] == value + def test_unsupported_events_are_skipped(self, tmp_path, capsys): self._install(tmp_path, { "session_start": [{"command": "speckit.agent-context.update"}], @@ -276,6 +314,80 @@ def test_merge_and_teardown_preserve_user_hooks(self, tmp_path): remaining = self._parse(tmp_path)["hooks"] assert [h["name"] for h in remaining] == ["deny-rm-rf"] + @pytest.mark.parametrize("newline", ("\n", "\r\n"), ids=("lf", "crlf")) + def test_merge_preserves_user_physical_line_bytes(self, tmp_path, newline): + """A native Windows write must not expand retained CRLF to CRCRLF.""" + user_content = ( + f"[[hooks]]{newline}" + f'name = "user-hook"{newline}' + f'type = "pre_tool"{newline}' + f'command = "guard-bash"' + ) + path = tmp_path / ".vibe" / "hooks.toml" + path.parent.mkdir(parents=True) + path.write_bytes((user_content + newline).encode("utf-8")) + + self._install(tmp_path, {"pre_tool_use": [{"command": "speckit.tdd.validate"}]}) + + merged = path.read_bytes() + assert merged.startswith(user_content.encode("utf-8") + b"\n\n") + assert b"\r\r\n" not in merged + assert [hook["name"] for hook in self._parse(tmp_path)["hooks"]] == [ + "user-hook", + "speckit-pre_tool-validate", + ] + + def test_merge_preserves_user_command_with_marker_text(self, tmp_path): + """A marker-looking command string does not claim a user hook block.""" + config_path = tmp_path / ".vibe" / "hooks.toml" + config_path.parent.mkdir(parents=True) + user_block = ( + "[[hooks]]\n" + 'name = "user-hook"\n' + 'type = "pre_tool"\n' + 'command = "echo speckit_marker = true"\n' + ) + config_path.write_text(user_block, encoding="utf-8") + + self._install(tmp_path, {"pre_tool_use": [{"command": "speckit.tdd.validate"}]}) + + assert user_block in config_path.read_text(encoding="utf-8") + assert [hook["name"] for hook in self._parse(tmp_path)["hooks"]] == [ + "user-hook", + "speckit-pre_tool-validate", + ] + + @pytest.mark.parametrize("separator", ("\u0085", "\u2028", "\u2029")) + def test_comment_unicode_separator_cannot_spoof_marker(self, tmp_path, separator): + """Only TOML LF and CRLF can end a comment for ownership scanning.""" + user_block = ( + "[[hooks]]\n" + 'name = "user-hook"\n' + 'type = "pre_tool"\n' + f"# user note{separator}speckit_marker = true\n" + 'command = "guard-bash"\n' + ) + path = tmp_path / ".vibe" / "hooks.toml" + path.parent.mkdir(parents=True) + original = user_block.encode() + path.write_bytes(original) + mtime_ns = 1_700_000_000_123_456_789 + os.utime(path, ns=(mtime_ns, mtime_ns)) + + integration = get_integration("vibe") + assert integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) is False + assert path.read_bytes() == original + assert path.stat().st_mtime_ns == mtime_ns + + self._install(tmp_path, {"pre_tool_use": [{"command": "speckit.tdd.validate"}]}) + assert user_block in path.read_bytes().decode("utf-8") + assert [hook["name"] for hook in self._parse(tmp_path)["hooks"]] == [ + "user-hook", + "speckit-pre_tool-validate", + ] + def test_commands_carry_structured_output_envelope(self, tmp_path): """Vibe parses non-empty hook stdout as JSON (HookStructuredResponse); plain text is reported as a hook failure. Every generated hook command @@ -293,7 +405,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", @@ -306,9 +420,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", @@ -348,6 +464,308 @@ 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_removal_preserves_user_comment_with_marker_text(self, tmp_path): + integration = get_integration("vibe") + original = ( + b"[[hooks]]\n" + b'name = "user-hook"\n' + b'type = "pre_tool"\n' + b"# speckit_marker = true\n" + b'command = "guard-bash"\n' + ) + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + self._assert_untouched(path, original, original_mtime_ns) + + @pytest.mark.parametrize("newline", ("\n", "\r\n"), ids=("lf", "crlf")) + def test_removal_recognizes_valid_marker_with_toml_line_endings(self, tmp_path, newline): + """Removal uses both TOML line endings without expanding retained CRLF.""" + managed = ( + f"[[hooks]]{newline}" + f"name = \"managed\"{newline}" + f"speckit_marker = true{newline}" + ) + user = ( + f"[[hooks]]{newline}" + f"name = \"user-hook\"{newline}" + f"command = \"guard-bash\"{newline}" + ) + path, _ = self._user_hooks_file(tmp_path, (managed + user).encode()) + + removed_file = get_integration("vibe").remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + assert path.read_bytes() == user.encode() + + def test_removal_preserves_marker_inside_multiline_strings(self, tmp_path): + integration = get_integration("vibe") + for name, delimiter in (("basic", '"""'), ("literal", "'''")): + original = ( + "[[hooks]]\n" + f"command = {delimiter}\n" + "speckit_marker = true\n" + f"{delimiter}\n" + ).encode() + path, original_mtime_ns = self._user_hooks_file(tmp_path / name, original) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path / name, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + self._assert_untouched(path, original, original_mtime_ns) + + def test_removal_accepts_spaced_hooks_header(self, tmp_path): + integration = get_integration("vibe") + path = tmp_path / ".vibe" / "hooks.toml" + path.parent.mkdir(parents=True) + path.write_text( + "[[ hooks ]]\n" + "speckit_marker = true\n" + "[[hooks]]\n" + 'name = "user-hook"\n', + encoding="utf-8", + ) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + assert path.read_text(encoding="utf-8") == '[[hooks]]\nname = "user-hook"\n' + + def test_removal_does_not_cross_an_intervening_table(self, tmp_path): + integration = get_integration("vibe") + original = ( + b"[[hooks]]\n" + b'command = "user-hook"\n' + b"[other]\n" + b"speckit_marker = true\n" + ) + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + self._assert_untouched(path, original, original_mtime_ns) + + def test_removal_ignores_headers_inside_a_managed_multiline_string(self, tmp_path): + integration = get_integration("vibe") + path = tmp_path / ".vibe" / "hooks.toml" + path.parent.mkdir(parents=True) + path.write_text( + "[[hooks]]\n" + 'command = """\n' + "[[hooks]]\n" + "not a header\n" + '"""\n' + "speckit_marker = true\n" + "[other]\n" + 'value = "preserved"\n', + encoding="utf-8", + ) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + assert path.read_text(encoding="utf-8") == '[other]\nvalue = "preserved"\n' + + def test_removal_ignores_header_like_nested_arrays(self, tmp_path): + """An array value must not split a managed array-of-tables block.""" + integration = get_integration("vibe") + path = tmp_path / ".vibe" / "hooks.toml" + path.parent.mkdir(parents=True) + path.write_text( + "[[hooks]]\n" + "values = [\n" + ' [[ "hooks" ]]\n' + "]\n" + "speckit_marker = true\n" + "[other]\n" + 'value = "preserved"\n', + encoding="utf-8", + ) + assert tomllib.loads(path.read_text(encoding="utf-8"))["hooks"][0]["values"] == [ + [["hooks"]] + ] + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + assert path.read_text(encoding="utf-8") == '[other]\nvalue = "preserved"\n' + + def test_removal_leaves_malformed_multiline_source_untouched(self, tmp_path): + integration = get_integration("vibe") + original = ( + b"[[hooks]]\n" + b'command = """\n' + b"speckit_marker = true\n" + ) + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + self._assert_untouched(path, original, original_mtime_ns) + + @pytest.mark.parametrize( + "original", + [ + ( + "[[hooks]]\r\n" + 'name = "\u7528\u6237-hook"\r\n' + "values = [speckit_marker = true]\r\n" + "[other]\r\n" + 'keep = "\u4fdd\u7559"\r\n' + ).encode(), + ( + "[[hooks]]\n" + 'name = "caf\u00e9-user"\n' + "bare invalid text\n" + "speckit_marker = true\n" + "[other]\n" + 'keep = "\u4fdd\u7559"\n' + ).encode(), + ( + "[[hooks]]\r" + 'name = "user-hook"\r' + "speckit_marker = true\r" + ).encode(), + ], + ids=("marker-like-array", "bare-invalid-text", "bare-cr"), + ) + def test_malformed_toml_leaves_removal_and_merge_unowned_and_untouched( + self, tmp_path, original + ): + """Invalid TOML must fail closed even when it resembles a managed hook.""" + integration = get_integration("vibe") + with pytest.raises(tomllib.TOMLDecodeError): + tomllib.loads(original.decode()) + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + self._assert_untouched(path, original, original_mtime_ns) + + manifest = _vibe_manifest() + install_integration_events( + integration, + tmp_path, + manifest, + {"pre_tool_use": [{"command": "speckit.tdd.validate"}]}, + ) + + self._assert_untouched(path, original, original_mtime_ns) + manifest.record_existing.assert_not_called() + + def test_malformed_header_with_triple_quotes_leaves_removal_and_merge_untouched( + self, tmp_path + ): + """Malformed header tokens must not let a user hook gain our marker.""" + integration = get_integration("vibe") + original = ( + b"[[hooks]]\n" + b'name = "user-owned"\n' + b'[ """malformed-header""" ]\n' + b"speckit_marker = true\n" + b"[other]\n" + b'keep = "survives"\n' + ) + with pytest.raises(tomllib.TOMLDecodeError): + tomllib.loads(original.decode("utf-8")) + path, original_mtime_ns = self._user_hooks_file(tmp_path, original) + + removed_file = integration.remove_vibe_event_hooks( + tmp_path, ensure_safe_destination=lambda _path: None + ) + + assert removed_file is False + self._assert_untouched(path, original, original_mtime_ns) + + install_integration_events( + integration, + tmp_path, + _vibe_manifest(), + {"pre_tool_use": [{"command": "speckit.tdd.validate"}]}, + ) + + 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 c92e9e30ad..9a2b52445a 100644 --- a/tests/specify_cli/events/test_events.py +++ b/tests/specify_cli/events/test_events.py @@ -2169,6 +2169,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."""