"""Harm-classifier robustness scanner: an interactive console.
An editorial-styled console: a Front Page that leads with the findings, a Lab to
attack a classifier live, a Scan Report with the full evidence, and a Method &
Ethics note. Built for a Trust & Safety reviewer to read in 60 seconds.
pip install -r requirements-app.txt
streamlit run scripts/scanner_app.py
Runs on PUBLIC PROXY DATA ONLY. It demonstrates the *methodology* a content-
safety team would apply to a production detector; it does not, and must not, be
pointed at egregious-harms material. The methods transfer; the data does not.
"""
import datetime
import html
import json
import os
import sys
import pandas as pd
import streamlit as st
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
OUT = os.path.join(ROOT, "outputs")
from src.perturbations import EVASIONS, ZERO_WIDTH
from src.defenses import normalize_pipeline
EXAMPLES = [
"you are an idiot and everyone knows it",
"what a stupid worthless take",
"i hate this loser so much",
"people like that are pathetic trash",
]
st.set_page_config(page_title="The Robustness Review · Harm-Classifier Scanner",
page_icon="🛡️", layout="wide")
# --------------------------------------------------------------------------- #
# Editorial (NYT-style) skin
# --------------------------------------------------------------------------- #
st.markdown("""
""", unsafe_allow_html=True)
# --------------------------------------------------------------------------- #
# Data / model helpers
# --------------------------------------------------------------------------- #
@st.cache_resource(show_spinner="Loading classifier…")
def _load_model(choice: str):
if choice.startswith("Detoxify"):
from src.models import DetoxifyModel
return DetoxifyModel("unbiased")
from src.models import ToyClassifier
return ToyClassifier()
def load_model(choice: str):
"""Return (model, warning|None). Falls back to the offline ToyClassifier if
the real model can't load (e.g. detoxify isn't installed on a lightweight
public deployment), so the demo never hard-crashes on model choice."""
try:
return _load_model(choice), None
except Exception as e:
return _load_model("ToyClassifier"), (
f"Real model unavailable here ({type(e).__name__}); using the offline "
"ToyClassifier. Install `detoxify` to run the real model locally.")
@st.cache_data(show_spinner=False)
def load_cache(path: str):
with open(path, "r", encoding="utf-8") as f:
d = json.load(f)
return d["results"], d["meta"]
def available_caches():
if not os.path.isdir(OUT):
return {}
return {fn.replace(".results.json", ""): os.path.join(OUT, fn)
for fn in sorted(os.listdir(OUT)) if fn.endswith(".results.json")}
def slice_recall(results, target):
for col, rows in results.get("slices", {}).items():
for r in rows:
if str(r.get(col)) == target:
return r.get("recall")
return None
# --------------------------------------------------------------------------- #
# Sidebar
# --------------------------------------------------------------------------- #
st.sidebar.markdown("### The Robustness Review")
st.sidebar.caption("Adversarial evaluation desk")
st.sidebar.markdown("---")
model_choice = st.sidebar.selectbox(
"Classifier under test",
["ToyClassifier (offline, instant)", "Detoxify (real model, slower)"],
help="ToyClassifier is a brittle keyword baseline that runs instantly. "
"Detoxify is a real production-grade RoBERTa model (needs `pip install "
"detoxify`; first load downloads weights).",
)
threshold = st.sidebar.slider("Flagging threshold", 0.1, 0.9, 0.5, 0.05,
help="Score at or above this counts as 'flagged'.")
st.sidebar.markdown("---")
st.sidebar.caption("Public proxy data only. A methodology demonstrator, not an "
"egregious-harms detector.")
# --------------------------------------------------------------------------- #
# Masthead
# --------------------------------------------------------------------------- #
_today = datetime.date.today().strftime("%A, %B %-d, %Y") if os.name != "nt" \
else datetime.date.today().strftime("%A, %B %d, %Y")
st.markdown(f"""
""", unsafe_allow_html=True)
tab_front, tab_lab, tab_scan, tab_about = st.tabs(
["Front Page", "Attack Lab", "Scan Report", "Method & Ethics"])
_caches = available_caches()
# --------------------------------------------------------------------------- #
# Tab:FRONT PAGE
# --------------------------------------------------------------------------- #
with tab_front:
# headline numbers, pulled live from the committed caches where possible
civ = load_cache(_caches["civil_comments"])[0] if "civil_comments" in _caches else None
hate = load_cache(_caches["hatecheck"])[0] if "hatecheck" in _caches else None
f1_civ = civ["baseline"]["f1"] if civ else 0.70
f1_hate = hate["baseline"]["f1"] if hate else 0.76
impl = slice_recall(hate, "derog_impl_h") if hate else 0.53
impl = impl if impl is not None else 0.53
sem = next((r for r in (civ["adversarial"] if civ else []) if r["evasion"] == "llm_paraphrase"), None)
sem_esr = sem["esr"] if sem else 0.57
ece_civ = civ["baseline"].get("ece", 0.022) if civ else 0.022
ece_hate = hate["baseline"].get("ece", 0.226) if hate else 0.226
st.markdown('Lead Investigation · Detoxify (unbiased)
',
unsafe_allow_html=True)
st.markdown('A Healthy F1 Score Hides Three Real Failures
',
unsafe_allow_html=True)
st.markdown('A production toxicity classifier looks fine on '
'the one number everyone reports. Read below the aggregate, by slice, '
'under attack, and at its operating point, and the failures a Trust & '
'Safety team is paid to find appear at once.
', unsafe_allow_html=True)
st.markdown(f"""
{f1_civ:.2f}/{f1_hate:.2f}
Aggregate F1
(Civil Comments / HateCheck)
{impl:.2f}
Recall on implicit hate
~{round((1-impl)*100)}% of it missed
{sem_esr:.2f}
LLM-paraphrase evasion rate
defense cannot recover it
{ece_civ:.3f}→{ece_hate:.2f}
Calibration error
does not transfer across data
""", unsafe_allow_html=True)
st.markdown(f"""
The slice cliff
Aggregate recall near 0.77 averages away the misses. The model catches
99% of blunt threats and just {impl:.2f}
of implicit derogation, the coded hostility that matters most.
Two kinds of evasion
Character tricks (homoglyphs, leetspeak, invisible characters) hit ~100%
evasion but a normalizer recovers them. An LLM paraphrase evades
{sem_esr:.0%} and a normalizer cannot touch it.
Calibration drifts
The same model is well-calibrated on one dataset (ECE {ece_civ:.3f}) and
badly off on another (ECE {ece_hate:.2f}). A threshold tuned in one place
misfires in the next.
""", unsafe_allow_html=True)
st.markdown('
', unsafe_allow_html=True)
st.markdown('Scope & ethics'
'Runs on public toxic-comment proxy data only. It never touches, and '
'must not be pointed at, CSAM, NCII, or violent-extremism material, '
'which belong in a sanctioned, legally-authorized pipeline. The '
'methods transfer to that setting; the data deliberately does not.'
'
', unsafe_allow_html=True)
st.markdown("**Inside this issue:** open **Attack Lab** to break a classifier "
"yourself, **Scan Report** for the full evidence, or **Method & Ethics** "
"for how it is framed and why.")
# --------------------------------------------------------------------------- #
# Tab:ATTACK LAB
# --------------------------------------------------------------------------- #
with tab_lab:
st.markdown('The Lab · Live adversarial test
',
unsafe_allow_html=True)
st.markdown('Break the filter '
'yourself
', unsafe_allow_html=True)
st.markdown('Type a comment the '
'filter should catch, then watch which cheap disguises slip it past, '
'and which a normalization defense recovers.
', unsafe_allow_html=True)
c1, c2 = st.columns([3, 1])
seed_text = c1.text_input("Comment to test", value=EXAMPLES[0])
pick = c2.selectbox("…or pick an example", ["(none)"] + EXAMPLES, index=0)
if pick != "(none)":
seed_text = pick
model, warn = load_model(model_choice)
if warn:
st.warning(warn)
base = float(model.predict_proba([seed_text])[0])
flagged = base >= threshold
if not flagged:
st.markdown(f'{base:.2f}
'
'
Baseline score
'
'
'
'The filter does not flag this at baseline, so there is nothing to '
'evade. Try a clearly toxic example.
',
unsafe_allow_html=True)
else:
rows = []
for name, meta in EVASIONS.items():
atk = meta["fn"](seed_text)
s_atk = float(model.predict_proba([atk])[0])
s_def = float(model.predict_proba([normalize_pipeline(atk)])[0])
shown = atk.replace(ZERO_WIDTH, "·")
rows.append({
"name": name, "effort": meta["effort"],
"stealth": meta["fluency_penalty"],
"shown": shown if len(shown) < 52 else shown[:50] + "…",
"s_atk": s_atk, "evaded": s_atk < threshold,
"s_def": s_def, "recovered": s_def >= threshold,
})
rows.sort(key=lambda r: r["s_atk"])
n_evaded = sum(r["evaded"] for r in rows)
st.markdown(f"""
{base:.2f}
Baseline · flagged
{n_evaded}/{len(rows)}
Evasions that slip through
""", unsafe_allow_html=True)
trows = []
for r in rows:
cls = "row-evaded" if r["evaded"] else ""
res = ('slips through' if r["evaded"]
else 'caught')
rec = ('recovered' if r["recovered"]
else 'no')
trows.append(
f'| {html.escape(r["name"])} | '
f'{r["effort"]} | {r["stealth"]} | '
f'{html.escape(r["shown"])} | '
f'{r["s_atk"]:.2f} | {res} | '
f'{r["s_def"]:.2f} | {rec} |
')
st.markdown(
'| Evasion | Effort | '
'Stealth | Disguised text | Score | Result | '
'After defense | Recovered |
'
+ "".join(trows) + "
", unsafe_allow_html=True)
st.caption("`·` marks an injected invisible (zero-width) character. **Score** "
"is after the attack; **after defense** is after the normalization "
"preprocessor cleans the text first.")
with st.expander("Add a semantic attack (LLM paraphrase)"):
if "ANTHROPIC_API_KEY" not in os.environ:
st.caption("Set `ANTHROPIC_API_KEY` to enable. The LLM rewrites the "
"comment into clean prose with the same intent, the attack "
"a normalization defense cannot reverse. Disabled on the "
"public demo by design.")
elif st.button("Generate paraphrase attacks"):
try:
from src.redteam import generate_variants, judge_preserved
with st.spinner("Asking the model for label-preserving rewrites…"):
prows = []
for v in generate_variants(seed_text, n=3):
if not isinstance(v, str) or not v.strip():
continue
keep = judge_preserved(seed_text, v)
sv = float(model.predict_proba([v])[0])
prows.append((v, keep, sv, sv < threshold and keep))
tr = "".join(
f'| {html.escape(v)} | '
f'{"yes" if keep else "drifted"} | '
f'{sv:.2f} | '
f'{"slips through" if ev else "caught"} |
'
for (v, keep, sv, ev) in prows)
st.markdown('| Paraphrase | '
'Intent kept | Score | Result |
'
'' + tr + "
",
unsafe_allow_html=True)
st.caption("Normalization is not applied here, there is no surface "
"disguise to strip. The fix is training-data augmentation.")
except Exception as e: # pragma: no cover
st.error(f"Red-team call failed: {e}")
# --------------------------------------------------------------------------- #
# Tab:SCAN REPORT
# --------------------------------------------------------------------------- #
with tab_scan:
st.markdown('The Report · Full weakness audit
',
unsafe_allow_html=True)
if not _caches:
st.warning("No scan results found in `outputs/`. Generate one first:\n\n"
"`python scripts/run_eval.py --dataset civil_comments --redteam`")
else:
ds = st.selectbox("Dataset scan", list(_caches.keys()))
results, meta = load_cache(_caches[ds])
b = results["baseline"]
st.markdown(f''
f'{html.escape(str(meta.get("model_name","classifier")))} '
f'on {html.escape(str(meta.get("dataset_name", ds)))}
',
unsafe_allow_html=True)
st.caption(f"n={results['n']} · operating threshold {results['threshold']}")
k = st.columns(5)
k[0].metric("F1", f"{b['f1']:.2f}")
k[1].metric("Precision", f"{b['precision']:.2f}")
k[2].metric("Recall", f"{b['recall']:.2f}")
k[3].metric("False-positive rate", f"{b['fpr']:.2f}")
k[4].metric("Calibration (ECE)", f"{b.get('ece', float('nan')):.3f}")
st.markdown('
', unsafe_allow_html=True)
st.markdown("#### Prioritized weaknesses")
head = results["headline"]
ws, cb = head.get("worst_slice"), head.get("cheapest_break")
if ws:
gap = b["recall"] - ws["recall"]
st.markdown(f"- **Slice cliff**: `{ws['column']}={ws['value']}` recall "
f"**{ws['recall']:.2f}** vs {b['recall']:.2f} overall "
f"(gap {gap:.2f}, on {ws['support']} positives).")
if cb:
st.markdown(f"- **Cheap evasion**: `{cb['evasion']}` (effort {cb['effort']}) "
f"drops recall to **{cb['recall_after']:.2f}** (ESR {cb['esr']:.2f}); "
f"defense recovers it to {cb['recall_after_defense']:.2f}.")
sem = next((r for r in results["adversarial"] if r["evasion"] == "llm_paraphrase"), None)
if sem:
st.markdown(f"- **Semantic evasion**: `llm_paraphrase` ESR **{sem['esr']:.2f}**; "
f"defense does *not* recover it ({sem['recall_after']:.2f} → "
f"{sem['recall_after_defense']:.2f}). Fix is training data, not preprocessing.")
if b.get("ece", 0) > 0.1:
st.markdown(f"- **Miscalibrated**: ECE {b['ece']:.2f}; scores can't be trusted "
f"to mean what they say on this distribution.")
if results["slices"]:
st.markdown('
', unsafe_allow_html=True)
st.markdown("#### Recall by slice")
col = list(results["slices"].keys())[0]
sl = sorted([r for r in results["slices"][col] if r.get("support", 0) >= 1],
key=lambda r: r["recall"])[:12]
bars = "".join(
f'{html.escape(str(r[col]))}'
f''
f'{r["recall"]:.2f}
' for r in sl)
st.markdown(f'{bars}
', unsafe_allow_html=True)
st.caption("Red bars: recall below 0.60, the failing slices the aggregate hides.")
st.markdown('
', unsafe_allow_html=True)
st.markdown("#### Adversarial robustness")
trows = ""
for r in results["adversarial"]:
ev_red = "color:var(--red);" if r["recall_after"] < 0.5 else ""
rec_grn = "color:var(--green);font-weight:700;" if r["recall_after_defense"] >= 0.8 else "color:var(--red);font-weight:700;"
sem_tag = ' semantic' if r["evasion"] == "llm_paraphrase" else ""
trows += (f'| {html.escape(r["evasion"])}{sem_tag} | '
f'{r["effort"]} | {r["fluency_penalty"]} | '
f'{r["esr"]:.2f} | '
f'{r["recall_after"]:.2f} | '
f'{r["recall_after_defense"]:.2f} |
')
st.markdown('| Evasion | Effort | '
'Stealth | ESR | Recall · attacked | '
'Recall · defended |
' + trows
+ "
", unsafe_allow_html=True)
try:
import altair as alt
sc = pd.DataFrame(results["adversarial"])
sc["recovered"] = sc["recall_after_defense"] >= 0.8
chart = (alt.Chart(sc).mark_circle(size=150, opacity=0.85).encode(
x=alt.X("esr:Q", title="evasion success rate (stronger →)",
scale=alt.Scale(domain=[0, 1])),
y=alt.Y("recall_after_defense:Q", title="recall the defense recovers",
scale=alt.Scale(domain=[0, 1])),
color=alt.Color("recovered:N",
scale=alt.Scale(domain=[True, False], range=["#111", "#b91c1c"]),
legend=alt.Legend(title="recovered by defense")),
tooltip=["evasion", "esr", "recall_after", "recall_after_defense"],
).properties(height=320))
rule = alt.Chart(pd.DataFrame({"y": [0.8]})).mark_rule(
strokeDash=[4, 4], color="#999").encode(y="y:Q")
st.altair_chart(chart + rule, width="stretch")
st.caption("Bottom-right (red) = strong attacks the defense can't recover. "
"That is where `llm_paraphrase` lands, the real threat.")
except Exception:
pass
# --------------------------------------------------------------------------- #
# Tab:METHOD & ETHICS
# --------------------------------------------------------------------------- #
with tab_about:
st.markdown('Editor\'s Note · Method & Ethics
',
unsafe_allow_html=True)
st.markdown("""
This is a **methodology demonstrator** for evaluating content-safety classifiers,
built to be read by a Trust & Safety reviewer in under a minute.
**The thesis.** A single aggregate accuracy score hides the failures a T&S team
is paid to find. The tool reads *below* the aggregate in three ways:
- **By slice:** where does recall collapse? (Implicit/coded hate, an
under-protected group, a language the model never learned.)
- **Under attack:** which evasions break it, how cheap are they, and does a
normalization preprocessor recover them? Mechanical character tricks are cheap
but defendable; **LLM paraphrase attacks are not**, and those need training
data, not filters.
- **At the operating point:** are the confidence scores calibrated, and does a
threshold tuned on one distribution transfer to another?
**Why public proxy data.** Egregious-harms detection is an adversarial,
cat-and-mouse problem, which is exactly what this measures. But the worst content
can only be handled inside a sanctioned, legally-authorized pipeline. So the
engine is proven on ordinary public toxic-comment data and the **methods
transfer**, deliberately demonstrating judgment about what *not* to touch.
**Both error directions matter.** For an egregious flag, a false negative lets
harm through, but a false positive can mean a wrongful report against a real
person. The report surfaces precision, false-positive rate, and calibration
alongside recall for that reason.
""")
st.markdown('
', unsafe_allow_html=True)
st.markdown("**Code & paper:** "
"[github.com/Prakharanand000/TandS-harm-classifier-eval]"
"(https://github.com/Prakharanand000/TandS-harm-classifier-eval)")