#!/usr/bin/env python3
"""exam - Kubernetes certification exam simulator (CKA / CKAD / CKS tracks).
Hands-on tasks graded per-check against the live cluster, exam clock, attempt history,
weak-spot log, drill mode, and a browser front end modelled on the real exam."""
import http.server
import importlib
import json
import os
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.05.1"
ROOT = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, ROOT)

# ------------------------------------------------------------ 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 = ""
STUDY_FILE = ""   # tasks solved in study mode (the objective checklist reads it; history never does)
OBJECTIVES = []   # the track's curriculum, objective by objective (tracks/<id>/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.35"
CLUSTER = ""      # kind cluster for that minor
CONTEXT = ""
NODES = {}

# One kind cluster per Kubernetes minor an exam runs on, so tracks on the same minor share one
# (CKA and CKS today) and moving an exam to a new minor is a one-line change to its TRACK["k8s"].
# The images are the ones the kind v0.33.0 release notes publish; kind only guarantees an image
# built for its own release when it is pinned by digest.
KIND_IMAGES = {
    "v1.35": "kindest/node:v1.35.8@sha256:07b2536e30b803ed61d1677a79df6115f798ce64c80f9e22f6ed45afd09323c0",
    "v1.37": "kindest/node:v1.37.0@sha256:a1ed56cfb0e7b93589bdf97c8cd566405a265939e3620fc4f5de89adff580ae5",
}

def kind_cluster(k8s):
    """Name of the kind cluster for a Kubernetes minor: v1.35 -> cka-sim-135."""
    return "cka-sim-" + k8s.lstrip("v").replace(".", "")


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"tracks.{track_id}.questions")
    except ModuleNotFoundError:
        sys.exit(f"unknown track '{track_id}' (no tracks/{track_id}/questions.py)")
    importlib.reload(mod)
    track, questions = mod.TRACK, list(mod.QUESTIONS)
    try:
        cur = importlib.import_module(f"tracks.{track_id}.curriculum")
        cur = importlib.reload(cur)
        objectives, study_order = list(cur.OBJECTIVES), list(getattr(cur, "STUDY_ORDER", []))
    except ModuleNotFoundError as e:
        if e.name != f"tracks.{track_id}.curriculum":
            raise
        objectives, study_order = [], []  # no objective map for this track: study/drill by domain only
    for q in questions:
        q["weight"] = sum(c["pts"] for c in q["checks"])
        q.setdefault("difficulty", "normal")
        q.setdefault("set", None)
    cluster = kind_cluster(track["k8s"])
    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"),
        STUDY_FILE=os.path.join(ROOT, f"study-{track_id}.json"), OBJECTIVES=objectives, STUDY_ORDER=study_order,
        # The CKA_* variables point every track at your own cluster instead (see the README).
        K8S=track["k8s"], CLUSTER=cluster,
        CONTEXT=os.environ.get("CKA_CONTEXT", f"kind-{cluster}"),
        NODES={"CP": os.environ.get("CKA_CP", f"{cluster}-control-plane"),
               "W1": os.environ.get("CKA_W1", f"{cluster}-worker"),
               "W2": os.environ.get("CKA_W2", f"{cluster}-worker2"),
               "W3": os.environ.get("CKA_W3", f"{cluster}-worker3")},
    )


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


STATE_DIR = os.path.join(ROOT, ".state")
WORKSPACE = os.path.join(ROOT, "workspace")
# How {WS} is shown in task text and solutions. Relative to the repo root by default, since that
# is where you are working; graders always use the absolute $WS internally.
WS_DISPLAY = os.environ.get("CKA_WS_DISPLAY") or os.path.relpath(WORKSPACE, ROOT)
DRILL_SECONDS_PER_TASK = 6 * 60
DRILL_SECONDS_PER_TASK_HARD = 5 * 60
DRILL_SIZE = 5
import random
NAMESPACES_TO_SETTLE = ["apps", "netpol", "svc", "ingress-app", "dns-check", "storage"]

# --- cluster targeting: CONTEXT and NODES follow the track (load_track); env vars override ---
def _default_access():
    try:
        return open(os.path.join(STATE_DIR, "runtime")).read().strip()  # written by setup.sh
    except OSError:
        return "podman"
NODE_ACCESS = os.environ.get("CKA_NODE_ACCESS", _default_access())  # podman | docker | ssh | vagrant
# kubectl matching each cluster's server version, fetched by setup.sh: .state/kubectl/v1.35/kubectl.
# .state/kubectl/current links to the one for the cluster of the latest session; examrc puts it
# first on PATH, so your terminals switch kubectl version together with the cluster.
KUBECTL_DIR = os.path.join(STATE_DIR, "kubectl")
def shell_hint(node):
    return {"podman": f"podman exec -it {node} bash",
            "docker": f"docker exec -it {node} bash",
            "ssh": f"ssh {node}",
            "vagrant": f"vagrant ssh {node}"}.get(NODE_ACCESS, f"ssh {node}")

PRELUDE = r"""
set -uo pipefail
K="kubectl --context ${CKA_CONTEXT:?}"     # grading never depends on the candidate's current context
WS="$WORKSPACE"
# 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="${CKA_CP:?}"; W1="${CKA_W1:?}"; W2="${CKA_W2:?}"; W3="${CKA_W3:?}"   # run() exports the track's nodes
# nx NODE 'command' -> run a shell command on a node (stdin is forwarded)
nx() {
  local n="$1"; shift
  case "${CKA_NODE_ACCESS:-podman}" in
    ssh)     ssh -o BatchMode=yes -o StrictHostKeyChecking=accept-new "$n" "$@" ;;
    vagrant) vagrant ssh "$n" -c "$*" -- -o LogLevel=QUIET ;;
    docker)  docker exec -i "$n" sh -c "$*" ;;
    *)       podman exec -i "$n" sh -c "$*" ;;
  esac
}
# ensure_etcd_tools -> etcdctl and etcdutl on the control-plane node, as on the real exam.
# The etcd image is distroless, so they cannot be copied out of the pod; etcd-tools.sh takes
# them from the image in the node's containerd (and falls back to a download / apt).
ensure_etcd_tools() {
  nx "$CP" 'command -v etcdctl >/dev/null 2>&1 && command -v etcdutl >/dev/null 2>&1' </dev/null && return 0
  bash "${CKA_ROOT:-.}/etcd-tools.sh" --emit | nx "$CP" 'sh -s'
}
# ensure_gateway_crds -> the Gateway API CRDs (standard channel), which kind does not ship. Q56 and
# Q74 need them at seed time; a cluster rebuilt by ./setup.sh has none until one of them installs it.
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` (CKAD Q23). kind ships none, and
# kind's kubelet serving certs have no IP SANs, so it only becomes Ready with --kubelet-insecure-tls.
# CKA Q23 strips that flag on purpose; the baseline puts it back before every session.
# Pinned: v0.9.x supports Kubernetes 1.34+, so it serves the v1.35 and v1.37 clusters alike, and a
# future "latest" that drops 1.35 cannot break the CKA track.
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
}
"""

def _pick_track_arg(argv):
    t = os.environ.get("CKA_TRACK", "cka")
    if "--track" in argv:
        i = argv.index("--track"); t = argv[i + 1]; del argv[i:i + 2]
    return t


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 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 untimed(st: dict[str, Any] | None) -> bool:
    return bool(st and st.get("untimed"))

