Skip to content

Commit bf8bf72

Browse files
committed
Push unit-sum incentive maps to mock-master raw-weights.
Ensure egress body sum-normalizes when mass>0, monochronic epoch/revision and payload_digest stay on the normalized payload (VAL-WGT-015).
1 parent 48077aa commit bf8bf72

2 files changed

Lines changed: 166 additions & 9 deletions

File tree

src/hypercluster/weight_push.py

Lines changed: 32 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,14 @@
11
"""Raw-weight push: monochronic snapshots, digest, mock-master ack.
22
3-
Fulfills VAL-SCORE-013, 014, 015, 017, 023, 024, 030.
3+
Fulfills VAL-SCORE-013, 014, 015, 017, 023, 024, 030 and VAL-WGT-015.
44
55
Challenge builds ``RawWeightPushRequest`` (protocol 1.x), stores
66
``weight_snapshots`` with UNIQUE(epoch, revision), POSTs to Base master
77
``POST /internal/v1/challenges/{slug}/raw-weights`` (mock-master :3201 in
8-
mission). Never calls on-chain ``set_weights``. Push loop is cooperative
9-
async and MUST NOT block ``/health``.
8+
mission). Posted ``weights`` are the **sum-normalized** incentive map
9+
(sum ≈ 1.0 when non-empty); ``raw_mass_json`` retains pre-normalize mass.
10+
Never calls on-chain ``set_weights`` and never product-Verda egress. Push
11+
loop is cooperative async and MUST NOT block ``/health``.
1012
"""
1113

