#!/usr/bin/env python3
"""ksim - Kubernetes certification exam simulator (CKA / CKAD / CKS tracks).
Hands-on tasks graded per-check, each on its own cluster that you reach with `ssh <host>` from a
remote desktop in the browser - the way the real exams run - with an exam clock, attempt history,
a weak-spot log, drills, and a path through each curriculum - every session against the clock."""
import concurrent.futures
import http.server
import importlib
import json
import os
import random
import re
import shutil
import signal
import subprocess
import sys
import textwrap
import threading
import time
import urllib.parse
from collections.abc import Callable
from datetime import datetime
from typing import Any

VERSION = "2026.10.19.1"
ROOT = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, ROOT)
from sim import desktop, envs  # noqa: E402

# ------------------------------------------------------------ track ----
ARGV = sys.argv[1:]
TRACKS = ["cka", "ckad", "cks"]
TRACK_ID = ""
TRACK = {}
QUESTIONS = []
DOMAINS = {}
QUOTA = {}
QUOTA_HARD = {}
QBYID = {}
SETS = {}
POOL = []
EXAM_SECONDS = 0
PASS_PCT = 0
STATE_FILE = ""
HISTORY_FILE = ""
WEAK_FILE = ""
PROGRESS_FILE = ""   # study-<track>.json: when each objective was done, your focus, what you marked known
OBJECTIVES = []   # the track's curriculum, objective by objective (<track>/curriculum.py), if it has one
STUDY_ORDER = []  # domains in the order the path walks them, when that differs from the curriculum's own
K8S = ""          # Kubernetes minor the track's real exam runs, e.g. "v1.37": picks the exam host image
PENDING = {}      # tasks of the bank that are held back, id -> why (PENDING in <track>/questions.py)
LESSONS = []      # guided labs, one per objective (<track>/lessons.py), each run like a task on a host of its own


def load_track(track_id):
    """Bind every track-dependent global. Called at startup and when the UI switches track."""
    try:
        mod = importlib.import_module(f"{track_id}.questions")
    except ModuleNotFoundError as e:
        if e.name not in (track_id, f"{track_id}.questions"):
            raise
        sys.exit(f"unknown track '{track_id}' (no {track_id}/questions.py)")
    importlib.reload(mod)
    track = mod.TRACK
    pending = dict(getattr(mod, "PENDING", {}))
    workers = dict(getattr(mod, "WORKERS", {}))
    try:   # scripted solutions, for ./ksim audit (<track>/solve.py)
        solve = dict(importlib.reload(importlib.import_module(f"{track_id}.solve")).SOLVE)
    except ModuleNotFoundError as e:
        if e.name != f"{track_id}.solve":
            raise
        solve = {}
    if not envs.vm_ready(track["k8s"]):
        for q in mod.QUESTIONS:
            if q.get("backend") == "vm":
                pending.setdefault(q["id"], f"needs the VM exam host - build it with: ./ksim build {track_id} --vm")
    questions = [q for q in mod.QUESTIONS if q["id"] not in pending]
    try:
        cur = importlib.import_module(f"{track_id}.curriculum")
        cur = importlib.reload(cur)
        objectives = [dict(o, tasks=[i for i in o["tasks"] if i not in pending]) for o in cur.OBJECTIVES]
        study_order = list(getattr(cur, "STUDY_ORDER", []))
    except ModuleNotFoundError as e:
        if e.name != f"{track_id}.curriculum":
            raise
        objectives, study_order = [], []  # no objective map for this track: drill by domain only
    try:
        lessons = [lesson_task(l, objectives) for l in importlib.reload(importlib.import_module(f"{track_id}.lessons")).LESSONS]
    except ModuleNotFoundError as e:
        if e.name != f"{track_id}.lessons":
            raise
        lessons = []
    for q in questions:
        q["weight"] = sum(c["pts"] for c in q["checks"])
        q.setdefault("difficulty", "normal")
        q.setdefault("set", None)
        if q["id"] in workers:
            q["workers"] = workers[q["id"]]
        if q["id"] in solve:
            q["solve"] = solve[q["id"]]
        # Every task has a cluster to itself, so nothing it breaks can reach another task: the flags
        # that kept such tasks apart on a shared cluster ("breaks", "solo", "disruptive") are void.
        for k in ("breaks", "solo", "disruptive"):
            q.pop(k, None)
    g = globals()
    g.update(
        TRACK_ID=track_id, TRACK=track, QUESTIONS=questions, DOMAINS=track["domains"],
        QUOTA=track.get("quota", {}), QUOTA_HARD=track.get("quota_hard", track.get("quota", {})),
        QBYID={q["id"]: q for q in questions},
        SETS=track.get("sets", {}),
        POOL=[q for q in questions if not q.get("set")],
        EXAM_SECONDS=track["minutes"] * 60, PASS_PCT=track["pass_pct"],
        STATE_FILE=os.path.join(ROOT, ".state", f"state-{track_id}.json"),
        HISTORY_FILE=os.path.join(ROOT, f"history-{track_id}.jsonl"),
        WEAK_FILE=os.path.join(ROOT, f"weak-spots-{track_id}.md"),
        PROGRESS_FILE=os.path.join(ROOT, f"study-{track_id}.json"), OBJECTIVES=objectives, STUDY_ORDER=study_order,
        K8S=track["k8s"], PENDING=pending, LESSONS=lessons,
    )


def lesson_task(l, objectives):
    """A lesson as the engine runs it: a task-shaped dict, so its host, setup and checks go through
    the same code as a task's (host_of, provision, run). Its steps' checks are its checks."""
    o = next((o for o in objectives if o["id"] == l["objective"]), None)
    return dict(l, lesson=True, domain=o["domain"] if o else "", difficulty="normal", set=None,
                checks=[{"name": s["title"], "pts": 1, "script": s["check"]} for s in l["steps"]], weight=len(l["steps"]),
                task=l["intro"], solution="\n\n".join(s["show"] for s in l["steps"] if s["show"]))


def set_questions(name):
    return [q for q in QUESTIONS if q.get("set") == name]


