Spaces:
Paused
Paused
File size: 2,513 Bytes
6b3874e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """Persistence for retro artifacts under runs/retro/.
runs/retro/analyses/<evaluation_id>.json scorecard + metadata
runs/retro/audit/<analysis_id>.jsonl tool-call audit trail
runs/retro/eval_cases/<case_id>.json cases created by the write tool
runs/retro/uploads/<name>.json uploaded logs (allowlisted reads)
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Dict, List, Optional
def root() -> Path:
"""Resolved per call so tests can isolate via RETRO_RUNS_DIR."""
return Path(os.environ.get("RETRO_RUNS_DIR", "runs/retro"))
def _dir(name: str) -> Path:
d = root() / name
d.mkdir(parents=True, exist_ok=True)
return d
def analyses_dir() -> Path:
return _dir("analyses")
def audit_dir() -> Path:
return _dir("audit")
def eval_cases_dir() -> Path:
return _dir("eval_cases")
def uploads_dir() -> Path:
return _dir("uploads")
def save_analysis(evaluation_id: str, payload: Dict) -> Path:
path = analyses_dir() / f"{evaluation_id}.json"
path.write_text(json.dumps(payload, indent=2))
return path
def load_analysis(evaluation_id: str) -> Optional[Dict]:
path = analyses_dir() / f"{evaluation_id}.json"
if not path.exists():
return None
return json.loads(path.read_text())
def list_analyses() -> List[Dict]:
out = []
for p in sorted(analyses_dir().glob("*.json")):
try:
data = json.loads(p.read_text())
sc = data.get("scorecard", {})
out.append({
"evaluation_id": sc.get("evaluation_id", p.stem),
"created_at": sc.get("created_at"),
"failure_type": sc.get("failure_type"),
"severity": sc.get("severity"),
"launch_recommendation": sc.get("launch_recommendation"),
"backend_used": sc.get("backend_used"),
"human_review_required": sc.get("human_review_required"),
})
except (json.JSONDecodeError, OSError):
continue
return out
def append_audit(analysis_id: str, record: Dict) -> None:
path = audit_dir() / f"{analysis_id}.jsonl"
with path.open("a") as f:
f.write(json.dumps(record) + "\n")
def read_audit(analysis_id: str) -> List[Dict]:
path = audit_dir() / f"{analysis_id}.jsonl"
if not path.exists():
return []
return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
|