diff --git a/backend/packages/harness/deerflow/config/managed_models.py b/backend/packages/harness/deerflow/config/managed_models.py
index 0190a059ee..36fcfd3911 100644
--- a/backend/packages/harness/deerflow/config/managed_models.py
+++ b/backend/packages/harness/deerflow/config/managed_models.py
@@ -25,6 +25,7 @@ from deerflow.config.managed_model_providers import get_managed_model_provider,
 from deerflow.config.model_config import ModelConfig, ReasoningCapabilities
 from deerflow.config.runtime_paths import runtime_home
 from deerflow.config.secret_file import read_or_create_secret_file
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 
 if TYPE_CHECKING:
     from deerflow.config.app_config import AppConfig
@@ -224,7 +225,9 @@ class ManagedModelStore:
             os.replace(temporary, path)
         finally:
             if temporary is not None:
-                temporary.unlink(missing_ok=True)
+                # Discarding the temporary must not replace the outcome of the
+                # write (or of a failed ``os.replace``) with a cleanup error.
+                cleanup_failed_atomic_write(temporary)
 
 
 def merge_managed_models(config: AppConfig) -> AppConfig:
diff --git a/backend/packages/harness/deerflow/config/secret_file.py b/backend/packages/harness/deerflow/config/secret_file.py
index 5c645a6239..8e99d6a3b1 100644
--- a/backend/packages/harness/deerflow/config/secret_file.py
+++ b/backend/packages/harness/deerflow/config/secret_file.py
@@ -36,6 +36,8 @@ import time
 from collections.abc import Callable
 from pathlib import Path
 
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
+
 logger = logging.getLogger(__name__)
 
 _POLL_INTERVAL_SECONDS = 0.05
@@ -147,7 +149,11 @@ def _create_exclusive(path: Path, generate: Callable[[], str]) -> bool:
             return _publish_under_claim(path, generate, replace_empty=False)
         return True
     finally:
-        Path(temporary).unlink(missing_ok=True)
+        # Removing our own temporary name is best-effort on every exit path —
+        # whether the hard link published the value, a peer won the name, a
+        # hard-link-less filesystem published through the replacement claim, or
+        # an earlier step already failed — and must never mask that outcome.
+        cleanup_failed_atomic_write(Path(temporary))
 
 
 def _claim_path(path: Path) -> Path:
@@ -186,4 +192,6 @@ def _publish_under_claim(path: Path, generate: Callable[[], str], *, replace_emp
         return True
     finally:
         if not published:
-            claim.unlink(missing_ok=True)
+            # Dropping our unpublished claim is best-effort: a failure here must
+            # not replace the error that aborted the publication.
+            cleanup_failed_atomic_write(claim)
diff --git a/backend/tests/test_config_atomic_cleanup.py b/backend/tests/test_config_atomic_cleanup.py
new file mode 100644
index 0000000000..321a171b23
--- /dev/null
+++ b/backend/tests/test_config_atomic_cleanup.py
@@ -0,0 +1,92 @@
+"""Cleanup failures in the secret-file and managed-model writers.
+
+Both writers stage a temporary file and then discard it. That removal runs on
+every exit path, not only after a successful publication: it also runs when the
+write, the close, or the replacement already failed. On Windows a scanner, a peer
+holding the handle, or a read-only mount can make the removal itself fail, so it
+has to stay best-effort — a cleanup error must never replace the outcome the
+caller needed, whether that outcome is a published value or an earlier error.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from deerflow.config.managed_models import ManagedModelStore
+from deerflow.config.secret_file import read_or_create_secret_file
+
+
+def _fail_unlink_where(predicate: Callable[[Path], bool], monkeypatch: pytest.MonkeyPatch) -> list[Path]:
+    """Make ``Path.unlink`` fail for matching paths and record the attempts."""
+    real_unlink = Path.unlink
+    attempts: list[Path] = []
+
+    def unlink(self: Path, missing_ok: bool = False) -> None:
+        if predicate(self):
+            attempts.append(self)
+            raise PermissionError(13, "The process cannot access the file", str(self))
+        real_unlink(self, missing_ok=missing_ok)
+
+    monkeypatch.setattr(Path, "unlink", unlink)
+    return attempts
+
+
+def test_secret_creation_survives_a_failing_temporary_removal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    path = tmp_path / ".secret"
+    attempts = _fail_unlink_where(lambda candidate: candidate.suffix == ".tmp", monkeypatch)
+
+    value = read_or_create_secret_file(path, lambda: "generated-value")
+
+    assert value == "generated-value"
+    assert path.read_text(encoding="utf-8") == "generated-value"
+    assert attempts, "the temporary file should have been handed to the cleanup helper"
+
+
+def test_secret_creation_keeps_the_replace_error_when_cleanup_also_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    path = tmp_path / ".secret"
+    path.write_text("", encoding="utf-8")  # an abandoned empty file takes the claim path
+
+    def no_hard_links(*_args, **_kwargs):
+        raise OSError(95, "Operation not supported")
+
+    def failing_replace(*_args, **_kwargs):
+        raise OSError(5, "publish failed")
+
+    monkeypatch.setattr("deerflow.config.secret_file.os.link", no_hard_links)
+    monkeypatch.setattr("deerflow.config.secret_file.os.replace", failing_replace)
+    attempts = _fail_unlink_where(lambda candidate: candidate.suffix == ".replacing", monkeypatch)
+
+    with pytest.raises(OSError) as raised:
+        read_or_create_secret_file(path, lambda: "generated-value")
+
+    assert "publish failed" in str(raised.value)
+    assert attempts, "the unpublished claim should still have been cleaned up"
+
+
+def test_managed_model_write_survives_a_failing_temporary_removal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    target = tmp_path / "managed.json"
+    attempts = _fail_unlink_where(lambda candidate: candidate.parent == tmp_path and candidate != target, monkeypatch)
+
+    ManagedModelStore._write(target, b"payload")
+
+    assert target.read_bytes() == b"payload"
+    assert attempts
+
+
+def test_managed_model_write_keeps_the_replace_error_when_cleanup_also_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    target = tmp_path / "managed.json"
+
+    def failing_replace(*_args, **_kwargs):
+        raise OSError(5, "publish failed")
+
+    monkeypatch.setattr("deerflow.config.managed_models.os.replace", failing_replace)
+    attempts = _fail_unlink_where(lambda candidate: candidate.parent == tmp_path and candidate != target, monkeypatch)
+
+    with pytest.raises(OSError) as raised:
+        ManagedModelStore._write(target, b"payload")
+
+    assert "publish failed" in str(raised.value)
+    assert attempts