app.py

import os
import json
import time
import random
from datetime import datetime, timedelta
from typing import Dict, Any, List, Tuple

import gradio as gr
import pandas as pd
import numpy as np

# ============================
# Branding
# ============================
PROCELEVATE_BLUE = "#0F2C59"
CUSTOM_CSS = f"""
/* Primary buttons */
.gr-button.gr-button-primary,
button.primary {{
  background: {PROCELEVATE_BLUE} !important;
  border-color: {PROCELEVATE_BLUE} !important;
  color: white !important;
  font-weight: 650 !important;
}}
.gr-button.gr-button-primary:hover,
button.primary:hover {{
  filter: brightness(0.92);
}}
/* Tabs: selected */
button[data-testid="tab-button"][aria-selected="true"] {{
  border-bottom: 3px solid {PROCELEVATE_BLUE} !important;
  color: {PROCELEVATE_BLUE} !important;
  font-weight: 750 !important;
}}
.block, .gr-box, .gr-panel {{
  border-radius: 14px !important;
}}
"""

# ============================
# Storage
# ============================
DATA_DIR = "data"
LOG_FILE = os.path.join(DATA_DIR, "staff_enablement_logs.json")
SOP_FILE = os.path.join(DATA_DIR, "hotel_sops.json")

SUPERVISOR_PIN = os.environ.get("SUPERVISOR_PIN", "2580")  # demo PIN

def ensure_data_dir():
    os.makedirs(DATA_DIR, exist_ok=True)

def load_json(path: str, default):
    ensure_data_dir()
    if not os.path.exists(path):
        return default
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return default

def save_json(path: str, data):
    ensure_data_dir()
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

