validator: per arena scoring rules, envelope budgets and footprint, burned empty pools, router and harness limits, and the context compression grader

docs/miner_setup.md

@@ -14,6 +14,6 @@ Validators score your weights under the arena's own prompt, and provenance fails
 2. Harness: set the adapter prompt, the verifier and retries, measure again.
 3. Distil: frontier answers with reasoning on schemas and question shapes like the arena's, filtered by the grader, fine tuned from the base (LoRA or full) and recorded under `training/`.
 4. Reinforce: reward a query that returns the reference result on every database state; or preference pairs of a right teacher answer against a wrong draft.
-5. Check: hold out whole domains, rename and revalue as the twin rule does, measure latency on a 24 GB card, train the router on the result.
+5. Check: hold out whole domains, rename and revalue as the twin rule does, measure latency and memory inside your arena's envelope group (https://www.microtensor.cloud/docs/certified-inference/system/envelope-groups), train the router on the result.
 
 Items are fresh every epoch and twins deduct memorisation, so train the skill, not the sample databases. Full page: https://www.microtensor.cloud/docs/certified-inference/guides/training

docs/validator_setup.md

@@ -7,7 +7,7 @@ Netuid **576** on Bittensor testnet. One validator evaluates every submission of
 | | |
 |---|---|
 | OS | Ubuntu 22.04 or 24.04 |
-| GPU | 3 × 24 GB (RTX 4090, L4, A10G or better), CUDA 12 driver |
+| GPU | 3 × 24 GB (RTX 4090, L4, A10G or better), CUDA 12 driver. Arenas graded by a probe agent (context compression) take the last GPU in `--gpus` for it, so give them one more |
 | CPU | 16 cores |
 | RAM | 64 GB |
 | Disk | NVMe, 400 GB free. Models are cached per revision and pruned after each epoch |

microtensor_subnet/artifact/contract.py

@@ -4,7 +4,8 @@
 
 SCHEMA_VERSION: Final[int] = 1
 COMPONENTS: Final[tuple[str, ...]] = ("slm", "harness", "router", "escalation")
-HARDWARE_CLASSES: Final[tuple[str, ...]] = ("gpu_24gb",)
+HARDWARE_CLASSES: Final[tuple[str, ...]] = ("mt-1", "mt-2", "mt-3")
+DEFAULT_HARDWARE_CLASS: Final[str] = "mt-3"
 
 ARTIFACT_REPO_PREFIX: Final[str] = "microtensor_"
 SLM_REPO_PREFIX: Final[str] = "mt_slm_"
@@ -36,7 +37,8 @@
 SLM_METRICS_GLOB: Final[str] = "training/runs/*/metrics.jsonl"
 SLM_MIN_PARAMETERS: Final[int] = 1_000_000_000
 SLM_MAX_PARAMETERS: Final[int] = 8_000_000_000
-ROUTER_MAX_PARAMETERS: Final[int] = 100_000_000
+ROUTER_MAX_PARAMETERS: Final[int] = 1_500_000_000
+ROUTER_CPU_PARAMETERS: Final[int] = 100_000_000
 
 ENV_PORT: Final[str] = "MT_PORT"
 ENV_SLM_URL: Final[str] = "MT_SLM_URL"

microtensor_subnet/artifact/params.py

@@ -11,6 +11,7 @@
 QUANT_SUFFIXES = (".qweight", ".qzeros", ".scales", ".g_idx")
 INT4_PACK = 8
 INT8_PACK = 4
+KV_BYTES = 2
 SKIPPED_SUFFIXES = (".absmax", ".quant_map", ".nested_absmax", ".nested_quant_map", ".quant_state")
 
 
@@ -68,3 +69,29 @@ def count_parameters(files: Iterable[Path]) -> int:
 
 def safetensors_in(root: Path) -> list[Path]:
     return sorted(p for p in root.glob("*.safetensors") if p.is_file())
