Add Wave 1 training pipeline (SFT/GRPO/eval/preflight/launch) + gitignore hardening

.gitignore

@@ -1,13 +1,34 @@
 # Documentation and research (not part of the submission)
 Docs/
-
-# OpenEnv preparatory course (dev reference only, not part of submission)
+DocsR2/
 openenv-course/
+playbook/
+
+# Secrets — NEVER commit
+.env
+*.env
+api.env
+hg.env
+
+# Training artefacts
+training/data/raw/
+training/.preflight_grpo/
+training/runs.jsonl
+outputs/
+out/
+wandb/
+trackio/
 
 # Python
 __pycache__/
 *.pyc
 *.pyo
+*.egg-info/
+.venv/
+venv/
 
-# Environment
-.env
+# OS / editor
+.DS_Store
+Thumbs.db
+.idea/
+.vscode/

training/README.md

@@ -0,0 +1,72 @@
+# SevZero — training (Round 2)
+
+One-liner per script:
+
+- **`train_sft.py`**: SFT on `Mist-ic/sevzero-expert-trajectories` with QLoRA (Unsloth or PEFT fallback) → push adapter with `HF_TOKEN`.
+- **`train_grpo.py`**: GRPO with `rollout_func` + remote env (`SEVZERO_ENV_URL`); vLLM colocate, Trackio `Mist-ic/sevzero-trackio`.
+- **`eval.py`**: Compare HF adapters and frontier models; write `eval_results.csv`, push `Mist-ic/sevzero-eval-results` with `HF_MAIN_TOKEN`.
+- **`preflight.py`**: In-process grader + tiny GRPO smoke (5 steps) on CPU; starts local uvicorn.
+- **`launch_hf_job.py`**: `huggingface_hub.run_job` wrapper; `--hardware l40sx1` (verify with `hf jobs hardware`).
+
+## Env files
+
+Load with `python-dotenv` (auto-tried in `config_utils`):
+
+- `hg.env` — `HF_TOKEN` (worker), `HF_MAIN_TOKEN` (Mist-ic, Trackio + eval dataset)
+- `api.env` — `GEMINI_API_KEY`, `AZURE_*` for `eval.py`
+
+| Variable | Role |
+|----------|------|
+| `HF_TOKEN` | Worker: train pushes, private adapter pulls |
+| `HF_MAIN_TOKEN` | `Mist-ic`: Trackio + `sevzero-eval-results` only |
+| `SEVZERO_ENV_URL` | HTTP base of SevZero Space/ server for GRPO + eval + preflight |
+| `GEMINI_API_KEY` | Direct Gemini in eval |
+| `AZURE_API_KEY` | Azure OpenAI + Azure AI Inference |
+| `AZURE_OPENAI_ENDPOINT` | Deployment base for gpt-5.4-pro |
+| `AZURE_AI_INFERENCE_ENDPOINT` | For grok / kimi / DeepSeek in eval |
+| `AZURE_API_VERSION` | OpenAI client version header if needed |
+| `GEMINI_EVAL_MODEL` | Optional override (default set in `eval.py`) |
+
+## Local debug (from repo root)
+
+```bash
+# Install (pin versions in comments / orchestrator)
+pip install -e ".[training]"
+
+# SFT
+python training/train_sft.py --output_dir ./out/sft --max_steps 10 --push_to_hub_repo "" --variant_name test
+
+# GRPO (remote env required)
+$env:SEVZERO_ENV_URL="https://<your-sevzero-space>.hf.space"
+python training/train_grpo.py --sft_adapter_repo YOUR/adapters --max_steps 5 --output_dir ./out/grpo
+```
+
+## Wave 3 — three GRPO variants (see `playbook/00-orchestration.md`)
+
+Primary (PhaseOfCode):
+
+```bash
+python training/train_grpo.py --sft_adapter_repo PhaseOfCode/sevzero-llama3-8b-sft --K 4 --lr 7e-6 --max_steps 350 --variant_name primary
+```
+
+Stability (NoahInOblivion):
+
+```bash
+python training/train_grpo.py --sft_adapter_repo NoahInOblivion/sevzero-llama3-8b-sft --K 8 --lr 5e-6 --max_steps 350 --variant_name stability
+```
+
+Innovation (NoxIsOblivion, env flags on):
+
+```bash
+python training/train_grpo.py --sft_adapter_repo NoxIsOblivion/sevzero-llama3-8b-sft --enable_schema_drift --enable_curriculum --K 4 --max_steps 350 --variant_name innovation
+```
+
+**HF Job (after merge + public git URL or bucket):**
+
+```bash
+$env:HF_TOKEN="<worker>"
+$env:SEVZERO_ENV_URL="https://....hf.space"
+python training/launch_hf_job.py --script grpo --variant_name primary -- --sft_adapter_repo YOUR/sevzero-llama3-8b-sft
+```
+
+**Dependency pins:** run `pip index versions trl openenv-core unsloth` and `python -c "import trl; print(trl.__version__)"` after install; pin in the orchestrator’s lock, not in this file.

training/__init__.py

@@ -0,0 +1 @@
+# Training / trajectory pipeline (Round 2)

training/build_dataset.py

