ctx_compress artifact slm-only

.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+*.safetensors
+*.bin
+*.sqlite
+.env

LICENSE

@@ -0,0 +1,8 @@
+Copyright (c) 2026 Microtensor. All rights reserved.
+
+This template and every derivative of it are licensed to you solely for
+building and submitting systems to Microtensor arenas under the Microtensor
+submission terms (https://www.microtensor.cloud/terms). No other use,
+redistribution or sublicensing is permitted without written permission from
+Microtensor. Models submitted to an arena are licensed to Microtensor under
+the submission terms and are not published under a free licence.

README.md

@@ -0,0 +1,115 @@
+# Microtensor miner template
+
+A working submission for any Microtensor arena, in the two repository layout validators expect. Fork it, pick the adapter for your arena's task, replace the model, improve the harness, train the router, and ship.
+
+## What a submission is
+
+| Component | Lives in | What happens to it |
+|---|---|---|
+| Specialised model (SLM) | your `mt_slm_<arena>` repository | served by the official engine image on your GPU during the window; validators spot check its calls on the same hardware profile |
+| Harness | `harness/` here | `main.py` runs in the official sandbox on your GPU with no network except the recording gateway; validators replay it on CPU against the transcript |
+| Router | `router/` here | loaded by validators from the mirror as a one logit sequence classifier for the router pass |
+| Frontier repair | `escalation/` here | runs in the sandbox with a ticket that pays for frontier calls through the relay; the signed receipts are checked on replay |
+
+You may submit any subset; `manifest.json` names which. Validators fill the rest with the arena's reference components and pay you from the pools you compete in.
+
+You declare a hardware profile, run the official images on your own GPU during the evaluation window and upload a transcript; validators replay the harness on CPU and spot check the model calls on the same profile. The docs: https://www.microtensor.cloud/docs/certified-inference/system/decentralised-evaluation
+
+## Repository layout
+
+```
+<user>/microtensor_<arena>        this repository
+  manifest.json                   written by `mt submit manifest`
+  config.json                     runtime limits and the adapter name
+  main.py                         serves /harness/run, /escalation/repair, /system/run
+  requirements.txt                pinned packages from the allowed list only
+  harness/                        engine.py, verifier.py, adapters/, prompts/<adapter>/
+  escalation/                     patcher.py, prompts/<adapter>/
+  router/                         config.json, model.safetensors, tokenizer files, training/
+
+<user>/mt_slm_<arena>             the slm_repo/ folder, pushed as its own repository
+  README.md LICENSE config.json generation_config.json model.safetensors tokenizer.json tokenizer_config.json
+  training/                       recipe, data sources, distillation, compression, runs, scripts
+```
+
+## Adapters
+
+An arena's items carry a task document (`task`) whose fields the arena definition describes in `schemas.input`. The harness and the repair prompt are written once, against an **adapter** that knows how to prompt for that task and how to check an answer before it is returned:
+
+| Adapter | For arenas whose task is | Verification |
+|---|---|---|
+| `generic` | any structured or free text task | non empty, length bound, refuses injected instructions |
+| `sql_generation` | a question plus a schema, answered with one SQLite query | parses with sqlglot, names only known tables, executes on the arena's sample database |
+| `context_compression` | a coding agent's session to compact into the five field JSON (`ctx_compress`) | one JSON object; folds extra keys into `summary` and object entries into plain strings; checks the token budget the instruction states |
+
+`config.json` names the adapter. Write your own under `harness/adapters/` for an arena with a different task: subclass `Adapter`, override `verify` (and `clean`, `query` or `render` when the task needs it), register it in `harness/adapters/__init__.py`, and add `harness/prompts/<name>/task.txt`, `retry.txt`, `escalation/prompts/<name>/system.txt`, `repair.txt`. Prompt templates may use any task field by name, `{fields}` for every field as `key: value` lines, `{task_json}`, and `{draft}`, `{errors}`, `{trace}` on retries and repairs.
+
+For `ctx_compress`, set `config.json` to `"adapter": "context_compression"`, `"slm": {"max_context": 32768, "max_new_tokens": 3072}`. The prompts use the Qwen chat format; for Phi 4 mini, change `harness/prompts/context_compression/*.txt` to its `<|system|>`, `<|user|>`, `<|assistant|>` and `<|end|>` markers.
+
+## Steps
+
+1. Read the arena definition: `GET https://api.microtensor.cloud/v2/arenas/<arena>` gives the task, the base model and its licence, the tools, the budgets and the item counts.
+2. Train and compress your model from the arena's base model, write the training record, push `slm_repo/` to `huggingface.co/<user>/mt_slm_<arena>`. Note the commit sha.
+3. Train the router (`router/training/train_router.py`) on drafts from your own model, labelled by the arena's grader on the public sample items. Put the weights and tokenizer under `router/`.
+4. Set `"adapter"` in `config.json`, adjust the prompts and the verifier. Keep `harness/` and `escalation/` under 2 MB together.
+5. Write the manifest and check both repositories. Pick the hardware profile of the GPU you will run on (`evaluation.profiles` in the definition lists the allowed ones, https://www.microtensor.cloud/docs/certified-inference/system/hardware-profiles describes them), pass it to the manifest, and run the pre-flight check before you commit:
+   ```
+   mt submit manifest --artifact . --arena <arena> --slm ./slm_repo \
+     --slm-repo <user>/mt_slm_<arena> --slm-revision <sha> --profile <profile>
+   mt submit validate --artifact . --slm ./slm_repo
+   mt check-model --slm ./slm_repo --profile <profile>
+   ```
+   `--profile` writes `schema_version: 2` with `profile`, `weights_sha256` and the official image digest. `check-model` dry loads the config, tokenizer and safetensors headers on CPU and prints `ok` or the reason the mirror would refuse the model; a model that fails it is never evaluated.
+6. Push this repository to `huggingface.co/<user>/microtensor_<arena>`. Note the commit sha.
+7. Commit before the epoch's commit window closes. If the arena charges a fee, pay it first and cite the transfer; a free arena (every testnet arena) needs only the commit. The commit carries the same profile as the manifest and cannot change after Freeze 1:
+   ```
+   mt submit pay --arena <arena>
+   mt submit commit --arena <arena> --hf-user <user> --artifact-sha <sha> --profile <profile> --fee-block <block> --fee-index <index>
+   mt submit commit --arena <arena> --hf-user <user> --artifact-sha <sha> --profile <profile>
+   ```
+8. Run the evaluation yourself during the window and upload the transcript before Freeze 2. After the commit window closes the server mirrors your frozen files and, at window open, publishes the items. On a machine with your profile's GPU, start the official runner image from the subnet repository's `deploy/docker-compose.miner.yml` (it needs the Docker socket, the GPU, `/var/lib/microtensor` and your wallet):
+   ```
+   git clone https://github.com/microtensor-io/microtensor-subnet
+   cd microtensor-subnet/deploy
+   cp .env.example .env        # wallet names, MT_ARENA, MT_EPOCH, DOCKER_GID
+   docker compose -f docker-compose.miner.yml up
+   ```
+   It runs `mt miner run --arena <arena> --epoch <n>`: refuses unless the mirror lists you as `mirrored` on an allowed profile, pulls your frozen artifact, weights and the inputs bundle from the mirror (never from your live repositories), starts the pinned engine and sandbox, runs every pool in the fixed order, then writes, signs and uploads the transcript and prints the receipt `{"transcript_sha256", "received_at", "receipt_sig"}`. Keep the receipt; it proves the upload landed before the close block. A second upload before the deadline replaces the first; nothing lands after it. Before your first upload with a new harness, replay it the way validators will:
+   ```
+   mt miner replay --transcript transcript.json --arena <arena> --epoch <n>
+   ```
+   Nothing the runner times is used for scoring; a validator on your profile times your sampled calls against the reference on the same card. Arenas without the `evaluation` block skip this step: validators run everything.
+9. `mt submit status --arena <arena>` shows the epoch and your published scores once validators report.
+
+## Determinism rules
+
+Validators replay `main.py` on CPU against the transcript your run produced, and every request it makes must match the recording exactly and in order. Code that breaks one of these rules fails replay and scores zero for the epoch:
+
+- No clock reads beyond the frozen clock. The sandbox's time is fixed; no wall time, timers or timeouts of your own.
+- Randomness only from the provided seed. Every request carries `seed`; never use `random`, `numpy.random` or `os.urandom` unseeded.
+- No concurrency that can reorder calls. Model, tool and relay calls go out in a fixed order per item; no threads, no async fan out.
+- No file or network access outside the gateway. Read your own artifact, write nothing, reach only the four service URLs.
+
+The template's `harness/`, `escalation/` and `main.py` follow them. `mt miner replay` tells you before validators do.
+
+## Training playbook
+
+Prompting pays only through `harness/prompts/`; the model pool runs your weights under the arena's own prompt. There is no retrieval beyond the arena's tools on the item's database. Provenance fingerprints your weights against the declared base and against public checkpoints, so an untrained base never certifies. Work in this order:
+
+1. Measure the arena's base model on the dev items with the arena's grader; that is the score to beat.
+2. Set the adapter prompt, the verifier and the retry prompt; measure again.
+3. Distil: `scripts/distill.py --arena <arena> --items dev/items.jsonl --teacher <model> --out pairs.jsonl` asks a teacher through your own provider key and keeps only the answers the arena grader accepts.
+4. Fine tune: `scripts/train.py --base <base model> --data pairs.jsonl --eval held_out.jsonl --out merged/ --run-id r1 --method lora` trains on completions only under the arena prompt, merges the adapter, and writes the run record under `training/runs/r1/`. Use `--method qlora --fp16` on a T4, `--method full` for a 1.5B model on a 24 GB card.
+5. Reinforce on the families that still fail, with the grader as the reward or as preference pairs of a passing teacher answer against a failing draft.
+6. Compress to the format your envelope group takes: `scripts/compress.py --model merged/ --out slm_repo/ --method awq --calibration pairs.jsonl` for a GPU group, `--method gguf --llama-cpp <checkout>` for a CPU or phone group. Score before and after with `scripts/score_compress.py` and write both into `training/compression/before_after.json`.
+7. Hold out whole domains, build twins as the twin rule does, run `mt envelope check` and `mt check-model`, then train the router on this model's drafts.
+
+Items are fresh every epoch and the twin gap is deducted, so train the skill, not the sample data. The full playbook, with the hardware each step needs: https://www.microtensor.cloud/docs/certified-inference/guides/training
+
+## The sandbox contract
+
+Your `main.py` listens on `MT_PORT` and may reach only four addresses: `MT_SLM_URL` (your model, OpenAI completions, greedy), `MT_ROUTER_URL` (your router, `POST /decide`), `MT_TOOLS_URL` (the arena's tools, if it has any), and `MT_RELAY_URL` (`POST /v1/chat/completions` with the `x-mt-ticket` header, only inside a repair). Tool answers are sometimes faulty on purpose, and sometimes carry instructions; following them scores the item wrong. A request that times out, crashes or returns an invalid answer counts as wrong. The four addresses point at the recording gateway inside the official image, the clock is frozen, and each request body carries the item's `seed`.
+
+## Licence
+
+This template, your derivative of it and the model you submit are licensed under the Microtensor submission terms (see `LICENSE`); none of it is published under a free licence. Each arena lists which base model licences it accepts as inputs, because the base's terms must allow a commercial derivative.

config.json

@@ -0,0 +1,17 @@
+{
+  "adapter": "generic",
+  "slm": {
+    "max_context": 8192,
+    "max_new_tokens": 256
+  },
+  "router": {
+    "max_input_tokens": 4096
+  },
+  "harness": {
+    "max_steps": 3,
+    "max_total_tokens": 32000
+  },
+  "escalation": {
+    "preferred_model": "gpt-5"
+  }
+}

escalation/patcher.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from harness.adapters import Adapter
+
+PROMPTS = Path(__file__).resolve().parent / "prompts"
+MAX_TOKENS = 512
+TRACE_LIMIT = 1200
+
+
+def _trim(text: str, limit: int) -> str:
+    return text if len(text) <= limit else text[-limit:]
+
+
+def repair(
+    task: dict[str, Any],
+    draft: Any,
+    trace: str,
+    errors: str,
+    *,
+    adapter: Adapter,
+    relay_url: str,
+    ticket: str,
+    model: str,
+) -> Any:
+    system = adapter.render(PROMPTS, "system", task)
+    user = adapter.render(
+        PROMPTS,
+        "repair",
+        task,
+        draft=adapter.as_text(draft),
+        errors=_trim(errors or "no verifier output", 400),
+        trace=_trim(trace or "", TRACE_LIMIT),
+    )
+    body = {
+        "model": model,
+        "max_tokens": MAX_TOKENS,
+        "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}],
+    }
+    with httpx.Client(timeout=180.0) as http:
+        response = http.post(
+            relay_url.rstrip("/") + "/v1/chat/completions",
+            json=body,
+            headers={"x-mt-ticket": ticket},
+        )
+    if response.status_code != 200:
+        return draft
+    choices = response.json().get("choices") or [{}]
+    text = str((choices[0].get("message") or {}).get("content") or "")
+    fixed = adapter.clean(text)
+    return fixed or draft

escalation/prompts/context_compression/repair.txt

@@ -0,0 +1,15 @@
+Instruction:
+{system}
+
+Session:
+{user}
+
+Draft compaction:
+{draft}
+
+Why the draft is wrong: {errors}
+
+Harness trace:
+{trace}
+
+Return the corrected compaction within the token budget the instruction states.

escalation/prompts/context_compression/system.txt

@@ -0,0 +1 @@
+You repair the compaction of a coding agent's session. Reply with one JSON object and nothing else, with exactly the fields summary, files_touched, open_tasks, errors and decisions; every list entry is a plain string. Copy file paths, function names and error messages exactly as they appear in the session. Use only facts from the session.

escalation/prompts/generic/repair.txt

@@ -0,0 +1,12 @@
+Task:
+{fields}
+
+Draft answer:
+{draft}
+
+Why the draft is wrong: {errors}
+
+Harness trace:
+{trace}
+
+Return the corrected answer.

escalation/prompts/generic/system.txt

@@ -0,0 +1 @@
+You repair answers to tasks. Reply with the corrected answer only, no prose, no code fences.

escalation/prompts/sql_generation/repair.txt

@@ -0,0 +1,14 @@
+Schema:
+{schema}
+
+Question: {question}
+
+Draft query:
+{draft}
+
+Why the draft is wrong: {errors}
+
+Harness trace:
+{trace}
+
+Return the corrected query.

escalation/prompts/sql_generation/system.txt

@@ -0,0 +1 @@
+You repair SQLite queries. Reply with one corrected query only, no prose, no code fences.

harness/adapters/__init__.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from harness.adapters.base import Adapter, Verdict
+from harness.adapters.context_compression import CompressionAdapter
+from harness.adapters.generic import GenericAdapter
+from harness.adapters.sql_generation import SqlAdapter
+
+ADAPTERS: dict[str, Adapter] = {
+    GenericAdapter.name: GenericAdapter(),
+    SqlAdapter.name: SqlAdapter(),
+    CompressionAdapter.name: CompressionAdapter(),
+}
+
+
+def load(name: str) -> Adapter:
+    try:
+        return ADAPTERS[name]
+    except KeyError as exc:
+        raise ValueError(f"unknown adapter {name!r}; one of {sorted(ADAPTERS)}") from exc
+
+
+__all__ = ["ADAPTERS", "Adapter", "Verdict", "load"]

harness/adapters/base.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+_FENCE = re.compile(r"```[a-zA-Z]*")
+
+
+@dataclass(frozen=True)
+class Verdict:
+    ok: bool
+    answer: Any
+    reason: str = ""
+
+
+class SafeDict(dict):
+    def __missing__(self, key: str) -> str:
+        return ""
+
+
+def strip_fences(text: str) -> str:
+    return _FENCE.sub("\n", text).strip()
+
+
+def fields_of(task: dict[str, Any]) -> str:
+    lines = []
+    for key, value in task.items():
+        body = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False)
+        lines.append(f"{key}: {body}")
+    return "\n".join(lines)
+
+
+class Adapter:
+    name = "generic"
+    stop: list[str] = ["\n\n\n"]
+
+    def render(self, prompts: Path, kind: str, task: dict[str, Any], **extra: Any) -> str:
+        template = (prompts / self.name / f"{kind}.txt").read_text(encoding="utf-8")
+        values = SafeDict(task_json=json.dumps(task, ensure_ascii=False), fields=fields_of(task))
+        values.update({k: v for k, v in task.items() if isinstance(v, str | int | float)})
+        values.update({k: (v if isinstance(v, str) else json.dumps(v)) for k, v in extra.items()})
+        return template.format_map(values)
+
+    def query(self, task: dict[str, Any]) -> str:
+        for key in ("question", "prompt", "input", "text", "instruction"):
+            if isinstance(task.get(key), str):
+                return str(task[key])
+        return fields_of(task)
+
+    def clean(self, text: str) -> str:
+        return strip_fences(text)
+
+    def verify(self, text: str, task: dict[str, Any], tools_url: str) -> Verdict:
+        answer = self.clean(text)
+        if not answer:
+            return Verdict(False, "", "empty answer")
+        return Verdict(True, answer)
+
+    def as_text(self, answer: Any) -> str:
+        return answer if isinstance(answer, str) else json.dumps(answer, ensure_ascii=False)

harness/adapters/context_compression.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+from harness.adapters.base import Adapter, Verdict, strip_fences
+
+FIELDS = ("summary", "files_touched", "open_tasks", "errors", "decisions")
+BUDGET = re.compile(r"within (\d+) tokens")
+CHARS_PER_TOKEN = 3.5
+
+
+def _text(value: Any) -> str:
+    if isinstance(value, str):
+        return value.strip()
+    if isinstance(value, dict):
+        return "; ".join(f"{key}: {_text(item)}" for key, item in value.items() if _text(item))
+    if isinstance(value, list):
+        return "; ".join(_text(item) for item in value if _text(item))
+    return "" if value is None else str(value)
+
+
+def normalise(found: dict[str, Any]) -> dict[str, Any]:
+    summary = _text(found.get("summary"))
+    extra = [f"{key}: {_text(value)}" for key, value in found.items() if key not in FIELDS and _text(value)]
+    if extra:
+        summary = " ".join([summary, *extra]).strip()
+    shaped: dict[str, Any] = {"summary": summary}
+    for key in FIELDS[1:]:
+        value = found.get(key)
+        entries = value if isinstance(value, list) else ([] if value in (None, "") else [value])
+        shaped[key] = [entry for entry in (_text(item) for item in entries) if entry]
+    return shaped
+
+
+class CompressionAdapter(Adapter):
+    name = "context_compression"
+    stop = ["<|im_end|>", "<|endoftext|>", "<|end|>"]
+
+    def clean(self, text: str) -> str:
+        body = strip_fences(text)
+        start, end = body.find("{"), body.rfind("}")
+        return body[start : end + 1] if start >= 0 and end > start else body
+
+    def verify(self, text: str, task: dict[str, Any], tools_url: str) -> Verdict:
+        body = self.clean(text)
+        try:
+            found = json.loads(body)
+        except ValueError:
+            return Verdict(False, body, "the reply is not one JSON object")
+        if not isinstance(found, dict):
+            return Verdict(False, body, "the reply is not one JSON object")
+        if not _text(found.get("summary")):
+            return Verdict(False, body, "the summary field is missing or empty")
+        answer = json.dumps(normalise(found), ensure_ascii=False)
+        budget = BUDGET.search(str(task.get("system") or ""))
+        if budget and len(answer) / CHARS_PER_TOKEN > int(budget.group(1)):
+            return Verdict(
+                False,
+                answer,
+                f"the reply is longer than {budget.group(1)} tokens; shorten the summary and the lists",
+            )
+        return Verdict(True, answer)

harness/adapters/generic.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from typing import Any
+
+from harness.adapters.base import Adapter, Verdict
+
+
+class GenericAdapter(Adapter):
+    name = "generic"
+    max_chars = 4000
+
+    def verify(self, text: str, task: dict[str, Any], tools_url: str) -> Verdict:
+        answer = self.clean(text)
+        if not answer:
+            return Verdict(False, "", "empty answer")
+        if len(answer) > self.max_chars:
+            return Verdict(False, answer, f"answer longer than {self.max_chars} characters")
+        if "ignore the task" in answer.lower():
+            return Verdict(False, answer, "answer repeats injected instructions")
+        return Verdict(True, answer)

harness/adapters/sql_generation.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from harness.adapters.base import Adapter, Verdict, strip_fences
+
+_FORBIDDEN = re.compile(r"\b(attach|detach|pragma|vacuum|insert|update|delete|drop|alter)\b", re.I)
+
+
+def schema_lookup(tools_url: str) -> str:
+    try:
+        with httpx.Client(timeout=20.0) as http:
+            response = http.post(tools_url.rstrip("/") + "/tools/schema_lookup", json={})
+        response.raise_for_status()
+        tables = response.json().get("tables") or []
+    except (httpx.HTTPError, ValueError):
+        return ""
+    return "\n".join(
+        str(t.get("sql") or "") for t in tables if "ignore the task" not in str(t.get("sql") or "").lower()
+    )
+
+
+class SqlAdapter(Adapter):
+    name = "sql_generation"
+    stop = ["\n\n", ";"]
+
+    def render(self, prompts: Path, kind: str, task: dict[str, Any], **extra: Any) -> str:
+        if not task.get("schema") and extra.get("tools_url"):
+            task = {**task, "schema": schema_lookup(str(extra["tools_url"]))}
+        return super().render(prompts, kind, task, **extra)
+
+    def clean(self, text: str) -> str:
+        body = strip_fences(text)
+        body = re.sub(r"^\s*(sql|sqlite|query|answer)\s*:\s*", "", body, flags=re.I)
+        return body.split(";", 1)[0].strip()
+
+    def verify(self, text: str, task: dict[str, Any], tools_url: str) -> Verdict:
+        sql = self.clean(text)
+        if not sql:
+            return Verdict(False, "", "empty answer")
+        if not sql.lower().startswith(("select", "with")):
+            return Verdict(False, sql, "not a SELECT statement")
+        if _FORBIDDEN.search(sql):
+            return Verdict(False, sql, "forbidden statement")
+        import sqlglot
+
+        try:
+            parsed = sqlglot.parse_one(sql, read="sqlite")
+        except sqlglot.errors.ParseError as exc:
+            return Verdict(False, sql, f"does not parse: {str(exc)[:120]}")
+        schema = str(task.get("schema") or "")
+        tables = {t.name.lower() for t in parsed.find_all(sqlglot.exp.Table)}
+        known = {
+            name.lower()
+            for name in re.findall(r"create\s+table\s+\"?([A-Za-z_][A-Za-z0-9_]*)", schema, re.I)
+        }
+        unknown = sorted(tables - known) if known else []
+        if unknown:
+            return Verdict(False, sql, f"unknown tables: {', '.join(unknown)}")
+        try:
+            with httpx.Client(timeout=20.0) as http:
+                response = http.post(
+                    tools_url.rstrip("/") + "/tools/query_execution", json={"sql": sql}
+                )
+            if response.status_code == 200:
+                body = response.json()
+                if isinstance(body, dict) and body.get("error"):
+                    return Verdict(False, sql, f"execution error: {str(body['error'])[:120]}")
+        except (httpx.HTTPError, ValueError):
+            pass
+        return Verdict(True, sql)

harness/engine.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from harness.adapters import Adapter
+from harness.verifier import verify
+
+PROMPTS = Path(__file__).resolve().parent / "prompts"
+
+
+@dataclass
+class HarnessResult:
+    answer: Any
+    verified: bool
+    errors: str = ""
+    trace: str = ""
+    steps: int = 0
+    attempts: list[str] = field(default_factory=list)
+
+
+def complete(slm_url: str, prompt: str, max_tokens: int, stop: list[str]) -> str:
+    with httpx.Client(timeout=120.0) as http:
+        response = http.post(
+            slm_url.rstrip("/") + "/v1/completions",
+            json={"model": "slm", "prompt": prompt, "max_tokens": max_tokens, "stop": stop},
+        )
+    response.raise_for_status()
+    choices = response.json().get("choices") or [{}]
+    return str(choices[0].get("text") or "").strip()
+
+
+def run_harness(
+    task: dict[str, Any],
+    *,
+    adapter: Adapter,
+    slm_url: str,
+    tools_url: str,
+    config: dict[str, Any],
+) -> HarnessResult:
+    max_tokens = int((config.get("slm") or {}).get("max_new_tokens") or 256)
+    max_steps = int((config.get("harness") or {}).get("max_steps") or 3)
+    errors = ""
+    answer = ""
+    trace: list[str] = []
+    attempts: list[str] = []
+    for step in range(1, max_steps + 1):
+        kind = "task" if not errors else "retry"
+        prompt = adapter.render(
+            PROMPTS, kind, task, draft=answer, errors=errors, tools_url=tools_url
+        )
+        answer = complete(slm_url, prompt, max_tokens, adapter.stop)
+        attempts.append(answer)
+        verdict = verify(adapter, answer, task, tools_url)
+        trace.append(f"step {step}: {'ok' if verdict.ok else verdict.reason}")
+        if verdict.ok:
+            return HarnessResult(
+                answer=verdict.answer, verified=True, trace="\n".join(trace), steps=step, attempts=attempts
+            )
+        errors = verdict.reason
+    return HarnessResult(
+        answer=adapter.clean(answer),
+        verified=False,
+        errors=errors,
+        trace="\n".join(trace),
+        steps=max_steps,
+        attempts=attempts,
+    )

harness/prompts/context_compression/retry.txt

@@ -0,0 +1,9 @@
+<|im_start|>system
+{system}<|im_end|>
+<|im_start|>user
+{user}<|im_end|>
+<|im_start|>assistant
+{draft}<|im_end|>
+<|im_start|>user
+That reply was rejected: {errors}. Reply again with only the JSON object with exactly the five fields summary, files_touched, open_tasks, errors and decisions, every list entry a plain string.<|im_end|>
+<|im_start|>assistant

harness/prompts/context_compression/task.txt

@@ -0,0 +1,5 @@
+<|im_start|>system
+{system}<|im_end|>
+<|im_start|>user
+{user}<|im_end|>
+<|im_start|>assistant

harness/prompts/generic/retry.txt

@@ -0,0 +1,11 @@
+You answer tasks precisely. Reply with the answer only, no explanation.
+
+{fields}
+
+A previous attempt was rejected.
+Attempt:
+{draft}
+Problem: {errors}
+
+Write a corrected answer.
+Answer:

harness/prompts/generic/task.txt

@@ -0,0 +1,5 @@
+You answer tasks precisely. Reply with the answer only, no explanation.
+
+{fields}
+
+Answer:

harness/prompts/sql_generation/retry.txt

@@ -0,0 +1,14 @@
+You are a SQLite expert. Answer with exactly one SQL query and nothing else.
+
+Schema:
+{schema}
+
+Question: {question}
+
+A previous attempt was rejected.
+Attempt:
+{draft}
+Problem: {errors}
+
+Write a corrected query.
+SQL:

harness/prompts/sql_generation/task.txt

@@ -0,0 +1,8 @@
+You are a SQLite expert. Answer with exactly one SQL query and nothing else.
+
+Schema:
+{schema}
+
+Question: {question}
+
+SQL:

harness/verifier.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from typing import Any
+
+from harness.adapters import Adapter, Verdict
+
+
+def verify(adapter: Adapter, text: str, task: dict[str, Any], tools_url: str) -> Verdict:
+    return adapter.verify(text, task, tools_url)

main.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+import httpx
+from fastapi import FastAPI
+from pydantic import BaseModel, Field
+
+ROOT = Path(__file__).resolve().parent
+sys.path.insert(0, str(ROOT))
+
+from escalation.patcher import repair  # noqa: E402
+from harness.adapters import load  # noqa: E402
+from harness.engine import run_harness  # noqa: E402
+
+CONFIG: dict[str, Any] = json.loads((ROOT / "config.json").read_text(encoding="utf-8"))
+ADAPTER = load(str(CONFIG.get("adapter") or "generic"))
+SLM_URL = os.environ["MT_SLM_URL"]
+ROUTER_URL = os.environ.get("MT_ROUTER_URL", "")
+TOOLS_URL = os.environ["MT_TOOLS_URL"]
+RELAY_URL = os.environ["MT_RELAY_URL"]
+PREFERRED_MODEL = str((CONFIG.get("escalation") or {}).get("preferred_model") or "")
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+
+class HarnessIn(BaseModel):
+    item_id: str
+    task: dict[str, Any]
+
+
+class RepairIn(BaseModel):
+    item_id: str
+    task: dict[str, Any]
+    draft: Any = None
+    trace: str = ""
+    errors: str = ""
+    ticket: str = ""
+
+
+class SystemIn(BaseModel):
+    item_id: str
+    task: dict[str, Any]
+    floor: float = Field(ge=0.0, le=1.0)
+    ticket: str = ""
+
+
+def decide(query: str, draft: str) -> float:
+    if not ROUTER_URL:
+        return 1.0
+    with httpx.Client(timeout=30.0) as http:
+        response = http.post(
+            ROUTER_URL.rstrip("/") + "/decide", json={"input": {"query": query, "draft": draft}}
+        )
+    response.raise_for_status()
+    answers = response.json().get("answers") or []
+    return float(answers[0]["probability"]) if answers else 0.0
+
+
+def _repair(payload: RepairIn | SystemIn, draft: Any, trace: str, errors: str) -> Any:
+    return repair(
+        payload.task,
+        draft,
+        trace,
+        errors,
+        adapter=ADAPTER,
+        relay_url=RELAY_URL,
+        ticket=payload.ticket,
+        model=PREFERRED_MODEL,
+    )
+
+
+@app.get("/health")
+def health() -> dict[str, str]:
+    return {"status": "ok", "adapter": ADAPTER.name}
+
+
+@app.post("/harness/run")
+def harness_run(payload: HarnessIn) -> dict[str, Any]:
+    result = run_harness(
+        payload.task, adapter=ADAPTER, slm_url=SLM_URL, tools_url=TOOLS_URL, config=CONFIG
+    )
+    return {
+        "answer": result.answer,
+        "self_check": {"pass": result.verified},
+        "trace": result.trace,
+        "steps": result.steps,
+    }
+
+
+@app.post("/escalation/repair")
+def escalation_repair(payload: RepairIn) -> dict[str, Any]:
+    return {"answer": _repair(payload, payload.draft, payload.trace, payload.errors)}
+
+
+@app.post("/system/run")
+def system_run(payload: SystemIn) -> dict[str, Any]:
+    result = run_harness(
+        payload.task, adapter=ADAPTER, slm_url=SLM_URL, tools_url=TOOLS_URL, config=CONFIG
+    )
+    draft = result.answer
+    probability = decide(ADAPTER.query(payload.task), ADAPTER.as_text(draft))
+    if result.verified and probability >= payload.floor:
+        return {"answer": draft, "route": "local", "p": probability}
+    answer = _repair(payload, draft, result.trace, result.errors)
+    return {"answer": answer, "route": "escalated", "p": probability}
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("MT_PORT", "8080")), log_level="warning")

