Add miner listener registry for off-chain deploy

CLAUDE.md

@@ -239,6 +239,7 @@ frontier/latest.json                                        # Current Pareto fro
 - [x] DatabaseClient for validator -> DB server communication (`shared/db_client.py`)
 - [x] Validator / dashboard deploy split (`RADAR_NEURON_MODE`)
 - [x] Hippius (Substrate) artifact backend with R2 legacy fallback (`shared/r2_audit.py`)
+- [x] Off-chain miner listener registry (miners self-register via `POST /miners/me/listener`; validators discover via `GET /miners/active`)
 
 ## Artifact Storage (Hippius / R2)
 

database/migrations/006_agent_listener_registry.sql

@@ -0,0 +1,34 @@
+-- Off-chain miner listener registry.
+--
+-- The post-bittensor deploy has no on-chain ImageCommitment to carry a
+-- miner's listener_url. Miners now self-register their URL via
+-- POST /agent_code (carried as a new optional field) and refresh it
+-- via POST /miners/me/listener. Validators discover live miners
+-- through GET /miners/active using the listener_seen_at timestamp.
+--
+-- All DDL is idempotent. The block is wrapped in an
+-- information_schema guard because the CI migration runner applies
+-- this file against an empty schema before init_schema() has
+-- bootstrapped the agent_submissions table — same pattern as
+-- 005_miner_registry_and_prompt_id.sql. On a fresh deploy the base
+-- CREATE TABLE in shared/pg_schema.py already declares the new
+-- columns, so the guarded block is a no-op there.
+
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1
+        FROM information_schema.tables
+        WHERE table_schema = 'CURRENT_SCHEMA'
+          AND table_name = 'agent_submissions'
+    ) THEN
+        ALTER TABLE CURRENT_SCHEMA.agent_submissions
+            ADD COLUMN IF NOT EXISTS listener_url TEXT NOT NULL DEFAULT '';
+        ALTER TABLE CURRENT_SCHEMA.agent_submissions
+            ADD COLUMN IF NOT EXISTS listener_seen_at
+                DOUBLE PRECISION NOT NULL DEFAULT 0.0;
+        CREATE INDEX IF NOT EXISTS idx_agent_listener_seen
+            ON CURRENT_SCHEMA.agent_submissions(listener_seen_at)
+            WHERE listener_url <> '';
+    END IF;
+END $$;

database/server.py

@@ -720,6 +720,10 @@ class SubmitAgentCodeRequest(BaseModel):
     """Miner POSTs their agent code bundle."""
     files: dict[str, str]          # filename -> source code
     entry_point: str = "agent.py"  # which file has design_architecture()
+    # Where the miner's warm-standby listener is reachable. Optional in
+    # the body so existing clients still work; validators that need to
+    # send TrainerRequests look this up via GET /miners/active.
+    listener_url: str = ""
 
 
 @validator_router.post("/agent_code")
@@ -794,25 +798,36 @@ async def submit_agent_code(request: Request, req: SubmitAgentCodeRequest):
     import json as _json
 
     bundle_json = _json.dumps(bundle)
