fix: atomically install preview bundles

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

src/quant_advisor_research/preview_bundle.py

@@ -106,45 +106,46 @@ def artifact(name: str, role: str, content: bytes) -> dict[str, str]:
     }
 
 
-def _output_dir(path: str | Path) -> Path:
+def _output_parent(path: str | Path) -> Path:
     output = Path(path)
     try:
-        if not output.is_dir():
-            raise _error("output_directory_invalid")
-        if any(output.iterdir()):
-            raise _error("output_not_empty")
+        if output.exists():
+            raise _error("output_exists")
+        if not output.parent.is_dir():
+            raise _error("output_parent_invalid")
     except PreviewBundleError:
         raise
     except (OSError, TypeError, ValueError):
-        raise _error("output_directory_invalid") from None
+        raise _error("output_parent_invalid") from None
     return output
 
 
 def build_preview_bundle(report: Mapping[str, Any], output_dir: str | Path) -> PreviewBundleEvidence:
     """Validate once, build three deterministic bytes, then write an empty directory."""
     snapshot = _validated_source(report)
-    output = _output_dir(output_dir)
+    output = _output_parent(output_dir)
     report_bytes = _canonical_json(snapshot)
     html_bytes = _render_html(snapshot, report_bytes)
     manifest = _manifest(snapshot, report_bytes, html_bytes)
     manifest_bytes = _canonical_json(manifest)
     files = {"report.json": report_bytes, "report.html": html_bytes, "manifest.json": manifest_bytes}
-    temp_dir: str | None = None
+    staging_dir: str | None = None
     try:
-        temp_dir = tempfile.mkdtemp(prefix=".qar-preview-", dir=output)
+        staging_dir = tempfile.mkdtemp(prefix=f".{output.name}.staging-", dir=output.parent)
         for name, content in files.items():
-            Path(temp_dir, name).write_bytes(content)
-        for name in files:
-            os.replace(Path(temp_dir, name), output / name)
-        os.rmdir(temp_dir)
-        temp_dir = None
+            Path(staging_dir, name).write_bytes(content)
+        read_preview_bundle(staging_dir)
+        os.rename(staging_dir, output)
+        staging_dir = None
+    except FileExistsError:
+        raise _error("output_exists") from None
     except (OSError, TypeError, ValueError):
         raise _error("output_write_failed") from None
     finally:
-        if temp_dir is not None:
-            for child in Path(temp_dir).iterdir():
+        if staging_dir is not None:
+            for child in Path(staging_dir).iterdir():
                 child.unlink(missing_ok=True)
-            Path(temp_dir).rmdir()
+            Path(staging_dir).rmdir()
     return PreviewBundleEvidence(MappingProxyType(snapshot), MappingProxyType(manifest))
 
 

tests/test_preview_bundle.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+import quant_advisor_research.preview_bundle as preview_bundle
 from quant_advisor_research.advisory_report import build_advisory_report
 from quant_advisor_research.preview_bundle import (
     BUNDLE_CONTRACT,
@@ -29,7 +30,6 @@ def report(*, cadence="daily", as_of="2026-06-20"):
 
 def test_daily_bundle_builds_and_readback_validates(tmp_path):
     output = tmp_path / "preview"
-    output.mkdir()
     result = build_preview_bundle(report(), output)
 
     assert result.bundle_contract == BUNDLE_CONTRACT
@@ -64,10 +64,9 @@ def test_daily_bundle_builds_and_readback_validates(tmp_path):
 @pytest.mark.parametrize("cadence", ["weekly", "monthly"])
 def test_non_daily_source_is_rejected_before_output(cadence, tmp_path):
     output = tmp_path / "preview"
-    output.mkdir()
     with pytest.raises(PreviewBundleError, match="daily_only"):
         build_preview_bundle(report(cadence=cadence), output)
-    assert list(output.iterdir()) == []
+    assert not output.exists()
 
 
 @pytest.mark.parametrize("mutation", [
@@ -79,17 +78,14 @@ def test_source_contract_mutations_fail_closed_without_partial_output(mutation,
     value = report()
     mutation(value)
     output = tmp_path / "preview"
-    output.mkdir()
     with pytest.raises(PreviewBundleError):
         build_preview_bundle(value, output)
-    assert list(output.iterdir()) == []
+    assert not output.exists()
 
 
 def test_build_is_deterministic_for_equivalent_mapping_order(tmp_path):
     left = tmp_path / "left"
     right = tmp_path / "right"
-    left.mkdir()
-    right.mkdir()
     value = report()
     reordered = {key: value[key] for key in reversed(list(value))}
     build_preview_bundle(value, left)
@@ -101,7 +97,6 @@ def test_build_is_deterministic_for_equivalent_mapping_order(tmp_path):
 
 def test_html_escapes_snapshot_and_has_only_fixed_relative_links(tmp_path):
     output = tmp_path / "preview"
-    output.mkdir()
     value = report()
     value["source_artifacts"]["political_events"] = "<script>alert('x')</script>"
     build_preview_bundle(value, output)
@@ -121,7 +116,6 @@ def test_html_escapes_snapshot_and_has_only_fixed_relative_links(tmp_path):
 ])
 def test_readback_tamper_and_extra_file_fail_closed(tamper, tmp_path):
     output = tmp_path / "preview"
-    output.mkdir()
     build_preview_bundle(report(), output)
     tamper(output)
     with pytest.raises(PreviewBundleError):
@@ -133,6 +127,51 @@ def test_non_empty_output_fails_without_touching_sentinel(tmp_path):
     output.mkdir()
     sentinel = output / "sentinel"
     sentinel.write_text("keep")
-    with pytest.raises(PreviewBundleError, match="output_not_empty"):
+    with pytest.raises(PreviewBundleError, match="output_exists"):
         build_preview_bundle(report(), output)
     assert sentinel.read_text() == "keep"
+
+
+def test_destination_is_not_visible_during_staging_readback(tmp_path, monkeypatch):
+    output = tmp_path / "preview"
+    observed = []
+    original = preview_bundle.read_preview_bundle
+
+    def inspect(path):
+        observed.append(Path(path))
+        assert not output.exists()
+        return original(path)
+
+    monkeypatch.setattr(preview_bundle, "read_preview_bundle", inspect)
+    build_preview_bundle(report(), output)
+    assert len(observed) == 1
+    assert output.is_dir()
+
+
+def test_concurrent_destination_winner_is_not_overwritten_or_cleaned(tmp_path, monkeypatch):
+    output = tmp_path / "preview"
+    real_rename = preview_bundle.os.rename
+
+    def concurrent_winner(source, destination):
+        Path(destination).mkdir()
+        raise FileExistsError(destination)
+
+    monkeypatch.setattr(preview_bundle.os, "rename", concurrent_winner)
+    with pytest.raises(PreviewBundleError, match="output_exists"):
+        build_preview_bundle(report(), output)
+    assert output.is_dir()
+    assert not list(tmp_path.glob(".preview.staging-*"))
+    monkeypatch.setattr(preview_bundle.os, "rename", real_rename)
+
+
+def test_staging_directory_is_cleaned_when_install_fails(tmp_path, monkeypatch):
+    output = tmp_path / "preview"
+
+    def fail_readback(_path):
+        raise PreviewBundleError("forced_readback_failure")
+
+    monkeypatch.setattr(preview_bundle, "read_preview_bundle", fail_readback)
+    with pytest.raises(PreviewBundleError, match="output_write_failed"):
+        build_preview_bundle(report(), output)
+    assert not output.exists()
+    assert not list(tmp_path.glob(".preview.staging-*"))