import os import json from dataclasses import asdict from functools import lru_cache from html import escape from pathlib import Path from time import perf_counter, sleep from typing import Any import gradio as gr from fastapi.responses import HTMLResponse from jawbreaker.analyzers import ( build_llama_cpp_analyzer, build_transformers_analyzer, heuristic_analyzer, prediction_to_analysis, repair_prediction, should_apply_heuristic_guard, validate_prediction, ) from jawbreaker.render import render_analysis_html, render_memory_html, render_scanning_html from jawbreaker.schema import ScamAnalysis from jawbreaker.trust import ( build_trusted_note, confidence_metadata, is_low_context_message, low_context_analysis, ) try: import spaces except ImportError: spaces = None EXAMPLES = [ "USPS: Your package is held due to an unpaid fee. Verify now: http://usps-track-secure.example", "Hi Grandma, I lost my phone. This is my new number. Can you send $800 for rent today? Please don't tell Mom.", "Coinbase alert: We received a request to update the phone number on your account. If this wasn't you, call support immediately at [callback number].", "Hi! I'm a recruiter from TikTok Shop. We are looking for a part-time assistant. Flexible remote work, 60 minutes per day, $330-$750 per day. Contact me on WhatsApp at [phone number].", "Your dentist appointment is confirmed for Tuesday at 2:00 PM. Reply C to confirm or R to reschedule.", "Your verification code is 482913. Do not share this code with anyone. We will never ask for it by phone.", ] DEFAULT_TRANSFORMERS_MODEL_ID = "openbmb/MiniCPM5-1B" DEFAULT_ADAPTER_ID = "build-small-hackathon/jawbreaker-minicpm5-1b-lora-v8" DEFAULT_TRANSFORMERS_MAX_TOKENS = 512 LOGO_PATH = Path("jawbreaker_logo.png") FORCE_LIGHT_JS = """() => { const forceLight = () => document.body.classList.remove('dark'); forceLight(); new MutationObserver(forceLight).observe(document.body, { attributes: true, attributeFilter: ['class'], }); }""" FORCE_LIGHT_HEAD = """ """ def app_theme() -> gr.Theme: theme = gr.themes.Soft( primary_hue="red", secondary_hue="slate", neutral_hue="zinc", radius_size="sm", ) theme.set( block_label_background_fill="transparent", block_label_border_color="transparent", block_label_text_color="#6b5144", block_label_text_size="13px", block_label_text_weight="700", ) return theme def app_css() -> str: return Path("style.css").read_text(encoding="utf-8") def gpu_callback(fn): if spaces is None: return fn return spaces.GPU(fn) def _env_int(name: str, default: int | None = None) -> int | None: value = os.getenv(name) if value is None or value == "": return default return int(value) def _env_bool(name: str, default: bool | None = None) -> bool | None: value = os.getenv(name) if value is None or value == "": return default return value.strip().lower() in {"1", "true", "yes", "on"} def default_adapter_id(model_id: str) -> str | None: configured = os.getenv("JAWBREAKER_ADAPTER_ID") if configured is not None: return configured or None if model_id == DEFAULT_TRANSFORMERS_MODEL_ID: return DEFAULT_ADAPTER_ID return None @lru_cache(maxsize=1) def get_analyzer(): backend = current_backend() if backend == "heuristic": return heuristic_analyzer if backend == "llama-cpp": model_path = resolve_model_path() return build_llama_cpp_analyzer( model_path, chat_format=os.getenv("JAWBREAKER_CHAT_FORMAT") or None, n_ctx=_env_int("JAWBREAKER_N_CTX", 1024) or 1024, n_threads=_env_int("JAWBREAKER_N_THREADS", 2), n_gpu_layers=_env_int("JAWBREAKER_N_GPU_LAYERS", 0) or 0, n_batch=_env_int("JAWBREAKER_N_BATCH", 128) or 128, n_ubatch=_env_int("JAWBREAKER_N_UBATCH", 128) or 128, offload_kqv=_env_bool("JAWBREAKER_OFFLOAD_KQV", True), op_offload=_env_bool("JAWBREAKER_OP_OFFLOAD"), max_tokens=_env_int("JAWBREAKER_MAX_TOKENS", 256) or 256, temperature=float(os.getenv("JAWBREAKER_TEMPERATURE", "0")), ) if backend in {"transformers", "zerogpu"}: model_id = os.getenv("JAWBREAKER_TRANSFORMERS_MODEL_ID", DEFAULT_TRANSFORMERS_MODEL_ID) return build_transformers_analyzer( model_id, adapter_id=default_adapter_id(model_id), max_new_tokens=_env_int("JAWBREAKER_MAX_TOKENS", DEFAULT_TRANSFORMERS_MAX_TOKENS) or DEFAULT_TRANSFORMERS_MAX_TOKENS, temperature=float(os.getenv("JAWBREAKER_TEMPERATURE", "0")), device_map=os.getenv("JAWBREAKER_DEVICE_MAP", "auto"), dtype=os.getenv("JAWBREAKER_TORCH_DTYPE", "auto"), trust_remote_code=_env_bool("JAWBREAKER_TRUST_REMOTE_CODE", True) or False, attn_implementation=os.getenv("JAWBREAKER_ATTENTION_IMPLEMENTATION", "eager") or None, ) raise ValueError(f"Unsupported JAWBREAKER_BACKEND: {backend}") def current_backend() -> str: return os.getenv("JAWBREAKER_BACKEND", "llama-cpp").strip().lower() def should_warm_model() -> bool: return _env_bool("JAWBREAKER_WARM_MODEL", True) and current_backend() in {"transformers", "zerogpu"} def resolve_model_path() -> Path: model_path = Path(os.getenv("JAWBREAKER_MODEL_PATH", "models/qwen3-0.6b-gguf/Qwen3-0.6B-Q4_K_M.gguf")) if model_path.exists(): return model_path repo_id = os.getenv("JAWBREAKER_MODEL_REPO", "unsloth/Qwen3-0.6B-GGUF") filename = os.getenv("JAWBREAKER_MODEL_FILE", "Qwen3-0.6B-Q4_K_M.gguf") cache_dir = os.getenv("JAWBREAKER_MODEL_CACHE") try: from huggingface_hub import hf_hub_download except ImportError as exc: raise RuntimeError("huggingface_hub is required to download the configured model file.") from exc return Path(hf_hub_download(repo_id=repo_id, filename=filename, cache_dir=cache_dir)) def run_analysis(message: str, memory: list[dict] | None) -> ScamAnalysis: memory = memory or [] heuristic = ScamAnalysis.from_heuristics(message, memory) started = perf_counter() try: prediction = get_analyzer()(message) except Exception as exc: print( "jawbreaker analyzer_error fallback=heuristic " f"elapsed={perf_counter() - started:.2f}s error={exc!r}", flush=True, ) heuristic.summary = f"{heuristic.summary} Jawbreaker used its safety fallback because the model response was not usable." return heuristic raw_validation_errors = validate_prediction(prediction) prediction = repair_prediction(prediction) validation_errors = validate_prediction(prediction) model_analysis = prediction_to_analysis(prediction, similar_memory=heuristic.similar_memory) print(f"jawbreaker analyze elapsed={perf_counter() - started:.2f}s", flush=True) if should_use_heuristic_guard(model_analysis, heuristic, validation_errors, message): print( "jawbreaker guard=heuristic " f"model_risk={model_analysis.risk_level} heuristic_risk={heuristic.risk_level}", flush=True, ) return heuristic if raw_validation_errors: print(f"jawbreaker schema_repaired errors={raw_validation_errors}", flush=True) return model_analysis def warm_model_impl() -> None: backend = current_backend() if not should_warm_model(): print(f"jawbreaker warm_model skip backend={backend}", flush=True) return None started = perf_counter() print(f"jawbreaker warm_model start backend={backend}", flush=True) get_analyzer() print(f"jawbreaker warm_model ready elapsed={perf_counter() - started:.2f}s", flush=True) return None @gpu_callback def warm_model() -> None: return warm_model_impl() def memory_entry_from_analysis(message: str, analysis: ScamAnalysis) -> dict[str, Any]: return { "summary": analysis.summary, "scam_type": analysis.scam_type, "risk_level": analysis.risk_level, "fingerprint": analysis.scam_dna, "text": message[:240], } def render_current_memory(memory: list[dict]) -> str: return render_memory_html(ScamAnalysis.from_heuristics("", memory), memory) def render_save_status(message: str = "") -> str: if not message: return "
" return f"
{message}
" def model_status_label() -> str: model_id = os.getenv("JAWBREAKER_TRANSFORMERS_MODEL_ID", DEFAULT_TRANSFORMERS_MODEL_ID) adapter_id = default_adapter_id(model_id) if model_id == "openbmb/MiniCPM5-1B" and adapter_id == "build-small-hackathon/jawbreaker-minicpm5-1b-lora-v8": return "MiniCPM5-1B + Jawbreaker LoRA v8" if adapter_id: return adapter_id.rsplit("/", 1)[-1].replace("jawbreaker-", "").upper() return model_id.rsplit("/", 1)[-1].upper() def backend_status_label() -> str: backend = current_backend() if backend == "zerogpu": return "ZEROGPU" if backend == "transformers": return "TRANSFORMERS" if backend == "llama-cpp": return "LLAMA.CPP" if backend == "heuristic": return "LOCAL FALLBACK" return backend.upper() def analysis_payload(message: str, memory: list[dict] | None = None) -> dict[str, Any]: memory = memory or [] started = perf_counter() if not message.strip(): analysis = ScamAnalysis.from_heuristics(message, memory) elif is_low_context_message(message): analysis = low_context_analysis(message, memory) else: analysis = run_analysis(message, memory) entry = memory_entry_from_analysis(message, analysis) if message.strip() and (not memory or memory[-1].get("text") != entry.get("text")): memory.append(entry) return { "analysis": asdict(analysis), "confidence": confidence_metadata(message, analysis), "copy_plan": build_handoff_message(message, analysis), "memory": memory[-6:], "message": message, "model_label": model_status_label(), "elapsed_seconds": round(perf_counter() - started, 2), } def logo_data_uri() -> str: import base64 if not LOGO_PATH.exists(): return "" encoded = base64.b64encode(LOGO_PATH.read_bytes()).decode("ascii") return f"data:image/png;base64,{encoded}" def paper_shield_html() -> str: examples_json = json.dumps(EXAMPLES).replace(" Jawbreaker

