import os # MPS lacks a few VLM ops in bf16; let those individual ops fall back to CPU # instead of hard-crashing. Must be set BEFORE torch is imported. os.environ.setdefault("PYTORCH_ENABLE_MPS_FALLBACK", "1") import json import time from pathlib import Path import gradio as gr import pypdfium2 as pdfium # PDF → PIL render, server-side (INPUT-02, no system binary needed) import spaces # ZeroGPU decorator; a harmless no-op when running locally import torch from transformers import ( AutoProcessor, Qwen3VLForConditionalGeneration, # Actual class name in transformers 5.10.2 is Qwen2_5_VLForConditionalGeneration # Aliased to Qwen2_5VLForConditionalGeneration to match the bake-off contract. Qwen2_5_VLForConditionalGeneration as Qwen2_5VLForConditionalGeneration, ) # eval/grounded.py is stdlib-only; importing it never triggers model loading. # Used to drive the verifying mascot state (D2-05) — pure Python check, no GPU. from eval.grounded import check_no_invention # noqa: F401 (used in Phase 2 UI plan) # --------------------------------------------------------------------------- # Device / dtype — SPACE_ID sentinel (D-02 ratified 2026-06-05) # The is_available() CUDA check is NOT reliable at module scope on ZeroGPU: # CUDA emulation is active at import time but that check may return False. # SPACE_ID is set on all HF Spaces; absent locally → MPS or CPU. # --------------------------------------------------------------------------- if os.getenv("SPACE_ID"): DEVICE = "cuda" elif torch.backends.mps.is_available(): DEVICE = "mps" else: DEVICE = "cpu" DTYPE = torch.float32 if DEVICE == "cpu" else torch.bfloat16 print(f"[Bureaucat] device={DEVICE} dtype={DTYPE}") # --------------------------------------------------------------------------- # Model-agnostic bake-off loader (D-16) # Switch between candidates by setting BUREAUCAT_MODEL env var. # IMAGE_PATCH_SIZE: 16 for Qwen3-VL (patch_size=16), 14 for Qwen2.5-VL. # --------------------------------------------------------------------------- MODEL_VARIANTS = { "qwen3": { "model_id": "Qwen/Qwen3-VL-8B-Instruct", "model_class": Qwen3VLForConditionalGeneration, "image_patch_size": 16, }, "qwen25": { "model_id": "Qwen/Qwen2.5-VL-7B-Instruct", "model_class": Qwen2_5VLForConditionalGeneration, "image_patch_size": 14, }, } # LOCKED MODEL (MODEL-01, Plan 01-04 bake-off, 2026-06-05): "qwen3" = Qwen3-VL-8B-Instruct. # Verdict: on the calibrated 5-letter gold gate, qwen3 passed 5/5 (zero invented values, # 100% recall both standard+beginner, severity parseable everywhere); the smaller # Qwen2.5-VL-7B passed only 3/5 (dropped case/dossier reference numbers on the standard # pass). "Smallest model that passes wins" → the smaller model does not pass, so qwen3 is # locked. Default read from env so the harness can still drive qwen25; the Space does NOT # set BUREAUCAT_MODEL → production runs qwen3. MODEL_VARIANT = os.getenv("BUREAUCAT_MODEL", "qwen3") # Resolve patch size once at module scope so run_inference can reference it. IMAGE_PATCH_SIZE = MODEL_VARIANTS[MODEL_VARIANT]["image_patch_size"] # Max new tokens: raised 1200 -> 1600 (Phase 3, MODEL-01 amendment 2026-06-06). The # Phase-1 lock at 1200 was established under stochastic sampling and a single output # language. The Phase-3 5x5 multilingual matrix surfaced truncation on the token-heaviest # combination (Hindi + beginner mode: Devanagari is token-dense and beginner mode adds # inline explanations) — csn-aterkrav hit exactly 1200 mid-output and dropped the SEVERITY # line. Under greedy decoding (see run_inference_multi) that worst case lands at ~1068 # tokens, so 1600 gives comfortable (~530-token) headroom across all five languages. # Latency note: more tokens => longer generation; the D-03 <40s ZeroGPU budget (DEFERRED) # must be re-checked on the Space now that the ceiling is higher. MAX_NEW_TOKENS = 1600 # --------------------------------------------------------------------------- # Page-cap constants for multi-image input (INPUT-01) # DEFERRED tuning targets (D-03): the <40s ZeroGPU latency check is not measurable # on local Apple MPS. Tune after measuring real inference time on the dev Space. # --------------------------------------------------------------------------- MAX_PAGES_SOFT = 3 # Warn user above this count (soft advisory threshold) MAX_PAGES_HARD = 5 # Hard cap: truncate to this many pages before inference # Vision-token budget (single knob — keep processor + per-image content in sync). # LOCKED at 1280×28×28 ≈ 1M px: a 1024×28×28 speed experiment (2026-06-12) # FAILED the eval gate — 4/5 gold recall and the non-Swedish fixture was # analyzed instead of refused. Any change MUST re-pass the full gate # (python -m eval.run_eval --model qwen3) before shipping. MIN_PIXELS = 256 * 28 * 28 # ~200K pixel budget floor (OCR-safe) MAX_PIXELS = 1280 * 28 * 28 # ~1M pixel budget ceiling (doc images) def pdf_to_images(pdf_bytes: bytes, max_pages: int = MAX_PAGES_HARD, dpi: int = 200) -> list: """ Render the first ≤max_pages pages of a PDF to PIL Images at the given DPI. Accepts bytes only — never a filesystem path (D3-09 in-memory constraint / T-03-05 path-traversal mitigation). 200 DPI renders A4 to ~1654×2339 px, within the qwen-vl-utils max_pixels budget of 1280×28×28 ≈ 1M px (D3-10). Returns a list[PIL.Image.Image] (possibly empty if the PDF has no pages). Raises PdfiumError on corrupt / malformed input (caller wraps in try/except). """ pdf = pdfium.PdfDocument(pdf_bytes) images = [] for i in range(min(len(pdf), max_pages)): images.append(pdf[i].render(scale=dpi / 72.0).to_pil()) return images def load_model(variant: str): """ Load model + processor for the given variant key ("qwen3" or "qwen25"). Applies: - AutoProcessor with pixel budget controls - attn_implementation="sdpa" (safe built-in, avoids flash-attn dependency) - dtype=DTYPE (keeps working baseline kwarg; NOT torch_dtype) - unconditional .to(DEVICE) — ZeroGPU emulation layer requires .to(), not the accelerate multi-device dispatch path - model.eval() Returns (model, processor). """ v = MODEL_VARIANTS[variant] model_id = v["model_id"] model_class = v["model_class"] print(f"[Bureaucat] loading {model_id} ...") print("[Bureaucat] (first run downloads weights to ~/.cache/huggingface)") proc = AutoProcessor.from_pretrained( model_id, min_pixels=MIN_PIXELS, max_pixels=MAX_PIXELS, ) mdl = model_class.from_pretrained( model_id, dtype=DTYPE, attn_implementation="sdpa", ) mdl = mdl.to(DEVICE) mdl.eval() print("[Bureaucat] model ready.") return mdl, proc # --------------------------------------------------------------------------- # TEST ESCAPE HATCH # When BUREAUCAT_NO_MODEL is set (e.g. by unit tests), skip the heavy load. # The Space and the bake-off do NOT set this var, so the D-02 module-scope # cuda load still happens in production. # --------------------------------------------------------------------------- if os.getenv("BUREAUCAT_NO_MODEL"): model = None processor = None else: model, processor = load_model(MODEL_VARIANT) # --------------------------------------------------------------------------- # Output schema + parser (D-04–D-08) # --------------------------------------------------------------------------- import re from dataclasses import dataclass from typing import Optional @dataclass class StructuredResult: transcription: str # raw verbatim OCR; used by eval harness (D-04) quip: str # "Bureaucat says:" value (D-07) tldr: str why: str actions: str deadlines: str severity: Optional[int] # None if output truncated (D-06) raw: str # full raw model output doctype: str = "letter" # DOCTYPE sentinel: letter | unreadable | not_letter | non_swedish (D3-01) SECTION_ANCHORS = [ ("tldr", r"##\s*TL;?DR"), ("why", r"##\s*Why you got this"), ("actions", r"##\s*What you need to do"), ("deadlines", r"##\s*Deadlines\s*&\s*money"), ] def _split_sections(text: str) -> dict: """Split body text into four sections by fixed Markdown heading anchors (D-05).""" result = {} for i, (key, pattern) in enumerate(SECTION_ANCHORS): m = re.search(pattern, text, re.IGNORECASE) if not m: continue start = m.end() if i + 1 < len(SECTION_ANCHORS): next_m = re.search(SECTION_ANCHORS[i + 1][1], text, re.IGNORECASE) end = next_m.start() if next_m else len(text) else: end = len(text) result[key] = text[start:end].strip() return result DOCTYPE_RE = re.compile( r'^DOCTYPE:\s*(letter|unreadable|not_letter|non_swedish)', re.MULTILINE | re.IGNORECASE, ) def parse_output(raw: str) -> StructuredResult: """ Parse raw model output into a StructuredResult. Language-invariant: anchors on fixed English sentinels regardless of the prose language. Never raises — returns empty fields and severity=None on malformed/truncated output (T-02-02). DOCTYPE sentinel (D3-01): extracted after block, before section split. Regex accepts only the four enumerated tokens; any other/absent value defaults to "letter" (D3-02 lean-toward-analyzing). """ if not raw: return StructuredResult( transcription="", quip="", tldr="", why="", actions="", deadlines="", severity=None, raw=raw, doctype="letter", ) # 1. Extract and strip the block (D-04). transcription = "" trans_match = re.search( r"(.*?)", raw, re.DOTALL | re.IGNORECASE ) if trans_match: transcription = trans_match.group(1).strip() after_trans = re.sub( r".*?", "", raw, flags=re.DOTALL | re.IGNORECASE ).strip() # 2. Parse DOCTYPE: first matching DOCTYPE line (D3-01). Accepts only the four # enumerated tokens; defaults to "letter" if absent or unrecognised (D3-02). doctype_m = DOCTYPE_RE.search(after_trans) doctype = doctype_m.group(1).lower() if doctype_m else "letter" # Strip the DOCTYPE line from body so it does not leak into section text. after_trans = re.sub( r'^DOCTYPE:\s*\S+\s*\n?', "", after_trans, flags=re.MULTILINE | re.IGNORECASE ).strip() # 3. Parse severity: LAST matching SEVERITY: N line (D-06). The schema puts # SEVERITY on the final line; taking the last match means a stray earlier # "SEVERITY:" mention in prose never mis-drives the Panic Meter (WR-02). severity = None sev_matches = re.findall(r"SEVERITY:\s*([1-5])\s*$", after_trans, re.MULTILINE) if sev_matches: severity = int(sev_matches[-1]) body = re.sub(r"\nSEVERITY:\s*[1-5]\s*$", "", after_trans, flags=re.MULTILINE).strip() # 4. Parse the "Bureaucat says:" quip (D-07). quip = "" quip_match = re.search(r"Bureaucat says:\s*(.+?)(?:\n|$)", body) if quip_match: quip = quip_match.group(1).strip() # 5. Split four sections by fixed English headings (D-05, Pitfall 3). sections = _split_sections(body) return StructuredResult( transcription=transcription, quip=quip, tldr=sections.get("tldr", ""), why=sections.get("why", ""), actions=sections.get("actions", ""), deadlines=sections.get("deadlines", ""), severity=severity, raw=raw, doctype=doctype, ) # --------------------------------------------------------------------------- # Prompt (D-04–D-08) # --------------------------------------------------------------------------- SYSTEM_PROMPT = """\ You are Bureaucat, an assistant that helps expats understand Swedish official letters. IMPORTANT — output format rules (do not deviate): 1. First, write a full verbatim OCR transcription of the letter wrapped in XML tags: [exact text of the letter, every word, number, date] 2. Immediately after , classify the document. Write exactly this line (always in English, never translated, machine-parsed — do not translate it): DOCTYPE: [letter|unreadable|not_letter|non_swedish] Valid values: - letter = readable Swedish authority or institutional document (analysable) - unreadable = too blurry, dark, or low-resolution to read reliably - not_letter = readable image but NOT a letter (e.g. photo, receipt, form, ID card) - non_swedish = document's PRIMARY language is not Swedish (a mostly-Swedish letter with embedded English phrases is still: letter) When uncertain, use: letter IMPORTANT: if DOCTYPE is not "letter", you MUST: - Still write the "Bureaucat says:" quip line (in-voice, playful, one-liner) - Still write the DOCTYPE line - SKIP the four ## sections (TL;DR, Why you got this, What you need to do, Deadlines & money) - SKIP the SEVERITY line 3. Then write exactly this line (always in English, always playful): Bureaucat says: [your witty one-liner about this letter] 4. If DOCTYPE is "letter", write the four sections using EXACTLY these English headings: ## TL;DR ## Why you got this ## What you need to do ## Deadlines & money (In "Deadlines & money" you MUST list EVERY date, EVERY amount of money, AND EVERY reference number found anywhere in the letter. Reference numbers include case numbers (ärendenummer), file/dossier numbers (dossiernummer), booking numbers (bokningsnummer), and OCR/payment numbers — list them here EVEN IF the letter has no payment or deadline. Never leave a reference number out of this section. Write one item per line. Start each line with the verbatim value from the letter, then add a dash and your interpretation. Examples: - 15 juni 2026 — last day to file your tax return - 1 234 kr — amount to pay - 9988776 — case number (ärendenummer) Only write "None found." if the letter contains no dates, amounts, OR reference numbers of any kind.) 5. If DOCTYPE is "letter", add a last line, always in English, never translated: SEVERITY: [1-5] Rate how worried the reader should be using the FULL 1-5 range. Do NOT default to 3 — most letters are NOT a 3. Pick the single number that best fits THIS letter's real-world stakes: - 1 = purely informational; nothing to do, no deadline, no money owed (a confirmation, a receipt, an FYI notice) - 2 = minor or routine action, low stakes, soft/distant or no hard deadline (book or attend a routine appointment, a small optional fee) - 3 = a genuine task with a clear deadline OR a modest amount to pay; manageable - 4 = a significant amount owed, OR a firm deadline whose miss has real consequences (a repayment demand, a required document submission) - 5 = urgent and high-stakes: a large sum, an imminent deadline, or a severe consequence such as rejection, debt collection, eviction, or loss of a permit / residence status Hard rules (these OVERRIDE any instinct to pick 3): - If the letter warns of rejection, having to leave Sweden, losing a permit, residence status, or benefit, debt collection, or eviction → SEVERITY 5. - If the letter demands repayment of a specific sum, or sets a firm deadline to submit documents or the case is closed/denied → SEVERITY at least 4. - Use 3 only for a routine task with a clear but low-consequence deadline. - Use 1-2 for informational notices and routine appointments with no real risk. Write all prose in the language requested in the user's message. Keep the four section headings, the DOCTYPE line, and the SEVERITY line in English, never translated. Quote all extracted values (dates, amounts, reference numbers) verbatim as they appear in the letter — never invent, approximate, or omit them. If something is unclear, say so. Do not invent details.""" def build_user_prompt(language: str, beginner_mode: bool) -> str: """ Build the per-call user prompt. beginner_mode adds ONLY inline-explanation guidance within prose — it never adds/removes sections or alters the SEVERITY line or transcription block (D-08). """ # Reference-completeness reminder lives in the BASE prompt (both modes). The Phase-3 # 5x5 matrix surfaced the model dropping a clearly-labelled reference number from # "Deadlines & money" — writing "None found." with e.g. "Ärendenummer: 9988776" sitting # in the transcription. Under greedy decoding this is deterministic (sampling had merely # masked it). The model's *default* is to omit references it doesn't tie to a deadline or # amount; this reminder re-anchors the Finding-3 completeness rule per-call without # touching the SYSTEM_PROMPT. It is mode-independent (standard mode failed too), so it # belongs in the base prompt, not the beginner branch. NOTE: this is a prompt-level # mitigation of a genuine model fragility — a deterministic transcription→Deadlines # cross-check is the more robust follow-up (tracked for the user). prompt = ( f"Please analyse this letter and respond in {language}." "\n\nBefore you write the 'Deadlines & money' section, re-scan the full " "transcription and list EVERY date, EVERY amount, AND EVERY reference number " "(ärendenummer, dossiernummer, bokningsnummer, OCR-nummer) that appears anywhere " "in the letter — each on its own line, verbatim. A reference number must be listed " "even when it has no associated deadline or amount. Only write 'None found.' if " "there is truly no date, amount, or reference number of any kind in the letter." ) if beginner_mode: prompt += ( "\n\nBeginner mode: within each section's prose, add brief " "parenthetical explanations of Swedish institutions or terms " "(e.g. Skatteverket, personnummer, OCR-nummer, etc.). " "Do not add new sections." ) return prompt # --------------------------------------------------------------------------- # Inference — primary multi-image path + backward-compat single-image wrapper # --------------------------------------------------------------------------- def run_inference_multi( images: list, language: str, beginner_mode: bool, mdl, proc, image_patch_size: int, ) -> StructuredResult: """ Encode-generate-parse for one or more images in a single inference call. This is the single shared inference path for both the eval harness and the Gradio UI (INPUT-01 multi-page). run_inference() delegates here so there is exactly one code path — no drift between harness and app. process_vision_info is imported lazily so importing app under BUREAUCAT_NO_MODEL=1 never requires qwen_vl_utils. """ from qwen_vl_utils import process_vision_info # lazy: not needed for parse_output tests image_content = [ { "type": "image", "image": img, "min_pixels": MIN_PIXELS, "max_pixels": MAX_PIXELS, } for img in images ] messages = [ {"role": "system", "content": [{"type": "text", "text": SYSTEM_PROMPT}]}, {"role": "user", "content": image_content + [ {"type": "text", "text": build_user_prompt(language, beginner_mode)}, ]}, ] chat_text = proc.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) image_inputs, video_inputs = process_vision_info( messages, image_patch_size=image_patch_size ) inputs = proc( text=[chat_text], images=image_inputs, videos=video_inputs, do_resize=False, return_tensors="pt", ).to(DEVICE) inputs.pop("token_type_ids", None) # Required for Qwen3-VL; harmless on Qwen2.5-VL # Sanity log: if the image didn't become pixel tensors, the model is # "reading" a blank page — that would look like plausible nonsense, not an error. pv = inputs.get("pixel_values") if pv is None: return StructuredResult( transcription="", quip="", tldr="", why="", actions="", deadlines="", severity=None, raw="ERROR: pixel_values missing from inputs", ) print(f"[Bureaucat] pixel_values={tuple(pv.shape)}") # Greedy decoding (do_sample=False), pinned deliberately (MODEL-01 amendment, # Phase 3). The shipped Qwen3-VL generation_config defaults to do_sample=true / # temperature=0.7 / top_p=0.8 / top_k=20 — i.e. stochastic sampling. For a tool # whose core promise is faithful, never-invented extraction, sampling is the wrong # regime: it makes the same letter yield different outputs run-to-run (the source of # the Phase-3 matrix's intermittent beginner-mode failures) and raises invention risk. # Greedy is deterministic and follows the fixed output-format rules more faithfully. # NOTE: repetition_penalty / no_repeat_ngram_size are NOT applied to the primary # pass — they regressed the gold gate (5/5 → 0/5: no_repeat_ngram_size kills the # legitimately-repeating "- value — label" deadline list, emptying whole sections in # beginner mode) and the adversarial refusal pass. Pure greedy stays the primary path. with torch.no_grad(): out_ids = mdl.generate( **inputs, max_new_tokens=MAX_NEW_TOKENS, do_sample=False, ) trimmed = [o[len(i):] for i, o in zip(inputs["input_ids"], out_ids)] raw_text = proc.batch_decode( trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False )[0] result = parse_output(raw_text) # SEVERITY-recovery fallback (gate-safe by construction). On very dense real letters # (long reference-number lists), pure greedy can fall into a degenerate repetition loop # in "Deadlines & money" and exhaust the token budget BEFORE emitting the trailing # SEVERITY line → severity=None → the Panic Meter shows "unclear" on an otherwise # perfectly-read letter. When that happens (letter parsed WITH content but no severity), # make ONE tiny, bounded text-only follow-up call asking only for the severity, scored # from the transcription the model already produced. It is capped at a handful of tokens # so it physically cannot loop (no long list to emit) — unlike a full re-generation, # which on the densest letters loops again. This never runs on the gold set (those always # emit SEVERITY on pass one), so it cannot affect the eval gate. if ( result.doctype == "letter" and result.severity is None and (result.tldr or result.why or result.actions or result.deadlines) ): print("[Bureaucat] SEVERITY missing (dense-letter loop) — bounded severity-only retry") result.severity = _recover_severity(result, mdl, proc) if result.severity is not None: print(f"[Bureaucat] recovered severity={result.severity}") return result # Compact rubric for the severity-only recovery call — mirrors the SYSTEM_PROMPT scale # and hard rules so a recovered severity matches what the primary pass would have produced. SEVERITY_ONLY_PROMPT = """You rate how worried the reader of a Swedish authority letter should be, on a 1-5 scale. Use the FULL range; do NOT default to 3. - 1 = purely informational; nothing to do, no deadline, no money owed (a refund, a confirmation, an FYI) - 2 = minor or routine action, low stakes (a routine appointment, a small fee) - 3 = a genuine task with a clear deadline OR a modest amount to pay; manageable - 4 = a significant amount owed, OR a firm deadline whose miss has real consequences (a repayment demand, a required document submission) - 5 = urgent and high-stakes: a large sum, an imminent deadline, or a severe consequence (rejection, debt collection, eviction, loss of a permit / residence status) Hard rules: warns of rejection / leaving Sweden / losing a permit or benefit / debt collection / eviction → 5. Demands repayment of a specific sum, or a firm deadline to submit documents or the case is closed → at least 4. Reply with EXACTLY one line and nothing else: SEVERITY: [1-5]""" def _recover_severity(result, mdl, proc) -> Optional[int]: """Bounded, loop-proof text-only severity classification from already-extracted text. Uses the model's own transcription (falling back to the parsed sections) as context. Returns an int 1-5, or None if the model still does not emit a parseable line. """ context = result.transcription.strip() if not context: context = "\n\n".join( s for s in (result.tldr, result.why, result.actions, result.deadlines) if s ).strip() if not context: return None messages = [ {"role": "system", "content": [{"type": "text", "text": SEVERITY_ONLY_PROMPT}]}, {"role": "user", "content": [{"type": "text", "text": ( "Swedish authority letter (already transcribed):\n\n" + context + "\n\nOutput only: SEVERITY: [1-5]" )}]}, ] chat_text = proc.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) sev_inputs = proc(text=[chat_text], return_tensors="pt").to(DEVICE) sev_inputs.pop("token_type_ids", None) with torch.no_grad(): sev_ids = mdl.generate(**sev_inputs, max_new_tokens=12, do_sample=False) sev_trimmed = [o[len(i):] for i, o in zip(sev_inputs["input_ids"], sev_ids)] sev_raw = proc.batch_decode(sev_trimmed, skip_special_tokens=True)[0] m = re.search(r"SEVERITY:\s*([1-5])", sev_raw) if not m: m = re.search(r"\b([1-5])\b", sev_raw) # last-ditch: a bare digit return int(m.group(1)) if m else None def run_inference( image, language: str, beginner_mode: bool, mdl, proc, image_patch_size: int, ) -> StructuredResult: """ Single-image entry point — delegates to run_inference_multi([image], ...). Preserved for backward compatibility so eval/run_eval.py calls (lines 329, 335) continue to work without modification. Guarantees a single shared code path. """ return run_inference_multi([image], language, beginner_mode, mdl, proc, image_patch_size) # Accurate short duration = better queue priority; visitors only have 3.5 min/day. # duration=55: placeholder until measured on dev Space (D-03). @spaces.GPU(duration=55) def decode(files, language, beginner_mode) -> StructuredResult: """ Gradio entry point — accepts a gr.File payload: list[str] of temp file paths. With file_count="multiple" and type="filepath" (Gradio 6.16.0, confirmed), Gradio passes a list of temporary file paths (strings) or None when no file has been uploaded yet. Dispatches per file: - *.pdf → read to bytes immediately (T-03-05 path-traversal mitigation), render via pdf_to_images(); a corrupt/malformed PDF catches PdfiumError and returns doctype="unreadable" so it routes through the slice-1 refusal path (confused mascot, no crash) — T-03-04. - other → Image.open(path).convert("RGB") as before. All rendered pages are concatenated and capped at MAX_PAGES_HARD before being passed to run_inference_multi (unchanged). """ if not files: return StructuredResult( transcription="", quip="", tldr="", why="", actions="", deadlines="", severity=None, raw="Please upload or photograph a letter first.", ) from PIL import Image as _PILImage from io import BytesIO as _BytesIO images: list = [] for path in files: if path is None: continue # Read to bytes immediately — never trust or retain the temp path (T-03-05). # Detect type by CONTENT, not filename: the custom gr.Server frontend uploads # via gradio_client, which lands files as extensionless "blob" temp files, so # endswith(".pdf") alone misses every real PDF (→ "No valid images found." on # the error card). Sniff the %PDF magic bytes instead; extension is a fallback. try: with open(path, "rb") as fh: data = fh.read() except Exception: continue is_pdf = data[:5].startswith(b"%PDF") or path.lower().endswith(".pdf") if is_pdf: try: page_images = pdf_to_images(data) images.extend(page_images) except Exception: # Any PDF parse failure (corrupt, malicious, truncated) → refusal. return StructuredResult( transcription="", quip="", tldr="", why="", actions="", deadlines="", severity=None, doctype="unreadable", raw="Could not read the PDF. Please try a different file.", ) else: try: images.append(_PILImage.open(_BytesIO(data)).convert("RGB")) except Exception: continue # skip unreadable image files; process remaining pages if not images: return StructuredResult( transcription="", quip="", tldr="", why="", actions="", deadlines="", severity=None, raw="No valid images found.", ) # Hard page cap — truncate before inference (DEFERRED tuning target D-03) if len(images) > MAX_PAGES_HARD: images = images[:MAX_PAGES_HARD] return run_inference_multi(images, language, beginner_mode, model, processor, IMAGE_PATCH_SIZE) import html as _html # --------------------------------------------------------------------------- # Example gallery — pre-computed results for zero-GPU demo (UI-01) # Each entry: slug, display label, source image path, cached result JSON path. # Paths are hardcoded constants (T-02-04-01 mitigation: no user-supplied path). # --------------------------------------------------------------------------- # Ordered ascending by severity so the gallery tells a left→right story: # all-clear (refund) → routine → action-needed → money owed → urgent. EXAMPLE_LETTERS = [ { "slug": "skatteverket-slutskattebesked", "label": "Skatteverket — tax refund, good news (severity 1)", "image": "data/letters/public/skatteverket-slutskattebesked.png", "cached": "data/gallery/skatteverket-slutskattebesked-result.json", }, { "slug": "vardcentral-kallelse", "label": "Vårdcentral — appointment reminder (severity 2)", "image": "data/letters/public/vardcentral-kallelse.png", "cached": "data/gallery/vardcentral-kallelse-result.json", }, { "slug": "forsakringskassan-komplettering", "label": "Försäkringskassan — submit documents (severity 3)", "image": "data/letters/public/forsakringskassan-komplettering.png", "cached": "data/gallery/forsakringskassan-komplettering-result.json", }, { "slug": "csn-aterkrav", "label": "CSN — student-aid repayment demand (severity 4)", "image": "data/letters/public/csn-aterkrav.png", "cached": "data/gallery/csn-aterkrav-result.json", }, { "slug": "migrationsverket-uppehallstillstand", "label": "Migrationsverket — residence permit at risk (severity 5)", "image": "data/letters/public/migrationsverket-uppehallstillstand.png", "cached": "data/gallery/migrationsverket-uppehallstillstand-result.json", }, ] def load_example(evt: gr.SelectData) -> tuple: """ Zero-GPU gallery loader — reads cached JSON from disk, returns full render tuple. NOT decorated with @spaces.GPU. Never calls the model. Security: evt.index selects from a hardcoded EXAMPLE_LETTERS list (T-02-04-01: no user-supplied path string — load_example is immune to path traversal). Returns the same 7-element tuple as render_result() so gallery.select() and the decode .then() chain share identical output bindings. """ entry = EXAMPLE_LETTERS[evt.index] data = json.loads(Path(entry["cached"]).read_text(encoding="utf-8")) result = StructuredResult(**data) # Always render in English (gallery examples are pre-computed in English) return render_result(result, "English") # --------------------------------------------------------------------------- # Pure-Python UI renderer functions (no GPU, no model loading) # --------------------------------------------------------------------------- def render_quip(quip: str) -> str: """ Render the Bureaucat quip line as a Markdown string. Returns empty string if quip is blank (hides the component). The "Bureaucat says:" prefix is prepended HERE at render time — parse_output already strips the prefix from the stored .quip field (line 220), so never add it at parse time or you get "Bureaucat says: Bureaucat says: …". """ if not quip or not quip.strip(): return "" escaped_quip = _html.escape(quip.strip()) return f'**Bureaucat says:** *{escaped_quip}*' def render_deadlines_html(deadlines: str) -> str: """ Render the Deadlines & money section as an HTML string (for gr.HTML). Uses gr.HTML (not gr.Markdown) because Markdown sanitization strips and inline style= attributes (RESEARCH Pitfall 2, T-02-02-01 mitigated). Model text in pills is HTML-escaped before wrapping (T-02-02-01). If deadlines is empty or "None found." → returns plain body text, no pill. Otherwise: each "- VALUE — interpretation" line becomes a
  • with the verbatim value highlighted in a warm amber pill. Pill spec (UI-SPEC §Deadlines & Money Card): background: #FFF3CD, color: #8B6914, padding: 4px, border-radius: 4px """ if not deadlines or not deadlines.strip(): return "

    None found.

    " # Check for "None found" (case-insensitive) if re.match(r"\s*none found\.?\s*$", deadlines, re.IGNORECASE): return f"

    {_html.escape(deadlines.strip())}

    " lines = deadlines.strip().splitlines() items = [] has_pill_line = False for raw_line in lines: line = raw_line.strip() if not line: continue # Strip leading bullet markers (-, *, •) stripped = re.sub(r"^[-*•]\s*", "", line).strip() if not stripped: continue # Check for "None found" lines if re.match(r"none found\.?$", stripped, re.IGNORECASE): items.append(f"
  • {_html.escape(stripped)}
    • ") continue # Split on first interpretation separator (same convention as extract_values_from_section) sep_match = re.search(r"\s+[—–\-]\s+", stripped) if sep_match: value_raw = stripped[: sep_match.start()].strip() interpretation_raw = stripped[sep_match.end():].strip() value_esc = _html.escape(value_raw) interp_esc = _html.escape(interpretation_raw) pill = ( f'{value_esc}' ) items.append(f"
  • {pill} — {interp_esc}
    • ") has_pill_line = True else: # No separator — render as plain text pill (value = whole line) value_esc = _html.escape(stripped) pill = ( f'{value_esc}' ) items.append(f"
  • {pill}
    • ") has_pill_line = True if not items: return "

    None found.

    " if not has_pill_line: # Only "None found" lines — render as plain paragraph(s) return "

    " + " ".join( _html.escape(re.sub(r"^[-*•]\s*", "", l.strip()).strip()) for l in lines if l.strip() ) + "

    " return ( '" ) def render_footer() -> str: """ Always-visible footer HTML containing both required trust statements. TRUST-01: privacy note (nothing stored, no external APIs) TRUST-05: legal disclaimer (not legal advice) """ return ( '
    ' "

    Everything runs on a small model inside this Space. " "Nothing is sent to external APIs. " "Nothing is stored after your session ends.

    " "

    Bureaucat explains letters — it does not give legal advice. " "For legal matters, consult a qualified professional.

    " "
    " ) # --------------------------------------------------------------------------- # Mascot assets map — Phase 2 wires 6 states; Phase 3 adds confused/wrong_document # --------------------------------------------------------------------------- MASCOT_ASSETS = { "idle": "assets/mascot/idle.png", "reading": "assets/mascot/reading.png", "verifying": "assets/mascot/verifying.png", "allclear": "assets/mascot/allclear.png", "deadline": "assets/mascot/deadline.png", "money": "assets/mascot/money.png", "confused": "assets/mascot/confused.png", # Phase 3: unreadable input "wrong_document": "assets/mascot/wrong_document.png", # Phase 3: not_letter / non_swedish } # Result-state detection regexes (D2-06 — do not change without plan approval) # Money: numeric amount followed by kr / SEK / currency symbol _MONEY_RE = re.compile(r'\d[\d\s]*(?:kr|SEK|€|\$|£)', re.IGNORECASE) # Date: "dd Month yyyy" (Swedish month names) OR ISO "yyyy-mm-dd" _DATE_RE = re.compile( r'\b\d{1,2}\s+(?:jan|feb|mar|apr|maj|jun|jul|aug|sep|okt|nov|dec)\w*\s+\d{4}\b' r'|\b\d{4}-\d{2}-\d{2}\b', re.IGNORECASE, ) # --------------------------------------------------------------------------- # Renderers — Plan 03 fills bodies # (wired in Plan 02 so the .then() chain is complete; implementations in Plan 03) # --------------------------------------------------------------------------- _SEVERITY_COLORS = { 1: "#27AE60", 2: "#8BC34A", 3: "#F39C12", 4: "#E67E22", 5: "#C0392B", } _SEVERITY_LABELS = { 1: "1 — Informational", 2: "2 — Low concern", 3: "3 — Action needed", 4: "4 — Urgent", 5: "5 — Act immediately", } def render_panic_meter(severity) -> str: """Return HTML for the Panic Meter — a color-coded *verdict badge* (FUN-01). severity int 1-5: solid color-tinted badge showing a big "{N}/5" + the word label. This deliberately is NOT a horizontal fill bar — a fill bar reads as a progress/loading bar (UAT feedback, 2026-06-06). severity None: gray "Result unavailable" badge. Never color-only (accessibility, UI-SPEC line 92): the badge always shows the integer and the word, and the aria-label carries "{N} — {word}". """ if severity is None: return ( '
    ' '
    Result unavailable — try re-uploading
    ' "
    " ) color = _SEVERITY_COLORS.get(severity, "#9E9E9E") label = _SEVERITY_LABELS.get(severity, f"{severity}") # e.g. "3 — Action needed" word = label.split("—", 1)[1].strip() if "—" in label else label return ( f'
    ' f'
    {severity}/5
    ' f'
    {word}
    ' f"
    " ) def render_panic_placeholder() -> str: """Neutral initial state for the panic area (before any analysis) so it never reads as an empty/stalled progress bar (UAT feedback, 2026-06-06).""" return ( '
    ' '
    🐾 Feed me a letter and I\'ll tell you how worried to be
    ' "
    " ) def render_mascot(state: str) -> str: """Return HTML for the mascot gr.HTML component (img tag + CSS class). CSS class mascot-{state} drives keyframe animation (CSS-only, no JS). Unknown state falls back to idle asset. """ src = MASCOT_ASSETS.get(state, MASCOT_ASSETS["idle"]) # Gradio 6 serves allowed_paths files under /gradio_api/file= (the bare # `file=` route from older Gradio 404s on 6.x). Leading slash → resolves # from server root regardless of page path. allowed_paths=["assets"] at launch. return ( f'' ) def select_result_state(result) -> str: """Return the mascot result state based on StructuredResult content (D2-06). Severity 1 is genuinely good / no-stakes news (e.g. a tax REFUND) → the happy "allclear" cat, which also drives the celebratory screen effects. This takes priority over content: a refund has a money amount but is NOT alarming, so it must not get the "money" (shocked) cat. Otherwise content-priority: money > deadline > allclear. "none found" in deadlines text short-circuits to allclear immediately. """ if getattr(result, "severity", None) == 1: return "allclear" deadlines_text = result.deadlines or "" if "none found" in deadlines_text.lower(): return "allclear" if _MONEY_RE.search(deadlines_text): return "money" if _DATE_RE.search(deadlines_text): return "deadline" return "allclear" # --------------------------------------------------------------------------- # Legacy helper kept for backward compat (replaced by structured render path) # --------------------------------------------------------------------------- def _render_result(result: StructuredResult) -> str: """Extract displayable text from StructuredResult for the Markdown pane.""" return result.raw or "No output generated." # --------------------------------------------------------------------------- # CSS — Global styles (brand palette, card styles, RTL text-align rule) # Applied via demo.launch(css=GLOBAL_CSS) so it covers the full page. # DO NOT pass css= to gr.Blocks() — that emits a UserWarning in Gradio 6.x. # The [dir="rtl"] rule supplies the text-align half of the RTL contract # (UI-SPEC §Typography line 62) when gr.Markdown's rtl prop sets dir="rtl". # --------------------------------------------------------------------------- GLOBAL_CSS = """ @import url('https://fonts.googleapis.com/css2?family=Fredoka:wght@400;500;600;700&family=Baloo+2:wght@600;700;800&display=swap'); .gradio-container { max-width: 1040px !important; margin: 0 auto !important; } /* Page background is set per-mode via the theme (.set body_background_fill*) — NOT hardcoded here, which was the dark-mode break (light body + light theme text). */ /* Hero header — chunky, cartoon, a little bouncy */ .bcat-hero { display: flex; align-items: center; gap: 16px; padding: 14px 4px 6px; } .bcat-hero__emoji { font-size: 54px; line-height: 1; animation: bcat-bounce 2.2s ease-in-out infinite; transform-origin: bottom center; } .bcat-hero__title { font-family: "Baloo 2", "Fredoka", sans-serif; font-size: 42px; font-weight: 800; letter-spacing: -0.01em; margin: 0; line-height: 1; background: linear-gradient(90deg, #E91E63, #84CC16); -webkit-background-clip: text; background-clip: text; -webkit-text-fill-color: transparent; } .bcat-hero__tag { font-size: 16px; color: var(--body-text-color-subdued); margin: 4px 0 0; } @keyframes bcat-bounce { 0%,100%{transform:translateY(0) scale(1)} 30%{transform:translateY(-10px) scale(1.06)} 50%{transform:translateY(0) scale(0.97)} } /* Big chunky tab labels */ .tabs button, .tab-nav button { font-family: "Fredoka", sans-serif !important; font-size: 17px !important; font-weight: 600 !important; } /* Sassy chunky call-to-action button */ .cta-btn { font-family: "Fredoka", sans-serif !important; font-size: 18px !important; font-weight: 700 !important; border-radius: 14px !important; padding: 12px 18px !important; box-shadow: 0 6px 0 0 rgba(0,0,0,0.12) !important; transition: transform 0.08s ease, box-shadow 0.08s ease !important; } .cta-btn:hover { transform: translateY(-2px) !important; box-shadow: 0 8px 0 0 rgba(0,0,0,0.14) !important; } .cta-btn:active { transform: translateY(3px) !important; box-shadow: 0 2px 0 0 rgba(0,0,0,0.14) !important; } /* Sassy quip — speech-bubble feel */ .quip-line { font-size: 17px !important; font-style: italic; margin-top: 8px !important; } .quip-line p { background: var(--background-fill-secondary); border-radius: 14px; padding: 12px 16px; margin: 0; position: relative; } /* Surfaces — theme-aware so both light and dark render correctly. gr.Group(elem_classes="section-card") IS the card; inner blocks are flattened so the heading + prose read as one surface (not nested boxes). */ .section-card, .input-card { background: var(--block-background-fill); border: 1px solid var(--border-color-primary); color: var(--body-text-color); border-radius: 14px !important; padding: 14px 18px; margin-bottom: 12px; box-shadow: var(--block-shadow, 0 1px 3px rgba(0,0,0,0.06)); overflow: hidden; } .section-card > *, .section-card .block, .section-card .form, .section-card .prose, .section-card .md { background: transparent !important; border: none !important; box-shadow: none !important; } .section-card h3 { font-size: 12px; font-weight: 700; color: var(--body-text-color); opacity: 0.55; margin: 0 0 6px 0; text-transform: uppercase; letter-spacing: 0.07em; } /* Panic verdict badge — status chip, deliberately NOT a fill/progress bar */ .panic-badge { border-radius: 16px; padding: 16px 20px; color: #fff; display: flex; align-items: baseline; gap: 14px; box-shadow: 0 6px 18px rgba(0,0,0,0.16); margin-bottom: 10px; } .panic-badge__num { font-size: 40px; font-weight: 800; line-height: 1; } .panic-badge__denom { font-size: 18px; font-weight: 600; opacity: 0.8; margin-left: 2px; } .panic-badge__word { font-size: 17px; font-weight: 800; text-transform: uppercase; letter-spacing: 0.06em; align-self: center; } .panic-badge--none, .panic-badge--placeholder { background: var(--background-fill-secondary); color: var(--body-text-color-subdued); border: 1.5px dashed var(--border-color-primary); box-shadow: none; justify-content: center; } .panic-badge--none .panic-badge__word, .panic-badge--placeholder .panic-badge__word { text-transform: none; font-weight: 600; letter-spacing: 0; font-size: 15px; } /* Bouncy pop-in when a real verdict lands (game-feel) */ @keyframes pop-in { 0%{transform:scale(0.85);opacity:0} 60%{transform:scale(1.04)} 100%{transform:scale(1);opacity:1} } .panic-badge.severity-1,.panic-badge.severity-2,.panic-badge.severity-3,.panic-badge.severity-4,.panic-badge.severity-5 { animation: pop-in 0.45s cubic-bezier(.22,1.2,.36,1); } /* Mascot — compact beside the verdict; bigger & bouncier on the fun tab */ .mascot-panel { text-align: center; padding: 0; } .mascot-panel img { transition: opacity 0.2s ease; } .mascot-big img { max-width: 180px !important; } .mascot-big .mascot-idle { animation: bcat-bounce 2s ease-in-out infinite; transform-origin: bottom center; } /* Footer — theme-aware trust note */ .bcat-footer { padding: 16px; border-top: 1px solid var(--border-color-primary); font-size: 13px; color: var(--body-text-color-subdued); } .bcat-footer p { margin: 4px 0; } [dir="rtl"] { text-align: right; } """ # --------------------------------------------------------------------------- # CSS animation keyframes — injected as a gr.HTML style block (first element # inside gr.Blocks context). gr.HTML passes through """ # --------------------------------------------------------------------------- # Event handler functions (Task 2b) # These run outside the GPU context; only decode() is GPU-decorated. # --------------------------------------------------------------------------- def set_reading_state() -> str: """Fires immediately on button click — no GPU. Sets mascot to reading.""" return render_mascot("reading") def run_verifying_state(result) -> str: """ Fires after decode() completes — pure Python, no GPU. On refusals (doctype != "letter"): returns the appropriate error mascot immediately, bypassing check_no_invention (which would operate on empty values and could raise — Pitfall 2 / T-03-03). On successful reads: runs the grounded no-invention check (D2-05), then dwells for 0.6 s so the verifying mascot state is visible in the browser. Without the dwell, Gradio may coalesce this update with the preceding/ following step and the verifying state is invisible (RESEARCH Pattern 3 — NON-GPU .then() step, zero GPU quota cost). """ # Refusal guard (T-03-03): skip check_no_invention on non-letter doctypes. # render_result's own refusal branch then renders the final panes. if getattr(result, "doctype", "letter") != "letter": mascot_state = "confused" if getattr(result, "doctype", "") == "unreadable" else "wrong_document" return render_mascot(mascot_state) check_no_invention(result) # pure-Python grounded check (eval/grounded.py) time.sleep(0.6) # deliberate dwell — verifying state must be visible return render_mascot("verifying") REFUSAL_GUIDANCE = { "unreadable": ( "The photo is too blurry or dark to read. " "Retake it in better light and make sure the whole letter is in frame and in focus." ), "not_letter": ( "That's a fine image, but it doesn't look like a Swedish authority letter. " "Upload a letter from Skatteverket, Försäkringskassan, Migrationsverket, CSN, or similar." ), "non_swedish": ( "Bureaucat reads Swedish authority letters. " "This one looks like it's in another language — upload a letter written in Swedish." ), } def render_refusal(result, language: str) -> tuple: """ Render a bad-input refusal as the same 7-element tuple as render_result. Returns: [0] panic_html = "" (no Panic Meter for refusals — D3-03) [1] mascot_html = confused for unreadable; wrong_document for not_letter/non_swedish [2] quip_md = model's in-voice refusal quip (already escaped in render_quip) [3] tldr_out = app-side fixed guidance text keyed by doctype (NOT model-driven) [4] why_out = "" (no section) [5] actions_out = "" (no section) [6] deadlines_html = "" (no deadlines) Guidance copy is app-side (REFUSAL_GUIDANCE constant) — model only provides the quip. This keeps the function unit-testable without a model (BUREAUCAT_NO_MODEL=1). """ is_rtl = language == "Arabic" def _wrap_rtl_html(html_str: str) -> str: if is_rtl: return f'
    {html_str}
    ' return html_str def _md_update(text: str) -> dict: return gr.update(value=text, rtl=is_rtl) doctype = getattr(result, "doctype", "not_letter") mascot_state = "confused" if doctype == "unreadable" else "wrong_document" guidance = REFUSAL_GUIDANCE.get(doctype, REFUSAL_GUIDANCE["not_letter"]) return ( "", # 0 panic_html — no Panic Meter (D3-03) render_mascot(mascot_state), # 1 mascot_html render_quip(result.quip), # 2 quip_md — model's in-voice refusal quip _md_update(guidance), # 3 tldr_out — app-side guidance (not model-driven) _md_update(""), # 4 why_out _md_update(""), # 5 actions_out _wrap_rtl_html(""), # 6 deadlines_html ) def render_result(result, language: str) -> tuple: """ Fires after run_verifying_state — renders all output panes. Returns a 7-element tuple matching the outputs order: [panic_html(0), mascot_html(1), quip_md(2), tldr_out(3), why_out(4), actions_out(5), deadlines_html(6)] RTL delivery-layer contract (UI-SPEC §Typography, T-02-02-01): - Prose panes (3, 4, 5) are gr.Markdown. Their rtl prop is toggled via gr.update(value=..., rtl=True/False) — explicit bool both ways so a component set RTL on a prior Arabic run is reset for the next English run. gr.HTML(
    ) would be STRIPPED by sanitize_html=True on gr.Markdown — that is the silent bug this approach avoids. - Deadlines & Panic gr.HTML (0, 6) are pass-through. For Arabic, wrap in
    . For other languages, no wrapper. - Extracted verbatim values remain Swedish-locale regardless of direction. Error handling: severity=None / malformed result → 7-element tuple with error copy in tldr pane; no crash (RESEARCH Pattern 3, UI-SPEC Copywriting Contract). """ ERROR_COPY = ( "Bureaucat had trouble reading that letter. Try uploading a clearer photo." ) # Determine if RTL is needed (Arabic only) is_rtl = language == "Arabic" def _wrap_rtl_html(html_str: str) -> str: """Wrap gr.HTML content in dir=rtl div for Arabic; leave others bare.""" if is_rtl: return f'
    {html_str}
    ' return html_str def _md_update(text: str) -> dict: """Return gr.update dict for a gr.Markdown prose pane with explicit rtl.""" return gr.update(value=text, rtl=is_rtl) # 1. Intentional refusal — doctype check FIRST (D3-03). # Refusals legitimately have severity=None, so this branch MUST precede # the severity=None malformed check below (routing order is load-bearing). if result is not None and getattr(result, "doctype", "letter") != "letter": return render_refusal(result, language) # 2. Malformed / error path — only when there is NOTHING to show. A letter that # produced an analysis but whose trailing SEVERITY line got lost (e.g. a greedy # repetition loop on a dense letter) still has real content — salvage it below # rather than discarding it with a false "couldn't read" error. _has_content = result is not None and ( getattr(result, "tldr", "") or getattr(result, "why", "") or getattr(result, "actions", "") or getattr(result, "deadlines", "") ) if result is None or not hasattr(result, "severity") or (result.severity is None and not _has_content): return ( render_panic_meter(None), # 0 panic_html render_mascot("idle"), # 1 mascot_html "", # 2 quip_md _md_update(ERROR_COPY), # 3 tldr_out _md_update(""), # 4 why_out _md_update(""), # 5 actions_out _wrap_rtl_html(""), # 6 deadlines_html ) # Render the result state (money / deadline / allclear) for the mascot state = select_result_state(result) return ( _wrap_rtl_html(render_panic_meter(result.severity)), # 0 panic_html render_mascot(state), # 1 mascot_html render_quip(result.quip), # 2 quip_md (always English) _md_update(result.tldr or ""), # 3 tldr_out _md_update(result.why or ""), # 4 why_out _md_update(result.actions or ""), # 5 actions_out _wrap_rtl_html(render_deadlines_html(result.deadlines)), # 6 deadlines_html ) # --------------------------------------------------------------------------- # Gradio UI — Full gr.Blocks layout (Task 2a: static structure; Task 2b: event wiring) # --------------------------------------------------------------------------- # Game-feel confetti — runs in the browser via demo.load(js=...) because a #