diff --git a/packages/challenges/agent-challenge/src/agent_challenge/evaluation/raw_weight_push.py b/packages/challenges/agent-challenge/src/agent_challenge/evaluation/raw_weight_push.py index b4f088e00..c8aecc7bd 100644 --- a/packages/challenges/agent-challenge/src/agent_challenge/evaluation/raw_weight_push.py +++ b/packages/challenges/agent-challenge/src/agent_challenge/evaluation/raw_weight_push.py @@ -92,6 +92,56 @@ class PushAttemptResult: error: str | None = None +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def _pending_still_fresh( + payload: RawWeightPushRequest, + *, + now: datetime, + skew_seconds: int = 30, +) -> bool: + """Mirror master receipt policy lower/upper bound for local pending reuse.""" + + expires = _as_utc(payload.expires_at) + receipt = _as_utc(now) + return receipt < (expires + timedelta(seconds=int(skew_seconds))) + + +def _safe_response_detail(response: httpx.Response) -> str: + """Extract a short rejection detail without logging secrets or full bodies.""" + + try: + data = response.json() + except Exception: # noqa: BLE001 + return "" + if not isinstance(data, dict): + return "" + detail = data.get("detail") + if isinstance(detail, str): + return detail[:300] + if isinstance(detail, list): + # FastAPI validation error shape — field locs only. + parts: list[str] = [] + for item in detail[:5]: + if isinstance(item, dict): + loc = ".".join(str(x) for x in item.get("loc", ())) + if loc: + parts.append(loc) + return ",".join(parts)[:300] + return "" + + +def _is_all_zero_weights(weights: Mapping[str, float]) -> bool: + return bool(weights) and all(float(value) <= 0.0 for value in weights.values()) + + +NON_RETRYABLE_PUSH_STATUSES = frozenset({409, 413, 415, 422}) + + class RawWeightPushLedger(Base): """Singleton SQLAlchemy row for durable attempt/ack checkpointing. @@ -196,6 +246,22 @@ async def record_pending( row.updated_at = attempted_at await session.commit() + async def clear_pending(self) -> None: + """Drop a non-retryable pending snapshot so the next push rebuilds fresh bytes.""" + + async with self.database.session() as session: + row = await self._get_row(session) + if row is None: + return + row.pending_epoch = None + row.pending_revision = None + row.pending_payload_digest = None + row.pending_canonical_payload = None + row.pending_nonce = None + row.pending_attempted_at = None + row.updated_at = datetime.now(UTC).isoformat() + await session.commit() + async def acknowledge( self, *, @@ -315,6 +381,8 @@ def _build_payload( } if not body["weights"]: raise ValueError("empty weight map has no push surface") + if _is_all_zero_weights(body["weights"]): + raise ValueError("all-zero weight map has no push surface") digest = RawWeightPushRequest.compute_digest(body) body["payload_digest"] = digest payload = RawWeightPushRequest.model_validate(body) @@ -375,6 +443,19 @@ async def push_once( pending_bytes = str(pending["canonical_payload"]).encode("utf-8") payload = RawWeightPushRequest.model_validate_json(pending_bytes) raw_bytes = payload.canonical_bytes() + if not _pending_still_fresh(payload, now=now): + logger.info( + "raw weight pending expired; rebuilding", + extra={ + "epoch": payload.epoch, + "revision": payload.revision, + "digest": payload.payload_digest[:12], + }, + ) + await self.store.clear_pending() + pending = None + payload = None + raw_bytes = None except Exception: # noqa: BLE001 - corrupt pending is rebuilt pending = None payload = None @@ -402,6 +483,16 @@ async def push_once( cursor_advanced=False, error="empty weights", ) + if _is_all_zero_weights(cleaned): + return PushAttemptResult( + status="skipped_zero", + epoch=0, + revision=0, + payload_digest="", + snapshot_id=None, + cursor_advanced=False, + error="all-zero weight map", + ) # Host-trust path: never silently push unattested scores as if TEE-backed. _log_unattested_weight_push_if_no_phala(cleaned) resolved_epoch = ( @@ -477,6 +568,22 @@ async def push_once( error=f"status={response.status_code}", ) if response.status_code not in {200, 201}: + detail = _safe_response_detail(response) + logger.info( + "raw weight push rejected", + extra={ + "status": response.status_code, + "detail": detail, + "epoch": payload.epoch, + "revision": payload.revision, + "digest": payload.payload_digest[:12], + }, + ) + if response.status_code in NON_RETRYABLE_PUSH_STATUSES: + await self.store.clear_pending() + err = f"status={response.status_code}" + if detail: + err = f"{err} detail={detail}" return PushAttemptResult( status="rejected", epoch=payload.epoch, @@ -484,7 +591,7 @@ async def push_once( payload_digest=payload.payload_digest, snapshot_id=None, cursor_advanced=False, - error=f"status={response.status_code}", + error=err, ) try: ack = RawWeightPushAcknowledgement.model_validate(response.json()) diff --git a/packages/challenges/agent-challenge/tests/test_ac_raw_weight_push.py b/packages/challenges/agent-challenge/tests/test_ac_raw_weight_push.py index d8bdcb716..64141eb29 100644 --- a/packages/challenges/agent-challenge/tests/test_ac_raw_weight_push.py +++ b/packages/challenges/agent-challenge/tests/test_ac_raw_weight_push.py @@ -277,3 +277,99 @@ def handler(request: httpx.Request) -> httpx.Response: for record in caplog.records: assert TOKEN not in str(record.__dict__) await http.aclose() + + +@pytest.mark.asyncio +async def test_expired_pending_is_rebuilt_not_retried(database: Database) -> None: + clock = FakeClock() + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + parsed = RawWeightPushRequest.model_validate_json(request.content) + seen.append(parsed.nonce) + return httpx.Response( + 200, + json={ + "protocol_version": "1.0", + "challenge_slug": SLUG, + "epoch": parsed.epoch, + "revision": parsed.revision, + "snapshot_id": "snap-ac-rebuilt", + "payload_digest": parsed.payload_digest, + "accepted": True, + "idempotent": False, + }, + ) + + http = httpx.AsyncClient( + transport=httpx.MockTransport(handler), + base_url="http://master.test", + ) + client = RawWeightPushClient( + database=database, + challenge_slug=SLUG, + master_base_url="http://master.test", + shared_token=TOKEN, + now_fn=clock.now, + http_client=http, + epoch_fn=lambda: 99, + freshness_seconds=60, + ) + await client.init() + old = clock.now() - timedelta(hours=12) + stale_payload, stale_raw = client._build_payload( + weights={WINNER: 1.0}, + epoch=1, + revision=1, + nonce="ac-stale-nonce", + now=old, + ) + await client.store.record_pending( + epoch=stale_payload.epoch, + revision=stale_payload.revision, + payload_digest=stale_payload.payload_digest, + canonical_payload=stale_raw.decode("utf-8"), + nonce=stale_payload.nonce, + attempted_at=old.isoformat(), + ) + with activate_role(Role.CHALLENGE): + result = await client.push_once(weights={WINNER: 1.0}, epoch=99, reuse_pending=True) + assert result.status == "acknowledged" + assert len(seen) == 1 + assert seen[0] != "ac-stale-nonce" + assert await client.store.get_pending() is None + await http.aclose() + + +@pytest.mark.asyncio +async def test_all_zero_weights_skipped_locally(database: Database) -> None: + clock = FakeClock() + calls: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(request) + return httpx.Response(500) + + http = httpx.AsyncClient( + transport=httpx.MockTransport(handler), + base_url="http://master.test", + ) + client = RawWeightPushClient( + database=database, + challenge_slug=SLUG, + master_base_url="http://master.test", + shared_token=TOKEN, + now_fn=clock.now, + http_client=http, + epoch_fn=lambda: 1, + ) + await client.init() + with activate_role(Role.CHALLENGE): + result = await client.push_once( + weights={"5GziQCcRpN8NCJktX343brnfuVe3w6gUYieeStXPD1Dag2At": 0.0}, + epoch=1, + ) + assert result.cursor_advanced is False + assert result.status in {"skipped_empty", "skipped_zero"} + assert calls == [] + await http.aclose() diff --git a/packages/challenges/prism/src/prism_challenge/raw_weight_push.py b/packages/challenges/prism/src/prism_challenge/raw_weight_push.py index f838ac2d2..bc1e84226 100644 --- a/packages/challenges/prism/src/prism_challenge/raw_weight_push.py +++ b/packages/challenges/prism/src/prism_challenge/raw_weight_push.py @@ -84,6 +84,55 @@ class PushAttemptResult: error: str | None = None +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def _pending_still_fresh( + payload: RawWeightPushRequest, + *, + now: datetime, + skew_seconds: int = 30, +) -> bool: + """Mirror master receipt policy lower/upper bound for local pending reuse.""" + + expires = _as_utc(payload.expires_at) + receipt = _as_utc(now) + return receipt < (expires + timedelta(seconds=int(skew_seconds))) + + +def _safe_response_detail(response: httpx.Response) -> str: + """Extract a short rejection detail without logging secrets or full bodies.""" + + try: + data = response.json() + except Exception: # noqa: BLE001 + return "" + if not isinstance(data, dict): + return "" + detail = data.get("detail") + if isinstance(detail, str): + return detail[:300] + if isinstance(detail, list): + # FastAPI validation error shape — field locs only. + parts: list[str] = [] + for item in detail[:5]: + if isinstance(item, dict): + loc = ".".join(str(x) for x in item.get("loc", ())) + if loc: + parts.append(loc) + return ",".join(parts)[:300] + return "" + + +def _is_all_zero_weights(weights: Mapping[str, float]) -> bool: + return bool(weights) and all(float(value) <= 0.0 for value in weights.values()) + + +NON_RETRYABLE_PUSH_STATUSES = frozenset({409, 413, 415, 422}) + RAW_WEIGHT_PUSH_SCHEMA = ( "CREATE TABLE IF NOT EXISTS raw_weight_push_ledger (" "id INTEGER PRIMARY KEY CHECK (id = 1)," @@ -204,6 +253,20 @@ async def record_pending( ), ) + async def clear_pending(self) -> None: + """Drop a non-retryable pending snapshot so the next push rebuilds fresh bytes.""" + + async with self.database.connect() as conn: + await ensure_raw_weight_push_schema(conn) + await conn.execute( + "UPDATE raw_weight_push_ledger SET " + "pending_epoch=NULL, pending_revision=NULL, " + "pending_payload_digest=NULL, pending_canonical_payload=NULL, " + "pending_nonce=NULL, pending_attempted_at=NULL, " + "updated_at=? WHERE id = 1", + (datetime.now(UTC).isoformat(),), + ) + async def acknowledge( self, *, @@ -326,8 +389,9 @@ def _build_payload( # push. Caller must supply either positive weights or an explicit zero map # with at least one hotkey. Empty still rejects at schema if empty dict. if not body["weights"]: - # Persist no network attempt; local explicit zero snapshot identity. raise ValueError("empty weight map has no push surface") + if _is_all_zero_weights(body["weights"]): + raise ValueError("all-zero weight map has no push surface") digest = RawWeightPushRequest.compute_digest(body) body["payload_digest"] = digest payload = RawWeightPushRequest.model_validate(body) @@ -388,6 +452,19 @@ async def push_once( pending_bytes = str(pending["canonical_payload"]).encode("utf-8") payload = RawWeightPushRequest.model_validate_json(pending_bytes) raw_bytes = payload.canonical_bytes() + if not _pending_still_fresh(payload, now=now): + logger.info( + "raw weight pending expired; rebuilding", + extra={ + "epoch": payload.epoch, + "revision": payload.revision, + "digest": payload.payload_digest[:12], + }, + ) + await self.store.clear_pending() + pending = None + payload = None + raw_bytes = None except Exception: # noqa: BLE001 - corrupt pending is rebuilt pending = None payload = None @@ -421,6 +498,16 @@ async def push_once( cursor_advanced=False, error="empty weights", ) + if _is_all_zero_weights(cleaned): + return PushAttemptResult( + status="skipped_zero", + epoch=0, + revision=0, + payload_digest="", + snapshot_id=None, + cursor_advanced=False, + error="all-zero weight map", + ) resolved_epoch = ( int(epoch) if epoch is not None @@ -493,6 +580,22 @@ async def push_once( error=f"status={response.status_code}", ) if response.status_code not in {200, 201}: + detail = _safe_response_detail(response) + logger.info( + "raw weight push rejected", + extra={ + "status": response.status_code, + "detail": detail, + "epoch": payload.epoch, + "revision": payload.revision, + "digest": payload.payload_digest[:12], + }, + ) + if response.status_code in NON_RETRYABLE_PUSH_STATUSES: + await self.store.clear_pending() + err = f"status={response.status_code}" + if detail: + err = f"{err} detail={detail}" return PushAttemptResult( status="rejected", epoch=payload.epoch, @@ -500,7 +603,7 @@ async def push_once( payload_digest=payload.payload_digest, snapshot_id=None, cursor_advanced=False, - error=f"status={response.status_code}", + error=err, ) try: ack = RawWeightPushAcknowledgement.model_validate(response.json()) diff --git a/packages/challenges/prism/tests/test_raw_weight_push.py b/packages/challenges/prism/tests/test_raw_weight_push.py index c78d8569e..116d152ba 100644 --- a/packages/challenges/prism/tests/test_raw_weight_push.py +++ b/packages/challenges/prism/tests/test_raw_weight_push.py @@ -364,3 +364,141 @@ def internal_token(self) -> str: assert client is not None assert client.epoch_fn is not None assert client.epoch_fn() == last_epoch + 1 + + +@pytest.mark.asyncio +async def test_expired_pending_is_rebuilt_not_retried(database: Database) -> None: + """Prod incident: stale pending with expired freshness must not loop 422 forever.""" + + clock = FakeClock() + seen: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + parsed = RawWeightPushRequest.model_validate_json(request.content) + seen.append( + { + "nonce": parsed.nonce, + "computed_at": parsed.computed_at.isoformat(), + "expires_at": parsed.expires_at.isoformat(), + "epoch": parsed.epoch, + } + ) + return httpx.Response( + 200, + json={ + "protocol_version": "1.0", + "challenge_slug": SLUG, + "epoch": parsed.epoch, + "revision": parsed.revision, + "snapshot_id": "snap-rebuilt", + "payload_digest": parsed.payload_digest, + "accepted": True, + "idempotent": False, + }, + ) + + http = httpx.AsyncClient( + transport=httpx.MockTransport(handler), + base_url="http://master.test", + ) + client = RawWeightPushClient( + database=database, + challenge_slug=SLUG, + master_base_url="http://master.test", + shared_token=TOKEN, + now_fn=clock.now, + http_client=http, + epoch_fn=lambda: 20, + freshness_seconds=60, + ) + await client.init() + # Seed an expired pending snapshot (mirrors prod ledger). + old = clock.now() - timedelta(hours=24) + stale_payload, stale_raw = client._build_payload( + weights={HOTKEY: 1.0}, + epoch=1, + revision=1, + nonce="stale-nonce", + now=old, + ) + await client.store.record_pending( + epoch=stale_payload.epoch, + revision=stale_payload.revision, + payload_digest=stale_payload.payload_digest, + canonical_payload=stale_raw.decode("utf-8"), + nonce=stale_payload.nonce, + attempted_at=old.isoformat(), + ) + with activate_role(Role.CHALLENGE): + result = await client.push_once(weights={HOTKEY: 1.0}, epoch=20, reuse_pending=True) + assert result.status == "acknowledged" + assert result.cursor_advanced is True + assert len(seen) == 1 + assert seen[0]["nonce"] != "stale-nonce" + assert seen[0]["epoch"] == 20 + assert await client.store.get_pending() is None + await http.aclose() + + +@pytest.mark.asyncio +async def test_non_retryable_rejection_clears_pending_and_surfaces_detail( + database: Database, +) -> None: + clock = FakeClock() + transport = TransportQueue( + [ + httpx.Response(422, json={"detail": "snapshot outside freshness window"}), + ] + ) + http = httpx.AsyncClient( + transport=httpx.MockTransport(transport.handler), + base_url="http://master.test", + ) + client = RawWeightPushClient( + database=database, + challenge_slug=SLUG, + master_base_url="http://master.test", + shared_token=TOKEN, + now_fn=clock.now, + http_client=http, + epoch_fn=lambda: 5, + ) + await client.init() + with activate_role(Role.CHALLENGE): + result = await client.push_once(weights={HOTKEY: 1.0}, epoch=5) + assert result.status == "rejected" + assert result.cursor_advanced is False + assert result.error is not None + assert "422" in result.error + assert "snapshot outside freshness window" in result.error + assert await client.store.get_pending() is None + await http.aclose() + + +@pytest.mark.asyncio +async def test_all_zero_weights_skipped_locally(database: Database) -> None: + clock = FakeClock() + transport = TransportQueue([]) + http = httpx.AsyncClient( + transport=httpx.MockTransport(transport.handler), + base_url="http://master.test", + ) + client = RawWeightPushClient( + database=database, + challenge_slug=SLUG, + master_base_url="http://master.test", + shared_token=TOKEN, + now_fn=clock.now, + http_client=http, + epoch_fn=lambda: 1, + ) + await client.init() + with activate_role(Role.CHALLENGE): + result = await client.push_once( + weights={"5GziQCcRpN8NCJktX343brnfuVe3w6gUYieeStXPD1Dag2At": 0.0}, + epoch=1, + ) + assert result.cursor_advanced is False + assert result.status in {"skipped_empty", "skipped_zero"} + assert transport.requests == [] + await http.aclose() diff --git a/src/base/master/raw_weight_ingress.py b/src/base/master/raw_weight_ingress.py index c179f7b9e..71097fed2 100644 --- a/src/base/master/raw_weight_ingress.py +++ b/src/base/master/raw_weight_ingress.py @@ -16,6 +16,7 @@ import asyncio import hmac import inspect +import logging import uuid from collections.abc import Callable from dataclasses import dataclass @@ -65,6 +66,54 @@ UNSUPPORTED_MEDIA_DETAIL = "unsupported media type" UNSUPPORTED_VERSION_DETAIL = "unsupported protocol version" FUTURE_EPOCH_DETAIL = "epoch too far in the future" +ZERO_WEIGHT_DETAIL = "all-zero weight map rejected" + + +logger = logging.getLogger(__name__) + + +def _attributable_schema_detail(exc: ValidationError) -> str: + """Map pydantic errors to stable, non-secret rejection details.""" + + parts: list[str] = [] + for err in exc.errors()[:8]: + loc_parts = [str(item) for item in err.get("loc", ())] + loc = ".".join(loc_parts) + msg = str(err.get("msg", "") or "") + msg_l = msg.lower() + if "payload_digest" in loc or "payload_digest" in msg_l: + return DIGEST_DETAIL + if "expires_at" in msg_l and "computed_at" in msg_l: + return f"{SCHEMA_DETAIL} (expires_at)" + err_type = str(err.get("type", "") or "value_error") + if loc: + parts.append(f"{loc}:{err_type}") + else: + parts.append(err_type) + if not parts: + return SCHEMA_DETAIL + joined = ", ".join(parts) + return f"{SCHEMA_DETAIL} ({joined})" + + +def _log_rejection( + *, + challenge_slug: str, + reason: str, + status_code: int, + epoch: int | None = None, + revision: int | None = None, +) -> None: + logger.info( + "raw weight push rejected", + extra={ + "challenge_slug": challenge_slug, + "reason": reason, + "status_code": status_code, + "epoch": epoch, + "revision": revision, + }, + ) class RawWeightAuthError(PermissionError): @@ -374,13 +423,17 @@ def _parse_payload(self, raw_body: bytes) -> RawWeightPushRequest: # model_validate_json rejects non-object roots and coercion. payload = RawWeightPushRequest.model_validate_json(text) except ValidationError as exc: - raise RawWeightSchemaError(SCHEMA_DETAIL) from exc + raise RawWeightSchemaError(_attributable_schema_detail(exc)) from exc if payload.protocol_version not in self.accepted_versions: raise RawWeightSchemaError(UNSUPPORTED_VERSION_DETAIL) if not payload.protocol_version.startswith(f"{PROTOCOL_MAJOR}."): raise RawWeightSchemaError(UNSUPPORTED_VERSION_DETAIL) if len(payload.weights) > self.max_weight_keys: raise RawWeightSchemaError(PAYLOAD_TOO_LARGE_DETAIL) + if payload.weights and all( + float(weight) <= 0.0 for weight in payload.weights.values() + ): + raise RawWeightSchemaError(ZERO_WEIGHT_DETAIL) # Recompute server digest over the signed field set (exclude digest). canonical = payload.model_dump(mode="json", exclude={"payload_digest"}) expected = RawWeightPushRequest.compute_digest(canonical) @@ -883,6 +936,11 @@ async def push_raw_weights( get_weight_flow_metrics().record_push( outcome="rejected_auth", challenge_slug=slug ) + _log_rejection( + challenge_slug=slug, + reason=UNAUTHORIZED_DETAIL, + status_code=status.HTTP_401_UNAUTHORIZED, + ) raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=UNAUTHORIZED_DETAIL, @@ -891,6 +949,11 @@ async def push_raw_weights( get_weight_flow_metrics().record_push( outcome="rejected_forbidden", challenge_slug=slug ) + _log_rejection( + challenge_slug=slug, + reason=FORBIDDEN_DETAIL, + status_code=status.HTTP_403_FORBIDDEN, + ) raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=FORBIDDEN_DETAIL, @@ -899,6 +962,11 @@ async def push_raw_weights( get_weight_flow_metrics().record_push( outcome="conflict", challenge_slug=slug ) + _log_rejection( + challenge_slug=slug, + reason=CONFLICT_DETAIL, + status_code=status.HTTP_409_CONFLICT, + ) raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=CONFLICT_DETAIL, @@ -907,14 +975,25 @@ async def push_raw_weights( get_weight_flow_metrics().record_push( outcome="rejected_sealed", challenge_slug=slug ) + sealed_detail = str(exc) or SEALED_REVISION_DETAIL + _log_rejection( + challenge_slug=slug, + reason=sealed_detail, + status_code=status.HTTP_409_CONFLICT, + ) raise HTTPException( status_code=status.HTTP_409_CONFLICT, - detail=str(exc) or SEALED_REVISION_DETAIL, + detail=sealed_detail, ) from exc except RawWeightFreshnessError as exc: get_weight_flow_metrics().record_push( outcome="rejected_freshness", challenge_slug=slug ) + _log_rejection( + challenge_slug=slug, + reason=FRESHNESS_DETAIL, + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + ) raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=FRESHNESS_DETAIL, @@ -925,15 +1004,30 @@ async def push_raw_weights( ) message = str(exc) or SCHEMA_DETAIL if message == PAYLOAD_TOO_LARGE_DETAIL: + _log_rejection( + challenge_slug=slug, + reason=PAYLOAD_TOO_LARGE_DETAIL, + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + ) raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail=PAYLOAD_TOO_LARGE_DETAIL, ) from exc if message == UNSUPPORTED_MEDIA_DETAIL: + _log_rejection( + challenge_slug=slug, + reason=UNSUPPORTED_MEDIA_DETAIL, + status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, + ) raise HTTPException( status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, detail=UNSUPPORTED_MEDIA_DETAIL, ) from exc + _log_rejection( + challenge_slug=slug, + reason=message, + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + ) raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message, @@ -958,6 +1052,7 @@ async def push_raw_weights( "DEFAULT_MAX_CLOCK_SKEW_SECONDS", "DEFAULT_MAX_FUTURE_EPOCH_AHEAD", "DEFAULT_MAX_WEIGHT_KEYS", + "ZERO_WEIGHT_DETAIL", "PushOutcome", "RAW_WEIGHT_FRESHNESS_POLICY_VERSION", "RawWeightAuthError", diff --git a/tests/unit/test_raw_weight_ingress.py b/tests/unit/test_raw_weight_ingress.py index ea239606c..6bb48e718 100644 --- a/tests/unit/test_raw_weight_ingress.py +++ b/tests/unit/test_raw_weight_ingress.py @@ -709,3 +709,92 @@ async def test_slug_body_route_mismatch_forbidden(harness: dict[str, Any]) -> No headers = _signed_headers(path=path, body=body, token=TOKEN, slug=SLUG, clock=clock) r = await client.post(path, content=body, headers=headers) assert r.status_code == 403 + + +@pytest.mark.asyncio +async def test_rejection_details_are_attributable(harness: dict[str, Any]) -> None: + """Each rejection reason surfaces its own detail string.""" + + client = harness["client"] + clock = harness["clock"] + path = f"/internal/v1/challenges/{SLUG}/raw-weights" + + # Freshness + past = clock.now() - timedelta(hours=1) + body = _build_body( + clock=clock, + nonce="attr-fresh", + computed_at=past, + expires_at=past + timedelta(minutes=1), + ) + r = await client.post( + path, content=body, headers=_signed_headers(path=path, body=body, clock=clock) + ) + assert r.status_code == 422 + assert r.json()["detail"] == "snapshot outside freshness window" + + # Digest mismatch + body = _build_body(clock=clock, nonce="attr-digest", bad_digest="a" * 64) + r = await client.post( + path, content=body, headers=_signed_headers(path=path, body=body, clock=clock) + ) + assert r.status_code == 422 + assert r.json()["detail"] == "payload_digest mismatch" + + # Schema field failure includes field path (not bare generic only) + bad = json.loads(_build_body(clock=clock, nonce="attr-schema").decode()) + bad["weights"] = {"not-a-hotkey!!!": 1.0} + bad.pop("payload_digest", None) + bad["payload_digest"] = RawWeightPushRequest.compute_digest(bad) + raw = json.dumps(bad, sort_keys=True, separators=(",", ":")).encode() + r = await client.post( + path, content=raw, headers=_signed_headers(path=path, body=raw, clock=clock) + ) + assert r.status_code == 422 + detail = r.json()["detail"] + assert "invalid raw weight payload" in detail + assert "weights" in detail + + +@pytest.mark.asyncio +async def test_all_zero_weight_map_rejected(harness: dict[str, Any]) -> None: + """All-zero maps must not enter aggregation.""" + + client = harness["client"] + clock = harness["clock"] + session_factory = harness["session_factory"] + path = f"/internal/v1/challenges/{SLUG}/raw-weights" + # Validator-shaped hotkey with weight 0.0 only — must fail closed. + body = _build_body( + clock=clock, + nonce="zero-1", + weights={"5GziQCcRpN8NCJktX343brnfuVe3w6gUYieeStXPD1Dag2At": 0.0}, + ) + r = await client.post( + path, content=body, headers=_signed_headers(path=path, body=body, clock=clock) + ) + assert r.status_code == 422 + assert "all-zero" in r.json()["detail"] + assert await _count_snapshots(session_factory) == 0 + + +@pytest.mark.asyncio +async def test_freshness_detail_distinct_from_schema( + harness: dict[str, Any], +) -> None: + client = harness["client"] + clock = harness["clock"] + path = f"/internal/v1/challenges/{SLUG}/raw-weights" + past = clock.now() - timedelta(hours=2) + body = _build_body( + clock=clock, + nonce="fresh-detail", + computed_at=past, + expires_at=past + timedelta(seconds=30), + ) + r = await client.post( + path, content=body, headers=_signed_headers(path=path, body=body, clock=clock) + ) + assert r.status_code == 422 + assert r.json()["detail"] == "snapshot outside freshness window" + assert r.json()["detail"] != "invalid raw weight payload"