#!/usr/bin/env python3
"""
colibri — tiny engine, immense model.
Run GLM-5.2 (744B) locally on CPU with roughly 15-26 GB of RAM.

  coli chat                 interactive chat (loads the model once)
  coli serve                OpenAI-compatible HTTP API (persistent engine)
  coli run "prompt"         one-shot generation
  coli info                 model, RAM, disk, and configuration status
  coli plan                 Disk / RAM / VRAM resource plan
  coli mirror               Plan, stage, or verify a learned partial mirror
  coli doctor               installation and execution-plan diagnostics
  coli bench [task...]      quality benchmarks (MMLU/HellaSwag/...)
  coli convert              convert GLM-5.2-FP8 to int4, one shard at a time
  coli build                build the engine

Configuration through environment variables or flags (also valid after the subcommand):
  COLI_MODEL=<dir>   model directory (required; or pass --model <dir>)
  COLI_MODEL_MIRROR=<dir>  second copy of the model on another drive: expert reads
                     are split across both SSDs (COLI_DISK_WEIGHTS=9,3 sets the
                     primary,mirror bandwidth ratio; default: measured at startup)
  --ram N            RAM budget in GB (automatically sizes the expert cache)
  --repin N          adapt RAM/VRAM experts every N tokens
  --topp P           adaptive expert top-p             --topk N   fixed top-k
  --ngen N           maximum response tokens           --cap N    cache slots/layer
"""
import os, sys, subprocess, argparse, json, time, signal, shutil, threading, re, codecs, tempfile, textwrap, struct, shlex

# input() only gets line editing and UP/DOWN history once the readline module
# has been LOADED — without it the terminal's arrow-key escape sequences are
# echoed raw into the prompt (#922). Guarded: Windows has no readline module
# and the console provides its own editing; on FreeBSD Python links libedit,
# which this activates the same way.
try:
    import readline  # noqa: F401 — importing it is the activation
except ImportError:
    pass

# The engine mmaps every shard (144+ files); macOS default RLIMIT_NOFILE is 256.
if sys.platform != "win32":
    try:
        import resource
        _soft, _hard = resource.getrlimit(resource.RLIMIT_NOFILE)
        _want = min(65536 if _hard == resource.RLIM_INFINITY else _hard, 65536)
        if _soft < _want:
            resource.setrlimit(resource.RLIMIT_NOFILE, (_want, _hard))
    except (ImportError, ValueError, OSError):
        pass

# Windows: forza output UTF-8 (console cp1252 tronca Unicode box-drawing/emoji)
if sys.platform == "win32":
    for s in (sys.stdout, sys.stderr):
        try: s.reconfigure(encoding="utf-8")
        except (AttributeError, OSError): pass

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
# version.py sits next to this script in a source checkout, but an installed
# layout puts the launcher in $(PREFIX)/bin while the support modules live in
# $(PREFIX)/libexec/colibri — and a packager may strip it entirely (#575,
# FreeBSD port). The launcher must never crash over a version string.
try:
    from version import __version__ as _version
except ModuleNotFoundError:
    sys.path.insert(0, os.path.join(os.path.dirname(HERE), "libexec", "colibri"))
    try:
        from version import __version__ as _version
    except ModuleNotFoundError:
        _version = "unknown"

from family_registry import (FamilyConfigError, UnknownFamilyError, all_families,
                             display_for, family_by_id, family_for_config, resolve_model,
                             tuning_replay_prompt)

_TUNE_ROTATION_PROMPT = (
    "A service has high median throughput but poor tail latency. Explain how "
    "you would distinguish a CPU, memory, and storage bottleneck."
)

# Run-in-place (source checkout, "cd c && ./coli ..."): the engine, the
# support modules (resource_plan.py, doctor.py, autotune.py, openai_server.py) and
# tools/ all live next to this script — unchanged from before.
#
# Installed layout ("make install"): this script is $(PREFIX)/bin/coli,
# while the engine binaries and support files live in
# $(PREFIX)/libexec/colibri, since they aren't meant to be run directly
# by users. COLI_ENGINE overrides the engine path explicitly if neither
# guess is right (e.g. a custom packaging layout).
_EXE = ".exe" if sys.platform == "win32" else ""
_LIBEXEC = os.path.join(os.path.dirname(HERE), "libexec", "colibri")
_here_colibri = os.path.join(HERE, "colibri" + _EXE)
_here_glm = os.path.join(HERE, "glm" + _EXE)

if os.environ.get("COLI_ENGINE"):
    GLM = os.environ["COLI_ENGINE"]
    TOOLS = os.path.join(os.path.dirname(GLM), "tools")
elif os.path.exists(_here_colibri):
    GLM = _here_colibri
    TOOLS = os.path.join(HERE, "tools")
elif os.path.exists(_here_glm):
    GLM = _here_glm
    TOOLS = os.path.join(HERE, "tools")
else:
    GLM = os.path.join(_LIBEXEC, "colibri" + _EXE)
    TOOLS = os.path.join(_LIBEXEC, "tools")
    sys.path.insert(0, _LIBEXEC)   # so `import resource_plan`, `doctor`, `openai_server` still resolve

# No invented default: a path from whoever's machine this was written on is worse than
# no path at all, because every downstream error then names a directory the user never
# chose and cannot act on (#724). None here, and each command says so in its own voice.
DEF_MODEL = os.environ.get("COLI_MODEL")
NO_MODEL_HINT = "pass --model <dir>, or set COLI_MODEL=<dir>"
END   = b"\x01\x01END\x01\x01\n"
READY = b"\x01\x01READY\x01\x01\n"

# ---------- palette & stile ----------
def _c(n): return f"\033[38;5;{n}m"
class C:
    teal=_c(37); cyan=_c(80); mag=_c(170); org=_c(208); grn=_c(78); yel=_c(179)
    dim="\033[2m"; b="\033[1m"; r="\033[0m"; gray=_c(242); dgray=_c(238)
    @staticmethod
    def off():
        for k,v in vars(C).items():
            if isinstance(v,str) and v.startswith("\033"): setattr(C,k,"")
TTY = sys.stdout.isatty() or os.environ.get("COLI_COLOR")=="1"
if not TTY: C.off()

# ---------- colibri 8-bit (pixel art, 2 pixel verticali per carattere) ----------
SPRITE = [
    "....MMM.........",
    "...MMMMM..w.....",
    "....MMMM.ww.....",
    "OOOOTTeTCC......",
    "....TTTTTCC.....",
    ".....TTTTCC.....",
    "......TTCC......",
    ".......TC.......",
    "........C.......",
    "................",
]
PAL = {"M":170, "T":37, "C":80, "O":208, "e":231, "w":80, ".":None}

def sprite_lines():
    if not TTY:
        return ["  (\\   ", "   )·>  ", "  / \\   ", "        ", "        "]
    out=[]
    for y in range(0,len(SPRITE),2):
        top, bot = SPRITE[y], SPRITE[y+1] if y+1<len(SPRITE) else "."*len(SPRITE[y])
        row=""
        for x in range(len(top)):
            ct, cb = PAL.get(top[x]), PAL.get(bot[x])
            if ct is None and cb is None: row+= "\033[0m "
            elif ct is not None and cb is None: row+= f"\033[38;5;{ct}m\033[49m▀"
            elif ct is None and cb is not None: row+= f"\033[38;5;{cb}m\033[49m▄"
            else: row+= f"\033[38;5;{ct}m\033[48;5;{cb}m▀"
        out.append(row+"\033[0m")
    return out

# Display names for the banner, keyed on the RAW model_type -- deliberately not
# on model_arch(), which returns "glm" for everything it does not recognise.
# That fallback is right for picking an engine (colibri.c is the general path)
# and wrong for telling someone what they loaded: it is why the banner said
# GLM-5.2 over an Inkling, a Kimi K3 and a DeepSeek V4 alike.
#
# Parameter counts are the roster's, i.e. the README's, so the two cannot drift
# apart quietly. A model_type absent from this table is NOT forced into a name
# here -- it prints its own model_type and its measured geometry instead.
_BANNER_MODELS = tuple((family.model_types[0], family.display_name, family.display_scale)
                       for family in all_families())

def model_banner_line(model):
    """Third banner line: what is actually loaded, or the generic tagline.

    Reads config.json and stats the shards. No safetensors headers are parsed:
    this runs before every command, and a banner may not cost a scan of a
    400 GB checkpoint.
    """
    # Senza un modello non si nomina un modello. Diceva "GLM-5.2 · 744B", che
    # era la famiglia di punta quando e' stata scritta e che dal #1367 non si
    # chiama nemmeno piu' cosi'; una riga che invecchia perche' cita un caso
    # invece di descrivere il programma. Il conto viene dal registry: aggiungere
    # una famiglia lo aggiorna, dimenticarsene non lo sbaglia.
    generic = (f"{len(_BANNER_MODELS)} model families · "
               f"MoE experts streamed from disk · CPU")
    if not model:
        return generic
    try:
        with open(os.path.join(model, "config.json"), encoding="utf-8") as f:
            cfg = json.load(f)
    except (OSError, ValueError, TypeError):
        return generic
    kind = (cfg.get("model_type") or "").lower()
    geometry = cfg
    try:
        resolved = resolve_model(model)
        family = resolved.descriptor
        # Size-aware: a family's model_type can span checkpoints an order of
        # magnitude apart (Qwen3.6-35B and Qwen3.8-2.4T both resolve to
        # qwen36). display_for names the geometry actually on disk, and
        # falls back to the model_type when it recognises none.
        name, scale = display_for(resolved)
        geometry = resolved.family_config
        # display_scale e' la taglia del checkpoint di riferimento. Un REAP ha
        # lo stesso model_type e meno esperti: "284B" su un 150B e' #1367
        # di nuovo. Quando la geometria non e' quella del riferimento si
        # lascia parlare la geometria, che e' misurata e non puo' mentire.
        experts = cfg.get("n_routed_experts")
        if family.reference_experts and experts and experts != family.reference_experts:
            scale = ""
    except (FamilyConfigError, UnknownFamilyError):
        # Unknown checkpoint: say so with its own words rather than guess.
        name = kind or "unknown model"
        scale = ""

    bits = [name]
    experts = geometry.get("n_routed_experts") or geometry.get("num_experts")
    if scale:
        bits.append(f"{scale} MoE" if experts else scale)
    else:
        layers = geometry.get("num_hidden_layers")
        if layers and experts:
            bits.append(f"{layers}L x {experts}E MoE")
        elif layers:
            bits.append(f"{layers} layer")
    try:
        size = sum(os.path.getsize(os.path.join(model, x))
                   for x in os.listdir(model) if x.endswith(".safetensors"))
        if size >= 1e9:
            # One decimal below 10 GB so a small checkpoint does not read as
            # "0 GB"; whole numbers above, where a decimal is noise.
            bits.append(f"{size/1e9:.1f} GB on disk" if size < 1e10
                        else f"{size/1e9:.0f} GB on disk")
        elif size:
            bits.append(f"{size/1e6:.0f} MB on disk")
    except OSError:
        pass
    return " · ".join(bits)

def banner(sub="", *, model=None):
    # model= is keyword-only on purpose: banner(sub) is called from a dozen
    # places and from open PRs, and a second positional argument here would
    # silently be read as a path.
    sp=sprite_lines()
    txt=[
        f"{C.teal}{C.b}colibri{C.r} {C.dim}v{_version}{C.r}",
        f"{C.dim}tiny engine, immense model{C.r}",
        f"{C.gray}{model_banner_line(model)}{C.r}",
        f"{C.dgray}{sub}{C.r}" if sub else "",
        "",
    ]
    print()
    for i,s in enumerate(sp):
        t = txt[i] if i<len(txt) else ""
        print(f"  {s}   {t}")
    print(f"  {C.dgray}{'─'*58}{C.r}")

def hline(w): return f"{C.dgray}{'─'*w}{C.r}"

# ---------- util ----------
def term_w(): return min(shutil.get_terminal_size((80,20)).columns, 100)

def _read_pasted_lines(stream):
    """Collect complete lines already queued after an interactive paste.

    ``input()`` is line-oriented on libedit (including the FreeBSD build), so a
    pasted block can return its first line while the remaining lines stay in
    the terminal input queue.  Read only data that is immediately ready; a
    normal Enter therefore remains a one-line prompt and typing for the next
    turn is not blocked here.
    """
    lines = []
    try:
        if sys.platform == "win32":
            import msvcrt
            pending = []
            while msvcrt.kbhit():
                ch = msvcrt.getwch()
                if ch in ("\r", "\n"):
                    lines.append("".join(pending))
                    pending = []
                elif ch not in ("\x00", "\xe0"):
                    pending.append(ch)
            if pending:
                lines.append("".join(pending))
            return lines
        import select
        while select.select([stream], [], [], 0)[0]:
            line = stream.readline()
            if not line:
                break
            lines.append(line.rstrip("\r\n"))
    except (ImportError, OSError, ValueError):
        # Some stdin implementations (notably embedded consoles) cannot be
        # polled.  Keeping the first line is safer than blocking the chat loop.
        pass
    return lines

def read_prompt(prompt=""):
    """Read one prompt, retaining additional lines from a pasted block."""
    first = input(prompt)
    if not TTY:
        return first
    return "\n".join([first, *_read_pasted_lines(sys.stdin)])

def prompt_box_lines(message, width):
    """Wrap a prompt for the terminal box without losing explicit newlines."""
    lines = []
    for logical in message.splitlines() or [""]:
        lines.extend(textwrap.wrap(logical, width=max(1, width),
                                   replace_whitespace=False,
                                   drop_whitespace=False) or [""])
    return lines