@@ -0,0 +1,241 @@
+"""
+Build Llama-3.1-8B-Instruct SFT jsonl from raw trajectory jsonl (score ≥ 0.85).
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import random
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, Set, Tuple
+
+from dotenv import load_dotenv
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from inference import SYSTEM_PROMPT  # noqa: E402
+
+load_dotenv(REPO_ROOT / "api.env")
+load_dotenv(REPO_ROOT / "hg.env")
+
+DATA_DIR = REPO_ROOT / "training" / "data"
+RAW_GLOB = "raw/*.jsonl"
+OUT_TRAIN = DATA_DIR / "sft_train.jsonl"
+OUT_EVAL = DATA_DIR / "sft_eval.jsonl"
+OUT_STATS = DATA_DIR / "build_stats.json"
+
+MAX_OBS_TOKENS = 2048
+
+
+def _get_tokenizer():
+    import os
+
+    try:
+        from transformers import AutoTokenizer
+    except Exception:
+        return None
+    name = "meta-llama/Llama-3.1-8B-Instruct"
+    try:
+        tok = AutoTokenizer.from_pretrained(
+            name, token=os.environ.get("HF_MAIN_TOKEN")
+        )
+        return tok
+    except Exception:
+        try:
+            return AutoTokenizer.from_pretrained(
+                "hf-internal-testing/llama-tokenizer"
+            )
+        except Exception:
+            return None
+
+
+def _count_tokens(toker, text: str) -> int:
+    if toker is not None:
+        return len(toker.encode(text, add_special_tokens=False))
+    return max(1, len(text) // 4)
+
+
+def _shrink_observation(obs: Dict[str, Any], toker, max_toks: int) -> str:
+    """Serialize observation to JSON, shrink until user message fits max_toks (approximate)."""
+    o = {k: v for k, v in obs.items() if k not in ("reward",)}
+    order_drop = [
+        "metric_history",
+        "traces",
+        "logs",
+        "actions_taken",
+        "recent_deploys",
+    ]
+    for _ in range(40):
+        text = json.dumps(o, ensure_ascii=False, separators=(",", ":"), default=str)
+        tcount = _count_tokens(toker, text)
+        if tcount <= max_toks:
+            return text
+        shrunk = False
+        for k in order_drop:
+            if k in o and o[k]:
+                o[k] = None
+                if k == "actions_taken":
+                    o[k] = []
+                elif k in ("metric_history", "recent_deploys"):
+                    o[k] = []
+                shrunk = True
+                break
+        if shrunk:
+            continue
+        if "services" in o and isinstance(o["services"], list) and len(o["services"]) > 2:
+            o["services"] = o["services"][: max(1, len(o["services"]) - 1)]
+            continue
+        if "alerts" in o and isinstance(o["alerts"], list) and len(o["alerts"]) > 1:
+            o["alerts"] = o["alerts"][: max(0, len(o["alerts"]) - 1)]
+            continue
+        o["__truncated__"] = True
+        break
+    return json.dumps(o, ensure_ascii=False, separators=(",", ":"), default=str)
+
+
+def _episode_id(ep: Dict[str, Any]) -> str:
+    return f"{ep.get('model', '')}|{ep.get('task_id', '')}|{ep.get('seed', 0)}"
+
+
+def _assistant_action_json(action: Any) -> str:
+    if not isinstance(action, dict):
+        return json.dumps(
+            {"action_type": "noop", "params": {}}, ensure_ascii=False
+        )
+    a = {
+        "action_type": str(action.get("action_type", "noop")),
+        "params": action.get("params") or {},
+    }
+    return json.dumps(a, ensure_ascii=False)
+
+
+def _load_episodes_from_raw(raw_dir: Path) -> List[Dict[str, Any]]:
+    out: List[Dict[str, Any]] = []
+    for p in sorted(raw_dir.glob("*.jsonl")):
+        with p.open(encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if not line:
+                    continue
+                out.append(json.loads(line))
+    return out
+
+
+def build(
+    min_score: float = 0.85,
+) -> Dict[str, Any]:
+    toker = _get_tokenizer()
+    raw_dir = DATA_DIR / "raw"
+    episodes = _load_episodes_from_raw(raw_dir)
+    kept: List[Dict[str, Any]] = []
+    dropped: List[Dict[str, Any]] = []
+    for ep in episodes:
+        sc = float(ep.get("final_score", 0.0) or 0.0)
+        if sc >= min_score and ep.get("steps"):
+            kept.append(ep)
+        else:
+            dropped.append(ep)
+
+    eids = [_episode_id(e) for e in kept]
+    unique_eids = list(dict.fromkeys(eids))
+    n_ep = len(unique_eids)
+    rng = random.Random(42)
+    rng.shuffle(unique_eids)
+    if n_ep <= 1:
+        n_eval = 0
+    else:
+        n_eval = max(1, n_ep // 10)
+    eval_ids: Set[str] = set(unique_eids[:n_eval]) if n_eval else set()
+
+    train_rows: List[Dict[str, Any]] = []
+    eval_rows: List[Dict[str, Any]] = []
+    max_prompt_toks = 0
+
+    for ep in kept:
+        eid = _episode_id(ep)
+        is_eval = eid in eval_ids
+        for st in ep.get("steps", []):
+            obs = st.get("observation", {})
+            if not isinstance(obs, dict):
+                continue
+            user_str = _shrink_observation(obs, toker, MAX_OBS_TOKENS)
+            messages = [
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": user_str},
+                {
+                    "role": "assistant",
+                    "content": _assistant_action_json(st.get("action", {})),
+                },
+            ]
+            if toker is not None:
+                try:
+                    plen = len(
+                        toker.apply_chat_template(
+                            messages, tokenize=True, add_generation_prompt=False
+                        )
+                    )
+                except Exception:
+                    plen = _count_tokens(
+                        toker, SYSTEM_PROMPT + "\n" + user_str
+                    )
+            else:
+                plen = _count_tokens(
+                    None, SYSTEM_PROMPT + "\n" + user_str
+                )
+            max_prompt_toks = max(max_prompt_toks, plen)
+            row = {
+                "messages": messages,
+                "meta": {
+                    "episode_id": eid,
+                    "model": ep.get("model"),
+                    "task_id": ep.get("task_id"),
+                    "seed": ep.get("seed"),
+                    "step": st.get("step"),
+                    "episode_score": ep.get("final_score"),
+                },
+            }
+            if is_eval:
+                eval_rows.append(row)
+            else:
+                train_rows.append(row)
+
+    scores = [float(x.get("final_score", 0) or 0) for x in kept]
+    mean_sc = sum(scores) / len(scores) if scores else 0.0
+
+    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    with OUT_TRAIN.open("w", encoding="utf-8") as ft:
+        for r in train_rows:
+            ft.write(json.dumps(r, ensure_ascii=False) + "\n")
+    with OUT_EVAL.open("w", encoding="utf-8") as fe:
+        for r in eval_rows:
+            fe.write(json.dumps(r, ensure_ascii=False) + "\n")
+
+    stats: Dict[str, Any] = {
+        "episodes_total_seen": len(episodes),
+        "episodes_kept": len(kept),
+        "episodes_dropped": len(dropped),
+        "mean_episode_score_kept": round(mean_sc, 6),
+        "train_rows": len(train_rows),
+        "eval_rows": len(eval_rows),
+        "max_prompt_token_length": max_prompt_toks,
+        "max_observation_user_token_budget": MAX_OBS_TOKENS,
+        "min_score_filter": min_score,
+    }
+    with OUT_STATS.open("w", encoding="utf-8") as f:
+        json.dump(stats, f, indent=2)
+    print(json.dumps(stats, indent=2), flush=True)
+    return stats
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--min-score", type=float, default=0.85)
+    args = ap.parse_args()
+    build(min_score=args.min_score)
+
+
+if __name__ == "__main__":
+    main()

training/collect_trajectories.py

@@ -0,0 +1,764 @@
+"""
+Collect expert trajectories for SevZero SFT (Round 2).
+
+Loads API keys from api.env and hg.env (gitignored). Does not log secrets.
+"""
+from __future__ import annotations
+
+import argparse
+import copy
+import difflib
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+import httpx
+from dotenv import load_dotenv
+from openai import AzureOpenAI
+from pydantic import BaseModel, Field
+
+# Repo root: parent of training/
+REPO_ROOT = Path(__file__).resolve().parent.parent
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from inference import (  # noqa: E402
+    build_observation_prompt,
+    parse_action,
+)
+from inference import SYSTEM_PROMPT as _BASE_SYSTEM  # noqa: E402
+
+load_dotenv(REPO_ROOT / "api.env")
+load_dotenv(REPO_ROOT / "hg.env")
+
+# ---------------------------------------------------------------------------
+# Config matrix (must match spec)
+# ---------------------------------------------------------------------------
+
+GEMINI_SEEDS = [
+    42, 123, 7, 11, 23, 31, 47, 59, 67, 71, 83, 89, 97, 101, 109, 113, 127, 131, 137, 149
+]
+GPT_SEEDS = [
+    42, 123, 7, 13, 17, 19, 29, 37, 41, 43, 53, 61, 73, 79, 83, 89, 97, 101, 103, 107
+]
+GROK_EXTRA_SEEDS = [13, 17, 19, 29, 37, 41, 43, 53, 61, 73]
+
+# Combined pool for grok / kimi / deepseek (any from grok list + full Gemini list)
+GROK_KIMI_POOL: List[int] = sorted(set(GEMINI_SEEDS) | set(GROK_EXTRA_SEEDS))
+
+MODEL_GEMINI = "gemini-3.1-pro-preview"
+MODEL_GPT = "gpt-5.4-pro"
+MODEL_GROK = "grok-4.20-reasoning"
+MODEL_KIMI = "kimi-k2.6"
+MODEL_DEEPSEEK = "DeepSeek-V3.2"
+ALL_CANON = {MODEL_GEMINI, MODEL_GPT, MODEL_GROK, MODEL_KIMI, MODEL_DEEPSEEK}
+
+
+def _split_seeds(
+    pool: List[int], counts: Tuple[int, int, int], offset: int
+) -> List[Tuple[str, int]]:
+    """Return list of (task_id, seed) in order easy, medium, hard."""
+    c_e, c_m, c_h = counts
+    n = len(pool)
+    if n == 0:
+        return []
+    o = [pool[(i + offset) % n] for i in range(n)]
+    out: List[Tuple[str, int]] = []
+    i = 0
+    for _ in range(c_e):
+        out.append(("easy", o[i % len(o)]))
+        i += 1
+    for _ in range(c_m):
+        out.append(("medium", o[i % len(o)]))
+        i += 1
+    for _ in range(c_h):
+        out.append(("hard", o[i % len(o)]))
+        i += 1
+    return out
+
+
+def plan_gemini(c_e: int, c_m: int, c_h: int) -> List[Tuple[str, str, int]]:
+    return [
+        (MODEL_GEMINI, t, s)
+        for t, s in _split_seeds(GEMINI_SEEDS, (c_e, c_m, c_h), offset=0)
+    ]
+
+
+def plan_gpt(c_e: int, c_m: int, c_h: int) -> List[Tuple[str, str, int]]:
+    return [
+        (MODEL_GPT, t, s)
+        for t, s in _split_seeds(GPT_SEEDS, (c_e, c_m, c_h), offset=0)
+    ]
+
+
+def plan_grok(c_e: int, c_m: int, c_h: int) -> List[Tuple[str, str, int]]:
+    return [
+        (MODEL_GROK, t, s)
+        for t, s in _split_seeds(GROK_KIMI_POOL, (c_e, c_m, c_h), offset=0)
+    ]
+
+
+def plan_kimi(c_e: int, c_m: int, c_h: int) -> List[Tuple[str, str, int]]:
+    return [
+        (MODEL_KIMI, t, s)
+        for t, s in _split_seeds(GROK_KIMI_POOL, (c_e, c_m, c_h), offset=7)
+    ]
+
+
+def plan_deepseek(c_e: int, c_m: int, c_h: int) -> List[Tuple[str, str, int]]:
+    return [
+        (MODEL_DEEPSEEK, t, s)
+        for t, s in _split_seeds(GROK_KIMI_POOL, (c_e, c_m, c_h), offset=3)
+    ]
+
+
+def full_plan(c_e: int, c_m: int, c_h: int) -> List[Tuple[str, str, int]]:
+    return (
+        plan_gemini(c_e, c_m, c_h)
+        + plan_gpt(c_e, c_m, c_h)
+        + plan_grok(c_e, c_m, c_h)
+        + plan_kimi(c_e, c_m, c_h)
+        + plan_deepseek(c_e, c_m, c_h)
+    )
+
+
+# Rough USD cost tracking (tunable; for guardrail only)
+@dataclass
+class CostTracker:
+    usd: float = 0.0
+    budget: float = 5.0
+    by_model: Dict[str, float] = field(default_factory=dict)
+    per_model_max: float = 2.0
+
+    def add(self, model: str, usd: float) -> None:
+        self.usd += usd
+        self.by_model[model] = self.by_model.get(model, 0.0) + usd
+        m = self.by_model[model]
+        cap = self.per_model_max
+        if m > cap:
+            raise RuntimeError(
+                f"Model {model} exceeded ${cap:.2f} in estimated spend (${m:.2f}); stopping per cap."
+            )
+        if self.usd > self.budget:
+            raise RuntimeError(
+                f"Total estimated API spend ${self.usd:.2f} exceeded budget ${self.budget:.2f}."
+            )
+
+
+def _estimate_openai_style_cost(
+    model: str, prompt_tokens: int, completion_tokens: int
+) -> float:
+    # Conservative blended rate per 1K tokens (USD) — for guardrails only
+    if "gemini" in model:
+        p, c = 0.00125, 0.01
+    elif "gpt" in model.lower() or "5.4" in model:
+        p, c = 0.0025, 0.01
+    else:
+        p, c = 0.001, 0.006
+    return (prompt_tokens * p + completion_tokens * c) / 1000.0
+
+
+# ---------------------------------------------------------------------------
+# Pydantic for Gemini structured action JSON
+# ---------------------------------------------------------------------------
+
+
+class AgentActionOut(BaseModel):
+    action_type: str
+    params: Dict[str, Any] = Field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# Azure deployment self-heal
+# ---------------------------------------------------------------------------
+
+
+def _is_not_found(err: str) -> bool:
+    s = (err or "").lower()
+    return "deploymentnotfound" in s or "deployment" in s and "not found" in s
+
+
+def list_azure_openai_deployments() -> List[str]:
+    key = os.environ.get("AZURE_API_KEY", "")
+    ep = (os.environ.get("AZURE_OPENAI_ENDPOINT", "") or "").rstrip("/")
+    ver = os.environ.get("AZURE_API_VERSION", "2024-12-01-preview")
+    if not key or not ep:
+        return []
+    url = f"{ep}/openai/deployments?api-version={ver}"
+    try:
+        r = httpx.get(url, headers={"api-key": key}, timeout=30.0)
+        r.raise_for_status()
+        data = r.json()
+        return [d.get("id", "") for d in data.get("value", []) if d.get("id")]
+    except Exception:
+        return []
+
+
+def list_foundry_deployments() -> List[str]:
+    """
+    Best-effort: project endpoint may expose deployments; schema varies.
+    """
+    fe = (os.environ.get("AZURE_FOUNDRY_PROJECT_ENDPOINT", "") or "").rstrip("/")
+    key = os.environ.get("AZURE_API_KEY", "")
+    if not fe or not key:
+        return []
+    for suffix in ("/deployments", "/openai/models"):
+        try:
+            url = f"{fe}{suffix}"
+            r = httpx.get(
+                url, headers={"api-key": key}, params={"api-version": "2024-12-01-preview"}, timeout=30.0
+            )
+            if r.status_code != 200:
+                continue
+            data = r.json()
+            if isinstance(data, list):
+                return [str(x.get("id", x)) for x in data if isinstance(x, dict)]
+            if "value" in data:
+                return [d.get("id", "") for d in data.get("value", []) if d.get("id")]
+        except Exception:
+            continue
+    return []
+
+
+def pick_closest(name: str, options: List[str]) -> str:
+    if not options:
+        return name
+    if name in options:
+        return name
+    ranked = difflib.get_close_matches(name, options, n=1, cutoff=0.2)
+    if ranked:
+        return ranked[0]
+    return options[0]
+
+
+# ---------------------------------------------------------------------------
+# LLM backends
+# ---------------------------------------------------------------------------
+
+
+class LLMClient:
+    def __init__(self, model: str) -> None:
+        self.model = model
+        self.gemini_client: Any = None
+        self.azure_openai: Any = None
+        self.azure_inf: Any = None
+        if model == MODEL_GEMINI:
+            from google import genai
+
+            key = os.environ.get("GEMINI_API_KEY", "")
+            if not key:
+                raise ValueError("GEMINI_API_KEY missing for Gemini collection.")
+            self.gemini_client = genai.Client(api_key=key)
+        elif model == MODEL_GPT:
+            if not all(
+                os.environ.get(x)
+                for x in (
+                    "AZURE_API_KEY",
+                    "AZURE_OPENAI_ENDPOINT",
+                    "AZURE_API_VERSION",
+                )
+            ):
+                raise ValueError("AZURE_API_KEY, AZURE_OPENAI_ENDPOINT, AZURE_API_VERSION required for gpt-5.4-pro.")
+            self.azure_openai = AzureOpenAI(
+                api_key=os.environ["AZURE_API_KEY"],
+                azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
+                api_version=os.environ["AZURE_API_VERSION"],
+            )
+        else:
+            if not all(os.environ.get(x) for x in ("AZURE_API_KEY", "AZURE_AI_INFERENCE_ENDPOINT")):
+                raise ValueError("AZURE_API_KEY and AZURE_AI_INFERENCE_ENDPOINT required for inference models.")
+            from azure.ai.inference import ChatCompletionsClient
+            from azure.core.credentials import AzureKeyCredential
+
+            self.azure_inf = ChatCompletionsClient(
+                endpoint=os.environ["AZURE_AI_INFERENCE_ENDPOINT"],
+                credential=AzureKeyCredential(os.environ["AZURE_API_KEY"]),
+            )
+
+    def _deployment_name(self) -> str:
+        m = {MODEL_GPT: "AZURE_MODEL_GPT", MODEL_GROK: "AZURE_MODEL_GROK", MODEL_KIMI: "AZURE_MODEL_KIMI", MODEL_DEEPSEEK: "AZURE_MODEL_DEEPSEEK"}.get(self.model)
+        if m:
+            v = os.environ.get(m, "").strip()
+            if v:
+                return v
+        return self.model
+
+    def call(
+        self,
+        messages: List[Dict[str, str]],
+    ) -> Tuple[str, int, int]:
+        """Return (raw_text, prompt_tokens, completion_tokens)."""
+        p_tok, c_tok = 0, 0
+        if self.gemini_client is not None:
+            return self._call_gemini(messages, p_tok, c_tok)
+        if self.azure_openai is not None:
+            return self._call_azure_openai(messages, p_tok, c_tok)
+        if self.azure_inf is not None:
+            return self._call_azure_inference(messages, p_tok, c_tok)
+        raise RuntimeError("No backend initialised")
+
+    def _call_gemini(
+        self, messages: List[Dict[str, str]], p0: int, c0: int
+    ) -> Tuple[str, int, int]:
+        from google.genai import types
+
+        if not messages:
+            return '{"action_type": "noop", "params": {}}', 0, 0
+        system = messages[0]["content"] if messages[0]["role"] == "system" else _BASE_SYSTEM
+        rest = messages[1:] if messages[0]["role"] == "system" else messages
+        name = os.environ.get("GEMINI_MODEL_PRO", MODEL_GEMINI)
+        config = types.GenerateContentConfig(
+            system_instruction=system,
+            response_mime_type="application/json",
+            response_json_schema=AgentActionOut,
+            temperature=0.0,
+            max_output_tokens=512,
+        )
+        # Build contents: alternating user / model for few-shot tail
+        contents: List[Any] = []
+        for m in rest:
+            if m["role"] == "user":
+                contents.append(
+                    types.Content(role="user", parts=[types.Part.from_text(text=m["content"])])
+                )
+            else:
+                contents.append(
+                    types.Content(
+                        role="model",
+                        parts=[types.Part.from_text(text=m["content"])],
+                    )
+                )
+        for attempt in range(3):
+            try:
+                resp = self.gemini_client.models.generate_content(
+                    model=name, contents=contents, config=config
+                )
+                text = (resp.text or "").strip() if hasattr(resp, "text") else ""
+                u = getattr(resp, "usage_metadata", None) or getattr(resp, "usage", None)
+                pt = int(getattr(u, "prompt_token_count", None) or getattr(u, "prompt_tokens", 0) or 0) if u else 0
+                ct = int(getattr(u, "candidates_token_count", None) or getattr(u, "completion_tokens", 0) or 0) if u else 0
+                if not text and hasattr(resp, "candidates") and resp.candidates:
+                    p0x = resp.candidates[0].content.parts[0] if resp.candidates[0].content.parts else None
+                    text = getattr(p0x, "text", "") or ""
+                return text, pt, ct
+            except Exception:
+                if attempt < 2:
+                    time.sleep(1.0 + attempt)
+                else:
+                    return '{"action_type": "noop", "params": {}}', p0, c0
+
+    def _call_azure_openai(
+        self, messages: List[Dict[str, str]], p0: int, c0: int
+    ) -> Tuple[str, int, int]:
+        dep = self._deployment_name()
+        for attempt in range(3):
+            try:
+                comp = self.azure_openai.chat.completions.create(
+                    model=dep,
+                    messages=messages,  # type: ignore[arg-type]
+                    temperature=0.0,
+                    max_tokens=512,
+                    timeout=90.0,
+                )
+                text = (comp.choices[0].message.content or "").strip()
+                u = comp.usage
+                pt = u.prompt_tokens if u else 0
+                ct = u.completion_tokens if u else 0
+                return text, pt, ct
+            except Exception as e:
+                err = str(e)
+                if _is_not_found(err):
+                    names = list_azure_openai_deployments()
+                    if names:
+                        dep = pick_closest(dep, names)
+                if attempt == 2:
+                    return '{"action_type": "noop", "params": {}}', p0, c0
+                time.sleep(1.0 + attempt)
+        return '{"action_type": "noop", "params": {}}', p0, c0
+
+    def _call_azure_inference(
+        self, messages: List[Dict[str, str]], p0: int, c0: int
+    ) -> Tuple[str, int, int]:
+        dep = self._deployment_name()
+        for attempt in range(3):
+            try:
+                resp = self.azure_inf.complete(
+                    model=dep,
+                    messages=messages,  # type: ignore[arg-type]
+                    temperature=0.0,
+                    max_tokens=512,
+                )
+                ch = resp.choices[0].message
+                text = (ch.content or "").strip() if ch else ""
+                u = getattr(resp, "usage", None)
+                pt = int(getattr(u, "prompt_tokens", 0) or 0) if u else 0
+                ct = int(getattr(u, "completion_tokens", 0) or 0) if u else 0
+                return text, pt, ct
+            except Exception as e:
+                err = str(e)
+                if _is_not_found(err) or "404" in err or "not found" in err.lower():
+                    names = [n for n in list_foundry_deployments() + list_azure_openai_deployments() if n]
+                    if names:
+                        dep = pick_closest(dep, names)
+                if attempt == 2:
+                    return '{"action_type": "noop", "params": {}}', p0, c0
+                time.sleep(1.0 + attempt)
+        return '{"action_type": "noop", "params": {}}', p0, c0
+
+
+# ---------------------------------------------------------------------------
+# Episode (mirrors inference.run_episode; logs full trace)
+# ---------------------------------------------------------------------------
+
+
+def _memory_block(tried_actions: Dict[str, List[str]], resolved_services: List[str]) -> str:
+    if not tried_actions and not resolved_services:
+        return ""
+    lines = ["## Episode Memory (do not repeat failed approaches)"]
+    if resolved_services:
+        lines.append(f"  Resolved: {', '.join(resolved_services)}")
+    for act, targets in tried_actions.items():
+        lines.append(f"  {act}: {'; '.join(targets)}")
+    return "\n".join(lines)
+
+
+def run_one_episode(
+    llm: LLMClient,
+    model_id: str,
+    base: str,
+    task_id: str,
+    seed: int,
+    cost: CostTracker,
+) -> Dict[str, Any]:
+    grade: Dict[str, Any] = {}
+    with httpx.Client(timeout=60.0) as http:
+        r = http.post(
+            f"{base}/reset", json={"seed": seed, "task_id": task_id}
+        )
+        r.raise_for_status()
+        resp_data = r.json()
+        obs: Dict[str, Any] = dict(resp_data.get("observation", resp_data))
+        max_steps = int(obs.get("max_steps", 10))
+        done = bool(resp_data.get("done", False))
+        conv: List[Dict[str, Any]] = []
+        tried: Dict[str, List[str]] = {}
+        resolved: List[str] = []
+        steps_out: List[Dict[str, Any]] = []
+        for step_num in range(1, max_steps + 1):
+            if done:
+                break
+            obs_pre = copy.deepcopy(obs)
+            user_msg = build_observation_prompt(obs_pre)
+            conv.append({"role": "user", "content": user_msg})
+            trimmed = conv[-6:]
+            memory = _memory_block(tried, resolved)
+            system_content = _BASE_SYSTEM + ("\n\n" + memory if memory else "")
+            messages: List[Dict[str, str]] = (
+                [{"role": "system", "content": system_content}] + trimmed
+            )
+            raw, pt, ct = llm.call(messages)
+            cost.add(
+                model_id, _estimate_openai_style_cost(model_id, pt, ct)
+            )
+            try:
+                action = parse_action(raw)
+            except Exception:
+                action = {"action_type": "noop", "params": {}}
+            if isinstance(action, dict) and "action_type" in action and model_id == MODEL_GEMINI:
+                try:
+                    a2 = (
+                        json.loads(raw[raw.find("{") : raw.rfind("}") + 1])
+                        if "{" in raw
+                        else None
+                    )
+                    if a2 and isinstance(a2, dict) and "action_type" in a2:
+                        action = a2
+                except Exception:
+                    pass
+            act_params = action.get("params", {}) or {}
+            if "replicas" in act_params:
+                try:
+                    act_params["replicas"] = int(act_params["replicas"])
+                except (ValueError, TypeError):
+                    act_params["replicas"] = 2
+            act_type = action.get("action_type", "noop")
+            target = act_params.get("service_id") or act_params.get("cache_name") or act_params.get("from_region") or ""
+            step_resp = http.post(
+                f"{base}/step",
+                json={"action": {"action_type": act_type, "params": act_params}},
+            )
+            sdata = step_resp.json() if step_resp.status_code == 200 else {}
+            obs = dict(sdata.get("observation", sdata))
+            done = bool(sdata.get("done", False))
+            reward = float(
+                obs.get("reward", sdata.get("reward", 0.0)) or 0.0
+            )
+            conv.append({"role": "assistant", "content": raw})
+            if act_type not in (
+                "inspect_logs",
+                "inspect_metrics",
+                "inspect_traces",
+                "noop",
+            ) and target:
+                new_slo = obs.get("global_slo_score", 0.0)
+                for svc in obs.get("services", []):
+                    if svc.get("id") == target and svc.get("status") == "healthy":
+                        if target not in resolved:
+                            resolved.append(target)
+                entry = f"{target} (slo={new_slo:.0%})"
+                tried.setdefault(str(act_type), [])
+                if entry not in tried[str(act_type)]:
+                    tried[str(act_type)].append(entry)
+            obs_ser = json.loads(
+                json.dumps(
+                    {k: v for k, v in obs_pre.items() if k != "reward"},
+                    default=str,
+                )
+            )
+            steps_out.append(
+                {
+                    "step": step_num,
+                    "observation": obs_ser,
+                    "prompt": user_msg,
+                    "messages": messages,
+                    "completion": raw,
+                    "action": action,
+                    "reward": reward,
+                    "info": {k: v for k, v in sdata.items() if k not in ("observation",)},
+                }
+            )
+        try:
+            final_state = http.get(f"{base}/state").json()
+        except Exception:
+            final_state = {}
+        try:
+            grade = http.post(
+                f"{base}/grader",
+                json={
+                    "final_slo_score": final_state.get("global_slo_score", 0.0),
+                    "steps_taken": final_state.get("step_count", 0),
+                    "max_steps": max_steps,
+                    "actions_taken": obs.get("actions_taken", []),
+                    "terminated": final_state.get("terminated", True),
+                    "termination_reason": final_state.get("termination_reason"),
+                },
+            ).json()
+        except Exception:
+            grade = {}
+    score = float(grade.get("score", 0.0) or 0.0)
+    return {
+        "model": model_id,
+        "task_id": task_id,
+        "seed": seed,
+        "steps": steps_out,
+        "grader": grade,
+        "final_score": score,
+        "max_steps": max_steps,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+
+def _raw_path(model: str) -> Path:
+    safe = re.sub(r"[^a-zA-Z0-9._-]+", "_", model)
+    d = REPO_ROOT / "training" / "data" / "raw"
+    d.mkdir(parents=True, exist_ok=True)
+    return d / f"{safe}.jsonl"
+
+
+def _wait_health(base: str, timeout: float = 45.0) -> None:
+    t0 = time.time()
+    while time.time() - t0 < timeout:
+        try:
+            r = httpx.get(f"{base}/health", timeout=3.0)
+            if r.status_code == 200:
+                return
+        except Exception:
+            pass
+        time.sleep(1.0)
+    print(f"[collect] health check timeout for {base} — continuing", flush=True)
+
+
+def start_server(port: int) -> subprocess.Popen:
+    env = os.environ.copy()
+    pp = str(REPO_ROOT)
+    env["PYTHONPATH"] = pp if not env.get("PYTHONPATH") else pp + os.pathsep + env["PYTHONPATH"]
+    return subprocess.Popen(
+        [sys.executable, "-m", "uvicorn", "server.app:app", "--host", "127.0.0.1", "--port", str(port)],
+        cwd=REPO_ROOT,
+        env=env,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.STDOUT,
+    )
+
+
+def parse_models(s: str) -> List[str]:
+    return [m.strip() for m in s.split(",") if m.strip()]
+
+
+def _plan_for_model(
+    model: str, c_e: int, c_m: int, c_h: int
+) -> List[Tuple[str, str, int]]:
+    p = {
+        MODEL_GEMINI: plan_gemini,
+        MODEL_GPT: plan_gpt,
+        MODEL_GROK: plan_grok,
+        MODEL_KIMI: plan_kimi,
+        MODEL_DEEPSEEK: plan_deepseek,
+    }
+    fn = p.get(model)
+    if not fn:
+        return []
+    return fn(c_e, c_m, c_h)
+
+
+def sanity_runs() -> List[Tuple[str, str, int]]:
+    return [
+        (MODEL_GEMINI, "easy", 42),
+        (MODEL_GPT, "easy", 42),
+        (MODEL_GROK, "easy", 13),
+    ]
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument(
+        "--models",
+        type=str,
+        default=",".join(sorted(ALL_CANON)),
+        help="Comma-separated model ids (default: all)",
+    )
+    ap.add_argument("--port", type=int, default=7860)
+    ap.add_argument("--no-start-server", action="store_true")
+    ap.add_argument("--sanity-only", action="store_true", help="Run only 3 smoke episodes (gemini, gpt, grok easy).")
+    ap.add_argument("--no-sanity", action="store_true", help="Skip pre-flight sanity runs.")
+    ap.add_argument(
+        "--budget-usd",
+        type=float,
+        default=5.0,
+        help="Total estimated-spend cap (heuristic) across all models.",
+    )
+    ap.add_argument(
+        "--per-model-budget-usd",
+        type=float,
+        default=0.0,
+        help="Per-model cap (0 = auto: max(2, budget/num selected models)).",
+    )
+    ap.add_argument(
+        "--episodes-easy",
+        type=int,
+        default=15,
+        help="Number of easy-task episodes per model (default 15, Wave 1.5).",
+    )
+    ap.add_argument(
+        "--episodes-medium",
+        type=int,
+        default=15,
+        help="Number of medium-task episodes per model (default 15).",
+    )
+    ap.add_argument(
+        "--episodes-hard",
+        type=int,
+        default=20,
+        help="Number of hard-task episodes per model (default 20).",
+    )
+    args = ap.parse_args()
+    want = set(parse_models(args.models))
+    bad = want - ALL_CANON
+    if bad:
+        raise SystemExit(f"Unknown model(s): {bad}. Valid: {sorted(ALL_CANON)}")
+
+    c_e, c_m, c_h = args.episodes_easy, args.episodes_medium, args.episodes_hard
+    if min(c_e, c_m, c_h) < 0:
+        raise SystemExit("--episodes-* must be non-negative.")
+    if c_e + c_m + c_h == 0:
+        raise SystemExit("At least one of --episodes-easy/medium/hard must be > 0.")
+
+    _ = full_plan(c_e, c_m, c_h)  # exercise planner (raises if misconfigured)
+
+    # Required keys
+    for m in want:
+        if m == MODEL_GEMINI and not os.environ.get("GEMINI_API_KEY"):
+            raise SystemExit("GEMINI_API_KEY missing (needed for gemini-3.1-pro-preview).")
+        if m == MODEL_GPT and not all(
+            os.environ.get(x) for x in ("AZURE_API_KEY", "AZURE_OPENAI_ENDPOINT", "AZURE_API_VERSION")
+        ):
+            raise SystemExit("Azure OpenAI env vars missing for gpt-5.4-pro.")
+        if m in (MODEL_GROK, MODEL_KIMI, MODEL_DEEPSEEK) and not all(
+            os.environ.get(x) for x in ("AZURE_API_KEY", "AZURE_AI_INFERENCE_ENDPOINT")
+        ):
+            raise SystemExit("Azure inference env missing for " + m)
+
+    proc: Optional[subprocess.Popen] = None
+    if not args.no_start_server:
+        proc = start_server(args.port)
+    base = f"http://127.0.0.1:{args.port}"
+    _wait_health(base)
+    n_m = max(1, len(want))
+    per_cap = args.per_model_budget_usd
+    if per_cap <= 0.0:
+        per_cap = max(2.0, args.budget_usd / n_m)
+    cost = CostTracker(budget=args.budget_usd, per_model_max=per_cap)
+    # LLM clients (lazy)
+    _clients: Dict[str, LLMClient] = {}
+    def get_llm(mid: str) -> LLMClient:
+        if mid not in _clients:
+            _clients[mid] = LLMClient(mid)
+        return _clients[mid]
+
+    try:
+        already: Set[Tuple[str, str, int]] = set()
+        if args.sanity_only:
+            final_list = [r for r in sanity_runs() if r[0] in want]
+        else:
+            if not args.no_sanity:
+                for mid, task_id, seed in (r for r in sanity_runs() if r[0] in want):
+                    print(f"[sanity] {mid} {task_id} seed={seed}", flush=True)
+                    llm = get_llm(mid)
+                    _ = run_one_episode(llm, mid, base, task_id, seed, cost)
+                    already.add((mid, task_id, seed))
+                print("[sanity] pre-flight ok", flush=True)
+            final_list = []
+            for m in want:
+                for x in _plan_for_model(m, c_e, c_m, c_h):
+                    if x in already:
+                        continue
+                    final_list.append(x)
+        n_done = 0
+        for mid, task_id, seed in final_list:
+            print(f"[episode] {mid} {task_id} seed={seed}", flush=True)
+            try:
+                llm = get_llm(mid)
+                ep = run_one_episode(llm, mid, base, task_id, seed, cost)
+            except RuntimeError as e:
+                print(f"[collect] Stopped: {e}", flush=True)
+                break
+            p = _raw_path(mid)
+            with p.open("a", encoding="utf-8") as f:
+                f.write(json.dumps(ep, ensure_ascii=False) + "\n")
+            n_done += 1
+            print(
+                f"  -> score={ep.get('final_score', 0):.4f} lines->{p.name} (total est ${cost.usd:.2f})",
+                flush=True,
+            )
+        print(f"Done. Episodes written: {n_done}. Estimated spend: ${cost.usd:.2f}", flush=True)
+    finally:
+        if proc is not None:
+            proc.terminate()
+            try:
+                proc.wait(timeout=5)
+            except Exception:
+                proc.kill()
+
+
+if __name__ == "__main__":
+    main()

training/config_utils.py

@@ -0,0 +1,32 @@
+"""Load dotenv from repo api.env + hg.env (optional). Does not read secrets into logs."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+
+
+def try_load_env_files() -> None:
+    for name in ("api.env", "hg.env"):
+        p = _REPO_ROOT / name
+        if not p.is_file():
+            continue
+        try:
+            from dotenv import load_dotenv
+
+            load_dotenv(p, override=False)
+        except ImportError:
+            _manual_load(p)
+
+
+def _manual_load(path: Path) -> None:
+    for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        k, v = line.split("=", 1)
+        k, v = k.strip(), v.strip().strip('"').strip("'")
+        if k and k not in os.environ:
+            os.environ[k] = v

