| """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) |
|
|
