refactor: remove redundant pipeline stages (merge_cve + normalize)

Two redundant phases eliminated after verifying output equivalence:

- merge_cve.py was a strict subset of merge_crawl_csvs.py (same CSV glob,
  source->source_platform normalize, md5 id, dedup, append). CVE CSVs share the
  crawl schema, so they now ride the single merge_crawl_csvs --src-dirs pass.
  Deleted merge_cve.py (-106 lines). Side benefit: CVE rows now get the same
  severity normalization as every other source (6 correctly promoted to the
  authoritative tier instead of defaulting to Info).
- The `normalize` stage (inline python heredoc filling stride/cwe) is
  unnecessary: build_security_dataset now canonicalizes the ''/nan/Other
  sentinels itself. Verified curate-from-crossref == curate-after-normalize
  (identical rows + tiers). Stage 8 is now a plain copy of the crossref parquet.

Pipeline: 9 stages -> cleaner 8 (merge is one pass; publish is a copy).
Curated 1878 -> 1880, tests 7/7 green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: grandchildrice

collection/README.md

@@ -10,7 +10,7 @@ The curated dataset is derived from the raw snapshot **offline** by
 | Per-client repo crawl | `crawl_eth_past_fixes.py`, `grep_eth_commits.py`, `mine_eth_releases.py`, `mine_stealth_prs.py`, `mine_direct_pulls.py`, `parse_eth_changelogs.py`, `extract_nimbus_urgency.py` |
 | Advisory databases | `crawl_cve.py`, `crawl_osv.py`, `crawl_rustsec.py`, `crawl_govulncheck.py`, `crawl_teku_jira_refs.py` |
 | Cross-client / specs | `crawl_cross_client.py`, `crawl_specs_divergence.py` |