Jawbreaker

Private scam defense for someone you love.

{model_label} READY

Is this message safe?

Paste a text, email, or DM. Jawbreaker will show the risk, the warning signs, and the safest next step before you reply, click, or pay.

Safety aid only, not legal, financial, or cybersecurity advice. Always verify through official channels. First check may take up to 30s while the secure system wakes up.
""" def kitchen_table_html() -> str: examples_json = json.dumps(EXAMPLES).replace(" Jawbreaker

Jawbreaker

When something feels off, check it here.

__MODEL_LABEL__ Ready

Pull up a chair. Let's read it together.

Paste a text, email, or DM that feels off. Jawbreaker turns it into a simple safety note: the risk, the warning signs, and the safest next step before you click, reply, or pay.

The kitchen-table rule If a message creates panic, asks for money, or sends you to a link, bring it here before you act.
Safety aid only. Always verify through official channels.
First check may take up to 30s while the secure system wakes up.
""" return ( template.replace("__LOGO_URI__", logo_uri) .replace("__MODEL_LABEL__", model_label) .replace("__BACKEND_LABEL__", backend_label) .replace("__EXAMPLES_JSON__", examples_json) ) def build_server() -> gr.Server: app = gr.Server() @app.api(name="analyze") @gpu_callback def analyze_api(message: str, memory: list[dict] | None = None) -> dict[str, Any]: return analysis_payload(message, memory) @app.api(name="warm") @gpu_callback def warm_api() -> dict[str, str]: warm_model_impl() return {"status": "ready", "backend": current_backend(), "model": model_status_label()} @app.get("/", response_class=HTMLResponse) async def homepage() -> HTMLResponse: return HTMLResponse(kitchen_table_html()) @app.get("/health") async def health() -> dict[str, str]: return {"status": "ok", "backend": current_backend(), "model": model_status_label()} return app def build_handoff_message(message: str, analysis: ScamAnalysis) -> str: return build_trusted_note(message, analysis) def should_use_heuristic_guard( model_analysis: ScamAnalysis, heuristic: ScamAnalysis, validation_errors: list[str], message: str = "", ) -> bool: return should_apply_heuristic_guard(message, model_analysis, heuristic, validation_errors) @gpu_callback def analyze_message( message: str, memory: list[dict] | None, ): memory = memory or [] if not message.strip(): analysis = ScamAnalysis.from_heuristics(message, memory) yield ( render_analysis_html(message, analysis), render_memory_html(analysis, memory), memory, {}, gr.update(interactive=True), gr.update(interactive=True), ) return if is_low_context_message(message): analysis = low_context_analysis(message, memory) entry = memory_entry_from_analysis(message, analysis) if not memory or memory[-1].get("text") != entry.get("text"): memory.append(entry) yield ( render_analysis_html(message, analysis), render_memory_html(analysis, memory), memory, {"message": message, "entry": entry}, gr.update(interactive=True), gr.update(interactive=True), ) return for active_step, progress in [(0, 12), (1, 30), (2, 54), (3, 76)]: yield ( render_scanning_html(active_step=active_step, progress=progress), render_current_memory(memory), memory, {}, gr.update(interactive=False), gr.update(interactive=False), ) sleep(0.25) try: analysis = run_analysis(message, memory) except Exception as exc: analysis = analysis_error(exc) yield ( render_analysis_html(message, analysis), render_memory_html(analysis, memory), memory, {}, gr.update(interactive=True), gr.update(interactive=True), ) return last_scan = { "message": message, "entry": memory_entry_from_analysis(message, analysis), } entry = last_scan["entry"] if not memory or memory[-1].get("text") != entry.get("text"): memory.append(entry) yield ( render_analysis_html(message, analysis), render_memory_html(analysis, memory), memory, last_scan, gr.update(interactive=True), gr.update(interactive=True), ) def analysis_error(exc: Exception) -> ScamAnalysis: return ScamAnalysis( risk_level="needs_check", scam_type="analysis_error", summary="Jawbreaker could not finish the model scan in this environment.", tactics=["runtime error"], safest_action="Do not click links or reply yet. Verify through an official app, website, or known phone number.", trusted_person_message=f"Can you check this for me? Jawbreaker hit an analysis error: {exc}", scam_dna={ "Impersonates": "Unknown", "Pressure": "Unknown", "Ask": "Unknown", "Risk": "Could not analyze", }, ) def remember_current(message: str, memory: list[dict] | None, last_scan: dict | None) -> tuple[str, str, list[dict]]: memory = memory or [] if not message.strip(): return render_save_status("Paste a message first."), render_current_memory(memory), memory last_scan = last_scan or {} entry = last_scan.get("entry") if last_scan.get("message") != message or not isinstance(entry, dict): return ( render_save_status("Analyze this message first, then save the pattern."), render_current_memory(memory), memory, ) if memory and memory[-1].get("text") == entry.get("text"): return ( render_save_status("This pattern is already saved for this session."), render_current_memory(memory), memory, ) memory.append(entry) return render_save_status("Saved this scam pattern for this session."), render_current_memory(memory), memory def build_app() -> gr.Blocks: gr.set_static_paths(paths=[LOGO_PATH.resolve()]) with gr.Blocks(title="Jawbreaker") as demo: memory_state = gr.State([]) last_scan_state = gr.State({}) gr.HTML( """
JAWBREAKER
[ SECURE_ENV: ACTIVE ] [ STATUS: READY ] [ MODEL: MINICPM5_1B_LORA_V8 ] [ BUILT_FOR: BUILD_SMALL_HACKATHON ]
""" ) with gr.Row(elem_classes=["main-grid"]): with gr.Column(scale=5, elem_classes=["scan-panel"]): gr.HTML( """
Message to check

MESSAGE TO CHECK

""" ) message = gr.Textbox( label=None, show_label=False, placeholder="Paste a text, email, or DM here.", lines=10, max_lines=16, ) analyze = gr.Button("Check message", variant="primary", elem_classes=["check-btn"]) gr.Examples(examples=EXAMPLES, inputs=message, label="Try a sample") memory = gr.HTML(render_current_memory([])) with gr.Column(scale=7): result = gr.HTML( """
Ready to check

SYSTEM STANDING BY

Jawbreaker is ready to shield your loved ones from digital fraud.

Paste any text message, email, or DM on the left. The local model will evaluate risk factors, unpack the scam strategy, and deliver a plain-English protection plan.

How to use it

1. Copy a text message from your phone or an email that feels off.

2. Paste it into the input area on the left of this screen.

3. Click RUN SCAM DETECTOR to analyze it with private local AI.

""" ) analyze.click( fn=analyze_message, inputs=[message, memory_state], outputs=[result, memory, memory_state, last_scan_state, message, analyze], show_progress="hidden", ) demo.load( fn=warm_model, inputs=None, outputs=None, show_progress="hidden", api_visibility="private", ) return demo if __name__ == "__main__": build_server().launch(favicon_path=str(LOGO_PATH))