Merge PR #705: bin/soul_eval_gate.py

bin/soul_eval_gate.py

@@ -33,6 +33,11 @@ from pathlib import Path
 from typing import Optional
 
 
+# ── Gate File Rotation ──────────────────────────────────────────────
+GATE_FILE_MAX_AGE_DAYS = 7
+GATE_FILE_MAX_COUNT = 50
+
+
 # ── SOUL.md Constraints ──────────────────────────────────────────────
 #
 # These are the non-negotiable categories from SOUL.md and the
@@ -240,6 +245,9 @@ def evaluate_candidate(
     latest_file = gate_dir / "eval_gate_latest.json"
     latest_file.write_text(json.dumps(result, indent=2))
 
+    # Rotate old gate files to prevent unbounded growth
+    _rotate_gate_files(gate_dir)
+
     return result
 
 
@@ -249,6 +257,48 @@ def _load_json(path: str | Path) -> dict:
     return json.loads(Path(path).read_text())
 
 
+def _rotate_gate_files(gate_dir: Path) -> None:
+    """Clean up old gate files to prevent unbounded directory growth.
+
+    - Deletes files older than GATE_FILE_MAX_AGE_DAYS
+    - Caps total count at GATE_FILE_MAX_COUNT (oldest first)
+    - Always preserves eval_gate_latest.json
+    """
+    if not gate_dir.exists():
+        return
+
+    latest_name = "eval_gate_latest.json"
+    cutoff = datetime.now(timezone.utc).timestamp() - (GATE_FILE_MAX_AGE_DAYS * 86400)
+
+    gate_files = []
+    for f in gate_dir.iterdir():
+        if f.name == latest_name or not f.name.startswith("eval_gate_") or f.suffix != ".json":
+            continue
+        try:
+            mtime = f.stat().st_mtime
+        except OSError:
+            continue
+        gate_files.append((mtime, f))
+
+    # Sort oldest first
+    gate_files.sort(key=lambda x: x[0])
+
+    deleted = 0
+    for mtime, f in gate_files:
+        should_delete = False
+        if mtime < cutoff:
+            should_delete = True
+        elif len(gate_files) - deleted > GATE_FILE_MAX_COUNT:
+            should_delete = True
+
+        if should_delete:
+            try:
+                f.unlink()
+                deleted += 1
+            except OSError:
+                pass
+
+
 def _find_category_score(
     sessions: dict[str, dict],
     category: str,