Add deterministic privacy-safe trace dataset

Co-Authored-By: Codex <codex@openai.com>

.gitignore

@@ -216,3 +216,7 @@ __marimo__/
 
 # Streamlit
 .streamlit/secrets.toml
+
+# Privacy-safe runtime trace shards are uploaded separately.
+traces/pending/
+traces/export/

README.md

@@ -68,6 +68,10 @@ OpenAI-compatible endpoint. It does not call OpenAI cloud APIs by default.
 | `MODEL_TIMEOUT_SECONDS` | Optional request timeout; default is 180 seconds |
 | `MODAL_PROXY_KEY` | Optional Modal proxy authentication key |
 | `MODAL_PROXY_SECRET` | Optional Modal proxy authentication secret |
+| `HF_TOKEN` | Scoped Hugging Face token used by the background trace uploader |
+| `HF_TRACE_DATASET_REPO` | Trace dataset repo; defaults to `build-small-hackathon/pakistan-notice-helper-traces` |
+| `TRACE_BATCH_SIZE` | Trace records per shard; default is 20 |
+| `TRACE_FLUSH_SECONDS` | Maximum batching delay; default is 60 seconds |
 
 The current defaults are:
 
@@ -115,6 +119,37 @@ unsloth/Qwen3.6-27B-MTP-GGUF
 All frontend assets are local. The app has no runtime CDN, analytics, OCR, MCP,
 or OpenAI Agents SDK. Analysis currently depends on the deployed Modal model.
 
+## Sharing is Caring: Open Pipeline Traces
+
+The app publishes optional privacy-safe backend traces to
+[`build-small-hackathon/pakistan-notice-helper-traces`](https://huggingface.co/datasets/build-small-hackathon/pakistan-notice-helper-traces).
+The checkbox is visible and enabled by default on each request, and users can
+turn it off before submitting.
+
+Trace creation is deterministic Python logic and makes no additional model
+request. It records the actual pipeline stages, cache/Modal status, duration
+buckets, fixed signal categories, result counts, and sanitized failure types.
+It never stores raw or redacted messages, screenshots, links, identifiers,
+model explanations, reply text, exceptions, or credentials.
+
+Safe records are queued without blocking the response, written in batches of
+20 or after 60 seconds, and uploaded as unique JSONL shards. Hub failures leave
+the shard pending for a later retry and do not affect scam analysis.
+
+Operator commands:
+
+```bash
+python scripts/seed_trace_dataset.py
+python scripts/validate_traces.py data/trace_samples.jsonl
+python scripts/create_trace_dataset.py --dry-run
+python scripts/create_trace_dataset.py
+python scripts/export_pending_traces.py --dry-run
+python scripts/upload_trace_shards.py --dry-run
+```
+
+See [the dataset card](docs/trace_dataset_card.md) for the schema, privacy
+policy, provenance, and limitations.
+
 ## Hugging Face Spaces
 
 Push this repository to a new Gradio Space. The metadata at the top of this
@@ -127,7 +162,8 @@ available as overrides for a future local deployment.
 
 - Submitted text and images are sent to the configured Modal endpoint and are
   not saved by this app.
-- The `traces/` directory contains only a placeholder; runtime tracing is off.
+- Public traces contain only allow-listed metadata, buckets, booleans, counts,
+  and fixed summaries. Tracing can be disabled per request.
 - Do not upload private personal data unless you trust the Modal deployment.
 - No automated result proves that a notice is genuine or fraudulent.
 - Image analysis requires a multimodal endpoint with its vision projector.
@@ -145,9 +181,15 @@ docs/
   research_notes.md
   model_experiment_notes.md
 data/
-  examples.jsonl
-sample_inputs/
+  example_assessments.json
+  trace_samples.jsonl
 traces/
+scripts/
+  create_trace_dataset.py
+  seed_trace_dataset.py
+  validate_traces.py
+  export_pending_traces.py
+  upload_trace_shards.py
 static/
   index.html
   styles.css
@@ -156,9 +198,8 @@ experiments/
   modal_qwen36_mtp/
 ```
 
-Existing public and synthetic examples in `data/examples.jsonl` cover courier,
-traffic challan, bank, FBR, wallet, job, utility, WhatsApp, and education scam
-patterns. Source screenshots are stored under `sample_inputs/`.
+The six bundled examples have cached Modal assessments and deterministic seed
+traces. Runtime trace shards are kept out of Git and uploaded separately.
 
 ## Official reporting channels
 

app.py

@@ -15,6 +15,7 @@ from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 from gradio import Server
 from openai import APIConnectionError, APIStatusError, APITimeoutError, OpenAI
+from trace_runtime import queue_trace, start_trace_worker, trace_status
 
 ROOT = Path(__file__).resolve().parent
 STATIC_DIR = ROOT / "static"
@@ -176,18 +177,33 @@ def load_example_cache() -> dict[str, dict[str, Any]]:
 EXAMPLE_ASSESSMENTS = load_example_cache()
 
 
-def parse_model_json(content: str) -> dict[str, Any]:
+def parse_model_json(
+    content: str, telemetry: dict[str, Any] | None = None
+) -> dict[str, Any]:
+    telemetry = telemetry if telemetry is not None else {}
     candidate = content.strip()
     if candidate.startswith("```"):
         candidate = re.sub(r"^```(?:json)?\s*", "", candidate, flags=re.I)
         candidate = re.sub(r"\s*```$", "", candidate)
+    parse_started = time.perf_counter()
     try:
-        return normalize_assessment(json.loads(candidate))
+        value = json.loads(candidate)
     except json.JSONDecodeError:
         match = re.search(r"\{.*\}", candidate, re.S)
         if not match:
             raise ValueError("Model did not return JSON.") from None
-        return normalize_assessment(json.loads(match.group(0)))
+        value = json.loads(match.group(0))
+    telemetry["parse_ms"] = (time.perf_counter() - parse_started) * 1000
+    telemetry["parse_completed"] = True
+    normalize_started = time.perf_counter()
+    try:
+        result = normalize_assessment(value)
+    finally:
+        telemetry["normalize_ms"] = (
+            time.perf_counter() - normalize_started
+        ) * 1000
+    telemetry["normalize_completed"] = True
+    return result
 
 
 def create_model_client() -> tuple[OpenAI, str]:
@@ -215,7 +231,12 @@ def create_model_client() -> tuple[OpenAI, str]:
     )
 
 
