From 5316b568b85375e107c7494f8f8d3f47d758ea23 Mon Sep 17 00:00:00 2001 From: Quratulain-bilal Date: Tue, 4 Aug 2026 02:10:36 +0500 Subject: [PATCH] fix: use atomic writes for composed preset command files Three write_text() calls for composed command files were not atomic. A crash mid-write leaves a partial .md file which can cause errors when the command is later read. Now uses a shared _atomic_write_text helper with tempfile.mkstemp + os.replace. --- src/specify_cli/presets/__init__.py | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/src/specify_cli/presets/__init__.py b/src/specify_cli/presets/__init__.py index cc5308f3fc..30a7c90f7b 100644 --- a/src/specify_cli/presets/__init__.py +++ b/src/specify_cli/presets/__init__.py @@ -62,6 +62,23 @@ def _content_sha256(content: bytes) -> str: return hashlib.sha256(content).hexdigest() +def _atomic_write_text(path: Path, content: str) -> None: + """Write *content* to *path* atomically via mkstemp + os.replace.""" + fd, tmp = tempfile.mkstemp( + dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp" + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + def _constitution_is_generated( project_root: Path, memory_constitution: Path, @@ -848,7 +865,7 @@ def _register_commands( composed_dir = preset_dir / ".composed" composed_dir.mkdir(parents=True, exist_ok=True) composed_file = composed_dir / f"{cmd['name']}.md" - composed_file.write_text(composed, encoding="utf-8") + _atomic_write_text(composed_file, composed) commands_to_register.append({ **cmd, "file": f".composed/{cmd['name']}.md", @@ -1838,7 +1855,7 @@ def record_written(written: Dict[str, List[str]]) -> None: composed_dir = pack_dir / ".composed" composed_dir.mkdir(parents=True, exist_ok=True) composed_file = composed_dir / f"{cmd_name}.md" - composed_file.write_text(composed, encoding="utf-8") + _atomic_write_text(composed_file, composed) written = self._register_for_non_skill_agents( registrar, [{**tmpl, "file": f".composed/{cmd_name}.md"}], @@ -1858,7 +1875,7 @@ def record_written(written: Dict[str, List[str]]) -> None: shared_composed = self.presets_dir / ".composed" shared_composed.mkdir(parents=True, exist_ok=True) composed_file = shared_composed / f"{cmd_name}.md" - composed_file.write_text(composed, encoding="utf-8") + _atomic_write_text(composed_file, composed) source = layers[0]["source"] if source.startswith("extension:"): source_id = source.split(":", 1)[1].split(" ", 1)[0]