diff --git a/backend/app/gateway/routers/artifacts.py b/backend/app/gateway/routers/artifacts.py
index f2f82ad321..d19afc02e4 100644
--- a/backend/app/gateway/routers/artifacts.py
+++ b/backend/app/gateway/routers/artifacts.py
@@ -25,6 +25,7 @@ from deerflow.config.paths import make_safe_user_id
 from deerflow.runtime import ConflictError, ThreadOperationKind
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.sandbox.sandbox_provider import get_sandbox_provider
+from deerflow.utils.atomic_write import cleanup_failed_atomic_write
 from deerflow.utils.file_io import await_drained
 from deerflow.utils.text_detection import _is_active_content_mime_type, is_text_file_by_content
 from deerflow.utils.thread_id import ThreadId
@@ -142,12 +143,10 @@ def _replace_artifact_atomically(actual_path: Path, content: bytes, file_stat: o
         # clearing the whole 256-entry LRU costs nothing (see PR review).
         _sha256_of_file_cached.cache_clear()
     finally:
-        if temp_fd >= 0:
-            os.close(temp_fd)
-        try:
-            temp_path.unlink()
-        except FileNotFoundError:
-            pass
+        # The helper keeps the published case silent (the temp name is already
+        # gone), warns when a leftover temp cannot be removed, and never lets a
+        # cleanup error replace the edit's outcome.
+        cleanup_failed_atomic_write(temp_path, temp_fd if temp_fd >= 0 else None)
 
 
 def _sync_artifact_to_sandbox(sandbox, virtual_path: str, content: bytes) -> None:
diff --git a/backend/packages/harness/deerflow/integrations/lark_cli.py b/backend/packages/harness/deerflow/integrations/lark_cli.py
index cfc788966c..434a0eb2e3 100644
--- a/backend/packages/harness/deerflow/integrations/lark_cli.py
+++ b/backend/packages/harness/deerflow/integrations/lark_cli.py
@@ -1445,7 +1445,7 @@ def _write_json_file_atomic(path: Path, payload: dict[str, Any]) -> None:
         temp_path.chmod(0o600)
         os.replace(temp_path, path)
     finally:
-        temp_path.unlink(missing_ok=True)
+        cleanup_failed_atomic_write(temp_path)
 
 
 _LARK_FLOW_DEVICE_CODE_UNSET: Any = object()
diff --git a/backend/tests/test_atomic_cleanup_writers.py b/backend/tests/test_atomic_cleanup_writers.py
new file mode 100644
index 0000000000..7b324c2c75
--- /dev/null
+++ b/backend/tests/test_atomic_cleanup_writers.py
@@ -0,0 +1,94 @@
+"""Staged-temp cleanup in the artifact and Lark writers must keep the real outcome.
+
+Both writers publish through a same-directory temporary and discard it on exit.
+A failing discard used to replace the error the caller needed — the artifact
+editor returned a cleanup ``OSError`` instead of the failed write, and the Lark
+flow-state writer did the same.
+"""
+
+from __future__ import annotations
+
+import os
+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 test_artifact_edit_keeps_the_publish_error_when_cleanup_fails(tmp_path, monkeypatch, real_unlink):
+    from app.gateway.routers import artifacts
+
+    target = tmp_path / "note.txt"
+    target.write_text("original", encoding="utf-8")
+    file_stat = os.stat(target)
+
+    def _failing_replace(*_args, **_kwargs):
+        raise OSError("replace failed")
+
+    monkeypatch.setattr(artifacts.os, "replace", _failing_replace)
+    _fail_unlink_on(monkeypatch, lambda path: path.name.startswith(".") and path.suffix == "", real_unlink)
+
+    with pytest.raises(OSError, match="replace failed"):
+        artifacts._replace_artifact_atomically(target, b"replacement", file_stat)
+
+    assert target.read_text(encoding="utf-8") == "original"
+
+
+def test_artifact_edit_success_is_quiet(tmp_path, caplog):
+    """The published path must not warn: its temp name is already gone."""
+    from app.gateway.routers import artifacts
+
+    target = tmp_path / "note.txt"
+    target.write_text("original", encoding="utf-8")
+    file_stat = os.stat(target)
+    caplog.clear()
+
+    with caplog.at_level("WARNING", logger="deerflow.utils.atomic_write"):
+        artifacts._replace_artifact_atomically(target, b"replacement", file_stat)
+
+    assert target.read_bytes() == b"replacement"
+    assert [p.name for p in tmp_path.iterdir() if p.name.startswith(artifacts._ARTIFACT_EDIT_TEMP_PREFIX)] == []
+    assert caplog.text == ""
+
+
+def test_lark_flow_state_write_keeps_the_write_error_when_cleanup_fails(tmp_path, monkeypatch, real_unlink):
+    from deerflow.integrations import lark_cli
+
+    target = tmp_path / "flow-state.json"
+
+    def _failing_replace(*_args, **_kwargs):
+        raise OSError("replace failed")
+
+    monkeypatch.setattr(lark_cli.os, "replace", _failing_replace)
+    _fail_unlink_on(monkeypatch, lambda path: path.name.startswith(".flow-state.json"), real_unlink)
+
+    with pytest.raises(OSError, match="replace failed"):
+        lark_cli._write_json_file_atomic(target, {"state": "pending"})
+
+    assert not target.exists()
+
+
+def test_lark_flow_state_write_still_succeeds(monkeypatch, tmp_path):
+    """The success path keeps its behaviour: payload written, no temp left."""
+    from deerflow.integrations import lark_cli
+
+    target = tmp_path / "flow-state.json"
+    lark_cli._write_json_file_atomic(target, {"state": "pending"})
+
+    assert target.read_text(encoding="utf-8") == '{"state": "pending"}\n'
+    assert [p.name for p in tmp_path.iterdir() if p.name.startswith(".flow-state")] == []