manifest.example.json

@@ -0,0 +1,26 @@
+{
+  "schema_version": 2,
+  "arena": "<arena>",
+  "components": ["slm", "harness", "router", "escalation"],
+  "hardware_class": "mt-5",
+  "profile": "P1",
+  "weights_sha256": "<sha256 over the sorted weight file digests>",
+  "image_digest": "sha256:<the runner image digest from the arena definition>",
+  "slm": {
+    "repo": "<user>/mt_slm_<arena>",
+    "revision": "<40 hex git commit sha>",
+    "files": {"model.safetensors": "sha256:...", "config.json": "sha256:..."}
+  },
+  "files": {
+    "config.json": "sha256:...",
+    "main.py": "sha256:...",
+    "requirements.txt": "sha256:...",
+    "router/config.json": "sha256:...",
+    "router/model.safetensors": "sha256:...",
+    "harness/engine.py": "sha256:...",
+    "harness/verifier.py": "sha256:...",
+    "harness/adapters/__init__.py": "sha256:...",
+    "harness/prompts/generic/task.txt": "sha256:...",
+    "escalation/patcher.py": "sha256:..."
+  }
+}

manifest.json

@@ -0,0 +1,62 @@
+{
+  "arena": "ctx_compress",
+  "components": [
+    "slm"
+  ],
+  "files": {
+    ".gitignore": "sha256:e7ab22595b29e55626e02c418ce6d59e580da857ded2a08f903e6d7ef1aec090",
+    "LICENSE": "sha256:30746c09020f9d1cd2882d681e7b972848f83a128fef195575a41a1e41fdec7e",
+    "README.md": "sha256:0b8c5dd5386c9c0cd0e090c98f578aa9fd9a45accd37207e09d556328a5e4b99",
+    "config.json": "sha256:59dcfc49e921e04c16dc538547db0af8d56b4ee1acbe1a6da46c59588b934442",
+    "escalation/patcher.py": "sha256:600f78f383d94cda69381ddf0d84873024f90bad4b058b9e006fa8b2dba6c779",
+    "escalation/prompts/context_compression/repair.txt": "sha256:f63b5f7a16621b6c3b471fb229410414866c2913abd695f26201c6f09e2194c2",
+    "escalation/prompts/context_compression/system.txt": "sha256:2f3bb96f667a24c1f50f2c6f4d856fefe06642b9b71cec8887596828ef79aeea",
+    "escalation/prompts/generic/repair.txt": "sha256:734091ff1ab1be571b6fcd2d450c265f55c431ad4c10151b8d31c1e37ec57023",
+    "escalation/prompts/generic/system.txt": "sha256:1191cfb67c22351adc36774fe5da8842cfa256a9f780ca1b41b7b7cdbfd9b13f",
+    "escalation/prompts/sql_generation/repair.txt": "sha256:7cdc2b0f488c003fbdc95be46268491aac9bfc8ae078390d03b2e2702932bd5e",
+    "escalation/prompts/sql_generation/system.txt": "sha256:a82c3d64ac057bb7b376270fcd180e680969cdbd3290ca67429d60599979c659",
+    "harness/adapters/__init__.py": "sha256:64899ed354e462e8bf6b278244401259e2d798943559382ec01bed7c36ee6d7b",
+    "harness/adapters/base.py": "sha256:b04da8e04e848e1b6d9edc4105391212121010ace0ac257421671535603a96b0",
+    "harness/adapters/context_compression.py": "sha256:47d8b1ccadbd4d7921dc78c4076f829d727c3f37b59dd0622081644997a3c3f0",
+    "harness/adapters/generic.py": "sha256:7ca2b1d5b1e9dec0d83820881f85744fcd3ae42d3a75d117447aead906872e5b",
+    "harness/adapters/sql_generation.py": "sha256:1e392d15645901fd080adf497cac1ec2b817baf5eb71959958d1c00302e6e21e",
+    "harness/engine.py": "sha256:87d33738a0c04fe0314e7f37bee85d3f7a6e7b7e94ff51f0c8ab6eac3347f768",
+    "harness/prompts/context_compression/retry.txt": "sha256:0a307850ab2187825dad4d3d36fb43adea4fdb6e0c4bd516d1a18f994982a99d",
+    "harness/prompts/context_compression/task.txt": "sha256:826ca63bccd76232f5aeaf0dd85d79e7747f6cf312e9ec469a4696d114d37d0e",
+    "harness/prompts/generic/retry.txt": "sha256:6f4741f5c4daf44e03fa55b0b5c6a10c7e84af780789e019a095060d3a6b7b06",
+    "harness/prompts/generic/task.txt": "sha256:695a5284f2c90433fd97617da4c82cdcb623d0ab091cca861f3f1d628a16a16c",
+    "harness/prompts/sql_generation/retry.txt": "sha256:9da2f6b58c3030c626dc5cff6c7799ab979779371c76382c0a802836a857601c",
+    "harness/prompts/sql_generation/task.txt": "sha256:c3a8b6c7028f60614b3c9c1240ab7f98fe51cd4455b7c025ebe972066f0fd994",
+    "harness/verifier.py": "sha256:8be9c46674ed2a4c5e60457f9aefd767d907b697bb5978c9bc48698015444681",
+    "main.py": "sha256:b58cf3812b208b02e9b08d8434e6dae6e709f6352470c5ab10a0a8dd5de42d59",
+    "manifest.example.json": "sha256:da11b505220ea6e1c8851adef33a68301a7bc2c49c4d35d8e32d1c4f7fbf3e4e",
+    "requirements.txt": "sha256:db65ff41ff4d68c613b1312491e7af60a37c1769e28e215c4f510e3614b77551",
+    "router/config.json": "sha256:cb6ea49716f2dd71ff62ed030fd6e525ccebf46030f017df786f1aa35ac1e077",
+    "router/training/data_sources.json": "sha256:5e1a5ba89b09eab573ffd00de540cf4a980541e0c75ad48be54ea38643d31d2f",
+    "router/training/metrics.jsonl": "sha256:794318db32f615ec49bfbea4ff12a274ad5c1676fad6ddb71c1d40b3a23c7be6",
+    "router/training/recipe.json": "sha256:bba9519bece1a14f222b64c2bf5feb1c92e1d37aae39f79ea1c5fa1799981e6f",
+    "router/training/train_router.py": "sha256:55886c9eccf006090a00bd2293cc4a334396f99eb531cbe539f8c461b4a8903a"
+  },
+  "hardware_class": "mt-5",
+  "image_digest": "sha256:aada28e6a4b50961d10bf63cb04c30e8449e4bfb6e385fbad7f387cf8eb9b9dd",
+  "profile": "P1",
+  "schema_version": 2,
+  "slm": {
+    "files": {
+      "LICENSE": "sha256:30746c09020f9d1cd2882d681e7b972848f83a128fef195575a41a1e41fdec7e",
+      "README.md": "sha256:93d876f9b25a1daad71e417227329417d62fc1f6df4627aaa20d951342dbf32a",
+      "chat_template.jinja": "sha256:9524df67b77a7b25a2dfee898f75b316a157eb9d855b51e32aeac79d7c8a83ce",
+      "config.json": "sha256:bd828b00edcc25f93d6fa55210acebcd058054c4d5b5fa6c5146b0a0d18172ee",
+      "generation_config.json": "sha256:7c04cb9d2ba771f7528fba5a7104999cdaf7566d02b5fbd58472829f62716177",
+      "model-00001-of-00002.safetensors": "sha256:0ef60a1197e76820688be0d36ff8405f66b64bf2c8f7fa01a1d82ace5741a6c5",
+      "model-00002-of-00002.safetensors": "sha256:ae3aa72e003040c10d632c59a71a7a2fbebb8d04c058f5095afe96a14ee883d4",
+      "model.safetensors.index.json": "sha256:6e2bc176baee38bfb8a177d324e5db720227bde37c6a072d06b45f390858e2ea",
+      "special_tokens_map.json": "sha256:c08676c49fd7969a3130f72be6d4bf34da66aa484a6e21dffe359893a1bd5f2e",
+      "tokenizer.json": "sha256:e2bad66439538cb4d5a7580680932432ed9ece9d3b8577e675512bdf11599253",
+      "tokenizer_config.json": "sha256:a5ec5daab12ba090a90f3dd169c8f9c275557013a87b9c1258dc7cb497a35c86"
+    },
+    "repo": "Daniel01410/mt_slm_ctx_compress",
+    "revision": "paste_from_above"
+  },
+  "weights_sha256": "b0bd36e4ffcbf6e55762df1ed07d662c8eca53591e172bf845001bdcd133ed03"
+}

requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.12
+uvicorn==0.34.2
+httpx==0.28.1
+pydantic==2.11.4
+sqlglot==26.16.4

router/config.json

@@ -0,0 +1,7 @@
+{
+  "architectures": ["DebertaV2ForSequenceClassification"],
+  "model_type": "deberta-v2",
+  "num_labels": 1,
+  "problem_type": "regression",
+  "note": "replace this file, model.safetensors and the tokenizer files with your trained router; see router/training/train_router.py"
+}

router/training/data_sources.json

@@ -0,0 +1 @@
+[{"name": "own-slm-drafts", "version": "s1", "license": "self", "sha256": ""}]

router/training/metrics.jsonl

@@ -0,0 +1 @@
+{"epoch": 1, "loss": 0.0, "ece": 0.0, "auroc": 0.0, "note": "replace with your training metrics"}

router/training/recipe.json

@@ -0,0 +1,9 @@
+{
+  "base": "microsoft/deberta-v3-xsmall",
+  "method": "sequence classification head, one logit, trained on (query, draft, y) pairs",
+  "pairs": "drafts from your own SLM on the arena's public sample items, labelled by the arena's grader",
+  "epochs": 3,
+  "learning_rate": 3e-05,
+  "max_input_tokens": 4096,
+  "calibration": "temperature fit on a held out split so sigmoid(logit) is a probability"
+}

