validator: report each miner's evaluation stage to the server as it runs

microtensor_subnet/validator/client.py

@@ -14,6 +14,7 @@
 SIGNATURE_HEADER: Final[str] = "x-mt-signature"
 TIMEOUT_SECONDS: Final[float] = 60.0
 BUNDLE_TIMEOUT_SECONDS: Final[float] = 600.0
+PROGRESS_TIMEOUT_SECONDS: Final[float] = 5.0
 
 
 ARENA_NETWORKS: dict[str, str] = {"test": "testnet", "finney": "mainnet"}
@@ -46,7 +47,13 @@ def _headers(self, method: str, path: str, body: bytes) -> dict[str, str]:
         return headers
 
     def _request(
-        self, method: str, path: str, body: dict[str, Any] | None = None, *, signed: bool = False
+        self,
+        method: str,
+        path: str,
+        body: dict[str, Any] | None = None,
+        *,
+        signed: bool = False,
+        timeout: float | None = None,
     ) -> Any:
         raw = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() if body else b""
         headers = (
@@ -55,7 +62,9 @@ def _request(
             else {"accept": "application/json", "content-type": "application/json"}
         )
         try:
-            with httpx.Client(base_url=self.base_url.rstrip("/"), timeout=self.timeout) as http:
+            with httpx.Client(
+                base_url=self.base_url.rstrip("/"), timeout=timeout or self.timeout
+            ) as http:
                 response = http.request(method, path, content=raw or None, headers=headers)
         except httpx.HTTPError as exc:
             raise ServerError(f"{method} {path}: {exc}") from exc
@@ -133,3 +142,9 @@ def submit_artifact_result(self, body: dict[str, Any]) -> dict[str, Any]:
         if not isinstance(found, dict) or not found.get("accepted"):
             raise ServerError(f"POST /v2/reports/artifact was not accepted: {found!r}"[:300])
         return dict(found)
+
+    def report_progress(self, body: dict[str, Any]) -> dict[str, Any]:
+        found = self._request(
+            "POST", "/v2/reports/progress", body, signed=True, timeout=PROGRESS_TIMEOUT_SECONDS
+        )
+        return dict(found or {})

microtensor_subnet/validator/collect.py

@@ -178,6 +178,7 @@ def materialise(
     bounds: tuple[int, int] | None = None,
     formats: list[str] | None = None,
     hold: Callable[[Path], None] | None = None,
+    on_downloaded: Callable[[], None] | None = None,
 ) -> Artifact:
     commitment = candidate.commitment
     if hold is not None:
@@ -186,6 +187,8 @@ def materialise(
         artifact_dir = download(commitment.artifact_repo, commitment.artifact_sha, cache_dir)
     except Exception as exc:
         raise CollectError(f"artifact repository could not be fetched: {exc}") from exc
+    if on_downloaded is not None:
+        on_downloaded()
     try:
         manifest = load_manifest(artifact_dir)
     except ManifestError as exc:

microtensor_subnet/validator/context.py

@@ -110,6 +110,11 @@ class EpochContext:
     def grader(self) -> str:
         return str(self.definition.get("grader") or "sqlite_execution")
 
+    def pool_size(self, test: str) -> int:
+        counts = self.definition.get("item_counts")
+        found = counts.get(test) if isinstance(counts, Mapping) else None
+        return int(found or 0) or len(self.items.originals(test))
+
     @property
     def stop(self) -> tuple[str, ...]:
         block = self.definition.get("generation") or {}

microtensor_subnet/validator/epoch.py

@@ -19,13 +19,23 @@
 from microtensor_subnet.core.reports import PoolScore
 from microtensor_subnet.scoring import pools
 from microtensor_subnet.serving.probes import sets as probe_sets
-from microtensor_subnet.validator import collect, compress, items, provenance, report, score
+from microtensor_subnet.validator import (
+    collect,
+    compress,
+    items,
+    progress,
+    provenance,
+    report,
+    score,
+)
 from microtensor_subnet.validator.client import ServerClient, ServerError
 from microtensor_subnet.validator.config import EpochConfig
 from microtensor_subnet.validator.context import (
     ESCALATION,
+    HARNESS,
     ROUTER,
     SLM,
+    SYSTEM,
     Artifact,
     EpochContext,
     Reference,
@@ -228,6 +238,7 @@ class Runner:
     reference_router: RouterRun | None = None
     reference_escalation: Any = None
     lock: threading.Lock = field(default_factory=threading.Lock)
+    reporter: progress.Reporter = field(default_factory=progress.Reporter)
 
     def _memory(self, model_dir: Path, config: Mapping[str, Any]) -> int:
         slm = config.get("slm") or {}
@@ -390,6 +401,7 @@ def slm_phase(self, artifact: Artifact, slot: int, gpu: str) -> None:
         outcome = self.outcomes.setdefault(artifact.hotkey, ArtifactOutcome(hotkey=artifact.hotkey))
         if artifact.slm_dir is None:
             return
+        self.reporter.post(artifact.hotkey, SLM, 0, self.context.pool_size(SLM))
         try:
             sandbox = self._open(
                 self.context.reference.harness_dir,
@@ -405,6 +417,7 @@ def slm_phase(self, artifact: Artifact, slot: int, gpu: str) -> None:
                     SlmClient(sandbox.slm_url),
                     memory_bytes=self._memory(artifact.slm_dir, artifact.config),
                     parameters=artifact.slm_parameters,
+                    progress=self.reporter.tracker(artifact.hotkey, SLM),
                 )
                 outcome.slm.vram_peak_bytes = self._measured(sandbox, outcome.slm.vram_peak_bytes)
             finally:
@@ -438,7 +451,11 @@ def full_phase(self, artifact: Artifact, slot: int, gpu: str) -> None:
             try:
                 if "harness" in artifact.has:
                     outcome.harness = harness.run(
-                        self.context, self.host, sandbox, timeout=self.config.request_seconds
+                        self.context,
+                        self.host,
+                        sandbox,
+                        timeout=self.config.request_seconds,
+                        progress=self.reporter.tracker(artifact.hotkey, HARNESS),
                     )
                     self.host.restart_main(sandbox, artifact_dir)
                 if "escalation" in artifact.has and self.reference_run is not None:
@@ -454,6 +471,7 @@ def full_phase(self, artifact: Artifact, slot: int, gpu: str) -> None:
                         reference=self.reference_run,
                         timeout=self.config.request_seconds,
                         token_price=token_price,
+                        progress=self.reporter.tracker(artifact.hotkey, ESCALATION),
                     )
                 if self.reference_run is not None:
                     outcome.system = system.run(
@@ -464,6 +482,7 @@ def full_phase(self, artifact: Artifact, slot: int, gpu: str) -> None:
                         reference=self.reference_run,
                         timeout=self.config.request_seconds,
                         token_price=token_price,
+                        progress=self.reporter.tracker(artifact.hotkey, SYSTEM),
                     )
             finally:
                 outcome.usage = self.host.close(sandbox)
@@ -494,6 +513,7 @@ def router_run(self, record: MinerRecord, sources: Mapping[str, Any]) -> RouterR
             self.gpus.give(slot, gpu)
 
     def close(self) -> None:
+        self.reporter.close()
         self.host.services.stop()
         if self.agent is not None:
             self.agent.stop()
@@ -508,6 +528,7 @@ def set_up_baselines(
     *,
     wallet: Any,
     hotkey: str,
+    reporter: progress.Reporter | None = None,
 ) -> Runner:
     limits = Limits.from_definition(context.definition, context.max_model_len)
     placement: probe_judge.Placement | None = None
@@ -566,6 +587,7 @@ def set_up_baselines(
         base_dirs=base_dirs,
         known_public=provenance.public_checkpoint_hashes(config.home / "public_checkpoints.json"),
         reference_slm_dir=reference_dir,
+        reporter=reporter or progress.Reporter(),
         protected={
             context.reference.slm_dir,
             *([reference_dir] if reference_dir is not None else []),
@@ -734,6 +756,19 @@ def build_queue(
     return queue, failures
 
 
+def announce(
+    reporter: progress.Reporter,
+    context: EpochContext,
+    queue: Sequence[collect.Candidate],
+    failures: Mapping[str, str],
+) -> None:
+    for position, candidate in enumerate(queue):
+        reporter.post(candidate.hotkey, progress.QUEUED, position + 1, len(queue))
+    for hotkey, reason in failures.items():
+        if hotkey in context.roster:
+            reporter.post(hotkey, progress.REFUSED, note=reason)
+
+
 def miner_loop(runner: Runner, queue: Sequence[collect.Candidate]) -> None:
     workers = (
         1 if runner.limits.on_cpu else max(1, min(runner.config.concurrency, runner.slot_count))
@@ -759,6 +794,7 @@ def process_miner(runner: Runner, position: int, candidate: collect.Candidate) -
             record.drafts = outcome.slm.drafts() if outcome.slm is not None else None
             record.failure = outcome.failure
             if not outcome.failure:
+                runner.reporter.post(candidate.hotkey, progress.SCORING)
                 record.scores = score_miner(runner, artifact, outcome)
                 record.provenance = dict(outcome.provenance)
     except Exception as exc:
@@ -781,6 +817,7 @@ def hold(path: Path) -> None:
             runner.holding[candidate.hotkey] = (*runner.holding.get(candidate.hotkey, ()), path)
 
     try:
+        runner.reporter.post(candidate.hotkey, progress.FETCHING)
         artifact = collect.materialise(
             candidate,
             cache_dir=runner.config.cache_dir,
@@ -790,6 +827,7 @@ def hold(path: Path) -> None:
             bounds=_bounds(definition),
             formats=[str(f) for f in definition.get("allowed_formats") or []],
             hold=hold,
+            on_downloaded=lambda: runner.reporter.post(candidate.hotkey, progress.CHECKING),
         )
         with runner.lock:
             runner.holding[candidate.hotkey] = _paths_of(artifact)
@@ -955,6 +993,8 @@ def post_result(runner: Runner, record: MinerRecord) -> bool:
         log.warning("%s result not accepted, kept for the retry at close: %s", record.hotkey, exc)
         return False
     record.posted = True
+    stage = progress.REFUSED if record.failure else progress.SENT
+    runner.reporter.post(record.hotkey, stage, note=record.failure)
     return True
 
 
@@ -1107,12 +1147,26 @@ def run_epoch(
     hotkey = hotkey_address(wallet)
     client = ServerClient(config.server_url, hotkey, wallet)
     context = load_context(config, client, chain, epoch_index=epoch_index, hotkey=hotkey)
-    runner = set_up_baselines(
-        config, client, context, docker or Docker(), wallet=wallet, hotkey=hotkey
+    reporter = progress.Reporter(
+        client, config.arena_id, context.epoch_index, enabled=not config.dry_run
     )
     try:
         queue, failures = build_queue(config, chain, context)
         log.info("epoch %d: %d queued, %d refused", epoch_index, len(queue), len(failures))
+        announce(reporter, context, queue, failures)
+        runner = set_up_baselines(
+            config,
+            client,
+            context,
+            docker or Docker(),
+            wallet=wallet,
+            hotkey=hotkey,
+            reporter=reporter,
+        )
+    except BaseException:
+        reporter.close()
+        raise
+    try:
         miner_loop(runner, queue)
         router_pass(runner)
         retry_unposted(runner)

microtensor_subnet/validator/pools/escalation.py

@@ -7,6 +7,7 @@
 from microtensor_subnet.artifact import contract
 from microtensor_subnet.core.items import Item
 from microtensor_subnet.validator.context import ESCALATION, EpochContext
+from microtensor_subnet.validator.progress import Progress, Tally
 from microtensor_subnet.validator.relay import REPAIR, RelayError, TicketDesk
 from microtensor_subnet.validator.results import EscalationRun, ItemRun, ReferenceRun
 from microtensor_subnet.validator.sandbox import Sandbox, call_main
@@ -39,10 +40,12 @@ def run(
     reference: ReferenceRun,
     timeout: float,
     token_price: float,
+    progress: Progress | None = None,
 ) -> EscalationRun:
     repairs: list[ItemRun] = []
     session = sandbox.session
-    for item, draft in drafts:
+    tally = Tally(progress, context.pool_size(ESCALATION), len(drafts))
+    for item, draft in tally.over(drafts):
         redo = reference.redo_answers.get(item.item_id)
         redo_cost = redo.cost_micros if redo is not None else 0
         cap = max(1, int(redo_cost * TICKET_MULTIPLE))

microtensor_subnet/validator/pools/harness.py

@@ -1,14 +1,15 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Sequence
+from collections.abc import Iterable, Sequence
 from pathlib import Path
 from typing import Any
 
 from microtensor_subnet.artifact import contract
 from microtensor_subnet.core.items import Item
 from microtensor_subnet.scoring.stats import percentile
 from microtensor_subnet.validator.context import HARNESS, EpochContext, render_prompt
+from microtensor_subnet.validator.progress import Progress, Tally
 from microtensor_subnet.validator.results import Condition, HarnessRun, ItemRun
 from microtensor_subnet.validator.sandbox import Sandbox, SandboxHost, call_main
 from microtensor_subnet.validator.sandbox.slm import SlmClient, SlmError
@@ -47,7 +48,7 @@ def run_condition(
     context: EpochContext,
     sandbox: Sandbox,
     name: str,
-    items: Sequence[Item],
+    items: Iterable[Item],
     *,
     timeout: float,
 ) -> Condition:
@@ -76,7 +77,7 @@ def run_condition(
 def run_bare(
     context: EpochContext,
     client: SlmClient,
-    items: Sequence[Item],
+    items: Iterable[Item],
     *,
     max_tokens: int | None = None,
 ) -> Condition:
@@ -102,20 +103,25 @@ def run(
     *,
     items: Sequence[Item] | None = None,
     timeout: float,
+    progress: Progress | None = None,
 ) -> HarnessRun:
     chosen = list(items) if items is not None else context.items.items(HARNESS)
-    miner = run_condition(context, sandbox, "H", chosen, timeout=timeout)
-    baselines = [run_bare(context, SlmClient(sandbox.slm_url), chosen)]
+    variants: list[Path] = [context.reference.harness_dir, *context.reference.variant_dirs]
+    tally = Tally(progress, context.pool_size(HARNESS), len(chosen) * (len(variants) + 2))
+    miner = run_condition(context, sandbox, "H", tally.over(chosen), timeout=timeout)
+    baselines = [run_bare(context, SlmClient(sandbox.slm_url), tally.over(chosen))]
     limit = float(context.harness_limits.get("added_wall_p95_ms") or 0)
     added = percentile(miner.added_ms, 0.95)
     limits_ok = not limit or added <= limit
     reason = "" if limits_ok else f"added wall time p95 {added:.0f} ms is over {limit:.0f} ms"
-    variants: list[Path] = [context.reference.harness_dir, *context.reference.variant_dirs]
     for index, variant in enumerate(variants, start=1):
         try:
             host.restart_main(sandbox, variant)
         except Exception as exc:
             log.warning("reference harness %d could not start: %s", index, exc)
+            tally.advance(len(chosen))
             continue
-        baselines.append(run_condition(context, sandbox, f"B{index}", chosen, timeout=timeout))
+        baselines.append(
+            run_condition(context, sandbox, f"B{index}", tally.over(chosen), timeout=timeout)
+        )
     return HarnessRun(miner=miner, baselines=baselines, limits_ok=limits_ok, limits_reason=reason)

microtensor_subnet/validator/pools/slm.py

@@ -6,6 +6,7 @@
 from microtensor_subnet.core.items import Item
 from microtensor_subnet.scoring.stats import percentile
 from microtensor_subnet.validator.context import SLM, EpochContext, render_prompt
+from microtensor_subnet.validator.progress import Progress, Tally
 from microtensor_subnet.validator.results import ItemRun, SlmRun
 from microtensor_subnet.validator.sandbox.slm import SlmClient, SlmError
 
@@ -42,13 +43,15 @@ def run(
     parameters: int,
     max_tokens: int | None = None,
     items: Sequence[Item] | None = None,
+    progress: Progress | None = None,
 ) -> SlmRun:
     budget = max_tokens or context.max_tokens
     chosen = list(items) if items is not None else context.items.originals(SLM)
     twins = context.items.twins_of(SLM)
     runs: list[ItemRun] = []
     twin_runs: list[ItemRun] = []
-    for item in chosen:
+    tally = Tally(progress, context.pool_size(SLM), len(chosen))
+    for item in tally.over(chosen):
         runs.append(_run_item(context, client, item, budget))
         twin = twins.get(item.item_id)
         if twin is not None:

microtensor_subnet/validator/pools/system.py

@@ -6,6 +6,7 @@
 from microtensor_subnet.artifact import contract
 from microtensor_subnet.core.items import Item
 from microtensor_subnet.validator.context import SYSTEM, EpochContext
+from microtensor_subnet.validator.progress import Progress, Tally
 from microtensor_subnet.validator.relay import SYSTEM as SYSTEM_PURPOSE
 from microtensor_subnet.validator.relay import RelayError, TicketDesk
 from microtensor_subnet.validator.results import FloorRun, ItemRun, ReferenceRun, SystemRun
@@ -25,13 +26,15 @@ def run(
     timeout: float,
     token_price: float,
     items: Sequence[Item] | None = None,
+    progress: Progress | None = None,
 ) -> SystemRun:
     chosen = list(items) if items is not None else context.items.items(SYSTEM)
     session = sandbox.session
     floors: list[FloorRun] = []
+    tally = Tally(progress, context.pool_size(SYSTEM), len(context.floors) * len(chosen))
     for floor in context.floors:
         runs: list[ItemRun] = []
-        for item in chosen:
+        for item in tally.over(chosen):
             redo = reference.system_answers.get(item.item_id)
             cap = max(1, int((redo.cost_micros if redo else 0) * TICKET_MULTIPLE))
             try: