diff --git a/backend/app/gateway/routers/AGENTS.md b/backend/app/gateway/routers/AGENTS.md
index a460204fc0..91f0f0088b 100644
--- a/backend/app/gateway/routers/AGENTS.md
+++ b/backend/app/gateway/routers/AGENTS.md
@@ -67,6 +67,14 @@ Plugin actions stamp `ActionContext.interaction=session` only from authenticated
 session provenance. Never accept it from payloads or propagate it to model tools;
 extensions use it to distinguish owner review from model/tool output.
 
+Skill archive upload copies (`_copy_uploaded_skill_archive`) report the original
+size-limit or read error when temporary cleanup or its warning log fails.
+Successful copies still return the uploaded bytes unchanged. The upload route's
+`finally` discards that archive through the same helper before closing the
+multipart form. Form close is suppressed, so a close, cleanup, or warning
+failure cannot turn a successful install into a 500, replace an install
+error, or skip the discard. Coverage: `tests/test_skill_archive_upload_cleanup.py`.
+
 State updates synchronize titles into thread metadata only for the root checkpoint
 namespace. Nonroot writes preserve the conversation title, branch-title sequence
 and activity timestamp. `tests/test_threads_router.py` checks metadata/search
diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py
index 0d663c8139..b4b7833662 100644
--- a/backend/app/gateway/routers/skills.py
+++ b/backend/app/gateway/routers/skills.py
@@ -4,7 +4,7 @@ import logging
 import tempfile
 import threading
 from collections.abc import AsyncGenerator, Callable, Coroutine, Iterator
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from pathlib import Path
 from typing import Any, BinaryIO, Literal
 
@@ -54,6 +54,7 @@ from deerflow.skills.security_static_scanner import (
 )
 from deerflow.skills.storage import SkillStorage, get_or_new_user_skill_storage
 from deerflow.skills.types import SKILL_MD_FILE, SkillCategory
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 from deerflow.utils.file_io import await_drained
 from deerflow.utils.thread_id import ThreadId
 
@@ -347,7 +348,11 @@ def _load_user_skills(config: AppConfig) -> list[Skill]:
 
 
 def _copy_uploaded_skill_archive(source: BinaryIO) -> Path:
-    """Copy an uploaded archive to a bounded temporary file off the event loop."""
+    """Copy an uploaded archive to a bounded temporary file off the event loop.
+
+    A size-limit or read failure is the error the caller sees. Removing the
+    partial temporary is best-effort and must not replace that error.
+    """
     destination: Path | None = None
     try:
         with tempfile.NamedTemporaryFile(prefix="deerflow-skill-", suffix=".skill", delete=False) as target:
@@ -361,7 +366,7 @@ def _copy_uploaded_skill_archive(source: BinaryIO) -> Path:
         return destination
     except Exception:
         if destination is not None:
-            destination.unlink(missing_ok=True)
+            cleanup_failed_atomic_write(destination)
         raise
 
 