1214
from __future__ import annotations
@@ -37,7 +39,10 @@
3739
compute_raw_weights,
3840
sanitize_weights_map,
3941
)
40-
from hypercluster.domain.incentive import finalize_incentives_with_settings
42+
from hypercluster.domain.incentive import (
43+
finalize_incentives,
44+
finalize_incentives_with_settings,
45+
)
4146
from hypercluster.no_verda import (
4247
VerdaForbiddenError,
4348
assert_challenge_outbound_allowed,
@@ -191,9 +196,15 @@ def build_raw_weight_push_body(
191196
computed_at: datetime,
192197
expires_at: datetime,
193198
protocol_version: str = PROTOCOL_VERSION,
199+
already_normalized: bool = False,
194200
) -> tuple[RawWeightPushRequest, bytes]:
195201
"""Build digest-bound Base RawWeightPushRequest bytes.
196202
203+
Egress maps are sum-normalized incentives (VAL-WGT-015): when mass > 0 the
204+
posted ``weights`` sum ≈ 1.0; empty remains burn-safe (no push). Callers
205+
that already finalized a unit-sum map may pass ``already_normalized=True``
206+
to skip a second normalize (idempotent: unit maps stay unit).
207+
197208
Raises WeightPushValidationError on inverted/expired window or empty map.
198209
"""
199210

@@ -204,6 +215,17 @@ def build_raw_weight_push_body(
204215
"empty_weights",
205216
"empty weight map has no push surface",
206217
)
218+
# VAL-WGT-015: mock-master / master raw-weights body carries unit-sum
219+
# incentives, never absolute composite mass alone.
220+
if already_normalized:
221+
emission = cleaned
222+
else:
223+
emission = filter_ss58_weights(finalize_incentives(cleaned, sum_normalize=True))
224+
if not emission:
225+
raise WeightPushValidationError(
226+
"empty_weights",
227+
"empty weight map has no push surface",
228+
)
207229
# Ensure finite ≥0 (already sanitized) and Base schema alphabet keys.
208230
body: dict[str, Any] = {
209231
"protocol_version": protocol_version,
@@ -213,7 +235,7 @@ def build_raw_weight_push_body(
213235
"computed_at": as_utc(computed_at).replace(microsecond=0),
214236
"expires_at": as_utc(expires_at).replace(microsecond=0),
215237
"nonce": str(nonce),
216-
"weights": {str(k): float(v) for k, v in cleaned.items()},
238+
"weights": {str(k): float(v) for k, v in emission.items()},
217239
}
218240
# Digests exclude payload_digest field.
219241
digest_src = {
@@ -224,7 +246,7 @@ def build_raw_weight_push_body(
224246
"computed_at": isoformat_utc(as_utc(computed_at).replace(microsecond=0)),
225247
"expires_at": isoformat_utc(as_utc(expires_at).replace(microsecond=0)),
226248
"nonce": str(nonce),
227-
"weights": {str(k): float(v) for k, v in cleaned.items()},
249+
"weights": {str(k): float(v) for k, v in emission.items()},
228250
}
229251
digest = RawWeightPushRequest.compute_digest(digest_src)
230252
body["payload_digest"] = digest
@@ -422,6 +444,9 @@ async def create_pending_snapshot(
422444

423445
local_id = str(uuid.uuid4())
424446
n = nonce or f"hyper-{uuid.uuid4().hex}"
447+
# weight_map is already unit-sum from finalize_incentives_with_settings
448+
# (or live compute_raw_weights); flag already_normalized to keep digest
449+
# stable and avoid a redundant second pass (still idempotent if reused).
425450
payload, raw = build_raw_weight_push_body(
426451
challenge_slug=challenge_slug,
427452
epoch=resolved_epoch,
@@ -430,6 +455,7 @@ async def create_pending_snapshot(
430455
nonce=n,
431456
computed_at=computed,
432457
expires_at=expires,
458+
already_normalized=True,
433459
)
434460
row = _snapshot_from_payload(
435461
local_id=local_id,

tests/domain/test_incentive_normalize.py

Lines changed: 134 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
1-
"""Incentive clamp + sum-normalize + snapshot raw mass (VAL-WGT-010..014, 022).
1+
"""Incentive clamp + sum-normalize + snapshot raw mass + push (VAL-WGT-010..015, 022).
22
33
M10 default: finite ≥0 clamp; optional top-k / max-fraction; sum≈1 when mass>0;
4-
empty → {}; weight_snapshots store normalized map + retain raw mass.
4+
empty → {}; weight_snapshots store normalized map + retain raw mass; mock-master
5+
push payload is the unit-sum incentive map (VAL-WGT-015).
56
"""
67

78
from __future__ import annotations
@@ -12,7 +13,9 @@
1213
from pathlib import Path
1314
from typing import Any
1415

16+
import httpx
1517
import pytest
18+
from base.challenge_sdk.schemas import RawWeightPushRequest
1619
from httpx import ASGITransport, AsyncClient
1720

1821
from hypercluster.db.models import Job, JobAttempt
@@ -27,7 +30,15 @@
2730
)
2831
from hypercluster.domain.scoring_tee import persist_score_for_attempt
2932
from hypercluster.settings import HyperSettings
30-
from hypercluster.weight_push import create_pending_snapshot
33+
from hypercluster.sim.mock_master import app as mock_master_app
34+
from hypercluster.sim.mock_master import configure_token, reset_store
35+
from hypercluster.weight_push import (
36+
WeightPushClient,
37+
build_raw_weight_push_body,
38+
compute_payload_digest_for_body,
39+
create_pending_snapshot,
40+
get_snapshot_by_epoch_revision,
41+
)
3142
from hypercluster.weights import get_weights, weight_preview_payload
3243

3344
# Base-like ss58 keys for snapshot / get_weights integration.
@@ -395,3 +406,123 @@ async def test_poisoned_composites_never_yield_nan_weights(
395406
out2 = finalize_incentives(mixed, sum_normalize=True)
396407
assert weight_sum(out2) == pytest.approx(1.0, abs=UNIT_SUM_TOLERANCE)
397408
assert all(math.isfinite(v) and v >= 0.0 for v in out2.values())
409+
410+
411+
# ----- push payload: VAL-WGT-015 ---------------------------------------------
412+
413+
414+
def test_build_push_body_normalizes_absolute_mass() -> None:
415+
"""VAL-WGT-015: raw-weights builder coerces absolute mass → unit-sum on egress."""
416+
417+
from datetime import UTC, datetime, timedelta
418+
419+
now = datetime.now(UTC).replace(microsecond=0)
420+
expires = now + timedelta(seconds=300)
421+
absolute = {HOTKEY_A: 15.0, HOTKEY_B: 5.0}
422+
payload, raw = build_raw_weight_push_body(
423+
challenge_slug=SLUG,
424+
epoch=77,
425+
revision=1,
426+
weights=absolute,
427+
nonce="n-unit-sum-push",
428+
computed_at=now,
429+
expires_at=expires,
430+
)
431+
body = json.loads(raw)
432+
wmap = {str(k): float(v) for k, v in body["weights"].items()}
433+
assert weight_sum(wmap) == pytest.approx(1.0, abs=UNIT_SUM_TOLERANCE)
434+
assert wmap[HOTKEY_A] == pytest.approx(0.75)
435+
assert wmap[HOTKEY_B] == pytest.approx(0.25)
436+
# Digest matches independent recompute over the normalized body.
437+
expected = compute_payload_digest_for_body(
438+
{k: v for k, v in body.items() if k != "payload_digest"}
439+
)
440+
assert payload.payload_digest == expected
441+
# Absolute mass must not appear as total\approx20 on the wire.
442+
assert weight_sum(wmap) != pytest.approx(20.0)
443+
444+
445+
@pytest.mark.asyncio
446+
async def test_push_to_mock_master_uses_normalized_weights(
447+
settings_factory: Any, tmp_path: Path
448+
) -> None:
449+
"""VAL-WGT-015: POST raw-weights body is unit-sum; digest matches; acked."""
450+
451+
from hypercluster.app import create_app
452+
453+
reset_store()
454+
configure_token(TOKEN)
455+
456+
settings = settings_factory(
457+
database_url=f"sqlite+aiosqlite:///{tmp_path / 'inc-push-norm.sqlite3'}",
458+
shared_token=TOKEN,
459+
shared_token_file=None,
460+
)
461+
hyper = _hyper()
462+
app = create_app(settings, hyper_settings=hyper)
463+
464+
transport = httpx.ASGITransport(app=mock_master_app)
465+
async with httpx.AsyncClient(
466+
transport=transport, base_url="http://mock-master.test"
467+
) as master_http:
468+
async with app.router.lifespan_context(app):
469+
db = app.state.database
470+
async with db.session() as session:
471+
# Absolute mass 12 + 4 = 16; unit shares must be 0.75 / 0.25.
472+
await _seed_score(session, hotkey=HOTKEY_A, efficiency=12.0, hyper=hyper)
473+
await _seed_score(session, hotkey=HOTKEY_B, efficiency=4.0, hyper=hyper)
474+
await session.commit()
475+
476+
client = WeightPushClient(
477+
database=db,
478+
challenge_slug=SLUG,
479+
master_base_url="http://mock-master.test",
480+
shared_token=TOKEN,
481+
hyper=hyper,
482+
http_client=master_http,
483+
)
484+
result = await client.push_once(epoch=515)
485+
assert result.status == "acknowledged", result.error
486+
assert result.push_status == "acked"
487+
assert result.payload_digest
488+
assert len(result.payload_digest) == 64
489+
490+
# Capture from mock-master debug store (wire payload weights).
491+
listed = await master_http.get(f"/internal/v1/challenges/{SLUG}/raw-weights")
492+
assert listed.status_code == 200, listed.text
493+
items = listed.json()["items"]
494+
assert items, "mock-master must store accepted push"
495+
pushed = items[-1]
496+
wire_weights = {str(k): float(v) for k, v in pushed["weights"].items()}
497+
assert weight_sum(wire_weights) == pytest.approx(1.0, abs=UNIT_SUM_TOLERANCE)
498+
assert wire_weights[HOTKEY_A] == pytest.approx(0.75)
499+
assert wire_weights[HOTKEY_B] == pytest.approx(0.25)
500+
# Absolute mass must not leak as the egress map sum.
501+
assert weight_sum(wire_weights) != pytest.approx(16.0)
502+
assert pushed["payload_digest"] == result.payload_digest
503+
assert int(pushed["epoch"]) == 515
504+
505+
# Local snapshot: monochronic, unit-sum weights, raw mass retained.
506+
async with db.session() as session:
507+
row = await get_snapshot_by_epoch_revision(
508+
session, epoch=515, revision=result.revision
509+
)
510+
assert row is not None
511+
assert row.push_status == "acked"
512+
assert row.payload_digest == result.payload_digest
513+
assert weight_sum(row.weights_map()) == pytest.approx(
514+
1.0, abs=UNIT_SUM_TOLERANCE
515+
)
516+
raw_mass = row.raw_mass_map()
517+
assert raw_mass[HOTKEY_A] == pytest.approx(12.0)
518+
assert raw_mass[HOTKEY_B] == pytest.approx(4.0)
519+
# Canonical submit bytes recompute to same digest (VAL-WGT-015).
520+
if row.canonical_payload:
521+
reparsed = RawWeightPushRequest.model_validate_json(row.canonical_payload)
522+
assert reparsed.payload_digest == result.payload_digest
523+
assert weight_sum(reparsed.weights) == pytest.approx(
524+
1.0, abs=UNIT_SUM_TOLERANCE
525+
)
526+
527+
# Never challenge set_weights — push client has no chain setter.
528+
assert not hasattr(client, "set_weights")

0 commit comments

Comments
 (0)