fix(challenge): answer a repeated relay receipt idempotently

docs/network-v2.md

@@ -192,7 +192,10 @@ uv run --frozen hypertrain-miner run-v2 --config miner.toml --round 0
 This follows accept, authenticated inputs, exact-N launcher, one island commit,
 original artifacts, regional custody, master acceptance and delta submission.
 Tokens derive from assigned samples times sequence length, not self-reported GPU
-count. `UPLOADED` is not audit success, settlement or payment. Unlike v1 `run`,
+count. Rerunning a round that is already `UPLOADED` exits 0 with `UPLOADED`: the
+master returns its first acceptance for the same receipt without a new relay
+retrieval, and answers 409 `RELAY_RECEIPT_CONFLICT` for a different receipt.
+`UPLOADED` is not audit success, settlement or payment. Unlike v1 `run`,
 `run-v2` has no `--rounds`/`--start` option. Preserve the work directory; do not
 claim the CLI is a complete unattended production scheduler.
 

scripts/mine_multi_round_v2.py

@@ -282,10 +282,11 @@ def count(kind: str, identity: str | None = None) -> int:
         "relay_receipts_unchanged": count("relay-receipt") == receipts,
     }
     if not (
-        probe["after_upload"]["delta_unchanged"]
+        probe["after_upload"]["rc"] == 0
+        and probe["after_upload"]["delta_unchanged"]
         and probe["after_upload"]["relay_receipts_unchanged"]
     ):
-        raise RuntimeError(f"rerun after UPLOADED changed the accepted delta: {probe}")
+        raise RuntimeError(f"rerun after UPLOADED not a clean no-op: {probe}")
 
     w = open_round(svc, n, ROUNDS + 1, settle(svc, n, w)).w
     for i, conf in enumerate(configs[1:], 1):
@@ -312,6 +313,8 @@ def count(kind: str, identity: str | None = None) -> int:
     probe["kill_mid_round"] = {
         "killed_after_job_staged": alive and bool(count("island-job", f"{w}:{hot}"))
     }
+    if not probe["kill_mid_round"]["killed_after_job_staged"]:
+        raise RuntimeError("kill probe did not interrupt a staged job")
     try:
         run_v2(stages, configs[0], w, 0)
         probe["kill_mid_round"]["resume"] = "UPLOADED"

src/hypertrain/challenge/store.py

@@ -9929,6 +9929,25 @@ def upload_grant_v2(
             self._put_record_v2(run_id, "chunk-manifest", grant.digest(), chunk.body())
             return signed
 
+    def _accepted_receipt_v2(
+        self, run_id: str, w: int, hotkey: str, receipt_hash: str
+    ) -> dict[str, Any] | None:
+        """Stored acceptance of receipt_hash for (w, hotkey); 409 if another was accepted."""
+        accepted: list[dict[str, Any]] = [
+            json.loads(stored)
+            for (stored,) in self._db.execute(
+                "SELECT data FROM records_v2 WHERE run_id=? AND kind='master-acceptance' "
+                "AND json_extract(data,'$.body.w')=? AND json_extract(data,'$.body.hotkey')=?",
+                (run_id, w, hotkey),
+            )
+        ]
+        for stored in accepted:
+            if stored["body"]["receipt_hash"] == receipt_hash:
+                return stored
+        if accepted:
+            raise ChallengeError(409, "RELAY_RECEIPT_CONFLICT")
+        return None
+
     async def relay_receipt_v2(self, raw: bytes, client: Any) -> dict[str, Any]:
         import base64
 
@@ -9959,6 +9978,12 @@ async def relay_receipt_v2(self, raw: bytes, client: Any) -> dict[str, Any]:
                 grant=grant,
                 received_before=deadline,
             )
+            # A rerun must not challenge the relay again: it refuses a second live
+            # retrieval of the same object, and the first acceptance is authoritative.
+            if (
+                stored := self._accepted_receipt_v2(run_id, grant.w, grant.hotkey, receipt.digest())
+            ) is not None:
+                return stored
             assignment = self._record_v2(run_id, "relay-assignment", f"{grant.w}:{grant.hotkey}")[
                 "body"
             ]
@@ -9989,18 +10014,30 @@ async def relay_receipt_v2(self, raw: bytes, client: Any) -> dict[str, Any]:
         capability = base64.b64encode(
             canonicalize(self._record_v2(run_id, "grant", grant.digest()))
         ).decode()
-        custody_response = await client.get(
-            spec.https_url + f"/v1/uploads/{grant.digest()}/acks",
-            headers={"X-Upload-Grant": capability},
-        )
-        custody_response.raise_for_status()
-        custody = custody_response.json()
-        relay = RelayClient(registry, client)
-        with spool() as payload:
-            response = await relay.retrieval(request_raw, payload, relay_public=env.signer)
-            payload.seek(0)
-            original = payload.read()
+        import httpx
+
+        try:
+            custody_response = await client.get(
+                spec.https_url + f"/v1/uploads/{grant.digest()}/acks",
+                headers={"X-Upload-Grant": capability},
+            )
+            custody_response.raise_for_status()
+            custody = custody_response.json()
+            relay = RelayClient(registry, client)
+            with spool() as payload:
+                response = await relay.retrieval(request_raw, payload, relay_public=env.signer)
+                payload.seek(0)
+                original = payload.read()
+        except httpx.HTTPStatusError as exc:
+            raise ChallengeError(
+                502, f"relay refused master retrieval: HTTP {exc.response.status_code}"
+            ) from exc
         with self._tx():
+            # A concurrent submission may have been accepted while this one awaited the relay.
+            if (
+                stored := self._accepted_receipt_v2(run_id, grant.w, grant.hotkey, receipt.digest())
+            ) is not None:
+                return stored
             current = self._fresh_v2(
                 manifest, f"relay-accept:{receipt.digest()}:{self._now(self._db)}"
             )

tests/challenge/test_capacity_outer_v2.py

@@ -936,6 +936,8 @@ def test_prepared_profile_opening_metadata_without_tensor_calls(
     tiny, tmp_path, monkeypatch, attack, metadata_genesis
 ):
     """Signed hermetic metadata/state oracle; opening never gains kernel approval."""
+    # Store imports the worker lazily; a first import under the trap would bind forbidden.
+    import hypertrain.auditor.worker  # noqa: F401
     from hypertrain.aggregator.core import OuterState
     from hypertrain.auditor.replay import AnchorCache
     from hypertrain.challenge.store import ChallengeError, ChallengeStore