diff --git a/config/config.yaml b/config/config.yaml
index 6bc2735..9ebef73 100644
--- a/config/config.yaml
+++ b/config/config.yaml
@@ -313,6 +313,10 @@ resources:
     mem_mb: 32768
     runtime: 30
     threads: 2
+  process_geojson:
+    mem_mb: 4096
+    runtime: 30
+    threads: 2
   qc_sweep:
     mem_mb: 49152
     runtime: 180
diff --git a/workflow/Snakefile b/workflow/Snakefile
index 9df8b03..b529e9d 100644
--- a/workflow/Snakefile
+++ b/workflow/Snakefile
@@ -7,6 +7,7 @@ postprocessing → exploration). Rules are split across rules/*.smk and included
 directly below; helper functions live in rules/common.smk.
 """
 
+import glob
 import os
 import sys
 import pandas as pd
@@ -132,6 +133,17 @@ if CORE_SAMPLES is None:
         )
     SAMPLE_IDS = list(SAMPLES.keys())
 
+# Optional per-sample region GeoJSON (ROI-based analysis). Presence is optional per
+# sample; a geojson-like file that matches neither accepted name is a fatal config
+# error (see _find_geojson).
+SAMPLES_WITH_GEOJSON = [s for s in SAMPLE_IDS if _find_geojson(s) is not None]
+for _s in SAMPLE_IDS:
+    if _s not in SAMPLES_WITH_GEOJSON:
+        print(
+            f"[process_geojson] No GeoJSON found for sample '{_s}' — "
+            "no ROI-based analysis will be performed for this sample."
+        )
+
 # Optional per-column colour palettes (sample + design columns) for downstream visualization
 SAMPLE_COLORS = config.get("sample_colors", {}) or {}
 
@@ -256,6 +268,7 @@ elif MODE == "merscope":
 include: "rules/validate_input.smk"
 include: "rules/qc_sweep.smk"
 include: "rules/preprocess_umap.smk"
+include: "rules/process_geojson.smk"
 include: "rules/leiden_analysis.smk"
 include: "rules/generate_annotation_template.smk"
 include: "rules/ingest_ref.smk"
@@ -311,6 +324,7 @@ rule run_preprocessing:
         expand(rules.preprocess_umap.output.adata, sample=SAMPLE_IDS),
         expand(rules.preprocess_umap.output.metadata, sample=SAMPLE_IDS),
         expand(rules.preprocess_umap.output.report, sample=SAMPLE_IDS),
+        expand(rules.process_geojson.output.report, sample=SAMPLES_WITH_GEOJSON),
         *QUPATH_IMAGES,
         *(
             expand(
diff --git a/workflow/rules/annotate_cells.smk b/workflow/rules/annotate_cells.smk
index 22be00f..94defc5 100644
--- a/workflow/rules/annotate_cells.smk
+++ b/workflow/rules/annotate_cells.smk
@@ -6,6 +6,7 @@ rule annotate_cells:
     input:
         adata=_annotate_input_adata,
         metadata=rules.preprocess_umap.output.metadata,
+        external_metadata=_external_metadata_input,
         cluster_annotations=config.get("cluster_annotations", "") or [],
         spatial_niche=_annotate_niche_input,
     output:
@@ -28,7 +29,6 @@ rule annotate_cells:
         de_n_genes=ANALYSIS.get("de_n_genes", 10),
         use_precomputed=USE_PRECOMPUTED,
         external_annotation=config.get("external_annotation", {}),
-        precomputed_metadata_dir=config.get("precomputed_metadata_dir", ""),
         annotation_colors=config.get("annotation_colors", {}),
         region_colors=ANALYSIS.get("region_colors", {}),
         dpi=ANALYSIS.get("plot_dpi", 300),
diff --git a/workflow/rules/common.smk b/workflow/rules/common.smk
index 2eb1183..2db86c5 100644
--- a/workflow/rules/common.smk
+++ b/workflow/rules/common.smk
@@ -195,17 +195,64 @@ def _ate_prepare_inputs(wildcards):
                 sample=wildcards.sample
             )
         )
-    # Track the region GeoJSON when present, so editing/renaming it retriggers the
-    # contract build. It is read by filename (regions are optional), not hard-required —
-    # same "declare only when present" pattern as _preprocess_inputs' precomputed_meta.
-    for _suffix in ("_he_background.geojson", "_morphology.geojson", "_he.geojson"):
-        _gj = os.path.join(GEOJ_DIR, f"{wildcards.sample}{_suffix}")
-        if os.path.isfile(_gj):
-            inputs["geojson"] = _gj
-            break
+    # Region annotations are optional. When one is present, track both the source
+    # and its validation report so the contract is rebuilt only after validation.
+    _gj = _find_geojson(wildcards.sample)
+    if _gj:
+        inputs["geojson"] = _gj
+        inputs["geojson_validation"] = _geojson_validation_report(wildcards.sample)
     return inputs
 
 
+def _geojson_suffixes():
+    """Accepted region-annotation filenames for the active head, in priority order."""
+    return {
+        "visiumhd": ("_tissue_hires_image.geojson",),
+        "xenium5k": ("_morphology.geojson",),
+        "atera": ("_he_background.geojson", "_morphology.geojson"),
+        "merscope": ("_morphology.geojson",),
+    }.get(MODE, ())
+
+
+def _find_geojson(sample):
+    """Resolve an optional region GeoJSON using filenames supported by the active head."""
+    if not GEOJ_DIR or not _geojson_suffixes():
+        return None
+
+    accepted = []
+    for _suffix in _geojson_suffixes():
+        accepted.append(os.path.join(GEOJ_DIR, f"{sample}{_suffix}"))
+
+    # A sample-prefixed GeoJSON with any other suffix is almost certainly a typo or
+    # an export for a different coordinate system. Fail before silently ignoring it.
+    discovered = sorted(glob.glob(os.path.join(GEOJ_DIR, f"{sample}_*.geojson")))
+    unsupported = [path for path in discovered if path not in accepted]
+    if unsupported:
+        sys.exit(
+            f"[config error] Unsupported GeoJSON filename(s) for sample '{sample}' "
+            f"in mode '{MODE}': {unsupported}. Expected one of: {accepted}"
+        )
+    return next((path for path in accepted if os.path.isfile(path)), None)
+
+
+def _geojson_validation_report(sample):
+    """Validation marker consumed by a head only when that sample has a GeoJSON."""
+    if _find_geojson(sample) is None:
+        return []
+    return f"{SAMPLES_DIR}/{sample}/validation/geojson_validation.json"
+
+
+def _external_metadata_input(wildcards):
+    """Tracked per-sample external metadata, or no input when the feature is off."""
+    cfg = config.get("external_annotation", {}) or {}
+    if not cfg.get("enabled", False):
+        return []
+    return os.path.join(
+        config.get("precomputed_metadata_dir", "") or "",
+        f"metadata_{wildcards.sample}.tsv",
+    )
+
+
 # ── HEAD (MERSCOPE) input helper ─────────────────────────────────────────────
 def merscope_dir_for(sample):
     """MERSCOPE region directory for a sample, from the merscope.merscope_dir
@@ -310,6 +357,7 @@ def get_all_targets(wildcards):
     targets += expand(rules.preprocess_umap.output.adata, sample=SAMPLE_IDS)
     targets += expand(rules.preprocess_umap.output.metadata, sample=SAMPLE_IDS)
     targets += expand(rules.preprocess_umap.output.report, sample=SAMPLE_IDS)
+    targets += expand(rules.process_geojson.output.report, sample=SAMPLES_WITH_GEOJSON)
     targets += list(QUPATH_IMAGES)  # head QuPath image(s) for the active mode (or [])
     if RUN_LEIDEN_ANALYSIS:
         targets += expand(
diff --git a/workflow/rules/prepare_input_mer.smk b/workflow/rules/prepare_input_mer.smk
index d3829df..6403a2a 100644
--- a/workflow/rules/prepare_input_mer.smk
+++ b/workflow/rules/prepare_input_mer.smk
@@ -8,6 +8,8 @@ rule prepare_input_mer:
     input:
         qupath_meta=rules.generate_qupath_mer.output.qupath_meta,  # px<->µm mapping + p0
         background=rules.generate_qupath_mer.output.background,  # grey contract image
+        geojson=lambda wc: _find_geojson(wc.sample) or [],
+        geojson_validation=lambda wc: _geojson_validation_report(wc.sample),
     output:
         h5ad=config["contract"]["unfiltered_h5ad"],
     log:
diff --git a/workflow/rules/prepare_input_vhd.smk b/workflow/rules/prepare_input_vhd.smk
index b3950ac..f7d5134 100644
--- a/workflow/rules/prepare_input_vhd.smk
+++ b/workflow/rules/prepare_input_vhd.smk
@@ -3,6 +3,8 @@ rule prepare_input_vhd:
     input:
         sr_done=f"{OUTDIR_SR}/{{sample}}/.done",
         hires_png=rules.generate_qupath_vhd.output.qupath_image,
+        geojson=lambda wc: _find_geojson(wc.sample) or [],
+        geojson_validation=lambda wc: _geojson_validation_report(wc.sample),
     output:
         h5ad=config["contract"]["unfiltered_h5ad"],
     log:
diff --git a/workflow/rules/prepare_input_x5k.smk b/workflow/rules/prepare_input_x5k.smk
index f9bffc5..a54c9b3 100644
--- a/workflow/rules/prepare_input_x5k.smk
+++ b/workflow/rules/prepare_input_x5k.smk
@@ -7,6 +7,8 @@ rule prepare_input_x5k:
     input:
         done=rules.convert_zarr_x5k.output.done,  # zarr completion marker
         qupath_meta=rules.generate_qupath_x5k.output.qupath_meta,  # geojson px→µm scale
+        geojson=lambda wc: _find_geojson(wc.sample) or [],
+        geojson_validation=lambda wc: _geojson_validation_report(wc.sample),
     output:
         h5ad=config["contract"]["unfiltered_h5ad"],
     log:
diff --git a/workflow/rules/preprocess_umap.smk b/workflow/rules/preprocess_umap.smk
index cf057f2..654ddcf 100644
--- a/workflow/rules/preprocess_umap.smk
+++ b/workflow/rules/preprocess_umap.smk
@@ -35,7 +35,6 @@ rule preprocess_umap:
         mito_prefix=config.get("contract", {}).get("mito_prefix", ["MT-", "mt-"]),
         random_seed=RANDOM_SEED,
         use_precomputed=USE_PRECOMPUTED,
-        precomputed_metadata_dir=config.get("precomputed_metadata_dir", ""),
         external_enabled=EXTERNAL_ENABLED,
         external_column=(config.get("external_annotation", {}) or {}).get("column", ""),
         keep_unannotated=(config.get("external_annotation", {}) or {}).get(
diff --git a/workflow/rules/process_geojson.smk b/workflow/rules/process_geojson.smk
new file mode 100644
index 0000000..d98a1c8
--- /dev/null
+++ b/workflow/rules/process_geojson.smk
@@ -0,0 +1,21 @@
+rule process_geojson:
+    """Validate an optional region GeoJSON without modifying the source file."""
+    input:
+        geojson=lambda wc: _find_geojson(wc.sample),
+    output:
+        report=f"{SAMPLES_DIR}/{{sample}}/validation/geojson_validation.json",
+    log:
+        out=f"{LOGDIR}/process_geojson/{{sample}}.out",
+        err=f"{LOGDIR}/process_geojson/{{sample}}.err",
+    benchmark:
+        f"{LOGDIR}/benchmarks/process_geojson/{{sample}}.tsv"
+    conda:
+        "../envs/visiumhd.yaml"
+    threads: get_resource("process_geojson", "threads")
+    resources:
+        mem_mb=mem_mb_attempt("process_geojson"),
+        runtime=get_resource("process_geojson", "runtime"),
+    params:
+        sample_id=lambda wc: wc.sample,
+    script:
+        "../scripts/process_geojson.py"
diff --git a/workflow/rules/validate_input.smk b/workflow/rules/validate_input.smk
index f16b614..952fc18 100644
--- a/workflow/rules/validate_input.smk
+++ b/workflow/rules/validate_input.smk
@@ -31,6 +31,7 @@ rule validate_input:
     """Assert the unfiltered h5ad satisfies the core contract; record image presence."""
     input:
         h5ad=lambda wc: _CONTRACT["unfiltered_h5ad"].format(sample=wc.sample),
+        external_metadata=_external_metadata_input,
     output:
         report=f"{_OUT}/{{sample}}/validation/input_validation.json",
     log:
@@ -51,6 +52,5 @@ rule validate_input:
         mito_prefix=_CONTRACT.get("mito_prefix", ["MT-", "mt-"]),
         external_enabled=EXTERNAL_ENABLED,
         external_column=(config.get("external_annotation", {}) or {}).get("column", ""),
-        external_meta_dir=config.get("precomputed_metadata_dir", ""),
     script:
         "../scripts/validate_input.py"
diff --git a/workflow/schemas/config.schema.yaml b/workflow/schemas/config.schema.yaml
index 3dcdb33..a8f9cce 100644
--- a/workflow/schemas/config.schema.yaml
+++ b/workflow/schemas/config.schema.yaml
@@ -351,7 +351,9 @@ properties:
 
   precomputed_metadata_dir:
     type: string
-    description: Directory of precomputed metadata_{sample}.tsv (with use_precomputed_clusters).
+    description: >-
+      Directory of read-only metadata_{sample}.tsv files. Cell IDs may be stored in
+      the first column or in a cell_id column.
 
   external_annotation:
     type: ["object", "null"]
diff --git a/workflow/scripts/annotate_cells.py b/workflow/scripts/annotate_cells.py
index 719a896..9a495fd 100644
--- a/workflow/scripts/annotate_cells.py
+++ b/workflow/scripts/annotate_cells.py
@@ -32,6 +32,7 @@
     _here = os.getcwd()
 sys.path.insert(0, _here)
 from composition_barplots import composition_pair, find_niche_column
+from external_metadata import normalized_labels, optional_input_path, read_cell_metadata
 
 
 # ── Logging ──────────────────────────────────────────────────────────────────
@@ -175,7 +176,6 @@ def generate_annotation_plots(adata, annot_key, label, plots_dir, sample_id,
 DE_N_GENES          = int(snakemake.params.de_n_genes)
 USE_PRECOMPUTED     = bool(snakemake.params.use_precomputed)
 EXT_ANNOT_CFG       = snakemake.params.external_annotation
-PRECOMPUTED_DIR     = str(snakemake.params.precomputed_metadata_dir)
 ANNOTATION_COLORS   = snakemake.params.annotation_colors
 REGION_COLORS       = snakemake.params.region_colors
 DPI          = int(getattr(snakemake.params, "dpi", 300))
@@ -184,6 +184,9 @@ def generate_annotation_plots(adata, annot_key, label, plots_dir, sample_id,
 adata_path     = str(snakemake.input.adata)
 metadata_path  = str(snakemake.input.metadata)
 annot_tsv_path = str(snakemake.input.cluster_annotations)
+external_metadata_path = optional_input_path(
+    getattr(snakemake.input, "external_metadata", None)
+)
 out_adata_path = str(snakemake.output.adata_annot)
 plots_dir      = str(snakemake.output.plots_dir)
 
@@ -209,9 +212,8 @@ def generate_annotation_plots(adata, annot_key, label, plots_dir, sample_id,
 
     # Reload precomputed clusters from metadata TSV if configured
     if USE_PRECOMPUTED:
-        _ext_meta = os.path.join(PRECOMPUTED_DIR, f"metadata_{sample_id}.tsv") if PRECOMPUTED_DIR else ""
-        if _ext_meta and os.path.isfile(_ext_meta):
-            _meta_source = _ext_meta
+        if external_metadata_path:
+            _meta_source = external_metadata_path
         elif os.path.isfile(metadata_path):
             _meta_source = metadata_path
         else:
@@ -219,9 +221,7 @@ def generate_annotation_plots(adata, annot_key, label, plots_dir, sample_id,
 
         if _meta_source:
             log.info("Reloading clusters from metadata: %s", _meta_source)
-            saved = pd.read_csv(_meta_source, sep="\t", index_col=0, comment="#")
-            saved.index = saved.index.astype(str)   # obs_names are str; numeric cell ids
-                                                    # would be inferred as int and misalign
+            saved = read_cell_metadata(_meta_source)
             leiden_cols = [c for c in saved.columns if c.startswith("leiden_")]
             for col in leiden_cols:
                 if col not in adata.obs.columns:
@@ -302,29 +302,25 @@ def generate_annotation_plots(adata, annot_key, label, plots_dir, sample_id,
         ext_col = EXT_ANNOT_CFG.get("column", "")
         if not ext_col:
             raise ValueError("external_annotation.enabled is true but no 'column' is set in config.")
-        # Resolve metadata source for external annotation
-        _ext_meta = os.path.join(PRECOMPUTED_DIR, f"metadata_{sample_id}.tsv") if PRECOMPUTED_DIR else ""
-        if _ext_meta and os.path.isfile(_ext_meta):
-            _ext_source = _ext_meta
-        elif os.path.isfile(metadata_path):
-            _ext_source = metadata_path
-        else:
-            _ext_source = None
-
-        if _ext_source is None:
+        if not external_metadata_path:
             raise FileNotFoundError(
                 f"external_annotation enabled but no metadata found for sample "
-                f"'{sample_id}' (looked in precomputed_metadata_dir and {metadata_path}).")
+                f"'{sample_id}'.")
 
-        log.info("Loading external annotation from '%s' in %s …", ext_col, _ext_source)
-        saved = pd.read_csv(_ext_source, sep="\t", index_col=0, comment="#")
-        saved.index = saved.index.astype(str)       # see note above
-        if ext_col not in saved.columns:
-            raise ValueError(
-                f"external_annotation column '{ext_col}' not found in {_ext_source} "
-                f"(sample '{sample_id}'). Available columns: {list(saved.columns)}")
+        log.info(
+            "Loading external annotation from '%s' in %s …",
+            ext_col,
+            external_metadata_path,
+        )
+        saved = read_cell_metadata(
+            external_metadata_path, required_columns=(ext_col,)
+        )
+        labels = normalized_labels(saved, ext_col)
         adata.obs["cell_type_external"] = (
-            saved[ext_col].reindex(adata.obs_names).fillna("Unannotated").astype("category"))
+            labels.reindex(adata.obs_names.astype(str))
+            .fillna("Unannotated")
+            .astype("category")
+        )
         ext_enabled = True
         log.info("  External annotation: %d types (%d cells unmatched → Unannotated)",
                  adata.obs["cell_type_external"].nunique(),
diff --git a/workflow/scripts/external_metadata.py b/workflow/scripts/external_metadata.py
new file mode 100644
index 0000000..1d8b8e7
--- /dev/null
+++ b/workflow/scripts/external_metadata.py
@@ -0,0 +1,77 @@
+"""Read cell metadata without changing the user-provided TSV."""
+from __future__ import annotations
+
+from collections.abc import Iterable
+from os import PathLike
+
+import pandas as pd
+
+
+_MISSING_LABELS = {"", "na", "n/a", "nan", "none", "null", "unannotated"}
+
+
+def optional_input_path(value) -> str:
+    """Return a path from an optional Snakemake input value."""
+    if value is None:
+        return ""
+    if isinstance(value, (str, PathLike)):
+        return str(value)
+    return str(value[0]) if len(value) else ""
+
+
+def read_cell_metadata(
+    path: str,
+    *,
+    id_column: str = "cell_id",
+    required_columns: Iterable[str] = (),
+) -> pd.DataFrame:
+    """Load a TSV and use ``cell_id`` as its index when that column is present.
+
+    All input columns are retained. Empty and duplicate cell identifiers are rejected
+    because either case would make downstream joins ambiguous.
+    """
+    metadata = pd.read_csv(path, sep="\t", index_col=0, comment="#", dtype=str)
+    if metadata.empty:
+        raise ValueError(f"Cell metadata is empty: {path}")
+
+    missing_columns = [column for column in required_columns if column not in metadata]
+    if missing_columns:
+        raise ValueError(
+            f"Required column(s) {missing_columns} not found in {path}; "
+            f"available columns: {list(metadata.columns)}"
+        )
+
+    if id_column in metadata.columns:
+        cell_ids = metadata[id_column].astype("string").str.strip()
+    else:
+        cell_ids = pd.Series(metadata.index, index=metadata.index, dtype="string").str.strip()
+
+    missing_ids = cell_ids.isna() | cell_ids.eq("")
+    if bool(missing_ids.any()):
+        raise ValueError(
+            f"Cell metadata {path} contains {int(missing_ids.sum())} empty "
+            f"'{id_column}' value(s)."
+        )
+
+    duplicate_ids = cell_ids.duplicated(keep=False)
+    if bool(duplicate_ids.any()):
+        examples = cell_ids[duplicate_ids].drop_duplicates().head(5).tolist()
+        raise ValueError(
+            f"Cell metadata {path} contains {int(duplicate_ids.sum())} rows with "
+            f"duplicate '{id_column}' values (examples: {examples})."
+        )
+
+    metadata.index = pd.Index(cell_ids.astype(str), name=id_column)
+    return metadata
+
+
+def normalized_labels(metadata: pd.DataFrame, label_column: str) -> pd.Series:
+    """Return trimmed labels, representing blank/unannotated values as missing."""
+    labels = metadata[label_column].astype("string").str.strip()
+    missing = labels.isna() | labels.str.casefold().isin(_MISSING_LABELS)
+    return labels.mask(missing)
+
+
+def annotated_cell_ids(metadata: pd.DataFrame, label_column: str) -> set[str]:
+    """Return IDs whose external annotation is non-empty and not ``Unannotated``."""
+    return set(normalized_labels(metadata, label_column).dropna().index.astype(str))
diff --git a/workflow/scripts/prepare_input_ate.py b/workflow/scripts/prepare_input_ate.py
index ee1a4f5..d09af1f 100644
--- a/workflow/scripts/prepare_input_ate.py
+++ b/workflow/scripts/prepare_input_ate.py
@@ -79,8 +79,21 @@
     adata.obs["sample"] = sample_id
     if "cell_id" not in adata.obs.columns:
         adata.obs["cell_id"] = adata.obs_names.astype(str)
-    # cell_id must equal obs_names as strings — downstream joins key on it.
+    # Contract invariant: obs_names == obs['cell_id']. Every downstream join (external
+    # annotation, spatial niches, region mapping) keys on obs_names, but the SpatialData
+    # zarr table carries the real cell_id in a column while obs_names is a positional
+    # integer index — so the two silently diverge and those joins match nothing (all cells
+    # fall through to 'Unannotated'). Stringify cell_id and align obs_names to it here.
     adata.obs["cell_id"] = adata.obs["cell_id"].astype(str)
+    adata.obs_names = adata.obs["cell_id"].values
+    if not adata.obs_names.is_unique:
+        n_dup = int(adata.obs_names.duplicated().sum())
+        examples = adata.obs_names[adata.obs_names.duplicated(keep=False)].unique()[:5]
+        raise ValueError(
+            f"cell_id has {n_dup} duplicate value(s) (examples: {list(examples)}). "
+            "Cell identifiers must be unique; suffixing them would break external "
+            "metadata and region joins."
+        )
     log.info("AnnData: %d cells × %d genes (raw, unfiltered)", adata.n_obs, adata.n_vars)
 
     # ── 2b. Contract dtype hygiene ───────────────────────────────────────
diff --git a/workflow/scripts/prepare_input_x5k.py b/workflow/scripts/prepare_input_x5k.py
index 319dd27..08d67b7 100644
--- a/workflow/scripts/prepare_input_x5k.py
+++ b/workflow/scripts/prepare_input_x5k.py
@@ -59,6 +59,21 @@
     adata.obs["sample"] = sample_id
     if "cell_id" not in adata.obs.columns:
         adata.obs["cell_id"] = adata.obs_names.astype(str)
+    # Contract invariant: obs_names == obs['cell_id']. Every downstream join (external
+    # annotation, spatial niches, region mapping) keys on obs_names, but the SpatialData
+    # zarr table carries the real Xenium cell_id in a column while obs_names is a positional
+    # integer index — so the two silently diverge and those joins match nothing (all cells
+    # fall through to 'Unannotated'). Align obs_names to cell_id here.
+    adata.obs["cell_id"] = adata.obs["cell_id"].astype(str)
+    adata.obs_names = adata.obs["cell_id"].values
+    if not adata.obs_names.is_unique:
+        n_dup = int(adata.obs_names.duplicated().sum())
+        examples = adata.obs_names[adata.obs_names.duplicated(keep=False)].unique()[:5]
+        raise ValueError(
+            f"cell_id has {n_dup} duplicate value(s) (examples: {list(examples)}). "
+            "Cell identifiers must be unique; suffixing them would break external "
+            "metadata and region joins."
+        )
     log.info("AnnData: %d cells × %d genes (raw, unfiltered)", adata.n_obs, adata.n_vars)
 
     # ── 3. Embed greyscale morphology composite in uns['spatial'] ────────
diff --git a/workflow/scripts/preprocess_umap.py b/workflow/scripts/preprocess_umap.py
index a72f82d..3ba58b3 100644
--- a/workflow/scripts/preprocess_umap.py
+++ b/workflow/scripts/preprocess_umap.py
@@ -23,6 +23,13 @@
 import pandas as pd
 import scanpy as sc
 
+try:
+    _here = os.path.dirname(os.path.abspath(__file__))
+except NameError:  # very old Snakemake
+    _here = os.getcwd()
+sys.path.insert(0, _here)
+from external_metadata import annotated_cell_ids, optional_input_path, read_cell_metadata
+
 # ── Logging ──────────────────────────────────────────────────────────────────
 log_handlers = [logging.StreamHandler(sys.stderr)]
 if hasattr(snakemake, "log"):
@@ -50,7 +57,6 @@
 RES_SCAN_STEP  = float(snakemake.params.resolution_scan_step)
 RANDOM_SEED    = int(snakemake.params.random_seed)
 USE_PRECOMPUTED = bool(snakemake.params.use_precomputed)
-PRECOMPUTED_DIR = str(snakemake.params.precomputed_metadata_dir)
 REGION_COLORS   = snakemake.params.region_colors
 # Mitochondrial gene prefixes. Honour contract.mito_prefix when the rule passes it,
 # else default to human + mouse. The old hardcoded "MT-" silently disabled the
@@ -63,6 +69,7 @@
 EXTERNAL_COLUMN  = str(getattr(snakemake.params, "external_column", "") or "")
 KEEP_UNANNOTATED = bool(getattr(snakemake.params, "keep_unannotated", True))
 _ext_meta_in     = getattr(snakemake.input, "external_meta", None)
+_precomputed_meta_in = getattr(snakemake.input, "precomputed_meta", None)
 EXTERNAL_MASK    = bool(EXTERNAL_ENABLED and not KEEP_UNANNOTATED and _ext_meta_in)
 
 out_adata      = str(snakemake.output.adata)
@@ -192,13 +199,9 @@ def _range(series):
         # and SKIP the pipeline QC thresholds (the user's external labels ARE the QC
         # decision). Genes expressed in no kept cell are dropped (data hygiene, not a
         # cell-QC threshold, so PCA/normalisation stay well-behaved).
-        ext_path = str(_ext_meta_in if isinstance(_ext_meta_in, str) else _ext_meta_in[0])
-        _saved = pd.read_csv(ext_path, sep="\t", index_col=0, comment="#")
-        if EXTERNAL_COLUMN not in _saved.columns:
-            raise ValueError(f"external_annotation column '{EXTERNAL_COLUMN}' not in {ext_path}")
-        _lab = _saved[EXTERNAL_COLUMN].dropna().astype(str).str.strip()
-        _annot_bc = set(_lab[(_lab != "") & (_lab.str.lower() != "nan")
-                             & (_lab.str.lower() != "unannotated")].index.astype(str))
+        ext_path = optional_input_path(_ext_meta_in)
+        _saved = read_cell_metadata(ext_path, required_columns=(EXTERNAL_COLUMN,))
+        _annot_bc = annotated_cell_ids(_saved, EXTERNAL_COLUMN)
         _keep = adata.obs_names.astype(str).isin(_annot_bc)
         adata = adata[_keep].copy()
         sc.pp.filter_genes(adata, min_cells=1)
@@ -279,8 +282,7 @@ def _compute_all():
             log.info("  res=%.1f → %d clusters", res, adata.obs[key].nunique())
 
     if USE_PRECOMPUTED:
-        ext_meta = (os.path.join(PRECOMPUTED_DIR, f"metadata_{sample_id}.tsv")
-                    if PRECOMPUTED_DIR else "")
+        ext_meta = optional_input_path(_precomputed_meta_in)
         if ext_meta and os.path.isfile(ext_meta):
             meta_source = ext_meta
             log.info("Reloading precomputed clusters from EXTERNAL: %s", meta_source)
@@ -292,11 +294,7 @@ def _compute_all():
             log.warning("use_precomputed is true but no metadata file found. Computing fresh.")
 
         if meta_source:
-            saved_meta = pd.read_csv(meta_source, sep="\t", index_col=0, comment="#")
-            # obs_names are strings; an all-numeric cell-id index (e.g. Xenium) is inferred as
-            # int by read_csv and would misalign on reindex -> force str (as the external path
-            # above does) so the saved clusters line up with the current cells.
-            saved_meta.index = saved_meta.index.astype(str)
+            saved_meta = read_cell_metadata(meta_source)
             saved_res = [c for c in saved_meta.columns if c.startswith("leiden_")]
             for key in leiden_keys:
                 if key not in saved_meta.columns:
diff --git a/workflow/scripts/process_geojson.py b/workflow/scripts/process_geojson.py
new file mode 100644
index 0000000..6f786d9
--- /dev/null
+++ b/workflow/scripts/process_geojson.py
@@ -0,0 +1,78 @@
+"""Validate an optional region-annotation GeoJSON without modifying it."""
+import json
+import logging
+import sys
+import traceback
+from pathlib import Path
+
+import geopandas as gpd
+
+# ── Logging ──────────────────────────────────────────────────────────────────
+log_handlers = [logging.StreamHandler(sys.stderr)]
+if hasattr(snakemake, "log"):
+    if snakemake.log.out:
+        Path(snakemake.log.out).parent.mkdir(parents=True, exist_ok=True)
+        log_handlers.append(logging.FileHandler(snakemake.log.out, mode="w"))
+    if snakemake.log.err:
+        Path(snakemake.log.err).parent.mkdir(parents=True, exist_ok=True)
+        sys.stderr = open(snakemake.log.err, "w")
+logging.basicConfig(level=logging.INFO,
+                    format="%(asctime)s  %(levelname)-8s  %(message)s",
+                    datefmt="%Y-%m-%d %H:%M:%S", handlers=log_handlers)
+log = logging.getLogger("process_geojson")
+
+# ── Parameters ───────────────────────────────────────────────────────────────
+in_geojson = str(snakemake.input.geojson)
+sample_id = str(snakemake.params.sample_id)
+out_report = str(snakemake.output.report)
+
+try:
+    log.info("=" * 70)
+    log.info("Processing region GeoJSON for sample: %s", sample_id)
+    log.info("  input: %s", in_geojson)
+    log.info("=" * 70)
+
+    # ── 1. Load ────────────────────────────────────────────────────────
+    gdf = gpd.read_file(in_geojson)
+    n_total = len(gdf)
+    log.info("Loaded %d feature(s)", n_total)
+
+    if n_total == 0:
+        raise ValueError(f"'{in_geojson}' contains no features.")
+
+    # ── 2. Validate ─────────────────────────────────────────────────────
+    if "classification" not in gdf.columns:
+        raise ValueError(
+            f"'{in_geojson}' has no 'classification' property. Assign a QuPath "
+            f"class to the annotations before exporting. Columns: {list(gdf.columns)}"
+        )
+
+    missing_mask = gdf.geometry.isna()
+    n_missing = int(missing_mask.sum())
+    empty_mask = gdf.geometry.is_empty & ~missing_mask
+    n_empty = int(empty_mask.sum())
+    invalid_mask = ~gdf.geometry.is_valid & ~missing_mask & ~empty_mask
+    n_invalid = int(invalid_mask.sum())
+    if n_missing or n_empty or n_invalid:
+        raise ValueError(
+            f"'{in_geojson}' contains unusable geometries: missing={n_missing}, "
+            f"empty={n_empty}, invalid={n_invalid}. Fix them in QuPath and re-export; "
+            "the workflow does not rewrite source annotations."
+        )
+
+    # ── 3. Write success report ─────────────────────────────────────────
+    report = {
+        "sample": sample_id,
+        "passed": True,
+        "file": str(Path(in_geojson).resolve()),
+        "feature_count": n_total,
+        "geometry_types": sorted(gdf.geometry.geom_type.unique().tolist()),
+        "classification_present": True,
+    }
+    Path(out_report).parent.mkdir(parents=True, exist_ok=True)
+    Path(out_report).write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+    log.info("GeoJSON validation passed; report written to %s", out_report)
+
+except Exception:
+    log.error("FAILED for %s:\n%s", sample_id, traceback.format_exc())
+    raise
diff --git a/workflow/scripts/spatial_niches.py b/workflow/scripts/spatial_niches.py
index 19b3ae7..2143aa3 100644
--- a/workflow/scripts/spatial_niches.py
+++ b/workflow/scripts/spatial_niches.py
@@ -268,7 +268,7 @@ def _resolve_precomputed():
     # preprocess_umap already sets obs["sample"] = sample_id per cell, so we do
     # NOT pass label= (which would clash with that column). keys + index_unique
     # only disambiguate obs_names across samples.
-    adata = sc.concat(adatas, join="inner", keys=sample_ids, index_unique="-")
+    adata = sc.concat(adatas, join="inner", label="sample_batch", keys=sample_ids)
     del adatas
     log.info("Concatenated: %d cells, %d genes", adata.n_obs, adata.n_vars)
 
diff --git a/workflow/scripts/validate_input.py b/workflow/scripts/validate_input.py
index 7f659b8..acdd6b9 100644
--- a/workflow/scripts/validate_input.py
+++ b/workflow/scripts/validate_input.py
@@ -34,10 +34,16 @@
 from pathlib import Path
 
 import numpy as np
-import pandas as pd
 import scanpy as sc
 import scipy.sparse as sp
 
+try:
+    _here = os.path.dirname(os.path.abspath(__file__))
+except NameError:  # very old Snakemake
+    _here = os.getcwd()
+sys.path.insert(0, _here)
+from external_metadata import annotated_cell_ids, optional_input_path, read_cell_metadata
+
 # ── Logging ──────────────────────────────────────────────────────────────────
 log_handlers = [logging.StreamHandler(sys.stderr)]
 if hasattr(snakemake, "log"):
@@ -91,7 +97,7 @@ def _valid_image_entry(entry):
     mito_prefix = tuple(snakemake.params.mito_prefix)
     ext_enabled  = bool(getattr(snakemake.params, "external_enabled", False))
     ext_column   = str(getattr(snakemake.params, "external_column", "") or "")
-    ext_meta_dir = str(getattr(snakemake.params, "external_meta_dir", "") or "")
+    ext_path = optional_input_path(getattr(snakemake.input, "external_metadata", None))
     out_report  = str(snakemake.output.report)
 
     log.info("=" * 70)
@@ -187,32 +193,29 @@ def _valid_image_entry(entry):
     # Catch a barcode-format mismatch here rather than silently producing all-
     # "Unannotated" downstream (or 0 cells when keep_unannotated=false).
     ext_overlap = None
-    if ext_enabled and ext_column and ext_meta_dir:
-        ext_path = os.path.join(ext_meta_dir, f"metadata_{sample_id}.tsv")
-        if not os.path.isfile(ext_path):
-            errors.append(f"external_annotation enabled but metadata missing: {ext_path}")
-        else:
-            try:
-                _sv = pd.read_csv(ext_path, sep="\t", index_col=0, comment="#")
-                if ext_column not in _sv.columns:
-                    errors.append(f"external_annotation column '{ext_column}' not in {ext_path}")
-                else:
-                    _lab = _sv[ext_column].dropna().astype(str).str.strip()
-                    _bc = set(_lab[(_lab != "") & (_lab.str.lower() != "nan")
-                                   & (_lab.str.lower() != "unannotated")].index.astype(str))
-                    _match = len(_bc & set(adata.obs_names.astype(str)))
-                    frac = round(_match / len(_bc), 4) if _bc else 0.0
-                    ext_overlap = {"annotated_barcodes": len(_bc),
-                                   "matched_obs_names": _match,
-                                   "fraction_matched": frac}
-                    if _bc and _match == 0:
-                        errors.append(f"external_annotation: 0 of {len(_bc)} annotated "
-                                      f"barcodes match obs_names — check barcode formatting")
-                    elif _bc and frac < 0.5:
-                        warnings.append(f"external_annotation: only {_match}/{len(_bc)} "
-                                        f"annotated barcodes match obs_names ({frac:.0%})")
-            except Exception as e:
-                warnings.append(f"external_annotation overlap check failed: {e}")
+    if ext_enabled and ext_column:
+        try:
+            _sv = read_cell_metadata(ext_path, required_columns=(ext_column,))
+            _bc = annotated_cell_ids(_sv, ext_column)
+            _match = len(_bc & set(adata.obs_names.astype(str)))
+            frac = round(_match / len(_bc), 4) if _bc else 0.0
+            ext_overlap = {
+                "annotated_barcodes": len(_bc),
+                "matched_obs_names": _match,
+                "fraction_matched": frac,
+            }
+            if _bc and _match == 0:
+                errors.append(
+                    f"external_annotation: 0 of {len(_bc)} annotated "
+                    "barcodes match obs_names — check barcode formatting"
+                )
+            elif _bc and frac < 0.5:
+                warnings.append(
+                    f"external_annotation: only {_match}/{len(_bc)} annotated "
+                    f"barcodes match obs_names ({frac:.0%})"
+                )
+        except Exception as e:
+            errors.append(f"external_annotation metadata is invalid: {e}")
 
     # ── Verdict ──────────────────────────────────────────────────────────
     for w in warnings: