fix(io): route the remaining graph/manifest writers through the atomic helper (follow-up to #1952)

The atomic-write PR left several writers on the old truncate-then-write path.
Route them through write_json_atomic so a crash mid-write can't corrupt them:
- the --no-cluster raw graph.json dump (a core graph.json writer)
- merge-graphs / merge-chunks / merge-semantic output
- .graphify_analysis.json and .graphify_labels.json sidecars
- global_graph.py's global-graph.json and global-manifest.json

write_json_atomic gains an ensure_ascii flag so the raw-UTF-8 writers
(labels, merge outputs) keep byte-for-byte output. Adds tests for the
Windows PermissionError copy fallback and ensure_ascii=False.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

graphify/cli.py

@@ -1400,7 +1400,8 @@ def dispatch_command(cmd: str) -> None:
             encoding="utf-8",
         )
         to_json(G, communities, str(out / "graph.json"), community_labels=labels)
-        labels_path.write_text(json.dumps({str(k): v for k, v in labels.items()}, ensure_ascii=False), encoding="utf-8")
+        from graphify.paths import write_json_atomic as _wja
+        _wja(labels_path, {str(k): v for k, v in labels.items()}, ensure_ascii=False)
         # Membership signatures beside the labels so a later cluster-only can detect
         # which communities changed and avoid reusing a stale label (see reuse above).
         from graphify.cluster import community_member_sigs as _cms
@@ -1690,7 +1691,8 @@ def dispatch_command(cmd: str) -> None:
         except TypeError:
             out_data = _jg.node_link_data(merged)
         out_path.parent.mkdir(parents=True, exist_ok=True)
-        out_path.write_text(json.dumps(out_data, indent=2), encoding="utf-8")
+        from graphify.paths import write_json_atomic as _wja
+        _wja(out_path, out_data, indent=2)
         print(f"Merged {len(graphs)} graphs -> {merged.number_of_nodes()} nodes, {merged.number_of_edges()} edges")
         print(f"Written to: {out_path}")
 
@@ -2846,9 +2848,8 @@ def dispatch_command(cmd: str) -> None:
                     )
                     sys.exit(1)
             _backup(graphify_out)
-            graph_json_path.write_text(
-                json.dumps(merged, indent=2), encoding="utf-8"
-            )
+            from graphify.paths import write_json_atomic as _write_json_atomic
+            _write_json_atomic(graph_json_path, merged, indent=2)
             stages.mark("write")
             cost = _estimate_cost(
                 backend, merged["input_tokens"], merged["output_tokens"]
@@ -3000,7 +3001,8 @@ def dispatch_command(cmd: str) -> None:
                 "output": merged["output_tokens"],
             },
         }
-        analysis_path.write_text(json.dumps(analysis, indent=2), encoding="utf-8")
+        from graphify.paths import write_json_atomic as _wja
+        _wja(analysis_path, analysis, indent=2)
         try:
             _save_manifest(_manifest_files, manifest_path=str(manifest_path), kind="both", root=target, scan_corpus=_scan_corpus)
         except Exception as exc:
@@ -3158,7 +3160,8 @@ def dispatch_command(cmd: str) -> None:
                 _v = chunk.get(_tok, 0)
                 merged[_tok] += _v if isinstance(_v, (int, float)) else 0
         out_path.parent.mkdir(parents=True, exist_ok=True)
-        out_path.write_text(json.dumps(merged, ensure_ascii=False), encoding="utf-8")
+        from graphify.paths import write_json_atomic as _wja
+        _wja(out_path, merged, ensure_ascii=False)
         print(
             f"Merged {len(chunk_files)} chunks: {len(merged['nodes'])} nodes, {len(merged['edges'])} edges, "
             f"{merged['input_tokens']:,} in / {merged['output_tokens']:,} out tokens"
@@ -3202,7 +3205,8 @@ def dispatch_command(cmd: str) -> None:
             "hyperedges": cached_data.get("hyperedges", []) + new_data.get("hyperedges", []),
         }
         out_path2.parent.mkdir(parents=True, exist_ok=True)
