diff --git a/src/g3dt/cli/_internal/safety.py b/src/g3dt/cli/_internal/safety.py
index 13ff7ee..7ae6dec 100644
--- a/src/g3dt/cli/_internal/safety.py
+++ b/src/g3dt/cli/_internal/safety.py
@@ -33,18 +33,6 @@ def require_test_env(env: str) -> None:
         raise typer.Exit(2)
 
 
-def abort_if_prod(env: str) -> None:
-    """Hard abort for bulk operations that must never touch production."""
-    if is_prod(env):
-        typer.secho(
-            f"Refusing bulk operation against a production environment "
-            f"('{env}').",
-            fg=typer.colors.RED,
-            err=True,
-        )
-        raise typer.Exit(2)
-
-
 def confirm_destructive(action: str, target: str, env: str, assume_yes: bool) -> None:
     """Gate a destructive operation with an appropriate confirmation.
 
diff --git a/src/g3dt/cli/delete_cmds.py b/src/g3dt/cli/delete_cmds.py
index 8ea8306..aef43b6 100644
--- a/src/g3dt/cli/delete_cmds.py
+++ b/src/g3dt/cli/delete_cmds.py
@@ -12,6 +12,8 @@
 """
 from __future__ import annotations
 
+import re
+
 import typer
 
 from g3dt.cli._internal import dispatch, safety
@@ -22,6 +24,37 @@
 
 _DELETE_METADATA = "services/delete/delete_metadata.sh"
 