-def call_model(text: str, image_data_url: str) -> dict[str, Any]:
+def call_model(
+    text: str,
+    image_data_url: str,
+    telemetry: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    telemetry = telemetry if telemetry is not None else {}
     client, model_name = create_model_client()
     prompt = (
         "Assess the following Pakistani notice or message for scam risk. "
@@ -233,8 +254,22 @@ def call_model(text: str, image_data_url: str) -> dict[str, Any]:
 
     retries = max(1, int(os.getenv("MODEL_MAX_ATTEMPTS", "4")))
     retry_delay = max(0.0, float(os.getenv("MODEL_RETRY_DELAY_SECONDS", "5")))
+    telemetry.update(
+        {
+            "modal_called": False,
+            "modal_ms": 0.0,
+            "retry_count": 0,
+            "attempt_count": 0,
+            "parse_ms": 0.0,
+            "normalize_ms": 0.0,
+        }
+    )
     for attempt in range(1, retries + 1):
+        telemetry["attempt_count"] = attempt
+        telemetry["retry_count"] = attempt - 1
         try:
+            request_started = time.perf_counter()
+            telemetry["modal_called"] = True
             completion = client.chat.completions.create(
                 model=model_name,
                 messages=[
@@ -253,16 +288,27 @@ def call_model(text: str, image_data_url: str) -> dict[str, Any]:
                 },
                 extra_body={"chat_template_kwargs": {"enable_thinking": False}},
             )
+            telemetry["modal_ms"] += (
+                time.perf_counter() - request_started
+            ) * 1000
             raw = completion.choices[0].message.content
             if not raw:
                 raise ValueError("Model returned an empty response.")
-            return parse_model_json(raw)
+            return parse_model_json(raw, telemetry)
         except APIStatusError as exc:
+            telemetry["modal_ms"] += max(
+                0.0,
+                (time.perf_counter() - request_started) * 1000,
+            )
             if exc.status_code == 503 and attempt < retries:
                 time.sleep(retry_delay)
                 continue
             raise
         except (APIConnectionError, APITimeoutError):
+            telemetry["modal_ms"] += max(
+                0.0,
+                (time.perf_counter() - request_started) * 1000,
+            )
             if attempt == retries:
                 raise
             time.sleep(retry_delay)
@@ -271,55 +317,184 @@ def call_model(text: str, image_data_url: str) -> dict[str, Any]:
 
 
 def analyze_notice(
-    text: str = "", image_data_url: str = "", example_id: str = ""
+    text: str = "",
+    image_data_url: str = "",
+    example_id: str = "",
+    save_trace: bool = True,
 ) -> dict[str, Any]:
     """Analyze supplied text/image using the configured model only."""
+    request_started = time.perf_counter()
+    pipeline_status = {step: "skipped" for step in (
+        "receive",
+        "validate",
+        "cache_lookup",
+        "modal_request",
+        "parse_json",
+        "normalize_result",
+        "reply_filter",
+        "response",
+    )}
+    pipeline_ms: dict[str, float] = {}
+    pipeline_status["receive"] = "completed"
     text = (text or "").strip()
     image_data_url = image_data_url or ""
     example_id = (example_id or "").strip()
+
+    def finish(
+        response: dict[str, Any],
+        *,
+        request_source: str,
+        telemetry: dict[str, Any] | None = None,
+        failure_category: str = "none",
+        failure_stage: str = "none",
+    ) -> dict[str, Any]:
+        telemetry = telemetry or {}
+        pipeline_status["response"] = "completed"
+        pipeline_ms["response"] = (time.perf_counter() - request_started) * 1000
+        if save_trace:
+            trace_id, queued = queue_trace(
+                text=text,
+                image_data_url=image_data_url,
+                example_id=example_id,
+                request_source=request_source,
+                pipeline_status=pipeline_status,
+                pipeline_ms=pipeline_ms,
+                modal_called=bool(telemetry.get("modal_called", False)),
+                modal_ms=float(telemetry.get("modal_ms", 0.0)),
+                retry_count=int(telemetry.get("retry_count", 0)),
+                assessment=response.get("assessment"),
+                failure_category=failure_category,
+                failure_stage=failure_stage,
+            )
+            response["trace"] = {"trace_id": trace_id, "status": queued}
+        else:
+            response["trace"] = {"trace_id": "", "status": "disabled"}
+        return response
+
+    validation_started = time.perf_counter()
+    valid_example = example_id in EXAMPLE_ASSESSMENTS
+    if not text and not image_data_url and not valid_example:
+        pipeline_status["validate"] = "rejected"
+        pipeline_ms["validate"] = (time.perf_counter() - validation_started) * 1000
+        return finish(
+            {
+                "ok": False,
+                "error": "Paste a message or upload a screenshot to continue.",
+                "status": model_status(),
+            },
+            request_source="user",
+            failure_category="validation_empty",
+            failure_stage="validate",
+        )
+    pipeline_status["validate"] = "completed"
+    pipeline_ms["validate"] = (time.perf_counter() - validation_started) * 1000
+
+    cache_started = time.perf_counter()
     if example_id in EXAMPLE_ASSESSMENTS:
-        return {
-            "ok": True,
-            "assessment": dict(EXAMPLE_ASSESSMENTS[example_id]),
-            "status": model_status(),
-            "source": "cached_modal_example",
-        }
-    if not text and not image_data_url:
-        return {
-            "ok": False,
-            "error": "Paste a message or upload a screenshot to continue.",
-            "status": model_status(),
-        }
+        pipeline_status["cache_lookup"] = "hit"
+        pipeline_ms["cache_lookup"] = (time.perf_counter() - cache_started) * 1000
+        pipeline_status["reply_filter"] = "completed"
+        return finish(
+            {
+                "ok": True,
+                "assessment": dict(EXAMPLE_ASSESSMENTS[example_id]),
+                "status": model_status(),
+                "source": "cached_modal_example",
+            },
+            request_source="cached_modal_example",
+        )
+    pipeline_status["cache_lookup"] = "miss"
+    pipeline_ms["cache_lookup"] = (time.perf_counter() - cache_started) * 1000
 
     status = model_status()
     if not status["connected"]:
-        return {
-            "ok": False,
-            "error": (
-                "The Modal model requires MODAL_PROXY_KEY and "
-                "MODAL_PROXY_SECRET. Add them as environment variables or "
-                "Hugging Face Space secrets."
-            ),
-            "status": status,
-        }
+        pipeline_status["modal_request"] = "skipped"
+        return finish(
+            {
+                "ok": False,
+                "error": (
+                    "The Modal model requires MODAL_PROXY_KEY and "
+                    "MODAL_PROXY_SECRET. Add them as environment variables or "
+                    "Hugging Face Space secrets."
+                ),
+                "status": status,
+            },
+            request_source="user",
+            failure_category="credentials_missing",
+            failure_stage="modal_request",
+        )
+    telemetry: dict[str, Any] = {}
     try:
-        result = call_model(text, image_data_url)
-        return {"ok": True, "assessment": result, "status": status, "source": "model"}
+        result = call_model(text, image_data_url, telemetry)
+        pipeline_status["modal_request"] = "completed"
+        pipeline_status["parse_json"] = "completed"
+        pipeline_status["normalize_result"] = "completed"
+        pipeline_status["reply_filter"] = "completed"
+        pipeline_ms["modal_request"] = float(telemetry.get("modal_ms", 0.0))
+        pipeline_ms["parse_json"] = float(telemetry.get("parse_ms", 0.0))
+        pipeline_ms["normalize_result"] = float(telemetry.get("normalize_ms", 0.0))
+        return finish(
+            {
+                "ok": True,
+                "assessment": result,
+                "status": status,
+                "source": "model",
+            },
+            request_source="user",
+            telemetry=telemetry,
+        )
     except APIStatusError as exc:
+        pipeline_status["modal_request"] = "failed"
+        pipeline_ms["modal_request"] = float(telemetry.get("modal_ms", 0.0))
         message = (
             "The Modal model rejected the request. Check the proxy credentials."
             if exc.status_code in {401, 403}
             else f"The Modal model returned HTTP {exc.status_code}. Try again shortly."
         )
-    except (APIConnectionError, APITimeoutError):
+        failure_category = "http_auth" if exc.status_code in {401, 403} else "http_error"
+    except APITimeoutError:
+        pipeline_status["modal_request"] = "failed"
+        pipeline_ms["modal_request"] = float(telemetry.get("modal_ms", 0.0))
+        message = "The Modal model is unavailable or still starting. Try again shortly."
+        failure_category = "timeout"
+    except APIConnectionError:
+        pipeline_status["modal_request"] = "failed"
+        pipeline_ms["modal_request"] = float(telemetry.get("modal_ms", 0.0))
         message = "The Modal model is unavailable or still starting. Try again shortly."
+        failure_category = "connection_error"
     except (ValueError, RuntimeError):
+        pipeline_status["modal_request"] = (
+            "completed" if telemetry.get("modal_called") else "skipped"
+        )
+        pipeline_ms["modal_request"] = float(telemetry.get("modal_ms", 0.0))
+        pipeline_ms["parse_json"] = float(telemetry.get("parse_ms", 0.0))
+        pipeline_ms["normalize_result"] = float(
+            telemetry.get("normalize_ms", 0.0)
+        )
+        if telemetry.get("parse_completed"):
+            pipeline_status["parse_json"] = "completed"
+            pipeline_status["normalize_result"] = "failed"
+            failure_stage = "normalize_result"
+        else:
+            pipeline_status["parse_json"] = "failed"
+            failure_stage = "parse_json"
         message = "The model returned an invalid response. Please try again."
-    return {
-        "ok": False,
-        "error": message,
-        "status": {**status, "connected": False, "label": "Modal model unavailable"},
-    }
+        failure_category = "invalid_model_output"
+    return finish(
+        {
+            "ok": False,
+            "error": message,
+            "status": {**status, "connected": False, "label": "Modal model unavailable"},
+        },
+        request_source="user",
+        telemetry=telemetry,
+        failure_category=failure_category,
+        failure_stage=(
+            failure_stage
+            if failure_category == "invalid_model_output"
+            else "modal_request"
+        ),
+    )
 
 
 app = Server()
@@ -328,9 +503,12 @@ app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
 
 @app.api(name="analyze", description="Assess a notice for common scam signals.", concurrency_limit=1)
 def analyze_api(
-    text: str = "", image_data_url: str = "", example_id: str = ""
+    text: str = "",
+    image_data_url: str = "",
+    example_id: str = "",
+    save_trace: bool = True,
 ) -> dict[str, Any]:
-    return analyze_notice(text, image_data_url, example_id)
+    return analyze_notice(text, image_data_url, example_id, save_trace)
 
 
 @app.api(name="status", description="Return model and privacy status.", queue=False)
@@ -338,6 +516,11 @@ def status_api() -> dict[str, Any]:
     return model_status()
 
 
+@app.api(name="trace_status", description="Return privacy-safe trace queue status.", queue=False)
+def trace_status_api() -> dict[str, Any]:
+    return trace_status()
+
+
 @app.get("/", include_in_schema=False)
 async def index() -> FileResponse:
     return FileResponse(STATIC_DIR / "index.html")
@@ -382,11 +565,11 @@ def run_self_tests() -> None:
         }
     )
     assert inappropriate["reply_draft"] == ""
