File size: 2,581 Bytes
12496fc | 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 | """Explicit opt-in, local failure triage; no automatic upload or training admission."""
from datetime import datetime, timezone
from pathlib import Path
import json
import uuid
from .data import SECRET, EMAIL
CATEGORIES = {"reasoning", "code", "tool_misuse", "hallucination", "premature_completion", "looping", "context", "memory", "voice", "latency"}
STAGES = ["captured", "classified", "reproduced", "minimized", "test_added", "candidate", "trained", "reevaluated"]
class FailureStore:
def __init__(self, root):
self.root = Path(root)
self.root.mkdir(parents=True, exist_ok=True)
def capture(self, category, summary, source, *, consent=False):
if not consent:
raise PermissionError("Failure content capture requires explicit consent")
if category not in CATEGORIES or not summary or not source:
raise ValueError("Invalid failure record")
if SECRET.search(summary) or SECRET.search(source):
raise ValueError("Credential pattern in failure record")
record = {"id": uuid.uuid4().hex, "category": category, "summary": EMAIL.sub("[EMAIL]", summary),
"source": EMAIL.sub("[EMAIL]", source), "stage": "captured", "created": datetime.now(timezone.utc).isoformat(),
"evidence": [], "training_approved": False}
self._write(record)
return record["id"]
def advance(self, record_id, stage, evidence, *, training_approved=False):
if len(record_id) != 32 or any(c not in "0123456789abcdef" for c in record_id):
raise ValueError("Invalid failure ID")
path = self.root / (record_id + ".json")
record = json.loads(path.read_text(encoding="utf-8"))
if stage not in STAGES or STAGES.index(stage) != STAGES.index(record["stage"])+1 or not evidence:
raise ValueError("Require next stage and evidence")
if SECRET.search(evidence):
raise ValueError("Credential pattern in evidence")
if stage == "candidate" and not training_approved:
raise PermissionError("Training admission needs explicit provenance/consent review")
record["stage"] = stage
record["evidence"].append(EMAIL.sub("[EMAIL]", evidence))
record["training_approved"] = record["training_approved"] or training_approved
self._write(record)
def _write(self, record):
path = self.root / (record["id"] + ".json")
temp = path.with_suffix(".tmp")
temp.write_text(json.dumps(record, indent=2), encoding="utf-8")
temp.replace(path)
|