fix: prevent communication transfer source-sidecar overwrite

core/communication_transfer.py

@@ -15,6 +15,7 @@
 import hashlib
 import json
 import os
+import tempfile
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
@@ -201,10 +202,30 @@ def build_communication_transfer(
 
 def write_communication_transfer(record: dict, output: str | Path) -> Path:
     path = Path(output)
+    source_path = (record.get("source_figure_evidence") or {}).get("path")
+    if source_path and path.resolve() == Path(source_path).resolve():
+        raise ValueError("communication transfer output must not overwrite the source figure-evidence sidecar")
+
     path.parent.mkdir(parents=True, exist_ok=True)
-    temp = path.with_suffix(path.suffix + ".tmp")
-    temp.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-    os.replace(temp, path)
+    staged = None
+    try:
+        # Use an exclusive, same-directory temporary file: a fixed .tmp name
+        # could itself be an existing source sidecar.
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=path.parent,
+            prefix=f".{path.name}.",
+            suffix=".tmp",
+            delete=False,
+        ) as handle:
+            staged = Path(handle.name)
+            json.dump(record, handle, ensure_ascii=False, indent=2)
+            handle.write("\n")
+        os.replace(staged, path)
+    finally:
+        if staged is not None:
+            staged.unlink(missing_ok=True)
     return path
 
 

tests/test_tool_cli.py

@@ -248,5 +248,45 @@ def test_transfer_wrong_source_and_bad_profile(self):
         self.assertEqual(invalid["error"]["code"], "EXECUTION_FAILED")
 
 
+
+    def test_handoff_cli_preserves_source_when_output_collides(self):
+        with tempfile.TemporaryDirectory() as temp:
+            source = Path(temp) / "source.json"
+            original = json.dumps({"profile": "sci-render-kit/figure-evidence"}).encode("utf-8")
+            source.write_bytes(original)
+            command = [sys.executable, str(ROOT / "core" / "communication_transfer.py"), str(source)]
+            rejected = subprocess.run(
+                command + ["--output", str(source)],
+                cwd=ROOT, capture_output=True, text=True,
+            )
+            self.assertNotEqual(rejected.returncode, 0)
+            self.assertIn("communication transfer output must not overwrite", rejected.stderr)
+            self.assertEqual(source.read_bytes(), original)
+
+            output = Path(temp) / "handoff.json"
+            accepted = subprocess.run(
+                command + ["--output", str(output)],
+                cwd=ROOT, capture_output=True, text=True,
+            )
+            self.assertEqual(accepted.returncode, 0, accepted.stderr)
+            self.assertTrue(output.is_file())
+            self.assertEqual(source.read_bytes(), original)
+
+    def test_handoff_cli_does_not_clobber_legacy_temp_name(self):
+        with tempfile.TemporaryDirectory() as temp:
+            output = Path(temp) / "handoff.json"
+            source = Path(temp) / "handoff.json.tmp"
+            original = json.dumps({"profile": "sci-render-kit/figure-evidence"}).encode("utf-8")
+            source.write_bytes(original)
+            command = [sys.executable, str(ROOT / "core" / "communication_transfer.py"), str(source)]
+            completed = subprocess.run(
+                command + ["--output", str(output)],
+                cwd=ROOT, capture_output=True, text=True,
+            )
+            self.assertEqual(completed.returncode, 0, completed.stderr)
+            self.assertTrue(output.is_file())
+            self.assertEqual(source.read_bytes(), original)
+
+
 if __name__ == "__main__":
     unittest.main()