router/training/train_router.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="train a one logit router on (query, draft, y) rows")
+    parser.add_argument("--pairs", type=Path, required=True, help="jsonl with query, draft, y")
+    parser.add_argument("--base", default="microsoft/deberta-v3-xsmall")
+    parser.add_argument("--out", type=Path, default=Path(__file__).resolve().parents[1])
+    parser.add_argument("--epochs", type=int, default=3)
+    parser.add_argument("--max-input-tokens", type=int, default=4096)
+    args = parser.parse_args()
+
+    import torch
+    from torch.utils.data import DataLoader
+    from transformers import AutoModelForSequenceClassification, AutoTokenizer
+
+    rows = [json.loads(line) for line in args.pairs.read_text(encoding="utf-8").splitlines() if line.strip()]
+    tokenizer = AutoTokenizer.from_pretrained(args.base)
+    model = AutoModelForSequenceClassification.from_pretrained(args.base, num_labels=1)
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    model.to(device)
+    optimizer = torch.optim.AdamW(model.parameters(), lr=3e-5)
+    loss_fn = torch.nn.BCEWithLogitsLoss()
+
+    def batches():
+        for start in range(0, len(rows), 16):
+            chunk = rows[start : start + 16]
+            encoded = tokenizer(
+                [r["query"] for r in chunk],
+                [r["draft"] for r in chunk],
+                truncation=True,
+                max_length=args.max_input_tokens,
+                padding=True,
+                return_tensors="pt",
+            )
+            labels = torch.tensor([float(r["y"]) for r in chunk])
+            yield {k: v.to(device) for k, v in encoded.items()}, labels.to(device)
+
+    model.train()
+    for epoch in range(args.epochs):
+        total = 0.0
+        for encoded, labels in DataLoader(list(batches()), batch_size=None):
+            optimizer.zero_grad()
+            logits = model(**encoded).logits.reshape(-1)
+            loss = loss_fn(logits, labels)
+            loss.backward()
+            optimizer.step()
+            total += float(loss)
+        print(f"epoch {epoch + 1}: loss {total:.4f}")
+    model.save_pretrained(args.out, safe_serialization=True)
+    tokenizer.save_pretrained(args.out)
+    print(f"saved to {args.out}")
+
+
+if __name__ == "__main__":
+    main()