+#: A version token in the form the Athena ``version`` column stores it. The
+#: uploader writes ``group(1)`` of this same pattern (metadata_submitter's
+#: ``_find_version_from_path``), i.e. WITHOUT any leading ``v``. The delete
+#: query interpolates the string straight into SQL, so a ``v``-prefixed version
+#: matches zero rows and is reported as "skipped" rather than as an error — a
+#: silent no-op that reads as a clean run. Normalising here closes that.
+_VERSION_RE = re.compile(r"^v?(\d+\.\d+\.\d+)$", re.IGNORECASE)
+
+
+def _normalise_version(raw: str, where: str) -> str:
+    """Canonicalise one version token to the form stored in Athena.
+
+    ``all`` in any case becomes ``all``; ``v1.5.4`` and ``1.5.4`` both become
+    ``1.5.4``. Anything else is a usage error: the column only ever holds
+    three-part semver, so a truncated version like ``0.9`` would match nothing
+    and be counted as a skip.
+    """
+    token = raw.strip()
+    if token.lower() == "all":
+        return "all"
+    match = _VERSION_RE.match(token)
+    if not match:
+        typer.secho(
+            f"Invalid version '{raw}' {where}: expected x.y.z (e.g. 0.9.8) "
+            "or 'all'.",
+            fg=typer.colors.RED,
+            err=True,
+        )
+        raise typer.Exit(2)
+    return match.group(1)
+
 
 @app.command()
 def metadata(
@@ -54,10 +87,12 @@ def metadata(
         )
         raise typer.Exit(2)
 
+    version = _normalise_version(version, "for --version")
+
     names = [s.strip() for s in studies.split(",") if s.strip()]
     keys = [study_of(name, env).key for name in names]
     target = ",".join(keys)
-    all_versions = version.strip().lower() == "all"
+    all_versions = version == "all"
 
     if all_versions:
         # Deleting every version is the most destructive path: always prompt
diff --git a/src/g3dt/cli/indexd_cmds.py b/src/g3dt/cli/indexd_cmds.py
index cbe40b8..c0f9d4d 100644
--- a/src/g3dt/cli/indexd_cmds.py
+++ b/src/g3dt/cli/indexd_cmds.py
@@ -27,10 +27,18 @@ def register(
     dry_run: bool = typer.Option(
         False, "--dry-run", help="Scan + write file_metadata only; skip indexd."
     ),
+    force: bool = typer.Option(
+        False, "--force",
+        help="Re-register files already in the registry with the same md5.",
+    ),
     on: Target = typer.Option(Target.local, "--on", help="Run local or on ec2."),
 ) -> None:
     """Scan S3 prefixes and register the files with Gen3 indexd.
 
+    Files already registered for this study at this endpoint with an unchanged
+    md5 are skipped (each re-registration would create a new indexd revision
+    and duplicate the registry). ``--force`` re-registers everything.
+
     Examples:
       g3dt indexd register --s3-paths s3://bucket/edcad/ --study edcad --env staging
       g3dt indexd register --s3-paths s3://b/a/ --s3-paths s3://b/c/ --study edcad --env staging --on ec2
@@ -41,6 +49,8 @@ def build_args(env_name):
         a = ["--s3-paths", *s3_paths, "--study", s.key, "--env", env_name]
         if dry_run:
             a.append("--dry-run")
+        if force:
+            a.append("--force")
         return a
 
     def remote_cli(env_name):
@@ -50,6 +60,8 @@ def remote_cli(env_name):
         a += ["--study", study, "--env", env_name]
         if dry_run:
             a.append("--dry-run")
+        if force:
+            a.append("--force")
         return a
 
     dispatch.run_or_dispatch(
diff --git a/src/g3dt/cli/metadata.py b/src/g3dt/cli/metadata.py
index 8e2c460..46cd655 100644
--- a/src/g3dt/cli/metadata.py
+++ b/src/g3dt/cli/metadata.py
@@ -7,7 +7,7 @@
 
 import typer
 
-from g3dt.cli._internal import dispatch
+from g3dt.cli._internal import dispatch, safety
 from g3dt.cli._internal.dispatch import Target
 from g3dt.cli._internal.resolve import study_of
 
@@ -55,20 +55,55 @@ def upload_all(
         ..., "--studies", help="Comma-separated studies, e.g. ausdiab,caughtcad."
     ),
     env: str = typer.Option(..., "--env", "-e", help="Environment, e.g. test."),
+    allow_prod: bool = typer.Option(
+        False, "--allow-prod",
+        help="Allow bulk upload against production (typed confirmation required).",
+    ),
+    prod_confirmed: bool = typer.Option(
+        False, "--prod-confirmed", hidden=True,
+        help="Internal: set by the remote re-entry after the typed "
+        "confirmation already happened locally. Never pass by hand.",
+    ),
     on: Target = typer.Option(Target.local, "--on", help="Run local or on ec2."),
 ) -> None:
     """Upload several studies sequentially (wraps upload_all_studies.sh).
 
-    The wrapped script aborts on any 'prod' environment.
+    Production needs ``--allow-prod`` AND a typed confirmation of the env
+    name. The confirmation happens locally, before any EC2 dispatch (SSM has
+    no TTY, so a remote prompt would abort); the remote re-entry carries the
+    hidden ``--prod-confirmed`` marker instead of re-prompting, and
+    ``--allow-prod`` is forwarded so the wrapped script's own guard passes.
     """
     names = [s.strip() for s in studies.split(",") if s.strip()]
     keys = [study_of(name, env).key for name in names]
 
+    # Prod is detected on the resolved study keys as well as on --env:
+    # `--env staging --studies ausdiab_prod` is a production write.
+    if safety.is_prod(env) or any(safety.is_prod(k) for k in keys):
+        if not allow_prod:
+            typer.secho(
+                "Refusing bulk upload against a production environment. "
+                "Re-run with --allow-prod to confirm interactively.",
+                fg=typer.colors.RED,
+                err=True,
+            )
+            raise typer.Exit(2)
+        if not prod_confirmed:
+            safety.confirm_prod_strict("bulk metadata upload", env)
+
     def build_args(env_name):
-        return ["--studies", ",".join(keys), "--env", env_name]
+        a = ["--studies", ",".join(keys), "--env", env_name]
+        if allow_prod:
+            a.append("--allow-prod")
+        return a
 
     def remote_cli(env_name):
-        return ["metadata", "upload-all", "--studies", studies, "--env", env_name]
+        a = ["metadata", "upload-all", "--studies", studies, "--env", env_name]
+        if allow_prod:
+            # The typed confirmation already happened locally above; the box
+            # has no TTY, so the re-entry must not prompt again.
+            a += ["--allow-prod", "--prod-confirmed"]
+        return a
 
     dispatch.run_or_dispatch(
         on, env, _UPLOAD_ALL, build_args, "metadata-upload-all",
diff --git a/src/g3dt/indexd/indexd_registrar.py b/src/g3dt/indexd/indexd_registrar.py
index c56605e..22b35d0 100644
--- a/src/g3dt/indexd/indexd_registrar.py
+++ b/src/g3dt/indexd/indexd_registrar.py
@@ -115,6 +115,95 @@ def scan_s3_files(
 
 # ---------- indexd registration ----------
 
+def fetch_registered_files(
+    database: str,
+    table: str,
+    study_id: str,
+    indexd_endpoint: str,
+    athena_s3_output: str,
+    workgroup: str = "primary",
+    boto3_session: Optional[boto3.Session] = None,
+) -> pd.DataFrame:
+    """Return the ``file_name``/``md5`` pairs already registered for a study.
+
+    Scoped to one study and one indexd endpoint so a staging registration can
+    never mask a missing prod one (and vice versa).
+
+    A missing table means nothing has ever been registered for this
+    environment, which is a normal first-run state, not an error — an empty
+    frame is returned so the caller registers everything.
+
+    Returns
+    -------
+    pd.DataFrame
+        Columns ``file_name`` and ``md5``; empty if the table does not exist.
+    """
+    sql = f"""
+        SELECT DISTINCT file_name, md5
+        FROM "{database}"."{table}"
+        WHERE study_id = '{study_id}'
+          AND indexd_endpoint = '{indexd_endpoint}'
+    """
+    try:
+        return wr.athena.read_sql_query(
+            sql,
+            database=database,
+            ctas_approach=False,
+            workgroup=workgroup,
+            s3_output=athena_s3_output,
+            boto3_session=boto3_session,
+        )
+    except Exception as exc:
+        logger.warning(
+            "Could not read existing registrations from %s.%s (%s). "
+            "Treating every scanned file as unregistered.",
+            database, table, exc,
+        )
+        return pd.DataFrame(columns=["file_name", "md5"])
+
+
+def filter_unregistered(
+    df: pd.DataFrame,
+    registered: pd.DataFrame,
+) -> pd.DataFrame:
+    """Drop scanned files already registered with the same name and md5.
+
+    Re-submitting a file to indexd does not overwrite it: because ``baseid``
+    is derived from the filename, indexd creates a *new revision with a new
+    did* every time. Those revisions accumulate in the registry table (which
+    merges on ``did``), so an unfiltered re-run doubles the corpus and forces
+    every downstream join to de-duplicate. Skipping files whose content has
+    not changed makes a re-run a cheap no-op.
+
+    An md5 that differs from the registered one means the file genuinely
+    changed, so it is *not* skipped — that is exactly when a new revision is
+    wanted.
+
+    Parameters
+    ----------
+    df : pd.DataFrame
+        Scanned files; must have ``file_name`` and ``md5``.
+    registered : pd.DataFrame
+        Existing registrations (see :func:`fetch_registered_files`).
+
+    Returns
+    -------
+    pd.DataFrame
+        The subset of *df* still needing registration.
+    """
+    if df.empty or registered.empty:
+        return df
+
+    already = set(
+        zip(registered["file_name"].astype(str), registered["md5"].astype(str))
+    )
+    keep = [
+        (str(name), str(md5)) not in already
+        for name, md5 in zip(df["file_name"], df["md5"])
+    ]
+    return df[pd.Series(keep, index=df.index)]
+
+
 def register_files_with_indexd(
     index: Gen3Index,
     df: pd.DataFrame,
@@ -122,10 +211,12 @@ def register_files_with_indexd(
 ) -> pd.DataFrame:
     """Register files with Gen3 indexd and return results.
 
-    For each row in *df*, calls ``Gen3Index.create_record`` with the
-    file's hashes, size, S3 URL, baseid, and authz.  Re-submitting
-    the same baseid will create a new revision in indexd, making this
-    safe to re-run (idempotent at the API level).
+    For each row in *df*, calls ``Gen3Index.create_record`` with the file's
+    hashes, size, S3 URL, baseid, and authz. Registers exactly what it is
+    given: re-submitting a baseid creates a NEW revision with a new ``did``
+    (indexd never overwrites), so callers must pre-filter already-registered
+    files with :func:`fetch_registered_files` + :func:`filter_unregistered`
+    unless duplicate revisions are intended.
 
     Parameters
     ----------
diff --git a/src/g3dt/services/indexd/register_indexd.py b/src/g3dt/services/indexd/register_indexd.py
index 2a97421..59da4b3 100644
--- a/src/g3dt/services/indexd/register_indexd.py
+++ b/src/g3dt/services/indexd/register_indexd.py
@@ -30,6 +30,8 @@
 )
 from g3dt.indexd.indexd_registrar import (
     scan_s3_files,
+    fetch_registered_files,
+    filter_unregistered,
     register_files_with_indexd,
     write_to_glue,
 )