-        out_path2.write_text(json.dumps(merged2, ensure_ascii=False), encoding="utf-8")
+        from graphify.paths import write_json_atomic as _wja
+        _wja(out_path2, merged2, ensure_ascii=False)
         print(f"Merged: {len(merged2['nodes'])} nodes, {len(merged2['edges'])} edges")
 
     elif Path(cmd).exists() or cmd in (".", "..") or cmd.startswith(("./", "../", "/", "~")):

graphify/global_graph.py

@@ -42,7 +42,8 @@ def _load_manifest() -> dict:
 
 def _save_manifest(manifest: dict) -> None:
     _GLOBAL_DIR.mkdir(parents=True, exist_ok=True)
-    _GLOBAL_MANIFEST.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
+    from graphify.paths import write_json_atomic
+    write_json_atomic(_GLOBAL_MANIFEST, manifest, indent=2)
 
 
 def _load_global_graph() -> nx.Graph:
@@ -65,7 +66,8 @@ def _save_global_graph(G: nx.Graph) -> None:
         data = _jg.node_link_data(G, edges="links")
     except TypeError:
         data = _jg.node_link_data(G)
-    _GLOBAL_GRAPH.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    from graphify.paths import write_json_atomic
+    write_json_atomic(_GLOBAL_GRAPH, data, indent=2)
 
 
 def _file_hash(path: Path) -> str:

graphify/paths.py

@@ -85,11 +85,12 @@ def write_text_atomic(path: "str | Path", text: str) -> None:
     _atomic_replace(path, lambda f: f.write(text))
 
 
-def write_json_atomic(path: "str | Path", obj, *, indent: "int | None" = None) -> None:
+def write_json_atomic(path: "str | Path", obj, *, indent: "int | None" = None, ensure_ascii: bool = True) -> None:
     """Atomically write ``obj`` as JSON to ``path``, streaming the encode into the
     temp file rather than materializing the whole string first (matters for very
-    large graphs). See :func:`_atomic_replace`."""
-    _atomic_replace(path, lambda f: json.dump(obj, f, indent=indent))
+    large graphs). ``ensure_ascii`` mirrors ``json.dump`` so callers that emit raw
+    UTF-8 (non-ASCII labels/paths) keep byte-for-byte output. See :func:`_atomic_replace`."""
+    _atomic_replace(path, lambda f: json.dump(obj, f, indent=indent, ensure_ascii=ensure_ascii))
 
 # Directory segments that, when they appear as a whole path component, mark the
 # whole path as a test location. Matched against path *segments* (not raw

tests/test_atomic_writes.py

@@ -99,3 +99,36 @@ def test_save_manifest_writes_atomically(tmp_path):
                   kind="both", root=tmp_path)
     assert json.loads(mpath.read_text())  # non-empty, valid JSON
     assert not any(x.name.endswith(".tmp") for x in mpath.parent.iterdir())
+
+
+def test_write_text_atomic_windows_permission_fallback(tmp_path, monkeypatch):
+    """On Windows os.replace raises PermissionError when the destination is
+    briefly locked (antivirus, an open reader); the copy-then-delete fallback
+    must still land the new content and leave no temp file."""
+    p = tmp_path / "graph.json"
+    p.write_text("original", encoding="utf-8")
+
+    real_replace = os.replace
+    calls = {"n": 0}
+
+    def flaky_replace(src, dst):
+        calls["n"] += 1
+        raise PermissionError("simulated WinError 5")
+
+    monkeypatch.setattr(os, "replace", flaky_replace)
+    write_text_atomic(p, "new-content")
+
+    assert calls["n"] == 1  # the fallback path was actually exercised
+    assert p.read_text() == "new-content"
+    assert sorted(x.name for x in tmp_path.iterdir()) == ["graph.json"]
+
+
+def test_write_json_atomic_ensure_ascii_false_preserves_utf8(tmp_path):
+    from graphify.paths import write_json_atomic
+
+    p = tmp_path / "g.json"
+    write_json_atomic(p, {"label": "Wörker 数据"}, ensure_ascii=False)
+    raw = p.read_text(encoding="utf-8")
+    assert "Wörker 数据" in raw  # raw UTF-8, not \\uXXXX escapes
+    assert "\\u" not in raw
+    assert json.loads(raw) == {"label": "Wörker 数据"}