# ============================
# Demo SOP Knowledge Pack
# Replace with real hotel SOP later
# ============================
DEFAULT_SOPS: Dict[str, Dict[str, Any]] = {
    "Front Desk": {
        "topics": {
            "Early check-in": {
                "steps": [
                    "Confirm booking details and expected arrival time.",
                    "Check room readiness status (clean & inspected).",
                    "If room is not ready: offer luggage hold + lobby welcome + estimated ready time.",
                    "If early check-in fee applies: communicate clearly and confirm guest approval.",
                    "Update PMS notes and inform housekeeping if priority cleaning is needed."
                ],
                "escalation": "Escalate to Duty Manager if guest is VIP, irate, or if fee waiver is requested.",
                "policy": "Early check-in is subject to availability. Fees may apply based on arrival time and occupancy."
            },
            "Late checkout": {
                "steps": [
                    "Verify occupancy and next-day arrivals for the room type.",
                    "Offer options: 1-hour courtesy (if available) or paid late checkout.",
                    "Confirm cutoff time and any charges in writing/receipt notes.",
                    "Update PMS and inform housekeeping schedule."
                ],
                "escalation": "Escalate to Duty Manager for VIPs or if occupancy is high and exceptions are requested.",
                "policy": "Late checkout is subject to availability. Charges may apply after standard checkout time."
            },
            "Noise complaint": {
                "steps": [
                    "Apologize and acknowledge quickly. Confirm location/room number and time.",
                    "Call the source room politely with a first warning.",
                    "If persists: send security for a discreet check.",
                    "Offer room change or goodwill gesture if required.",
                    "Log incident in daily report."
                ],
                "escalation": "Escalate to Security Supervisor/Duty Manager if repeated, aggressive behavior, or safety risk.",
                "policy": "Quiet hours apply as per hotel policy; repeated disturbances may lead to eviction per management decision."
            },
            "ID verification": {
                "steps": [
                    "Request passport/ID as per local regulations.",
                    "Verify name matches booking and capture required fields securely.",
                    "If mismatch: confirm with booking channel or manager before proceeding.",
                    "Return ID promptly and thank guest."
                ],
                "escalation": "Escalate to Duty Manager if ID is missing/expired or guest refuses compliance.",
                "policy": "ID verification is mandatory for check-in as per regulatory compliance."
            },
            "Refund / cancellation": {
                "steps": [
                    "Check booking channel (direct/OTA) and cancellation policy.",
                    "Confirm timeline (cutoff) and fees.",
                    "If within policy: process refund workflow or advise OTA route.",
                    "Document communication and outcome in PMS/CRM notes."
                ],
                "escalation": "Escalate to Finance/Duty Manager for exception approvals or disputes.",
                "policy": "Refunds follow rate plan & channel policy; exceptions require approval."
            },
        }
    },
    "Housekeeping": {
        "topics": {
            "Room turnaround (standard)": {
                "steps": [
                    "Knock, announce, and confirm room is vacant/guest permission granted.",
                    "Strip linens, collect trash, and separate lost-and-found items.",
                    "Clean bathroom first (sanitation focus), then bedroom surfaces.",
                    "Replace linens, replenish amenities, and check minibar (if applicable).",
                    "Final inspection checklist: lights, AC, TV, odors, floor, bathroom shine."
                ],
                "escalation": "Escalate to HK Supervisor if room damage, biohazard, or missing inventory is found.",
                "policy": "Follow sanitation standards; document lost-and-found immediately."
            },
            "Extra towel request": {
                "steps": [
                    "Confirm quantity and delivery time preference.",
                    "Prepare towels and verify quality (no stains/tears).",
                    "Deliver within SLA (e.g., 10–15 minutes).",
                    "Update request log/notes if system exists."
                ],
                "escalation": "Escalate to HK Supervisor if repeated delays or stock shortage.",
                "policy": "Standard amenity fulfillment within defined SLA."
            },
        }
    },
    "F&B": {
        "topics": {
            "Dinner menu query": {
                "steps": [
                    "Ask preference: vegetarian/non-veg, allergies, spice level.",
                    "Share 3–5 popular items + price range.",
                    "Offer reservation or takeaway/room service option.",
                    "Confirm service timings and last order time."
                ],
                "escalation": "Escalate to Restaurant Supervisor for large groups, special diets, or VIP requests.",
                "policy": "Menu availability may vary; confirm specials with kitchen."
            },
            "Room service order": {
                "steps": [
                    "Confirm room number/name and order details.",
                    "Confirm allergies and cooking preferences.",
                    "Provide ETA and any service charges.",
                    "Hand off to kitchen + runner; confirm delivery completion."
                ],
                "escalation": "Escalate to F&B Supervisor if delays, complaints, or refunds required.",
                "policy": "Room service ETA targets apply; communicate proactively if delayed."
            },
        }
    },
    "Maintenance": {
        "topics": {
            "AC not cooling": {
                "steps": [
                    "Confirm room number and symptoms (not cooling, noise, leak).",
                    "Check thermostat settings and power cycle if safe.",
                    "Inspect filter and airflow; check for blockage.",
                    "If unresolved: dispatch technician and provide ETA to front desk/guest.",
                    "Log issue and resolution steps."
                ],
                "escalation": "Escalate to Chief Engineer for repeat failures, leaks, or safety risk.",
                "policy": "Guest comfort is priority; offer room move if repair exceeds threshold time."
            },
            "Wi-Fi complaint": {
                "steps": [
                    "Confirm floor/room and device type.",
                    "Guide basic steps: reconnect, forget network, restart.",
                    "Check AP status for affected area (if system exists).",
                    "If recurring: dispatch IT/maintenance support.",
                    "Log complaint for trend analysis."
                ],
                "escalation": "Escalate to IT/Engineering lead if multiple rooms affected.",
                "policy": "Connectivity issues should be acknowledged quickly; proactive updates reduce dissatisfaction."
            },
        }
    },
    "Security": {
        "topics": {
            "Suspicious activity": {
                "steps": [
                    "Observe discreetly; do not escalate publicly.",
                    "Confirm with CCTV (if available) and patrol report.",
                    "Approach politely if needed; maintain guest privacy.",
                    "If risk confirmed: follow incident protocol and notify Duty Manager."
                ],
                "escalation": "Immediate escalation for safety threats or illegal activity.",
                "policy": "Guest safety and discretion are top priorities."
            }
        }
    }
}

# Load SOP knowledge base (persisted) or default
SOPS = load_json(SOP_FILE, DEFAULT_SOPS)
if not SOPS:
    SOPS = DEFAULT_SOPS
    save_json(SOP_FILE, SOPS)

# Logs store
LOGS: List[Dict[str, Any]] = load_json(LOG_FILE, [])

# ============================
# Helper utilities
# ============================
def now_str():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def normalize(text: str) -> str:
    return (text or "").strip().lower()