+    listener_url = (req.listener_url or "").strip()
     try:
         async with _pool.acquire() as conn:
             async with conn.transaction():
                 await conn.execute(
                     """
                     INSERT INTO agent_submissions
                         (hotkey, miner_uid, code_hash, entry_point, r2_key,
-                         round_submitted, timestamp)
-                    VALUES ($1, $2, $3, $4, $5, $6, extract(epoch from now()))
+                         round_submitted, timestamp,
+                         listener_url, listener_seen_at)
+                    VALUES ($1, $2, $3, $4, $5, $6,
+                            extract(epoch from now()),
+                            $7,
+                            CASE WHEN $7 = '' THEN 0.0
+                                 ELSE extract(epoch from now()) END)
                     ON CONFLICT (hotkey) DO UPDATE SET
                         miner_uid = $2,
                         code_hash = $3,
                         entry_point = $4,
                         r2_key = $5,
                         round_submitted = $6,
-                        timestamp = extract(epoch from now())
+                        timestamp = extract(epoch from now()),
+                        listener_url = CASE WHEN $7 = ''
+                            THEN agent_submissions.listener_url ELSE $7 END,
+                        listener_seen_at = CASE WHEN $7 = ''
+                            THEN agent_submissions.listener_seen_at
+                            ELSE extract(epoch from now()) END
                     """,
                     hotkey, miner_uid, code_hash, req.entry_point,
-                    immutable_key, round_submitted,
+                    immutable_key, round_submitted, listener_url,
                 )
                 await conn.execute(
                     """
@@ -951,6 +966,119 @@ async def get_agent_code_by_hash(code_hash: str):
     return bundle
 
 
+# ── Miner listener registry ────────────────────────────────────────
+#
+# The post-bittensor deploy has no on-chain ImageCommitment to carry a
+# miner's listener_url. Miners self-register by POSTing it here (with
+# the bearer token they already use for /agent_code), refreshing
+# `listener_seen_at` on each call. Validators then call /miners/active
+# at the start of every round to discover where to send TrainerRequests.
+
+# Active = listener_seen_at within this many seconds. 6h tolerates
+# brief miner restarts without re-registration while still pruning
+# truly-gone miners before the next round.
+_ACTIVE_MINER_WINDOW_SECONDS: float = 6 * 60 * 60
+
+
+class RegisterListenerRequest(BaseModel):
+    """Miner POSTs the URL of its warm-standby listener.
+
+    The miner must have already submitted agent code (so a row in
+    agent_submissions exists for this hotkey); otherwise the endpoint
+    returns 404 and the miner should call submit_agent_code first.
+    """
+    listener_url: str
+
+
+@validator_router.post("/miners/me/listener")
+async def register_listener(request: Request, req: RegisterListenerRequest):
+    if _pool is None:
+        raise HTTPException(status_code=503, detail="DB pool not configured")
+
+    # _bearer_auth_then_call (the /miners/me/* path) sets miner_identity;
+    # the agent_code helper sets caller_hotkey. Accept either so this
+    # endpoint works under both auth wrappers (and is testable on its own).
+    hotkey = getattr(request.state, "caller_hotkey", "") or ""
+    if not hotkey:
+        ident = getattr(request.state, "miner_identity", None)
+        if ident is not None:
+            hotkey = ident.hotkey or ident.miner_id
+    if not hotkey:
+        raise HTTPException(status_code=403, detail="Auth required")
+
+    listener_url = (req.listener_url or "").strip()
+    if not listener_url:
+        raise HTTPException(status_code=400, detail="listener_url required")
+
+    result = await _pool.execute(
+        """
+        UPDATE agent_submissions
+           SET listener_url = $2,
+               listener_seen_at = extract(epoch from now())
+         WHERE hotkey = $1
+        """,
+        hotkey, listener_url,
+    )
+    try:
+        updated = int(result.split()[-1]) > 0
+    except (ValueError, IndexError):
+        updated = False
+    if not updated:
+        raise HTTPException(
+            status_code=404,
+            detail="No agent_submissions row — submit agent code first",
+        )
+    logger.info(
+        "Listener registered: hotkey=%s url=%s",
+        hotkey[:16], listener_url,
+    )
+    return {"status": "ok", "listener_url": listener_url}
+
+
+@validator_router.get("/miners/active")
+async def list_active_miners(
+    window_seconds: float = _ACTIVE_MINER_WINDOW_SECONDS,
+):
+    """Recent miners with live listener registrations.
+
+    Validators call this at the start of every round to build the set
+    of miners eligible to receive TrainerRequests. Rows are filtered
+    by `listener_seen_at` so a stale miner whose listener is no
+    longer reachable drops out automatically after the TTL.
+    """
+    if _pool is None:
+        raise HTTPException(status_code=503, detail="DB pool not configured")
+
+    cutoff = max(60.0, float(window_seconds))
+    rows = await _pool.fetch(
+        """
+        SELECT hotkey, miner_uid, code_hash, entry_point,
+               listener_url, listener_seen_at,
+               round_submitted, timestamp
+          FROM agent_submissions
+         WHERE listener_url <> ''
+           AND listener_seen_at > extract(epoch from now()) - $1
+         ORDER BY listener_seen_at DESC
+        """,
+        cutoff,
+    )
+    return {
+        "miners": [
+            {
+                "hotkey": r["hotkey"],
+                "miner_uid": r["miner_uid"],
+                "code_hash": r["code_hash"],
+                "entry_point": r["entry_point"],
+                "listener_url": r["listener_url"],
+                "listener_seen_at": float(r["listener_seen_at"]),
+                "round_submitted": int(r["round_submitted"]),
+                "timestamp": float(r["timestamp"]),
+            }
+            for r in rows
+        ],
+    }
+
+
 def include_validator_routes(target_app: FastAPI) -> None:
     """Mount the HMAC-authed validator/miner surface onto ``target_app``.
 

miner/neuron.py

@@ -167,16 +167,18 @@ async def submit_agent_code(self):
             return
         from shared.db_client import DatabaseClient
         db = DatabaseClient(db_url=db_url, api_key=self.api_key)
+        listener_url = self._listener_external_url()
         try:
             result = await db.submit_agent_code(
                 files=bundle["files"],
                 entry_point=bundle["entry_point"],
+                listener_url=listener_url,
             )
             if result:
                 logger.info(
-                    "Agent code submitted to %s: hash=%s files=%s",
+                    "Agent code submitted to %s: hash=%s listener=%s files=%s",
                     db_url, result.get("code_hash", "?")[:24],
-                    sorted(bundle["files"].keys()),
+                    listener_url, sorted(bundle["files"].keys()),
                 )
             else:
                 logger.error("Agent code submission failed")
@@ -186,6 +188,26 @@ async def submit_agent_code(self):
 
         self._code_hash = code_hash
 
+    async def heartbeat_listener(self):
+        """Refresh listener_url on the DB so validators keep seeing us.
+
+        The /miners/active endpoint filters on listener_seen_at, so
+        without periodic re-registration the miner drops out of the
+        active set after the TTL window even if it's still online.
+        """
+        db_url = Config.DB_API_URL
+        if not db_url or not self.api_key:
+            return
+        listener_url = self._listener_external_url()
+        from shared.db_client import DatabaseClient
+        db = DatabaseClient(db_url=db_url, api_key=self.api_key)
+        try:
+            await db.register_listener(listener_url)
+        except Exception as e:
+            logger.warning("Listener heartbeat failed: %s", e)
+        finally:
+            await db.close()
+
     async def _post_trainer_ready(
         self, round_id: int, deployment: Deployment, validator_db_url: str,
     ):
@@ -736,6 +758,8 @@ async def run(self):
                 )
                 if not self._code_hash:
                     await self.submit_agent_code()
