fix(export): write graph.canvas and Obsidian notes atomically

graphify/export.py

@@ -16,7 +16,7 @@
 from graphify.security import sanitize_label
 from graphify.analyze import _node_community_map
 from graphify.build import edge_data
-from graphify.paths import stem_filename_budget
+from graphify.paths import stem_filename_budget, write_json_atomic, write_text_atomic
 
 from graphify.exporters.graphdb import push_to_falkordb, push_to_neo4j  # noqa: E402,F401
 
@@ -726,7 +726,7 @@ def _owned_write(rel_name: str, content: str) -> bool:
             _skipped.append(rel_name)
             return False
         target.parent.mkdir(parents=True, exist_ok=True)
-        target.write_text(content, encoding="utf-8")  # nosec
+        write_text_atomic(target, content)
         _written.append(rel_name)
         return True
 
@@ -1199,7 +1199,7 @@ def to_canvas(
         })
 
     canvas_data = {"nodes": canvas_nodes, "edges": canvas_edges}
-    Path(output_path).write_text(json.dumps(canvas_data, indent=2), encoding="utf-8")  # nosec
+    write_json_atomic(output_path, canvas_data, indent=2)
 
 
 def to_graphml(

tests/test_atomic_canvas_export.py

@@ -0,0 +1,73 @@
+"""Regression: to_canvas / Obsidian vault writes must be atomic (#3282)."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import networkx as nx
+import pytest
+
+from graphify.export import to_canvas, to_obsidian
+
+
+def _tiny_graph():
+    G = nx.Graph()
+    G.add_node("a", label="Alpha", file_type="code", source_file="a.py")
+    G.add_node("b", label="Beta", file_type="code", source_file="b.py")
+    G.add_edge("a", "b", relation="calls", confidence="EXTRACTED", weight=1.0)
+    return G, {0: ["a", "b"]}
+
+
+def test_to_canvas_uses_atomic_replace(tmp_path, monkeypatch):
+    G, communities = _tiny_graph()
+    out = tmp_path / "graph.canvas"
+    out.write_text('{"nodes":[],"edges":[]}', encoding="utf-8")
+
+    real_replace = os.replace
+    calls: list[tuple[str, str]] = []
+
+    def tracking_replace(src, dst):
+        calls.append((str(src), str(dst)))
+        return real_replace(src, dst)
+
+    monkeypatch.setattr(os, "replace", tracking_replace)
+    to_canvas(G, communities, str(out))
+
+    assert any(Path(dst).resolve() == out.resolve() for _, dst in calls), calls
+    data = json.loads(out.read_text(encoding="utf-8"))
+    assert "nodes" in data and "edges" in data
+    assert len(data["nodes"]) >= 2
+
+
+def test_to_canvas_preserves_existing_when_replace_fails(tmp_path, monkeypatch):
+    G, communities = _tiny_graph()
+    out = tmp_path / "graph.canvas"
+    original = '{"nodes":[],"edges":[],"preserved":true}'
+    out.write_text(original, encoding="utf-8")
+
+    def boom(src, dst):
+        raise OSError("simulated failure")
+
+    monkeypatch.setattr(os, "replace", boom)
+    with pytest.raises(OSError):
+        to_canvas(G, communities, str(out))
+
+    assert out.read_text(encoding="utf-8") == original
+
+
+def test_to_obsidian_owned_writes_are_atomic(tmp_path, monkeypatch):
+    G, communities = _tiny_graph()
+    real_replace = os.replace
+    calls: list[str] = []
+
+    def tracking_replace(src, dst):
+        calls.append(str(dst))
+        return real_replace(src, dst)
+
+    monkeypatch.setattr(os, "replace", tracking_replace)
+    to_obsidian(G, communities, str(tmp_path), community_labels={0: "Core"})
+
+    # At least one vault artifact should land via os.replace (notes / graph.json).
+    assert calls, "expected atomic replaces for Obsidian vault writes"
+    assert any(Path(p).name.endswith(".md") or p.endswith("graph.json") for p in calls)