From da206c8c44db6b5c276df67f434a0e7177943167 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 5 May 2026 00:20:04 +0000 Subject: [PATCH 1/2] 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 Date: Tue, 5 May 2026 00:37:18 +0000 Subject: [PATCH 2/2] 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(