@@ -81,6 +83,12 @@ def main():
         help="Scan and write file_metadata only; skip indexd "
         "registration",
     )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Re-register files even if already recorded in the "
+        "registry with the same md5",
+    )
 
     args = parser.parse_args()
 
@@ -186,6 +194,36 @@ def main():
         logger.info("Dry run — skipping indexd registration.")
         sys.exit(0)
 
+    # --- Skip files already registered at this endpoint (same name + md5) ---
+    # Re-submitting creates a NEW indexd revision per run (baseid never
+    # overwrites), so an unfiltered re-run duplicates the corpus. A changed
+    # md5 is deliberately NOT skipped — that is when a new revision is wanted.
+    if args.force:
+        logger.info(
+            "--force given: re-registering all %d scanned file(s).",
+            len(file_df),
+        )
+    else:
+        registered = fetch_registered_files(
+            database=reg_database,
+            table=reg_table,
+            study_id=args.study,
+            indexd_endpoint=indexd_endpoint,
+            athena_s3_output=athena_s3_output,
+            workgroup=workgroup,
+            boto3_session=session,
+        )
+        before = len(file_df)
+        file_df = filter_unregistered(file_df, registered)
+        logger.info(
+            "Skipping %d / %d file(s) already registered with the same md5 "
+            "(use --force to re-register).",
+            before - len(file_df), before,
+        )
+        if file_df.empty:
+            logger.info("Nothing new to register. Done.")
+            sys.exit(0)
+
     auth = Gen3Auth(refresh_token=api_key)
     index = Gen3Index(auth)
 
diff --git a/src/g3dt/services/upload/metadata/upload_all_studies.sh b/src/g3dt/services/upload/metadata/upload_all_studies.sh
index dcfdfe1..475da11 100755
--- a/src/g3dt/services/upload/metadata/upload_all_studies.sh
+++ b/src/g3dt/services/upload/metadata/upload_all_studies.sh
@@ -5,13 +5,17 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 
 usage() {
     cat <<EOF
-Usage: $(basename "$0") --studies <comma-separated-studies> --env <environment>
+Usage: $(basename "$0") --studies <comma-separated-studies> --env <environment> [--allow-prod]
 
 Run upload_metadata.py sequentially for each study.
 
 Arguments:
-  --studies   Comma-separated list of study config keys (e.g. ausdiab_staging,caughtcad_staging)
-  --env       Environment string passed to the Python script (e.g. staging_ec2)
+  --studies     Comma-separated list of study config keys (e.g. ausdiab_staging,caughtcad_staging)
+  --env         Environment string passed to the Python script (e.g. staging_ec2)
+  --allow-prod  Permit running against a production environment. Without it any
+                'prod' in --env or --studies aborts. The g3dt CLI adds this flag
+                only after a local typed confirmation — prefer invoking via
+                'g3dt metadata upload-all' rather than passing it by hand.
 
 Run via the g3dt CLI:
   g3dt metadata upload-all \\
@@ -26,6 +30,7 @@ EOF
 # ---------- Parse arguments ----------
 STUDIES=""
 ENV=""
+ALLOW_PROD="false"
 
 while [[ $# -gt 0 ]]; do
     case "$1" in
@@ -37,6 +42,10 @@ while [[ $# -gt 0 ]]; do
             ENV="$2"
             shift 2
             ;;
+        --allow-prod)
+            ALLOW_PROD="true"
+            shift
+            ;;
         *)
             echo "ERROR: Unknown argument: $1"
             usage
@@ -51,9 +60,13 @@ fi
 
 # ---------- Production safety check ----------
 if echo "$ENV $STUDIES" | grep -qi "prod"; then
-    echo "ERROR: Production environment detected in arguments."
-    echo "       This script is not intended for production use. Aborting."
-    exit 1
+    if [[ "$ALLOW_PROD" != "true" ]]; then
+        echo "ERROR: Production environment detected in arguments."
+        echo "       Re-run with --allow-prod to permit this (the g3dt CLI adds"
+        echo "       it after a typed confirmation). Aborting."
+        exit 1
+    fi
+    echo "WARNING: --allow-prod given — running bulk upload against PRODUCTION."
 fi
 
 # ---------- Setup ----------
diff --git a/src/g3dt/upload/metadata_submitter.py b/src/g3dt/upload/metadata_submitter.py
index dc5a190..15cedaa 100644
--- a/src/g3dt/upload/metadata_submitter.py
+++ b/src/g3dt/upload/metadata_submitter.py
@@ -475,12 +475,40 @@ def get_gen3_api_key_aws_secret(secret_name: str, region_name: str, session) ->
         raise
 
 
+def commons_url_from_jwt(jwt_token: str) -> str:
+    """Derive the bare commons base URL from an API key's JWT.
+
+    The ``iss`` claim is ``https://<commons>/user``; stripping the service
+    suffix yields the base URL the key authenticates against — so the API key
+    itself selects the environment (test/staging/prod), and there is no URL
+    for an operator to get wrong.
+
+    This matters because ``gen3.auth.Gen3Auth`` silently falls back to the
+    Workspace Token Service whenever an explicitly-passed ``endpoint``
+    disagrees with the credential's issuer. WTS is not deployed on these
+    commons, so that path dies with a misleading ``502 Bad Gateway`` on
+    ``/wts/external_oidc/``. Constructing ``Gen3Auth(refresh_token=key)``
+    with NO endpoint never enters that branch — keep it structurally
+    impossible rather than merely guarded.
+    """
+    url = jwt.decode(
+        jwt_token,
+        options={"verify_signature": False},
+    ).get('iss', '')
+    if url.endswith('/user'):
+        url = url[: -len('/user')]
+    return url.rstrip('/')
+
+
 def infer_api_endpoint_from_jwt(
     jwt_token: str,
     api_version: str = 'v0',
 ) -> str:
     """
-    Extract the API endpoint URL from a JSON Web Token (JWT) credential.
+    Extract the sheepdog API endpoint URL from a JWT credential.
+
+    Delegates the base-URL derivation to :func:`commons_url_from_jwt` and
+    appends the API suffix.
 
     Args:
         jwt_token (str): The JSON Web Token (JWT) credential.
@@ -491,13 +519,7 @@ def infer_api_endpoint_from_jwt(
         str: The extracted API endpoint URL.
     """
     logger.info("Decoding JWT to extract API URL.")
-    url = jwt.decode(
-        jwt_token,
-        options={"verify_signature": False},
-    ).get('iss', '')
-    if '/user' in url:
-        url = url.split('/user')[0]
-    url = f"{url}/api/{api_version}"
+    url = f"{commons_url_from_jwt(jwt_token)}/api/{api_version}"
     logger.info("Extracted API URL from JWT: %s", url)
     return url
 
diff --git a/tests/test_cli_commands.py b/tests/test_cli_commands.py
index a48bd10..688a0be 100644
--- a/tests/test_cli_commands.py
+++ b/tests/test_cli_commands.py
@@ -175,6 +175,100 @@ def test_metadata_upload_all_resolves_each_study(mock_run, _study, _env):
     assert argv[i + 1] == "ausdiab_staging,caughtcad_staging"
 
 
+@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg)
+@patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_metadata_upload_all_prod_refused_without_allow_prod(mock_run, _study, _env):
+    """
+    Background:
+        Bulk upload against production used to be impossible (the wrapped
+        script hard-aborted on any 'prod') while showing no local guard at
+        all. The gate makes prod possible but deliberate: without
+        --allow-prod the CLI refuses before anything is dispatched.
+
+    Inputs:  --env prod, no --allow-prod
+    Expected Output: exit 2, the wrapped script is never invoked.
+    """
+    result = runner.invoke(
+        app,
+        ["metadata", "upload-all", "--studies", "ausdiab", "--env", "prod"],
+    )
+    assert result.exit_code == 2
+    mock_run.assert_not_called()
+    assert "--allow-prod" in result.output
+
+
+@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg)
+@patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_metadata_upload_all_prod_with_flag_requires_typed_confirmation(
+    mock_run, _study, _env
+):
+    """
+    Background:
+        --allow-prod alone is not enough: the operator must type the env
+        name exactly (confirm_prod_strict), locally, BEFORE any dispatch —
+        SSM has no TTY so a remote prompt could never be answered.
+
+    Inputs:  --env prod --allow-prod, then 'prod' typed at the prompt
+    Expected Output: exit 0 and the wrapped script receives --allow-prod.
+    """
+    result = runner.invoke(
+        app,
+        ["metadata", "upload-all", "--studies", "ausdiab", "--env", "prod",
+         "--allow-prod"],
+        input="prod\n",
+    )
+    assert result.exit_code == 0, result.output
+    argv = _argv(mock_run)
+    assert "--allow-prod" in argv
+    i = argv.index("--studies")
+    assert argv[i + 1] == "ausdiab_prod"
+
+
+@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg)
+@patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_metadata_upload_all_prod_mismatched_confirmation_aborts(
+    mock_run, _study, _env
+):
+    """
+    Inputs:  --env prod --allow-prod, but 'staging' typed at the prompt
+    Expected Output: non-zero exit, nothing dispatched.
+    """
+    result = runner.invoke(
+        app,
+        ["metadata", "upload-all", "--studies", "ausdiab", "--env", "prod",
+         "--allow-prod"],
+        input="staging\n",
+    )
+    assert result.exit_code != 0
+    mock_run.assert_not_called()
+
+
+@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg)
+@patch("g3dt.cli.metadata.study_of", side_effect=_study_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_metadata_upload_all_prod_study_key_trips_the_gate(mock_run, _study, _env):
+    """
+    Background:
+        The env alone does not determine where the write lands: a study key
+        can resolve to another environment's commons. `--env staging
+        --studies ausdiab_prod` is a production write and must be gated
+        exactly like --env prod.
+
+    Inputs:  --env staging with a study whose resolved key contains 'prod'
+    Expected Output: exit 2 without --allow-prod.
+    """
+    result = runner.invoke(
+        app,
+        ["metadata", "upload-all", "--studies", "ausdiab_prod",
+         "--env", "staging"],
+    )
+    assert result.exit_code == 2
+    mock_run.assert_not_called()
+
+
 @patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg)
 @patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg)
 @patch("g3dt.cli._internal.runner.run")