def best_topic_match(role: str, question: str) -> Tuple[str, float]:
    """
    Simple keyword match against topic titles.
    Returns (topic_title, score)
    """
    q = normalize(question)
    topics = SOPS.get(role, {}).get("topics", {})
    if not topics:
        return "", 0.0

    # Token set
    q_tokens = set([t for t in q.replace("/", " ").replace("-", " ").split() if len(t) > 2])

    best_title, best_score = "", 0.0
    for title in topics.keys():
        t = normalize(title)
        t_tokens = set([x for x in t.replace("/", " ").replace("-", " ").split() if len(x) > 2])
        # overlap score
        overlap = len(q_tokens.intersection(t_tokens))
        score = overlap / max(len(t_tokens), 1)
        # boost if title phrase occurs
        if t in q:
            score += 0.6
        if score > best_score:
            best_score = score
            best_title = title

    return best_title, float(best_score)

def format_sop_answer(role: str, topic: str) -> str:
    data = SOPS.get(role, {}).get("topics", {}).get(topic)
    if not data:
        return "I couldn't find a matching SOP for that. Please rephrase or check with your supervisor."

    steps = data.get("steps", [])
    escalation = data.get("escalation", "")
    policy = data.get("policy", "")

    step_lines = "\n".join([f"{i+1}. {s}" for i, s in enumerate(steps)]) if steps else "No steps defined."
    esc = f"**Escalation:** {escalation}" if escalation else "**Escalation:** N/A"
    pol = f"**Policy note:** {policy}" if policy else "**Policy note:** N/A"

    return (
        f"### ✅ SOP Guidance — {role}\n"
        f"**Topic:** {topic}\n\n"
        f"**Step-by-step:**\n{step_lines}\n\n"
        f"{esc}\n\n"
        f"{pol}\n"
    )

def log_interaction(entry: Dict[str, Any]):
    LOGS.append(entry)
    save_json(LOG_FILE, LOGS)

def roles_list() -> List[str]:
    return list(SOPS.keys())

def topics_for_role(role: str) -> List[str]:
    return list(SOPS.get(role, {}).get("topics", {}).keys())

# ============================
# Micro-training scenarios (demo)
# ============================
SCENARIOS = {
    "Front Desk": [
        {
            "q": "Guest requests early check-in at 10 AM. What should you do first?",
            "options": [
                "Immediately confirm early check-in is guaranteed",
                "Check room readiness and explain availability/fee policy",
                "Ask housekeeping to clean all rooms now",
                "Refuse early check-in without checking"
            ],
            "answer_index": 1,
            "topic": "Early check-in"
        },
        {
            "q": "A guest complains about noise after 11 PM. What is the best next step?",
            "options": [
                "Ignore; it will stop",
                "Call the guest back tomorrow",
                "Acknowledge, warn source room, escalate if repeated",
                "Refund the guest immediately"
            ],
            "answer_index": 2,
            "topic": "Noise complaint"
        }
    ],
    "Housekeeping": [
        {
            "q": "Guest asks for 2 extra towels. What’s the correct response?",
            "options": [
                "Deliver within SLA and update request log",
                "Ask guest to come collect towels",
                "Deliver only if guest tips",
                "Deliver next day"
            ],
            "answer_index": 0,
            "topic": "Extra towel request"
        }
    ],
    "F&B": [
        {
            "q": "Guest asks for dinner menu before arrival. What should you do?",
            "options": [
                "Say you can’t share menu",
                "Ask preferences and share top items + timings",
                "Only give one item",
                "Ask guest to Google"
            ],
            "answer_index": 1,
            "topic": "Dinner menu query"
        }
    ],
    "Maintenance": [
        {
            "q": "AC is not cooling. What should be done first?",
            "options": [
                "Tell guest to wait",
                "Confirm details and check thermostat/power cycle if safe",
                "Change guest room immediately without checking",
                "Ignore if it’s late night"
            ],
            "answer_index": 1,
            "topic": "AC not cooling"
        }
    ],
}

def get_random_scenario(role: str) -> Dict[str, Any]:
    pool = SCENARIOS.get(role, [])
    if not pool:
        return {}
    return random.choice(pool)

# ============================
# Staff Assistant functions
# ============================
def handle_staff_question(role: str, staff_id: str, question: str):
    role = role or "Front Desk"
    staff_id = (staff_id or "").strip() or "Staff-Unknown"
    question = (question or "").strip()

    if not question:
        return "Please type a question."

    topic, score = best_topic_match(role, question)

    # If not confident, offer nearest topics
    if not topic or score < 0.25:
        suggestions = topics_for_role(role)[:8]
        sug = "\n".join([f"- {t}" for t in suggestions]) if suggestions else "- (No topics configured)"
        answer = (
            f"### ⚠️ Not confident about the match\n"
            f"I couldn't confidently map your question to an SOP topic.\n\n"
            f"Try asking using one of these topics:\n{sug}\n\n"
            f"Or rephrase with keywords (e.g., “late checkout policy”, “noise complaint steps”)."
        )
        log_interaction({
            "timestamp": now_str(),
            "type": "qa",
            "role": role,
            "staff_id": staff_id,
            "question": question,
            "matched_topic": "",
            "match_score": score,
            "result": "no_match"
        })
        return answer

    answer = format_sop_answer(role, topic)

    log_interaction({
        "timestamp": now_str(),
        "type": "qa",
        "role": role,
        "staff_id": staff_id,
        "question": question,
        "matched_topic": topic,
        "match_score": score,
        "result": "matched"
    })
    return answer