training/data/DATASET_README_HF.md

@@ -0,0 +1,35 @@
+# SevZero expert trajectories (SFT)
+
+## Sources
+
+- Synthetic expert rollouts from frontier models (Gemini 3.1 Pro, Azure OpenAI, Azure AI Inference)
+  against the local OpenEnv `server.app` SevZero environment.
+
+## Filtering
+
+- Episodes with final grader `score` **≥** `0.75` are included.
+
+## Schema
+
+- Each example has a `messages` list (Llama-3.1-8B-Instruct–style SFT) and `meta` (episode / step provenance):
+  - `system`: SRE on-call system prompt (same as `inference.SYSTEM_PROMPT` in the repo)
+  - `user`: JSON-serialized observation (shrink to ≤ 2048 tokens for the user part)
+  - `assistant`: one JSON object `{"action_type": "...", "params": {...}}`
+
+## Stats (from `build_stats.json` at publish time)
+
+{
+  "episodes_total_seen": 90,
+  "episodes_kept": 42,
+  "episodes_dropped": 48,
+  "mean_episode_score_kept": 0.836021,
+  "train_rows": 853,
+  "eval_rows": 80,
+  "max_prompt_token_length": 2,
+  "max_observation_user_token_budget": 2048,
+  "min_score_filter": 0.75
+}
+
+## Parquet
+
+- Splits `train` and `eval` are also pushed in Parquet for fast `datasets.load_dataset`.

