diff --git a/apps/observer/tests/test_utils.py b/apps/observer/tests/test_utils.py
index 65ee54d11..b7d6489cf 100644
--- a/apps/observer/tests/test_utils.py
+++ b/apps/observer/tests/test_utils.py
@@ -218,6 +218,61 @@ class TestIncrementStat:
         increment_stat("nonexistent", "segments_observed")
 
 
+class TestAtomicWriteCrashSafety:
+    """Tests for atomic write crash safety."""
+
+    def test_save_observer_crash_preserves_existing_file(
+        self, storage_env, monkeypatch
+    ):
+        """save_observer leaves prior observer data intact on replace failure."""
+        observer = {
+            "key": "testkey123456789",
+            "name": "original",
+            "stats": {},
+        }
+        assert save_observer(observer) is True
+
+        def raising_stub(*args, **kwargs):
+            raise OSError("simulated crash")
+
+        monkeypatch.setattr("think.entities.core.os.replace", raising_stub)
+
+        updated_observer = {
+            "key": "testkey123456789",
+            "name": "updated",
+            "stats": {},
+        }
+        assert save_observer(updated_observer) is False
+
+        loaded = load_observer("testkey123456789")
+        assert loaded is not None
+        assert loaded["name"] == "original"
+        assert list(storage_env.observers_dir.glob(".tmp_*")) == []
+
+    def test_increment_stat_crash_preserves_existing_file(
+        self, storage_env, monkeypatch
+    ):
+        """increment_stat leaves prior observer data intact on replace failure."""
+        observer = {
+            "key": "testkey123456789",
+            "name": "test",
+            "stats": {"events_received": 5},
+        }
+        assert save_observer(observer) is True
+
+        def raising_stub(*args, **kwargs):
+            raise OSError("simulated crash")
+
+        monkeypatch.setattr("think.entities.core.os.replace", raising_stub)
+
+        increment_stat("testkey1", "events_received")
+
+        loaded = load_observer("testkey123456789")
+        assert loaded is not None
+        assert loaded["stats"]["events_received"] == 5
+        assert list(storage_env.observers_dir.glob(".tmp_*")) == []
+
+
 class TestFindSegmentBySha256:
     """Tests for find_segment_by_sha256."""
 
diff --git a/apps/observer/utils.py b/apps/observer/utils.py
index d47c85b8f..355ffca98 100644
--- a/apps/observer/utils.py
+++ b/apps/observer/utils.py
@@ -15,6 +15,7 @@ import os
 from pathlib import Path
 
 from apps.utils import get_app_storage_path
+from think.entities.core import atomic_write
 
 logger = logging.getLogger(__name__)
 
@@ -78,8 +79,7 @@ def save_observer(data: dict) -> bool:
     observers_dir = get_observers_dir()
     observer_path = observers_dir / f"{key[:8]}.json"
     try:
-        with open(observer_path, "w") as f:
-            json.dump(data, f, indent=2)
+        atomic_write(observer_path, json.dumps(data, indent=2))
         os.chmod(observer_path, 0o600)
         return True
     except OSError:
@@ -179,8 +179,7 @@ def increment_stat(key_prefix: str, stat_name: str) -> None:
 
         data["stats"][stat_name] = data["stats"].get(stat_name, 0) + 1
 
-        with open(observer_path, "w") as f:
-            json.dump(data, f, indent=2)
+        atomic_write(observer_path, json.dumps(data, indent=2))
         os.chmod(observer_path, 0o600)
     except (json.JSONDecodeError, OSError, KeyError) as e:
         logger.warning(f"Failed to update {stat_name} for {key_prefix}: {e}")