def start_training(role: str, staff_id: str):
    role = role or "Front Desk"
    staff_id = (staff_id or "").strip() or "Staff-Unknown"
    sc = get_random_scenario(role)
    if not sc:
        return "No training scenarios configured for this role yet.", None, None, None

    log_interaction({
        "timestamp": now_str(),
        "type": "training_start",
        "role": role,
        "staff_id": staff_id,
        "scenario_question": sc["q"],
        "topic": sc.get("topic", "")
    })

    return sc["q"], sc["options"], sc["answer_index"], sc.get("topic", "")

def submit_training_answer(role: str, staff_id: str, scenario_q: str, options: List[str], correct_idx: int, chosen: str, topic: str):
    role = role or "Front Desk"
    staff_id = (staff_id or "").strip() or "Staff-Unknown"

    if not scenario_q or options is None or correct_idx is None:
        return "Please click “Start Micro-Training” first."

    try:
        chosen_idx = options.index(chosen)
    except Exception:
        chosen_idx = -1

    ok = (chosen_idx == int(correct_idx))
    feedback = "✅ Correct." if ok else f"❌ Not correct. Best answer: **{options[int(correct_idx)]}**"

    # Attach SOP guidance for reinforcement
    sop = format_sop_answer(role, topic) if topic else ""
    msg = f"{feedback}\n\n{sop}"

    log_interaction({
        "timestamp": now_str(),
        "type": "training_answer",
        "role": role,
        "staff_id": staff_id,
        "scenario_question": scenario_q,
        "topic": topic,
        "chosen": chosen,
        "is_correct": ok
    })

    return msg

# ============================
# Supervisor analytics
# ============================
def supervisor_unlock(pin: str):
    if (pin or "").strip() == SUPERVISOR_PIN:
        return gr.update(visible=False), gr.update(visible=True), "✅ Supervisor access granted."
    return gr.update(visible=True), gr.update(visible=False), "❌ Incorrect PIN."

def logs_df() -> pd.DataFrame:
    if not LOGS:
        return pd.DataFrame(columns=["timestamp","type","role","staff_id","question","matched_topic","result","is_correct"])
    df = pd.DataFrame(LOGS)
    # Ensure columns exist
    for c in ["timestamp","type","role","staff_id","question","matched_topic","result","is_correct","topic"]:
        if c not in df.columns:
            df[c] = ""
    return df

def compute_readiness(df: pd.DataFrame) -> pd.DataFrame:
    """
    Demo readiness score:
    - training accuracy weighted higher
    - fewer repeated SOP Qs => higher
    - normalize to 0-100
    """
    if df.empty:
        return pd.DataFrame(columns=["staff_id","role","qa_count","no_match_count","training_attempts","training_accuracy","readiness_score"])

    qa = df[df["type"] == "qa"].copy()
    tr = df[df["type"] == "training_answer"].copy()

    # QA stats
    qa_stats = (
        qa.groupby(["staff_id","role"])
        .agg(
            qa_count=("type","count"),
            no_match_count=("result", lambda x: int((x=="no_match").sum())),
        )
        .reset_index()
    )

    # Repeated topics penalty (ask same topic too often)
    if not qa.empty:
        rep = (
            qa[qa["matched_topic"].fillna("") != ""]
            .groupby(["staff_id","role","matched_topic"])
            .size()
            .reset_index(name="topic_count")
        )
        rep_pen = (
            rep.groupby(["staff_id","role"])["topic_count"]
            .apply(lambda s: int((s >= 3).sum()))  # count topics asked 3+ times
            .reset_index(name="repeated_topics_3plus")
        )
    else:
        rep_pen = pd.DataFrame(columns=["staff_id","role","repeated_topics_3plus"])

    # Training stats
    if not tr.empty:
        tr["is_correct"] = tr["is_correct"].fillna(False).astype(bool)
        tr_stats = (
            tr.groupby(["staff_id","role"])
            .agg(
                training_attempts=("type","count"),
                training_correct=("is_correct","sum")
            )
            .reset_index()
        )
        tr_stats["training_accuracy"] = (tr_stats["training_correct"] / tr_stats["training_attempts"]).replace([np.inf, np.nan], 0.0)
    else:
        tr_stats = pd.DataFrame(columns=["staff_id","role","training_attempts","training_correct","training_accuracy"])

    # Merge
    out = pd.merge(qa_stats, tr_stats, on=["staff_id","role"], how="outer")
    out = pd.merge(out, rep_pen, on=["staff_id","role"], how="left")
    out = out.fillna(0)

    # Score formula (demo)
    # Base from training accuracy (0-70)
    # Penalty for no_match (0-10)
    # Penalty for repeated topics (0-10)
    # Bonus for healthy usage (0-10) (asking questions is good early on; too many is not)
    out["base"] = (out["training_accuracy"] * 70.0).clip(0, 70)
    out["pen_no_match"] = (out["no_match_count"] * 2.0).clip(0, 10)
    out["pen_repeat"] = (out["repeated_topics_3plus"] * 3.0).clip(0, 10)

    # usage bonus: if QA count in a reasonable range (1..20)
    out["bonus_usage"] = out["qa_count"].apply(lambda x: 10.0 if 3 <= x <= 15 else (6.0 if 1 <= x <= 25 else 2.0)).clip(0, 10)

    out["readiness_score"] = (out["base"] + out["bonus_usage"] - out["pen_no_match"] - out["pen_repeat"]).clip(0, 100).round(0)

    cols = ["staff_id","role","qa_count","no_match_count","training_attempts","training_accuracy","readiness_score"]
    out = out[cols].sort_values(["readiness_score","training_accuracy"], ascending=False)
    out["training_accuracy"] = (out["training_accuracy"]*100).round(0).astype(int).astype(str) + "%"
    return out

def top_questions(df: pd.DataFrame, role_filter: str = "All", n: int = 10) -> pd.DataFrame:
    if df.empty:
        return pd.DataFrame(columns=["role","question","count"])
    qa = df[df["type"] == "qa"].copy()
    if role_filter != "All":
        qa = qa[qa["role"] == role_filter]
    if qa.empty:
        return pd.DataFrame(columns=["role","question","count"])
    out = (
        qa.groupby(["role","question"])
        .size()
        .reset_index(name="count")
        .sort_values("count", ascending=False)
        .head(n)
    )
    return out

def confusion_hotspots(df: pd.DataFrame, n: int = 10) -> pd.DataFrame:
    if df.empty:
        return pd.DataFrame(columns=["role","hotspot","count"])
    qa = df[df["type"] == "qa"].copy()
    qa["matched_topic"] = qa["matched_topic"].fillna("")
    qa["hotspot"] = qa.apply(lambda r: r["matched_topic"] if r["matched_topic"] else "Unmapped / unclear SOP", axis=1)
    out = (
        qa.groupby(["role","hotspot"])
        .size()
        .reset_index(name="count")
        .sort_values("count", ascending=False)
        .head(n)
    )
    return out

def export_logs():
    df = logs_df()
    ensure_data_dir()
    path = os.path.join(DATA_DIR, "staff_enablement_logs_export.csv")
    df.to_csv(path, index=False)
    return path

def supervisor_clear(pin: str):
    global LOGS
    if (pin or "").strip() != SUPERVISOR_PIN:
        return "❌ Incorrect PIN. Cannot clear logs."
    LOGS = []
    save_json(LOG_FILE, LOGS)
    return f"✅ Cleared logs at {now_str()}."

# ============================
# SOP editor (optional)
# ============================
def get_sop_json_text():
    return json.dumps(SOPS, ensure_ascii=False, indent=2)

def save_sop_json_text(pin: str, text: str):
    global SOPS
    if (pin or "").strip() != SUPERVISOR_PIN:
        return "❌ Incorrect PIN. Cannot update SOPs."
    try:
        parsed = json.loads(text)
        if not isinstance(parsed, dict):
            return "❌ SOP JSON must be an object/dict at top level."
        SOPS = parsed
        save_json(SOP_FILE, SOPS)
        return f"✅ SOP knowledge updated at {now_str()}."
    except Exception as e:
        return f"❌ Invalid JSON: {e}"