training/data/HANDOFF.md

@@ -0,0 +1,5 @@
+- **Dataset URL (after `python -m training.push_dataset`):** https://huggingface.co/datasets/Mist-ic/sevzero-expert-trajectories
+- **Rows:** see `build_stats.json` for `train_rows` and `eval_rows` after you run `build_dataset.py` on real raw JSONL.
+- **Max prompt tokens:** see `max_prompt_token_length` in `build_stats.json` — set SFT/GRPO `max_seq_length` to this + `max_completion_length` (e.g. +1024).
+- **Mean episode score:** `mean_episode_score_kept` in `build_stats.json` (episodes with final grader ≥ 0.85).
+- **Caveats:** run `collect_trajectories.py` with working `api.env`/`hg.env`; use `--no-sanity` to skip the 3 pre-flight API calls; install extras (`python-dotenv`, `google-genai`, `azure-ai-inference`, `huggingface_hub`, `datasets`, `transformers`, `pydantic`) as needed — `pyproject.toml` is unchanged.

training/data/build_stats.json

@@ -0,0 +1,11 @@
+{
+  "episodes_total_seen": 90,
+  "episodes_kept": 42,
+  "episodes_dropped": 48,
+  "mean_episode_score_kept": 0.836021,
+  "train_rows": 853,
+  "eval_rows": 80,
+  "max_prompt_token_length": 2,
+  "max_observation_user_token_budget": 2048,
+  "min_score_filter": 0.75
+}

training/data/dataset_info.json

@@ -0,0 +1,16 @@
+{
+  "description": "SevZero SFT expert trajectories for Llama-3.1-8B-Instruct style chat training.",
+  "version": "1.0.0",
+  "license": "apache-2.0",
+  "build": {
+    "episodes_total_seen": 90,
+    "episodes_kept": 42,
+    "episodes_dropped": 48,
+    "mean_episode_score_kept": 0.836021,
+    "train_rows": 853,
+    "eval_rows": 80,
+    "max_prompt_token_length": 2,
+    "max_observation_user_token_budget": 2048,
+    "min_score_filter": 0.75
+  }
+}

training/env_client.py

@@ -0,0 +1,159 @@
+"""
+Async HTTP client for the SevZero OpenEnv server (stateful /reset, /step, /state, /grader).
+Used by train_grpo rollout_func. Does not use root client.py (WebSocket); mirrors inference.py HTTP usage.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+_DEFAULT_TIMEOUT = 120.0
+_MAX_RETRIES = 5
+_BACKOFF = 1.6
+
+
+def _space_id_to_runtime_url(space_id: str) -> str:
+    """HF Space 'org/name' -> https://org-name.hf.space (common runtime URL)."""
+    space_id = space_id.strip()
+    if space_id.startswith("http"):
+        return space_id.rstrip("/")
+    parts = space_id.split("/")
+    if len(parts) == 2:
+        org, name = parts[0], parts[1]
+        # HF uses lowercase, slashes -> dashes in subdomains
+        sub = f"{org}-{name}".replace("_", "-").lower()
+        return f"https://{sub}.hf.space"
+    raise ValueError(f"Invalid space_id (expected 'org/name' or URL): {space_id!r}")
+
+
+def _backoff_delay(attempt: int) -> float:
+    return min(30.0, _BACKOFF**attempt)
+
+
+def _is_transient_status(code: int) -> bool:
+    return code in (429, 500, 502, 503, 504)
+
+
+class AsyncSevZeroEnvClient:
+    """
+    Minimal async env client: reset / step / state / grader.
+    Pass base_url from SEVZERO_ENV_URL or from_hf_space().
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        *,
+        token: Optional[str] = None,
+        timeout: float = _DEFAULT_TIMEOUT,
+    ) -> None:
+        self._base = base_url.rstrip("/")
+        self._token = token
+        headers: Dict[str, str] = {"Content-Type": "application/json"}
+        if token:
+            headers["Authorization"] = f"Bearer {token}"
+        self._client = httpx.AsyncClient(
+            base_url=self._base,
+            headers=headers,
+            timeout=timeout,
+        )
+
+    @classmethod
+    def from_hf_space(
+        cls,
+        space_id: str,
+        token: Optional[str] = None,
+    ) -> "AsyncSevZeroEnvClient":
+        """
+        space_id: 'organization/space_name' (HF Space) or a full http(s) URL.
+        For private Spaces, pass a read token with Space access.
+        """
+        return cls(_space_id_to_runtime_url(space_id), token=token or os.environ.get("HF_TOKEN"))
+
+    async def aclose(self) -> None:
+        await self._client.aclose()
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        json: Any = None,
+    ) -> httpx.Response:
+        last_err: Optional[Exception] = None
+        for attempt in range(_MAX_RETRIES):
+            try:
+                r = await self._client.request(method, path, json=json)
+                if r.status_code < 400:
+                    return r
+                if _is_transient_status(r.status_code) and attempt < _MAX_RETRIES - 1:
+                    await asyncio.sleep(_backoff_delay(attempt + 1))
+                    continue
+                return r
+            except (httpx.TimeoutException, httpx.NetworkError) as e:
+                last_err = e
+                if attempt < _MAX_RETRIES - 1:
+                    await asyncio.sleep(_backoff_delay(attempt + 1))
+                    continue
+                raise
+        if last_err:
+            raise last_err
+        raise RuntimeError("request failed")
+
+    async def reset(
+        self,
+        *,
+        task_id: str = "hard",
+        seed: int = 13,
+        episode_id: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"task_id": task_id, "seed": seed}
+        if episode_id:
+            body["episode_id"] = episode_id
+        r = await self._request("POST", "/reset", json=body)
+        r.raise_for_status()
+        return r.json()
+
+    async def step(self, action: Dict[str, Any]) -> Dict[str, Any]:
+        r = await self._request("POST", "/step", json={"action": action})
+        r.raise_for_status()
+        return r.json()
+
+    async def get_state(self) -> Dict[str, Any]:
+        r = await self._request("GET", "/state")
+        r.raise_for_status()
+        return r.json()
+
+    async def grade_episode(
+        self,
+        *,
+        final_slo_score: float,
+        steps_taken: int,
+        max_steps: int,
+        actions_taken: List[Dict[str, Any]],
+        terminated: bool,
+        termination_reason: Optional[str],
+    ) -> Dict[str, Any]:
+        r = await self._request(
+            "POST",
+            "/grader",
+            json={
+                "final_slo_score": final_slo_score,
+                "steps_taken": steps_taken,
+                "max_steps": max_steps,
+                "actions_taken": actions_taken,
+                "terminated": terminated,
+                "termination_reason": termination_reason,
+            },
+        )
+        r.raise_for_status()
+        return r.json()
+
+
+def run_async(coro):
+    """Run async coroutine from sync context (rollout_func)."""
+    return asyncio.run(coro)

training/eval.py

@@ -0,0 +1,269 @@
+#!/usr/bin/env python3
+"""
+Eval: local HF adapters + Gemini (google-genai) + Azure OpenAI + Azure AI Inference.
+Writes eval_results.csv; pushes Mist-ic/sevzero-eval-results with HF_MAIN_TOKEN. No Claude.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import os
+import sys
+from pathlib import Path
+from typing import Any, Callable, Dict, List
+
+_REPO = Path(__file__).resolve().parent.parent
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+
+from training.config_utils import try_load_env_files
+from training.rollout_sevzero import SRE_SYSTEM_PROMPT, build_observation_prompt, parse_action
+
+try_load_env_files()
+
+HELD_OUT = (13, 99, 777)
+DEFAULT_TASKS = ("easy", "medium", "hard")
+DATASET_HUB = "Mist-ic/sevzero-eval-results"
+
+BUILTIN: Dict[str, str] = {
+    "untrained-llama": "base:meta-llama/Llama-3.1-8B-Instruct",
+    "sft-primary": os.getenv("SFT_ADAPTER_PRIMARY", "PhaseOfCode/sevzero-llama3-8b-sft"),
+    "sft-backup": os.getenv("SFT_ADAPTER_BACKUP", "NoahInOblivion/sevzero-llama3-8b-sft"),
+    "sft-innovation": os.getenv("SFT_ADAPTER_INNOVATION", "NoxIsOblivion/sevzero-llama3-8b-sft"),
+    "grpo-primary": os.getenv("GRPO_ADAPTER_PRIMARY", "PhaseOfCode/sevzero-llama3-8b-grpo-primary"),
+    "grpo-stability": os.getenv("GRPO_ADAPTER_STABILITY", "NoahInOblivion/sevzero-llama3-8b-grpo-stability"),
+    "grpo-innovation": os.getenv("GRPO_ADAPTER_INNOVATION", "NoxIsOblivion/sevzero-llama3-8b-grpo-innovation"),
+}
+
+AZURE_INF = {
+    "grok-4.20-reasoning": "grok-2-latest",
+    "kimi-k2.6": "kimi-k2-6-2025",
+    "DeepSeek-V3.2": "DeepSeek-V3-2",
+}
+
+
+def run_episode(
+    base: str, task: str, seed: int, answer: Callable[[str, str], str]
+) -> Dict[str, Any]:
+    import httpx
+
+    with httpx.Client(base_url=base.rstrip("/"), timeout=120.0) as client:
+        r = client.post("/reset", json={"task_id": task, "seed": seed})
+        r.raise_for_status()
+        ro = r.json()
+        obs = ro.get("observation", ro)
+        done = ro.get("done", False)
+        user_pfx = f"You are the on-call SRE. task={task!r} seed={seed}.\n\n## Session\n"
+        for _ in range(1 + int(obs.get("max_steps", 20))):
+            if done:
+                break
+            user_block = user_pfx + build_observation_prompt(obs)
+            text = answer(SRE_SYSTEM_PROMPT, user_block)
+            act = parse_action(text)
+            sr = client.post(
+                "/step",
+                json={"action": {"action_type": str(act.get("action_type", "noop")), "params": act.get("params") or {}}},
+            )
+            sr.raise_for_status()
+            out = sr.json()
+            obs = out.get("observation", out)
+            done = out.get("done", False)
+        stt = client.get("/state")
+        stt.raise_for_status()
+        fs = stt.json()
+        g = client.post(
+            "/grader",
+            json={
+                "final_slo_score": float(fs.get("global_slo_score", 0.0)),
+                "steps_taken": int(fs.get("step_count", 0)),
+                "max_steps": int((obs or {}).get("max_steps", 10)),
+                "actions_taken": list((obs or {}).get("actions_taken", [])),
+                "terminated": bool(fs.get("terminated", True)),
+                "termination_reason": fs.get("termination_reason"),
+            },
+        )
+        js: Dict[str, Any] = {}
+        if g.status_code < 400:
+            js = g.json()
+    return {
+        "score": float(js.get("score", 0.0)),
+        "slo_recovery": float(js.get("slo_recovery", 0.0)),
+        "action_efficiency": float(js.get("action_efficiency", 0.0)),
+        "time_efficiency": float(js.get("time_efficiency", 0.0)),
+        "steps_used": int(fs.get("step_count", 0)),
+        "terminated": fs.get("terminated", True),
+        "termination_reason": str(fs.get("termination_reason", "")),
+    }
+
+
+def load_llama_peft(adapter_id: str | None):
+    import torch
+    from peft import PeftModel
+    from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
+
+    base_id = "meta-llama/Llama-3.1-8B-Instruct"
+    tok = AutoTokenizer.from_pretrained(base_id, use_fast=True, token=os.environ.get("HF_TOKEN"))
+    if tok.pad_token is None:
+        tok.pad_token = tok.eos_token
+    bnb = BitsAndBytesConfig(
+        load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16
+    )
+    m = AutoModelForCausalLM.from_pretrained(
+        base_id, quantization_config=bnb, device_map="auto", torch_dtype=torch.bfloat16, token=os.environ.get("HF_TOKEN")
+    )
+    if adapter_id:
+        m = PeftModel.from_pretrained(m, adapter_id, token=os.environ.get("HF_TOKEN"))
+    m.eval()
+    return tok, m
+
+
+def hf_answer(tok, mdl):
+    import torch
+
+    def answer(system: str, user: str) -> str:
+        messages = [{"role": "system", "content": system}, {"role": "user", "content": user}]
+        p = tok.apply_chat_template(messages, add_generation_prompt=True, tokenize=False)
+        inputs = tok(p, return_tensors="pt").to(mdl.device)
+        with torch.no_grad():
+            o = mdl.generate(**inputs, max_new_tokens=256, do_sample=True, temperature=0.0)
+        gen = o[0, inputs["input_ids"].shape[1] :]
+        return tok.decode(gen, skip_special_tokens=True)
+
+    return answer
+
+
+def answer_gemini(system: str, user: str) -> str:
+    from google import genai
+
+    model = os.environ.get(
+        "GEMINI_EVAL_MODEL",
+        os.environ.get("GEMINI_MODEL_PRO", "gemini-3.1-pro-preview"),
+    )
+    c = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
+    r = c.models.generate_content(model=model, contents=f"{system}\n\n{user}")
+    return (r.text or "").strip()
+
+
+def answer_azure_openai(system: str, user: str) -> str:
+    from openai import OpenAI
+
+    ep = os.environ.get("AZURE_OPENAI_ENDPOINT", "").rstrip("/")
+    c = OpenAI(
+        api_key=os.environ.get("AZURE_API_KEY", ""),
+        base_url=ep + "/openai/v1",
+    )
+    dep = os.environ.get("AZURE_GPT_DEPLOYMENT", "gpt-5.4-pro")
+    r = c.chat.completions.create(
+        model=dep,
+        messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
+        temperature=0.0,
+        max_tokens=512,
+    )
+    return (r.choices[0].message.content or "").strip()
+
+
+def answer_azure_inference(model_name: str, system: str, user: str) -> str:
+    from azure.ai.inference import ChatCompletionsClient
+    from azure.core.credentials import AzureKeyCredential
+
+    ep = os.environ.get("AZURE_AI_INFERENCE_ENDPOINT", "").rstrip("/") + "/"
+    c = ChatCompletionsClient(endpoint=ep, credential=AzureKeyCredential(os.environ.get("AZURE_API_KEY", "")))
+    r = c.complete(
+        model_name=model_name,
+        messages=[{"role": "user", "content": f"{system}\n\n{user}"}],
+    )
+    return (r.choices[0].message.content or "").strip()
+
+
+def pick_answer_fn(name: str) -> Callable[[str, str], str]:
+    n = name.strip()
+    if n in BUILTIN:
+        spec = BUILTIN[n]
+        aid = None if spec.startswith("base:") else spec
+        tok, m = load_llama_peft(aid)
+        return hf_answer(tok, m)
+    if "/" in n and n.count("/") == 1 and not n.startswith("meta-llama/"):
+        tok, m = load_llama_peft(n)
+        return hf_answer(tok, m)
+    if n.startswith("gemini"):
+        return answer_gemini
+    if "gpt" in n.lower() or n == "gpt-5.4-pro":
+        return answer_azure_openai
+    if n in AZURE_INF:
+        mid = AZURE_INF[n]
+
+        def _fn(s: str, u: str) -> str:
+            return answer_azure_inference(mid, s, u)
+
+        return _fn
+    raise ValueError(f"Unknown model key: {name!r}")
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--models", type=str, default="untrained-llama")
+    ap.add_argument("--out", type=str, default="eval_results.csv")
+    ap.add_argument("--seeds", type=str, default=",".join(str(s) for s in HELD_OUT))
+    ap.add_argument("--tasks", type=str, default=",".join(DEFAULT_TASKS))
+    a = ap.parse_args()
+
+    base = (os.environ.get("SEVZERO_ENV_URL") or "").rstrip("/")
+    if not base:
+        raise SystemExit("SEVZERO_ENV_URL required")
+
+    models = [m.strip() for m in a.models.split(",") if m.strip()]
+    seeds = [int(x) for x in a.seeds.split(",")]
+    tasks = [t.strip() for t in a.tasks.split(",")]
+
+    rows: List[Dict[str, Any]] = []
+    for mname in models:
+        try:
+            answer = pick_answer_fn(mname)
+        except ValueError as e:
+            print(f"SKIP {mname}: {e}", flush=True)
+            continue
+        for task in tasks:
+            for seed in seeds:
+                r = run_episode(base, task, seed, answer)
+                rows.append(
+                    {
+                        "model": mname,
+                        "task": task,
+                        "seed": seed,
+                        **r,
+                    }
+                )
+                print(rows[-1], flush=True)
+
+    with Path(a.out).open("w", newline="", encoding="utf-8") as f:
+        fieldnames = [
+            "model",
+            "task",
+            "seed",
+            "score",
+            "slo_recovery",
+            "action_efficiency",
+            "time_efficiency",
+            "steps_used",
+            "terminated",
+            "termination_reason",
+        ]
+        w = csv.DictWriter(f, fieldnames=fieldnames)
+        w.writeheader()
+        for r in rows:
+            w.writerow(r)
+
+    tok_m = os.environ.get("HF_MAIN_TOKEN", "")
+    if not tok_m:
+        print("HF_MAIN_TOKEN not set — skip Hub push", flush=True)
+        return
+    from datasets import Dataset
+
+    ds = Dataset.from_list([dict(x) for x in rows])
+    ds.push_to_hub(DATASET_HUB, token=tok_m, private=False)
+    print(f"OK: pushed hf.co/datasets/{DATASET_HUB}", flush=True)
+
+
+if __name__ == "__main__":
+    main()

training/launch_hf_job.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""
+Submit a HuggingFace Job to run training/train_sft.py or training/train_grpo.py.
+Uses huggingface_hub.run_job; prints job URL; appends training/runs.jsonl.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+_REPO = Path(__file__).resolve().parent.parent
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+
+from training.config_utils import try_load_env_files
+
+try_load_env_files()
+
+
+def _default_git_url() -> str:
+    r = subprocess.run(
+        ["git", "remote", "get-url", "origin"],
+        cwd=str(_REPO),
+        capture_output=True,
+        text=True,
+    )
+    return (r.stdout or "").strip() if r.returncode == 0 else ""
+
+
+def main() -> None:
+    p = argparse.ArgumentParser()
+    p.add_argument("--account_token", type=str, default=os.environ.get("HF_TOKEN", ""))
+    p.add_argument("--script", type=str, choices=("sft", "grpo"), required=True)
+    p.add_argument("--variant_name", type=str, default="run")
+    p.add_argument("--hardware", type=str, default="l40sx1")
+    p.add_argument(
+        "--image",
+        type=str,
+        default="pytorch/pytorch:2.6.0-cuda12.4-cudnn9-runtime",
+    )
+    p.add_argument("--git-url", type=str, default="")
+    p.add_argument(
+        "--env_vars",
+        type=str,
+        default="",
+        help="KEY=val pairs comma-separated, e.g. SEVZERO_ENV_URL=https://x.hf.space,HF_MAIN_TOKEN=...",
+    )
+    a, rest = p.parse_known_args()
+    if not a.account_token:
+        raise SystemExit("Need HF_TOKEN or --account_token")
+    git_url = a.git_url or _default_git_url()
+    if not git_url:
+        raise SystemExit("Set --git-url or configure git origin")
+    ev = {k: v for k, v in [x.split("=", 1) for x in a.env_vars.split(",") if "=" in x]}
+    if "SEVZERO_ENV_URL" not in ev and os.environ.get("SEVZERO_ENV_URL"):
+        ev["SEVZERO_ENV_URL"] = os.environ["SEVZERO_ENV_URL"]
+
+    which = f"training/train_{a.script}.py"
+    extra = " ".join(rest)
+    inner = (
+        f"set -euo pipefail && git clone --depth 1 {git_url!r} /work/r && cd /work/r && "
+        "pip install -U pip 'trl>=0.20' 'peft' 'transformers' 'accelerate' 'bitsandbytes' 'datasets' "
+        "'huggingface_hub' 'httpx' 'python-dotenv' 'vllm' 'unsloth' 2>/dev/null || true && "
+        f"python {which} --variant_name {a.variant_name!r} {extra}"
+    )
+    from huggingface_hub import run_job
+
+    job = run_job(
+        image=a.image,
+        command=["bash", "-lc", inner],
+        env=ev,
+        secrets={"HF_TOKEN": a.account_token},
+        flavor=a.hardware,
+    )
+    with (_REPO / "training" / "runs.jsonl").open("a", encoding="utf-8") as f:
+        f.write(
+            json.dumps(
+                {
+                    "account_token_tail": a.account_token[-4:] if len(a.account_token) > 4 else "",
+                    "job_id": str(getattr(job, "id", job)),
+                    "variant_name": a.variant_name,
+                    "started_at": datetime.now(timezone.utc).isoformat(),
+                }
+            )
+            + "\n"
+        )
+    print(getattr(job, "url", f"https://huggingface.co/jobs/{getattr(job, 'id', job)}"), flush=True)
+
+
+if __name__ == "__main__":
+    main()

training/loader.py