@@ -224,6 +318,56 @@ def test_delete_metadata_all_versions_passes_all(mock_run, _study, _env):
     assert argv[j + 1] == "all"
 
 
+@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg)
+@patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_delete_metadata_strips_leading_v_from_version(mock_run, _study, _env):
+    """
+    Background:
+        Athena stores the release version without a leading 'v' — the uploader
+        parses it out of the S3 path and keeps only group(1) of ^v?(x.y.z)$.
+        The delete query interpolates the operator's string literally, so a
+        'v'-prefixed version matches zero rows; the bulk wrapper then reports
+        exit 3 as "skipped" and the run looks clean while deleting nothing.
+        Normalising in the CLI closes that silent no-op.
+
+    Inputs:  --version V0.9.8 (case and prefix both wrong)
+    Expected Output: the wrapper receives exactly 0.9.8.
+    """
+    result = runner.invoke(
+        app,
+        ["delete", "metadata", "--studies", "ausdiab",
+         "--env", "staging", "--version", "V0.9.8", "--yes"],
+    )
+    assert result.exit_code == 0, result.output
+    argv = _argv(mock_run)
+    assert argv[argv.index("--version") + 1] == "0.9.8"
+
+
+@patch("g3dt.cli._internal.dispatch.resolve_env", side_effect=_env_cfg)
+@patch("g3dt.cli.delete_cmds.study_of", side_effect=_study_cfg)
+@patch("g3dt.cli._internal.runner.run")
+def test_delete_metadata_rejects_malformed_version(mock_run, _study, _env):
+    """
+    Background:
+        The upload path only ever writes three-part semver into the version
+        column, so a truncated version like 0.9 can never match a row. Left
+        unchecked it deletes nothing and is counted as a skip — invisible to
+        the operator. Rejecting it loudly refuses no valid input.
+
+    Inputs:  --version 0.9
+    Expected Output: exit 2, nothing dispatched.
+    """
+    result = runner.invoke(
+        app,
+        ["delete", "metadata", "--studies", "ausdiab",
+         "--env", "staging", "--version", "0.9", "--yes"],
+    )
+    assert result.exit_code == 2
+    mock_run.assert_not_called()
+    assert "Invalid version" in result.output
+
+
 @patch("g3dt.cli.k8s.env_of", side_effect=_env_cfg)
 @patch("g3dt.cli._internal.runner.run")
 def test_k8s_restart_schema_passes_env_argo_args(mock_run, _env):
