Repair Explorer for Unified-3000 (E1.1/E2)

Dataset revision: 1d01f3649ace0301ac3bbe9ee875eea660347a29
Local cumulative diff SHA-256: 35ffcb4f6551646fa5af2f70592c0c65045424b048d30b59d59de3be3fd6a271
Space payload diff SHA-256: 148c00a691ab9208cd70949e7bdaccfb3a70de92e5ccf8bfa3e7ec78935b80ba

README.md

@@ -9,24 +9,54 @@ license: cc-by-nc-4.0
 
 # H²EPR-Bench Explorer
 
-H²EPR-Bench Explorer is the interactive browsing layer for `AgenticFinLab/H2EPR-Bench`. It is separate from the canonical dataset repository: the dataset repo remains the release package, while this Docker Space runs a Streamlit app for search, event detail, stage inspection, public FinalCascade JSON browsing, and Gantt-style timelines.
+H²EPR-Bench Explorer is the interactive browsing layer for the Unified-3000
+release of `AgenticFinLab/H2EPR-Bench`. The existing Docker Space runs a
+Streamlit interface for searching all 3,000 catalog events, inspecting current
+Draft EPG summaries and stage timelines, and previewing or downloading the
+2,876 public per-event Draft EPG files.
 
-**Release boundary:** public records are intended for browsing, reuse, and presentation. Official scoring uses the [manual-gated Gold companion](https://huggingface.co/datasets/AgenticFinLab/H2EPR-Bench-Gold). Public FinalCascade and Gantt views are supplementary inspection assets, not official scoring references.
+The other 124 catalog events remain fully browsable and show a neutral local
+empty state because no public Draft EPG is available for them in this release.
 
-## Data Sources
+**Release boundary:** public Draft EPGs are sanitized FinMycelium construction
+artifacts. Official benchmark scoring uses expert-adjudicated reference EPGs in
+the [manual-gated companion](https://huggingface.co/datasets/AgenticFinLab/H2EPR-Bench-Gold).
+The Explorer does not load reference EPGs or frozen evidence packages.
 
-| Source | Role |
+## Public data contract
+
+The Explorer loads five Parquet tables from one immutable dataset revision:
+
+| Table | Role |
 |---|---|
-| `AgenticFinLab/H2EPR-Bench` | Public event catalog, stage table, public sanitized FinalCascade, and Gantt artifact paths. |
-| `AgenticFinLab/H2EPR-Bench-Gold` | Manual-gated Gold companion for official scoring references. Linked for users who need scoring access; not loaded by this app. |
+| `event_catalog.parquet` | Stable event discovery metadata |
+| `event_instances.parquet` | Public access-state fields |
+| `finalcascade_summary.parquet` | Draft graph counts and event-level temporal summary |
+| `draft_availability.parquet` | Per-event Draft EPG availability and integrity metadata |
+| `event_stages.parquet` | Ordered stage rows for available drafts |
+
+Selected Draft EPGs are loaded directly from
+`draft_events/<H2EPR-ID>/draft_epg.json`. The path is derived only after the
+canonical event ID and availability state have been validated. The availability
+table's `draft_asset` value identifies the consolidated release asset and is
+not used as the selected-event path.
+
+Default dataset revision:
+
+```text
+1d01f3649ace0301ac3bbe9ee875eea660347a29
+```
 
-## Local Development
+## Local development
 
-The app can run from a local staged dataset package before upload:
+The app can use the frozen local Unified-3000 release candidate without any
+network access:
 
 ```bash
-export H2EPR_EXPLORER_LOCAL_DATASET_DIR=../../build/hf_dataset_repo_staging/eventmycelium-v1_1000-public
+export H2EPR_EXPLORER_LOCAL_DATASET_DIR=../../build/hf_unified3000_inplace_upgrade_rc_v3_dataset_card/H2EPR-Bench
 streamlit run app.py
 ```
 
-Without `H2EPR_EXPLORER_LOCAL_DATASET_DIR`, the app downloads public files from the Hugging Face dataset repo.
+Without `H2EPR_EXPLORER_LOCAL_DATASET_DIR`, all public table and selected-event
+downloads use the pinned revision above. The application exposes no runtime
+revision override, so one process cannot mix dataset revisions.

app.py

@@ -4,6 +4,9 @@ import json
 from pathlib import Path
 import sys
 
+import pandas as pd
+
+
 APP_ROOT = Path(__file__).resolve().parent
 sys.path.insert(0, str(APP_ROOT / "src"))
 
@@ -11,11 +14,22 @@ import streamlit as st
 
 from h2epr_explorer.constants import (
     CATALOG_COLUMNS,
+    DRAFT_UNAVAILABLE_MESSAGE,
     GOLD_COMPANION_REPO,
+    PROFILE_COLUMNS,
     PUBLIC_DATASET_REPO,
+    PUBLIC_DATASET_REVISION,
     RELEASE_BOUNDARY_NOTICE,
 )
-from h2epr_explorer.data_loader import load_catalog, load_event_graph, load_finalcascade_summary, load_stages
+from h2epr_explorer.data_loader import (
+    DatasetTransportError,
+    DraftAssetMissing,
+    DraftIntegrityError,
+    DraftUnavailable,
+    ReleaseContractError,
+    load_event_graph,
+    load_release,
+)
 from h2epr_explorer.filters import event_description, event_display_label, event_name, filter_catalog
 from h2epr_explorer.navigation import (
     build_event_links,
@@ -26,25 +40,37 @@ from h2epr_explorer.navigation import (
 from h2epr_explorer.render_gantt import build_timeline_figure
 
 
-def _as_records(frame):
+FILTER_SEARCH_KEY = "h2epr_filter_search"
+FILTER_DOMAIN_KEY = "h2epr_filter_domains"
+FILTER_CATEGORY_KEY = "h2epr_filter_categories"
+FILTER_MIN_STAGE_KEY = "h2epr_filter_min_stage_count"
+FILTER_RESET_KEY = "h2epr_filter_reset"
+FILTER_DEFAULTS = {
+    FILTER_SEARCH_KEY: "",
+    FILTER_DOMAIN_KEY: (),
+    FILTER_CATEGORY_KEY: (),
+    FILTER_MIN_STAGE_KEY: 0,
+}
+
+
+def _as_records(frame: pd.DataFrame) -> list[dict]:
     return frame.to_dict(orient="records")
 
 
-def _select_columns(frame, columns):
-    present = [column for column in columns if column in frame.columns]
-    return frame[present] if present else frame
+def _select_columns(frame: pd.DataFrame, columns: tuple[str, ...] | list[str]) -> pd.DataFrame:
+    return frame.loc[:, list(columns)]
 
 
-def _sort_stage_frame(frame):
-    sort_columns = [column for column in ("stage_index", "stage_order", "stage_id") if column in frame.columns]
-    return frame.sort_values(sort_columns) if sort_columns else frame
+def _metric_value(value) -> str:
+    if value is None or bool(pd.isna(value)):
+        return "—"
+    return f"{int(value):,}"
 
 
-def _safe_int(value, default=0):
-    try:
-        return int(value)
-    except (TypeError, ValueError):
-        return default
+def _reset_filters(state=None) -> None:
+    target = st.session_state if state is None else state
+    for key, value in FILTER_DEFAULTS.items():
+        target[key] = list(value) if isinstance(value, tuple) else value
 
 
 st.set_page_config(page_title="H2EPR-Bench Explorer", layout="wide")
@@ -83,34 +109,71 @@ div[data-testid="stMetric"] {
 st.markdown('<div class="h2epr-kicker">H²EPR-Bench · public release explorer</div>', unsafe_allow_html=True)
 st.markdown('<div class="h2epr-title">Event-process graph browser</div>', unsafe_allow_html=True)
 st.markdown(
-    '<div class="h2epr-subtitle">Browse public event metadata, stage rows, FinalCascade summaries, and Gantt-style timelines for the H²EPR-Bench release.</div>',
+    '<div class="h2epr-subtitle">Browse 3,000 public event records, Draft EPG summaries, and stage timelines from one pinned Unified-3000 release.</div>',
     unsafe_allow_html=True,
 )
 st.info(RELEASE_BOUNDARY_NOTICE)
 
-catalog = load_catalog()
-stages = load_stages()
-summary = load_finalcascade_summary()
+try:
+    release = load_release()
+except DatasetTransportError as exc:
+    st.error(f"The pinned public dataset could not be retrieved. Please retry. Details: {exc}")
+    st.stop()
+except ReleaseContractError as exc:
+    st.error(f"The public release failed Explorer contract validation: {exc}")
+    st.stop()
 
+catalog = release.events
 catalog_rows = _as_records(catalog)
+query_resolution = query_param_event_id(st.query_params)
 
 with st.sidebar:
     st.header("Filter events")
-    query = st.text_input("Search", placeholder="event name, ID, category, keyword")
-    domains = st.multiselect("Domain", sorted(catalog["domain"].dropna().unique().tolist()))
-    categories = st.multiselect("Category", sorted(catalog["event_category"].dropna().unique().tolist()))
-    min_source_count = st.slider("Minimum sources", 0, int(catalog["source_count"].max()), 0)
-    min_stage_count = st.slider("Minimum stages", 0, int(catalog["stage_count"].max()), 0)
+    query = st.text_input(
+        "Search",
+        placeholder="event name, ID, category, keyword",
+        key=FILTER_SEARCH_KEY,
+    )
+    domains = st.multiselect(
+        "Domain",
+        sorted(catalog["domain"].dropna().unique().tolist()),
+        key=FILTER_DOMAIN_KEY,
+    )
+    categories = st.multiselect(
+        "Category",
+        sorted(catalog["category"].dropna().unique().tolist()),
+        key=FILTER_CATEGORY_KEY,
+    )
+    max_stage_count = int(catalog["stage_count"].dropna().max())
+    min_stage_count = st.slider(
+        "Minimum stages",
+        0,
+        max_stage_count,
+        key=FILTER_MIN_STAGE_KEY,
+    )
+    st.button(
+        "Reset filters",
+        key=FILTER_RESET_KEY,
+        on_click=_reset_filters,
+        width="stretch",
+    )
     st.divider()
-    st.link_button("Dataset repository", f"https://huggingface.co/datasets/{PUBLIC_DATASET_REPO}", use_container_width=True)
-    st.link_button("Request Gold access", f"https://huggingface.co/datasets/{GOLD_COMPANION_REPO}", use_container_width=True)
+    st.link_button(
+        "Dataset repository",
+        f"https://huggingface.co/datasets/{PUBLIC_DATASET_REPO}/tree/{PUBLIC_DATASET_REVISION}",
+        width="stretch",
+    )
+    st.link_button(
+        "Reference EPG access",
+        f"https://huggingface.co/datasets/{GOLD_COMPANION_REPO}",
+        width="stretch",
+    )
 
 filtered_rows = filter_catalog(
     catalog_rows,
     query=query,
     domains=domains,
     categories=categories,
-    min_source_count=min_source_count,
     min_stage_count=min_stage_count,
 )
 
@@ -118,113 +181,131 @@ if not filtered_rows:
     st.warning("No event matches the current filters.")
     st.stop()
 
-event_labels = {row["event_id"]: event_display_label(row) for row in catalog_rows}
-requested_event_id = query_param_event_id(st.query_params)
-selected_index = resolve_selected_event_index(filtered_rows, requested_event_id)
+if query_resolution.used_legacy_mapping:
+    st.caption(f"Historical event link resolved to canonical ID {query_resolution.canonical_id}.")
+elif query_resolution.unresolved:
+    st.warning("The requested event link is malformed or outside this release; showing a valid event instead.")
 
+event_labels = {row["event_id"]: event_display_label(row) for row in catalog_rows}
+selected_index = resolve_selected_event_index(filtered_rows, query_resolution.canonical_id)
 selected_event = st.selectbox(
     "Selected event",
     [row["event_id"] for row in filtered_rows],
     index=selected_index,
     format_func=lambda event_id: event_labels.get(event_id, event_id),
 )
-try:
+if st.query_params.get("event_id") != selected_event:
     st.query_params["event_id"] = selected_event
-except Exception:
-    pass
 
-event_row = catalog[catalog["event_id"] == selected_event].iloc[0]
+event_row = release.event_row(selected_event)
 event_record = event_row.to_dict()
-event_stages = _sort_stage_frame(stages[stages["event_id"] == selected_event])
-summary_row = summary[summary["event_id"] == selected_event]
-event_links = build_event_links(selected_event, str(event_record.get("gantt_html_path") or ""))
+event_stages = release.stage_frame(selected_event)
+stage_records = _as_records(event_stages)
+draft_available = event_record["draft_status"] == "draft_available"
+event_links = build_event_links(selected_event, draft_available=draft_available)
 
 st.caption(filter_summary_text(len(filtered_rows), len(catalog_rows)))
 
-tabs = st.tabs(["Catalog", "Event detail", "Timeline", "Stages", "FinalCascade JSON", "Access and boundary"])
+tabs = st.tabs(
+    ["Catalog", "Event detail", "Timeline", "Stages", "Draft EPG JSON", "Access and boundary"]
+)
 
 with tabs[0]:
     st.subheader("Event catalog")
-    st.dataframe(_select_columns(catalog[catalog["event_id"].isin([row["event_id"] for row in filtered_rows])], CATALOG_COLUMNS), use_container_width=True, height=520)
+    filtered_ids = [row["event_id"] for row in filtered_rows]
+    table = catalog.loc[catalog["event_id"].isin(filtered_ids)]
+    st.dataframe(_select_columns(table, CATALOG_COLUMNS), width="stretch", height=520)
 
 with tabs[1]:
     st.subheader(event_name(event_record))
     st.write(event_description(event_record))
-    c1, c2, c3, c4, c5 = st.columns(5)
-    c1.metric("Sources", _safe_int(event_row.get("source_count", 0)))
-    c2.metric("Stages", _safe_int(event_row.get("stage_count", 0)))
-    c3.metric("Episodes", _safe_int(event_row.get("episode_count", 0)))
-    c4.metric("Participants", _safe_int(event_row.get("participant_count", 0)))
-    c5.metric("Relations", _safe_int(event_row.get("relation_count", 0)))
+    metric_columns = st.columns(5)
+    for column, label, widget in zip(
+        ("stage_count", "episode_count", "participant_count", "action_count", "relation_count"),
+        ("Stages", "Episodes", "Participants", "Actions", "Relations"),
+        metric_columns,
+    ):
+        widget.metric(label, _metric_value(event_record.get(column)))
 
     st.markdown("#### Event profile")
-    profile_columns = [
+    selected_frame = catalog.loc[catalog["event_id"].eq(selected_event)]
+    st.dataframe(
+        _select_columns(selected_frame, PROFILE_COLUMNS),
+        width="stretch",
+    )
+
+    link_columns = st.columns(3 if draft_available else 2)
+    link_columns[0].link_button("Open dataset", event_links["public_dataset"], width="stretch")
+    link_columns[1].link_button(
+        "Reference EPG access", event_links["reference_access"], width="stretch"
+    )
+    if draft_available:
+        link_columns[2].link_button(
+            "Open Draft EPG file", event_links["draft_epg"], width="stretch"
+        )
+
+    st.markdown("#### Draft EPG summary")
+    summary_columns = [
         "event_id",
-        "display_name",
-        "domain",
-        "event_category",
-        "event_scope_label",
-        "keywords",
+        "stage_count",
+        "episode_count",
+        "participant_count",
+        "action_count",
+        "transaction_count",
+        "relation_count",
         "event_boundary_time_status",
-        "temporal_anchor_summary",
-        "gold_reference_access_level",
-        "finalcascade_access_level",
+        "known_action_time_anchor_count",
     ]
-    st.dataframe(_select_columns(catalog[catalog["event_id"] == selected_event], profile_columns), use_container_width=True)
-
-    link_cols = st.columns(4)
-    link_cols[0].link_button("Open dataset", event_links["public_dataset"], use_container_width=True)
-    link_cols[1].link_button("Gold access", event_links["gold_request"], use_container_width=True)
-    link_cols[2].link_button("FinalCascade file", event_links["finalcascade_jsonl"], use_container_width=True)
-    if "gantt_html" in event_links:
-        link_cols[3].link_button("Gantt artifact", event_links["gantt_html"], use_container_width=True)
-
-    if not summary_row.empty:
-        st.markdown("#### Public FinalCascade summary")
-        summary_columns = [
-            "event_id",
-            "stage_count",
-            "episode_count",
-            "participant_count",
-            "transaction_count",
-            "relation_count",
-            "event_boundary_time_status",
-            "known_action_time_anchor_count",
-            "not_gold_warning",
-        ]
-        st.dataframe(_select_columns(summary_row, summary_columns), use_container_width=True)
+    st.dataframe(selected_frame.loc[:, summary_columns], width="stretch")
 
 with tabs[2]:
-    figure = build_timeline_figure(_as_records(event_stages), selected_event)
+    figure = build_timeline_figure(stage_records, selected_event)
     if figure is None:
-        st.warning("No public stage rows are available for this event.")
+        st.info(DRAFT_UNAVAILABLE_MESSAGE)
     else:
-        st.plotly_chart(figure, use_container_width=True)
-    if "gantt_html_path" in event_row and event_row.get("gantt_html_path"):
-        st.markdown(f"Gantt HTML artifact path: `{event_row.get('gantt_html_path')}`")
+        st.plotly_chart(figure, width="stretch")
 
 with tabs[3]:
-    st.dataframe(event_stages, use_container_width=True, height=520)
+    if event_stages.empty:
+        st.info(DRAFT_UNAVAILABLE_MESSAGE)
+    else:
+        st.dataframe(event_stages, width="stretch", height=520)
 
 with tabs[4]:
-    graph = load_event_graph(selected_event)
-    st.download_button(
-        "Download selected public FinalCascade JSON",
-        data=json.dumps(graph, ensure_ascii=False, indent=2),
-        file_name=f"{selected_event}_finalcascade_public.json",
-        mime="application/json",
-    )
-    st.json(graph, expanded=False)
+    if not draft_available:
+        st.info(DRAFT_UNAVAILABLE_MESSAGE)
+    else:
+        try:
+            graph_result = load_event_graph(selected_event, release=release)
+        except DatasetTransportError as exc:
+            st.error(f"The selected Draft EPG could not be retrieved. Please retry. Details: {exc}")
+        except DraftAssetMissing as exc:
+            st.error(f"The selected event is marked available but its Draft EPG file is missing: {exc}")
+        except DraftIntegrityError as exc:
+            st.error(f"The selected Draft EPG failed integrity validation and was not shown: {exc}")
+        else:
+            if isinstance(graph_result, DraftUnavailable):
+                st.info(graph_result.message)
+            else:
+                graph_json = json.dumps(graph_result, ensure_ascii=False, indent=2)
+                st.download_button(
+                    "Download selected Draft EPG JSON",
+                    data=graph_json,
+                    file_name=f"{selected_event}_draft_epg.json",
+                    mime="application/json",
+                )
+                st.link_button("Open exact public file", event_links["draft_epg"])
+                st.json(graph_result, expanded=False)
 
 with tabs[5]:
     st.markdown(
         f"""
 ### Release boundary
 
-- Public dataset repo: [`{PUBLIC_DATASET_REPO}`](https://huggingface.co/datasets/{PUBLIC_DATASET_REPO})
-- Manual-gated Gold companion: [`{GOLD_COMPANION_REPO}`](https://huggingface.co/datasets/{GOLD_COMPANION_REPO})
-- This Explorer loads public event metadata, public stages, public sanitized FinalCascade records, and public visualization paths.
-- It does not load gated Gold references.
-- Public FinalCascade and Gantt views are supplementary inspection assets, not official scoring references.
+- Public dataset: [`{PUBLIC_DATASET_REPO}`](https://huggingface.co/datasets/{PUBLIC_DATASET_REPO}/tree/{PUBLIC_DATASET_REVISION}) at `{PUBLIC_DATASET_REVISION}`.
+- Manual-gated reference EPG companion: [`{GOLD_COMPANION_REPO}`](https://huggingface.co/datasets/{GOLD_COMPANION_REPO}).
+- Public Draft EPGs are sanitized FinMycelium construction artifacts, not scoring references.
+- Official benchmark scoring uses expert-adjudicated reference EPGs in the gated companion.
+- This Explorer loads neither reference EPGs nor frozen evidence packages.
 """
     )

src/h2epr_explorer/constants.py

@@ -1,30 +1,172 @@
+from __future__ import annotations
+
 PUBLIC_DATASET_REPO = "AgenticFinLab/H2EPR-Bench"
 GOLD_COMPANION_REPO = "AgenticFinLab/H2EPR-Bench-Gold"
+
+DEFAULT_PUBLIC_DATASET_REVISION = "1d01f3649ace0301ac3bbe9ee875eea660347a29"
+PUBLIC_DATASET_REVISION = DEFAULT_PUBLIC_DATASET_REVISION
 LOCAL_DATASET_ENV = "H2EPR_EXPLORER_LOCAL_DATASET_DIR"
 
+EVENT_ID_PATTERN = r"^H2EPR-[0-9]{4}$"
+EVENT_ID_MIN = 1
+EVENT_ID_MAX = 3000
+
 CATALOG_PARQUET = "data/viewer_mirrors/event_catalog.parquet"
-CATALOG_JSONL = "data/event_catalog.jsonl"
-STAGES_PARQUET = "data/viewer_mirrors/event_stages.parquet"
-STAGES_JSONL = "data/event_stages.jsonl"
-FINALCASCADE_JSONL = "data/finmycelium_finalcascade_public.jsonl"
+EVENT_INSTANCES_PARQUET = "data/viewer_mirrors/event_instances.parquet"
 FINALCASCADE_SUMMARY_PARQUET = "data/viewer_mirrors/finalcascade_summary.parquet"
+DRAFT_AVAILABILITY_PARQUET = "data/viewer_mirrors/draft_availability.parquet"
+STAGES_PARQUET = "data/viewer_mirrors/event_stages.parquet"
+DRAFT_EPG_PATH_TEMPLATE = "draft_events/{event_id}/draft_epg.json"
 
-RELEASE_BOUNDARY_NOTICE = (
-    "Official scoring uses the manual-gated Gold companion repository. "
-    "Public FinalCascade and Gantt views are supplementary inspection assets, "
-    "not official scoring references."
+EXPECTED_EVENT_COUNT = 3000
+EXPECTED_AVAILABLE_DRAFT_COUNT = 2876
+EXPECTED_UNAVAILABLE_DRAFT_COUNT = 124
+EXPECTED_STAGE_ROW_COUNT = 8500
+
+CATALOG_SCHEMA = (
+    "public_event_id",
+    "event_id",
+    "title",
+    "display_name",
+    "event_descriptor",
+    "domain",
+    "category",
+    "keywords",
+    "release_split",
+    "version",
+    "schema_version",
+    "draft_status",
+    "has_gold_reference",
 )
 
-CATALOG_COLUMNS = [
+EVENT_INSTANCES_SCHEMA = (
+    "public_event_id",
     "event_id",
+    "title",
     "display_name",
+    "event_descriptor",
     "domain",
-    "event_category",
-    "event_descriptor_en",
+    "category",
     "keywords",
-    "source_count",
+    "release_split",
+    "version",
+    "schema_version",
+    "has_finalcascade",
+    "draft_status",
+    "has_gold_reference",
+    "finalcascade_access_level",
+    "gold_reference_access_level",
+    "evidence_context_access_level",
+)
+
+FINALCASCADE_SUMMARY_SCHEMA = (
+    "public_event_id",
+    "event_id",
+    "title",
+    "domain",
+    "category",
+    "draft_status",
     "stage_count",
+    "episode_count",
+    "participant_count",
+    "action_count",
+    "transaction_count",
+    "relation_count",
+    "event_start_time",
+    "event_end_time",
     "event_boundary_time_status",
     "known_action_time_anchor_count",
-    "gantt_html_path",
-]
+    "known_action_time_anchors",
+    "relative_order_available",
+)
+
+DRAFT_AVAILABILITY_SCHEMA = (
+    "public_event_id",
+    "draft_status",
+    "draft_source_kind",
+    "draft_schema",
+    "draft_asset",
+    "draft_record_index",
+    "draft_sha256",
+    "source_payload_sha256",
+    "has_reference_epg",
+)
+
+STAGES_SCHEMA = (
+    "public_event_id",
+    "event_id",
+    "stage_id",
+    "stage_index",
+    "stage_title",
+    "stage_start_time",
+    "stage_end_time",
+    "stage_boundary_time_status",
+    "episode_count",
+    "participant_count",
+    "action_count",
+    "transaction_count",
+    "relation_count",
+    "known_action_time_anchor_count",
+    "known_action_time_anchors",
+    "relative_order_available",
+    "release_split",
+    "version",
+    "schema_version",
+)
+
+GRAPH_COUNT_COLUMNS = (
+    "stage_count",
+    "episode_count",
+    "participant_count",
+    "action_count",
+    "transaction_count",
+    "relation_count",
+)
+
+ARROW_INT64_COLUMNS = frozenset(
+    {
+        *GRAPH_COUNT_COLUMNS,
+        "stage_index",
+        "known_action_time_anchor_count",
+        "draft_record_index",
+    }
+)
+
+ARROW_BOOL_COLUMNS = frozenset(
+    {
+        "has_gold_reference",
+        "has_finalcascade",
+        "relative_order_available",
+        "has_reference_epg",
+    }
+)
+
+CATALOG_COLUMNS = (
+    "event_id",
+    "display_name",
+    "domain",
+    "category",
+    "event_descriptor",
+    "keywords",
+    "stage_count",
+)
+
+PROFILE_COLUMNS = (
+    "event_id",
+    "display_name",
+    "domain",
+    "category",
+    "keywords",
+    "event_boundary_time_status",
+    "known_action_time_anchors",
+    "gold_reference_access_level",
+    "finalcascade_access_level",
+)
+
+DRAFT_UNAVAILABLE_MESSAGE = "No public Draft EPG is available for this event in this release."
+
+RELEASE_BOUNDARY_NOTICE = (
+    "Public Draft EPGs are FinMycelium construction artifacts. Official benchmark "
+    "scoring uses reference EPGs in the manual-gated companion repository; this "
+    "Explorer loads neither reference EPGs nor frozen evidence packages."
+)

src/h2epr_explorer/data_loader.py

@@ -1,36 +1,100 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
+from functools import lru_cache
+import hashlib
 import json
 import os
-from functools import lru_cache
 from pathlib import Path
+import re
 from typing import Any
 
+from huggingface_hub import hf_hub_download
+from huggingface_hub.errors import EntryNotFoundError
+import pandas as pd
+import pyarrow.parquet as parquet
+
 from .constants import (
-    CATALOG_JSONL,
+    ARROW_BOOL_COLUMNS,
+    ARROW_INT64_COLUMNS,
     CATALOG_PARQUET,
-    FINALCASCADE_JSONL,
+    CATALOG_SCHEMA,
+    DRAFT_AVAILABILITY_PARQUET,
+    DRAFT_AVAILABILITY_SCHEMA,
+    DRAFT_EPG_PATH_TEMPLATE,
+    DRAFT_UNAVAILABLE_MESSAGE,
+    EVENT_ID_MAX,
+    EVENT_ID_MIN,
+    EVENT_ID_PATTERN,
+    EVENT_INSTANCES_PARQUET,
+    EVENT_INSTANCES_SCHEMA,
+    EXPECTED_AVAILABLE_DRAFT_COUNT,
+    EXPECTED_EVENT_COUNT,
+    EXPECTED_STAGE_ROW_COUNT,
+    EXPECTED_UNAVAILABLE_DRAFT_COUNT,
     FINALCASCADE_SUMMARY_PARQUET,
+    FINALCASCADE_SUMMARY_SCHEMA,
+    GRAPH_COUNT_COLUMNS,
     LOCAL_DATASET_ENV,
     PUBLIC_DATASET_REPO,
-    STAGES_JSONL,
+    PUBLIC_DATASET_REVISION,
     STAGES_PARQUET,
+    STAGES_SCHEMA,
 )
 
 
-class SimpleTable:
-    """Small fallback table used when pandas is unavailable in local checks."""
+class ExplorerDataError(RuntimeError):
+    """Base error for public Explorer data access."""
+
+
+class DatasetTransportError(ExplorerDataError):
+    """A pinned public dataset asset could not be retrieved."""
+
+
+class ReleaseContractError(ExplorerDataError):
+    """The loaded files do not form the expected Unified-3000 release."""
+
+
+class DraftAssetMissing(ExplorerDataError):
+    """An available event is missing its required direct Draft EPG file."""
+
 
-    def __init__(self, rows: list[dict[str, Any]]):
-        self._rows = rows
+class DraftIntegrityError(ExplorerDataError):
+    """A direct Draft EPG failed identity or digest validation."""
 
-    def __len__(self) -> int:
-        return len(self._rows)
 
-    def to_dict(self, orient: str = "records") -> list[dict[str, Any]]:
-        if orient != "records":
-            raise ValueError("SimpleTable only supports orient='records'")
-        return list(self._rows)
+class InvalidEventId(ValueError):
+    """An identifier is not a canonical Unified-3000 event ID."""
+
+
+@dataclass(frozen=True)
+class DraftUnavailable:
+    event_id: str
+    message: str = DRAFT_UNAVAILABLE_MESSAGE
+
+
+@dataclass
+class ExplorerRelease:
+    events: pd.DataFrame
+    stages: pd.DataFrame
+    stages_by_event: dict[str, pd.DataFrame]
+    revision: str = PUBLIC_DATASET_REVISION
+
+    def event_row(self, event_id: str) -> pd.Series:
+        validate_event_id(event_id)
+        rows = self.events.loc[self.events["event_id"].eq(event_id)]
+        if rows.empty:
+            raise InvalidEventId(f"Unknown Unified-3000 event ID: {event_id}")
+        if len(rows) != 1:
+            raise ReleaseContractError(f"Duplicate event identity in joined view: {event_id}")
+        return rows.iloc[0]
+
+    def stage_frame(self, event_id: str) -> pd.DataFrame:
+        self.event_row(event_id)
+        frame = self.stages_by_event.get(event_id)
+        if frame is None:
+            return self.stages.iloc[0:0].copy()
+        return frame.copy()
 
 
 def _as_local_root(local_dataset_dir: Path | str | None = None) -> Path | None:
@@ -40,87 +104,388 @@ def _as_local_root(local_dataset_dir: Path | str | None = None) -> Path | None:
     return Path(value).expanduser().resolve()
 
 
+def validate_event_id(event_id: str) -> str:
+    if not isinstance(event_id, str) or not re.fullmatch(EVENT_ID_PATTERN, event_id):
+        raise InvalidEventId(f"Invalid Unified-3000 event ID: {event_id!r}")
+    number = int(event_id.rsplit("-", 1)[1])
+    if not EVENT_ID_MIN <= number <= EVENT_ID_MAX:
+        raise InvalidEventId(f"Unified-3000 event ID is out of range: {event_id}")
+    return event_id
+
+
 def resolve_dataset_file(filename: str, local_dataset_dir: Path | str | None = None) -> Path:
     local_root = _as_local_root(local_dataset_dir)
     if local_root is not None:
         path = (local_root / filename).resolve()
         if not path.is_relative_to(local_root):
             raise ValueError(f"Refusing to read outside local dataset root: {filename}")
-        if not path.exists():
+        if not path.is_file():
             raise FileNotFoundError(path)
         return path
 
-    from huggingface_hub import hf_hub_download
+    try:
+        return Path(
+            hf_hub_download(
+                repo_id=PUBLIC_DATASET_REPO,
+                repo_type="dataset",
+                filename=filename,
+                revision=PUBLIC_DATASET_REVISION,
+            )
+        )
+    except EntryNotFoundError as exc:
+        raise FileNotFoundError(filename) from exc
+    except Exception as exc:
+        raise DatasetTransportError(
+            f"Unable to retrieve {filename!r} from pinned dataset revision "
+            f"{PUBLIC_DATASET_REVISION}."
+        ) from exc
+
+
+def _read_required_parquet(
+    filename: str,
+    expected_schema: tuple[str, ...],
+    local_dataset_dir: Path | str | None = None,
+) -> pd.DataFrame:
+    try:
+        path = resolve_dataset_file(filename, local_dataset_dir=local_dataset_dir)
+        arrow_schema = parquet.read_schema(path)
+        frame = pd.read_parquet(path)
+    except DatasetTransportError:
+        raise
+    except Exception as exc:
+        raise ReleaseContractError(f"Unable to read required Parquet table: {filename}") from exc
 
-    return Path(
-        hf_hub_download(
-            repo_id=PUBLIC_DATASET_REPO,
-            repo_type="dataset",
-            filename=filename,
+    observed = tuple(frame.columns)
+    observed_arrow = tuple((field.name, str(field.type)) for field in arrow_schema)
+    expected_arrow = tuple(
+        (
+            column,
+            "int64"
+            if column in ARROW_INT64_COLUMNS
+            else "bool"
+            if column in ARROW_BOOL_COLUMNS
+            else "string",
         )
+        for column in expected_schema
     )
+    if observed != expected_schema or observed_arrow != expected_arrow:
+        raise ReleaseContractError(
+            f"Schema mismatch for {filename}: expected {expected_arrow}, observed {observed_arrow}"
+        )
+    return frame
 
 
-def load_jsonl_rows(filename: str, local_dataset_dir: Path | str | None = None) -> list[dict[str, Any]]:
-    path = resolve_dataset_file(filename, local_dataset_dir=local_dataset_dir)
-    rows: list[dict[str, Any]] = []
-    with path.open("r", encoding="utf-8") as handle:
-        for line in handle:
-            line = line.strip()
-            if line:
-                rows.append(json.loads(line))
-    return rows
+def _require_row_count(frame: pd.DataFrame, expected: int, table_name: str) -> None:
+    if len(frame) != expected:
+        raise ReleaseContractError(
+            f"{table_name} row count mismatch: expected {expected}, observed {len(frame)}"
+        )
 
 
-def read_event_graph_from_jsonl(
-    event_id: str, local_dataset_dir: Path | str | None = None
-) -> dict[str, Any]:
-    path = resolve_dataset_file(FINALCASCADE_JSONL, local_dataset_dir=local_dataset_dir)
-    with path.open("r", encoding="utf-8") as handle:
-        for line in handle:
-            if not line.strip():
-                continue
-            row = json.loads(line)
-            if row.get("event_id") == event_id:
-                return row
-    raise KeyError(f"Event graph not found: {event_id}")
-
-
-def _read_table(filename: str, fallback_jsonl: str, local_dataset_dir: Path | str | None = None):
-    pd = None
-    try:
-        import pandas as pandas_module
+def _require_unique(frame: pd.DataFrame, columns: list[str], table_name: str) -> None:
+    if frame.duplicated(columns).any():
+        raise ReleaseContractError(f"{table_name} has duplicate identity rows for {columns}")
+
+
+def _require_equal_ids(frame: pd.DataFrame, table_name: str) -> None:
+    if not frame["event_id"].equals(frame["public_event_id"]):
+        raise ReleaseContractError(f"{table_name} event_id/public_event_id mismatch")
+
+
+def _require_semantic_equality(
+    catalog: pd.DataFrame,
+    other: pd.DataFrame,
+    columns: tuple[str, ...],
+    other_name: str,
+) -> None:
+    left = catalog.set_index("event_id").sort_index()
+    right = other.set_index("event_id").sort_index()
+    for column in columns:
+        if not left[column].equals(right[column]):
+            raise ReleaseContractError(f"{column} disagrees between catalog and {other_name}")
+
+
+def build_explorer_view(
+    catalog: pd.DataFrame,
+    instances: pd.DataFrame,
+    summary: pd.DataFrame,
+    availability: pd.DataFrame,
+) -> pd.DataFrame:
+    """Build the validated one-row-per-event Unified-3000 Explorer view."""
+
+    for frame, schema, name in (
+        (catalog, CATALOG_SCHEMA, "event_catalog"),
+        (instances, EVENT_INSTANCES_SCHEMA, "event_instances"),
+        (summary, FINALCASCADE_SUMMARY_SCHEMA, "finalcascade_summary"),
+        (availability, DRAFT_AVAILABILITY_SCHEMA, "draft_availability"),
+    ):
+        if tuple(frame.columns) != schema:
+            raise ReleaseContractError(f"Schema mismatch for {name}")
+        _require_row_count(frame, EXPECTED_EVENT_COUNT, name)
+
+    expected_ids = {f"H2EPR-{index:04d}" for index in range(EVENT_ID_MIN, EVENT_ID_MAX + 1)}
+    _require_unique(catalog, ["public_event_id", "event_id"], "event_catalog")
+    _require_unique(instances, ["public_event_id", "event_id"], "event_instances")
+    _require_unique(summary, ["public_event_id", "event_id"], "finalcascade_summary")
+    _require_unique(availability, ["public_event_id"], "draft_availability")
+    _require_equal_ids(catalog, "event_catalog")
+    _require_equal_ids(instances, "event_instances")
+    _require_equal_ids(summary, "finalcascade_summary")
+    if set(catalog["event_id"]) != expected_ids:
+        raise ReleaseContractError("event_catalog does not contain the exact Unified-3000 ID set")
+    if set(instances["event_id"]) != expected_ids or set(summary["event_id"]) != expected_ids:
+        raise ReleaseContractError("instance/summary event identity does not match the catalog")
+    if set(availability["public_event_id"]) != expected_ids:
+        raise ReleaseContractError("availability identity does not match the catalog")
+
+    _require_semantic_equality(
+        catalog, instances, ("domain", "category", "draft_status"), "event_instances"
+    )
+    _require_semantic_equality(
+        catalog, summary, ("domain", "category", "draft_status"), "finalcascade_summary"
+    )
+    catalog_status = catalog.set_index("public_event_id")["draft_status"].sort_index()
+    availability_status = availability.set_index("public_event_id")["draft_status"].sort_index()
+    if not catalog_status.equals(availability_status):
+        raise ReleaseContractError("draft_status disagrees between catalog and availability")
+
+    status_counts = availability["draft_status"].value_counts(dropna=False).to_dict()
+    expected_status_counts = {
+        "draft_available": EXPECTED_AVAILABLE_DRAFT_COUNT,
+        "draft_unavailable": EXPECTED_UNAVAILABLE_DRAFT_COUNT,
+    }
+    if status_counts != expected_status_counts:
+        raise ReleaseContractError(
+            f"Draft availability mismatch: expected {expected_status_counts}, observed {status_counts}"
+        )
+
+    access_fields = [
+        "public_event_id",
+        "event_id",
+        "has_finalcascade",
+        "finalcascade_access_level",
+        "gold_reference_access_level",
+        "evidence_context_access_level",
+    ]
+    summary_fields = [
+        "event_id",
+        *GRAPH_COUNT_COLUMNS,
+        "event_start_time",
+        "event_end_time",
+        "event_boundary_time_status",
+        "known_action_time_anchor_count",
+        "known_action_time_anchors",
+        "relative_order_available",
+    ]
+    availability_fields = [
+        "public_event_id",
+        "draft_source_kind",
+        "draft_schema",
+        "draft_asset",
+        "draft_record_index",
+        "draft_sha256",
+        "source_payload_sha256",
+        "has_reference_epg",
+    ]
 
-        pd = pandas_module
-    except ImportError:
-        pass
     try:
-        path = resolve_dataset_file(filename, local_dataset_dir=local_dataset_dir)
-        if pd is None:
-            raise ImportError("pandas is unavailable")
-        return pd.read_parquet(path)
-    except (FileNotFoundError, ImportError, ValueError):
-        rows = load_jsonl_rows(fallback_jsonl, local_dataset_dir=local_dataset_dir)
-        if pd is None:
-            return SimpleTable(rows)
-        return pd.DataFrame(rows)
+        joined = catalog.merge(
+            instances[access_fields],
+            on=["public_event_id", "event_id"],
+            how="left",
+            validate="one_to_one",
+        )
+        joined = joined.merge(
+            summary[summary_fields], on="event_id", how="left", validate="one_to_one"
+        )
+        joined = joined.merge(
+            availability[availability_fields],
+            on="public_event_id",
+            how="left",
+            validate="one_to_one",
+        )
+    except Exception as exc:
+        raise ReleaseContractError("Unified-3000 Explorer join multiplicity failure") from exc
+
+    _require_row_count(joined, EXPECTED_EVENT_COUNT, "joined Explorer view")
+    _require_unique(joined, ["public_event_id", "event_id"], "joined Explorer view")
+    if joined[access_fields[2:] + availability_fields[1:]].isna().all(axis=1).any():
+        raise ReleaseContractError("Joined Explorer view contains unmatched access/availability rows")
+
+    available = joined["draft_status"].eq("draft_available")
+    unavailable = joined["draft_status"].eq("draft_unavailable")
+    if joined.loc[available, list(GRAPH_COUNT_COLUMNS)].isna().any().any():
+        raise ReleaseContractError("Available drafts have null graph counts")
+    if not joined.loc[unavailable, list(GRAPH_COUNT_COLUMNS)].isna().all().all():
+        raise ReleaseContractError("Unavailable drafts contain observed graph counts")
+    if not joined.loc[available, "has_finalcascade"].eq(True).all():
+        raise ReleaseContractError("Available draft rows disagree with has_finalcascade")
+    if not joined.loc[unavailable, "has_finalcascade"].eq(False).all():
+        raise ReleaseContractError("Unavailable draft rows disagree with has_finalcascade")
+    return joined.sort_values("event_id", kind="stable").reset_index(drop=True)
+
+
+def _validate_stages(stages: pd.DataFrame, events: pd.DataFrame) -> None:
+    if tuple(stages.columns) != STAGES_SCHEMA:
+        raise ReleaseContractError("Schema mismatch for event_stages")
+    _require_row_count(stages, EXPECTED_STAGE_ROW_COUNT, "event_stages")
+    _require_equal_ids(stages, "event_stages")
+    if stages[["event_id", "stage_id"]].duplicated().any():
+        raise ReleaseContractError("event_stages contains duplicate stage identity")
+    if stages[["event_id", "stage_index"]].duplicated().any():
+        raise ReleaseContractError("event_stages contains duplicate stage_index")
+    invalid_stage_index = stages["stage_index"].map(
+        lambda value: pd.isna(value) or int(value) != value or int(value) <= 0
+    )
+    if invalid_stage_index.any():
+        raise ReleaseContractError("event_stages stage_index values must be positive integers")
+    if not stages["event_id"].map(
+        lambda value: isinstance(value, str) and bool(re.fullmatch(EVENT_ID_PATTERN, value))
+    ).all():
+        raise ReleaseContractError("event_stages contains a malformed event ID")
+    available_ids = set(events.loc[events["draft_status"].eq("draft_available"), "event_id"])
+    if set(stages["event_id"]) != available_ids:
+        raise ReleaseContractError("Stage coverage does not equal the draft-available event set")
+
+    available_summary = events.loc[
+        events["draft_status"].eq("draft_available"),
+        ["event_id", *GRAPH_COUNT_COLUMNS],
+    ].set_index("event_id")
+    for event_id in sorted(available_ids):
+        event_stages = stages.loc[stages["event_id"].eq(event_id)]
+        expected_stage_count = available_summary.at[event_id, "stage_count"]
+        if (
+            pd.isna(expected_stage_count)
+            or int(expected_stage_count) != expected_stage_count
+            or int(expected_stage_count) <= 0
+            or len(event_stages) != int(expected_stage_count)
+        ):
+            raise ReleaseContractError(f"stage_count closure mismatch for {event_id}")
+
+        observed_indices = sorted(int(value) for value in event_stages["stage_index"])
+        expected_indices = list(range(1, int(expected_stage_count) + 1))
+        if observed_indices != expected_indices:
+            raise ReleaseContractError(f"Non-contiguous stage_index values for {event_id}")
+
+        for column in GRAPH_COUNT_COLUMNS[1:]:
+            expected_count = available_summary.at[event_id, column]
+            observed_count = event_stages[column].sum(min_count=1)
+            if pd.isna(expected_count) or pd.isna(observed_count) or observed_count != expected_count:
+                raise ReleaseContractError(f"{column} closure mismatch for {event_id}")
+
 
+@lru_cache(maxsize=8)
+def _load_release_cached(local_root_value: str | None) -> ExplorerRelease:
+    local_root = Path(local_root_value) if local_root_value else None
+    catalog = _read_required_parquet(CATALOG_PARQUET, CATALOG_SCHEMA, local_root)
+    instances = _read_required_parquet(
+        EVENT_INSTANCES_PARQUET, EVENT_INSTANCES_SCHEMA, local_root
+    )
+    summary = _read_required_parquet(
+        FINALCASCADE_SUMMARY_PARQUET, FINALCASCADE_SUMMARY_SCHEMA, local_root
+    )
+    availability = _read_required_parquet(
+        DRAFT_AVAILABILITY_PARQUET, DRAFT_AVAILABILITY_SCHEMA, local_root
+    )
+    stages = _read_required_parquet(STAGES_PARQUET, STAGES_SCHEMA, local_root)
+    events = build_explorer_view(catalog, instances, summary, availability)
+    _validate_stages(stages, events)
+    stages = stages.sort_values(["event_id", "stage_index", "stage_id"], kind="stable")
+    stages_by_event = {
+        event_id: group.reset_index(drop=True)
+        for event_id, group in stages.groupby("event_id", sort=False)
+    }
+    return ExplorerRelease(
+        events=events,
+        stages=stages.reset_index(drop=True),
+        stages_by_event=stages_by_event,
+    )
+
+
+def load_release(local_dataset_dir: Path | str | None = None) -> ExplorerRelease:
+    local_root = _as_local_root(local_dataset_dir)
+    return _load_release_cached(str(local_root) if local_root is not None else None)
+
+
+def clear_caches() -> None:
+    _load_release_cached.cache_clear()
+    _load_event_graph_cached.cache_clear()
 
-@lru_cache(maxsize=1)
-def load_catalog():
-    return _read_table(CATALOG_PARQUET, CATALOG_JSONL)
 
+def _canonical_graph_sha256(payload: dict[str, Any]) -> str:
+    canonical = json.dumps(
+        payload,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    return hashlib.sha256(canonical).hexdigest()
 
-@lru_cache(maxsize=1)
-def load_stages():
-    return _read_table(STAGES_PARQUET, STAGES_JSONL)
 
+def _validate_graph(payload: Any, event_id: str, event_row: pd.Series) -> dict[str, Any]:
+    if not isinstance(payload, dict):
+        raise DraftIntegrityError(f"Draft EPG for {event_id} is not a JSON object")
+    if payload.get("event_id") != event_id or payload.get("public_event_id") != event_id:
+        raise DraftIntegrityError(f"Draft EPG identity mismatch for {event_id}")
+    nested_event = payload.get("event")
+    if not isinstance(nested_event, dict):
+        raise DraftIntegrityError(f"Nested Draft EPG event must be an object for {event_id}")
+    if nested_event.get("event_id") != event_id:
+        raise DraftIntegrityError(f"Nested Draft EPG identity mismatch for {event_id}")
+    if payload.get("source_payload_sha256") != event_row.get("source_payload_sha256"):
+        raise DraftIntegrityError(f"Draft EPG source payload digest mismatch for {event_id}")
+    if _canonical_graph_sha256(payload) != event_row.get("draft_sha256"):
+        raise DraftIntegrityError(f"Draft EPG canonical digest mismatch for {event_id}")
+    return payload
 
-@lru_cache(maxsize=1)
-def load_finalcascade_summary():
-    return _read_table(FINALCASCADE_SUMMARY_PARQUET, CATALOG_JSONL)
+
+@lru_cache(maxsize=256)
+def _load_event_graph_cached(
+    event_id: str,
+    local_root_value: str | None,
+    expected_source_sha256: str,
+    expected_draft_sha256: str,
+) -> dict[str, Any]:
+    filename = DRAFT_EPG_PATH_TEMPLATE.format(event_id=event_id)
+    local_root = Path(local_root_value) if local_root_value else None
+    try:
+        path = resolve_dataset_file(filename, local_dataset_dir=local_root)
+    except FileNotFoundError as exc:
+        raise DraftAssetMissing(f"Available Draft EPG file is missing: {filename}") from exc
+    try:
+        with path.open("r", encoding="utf-8") as handle:
+            payload = json.load(handle)
+    except Exception as exc:
+        raise DraftIntegrityError(f"Unable to parse Draft EPG JSON for {event_id}") from exc
+    expected = pd.Series(
+        {
+            "source_payload_sha256": expected_source_sha256,
+            "draft_sha256": expected_draft_sha256,
+        }
+    )
+    return _validate_graph(payload, event_id, expected)
 
 
-@lru_cache(maxsize=128)
-def load_event_graph(event_id: str) -> dict[str, Any]:
-    return read_event_graph_from_jsonl(event_id)
+def load_event_graph(
+    event_id: str,
+    *,
+    release: ExplorerRelease | None = None,
+    local_dataset_dir: Path | str | None = None,
+) -> dict[str, Any] | DraftUnavailable:
+    """Load one direct public Draft EPG after catalog and availability checks."""
+
+    validate_event_id(event_id)
+    selected_release = release or load_release(local_dataset_dir=local_dataset_dir)
+    event_row = selected_release.event_row(event_id)
+    if event_row["draft_status"] == "draft_unavailable":
+        return DraftUnavailable(event_id)
+    if event_row["draft_status"] != "draft_available":
+        raise ReleaseContractError(f"Unknown draft_status for {event_id}")
+
+    local_root = _as_local_root(local_dataset_dir)
+    return _load_event_graph_cached(
+        event_id,
+        str(local_root) if local_root is not None else None,
+        str(event_row["source_payload_sha256"]),
+        str(event_row["draft_sha256"]),
+    )

src/h2epr_explorer/filters.py

@@ -1,29 +1,57 @@
 from __future__ import annotations
 
+import math
+import re
 from typing import Any, Iterable
 
+from .constants import EVENT_ID_MAX, EVENT_ID_MIN, EVENT_ID_PATTERN
 
-NAME_FIELDS = ("display_name", "event_name_en", "event_name", "event_name_zh")
-DESCRIPTION_FIELDS = ("short_description", "event_descriptor_en", "event_description_en", "event_description_zh")
+
+NAME_FIELDS = ("display_name", "title")
+DESCRIPTION_FIELDS = ("event_descriptor",)
 SEARCH_FIELDS = (
     "event_id",
+    "public_event_id",
+    "title",
     "display_name",
-    "event_name_en",
-    "event_name",
-    "event_name_zh",
-    "short_description",
-    "event_descriptor_en",
-    "event_description_zh",
+    "event_descriptor",
     "domain",
-    "event_category",
+    "category",
     "keywords",
 )
 
 
 def _text_value(value: Any) -> str:
-    if isinstance(value, list):
+    if isinstance(value, (list, tuple, set)):
         return " ".join(str(item) for item in value)
-    return str(value or "")
+    if value is None:
+        return ""
+    try:
+        if value != value:
+            return ""
+    except (TypeError, ValueError):
+        pass
+    return str(value)
+
+
+def _optional_int(value: Any) -> int | None:
+    if value is None:
+        return None
+    try:
+        if isinstance(value, float) and math.isnan(value):
+            return None
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _require_current_event_id(row: dict[str, Any]) -> None:
+    event_id = _text_value(row.get("event_id")).strip()
+    if not re.fullmatch(EVENT_ID_PATTERN, event_id):
+        raise ValueError(f"Catalog filter received a non-canonical event ID: {event_id!r}")
+    number = int(event_id.rsplit("-", 1)[1])
+    if not EVENT_ID_MIN <= number <= EVENT_ID_MAX:
+        raise ValueError(f"Catalog filter received an out-of-range event ID: {event_id}")
 
 
 def event_name(row: dict[str, Any]) -> str:
@@ -61,22 +89,21 @@ def filter_catalog(
     query: str = "",
     domains: list[str] | None = None,
     categories: list[str] | None = None,
-    min_source_count: int = 0,
     min_stage_count: int = 0,
 ) -> list[dict[str, Any]]:
     domain_set = set(domains or [])
     category_set = set(categories or [])
     filtered: list[dict[str, Any]] = []
     for row in rows:
+        _require_current_event_id(row)
         if domain_set and row.get("domain") not in domain_set:
             continue
-        if category_set and row.get("event_category") not in category_set:
-            continue
-        if int(row.get("source_count") or 0) < min_source_count:
+        if category_set and row.get("category") not in category_set:
             continue
-        if int(row.get("stage_count") or 0) < min_stage_count:
+        stage_count = _optional_int(row.get("stage_count"))
+        if min_stage_count > 0 and (stage_count is None or stage_count < min_stage_count):
             continue
-        if not _contains_query(row, query):
+        if not _contains_query(row, query.strip()):
             continue
         filtered.append(row)
     return filtered

src/h2epr_explorer/navigation.py

@@ -1,13 +1,35 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
+import re
 from typing import Any, Mapping
 
-from .constants import GOLD_COMPANION_REPO, PUBLIC_DATASET_REPO
+from .constants import (
+    DRAFT_EPG_PATH_TEMPLATE,
+    EVENT_ID_MAX,
+    EVENT_ID_MIN,
+    EVENT_ID_PATTERN,
+    GOLD_COMPANION_REPO,
+    PUBLIC_DATASET_REPO,
+    PUBLIC_DATASET_REVISION,
+)
 
 
 SPACE_URL = "https://huggingface.co/spaces/AgenticFinLab/H2EPR-Bench-Explorer"
 PUBLIC_DATASET_URL = f"https://huggingface.co/datasets/{PUBLIC_DATASET_REPO}"
 GOLD_COMPANION_URL = f"https://huggingface.co/datasets/{GOLD_COMPANION_REPO}"
+LEGACY_EVENT_ID_PATTERN = r"^P1000-([0-9]{4})$"
+
+
+@dataclass(frozen=True)
+class QueryEventResolution:
+    raw_value: str
+    canonical_id: str
+    used_legacy_mapping: bool = False
+
+    @property
+    def unresolved(self) -> bool:
+        return bool(self.raw_value) and not self.canonical_id
 
 
 def _first_value(value: Any) -> str:
@@ -16,8 +38,42 @@ def _first_value(value: Any) -> str:
     return str(value or "").strip()
 
 
-def query_param_event_id(query_params: Mapping[str, Any]) -> str:
-    return _first_value(query_params.get("event_id"))
+def _canonical_id(value: str) -> str:
+    if not re.fullmatch(EVENT_ID_PATTERN, value):
+        return ""
+    number = int(value.rsplit("-", 1)[1])
+    return value if EVENT_ID_MIN <= number <= EVENT_ID_MAX else ""
+
+
+def normalize_query_event_id(value: Any) -> QueryEventResolution:
+    """Normalize only inbound navigation IDs; all returned IDs are canonical."""
+
+    raw_value = _first_value(value)
+    canonical = _canonical_id(raw_value)
+    if canonical:
+        return QueryEventResolution(raw_value, canonical)
+
+    match = re.fullmatch(LEGACY_EVENT_ID_PATTERN, raw_value)
+    if not match:
+        return QueryEventResolution(raw_value, "")
+    legacy_number = int(match.group(1))
+    if not 1 <= legacy_number <= 1000:
+        return QueryEventResolution(raw_value, "")
+    if legacy_number <= 359:
+        canonical_number = legacy_number
+    elif legacy_number == 360:
+        canonical_number = 87
+    else:
+        canonical_number = legacy_number - 1
+    return QueryEventResolution(
+        raw_value,
+        f"H2EPR-{canonical_number:04d}",
+        used_legacy_mapping=True,
+    )
+
+
+def query_param_event_id(query_params: Mapping[str, Any]) -> QueryEventResolution:
+    return normalize_query_event_id(query_params.get("event_id"))
 
 
 def resolve_selected_event_index(rows: list[dict[str, Any]], requested_event_id: str = "") -> int:
@@ -30,16 +86,18 @@ def resolve_selected_event_index(rows: list[dict[str, Any]], requested_event_id:
     return 0
 
 
-def build_event_links(event_id: str, gantt_html_path: str = "") -> dict[str, str]:
-    event_id = event_id.strip()
+def build_event_links(event_id: str, *, draft_available: bool = False) -> dict[str, str]:
+    canonical = _canonical_id(event_id.strip())
+    if not canonical:
+        raise ValueError(f"Cannot build current-release links for event ID: {event_id!r}")
     links = {
-        "explorer": f"{SPACE_URL}?event_id={event_id}",
-        "public_dataset": PUBLIC_DATASET_URL,
-        "gold_request": GOLD_COMPANION_URL,
-        "finalcascade_jsonl": f"{PUBLIC_DATASET_URL}/blob/main/data/finmycelium_finalcascade_public.jsonl",
+        "explorer": f"{SPACE_URL}?event_id={canonical}",
+        "public_dataset": f"{PUBLIC_DATASET_URL}/tree/{PUBLIC_DATASET_REVISION}",
+        "reference_access": GOLD_COMPANION_URL,
     }
-    if gantt_html_path:
-        links["gantt_html"] = f"{PUBLIC_DATASET_URL}/blob/main/{gantt_html_path.lstrip('/')}"
+    if draft_available:
+        draft_path = DRAFT_EPG_PATH_TEMPLATE.format(event_id=canonical)
+        links["draft_epg"] = f"{PUBLIC_DATASET_URL}/blob/{PUBLIC_DATASET_REVISION}/{draft_path}"
     return links
 
 

src/h2epr_explorer/render_gantt.py

@@ -1,14 +1,30 @@
 from __future__ import annotations
 
+import json
+import re
 from typing import Any
 
+from .constants import EVENT_ID_MAX, EVENT_ID_MIN, EVENT_ID_PATTERN
+
+
+def _text_value(value: Any) -> str:
+    if value is None:
+        return ""
+    try:
+        if value != value:
+            return ""
+    except (TypeError, ValueError):
+        pass
+    return str(value).strip()
+
 
 def _is_known_time(value: Any) -> bool:
-    return bool(value) and str(value).strip().lower() not in {"unknown", "none", "nan", "nat"}
+    text = _text_value(value).lower()
+    return bool(text) and text not in {"unknown", "none", "nan", "nat"}
 
 
 def _stage_order(row: dict[str, Any], fallback_index: int = 0) -> int:
-    value = row.get("stage_index", row.get("stage_order", fallback_index))
+    value = row.get("stage_index", fallback_index)
     try:
         return int(value)
     except (TypeError, ValueError):
@@ -16,40 +32,62 @@ def _stage_order(row: dict[str, Any], fallback_index: int = 0) -> int:
 
 
 def _stage_label(row: dict[str, Any]) -> str:
-    for field in ("stage_title", "stage_label_public", "stage_label", "stage_id"):
-        value = str(row.get(field) or "").strip()
+    for field in ("stage_title", "stage_id"):
+        value = _text_value(row.get(field))
         if value:
             return value
     return "Unnamed stage"
 
 
+def _time_note(row: dict[str, Any]) -> str:
+    anchors = row.get("known_action_time_anchors")
+    if isinstance(anchors, (list, tuple)):
+        return "; ".join(_text_value(anchor) for anchor in anchors if _text_value(anchor))
+    text = _text_value(anchors)
+    if not text:
+        return ""
+    try:
+        parsed = json.loads(text)
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return text
+    if isinstance(parsed, list):
+        return "; ".join(_text_value(anchor) for anchor in parsed if _text_value(anchor))
+    return text
+
+
+def _require_current_event_id(row: dict[str, Any]) -> None:
+    event_id = _text_value(row.get("event_id"))
+    if not re.fullmatch(EVENT_ID_PATTERN, event_id):
+        raise ValueError(f"Timeline received a non-canonical event ID: {event_id!r}")
+    number = int(event_id.rsplit("-", 1)[1])
+    if not EVENT_ID_MIN <= number <= EVENT_ID_MAX:
+        raise ValueError(f"Timeline received an out-of-range event ID: {event_id}")
+
+
 def prepare_gantt_rows(stage_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    for row in stage_rows:
+        _require_current_event_id(row)
     ordered = sorted(
         stage_rows,
         key=lambda row: (_stage_order(row), str(row.get("stage_id", ""))),
     )
-    calendar_axis = all(
-        _is_known_time(row.get("stage_start_time")) and _is_known_time(row.get("stage_end_time"))
+    calendar_axis = bool(ordered) and all(
+        _is_known_time(row.get("stage_start_time"))
+        and _is_known_time(row.get("stage_end_time"))
         for row in ordered
     )
     prepared: list[dict[str, Any]] = []
     for fallback_index, row in enumerate(ordered, start=1):
-        start = row.get("stage_start_time")
-        end = row.get("stage_end_time")
         stage_order = _stage_order(row, fallback_index)
         if calendar_axis:
-            display_start = start
-            display_end = end
+            display_start = row.get("stage_start_time")
+            display_end = row.get("stage_end_time")
             axis_mode = "calendar"
         else:
             display_start = stage_order
             display_end = display_start + 0.85
             axis_mode = "relative_order"
 
-        time_note = row.get("temporal_anchor_summary") or ""
-        if not time_note and int(row.get("known_action_time_anchor_count") or 0) > 0:
-            time_note = "Action-level time anchors available"
-
         prepared.append(
             {
                 **row,
@@ -58,12 +96,18 @@ def prepare_gantt_rows(stage_rows: list[dict[str, Any]]) -> list[dict[str, Any]]
                 "display_start": display_start,
                 "display_end": display_end,
                 "axis_mode": axis_mode,
-                "time_note": time_note,
+                "time_note": _time_note(row),
             }
         )
     return prepared
 
 
+def _calendar_datetime_values(values: list[Any]) -> list[Any]:
+    import pandas as pd
+
+    return [pd.to_datetime(value, errors="raise") for value in values]
+
+
 def build_timeline_figure(stage_rows: list[dict[str, Any]], event_id: str):
     import pandas as pd
     import plotly.express as px
@@ -74,6 +118,8 @@ def build_timeline_figure(stage_rows: list[dict[str, Any]], event_id: str):
 
     frame = pd.DataFrame(prepared)
     if set(frame["axis_mode"]) == {"calendar"}:
+        frame["display_start"] = _calendar_datetime_values(frame["display_start"].tolist())
+        frame["display_end"] = _calendar_datetime_values(frame["display_end"].tolist())
         fig = px.timeline(
             frame,
             x_start="display_start",