def remaining(st: dict[str, Any] | None) -> float:
    if not st:
        return 0.0
    if untimed(st):
        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 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:  ./exam start   (or ./exam 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"tracks.{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}
    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:
        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 {}

def render(text, params):
    """Substitute %{param}, {WS}, {CP}.. and {SHELL_CP}.. placeholders."""
    t = text.replace("{WS}", WS_DISPLAY)
    for k, v in (params or {}).items():
        t = t.replace("%{" + k + "}", str(v))
    for k, v in NODES.items():
        t = t.replace("{" + k + "}", v).replace("{SHELL_" + k + "}", shell_hint(v))
    return t

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 render(q["task"].strip(), params if params is not None else session_params(load_state(), q))

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

# -------------------------------------------------------------- shell ----
def kubectl_path():
    """PATH with the kubectl matching this track's Kubernetes minor first, when setup.sh fetched one."""
    d = os.path.join(KUBECTL_DIR, K8S)
    path = os.environ.get("PATH", "")
    return f"{d}{os.pathsep}{path}" if os.path.isfile(os.path.join(d, "kubectl")) else path

def run(script, timeout=180, params=None):
    env = dict(os.environ, WORKSPACE=WORKSPACE, CKA_ROOT=ROOT, CKA_NODE_ACCESS=NODE_ACCESS, CKA_CONTEXT=CONTEXT,
               PATH=kubectl_path(), **{f"CKA_{k}": v for k, v in NODES.items()})
    p = subprocess.run(["bash", "-c", PRELUDE + render(script, params)], env=env, capture_output=True, text=True,
                       timeout=timeout, stdin=subprocess.DEVNULL)
    return p.returncode, (p.stdout + p.stderr).strip()

def server_version():
    """The cluster's gitVersion (e.g. v1.35.8), or None when it cannot be reached."""
    try:
        p = subprocess.run(["kubectl", "--context", CONTEXT, "get", "--raw=/version"],
                           capture_output=True, text=True, timeout=15)
        return json.loads(p.stdout).get("gitVersion") if p.returncode == 0 else None
    except Exception:
        return None

def kubectl_ok():
    return server_version() is not None

def setup_hint():
    """How to (re)build this track's cluster."""
    return f"./setup.sh {TRACK_ID}"

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(ch["script"], timeout=240, params=params)
        except subprocess.TimeoutExpired:
            rc, out = 1, "timed out"
        checks.append({"name": render(ch["name"], params), "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 ./exam start. Better to hand over a hard task than to silently skip the domain.
        got = no_conflicts(build(True), n, taken)
    return got

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()
    pool = [q for q in POOL if q["id"] not in seen and not q.get("solo") and (hard or q["difficulty"] == "normal")]
    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([q for q in POOL if hard or q["difficulty"] == "normal"])
    return f"{total - len(pool)}/{total} attempted", qs

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

def cmd_solution(args):
    if not args:
        die("usage: ./exam 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 (./exam grade N); in an exam, submit first. --force overrides.")
    print(bold(f"Solution - Q{q['id']} {q['title']}")); print("-" * 64)
    sol = str(q.get("solution", ""))
    print(render(sol, session_params(st, q))); print("-" * 64)
    print(dim(f"One correct way, not the only one. $do = --dry-run=client -o yaml, k = kubectl. Paths are relative to {ROOT}"))

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.  ./exam 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  (`./exam 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 - study an objective with
# the solutions open, then prove it against the clock - and `./exam 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 path_objectives():
    """Objectives in the order the path walks them: the curriculum's order, unless the track names
    a STUDY_ORDER of domains (CKA leaves Troubleshooting for last - it builds on all the rest)."""
    order = [d for d in STUDY_ORDER if d in DOMAINS] + [d for d in objective_domains() if d not in STUDY_ORDER]
    return sorted(OBJECTIVES, key=lambda o: order.index(o["domain"]))   # stable: curriculum order within a domain

def objective_session(o, how, with_sets=False, n=None):
    """The tasks of one sitting on an objective: what is still open first ("study": never solved;
    "drill": not yet passed against the clock), at most n, never two that break the same thing.
    A big objective therefore takes several sittings, each on new ground."""
    agg, studied = stats(), load_study()
    closed = ((lambda q: task_solved(q["id"], agg, studied)) if how == "study"
              else (lambda q: task_state(q["id"], agg, studied) == "passed"))
    qs = objective_tasks(o, with_sets)
    todo = [q for q in qs if not closed(q)] 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:
      tasks       task id -> its latest result in a study session ("passed" sticks once earned)
      objectives  objective id -> when it was first studied and first completed
    Study results are kept apart from the history on purpose: a pass with the solution open is
    practice, not proof, so it never reaches the statistics, the weak list or the exam gate."""
    try:
        with open(STUDY_FILE) as f:
            d = json.load(f)
    except (OSError, ValueError):
        d = {}
    return {"tasks": {int(k): v for k, v in d.get("tasks", {}).items()}, "objectives": dict(d.get("objectives", {}))}

def save_progress(prog):
    with open(STUDY_FILE, "w") as f:
        json.dump({"tasks": {str(k): prog["tasks"][k] for k in sorted(prog["tasks"])},
                   "objectives": {k: prog["objectives"][k] for k in sorted(prog["objectives"])}}, f, indent=1)
        f.write("\n")

def load_study():
    return load_progress()["tasks"]

def record_study(st, ids=None):
    """Note what a study session graded (all of it, or just `ids`)."""
    if not untimed(st):
        return
    prog = load_progress(); log = prog["tasks"]
    for k, r in st.get("results", {}).items():
        if ids is not None and int(k) not in ids:
            continue
        log[int(k)] = {"at": datetime.now().isoformat(timespec="minutes"), "pass": r["pass"], "earned": r["earned"],
                       "total": r["total"], "passed": bool(r["pass"] or log.get(int(k), {}).get("passed"))}
    save_progress(prog)

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

def task_solved(qid, agg, studied):
    """Solved at least once, with the solution open or against the clock."""
    a = agg.get(int(qid))
    return bool(studied.get(int(qid), {}).get("passed") or (a is not None and a.attempts > a.fails))

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

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

def objective_state(o, agg, studied):
    """The two stages every objective goes through, and where it is:
      studied  every pool task solved at least once (study mode counts)
      done     every pool task passed against the clock on its latest attempt - exam conditions
    before that: started (something attempted) or new."""
    qs = objective_tasks(o)
    states = [task_state(q["id"], agg, studied) for q in qs]
    if qs and all(x == "passed" for x in states):
        return "done"
    if qs and all(task_solved(q["id"], agg, studied) for q in qs):
        return "studied"
    return "started" if any(x != "new" for x in states) else "new"

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

def next_objective(agg=None, studied=None):
    """(objective, how) for the first objective in curriculum order that is not done: "study" it
    until every task has been solved once, then "drill" it against the clock. (None, None) when
    every objective is done."""
    agg = stats() if agg is None else agg
    studied = load_study() if studied is None else studied
    for o in path_objectives():
        state = objective_state(o, agg, studied)
        if state != "done":
            return o, ("drill" if state == "studied" else "study")
    return None, 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, studied=None):
    """The next move on the path to the exam, as {kind, cmd, why[, objective | set]}:
    study an objective -> drill it 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
    studied = load_study() if studied is None else studied
    o, how = next_objective(agg, studied)
    if o and how == "study":
        return {"kind": "study", "objective": o, "cmd": f"./exam study {o['id']}",
                "why": f"{o['id']} {o['text']} - notes, then its tasks with no clock and the solutions open"}
    if o:
        return {"kind": "drill", "objective": o, "cmd": f"./exam drill {o['id']}",
                "why": f"{o['id']} {o['text']} - studied; now the same tasks against the clock, no solution until graded"}
    full, fixed = mock_scores()
    if not any(x >= PASS_PCT for x in full):
        return {"kind": "mock", "cmd": "./exam start",
                "why": f"every objective is done - a full mock: {sum(QUOTA.values())} tasks, {TRACK['minutes']} minutes, no feedback until you submit"}
    for k in SETS:
        if not any(x >= PASS_PCT for x in fixed[k]):
            return {"kind": "sim", "set": k, "cmd": f"./exam sim {k}",
                    "why": f"{SETS[k]}: {len(set_questions(k))} fixed tasks, {TRACK['minutes']} minutes, exam rules"}
    return {"kind": "done", "cmd": "./exam 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 studied and the first time it is done, and
    return what was newly reached as [(objective, "studied" | "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:
        state = objective_state(o, agg, prog["tasks"])
        seen = prog["objectives"].setdefault(o["id"], {})
        for level in ("studied", "done"):
            if state in (level, "done") and level not in seen:
                seen[level] = now
                new.append((o, level))
        if not seen:
            del prog["objectives"][o["id"]]
    if new:
        save_progress(prog)
    # reaching "done" in one go also stamps "studied"; only the higher one is news
    return [(o, lv) for o, lv in new if not (lv == "studied" and (o, "done") in new)]

def milestone_lines(new):
    out = []
    for o, level in new:
        n = len(objective_tasks(o))
        if level == "done":
            out.append(f"✓ Objective {o['id']} {o['key']} complete - {n}/{n} tasks passed against the clock.")
        else:
            out.append(f"~ Objective {o['id']} {o['key']} studied - every task solved once. Prove it against the clock: ./exam drill {o['id']}")
    return out

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 study."""
    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"  Study it by domain instead:  ./exam study {next(iter(DOMAINS))}")

def cmd_objectives(args):
    """The curriculum as a checklist: every objective, its tasks, and how far you are."""
    require_objectives()
    agg, studied = stats(), load_study()
    cols = shutil.get_terminal_size((120, 24)).columns
    sets = len(QUESTIONS) - len(POOL)
    print(bold(f"{TRACK['name']} - study 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  ~ solved in study mode  ✗ failed last time  · not yet"
              "      objectives: ✓ complete  ~ studied  - started"))
    def cell(q):
        return f"{MARKS[task_state(q['id'], agg, studied)][0]}{q['id']}", f"{mark(task_state(q['id'], agg, studied))}{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)))
    for d in objective_domains():
        print(bold(f"\n{DOMAINS[d]}"))
        for o, plain, shown in rows:
            if o["domain"] != d:
                continue
            state = objective_state(o, agg, studied)
            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, studied)
    for g in curriculum_gaps():
        print(yellow(f"curriculum.py: {g}"))

def print_path(agg=None, studied=None):
    """The ladder from first reading to exam conditions, how far up you are, and the next rung."""
    agg = stats() if agg is None else agg
    studied = load_study() if studied is None else studied
    states = [objective_state(o, agg, studied) for o in OBJECTIVES]
    n, done = len(OBJECTIVES), states.count("done")
    full, fixed = mock_scores()
    step = next_step(agg, studied)
    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 = [("study", "Study", "each objective's tasks, notes and solutions open", f"{done + states.count('studied')}/{n} objectives"),
            ("drill", "Clock", "the same tasks as a timed drill, graded per check", f"{done}/{n} objectives")]
    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}")
    print(bold(f"\nNext:  {step['cmd']}") + dim("   (or just ./exam next)"))
    print(dim(f"       {step['why']}"))
    if step.get("objective"):
        print(dim(f"       ./exam 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: study an objective, drill it, 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"] == "study":
        cmd_study([step["objective"]["id"]] + args)
    elif 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 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, studied=None):
    agg = stats() if agg is None else agg
    studied = load_study() if studied is None else studied
    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, studied))} 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 (./exam 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 `./exam --track {TRACK_ID} notes all --md` from `tracks/{TRACK_ID}/curriculum.py` - edit that file, not this one._", "",
         "```", f"./exam objectives        # the path and your progress", "./exam study 3.2         # one objective: these notes, then its tasks, no clock, solutions open",
         "./exam drill 3.2         # the same 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"`./exam study {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()[0] 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, studied = stats(), load_study()
    for i, o in enumerate(objs):
        if i:
            print()
        print_notes(o, agg, studied)
    if len(objs) == 1:
        print(dim(f"./exam study {objs[0]['id']}   its tasks, no clock, solutions open        ./exam drill {objs[0]['id']}   against the clock"))

# ------------------------------------------------------------ session ----
def preflight():
    out = subprocess.run(["kubectl", "--context", CONTEXT, "get", "nodes", "-o", "jsonpath={.items[*].metadata.name}"],
                         capture_output=True, text=True).stdout.split()
    missing = [v for v in NODES.values() if v not in out]
    if missing:
        die(f"Nodes not found in cluster: {missing}\n  cluster has: {out}\n"
            "  Set CKA_CP, CKA_W1, CKA_W2, CKA_W3 to your node names (needs 1 control-plane + 3 workers).")
    rc, _ = run('nx "$CP" "true" </dev/null && nx "$W1" "true" </dev/null && nx "$W2" "true" </dev/null && nx "$W3" "true" </dev/null', timeout=60)
    if rc != 0:
        die(f"Cannot run commands on nodes with CKA_NODE_ACCESS={NODE_ACCESS}.\n"
            f"  podman/docker -> node names must be kind container names (podman ps --filter label=io.x-k8s.kind.cluster={CLUSTER})\n"
            "  ssh     -> passwordless ssh to each node name must work\n"
            "  vagrant -> run from the Vagrantfile directory; names are vagrant machine names")
    ver = server_version() or "?"
    print(dim(f"  track: {TRACK_ID}  cluster: {CONTEXT} ({ver})  node access: {NODE_ACCESS}  nodes: {', '.join(NODES.values())}"))
    # The real exams pin one minor each; practising on another one teaches the wrong defaults.
    if not ver.startswith(K8S + "."):
        fix = f"rebuild it:  {setup_hint()} --recreate" if CONTEXT == f"kind-{CLUSTER}" else "point CKA_CONTEXT at a cluster on that minor"
        print(yellow(f"  warning: this cluster runs {ver}, but the real {TRACK_ID.upper()} exam runs Kubernetes {K8S} - {fix}"))
    for tool, why in (("helm", "Helm tasks (the real exam has helm on the terminal)"), ("openssl", "certificate tasks"), ("curl", "tasks that fetch manifests")):
        if not shutil.which(str(tool)):
            print(yellow(f"  warning: '{tool}' not found on this machine - needed for {why}"))

HARD_CONTEXTS = {"k8s-c1-H": "default", "k8s-c2-AC": "%{ns}", "k8s-c3-CCC": "kube-system"}

def context_namespace(ctx, template, params_by_id):
    """Default namespace for one exam context.

    Several tasks can name the same context while working in different namespaces, but a
    context has only one default namespace - so it belongs to whichever task actually reads it
    ("ctx_ns": True, i.e. the task says to work in the context's namespace without -n). Every
    other task states its namespace in its own text, so the value does not matter to them.
    """
    if "%{ns}" not in template:
        return template
    def ns_of(pred):
        return [p["ns"] for i, p in params_by_id.items()
                if p.get("ctx") == ctx and p.get("ns") and pred(QBYID.get(int(i)) or {})]
    return (ns_of(lambda q: q.get("ctx_ns")) or ns_of(lambda q: True)
            or [p["ns"] for p in params_by_id.values() if p.get("ns")] or ["default"])[0]

def setup_contexts(hard, params_by_id):
    """Exam-style contexts on the same cluster with different default namespaces.

    Created whenever a task in the session asks for one - 24 tasks open with
    `kubectl config use-context k8s-c?-?` and are unfollowable without it - not only when the
    whole session was started with --hard. A drill that names hard task ids explicitly
    ('./exam drill 121', './exam drill weak') pulls those tasks in with hard=False.

    Only tracks with TRACK["exam_contexts"] (CKA) use them, so only those tracks create or delete
    them: the others run on another cluster, and must not repoint or remove the CKA contexts.
    """
    if not TRACK.get("exam_contexts"):
        subprocess.run(["kubectl", "config", "use-context", CONTEXT], capture_output=True)
        return
    cluster = subprocess.run(["kubectl", "config", "view", "-o", f"jsonpath={{.contexts[?(@.name=='{CONTEXT}')].context.cluster}}"],
                             capture_output=True, text=True).stdout.strip()
    user = subprocess.run(["kubectl", "config", "view", "-o", f"jsonpath={{.contexts[?(@.name=='{CONTEXT}')].context.user}}"],
                          capture_output=True, text=True).stdout.strip()
    for name in HARD_CONTEXTS:
        subprocess.run(["kubectl", "config", "delete-context", name], capture_output=True)
    wanted = set(HARD_CONTEXTS) if hard else {p.get("ctx") for p in params_by_id.values()} & set(HARD_CONTEXTS)
    if not wanted:
        subprocess.run(["kubectl", "config", "use-context", CONTEXT], capture_output=True)
        return
    for name in sorted(wanted):
        subprocess.run(["kubectl", "config", "set-context", name, f"--cluster={cluster}", f"--user={user}",
                        f"--namespace={context_namespace(name, HARD_CONTEXTS[name], params_by_id)}"], capture_output=True)
    # Hard mode starts you in c1 like the real exam; a mixed drill stays on the normal context
    # so its non-hard tasks behave as usual and the hard ones still switch as instructed.
    start = "k8s-c1-H" if hard and "k8s-c1-H" in wanted else CONTEXT
    subprocess.run(["kubectl", "config", "use-context", start], capture_output=True)

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


# Disruptive tasks (see "breaks": ... in the question bank) corrupt shared control-plane /
# node state. A session's setup only touches the questions it actually contains, so damage
# from a task in an earlier session lingers if that task isn't picked again - e.g. Q39 wrecks
# W1's kubelet config, and a later session with different questions never undoes it. This
# restores every known "breaks" target to a good baseline before each session seeds, so
# leftover damage never leaks across sessions regardless of which questions get sampled.
RESTORE_BASELINE = r"""
# Static-pod manifests are only ever rewritten when an edit actually changes them, and through a
# hidden temp file the kubelet ignores. `sed -i` writes its temp file (sedXXXXXX) into the SAME
# directory, the kubelet loads that as one more static pod and kills the running one: every session
# start used to knock the scheduler and controller-manager into CrashLoopBackOff for up to minutes
# while the session's tasks were being seeded (new Namespaces got no ServiceAccount, so bare Pods
# were rejected; Pods sat unscheduled). The user's cluster had 10 such restarts by 2026-09-24.
MEDIT='medit() { f=$1; shift; t=$(mktemp "${f%/*}/.medit.XXXXXX") || return 1
  if sed "$@" "$f" > "$t" 2>/dev/null && [ "$(md5sum < "$t")" != "$(md5sum < "$f")" ]; then
    chmod --reference="$f" "$t" 2>/dev/null; mv -f "$t" "$f"; else rm -f "$t"; fi; }'

# apiserver FIRST: Q66 renames the command to "kube-apiserverr", which takes the whole API
# down - every kubectl-based step below would fail until this is undone. The repair goes over
# podman/ssh (nx), not kubectl, so it works precisely when the cluster is unreachable.
if nx "$CP" 'grep -q "^ *- kube-apiserverr$" /etc/kubernetes/manifests/kube-apiserver.yaml' </dev/null 2>/dev/null; then
  nx "$CP" "$MEDIT"'
medit /etc/kubernetes/manifests/kube-apiserver.yaml -e "s|^\( *\)- kube-apiserverr$|\1- kube-apiserver|"' </dev/null
  # the kubelet re-reads the manifest on its own; wait for the API to answer again
  for _ in $(seq 1 30); do $K get --raw=/readyz >/dev/null 2>&1 && break; sleep 2; done
fi

# apiserver audit flags (CKS Q9, whose track shares this cluster with CKA): a half-finished answer -
# flags without the volume, or a hostPath of type File whose policy file is gone - keeps the API server
# down for every later session. Touched only when the API server is down and audit flags are present,
# so a working audit setup is left alone.
if nx "$CP" 'grep -q -- "--audit-" /etc/kubernetes/manifests/kube-apiserver.yaml' </dev/null 2>/dev/null; then
  for _ in $(seq 1 10); do $K get --raw=/readyz >/dev/null 2>&1 && break; sleep 3; done
  if ! $K get --raw=/readyz >/dev/null 2>&1; then
    nx "$CP" 'mkdir -p /var/log/kubernetes
[ -f /etc/kubernetes/audit-policy.yaml ] || printf "apiVersion: audit.k8s.io/v1\nkind: Policy\nrules:\n- level: None\n" > /etc/kubernetes/audit-policy.yaml' </dev/null
    nx "$CP" "$MEDIT"'
medit /etc/kubernetes/manifests/kube-apiserver.yaml -e "/--audit-/d"' </dev/null
    for _ in $(seq 1 45); do $K get --raw=/readyz >/dev/null 2>&1 && break; sleep 2; done
  fi
fi

# scheduler: undo the "kube-scheduler" -> "kube-schedulerr" command typo (Q20/Q61), and put the
# manifest back if a task had the candidate stop the scheduler by moving it out (Q97) and they
# never moved it back - that setup keeps a pristine copy outside the static pod directory.
nx "$CP" "$MEDIT"'
f=/etc/kubernetes/manifests/kube-scheduler.yaml
[ -f $f ] || { [ -f /etc/kubernetes/.cka-sim/kube-scheduler.yaml ] && cp /etc/kubernetes/.cka-sim/kube-scheduler.yaml $f; }
medit $f -e "s|^\( *\)- kube-schedulerr$|\1- kube-scheduler|"
' </dev/null

# controller-manager: undo Q38's injected flag and Q61's broken image tag. The last two lines
# undo the retired set B's Q317 (the CSR signing controller switched off in --controllers, or in
# its older form --cluster-signing-cert-file pointed at a missing file, which crash-looped the WHOLE
# controller-manager); a cluster that ever ran that set may still carry either.
nx "$CP" "$MEDIT"'
medit /etc/kubernetes/manifests/kube-controller-manager.yaml \
  -e "/--leader-elect-lease-duration=1s/d" \
  -e "s|\(image: registry.k8s.io/kube-controller-manager:v[0-9.]*\)-broken|\1|" \
  -e "s|--cluster-signing-cert-file=/etc/kubernetes/pki/ca-signing.crt|--cluster-signing-cert-file=/etc/kubernetes/pki/ca.crt|" \
  -e "s|,-certificatesigningrequest-signing-controller||; s|,-csrsigning||"
' </dev/null

# coredns: restore the default upstream forward (Q37/Q57 point it somewhere bogus)
cf=$($K -n kube-system get cm coredns -o jsonpath='{.data.Corefile}' 2>/dev/null)
if [ -n "$cf" ] && ! printf '%s' "$cf" | grep -Eq '^[[:space:]]*forward \. /etc/resolv\.conf'; then
  printf '%s' "$cf" | sed -E 's|^([[:space:]]*forward \. )[^[:space:]{]+|\1/etc/resolv.conf|' > /tmp/Corefile.restore
  $K -n kube-system create cm coredns --from-file=Corefile=/tmp/Corefile.restore --dry-run=client -o yaml | $K apply -f - >/dev/null 2>&1
  $K -n kube-system rollout restart deploy coredns >/dev/null 2>&1
fi
# ...and Q57's other two faults. kindnet ENFORCES NetworkPolicy, so a leftover lockdown policy on
# the CoreDNS pods kills name resolution for the whole cluster in every later session, and so does
# a Deployment left at 0 replicas.
$K -n kube-system delete netpol maintenance-lockdown --ignore-not-found >/dev/null 2>&1
r=$($K -n kube-system get deploy coredns -o jsonpath='{.spec.replicas}' 2>/dev/null)
if [ -n "$r" ] && [ "$r" -lt 2 ]; then $K -n kube-system scale deploy coredns --replicas=2 >/dev/null 2>&1; fi

# w1/w2/w3 kubelet: undo Q17/Q39-style tampering, then make sure it's enabled and running.
# staticPodPath is normalised unconditionally: Q46 relocates it to /etc/kubernetes/static, and
# the config.yaml.orig copy below only helps if something made that backup first - nothing does,
# so a relocated path used to survive every later session and silently break Q05 (whose check
# looks in /etc/kubernetes/manifests). Only restart the kubelet if we actually changed the file.
# clientCAFile gets the same treatment: Q39 points it at ca-bundle.crt and Q60 at ca-wrong.crt, and
# only W1 ever gets a config.yaml.orig (Q39 makes it), so an unfinished Q60 left W2's kubelet
# crash-looping in every later session - enable/start below cannot help a kubelet that exits.
# The systemd drop-in is checked too: Q92 points ExecStart at a kubelet binary that does not exist.
for n in "$W1" "$W2" "$W3"; do
  nx "$n" '
    c=/var/lib/kubelet/config.yaml; d=/etc/systemd/system/kubelet.service.d/10-kubeadm.conf
    [ -f $c.orig ] && cp -f $c.orig $c
    restart=0
    if grep -q "^staticPodPath:" $c 2>/dev/null && ! grep -q "^staticPodPath: /etc/kubernetes/manifests$" $c; then
      sed -i "s|^staticPodPath: .*|staticPodPath: /etc/kubernetes/manifests|" $c; restart=1
    fi
    ca=$(awk "/clientCAFile:/{print \$2}" $c 2>/dev/null)
    if [ -n "$ca" ] && [ ! -f "$ca" ]; then
      sed -i "s|clientCAFile: .*|clientCAFile: /etc/kubernetes/pki/ca.crt|" $c; restart=1
    fi
    bin=$(sed -n "s|^ExecStart=\([^ ]*kubelet\) .*|\1|p" $d 2>/dev/null | tail -1)
    if [ -n "$bin" ] && [ ! -x "$bin" ]; then
      sed -i "s|^ExecStart=$bin |ExecStart=/usr/bin/kubelet |" $d; systemctl daemon-reload; restart=1
    fi
    [ $restart = 1 ] && systemctl restart kubelet >/dev/null 2>&1
    systemctl enable kubelet >/dev/null 2>&1
    systemctl start kubelet  >/dev/null 2>&1
  ' </dev/null
done

# static pods the bank drops on nodes: Q05's static-web on W3, Q46's relocated copies, Q94's
# ops-beacon on the control plane - and edge-cache, left on W3 by the retired set B (it sat in
# ContainerCreating for days, its hostPath deliberately missing). Each setup recreates what its
# own session needs.
nx "$W3" 'rm -f /etc/kubernetes/manifests/static-web*.yaml /etc/kubernetes/manifests/edge-cache.yaml /etc/kubernetes/static/*.yaml' </dev/null
nx "$CP" 'rm -f /etc/kubernetes/manifests/ops-beacon*.yaml' </dev/null

# default StorageClass: Q16 makes fast-local the default and strips `standard`. Leave exactly one
# default - `standard` - so "the default class" means the same thing in every session.
if $K get sc standard >/dev/null 2>&1; then
  for sc in $($K get sc -o jsonpath='{.items[*].metadata.name}'); do
    [ "$sc" = standard ] && continue
    [ "$($K get sc "$sc" -o jsonpath='{.metadata.annotations.storageclass\.kubernetes\.io/is-default-class}')" = true ] \
      && $K annotate sc "$sc" storageclass.kubernetes.io/is-default-class- >/dev/null 2>&1
  done
  $K annotate sc standard storageclass.kubernetes.io/is-default-class=true --overwrite >/dev/null 2>&1
fi

# w1-taint: clear every randomized demo taint key the bank can apply. Q25 draws from
# dedicated/team/workload and Q50 from pool/lane; `maintenance` was the retired set B's and may
# still be on a cluster that ran it. Miss one and that node silently stops accepting pods.
# One key per call: `kubectl taint node N a- b-` aborts on the first key that is not present and
# removes NOTHING, so the single taint that is actually set survives the "cleanup" and W1 keeps
# rejecting pods in every later session - the exact failure this block exists to prevent.
for _t in dedicated workload team pool lane maintenance; do
  $K taint node "$W1" "$_t-" >/dev/null 2>&1 || true
done

# metrics-server: Q23 seeds its scenario by stripping --kubelet-insecure-tls, and kind's kubelet
# serving certs have no IP SANs, so without that flag metrics-server never becomes Ready. Left
# broken it does far more than disable `kubectl top`: the v1beta1.metrics.k8s.io APIService has no
# endpoints, aggregated discovery goes stale, and then EVERY namespace deletion in the cluster
# hangs on NamespaceDeletionDiscoveryFailure. Baseline runs before seeding, so a session that
# actually contains Q23 re-breaks it a moment later, exactly as intended.
if $K -n kube-system get deploy metrics-server >/dev/null 2>&1 \
   && ! $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 || true
fi

# cordons: Q60 cordons W2, Q109 has the control-plane node drained, and a drain task left half-done
# leaves a node cordoned. A cordon left behind fails any later task that needs every worker (Q26's
# DaemonSet coverage, Q50's spread) or a Pod on the control-plane node (Q96, Q109 itself).
for n in "$CP" "$W1" "$W2" "$W3"; do $K uncordon "$n" >/dev/null 2>&1 || true; done

# Deliberately NOT restored, and why:
#   etcd (Q45) and certs (Q65). Neither setup damages anything - Q45 only takes a snapshot
#   and deletes a ConfigMap, Q65 only records the current notBefore. Any damage comes from a
#   half-finished candidate attempt (etcd.yaml left pointing at a restored data-dir, static pod
#   manifests moved to /tmp and never moved back), which is not deterministic: there is no way
#   to tell "mid-attempt" from "correctly completed" without guessing, and guessing wrong here
#   rewrites etcd's data-dir or re-issues certs on a healthy cluster. Re-run ./setup.sh (or
#   ./teardown.sh && ./setup.sh) if one of those leaves the control plane wedged.
"""

def restore_baseline(log: Callable[..., Any] = _log_msg):
    log(dim("  resetting control-plane / node state to a clean baseline..."))
    rc, out = run(RESTORE_BASELINE, timeout=90)
    if rc != 0:
        log(yellow(f"  warning: baseline restore had issues (continuing anyway):\n{dim(out[-500:])}"))


def seed(qs, log: Callable[..., Any] = _log_msg, params_by_id=None):
    params_by_id = params_by_id or {}
    normal = [q for q in qs if not q.get("disruptive")]
    disruptive = [q for q in qs if q.get("disruptive")]
    def one(q):
        try:
            rc, out = run(q["setup"], params=params_by_id.get(str(q["id"])), timeout=240)
        except subprocess.TimeoutExpired:
            rc, out = 1, "setup timed out after 240s"
        log(f"  {'✓' if rc == 0 else '!'} Q{q['id']:02d} {q['title']}" + ("" if rc == 0 else f"\n{dim(out[-300:])}"))
    for q in normal:
        one(q)
    if disruptive:
        log(dim("  waiting up to 45s for seeded pods to be scheduled..."))
        for _ in range(15):
            pending = subprocess.run(["kubectl", "--context", CONTEXT, "get", "pods", "-A", "--field-selector=spec.nodeName=",
                                      "-o", "name"], capture_output=True, text=True).stdout.strip()
            if not pending:
                break
            time.sleep(3)
    for q in disruptive:
        one(q)

def open_sessions_elsewhere():
    """Tracks other than this one with a session that has not ended. Sessions share the kubeconfig's
    current context (and kubectl version), so starting one would pull the other's terminal over to
    another cluster while its grading stays on the old one: silent zeros."""
    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 use_kubectl():
    """Point .state/kubectl/current at the kubectl for this track's minor, or at your own kubectl
    ('host') when setup.sh could not fetch one. examrc puts current/ first on PATH; it always exists
    once setup.sh has run, so a shell that already hashed current/kubectl follows the switch."""
    if not os.path.isdir(KUBECTL_DIR):
        return
    target = K8S
    if not os.path.isfile(os.path.join(KUBECTL_DIR, K8S, "kubectl")):
        target = "host"
        mine = os.path.realpath(KUBECTL_DIR)  # never our own current/ - PATH may spell it via a symlink
        own = shutil.which("kubectl", path=os.pathsep.join(
            p for p in os.environ.get("PATH", "").split(os.pathsep) if not os.path.realpath(p).startswith(mine)))
        if not own:
            return
        link = os.path.join(KUBECTL_DIR, "host", "kubectl")
        os.makedirs(os.path.dirname(link), exist_ok=True)
        if os.path.realpath(link) != os.path.realpath(own):
            if os.path.lexists(link):
                os.remove(link)
            os.symlink(os.path.realpath(own), link)
    cur = os.path.join(KUBECTL_DIR, "current")
    tmp = cur + ".new"
    try:
        if os.path.lexists(tmp):
            os.remove(tmp)
        os.symlink(target, tmp)
        os.replace(tmp, cur)  # atomic: a terminal never sees current/ missing
    except OSError:
        pass

def begin_session(qs, mode, duration, force, log: Callable[..., Any] = _log_msg, hard=False, untimed_session=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. './exam end' it first, or add --force.")
    others = open_sessions_elsewhere()
    if others and not (force if other_ok is None else other_ok):
        t = others[0]
        how = (f"switch to {t.upper()} in the lobby and submit it, or run  ./exam --track {t} end  (abort --yes discards it)"
               if other_ok is False else f"./exam --track {t} end   (or abort --yes),  or add --force here")
        die(f"A {t.upper()} session is still open. Tracks run on their own clusters but share your kubectl context,\n"
            f"  so finish that one first: {how}.")
    if others:
        log(yellow(f"  warning: the open {'/'.join(o.upper() for o in others)} session loses its kubectl context and kubectl"
                   f" version to this {TRACK_ID.upper()} session; end it before going back to it."))
    if not kubectl_ok():
        die(f"Cannot reach the {TRACK_ID.upper()} cluster through kubectl context '{CONTEXT}' (Kubernetes {K8S}).\n"
            f"  Create or restart it:  {setup_hint()}     (or set CKA_CONTEXT for your own cluster)")
    if not qs:
        die(f"Track '{TRACK_ID}' has no questions yet.")
    preflight()
    update_milestones()   # stamp what the history already proves, so this session only announces its own
    os.makedirs(WORKSPACE, exist_ok=True)
    rng = random.Random()
    params_by_id = {str(q["id"]): make_params(q, rng) for q in qs}
    setup_contexts(hard, params_by_id)
    use_kubectl()
    restore_baseline(log)
    log(bold(f"Seeding {len(qs)} scenario(s){' [HARD]' if hard else ''}..."))
    seed(qs, log, params_by_id)
    st = {"mode": mode, "track": TRACK_ID, "hard": hard, "untimed": untimed_session, "ids": [q["id"] for q in qs],
          "params": params_by_id, "duration": duration, "started_at": time.time(), "ended_at": None,
          "k8s": server_version(), "results": {}, "flags": [], "notes": "",
          "objective": objective}   # the curriculum objective a study/drill session was started for, if any
    save_state(st)
    log("")
    log(bold("=" * 64))
    kind = "STUDY" if untimed_session else ("EXAM" if mode == "exam" else "DRILL")
    log(bold(f"  {TRACK_ID.upper()} {kind}{' · HARD' if hard else ''} - {'no clock, solutions open' if untimed_session else 'clock started'}"))
    log(f"  Cluster : {CONTEXT}  (Kubernetes {st['k8s'] or K8S}, as the real {TRACK_ID.upper()} runs {K8S})")
    if hard and TRACK.get("exam_contexts"):
        log(yellow("  Contexts: k8s-c1-H (default) · k8s-c2-AC · k8s-c3-CCC  - each task tells you which to use; you are on k8s-c1-H now"))
    if not untimed_session:
        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)}"
        + ("" if untimed_session else f"   Pass mark: {PASS_PCT}%"))
    log(bold("=" * 64))
    log("  ./exam web         exam-style browser UI (task pane + terminal + clock)\n"
        + ("  ./exam list / show N / grade N / solution N / end   terminal workflow\n" if untimed_session else
           "  ./exam list / show N / grade N / timer / end   terminal workflow\n")
        + ("  Tip: read ALL tasks first. Some are cluster-wide failures that block others.\n" if mode == "exam" else "")
        + f"  Files the tasks ask for go under {WS_DISPLAY}/ (relative to this directory)\n"
        + "  Your own terminal: source ./examrc   (fish: source ./examrc.fish)  -> k, $do, $now, completion")
    return st

def finish_session(st, log: Callable[..., Any] = _log_msg):
    qs = session_questions(st)
    over = remaining(st) <= 0
    set_paused(st, False)
    log(bold(f"Submitting - grading {len(qs)} task(s)..."))
    for q in qs:
        r = grade_one(q)
        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)
    setup_contexts(False, {})
    entry = None
    if untimed(st):
        record_study(st)
    else:
        entry = record_attempt(st)
    # objectives this session got over a line: shown by ./exam 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
    undrilled = [d for d, n in drilled_domains().items() if n == 0]
    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:  ./exam drill      (picks the next domain)\n"
            "  or:   ./exam 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()
    domain = min(DOMAINS, key=lambda d: (cov[d], list(DOMAINS).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 ./exam 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 './exam end' first, or name tasks: ./exam 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 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), "drill")
                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)}), a set "
                    f"({', '.join('set' + k for k in SETS) or 'none'})"
                    + (" or an objective (./exam 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: ./exam 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 cmd_study(args):
    """No clock, solutions unlocked from the start, nothing written to history."""
    hard = "--hard" in args; force = "--force" in args; with_sets = "--all" in args
    rest = [a for a in args if not a.startswith("--")]
    n = DRILL_SIZE
    if "--count" in args:
        n = int(args[args.index("--count") + 1]); rest = [a for a in rest if a != str(n)]
    rng = random.Random()
    obj = None
    if rest and rest[0].upper() in DOMAINS:
        domain = rest[0].upper()
        qs = pick(domain, n, hard, rng)
        print(bold(f"Study: {DOMAINS[domain]}"))
    elif rest and rest[0] == "weak":
        qs = no_conflicts(split_solo([q for q in QUESTIONS if q["id"] in weak_ids()])[0], n)
        print(bold("Study: tasks you have been failing"))
    elif OBJECTIVES and (not rest or rest[0] == "next" or find_objective(rest[0])):
        # one objective of the curriculum: the one named, else the next one on the path
        if rest and rest[0] != "next":
            obj = find_objective(rest[0])
        else:
            agg, studied = stats(), load_study()
            obj = next((o for o in path_objectives() if objective_state(o, agg, studied) in ("new", "started")), None)
            if obj is None:
                die("Every objective has been studied - name one to revisit (./exam objectives lists them),\n"
                    f"  or carry on up the path:  {next_step(agg, studied)['cmd']}   (./exam next)")
        qs, open_n = objective_session(obj, "study", with_sets, n if "--count" in args else None)
        print(bold(f"Study: {obj['id']} {obj['text']}") + dim(f"   {DOMAINS[obj['domain']]}"))
        if open_n > len(qs):
            print(dim(f"  {len(qs)} of its {open_n} open tasks in this sitting - run the same command again for the rest."))
    elif rest:
        qs = no_conflicts(split_solo([q_by_id(int(a)) if a.isdigit() else die(
            f"'{a}' is not a task id, a domain ({', '.join(DOMAINS)}), 'weak'"
            + (" or an objective (./exam objectives)." if OBJECTIVES else ".")) for a in rest])[0], len(rest))
        print(bold("Study: " + ", ".join(f"Q{q['id']}" for q in qs)))
    else:
        domain, qs = next_drill(hard, rng)
        print(bold(f"Study: {DOMAINS[domain]}"))
    begin_session(qs, "drill", EXAM_SECONDS, force, hard=hard, untimed_session=True, objective=obj and obj["id"])
    if obj:
        print(); print_notes(obj)
        print(dim("  ./exam show N   read a task      ./exam grade N   check it      ./exam solution N   reference answer\n"
                  "  ./exam notes    these notes again      ./exam end   finish (./exam objectives then shows what you solved)"))

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: ./exam 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: ./exam 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("-" * 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: ./exam 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 ./exam end (--force to peek anyway).")
    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)
    record_study(st, set(ids))
    if untimed(st):
        reached = milestone_lines(update_milestones())
        if reached:   # kept with the session, so ./exam end and the results page show them again
            st["milestones"] = st.get("milestones", []) + reached; save_state(st)
        for line in reached:
            print(green(line))
    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):
    if untimed(st):
        return green("study mode - no clock") + dim(f"  [{len(session_questions(st))} tasks, kept out of the history]")
    r = remaining(st)
    if st.get("paused_at"):
        return yellow(f"⏸  PAUSED - {fmt(r)} remaining") + dim("  ./exam resume")
    if r <= 0:
        return red("⏰ TIME IS UP - run ./exam 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 cmd_timer(args):
    st = require_started()
    if st.get("ended_at"):
        die("Session already ended.")
    if untimed(st):
        die("This is a study session - there is no clock.")
    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("   ./exam resume        ")); sys.stdout.flush(); time.sleep(1); continue
            if r <= 0:
                print("\r" + red(bold("⏰  00:00:00  TIME IS UP - run ./exam 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.")
    set_paused(st, True); print(status_line(st))

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

def cmd_end(args):
    st = require_started()
    if st.get("ended_at"):
        die("Session already ended and recorded.")
    if not kubectl_ok():
        die(f"Cannot reach the {TRACK_ID.upper()} cluster ({CONTEXT}), so every check would fail and a 0% attempt would be recorded.\n"
            f"  Start it first:  {setup_hint()}     or discard the session:  ./exam abort --yes")
    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']}"))
    if entry is None:
        print(dim("Study session - not added to history or weak-spots.")); path_note(); return
    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:  ./exam drill weak") + dim(f"   (Q{', Q'.join(map(str, failed))})"))
    path_note()

def cmd_abort(args):
    """End a session without grading it and restore the kubectl context."""
    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:  ./exam abort --yes        (to grade and keep it instead: ./exam end)")
    st["ended_at"] = time.time(); st["aborted"] = True
    save_state(st)
    setup_contexts(False, {})
    print(yellow("Session discarded. Nothing was graded or written to history."))
    print(dim(f"  kubectl context restored to {CONTEXT}. Seeded objects are still on the cluster;"))
    print(dim(f"  './exam drill ...' or './exam start' re-seeds, './teardown.sh {TRACK_ID}' removes the cluster."))

def cmd_weak(args):
    agg = stats(); hist = load_history()
    if not hist:
        die("No history yet - finish a session with './exam 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: ./exam 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):
    if not args:
        die("usage: ./exam reset N   (re-seed one task's scenario)")
    q = q_by_id(int(args[0])); rc, out = run(str(q.get("setup", "")), params=session_params(load_state(), q))
    print(("re-seeded " if rc == 0 else "problem re-seeding ") + f"Q{q['id']:02d}: {q['title']}")
    if out:
        print(dim(out))

# --------------------------------------------------------------- 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
    ttyd_port = 7681
    shell = "bash"

    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/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("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:
                            objective = find_objective(body["objective"]) if body.get("objective") else None
                            if objective:
                                qs, _ = objective_session(objective, "study" if body.get("study") else "drill", 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)
                            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))
                            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,
                                          untimed_session=bool(body.get("study")), 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 ./exam 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 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/pause":
                set_paused(st, bool(body.get("paused", True)))
            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)
                record_study(st, {q["id"]})
                if untimed(st):
                    st["milestones"] = st.get("milestones", []) + milestone_lines(update_milestones()); save_state(st)
            elif p.path == "/api/end":
                if not st.get("ended_at"):
                    if not kubectl_ok():  # grading an unreachable cluster would record a 0% attempt
                        return self._json({"error": f"Cannot reach the {TRACK_ID.upper()} cluster ({CONTEXT}). Start it with {setup_hint()}, then submit."}, 503)
                    finish_session(st, log=lambda *_: None)
            elif p.path == "/api/reset":
                q = q_by_id(int(body["id"])); run(str(q.get("setup", "")), params=session_params(st, q))
            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, "ttyd_port": Api.ttyd_port,
            "ttyd": shutil.which("ttyd") is not None, "shell": Api.shell, "setup_log": Api.log_lines,
            "questions": [{"id": q["id"], "domain": q["domain"], "title": q["title"], "weight": q["weight"], "difficulty": q["difficulty"],
                           "task": task_text(q, session_params(st, q)), "checks": [{"name": c["name"], "pts": c["pts"]} for c in q["checks"]],
                           "solution": render(str(q.get("solution", "")), 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(),
            "cluster": CONTEXT, "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),
                                  "untimed": untimed(st), "notes": st.get("notes", ""),
                                  "objective": st.get("objective"), "milestones": st.get("milestones", []),
                                  "results": st["results"], "score": {"earned": earned, "total": total, "pct": pct, "pass_pct": PASS_PCT},
                                  "domains": self.domain_scores(st), "server_time": time.time()}
        return payload

    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 studied and done, and the next step on the path. Empty for a track without one."""
        if not OBJECTIVES:
            return {"objectives": [], "path": None}
        agg, studied = stats(), load_study()
        objs = [{"id": o["id"], "key": o["key"], "domain": o["domain"], "text": o["text"], "state": objective_state(o, agg, studied),
                 "know": o["know"], "fast": o["fast"].strip("\n"), "traps": o["traps"], "docs": o["docs"],
                 "tasks": [{"id": q["id"], "title": q["title"], "set": q.get("set"), "state": task_state(q["id"], agg, studied)}
                           for q in objective_tasks(o, True)]} for o in OBJECTIVES]
        step = next_step(agg, studied); full, fixed = mock_scores()
        states = [o["state"] for o in objs]
        return {"objectives": objs,
                "path": {"total": len(objs), "done": states.count("done"), "studied": states.count("done") + states.count("studied"),
                         "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 cmd_web(args):
    port = 8080
    if "--port" in args:
        port = int(args[args.index("--port") + 1])
    Api.shell = os.environ.get("CKA_SHELL", "bash")
    if "--shell" in args:
        Api.shell = args[args.index("--shell") + 1]
    os.makedirs(WORKSPACE, exist_ok=True)
    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 './exam web' still running.\n"
            f"  Find it:  ss -ltnp | grep :{port}      or start on another port:  ./exam web --port {port + 10}")
    if shutil.which("ttyd") and not port_free(Api.ttyd_port):
        print(yellow(f"  ttyd port {Api.ttyd_port} already in use - reusing the existing terminal server."))
    ttyd = None
    if shutil.which("ttyd") and port_free(Api.ttyd_port):
        env = dict(os.environ, WORKSPACE=WORKSPACE, KUBECONFIG=os.environ.get("KUBECONFIG", os.path.expanduser("~/.kube/config")))
        rc = os.path.join(ROOT, "examrc.fish" if "fish" in Api.shell else "examrc")
        if "fish" in Api.shell:
            shell_cmd = [Api.shell, "-C", f"source {rc}"]
        else:
            shell_cmd = [Api.shell, "--rcfile", rc, "-i"]
        # -W makes the terminal writable, so it must never be reachable off this machine: it is an
        # unauthenticated root-equivalent shell with the cluster kubeconfig. ttyd binds every
        # interface by default, so pin it to loopback (macOS needs lo0; override if yours differs).
        iface = os.environ.get("CKA_TTYD_INTERFACE", "lo")
        ttyd = subprocess.Popen(["ttyd", "-p", str(Api.ttyd_port), "-i", iface, "-W",
                                 "-t", "fontSize=14", "-t", "theme={\"background\":\"#1a1b1e\"}",
                                 "--cwd", ROOT] + shell_cmd, env=env,
                                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        print(dim(f"  terminal: ttyd on {iface}:{Api.ttyd_port} running {Api.shell} in {ROOT} with k=kubectl, $do and $now preset"))
    else:
        print(yellow("  ttyd not found - the terminal pane will show install instructions.\n"
                     "  (dnf/apt/pacman/brew install ttyd, or https://github.com/tsl0922/ttyd/releases)"))
    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}   (exam {VERSION})\n") + dim("  Ctrl-C stops the UI (not the exam clock)."))
    def stop(*_):
        if ttyd:
            ttyd.terminate()
        srv.shutdown()
    signal.signal(signal.SIGINT, lambda *_: threading.Thread(target=stop).start())
    signal.signal(signal.SIGTERM, lambda *_: threading.Thread(target=stop).start())
    try:
        srv.serve_forever()
    finally:
        if ttyd:
            ttyd.terminate()

def cmd_version(args):
    print(f"exam {VERSION}  track={TRACK_ID}  k8s={K8S}  cluster={CONTEXT}  bank={len(QUESTIONS)} tasks  python={sys.version.split()[0]}")

def cmd_cluster_env(args):
    """Shell assignments for this track's kind cluster - setup.sh, teardown.sh and the tool scripts
    eval this, so the track -> Kubernetes version -> cluster/image mapping lives only here."""
    print(f"TRACK={TRACK_ID}\nK8S={K8S}\nCLUSTER={CLUSTER}\nNODE_IMAGE={KIND_IMAGES.get(K8S, '')}")

def cmd_clusters(args):
    """Which cluster each track runs on, and whether it is up."""
    print(bold(f"{'track':<7}{'exam runs':<11}{'kind cluster':<15}{'context':<19}status"))
    for t in TRACKS:
        load_track(t)
        ver = server_version()
        status = (green(f"up, {ver}") if ver and ver.startswith(K8S + ".") else yellow(f"up, but {ver}") if ver
                  else red(f"not reachable  ({setup_hint()})"))
        print(f"{t:<7}{K8S:<11}{CLUSTER:<15}{CONTEXT:<19}{status}")

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 (study an objective, then drill it, then the mocks)\n"
        "  notes [OBJ|DOMAIN|all]   what to know, fastest commands, traps and docs for an objective (3.2 or probes)\n"
        "  study [OBJ|next]         an objective's notes and tasks; with no argument, the next one on the path\n"
        "                           (--all adds its Simulator Set tasks);  drill OBJ = the same tasks against the clock\n")
    print(f"""usage: ./exam [--track cka|ckad|cks] <command>        (current track: {TRACK_ID}, {len(QUESTIONS)} tasks)
       Each track runs on a kind cluster at its exam's Kubernetes version - {TRACK_ID}: {K8S} on {CONTEXT}.

  web [--port 8080] [--shell bash]   exam-style browser UI: tasks, terminal, clock, submit
  study [DOMAIN|weak|N..]  no clock, solutions unlocked, kept out of the history  (--count N, default {DRILL_SIZE})
{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)
  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' if not TRACK.get('exam_contexts') else ', collateral checks and exam-style contexts'}.
  list                     tasks in this session, with current + historical results
  show N                   display task N
  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)
  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                  re-seed the scenario for task N
  clusters                 each track's Kubernetes version and cluster, and whether it is up
""")

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, "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, "study": cmd_study, "version": cmd_version, "help": cmd_help,
            "clusters": cmd_clusters, "cluster-env": cmd_cluster_env,
            "objectives": cmd_objectives, "curriculum": cmd_objectives, "notes": cmd_notes, "next": cmd_next}

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