diff --git a/tests/test_indexd_registrar.py b/tests/test_indexd_registrar.py
index a0c91ff..e460dcb 100644
--- a/tests/test_indexd_registrar.py
+++ b/tests/test_indexd_registrar.py
@@ -264,7 +264,6 @@ def test_register_single_file(
                      "rev": "aabb"}
         Output: DataFrame with 1 row; did="PREFIX/did-1", rev="aabb"
         """
-        mock_gen3_index.get_record.side_effect = Exception("not found")
         mock_gen3_index.create_record.return_value = {
             "did": "PREFIX/did-1",
             "baseid": single_file_df.iloc[0]["baseid"],
@@ -302,7 +301,6 @@ def test_register_multiple_files(
                 create_record returns distinct dids
         Output: DataFrame with 3 rows, each with a unique did
         """
-        mock_gen3_index.get_record.side_effect = Exception("not found")
         mock_gen3_index.create_record.side_effect = [
             {"did": f"PREFIX/did-{i}", "baseid": f"b-{i}", "rev": f"r-{i}"}
             for i in range(1, 4)
@@ -317,12 +315,13 @@ def test_register_multiple_files(
             "PREFIX/did-1", "PREFIX/did-2", "PREFIX/did-3",
         ]
 
-    def test_register_always_uploads(self, mock_gen3_index):
+    def test_register_registers_exactly_what_it_is_given(self, mock_gen3_index):
         """
-        All files should be submitted to indexd regardless of whether
-        the baseid already exists.  Re-submitting the same baseid
-        creates a new revision in indexd, so every file produces a
-        result row.
+        register_files_with_indexd submits every row it receives — the
+        already-registered skip is deliberately the CALLER's job (via
+        fetch_registered_files + filter_unregistered), because re-submitting
+        a baseid creates a new indexd revision with a new did each time.
+        This pins the function's half of that contract: no hidden filtering.
 
         Input:  2-row DataFrame; create_record succeeds for both rows
         Output: DataFrame with 2 rows, create_record called twice
@@ -399,7 +398,6 @@ def test_register_handles_api_error(
             ]
         )
 
-        mock_gen3_index.get_record.side_effect = Exception("not found")
         mock_gen3_index.create_record.side_effect = [
             Exception("API error"),
             {"did": "PREFIX/good-did", "baseid": "baseid-good", "rev": "r1"},
@@ -462,3 +460,125 @@ def test_write_calls_iceberg_correctly(self, mock_write):
             schema_evolution=False,
             boto3_session=None,
         )
+
+
+# --- Tests for the registration skip (fetch_registered_files / filter_unregistered) ---
+
+def test_filter_unregistered_skips_files_with_the_same_name_and_md5():
+    """
+    An unchanged file already in the registry is not re-registered.
+
+    Background:
+        Re-submitting a file to indexd is NOT a no-op: because baseid is
+        derived from the filename, indexd creates a new revision with a new
+        did every time, and the registry table (which merges on did) gains a
+        row per run. A re-run over an unchanged corpus therefore doubles it —
+        measured on a live deployment as 46,598 registry rows for 23,295
+        unique files. Skipping unchanged files makes a re-run cheap and keeps
+        the registry one row per file.
+
+    Inputs:  two scanned files, one already registered with an identical md5
+    Expected Output: only the unregistered file is returned.
+    """
+    scanned = pd.DataFrame([
+        {"file_name": "a.csv", "md5": "aaa"},
+        {"file_name": "b.csv", "md5": "bbb"},
+    ])
+    registered = pd.DataFrame([{"file_name": "a.csv", "md5": "aaa"}])
+
+    result = registrar.filter_unregistered(scanned, registered)
+
+    assert list(result["file_name"]) == ["b.csv"]
+
+
+def test_filter_unregistered_keeps_a_file_whose_md5_changed():
+    """
+    Changed content still gets registered.
+
+    Background:
+        The skip is keyed on name AND md5 precisely so that a genuinely
+        edited file still produces a new indexd revision. Matching on the
+        filename alone would silently strand updated content at the old did.
+
+    Inputs:  a scanned file whose md5 differs from the registered one
+    Expected Output: the file is returned for registration.
+    """
+    scanned = pd.DataFrame([{"file_name": "a.csv", "md5": "NEW"}])
+    registered = pd.DataFrame([{"file_name": "a.csv", "md5": "OLD"}])
+
+    result = registrar.filter_unregistered(scanned, registered)
+
+    assert list(result["file_name"]) == ["a.csv"]
+
+
+def test_filter_unregistered_registers_everything_on_a_first_run():
+    """
+    First-run case: nothing registered yet, so everything is kept.
+
+    Inputs:  scanned files and an empty registry frame
+    Expected Output: every scanned file is returned unchanged.
+    """
+    scanned = pd.DataFrame([
+        {"file_name": "a.csv", "md5": "aaa"},
+        {"file_name": "b.csv", "md5": "bbb"},
+    ])
+
+    result = registrar.filter_unregistered(
+        scanned, pd.DataFrame(columns=["file_name", "md5"])
+    )
+
+    assert len(result) == 2
+
+
+def test_fetch_registered_files_returns_empty_when_the_table_is_missing():
+    """
+    A missing registry table is treated as "nothing registered".
+
+    Background:
+        A brand-new environment has no indexd_registry table yet. That is a
+        normal first-run state, not an error — the scan must proceed and
+        register everything rather than aborting.
+
+    Inputs:  an Athena query that raises (table does not exist)
+    Expected Output: an empty DataFrame with the expected columns, no raise.
+    """
+    with patch(
+        "g3dt.indexd.indexd_registrar.wr.athena.read_sql_query",
+        side_effect=Exception("TABLE_NOT_FOUND"),
+    ):
+        result = registrar.fetch_registered_files(
+            database="db", table="indexd_registry", study_id="edcad",
+            indexd_endpoint="https://commons.example.org/index/index",
+            athena_s3_output="s3://out/",
+        )
+
+    assert result.empty
+    assert list(result.columns) == ["file_name", "md5"]
+
+
+def test_fetch_registered_files_scopes_to_study_and_endpoint():
+    """
+    The lookup cannot mix environments or studies.
+
+    Background:
+        indexd_registry holds every study and every commons endpoint the
+        deployment has registered against. Without both predicates, a
+        staging registration would mask a missing prod one and a prod
+        release would ship files that were never registered there.
+
+    Inputs:  study 'edcad' and a specific indexd endpoint
+    Expected Output: both appear in the WHERE clause of the query.
+    """
+    with patch(
+        "g3dt.indexd.indexd_registrar.wr.athena.read_sql_query",
+        return_value=MagicMock(),
+    ) as mock_query:
+        registrar.fetch_registered_files(
+            database="db", table="indexd_registry", study_id="edcad",
+            indexd_endpoint="https://commons.example.org/index/index",
+            athena_s3_output="s3://out/",
+        )
+
+    sql = mock_query.call_args[0][0]
+    assert "study_id = 'edcad'" in sql
+    assert "indexd_endpoint = 'https://commons.example.org/index/index'" in sql
diff --git a/tests/test_metadata_submitter.py b/tests/test_metadata_submitter.py
index ade3f21..ad02a29 100644
--- a/tests/test_metadata_submitter.py
+++ b/tests/test_metadata_submitter.py
@@ -387,3 +387,64 @@ def test_submit_metadata_specific_node_not_found(submitter_instance):
 
         with pytest.raises(ValueError, match="Node 'nonexistent_node' not found in data import order"):
             submitter_instance.submit_metadata(specific_node='nonexistent_node')
+
+
+def _fake_jwt(iss: str) -> str:
+    """Build an unsigned JWT with the given iss claim (signature unverified)."""
+    import base64
+    import json as _json
+
+    def _b64(obj):
+        return base64.urlsafe_b64encode(
+            _json.dumps(obj).encode()
+        ).decode().rstrip("=")
+
+    header = _b64({"alg": "RS256", "typ": "JWT"})
+    sig = base64.urlsafe_b64encode(b"sig").decode().rstrip("=")
+    return f"{header}.{_b64({'iss': iss})}.{sig}"
+
+
+def test_commons_url_from_jwt_strips_user_suffix():
+    """
+    Test that the commons base URL is extracted from a Gen3 API key's JWT.
+
+    Background:
+        A Gen3 API key's JWT carries an `iss` claim of the form
+        "https://<commons>/user". Every tool infers the target environment
+        from that claim — the key IS the environment selector — so a staging
+        key targets staging and a prod key targets prod without anyone
+        passing a URL. Passing a mismatched URL to the Gen3 SDK makes it
+        silently fall back to the Workspace Token Service (not deployed on
+        these commons), which fails with a confusing 502 on
+        /wts/external_oidc/.
+
+    Inputs:  JWTs whose iss ends in /user (with and without trailing slash)
+    Expected Output: the bare commons URL — no /user, no trailing slash.
+    """
+    from g3dt.upload.metadata_submitter import commons_url_from_jwt
+
+    assert commons_url_from_jwt(
+        _fake_jwt("https://staging.commons.example.org/user")
+    ) == "https://staging.commons.example.org"
+    assert commons_url_from_jwt(
+        _fake_jwt("https://commons.example.org/")
+    ) == "https://commons.example.org"
+
+
+def test_infer_api_endpoint_from_jwt_appends_api_version():
+    """
+    Test that the sheepdog API endpoint keeps its /api/<version> suffix.
+
+    Background:
+        infer_api_endpoint_from_jwt delegates the JWT decoding to
+        commons_url_from_jwt. This pins the delegation contract so a future
+        refactor cannot change what the metadata uploader submits to.
+
+    Inputs:  a JWT issued by a staging commons
+    Expected Output: "<commons>/api/v0"
+    """
+    from g3dt.upload.metadata_submitter import infer_api_endpoint_from_jwt
+
+    assert infer_api_endpoint_from_jwt(
+        _fake_jwt("https://staging.commons.example.org/user")
+    ) == "https://staging.commons.example.org/api/v0"