Preserve mixed evaluation evidence with reserved atomic report checkpoints

tools/mixed-eval/README.md

@@ -36,3 +36,5 @@ The raw report retains queries, expected IDs, ordered hits, source locations,
 scores, modes/scopes/languages and MCP roundtrip times. Those times exclude
 process/model initialization and do not represent native UI latency. Keep raw
 reports under ignored `.searchi`; review deliberate evidence before publishing.
+Use a new output filename: existing evidence is rejected before inference.
+Checkpoints replace JSON atomically so a failed write preserves the prior report.

tools/mixed-eval/evaluate.py

@@ -16,6 +16,28 @@ ROOT = Path(__file__).resolve().parents[2]
 FIXTURE = ROOT / 'fixtures/mixed'
 
 
+def reserve_report(path):
+    """Refuse to overwrite prior evidence before starting expensive inference."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open('x', encoding='utf-8'):
+        pass
+
+
+def write_report(path, report):
+    """Publish a complete JSON checkpoint without truncating the prior one."""
+    data = json.dumps(report, indent=2, ensure_ascii=False)
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=path.parent,
+                                         suffix='.partial', delete=False) as stream:
+            temporary = Path(stream.name)
+            stream.write(data)
+        os.replace(temporary, path)
+    finally:
+        if temporary is not None:
+            temporary.unlink(missing_ok=True)
+
+
 def digest(path):
     with path.open('rb') as stream:
         return hashlib.file_digest(stream, 'sha256').hexdigest()
@@ -92,14 +114,14 @@ def run(args):
             raise ValueError('Model bytes differ from the pinned Google manifest')
     spec = importlib.util.spec_from_file_location('searchi_benchmark', ROOT / 'tools/embeddinggemma2/benchmark.py')
     benchmark = importlib.util.module_from_spec(spec); spec.loader.exec_module(benchmark)
-    args.output.parent.mkdir(parents=True, exist_ok=True)
+    reserve_report(args.output)
     rows, stages = [], []
     provenance = {'fixture_sha256': lock['fixture_sha256'], 'model_manifest_sha256': digest(manifest_path),
                   'model': manifest, 'device': 'cpu', 'cpu_threads': 1,
                   'cli_sha256': digest(args.cli), 'worker_sha256': digest(args.worker),
                   'platform': platform.platform(), 'python': platform.python_version()}
     def checkpoint():
-        args.output.write_text(json.dumps({**provenance, 'complete': False, 'stages': stages, 'rows': rows}, indent=2, ensure_ascii=False), encoding='utf-8')
+        write_report(args.output, {**provenance, 'complete': False, 'stages': stages, 'rows': rows})
     def stage(name, command, timeout):
         started = time.perf_counter()
         try:
@@ -167,7 +189,7 @@ def run(args):
               'device': 'cpu', 'cpu_threads': 1, 'cli_sha256': digest(args.cli), 'worker_sha256': digest(args.worker),
               'scope_note': 'Small mixed diagnostic, reused previously tested media labels; file-level relevance only. No temporal-localization or production-scale claims. MCP roundtrip excludes initialization and is not UI latency.',
               'rows': rows, 'summary': summarize(rows)}
-    args.output.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding='utf-8')
+    write_report(args.output, report)
 
 
 if __name__ == '__main__':

tools/mixed-eval/test_evaluate.py

@@ -3,12 +3,38 @@ import importlib.util
 from pathlib import Path
 import tempfile
 import unittest
+from unittest.mock import patch
 
 spec = importlib.util.spec_from_file_location('mixed_eval', Path(__file__).with_name('evaluate.py'))
 evaluate = importlib.util.module_from_spec(spec); spec.loader.exec_module(evaluate)
 
 
 class EvaluationTests(unittest.TestCase):
+    def test_existing_report_is_preserved_and_new_report_is_reserved(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'nested' / 'report.json'
+            evaluate.reserve_report(path)
+            evaluate.write_report(path, {'complete': False, 'query': 'Straße'})
+            previous = path.read_bytes()
+            with self.assertRaises(FileExistsError):
+                evaluate.reserve_report(path)
+            self.assertEqual(path.read_bytes(), previous)
+            evaluate.write_report(path, {'complete': True})
+            self.assertIn('true', path.read_text(encoding='utf-8'))
+            self.assertEqual(list(path.parent.glob('*.partial')), [])
+
+    def test_failed_checkpoint_publication_keeps_previous_report(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'report.json'
+            evaluate.reserve_report(path)
+            evaluate.write_report(path, {'complete': False, 'rows': [1]})
+            previous = path.read_bytes()
+            with patch.object(evaluate.os, 'replace', side_effect=OSError('disk failure')):
+                with self.assertRaises(OSError):
+                    evaluate.write_report(path, {'complete': True})
+            self.assertEqual(path.read_bytes(), previous)
+            self.assertEqual(list(path.parent.glob('*.partial')), [])
+
     def test_frozen_real_assets_verify(self):
         fixture, _ = evaluate.load_fixture()
         self.assertEqual(len(fixture['assets']), 32)