Deploy grading app

.env.example

@@ -0,0 +1,16 @@
+# Local grading data root (skips Hugging Face for templates + user answers).
+# Layout matches the HF dataset layout:
+#   {GRADING_DATA_DIR}/{jurisdiction}/grading_template.parquet
+#   {GRADING_DATA_DIR}/{jurisdiction}/users/users.json
+#   {GRADING_DATA_DIR}/{jurisdiction}/users/{username}_answers.parquet
+GRADING_DATA_DIR=./data
+
+# Optional: single-template shortcut when you only have one parquet file.
+# Takes precedence over the jurisdiction file under GRADING_DATA_DIR for template load.
+# Jurisdiction list still comes from GRADING_DATA_DIR subfolders, or from GRADING_JURISDICTION.
+# GRADING_TEMPLATE_PATH=./data/en-us/grading_template.parquet
+# GRADING_JURISDICTION=en-us
+
+# Hugging Face mode (used when GRADING_DATA_DIR / GRADING_TEMPLATE_PATH are unset)
+# HF_DATASET_REPO=TransLegal/grading-answers
+# HF_TOKEN=hf_...

.gitignore

@@ -0,0 +1 @@
+data/_downloads/

README.md

@@ -80,13 +80,14 @@ To add a new jurisdiction to the repository:
      - `concept` (string) - The concept class (drives which questions apply)
      - `section` (string) - The section a question belongs to (one page per section)
      - `section_index` (integer) - Display order of sections within a result
-     - `question_id` (string) - Unique question id, e.g. `A2-Q13a`
+     - `question_id` (string) - Unique question id, e.g. `A2-Q01`
      - `group` (string) - Groups split questions that share a source field, e.g. `A2-Q13`
      - `part` (string) - Split marker (`a`, `b`, …) or empty
      - `question` (string) - The PASS/FAIL check text
      - `question_index` (integer) - Order of questions within a section for a result
-     - `answer` (string) - Full model output for the Analytical Plane block this check covers
-     - `source_fields` (string) - Comma-joined structured-analysis fields in that block
+     - `answer` (string) - Generated text for the AP prompt(s) this check covers, and nothing else from their block
+     - `source_fields` (string) - Comma-joined structured-analysis fields for those prompts
+     - `source_prompts` (string) - Semicolon-joined AP prompt ids (e.g. `A1.3-P01`) supplying the evidence
    - **Display Order:** swipe deck is **concept → section (`section_index`) → term → questions (`question_index`)**. Within a section, all checks for one term are asked before the next term.
 
 3. **Create users directory:**

WORKFLOW.md