@@ -0,0 +1,56 @@
+"""
+Load SevZero SFT data for a trainer: local JSONL or the Hub Parquet copy.
+
+The training config should set `max_seq_length` to at least
+`max_prompt_token_length` from `build_stats.json` (plus max completion length).
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any, Optional, Union
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+DATA_DIR = REPO_ROOT / "training" / "data"
+
+try:
+    from datasets import Dataset, DatasetDict, load_dataset
+except ImportError as e:
+    raise ImportError("Install `datasets` to use the loader.") from e
+
+
+def load_local_jsonl(
+    train_path: Optional[Path] = None,
+    eval_path: Optional[Path] = None,
+) -> DatasetDict:
+    train_path = train_path or (DATA_DIR / "sft_train.jsonl")
+    eval_path = eval_path or (DATA_DIR / "sft_eval.jsonl")
+    train = load_dataset("json", data_files=str(train_path), split="train")
+    if eval_path.is_file() and eval_path.stat().st_size > 0:
+        ev = load_dataset("json", data_files=str(eval_path), split="train")
+    else:
+        ev = train.select([])
+    return DatasetDict(train=train, eval=ev)
+
+
+def load_from_hub(
+    repo_id: str = "Mist-ic/sevzero-expert-trajectories",
+    token: Optional[str] = None,
+) -> DatasetDict:
+    tok = token or os.environ.get("HF_MAIN_TOKEN")
+    return load_dataset(repo_id, token=tok)  # type: ignore[return-value]
+
+
+def read_build_stats() -> dict[str, Any]:
+    p = DATA_DIR / "build_stats.json"
+    if not p.is_file():
+        return {}
+    return json.loads(p.read_text(encoding="utf-8"))
+
+
+def recommended_max_seq_length(plus_completion: int = 1024) -> int:
+    s = read_build_stats()
+    m = int(s.get("max_prompt_token_length", 0) or 0)
+    return m + plus_completion

training/preflight.py

@@ -0,0 +1,250 @@
+#!/usr/bin/env python3
+"""
+(1) In-process Sim + grader: golden remediation plan → score >= 0.9 when possible
+(2) Uvicorn /health (optional) + 5 CPU GRPO steps with rollout_func + tiny model
+"""
+
+from __future__ import annotations
+
+import os
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Tuple
+
+_REPO = Path(__file__).resolve().parent.parent
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+
+from training.config_utils import try_load_env_files
+
+try_load_env_files()
+
+
+def _action_plan(seed: int, task_id: str) -> List[Tuple[str, Dict[str, Any]]]:
+    from server.failures import FailureType
+    from server.scenarios import generate_scenario
+
+    sc = generate_scenario(seed, task_id)
+    if not sc.failure_specs:
+        return [("noop", {})]
+    spec = sc.failure_specs[0]
+    sid = spec.service_id
+    ft = spec.failure_type
+    if ft == FailureType.BAD_DEPLOY:
+        return [("rollback_service", {"service_id": sid})]
+    if ft in (FailureType.CONFIG_STARTUP, FailureType.CONFIG_RUNTIME):
+        k = spec.broken_config_key or "timeout_ms"
+        out = [("tune_config", {"service_id": sid, "key": k, "value": "correct"})]
+        if ft == FailureType.CONFIG_STARTUP:
+            out.append(("restart_service", {"service_id": sid}))
+        return out
+    if ft == FailureType.CACHE_FAILURE:
+        return [("clear_cache", {"cache_name": sid})]
+    if ft == FailureType.CASCADING_LATENCY:
+        return [("scale_service", {"service_id": sid, "replicas": 4})]
+    if ft == FailureType.NETWORK_ERROR:
+        return [("noop", {}), ("noop", {})]
+    return [("restart_service", {"service_id": sid})]
+
+
+def _inproc_golden_score(seed: int, task_id: str) -> float:
+    from server.grader import grade_episode
+    from server.scenarios import generate_scenario
+    from server.simulator import Simulator
+
+    sc = generate_scenario(seed, task_id)
+    sim = Simulator()
+    sim.reset(seed=seed, difficulty=sc.difficulty, failure_specs=sc.failure_specs)
+    for at, p in _action_plan(seed, task_id):
+        sim.step(at, p)
+        for _ in range(4):
+            if sim.terminated:
+                break
+            sim.step("noop", {})
+    g = grade_episode(
+        final_slo_score=sim.get_slo_score(),
+        steps_taken=len(sim.actions_taken),
+        max_steps=sc.max_steps,
+        actions_taken=sim.actions_taken,
+        terminated=sim.terminated,
+        termination_reason=sim.termination_reason,
+    )
+    return float(g.score)
+
+
+def _grpo_tiny() -> bool:
+    try:
+        import trl  # noqa: F401
+    except ImportError:
+        print("GRPO preflight: trl not installed — skip (pip install trl)", flush=True)
+        return True
+    os.environ["UNSLOTH_DISABLE"] = "1"
+    os.environ["CUDA_VISIBLE_DEVICES"] = os.environ.get("CUDA_VISIBLE_DEVICES", "")
+
+    from datasets import Dataset
+    from peft import LoraConfig, get_peft_model
+    from transformers import AutoModelForCausalLM, AutoTokenizer
+    from trl import GRPOConfig, GRPOTrainer
+    from trl.experimental.openenv import generate_rollout_completions
+
+    from training.env_client import AsyncSevZeroEnvClient, run_async
+    from training.rollout_sevzero import SRE_SYSTEM_PROMPT, build_observation_prompt, parse_action
+
+    base = (os.environ.get("SEVZERO_ENV_URL") or "").rstrip("/")
+    if not base:
+        print("SEVZERO_ENV_URL unset — skip GRPO smoke", flush=True)
+        return True
+
+    tok = AutoTokenizer.from_pretrained("HuggingFaceTB/SmolLM2-135M-Instruct")
+    m = AutoModelForCausalLM.from_pretrained("HuggingFaceTB/SmolLM2-135M-Instruct", device_map="cpu")
+    m = get_peft_model(
+        m,
+        LoraConfig(
+            r=4,
+            lora_alpha=8,
+            target_modules=["q_proj", "v_proj"],
+            lora_dropout=0.0,
+            task_type="CAUSAL_LM",
+        ),
+    )
+
+    def rollout_func(prompts, trainer):
+        ep_ids: List[int] = []
+        ec_ids: List[int] = []
+        elp: List[float] = []
+        env_r: List[float] = []
+        for pr in prompts:
+            client = AsyncSevZeroEnvClient(base, None)
+
+            async def run_one():
+                p_ids, c_ids, lps = [], [], []
+                step_sum = 0.0
+                try:
+                    ro = await client.reset(task_id="easy", seed=7)
+                    obs = ro.get("observation", ro)
+                    done = ro.get("done", False)
+                    for _ in range(2):
+                        if done:
+                            break
+                        u = build_observation_prompt(obs)
+                        msg = [
+                            {"role": "system", "content": SRE_SYSTEM_PROMPT},
+                            {"role": "user", "content": f"{pr}\n{u}"},
+                        ]
+                        ptxt = tok.apply_chat_template(msg, add_generation_prompt=True, tokenize=False)
+                        out = generate_rollout_completions(trainer, [ptxt])[0]
+                        p_ids.extend(out.get("prompt_ids", []))
+                        c_ids.extend(out.get("completion_ids", []))
+                        lps.extend(out.get("logprobs", []))
+                        ctext = out.get("text")
+                        if not ctext and cids:
+                            ctext = tok.decode(cids, skip_special_tokens=True)
+                        a = parse_action(ctext or "")
+                        sr = await client.step(
+                            {
+                                "action": {
+                                    "action_type": str(a.get("action_type", "noop")),
+                                    "params": a.get("params") or {},
+                                }
+                            }
+                        )
+                        obs = sr.get("observation", sr)
+                        done = sr.get("done", False)
+                        step_sum += float(obs.get("reward", sr.get("reward", 0.0) or 0.0))
+                    return p_ids, c_ids, lps, step_sum
+                finally:
+                    await client.aclose()
+
+            p, c, lp, s = run_async(run_one())
+            ep_ids.append(p)
+            ec_ids.append(c)
+            elp.append(lp)
+            env_r.append(s)
+        return {
+            "prompt_ids": ep_ids,
+            "completion_ids": ec_ids,
+            "logprobs": elp,
+            "env_reward": env_r,
+        }
+
+    def rf(completions, **kwargs):
+        return [float(x) for x in kwargs.get("env_reward", [0.0] * len(completions))]
+
+    out_dir = str(_REPO / "training" / ".preflight_grpo")
+    os.makedirs(out_dir, exist_ok=True)
+    tr = GRPOTrainer(
+        model=m,
+        processing_class=tok,
+        args=GRPOConfig(
+            output_dir=out_dir,
+            per_device_train_batch_size=1,
+            max_steps=5,
+            num_generations=1,
+            use_vllm=False,
+            learning_rate=1e-5,
+            max_completion_length=32,
+        ),
+        train_dataset=Dataset.from_list([{"text": "x"}] * 2),
+        reward_funcs=[rf],
+        rollout_func=rollout_func,
+    )
+    tr.train()
+    return True
+
+
+def main() -> None:
+    # --- Part A: in-process (no network)
+    for seed, task in ((100, "easy"), (13, "easy"), (7, "easy")):
+        s = _inproc_golden_score(seed, task)
+        print(f"in-proc grader: seed={seed} task={task} score={s:.3f}", flush=True)
+        if s >= 0.9:
+            print("OK: in-process golden path reached >=0.9", flush=True)
+            break
+    else:
+        print("WARN: no seed reached 0.9 in in-proc test — check failure coverage", flush=True)
+
+    # --- B: Uvicorn + optional GRPO (requires same deps as the project)
+    try:
+        import uvicorn  # noqa: F401
+    except ImportError:
+        print("SKIP: uvicorn not installed — pip install the project (see training/README.md)", flush=True)
+        print("OK", flush=True)
+        return
+
+    port = int(os.environ.get("PREFLIGHT_PORT", "8765"))
+    base = f"http://127.0.0.1:{port}"
+    os.environ["SEVZERO_ENV_URL"] = base
+    import urllib.request
+
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "uvicorn", "server.app:app", "--host", "127.0.0.1", "--port", str(port)],
+        cwd=str(_REPO),
+    )
+    try:
+        for _ in range(25):
+            try:
+                with urllib.request.urlopen(f"{base}/health", timeout=2) as r:
+                    if getattr(r, "status", 200) < 500:
+                        break
+            except Exception:
+                time.sleep(0.5)
+        else:
+            raise RuntimeError("uvicorn not up")
+        try:
+            _grpo_tiny()
+        except Exception as e:
+            print(f"GRPO smoke failed (env OK): {e}", flush=True)
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=10)
+        except Exception:
+            proc.kill()
+    print("OK", flush=True)
+
+
+if __name__ == "__main__":
+    main()

training/push_dataset.py

@@ -0,0 +1,127 @@
+"""
+Upload SFT jsonl to Hugging Face (Mist-ic Main account) as a public dataset with Parquet.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv
+from huggingface_hub import HfApi
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+load_dotenv(REPO_ROOT / "api.env")
+load_dotenv(REPO_ROOT / "hg.env")
+
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+DATA_DIR = REPO_ROOT / "training" / "data"
+STATS_PATH = DATA_DIR / "build_stats.json"
+
+
+def _readme(stats: dict) -> str:
+    return f"""# SevZero expert trajectories (SFT)
+
+## Sources
+
+- Synthetic expert rollouts from frontier models (Gemini 3.1 Pro, Azure OpenAI, Azure AI Inference)
+  against the local OpenEnv `server.app` SevZero environment.
+
+## Filtering
+
+- Episodes with final grader `score` **≥** `{stats.get("min_score_filter", 0.85)}` are included.
+
+## Schema
+
+- Each example has a `messages` list (Llama-3.1-8B-Instruct–style SFT) and `meta` (episode / step provenance):
+  - `system`: SRE on-call system prompt (same as `inference.SYSTEM_PROMPT` in the repo)
+  - `user`: JSON-serialized observation (shrink to ≤ {stats.get("max_observation_user_token_budget", 2048)} tokens for the user part)
+  - `assistant`: one JSON object `{{"action_type": "...", "params": {{...}}}}`
+
+## Stats (from `build_stats.json` at publish time)
+
+{json.dumps(stats, indent=2)}
+
+## Parquet
+
+- Splits `train` and `eval` are also pushed in Parquet for fast `datasets.load_dataset`.
+"""
+
+
+def _dataset_info(stats: dict) -> dict:
+    return {
+        "description": "SevZero SFT expert trajectories for Llama-3.1-8B-Instruct style chat training.",
+        "version": "1.0.0",
+        "license": "apache-2.0",
+        "build": stats,
+    }
+
+
+def main() -> None:
+    token = os.environ.get("HF_MAIN_TOKEN", "")
+    if not token:
+        raise SystemExit("HF_MAIN_TOKEN missing (set in api.env or hg.env).")
+    user = (os.environ.get("HF_MAIN_USERNAME", "") or "").strip() or "Mist-ic"
+    repo_id = f"{user}/sevzero-expert-trajectories"
+    if not (DATA_DIR / "sft_train.jsonl").is_file():
+        raise SystemExit(f"Missing {DATA_DIR / 'sft_train.jsonl'} — run build_dataset.py first.")
+    stats: dict = {}
+    if STATS_PATH.is_file():
+        stats = json.loads(STATS_PATH.read_text(encoding="utf-8"))
+    readme = _readme(stats)
+    info = _dataset_info(stats)
+    (DATA_DIR / "DATASET_README_HF.md").write_text(readme, encoding="utf-8")
+    (DATA_DIR / "dataset_info.json").write_text(
+        json.dumps(info, indent=2), encoding="utf-8"
+    )
+
+    api = HfApi(token=token)
+    api.create_repo(
+        repo_id=repo_id,
+        repo_type="dataset",
+        private=False,
+        exist_ok=True,
+    )
+    for name in (
+        "sft_train.jsonl",
+        "sft_eval.jsonl",
+        "build_stats.json",
+        "dataset_info.json",
+    ):
+        p = DATA_DIR / name
+        if p.is_file():
+            api.upload_file(
+                path_or_fileobj=str(p),
+                path_in_repo=name,
+                repo_id=repo_id,
+                repo_type="dataset",
+                commit_message="Add SFT files and metadata",
+            )
+    api.upload_file(
+        path_or_fileobj=readme.encode("utf-8"),
+        path_in_repo="README.md",
+        repo_id=repo_id,
+        repo_type="dataset",
+        commit_message="Add dataset README",
+    )
+
+    from datasets import DatasetDict, load_dataset
+
+    train = load_dataset("json", data_files=str(DATA_DIR / "sft_train.jsonl"))["train"]
+    evp = DATA_DIR / "sft_eval.jsonl"
+    if evp.is_file() and evp.stat().st_size > 0:
+        ev = load_dataset("json", data_files=str(evp))["train"]
+    else:
+        ev = train.select([])
+    dd = DatasetDict(train=train, eval=ev)
+    dd.push_to_hub(repo_id, private=False, token=token)
+
+    url = f"https://huggingface.co/datasets/{repo_id}"
+    print(url, flush=True)
+
+
+if __name__ == "__main__":
+    main()