# ============================
# UI
# ============================
with gr.Blocks(title="AI Staff Enablement & Continuity Assistant (Prototype)", css=CUSTOM_CSS) as demo:
    gr.Markdown(
        """
# 👥 AI Staff Enablement & Continuity Assistant (Prototype)

A role-based “AI Buddy” that helps hotel staff **learn while working**, reduces dependency on seniors, and preserves SOP knowledge even when employees join/leave.

✅ Role-based SOP guidance • Micro-training • Supervisor insights • Readiness scoring  
*(Demo uses sample SOPs; replace with your hotel SOPs during pilot.)*
"""
    )

    with gr.Tab("Staff Assistant"):
        with gr.Row():
            role = gr.Dropdown(roles_list(), value=roles_list()[0], label="Select your role")
            staff_id = gr.Textbox(label="Staff ID / Name (for demo)", placeholder="e.g., FD-021 / Anita")

        gr.Markdown("### Ask a work question (learn while doing)")
        question = gr.Textbox(
            label="Your question",
            placeholder="e.g., Guest wants early check-in at 10AM. What should I do?",
            lines=2
        )
        ask_btn = gr.Button("Get SOP Guidance", variant="primary")
        answer_md = gr.Markdown("")

        gr.Markdown("---")
        gr.Markdown("### Micro-Training (2 minutes)")
        train_btn = gr.Button("Start Micro-Training", variant="primary")
        scenario_q = gr.Textbox(label="Scenario", interactive=False, lines=2)
        scenario_options = gr.Radio(choices=[], label="Choose the best answer")
        hidden_correct = gr.State(None)
        hidden_topic = gr.State("")
        submit_ans_btn = gr.Button("Submit Answer", variant="primary")
        train_feedback = gr.Markdown("")

        ask_btn.click(handle_staff_question, inputs=[role, staff_id, question], outputs=[answer_md])

        train_btn.click(
            start_training,
            inputs=[role, staff_id],
            outputs=[scenario_q, scenario_options, hidden_correct, hidden_topic],
        )

        submit_ans_btn.click(
            submit_training_answer,
            inputs=[role, staff_id, scenario_q, scenario_options, hidden_correct, scenario_options, hidden_topic],
            outputs=[train_feedback],
        )

    with gr.Tab("Supervisor Dashboard (PIN)"):
        gr.Markdown("### Supervisor access (PIN protected)")
        pin_box = gr.Textbox(label="Enter Supervisor PIN", type="password", placeholder="PIN")
        unlock_btn = gr.Button("Unlock Dashboard", variant="primary")
        unlock_status = gr.Markdown("")

        dash = gr.Column(visible=False)
        with dash:
            df_state = gr.State(None)

            with gr.Row():
                refresh_btn = gr.Button("Refresh Insights", variant="primary")
                export_btn = gr.Button("Export Logs CSV")
            export_file = gr.File(label="Exported file", interactive=False)

            with gr.Row():
                role_filter = gr.Dropdown(["All"] + roles_list(), value="All", label="Filter by role")

            readiness_table = gr.Dataframe(label="Staff Readiness (Demo Score)", interactive=False, wrap=True)
            topq_table = gr.Dataframe(label="Top Questions", interactive=False, wrap=True)
            hotspot_table = gr.Dataframe(label="Confusion Hotspots (SOP improvement areas)", interactive=False, wrap=True)

            gr.Markdown("---")
            gr.Markdown("### SOP Knowledge Base (Editable JSON) — optional")
            sop_json = gr.Textbox(label="SOP JSON", value=get_sop_json_text(), lines=18)
            save_sop_btn = gr.Button("Save SOP Updates (PIN required)", variant="primary")
            sop_save_status = gr.Markdown("")

            gr.Markdown("---")
            clear_btn = gr.Button("Clear Logs (PIN required)")
            clear_status = gr.Markdown("")

        def _refresh(role_filter_val: str):
            df = logs_df()
            readiness = compute_readiness(df)
            topq = top_questions(df, role_filter_val, n=10)
            hot = confusion_hotspots(df, n=10)
            return readiness, topq, hot

        refresh_btn.click(_refresh, inputs=[role_filter], outputs=[readiness_table, topq_table, hotspot_table])
        export_btn.click(export_logs, inputs=[], outputs=[export_file])
        clear_btn.click(supervisor_clear, inputs=[pin_box], outputs=[clear_status])
        save_sop_btn.click(save_sop_json_text, inputs=[pin_box, sop_json], outputs=[sop_save_status])

        unlock_btn.click(supervisor_unlock, inputs=[pin_box], outputs=[pin_box, dash, unlock_status])

demo.launch()