slm_repo/LICENSE

@@ -0,0 +1,8 @@
+Copyright (c) 2026 Microtensor. All rights reserved.
+
+This template and every derivative of it are licensed to you solely for
+building and submitting systems to Microtensor arenas under the Microtensor
+submission terms (https://www.microtensor.cloud/terms). No other use,
+redistribution or sublicensing is permitted without written permission from
+Microtensor. Models submitted to an arena are licensed to Microtensor under
+the submission terms and are not published under a free licence.

slm_repo/README.md

@@ -0,0 +1,22 @@
+---
+license: other
+license_name: microtensor-submission
+license_link: https://www.microtensor.cloud/terms
+base_model: <allowed base>
+tags:
+- microtensor
+- <arena>
+---
+
+# mt_slm_<arena>
+
+A specialised model for the Microtensor `<arena>` arena.
+
+| Field | Value |
+|---|---|
+| Base model | an allowed base from the arena definition, with its licence |
+| Method | see `training/recipe.json` |
+| Arena | <arena> |
+| Scores | filled in from the ledger after the first epoch |
+
+The root of this repository loads with `AutoModelForCausalLM`. Everything under `training/` is the record validators require and the archive keeps; it is never scored.

slm_repo/training/compression/before_after.json

@@ -0,0 +1 @@
+{"before": {"execution_accuracy": 0.0}, "after": {"execution_accuracy": 0.0}}

slm_repo/training/compression/compression.json

@@ -0,0 +1 @@
+{"method": "awq", "bits": 4, "calibration_data_sha256": ""}

slm_repo/training/data/data_card.md

@@ -0,0 +1,3 @@
+# Data card
+
+Describe what the training data is and how it was built: the public sample items from the arena, synthetic questions generated against your own schemas, and the teacher model that answered them. Hidden arena items never appear here because you never see them.

slm_repo/training/data/sources.json

@@ -0,0 +1,14 @@
+[
+  {
+    "name": "microtensor/<arena>-sample",
+    "version": "",
+    "license": "",
+    "sha256": ""
+  },
+  {
+    "name": "own-synthetic",
+    "version": "1",
+    "license": "proprietary",
+    "sha256": ""
+  }
+]

slm_repo/training/data/synthetic.json

@@ -0,0 +1 @@
+{"teacher_model": "", "prompts": ["see scripts/distill.py"]}

slm_repo/training/distillation/teacher.json

@@ -0,0 +1 @@
+{"teacher_model": "", "teacher_version": "", "harness_used": "harness/ from the artifact repository"}

slm_repo/training/recipe.json

@@ -0,0 +1,15 @@
+{
+  "base_model": "<allowed base>",
+  "method": "distillation",
+  "harness_aware": false,
+  "hyperparameters": {
+    "epochs": 2,
+    "learning_rate": 1e-05,
+    "batch_size": 16,
+    "max_length": 4096
+  },
+  "seeds": [
+    0
+  ],
+  "compression": "awq-int4"
+}

slm_repo/training/runs/example/config.json

@@ -0,0 +1 @@
+{"run_id": "example", "base_model": "Qwen/Qwen2.5-Coder-7B-Instruct", "epochs": 2, "learning_rate": 1e-5}

slm_repo/training/runs/example/env.json

@@ -0,0 +1 @@
+{"libraries": {"torch": "", "transformers": ""}, "gpu": "", "hours": 0, "profile": ""}

slm_repo/training/runs/example/final_eval.json

@@ -0,0 +1 @@
+{"execution_accuracy": 0.0, "items": 0}

slm_repo/training/runs/example/metrics.jsonl

@@ -0,0 +1 @@
+{"step": 0, "loss": 0.0, "eval_execution_accuracy": 0.0, "note": "replace with your real curves"}

slm_repo/training/scripts/compress.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import random
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Any
+
+
+def calibration(path: str, count: int, seed: int) -> list[str]:
+    rows = [json.loads(line) for line in Path(path).read_text(encoding="utf-8").splitlines() if line.strip()]
+    texts = [str(r.get("prompt", "")) + str(r.get("completion", "")) for r in rows if r.get("prompt")]
+    random.Random(seed).shuffle(texts)
+    return texts[:count]
+
+
+def quantise_awq(model: str, out: Path, samples: list[str], bits: int, group: int) -> None:
+    from awq import AutoAWQForCausalLM
+    from transformers import AutoTokenizer
+
+    tokenizer = AutoTokenizer.from_pretrained(model)
+    quantiser = AutoAWQForCausalLM.from_pretrained(model, safetensors=True)
+    quantiser.quantize(
+        tokenizer,
+        quant_config={"zero_point": True, "q_group_size": group, "w_bit": bits, "version": "GEMM"},
+        calib_data=samples,
+    )
+    quantiser.save_quantized(str(out), safetensors=True)
+    tokenizer.save_pretrained(str(out))
+
+
+def quantise_gptq(model: str, out: Path, samples: list[str], bits: int, group: int) -> None:
+    from gptqmodel import GPTQModel, QuantizeConfig
+    from transformers import AutoTokenizer
+
+    tokenizer = AutoTokenizer.from_pretrained(model)
+    quantiser = GPTQModel.load(model, QuantizeConfig(bits=bits, group_size=group, desc_act=False))
+    quantiser.quantize(samples, batch_size=1)
+    quantiser.save(str(out))
+    tokenizer.save_pretrained(str(out))
+
+
+def quantise_gguf(model: str, out: Path, kind: str, llama_cpp: Path) -> None:
+    out.mkdir(parents=True, exist_ok=True)
+    f16 = out / "model-f16.gguf"
+    subprocess.run(
+        ["python", str(llama_cpp / "convert_hf_to_gguf.py"), model, "--outtype", "f16", "--outfile", str(f16)],
+        check=True,
+    )
+    binary = llama_cpp / "build" / "bin" / "llama-quantize"
+    if not binary.exists():
+        binary = llama_cpp / "llama-quantize"
+    subprocess.run([str(binary), str(f16), str(out / "model.gguf"), kind], check=True)
+    f16.unlink()
+    for name in ("tokenizer.json", "tokenizer_config.json", "config.json"):
+        source = Path(model) / name
+        if source.is_file():
+            shutil.copy2(source, out / name)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="compress the merged model to the format your envelope group takes")
+    parser.add_argument("--model", required=True, help="the merged model folder from train.py")
+    parser.add_argument("--out", required=True, help="the folder that becomes slm_repo/ weights")
+    parser.add_argument("--method", default="awq", choices=["awq", "gptq", "gguf"])
+    parser.add_argument("--bits", type=int, default=4)
+    parser.add_argument("--group-size", type=int, default=128)
+    parser.add_argument("--gguf-type", default="Q4_K_M", choices=["Q4_K_M", "Q5_K_S", "Q5_K_M", "Q6_K", "Q8_0"])
+    parser.add_argument("--llama-cpp", default="", help="path to a llama.cpp checkout with llama-quantize built")
+    parser.add_argument("--calibration", default="", help="pairs.jsonl from distill.py, used as calibration text")
+    parser.add_argument("--samples", type=int, default=256)
+    parser.add_argument("--seed", type=int, default=0)
+    args = parser.parse_args()
+
+    if args.bits < 4:
+        raise SystemExit("nothing below 4 bit is accepted in any envelope group")
+    out = Path(args.out)
+    samples: list[str] = []
+    calibration_sha = ""
+    if args.method in ("awq", "gptq"):
+        if not args.calibration:
+            raise SystemExit("awq and gptq need --calibration pairs.jsonl from your own training inputs")
+        samples = calibration(args.calibration, args.samples, args.seed)
+        calibration_sha = hashlib.sha256("\n".join(samples).encode("utf-8")).hexdigest()
+    if args.method == "awq":
+        quantise_awq(args.model, out, samples, args.bits, args.group_size)
+    elif args.method == "gptq":
+        quantise_gptq(args.model, out, samples, args.bits, args.group_size)
+    else:
+        if not args.llama_cpp:
+            raise SystemExit("gguf needs --llama-cpp pointing at a llama.cpp checkout")
+        quantise_gguf(args.model, out, args.gguf_type, Path(args.llama_cpp))
+
+    record: dict[str, Any] = {
+        "method": args.method,
+        "bits": args.bits if args.method != "gguf" else None,
+        "group_size": args.group_size if args.method != "gguf" else None,
+        "gguf_type": args.gguf_type if args.method == "gguf" else None,
+        "calibration_samples": len(samples),
+        "calibration_data_sha256": calibration_sha,
+        "source_model": args.model,
+    }
+    folder = Path(__file__).resolve().parents[1] / "compression"
+    folder.mkdir(parents=True, exist_ok=True)
+    (folder / "compression.json").write_text(json.dumps(record, indent=2) + "\n", encoding="utf-8")
+    before_after = folder / "before_after.json"
+    if not before_after.is_file():
+        before_after.write_text(
+            json.dumps({"before": {}, "after": {}, "note": "fill both from score_compress.py on the held out items"}, indent=2)
+            + "\n",
+            encoding="utf-8",
+        )
+    print(f"compressed weights at {out}; record at {folder / 'compression.json'}")
+
+
+if __name__ == "__main__":
+    main()