STATE_DIR = os.path.join(ROOT, ".state")
# How {WS} reads in a task and its solution: the directory of the task's files on its own host.
# Graders see the same directory on this machine as $WS (it is mounted into the host).
WS_DISPLAY = envs.COURSE
DRILL_SECONDS_PER_TASK = 6 * 60
DRILL_SECONDS_PER_TASK_HARD = 5 * 60
DRILL_SIZE = 5
# Clusters provisioned (and tasks graded) at the same time: one per four cores, at most 8.
PARALLEL = max(1, int(os.environ.get("KSIM_PARALLEL") or max(2, min(8, (os.cpu_count() or 8) // 4))))

# ---------------------------------------------- a task's own cluster ----
_WORKER_REF = re.compile(r"(?:\$\{?|\{(?:SHELL_)?)W([123])\b")

def node_image(q=None):
    """The exam host image of a task: its track's Kubernetes minor, in the track's own flavour if it
    has one - unless the task names a minor ("k8s") or a flavour ("image") of its own, as a task
    that upgrades its cluster to the track's minor does."""
    q = q or {}
    return envs.node_image(q.get("k8s", K8S), q.get("image", TRACK.get("image")))


def is_vm(q):
    """The task's host is a virtual machine ("backend": "vm") - for what a container host cannot do."""
    return q.get("backend") == "vm"


def make_host(q):
    """Create the host of a task: a kind cluster of containers, or a VM. Raises envs.EnvError."""
    if is_vm(q):
        return envs.create_vm(host_of(q), K8S)
    return envs.create(host_of(q), node_image(q), workers_of(q))


def workers_of(q):
    """Worker nodes in a task's cluster. A task says so with "workers" (or its track's WORKERS
    map); otherwise one that names a worker anywhere gets the three the bank was written against,
    and every other task the track's default. A VM host is a single node."""
    if is_vm(q):
        return 0
    if "workers" in q:
        return int(q["workers"])
    if "_workers" not in q:
        text = " ".join([str(q.get("task", "")), str(q.get("setup", "")), str(q.get("solution", ""))]
                        + [str(c["script"]) for c in q["checks"]])
        q["_workers"] = 3 if _WORKER_REF.search(text) else int(TRACK.get("workers", 0))
    return q["_workers"]

def host_of(q):
    """The host a task is solved on - `ssh cka0037` - which is also the name of its cluster."""
    return envs.host_of(TRACK_ID, q["id"])

def nodes_of(q):
    return envs.node_names(host_of(q), workers_of(q))

def shell_hint(q, key):
    """How a task tells you to reach a node: its host is the control plane; workers are a hop from it."""
    return f"ssh {host_of(q)}" if key == "CP" else f"ssh {nodes_of(q)[key]}"

PRELUDE = r"""
set -uo pipefail
: "${KUBECONFIG:?}"; K=kubectl               # KUBECONFIG is this task's own cluster - never your kubectl context
WS="$WORKSPACE"                            # the task's /opt/course, as a directory on this machine
PRIV="${WS%/*}/private"; mkdir -p "$PRIV"  # beside it and not mounted into the host: what a setup records for its checks
# j [kubectl get args...] '{jsonpath}'  -> prints the jsonpath result ("" on error)
j() { local n=$#; local path="${!n}"; $K get "${@:1:n-1}" -o "jsonpath=$path" 2>/dev/null; }
eq() { [ "$1" = "$2" ]; }
CP="${KSIM_CP:?}"; W1="${KSIM_W1:-}"; W2="${KSIM_W2:-}"; W3="${KSIM_W3:-}"   # the nodes of this task's cluster
# nx NODE 'command' -> run a shell command as root on a node (stdin is forwarded). The node of a VM
# host is a container that only runs QEMU: there the command goes on into the machine over ssh.
nx() {
  local n="$1"; shift
  [ -n "$n" ] || { echo "nx: this task's cluster has no such node (set \"workers\" on the task)" >&2; return 1; }
  if [ -n "${KSIM_VM:-}" ]; then command "${KSIM_RUNTIME:-podman}" exec -i "$n" ssh vm "$*"
  else command "${KSIM_RUNTIME:-podman}" exec -i "$n" sh -c "$*"; fi
}
# podman ARGS -> root's podman on the task's host: the store `sudo podman` (and the host's podman
# and docker aliases) builds into. A path under $WS becomes the host's own path for it.
podman() {
  local a q=""
  for a in "$@"; do a=${a//"$WS"//opt/course}; q+=" $(printf '%q' "$a")"; done
  command "${KSIM_RUNTIME:-podman}" exec -i "$CP" bash -c "podman$q"
}
# ensure_etcd_tools -> etcdctl and etcdutl on the control-plane node, as on the real exam. The exam
# host image carries them, at the version of the cluster's etcd (images/node/Containerfile).
ensure_etcd_tools() {
  nx "$CP" 'command -v etcdctl >/dev/null 2>&1 && command -v etcdutl >/dev/null 2>&1' </dev/null
}
# ensure_gateway_crds -> the Gateway API CRDs (standard channel), which kind does not ship.
ensure_gateway_crds() {
  $K get crd gateways.gateway.networking.k8s.io >/dev/null 2>&1 && return 0
  curl -sfL --max-time 60 https://github.com/kubernetes-sigs/gateway-api/releases/download/v1.2.1/standard-install.yaml \
    | $K apply -f - >/dev/null 2>&1 \
    && $K wait --for=condition=Established --timeout=60s crd/gatewayclasses.gateway.networking.k8s.io \
         crd/gateways.gateway.networking.k8s.io crd/httproutes.gateway.networking.k8s.io >/dev/null 2>&1
}
# ensure_metrics_server -> a working metrics API for `kubectl top`. kind ships none, and kind's
# kubelet serving certs have no IP SANs, so it only becomes Ready with --kubelet-insecure-tls.
# Pinned: v0.9.x supports Kubernetes 1.34+, so it serves the v1.37 hosts.
METRICS_SERVER_URL=https://github.com/kubernetes-sigs/metrics-server/releases/download/v0.9.0/components.yaml
ensure_metrics_server() {
  if ! $K -n kube-system get deploy metrics-server >/dev/null 2>&1; then
    curl -sfL --max-time 60 "$METRICS_SERVER_URL" | $K apply -f - >/dev/null 2>&1 || return 1
  fi
  if ! $K -n kube-system get deploy metrics-server -o jsonpath='{.spec.template.spec.containers[0].args}' | grep -q kubelet-insecure-tls; then
    $K -n kube-system patch deploy metrics-server --type=json \
      -p '[{"op":"add","path":"/spec/template/spec/containers/0/args/-","value":"--kubelet-insecure-tls"}]' >/dev/null 2>&1
  fi
  $K -n kube-system rollout status deploy metrics-server --timeout=150s >/dev/null 2>&1
}
# ensure_csi_hostpath -> the CSI hostpath driver with snapshots, and StorageClass csi-hostpath-sc
# (expandable), as the real CKA's clusters have it: what local-path cannot do - expanding a claim,
# VolumeSnapshots, cloning. Pinned to csi-driver-host-path v1.18.0 (its deploy/kubernetes-1.34
# directory, which kubernetes-1.35 and -latest link to); each sidecar's RBAC at the version of its
# image there, and the snapshot CRDs and controller of external-snapshotter at the snapshotter's.
CSI_GH=https://raw.githubusercontent.com/kubernetes-csi
ensure_csi_hostpath() {
  $K get csidriver hostpath.csi.k8s.io >/dev/null 2>&1 && $K get sc csi-hostpath-sc >/dev/null 2>&1 \
    && $K -n default get pod csi-hostpathplugin-0 -o jsonpath='{.status.containerStatuses[*].ready}' 2>/dev/null | grep -qv false && return 0
  local f s=$CSI_GH/external-snapshotter/v8.6.0
  for f in volumesnapshotclasses volumesnapshotcontents volumesnapshots; do
    curl -sfL --max-time 60 "$s/client/config/crd/snapshot.storage.k8s.io_$f.yaml" | $K apply -f - >/dev/null 2>&1 || return 1
  done
  $K wait --for condition=established --timeout=60s crd/volumesnapshotclasses.snapshot.storage.k8s.io \
    crd/volumesnapshotcontents.snapshot.storage.k8s.io crd/volumesnapshots.snapshot.storage.k8s.io >/dev/null 2>&1 || return 1
  for f in $s/deploy/kubernetes/snapshot-controller/rbac-snapshot-controller.yaml $s/deploy/kubernetes/snapshot-controller/setup-snapshot-controller.yaml \
           $CSI_GH/external-provisioner/v6.3.0/deploy/kubernetes/rbac.yaml $CSI_GH/external-attacher/v4.12.0/deploy/kubernetes/rbac.yaml \
           $s/deploy/kubernetes/csi-snapshotter/rbac-csi-snapshotter.yaml $CSI_GH/external-resizer/v2.2.1/deploy/kubernetes/rbac.yaml \
           $CSI_GH/external-health-monitor/v0.18.0/deploy/kubernetes/external-health-monitor-controller/rbac.yaml; do
    curl -sfL --max-time 60 "$f" | $K apply -f - >/dev/null 2>&1 || return 1
  done
  for f in driverinfo plugin snapshotclass; do
    curl -sfL --max-time 60 "$CSI_GH/csi-driver-host-path/v1.18.0/deploy/kubernetes-1.34/hostpath/csi-hostpath-$f.yaml" | $K apply -f - >/dev/null 2>&1 || return 1
  done
  $K apply -f - >/dev/null 2>&1 <<'CSI_SC' || return 1
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: csi-hostpath-sc
provisioner: hostpath.csi.k8s.io
reclaimPolicy: Delete
volumeBindingMode: Immediate
allowVolumeExpansion: true
CSI_SC
  $K -n kube-system rollout status deploy snapshot-controller --timeout=180s >/dev/null 2>&1 \
    && $K -n default rollout status sts csi-hostpathplugin --timeout=240s >/dev/null 2>&1
}
"""

def _session_tracks():
    """The tracks that have a session under way (a state file that has not been ended)."""
    out = []
    for t in TRACKS:
        try:
            with open(os.path.join(STATE_DIR, f"state-{t}.json")) as f:
                if not json.load(f).get("ended_at"):
                    out.append(t)
        except (OSError, ValueError):
            pass
    return out

def _pick_track_arg(argv):
    """The track a command is for: --track, else $KSIM_TRACK, else the one track with a session under
    way - after `./ksim --track ckad drill`, a plain `./ksim show 13` means that drill's task 13,
    not CKA's - else cka."""
    if "--track" in argv:
        i = argv.index("--track"); t = argv[i + 1]; del argv[i:i + 2]
        return t
    if os.environ.get("KSIM_TRACK"):
        return os.environ["KSIM_TRACK"]
    running = _session_tracks()
    return running[0] if len(running) == 1 else "cka"


load_track(_pick_track_arg(ARGV))

TTY = sys.stdout.isatty()
def c(code, s): return f"\033[{code}m{s}\033[0m" if TTY else s
def green(s): return c("32", s)
def red(s): return c("31", s)
def yellow(s): return c("33", s)
def cyan(s): return c("36", s)
def bold(s): return c("1", s)
def dim(s): return c("2", s)
class Die(SystemExit):
    """SystemExit that carries its message, so the web UI can show why a start was refused."""

def die(msg, code=1):
    print(red(msg))
    e = Die(code); e.msg = msg
    raise e

# ------------------------------------------------------------- state ----
def load_state() -> dict[str, Any] | None:
    if not os.path.exists(STATE_FILE):
        return None
    with open(STATE_FILE) as f:
        return json.load(f)

def save_state(st: dict[str, Any]) -> None:
    os.makedirs(STATE_DIR, exist_ok=True)
    with open(STATE_FILE, "w") as f:
        json.dump(st, f, indent=2)

def paused_total(st: dict[str, Any] | None) -> float:
    """Seconds spent paused so far, including the current pause if any."""
    if not st:
        return 0.0
    t = float(st.get("paused_total", 0))
    if st.get("paused_at"):
        t += time.time() - float(st["paused_at"])
    return t

def remaining(st: dict[str, Any] | None) -> float:
    if not st:
        return 0.0
    if st.get("mode") == "lesson":   # a lesson teaches: it has no clock
        return float("inf")
    return float(st.get("started_at", 0)) + float(st.get("duration", EXAM_SECONDS)) + paused_total(st) - time.time()

def set_paused(st: dict[str, Any], on: bool) -> None:
    if on and not st.get("paused_at"):
        st["paused_at"] = time.time()
    elif not on and st.get("paused_at"):
        st["paused_total"] = float(st.get("paused_total", 0)) + (time.time() - float(st["paused_at"])); st["paused_at"] = None
    save_state(st)

def restart_clock(st: dict[str, Any]) -> None:
    """The session's full time again, from now and running. Hosts, the work on them, results,
    flags and notes stay as they are."""
    st["started_at"] = time.time(); st["paused_total"] = 0.0; st["paused_at"] = None
    save_state(st)

def fmt(secs: float | int) -> str:
    secs_int = max(0, int(secs))
    return f"{secs_int // 3600:02d}:{(secs_int % 3600) // 60:02d}:{secs_int % 60:02d}"

def require_started() -> dict[str, Any]:
    st = load_state()
    if not st:
        die("No session in progress. Run:  ./ksim start   (or ./ksim drill weak)")
    assert st is not None
    return st

_TRACK_K8S: dict[str, str] = {}
def track_k8s():
    """Each track's Kubernetes minor, for the lobby's track picker (looked up once per process)."""
    if not _TRACK_K8S:
        for t in TRACKS:
            try:
                _TRACK_K8S[t] = importlib.import_module(f"{t}.questions").TRACK.get("k8s", "?")
            except Exception:  # noqa: BLE001 - a bank mid-edit must not take the other tracks' UI down
                _TRACK_K8S[t] = "?"
    return _TRACK_K8S

def session_questions(st: dict[str, Any] | None) -> list[dict[str, Any]]:
    if not st:
        return list(QUESTIONS)
    ids = st.get("ids") or [q["id"] for q in QUESTIONS]
    # Preserve the session's own order. Filtering QUESTIONS (which is sorted by id) threw it away,
    # so compose_exam's rng.shuffle never reached the candidate and every exam arrived grouped by
    # domain and difficulty - base tasks first, then hard, then the sim sets. Real exams are not
    # ordered by difficulty, and practising triage on a sorted list teaches the wrong instinct.
    by_id = {q["id"]: q for q in QUESTIONS + LESSONS}
    return [by_id[i] for i in ids if i in by_id]

def q_by_id(n: int) -> dict[str, Any]:
    for q in QUESTIONS + LESSONS:   # a lesson has a host like a task: ./ksim ssh 9201 reaches it
        if q["id"] == n:
            return q
    # ids run 1..N without gaps today, but print the real ranges rather than "1-<count>" so a
    # future gap can never make this message name ids that do not exist.
    ids = sorted(q["id"] for q in QUESTIONS)
    ranges, start, prev = [], ids[0], ids[0]
    for i in ids[1:] + [None]:
        if i != prev + 1:
            ranges.append(str(start) if start == prev else f"{start}-{prev}")
            start = i
        prev = i
    die(f"No question {n}. Valid: {', '.join(ranges)}")
    return {}

_USE_CONTEXT = re.compile(r"^.*config use-context %\{ctx\}.*\n?", re.M)

def render(text, params, q=None):
    """Substitute %{param} and {WS}, and for a task its own nodes: {CP}, {W1}.. and {SHELL_CP}.."""
    t = text.replace("{WS}", WS_DISPLAY)
    for k, v in (params or {}).items():
        t = t.replace("%{" + k + "}", str(v))
    if q is not None:
        for k, v in nodes_of(q).items():
            t = t.replace("{SHELL_" + k + "}", shell_hint(q, k)).replace("{" + k + "}", v)
    return t

def prose(text, params, q):
    """Task or solution text as the candidate reads it. Tasks written for the pre-2025 exams open
    by switching kubectl context; here every task has a host of its own instead, so that line goes."""
    return render(_USE_CONTEXT.sub("", text), params, q).strip()

def session_params(st, q):
    """Params drawn for this session; outside a session, a fixed sample so text still reads naturally."""
    p = (st or {}).get("params", {}).get(str(q["id"]))
    return p if p is not None else make_params(q, random.Random(q["id"]))

def task_text(q, params=None):
    return prose(q["task"], params if params is not None else session_params(load_state(), q), q)

def solution_text(q, params=None):
    return prose(str(q.get("solution", "")), params if params is not None else session_params(load_state(), q), q)

def make_params(q, rng):
    fn = q.get("params")
    return fn(rng) if fn else {}

# -------------------------------------------------------------- shell ----
def run(q, script, timeout=180, params=None):
    """Run a setup or check script of task q against q's own cluster."""
    host = host_of(q)
    env = dict(os.environ, KUBECONFIG=envs.kubeconfig(host), WORKSPACE=envs.course_dir(host), KSIM_ROOT=ROOT,
               KSIM_HOST=host, KSIM_RUNTIME=envs.RUNTIME, KSIM_VM="1" if is_vm(q) else "",
               **{f"KSIM_{k}": v for k, v in nodes_of(q).items()})
    p = subprocess.run(["bash", "-c", PRELUDE + render(script, params, q)], env=env, capture_output=True, text=True,
                       timeout=timeout, stdin=subprocess.DEVNULL)
    return p.returncode, (p.stdout + p.stderr).strip()

# `podman exec` as the candidate, with the environment an ssh login has: a kind node's container
# environment points KUBECONFIG at admin.conf, which only root can read and sshd never passes on.
AS_CANDIDATE = ["-u", "candidate", "-w", "/home/candidate", "-e", "KUBECONFIG=/home/candidate/.kube/config"]

def solve_on_host(q, params, timeout=600, script=None):
    """Run a task's scripted solution (or another script: a lesson step's commands) the way a
    candidate would type it: as `candidate`, in an interactive shell on the task's host, so the
    host's aliases and profile apply."""
    into = ["ssh", "candidate@vm", "bash -i"] if is_vm(q) else ["bash", "-i"]
    try:
        p = subprocess.run([envs.RUNTIME, "exec", "-i", *([] if is_vm(q) else AS_CANDIDATE), nodes_of(q)["CP"], *into],
                           input=render(q["solve"] if script is None else script, params, q), capture_output=True, text=True,
                           errors="replace", timeout=timeout)   # a script may print binary (an etcd value, say)
    except subprocess.TimeoutExpired:
        return 1, f"solution script timed out after {timeout}s"
    return p.returncode, p.stdout[-400:]

def k8s_version():
    """The Kubernetes release the track's exam hosts run, e.g. v1.37.0 (from the pinned node image)."""
    m = re.search(r":(v[\d.]+)", envs.BASE_IMAGES.get(K8S, ""))
    return m.group(1) if m else K8S

def hosts_down(st):
    """Hosts of a session that are not running - after a reboot, say - so grading them would
    record zeros for work that was never looked at."""
    return [h for h in (envs.host_of(TRACK_ID, i) for i in (st or {}).get("ids", [])) if not envs.running(h)]

def down_message(down):
    return (f"{len(down)} of this session's hosts are not running ({', '.join(down[:6])}{'...' if len(down) > 6 else ''}), "
            "so their checks would all fail and be recorded.\n"
            "  Start them again:  ./ksim hosts start     or discard the session:  ./ksim abort --yes")

def grade_one(q, params=None):
    """Run every check independently. Returns a result dict with partial credit."""
    if params is None:
        params = session_params(load_state(), q)
    checks = []
    for ch in q["checks"]:
        try:
            rc, out = run(q, ch["script"], timeout=240, params=params)
        except subprocess.TimeoutExpired:
            rc, out = 1, "timed out"
        checks.append({"name": render(ch["name"], params, q), "pts": ch["pts"], "pass": rc == 0, "out": out[-400:]})
    earned = sum(ch["pts"] for ch in checks if ch["pass"])
    return {"pass": earned == q["weight"], "earned": earned, "total": q["weight"],
            "checks": checks, "at": time.time(), "difficulty": q["difficulty"]}

def result_detail(r):
    return "\n".join(("  ✓ " if ch["pass"] else "  ✗ ") + ch["name"] + f" ({ch['pts']} pt)" for ch in r["checks"])

def score(st, results=None):
    results = results if results is not None else st["results"]
    qs = session_questions(st)
    total = sum(q["weight"] for q in qs)
    earned = sum(results.get(str(q["id"]), {}).get("earned", 0) for q in qs)
    return earned, total, (round(100 * earned / total, 1) if total else 0)

# ----------------------------------------------------------- history ----
def load_history(include_synthetic=False):
    """Attempts from history-<track>.jsonl. Entries marked "synthetic" (a scripted verification
    run, not a person sitting the tasks) stay in the file for the record but are left out of every
    statistic, weak list and "latest exam" unless explicitly asked for."""
    if not os.path.exists(HISTORY_FILE):
        return []
    with open(HISTORY_FILE) as f:
        entries = [json.loads(l) for l in f if l.strip()]
    return entries if include_synthetic else [e for e in entries if not e.get("synthetic")]

def record_attempt(st):
    earned, total, pct = score(st)
    entry = {"at": datetime.now().isoformat(timespec="minutes"), "mode": st.get("mode", "exam"), "hard": st.get("hard", False),
             "k8s": st.get("k8s"),  # server version it ran on; older entries predate per-track clusters
             "ids": [q["id"] for q in session_questions(st)], "score_pct": pct, "earned": earned, "total": total,
             "time_used": int(st["ended_at"] - st["started_at"] - paused_total(st)),
             "results": {k: {"pass": v["pass"], "earned": v["earned"], "total": v["total"],
                             "failed": [ch["name"] for ch in v["checks"] if not ch["pass"]]}
                         for k, v in st["results"].items()}}
    with open(HISTORY_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")
    write_weak_spots()
    return entry

class AggEntry:
    def __init__(self) -> None:
        self.attempts: int = 0
        self.fails: int = 0
        self.last: Any = None
        self.last_fail_at: str = ""
        self.failed: list[str] = []
        self.last_earned: str = ""
        self.streak: int = 0          # consecutive full passes at the end of the history

    def __getitem__(self, item: str) -> Any:
        return getattr(self, item)

    def __setitem__(self, item: str, value: Any) -> None:
        setattr(self, item, value)

    def to_dict(self) -> dict[str, Any]:
        return {
            "attempts": self.attempts,
            "fails": self.fails,
            "last": self.last,
            "last_fail_at": self.last_fail_at,
            "failed": self.failed,
            "last_earned": self.last_earned,
            "streak": self.streak,
        }


def stats() -> dict[int, AggEntry]:
    agg: dict[int, AggEntry] = {int(q["id"]): AggEntry() for q in QUESTIONS}
    for e in load_history():
        for k, r in e.get("results", {}).items():
            a = agg.get(int(k))
            if a is None:
                continue
            a.attempts += 1
            a.last = r.get("pass")
            a.last_earned = f"{r.get('earned', 0)}/{r.get('total', 0)}"
            a.streak = a.streak + 1 if r.get("pass") else 0
            if not r.get("pass"):
                a.fails += 1
                a.last_fail_at = str(e.get("at", ""))
                a.failed = list(r.get("failed", []))
    return agg

def last_seen():
    """question id -> index of the most recent session it appeared in (higher = more recent)."""
    seen = {}
    for i, e in enumerate(load_history()):
        for k in e["results"]:
            seen[int(k)] = i
    return seen

def drilled_domains():
    out = {d: 0 for d in DOMAINS}
    for e in load_history():
        if e["mode"] != "drill":
            continue
        for k in e["results"]:
            q = QBYID.get(int(k))
            if q:
                out[q["domain"]] += 1
    return out

def breaks_of(q):
    """A task's "breaks" as a set. It is a string or a list: Q61 damages the scheduler AND the
    controller-manager, and declaring only one let it share a session with Q38."""
    b = q.get("breaks")
    return set() if not b else ({b} if isinstance(b, str) else set(b))

def no_conflicts(cands, n, taken=()):
    """Take up to n tasks, never two that break the same thing (or clash with tasks already taken).
    A "solo" task (Q45's etcd restore rolls the WHOLE cluster back to its snapshot, wiping every
    other task's seed and answers) is only ever taken on its own."""
    used = set().union(*(breaks_of(q) for q in taken)) if taken else set()
    out = []
    for q in cands:
        if q.get("solo") and (taken or out):
            continue
        if any(x.get("solo") for x in list(taken) + out):
            break
        b = breaks_of(q)
        if b & used:
            continue
        out.append(q)
        used |= b
        if len(out) == n:
            break
    return out

def split_solo(qs):
    """Pull solo tasks out of a multi-task selection; they have to be drilled alone."""
    if len(qs) <= 1:
        return list(qs), []
    return [q for q in qs if not q.get("solo")], [q for q in qs if q.get("solo")]

def pick(domain, n, hard, rng, exclude=(), taken=(), unseen_only=False):
    """n tasks from a domain: unseen first, then least recently seen; hard sessions prefer hard tasks."""
    seen = last_seen()

    def build(allow_hard):
        pool = [q for q in POOL if q["domain"] == domain and q["id"] not in exclude and not q.get("solo")
                and (allow_hard or q["difficulty"] == "normal") and not (unseen_only and q["id"] in seen)]
        rng.shuffle(pool)
        pool.sort(key=lambda q: (0 if (hard and q["difficulty"] == "hard") else 1, seen.get(q["id"], -1)))
        return pool

    got = no_conflicts(build(hard), n, taken)
    if len(got) < n and not hard:
        # A domain whose only tasks are "hard" contributes NOTHING to a normal session and drops
        # out of the blueprint entirely - CKS SUPPLY and RUNTIME (20% each) were both unreachable
        # via ./ksim start. Better to hand over a hard task than to silently skip the domain.
        got = no_conflicts(build(True), n, taken)
    return got

_TOUCH = re.compile(r"([A-Za-z]+):(\d+)")

def touch_arg(a):
    """DOMAIN:N as (domain, n), or None if the argument is something else."""
    m = _TOUCH.fullmatch(a)
    return (m.group(1).upper(), int(m.group(2))) if m and m.group(1).upper() in DOMAINS else None

def spot_check(domain, n, hard, rng, taken=()):
    """A few tasks of one domain, spread over its objectives - to touch a domain rather than work
    through it (./ksim drill OBS:3): one task from each objective in turn, starting with the
    objectives you have solved least of, and within one the tasks you have not solved, then the
    least recently seen. Three tasks therefore come from three objectives, and the next three from
    the others."""
    taken = list(taken); out = []
    objs = [o for o in OBJECTIVES if o["domain"] == domain]
    if not objs:
        return pick(domain, n, hard, rng, exclude=[q["id"] for q in taken], taken=taken)
    agg, seen = stats(), last_seen()
    rng.shuffle(objs)
    objs.sort(key=lambda o: sum(1 for q in objective_tasks(o) if task_solved(q["id"], agg)))
    def cands(o, allow_hard):
        qs = [q for q in objective_tasks(o) if q["domain"] == domain and not q.get("solo") and q not in taken and q not in out
              and (allow_hard or q["difficulty"] == "normal")]
        rng.shuffle(qs)
        qs.sort(key=lambda q: (task_solved(q["id"], agg), seen.get(q["id"], -1)))
        return qs
    cover = {o["id"]: 0 for o in objs}   # tasks taken so far that each objective lists (a task may sit under two)
    for allow_hard in ((True,) if hard else (False, True)):   # normal tasks first; hard ones only to fill up
        while len(out) < n:
            for o in sorted(objs, key=lambda o: cover[o["id"]]):   # the objective with least so far; ties keep the order above
                got = no_conflicts(cands(o, allow_hard), 1, taken + out)
                if got:
                    out += got
                    for other in objectives_of(got[0]["id"]):
                        if other["id"] in cover:
                            cover[other["id"]] += 1
                    break
            else:
                break   # no objective has a task left
    return out

def compose_exam(hard, rng):
    quota = QUOTA_HARD if hard else QUOTA
    # Scarcest domain first. Two tasks that "breaks" the same thing can't share a session, so a
    # one-task domain loses every such tie to a bigger domain that claimed the shared slot earlier
    # and then contributes nothing: CKS RUNTIME (only Q9, apiserver-flag) was never drawn in ANY
    # mode because SETUP was iterated first and took that slot with Q2.
    order = sorted(quota, key=lambda d: len([q for q in POOL if q["domain"] == d]))
    qs, short = [], {}
    for d in order:
        got = pick(d, quota[d], hard, rng, taken=qs)
        if len(got) < quota[d]:
            short[d] = (len(got), quota[d])
        qs += got
    if short:
        # Never truncate in silence - a short session still gets scored out of its own total, so
        # the missing domains look like they were simply not on the blueprint.
        print(yellow("  warning: bank too small to fill the blueprint - "
                     + ", ".join(f"{d} {g}/{w}" for d, (g, w) in short.items())))
    rng.shuffle(qs)
    return qs

def campaign_batch(hard, rng):
    """'drill all': DRILL_SIZE tasks you have never attempted, round-robin across domains,
    hard tasks included when --hard. When everything has been seen once, falls back to weak tasks."""
    seen = last_seen(); known = load_known()
    left = [q for q in POOL if (hard or q["difficulty"] == "normal") and not known_task(q, known)]   # what you marked known is not "left"
    pool = [q for q in left if q["id"] not in seen and not q.get("solo")]
    if not pool:
        return None, []
    qs, domains = [], list(DOMAINS)
    rng.shuffle(pool)
    pool.sort(key=lambda q: 0 if (hard and q["difficulty"] == "hard") else 1)
    while len(qs) < DRILL_SIZE and pool:
        progressed = False
        for d in domains:
            cand = [q for q in pool if q["domain"] == d and q not in qs]
            got = no_conflicts(cand, 1, qs)
            if got:
                qs += got; progressed = True
                if len(qs) == DRILL_SIZE:
                    break
        if not progressed:
            break
    total = len(left)
    return f"{total - len(pool)}/{total} attempted", qs

def solution_allowed(st: dict[str, Any] | None, qid: int | str) -> bool:
    """Drill: once the task has been graded. Exam: only after submit."""
    if not st or st.get("ended_at"):
        return True
    return bool(st.get("mode") == "drill" and str(qid) in st.get("results", {}))

def cmd_solution(args):
    if not args:
        die("usage: ./ksim solution N   (drills: after grading the task; exams: after submit; --force to peek anyway)")
    force = "--force" in args; args = [a for a in args if a != "--force"]
    q = q_by_id(int(args[0])); st = load_state()
    if not solution_allowed(st, int(q["id"])) and not force:
        die("Not yet: in a drill, grade the task first (./ksim grade N); in an exam, submit first. --force overrides.")
    print(bold(f"Solution - Q{q['id']} {q['title']}")); print("-" * 64)
    print(solution_text(q, session_params(st, q))); print("-" * 64)
    print(dim(f"One correct way, not the only one. $do = --dry-run=client -o yaml, k = kubectl. Run it on the task's host: ssh {host_of(q)}"))

def cmd_progress(args):
    seen = last_seen(); agg = stats()
    print(bold(f"{'Domain':<8}{'attempted':>11}{'passed':>9}   {'hard attempted':>15}{'hard passed':>13}"))
    for d in DOMAINS:
        dq = [q for q in QUESTIONS if q["domain"] == d]
        n = [q for q in dq if q["difficulty"] == "normal"]; h = [q for q in dq if q["difficulty"] == "hard"]
        att = lambda L: sum(1 for q in L if q["id"] in seen); ok = lambda L: sum(1 for q in L if agg.get(int(q["id"])) and agg[int(q["id"])].last is True)
        print(f"{d:<8}{att(n):>6}/{len(n):<4}{ok(n):>6}/{len(n):<4}   {att(h):>9}/{len(h):<4}{ok(h):>8}/{len(h):<4}")
    left = [q["id"] for q in QUESTIONS if q["id"] not in seen]
    print(dim(f"\n{len(QUESTIONS) - len(left)}/{len(QUESTIONS)} tasks attempted at least once." + (f"  Never seen: Q{', Q'.join(map(str, left))}" if left and len(left) <= 25 else "")))
    print(dim("passed = latest attempt was a full pass.  ./ksim drill all [--hard] works through what's left."))

def is_weak(q, a):
    """Weak = failed on the latest attempt, or failed at least twice and not yet passed twice in a
    row since - a flaky task has to prove itself - or never attempted (outside the fixed sets).
    The old rule ("failed twice, ever") kept a task weak forever: 13 of 29 listed were passing."""
    if a.attempts == 0:
        return not q.get("set")
    return a.last is False or (a.fails >= 2 and a.streak < 2)

def weak_ids(agg=None):
    agg = agg or stats()
    return [q["id"] for q in QUESTIONS if is_weak(q, agg[q["id"]])]

def weak_lines(a):
    """What to show under a weak task: the failing checks if it failed last time, otherwise how
    close it is to dropping off the list - not the ✗ list of some older, since-fixed attempt."""
    if a.last is False:
        return [f"✗ {n}" for n in (a.failed or [])]
    return [f"passed last time ({a.last_earned}) - one more clean pass takes it off this list"]

def exam_domain_scores(e):
    """domain -> (earned, total) over the tasks that were actually IN that attempt. Summing the
    weight of every task in the bank instead made a 100% run read "ARCH 38/266"."""
    out = {}
    for k, r in e.get("results", {}).items():
        q = QBYID.get(int(k))
        if q:
            de, dt = out.get(q["domain"], (0, 0))
            out[q["domain"]] = (de + r.get("earned", 0), dt + r.get("total", 0))
    return out

def write_weak_spots():
    agg = stats(); hist = load_history()
    L = [f"# {TRACK_ID.upper()} weak spots", "", f"_Generated by exam from {len(hist)} attempt(s); regenerated on every `end`._", ""]
    exams = [e for e in hist if e["mode"] == "exam"]
    if exams:
        e = exams[-1]; ds = exam_domain_scores(e)
        L += [f"## Domain scores (latest full exam, {e['at']}, {e['score_pct']}%)", ""]
        for d, name in DOMAINS.items():
            if d in ds:
                L.append(f"- **{d}** {ds[d][0]}/{ds[d][1]} — {name}")
        L.append("")
    L += ["## Drill next  (`./ksim drill weak`)", ""]
    weak = [q for q in QUESTIONS if q["id"] in weak_ids(agg) and agg[q["id"]]["attempts"] > 0]
    if not weak:
        L.append("Nothing failing right now.")
    for q in weak:
        a = agg[q["id"]]
        L.append(f"### Q{q['id']:02d} · {q['domain']} · {q['title']}  ({a['fails']}/{a['attempts']} failed, last {a['last_earned']} pts, last failed {a['last_fail_at']})")
        L += [f"- {n}" for n in weak_lines(a)]
        L.append("")
    L += ["## Attempt matrix", "", "| Q | Domain | Title | " + " | ".join(f"{e['mode'][0].upper()}{i + 1}" for i, e in enumerate(hist)) + " |",
          "|---|---|---|" + "---|" * len(hist)]
    for q in QUESTIONS:
        cells = []
        for e in hist:
            r = e["results"].get(str(q["id"]))
            cells.append("·" if r is None else ("✓" if r["pass"] else f"{r['earned']}/{r['total']}"))
        L.append(f"| {q['id']} | {q['domain']} | {q['title']} | " + " | ".join(cells) + " |")
    L += ["", "_E = full exam, D = drill, · = not in that session, n/m = partial credit._", ""]
    with open(WEAK_FILE, "w") as f:
        f.write("\n".join(L))

# --------------------------------------------------------- curriculum ----
# A track may ship tracks/<id>/curriculum.py: the exam's objectives in order, each with study
# notes and the tasks that practise it. It turns the bank into a path - drill one objective at a
# time against the clock until every task passes - and `./ksim objectives` shows where you are.
def find_objective(arg):
    """An objective by id (3.2) or key (probes, or any unambiguous start of it); None if none fits."""
    a = str(arg).lower()
    for o in OBJECTIVES:
        if a in (o["id"], o["key"]):
            return o
    hits = [o for o in OBJECTIVES if len(a) >= 3 and o["key"].startswith(a)]
    return hits[0] if len(hits) == 1 else None

def objective_tasks(o, with_sets=False):
    """An objective's tasks in its own order. Tasks of the fixed simulator sets stay out unless
    asked for: meeting them one objective at a time would spoil the mock they are kept for."""
    return [QBYID[i] for i in o["tasks"] if i in QBYID and (with_sets or not QBYID[i].get("set"))]

def objectives_of(qid):
    return [o for o in OBJECTIVES if qid in o["tasks"]]

def objective_domains():
    """Domain codes in the order the curriculum lists them (TRACK["domains"] may order them otherwise)."""
    return list(dict.fromkeys(o["domain"] for o in OBJECTIVES))

def load_focus():
    """Domains you put at the front of the path with ./ksim focus, in that order."""
    return load_progress()["focus"]

def path_domains():
    """Domain codes in the order the path walks them: your focus first (./ksim focus - the domains
    a real exam report or a mock said are weakest), then the track's STUDY_ORDER (CKA leaves
    Troubleshooting for last - it builds on all the rest), then the curriculum's own order."""
    return list(dict.fromkeys([*load_focus(), *(d for d in STUDY_ORDER if d in DOMAINS), *objective_domains()]))

def path_objectives():
    """Objectives in the order the path walks them: by path_domains(), curriculum order within a domain."""
    order = path_domains()
    return sorted(OBJECTIVES, key=lambda o: order.index(o["domain"]))   # stable sort

def objective_session(o, with_sets=False, n=None):
    """The tasks of one sitting on an objective: what has not yet passed against the clock first,
    at most n, never two that break the same thing. A big objective therefore takes several
    sittings, each on new ground."""
    agg = stats()
    qs = objective_tasks(o, with_sets)
    todo = [q for q in qs if task_state(q["id"], agg) != "passed"] or qs     # nothing open: go over all of it again
    return no_conflicts(todo, n or DRILL_SIZE), len(todo)

def load_progress():
    """study-<track>.json, next to the history so it survives ./teardown.sh as the history does:
      objectives  objective id -> when it was first done
      focus       domains the path walks first, in that order (./ksim focus)
      known       domains and objectives you declared known, which the path leaves out (./ksim skip)
      lessons     objective id -> when its lesson was first done (every step checked)
      tasks       what the study sessions of earlier versions solved - kept as it was, read by nothing.
    Results against the clock live in the history; this file only steers the path."""
    try:
        with open(PROGRESS_FILE) as f:
            d = json.load(f)
    except (OSError, ValueError):
        d = {}
    oids = {o["id"] for o in OBJECTIVES}
    return {"tasks": dict(d.get("tasks", {})), "objectives": dict(d.get("objectives", {})),
            "focus": [x for x in d.get("focus", []) if x in DOMAINS],
            "known": [x for x in dict.fromkeys(d.get("known", [])) if x in DOMAINS or x in oids],
            "lessons": dict(d.get("lessons", {}))}

def save_progress(prog):
    with open(PROGRESS_FILE, "w") as f:
        json.dump({**({"tasks": prog["tasks"]} if prog.get("tasks") else {}),
                   "objectives": {k: prog["objectives"][k] for k in sorted(prog["objectives"])},
                   **({"focus": prog["focus"]} if prog.get("focus") else {}),
                   **({"known": prog["known"]} if prog.get("known") else {}),
                   **({"lessons": {k: prog["lessons"][k] for k in sorted(prog["lessons"])}} if prog.get("lessons") else {})}, f, indent=1)
        f.write("\n")

def load_known():
    """What you declared known with ./ksim skip: domain codes and objective ids, in the order you
    marked them. A statement, not a result - it steers the path and nothing else."""
    return load_progress()["known"]

def is_known(o, known):
    """An objective you declared known, by itself or with its whole domain."""
    return o["domain"] in known or o["id"] in known

def known_task(q, known):
    """A task the path no longer brings up: its domain is marked known, or every objective that
    practises it is."""
    if q["domain"] in known:
        return True
    objs = objectives_of(q["id"])
    return bool(objs) and all(is_known(o, known) for o in objs)

# ------------------------------------------------------------ lessons ----
# A lesson (<track>/lessons.py) is a guided lab on one objective: an introduction, steps that you do
# on a host of your own - each with a check, a hint and the commands that do it - and a wrap-up that
# sends you to the objective's timed drill. It runs as a session of mode "lesson": no clock, its
# steps checked one at a time and in order, nothing written to the history or the statistics; a
# lesson whose every step was checked is stamped done in study-<track>.json.
def find_lesson(arg):
    """A lesson by its objective (2.1, or the objective's key) or by its id (9201); None if none fits."""
    o = find_objective(arg)
    return next((l for l in LESSONS if (o and l["objective"] == o["id"]) or str(l["id"]) == str(arg)), None)

def lesson_of(st):
    """The lesson a session is, or None."""
    if not st or st.get("mode") != "lesson":
        return None
    return next((l for l in LESSONS if l["id"] in st.get("ids", [])), None)

def lessons_done():
    """objective id -> when its lesson was first done."""
    return load_progress()["lessons"]

def check_step(l, params, i):
    """Run the check of step i on the lesson's host."""
    try:
        rc, out = run(l, l["steps"][i]["check"], timeout=240, params=params)
    except subprocess.TimeoutExpired:
        rc, out = 1, "timed out"
    return {"pass": rc == 0, "out": out[-600:], "at": time.time()}

def open_step(st, l):
    """The first step not checked yet, or None when every one is."""
    done = st.get("steps", {})
    return next((i for i in range(len(l["steps"])) if not done.get(str(i), {}).get("pass")), None)

def lesson_check(st, i):
    """Check step i of the lesson in progress and keep the result. Steps count in order - a step
    builds on the one before - so only the first open step or one already done can be checked, and
    a pass stays a pass when a later step changes what an earlier check looked at.
    Returns (result, None) or (None, why not)."""
    l = lesson_of(st)
    if not l or st.get("ended_at"):
        return None, "No lesson in progress."
    if not 0 <= i < len(l["steps"]):
        return None, f"The lesson has steps 1 to {len(l['steps'])}."
    first = open_step(st, l)
    if first is not None and i > first:
        return None, f"Do step {first + 1} first: steps count in order."
    r = check_step(l, session_params(st, l), i)
    steps = st.setdefault("steps", {})
    if r["pass"] or not steps.get(str(i), {}).get("pass"):
        steps[str(i)] = r
    save_state(st)
    return r, None

def task_state(qid, agg):
    """passed = the latest attempt against the clock was a full pass; failed = it was not;
    new = never attempted."""
    a = agg.get(int(qid))
    if a is not None and a.attempts:
        return "passed" if a.last else "failed"
    return "new"

def task_solved(qid, agg):
    """Passed against the clock at least once, if not necessarily on the latest attempt."""
    a = agg.get(int(qid))
    return bool(a is not None and a.attempts > a.fails)

MARKS = {"passed": ("✓", green), "failed": ("✗", red), "new": ("·", dim), "started": ("-", yellow), "known": ("»", cyan)}

def mark(state):
    ch, col = MARKS[state]
    return col(ch)

def objective_state(o, agg, known=None):
    """Where an objective is:
      done     every pool task passed against the clock on its latest attempt - exam conditions
    before that: started (something attempted) or new. And beside them:
      known    you said you are good with it (./ksim skip), so the path leaves it out. Proof still
               outranks the statement: a known objective you did complete shows as done."""
    qs = objective_tasks(o)
    states = [task_state(q["id"], agg) for q in qs]
    if qs and all(x == "passed" for x in states):
        return "done"
    if is_known(o, load_known() if known is None else known):
        return "known"
    return "started" if any(x != "new" for x in states) else "new"

OBJECTIVE_MARK = {"done": "passed", "started": "started", "new": "new", "known": "known"}

def next_objective(agg=None):
    """The first objective in path order that is neither done nor marked known - drilled against
    the clock until every task passes. None when nothing is left."""
    agg = stats() if agg is None else agg
    known = load_known()
    return next((o for o in path_objectives() if objective_state(o, agg, known) not in ("done", "known")), None)

def mock_scores():
    """Scores of the full mocks and of each fixed simulator set, from the history."""
    full, fixed = [], {k: [] for k in SETS}
    members = {k: {q["id"] for q in set_questions(k)} for k in SETS}
    for e in load_history():
        if e["mode"] != "exam":
            continue
        hit = next((k for k in SETS if set(e["ids"]) == members[k]), None)
        (fixed[hit] if hit else full).append(e["score_pct"])
    return full, fixed

def next_step(agg=None):
    """The next move on the path to the exam, as {kind, cmd, why[, objective | set]}:
    drill an objective against the clock -> ... every objective -> a full mock -> the fixed
    simulator sets. Each stage is closer to the real exam than the one before."""
    agg = stats() if agg is None else agg
    o = next_objective(agg)
    if o:
        return {"kind": "drill", "objective": o, "cmd": f"./ksim drill {o['id']}",
                "why": f"{o['id']} {o['text']} - its tasks against the clock, no solution until graded"}
    full, fixed = mock_scores()
    if not any(x >= PASS_PCT for x in full):
        skipped = any(is_known(x, load_known()) for x in OBJECTIVES)
        return {"kind": "mock", "cmd": "./ksim start",
                "why": f"every objective is done{' or marked known' if skipped else ''} - a full mock: {sum(QUOTA.values())} tasks, "
                       f"{TRACK['minutes']} minutes, no feedback until you submit"
                       + (" (it draws from every domain, the known ones included)" if skipped else "")}
    for k in SETS:
        if not any(x >= PASS_PCT for x in fixed[k]):
            return {"kind": "sim", "set": k, "cmd": f"./ksim sim {k}",
                    "why": f"{SETS[k]}: {len(set_questions(k))} fixed tasks, {TRACK['minutes']} minutes, exam rules"}
    return {"kind": "done", "cmd": "./ksim start --hard",
            "why": f"every objective done and every mock passed at {PASS_PCT}% or better - keep sharp with a fresh mock"}

def update_milestones():
    """Stamp each objective the first time it is done, and return the objectives newly done.
    Called whenever results are recorded, so the checklist keeps the date you got there."""
    if not OBJECTIVES:
        return []
    prog = load_progress(); agg = stats(); new = []
    now = datetime.now().isoformat(timespec="minutes")
    for o in OBJECTIVES:
        if objective_state(o, agg, prog["known"]) == "done" and "done" not in prog["objectives"].get(o["id"], {}):
            prog["objectives"].setdefault(o["id"], {})["done"] = now
            new.append(o)
    if new:
        save_progress(prog)
    return new

def milestone_lines(new):
    return [f"✓ Objective {o['id']} {o['key']} complete - {len(objective_tasks(o))}/{len(objective_tasks(o))} tasks passed against the clock."
            for o in new]

def curriculum_gaps():
    """What would make the path lie: a task no objective lists, a listed id that does not exist,
    an objective too thin to drill."""
    listed = {i for o in OBJECTIVES for i in o["tasks"]}
    out = []
    orphans = [q["id"] for q in QUESTIONS if q["id"] not in listed]
    if orphans:
        out.append(f"not listed under any objective: Q{', Q'.join(map(str, orphans))}")
    ghosts = sorted(listed - set(QBYID))
    if ghosts:
        out.append(f"listed but not in the bank: Q{', Q'.join(map(str, ghosts))}")
    thin = [o["id"] for o in OBJECTIVES if len(objective_tasks(o)) < 2]
    if thin:
        out.append(f"fewer than two pool tasks: {', '.join(thin)}")
    wrong = [f"{o['id']} ({o['domain']})" for o in OBJECTIVES if o["domain"] not in DOMAINS]
    if wrong:
        out.append(f"unknown domain: {', '.join(wrong)}")
    return out

def require_objectives():
    if not OBJECTIVES:
        die(f"Track '{TRACK_ID}' has no objective map (tracks/{TRACK_ID}/curriculum.py).\n"
            f"  Drill it by domain instead:  ./ksim drill {next(iter(DOMAINS))}")

def cmd_objectives(args):
    """The curriculum as a checklist: every objective, its tasks, and how far you are."""
    require_objectives()
    agg, known = stats(), load_known()
    cols = shutil.get_terminal_size((120, 24)).columns
    sets = len(QUESTIONS) - len(POOL)
    print(bold(f"{TRACK['name']} - path") + dim(f"   {len(OBJECTIVES)} objectives · {len(POOL)} tasks"
                                                 + (f" · {sets} more in the fixed sets [in brackets]" if sets else "")))
    print(dim("tasks: ✓ passed against the clock  ✗ failed last time  · not yet"
              "      objectives: ✓ complete  - started  » marked known, off the path"))
    def cell(q):
        return f"{MARKS[task_state(q['id'], agg)][0]}{q['id']}", f"{mark(task_state(q['id'], agg))}{q['id']}"
    rows = []
    for o in OBJECTIVES:
        pool = [cell(q) for q in objective_tasks(o)]
        fixed = [cell(q) for q in objective_tasks(o, True) if q.get("set")]
        plain = " ".join(c[0] for c in pool) + (" [" + " ".join(c[0] for c in fixed) + "]" if fixed else "")
        shown = " ".join(c[1] for c in pool) + (dim(" [") + " ".join(c[1] for c in fixed) + dim("]") if fixed else "")
        rows.append((o, plain, shown))
    tw = max(28, min(max(len(o["text"]) for o in OBJECTIVES), cols - 26 - max(len(r[1]) for r in rows)))
    focus = load_focus()
    for d in objective_domains():
        print(bold(f"\n{DOMAINS[d]}") + (yellow(f"   focus {focus.index(d) + 1}") if d in focus else "")
              + (cyan(f"   marked known - off the path (./ksim unskip {d} brings it back)") if d in known else ""))
        for o, plain, shown in rows:
            if o["domain"] != d:
                continue
            state = objective_state(o, agg, known)
            text = o["text"] if len(o["text"]) <= tw else o["text"][:tw - 1] + "…"
            print(f" {mark(OBJECTIVE_MARK[state])} {o['id']:<4}{o['key']:<16}{text:<{tw}}  {shown}")
    print_path(agg)
    for g in curriculum_gaps():
        print(yellow(f"curriculum.py: {g}"))

def print_path(agg=None):
    """The ladder from one objective at a time to exam conditions, how far up you are, and the next rung."""
    agg = stats() if agg is None else agg
    known = load_known()
    states = [objective_state(o, agg, known) for o in OBJECTIVES]
    skipped = states.count("known")
    n, done = len(OBJECTIVES) - skipped, states.count("done")
    aside = dim(f"  + {skipped} marked known") if skipped else ""
    full, fixed = mock_scores()
    step = next_step(agg)
    def best(scores):
        if not scores:
            return dim("not sat yet")
        b = max(scores)
        return (green if b >= PASS_PCT else red)(f"best {b}%") + dim(f" of {len(scores)}, pass mark {PASS_PCT}%")
    rows = [("drill", "Clock", "each objective's tasks, timed, graded per check", f"{done}/{n} objectives{aside}")]
    rows.append(("mock", "Mock exam", f"{sum(QUOTA.values())} random tasks, {TRACK['minutes']} min, no feedback until submit", best(full)))
    rows += [("sim", f"Set {k.upper()}", f"{len(set_questions(k))} fixed tasks, {TRACK['minutes']} min, exam rules", best(fixed[k])) for k in SETS]
    print(bold("\nPath to the exam"))
    for i, (kind, name, what, where) in enumerate(rows, 1):
        here = step["kind"] == kind and (kind != "sim" or name == f"Set {step['set'].upper()}")
        print(f" {green('→') if here else ' '} {i}  {name:<11}{dim(f'{what:<54}')}{where}")
    focus = load_focus()
    if focus:
        print(dim(f"\nFocus: {', '.join(focus)} first, then the rest of the curriculum   (./ksim focus shows it, ./ksim focus clear undoes it)"))
    if known:
        print(dim(f"\nKnown: {', '.join(known)} - left out of the path because you said so   (./ksim skip shows it, ./ksim unskip X brings X back)"))
    print(bold(f"\nNext:  {step['cmd']}") + dim("   (or just ./ksim next)"))
    print(dim(f"       {step['why']}"))
    if step.get("objective"):
        print(dim(f"       ./ksim notes {step['objective']['id']}   what to know, the fastest commands, the traps, the docs to open"))

def cmd_next(args):
    """Start whatever is next on the path: drill an objective, a mock, a simulator set."""
    require_objectives()
    step = next_step()
    print(bold(f"Next on the path:  {step['cmd']}") + dim(f"\n  {step['why']}"))
    if step["kind"] == "drill":
        cmd_drill([step["objective"]["id"]] + args)
    elif step["kind"] == "sim":
        cmd_sim([step["set"]] + args)
    else:
        cmd_start((["--hard"] if step["kind"] == "done" else []) + args)

def cmd_focus(args):
    """Put domains at the front of the path - the ones an exam report or a mock said are weakest:
    ./ksim focus DEPLOY NET ENV, ./ksim focus clear, ./ksim focus (show). Nothing else changes: a
    mock still samples every domain, and any objective can still be drilled at any time."""
    prog = load_progress(); before = list(prog["focus"])
    if args:
        if args[0].lower() in ("clear", "none", "off"):
            prog["focus"] = []
        else:
            want = list(dict.fromkeys(a.upper() for a in args))
            bad = [a for a in want if a not in DOMAINS]
            if bad:
                die(f"'{bad[0]}' is not a domain of the {TRACK_ID.upper()} track. Domains:\n  "
                    + "\n  ".join(f"{d:<8}{name}" for d, name in DOMAINS.items()) + "\n  ./ksim focus clear   back to the usual order")
            prog["focus"] = want
        if prog["focus"] != before:
            save_progress(prog)
    focus = prog["focus"]
    if not focus:
        print("No focus set - the path follows " + ("the track's study order." if STUDY_ORDER else "the curriculum's order.")
              + dim(f"\n  ./ksim focus {' '.join(list(DOMAINS)[:2])}   puts those domains first"))
        return
    agg = stats()
    print(bold("Focus") + dim("   these domains come first on the path, in this order"))
    for i, d in enumerate(focus, 1):
        objs = [o for o in OBJECTIVES if o["domain"] == d]
        states = [objective_state(o, agg) for o in objs]
        pool = [q for q in POOL if q["domain"] == d]
        solved = sum(1 for q in pool if task_solved(q["id"], agg))
        passed = sum(1 for q in pool if task_state(q["id"], agg) == "passed")
        where = f"objectives {states.count('done')}/{len(objs)} done · " if objs else ""
        print(f" {i}  {DOMAINS[d]:<58}" + dim(f"{where}tasks {solved}/{len(pool)} passed once, {passed}/{len(pool)} on the latest try"))
    if OBJECTIVES:
        step = next_step(agg)
        print(bold(f"\nNext:  {step['cmd']}") + dim("   (or just ./ksim next)") + dim(f"\n       {step['why']}"))
    else:
        print(dim("\n./ksim drill with no argument picks these first among the domains you have drilled least."))

def _known_what(x):
    """A known entry as text: a domain with its objectives and tasks, or one objective."""
    if x in DOMAINS:
        objs = [o for o in OBJECTIVES if o["domain"] == x]
        return f"{x:<8}{DOMAINS[x]} - {len(objs)} objective(s), {len([q for q in POOL if q['domain'] == x])} pool tasks"
    o = find_objective(x)
    return f"{o['id']:<8}{o['text']}  ({o['domain']})"

def set_known(mark=(), unmark=(), clear=False):
    """Change what is marked known and return (known, added, removed). Entries are domain codes or
    objective ids; a domain covers its objectives, so marking it drops their own entries, and
    unmarking one objective of a known domain leaves the others of that domain marked."""
    prog = load_progress(); before = list(prog["known"]); known = [] if clear else list(before)
    for x in unmark:
        if x in known:
            known.remove(x)
        elif x not in DOMAINS:
            o = find_objective(x)
            if o and o["domain"] in known:   # one objective back on the path: its siblings stay known
                i = known.index(o["domain"])
                known[i:i + 1] = [p["id"] for p in OBJECTIVES if p["domain"] == o["domain"] and p["id"] != o["id"]]
    for x in mark:
        if x in DOMAINS:
            known = [k for k in known if k == x or k in DOMAINS or find_objective(k)["domain"] != x]
            if x not in known:
                known.append(x)
        elif x not in known and find_objective(x)["domain"] not in known:
            known.append(x)
    prog["known"] = known
    prog["focus"] = [d for d in prog["focus"] if d not in known]   # a domain cannot be first on the path and off it
    if known != before or len(prog["focus"]) != len(load_focus()):
        save_progress(prog)
    return known, [k for k in known if k not in before], [k for k in before if k not in known]

def _known_args(args, cmd):
    out = []
    for a in args:
        if a.upper() in DOMAINS:
            out.append(a.upper())
        elif find_objective(a):
            out.append(find_objective(a)["id"])
        else:
            die(f"'{a}' is not a domain or an objective of the {TRACK_ID.upper()} track. Domains:\n  "
                + "\n  ".join(f"{d:<8}{name}" for d, name in DOMAINS.items())
                + f"\n  objectives: ./ksim objectives      e.g.  ./ksim {cmd} {next(iter(DOMAINS))}   ./ksim {cmd} {OBJECTIVES[0]['id'] if OBJECTIVES else '1.1'}")
    return list(dict.fromkeys(out))

def print_known(known):
    if not known:
        print("Nothing is marked known - the path walks the whole curriculum."
              + dim(f"\n  ./ksim skip {next(iter(DOMAINS))}   leaves that domain out because you are good with it"))
        return
    agg = stats()
    print(bold("Marked known") + dim("   left out of the path because you said so - a statement, not a result"))
    for x in known:
        pool = [q for q in POOL if (q["domain"] == x if x in DOMAINS else q["id"] in find_objective(x)["tasks"])]
        solved = sum(1 for q in pool if task_solved(q["id"], agg))
        passed = sum(1 for q in pool if task_state(q["id"], agg) == "passed")
        print(f" {cyan('»')} {_known_what(x)}" + dim(f"   on record: {solved}/{len(pool)} tasks passed once, {passed}/{len(pool)} on the latest try"))
    print(dim("  Nothing is written to the history, the statistics or the weak list. A mock exam still draws from every\n"
              "  domain, as the real one will, and ./ksim drill X still works for anything.\n"
              "  ./ksim unskip X   puts X back on the path      ./ksim skip clear   puts everything back"))

def cmd_skip(args):
    """Mark domains or objectives as known - "I am good with this" - so the path leaves them out:
    ./ksim skip DESIGN, ./ksim skip 2.3 kustomize, ./ksim skip (show), ./ksim skip clear."""
    if args and args[0].lower() in ("clear", "none", "off"):
        known, _, removed = set_known(clear=True)
        if removed:
            print(f"Back on the path: {', '.join(removed)}")
    elif args:
        known, added, _ = set_known(mark=_known_args(args, "skip"))
        for x in added:
            print(f"Marked known:  {_known_what(x)}")
    else:
        known = load_known()
    print_known(known)
    if OBJECTIVES and (args or known):
        step = next_step()
        print(bold(f"\nNext:  {step['cmd']}") + dim("   (or just ./ksim next)") + dim(f"\n       {step['why']}"))

def cmd_unskip(args):
    """Put domains or objectives you marked known back on the path: ./ksim unskip DESIGN 2.3, ./ksim unskip all."""
    if not args:
        die("usage: ./ksim unskip DOMAIN|OBJECTIVE..   or   ./ksim unskip all      (./ksim skip shows what is marked known)")
    if args[0].lower() == "all":
        known, _, removed = set_known(clear=True)
    else:
        known, _, removed = set_known(unmark=_known_args(args, "unskip"))
    print(f"Back on the path: {', '.join(removed)}" if removed else "Nothing changed - that was not marked known.")
    print_known(known)
    if OBJECTIVES:
        step = next_step()
        print(bold(f"\nNext:  {step['cmd']}") + dim("   (or just ./ksim next)") + dim(f"\n       {step['why']}"))

def paint_code(text):
    """`code` in a note: highlighted on a terminal, left as written (it is Markdown) in a pipe."""
    return re.sub(r"`([^`]+)`", lambda m: bold(m.group(1)), text) if TTY else text

def print_notes(o, agg=None):
    agg = stats() if agg is None else agg
    w = min(shutil.get_terminal_size((100, 24)).columns, 104)
    def bullets(items):
        for it in items:
            print(paint_code("\n".join(textwrap.wrap(it, w, initial_indent="  - ", subsequent_indent="    "))))
    print(bold(f"{o['id']} {o['key']}") + dim(f"  ·  {DOMAINS[o['domain']]}"))
    print(bold(o["text"])); print("-" * 64)
    print(bold("Know")); bullets(o["know"])
    print(bold("\nFast")); print("\n".join("  " + line for line in o["fast"].strip("\n").splitlines()))
    print(bold("\nTraps")); bullets(o["traps"])
    print(bold("\nDocs") + dim("  (pages you may open during the exam)"))
    for u in o["docs"]:
        print("  " + u)
    print(bold("\nTasks"))
    for q in objective_tasks(o, True):
        print(f"  {mark(task_state(q['id'], agg))} Q{q['id']:<3} {q['title']}" + (dim(f"   [{SETS.get(q['set'], q['set'])}]") if q.get("set") else ""))
    print("-" * 64)

def notes_markdown(objs):
    """The same notes as a Markdown document (./ksim notes all --md > CKAD_STUDY_GUIDE.md)."""
    L = [f"# {TRACK_ID.upper()} study guide", "",
         f"The {len(OBJECTIVES)} objectives of the {TRACK['name']} curriculum, in its own words and order: what to know, "
         "the fastest commands, the traps, the documentation pages the exam lets you open, and the tasks in this "
         "simulator that practise each one.", "",
         f"_Generated by `./ksim --track {TRACK_ID} notes all --md` from `tracks/{TRACK_ID}/curriculum.py` - edit that file, not this one._", "",
         "```", f"./ksim objectives        # the path and your progress", "./ksim notes 3.2         # one objective's notes",
         "./ksim drill 3.2         # its tasks against the clock", "```", ""]
    for d in objective_domains():
        mine = [o for o in objs if o["domain"] == d]
        if not mine:
            continue
        L += [f"## {DOMAINS[d]}", ""]
        for o in mine:
            L += [f"### {o['id']} {o['text']}", "", f"`./ksim drill {o['id']}` (or `{o['key']}`)", "", "**Know**", ""]
            L += [f"- {x}" for x in o["know"]]
            L += ["", "**Fast**", "", "```", o["fast"].strip("\n"), "```", "", "**Traps**", ""]
            L += [f"- {x}" for x in o["traps"]]
            L += ["", "**Docs**", ""] + [f"- <{u}>" for u in o["docs"]]
            L += ["", "**Tasks**", ""]
            L += [f"- Q{q['id']} {q['title']}" + (f" _({SETS.get(q['set'], q['set'])})_" if q.get("set") else "") for q in objective_tasks(o, True)]
            L.append("")
    return "\n".join(L)

def cmd_notes(args):
    """Study notes for an objective, a domain or the whole curriculum."""
    require_objectives()
    rest = [a for a in args if not a.startswith("--")]
    if not rest:
        st = load_state()
        o = find_objective(st.get("objective")) if st and not st.get("ended_at") and st.get("objective") else None
        objs = [o or next_objective() or OBJECTIVES[0]]
    elif rest[0] == "all":
        objs = list(OBJECTIVES)
    elif rest[0].upper() in DOMAINS:
        objs = [o for o in OBJECTIVES if o["domain"] == rest[0].upper()]
    else:
        objs = [find_objective(a) for a in rest]
        if None in objs:
            die(f"'{rest[objs.index(None)]}' is not an objective. Ids and keys:\n  "
                + "\n  ".join(f"{o['id']:<5}{o['key']:<16}{o['text']}" for o in OBJECTIVES))
    if "--md" in args:
        print(notes_markdown(objs)); return
    agg = stats()
    for i, o in enumerate(objs):
        if i:
            print()
        print_notes(o, agg)
    if len(objs) == 1:
        print(dim(f"./ksim drill {objs[0]['id']}   its tasks against the clock"))

# ------------------------------------------------------------ session ----
def preflight(qs):
    """Everything a session needs before any cluster is made: the tools, the images, the memory."""
    for tool, why in (("podman", "runs the exam hosts and the desktop"), ("kind", "builds each task's cluster"),
                      ("kubectl", "seeds and grades the tasks"), ("ssh-keygen", "makes the key the desktop logs in with")):
        if not shutil.which(tool):
            die(f"'{tool}' is not installed - it {why}.")
    missing = [i for i in dict.fromkeys([node_image(q) for q in qs if not is_vm(q)] + [envs.DESKTOP_IMAGE]) if not envs.image_exists(i)]
    if missing:
        die(f"Not built yet: {', '.join(missing)}\n  Build the exam host and desktop images (once, a few minutes):  ./ksim build")
    if any(is_vm(q) for q in qs) and not (envs.vm_ready(K8S) and envs.image_exists(envs.VM_IMAGE)):
        die(f"Not built yet: the VM exam host, which {sum(is_vm(q) for q in qs)} of these tasks need.\n"
            f"  Build it (once, about ten minutes):  ./ksim build {TRACK_ID} --vm")
    need, have = sum(envs.memory_mb(workers_of(q), is_vm(q)) for q in qs), envs.available_mb()
    nodes = sum(1 + workers_of(q) for q in qs)
    print(dim(f"  track: {TRACK_ID}  Kubernetes {k8s_version()}  {len(qs)} task(s) = {len(qs)} cluster(s), {nodes} node(s), about {need / 1024:.1f} GB"))
    if have is not None and need > have - 1536 and not os.environ.get("KSIM_FORCE"):
        die(f"These {len(qs)} clusters need about {need / 1024:.1f} GB and {have / 1024:.1f} GB is available.\n"
            "  Free some memory, pick fewer tasks, or set KSIM_FORCE=1 to try anyway.")
    for tool, why in (("helm", "seeding the Helm tasks"), ("openssl", "seeding the certificate tasks"), ("curl", "seeding tasks that fetch manifests")):
        if not shutil.which(str(tool)):
            print(yellow(f"  warning: '{tool}' not found on this machine - needed for {why}"))

def _log_msg(*args: Any, **kwargs: Any) -> None:
    print(*args, **kwargs)


def provision(q, params, log: Callable[..., Any] = _log_msg):
    """One task, start to finish: its cluster, its scenario, and the hand-over to `candidate`.
    Returns None, or what went wrong."""
    host, tag = host_of(q), (f"lesson {q['objective']} {q['title']}" if q.get("lesson") else f"Q{q['id']:02d} {q['title']}")
    try:
        make_host(q)
    except envs.EnvError as e:
        log(f"  ! {host}  {tag}\n{dim(str(e)[-500:])}")
        return f"{host}: no cluster"
    try:
        rc, out = run(q, q["setup"], params=params, timeout=300)
    except subprocess.TimeoutExpired:
        rc, out = 1, "setup timed out after 300s"
    # a task that says "the current namespace" (ctx_ns) starts with its kubeconfig pointing there
    envs.hand_over(host, (params or {}).get("ns") if q.get("ctx_ns") else None)
    log(f"  {'✓' if rc == 0 else '!'} {host}  {tag}" + ("" if rc == 0 else f"\n{dim(out[-300:])}"))
    return None if rc == 0 else f"{host}: setup failed"

def seed(qs, log: Callable[..., Any] = _log_msg, params_by_id=None):
    """Provision and seed every task of a session, PARALLEL clusters at a time. Tasks cannot
    disturb each other any more, so there is no order to respect."""
    params_by_id = params_by_id or {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=PARALLEL) as pool:
        problems = [p for p in pool.map(lambda q: provision(q, params_by_id.get(str(q["id"])), log), qs) if p]
    return problems

def audits_running():
    """PIDs of `./ksim audit` runs in progress. An audit and a session share the host names, and
    starting a session deletes every host there is - so a session waits for an audit to finish."""
    pids = []
    for name in os.listdir(STATE_DIR) if os.path.isdir(STATE_DIR) else []:
        m = re.match(r"audit-(\d+)\.lock$", name)
        if not m:
            continue
        if os.path.isdir(f"/proc/{m.group(1)}") and int(m.group(1)) != os.getpid():
            pids.append(int(m.group(1)))
        elif int(m.group(1)) != os.getpid():
            os.remove(os.path.join(STATE_DIR, name))   # left by an audit that was killed
    return pids

def clear_hosts(keep=()):
    """Delete the exam hosts left by earlier sessions (they stay up after a session for review)."""
    old = [h for h in envs.existing() if h not in keep]
    with concurrent.futures.ThreadPoolExecutor(max_workers=PARALLEL) as pool:
        list(pool.map(envs.destroy, old))
    return old

def drop_hosts(st):
    """Delete the hosts of one session's tasks and nothing else - the browser's Terminate session
    and Delete the hosts, which must not reach another track's session."""
    mine = {host_of(q) for q in session_questions(st)}
    gone = [h for h in envs.existing() if h in mine]
    with concurrent.futures.ThreadPoolExecutor(max_workers=PARALLEL) as pool:
        list(pool.map(envs.destroy, gone))
    return gone

def open_sessions_elsewhere():
    """Tracks other than this one with a session that has not ended. One session at a time: they
    share the desktop, and a new session clears the hosts of the last one."""
    out = []
    for t in TRACKS:
        if t == TRACK_ID:
            continue
        try:
            with open(os.path.join(STATE_DIR, f"state-{t}.json")) as f:
                if not json.load(f).get("ended_at"):
                    out.append(t)
        except (OSError, ValueError):
            pass
    return out

def begin_session(qs, mode, duration, force, log: Callable[..., Any] = _log_msg, hard=False, other_ok=None, objective=None):
    st = load_state()
    if st and not st.get("ended_at") and not force:
        die("A session is already in progress. './ksim end' it first, or add --force.")
    others = open_sessions_elsewhere()
    if others:
        t = others[0]
        die(f"A {t.upper()} session is still open, and its hosts would be deleted by this one.\n"
            f"  Finish it first:  ./ksim --track {t} end   (or  ./ksim --track {t} abort --yes  to discard it)"
            + ("" if other_ok is None else "  - in the browser: switch to it in the lobby and submit."))
    if not qs:
        die(f"Track '{TRACK_ID}' has no questions yet.")
    running = audits_running()
    if running:
        die(f"A bank audit is running (./ksim audit, pid {running[0]}). Starting a session deletes every host, the audit's included.\n"
            "  Let it finish, or stop it, then start the session.")
    preflight(qs)
    update_milestones()   # stamp what the history already proves, so this session only announces its own
    rng = random.Random()
    params_by_id = {str(q["id"]): make_params(q, rng) for q in qs}
    hosts = [host_of(q) for q in qs]
    if not envs.ensure_registry():
        log(yellow("  warning: the image cache (ksim-registry) is not running - hosts pull straight from Docker Hub, which allows 100 pulls an hour"))
    old = clear_hosts()
    if old:
        log(dim(f"  removed {len(old)} host(s) of the previous session"))
    # a fresh desktop for every session, as on the day: no shell history, no open tabs
    desktop.up(fresh=True)
    desktop.configure(TRACK_ID, hosts, open_web=bool(os.environ.get("KSIM_OPEN_WEB")))
    log(bold(f"Preparing {len(qs)} exam host(s){' [HARD]' if hard else ''} - one cluster per task, {PARALLEL} at a time..."))
    problems = seed(qs, log, params_by_id)
    if any("no cluster" in p for p in problems):
        clear_hosts()
        die("Could not create every host (see above), so the session was not started.\n"
            "  Usual causes: not enough memory, or podman's limits - try fewer tasks or KSIM_PARALLEL=3.")
    st = {"mode": mode, "track": TRACK_ID, "hard": hard, "ids": [q["id"] for q in qs],
          "params": params_by_id, "duration": duration, "started_at": time.time(), "ended_at": None,
          "k8s": k8s_version(), "results": {}, "flags": [], "notes": "",
          "objective": objective}   # the curriculum objective a drill or a lesson was started for, if any
    lesson = mode == "lesson"
    if lesson:
        st["steps"] = {}   # step index -> its latest check (a pass sticks)
    save_state(st)
    log("")
    log(bold("=" * 64))
    log(bold(f"  {TRACK_ID.upper()} {'LESSON' if lesson else 'EXAM' if mode == 'exam' else 'DRILL'}{' · HARD' if hard else ''} - "
             + ("no clock: take your time" if lesson else "clock started")))
    log(f"  Hosts   : {hosts[0]}" + (f" .. {len(hosts)} in all" if len(hosts) > 1 else "")
        + f"  (Kubernetes {st['k8s']}, as the real {TRACK_ID.upper()} runs {K8S})")
    if lesson:
        log(f"  Lesson  : {objective} {qs[0]['title']} - {len(qs[0]['steps'])} steps")
    else:
        log(f"  Ends    : {datetime.fromtimestamp(st['started_at'] + duration).strftime('%H:%M:%S')}  ({duration // 60} min)")
        log(f"  Tasks   : {', '.join('Q%02d' % q['id'] for q in qs) if mode == 'drill' else len(qs)}   Pass mark: {PASS_PCT}%")
    if problems:
        log(yellow(f"  Seeding had problems on: {', '.join(problems)}  -  ./ksim reset N  rebuilds one task's host"))
    log(bold("=" * 64))
    log("  ./ksim web         " + ("the lesson in the browser: its steps beside the remote desktop\n" if lesson else
                                   "the exam in the browser: task pane, remote desktop, clock\n")
        + ("  ./ksim lesson step / check / hint / show, ./ksim end   terminal workflow\n" if lesson else
           "  ./ksim list / show N / grade N / timer / end   terminal workflow\n")
        + f"  {'The lesson runs on' if lesson else 'Every task names its host:'}  ssh {hosts[0]}  from the desktop's terminal, then  sudo -i  when you need root.\n"
        + f"  Files go under {WS_DISPLAY}/ on the host.")
    return st

def finish_lesson(st):
    """End a lesson: nothing is graded or recorded in the history; a lesson whose every step was
    checked is stamped done on its objective (the first time). Returns the lines to show."""
    l = lesson_of(st)
    st["ended_at"] = time.time()
    n, done = len(l["steps"]), sum(1 for i in range(len(l["steps"])) if st.get("steps", {}).get(str(i), {}).get("pass"))
    if done == n:
        prog = load_progress()
        prog["lessons"].setdefault(l["objective"], datetime.now().isoformat(timespec="minutes"))
        save_progress(prog)
        lines = [f"✓ Lesson {l['objective']} done - {n}/{n} steps checked. Now prove it against the clock: ./ksim drill {l['objective']}"]
    else:
        lines = [f"Lesson {l['objective']} finished with {done}/{n} steps checked - take it again any time: ./ksim lesson {l['objective']}"]
    st["milestones"] = lines
    save_state(st)
    return lines

def finish_session(st, log: Callable[..., Any] = _log_msg):
    if st.get("mode") == "lesson":
        finish_lesson(st)
        return None
    qs = session_questions(st)
    over = remaining(st) <= 0
    set_paused(st, False)
    log(bold(f"Submitting - grading {len(qs)} task(s)..."))
    # tasks are graded side by side: each one's checks only ever touch its own cluster
    with concurrent.futures.ThreadPoolExecutor(max_workers=PARALLEL) as pool:
        graded = list(pool.map(lambda q: grade_one(q, session_params(st, q)), qs))
    for q, r in zip(qs, graded):
        st["results"][str(q["id"])] = r
        tag = green("PASS") if r["pass"] else (yellow(f"{r['earned']}/{r['total']}") if r["earned"] else red("FAIL"))
        log(f"Q{q['id']:02d} {tag}  {q['title']}" + ("" if r["pass"] else f"\n{dim(result_detail(r))}"))
    st["ended_at"] = time.time()
    st["over_time"] = over
    save_state(st)
    entry = record_attempt(st)
    # objectives this session got over a line: shown by ./ksim end and by the results page
    st["milestones"] = st.get("milestones", []) + milestone_lines(update_milestones())
    save_state(st)
    return entry

# ----------------------------------------------------------- commands ----
def cmd_start(args):
    hard = "--hard" in args; force = "--force" in args
    known = load_known()   # a domain you marked known does not have to be drilled first
    undrilled = [d for d, n in drilled_domains().items() if n == 0 and d not in known]
    if undrilled and not force:
        die("Drill every domain at least once before a full exam (that is the point of the drills).\n"
            f"  Not yet drilled: {', '.join(undrilled)}\n  Run:  ./ksim drill      (picks the next domain)\n"
            "  or:   ./ksim start --force   to skip the gate.")
    begin_session(compose_exam(hard, random.Random()), "exam", EXAM_SECONDS, force, hard=hard)

def next_drill(hard, rng):
    """No-arg drill: the least-drilled domain, DRILL_SIZE tasks, least-recently-seen first."""
    cov = drilled_domains()
    order = list(dict.fromkeys([*load_focus(), *DOMAINS]))   # equally drilled: your focus first
    known = load_known()
    doms = [d for d in DOMAINS if d not in known] or list(DOMAINS)   # not the domains you marked known
    domain = min(doms, key=lambda d: (cov[d], order.index(d)))
    return domain, pick(domain, DRILL_SIZE, hard, rng)

def cmd_drill(args):
    force = "--force" in args; hard = "--hard" in args
    args = [a for a in args if a not in ("--force", "--hard")]
    minutes = None
    if "--time" in args:
        i = args.index("--time"); minutes = int(args[i + 1]); del args[i:i + 2]
    per_task = DRILL_SECONDS_PER_TASK_HARD if hard else DRILL_SECONDS_PER_TASK
    if not args:
        domain, qs = next_drill(hard, random.Random())
        cov = drilled_domains()
        print(bold(f"Drill: {DOMAINS[domain]}") + dim(f"  ({sum(1 for v in cov.values() if v)}/{len(DOMAINS)} domains drilled so far)"))
        begin_session(qs, "drill", (minutes * 60) if minutes else per_task * len(qs), force, hard=hard)
        return
    if args[0] == "all":
        label, qs = campaign_batch(hard, random.Random())
        if not qs:
            print(bold("Every task in the bank has been attempted." ) + " Switching to weak tasks.")
            qs, _ = split_solo([q for q in QUESTIONS if q["id"] in weak_ids()])
            qs = no_conflicts(qs, DRILL_SIZE)
            if not qs:
                die("Nothing weak either - you have passed everything on its latest attempt. Time for ./ksim start --hard.")
        else:
            print(bold(f"Campaign: {label}") + dim(f"  this batch: Q{', Q'.join(str(q['id']) for q in qs)}"))
        begin_session(qs, "drill", (minutes * 60) if minutes else per_task * len(qs), force, hard=hard)
        return
    objective = None
    if args[0] == "weak":
        ids = weak_ids()
        if not ids:
            die("Nothing marked weak - finish a session with './ksim end' first, or name tasks: ./ksim drill 2 17")
    else:
        ids = []
        for a in args:
            if a.lower().replace("set", "") in SETS and a.lower().startswith("set"):
                ids += [q["id"] for q in set_questions(a.lower().replace("set", ""))]
            elif touch_arg(a):
                d, k = touch_arg(a)
                ids += [q["id"] for q in spot_check(d, k, hard, random.Random(), taken=[q for q in QUESTIONS if q["id"] in ids])]
            elif a.upper() in DOMAINS:
                ids += [q["id"] for q in QUESTIONS if q["domain"] == a.upper()]
            elif a.isdigit():
                ids.append(q_by_id(int(a))["id"])
            elif find_objective(a):
                sitting, open_n = objective_session(find_objective(a))
                ids += [q["id"] for q in sitting]
                objective = objective or find_objective(a)["id"]
                if open_n > len(sitting):
                    print(dim(f"  {find_objective(a)['id']}: {len(sitting)} of its {open_n} open tasks in this sitting - run it again for the rest."))
            else:
                die(f"'{a}' is not a task id, a domain ({', '.join(DOMAINS)}; {next(iter(DOMAINS))}:3 = three of its tasks), a set "
                    f"({', '.join('set' + k for k in SETS) or 'none'})"
                    + (" or an objective (./ksim objectives)." if OBJECTIVES else "."))
    ids = sorted(set(ids))
    qs, solo = split_solo([q for q in QUESTIONS if q["id"] in ids])
    if solo:
        print(yellow(f"Dropped Q{', Q'.join(str(q['id']) for q in solo)}: restoring it rolls the whole cluster back and would "
                     f"wipe the other tasks. Drill it on its own: ./ksim drill {solo[0]['id']}"))
    kept = no_conflicts(qs, len(qs))
    if len(kept) < len(qs):
        dropped = [q["id"] for q in qs if q not in kept]
        print(yellow(f"Dropped Q{', Q'.join(map(str, dropped))}: they break the same thing as another task in this set. Drill them separately."))
    # Shuffle AFTER resolving conflicts, so which task gets dropped stays deterministic while the
    # order you meet them in does not. A drill listed by id walks you from easy to hard in
    # lockstep, which is the one ordering the real exam will never give you.
    random.Random().shuffle(kept)
    begin_session(kept, "drill", (minutes * 60) if minutes else per_task * len(kept), force, hard=hard,
                  objective=objective if len(args) == 1 else None)

def cli_md(text):
    """A lesson's Markdown for a terminal: fenced blocks indented, `code` highlighted."""
    out, fence = [], False
    for line in text.strip("\n").split("\n"):
        if line.strip().startswith("```"):
            fence = not fence
            continue
        out.append("    " + line if fence else paint_code(line))
    return "\n".join(out)

def print_step(l, params, i, st=None):
    s = l["steps"][i]
    done = bool(st and st.get("steps", {}).get(str(i), {}).get("pass"))
    print(bold(f"Step {i + 1} of {len(l['steps'])}: {render(s['title'], params, l)}") + (green("   ✓ checked") if done else ""))
    print(cli_md(render(s["text"], params, l)))
    print(dim("  ./ksim lesson check   when it is done        ./ksim lesson hint / show   if you are stuck"))

def print_outro(l, params):
    print(bold("Every step is checked.")); print("-" * 64)
    print(cli_md(render(l["outro"], params, l))); print("-" * 64)
    print(dim("  ./ksim end   finishes the lesson"))

def cmd_lesson(args):
    """Lessons (<track>/lessons.py): guided labs on a host of your own, with no clock.
    ./ksim lesson | lesson OBJ | lesson step|check|hint|show [N] | lesson audit [OBJ..]"""
    sub = args[0] if args else ""
    if sub == "audit":
        return lesson_audit(args[1:])
    st = load_state()
    live = lesson_of(st) if st and not st.get("ended_at") else None
    if sub in ("check", "step", "hint", "show"):
        if not live:
            die("No lesson in progress.  ./ksim lesson   lists them;  ./ksim lesson 2.1   starts one.")
        params = session_params(st, live)
        num = [a for a in args[1:] if a.isdigit()]
        i = int(num[0]) - 1 if num else open_step(st, live)
        if i is None:
            return print_outro(live, params)
        if not 0 <= i < len(live["steps"]):
            die(f"The lesson has steps 1 to {len(live['steps'])}.")
        s = live["steps"][i]
        if sub == "step":
            return print_step(live, params, i, st)
        if sub in ("hint", "show"):
            print(bold(f"{'Hint' if sub == 'hint' else 'The commands'} - step {i + 1}: {render(s['title'], params, live)}"))
            text = render(s[sub], params, live)
            print((cli_md(text) if sub == "hint" else text) or dim("Nothing for this step."))
            return
        r, why = lesson_check(st, i)
        if why:
            die(why)
        if not r["pass"]:
            print(red(f"✗ Step {i + 1} is not done yet: {render(s['title'], params, live)}"))
            if r["out"]:
                print(dim("  " + r["out"][-300:].replace("\n", "\n  ")))
            print(dim("  ./ksim lesson hint   a hint        ./ksim lesson show   the commands"))
            return
        print(green(f"✓ Step {i + 1} done: {render(s['title'], params, live)}")); print()
        nxt = open_step(load_state(), live)
        return print_outro(live, params) if nxt is None else print_step(live, params, nxt)
    if sub:
        l = find_lesson(sub)
        if not l:
            die(f"No lesson for '{sub}'. The {TRACK_ID.upper()} lessons: "
                + (", ".join(f"{x['objective']} ({x['title']})" for x in LESSONS) or "none yet") + "   (./ksim lesson lists them)")
        st = begin_session([l], "lesson", 0, "--force" in args, objective=l["objective"])
        params = session_params(st, l)
        print(); print(bold(f"Lesson {l['objective']} - {l['title']}")); print("-" * 64)
        print(cli_md(render(l["intro"], params, l))); print("-" * 64)
        print_step(l, params, 0, st)
        return
    if not LESSONS:
        print(f"No lessons for {TRACK_ID.upper()} yet.")
        return
    done = lessons_done()
    print(bold(f"{TRACK['name']} - lessons") + dim("   guided labs on a host of your own, with no clock"))
    for l in LESSONS:
        print(f" {green('✓') if l['objective'] in done else dim('·')} {l['objective']:<5} {l['title']}"
              + dim(f"   {len(l['steps'])} steps · {DOMAINS.get(l['domain'], l['domain'])}"))
    print(dim(f"\n  ./ksim lesson {LESSONS[0]['objective']}   starts that lesson      ./ksim lesson audit   proves them on real clusters"))

def lesson_audit(args):
    """Prove lessons on real clusters: each is seeded on a host of its own; then every step must
    fail its check, have its commands run as `candidate`, and pass its check - so a lesson can be
    done exactly as it shows, and no step is done before you do it."""
    for t in TRACKS:
        try:
            with open(os.path.join(STATE_DIR, f"state-{t}.json")) as f:
                if not json.load(f).get("ended_at"):
                    die(f"A {t.upper()} session is open, and a new session would delete the audit's hosts. End it first.")
        except (OSError, ValueError):
            pass
    names = [a for a in args if not a.startswith("--")]
    ls = list(LESSONS) if not names or names == ["all"] else [find_lesson(n) for n in names]
    if not ls or None in ls:
        die(f"usage: ./ksim lesson audit [all|OBJECTIVE..] [--keep]   (lessons: {', '.join(l['objective'] for l in LESSONS) or 'none'})")
    preflight(ls[:PARALLEL])
    envs.ensure_registry()
    lock = os.path.join(STATE_DIR, f"audit-{os.getpid()}.lock")   # keeps sessions from starting meanwhile
    os.makedirs(STATE_DIR, exist_ok=True)
    open(lock, "w").close()
    import atexit
    atexit.register(lambda: os.path.exists(lock) and os.remove(lock))
    print(bold(f"Lesson audit of {len(ls)} {TRACK_ID.upper()} lesson(s), {PARALLEL} at a time - each step must fail, be done by its own commands, then pass."))
    def one(l):
        params = make_params(l, random.Random(l["id"])); host = host_of(l); t0 = time.time()
        res = {"id": l["id"], "objective": l["objective"], "title": l["title"], "host": host, "steps": len(l["steps"]), "proven": 0}
        try:
            make_host(l)
        except envs.EnvError as e:
            res.update(verdict="no cluster", detail=str(e)[-400:], secs=int(time.time() - t0))
            return res
        try:
            rc, out = run(l, l["setup"], params=params, timeout=300)
        except subprocess.TimeoutExpired:
            rc, out = 1, "setup timed out after 300s"
        envs.hand_over(host, None)
        if rc != 0:
            res.update(verdict="seed failed", detail=out[-400:])
        else:
            for i, s in enumerate(l["steps"]):
                if check_step(l, params, i)["pass"]:
                    res.update(verdict=f"step {i + 1} passes before its commands")
                    break
                _, sout = solve_on_host(l, params, script=s["show"])
                # what the commands created may still be starting: a learner would watch it come up
                r, waited = check_step(l, params, i), 0
                while not r["pass"] and waited < 90:
                    time.sleep(10); waited += 10
                    r = check_step(l, params, i)
                if not r["pass"]:
                    res.update(verdict=f"step {i + 1} fails after its commands", detail=(r["out"] + "\n--- the commands printed:\n" + sout)[-600:])
                    break
                res["proven"] = i + 1
            else:
                res["verdict"] = "ok"
        if "--keep" not in args:
            envs.destroy(host)
        res["secs"] = int(time.time() - t0)
        return res
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=PARALLEL) as pool:
        for fut in concurrent.futures.as_completed([pool.submit(one, l) for l in ls]):
            res = fut.result(); results.append(res)
            col = green if res["verdict"] == "ok" else red
            print(f"  {col(res['verdict']):<44} {res['objective']:<5}{res['proven']}/{res['steps']} steps  {res['secs']:>4}s  {res['title']}", flush=True)
            if res.get("detail"):
                print(dim("      " + res["detail"].replace("\n", "\n      ")), flush=True)
    results.sort(key=lambda r: r["id"])
    out_dir = os.path.join(STATE_DIR, "audit"); os.makedirs(out_dir, exist_ok=True)
    path = os.path.join(out_dir, f"lessons-{TRACK_ID}.json")
    with open(path, "w") as f:
        json.dump({"k8s": k8s_version(), "at": datetime.now().isoformat(timespec="minutes"), "results": results}, f, indent=1)
    bad = [r for r in results if r["verdict"] != "ok"]
    print(bold(f"\n{len(results) - len(bad)}/{len(results)} lessons proven step by step.") + (red(f"  {len(bad)} need work: " + ", ".join(r["objective"] for r in bad)) if bad else ""))
    print(dim(f"  report: {os.path.relpath(path, ROOT)}"))

def cmd_sim(args):
    """A fixed simulator set, full exam rules and clock."""
    force = "--force" in args
    names = [a.lower() for a in args if not a.startswith("--")]
    if not names or names[0] not in SETS:
        die(f"usage: ./ksim sim {'|'.join(SETS) or '<set>'} [--force]\n  "
            + "\n  ".join(f"{k}: {v} ({len(set_questions(k))} tasks)" for k, v in SETS.items()))
    name = names[0]
    qs = set_questions(name)
    print(bold(f"{SETS[name]} - {len(qs)} tasks, {TRACK['minutes']} minutes, exam rules (no feedback until you submit)"))
    begin_session(qs, "exam", EXAM_SECONDS, force, hard=True)

def cmd_list(args):
    st = load_state()
    results = st["results"] if st and "results" in st else {}
    qs = session_questions(st) if st and not st.get("ended_at") else list(QUESTIONS)
    agg = stats()
    print(bold(f"{'#':>3}  {'Domain':<8}{'Pts':>4}  {'Now':<8} {'History':<9} Title"))
    for q in qs:
        r = results.get(str(q["id"]))
        now = "-" if r is None else (green("PASS") if r.get("pass") else yellow(f"{r.get('earned')}/{r.get('total')}"))
        a = agg.get(int(q["id"])) or AggEntry()
        hist = "" if a.attempts == 0 else f"{a.attempts - a.fails}/{a.attempts}"
        if a.attempts and a.last is False:
            hist = red(hist + " ✗")
        print(f"{q['id']:>3}  {q['domain']:<8}{q['weight']:>4}  {now:<17} {hist:<18} {q['title']}{'  [hard]' if q['difficulty'] == 'hard' else ''}")
    print(dim("\nHistory = full passes/attempts across sessions, ✗ = failed most recently."))
    if st and not st.get("ended_at"):
        print(status_line(st))

def cmd_show(args):
    if not args:
        die("usage: ./ksim show N")
    q = q_by_id(int(args[0]))
    st0 = load_state()
    sq = session_questions(st0) if st0 and not st0.get("ended_at") else []
    pos = f"{sq.index(q) + 1} of {len(sq)}  |  " if q in sq else ""
    print(bold(f"{pos}Q{q['id']}  |  {DOMAINS[q['domain']]}  |  {q['weight']} pts{'  |  HARD' if q['difficulty'] == 'hard' else ''}"))
    print(bold(q["title"])); print(f"Solve this question on:  ssh {host_of(q)}")
    # The host's name only resolves on the simulator's own network, and only while the task is in a session.
    if q in sq:
        print(dim(f"  (in the desktop's terminal - from this one: ./ksim --track {TRACK_ID} ssh {q['id']})"))
    else:
        others = [t for t in _session_tracks() if t != TRACK_ID]
        print(dim(f"  (no {TRACK_ID.upper()} session holds this task, so that host does not exist yet: ./ksim --track {TRACK_ID} drill {q['id']} starts one"
                  + (f" - the session under way is {others[0].upper()}'s: ./ksim --track {others[0]} show N" if others else "") + ")"))
    print("-" * 64); print(task_text(q)); print("-" * 64)
    st = load_state()
    if st and not st.get("ended_at"):
        print(status_line(st))

def cmd_grade(args):
    st = require_started()
    force = "--force" in args; args = [a for a in args if a != "--force"]
    if not args:
        die("usage: ./ksim grade N | all")
    # The web UI already refused this; the CLI handed out per-check feedback mid-exam, which a sim
    # meant to rehearse the real thing (no feedback until you submit) must not do by default.
    if st.get("mode") == "exam" and not st.get("ended_at") and not force:
        die("No feedback during an exam - the real one gives none. Submit with ./ksim end (--force to peek anyway).")
    if st.get("mode") == "lesson":
        die("A lesson is checked step by step:  ./ksim lesson check")
    ids = [q["id"] for q in session_questions(st)] if args[0] == "all" else [int(a) for a in args]
    for n in ids:
        q = q_by_id(n); r = grade_one(q)
        st["results"][str(n)] = r
        tag = green("PASS") if r.get("pass") else yellow(f"{r.get('earned')}/{r.get('total')}")
        print(f"Q{n:02d} {tag}  {q['title']}"); print(dim(result_detail(r)))
    save_state(st)
    if args[0] == "all":
        print_score(st)

def print_score(st):
    qs = session_questions(st)
    earned, total, pct = score(st)
    print(); print(bold("Per-domain:"))
    for d, name in DOMAINS.items():
        dq = [q for q in qs if q["domain"] == d]
        if not dq:
            continue
        dt = sum(q["weight"] for q in dq); de = sum(st["results"].get(str(q["id"]), {}).get("earned", 0) for q in dq)
        n = int(20 * de / dt)
        print(f"  {d:<7} {'█' * n}{'░' * (20 - n)} {de:>3}/{dt:<3} {dim(name)}")
    verdict = green("PASS") if pct >= PASS_PCT else red("FAIL")
    print(f"\n{bold('Score:')} {earned}/{total} = {pct}%  ->  {verdict} (pass mark {PASS_PCT}%)")

def status_line(st):
    l = lesson_of(st)
    if l:
        n = len(l["steps"]); done = n if open_step(st, l) is None else open_step(st, l)
        return green("lesson - no clock") + dim(f"  [{done}/{n} steps checked]")
    r = remaining(st)
    if st.get("paused_at"):
        return yellow(f"⏸  PAUSED - {fmt(r)} remaining") + dim("  ./ksim resume")
    if r <= 0:
        return red("⏰ TIME IS UP - run ./ksim end")
    col = green if r > 30 * 60 else (yellow if r > 10 * 60 else red)
    return col(f"⏱  {fmt(r)} remaining") + dim(f"  [{st.get('mode', 'exam')}]")

def cmd_status(args):
    st = require_started()
    if st.get("ended_at"):
        print("Session ended."); print_score(st); return
    print(status_line(st))
    print(dim(f"{sum(1 for v in st['results'].values() if v['pass'])} passed / {len(st['results'])} graded / {len(session_questions(st))} in session"))

def no_clock(st):
    if st.get("mode") == "lesson":
        die("A lesson has no clock - take your time.  (./ksim lesson step shows where you are)")

def cmd_timer(args):
    st = require_started()
    if st.get("ended_at"):
        die("Session already ended.")
    no_clock(st)
    try:
        while True:
            st = load_state(); r = remaining(st)
            if st.get("paused_at"):
                sys.stdout.write("\r" + yellow(bold(f"⏸  {fmt(r)} PAUSED   ")) + dim("   ./ksim resume        ")); sys.stdout.flush(); time.sleep(1); continue
            if r <= 0:
                print("\r" + red(bold("⏰  00:00:00  TIME IS UP - run ./ksim end       "))); break
            col = green if r > 30 * 60 else (yellow if r > 10 * 60 else red)
            sys.stdout.write("\r" + col(bold(f"⏱  {fmt(r)} remaining")) + dim("   (Ctrl-C to hide)")); sys.stdout.flush()
            if int(r) in (30 * 60, 15 * 60, 5 * 60) and TTY:
                sys.stdout.write("\a")
            time.sleep(1)
    except KeyboardInterrupt:
        print()

def cmd_pause(args):
    st = require_started()
    if st.get("ended_at"):
        die("Session already ended.")
    no_clock(st)
    set_paused(st, True); print(status_line(st))

def cmd_resume(args):
    st = require_started()
    no_clock(st)
    set_paused(st, False); print(status_line(st))

def cmd_restart(args):
    """Start the clock again from the session's full time, keeping its hosts and your work. Like
    pause, not recorded anywhere: the attempt's time used counts from the restart."""
    st = require_started()
    if st.get("ended_at"):
        die("Session already ended - nothing to restart. Start a new one: ./ksim drill, ./ksim start, ./ksim next")
    no_clock(st)
    restart_clock(st)
    print(status_line(st) + dim("   clock restarted - the hosts and your work are as they were"))

def cmd_end(args):
    st = require_started()
    if st.get("ended_at"):
        die("Session already ended and recorded.")
    if st.get("mode") == "lesson":   # nothing to grade: the steps were checked as you went
        for line in finish_lesson(st):
            print(green(line) if line.startswith("✓") else line)
        print(dim("A lesson writes nothing to the history or the scores. Its host stays up until the next session."))
        return
    down = hosts_down(st)
    if down:
        die(down_message(down))
    if remaining(st) <= 0:
        print(yellow("Note: time expired before submission."))
    entry = finish_session(st)
    print_score(st)
    def path_note():
        if not OBJECTIVES:
            return
        print()
        for line in st.get("milestones", []):
            print(green(line))
        step = next_step()
        print(bold(f"Next on the path:  {step['cmd']}") + dim(f"   {step['why']}"))
    print(dim(f"Time used: {fmt(st['ended_at'] - st['started_at'] - paused_total(st))}" + ("  (over time)" if st.get("over_time") else "")))
    failed = sorted(int(k) for k, v in entry["results"].items() if not v["pass"])
    print(dim(f"Attempt #{len(load_history())} logged to {os.path.basename(HISTORY_FILE)}; {os.path.basename(WEAK_FILE)} updated."))
    if failed:
        print(bold("\nTo drill what you missed:  ./ksim drill weak") + dim(f"   (Q{', Q'.join(map(str, failed))})"))
    path_note()

def cmd_abort(args):
    """End a session without grading it, and delete its hosts."""
    st = load_state()
    if not st or st.get("ended_at"):
        die("No session in progress.")
    assert st is not None
    if "--yes" not in args:
        die(f"This discards the {st.get('mode', 'exam')} in progress without grading or recording it.\n"
            "  Confirm:  ./ksim abort --yes        (to grade and keep it instead: ./ksim end)")
    st["ended_at"] = time.time(); st["aborted"] = True
    save_state(st)
    gone = clear_hosts()
    print(yellow("Session discarded. Nothing was graded or written to history."))
    print(dim(f"  Its {len(gone)} host(s) are deleted."))

def cmd_weak(args):
    agg = stats(); hist = load_history()
    if not hist:
        die("No history yet - finish a session with './ksim end'.")
    ids = [i for i in weak_ids(agg) if agg[i]["attempts"] > 0]
    print(bold(f"Weak spots after {len(hist)} attempt(s):"))
    if not ids:
        print(green("  none - everything attempted has passed on its latest try"))
    for i in ids:
        q = q_by_id(i); a = agg[i]
        print(f"\n{red('Q%02d' % i)} {q['domain']:<7} {q['title']}  " + dim(f"{a['fails']}/{a['attempts']} failed, last {a['last_earned']} pts, last failed {a['last_fail_at']}"))
        for n in weak_lines(a):
            print(dim("    " + n))
    never = [q["id"] for q in QUESTIONS if agg[q["id"]]["attempts"] == 0]
    if never:
        print(dim(f"\nNever attempted: Q{', Q'.join(map(str, never))}"))
    print(dim(f"\nFull log: {WEAK_FILE}     Drill them: ./ksim drill weak"))

def cmd_history(args):
    hist = load_history(include_synthetic=True)
    if not hist:
        die("No history yet.")
    print(bold(f"{'#':>3}  {'when':<17}{'mode':<7}{'score':>7}  {'time':<9} tasks"))
    for i, e in enumerate(hist, 1):
        col = green if e["score_pct"] >= PASS_PCT else red
        tasks = f"all {len(QUESTIONS)}" if len(e["ids"]) == len(QUESTIONS) else "Q" + ",".join(map(str, e["ids"]))
        tag = dim("  [scripted verification - not counted]") if e.get("synthetic") else ""
        print(f"{i:>3}  {e['at']:<17}{(e['mode'] + ('*' if e.get('hard') else '')):<7}{col('%5.1f%%' % e['score_pct']):>16}  {fmt(e['time_used']):<9} {tasks}{tag}")
    exams = [e for e in hist if e["mode"] == "exam" and not e.get("synthetic")]
    print(dim("* = hard"))
    if len(exams) >= 2:
        print(bold("\nDomain trend (full exams, % of that exam's points in the domain):"))
        for d in DOMAINS:
            cells = []
            for e in exams:
                de, dt = exam_domain_scores(e).get(d, (0, 0))
                cells.append(f"{round(100 * de / dt):>3}%" if dt else "   -")
            print(f"  {d:<7} " + " → ".join(cells))

def cmd_reset(args):
    """Throw a task's cluster away and build it again, scenario included."""
    if not args:
        die("usage: ./ksim reset N   (rebuild one task's host from scratch: new cluster, scenario seeded again)")
    st = require_started(); q = q_by_id(int(args[0]))
    print(dim(f"  rebuilding {host_of(q)} - about a minute..."))
    problem = provision(q, session_params(st, q))
    print(("rebuilt " if not problem else "problem rebuilding ") + f"Q{q['id']:02d}: {q['title']}")

# --------------------------------------------------------------- web ----
class Api(http.server.BaseHTTPRequestHandler):
    """Serves web/index.html plus a small JSON API. State lives in the same files the CLI uses."""
    lock = threading.Lock()
    log_lines = []
    seeding = False
    seed_error = None
    desktop_ok = False
    desktop_error = None

    def log_message(self, format: str, *args: Any) -> None:  # quiet
        pass

    def _json(self, obj, code=200):
        body = json.dumps(obj).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)

    def _read_json(self):
        n = int(self.headers.get("Content-Length") or 0)
        return json.loads(self.rfile.read(n) or b"{}")

    def do_GET(self):
        p = urllib.parse.urlparse(self.path)
        if p.path in ("/", "/index.html"):
            with open(os.path.join(ROOT, "web", "index.html"), "rb") as f:
                body = f.read()
            self.send_response(200); self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body); return
        if p.path == "/api/session":
            with Api.lock:
                return self._json(self.session_payload())
        if p.path == "/api/history":
            return self._json({"history": load_history(), "weak": weak_ids(), "stats": {str(k): v.to_dict() for k, v in stats().items()}})
        self._json({"error": "not found"}, 404)

    def do_POST(self):
        p = urllib.parse.urlparse(self.path); body = self._read_json()
        with Api.lock:
            st = load_state()
            if p.path == "/api/track":
                t = body.get("track")
                if t not in TRACKS:
                    return self._json({"error": f"unknown track {t}"}, 400)
                if Api.seeding:
                    return self._json({"error": "a session is being seeded"}, 409)
                load_track(t)
                Api.log_lines = []
                return self._json(self.session_payload())
            if p.path == "/api/known":   # mark a domain or an objective as known, or put it back on the path
                x = str(body.get("mark", ""))
                if x not in DOMAINS and not any(o["id"] == x for o in OBJECTIVES):
                    return self._json({"error": f"unknown domain or objective {x}"}, 400)
                set_known(mark=[x]) if body.get("known", True) else set_known(unmark=[x])
                return self._json(self.session_payload())
            if p.path == "/api/clip":   # a value clicked in a task lands on the desktop's clipboard
                desktop.clip(str(body.get("text", ""))[:20000])
                return self._json({"ok": True})
            if p.path == "/api/start":
                if Api.seeding:
                    return self._json({"error": "already seeding"}, 409)
                mode = body.get("mode", "exam"); hard = bool(body.get("hard")); rng = random.Random()
                Api.log_lines = []; Api.seed_error = None; Api.seeding = True
                ansi = __import__("re").compile(r"\x1b\[[0-9;]*m")
                ulog = lambda line: Api.log_lines.append(ansi.sub("", line))
                def work():
                    try:
                        if body.get("lesson"):
                            l = find_lesson(body["lesson"])
                            if not l:
                                die(f"No lesson for {body['lesson']}.")
                            begin_session([l], "lesson", 0, True, log=ulog, other_ok=False, objective=l["objective"])
                        elif body.get("sim") in SETS:
                            begin_session(set_questions(body["sim"]), "exam", EXAM_SECONDS, True, log=ulog, hard=True, other_ok=False)
                        elif mode == "exam":
                            begin_session(compose_exam(hard, rng), "exam", EXAM_SECONDS, True, log=ulog, hard=hard, other_ok=False)
                        else:
                            # touch: [[domain, n]..] - a few tasks of a domain, on top of the rest or on their own
                            touch = [(str(t[0]).upper(), int(t[1])) for t in (body.get("touch") or []) if str(t[0]).upper() in DOMAINS]
                            objective = find_objective(body["objective"]) if body.get("objective") else None
                            if objective:
                                qs, _ = objective_session(objective, bool(body.get("all")))
                            elif body.get("domain") in DOMAINS:
                                qs = pick(body["domain"], DRILL_SIZE, hard, rng)
                            elif body.get("campaign"):
                                _, qs = campaign_batch(hard, rng)
                                if not qs:
                                    qs = no_conflicts(split_solo([q for q in QUESTIONS if q["id"] in weak_ids()])[0], DRILL_SIZE)
                            elif body.get("next"):
                                _, qs = next_drill(hard, rng)
                            elif touch and not body.get("ids"):
                                qs = []
                            else:
                                ids = body.get("ids") or weak_ids()
                                qs = no_conflicts(split_solo([q for q in QUESTIONS if q["id"] in ids])[0], len(ids))
                            for d, n in touch:
                                qs = list(qs) + spot_check(d, n, hard, rng, taken=qs)
                            per = DRILL_SECONDS_PER_TASK_HARD if hard else DRILL_SECONDS_PER_TASK
                            begin_session(qs, "drill", per * len(qs), True, log=ulog, hard=hard, other_ok=False,
                                          objective=objective and objective["id"])
                    except SystemExit as e:
                        Api.seed_error = (getattr(e, "msg", None) or "\n".join(l for l in Api.log_lines if l.strip())
                                          or "start failed - see the terminal running ./ksim web")
                    except Exception as e:  # noqa: BLE001
                        Api.seed_error = f"{type(e).__name__}: {e}"
                    finally:
                        Api.seeding = False
                threading.Thread(target=work, daemon=True).start()
                return self._json({"seeding": True})
            if not st:
                return self._json({"error": "no session"}, 400)
            if st.get("mode") == "lesson" and p.path in ("/api/pause", "/api/restart", "/api/grade"):
                return self._json({"error": "A lesson has no clock and is checked step by step."}, 400)
            if p.path == "/api/flag":
                qid = int(body["id"]); flags = set(st.get("flags", []))
                flags.symmetric_difference_update({qid}); st["flags"] = sorted(flags); save_state(st)
            elif p.path == "/api/lesson/check":   # one step of the lesson in progress
                _, why = lesson_check(st, int(body.get("step", -1)))
                if why:
                    return self._json({"error": why}, 400)
            elif p.path == "/api/pause":
                set_paused(st, bool(body.get("paused", True)))
            elif p.path == "/api/restart":
                if st.get("ended_at"):
                    return self._json({"error": "This session has ended - there is no clock to restart."}, 400)
                restart_clock(st)
            elif p.path == "/api/terminate":   # an open session ends ungraded, as ./ksim abort; an ended one only loses its hosts
                if Api.seeding:
                    return self._json({"error": "A session is being prepared - wait until it is ready, then terminate it."}, 409)
                if not st.get("ended_at"):
                    st["ended_at"] = time.time(); st["aborted"] = True; save_state(st)
                drop_hosts(st)
            elif p.path == "/api/notes":
                st["notes"] = body.get("notes", ""); save_state(st)
            elif p.path == "/api/grade":
                if st.get("mode") != "drill" and not st.get("ended_at"):
                    return self._json({"error": "No feedback during an exam. Submit to see results."}, 403)
                q = q_by_id(int(body["id"]))
                if "results" not in st or not isinstance(st["results"], dict):
                    st["results"] = {}
                st["results"][str(q["id"])] = grade_one(q, session_params(st, q)); save_state(st)
            elif p.path == "/api/end":
                if not st.get("ended_at"):
                    down = hosts_down(st) if st.get("mode") != "lesson" else []  # grading hosts that are not running would record zeros
                    if down:
                        return self._json({"error": down_message(down)}, 503)
                    finish_session(st, log=lambda *_: None)
            elif p.path == "/api/reset":
                q = q_by_id(int(body["id"])); provision(q, session_params(st, q), log=lambda *_: None)
            else:
                return self._json({"error": "not found"}, 404)
            return self._json(self.session_payload())

    def session_payload(self) -> dict[str, Any]:
        st = load_state()
        qs = session_questions(st) if st else QUESTIONS
        payload: dict[str, Any] = {
            "track": TRACK, "track_id": TRACK_ID, "workspace": WS_DISPLAY, "version": VERSION,
            "desktop": desktop.url() if Api.desktop_ok else None, "desktop_error": Api.desktop_error, "setup_log": Api.log_lines,
            "questions": [{"id": q["id"], "domain": q["domain"], "title": q["title"], "weight": q["weight"], "difficulty": q["difficulty"],
                           "host": host_of(q), "nodes": 1 + workers_of(q),
                           "task": task_text(q, session_params(st, q)), "checks": [{"name": c["name"], "pts": c["pts"]} for c in q["checks"]],
                           "solution": solution_text(q, session_params(st, q)) if solution_allowed(st, int(q["id"])) else None,
                           # which objective a task belongs to is a hint, so it follows the solution's rule
                           "objectives": [o["id"] for o in objectives_of(q["id"])] if solution_allowed(st, int(q["id"])) else []}
                          for q in qs],
            "bank": [{"id": q["id"], "domain": q["domain"], "title": q["title"], "difficulty": q["difficulty"]} for q in QUESTIONS],
            "tracks": TRACKS, "track_k8s": track_k8s(),
            "k8s": k8s_version(), "pending": len(PENDING), "drilled": drilled_domains(), "sets": {k: {"name": v, "count": len(set_questions(k))} for k, v in SETS.items()},
            "seen": sorted(last_seen()),
            "seeding": Api.seeding, "seed_error": Api.seed_error,
            "session": None,
        }
        payload.update(self.path_payload())
        if st:
            earned, total, pct = score(st)
            payload["session"] = {"mode": st.get("mode"), "hard": st.get("hard", False), "started_at": st["started_at"], "duration": st.get("duration", EXAM_SECONDS),
                                  "ends_at": st["started_at"] + st.get("duration", EXAM_SECONDS), "ended_at": st.get("ended_at"),
                                  "over_time": st.get("over_time", False), "flags": st.get("flags", []),
                                  "paused": bool(st.get("paused_at")), "paused_total": paused_total(st),
                                  "notes": st.get("notes", ""),
                                  "objective": st.get("objective"), "milestones": st.get("milestones", []), "aborted": bool(st.get("aborted")),
                                  "results": st["results"], "score": {"earned": earned, "total": total, "pct": pct, "pass_pct": PASS_PCT},
                                  "domains": self.domain_scores(st), "server_time": time.time()}
            l = lesson_of(st)
            if l:
                payload["session"]["steps"] = st.get("steps", {})
                payload["lesson"] = self.lesson_payload(l, session_params(st, l))
        return payload

    def lesson_payload(self, l: dict[str, Any], params: dict[str, Any]) -> dict[str, Any]:
        """A lesson as the page shows it: introduction, steps (with their hint and commands, which the
        page keeps behind a click), wrap-up, its host and the objective's drill."""
        r = lambda t: render(t, params, l).strip("\n")
        return {"id": l["id"], "objective": l["objective"], "title": l["title"], "host": host_of(l), "nodes": 1 + workers_of(l),
                "intro": r(l["intro"]), "outro": r(l["outro"]),
                "steps": [{"title": r(s["title"]), "text": r(s["text"]), "hint": r(s["hint"]), "show": r(s["show"])} for s in l["steps"]]}

    def path_payload(self) -> dict[str, Any]:
        """The curriculum as the UI shows it: every objective with its notes, tasks and state, how
        many are done, and the next step on the path. Empty for a track without one."""
        if not OBJECTIVES:
            return {"objectives": [], "focus": [], "known": load_known(), "path": None}
        agg, known, done = stats(), load_known(), lessons_done()
        lesson = {l["objective"]: l for l in LESSONS}
        objs = [{"id": o["id"], "key": o["key"], "domain": o["domain"], "text": o["text"], "state": objective_state(o, agg, known),
                 "know": o["know"], "fast": o["fast"].strip("\n"), "traps": o["traps"], "docs": o["docs"],
                 "lesson": ({"id": lesson[o["id"]]["id"], "title": lesson[o["id"]]["title"], "steps": len(lesson[o["id"]]["steps"]),
                             "done": o["id"] in done} if o["id"] in lesson else None),
                 "tasks": [{"id": q["id"], "title": q["title"], "set": q.get("set"), "state": task_state(q["id"], agg)}
                           for q in objective_tasks(o, True)]} for o in OBJECTIVES]
        step = next_step(agg); full, fixed = mock_scores()
        states = [o["state"] for o in objs]
        return {"objectives": objs, "focus": load_focus(), "known": known,
                "path": {"total": len(objs), "known": states.count("known"), "done": states.count("done"),
                         "mock_best": max(full) if full else None, "set_best": {k: (max(v) if v else None) for k, v in fixed.items()},
                         "next": {"kind": step["kind"], "cmd": step["cmd"], "why": step["why"], "set": step.get("set"),
                                  "objective": step["objective"]["id"] if step.get("objective") else None}}}

    def domain_scores(self, st: dict[str, Any]) -> dict[str, Any]:
        qs = session_questions(st); out = {}
        st_results: dict[str, Any] = st.get("results", {})
        for d in DOMAINS:
            dq = [q for q in qs if q["domain"] == d]
            if dq:
                out[d] = {"earned": sum(st_results.get(str(q["id"]), {}).get("earned", 0) for q in dq),
                          "total": sum(q["weight"] for q in dq)}
        return out

def ensure_desktop():
    """The desktop, running and pointed at the open session if there is one. A running desktop is
    kept, so restarting the UI in the middle of a session does not cost you your terminals."""
    was_running = desktop.running()
    desktop.up()
    st = load_state()
    if not was_running and st and not st.get("ended_at"):
        desktop.configure(TRACK_ID, [envs.host_of(TRACK_ID, i) for i in st.get("ids", [])],
                          open_web=bool(os.environ.get("KSIM_OPEN_WEB")))

def cmd_web(args):
    port = 8080
    if "--port" in args:
        port = int(args[args.index("--port") + 1])
    if "--desktop-port" in args:
        desktop.PORT = int(args[args.index("--desktop-port") + 1])
    import socket
    def port_free(pt):
        with socket.socket() as sk:
            sk.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            try:
                sk.bind(("127.0.0.1", pt)); return True
            except OSError:
                return False
    if not port_free(port):
        die(f"Port {port} is already in use - probably another './ksim web' still running.\n"
            f"  Find it:  ss -ltnp | grep :{port}      or start on another port:  ./ksim web --port {port + 10}")
    try:
        ensure_desktop()
        Api.desktop_ok = True
        print(dim(f"  desktop: {desktop.NAME} on 127.0.0.1:{desktop.PORT} (XFCE over noVNC, loopback only)"))
    except envs.EnvError as e:
        Api.desktop_error = str(e)
        print(yellow(f"  no desktop: {e}"))
    http.server.ThreadingHTTPServer.allow_reuse_address = True
    srv = http.server.ThreadingHTTPServer(("127.0.0.1", port), Api)  # type: ignore[arg-type]
    print(bold(f"\n  {TRACK_ID.upper()} exam UI:  http://127.0.0.1:{port}   (ksim {VERSION})\n")
          + dim("  Ctrl-C stops the UI - not the exam clock, the desktop or the hosts (./ksim down removes those)."))
    def stop(*_):
        srv.shutdown()
    signal.signal(signal.SIGINT, lambda *_: threading.Thread(target=stop).start())
    signal.signal(signal.SIGTERM, lambda *_: threading.Thread(target=stop).start())
    srv.serve_forever()

def cmd_version(args):
    print(f"ksim {VERSION}  track={TRACK_ID}  k8s={k8s_version()}  bank={len(QUESTIONS)} tasks"
          + (f" (+{len(PENDING)} held back)" if PENDING else "") + f"  python={sys.version.split()[0]}")

def cmd_build(args):
    """Build the exam host image of a track's Kubernetes minor, and the desktop image."""
    names = [a for a in args if not a.startswith("--")] or [TRACK_ID]
    tracks = TRACKS if "all" in names else names
    for t in tracks:
        if t not in TRACKS:
            die(f"usage: ./ksim build [{'|'.join(TRACKS)}|all] [--vm]   (default: the current track, {TRACK_ID})")
    minors, variants, vm_minors = [], [], []
    for t in tracks:
        load_track(t)
        if any(is_vm(q) for q in QUESTIONS) and K8S not in vm_minors:
            vm_minors.append(K8S)
        for q in [{}] + [q for q in QUESTIONS + LESSONS if not is_vm(q)]:   # the track's own, then any a task names
            k8s, variant = q.get("k8s", K8S), q.get("image", TRACK.get("image"))
            if k8s not in minors:
                minors.append(k8s)
            if variant and (k8s, variant) not in variants:
                variants.append((k8s, variant))
    print(bold(f"Building images for {', '.join(t.upper() for t in tracks)} - the first build downloads a few hundred MB..."))
    try:
        for k8s in minors:
            envs.build_node_image(k8s)       # unchanged layers come from the build cache
        for k8s, variant in variants:        # a track's own flavour, on top of the common image
            envs.build_variant_image(k8s, variant)
        envs.build_desktop_image()
        if "--vm" in args:                   # the VM exam host, for tasks that need a kernel of their own
            for k8s in vm_minors:
                envs.build_vm_image(k8s)
    except envs.EnvError as e:
        die(str(e))
    print(green("Done.") + dim("   ./ksim web   opens the exam UI"))

def cmd_hosts(args):
    """The hosts of the session, or - `hosts start` - bring stopped ones back after a reboot."""
    st = load_state()
    ids = (st or {}).get("ids", [])
    if args and args[0] == "start":
        for i in ids:
            h = envs.host_of(TRACK_ID, i)
            if not envs.running(h):
                print(f"  starting {h}" if envs.start(h) else red(f"  {h} no longer exists - ./ksim reset {i} rebuilds it"))
        try:
            ensure_desktop()
        except envs.EnvError as e:
            print(yellow(f"  no desktop: {e}"))
        print(dim("  Give the API servers half a minute before grading."))
        return
    alive = set(envs.existing())
    if not ids:
        print("No session." + (f"  Hosts still up: {', '.join(sorted(alive))}  (./ksim down removes them)" if alive else ""))
        return
    print(bold(f"{'task':<6}{'host':<11}{'nodes':<7}{'state':<10}title"))
    for i in ids:
        q = QBYID.get(i)
        if not q:
            continue
        h = host_of(q)
        state = green("up") if envs.running(h) else (yellow("stopped") if h in alive else red("gone"))
        print(f"Q{i:<5}{h:<11}{1 + workers_of(q):<7}{state:<19}{q['title']}")
    print(dim(f"\ndesktop: {'up' if desktop.running() else 'not running'}     ./ksim ssh N   a shell on a task's host from this terminal"))

def cmd_ssh(args):
    """A shell on a task's host from your own terminal - the desktop's `ssh <host>` without the desktop."""
    if not args:
        die("usage: ./ksim ssh N")
    q = q_by_id(int(args[0])); node = nodes_of(q)["CP"]
    if not envs.running(host_of(q)):
        die(f"{host_of(q)} is not running - it exists only while a session with Q{q['id']} is open (./ksim hosts).")
    if is_vm(q):
        os.execvp(envs.RUNTIME, [envs.RUNTIME, "exec", "-it", node, "ssh", "-t", "candidate@vm"])
    os.execvp(envs.RUNTIME, [envs.RUNTIME, "exec", "-it", *AS_CANDIDATE, node, "bash", "-l"])

TARGET_PER_DOMAIN = (17, 20)   # pool tasks a domain should have, so every concept in it gets practised

def cmd_coverage(args):
    """Tasks per domain and per objective against the target - what the bank still needs."""
    lo, hi = TARGET_PER_DOMAIN
    print(bold(f"{TRACK['name']} - bank coverage") + dim(f"   target: {lo}-{hi} pool tasks per domain (fixed simulator sets are counted apart)"))
    print(bold(f"\n{'domain':<9}{'pool':>5}{'in sets':>9}{'to write':>10}   name"))
    need = 0
    for d, name in DOMAINS.items():
        dq = [q for q in QUESTIONS if q["domain"] == d]; pool = [q for q in dq if not q.get("set")]
        gap = max(0, lo - len(pool)); need += gap
        col = green if not gap else (yellow if gap <= 5 else red)
        print(f"{d:<9}{len(pool):>5}{len(dq) - len(pool):>9}{col((('+' + str(gap)) if gap else 'ok').rjust(10))}   {name}")
    print(dim(f"\n{len(POOL)} pool tasks, {len(QUESTIONS) - len(POOL)} in fixed sets" + (f", {len(PENDING)} held back" if PENDING else "")
              + f".  {need} more pool tasks reach {lo} in every domain."))
    if OBJECTIVES:
        print(bold(f"\n{'obj':<6}{'domain':<8}{'pool':>5}{'sets':>6}   objective"))
        for o in OBJECTIVES:
            ts = [QBYID[i] for i in o["tasks"] if i in QBYID]; pool = [q for q in ts if not q.get("set")]
            col = red if len(pool) < 2 else (yellow if len(pool) < 4 else (lambda x: x))
            print(col(f"{o['id']:<6}{o['domain']:<8}{len(pool):>5}{len(ts) - len(pool):>6}   {o['text']}"))
        print(dim("  a task may sit under several objectives, so these do not add up to the domain totals"))
    else:
        print(yellow(f"\nNo {TRACK_ID}/curriculum.py yet: no objectives, notes or path for this track."))
    for i, why in PENDING.items():
        print(dim(f"  held back: Q{i} - {why}"))

def cmd_audit(args):
    """Bank audit: build each task's host, seed it, grade it untouched, then - for a task with a
    scripted solution in <track>/solve.py - solve it on the host and grade it again. A clean seed
    and no full marks untouched is the least a task must manage; only the script proves it can be
    solved as written."""
    for t in TRACKS:
        try:
            with open(os.path.join(STATE_DIR, f"state-{t}.json")) as f:
                if not json.load(f).get("ended_at"):
                    die(f"A {t.upper()} session is open and the audit rebuilds hosts under the same names. End it first.")
        except (OSError, ValueError):
            pass
    settle = 0   # seconds between seeding and the untouched grade: a stopped kubelet takes ~40 s to show as NotReady
    if "--settle" in args:
        i = args.index("--settle"); settle = int(args[i + 1]); args = args[:i] + args[i + 2:]
    names = [a for a in args if not a.startswith("--")]
    if not names or names == ["all"]:
        qs = list(QUESTIONS)
    else:
        qs = [q for q in QUESTIONS if str(q["id"]) in names or q["domain"] in [n.upper() for n in names]]
    if not qs:
        die(f"usage: ./ksim audit [all|DOMAIN..|N..] [--settle SECONDS] [--keep]   (domains: {', '.join(DOMAINS)})")
    preflight(qs[:PARALLEL])
    envs.ensure_registry()
    lock = os.path.join(STATE_DIR, f"audit-{os.getpid()}.lock")   # keeps sessions from starting meanwhile
    os.makedirs(STATE_DIR, exist_ok=True)
    open(lock, "w").close()
    import atexit
    atexit.register(lambda: os.path.exists(lock) and os.remove(lock))
    print(bold(f"Bank audit of {len(qs)} {TRACK_ID.upper()} task(s), {PARALLEL} at a time - each gets its own cluster, is seeded, graded untouched and removed."))
    def one(q):
        params = make_params(q, random.Random(q["id"])); host = host_of(q); t0 = time.time()
        res = {"id": q["id"], "domain": q["domain"], "title": q["title"], "host": host, "workers": workers_of(q), "total": q["weight"]}
        try:
            make_host(q)
        except envs.EnvError as e:
            res.update(verdict="no cluster", detail=str(e)[-400:], secs=int(time.time() - t0))
            return res
        try:
            rc, out = run(q, q["setup"], params=params, timeout=300)
        except subprocess.TimeoutExpired:
            rc, out = 1, "setup timed out after 300s"
        envs.hand_over(host, params.get("ns") if q.get("ctx_ns") else None)
        res["seed_secs"] = int(time.time() - t0)
        time.sleep(settle)
        r = grade_one(q, params)
        res.update(earned=r["earned"], passing=[c["name"] for c in r["checks"] if c["pass"]])
        if rc != 0:
            res.update(verdict="seed failed", detail=out[-400:])
        elif r["earned"] == r["total"]:
            res.update(verdict="full marks unsolved")
        elif q.get("solve"):
            _, sout = solve_on_host(q, params)
            r2 = grade_one(q, params)
            # What a script created may still be starting. A candidate watches it come up before
            # moving on, so the task counts as solved if it reaches full marks within 90 s.
            waited = 0
            while not r2["pass"] and waited < 90:
                time.sleep(15); waited += 15
                r2 = grade_one(q, params)
            res.update(solved=r2["earned"], verdict="ok" if r2["pass"] else "solve failed")
            if not r2["pass"]:
                res["detail"] = "; ".join(f"✗ {c['name']}" for c in r2["checks"] if not c["pass"])[-400:]
        else:
            res.update(verdict="ok", setup_note=out[-200:] if out else "")
        if "--keep" not in args:
            envs.destroy(host)
        res["secs"] = int(time.time() - t0)
        return res
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=PARALLEL) as pool:
        for fut in concurrent.futures.as_completed([pool.submit(one, q) for q in qs]):   # as they finish
            res = fut.result()
            results.append(res)
            col = green if res["verdict"] == "ok" else red
            free = f"  {res['earned']}/{res['total']} pts for nothing" if res.get("earned") else ""
            proven = dim("  solved by script") if res["verdict"] == "ok" and "solved" in res else ""
            print(f"  {col(res['verdict']):<28} Q{res['id']:<4}{res['domain']:<8}{res['secs']:>4}s  {res['title']}{yellow(free) if free else ''}{proven}", flush=True)
            if res.get("detail"):
                print(dim("      " + res["detail"].replace("\n", "\n      ")), flush=True)
    results.sort(key=lambda r: r["id"])
    out_dir = os.path.join(STATE_DIR, "audit"); os.makedirs(out_dir, exist_ok=True)
    path = os.path.join(out_dir, f"audit-{TRACK_ID}.json")
    now = datetime.now().isoformat(timespec="minutes")
    for r in results:
        r["at"] = now
    try:   # the report holds the latest result of every task ever audited, not only this run's
        with open(path) as f:
            merged = {r["id"]: r for r in json.load(f)["results"]}
    except (OSError, ValueError, KeyError):
        merged = {}
    merged.update({r["id"]: r for r in results})
    with open(path, "w") as f:
        json.dump({"k8s": k8s_version(), "results": [merged[i] for i in sorted(merged)]}, f, indent=1)
    bad = [r for r in results if r["verdict"] != "ok"]
    proven = sum(1 for r in results if r["verdict"] == "ok" and "solved" in r)
    print(bold(f"\n{len(results) - len(bad)}/{len(results)} seed cleanly and give no full marks untouched; {proven} of them also solved by their script.") + (red(f"  {len(bad)} need work: Q" + ", Q".join(str(r['id']) for r in bad)) if bad else ""))
    print(dim(f"  report: {os.path.relpath(path, ROOT)}"))

def cmd_down(args):
    """Delete every exam host and the desktop. History, path progress and images are kept."""
    for t in TRACKS:
        try:
            with open(os.path.join(STATE_DIR, f"state-{t}.json")) as f:
                if not json.load(f).get("ended_at") and "--yes" not in args:
                    die(f"A {t.upper()} session is still open; its hosts hold your work.\n"
                        f"  Submit it (./ksim --track {t} end), or confirm:  ./ksim down --yes")
        except (OSError, ValueError):
            pass
    gone = clear_hosts()
    desktop.down()
    envs.stop_registry()
    print(f"Removed {len(gone)} host(s) and the desktop." + dim("  History, images and the image cache's contents are kept."))

def cmd_help(args):
    print(__doc__ + f"\nversion {VERSION}")
    objective_help = "" if not OBJECTIVES else (
        f"  objectives               the curriculum as a path to the exam: {len(OBJECTIVES)} objectives, their tasks, your progress\n"
        "  next                     start the next step on that path (drill an objective against the clock, then the mocks)\n"
        "  focus [DOMAIN..|clear]   walk these domains first on the path - the ones an exam report said are weakest\n"
        "  skip [DOMAIN|OBJ..]      mark a domain or an objective as known: the path leaves it out, nothing is recorded\n"
        "                           as passed, and a mock still draws from it  (skip clear / unskip X bring it back)\n"
        "  notes [OBJ|DOMAIN|all]   what to know, fastest commands, traps and docs for an objective (3.2 or probes)\n"
        "  drill OBJ                an objective's tasks against the clock, those not yet passed first\n")
    objective_help += "" if not LESSONS else (
        f"  lesson [OBJ]             the track's lessons ({len(LESSONS)}), or start one: a guided lab on a host of your own, no clock\n"
        "  lesson step|check|hint|show [N]   the step you are on, check it, a hint, the commands;  ./ksim end finishes\n"
        "  lesson audit [OBJ..]     prove lessons: every step fails before its commands and passes after them\n")
    print(f"""usage: ./ksim [--track cka|ckad|cks] <command>        (current track: {TRACK_ID}, {len(QUESTIONS)} tasks)
       Every task runs on its own cluster at the exam's Kubernetes version ({TRACK_ID}: {k8s_version()}),
       reached with `ssh <host>` from the remote desktop.

  build [TRACK|all]        build the exam host and desktop images (once per Kubernetes minor)
  build cks --vm           also build the VM exam host that the Falco and AppArmor tasks need (about ten minutes)
  web [--port 8080]        the exam in the browser: task pane, remote desktop, clock, submit
{objective_help}  drill                    next domain you have drilled least: {DRILL_SIZE} tasks, 6 min each  (--hard: 5 min, hard tasks)
  drill all [--hard]       campaign: {DRILL_SIZE} never-attempted tasks per session until the bank is done, then weak ones
  drill weak|N..|DOMAIN..  chosen tasks  (--time MIN, --force, --hard)
  drill .. DOMAIN:3        three tasks of that domain, spread over its objectives - on their own, or added to
                           whatever else you named: ./ksim next OBS:3
  progress                attempted/passed per domain, normal and hard
  solution N               reference solution (drills: after grading N; exams: after submit)
  sim {'|'.join(SETS) or '<set>'}{' ' * max(1, 20 - len('|'.join(SETS)))}fixed simulator set, exam rules, {TRACK['minutes']} min
  start [--hard]           full exam: {sum(QUOTA.values()) or '?'} tasks sampled by domain ({sum(QUOTA_HARD.values()) or '?'} in hard mode), {TRACK['minutes']} min.
                           Requires every domain drilled once first (--force skips). Hard mode adds traps,
                           exact-output tasks, compound failures and collateral checks.
  list                     tasks in this session, with current + historical results
  show N                   display task N and the host it is solved on
  grade N|all              grade one task (or all) per check - drills only (--force to peek in an exam)
  status / timer           time remaining (timer = live countdown)
  pause / resume           stop the clock (not recorded anywhere - but the real exam has no pause)
  restart                  the clock from the top: the session's full time again, hosts and work kept (not recorded)
  end                      submit: grade per check, score, log attempt, update weak-spots
  weak                     what you keep getting wrong, with the failing checks
  history                  every attempt, scores, and per-domain trend
  reset N                  rebuild the host of task N from scratch (new cluster, scenario seeded again)
  hosts [start]            the session's hosts and their state; `start` brings them back after a reboot
  ssh N                    a shell on the host of task N from this terminal
  down                     delete every host and the desktop (history and images are kept)
  coverage                 tasks per domain and objective against the 17-20 per domain target
  audit [all|DOMAIN|N..]   bank check: build each task's host, seed it, grade it untouched, solve it with its
                           script if it has one (<track>/solve.py), remove it   (--settle SECONDS, --keep)
""")

COMMANDS = {"start": cmd_start, "drill": cmd_drill, "list": cmd_list, "show": cmd_show, "grade": cmd_grade,
            "status": cmd_status, "timer": cmd_timer, "pause": cmd_pause, "resume": cmd_resume, "restart": cmd_restart, "end": cmd_end, "weak": cmd_weak, "history": cmd_history,
            "reset": cmd_reset, "web": cmd_web, "abort": cmd_abort, "progress": cmd_progress, "solution": cmd_solution, "sim": cmd_sim, "version": cmd_version, "help": cmd_help,
            "build": cmd_build, "hosts": cmd_hosts, "ssh": cmd_ssh, "down": cmd_down, "audit": cmd_audit, "coverage": cmd_coverage,
            "objectives": cmd_objectives, "curriculum": cmd_objectives, "notes": cmd_notes, "next": cmd_next, "focus": cmd_focus,
            "skip": cmd_skip, "known": cmd_skip, "unskip": cmd_unskip, "lesson": cmd_lesson, "lessons": cmd_lesson}

if __name__ == "__main__":
    cmd = ARGV[0] if ARGV else "help"
    COMMANDS.get(cmd, cmd_help)(ARGV[1:])
