From da206c8c44db6b5c276df67f434a0e7177943167 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 5 May 2026 00:20:04 +0000
Subject: [PATCH 1/6] Anonymize Phase B dispatch with opaque submission_id +
 post-Phase-C reveal

Trainer-hosts (themselves miners under cross-eval) used to receive
miner_hotkey explicitly in /train and the bucket path
round_{id}/miner_{hotkey}/..., letting a hostile host selectively
sandbag specific rivals. Replace with a per-job opaque submission_id
minted by the dispatching validator at dispatch time:
  - dispatch payload field renamed miner_hotkey -> submission_id
  - artifact paths become round_{id}/submission_{sid}/...
  - TrainingMeta carries submission_id; validator resolves identity
    via in-memory job state plus published dispatch records
  - cross-validator Phase C eval reads dispatch records to assemble
    the union arch_uid -> submission_id map

After Phase C closes, validators publish the (submission_id,
miner_hotkey, miner_uid) reveal map via POST /round_submissions/reveal,
backed by a new round_submissions Postgres table (migration 004).
The dashboard (logs.py, api.py, views.py) now resolves
(round_id, miner_hotkey) -> submission_id through the reveal table
before fetching artifacts, and falls back gracefully when the reveal
hasn't landed yet.
---
 database/dashboard/api.py                     |  18 +-
 database/dashboard/logs.py                    | 129 +++++++++---
 database/dashboard/views.py                   |  20 +-
 database/migrations/004_round_submissions.sql |  38 ++++
 database/server.py                            |  94 ++++++++-
 runner/harness.py                             |   6 +-
 runner/sandbox.py                             |  11 +-
 runner/sandbox_runner.py                      |   2 +-
 runner/server.py                              |  48 +++--
 runner/uploads.py                             |  34 +--
 shared/artifacts.py                           | 194 +++++++++---------
 shared/db_client.py                           |  15 ++
 shared/pg_schema.py                           |  19 ++
 tests/test_artifacts.py                       | 111 +++++-----
 tests/test_coordinator.py                     | 135 ++++++++++++
 tests/test_dashboard.py                       |  44 +++-
 tests/test_database_server.py                 |  71 +++++++
 tests/test_generalist_server.py               |  10 +-
 validator/coordinator.py                      | 167 ++++++++++-----
 validator/evaluator.py                        |  13 +-
 validator/neuron.py                           |  95 +++++++++
 21 files changed, 979 insertions(+), 295 deletions(-)
 create mode 100644 database/migrations/004_round_submissions.sql

diff --git a/database/dashboard/api.py b/database/dashboard/api.py
index 2c4e3b0..4526fb9 100644
--- a/database/dashboard/api.py
+++ b/database/dashboard/api.py
@@ -206,8 +206,8 @@ async def loss_curve(index: int) -> dict:
     meta = _decode_jsonb(row["meta"], {}) or {}
     train, val = _aligned_train_val(meta)
     if not train and not val and row["round_id"] is not None and row["miner_hotkey"]:
-        r2_meta = log_helpers.fetch_meta(
-            state.r2, int(row["round_id"]), row["miner_hotkey"],
+        r2_meta = await log_helpers.fetch_meta(
+            state.pool, state.r2, int(row["round_id"]), row["miner_hotkey"],
         )
         if r2_meta:
             train, val = _aligned_train_val(r2_meta)
@@ -741,11 +741,13 @@ async def logs_meta_json(round_id: int, hotkey: str):
 async def logs_stdout_json(round_id: int, hotkey: str, direct: int = 0):
     state = get_state()
     if direct:
-        url = log_helpers.presigned_stdout_url(state.r2, round_id, hotkey)
+        url = await log_helpers.presigned_stdout_url(
+            state.pool, state.r2, round_id, hotkey,
+        )
         if not url:
             raise HTTPException(status_code=503, detail="R2 presign unavailable")
         return RedirectResponse(url=url, status_code=302)
-    payload = log_helpers.fetch_stdout(state.r2, round_id, hotkey)
+    payload = await log_helpers.fetch_stdout(state.pool, state.r2, round_id, hotkey)
     if payload is None:
         raise HTTPException(status_code=404, detail="No stdout.log in R2")
     return JSONResponse(payload)
@@ -755,11 +757,15 @@ async def logs_stdout_json(round_id: int, hotkey: str, direct: int = 0):
 async def logs_architecture_json(round_id: int, hotkey: str, direct: int = 0):
     state = get_state()
     if direct:
-        url = log_helpers.presigned_architecture_url(state.r2, round_id, hotkey)
+        url = await log_helpers.presigned_architecture_url(
+            state.pool, state.r2, round_id, hotkey,
+        )
         if not url:
             raise HTTPException(status_code=503, detail="R2 presign unavailable")
         return RedirectResponse(url=url, status_code=302)
-    payload = log_helpers.fetch_architecture(state.r2, round_id, hotkey)
+    payload = await log_helpers.fetch_architecture(
+        state.pool, state.r2, round_id, hotkey,
+    )
     if payload is None:
         raise HTTPException(status_code=404, detail="No architecture.py in R2")
     return JSONResponse(payload)
diff --git a/database/dashboard/logs.py b/database/dashboard/logs.py
index 05f319a..71b3a29 100644
--- a/database/dashboard/logs.py
+++ b/database/dashboard/logs.py
@@ -3,6 +3,12 @@
 Wraps ``shared.r2_audit.R2AuditLog`` with a byte cap so the dashboard
 never streams pathologically large stdout blobs over HTTP. Callers that
 want the raw file use ``?direct=1`` to get a short-lived presigned URL.
+
+Artifacts live at ``round_{id}/submission_{sid}/...`` — the bucket path
+hides miner identities from the trainer-host during Phase B. The
+dashboard resolves ``(round_id, miner_hotkey) → submission_id`` via the
+``round_submissions`` reveal table that the validator populates after
+Phase C closes.
 """
 
 from __future__ import annotations
@@ -15,26 +21,65 @@
 logger = logging.getLogger(__name__)
 
 
-def _stdout_key(round_id: int, hotkey: str) -> str:
-    return f"round_{round_id}/miner_{hotkey}/stdout.log"
+def _stdout_key_sid(round_id: int, submission_id: str) -> str:
+    return f"round_{round_id}/submission_{submission_id}/stdout.log"
+
+
+def _meta_key_sid(round_id: int, submission_id: str) -> str:
+    return f"round_{round_id}/submission_{submission_id}/training_meta.json"
+
 
+def _arch_key_sid(round_id: int, submission_id: str) -> str:
+    return f"round_{round_id}/submission_{submission_id}/architecture.py"
 
-def _meta_key(round_id: int, hotkey: str) -> str:
-    return f"round_{round_id}/miner_{hotkey}/training_meta.json"
 
+async def submission_id_for(
+    pool, round_id: int, hotkey: str,
+) -> Optional[str]:
+    """Look up the opaque submission_id for ``(round_id, miner_hotkey)``.
+
+    Returns None when the validator hasn't yet published the reveal map
+    (Phase C still in progress, or this validator never dispatched a
+    job for that miner this round).
+    """
+    if pool is None or not hotkey:
+        return None
+    try:
+        row = await pool.fetchrow(
+            """
+            SELECT submission_id FROM round_submissions
+            WHERE round_id = $1 AND miner_hotkey = $2
+            ORDER BY created_at DESC LIMIT 1
+            """,
+            int(round_id), hotkey,
+        )
+    except Exception:
+        logger.exception(
+            "submission_id lookup failed: round=%s hotkey=%s",
+            round_id, hotkey[:16],
+        )
+        return None
+    return row["submission_id"] if row else None
 
-def _arch_key(round_id: int, hotkey: str) -> str:
-    return f"round_{round_id}/miner_{hotkey}/architecture.py"
 
+async def fetch_meta(pool, r2, round_id: int, hotkey: str) -> Optional[dict]:
+    """Return ``training_meta.json`` for a (round, miner) or None if missing.
 
-def fetch_meta(r2, round_id: int, hotkey: str) -> Optional[dict]:
-    """Return ``training_meta.json`` for a (round, miner) or None if missing."""
+    Resolves the bucket path through the ``round_submissions`` reveal
+    table. Returns None when the reveal hasn't been published yet.
+    """
     if r2 is None:
         return None
+    sid = await submission_id_for(pool, round_id, hotkey)
+    if not sid:
+        return None
     try:
-        return r2.download_json(_meta_key(round_id, hotkey))
+        return r2.download_json(_meta_key_sid(round_id, sid))
     except Exception:
-        logger.exception("fetch_meta failed: round=%s hotkey=%s", round_id, hotkey[:16])
+        logger.exception(
+            "fetch_meta failed: round=%s hotkey=%s sid=%s",
+            round_id, hotkey[:16], sid[:12],
+        )
         return None
 
 
@@ -70,54 +115,64 @@ async def fetch_meta_cached_or_r2(
     meta = await cached_meta(pool, round_id, hotkey)
     if meta is not None:
         return meta
-    return fetch_meta(r2, round_id, hotkey)
+    return await fetch_meta(pool, r2, round_id, hotkey)
 
 
-def fetch_stdout(
+async def fetch_stdout(
+    pool,
     r2,
     round_id: int,
     hotkey: str,
     max_bytes: Optional[int] = None,
 ) -> Optional[dict]:
-    """Return ``{text, truncated, size}`` for the stdout file, or None.
-
-    ``text`` is capped at ``max_bytes`` (defaults to
-    ``Config.DASHBOARD_MAX_LOG_BYTES``). If the file is larger we take the
-    tail so operators see the most recent lines.
-    """
+    """Return ``{text, truncated, size}`` for the stdout file, or None."""
     if r2 is None:
         return None
+    sid = await submission_id_for(pool, round_id, hotkey)
+    if not sid:
+        return None
     cap = max_bytes if max_bytes is not None else Config.DASHBOARD_MAX_LOG_BYTES
     try:
-        raw = r2.download_text(_stdout_key(round_id, hotkey))
+        raw = r2.download_text(_stdout_key_sid(round_id, sid))
     except Exception:
-        logger.exception("fetch_stdout failed: round=%s hotkey=%s", round_id, hotkey[:16])
+        logger.exception(
+            "fetch_stdout failed: round=%s hotkey=%s sid=%s",
+            round_id, hotkey[:16], sid[:12],
+        )
         return None
     if raw is None:
         return None
     data = raw.encode("utf-8", errors="replace")
     size = len(data)
     if size > cap:
-        # Keep the tail — most recent lines are the useful ones
         tail = data[-cap:].decode("utf-8", errors="replace")
         return {"text": tail, "truncated": True, "size": size}
     return {"text": raw, "truncated": False, "size": size}
 
 
-def presigned_stdout_url(r2, round_id: int, hotkey: str, ttl: int = 900) -> str:
+async def presigned_stdout_url(
+    pool, r2, round_id: int, hotkey: str, ttl: int = 900,
+) -> str:
     """Return a 15-minute presigned GET URL for the stdout file."""
     if r2 is None:
         return ""
+    sid = await submission_id_for(pool, round_id, hotkey)
+    if not sid:
+        return ""
     try:
-        return r2.generate_presigned_get_url(_stdout_key(round_id, hotkey), ttl=ttl)
+        return r2.generate_presigned_get_url(
+            _stdout_key_sid(round_id, sid), ttl=ttl,
+        )
     except Exception:
         logger.exception(
-            "presigned_stdout_url failed: round=%s hotkey=%s", round_id, hotkey[:16],
+            "presigned_stdout_url failed: round=%s hotkey=%s sid=%s",
+            round_id, hotkey[:16], sid[:12],
         )
         return ""
 
 
-def fetch_architecture(
+async def fetch_architecture(
+    pool,
     r2,
     round_id: int,
     hotkey: str,
@@ -126,12 +181,16 @@ def fetch_architecture(
     """Return ``{text, truncated, size}`` for architecture.py, or None."""
     if r2 is None:
         return None
+    sid = await submission_id_for(pool, round_id, hotkey)
+    if not sid:
+        return None
     cap = max_bytes if max_bytes is not None else Config.DASHBOARD_MAX_LOG_BYTES
     try:
-        raw = r2.download_text(_arch_key(round_id, hotkey))
+        raw = r2.download_text(_arch_key_sid(round_id, sid))
     except Exception:
         logger.exception(
-            "fetch_architecture failed: round=%s hotkey=%s", round_id, hotkey[:16],
+            "fetch_architecture failed: round=%s hotkey=%s sid=%s",
+            round_id, hotkey[:16], sid[:12],
         )
         return None
     if raw is None:
@@ -144,21 +203,29 @@ def fetch_architecture(
     return {"text": raw, "truncated": False, "size": size}
 
 
-def presigned_architecture_url(r2, round_id: int, hotkey: str, ttl: int = 900) -> str:
+async def presigned_architecture_url(
+    pool, r2, round_id: int, hotkey: str, ttl: int = 900,
+) -> str:
     """Return a 15-minute presigned GET URL for architecture.py."""
     if r2 is None:
         return ""
+    sid = await submission_id_for(pool, round_id, hotkey)
+    if not sid:
+        return ""
     try:
-        return r2.generate_presigned_get_url(_arch_key(round_id, hotkey), ttl=ttl)
+        return r2.generate_presigned_get_url(
+            _arch_key_sid(round_id, sid), ttl=ttl,
+        )
     except Exception:
         logger.exception(
-            "presigned_architecture_url failed: round=%s hotkey=%s",
-            round_id, hotkey[:16],
+            "presigned_architecture_url failed: round=%s hotkey=%s sid=%s",
+            round_id, hotkey[:16], sid[:12],
         )
         return ""
 
 
 __all__ = [
+    "submission_id_for",
     "cached_meta",
     "fetch_architecture",
     "fetch_meta",
diff --git a/database/dashboard/views.py b/database/dashboard/views.py
index bb09ffd..9210cc2 100644
--- a/database/dashboard/views.py
+++ b/database/dashboard/views.py
@@ -321,8 +321,10 @@ async def logs_view(request: Request, round_id: int, hotkey: str):
     meta = await log_helpers.fetch_meta_cached_or_r2(
         state.pool, state.r2, round_id, hotkey,
     )
-    stdout = log_helpers.fetch_stdout(state.r2, round_id, hotkey)
-    architecture = log_helpers.fetch_architecture(state.r2, round_id, hotkey)
+    stdout = await log_helpers.fetch_stdout(state.pool, state.r2, round_id, hotkey)
+    architecture = await log_helpers.fetch_architecture(
+        state.pool, state.r2, round_id, hotkey,
+    )
     return _html(
         "logs.html",
         request,
@@ -354,11 +356,13 @@ async def logs_meta(round_id: int, hotkey: str):
 async def logs_stdout(round_id: int, hotkey: str, direct: int = 0):
     state = get_state()
     if direct:
-        url = log_helpers.presigned_stdout_url(state.r2, round_id, hotkey)
+        url = await log_helpers.presigned_stdout_url(
+            state.pool, state.r2, round_id, hotkey,
+        )
         if not url:
             raise HTTPException(status_code=503, detail="R2 presign unavailable")
         return RedirectResponse(url=url, status_code=302)
-    payload = log_helpers.fetch_stdout(state.r2, round_id, hotkey)
+    payload = await log_helpers.fetch_stdout(state.pool, state.r2, round_id, hotkey)
     if payload is None:
         raise HTTPException(status_code=404, detail="No stdout.log in R2")
     return JSONResponse(payload)
@@ -368,11 +372,15 @@ async def logs_stdout(round_id: int, hotkey: str, direct: int = 0):
 async def logs_architecture(round_id: int, hotkey: str, direct: int = 0):
     state = get_state()
     if direct:
-        url = log_helpers.presigned_architecture_url(state.r2, round_id, hotkey)
+        url = await log_helpers.presigned_architecture_url(
+            state.pool, state.r2, round_id, hotkey,
+        )
         if not url:
             raise HTTPException(status_code=503, detail="R2 presign unavailable")
         return RedirectResponse(url=url, status_code=302)
-    payload = log_helpers.fetch_architecture(state.r2, round_id, hotkey)
+    payload = await log_helpers.fetch_architecture(
+        state.pool, state.r2, round_id, hotkey,
+    )
     if payload is None:
         raise HTTPException(status_code=404, detail="No architecture.py in R2")
     return JSONResponse(payload)
diff --git a/database/migrations/004_round_submissions.sql b/database/migrations/004_round_submissions.sql
new file mode 100644
index 0000000..2d46d54
--- /dev/null
+++ b/database/migrations/004_round_submissions.sql
@@ -0,0 +1,38 @@
+-- Per-round opaque submission_id -> miner_hotkey reveal map.
+--
+-- Phase B dispatch now hides the architecture owner's hotkey from the
+-- trainer-host (which is itself another miner's pod under cross-eval).
+-- The validator mints a per-job ``submission_id`` (32 hex chars) at
+-- dispatch time and uses it as the bucket path component:
+--     round_{id}/submission_{sid}/checkpoint.safetensors
+--     round_{id}/submission_{sid}/architecture.py
+--     round_{id}/submission_{sid}/training_meta.json
+--     round_{id}/submission_{sid}/stdout.log
+-- Trainers never see the architecture owner's hotkey on the wire.
+--
+-- After Phase C closes, the dispatching validator POSTs the reveal map
+-- to ``/round_submissions/reveal``. This table stores it so:
+--   * the public dashboard can resolve historical artifact paths back
+--     to the (miner_hotkey, miner_uid) pair for log/loss views;
+--   * miners can independently confirm their own submission_id was the
+--     one trained for any past round.
+--
+-- Idempotent upsert on (round_id, submission_id): repeated submissions
+-- from the same dispatching validator are no-ops, and another
+-- dispatcher can append entries for jobs it owned without conflict.
+--
+-- Bootstrap-order safety: matches the pattern in 003.
+
+CREATE TABLE IF NOT EXISTS round_submissions (
+    round_id BIGINT NOT NULL,
+    submission_id TEXT NOT NULL,
+    miner_hotkey TEXT NOT NULL,
+    miner_uid INTEGER NOT NULL DEFAULT -1,
+    created_at DOUBLE PRECISION NOT NULL DEFAULT extract(epoch from now()),
+    PRIMARY KEY (round_id, submission_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_rs_round_hotkey
+    ON round_submissions(round_id, miner_hotkey);
+CREATE INDEX IF NOT EXISTS idx_rs_hotkey
+    ON round_submissions(miner_hotkey);
diff --git a/database/server.py b/database/server.py
index e37b1ad..30164f8 100644
--- a/database/server.py
+++ b/database/server.py
@@ -246,10 +246,22 @@ class SubmitTrainingMetaRequest(BaseModel):
     meta: dict
 
 
+class SubmissionRevealEntry(BaseModel):
+    submission_id: str
+    miner_hotkey: str
+    miner_uid: int = -1
+
+
+class SubmissionRevealRequest(BaseModel):
+    """Post-Phase-C reveal of submission_id -> miner_hotkey for a round."""
+    round_id: int
+    entries: list[SubmissionRevealEntry]
+
+
 # Routes that only validators (hotkeys with permit) may call
 _VALIDATOR_ONLY_PREFIXES = (
     "/experiments/add", "/frontier/update", "/provenance/record",
-    "/training_metas",
+    "/training_metas", "/round_submissions/reveal",
 )
 
 
@@ -300,6 +312,7 @@ async def auth_middleware(request: Request, call_next):
         or path.startswith("/agent_code")
         or path.startswith("/desearch")
         or path.startswith("/llm")
+        or path.startswith("/round_submissions")
     )
     if needs_auth:
         # ── Reject oversized bodies before reading fully ──
@@ -638,6 +651,85 @@ async def submit_training_meta(req: SubmitTrainingMetaRequest):
     return {"status": "ok"}
 
 
+@validator_router.post("/round_submissions/reveal")
+async def submit_submission_reveal(req: SubmissionRevealRequest):
+    """Validator publishes the per-round submission_id -> miner_hotkey map.
+
+    Idempotent — every entry upserts on (round_id, submission_id), so two
+    validators that both happen to dispatch the same UID (e.g. fallback
+    paths) won't conflict. Only the dispatching validator owns the truth
+    for its submission_ids; another dispatcher's entries are appended
+    without overwrite.
+    """
+    if _pool is None:
+        raise HTTPException(status_code=503, detail="DB pool not configured")
+    if not req.entries:
+        return {"status": "ok", "inserted": 0}
+
+    # Defensive cap so a misbehaving caller can't bloat the table.
+    if len(req.entries) > 1024:
+        raise HTTPException(status_code=400, detail="Too many reveal entries")
+
+    rows = [
+        (int(req.round_id), e.submission_id, e.miner_hotkey, int(e.miner_uid))
+        for e in req.entries
+        if e.submission_id and e.miner_hotkey
+    ]
+    if not rows:
+        return {"status": "ok", "inserted": 0}
+
+    try:
+        async with _pool.acquire() as conn:
+            await conn.executemany(
+                """
+                INSERT INTO round_submissions
+                    (round_id, submission_id, miner_hotkey, miner_uid)
+                VALUES ($1, $2, $3, $4)
+                ON CONFLICT (round_id, submission_id) DO UPDATE SET
+                    miner_hotkey = EXCLUDED.miner_hotkey,
+                    miner_uid    = EXCLUDED.miner_uid
+                """,
+                rows,
+            )
+    except Exception as e:
+        logger.error("submission reveal write failed: round=%s err=%s", req.round_id, e)
+        raise HTTPException(status_code=500, detail="Failed to persist reveal")
+    return {"status": "ok", "inserted": len(rows)}
+
+
+@validator_router.get("/round_submissions/{round_id}")
+async def get_round_submissions(round_id: int):
+    """Return the reveal map for a round.
+
+    Validator-authed (matches the rest of the validator surface). The
+    public dashboard reads this through ``/dashboard/api/...`` helpers
+    which resolve via the same Postgres rows server-side.
+    """
+    if _pool is None:
+        raise HTTPException(status_code=503, detail="DB pool not configured")
+    rows = await _pool.fetch(
+        """
+        SELECT submission_id, miner_hotkey, miner_uid, created_at
+        FROM round_submissions
+        WHERE round_id = $1
+        ORDER BY created_at ASC
+        """,
+        int(round_id),
+    )
+    return {
+        "round_id": int(round_id),
+        "entries": [
+            {
+                "submission_id": r["submission_id"],
+                "miner_hotkey": r["miner_hotkey"],
+                "miner_uid": int(r["miner_uid"]),
+                "created_at": float(r["created_at"]),
+            }
+            for r in rows
+        ],
+    }
+
+
 @validator_router.post("/frontier/update")
 async def update_frontier(req: UpdateFrontierRequest):
     """Validator pushes frontier data."""
diff --git a/runner/harness.py b/runner/harness.py
index 455bd5a..b7e6e7e 100644
--- a/runner/harness.py
+++ b/runner/harness.py
@@ -117,7 +117,7 @@ class TrainingConfig:
     round_id: int = 0
     min_flops: int = 0
     max_flops: int = 0
-    miner_hotkey: str = "unknown"
+    submission_id: str = "unknown"
     time_budget: int = 300
 
     @classmethod
@@ -222,7 +222,7 @@ def run_training(runner: TaskRunner, architecture_code: str, config: TrainingCon
 
     return {
         "round_id": config.round_id,
-        "miner_hotkey": config.miner_hotkey,
+        "submission_id": config.submission_id,
         "status": "success",
         "flops_equivalent_size": flops_equiv,
         "training_time_seconds": time.time() - start,
@@ -733,7 +733,7 @@ def _check_size_gate(config: TrainingConfig, flops_equiv: int) -> dict | None:
 def _fail(config: TrainingConfig, status: str, error: str, **kw) -> dict:
     return {
         "round_id": config.round_id,
-        "miner_hotkey": config.miner_hotkey,
+        "submission_id": config.submission_id,
         "status": status,
         "error": error,
         **kw,
diff --git a/runner/sandbox.py b/runner/sandbox.py
index 4524f46..34789f3 100644
--- a/runner/sandbox.py
+++ b/runner/sandbox.py
@@ -236,8 +236,9 @@ async def run_sandbox(
     cmd = _build_command(config_path)
 
     logger.info(
-        "Spawning sandbox: cmd=%s timeout=%ds (round=%s miner=%s)",
-        cmd[0], timeout, config.get("round_id"), config.get("miner_hotkey", "?")[:16],
+        "Spawning sandbox: cmd=%s timeout=%ds (round=%s submission=%s)",
+        cmd[0], timeout, config.get("round_id"),
+        config.get("submission_id", "?")[:12],
     )
 
     try:
@@ -309,7 +310,7 @@ async def _drain(stream, cap: int, label: str) -> bytes:
         return (
             {
                 "round_id": config.get("round_id", 0),
-                "miner_hotkey": config.get("miner_hotkey", "unknown"),
+                "submission_id": config.get("submission_id", "unknown"),
                 "status": "failed",
                 "error": f"sandbox timed out after {timeout}s",
             },
@@ -320,7 +321,7 @@ async def _drain(stream, cap: int, label: str) -> bytes:
         return (
             {
                 "round_id": config.get("round_id", 0),
-                "miner_hotkey": config.get("miner_hotkey", "unknown"),
+                "submission_id": config.get("submission_id", "unknown"),
                 "status": "failed",
                 "error": "sandbox killed: " + ", ".join(sorted(overflow.values())),
             },
@@ -333,7 +334,7 @@ async def _drain(stream, cap: int, label: str) -> bytes:
         return (
             {
                 "round_id": config.get("round_id", 0),
-                "miner_hotkey": config.get("miner_hotkey", "unknown"),
+                "submission_id": config.get("submission_id", "unknown"),
                 "status": "failed",
                 "error": (
                     f"sandbox produced no JSON result (exit={proc.returncode}); "
diff --git a/runner/sandbox_runner.py b/runner/sandbox_runner.py
index 78914e2..9e567b7 100644
--- a/runner/sandbox_runner.py
+++ b/runner/sandbox_runner.py
@@ -239,7 +239,7 @@ def main() -> int:
     except Exception as e:
         _emit({
             "round_id": tc.round_id,
-            "miner_hotkey": tc.miner_hotkey,
+            "submission_id": tc.submission_id,
             "status": "failed",
             "error": f"sandbox: harness raised {type(e).__name__}: {e}",
         })
diff --git a/runner/server.py b/runner/server.py
index badc4be..5861a62 100644
--- a/runner/server.py
+++ b/runner/server.py
@@ -162,7 +162,9 @@ async def train(request: Request):
         "round_id": data.get("round_id", 0),
         "min_flops": data.get("min_flops_equivalent", 0),
         "max_flops": data.get("max_flops_equivalent", 0),
-        "miner_hotkey": data.get("miner_hotkey", "unknown"),
+        # Opaque submission_id from the dispatching validator; the
+        # trainer-host never sees the architecture owner's hotkey.
+        "submission_id": data.get("submission_id", "unknown"),
         "time_budget": data.get("time_budget", 300),
         "task_name": task_name,
     }
@@ -179,8 +181,8 @@ async def train(request: Request):
     ))
 
     logger.info(
-        "Training job accepted: round=%d miner=%s task=%s",
-        training_config["round_id"], training_config["miner_hotkey"], task_name,
+        "Training job accepted: round=%d submission=%s task=%s",
+        training_config["round_id"], training_config["submission_id"][:12], task_name,
     )
     return JSONResponse(
         status_code=202,
@@ -207,10 +209,10 @@ async def _train_and_upload(
     credentials and no network-capable Python imports.
     """
     round_id = training_config["round_id"]
-    miner_hotkey = training_config["miner_hotkey"]
+    submission_id = training_config["submission_id"]
     logger.info(
-        "Starting train+upload: round=%d miner=%s upload_url_keys=%s",
-        round_id, miner_hotkey, sorted(upload_urls.keys()) if upload_urls else "(none)",
+        "Starting train+upload: round=%d submission=%s upload_url_keys=%s",
+        round_id, submission_id[:12], sorted(upload_urls.keys()) if upload_urls else "(none)",
     )
     try:
         # ── Prefetch data with the parent's network credentials ──
@@ -250,41 +252,45 @@ async def _train_and_upload(
         if sandbox_log:
             tail = sandbox_log[-4096:]
             logger.info(
-                "Sandbox stderr tail (round %d miner %s):\n%s",
-                round_id, miner_hotkey, tail,
+                "Sandbox stderr tail (round %d submission %s):\n%s",
+                round_id, submission_id[:12], tail,
             )
 
         logger.info(
-            "Training complete: round=%d miner=%s status=%s elapsed=%.1fs",
-            round_id, miner_hotkey, result.get("status", "?"), elapsed,
+            "Training complete: round=%d submission=%s status=%s elapsed=%.1fs",
+            round_id, submission_id[:12], result.get("status", "?"), elapsed,
         )
 
         if result.get("status") in ("build_failed", "size_violation", "failed", "timeout"):
             logger.warning(
-                "Training failed for round %d miner %s: %s — %s",
-                round_id, miner_hotkey, result.get("status"), result.get("error", ""),
+                "Training failed for round %d submission %s: %s — %s",
+                round_id, submission_id[:12], result.get("status"), result.get("error", ""),
             )
-            # Upload a failure meta so the validator knows what happened
-            upload_failure_meta(round_id, miner_hotkey, upload_urls, result)
+            upload_failure_meta(round_id, submission_id, upload_urls, result)
             return
 
-        # Upload artifacts to R2
-        upload_artifacts(result, architecture_code, round_id, miner_hotkey, upload_urls)
+        upload_artifacts(result, architecture_code, round_id, submission_id, upload_urls)
     except Exception as e:
         logger.error(
-            "Background train+upload failed for round %d miner %s: %s",
-            round_id, miner_hotkey, e, exc_info=True,
+            "Background train+upload failed for round %d submission %s: %s",
+            round_id, submission_id[:12], e, exc_info=True,
         )
         try:
-            upload_failure_meta(round_id, miner_hotkey, upload_urls, {
+            upload_failure_meta(round_id, submission_id, upload_urls, {
                 "status": "failed",
                 "error": f"Unhandled exception: {e}",
             })
         except Exception:
-            logger.error("Failed to upload failure meta for round %d miner %s", round_id, miner_hotkey)
+            logger.error(
+                "Failed to upload failure meta for round %d submission %s",
+                round_id, submission_id[:12],
+            )
     finally:
         _train_semaphore.release()
-        logger.info("Training semaphore released (round %d miner %s)", round_id, miner_hotkey)
+        logger.info(
+            "Training semaphore released (round %d submission %s)",
+            round_id, submission_id[:12],
+        )
 
 
 @app.get("/health")
diff --git a/runner/uploads.py b/runner/uploads.py
index c05cfbb..c0729c4 100644
--- a/runner/uploads.py
+++ b/runner/uploads.py
@@ -1,8 +1,13 @@
 """Artifact upload helpers for the trainer server.
 
 Split out of ``runner/server.py`` so the request-handling module stays
-under the 300-line file cap.  All functions assume the caller already
+under the 300-line file cap. All functions assume the caller already
 prefetched data and the sandbox already produced a ``result`` dict.
+
+The trainer never sees the architecture owner's hotkey — only the opaque
+``submission_id`` minted by the dispatching validator. Bucket paths and
+TrainingMeta live entirely in submission_id space until the validator
+publishes the post-Phase-C reveal map.
 """
 
 from __future__ import annotations
@@ -26,7 +31,7 @@ def _get_r2():
 
 def upload_failure_meta(
     round_id: int,
-    miner_hotkey: str,
+    submission_id: str,
     upload_urls: dict,
     result: dict,
 ) -> None:
@@ -35,7 +40,7 @@ def upload_failure_meta(
 
     meta = TrainingMeta(
         round_id=round_id,
-        miner_hotkey=miner_hotkey,
+        submission_id=submission_id,
         status=result.get("status", "failed"),
         error=result.get("error", ""),
         flops_equivalent_size=result.get("flops_equivalent_size", 0),
@@ -49,7 +54,10 @@ def upload_failure_meta(
         body = json.dumps(meta.to_dict(), indent=2).encode()
         resp = httpx.put(meta_url, content=body, timeout=30)
         resp.raise_for_status()
-        logger.info("Uploaded failure meta for round %d miner %s", round_id, miner_hotkey)
+        logger.info(
+            "Uploaded failure meta for round %d submission %s",
+            round_id, submission_id[:12],
+        )
     except Exception as e:
         logger.error("Failed to upload failure meta: %s", e)
 
@@ -58,22 +66,22 @@ def upload_artifacts(
     result: dict,
     architecture_code: str,
     round_id: int,
-    miner_hotkey: str,
+    submission_id: str,
     upload_urls: dict,
 ) -> dict:
-    """Upload checkpoint + architecture + meta to R2.  Mutates and returns ``result``."""
+    """Upload checkpoint + architecture + meta. Mutates and returns ``result``."""
     from shared.artifacts import (
         TrainingMeta, checkpoint_key as ck_fn, architecture_key as ak_fn,
     )
 
     checkpoint_path = result.get("checkpoint_path", "")
     stdout_log = result.get("stdout_log", "")
-    ck_str = ck_fn(round_id, miner_hotkey)
-    ak_str = ak_fn(round_id, miner_hotkey)
+    ck_str = ck_fn(round_id, submission_id)
+    ak_str = ak_fn(round_id, submission_id)
 
     meta = TrainingMeta(
         round_id=round_id,
-        miner_hotkey=miner_hotkey,
+        submission_id=submission_id,
         status="success",
         flops_equivalent_size=result.get("flops_equivalent_size", 0),
         training_time_seconds=result.get("training_time_seconds", 0),
@@ -108,7 +116,7 @@ def upload_artifacts(
             upload_ok = upload_training_artifacts(
                 r2=_get_r2(),
                 round_id=round_id,
-                miner_hotkey=miner_hotkey,
+                submission_id=submission_id,
                 checkpoint_path=checkpoint_path,
                 architecture_code=architecture_code,
                 stdout_log=stdout_log,
@@ -120,9 +128,11 @@ def upload_artifacts(
 
     status = "success" if upload_ok else "upload_failed"
     if not upload_ok:
-        logger.error("Artifact upload incomplete for round %d miner %s", round_id, miner_hotkey)
+        logger.error(
+            "Artifact upload incomplete for round %d submission %s",
+            round_id, submission_id[:12],
+        )
 
-    # Remove internal-only fields, add R2 keys
     result.pop("checkpoint_path", None)
     result.pop("stdout_log", None)
     result["status"] = status
diff --git a/shared/artifacts.py b/shared/artifacts.py
index 6f8261b..29c8b93 100644
--- a/shared/artifacts.py
+++ b/shared/artifacts.py
@@ -1,10 +1,16 @@
 """Training artifact storage — upload, download, verify.
 
-Handles the R2 storage protocol for training artifacts:
-  round_{round_id}/miner_{hotkey}/checkpoint.safetensors
-  round_{round_id}/miner_{hotkey}/architecture.py
-  round_{round_id}/miner_{hotkey}/training_meta.json
-  round_{round_id}/miner_{hotkey}/stdout.log
+Handles the R2/Hippius storage protocol for training artifacts:
+  round_{round_id}/submission_{submission_id}/checkpoint.safetensors
+  round_{round_id}/submission_{submission_id}/architecture.py
+  round_{round_id}/submission_{submission_id}/training_meta.json
+  round_{round_id}/submission_{submission_id}/stdout.log
+
+`submission_id` is an opaque per-round per-job ID minted by the dispatching
+validator. It hides the miner's hotkey from the trainer-host (which is itself
+a miner's pod under cross-eval). The validator publishes the
+``submission_id → miner_hotkey`` reveal map after Phase C closes — see
+``database/server.py:/round_submissions/reveal``.
 
 Hash verification chain: training_meta.json contains sha256 hashes
 binding all artifact files together.
@@ -16,6 +22,7 @@
 import json
 import logging
 import os
+import re as _re
 from dataclasses import dataclass, field
 from typing import Optional, TYPE_CHECKING
 
@@ -25,39 +32,53 @@
 logger = logging.getLogger(__name__)
 
 
-# ── R2 path helpers ──────────────────────────────────────────────
+# ── Path helpers ─────────────────────────────────────────────────
+
+_SAFE_ID = _re.compile(r"^[A-Za-z0-9_-]+$")
 
-import re as _re
 
-_SAFE_HOTKEY = _re.compile(r"^[A-Za-z0-9_-]+$")
+def _validate_submission_id(submission_id: str) -> str:
+    """Validate submission_id contains no path-traversal characters."""
+    if not submission_id or not _SAFE_ID.match(submission_id):
+        raise ValueError(f"Invalid submission_id (unsafe characters): {submission_id!r}")
+    return submission_id
 
 
 def _validate_hotkey(miner_hotkey: str) -> str:
     """Validate hotkey contains no path-traversal characters."""
-    if not _SAFE_HOTKEY.match(miner_hotkey):
+    if not _SAFE_ID.match(miner_hotkey):
         raise ValueError(f"Invalid miner_hotkey (unsafe characters): {miner_hotkey!r}")
     return miner_hotkey
 
 
-def checkpoint_key(round_id: int, miner_hotkey: str) -> str:
-    _validate_hotkey(miner_hotkey)
-    return f"round_{round_id}/miner_{miner_hotkey}/checkpoint.safetensors"
+def checkpoint_key(round_id: int, submission_id: str) -> str:
+    _validate_submission_id(submission_id)
+    return f"round_{round_id}/submission_{submission_id}/checkpoint.safetensors"
 
 
-def architecture_key(round_id: int, miner_hotkey: str) -> str:
-    return f"round_{round_id}/miner_{miner_hotkey}/architecture.py"
+def architecture_key(round_id: int, submission_id: str) -> str:
+    _validate_submission_id(submission_id)
+    return f"round_{round_id}/submission_{submission_id}/architecture.py"
 
 
-def meta_key(round_id: int, miner_hotkey: str) -> str:
-    return f"round_{round_id}/miner_{miner_hotkey}/training_meta.json"
+def meta_key(round_id: int, submission_id: str) -> str:
+    _validate_submission_id(submission_id)
+    return f"round_{round_id}/submission_{submission_id}/training_meta.json"
 
 
-def stdout_key(round_id: int, miner_hotkey: str) -> str:
-    return f"round_{round_id}/miner_{miner_hotkey}/stdout.log"
+def stdout_key(round_id: int, submission_id: str) -> str:
+    _validate_submission_id(submission_id)
+    return f"round_{round_id}/submission_{submission_id}/stdout.log"
 
 
 def scratchpad_key(miner_hotkey: str) -> str:
-    """R2 key for a miner's persistent scratchpad archive."""
+    """R2 key for a miner's persistent scratchpad archive.
+
+    Scratchpads stay keyed by hotkey — they're miner-private state the
+    miner reads/writes themselves through their own creds; no cross-miner
+    anonymity concern applies.
+    """
+    _validate_hotkey(miner_hotkey)
     return f"scratchpad/{miner_hotkey}/state.tar.gz"
 
 
@@ -83,9 +104,14 @@ def generate_scratchpad_urls(
 
 @dataclass
 class TrainingMeta:
-    """Metadata written alongside checkpoint after training."""
+    """Metadata written alongside checkpoint after training.
+
+    The trainer fills ``submission_id`` (from the dispatch payload). The
+    validator only resolves ``submission_id → miner_hotkey`` via its own
+    in-memory dispatch state and the post-Phase-C reveal map.
+    """
     round_id: int = 0
-    miner_hotkey: str = ""
+    submission_id: str = ""
     status: str = ""                    # success | failed | timeout | build_failed | size_violation
     error: str = ""
     flops_equivalent_size: int = 0
@@ -118,7 +144,7 @@ def to_json(self) -> str:
     def to_dict(self) -> dict:
         return {
             "round_id": self.round_id,
-            "miner_hotkey": self.miner_hotkey,
+            "submission_id": self.submission_id,
             "status": self.status,
             "error": self.error,
             "flops_equivalent_size": self.flops_equivalent_size,
@@ -170,25 +196,21 @@ def sha256_text(text: str) -> str:
 def upload_training_artifacts(
     r2: "R2AuditLog",
     round_id: int,
-    miner_hotkey: str,
+    submission_id: str,
     checkpoint_path: str,
     architecture_code: str,
     stdout_log: str,
     meta: TrainingMeta,
 ) -> bool:
-    """Upload all training artifacts to R2 with hash verification chain.
-
-    Computes sha256 of each artifact and stores in training_meta.json.
-    """
-    # Compute hashes
+    """Upload all training artifacts with hash verification chain."""
     meta.checkpoint_sha256 = sha256_file(checkpoint_path)
     meta.architecture_sha256 = sha256_text(architecture_code)
     meta.stdout_sha256 = sha256_text(stdout_log)
 
-    ck = checkpoint_key(round_id, miner_hotkey)
-    ak = architecture_key(round_id, miner_hotkey)
-    mk = meta_key(round_id, miner_hotkey)
-    sk = stdout_key(round_id, miner_hotkey)
+    ck = checkpoint_key(round_id, submission_id)
+    ak = architecture_key(round_id, submission_id)
+    mk = meta_key(round_id, submission_id)
+    sk = stdout_key(round_id, submission_id)
 
     ok = True
     ok = r2.upload_file_from_disk(checkpoint_path, ck) and ok
@@ -197,9 +219,9 @@ def upload_training_artifacts(
     ok = r2.upload_json(mk, meta.to_dict()) and ok
 
     if ok:
-        logger.info("Uploaded artifacts for round %d miner %s", round_id, miner_hotkey)
+        logger.info("Uploaded artifacts for round %d submission %s", round_id, submission_id[:12])
     else:
-        logger.error("Some artifacts failed to upload for round %d miner %s", round_id, miner_hotkey)
+        logger.error("Some artifacts failed to upload for round %d submission %s", round_id, submission_id[:12])
 
     return ok
 
@@ -207,21 +229,20 @@ def upload_training_artifacts(
 def generate_upload_urls(
     r2: "R2AuditLog",
     round_id: int,
-    miner_hotkey: str,
+    submission_id: str,
     ttl: int = 5400,
 ) -> dict[str, str]:
     """Generate pre-signed PUT URLs for all training artifacts.
 
     TTL defaults to 5400s (~90 min). URLs are path-locked to the specific
-    round/miner key.
-
-    Returns dict mapping artifact name to presigned URL.
+    round/submission key, so the trainer-host can only write into its
+    assigned opaque submission slot.
     """
     keys = {
-        "checkpoint": checkpoint_key(round_id, miner_hotkey),
-        "architecture": architecture_key(round_id, miner_hotkey),
-        "meta": meta_key(round_id, miner_hotkey),
-        "stdout": stdout_key(round_id, miner_hotkey),
+        "checkpoint": checkpoint_key(round_id, submission_id),
+        "architecture": architecture_key(round_id, submission_id),
+        "meta": meta_key(round_id, submission_id),
+        "stdout": stdout_key(round_id, submission_id),
     }
     urls = {}
     for name, key in keys.items():
@@ -236,25 +257,21 @@ def generate_upload_urls(
 def verify_uploaded_artifacts(
     r2: "R2AuditLog",
     round_id: int,
-    miner_hotkey: str,
+    submission_id: str,
 ) -> tuple[bool, str]:
-    """Verify that expected artifacts exist in R2 after training upload.
+    """Verify expected artifacts exist after training upload.
 
-    Checks that checkpoint and meta files exist at the expected keys.
-    This catches cases where a presigned URL was used to write to an
-    unexpected path (which S3 presigned URLs prevent, but defense-in-depth).
-
-    Returns (ok, error_message).
+    Checks that checkpoint and meta files exist at the expected keys, and
+    that the meta self-reports the same round_id and submission_id.
     """
-    ck = checkpoint_key(round_id, miner_hotkey)
-    mk = meta_key(round_id, miner_hotkey)
+    ck = checkpoint_key(round_id, submission_id)
+    mk = meta_key(round_id, submission_id)
 
     if not r2.key_exists(mk):
         return False, f"training_meta.json missing at {mk}"
     if not r2.key_exists(ck):
         return False, f"checkpoint missing at {ck}"
 
-    # Verify meta contains correct round_id and miner_hotkey
     meta_dict = r2.download_json(mk)
     if not meta_dict:
         return False, "Failed to download training_meta.json for verification"
@@ -264,10 +281,10 @@ def verify_uploaded_artifacts(
             f"Meta round_id mismatch: expected {round_id}, "
             f"got {meta_dict.get('round_id')}"
         )
-    if meta_dict.get("miner_hotkey") != miner_hotkey:
+    if meta_dict.get("submission_id") != submission_id:
         return False, (
-            f"Meta miner_hotkey mismatch: expected {miner_hotkey}, "
-            f"got {meta_dict.get('miner_hotkey')}"
+            f"Meta submission_id mismatch: expected {submission_id}, "
+            f"got {meta_dict.get('submission_id')}"
         )
 
     return True, ""
@@ -286,14 +303,12 @@ def upload_training_artifacts_presigned(
     """
     import httpx
 
-    # Compute hashes before upload
     meta.checkpoint_sha256 = sha256_file(checkpoint_path)
     meta.architecture_sha256 = sha256_text(architecture_code)
     meta.stdout_sha256 = sha256_text(stdout_log)
 
     ok = True
 
-    # Upload checkpoint (binary file)
     if "checkpoint" in presigned_urls:
         try:
             with open(checkpoint_path, "rb") as f:
@@ -306,7 +321,7 @@ def upload_training_artifacts_presigned(
         logger.error("No presigned URL for checkpoint")
         ok = False
 
-    # Upload architecture (text) — retry once on failure since this is critical for eval
+    # Architecture: retry once — eval depends on it.
     if "architecture" in presigned_urls:
         arch_bytes = architecture_code.encode()
         arch_uploaded = False
@@ -332,7 +347,6 @@ def upload_training_artifacts_presigned(
         logger.error("No presigned URL for architecture — upload_urls keys: %s", list(presigned_urls.keys()))
         ok = False
 
-    # Upload stdout (text)
     if "stdout" in presigned_urls:
         try:
             resp = httpx.put(presigned_urls["stdout"], content=stdout_log.encode(), timeout=30)
@@ -341,7 +355,6 @@ def upload_training_artifacts_presigned(
             logger.error("Presigned upload failed for stdout: %s", e)
             ok = False
 
-    # Upload meta (JSON) — must be last since it contains hashes
     if "meta" in presigned_urls:
         try:
             body = json.dumps(meta.to_dict(), indent=2).encode()
@@ -352,13 +365,13 @@ def upload_training_artifacts_presigned(
             ok = False
 
     if ok:
-        logger.info("Uploaded artifacts via presigned URLs for miner %s", meta.miner_hotkey)
+        logger.info("Uploaded artifacts via presigned URLs for submission %s", meta.submission_id[:12])
     return ok
 
 
 @dataclass
 class DownloadedArtifacts:
-    """All artifacts downloaded from R2 for a single miner's training run."""
+    """All artifacts downloaded for a single miner's training run."""
     meta: TrainingMeta
     checkpoint_path: str = ""    # local path where checkpoint was saved
     architecture_code: str = ""
@@ -370,57 +383,52 @@ class DownloadedArtifacts:
 def download_training_artifacts(
     r2: "R2AuditLog",
     round_id: int,
-    miner_hotkey: str,
+    submission_id: str,
     download_dir: str,
 ) -> Optional[DownloadedArtifacts]:
-    """Download all training artifacts from R2 and verify hashes.
+    """Download all training artifacts for a submission and verify hashes.
 
     Returns None if meta can't be downloaded.
     Sets verified=False with verification_error if hash mismatch.
     """
-    hk_short = miner_hotkey[:16]
-    logger.info("Downloading artifacts for miner %s... round %d", hk_short, round_id)
+    sid_short = submission_id[:12]
+    logger.info("Downloading artifacts for submission %s round %d", sid_short, round_id)
 
-    # Download meta
-    mk = meta_key(round_id, miner_hotkey)
+    mk = meta_key(round_id, submission_id)
     meta_dict = r2.download_json(mk)
     if not meta_dict:
-        logger.warning("No training_meta.json found for miner %s... round %d", hk_short, round_id)
+        logger.warning("No training_meta.json found for submission %s round %d", sid_short, round_id)
         return None
 
     meta = TrainingMeta.from_dict(meta_dict)
     logger.info(
-        "Training meta for %s...: status=%s steps=%d time=%.1fs params=%.2fM",
-        hk_short, meta.status, meta.num_steps,
+        "Training meta for %s: status=%s steps=%d time=%.1fs params=%.2fM",
+        sid_short, meta.status, meta.num_steps,
         meta.training_time_seconds, meta.num_params_M,
     )
 
-    # Download checkpoint
-    ck = checkpoint_key(round_id, miner_hotkey)
+    ck = checkpoint_key(round_id, submission_id)
     os.makedirs(download_dir, exist_ok=True)
-    local_checkpoint = os.path.join(download_dir, f"checkpoint_{miner_hotkey}.safetensors")
+    local_checkpoint = os.path.join(download_dir, f"checkpoint_{submission_id}.safetensors")
     if not r2.download_file_to_disk(ck, local_checkpoint):
-        logger.warning("Failed to download checkpoint for miner %s... round %d", hk_short, round_id)
+        logger.warning("Failed to download checkpoint for submission %s round %d", sid_short, round_id)
         return DownloadedArtifacts(
             meta=meta, verification_error="Failed to download checkpoint",
         )
 
     ckpt_size_mb = os.path.getsize(local_checkpoint) / (1024 * 1024)
-    logger.info("Checkpoint downloaded: %s... (%.2f MB)", hk_short, ckpt_size_mb)
+    logger.info("Checkpoint downloaded: %s (%.2f MB)", sid_short, ckpt_size_mb)
 
-    # Download architecture
-    ak = architecture_key(round_id, miner_hotkey)
+    ak = architecture_key(round_id, submission_id)
     architecture_code = r2.download_text(ak) or ""
 
-    # Download stdout
-    sk = stdout_key(round_id, miner_hotkey)
+    sk = stdout_key(round_id, submission_id)
     stdout_log = r2.download_text(sk) or ""
     logger.info(
-        "Artifacts downloaded for %s...: checkpoint=%.2fMB arch=%d bytes stdout=%d bytes",
-        hk_short, ckpt_size_mb, len(architecture_code), len(stdout_log),
+        "Artifacts downloaded for %s: checkpoint=%.2fMB arch=%d bytes stdout=%d bytes",
+        sid_short, ckpt_size_mb, len(architecture_code), len(stdout_log),
     )
 
-    # Verify hashes
     result = DownloadedArtifacts(
         meta=meta,
         checkpoint_path=local_checkpoint,
@@ -437,19 +445,19 @@ def download_training_artifacts(
     if meta.architecture_sha256 and architecture_code:
         actual = sha256_text(architecture_code)
         if actual != meta.architecture_sha256:
-            errors.append(f"architecture hash mismatch")
+            errors.append("architecture hash mismatch")
 
     if meta.stdout_sha256 and stdout_log:
         actual = sha256_text(stdout_log)
         if actual != meta.stdout_sha256:
-            errors.append(f"stdout hash mismatch")
+            errors.append("stdout hash mismatch")
 
     if errors:
         result.verification_error = "; ".join(errors)
-        logger.warning("Hash verification failed for %s...: %s", hk_short, result.verification_error)
+        logger.warning("Hash verification failed for %s: %s", sid_short, result.verification_error)
     else:
         result.verified = True
-        logger.info("All hashes verified for miner %s...", hk_short)
+        logger.info("All hashes verified for submission %s", sid_short)
 
     return result
 
@@ -458,16 +466,16 @@ def list_round_artifacts(
     r2: "R2AuditLog",
     round_id: int,
 ) -> list[str]:
-    """List all miner hotkeys that have training_meta.json for a round."""
+    """List all submission_ids that have training_meta.json for a round."""
     prefix = f"round_{round_id}/"
     keys = r2.list_keys(prefix)
 
-    hotkeys = set()
+    sids: set[str] = set()
     for key in keys:
-        # round_{id}/miner_{hotkey}/training_meta.json
+        # round_{id}/submission_{sid}/training_meta.json
         if key.endswith("/training_meta.json"):
             parts = key.split("/")
-            if len(parts) >= 3 and parts[1].startswith("miner_"):
-                hotkeys.add(parts[1][len("miner_"):])
+            if len(parts) >= 3 and parts[1].startswith("submission_"):
+                sids.add(parts[1][len("submission_"):])
 
-    return sorted(hotkeys)
+    return sorted(sids)
diff --git a/shared/db_client.py b/shared/db_client.py
index 1c50277..dd6f4c3 100644
--- a/shared/db_client.py
+++ b/shared/db_client.py
@@ -215,6 +215,21 @@ async def submit_training_meta(
         })
         return result is not None
 
+    async def submit_submission_reveal(
+        self, round_id: int, entries: list[dict],
+    ) -> bool:
+        """POST the per-round submission_id -> miner_hotkey reveal map.
+
+        Called after Phase C closes. Each entry is a dict with keys
+        ``submission_id``, ``miner_hotkey``, ``miner_uid``. Idempotent on
+        (round_id, submission_id).
+        """
+        result = await self._post("/round_submissions/reveal", {
+            "round_id": int(round_id),
+            "entries": entries,
+        })
+        return result is not None
+
     async def close(self):
         """Close the underlying HTTP client."""
         if self._client and not self._client.is_closed:
diff --git a/shared/pg_schema.py b/shared/pg_schema.py
index 32fa1f9..24bd33d 100644
--- a/shared/pg_schema.py
+++ b/shared/pg_schema.py
@@ -208,6 +208,25 @@
 );
 CREATE INDEX IF NOT EXISTS idx_tm_round ON training_metas(round_id);
 CREATE INDEX IF NOT EXISTS idx_tm_hotkey ON training_metas(hotkey);
+
+-- Post-Phase-C reveal of the opaque submission_id -> miner_hotkey map.
+-- During Phase B the trainer-host only sees submission_id; once Phase C
+-- closes, validators publish this mapping so:
+--   1. miners can verify the submission they queued was the one trained
+--   2. the public dashboard can resolve historical bucket paths
+--      (``round_{id}/submission_{sid}/...``) back to miner identities
+--      for loss-curve / log views.
+CREATE TABLE IF NOT EXISTS round_submissions (
+    round_id BIGINT NOT NULL,
+    submission_id TEXT NOT NULL,
+    miner_hotkey TEXT NOT NULL,
+    miner_uid INTEGER NOT NULL DEFAULT -1,
+    created_at DOUBLE PRECISION NOT NULL DEFAULT extract(epoch from now()),
+    PRIMARY KEY (round_id, submission_id)
+);
+CREATE INDEX IF NOT EXISTS idx_rs_round_hotkey
+    ON round_submissions(round_id, miner_hotkey);
+CREATE INDEX IF NOT EXISTS idx_rs_hotkey ON round_submissions(miner_hotkey);
 """
 
 ACCESS_LOG_SCHEMA = """
diff --git a/tests/test_artifacts.py b/tests/test_artifacts.py
index 704553b..b3a46b1 100644
--- a/tests/test_artifacts.py
+++ b/tests/test_artifacts.py
@@ -19,15 +19,25 @@
 
 
 def test_path_helpers():
-    assert checkpoint_key(42, "5Foo") == "round_42/miner_5Foo/checkpoint.safetensors"
-    assert architecture_key(42, "5Foo") == "round_42/miner_5Foo/architecture.py"
-    assert meta_key(42, "5Foo") == "round_42/miner_5Foo/training_meta.json"
-    assert stdout_key(42, "5Foo") == "round_42/miner_5Foo/stdout.log"
+    """Bucket paths use the opaque submission_id, not miner_hotkey."""
+    sid = "abc123"
+    assert checkpoint_key(42, sid) == "round_42/submission_abc123/checkpoint.safetensors"
+    assert architecture_key(42, sid) == "round_42/submission_abc123/architecture.py"
+    assert meta_key(42, sid) == "round_42/submission_abc123/training_meta.json"
+    assert stdout_key(42, sid) == "round_42/submission_abc123/stdout.log"
+
+
+def test_path_helpers_reject_path_traversal():
+    """Submission IDs with unsafe characters must be rejected."""
+    import pytest
+    for bad in ("../oops", "a/b", "abc def", ""):
+        with pytest.raises(ValueError):
+            checkpoint_key(1, bad)
 
 
 def test_training_meta_roundtrip():
     meta = TrainingMeta(
-        round_id=1, miner_hotkey="5Bar", status="success",
+        round_id=1, submission_id="sid_bar", status="success",
         flops_equivalent_size=500_000, training_time_seconds=120.5,
         num_steps=1000, num_params_M=2.5,
         peak_vram_mb=1024.0,
@@ -40,7 +50,7 @@ def test_training_meta_roundtrip():
     text = meta.to_json()
     restored = TrainingMeta.from_json(text)
     assert restored.round_id == 1
-    assert restored.miner_hotkey == "5Bar"
+    assert restored.submission_id == "sid_bar"
     assert restored.train_loss_history == [{"step": 10, "loss": 1.0}, {"step": 20, "loss": 0.8}]
     assert restored.val_loss_history == [{"step": 10, "loss": 0.9}]
     assert restored.best_val_loss == 0.9
@@ -51,7 +61,7 @@ def test_training_meta_roundtrip():
 def test_training_meta_old_format_without_new_fields():
     """Old metas (pre-loss-tracking) must still parse via from_dict."""
     old = {
-        "round_id": 7, "miner_hotkey": "5Old", "status": "success",
+        "round_id": 7, "submission_id": "sid_old", "status": "success",
         "loss_curve": [1.0, 0.5],  # legacy field — should be silently dropped
         "num_steps": 100,
     }
@@ -85,7 +95,7 @@ def test_training_meta_roundtrip_with_schedule_policy_fields():
         {"step": 100, "flops": 6_000_000, "loss": 1.9},
     ]
     meta = TrainingMeta(
-        round_id=11, miner_hotkey="5Sched", status="success",
+        round_id=11, submission_id="sid_sched", status="success",
         train_loss_history=train_hist,
         val_loss_history=val_hist,
         best_val_loss=1.2, best_val_step=200,
@@ -97,7 +107,6 @@ def test_training_meta_roundtrip_with_schedule_policy_fields():
         reference_eval_loss_history=ref_hist,
     )
 
-    # to_dict / from_dict round-trip
     restored = TrainingMeta.from_dict(meta.to_dict())
     assert restored.val_cadence_unit == "flops"
     assert restored.val_base == 1e15
@@ -108,7 +117,6 @@ def test_training_meta_roundtrip_with_schedule_policy_fields():
     assert restored.train_loss_history == train_hist
     assert restored.val_loss_history == val_hist
 
-    # to_json / from_json round-trip
     restored2 = TrainingMeta.from_json(meta.to_json())
     assert restored2.val_cadence_unit == "flops"
     assert restored2.val_base == 1e15
@@ -116,7 +124,6 @@ def test_training_meta_roundtrip_with_schedule_policy_fields():
     assert restored2.val_eval_tokens == 4096
     assert restored2.flops_per_step_estimate == 6e8
     assert restored2.reference_eval_loss_history == ref_hist
-    # Loss histories with the new flops key survive JSON serialization intact.
     assert restored2.train_loss_history == train_hist
     assert restored2.val_loss_history == val_hist
 
@@ -149,34 +156,34 @@ def test_sha256_file():
 
 
 def test_upload_and_download(mock_r2):
-    """Full upload → download → verify cycle."""
+    """Full upload → download → verify cycle, keyed by submission_id."""
     with tempfile.NamedTemporaryFile(delete=False, suffix=".safetensors") as f:
         f.write(b"fake checkpoint data")
         ckpt_path = f.name
 
     try:
+        sid = "sid_test"
         meta = TrainingMeta(
-            round_id=1, miner_hotkey="5Test", status="success",
+            round_id=1, submission_id=sid, status="success",
             flops_equivalent_size=200_000, num_steps=500,
         )
 
         ok = upload_training_artifacts(
             r2=mock_r2,
             round_id=1,
-            miner_hotkey="5Test",
+            submission_id=sid,
             checkpoint_path=ckpt_path,
             architecture_code="def build_model(): pass",
             stdout_log="step 1 | loss: 0.5\nstep 2 | loss: 0.4\n",
             meta=meta,
         )
         assert ok
-        assert meta.checkpoint_sha256  # hashes populated
+        assert meta.checkpoint_sha256
         assert meta.architecture_sha256
         assert meta.stdout_sha256
 
-        # Download
         with tempfile.TemporaryDirectory() as dl_dir:
-            result = download_training_artifacts(mock_r2, 1, "5Test", dl_dir)
+            result = download_training_artifacts(mock_r2, 1, sid, dl_dir)
             assert result is not None
             assert result.verified
             assert result.verification_error == ""
@@ -188,7 +195,7 @@ def test_upload_and_download(mock_r2):
 
 
 def test_download_missing_meta(mock_r2):
-    result = download_training_artifacts(mock_r2, 99, "nonexistent", "/tmp")
+    result = download_training_artifacts(mock_r2, 99, "sid_nonexistent", "/tmp")
     assert result is None
 
 
@@ -199,18 +206,18 @@ def test_download_hash_mismatch(mock_r2):
         ckpt_path = f.name
 
     try:
-        meta = TrainingMeta(round_id=2, miner_hotkey="5Tamper", status="success")
+        sid = "sid_tamper"
+        meta = TrainingMeta(round_id=2, submission_id=sid, status="success")
         upload_training_artifacts(
-            mock_r2, 2, "5Tamper", ckpt_path,
+            mock_r2, 2, sid, ckpt_path,
             "code", "stdout", meta,
         )
 
-        # Tamper with checkpoint in R2
-        ck = checkpoint_key(2, "5Tamper")
+        ck = checkpoint_key(2, sid)
         mock_r2._s3.put_object(Bucket=mock_r2.bucket, Key=ck, Body=b"tampered data")
 
         with tempfile.TemporaryDirectory() as dl_dir:
-            result = download_training_artifacts(mock_r2, 2, "5Tamper", dl_dir)
+            result = download_training_artifacts(mock_r2, 2, sid, dl_dir)
             assert result is not None
             assert not result.verified
             assert "checkpoint hash mismatch" in result.verification_error
@@ -219,25 +226,24 @@ def test_download_hash_mismatch(mock_r2):
 
 
 def test_list_round_artifacts(mock_r2):
-    # Upload artifacts for two miners
+    """list_round_artifacts returns the submission_ids present for a round."""
     with tempfile.NamedTemporaryFile(delete=False, suffix=".safetensors") as f:
         f.write(b"data")
         ckpt_path = f.name
 
     try:
-        for hotkey in ["5Alice", "5Bob"]:
-            meta = TrainingMeta(round_id=10, miner_hotkey=hotkey, status="success")
+        for sid in ["sid_alice", "sid_bob"]:
+            meta = TrainingMeta(round_id=10, submission_id=sid, status="success")
             upload_training_artifacts(
-                mock_r2, 10, hotkey, ckpt_path,
+                mock_r2, 10, sid, ckpt_path,
                 "code", "stdout", meta,
             )
 
-        hotkeys = list_round_artifacts(mock_r2, 10)
-        assert len(hotkeys) == 2
-        assert "5Alice" in hotkeys
-        assert "5Bob" in hotkeys
+        sids = list_round_artifacts(mock_r2, 10)
+        assert len(sids) == 2
+        assert "sid_alice" in sids
+        assert "sid_bob" in sids
 
-        # Different round should be empty
         assert list_round_artifacts(mock_r2, 99) == []
     finally:
         os.unlink(ckpt_path)
@@ -257,12 +263,13 @@ def test_verify_uploaded_artifacts_success(mock_r2):
         ckpt_path = f.name
 
     try:
-        meta = TrainingMeta(round_id=5, miner_hotkey="5Valid", status="success")
+        sid = "sid_valid"
+        meta = TrainingMeta(round_id=5, submission_id=sid, status="success")
         upload_training_artifacts(
-            mock_r2, 5, "5Valid", ckpt_path, "code", "stdout", meta,
+            mock_r2, 5, sid, ckpt_path, "code", "stdout", meta,
         )
 
-        ok, err = verify_uploaded_artifacts(mock_r2, 5, "5Valid")
+        ok, err = verify_uploaded_artifacts(mock_r2, 5, sid)
         assert ok
         assert err == ""
     finally:
@@ -271,17 +278,18 @@ def test_verify_uploaded_artifacts_success(mock_r2):
 
 def test_verify_uploaded_artifacts_missing_meta(mock_r2):
     """Missing meta file fails verification."""
-    ok, err = verify_uploaded_artifacts(mock_r2, 99, "5Missing")
+    ok, err = verify_uploaded_artifacts(mock_r2, 99, "sid_missing")
     assert not ok
     assert "training_meta.json missing" in err
 
 
 def test_verify_uploaded_artifacts_missing_checkpoint(mock_r2):
     """Meta exists but checkpoint missing fails verification."""
-    meta = TrainingMeta(round_id=7, miner_hotkey="5NoCkpt", status="success")
-    mock_r2.upload_json(meta_key(7, "5NoCkpt"), meta.to_dict())
+    sid = "sid_no_ckpt"
+    meta = TrainingMeta(round_id=7, submission_id=sid, status="success")
+    mock_r2.upload_json(meta_key(7, sid), meta.to_dict())
 
-    ok, err = verify_uploaded_artifacts(mock_r2, 7, "5NoCkpt")
+    ok, err = verify_uploaded_artifacts(mock_r2, 7, sid)
     assert not ok
     assert "checkpoint missing" in err
 
@@ -293,36 +301,35 @@ def test_verify_uploaded_artifacts_round_id_mismatch(mock_r2):
         ckpt_path = f.name
 
     try:
-        # Upload with round_id=3 but meta claims round_id=999
-        meta = TrainingMeta(round_id=999, miner_hotkey="5Wrong", status="success")
+        sid = "sid_wrong"
+        meta = TrainingMeta(round_id=999, submission_id=sid, status="success")
         upload_training_artifacts(
-            mock_r2, 3, "5Wrong", ckpt_path, "code", "stdout", meta,
+            mock_r2, 3, sid, ckpt_path, "code", "stdout", meta,
         )
-        # Meta was uploaded with round_id=999 inside the JSON
 
-        ok, err = verify_uploaded_artifacts(mock_r2, 3, "5Wrong")
+        ok, err = verify_uploaded_artifacts(mock_r2, 3, sid)
         assert not ok
         assert "round_id mismatch" in err
     finally:
         os.unlink(ckpt_path)
 
 
-def test_verify_uploaded_artifacts_hotkey_mismatch(mock_r2):
-    """Meta with wrong miner_hotkey fails verification."""
+def test_verify_uploaded_artifacts_submission_id_mismatch(mock_r2):
+    """Meta with wrong submission_id fails verification."""
     with tempfile.NamedTemporaryFile(delete=False, suffix=".safetensors") as f:
         f.write(b"data")
         ckpt_path = f.name
 
     try:
-        meta = TrainingMeta(round_id=4, miner_hotkey="5Impostor", status="success")
-        # Upload to 5Real's path but meta says 5Impostor
-        ck = checkpoint_key(4, "5Real")
-        mk = meta_key(4, "5Real")
+        # Meta claims sid_impostor, but we upload to sid_real's path
+        meta = TrainingMeta(round_id=4, submission_id="sid_impostor", status="success")
+        ck = checkpoint_key(4, "sid_real")
+        mk = meta_key(4, "sid_real")
         mock_r2.upload_file_from_disk(ckpt_path, ck)
         mock_r2.upload_json(mk, meta.to_dict())
 
-        ok, err = verify_uploaded_artifacts(mock_r2, 4, "5Real")
+        ok, err = verify_uploaded_artifacts(mock_r2, 4, "sid_real")
         assert not ok
-        assert "miner_hotkey mismatch" in err
+        assert "submission_id mismatch" in err
     finally:
         os.unlink(ckpt_path)
diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py
index 33339bf..9c8d73d 100644
--- a/tests/test_coordinator.py
+++ b/tests/test_coordinator.py
@@ -231,3 +231,138 @@ async def test_dispatch_retries_on_403():
     assert results[0].status == "accepted"
     # POST called twice: first attempt (403) + retry (202)
     assert mock_post.call_count == 2
+
+
+# ── Submission-ID anonymity tests ─────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_dispatch_payload_omits_miner_hotkey_and_carries_submission_id():
+    """Trainer-host must never see the architecture owner's hotkey on the wire.
+
+    The dispatch payload must carry an opaque ``submission_id`` instead.
+    The coordinator mints a fresh sid per job at dispatch time.
+    """
+    import json as _json
+
+    coordinator = TrainingCoordinator(
+        wallet=MagicMock(), metagraph=MagicMock(hotkeys=["hk0", "hk1"]),
+        r2=MagicMock(), my_uid=10,
+    )
+    jobs = [Job(arch_owner=0, trainer_uid=1, dispatcher=10, round_id=1)]
+    challenge = MagicMock(
+        seed=42, round_id=1, min_flops_equivalent=0,
+        max_flops_equivalent=1000000, task={"time_budget": 300},
+    )
+    submissions = {0: Proposal(code="code_a")}
+    endpoints = {1: "http://trainer:8080"}
+
+    captured_payload: dict = {}
+
+    async def fake_post(url, content=None, headers=None):
+        captured_payload.update(_json.loads(content))
+        resp = MagicMock()
+        resp.status_code = 202
+        resp.json.return_value = {"status": "accepted"}
+        return resp
+
+    with patch("httpx.AsyncClient") as mock_client, \
+         patch("shared.artifacts.generate_upload_urls", return_value={
+             "checkpoint": "http://fake/ckpt",
+             "architecture": "http://fake/arch",
+             "meta": "http://fake/meta",
+             "stdout": "http://fake/std",
+         }), \
+         patch("validator.coordinator.sign_request", return_value={"X-Epistula-Signed-By": "hk0"}):
+        mock_client.return_value.__aenter__ = AsyncMock(return_value=MagicMock(
+            post=AsyncMock(side_effect=fake_post),
+        ))
+        mock_client.return_value.__aexit__ = AsyncMock(return_value=False)
+
+        results = await coordinator.dispatch_jobs(
+            jobs, challenge, submissions, endpoints,
+        )
+
+    # Payload privacy invariants
+    assert "miner_hotkey" not in captured_payload
+    assert captured_payload.get("submission_id"), "submission_id must be set"
+    sid = captured_payload["submission_id"]
+    assert len(sid) >= 16  # 16 random bytes hex = 32 chars
+    # The job and result both carry the minted sid
+    assert jobs[0].submission_id == sid
+    assert len(results) == 1
+    assert results[0].submission_id == sid
+
+
+@pytest.mark.asyncio
+async def test_dispatch_record_includes_submission_reveal():
+    """write_dispatch_record publishes (submission_id, miner_hotkey) so other
+    validators can resolve checkpoints minted by this dispatcher in Phase C."""
+    from validator.coordinator import TrainingResult
+
+    r2 = MagicMock()
+    uploaded: dict = {}
+
+    def fake_upload_json(key, body):
+        uploaded[key] = body
+        return True
+
+    r2.upload_json = fake_upload_json
+    wallet = MagicMock()
+    wallet.hotkey.ss58_address = "validator_hk"
+    metagraph = MagicMock(hotkeys=["miner_hk_0", "miner_hk_1"])
+
+    coordinator = TrainingCoordinator(
+        wallet=wallet, metagraph=metagraph, r2=r2, my_uid=10,
+    )
+    results = [
+        TrainingResult(
+            round_id=42, arch_owner=0, trainer_uid=1, dispatcher=10,
+            seed=0, status="accepted", submission_id="sid_one",
+        ),
+        TrainingResult(
+            round_id=42, arch_owner=1, trainer_uid=0, dispatcher=10,
+            seed=0, status="accepted", submission_id="sid_two",
+        ),
+    ]
+    await coordinator.write_dispatch_record(42, results)
+
+    assert "round_42/dispatch/vali_validator_hk.json" in uploaded
+    body = uploaded["round_42/dispatch/vali_validator_hk.json"]
+    sids = {j["submission_id"]: j["arch_owner_hotkey"] for j in body["jobs"]}
+    assert sids == {"sid_one": "miner_hk_0", "sid_two": "miner_hk_1"}
+
+
+@pytest.mark.asyncio
+async def test_collect_submission_map_unions_dispatch_records():
+    """Every dispatcher's record contributes its submission_ids; Phase C
+    consumers union the lot to fetch every checkpoint."""
+    r2 = MagicMock()
+    r2.list_keys = MagicMock(return_value=[
+        "round_5/dispatch/vali_a.json",
+        "round_5/dispatch/vali_b.json",
+        "round_5/dispatch/something_else.txt",  # ignored
+    ])
+    r2.download_json = MagicMock(side_effect=[
+        {"jobs": [
+            {"arch_owner": 0, "submission_id": "sid_0_from_a"},
+            {"arch_owner": 1, "submission_id": "sid_1_from_a"},
+        ]},
+        {"jobs": [
+            {"arch_owner": 2, "submission_id": "sid_2_from_b"},
+            # First-write-wins: another validator can't overwrite arch_owner=0
+            {"arch_owner": 0, "submission_id": "sid_0_dup_from_b"},
+        ]},
+    ])
+
+    coordinator = TrainingCoordinator(
+        wallet=MagicMock(), metagraph=MagicMock(hotkeys=[]),
+        r2=r2, my_uid=10,
+    )
+    mapping = await coordinator.collect_submission_map(round_id=5)
+
+    assert mapping == {
+        0: "sid_0_from_a",
+        1: "sid_1_from_a",
+        2: "sid_2_from_b",
+    }
diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py
index 70f8c4a..869a1eb 100644
--- a/tests/test_dashboard.py
+++ b/tests/test_dashboard.py
@@ -105,6 +105,7 @@ def __init__(
         agent_history: Optional[list[dict]] = None,
         agent_bundles: Optional[dict[str, dict]] = None,
         training_metas: Optional[dict[tuple[int, str], dict]] = None,
+        round_submissions: Optional[dict[tuple[int, str], str]] = None,
     ):
         self.elements = elements
         self.access_log = access_log or []
@@ -113,6 +114,8 @@ def __init__(
         self.agent_bundles = agent_bundles or {}
         # (round_id, hotkey) -> meta dict (mirrors training_metas table)
         self.training_metas = training_metas or {}
+        # (round_id, miner_hotkey) -> submission_id (mirrors round_submissions table)
+        self.round_submissions = round_submissions or {}
 
     def _rows(self):
         return [self._row(e) for e in self.elements]
@@ -327,6 +330,12 @@ async def fetchrow(self, sql: str, *params):
             if meta is None:
                 return None
             return {"meta": json.dumps(meta)}
+        if "from round_submissions" in sql_l and "miner_hotkey = $2" in sql_l:
+            key = (int(params[0]), params[1])
+            sid = self.round_submissions.get(key)
+            if sid is None:
+                return None
+            return {"submission_id": sid}
         return None
 
 
@@ -430,7 +439,7 @@ def dashboard_client():
     # Round 7's meta lives in the Postgres cache; round 8's only in R2 — this
     # keeps the R2 fallback path covered while round 7 exercises the cache.
     cached_meta_round7 = {
-        "round_id": 7, "miner_hotkey": "hk_a",
+        "round_id": 7, "submission_id": "sid_round7_hka",
         "flops": 1234, "ok": True,
         "train_loss_history": [
             {"step": 10, "loss": 22.19}, {"step": 20, "loss": 14.71},
@@ -440,7 +449,7 @@ def dashboard_client():
         ],
     }
     r2_meta_round8 = {
-        "round_id": 8, "miner_hotkey": "hk_a",
+        "round_id": 8, "submission_id": "sid_round8_hka",
         "flops": 5678, "ok": True,
     }
     pool = FakePool(
@@ -453,24 +462,39 @@ def dashboard_client():
             # Element 2 (round 8, hk_b) has an empty experiments.loss_curve —
             # this meta exercises the training_metas fallback in loss_curve.json.
             (8, "hk_b"): {
-                "round_id": 8, "miner_hotkey": "hk_b",
+                "round_id": 8, "submission_id": "sid_round8_hkb",
                 "train_loss_history": [
                     {"step": 1, "loss": 3.3}, {"step": 2, "loss": 2.2},
                 ],
             },
         },
+        # Reveal map: dashboard resolves (round_id, miner_hotkey) → submission_id
+        # via this table, then fetches artifacts from
+        # round_{id}/submission_{sid}/... in R2.
+        round_submissions={
+            (7, "hk_a"): "sid_round7_hka",
+            (8, "hk_a"): "sid_round8_hka",
+            (8, "hk_b"): "sid_round8_hkb",
+        },
     )
     r2 = FakeR2()
     r2.upload_json("agents/hk_a/hash_v1.json", bundle_v1)
     r2.upload_json("agents/hk_a/hash_v2.json", bundle_v2)
-    # Seed a training log file so the log route has something to read.
-    r2.upload_text("round_7/miner_hk_a/stdout.log", "epoch 0 loss=2.0\nepoch 1 loss=1.5\n")
-    # Round 7 meta also goes to R2 so existing tests keep working — the cache
-    # is preferred when both exist.
-    r2.upload_json("round_7/miner_hk_a/training_meta.json", cached_meta_round7)
-    r2.upload_json("round_8/miner_hk_a/training_meta.json", r2_meta_round8)
+    # Seed training artifacts at the new submission_id-keyed path.
+    r2.upload_text(
+        "round_7/submission_sid_round7_hka/stdout.log",
+        "epoch 0 loss=2.0\nepoch 1 loss=1.5\n",
+    )
+    r2.upload_json(
+        "round_7/submission_sid_round7_hka/training_meta.json",
+        cached_meta_round7,
+    )
+    r2.upload_json(
+        "round_8/submission_sid_round8_hka/training_meta.json",
+        r2_meta_round8,
+    )
     r2.upload_text(
-        "round_7/miner_hk_a/architecture.py",
+        "round_7/submission_sid_round7_hka/architecture.py",
         "class Model:\n    def __init__(self):\n        self.name = 'unit-test-model'\n",
     )
 
diff --git a/tests/test_database_server.py b/tests/test_database_server.py
index 4553f3d..5e95ba6 100644
--- a/tests/test_database_server.py
+++ b/tests/test_database_server.py
@@ -479,6 +479,8 @@ class _FakePool:
     def __init__(self):
         self.registry: dict[str, dict] = {}
         self.history: list[dict] = []
+        # (round_id, submission_id) -> {miner_hotkey, miner_uid}
+        self.round_submissions: dict[tuple[int, str], dict] = {}
 
     class _Conn:
         def __init__(self, parent):
@@ -487,6 +489,10 @@ def __init__(self, parent):
         async def execute(self, sql, *params):
             self.parent._execute(sql, *params)
 
+        async def executemany(self, sql, args):
+            for row in args:
+                self.parent._execute(sql, *row)
+
         def transaction(self):
             parent = self.parent
 
@@ -529,6 +535,11 @@ def _execute(self, sql: str, *params):
                 "entry_point": ep, "r2_key": r2, "round_submitted": rs,
                 "timestamp": _time.time(),
             })
+        elif "insert into round_submissions" in sql_l:
+            round_id, sid, hk, uid = params
+            self.round_submissions[(int(round_id), sid)] = {
+                "miner_hotkey": hk, "miner_uid": int(uid),
+            }
 
     async def execute(self, sql, *params):
         self._execute(sql, *params)
@@ -671,3 +682,63 @@ def test_agent_code_by_hash_unknown_returns_404():
     finally:
         srv._r2 = None
         srv._pool = None
+
+
+def test_submission_reveal_persists_mapping():
+    """POST /round_submissions/reveal upserts (round_id, submission_id) rows."""
+    from database import server as srv
+    from database.server import (
+        SubmissionRevealRequest, SubmissionRevealEntry,
+        submit_submission_reveal,
+    )
+
+    fake_pool = _FakePool()
+    srv._pool = fake_pool
+    try:
+        req = SubmissionRevealRequest(
+            round_id=42,
+            entries=[
+                SubmissionRevealEntry(
+                    submission_id="sid_one", miner_hotkey="hk_a", miner_uid=3,
+                ),
+                SubmissionRevealEntry(
+                    submission_id="sid_two", miner_hotkey="hk_b", miner_uid=4,
+                ),
+            ],
+        )
+        result = _call_async(submit_submission_reveal(req))
+        assert result["status"] == "ok"
+        assert result["inserted"] == 2
+
+        assert fake_pool.round_submissions[(42, "sid_one")]["miner_hotkey"] == "hk_a"
+        assert fake_pool.round_submissions[(42, "sid_two")]["miner_uid"] == 4
+    finally:
+        srv._pool = None
+
+
+def test_submission_reveal_skips_empty_entries():
+    """Entries with empty submission_id or hotkey are dropped, not stored."""
+    from database import server as srv
+    from database.server import (
+        SubmissionRevealRequest, SubmissionRevealEntry,
+        submit_submission_reveal,
+    )
+
+    fake_pool = _FakePool()
+    srv._pool = fake_pool
+    try:
+        req = SubmissionRevealRequest(
+            round_id=99,
+            entries=[
+                SubmissionRevealEntry(submission_id="", miner_hotkey="hk_a"),
+                SubmissionRevealEntry(submission_id="sid", miner_hotkey=""),
+                SubmissionRevealEntry(submission_id="sid_ok", miner_hotkey="hk_ok", miner_uid=1),
+            ],
+        )
+        result = _call_async(submit_submission_reveal(req))
+        assert result["inserted"] == 1
+        assert (99, "sid_ok") in fake_pool.round_submissions
+        assert (99, "") not in fake_pool.round_submissions
+        assert (99, "sid") not in fake_pool.round_submissions
+    finally:
+        srv._pool = None
diff --git a/tests/test_generalist_server.py b/tests/test_generalist_server.py
index c1627cd..5e2e965 100644
--- a/tests/test_generalist_server.py
+++ b/tests/test_generalist_server.py
@@ -28,7 +28,7 @@ def _make_body(**overrides):
         "architecture": "def build_model(c, p, n, q): pass\ndef build_optimizer(m): pass",
         "seed": 42,
         "round_id": 1,
-        "miner_hotkey": "miner_abc",
+        "submission_id": "miner_abc",
         "time_budget": 10,
         "task_name": "ts_forecasting",
     }
@@ -40,7 +40,7 @@ def _success_result():
     return ({
         "status": "success",
         "round_id": 1,
-        "miner_hotkey": "miner_abc",
+        "submission_id": "miner_abc",
         "flops_equivalent_size": 100000,
         "training_time_seconds": 5.0,
         "num_steps": 10,
@@ -105,7 +105,7 @@ def test_train_and_upload_dispatches_sandbox(self):
         mock_sandbox = AsyncMock(return_value=_success_result())
         mock_upload = MagicMock(return_value={"status": "success"})
         config = {
-            "seed": 42, "round_id": 1, "miner_hotkey": "miner_abc",
+            "seed": 42, "round_id": 1, "submission_id": "miner_abc",
             "min_flops": 0, "max_flops": 0, "time_budget": 5,
             "task_name": "ts_forecasting",
         }
@@ -132,7 +132,7 @@ def test_train_and_upload_failed_sandbox_skips_upload(self):
 
         mock_sandbox = AsyncMock(return_value=({
             "status": "build_failed", "error": "Missing build_model()",
-            "round_id": 1, "miner_hotkey": "miner_abc",
+            "round_id": 1, "submission_id": "miner_abc",
         }, ""))
         mock_upload = MagicMock()
         mock_failure = MagicMock()
@@ -143,7 +143,7 @@ def test_train_and_upload_failed_sandbox_skips_upload(self):
              patch("runner.server.prefetch_gift_eval", AsyncMock(return_value=None)):
             asyncio.run(srv._train_and_upload(
                 "code", {
-                    "seed": 0, "round_id": 1, "miner_hotkey": "miner_abc",
+                    "seed": 0, "round_id": 1, "submission_id": "miner_abc",
                     "min_flops": 0, "max_flops": 0, "time_budget": 5,
                     "task_name": "ts_forecasting",
                 },
diff --git a/validator/coordinator.py b/validator/coordinator.py
index 7c2e818..0fd6e01 100644
--- a/validator/coordinator.py
+++ b/validator/coordinator.py
@@ -11,6 +11,7 @@
 import logging
 import os
 import random
+import secrets
 from dataclasses import dataclass, field
 from typing import Optional, TYPE_CHECKING
 
@@ -28,11 +29,18 @@
 
 @dataclass
 class Job:
-    """A training job assignment."""
+    """A training job assignment.
+
+    ``submission_id`` is an opaque per-job ID minted by the dispatching
+    validator at dispatch time (left empty for jobs we don't dispatch).
+    It hides the architecture owner's hotkey from the trainer-host (which
+    is itself another miner's pod under cross-eval).
+    """
     arch_owner: int      # miner whose architecture is being trained
     trainer_uid: int     # miner whose trainer runs the job (self-training allowed)
     dispatcher: int      # validator UID responsible for dispatching
     round_id: int = 0
+    submission_id: str = ""
 
 
 @dataclass
@@ -49,6 +57,7 @@ class TrainingResult:
     training_time_seconds: float = 0.0
     checkpoint_key: str = ""
     architecture_key: str = ""
+    submission_id: str = ""
 
 
 def compute_assignments(
@@ -209,31 +218,35 @@ async def dispatch_jobs(
                     dispatcher=self.my_uid,
                     status="failed",
                     error="No trainer endpoint",
+                    submission_id=job.submission_id,
                 ))
                 continue
 
-            miner_hotkey = (
-                self.metagraph.hotkeys[job.arch_owner]
-                if job.arch_owner < len(self.metagraph.hotkeys)
-                else f"uid_{job.arch_owner}"
-            )
+            # Mint an opaque submission_id per job at dispatch time.
+            # The trainer-host receives only this ID — the miner's hotkey
+            # is never on the wire to the trainer or in the bucket path.
+            # `secrets.token_hex(16)` → 32-char unguessable hex.
+            if not job.submission_id:
+                job.submission_id = secrets.token_hex(16)
+            submission_id = job.submission_id
 
             # Generate presigned PUT URLs so trainer can upload without R2 creds.
             from shared.artifacts import generate_upload_urls
             presigned_ttl = int(os.getenv("RADAR_PRESIGNED_TTL", "5400"))
             upload_urls = generate_upload_urls(
-                self.r2, challenge.round_id, miner_hotkey, ttl=presigned_ttl,
+                self.r2, challenge.round_id, submission_id, ttl=presigned_ttl,
             )
 
             if not upload_urls.get("architecture"):
                 logger.error(
-                    "Missing architecture presigned URL for arch_owner=%d miner=%s",
-                    job.arch_owner, miner_hotkey[:16],
+                    "Missing architecture presigned URL for arch_owner=%d submission=%s",
+                    job.arch_owner, submission_id[:12],
                 )
             else:
                 logger.info(
-                    "Generated %d presigned URLs for arch_owner=%d: %s",
-                    len(upload_urls), job.arch_owner, list(upload_urls.keys()),
+                    "Generated %d presigned URLs for arch_owner=%d submission=%s: %s",
+                    len(upload_urls), job.arch_owner, submission_id[:12],
+                    list(upload_urls.keys()),
                 )
 
             task_name = challenge.task.get("name", "")
@@ -244,7 +257,7 @@ async def dispatch_jobs(
                 "round_id": challenge.round_id,
                 "min_flops_equivalent": challenge.min_flops_equivalent,
                 "max_flops_equivalent": challenge.max_flops_equivalent,
-                "miner_hotkey": miner_hotkey,
+                "submission_id": submission_id,
                 "time_budget": time_budget,
                 "upload_urls": upload_urls,
                 "task_name": task_name,
@@ -323,6 +336,7 @@ async def _dispatch_one(
                             dispatcher=self.my_uid,
                             seed=challenge.seed,
                             status="already_running",
+                            submission_id=job.submission_id,
                         )
 
                     try:
@@ -340,6 +354,7 @@ async def _dispatch_one(
                             dispatcher=self.my_uid,
                             status="failed",
                             error=f"Non-JSON response (HTTP {resp.status_code}): {body_preview}",
+                            submission_id=job.submission_id,
                         )
 
                     # 202 Accepted = trainer acknowledged, training in background
@@ -356,6 +371,7 @@ async def _dispatch_one(
                             dispatcher=self.my_uid,
                             seed=challenge.seed,
                             status="accepted",
+                            submission_id=job.submission_id,
                         )
 
                     # Log HTTP errors with the response body for diagnostics
@@ -377,6 +393,7 @@ async def _dispatch_one(
                         training_time_seconds=float(data.get("training_time_seconds", 0)),
                         checkpoint_key=data.get("checkpoint_key", ""),
                         architecture_key=data.get("architecture_key", ""),
+                        submission_id=job.submission_id,
                     )
                 except httpx.TimeoutException:
                     logger.error("Dispatch to trainer UID %d timed out", job.trainer_uid)
@@ -387,6 +404,7 @@ async def _dispatch_one(
                         dispatcher=self.my_uid,
                         status="timeout",
                         error="Request timed out",
+                        submission_id=job.submission_id,
                     )
                 except Exception as e:
                     if attempt < max_retries:
@@ -405,12 +423,14 @@ async def _dispatch_one(
                         dispatcher=self.my_uid,
                         status="failed",
                         error=str(e),
+                        submission_id=job.submission_id,
                     )
             # Should not reach here, but handle defensively
             return TrainingResult(
                 round_id=job.round_id, arch_owner=job.arch_owner,
                 trainer_uid=job.trainer_uid, dispatcher=self.my_uid,
                 status="failed", error="Retries exhausted",
+                submission_id=job.submission_id,
             )
 
         async with httpx.AsyncClient(timeout=job_timeout) as client:
@@ -426,34 +446,43 @@ async def wait_for_checkpoints(
         self,
         round_id: int,
         expected_miners: list[int],
+        uid_to_submission_id: dict[int, str],
         timeout: int = 300,
     ) -> dict[int, dict]:
-        """Poll R2 for training_meta.json from each miner.
-
-        Uses direct key lookups (strongly consistent) instead of
-        list_objects (eventually consistent) to avoid missing recently
-        uploaded artifacts.
-
-        Performs post-upload verification: checks that the meta's round_id
-        and miner_hotkey match expectations, rejecting tampered artifacts.
+        """Poll R2/Hippius for training_meta.json by submission_id.
+
+        Artifacts live at ``round_{id}/submission_{sid}/...``. Resolution
+        from miner UID → submission_id comes from the dispatching
+        validator's in-memory job state plus other validators' published
+        dispatch records — assembled by the caller and passed in via
+        ``uid_to_submission_id``.
+
+        UIDs missing from the mapping are skipped (no submission_id known
+        yet — typically jobs dispatched by another validator that hasn't
+        written its dispatch record). Successful collections include the
+        miner's hotkey/uid in the returned meta so the rest of the
+        validator pipeline keeps working unchanged.
         """
         from shared.artifacts import meta_key as mk_fn, checkpoint_key as ck_fn
 
         results: dict[int, dict] = {}
         deadline = asyncio.get_event_loop().time() + timeout
 
-        # Build uid→hotkey mapping
+        # Build uid→hotkey mapping for stamping the result so downstream
+        # scoring/eval still has the resolved hotkey.
         hotkeys = self.metagraph.hotkeys if hasattr(self.metagraph, "hotkeys") else []
-        uid_to_hotkey = {}
-        for uid in expected_miners:
-            hk = hotkeys[uid] if uid < len(hotkeys) else f"uid_{uid}"
-            uid_to_hotkey[uid] = hk
 
-        # Log expected R2 keys for debugging
-        for uid, hk in uid_to_hotkey.items():
+        # Log expected keys for debugging
+        for uid in expected_miners:
+            sid = uid_to_submission_id.get(uid, "")
+            if not sid:
+                logger.info(
+                    "UID %d: no submission_id known yet — skipping until reveal",
+                    uid,
+                )
+                continue
             logger.info(
-                "Expecting R2 meta key for UID %d: %s",
-                uid, mk_fn(round_id, hk),
+                "Expecting meta key for UID %d: %s", uid, mk_fn(round_id, sid),
             )
 
         poll_count = 0
@@ -464,11 +493,14 @@ async def wait_for_checkpoints(
                 "Checkpoint poll #%d: %d/%d collected, %.0fs remaining (round %d)",
                 poll_count, len(results), len(expected_miners), remaining, round_id,
             )
-            # Check each pending miner directly (strongly consistent)
-            for uid, hk in uid_to_hotkey.items():
+            for uid in expected_miners:
                 if uid in results:
                     continue
-                mk = mk_fn(round_id, hk)
+                sid = uid_to_submission_id.get(uid, "")
+                if not sid:
+                    continue
+                hk = hotkeys[uid] if uid < len(hotkeys) else f"uid_{uid}"
+                mk = mk_fn(round_id, sid)
                 try:
                     meta = self.r2.download_json(mk)
                     if not meta:
@@ -484,24 +516,19 @@ async def wait_for_checkpoints(
                         uid, mk, meta.get("status", "?"),
                     )
 
-                    # Verify meta fields match expectations
                     if meta.get("round_id") != round_id:
                         logger.warning(
                             "Meta round_id mismatch for UID %d: expected %d, got %s",
                             uid, round_id, meta.get("round_id"),
                         )
                         continue
-                    if meta.get("miner_hotkey") != hk:
+                    if meta.get("submission_id") != sid:
                         logger.warning(
-                            "Meta miner_hotkey mismatch for UID %d: expected %s, got %s",
-                            uid, hk[:16], str(meta.get("miner_hotkey"))[:16],
+                            "Meta submission_id mismatch for UID %d: expected %s, got %s",
+                            uid, sid[:12], str(meta.get("submission_id"))[:12],
                         )
                         continue
 
-                    # Failure statuses: collect the meta without requiring a checkpoint.
-                    # The trainer only uploads training_meta.json for failures
-                    # (no checkpoint is produced). Downstream scoring handles these
-                    # via penalties / size gate rejection.
                     _FAILURE_STATUSES = ("build_failed", "size_violation", "failed", "timeout")
                     meta_status = meta.get("status", "")
                     if meta_status in _FAILURE_STATUSES:
@@ -514,8 +541,7 @@ async def wait_for_checkpoints(
                         )
                         continue
 
-                    # Verify checkpoint file exists (only for success status)
-                    ck = ck_fn(round_id, hk)
+                    ck = ck_fn(round_id, sid)
                     if not self.r2.key_exists(ck):
                         logger.info(
                             "Meta exists but checkpoint missing for UID %d at %s (round %d)",
@@ -539,34 +565,46 @@ async def wait_for_checkpoints(
                 break
             await asyncio.sleep(30)
         else:
-            # On timeout, check what actually exists in R2 for this round
             round_prefix = f"round_{round_id}/"
             try:
                 existing_keys = self.r2.list_keys(round_prefix)
                 logger.warning(
                     "Checkpoint collection timed out: %d/%d collected after %ds (round %d). "
-                    "R2 keys under %s: %s",
+                    "Keys under %s: %s",
                     len(results), len(expected_miners), timeout, round_id,
                     round_prefix, existing_keys[:20] if existing_keys else "(none)",
                 )
             except Exception:
                 logger.warning(
                     "Checkpoint collection timed out: %d/%d collected after %ds (round %d). "
-                    "R2 list_keys also failed — check R2 connectivity",
+                    "list_keys also failed — check connectivity",
                     len(results), len(expected_miners), timeout, round_id,
                 )
 
         return results
 
     async def write_dispatch_record(self, round_id: int, results: list[TrainingResult]):
-        """Write dispatch record to R2 (and Hippius if dual-write enabled)."""
+        """Write dispatch record (and the submission_id reveal map for our jobs).
+
+        Each record carries (arch_owner, submission_id, miner_hotkey) so any
+        validator can resolve every checkpoint they need for Phase C eval
+        — the dispatching validator is the only entity that knows the
+        opaque submission_id minted at dispatch.
+        """
         hotkey = self.wallet.hotkey.ss58_address
         key = f"round_{round_id}/dispatch/vali_{hotkey}.json"
+        hotkeys = self.metagraph.hotkeys if hasattr(self.metagraph, "hotkeys") else []
         records = []
         for r in results:
+            arch_hk = (
+                hotkeys[r.arch_owner]
+                if 0 <= r.arch_owner < len(hotkeys) else ""
+            )
             records.append({
                 "arch_owner": r.arch_owner,
+                "arch_owner_hotkey": arch_hk,
                 "trainer_uid": r.trainer_uid,
+                "submission_id": r.submission_id,
                 "status": r.status,
                 "flops_equivalent_size": r.flops_equivalent_size,
                 "training_time_seconds": r.training_time_seconds,
@@ -581,6 +619,43 @@ async def write_dispatch_record(self, round_id: int, results: list[TrainingResul
         else:
             self.r2.upload_json(key, body)
 
+    async def collect_submission_map(
+        self, round_id: int,
+    ) -> dict[int, str]:
+        """Read every validator's dispatch record for ``round_id`` and
+        assemble the union ``arch_owner_uid → submission_id`` map.
+
+        Used for Phase C cross-eval: every validator must independently
+        download every checkpoint, so each needs the submission_ids minted
+        by the *other* validators. Best-effort — missing records mean a
+        UID falls out of the map and its checkpoint is skipped this round.
+        """
+        prefix = f"round_{round_id}/dispatch/"
+        try:
+            keys = self.r2.list_keys(prefix)
+        except Exception as e:
+            logger.warning("Failed to list dispatch records for round %d: %s", round_id, e)
+            return {}
+        mapping: dict[int, str] = {}
+        for key in keys:
+            if not key.endswith(".json"):
+                continue
+            try:
+                body = self.r2.download_json(key)
+            except Exception as e:
+                logger.warning("Failed to read dispatch record %s: %s", key, e)
+                continue
+            if not isinstance(body, dict):
+                continue
+            for job in body.get("jobs", []) or []:
+                if not isinstance(job, dict):
+                    continue
+                sid = job.get("submission_id", "")
+                uid = job.get("arch_owner")
+                if isinstance(uid, int) and sid and uid not in mapping:
+                    mapping[uid] = sid
+        return mapping
+
     async def write_frontier(
         self, frontier_data: list[dict], task_name: str = "",
     ):
diff --git a/validator/evaluator.py b/validator/evaluator.py
index af432b3..beb7d23 100644
--- a/validator/evaluator.py
+++ b/validator/evaluator.py
@@ -225,11 +225,18 @@ async def evaluate_all_checkpoints(
         if meta.get("status") != "success":
             continue
 
+        submission_id = meta.get("submission_id", "")
+        if not submission_id:
+            logger.warning("UID %d: meta missing submission_id — skipping eval", uid)
+            continue
         miner_hotkey = meta.get("miner_hotkey", f"uid_{uid}")
-        logger.info("Evaluating UID %d (miner %s...) round %d", uid, miner_hotkey[:16], round_id)
+        logger.info(
+            "Evaluating UID %d (miner %s submission %s) round %d",
+            uid, miner_hotkey[:16], submission_id[:12], round_id,
+        )
 
-        # Download and verify all artifacts
-        artifacts = download_training_artifacts(r2, round_id, miner_hotkey, tmp_dir)
+        # Download and verify all artifacts (keyed by submission_id, not hotkey)
+        artifacts = download_training_artifacts(r2, round_id, submission_id, tmp_dir)
         if not artifacts:
             logger.warning("UID %d: failed to download artifacts", uid)
             continue
diff --git a/validator/neuron.py b/validator/neuron.py
index 8d8d0e2..44ef8d5 100644
--- a/validator/neuron.py
+++ b/validator/neuron.py
@@ -440,6 +440,55 @@ async def _cache_training_metas(
                 "Cached %d training_meta blobs for round %d", cached, round_id,
             )
 
+    async def _publish_submission_reveal(
+        self,
+        round_id: int,
+        uid_to_sid: dict[int, str],
+        hotkeys: list[str],
+    ) -> None:
+        """Push the per-round submission_id → (miner_hotkey, miner_uid) map
+        to the centralised DB after Phase C closes.
+
+        Best-effort — if the cache write fails the dashboard simply falls
+        back to the historic ``miner_{hotkey}/`` paths it knows for older
+        rounds. Scoring never depends on this call. Only the dispatching
+        validator owns the truth for its submission_ids; idempotent
+        upsert in the DB merges across validators safely.
+        """
+        if not uid_to_sid:
+            return
+        entries = []
+        for uid, sid in uid_to_sid.items():
+            if not sid:
+                continue
+            hk = hotkeys[uid] if 0 <= uid < len(hotkeys) else ""
+            if not hk:
+                continue
+            entries.append({
+                "submission_id": sid,
+                "miner_hotkey": hk,
+                "miner_uid": int(uid),
+            })
+        if not entries:
+            return
+        try:
+            ok = await self.db_client.submit_submission_reveal(
+                round_id=round_id, entries=entries,
+            )
+            if ok:
+                logger.info(
+                    "Published submission reveal: round=%d entries=%d",
+                    round_id, len(entries),
+                )
+            else:
+                logger.warning(
+                    "Submission reveal returned non-OK: round=%d", round_id,
+                )
+        except Exception as e:
+            logger.warning(
+                "Submission reveal failed: round=%d err=%s", round_id, e,
+            )
+
     async def run_round(self):
         """Execute one full 3-phase round."""
         # ── SETUP ──────────────────────────────────────────────
@@ -672,6 +721,14 @@ async def _no_endpoints():
 
         await self.coordinator.write_dispatch_record(challenge.round_id, my_results)
 
+        # Build the uid → submission_id map. Start with my own jobs (in-memory
+        # source of truth for the submission_ids I just minted), then merge in
+        # other validators' submission_ids by reading their dispatch records.
+        uid_to_sid: dict[int, str] = {
+            r.arch_owner: r.submission_id
+            for r in my_results if r.submission_id
+        }
+
         if Config.SKIP_TRAINING_WAIT:
             logger.info("SKIP_TRAINING_WAIT enabled — skipping block wait after dispatch")
         else:
@@ -679,10 +736,21 @@ async def _no_endpoints():
                 round_start + Config.SUBMISSION_WINDOW_BLOCKS + Config.TRAINING_WINDOW_BLOCKS
             )
 
+        # Pull other validators' dispatch records so we can poll for their
+        # checkpoints too (Phase C requires every validator to evaluate every
+        # checkpoint, so we need the union of submission_ids).
+        try:
+            other_sids = await self.coordinator.collect_submission_map(challenge.round_id)
+            for uid, sid in other_sids.items():
+                uid_to_sid.setdefault(uid, sid)
+        except Exception as e:
+            logger.warning("Failed to collect cross-validator submission map: %s", e)
+
         # ── Checkpoint collection ─────────────────────────────
         training_metas = await self.coordinator.wait_for_checkpoints(
             challenge.round_id,
             expected_miners=list(filtered.keys()),
+            uid_to_submission_id=uid_to_sid,
             timeout=Config.EVAL_WINDOW_BLOCKS * 12,
         )
 
@@ -744,6 +812,9 @@ async def _no_endpoints():
                     await self.coordinator.write_dispatch_record(
                         challenge.round_id, fb_results,
                     )
+                    for r in fb_results:
+                        if r.submission_id:
+                            uid_to_sid[r.arch_owner] = r.submission_id
 
                 await self._wait_until_block(
                     round_start + Config.SUBMISSION_WINDOW_BLOCKS
@@ -751,9 +822,22 @@ async def _no_endpoints():
                     + Config.FALLBACK_WINDOW_BLOCKS
                 )
 
+                # Re-pull other validators' fallback dispatch records too.
+                try:
+                    other_sids = await self.coordinator.collect_submission_map(
+                        challenge.round_id,
+                    )
+                    for uid, sid in other_sids.items():
+                        uid_to_sid.setdefault(uid, sid)
+                except Exception as e:
+                    logger.warning(
+                        "Failed to collect fallback submission map: %s", e,
+                    )
+
                 fb_metas = await self.coordinator.wait_for_checkpoints(
                     challenge.round_id,
                     expected_miners=missing_trainers,
+                    uid_to_submission_id=uid_to_sid,
                     timeout=Config.FALLBACK_WINDOW_BLOCKS * 12,
                 )
                 training_metas.update(fb_metas)
@@ -794,6 +878,17 @@ async def _no_endpoints():
         )
 
         logger.info("Phase C: evaluated %d checkpoints", len(eval_results))
+
+        # ── REVEAL: publish submission_id → miner_hotkey mapping ──
+        # Phase B is over and Phase C has read every checkpoint, so
+        # revealing the mapping no longer lets a trainer-host selectively
+        # sandbag a rival. The dashboard uses this map to render
+        # per-miner training history for past rounds.
+        await self._publish_submission_reveal(
+            round_id=challenge.round_id,
+            uid_to_sid=uid_to_sid,
+            hotkeys=self.metagraph.hotkeys or [],
+        )
         primary_obj = round_task.primary_objective
         primary_name = primary_obj.name if primary_obj else "metric"
         primary_default = primary_obj.default if primary_obj else float("inf")

From 72fb7204c18eb46ccf9bd6df56342e3eb9e6cc4d Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 5 May 2026 00:37:18 +0000
Subject: [PATCH 2/6] Defer dispatch-record write to end of Phase B (Hippius
 anonymity)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Hippius is decentralized object storage — anything written there is
world-readable for the lifetime of the substrate chain. The dispatch
record carries the submission_id ↔ arch_owner mapping that lets other
validators resolve checkpoints in Phase C, but writing it right after
dispatch let a trainer-host fetch the record from Hippius mid-training
and de-anonymise the architecture they were running.

Move both the primary and fallback write_dispatch_record calls to
after the corresponding training-window block wait. By the time the
record becomes readable, training is already complete and selective
sandbagging is no longer possible.
---
 validator/neuron.py | 23 +++++++++++++++++------
 1 file changed, 17 insertions(+), 6 deletions(-)

diff --git a/validator/neuron.py b/validator/neuron.py
index 44ef8d5..7fe48ac 100644
--- a/validator/neuron.py
+++ b/validator/neuron.py
@@ -719,11 +719,9 @@ async def _no_endpoints():
                     r.arch_owner, r.trainer_uid, r.status, r.error,
                 )
 
-        await self.coordinator.write_dispatch_record(challenge.round_id, my_results)
-
         # Build the uid → submission_id map. Start with my own jobs (in-memory
         # source of truth for the submission_ids I just minted), then merge in
-        # other validators' submission_ids by reading their dispatch records.
+        # other validators' submission_ids after Phase B closes.
         uid_to_sid: dict[int, str] = {
             r.arch_owner: r.submission_id
             for r in my_results if r.submission_id
@@ -736,6 +734,14 @@ async def _no_endpoints():
                 round_start + Config.SUBMISSION_WINDOW_BLOCKS + Config.TRAINING_WINDOW_BLOCKS
             )
 
+        # Publish the dispatch record AFTER the Phase B training window closes.
+        # The record carries the submission_id ↔ arch_owner mapping that lets
+        # other validators resolve checkpoints in Phase C. Hippius is public
+        # by design (decentralized object store), so writing the record any
+        # earlier would let a trainer-host fetch it mid-training and learn
+        # whose architecture they're running — defeating the anonymisation.
+        await self.coordinator.write_dispatch_record(challenge.round_id, my_results)
+
         # Pull other validators' dispatch records so we can poll for their
         # checkpoints too (Phase C requires every validator to evaluate every
         # checkpoint, so we need the union of submission_ids).
@@ -802,6 +808,7 @@ async def _no_endpoints():
                     block_hash, missing_dispatchers, all_jobs, remaining_valis,
                 )
                 my_fallback = [j for j in fallback_jobs if j.dispatcher == my_uid]
+                fb_results: list = []
                 if my_fallback:
                     fb_extras = self._build_dispatch_extras(challenge, round_task)
                     fb_results = await self.coordinator.dispatch_jobs(
@@ -809,9 +816,6 @@ async def _no_endpoints():
                         commitments=commitments,
                         extras=fb_extras,
                     )
-                    await self.coordinator.write_dispatch_record(
-                        challenge.round_id, fb_results,
-                    )
                     for r in fb_results:
                         if r.submission_id:
                             uid_to_sid[r.arch_owner] = r.submission_id
@@ -822,6 +826,13 @@ async def _no_endpoints():
                     + Config.FALLBACK_WINDOW_BLOCKS
                 )
 
+                # Publish fallback dispatch record AFTER the fallback training
+                # window closes — same anonymity reason as the primary record.
+                if fb_results:
+                    await self.coordinator.write_dispatch_record(
+                        challenge.round_id, fb_results,
+                    )
+
                 # Re-pull other validators' fallback dispatch records too.
                 try:
                     other_sids = await self.coordinator.collect_submission_map(

From b0e539686ca39620d8877ba7104247781cb503dd Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 5 May 2026 09:25:45 +0000
Subject: [PATCH 3/6] Stop leaking Basilica agent pods on HTTP errors and
 crashes

Each Basilica retry inside run_agent_on_pod spawns a fresh deployment
(timestamp-suffixed name), but env.cleanup() only knows about the most
recent one. With AGENT_POD_RETRIES=3 and 8-way agent concurrency, a
round of 502s could orphan up to 32 GPU pods that bill until TTL.
There was also no validator-side reaper, so a process crash mid-round
left every active pod stranded.

* Track every deployment name affinetes assigns across retries on the
  env, then force-delete each one via BasilicaClient.delete_deployment
  in cleanup_agent_env (called from _run_single_agent's finally block).
* Move launch_agent_pod inside the try/finally so the cleanup path
  also runs when launch itself raises.
* Add reap_orphan_agent_pods, a Basilica orphan sweeper that mirrors
  the miner's _reap_stale_deployments. Run at validator startup and
  after every round.
* Lower AGENT_POD_RETRIES default 3 -> 1 (each retry costs real GPU
  time and rarely succeeds when Basilica is degraded).
* Add 13 unit tests covering name tracking, force-delete on cleanup,
  the reaper's prefix/age filtering, and SDK-missing fallbacks.

https://claude.ai/code/session_01Xi8m7KpYdPwiTJUuTo3Srb
---
 config.py                 |  13 ++-
 tests/test_pod_manager.py | 160 ++++++++++++++++++++++++++++++-
 validator/collection.py   |  28 +++---
 validator/neuron.py       |  20 +++-
 validator/pod_manager.py  | 194 ++++++++++++++++++++++++++++++++++++++
 5 files changed, 400 insertions(+), 15 deletions(-)

diff --git a/config.py b/config.py
index e2aebc5..8172dc3 100644
--- a/config.py
+++ b/config.py
@@ -171,7 +171,18 @@ class Config:
     # `agent_seconds` in the task YAML; this is the fallback when a task
     # doesn't set one. Must fit within SUBMISSION_WINDOW_BLOCKS × 12s.
     AGENT_TIMEOUT: int = int(os.getenv("RADAR_AGENT_TIMEOUT", "600"))
-    AGENT_POD_RETRIES: int = int(os.getenv("RADAR_AGENT_POD_RETRIES", "3"))
+    # Each retry on Basilica spawns a brand-new deployment (timestamp
+    # suffix in the name), so a high retry count multiplies pod cost
+    # under transient HTTP errors. Default to 1 retry — anything more
+    # rarely succeeds and reliably leaks pods until the orphan reaper
+    # cleans them.
+    AGENT_POD_RETRIES: int = int(os.getenv("RADAR_AGENT_POD_RETRIES", "1"))
+    # Max age (seconds) for orphan agent pods picked up by the reaper.
+    # A round's full agent budget is AGENT_TIMEOUT (default 600s); 1800s
+    # is 3x that, so anything older was definitely abandoned.
+    AGENT_POD_REAP_MAX_AGE_S: int = int(
+        os.getenv("RADAR_AGENT_POD_REAP_MAX_AGE_S", "1800"),
+    )
 
     # Max number of agent pods this validator runs concurrently in Phase A.
     # Each pod is on a separate Basilica node so there's no local resource
diff --git a/tests/test_pod_manager.py b/tests/test_pod_manager.py
index e0805cf..c40bcac 100644
--- a/tests/test_pod_manager.py
+++ b/tests/test_pod_manager.py
@@ -7,7 +7,8 @@
 from validator.pod_manager import (
     get_mode, pre_validate_code, verify_miner_pod, _parse_image_ref,
     _normalise_agent_code, _write_agent_code, launch_agent_pod,
-    run_agent_on_pod,
+    run_agent_on_pod, cleanup_agent_env, reap_orphan_agent_pods,
+    _record_deployment_name,
 )
 
 
@@ -369,3 +370,160 @@ async def test_task_id_omitted_when_not_provided(self):
         await run_agent_on_pod(mock_env, '{}', timeout=60)
         call_kwargs = mock_env.process_challenge.call_args.kwargs
         assert "task_id" not in call_kwargs
+
+    @pytest.mark.asyncio
+    async def test_records_deployment_names_across_retries(self):
+        """Each retry's deployment name is captured for cleanup."""
+        env = MagicMock(spec=["_agent_code", "_deployment_name", "process_challenge"])
+        env._agent_code = None
+        # Simulate Basilica re-naming the deployment on every retry
+        names = ["radar-agent-pc-1001", "radar-agent-pc-1002", "radar-agent-pc-1003"]
+
+        async def _proc(**kw):
+            env._deployment_name = names.pop(0)
+            if names:
+                raise RuntimeError("502 Bad Gateway")
+            return {"code": "x"}
+
+        env.process_challenge = _proc
+        with patch("config.Config.AGENT_POD_RETRIES", 2), \
+             patch("asyncio.sleep", new_callable=AsyncMock):
+            result = await run_agent_on_pod(env, '{}', timeout=60)
+        assert result == {"code": "x"}
+        assert env._radar_deployment_names == [
+            "radar-agent-pc-1001",
+            "radar-agent-pc-1002",
+            "radar-agent-pc-1003",
+        ]
+
+
+class TestRecordDeploymentName:
+    def test_picks_up_string_attr(self):
+        env = MagicMock(spec=["_deployment_name"])
+        env._deployment_name = "radar-agent-1"
+        _record_deployment_name(env)
+        assert env._radar_deployment_names == ["radar-agent-1"]
+
+    def test_dedupes_repeated_names(self):
+        env = MagicMock(spec=["deployment_name"])
+        env.deployment_name = "radar-agent-1"
+        _record_deployment_name(env)
+        _record_deployment_name(env)
+        assert env._radar_deployment_names == ["radar-agent-1"]
+
+    def test_falls_back_to_deployment_object(self):
+        dep = MagicMock()
+        dep.name = "radar-agent-deep"
+        env = MagicMock(spec=["_deployment"])
+        env._deployment = dep
+        _record_deployment_name(env)
+        assert env._radar_deployment_names == ["radar-agent-deep"]
+
+    def test_silent_when_no_name_available(self):
+        env = MagicMock(spec=[])
+        _record_deployment_name(env)
+        assert not hasattr(env, "_radar_deployment_names")
+
+
+class TestCleanupAgentEnv:
+    @pytest.mark.asyncio
+    async def test_none_env_is_noop(self):
+        await cleanup_agent_env(None)  # must not raise
+
+    @pytest.mark.asyncio
+    async def test_calls_env_cleanup(self):
+        env = MagicMock()
+        env.cleanup = AsyncMock()
+        env._radar_deployment_names = []
+        await cleanup_agent_env(env)
+        env.cleanup.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_swallows_cleanup_exception(self):
+        env = MagicMock()
+        env.cleanup = AsyncMock(side_effect=RuntimeError("boom"))
+        env._radar_deployment_names = []
+        await cleanup_agent_env(env)  # must not raise
+
+    @pytest.mark.asyncio
+    async def test_force_deletes_recorded_names(self):
+        """Every recorded deployment is force-deleted via BasilicaClient."""
+        env = MagicMock()
+        env.cleanup = AsyncMock()
+        env._radar_deployment_names = ["a", "b", "c"]
+
+        fake_client = MagicMock()
+        fake_client.delete_deployment = MagicMock()
+        fake_module = ModuleType("basilica")
+        fake_module.BasilicaClient = MagicMock(return_value=fake_client)
+        with patch.dict(sys.modules, {"basilica": fake_module}):
+            await cleanup_agent_env(env)
+
+        deleted = [c.args[0] for c in fake_client.delete_deployment.call_args_list]
+        assert deleted == ["a", "b", "c"]
+
+    @pytest.mark.asyncio
+    async def test_force_delete_failures_swallowed(self):
+        env = MagicMock()
+        env.cleanup = AsyncMock()
+        env._radar_deployment_names = ["a"]
+
+        fake_client = MagicMock()
+        fake_client.delete_deployment = MagicMock(side_effect=RuntimeError("404"))
+        fake_module = ModuleType("basilica")
+        fake_module.BasilicaClient = MagicMock(return_value=fake_client)
+        with patch.dict(sys.modules, {"basilica": fake_module}):
+            await cleanup_agent_env(env)  # must not raise
+
+
+class TestReapOrphanAgentPods:
+    @pytest.mark.asyncio
+    async def test_no_op_when_not_basilica_mode(self):
+        with patch("validator.pod_manager.get_mode", return_value="docker"):
+            n = await reap_orphan_agent_pods()
+        assert n == 0
+
+    @pytest.mark.asyncio
+    async def test_deletes_old_matching_deployments(self):
+        old = MagicMock()
+        old.name = "radar-agent-pc-old"
+        old.uptime_seconds = 4000  # > 1800
+
+        young = MagicMock()
+        young.name = "radar-agent-pc-young"
+        young.uptime_seconds = 60  # fresh
+
+        other = MagicMock()
+        other.name = "radar-trainer-x"
+        other.uptime_seconds = 9999  # wrong prefix
+
+        no_age = MagicMock()
+        no_age.name = "radar-agent-pc-noage"
+        no_age.uptime_seconds = None
+        no_age.created_at = None  # unparseable → skip
+
+        fake_client = MagicMock()
+        fake_client.list_deployments = MagicMock(
+            return_value=[old, young, other, no_age],
+        )
+        fake_client.delete_deployment = MagicMock()
+        fake_module = ModuleType("basilica")
+        fake_module.BasilicaClient = MagicMock(return_value=fake_client)
+
+        with patch("validator.pod_manager.get_mode", return_value="basilica"), \
+             patch.dict(sys.modules, {"basilica": fake_module}):
+            n = await reap_orphan_agent_pods(
+                prefix="radar-agent", max_age_seconds=1800,
+            )
+
+        assert n == 1
+        deleted = [c.args[0] for c in fake_client.delete_deployment.call_args_list]
+        assert deleted == ["radar-agent-pc-old"]
+
+    @pytest.mark.asyncio
+    async def test_basilica_sdk_missing_returns_zero(self):
+        with patch("validator.pod_manager.get_mode", return_value="basilica"), \
+             patch.dict(sys.modules, {"basilica": None}):
+            # Importing a module set to None raises ImportError
+            n = await reap_orphan_agent_pods()
+        assert n == 0
diff --git a/validator/collection.py b/validator/collection.py
index 535bb53..2e82f10 100644
--- a/validator/collection.py
+++ b/validator/collection.py
@@ -34,7 +34,8 @@
 from shared.protocol import Proposal
 from shared.url_gate import parse_allowed_urls
 from validator.pod_manager import (
-    launch_agent_pod, pre_validate_agent_code, run_agent_on_pod,
+    cleanup_agent_env, launch_agent_pod, pre_validate_agent_code,
+    run_agent_on_pod,
 )
 
 logger = logging.getLogger(__name__)
@@ -234,15 +235,21 @@ async def _run_single_agent(
     except (json.JSONDecodeError, TypeError, ValueError):
         agent_timeout = Config.AGENT_TIMEOUT
 
-    agent_env = await launch_agent_pod(
-        image_url=Config.OFFICIAL_AGENT_IMAGE,
-        mem_limit="8192Mi",
-        agent_code=bundle,
-        allowed_urls=allowed_urls,
-    )
-
+    # Wrap launch + run in a single try/finally so a failure inside
+    # launch_agent_pod (or anywhere before cleanup) still triggers
+    # teardown. Each Basilica retry inside run_agent_on_pod can spawn a
+    # NEW deployment; cleanup_agent_env force-deletes every name we
+    # recorded, not just the latest, which prevents orphan accumulation
+    # on HTTP-error paths.
+    agent_env = None
     started_at = time.monotonic()
     try:
+        agent_env = await launch_agent_pod(
+            image_url=Config.OFFICIAL_AGENT_IMAGE,
+            mem_limit="8192Mi",
+            agent_code=bundle,
+            allowed_urls=allowed_urls,
+        )
         result = await run_agent_on_pod(
             agent_env, miner_challenge_json,
             timeout=agent_timeout,
@@ -298,10 +305,7 @@ async def _run_single_agent(
                 type(result).__name__, result_keys,
             )
     finally:
-        try:
-            await agent_env.cleanup()
-        except Exception:
-            pass
+        await cleanup_agent_env(agent_env)
     return None, _empty_meta()
 
 
diff --git a/validator/neuron.py b/validator/neuron.py
index 7fe48ac..2948d36 100644
--- a/validator/neuron.py
+++ b/validator/neuron.py
@@ -44,7 +44,7 @@
     set_hotkey_map, rotate_agent_token,
 )
 from validator.evaluator import evaluate_all_checkpoints
-from validator.pod_manager import pre_validate_code
+from validator.pod_manager import pre_validate_code, reap_orphan_agent_pods
 from validator.substrate_publisher import run_substrate_publish_step
 
 logger = logging.getLogger(__name__)
@@ -1132,6 +1132,15 @@ def _set_weights(self):
     async def run(self):
         """Main validator loop."""
         self.start_proxy_server()
+        # Reap any agent pods left over from a prior crash/redeploy
+        # before we start spawning new ones. Pure safety-net — never
+        # raises, never blocks startup on Basilica latency.
+        try:
+            await reap_orphan_agent_pods(
+                max_age_seconds=Config.AGENT_POD_REAP_MAX_AGE_S,
+            )
+        except Exception as e:
+            logger.warning("Startup orphan reap failed: %s", e)
         while True:
             try:
                 current_block = self.subtensor.block
@@ -1168,6 +1177,15 @@ async def run(self):
                 self._last_completed_round = round_start
             except Exception:
                 logger.error("Round error:\n%s", traceback.format_exc())
+            # Sweep any agent pods that the round left behind — covers
+            # mid-round exceptions, retry-spawned pods cleanup missed,
+            # and load-shed scenarios.
+            try:
+                await reap_orphan_agent_pods(
+                    max_age_seconds=Config.AGENT_POD_REAP_MAX_AGE_S,
+                )
+            except Exception as e:
+                logger.debug("Post-round orphan reap failed: %s", e)
             await asyncio.sleep(60)
 
 
diff --git a/validator/pod_manager.py b/validator/pod_manager.py
index ab3fcc7..e8b3f8e 100644
--- a/validator/pod_manager.py
+++ b/validator/pod_manager.py
@@ -16,14 +16,22 @@
 
 import asyncio
 import ast
+import functools
 import json
 import logging
 import os
 import tempfile
+import time
 from typing import Optional
 
 logger = logging.getLogger(__name__)
 
+# Default prefix used to identify validator-owned agent deployments on
+# Basilica. Anything matching this prefix that's older than the reap age
+# is assumed to be an orphan (the validator that created it crashed,
+# was redeployed, or hit an HTTP error path that skipped cleanup).
+_DEFAULT_AGENT_PREFIX = "radar-agent"
+
 
 def _af():
     """Lazy import of affinetes to avoid hard dependency at import time."""
@@ -267,6 +275,7 @@ async def run_agent_on_pod(
     for attempt in range(1 + max_retries):
         try:
             result = await env.process_challenge(**call_kwargs)
+            _record_deployment_name(env)
             if attempt > 0:
                 logger.info(
                     "Agent pod succeeded on attempt %d/%d",
@@ -274,6 +283,11 @@ async def run_agent_on_pod(
                 )
             return result
         except Exception as e:
+            # Each Basilica retry spawns a NEW deployment (timestamp suffix
+            # in the name). Record whatever name was assigned this attempt
+            # so finally-block cleanup can delete every pod we created,
+            # not just the last one.
+            _record_deployment_name(env)
             err_str = str(e)
             is_502 = "502" in err_str
             if attempt < max_retries:
@@ -297,6 +311,186 @@ async def run_agent_on_pod(
     return None
 
 
+# ── Cleanup ──────────────────────────────────────────────────────
+
+def _record_deployment_name(env) -> None:
+    """Capture the Basilica deployment name affinetes assigned to ``env``.
+
+    affinetes exposes the active deployment under one of a few attribute
+    names depending on backend version (``_deployment_name``,
+    ``deployment_name``, or a ``_deployment`` object with a ``.name``).
+    We accumulate every distinct name we observe across retries so
+    cleanup can hit them all.
+    """
+    name = None
+    for attr in ("_deployment_name", "deployment_name", "instance_name"):
+        candidate = getattr(env, attr, None)
+        if isinstance(candidate, str) and candidate:
+            name = candidate
+            break
+    if name is None:
+        dep = getattr(env, "_deployment", None) or getattr(env, "deployment", None)
+        candidate = getattr(dep, "name", None)
+        if isinstance(candidate, str) and candidate:
+            name = candidate
+    if not name:
+        return
+    seen = getattr(env, "_radar_deployment_names", None)
+    if seen is None:
+        seen = []
+        try:
+            env._radar_deployment_names = seen
+        except Exception:
+            return
+    if name not in seen:
+        seen.append(name)
+
+
+async def cleanup_agent_env(env) -> None:
+    """Best-effort teardown of every Basilica deployment tied to ``env``.
+
+    Calls ``env.cleanup()`` first (the affinetes path), then force-deletes
+    every deployment name we recorded during retries via the Basilica
+    SDK. Always swallows exceptions — this runs in ``finally`` blocks and
+    must never mask the original error.
+    """
+    if env is None:
+        return
+    cleanup = getattr(env, "cleanup", None)
+    if cleanup is not None:
+        try:
+            res = cleanup()
+            if asyncio.iscoroutine(res):
+                await res
+        except Exception as e:
+            logger.debug("env.cleanup() failed: %s", e)
+
+    names = list(getattr(env, "_radar_deployment_names", []) or [])
+    if not names:
+        return
+    try:
+        from basilica import BasilicaClient
+    except ImportError:
+        return
+    try:
+        client = BasilicaClient()
+    except Exception as e:
+        logger.debug("BasilicaClient init failed during cleanup: %s", e)
+        return
+    loop = asyncio.get_event_loop()
+    for name in names:
+        try:
+            await loop.run_in_executor(
+                None, functools.partial(client.delete_deployment, name),
+            )
+            logger.info("Force-deleted leaked agent deployment: %s", name)
+        except Exception as e:
+            logger.debug(
+                "Force-delete of %s failed (TTL will reap): %s", name, e,
+            )
+
+
+# ── Orphan Reaper ────────────────────────────────────────────────
+
+def _list_basilica_deployments(client):
+    """Call whichever list method the installed basilica SDK exposes.
+
+    Returns an iterable of deployment objects, or [] if no list API is
+    available (older SDKs). Each object is expected to have ``.name`` and
+    one of ``.created_at`` / ``.uptime_seconds`` we can use to age it.
+    """
+    for method in ("list_deployments", "deployments", "list"):
+        fn = getattr(client, method, None)
+        if callable(fn):
+            try:
+                return list(fn())
+            except Exception as e:
+                logger.debug("%s() failed: %s", method, e)
+                return []
+    return []
+
+
+def _deployment_age_seconds(dep) -> Optional[float]:
+    """Return the age of a Basilica deployment in seconds, or None.
+
+    Tries ``uptime_seconds`` first (a single number), then ``created_at``
+    (epoch or ISO 8601). Returns None when neither is parseable so the
+    caller can skip rather than mistakenly delete a fresh pod.
+    """
+    uptime = getattr(dep, "uptime_seconds", None)
+    if isinstance(uptime, (int, float)) and uptime >= 0:
+        return float(uptime)
+    created = getattr(dep, "created_at", None)
+    if isinstance(created, (int, float)):
+        return max(0.0, time.time() - float(created))
+    if isinstance(created, str):
+        try:
+            from datetime import datetime
+            ts = datetime.fromisoformat(created.replace("Z", "+00:00"))
+            return max(0.0, time.time() - ts.timestamp())
+        except Exception:
+            return None
+    return None
+
+
+async def reap_orphan_agent_pods(
+    prefix: str = _DEFAULT_AGENT_PREFIX,
+    max_age_seconds: int = 1800,
+) -> int:
+    """Delete Basilica deployments matching ``prefix`` older than the cap.
+
+    Safety net for pods leaked by validator crashes, retry storms, or
+    cleanup paths that didn't run. Mirrors ``Miner._reap_stale_deployments``
+    on the miner side. Returns the number of deployments deleted.
+
+    Skips deployments whose age can't be determined to avoid deleting
+    fresh pods on SDK versions that don't expose timestamps.
+    """
+    if get_mode() != "basilica":
+        return 0
+    try:
+        from basilica import BasilicaClient
+    except ImportError:
+        return 0
+    try:
+        client = BasilicaClient()
+    except Exception as e:
+        logger.debug("BasilicaClient init failed during reap: %s", e)
+        return 0
+
+    loop = asyncio.get_event_loop()
+    try:
+        deps = await loop.run_in_executor(
+            None, functools.partial(_list_basilica_deployments, client),
+        )
+    except Exception as e:
+        logger.debug("Basilica list failed: %s", e)
+        return 0
+
+    deleted = 0
+    for dep in deps:
+        name = getattr(dep, "name", None) or getattr(dep, "instance_name", None)
+        if not isinstance(name, str) or not name.startswith(prefix):
+            continue
+        age = _deployment_age_seconds(dep)
+        if age is None or age < max_age_seconds:
+            continue
+        try:
+            await loop.run_in_executor(
+                None, functools.partial(client.delete_deployment, name),
+            )
+            deleted += 1
+            logger.info(
+                "Reaped orphan agent deployment %s (%.0f min old)",
+                name, age / 60,
+            )
+        except Exception as e:
+            logger.debug("Reap delete of %s failed: %s", name, e)
+    if deleted:
+        logger.info("Orphan reaper deleted %d agent deployments", deleted)
+    return deleted
+
+
 # ── Pod Verification (Basilica public metadata) ──────────────────
 
 

From 9699b4c11224c8c2c2ec7c0edb7ddf397dd0aac9 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Tue, 5 May 2026 20:10:51 +0000
Subject: [PATCH 4/6] Await provenance write so the heatmap actually fills
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The middleware was scheduling _access_logger.log_access via
asyncio.create_task. Per the asyncio docs, the event loop only keeps
*weak* references to tasks created that way — under load the task can
be garbage-collected before the INSERT runs, and the miner_access_log
table never gets the row that drives the dashboard heatmap.

Every test for this code path had to manually
loop.run_until_complete(asyncio.sleep(0.05)) to give the task a tick,
which is itself proof that the production path was racy.

Await the call inside the middleware. A single Postgres INSERT is
cheap, and we wrap it in try/except so a logger failure can't break
the user-facing response. Drop the sleep workarounds in the existing
tests and add a regression test using a deliberately slow logger:
if anyone reverts to fire-and-forget, the assertion that the write
completed before the response was observed will fail.

https://claude.ai/code/session_017JNvSgYhBbjVmHVWZSVozE
---
 database/server.py            | 21 +++++++++-----
 tests/test_database_server.py | 52 +++++++++++++++++++++--------------
 2 files changed, 45 insertions(+), 28 deletions(-)

diff --git a/database/server.py b/database/server.py
index 30164f8..e24ee46 100644
--- a/database/server.py
+++ b/database/server.py
@@ -380,13 +380,20 @@ async def auth_middleware(request: Request, call_next):
             miner_uid = int(miner_uid_str)
         except ValueError:
             miner_uid = -1
-        asyncio.create_task(_access_logger.log_access(
-            hotkey=miner_hotkey or "validator",
-            miner_uid=miner_uid,
-            endpoint=path,
-            experiment_ids=exp_ids,
-            method=request.method,
-        ))
+        # Await the write: asyncio.create_task gives the event loop only a
+        # weak reference, so the task can be GC'd before the INSERT runs and
+        # the heatmap stays empty. A single Postgres INSERT is cheap; any
+        # failure must not break the user-facing response.
+        try:
+            await _access_logger.log_access(
+                hotkey=miner_hotkey or "validator",
+                miner_uid=miner_uid,
+                endpoint=path,
+                experiment_ids=exp_ids,
+                method=request.method,
+            )
+        except Exception as e:
+            logger.warning("access_log write failed: %s", e)
 
     return response
 
diff --git a/tests/test_database_server.py b/tests/test_database_server.py
index 5e95ba6..2112433 100644
--- a/tests/test_database_server.py
+++ b/tests/test_database_server.py
@@ -324,11 +324,6 @@ def test_access_log_captures_ids_from_experiment_detail():
         assert r.status_code == 200
         # Body should still be delivered intact
         assert r.json()["name"] == "exp_0"
-        # asyncio.create_task was scheduled — give it a tick to run
-        import asyncio as _asyncio
-        loop = _asyncio.new_event_loop()
-        loop.run_until_complete(_asyncio.sleep(0.05))
-        loop.close()
         assert logger.calls, "middleware did not log the request"
         call = logger.calls[-1]
         assert 0 in call["experiment_ids"]
@@ -345,10 +340,6 @@ def test_access_log_captures_ids_from_list_response():
         r = client.get("/experiments/recent?n=2")
         assert r.status_code == 200
         assert len(r.json()) == 2
-        import asyncio as _asyncio
-        loop = _asyncio.new_event_loop()
-        loop.run_until_complete(_asyncio.sleep(0.05))
-        loop.close()
         # Captured IDs match the response payload
         payload_ids = {item["index"] for item in r.json()}
         captured = set(logger.calls[-1]["experiment_ids"])
@@ -369,10 +360,6 @@ def test_access_log_skips_non_experiment_paths():
         client = TestClient(app)
         r = client.get("/provenance/component_stats")
         assert r.status_code == 200
-        import asyncio as _asyncio
-        loop = _asyncio.new_event_loop()
-        loop.run_until_complete(_asyncio.sleep(0.05))
-        loop.close()
         # Still logs the access. The aggregate /provenance/component_stats
         # response carries no per-experiment IDs and the path itself isn't
         # an /provenance/{id} shape, so nothing gets extracted.
@@ -396,10 +383,6 @@ def test_access_log_captures_id_from_provenance_path():
         client = TestClient(app)
         r = client.get("/provenance/0/influences")
         assert r.status_code == 200
-        import asyncio as _asyncio
-        loop = _asyncio.new_event_loop()
-        loop.run_until_complete(_asyncio.sleep(0.05))
-        loop.close()
         assert logger.calls, "middleware did not log /provenance/0/influences"
         assert 0 in logger.calls[-1]["experiment_ids"]
         assert logger.calls[-1]["endpoint"] == "/provenance/0/influences"
@@ -435,10 +418,6 @@ def test_access_log_runs_on_proxy_api_key_path():
             },
         )
         assert r.status_code == 200
-        import asyncio as _asyncio
-        loop = _asyncio.new_event_loop()
-        loop.run_until_complete(_asyncio.sleep(0.05))
-        loop.close()
         assert logger.calls, "proxy path did not reach access logger"
         call = logger.calls[-1]
         assert call["hotkey"] == "miner_hk_abc"
@@ -449,6 +428,37 @@ def test_access_log_runs_on_proxy_api_key_path():
         _uninstall_access_logger()
 
 
+def test_access_log_is_awaited_not_fire_and_forget():
+    """Regression: middleware used to schedule log_access via
+    asyncio.create_task, which the event loop only weakly references — under
+    load the task could be GC'd before the INSERT ran and the heatmap stayed
+    empty. The write must complete before the response returns.
+    """
+    _setup()
+
+    class _SlowLogger:
+        def __init__(self):
+            self.completed = False
+
+        async def log_access(self, **kw):
+            import asyncio as _a
+            await _a.sleep(0.02)
+            self.completed = True
+
+    from database import server as srv
+    slow = _SlowLogger()
+    srv._access_logger = slow
+    try:
+        client = TestClient(app)
+        r = client.get("/experiments/0")
+        assert r.status_code == 200
+        # If the write were fire-and-forget, .completed would still be False
+        # at the moment the response is observed by the client.
+        assert slow.completed, "access log write was not awaited"
+    finally:
+        srv._access_logger = None
+
+
 # ── Agent code submission history ──────────────────────────────
 
 

From 009de29bb78cf79ed26da34002985f002f7a4127 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 6 May 2026 10:08:07 +0000
Subject: [PATCH 5/6] Lower val_base_flops default so the val curve actually
 renders
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The FLOPs-keyed validation cadence introduced in fd6cfc9 set
val_base_flops=1e15 as the default first-trigger threshold. A typical
Phase B run only accumulates ~1e13 cumulative FLOPs (Tiny bucket,
500K params × ~15K steps in the scaled time budget) up to ~1e16
(Large bucket, 50M params × ~10K steps). For every bucket but Large,
that ceiling is never crossed during training — val_loss_history
ends up with at most one entry from the end-of-run flush at the
bottom of _training_loop, which the dashboard chart can't render as
a curve. That's why "validation loss from Phase B training does not
seem to show on most recent experiments on the dashboard": the data
point literally isn't there.

Drop the default to 1e10 and lower the clamp floor to 1e8 to match.
At 1e10 with val_growth=2.0:
  - Tiny    (~6e9 flops/step):   first val ~step 2,  ~10–12 vals total
  - Small   (~6e10 flops/step):  first val ~step 1,  ~12–14 vals total
  - Large   (~6e11 flops/step):  first val on step 1, dense early then
    sparser as next_val_flops doubles past cumulative_flops — still ~15
    log-spaced points across the run

Update the two tests pinning the prior numeric defaults to assert the
new ones, with comments explaining why the threshold moved.

https://claude.ai/code/session_017JNvSgYhBbjVmHVWZSVozE
---
 runner/harness.py         | 10 ++++++++--
 tests/test_harness_val.py | 11 ++++++++---
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git a/runner/harness.py b/runner/harness.py
index b7e6e7e..8b16404 100644
--- a/runner/harness.py
+++ b/runner/harness.py
@@ -253,7 +253,13 @@ def run_training(runner: TaskRunner, architecture_code: str, config: TrainingCon
     "grad_clip": 1.0,
     "log_every_n_steps": 10,
     "val_base_step": 10,
-    "val_base_flops": 1e15,
+    # First val triggers at this many cumulative FLOPs, then doubles each
+    # time (val_growth=2.0). 1e10 gives ~10 log-spaced val points across the
+    # full bucket range (Tiny ≈ 1e13 cumulative → ~10 vals; Large ≈ 1e16 →
+    # ~20 vals). The previous 1e15 default was so high that Phase B runs
+    # never crossed it for any bucket but Large, leaving val_loss_history
+    # with only the end-of-run flush — unrenderable as a curve.
+    "val_base_flops": 1e10,
     "val_growth": 2.0,
 }
 _CLAMPS = {
@@ -262,7 +268,7 @@ def run_training(runner: TaskRunner, architecture_code: str, config: TrainingCon
     "grad_clip": (0.0, 100.0),
     "log_every_n_steps": (1, 1000),
     "val_base_step": (1, 10000),
-    "val_base_flops": (1e12, 1e22),
+    "val_base_flops": (1e8, 1e22),
     "val_growth": (1.1, 10.0),
 }
 _VAL_SCHEDULES = ("logarithmic", "fixed", "none")
diff --git a/tests/test_harness_val.py b/tests/test_harness_val.py
index 0e6ef26..3f68344 100644
--- a/tests/test_harness_val.py
+++ b/tests/test_harness_val.py
@@ -491,7 +491,12 @@ def test_read_config_default_val_cadence_unit_is_flops():
     sub = types.ModuleType("s")
     cfg = _read_config(sub)
     assert cfg["val_cadence_unit"] == "flops"
-    assert cfg["val_base_flops"] == 1e15
+    # 1e10 produces ~10 log-spaced val samples across the full size-bucket
+    # range. The original 1e15 default was so high that Phase B runs in any
+    # bucket but Large never crossed it — val_loss_history stayed empty
+    # except for the end-of-run flush, so the dashboard chart had nothing
+    # to draw.
+    assert cfg["val_base_flops"] == 1e10
     # Step-based defaults are unchanged so mixed-cadence runs stay comparable.
     assert cfg["val_base_step"] == 10
     assert cfg["val_growth"] == 2.0
@@ -517,9 +522,9 @@ def test_read_config_invalid_val_cadence_unit_falls_back():
 
 def test_read_config_clamps_val_base_flops():
     sub = types.ModuleType("s")
-    sub.training_config = lambda: {"val_base_flops": 1.0}  # well below 1e12
+    sub.training_config = lambda: {"val_base_flops": 1.0}  # well below the floor
     cfg = _read_config(sub)
-    assert cfg["val_base_flops"] == 1e12
+    assert cfg["val_base_flops"] == 1e8
 
 
 # ── FLOPs-keyed val cadence ───────────────────────────────────────

From 552ccca0847b393ed92699e14ca7aac5ce069ac9 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 6 May 2026 10:19:29 +0000
Subject: [PATCH 6/6] Anchor first val_flops to val_base_step so cadence stays
 log-spaced
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The previous commit lowered val_base_flops's default to 1e10 to make
sure the threshold actually fires during a Phase B run, but with an
absolute floor a high-FLOPs/step model overshoots the threshold on
its very first batch and then fires val on consecutive steps until
the doubling chain (val_growth=2.0) outpaces cumulative_flops. That
produces a dense early cluster, not a log axis — exactly the
spacing artefact called out in review.

Mirror val_base_step's role in step-cadence: when initialising
next_val_flops on the first batch, take

    max(val_base_flops, flops_per_optim_step * val_base_step)

so the first val always lands ~val_base_step optim steps in, for
every bucket. Subsequent vals double from there, giving a clean
log axis. val_base_flops is reframed as an absolute floor for the
case where flops_per_optim_step is essentially zero.

The flops-cadence growth-ratio test still passes: with batch=2,
seq=4, and a 20-param Linear model, flops_per_optim_step=960, so
the new floor (9600) supersedes the test's 1500 setting and pushes
the first val to step 10. Later vals land at steps 20, 40, 80, 160,
320 — six entries with 2× ratios, well within the test's
[1.5, 3.5) bound.

https://claude.ai/code/session_017JNvSgYhBbjVmHVWZSVozE
---
 runner/harness.py | 26 +++++++++++++++++++-------
 1 file changed, 19 insertions(+), 7 deletions(-)

diff --git a/runner/harness.py b/runner/harness.py
index 8b16404..fd23d7e 100644
--- a/runner/harness.py
+++ b/runner/harness.py
@@ -253,12 +253,12 @@ def run_training(runner: TaskRunner, architecture_code: str, config: TrainingCon
     "grad_clip": 1.0,
     "log_every_n_steps": 10,
     "val_base_step": 10,
-    # First val triggers at this many cumulative FLOPs, then doubles each
-    # time (val_growth=2.0). 1e10 gives ~10 log-spaced val points across the
-    # full bucket range (Tiny ≈ 1e13 cumulative → ~10 vals; Large ≈ 1e16 →
-    # ~20 vals). The previous 1e15 default was so high that Phase B runs
-    # never crossed it for any bucket but Large, leaving val_loss_history
-    # with only the end-of-run flush — unrenderable as a curve.
+    # Absolute floor for the first-val FLOPs threshold. The actual
+    # threshold is max(this, flops_per_optim_step * val_base_step) so the
+    # curve stays log-spaced across model sizes — see the cadence-init
+    # block in _training_loop. The previous 1e15 default was so high that
+    # Phase B runs never crossed it for any bucket but Large, leaving
+    # val_loss_history with only the end-of-run flush.
     "val_base_flops": 1e10,
     "val_growth": 2.0,
 }
@@ -443,7 +443,19 @@ def _training_loop(runner: TaskRunner, sub, model, device: str, time_budget: int
 
             if val_enabled and use_flops_cadence_requested:
                 if flops_per_optim_step > 0:
-                    next_val_flops = int(cfg["val_base_flops"])
+                    # Anchor the first val to roughly val_base_step optim
+                    # steps' worth of FLOPs so the cadence stays log-spaced
+                    # across model sizes. With a fixed absolute threshold,
+                    # high-FLOPs/step models overshoot it on the first batch
+                    # and then fire val on every consecutive step until
+                    # next_val_flops's doublings outpace cumulative_flops —
+                    # producing a dense early cluster instead of a log axis.
+                    # Mirroring val_base_step here keeps the first val at
+                    # ~step val_base_step for every bucket.
+                    next_val_flops = max(
+                        int(cfg["val_base_flops"]),
+                        flops_per_optim_step * cfg["val_base_step"],
+                    )
                 else:
                     next_val = cfg["val_base_step"]