+
+
+def _text_config(root: Path) -> Mapping[str, Any]:
+    try:
+        config = json.loads((root / "config.json").read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        raise ParameterError(f"config.json in {root.name} cannot be read") from exc
+    if not isinstance(config, Mapping):
+        raise ParameterError(f"config.json in {root.name} is not an object")
+    nested = config.get("text_config")
+    return nested if isinstance(nested, Mapping) else config
+
+
+def kv_cache_bytes(root: Path, max_len: int) -> int:
+    config = _text_config(root)
+    layers = int(config.get("num_hidden_layers") or 0)
+    heads = int(config.get("num_attention_heads") or 0)
+    kv_heads = int(config.get("num_key_value_heads") or heads)
+    hidden = int(config.get("hidden_size") or 0)
+    head_dim = int(config.get("head_dim") or (hidden // heads if heads else 0))
+    return 2 * layers * kv_heads * head_dim * max_len * KV_BYTES
+
+
+def footprint_bytes(root: Path, max_len: int) -> int:
+    weights = sum(path.stat().st_size for path in safetensors_in(root))
+    return weights + kv_cache_bytes(root, max_len)

microtensor_subnet/chain/client.py

@@ -455,6 +455,20 @@ def commit_reveal_enabled(self) -> bool:
             return bool(result) and int(result) > 0
         return False
 
+    @_serialised
+    def owner_hotkey(self) -> str:
+        reader = getattr(self.subtensor, "get_subnet_owner_hotkey", None)
+        if callable(reader):
+            return str(with_retry("owner hotkey", lambda: reader(self._config.netuid)) or "")
+        substrate = getattr(self.subtensor, "substrate", None)
+        if substrate is None:
+            raise ChainError(UNADAPTED.format(what="owner hotkey reader"))
+        found = with_retry(
+            "owner hotkey",
+            lambda: substrate.query("SubtensorModule", "SubnetOwnerHotkey", [self._config.netuid]),
+        )
+        return str(getattr(found, "value", found) or "")
+
     @_serialised
     def set_weights(self, vector: WeightVector) -> tuple[bool, str]:
         if vector.is_empty:

microtensor_subnet/cli/submit.py

@@ -47,7 +47,11 @@ def register(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) ->
     manifest.add_argument("--slm", type=Path, help="slm repository root, when slm is submitted")
     manifest.add_argument("--slm-repo", default="", help="<user>/mt_slm_<arena>")
     manifest.add_argument("--slm-revision", default="", help="the slm repository's git commit sha")
-    manifest.add_argument("--hardware-class", default=contract.HARDWARE_CLASSES[0])
+    manifest.add_argument(
+        "--hardware-class",
+        default=contract.DEFAULT_HARDWARE_CLASS,
+        choices=contract.HARDWARE_CLASSES,
+    )
     add_common_arguments(manifest)
     manifest.set_defaults(handler=_manifest)
 

microtensor_subnet/scoring/pools.py

@@ -51,13 +51,17 @@ def combine(
     pools: Mapping[str, Sequence[Standing]],
     shares: Mapping[str, float] = POOL_SHARES,
     schedule: Sequence[float] = RANK_PAYOUT,
+    *,
+    burn_to: str = "",
 ) -> tuple[dict[str, float], dict[str, float]]:
     weights: dict[str, float] = {}
     burned: dict[str, float] = {}
     for pool, share in shares.items():
         paid = pool_payout(pools.get(pool, ()), schedule)
         if not paid:
             burned[pool] = share
+            if burn_to:
+                weights[burn_to] = weights.get(burn_to, 0.0) + share
             continue
         for hotkey, fraction in paid.items():
             weights[hotkey] = weights.get(hotkey, 0.0) + share * fraction

microtensor_subnet/validator/compress.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from typing import Any, Final
+
+GRADER: Final[str] = "context_compression"
+FIELDS: Final[dict[str, type]] = {
+    "summary": str,
+    "files_touched": list,
+    "open_tasks": list,
+    "errors": list,
+    "decisions": list,
+}
+EXTENSIONS: Final[str] = (
+    "py|pyi|js|ts|tsx|jsx|go|rs|java|rb|c|h|cc|cpp|hpp|toml|cfg|ini|yaml|yml|json|md|rst|txt|sh"
+)
+PATH = re.compile(rf"(?<![\w/.-])((?:[\w.-]+/)*[\w-]+\.(?:{EXTENSIONS}))(?![\w-])")
+ERROR = re.compile(r"\b([A-Za-z_][\w.]*(?:Error|Exception))\b")
+TICKED = re.compile(r"`([^`\s]{2,80})`")
+CALL = re.compile(r"\b([A-Za-z_]\w*)\(")
+SNAKE = re.compile(r"\b([a-z][a-z0-9]*_[a-z0-9_]+)\b")
+CAMEL = re.compile(r"\b([A-Z][a-z0-9]+[A-Z][A-Za-z0-9]*)\b")
+FENCE = re.compile(r"^```[A-Za-z]*\s*|\s*```$")
+BOUNDARY_BEFORE: Final[str] = r"(?<![\w.-])"
+LOOSE_BEFORE: Final[str] = r"(?<![\w-])"
+BOUNDARY_AFTER: Final[str] = r"(?![\w-])"
+MAX_ANSWER_WORDS: Final[int] = 8
+
+
+@dataclass(frozen=True, slots=True)
+class Judge:
+    count_tokens: Callable[[str], int]
+    answer: Callable[[str, Sequence[str]], list[str]]
+    pass_threshold: float = 0.8
+    min_output_tokens: int = 300
+    max_output_share: float = 0.1
+
+
+@dataclass(frozen=True, slots=True)
+class Verdict:
+    score: float
+    success: bool
+    reason: str
+
+
+def normalise(text: str) -> str:
+    cleaned = text.replace("\\", "/")
+    for mark in ("`", '"', "'"):
+        cleaned = cleaned.replace(mark, " ")
+    return " ".join(cleaned.lower().split())
+
+
+def mentioned(normalised_text: str, alias: str, *, loose: bool = False) -> bool:
+    before = LOOSE_BEFORE if loose else BOUNDARY_BEFORE
+    pattern = before + re.escape(normalise(alias)) + BOUNDARY_AFTER
+    return re.search(pattern, normalised_text) is not None
+
+
+def parse(output: Any) -> Mapping[str, Any] | None:
+    if isinstance(output, Mapping):
+        return output
+    text = FENCE.sub("", str(output if output is not None else "").strip())
+    try:
+        found = json.loads(text)
+    except ValueError:
+        return None
+    return found if isinstance(found, Mapping) else None
+
+
+def valid(found: Mapping[str, Any] | None) -> bool:
+    if found is None or set(found) != set(FIELDS):
+        return False
+    for key, kind in FIELDS.items():
+        value = found[key]
+        if not isinstance(value, kind):
+            return False
+        if isinstance(value, list) and not all(isinstance(entry, str) for entry in value):
+            return False
+    return True
+
+
+def values_text(found: Mapping[str, Any]) -> str:
+    parts: list[str] = []
+    for key in FIELDS:
+        value = found.get(key)
+        parts += value if isinstance(value, list) else [str(value or "")]
+    return "\n".join(str(p) for p in parts)
+
+
+def identifiers(text: str) -> set[str]:
+    found: set[str] = set()
+    for pattern in (PATH, ERROR, TICKED, CALL, SNAKE, CAMEL):
+        found.update(match for match in pattern.findall(text) if len(match) > 2)
+    return found
+
+
+def recall(text: str, facts: Sequence[Mapping[str, Any]]) -> float:
+    if not facts:
+        return 1.0
+    haystack = normalise(text)
+    kept = 0
+    for fact in facts:
+        if any(mentioned(haystack, alias) for alias in fact.get("aliases") or []):
+            kept += 1
+    return kept / len(facts)
+
+
+def hallucination(text: str, session: str) -> float:
+    found = identifiers(text)
+    if not found:
+        return 0.0
+    haystack = normalise(session)
+    invented = sum(1 for name in found if not mentioned(haystack, name, loose=True))
+    return invented / len(found)
+
+
+def continuation(
+    found: Mapping[str, Any], probes: Sequence[Mapping[str, Any]], judge: Judge
+) -> float:
+    if not probes:
+        return 1.0
+    summary = json.dumps(dict(found), ensure_ascii=False, indent=1)
+    answers = judge.answer(summary, [str(p.get("question", "")) for p in probes])
+    correct = 0
+    for probe, answer in zip(probes, answers, strict=False):
+        said = normalise(answer)
+        if len(said.split()) > MAX_ANSWER_WORDS:
+            continue
+        if any(mentioned(said, alias) for alias in probe.get("answers") or []):
+            correct += 1
+    return correct / len(probes)
+
+
+def grade(
+    output: Any, label: Mapping[str, Any], task: Mapping[str, Any], judge: Judge | None
+) -> Verdict:
+    if judge is None:
+        raise ValueError("context compression needs the probe agent and the tokenizer")
+    found = parse(output)
+    if not valid(found) or found is None:
+        return Verdict(0.0, False, "not a JSON object with the five fields")
+    raw = output if isinstance(output, str) else json.dumps(output, ensure_ascii=False)
+    session = str(task.get("user", ""))
+    input_tokens = int(label.get("input_tokens") or 0) or judge.count_tokens(
+        str(task.get("system", "")) + session
+    )
+    budget = max(judge.min_output_tokens, int(judge.max_output_share * input_tokens))
+    spent = judge.count_tokens(raw)
+    if spent > budget:
+        return Verdict(0.0, False, f"{spent} output tokens, over {budget}")
+    text = values_text(found)
+    kept = recall(text, label.get("facts") or [])
+    invented = hallucination(text, session)
+    carried = continuation(found, label.get("probes") or [], judge)
+    score = (1 - invented) * (0.5 * kept + 0.5 * carried)
+    reason = f"recall {kept:.2f}, invented {invented:.2f}, continuation {carried:.2f}"
+    return Verdict(round(score, 6), score >= judge.pass_threshold, reason)

microtensor_subnet/validator/context.py

@@ -10,6 +10,7 @@
 from microtensor_subnet.artifact.validate import Manifest
 from microtensor_subnet.chain.commitment import SubmissionCommitment
 from microtensor_subnet.core.items import Item
+from microtensor_subnet.validator import compress, grade
 
 SLM = "slm"
 HARNESS = "harness"
@@ -103,6 +104,7 @@ class EpochContext:
     sample_database: bytes = b""
     r_class: dict[str, float] = field(default_factory=dict)
     floors: tuple[float, ...] = (0.8, 0.9, 0.95, 0.99)
+    judge: compress.Judge | None = None
 
     @property
     def grader(self) -> str:
@@ -120,6 +122,42 @@ def max_tokens(self) -> int:
         found = block.get("max_tokens") if isinstance(block, Mapping) else None
         return int(found or SQL_MAX_TOKENS)
 
+    @property
+    def budgets(self) -> dict[str, Any]:
+        found = self.definition.get("budgets") or {}
+        return dict(found) if isinstance(found, Mapping) else {}
+
+    @property
+    def max_model_len(self) -> int | None:
+        budgets = self.budgets
+        if budgets.get("max_input_tokens") and budgets.get("max_output_tokens"):
+            return int(budgets["max_input_tokens"]) + int(budgets["max_output_tokens"])
+        return None
+
+    @property
+    def harness_limits(self) -> dict[str, Any]:
+        limits = self.definition.get("limits") or {}
+        found = limits.get("harness") if isinstance(limits, Mapping) else None
+        return dict(found) if isinstance(found, Mapping) else {}
+
+    @property
+    def grading(self) -> dict[str, Any]:
+        found = self.definition.get("grading") or {}
+        return dict(found) if isinstance(found, Mapping) else {}
+
+    def grade(self, item: Item, output: Any) -> grade.Grade:
+        try:
+            return grade.grade(
+                self.grader,
+                output,
+                item.label,
+                self.items.databases,
+                task=item.task,
+                judge=self.judge,
+            )
+        except grade.GradeError as exc:
+            return grade.Grade(0.0, False, str(exc))
+
     @property
     def slm_prompt(self) -> str:
         return str(