-| Merge + classify + enrich | `merge_crawl_csvs.py`, `merge_cve.py`, `build_derived.py`, `classify_stride_cwe.py`, `classify_stride_cwe_sdk.py`, `cross_reference.py`, `blame_walk.py` |
+| Merge + classify + enrich | `merge_crawl_csvs.py`, `build_derived.py`, `classify_stride_cwe.py`, `classify_stride_cwe_sdk.py`, `cross_reference.py`, `blame_walk.py` |
 
 Collection methodology (per-client security-label taxonomies, a body-keyword path
 filter for unlabeled "stealth" fixes, and the rule that a severity from a release

collection/merge_cve.py

@@ -8,10 +8,10 @@
 #   Stage 2  supplementary crawlers        -> $WORK/supp/*.csv
 #   Stage 3  advisory DBs (CVE)            -> $WORK/cve/<client>.cve.csv
 #   Stage 4  build_derived (canonical)     -> $WORK/derived/ethereum/train.parquet
-#   Stage 5  merge supp + cve              -> train.parquet (in place)
+#   Stage 5  merge supp + cve (one pass)   -> train.parquet (in place)
 #   Stage 6  cross_reference (dedup)       -> train.parquet
 #   Stage 7  blame_walk (FULL only)        -> enrich introduced_in_commit
-#   Stage 8  normalize stride/cwe          -> data/raw/train.classified.parquet
+#   Stage 8  publish raw snapshot (copy)   -> data/raw/train.classified.parquet
 #   Stage 9  curate                        -> data/ethereum_vulns.parquet (+manifest)
 #
 # LLM STRIDE/CWE classification is intentionally skipped (stride=Other, cwe=N/A);
@@ -123,9 +123,10 @@ stage build_derived PY collection/build_derived.py --domain ethereum \
     --filter-platforms "" --out-dir "$DERIVED" "${SRC_ARGS[@]}"
 [ -f "$TRAIN" ] || { echo "FATAL: $TRAIN not produced"; exit 1; }
 
-# --- Stage 5: merge supplementary + CVE CSVs -------------------------------
-stage merge_supp PY collection/merge_crawl_csvs.py --src-dirs "$SUPP" --parquet "$TRAIN" --out "$TRAIN"
-stage merge_cve  PY collection/merge_cve.py        --cve-dir "$CVE"  --parquet "$TRAIN" --out "$TRAIN"
+# --- Stage 5: merge supplementary + CVE CSVs (one pass) --------------------
+# CVE CSVs share the crawl schema, so merge_crawl_csvs ingests them alongside
+# the supplementary dir — no separate merge_cve step needed.
+stage merge PY collection/merge_crawl_csvs.py --src-dirs "$SUPP" "$CVE" --parquet "$TRAIN" --out "$TRAIN"
 
 # --- Stage 6: cross_reference (de-dup GHSA/PR/CVE) -------------------------
 stage cross_ref PY collection/cross_reference.py --in "$TRAIN" --out "$DERIVED/ethereum/train.crossref.parquet" --quiet
@@ -137,19 +138,11 @@ if [ "$MODE" = "full" ] && [ "${SKIP_BLAME:-0}" != "1" ]; then
       --manifest "$DERIVED/ethereum/blame_walk_manifest.json"
 fi
 
-# --- Stage 8: normalize stride/cwe (classification skipped) ----------------
+# --- Stage 8: publish the raw snapshot -------------------------------------
+# No stride/cwe munging here: build_security_dataset canonicalizes the
+# unclassified sentinels itself, so this is just a copy of the crossref output.
 mkdir -p data/raw
-stage normalize PY - "$TRAIN" data/raw/train.classified.parquet <<'PYEOF'
-import sys, pandas as pd
-src, dst = sys.argv[1], sys.argv[2]
-df = pd.read_parquet(src)
-if "stride" not in df.columns: df["stride"] = "Other"
-if "cwe_top25" not in df.columns: df["cwe_top25"] = "N/A"
-df["stride"] = df["stride"].fillna("Other").replace("", "Other")
-df["cwe_top25"] = df["cwe_top25"].fillna("N/A").replace("", "N/A")
-df.to_parquet(dst, index=False)
-print(f"normalized {len(df)} rows -> {dst}")
-PYEOF
+stage publish_raw cp "$TRAIN" data/raw/train.classified.parquet
 
 # --- Stage 9: curate --------------------------------------------------------
 stage curate PY pipeline/build_security_dataset.py \

collection/run_pipeline.sh

@@ -1,6 +1,6 @@
 {
   "domain": "ethereum",
-  "n_rows": 1878,
+  "n_rows": 1880,
   "schema": [
     "id",
     "source_platform",
@@ -32,27 +32,27 @@
     "t2_noise_dropped": 1417,
     "t2b_nvd_fp_dropped": 49,
     "after_t1": 16998,
-    "security_rows": 1878,
-    "low_signal_dropped": 15120,
+    "security_rows": 1880,
+    "low_signal_dropped": 15118,
     "by_confidence": {
-      "medium": 1544,
-      "high": 333,
+      "medium": 1542,
+      "high": 337,
       "low": 1
     },
     "by_authority_tier": {
-      "B_corroborated": 1134,
+      "B_corroborated": 1130,
       "C_candidate": 511,
-      "A_authoritative": 233
+      "A_authoritative": 239
     },
     "by_n_signals": {
       "1": 556,
       "2": 907,
-      "3": 345,
-      "4": 65,
+      "3": 343,
+      "4": 69,
       "5": 5
     },
     "by_source": {
-      "geth": 374,
+      "geth": 376,
       "erigon": 358,
       "nimbus": 198,
       "lodestar": 194,
@@ -66,20 +66,20 @@
       "consensus-specs": 2
     },
     "by_severity": {
-      "Unrated": 964,
+      "Unrated": 960,
       "Info": 773,
       "High": 63,
-      "Medium": 54,
+      "Medium": 60,
       "Low": 21,
       "Critical": 3
     },
     "by_score": {
       "0.0": 35,
       "0.3": 4,
       "0.5": 1208,
-      "0.8": 298,
-      "0.9": 172,
-      "1.0": 161
+      "0.8": 296,
+      "0.9": 170,
+      "1.0": 167
     },
     "residual_boilerplate_fp": 0
   },

data/ethereum_vulns.parquet

@@ -282,6 +282,11 @@ def build(df: pd.DataFrame) -> tuple[pd.DataFrame, dict]:
     for col in ("title", "description", "severity", "stride", "cwe_top25", "source_platform"):
         if col not in df.columns:
             df[col] = ""
+    # Canonicalize the "unclassified" sentinels once, so the gate/signal logic
+    # can treat NaN / "" / "nan" identically to Other/N/A. (Makes the former
+    # external `normalize` collection step unnecessary — see run_pipeline.sh.)
+    df["stride"] = df["stride"].fillna("Other").replace({"": "Other", "nan": "Other"})
+    df["cwe_top25"] = df["cwe_top25"].fillna("N/A").replace({"": "N/A", "nan": "N/A"})
     blob = df["title"].fillna("").astype(str) + " " + df["description"].fillna("").astype(str)
 
     # T1