def prompt_input_rows(message, columns):
    """Return the number of terminal rows occupied by the readline input."""
    rows = 0
    for index, logical in enumerate(message.splitlines() or [""]):
        prefix = 6 if index == 0 else 0  # ``  │ › `` before the first line
        rows += max(1, (prefix + len(logical) + columns - 1) // columns)
    return rows

def redraw_prompt_box(message, width):
    """Replace readline's raw echo with a wrapped, multiline prompt box."""
    clean = message.strip("\r\n")
    columns = shutil.get_terminal_size((80,20)).columns
    used = prompt_input_rows(clean, columns)
    sys.stdout.write(f"\x1b[{used}A\x1b[0J")
    inner = width - 3
    for index, line in enumerate(prompt_box_lines(clean, inner)):
        prefix = f"{C.teal}{C.b}›{C.r}" if index == 0 else " "
        print(f"  {C.dgray}│{C.r} {prefix} {line}{' '*(inner-len(line))}{C.dgray}│{C.r}")
    print(f"  {C.dgray}╰{'─'*width}╯{C.r}")
    return clean

def model_arch(model):
    return resolve_model(model).descriptor.id

def engine_for(model):
    family = resolve_model(model).descriptor
    if os.environ.get("COLI_ENGINE"):
        if (family.id != "glm" and
                os.environ.get("COLI_DOCKER_GLM_ONLY") == "1"):
            raise UnknownFamilyError(
                f"this image contains only the GLM engine; {family.display_name} "
                "needs a full release archive or source build")
        return GLM
    if family.id == "glm":
        return GLM
    name = family.engine_artifact
    local = os.path.join(HERE, name + _EXE)
    return local if os.path.exists(local) else os.path.join(_LIBEXEC, name + _EXE)

def require_model(model, file="tokenizer.json", engine=None, target="", build="coli build"):
    """Shared model-directory / required-file / engine presence check.

    `need_model` and `need_worker_model` differ only in which file they require
    (tokenizer.json vs config.json) and whether non-GLM families are allowed
    (expert workers live in the GLM engine only). The exit wording and
    the check order are shared so the two validators cannot drift apart.
    """
    if not model:
        sys.exit(f"{C.yel}no model directory given.{C.r}\n  {NO_MODEL_HINT}")
    if not os.path.isdir(model):
        sys.exit(f"{C.yel}model not found:{C.r} {model}\n  set COLI_MODEL or use --model")
    if file and not os.path.exists(os.path.join(model, file)):
        sys.exit(f"{C.yel}{file} is missing from {model}{C.r}")
    if engine and not os.path.exists(engine):
        label = f"{target} " if target else ""
        sys.exit(f"{C.yel}{label}engine is not built.{C.r} Run: {build}")

def need_model(model, engine=None):
    # Directory/file checks come FIRST: resolving the family reads config.json,
    # and a missing model dir must say "model not found", not "unsupported model".
    require_model(model, file="tokenizer.json")
    engine = engine or engine_for(model)
    target = resolve_model(model).descriptor.build_target
    require_model(model, file="tokenizer.json", engine=engine, target=target,
                  build=f"make -C c {target}")

# One-shot runs keep the historic 1024. Interactive sessions get 16384, because
# the browser and the TUI both offer a per-request limit and the server treats
# --ngen as a hard ceiling it clamps to (#260) -- so a low ceiling makes the
# user's own control inert instead of merely cautious. Still a ceiling, not a
# target: generation ends at EOS either way.
def ngen_for(a, interactive=False, family=None):
    if getattr(a, "ngen", None): return a.ngen
    if family is not None:
        if isinstance(family, str): family = family_by_id(family)
        limits = family.limits
        return limits.interactive_max_output if interactive else limits.default_max_output
    return 16384 if interactive else 1024

_PROFILE_CAP_ENV = "COLI_PROFILE_CAP"
_PLAN_CAP_ENV = "COLI_PLAN_CAP"

def apply_measured_profile(env, profile, explicit_env, explicit_cap):
    """Apply scheduling/resource results while preserving CLI precedence."""
    from autotune import apply_profile
    result=apply_profile(env,profile,explicit_env)
    # This is a launcher-private bridge for engines whose cache capacity is an
    # argv value rather than an environment variable. It must never leak from a
    # parent shell or override an explicit --cap.
    result.pop(_PROFILE_CAP_ENV,None)
    measured=profile.get("winner",{}).get("cap")
    if explicit_cap is None and isinstance(measured,int) and not isinstance(measured,bool):
        result[_PROFILE_CAP_ENV]=str(measured)
    return result

def cap_for_launch(explicit_cap, env, fallback):
    """Resolve a measured cap for direct-engine launchers; --cap always wins."""
    if explicit_cap is not None:
        return explicit_cap
    try:
        measured=int(env.pop(_PROFILE_CAP_ENV,""))
    except (TypeError,ValueError):
        measured=0
    try:
        planned=int(env.pop(_PLAN_CAP_ENV,""))
    except (TypeError,ValueError):
        planned=0
    return measured if measured>=1 else planned if planned>=1 else fallback

def operator_cap(a, arch):
    """Return a real operator cap, including GLM's documented CAP channel."""
    cap=getattr(a,"cap",None)
    if cap is not None or arch!="glm":
        return cap
    try:
        cap=int(os.environ.get("CAP","0"))
    except (TypeError,ValueError):
        cap=0
    return cap if cap else None

def env_for_engine(a, arch, plan=None):
    if arch == "glm":
        return env_for(a)
    explicit_env = set(os.environ)
    env = os.environ.copy()
    env.pop(_PROFILE_CAP_ENV,None)
    env.pop(_PLAN_CAP_ENV,None)
    family=family_by_id(arch)
    if getattr(a,"ram",0): explicit_env.add("RAM_GB")
    # OMP_NUM_THREADS for the sister engines that do not size their own team.
    #
    # #805 set this from physical cores "on every platform" -- but only inside
    # env_for(), which this function calls only for glm. inkling, kimi_k3,
    # olmoe and deepseek_v4 all took the branch below and got libgomp's
    # default, nproc, i.e. LOGICAL cores. On any SMT host that is a 2x
    # over-subscription of a memory-bound int4 GEMV, which is the collapse
    # measured in #718 (2.3x on Zen3) and again here: an i7-1355U reports 6
    # physical and 12 logical, and DeepSeek V4 was running the 12.
    #
    # Same defect shape as the one #805 fixed, one level out: a mechanism that
    # reaches one engine and not its siblings.
    #
    # DeepSeek V4 is now the exception: its C runtime sizes the team from
    # logical CPUs minus its expert-loader reservation. Setting the physical
    # count here makes that runtime mistake a launcher default for a user
    # override and bypass its loader-aware policy. The other engines still
    # need the physical-core default below.
    #
    # setdefault, so an explicit OMP_NUM_THREADS still wins.
    if not env.get("COLI_NO_OMP_TUNE"):
        if arch != "deepseek_v4":
            from resource_plan import physical_cpu_count
            env.setdefault("OMP_NUM_THREADS", str(physical_cpu_count()))
        else:
            env.setdefault("OMP_WAIT_POLICY", "active")
            env.setdefault("GOMP_SPINCOUNT", "200000")
            env.setdefault("OMP_DYNAMIC", "FALSE")
            if sys.platform != "win32":
                env.setdefault("OMP_PROC_BIND", "close")
                env.setdefault("OMP_PLACES", "cores")
    env["NGEN"] = str(ngen_for(a, family=arch))
    if a.temp is not None: env["COLI_TEMP"] = str(a.temp)
    # SNAP is how every engine but GLM learns its model directory. env_for()
    # sets it for GLM and openai_server.py sets it for the gateway, so `coli
    # chat` and `coli serve` worked while `coli run` handed the sister engines
    # an environment without it: OLMoE exited with "started without a model"
    # (#1501). Set it here, once, for all of them.
    if getattr(a, "model", None):
        env.setdefault("SNAP", os.path.abspath(a.model))
    if arch == "olmoe":
        env["CHAT"] = "1"
        env["MAX_NEW"] = str(ngen_for(a, family=arch))
        # olmoe reads COLI_TEMP like every other engine (line above). Setting the
        # legacy TEMP alias here also clobbered %TEMP% with "0.7" for every child
        # of the chat process on Windows — and olmoe never read COLI_TEMP, so the
        # ONLY working channel was a poisoned system variable (#509's exact shape).
    # --ram reaches the engines that read it. kimi_k3 gained a RAM budget in
    # #855; before that `grep -c RAM_GB c/kimi_k3.c` returned 0, so `coli chat
    # --ram 242` on Kimi K3 set an environment variable nobody looked at and the
    # user's session ran itself out of memory with the flag apparently set.
    if arch in ("deepseek_v4", "kimi", "glm53", "olmoe"):
        if a.ram: env["RAM_GB"] = str(a.ram)
    if arch == "glm53":
        # I densi vanno a int4 di default: sono 9,7 B parametri su 321, e in
        # BF16 sono 19 GB che nessuno recupera. Chi vuole la precisione piena
        # mette GLM53_BITS=32, e questo setdefault gli lascia la scelta.
        env.setdefault("GLM53_BITS", "4")
        # Budget della cache degli esperti. Se l'utente ha dato --ram, si lascia
        # spazio a densi, stato e sistema invece di prendersi tutto.
        if a.ram:
            try:
                spare = max(2.0, float(a.ram) - 8.0)
                env.setdefault("GLM53_EXPERT_GB", f"{spare:.1f}")
            except (TypeError, ValueError):
                pass
    if a.ctx:
        limits = family_by_id(arch).limits
        # #1376: the engine clamps silently to its hard ceiling. A flag that is
        # accepted and then ignored is worse than one refused with the number.
        if limits.max_context and a.ctx > limits.max_context:
            sys.exit(f"{C.yel}--ctx {a.ctx} exceeds what {family_by_id(arch).display_name} "
                     f"supports ({limits.max_context} tokens).{C.r}\n"
                     f"  Pass --ctx {limits.max_context} or less.")
        env[limits.context_env] = str(a.ctx)
    if arch == "deepseek_v4":
        # Speculation stays opt-in. Real multi-turn chat measured only 1/15
        # accepted prompt-lookup candidates; recurrent-state replay dominated
        # the visible decode. V4_DRAFT=5 remains available for repeated code.
        env.setdefault("V4_DRAFT", "0")
        # Full three-stage DSpark remains opt-in.  In a real multi-turn chat it
        # accepted only 10/24 candidates and rejected-suffix replay turned a
        # 14-token answer into 495 seconds.  V4_MTP=1 still enables A/B runs;
        # exact zero-I/O prompt lookup above remains enabled by default.
        env.setdefault("V4_MTP", "0")
        env.setdefault("V4_MTP_DRAFT", "3")
        env.setdefault("V4_MTP_GB", "0.45")
        env.setdefault("V4_MTP_MISS", "96")
        env.setdefault("V4_MTP_MIN", "3")
        env.setdefault("V4_MTP_CONF", "0.55")
        # GPU mirroring of the MTP draft experts stays opt-in: the CUDA kernels
        # accumulate fp32 differently from the CPU refs and speculative
        # acceptance needs a bit-exact draft. V4_MTP_GPU=1 mirrors GLM's
        # COLI_CUDA_MTP switch for CUDA-driven speculative decoding.
        env.setdefault("V4_MTP_GPU", "0")
    # getattr: tests build partial Namespaces for env_for_engine.
    gpu = getattr(a, "gpu", None)
    vram = getattr(a, "vram", None)
    if not family.supports_accelerator:
        if gpu not in (None,"none"):
            sys.exit(f"{C.yel}{family.display_name} currently supports CPU only; --gpu is unavailable{C.r}")
        if vram and gpu!="none":
            sys.exit(f"{C.yel}{family.display_name} currently supports CPU only; --vram is unavailable{C.r}")
    if gpu is not None:
        env.pop("COLI_GPU", None); env.pop("COLI_GPUS", None)
        if gpu == "none":
            env["COLI_CUDA"] = "0"
            env.pop("CUDA_EXPERT_GB", None); env.pop("CUDA_DENSE", None)
        else:
            if arch == "deepseek_v4":
                if not dsv4_cuda_available(a.model):
                    sys.exit(f"{C.yel}--gpu needs the CUDA build:{C.r} {dsv4_cuda_build_hint()}")
            elif not cuda_binary(engine_for_gpu_check(a)):
                sys.exit(f"{C.yel}--gpu needs the CUDA build:{C.r} the engine binary is CPU-only")
            env["COLI_CUDA"] = "1"
            if gpu != "auto":
                env["COLI_GPUS"] = gpu
            env.setdefault("CUDA_DENSE", "1")
    if vram and gpu != "none":
        if arch == "deepseek_v4":
            if not dsv4_cuda_available(a.model):
                sys.exit(f"{C.yel}--vram needs the CUDA build:{C.r} {dsv4_cuda_build_hint()}")
        elif not cuda_binary(engine_for_gpu_check(a)):
            sys.exit(f"{C.yel}--vram needs the CUDA build:{C.r} the engine binary is CPU-only")
        env["COLI_CUDA"] = "1"
        env["CUDA_EXPERT_GB"] = str(vram)
    if getattr(a, "auto_tier", False):
        # GLM has historically applied both the resource plan and its measured
        # profile inside env_for().  The sibling launcher returned above that
        # code path, so #1196 could measure and save a perfectly valid Kimi,
        # Inkling, OLMoE, Qwen or V4 profile that chat/serve never loaded.
        from resource_plan import build_plan, environment_for_plan
        try:
            if plan is None:
                ram,ctx,devices,vram=resource_request(a,env)
                plan=build_plan(a.model,ram,ctx,devices,vram,policy=a.policy,
                                kv_slots=requested_kv_slots(a,env))
            plan_must_not_choose_omp=("OMP_NUM_THREADS" not in env
                                      and (arch == "deepseek_v4"
                                           or env.get("COLI_NO_OMP_TUNE")))
            gpu_off_by_request = env.get("COLI_CUDA") == "0"
            cuda_on = plan_cuda_enabled(a, arch, env)
            env=environment_for_plan(plan,env,cuda_enabled=cuda_on)
        except (OSError,ValueError,json.JSONDecodeError) as error:
            sys.exit(f"{C.yel}invalid resource plan:{C.r} {error}")
        # Outside the try on purpose: this only prints, and a display helper
        # that stumbled inside it would be reported as "invalid resource plan"
        # and take the launch down with it.
        if not cuda_on: report_unapplied_vram_tier(plan, gpu_off_by_request)
        if plan_must_not_choose_omp:
            # environment_for_plan supplies the physical-core default used by
            # the other engines. V4 deliberately computes logical CPUs minus
            # its loader reservation at startup; COLI_NO_OMP_TUNE likewise
            # promises no automatic team. Keep those baselines unless a
            # measured profile (V4 only, below) or the user explicitly chooses
            # a team.
            env.pop("OMP_NUM_THREADS",None)
        if not getattr(a, "no_tune_profile", False):
            from autotune import load_profile
            profile=load_profile(plan,a.model,engine_for(a.model))
            if profile:
                env=apply_measured_profile(env,profile,explicit_env,
                                           getattr(a,"cap",None))
                gain=100.0*profile["gain"]
                print(f"  {C.dim}[TUNE] applied measured profile · +{gain:.1f}% "
                      f"calibration throughput{C.r}",file=sys.stderr)
    return env

def dsv4_cuda_available(model=None):
    # Windows loads the tier as a runtime DLL next to the engine; Linux links
    # it straight into the binary (nvcc, -lcudart), so the DLL check there
    # rejected every valid `make deepseek-v4 CUDA=1` build (#1219). Detect the
    # Linux build through the dynamic libcudart entry added by the in-tree
    # CUDA=1 target. DeepSeek V4 has no HIP backend, so libamdhip64 alone must
    # not make a CPU engine look GPU-capable.
    eng = engine_for(model) if model else GLM
    if sys.platform == "win32":
        # backend_loader_dsv4.c tries the DeepGEMM build first and the generic
        # one second, so either next to the engine starts the tier (the same
        # two names `coli doctor` accepts).
        return any(os.path.exists(os.path.join(os.path.dirname(eng), name))
                   for name in ("coli_cuda_dsv4_dg.dll", "coli_cuda_dsv4.dll"))
    if sys.platform == "linux" and os.path.exists(eng):
        try:
            linked = subprocess.run(["ldd", eng], capture_output=True, text=True, timeout=3)
            return any("libcudart" in line and "not found" not in line
                       for line in linked.stdout.splitlines())
        except (OSError, subprocess.SubprocessError):
            return False
    return False

def dsv4_cuda_build_hint():
    if sys.platform == "win32":
        return "build coli_cuda_dsv4.dll (make cuda-dsv4-dll)"
    if sys.platform == "linux":
        return "make deepseek-v4 CUDA=1 (this binary is CPU-only)"
    return "the DeepSeek V4 CUDA tier is supported only on Linux and Windows"

def need_worker_model(model):
    require_model(model, file="config.json")
    family = resolve_model(model).descriptor
    if family.id != "glm":
        sys.exit(f"{C.yel}cluster expert workers currently support the GLM engine only;{C.r} "
                 f"{family.display_name} models cannot serve experts")
    engine = engine_for(model)
    require_model(model, file="config.json", engine=engine, target=family.build_target,
                  build=f"make -C c {family.build_target}")
    return engine

def engine_for_gpu_check(a):
    """The family's engine path for cuda_binary(), or None (the GLM default)
    when the model cannot be resolved here: the check must not turn a missing
    config.json into a different error than the one need_model() gives."""
    model = getattr(a, "model", None)
    if not model: return None
    try: return engine_for(model)
    except (FamilyConfigError, UnknownFamilyError, OSError): return None

def cuda_binary(engine=None):
    """Was `engine` (default: the GLM binary) built with the GPU backend?
    Callers that know the family pass its engine: checking colibri.exe for a
    qwen36 run refused --gpu on a correct qwen36 CUDA build (#1533)."""
    engine = engine or GLM
    if not os.path.exists(engine): return False
    if sys.platform == "linux":
        try:
            linked=subprocess.run(["ldd",engine],capture_output=True,text=True,timeout=3)
            # Detect both CUDA (libcudart) and HIP (libamdhip64) builds.
            return any(("libcudart" in line or "libamdhip64" in line) and "not found" not in line
                       for line in linked.stdout.splitlines())
        except (OSError,subprocess.SubprocessError): return False
    if sys.platform == "win32":
        # Windows CUDA_DLL=1 builds never link libcudart directly: glm.exe loads
        # coli_cuda.dll at runtime via LoadLibrary (backend_loader.c), so there's no
        # import-table entry for ldd/dumpbin to see. Detect the COLI_CUDA build via a
        # marker string baked into glm.c's #ifdef COLI_CUDA block instead, and require
        # coli_cuda.dll to actually sit next to glm.exe (else CUDA init fails at startup).
        try:
            with open(engine,"rb") as f: built=b"[CUDA] mode: routed experts" in f.read()
        except OSError: return False
        return built and os.path.exists(os.path.join(os.path.dirname(engine),"coli_cuda.dll"))
    return False

def resource_request(a, env):
    family=resolve_model(a.model).descriptor if getattr(a,"model",None) else family_by_id("glm")
    ctx=a.ctx or int(env.get(family.limits.context_env,family.limits.default_context))
    if ctx<1 or ctx>family.limits.max_context:
        raise ValueError(f"--ctx must be between 1 and {family.limits.max_context} "
                         f"for {family.display_name}")
    def num(v):
        try: return float(v)
        except ValueError: return 0.0  # "auto": the planner sizes it; the engine reads the string itself
    ram=a.ram or num(env.get("RAM_GB",0))
    vram=a.vram or num(env.get("CUDA_EXPERT_GB",0))
    gpu=a.gpu
    if gpu is None:
        gpu=env.get("COLI_GPUS",env.get("COLI_GPU","auto"))
    devices=None if gpu=="auto" else ([] if gpu=="none" else
        [int(value) for value in gpu.split(",")])
    return ram,ctx,devices,vram


def requested_kv_slots(a, env):
    value=getattr(a,"kv_slots",None)
    if value is None: value=int(env.get("COLI_KV_SLOTS","1"))
    return value

def plan_cuda_enabled(a, arch, env):
    """Should --auto-tier apply the VRAM tier that `coli plan` printed?

    Yes exactly when the engine that is about to run was built with a GPU
    backend, which is the question the GLM path has always asked
    (`has_cuda=cuda_binary()`), and which is what --auto-tier documents: apply
    the plan.

    The sibling path used to answer a narrower question -- "did the user pass
    --gpu or --vram" -- and the comment that guarded it was right at the time:
    cuda_binary() could only inspect the GLM binary, so a perfectly good qwen36
    CUDA build looked CPU-only to it and auto-tier had to assume the worst
    rather than risk launching a CPU-only sibling as a CUDA one. Since #1533
    the check takes the engine that will actually run, so the thing that
    contract was protecting against can be TESTED instead of assumed, and
    assuming it costs the user the tier they were just shown. On the box in
    #1581 that was 11.8 tok/s against 21, with nothing on screen to explain it.

    A CPU-only build still never gets a CUDA launch: that is what the check
    below says. `--gpu none` (COLI_CUDA=0) is still the off switch, and an
    explicit --gpu/--vram has already proved the build before we get here.
    """
    value = env.get("COLI_CUDA")
    if value == "1": return True      # --gpu / --vram validated the build above
    if value == "0": return False     # --gpu none, or the user set it by hand
    # DeepSeek V4 links its tier differently on each platform, so it has its own
    # probe; using cuda_binary() here would reject valid V4 CUDA builds (#1219).
    if arch == "deepseek_v4": return dsv4_cuda_available(getattr(a, "model", None))
    return cuda_binary(engine_for_gpu_check(a))


def report_unapplied_vram_tier(plan, gpu_off_by_request=False):
    """Say so when a real VRAM tier from the plan is not going into the launch.

    After plan_cuda_enabled() there are only two ways to get here, and the user
    can act on exactly one of them, so the notice says which: the GPU was turned
    off by request, or the engine binary has no GPU backend to turn on.

    Being told matters because `coli plan` and `coli doctor` print the VRAM tier
    as part of the plan and --auto-tier is documented as applying that plan. In
    #1581 the tier vanished with nothing on screen connecting the two, and the
    reporter only found it by watching nvidia-smi.

    The test for "real" is the planner's own: a device qualified to drive
    placement, and a budget above zero. Anything else was not going to be
    applied with the GPU on either, and saying so would be noise.
    """
    try:
        from resource_plan import plans_placement, GB
        vram = plan["tiers"]["vram"]
        devices = [d for d in vram["devices"] if plans_placement(d)]
        budget = vram["budget_bytes"]
    except (ImportError, KeyError, TypeError, AttributeError):
        return
    if not devices or budget <= 0:
        return
    where = ", ".join(f"{d['index']}:{d.get('name') or 'GPU'}" for d in devices)
    why = ("--gpu none turns the GPU off, so the tier is not applied. Drop it "
           "to use the plan as printed."
           if gpu_off_by_request else
           "the engine binary has no GPU backend, so there is nothing to apply "
           "it with. Rebuild the engine with CUDA=1 to use it.")
    print(f"  {C.yel}[PLAN] the plan's VRAM tier is not being applied{C.r}\n"
          f"  {C.dim}{budget / GB:.1f} GB on {where} stays unused: {why}{C.r}",
          file=sys.stderr)


def env_for(a):
    explicit_env = set(os.environ)
    e = dict(os.environ, SNAP=a.model)
    e.pop(_PROFILE_CAP_ENV,None)
    e.pop(_PLAN_CAP_ENV,None)
    if getattr(a,"ram",0): explicit_env.add("RAM_GB")
    # A GLM resource winner is a measured pair (RAM ceiling + argv cap). An
    # explicit cap asks the engine to size against the current RAM plan, not a
    # smaller ceiling remembered from that old pair.
    requested_cap=operator_cap(a,"glm")
    if requested_cap is not None: explicit_env.add("RAM_GB")
    # OMP_NUM_THREADS on EVERY platform, not only Windows.
    #
    # glm.c's self-exec tuning (colibri.c, the COLI_OMP_TUNED block) sets
    # OMP_WAIT_POLICY and GOMP_SPINCOUNT but never the thread count -- and it
    # skips itself entirely when COLI_CUDA or COLI_METAL is on. So on Linux
    # nothing ever sets OMP_NUM_THREADS and libgomp falls back to nproc, i.e.
    # LOGICAL cores. On any SMT host that over-subscribes by 2x, which is
    # precisely what physical_cpu_count()'s own docstring warns about:
    # "two SMT siblings share one AVX-512 unit and contend".
    #
    # Measured, GLM-5.2 744B on 4x RTX A6000 + EPYC 7402P (24 cores / 48
    # threads), CTX=32768, .coli_usage restored between runs:
    #
    #   OMP_NUM_THREADS   routed CPU read   decode      per thread
    #      4               6.48 GB/s        1.28 tok/s   1.62 GB/s
    #      8              12.50             1.97         1.56
    #     16              23.97             2.63         1.49
    #     24 (physical)   30.10             2.90         1.25
    #     48 (logical)    17.35             2.06         0.36   <- today's default
    #
    # setdefault, so an explicit OMP_NUM_THREADS still wins, and the whole
    # thing stays behind COLI_NO_OMP_TUNE like the rest of the OMP tuning.
    if not e.get("COLI_NO_OMP_TUNE"):
        from resource_plan import physical_cpu_count
        e.setdefault("OMP_NUM_THREADS", str(physical_cpu_count()))
    if sys.platform == "win32":
        # COLI_NO_OMP_TUNE spegne SOLO il blocco OMP (stesso perimetro del
        # self-exec di glm.c; presence-based come nel motore: impostarla a
        # qualsiasi valore, anche 0, disattiva). I default I/O piu' sotto
        # restano attivi: kill-switch dedicati = le var stesse (DIRECT=0 ecc.)
        if not e.get("COLI_NO_OMP_TUNE"):
            # parita' col tuning OMP self-exec di glm.c (solo Linux/FreeBSD, e
            # comunque saltato sotto CUDA/Metal): libgomp legge queste variabili
            # prima di main, quindi su Windows vanno nell'ambiente del figlio.
            # niente OMP_PROC_BIND/OMP_PLACES: la libgomp di MinGW non supporta
            # l'affinity su Windows ("Affinity not supported on this configuration")
            for k, v in (("OMP_WAIT_POLICY", "active"),
                         ("GOMP_SPINCOUNT", "200000"),
                         ("OMP_DYNAMIC", "FALSE")):
                e.setdefault(k, v)
        # Default Windows misurati sul box di riferimento (docs/tuning-9950x3d-5090.md),
        # tutti lossless e tutti setdefault (un override esplicito vince sempre):
        # - DIRECT=1: 10.7 GB/s O_DIRECT vs 9.0 buffered (iobench, anche a cache
        #   calda); nel motore 0.48 -> 1.02 tok/s. Upstream #162: 1.47x.
        # - PIPE=1: overlap load/matmul, +8% sopra DIRECT (byte-identico, riordina
        #   solo l'I/O). PIPE_WORKERS resta al default 8 (sweep 4/8/16 piatto).
        # - PILOT_REAL=1: prefetch cross-layer con load veri (unico prefetch
        #   funzionante su Windows: fadvise e' no-op), +11%, hit rate +19 punti.
        e.setdefault("DIRECT", "1")
        e.setdefault("PIPE", "1")
        e.setdefault("PILOT_REAL", "1")
    e["COLI_POLICY"]=a.policy
    if getattr(a, "cluster_workers", None): e["CLUSTER_WORKERS"] = a.cluster_workers
    if getattr(a, "cluster_coordinator", None): e["CLUSTER_COORDINATOR"] = a.cluster_coordinator
    if a.ram:  e["RAM_GB"]=str(a.ram)
    e["NGEN"]=str(ngen_for(a))
    if a.topp: e["TOPP"]=str(a.topp)
    if a.topk: e["TOPK"]=str(a.topk)
    if a.temp is not None: e["COLI_TEMP"]=str(a.temp)   # 0 = greedy; default motore: 1.0 + nucleus 0.95
                                                        # COLI_TEMP, non TEMP: ROCm e Windows leggono $TEMP come dir temporanea (#509)
    if a.repin: e["REPIN"]=str(a.repin)
    if a.ctx: e["CTX"]=str(a.ctx)
    if a.auto_tier:
        from resource_plan import build_plan, environment_for_plan, format_bytes
        if a.gpu is not None:
            e.pop("COLI_GPU",None); e.pop("COLI_GPUS",None)
            if a.gpu=="none":
                e["COLI_CUDA"]="0"; e.pop("CUDA_EXPERT_GB",None); e.pop("CUDA_DENSE",None)
            else: e.pop("COLI_CUDA",None)
        elif e.get("COLI_CUDA")=="0":
            e.pop("COLI_GPU",None); e.pop("COLI_GPUS",None)
            e.pop("CUDA_EXPERT_GB",None); e.pop("CUDA_DENSE",None)
        if a.vram and a.gpu!="none": e["CUDA_EXPERT_GB"]=str(a.vram)
        try:
            ram,ctx,devices,vram=resource_request(a,e)
            plan=build_plan(a.model,ram,ctx,devices,vram,policy=a.policy,
                            kv_slots=requested_kv_slots(a,e))
        except (OSError,ValueError,json.JSONDecodeError) as error:
            sys.exit(f"{C.yel}invalid resource plan:{C.r} {error}")
        has_cuda=cuda_binary()
        e=environment_for_plan(plan,e,has_cuda)
        if not getattr(a, "no_tune_profile", False):
            from autotune import load_profile
            profile=load_profile(plan,a.model,GLM)
            if profile:
                e=apply_measured_profile(e,profile,explicit_env,requested_cap)
                gain=100.0*profile["gain"]
                print(f"  {C.dim}[TUNE] applied measured profile · +{gain:.1f}% calibration throughput{C.r}",
                      file=sys.stderr)
        rt=plan["tiers"]["ram"]; vt=plan["tiers"]["vram"]
        # The plan label used to derive the backend from CUDA alone, so on
        # Apple Silicon it printed " · CPU" even when the calibration child
        # processes ran on Metal (COLI_METAL is inherited through env_for and
        # environment_for_plan, so the measurements are real). Report the
        # active backend: CUDA VRAM first, then Metal, then CPU.
        if has_cuda and vt["devices"]:
            gpu=f" · VRAM {format_bytes(vt['budget_bytes'])}"
        elif str(e.get("COLI_METAL","")).strip() not in ("","0"):
            gpu=" · Metal (unified memory)"
        else:
            gpu=" · CPU"
        print(f"  {C.dim}[PLAN] RAM {format_bytes(rt['budget_bytes'])} · cap {rt['cache_slots_per_layer']}/layer{gpu}{C.r}",file=sys.stderr)
    else:
        # Windows: a bare `coli chat` (no --gpu/--vram/--auto-tier) used to ALWAYS
        # run CPU-only, even on a CUDA build with a GPU present — cuda_binary()
        # returned False on Windows (see above), and nothing set COLI_CUDA without
        # an explicit flag. Now that detection works, auto-enable the GPU when one
        # is detected so `coli chat` Just Works. Scoped to Windows: Linux already
        # has working detection + the explicit-flag UX, and changing bare-chat
        # semantics there is out of scope. Falls back to CPU with a warning if
        # nvidia-smi is missing (discover_gpus can't size VRAM without it).
        # An explicit COLI_CUDA=0 in the environment must win over the implicit
        # auto-enable: before this check, a Windows user setting COLI_CUDA=0 for
        # a CPU baseline silently got a ~12.6 GB VRAM expert tier anyway (the
        # engine's "CPU" rows were GPU-assisted). --gpu none remains the
        # canonical hard off-switch (works on every platform, also clears the
        # CUDA_* sizing vars).
        if (sys.platform == "win32" and a.gpu is None and not a.vram
                and e.get("COLI_CUDA") != "0"):
            if cuda_binary():
                from resource_plan import (discover_gpus, plans_placement, build_plan,
                                           environment_for_plan, format_bytes)
                # Auto-enable is an automatic placement decision, so it may only
                # be made from devices whose free memory is a qualified budget.
                # A discovered-but-unqualified device (free_bytes None: a Windows
                # AMD part found through hipInfo) reaches the engine env here
                # WITHOUT passing through environment_for_plan, so the planner's
                # own gate cannot cover this path -- filter at the source.
                gpus = [g for g in discover_gpus() if plans_placement(g)]
                if gpus:
                    e["COLI_CUDA"]="1"
                    e.setdefault("COLI_GPUS", ",".join(str(g["index"]) for g in gpus))
                    # Reuse the planner so the expert-tier VRAM budget is the real
                    # free VRAM minus the 2 GB reserve — not a guess. Same machinery
                    # as --auto-tier, just without requiring the user to pass it.
                    ram,ctx,devices,vram_req = resource_request(a, e)
                    try:
                        plan=build_plan(a.model,ram,ctx,devices,vram_req,policy=a.policy,
                                        kv_slots=requested_kv_slots(a,e))
                        e.update(environment_for_plan(plan,e,cuda_enabled=True))
                        vt=plan["tiers"]["vram"]
                        # #1409: the dense trunk is read on every token; leaving
                        # it on the CPU next to a 30 GB expert tier is the
                        # slowest possible placement of a card that could hold
                        # it. Put it on the GPU when the plan says it fits with
                        # room for experts, and take its bytes out of the expert
                        # budget: dense uploads are lazy, and a tier that already
                        # filled the card makes them fail (#687). Explicit
                        # CUDA_DENSE / CUDA_EXPERT_GB from the user always win.
                        dense=int(plan["tiers"].get("ram",{}).get("dense_bytes",0) or 0)
                        room=int(vt.get("budget_bytes",0) or 0)-dense
                        # An explicit expert budget owns the placement decision:
                        # do not shrink it or add a trunk outside that budget.
                        if (dense>0 and room>=4*(1024**3) and "CUDA_DENSE" not in e
                                and "CUDA_EXPERT_GB" not in explicit_env):
                            e["CUDA_DENSE"]="1"
                            if "CUDA_EXPERT_GB" in e:
                                e["CUDA_EXPERT_GB"]=f"{room/(1024**3):.3f}"
                        names=",".join(g["name"].strip() for g in gpus)
                        print(f"  {C.dim}[GPU] auto-enabled CUDA · {names} · "
                              f"{format_bytes(vt['budget_bytes'])} expert tier{C.r}", file=sys.stderr)
                    except (OSError,ValueError,json.JSONDecodeError) as error:
                        # Plan failed (e.g. model dir unreadable): don't block the
                        # run, just leave the unsized COLI_CUDA=1 and let the engine
                        # pick its own budget. Engine handles a missing budget.
                        print(f"  {C.yel}[GPU] auto-enable: could not size VRAM ({error}); "
                              f"using engine default{C.r}", file=sys.stderr)
                else:
                    print(f"  {C.yel}[GPU] coli_cuda.dll present but nvidia-smi not found on PATH "
                          f"(cannot size VRAM); running CPU-only. Add nvidia-smi to PATH or pass "
                          f"--vram N to enable CUDA.{C.r}", file=sys.stderr)
            # else: CPU build (no coli_cuda.dll) — stay silent, CPU is correct.
        elif e.get("COLI_CUDA") == "0":
            # honoured off-switch: also drop stale device/sizing vars so the
            # engine can't be re-enabled by leftovers (same as --gpu none).
            e.pop("COLI_GPU",None); e.pop("COLI_GPUS",None)
            e.pop("CUDA_EXPERT_GB",None); e.pop("CUDA_DENSE",None)
        # --gpu/--vram SENZA --auto-tier: prima venivano ignorati in silenzio e il run
        # partiva CPU-only senza alcun avviso — benchmark "GPU" pubblicati per errore (#121).
        if a.gpu is not None:
            e.pop("COLI_GPU",None); e.pop("COLI_GPUS",None)
            if a.gpu=="none":
                e["COLI_CUDA"]="0"; e.pop("CUDA_EXPERT_GB",None); e.pop("CUDA_DENSE",None)
            else:
                if not cuda_binary():
                    sys.exit(f"{C.yel}--gpu needs the CUDA build:{C.r} make colibri CUDA=1 (this binary is CPU-only)")
                e["COLI_CUDA"]="1"
                if a.gpu!="auto": e["COLI_GPUS"]=a.gpu
                e.setdefault("CUDA_DENSE","1")
        if a.vram and a.gpu!="none":
            if not cuda_binary():
                sys.exit(f"{C.yel}--vram needs the CUDA build:{C.r} make colibri CUDA=1 (this binary is CPU-only)")
            e["COLI_CUDA"]="1"; e["CUDA_EXPERT_GB"]=str(a.vram)
    return e

# ---------- rendering markdown in STREAMING per il terminale ----------
class MDStream:
    """Interpreta il markdown della risposta mentre arriva: i ``` diventano riquadri,
    **x** grassetto vero, `x` colorato, # titoli, - puntini. I marker non si vedono mai.
    Regge i chunk spezzati a meta' marker (hold-back) e l'output sporco (``` doppi)."""
    def __init__(self, indent="  "):
        self.ind=indent
        self.cur=""                      # riga parziale non ancora emessa
        self.code=False; self.lang=""
        self.bold=False; self.icode=False
        self.justclosed=False            # l'ultima riga era una chiusura ```? (anti ``` doppi)
        self.printed=0                   # caratteri della riga corrente gia' emessi
    def _fence(self, line):
        lang=line.strip()[3:].strip().strip("`")
        if not self.code:
            if not lang and self.justclosed: return   # ``` orfano dopo una chiusura: rumore, ignora
            self.code=True; self.lang=lang
            sys.stdout.write(f"{self.ind}{C.dgray}\u256d\u2500 {lang or 'code'}{C.r}\n")
        elif lang:                       # ```lang mentre siamo GIA' in code: chiudi e riapri
            sys.stdout.write(f"{self.ind}{C.dgray}\u2570\u2500{C.r}\n{self.ind}{C.dgray}\u256d\u2500 {lang}{C.r}\n")
            self.lang=lang
        else:
            self.code=False; self.justclosed=True
            sys.stdout.write(f"{self.ind}{C.dgray}\u2570\u2500{C.r}\n")
    def _inline(self, txt, out):
        i=0
        while i<len(txt):
            ch=txt[i]
            if ch=="`":
                self.icode=not self.icode
                out.append(C.org if self.icode else C.r); i+=1; continue
            if ch=="*":
                j=i
                while j<len(txt) and txt[j]=="*": j+=1
                if j-i>=2:               # **/***: grassetto on/off, gli asterischi spariscono
                    self.bold=not self.bold
                    out.append(C.b if self.bold else C.r)
                else: out.append("*")    # * singolo: lascialo (moltiplicazioni ecc.)
                i=j; continue
            out.append(ch); i+=1
    def _line(self, line, partial=False):
        if not partial and line.lstrip().startswith("```"):
            self._fence(line); self.printed=0; return
        if line.strip(): self.justclosed=False
        seg=line[self.printed:]          # emetti solo la parte nuova della riga
        out=[]
        if self.code:
            if self.printed==0: out.append(f"{self.ind}{C.dgray}\u2502{C.r} {C.cyan}")
            out.append(seg)
        else:
            if self.printed==0:
                out.append(self.ind)
                st=seg.lstrip()
                if st.startswith("#"):           # titolo: via i #, grassetto teal
                    seg=st.lstrip("#").strip(); out.append(f"{C.teal}{C.b}"); self.bold=True
                elif st.startswith(("- ","* ")): # lista: puntino vero
                    seg=st[2:]; out.append(f"{C.teal}\u2022{C.r} ")
            self._inline(seg,out)
        sys.stdout.write("".join(out)); sys.stdout.flush()
        self.printed=len(line)
        if not partial:                  # fine riga: reset stati inline (robusto ai marker orfani)
            sys.stdout.write(C.r+"\n"); sys.stdout.flush()
            self.bold=self.icode=False
            self.printed=0
    def feed(self, s):
        self.cur+=s
        while "\n" in self.cur:
            line,self.cur=self.cur.split("\n",1)
            self._line(line)
        st=self.cur.lstrip()             # riga parziale: possibile fence? aspetta il newline
        if st and (st.startswith("```") or (len(st)<3 and "```".startswith(st))):
            return
        if st.startswith("#") and self.printed==0:
            return                        # titolo: rendi la riga intera al newline
        hold=0                            # trattieni marker potenzialmente spezzati in coda
        while hold<len(self.cur) and self.cur[-1-hold] in "*`": hold+=1
        safe=self.cur[:len(self.cur)-hold] if hold else self.cur
        if len(safe)>self.printed: self._line(safe, partial=True)
    def close(self):
        if self.cur: self._line(self.cur); self.cur=""
        if self.code:
            sys.stdout.write(f"\n{self.ind}{C.dgray}\u2570\u2500{C.r}"); self.code=False
        sys.stdout.write(C.r); sys.stdout.flush()

class Spinner:
    FRAMES=["⠋","⠙","⠹","⠸","⠼","⠴","⠦","⠧","⠇","⠏"]
    def __init__(self,label,tick=None):
        self.label=label; self.tick=tick; self.suffix=""
        self.stop_evt=threading.Event(); self.t0=time.time(); self.th=None
    def start(self):
        if not TTY: return
        def run():
            i=0
            while not self.stop_evt.is_set():
                el=time.time()-self.t0
                if self.tick and i%8==0:              # ~1 Hz: legge il progresso dal log
                    try: self.suffix=self.tick() or self.suffix
                    except Exception: pass
                suf=f" {C.dgray}· {self.suffix}{C.r}" if self.suffix else ""
                sys.stdout.write(f"\r  {C.teal}{self.FRAMES[i%10]}{C.r} {C.dim}{self.label} {el:.0f}s{C.r}{suf}\033[K")
                sys.stdout.flush(); i+=1; time.sleep(0.12)
        self.th=threading.Thread(target=run,daemon=True); self.th.start()
    def stop(self):
        self.stop_evt.set()
        if self.th: self.th.join(timeout=0.4)
        if TTY: sys.stdout.write("\r\033[K"); sys.stdout.flush()

def engine_diag(p, errlog=None):
    """Lo stdout del motore ha chiuso: il processo e' morto. Dire PERCHE'.
    Un SIGKILL dell'OOM-killer del kernel e' altrimenti invisibile — niente errore,
    niente exit code, il motore muore muto e sembra un bug nostro (issue #305).
    Quasi sempre e' memoria: il picco reale ha sforato la RAM della macchina."""
    try: rc=p.wait(timeout=5)
    except Exception: rc=p.poll()
    if rc is None:                      why="its output closed but the process is still alive"
    elif rc<0:                                        # POSIX: morte per segnale
        try: why=f"killed by {signal.Signals(-rc).name}"
        except Exception: why=f"killed by signal {-rc}"
    elif rc>0:                          why=f"exit code {rc}"
    else:                               why="exited cleanly"
    print(f"\n  {C.yel}[engine terminated: {why}]{C.r}")
    if rc is not None and rc<0 and -rc==getattr(signal,"SIGKILL",-1):
        print(f"  {C.yel}nothing in the engine sends SIGKILL to itself: this is the kernel's\n"
              f"  OOM-killer. The peak RSS exceeded the machine's free memory.\n"
              f"  Lower --ram, lower PIN_GB, or shorten the context.{C.r}")
    if errlog is not None:
        try:
            errlog.flush()
            for _ in range(20):         # il drain thread scrive su errlog quando il child chiude
                tail=open(errlog.name, encoding="utf-8", errors="replace").read().strip()
                if tail: break
                time.sleep(0.05)
            if tail: print(f"  {C.dgray}" + "\n  ".join(tail.splitlines()[-6:]) + f"{C.r}")
        except Exception: pass

def stream_turn(p, sentinel, on_bytes):
    """legge fino alla sentinella; on_bytes riceve i chunk della risposta. Poi legge la riga STAT.
    Il PRIMO Ctrl-C durante lo stream non chiude la sessione: il motore (handler SIGINT)
    chiude il turno per la via del tetto NGEN e noi dreniamo fino alla sentinella.
    Un SECONDO Ctrl-C esce davvero."""
    pend=b""; interrupted=False
    while True:
        try:
            b=p.stdout.read(1)
        except KeyboardInterrupt:
            if interrupted or p.poll() is not None: raise
            interrupted=True
            try: p.send_signal(signal.SIGINT)   # non-TTY: il motore potrebbe non aver visto il Ctrl-C
            except Exception: pass
            print(f"\n  {C.yel}⏹ stopping… (Ctrl-C again to quit){C.r}", flush=True)
            continue
        if b==b"": return None
        pend+=b
        if pend.endswith(sentinel):
            rest=pend[:-len(sentinel)]
            if rest: on_bytes(rest)
            line=p.stdout.readline().decode("utf-8","replace").strip()   # STAT tok tps hit rss
            m=re.match(r"STAT (\S+) (\S+) (\S+) (\S+)", line)
            st={"tok":int(m.group(1)),"tps":float(m.group(2)),"hit":float(m.group(3)),"rss":float(m.group(4))} if m else {}
            if interrupted: st["interrupted"]=True
            return st
        if len(pend)>len(sentinel):
            out=pend[:-len(sentinel)]; pend=pend[-len(sentinel):]
            on_bytes(out)

# ---------- comandi ----------
def cmd_build(a):
    banner("build")
    if not os.path.exists(os.path.join(HERE, "Makefile")):
        sys.exit(f"{C.yel}coli build{C.r} only works from a source checkout (this is an installed copy).\n"
                  f"  Clone https://github.com/JustVugg/colibri and run ./setup.sh, or make -C c colibri.")
    family=resolve_model(a.model).descriptor if a.model else family_by_id("glm")
    target=family.build_target
    sys.exit(subprocess.call(["make","-C",HERE,target]))

def cmd_info(a):
    banner("info", model=a.model)
    if not a.model:
        sys.exit(f"{C.yel}no model directory given.{C.r}\n  {NO_MODEL_HINT}")
    cfgp=os.path.join(a.model,"config.json")
    resolved=None
    try:
        resolved=resolve_model(a.model)
        family=resolved.descriptor
    except (FamilyConfigError,UnknownFamilyError) as error:
        if os.path.exists(cfgp):
            sys.exit(f"{C.yel}unsupported model:{C.r} {error}")
        family=None
    def row(k,v): print(f"   {C.gray}{k:<10}{C.r} {v}")
    if os.path.exists(cfgp):
        c=json.load(open(cfgp))
        # Multimodal wrappers such as Qwen3.8 keep the language architecture
        # below text_config.  The registry has already resolved that section;
        # using the root here produced a recognized family followed by four
        # misleading `None` dimensions in `coli info`.
        arch_c=resolved.family_config if resolved else c
        experts=arch_c.get("n_routed_experts",arch_c.get("num_experts"))
        row("model", a.model)
        row("arch", f"hidden {arch_c.get('hidden_size')} · {arch_c.get('num_hidden_layers')} layer · "
                    f"{experts} expert/layer · top-{arch_c.get('num_experts_per_tok')}")
        sts=[x for x in os.listdir(a.model) if x.endswith('.safetensors')]
        sz=sum(os.path.getsize(os.path.join(a.model,x)) for x in sts)
        row("shards", f"{len(sts)} files · {sz/1e9:.0f} GB on disk")
    else:
        # No config.json means no family, and no family means no engine: say so
        # instead of letting the GLM default below stand in. A directory of shards
        # copied without the small files is the usual shape of this (the Qwen3.8
        # support case: the GLM engine then died on a tensor name it never had).
        try: n_shards=len([x for x in os.listdir(a.model) if x.endswith('.safetensors')])
        except OSError: n_shards=0
        print(f"   {C.yel}config.json is missing{C.r}: coli picks the engine from it, so nothing can run here yet.")
        print(f"   Copy the checkpoint's config.json (with tokenizer.json and model.safetensors.index.json)")
        print(f"   from the model repo next to the {n_shards} shard(s) found here, then run coli info again.")
    try:
        mi=open('/proc/meminfo').read()
        tot=int(re.search(r'MemTotal:\s+(\d+)',mi).group(1))/1e6
        av=int(re.search(r'MemAvailable:\s+(\d+)',mi).group(1))/1e6
        row("RAM", f"{tot:.0f} GB total · {av:.1f} GB available")
    except Exception: pass
    try:
        fs = shutil.disk_usage(a.model if os.path.isdir(a.model) else HERE)
        row("disk", f"{fs.free/1e9:.0f} GB free")
    except OSError:
        row("disk", "? GB (unavailable)")
    if family:
        engine=engine_for(a.model)
        row("engine", "ready ✓" if os.path.exists(engine) else "not built (coli build)")
    else:
        row("engine", "unknown until config.json is present")
    knobs=[]
    if a.ram: knobs.append(f"ram {a.ram}GB")
    if a.topp: knobs.append(f"topp {a.topp}")
    if a.topk: knobs.append(f"topk {a.topk}")
    if knobs: row("tuning", " · ".join(knobs))
    print()

def cmd_plan(a):
    from resource_plan import build_plan, format_plan
    if not a.model:
        sys.exit(f"{C.yel}no model directory given.{C.r}\n  {NO_MODEL_HINT}")
    try:
        ram,ctx,devices,vram=resource_request(a,os.environ)
        if ctx<1: raise ValueError("--ctx must be positive")
        if a.vram<0: raise ValueError("--vram cannot be negative")
        plan=build_plan(a.model,ram,ctx,devices,vram,policy=a.policy,
                        kv_slots=requested_kv_slots(a,os.environ))
    except (OSError, ValueError, json.JSONDecodeError) as error:
        sys.exit(f"{C.yel}cannot create resource plan:{C.r} {error}")
    if a.json:
        print(json.dumps(plan,indent=2))
        return
    banner("plan · Disk / RAM / VRAM", model=a.model)
    print(textwrap.indent(format_plan(plan),"  "))
    print()

def cmd_doctor(a):
    from doctor import exit_code, format_doctor, run_doctor
    if not a.model:
        report={"schema_version":1,"status":"error","model":None,
                "mode":"deep" if a.deep else "standard",
                "checks":[{"id":"config.arguments","status":"fail",
                           "summary":f"no model directory given: {NO_MODEL_HINT}"}],
                "plan":None}
        print(json.dumps(report,indent=2) if a.json else format_doctor(report)); return 2
    try:
        # doctor reports rather than exits, so a missing model is a failed check and not
        # a sys.exit like everywhere else — it is exactly the thing doctor exists to say.
        ram,ctx,devices,vram=resource_request(a,os.environ)
        if ctx<1: raise ValueError("--ctx must be positive")
        if ram<0: raise ValueError("--ram cannot be negative")
        if vram<0: raise ValueError("--vram cannot be negative")
    except ValueError as error:
        report={"schema_version":1,"status":"error","model":os.path.abspath(a.model) if a.model else None,
                "mode":"deep" if a.deep else "standard",
                "checks":[{"id":"config.arguments","status":"fail","summary":str(error)}],
                "plan":None}
        print(json.dumps(report,indent=2) if a.json else format_doctor(report))
        return 2
    engine_error=None
    try:
        engine=engine_for(a.model)
    except (FamilyConfigError,UnknownFamilyError) as error:
        engine=GLM
        engine_error=error
    report=run_doctor(
        # engine_path was hardcoded to GLM, so `coli doctor` inspected colibri
        # no matter which model it was pointed at: a built kimi_k3/inkling was
        # reported "engine is not built", and a stale colibri would have been
        # reported ready for a model that never runs on it (#783).
        a.model, ram, ctx, devices, vram, engine_path=engine, deep=a.deep,
        kv_slots=requested_kv_slots(a,os.environ),
        engine_error=engine_error,
        mirror_dir=os.environ.get("COLI_MODEL_MIRROR"),
    )
    print(json.dumps(report,indent=2) if a.json else format_doctor(report))
    return exit_code(report)

def cmd_tune(a):
    """Measure this exact model/hardware pair and persist the fastest safe path."""
    need_model(a.model)
    if a.tokens < 4:
        sys.exit(f"{C.yel}--tokens must be at least 4{C.r}")
    if a.timeout < 1:
        sys.exit(f"{C.yel}--timeout must be positive{C.r}")
    if not 0.0 <= a.min_gain <= 1.0:
        sys.exit(f"{C.yel}--min-gain must be between 0 and 1{C.r}")
    from autotune import run_tune
    from resource_plan import build_plan
    try:
        ram,ctx,devices,vram=resource_request(a,os.environ)
        plan=build_plan(a.model,ram,ctx,devices,vram,policy=a.policy,
                        kv_slots=requested_kv_slots(a,os.environ))
    except (OSError,ValueError,json.JSONDecodeError) as error:
        sys.exit(f"{C.yel}cannot create tuning plan:{C.r} {error}")
    a.auto_tier=True
    a.no_tune_profile=True
    # Dispatch like every other command. cmd_tune used to hardcode GLM: the
    # banner read config.json and printed "DeepSeek V4 Flash", then calibration
    # launched the GLM engine, which died on "this engine requires n_group=1"
    # (#898). Same shape as #879, one command over -- the launcher naming a
    # family it then mis-dispatched.
    family=resolve_model(a.model).descriptor
    arch=family.id
    if not family.has_gateway_adapter:
        sys.exit(f"{family.display_name}: gateway adapter is not wired")
    engine=engine_for(a.model)
    need_model(a.model,engine)
    base_env=(env_for_engine(a,arch,plan=plan) if arch!="glm" else env_for(a))
    # Prompts have to use the engine's own template: GLM's [gMASK]<sop>
    # markers are ordinary text to every other tokenizer. Sibling engines keep
    # one serve process alive per candidate and alternate these prompts; this
    # preserves their real expert-cache state without training it on one exact
    # request. GLM retains its stronger fixed-token replay protocol.
    raw_prompts=[a.prompt,*(a.rotate_prompt or [_TUNE_ROTATION_PROMPT])]
    if len(set(raw_prompts)) < 2:
        sys.exit(f"{C.yel}tuning rotation needs at least two distinct prompts{C.r}")
    prompts=[tuning_replay_prompt(family,item) for item in raw_prompts]
    prompt=prompts[0]
    banner("tune · measured execution profile", model=a.model)
    if arch=="glm":
        print(f"  fixed replay: {a.tokens} tokens · {a.repeats} run(s) per candidate")
    else:
        active=min(a.repeats,len(prompts))
        print(f"  persistent rotation: {a.tokens} tokens · {a.repeats} request(s) "
              f"per candidate · {active} distinct prompt(s)")
    print("  only quality-preserving scheduling knobs are eligible\n")
    try:
        explicit_resources={key for key in ("RAM_GB","K3_EXPERT_GB")
                            if key in os.environ}
        if a.ram: explicit_resources.add("RAM_GB")
        tuning_cap=operator_cap(a,arch)
        if tuning_cap is not None: explicit_resources.add("cap")
        profile,path=run_tune(
            engine,tuning_cap,base_env,plan,a.model,prompt,arch=arch,family=family,
            tokens=a.tokens,
            repeats=a.repeats,timeout=a.timeout,min_gain=a.min_gain,
            profile_dir=a.profile_dir,
            prompts=prompts,
            explicit_resources=explicit_resources,
            progress=lambda message: print(f"  {C.dim}· {message}{C.r}",flush=True),
        )
    except (OSError,ValueError,RuntimeError,subprocess.SubprocessError) as error:
        sys.exit(f"{C.yel}tuning failed:{C.r} {error}")
    if profile["accepted"]:
        winner=profile["winner"]
        baseline=profile["validation"]["baseline"]["tok_s"]
        print(f"\n  {C.grn}✓ accepted{C.r} {winner['name']}: "
              f"{baseline:.2f} → {winner['tok_s']:.2f} tok/s "
              f"(+{100*profile['gain']:.1f}%)")
        if winner["env"]:
            print(f"  env: {' '.join(f'{key}={value}' for key,value in winner['env'].items())}")
        if "cap" in winner:
            print(f"  cache cap: {winner['cap']} slots/layer")
    else:
        print(f"\n  {C.grn}✓ baseline retained{C.r}: no candidate cleared the "
              f"{100*a.min_gain:.1f}% gain and safety gates")
    print(f"  profile: {path}\n")

def cmd_run(a):
    need_model(a.model)
    family=resolve_model(a.model).descriptor
    arch=family.id
    engine=engine_for(a.model)
    need_model(a.model,engine)
    prompt=" ".join(a.prompt) if a.prompt else sys.exit('usage: coli run "your prompt"')
    banner("run", model=a.model)
    if not family.has_cli_adapter:
        sys.exit(f"{C.yel}coli run is not wired for {family.display_name};{C.r} "
                 f"use coli chat or coli serve")
    if arch=="deepseek_v4":
        prompt_path=None
        try:
            e=env_for_engine(a,arch)
            cmd=[engine,os.path.abspath(a.model)]
            if sys.platform in ("win32", "msys", "cygwin"):
                with tempfile.NamedTemporaryFile(
                        mode="w",encoding="utf-8",newline="",delete=False,
                        prefix="coli-v4-prompt-",suffix=".txt") as stream:
                    stream.write(prompt)
                    prompt_path=stream.name
                cmd += ["--prompt-file",prompt_path]
            else:
                cmd.append(prompt)
            cmd += ["--max-tokens",str(ngen_for(a,interactive=True,family=family))]
            try: memory_gb=float(e.get("RAM_GB","0"))
            except (TypeError,ValueError): memory_gb=0.0
            if memory_gb>0: cmd += ["--memory-gb",str(e["RAM_GB"])]
            if os.environ.get("COLI_THINK","0")=="1": cmd.append("--thinking")
            result=subprocess.call(cmd,env=e)
        finally:
            if prompt_path:
                try: os.unlink(prompt_path)
                except OSError: pass
        sys.exit(result)
    if arch=="glm53":
        e=env_for_engine(a,arch)
        # GLM53 accepts cache/layer as a positional numeric argument.  Keep the
        # same cap_for_launch contract used by the persistent gateway so an
        # explicit `coli run --cap N` reaches the engine instead of being
        # silently ignored by the one-shot path.
        cap = cap_for_launch(a.cap, e, 0)
        cmd=[engine,str(cap),"--model",os.path.abspath(a.model),"--prompt",prompt,
             "--greedy",str(ngen_for(a,interactive=True,family=family))]
        sys.exit(subprocess.call(cmd,env=e))
    if arch=="olmoe":
        # olmoe.c now speaks the gateway's SERVE protocol too (`coli chat`/
        # `coli web`/`coli serve` use it), but `coli run` predates that and
        # its one-shot contract is simpler without it: feed exactly one turn
        # over the engine's plain stdin chat loop and close stdin, the engine
        # emits the answer and exits cleanly. No reason to spin up the gateway
        # for a single non-interactive prompt.
        e=env_for_engine(a,arch)
        # The engine tokenizes the bytes it reads as UTF-8. A text-mode pipe
        # would encode the prompt in the locale's code page instead: a
        # UnicodeEncodeError on cp949/cp932 or the C locale, garbled bytes on
        # cp1252. Hand it the bytes, like the V4 path's UTF-8 prompt file.
        result=subprocess.run(
            [engine, str(cap_for_launch(a.cap,e,16)), "8"],
            input=(prompt+"\n").encode("utf-8"), env=e, check=False)
        sys.exit(result.returncode)
    # template ufficiale GLM-5.2: niente \n dopo i ruoli; <think></think> = risposta diretta (nothink).
    # THINK=1 lascia <think> aperto, stessa convenzione del serve mode (glm.c). EN: THINK=1 leaves
    # <think> open so the engine emits its reasoning block; the default stays nothink.
    tk="<think>" if os.environ.get("THINK","0")=="1" else "<think></think>"
    e=env_for(a); e["PROMPT"]=f"[gMASK]<sop><|user|>{prompt}<|assistant|>{tk}"
    sys.exit(subprocess.call([GLM, str(cap_for_launch(a.cap,e,0))], env=e))

def server_probe(base, api_key=None, timeout=1.5):
    """Is a coli serve alive at `base`? Returns its model_id, or None.
    Probes /health then /v1/models — both cheap, neither touches the engine."""
    import urllib.request, urllib.error
    def get(path):
        req=urllib.request.Request(base.rstrip("/")+path)
        if api_key: req.add_header("Authorization", f"Bearer {api_key}")
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return json.loads(r.read().decode("utf-8","replace"))
    try:
        if get("/health").get("status")!="ok": return None
        data=get("/v1/models").get("data") or []
        return data[0]["id"] if data else None
    except Exception:
        return None

# ---- generation statistics for the chat footer (#1475) ----------------------
# Measurement and rendering apart: gen_stats() counts (exact completion tokens
# from the server's usage block when it sends one, the historical chars/4
# estimate otherwise), render_gen_stats() lays it out for one of three modes.
# The private-engine chat already prints the engine's own exact counters; the
# attached path is where the estimate lived and where tok/s was missing.
CHAT_STATS_MODES = ("full", "compact", "off")

def chat_stats_mode(a):
    mode = getattr(a, "stats", None) or os.environ.get("COLI_CHAT_STATS", "full")
    return mode if mode in CHAT_STATS_MODES else "full"

def gen_stats(reply_text, elapsed, exact_tokens=None, interrupted=False):
    exact = isinstance(exact_tokens, int) and not isinstance(exact_tokens, bool) and exact_tokens >= 0
    tokens = exact_tokens if exact else len(reply_text) // 4
    tps = tokens / elapsed if elapsed > 0 else None
    return {"tokens": tokens, "exact": exact, "elapsed": elapsed, "tps": tps,
            "interrupted": bool(interrupted)}

def render_gen_stats(st, mode):
    """'' for off; otherwise the footer body, without the box glyph or colour:
    full    ~370 tok · 313s · ~1.18 tok/s   (372 tok · 313s · 1.19 tok/s when exact)
    compact ~370 tok · ~1.18 tok/s
    plus ' · ⏹ interrupted' when the answer was cut short."""
    if mode == "off": return ""
    approx = "" if st["exact"] else "~"
    parts = [f"{approx}{st['tokens']} tok"]
    if mode == "full": parts.append(f"{st['elapsed']:.0f}s")
    if st["tps"] is not None: parts.append(f"{approx}{st['tps']:.2f} tok/s")
    if st["interrupted"]: parts.append("⏹ interrupted")
    return " · ".join(parts)



# ---- comandi della chat ------------------------------------------------------
#
# Stanno in un posto solo perche servono a tre cose che devono restare
# d'accordo: il completamento col TAB, l'elenco di /help, e il controllo che
# impedisce a un comando scritto male di partire come messaggio al modello.
# Quest'ultimo e il motivo per cui esistono: prima "/brio merge" con una lettera
# sbagliata finiva dritto al modello, che rispondeva educatamente a una riga di
# comando. Ogni forma si accetta sia con / sia con : -- i comandi storici usano
# i due punti e non si toglie a nessuno l'abitudine.
CHAT_COMMANDS = {
    "brio":  "a | b | c   score the options instead of generating; no options returns to chat",
    "reset": "            start the conversation again",
    "help":  "            this list",
    "quit":  "            leave (also :q)",
}

def chat_command(message):
    """(nome, resto) se la riga e un comando, (None, None) se e un messaggio.

    Il nome torna anche quando NON esiste: e il chiamante a decidere che farne,
    e deve poter distinguere "comando sbagliato" da "testo che comincia con /"."""
    if not message or message[0] not in "/:":
        return None, None
    head, _, rest = message[1:].partition(" ")
    return head.strip().lower(), rest.strip()

def chat_commands_help():
    print(f"  {C.dim}commands:{C.r}")
    for name, what in CHAT_COMMANDS.items():
        print(f"    {C.teal}/{name}{C.r} {C.dim}{what}{C.r}")
    print()

def install_chat_completer():
    """TAB completa i comandi. Senza readline (Windows) si perde solo il TAB."""
    try:
        import readline
    except ImportError:
        return
    names = [p + n for n in CHAT_COMMANDS for p in ("/", ":")]
    def complete(text, state):
        hits = [n + " " for n in names if n.startswith(text)]
        return hits[state] if state < len(hits) else None
    readline.set_completer_delims(" \t\n")
    readline.parse_and_bind("set show-all-if-ambiguous on")   # un TAB solo li mostra tutti
    readline.parse_and_bind("tab: complete")
    readline.set_completer(complete)

def chat_brio(a, base, model_id, msgs, question, options, w):
    """Un giro di modalita brio dalla TUI: una richiesta, una distribuzione.

    Il ciclo (fotografia del prefisso, una lettura per opzione, normalizzazione
    per lunghezza) sta nel gateway, non qui: sono tre cose facili da sbagliare e
    devono stare in un posto solo. Qui si manda la conversazione come contesto
    e si disegnano le barre."""
    import urllib.request, urllib.error
    body=json.dumps({"model":model_id,"messages":msgs,"question":question,
                     "options":options}).encode()
    req=urllib.request.Request(base.rstrip("/")+"/v1/brio", data=body,
                               headers={"Content-Type":"application/json"})
    if a.api_key: req.add_header("Authorization", f"Bearer {a.api_key}")
    print(f"\n  {C.teal}◆ brio{C.r}")
    sp=Spinner("scoring…"); sp.start(); t0=time.time()
    try:
        with urllib.request.urlopen(req) as r:
            out=json.loads(r.read().decode("utf-8","replace"))
    except urllib.error.HTTPError as e:
        sp.stop()
        try: detail=json.loads(e.read().decode("utf-8","replace")).get("error",{}).get("message","")
        except Exception: detail=""
        print(f"  {C.yel}[il server ha rifiutato: HTTP {e.code}{' — '+detail if detail else ''}]{C.r}\n")
        return
    except OSError as e:
        sp.stop(); print(f"  {C.yel}[server irraggiungibile: {e}]{C.r}\n"); return
    sp.stop()
    bars=max(10, min(30, w-34))
    for choice in out.get("choices", []):
        p=choice.get("p",0.0); filled=int(round(p*bars))
        colour=C.grn if choice["option"]==out.get("answer") else C.dgray
        print(f"     {colour}{choice['option'][:22]:<22}{C.r} {colour}{'█'*filled}{C.r}"
              f"{C.dgray}{'░'*(bars-filled)}{C.r} {p*100:5.1f}%"
              f"  {C.dim}{choice.get('tokens',0)} tok{C.r}")
    h=out.get("entropy",1.0)
    # L'entropia e la parte che la generazione non sa dare: non "qual e la
    # risposta" ma "quanto il modello sa di saperlo".
    verdict=(f"{C.grn}sure{C.r}" if h<.4 else
             f"{C.yel}unsure{C.r}" if h<.8 else f"{C.yel}DOES NOT KNOW{C.r}")
    usage=out.get("usage",{})
    print(f"     {C.b}→ {out.get('answer','?')}{C.r}  {C.dim}entropy {h:.3f}{C.r} ({verdict})")
    print(f"  {C.dim}{time.time()-t0:.2f}s · {usage.get('read_tokens',0)} tokens read · "
          f"{usage.get('completion_tokens',0)} generated{C.r}\n")

def chat_attached(a, base, model_id):
    """The chat REPL over HTTP against a running `coli serve`.

    Why this exists (the cold-chat cost, measured): spawning a private engine
    pays 34-136 s of resident load on EVERY start, and begins with an empty
    expert cache — hit rate 4% cold vs 55% warm, a ~10x on early decode. A
    resident server pays load once and keeps the LRU warm across sessions;
    its KV slots reuse the conversation prefix, so a continued chat skips
    re-prefill too. The engine byte-protocol stays untouched — this is plain
    OpenAI SSE over localhost, stdlib only."""
    import urllib.request, urllib.error
    print(f"  {C.grn}✦ attached{C.r} {C.dim}to {base} · model {model_id} · the engine stays warm after you quit{C.r}")
    print(f"  {C.dim}type and press Enter · Ctrl-C stops the answer · :reset starts a new conversation · :q exits{C.r}")
    print(f"  {C.dim}/ for commands · TAB completes · /brio puts the SAME model in brio mode{C.r}\n")
    install_chat_completer()
    msgs=[]
    brio_options=[]          # non vuoto = modalita brio accesa
    w=term_w()-4
    while True:
        if TTY:
            print(f"  {C.dgray}╭{'─'*w}╮{C.r}")
            try: msg=read_prompt(f"  {C.dgray}│{C.r} {C.teal}{C.b}›{C.r} ")
            except EOFError: print(); break
            try: msg=redraw_prompt_box(msg, w)
            except Exception: print(f"  {C.dgray}╰{'─'*w}╯{C.r}")
        else:
            try: msg=input()
            except EOFError: break
        msg=msg.strip("\r\n")
        if msg in (":q","/q","exit"): break
        if not msg.strip(): continue
        # Una riga che comincia con / o : e un COMANDO, giusto o sbagliato che
        # sia: non finisce mai al modello. Prima un refuso partiva come domanda.
        name, rest = chat_command(msg)
        if name is not None and name not in CHAT_COMMANDS:
            print(f"  {C.yel}  comando sconosciuto: {msg.split()[0]}{C.r}")
            chat_commands_help(); continue
        if name in ("quit",): break
        if name == "help": chat_commands_help(); continue
        if name == "reset":
            msgs=[]; brio_options=[]
            print(f"  {C.dim}✦ new conversation{C.r}\n"); continue
        # ---- modalita brio -------------------------------------------------
        # Lo stesso modello con cui stai chattando smette di generare e comincia
        # a punteggiare: gli si danno le opzioni ammesse e lui dice quanto e
        # probabile ciascuna. La conversazione fin qui FA da contesto, quindi
        # /brio a meta chat continua il discorso invece di ricominciarlo.
        if name == "brio":
            spec=rest
            if not spec:
                if brio_options:
                    brio_options=[]
                    print(f"  {C.dim}✦ chat{C.r}\n")
                else:
                    print(f"  {C.yel}  /brio needs the options: /brio merge | request changes | close{C.r}\n")
                continue
            brio_options=[o.strip() for o in spec.split("|") if o.strip()]
            if len(brio_options)<2:
                brio_options=[]
                print(f"  {C.yel}  at least two options are needed, separated by |{C.r}\n")
                continue
            print(f"  {C.grn}✦ brio{C.r} {C.dim}· {len(brio_options)} options · "
                  f"the model stops generating and scores instead · /brio returns to chat{C.r}\n")
            continue
        if brio_options:
            chat_brio(a, base, model_id, msgs, msg, brio_options, w)
            continue
        # --------------------------------------------------------------------
        payload = message_with_images(msg)
        if payload is None: continue          # percorso sbagliato: gia' detto
        msgs.append({"role":"user","content":payload})
        # Il livello di ragionamento e' una leva di tempo, non di gusto: su un
        # motore che streamma gli esperti da disco, "max" vuol dire centinaia
        # di token di riflessione prima della prima parola di risposta.
        request = {"model":model_id,"messages":msgs,"stream":True,
                   "max_tokens":ngen_for(a,interactive=True),
                   "stream_options":{"include_usage":True}}   # exact completion_tokens in the last chunk
        if getattr(a,"effort",None): request["reasoning_effort"]=a.effort
        if getattr(a,"think",None) is not None: request["enable_thinking"]=a.think
        body=json.dumps(request).encode()
        req=urllib.request.Request(base.rstrip("/")+"/v1/chat/completions", data=body,
                                   headers={"Content-Type":"application/json"})
        if a.api_key: req.add_header("Authorization", f"Bearer {a.api_key}")
        print(f"\n  {C.teal}◆ colibri{C.r}")
        sp=Spinner("thinking…"); sp.start()
        md=MDStream("  "); reply=[]; first=True; t0=time.time(); interrupted=False; think_open=False
        exact_tokens=None
        try:
            with urllib.request.urlopen(req) as r:
                for raw in r:
                    line=raw.decode("utf-8","replace").strip()
                    if not line.startswith("data: "): continue
                    data=line[6:]
                    if data=="[DONE]": break
                    try: ev=json.loads(data)
                    except ValueError: continue
                    usage=ev.get("usage")
                    if isinstance(usage,dict) and isinstance(usage.get("completion_tokens"),int):
                        exact_tokens=usage["completion_tokens"]
                    for ch in ev.get("choices",[]):
                        d=ch.get("delta",{})
                        rtxt=d.get("reasoning_content")
                        if rtxt and os.environ.get("COLI_SHOW_THINK","1")=="0":
                            continue              # thinking nascosto: lo spinner resta a girare
                        if rtxt:
                            # Thinking arrives as its own delta field, and this loop
                            # used to drop it on the floor: `coli run` streamed the
                            # model's reasoning while `coli chat` silently hid it
                            # (#979). Render it dim and unformatted — it is the
                            # model's scratchpad, not the answer — and keep it out
                            # of the saved reply so history carries content only.
                            # The keepalive ping reuses this field with "."/"" —
                            # dim dots read as heartbeat, which is what they are.
                            if first: sp.stop(); first=False
                            if not think_open:
                                think_open=True
                                sys.stdout.write(f"  {C.dgray}┌ thinking{C.r}\n  {C.dgray}│{C.r} ")
                            sys.stdout.write(C.dim+rtxt.replace("\n","\n  "+C.dgray+"│"+C.r+" "+C.dim)+C.r)
                            sys.stdout.flush(); continue
                        txt=d.get("content")
                        if not txt: continue          # ping/ruolo: non è testo
                        if think_open:
                            think_open=False
                            sys.stdout.write(f"{C.r}\n  {C.dgray}└─{C.r}\n")
                        if first: sp.stop(); first=False
                        md.feed(txt); reply.append(txt)
        except KeyboardInterrupt:
            interrupted=True                          # il server annulla la richiesta alla disconnessione
        except urllib.error.HTTPError as e:
            # HTTPError IS-A OSError: without this arm a 4xx/5xx fell into the
            # branch below and read as "server unreachable" — hiding the API's
            # own explanation of what was wrong with the request (#975: a V4
            # CONTEXT_EXCEEDED 400 took a screenshot to diagnose instead of a
            # sentence). The body is OpenAI-shaped; show its message.
            sp.stop()
            try:
                detail=json.loads(e.read().decode("utf-8","replace")).get("error",{}).get("message","")
            except Exception:
                detail=""
            print(f"\n  {C.yel}[the server rejected the request: HTTP {e.code}"
                  f"{' — '+detail if detail else ''}]{C.r}")
            msgs.pop(); continue                      # la richiesta non è mai partita: togli il turno e riprova
        except OSError as e:
            sp.stop()
            print(f"\n  {C.yel}[server unreachable: {e}]{C.r}"); break
        if think_open:                            # stream finito dentro il thinking (interrupt o
            think_open=False                      # risposta solo-reasoning): chiudi il riquadro
            sys.stdout.write(f"{C.r}\n  {C.dgray}\u2514\u2500{C.r}\n")
        md.close(); sp.stop()
        if reply: msgs.append({"role":"assistant","content":"".join(reply)})
        else: msgs.pop()                              # turno vuoto: non sporcare la history
        el=time.time()-t0
        footer=render_gen_stats(gen_stats("".join(reply), el, exact_tokens, interrupted), chat_stats_mode(a))
        if footer: print(f"\r  {C.dgray}└─ {footer}{C.r}\n")
        else: print()
    print(f"  {C.dim}goodbye — the engine keeps running for the next chat 🐦{C.r}")

def kv_resume_notice(model_dir):
    """SERVE mode silently resumes .coli_kv from disk (glm.c kv_disk_load): a chat
    started today continues a conversation from days ago, with `first=0` so the
    turn is appended WITHOUT the [gMASK]<sop> prefix. The engine does announce it
    on stderr — but nothing here ever shows that: the drain thread's
    p.stderr.read() blocks until EOF, so on a healthy start errlog is still empty
    when the status lines are printed. The warning only appeared once the engine
    DIED, which is exactly when it no longer mattered.

    Measured cost of the silence: a chat inherited 670 tokens of an old Italian
    session ("il mio numero preferito e 7, ricordalo!"). Every later reply came
    back in Italian, and "explain fibonacci in short" was answered about the
    number 7 — the model was being coherent with a context nobody could see, and
    it read as a quantization bug for a day.

    So say it here, in Python, from the file itself: no pipe, no thread, no
    Windows deadlock risk (see the stderr comment below)."""
    p=os.path.join(model_dir, ".coli_kv")
    try:
        with open(p,"rb") as f:
            if f.read(8)!=b"COLIKV1\0": return
            h=struct.unpack("<8i", f.read(32))
        n=h[6]
        if n<1: return
        age=time.time()-os.path.getmtime(p)
        when=f"{age/86400:.0f}d ago" if age>86400 else f"{age/3600:.0f}h ago" if age>3600 else "just now"
        print(f"  {C.yel}↺ resuming a saved conversation: {n} tokens, last written {when}{C.r}")
        print(f"  {C.dgray}  it steers tone, language and topic. :reset clears it · "
              f"KVSAVE=0 disables saving · delete {p} to start clean{C.r}")
    except (OSError, struct.error): pass

IMAGE_SUFFIXES = (".png",".jpg",".jpeg",".webp",".bmp",".gif",".tif",".tiff")
# Un percorso che finisce in un'estensione di immagine, nelle due forme che
# capita di incollare: Windows (C:\Users\...) e POSIX (/home/...  ~/foto.png).
#
# Il confine e' l'ESTENSIONE, non lo spazio: le cartelle di Windows si chiamano
# "Nuova cartella" e spezzare sugli spazi le taglierebbe a meta'. E non si passa
# da shlex, che in modalita' POSIX mangia le barre rovesciate e va in errore su
# un apostrofo -- cioe' su una frase italiana su due.
_ESTENSIONI = "|".join(suffix[1:] for suffix in IMAGE_SUFFIXES)
IMAGE_PATH = re.compile(
    r"(?:[A-Za-z]:\\|~?/|\./)[^\"'<>|?*\n]*?\.(?:" + _ESTENSIONI + r")\b",
    re.IGNORECASE)


def to_local_path(token):
    """Il percorso come lo vede questa macchina.

    Un percorso Windows incollato da Esplora risorse dentro WSL indica un file
    che da qui si chiama /mnt/<lettera>/...: senza tradurlo non esiste."""
    token = token.strip().strip('"').strip("'")
    if re.fullmatch(r"[A-Za-z]:\\.*", token, re.S):
        if sys.platform == "win32":
            return token
        return f"/mnt/{token[0].lower()}/" + token[2:].replace("\\", "/").lstrip("/")
    return os.path.expanduser(token)


def message_with_images(text):
    """Un percorso incollato diventa un'immagine, il resto resta testo.

    Chi scrive in chat non ha voglia di comporre JSON: incolla il percorso, o
    ce lo trascina sopra e il terminale glielo mette fra virgolette.

    Restituisce None quando qualcosa SEMBRA un'immagine e il file non c'e',
    dopo averlo detto. Il silenzio qui costa caro: si aspetta la risposta per
    minuti e solo alla fine si scopre che il modello non ha ricevuto nulla."""
    if not text.strip():
        return text

    found, rest = [], text
    for match in IMAGE_PATH.finditer(text):
        raw = match.group(0)
        candidate = to_local_path(raw)
        if not os.path.isfile(candidate):
            print(f"  {C.yel}✗ non trovo{C.r} {raw}"
                  + (f"\n    da qui si chiamerebbe {candidate}" if candidate != raw else ""))
            return None
        found.append(candidate)
        rest = rest.replace(raw, " ", 1)

    if not found:
        return text
    if len(found) > 1:
        # Il motore ne tiene una in sospeso per volta. Dirlo adesso costa un
        # secondo; scoprirlo dal server dopo il prefill ne costa centinaia.
        print(f"  {C.yel}✗ una immagine per messaggio{C.r}: ne hai indicate "
              f"{len(found)}")
        return None
    content = []
    rest = " ".join(rest.split()).strip()
    if rest:
        content.append({"type":"text","text":rest})
    for path in found:
        # The file is read HERE, with the user's own rights, and travels as a
        # data: URI. The server no longer opens local paths for a client (an
        # inference client is not the operator; with the API key it could read
        # any file the server process can), unless the operator sets
        # COLI_IMAGE_ROOT, and a path sent by mistake would be refused.
        content.append({"type":"image_url","image_url":{"url":image_data_uri(path)}})
    return content

def image_data_uri(path):
    """base64 data: URI of a local image, MIME from the extension (PNG when unknown)."""
    import base64, mimetypes
    mime = mimetypes.guess_type(path)[0] or "image/png"
    with open(path, "rb") as handle:
        return f"data:{mime};base64," + base64.b64encode(handle.read()).decode("ascii")


def cmd_chat(a):
    # ATTACH: a running `coli serve` beats a private engine every time — the load
    # (34-136 s) and the cache warmth survive between sessions. Explicit --attach
    # wins; otherwise probe localhost quietly and use it if it's there. --no-attach
    # forces the old behaviour. The probe costs ~1 ms when nothing is listening.
    if not getattr(a,"no_attach",False):
        base=getattr(a,"attach",None) or "http://127.0.0.1:8000"
        mid=server_probe(base, getattr(a,"api_key",None))
        if mid:
            banner(f"chat · {mid} · attached")
            chat_attached(a, base, mid); return
        if getattr(a,"attach",None):
            sys.exit(f"--attach: no coli serve answering at {base} (start one with: coli serve --model <dir>)")
    need_model(a.model)
    family=resolve_model(a.model).descriptor
    arch=family.id
    if not family.has_gateway_adapter:
        sys.exit(f"{family.display_name}: gateway adapter is not wired")
    if arch!="glm":
        engine=engine_for(a.model)
        need_model(a.model,engine)
        model_id=family.default_model_id
        banner(f"chat · {model_id} · local server", model=a.model)
        import openai_server
        # --cap rides along so an explicit `coli chat --cap N` reaches the
        # engine (it was silently eaten for years -- disclosed behavior
        # change); absent stays absent, and openai_server translates the
        # missing flag per model arch (cap_for_arch, #379: non-glm gets the
        # legacy 8, an explicit value -- 0 included -- passes verbatim).
        cmd=[sys.executable,openai_server.__file__,"--model",a.model,
             "--engine",engine,"--arch",arch,"--model-id",model_id,
             "--host","127.0.0.1","--port","8000","--max-tokens",
             str(ngen_for(a,interactive=True,family=family))]
        if a.cap is not None: cmd+=["--cap",str(a.cap)]
        if a.api_key: cmd+=["--api-key",a.api_key]
        p=subprocess.Popen(cmd,env=env_for_engine(a,arch))
        try:
            sp=Spinner(f"loading {model_id}…"); sp.start()
            for _ in range(1800):
                if p.poll() is not None:
                    sp.stop(); sys.exit(f"{model_id} server exited while loading")
                if server_probe("http://127.0.0.1:8000",a.api_key,timeout=1.0):
                    sp.stop()
                    chat_attached(a,"http://127.0.0.1:8000",model_id)
                    return
                time.sleep(1)
            sp.stop(); sys.exit(f"timed out loading {model_id}")
        finally:
            if p.poll() is None:
                p.terminate()
                try: p.wait(timeout=10)
                except subprocess.TimeoutExpired: p.kill()
    need_model(a.model)
    banner(f"chat · {os.path.basename(a.model)} · ram {a.ram or '-'}GB · topp {a.topp or 'off'}", model=a.model)
    kv_resume_notice(a.model)
    # UTF-8, not the locale's code page: the engine writes "·" and "—", which
    # cp932/cp949 and the C locale cannot encode. The drain thread below
    # caught that UnicodeEncodeError (a ValueError) and stopped, and once
    # nobody read the pipe the engine blocked on its next [prefill] line.
    errlog=tempfile.NamedTemporaryFile(mode="w+", suffix=".log", delete=False, encoding="utf-8")
    e=env_for(a); e["SERVE"]="1"
    # stderr -> PIPE, NOT stderr=errlog (file). On Windows/MinGW, pointing the
    # child's stderr at a file/DEVNULL handle stalls the CRT so stdout (the byte
    # protocol coli reads one byte at a time) never flushes and chat hangs at
    # ~10 GB resident. A PIPE whose read end nobody drains still works: the
    # engine emits only ~400 bytes of status to stderr, which fits comfortably
    # in the OS pipe buffer, so it never blocks. We snapshot stderr into errlog
    # once the READY sentinel arrives, so the status-line display below works
    # exactly as before. (Do NOT add a concurrent stderr drain thread: on
    # Windows, reading two child pipes simultaneously deadlocks CPython's IO.)
    p=subprocess.Popen([GLM,str(cap_for_launch(a.cap,e,0))], env=e, stdin=subprocess.PIPE,
                       stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0)
    sp=Spinner("waking the giant (744B)…"); sp.start()
    # Keep the pre-READY stdout preamble instead of discarding it: the engine's
    # "loaded in Xs | resident dense: Y MB" line is a printf, i.e. STDOUT
    # (colibri.c:9595 — same in inkling.c/olmoe.c), so it arrives HERE and never
    # in errlog. The ready-line regex below used to scan stderr for it and could
    # never match. A few hundred bytes of banner; nothing else reads them.
    _preamble=bytearray()
    st=stream_turn(p, READY, _preamble.extend)
    sp.stop()
    if st is None:
        try: errlog.write(p.stderr.read().decode("utf-8","replace"))
        except (OSError, ValueError): pass
        errlog.flush(); errlog.seek(0); print(errlog.read()[-1500:])
        engine_diag(p)        # perche' e' morto (OOM-kill compreso); errlog e' gia' stampato sopra
        sys.exit("the engine exited while loading")
    p.stdout.readline()   # TIERS line (web-dashboard protocol): emitted once right after STAT;
                          # left unread it leaks into the first answer's text
    # READY received. Drain the child's stderr into errlog without blocking:
    # the engine is still alive (blocked on stdin), so a plain read() would
    # hang forever waiting for EOF. A short bounded drain grabs the ~400 bytes
    # of load-time status ([RAM_GB], [MTP], ...) that were already emitted.
    _drain_box={"done":False}
    def _drain():
        # readline(), NOT read(): read() returns only at EOF, and the engine is alive
        # on stdin, so it never returned and errlog.write() below it never ran — the
        # bounded wait always timed out on an EMPTY file and the whole status block
        # (including "ready in Xs") silently displayed nothing. Per-line write+flush
        # makes each status line visible to the reader below as soon as it arrives.
        try:
            while True:
                raw=p.stderr.readline()
                if not raw: break
                errlog.write(raw.decode("utf-8","replace")); errlog.flush()
        except (OSError, ValueError): pass
        _drain_box["done"]=True
    threading.Thread(target=_drain, daemon=True).start()
    _drain_box["th"]=threading.current_thread()
    for _ in range(20):           # up to ~1s for the load-status lines
        if _drain_box["done"]: break
        time.sleep(0.05)
    errlog.flush()
    mload=re.search(r"loaded in ([0-9.]+)s \| resident dense: ([0-9.]+) MB",
                    _preamble.decode("utf-8","replace"))
    if mload: print(f"  {C.grn}✓{C.r} ready in {mload.group(1)}s {C.dim}· resident {float(mload.group(2))/1000:.1f} GB · RSS {st.get('rss','?')} GB{C.r}")
    try:
        elog=open(errlog.name, encoding="utf-8", errors="replace").read()
        for l in elog.splitlines():                     # una riga di stato per riga, senza path
            # [METAL]/[CUDA] first: the splash tagline (model_banner_line) names the
            # checkpoint, not the backend — this engine line is the only confirmation
            # the GPU tier actually engaged (it can still fall back per-block). The
            # engine prints it to stderr, we drain it into errlog; keep it whitelisted.
            if l.startswith(("[METAL]","[CUDA]","[RAM_GB","[PIN]","[MTP]","[USAGE]","[DSA]","[KV]")):
                l=re.sub(r" ?\(?/[^ )]+\)?","",l.strip())       # via i percorsi lunghi
                l=re.sub(r" from$","",l)
                for chunk in textwrap.wrap(l, term_w()-4) or [l]:
                    print(f"  {C.dgray}{chunk}{C.r}")
    except Exception: pass
    print(f"  {C.dim}type and press Enter · Ctrl-C stops the answer · :more continues · :reset clears memory · :q exits{C.r}\n")
    w=term_w()-4
    try:
        while True:
            if TTY:
                print(f"  {C.dgray}╭{'─'*w}╮{C.r}")
                try: msg=read_prompt(f"  {C.dgray}│{C.r} {C.teal}{C.b}›{C.r} ")
                except EOFError: print(); break
                try: msg=redraw_prompt_box(msg, w)
                except Exception: print(f"  {C.dgray}╰{'─'*w}╯{C.r}")
            else:
                try: msg=input()
                except EOFError: break
            msg=msg.strip("\r\n")
            if msg in (":q",":quit","exit"): break
            if not msg.strip(): continue
            if msg==":reset":
                p.stdin.write(b"\x02RESET\n"); p.stdin.flush()
                stream_turn(p, END, lambda b: None)
                print(f"  {C.dim}✦ memory cleared{C.r}\n"); continue
            if msg in (":piu",":più",":more",":continua"):
                p.stdin.write(b"\x02MORE\n"); p.stdin.flush()
            else:
                p.stdin.write((msg.replace("\n"," ")+"\n").encode()); p.stdin.flush()
            print(f"\n  {C.teal}◆ colibri{C.r}")
            dec=codecs.getincrementaldecoder("utf-8")("replace")
            state={"first":True}
            def prefill_tick(path=errlog.name):
                try:
                    with open(path, encoding="utf-8", errors="replace") as f:
                        f.seek(max(0, os.path.getsize(path)-1500)); tail=f.read()
                    pl=[l for l in tail.splitlines() if l.startswith("[prefill]")]
                    return pl[-1].replace("[prefill] ","prefill ") if pl else ""
                except Exception: return ""
            sp2=Spinner("thinking…", tick=prefill_tick); sp2.start()
            md=MDStream("  ")            # markdown -> terminale, in streaming
            raw=os.environ.get("COLI_RAW")=="1"
            def echo(bs, _dec=dec, _st=state):
                if _st["first"]:
                    sp2.stop(); _st["first"]=False
                    if raw: sys.stdout.write("  ")
                s=_dec.decode(bs)
                if not s: return
                if raw: sys.stdout.write(s.replace("\n","\n  ")); sys.stdout.flush()
                else: md.feed(s)
            t0=time.time()
            st=stream_turn(p, END, echo)
            if not raw: md.close()
            sp2.stop()
            if st is None: engine_diag(p, errlog); break
            el=time.time()-t0
            if st.get("tok") and chat_stats_mode(a) == "off":
                if st.get("interrupted"): print(f"  {C.yel}⏹ interrupted; type :more to continue the response{C.r}")
                print()
            elif st.get("tok"):
                print(f"\r  {C.dgray}└─ {st['tok']} tok · {st['tps']:.2f} tok/s · hit {st['hit']:.0f}% · RSS {st['rss']:.1f} GB · {el:.0f}s{C.r}")
                if st.get("interrupted"):
                    print(f"  {C.yel}⏹ interrupted; type :more to continue the response{C.r}")
                elif st["tok"]>=ngen_for(a,interactive=True):
                    print(f"  {C.yel}…stopped at --ngen ({ngen_for(a,interactive=True)}); type :more to continue the response{C.r}")
                print()
            else:
                if st.get("interrupted"): print(f"  {C.yel}⏹ interrupted{C.r}")
                print()
    except KeyboardInterrupt:
        print(f"\n  {C.dim}interrupted{C.r}")
    finally:
        try: p.stdin.close(); p.terminate()
        except Exception: pass
        try: os.unlink(errlog.name)
        except Exception: pass
    print(f"  {C.teal}goodbye{C.r} {C.dim}— the hummingbird returns to its nest{C.r} 🐦\n")

def serve_pidfile(port): return os.path.join(tempfile.gettempdir(), f"coli-serve-{port}.pid")

def _serve_cmdline_matches_port(raw, port):
    argv=[part.decode("utf-8","replace") for part in raw.split(b"\0") if part]
    try: coli_i=next(i for i,arg in enumerate(argv) if os.path.basename(arg)=="coli")
    except StopIteration: return False
    if coli_i+1>=len(argv) or argv[coli_i+1]!="serve":
        return False
    configured=8000
    for i,arg in enumerate(argv[coli_i+2:], start=coli_i+2):
        if arg=="--port" and i+1<len(argv):
            try: configured=int(argv[i+1])
            except ValueError: return False
        elif arg.startswith("--port="):
            try: configured=int(arg.split("=",1)[1])
            except ValueError: return False
    return configured==port

def _serve_environ_matches_port(raw, port):
    env={}
    for entry in raw.split(b"\0"):
        if b"=" in entry:
            key,value=entry.split(b"=",1); env[key]=value
    return env.get(b"SERVE")==b"1" and env.get(b"COLI_SERVE_PORT")==str(port).encode()

def _serve_engine_env(a, arch):
    env=env_for_engine(a,arch)
    env["COLI_SERVE_PORT"]=str(a.port)
    return env

def cmd_serve(a):
    need_model(a.model)
    family=resolve_model(a.model).descriptor
    arch=family.id
    if not family.has_gateway_adapter:
        sys.exit(f"{family.display_name}: gateway adapter is not wired")
    engine=engine_for(a.model)
    need_model(a.model,engine)
    if a.kv_slots>family.limits.max_kv_slots:
        sys.exit(f"{arch} currently supports at most {family.limits.max_kv_slots} KV slot(s)")
    # pidfile: cosi' `coli stop` spegne tutto con un comando, senza pkill a mano.
    # EN: pidfile so `coli stop` can shut everything down without manual pkill.
    try:
        with open(serve_pidfile(a.port),"w") as f: f.write(f"{os.getpid()} {a.model}\n")
    except OSError: pass
    import openai_server
    openai_server.ARCH=arch
    model_id=a.model_id or family.default_model_id
    try:
        if a.temp is not None: os.environ["COLI_TEMP"] = str(a.temp)
        env=_serve_engine_env(a,arch)
        if a.cluster_coordinator and not a.cluster_workers:
            from cluster import discover_workers
            try:
                workers=discover_workers(a.cluster_coordinator)
            except OSError as error:
                sys.exit(f"{C.yel}cannot discover cluster workers:{C.r} {error}")
            if not workers:
                sys.exit(f"{C.yel}cluster coordinator has no live expert workers{C.r}")
            env["CLUSTER_WORKERS"] = ",".join(workers)
            print(f"  {C.dim}[CLUSTER] discovered {len(workers)} expert worker(s){C.r}", file=sys.stderr)
        openai_server.serve(a.model,a.host,a.port,model_id,a.api_key,
              a.cap,ngen_for(a,interactive=True,family=family),engine,
              env,a.cors_origin,
              a.max_queue,a.queue_timeout,a.kv_slots,allowed_hosts=a.allowed_host,
              family=family)
    finally:
        try: os.unlink(serve_pidfile(a.port))
        except OSError: pass

def _pid_alive(pid):
    """Is `pid` a live process?

    `os.kill(pid, 0)` is the POSIX idiom, but on Windows os.kill() has no signal
    semantics: for a pid this process did not spawn it raises
    OSError [WinError 87] "The parameter is incorrect" instead of reporting
    liveness. cmd_stop caught that as "not running", so `coli stop` printed
    "nothing running" while the port was still LISTENING and ~5 GB of engines
    were resident (#1049, reported with a reproduction on Windows 11 native).
    On win32 ask the OS directly instead."""
    if sys.platform != "win32":
        try: os.kill(pid, 0); return True
        except OSError: return False
    import ctypes
    from ctypes import wintypes
    PROCESS_QUERY_LIMITED_INFORMATION, STILL_ACTIVE = 0x1000, 259
    k32 = ctypes.WinDLL("kernel32", use_last_error=True)
    k32.OpenProcess.restype = wintypes.HANDLE
    handle = k32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
    if not handle: return False
    try:
        code = wintypes.DWORD()
        if not k32.GetExitCodeProcess(handle, ctypes.byref(code)): return False
        # STILL_ACTIVE is also a legal exit code; a process that exited with 259
        # reads as alive here. SIGTERM on a dead pid is harmless, so the failure
        # mode is a redundant kill, never a missed one.
        return code.value == STILL_ACTIVE
    finally:
        k32.CloseHandle(handle)

def cmd_cluster_coordinator(a):
    from cluster import serve
    serve(a.host, a.port, a.stale_after, a.allowed_host)

def cmd_cluster_worker(a):
    engine=need_worker_model(a.model)
    coordinator=a.coordinator or os.environ.get("CLUSTER_COORDINATOR")
    host=a.advertise_host or os.environ.get("CLUSTER_ADVERTISE_HOST", "127.0.0.1")
    node_id=a.node_id or f"{host}:{a.port}"
    stop_heartbeat=threading.Event()
    if coordinator:
        from cluster import heartbeat, register
        register(coordinator, {"node_id":node_id, "host":host, "port":a.port,
                               "role":"expert", "layers":a.layers})
        def keep_registered():
            while not stop_heartbeat.wait(10):
                try: heartbeat(coordinator, node_id)
                except OSError: pass
        threading.Thread(target=keep_registered, name="colibri-cluster-heartbeat", daemon=True).start()
    e=env_for(a)
    e.update({"EXPERT_WORKER":"1", "CLUSTER_WORKER_PORT":str(a.port),
              "COLI_MMAP":os.environ.get("COLI_MMAP", "1")})
    print(f"  {C.dim}[CLUSTER] expert worker · {host}:{a.port} · layers {a.layers}{C.r}")
    try:
        # cap_for_launch, not a.cap: --cap defaults to None and str(None) is the
        # literal "None", which the engine's argument parser refuses with
        # `cache/layer: expected a whole number, got "None"` -- so the worker
        # could only ever start with an explicit --cap (#1452). Fallback 0 is
        # what the other GLM launch sites pass: the engine resolves its own cap
        # when the argument is zero.
        return subprocess.call([engine,str(cap_for_launch(a.cap,e,0)),
                                str(a.ebits),str(a.dbits)],env=e)
    finally:
        stop_heartbeat.set()

def _stop_proc_table(port):
    """(pid, cmdline, comm) for every process, in /proc's NUL-separated byte format.

    macOS and the BSDs have no /proc, so cmd_stop's scan used to iterate an empty list and
    report "nothing running" while a live serve and its multi-GB engine kept the box. That is
    worse than a crash: this command exists to stop the ghost engines that once OOM'd a
    machine, and a silent success invites the operator to walk away.

    `ps` is asked for everything ONCE rather than per pid, and its space-separated output is
    normalised to NUL-separated bytes so _serve_cmdline_matches_port keeps working unchanged."""
    try:
        pids=[int(d) for d in os.listdir("/proc") if d.isdigit()]
    except OSError:
        pids=None
    if pids is not None:
        table=[]
        for pid in pids:
            try:
                with open(f"/proc/{pid}/cmdline","rb") as f: cmd=f.read()
                with open(f"/proc/{pid}/comm") as f: comm=f.read().strip()
            except (OSError,PermissionError): continue
            table.append((pid,cmd,comm))
        return table
    try:
        out=subprocess.run(["ps","-Ao","pid=,comm=,command="],capture_output=True,text=True,
                           timeout=10).stdout
    except (OSError,subprocess.SubprocessError):
        return []
    table=[]
    for line in out.splitlines():
        parts=line.strip().split(None,2)
        if len(parts)<3 or not parts[0].isdigit(): continue
        pid=int(parts[0]); comm=os.path.basename(parts[1])
        try: argv=shlex.split(parts[2])
        except ValueError: argv=parts[2].split()
        table.append((pid,b"\0".join(a.encode("utf-8","replace") for a in argv)+b"\0",comm))
    return table

def _stop_environ(pid):
    """A process's environment as NUL-separated KEY=VAL bytes, /proc or `ps eww`."""
    try:
        with open(f"/proc/{pid}/environ","rb") as f: return f.read()
    except OSError:
        pass
    try:
        out=subprocess.run(["ps","eww","-o","command=","-p",str(pid)],capture_output=True,
                           text=True,timeout=5).stdout
    except (OSError,subprocess.SubprocessError):
        return b""
    # `ps eww` appends the environment after the command line; keep only KEY=VAL tokens.
    pairs=[t for t in out.split() if "=" in t and t.split("=",1)[0].replace("_","").isalnum()]
    return b"\0".join(t.encode("utf-8","replace") for t in pairs)+b"\0"

def cmd_stop(a):
    """Shut down a running `coli serve` AND its engine — one command, no pkill.
    The engine re-execs itself for OMP tuning, so its process is named `exe`,
    not `glm`: every `pkill -x glm` in history silently killed nothing (that is
    how two 17+5 GB ghost engines OOM'd this box on 2026-07-16). This finds the
    real processes: the pidfile first, then /proc by cmdline/environ — only
    the requested port's tagged engine and matching `coli serve` wrapper."""
    banner("stop")
    targets={}  # pid -> descrizione
    pf=serve_pidfile(a.port)
    try:
        with open(pf) as f: pid=int(f.read().split()[0])
        if _pid_alive(pid): targets[pid]=f"coli serve (pidfile, port {a.port})"
    except (OSError,ValueError,IndexError): pass
    process_names={"exe"}
    for family in all_families(): process_names.update(family.process_names)
    for pid,cmd,comm in _stop_proc_table(a.port):
        try:
            if pid!=os.getpid() and _serve_cmdline_matches_port(cmd,a.port):
                targets.setdefault(pid,f"coli serve (cmdline, port {a.port})")
            if comm in process_names and _serve_environ_matches_port(_stop_environ(pid),a.port):
                targets.setdefault(pid,f"engine `{comm}` (port {a.port})")
        except (OSError,PermissionError): continue
    if not targets:
        print(f"  nothing running — no serve on port {a.port}, no SERVE engines"); return
    for pid,desc in targets.items(): print(f"  {'would stop' if a.dry_run else 'stopping'} {pid}: {desc}")
    if a.dry_run: return
    for pid in targets:
        try: os.kill(pid, signal.SIGTERM)
        except OSError: pass
    time.sleep(2.0)
    for pid in targets:
        # signal.SIGKILL does not exist on win32 and AttributeError is not OSError,
        # so the bare `except OSError` let it escape and abort the command AFTER
        # the SIGTERMs above (#1049). Line 752 in this file already uses the
        # getattr form; os.kill(SIGTERM) is TerminateProcess on Windows anyway.
        try: os.kill(pid, getattr(signal, "SIGKILL", signal.SIGTERM)); print(f"  {pid}: forced")
        except OSError: pass       # gia' morto: bene
    try: os.unlink(pf)
    except OSError: pass
    print(f"  {C.grn}✓ stopped{C.r} — RAM released")

def cmd_web(a):
    """serve + open the dashboard in the browser once the API answers."""
    need_model(a.model)
    # Same two layouts openai_server.py resolves: next to this script (release
    # archive / installed tree) or one level up (source checkout, where this is c/coli).
    cands = [os.path.join(HERE, "web", "dist"),
             os.path.join(os.path.dirname(os.path.abspath(HERE)), "web", "dist")]
    dist = next((d for d in cands if os.path.exists(os.path.join(d, "index.html"))), cands[-1])
    if not os.path.exists(os.path.join(dist, "index.html")):
        print(f"{C.yel}web UI not built:{C.r} run  cd web && npm install && npm run build  first;")
        print("serving the API anyway (the dashboard will 404 until built).")
    url = f"http://{a.host}:{a.port}/"
    if not getattr(a, "no_browser", False):
        import threading, urllib.request, webbrowser
        def opener():
            # No iteration cap. It used to be `for _ in range(600)` with a 2 s sleep --
            # a 20-minute budget, sized when "the engine takes minutes to load" meant
            # GLM. #854 reported Kimi K3 never opening the browser on a 256 GB Windows
            # box, and the log explains itself: `[K3] init done in 2540.5s`. The thread
            # gave up 22 minutes before the server first answered /health, so the
            # message promising it opens automatically was simply false.
            #
            # A ceiling here can only ever be wrong in one direction: too low fails
            # silently on a slow load, too high costs a sleeping daemon thread. The
            # thread is a daemon, so it cannot outlive the server it is waiting for --
            # there is nothing for the cap to protect against.
            waited = 0
            while True:
                time.sleep(2); waited += 2
                try:
                    urllib.request.urlopen(f"http://{a.host}:{a.port}/health", timeout=5)
                except OSError:
                    # Say something once, rather than looking hung. A 93-layer K3 load
                    # is 40+ minutes and the user has no way to tell waiting from stuck.
                    if waited == 300:
                        print(f"still loading; the dashboard will open at {url} when the "
                              f"engine is ready (Ctrl-C to stop, or --no-browser to skip)")
                    continue
                webbrowser.open(url); return
        threading.Thread(target=opener, daemon=True).start()
    print(f"dashboard: {url}  (opens automatically when the engine is ready)")
    cmd_serve(a)

def project_python():
    if sys.platform == "win32":
        candidate = os.path.join(HERE, "mio_env", "Scripts", "python.exe")
    else:
        candidate = os.path.join(HERE, "mio_env", "bin", "python3")
    return candidate if os.path.exists(candidate) else sys.executable


def cmd_bench(a):
    need_model(a.model)
    banner("bench", model=a.model)
    # tools/eval_glm.py drives the GLM engine through a scoring protocol no
    # sibling engine speaks; handed any other model it started the GLM engine
    # on it, which died on "this engine requires n_group=1" (#898's shape).
    family=resolve_model(a.model).descriptor
    if family.id!="glm":
        sys.exit(f"{C.yel}coli bench is not wired for {family.display_name};{C.r} "
                 f"its harness scores the GLM-5.2/5.3 engine only")
    # python con `tokenizers`: l'ambiente del progetto se c'e', altrimenti quello corrente
    py = project_python()
    tasks = ",".join(a.tasks) if a.tasks else "hellaswag,arc_challenge,mmlu"
    # dataset mancanti -> li scarica una volta (fetch_benchmarks.py li mette in --data come JSONL)
    missing=[t for t in tasks.split(",") if not os.path.exists(os.path.join(a.data,f"{t}.jsonl"))]
    if missing:
        print(f"  {C.dim}downloading missing datasets: {', '.join(missing)}{C.r}")
        subprocess.call([py, os.path.join(TOOLS,"fetch_benchmarks.py"),
                         "--out", a.data, "--tasks", ",".join(missing), "--limit", str(max(a.limit,200))])
        # il fetch riprova da solo (#304), ma se l'hub resta giu' il bench gira sui task
        # disponibili invece di passare a eval file inesistenti.
        # EN: the fetch retries on its own (#304), but if the hub stays down the bench
        # runs on the available tasks instead of handing eval nonexistent files.
        still=[t for t in tasks.split(",") if not os.path.exists(os.path.join(a.data,f"{t}.jsonl"))]
        if still:
            tasks=",".join(t for t in tasks.split(",") if t not in still)
            print(f"  {C.yel}skipping (download failed, rerun later): {', '.join(still)}{C.r}")
            if not tasks:
                print(f"  {C.yel}no datasets available — nothing to bench{C.r}"); sys.exit(1)
    cmd=[py, os.path.join(TOOLS,"eval_glm.py"), "--glm", GLM, "--snap",a.model,
         "--tasks", tasks, "--limit", str(a.limit), "--data", a.data]
    if a.ram: cmd+=["--ram",str(a.ram)]
    e=env_for(a)
    print(f"  {C.dim}decode is disk-bound: this takes HOURS on slow hardware. Raise --limit on faster machines.{C.r}\n")
    sys.exit(subprocess.call(cmd, env=e))

#: Le opzioni di precisione di `coli convert`: nome sulla riga di comando e
#: default. Il parser le dichiara con default=None apposta, cosi' "l'utente
#: l'ha scritta" e' semplicemente "non e' None" -- vale per --ebits 3,
#: --ebits=3 e per le abbreviazioni che argparse accetta, senza frugare in
#: sys.argv. Un default non passato al convertitore non fa danno; un valore
#: scritto e ignorato si'.
_CONVERT_PRECISION_FLAGS = {"ebits": "--ebits", "io_bits": "--io-bits",
                            "xbits": "--xbits", "group_size": "--group-size"}
_CONVERT_DEFAULTS = {"ebits": 4, "io_bits": 8, "xbits": 0, "group_size": 64}


def checkpoint_family(repo):
    """La famiglia del checkpoint, letta dal solo config.json.

    Cinque KB decidono quale convertitore ha senso per i prossimi 300 GB, ed e'
    lo stesso appiglio che usa la guardia dentro convert_fp8_to_int4.py.

    None quando non si riesce a stabilirlo -- rete assente, huggingface_hub non
    installato, repo senza config.json. In quel caso si prosegue come prima: la
    guardia del convertitore scarica lo stesso file e rifiuta lei. Meglio
    proseguire che fermare una conversione valida per un errore di rete.
    """
    try:
        local = os.path.join(os.path.expanduser(repo), "config.json")
        if os.path.isfile(local):
            with open(local, encoding="utf-8") as handle:
                return family_for_config(json.load(handle))
        from huggingface_hub import hf_hub_download
        with open(hf_hub_download(repo, "config.json"), encoding="utf-8") as handle:
            return family_for_config(json.load(handle))
    except SystemExit:
        raise
    except Exception:
        return None


CONVERT_DEFAULT_REPO = "zai-org/GLM-5.2-FP8"

def convert_output_refusal(out_dir):
    """Why `coli convert --model out_dir` must not run, or None. A checkpoint
    already living there is the tell: --model names the OUTPUT directory,
    and a family that runs its official checkpoint as is needs no conversion
    at all, which is the usual reason someone points convert at one."""
    if not os.path.isdir(out_dir):
        return None
    try:
        shards = [x for x in os.listdir(out_dir) if x.endswith(".safetensors")]
    except OSError:
        return None
    has_config = os.path.isfile(os.path.join(out_dir, "config.json"))
    if not shards and not has_config:
        return None
    what = f"{len(shards)} shard(s)" if shards else "a config.json"
    family = None
    if has_config:
        try:
            family = resolve_model(out_dir).descriptor
        except (FamilyConfigError, UnknownFamilyError):
            family = None
    head = (f"{C.yel}{out_dir} already holds a checkpoint ({what}"
            f"{', ' + family.display_name if family else ''}).{C.r}\n"
            f"  --model is the OUTPUT directory of coli convert; the source is --repo <hf repo>.\n"
            f"  Nothing was written.")
    if family and not family.converter:
        return (head + f"\n  {family.display_name} runs its official checkpoint as it is: no conversion needed.\n"
                       f"  Point the engine at it: coli chat --model {out_dir}")
    if family and family.converter:
        return (head + f"\n  To convert shards already on disk, run the family's converter on them:\n"
                       f"      python3 tools/{family.converter} --indir {out_dir} --outdir <new dir>")
    return head + "\n  To convert into a fresh directory: coli convert --repo <hf repo> --model <new dir>"

def cmd_convert(a):
    banner("convert")
    # here --model is the DESTINATION the converted weights are written to, not an
    # existing snapshot, so it needs its own wording rather than need_model's.
    if not a.model:
        sys.exit(f"{C.yel}no output directory given.{C.r}\n"
                 f"  --model <dir> is where the converted model is written")
    # A support case ran `coli convert --model <dir of a downloaded Qwen3.8>`,
    # read --model as "the model to convert", and watched coli start fetching
    # the default repo (GLM-5.2) INTO that directory. --model is the output:
    # refuse to write into a directory that already holds a checkpoint, and
    # say what the user most likely wanted instead.
    refusal = convert_output_refusal(a.model)
    if refusal:
        sys.exit(refusal)
    if a.repo is None:
        a.repo = CONVERT_DEFAULT_REPO
    # python con torch/safetensors: l'ambiente del progetto se c'e', altrimenti quello corrente
    py = project_python()

    # #1368: quale convertitore, deciso dal checkpoint e non dal fatto che GLM-5.2
    # e' arrivato per primo. Prima di questo, una GLM-5.3-Flash passava per il
    # convertitore di GLM-5.2, che le quantizzava l'embedding perche' il suo nome
    # annidato non corrisponde a nessuna delle sue regole; il contenitore usciva
    # senza un errore e moriva ore dopo, dentro `coli web`.
    family = checkpoint_family(a.repo)
    script = family.converter if family else "convert_fp8_to_int4.py"
    if family and not script:
        # Famiglia nota che coli non guida: il suo convertitore ha un'altra riga
        # di comando, o non serve. La guardia dentro convert_fp8_to_int4.py tiene
        # le indicazioni per famiglia, sotto test; le si lascia dire a lei
        # invece di tenerne una seconda copia qui.
        script = "convert_fp8_to_int4.py"
    written_names = [name for name in _CONVERT_PRECISION_FLAGS
                     if getattr(a, name) is not None]
    for name, value in _CONVERT_DEFAULTS.items():
        if getattr(a, name) is None:
            setattr(a, name, value)
    if family:
        print(f"  {C.dim}checkpoint: {family.display_name} -> tools/{script}{C.r}")
        written = [_CONVERT_PRECISION_FLAGS[name] for name in written_names
                   if name not in family.converter_accepts]
        if written and family.converter:
            sys.exit(f"{C.yel}{' '.join(written)}: tools/{script} does not take "
                     f"{'them' if len(written) > 1 else 'it'}.{C.r}\n"
                     f"  {family.display_name} keeps its dense weights and embedding "
                     f"wide and picks the precision when the engine loads them.\n"
                     f"  Re-run without {' '.join(written)}.")

    # Quando la famiglia non si e' potuta stabilire si passano tutte le opzioni,
    # cioe' esattamente il comando di prima: nessuna conversione che funzionava
    # cambia comportamento perche' la rete non ha risposto.
    accepts = family.converter_accepts if family and family.converter else \
        tuple(_CONVERT_PRECISION_FLAGS)
    base = [py, os.path.join(TOOLS, script), "--repo", a.repo, "--outdir", a.model]
    if "ebits" in accepts:
        base += ["--ebits", str(a.ebits), "--io-bits", str(a.io_bits)]
    if "group_size" in accepts:
        base += ["--group-size", str(a.group_size)]
    if a.xbits and "xbits" in accepts:
        base += ["--xbits", str(a.xbits)]

    mtp_pass = family.converter_mtp_pass if family and family.converter else True
    steps = 2 if (mtp_pass and not a.no_mtp) else 1
    print(f"  {C.dim}[1/{steps}] model: {' '.join(base)}{C.r}")
    rc = subprocess.call(base)
    if rc != 0: sys.exit(rc)
    if steps == 1:
        # Non tutte le famiglie hanno una testa MTP da convertire a parte, e
        # convert_glm53.py non ha proprio --mtp: chiamarlo con quel flag e' un
        # errore di argomenti dopo che il modello e' gia' stato scritto.
        sys.exit(0)
    # passo 2: testa MTP (layer 78). SEMPRE int8: a int4 i draft sbagliano quasi sempre
    # (acceptance 0-4% vs 39-59%, misurato — issue #8) e la speculazione non parte mai.
    mtp_cmd=list(base); i=mtp_cmd.index("--ebits"); mtp_cmd[i+1]=str(max(8,a.ebits))
    print(f"  {C.dim}[2/2] int8 MTP head (speculative drafts){C.r}")
    sys.exit(subprocess.call(mtp_cmd+["--mtp"]))


def cmd_mirror(a):
    if not a.mirror:
        sys.exit("mirror path required: pass --mirror or set COLI_MODEL_MIRROR")
    command = [sys.executable, os.path.join(TOOLS, "mirror_plan.py"), a.action,
               "--model", a.model, "--mirror", a.mirror]
    source_dirs = list(a.source_dir)
    configured = os.environ.get("COLI_MODEL_DIRS", "")
    if configured:
        source_dirs.extend(part.strip() for part in re.split(r"[;,]", configured)
                           if part.strip())
    for directory in source_dirs:
        command += ["--source-dir", directory]
    if a.usage:
        command += ["--usage", a.usage]
    if a.action != "verify":
        command += ["--budget-gib", str(a.budget_gib),
                    "--reserve-gib", str(a.reserve_gib)]
    return subprocess.call(command)


def main():
    common=argparse.ArgumentParser(add_help=False)
    common.add_argument("--model", default=DEF_MODEL); common.add_argument("--ram", type=int, default=0)  # 0 = auto (il motore usa l'88% della RAM disponibile)
    common.add_argument("--auto-tier",action="store_true",help="automatically apply the RAM/VRAM plan")
    common.add_argument("--no-tune-profile",action="store_true",
                        help="ignore a saved measured tuning profile")
    common.add_argument("--ctx",type=int,default=0)
    common.add_argument("--gpu",default=None,help="auto, none, or a device list such as 0,1")
    common.add_argument("--vram",type=float,default=0,help="total VRAM budget in GB (0=auto)")
    common.add_argument("--policy",choices=("quality","balanced","experimental-fast"),
                        default=os.environ.get("COLI_POLICY","quality"),
                        help="resource policy (explicit --topk/--topp overrides warn and proceed)")
    common.add_argument("--repin", type=int, default=0, help="adapt RAM/VRAM experts every N tokens")
    # Absent (None) = not explicitly set: the glm engine gets the 0 sentinel and picks
    # (8 historically; 1 on Metal+darwin+fast SSD, #379 -- an honest F_NOCACHE probe
    # measures "fast", cached in <model>/.coli_ssd), while non-glm arches get the
    # legacy 8 (openai_server.cap_for_arch). An explicit --cap N -- 0 included --
    # reaches the engine verbatim; see colibri.c's coli_resolve_cap().
    common.add_argument("--cap", type=int, default=None, help="cache slots/layer (default: auto)")
    # default None, resolved per command by ngen_for() below. 1024 is a sensible
    # safety net for a one-shot `coli run`; it is the wrong number for a session
    # where the browser shows a "max output tokens" control, because the server
    # clamps to it and the control silently does nothing above 1024 (#889).
    common.add_argument("--ngen", type=int, default=None)  # rete di sicurezza: la fine vera la decidono gli stop token
    common.add_argument("--topp", type=float, default=0); common.add_argument("--topk", type=int, default=0)
    common.add_argument("--effort", choices=("minimal","low","medium","high","xhigh"),
                        default=None)   # quanto il modello riflette prima di rispondere
    # --think / --no-think: se riflettere del tutto. Spento, il prompt chiude
    # subito il blocco e la risposta arriva alla prima parola.
    common.add_argument("--think", action="store_true", default=None)
    common.add_argument("--no-think", dest="think", action="store_false")
    common.add_argument("--temp", type=float, default=None)  # temperatura token (0=greedy, default 1.0+nucleus .95)
    common.add_argument("--cluster-workers", default=os.environ.get("CLUSTER_WORKERS"),
                        help="comma-separated expert workers, host:port,...")
    common.add_argument("--cluster-coordinator", default=os.environ.get("CLUSTER_COORDINATOR"),
                        help="control-plane URL used to discover expert workers")
    ap=argparse.ArgumentParser(prog="coli", parents=[common], description="colibri — run GLM-5.2 locally")
    ap.add_argument("--version", action="version", version=f"colibri {_version}")
    sub=ap.add_subparsers(dest="cmd")
    sub.add_parser("build", parents=[common]); sub.add_parser("info", parents=[common])
    pp=sub.add_parser("plan",parents=[common])
    pp.add_argument("--json",action="store_true")
    pm=sub.add_parser("mirror", parents=[common],
                      help="plan, stage, or verify a usage-ranked partial model mirror")
    pm.add_argument("action", choices=("plan", "stage", "verify"))
    pm.add_argument("--mirror", default=os.environ.get("COLI_MODEL_MIRROR"))
    pm.add_argument("--source-dir", action="append", default=[])
    pm.add_argument("--usage")
    pm.add_argument("--budget-gib", type=float, default=0)
    pm.add_argument("--reserve-gib", type=float, default=10)
    pd=sub.add_parser("doctor",parents=[common])
    pd.add_argument("--json",action="store_true",help="emit a versioned JSON report")
    pd.add_argument("--deep",action="store_true",
                    help="strictly check every tensor layout, model index, and configured mirror")
    pt=sub.add_parser("tune",parents=[common],
                      description="measure and save the fastest quality-preserving execution profile",
                      help="measure and save the fastest quality-preserving execution profile")
    pt.add_argument("--prompt",default="Explain why fixed-token replay makes a benchmark reproducible.")
    pt.add_argument("--rotate-prompt",action="append",default=None,
                    help="additional persistent-workload prompt; repeat for a custom rotation")
    pt.add_argument("--tokens",type=int,default=16,help="calibration continuation length")
    pt.add_argument("--repeats",type=int,choices=range(1,6),default=2,
                    help="requests per candidate (one persistent engine on non-GLM)")
    pt.add_argument("--timeout",type=int,default=900,help="seconds allowed for each engine run")
    pt.add_argument("--min-gain",type=float,default=0.03,
                    help="minimum fractional throughput gain required to accept a candidate")
    pt.add_argument("--profile-dir",default=None,help=argparse.SUPPRESS)
    pr=sub.add_parser("run", parents=[common]);  pr.add_argument("prompt", nargs="*")
    pc=sub.add_parser("chat", parents=[common])
    pc.add_argument("--attach", nargs="?", const="http://127.0.0.1:8000", default=None,
                    help="chat against a running `coli serve` instead of spawning an engine "
                         "(keeps the model loaded and the expert cache warm across chat sessions). "
                         "Bare --attach probes localhost:8000.")
    pc.add_argument("--no-attach", action="store_true",
                    help="never auto-attach, always spawn a private engine")
    pc.add_argument("--api-key", default=os.environ.get("COLI_API_KEY"))
    pc.add_argument("--stats", choices=list(CHAT_STATS_MODES), default=None,
                    help="generation statistics footer after each answer: full (tokens, seconds, tok/s), "
                         "compact (tokens, tok/s), off. Default from COLI_CHAT_STATS, else full. "
                         "Counts are exact when the server reports completion_tokens, ~estimated otherwise.")
    ps=sub.add_parser("serve", parents=[common])
    ps.add_argument("--host",default="127.0.0.1"); ps.add_argument("--port",type=int,default=8000)
    ps.add_argument("--model-id",default=os.environ.get("COLI_MODEL_ID"))
    ps.add_argument("--api-key",default=os.environ.get("COLI_API_KEY"))
    ps.add_argument("--cors-origin",action="append",default=None)
    ps.add_argument("--max-queue",type=int,default=int(os.environ.get("COLI_MAX_QUEUE","8")))
    ps.add_argument("--queue-timeout",type=float,default=float(os.environ.get("COLI_QUEUE_TIMEOUT","300")))
    ps.add_argument("--kv-slots",type=int,default=int(os.environ.get("COLI_KV_SLOTS","1")))
    ps.add_argument("--allowed-host",action="append",
        default=[h.strip() for h in os.environ.get("COLI_ALLOWED_HOSTS","").split(",") if h.strip()],
        help="additional Host header accepted by the DNS-rebinding guard; repeat as needed")
    pcluster=sub.add_parser("cluster", help="run the local-cluster control plane or an expert worker")
    cluster_sub=pcluster.add_subparsers(dest="cluster_cmd")
    pcoord=cluster_sub.add_parser("coordinator", help="serve worker registration and discovery")
    pcoord.add_argument("--host",default="127.0.0.1"); pcoord.add_argument("--port",type=int,default=8765)
    pcoord.add_argument("--stale-after",type=float,default=30.0)
    pcoord.add_argument("--allowed-host",action="append",
                        default=[h.strip() for h in os.environ.get("COLI_ALLOWED_HOSTS","").split(",") if h.strip()],
                        help="additional Host header accepted by the DNS-rebinding guard; repeat as needed")
    pworker=cluster_sub.add_parser("worker", parents=[common], help="serve disk-backed expert compute")
    pworker.add_argument("--port",type=int,default=int(os.environ.get("CLUSTER_WORKER_PORT","9100")))
    pworker.add_argument("--coordinator",default=os.environ.get("CLUSTER_COORDINATOR"))
    pworker.add_argument("--node-id",default=None); pworker.add_argument("--advertise-host",default=None)
    pworker.add_argument("--layers",default="all",help="topology label, e.g. 0-37")
    pworker.add_argument("--ebits",type=int,default=8); pworker.add_argument("--dbits",type=int,default=8)
    pst=sub.add_parser("stop", parents=[common], help="shut down a running coli serve and its engine")
    pst.add_argument("--port",type=int,default=8000); pst.add_argument("--dry-run",action="store_true")
    pw=sub.add_parser("web", parents=[common], help="serve + open the dashboard in a browser")
    for arg,kw in (("--host",dict(default="127.0.0.1")),("--port",dict(type=int,default=8000)),
                   ("--model-id",dict(default=os.environ.get("COLI_MODEL_ID"))),
                   ("--api-key",dict(default=os.environ.get("COLI_API_KEY"))),
                   ("--cors-origin",dict(action="append",default=None)),
                   ("--max-queue",dict(type=int,default=int(os.environ.get("COLI_MAX_QUEUE","8")))),
                   ("--queue-timeout",dict(type=float,default=float(os.environ.get("COLI_QUEUE_TIMEOUT","300")))),
                   ("--kv-slots",dict(type=int,default=int(os.environ.get("COLI_KV_SLOTS","1")))),
                   ("--allowed-host",dict(action="append",
                       default=[h.strip() for h in os.environ.get("COLI_ALLOWED_HOSTS","").split(",") if h.strip()],
                       help="additional Host header accepted by the DNS-rebinding guard; repeat as needed"))):
        pw.add_argument(arg,**kw)
    pw.add_argument("--no-browser",action="store_true",help="don't auto-open the browser")
    pb=sub.add_parser("bench", parents=[common]); pb.add_argument("tasks", nargs="*")
    pb.add_argument("--limit",type=int,default=40)
    if sys.platform == "win32":
        _cache_root = os.environ.get("LOCALAPPDATA", os.path.expanduser("~\\AppData\\Local"))
    else:
        _cache_root = os.environ.get("XDG_CACHE_HOME", os.path.expanduser("~/.cache"))
    _bench_cache = os.path.join(_cache_root, "colibri", "bench")
    pb.add_argument("--data",default=_bench_cache)
    pc=sub.add_parser("convert", parents=[common])
    pc.add_argument("--repo",default=None,
                    help="HF repo to download and convert (default zai-org/GLM-5.2-FP8). "
                         "--model is the OUTPUT directory, not a checkpoint to convert.")
    # default=None e non il valore: cosi' e' argparse a dire se l'opzione e'
    # stata SCRITTA (--ebits 3, --ebits=3, o abbreviata), e cmd_convert puo'
    # rifiutarla su un convertitore che non la prende invece di ignorarla.
    # I valori di default veri stanno in _CONVERT_DEFAULTS.
    pc.add_argument("--ebits",type=int,default=None,help="routed expert bits (default 4)")
    pc.add_argument("--io-bits",type=int,default=None,help="embedding/head bits (default 8)")
    pc.add_argument("--xbits",type=int,default=None,help="streaming expert bits (default = ebits)")
    pc.add_argument("--group-size",type=int,default=None,
        help="int4 scale group size: 64 (default, group-scaled quality) or 0 (legacy per-row)")
    pc.add_argument("--no-mtp",action="store_true",help="skip the MTP head (no speculative drafts)")
    a=ap.parse_args()
    handler={"build":cmd_build,"info":cmd_info,"plan":cmd_plan,"mirror":cmd_mirror,
             "doctor":cmd_doctor,"tune":cmd_tune,
             "run":cmd_run,"chat":cmd_chat,"serve":cmd_serve,"stop":cmd_stop,"bench":cmd_bench,
             "convert":cmd_convert,"web":cmd_web}.get(a.cmd)
    if a.cmd=="cluster":
        if a.cluster_cmd=="coordinator": handler=cmd_cluster_coordinator
        elif a.cluster_cmd=="worker": handler=cmd_cluster_worker
    if handler:
        try:
            sys.exit(handler(a) or 0)
        except (FamilyConfigError,UnknownFamilyError) as error:
            sys.exit(f"{C.yel}unsupported model:{C.r} {error}")
    banner(); print(__doc__)

if __name__=="__main__":
    signal.signal(signal.SIGINT, signal.default_int_handler)
    main()
