feat(challenge): serve a joined miner its own admission-trial job (S2)

docs/network-v2.md

@@ -143,6 +143,23 @@ uv run --frozen hypertrain-miner probe-v2 --config miner.toml \
 uv run --frozen hypertrain-miner proof-v2 --config miner.toml --proof proof.json
 ```
 
+A joined miner can skip the export: `probe-v2 --config miner.toml > proof.json`
+without `--job`/`--challenge` reads its admission ID (`status-v2`), GETs the
+challenge (which issues the trial when none is open), then
+`GET /v2/runs/{run_id}/join/{admission_id}/job` with headers `X-Hotkey` and
+`X-Trial-Signature` (hex hotkey signature of
+`hypertrain/trial-job/2|{run_id}|{admission_id}`). The route returns the admin
+export's JSON only to the currently registered hotkey that owns the admission:
+unknown run 404, unknown admission 409 `UNKNOWN_ADMISSION`, a deregistered or
+revoked owner 403 like the challenge route, any other hotkey or a missing or bad
+signature 403 (the admin token is no substitute), then the export's own refusals
+(409 `NO_CHALLENGE`, 409 `REFERENCE_TRIAL_CLOSED` once the trial is finalized,
+503 on a beacon pause). The miner checks the job, objects and signed challenge as
+`trial_export` does, streams each input (one uncompressed, sha256-verified
+object; about 4 bytes per parameter for a `reset` start state, more under
+`carry`) into a private temporary folder under its workdir, probes, and deletes
+the folder, also on failure: a re-run downloads and probes again.
+
 `probe-v2` executes the screen and uploads original rank-0 state/EF/delta/leaves;
 stdout contains the signed proof/screen pair. The operator independently prepares
 and finalizes references through authenticated admin routes; miners cannot

src/hypertrain/challenge/app.py

@@ -453,6 +453,32 @@ async def challenge_v2(run_id: str, admission_id: str) -> Any:
         await subnet_member((await run(admission.store.by_id, admission_id)).hotkey)
         return await run(store.for_run(run_id).admission_v2, run_id, "challenge", b"", admission_id)
 
+    @app.get("/v2/runs/{run_id}/join/{admission_id}/job")
+    async def trial_job_v2(
+        run_id: str,
+        admission_id: str,
+        x_hotkey: Annotated[str | None, Header()] = None,
+        x_trial_signature: Annotated[str | None, Header()] = None,
+    ) -> Any:
+        from hypertrain.protocol.keys import verify
+
+        # The admin export's payload, only to the registered hotkey that owns the admission:
+        # admission ids are public (admission status), so the caller signs for its hotkey.
+        _, admission, _ = await run(store.for_run(run_id)._services, run_id)
+        hotkey = (await run(admission.store.by_id, admission_id)).hotkey
+        await subnet_member(hotkey)
+        try:
+            valid = x_hotkey == hotkey and verify(
+                decode_hotkey(hotkey),
+                f"hypertrain/trial-job/2|{run_id}|{admission_id}".encode(),
+                bytes.fromhex(x_trial_signature or ""),
+            )
+        except (ValueError, KeyError_):
+            valid = False
+        if not valid:
+            raise ChallengeError(403, "hotkey is not current accepted identity")
+        return await run(store.for_run(run_id).export_trial_v2, run_id, admission_id)
+
     @app.get("/v2/runs/{run_id}/admission/{hotkey}")
     async def admission_status_v2(run_id: str, hotkey: str) -> Any:
         return await run(store.for_run(run_id).admission_v2, run_id, "status", b"", hotkey)

src/hypertrain/miner/cli.py

@@ -385,11 +385,11 @@ def main(argv: list[str] | None = None) -> int:
             s.add_argument("--request-id", required=True)
             s.add_argument("--expiry", type=int, required=True)
         if name in ("probe-v2", "run-v2", "watch"):
-            s.add_argument("--job", type=Path, required=name == "probe-v2")
+            s.add_argument("--job", type=Path)
         if name == "run-v2":
             s.add_argument("--round", type=int)
         if name == "probe-v2":
-            s.add_argument("--challenge", type=Path, required=True)
+            s.add_argument("--challenge", type=Path)
         if name == "proof-v2":
             s.add_argument("--proof", type=Path, required=True)
         if name == "watch":
@@ -465,7 +465,12 @@ def main(argv: list[str] | None = None) -> int:
                     case "status-v2":
                         result = network.status()
                     case "probe-v2":
-                        result = network.probe(args.job, args.challenge)
+                        if (args.job is None) != (args.challenge is None):
+                            raise MinerError("probe-v2 takes --job and --challenge, or neither")
+                        if args.job is None:
+                            result = network.probe_trial()
+                        else:
+                            result = network.probe(args.job, args.challenge)
                     case "proof-v2":
                         result = network.proof(args.proof)
                     case "run-v2":

src/hypertrain/miner/core.py

@@ -890,6 +890,109 @@ def traced_launch(
             raise MinerError("probe deadline/cancellation")
         return result
 
+    def probe_trial(self) -> dict[str, Any]:
+        """probe-v2 without --job: this hotkey's open trial, fetched into a private temp folder."""
+        with tempfile.TemporaryDirectory(prefix="trial-", dir=self.directory) as folder:
+            job, challenge = self._fetch_trial(Path(folder).resolve())
+            return self.probe(job, challenge)
+
+    def _fetch_trial(self, root: Path) -> tuple[Path, Path]:
+        """The folder trial_export writes, from the miner's own signed job read.
+
+        The challenge GET issues the trial when none is open (else returns the stored one);
+        the job, its inputs and the signed challenge are checked as trial_export checks them."""
+        from pydantic import TypeAdapter
+
+        from hypertrain.protocol import envelope_v2
+        from hypertrain.protocol.messages_v2 import Hex64, IslandJobV1, JoinChallenge
+
+        admission_id = self.status()["record"]["admission_id"]
+        url = self.url + f"/join/{admission_id}"
+        envelope = self.api.call("GET", url + "/challenge")
+        signature = self.kp.sign(f"hypertrain/trial-job/2|{self.run_id}|{admission_id}".encode())
+        data = self.api.call(
+            "GET",
+            url + "/job",
+            headers={"X-Hotkey": self.kp.ss58, "X-Trial-Signature": signature.hex()},
+        )
+        try:
+            job = IslandJobV1.model_validate(data["job"])
+            challenge = JoinChallenge.model_validate(data["challenge"])
+            objects = TypeAdapter(dict[str, Hex64]).validate_python(data["objects"])
+            nonce = data["nonce"]
+            signed = envelope_v2.parse_envelope(envelope)
+        except (ValueError, KeyError, TypeError) as error:
+            raise MinerError("malformed trial job response") from error
+        if (
+            job.run_id != self.run_id
+            or challenge.manifest_hash != self.run_id
+            or challenge.admission_id != admission_id
+            or challenge.nonce != nonce
+            or set(objects) != set(job.object_paths)
+        ):
+            raise MinerError("trial job/challenge/object binding differs")
+        for name in ("start_state", "ef_in", "v0"):
+            if objects.get(name) != getattr(job, name + "_sha256"):
+                raise MinerError(f"trial {name} sha256 differs from job")
+        if (
+            signed.type != "JoinChallenge"
+            or signed.run_id != self.run_id
+            or signed.signer != self.manifest.training.coord_pubkey
+            or signed.body != challenge.body()
+            or not envelope_v2.verify_envelope(envelope)
+        ):
+            raise MinerError("signed challenge differs from the trial job")
+        destinations = {root / "job.json", root / "challenge.json"}
+        paths: dict[str, Path] = {}
+        for name, relative in job.object_paths.items():
+            target = (root / relative).resolve()
+            if (
+                target == root
+                or not target.is_relative_to(root)
+                or any(
+                    target == p or target in p.parents or p in target.parents for p in destinations
+                )
+            ):
+                raise MinerError(f"unsafe trial object path: {relative}")
+            destinations.add(target)
+            paths[name] = target
+        for name, digest in objects.items():
+            paths[name].parent.mkdir(parents=True, exist_ok=True)
+            self._get_whole(digest, paths[name])
+        (root / "job.json").write_bytes(canonicalize(job.body()))
+        (root / "challenge.json").write_bytes(canonicalize(envelope))
+        return root / "job.json", root / "challenge.json"
+
+    def _get_whole(self, digest: str, target: Path) -> None:
+        """Stream one whole plain object to `target` and verify it: the trial export stores each
+        input as a single object, so the 1 MiB part bound of _get_object does not apply; the
+        bound is the largest object a part index can describe."""
+        from hypertrain.protocol.messages_v2 import MAX_PARTS, PART_BYTES
+
+        def get() -> None:
+            hasher, size = hashlib.sha256(), 0
+            with self.api.c.stream(
+                "GET",
+                self.api.base + self.url + "/objects/" + digest,
+                headers={"Accept-Encoding": "identity"},
+            ) as r:
+                if not r.is_success:
+                    raise ChallengeHTTPError(r.status_code, "trial input object unavailable")
+                # As _get_object: httpx would decode gzip/br before the bound check.
+                if r.headers.get("Content-Encoding", "identity").lower() != "identity":
+                    raise MinerError("trial input object has a non-identity content encoding")
+                with target.open("wb") as stream:
+                    for chunk in r.iter_bytes(chunk_size=64 << 10):
+                        size += len(chunk)
+                        if size > MAX_PARTS * PART_BYTES:
+                            raise MinerError("trial input object exceeds the object size bound")
+                        hasher.update(chunk)
+                        stream.write(chunk)
+            if hasher.hexdigest() != digest:
+                raise MinerError(f"trial input {target.name} sha256 mismatch")
+
+        _retrying("GET object " + digest, get)
+
     def _bounded_put(
         self, deadline: float, cancel: threading.Event | None
     ) -> Callable[[bytes], str]:

tests/challenge/test_subnet_gate_v2.py

@@ -313,6 +313,36 @@ def test_deregistered_after_join_refused(gated, subnet_master):
     )
 
 
+def test_trial_job_refused_like_challenge_once_owner_deregistered(gated, subnet_master):
+    # Given: HOT[0] joined while registered and holds an open trial.
+    n = gated
+    subnet_master.registered.add(HOT[0].ss58)
+    identity = n.join(0).json()["admission_id"]
+    n.push(2)
+    assert n.client.get(n.url + "/join/" + identity + "/challenge").status_code == 200
+    message = f"hypertrain/trial-job/2|{n.manifest.run_id()}|{identity}".encode()
+    signed = {"x-hotkey": HOT[0].ss58, "x-trial-signature": HOT[0].sign(message).hex()}
+    url = n.url + "/join/" + identity + "/job"
+    # When: the hotkey leaves the subnet.
+    subnet_master.registered.clear()
+    refresh(n)
+    # Then: the owner's signature, the admin token, or both get the challenge route's refusal.
+    challenge = n.client.get(n.url + "/join/" + identity + "/challenge")
+    assert challenge.status_code == 403
+    assert challenge.json() == {"detail": "the hotkey is not registered on the subnet"}
+    for headers in (signed, service.admin(), {**signed, **service.admin()}):
+        response = n.client.get(url, headers=headers)
+        assert (response.status_code, response.json()) == (403, challenge.json()), headers
+    assert not (n.store.state_dir / "trials-v2").exists()
+    # And: registered again, the admin token still does not stand in for the signature.
+    subnet_master.registered.add(HOT[0].ss58)
+    refresh(n)
+    response = n.client.get(url, headers=service.admin())
+    assert response.status_code == 403
+    assert response.json() == {"detail": "hotkey is not current accepted identity"}
+    assert n.client.get(url, headers=signed).status_code == 200
+
+
 def test_weights_exclude_deregistered_hotkeys(tmp_path, secrets_dir, clock):
     # Given: two vested entitlements; one hotkey later deregistered.
     master = Master()

tests/challenge/test_trial_export.py

@@ -40,6 +40,16 @@ def export(n, identity):
     return n.client.post(n.url + f"/admin/trial/{identity}/export", headers=service.admin())
 
 
+def job(n, identity, key=service.HOT[0], hotkey=None, run_id=None):
+    """Miner-facing job read, signed by `key` (X-Hotkey defaults to the signer)."""
+    run_id = run_id or n.manifest.run_id()
+    signature = key.sign(f"hypertrain/trial-job/2|{run_id}|{identity}".encode()).hex()
+    return n.client.get(
+        f"/v2/runs/{run_id}/join/{identity}/job",
+        headers={"x-hotkey": hotkey or key.ss58, "x-trial-signature": signature},
+    )
+
+
 def test_export_matches_prepare_reference_inputs(trial):
     n, identity, envelope = trial
     response = export(n, identity)
@@ -137,6 +147,41 @@ def test_export_requires_admin(trial):
     assert not (n.store.state_dir / "trials-v2").exists()
 
 
+def test_owner_reads_the_admin_export_without_admin_token(trial):
+    n, identity, envelope = trial
+    response = job(n, identity)
+    assert response.status_code == 200, response.text
+    assert response.json() == export(n, identity).json()
+    assert response.json()["challenge"] == envelope["body"]
+
+
+def test_job_route_refuses_another_miners_admission(trial):
+    n, identity, _ = trial
+    other = service.HOT[1]  # registered on the fixture subnet, not this admission's hotkey
+    for response in (
+        job(n, identity, key=other),
+        job(n, identity, key=other, hotkey=service.HOT[0].ss58),
+        job(n, identity, hotkey=other.ss58),
+    ):
+        assert response.status_code == 403, response.text
+        assert response.json() == {"detail": "hotkey is not current accepted identity"}
+    assert not (n.store.state_dir / "trials-v2").exists()
+    assert job(n, identity).status_code == 200
+
+
+@pytest.mark.parametrize(
+    "unknown,status,detail", [("run", 404, "unknown run"), ("admission", 409, "UNKNOWN_ADMISSION")]
+)
+def test_job_route_unknown_ids_answer_like_challenge(trial, unknown, status, detail):
+    n, identity, _ = trial
+    run_id = "cd" * 32 if unknown == "run" else n.manifest.run_id()
+    identity = "ab" * 32 if unknown == "admission" else identity
+    challenge = n.client.get(f"/v2/runs/{run_id}/join/{identity}/challenge")
+    response = job(n, identity, run_id=run_id)
+    assert (response.status_code, response.json()) == (status, {"detail": detail})
+    assert (challenge.status_code, challenge.json()) == (status, {"detail": detail})
+
+
 def test_export_refuses_changed_job(trial):
     n, identity, envelope = trial
     directory = n.store.state_dir / "trials-v2" / envelope["body"]["nonce"]