gdpr / server /app.py
Developer
Fix inference JSONL logs, judge_pack guard, and grading calibration
3e6445f
Raw History Blame Contribute Delete
34.3 kB
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 """
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>GDPR EraseOps</title>
<style>
@import url('https://fonts.googleapis.com/css2?family=Space+Grotesk:wght@500;700&family=IBM+Plex+Sans:wght@400;500;600&display=swap');
:root {
--bg: #f4f7f3;
--panel: #ffffff;
--text: #122015;
--muted: #4a5f4d;
--accent: #1f7a45;
--accent-2: #0f4f2c;
--line: #d7e4d8;
}
* { box-sizing: border-box; }
body {
margin: 0;
font-family: 'IBM Plex Sans', sans-serif;
color: var(--text);
background:
radial-gradient(circle at 10% -10%, #d6f0dd 0%, transparent 45%),
radial-gradient(circle at 110% 0%, #e4efe8 0%, transparent 40%),
var(--bg);
}
.wrap {
max-width: 1080px;
margin: 0 auto;
padding: 28px 18px 40px;
}
.hero {
background: linear-gradient(150deg, #ffffff 0%, #f0f6f2 100%);
border: 1px solid var(--line);
border-radius: 22px;
padding: 22px;
box-shadow: 0 14px 40px rgba(18, 32, 21, 0.08);
}
h1, h2 {
font-family: 'Space Grotesk', sans-serif;
margin: 0;
}
h1 {
font-size: clamp(1.9rem, 4.6vw, 3rem);
line-height: 1.02;
letter-spacing: -0.02em;
}
.sub {
color: var(--muted);
margin-top: 10px;
line-height: 1.55;
}
.pill {
display: inline-flex;
align-items: center;
gap: 8px;
margin-top: 14px;
background: #e8f6ee;
color: var(--accent-2);
border: 1px solid #b7ddc2;
border-radius: 999px;
padding: 8px 12px;
font-size: 0.92rem;
font-weight: 600;
}
.dot {
width: 8px;
height: 8px;
border-radius: 50%;
background: var(--accent);
box-shadow: 0 0 0 5px rgba(31, 122, 69, 0.16);
}
.grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(230px, 1fr));
gap: 12px;
margin-top: 18px;
}
.card {
background: var(--panel);
border: 1px solid var(--line);
border-radius: 14px;
padding: 14px;
}
.card h2 {
font-size: 1.05rem;
margin-bottom: 8px;
}
.small {
color: var(--muted);
font-size: 0.93rem;
line-height: 1.45;
}
.links {
margin-top: 18px;
display: flex;
flex-wrap: wrap;
gap: 10px;
}
a {
text-decoration: none;
color: #ffffff;
background: var(--accent);
border: 1px solid #156238;
border-radius: 10px;
padding: 9px 12px;
font-weight: 600;
font-size: 0.92rem;
}
a.alt {
color: var(--accent-2);
background: #edf5ef;
border-color: #c2d8c8;
}
</style>
</head>
<body>
<main class="wrap">
<section class="hero">
<h1>GDPR EraseOps</h1>
<p class="sub">
Enterprise-grade reinforcement learning benchmark for GDPR right-to-erasure workflows,
with legal-hold constraints, dependency-order safety, and contamination risk controls.
</p>
<div class="pill"><span class="dot"></span>Service status: running</div>
<div class="grid">
<article class="card">
<h2>Live Health</h2>
<p class="small">Check deployment liveness and current episode pointer.</p>
<div class="links"><a class="alt" href="/health">Open /health</a></div>
</article>
<article class="card">
<h2>Task Catalog</h2>
<p class="small">List canonical tasks with budget and grading metadata.</p>
<div class="links"><a class="alt" href="/tasks">Open /tasks</a></div>
</article>
<article class="card">
<h2>Judge Snapshot</h2>
<p class="small">Single payload for current state, metrics, and score.</p>
<div class="links"><a class="alt" href="/judge_pack">Open /judge_pack</a></div>
</article>
<article class="card">
<h2>Advanced Reports</h2>
<p class="small">Run benchmark matrix, forensic analysis, and weighted evaluation tracks.</p>
<div class="links">
<a class="alt" href="/benchmark_matrix?episodes_per_task=1&seed_start=42">Benchmark</a>
<a class="alt" href="/forensic_audit?task_name=legal_minefield&seed=42&policy=greedy">Forensic</a>
<a class="alt" href="/evaluation_report?policy=reasoning&episodes_per_task=1&seed_start=42&include_hidden=true&hidden_weight=0.35">Eval</a>
</div>
</article>
</div>
<div class="links">
<a href="/docs">API Docs</a>
<a href="/metrics">Metrics</a>
<a href="/replay">Replay</a>
<a href="/judge_quickstart">Judge Quickstart</a>
<a href="/showcase">Showcase</a>
</div>
</section>
</main>
</body>
</html>
"""
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 """
<!doctype html>
<html lang=\"en\">
<head>
<meta charset=\"utf-8\" />
<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\" />
<title>GDPR EraseOps Showcase</title>
<style>
body { margin:0; font-family: 'IBM Plex Sans', sans-serif; background:#f3f7f4; color:#132117; }
.wrap { max-width:1000px; margin:0 auto; padding:24px 16px 36px; }
.hero { background:#fff; border:1px solid #d6e3d8; border-radius:16px; padding:18px; }
h1 { margin:0 0 8px; font-family:'Space Grotesk', sans-serif; font-size:2rem; }
.muted { color:#49604d; }
.grid { display:grid; grid-template-columns:repeat(auto-fit,minmax(280px,1fr)); gap:12px; margin-top:14px; }
.card { background:#fff; border:1px solid #d6e3d8; border-radius:12px; padding:14px; }
a { text-decoration:none; color:#0f5f33; font-weight:600; }
code { background:#eef5ef; padding:2px 6px; border-radius:6px; }
</style>
</head>
<body>
<main class=\"wrap\">
<section class=\"hero\">
<h1>GDPR EraseOps Showcase</h1>
<p class=\"muted\">Judge-first panel for quick evidence of robustness, safety, and reproducibility.</p>
</section>
<section class=\"grid\">
<article class=\"card\"><h3>Compliance Safety</h3><p class=\"muted\">Legal-hold, dependency-order, and contamination failures are first-class scored signals.</p></article>
<article class=\"card\"><h3>Auditability</h3><p class=\"muted\">Use <code>/judge_pack</code>, <code>/forensic_audit</code>, and <code>/replay</code> for deterministic trace review.</p></article>
<article class=\"card\"><h3>Benchmark Rigor</h3><p class=\"muted\">Compare policies through <code>/benchmark_matrix</code> and hidden-track weighted reports.</p></article>
</section>
</main>
</body>
</html>
"""
def main() -> None:
uvicorn.run(app, host="0.0.0.0", port=7860)
if __name__ == "__main__":
main()