@@ -0,0 +1,203 @@
+# Grading app — end-to-end workflow
+
+How to stand up a grading round from scratch: author the questions, run the
+pipeline, pull the results, build the template, and launch the swipe app.
+Follow this when the **real questions** and **real results** are ready.
+
+All commands run from the **TransLegal repo root** unless noted, using `uv`.
+
+---
+
+## 0. Mental model
+
+```
+question_specs.iter_question_specs     structured_analysis run (HF)
+ (PASS/FAIL checks per concept class)      (term × field_of_law × answers)
+                     \                    /
+                      \                  /
+                 build_grading_template.py   ──►  data/{jurisdiction}/grading_template.parquet
+                                                    (one row per result × question, answer attached)
+                                                                 │
+                                                                 ▼
+                                                   Streamlit swipe app (Pass/Fail)
+                                                                 │
+                                                                 ▼
+                                       data/{jurisdiction}/users/{username}_answers.parquet
+                                                    (the expensive human verdicts)
+```
+
+Key idea: **questions are concept-specific.** Every result only ever shows the
+questions for *its* concept class. This is enforced by the builder once
+`iter_question_specs` is wired.
+
+---
+
+## 1. Author / update the questions
+
+**Analysis prompts** (Stage A) live in `concepts/analytical_plane/blocks/*.py`
+and `bundles.py`. Those are independent of grading.
+
+**PASS/FAIL grading questions** live in `pipeline/llm_judge/questions.tsv` and
+are exposed via `pipeline.llm_judge.question_specs.iter_question_specs`.
+
+---
+
+## 2. Produce the results (pipeline run)
+
+Grading needs a `structured_analysis` parquet: one row per
+`(term, field_of_law)` with `concept` and `structured_answers_json`.
+
+- These are produced by the Dagster `structured_analysis` asset and published to
+  the private HF dataset **`TransLegal/structured_analysis`**, under
+  `{experiment}/{variant}/{jurisdiction}/{timestamp}.parquet` on a **branch**
+  (e.g. `main`, `showcase`).
+- For the real round, run the pipeline for your chosen term set (e.g. 160 terms,
+  20 per concept class) and note the branch + path it wrote to.
+
+You need a HF token that is a **member of the `TransLegal` org** (read access).
+Set it in `.env` as `HUGGING_FACE_HUB_TOKEN_TRANSLEGAL`. Verify:
+
+```bash
+uv run hf auth whoami        # 'TransLegal' must appear under orgs
+```
+
+---
+
+## 3. Ingest results into the grading app
+
+Preferred path: **`ingest_structured_analysis.py`** downloads the latest
+`structured_analysis` parquet from HF and builds
+`data/<jurisdiction>/grading_template.parquet` in one step.
+
+### Tiny development run
+
+Config:
+`orchestration/translegal_dagster/run_configs/definition_pipeline/new-llm-judge-tiny-example.yaml`
+→ publishes to branch **`dev`**, path
+`new-llm-judge/tiny-example/en_us/<timestamp>.parquet`.
+
+Once the run has published (or while it is still running):
+
+```bash
+# One-shot after publish:
+uv run python hf-spaces/grading-answers/ingest_structured_analysis.py
+
+# Or poll until the parquet appears (checks every 30s, up to 1h):
+uv run python hf-spaces/grading-answers/ingest_structured_analysis.py --poll
+
+# Or build from a local parquet you already have:
+uv run python hf-spaces/grading-answers/ingest_structured_analysis.py \
+  --input /path/to/structured_analysis.parquet
+```
+
+Defaults: `--branch dev --experiment new-llm-judge --variant tiny-example
+--jurisdiction en-us`, output
+`hf-spaces/grading-answers/data/en-us/grading_template.parquet`.
+
+Restart the Streamlit app after ingest so `@st.cache_data` reloads the template.
+
+### Manual download + build (any run)
+
+List a branch to find the exact path (paths use `en_us`, not `en-us`):
+
+```bash
+uv run python -c "from huggingface_hub import HfApi; \
+print('\n'.join(HfApi().list_repo_files('TransLegal/structured_analysis', repo_type='dataset', revision='<branch>')))"
+
+uv run hf download TransLegal/structured_analysis \
+  "<experiment>/<variant>/<jurisdiction>/<timestamp>.parquet" \
+  --repo-type dataset --revision <branch> --local-dir /tmp/run
+```
+
+Then:
+
+```bash
+uv run python hf-spaces/grading-answers/build_grading_template.py \
+  --input /tmp/run/<...>.parquet \
+  --output hf-spaces/grading-answers/data/<jurisdiction>/grading_template.parquet
+```
+
+Selection flags (same on `ingest_structured_analysis.py` and
+`build_grading_template.py`):
+
+| Flag | Effect |
+|------|--------|
+| *(default)* | **One representative field of law per term** (first alphabetically). Use for the real "one field of law per term" round. |
+| `--field-of-law "Contract Law"` | Pin a single field of law for every term. |
+| `--all-fields-of-law` | Keep **every** `(term, field_of_law)` result. Use for the showcase set (same term across many fields). |
+| `--max-terms N` | Cap number of terms (smoke tests). |
+| `--include-failed` | Include rows whose `generation_status != ok` (off by default). |
+
+Output columns: `term, field_of_law, concept, section, section_index,
+question_id, group, part, question, question_index, answer, source_fields,
+source_prompts`.
+
+> Note the app data folder uses the hyphenated jurisdiction (`en-us`), even
+> though the pipeline path uses `en_us`.
+
+---
+
+## 4. Run the app (local mode)
+
+```bash
+cd hf-spaces/grading-answers
+GRADING_DATA_DIR=./data \
+uv run --with streamlit --with streamlit-shortcuts --with pandas --with pyarrow \
+  --with huggingface-hub --with pycountry --with babel --with python-dotenv \
+  streamlit run src/streamlit_app.py --server.port 8502
+```
+
+Or put `GRADING_DATA_DIR=./data` in `hf-spaces/grading-answers/.env` and just run
+`streamlit run src/streamlit_app.py`.
+
+In the app: pick the jurisdiction → register a user → grade. Swipe deck controls:
+**→** pass, **←** fail, **↓** skip, **↑** undo (or the buttons). Verdicts autosave
+and resume where you left off.
+
+Verdicts land in `data/<jurisdiction>/users/<username>_answers.parquet`
+(template columns + `verdict` + `time_stamp`). **These are the expensive human
+outputs — never overwrite this folder when you refresh a template.**
+
+---
+
+## 5. Deploy to Hugging Face (shared graders)
+
+The app is a Docker Space (`TransLegal/grading-answers`). In HF mode it reads/writes
+the private dataset `TransLegal/grading-answers`:
+
+1. Upload only the built `grading_template.parquet` (+ `grading_template.meta.yaml`)
+   to `<jurisdiction>/grading_template.parquet` — **leave `<jurisdiction>/users/`
+   untouched.**
+2. Ensure the Space has `HF_TOKEN` (write) and `HF_DATASET_REPO` set (see README).
+3. `git push` the Space repo to redeploy.
+
+Graders then register/sign in per jurisdiction; their verdicts are written to
+`<jurisdiction>/users/<username>_answers.parquet` in the dataset.
+
+---
+
+## 6. Refreshing a round (real questions / new results)
+
+1. Update questions in `pipeline/llm_judge/questions.tsv` (step 1).
+2. Run the pipeline for the real term set → note branch/path (step 2).
+3. Re-ingest with `ingest_structured_analysis.py` (or download + build) (step 3).
+4. Replace `grading_template.parquet` (+ meta) in `data/<jurisdiction>/` (local) or
+   in the HF dataset (step 5). **Do not touch `users/`.**
+5. Launch/redeploy (steps 4–5).
+
+---
+
+## File map
+
+| Path | Role |
+|------|------|
+| `concepts/analytical_plane/blocks/*.py` | Stage A prompt blocks (docstrings) |
+| `pipeline/llm_judge/questions.tsv` | PASS/FAIL question bank |
+| `pipeline/llm_judge/question_specs.py` | `iter_question_specs()` |
+| `hf-spaces/grading-answers/build_grading_template.py` | Results × questions → app template (+ meta sidecar) |
+| `hf-spaces/grading-answers/ingest_structured_analysis.py` | HF download + template build (tiny-example defaults) |
+| `hf-spaces/grading-answers/data/<j>/grading_template.parquet` | Local template the app loads |
+| `hf-spaces/grading-answers/data/<j>/users/<user>_answers.parquet` | Human verdicts (precious) |
+| `hf-spaces/grading-answers/src/streamlit_app.py` | The swipe grading app |
+| `TransLegal/structured_analysis` (HF) | Source pipeline results |
+| `TransLegal/grading-answers` (HF) | Deployed app storage (templates + verdicts) |