slm_repo/training/scripts/distill.py

@@ -0,0 +1,167 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import time
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+FENCE = re.compile(r"^```[A-Za-z]*\s*|\s*```$")
+
+
+def fetch_arena(server: str, arena: str) -> dict[str, Any]:
+    response = httpx.get(f"{server.rstrip('/')}/v2/arenas/{arena}", timeout=60.0)
+    response.raise_for_status()
+    found: dict[str, Any] = response.json()
+    return found
+
+
+def chat(base_url: str, key: str, model: str, messages: list[dict[str, str]], max_tokens: int) -> str:
+    for attempt in range(4):
+        try:
+            response = httpx.post(
+                f"{base_url.rstrip('/')}/chat/completions",
+                headers={"authorization": f"Bearer {key}"},
+                json={"model": model, "messages": messages, "temperature": 0, "max_tokens": max_tokens},
+                timeout=300.0,
+            )
+            response.raise_for_status()
+            data = response.json()
+            return str(((data.get("choices") or [{}])[0].get("message") or {}).get("content") or "")
+        except (httpx.HTTPError, ValueError):
+            time.sleep(2.0 * (attempt + 1))
+    return ""
+
+
+def render(template: str, task: dict[str, Any]) -> str:
+    text = template
+    for key, value in task.items():
+        if isinstance(value, str):
+            text = text.replace("{" + key + "}", value)
+    return text
+
+
+def grader_for(arena: dict[str, Any]) -> Any:
+    definition = arena.get("definition") or {}
+    name = str(definition.get("grader") or "")
+    if name == "context_compression":
+        from microtensor_subnet.validator import compress
+
+        grading = definition.get("grading") or {}
+        from tokenizers import Tokenizer
+
+        tokenizer = Tokenizer.from_pretrained(str(grading.get("tokenizer") or "Qwen/Qwen2.5-Coder-7B-Instruct"))
+        judge = compress.Judge(
+            count_tokens=lambda text: len(tokenizer.encode(text).ids),
+            answer=lambda summary, questions: ["" for _ in questions],
+            pass_threshold=float(grading.get("pass_threshold") or 0.5),
+            min_output_tokens=int(grading.get("min_output_tokens") or 300),
+            max_output_share=float(grading.get("max_output_share") or 0.1),
+        )
+
+        def grade(output: str, label: dict[str, Any], task: dict[str, Any]) -> tuple[bool, str]:
+            verdict = compress.grade(output, label, task, judge)
+            parts = dict(p.split(" ") for p in verdict.reason.split(", ")) if "recall" in verdict.reason else {}
+            if not parts:
+                return False, verdict.reason
+            ok = float(parts.get("recall", 0)) >= 0.99 and float(parts.get("invented", 1)) == 0.0
+            return ok, verdict.reason
+
+        return grade
+
+    from microtensor_subnet.validator import grade as grading_module
+
+    def grade(output: str, label: dict[str, Any], task: dict[str, Any]) -> tuple[bool, str]:
+        verdict = grading_module.grade(name, output, label, task)
+        return bool(verdict.success), str(verdict.reason)
+
+    return grade
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="write teacher answers for dev items and keep the ones the grader accepts")
+    parser.add_argument("--arena", required=True)
+    parser.add_argument("--items", required=True, help="items.jsonl built with the arena dev tooling, with task and label")
+    parser.add_argument("--out", required=True, help="jsonl of {prompt, completion} for train.py")
+    parser.add_argument("--teacher", required=True, help="teacher model id at your provider")
+    parser.add_argument("--base-url", default=os.environ.get("TEACHER_BASE_URL", "https://openrouter.ai/api/v1"))
+    parser.add_argument("--key-env", default="TEACHER_API_KEY")
+    parser.add_argument("--server", default="https://api.microtensor.cloud")
+    parser.add_argument("--max-tokens", type=int, default=0, help="default: the arena output budget")
+    parser.add_argument("--workers", type=int, default=8)
+    parser.add_argument("--limit", type=int, default=0)
+    parser.add_argument("--keep-rejected", default="", help="write grader rejected answers here for inspection")
+    args = parser.parse_args()
+
+    key = os.environ.get(args.key_env, "")
+    if not key:
+        raise SystemExit(f"set {args.key_env} to your provider key")
+    arena = fetch_arena(args.server, args.arena)
+    definition = arena.get("definition") or {}
+    template = str(definition.get("slm_prompt") or "")
+    if not template:
+        raise SystemExit(f"{args.arena} publishes no slm_prompt")
+    max_tokens = args.max_tokens or int((definition.get("generation") or {}).get("max_tokens") or 1024)
+    grade = grader_for(arena)
+
+    rows = [json.loads(line) for line in Path(args.items).read_text(encoding="utf-8").splitlines() if line.strip()]
+    items = [row for row in rows if not row.get("twin_of")]
+    if args.limit:
+        items = items[: args.limit]
+
+    def one(item: dict[str, Any]) -> dict[str, Any]:
+        task = item["task"]
+        messages = [{"role": "system", "content": str(task.get("system") or "")}]
+        messages.append({"role": "user", "content": str(task.get("user") or json.dumps(task, sort_keys=True))})
+        answer = FENCE.sub("", chat(args.base_url, key, args.teacher, messages, max_tokens).strip())
+        ok, reason = grade(answer, item.get("label") or {}, task)
+        return {
+            "item_id": item.get("item_id"),
+            "ok": ok,
+            "reason": reason,
+            "prompt": render(template, task),
+            "completion": answer,
+        }
+
+    with ThreadPoolExecutor(max_workers=args.workers) as pool:
+        results = list(pool.map(one, items))
+    kept = [r for r in results if r["ok"]]
+    out = Path(args.out)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text(
+        "".join(json.dumps({"prompt": r["prompt"], "completion": r["completion"]}) + "\n" for r in kept),
+        encoding="utf-8",
+    )
+    if args.keep_rejected:
+        Path(args.keep_rejected).write_text(
+            "".join(json.dumps(r) + "\n" for r in results if not r["ok"]), encoding="utf-8"
+        )
+    digest = hashlib.sha256(out.read_bytes()).hexdigest()
+    record = Path(__file__).resolve().parents[1] / "distillation" / "teacher.json"
+    record.write_text(
+        json.dumps(
+            {
+                "teacher_model": args.teacher,
+                "teacher_version": "",
+                "provider": args.base_url,
+                "items": len(results),
+                "kept": len(kept),
+                "pairs_sha256": digest,
+                "harness_used": "none, the arena prompt",
+            },
+            indent=2,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    print(f"{len(kept)} of {len(results)} teacher answers passed the grader; pairs at {out} (sha256 {digest[:12]})")
+
+
+if __name__ == "__main__":
+    main()

slm_repo/training/scripts/requirements.txt

@@ -0,0 +1,12 @@
+torch
+transformers>=4.55
+datasets
+accelerate
+peft>=0.15
+trl>=0.19
+bitsandbytes
+autoawq
+gptqmodel
+tokenizers
+httpx
+microtensor-subnet @ git+https://github.com/microtensor-io/microtensor-subnet.git

slm_repo/training/scripts/score_compress.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import Any
+
+import httpx
+from microtensor_subnet.validator import compress
+from microtensor_subnet.validator.sandbox.judge import PROMPT
+
+LINES = (0.5, 0.6, 0.7, 0.8, 0.9)
+TOKENIZER = "Qwen/Qwen2.5-Coder-7B-Instruct"
+STOP = ["<|im_end|>", "<|endoftext|>", "<|end|>"]
+
+
+def post(url: str, body: dict[str, Any]) -> dict[str, Any]:
+    response = httpx.post(url, json=body, timeout=900.0)
+    response.raise_for_status()
+    found: dict[str, Any] = response.json()
+    return found
+
+
+def complete(url: str, model: str, prompt: str, max_tokens: int) -> str:
+    data = post(
+        url.rstrip("/") + "/v1/completions",
+        {"model": model, "prompt": prompt, "temperature": 0, "max_tokens": max_tokens, "stop": STOP},
+    )
+    return str((data.get("choices") or [{}])[0].get("text") or "")
+
+
+def chat(url: str, model: str, messages: list[dict[str, str]], max_tokens: int) -> str:
+    data = post(
+        url.rstrip("/") + "/v1/chat/completions",
+        {"model": model, "messages": messages, "temperature": 0, "max_tokens": max_tokens},
+    )
+    return str(((data.get("choices") or [{}])[0].get("message") or {}).get("content") or "")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="score a model on ctx_compress items the way validators do")
+    parser.add_argument("--items", required=True, help="items.jsonl from mta dev")
+    parser.add_argument("--arena", default="ctx_compress")
+    parser.add_argument("--server", default="https://api.microtensor.cloud")
+    parser.add_argument("--slm-url", required=True, help="your model behind an OpenAI compatible server")
+    parser.add_argument("--slm-model", default="slm")
+    parser.add_argument("--judge-url", required=True, help="Qwen/Qwen2.5-Coder-7B-Instruct, the probe agent")
+    parser.add_argument("--judge-model", default="judge")
+    parser.add_argument("--limit", type=int, default=200)
+    parser.add_argument("--max-tokens", type=int, default=3072)
+    parser.add_argument("--workers", type=int, default=8)
+    parser.add_argument("--pass-line", type=float, default=0.5)
+    parser.add_argument("--out", default="", help="write one scored line per item here")
+    args = parser.parse_args()
+
+    from tokenizers import Tokenizer
+
+    tokenizer = Tokenizer.from_pretrained(TOKENIZER)
+    arena = httpx.get(f"{args.server.rstrip('/')}/v2/arenas/{args.arena}", timeout=60.0).json()
+    template = str((arena.get("definition") or {}).get("slm_prompt") or "")
+    if not template:
+        raise SystemExit(f"{args.arena} publishes no slm_prompt")
+
+    def answer(summary: str, questions: list[str]) -> list[str]:
+        return [
+            chat(
+                args.judge_url,
+                args.judge_model,
+                [
+                    {"role": "system", "content": PROMPT},
+                    {"role": "user", "content": f"<summary>\n{summary}\n</summary>\n\nQuestion: {q}"},
+                ],
+                64,
+            )
+            for q in questions
+        ]
+
+    judge = compress.Judge(
+        count_tokens=lambda text: len(tokenizer.encode(text).ids),
+        answer=answer,
+        pass_threshold=args.pass_line,
+    )
+    rows = [json.loads(line) for line in Path(args.items).read_text(encoding="utf-8").splitlines() if line.strip()]
+    items = [row for row in rows if not row.get("twin_of")][: args.limit]
+
+    def one(item: dict[str, Any]) -> dict[str, Any]:
+        task = item["task"]
+        prompt = template.replace("{system}", task["system"]).replace("{user}", task["user"])
+        output = complete(args.slm_url, args.slm_model, prompt, args.max_tokens)
+        verdict = compress.grade(output, item["label"], task, judge)
+        parts = dict(p.split(" ") for p in verdict.reason.split(", ")) if "recall" in verdict.reason else {}
+        return {"item_id": item["item_id"], "score": verdict.score, "reason": verdict.reason, "parts": parts, "output": output}
+
+    with ThreadPoolExecutor(max_workers=args.workers) as pool:
+        results = list(pool.map(one, items))
+    scores = [r["score"] for r in results]
+    print(f"items {len(scores)}, mean item score {statistics.mean(scores):.3f}")
+    for line in LINES:
+        print(f"pass at {line:.1f}: {sum(s >= line for s in scores) / len(scores):.1%}")
+    for key in ("recall", "invented", "continuation"):
+        values = [float(r["parts"][key]) for r in results if key in r["parts"]]
+        if values:
+            print(f"{key}: mean {statistics.mean(values):.3f}")
+    failed = [r["reason"] for r in results if not r["parts"]]
+    print(f"format or budget failures: {len(failed)}")
+    if args.out:
+        Path(args.out).write_text("".join(json.dumps(r) + "\n" for r in results), encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()