+                else:
+                    await self.heartbeat_listener()
             except Exception as e:
                 logger.warning("Heartbeat tick failed: %s", e)
 

shared/db_client.py

@@ -187,11 +187,42 @@ async def get_agent_code_by_hash(self, code_hash: str) -> Optional[dict]:
         return await self._get(f"/agent_code/by_hash/{code_hash}")
 
     async def submit_agent_code(
-        self, files: dict[str, str], entry_point: str = "agent.py",
+        self,
+        files: dict[str, str],
+        entry_point: str = "agent.py",
+        listener_url: str = "",
     ) -> Optional[dict]:
-        return await self._post("/agent_code", {
-            "files": files, "entry_point": entry_point,
-        })
+        body: dict = {"files": files, "entry_point": entry_point}
+        if listener_url:
+            body["listener_url"] = listener_url
+        return await self._post("/agent_code", body)
+
+    async def register_listener(self, listener_url: str) -> Optional[dict]:
+        """Refresh this miner's listener URL on the DB registry.
+
+        Bearer-authed — the miner must already have an agent_submissions
+        row (i.e. submit_agent_code has been called at least once).
+        """
+        return await self._post(
+            "/miners/me/listener", {"listener_url": listener_url},
+        )
+
+    async def get_active_miners(
+        self, window_seconds: float = 6 * 60 * 60,
+    ) -> list[dict]:
+        """Recent miners with live listener registrations.
+
+        HMAC-authed. Returns the list raw from the DB; callers project
+        the fields they need (typically hotkey + listener_url + code_hash).
+        """
+        result = await self._get(
+            "/miners/active", params={"window_seconds": window_seconds},
+        )
+        if isinstance(result, dict):
+            miners = result.get("miners")
+            if isinstance(miners, list):
+                return miners
+        return []
 
     async def submit_training_meta(
         self, round_id: int, hotkey: str, meta: dict,

shared/pg_schema.py

@@ -157,11 +157,20 @@
     r2_key TEXT NOT NULL,
     round_submitted BIGINT NOT NULL DEFAULT -1,
     timestamp DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    listener_url TEXT NOT NULL DEFAULT '',
+    listener_seen_at DOUBLE PRECISION NOT NULL DEFAULT 0.0,
     UNIQUE(hotkey)
 );
 ALTER TABLE agent_submissions ALTER COLUMN round_submitted TYPE BIGINT;
+ALTER TABLE agent_submissions
+    ADD COLUMN IF NOT EXISTS listener_url TEXT NOT NULL DEFAULT '';
+ALTER TABLE agent_submissions
+    ADD COLUMN IF NOT EXISTS listener_seen_at DOUBLE PRECISION NOT NULL DEFAULT 0.0;
 CREATE INDEX IF NOT EXISTS idx_agent_hotkey ON agent_submissions(hotkey);
 CREATE INDEX IF NOT EXISTS idx_agent_hash ON agent_submissions(code_hash);
+CREATE INDEX IF NOT EXISTS idx_agent_listener_seen
+    ON agent_submissions(listener_seen_at)
+    WHERE listener_url <> '';
 
 -- Append-only history of every agent submission. Unlike agent_submissions
 -- (one row per hotkey, overwritten on each upload), this preserves the full