build_grading_template.py

@@ -0,0 +1,440 @@
+#!/usr/bin/env python3
+"""Build a PASS/FAIL question grading template from structured_analysis output.
+
+Reads Dagster ``structured_analysis`` output (term, field_of_law,
+concept_class / concept, structured_answers_json) and, for each graded
+result, emits one row per PASS/FAIL question from
+``pipeline.llm_judge.question_specs``.
+
+Question specs are loaded from ``pipeline/llm_judge/questions.tsv``.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+import yaml
+
+# Allow running from repo root; concepts lives at TransLegal/concepts.
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+
+from pipeline.llm_judge.answers import answer_for_spec  # noqa: E402
+from pipeline.llm_judge.question_specs import (  # noqa: E402
+    QUESTION_BANK_VERSION,
+    iter_question_specs,
+)
+
+logger = logging.getLogger(__name__)
+
+SCHEMA_VERSION = 3
+
+REQUIRED_INPUT_COLUMNS = (
+    "term",
+    "field_of_law",
+    "structured_answers_json",
+)
+# Pipeline output uses ``concept_class``; accept ``concept`` as an alias.
+_CONCEPT_COLUMN_CANDIDATES = ("concept_class", "concept")
+
+OUTPUT_COLUMNS = (
+    "term",
+    "field_of_law",
+    "concept",
+    "section",
+    "section_index",
+    "question_id",
+    "group",
+    "part",
+    "question",
+    "question_index",
+    "answer",
+    "source_fields",
+    "source_prompts",
+)
+
+
+def flatten_result_row(
+    *,
+    term: str,
+    field_of_law: str,
+    concept: str,
+    structured_answers: dict[str, Any],
+) -> list[dict[str, Any]]:
+    """Expand one (term, field_of_law) result into per-question grading rows."""
+    rows: list[dict[str, Any]] = []
+    for spec in iter_question_specs(concept):
+        rows.append(
+            {
+                "term": term,
+                "field_of_law": field_of_law,
+                "concept": concept,
+                "section": spec["section"],
+                "section_index": spec["section_index"],
+                "question_id": spec["id"],
+                "group": spec["group"],
+                "part": spec["part"],
+                "question": spec["question"],
+                "question_index": spec["question_index"],
+                "answer": answer_for_spec(spec, structured_answers),
+                "source_fields": ",".join(spec["source_fields"]),
+                "source_prompts": ";".join(spec["source_prompts"]),
+            }
+        )
+    return rows
+
+
+def select_results(
+    df: pd.DataFrame,
+    *,
+    field_of_law: str | None,
+    one_per_term: bool,
+) -> pd.DataFrame:
+    """Reduce a multi-field_of_law run to the results to grade.
+
+    - ``field_of_law`` pins a single field of law (kept only where present).
+    - ``one_per_term`` keeps one representative result per term (the first
+      field of law alphabetically), which yields one result per term.
+    - otherwise every (term, field_of_law) result is kept.
+    """
+    if field_of_law is not None:
+        return df[df["field_of_law"].astype(str).str.strip() == field_of_law].copy()
+    if not one_per_term:
+        return df.copy()
+    ordered = df.sort_values(by=["term", "field_of_law"], kind="mergesort")
+    return ordered.drop_duplicates(subset=["term"], keep="first").copy()
+
+
+def _resolve_concept_column(columns: pd.Index) -> str:
+    for name in _CONCEPT_COLUMN_CANDIDATES:
+        if name in columns:
+            return name
+    raise ValueError(
+        "structured_analysis is missing a concept column "
+        f"(expected one of {_CONCEPT_COLUMN_CANDIDATES}). "
+        f"Found: {list(columns)}"
+    )
+
+
+def build_grading_template(
+    structured_analysis: pd.DataFrame,
+    *,
+    only_ok: bool = True,
+    field_of_law: str | None = None,
+    one_per_term: bool = True,
+    max_terms: int | None = None,
+) -> pd.DataFrame:
+    """Flatten a structured_analysis DataFrame into per-question grading rows."""
+    missing = [
+        c for c in REQUIRED_INPUT_COLUMNS if c not in structured_analysis.columns
+    ]
+    if missing:
+        raise ValueError(
+            f"structured_analysis is missing required columns: {missing}. "
+            f"Found: {list(structured_analysis.columns)}"
+        )
+    concept_col = _resolve_concept_column(structured_analysis.columns)
+
+    df = structured_analysis.copy()
+    if only_ok and "generation_status" in df.columns:
+        before = len(df)
+        df = df[df["generation_status"] == "ok"]
+        dropped = before - len(df)
+        if dropped:
+            logger.info("Dropped %d rows with generation_status != ok", dropped)
+
+    df = select_results(df, field_of_law=field_of_law, one_per_term=one_per_term)
+
+    if max_terms is not None:
+        keep_terms = (
+            df.sort_values(by=["term"], kind="mergesort")["term"]
+            .drop_duplicates()
+            .head(max_terms)
+        )
+        df = df[df["term"].isin(set(keep_terms))]
+
+    all_rows: list[dict[str, Any]] = []
+    skipped = 0
+    for _, row in df.iterrows():
+        term = str(row["term"] or "").strip()
+        field = str(row["field_of_law"] or "").strip()
+        concept = str(row[concept_col] or "").strip()
+        raw_json = str(row["structured_answers_json"] or "").strip()
+
+        if not term or not concept or not raw_json or raw_json == "{}":
+            skipped += 1
+            logger.warning(
+                "Skipping incomplete row term=%r field_of_law=%r concept=%r",
+                term,
+                field,
+                concept,
+            )
+            continue
+
+        try:
+            parsed = json.loads(raw_json)
+        except json.JSONDecodeError:
+            skipped += 1
+            logger.exception("Invalid JSON for term=%r field_of_law=%r", term, field)
+            continue
+
+        if not isinstance(parsed, dict):
+            skipped += 1
+            logger.warning(
+                "Expected object JSON for term=%r field_of_law=%r, got %s",
+                term,
+                field,
+                type(parsed).__name__,
+            )
+            continue
+
+        try:
+            all_rows.extend(
+                flatten_result_row(
+                    term=term,
+                    field_of_law=field,
+                    concept=concept,
+                    structured_answers=parsed,
+                )
+            )
+        except (ValueError, KeyError):
+            skipped += 1
+            logger.exception(
+                "Cannot flatten term=%r field_of_law=%r concept=%r",
+                term,
+                field,
+                concept,
+            )
+
+    if skipped:
+        logger.info("Skipped %d input rows", skipped)
+
+    out = pd.DataFrame(all_rows, columns=list(OUTPUT_COLUMNS))
+    if out.empty:
+        return out
+
+    return out.sort_values(
+        by=["concept", "section_index", "term", "field_of_law", "question_index"],
+        kind="mergesort",
+    ).reset_index(drop=True)
+
+
+def _relative_to_repo(path: Path) -> str:
+    try:
+        return str(path.resolve().relative_to(_REPO_ROOT))
+    except ValueError:
+        return str(path.resolve())
+
+
+def _load_dagster_sidecar(input_parquet: Path) -> dict[str, Any]:
+    """Load the Dagster tabular sidecar yaml next to a structured_analysis parquet."""
+    sidecar = input_parquet.with_suffix(".yaml")
+    if not sidecar.is_file():
+        return {}
+    try:
+        with sidecar.open(encoding="utf-8") as f:
+            data = yaml.safe_load(f)
+        return data if isinstance(data, dict) else {}
+    except Exception:
+        logger.exception("Could not read Dagster sidecar %s", sidecar)
+        return {}
+
+
+def build_template_metadata(
+    *,
+    input_parquet: Path,
+    output_parquet: Path,
+    template: pd.DataFrame,
+    structured_input: pd.DataFrame,
+    only_ok: bool,
+    field_of_law: str | None,
+    one_per_term: bool,
+    max_terms: int | None,
+) -> dict[str, Any]:
+    """Build provenance metadata for a grading_template.parquet artifact."""
+    sidecar = _load_dagster_sidecar(input_parquet)
+    op_config = (
+        sidecar.get("op_config") if isinstance(sidecar.get("op_config"), dict) else {}
+    )
+
+    results = (
+        int(template[["term", "field_of_law"]].drop_duplicates().shape[0])
+        if not template.empty
+        else 0
+    )
+
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "question_bank_version": QUESTION_BANK_VERSION,
+        "artifact_type": "grading_template",
+        "built_at": datetime.now(UTC).replace(microsecond=0).isoformat(),
+        "builder": {
+            "script": _relative_to_repo(Path(__file__)),
+        },
+        "output": {
+            "parquet": _relative_to_repo(output_parquet),
+            "meta": _relative_to_repo(
+                output_parquet.with_name("grading_template.meta.yaml")
+            ),
+        },
+        "source": {
+            "structured_analysis_parquet": _relative_to_repo(input_parquet),
+            "dagster_sidecar": _relative_to_repo(input_parquet.with_suffix(".yaml"))
+            if input_parquet.with_suffix(".yaml").is_file()
+            else None,
+            "run_id": sidecar.get("run_id"),
+            "run_ts": sidecar.get("run_ts"),
+            "git_sha": sidecar.get("git_sha"),
+            "jurisdiction": sidecar.get("jurisdiction"),
+            "dataset_branch": sidecar.get("dataset_branch"),
+            "experiment_name": sidecar.get("experiment_name"),
+            "experiment_variant": sidecar.get("experiment_variant"),
+            "run_comment": sidecar.get("run_comment"),
+            "llm_provider": op_config.get("llm_provider"),
+            "llm_model": op_config.get("llm_model"),
+            "publish_to_hf": sidecar.get("publish_to_hf"),
+        },
+        "build": {
+            "only_ok": only_ok,
+            "field_of_law": field_of_law,
+            "one_per_term": one_per_term,
+            "max_terms": max_terms,
+            "input_rows": int(len(structured_input)),
+            "output_rows": int(len(template)),
+            "result_count": results,
+            "term_count": int(template["term"].nunique()) if not template.empty else 0,
+            "field_of_law_count": (
+                int(template["field_of_law"].nunique()) if not template.empty else 0
+            ),
+            "question_count": (
+                int(template["question_id"].nunique()) if not template.empty else 0
+            ),
+            "concepts": sorted(template["concept"].unique().tolist())
+            if not template.empty
+            else [],
+            "columns": list(template.columns),
+        },
+    }
+
+
+def write_template_metadata(metadata: dict[str, Any], output_parquet: Path) -> Path:
+    """Write grading_template.meta.yaml beside grading_template.parquet."""
+    meta_path = output_parquet.with_name("grading_template.meta.yaml")
+    meta_path.parent.mkdir(parents=True, exist_ok=True)
+    with meta_path.open("w", encoding="utf-8") as f:
+        yaml.safe_dump(metadata, f, sort_keys=False, allow_unicode=True)
+    return meta_path
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Build a grading_template.parquet from structured_analysis output "
+            "(Analytical Plane bundles)."
+        )
+    )
+    parser.add_argument(
+        "--input",
+        type=Path,
+        required=True,
+        help="Path to structured_analysis parquet (Dagster tabular output).",
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        required=True,
+        help="Path to write grading_template.parquet.",
+    )
+    parser.add_argument(
+        "--max-terms",
+        type=int,
+        default=None,
+        help="Optional cap on the number of terms for smoke tests.",
+    )
+    parser.add_argument(
+        "--field-of-law",
+        type=str,
+        default=None,
+        help="Pin a single field of law to grade (exact match).",
+    )
+    parser.add_argument(
+        "--all-fields-of-law",
+        action="store_true",
+        help=(
+            "Grade every (term, field_of_law) result. Default keeps one "
+            "representative result per term."
+        ),
+    )
+    parser.add_argument(
+        "--include-failed",
+        action="store_true",
+        help="Include rows whose generation_status is not 'ok'.",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    args = _parse_args()
+    if not args.input.exists():
+        raise FileNotFoundError(f"Input parquet not found: {args.input}")
+
+    logger.info("Loading %s", args.input)
+    structured = pd.read_parquet(args.input)
+    one_per_term = not args.all_fields_of_law and args.field_of_law is None
+    template = build_grading_template(
+        structured,
+        only_ok=not args.include_failed,
+        field_of_law=args.field_of_law,
+        one_per_term=one_per_term,
+        max_terms=args.max_terms,
+    )
+
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    template.to_parquet(args.output, index=False)
+
+    metadata = build_template_metadata(
+        input_parquet=args.input,
+        output_parquet=args.output,
+        template=template,
+        structured_input=structured,
+        only_ok=not args.include_failed,
+        field_of_law=args.field_of_law,
+        one_per_term=one_per_term,
+        max_terms=args.max_terms,
+    )
+    meta_path = write_template_metadata(metadata, args.output)
+
+    n_results = (
+        template[["term", "field_of_law"]].drop_duplicates().shape[0]
+        if not template.empty
+        else 0
+    )
+    logger.info(
+        "Wrote %d grading rows for %d results (term, field_of_law) → %s",
+        len(template),
+        n_results,
+        args.output,
+    )
+    logger.info("Wrote metadata sidecar → %s", meta_path)
+    if not template.empty:
+        logger.info("Concepts: %s", sorted(template["concept"].unique()))
+        logger.info(
+            "Columns: %s",
+            list(template.columns),
+        )
+
+
+if __name__ == "__main__":
+    main()

ingest_structured_analysis.py

@@ -0,0 +1,366 @@
+#!/usr/bin/env python3
+"""Download a structured_analysis parquet and build the local grading template.
+
+Default target is the tiny development run::
+
+    uv run python hf-spaces/grading-answers/ingest_structured_analysis.py
+
+That pulls the latest ``new-llm-judge/tiny-example/en_us/*.parquet`` from
+``TransLegal/structured_analysis`` (branch ``dev``) and writes::
+
+    hf-spaces/grading-answers/data/en-us/grading_template.parquet
+
+Pass ``--input`` to skip Hugging Face and build from a local parquet instead.
+Use ``--poll`` while a Dagster run is still publishing.
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+import time
+from pathlib import Path
+
+import pandas as pd
+from huggingface_hub import hf_hub_download
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+
+# Sibling module (same directory).
+from build_grading_template import (  # noqa: E402
+    build_grading_template,
+    build_template_metadata,
+    write_template_metadata,
+)
+
+from config import get_settings  # noqa: E402
+from config.constants import REPOS  # noqa: E402
+from utils.huggingface import get_hf_api_wrapper  # noqa: E402
+
+logger = logging.getLogger(__name__)
+
+_SPACE_ROOT = Path(__file__).resolve().parent
+_DEFAULT_OUTPUT = _SPACE_ROOT / "data" / "en-us" / "grading_template.parquet"
+
+_STRUCTURED_ANALYSIS_REPO = REPOS["structured_answers"]
+
+
+def _hyphen_to_underscore(jurisdiction: str) -> str:
+    return jurisdiction.replace("-", "_")
+
+
+def _list_run_parquets(
+    *,
+    branch: str,
+    experiment: str,
+    variant: str,
+    jurisdiction: str,
+) -> list[str]:
+    """Return matching parquet paths under experiment/variant/jurisdiction/."""
+    prefix = f"{experiment}/{variant}/{_hyphen_to_underscore(jurisdiction)}/"
+    hf = get_hf_api_wrapper()
+    files = hf.api.list_repo_files(
+        repo_id=f"{hf.config.organization}/{_STRUCTURED_ANALYSIS_REPO}",
+        repo_type="dataset",
+        revision=branch,
+    )
+    return sorted(f for f in files if f.startswith(prefix) and f.endswith(".parquet"))
+
+
+def resolve_hf_parquet(
+    *,
+    branch: str,
+    experiment: str,
+    variant: str,
+    jurisdiction: str,
+    path: str | None,
+) -> str:
+    """Return the repo-relative parquet path to download."""
+    if path:
+        return path
+    matches = _list_run_parquets(
+        branch=branch,
+        experiment=experiment,
+        variant=variant,
+        jurisdiction=jurisdiction,
+    )
+    if not matches:
+        raise FileNotFoundError(
+            f"No structured_analysis parquet yet under "
+            f"{experiment}/{variant}/{_hyphen_to_underscore(jurisdiction)}/ "
+            f"on branch {branch!r}. Wait for the pipeline publish, or pass --input."
+        )
+    chosen = matches[-1]
+    if len(matches) > 1:
+        logger.info("Found %d parquet(s); using latest: %s", len(matches), chosen)
+    return chosen
+
+
+def download_hf_parquet(
+    *,
+    repo_path: str,
+    branch: str,
+    local_dir: Path,
+) -> Path:
+    settings = get_settings()
+    org = settings.huggingface.organization
+    token = settings.huggingface.token
+    local_dir.mkdir(parents=True, exist_ok=True)
+    downloaded = hf_hub_download(
+        repo_id=f"{org}/{_STRUCTURED_ANALYSIS_REPO}",
+        filename=repo_path,
+        repo_type="dataset",
+        revision=branch,
+        token=token,
+        local_dir=str(local_dir),
+    )
+    return Path(downloaded)
+
+
+def poll_until_available(
+    *,
+    branch: str,
+    experiment: str,
+    variant: str,
+    jurisdiction: str,
+    path: str | None,
+    interval_s: float,
+    timeout_s: float,
+) -> str:
+    """Block until a matching parquet appears on HF (or timeout)."""
+    deadline = time.monotonic() + timeout_s
+    last_error: Exception | None = None
+    while time.monotonic() < deadline:
+        try:
+            return resolve_hf_parquet(
+                branch=branch,
+                experiment=experiment,
+                variant=variant,
+                jurisdiction=jurisdiction,
+                path=path,
+            )
+        except FileNotFoundError as exc:
+            last_error = exc
+            remaining = max(0, int(deadline - time.monotonic()))
+            logger.info(
+                "Waiting for HF publish (%ds left)… %s",
+                remaining,
+                exc,
+            )
+            time.sleep(interval_s)
+    raise TimeoutError(
+        f"Timed out after {timeout_s:.0f}s waiting for structured_analysis on "
+        f"branch {branch!r}."
+    ) from last_error
+
+
+def build_and_write(
+    *,
+    input_parquet: Path,
+    output_parquet: Path,
+    all_fields_of_law: bool,
+    field_of_law: str | None,
+    max_terms: int | None,
+    include_failed: bool,
+) -> pd.DataFrame:
+    logger.info("Loading %s", input_parquet)
+    structured = pd.read_parquet(input_parquet)
+    one_per_term = not all_fields_of_law and field_of_law is None
+    template = build_grading_template(
+        structured,
+        only_ok=not include_failed,
+        field_of_law=field_of_law,
+        one_per_term=one_per_term,
+        max_terms=max_terms,
+    )
+    output_parquet.parent.mkdir(parents=True, exist_ok=True)
+    template.to_parquet(output_parquet, index=False)
+    metadata = build_template_metadata(
+        input_parquet=input_parquet,
+        output_parquet=output_parquet,
+        template=template,
+        structured_input=structured,
+        only_ok=not include_failed,
+        field_of_law=field_of_law,
+        one_per_term=one_per_term,
+        max_terms=max_terms,
+    )
+    meta_path = write_template_metadata(metadata, output_parquet)
+    n_results = (
+        template[["term", "field_of_law"]].drop_duplicates().shape[0]
+        if not template.empty
+        else 0
+    )
+    logger.info(
+        "Wrote %d grading rows for %d results → %s",
+        len(template),
+        n_results,
+        output_parquet,
+    )
+    logger.info("Wrote metadata → %s", meta_path)
+    if not template.empty:
+        logger.info("Concepts: %s", sorted(template["concept"].unique()))
+        logger.info(
+            "Questions per concept: %s",
+            template.groupby("concept")["question_id"].nunique().to_dict(),
+        )
+    return template
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Ingest structured_analysis into the local grading app "
+            "(download from HF and/or build grading_template.parquet)."
+        )
+    )
+    parser.add_argument(
+        "--input",
+        type=Path,
+        default=None,
+        help="Local structured_analysis parquet (skips Hugging Face download).",
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=_DEFAULT_OUTPUT,
+        help=f"grading_template.parquet path (default: {_DEFAULT_OUTPUT}).",
+    )
+    parser.add_argument(
+        "--branch",
+        default="dev",
+        help="HF dataset branch (default: dev).",
+    )
+    parser.add_argument(
+        "--experiment",
+        default="new-llm-judge",
+        help="Experiment folder on HF (default: new-llm-judge).",
+    )
+    parser.add_argument(
+        "--variant",
+        default="tiny-example",
+        help="Variant folder on HF (default: tiny-example).",
+    )
+    parser.add_argument(
+        "--jurisdiction",
+        default="en-us",
+        help="Jurisdiction code with hyphen (default: en-us).",
+    )
+    parser.add_argument(
+        "--hf-path",
+        default=None,
+        help=(
+            "Exact repo-relative parquet path. Default: latest under "
+            "experiment/variant/jurisdiction/."
+        ),
+    )
+    parser.add_argument(
+        "--download-dir",
+        type=Path,
+        default=_SPACE_ROOT / "data" / "_downloads",
+        help="Where to cache the downloaded HF parquet.",
+    )
+    parser.add_argument(
+        "--poll",
+        action="store_true",
+        help="Keep checking HF until a matching parquet appears.",
+    )
+    parser.add_argument(
+        "--poll-interval",
+        type=float,
+        default=30.0,
+        help="Seconds between HF checks when --poll is set (default: 30).",
+    )
+    parser.add_argument(
+        "--poll-timeout",
+        type=float,
+        default=3600.0,
+        help="Max seconds to wait with --poll (default: 3600).",
+    )
+    parser.add_argument(
+        "--all-fields-of-law",
+        action="store_true",
+        help="Keep every (term, field_of_law) row (default: one field per term).",
+    )
+    parser.add_argument(
+        "--field-of-law",
+        default=None,
+        help="Pin a single field of law (exact match).",
+    )
+    parser.add_argument(
+        "--max-terms",
+        type=int,
+        default=None,
+        help="Optional cap on terms.",
+    )
+    parser.add_argument(
+        "--include-failed",
+        action="store_true",
+        help="Include generation_status != ok rows.",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    args = _parse_args()
+
+    if args.input is not None:
+        input_parquet = args.input.expanduser().resolve()
+        if not input_parquet.is_file():
+            raise FileNotFoundError(f"Input parquet not found: {input_parquet}")
+    else:
+        if args.poll:
+            repo_path = poll_until_available(
+                branch=args.branch,
+                experiment=args.experiment,
+                variant=args.variant,
+                jurisdiction=args.jurisdiction,
+                path=args.hf_path,
+                interval_s=args.poll_interval,
+                timeout_s=args.poll_timeout,
+            )
+        else:
+            repo_path = resolve_hf_parquet(
+                branch=args.branch,
+                experiment=args.experiment,
+                variant=args.variant,
+                jurisdiction=args.jurisdiction,
+                path=args.hf_path,
+            )
+        logger.info(
+            "Downloading hf://datasets/.../%s@%s → %s",
+            repo_path,
+            args.branch,
+            args.download_dir,
+        )
+        input_parquet = download_hf_parquet(
+            repo_path=repo_path,
+            branch=args.branch,
+            local_dir=args.download_dir,
+        )
+
+    build_and_write(
+        input_parquet=input_parquet,
+        output_parquet=args.output.expanduser().resolve(),
+        all_fields_of_law=args.all_fields_of_law,
+        field_of_law=args.field_of_law,
+        max_terms=args.max_terms,
+        include_failed=args.include_failed,
+    )
+    logger.info(
+        "App-ready. From hf-spaces/grading-answers run: "
+        "uv run --with streamlit --with streamlit-shortcuts --with pandas "
+        "--with pyarrow --with huggingface-hub --with pycountry --with babel "
+        "--with python-dotenv streamlit run src/streamlit_app.py"
+    )
+
+
+if __name__ == "__main__":
+    main()

src/streamlit_app.py

@@ -8,9 +8,8 @@ import pandas as pd
 import pycountry
 import streamlit as st
 from babel import Locale
-from huggingface_hub import HfApi, hf_hub_download, login
-
 from glossary import GLOSSARY_TIP_CSS, format_question_html
+from huggingface_hub import HfApi, hf_hub_download, login
 
 st.set_page_config(
     page_title="Answer grading",
@@ -248,6 +247,7 @@ TEMPLATE_COLUMNS = (
     "question_index",
     "answer",
     "source_fields",
+    "source_prompts",
 )
 
 
@@ -437,6 +437,7 @@ def get_cards(df):
         "concept",
         "section",
         "question_id",
+        "source_prompts",
         "question",
         "answer",
     ]
@@ -722,9 +723,9 @@ st.caption(
     f":material/gavel: **{card['term']}**  ·  {card['field_of_law']}  ·  {card['concept']}"
 )
 st.progress((card_index) / total_cards if total_cards else 0.0)
-meta = (
-    f"Card {card_index + 1} of {total_cards}  ·  {card['section']}  ·  {card['question_id']}"
-)
+meta = f"Card {card_index + 1} of {total_cards}  ·  {card['section']}  ·  {card['question_id']}"
+if card.get("source_prompts"):
+    meta += f"  ·  {card['source_prompts'].replace(';', ', ')}"
 if existing:
     badge = ":green-badge[Pass]" if existing == VERDICT_PASS else ":red-badge[Fail]"
     meta += f"  ·  current: {badge}"