diff --git a/gateway/channel_directory.py b/gateway/channel_directory.py
index cdd2ff9a241..ecc54e64464 100644
--- a/gateway/channel_directory.py
+++ b/gateway/channel_directory.py
@@ -12,6 +12,7 @@ from datetime import datetime
 from typing import Any, Dict, List, Optional
 
 from hermes_cli.config import get_hermes_home
+from utils import atomic_json_write
 
 logger = logging.getLogger(__name__)
 
@@ -86,9 +87,7 @@ def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
     }
 
     try:
-        DIRECTORY_PATH.parent.mkdir(parents=True, exist_ok=True)
-        with open(DIRECTORY_PATH, "w", encoding="utf-8") as f:
-            json.dump(directory, f, indent=2, ensure_ascii=False)
+        atomic_json_write(DIRECTORY_PATH, directory)
     except Exception as e:
         logger.warning("Channel directory: failed to write: %s", e)
 
diff --git a/tests/gateway/test_channel_directory.py b/tests/gateway/test_channel_directory.py
index 8981be6befb..50d5b04b747 100644
--- a/tests/gateway/test_channel_directory.py
+++ b/tests/gateway/test_channel_directory.py
@@ -6,6 +6,7 @@ from pathlib import Path
 from unittest.mock import patch
 
 from gateway.channel_directory import (
+    build_channel_directory,
     resolve_channel_name,
     format_directory_for_display,
     load_directory,
@@ -45,6 +46,27 @@ class TestLoadDirectory:
         assert result["updated_at"] is None
 
 
+class TestBuildChannelDirectoryWrites:
+    def test_failed_write_preserves_previous_cache(self, tmp_path, monkeypatch):
+        cache_file = _write_directory(tmp_path, {
+            "telegram": [{"id": "123", "name": "Alice", "type": "dm"}]
+        })
+        previous = json.loads(cache_file.read_text())
+
+        def broken_dump(data, fp, *args, **kwargs):
+            fp.write('{"updated_at":')
+            fp.flush()
+            raise OSError("disk full")
+
+        monkeypatch.setattr(json, "dump", broken_dump)
+
+        with patch("gateway.channel_directory.DIRECTORY_PATH", cache_file):
+            build_channel_directory({})
+            result = load_directory()
+
+        assert result == previous
+
+
 class TestResolveChannelName:
     def _setup(self, tmp_path, platforms):
         cache_file = _write_directory(tmp_path, platforms)