fix(retrieval): validate owned index data

scripts/lib/semantic-retrieval.py

@@ -90,10 +90,17 @@ def check_runtime(target, state):
         path = target / name
         if path.is_symlink() or not path.is_file() or digest(path) != expected:
             raise ValueError(f'Owned semantic retrieval file changed: {name}')
-    allowed = set(state['files']) | {'.venv', 'model-cache', '__pycache__'}
+    allowed = set(state['files']) | {'.venv', 'model-cache', '__pycache__', 'data'}
     for path in target.iterdir():
         if path.is_symlink() or path.name not in allowed:
             raise ValueError(f'Foreign or linked semantic retrieval path: {path.name}')
+        if path.name == 'data' and (not path.is_dir() or any(
+                child.is_symlink() or not (
+                    child.is_file() and (child.name in ('daemon.key', 'daemon.starting') or re.fullmatch(
+                        r'[0-9a-f]{64}\.sqlite3(?:-(?:wal|shm|journal))?', child.name))
+                    or child.is_socket() and re.fullmatch(r'daemon-[0-9a-f]{16}\.sock', child.name))
+                for child in path.iterdir())):
+            raise ValueError('Foreign or linked semantic retrieval path: data')
         if path.name == '__pycache__' and (not path.is_dir() or any(
                 child.is_symlink() or not child.is_file() or not re.fullmatch(
                     r'(?:server|benchmark)\.[A-Za-z0-9_-]+(?:\.opt-\d+)?\.pyc', child.name)
@@ -188,7 +195,7 @@ def benchmark(root, home, dry_run, summary=False):
         shutil.rmtree(target)
         state_path.unlink(missing_ok=True)
     if not summary:
-        print('Semantic retrieval benchmark: ' + ('recommended' if outcome['recommended'] else 'not recommended') + f" (embedding {outcome['embeddingSeconds']}s, reranking {outcome['rerankingSeconds']}s)")
+        print('Semantic retrieval warm-model check: ' + ('passed' if outcome['recommended'] else 'failed') + f" (embedding {outcome['embeddingSeconds']}s, reranking {outcome['rerankingSeconds']}s; excludes model startup and indexing)")
     return outcome['recommended']
 
 

scripts/tests/test-semantic-retrieval.py

@@ -5,6 +5,7 @@ import json
 from contextlib import redirect_stdout
 from pathlib import Path
 import py_compile
+import subprocess
 import tempfile
 import unittest
 from unittest.mock import patch
@@ -105,6 +106,39 @@ class SemanticRetrievalTests(unittest.TestCase):
             with self.assertRaisesRegex(ValueError, '__pycache__'):
                 MODULE.sync(ROOT, home, ['codex'], True, False, True)
 
+    def test_install_search_update_and_remove_owned_data(self):
+        with tempfile.TemporaryDirectory() as directory:
+            home = Path(directory) / 'home'
+            repository = Path(directory) / 'repository'
+            repository.mkdir()
+            subprocess.run(['git', '-C', str(repository), 'init', '--quiet'], check=True)
+            target = home / '.my-ai-configuration/semantic-retrieval'
+            state_path = target.parent / 'semantic-retrieval.json'
+            with patch.object(MODULE, 'ensure_runtime', return_value=Path('python')), \
+                 patch.object(MODULE, 'ensure_models'):
+                MODULE.sync(ROOT, home, ['codex'], True, False, False)
+                installed = importlib.util.spec_from_file_location('installed_retrieval', target / 'server.py')
+                server = importlib.util.module_from_spec(installed)
+                installed.loader.exec_module(server)
+                index = server.Index(repository, target / 'data')
+                try:
+                    self.assertEqual(index.search('query'), [])
+                finally:
+                    index.db.close()
+                self.assertEqual(len(list((target / 'data').glob('*.sqlite3'))), 1)
+                server.daemon_authkey(target / 'data')
+                MODULE.sync(ROOT, home, ['codex'], True, False, True)
+                self.assertTrue((target / 'data').is_dir())
+                foreign = target / 'data' / 'foreign.txt'
+                foreign.write_text('keep')
+                with self.assertRaisesRegex(ValueError, 'data'):
+                    MODULE.sync(ROOT, home, ['codex'], False, False, False)
+                self.assertTrue(foreign.exists())
+                foreign.unlink()
+                MODULE.sync(ROOT, home, ['codex'], False, False, False)
+            self.assertFalse(target.exists())
+            self.assertFalse(state_path.exists())
+
     def test_negative_benchmark_keeps_active_runtime(self):
         with tempfile.TemporaryDirectory() as directory:
             home = Path(directory) / 'home'