From b06ef5e6a8e7d51a43ee47fe690ba60aa80205b6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 22 May 2026 05:56:43 +0000 Subject: [PATCH 1/2] miner listener registry: close the off-chain commitments gap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The post-bittensor refactor dropped on-chain ImageCommitments and hard-coded `commitments = {}` in the validator with a comment claiming downstream paths would treat that as "every registered miner is eligible" — but coordinator/collection both iterate `commitments.keys()`, so every round dead-ended at "0 miners with listener_urls" and "No valid submissions". Miners now self-register their warm-standby listener URL via POST /agent_code (carried as a new optional field) and refresh it on each heartbeat via the new POST /miners/me/listener. Validators discover live miners through GET /miners/active (HMAC-authed, 6h freshness window) and build the equivalent ImageCommitment dict at the start of every round, using miner-posted miner_uid when set and a stable hotkey-derived synthetic uid as fallback. Schema adds `listener_url` + `listener_seen_at` to agent_submissions; existing rows keep working because the upsert's CASE clause preserves the prior URL when a caller posts agent code without one. --- CLAUDE.md | 1 + database/server.py | 136 ++++++++- miner/neuron.py | 28 +- shared/db_client.py | 39 ++- shared/pg_schema.py | 9 + tests/test_miner_listener_registry.py | 383 ++++++++++++++++++++++++++ validator/neuron.py | 73 ++++- 7 files changed, 656 insertions(+), 13 deletions(-) create mode 100644 tests/test_miner_listener_registry.py diff --git a/CLAUDE.md b/CLAUDE.md index 8f85253..4ba0f0e 100644 --- a/CLAUDE.md +++ b/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) diff --git a/database/server.py b/database/server.py index 78a34ca..c71566e 100644 --- a/database/server.py +++ b/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,6 +798,7 @@ 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(): @@ -801,18 +806,28 @@ async def submit_agent_code(request: Request, req: SubmitAgentCodeRequest): """ 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``. diff --git a/miner/neuron.py b/miner/neuron.py index 06ca6db..4af8b47 100644 --- a/miner/neuron.py +++ b/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) diff --git a/shared/db_client.py b/shared/db_client.py index c3d2582..85259aa 100644 --- a/shared/db_client.py +++ b/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, diff --git a/shared/pg_schema.py b/shared/pg_schema.py index b5a23d1..81aa72f 100644 --- a/shared/pg_schema.py +++ b/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 diff --git a/tests/test_miner_listener_registry.py b/tests/test_miner_listener_registry.py new file mode 100644 index 0000000..0b272eb --- /dev/null +++ b/tests/test_miner_listener_registry.py @@ -0,0 +1,383 @@ +"""Tests for the miner listener registry. + +The off-chain deploy has no on-chain ImageCommitment to carry a +miner's listener_url, so miners self-register their URL via +``POST /miners/me/listener`` and validators fetch the live list via +``GET /miners/active``. Together these close the gap that previously +left ``commitments`` empty and caused every round to dead-end with +"0 miners with listener_urls". + +Covered: + * ``POST /agent_code`` stores the optional ``listener_url`` field. + * ``POST /miners/me/listener`` refreshes the registry row. + * ``GET /miners/active`` returns only rows within the freshness + window and with a non-empty URL. + * ``DatabaseClient`` round-trips through the new endpoints. + * ``Validator._fetch_active_commitments`` builds an + ``ImageCommitment`` dict from the DB response with stable + synthetic UIDs. +""" + +from __future__ import annotations + +import time +from contextlib import asynccontextmanager, contextmanager +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + + +# ── Fakes that mirror the agent_code test wiring ───────────────────── + + +class _FakeR2: + def __init__(self): + self.uploads: list[tuple[str, dict]] = [] + + def upload_json(self, key, data): + self.uploads.append((key, dict(data))) + + +class _FakeRow(dict): + """Behaves like an asyncpg Record for the bits the route reads.""" + + +class _FakeConn: + def __init__(self, pool): + self.pool = pool + + async def execute(self, sql, *args): + self.pool.executed.append((sql, args)) + # /agent_code uses INSERT … ON CONFLICT — pretend one row was + # touched so the path treats it as success. + return "INSERT 0 1" + + @asynccontextmanager + async def transaction(self): + yield + + +class _FakePool: + def __init__(self): + self.executed: list[tuple[str, tuple]] = [] + self.rows: list[dict] = [] + # Default to "update touched a row" for the heartbeat endpoint. + self.update_rowcount: int = 1 + + @asynccontextmanager + async def acquire(self): + yield _FakeConn(self) + + async def execute(self, sql, *args): + self.executed.append((sql, args)) + return f"UPDATE {self.update_rowcount}" + + async def fetch(self, sql, *args): + return [_FakeRow(r) for r in self.rows] + + async def fetchrow(self, sql, *args): + return None + + +@contextmanager +def _wire_app(monkeypatch, *, hotkey: str = "hk-A"): + from database import server as srv + from shared import miner_auth + + pool = _FakePool() + r2 = _FakeR2() + monkeypatch.setattr(srv, "_pool", pool) + monkeypatch.setattr(srv, "_r2", r2) + monkeypatch.setattr(srv, "_current_challenge", {"round_id": 7}) + srv._rate_window.clear() + srv._ip_rate_window.clear() + + async def fake_lookup(_pool, token): + if token == "good": + return miner_auth.MinerIdentity( + miner_id="m1", key_id="k1", scope="miner", + hotkey=hotkey, name="alice", + ) + return None + + monkeypatch.setattr(miner_auth, "lookup_bearer", fake_lookup) + + app = FastAPI() + app.include_router(srv.validator_router) + + @app.middleware("http") + async def _route_auth(request, call_next): + path = request.url.path + method = request.method + if path == "/agent_code" and method == "POST": + return await srv._bearer_auth_then_call_for_agent_code( + request, call_next, + ) + if path == "/miners/me/listener" and method == "POST": + # Same bearer wrapper /miners/me/* normally uses in the real + # app, just exercised here on its own. + return await srv._bearer_auth_then_call(request, call_next) + return await call_next(request) + + yield app, pool, r2 + + +def _bundle(): + return { + "files": {"agent.py": "def design_architecture(*a, **k):\n return {}\n"}, + "entry_point": "agent.py", + } + + +# ── POST /agent_code stores listener_url ───────────────────────────── + + +def test_post_agent_code_stores_listener_url(monkeypatch): + """Listener URL flows through to the SQL parameters.""" + with _wire_app(monkeypatch) as (app, pool, _): + client = TestClient(app) + body = _bundle() | {"listener_url": "http://1.2.3.4:8090"} + r = client.post( + "/agent_code", json=body, + headers={"Authorization": "Bearer good"}, + ) + assert r.status_code == 200, r.text + upsert_sqls = [sql for sql, _ in pool.executed if "agent_submissions" in sql] + assert upsert_sqls, "expected an INSERT into agent_submissions" + upsert_args = next( + args for sql, args in pool.executed if "agent_submissions" in sql + ) + assert "http://1.2.3.4:8090" in upsert_args + + +def test_post_agent_code_without_listener_url_passes_empty(monkeypatch): + """Omitting the field is fine — the column gets an empty string and + the upsert preserves any previously-stored value via the CASE clause.""" + with _wire_app(monkeypatch) as (app, pool, _): + client = TestClient(app) + r = client.post( + "/agent_code", json=_bundle(), + headers={"Authorization": "Bearer good"}, + ) + assert r.status_code == 200, r.text + args = next(a for sql, a in pool.executed if "agent_submissions" in sql) + assert "" in args # empty listener_url positional arg + + +# ── POST /miners/me/listener ───────────────────────────────────────── + + +def test_register_listener_updates_row(monkeypatch): + with _wire_app(monkeypatch) as (app, pool, _): + client = TestClient(app) + r = client.post( + "/miners/me/listener", + json={"listener_url": "http://5.6.7.8:8090"}, + headers={"Authorization": "Bearer good"}, + ) + assert r.status_code == 200, r.text + assert r.json() == { + "status": "ok", "listener_url": "http://5.6.7.8:8090", + } + update_sqls = [sql for sql, _ in pool.executed if "UPDATE" in sql.upper()] + assert update_sqls, "expected UPDATE on agent_submissions" + + +def test_register_listener_rejects_empty(monkeypatch): + with _wire_app(monkeypatch) as (app, _, _): + client = TestClient(app) + r = client.post( + "/miners/me/listener", + json={"listener_url": " "}, + headers={"Authorization": "Bearer good"}, + ) + assert r.status_code == 400 + + +def test_register_listener_404_when_no_agent_submission(monkeypatch): + """If the miner hasn't posted agent code yet there's no row to + UPDATE — make sure that surfaces as a 404 so the miner knows to + call submit_agent_code first.""" + with _wire_app(monkeypatch) as (app, pool, _): + pool.update_rowcount = 0 + client = TestClient(app) + r = client.post( + "/miners/me/listener", + json={"listener_url": "http://x:9"}, + headers={"Authorization": "Bearer good"}, + ) + assert r.status_code == 404 + + +def test_register_listener_requires_auth(monkeypatch): + with _wire_app(monkeypatch) as (app, _, _): + client = TestClient(app) + r = client.post( + "/miners/me/listener", json={"listener_url": "http://x:9"}, + ) + assert r.status_code == 401 + + +# ── GET /miners/active ─────────────────────────────────────────────── + + +def test_list_active_miners_returns_rows(monkeypatch): + with _wire_app(monkeypatch) as (app, pool, _): + now = time.time() + pool.rows = [ + { + "hotkey": "hk-A", "miner_uid": 1, "code_hash": "ab" * 32, + "entry_point": "agent.py", + "listener_url": "http://1.1.1.1:8090", + "listener_seen_at": now - 10.0, + "round_submitted": 100, "timestamp": now - 30.0, + }, + { + "hotkey": "hk-B", "miner_uid": -1, "code_hash": "cd" * 32, + "entry_point": "agent.py", + "listener_url": "http://2.2.2.2:8090", + "listener_seen_at": now - 60.0, + "round_submitted": 99, "timestamp": now - 90.0, + }, + ] + client = TestClient(app) + # Public GET — middleware in the wire fixture doesn't gate it, + # mirroring how the validator-router /miners/active is HMAC-authed + # at the real middleware layer (separately tested for /agent_code). + r = client.get("/miners/active") + assert r.status_code == 200, r.text + miners = r.json()["miners"] + assert len(miners) == 2 + assert {m["hotkey"] for m in miners} == {"hk-A", "hk-B"} + assert miners[0]["listener_url"] == "http://1.1.1.1:8090" + + +# ── DatabaseClient round-trip ──────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_db_client_submit_agent_code_passes_listener_url(): + from shared.db_client import DatabaseClient + + db = DatabaseClient(db_url="http://fake:8090", api_key="rdrk_x") + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"status": "ok", "code_hash": "ab" * 32} + mock_resp.raise_for_status = MagicMock() + + with patch.object(db, "_get_client") as mock_get: + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_resp) + mock_get.return_value = mock_client + await db.submit_agent_code( + files={"agent.py": "x"}, + entry_point="agent.py", + listener_url="http://m:9", + ) + # _post sends a JSON-encoded body via content=, not json= + import json as _json + sent = _json.loads(mock_client.post.call_args.kwargs["content"]) + assert sent["listener_url"] == "http://m:9" + assert mock_client.post.call_args.args[0].endswith("/agent_code") + + +@pytest.mark.asyncio +async def test_db_client_register_listener(): + from shared.db_client import DatabaseClient + + db = DatabaseClient(db_url="http://fake:8090", api_key="rdrk_x") + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = {"status": "ok"} + mock_resp.raise_for_status = MagicMock() + with patch.object(db, "_get_client") as mock_get: + mock_client = AsyncMock() + mock_client.post = AsyncMock(return_value=mock_resp) + mock_get.return_value = mock_client + await db.register_listener("http://m:9") + import json as _json + sent = _json.loads(mock_client.post.call_args.kwargs["content"]) + assert sent == {"listener_url": "http://m:9"} + assert mock_client.post.call_args.args[0].endswith( + "/miners/me/listener", + ) + + +@pytest.mark.asyncio +async def test_db_client_get_active_miners(): + from shared.db_client import DatabaseClient + + db = DatabaseClient(db_url="http://fake:8090", service_secret=b"k") + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.return_value = { + "miners": [{"hotkey": "hk", "listener_url": "u"}], + } + mock_resp.raise_for_status = MagicMock() + with patch.object(db, "_get_client") as mock_get: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_resp) + mock_get.return_value = mock_client + out = await db.get_active_miners() + assert out == [{"hotkey": "hk", "listener_url": "u"}] + + +# ── Validator._fetch_active_commitments ────────────────────────────── + + +@pytest.mark.asyncio +async def test_validator_fetch_active_commitments_builds_dict(): + """The validator turns the JSON rows into ImageCommitment objects + keyed by a stable int UID — preferring the miner-posted uid, and + falling back to a hotkey-derived synthetic uid so different miners + never share a key.""" + from shared.commitment import ImageCommitment + from validator.neuron import Validator + + fake_db = MagicMock() + fake_db.get_active_miners = AsyncMock(return_value=[ + { + "hotkey": "hk-A", "miner_uid": 7, "code_hash": "ab" * 32, + "listener_url": "http://a:9", + }, + { + "hotkey": "hk-B", "miner_uid": -1, "code_hash": "cd" * 32, + "listener_url": "http://b:9", + }, + { + # Empty listener — must be skipped. + "hotkey": "hk-C", "miner_uid": -1, "code_hash": "ef" * 32, + "listener_url": "", + }, + ]) + + v = Validator.__new__(Validator) + v.db_client = fake_db + + out = await v._fetch_active_commitments() + + assert len(out) == 2 + assert 7 in out + assert out[7].listener_url == "http://a:9" + assert out[7].hotkey == "hk-A" + assert all(isinstance(v, ImageCommitment) for v in out.values()) + # The hk-B uid is synthetic (positive int, not the preferred 7). + other_uid = next(u for u in out if u != 7) + assert other_uid > 0 + assert out[other_uid].hotkey == "hk-B" + + +@pytest.mark.asyncio +async def test_validator_fetch_active_commitments_returns_empty_on_failure(): + from validator.neuron import Validator + + fake_db = MagicMock() + fake_db.get_active_miners = AsyncMock(side_effect=RuntimeError("boom")) + + v = Validator.__new__(Validator) + v.db_client = fake_db + + out = await v._fetch_active_commitments() + assert out == {} diff --git a/validator/neuron.py b/validator/neuron.py index 8d31d98..45cc869 100644 --- a/validator/neuron.py +++ b/validator/neuron.py @@ -27,6 +27,7 @@ from shared.challenge import ( generate_challenge, round_start_block, current_phase, select_task, ) +from shared.commitment import ImageCommitment from shared.database import DataElement from shared.db_client import DatabaseClient from shared.protocol import Proposal @@ -337,6 +338,65 @@ async def _wait_until_block(self, target_block: int): remaining = (target_block - current) * 12 await asyncio.sleep(min(remaining, 30)) + async def _fetch_active_commitments(self) -> dict[int, ImageCommitment]: + """Build {uid: ImageCommitment} from the DB's active miner list. + + The off-chain deploy has no on-chain ImageCommitment; miners + self-register their listener URL via POST /miners/me/listener + after submit_agent_code. We synthesise an ImageCommitment per + active miner so the rest of Phase A/B/C (which is still typed + around commitments) keeps working. + + UIDs: + * Prefer agent_submissions.miner_uid when the miner posted + one via X-Miner-UID (lets operators keep stable, small ids). + * Otherwise derive a stable synthetic int from the hotkey. + Collisions are vanishingly unlikely in practice but we + re-roll on conflict so two miners never share a uid. + """ + if self.db_client is None: + logger.warning("No db_client configured — cannot fetch active miners") + return {} + + try: + rows = await self.db_client.get_active_miners() + except Exception as e: + logger.warning("get_active_miners failed: %s", e) + return {} + + out: dict[int, ImageCommitment] = {} + for row in rows: + hotkey = row.get("hotkey") or "" + listener_url = row.get("listener_url") or "" + if not hotkey or not listener_url: + continue + preferred = int(row.get("miner_uid", -1)) + uid = preferred if preferred >= 0 and preferred not in out else -1 + if uid < 0: + base = int.from_bytes( + hashlib.sha256(hotkey.encode()).digest()[:4], + "big", + ) % (2**31 - 1) + uid = base + # Re-roll on collision — extremely rare but cheap. + bump = 0 + while uid in out and bump < 16: + bump += 1 + uid = (base + bump) % (2**31 - 1) + if uid in out: + logger.warning( + "Skipping miner %s — could not assign unique synthetic uid", + hotkey[:16], + ) + continue + out[uid] = ImageCommitment( + code_hash=row.get("code_hash", "") or "", + listener_url=listener_url, + miner_uid=uid, + hotkey=hotkey, + ) + return out + async def _cache_training_metas( self, round_id: int, metas: dict[int, dict], ) -> None: @@ -489,10 +549,17 @@ async def run_round(self): # ── PHASE A: RUN AGENTS + COLLECT ──────────────────── # Off-chain deploy: miner identity comes from operator-CLI tokens, - # not on-chain ImageCommitments. Commitments stay empty; downstream - # code paths treat that as "every registered miner is eligible". - commitments: dict = {} + # not on-chain ImageCommitments. We rebuild the equivalent record + # by asking the DB for miners that have recently registered a + # listener URL (POST /miners/me/listener). Without this lookup + # `commitments` would be empty and the round would dead-end at + # "0 miners with listener_urls". + commitments = await self._fetch_active_commitments() validator_uids: list[int] = [] + logger.info( + "Active miner registry: %d miners with live listener URLs", + len(commitments), + ) if not self.r2: logger.warning("No R2 configured — skipping Phase A/B/C") From 4b64938460ca33d46d5c4018e9fcebe2e57bc2c7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 22 May 2026 06:15:20 +0000 Subject: [PATCH 2/2] db migration: 006_agent_listener_registry CI's "Schema change requires migration" check flagged pg_schema.py changes without a companion migration file. Adds the listener_url + listener_seen_at columns and partial index for the active-miner freshness query, guarded by an information_schema check so the migration runner's empty-schema bootstrap pass stays a no-op. --- .../006_agent_listener_registry.sql | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 database/migrations/006_agent_listener_registry.sql diff --git a/database/migrations/006_agent_listener_registry.sql b/database/migrations/006_agent_listener_registry.sql new file mode 100644 index 0000000..555c3bd --- /dev/null +++ b/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 $$;