fix(retrieval): write ownership state atomically

TODO.md

@@ -9,7 +9,7 @@
 - [ ] Preserve pre-existing global CLI packages. `ensure_cli()` and `ensure_qmd()` install and claim npm packages without checking whether they were already installed; later removal can uninstall a user's package. (`scripts/lib/managed-extensions.py:342`, `scripts/lib/managed-extensions.py:378`)
 - [ ] Fetch skill archives by an immutable commit and record that same commit. The archive uses GitHub `HEAD`, while `resolvedRevision` comes from a separate later request, so the recorded revision may not identify the installed payload. (`scripts/lib/managed-extensions.py:94`, `scripts/lib/managed-extensions.py:140`, `scripts/lib/managed-extensions.py:409`)
 - [ ] Protect the semantic retrieval daemon's IPC protocol. It uses a hardcoded authentication key and `multiprocessing.connection.recv()`, which deserializes messages; a process that can reach its socket or pipe can submit arbitrary roots and potentially crafted objects. Use a private per-install credential and a non-pickle wire format, or restrict the endpoint to the current user. (`shared/retrieval/server.py:47`, `shared/retrieval/server.py:306`, `shared/retrieval/server.py:314`)
-- [ ] Write semantic retrieval ownership state atomically. `save_state()` replaces the JSON with a direct write, unlike the extension and selection ledgers; interruption can leave it unreadable and block later reconciliation. (`scripts/lib/semantic-retrieval.py:63`)
+- [x] Write semantic retrieval ownership state atomically. `save_state()` replaces the JSON with a direct write, unlike the extension and selection ledgers; interruption can leave it unreadable and block later reconciliation. (`scripts/lib/semantic-retrieval.py:63`)
 
 ## Menus and reconciliation
 

scripts/lib/semantic-retrieval.py

@@ -6,6 +6,7 @@ from pathlib import Path
 import shutil
 import subprocess
 import sys
+import tempfile
 
 
 FILES = ('server.py', 'benchmark.py', 'requirements.txt')
@@ -63,7 +64,18 @@ def load_state(path):
 def save_state(path, state, dry_run):
     if not dry_run:
         path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text(json.dumps(state, indent=2) + '\n', encoding='utf-8')
+        temporary = None
+        try:
+            with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=path.parent, delete=False) as stream:
+                temporary = Path(stream.name)
+                json.dump(state, stream, indent=2)
+                stream.write('\n')
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(temporary, path)
+        finally:
+            if temporary is not None:
+                temporary.unlink(missing_ok=True)
 
 
 def sync(root, home, clients, enabled, dry_run, update, summary=False):