-    cached = analyze_notice(example_id="text-bank")
+    cached = analyze_notice(example_id="text-bank", save_trace=False)
     assert cached["ok"] is True
     assert cached["source"] == "cached_modal_example"
     assert cached["assessment"]["risk_label"] == "Likely scam"
-    assert analyze_notice("", "")["ok"] is False
+    assert analyze_notice("", "", save_trace=False)["ok"] is False
     try:
         normalize_assessment({"risk_label": "Looks normal"})
     except ValueError:
@@ -431,6 +614,7 @@ def main() -> int:
         if args.test_endpoint:
             test_endpoint()
             return 0
+        start_trace_worker()
         app.launch(server_name=args.host, server_port=args.port)
         return 0
     except (

data/trace_samples.jsonl

@@ -0,0 +1,6 @@
+{"app_commit": "unknown", "cache": {"example_id": "text-courier", "hit": true}, "failure": {"category": "none", "stage": "none"}, "input": {"category": "courier", "image_size_bucket": "none", "language_hint": "latin_script", "safe_summary": "Courier-style text input with link, urgency, payment, courier signals", "signals": {"account_threat": false, "challan": false, "cnic": false, "courier": true, "credentials": false, "link": true, "otp": false, "payment": true, "refund_or_prize": false, "urgency": true}, "text_byte_bucket": "1-160", "text_character_bucket": "1-160", "type": "text"}, "modal": {"called": false, "latency_bucket": "0-1ms", "model_family": "qwen3.6-27b-mtp", "outcome": "not_called", "retry_count": 0}, "pipeline_steps": [{"duration_bucket": "0-1ms", "status": "completed", "step": "receive"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "validate"}, {"duration_bucket": "0-1ms", "status": "hit", "step": "cache_lookup"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "modal_request"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "parse_json"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "normalize_result"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "reply_filter"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "response"}], "privacy": {"exception_text_stored": false, "identifiers_stored": false, "raw_image_stored": false, "raw_input_stored": false, "raw_model_output_stored": false}, "request_source": "cached_modal_example", "result": {"red_flag_count": 4, "reply_draft_policy": "suppressed", "reply_draft_returned": false, "risk_label": "Likely scam", "safe_next_step_count": 4}, "schema_version": "1.0", "timestamp": "2026-06-07T06:08:05.356371+00:00", "trace_id": "72c1dd56-9ae4-42d2-a8cd-7333f48d195e"}
+{"app_commit": "unknown", "cache": {"example_id": "text-fbr", "hit": true}, "failure": {"category": "none", "stage": "none"}, "input": {"category": "fbr", "image_size_bucket": "none", "language_hint": "latin_script", "safe_summary": "FBR-style text input with CNIC, credential, urgency, payment signals", "signals": {"account_threat": false, "challan": false, "cnic": true, "courier": false, "credentials": true, "link": false, "otp": false, "payment": true, "refund_or_prize": true, "urgency": true}, "text_byte_bucket": "1-160", "text_character_bucket": "1-160", "type": "text"}, "modal": {"called": false, "latency_bucket": "0-1ms", "model_family": "qwen3.6-27b-mtp", "outcome": "not_called", "retry_count": 0}, "pipeline_steps": [{"duration_bucket": "0-1ms", "status": "completed", "step": "receive"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "validate"}, {"duration_bucket": "0-1ms", "status": "hit", "step": "cache_lookup"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "modal_request"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "parse_json"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "normalize_result"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "reply_filter"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "response"}], "privacy": {"exception_text_stored": false, "identifiers_stored": false, "raw_image_stored": false, "raw_input_stored": false, "raw_model_output_stored": false}, "request_source": "cached_modal_example", "result": {"red_flag_count": 4, "reply_draft_policy": "suppressed", "reply_draft_returned": false, "risk_label": "Likely scam", "safe_next_step_count": 4}, "schema_version": "1.0", "timestamp": "2026-06-07T06:08:05.357519+00:00", "trace_id": "12bb3cf0-43e1-464b-b004-e9948cc4ed2a"}
+{"app_commit": "unknown", "cache": {"example_id": "text-bank", "hit": true}, "failure": {"category": "none", "stage": "none"}, "input": {"category": "bank", "image_size_bucket": "none", "language_hint": "latin_script", "safe_summary": "Bank-style text input with OTP, urgency, account-threat signals", "signals": {"account_threat": true, "challan": false, "cnic": false, "courier": false, "credentials": false, "link": false, "otp": true, "payment": false, "refund_or_prize": false, "urgency": true}, "text_byte_bucket": "1-160", "text_character_bucket": "1-160", "type": "text"}, "modal": {"called": false, "latency_bucket": "0-1ms", "model_family": "qwen3.6-27b-mtp", "outcome": "not_called", "retry_count": 0}, "pipeline_steps": [{"duration_bucket": "0-1ms", "status": "completed", "step": "receive"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "validate"}, {"duration_bucket": "0-1ms", "status": "hit", "step": "cache_lookup"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "modal_request"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "parse_json"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "normalize_result"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "reply_filter"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "response"}], "privacy": {"exception_text_stored": false, "identifiers_stored": false, "raw_image_stored": false, "raw_input_stored": false, "raw_model_output_stored": false}, "request_source": "cached_modal_example", "result": {"red_flag_count": 3, "reply_draft_policy": "suppressed", "reply_draft_returned": false, "risk_label": "Likely scam", "safe_next_step_count": 5}, "schema_version": "1.0", "timestamp": "2026-06-07T06:08:05.357666+00:00", "trace_id": "80403716-5ed5-4ee6-8432-e0677f3a75c4"}
+{"app_commit": "unknown", "cache": {"example_id": "image-courier", "hit": true}, "failure": {"category": "none", "stage": "none"}, "input": {"category": "courier", "image_size_bucket": "up-to-100KB", "language_hint": "unknown", "safe_summary": "Courier-style image input with link, urgency, courier signals", "signals": {"account_threat": false, "challan": false, "cnic": false, "courier": true, "credentials": false, "link": true, "otp": false, "payment": false, "refund_or_prize": false, "urgency": true}, "text_byte_bucket": "empty", "text_character_bucket": "empty", "type": "image"}, "modal": {"called": false, "latency_bucket": "0-1ms", "model_family": "qwen3.6-27b-mtp", "outcome": "not_called", "retry_count": 0}, "pipeline_steps": [{"duration_bucket": "0-1ms", "status": "completed", "step": "receive"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "validate"}, {"duration_bucket": "0-1ms", "status": "hit", "step": "cache_lookup"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "modal_request"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "parse_json"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "normalize_result"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "reply_filter"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "response"}], "privacy": {"exception_text_stored": false, "identifiers_stored": false, "raw_image_stored": false, "raw_input_stored": false, "raw_model_output_stored": false}, "request_source": "cached_modal_example", "result": {"red_flag_count": 4, "reply_draft_policy": "suppressed", "reply_draft_returned": false, "risk_label": "Likely scam", "safe_next_step_count": 4}, "schema_version": "1.0", "timestamp": "2026-06-07T06:08:05.357884+00:00", "trace_id": "2b7fbebf-43dd-4a5a-bf45-06059f3dc4ff"}
+{"app_commit": "unknown", "cache": {"example_id": "image-mobile", "hit": true}, "failure": {"category": "none", "stage": "none"}, "input": {"category": "marketplace", "image_size_bucket": "500KB-2MB", "language_hint": "unknown", "safe_summary": "Marketplace-style image input with credential signals", "signals": {"account_threat": false, "challan": false, "cnic": false, "courier": false, "credentials": true, "link": false, "otp": false, "payment": false, "refund_or_prize": false, "urgency": false}, "text_byte_bucket": "empty", "text_character_bucket": "empty", "type": "image"}, "modal": {"called": false, "latency_bucket": "0-1ms", "model_family": "qwen3.6-27b-mtp", "outcome": "not_called", "retry_count": 0}, "pipeline_steps": [{"duration_bucket": "0-1ms", "status": "completed", "step": "receive"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "validate"}, {"duration_bucket": "0-1ms", "status": "hit", "step": "cache_lookup"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "modal_request"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "parse_json"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "normalize_result"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "reply_filter"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "response"}], "privacy": {"exception_text_stored": false, "identifiers_stored": false, "raw_image_stored": false, "raw_input_stored": false, "raw_model_output_stored": false}, "request_source": "cached_modal_example", "result": {"red_flag_count": 3, "reply_draft_policy": "suppressed", "reply_draft_returned": false, "risk_label": "Likely scam", "safe_next_step_count": 4}, "schema_version": "1.0", "timestamp": "2026-06-07T06:08:05.358366+00:00", "trace_id": "2d890292-43af-4e83-9afd-8069df37dc16"}
+{"app_commit": "unknown", "cache": {"example_id": "image-traffic", "hit": true}, "failure": {"category": "none", "stage": "none"}, "input": {"category": "traffic_challan", "image_size_bucket": "up-to-100KB", "language_hint": "unknown", "safe_summary": "Traffic-challan-style image input with link, urgency, payment, challan signals", "signals": {"account_threat": false, "challan": true, "cnic": false, "courier": false, "credentials": false, "link": true, "otp": false, "payment": true, "refund_or_prize": false, "urgency": true}, "text_byte_bucket": "empty", "text_character_bucket": "empty", "type": "image"}, "modal": {"called": false, "latency_bucket": "0-1ms", "model_family": "qwen3.6-27b-mtp", "outcome": "not_called", "retry_count": 0}, "pipeline_steps": [{"duration_bucket": "0-1ms", "status": "completed", "step": "receive"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "validate"}, {"duration_bucket": "0-1ms", "status": "hit", "step": "cache_lookup"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "modal_request"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "parse_json"}, {"duration_bucket": "0-1ms", "status": "skipped", "step": "normalize_result"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "reply_filter"}, {"duration_bucket": "0-1ms", "status": "completed", "step": "response"}], "privacy": {"exception_text_stored": false, "identifiers_stored": false, "raw_image_stored": false, "raw_input_stored": false, "raw_model_output_stored": false}, "request_source": "cached_modal_example", "result": {"red_flag_count": 3, "reply_draft_policy": "suppressed", "reply_draft_returned": false, "risk_label": "Likely scam", "safe_next_step_count": 4}, "schema_version": "1.0", "timestamp": "2026-06-07T06:08:05.358581+00:00", "trace_id": "872ba35e-76b8-41e5-9cfc-052656cca1ff"}

docs/trace_dataset_card.md

@@ -0,0 +1,101 @@
+---
+license: cc-by-4.0
+pretty_name: Pakistan Notice Helper Privacy-Safe Pipeline Traces
+task_categories:
+  - text-classification
+language:
+  - en
+  - ur
+tags:
+  - safety
+  - scams
+  - pakistan
+  - pipeline-traces
+  - privacy
+configs:
+  - config_name: default
+    data_files:
+      - split: train
+        path: data/**/*.jsonl
+---
+
+# Pakistan Notice Helper Privacy-Safe Pipeline Traces
+
+## Purpose
+
+This dataset shows how Pakistan Notice Helper processes scam-check requests.
+It contains deterministic application-pipeline traces, not hidden model
+reasoning and not 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.
+
+## Pipeline
+
+Each record follows these ordered stages:
+
+1. `receive`
+2. `validate`
+3. `cache_lookup`
+4. `modal_request`
+5. `parse_json`
+6. `normalize_result`
+7. `reply_filter`
+8. `response`
+
+Stages may be completed, skipped, rejected, failed, hit, or miss.
+
+## Fields
+
+- Trace identity: schema version, random trace ID, UTC timestamp, app commit
+- Input profile: type, size buckets, category, script/language hint
+- Deterministic signals: OTP, CNIC, credentials, link, urgency, payment,
+  refund/prize, courier, challan, and account threat
+- Fixed-template safe summary
+- Cache and Modal-call metadata
+- Pipeline status and duration buckets
+- Final risk label and output item counts
+- Sanitized failure category and stage
+- Explicit privacy flags
+
+## Privacy
+
+The dataset never stores:
+
+- Raw or redacted message text
+- Screenshots, image bytes, or base64
+- URLs, phone numbers, CNICs, names, addresses, account/card numbers, or
+  tracking numbers
+- Model explanations, red flags, safe-step text, reply drafts, or raw model
+  output
+- Exceptions, credentials, tokens, or endpoint headers
+
+Summaries use fixed templates. Regex detection happens transiently in memory.
+Users see a checked trace disclosure in the app and may opt out before each
+request.
+
+## Provenance
+
+Seed traces represent the six public examples bundled with Pakistan Notice
+Helper. Runtime traces may represent successful, cached, rejected, or failed
+requests. A trace reports whether the existing Modal request occurred, but
+trace generation itself does not invoke the model.
+
+## Limitations
+
+- Regex signals and category detection are approximate.
+- Duration and input sizes are deliberately bucketed.
+- The dataset cannot reproduce original messages or screenshots.
+- A risk label is safety guidance, not official verification.
+- Records with `app_commit: unknown` were generated where commit metadata was
+  unavailable.
+
+## Links
+
+- App: https://huggingface.co/spaces/build-small-hackathon/pakistan-notice-helper
+- Source: https://github.com/kingabzpro/pakistan-notice-helper
+
+## License
+
+CC BY 4.0.

requirements.txt

@@ -1,2 +1,3 @@
 gradio==6.15.1
 openai==2.33.0
+huggingface_hub==1.18.0

scripts/__init__.py

@@ -0,0 +1 @@
+"""Trace dataset administration scripts."""

scripts/create_trace_dataset.py

@@ -0,0 +1,52 @@
+"""Create and initialize the public Hugging Face trace dataset."""
+
+from __future__ import annotations
+
+import argparse
+import os
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_REPO = "build-small-hackathon/pakistan-notice-helper-traces"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--repo-id",
+        default=os.getenv("HF_TRACE_DATASET_REPO", DEFAULT_REPO),
+    )
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+    files = {
+        ROOT / "docs" / "trace_dataset_card.md": "README.md",
+        ROOT / "data" / "trace_samples.jsonl": "data/seed/trace_samples.jsonl",
+    }
+    if args.dry_run:
+        print(f"Would create public dataset: {args.repo_id}")
+        for local, remote in files.items():
+            print(f"Would upload {local} -> {remote}")
+        return 0
+    api = HfApi(token=os.getenv("HF_TOKEN") or None)
+    api.create_repo(
+        repo_id=args.repo_id,
+        repo_type="dataset",
+        private=False,
+        exist_ok=True,
+    )
+    for local, remote in files.items():
+        api.upload_file(
+            path_or_fileobj=str(local),
+            path_in_repo=remote,
+            repo_id=args.repo_id,
+            repo_type="dataset",
+            commit_message=f"Add {remote}",
+        )
+    print(f"Initialized https://huggingface.co/datasets/{args.repo_id}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/export_pending_traces.py

@@ -0,0 +1,49 @@
+"""Export validated pending trace shards into one JSONL file."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from trace_runtime import PENDING_DIR, validate_trace
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=ROOT / "traces" / "export" / "trace_export.jsonl",
+    )
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+    records = []
+    for path in sorted(PENDING_DIR.glob("*.jsonl")):
+        for line in path.read_text(encoding="utf-8").splitlines():
+            if not line:
+                continue
+            record = json.loads(line)
+            errors = validate_trace(record)
+            if errors:
+                raise RuntimeError(f"{path}: {'; '.join(errors)}")
+            records.append(record)
+    content = "".join(
+        json.dumps(record, sort_keys=True, ensure_ascii=True) + "\n"
+        for record in records
+    )
+    if args.dry_run:
+        print(content, end="")
+        return 0
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(content, encoding="utf-8")
+    print(f"Exported {len(records)} traces to {args.output}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/seed_trace_dataset.py

@@ -0,0 +1,106 @@
+"""Generate deterministic seed traces for the six built-in examples."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from trace_runtime import PIPELINE_STEPS, build_trace_record, validate_trace
+
+TEXT_EXAMPLES = {
+    "text-courier": (
+        "PAKISTAN POST: Your parcel address is incomplete. Pay Rs. 85 today "
+        "at http://pakpost-delivery.xyz or the parcel will be destroyed."
+    ),
+    "text-fbr": (
+        "FBR REFUND: You are eligible for Rs 42,500. Submit your CNIC and bank "
+        "card details at the link today to receive payment."
+    ),
+    "text-bank": (
+        "HBL Security: Your account will be suspended. Share the OTP sent to "
+        "your phone with our support team immediately."
+    ),
+}
+IMAGE_EXAMPLES = {
+    "image-courier": ROOT / "static" / "example-courier.jpeg",
+    "image-mobile": ROOT / "static" / "example-mobile.png",
+    "image-traffic": ROOT / "static" / "example-trafic.png",
+}
+
+
+def build_seed_records() -> list[dict]:
+    assessments = json.loads(
+        (ROOT / "data" / "example_assessments.json").read_text(encoding="utf-8")
+    )["examples"]
+    pipeline_status = {
+        step: (
+            "completed"
+            if step
+            in {
+                "receive",
+                "validate",
+                "reply_filter",
+                "response",
+            }
+            else "hit"
+            if step == "cache_lookup"
+            else "skipped"
+        )
+        for step in PIPELINE_STEPS
+    }
+    records = []
+    for example_id, assessment in assessments.items():
+        text = TEXT_EXAMPLES.get(example_id, "")
+        image_path = IMAGE_EXAMPLES.get(example_id)
+        image_placeholder = ""
+        if image_path:
+            image_placeholder = "x" * int(image_path.stat().st_size * 4 / 3)
+        record = build_trace_record(
+            text=text,
+            image_data_url=image_placeholder,
+            example_id=example_id,
+            request_source="cached_modal_example",
+            pipeline_status=pipeline_status,
+            pipeline_ms={},
+            modal_called=False,
+            modal_ms=0,
+            retry_count=0,
+            assessment=assessment,
+        )
+        errors = validate_trace(record)
+        if errors:
+            raise RuntimeError(f"{example_id}: {'; '.join(errors)}")
+        records.append(record)
+    return records
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=ROOT / "data" / "trace_samples.jsonl",
+    )
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+    records = build_seed_records()
+    content = "".join(
+        json.dumps(record, sort_keys=True, ensure_ascii=True) + "\n"
+        for record in records
+    )
+    if args.dry_run:
+        print(content, end="")
+        return 0
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(content, encoding="utf-8")
+    print(f"Wrote {len(records)} traces to {args.output}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/upload_trace_shards.py

@@ -0,0 +1,57 @@
+"""Validate and upload pending privacy-safe trace shards."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from scripts.validate_traces import validate_file
+from trace_runtime import DATASET_REPO, PENDING_DIR
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--repo-id",
+        default=os.getenv("HF_TRACE_DATASET_REPO", DATASET_REPO),
+    )
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--keep", action="store_true")
+    args = parser.parse_args()
+    paths = sorted(PENDING_DIR.glob("*.jsonl"))
+    if not paths:
+        print("No pending trace shards.")
+        return 0
+    api = HfApi(token=os.getenv("HF_TOKEN") or None)
+    date_path = datetime.now(timezone.utc).strftime("%Y/%m/%d")
+    for path in paths:
+        count, errors = validate_file(path)
+        if errors:
+            raise RuntimeError("\n".join(errors))
+        remote = f"data/{date_path}/{path.name}"
+        if args.dry_run:
+            print(f"Would upload {count} records: {path} -> {remote}")
+            continue
+        api.upload_file(
+            path_or_fileobj=str(path),
+            path_in_repo=remote,
+            repo_id=args.repo_id,
+            repo_type="dataset",
+            commit_message=f"Add privacy-safe trace shard {path.name}",
+        )
+        if not args.keep:
+            path.unlink()
+        print(f"Uploaded {count} records to {remote}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/validate_traces.py

@@ -0,0 +1,58 @@
+"""Validate privacy-safe trace JSONL files."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from trace_runtime import validate_trace
+
+
+def validate_file(path: Path) -> tuple[int, list[str]]:
+    count = 0
+    errors: list[str] = []
+    for line_number, line in enumerate(
+        path.read_text(encoding="utf-8").splitlines(),
+        start=1,
+    ):
+        if not line.strip():
+            continue
+        count += 1
+        try:
+            record = json.loads(line)
+        except json.JSONDecodeError as exc:
+            errors.append(f"{path}:{line_number}: invalid JSON: {exc}")
+            continue
+        for error in validate_trace(record):
+            errors.append(f"{path}:{line_number}: {error}")
+    return count, errors
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("paths", nargs="*", type=Path)
+    args = parser.parse_args()
+    paths = args.paths or [ROOT / "data" / "trace_samples.jsonl"]
+    total = 0
+    all_errors: list[str] = []
+    for path in paths:
+        if not path.exists():
+            all_errors.append(f"{path}: file does not exist")
+            continue
+        count, errors = validate_file(path)
+        total += count
+        all_errors.extend(errors)
+    if all_errors:
+        print("\n".join(all_errors), file=sys.stderr)
+        return 1
+    print(f"Validated {total} trace records.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

static/app.js

@@ -15,6 +15,7 @@ const elements = {
   source: document.querySelector("#resultSource"),
   uploadHint: document.querySelector("#uploadHint"),
   textHint: document.querySelector("#textHint"),
+  saveTrace: document.querySelector("#saveTrace"),
 };
 
 let imageDataUrl = "";
@@ -250,7 +251,7 @@ elements.form.addEventListener("submit", async (event) => {
     const submittedImage = activeExampleId ? "" : imageDataUrl;
     renderResult(await callGradioApi(
       "analyze",
-      [elements.text.value, submittedImage, activeExampleId],
+      [elements.text.value, submittedImage, activeExampleId, elements.saveTrace.checked],
     ));
   } catch (error) {
     showError(error.message || "The request could not be completed.");

static/index.html

@@ -74,6 +74,13 @@
         </div>
 
         <div id="formError" class="form-error" role="alert"></div>
+        <label class="trace-consent" for="saveTrace">
+          <input id="saveTrace" type="checkbox" checked>
+          <span>
+            <strong>Publish privacy-safe trace</strong>
+            <small>No raw message, screenshot, link, identifier, or model text is stored.</small>
+          </span>
+        </label>
         <div class="form-actions">
           <button id="analyzeButton" class="primary-button" type="submit">
             <span class="button-label">Check this notice</span>

static/styles.css

@@ -136,6 +136,14 @@ textarea:disabled { opacity: .45; background: #f0f5f2; cursor: not-allowed; }
 .form-actions { display: flex; justify-content: center; align-items: center; gap: 14px; margin-top: 26px; }
 .form-error { display: none; margin-top: 16px; padding: 12px 14px; border-radius: 12px; background: #fff0ef; color: #8d2722; font-size: 13px; }
 .form-error.visible { display: block; }
+.trace-consent {
+  margin-top: 18px; padding: 13px 15px; display: flex; align-items: flex-start; gap: 11px;
+  border: 1px solid var(--line); border-radius: 14px; background: #f8fbf9; cursor: pointer;
+}
+.trace-consent input { width: 18px; height: 18px; margin: 2px 0 0; accent-color: var(--green-700); }
+.trace-consent span, .trace-consent strong, .trace-consent small { display: block; }
+.trace-consent strong { color: var(--green-950); font-size: 13px; }
+.trace-consent small { margin-top: 3px; color: var(--muted); font-size: 11px; line-height: 1.45; }
 
 .mode-hint {
   display: none; margin-top: 10px; padding: 8px 14px; border-radius: 10px;

tests/__init__.py

@@ -0,0 +1 @@
+"""Test package."""

tests/test_tracing.py

@@ -0,0 +1,292 @@
+"""Tests for deterministic privacy-safe tracing."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import httpx
+from openai import APIStatusError, APITimeoutError
+
+import app
+import trace_runtime
+
+
+class TraceTests(unittest.TestCase):
+    def sample_record(self) -> dict:
+        return trace_runtime.build_trace_record(
+            text=(
+                "Ali call +92 300 1234567 CNIC 35202-1234567-1 "
+                "card 4111111111111111 https://private.example/a "
+                "account PK00TEST address House 10 tracking ABC123"
+            ),
+            image_data_url="data:image/png;base64,PRIVATE_IMAGE_BYTES",
+            example_id="",
+            request_source="user",
+            pipeline_status={step: "completed" for step in trace_runtime.PIPELINE_STEPS},
+            pipeline_ms={},
+            modal_called=True,
+            modal_ms=120,
+            retry_count=0,
+            assessment={
+                "risk_label": "Likely scam",
+                "simple_explanation": "PRIVATE MODEL EXPLANATION",
+                "red_flags": ["PRIVATE FLAG"],
+                "safe_next_steps": ["PRIVATE STEP"],
+                "reply_draft": "PRIVATE REPLY",
+            },
+        )
+
+    def test_trace_has_no_private_canaries(self) -> None:
+        serialized = json.dumps(self.sample_record())
+        for canary in (
+            "Ali",
+            "+92 300 1234567",
+            "35202-1234567-1",
+            "4111111111111111",
+            "private.example",
+            "PK00TEST",
+            "House 10",
+            "ABC123",
+            "PRIVATE_IMAGE_BYTES",
+            "PRIVATE MODEL EXPLANATION",
+            "PRIVATE FLAG",
+            "PRIVATE STEP",
+            "PRIVATE REPLY",
+        ):
+            self.assertNotIn(canary, serialized)
+
+    def test_trace_validation_and_performance(self) -> None:
+        started = time.perf_counter()
+        records = [self.sample_record() for _ in range(200)]
+        elapsed_ms = (time.perf_counter() - started) * 1000 / len(records)
+        self.assertLess(elapsed_ms, 10)
+        self.assertFalse(trace_runtime.validate_trace(records[0]))
+
+    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)
+        queue_mock.assert_not_called()
+        self.assertEqual(result["trace"]["status"], "disabled")
+
+    def test_cached_trace_does_not_call_model(self) -> None:
+        with patch("app.call_model") as model_mock, patch(
+            "app.queue_trace",
+            return_value=("trace-id", "queued"),
+        ):
+            result = app.analyze_notice(example_id="text-bank")
+        model_mock.assert_not_called()
+        self.assertEqual(result["source"], "cached_modal_example")
+
+    def test_empty_input_traces_sanitized_validation_failure(self) -> None:
+        with patch(
+            "app.queue_trace",
+            return_value=("trace-id", "queued"),
+        ) as queue_mock:
+            result = app.analyze_notice("")
+        self.assertFalse(result["ok"])
+        self.assertEqual(
+            queue_mock.call_args.kwargs["failure_category"],
+            "validation_empty",
+        )
+
+    def test_missing_credentials_traces_without_model_call(self) -> None:
+        with patch(
+            "app.model_status",
+            return_value={"connected": False, "label": "missing"},
+        ), patch("app.call_model") as model_mock, patch(
+            "app.queue_trace",
+            return_value=("trace-id", "queued"),
+        ) as queue_mock:
+            result = app.analyze_notice("test message")
+        self.assertFalse(result["ok"])
+        model_mock.assert_not_called()
+        self.assertEqual(
+            queue_mock.call_args.kwargs["failure_category"],
+            "credentials_missing",
+        )
+
+    def test_success_uses_existing_model_call_once(self) -> None:
+        assessment = {
+            "risk_label": "Verify first",
+            "simple_explanation": "Check independently.",
+            "red_flags": ["Unverified sender"],
+            "safe_next_steps": ["Use an official channel."],
+            "reply_draft": "Please confirm through your official channel.",
+        }
+
+        def fake_call(_text, _image, telemetry):
+            telemetry.update(
+                {
+                    "modal_called": True,
+                    "modal_ms": 120,
+                    "retry_count": 1,
+                    "parse_ms": 1,
+                    "normalize_ms": 1,
+                }
+            )
+            return assessment
+
+        with patch(
+            "app.model_status",
+            return_value={"connected": True, "label": "ready"},
+        ), patch("app.call_model", side_effect=fake_call) as model_mock, patch(
+            "app.queue_trace",
+            return_value=("trace-id", "queued"),
+        ) as queue_mock:
+            result = app.analyze_notice("test message")
+        self.assertTrue(result["ok"])
+        model_mock.assert_called_once()
+        self.assertTrue(queue_mock.call_args.kwargs["modal_called"])
+        self.assertEqual(queue_mock.call_args.kwargs["retry_count"], 1)
+
+    def test_timeout_is_sanitized(self) -> None:
+        timeout = APITimeoutError(request=httpx.Request("POST", "https://example.invalid"))
+        with patch(
+            "app.model_status",
+            return_value={"connected": True, "label": "ready"},
+        ), patch("app.call_model", side_effect=timeout), patch(
+            "app.queue_trace",
+            return_value=("trace-id", "queued"),
+        ) as queue_mock:
+            result = app.analyze_notice("test message")
+        self.assertFalse(result["ok"])
+        self.assertEqual(queue_mock.call_args.kwargs["failure_category"], "timeout")
+
+    def test_http_failure_is_sanitized(self) -> None:
+        request = httpx.Request("POST", "https://example.invalid")
+        error = APIStatusError(
+            "service unavailable",
+            response=httpx.Response(503, request=request),
+            body={"private": "must not be traced"},
+        )
+        with patch(
+            "app.model_status",
+            return_value={"connected": True, "label": "ready"},
+        ), patch("app.call_model", side_effect=error), patch(
+            "app.queue_trace",
+            return_value=("trace-id", "queued"),
+        ) as queue_mock:
+            result = app.analyze_notice("test message")
+        self.assertFalse(result["ok"])
+        self.assertEqual(queue_mock.call_args.kwargs["failure_category"], "http_error")
+        self.assertNotIn("private", json.dumps(queue_mock.call_args.kwargs))
+
+    def test_malformed_output_is_sanitized(self) -> None:
+        with patch(
+            "app.model_status",
+            return_value={"connected": True, "label": "ready"},
+        ), patch("app.call_model", side_effect=ValueError("PRIVATE RAW OUTPUT")), patch(
+            "app.queue_trace",
+            return_value=("trace-id", "queued"),
+        ) as queue_mock:
+            result = app.analyze_notice("test message")
+        self.assertFalse(result["ok"])
+        self.assertEqual(
+            queue_mock.call_args.kwargs["failure_category"],
+            "invalid_model_output",
+        )
+        self.assertNotIn("PRIVATE RAW OUTPUT", json.dumps(queue_mock.call_args.kwargs))
+
+    def test_normalization_failure_uses_normalize_stage(self) -> None:
+        telemetry: dict = {}
+        with self.assertRaises(ValueError):
+            app.parse_model_json('{"risk_label":"invalid"}', telemetry)
+        self.assertTrue(telemetry["parse_completed"])
+        self.assertNotIn("normalize_completed", telemetry)
+
+    def test_model_retry_is_counted_without_extra_trace_call(self) -> None:
+        request = httpx.Request("POST", "https://example.invalid")
+        unavailable = APIStatusError(
+            "unavailable",
+            response=httpx.Response(503, request=request),
+            body=None,
+        )
+
+        class Completions:
+            def __init__(self):
+                self.calls = 0
+
+            def create(self, **_kwargs):
+                self.calls += 1
+                if self.calls == 1:
+                    raise unavailable
+                message = type("Message", (), {"content": json.dumps({
+                    "risk_label": "Verify first",
+                    "simple_explanation": "Verify independently.",
+                    "red_flags": ["Unverified sender"],
+                    "safe_next_steps": ["Use an official channel."],
+                    "reply_draft": "Please confirm through an official channel.",
+                })})()
+                choice = type("Choice", (), {"message": message})()
+                return type("Completion", (), {"choices": [choice]})()
+
+        completions = Completions()
+        client = type(
+            "Client",
+            (),
+            {"chat": type("Chat", (), {"completions": completions})()},
+        )()
+        telemetry: dict = {}
+        with patch("app.create_model_client", return_value=(client, "model")), patch.dict(
+            "os.environ",
+            {"MODEL_MAX_ATTEMPTS": "2", "MODEL_RETRY_DELAY_SECONDS": "0"},
+        ):
+            result = app.call_model("test", "", telemetry)
+        self.assertEqual(result["risk_label"], "Verify first")
+        self.assertEqual(completions.calls, 2)
+        self.assertEqual(telemetry["retry_count"], 1)
+
+    def test_publisher_persists_batch(self) -> None:
+        publisher = trace_runtime.TracePublisher()
+        with tempfile.TemporaryDirectory() as directory, patch.object(
+            trace_runtime,
+            "PENDING_DIR",
+            Path(directory),
+        ):
+            records = [self.sample_record() for _ in range(20)]
+            publisher._persist_batch(records)
+            paths = list(Path(directory).glob("*.jsonl"))
+            self.assertEqual(len(paths), 1)
+            self.assertEqual(len(paths[0].read_text().splitlines()), 20)
+
+    def test_concurrent_enqueue_and_queue_limit(self) -> None:
+        with patch.object(trace_runtime, "MAX_QUEUE_SIZE", 20):
+            publisher = trace_runtime.TracePublisher()
+        with patch.object(publisher, "_ensure_worker"):
+            threads = [
+                threading.Thread(target=publisher.enqueue, args=(self.sample_record(),))
+                for _ in range(20)
+            ]
+            for thread in threads:
+                thread.start()
+            for thread in threads:
+                thread.join()
+            self.assertEqual(publisher.status()["queued"], 20)
+            self.assertEqual(publisher.queue.qsize(), 20)
+            self.assertEqual(publisher.enqueue(self.sample_record()), "dropped")
+            self.assertEqual(publisher.status()["dropped"], 1)
+
+    def test_hub_failure_keeps_pending_shard(self) -> None:
+        publisher = trace_runtime.TracePublisher()
+        with tempfile.TemporaryDirectory() as directory, patch.object(
+            trace_runtime,
+            "PENDING_DIR",
+            Path(directory),
+        ), patch.dict("os.environ", {"HF_TOKEN": "test-token"}), patch(
+            "huggingface_hub.HfApi.upload_file",
+            side_effect=RuntimeError("offline"),
+        ), patch("trace_runtime.time.sleep"):
+            publisher._persist_batch([self.sample_record()])
+            publisher._upload_pending()
+            self.assertEqual(len(list(Path(directory).glob("*.jsonl"))), 1)
+            self.assertEqual(publisher.status()["upload_failures"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

trace_runtime.py

@@ -0,0 +1,590 @@
+"""Fast, deterministic, privacy-safe pipeline tracing."""
+
+from __future__ import annotations
+
+import json
+import os
+import queue
+import re
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parent
+TRACE_ROOT = Path(os.getenv("TRACE_DIR", ROOT / "traces"))
+PENDING_DIR = TRACE_ROOT / "pending"
+DATASET_REPO = os.getenv(
+    "HF_TRACE_DATASET_REPO",
+    "build-small-hackathon/pakistan-notice-helper-traces",
+)
+SCHEMA_VERSION = "1.0"
+BATCH_SIZE = max(1, int(os.getenv("TRACE_BATCH_SIZE", "20")))
+FLUSH_SECONDS = max(1.0, float(os.getenv("TRACE_FLUSH_SECONDS", "60")))
+MAX_QUEUE_SIZE = max(1, int(os.getenv("TRACE_MAX_QUEUE_SIZE", "5000")))
+
+PIPELINE_STEPS = (
+    "receive",
+    "validate",
+    "cache_lookup",
+    "modal_request",
+    "parse_json",
+    "normalize_result",
+    "reply_filter",
+    "response",
+)
+RISK_LABELS = {
+    "Looks normal",
+    "Verify first",
+    "Suspicious",
+    "Likely scam",
+    "Inappropriate",
+    "none",
+}
+FAILURE_CATEGORIES = {
+    "none",
+    "validation_empty",
+    "credentials_missing",
+    "http_auth",
+    "http_error",
+    "connection_error",
+    "timeout",
+    "invalid_model_output",
+    "internal_error",
+}
+PIPELINE_STATUSES = {"completed", "skipped", "rejected", "failed", "hit", "miss"}
+REQUEST_SOURCES = {"user", "cached_modal_example"}
+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",
+    "account_threat": (
+        r"\b(?:account|sim|service|electricity)\b.{0,50}"
+        r"\b(?:block|blocked|suspend|closed|disconnect)\b"
+    ),
+}
+EXAMPLE_PROFILES = {
+    "text-courier": ("text", "courier", {"link", "urgency", "payment", "courier"}),
+    "text-fbr": ("text", "fbr", {"cnic", "credentials", "urgency", "refund_or_prize"}),
+    "text-bank": ("text", "bank", {"otp", "urgency", "account_threat"}),
+    "image-courier": ("image", "courier", {"link", "urgency", "courier"}),
+    "image-mobile": ("image", "marketplace", {"credentials"}),
+    "image-traffic": ("image", "traffic_challan", {"link", "urgency", "payment", "challan"}),
+}
+
+
+def _bucket_number(value: float, thresholds: tuple[tuple[float, str], ...]) -> str:
+    for maximum, label in thresholds:
+        if value <= maximum:
+            return label
+    return thresholds[-1][1]
+
+
+def duration_bucket(milliseconds: float) -> str:
+    return _bucket_number(
+        max(0.0, milliseconds),
+        (
+            (1, "0-1ms"),
+            (5, "2-5ms"),
+            (10, "6-10ms"),
+            (50, "11-50ms"),
+            (250, "51-250ms"),
+            (1000, "251-1000ms"),
+            (5000, "1-5s"),
+            (30000, "5-30s"),
+            (float("inf"), "30s+"),
+        ),
+    )
+
+
+def input_size_bucket(length: int) -> str:
+    return _bucket_number(
+        max(0, length),
+        (
+            (0, "empty"),
+            (160, "1-160"),
+            (500, "161-500"),
+            (2000, "501-2000"),
+            (6000, "2001-6000"),
+            (12000, "6001-12000"),
+            (float("inf"), "12000+"),
+        ),
+    )
+
+
+def image_size_bucket(data_url_length: int) -> str:
+    estimated_bytes = max(0, int(data_url_length * 0.75))
+    return _bucket_number(
+        estimated_bytes,
+        (
+            (0, "none"),
+            (100_000, "up-to-100KB"),
+            (500_000, "100-500KB"),
+            (2_000_000, "500KB-2MB"),
+            (8_000_000, "2-8MB"),
+            (float("inf"), "8MB+"),
+        ),
+    )
+
+
+def detect_signals(text: str, example_id: str = "") -> dict[str, bool]:
+    detected = {
+        name: bool(re.search(pattern, text or "", re.I | re.S))
+        for name, pattern in SIGNAL_PATTERNS.items()
+    }
+    profile = EXAMPLE_PROFILES.get(example_id)
+    if profile:
+        for name in profile[2]:
+            detected[name] = True
+    return detected
+
+
+def detect_category(text: str, signals: dict[str, bool], example_id: str = "") -> str:
+    profile = EXAMPLE_PROFILES.get(example_id)
+    if profile:
+        return profile[1]
+    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:
+        if any(term in lowered for term in terms):
+            return category
+    if signals["challan"]:
+        return "traffic_challan"
+    if signals["courier"]:
+        return "courier"
+    return "unknown"
+
+
+def detect_language_hint(text: str) -> str:
+    has_urdu = bool(re.search(r"[\u0600-\u06ff]", text or ""))
+    has_latin = bool(re.search(r"[A-Za-z]", text or ""))
+    roman_terms = bool(
+        re.search(
+            r"\b(?:aap|apka|apki|hai|hain|karo|karein|paisa|rupay|bhej|jaldi)\b",
+            text or "",
+            re.I,
+        )
+    )
+    if has_urdu and has_latin:
+        return "mixed_urdu_latin"
+    if has_urdu:
+        return "urdu_script"
+    if roman_terms:
+        return "roman_urdu"
+    if has_latin:
+        return "latin_script"
+    return "unknown"
+
+
+def safe_summary(category: str, signals: dict[str, bool], input_type: str) -> str:
+    category_labels = {
+        "fbr": "FBR-style",
+        "bank": "Bank-style",
+        "wallet": "Wallet-style",
+        "utility": "Utility-style",
+        "traffic_challan": "Traffic-challan-style",
+        "courier": "Courier-style",
+        "customs": "Customs-style",
+        "university": "Education-style",
+        "job": "Job-style",
+        "marketplace": "Marketplace-style",
+        "unknown": "Unclassified",
+    }
+    signal_labels = {
+        "otp": "OTP",
+        "cnic": "CNIC",
+        "credentials": "credential",
+        "link": "link",
+        "urgency": "urgency",
+        "payment": "payment",
+        "refund_or_prize": "refund-or-prize",
+        "courier": "courier",
+        "challan": "challan",
+        "account_threat": "account-threat",
+    }
+    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"
+    return f"{category_labels[category]} {input_type} input{suffix}"
+
+
+def build_input_profile(text: str, image_data_url: str, example_id: str = "") -> dict[str, Any]:
+    profile = EXAMPLE_PROFILES.get(example_id)
+    if profile:
+        input_type = profile[0]
+    elif image_data_url and text:
+        input_type = "text_and_image"
+    elif image_data_url:
+        input_type = "image"
+    else:
+        input_type = "text"
+    signals = detect_signals(text, example_id)
+    category = detect_category(text, signals, example_id)
+    return {
+        "type": input_type,
+        "text_character_bucket": input_size_bucket(len(text or "")),
+        "text_byte_bucket": input_size_bucket(len((text or "").encode("utf-8"))),
+        "image_size_bucket": image_size_bucket(len(image_data_url or "")),
+        "category": category,
+        "language_hint": detect_language_hint(text),
+        "signals": signals,
+        "safe_summary": safe_summary(category, signals, input_type),
+    }
+
+
+def build_trace_record(
+    *,
+    text: str,
+    image_data_url: str,
+    example_id: str,
+    request_source: str,
+    pipeline_status: dict[str, str],
+    pipeline_ms: dict[str, float],
+    modal_called: bool,
+    modal_ms: float,
+    retry_count: int,
+    assessment: dict[str, Any] | None,
+    failure_category: str = "none",
+    failure_stage: str = "none",
+) -> dict[str, Any]:
+    trace_id = str(uuid.uuid4())
+    risk_label = str((assessment or {}).get("risk_label", "none"))
+    if risk_label not in RISK_LABELS:
+        risk_label = "none"
+    commit = (
+        os.getenv("SPACE_COMMIT")
+        or os.getenv("GIT_COMMIT")
+        or os.getenv("COMMIT_SHA")
+        or ""
+    )
+    commit = commit[:64] if re.fullmatch(r"[0-9a-fA-F]{7,64}", commit) else "unknown"
+    request_source = (
+        request_source if request_source in REQUEST_SOURCES else "user"
+    )
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "trace_id": trace_id,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "app_commit": commit,
+        "request_source": request_source,
+        "input": build_input_profile(text, image_data_url, example_id),
+        "pipeline_steps": [
+            {
+                "step": step,
+                "status": (
+                    pipeline_status.get(step, "skipped")
+                    if pipeline_status.get(step, "skipped") in PIPELINE_STATUSES
+                    else "skipped"
+                ),
+                "duration_bucket": duration_bucket(pipeline_ms.get(step, 0.0)),
+            }
+            for step in PIPELINE_STEPS
+        ],
+        "cache": {
+            "hit": request_source == "cached_modal_example",
+            "example_id": example_id if example_id in EXAMPLE_PROFILES else "none",
+        },
+        "modal": {
+            "called": bool(modal_called),
+            "model_family": (
+                "qwen3.6-27b-mtp"
+                if "qwen3.6-27b-mtp"
+                in os.getenv("MODEL_NAME", "qwen3.6-27b-mtp").lower()
+                else "other"
+            ),
+            "latency_bucket": duration_bucket(modal_ms),
+            "retry_count": max(0, min(int(retry_count), 20)),
+            "outcome": (
+                "success"
+                if modal_called and failure_category == "none"
+                else "failed"
+                if modal_called
+                else "not_called"
+            ),
+        },
+        "result": {
+            "risk_label": risk_label,
+            "red_flag_count": min(len((assessment or {}).get("red_flags", [])), 50),
+            "safe_next_step_count": min(
+                len((assessment or {}).get("safe_next_steps", [])),
+                50,
+            ),
+            "reply_draft_returned": bool((assessment or {}).get("reply_draft")),
+            "reply_draft_policy": (
+                "allowed"
+                if risk_label in {"Verify first", "Suspicious"}
+                else "suppressed"
+                if risk_label != "none"
+                else "not_applicable"
+            ),
+        },
+        "failure": {
+            "category": (
+                failure_category
+                if failure_category in FAILURE_CATEGORIES
+                else "internal_error"
+            ),
+            "stage": failure_stage if failure_stage in {*PIPELINE_STEPS, "none"} else "response",
+        },
+        "privacy": {
+            "raw_input_stored": False,
+            "raw_image_stored": False,
+            "raw_model_output_stored": False,
+            "exception_text_stored": False,
+            "identifiers_stored": False,
+        },
+    }
+
+
+def validate_trace(record: Any) -> list[str]:
+    errors: list[str] = []
+    if not isinstance(record, dict):
+        return ["Trace must be an object."]
+    required = {
+        "schema_version",
+        "trace_id",
+        "timestamp",
+        "app_commit",
+        "request_source",
+        "input",
+        "pipeline_steps",
+        "cache",
+        "modal",
+        "result",
+        "failure",
+        "privacy",
+    }
+    missing = required - record.keys()
+    if missing:
+        errors.append("Missing fields: " + ", ".join(sorted(missing)))
+    if record.get("schema_version") != SCHEMA_VERSION:
+        errors.append("Unsupported schema_version.")
+    if record.get("result", {}).get("risk_label") not in RISK_LABELS:
+        errors.append("Invalid risk label.")
+    privacy = record.get("privacy", {})
+    if not isinstance(privacy, dict) or any(privacy.get(key) is not False for key in privacy):
+        errors.append("Privacy flags must all be false.")
+    steps = record.get("pipeline_steps", [])
+    if not isinstance(steps, list) or [item.get("step") for item in steps] != list(
+        PIPELINE_STEPS
+    ):
+        errors.append("Pipeline steps are invalid or out of order.")
+    forbidden_keys = {
+        "raw_input",
+        "raw_text",
+        "image_data_url",
+        "raw_model_output",
+        "reply_draft",
+        "simple_explanation",
+        "error",
+        "exception",
+        "url",
+        "phone",
+        "account_number",
+    }
+
+    def walk(value: Any) -> None:
+        if isinstance(value, dict):
+            for key, child in value.items():
+                if key.lower() in forbidden_keys:
+                    errors.append(f"Forbidden field: {key}")
+                walk(child)
+        elif isinstance(value, list):
+            for child in value:
+                walk(child)
+
+    walk(record)
+    return sorted(set(errors))
+
+
+class TracePublisher:
+    def __init__(self) -> None:
+        self.queue: queue.Queue[dict[str, Any]] = queue.Queue(MAX_QUEUE_SIZE)
+        self.lock = threading.Lock()
+        self.thread: threading.Thread | None = None
+        self.counters = {
+            "queued": 0,
+            "persisted": 0,
+            "uploaded": 0,
+            "upload_failures": 0,
+            "dropped": 0,
+        }
+
+    def enqueue(self, record: dict[str, Any]) -> str:
+        if validate_trace(record):
+            with self.lock:
+                self.counters["dropped"] += 1
+            return "invalid"
+        try:
+            self.queue.put_nowait(record)
+        except queue.Full:
+            with self.lock:
+                self.counters["dropped"] += 1
+            return "dropped"
+        with self.lock:
+            self.counters["queued"] += 1
+        self._ensure_worker()
+        return "queued"
+
+    def status(self) -> dict[str, Any]:
+        with self.lock:
+            counters = dict(self.counters)
+        counters.update(
+            {
+                "queue_size": self.queue.qsize(),
+                "pending_shards": len(list(PENDING_DIR.glob("*.jsonl")))
+                if PENDING_DIR.exists()
+                else 0,
+                "dataset_repo": DATASET_REPO,
+            }
+        )
+        return counters
+
+    def _ensure_worker(self) -> None:
+        with self.lock:
+            if self.thread and self.thread.is_alive():
+                return
+            self.thread = threading.Thread(
+                target=self._worker,
+                name="privacy-safe-trace-publisher",
+                daemon=True,
+            )
+            self.thread.start()
+
+    def _worker(self) -> None:
+        batch: list[dict[str, Any]] = []
+        deadline: float | None = None
+        self._upload_pending()
+        while True:
+            timeout = (
+                max(0.05, deadline - time.monotonic())
+                if deadline is not None
+                else FLUSH_SECONDS
+            )
+            try:
+                record = self.queue.get(timeout=timeout)
+                batch.append(record)
+                if deadline is None:
+                    deadline = time.monotonic() + FLUSH_SECONDS
+            except queue.Empty:
+                pass
+            if batch and (
+                len(batch) >= BATCH_SIZE
+                or (deadline is not None and time.monotonic() >= deadline)
+            ):
+                self._persist_batch(batch)
+                batch = []
+                deadline = None
+                self._upload_pending()
+
+    def _persist_batch(self, records: list[dict[str, Any]]) -> None:
+        PENDING_DIR.mkdir(parents=True, exist_ok=True)
+        pending_count = self._pending_record_count()
+        capacity = max(0, MAX_QUEUE_SIZE - pending_count)
+        if capacity == 0:
+            with self.lock:
+                self.counters["dropped"] += len(records)
+            return
+        accepted = records[:capacity]
+        dropped = len(records) - len(accepted)
+        if dropped:
+            with self.lock:
+                self.counters["dropped"] += dropped
+        timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
+        filename = f"trace-{timestamp}-{uuid.uuid4().hex[:8]}.jsonl"
+        final_path = PENDING_DIR / filename
+        temporary_path = final_path.with_suffix(".tmp")
+        content = "".join(
+            json.dumps(record, sort_keys=True, ensure_ascii=True) + "\n"
+            for record in accepted
+        )
+        temporary_path.write_text(content, encoding="utf-8")
+        os.replace(temporary_path, final_path)
+        with self.lock:
+            self.counters["persisted"] += len(accepted)
+
+    def _pending_record_count(self) -> int:
+        if not PENDING_DIR.exists():
+            return 0
+        count = 0
+        for path in PENDING_DIR.glob("*.jsonl"):
+            try:
+                count += sum(
+                    1
+                    for line in path.read_text(encoding="utf-8").splitlines()
+                    if line
+                )
+            except OSError:
+                continue
+        return count
+
+    def _upload_pending(self) -> None:
+        token = os.getenv("HF_TOKEN", "").strip()
+        if not token or not PENDING_DIR.exists():
+            return
+        try:
+            from huggingface_hub import HfApi
+
+            api = HfApi(token=token)
+            for path in sorted(PENDING_DIR.glob("*.jsonl")):
+                date_path = datetime.now(timezone.utc).strftime("%Y/%m/%d")
+                uploaded = False
+                for attempt in range(3):
+                    try:
+                        api.upload_file(
+                            path_or_fileobj=str(path),
+                            path_in_repo=f"data/{date_path}/{path.name}",
+                            repo_id=DATASET_REPO,
+                            repo_type="dataset",
+                            commit_message=f"Add privacy-safe trace shard {path.name}",
+                        )
+                        uploaded = True
+                        break
+                    except Exception:
+                        if attempt < 2:
+                            time.sleep(2**attempt)
+                if not uploaded:
+                    with self.lock:
+                        self.counters["upload_failures"] += 1
+                    return
+                count = sum(1 for line in path.read_text(encoding="utf-8").splitlines() if line)
+                path.unlink(missing_ok=True)
+                with self.lock:
+                    self.counters["uploaded"] += count
+        except Exception:
+            with self.lock:
+                self.counters["upload_failures"] += 1
+
+
+PUBLISHER = TracePublisher()
+
+
+def start_trace_worker() -> None:
+    PUBLISHER._ensure_worker()
+
+
+def queue_trace(**kwargs: Any) -> tuple[str, str]:
+    record = build_trace_record(**kwargs)
+    return record["trace_id"], PUBLISHER.enqueue(record)
+
+
+def trace_status() -> dict[str, Any]:
+    return PUBLISHER.status()