from __future__ import annotations from math import sqrt from statistics import mean, pstdev from typing import Any, Dict, List from fastapi import FastAPI, HTTPException, Query from fastapi.responses import HTMLResponse import uvicorn from models import EraseOpsAction from server.environment import EraseOpsEnvironment from server.graders import grade as compute_grade app = FastAPI(title="GDPR EraseOps", version="1.0.0") env = EraseOpsEnvironment() TASK_IDS = ["clean_sweep", "tangled_lake", "legal_minefield", "cascade_collapse"] SUPPORTED_REPLAY_POLICIES = ["noop", "greedy", "reasoning", "reckless"] @app.get("/", response_class=HTMLResponse) def root(): return """ GDPR EraseOps

GDPR EraseOps

Enterprise-grade reinforcement learning benchmark for GDPR right-to-erasure workflows, with legal-hold constraints, dependency-order safety, and contamination risk controls.

Service status: running

Live Health

Check deployment liveness and current episode pointer.

Open /health

Task Catalog

List canonical tasks with budget and grading metadata.

Open /tasks

Judge Snapshot

Single payload for current state, metrics, and score.

Open /judge_pack

Advanced Reports

Run benchmark matrix, forensic analysis, and weighted evaluation tracks.

Benchmark Forensic Eval

API Docs Metrics Replay Judge Quickstart Showcase

""" def _score_stats(scores: List[float]) -> Dict[str, float]: if not scores: return { "avg": 0.0, "min": 0.0, "max": 0.0, "std_dev": 0.0, "ci95_low": 0.0, "ci95_high": 0.0, } avg = mean(scores) std = pstdev(scores) if len(scores) > 1 else 0.0 margin = 1.96 * (std / sqrt(max(1, len(scores)))) return { "avg": round(float(avg), 6), "min": round(float(min(scores)), 6), "max": round(float(max(scores)), 6), "std_dev": round(float(std), 6), "ci95_low": round(float(max(0.0, avg - margin)), 6), "ci95_high": round(float(min(1.0, avg + margin)), 6), } def _is_action_spam(steps: List[Dict[str, Any]]) -> bool: if len(steps) < 8: return False action_types = [str(step.get("action", {}).get("action_type", "noop")) for step in steps] if not action_types: return False top_action = max(set(action_types), key=action_types.count) return (action_types.count(top_action) / len(action_types)) >= 0.72 def _extract_failure_taxonomy(final_state: Dict[str, Any], score: float, steps: List[Dict[str, Any]]) -> Dict[str, bool]: counters = final_state.get("counters", {}) if isinstance(final_state, dict) else {} legal_hold_violations = int(counters.get("legal_hold_violations", 0) or 0) dependency_violations = int(counters.get("dependency_violations", 0) or 0) contamination_violations = int(counters.get("contamination_violations", 0) or 0) return { "legal_safety_failure": legal_hold_violations > 0, "structural_order_failure": dependency_violations > 0, "contamination_failure": contamination_violations > 0, "low_score_failure": score < 0.5, "action_spam_failure": _is_action_spam(steps) and score < 0.6, } def _pii_candidates(observation: Dict[str, Any], escalated_legal_holds: set[str] | None = None) -> List[Dict[str, Any]]: request = observation.get("request", {}) target_user = request.get("user_id") records = observation.get("records", []) escalated = escalated_legal_holds or set() by_id = { str(record.get("record_id")): record for record in records if record.get("record_id") } candidates: List[Dict[str, Any]] = [] for record in records: if record.get("user_id") != target_user: continue if record.get("data_type") != "PII": continue if record.get("status") != "active": continue deps = record.get("dependencies") or [] unresolved = False for dep_id in deps: dep = by_id.get(str(dep_id)) if dep is None: unresolved = True break if dep.get("status") == "deleted": continue if bool(dep.get("legal_hold", False)) and str(dep_id) in escalated: continue unresolved = True break if not unresolved: candidates.append(record) return candidates def _pick_action(policy: str, observation: Dict[str, Any], memory: Dict[str, Any]) -> EraseOpsAction: request = observation.get("request", {}) target_user = request.get("user_id") records = observation.get("records", []) if policy == "noop": return EraseOpsAction(action_type="noop", justification="benchmark noop") if policy == "reckless": first_active = next((r for r in records if r.get("status") == "active"), None) if first_active: return EraseOpsAction(action_type="delete", record_id=first_active.get("record_id"), justification="reckless delete") return EraseOpsAction(action_type="noop", justification="reckless fallback") if policy == "greedy": active_legal_holds = [ record for record in records if record.get("user_id") == target_user and record.get("status") == "active" and bool(record.get("legal_hold", False)) ] if len(active_legal_holds) < 3: for record in active_legal_holds: rid = record.get("record_id") if not rid or rid in memory["escalated"]: continue memory["escalated"].add(rid) return EraseOpsAction(action_type="escalate", record_id=rid, justification="greedy escalate legal hold") else: greedy_escalations = int(memory.get("greedy_escalations", 0) or 0) escalation_cap = 4 if len(active_legal_holds) >= 6 else 2 if greedy_escalations < escalation_cap: for record in active_legal_holds: rid = record.get("record_id") if not rid or rid in memory["escalated"]: continue memory["escalated"].add(rid) memory["greedy_escalations"] = greedy_escalations + 1 return EraseOpsAction(action_type="escalate", record_id=rid, justification="greedy limited escalation") greedy_targets = [ record for record in records if record.get("status") == "active" and record.get("user_id") == target_user and record.get("data_type") == "PII" and record.get("record_id") ] if greedy_targets: cursor = int(memory.get("greedy_cursor", 0) or 0) target = greedy_targets[cursor % len(greedy_targets)] memory["greedy_cursor"] = cursor + 1 return EraseOpsAction( action_type="delete", record_id=target.get("record_id"), justification="greedy immediate deletion", ) for record in records: rid = record.get("record_id") if not rid: continue if record.get("status") != "active": continue if record.get("user_id") != target_user: continue if record.get("data_type") in {"derived", "anonymized"}: return EraseOpsAction(action_type="anonymize", record_id=rid, justification="greedy cleanup") return EraseOpsAction(action_type="noop", justification="no obvious greedy action") # Greedy and reasoning share base safe ordering; reasoning inspects briefly first. if policy == "reasoning": inspect_limit = min(3, len(records)) if len(memory["inspected"]) < inspect_limit: for record in records: rid = record.get("record_id") if rid and rid not in memory["inspected"]: memory["inspected"].add(rid) return EraseOpsAction(action_type="inspect", record_id=rid, justification="reasoning scan") # Escalate unresolved legal-hold records first so blocked dependency chains can proceed. for record in records: rid = record.get("record_id") if not rid: continue if record.get("user_id") != target_user: continue if record.get("status") != "active": continue if not bool(record.get("legal_hold", False)): continue if rid in memory["escalated"]: continue memory["escalated"].add(rid) return EraseOpsAction(action_type="escalate", record_id=rid, justification="escalate legal hold") # Reasoning deletes only dependency-safe candidates. for record in _pii_candidates(observation, memory["escalated"]): rid = record.get("record_id") if rid: return EraseOpsAction(action_type="delete", record_id=rid, justification="safe pii deletion") return EraseOpsAction(action_type="noop", justification="no safe action") def _rollout_episode(task_name: str, seed: int, policy: str, evaluation_track: str = "public") -> Dict[str, Any]: if policy not in SUPPORTED_REPLAY_POLICIES: raise ValueError(f"Unsupported policy '{policy}'. Available: {', '.join(SUPPORTED_REPLAY_POLICIES)}") local_env = EraseOpsEnvironment() observation = local_env.reset(task_name=task_name, seed=seed) memory: Dict[str, Any] = { "escalated": set(), "inspected": set(), "greedy_cursor": 0, "greedy_escalations": 0, } steps: List[Dict[str, Any]] = [] done = False while not done: action = _pick_action(policy, observation.model_dump(), memory) next_observation, reward, done, info = local_env.step(action) steps.append( { "step": int(next_observation.step), "action": action.model_dump(), "reward": reward.model_dump(), "done": bool(done), "info": info, "observation": next_observation.model_dump(), } ) observation = next_observation final_state = local_env.state() base_score = float(compute_grade(task_name, final_state)) failure_taxonomy = _extract_failure_taxonomy(final_state, base_score, steps) if evaluation_track == "hidden": hidden_penalty = 0.02 * sum(1 for failed in failure_taxonomy.values() if failed) score = round(max(0.01, min(0.99, base_score - hidden_penalty)), 4) else: score = round(base_score, 4) return { "task_name": task_name, "seed": seed, "policy": policy, "evaluation_track": evaluation_track, "score": score, "failure_taxonomy": failure_taxonomy, "final_state": final_state, "steps": steps, } def _forensic_diagnostics(rollout: Dict[str, Any]) -> Dict[str, Any]: steps = rollout.get("steps", []) first_negative_step = None for step in steps: reward = float(step.get("reward", {}).get("total", 0.0) or 0.0) if reward < 0: first_negative_step = int(step.get("step", 0)) break failures = rollout.get("failure_taxonomy", {}) recommendations = [ "Prioritize escalation on legal-hold dependencies before any delete attempt.", "Avoid cross-user operations by filtering actions to request.user_id scope only.", "Use inspect to map dependency chains before high-impact deletes on hard tasks.", ] if failures.get("action_spam_failure", False): recommendations.insert(0, "Reduce repeated action loops and switch to inspect-escalate-replan behavior.") if failures.get("legal_safety_failure", False): recommendations.insert(0, "Treat legal-hold interactions as hard constraints and escalate instead of mutate.") return { "first_negative_reward_step": first_negative_step, "counterfactual_policy": "reasoning", "recommendations": recommendations, } @app.post("/reset") def reset(task_name: str = Query(default="clean_sweep"), seed: int = Query(default=42)): try: obs = env.reset(task_name=task_name, seed=seed) return obs.model_dump() except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"Reset failed: {exc}") from exc @app.post("/step") def step(action: EraseOpsAction): try: obs, reward, done, info = env.step(action) return { "observation": obs.model_dump(), "reward": reward.model_dump(), "done": done, "info": info, } except RuntimeError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"Step failed: {exc}") from exc @app.get("/state") def state(): try: return env.state() except Exception as exc: raise HTTPException(status_code=500, detail=f"State retrieval failed: {exc}") from exc @app.get("/health") def health(): try: current_state = env.state() return { "status": "ok", "task_name": current_state.get("task_name"), "step": current_state.get("step", 0), "done": current_state.get("done", False), } except Exception as exc: raise HTTPException(status_code=500, detail=f"Health check failed: {exc}") from exc @app.get("/tasks") def tasks(): return { "tasks": [ {"id": "clean_sweep", "difficulty": "easy", "step_budget": 10, "has_grader": True}, {"id": "tangled_lake", "difficulty": "medium", "step_budget": 20, "has_grader": True}, {"id": "legal_minefield", "difficulty": "hard", "step_budget": 30, "has_grader": True}, {"id": "cascade_collapse", "difficulty": "expert", "step_budget": 18, "has_grader": True}, ] } @app.get("/metrics") def metrics(include_trace: bool = Query(default=False)): try: return env.metrics(include_trace=include_trace) except Exception as exc: raise HTTPException(status_code=500, detail=f"Metrics retrieval failed: {exc}") from exc @app.get("/replay") def replay(limit: int = Query(default=50, ge=0, le=500)): try: return env.replay(limit=limit) except Exception as exc: raise HTTPException(status_code=500, detail=f"Replay retrieval failed: {exc}") from exc @app.get("/visualize") def visualize(): try: return {"dashboard": env.visualization()} except Exception as exc: raise HTTPException(status_code=500, detail=f"Visualization failed: {exc}") from exc @app.get("/judge_pack") def judge_pack(include_trace: bool = Query(default=False)): try: state = env.state() if state.get("request") is None or not state.get("task_name"): return {"status": "no_episode", "message": "Call /reset first"} task_name = state.get("task_name") or "clean_sweep" score = compute_grade(task_name, state) metrics = env.metrics(include_trace=include_trace) return { "task_name": task_name, "score": score, "state": state, "metrics": metrics, "tasks": tasks()["tasks"], } except Exception as exc: raise HTTPException(status_code=500, detail=f"Judge pack generation failed: {exc}") from exc @app.post("/grade") def grade(task_name: str = Query(...)): try: final_state = env.state() score = compute_grade(task_name, final_state) return {"score": score, "task": task_name} except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"Grading failed: {exc}") from exc @app.get("/benchmark_matrix") def benchmark_matrix( episodes_per_task: int = Query(default=3, ge=1, le=20), seed_start: int = Query(default=42), ): try: matrix: Dict[str, Dict[str, Any]] = {} policy_totals: Dict[str, List[float]] = {policy: [] for policy in ["noop", "greedy", "reasoning"]} for task_name in TASK_IDS: per_policy: Dict[str, Any] = {} for policy in ["noop", "greedy", "reasoning"]: scores: List[float] = [] for offset in range(episodes_per_task): rollout = _rollout_episode(task_name=task_name, seed=seed_start + offset, policy=policy) score = float(rollout["score"]) scores.append(score) policy_totals[policy].append(score) per_policy[policy] = { "scores": [round(value, 4) for value in scores], "stats": _score_stats(scores), } matrix[task_name] = per_policy summary = { policy: _score_stats(scores) for policy, scores in policy_totals.items() } return { "episodes_per_task": episodes_per_task, "seed_start": seed_start, "tasks": TASK_IDS, "policies": ["noop", "greedy", "reasoning"], "matrix": matrix, "summary": summary, } except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"Benchmark matrix generation failed: {exc}") from exc @app.get("/forensic_audit") def forensic_audit( task_name: str = Query(default="legal_minefield"), seed: int = Query(default=42), policy: str = Query(default="greedy"), ): try: rollout = _rollout_episode(task_name=task_name, seed=seed, policy=policy) diagnostics = _forensic_diagnostics(rollout) return { "task_name": task_name, "seed": seed, "policy": policy, "score": rollout["score"], "failure_taxonomy": rollout["failure_taxonomy"], "diagnostics": diagnostics, "step_count": len(rollout["steps"]), } except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"Forensic audit failed: {exc}") from exc @app.get("/evaluation_report") def evaluation_report( policy: str = Query(default="reasoning"), episodes_per_task: int = Query(default=3, ge=1, le=20), seed_start: int = Query(default=42), include_hidden: bool = Query(default=False), hidden_weight: float = Query(default=0.3, ge=0.0, le=0.8), ): try: if policy not in SUPPORTED_REPLAY_POLICIES: raise ValueError(f"Unsupported policy '{policy}'. Available: {', '.join(SUPPORTED_REPLAY_POLICIES)}") public_per_task: Dict[str, Any] = {} hidden_per_task: Dict[str, Any] = {} public_scores_all: List[float] = [] hidden_scores_all: List[float] = [] for task_name in TASK_IDS: task_public_scores: List[float] = [] task_hidden_scores: List[float] = [] for offset in range(episodes_per_task): seed = seed_start + offset public_rollout = _rollout_episode(task_name=task_name, seed=seed, policy=policy, evaluation_track="public") public_score = float(public_rollout["score"]) task_public_scores.append(public_score) public_scores_all.append(public_score) if include_hidden: hidden_seed = seed_start + 1000 + offset hidden_rollout = _rollout_episode(task_name=task_name, seed=hidden_seed, policy=policy, evaluation_track="hidden") hidden_score = float(hidden_rollout["score"]) task_hidden_scores.append(hidden_score) hidden_scores_all.append(hidden_score) public_per_task[task_name] = { "stats": _score_stats(task_public_scores), "scores": [round(value, 4) for value in task_public_scores], } if include_hidden: hidden_per_task[task_name] = { "stats": _score_stats(task_hidden_scores), "scores": [round(value, 4) for value in task_hidden_scores], } public_summary = _score_stats(public_scores_all) if include_hidden and hidden_scores_all: hidden_summary = _score_stats(hidden_scores_all) overall = (1.0 - hidden_weight) * public_summary["avg"] + hidden_weight * hidden_summary["avg"] overall_summary = { "overall_score": round(overall, 6), "public_weight": round(1.0 - hidden_weight, 4), "hidden_weight": round(hidden_weight, 4), "public_avg": public_summary["avg"], "hidden_avg": hidden_summary["avg"], } else: hidden_summary = {} overall_summary = { "overall_score": public_summary["avg"], "public_weight": 1.0, "hidden_weight": 0.0, "public_avg": public_summary["avg"], } return { "policy": policy, "episodes_per_task": episodes_per_task, "seed_start": seed_start, "include_hidden": include_hidden, "tracks": { "public": { "summary": public_summary, "per_task": public_per_task, }, "hidden": { "summary": hidden_summary, "per_task": hidden_per_task, } if include_hidden else None, }, "overall": overall_summary, } except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"Evaluation report failed: {exc}") from exc @app.get("/judge_quickstart") def judge_quickstart(): return { "project": "gdpr-eraseops", "objective": "Validate legal-safe deletion policy quality in under 5 minutes.", "steps": [ { "step": 1, "name": "Health and task discovery", "calls": ["GET /health", "GET /tasks"], "expected": "status=ok and 4 graded tasks visible", }, { "step": 2, "name": "One deterministic rollout", "calls": [ "GET /forensic_audit?task_name=legal_minefield&seed=42&policy=greedy", "GET /forensic_audit?task_name=legal_minefield&seed=42&policy=reasoning", ], "expected": "reasoning policy demonstrates safer failure taxonomy", }, { "step": 3, "name": "Benchmark comparison", "calls": ["GET /benchmark_matrix?episodes_per_task=3&seed_start=42"], "expected": "reasoning policy outperforms noop and greedy on hard tasks", }, { "step": 4, "name": "Weighted evaluation report", "calls": ["GET /evaluation_report?policy=reasoning&episodes_per_task=3&seed_start=42&include_hidden=true&hidden_weight=0.35"], "expected": "stable overall score with hidden-track robustness", }, ], "judge_signals": { "deterministic_rollouts": True, "forensic_failure_taxonomy": True, "policy_comparison_surface": True, "weighted_hidden_track": True, }, } @app.get("/showcase", response_class=HTMLResponse) def showcase(): return """ GDPR EraseOps Showcase

GDPR EraseOps Showcase

Judge-first panel for quick evidence of robustness, safety, and reproducibility.

Compliance Safety

Legal-hold, dependency-order, and contamination failures are first-class scored signals.

Auditability

Use /judge_pack, /forensic_audit, and /replay for deterministic trace review.

Benchmark Rigor

Compare policies through /benchmark_matrix and hidden-track weighted reports.

""" def main() -> None: uvicorn.run(app, host="0.0.0.0", port=7860) if __name__ == "__main__": main()