@@ -578,10 +583,13 @@ async def upload_and_install_skill(
     except MultiPartException as e:
         raise HTTPException(status_code=400, detail=e.message) from e
     finally:
-        if form is not None:
-            await form.close()
+        # Discard before close. SpooledTemporaryFile.close can raise while
+        # flushing, which would mask the install outcome and skip the discard.
         if temporary_path is not None:
-            await asyncio.to_thread(temporary_path.unlink, missing_ok=True)
+            await asyncio.to_thread(cleanup_failed_atomic_write, temporary_path)
+        if form is not None:
+            with suppress(Exception):
+                await form.close()
 
 
 @router.post(
diff --git a/backend/packages/harness/deerflow/extensions/AGENTS.md b/backend/packages/harness/deerflow/extensions/AGENTS.md
index bf6389fe83..0401949422 100644
--- a/backend/packages/harness/deerflow/extensions/AGENTS.md
+++ b/backend/packages/harness/deerflow/extensions/AGENTS.md
@@ -64,7 +64,14 @@ full dependency resolve invites a second interrupt that escapes the handler mid-
 File snapshots capture the original permission bits alongside the bytes; rollback applies
 that mode to a temporary file before atomically replacing the destination, including when
 restoring `config.yaml` after a failed removal. A failed write, chmod, or replace leaves the
-current destination untouched and removes the temporary file.
+current destination untouched. Temporary cleanup uses `cleanup_failed_atomic_write`, so a
+failed unlink or its warning log cannot replace the original error, including interruptions.
+Pass the raw descriptor only until `fdopen` succeeds; a close failure still attempts
+the unlink. The managed `plugins:` rewrite uses the same cleanup and records the
+temporary path before writing, so a short write still removes the hidden file.
+Restoring a file that did not exist uses the same helper, so an unlink or
+warning failure cannot replace the original operation error or skip later
+restores. Coverage: `tests/test_extension_atomic_write_cleanup.py`.
 Package mutation is deferred from environment mutation: after `uv add/remove`
 updates the declaration and lock, one `uv sync --locked --all-packages` preserves the same
 config-/environment-detected optional extras as normal startup. All three uv calls pin the
diff --git a/backend/packages/harness/deerflow/extensions/manager.py b/backend/packages/harness/deerflow/extensions/manager.py
index 9085d6d697..7d2bd2b229 100644
--- a/backend/packages/harness/deerflow/extensions/manager.py
+++ b/backend/packages/harness/deerflow/extensions/manager.py
@@ -24,6 +24,7 @@ from typing import Any
 import yaml
 from packaging.requirements import InvalidRequirement, Requirement
 
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 from deerflow.utils.file_lock import retry_until_locked as _retry_until_locked
 
 logger = logging.getLogger(__name__)
@@ -103,11 +104,15 @@ def _write_bytes_atomically(path: Path, content: bytes, mode: int | None) -> Non
     ``Path.write_bytes`` opens with ``O_TRUNC``, so a rollback that dies midway
     leaves pyproject.toml / uv.lock / config.yaml half written — the failed
     install would then take the checkout with it. The rest of the codebase
-    already publishes through a temporary file plus ``os.replace``.
+    already publishes through a temporary file plus ``os.replace``. Cleanup of
+    that temporary must not replace the publication error.
     """
-    fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
+    fd, temporary_name = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
+    temporary = Path(temporary_name)
     try:
-        with os.fdopen(fd, "wb") as handle:
+        handle = os.fdopen(fd, "wb")
+        fd = None  # The file object now owns the descriptor.
+        with handle:
             handle.write(content)
             handle.flush()
             if mode is not None:
@@ -115,7 +120,7 @@ def _write_bytes_atomically(path: Path, content: bytes, mode: int | None) -> Non
             os.fsync(handle.fileno())
         os.replace(temporary, path)
     except BaseException:
-        Path(temporary).unlink(missing_ok=True)
+        cleanup_failed_atomic_write(temporary, fd)
         raise
 
 
@@ -134,7 +139,9 @@ class _FileSnapshot:
 
     def restore(self) -> None:
         if self.content is None:
-            self.path.unlink(missing_ok=True)
+            # Rollback loops call this for every snapshot. A bare unlink can
+            # replace the original operation error and skip later restores.
+            cleanup_failed_atomic_write(self.path)
         else:
             _write_bytes_atomically(self.path, self.content, self.mode)
 
@@ -993,14 +1000,14 @@ def _write_plugins_block(path: Path, original: str, plugins: list[Any]) -> None:
             prefix=f".{path.name}.",
             delete=False,
         ) as stream:
-            stream.write(updated)
             temporary = Path(stream.name)
+            stream.write(updated)
         os.chmod(temporary, mode)
         os.replace(temporary, path)
         temporary = None
     finally:
         if temporary is not None:
-            temporary.unlink(missing_ok=True)
+            cleanup_failed_atomic_write(temporary)
 
 
 def _plugins_block_span(original: str) -> tuple[int, int] | None:
diff --git a/backend/packages/harness/deerflow/skills/AGENTS.md b/backend/packages/harness/deerflow/skills/AGENTS.md
index 76f78c9936..d5dd94eab6 100644
--- a/backend/packages/harness/deerflow/skills/AGENTS.md
+++ b/backend/packages/harness/deerflow/skills/AGENTS.md
@@ -21,6 +21,7 @@
 - **Tool policy (compatibility metadata)**: `allowed-tools` declarations are parsed, validated, exported, and described for interop with other agent frameworks, but they never filter model-visible schemas, refuse calls, narrow a subagent, or rewrite `tool_search` promotions. Real tool authority lives in Agent configuration and authorization: assembly-time provider filtering, runtime Guardrail enforcement, guardrail allowlists, and the deferred-promotion catalog. A lead custom Agent's explicit `skills` list is still enforced at the sandbox filesystem layer (see Sandbox projection); subagent skill lists scope discovery and activation only because concurrently delegated subagents share the lead thread sandbox. Subagents render skill discovery metadata at startup and reuse the same `SkillActivationMiddleware` as the lead; their configured `skills` field limits discovery and activation, not tools.
 - **Sandbox projection**: `skills/projection.py` materializes enabled-only shared trees at `{base_dir}/skills_view/public` and `{base_dir}/users/{user_id}/skills_view/{custom,legacy,integrations}`. A lead Agent with an explicit `skills` allowlist (including `[]`) gets the intersection of enabled public/user-visible skills and that allowlist at `{base_dir}/users/{user_id}/threads/{thread_id}/skills_view/{public,custom,legacy,integrations}`. `skills=None` keeps the shared zero-copy mount until a thread has used an explicit policy; later unrestricted runs repopulate the same stable thread root with all enabled skills (with host bash enabled, `LocalSandboxProvider` mounts the shared views instead). Rebuilds sign source state, view state, and normalized policy in a manifest, revoke every old category before adding the new policy, stage copies in temporary directories, and atomically replace files. Category root inodes stay stable for live bind mounts; concurrent readers can briefly see fewer skills during a policy change, never a skill revoked by the new policy. Policy-scoped copies reject absolute symlinks and relative symlinks that resolve outside their own skill package, preventing a permitted package from linking back to an omitted source. It copies files into the view (`_copy_into_view`) so a sandbox write cannot mutate the canonical skill inode; the operational trade-off is an O(total bytes) I/O and per-user/thread storage multiplier across rebuilds, prioritized for write isolation over zero-copy hardlinks. Steady-state freshness checks combine source and view metadata tree digests, so in-sandbox view tampering is detected and repaired on the next acquire. Storage writes, archive installs, deletes, and toggles rebuild shared scopes under a cross-process lock; Gateway boot ensures only the shared public view, user views are repaired lazily on acquire, and Agent thread views are recomputed before sandbox reuse. Managed integration packages are global, but their projected category is per-user because enabled state is isolated. User projection rebuilds re-read global enable state from disk instead of the process singleton, so a toggle handled by another Gateway worker is reflected on the next acquire. Gateway public-skill toggles take the public projection lock before the shared `extensions_config_write_lock`, re-read an existing config raw from disk, change only that skill's entry, and rebuild before responding; keep this as one worker-owned critical section so MCP writes cannot interleave and request cancellation cannot release either lock while the worker still runs. The shared public steady-state signature check runs without the global projection lock; stale/error paths take the lock and re-check before rebuilding or clearing. User and thread scope checks are serialized per scope. Projection failures clear the affected view before raising.
 - **Projection manifest failures**: `_write_manifest` keeps UTF-8 JSON, private `mkstemp` permissions and same-directory replacement. On ordinary errors or interruptions, reuse `utils.atomic_write.cleanup_failed_atomic_write`; transfer descriptor ownership only after `fdopen` succeeds. Descriptor close, sibling unlink and warning logs must not mask the original error. Coverage: `tests/test_skill_projection_manifest_cleanup.py`.
+- **Custom skill writes**: `LocalSkillStorage.write_custom_skill` and `UserScopedSkillStorage.write_custom_skill` publish through a same-directory temporary and `Path.replace`, then make the written path sandbox-readable. Cleanup uses `cleanup_failed_atomic_write`, including after a successful replace has already consumed the temporary name, so an unlink or warning failure cannot replace the write error. `_write_skill_states` keeps its indented JSON and transfers the descriptor only after `fdopen` succeeds; cleanup runs for `BaseException`, including `KeyboardInterrupt`, and warnings replace the previous silent `OSError` swallow. Coverage: `tests/test_skill_storage_atomic_write_cleanup.py`.
 - **Injection (legacy / default)**: Enabled skills are listed in the agent system prompt with full metadata and container paths (`<available_skills>` block). Controlled by `skills.deferred_discovery: false` (default).
 - **Deferred discovery** (`skills.deferred_discovery: true`): Skills are listed by name only in a compact `<skill_index>` block, keeping the system prompt prefix-cache friendly. The agent calls the `describe_skill` tool at runtime to fetch full metadata for skills it wants to use, then loads the SKILL.md via `read_file`. Two new modules support this path:
   - `skills/catalog.py` — immutable `SkillCatalog`; query forms: `select:a,b`, `+prefix`, free-text intent terms. Ranked queries use up to 256 characters / 16 unique literal terms, preferring name matches over description-only matches at equal coverage; ties keep catalog order. A lazy per-catalog index caches normalized names/descriptions. Unlike tool search, ranking uses literal intent terms, not regexes. `select:` is parsed before search limits and returns all exact catalog matches without query-length or result caps; other modes cap at `MAX_RESULTS=5`.
diff --git a/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py b/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py
index a0d189bcf8..8eeca6ab17 100644
--- a/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py
+++ b/backend/packages/harness/deerflow/skills/storage/local_skill_storage.py
@@ -17,6 +17,7 @@ from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
 from deerflow.skills.permissions import make_skill_written_path_sandbox_readable
 from deerflow.skills.storage.skill_storage import SKILL_MD_FILE, SkillStorage, walk_skill_directories
 from deerflow.skills.types import SkillCategory
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 
 logger = logging.getLogger(__name__)
 
@@ -115,7 +116,7 @@ class LocalSkillStorage(SkillStorage):
                 make_skill_written_path_sandbox_readable(self.get_custom_skill_dir(name), target)
             finally:
                 if tmp_path is not None:
-                    tmp_path.unlink(missing_ok=True)
+                    cleanup_failed_atomic_write(tmp_path)
 
     def remove_custom_skill_file(self, name: str, relative_path: str) -> str | None:
         removal = ((SkillCategory.CUSTOM, Path(name)),)
diff --git a/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py b/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py
index 9355a66cdf..ec055fef76 100644
--- a/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py
+++ b/backend/packages/harness/deerflow/skills/storage/user_scoped_skill_storage.py
@@ -45,6 +45,7 @@ from deerflow.skills.permissions import make_skill_written_path_sandbox_readable
 from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
 from deerflow.skills.storage.skill_storage import SKILL_MD_FILE, walk_skill_directories
 from deerflow.skills.types import SkillCategory
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 
 logger = logging.getLogger(__name__)
 
@@ -132,15 +133,13 @@ class UserScopedSkillStorage(LocalSkillStorage):
         )
         tmp_path = Path(tmp_path_str)
         try:
-            with os.fdopen(fd, "w", encoding="utf-8") as f:
-                json.dump(states, f, indent=2)
+            handle = os.fdopen(fd, "w", encoding="utf-8")
+            fd = None  # The file object now owns the descriptor.
+            with handle:
+                json.dump(states, handle, indent=2)
             tmp_path.replace(self._skill_states_file)
-        except Exception:
-            # Best-effort cleanup of the temp file on failure.
-            try:
-                tmp_path.unlink(missing_ok=True)
-            except OSError:
-                pass
+        except BaseException:
+            cleanup_failed_atomic_write(tmp_path, fd)
             raise
 
     def get_skill_enabled_state(self, skill_name: str) -> bool:
@@ -420,7 +419,7 @@ class UserScopedSkillStorage(LocalSkillStorage):
                 make_skill_written_path_sandbox_readable(self.get_custom_skill_dir(name), target)
             finally:
                 if tmp_path is not None:
-                    tmp_path.unlink(missing_ok=True)
+                    cleanup_failed_atomic_write(tmp_path)
 
     # ------------------------------------------------------------------
     # Public helpers
diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md
index 6b0757132b..326ba0e701 100644
--- a/backend/packages/harness/deerflow/utils/AGENTS.md
+++ b/backend/packages/harness/deerflow/utils/AGENTS.md
@@ -7,8 +7,12 @@ umask; replacements preserve existing permission bits before publication. Never
 read the process umask by changing it in this threaded runtime. Behavioral tests
 for both tool imports live in `tests/test_browser_outputs_atomic.py`.
 `cleanup_failed_atomic_write` shares best-effort cleanup across this publisher,
-`skills/projection.py`'s private manifest writer, project document staging, and
-Gateway upload staging.
+`skills/projection.py`'s private manifest writer, project document staging,
+extension snapshot rollback (including removal of a file that did not exist),
+the managed plugins-block rewrite, custom-skill file writes, per-user
+skill-state writes, Gateway upload staging, and the Gateway skill-archive
+upload copy and its post-install discard. The upload route discards that
+archive before closing the multipart form and suppresses close errors.
 Use it on publication failure or after deciding to discard an owned
 temporary; pass a raw descriptor only until `fdopen` succeeds, or an explicitly
 transferred duplicate descriptor owned by the cleanup caller. Shelf discards
diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md
index c0725dcf30..71dd0e9538 100644
--- a/backend/tests/AGENTS.md
+++ b/backend/tests/AGENTS.md
@@ -105,6 +105,22 @@ Browserless reservation cleanup tests require a warning with the leftover path
 and unlink error while preserving the original publication exception, including
 interruption. Successful cleanup stays quiet.
 
+Extension snapshot rollback and the managed plugins-block rewrite preserve the
+original publish error when temporary cleanup or its warning log fails.
+`tests/test_extension_atomic_write_cleanup.py` covers both writers, including
+interruptions, an `fdopen` failure whose descriptor close also fails, a plugins
+rewrite that fails while writing, a successful plugins rewrite's bytes and
+mode, and a delete-branch restore whose unlink failure still lets a later
+snapshot restore.
+Skill archive upload copies do the same in
+`tests/test_skill_archive_upload_cleanup.py`: a size-limit or read error stays
+visible, a successful copy returns the original bytes, and the upload route's
+post-install discard cannot replace a successful install or an install error.
+A form-close failure cannot replace that outcome or skip the discard.
+Custom-skill file writes and per-user skill-state writes do the same in
+`tests/test_skill_storage_atomic_write_cleanup.py`, including
+`KeyboardInterrupt` during a skill-state replace.
+
 `blocking_io/test_web_tool_url_validation.py` resolves a synthetic `.invalid`
 hostname to loopback by patching `_socket.getaddrinfo`, below the real
 `socket.getaddrinfo` wrapper. Do not replace that wrapper or the production URL
diff --git a/backend/tests/test_extension_atomic_write_cleanup.py b/backend/tests/test_extension_atomic_write_cleanup.py
new file mode 100644
index 0000000000..9d9efb24d1
--- /dev/null
+++ b/backend/tests/test_extension_atomic_write_cleanup.py
@@ -0,0 +1,315 @@
+# SPDX-License-Identifier: MIT
+"""Extension writers keep the original publish error when temporary cleanup fails."""
+
+from __future__ import annotations
+
+import logging
+import os
+import stat
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from deerflow.extensions.manager import _FileSnapshot, _write_plugins_block
+
+
+@pytest.mark.parametrize("primary_type", [OSError, KeyboardInterrupt])
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+def test_snapshot_restore_preserves_publish_error_when_cleanup_fails(tmp_path, monkeypatch, caplog, primary_type, failure):
+    target = tmp_path / "pyproject.toml"
+    target.write_bytes(b"mutated-before-rollback")
+    if os.name != "nt":
+        target.chmod(0o640)
+    mode = stat.S_IMODE(target.stat().st_mode)
+    snapshot = _FileSnapshot(target, b"original-bytes", 0o644 if os.name != "nt" else None)
+
+    primary = primary_type("publish failed")
+    staged: list[Path] = []
+    real_replace = os.replace
+    real_unlink = Path.unlink
+
+    def replace(source, destination):
+        staged.append(Path(source))
+        if Path(destination) == target:
+            raise primary
+        return real_replace(source, destination)
+
+    monkeypatch.setattr(os, "replace", replace)
+    if failure in {"unlink", "logging"}:
+        # The staged path is known only once replace runs; patch unlink lazily.
+        def unlink(path, *args, **kwargs):
+            if staged and path == staged[0]:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+        with pytest.raises(primary_type) as raised:
+            snapshot.restore()
+
+    assert raised.value is primary
+    assert target.read_bytes() == b"mutated-before-rollback"
+    assert stat.S_IMODE(target.stat().st_mode) == mode
+    assert len(staged) == 1
+    if failure == "none":
+        assert not staged[0].exists()
+        assert "Failed to remove temporary output file" not in caplog.text
+    else:
+        assert staged[0].read_bytes() == b"original-bytes"
+        if failure == "unlink":
+            assert "Failed to remove temporary output file" in caplog.text
+            assert str(staged[0]) in caplog.text
+            assert "temporary unlink failed" in caplog.text
+    real_unlink(staged[0], missing_ok=True)
+
+
+@pytest.mark.parametrize("primary_type", [OSError, KeyboardInterrupt])
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+def test_plugins_rewrite_preserves_publish_error_when_cleanup_fails(tmp_path, monkeypatch, caplog, primary_type, failure):
+    target = tmp_path / "config.yaml"
+    original = "kept: true\n"
+    target.write_text(original, encoding="utf-8")
+    if os.name != "nt":
+        target.chmod(0o640)
+    mode = stat.S_IMODE(target.stat().st_mode)
+
+    primary = primary_type("publish failed")
+    real_replace = os.replace
+    real_unlink = Path.unlink
+    staged: list[Path] = []
+
+    def replace(source, destination):
+        staged.append(Path(source))
+        if Path(destination) == target:
+            raise primary
+        return real_replace(source, destination)
+
+    monkeypatch.setattr(os, "replace", replace)
+    if failure in {"unlink", "logging"}:
+
+        def unlink(path, *args, **kwargs):
+            if staged and path == staged[0]:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+        with pytest.raises(primary_type) as raised:
+            _write_plugins_block(target, original, [{"name": "demo", "enabled": True}])
+
+    assert raised.value is primary
+    assert target.read_text(encoding="utf-8") == original
+    assert stat.S_IMODE(target.stat().st_mode) == mode
+    assert len(staged) == 1
+    if failure == "none":
+        assert not staged[0].exists()
+        assert "Failed to remove temporary output file" not in caplog.text
+    else:
+        assert "name: demo" in staged[0].read_text(encoding="utf-8")
+        if failure == "unlink":
+            assert "Failed to remove temporary output file" in caplog.text
+            assert str(staged[0]) in caplog.text
+    real_unlink(staged[0], missing_ok=True)
+
+
+def test_plugins_rewrite_success_keeps_neighbor_bytes_and_mode(tmp_path):
+    target = tmp_path / "config.yaml"
+    target.write_bytes(b"other: 1\n")
+    if os.name != "nt":
+        target.chmod(0o640)
+    mode = stat.S_IMODE(target.stat().st_mode)
+
+    _write_plugins_block(target, "other: 1\n", [{"name": "demo", "enabled": True}])
+
+    text = target.read_text(encoding="utf-8")
+    assert text.startswith("other: 1\n")
+    assert "name: demo" in text
+    assert stat.S_IMODE(target.stat().st_mode) == mode
+    assert list(tmp_path.glob(".config.yaml.*")) == []
+
+
+@pytest.mark.parametrize("close_fails", [False, True])
+@pytest.mark.parametrize("unlink_fails", [False, True])
+def test_snapshot_fdopen_failure_closes_descriptor_without_masking(tmp_path, monkeypatch, caplog, close_fails, unlink_fails):
+    from deerflow.extensions import manager as extension_manager
+
+    target = tmp_path / "pyproject.toml"
+    target.write_bytes(b"kept")
+    snapshot = _FileSnapshot(target, b"original-bytes", None)
+    primary = OSError("fdopen failed")
+    allocated: list[tuple[int, Path]] = []
+    closes: list[int] = []
+    real_mkstemp = tempfile.mkstemp
+    real_fdopen = os.fdopen
+    real_close = os.close
+    real_unlink = Path.unlink
+
+    def mkstemp(*args, **kwargs):
+        fd, name = real_mkstemp(*args, **kwargs)
+        allocated.append((fd, Path(name)))
+        return fd, name
+
+    def fdopen(fd, *args, **kwargs):
+        if allocated and fd == allocated[-1][0]:
+            raise primary
+        return real_fdopen(fd, *args, **kwargs)
+
+    def close(fd):
+        closes.append(fd)
+        real_close(fd)
+        if close_fails and allocated and fd == allocated[-1][0]:
+            raise OSError("descriptor cleanup failed")
+
+    def unlink(path, *args, **kwargs):
+        if unlink_fails and allocated and path == allocated[-1][1]:
+            raise OSError("temporary unlink failed")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(extension_manager.tempfile, "mkstemp", mkstemp)
+    monkeypatch.setattr(os, "fdopen", fdopen)
+    monkeypatch.setattr(os, "close", close)
+    monkeypatch.setattr(Path, "unlink", unlink)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            with pytest.raises(OSError) as raised:
+                snapshot.restore()
+        assert raised.value is primary
+        fd, temporary = allocated[0]
+        assert fd in closes
+        assert target.read_bytes() == b"kept"
+        if unlink_fails:
+            assert temporary.exists()
+            assert "Failed to remove temporary output file" in caplog.text
+        else:
+            assert not temporary.exists()
+        if close_fails:
+            assert "Failed to close temporary output descriptor" in caplog.text
+            assert str(temporary) in caplog.text
+        else:
+            assert "Failed to close temporary output descriptor" not in caplog.text
+    finally:
+        for fd, temporary in allocated:
+            real_unlink(temporary, missing_ok=True)
+
+
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+def test_plugins_short_write_still_removes_the_temporary(tmp_path, monkeypatch, caplog, failure):
+    from deerflow.extensions import manager as extension_manager
+
+    target = tmp_path / "config.yaml"
+    original = "kept: true\n"
+    target.write_text(original, encoding="utf-8")
+    primary = OSError("disk full")
+    created: list[Path] = []
+    real_named = tempfile.NamedTemporaryFile
+    real_unlink = Path.unlink
+
+    def tracking(*args, **kwargs):
+        handle = real_named(*args, **kwargs)
+        created.append(Path(handle.name))
+
+        def fail_write(_data):
+            raise primary
+
+        handle.write = fail_write
+        return handle
+
+    monkeypatch.setattr(extension_manager.tempfile, "NamedTemporaryFile", tracking)
+    if failure in {"unlink", "logging"}:
+
+        def unlink(path, *args, **kwargs):
+            if created and path == created[0]:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            with pytest.raises(OSError) as raised:
+                _write_plugins_block(target, original, [{"name": "demo", "enabled": True}])
+        assert raised.value is primary
+        assert target.read_text(encoding="utf-8") == original
+        assert len(created) == 1
+        if failure == "none":
+            assert not created[0].exists()
+            assert "Failed to remove temporary output file" not in caplog.text
+        else:
+            assert created[0].exists()
+            if failure == "unlink":
+                assert "Failed to remove temporary output file" in caplog.text
+                assert str(created[0]) in caplog.text
+    finally:
+        for path in created:
+            real_unlink(path, missing_ok=True)
+
+
+@pytest.mark.parametrize("failure", ["unlink", "logging"])
+def test_snapshot_delete_branch_cleanup_failure_does_not_skip_later_restores(tmp_path, monkeypatch, caplog, failure):
+    created = tmp_path / "uv.lock"
+    kept = tmp_path / "pyproject.toml"
+    kept.write_bytes(b"mutated")
+    created.write_bytes(b"created-by-install")
+    missing = _FileSnapshot(created, None, None)
+    later = _FileSnapshot(kept, b"original", 0o644 if os.name != "nt" else None)
+    operation_error = RuntimeError("install failed")
+
+    real_unlink = Path.unlink
+
+    def unlink(path, *args, **kwargs):
+        if path == created:
+            raise OSError("temporary unlink failed")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    def _rollback():
+        try:
+            raise operation_error
+        except RuntimeError:
+            for snapshot in (missing, later):
+                snapshot.restore()
+            raise
+
+    with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+        with pytest.raises(RuntimeError) as raised:
+            _rollback()
+
+    assert raised.value is operation_error
+    assert created.read_bytes() == b"created-by-install"
+    assert kept.read_bytes() == b"original"
+    if failure == "unlink":
+        assert "Failed to remove temporary output file" in caplog.text
+        assert str(created) in caplog.text
+        assert "temporary unlink failed" in caplog.text
+    if failure == "logging":
+        assert str(raised.value) == "install failed"
+        assert "log handler failed" not in str(raised.value)
diff --git a/backend/tests/test_skill_archive_upload_cleanup.py b/backend/tests/test_skill_archive_upload_cleanup.py
new file mode 100644
index 0000000000..164cdc37fa
--- /dev/null
+++ b/backend/tests/test_skill_archive_upload_cleanup.py
@@ -0,0 +1,238 @@
+# SPDX-License-Identifier: MIT
+"""Skill archive upload copies keep the original error when temporary cleanup fails."""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from io import BytesIO
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from app.gateway.routers import skills as skills_router
+
+
+class _FailingRead:
+    def __init__(self, error: BaseException):
+        self._error = error
+
+    def read(self, _size: int) -> bytes:
+        raise self._error
+
+
+@pytest.mark.parametrize("kind", ["read", "limit"])
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+def test_copy_preserves_original_error_when_cleanup_fails(monkeypatch, caplog, kind, failure):
+    created: list[Path] = []
+    real_named = tempfile.NamedTemporaryFile
+    real_unlink = Path.unlink
+    primary = OSError("read failed")
+
+    def tracking(*args, **kwargs):
+        handle = real_named(*args, **kwargs)
+        created.append(Path(handle.name))
+        return handle
+
+    monkeypatch.setattr(skills_router.tempfile, "NamedTemporaryFile", tracking)
+    if kind == "limit":
+        monkeypatch.setattr(skills_router, "_MAX_SKILL_ARCHIVE_UPLOAD_BYTES", 4)
+        source = BytesIO(b"12345")
+        expected_type = skills_router._SkillArchiveUploadTooLargeError
+    else:
+        source = _FailingRead(primary)
+        expected_type = OSError
+    if failure in {"unlink", "logging"}:
+
+        def unlink(path, *args, **kwargs):
+            if created and path == created[0]:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            with pytest.raises(expected_type) as raised:
+                skills_router._copy_uploaded_skill_archive(source)
+        if kind == "read":
+            assert raised.value is primary
+        else:
+            assert "upload limit" in str(raised.value)
+        assert "temporary unlink failed" not in str(raised.value)
+        assert len(created) == 1
+        if failure == "none":
+            assert not created[0].exists()
+            assert "Failed to remove temporary output file" not in caplog.text
+        else:
+            assert created[0].exists()
+            if failure == "unlink":
+                assert "Failed to remove temporary output file" in caplog.text
+                assert str(created[0]) in caplog.text
+                assert "temporary unlink failed" in caplog.text
+    finally:
+        for path in created:
+            real_unlink(path, missing_ok=True)
+
+
+def test_copy_success_returns_the_uploaded_bytes():
+    payload = b"PK\x03\x04exact-skill-bytes"
+    destination = skills_router._copy_uploaded_skill_archive(BytesIO(payload))
+    try:
+        assert destination.read_bytes() == payload
+        assert destination.name.startswith("deerflow-skill-")
+        assert destination.suffix == ".skill"
+    finally:
+        destination.unlink(missing_ok=True)
+
+
+class _Form:
+    def __init__(self, *, close_error: BaseException | None = None) -> None:
+        self.archive = skills_router.UploadFile(BytesIO(b"unused"), filename="demo.skill")
+        self.close_error = close_error
+        self.closed = False
+
+    async def close(self) -> None:
+        self.closed = True
+        if self.close_error is not None:
+            raise self.close_error
+        await self.archive.close()
+
+    def get(self, _key: str) -> skills_router.UploadFile:
+        return self.archive
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("outcome", ["success", "install_error"])
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+async def test_upload_finally_preserves_install_outcome(tmp_path, monkeypatch, caplog, outcome, failure):
+    archive = tmp_path / "uploaded.skill"
+    archive.write_bytes(b"archive-bytes")
+    response = skills_router.SkillInstallResponse(success=True, skill_name="demo", message="installed")
+    install_error = RuntimeError("install failed")
+
+    async def _admin(*_args, **_kwargs):
+        return None
+
+    async def _parse(_request):
+        return _Form()
+
+    def _copy(_source):
+        return archive
+
+    async def _install(path, _config):
+        assert path.read_bytes() == b"archive-bytes"
+        if outcome == "install_error":
+            raise install_error
+        return response
+
+    monkeypatch.setattr(skills_router, "require_admin_user", _admin)
+    monkeypatch.setattr(skills_router, "_parse_skill_archive_form", _parse)
+    monkeypatch.setattr(skills_router, "_copy_uploaded_skill_archive", _copy)
+    monkeypatch.setattr(skills_router, "_install_skill_archive", _install)
+
+    real_unlink = Path.unlink
+    if failure in {"unlink", "logging"}:
+
+        def unlink(path, *args, **kwargs):
+            if path == archive:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            if outcome == "success":
+                result = await skills_router.upload_and_install_skill(SimpleNamespace(), SimpleNamespace())
+                assert result == response
+            else:
+                with pytest.raises(RuntimeError) as raised:
+                    await skills_router.upload_and_install_skill(SimpleNamespace(), SimpleNamespace())
+                assert raised.value is install_error
+        if failure == "none":
+            assert not archive.exists()
+            assert "Failed to remove temporary output file" not in caplog.text
+        else:
+            assert archive.exists()
+            if failure == "unlink":
+                assert "Failed to remove temporary output file" in caplog.text
+                assert str(archive) in caplog.text
+                assert "temporary unlink failed" in caplog.text
+    finally:
+        real_unlink(archive, missing_ok=True)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("outcome", ["success", "install_error"])
+@pytest.mark.parametrize("failure", ["none", "unlink"])
+async def test_upload_finally_close_failure_still_discards_and_preserves_outcome(tmp_path, monkeypatch, caplog, outcome, failure):
+    archive = tmp_path / "uploaded.skill"
+    archive.write_bytes(b"archive-bytes")
+    response = skills_router.SkillInstallResponse(success=True, skill_name="demo", message="installed")
+    install_error = RuntimeError("install failed")
+    close_error = RuntimeError("form close failed")
+    form = _Form(close_error=close_error)
+
+    async def _admin(*_args, **_kwargs):
+        return None
+
+    async def _parse(_request):
+        return form
+
+    def _copy(_source):
+        return archive
+
+    async def _install(path, _config):
+        assert path.read_bytes() == b"archive-bytes"
+        if outcome == "install_error":
+            raise install_error
+        return response
+
+    monkeypatch.setattr(skills_router, "require_admin_user", _admin)
+    monkeypatch.setattr(skills_router, "_parse_skill_archive_form", _parse)
+    monkeypatch.setattr(skills_router, "_copy_uploaded_skill_archive", _copy)
+    monkeypatch.setattr(skills_router, "_install_skill_archive", _install)
+
+    real_unlink = Path.unlink
+    if failure == "unlink":
+
+        def unlink(path, *args, **kwargs):
+            if path == archive:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            if outcome == "success":
+                result = await skills_router.upload_and_install_skill(SimpleNamespace(), SimpleNamespace())
+                assert result == response
+            else:
+                with pytest.raises(RuntimeError) as raised:
+                    await skills_router.upload_and_install_skill(SimpleNamespace(), SimpleNamespace())
+                assert raised.value is install_error
+        assert form.closed
+        assert "form close failed" not in caplog.text
+        if failure == "none":
+            assert not archive.exists()
+        else:
+            assert archive.exists()
+            assert "Failed to remove temporary output file" in caplog.text
+            assert "temporary unlink failed" in caplog.text
+    finally:
+        real_unlink(archive, missing_ok=True)
diff --git a/backend/tests/test_skill_storage_atomic_write_cleanup.py b/backend/tests/test_skill_storage_atomic_write_cleanup.py
new file mode 100644
index 0000000000..f913014043
--- /dev/null
+++ b/backend/tests/test_skill_storage_atomic_write_cleanup.py
@@ -0,0 +1,198 @@
+# SPDX-License-Identifier: MIT
+"""Custom-skill writers keep the original save error when temporary cleanup fails."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from deerflow.config.paths import Paths
+from deerflow.skills.storage import get_or_new_skill_storage, reset_skill_storage
+from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
+
+
+@pytest.fixture(autouse=True)
+def _reset_storages():
+    reset_skill_storage()
+    yield
+    reset_skill_storage()
+
+
+@pytest.fixture()
+def storage(tmp_path):
+    return get_or_new_skill_storage(skills_path=str(tmp_path))
+
+
+@pytest.fixture()
+def user_storage(tmp_path):
+    with patch("deerflow.config.paths.get_paths", return_value=Paths(base_dir=tmp_path)):
+        with patch("deerflow.config.paths._paths", None):
+            created = UserScopedSkillStorage("test-user", host_path=str(tmp_path))
+    return created
+
+
+def _failing_replace(real_replace, destination_name: str, primary: BaseException, staged: list[Path]):
+    def replace(path, destination):
+        if Path(destination).name == destination_name and path.name != destination_name:
+            staged.append(path)
+            raise primary
+        return real_replace(path, destination)
+
+    return replace
+
+
+@pytest.mark.parametrize("storage_kind", ["local", "user"])
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+def test_write_custom_skill_preserves_replace_error_when_cleanup_fails(tmp_path, monkeypatch, caplog, storage, user_storage, storage_kind, failure):
+    selected = storage if storage_kind == "local" else user_storage
+    skill_dir = selected.get_custom_skill_dir("demo-skill")
+    target = skill_dir / "SKILL.md"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text("previous", encoding="utf-8")
+
+    primary = OSError("replace failed")
+    staged: list[Path] = []
+    real_replace = Path.replace
+    real_unlink = Path.unlink
+    monkeypatch.setattr(Path, "replace", _failing_replace(real_replace, "SKILL.md", primary, staged))
+    if failure in {"unlink", "logging"}:
+
+        def unlink(path, *args, **kwargs):
+            if staged and path == staged[0]:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            with pytest.raises(OSError) as raised:
+                selected.write_custom_skill("demo-skill", "SKILL.md", "replacement")
+        assert raised.value is primary
+        assert target.read_text(encoding="utf-8") == "previous"
+        assert len(staged) == 1
+        if failure == "none":
+            assert not staged[0].exists()
+            assert "Failed to remove temporary output file" not in caplog.text
+        else:
+            assert staged[0].exists()
+            if failure == "unlink":
+                assert "Failed to remove temporary output file" in caplog.text
+                assert str(staged[0]) in caplog.text
+                assert "temporary unlink failed" in caplog.text
+    finally:
+        for path in staged:
+            real_unlink(path, missing_ok=True)
+
+
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+def test_skill_states_preserve_save_error_when_cleanup_fails(monkeypatch, caplog, user_storage, failure):
+    target = user_storage._skill_states_file
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text('{"old-skill": {"enabled": false}}\n', encoding="utf-8")
+
+    primary = OSError("replace failed")
+    staged: list[Path] = []
+    real_replace = Path.replace
+    real_unlink = Path.unlink
+
+    def replace(path, destination):
+        if destination == target and path != target:
+            staged.append(path)
+            raise primary
+        return real_replace(path, destination)
+
+    monkeypatch.setattr(Path, "replace", replace)
+    if failure in {"unlink", "logging"}:
+
+        def unlink(path, *args, **kwargs):
+            if staged and path == staged[0]:
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            with pytest.raises(OSError) as raised:
+                user_storage._write_skill_states({"new-skill": {"enabled": True}})
+        assert raised.value is primary
+        assert target.read_text(encoding="utf-8") == '{"old-skill": {"enabled": false}}\n'
+        assert len(staged) == 1
+        if failure == "none":
+            assert not staged[0].exists()
+            assert "Failed to remove temporary output file" not in caplog.text
+        else:
+            assert staged[0].read_text(encoding="utf-8").strip().startswith("{")
+            if failure == "unlink":
+                assert "Failed to remove temporary output file" in caplog.text
+                assert str(staged[0]) in caplog.text
+    finally:
+        for path in staged:
+            real_unlink(path, missing_ok=True)
+
+
+@pytest.mark.parametrize("failure", ["none", "unlink", "logging"])
+def test_skill_states_keyboard_interrupt_still_cleans(monkeypatch, caplog, user_storage, failure):
+    target = user_storage._skill_states_file
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_text('{"old-skill": {"enabled": false}}\n', encoding="utf-8")
+
+    primary = KeyboardInterrupt()
+    staged: list[Path] = []
+    cleanup_unlinks: list[Path] = []
+    real_replace = Path.replace
+    real_unlink = Path.unlink
+
+    monkeypatch.setattr(Path, "replace", _failing_replace(real_replace, target.name, primary, staged))
+    if failure in {"unlink", "logging"}:
+
+        def unlink(path, *args, **kwargs):
+            if staged and path == staged[0]:
+                cleanup_unlinks.append(path)
+                raise OSError("temporary unlink failed")
+            return real_unlink(path, *args, **kwargs)
+
+        monkeypatch.setattr(Path, "unlink", unlink)
+    if failure == "logging":
+
+        def boom_log(*args, **kwargs):
+            raise RuntimeError("log handler failed")
+
+        monkeypatch.setattr("deerflow.utils.atomic_write.logger.warning", boom_log)
+
+    try:
+        with caplog.at_level(logging.WARNING, logger="deerflow.utils.atomic_write"):
+            with pytest.raises(KeyboardInterrupt) as raised:
+                user_storage._write_skill_states({"new-skill": {"enabled": True}})
+        assert raised.value is primary
+        assert target.read_text(encoding="utf-8") == '{"old-skill": {"enabled": false}}\n'
+        assert len(staged) == 1
+        if failure == "none":
+            assert not staged[0].exists()
+            assert "Failed to remove temporary output file" not in caplog.text
+        else:
+            assert staged[0].exists()
+            assert cleanup_unlinks == [staged[0]]
+            if failure == "unlink":
+                assert "Failed to remove temporary output file" in caplog.text
+                assert str(staged[0]) in caplog.text
+                assert "temporary unlink failed" in caplog.text
+    finally:
+        for path in staged:
+            real_unlink(path, missing_ok=True)