fix(sub_map): unique temp name per save and remove it if the dump fails

Follow the .tmp.{pid}.{thread} convention of alias_loader.py plus a
per-process counter: the SIGHUP handler runs on the main thread, the same
one the trimmer saves from, so pid and thread alone could still collide.
A failed pickle.dump no longer leaves an orphan temp file behind.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/adn_server/infrastructure/persistence/sub_map_store.py

@@ -25,13 +25,19 @@
 
 from __future__ import annotations
 
+import itertools
 import os
 import pickle
+import threading
 from collections.abc import Hashable
 from pathlib import Path
 
 from ...application.ports import SubMapStore
 
+# A SIGHUP handler runs on the main thread, so it can interrupt a save already in
+# progress there: pid and thread alone would give both writers the same temp file.
+_tmp_seq = itertools.count()
+
 
 class PickleSubMapStore(SubMapStore):
     """Persist SUB_MAP as pickle (legacy compatible)."""
@@ -53,10 +59,13 @@ class PickleSubMapStore(SubMapStore):
         p.parent.mkdir(parents=True, exist_ok=True)
         # Write aside and rename: a crash mid-dump must not leave a truncated
         # file, which load() would turn into an empty map.
-        tmp = p.with_name(p.name + ".tmp")
-        with open(tmp, "wb") as f:
-            pickle.dump(sub_map, f)
-        os.replace(tmp, p)
+        tmp = p.with_name(f"{p.name}.tmp.{os.getpid()}.{threading.get_ident()}.{next(_tmp_seq)}")
+        try:
+            with open(tmp, "wb") as f:
+                pickle.dump(sub_map, f)
+            tmp.replace(p)
+        finally:
+            tmp.unlink(missing_ok=True)
 
 
 # SUB_MAP is rewritten on every frame, but only the route of an entry (system,

tests/infrastructure/test_sub_map_save_on_change.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 
 import pickle
 
+import pytest
+
 from adn_server.infrastructure.persistence import PickleSubMapStore, SubMapSaver
 
 
@@ -78,3 +80,18 @@ def test_save_leaves_no_temp_file_and_replaces_atomically(tmp_path):
     PickleSubMapStore().save(str(path), {b"new": ("Y", 2, 1.0)})
     assert [p.name for p in tmp_path.iterdir()] == ["sub_map.pkl"]
     assert PickleSubMapStore().load(str(path)) == {b"new": ("Y", 2, 1.0)}
+
+
+def test_failed_dump_leaves_no_temp_file(tmp_path, monkeypatch):
+    path = tmp_path / "sub_map.pkl"
+    PickleSubMapStore().save(str(path), {b"\x00\x00\x01": ("SYS", 1, 1.0)})
+
+    def boom(*_a, **_k):
+        raise pickle.PicklingError("boom")
+
+    monkeypatch.setattr(pickle, "dump", boom)
+    with pytest.raises(pickle.PicklingError):
+        PickleSubMapStore().save(str(path), {b"\x00\x00\x02": ("SYS", 2, 2.0)})
+
+    assert sorted(p.name for p in tmp_path.iterdir()) == ["sub_map.pkl"]
+    assert PickleSubMapStore().load(str(path)) == {b"\x00\x00\x01": ("SYS", 1, 1.0)}