Improve image trace classification

Co-authored-by: Codex <codex@openai.com>

tests/test_tracing.py

@@ -127,6 +127,61 @@ class TraceTests(unittest.TestCase):
         ))
         self.assertEqual(next(iter(text_record)), "trace_id")
 
+    def test_image_trace_uses_model_assessment_for_fixed_description(self) -> None:
+        assessment = {
+            "risk_label": "Suspicious",
+            "simple_explanation": (
+                "The screenshot claims a Pakistan Post parcel delivery failed "
+                "and demands an immediate address update through a short link."
+            ),
+            "red_flags": [
+                "Unknown sender with no official courier branding.",
+                "Urgent delivery action through a suspicious URL.",
+            ],
+            "safe_next_steps": ["Contact the courier through its official app."],
+            "reply_draft": "",
+        }
+        record = trace_runtime.build_trace_record(
+            text="",
+            image_data_url="data:image/png;base64,PRIVATE_IMAGE_BYTES",
+            example_id="",
+            assessment=assessment,
+        )
+
+        self.assertEqual(record["input_category"], "courier")
+        self.assertTrue(record["urgency"])
+        self.assertEqual(
+            record["input"],
+            "image: Courier-style content with link, urgency, courier signals",
+        )
+        self.assertIn("Known courier pattern", record["result_summary"])
+        serialized = json.dumps(record)
+        self.assertNotIn(assessment["simple_explanation"], serialized)
+        self.assertNotIn(assessment["red_flags"][0], serialized)
+        self.assertNotIn("PRIVATE_IMAGE_BYTES", serialized)
+
+    def test_urdu_image_assessment_maps_to_traffic_challan(self) -> None:
+        record = trace_runtime.build_trace_record(
+            text="",
+            image_data_url="data:image/jpeg;base64,PRIVATE_IMAGE_BYTES",
+            example_id="",
+            assessment={
+                "risk_label": "Likely scam",
+                "simple_explanation": (
+                    "یہ ٹریفک چالان کی فوری ادائیگی کے لیے مشکوک لنک استعمال کرتا ہے۔"
+                ),
+                "red_flags": ["آج جرمانہ ادا کرنے کا دباؤ"],
+                "safe_next_steps": ["سرکاری ذریعے سے تصدیق کریں۔"],
+                "reply_draft": "",
+            },
+        )
+
+        self.assertEqual(record["input_category"], "traffic_challan")
+        self.assertTrue(record["urgency"])
+        self.assertIn("Traffic-challan-style", record["input"])
+        self.assertIn("link", record["scam_tactics"])
+        self.assertIn("payment", record["scam_tactics"])
+
     def test_opt_out_does_not_queue_trace(self) -> None:
         with patch("app.queue_trace") as queue_mock:
             result = app.analyze_notice("", "", save_trace=False)

traces/dataset_card.md

@@ -28,8 +28,11 @@ Helper scam-check requests. It does not contain hidden model reasoning or
 autonomous-agent trajectories.
 
 The application uses a Modal-hosted Qwen model for normal assessments. Creating
-a trace never calls an AI model. Traces only observe the existing request path
-and convert it into allow-listed categories, booleans, buckets, and counts.
+a trace never makes an additional AI model call. Traces only observe the
+existing request path and convert it into allow-listed categories, booleans,
+and fixed descriptions. For image submissions, the existing assessment's
+explanation and red flags may be inspected transiently for this mapping, but
+their text is not stored.
 
 ## Fields
 
@@ -70,9 +73,9 @@ credentials, addresses, tracking IDs, long numbers, and title-case names or
 entities. Regex redaction cannot guarantee removal of every possible
 identifier, so users should opt out when submitting sensitive content.
 
-Images store only fixed descriptions; screenshots and OCR text are not stored.
-Users see a checked trace disclosure in the app and may opt out before each
-request.
+Images store only fixed descriptions. Screenshots, OCR text, model explanations,
+and model red flags are not stored. Users see a checked trace disclosure in the
+app and may opt out before each request.
 
 ## Provenance
 