training/rollout_sevzero.py

@@ -0,0 +1,109 @@
+"""
+SevZero multi-turn rollout helpers for TRL GRPO (sync API for rollout_func).
+Builds chat prompts from observations and parses one JSON action per turn.
+"""
+
+from __future__ import annotations
+
+import json
+import textwrap
+from typing import Any, Dict, List, Optional, Tuple
+
+SRE_SYSTEM_PROMPT = textwrap.dedent(
+    """\
+    You are an expert Site Reliability Engineer (SRE) responding to a production incident.
+    You are managing a microservice cluster experiencing failures.
+    Your goal: restore all services to healthy SLO compliance as efficiently as possible.
+
+    Respond with EXACTLY one JSON object — no explanation, no markdown, just raw JSON:
+    {"action_type": "...", "params": {...}}
+
+    Param rules (STRICT — single service only, never a list):
+    - inspect_logs / inspect_metrics / inspect_traces / restart_service / rollback_service / scale_service:
+        {"action_type": "X", "params": {"service_id": "order-service"}}
+    - tune_config:
+        {"action_type": "tune_config", "params": {"service_id": "order-service", "key": "api_endpoint", "value": "correct"}}
+    - clear_cache:
+        {"action_type": "clear_cache", "params": {"cache_name": "redis-cache"}}
+    - rebalance_traffic:
+        {"action_type": "rebalance_traffic", "params": {"from_region": "us-east-1", "to_region": "us-west-2"}}
+    - noop:
+        {"action_type": "noop", "params": {}}
+"""
+)
+
+
+def build_observation_prompt(obs: Dict[str, Any]) -> str:
+    """Port of inference.build_observation_prompt (observation dict from HTTP JSON)."""
+    parts = [f"## Incident Status\n{obs.get('observation_summary', 'N/A')}"]
+    alerts = obs.get("alerts") or []
+    if alerts:
+        alert_lines = [f"  [{a['severity'].upper()}] {a['message']}" for a in alerts[:10]]
+        parts.append("## Active Alerts\n" + "\n".join(alert_lines))
+    services = obs.get("services") or []
+    degraded = [s for s in services if s.get("status") in ("degraded", "critical", "down")]
+    if degraded:
+        svc_lines = []
+        for s in degraded:
+            sid = s["id"]
+            svc_lines.append(
+                f"  {sid} [{s['status']}]: error={s['error_rate']:.1%}, "
+                f"p99={s['latency_p99_ms']:.0f}ms, cpu={s['cpu_pct']:.0f}%, "
+                f"mem={s['memory_pct']:.0f}%"
+            )
+        parts.append("## Degraded Services\n" + "\n".join(svc_lines))
+    deploys = obs.get("recent_deploys") or []
+    if deploys:
+        dep_lines = [f"  {d['service']} -> {d['version']} ({d['ticks_ago']} ticks ago)" for d in deploys]
+        parts.append("## Recent Deploys\n" + "\n".join(dep_lines))
+    actions = obs.get("actions_taken") or []
+    if actions:
+        act_lines = [
+            f"  tick {a['tick']}: {a['action']}({a.get('target', '')}) -> {'OK' if a['success'] else 'FAIL'}"
+            for a in actions[-5:]
+        ]
+        parts.append("## Recent Actions\n" + "\n".join(act_lines))
+    logs = obs.get("logs")
+    if logs:
+        parts.append(f"## Logs\n{logs}")
+    traces = obs.get("traces")
+    if traces:
+        spans = (traces.get("spans") or []) if isinstance(traces, dict) else []
+        error_spans = [s for s in spans if s.get("status") == "ERROR"]
+        if error_spans:
+            trace_lines = [
+                f"  {s.get('service')}: {s.get('tags', {}).get('error.message', 'ERROR')}"
+                for s in error_spans[:5]
+            ]
+            parts.append("## Trace Errors\n" + "\n".join(trace_lines))
+    legal = obs.get("legal_actions") or []
+    if legal:
+        legal_strs = [f"  {la.get('action_type', '')}: targets={la.get('valid_targets', [])[:5]}" for la in legal]
+        parts.append("## Available Actions\n" + "\n".join(legal_strs))
+    return "\n\n".join(parts)
+
+
+def parse_action(response_text: str) -> Dict[str, Any]:
+    text = (response_text or "").strip()
+    if "```json" in text:
+        text = text.split("```json", 1)[1].split("```", 1)[0].strip()
+    elif "```" in text:
+        text = text.split("```", 1)[1].split("```", 1)[0].strip()
+    start, end = text.find("{"), text.rfind("}") + 1
+    if start >= 0 and end > start:
+        try:
+            return json.loads(text[start:end])
+        except json.JSONDecodeError:
+            pass
+    return {"action_type": "noop", "params": {}}
+
+
+def _normalize_action(action: Dict[str, Any]) -> Dict[str, Any]:
+    act_type = action.get("action_type", "noop")
+    params = dict(action.get("params") or {})
+    if "replicas" in params:
+        try:
+            params["replicas"] = int(params["replicas"])
+        except (TypeError, ValueError):
+            params["replicas"] = 2
+    return {"action_type": act_type, "params": params}

training/train_grpo.py

