Improve privacy-safe trace quality

Strengthen schema validation, add live dataset auditing, improve image category and tactic mapping, and distinguish failed image assessments from genuinely unclassified content.

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

README.md

@@ -159,6 +159,12 @@ They do **not** include raw notice text, OCR text, screenshots, names, phone
 numbers, credentials, or full model responses. See
 the trace [`dataset card`](traces/dataset_card.md).
 
+Audit the live trace dataset with:
+
+```bash
+python -m traces.scripts.analyze_trace_dataset
+```
+
 ## Privacy and Limitations
 
 - Inputs are processed in memory and are not written to disk by the app.

app.py

@@ -37,6 +37,33 @@ REQUIRED_FIELDS = {
     "safe_next_steps",
     "reply_draft",
 }
+TRACE_CATEGORIES = (
+    "fbr",
+    "bank",
+    "wallet",
+    "utility",
+    "traffic_challan",
+    "courier",
+    "customs",
+    "university",
+    "job",
+    "marketplace",
+    "unknown",
+)
+TRACE_TACTICS = (
+    "otp",
+    "cnic",
+    "credentials",
+    "link",
+    "urgency",
+    "payment",
+    "refund_or_prize",
+    "courier",
+    "challan",
+    "account_threat",
+    "off_platform_contact",
+    "impersonation",
+)
 EXAMPLE_CACHE_PATH = ROOT / "data" / "example_assessments.json"
 
 SYSTEM_PROMPT = """Assess Pakistani notices and messages for scam risk.
@@ -67,6 +94,11 @@ Evidence rules:
   relevant notice; do not call it irrelevant merely because it looks harmless.
 
 Output rules:
+- trace_category: choose exactly one privacy-safe category from the schema based
+  on the visible content. Use unknown only when no listed category applies.
+- trace_tactics: choose every visible tactic from the schema. Use an empty
+  array when none applies. These values are metadata and must not contain raw
+  text, names, numbers, URLs, or explanations.
 - explanation: 1-3 short sentences naming the decisive visible evidence.
 - red_flags: 1-4 concise evidence-based items. For a normal relevant notice,
   use one item such as "No clear scam indicators in the supplied message."
@@ -109,8 +141,14 @@ OUTPUT_SCHEMA: dict[str, Any] = {
         "red_flags": {"type": "array", "items": {"type": "string"}},
         "safe_next_steps": {"type": "array", "items": {"type": "string"}},
         "reply_draft": {"type": "string"},
+        "trace_category": {"type": "string", "enum": list(TRACE_CATEGORIES)},
+        "trace_tactics": {
+            "type": "array",
+            "items": {"type": "string", "enum": list(TRACE_TACTICS)},
+            "uniqueItems": True,
+        },
     },
-    "required": sorted(REQUIRED_FIELDS),
+    "required": sorted(REQUIRED_FIELDS | {"trace_category", "trace_tactics"}),
     "additionalProperties": False,
 }
 
@@ -173,6 +211,15 @@ def normalize_assessment(value: Any) -> dict[str, Any]:
             else ""
         ),
     }
+    trace_category = value.get("trace_category")
+    trace_tactics = value.get("trace_tactics")
+    if trace_category in TRACE_CATEGORIES:
+        result["trace_category"] = trace_category
+    if isinstance(trace_tactics, list):
+        normalized_tactics = list(dict.fromkeys(
+            str(item) for item in trace_tactics if str(item) in TRACE_TACTICS
+        ))
+        result["trace_tactics"] = normalized_tactics
     for field in ("simple_explanation",):
         if not result[field]:
             raise ValueError(f"{field} must not be empty.")

tests/test_tracing.py

@@ -15,6 +15,7 @@ from openai import APIStatusError, APITimeoutError
 
 import app
 from traces import runtime as trace_runtime
+from traces.scripts.validate_traces import validate_file
 
 
 class TraceTests(unittest.TestCase):
@@ -62,6 +63,40 @@ class TraceTests(unittest.TestCase):
         self.assertLess(elapsed_ms, 10)
         self.assertFalse(trace_runtime.validate_trace(records[0]))
 
+    def test_trace_validation_rejects_extra_or_sensitive_columns(self) -> None:
+        record = self.sample_record()
+        record["notes"] = "PRIVATE RAW MESSAGE"
+        self.assertIn(
+            "Unexpected fields: notes",
+            trace_runtime.validate_trace(record),
+        )
+
+        record = self.sample_record()
+        record["input"] = "text: Visit https://private.example immediately"
+        self.assertIn(
+            "Input contains an unredacted URL.",
+            trace_runtime.validate_trace(record),
+        )
+
+        record = self.sample_record()
+        record["result_summary"] = "PRIVATE MODEL OUTPUT"
+        self.assertIn(
+            "Result summary does not match the deterministic fields.",
+            trace_runtime.validate_trace(record),
+        )
+
+    def test_file_validation_rejects_duplicate_trace_ids(self) -> None:
+        record = self.sample_record()
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "duplicates.jsonl"
+            path.write_text(
+                json.dumps(record) + "\n" + json.dumps(record) + "\n",
+                encoding="utf-8",
+            )
+            count, errors = validate_file(path)
+        self.assertEqual(count, 2)
+        self.assertTrue(any("Duplicate trace ID" in error for error in errors))
+
     def test_trace_uses_simplified_columns(self) -> None:
         image_record = self.sample_record()
         self.assertTrue(image_record["input"].startswith("image: "))
@@ -160,6 +195,51 @@ class TraceTests(unittest.TestCase):
         self.assertNotIn(assessment["red_flags"][0], serialized)
         self.assertNotIn("PRIVATE_IMAGE_BYTES", serialized)
 
+    def test_image_trace_rejects_unsupported_structured_model_metadata(self) -> None:
+        record = trace_runtime.build_trace_record(
+            text="",
+            image_data_url="data:image/png;base64,PRIVATE_IMAGE_BYTES",
+            example_id="",
+            assessment={
+                "risk_label": "Suspicious",
+                "simple_explanation": (
+                    "The buyer asks to move the marketplace conversation "
+                    "to WhatsApp."
+                ),
+                "red_flags": ["The sender identity is unverified."],
+                "safe_next_steps": ["Verify independently."],
+                "reply_draft": "",
+                "trace_category": "university",
+                "trace_tactics": ["refund_or_prize", "urgency"],
+            },
+        )
+
+        self.assertEqual(record["input_category"], "marketplace")
+        self.assertEqual(
+            record["scam_tactics"],
+            "off_platform_contact, impersonation",
+        )
+        self.assertEqual(
+            record["input"],
+            (
+                "image: Marketplace-style content with off-platform-contact, "
+                "impersonation signals"
+            ),
+        )
+        self.assertFalse(trace_runtime.validate_trace(record))
+
+    def test_failed_image_trace_is_assessment_unavailable(self) -> None:
+        record = trace_runtime.build_trace_record(
+            text="",
+            image_data_url="data:image/png;base64,PRIVATE_IMAGE_BYTES",
+            example_id="",
+            assessment=None,
+        )
+
+        self.assertEqual(record["input"], "image: Assessment unavailable")
+        self.assertEqual(record["risk_label"], "none")
+        self.assertFalse(trace_runtime.validate_trace(record))
+
     def test_urdu_image_assessment_maps_to_traffic_challan(self) -> None:
         record = trace_runtime.build_trace_record(
             text="",

traces/dataset_card.md

@@ -32,7 +32,10 @@ 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.
+their text is not stored. New model responses also include enum-only category
+and tactic hints. These hints are treated as untrusted: the trace mapper keeps
+only values supported by the explanation or red flags and falls back to its
+deterministic English/Urdu evidence rules when they disagree.
 
 ## Fields
 
@@ -77,15 +80,38 @@ 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.
 
+When image analysis fails before an assessment is available, new records use
+`image: Assessment unavailable`. This keeps service failures separate from
+successful assessments whose content is genuinely unclassified.
+
 ## Provenance
 
 Seed traces represent the six public examples bundled with Pakistan Notice
 Helper. Runtime traces may represent successful, rejected, or failed requests.
 Trace generation itself does not invoke the model.
 
+The seed rows are illustrative examples, not an evaluation split. All six
+currently have the `Likely scam` label, so they must not be used to estimate
+class balance, accuracy, recall, or real-world scam prevalence.
+
+Runtime rows are an operational log and intentionally preserve repeated
+requests. Consequently, repeated examples, unclassified image descriptions,
+and incomplete `none` assessments may be common. For training or evaluation,
+create a separate curated split that:
+
+- excludes `risk_label: none`
+- reviews or excludes unclassified image rows
+- deduplicates on the privacy-safe `input` and result columns
+- uses a task-appropriate class-balancing strategy
+
+The source repository includes `traces/scripts/analyze_trace_dataset.py` for
+schema validation and a reproducible summary of these quality indicators.
+
 ## Limitations
 
 - Regex signals and category detection are approximate.
+- Runtime frequencies may reflect testing or repeated usage, not population
+  prevalence.
 - Regex redaction may miss unusual personal or confidential information.
 - Novelty is not researched against external threat-intelligence sources.
 - The dataset cannot reproduce original messages or screenshots.

traces/runtime.py

@@ -67,6 +67,16 @@ SIGNAL_PATTERNS = {
         r"\b(?:account|sim|service|electricity)\b.{0,50}"
         r"\b(?:block|blocked|suspend|closed|disconnect)\b"
     ),
+    "off_platform_contact": (
+        r"\b(?:move|continue|contact|chat|message)\b.{0,50}"
+        r"\b(?:whatsapp|telegram|outside|another (?:app|platform)|phone number)\b|"
+        r"\bwhatsapp\b|واٹس ایپ"
+    ),
+    "impersonation": (
+        r"\b(?:claims? to be|pretends? to be|impersonat(?:e|es|ing|ion)|"
+        r"fake (?:sender|branding|authority)|unverified (?:sender|identity)|"
+        r"(?:sender )?identity is unverified|official branding)\b|جعلی|نقالی"
+    ),
 }
 EXAMPLE_PROFILES = {
     "text-courier": ("text", "courier", {"link", "urgency", "payment", "courier"}),
@@ -114,6 +124,141 @@ SENSITIVE_VALUE_PATTERN = re.compile(
     re.I,
 )
 TITLE_CASE_PATTERN = re.compile(r"\b[A-Z][a-z]{2,}\b")
+TRACE_FIELDS = (
+    "trace_id",
+    "timestamp",
+    "input",
+    "input_category",
+    "urgency",
+    "scam_tactics",
+    "result_summary",
+    "risk_label",
+    "reply_draft_policy",
+)
+REPLY_DRAFT_POLICIES = {"allowed", "suppressed", "not_applicable"}
+RESIDUAL_IDENTIFIER_PATTERNS = {
+    "URL": re.compile(r"https?://|www\.", re.I),
+    "email": re.compile(r"\b[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}\b"),
+    "CNIC": re.compile(r"\b\d{5}-\d{7}-\d\b"),
+    "phone number": re.compile(r"(?<!\d)(?:\+?92[- ]?|0)?3\d{2}[- ]?\d{7}(?!\d)"),
+    "account number": re.compile(r"\bPK\d{2}[A-Z0-9]{10,30}\b", re.I),
+    "card number": re.compile(r"(?<!\d)(?:\d[ -]?){12,19}(?!\d)"),
+}
+CATEGORY_TERMS = (
+    ("fbr", ("fbr", "taxpayer", "tax refund", "revenue board", "ایف بی آر", "ٹیکس")),
+    (
+        "bank",
+        (
+            "bank",
+            "banking",
+            "hbl",
+            "ubl",
+            "meezan",
+            "alfalah",
+            "debit card",
+            "credit card",
+            "بینک",
+        ),
+    ),
+    (
+        "wallet",
+        (
+            "easypaisa",
+            "easy paisa",
+            "jazzcash",
+            "mobile wallet",
+            "ایزی پیسہ",
+            "جاز کیش",
+        ),
+    ),
+    (
+        "utility",
+        (
+            "electricity",
+            "gas bill",
+            "utility bill",
+            "lesco",
+            "k-electric",
+            "meter",
+            "بجلی",
+            "گیس بل",
+        ),
+    ),
+    (
+        "traffic_challan",
+        (
+            "challan",
+            "traffic fine",
+            "traffic violation",
+            "e-challan",
+            "vehicle fine",
+            "چالان",
+            "ٹریفک جرمانہ",
+        ),
+    ),
+    (
+        "courier",
+        (
+            "parcel",
+            "package",
+            "courier",
+            "delivery",
+            "shipment",
+            "consignment",
+            "pakistan post",
+            "leopards",
+            "tcs",
+            "پارسل",
+            "کوریئر",
+            "ڈیلیوری",
+            "پاکستان پوسٹ",
+        ),
+    ),
+    ("customs", ("customs", "custom duty", "import duty", "کسٹمز")),
+    (
+        "university",
+        (
+            "university",
+            "admission",
+            "scholarship",
+            "student portal",
+            "tuition",
+            "hec",
+            "یونیورسٹی",
+            "داخلہ",
+            "اسکالرشپ",
+        ),
+    ),
+    (
+        "job",
+        (
+            "job",
+            "salary",
+            "recruiter",
+            "recruitment",
+            "employment",
+            "interview",
+            "work from home",
+            "نوکری",
+            "تنخواہ",
+        ),
+    ),
+    (
+        "marketplace",
+        (
+            "buyer",
+            "seller",
+            "marketplace",
+            "listing",
+            "product",
+            "whatsapp",
+            "move the conversation",
+            "خریدار",
+            "فروخت",
+            "واٹس ایپ",
+        ),
+    ),
+)
 
 
 def detect_signals(text: str, example_id: str = "") -> dict[str, bool]:
@@ -135,45 +280,7 @@ def detect_category(text: str, signals: dict[str, bool], example_id: str = "") -
     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", "خریدار", "فروخت", "واٹس ایپ"),
-        ),
-    )
-    for category, terms in categories:
+    for category, terms in CATEGORY_TERMS:
         if any(term in lowered for term in terms):
             return category
     if signals["courier"]:
@@ -206,6 +313,8 @@ def safe_description(category: str, signals: dict[str, bool]) -> str:
         "courier": "courier",
         "challan": "challan",
         "account_threat": "account-threat",
+        "off_platform_contact": "off-platform-contact",
+        "impersonation": "impersonation",
     }
     active = [signal_labels[name] for name, enabled in signals.items() if enabled][:4]
     suffix = f" with {', '.join(active)} signals" if active else " with no mapped signals"
@@ -268,6 +377,23 @@ def assessment_evidence(assessment: dict[str, Any] | None) -> str:
     return " ".join(values)[:4000]
 
 
+def structured_assessment_profile(
+    assessment: dict[str, Any] | None,
+) -> tuple[str, dict[str, bool]] | None:
+    if not isinstance(assessment, dict):
+        return None
+    category = assessment.get("trace_category")
+    tactics = assessment.get("trace_tactics")
+    if category not in CATEGORY_DISPLAY_NAMES or not isinstance(tactics, list):
+        return None
+    if any(tactic not in SIGNAL_PATTERNS for tactic in tactics):
+        return None
+    return category, {
+        name: name in tactics
+        for name in SIGNAL_PATTERNS
+    }
+
+
 def build_input_profile(
     text: str,
     image_data_url: str,
@@ -281,6 +407,11 @@ def build_input_profile(
         input_type = "image"
     else:
         input_type = "text"
+    structured_profile = (
+        structured_assessment_profile(assessment)
+        if input_type == "image" and not example_id
+        else None
+    )
     classification_text = text
     if input_type == "image" and not example_id:
         classification_text = " ".join(
@@ -288,12 +419,29 @@ def build_input_profile(
         )
     signals = detect_signals(classification_text, example_id)
     category = detect_category(classification_text, signals, example_id)
+    if structured_profile:
+        structured_category, structured_signals = structured_profile
+        confirmed_tactics = {
+            name
+            for name, enabled in structured_signals.items()
+            if enabled and signals[name]
+        }
+        if (
+            category == "unknown"
+            and structured_category != "unknown"
+            and confirmed_tactics
+        ):
+            category = structured_category
     tactics = [name for name, enabled in signals.items() if enabled]
+    if input_type == "image" and not assessment and not example_id:
+        input_description = "image: Assessment unavailable"
+    else:
+        input_description = f"image: {safe_description(category, signals)}"
     return {
         "input": (
             f"text: {redact_text(text)}"
             if input_type == "text"
-            else f"image: {safe_description(category, signals)}"
+            else input_description
         ),
         "input_category": category,
         "urgency": signals["urgency"],
@@ -318,13 +466,13 @@ def build_trace_record(
         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"]
+    tactic_values = (
+        []
+        if input_profile["scam_tactics"] == "none"
+        else input_profile["scam_tactics"].split(", ")
+    )
+    signals = {name: name in tactic_values for name in SIGNAL_PATTERNS}
     assessment = assessment or {}
     return {
         "trace_id": trace_id,
@@ -346,22 +494,34 @@ def validate_trace(record: Any) -> list[str]:
     errors: list[str] = []
     if not isinstance(record, dict):
         return ["Trace must be an object."]
-    required = {
-        "trace_id",
-        "timestamp",
-        "input",
-        "input_category",
-        "urgency",
-        "scam_tactics",
-        "result_summary",
-        "risk_label",
-        "reply_draft_policy",
-    }
+    required = set(TRACE_FIELDS)
     missing = required - record.keys()
     if missing:
         errors.append("Missing fields: " + ", ".join(sorted(missing)))
+    unexpected = record.keys() - required
+    if unexpected:
+        errors.append("Unexpected fields: " + ", ".join(sorted(unexpected)))
     if record and next(iter(record)) != "trace_id":
         errors.append("trace_id must be the first column.")
+    trace_id = record.get("trace_id")
+    try:
+        parsed_trace_id = uuid.UUID(trace_id) if isinstance(trace_id, str) else None
+        if parsed_trace_id is None or str(parsed_trace_id) != trace_id:
+            raise ValueError
+    except (ValueError, AttributeError):
+        errors.append("Trace ID must be a canonical UUID.")
+    timestamp = record.get("timestamp")
+    try:
+        parsed_timestamp = (
+            datetime.fromisoformat(timestamp) if isinstance(timestamp, str) else None
+        )
+        if (
+            parsed_timestamp is None
+            or parsed_timestamp.utcoffset() != timezone.utc.utcoffset(None)
+        ):
+            raise ValueError
+    except (ValueError, TypeError):
+        errors.append("Timestamp must be an ISO 8601 UTC value.")
     input_value = record.get("input")
     if not (
         isinstance(input_value, str)
@@ -371,14 +531,58 @@ def validate_trace(record: Any) -> list[str]:
         )
     ):
         errors.append("Input must use a fixed text: or image: description.")
-    if not isinstance(record.get("input_category"), str):
-        errors.append("Input category must be a string.")
+    elif len(input_value) > 506:
+        errors.append("Input exceeds the 500-character content limit.")
+    else:
+        for label, pattern in RESIDUAL_IDENTIFIER_PATTERNS.items():
+            if pattern.search(input_value):
+                errors.append(f"Input contains an unredacted {label}.")
+    if record.get("input_category") not in CATEGORY_DISPLAY_NAMES:
+        errors.append("Invalid input category.")
     if not isinstance(record.get("urgency"), bool):
         errors.append("Urgency must be boolean.")
-    if not isinstance(record.get("result_summary"), str):
-        errors.append("Result summary must be a string.")
+    tactics = record.get("scam_tactics")
+    tactic_values: list[str] = []
+    if not isinstance(tactics, str):
+        errors.append("Scam tactics must be a string.")
+    else:
+        tactic_values = [] if tactics == "none" else tactics.split(", ")
+        if (
+            any(value not in SIGNAL_PATTERNS for value in tactic_values)
+            or len(tactic_values) != len(set(tactic_values))
+        ):
+            errors.append("Invalid scam tactics.")
+    result = record.get("result_summary")
+    if not isinstance(result, str) or not result or len(result) > 500:
+        errors.append("Result summary must be a non-empty string of at most 500 characters.")
     if record.get("risk_label") not in RISK_LABELS:
         errors.append("Invalid risk label.")
+    if record.get("reply_draft_policy") not in REPLY_DRAFT_POLICIES:
+        errors.append("Invalid reply draft policy.")
+    risk_label = record.get("risk_label")
+    category = record.get("input_category")
+    if risk_label in RISK_LABELS and category in CATEGORY_DISPLAY_NAMES:
+        expected_policy = (
+            "allowed"
+            if risk_label in {"Verify first", "Suspicious"}
+            else "suppressed"
+            if risk_label != "none"
+            else "not_applicable"
+        )
+        if record.get("reply_draft_policy") != expected_policy:
+            errors.append("Reply draft policy does not match the risk label.")
+        signals = {name: name in tactic_values for name in SIGNAL_PATTERNS}
+        expected_summary = result_summary(risk_label, category, signals)
+        if result != expected_summary:
+            errors.append("Result summary does not match the deterministic fields.")
+        if isinstance(input_value, str) and input_value.startswith("image: "):
+            expected_input = f"image: {safe_description(category, signals)}"
+            unavailable_input = (
+                risk_label == "none"
+                and input_value == "image: Assessment unavailable"
+            )
+            if input_value != expected_input and not unavailable_input:
+                errors.append("Image input does not match the deterministic fields.")
     if any(isinstance(value, (dict, list)) for value in record.values()):
         errors.append("Trace columns must contain scalar values only.")
     forbidden_keys = {

traces/scripts/analyze_trace_dataset.py

@@ -0,0 +1,160 @@
+"""Analyze local or Hugging Face privacy-safe trace JSONL files."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from collections import Counter
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from traces.runtime import DATASET_REPO, validate_trace
+
+
+def load_records(paths: list[Path]) -> tuple[list[dict[str, Any]], list[str]]:
+    records: list[dict[str, Any]] = []
+    errors: list[str] = []
+    trace_ids: set[str] = set()
+    for path in paths:
+        for line_number, line in enumerate(
+            path.read_text(encoding="utf-8").splitlines(),
+            start=1,
+        ):
+            if not line.strip():
+                continue
+            try:
+                record = json.loads(line)
+            except json.JSONDecodeError as exc:
+                errors.append(f"{path}:{line_number}: invalid JSON: {exc}")
+                continue
+            if not isinstance(record, dict):
+                errors.append(f"{path}:{line_number}: Trace must be an object.")
+                continue
+            trace_id = record.get("trace_id")
+            if trace_id in trace_ids:
+                errors.append(f"{path}:{line_number}: Duplicate trace ID: {trace_id}")
+            elif isinstance(trace_id, str):
+                trace_ids.add(trace_id)
+            errors.extend(
+                f"{path}:{line_number}: {error}"
+                for error in validate_trace(record)
+            )
+            records.append(record)
+    return records, errors
+
+
+def summarize(records: list[dict[str, Any]], file_count: int) -> dict[str, Any]:
+    inputs = Counter(
+        record.get("input")
+        for record in records
+        if isinstance(record.get("input"), str)
+    )
+    duplicate_rows = sum(count - 1 for count in inputs.values() if count > 1)
+    return {
+        "files": file_count,
+        "rows": len(records),
+        "unique_inputs": len(inputs),
+        "duplicate_input_rows": duplicate_rows,
+        "duplicate_input_rate": (
+            round(duplicate_rows / len(records), 4) if records else 0
+        ),
+        "unclassified_images": sum(
+            record.get("input")
+            == "image: Unclassified content with no mapped signals"
+            for record in records
+        ),
+        "unavailable_image_assessments": sum(
+            record.get("input") == "image: Assessment unavailable"
+            for record in records
+        ),
+        "incomplete_assessments": sum(
+            record.get("risk_label") == "none" for record in records
+        ),
+        "risk_labels": dict(Counter(
+            str(record.get("risk_label", "[missing]")) for record in records
+        ).most_common()),
+        "input_categories": dict(Counter(
+            str(record.get("input_category", "[missing]")) for record in records
+        ).most_common()),
+        "input_types": dict(Counter(
+            record["input"].split(":", 1)[0]
+            for record in records
+            if isinstance(record.get("input"), str)
+        ).most_common()),
+        "top_repeated_inputs": [
+            {"count": count, "input": input_value}
+            for input_value, count in inputs.most_common(10)
+            if count > 1
+        ],
+    }
+
+
+def hub_paths(repo_id: str, directory: Path) -> list[Path]:
+    try:
+        from huggingface_hub import HfApi, hf_hub_download
+    except ImportError as exc:
+        raise RuntimeError(
+            "Install huggingface_hub to analyze a Hub dataset."
+        ) from exc
+    api = HfApi()
+    filenames = [
+        name
+        for name in api.list_repo_files(repo_id, repo_type="dataset")
+        if name.endswith(".jsonl")
+    ]
+    return [
+        Path(hf_hub_download(
+            repo_id,
+            filename,
+            repo_type="dataset",
+            local_dir=directory,
+        ))
+        for filename in filenames
+    ]
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("paths", nargs="*", type=Path)
+    parser.add_argument("--repo-id", default=DATASET_REPO)
+    parser.add_argument("--json", action="store_true", dest="as_json")
+    args = parser.parse_args()
+
+    with TemporaryDirectory() as directory:
+        paths = args.paths or hub_paths(args.repo_id, Path(directory))
+        records, errors = load_records(paths)
+        report = summarize(records, len(paths))
+    report["validation_errors"] = len(errors)
+
+    if args.as_json:
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+    else:
+        print(f"Files: {report['files']}")
+        print(f"Rows: {report['rows']}")
+        print(f"Validation errors: {report['validation_errors']}")
+        print(f"Unique inputs: {report['unique_inputs']}")
+        print(
+            "Repeated-input rows: "
+            f"{report['duplicate_input_rows']} "
+            f"({report['duplicate_input_rate']:.1%})"
+        )
+        print(f"Unclassified images: {report['unclassified_images']}")
+        print(
+            "Unavailable image assessments: "
+            f"{report['unavailable_image_assessments']}"
+        )
+        print(f"Incomplete assessments: {report['incomplete_assessments']}")
+        print(f"Risk labels: {report['risk_labels']}")
+        print(f"Input categories: {report['input_categories']}")
+        if errors:
+            print("\n".join(errors), file=sys.stderr)
+    return 1 if errors else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

traces/scripts/validate_traces.py

@@ -17,6 +17,7 @@ from traces.runtime import validate_trace
 def validate_file(path: Path) -> tuple[int, list[str]]:
     count = 0
     errors: list[str] = []
+    trace_ids: set[str] = set()
     for line_number, line in enumerate(
         path.read_text(encoding="utf-8").splitlines(),
         start=1,
@@ -29,6 +30,11 @@ def validate_file(path: Path) -> tuple[int, list[str]]:
         except json.JSONDecodeError as exc:
             errors.append(f"{path}:{line_number}: invalid JSON: {exc}")
             continue
+        trace_id = record.get("trace_id") if isinstance(record, dict) else None
+        if trace_id in trace_ids:
+            errors.append(f"{path}:{line_number}: Duplicate trace ID: {trace_id}")
+        elif isinstance(trace_id, str):
+            trace_ids.add(trace_id)
         for error in validate_trace(record):
             errors.append(f"{path}:{line_number}: {error}")
     return count, errors