traces/runtime.py

@@ -33,15 +33,36 @@ RISK_LABELS = {
     "none",
 }
 SIGNAL_PATTERNS = {
-    "otp": r"\b(?:otp|one[- ]time (?:pin|password)|verification code)\b",
-    "cnic": r"\bcnic\b",
-    "credentials": r"\b(?:pin|password|cvv|card details?|bank details?)\b",
-    "link": r"(?:https?://|www\.|bit\.ly|tinyurl\.|cutt\.ly|\.xyz\b|\.top\b)",
-    "urgency": r"\b(?:urgent|immediately|today|now|within \d+|last warning)\b",
-    "payment": r"\b(?:pay|payment|fee|fine|transfer|send money|rs\.?|pkr)\b",
-    "refund_or_prize": r"\b(?:refund|prize|winner|lottery|cashback|reward)\b",
-    "courier": r"\b(?:parcel|courier|delivery|pakistan post|leopards|tcs|customs)\b",
-    "challan": r"\b(?:challan|traffic fine|traffic violation|e-challan)\b",
+    "otp": r"\b(?:otp|one[- ]time (?:pin|password)|verification code)\b|او ٹی پی",
+    "cnic": r"\bcnic\b|شناختی کارڈ",
+    "credentials": (
+        r"\b(?:pin|password|cvv|card details?|bank details?)\b|"
+        r"پاس ورڈ|کارڈ کی تفصیلات"
+    ),
+    "link": (
+        r"(?:https?://|www\.|bit\.ly|tinyurl\.|cutt\.ly|\.xyz\b|\.top\b)|"
+        r"\b(?:link|url|domain)\b|لنک"
+    ),
+    "urgency": (
+        r"\b(?:urgent|immediately|today|now|within \d+|last warning)\b|"
+        r"فوری|فوراً|آج|آخری وارننگ"
+    ),
+    "payment": (
+        r"\b(?:pay|payment|fee|fine|transfer|send money|rs\.?|pkr)\b|"
+        r"ادائیگی|جرمانہ|رقم"
+    ),
+    "refund_or_prize": (
+        r"\b(?:refund|prize|winner|lottery|cashback|reward)\b|"
+        r"انعام|ریفنڈ|قرعہ اندازی"
+    ),
+    "courier": (
+        r"\b(?:parcel|courier|delivery|pakistan post|leopards|tcs|customs)\b|"
+        r"پارسل|کوریئر|ڈیلیوری|پاکستان پوسٹ"
+    ),
+    "challan": (
+        r"\b(?:challan|traffic fine|traffic violation|e-challan)\b|"
+        r"چالان|ٹریفک جرمانہ"
+    ),
     "account_threat": (
         r"\b(?:account|sim|service|electricity)\b.{0,50}"
         r"\b(?:block|blocked|suspend|closed|disconnect)\b"
@@ -111,24 +132,50 @@ def detect_category(text: str, signals: dict[str, bool], example_id: str = "") -
     profile = EXAMPLE_PROFILES.get(example_id)
     if profile:
         return profile[1]
+    if signals["challan"]:
+        return "traffic_challan"
     lowered = (text or "").lower()
     categories = (
-        ("fbr", ("fbr", "taxpayer", "tax refund")),
-        ("bank", ("bank", "hbl", "ubl", "meezan", "alfalah")),
-        ("wallet", ("easypaisa", "jazzcash", "wallet")),
-        ("utility", ("electricity", "gas bill", "utility", "lesco", "k-electric")),
-        ("traffic_challan", ("challan", "traffic fine", "traffic violation")),
-        ("courier", ("parcel", "courier", "delivery", "pakistan post", "leopards", "tcs")),
-        ("customs", ("customs", "duty")),
-        ("university", ("university", "admission", "scholarship", "hec")),
-        ("job", ("job", "salary", "recruiter", "employment")),
-        ("marketplace", ("buyer", "seller", "marketplace", "whatsapp")),
+        ("fbr", ("fbr", "taxpayer", "tax refund", "ایف بی آر", "ٹیکس")),
+        ("bank", ("bank", "hbl", "ubl", "meezan", "alfalah", "بینک")),
+        ("wallet", ("easypaisa", "jazzcash", "wallet", "ایزی پیسہ", "جاز کیش")),
+        (
+            "utility",
+            ("electricity", "gas bill", "utility", "lesco", "k-electric", "بجلی", "گیس بل"),
+        ),
+        (
+            "traffic_challan",
+            ("challan", "traffic fine", "traffic violation", "چالان", "ٹریفک جرمانہ"),
+        ),
+        (
+            "courier",
+            (
+                "parcel",
+                "courier",
+                "delivery",
+                "pakistan post",
+                "leopards",
+                "tcs",
+                "پارسل",
+                "کوریئر",
+                "ڈیلیوری",
+                "پاکستان پوسٹ",
+            ),
+        ),
+        ("customs", ("customs", "duty", "کسٹمز")),
+        (
+            "university",
+            ("university", "admission", "scholarship", "hec", "یونیورسٹی", "داخلہ"),
+        ),
+        ("job", ("job", "salary", "recruiter", "employment", "نوکری", "تنخواہ")),
+        (
+            "marketplace",
+            ("buyer", "seller", "marketplace", "whatsapp", "خریدار", "فروخت", "واٹس ایپ"),
+        ),
     )
     for category, terms in categories:
         if any(term in lowered for term in terms):
             return category
-    if signals["challan"]:
-        return "traffic_challan"
     if signals["courier"]:
         return "courier"
     return "unknown"
@@ -210,7 +257,23 @@ def result_summary(
     return f"{risk_label}: {pattern}. {novelty} {RESULT_GUIDANCE[risk_label]}"
 
 
-def build_input_profile(text: str, image_data_url: str, example_id: str = "") -> dict[str, Any]:
+def assessment_evidence(assessment: dict[str, Any] | None) -> str:
+    """Return transient model evidence used only for allow-list classification."""
+    if not isinstance(assessment, dict):
+        return ""
+    values: list[str] = [str(assessment.get("simple_explanation", ""))]
+    red_flags = assessment.get("red_flags", [])
+    if isinstance(red_flags, list):
+        values.extend(str(item) for item in red_flags)
+    return " ".join(values)[:4000]
+
+
+def build_input_profile(
+    text: str,
+    image_data_url: str,
+    example_id: str = "",
+    assessment: dict[str, Any] | None = None,
+) -> dict[str, Any]:
     profile = EXAMPLE_PROFILES.get(example_id)
     if profile:
         input_type = profile[0]
@@ -218,8 +281,13 @@ def build_input_profile(text: str, image_data_url: str, example_id: str = "") ->
         input_type = "image"
     else:
         input_type = "text"
-    signals = detect_signals(text, example_id)
-    category = detect_category(text, signals, example_id)
+    classification_text = text
+    if input_type == "image" and not example_id:
+        classification_text = " ".join(
+            part for part in (text, assessment_evidence(assessment)) if part
+        )
+    signals = detect_signals(classification_text, example_id)
+    category = detect_category(classification_text, signals, example_id)
     tactics = [name for name, enabled in signals.items() if enabled]
     return {
         "input": (
@@ -244,8 +312,18 @@ def build_trace_record(
     risk_label = str((assessment or {}).get("risk_label", "none"))
     if risk_label not in RISK_LABELS:
         risk_label = "none"
-    input_profile = build_input_profile(text, image_data_url, example_id)
-    signals = detect_signals(text, example_id)
+    input_profile = build_input_profile(
+        text,
+        image_data_url,
+        example_id,
+        assessment,
+    )
+    classification_text = text
+    if image_data_url and not example_id:
+        classification_text = " ".join(
+            part for part in (text, assessment_evidence(assessment)) if part
+        )
+    signals = detect_signals(classification_text, example_id)
     category = input_profile["input_category"]
     assessment = assessment or {}
     return {