@@ -0,0 +1,317 @@
+#!/usr/bin/env python3
+"""
+GRPO on SevZero via TRL rollout_func + trl.experimental.openenv.generate_rollout_completions.
+Verify API with Context7 before changing integration (rollout_func is required; environment_factory is deprecated).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import random
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+_REPO = Path(__file__).resolve().parent.parent
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+
+from training.config_utils import try_load_env_files
+
+try_load_env_files()
+
+BASE_MODEL = "meta-llama/Llama-3.1-8B-Instruct"
+METRICS_NAME = "metrics.jsonl"
+
+# Pinned in README: trl, unsloth, vllm — orchestrator sets exact versions
+
+
+def _parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser()
+    p.add_argument("--output_dir", type=str, default="./outputs/grpo")
+    p.add_argument("--sft_adapter_repo", type=str, required=True, help="HF adapter repo (worker account)")
+    p.add_argument("--env_url", type=str, default="", help="Override; else SEVZERO_ENV_URL")
+    p.add_argument("--max_steps", type=int, default=350)
+    p.add_argument("--lr", type=float, default=7e-6)
+    p.add_argument("--K", type=int, default=4, dest="K", help="num_generations")
+    p.add_argument("--seed", type=int, default=42)
+    p.add_argument(
+        "--reward_shaping",
+        type=str,
+        default="dense_v1",
+        choices=("dense_v1", "dense_v2", "sparse"),
+    )
+    p.add_argument("--enable_schema_drift", action="store_true")
+    p.add_argument("--enable_curriculum", action="store_true")
+    p.add_argument("--enable_oversight", action="store_true")
+    p.add_argument(
+        "--task_mix",
+        type=str,
+        default="hard",
+        choices=("hard", "mixed", "curriculum"),
+    )
+    p.add_argument("--push_to_hub_repo", type=str, default="")
+    p.add_argument("--variant_name", type=str, default="grpo")
+    p.add_argument("--rollout_max_steps", type=int, default=0, help="0 = from env observation max_steps")
+    return p.parse_args()
+
+
+def _pick_task_id(args, idx: int, step: int) -> str:
+    if args.task_mix == "hard":
+        return "hard"
+    if args.task_mix == "mixed":
+        return ["easy", "medium", "hard"][idx % 3]
+    # curriculum: escalate every ~50 steps
+    if args.enable_curriculum:
+        tier = min(2, step // 50)
+        return ["easy", "medium", "hard"][tier]
+    return "hard"
+
+
+def _compute_episode_return(
+    shaping: str,
+    step_rewards: List[float],
+    grader: Optional[Dict[str, Any]],
+) -> float:
+    if shaping == "sparse" and grader is not None:
+        return float(grader.get("score", 0.0))
+    if shaping == "dense_v2" and grader is not None:
+        # Slightly weight terminal score
+        s = sum(step_rewards) if step_rewards else 0.0
+        return 0.7 * s + 0.3 * float(grader.get("score", 0.0))
+    return float(sum(step_rewards)) if step_rewards else 0.0
+
+
+def _build_default_dataset():
+    from datasets import Dataset
+
+    rows = []
+    for i in range(64):
+        text = (
+            "You are the on-call SRE. Restore service health. "
+            f"Incident session {i} — triage, diagnose root cause, remediate, verify."
+        )
+        rows.append({"text": text, "row_id": i})
+    return Dataset.from_list(rows)
+
+
+def _reward_from_env(completions, **kwargs):
+    r = kwargs.get("env_reward")
+    if r is None:
+        return [0.0] * len(completions)
+    return [float(x) for x in r]
+
+
+def main() -> None:
+    args = _parse_args()
+    env_url = (args.env_url or os.environ.get("SEVZERO_ENV_URL", "")).rstrip("/")
+    if not env_url:
+        raise SystemExit("Set --env_url or SEVZERO_ENV_URL to the remote SevZero HTTP base URL")
+
+    worker_token = os.environ.get("HF_TOKEN", "")
+    main_token = os.environ.get("HF_MAIN_TOKEN", "")
+
+    try:
+        import trackio
+
+        trackio.init(
+            project="sevzero-grpo",
+            space_id="Mist-ic/sevzero-trackio",
+            **({"hf_token": main_token} if main_token else {}),
+        )
+    except Exception as e:
+        print(f"trackio init skipped: {e}", flush=True)
+
+    try:
+        from unsloth import FastLanguageModel, PatchFastRL
+    except ImportError as e:
+        raise SystemExit(
+            f"unsloth is required for GRPO on this path: {e}\n"
+            "Install training extras, or on unsupported platforms set UNSLOTH_DISABLE=1 and extend train_grpo."
+        ) from e
+
+    PatchFastRL(algorithm="grpo", FastLanguageModel=FastLanguageModel)
+
+    from peft import PeftModel
+    from trl import GRPOConfig, GRPOTrainer
+    from trl.experimental.openenv import generate_rollout_completions
+
+    from training.env_client import AsyncSevZeroEnvClient, run_async
+    from training.rollout_sevzero import (
+        SRE_SYSTEM_PROMPT,
+        build_observation_prompt,
+        parse_action,
+    )
+
+    max_seq = 4096
+    model, tokenizer = FastLanguageModel.from_pretrained(
+        model_name=BASE_MODEL,
+        max_seq_length=max_seq,
+        dtype=None,
+        load_in_4bit=True,
+    )
+    model = PeftModel.from_pretrained(model, args.sft_adapter_repo, token=worker_token or None)
+    # Optional env flags (future env upgrades) — no-op for baseline server
+    if args.enable_schema_drift:
+        os.environ["SEVZERO_SCHEMA_DRIFT"] = "1"
+    if args.enable_oversight:
+        os.environ["SEVZERO_OVERSIGHT"] = "1"
+
+    metrics_path = Path(args.output_dir) / METRICS_NAME
+    metrics_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # Capture trainer ref for step index in seeding
+    _trainer_holder: List[Any] = [None]
+    _global_episode: List[int] = [0]
+
+    def rollout_func(prompts: List[str], trainer) -> Dict[str, List[Any]]:
+        _trainer_holder[0] = trainer
+        episode_prompt_ids: List[List[int]] = []
+        episode_completion_ids: List[List[int]] = []
+        episode_logprobs: List[List[float]] = []
+        env_rewards: List[float] = []
+        tkn = os.environ.get("HF_TOKEN", "")  # for private Space
+        for batch_idx, prompt_text in enumerate(prompts):
+            tr = _trainer_holder[0]
+            state = getattr(tr, "state", None) if tr else None
+            step = getattr(state, "global_step", 0) if state else 0
+            _global_episode[0] += 1
+            task_id = _pick_task_id(args, batch_idx, step)
+            seed = 13 + (batch_idx * 997) + (step * 13) + _global_episode[0] + random.randint(0, 1_000_000) % 100_000
+
+            async def _one_ep() -> tuple:
+                client = AsyncSevZeroEnvClient(env_url, token=tkn or None)
+                try:
+                    p_ids: List[int] = []
+                    c_ids: List[int] = []
+                    lps: List[float] = []
+                    step_rewards: List[float] = []
+                    ro = await client.reset(task_id=task_id, seed=seed)
+                    obs = ro.get("observation", ro)
+                    done = ro.get("done", False)
+                    grader: Optional[Dict[str, Any]] = None
+                    user_prefix = f"{prompt_text}\n\n## Session\n"
+                    for _t in range(args.rollout_max_steps or int(obs.get("max_steps", 20))):
+                        if done:
+                            break
+                        user_msg = build_observation_prompt(obs)
+                        messages = [
+                            {"role": "system", "content": SRE_SYSTEM_PROMPT},
+                            {"role": "user", "content": user_prefix + user_msg},
+                        ]
+                        p_text = tokenizer.apply_chat_template(
+                            messages, add_generation_prompt=True, tokenize=False,
+                        )
+                        out = generate_rollout_completions(tr, [p_text])[0]
+                        p_ids.extend(out.get("prompt_ids", []))
+                        c_ids.extend(out.get("completion_ids", []))
+                        lps.extend(out.get("logprobs", []))
+                        gen_ids = out.get("completion_ids", [])
+                        raw = out.get("text")
+                        if not raw and gen_ids:
+                            raw = tokenizer.decode(gen_ids, skip_special_tokens=True)
+                        action = parse_action(raw or "")
+                        step_payload = {
+                            "action_type": str(action.get("action_type", "noop")),
+                            "params": action.get("params") or {},
+                        }
+                        sr = await client.step({"action": step_payload})
+                        obs = sr.get("observation", sr)
+                        done = sr.get("done", False)
+                        r = float(obs.get("reward", sr.get("reward", 0.0) or 0.0))
+                        step_rewards.append(r)
+                    st = await client.get_state()
+                    max_st = int(obs.get("max_steps", 10))
+                    try:
+                        grader = await client.grade_episode(
+                            final_slo_score=float(st.get("global_slo_score", 0.0)),
+                            steps_taken=int(st.get("step_count", 0)),
+                            max_steps=max_st,
+                            actions_taken=list(obs.get("actions_taken", [])),
+                            terminated=bool(st.get("terminated", True)),
+                            termination_reason=st.get("termination_reason"),
+                        )
+                    except Exception:
+                        grader = None
+                    R = _compute_episode_return(args.reward_shaping, step_rewards, grader)
+                    return p_ids, c_ids, lps, R
+                finally:
+                    await client.aclose()
+
+            p_ids, c_ids, lps, r_ep = run_async(_one_ep())
+            episode_prompt_ids.append(p_ids)
+            episode_completion_ids.append(c_ids)
+            episode_logprobs.append(lps)
+            env_rewards.append(r_ep)
+        return {
+            "prompt_ids": episode_prompt_ids,
+            "completion_ids": episode_completion_ids,
+            "logprobs": episode_logprobs,
+            "env_reward": env_rewards,
+        }
+
+    grpo = GRPOConfig(
+        output_dir=args.output_dir,
+        learning_rate=args.lr,
+        per_device_train_batch_size=1,
+        gradient_accumulation_steps=8,
+        max_completion_length=1024,
+        num_train_epochs=1,
+        max_steps=args.max_steps,
+        num_generations=args.K,
+        temperature=0.85,
+        max_prompt_length=4096,
+        beta=0.04,
+        lr_scheduler_type="cosine",
+        use_vllm=True,
+        vllm_mode="colocate",
+        vllm_gpu_memory_utilization=0.55,
+        report_to="trackio",
+        logging_steps=1,
+        save_steps=100,
+    )
+
+    train_ds = _build_default_dataset()
+
+    trainer = GRPOTrainer(
+        model=model,
+        processing_class=tokenizer,
+        args=grpo,
+        train_dataset=train_ds,
+        reward_funcs=[_reward_from_env],
+        rollout_func=rollout_func,
+    )
+
+    from transformers import TrainerCallback
+
+    class _MetricsJSONL(TrainerCallback):
+        def on_log(self, args, state, control, logs=None, **kwargs):
+            if not logs:
+                return
+            rec = {
+                "step": state.global_step,
+                "reward_mean": logs.get("rewards", logs.get("reward", None)),
+                "reward_std": logs.get("reward_std", None),
+                "kl": logs.get("kl", None),
+                "entropy": logs.get("entropy", None),
+                "grad_norm": logs.get("grad_norm", None),
+                "loss": logs.get("loss", None),
+                "frac_reward_zero_std": logs.get("frac_reward_zero", logs.get("frac_reward_zero_std", None)),
+                "lr": logs.get("learning_rate", None),
+            }
+            with metrics_path.open("a", encoding="utf-8") as f:
+                f.write(json.dumps(rec, default=str) + "\n")
+            print(json.dumps({"type": "grpo", **rec}, default=str), flush=True)
+
+    trainer.add_callback(_MetricsJSONL())
+    trainer.train()
+
+    if args.push_to_hub_repo:
+        model.push_to_hub(args.push_to_hub_repo, token=worker_token or None, private=True)
+        tokenizer.push_to_hub(args.push_to_hub_repo, token=worker_token or None, private=True)
+
+
+if __name__ == "__main__":
+    main()

training/train_sft.py

@@ -0,0 +1,236 @@
+#!/usr/bin/env python3
+"""
+SFT warmup: QLoRA on Mist-ic/sevzero-expert-trajectories (see training/data/HANDOFF.md).
+Target TRL / Unsloth versions: see comments after `pip index` in training/README.md.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+_REPO = Path(__file__).resolve().parent.parent
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+
+from training.config_utils import try_load_env_files
+
+try_load_env_files()
+
+# --- Pin guidance (orchestrator resolves exact pins): trl>=0.22, unsloth, bitsandbytes, peft, accelerate
+BASE_MODEL = "meta-llama/Llama-3.1-8B-Instruct"
+DATASET_ID = "Mist-ic/sevzero-expert-trajectories"
+DEFAULT_MAX_SEQ = 2048
+
+
+def _parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser()
+    p.add_argument("--output_dir", type=str, default="./outputs/sft")
+    p.add_argument("--max_steps", type=int, default=250)
+    p.add_argument("--lr", type=float, default=1e-5)
+    p.add_argument("--seed", type=int, default=42)
+    p.add_argument("--push_to_hub_repo", type=str, default="", help="e.g. PhaseOfCode/sevzero-llama3-8b-sft")
+    p.add_argument("--variant_name", type=str, default="default")
+    p.add_argument("--max_seq_length", type=int, default=0, help="0 = read HANDOFF / 2048")
+    return p.parse_args()
+
+
+def _read_default_max_seq() -> int:
+    handoff = _REPO / "training" / "data" / "HANDOFF.md"
+    if not handoff.is_file():
+        return DEFAULT_MAX_SEQ
+    text = handoff.read_text(encoding="utf-8", errors="ignore")
+    for line in text.splitlines():
+        if "max_seq" in line.lower() and "`" in line:
+            try:
+                return int(line.split("`")[1])
+            except (ValueError, IndexError):
+                pass
+    return DEFAULT_MAX_SEQ
+
+
+def _format_row_to_text(row: dict, tokenizer) -> str:
+    """Support 'text' column or OpenAI-style messages JSON."""
+    if "text" in row and row["text"]:
+        return str(row["text"])
+    if "messages" in row and row["messages"]:
+        msgs = row["messages"]
+        if isinstance(msgs, str):
+            import json as _j
+
+            msgs = _j.loads(msgs)
+        return tokenizer.apply_chat_template(msgs, tokenize=False, add_generation_prompt=False)
+    raise ValueError("Dataset row must have 'text' or 'messages'")
+
+
+def main() -> None:
+    args = _parse_args()
+    max_seq = args.max_seq_length or _read_default_max_seq()
+
+    worker_token = os.environ.get("HF_TOKEN", "")
+    main_token = os.environ.get("HF_MAIN_TOKEN", "")
+    if not worker_token:
+        print("warning: HF_TOKEN not set — Hub push and model download may fail.", flush=True)
+
+    # Trackio with main account (read-only space) while training pushes use HF_TOKEN
+    try:
+        import trackio
+
+        if main_token:
+            os.environ.setdefault("HF_TOKEN", worker_token)
+        trackio.init(
+            project="sevzero-sft",
+            space_id="Mist-ic/sevzero-trackio",
+            **({"hf_token": main_token} if main_token else {}),
+        )
+    except Exception as e:
+        print(f"trackio init skipped: {e}", flush=True)
+
+    from datasets import load_dataset
+    from transformers import TrainingArguments
+    from trl import SFTConfig, SFTTrainer
+
+    ds = load_dataset(DATASET_ID, split="train")
+
+    use_unsloth = os.environ.get("UNSLOTH_DISABLE", "").lower() not in ("1", "true", "yes")
+    model = None
+    tokenizer = None
+
+    if use_unsloth:
+        try:
+            from unsloth import FastLanguageModel
+
+            model, tokenizer = FastLanguageModel.from_pretrained(
+                model_name=BASE_MODEL,
+                max_seq_length=max_seq,
+                dtype=None,
+                load_in_4bit=True,
+            )
+            target_modules = [
+                "q_proj",
+                "k_proj",
+                "v_proj",
+                "o_proj",
+                "gate_proj",
+                "up_proj",
+                "down_proj",
+            ]
+            model = FastLanguageModel.get_peft_model(
+                model,
+                r=32,
+                lora_alpha=64,
+                lora_dropout=0.0,
+                target_modules=target_modules,
+                use_gradient_checkpointing="unsloth",
+            )
+        except Exception as e:
+            print(f"Unsloth path failed ({e}), falling back to PEFT+bnb.", flush=True)
+            use_unsloth = False
+
+    if not use_unsloth:
+        import torch
+        from peft import LoraConfig, get_peft_model
+        from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
+
+        bnb = BitsAndBytesConfig(
+            load_in_4bit=True,
+            bnb_4bit_quant_type="nf4",
+            bnb_4bit_compute_dtype=torch.bfloat16,
+        )
+        tokenizer = AutoTokenizer.from_pretrained(BASE_MODEL, use_fast=True)
+        if tokenizer.pad_token is None:
+            tokenizer.pad_token = tokenizer.eos_token
+        model = AutoModelForCausalLM.from_pretrained(
+            BASE_MODEL,
+            quantization_config=bnb,
+            device_map="auto",
+            torch_dtype=torch.bfloat16,
+        )
+        lora = LoraConfig(
+            r=32,
+            lora_alpha=64,
+            lora_dropout=0.0,
+            target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
+            task_type="CAUSAL_LM",
+        )
+        model = get_peft_model(model, lora)
+
+    def formatting_prompts(examples: dict) -> dict:
+        texts = []
+        n = len(next(iter(examples.values())))
+        keys = list(examples.keys())
+        for i in range(n):
+            row = {k: (examples[k][i] if k in examples else None) for k in keys}
+            texts.append(_format_row_to_text(row, tokenizer))
+        return {"text": texts}
+
+    cols = ds.column_names
+    if "text" not in ds.column_names:
+        if "messages" in ds.column_names:
+            ds = ds.map(
+                formatting_prompts,
+                batched=True,
+                remove_columns=[c for c in cols if c not in ("messages",)],
+            )
+        else:
+            raise ValueError("Dataset must include a 'text' or 'messages' column")
+    targs = SFTConfig(
+        output_dir=args.output_dir,
+        max_steps=args.max_steps,
+        learning_rate=args.lr,
+        per_device_train_batch_size=4,
+        gradient_accumulation_steps=8,
+        warmup_ratio=0.05,
+        lr_scheduler_type="cosine",
+        optim="paged_adamw_8bit",
+        bf16=True,
+        seed=args.seed,
+        logging_steps=1,
+        report_to="trackio",
+        save_total_limit=2,
+        max_seq_length=max_seq,
+    )
+
+    from transformers import TrainerCallback
+
+    class JsonStepLog(TrainerCallback):
+        def on_log(self, args, state, control, logs=None, **kwargs):
+            if not logs:
+                return
+            payload = {
+                "type": "sft_step",
+                "step": state.global_step,
+                "loss": logs.get("loss"),
+                "lr": logs.get("learning_rate"),
+            }
+            print(json.dumps(payload, default=str), flush=True)
+
+    trainer = SFTTrainer(
+        model=model,
+        processing_class=tokenizer,
+        args=targs,
+        train_dataset=ds,
+        dataset_text_field="text",
+        callbacks=[JsonStepLog()],
+    )
+    trainer.train()
+
+    if args.push_to_hub_repo:
+        print(json.dumps({"event": "push_to_hub", "repo": args.push_to_hub_repo}, default=str), flush=True)
+        model.push_to_hub(
+            args.push_to_hub_repo,
+            token=worker_token or None,
+            private=True,
+        )
+        tokenizer.push_to_hub(
+            args.push_to_hub_repo,
+            token=worker_token or None,
+            private=True,
+        )
+
+
+if __name__ == "__main__":
+    main()