fix(persistence): keep staged-write cleanup best-effort (#7485)

backend/packages/harness/deerflow/persistence/agents/file.py

@@ -37,6 +37,7 @@ from deerflow.persistence.agents.base import (
     parse_agent_config,
 )
 from deerflow.runtime.user_context import DEFAULT_USER_ID
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 
 logger = logging.getLogger(__name__)
 
@@ -257,7 +258,9 @@ class FileAgentStore(AgentStore):
                 staged.remove(tmp)
         finally:
             for tmp in staged:
-                tmp.unlink(missing_ok=True)
+                # A failing discard must not replace the error that aborted the
+                # publish, nor the result of a successful one.
+                cleanup_failed_atomic_write(tmp)
 
 
 def _stage_temp(target: Path, text: str) -> Path:

backend/packages/harness/deerflow/persistence/managed_subagents/file.py

@@ -16,6 +16,7 @@ from deerflow.persistence.managed_subagents.base import (
     ManagedSubagentStore,
     normalize_managed_subagent_name,
 )
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 
 logger = logging.getLogger(__name__)
 _write_lock = threading.RLock()
@@ -97,4 +98,4 @@ class FileManagedSubagentStore(ManagedSubagentStore):
             tmp_path = None
         finally:
             if tmp_path is not None:
-                tmp_path.unlink(missing_ok=True)
+                cleanup_failed_atomic_write(tmp_path)

backend/tests/test_atomic_cleanup_census.py

@@ -0,0 +1,71 @@
+"""Staged-write cleanup at the two persistence writers must not replace the outcome.
+
+Each writer stages a temporary file and discards it in a ``finally`` that also
+runs when the publish itself failed. Before this change a failing discard
+replaced that error with its own.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def real_unlink():
+    return Path.unlink
+
+
+def _fail_unlink_on(monkeypatch, predicate, real_unlink):
+    """Make ``Path.unlink`` raise for the paths a test owns."""
+
+    def unlink(self, *args, **kwargs):
+        if predicate(self):
+            raise PermissionError("temporary unlink failed")
+        return real_unlink(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", unlink)
+
+
+def _is_staged(path: Path) -> bool:
+    return path.suffix == ".tmp"
+
+
+def test_managed_subagent_write_keeps_the_replace_error_when_cleanup_fails(tmp_path, monkeypatch, real_unlink):
+    from deerflow.persistence.managed_subagents import file as store_module
+    from deerflow.persistence.managed_subagents.base import ManagedSubagentDefinition
+
+    definition = ManagedSubagentDefinition(
+        name="planner",
+        description="plans",
+        system_prompt="plan things",
+    )
+    target = tmp_path / "planner.json"
+
+    def _failing_replace(*_args, **_kwargs):
+        raise OSError("replace failed")
+
+    monkeypatch.setattr(store_module.os, "replace", _failing_replace)
+    _fail_unlink_on(monkeypatch, _is_staged, real_unlink)
+
+    with pytest.raises(OSError, match="replace failed"):
+        store_module.FileManagedSubagentStore._atomic_write(target, definition)
+
+    assert not target.exists()
+
+
+def test_agent_store_write_keeps_the_publish_error_when_cleanup_fails(tmp_path, monkeypatch, real_unlink):
+    from deerflow.persistence.agents.file import FileAgentStore
+
+    agent_dir = tmp_path / "agents"
+    agent_dir.mkdir()
+
+    def _failing_replace(self, target, *args, **kwargs):
+        raise OSError("replace failed")
+
+    monkeypatch.setattr(Path, "replace", _failing_replace)
+    _fail_unlink_on(monkeypatch, _is_staged, real_unlink)
+
+    with pytest.raises(OSError, match="replace failed"):
+        FileAgentStore._write(agent_dir, {"name": "planner"}, "# SOUL")