fix(examples): verify temporary prompt bundle

Author: YAO-001

examples/optimization/eval_optimize_loop/eval_loop/writeback.py

@@ -195,6 +195,14 @@ def temporary_prompt_bundle(
                 expected_sha256=expected_hashes[name],
             )
             written_hashes[name] = candidate_hashes[name]
+        installed_hashes = _current_hashes(snapshot)
+        candidate_mismatches = [
+            name for name, candidate_hash in candidate_hashes.items() if installed_hashes[name] != candidate_hash
+        ]
+        if candidate_mismatches:
+            raise ConcurrentPromptUpdateError(
+                "candidate prompt files changed before temporary evaluation: " f"{', '.join(candidate_mismatches)}"
+            )
         yield
     except BaseException as primary_error:
         failures = _restore_snapshot(snapshot, written_hashes)

examples/optimization/eval_optimize_loop/tests/test_writeback.py

@@ -173,6 +173,35 @@ def inject_external_before_second_precondition(file_descriptor: int):
     assert paths["user"].read_bytes() == external_content
 
 
+def test_temporary_verifies_complete_candidate_bundle_before_entering_body(tmp_path: Path, monkeypatch):
+    paths, baseline = _prompt_files(tmp_path)
+    snapshot = snapshot_prompt_files(paths)
+    external_content = b"external after second candidate replace"
+    original_replace = os.replace
+    replace_calls = 0
+    body_entered = False
+
+    def update_first_file_after_second_replace(source: str | bytes | Path, destination: str | bytes | Path):
+        nonlocal replace_calls
+        original_replace(source, destination)
+        replace_calls += 1
+        if replace_calls == 2:
+            paths["system"].write_bytes(external_content)
+
+    monkeypatch.setattr(writeback.os, "replace", update_first_file_after_second_replace)
+
+    with pytest.raises((ConcurrentPromptUpdateError, RuntimeError)):
+        with temporary_prompt_bundle(
+            snapshot,
+            {"system": "candidate system", "user": "candidate user"},
+        ):
+            body_entered = True
+
+    assert not body_entered
+    assert paths["system"].read_bytes() == external_content
+    assert paths["user"].read_bytes() == baseline["user"]
+
+
 def test_commit_rejects_concurrent_update_before_any_write(tmp_path: Path, monkeypatch):
     paths, baseline = _prompt_files(tmp_path)
     snapshot = snapshot_prompt_files(paths)