From b950bb6c724431ddc2c9a2c543a540c9b7d5f440 Mon Sep 17 00:00:00 2001
From: poria-lang <mosespsitito@gmail.com>
Date: Mon, 10 Aug 2026 20:39:41 +0300
Subject: [PATCH] fix(report): clean up temp file when the atomic write fails

---
 strix/report/writer.py      | 29 +++++++++++-------
 tests/test_report_writer.py | 59 +++++++++++++++++++++++++++++++++----
 2 files changed, 72 insertions(+), 16 deletions(-)

diff --git a/strix/report/writer.py b/strix/report/writer.py
index 1fa2ad498..c96989760 100644
--- a/strix/report/writer.py
+++ b/strix/report/writer.py
@@ -179,17 +179,24 @@ def write_vulnerabilities(
 
 def _atomic_write_text(path: Path, payload: str) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
-    with tempfile.NamedTemporaryFile(
-        mode="w",
-        encoding="utf-8",
-        dir=str(path.parent),
-        prefix=f".{path.name}.",
-        suffix=".tmp",
-        delete=False,
-    ) as tmp:
-        tmp.write(payload)
-        tmp_path = Path(tmp.name)
-    tmp_path.replace(path)
+    tmp_path: Path | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=str(path.parent),
+            prefix=f".{path.name}.",
+            suffix=".tmp",
+            delete=False,
+        ) as tmp:
+            tmp_path = Path(tmp.name)
+            tmp.write(payload)
+        assert tmp_path is not None
+        tmp_path.replace(path)
+    except Exception:  # leave no temp file behind on any failure
+        if tmp_path is not None:
+            tmp_path.unlink(missing_ok=True)
+        raise
 
 
 def render_vulnerability_md(report: dict[str, Any]) -> str:  # noqa: PLR0912, PLR0915
diff --git a/tests/test_report_writer.py b/tests/test_report_writer.py
index 7d424d883..8eead46b9 100644
--- a/tests/test_report_writer.py
+++ b/tests/test_report_writer.py
@@ -5,7 +5,8 @@ from __future__ import annotations
 import csv
 import json
 import tempfile
-from typing import TYPE_CHECKING, Any
+from pathlib import Path
+from typing import Any, Self
 
 import pytest
 
@@ -18,10 +19,6 @@ from strix.report.writer import (
 )
 
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
-
 def _sample_report(**overrides: Any) -> dict[str, Any]:
     base: dict[str, Any] = {
         "id": "vuln-0001",
@@ -197,3 +194,55 @@ def test_write_executive_report_failure_preserves_previous_report(
         write_executive_report(tmp_path, "new content")
 
     assert target.read_text(encoding="utf-8") == "previous report"
+
+
+def test_write_executive_report_write_failure_preserves_previous_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    target = tmp_path / "penetration_test_report.md"
+    target.write_text("previous report", encoding="utf-8")
+
+    temp_path = tmp_path / ".penetration_test_report.md.stub.tmp"
+    temp_path.touch()
+
+    class _FailingTempFile:
+        name = str(temp_path)
+
+        def __enter__(self) -> Self:
+            return self
+
+        def __exit__(self, *_exc: object) -> None:
+            return None
+
+        def write(self, _payload: str) -> int:
+            raise OSError("disk full")
+
+    monkeypatch.setattr(
+        tempfile,
+        "NamedTemporaryFile",
+        lambda *_args, **_kwargs: _FailingTempFile(),
+    )
+
+    with pytest.raises(OSError, match="disk full"):
+        write_executive_report(tmp_path, "new content")
+
+    assert target.read_text(encoding="utf-8") == "previous report"
+    assert not temp_path.exists()  # the partial temp file is cleaned up
+
+
+def test_write_executive_report_replace_failure_preserves_previous_report(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    target = tmp_path / "penetration_test_report.md"
+    target.write_text("previous report", encoding="utf-8")
+
+    def _failing_replace(_self: Path, _destination: Path) -> None:
+        raise OSError("disk full")
+
+    monkeypatch.setattr(Path, "replace", _failing_replace)
+
+    with pytest.raises(OSError, match="disk full"):
+        write_executive_report(tmp_path, "new content")
+
+    assert target.read_text(encoding="utf-8") == "previous report"
+    assert not list(tmp_path.glob(".*.tmp"))  # no orphaned temp files