perf(ops): streamline log capture and git error reporting

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

scripts/release_provenance_gate.py

@@ -77,7 +77,7 @@ def main(argv: list[str] | None = None) -> int:
         item.get("matches_expected_revision", False) for item in report["runtime_file_verification"]
     )
     oci_ok = bool(report["artifact_identity"]["oci_revision_matches_head"])
-    allow_dirty_passed = bool(args.allow_dirty and dirty_checkout and runtime_ok and oci_ok)
+    allow_dirty_passed = bool(args.allow_dirty and runtime_ok and oci_ok)
     report["gate"]["allow_dirty"] = bool(args.allow_dirty)
     report["gate"]["allow_dirty_passed"] = allow_dirty_passed
 

tests/adapters/persistence/test_ops_checks.py

@@ -12,6 +12,8 @@ import uuid
 from datetime import UTC, date, datetime
 from pathlib import Path
 
+import pytest
+
 from trex_core.adapters.persistence.ops_checks import (
     collect_release_provenance,
     find_ledger_gap_slots,
@@ -179,3 +181,18 @@ def test_collect_release_provenance_rejects_runtime_path_escape(tmp_path):
     assert runtime["path_within_checkout"] is False
     assert runtime["matches_expected_revision"] is False
     assert report["gate"]["passed"] is False
+
+
+def test_collect_release_provenance_reports_git_failure_context(tmp_path):
+    not_repo = tmp_path / "not-repo"
+    not_repo.mkdir()
+    with pytest.raises(RuntimeError, match="git command failed in checkout="):
+        collect_release_provenance(
+            actor="operator-a",
+            reason="deploy",
+            instance="inst-a",
+            command_class="manual-start",
+            checkout_dir=not_repo,
+            capture_commands={},
+            broker_log_path=None,
+        )

trex_core/adapters/persistence/ops_checks.py

@@ -15,6 +15,7 @@ import hashlib
 import json
 import shutil
 import subprocess
+from collections import deque
 from datetime import UTC, date, datetime
 from pathlib import Path
 from typing import Any
@@ -56,11 +57,18 @@ def find_ledger_gap_slots(
 
 
 def _run_git(checkout_dir: Path, *args: str) -> str:
+    command = ["git", *args]
+    try:
         return subprocess.check_output(
-        ["git", *args],
+            command,
             cwd=str(checkout_dir),
             text=True,
         ).strip()
+    except subprocess.CalledProcessError as exc:
+        joined = " ".join(command)
+        raise RuntimeError(
+            f"git command failed in checkout={checkout_dir}: {joined}"
+        ) from exc
 
 
 def _porcelain_paths(status_output: str) -> list[str]:
@@ -125,8 +133,9 @@ def _capture_command(command: list[str], checkout_dir: Path) -> dict:
 def _capture_broker_side_effect_logs(log_path: Path) -> dict:
     if not log_path.exists():
         return {"path": str(log_path), "exists": False, "matches": []}
-    matches: list[str] = []
-    for line in log_path.read_text(encoding="utf-8", errors="replace").splitlines():
+    matches: deque[str] = deque(maxlen=200)
+    with log_path.open("r", encoding="utf-8", errors="replace") as fp:
+        for line in fp:
             lowered = line.lower()
             if (
                 "event=order_submit" in lowered
@@ -134,8 +143,8 @@ def _capture_broker_side_effect_logs(log_path: Path) -> dict:
                 or "event=order_reject" in lowered
                 or "cancel" in lowered
             ):
-            matches.append(line)
-    return {"path": str(log_path), "exists": True, "matches": matches[-200:]}
+                matches.append(line.rstrip("\n"))
+    return {"path": str(log_path), "exists": True, "matches": list(matches)}
 
 
 def collect_release_provenance(