Claude/fix provenance x yac6

config.py

@@ -171,7 +171,18 @@ class Config:
     # `agent_seconds` in the task YAML; this is the fallback when a task
     # doesn't set one. Must fit within SUBMISSION_WINDOW_BLOCKS × 12s.
     AGENT_TIMEOUT: int = int(os.getenv("RADAR_AGENT_TIMEOUT", "600"))
-    AGENT_POD_RETRIES: int = int(os.getenv("RADAR_AGENT_POD_RETRIES", "3"))
+    # Each retry on Basilica spawns a brand-new deployment (timestamp
+    # suffix in the name), so a high retry count multiplies pod cost
+    # under transient HTTP errors. Default to 1 retry — anything more
+    # rarely succeeds and reliably leaks pods until the orphan reaper
+    # cleans them.
+    AGENT_POD_RETRIES: int = int(os.getenv("RADAR_AGENT_POD_RETRIES", "1"))
+    # Max age (seconds) for orphan agent pods picked up by the reaper.
+    # A round's full agent budget is AGENT_TIMEOUT (default 600s); 1800s
+    # is 3x that, so anything older was definitely abandoned.
+    AGENT_POD_REAP_MAX_AGE_S: int = int(
+        os.getenv("RADAR_AGENT_POD_REAP_MAX_AGE_S", "1800"),
+    )
 
     # Max number of agent pods this validator runs concurrently in Phase A.
     # Each pod is on a separate Basilica node so there's no local resource

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)

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",

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)

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);