From cc5379fe1b9f9177632d547d923448266cb82045 Mon Sep 17 00:00:00 2001
From: lostlight530 <13957086206@163.com>
Date: Sun, 25 Oct 2026 07:09:25 +0800
Subject: [PATCH 1/2] test: reproduce handoff source and staging collisions

---
 tests/test_tool_cli.py | 40 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 40 insertions(+)

diff --git a/tests/test_tool_cli.py b/tests/test_tool_cli.py
index 6ddc391..32a52db 100644
--- a/tests/test_tool_cli.py
+++ b/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()

From 48378c5b7f85c4d3044b8344a46e59770611b013 Mon Sep 17 00:00:00 2001
From: lostlight530 <13957086206@163.com>
Date: Sun, 25 Oct 2026 07:09:43 +0800
Subject: [PATCH 2/2] fix: preserve source sidecar and use exclusive staging
 for handoff output

---
 core/communication_transfer.py | 27 ++++++++++++++++++++++++---
 1 file changed, 24 insertions(+), 3 deletions(-)

diff --git a/core/communication_transfer.py b/core/communication_transfer.py
index e664152..f94b9b0 100644
--- a/core/communication_transfer.py
+++ b/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