diff --git a/src/specify_cli/agents.py b/src/specify_cli/agents.py index 8f3668812f..57c8cc9011 100644 --- a/src/specify_cli/agents.py +++ b/src/specify_cli/agents.py @@ -9,6 +9,8 @@ import hashlib import os import re +import secrets +import stat from copy import deepcopy from pathlib import Path from typing import Any, Dict, Iterable, List, Optional @@ -990,6 +992,69 @@ def register_commands( return registered + @staticmethod + def _replace_file_text(dest: Path, content: str) -> None: + """Write ``content`` to ``dest`` by replacing its directory entry. + + An in-place write reaches every hard link of the file, so an alias + linked to a file outside the project would carry the new text there + too. Writing a sibling temp file and renaming it over ``dest`` only + changes what ``dest`` names. + + The replacement keeps an existing file's permission bits, and the + temp file is created with (at most) those bits and set to them before + any content is written, so the new text is never readable more widely + than the destination. A new file is created with mode 0666 so the + kernel applies the process umask, the same as an in-place write; the + umask itself is never changed. + The temp name has a fixed short length, so any name valid for + ``dest`` stays writable. + """ + try: + dest_stat = os.lstat(dest) + except FileNotFoundError: + dest_stat = None + keep_mode = ( + stat.S_IMODE(dest_stat.st_mode) + if dest_stat is not None and stat.S_ISREG(dest_stat.st_mode) + else None + ) + # O_BINARY (Windows) leaves newline translation to the text wrapper + # below, as ``tempfile.mkstemp`` and ``Path.write_text`` do. + flags = ( + os.O_WRONLY + | os.O_CREAT + | os.O_EXCL + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_BINARY", 0) + ) + for _attempt in range(100): + temp_path = dest.parent / f".speckit-{secrets.token_hex(8)}.tmp" + try: + fd = os.open( + temp_path, flags, 0o666 if keep_mode is None else keep_mode + ) + except FileExistsError: + continue + break + else: + raise FileExistsError( + f"Could not create a temporary file next to {dest}" + ) + try: + # Wrap the descriptor first so a failing ``chmod`` cannot leak it: + # registration catches write errors and carries on. + with os.fdopen(fd, "w", encoding="utf-8") as fh: + if keep_mode is not None: + # The umask may have cleared some of the destination's + # bits; restore them before writing. This never widens + # the mode beyond the destination's own. + os.chmod(temp_path, keep_mode) + fh.write(content) + os.replace(temp_path, dest) + finally: + temp_path.unlink(missing_ok=True) + @staticmethod def _write_registered_output( dest_file: Path, @@ -1012,7 +1077,7 @@ def _write_registered_output( if not link_outputs or (agent_config or {}).get("dev_no_symlink"): if dest_file.is_symlink(): dest_file.unlink() - dest_file.write_text(content, encoding="utf-8") + CommandRegistrar._replace_file_text(dest_file, content) after_file_change(dest_file) return @@ -1024,7 +1089,7 @@ def _write_registered_output( try: before_file_change(cache_file) cache_file.parent.mkdir(parents=True, exist_ok=True) - cache_file.write_text(content, encoding="utf-8") + CommandRegistrar._replace_file_text(cache_file, content) after_file_change(cache_file) if dest_file.exists() or dest_file.is_symlink(): dest_file.unlink() @@ -1036,7 +1101,7 @@ def _write_registered_output( # functional by falling back to a copy. if dest_file.is_symlink(): dest_file.unlink() - dest_file.write_text(content, encoding="utf-8") + CommandRegistrar._replace_file_text(dest_file, content) after_file_change(dest_file) @staticmethod diff --git a/src/specify_cli/presets/_manager.py b/src/specify_cli/presets/_manager.py index cf6c61f434..2b4ad5c9e7 100644 --- a/src/specify_cli/presets/_manager.py +++ b/src/specify_cli/presets/_manager.py @@ -895,6 +895,10 @@ def remove(self, pack_id: str) -> bool: registered_commands = commands_to_unregister # Unregister non-skill command files from AI agents. + if registered_commands: + registered_commands = self._drop_survivor_owned_aliases( + pack_id, registered_commands + ) if registered_commands: self._unregister_commands(registered_commands) diff --git a/src/specify_cli/presets/_manager_skills.py b/src/specify_cli/presets/_manager_skills.py index 169f6d9215..63d6bb4cf6 100644 --- a/src/specify_cli/presets/_manager_skills.py +++ b/src/specify_cli/presets/_manager_skills.py @@ -150,14 +150,19 @@ def _reconcile_skills( # directory-independent, so it's computed once and then applied to # every affected directory below. preset_cmds: Dict[str, List[str]] = {} + # Commands won by a composing preset whose stack no longer has any + # replace base: nothing resolves, so their owned output is removed. + unresolvable_cmds: Set[str] = set() non_preset_skills: List[tuple] = [] managed_skill_names: set = set() reconciled_skill_commands: set[str] = set() + cmd_layers: Dict[str, List[Dict[str, Any]]] = {} for cmd_name in command_names: layers = resolver.collect_all_layers(cmd_name, "command") if not layers: continue + cmd_layers[cmd_name] = layers skill_name, legacy_skill_name = self._skill_names_for_command( cmd_name @@ -194,6 +199,11 @@ def _reconcile_skills( pack_dir = self.presets_dir / pack_id if top_path.is_relative_to(pack_dir): preset_cmds.setdefault(pack_id, []).append(cmd_name) + if ( + all(layer["strategy"] != "replace" for layer in layers) + and resolver.resolve_content(cmd_name, "command") is None + ): + unresolvable_cmds.add(cmd_name) found_preset = True break if not found_preset: @@ -301,6 +311,7 @@ def apply_to_dir( skills_dir, skill_subdir / "SKILL.md", skill_content, + preserve_mode=True, ) wrote_override = True if ( @@ -326,6 +337,18 @@ def apply_to_dir( for name in self._skill_names_for_command(cmd) ) ] + stale_cmds = [cmd for cmd in dir_cmds if cmd in unresolvable_cmds] + if stale_cmds: + self._remove_owned_skills( + skills_dir, + [ + name + for cmd in stale_cmds + for name in self._skill_names_for_command(cmd) + if name in dir_managed_names + ], + ) + dir_cmds = [cmd for cmd in dir_cmds if cmd not in unresolvable_cmds] if not dir_cmds: continue pack_dir = self.presets_dir / pack_id @@ -361,11 +384,14 @@ def apply_to_dir( # Preserve exact prior behaviour for the currently # active directory (including the ability to create # brand-new skill subdirectories when ai_skills is on). - written = self._register_skills(filtered_manifest, pack_dir) + written = self._register_skills( + filtered_manifest, pack_dir, only_skill_names=managed_names, + ) else: written = self._register_skills( filtered_manifest, pack_dir, target_dir=skills_dir, target_agent=dir_agent or "", + only_skill_names=managed_names, ) if target_agent is None: written_names = { @@ -390,7 +416,41 @@ def apply_to_dir( # reconciliation just wrote to on its behalf (#2948). self._merge_pack_registered_skills(pack_id, written) - extra_dirs = extra_skills_dirs or {} + extra_dirs = dict(extra_skills_dirs or {}) + # {(skills_dir, agent, owner_id): {cmd_name: [skill_name, ...]}} + owner_refreshes: Dict[tuple, Dict[str, List[str]]] = {} + if extra_skills_dirs is not None: + # _unregister_skills only restored what the removed preset owned. + # Skills a surviving preset composed on top of it are its own, so + # refresh those recorded destinations from that owner's layer and + # the layers below it. A replace owner never depended on lower + # layers, and a higher preset that never registered there must + # not be rendered into it (#2948), so the global winner is used + # only when it is the owner. + for skills_dir, agent, owner_id, cmd_name, name in self._surviving_owned_skills( + command_names, presets_by_priority, active_ai, active_skills_dir + ): + layers = cmd_layers.get(cmd_name, []) + owner_dir = self.presets_dir / owner_id + owner_idx = next( + ( + idx + for idx, layer in enumerate(layers) + if layer["path"].is_relative_to(owner_dir) + ), + None, + ) + if owner_idx is None or layers[owner_idx]["strategy"] == "replace": + continue + if owner_idx == 0: + prior_agent, prior_names = extra_dirs.get(skills_dir, (agent, [])) + extra_dirs[skills_dir] = ( + prior_agent, list(dict.fromkeys([*prior_names, name])) + ) + continue + owner_refreshes.setdefault( + (skills_dir, agent, owner_id), {} + ).setdefault(cmd_name, []).append(name) if active_skills_dir: active_provenance = extra_dirs.get(active_skills_dir) if extra_skills_dirs is None or active_provenance: @@ -415,8 +475,465 @@ def apply_to_dir( managed_names=set(extra_names), ) + for (skills_dir, agent, owner_id), owned in owner_refreshes.items(): + owner_dir = self.presets_dir / owner_id + try: + manifest = PresetManifest(owner_dir / "preset.yml") + except PresetValidationError: + continue + overrides: Dict[str, str] = {} + for cmd_name, names in owned.items(): + layers = cmd_layers[cmd_name] + owner_idx = next( + idx + for idx, layer in enumerate(layers) + if layer["path"].is_relative_to(owner_dir) + ) + owner_layers = layers[owner_idx:] + composed = resolver.resolve_content( + cmd_name, "command", layers=owner_layers + ) + if composed is not None: + overrides[cmd_name] = composed + elif all(layer["strategy"] != "replace" for layer in owner_layers): + # Nothing left under the owner to compose onto. + self._remove_owned_skills(skills_dir, names) + # Otherwise a surviving layer is unreadable: keep the + # existing output rather than deleting it. + if not overrides: + continue + written = self._register_skills( + self._FilteredManifest(manifest, set(overrides)), + owner_dir, + target_dir=skills_dir, + target_agent=agent or "", + only_skill_names={ + name + for cmd_name in overrides + for name in owned[cmd_name] + }, + content_overrides=overrides, + ) + if target_agent is None or agent == target_agent: + written_names = {n for names in written.values() for n in names} + reconciled_skill_commands.update( + cmd_name + for cmd_name, names in owned.items() + if written_names.intersection(names) + ) + + self._reconcile_preset_aliases( + command_names, + presets_by_priority, + active_ai, + active_skills_dir, + create_missing=extra_skills_dirs is None, + removal_pass=extra_skills_dirs is not None, + ) + return reconciled_skill_commands + def _reconcile_preset_aliases( + self, + command_names: List[str], + presets_by_priority: List[Any], + active_ai: Optional[str], + active_skills_dir: Optional[Path], + *, + create_missing: bool = False, + removal_pass: bool = False, + ) -> None: + """Re-render, retire or restore native-skill aliases of composed commands. + + For native skill agents, a preset's command aliases are written by + command registration (``:`` source marker) and recorded in + its ``registered_commands``, not in ``registered_skills``, so the + skill pass above never sees them. When a layer under a composing + preset is added or removed, its aliases would otherwise keep the old + composition. Rewrite each from the current stack, or remove it when + nothing is left to compose onto, the same as its primary skill. Only + outputs still carrying the preset's marker are touched. + + Args: + create_missing: Also render a recorded alias that is missing from + the active agent's directory (e.g. one retired when its base + was removed, once a new base makes it composable again). Like + the primary skill, this only happens for the active agent. + removal_pass: True when called after a preset was removed, where + the primary-skill pass also repairs historical agents' + directories. On install/activation passes, rendering stays + active-only (the single-active rule of #2948); an alias that + is no longer composable is still retired wherever it is + recorded, mirroring stale-output retirement. + """ + from ..integrations._file_changes import ( + after_file_change, + before_file_change, + unlink_in_directory, + ) + from ..shared_infra import ( + _ensure_safe_shared_destination, + _ensure_safe_shared_directory, + ) + + try: + registrar = self._command_registrar() + except ImportError: + return + registrar._ensure_configs() + resolver = PresetResolver(self.project_root) + wanted = set(command_names) + + for pack_id, meta in presets_by_priority: + if not isinstance(meta, dict): + continue + recorded = meta.get("registered_commands") + if not isinstance(recorded, dict): + continue + pack_dir = self.presets_dir / pack_id + manifest = resolver._get_manifest(pack_dir) + if not manifest: + continue + alias_to_tmpl: Dict[str, Dict[str, Any]] = {} + for tmpl in manifest.templates: + if tmpl.get("type") != "command" or tmpl.get("name") not in wanted: + continue + if tmpl.get("strategy", "replace") == "replace": + # A replace layer's own file did not change. + continue + for alias in tmpl.get("aliases") or []: + if isinstance(alias, str): + alias_to_tmpl[alias] = tmpl + if not alias_to_tmpl: + continue + + # Several native skill agents can resolve to one physical + # directory (agy/codex/zed all use ``.agents/skills``). Rendering + # it once per recorded agent would let whichever agent is iterated + # last win, writing another agent's command-reference syntax over + # the active one's. Group by resolved directory and render with + # the active agent when it shares that directory, exactly as + # ``_unregister_skills`` does. + groups: Dict[Path, Dict[str, Any]] = {} + for agent, names in recorded.items(): + agent_config = registrar.AGENT_CONFIGS.get(agent) + if ( + not isinstance(names, list) + or not agent_config + or agent_config.get("extension") != "/SKILL.md" + ): + continue + skills_dir = ( + active_skills_dir + if agent == active_ai + else self._safe_skills_dir_for_agent(agent) + ) + if skills_dir is None: + continue + group = groups.setdefault(skills_dir, {"agents": [], "aliases": []}) + group["agents"].append(agent) + for name in names: + if isinstance(name, str) and name not in group["aliases"]: + group["aliases"].append(name) + + for skills_dir, group in groups.items(): + agents = group["agents"] + agent = ( + active_ai if active_ai in agents else sorted(agents)[0] + ) + agent_config = registrar.AGENT_CONFIGS[agent] + # Rendering follows the single-active rule outside a removal + # pass; retiring an uncomposable alias is allowed anywhere. + may_render = removal_pass or skills_dir == active_skills_dir + for alias in group["aliases"]: + tmpl = alias_to_tmpl.get(alias) if isinstance(alias, str) else None + if tmpl is None: + continue + skill_name = registrar._compute_output_name(agent, alias, agent_config) + skill_dir = skills_dir / skill_name + skill_file = skill_dir / "SKILL.md" + missing = not skill_dir.exists() and not skill_dir.is_symlink() + if missing: + if not (create_missing and skills_dir == active_skills_dir): + continue + else: + if not self._validate_skill_subdir( + skill_dir, create=False, skills_root=skills_dir + ) or not skill_file.is_file(): + continue + try: + content = skill_file.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + metadata = registrar.parse_frontmatter(content)[0].get("metadata") + source = ( + metadata.get("source") if isinstance(metadata, dict) else None + ) + if not ( + isinstance(source, str) and source.startswith(f"{pack_id}:") + ): + continue + + primary = tmpl["name"] + layers = resolver.collect_all_layers(primary, "command") + owner_idx = next( + ( + idx + for idx, layer in enumerate(layers) + if layer["path"].is_relative_to(pack_dir) + ), + None, + ) + if owner_idx is None: + continue + # The alias belongs to the highest layer that declares + # it, not to whichever layer wins the primary command: a + # higher preset that does not declare this alias leaves + # it to us. Only a higher declarer shadows it. + alias_shadowed = any( + self._layer_declares_alias(resolver, layer, primary, alias) + for layer in layers[:owner_idx] + ) + if alias_shadowed and missing: + continue + # A shadowed alias is not ours to render outside a + # removal pass. It still goes through the retirement + # check below: a rescaffold re-registers it from the raw + # fragment, which must not stay executable once nothing + # is left to compose onto. On a removal pass an existing + # shadowed alias is refreshed from its own layer down, as + # its primary skill is. + shadowed = alias_shadowed and not removal_pass + owner_layers = layers[owner_idx:] + composed = resolver.resolve_content( + primary, "command", layers=owner_layers + ) + if composed is None: + if missing or any( + layer["strategy"] == "replace" for layer in owner_layers + ): + # A replace base is still there, so this is a + # resolution failure (e.g. an unreadable layer), + # not a missing base: keep the alias, as the + # primary-skill path keeps its output. + continue + alias_file = skill_dir / "SKILL.md" + unlink_in_directory( + alias_file, + root=self.project_root if alias_file.is_relative_to(self.project_root) else skills_dir, + ) + continue + if shadowed or not may_render: + continue + if missing and not self._validate_skill_subdir( + skill_dir, create=True, skills_root=skills_dir + ): + continue + composed_file = pack_dir / ".composed" / f"{primary}.md" + try: + # Every ancestor and the cache file itself must stay + # inside the project and off symlinks, so a planted + # ``.composed`` link cannot redirect this write. The + # entry is replaced atomically rather than written in + # place, so a hard link to an outside file is broken, + # not written through, and an existing cache keeps + # its permission bits. + _ensure_safe_shared_directory( + self.project_root, + composed_file.parent, + create=True, + context="preset composed cache directory", + ) + _ensure_safe_shared_destination( + self.project_root, composed_file + ) + before_file_change(composed_file) + registrar._replace_file_text(composed_file, composed) + after_file_change(composed_file) + except (ValueError, OSError): + continue + registrar.register_commands( + agent, + [{"name": alias, "file": f".composed/{primary}.md"}], + pack_id, + pack_dir, + self.project_root, + _resolved_dir=skills_dir, + ) + + def _layer_declares_alias( + self, resolver: Any, layer: Dict[str, Any], primary: str, alias: str + ) -> bool: + """Whether ``layer`` comes from a preset declaring ``alias`` for ``primary``.""" + try: + rel = layer["path"].relative_to(self.presets_dir) + except (KeyError, TypeError, ValueError): + return False + if not rel.parts: + return False + manifest = resolver._get_manifest(self.presets_dir / rel.parts[0]) + if not manifest: + return False + return any( + tmpl.get("type") == "command" + and tmpl.get("name") == primary + and alias in (tmpl.get("aliases") or []) + for tmpl in manifest.templates + ) + + def _drop_survivor_owned_aliases( + self, pack_id: str, registered_commands: Dict[str, List[str]] + ) -> Dict[str, List[str]]: + """Leave out native-skill outputs another installed preset now owns. + + Two presets can declare the same command alias. For native skill + agents both record it in ``registered_commands``, but only one file + exists, carrying the ``:`` marker of whichever registered + it last. Deleting it with the preset being removed would drop a + survivor's alias that post-removal reconciliation (which does not + create missing outputs on a removal pass) cannot bring back, so keep + those names out of the unregister list and let reconciliation + recompose them. + """ + try: + registrar = self._command_registrar() + except ImportError: + return registered_commands + registrar._ensure_configs() + survivors = [pid for pid in self.registry.list() if pid != pack_id] + if not survivors: + return registered_commands + result: Dict[str, List[str]] = {} + for agent, names in registered_commands.items(): + agent_config = registrar.AGENT_CONFIGS.get(agent) + if ( + not isinstance(names, list) + or not agent_config + or agent_config.get("extension") != "/SKILL.md" + ): + result[agent] = names + continue + skills_dir = self._safe_skills_dir_for_agent(agent) + kept: List[str] = [] + for name in names: + if skills_dir is not None and isinstance(name, str): + skill_dir = skills_dir / registrar._compute_output_name( + agent, name, agent_config + ) + skill_file = skill_dir / "SKILL.md" + source = None + if self._validate_skill_subdir( + skill_dir, create=False, skills_root=skills_dir + ) and skill_file.is_file(): + try: + metadata = registrar.parse_frontmatter( + skill_file.read_text(encoding="utf-8") + )[0].get("metadata") + except (OSError, UnicodeDecodeError): + metadata = None + if isinstance(metadata, dict): + source = metadata.get("source") + if isinstance(source, str) and any( + source.startswith(f"{other}:") for other in survivors + ): + continue + kept.append(name) + if kept: + result[agent] = kept + return result + + def _surviving_owned_skills( + self, + command_names: List[str], + presets_by_priority: List[Any], + active_ai: Optional[str], + active_skills_dir: Optional[Path], + ) -> List[tuple[Path, Optional[str], str, str, str]]: + """Recorded skill destinations of ``command_names`` still owned by a survivor. + + Reads each remaining preset's ``registered_skills`` (inferring per-agent + provenance for legacy flat lists) and keeps the names whose ``SKILL.md`` + still carries that preset's ``preset:`` marker, as + ``(skills_dir, agent, owner_id, command_name, skill_name)`` tuples. + """ + registrar = self._command_registrar() + affected = { + name: cmd_name + for cmd_name in command_names + for name in self._skill_names_for_command(cmd_name) + } + found: List[tuple[Path, Optional[str], str, str, str]] = [] + for pack_id, meta in presets_by_priority: + if not isinstance(meta, dict): + continue + recorded = meta.get("registered_skills") + if isinstance(recorded, list) and active_ai: + recorded = self._infer_legacy_skill_provenance( + [n for n in recorded if isinstance(n, str)], pack_id, active_ai + ) + if not isinstance(recorded, dict): + continue + for agent, names in recorded.items(): + if not isinstance(agent, str) or not isinstance(names, list): + continue + skills_dir = ( + active_skills_dir + if agent == active_ai + else self._safe_skills_dir_for_agent(agent) + ) + if skills_dir is None: + continue + for name in names: + if name not in affected: + continue + skill_file = skills_dir / name / "SKILL.md" + if not self._validate_skill_subdir( + skill_file.parent, create=False, skills_root=skills_dir + ) or not skill_file.is_file(): + continue + try: + content = skill_file.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + metadata = registrar.parse_frontmatter(content)[0].get("metadata") + if isinstance(metadata, dict) and metadata.get("source") == f"preset:{pack_id}": + found.append((skills_dir, agent, pack_id, affected[name], name)) + return found + + def _remove_owned_skills(self, skills_dir: Path, skill_names: List[str]) -> None: + """Delete preset-owned skills among ``skill_names`` (journaled). + + A skill is preset-owned when its ``SKILL.md`` source marker names an + installed preset, either as ``preset:`` (skill registration) or + ``:`` (command registration for native skill agents); + anything else is left untouched. + """ + from ..integrations._file_changes import unlink_in_directory + + registrar = self._command_registrar() + pack_ids = list(self.registry.list()) + for name in dict.fromkeys(skill_names): + skill_dir = skills_dir / name + skill_file = skill_dir / "SKILL.md" + if not self._validate_skill_subdir( + skill_dir, create=False, skills_root=skills_dir + ) or not skill_file.is_file(): + continue + try: + content = skill_file.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + metadata = registrar.parse_frontmatter(content)[0].get("metadata") + source = metadata.get("source") if isinstance(metadata, dict) else None + if isinstance(source, str) and any( + source == f"preset:{pack_id}" or source.startswith(f"{pack_id}:") + for pack_id in pack_ids + ): + unlink_in_directory( + skill_file, + root=self.project_root if skill_file.is_relative_to(self.project_root) else skills_dir, + ) + def _resolve_agent_skills_dir(self, agent_name: str) -> Path: """Resolve the real skill output directory for an integration.""" from .. import _get_skills_dir as _project_skills_dir @@ -625,6 +1142,8 @@ def _register_skills( *, target_dir: Optional[Path] = None, target_agent: Optional[str] = None, + only_skill_names: Optional[Set[str]] = None, + content_overrides: Optional[Dict[str, str]] = None, ) -> Dict[str, List[str]]: """Generate SKILL.md files for preset command overrides. @@ -650,6 +1169,12 @@ def _register_skills( subdirectories are never created for a non-active, explicitly targeted directory (that creation path is only meaningful for the currently active agent). + only_skill_names: When given, write only these skill + directory names; any other existing modern or legacy + skill for the same command is left untouched. + content_overrides: ``{command_name: content}`` to render instead + of the preset's command file or ``.composed`` cache, for a + composition that has already been resolved by the caller. Returns: ``{agent_name: [skill_name, ...]}`` for the single active @@ -735,11 +1260,19 @@ def _register_skills( missing_skill_dir = skills_dir / skill_name if not missing_skill_dir.exists(): target_skill_names.append(skill_name) + if only_skill_names is not None: + target_skill_names = [ + name for name in target_skill_names if name in only_skill_names + ] if not target_skill_names: continue # Parse the command file - content = source_file.read_text(encoding="utf-8") + override = (content_overrides or {}).get(cmd_name) + content = ( + override if override is not None + else source_file.read_text(encoding="utf-8") + ) frontmatter, body = registrar.parse_frontmatter(content) # A composition-strategy command (wrap/prepend/append) needs a @@ -758,9 +1291,12 @@ def _register_skills( or frontmatter.get("strategy") or "replace" ) + # A cached composition can outlive its base (e.g. one removed + # while its composer was disabled), so the resolver decides even + # when the cache exists. if ( - effective_strategy != "replace" - and not composed_file.exists() + override is None + and effective_strategy != "replace" and resolver.resolve_content(cmd_name, "command") is None ): continue @@ -819,7 +1355,7 @@ def _register_skills( skill_file = skill_subdir / "SKILL.md" _write_shared_text( - skills_dir, skill_file, skill_content + skills_dir, skill_file, skill_content, preserve_mode=True ) written.append(target_skill_name) self._merge_pack_registered_skills( @@ -1480,7 +2016,9 @@ def _unregister_skills_in_dir( skill_content = integration.post_process_skill_content( skill_content ) - _write_shared_text(skills_dir, skill_file, skill_content) + _write_shared_text( + skills_dir, skill_file, skill_content, preserve_mode=True + ) mutated_names.append(skill_name) continue @@ -1538,7 +2076,9 @@ def _unregister_skills_in_dir( skill_content = integration.post_process_skill_content( skill_content ) - _write_shared_text(skills_dir, skill_file, skill_content) + _write_shared_text( + skills_dir, skill_file, skill_content, preserve_mode=True + ) mutated_names.append(skill_name) else: # No core or extension template — remove only the generated file diff --git a/src/specify_cli/presets/_resolver.py b/src/specify_cli/presets/_resolver.py index 5700fa8378..912fae652e 100644 --- a/src/specify_cli/presets/_resolver.py +++ b/src/specify_cli/presets/_resolver.py @@ -736,6 +736,8 @@ def resolve_content( self, template_name: str, template_type: str = "template", + *, + layers: Optional[List[Dict[str, Any]]] = None, ) -> Optional[str]: """Resolve a template name and return composed content. @@ -751,11 +753,15 @@ def resolve_content( Args: template_name: Template name (e.g., "spec-template") template_type: Template type ("template", "command", or "script") + layers: Compose this slice of the stack (highest priority first, + as returned by :meth:`collect_all_layers`) instead of the + whole stack, e.g. one preset's layer and everything below it. Returns: Composed content string, or None if not found """ - layers = self.collect_all_layers(template_name, template_type) + if layers is None: + layers = self.collect_all_layers(template_name, template_type) if not layers: return None diff --git a/src/specify_cli/shared_infra.py b/src/specify_cli/shared_infra.py index 44960706a1..f1a19cb9c1 100644 --- a/src/specify_cli/shared_infra.py +++ b/src/specify_cli/shared_infra.py @@ -8,6 +8,7 @@ import logging import os import re +import stat import tempfile from pathlib import Path from typing import Any @@ -317,8 +318,26 @@ def _ensure_safe_shared_destination( raise ValueError(f"Shared infrastructure destination escapes project root: {label}") from None -def _write_shared_text(project_path: Path, dest: Path, content: str) -> None: - _write_shared_bytes(project_path, dest, content.encode("utf-8")) +def _write_shared_text( + project_path: Path, dest: Path, content: str, *, preserve_mode: bool = False +) -> None: + """Atomically write ``content`` to ``dest`` (see ``_write_shared_bytes``). + + With ``preserve_mode``, an existing regular file keeps its permission + bits instead of being reset to 0644, so refreshing a private output + does not widen it; a new file still gets 0644. The content is written + into a 0600 temp file, so it is never readable more widely than that + before the final mode is applied. + """ + mode = 0o644 + if preserve_mode: + try: + dest_stat = os.lstat(dest) + except FileNotFoundError: + dest_stat = None + if dest_stat is not None and stat.S_ISREG(dest_stat.st_mode): + mode = stat.S_IMODE(dest_stat.st_mode) + _write_shared_bytes(project_path, dest, content.encode("utf-8"), mode=mode) def _write_shared_bytes( diff --git a/tests/integrations/test_integration_generic.py b/tests/integrations/test_integration_generic.py index 3ea10f7b11..99e109403f 100644 --- a/tests/integrations/test_integration_generic.py +++ b/tests/integrations/test_integration_generic.py @@ -11,6 +11,7 @@ import pytest import yaml +from specify_cli.agents import CommandRegistrar as AgentCommandRegistrar from specify_cli.integrations import get_integration from specify_cli.integrations.base import MarkdownIntegration from specify_cli.integrations.manifest import IntegrationManifest @@ -876,16 +877,29 @@ def test_generic_refresh_write_error_restores_artifacts_and_registry( second.parent.rmdir() source = manager.extensions_dir / "sample/commands/run.md" source.write_text(source.read_text(encoding="utf-8") + "\nupdated source\n", encoding="utf-8") + # Skills-mode generic artifacts are written through Path.write_text; the + # command-file path replaces the entry via the registrar helper. original_write = Path.write_text + original_replace = AgentCommandRegistrar._replace_file_text - def fail_second_write(path, *args, **kwargs): + def fail_second(path): if path == second: if partial_second: original_write(path, "partial generated output", encoding="utf-8") raise OSError("simulated refresh write error") + + def fail_second_write(path, *args, **kwargs): + fail_second(path) return original_write(path, *args, **kwargs) + def fail_second_replace(path, content): + fail_second(path) + return original_replace(path, content) + monkeypatch.setattr(Path, "write_text", fail_second_write) + monkeypatch.setattr( + AgentCommandRegistrar, "_replace_file_text", staticmethod(fail_second_replace) + ) manager.register_enabled_extensions_for_agent("generic", force=skills) monkeypatch.undo() @@ -1105,13 +1119,22 @@ def test_generic_registration_write_error_rolls_back_partial_install( "speckit-sample-other/SKILL.md" if skills else "speckit.sample.other.md" ) original_write = Path.write_text + original_replace = AgentCommandRegistrar._replace_file_text def fail_second_write(path, *args, **kwargs): if path == second: raise OSError("simulated artifact write error") return original_write(path, *args, **kwargs) + def fail_second_replace(path, content): + if path == second: + raise OSError("simulated artifact write error") + return original_replace(path, content) + monkeypatch.setattr(Path, "write_text", fail_second_write) + monkeypatch.setattr( + AgentCommandRegistrar, "_replace_file_text", staticmethod(fail_second_replace) + ) with pytest.raises(OSError, match="simulated artifact write error"): manager.install_from_directory(generic_extension, "1.0.0") monkeypatch.undo() diff --git a/tests/specify_cli/presets/test_manager_skills.py b/tests/specify_cli/presets/test_manager_skills.py index 124e0bb3bb..2badc9ec03 100644 --- a/tests/specify_cli/presets/test_manager_skills.py +++ b/tests/specify_cli/presets/test_manager_skills.py @@ -1,7 +1,9 @@ """Tests for preset skill artifacts in specify_cli.presets._manager_skills.""" +import functools import os import shutil +import stat from pathlib import Path import pytest @@ -1898,6 +1900,1038 @@ def test_remove_preserves_non_owned_skill_during_reconciliation( "---\nname: speckit-specify\n---\n\nUser-owned body\n" ) + def _install_composed_specify_stack(self, project_dir, temp_dir, layers): + """Install ``(preset_id, body, strategy, priority)`` layers for speckit.specify.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-specify") + (project_dir / ".specify" / "templates" / "commands" / "specify.md").write_text( + "---\ndescription: Core specify\n---\n\nCore specify body\n", + encoding="utf-8", + ) + manager = PresetManager(project_dir) + for preset_id, body, strategy, priority in layers: + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.specify", preset_id, body + ) + if strategy != "replace": + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + return manager, project_dir / ".claude" / "skills" / "speckit-specify" / "SKILL.md" + + def _install_unbacked_custom_stack(self, project_dir, temp_dir, layers=None): + """Presets for ``speckit.custom``, which has no core template.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + manager = PresetManager(project_dir) + for preset_id, body, strategy, priority in layers or [ + ("base-preset", "Base preset body", "replace", 10), + ("append-preset", "Appended body", "append", 5), + ]: + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + if strategy != "replace": + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + skill_file = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + if layers is None: + assert "Base preset body" in skill_file.read_text(encoding="utf-8") + return manager, skill_file + + def test_remove_sole_base_preset_retires_stale_composed_skill( + self, project_dir, temp_dir + ): + """With no core or extension fallback, removing the only base leaves + nothing to compose onto: the composer's skill is removed, and a later + rescaffold does not render its stale cached composition back.""" + manager, skill_file = self._install_unbacked_custom_stack(project_dir, temp_dir) + + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + manager.register_enabled_presets_for_agent("claude") + + assert not skill_file.exists() + + def test_remove_sole_base_preset_keeps_hand_edited_skill( + self, project_dir, temp_dir + ): + manager, skill_file = self._install_unbacked_custom_stack(project_dir, temp_dir) + edited = "---\nname: speckit-custom\n---\n\nUser-owned body\n" + skill_file.write_text(edited, encoding="utf-8") + + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + + assert skill_file.read_text(encoding="utf-8") == edited + + def test_remove_sole_base_under_two_composers_survives_rescaffold( + self, project_dir, temp_dir + ): + """With two composers and no base left, the skill is removed and a + rescaffold does not render either composer's stale cache back.""" + manager, skill_file = self._install_unbacked_custom_stack( + project_dir, + temp_dir, + [ + ("base-preset", "Base preset body", "replace", 10), + ("middle-preset", "Middle body", "append", 7), + ("top-preset", "Top body", "append", 5), + ], + ) + + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + manager.register_enabled_presets_for_agent("claude") + + assert not skill_file.exists() + + def test_stale_skill_removal_is_journaled(self, project_dir, temp_dir): + """Removing the composer's stale skill is reported to the lifecycle journal.""" + from specify_cli.integrations._file_changes import file_change_observer + + manager, _skill = self._install_unbacked_custom_stack(project_dir, temp_dir) + events = [] + token = file_change_observer.set( + lambda path, before, removal: events.append((path, before, removal)) + ) + try: + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + finally: + file_change_observer.reset(token) + + skill_dir = project_dir / ".claude" / "skills" / "speckit-custom" + assert (skill_dir, True, True) in events + assert not skill_dir.exists() + + def _install_custom_layers(self, project_dir, temp_dir, *, core=False, alias=None): + """Base (replace) + append presets for ``speckit.custom`` under Claude.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + if core: + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "custom.md").write_text( + "---\ndescription: Core custom\n---\n\nCore custom body\n", + encoding="utf-8", + ) + manager = PresetManager(project_dir) + for preset_id, body, strategy, priority in [ + ("base-preset", "Base preset body", "replace", 10), + ("append-preset", "Appended body", "append", 5), + ]: + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + if alias and strategy != "replace": + manifest["provides"]["templates"][0]["aliases"] = [alias] + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + return manager + + _ALIAS = "speckit.custom-short" + + def _alias_skill(self, project_dir): + return project_dir / ".claude" / "skills" / "speckit-custom-short" / "SKILL.md" + + def test_remove_sole_base_retires_composer_alias_skill(self, project_dir, temp_dir): + """A composer's native alias goes with its primary when no base is left, + and a rescaffold does not render the removed base's text back into it.""" + from specify_cli.integrations._file_changes import file_change_observer + + manager = self._install_custom_layers(project_dir, temp_dir, alias=self._ALIAS) + alias_file = self._alias_skill(project_dir) + assert "Base preset body" in alias_file.read_text(encoding="utf-8") + events = [] + token = file_change_observer.set( + lambda path, before, removal: events.append((path, before, removal)) + ) + try: + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + finally: + file_change_observer.reset(token) + + assert not alias_file.parent.exists() + assert (alias_file.parent, True, True) in events + with pytest.warns(UserWarning, match="no base layer"): + manager.register_enabled_presets_for_agent("claude") + assert not alias_file.parent.exists() + + def test_remove_base_recomposes_composer_alias_skill(self, project_dir, temp_dir): + """With a core fallback, the alias is recomposed on removal exactly as a + rescaffold would render it.""" + manager = self._install_custom_layers( + project_dir, temp_dir, core=True, alias=self._ALIAS + ) + alias_file = self._alias_skill(project_dir) + + assert manager.remove("base-preset") is True + + after_remove = alias_file.read_text(encoding="utf-8") + assert "Base preset body" not in after_remove + assert "Core custom body" in after_remove + assert "Appended body" in after_remove + manager.register_enabled_presets_for_agent("claude") + assert alias_file.read_text(encoding="utf-8") == after_remove + + # The alias stays tracked as the composer's command output, so + # removing the composer after the rescaffold still cleans it up. + assert manager.remove("append-preset") is True + assert not alias_file.parent.exists() + + def test_new_base_restores_retired_composer_alias_skill(self, project_dir, temp_dir): + """An alias retired with its base comes back with its primary once a new + base makes the command composable again.""" + manager = self._install_custom_layers(project_dir, temp_dir, alias=self._ALIAS) + alias_file = self._alias_skill(project_dir) + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + assert not alias_file.parent.exists() + + new_base = self._create_command_preset( + temp_dir, "new-base", "speckit.custom", "new-base", "New base body" + ) + manager.install_from_directory(new_base, "0.1.5", priority=10) + + restored = alias_file.read_text(encoding="utf-8") + assert "New base body" in restored and "Appended body" in restored + assert "Base preset body" not in restored + assert "source: append-preset:" in restored + + def test_unreadable_surviving_base_keeps_alias_skill(self, project_dir, temp_dir): + """A surviving replace base that cannot be decoded makes resolution + fail; that is not a missing base, so the alias is kept like the + primary skill's output.""" + manager = self._install_custom_layers( + project_dir, temp_dir, alias=self._ALIAS + ) + other = self._create_command_preset( + temp_dir, "other-base", "speckit.custom", "other-base", "Other base body" + ) + # Lower-priority than base-preset, so it only becomes the base once + # base-preset is removed. + manager.install_from_directory(other, "0.1.5", priority=12) + alias_file = self._alias_skill(project_dir) + before = alias_file.read_text(encoding="utf-8") + assert "Base preset body" in before + manifest = yaml.safe_load( + (manager.presets_dir / "other-base" / "preset.yml").read_text() + ) + command_file = ( + manager.presets_dir / "other-base" + / manifest["provides"]["templates"][0]["file"] + ) + command_file.write_bytes(b"---\ndescription: x\n---\n\xff\xfe broken\n") + + assert manager.remove("base-preset") is True + + assert alias_file.read_text(encoding="utf-8") == before + + def test_install_does_not_rewrite_inactive_agent_alias_skill( + self, project_dir, temp_dir + ): + """Installing a new base while another agent is active must not write + an inactive agent's alias: registration is active-only (#2948).""" + manager = self._install_custom_layers( + project_dir, temp_dir, alias=self._ALIAS + ) + claude_alias = self._alias_skill(project_dir) + before = claude_alias.read_text(encoding="utf-8") + self._write_init_options(project_dir, ai="codex", ai_skills=True) + + new_base = self._create_command_preset( + temp_dir, "new-base", "speckit.custom", "new-base", "New base body" + ) + manager.install_from_directory(new_base, "0.1.5", priority=8) + + assert claude_alias.read_text(encoding="utf-8") == before + + def test_shared_dir_alias_rendered_once_with_active_renderer( + self, project_dir, temp_dir + ): + """Codex and agy share ``.agents/skills``. A shared alias must be + rendered once, with the active agent's command-reference syntax.""" + self._write_init_options(project_dir, ai="codex", ai_skills=True) + shared = project_dir / ".agents" / "skills" + self._create_skill(shared, "speckit-custom") + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "custom.md").write_text( + "---\ndescription: Core custom\n---\n\nCore custom body\n", + encoding="utf-8", + ) + manager = PresetManager(project_dir) + for preset_id, body, strategy, priority in [ + ("base-preset", "Base preset body", "replace", 10), + ("append-preset", "Run `__SPECKIT_COMMAND_PLAN__`.", "append", 5), + ]: + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + if strategy != "replace": + manifest["provides"]["templates"][0]["aliases"] = [self._ALIAS] + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + + # `integration use agy` then back to codex records both agent keys + # against the one shared directory. + self._write_init_options(project_dir, ai="agy", ai_skills=True) + manager.register_enabled_presets_for_agent("agy") + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + recorded = manager.registry.get("append-preset").get("registered_commands", {}) + assert {"codex", "agy"} <= set(recorded) + + assert manager.remove("base-preset") is True + + alias_text = (shared / "speckit-custom-short" / "SKILL.md").read_text( + encoding="utf-8" + ) + assert "Core custom body" in alias_text + # Codex is active, so references use its "$" prefix, not agy's "/". + assert "$speckit-plan" in alias_text + assert "/speckit-plan" not in alias_text + + def test_remove_sole_base_keeps_hand_edited_alias_skill(self, project_dir, temp_dir): + manager = self._install_custom_layers(project_dir, temp_dir, alias=self._ALIAS) + alias_file = self._alias_skill(project_dir) + edited = "---\nname: speckit-custom-short\n---\n\nUser-owned body\n" + alias_file.write_text(edited, encoding="utf-8") + + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + + assert alias_file.read_text(encoding="utf-8") == edited + + + @pytest.mark.parametrize("core", [False, True], ids=["no-fallback", "core-fallback"]) + def test_remove_base_reconciles_skill_owned_by_lower_survivor_in_historical_dir( + self, project_dir, temp_dir, core + ): + """Claude's skill was rendered by ``middle``; ``top`` was installed later + under Codex only. Removing the base must still refresh Claude's skill, + whose recorded owner is ``middle``, not the current winner ``top``.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + claude_skill = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + if core: + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "custom.md").write_text( + "---\ndescription: Core custom\n---\n\nCore custom body\n", + encoding="utf-8", + ) + manager = PresetManager(project_dir) + + def install(preset_id, body, strategy, priority): + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + + install("base-preset", "Base preset body", "replace", 10) + install("middle-preset", "Middle body", "append", 7) + assert "preset:middle-preset" in claude_skill.read_text(encoding="utf-8") + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + install("top-preset", "Top body", "append", 5) + + if core: + assert manager.remove("base-preset") is True + content = claude_skill.read_text(encoding="utf-8") + assert "Base preset body" not in content + assert "Core custom body" in content and "Middle body" in content + # top-preset was never registered under Claude. + assert "Top body" not in content + assert "preset:middle-preset" in content + else: + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + assert not claude_skill.exists() + + def test_remove_shadowed_base_leaves_replace_survivor_in_historical_dir( + self, project_dir, temp_dir + ): + """``middle`` replaces ``base`` in Claude's skill; ``top`` is installed + later under Codex only. Removing the shadowed base must not touch + Claude's skill, and must not render ``top`` into it.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + claude_skill = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + manager = PresetManager(project_dir) + + def install(preset_id, body, priority): + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + + install("base-preset", "Base preset body", 10) + install("middle-preset", "Middle body", 7) + before = claude_skill.read_text(encoding="utf-8") + assert "preset:middle-preset" in before + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + install("top-preset", "Top body", 5) + + assert manager.remove("base-preset") is True + + assert claude_skill.read_text(encoding="utf-8") == before + + @pytest.mark.parametrize("core", [False, True], ids=["no-fallback", "core-fallback"]) + def test_remove_base_reconciles_alias_owned_by_lower_survivor_in_historical_dir( + self, project_dir, temp_dir, core + ): + """Like the primary skill, Claude's alias skill rendered by ``middle`` + must drop the removed base's body even though ``top`` (Codex only) + now wins, and must not pick up ``top``'s body.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + skills = project_dir / ".claude" / "skills" + self._create_skill(skills, "speckit-custom") + alias_file = self._alias_skill(project_dir) + if core: + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "custom.md").write_text( + "---\ndescription: Core custom\n---\n\nCore custom body\n", + encoding="utf-8", + ) + manager = PresetManager(project_dir) + + def install(preset_id, body, strategy, priority): + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + if strategy != "replace": + manifest["provides"]["templates"][0]["aliases"] = [self._ALIAS] + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + + install("base-preset", "Base preset body", "replace", 10) + install("middle-preset", "Middle body", "append", 7) + assert "Base preset body" in alias_file.read_text(encoding="utf-8") + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + install("top-preset", "Top body", "append", 5) + + if core: + assert manager.remove("base-preset") is True + content = alias_file.read_text(encoding="utf-8") + assert "Base preset body" not in content + assert "Core custom body" in content and "Middle body" in content + assert "Top body" not in content + assert "middle-preset:" in content + else: + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + assert not alias_file.exists() + + @pytest.mark.parametrize("link", ["cache-file", "cache-dir"]) + def test_historical_alias_refresh_does_not_write_through_symlinked_cache( + self, project_dir, temp_dir, link + ): + """Refreshing a shadowed survivor's historical alias writes its + ``.composed`` cache; a symlinked cache directory or file must not + redirect that write outside the project.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "custom.md").write_text( + "---\ndescription: Core custom\n---\n\nCore custom body\n", + encoding="utf-8", + ) + manager = PresetManager(project_dir) + + def install(preset_id, body, strategy, priority): + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + if strategy != "replace": + manifest["provides"]["templates"][0]["aliases"] = [self._ALIAS] + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + + install("base-preset", "Base preset body", "replace", 10) + install("middle-preset", "Middle body", "append", 7) + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + install("top-preset", "Top body", "append", 5) + + outside = temp_dir / "outside" + outside.mkdir() + target = outside / "speckit.custom.md" + target.write_text("external\n", encoding="utf-8") + cache_dir = manager.presets_dir / "middle-preset" / ".composed" + if link == "cache-dir": + if cache_dir.exists(): + shutil.rmtree(cache_dir) + cache_dir.symlink_to(outside, target_is_directory=True) + else: + cache_dir.mkdir(exist_ok=True) + cache_file = cache_dir / "speckit.custom.md" + if cache_file.exists(): + cache_file.unlink() + cache_file.symlink_to(target) + + assert manager.remove("base-preset") is True + + assert target.read_text(encoding="utf-8") == "external\n" + assert sorted(p.name for p in outside.iterdir()) == ["speckit.custom.md"] + + def _install_alias_layer( + self, manager, temp_dir, preset_id, body, strategy, priority, *, alias=True + ): + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + if alias: + manifest["provides"]["templates"][0]["aliases"] = [self._ALIAS] + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + + def _write_core_custom(self, project_dir): + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "custom.md").write_text( + "---\ndescription: Core custom\n---\n\nCore custom body\n", + encoding="utf-8", + ) + + def test_historical_alias_refresh_does_not_write_through_hardlinked_cache( + self, project_dir, temp_dir + ): + """A ``.composed`` cache file hard-linked to a file outside the project + passes the symlink and containment checks; the refresh must replace + the entry rather than write through the shared inode.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + self._write_core_custom(project_dir) + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10, alias=False) + install("middle-preset", "Middle body", "append", 7) + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + install("top-preset", "Top body", "append", 5) + + outside = temp_dir / "outside" + outside.mkdir() + target = outside / "speckit.custom.md" + target.write_text("external\n", encoding="utf-8") + cache_dir = manager.presets_dir / "middle-preset" / ".composed" + cache_dir.mkdir(exist_ok=True) + cache_file = cache_dir / "speckit.custom.md" + cache_file.unlink(missing_ok=True) + try: + os.link(target, cache_file) + except OSError as exc: # pragma: no cover - filesystem without hard links + pytest.skip(f"hard links unsupported: {exc}") + + assert manager.remove("base-preset") is True + + assert target.read_text(encoding="utf-8") == "external\n" + assert "Core custom body" in self._alias_skill(project_dir).read_text( + encoding="utf-8" + ) + + @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") + def test_alias_refresh_preserves_existing_output_mode( + self, project_dir, temp_dir + ): + """Replacing the alias entry must keep a private (0600) output private + rather than resetting it to a fixed world-readable mode.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + self._write_core_custom(project_dir) + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10) + install("append-preset", "Appended body", "append", 5) + alias_file = self._alias_skill(project_dir) + alias_file.chmod(0o600) + + assert manager.remove("base-preset") is True + + assert "Core custom body" in alias_file.read_text(encoding="utf-8") + assert stat.S_IMODE(alias_file.stat().st_mode) == 0o600 + + @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") + def test_new_alias_output_honors_umask(self, project_dir, temp_dir): + """A newly created alias output gets the default creation mode under + the process umask, as an ordinary file write would.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + self._write_core_custom(project_dir) + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + old_umask = os.umask(0o077) + try: + install("append-preset", "Appended body", "append", 5) + finally: + os.umask(old_umask) + + alias_file = self._alias_skill(project_dir) + assert "Appended body" in alias_file.read_text(encoding="utf-8") + assert stat.S_IMODE(alias_file.stat().st_mode) == 0o600 + + @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") + def test_historical_alias_refresh_preserves_cache_mode( + self, project_dir, temp_dir + ): + """Refreshing a shadowed survivor's ``.composed`` cache keeps a private + (0600) cache file private.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + self._write_core_custom(project_dir) + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10, alias=False) + install("middle-preset", "Middle body", "append", 7) + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + install("top-preset", "Top body", "append", 5) + cache_file = ( + manager.presets_dir / "middle-preset" / ".composed" / "speckit.custom.md" + ) + cache_file.parent.mkdir(exist_ok=True) + cache_file.write_text("stale\n", encoding="utf-8") + cache_file.chmod(0o600) + + assert manager.remove("base-preset") is True + + assert "Core custom body" in cache_file.read_text(encoding="utf-8") + assert stat.S_IMODE(cache_file.stat().st_mode) == 0o600 + + def test_rescaffold_recomposes_alias_declared_only_by_lower_preset( + self, project_dir, temp_dir + ): + """``top`` wins the primary command but does not declare the alias, so + ``middle`` still owns it: a rescaffold must recompose it onto the base + instead of leaving the raw fragment it re-registers.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10, alias=False) + install("middle-preset", "Middle body", "append", 7) + install("top-preset", "Top body", "append", 5, alias=False) + alias_file = self._alias_skill(project_dir) + + manager.register_enabled_presets_for_agent("claude") + + content = alias_file.read_text(encoding="utf-8") + assert "Base preset body" in content and "Middle body" in content + assert "Top body" not in content + assert "source: middle-preset:" in content + + def test_new_base_restores_alias_declared_only_by_lower_preset( + self, project_dir, temp_dir + ): + """An alias retired with its base comes back when a new base arrives, + even though a higher preset (which does not declare the alias) wins + the primary command.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10, alias=False) + install("middle-preset", "Middle body", "append", 7) + install("top-preset", "Top body", "append", 5, alias=False) + alias_file = self._alias_skill(project_dir) + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + assert not alias_file.parent.exists() + + install("new-base", "New base body", "replace", 10, alias=False) + + content = alias_file.read_text(encoding="utf-8") + assert "New base body" in content and "Middle body" in content + assert "Top body" not in content + + @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") + def test_skill_refresh_on_removal_preserves_skill_mode( + self, project_dir, temp_dir + ): + """Recomposing a surviving preset's skill keeps a private (0600) + ``SKILL.md`` private instead of resetting it to 0644.""" + manager = self._install_custom_layers(project_dir, temp_dir, core=True) + skill_file = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + skill_file.chmod(0o600) + + assert manager.remove("base-preset") is True + + content = skill_file.read_text(encoding="utf-8") + assert "Core custom body" in content and "Appended body" in content + assert stat.S_IMODE(skill_file.stat().st_mode) == 0o600 + + @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") + def test_skill_restore_on_removal_preserves_skill_mode( + self, project_dir, temp_dir + ): + """Restoring the core skill when the last preset goes keeps the + existing ``SKILL.md`` mode.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + self._write_core_custom(project_dir) + manager = PresetManager(project_dir) + self._install_alias_layer( + manager, temp_dir, "base-preset", "Base preset body", "replace", 10, + alias=False, + ) + skill_file = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + assert "Base preset body" in skill_file.read_text(encoding="utf-8") + skill_file.chmod(0o600) + + assert manager.remove("base-preset") is True + + assert "Base preset body" not in skill_file.read_text(encoding="utf-8") + assert stat.S_IMODE(skill_file.stat().st_mode) == 0o600 + + def test_alias_refresh_does_not_write_through_hardlinked_skill( + self, project_dir, temp_dir + ): + """An alias ``SKILL.md`` hard-linked to a file outside the project + carries the preset's own source marker, so the ownership checks pass; + the refresh on removal must replace the directory entry rather than + write through the shared inode.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + self._write_core_custom(project_dir) + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10) + install("append-preset", "Appended body", "append", 5) + alias_file = self._alias_skill(project_dir) + before = alias_file.read_text(encoding="utf-8") + assert "Base preset body" in before + + outside = temp_dir / "outside" + outside.mkdir() + target = outside / "SKILL.md" + try: + os.link(alias_file, target) + except OSError as exc: # pragma: no cover - filesystem without hard links + pytest.skip(f"hard links unsupported: {exc}") + + assert manager.remove("base-preset") is True + + assert target.read_text(encoding="utf-8") == before + after = alias_file.read_text(encoding="utf-8") + assert "Base preset body" not in after + assert "Core custom body" in after and "Appended body" in after + assert alias_file.stat().st_nlink == 1 + + def test_remove_base_keeps_survivor_alias_both_layers_declare( + self, project_dir, temp_dir + ): + """When the removed base and the surviving composer declare the same + alias, the file on disk is the composer's; removing the base must not + delete it while the core fallback keeps the command composable.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + self._write_core_custom(project_dir) + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10) + install("append-preset", "Appended body", "append", 5) + alias_file = self._alias_skill(project_dir) + assert "source: append-preset:" in alias_file.read_text(encoding="utf-8") + + assert manager.remove("base-preset") is True + + content = alias_file.read_text(encoding="utf-8") + assert "Base preset body" not in content + assert "Core custom body" in content and "Appended body" in content + assert "source: append-preset:" in content + manager.register_enabled_presets_for_agent("claude") + assert alias_file.read_text(encoding="utf-8") == content + + def test_baseless_lower_composer_alias_stays_retired_after_rescaffold( + self, project_dir, temp_dir + ): + """With two composers and no base left, a rescaffold re-registers the + lower composer's alias from its raw fragment; reconciliation must + retire it again rather than skip it because another layer wins.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + manager = PresetManager(project_dir) + install = functools.partial(self._install_alias_layer, manager, temp_dir) + install("base-preset", "Base preset body", "replace", 10, alias=False) + install("middle-preset", "Middle body", "append", 7) + install("top-preset", "Top body", "append", 5, alias=False) + alias_file = self._alias_skill(project_dir) + assert alias_file.is_file() + + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + manager.register_enabled_presets_for_agent("claude") + + assert not alias_file.parent.exists() + + def test_remove_base_keeps_lower_survivor_skill_when_core_base_is_unreadable( + self, project_dir, temp_dir + ): + """The core base under ``middle`` exists but is not UTF-8, so its + composition fails. That is not "no base left": Claude's skill, owned by + ``middle``, must be kept rather than deleted.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + claude_skill = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + self._create_skill(project_dir / ".claude" / "skills", "speckit-custom") + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "custom.md").write_bytes(b"---\ndescription: x\n---\n\n\xff\xfe\n") + manager = PresetManager(project_dir) + + def install(preset_id, body, strategy, priority): + preset_dir = self._create_command_preset( + temp_dir, preset_id, "speckit.custom", preset_id, body + ) + manifest_path = preset_dir / "preset.yml" + manifest = yaml.safe_load(manifest_path.read_text()) + manifest["provides"]["templates"][0]["strategy"] = strategy + manifest_path.write_text(yaml.dump(manifest)) + manager.install_from_directory(preset_dir, "0.1.5", priority=priority) + + install("base-preset", "Base preset body", "replace", 10) + install("middle-preset", "Middle body", "append", 7) + before = claude_skill.read_text(encoding="utf-8") + assert "preset:middle-preset" in before + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + install("top-preset", "Top body", "append", 5) + + assert manager.remove("base-preset") is True + + assert claude_skill.read_text(encoding="utf-8") == before + + def test_remove_base_reconciles_survivor_with_legacy_flat_skill_record( + self, project_dir, temp_dir + ): + """A surviving composer whose ``registered_skills`` is still a legacy + flat list gets its inactive agent's skill refreshed too.""" + manager = self._install_custom_layers(project_dir, temp_dir, core=True) + claude_skill = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + manager.registry.update("append-preset", {"registered_skills": ["speckit-custom"]}) + + assert manager.remove("base-preset") is True + + content = claude_skill.read_text(encoding="utf-8") + assert "Base preset body" not in content + assert "Core custom body" in content and "Appended body" in content + + def test_reenabled_sole_composer_does_not_render_removed_base( + self, project_dir, temp_dir + ): + """The only composer is disabled while its base is removed; re-enabling + and rescaffolding neither renders its stale cache nor keeps its old + skill, since nothing is left to compose onto.""" + manager = self._install_custom_layers(project_dir, temp_dir) + skill = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + manager.registry.update("append-preset", {"enabled": False}) + manager.remove("base-preset") + manager.registry.update("append-preset", {"enabled": True}) + + with pytest.warns(UserWarning, match="no base layer"): + manager.register_enabled_presets_for_agent("claude") + + assert not skill.exists() + + @pytest.mark.parametrize("core", [False, True], ids=["no-fallback", "core-fallback"]) + def test_remove_base_reconciles_historical_agent_skill( + self, project_dir, temp_dir, core + ): + """A composer's skill in a historical (inactive) agent directory is + recomposed onto the fallback, or retired when there is none.""" + manager = self._install_custom_layers(project_dir, temp_dir, core=core) + claude_skill = project_dir / ".claude" / "skills" / "speckit-custom" / "SKILL.md" + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + + if core: + assert manager.remove("base-preset") is True + content = claude_skill.read_text(encoding="utf-8") + assert "Base preset body" not in content + assert "Core custom body" in content and "Appended body" in content + else: + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + assert not claude_skill.exists() + + def test_remove_sole_base_with_disabled_composer_survives_rescaffold( + self, project_dir, temp_dir + ): + """A composer disabled during the removal can't bring the removed base + back through its stale cache once it is re-enabled and rescaffolded.""" + manager, skill_file = self._install_unbacked_custom_stack( + project_dir, + temp_dir, + [ + ("base-preset", "Base preset body", "replace", 10), + ("middle-preset", "Middle body", "append", 7), + ("top-preset", "Top body", "append", 5), + ], + ) + manager.registry.update("middle-preset", {"enabled": False}) + + with pytest.warns(UserWarning, match="no base layer"): + assert manager.remove("base-preset") is True + manager.registry.update("middle-preset", {"enabled": True}) + manager.register_enabled_presets_for_agent("claude") + + assert not skill_file.exists() + + def test_unreadable_surviving_base_keeps_composed_output( + self, project_dir, temp_dir + ): + """Resolution also fails when a surviving base can't be decoded; that + must not be treated as a missing base that retires the output.""" + manager, skill_file = self._install_unbacked_custom_stack( + project_dir, + temp_dir, + [ + ("base-preset", "Base preset body", "replace", 10), + ("other-base", "Other base body", "replace", 8), + ("append-preset", "Appended body", "append", 5), + ], + ) + cache = manager.presets_dir / "append-preset" / ".composed" / "speckit.custom.md" + skill_before = skill_file.read_text(encoding="utf-8") + assert cache.is_file() and "Other base body" in skill_before + manifest = yaml.safe_load( + (manager.presets_dir / "other-base" / "preset.yml").read_text() + ) + command_file = manager.presets_dir / "other-base" / manifest["provides"]["templates"][0]["file"] + command_file.write_bytes(b"---\ndescription: x\n---\n\xff\xfe broken\n") + + manager.remove("base-preset") + + assert cache.is_file() + assert skill_file.read_text(encoding="utf-8") == skill_before + + def test_remove_base_preset_recomposes_surviving_composed_skill( + self, project_dir, temp_dir + ): + """Removing a composition's base layer must recompose the surviving skill. + + The appending preset owns the composed SKILL.md, so removing the + replace preset beneath it restored nothing in the active skills + directory and reconciliation skipped it: the skill kept the removed + preset's body while command-file agents were recomposed onto core. + """ + manager, skill_file = self._install_composed_specify_stack( + project_dir, + temp_dir, + [ + ("base-preset", "Base preset body", "replace", 10), + ("append-preset", "Appended body", "append", 5), + ], + ) + content = skill_file.read_text(encoding="utf-8") + assert "Base preset body" in content and "Appended body" in content + + assert manager.remove("base-preset") is True + + content = skill_file.read_text(encoding="utf-8") + assert "Base preset body" not in content + assert "Core specify body" in content + assert "Appended body" in content + assert "preset:append-preset" in content + + def test_remove_middle_preset_drops_its_layer_from_composed_skill( + self, project_dir, temp_dir + ): + """Removing a middle composing layer must drop its fragment from the skill.""" + manager, skill_file = self._install_composed_specify_stack( + project_dir, + temp_dir, + [ + ("base-preset", "Base preset body", "replace", 10), + ("middle-preset", "Middle body", "append", 7), + ("top-preset", "Top body", "append", 5), + ], + ) + assert "Middle body" in skill_file.read_text(encoding="utf-8") + + assert manager.remove("middle-preset") is True + + content = skill_file.read_text(encoding="utf-8") + assert "Middle body" not in content + assert "Base preset body" in content + assert "Top body" in content + + def test_remove_base_preset_preserves_hand_edited_composed_skill( + self, project_dir, temp_dir + ): + """Recomposition after removal must not overwrite a hand-edited skill.""" + manager, skill_file = self._install_composed_specify_stack( + project_dir, + temp_dir, + [ + ("base-preset", "Base preset body", "replace", 10), + ("append-preset", "Appended body", "append", 5), + ], + ) + edited = "---\nname: speckit-specify\n---\n\nUser-owned body\n" + skill_file.write_text(edited, encoding="utf-8") + + assert manager.remove("base-preset") is True + + assert skill_file.read_text(encoding="utf-8") == edited + + def test_remove_base_preset_preserves_user_owned_legacy_skill( + self, project_dir, temp_dir + ): + """Recomposition writes only the skill name that still carries the + winning preset's marker, not a user-owned legacy sibling.""" + manager, skill_file = self._install_composed_specify_stack( + project_dir, + temp_dir, + [ + ("base-preset", "Base preset body", "replace", 10), + ("append-preset", "Appended body", "append", 5), + ], + ) + legacy_file = skill_file.parent.parent / "speckit.specify" / "SKILL.md" + legacy_file.parent.mkdir(parents=True, exist_ok=True) + legacy = "---\nname: speckit.specify\n---\n\nUser-owned legacy body\n" + legacy_file.write_text(legacy, encoding="utf-8") + + assert manager.remove("base-preset") is True + + assert legacy_file.read_text(encoding="utf-8") == legacy + content = skill_file.read_text(encoding="utf-8") + assert "Base preset body" not in content + assert "Appended body" in content + def test_shared_skills_dir_restored_once_using_active_agent( self, project_dir, temp_dir ): @@ -2104,6 +3138,42 @@ def test_remove_reconciles_skill_for_every_historical_agent( "the currently active agent" ) + def test_remove_keeps_user_owned_legacy_skill_in_historical_dir( + self, project_dir, temp_dir + ): + """Reconciling an inactive agent's directory writes only the managed + skill name, not a user-owned legacy sibling beside it.""" + self._write_init_options(project_dir, ai="claude", ai_skills=True) + claude_skills_dir = project_dir / ".claude" / "skills" + core_cmds = project_dir / ".specify" / "templates" / "commands" + core_cmds.mkdir(parents=True, exist_ok=True) + (core_cmds / "specify.md").write_text( + "---\ndescription: Core specify command\n---\n\nCore specify body\n", + encoding="utf-8", + ) + preset_b_dir = self._create_command_preset( + temp_dir, "hist-legacy-b", "speckit.specify", "Preset B", "preset B body", + ) + preset_a_dir = self._create_command_preset( + temp_dir, "hist-legacy-a", "speckit.specify", "Preset A", "preset A body", + ) + manager = PresetManager(project_dir) + manager.install_from_directory(preset_b_dir, "0.1.5", priority=10) + manager.install_from_directory(preset_a_dir, "0.1.5", priority=1) + self._write_init_options(project_dir, ai="codex", ai_skills=True) + manager.register_enabled_presets_for_agent("codex") + legacy_file = claude_skills_dir / "speckit.specify" / "SKILL.md" + legacy_file.parent.mkdir(parents=True) + legacy = "---\nname: speckit.specify\n---\n\nUser-owned legacy body\n" + legacy_file.write_text(legacy, encoding="utf-8") + + assert manager.remove("hist-legacy-a") is True + + assert legacy_file.read_text(encoding="utf-8") == legacy + assert "preset:hist-legacy-b" in ( + claude_skills_dir / "speckit-specify" / "SKILL.md" + ).read_text() + def test_skill_reconciliation_preserves_per_directory_names( self, project_dir, temp_dir ): diff --git a/tests/specify_cli/test_agents_replace_file_text.py b/tests/specify_cli/test_agents_replace_file_text.py new file mode 100644 index 0000000000..881c7039bb --- /dev/null +++ b/tests/specify_cli/test_agents_replace_file_text.py @@ -0,0 +1,104 @@ +"""Tests for ``CommandRegistrar._replace_file_text`` (atomic artifact writes).""" + +import os +import stat + +import pytest + +from specify_cli.agents import CommandRegistrar + + +def test_replace_keeps_long_valid_filename_writable(tmp_path): + """The temp entry must not be longer than the destination allows: a + 250-byte name is valid and an in-place write accepted it.""" + dest = tmp_path / ("a" * 247 + ".md") + dest.write_text("old\n", encoding="utf-8") + + CommandRegistrar._replace_file_text(dest, "new\n") + + assert dest.read_text(encoding="utf-8") == "new\n" + assert sorted(p.name for p in tmp_path.iterdir()) == [dest.name] + + +def test_replace_creates_long_valid_filename(tmp_path): + dest = tmp_path / ("b" * 247 + ".md") + + CommandRegistrar._replace_file_text(dest, "new\n") + + assert dest.read_text(encoding="utf-8") == "new\n" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_replace_does_not_touch_process_umask(tmp_path, monkeypatch): + """Reading the umask by setting it to 0 is process-wide and racy; a new + file must get the creation mode from the kernel instead.""" + old_umask = os.umask(0o027) + calls = [] + monkeypatch.setattr(os, "umask", lambda mask: calls.append(mask) or 0o027) + try: + dest = tmp_path / "new.md" + CommandRegistrar._replace_file_text(dest, "new\n") + finally: + monkeypatch.undo() + os.umask(old_umask) + + assert calls == [] + assert stat.S_IMODE(dest.stat().st_mode) == 0o640 + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_replace_writes_private_content_only_into_private_temp(tmp_path, monkeypatch): + """Replacing a 0600 file under a 022 umask must not expose the new + content through a 0644 temp file before the final chmod.""" + dest = tmp_path / "private.md" + dest.write_text("old\n", encoding="utf-8") + dest.chmod(0o600) + modes_at_write = [] + real_fdopen = os.fdopen + + def recording_fdopen(fd, *args, **kwargs): + modes_at_write.append(stat.S_IMODE(os.fstat(fd).st_mode)) + return real_fdopen(fd, *args, **kwargs) + + old_umask = os.umask(0o022) + try: + monkeypatch.setattr(os, "fdopen", recording_fdopen) + CommandRegistrar._replace_file_text(dest, "secret\n") + finally: + monkeypatch.undo() + os.umask(old_umask) + + assert modes_at_write == [0o600] + assert dest.read_text(encoding="utf-8") == "secret\n" + assert stat.S_IMODE(dest.stat().st_mode) == 0o600 + + +def test_replace_closes_descriptor_when_chmod_fails(tmp_path, monkeypatch): + """A failing ``chmod`` must not leak the temp file's descriptor: callers + catch registration errors and continue, so a leak would accumulate.""" + dest = tmp_path / "owned.md" + dest.write_text("old\n", encoding="utf-8") + opened = [] + real_open = os.open + + def recording_open(path, *args, **kwargs): + fd = real_open(path, *args, **kwargs) + opened.append(fd) + return fd + + def failing_chmod(path, mode, *args, **kwargs): + raise PermissionError(f"chmod denied: {path}") + + monkeypatch.setattr(os, "open", recording_open) + monkeypatch.setattr(os, "chmod", failing_chmod) + try: + with pytest.raises(PermissionError): + CommandRegistrar._replace_file_text(dest, "new\n") + finally: + monkeypatch.undo() + + assert len(opened) == 1 + with pytest.raises(OSError): + os.fstat(opened[0]) + assert dest.read_text(encoding="utf-8") == "old\n" + assert sorted(p.name for p in tmp_path.iterdir()) == [dest.name] diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 8f370b1a64..91097c4a09 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -4794,14 +4794,18 @@ def test_dev_register_commands_falls_back_to_copy_when_cache_write_fails( """Dev-mode registration stays functional when the dev cache is unwritable.""" agents_dir = project_dir / ".github" / "agents" agents_dir.mkdir(parents=True) - original_write_text = Path.write_text + from specify_cli.agents import CommandRegistrar as AgentCommandRegistrar + + original_write = AgentCommandRegistrar._replace_file_text - def raise_cache_write_error(path, *args, **kwargs): + def raise_cache_write_error(path, content): if ".specify-dev" in path.parts: raise OSError("cache is not writable") - return original_write_text(path, *args, **kwargs) + return original_write(path, content) - monkeypatch.setattr(Path, "write_text", raise_cache_write_error) + monkeypatch.setattr( + AgentCommandRegistrar, "_replace_file_text", staticmethod(raise_cache_write_error) + ) manifest = ExtensionManifest(extension_dir / "extension.yml") registrar = CommandRegistrar()