minion.py

#!/usr/bin/env python3
"""minion — a deliberately tiny coding agent for self-hosted or remote models.

One file, one dep (`openai`), no TUI framework. Points at any OpenAI-compatible
endpoint (vLLM / llama.cpp / SGLang / Z.ai / OpenAI itself). Survives models
whose native tool-calling isn't wired up yet by falling back to parsing
<tool_call>...</tool_call> tags out of the text — the convention most open
models (Hermes/Qwen/Nemotron) emit.

  pip install openai
  export MINION_BASE_URL=http://localhost:8000/v1   # your served endpoint
  export MINION_MODEL=your-model-name
  export MINION_API_KEY=sk-noop                    # any string; local servers ignore it
  python minion.py

Multiple sources — define named endpoints and switch between them at runtime:

  MINION_SOURCES=local,zai
  MINION_SOURCE_LOCAL_BASE_URL=http://localhost:8080/v1
  MINION_SOURCE_ZAI_BASE_URL=https://api.z.ai/api/paas/v4
  MINION_SOURCE_ZAI_API_KEY=***                   # $name = key from env / ~/.env
  MINION_SOURCE_ZAI_MODEL=glm-x-preview

  python minion.py --source zai                     # start on Z.ai

Sessions — every chat is auto-saved to ~/.minion/sessions/ and resumable:

  python minion.py sessions           # list saved sessions (prints + exits)
  python minion.py sessions --page 2  # next page of saved sessions
  python minion.py sessions refactor  # …filtered by a substring query
  python minion.py --resume <id|short-id|prefix|title>  # resume a past session
  python minion.py --resume 1                        # resume the most recent
  /sessions                       # list recent sessions (with short ids)
  /sessions --page 2              # next page of recent sessions
  /resume <n|short-id|title>      # switch to another session mid-chat
  /save [title]                   # save the current session (title optional)

Toggles in-session: /source [name] [model]  /provider [source] [a,b,…|off]  /yolo  /approval [level]  /compress  /compact  /autocompress [pct|off|on]  /memory save|remember|list  /reset  /clear  /new  /recover  /sessions  /resume  /save  /delete  /quit
Flags: --yolo  --approval <all|low|medium|high|yolo>  --source <name>  --resume <target>  --session <id>
Env:   MINION_APPROVAL=<all|low|medium|high|yolo>  (persistent default approval mode; ~/.env or shell)
       MINION_AUTOCOMPRESS_PERCENT=<0-100>  (auto-compress threshold; 0=off; default 85)
       MINION_BACKEND=vllm  (disables llama.cpp-only recovery knobs like min_p /
                             repeat_penalty / DRY that vLLM's speculative decoder
                             rejects; omits them from extra_body on retries)
       TOGETHER_API_KEY  (auto-registers a built-in `together` source; default model zai-org/GLM-5.2)
       OPENROUTER_API_KEY  (auto-registers a built-in `openrouter` source; default model z-ai/glm-5.2,
            routed to parasail/fp8 — override with /provider or MINION_SOURCE_OPENROUTER_EXTRA_BODY)
"""
import json
import os
import random
import re
import secrets
import select
import shlex
import unicodedata
import shutil
import subprocess
import sys
import termios
import threading
import time
import urllib.error
import urllib.request

import httpx
from openai import OpenAI, APIConnectionError, APIError

# --- env file ---------------------------------------------------------------
# Load ~/.env (or MINION_ENV_FILE) into os.environ without clobbering vars
# already set in the shell. Lets source config / API keys live in one place
# instead of being exported in every terminal.
_ENV_FILE = os.path.expanduser(os.environ.get("MINION_ENV_FILE", "~/.env"))


def _load_env_file():
    try:
        with open(_ENV_FILE, encoding="utf-8") as f:
            for raw in f:
                line = raw.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                k, v = line.split("=", 1)
                k = k.strip()
                if k.startswith("export "):
                    k = k[len("export "):].strip()
                if not k or k in os.environ:
                    continue
                v = v.strip()
                if len(v) >= 2 and v[0] == v[-1] and v[0] in ("'", '"'):
                    v = v[1:-1]
                os.environ[k] = v
    except (OSError, IOError):
        pass


_load_env_file()


# --- sessions ---------------------------------------------------------------
# Chat history persistence. Each session is one JSON file under
# ~/.minion/sessions/ (override with MINION_HOME / MINION_SESSIONS_DIR).
# The file stores the exact `messages` array the model sees (system prompt
# + every turn + tool calls/results), plus a little metadata (id, title,
# created_at, updated_at, cwd, source). Greppable, human-readable, and it
# round-trips trivially — load it back in and you have a resumable chat.
#
# Design cribbed from Hermes (hermes_state.py / SessionDB), which uses a
# SQLite store + FTS5 search because it's a multi-platform gateway (web,
# CLI, Telegram, …) with billing and compression chains. minion is a single
# local agent, so a flat directory of JSON files gives the same UX
# (auto-save per turn, /sessions, /resume, /save) without the weight.

SESSION_HOME = os.path.expanduser(
    os.environ.get("MINION_HOME", "~/.minion")
)
SESSION_LIST_DEFAULT_LIMIT = 10
SESSION_LIST_MAX_LIMIT = 100


def _sessions_dir():
    """Where session files live. Honors MINION_SESSIONS_DIR, then MINION_HOME/sessions."""
    return os.path.expanduser(
        os.environ.get("MINION_SESSIONS_DIR",
                       os.path.join(SESSION_HOME, "sessions"))
    )


def _memories_dir():
    """Where memory files live. Always under SESSION_HOME/memories."""
    return os.path.join(SESSION_HOME, "memories")


def _new_session_id():
    """Short, unguessable, sortable-ish: YYYYMMDD-HHMMSS-<6 hex>."""
    stamp = time.strftime("%Y%m%d-%H%M%S")
    return f"{stamp}-{secrets.token_hex(3)}"


def _safe_title(text, maxlen=60):
    """Turn the first user message into a filesystem-safe-ish title.

    Collapses whitespace, strips control chars, clamps length. We don't
    scrub for path-separators beyond replacing them with spaces — the id
    (not the title) is the filename, so a weird title can't break lookup.
    """
    if not text:
        return None
    text = " ".join(str(text).split())
    text = "".join(c for c in text if c.isprintable())
    if len(text) > maxlen:
        text = text[:maxlen - 1] + "…"
    return text or None


def _is_empty_assistant_message(msg):
    if msg.get("role") != "assistant" or msg.get("tool_calls"):
        return False
    content = msg.get("content")
    if content is None:
        return True
    if isinstance(content, str):
        return not content.strip()
    if isinstance(content, list):
        for part in content:
            if isinstance(part, str) and part.strip():
                return False
            if isinstance(part, dict):
                text = part.get("text")
                if isinstance(text, str) and text.strip():
                    return False
        return True
    return False


def _prune_empty_assistant_messages(messages):
    """Drop assistant turns that have neither visible content nor tool calls."""
    if not isinstance(messages, list):
        return 0
    kept = [m for m in messages if not _is_empty_assistant_message(m)]
    removed = len(messages) - len(kept)
    if removed:
        messages[:] = kept
    return removed


def _session_path(session_id):
    return os.path.join(_sessions_dir(), f"{session_id}.json")


def _write_session(session_id, messages, meta=None):
    """Persist `messages` to the session file. Creates the dir if needed.

    Writes atomically (temp file + rename) so a crash mid-write can't
    corrupt the existing session. `meta` (title, source, cwd, …) is
    merged into the stored metadata.
    """
    d = _sessions_dir()
    os.makedirs(d, exist_ok=True)
    _prune_empty_assistant_messages(messages)
    path = _session_path(session_id)
    now = time.time()
    existing = {}
    try:
        with open(path, encoding="utf-8") as f:
            existing = json.load(f)
        if not isinstance(existing, dict):
            existing = {}
    except (OSError, IOError, json.JSONDecodeError):
        pass
    # Start from the existing metadata so partial writes (e.g. the session
    # description refresh, which only passes {"description", "desc_turns"})
    # don't clobber fields a prior _save_current wrote — token totals,
    # started_at, etc.  Then let `meta` override on top.
    data = dict(existing)
    data["id"] = session_id
    data["messages"] = messages
    data["created_at"] = existing.get("created_at", now)
    data["updated_at"] = now
    if meta:
        for k, v in meta.items():
            if v is not None:
                data[k] = v
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    os.replace(tmp, path)
    try:
        os.utime(path, (data["updated_at"], data["updated_at"]))
    except (OSError, TypeError, ValueError):
        pass


def _load_session(session_id):
    """Read a session file. Returns the dict (id, messages, meta…) or None."""
    path = _session_path(session_id)
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
        if isinstance(data, dict) and "messages" in data:
            _prune_empty_assistant_messages(data.get("messages"))
            return data
    except (OSError, IOError, json.JSONDecodeError):
        pass
    return None


def _session_files_newest():
    """Return session JSON filenames newest-first using filesystem mtimes.

    Listing many sessions should not require parsing every old transcript.
    The session writer keeps mtime aligned with updated_at, so this is a
    cheap index for recent-session browsing.
    """
    d = _sessions_dir()
    files = []
    try:
        with os.scandir(d) as entries:
            for entry in entries:
                if not entry.name.endswith(".json"):
                    continue
                try:
                    if not entry.is_file():
                        continue
                    files.append((entry.stat().st_mtime, entry.name))
                except OSError:
                    continue
    except OSError:
        return []
    files.sort(key=lambda item: (item[0], item[1]), reverse=True)
    return [name for _, name in files]


def _session_summary_from_file(fname):
    d = _sessions_dir()
    path = os.path.join(d, fname)
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict) or "messages" not in data:
            return None
    except (OSError, IOError, json.JSONDecodeError):
        return None

    sid = data.get("id") or fname[:-5]
    msgs = data.get("messages", [])
    preview = ""
    for m in msgs:
        if m.get("role") == "user" and m.get("content"):
            preview = _safe_title(m["content"]) or ""
            break
    return {
        "id": sid,
        "short": _short_id(sid),
        "title": data.get("title") or preview or "(empty)",
        "description": data.get("description"),
        "preview": preview,
        "updated_at": data.get("updated_at", 0),
        "n": len([m for m in msgs if m.get("role") != "system"]),
        "model": data.get("model"),
        "source": data.get("source"),
        "cwd": data.get("cwd"),
    }


def _session_matches_query(summary, query):
    if not query:
        return True
    q = query.lower()
    return (
        q in (summary.get("title") or "").lower()
        or q in (summary.get("description") or "").lower()
        or q in (summary.get("preview") or "").lower()
        or q in (summary.get("id") or "").lower()
        or q in (summary.get("short") or "").lower()
    )


def _list_sessions(limit=20, offset=0, query=None):
    """Return sessions newest-first as dicts: id, title, description, preview, n.

    `preview` is the first ~60 chars of the first user message; `description`
    is an optional model-generated one-liner refreshed every N turns (richer
    than the static first-message title). `n` is the turn count (non-system
    messages)."""
    limit = max(0, int(limit)) if limit is not None else None
    offset = max(0, int(offset or 0))
    files = _session_files_newest()
    out = []
    for fname in files:
        summary = _session_summary_from_file(fname)
        if not summary or not _session_matches_query(summary, query):
            continue
        out.append(summary)
        if query:
            continue
        if limit is not None and len(out) >= offset + limit:
            break
    if query:
        out.sort(key=lambda s: s["updated_at"], reverse=True)
    end = None if limit is None else offset + limit
    return out[offset:end]


def _delete_session(session_id):
    """Remove a session file. Returns True if something was deleted."""
    path = _session_path(session_id)
    try:
        os.remove(path)
        return True
    except OSError:
        return False


# After this many user turns, minion asks the model for a short (≤70 char)
# description of the whole conversation so far and stores it in the session
# file's `description` field. The description refreshes on every Nth turn
# thereafter, so it tracks what the chat is actually about as it evolves —
# far more useful in `minion sessions` than a static first-message title.
# 0 disables the refresh entirely (the auto-derived title is used as-is).
# The actual value is resolved after _env_int() is defined (below), so this
# is just a sentinel; see SESSION_DESC_REFRESH.
_DESC_REFRESH_DEFAULT = 6


def _short_id(session_id):
    """The scannable tail of a session id (the 6-hex suffix), for listings.

    The full id is `YYYYMMDD-HHMMSS-XXXXXX`; in a list of recent sessions the
    date+time prefix is shared/redundant, so we show just the 6 hex chars as a
    quick tag the user can grep or pass to --resume."""
    if session_id and "-" in session_id:
        return session_id.rsplit("-", 1)[-1]
    return session_id or ""


def _maybe_refresh_description(session_id, messages):
    """Ask the model for a one-line session description, if enough turns have
    passed since the last refresh.

    Refreshes at the configured interval (every SESSION_DESC_REFRESH user
    turns). The description is stored in the session file's `description`
    field and surfaced in `minion sessions` / `/sessions` listings. A failure
    (server down, empty response) leaves the existing description untouched.

    Returns the new description string, or None if no refresh happened.
    """
    if SESSION_DESC_REFRESH <= 0:
        return None
    # Count user turns (exclude synthetic runtime-note-only turns).
    user_turns = sum(
        1 for m in messages
        if m.get("role") == "user"
        and isinstance(m.get("content"), str)
        and not m["content"].lstrip().startswith("[")
    )
    existing = _load_session(session_id) or {}
    last_desc_turns = existing.get("desc_turns", 0)
    # Refresh when we've crossed a multiple of SESSION_DESC_REFRESH since the
    # last recorded refresh. Don't refresh before the first threshold (so a
    # 1-turn "hello" session doesn't burn a model call).
    if user_turns < SESSION_DESC_REFRESH:
        return None
    if user_turns - last_desc_turns < SESSION_DESC_REFRESH:
        return None
    # Build a compact transcript for the summarizer — truncate tool outputs
    # and skip the system prompt to keep the call cheap.
    def _trim(msgs, per_msg=500, cap=20):
        out = []
        for m in msgs:
            if m.get("role") == "system":
                continue
            c = m.get("content")
            if c is None and m.get("tool_calls"):
                calls = ", ".join(
                    f"{tc['function']['name']}(...)" for tc in m["tool_calls"])
                c = f"→ {calls}"
            elif isinstance(c, list):
                c = " ".join(p.get("text", "") for p in c if isinstance(p, dict))
            c = (c or "").replace("\n", " ").strip()
            if len(c) > per_msg:
                c = c[:per_msg - 1] + "…"
            out.append(f"[{m.get('role', '?')}] {c}")
            if len(out) >= cap:
                break
        return "\n".join(out)
    prompt = _trim(messages[-30:])  # last ~30 messages is plenty of context
    payload = [
        {"role": "system", "content": DESC_SYSTEM},
        {"role": "user", "content": prompt},
    ]
    try:
        _log_event("req", {"model": MODEL, "messages": payload,
                           "stream": False, "_purpose": "session_desc"})
        resp = client.chat.completions.create(
            model=MODEL, messages=payload, stream=False, timeout=20,
            **(ACTIVE.extra_request_kwargs() if ACTIVE else {}))
        try:
            _log_event("resp", {"_purpose": "session_desc", "data": resp.model_dump()})
        except Exception:
            pass
    except APIConnectionError:
        return None  # server down — leave existing description as-is
    except Exception:
        return None
    desc = (resp.choices[0].message.content or "").strip().splitlines()
    desc = desc[0].strip() if desc else ""
    if not desc:
        return None
    desc = _safe_title(desc, maxlen=70) or desc[:70]
    # Persist the description + the turn count it was generated at so we know
    # when to refresh next. Merge into existing meta (don't clobber messages).
    _write_session(session_id, messages, {"description": desc,
                                          "desc_turns": user_turns})
    return desc


def _resolve_session(target, sessions=None):
    """Resolve a user-typed target to a session id.

    Accepts: a full id, a numeric index into the recent-sessions list,
    a unique id prefix, or an exact title. Returns the id or None.
    """
    if not target:
        return None
    target = target.strip()
    sessions = sessions if sessions is not None else _list_sessions(limit=50)
    ids = [s["id"] for s in sessions]
    # numeric index → recent-sessions slot
    if target.isdigit():
        idx = int(target)
        if 1 <= idx <= len(sessions):
            return sessions[idx - 1]["id"]
    # exact id
    if target in ids:
        return target
    # unique prefix
    prefixed = [i for i in ids if i.startswith(target)]
    if len(prefixed) == 1:
        return prefixed[0]
    # short id (the 6-hex suffix shown in listings) — match the tail segment
    # since the date+time prefix is shared/redundant across sessions created
    # in the same minute.
    suffixed = [i for i in ids if i.endswith("-" + target) or _short_id(i) == target]
    if len(suffixed) == 1:
        return suffixed[0]
    # exact title
    titled = [s["id"] for s in sessions if s["title"] == target]
    if len(titled) == 1:
        return titled[0]
    return None


# --- model sources ----------------------------------------------------------
# minion talks to any OpenAI-compatible endpoint. A "source" bundles a
# base_url, api_key, and model name. Define sources with env vars:
#
#   MINION_SOURCES=local,zai
#   MINION_SOURCE_LOCAL_BASE_URL=http://localhost:8080/v1
#   MINION_SOURCE_ZAI_BASE_URL=https://api.z.ai/api/paas/v4
#   MINION_SOURCE_ZAI_API_KEY=$zai_test        ← $name = look up env/file key
#   MINION_SOURCE_ZAI_MODEL=glm-x-preview
#
# If no MINION_SOURCE_* vars are present, a single "local" source is built
# from the legacy MINION_BASE_URL / MINION_API_KEY / MINION_MODEL vars
# (same defaults as before, so existing setups keep working).
# Switch at runtime with /source.

class Source:
    def __init__(self, name, base_url, api_key, model=None, extra_body=None,
                 http_headers=None):
        self.name = name
        self.base_url = base_url
        self.api_key = api_key or "sk-noop"
        self.model = model or None  # None → ask the server at resolve time
        # Default HTTP headers sent on every request. Used by aggregators like
        # OpenRouter which read HTTP-Referer (your site URL) and X-Title (your
        # app name) to identify the calling app in their dashboard. Ignored by
        # non-consumers (llama.cpp, Together, …) so it's safe to always pass.
        self.http_headers = http_headers or None
        client_kwargs = {"base_url": base_url, "api_key": self.api_key}
        if self.http_headers:
            client_kwargs["default_headers"] = self.http_headers
        self.client = OpenAI(**client_kwargs)
        # Lazily-resolved max context window (in tokens) for the active model,
        # or None when no probe could determine it. Cached after first success
        # so the footer + /source list don't re-hit the network each turn.
        # Call resolve_context_window(force=True) to re-probe (e.g. after a
        # /source <name> <model> override pins a different model id).
        self._context_window = None
        # Per-request body fields the OpenAI SDK doesn't accept as kwargs, merged
        # into every chat.completions.create() call via extra_body. The canonical
        # use is OpenRouter provider routing — a `provider` object whose `order`
        # picks which upstream serves a given model id (different providers back
        # the same endpoint at different prices / precisions / latencies):
        #
        #   {"provider": {"order": ["parasail/fp8"], "allow_fallbacks": false,
        #                 "require_parameters": true, "data_collection": "deny"}}
        #
        # For local servers (llama.cpp) this is empty and contributes nothing;
        # unknown keys are silently ignored by non-consumers, so it's safe to
        # always pass it through. Set via MINION_SOURCE_<NAME>_EXTRA_BODY (a JSON
        # string) or the /provider slash command (ephemeral, per-session).
        self.extra_body = extra_body or None

    def extra_request_kwargs(self):
        """extra_body kwarg to merge into every chat.completions.create(), or
        {} when none is configured. Non-consumers ignore unknown body keys, so
        this is a no-op for backends that don't understand it (llama.cpp,
        Together, …)."""
        if not self.extra_body:
            return {}
        return {"extra_body": dict(self.extra_body)}

    # Server-advertised ids that carry no real identity — when /v1/models
    # returns one of these (bare llama.cpp with no --alias, or a generic
    # OpenAI-compat stub), fall through to a /props probe before giving up so
    # saved sessions don't all collapse into a meaningless "local-model".
    _GENERIC_MODEL_IDS = {"local-model", "model", "auto", "unknown", "gpt-3.5-turbo"}

    @staticmethod
    def _clean_model_id(advertised):
        """Shorten a server-advertised model id into a concise display name.

        Only touches local-server ids that look like file paths (llama.cpp,
        Ollama, …) where /v1/models returns the full GGUF/HF file path. Remote
        API ids in org/model form (e.g. "zai-org/GLM-5.2") are returned
        unchanged.

        Examples:
          /media/h/.../GLM-5.2-GGUF/UD-IQ4_NL/GLM-5.2-UD-IQ4_NL-00001-of-00009.gguf
            → GLM-5.2-UD-IQ4_NL
          /models/Meta-Llama-3-8B-Instruct-Q4_K_M.gguf
            → Meta-Llama-3-8B-Instruct-Q4_K_M
        Anything that doesn't look like a local model file path is returned
        as-is so we never mangle a legitimate remote model id.
        """
        if not advertised:
            return advertised
        # Only clean up local file paths — ones that end in a model file
        # extension. Anything else (org/model, bare names) is passed through.
        lower = advertised.lower()
        if not any(lower.endswith(ext) for ext in (".gguf", ".bin", ".safetensors")):
            return advertised
        name = advertised.rsplit("/", 1)[-1]
        # Strip shard suffix on multi-file quantizations
        # (e.g. "-00001-of-00009")
        name = re.sub(r"-\d{5}-of-\d{5}(?=$|\.)", "", name)
        # Strip extension
        for ext in (".gguf", ".bin", ".safetensors"):
            if name.lower().endswith(ext):
                name = name[: -len(ext)]
                break
        return name

    def resolve_model(self):
        if self.model:
            return self.model
        advertised = None
        try:
            advertised = self.client.models.list(timeout=10).data[0].id
        except Exception:
            advertised = None
        if advertised and advertised.strip().lower() not in self._GENERIC_MODEL_IDS:
            return self._clean_model_id(advertised)
        # The endpoint gave us nothing useful. Many local servers (llama.cpp's
        # llama-server) expose the actually-loaded model file at /props even
        # when /v1/models only advertises a generic id — recover the real name
        # (incl. quant) from there. Best-effort; any failure keeps the old value.
        probed = self._probe_loaded_model()
        return probed or advertised or "local-model"

    def _probe_loaded_model(self):
        """Best-effort GET <server>/props for the loaded model path/alias.
        llama.cpp serves /props at the root (not under /v1). Returns None on
        any failure — never raises, never required, makes no assumption about
        what (if anything) consumes minion's session metrics."""
        try:
            root = re.sub(r"/v\d+/?$", "", self.base_url.rstrip("/"))
            with urllib.request.urlopen(root + "/props", timeout=5) as r:
                props = json.loads(r.read().decode("utf-8"))
        except Exception:
            return None
        for key in ("model_alias", "model_path", "model"):
            val = props.get(key)
            if isinstance(val, str) and val.strip() and \
                    val.strip().lower() not in self._GENERIC_MODEL_IDS:
                return val.strip()
        return None

    def display_model(self):
        return self.model or "auto"

    # Regex for the provider-agnostic context-limit probe: Together (and any
    # OpenAI-compat backend that mirrors OpenAI's error wording) rejects an
    # over-large request with "This model's maximum context length is N tokens,
    # but the request …". We send a deliberately over-sized max_tokens with a
    # one-token prompt so the server tells us N without us ever generating.
    _CTX_LIMIT_RE = re.compile(r"maximum context length is (\d+) tokens", re.IGNORECASE)

    def _is_local(self):
        """True if this source points at a host on the local machine or LAN
        (llama.cpp, Ollama, etc.) rather than a remote API (Together, Z.ai,
        OpenAI, …). Local servers expose /v1/models meta and /props cheaply
        (sub-millisecond LAN round-trips); remote hosts return empty lists or
        404 there and waste a multi-second TLS round-trip, so the probe order
        is flipped for them (over-max_tokens first)."""
        host = (self.base_url or "").lower()
        # strip the scheme
        if "://" in host:
            host = host.split("://", 1)[1]
        host = host.split("/", 1)[0].split(":", 1)[0]
        if not host:
            return True  # blank → assume local
        if host in ("localhost", "0.0.0.0", "::1"):
            return True
        # 127.x.x.x / 10.x / 192.168.x / 169.254.x / 172.16-31.x are LAN
        if re.match(r"^(127\.|10\.|192\.168\.|169\.254\.|172\.(1[6-9]|2[0-9]|3[01])\.)", host):
            return True
        return False

    def resolve_context_window(self, model=None, force=False):
        """Best-effort max context window (tokens) for `model` on this source.

        Returns the cached value if present (unless force=True), else probes
        and caches. Never raises — any failure returns None and the footer
        just omits the "/<max>" suffix. Probe order depends on host locality:

          • Local (llama.cpp etc.) — /v1/models meta, then /props, then the
            over-max_tokens chat probe. Local metadata endpoints are cheap
            (sub-ms LAN) and llama.cpp stashes n_ctx in data[0].meta, so they
            win on the first try.

          • Remote (Together, Z.ai, OpenAI, …) — over-max_tokens chat probe
            FIRST (one cheap request, ~0.1s, works for any OpenAI-compat
            host that mirrors the "maximum context length is N tokens" 400),
            then /v1/models as a fallback. Remote /v1/models often returns an
            empty list or lacks context_length, and the TLS round-trip alone
            costs 2-3s — leading with it added multi-second latency to the
            background probe, so the first turn's footer rendered before the
            max resolved. The chat probe resolves in well under a second.

        `model` defaults to the source's configured model, falling back to a
        live resolve_model() so an auto source is probed against the model it
        will actually use. The result is keyed to that model id, so a
        /source <name> <other-model> override should call with force=True.

        A single probe attempt can hit a transient error — most notably
        llama.cpp's 503 "Loading model" while the weights are still being
        mmapped, which blocks /v1/models AND /props even though a chat slot
        may already be serving. So a failed attempt does NOT cache None;
        the caller (footer/listing) re-arms the probe on a later turn until
        it resolves. Use force=True to retry immediately (e.g. an explicit
        /source switch the user is waiting on)."""
        if not force and self._context_window is not None:
            return self._context_window
        mid = model or self.model or self.resolve_model()
        if self._is_local():
            n = self._ctx_from_models(mid) or self._ctx_from_props()
            if n is None:
                n = self._ctx_from_overrun_probe(mid)
        else:
            # Remote: the over-max_tokens chat probe is cheap and universal;
            # lead with it so the max resolves on the first turn instead of
            # after a multi-second /v1/models round-trip that returns nothing.
            n = self._ctx_from_overrun_probe(mid)
            if n is None:
                n = self._ctx_from_models(mid) or self._ctx_from_props()
        if isinstance(n, int) and n > 0:
            self._context_window = n
        # NOTE: a miss does NOT set _context_window — stays None so the
        # footer re-probes next turn (the server may have finished loading).
        return self._context_window

    def _ctx_from_models(self, mid):
        """Pull n_ctx / context_length from GET /v1/models. llama.cpp stashes
        it under data[0].meta.n_ctx; some generic hosts expose context_length
        as a top-level model field. Returns int or None."""
        try:
            data = self.client.models.list(timeout=10).data
        except Exception:
            return None
        # Prefer the entry whose id matches the active model (multi-model
        # hosts), else the first. Match loosely on a trailing token so a full
        # path id ("…/GLM-5.2-…gguf") still hits a short alias ("GLM-5.2").
        pick = None
        if mid:
            mtail = mid.rsplit("/", 1)[-1].lower()
            for m in data:
                if (m.id or "").lower() == mid.lower() or \
                   (m.id or "").rsplit("/", 1)[-1].lower() == mtail:
                    pick = m
                    break
        if pick is None and data:
            pick = data[0]
        if pick is None:
            return None
        # OpenAI SDK preserves unknown fields in __pydantic_extra__.
        extra = getattr(pick, "__pydantic_extra__", None) or {}
        meta = extra.get("meta") if isinstance(extra, dict) else None
        if isinstance(meta, dict):
            for k in ("n_ctx", "context_length", "context_window",
                      "max_context_length", "max_input_tokens"):
                v = meta.get(k)
                if isinstance(v, int) and v > 0:
                    return v
        # Some hosts put it directly on the model object (OpenAI-compat extra).
        for k in ("context_length", "context_window", "max_context_length",
                  "max_input_tokens", "max_tokens", "max_model_len"):
            v = extra.get(k) if isinstance(extra, dict) else None
            if v is None:
                v = getattr(pick, k, None)
            if isinstance(v, int) and v > 0:
                return v
        return None

    def _ctx_from_props(self):
        """llama.cpp /props fallback: default_generation_settings.n_ctx."""
        try:
            root = re.sub(r"/v\d+/?$", "", self.base_url.rstrip("/"))
            with urllib.request.urlopen(root + "/props", timeout=5) as r:
                props = json.loads(r.read().decode("utf-8"))
        except Exception:
            return None
        dgs = props.get("default_generation_settings")
        if isinstance(dgs, dict):
            for k in ("n_ctx", "context_length", "context_window"):
                v = dgs.get(k)
                if isinstance(v, int) and v > 0:
                    return v
        # Some builds put n_ctx at the top level of /props.
        for k in ("n_ctx", "context_length", "context_window"):
            v = props.get(k)
            if isinstance(v, int) and v > 0:
                return v
        return None

    def _ctx_from_overrun_probe(self, mid):
        """Send a chat completion with max_tokens far beyond any real context
        and a 1-token prompt; parse the 400's "maximum context length is N
        tokens" message. Works for Together and any host mirroring OpenAI's
        over-limit wording. No tokens are generated (the request is rejected
        before inference). Returns int or None."""
        if not mid:
            return None
        try:
            self.client.chat.completions.create(
                model=mid,
                messages=[{"role": "user", "content": "hi"}],
                max_tokens=10_000_000,
                stream=False,
                timeout=30,
            )
        except Exception as e:
            msg = str(e)
            body = getattr(e, "body", None)
            if isinstance(body, dict):
                # Together nests it under body["message"]["message"]; OpenAI
                # under body["error"]["message"]. Flatten to a string search.
                msg = json.dumps(body) + " " + msg
            m = self._CTX_LIMIT_RE.search(msg)
            if m:
                return int(m.group(1))
        return None


def _resolve_api_key(val):
    """$name → look up env var (populated from ~/.env if present); else literal."""
    if val and val.startswith("$"):
        return os.environ.get(val[1:], "")
    return val


def _parse_extra_body(raw):
    """Parse a MINION_SOURCE_<NAME>_EXTRA_BODY env value into a dict, or None.

    Accepts a JSON object string and merges it into every chat request as the
    OpenAI SDK's `extra_body` — the canonical channel for fields the SDK doesn't
    model as top-level kwargs. The primary use case is OpenRouter provider
    routing (a `provider` object whose `order` picks which upstream serves a
    model id). Returns None on a blank/invalid value (with a stderr warning on
    bad JSON) so a typo never silently drops routing — it fails loudly at load.

    Example:
        MINION_SOURCE_OR_EXTRA_BODY={"provider":{"order":["parasail/fp8"],"allow_fallbacks":false}}
    """
    if not raw or not raw.strip():
        return None
    try:
        obj = json.loads(raw)
    except json.JSONDecodeError as e:
        sys.stderr.write(
            f"minion: ignoring bad MINION_*_EXTRA_BODY JSON ({e.msg} at "
            f"char {e.pos}); provider routing is NOT set\n")
        return None
    if isinstance(obj, dict) and obj:
        return obj
    if isinstance(obj, dict):
        return None  # empty object {} → no routing, same as unset
    sys.stderr.write(
        "minion: MINION_*_EXTRA_BODY must be a JSON object, ignoring "
        f"(got {type(obj).__name__}); provider routing is NOT set\n")
    return None


def _build_http_headers(app_name=None, app_url=None):
    """Build default HTTP headers for API requests, for aggregators like
    OpenRouter which read HTTP-Referer (your site URL) and X-Title (your app
    name) to identify the calling app in their dashboard.

    Set via MINION_SOURCE_<NAME>_APP_NAME and MINION_SOURCE_<NAME>_APP_URL.
    Returns None when neither is provided so the OpenAI client is constructed
    without default_headers (no overhead for non-aggregator sources).
    """
    headers = {}
    if app_url:
        headers["HTTP-Referer"] = app_url
    if app_name:
        headers["X-Title"] = app_name
    return headers or None


def _discover_sources():
    """Build SOURCES + SOURCE_ORDER from MINION_SOURCE_* env vars, falling
    back to a single 'local' source from the legacy MINION_* vars."""
    names = []
    raw = os.environ.get("MINION_SOURCES", "")
    if raw:
        names = [n.strip() for n in raw.split(",") if n.strip()]
    # auto-discover from MINION_SOURCE_<NAME>_BASE_URL if MINION_SOURCES absent
    if not names:
        prefix = "MINION_SOURCE_"
        found = []
        for k in os.environ:
            if k.startswith(prefix) and k.endswith("_BASE_URL"):
                found.append(k[len(prefix):-len("_BASE_URL")].lower())
        names = sorted(found)
    for name in names:
        p = f"MINION_SOURCE_{name.upper()}_"
        base_url = os.environ.get(p + "BASE_URL")
        if not base_url:
            continue
        api_key = _resolve_api_key(os.environ.get(p + "API_KEY"))
        model = os.environ.get(p + "MODEL")
        extra_body = _parse_extra_body(os.environ.get(p + "EXTRA_BODY"))
        # HTTP headers for aggregators like OpenRouter: HTTP-Referer (site
        # URL) and X-Title (app name) identify the calling app in their
        # dashboard. Set via MINION_SOURCE_<NAME>_APP_NAME /
        # MINION_SOURCE_<NAME>_APP_URL.
        http_headers = _build_http_headers(
            os.environ.get(p + "APP_NAME"),
            os.environ.get(p + "APP_URL"))
        src = Source(name, base_url, api_key, model, extra_body=extra_body,
                     http_headers=http_headers)
        SOURCES[name] = src
        SOURCE_ORDER.append(name)
    if not SOURCES:
        # legacy fallback: one source from MINION_BASE_URL etc.
        src = Source(
            "local",
            os.environ.get("MINION_BASE_URL", "http://localhost:8080/v1"),
            os.environ.get("MINION_API_KEY", "sk-noop"),
            os.environ.get("MINION_MODEL"),
        )
        SOURCES["local"] = src
        SOURCE_ORDER.append("local")

    # Built-in convenience source: Together AI. It's a multi-model host (you
    # point it at any model by id), so unlike the local fallback it only
    # appears when a key is actually available — and only when the user
    # hasn't already defined a "together" source themselves (so an explicit
    # MINION_SOURCE_TOGETHER_* config always wins). Registered last, so the
    # user's first/source-listed source stays the default at startup and
    # `/source together` (or `--source together`) is purely opt-in.
    if "together" not in SOURCES:
        together_key = _resolve_api_key(os.environ.get("TOGETHER_API_KEY", ""))
        if together_key:
            SOURCES["together"] = Source(
                "together",
                "https://api.together.xyz/v1",
                together_key,
                "zai-org/GLM-5.2",  # default model; override per-switch via /source together <model>
            )
            SOURCE_ORDER.append("together")

    # Built-in convenience source: OpenRouter. Like Together it's a multi-model
    # host (you address any model by id), so it only registers when a key is
    # available and only when the user hasn't defined an "openrouter" source
    # themselves (an explicit MINION_SOURCE_OPENROUTER_* config always wins).
    # Registered last, so it never displaces the user's default startup source —
    # opt in with `/source openrouter` (or `--source openrouter`).
    #
    # The difference from Together is provider routing: OpenRouter fronts many
    # upstream providers behind one model id, each with its own price / precision
    # / latency / data-collection policy. Routing rides in the request body as a
    # `provider` object (extra_body), so the built-in seeds it from
    # MINION_SOURCE_OPENROUTER_EXTRA_BODY (a JSON string) if present, and the
    # ephemeral /provider command can override it per-session. The default model
    # is z-ai/glm-5.2; the default routing pins parasail/fp8 — cheap, fast, and a
    # deliberate single-provider choice so you know exactly what's serving you.
    if "openrouter" not in SOURCES:
        or_key = _resolve_api_key(os.environ.get("OPENROUTER_API_KEY", ""))
        if or_key:
            or_body = _parse_extra_body(
                os.environ.get("MINION_SOURCE_OPENROUTER_EXTRA_BODY", ""))
            if or_body is None:
                or_body = {"provider": {"order": ["parasail/fp8"],
                                        "allow_fallbacks": False}}
            or_headers = _build_http_headers(
                os.environ.get("MINION_SOURCE_OPENROUTER_APP_NAME", "Minion"),
                os.environ.get("MINION_SOURCE_OPENROUTER_APP_URL",
                               "https://github.com/Sentdex/minion"))
            SOURCES["openrouter"] = Source(
                "openrouter",
                "https://openrouter.ai/api/v1",
                or_key,
                "z-ai/glm-5.2",  # default model; override via /source openrouter <model>
                extra_body=or_body,
                http_headers=or_headers,
            )
            SOURCE_ORDER.append("openrouter")


SOURCES = {}        # name → Source
SOURCE_ORDER = []   # preserve definition order for /source listing
ACTIVE = None       # current Source

# `client` and `MODEL` are bare globals read throughout the file. They always
# mirror the active source; switch_source() reassigns both. Every function that
# needs them (open_stream, _assess_risk, compress, …) does a call-time global
# lookup, so a mid-session swap is picked up instantly — same pattern /yolo
# already uses for its own globals.
client = None
MODEL = None

# Updated by model_turn() on every turn that reports a prompt-token count
# (llama.cpp timings.prompt_n or OpenAI/Z.ai usage.prompt_tokens). Read by
# _maybe_autocompress() after a turn settles to decide whether the context
# window is full enough to warrant a silent auto-compress.
_LAST_PROMPT_TOKENS = 0


def switch_source(name, model_override=None):
    """Swap the active source. Reassigns client + MODEL globals. Returns True
    on success, False (with a message) if the name is unknown.

    `model_override` (optional) pins MODEL to a specific model id for this
    switch instead of resolving the source's default — used by
    `/source <name> <model>` so a multi-model host (e.g. Together) can be
    pointed at any of its models without a config edit. A bare switch (no
    override) always falls back to the source's configured/default model,
    so `/source together` returns to GLM-5.2 even after an override."""
    global ACTIVE, client, MODEL
    src = SOURCES.get(name)
    if not src:
        print(f"{RED}  ✗ unknown source {name!r}{RESET}")
        return False
    ACTIVE = src
    client = src.client
    MODEL = model_override if model_override else src.resolve_model()
    # A model change (different source, or a per-switch model override on a
    # multi-model host like Together) means the cached max-context may no longer
    # apply — drop it so the footer re-probes against the new model next turn.
    src._context_window = None
    return True


def restore_source(source_name, model=None):
    """Best-effort restore of the source (and optional model) a session was
    started on, used when resuming. Switches source if it's configured and
    differs from the active one; if `model` is set and not equal to the
    source's configured default, pins MODEL to it so a `/source <name>
    <model>` override survives a resume. Returns True if the source is now
    the requested one."""
    if not source_name or source_name not in SOURCES:
        return False
    src = SOURCES[source_name]
    pin = model if model and model != src.model else None
    if source_name == ACTIVE.name and MODEL == (pin or src.model):
        return True
    switch_source(source_name, model_override=pin)
    return True


def _set_provider_order(src, order, allow_fallbacks=None):
    """Set the OpenRouter `provider.order` routing for `src` in place.

    `order` is a list of provider names (e.g. ["parasail/fp8", "Together"]); a
    deliberate, non-empty list pins routing to exactly those providers and —
    unless `allow_fallbacks` is explicitly True — disables OpenRouter's usual
    fallback to other upstreams, so you know precisely what's serving the model.
    Pass `allow_fallbacks=True` to keep falling through the rest of OpenRouter's
    pool after the ordered list is exhausted. An empty `order` clears routing
    entirely (sets src.extra_body to None), letting OpenRouter pick freely.

    Only mutates `provider`-style routing; any other keys already in
    `extra_body` are preserved. Non-routing sources (local llama.cpp, Together,
    …) accept the field harmlessly — they ignore unknown request-body keys — so
    this is safe to call on any source, though it only meaningfully routes on
    OpenRouter."""
    if order:
        body = dict(src.extra_body or {})
        prov = dict(body.get("provider") or {})
        prov["order"] = list(order)
        if allow_fallbacks is not None:
            prov["allow_fallbacks"] = allow_fallbacks
        else:
            # Pinning an explicit order means "I chose these specifically" →
            # don't silently fall through to other providers. The caller can
            # override by passing allow_fallbacks=True.
            prov["allow_fallbacks"] = False
        body["provider"] = prov
        src.extra_body = body
    else:
        src.extra_body = None


def _provider_order(src):
    """The current `provider.order` list on `src`, or [] when unset. Read-only
    accessor for the /provider listing; never raises."""
    body = src.extra_body if isinstance(src.extra_body, dict) else None
    if not body:
        return []
    prov = body.get("provider")
    if not isinstance(prov, dict):
        return []
    order = prov.get("order")
    return list(order) if isinstance(order, list) else []


_discover_sources()

# Pick the starting source: --source flag, then MINION_ACTIVE env, then first.
_start = None
for _i, _arg in enumerate(sys.argv):
    if _arg == "--source" and _i + 1 < len(sys.argv):
        _start = sys.argv[_i + 1]
        break
_start = _start or os.environ.get("MINION_ACTIVE") or (SOURCE_ORDER[0] if SOURCE_ORDER else None)
if not _start or _start not in SOURCES:
    _start = SOURCE_ORDER[0] if SOURCE_ORDER else None
if _start:
    switch_source(_start)


# --- approval gating --------------------------------------------------------
# Three risk levels (low < medium < high) plus an explicit "prompt all" state
# and YOLO mode. APPROVE_LEVEL is the maximum risk level to AUTO-APPROVE:
# actions classified at ≤ APPROVE_LEVEL run without prompting; anything
# strictly above prompts. APPROVE_LEVEL=None means prompt for every classified
# action. YOLO=True short-circuits entirely and skips the classifier call.
#   None     → prompt low + medium + high
#   "low"    → auto-approve low (reads, grep, wc); prompt medium + high
#   "medium" → auto-approve low + medium (edits, cp, mv, tests); prompt high
#   "high"   → auto-approve low + medium + high
LEVEL_ORDER = {"low": 0, "medium": 1, "high": 2}

# Accept full words and common abbreviations (med, hi, lo, m, h, l …).
_LEVEL_ALIASES = {
    "l": "low", "lo": "low", "low": "low",
    "m": "medium", "med": "medium", "mid": "medium", "medium": "medium",
    "h": "high", "hi": "high", "high": "high",
}


def _normalize_level(arg):
    """Resolve 'med' → 'medium', 'hi' → 'high', etc. Returns the canonical
    level name or None if the input isn't recognised."""
    return _LEVEL_ALIASES.get(arg.lower().strip())


def _normalize_approval(arg):
    """Resolve approval setting aliases.

    Returns (kind, value):
      ("level", "low"|"medium"|"high") for auto-approval thresholds
      ("prompt_all", None) for strict approval of every classified action
      ("yolo", None) for never-prompt mode
      (None, None) when unrecognized
    """
    raw = arg.lower().strip()
    if raw in ("all", "prompt", "prompt-all", "strict", "none"):
        return "prompt_all", None
    if raw == "yolo":
        return "yolo", None
    lvl = _normalize_level(raw)
    if lvl:
        return "level", lvl
    return None, None


def _approval_display():
    if YOLO:
        return "off (yolo)"
    if APPROVE_LEVEL is None:
        return "all"
    return APPROVE_LEVEL


def _apply_approval(kind, level=None, *, update_default=True):
    global YOLO, APPROVE_LEVEL, DEFAULT_APPROVE_LEVEL
    if kind == "yolo":
        YOLO = True
        APPROVE_LEVEL = None
        if update_default:
            DEFAULT_APPROVE_LEVEL = None
    elif kind == "prompt_all":
        YOLO = False
        APPROVE_LEVEL = None
        if update_default:
            DEFAULT_APPROVE_LEVEL = None
    elif kind == "level":
        YOLO = False
        APPROVE_LEVEL = level
        if update_default:
            DEFAULT_APPROVE_LEVEL = level
    else:
        return False
    return True


# Approval default resolution (highest precedence wins):
#   1. --yolo / --approval CLI flags  (explicit, per-invocation)
#   2. MINION_APPROVAL env var         (persistent, per-user — set in ~/.env)
#   3. prompt-all built-in default     (safest: prompts on every action)
#
# This lets each developer keep their own comfort level in ~/.env (e.g.
# MINION_APPROVAL=medium) without typing --approval every time, while the
# repo ships with the safest default (prompt everything) so anyone
# cloning it starts in cautious mode and can raise it with /approval or the
# env var. Accepts all|low|medium|high|yolo.
YOLO = False
APPROVE_LEVEL = None
DEFAULT_APPROVE_LEVEL = None
_env_approval = os.environ.get("MINION_APPROVAL", "").strip().lower()
if _env_approval:
    _kind, _lvl = _normalize_approval(_env_approval)
    if not _apply_approval(_kind, _lvl):
        print(f"  ✗ unknown MINION_APPROVAL={_env_approval!r} "
              f"(want all|low|medium|high|yolo); prompting for all actions")
for _i, _arg in enumerate(sys.argv):
    if _arg == "--approval" and _i + 1 < len(sys.argv):
        _kind, _lvl = _normalize_approval(sys.argv[_i + 1])
        if not _apply_approval(_kind, _lvl):
            print(f"  ✗ unknown --approval level {sys.argv[_i + 1]!r} "
                  f"(want all|low|medium|high|yolo); keeping {_approval_display()}")
if "--yolo" in sys.argv:
    YOLO = True
    APPROVE_LEVEL = None  # --yolo overrides everything; never prompt

# --- --prompt-file: non-interactive single-shot mode ------------------------
# `--prompt-file <path>` (or `-f <path>`) reads the full task instruction from
# a file, runs exactly one model turn, and exits — no REPL, no banner, no
# session save. Designed for headless benchmark harnesses (Harbor/Terminal-Bench)
# that pipe a task in and wait for the agent to finish. When set, main() runs
# _run_one_shot() instead of the interactive loop. A value of "-" reads from
# stdin instead of a file (useful for `echo ... | minion -f -`).
_PROMPT_FILE = None
for _i, _arg in enumerate(sys.argv):
    if _arg in ("--prompt-file", "-f") and _i + 1 < len(sys.argv):
        _PROMPT_FILE = sys.argv[_i + 1]
        break

# --- base-level traffic log -------------------------------------------------
# Append-only JSONL record of every byte we ship to / receive from the server.
# Lives next to this script so it's easy to find; rotate by hand if it gets big.
LOG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs", "llamacpp.log")
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
_llog = open(LOG_PATH, "a", buffering=1)  # line-buffered; flushes per write


def _log_event(direction, payload):
    """direction: 'req' (outgoing) or 'resp' (incoming SSE chunk)."""
    _llog.write(json.dumps({"ts": time.time(), "dir": direction, "data": payload}) + "\n")

# --- ANSI -------------------------------------------------------------------
DIM, CYAN, GREEN, YELLOW, RED, MAGENTA, BOLD, RESET = (
    "\033[2m", "\033[36m", "\033[32m", "\033[33m", "\033[31m", "\033[35m",
    "\033[1m", "\033[0m",
)
CLEAR_LINE = "\033[2K\r"   # erase entire line, return cursor to col 0
HIDE_CURSOR = "\033[?25l"
SHOW_CURSOR = "\033[?25h"


# --- waiting animation (tiny Conway's Game of Life) -------------------------
# A spinner is boring. A 1-row toroidal Game of Life is the same shape on screen
# (one line of cells) but actually does something — patterns glide, blinkers
# flash, gliders crawl. Runs in a background thread; the main loop kills it
# the instant the first token arrives.
_GOL_W = 24
_GOL_ALIVE = "█"
_GOL_DEAD = "·"
_GOL_GLIDER = {(0, 0), (1, 1), (2, 1), (0, 2), (1, 0)}  # 5-cell, period-4


class LifeSpinner:
    def __init__(self, width=_GOL_W, tick_ms=90, label="thinking"):
        self.w = width
        self.tick = tick_ms / 1000
        self.label = label  # shown before the cells ("thinking" / "running" / ...)
        self._stop = threading.Event()
        self._t = None

    def _seed(self):
        row = [0] * self.w
        x = random.randrange(self.w)
        for dx, _ in _GOL_GLIDER:
            row[(x + dx) % self.w] = 1
        for _ in range(2):
            x = random.randrange(self.w)
            row[x] = row[(x + 1) % self.w] = row[(x + 2) % self.w] = 1
        for _ in range(self.w // 6):
            row[random.randrange(self.w)] = 1
        return row

    def _step(self, row):
        # A 1-row GoL is degenerate (cells have only 2 neighbors). Cheat: treat
        # the row as the middle of a 3-row toroidal world where the rows above
        # and below mirror the current one. Gives every cell the standard 8
        # neighbors, so gliders/blinkers/etc. actually work.
        w, above, below, nxt = self.w, row, row, [0] * self.w
        for x in range(w):
            n = (above[(x - 1) % w] + above[x] + above[(x + 1) % w] +
                 row[(x - 1) % w]                   + row[(x + 1) % w] +
                 below[(x - 1) % w] + below[x] + below[(x + 1) % w])
            cur = row[x]
            nxt[x] = 1 if (cur and n in (2, 3)) or (not cur and n == 3) else 0
        return nxt

    def _run(self):
        sys.stdout.write(HIDE_CURSOR)
        try:
            row = self._seed()
            # initial render — also reserve the line so subsequent prints don't shift things
            sys.stdout.write(CLEAR_LINE + "  " + DIM + f"{self.label} " + RESET +
                             "".join(_GOL_ALIVE if c else _GOL_DEAD for c in row))
            sys.stdout.flush()
            while not self._stop.is_set():
                time.sleep(self.tick)
                if self._stop.is_set():
                    break
                row = self._step(row)
                sys.stdout.write(CLEAR_LINE + "  " + DIM + f"{self.label} " + RESET +
                                 "".join(_GOL_ALIVE if c else _GOL_DEAD for c in row))
                sys.stdout.flush()
        finally:
            # wipe the spinner line and restore cursor
            sys.stdout.write(CLEAR_LINE + SHOW_CURSOR)
            sys.stdout.flush()

    def start(self):
        if not sys.stdout.isatty():
            return  # headless (piped/redirected) — no spinner animation
        self._stop.clear()
        self._t = threading.Thread(target=self._run, daemon=True)
        self._t.start()

    def stop(self):
        self._stop.set()
        if self._t:
            self._t.join(timeout=0.5)
            self._t = None


# --- interrupt watcher ------------------------------------------------------
# Lets the user press Esc during model generation to stop the stream and drop
# back to the prompt. A SINGLE daemon thread, started once and parked between
# turns, watches stdin for bare Esc. The main loop checks _USER_INTERRUPTED
# between chunks and closes the stream on interrupt. Tools are NOT cancelled —
# run_bash etc. run to completion. (Hard-cancelling a tool mid-flight is a
# separate follow-up.)
#
# Why a single persistent thread instead of spawn-per-turn:
#
#   The watcher reads stdin byte-by-byte and discards anything that isn't Esc
#   (line 1175ish). That makes it a stdin *thief* while alive. An earlier
#   implementation spawned a fresh thread per turn and stopped it with
#   `watcher.join(timeout=0.5)`. A timed join is not a guaranteed kill: if the
#   thread was mid-read when the timeout elapsed, it kept running — and the
#   `finally` then `_INTERRUPT_EVENT.clear()`d the very flag the thread needs
#   to notice it should stop. The leaked thread was then immortal (its exit
#   signal had already been lowered) and sat in a tight `select`+`os.read`
#   loop, eating 50–75% of keystrokes that the chatbox was supposed to get.
#   Worse, termios is global to the tty (not per-thread), so leaked watchers
#   fighting the chatbox over `VMIN` (watcher sets 0, chatbox sets 1) could
#   end up in a *blocking* `os.read` from which they could never re-check the
#   exit flag — so the next turn's `_INTERRUPT_EVENT.set()` couldn't reach
#   them and the leaks stacked. Result: a session got progressively more
#   laggy the longer it ran, and only Ctrl+C / restart (which tears down all
#   threads) cleared it. A persistent singleton thread with a clean
#   arm/disarm protocol has no spawn/join window to leak through.
#
# Four events:
#   _INTERRUPT_ARMED    — "watcher should actively read stdin right now"
#                         main sets it to start a generation; clears it to park.
#   _INTERRUPT_EXIT     — "watcher should die entirely" (set once at process exit)
#   _USER_INTERRUPTED   — "the user actually pressed Esc" — only the watcher sets
#                         this; main reads it after disarming. Separate from the
#                         arm flag so disarming to park ≠ "you pressed Esc".
#   _INTERRUPT_RESTORED — "watcher has restored termios to cooked mode" — the
#                         watcher sets this in its inner finally after tcsetattr.
#                         main clears it before arming and waits on it after
#                         disarming so the chatbox never captures the watcher's
#                         raw-mode termios as its "old" state. Without this
#                         handshake the disarm is asynchronous (the watcher may
#                         still be in select(0.1) when the chatbox runs) and the
#                         chatbox can inherit VMIN=0/ISIG-off, breaking Enter
#                         and leaving the terminal stuck in raw mode.
_INTERRUPT_ARMED = threading.Event()
_INTERRUPT_EXIT = threading.Event()
_USER_INTERRUPTED = threading.Event()
_INTERRUPT_RESTORED = threading.Event()
_INTERRUPT_RESTORED.set()  # initially: nothing to restore (OK to proceed)
_INTERRUPT_THREAD = None


def _interrupt_watcher():
    """Persistent daemon: park while idle, read stdin for bare Esc while armed.

    When armed (during model generation) stdin is put into raw mode (ISIG off so
    Ctrl+C still kills the process) so we can read without echo. A bare Esc (not
    the start of an arrow-key / bracketed-paste / etc. CSI sequence) sets
    _USER_INTERRUPTED. When disarmed (between turns, while the chatbox owns
    stdin) we restore termios and park on _INTERRUPT_ARMED — doing NOTHING with
    stdin, so the chatbox / approval prompt get every keystroke.

    Single-reader invariant: this is the only thread that ever calls os.read on
    stdin, and only while armed. Parked = hands off stdin entirely. That's what
    stops the keystroke-eating leaks the spawn-per-turn design suffered from.
    """
    try:
        fd = sys.stdin.fileno()
    except (AttributeError, OSError, ValueError):
        return  # stdin has no fileno (pytest capture, redirect, etc.)
    if not os.isatty(fd):
        return
    try:
        old = termios.tcgetattr(fd)
    except Exception:
        return
    new = old[:]
    new[3] &= ~(termios.ECHO | termios.ICANON | termios.ISIG)
    new[0] &= ~termios.ICRNL
    new[6][termios.VMIN] = 0
    new[6][termios.VTIME] = 0
    last_fire = 0.0
    while not _INTERRUPT_EXIT.is_set():
        # Park until a generation arms us. While parked we touch stdin NOT AT
        # ALL — the chatbox owns it. This is the whole point of the redesign.
        _INTERRUPT_ARMED.wait(0.25)
        if not _INTERRUPT_ARMED.is_set() or _INTERRUPT_EXIT.is_set():
            continue
        # We're armed: take stdin into raw mode for the duration of generation.
        try:
            termios.tcsetattr(fd, termios.TCSADRAIN, new)
        except Exception:
            _INTERRUPT_RESTORED.set()  # never armed — nothing to restore
            return
        try:
            while _INTERRUPT_ARMED.is_set() and not _INTERRUPT_EXIT.is_set():
                r, _, _ = select.select([fd], [], [], 0.1)
                if not r:
                    continue
                try:
                    c = os.read(fd, 1)
                except OSError:
                    _INTERRUPT_RESTORED.set()  # dying — unblock main's wait
                    return
                if c != b"\x1b":
                    continue  # discard anything that isn't Esc
                # Could be bare Esc OR the lead byte of an escape sequence
                # (arrow keys, Home/End, bracketed paste, etc.). Wait up to 50ms
                # for more bytes; if none arrive, it's a bare Esc.
                r2, _, _ = select.select([fd], [], [], 0.05)
                if r2:
                    try:
                        os.read(fd, 1)  # swallow the rest of the sequence
                    except OSError:
                        pass
                    continue
                now = time.time()
                if now - last_fire < 0.25:  # debounce — don't fire twice in a row
                    continue
                last_fire = now
                _USER_INTERRUPTED.set()
                # Leave _INTERRUPT_ARMED set: main will clear it on disarm,
                # which exits the inner loop and parks us.
        finally:
            try:
                termios.tcsetattr(fd, termios.TCSADRAIN, old)
            except Exception:
                pass
            # Signal main that termios has been restored to cooked mode.
            # main waits on this after clearing _INTERRUPT_ARMED so the chatbox
            # never captures our raw-mode termios as its "old" state.
            _INTERRUPT_RESTORED.set()


def _ensure_interrupt_watcher():
    """Start the singleton interrupt watcher thread if it isn't already alive.

    Called lazily before the first model turn (and defensively on every turn —
    cheap because a living thread short-circuits the check). Guaranteed exactly
    one reader of stdin across the whole process lifetime.
    """
    global _INTERRUPT_THREAD
    if _INTERRUPT_THREAD is not None and _INTERRUPT_THREAD.is_alive():
        return
    _INTERRUPT_EXIT.clear()
    _INTERRUPT_ARMED.clear()
    _INTERRUPT_RESTORED.set()  # fresh thread: nothing to restore yet
    _USER_INTERRUPTED.clear()
    _INTERRUPT_THREAD = threading.Thread(
        target=_interrupt_watcher, daemon=True, name="minion-interrupt")
    _INTERRUPT_THREAD.start()


# --- tools ------------------------------------------------------------------
def read_file(path, offset=1, limit=None, **_):
    """Read a file as numbered lines (1-based, `cat -n` style: right-aligned
    line number, a tab, then the line). Large files return only a window — pass
    offset (start line) and limit (max lines) to page through the rest. A header
    announces the visible range + total line count so the model can ask for more
    instead of being head/tail-capped by _sanitize_tool_result."""
    with open(path) as f:
        lines = f.readlines()
    total = len(lines)
    try:
        start = max(1, int(offset)) - 1
    except (TypeError, ValueError):
        start = 0
    if total == 0:
        return f"[{path}: empty file]"
    if start >= total:
        return f"[{path}: {total} lines; offset {start + 1} is past end of file]"
    if limit is None:
        limit = _env_int("MINION_READ_FILE_LINES", 400)
    try:
        limit = int(limit)
    except (TypeError, ValueError):
        limit = 400
    end = total if limit <= 0 else min(total, start + limit)
    body = "".join(f"{i + 1:6d}\t{lines[i]}" for i in range(start, end))
    if body and not body.endswith("\n"):
        body += "\n"
    if start > 0 or end < total:
        return (f"[{path}: lines {start + 1}-{end} of {total}; "
                f"call read_file with offset/limit to page]\n" + body)
    return body


def write_file(path, content, **_):
    if not _confirm(f"write {path} ({len(content)} bytes)"):
        return "DENIED by user"
    with open(path, "w") as f:
        f.write(content)
    return f"wrote {len(content)} bytes to {path}"


_LINE_NUM_PREFIX = re.compile(r"^ *\d+\t")


def _strip_line_numbers(text):
    """Remove read_file's `<n>\\t` line-number prefixes from a pasted block.
    Only strips when EVERY non-empty line carries one (i.e. the whole block came
    from numbered read_file output) — so ordinary edits, and code that merely
    starts a line with digits, are never altered."""
    lines = text.split("\n")
    nonempty = [ln for ln in lines if ln.strip()]
    if nonempty and all(_LINE_NUM_PREFIX.match(ln) for ln in nonempty):
        return "\n".join(_LINE_NUM_PREFIX.sub("", ln) for ln in lines)
    return text


def edit_file(path, old, new, **_):
    with open(path) as f:
        src = f.read()
    if src.count(old) != 1:
        # The model may have pasted read_file's numbered output into `old`.
        # Only as a fallback (exact match already failed), retry with the
        # numbering stripped — and strip `new` the same way so we never write
        # line numbers into the file. Normal edits hit the exact match above
        # and never reach here, so genuine content is never touched.
        stripped_old = _strip_line_numbers(old)
        if stripped_old != old and src.count(stripped_old) == 1:
            old, new = stripped_old, _strip_line_numbers(new)
    if src.count(old) != 1:
        return f"ERROR: `old` matched {src.count(old)} times (need exactly 1)"
    if not _confirm(f"edit {path}"):
        return "DENIED by user"
    with open(path, "w") as f:
        f.write(src.replace(old, new))
    return f"edited {path}"


def list_dir(path=".", **_):
    return "\n".join(sorted(os.listdir(path)))


def _run_bash(command):
    """Launch `command` detached in its own session and return (pid, log_path).

    The command is started with ``setsid`` so it runs in a **new session / new
    process group**, fully detached from minion's controlling terminal. This is
    important: with plain ``nohup`` the background job stays in minion's
    process group, so a terminal Ctrl+C (SIGINT to the foreground group) kills
    it — exactly the failure mode that kept canceling long ``docker pull``s
    mid-extraction. ``nohup`` only ignores SIGHUP, not SIGINT; ``setsid`` puts
    the job beyond the reach of terminal-generated signals entirely.

    Exit-code capture is inlined into a single ``sh -c`` invocation so no
    wrapper script is needed — avoids a race where the launcher starts after
    the temp file is already deleted.

    Stores a ``bg-pid-{pid}.map`` file (JSON ``{"log_path": "…"}``) so
    ``wait_background`` can locate logs by PID later.
    """
    _delete_old_bg_logs()
    bg_dir = os.path.join(SESSION_HOME, "bg-logs")
    os.makedirs(bg_dir, exist_ok=True)
    stamp = time.strftime("%Y%m%d-%H%M%S")
    suffix = f"{stamp}-{secrets.token_hex(3)}"
    log_path = os.path.join(bg_dir, f"bg-{suffix}.log")
    # Inline: run the command in a subshell, then echo its exit status.
    # setsid detaches into a new session so terminal Ctrl+C / SIGINT can't
    # reach the background job. Redirect to the log file ourselves.
    inner = f"({command}); echo \"[exit: $?]\""
    wrapped = (f"setsid sh -c {shlex.quote(inner)} > {shlex.quote(log_path)} 2>&1 < /dev/null &\n"
               f"echo $!")
    r = subprocess.run(wrapped, shell=True, capture_output=True, text=True)
    pid = r.stdout.strip()
    # Write a tiny pid→logpath map file so wait_background can find it.
    try:
        map_path = os.path.join(bg_dir, f"bg-pid-{pid}.map")
        with open(map_path, "w") as f:
            json.dump({"log_path": log_path}, f)
    except OSError:
        pass
    return pid, log_path


def _delete_old_bg_logs(max_logs=50, max_age_days=7):
    """Trim background logs so they don't pile up forever.

    Also removes stale ``bg-pid-*.map`` files whose target log no longer exists.
    """
    bg_dir = os.path.join(SESSION_HOME, "bg-logs")
    try:
        log_entries = []
        map_entries = []
        for name in os.listdir(bg_dir):
            if not name.startswith("bg-"):
                continue
            p = os.path.join(bg_dir, name)
            try:
                if name.endswith(".map"):
                    map_entries.append((os.path.getmtime(p), p, name))
                else:
                    log_entries.append((os.path.getmtime(p), p, name))
            except OSError:
                continue
        now = time.time()
        # remove logs by age
        for mtime, p, _ in log_entries:
            if (now - mtime) > max_age_days * 86400:
                try:
                    os.remove(p)
                except OSError:
                    pass
        # then cap log count (keep newest)
        log_entries.sort(key=lambda x: x[0], reverse=True)
        for _, p, _ in log_entries[max_logs:]:
            try:
                os.remove(p)
            except OSError:
                pass
        # Clean stale map files (no corresponding log) and age-prune maps too
        for mtime, p, _ in map_entries:
            try:
                with open(p) as f:
                    data = json.load(f)
                target = data.get("log_path", "")
                should_del = not os.path.exists(target)
            except (OSError, json.JSONDecodeError):
                should_del = True
            if not should_del and (now - mtime) > max_age_days * 86400:
                should_del = True
            if should_del:
                try:
                    os.remove(p)
                except OSError:
                    pass
    except OSError:
        pass


def _poll_pid(pid, timeout, check_esc=False):
    """Poll *pid* up to *timeout* seconds.

    Returns:
        "exited"      — process finished
        "timeout"     — poll window expired, process still alive
        "interrupted" — Esc was pressed (only if *check_esc* is True)

    Uses ``os.kill(pid, 0)`` (a kernel-level signal-0 probe) instead of
    shelling out to the external ``kill`` binary.  Some minimal Docker
    images (Alpine/busybox without ``procps``) don't ship a ``kill``
    executable in PATH, which caused *every* ``run_bash`` call to raise
    ``FileNotFoundError`` — the agent could run no shell commands at all.
    ``os.kill`` talks to the kernel directly, so it works in any
    environment with a POSIX layer, no external binary required.

    Signal-0 semantics:
      • returns normally    → process exists and is signalable → alive
      • ProcessLookupError  → ESRCH, no such process            → exited
      • PermissionError     → EPERM, process exists but not ours → alive
      • ValueError          → pid wasn't a valid int             → exited
    """
    deadline = time.time() + timeout
    try:
        pid_i = int(pid)
    except (TypeError, ValueError):
        return "exited"  # no valid pid → the command never started
    while time.time() < deadline:
        try:
            os.kill(pid_i, 0)
        except ProcessLookupError:
            return "exited"
        except PermissionError:
            pass  # process exists but owned by another uid — still alive
        except OSError:
            return "exited"  # unexpected (e.g. EINTR); bail to be safe
        if check_esc and _USER_INTERRUPTED.is_set():
            return "interrupted"
        time.sleep(0.1)
    return "timeout"


def _read_log(log_path):
    """Read *log_path*, parse the trailing ``[exit: N]`` marker, return (content, exit_code)."""
    try:
        with open(log_path, "rb") as f:
            out = f.read().decode("utf-8", errors="replace")
    except OSError:
        return "", None
    exit_code = None
    last_nl = out.rstrip("\n").rfind("\n")
    trailer = out[last_nl + 1:].strip() if last_nl != -1 else out.strip()
    m = re.match(r"\[exit:\s*(-?\d+)]", trailer)
    if m:
        exit_code = int(m.group(1))
        out = out[:last_nl] if last_nl != -1 else ""
    return out, exit_code


def _infer_timeout_from_sleep(command, default):
    """If *command* contains ``sleep N`` (with N in seconds), return a poll
    timeout that's ``max(N) + 10``, so the command finishes synchronously
    without the model having to remember ``timeout=N``.

    This compensates for the natural habit of writing
    ``run_bash("sleep 30 && cat …")`` — a pattern that *seems* like it should
    block (the shell waits, then cats) but is itself launched via nohup, so
    it gets backgrounded after ~3 s just like anything else.  By recognising
    the sleep we make the "obvious" pattern work again.
    """
    import re
    max_sleep = 0
    for m in re.finditer(r'(?:^|[;&|]\s*)sleep\s+(\d+)', command):
        n = int(m.group(1))
        if n > max_sleep:
            max_sleep = n
    if max_sleep > 0:
        return max_sleep + 10
    return default


def run_bash(command, timeout=None, **_):
    """Run a shell command.

    Commands always run via nohup — they never block the agent loop.
    If the command finishes within a short poll window (~3 s by default),
    output is returned directly and the log file is deleted.  If it's still
    running after the poll window, it continues in the background and you
    get a PID + log path to check later with ``read_file``.

    **Do NOT use** ``sleep N && …`` inside the command string as a way to
    wait — the entire command is launched via nohup so *it* will also be
    backgrounded after a few seconds.  Instead, pass ``timeout=N`` to tell
    minion to wait synchronously (e.g. ``timeout=35`` for a ``sleep 30``
    command).  As a convenience, if your command *does* start with
    ``sleep N …`` the tool auto-extends its poll window to ``N+10`` s
    so the "obvious" pattern still works.

    Use ``timeout=0`` to wait *indefinitely* (the tool blocks until the
    command completes).  To wait for an **already-backgrounded** command,
    use ``wait_background(pid=…)``.

    Long-running commands (servers, watchers, builds) produce background logs
    at ``~/.minion/bg-logs/``, automatically cleaned up (max ~50 files /
    7 days).  Pressing Esc during the poll window immediately backgrounds
    the command.
    """
    if not _confirm(f"run: {command}"):
        return "DENIED by user"
    if timeout is None:
        timeout = _env_int("MINION_BASH_POLL_SECONDS", 3)
        timeout = _infer_timeout_from_sleep(command, timeout)
    if timeout <= 0:
        timeout = 999999999  # ∞ — wait indefinitely
    pid, log_path = _run_bash(command)
    status = _poll_pid(pid, timeout, check_esc=True)
    if status == "interrupted":
        _log_event("tool", f"run_bash interrupted by user while polling (pid {pid})")
        return (f"[background] PID {pid} | log: {log_path}\n"
                "The command was interrupted (Esc pressed) and is now running "
                "in the background. "
                f"Check output with read_file('{os.path.abspath(log_path)}')")
    if status == "timeout":
        return (f"[background] PID {pid} | log: {log_path}\n"
                "The command is still running in the background. "
                f"Check output with read_file('{os.path.abspath(log_path)}')")
    time.sleep(0.2)  # brief settle so the log is flushed
    out, exit_code = _read_log(log_path)
    # Delete the log — quick commands leave no trace.
    try:
        os.remove(log_path)
    except OSError:
        pass
    if exit_code is not None:
        return f"[exit {exit_code}]\n{out}"
    return f"[exited]\n{out}"


def wait_background(pid, log_path=None, timeout=0, **_):
    """Wait for a backgrounded command (started by ``run_bash``) to finish.

    Polls *pid* until it exits (default: wait indefinitely).  When the command
    exits the output is returned and the log file is cleaned up.  Esc can be
    pressed at any time to interrupt the wait — the command keeps running and
    the log path is returned so you can check back later.

    Pass ``timeout=N`` only if you specifically want to bail after N seconds
    (rarely useful — you called this because you want to *wait*, not guess a
    ceiling).

    Parameters
    ----------
    pid : int or str
        The PID returned by the prior ``run_bash`` call.
    log_path : str, optional
        The log path from the background message.  If omitted the tool scans
        ``~/.minion/bg-logs/`` for a log whose PID matches.
    timeout : int
        Max seconds to wait (default 0 = wait indefinitely until the command
        finishes or Esc is pressed).
    """
    pid = int(pid)
    if not log_path:
        log_path = _find_bg_log_for_pid(pid)
        if not log_path:
            return (f"[error] No background log found for PID {pid}. "
                    "The log may have been cleaned up already.")
    timeout_s = 999999999 if timeout <= 0 else timeout
    status = _poll_pid(pid, timeout_s, check_esc=True)
    if status == "interrupted":
        return (f"[interrupted] PID {pid} still running. "
                f"Check later with read_file('{os.path.abspath(log_path)}')")
    if status == "timeout":
        return (f"[timeout after {timeout}s] PID {pid} still running. "
                f"Check output so far with read_file('{os.path.abspath(log_path)}')")
    time.sleep(0.2)
    out, exit_code = _read_log(log_path)
    try:
        os.remove(log_path)
    except OSError:
        pass
    if exit_code is not None:
        return f"[exit {exit_code}]\n{out}"
    return f"[exited]\n{out}"


def _find_bg_log_for_pid(pid):
    """Look up the log path for a background PID via the pid→log map files
    written by ``_run_bash``.  Returns the log path, or None if not found."""
    pid = str(pid)
    bg_dir = os.path.join(SESSION_HOME, "bg-logs")
    map_path = os.path.join(bg_dir, f"bg-pid-{pid}.map")
    try:
        with open(map_path) as f:
            data = json.load(f)
        return data.get("log_path")
    except (OSError, IOError, json.JSONDecodeError):
        pass
    # Fallback: try to find a log file that mentions this PID in the first 1K
    try:
        for name in os.listdir(bg_dir):
            if not name.startswith("bg-") or ".map" in name:
                continue
            p = os.path.join(bg_dir, name)
            try:
                with open(p, "rb") as f:
                    head = f.read(1024).decode("utf-8", errors="replace")
                if pid in head:
                    return p
            except OSError:
                continue
    except OSError:
        pass
    return None


DISPATCH = {
    "read_file": read_file, "write_file": write_file, "edit_file": edit_file,
    "list_dir": list_dir, "run_bash": run_bash, "wait_background": wait_background,
}

TOOLS = [
    {"type": "function", "function": {"name": "read_file", "description": "Read a file's contents. Returns lines numbered (1-based, like `cat -n`: a right-aligned number, a tab, then the line). Large files return only a window — pass `offset` (1-based start line) and `limit` (max lines, default 400) to page through the rest; a header shows the visible range and total line count.",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "offset": {"type": "integer", "description": "1-based line to start from (default 1)"}, "limit": {"type": "integer", "description": "max lines to return (default 400; <=0 reads to end)"}}, "required": ["path"]}}},
    {"type": "function", "function": {"name": "write_file", "description": "Write (overwrite) a file",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}},
    {"type": "function", "function": {"name": "edit_file", "description": "Replace one exact occurrence of `old` with `new` in a file. Prefer raw file text; if you paste read_file's numbered lines, the `<n>\\t` line-number prefixes are stripped automatically.",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "old": {"type": "string"}, "new": {"type": "string"}}, "required": ["path", "old", "new"]}}},
    {"type": "function", "function": {"name": "list_dir", "description": "List a directory",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}}}}},
    {"type": "function", "function": {"name": "run_bash", "description": "Run a shell command. Commands are launched detached (setsid — their own session/process group, immune to terminal Ctrl+C) and never block the agent loop. If the command finishes within a short poll window (~3 s by default), output is returned directly; otherwise it continues in the background and you get a PID + log path to check later with read_file() or wait_background(). Do NOT use `sleep N` as a chaining trick inside the command string — it'll background too. For long commands (docker pull, builds, servers), DON'T pass a big timeout — just let it background and then call wait_background(pid=...) which waits indefinitely. Only pass `timeout=N` for commands you KNOW finish in ~N seconds (e.g. timeout=35 for `sleep 30`). As a convenience, if your command does start with `sleep N ...` the tool auto-extends the poll window to N+10 s. Background logs at ~/.minion/bg-logs/, auto-cleaned (max ~50 files / 7 days).",
        "parameters": {"type": "object", "properties": {"command": {"type": "string"}, "timeout": {"type": "integer", "description": "Seconds to wait for the command to finish synchronously (default unset = ~3 s poll; 0 = wait indefinitely). Only set this when you KNOW the command finishes in ~N seconds — for unknown-duration commands, let it background and use wait_background instead."}}, "required": ["command"]}}},
    {"type": "function", "function": {"name": "wait_background", "description": "Wait for a backgrounded command (previously started by run_bash) to finish and return its output. By default waits indefinitely until the command completes or Esc is pressed (which leaves it running in the background — Esc cancels the WAIT, never the process). Pass `timeout=N` only if you specifically want to bail after N seconds — but usually you should just let it wait, since you can always Esc out. Locate the log via the `pid` (auto-located) or pass `log_path` explicitly. Note: this tool only polls the process (kill -0); it never sends a signal to the background command, so waiting cannot kill it.",
        "parameters": {"type": "object", "properties": {"pid": {"type": "integer", "description": "PID of the background command to wait for"}, "log_path": {"type": "string", "description": "Log file path from the background message (optional — auto-located by PID if omitted)"}, "timeout": {"type": "integer", "description": "Max seconds to wait (default 0 = wait indefinitely; almost never set this — Esc works as the cancel)"}}, "required": ["pid"]}}},
]

FINAL_ANSWER_TOOL = {"type": "function", "function": {
    "name": "final_answer",
    "description": "Return a complete, concise visible answer to the user. Use this when no more tool calls are needed.",
    "parameters": {
        "type": "object",
        "properties": {
            "answer": {
                "type": "string",
                "description": "The complete concise answer to show to the user. Keep it short and do not trail off.",
            },
            "status": {"type": "string", "enum": ["answered", "blocked"]},
        },
        "required": ["answer"],
    },
}}
FINAL_ANSWER_TOOL_CHOICE = {
    "type": "function",
    "function": {"name": "final_answer"},
}

SYSTEM = """You are a terminal coding agent working in the user's current directory.
Use the provided tools to inspect and modify code. Take one concrete step at a time.

If your runtime does NOT support native tool calls, emit a standalone text-protocol call exactly like:
[minion_tool_call]{"name": "read_file", "arguments": {"path": "foo.py"}}[/minion_tool_call]
Emit nothing before or after a tool call; wait for the Observation. When showing a tool-call example, use prose or a code block so it is not a standalone tool call. When the task is done, reply in plain prose.

For shell commands, you never need to worry about blocking — commands run detached (setsid) automatically. Quick ones (a few seconds) return output directly; long-running ones (servers, builds, docker pulls) get backgrounded and you'll get the PID + log path. **Do NOT use `sleep N && …`** as a chaining trick — the whole command is detached so it'll background too. For long commands of unknown duration (docker pull, builds), let them background and then call `wait_background(pid=...)` — it waits indefinitely until the command finishes, and you (or the user) can Esc out at any time (Esc cancels the WAIT, never the process). Only pass `timeout=N` to run_bash when you KNOW the command finishes in ~N seconds (e.g. `timeout=35` for `sleep 30`). Never guess a big timeout for an unknown-duration command.

Operating principles for long background commands (docker pull/build, git clone of big repos, torch.compile, CUDA graph builds):
- A process that has produced no new log output is NOT necessarily dead. Docker's layer-extraction phase, AOT compilation, and many CUDA/graph-cache builds are legitimately silent for minutes. Distinguish "no output" from "no progress": check `/proc/<pid>/io` read/write bytes, `du -sh` of the target dir, or CPU% via `ps` BEFORE declaring a hang.
- NEVER `pkill -f '<the thing we just started>'` as a "clean restart". That destroys a job that may be 99% done and restarts it from the same wall. If a backgrounded long command appears stuck, first PROVE it's stuck (ps, /proc/<pid>/io, du), and only kill it if it's genuinely hung.
- Treat exit 130 (SIGINT) / `context canceled` as a signal-of-a-signal, not a command failure. When a backgrounded command dies with 130, the first hypothesis should be "something interrupted it," not "the command failed." Check the command's own logs / journalctl for `context canceled` before retrying.
- For long silent-tail commands (docker pull, docker build, big git clone), prefer the background-and-wait pattern: `run_bash(cmd)` → let it background → `wait_background(pid)`. Don't hammer it with sleep-poll loops."""


def _env_int(name, default):
    try:
        return int(os.environ.get(name, str(default)))
    except ValueError:
        return default


def _env_float(name, default):
    try:
        return float(os.environ.get(name, str(default)))
    except ValueError:
        return default


# --- usage metrics (default off) --------------------------------------------
# minion already captures every model call's token usage (the streaming
# `usage` object from OpenAI/Z.ai, plus llama.cpp's `timings`) to print the
# stats footer — and then throws it away. This module retains those numbers
# so they can be written into the session JSON (always — it's free, the data
# is already in hand) and optionally pushed to an external metrics endpoint
# (only when MINION_METRICS_URL is set). The endpoint is generic: minion
# posts a small JSON blob of cumulative session totals after each turn. One
# such consumer is a personal dashboard; the format is dashboard-agnostic.
#
# Accounting matches the OpenAI usage convention (and Hermes's canonical
# usage): `input_tokens` EXCLUDES cached tokens (cache is broken out into
# `cache_read_tokens`), and `output_tokens` EXCLUDES reasoning tokens
# (reasoning is broken out into `reasoning_tokens`). Dashboards that want a
# "total in" sum input + cache_read; a "total out" sum output + reasoning.

def _normalize_usage(usage, timings, fallback_chars=0):
    """Pull a {input, output, cache_read, reasoning} token dict out of a
    finished model call, from whichever the server gave us: the standard
    streaming `usage` object (OpenAI/Z.ai) or the llama.cpp `timings` extra
    (where `usage` is null). Returns None if neither has anything to say.

    Numbers are the per-call totals reported by the server, not deltas —
    callers accumulate them. The split is:

      input_tokens        prompt tokens, MINUS cached (matches OpenAI's
                          prompt_tokens minus prompt_tokens_details.cached)
      cache_read_tokens   cached prompt tokens (prompt_tokens_details.cached
                          on OpenAI; cache_n on llama.cpp)
      output_tokens       completion tokens, MINUS reasoning
      reasoning_tokens    completion_tokens_details.reasoning_tokens, if any
                          (reasoning models: MiniMax-M3, DeepSeek-R1, …)

    When neither `usage` nor `timings` is available, a rough estimate is
    derived from `fallback_chars` (client-side character count of streamed
    content + tool-call arguments). This ensures backends like vLLM without
    `stream_options` support still report token counts to agentdash.
    """
    # llama.cpp: no `usage`, but a `timings` object on the final chunk with
    # prompt_n / predicted_n / cache_n. There's no reasoning split here —
    # predicted_n is all generated tokens.
    if (not usage) and timings and timings.get("predicted_n"):
        prompt_n = int(timings.get("prompt_n") or 0)
        cache_n = int(timings.get("cache_n") or 0)
        gen_n = int(timings.get("predicted_n") or 0)
        return {
            "input_tokens": max(0, prompt_n - cache_n),
            "output_tokens": gen_n,
            "cache_read_tokens": cache_n,
            "reasoning_tokens": 0,
        }
    if not usage:
        # Fallback: estimate from client-side character count (~4 chars/token)
        if fallback_chars > 0:
            estimated = max(1, fallback_chars // 4)
            return {
                "input_tokens": 0,
                "output_tokens": estimated,
                "cache_read_tokens": 0,
                "reasoning_tokens": 0,
            }
        return None
    prompt_total = int(getattr(usage, "prompt_tokens", 0) or 0)
    completion_total = int(getattr(usage, "completion_tokens", 0) or 0)
    # OpenAI/Z.ai: prompt_tokens_details.cached_tokens is the cache portion.
    ptd = getattr(usage, "prompt_tokens_details", None)
    cache_n = int(getattr(ptd, "cached_tokens", 0) or 0) if ptd else 0
    # Reasoning models break completion into content + reasoning.
    otd = getattr(usage, "output_tokens_details", None)
    reasoning_n = int(getattr(otd, "reasoning_tokens", 0) or 0) if otd else 0
    return {
        "input_tokens": max(0, prompt_total - cache_n),
        "output_tokens": max(0, completion_total - reasoning_n),
        "cache_read_tokens": cache_n,
        "reasoning_tokens": reasoning_n,
    }


class _Metrics:
    """Cumulative token usage for the current session, plus an optional push.

    A single module-global instance, `_METRICS`, is created in main() when a
    session starts (and recreated on /reset, reloaded on /resume). When
    MINION_METRICS_URL is unset it's just an in-memory accumulator that feeds
    the session-JSON totals — no network, no threads, no dependency. When the
    URL is set, each flush() also fires off a fire-and-forget POST so a live
    dashboard can plot tok/min. The push is best-effort: it disables itself
    after the first failure so a missing dashboard never slows the REPL.
    """

    def __init__(self, session_id, model=None, source=None, started_at=None,
                 push_url=None):
        self.session_id = session_id
        self.model = model
        self.source = source
        self.started_at = started_at or time.time()
        self.input_tokens = 0
        self.output_tokens = 0
        self.cache_read_tokens = 0
        self.reasoning_tokens = 0
        self.api_calls = 0
        self.push_url = push_url
        self._push_disabled = False   # latched off after the first failure

    def add(self, delta):
        """Accumulate a per-call usage dict from _normalize_usage()."""
        if not delta:
            return
        self.input_tokens += int(delta.get("input_tokens") or 0)
        self.output_tokens += int(delta.get("output_tokens") or 0)
        self.cache_read_tokens += int(delta.get("cache_read_tokens") or 0)
        self.reasoning_tokens += int(delta.get("reasoning_tokens") or 0)
        self.api_calls += 1

    def as_meta(self):
        """The fields merged into the session JSON on save."""
        return {
            "input_tokens": self.input_tokens,
            "output_tokens": self.output_tokens,
            "cache_read_tokens": self.cache_read_tokens,
            "reasoning_tokens": self.reasoning_tokens,
            "api_calls": self.api_calls,
            "started_at": self.started_at,
        }

    def flush(self):
        """Fire the optional HTTP push. Cumulative totals, so the endpoint
        can compute its own deltas (agentdash's /api/tokens/push does)."""
        if not self.push_url or self._push_disabled:
            return
        payload = {
            "session_id": self.session_id,
            "model": self.model,
            "source": self.source,
            "input_tokens": self.input_tokens,
            "output_tokens": self.output_tokens,
            "cache_read_tokens": self.cache_read_tokens,
            "reasoning_tokens": self.reasoning_tokens,
            "api_calls": self.api_calls,
            "started_at": self.started_at,
            "ended_at": time.time(),
        }
        try:
            data = json.dumps(payload).encode("utf-8")
            req = urllib.request.Request(
                self.push_url, data=data,
                headers={"Content-Type": "application/json"})
            urllib.request.urlopen(req, timeout=1.5)
        except Exception:
            # Dashboard down / wrong port / not running — stop trying so we
            # never add latency to the REPL on every turn.
            self._push_disabled = True


# Module-global recorder. None until main() builds one for the session.
# It's threaded down into the model-turn path so each API call's usage is
# accumulated; the REPL merges its .as_meta() into the session JSON on save.
_METRICS = None


def _env_metrics_url():
    """MINION_METRICS_URL — optional endpoint to receive cumulative token
    totals after each turn. Default unset (metrics stay local to the session
    JSON, nothing leaves the machine). No scheme/host validation beyond
    "looks like a URL" — a bad value just latches the push off on first try."""
    url = os.environ.get("MINION_METRICS_URL", "").strip()
    return url or None


def _abbr(n):
    """Compact token count for the stats footer: 832 → '832', 1500 → '1.5K',
    78825 → '78K', 1234567 → '1.2M'. Keeps the footer tidy once context grows."""
    n = int(n)
    if n < 1000:
        return str(n)
    if n < 10_000:
        return f"{n / 1000:.1f}K"
    if n < 1_000_000:
        return f"{n // 1000}K"
    if n < 10_000_000:
        return f"{n / 1_000_000:.1f}M"
    return f"{n // 1_000_000}M"


def _precise_abbr(n):
    """Like _abbr but keeps two decimals in the K/M/B ranges.

    Used for the live tool-call-args counter ("generating tool call write_file
    args 25.15K chars"), where watching the number tick up as args stream in
    is part of the feedback — so we don't want to round 25,152 down to '25K'.
    832 → '832', 25152 → '25.15K', 1234567 → '1.23M'."""
    n = int(n)
    if n < 1000:
        return str(n)
    if n < 1_000_000:
        return f"{n / 1000:.2f}K"
    if n < 1_000_000_000:
        return f"{n / 1_000_000:.2f}M"
    return f"{n / 1_000_000_000:.2f}B"

REASONING_ONLY_CHAR_LIMIT = _env_int("MINION_REASONING_ONLY_CHARS", 36000)
# In addition to the char limit, cut a reasoning-only stream if it exceeds
# this many *seconds* of wall-clock time.  The char limit (36K) assumes a
# healthy generation speed; at slow speeds (6 tok/s on some local models) the
# model can burn the entire timeout in a single reasoning spiral without ever
# reaching the char limit.  120s gives ample room for genuine deep thinking
# while preventing a single reasoning block from eating more than ~2 minutes.
# Set to 0 to disable the time-based check (char limit only).
REASONING_ONLY_TIME_LIMIT = _env_int("MINION_REASONING_ONLY_TIME", 120)
# When a reasoning-only turn stalls, capture the reasoning_content and feed it
# back to the model as an assistant message so it can *continue* from where it
# left off rather than starting from scratch.  This caps how many chars of
# reasoning we store (the rest is truncated to the tail, which usually contains
# the model's most recent conclusions / draft work).  0 disables capture
# (reverts to the old behavior: discard reasoning, nudge with no context).
REASONING_CAPTURE_CHARS = _env_int("MINION_REASONING_CAPTURE_CHARS", 8000)
MAX_COMPLETION_TOKENS = _env_int("MINION_MAX_TOKENS", 16000)
# Cap + dedup tool results before they enter the message history. Large,
# near-duplicate tool outputs (find/ls/grep path dumps) are the fuel for the
# context-copying repetition collapse some local models fall into — bounding
# them is the cheapest prevention. 0 disables the size cap (dedup still runs).
TOOL_RESULT_CHARS = _env_int("MINION_TOOL_RESULT_CHARS", 20000)
# Backend hint.  Set MINION_BACKEND=vllm to disable llama.cpp-only recovery
# knobs (min_p, repeat_penalty, DRY) that vLLM's speculative decoder rejects.
RECOVERY_BACKEND = os.environ.get("MINION_BACKEND", "").strip().lower()

# Recovery sampler. A gibberish/repetition collapse is a LOW-ENTROPY attractor:
# the model is copying long n-grams already in context (paths, flags). Escaping
# it needs MORE entropy (raise temperature, lower min_p) plus anti-repetition
# penalties that lower the looping token's logit — the OPPOSITE of sharpening
# toward greedy. So recovery now RAISES temperature (was 0.2, which made the
# loop more deterministic) and adds DRY + repeat_penalty. The non-OpenAI knobs
# ride in extra_body (llama.cpp reads them; other backends ignore unknown keys).
RECOVERY_TEMPERATURE = _env_float("MINION_RECOVERY_TEMPERATURE", 1.0)
RECOVERY_TOP_P = _env_float("MINION_RECOVERY_TOP_P", 0.95)
RECOVERY_MIN_P = None if RECOVERY_BACKEND == "vllm" else _env_float("MINION_RECOVERY_MIN_P", 0.02)
RECOVERY_REPEAT_PENALTY = None if RECOVERY_BACKEND == "vllm" else _env_float("MINION_RECOVERY_REPEAT_PENALTY", 1.2)
RECOVERY_REPEAT_LAST_N = None if RECOVERY_BACKEND == "vllm" else _env_int("MINION_RECOVERY_REPEAT_LAST_N", 512)
RECOVERY_DRY_MULTIPLIER = None if RECOVERY_BACKEND == "vllm" else _env_float("MINION_RECOVERY_DRY_MULTIPLIER", 0.8)
RECOVERY_DRY_BASE = None if RECOVERY_BACKEND == "vllm" else _env_float("MINION_RECOVERY_DRY_BASE", 1.75)
RECOVERY_DRY_ALLOWED_LENGTH = None if RECOVERY_BACKEND == "vllm" else _env_int("MINION_RECOVERY_DRY_ALLOWED_LENGTH", 2)
# Treat path/code punctuation as DRY sequence breakers so a long file path the
# agent must emit verbatim is never penalized as "repetition".
RECOVERY_DRY_SEQUENCE_BREAKERS = None if RECOVERY_BACKEND == "vllm" else ["\n", ":", "\"", "*", "/", "\\", "`", "'"]

# --- leaked special-token collapse ------------------------------------------
# Some models (notably DeepSeek-V4 served via vLLM with dspark speculative
# decoding) will, when destabilized (e.g. by a harness error like a missing
# `kill` binary), collapse into spamming their tokenizer-internal special
# tokens.  The clearest signal is the DeepSeek begin/end-of-sentence token
# with a FULL-WIDTH pipe (｜, U+FF5C) and SentencePiece marker (▁):
#   <｜begin▁of▁sentence｜><｜begin▁of▁sentence｜>...
# These NEVER appear in legitimate model output — they are the tokenizer's
# internal representation, not a string the model would legitimately emit or
# discuss.  (Contrast with half-width "<|endoftext|>" which GPT-2 tasks may
# legitimately reference.)  Once leaked, the spam enters the assistant message
# history and on the next turn the model sees its own gibberish in context,
# reinforcing the collapse and often drifting into regurgitated training data
# (Chinese-language passages are a common artifact).

# Tokenizer-internal markers that are ALWAYS leaks — never legitimate output.
# Full-width pipe (｜ = U+FF5C) + SentencePiece ▁ = U+2581.
_LEAKED_SPECIAL_TOKEN_RE = re.compile(
    r"<\｜(?:begin▁of▁sentence|end▁of▁sentence|tool▁calls▁begin|tool▁calls▁end"
    r"|tool▁call▁begin|tool▁call▁end|reasoning▁begin|reasoning▁end)\｜>"
)
# When a leak run is detected in-stream, cut after this many leaked chunks.
LEAKED_TOKEN_RUN_LIMIT = _env_int("MINION_LEAKED_TOKEN_RUN_LIMIT", 3)
# Also detect the degeneration via repeated identical content chunks.  If the
# last N streamed chunks are byte-identical AND short (<=40 chars), the model
# is in a repetition collapse — cut the stream and recover.
REPEAT_CHUNK_LIMIT = _env_int("MINION_REPEAT_CHUNK_LIMIT", 6)


def _strip_leaked_tokens(text):
    """Remove tokenizer-internal special tokens that leaked into model output.

    DeepSeek's ``<｜begin▁of▁sentence｜>`` (full-width pipe) is a tokenizer
    artifact that should never appear in content; it signals the model has
    degenerated into a special-token repetition collapse.  Stripping it
    (a) keeps the user-visible output clean, and crucially (b) prevents the
    spam from entering the assistant message history where the model would
    see it on the next turn and have the collapse reinforced.
    """
    if not text:
        return text
    return _LEAKED_SPECIAL_TOKEN_RE.sub("", text)


def _is_leak_chunk(chunk):
    """True if a streamed content/reasoning chunk is (mostly) leaked special
    tokens — i.e. the model has entered the special-token collapse."""
    if not chunk:
        return False
    stripped = _LEAKED_SPECIAL_TOKEN_RE.sub("", chunk)
    # If removing the leaked tokens leaves almost nothing, this chunk was spam.
    return len(stripped.strip()) <= 2 and bool(_LEAKED_SPECIAL_TOKEN_RE.search(chunk))

# Default sampling params for normal (non-recovery) completions.
# Set these to get deterministic/greedy output: MINION_TEMPERATURE=0
# MINION_TOP_P is only relevant when temperature > 0; set to 1 for full
# probability mass, or lower for nucleus sampling.
# Set either to a negative value to omit it from the request (let the
# server use its own default, typically temperature=1.0 / top_p=1.0).
RISK_CONNECTION_RETRIES = _env_int("MINION_RISK_RETRIES", 3)
RISK_CONNECTION_RETRY_SECONDS = _env_int("MINION_RISK_RETRY_SECONDS", 1)
# Resolved here (not at the sessions-section top) because _env_int() is
# defined just above. See the comment at _DESC_REFRESH_DEFAULT for behavior.
SESSION_DESC_REFRESH = _env_int("MINION_SESSION_DESC_REFRESH", _DESC_REFRESH_DEFAULT)
DESC_SYSTEM = (
    "You summarize coding-agent conversations into one short line. "
    "Given the transcript, reply with ONLY a single line of at most 70 "
    "characters describing what this session is about / working on — the "
    "current task, key files, or goal. No preamble, no quotes, no trailing "
    "punctuation. Write in the same language as the conversation."
)
REASONING_ONLY_RETRY_LIMIT = _env_int("MINION_REASONING_ONLY_RETRIES", 3)
MALFORMED_STREAM_RETRY_LIMIT = _env_int("MINION_MALFORMED_STREAM_RETRIES", 2)
FORCED_FINAL_MAX_TOKENS = _env_int("MINION_FORCED_FINAL_MAX_TOKENS", 2048)
# An empty assistant turn (no content, no tool calls) is a silent stall: the model
# produced nothing and the loop would otherwise return TURN_DONE, dropping the user
# to the chat prompt mid-task. That's confusing — it looks like the agent finished
# when it actually just went mute (common with quantized models on a short/empty
# tool result, where the model "knows" the answer but emits no tokens). Auto-nudge
# it back into producing either a tool call or a visible answer. 0 disables the
# auto-continue (an empty turn then drops to chat as before).
EMPTY_TURN_RETRY_LIMIT = _env_int("MINION_EMPTY_TURN_RETRIES", 3)
RUNTIME_NOTE_RE = re.compile(r"\n\n\[Runtime note: .*?\]\s*$", re.DOTALL)

# --- auto context-compression -----------------------------------------------
# When the context window fills past this fraction (default 85%), the next
# settled model turn silently compresses the older history so the chat can
# keep going instead of running out of room. Unlike manual /compress (which
# keeps only COMPRESS_KEEP=2 turns), auto-compress is conservative: it keeps
# more recent turns so in-progress work isn't disturbed. Set to 0 to disable.
AUTOCOMPRESS_PERCENT = _env_int("MINION_AUTOCOMPRESS_PERCENT", 85)
AUTOCOMPRESS_PERCENT = max(0, min(100, AUTOCOMPRESS_PERCENT))  # clamp 0–100
# What `/autocompress on` restores; falls back to 85 when configured off.
AUTOCOMPRESS_DEFAULT = AUTOCOMPRESS_PERCENT or 85

# When the model stalls in reasoning-only (thinks and thinks but never
# produces a tool call or visible text), the first nudge should push it to
# ACT — use tools, write code, run commands — rather than to produce a
# premature "final answer".  This is especially critical for coding-agent
# tasks where the model must manipulate the filesystem, not narrate.
# Only on the LAST retry do we ask for a final answer (forced_final=True).
REASONING_STALL_NUDGE = (
    "Your previous response was reasoning only — you produced no tool calls and "
    "no visible output. Stop planning in your head. ACT NOW, in this order:\n"
    "1. write_file — create the deliverable file the task asks for (e.g. /app/solution.py, "
    "/app/main.rs, the output file the verifier checks). Put your best current approach "
    "into code, even if it's a first draft. An imperfect file you can iterate on beats a "
    "perfect plan that lives only in your head.\n"
    "2. run_bash — run it to see what happens (python3 /app/solution.py, cargo run, etc.).\n"
    "3. edit_file / write_file — fix what the run told you was wrong, then re-run.\n"
    "Do not produce a final_answer until the deliverable file exists and runs. Pick the "
    "most promising approach from your reasoning and execute it right now."
)
FORCED_FINAL_NUDGE = (
    "Your previous streamed response produced reasoning only. Do not continue "
    "private reasoning. Use the final_answer tool if available. Return a complete "
    "visible answer now in at most six short bullets or paragraphs. If you are "
    "blocked, say exactly what is blocking you and what input is needed."
)
EMPTY_TURN_NUDGE = (
    "Your previous response was empty — you produced no visible text and no tool "
    "call, so nothing happened. Do not stop. If the task is not finished, emit the "
    "next tool call now. If you have everything you need, use the final_answer tool "
    "(or reply with a concise visible answer). Do not repeat the empty turn."
)
MANUAL_RECOVERY_NUDGE = (
    "Manual recovery requested by the user because the previous response appeared "
    "off the rails or corrupted. Discard any corrupted reasoning/output and do not "
    "continue private reasoning. Use the final_answer tool if available. Return a "
    "bounded visible checkpoint with: (1) the last valid result you can rely on, "
    "(2) the next concrete action you would take, and (3) any blocker or "
    "uncertainty. Keep it concise."
)


def _nudge_current_user_turn(messages, nudge):
    note = f"[Runtime note: {nudge}]"
    for msg in reversed(messages):
        if msg.get("role") != "user" or not isinstance(msg.get("content"), str):
            continue
        content = RUNTIME_NOTE_RE.sub("", msg["content"]).rstrip()
        msg["content"] = f"{content}\n\n{note}" if content else note
        return
    messages.append({"role": "user", "content": note})


def _last_is_dangling_tool(messages):
    """True if the last message is a `tool` result with no assistant turn after it.

    This is the layout right after a tool ran and before the model follows up: an
    empty assistant turn here is most suspicious — the model produced nothing in
    response to a fresh tool result, so the user sees the agent "finish" with no
    answer. Auto-continue is most warranted in exactly this case.
    """
    return bool(messages) and messages[-1].get("role") == "tool"


# --- risk classifier --------------------------------------------------------
# One cheap non-streaming call per write/bash action. Same model, tiny prompt,
# expects {"level": "low|medium|high", "reason": "<short>"}. Defensive parse —
# if the model rambles or returns garbage we fall back to "high" so we err on
# the side of asking. Connection failures get a few short retries first because
# Minion normally talks to reliable or local endpoints. Skipped entirely in YOLO
# mode (no point paying for a call we won't act on) and for read-only tools.

def _detect_project_root(cwd=None):
    """Return the git root for cwd when available, otherwise cwd itself."""
    cwd = os.path.realpath(cwd or os.getcwd())
    if shutil.which("git"):
        try:
            r = subprocess.run(
                ["git", "-C", cwd, "rev-parse", "--show-toplevel"],
                capture_output=True, text=True, timeout=2)
            root = r.stdout.strip()
            if r.returncode == 0 and root:
                return os.path.realpath(root)
        except (OSError, subprocess.SubprocessError):
            pass
    return cwd


PROJECT_ROOT = _detect_project_root()


def _is_under_path(path, root):
    try:
        return os.path.commonpath([path, root]) == root
    except (OSError, ValueError):
        return False


def _extract_action_path(action):
    """Pull the primary file path from Minion's short approval action string."""
    if not isinstance(action, str):
        return None
    action = action.strip()
    if action.startswith("edit "):
        return action[len("edit "):].strip() or None
    m = re.match(r"^write\s+(.+)\s+\(\d+\s+bytes\)$", action)
    if m:
        return m.group(1).strip() or None
    return None


def _classify_action_path(path, cwd=None, project_root=None):
    cwd = os.path.realpath(cwd or os.getcwd())
    project_root = os.path.realpath(project_root or PROJECT_ROOT)
    expanded = os.path.expanduser(path)
    if not os.path.isabs(expanded):
        expanded = os.path.join(cwd, expanded)
    abs_path = os.path.realpath(expanded)
    home = os.path.realpath(os.path.expanduser("~"))
    downloads = os.path.join(home, "Downloads")
    rel_home = None
    if _is_under_path(abs_path, home):
        try:
            rel_home = os.path.relpath(abs_path, home)
        except ValueError:
            rel_home = None
    home_parts = [] if rel_home in (None, ".") else rel_home.split(os.sep)
    touches_home_dotdir = bool(home_parts and home_parts[0].startswith("."))
    in_project = _is_under_path(abs_path, project_root)
    return {
        "primary_path": abs_path,
        "path_scope": "in_project" if in_project else "outside_project",
        "path_in_cwd": _is_under_path(abs_path, cwd),
        "path_in_downloads": _is_under_path(abs_path, downloads),
        "path_touches_home_dotdir": touches_home_dotdir,
    }


def _risk_user_message(action):
    cwd = os.path.realpath(os.getcwd())
    msg = {
        "action": action,
        "cwd": cwd,
        "project_root": os.path.realpath(PROJECT_ROOT),
        "primary_path": None,
        "path_scope": "unknown",
        "scope_guidance": (
            "Use project_root/path_scope for outside-project decisions. "
            "A file under project_root is in-project even when project_root "
            "is inside ~/Downloads."
        ),
    }
    path = _extract_action_path(action)
    if path:
        msg.update(_classify_action_path(path, cwd=cwd))
    return json.dumps(msg, sort_keys=True)


RISK_SYSTEM = (
    "You are a risk classifier for a coding agent's tool calls. "
    "Given one tool action as JSON, respond with ONLY a JSON object of the form "
    '{"level": "low"|"medium"|"high", "reason": "<one short sentence>"}.\n'
    "The user JSON includes cwd, project_root, and when possible path_scope. "
    "For outside-project decisions, trust project_root/path_scope over folder-name heuristics. "
    "If path_scope is in_project, do not classify high merely because the path is under ~/Downloads; "
    "projects can live in Downloads. If path_scope is outside_project, treat that as outside the project.\n"
    "Levels:\n"
    '- low: read-only or trivially reversible (ls, cat, grep, git status, mkdir, touch, file reads).\n'
    '- medium: modifies state but contained/reversible (writing a single file, editing a file, cp, mv, '
    'pip install in a venv, running tests, git commit).\n'
    '- high: destructive, hard to reverse, or broad scope (rm -rf, git push --force, git reset --hard, '
    'dd, chmod -R, writing outside the project, network sends to external hosts, killing processes, '
    'system-level changes, anything touching dotfiles in $HOME).\n'
    "When in doubt, classify higher. Output ONLY the JSON, no preamble."
)


def _assess_risk(action):
    """Return (level, reason). level is one of LEVEL_ORDER; reason is a short
    string. On any failure (server down, bad JSON, unknown level) returns
    ("high", "<error>") so the caller falls through to the prompt path."""
    payload = [
        {"role": "system", "content": RISK_SYSTEM},
        {"role": "user", "content": _risk_user_message(action)},
    ]
    max_conn = max(0, RISK_CONNECTION_RETRIES) + 1
    max_parse = 3
    try:
        for parse_attempt in range(max_parse):
            resp = None
            for conn_attempt in range(max_conn):
                try:
                    _log_event("req", {"model": MODEL, "messages": payload,
                                       "stream": False, "_purpose": "risk",
                                       "_attempt": conn_attempt + 1,
                                       "_parse_attempt": parse_attempt + 1})
                    resp = client.chat.completions.create(
                        model=MODEL, messages=payload, stream=False, timeout=15,
                        **(ACTIVE.extra_request_kwargs() if ACTIVE else {}))
                    break
                except APIConnectionError:
                    if conn_attempt >= max_conn - 1:
                        return ("high", f"server unreachable after {max_conn} attempts; defaulting to high")
                    time.sleep(max(0, RISK_CONNECTION_RETRY_SECONDS))
            if resp is None:
                return ("high", "no response; defaulting to high")
            try:
                _log_event("resp", {"_purpose": "risk", "data": resp.model_dump()})
            except Exception:
                pass
            text = (resp.choices[0].message.content or "").strip()
            level, reason = None, ""
            try:
                obj = json.loads(text)
                level = (obj.get("level") or "").strip().lower()
                reason = (obj.get("reason") or "").strip()
            except (json.JSONDecodeError, AttributeError, TypeError):
                m = re.search(r'\b(low|medium|high)\b', text, re.IGNORECASE)
                if m:
                    level = m.group(1).lower()
                reason = text[:120]
            if level in LEVEL_ORDER:
                return (level, reason or level)
            # Unparseable — add a sharp follow-up instruction and retry
            if parse_attempt < max_parse - 1:
                payload.append({"role": "assistant", "content": text or "(no output)"})
                payload.append({"role": "user", "content": (
                    "Respond with ONLY a single JSON object containing \"level\" and \"reason\" fields. "
                    "No preamble, no explanation, no markdown. Example:\n"
                    '{"level": "medium", "reason": "edits a file in the project"}'
                )})
        return ("high", f"unparseable risk response after {max_parse} attempts: {text[:80]!r}")
    except Exception as e:
        return ("high", f"risk call failed: {type(e).__name__}")


# A special exception that propagates up from _confirm (via the tool function
# body → run_tool → model_turn) when the user presses Esc at an approval prompt.
# It's not a real error — it's a control-flow signal meaning "stop this turn and
# drop the user back to the chat input so they can add more guidance."
class _EscToChat(Exception):
    pass


_ACTIVE_SPINNER = None  # set by run_tool() while a tool body is executing


def _ask_approval(prompt):
    """Read a single keypress for a Y/n/esc approval prompt. Returns one of
    'y', 'n', 'esc'. Esc is accepted both as the bare Esc key (read in raw
    mode) and as the letter 'e'/'E' for terminals where raw mode isn't
    usable. Falls back to line-based `input()` when stdin isn't a TTY."""
    if not sys.stdin.isatty() or not sys.stdout.isatty():
        ans = input(prompt).strip().lower()
        if ans in ("esc", "e", "\x1b"):
            return "esc"
        if ans == "n":
            return "n"
        return "y"
    fd = sys.stdin.fileno()
    old = termios.tcgetattr(fd)
    new = old[:]
    new[3] &= ~(termios.ECHO | termios.ICANON)
    new[6][termios.VMIN] = 1
    new[6][termios.VTIME] = 0
    try:
        termios.tcsetattr(fd, termios.TCSADRAIN, new)
        sys.stdout.write(prompt)
        sys.stdout.flush()
        c = os.read(fd, 1).decode("utf-8", "replace")
        if c == "\x1b":
            # Could be bare Esc OR the lead byte of an escape sequence
            # (arrow keys, Home/End, bracketed paste, …). Wait briefly; if
            # no more bytes arrive, it's a bare Esc.
            r, _, _ = select.select([fd], [], [], 0.05)
            if r:
                try:
                    os.read(fd, 1)  # swallow the rest of the sequence
                except OSError:
                    pass
                # Any non-bare-Esc sequence (arrow keys etc.) → default proceed.
                sys.stdout.write("\n")
                sys.stdout.flush()
                return "y"
            sys.stdout.write(f"{DIM}esc{RESET}\n")
            sys.stdout.flush()
            return "esc"
        # Echo the choice for feedback
        if c in ("\r", "\n"):
            shown, result = "Y (default)", "y"
        elif c in ("y", "Y"):
            shown, result = "Y", "y"
        elif c in ("n", "N"):
            shown, result = "n", "n"
        elif c in ("e", "E"):
            shown, result = "esc", "esc"
        else:
            shown, result = f"{c} → Y (default)", "y"
        sys.stdout.write(f"{DIM}{shown}{RESET}\n")
        sys.stdout.flush()
        return result
    finally:
        try:
            termios.tcsetattr(fd, termios.TCSADRAIN, old)
        except Exception:
            pass


def _confirm(action):
    """Decide whether to run `action`. Returns True to proceed, False to deny.
    Raises `_EscToChat` if the user presses Esc — propagating up so the turn
    stops and the REPL drops back to the chat input.

    Flow:
      1. YOLO → True (no call, no prompt).
      2. Ask the model for a risk level (skipped in step 1).
      3. If APPROVE_LEVEL is set and level ≤ APPROVE_LEVEL → auto-allow.
      4. Otherwise prompt (Y/n/esc), showing the level + reason so the user
         has context.

    Reads module globals YOLO and APPROVE_LEVEL at call time — /yolo and
    /approval reassign them mid-session, and we must always see the latest
    value, not a snapshot from when the tool function was defined.
    """
    if YOLO:
        return True
    # If a tool spinner is running, keep it alive during the risk-classifier
    # request and pause it only around our own terminal I/O. Otherwise a slow
    # classifier call looks exactly like a frozen tool.
    sp = _ACTIVE_SPINNER
    paused = False

    def pause_spinner():
        nonlocal paused
        if sp is not None and not paused:
            sp.stop()
            paused = True

    try:
        level, reason = _assess_risk(action)
        # APPROVE_LEVEL is the max level to AUTO-APPROVE. level ≤ threshold → run.
        if APPROVE_LEVEL is not None and LEVEL_ORDER[level] <= LEVEL_ORDER[APPROVE_LEVEL]:
            # Auto-allow. Show the assessment so the user has a paper trail.
            short = reason if len(reason) <= 80 else reason[:77] + "..."
            pause_spinner()
            print(f"{DIM}  ↳ auto-allow [{level}] {action}  ({short}){RESET}")
            return True
        short = reason if len(reason) <= 80 else reason[:77] + "..."
        lvl_color = {"low": DIM, "medium": YELLOW, "high": RED}[level]
        pause_spinner()
        choice = _ask_approval(
            f"{YELLOW}  allow {action}? {lvl_color}[risk: {level.upper()} — {short}]{RESET} "
            f"{YELLOW}[Y/n/esc] {RESET}")
        if choice == "esc":
            raise _EscToChat(action)
        return choice != "n"
    finally:
        if sp is not None and paused:
            sp.start()


# --- text-fallback parsing --------------------------------------------------
TOOL_TAG_PATTERN = r"<tool_call>\s*(\{.*?\})\s*</tool_call>"
MINION_TOOL_TAG_PATTERN = r"\[minion_tool_call\]\s*(\{.*?\})\s*\[/minion_tool_call\]"
TOOL_TAG = re.compile(TOOL_TAG_PATTERN, re.DOTALL)
MINION_TOOL_TAG = re.compile(MINION_TOOL_TAG_PATTERN, re.DOTALL)
TOOL_TAGS = (MINION_TOOL_TAG, TOOL_TAG)  # prefer Minion's bracketed protocol; keep legacy support
TOOL_PROTOCOL_TAG_RE = re.compile(r"</?tool_call\b[^>]*>")
TOOL_RESULT_PROTOCOL_NOTE = (
    "[minion note: escaped tool-call protocol delimiters from this tool result "
    "before sending it back to the model]\n"
)


def parse_text_calls(content):
    """Pull standalone text-protocol tool-call messages.

    Literal protocol tags can appear in files, docs, and code examples. Treat
    them as executable only when the whole assistant message is protocol text.
    """
    text = content or ""
    pos = 0
    calls = []
    while pos < len(text):
        while pos < len(text) and text[pos].isspace():
            pos += 1
        if pos >= len(text):
            break
        match = None
        for tag in TOOL_TAGS:
            match = tag.match(text, pos)
            if match:
                break
        if not match:
            return []
        try:
            obj = json.loads(match.group(1))
            calls.append((obj["name"], obj.get("arguments", {})))
        except (json.JSONDecodeError, KeyError):
            return []
        pos = match.end()
    return calls


def _escape_tool_protocol_delimiters(text):
    """Neutralize active tool-call delimiters in untrusted tool output."""
    if not isinstance(text, str):
        return text
    safe = TOOL_PROTOCOL_TAG_RE.sub(
        lambda m: m.group(0).replace("<", "&lt;").replace(">", "&gt;"),
        text)
    safe = safe.replace("[minion_tool_call]", "&#91;minion_tool_call&#93;")
    safe = safe.replace("[/minion_tool_call]", "&#91;/minion_tool_call&#93;")
    if safe != text:
        return TOOL_RESULT_PROTOCOL_NOTE + safe
    return text


def _sanitize_tool_result(text):
    """Neutralize active protocol delimiters, dedup, and cap tool results.

    Large, near-duplicate tool outputs (find/ls/grep path dumps) are the fuel
    for context-copying repetition collapse, so we collapse runs of >=3
    identical consecutive lines, then head/tail-cap to TOOL_RESULT_CHARS with a
    visible marker. Short, non-repetitive results pass through except for
    protocol-delimiter escaping.
    """
    if not isinstance(text, str):
        return text
    text = _escape_tool_protocol_delimiters(text)
    if len(text) <= 1000:
        return text
    lines = text.split("\n")
    out_lines = []
    i, n = 0, len(lines)
    while i < n:
        j = i + 1
        while j < n and lines[j] == lines[i]:
            j += 1
        run = j - i
        if run >= 3:
            out_lines.append(lines[i])
            out_lines.append(f"... [+{run - 1} identical lines elided]")
        else:
            out_lines.extend(lines[i:j])
        i = j
    result = "\n".join(out_lines)
    budget = TOOL_RESULT_CHARS
    if budget > 0 and len(result) > budget:
        head = budget * 2 // 3
        tail = budget - head
        elided = len(result) - head - tail
        result = (result[:head]
                  + f"\n... [{elided} chars elided to bound context; "
                    f"re-run more narrowly if you need the rest]\n"
                  + result[-tail:])
    return result


def run_tool(name, args):
    fn = DISPATCH.get(name)
    if not fn:
        return f"ERROR: unknown tool {name}"
    # newline so the tool arrow gets its own line — streamed text uses end=""
    # and would otherwise run straight into the indicator
    arg_preview = json.dumps(args)
    if len(arg_preview) > 120:
        arg_preview = arg_preview[:117] + "..."
    print(f"\n{CYAN}  ┌─ {name}{RESET}")
    print(f"{CYAN}  │ {RESET}{DIM}{arg_preview}{RESET}")
    # Animate the gap between "cyan args line" and "cyan result line". Tool
    # bodies can take a while — _confirm makes a network round-trip to the
    # risk classifier, run_bash can run for tens of seconds, write_file on a
    # big payload takes a beat — and without this the user just sees a frozen
    # screen after the green model output finishes. _confirm pauses/resumes
    # us around its own I/O so the auto-allow line / Y/n prompt aren't clobbered.
    spinner = LifeSpinner(label=f"running {name}")
    spinner.start()
    global _ACTIVE_SPINNER
    _ACTIVE_SPINNER = spinner
    try:
        result = fn(**args)
    except _EscToChat:
        # Close the tool box before propagating so it isn't left visually open.
        print(f"{CYAN}  └─{RESET} {DIM}(escaped){RESET}")
        raise  # not an error — control-flow signal back to model_turn/REPL
    except Exception as e:  # noqa: BLE001 — surface any tool error back to the model
        result = f"ERROR: {type(e).__name__}: {e}"
    finally:
        _ACTIVE_SPINNER = None
        spinner.stop()
    # Dedup + cap the result BEFORE it enters the message history (the model and
    # the on-screen preview both get this sanitized form). Bounding repetitive
    # tool output is the cheapest defense against context-copying collapse.
    result = _sanitize_tool_result(result)
    # box the result; the preview is a further display-only trim of the (already
    # context-bounded) result
    preview = result if len(result) < 800 else result[:800] + f"\n... [{len(result) - 800} more chars]"
    for line in preview.splitlines():
        print(f"{CYAN}  │ {RESET}{line}")
    print(f"{CYAN}  └─{RESET}")
    return result


def _tool_args_parse_error(c, err, finish_reasons=None):
    name = c.get("name") or "tool"
    args_text = c.get("args") or ""
    detail = f"{name} arguments are not valid JSON"
    if args_text:
        detail += f" ({_abbr(len(args_text))} chars"
        if finish_reasons:
            detail += f"; finish_reason={','.join(finish_reasons)}"
        detail += f"; {err.msg} at char {err.pos})"
    else:
        detail += f" ({err.msg} at char {err.pos})"
    return detail


def _recovery_sampling_opts():
    """Loop-breaking sampler for recovery retries (see RECOVERY_* constants).

    temperature/top_p are standard OpenAI params and go top-level. min_p,
    repeat_penalty, repeat_last_n and the DRY family are llama.cpp extensions the
    OpenAI SDK won't accept as kwargs, so they ride in extra_body — llama.cpp's
    server merges that into the request body; non-llama.cpp endpoints ignore
    unknown keys. Set any RECOVERY_* value negative to omit it.  When
    MINION_BACKEND=vllm the llama.cpp-only knobs are set to None instead of
    their defaults, so the checks below miss them automatically."""
    opts = {}
    if RECOVERY_TEMPERATURE is not None and RECOVERY_TEMPERATURE >= 0:
        opts["temperature"] = RECOVERY_TEMPERATURE
    if RECOVERY_TOP_P is not None and RECOVERY_TOP_P >= 0:
        opts["top_p"] = RECOVERY_TOP_P
    extra = {}
    if RECOVERY_MIN_P is not None and RECOVERY_MIN_P >= 0:
        extra["min_p"] = RECOVERY_MIN_P
    if RECOVERY_REPEAT_PENALTY is not None and RECOVERY_REPEAT_PENALTY >= 0:
        extra["repeat_penalty"] = RECOVERY_REPEAT_PENALTY
        if RECOVERY_REPEAT_LAST_N is not None:
            extra["repeat_last_n"] = RECOVERY_REPEAT_LAST_N
    if RECOVERY_DRY_MULTIPLIER is not None and RECOVERY_DRY_MULTIPLIER > 0:
        extra["dry_multiplier"] = RECOVERY_DRY_MULTIPLIER
        if RECOVERY_DRY_BASE is not None:
            extra["dry_base"] = RECOVERY_DRY_BASE
        if RECOVERY_DRY_ALLOWED_LENGTH is not None:
            extra["dry_allowed_length"] = RECOVERY_DRY_ALLOWED_LENGTH
        if RECOVERY_DRY_SEQUENCE_BREAKERS is not None:
            extra["dry_sequence_breakers"] = RECOVERY_DRY_SEQUENCE_BREAKERS
    if extra:
        opts["extra_body"] = extra
    return opts


def open_stream(messages, tools=TOOLS, tool_choice=None, max_tokens=None,
                recovery_sampling=False):
    """Open a streaming completion. Retries without tools= if the server rejects
    that param; returns None (after a friendly message) on connection/API failure."""
    try:
        _prune_empty_assistant_messages(messages)
        token_limit = MAX_COMPLETION_TOKENS if max_tokens is None else max_tokens
        request_opts = {}
        if token_limit > 0:
            request_opts["max_tokens"] = token_limit
        if recovery_sampling:
            request_opts.update(_recovery_sampling_opts())
        if tool_choice is not None:
            request_opts["tool_choice"] = tool_choice
        # Per-request body fields (e.g. OpenRouter provider routing) ride in
        # extra_body. Merged here once so both the tools and no-tools fallback
        # branches below carry it through unchanged.
        if ACTIVE:
            request_opts.update(ACTIVE.extra_request_kwargs())
        try:
            _log_event("req", {"model": MODEL, "messages": messages, "tools": tools, "stream": True, **request_opts})
            stream = client.chat.completions.create(
                model=MODEL, messages=messages, tools=tools, stream=True,
                stream_options={"include_usage": True}, **request_opts)
        except APIConnectionError:
            raise  # server unreachable — don't bother retrying without tools
        except httpx.HTTPError:
            raise  # transport failure — retrying without tools won't help
        except Exception:  # reachable but rejected tools= → text-protocol fallback
            # If extra_body (e.g. min_p / repeat_penalty / dry_*) is also
            # rejected — vLLM's speculative decoder doesn't support min_p etc. —
            # strip it too.  Try no-tools first, then no-tools + no-extra.
            for fallback_keys in ({"tool_choice"}, {"tool_choice", "extra_body"}):
                fallback_opts = {k: v for k, v in request_opts.items()
                                 if k not in fallback_keys}
                try:
                    _tag = "no-tools" if fallback_keys == {"tool_choice"} else "no-tools-no-extra"
                    _log_event("req", {"model": MODEL, "messages": messages, "stream": True, "_fallback": _tag, **fallback_opts})
                    stream = client.chat.completions.create(
                        model=MODEL, messages=messages, stream=True,
                        stream_options={"include_usage": True}, **fallback_opts)
                    break  # succeeded
                except (APIConnectionError, httpx.HTTPError):
                    raise  # transport failure — don't retry further
                except Exception:
                    continue  # try next fallback level
            else:
                raise  # all fallbacks exhausted
        # Wrap the stream so every chunk is captured to the log on its way out.
        return _LoggingStream(stream, _llog)
    except (APIConnectionError, httpx.HTTPError):
        print(f"{RED}  ✗ can't reach {client.base_url} — is the server up? "
              f"Set MINION_BASE_URL (and MINION_MODEL) to point at it.{RESET}")
    except Exception as e:
        print(f"{RED}  ✗ API error: {type(e).__name__}: {e}{RESET}")
    return None


# --- context compression ----------------------------------------------------
# Summarize the older turns of `messages` into a single user-role turn, keeping
# the system prompt and the last K turns verbatim. Frees context without losing
# the model's grip on what it was just doing.
COMPRESS_KEEP = 2  # how many recent turns to leave untouched


def compress(messages, keep=COMPRESS_KEEP, auto=False):
    """Ask the model to summarize everything except system + last `keep` turns.

    Mutates `messages` in place on success: replaces the middle slice with a
    single user-role summary turn. Returns (kept_n, summarized_n, summary_chars)
    or None on failure (in which case `messages` is untouched).

    Non-streaming on purpose — we want the whole summary before splicing it in,
    and a spinner for a one-shot summary would be visual noise.

    When auto=True the keep count is raised above COMPRESS_KEEP so
    auto-compression (triggered when the context window fills past
    AUTOCOMPRESS_PERCENT) is more conservative than a manual /compress —
    it keeps roughly the last third of the conversation verbatim so
    in-progress work and recent tool results survive the fold.
    """
    # Layout: [system?, ..., user, assistant, tool, ..., user, assistant(tool_calls)?, ...]
    # We assume messages[0] is the system prompt (matches how main() builds it).
    # Anything before the "tail" we want to summarize; the tail stays verbatim.
    sys_msg = messages[0] if messages and messages[0].get("role") == "system" else None
    if auto:
        body_len = len(messages) - (1 if sys_msg else 0)
        keep = max(keep, body_len // 3)
    if len(messages) <= 1 + keep:
        return None  # nothing to compress

    body = messages[1:] if sys_msg else messages
    if len(body) <= keep:
        return None

    head, tail = body[:-keep], body[-keep:]
    summarized_n = len(head)

    # The tail must start on a turn the chat template can render. A `tool` turn
    # with no preceding assistant(tool_calls) parent — or an assistant(tool_calls)
    # turn whose result got cut off into `head` — makes llama.cpp's Jinja template
    # raise "Message has tool role, but there was no previous assistant message
    # with a tool call!". Walk from the front of the tail and drop any leading
    # tool/half-tool-call turns until we land on something safe (user, plain
    # assistant, or system). Bump `summarized_n` so the user-visible count stays
    # honest about how many turns actually got folded into the summary.
    while tail and tail[0].get("role") in ("tool", "assistant"):
        first = tail[0]
        if first.get("role") == "tool":
            tail = tail[1:]
            summarized_n += 1
            continue
        # assistant: only safe if it has NO tool_calls, OR every tool_call has
        # its matching tool result later in the tail
        if first.get("tool_calls"):
            ids = {tc["id"] for tc in first["tool_calls"]}
            seen = set()
            for m in tail[1:]:
                tcid = m.get("tool_call_id")
                if m.get("role") == "tool" and tcid:
                    seen.add(tcid)
            if ids - seen:
                tail = tail[1:]
                summarized_n += 1
                continue
        break

    # Render the head as plain text the model can summarize. Tool outputs are
    # the bulkiest part of a real session — include them but truncate each one
    # so a single huge read_file doesn't blow up the summary prompt itself.
    def _render(msgs):
        out = []
        for m in msgs:
            role = m.get("role", "?")
            content = m.get("content")
            if content is None and m.get("tool_calls"):
                # assistant tool-call turn — show the calls so the summary knows what ran
                calls = ", ".join(
                    f"{c['function']['name']}({c['function']['arguments']})"
                    for c in m["tool_calls"]
                )
                out.append(f"[{role}] → {calls}")
            elif isinstance(content, list):
                # some servers return content as a list of parts; flatten it
                content = "".join(p.get("text", "") for p in content if isinstance(p, dict))
                out.append(f"[{role}] {content[:2000]}")
            else:
                out.append(f"[{role}] {(content or '')[:2000]}")
        return "\n\n".join(out)

    summary_prompt = (
        "Summarize the following conversation history for context retention. "
        "Preserve: the original user goal/task, key decisions made, file paths "
        "and identifiers touched, current state of any in-progress work, and "
        "any unresolved questions. Drop: raw tool outputs, full file contents, "
        "and verbose back-and-forth — keep it dense and information-rich. "
        "Write in the same language as the conversation. Output ONLY the "
        "summary, no preamble.\n\n"
        f"---\n{_render(head)}\n---"
    )

    payload = [{"role": "user", "content": summary_prompt}]
    try:
        _log_event("req", {"model": MODEL, "messages": payload, "stream": False, "_purpose": "compress"})
        resp = client.chat.completions.create(
            model=MODEL, messages=payload, stream=False,
            **(ACTIVE.extra_request_kwargs() if ACTIVE else {}))
        try:
            _log_event("resp", {"_purpose": "compress", "data": resp.model_dump()})
        except Exception:
            pass  # never let logging break the summary call
    except APIConnectionError:
        print(f"{RED}  ✗ can't reach {client.base_url} — context unchanged{RESET}")
        return None
    except Exception as e:
        print(f"{RED}  ✗ compress failed: {type(e).__name__}: {e}{RESET}")
        return None

    summary = (resp.choices[0].message.content or "").strip()
    if not summary:
        print(f"{RED}  ✗ compress returned empty summary — context unchanged{RESET}")
        return None

    header = f"[Compressed context — {summarized_n} earlier turns summarized; last {keep} turns kept verbatim]"
    new_mid = [{"role": "user", "content": f"{header}\n\n{summary}"}]
    messages[:] = ([sys_msg] if sys_msg else []) + new_mid + tail
    return len(tail), summarized_n, len(summary)


def _maybe_autocompress(messages, prompt_tokens):
    """Silently compress when context usage crosses the AUTOCOMPRESS_PERCENT
    threshold. Called after each settled model turn with the last turn's
    `prompt_tokens` (the size of the context the model just saw).

    Returns True if a compression happened, False otherwise. Guards:
      - AUTOCOMPRESS_PERCENT == 0 → disabled
      - ACTIVE context window unknown / ≤ 0 → can't compute a ratio
      - ratio < threshold → nothing to do
      - compress() itself bails when there's too little to fold
    """
    if AUTOCOMPRESS_PERCENT <= 0:
        return False
    if not isinstance(prompt_tokens, int) or prompt_tokens <= 0:
        return False
    if ACTIVE is None:
        return False
    mx = ACTIVE._context_window
    if not isinstance(mx, int) or mx <= 0:
        return False
    ratio = prompt_tokens / mx
    if ratio * 100 < AUTOCOMPRESS_PERCENT:
        return False
    result = compress(messages, auto=True)
    if result is None:
        return False
    kept_n, summarized_n, summary_chars = result
    pct = int(ratio * 100)
    print(f"{DIM}  ↻ auto-compressed {summarized_n} turns → 1 summary "
          f"({summary_chars} chars), kept last {kept_n} verbatim "
          f"(context was at {pct}% of {_abbr(mx)}){RESET}")
    return True


class _LoggingStream:
    """Iterator wrapper that tees each SSE chunk to llamacpp.log before yielding.
    Uses model_dump so we capture the chunk's full structure (incl. reasoning_content)."""
    def __init__(self, inner, log_file):
        self._inner = inner
        self._log = log_file

    def __iter__(self):
        for chunk in self._inner:
            try:
                self._log.write(json.dumps({"ts": time.time(), "dir": "resp",
                                             "data": chunk.model_dump()}) + "\n")
            except Exception:
                pass  # never let logging break the stream
            yield chunk

    def close(self):
        close = getattr(self._inner, "close", None)
        if close:
            try:
                close()
            except Exception:
                pass


def _tool_call_progress(tcs):
    parts = []
    for i in sorted(tcs):
        c = tcs[i]
        name = c["name"] or "tool"
        args_n = len(c["args"])
        parts.append(f"{name} args {_precise_abbr(args_n)} chars" if args_n else f"{name} ...")
    return ", ".join(parts) if parts else "tool call ..."


TURN_DONE = "done"
TURN_TOOL = "tool"
TURN_STREAM_CUT = "stream_cut"
TURN_FORCE_FINAL = "force_final"
TURN_STALL = "stall"  # reasoning-only stall → nudge to act (full tools retained)
TURN_EMPTY = "empty"  # assistant emitted no content and no tool calls → silent stall
TURN_ESC = "esc"  # user pressed Esc at an approval prompt → drop to chat input


def _ctx_field(prompt_n):
    """The colored `ctx` field for the stats footer: `<cur>/<max> ctx`.
    The max context window is cyan; the current context is colored by
    utilization — green under 30%, yellow under 60%, red at or above.
    When the max isn't known yet (probe still pending or unreachable)
    the field degrades to `<cur> ctx` with no color, and a background
    probe is kicked off so a later turn can fill in the max."""
    cur = _abbr(prompt_n)
    if ACTIVE is None:
        return f"{cur} ctx"
    mx = ACTIVE._context_window
    if not isinstance(mx, int) or mx <= 0:
        _ensure_ctx_probe(ACTIVE)
        return f"{cur} ctx"
    mx_s = _abbr(mx)
    ratio = prompt_n / mx if mx > 0 else 0.0
    if ratio < 0.30:
        color = GREEN
    elif ratio < 0.60:
        color = YELLOW
    else:
        color = RED
    # RESET after each colored run, then re-apply DIM so the rest of the
    # footer (joined around this field) stays dim.
    return f"{color}{cur}{RESET}{DIM}/{CYAN}{mx_s}{RESET}{DIM} ctx"


_CTX_PROBE_STARTED = set()  # id(Source) currently being probed (avoid dupe threads)


def _ensure_ctx_probe(src):
    """Start a background thread to resolve a Source's max context window if
    it isn't already known and no probe is in flight for it. The result lands
    in src._context_window; the footer picks it up on a later turn. Idempotent
    via _CTX_PROBE_STARTED (cleared when the thread finishes).

    A local server's metadata endpoints can 503 "Loading model" for tens of
    seconds at cold start (the weights are still mmapping) even while a chat
    slot is already serving — so a single attempt often misses. The thread
    retries with backoff until it resolves (or gives up after ~2 min), so the
    suffix fills in the first turn AFTER the server is ready rather than
    depending on the user to keep typing."""
    sid = id(src)
    if sid in _CTX_PROBE_STARTED:
        return
    _CTX_PROBE_STARTED.add(sid)

    def _work():
        # Backoff schedule (seconds between attempts): a handful of quick
        # tries for a fast local server, then wider spacing for a slow cold
        # start. Total ~2 min of probing before giving up (the footer will
        # re-arm on the next user turn anyway, so this is just a best effort
        # to resolve the max before the user has to wait for another turn).
        delays = [1, 2, 3, 5, 8, 10, 15, 20, 25, 30, 30]
        try:
            for delay in delays:
                if isinstance(src._context_window, int) and src._context_window > 0:
                    return  # resolved by another path (e.g. a /source switch)
                src.resolve_context_window()
                if isinstance(src._context_window, int) and src._context_window > 0:
                    return
                time.sleep(delay)
        except Exception:
            pass  # never raise from a background probe
        finally:
            _CTX_PROBE_STARTED.discard(sid)

    t = threading.Thread(target=_work, daemon=True, name="minion-ctx-probe")
    t.start()


# --- one model turn (streamed), returns TURN_* status -----------------------
def model_turn(messages, reasoning_loop_cut_count=0, malformed_stream_cut_count=0,
               empty_turn_count=0, forced_final=False, recovery_sampling=False):
    global _LAST_PROMPT_TOKENS
    # Start the spinner BEFORE open_stream() so the HTTP-handshake + interrupt-
    # watcher-setup window (which can be tens of ms on a warm local server but
    # seconds on a cold/wake-from-sleep one) isn't a frozen green-text gap.
    # The spinner is still killed by the first SSE chunk in the loop below.
    # t0 must be set BEFORE open_stream() — the HTTP request is sent inside
    # open_stream() (TCP/TLS handshake, request bytes, etc.), and all of that
    # latency is part of TTFT. If we set t0 after, the server has already been
    # processing and the first token may be sitting in the buffer by the time
    # we start the clock, making TTFT look implausibly tiny.
    t0 = time.time()
    spinner_label = (
        "forcing final answer · esc to interrupt"
        if forced_final else
        "recovering · esc to interrupt" if recovery_sampling else
        "thinking · esc to interrupt"
    )
    spinner = LifeSpinner(label=spinner_label)
    spinner.start()
    try:
        if forced_final:
            stream = open_stream(
                messages,
                tools=[FINAL_ANSWER_TOOL],
                tool_choice=FINAL_ANSWER_TOOL_CHOICE,
                max_tokens=FORCED_FINAL_MAX_TOKENS,
                recovery_sampling=recovery_sampling,
            )
        else:
            stream = open_stream(messages, recovery_sampling=recovery_sampling)
    except Exception:
        spinner.stop()
        raise
    if stream is None:
        spinner.stop()
        return TURN_DONE  # error already reported; REPL continues

    # Interrupt watcher: the singleton daemon thread watches stdin for bare Esc
    # while ARMED. On hit it sets _USER_INTERRUPTED; the main loop breaks out of
    # the chunk loop on its next iteration. We arm it BEFORE the spinner so the
    # user always has a moment to interrupt even if the first token takes a
    # while to arrive.
    _ensure_interrupt_watcher()
    _USER_INTERRUPTED.clear()
    # Clear the "termios restored" flag BEFORE arming so we can wait on it
    # after disarm. The watcher sets it in its inner finally once it has
    # restored cooked mode. Without this handshake the watcher's termios
    # restore is asynchronous (it may still be in select(0.1) when we
    # return), and the chatbox can capture the watcher's raw-mode termios
    # as its "old" state — leaving the terminal stuck in VMIN=0/ISIG-off
    # with broken Enter and garbled typing.
    _INTERRUPT_RESTORED.clear()
    _INTERRUPT_ARMED.set()
    content, tcs, mode = [], {}, None
    reasoning_parts = []  # captured reasoning_content chunks (for stall feedback)
    timings = None
    usage = None
    streamed_chars = 0    # client-side character count for fallback token estimate
    t_first = None   # time of first output token (for TTFT)
    loop_cut = False
    reasoning_only_chars = 0
    reasoning_only_start = None  # wall-clock when reasoning-only streaming began
    stream_error = None
    interrupted = False
    finish_reasons = []
    tool_status_active = False
    last_tool_status = 0.0
    leaked_token_run = 0   # consecutive leaked-special-token chunks seen
    repeat_chunk_run = 0   # consecutive identical short chunks (repetition collapse)
    last_chunk = None      # for repeat-chunk detection

    def show_tool_status(force=False):
        nonlocal tool_status_active, last_tool_status
        now = time.time()
        if not force and tool_status_active and now - last_tool_status < 0.2:
            return
        sys.stdout.write(CLEAR_LINE + f"{CYAN}  ↳ generating tool call{RESET} "
                         f"{DIM}{_tool_call_progress(tcs)}{RESET}")
        sys.stdout.flush()
        tool_status_active = True
        last_tool_status = now

    def clear_tool_status():
        nonlocal tool_status_active
        if tool_status_active:
            sys.stdout.write(CLEAR_LINE)
            sys.stdout.flush()
            tool_status_active = False

    try:
        for chunk in stream:
            if _USER_INTERRUPTED.is_set():
                interrupted = True
                # close() makes the next iteration raise StopIteration / a
                # connection error; we're breaking anyway, but be tidy
                close = getattr(stream, "close", None)
                if close:
                    try:
                        close()
                    except Exception:
                        pass
                break
            # first byte in: kill the spinner, let the real output take this line
            if spinner._t is not None:
                spinner.stop()
            # Capture streaming usage (OpenAI/Z.ai send a final chunk with
            # usage populated when stream_options.include_usage is True).
            # This chunk has an empty choices array, so check before the
            # choices guard below.
            if chunk.usage:
                usage = chunk.usage
            # The final usage-only chunk has an empty choices array — nothing
            # else to do with it.
            if not chunk.choices:
                continue
            choice = chunk.choices[0]
            if choice.finish_reason:
                finish_reasons.append(str(choice.finish_reason))
            d = choice.delta
            # Capture TTFT on the first chunk carrying real output (reasoning,
            # content, or tool calls).
            if t_first is None:
                rc_peek = (getattr(d, "reasoning_content", None) or
                           (getattr(d, "model_extra", None) or {}).get("reasoning_content") or
                           (getattr(d, "model_extra", None) or {}).get("reasoning"))
                if d.content or d.tool_calls or rc_peek:
                    t_first = time.time() - t0
            # llama.cpp attaches a `timings` object to the final chunk — grab it
            # for the stats footer. It's the only place we get real tok/s numbers
            # (streaming `usage` is always null on llama.cpp).
            extra = getattr(chunk, "model_extra", None) or {}
            if "timings" in extra:
                timings = extra["timings"]
            # reasoning models (e.g. MiniMax-M3) stream a separate reasoning_content
            # field before content/tool_calls. Header + dim text, then a blank line
            # so the green "actual response" always lands on its own row (reasoning
            # from the model often doesn't end in \n — without the gap it would
            # run straight into the answer).
            rc = (getattr(d, "reasoning_content", None) or
                  (d.model_extra or {}).get("reasoning_content") or
                  (d.model_extra or {}).get("reasoning"))
            if rc:
                if mode != "think":
                    print(f"{DIM}  ── reasoning ──{RESET}")
                    mode = "think"
                # Sanitize leaked special tokens so they never enter history.
                if _LEAKED_SPECIAL_TOKEN_RE.search(rc):
                    rc = _strip_leaked_tokens(rc)
                    leaked_token_run += 1
                else:
                    leaked_token_run = 0
                if rc.strip():
                    print(f"{DIM}{rc}{RESET}", end="", flush=True)
                    reasoning_parts.append(rc)
                if not content and not tcs:
                    reasoning_only_chars += len(rc)
                    if reasoning_only_start is None:
                        reasoning_only_start = time.time()
                # Cut the stream if the model has collapsed into special-token
                # spam (detected via consecutive leaked-token chunks).
                if LEAKED_TOKEN_RUN_LIMIT > 0 and leaked_token_run >= LEAKED_TOKEN_RUN_LIMIT:
                    print()
                    print(f"{RED}  ⚠ LEAKED-TOKEN COLLAPSE — {leaked_token_run} consecutive "
                          f"special-token chunks; cutting stream now{RESET}")
                    loop_cut = True
                    close = getattr(stream, "close", None)
                    if close:
                        try:
                            close()
                        except Exception:
                            pass
                    break
                # Check the char-based limit (fires when reasoning exceeds
                # REASONING_ONLY_CHAR_LIMIT chars without any content/tool call).
                if (REASONING_ONLY_CHAR_LIMIT > 0 and not content and not tcs
                        and reasoning_only_chars >= REASONING_ONLY_CHAR_LIMIT):
                    print()
                    print(f"{RED}  ⚠ REASONING-ONLY LIMIT HIT — "
                          f"{_abbr(reasoning_only_chars)} reasoning chars with no "
                          f"answer/tool call — cutting stream now{RESET}")
                    loop_cut = True
                    close = getattr(stream, "close", None)
                    if close:
                        close()
                    break
                # Check the time-based limit (fires when reasoning-only streaming
                # has run for > REASONING_ONLY_TIME_LIMIT seconds).  At slow
                # generation speeds the char limit may never be reached, so this
                # wall-clock guard prevents a single reasoning spiral from eating
                # the entire timeout budget.
                if (REASONING_ONLY_TIME_LIMIT > 0 and not content and not tcs
                        and reasoning_only_start is not None
                        and time.time() - reasoning_only_start >= REASONING_ONLY_TIME_LIMIT):
                    elapsed_reason = time.time() - reasoning_only_start
                    print()
                    print(f"{RED}  ⚠ REASONING-ONLY TIME LIMIT — "
                          f"{elapsed_reason:.0f}s of reasoning with no "
                          f"answer/tool call — cutting stream now{RESET}")
                    loop_cut = True
                    close = getattr(stream, "close", None)
                    if close:
                        close()
                    break
            if d.content:
                clear_tool_status()
                if mode == "think":
                    # close out the reasoning block; newline guarantees the green
                    # answer starts on a fresh line below the dim text
                    print()  # end the current reasoning line
                    print(f"{DIM}  ──────────────{RESET}")
                print(f"{GREEN}", end="")
                mode = "say"
                # Sanitize leaked special tokens before display + history.
                chunk = d.content
                if _LEAKED_SPECIAL_TOKEN_RE.search(chunk):
                    chunk = _strip_leaked_tokens(chunk)
                    leaked_token_run += 1
                else:
                    leaked_token_run = 0
                # Repetition-collapse detection: short identical chunks
                # repeated in a row (e.g. "import" over and over).
                if REPEAT_CHUNK_LIMIT > 0 and chunk and len(chunk) <= 40:
                    if chunk == last_chunk:
                        repeat_chunk_run += 1
                    else:
                        repeat_chunk_run = 1
                        last_chunk = chunk
                else:
                    repeat_chunk_run = 0
                if chunk.strip():
                    print(chunk, end="", flush=True)
                    content.append(chunk)
                    streamed_chars += len(chunk)
                # Cut the stream on a special-token collapse or repetition loop.
                if ((LEAKED_TOKEN_RUN_LIMIT > 0 and leaked_token_run >= LEAKED_TOKEN_RUN_LIMIT)
                        or (REPEAT_CHUNK_LIMIT > 0 and repeat_chunk_run >= REPEAT_CHUNK_LIMIT)):
                    kind = ("special-token collapse" if leaked_token_run >= LEAKED_TOKEN_RUN_LIMIT
                            else "repetition collapse")
                    print(RESET)
                    print(f"{RED}  ⚠ {kind.upper()} — cutting stream now; "
                          f"discarding degenerated output{RESET}")
                    loop_cut = True
                    close = getattr(stream, "close", None)
                    if close:
                        try:
                            close()
                        except Exception:
                            pass
                    break
            for tc in (d.tool_calls or []):
                # if we were mid-reasoning when tools kicked in, close it out so
                # the cyan tool box (which starts with its own \n) gets a clean line
                if mode == "think":
                    print()
                    print(f"{DIM}  ──────────────{RESET}")
                    mode = None
                elif mode == "say":
                    print(RESET)
                    mode = None
                s = tcs.setdefault(tc.index, {"id": "", "name": "", "args": ""})
                if tc.id:
                    s["id"] = tc.id
                if tc.function and tc.function.name:
                    s["name"] = tc.function.name
                if tc.function and tc.function.arguments:
                    s["args"] += tc.function.arguments
                    streamed_chars += len(tc.function.arguments)
                show_tool_status()
    except (APIError, APIConnectionError, httpx.HTTPError) as e:
        stream_error = e
        close = getattr(stream, "close", None)
        if close:
            try:
                close()
            except Exception:
                pass
    finally:
        spinner.stop()
        clear_tool_status()
        # Disarm the watcher: it stops reading stdin, restores termios in its
        # own inner finally, and parks on _INTERRUPT_ARMED until next turn.
        # We MUST wait for _INTERRUPT_RESTORED before returning — otherwise the
        # watcher's termios restore is asynchronous (it may still be in
        # select(0.1) when we return) and the chatbox can capture our raw-mode
        # termios as its "old", leaving the terminal stuck in VMIN=0/ISIG-off.
        # The watcher's inner select has a 0.1s timeout, so restore typically
        # completes in <0.1s; the 0.5s timeout is a safety net for the rare
        # case where the turn was too short for the watcher to even arm (in
        # which case termios was never changed and there's nothing to restore).
        _INTERRUPT_ARMED.clear()
        _INTERRUPT_RESTORED.wait(timeout=0.5)
        _USER_INTERRUPTED.clear()
    if stream_error is not None:
        if mode in ("think", "say"):
            print()
        if mode == "think":
            print(f"{DIM}  ──────────────{RESET}")
        print(RESET)
        if isinstance(stream_error, httpx.TimeoutException):
            print(f"{YELLOW}  ✂ STREAM TIMEOUT — server stopped delivering chunks; "
                  f"discarded partial output and returned to chat input{RESET}")
            print(f"{DIM}    {type(stream_error).__name__}: {stream_error}{RESET}")
            return TURN_DONE
        retry_limit = max(0, MALFORMED_STREAM_RETRY_LIMIT)
        if malformed_stream_cut_count >= retry_limit:
            print(f"{RED}  ✗ malformed stream from model/API after "
                  f"{malformed_stream_cut_count} recoveries — waiting for user input{RESET}")
            print(f"{DIM}    {type(stream_error).__name__}: {stream_error}{RESET}")
            return TURN_DONE
        print(f"{YELLOW}  ✂ MALFORMED STREAM — discarded partial output/tool args; "
              f"retrying cleanly ({malformed_stream_cut_count + 1}/{retry_limit}){RESET}")
        _nudge_current_user_turn(
            messages,
            "Your previous streamed response became malformed before it completed. "
            "Discard it. Retry the same task from the current conversation state, "
            "but emit either valid tool calls or a concise final answer only.")
        return TURN_STREAM_CUT
    # reasoning-only turn (no content, no tool_calls) — close out the block so
    # the stats footer doesn't run straight into the dim reasoning text
    if mode == "think":
        print()
        print(f"{DIM}  ──────────────{RESET}")
    print(RESET)
    text = "".join(content)
    # Final defense-in-depth sanitization: strip any leaked special tokens
    # that slipped through the per-chunk stripping (e.g. a single token at the
    # very end of a stream that got cut before the run limit triggered).
    text = _strip_leaked_tokens(text)
    elapsed = time.time() - t0

    if interrupted:
        print(f"{YELLOW}  ↳ interrupted by user (Esc) after {elapsed:4.1f}s, "
              f"{len(content)} chars streamed{RESET}")
        # Discard partial content — it's almost certainly a half-formed
        # sentence / tool-call args. Append a synthetic user turn so the
        # model has context for what just happened, then return False so the
        # REPL drops to the prompt instead of looping into another turn.
        messages.append({"role": "user", "content":
            "[User interrupted your previous response with Esc. "
            "Acknowledge briefly and wait for their next message.]"})
        return TURN_DONE

    if not loop_cut and not text.strip() and not tcs and reasoning_only_chars > 0:
        loop_cut = True

    if loop_cut:
        # A leaked-token or repetition collapse is a degeneration, not a
        # reasoning stall — the model was emitting real content (or reasoning)
        # but it degenerated into gibberish.  Treat it like a malformed stream:
        # discard the degenerated output and nudge the model to produce clean
        # output from the current conversation state.  This runs BEFORE the
        # reasoning-stall logic so we don't mis-classify a collapse as a stall.
        is_leak_collapse = (leaked_token_run >= LEAKED_TOKEN_RUN_LIMIT
                           or repeat_chunk_run >= REPEAT_CHUNK_LIMIT)
        if is_leak_collapse and not forced_final:
            collapse_type = ("special-token collapse" if leaked_token_run >= LEAKED_TOKEN_RUN_LIMIT
                             else "repetition collapse")
            print(f"{YELLOW}  ✂ {collapse_type.upper()} — discarding degenerated "
                  f"output and nudging clean recovery{RESET}")
            # Do NOT feed the degenerated reasoning/content back — it's spam
            # that would reinforce the collapse on the next turn.  Nudge fresh.
            _nudge_current_user_turn(messages,
                "Your previous output degenerated into a repetition collapse "
                "(repeated special tokens or identical chunks). This is a known "
                "failure mode — discard that output entirely. Re-orient: state "
                "what you were trying to do in one sentence, then emit the next "
                "concrete tool call (write_file, run_bash, or edit_file). Do not "
                "continue the previous stream. Start fresh.")
            return TURN_STREAM_CUT
        retry_limit = max(0, REASONING_ONLY_RETRY_LIMIT)
        cut_count = reasoning_loop_cut_count
        reasoning_secs = (time.time() - reasoning_only_start) if reasoning_only_start else 0
        if reasoning_only_start and REASONING_ONLY_TIME_LIMIT > 0 and \
                reasoning_secs >= REASONING_ONLY_TIME_LIMIT:
            loop_detail = (f"{reasoning_secs:.0f}s of reasoning "
                           f"({_abbr(reasoning_only_chars)} chars) without content/tool calls")
        else:
            loop_detail = (
                f"{_abbr(reasoning_only_chars)} reasoning chars without content/tool calls")
        if forced_final:
            print(f"{RED}  ✂ FORCED FINAL ANSWER FAILED — got {loop_detail}; "
                  f"returning to chat input{RESET}")
            return TURN_DONE
        if cut_count >= retry_limit:
            if retry_limit == 0:
                # Failsafes deliberately disabled (e.g. benchmark mode where
                # we let the model think as long as it wants).  The model
                # produced only reasoning and no action — that's an honest
                # model-capability signal, not a harness rescue scenario.
                print(f"{DIM}  • reasoning-only turn ended naturally "
                      f"({loop_detail}); failsafes disabled, ending turn{RESET}")
            else:
                print(f"{RED}  ✂ REASONING-ONLY RESCUE FAILED — gave up after "
                      f"{cut_count} stall "
                      f"attempt{'s' if cut_count != 1 else ''} "
                      f"× {loop_detail}; waiting for user input{RESET}")
            return TURN_DONE
        if reasoning_only_start and REASONING_ONLY_TIME_LIMIT > 0 and \
                reasoning_secs >= REASONING_ONLY_TIME_LIMIT:
            cut_msg = (f"{reasoning_secs:.0f}s of reasoning with no "
                       f"answer/tool call (limit {REASONING_ONLY_TIME_LIMIT}s)")
        else:
            cut_msg = (f"{_abbr(reasoning_only_chars)} reasoning chars with no "
                       f"answer/tool call (limit {_abbr(REASONING_ONLY_CHAR_LIMIT)})")

        # --- Feed captured reasoning back so the model can *continue* -----
        # The old behavior discarded reasoning_content entirely and appended a
        # nudge to the existing user message.  The model saw the same starting
        # state every time → same reasoning → same stall → literal loop.  Now
        # we store the reasoning as an assistant message (truncated to a cap so
        # a 46K-char spiral doesn't blow up context) and add the nudge as a NEW
        # user message.  The model sees: [task] → [my reasoning] → [now act]
        # and can pick up where it left off instead of starting from scratch.
        reasoning_text = "".join(reasoning_parts)
        if REASONING_CAPTURE_CHARS > 0 and reasoning_text.strip():
            cap = REASONING_CAPTURE_CHARS
            if len(reasoning_text) > cap:
                # Keep the tail — it has the model's most recent conclusions.
                reasoning_text = ("[…earlier reasoning truncated…]\n"
                                  + reasoning_text[-cap:])
            messages.append({"role": "assistant", "content": reasoning_text})
            print(f"{DIM}  ↳ captured {len(reasoning_text)} chars of reasoning "
                  f"for feedback{RESET}")

        # Two-tier recovery:
        #   • First (and any intermediate) stall: nudge the model to ACT —
        #     emit a tool call, write a file — with FULL tool access. This
        #     is critical for coding tasks where the model got stuck planning
        #     but never executed anything.  Returns TURN_STALL so the loop
        #     retries with normal tools (not the restricted FINAL_ANSWER_TOOL).
        #   • Last stall (cut_count == retry_limit - 1): force a final answer
        #     with restricted tools (FINAL_ANSWER_TOOL only).  This is the
        #     fallback when repeated nudges still don't break the reasoning
        #     loop, so we at least get a visible answer instead of a hang.
        is_last_retry = (cut_count == retry_limit - 1)
        if is_last_retry:
            print(f"{YELLOW}  ✂ REASONING-ONLY STALL — {cut_msg}; forcing final "
                  f"answer (attempt {cut_count + 1}/{retry_limit}){RESET}")
            if reasoning_text.strip():
                messages.append({"role": "user", "content":
                    "[Runtime note: Your reasoning has been preserved as your "
                    "previous message. Do not continue private reasoning. Use "
                    "the final_answer tool if available, or return a complete "
                    "visible answer now based on what you reasoned. Keep it "
                    "concise.]"})
            else:
                _nudge_current_user_turn(messages, FORCED_FINAL_NUDGE)
            return TURN_FORCE_FINAL
        print(f"{YELLOW}  ✂ REASONING-ONLY STALL — {cut_msg}; nudging to act "
              f"(attempt {cut_count + 1}/{retry_limit}){RESET}")
        if reasoning_text.strip():
            messages.append({"role": "user", "content":
                "[Runtime note: You were thinking about the task above. "
                "Your reasoning has been preserved as your previous message. "
                "Now act on it — do NOT continue reasoning. Concretely:\n"
                "  • write_file: create the deliverable file the task expects "
                "(e.g. /app/solution.py, /app/main.rs). Put your planned approach "
                "into actual code now, even as a first draft.\n"
                "  • run_bash: execute it to see what happens.\n"
                "  • edit_file / write_file: fix what the run reported and re-run.\n"
                "An imperfect file you iterate on beats a perfect plan that never "
                "gets written. Emit the write_file or run_bash tool call now.]"})
        else:
            _nudge_current_user_turn(messages, REASONING_STALL_NUDGE)
        return TURN_STALL

    # stats footer — prefer llama.cpp timings if present; otherwise fall back
    # to the standard streaming `usage` object (OpenAI, Z.ai, etc.); otherwise
    # fall back to wall-clock only. TTFT (time to first token) is shown when
    # available — for local it comes from llama.cpp timings, for remote we
    # measure it client-side.
    if _METRICS is not None:
        _METRICS.add(_normalize_usage(usage, timings, fallback_chars=streamed_chars))
    if timings and timings.get("predicted_n"):
        prompt_n = timings.get("prompt_n", 0)
        if prompt_n:
            _LAST_PROMPT_TOKENS = prompt_n
        gen_n = timings["predicted_n"]
        tps = timings.get("predicted_per_second", 0)
        ctx = _ctx_field(prompt_n)
        prompt_ms = timings.get("prompt_ms", 0)
        ttft = (prompt_ms / 1000.0) if prompt_ms else t_first
        parts = [f"{gen_n} tok", f"{tps:5.1f} tok/s", ctx]
        if ttft:
            parts.append(f"{ttft*1000:4.0f}ms ttft")
        parts.append(f"{elapsed:4.1f}s wall")
        print(f"{DIM}  └ {' · '.join(parts)}{RESET}")
    elif usage and (usage.completion_tokens or 0):
        gen_n = usage.completion_tokens or 0
        prompt_n = usage.prompt_tokens or 0
        if prompt_n:
            _LAST_PROMPT_TOKENS = prompt_n
        tps = gen_n / elapsed if elapsed > 0 else 0
        ctx = _ctx_field(prompt_n)
        parts = [f"{gen_n} tok", f"{tps:5.1f} tok/s", ctx]
        if t_first:
            parts.append(f"{t_first*1000:4.0f}ms ttft")
        parts.append(f"{elapsed:4.1f}s wall")
        print(f"{DIM}  └ {' · '.join(parts)}{RESET}")
    elif streamed_chars > 0:
        gen_n = max(1, streamed_chars // 4)
        tps = gen_n / elapsed if elapsed > 0 else 0
        print(f"{DIM}  └ ≈{_abbr(gen_n)} tok · {tps:5.1f} tok/s · "
              f"{elapsed:4.1f}s wall{RESET}")
    elif text or tcs:
        print(f"{DIM}  └ {elapsed:4.1f}s wall{RESET}")

    if forced_final and tcs:
        ordered = [tcs[i] for i in sorted(tcs)]
        for c in ordered:
            if c["name"] != "final_answer":
                continue
            try:
                args = json.loads(c["args"] or "{}")
            except json.JSONDecodeError:
                args = {}
            answer = str(args.get("answer") or "").strip()
            status = str(args.get("status") or "answered").strip()
            if answer:
                print(f"{GREEN}{answer}{RESET}")
                messages.append({"role": "assistant", "content": answer})
                return TURN_DONE
            print(f"{RED}  ✂ FORCED FINAL ANSWER EMPTY — status={status or 'unknown'}; "
                  f"returning to chat input{RESET}")
            return TURN_DONE
        names = ", ".join(c["name"] or "tool" for c in ordered)
        print(f"{RED}  ✂ FORCED FINAL ANSWER FAILED — model emitted {names}; "
              f"returning to chat input{RESET}")
        return TURN_DONE

    if tcs:  # native tool-calling path
        ordered = [tcs[i] for i in sorted(tcs)]
        parsed_args = []
        parse_error = None
        for c in ordered:
            try:
                parsed_args.append(json.loads(c["args"] or "{}"))
            except json.JSONDecodeError as e:
                parse_error = (c, e)
                break
        if parse_error is not None:
            c, err = parse_error
            detail = _tool_args_parse_error(c, err, finish_reasons)
            retry_limit = max(0, MALFORMED_STREAM_RETRY_LIMIT)
            if malformed_stream_cut_count >= retry_limit:
                print(f"{RED}  ✗ malformed tool call after "
                      f"{malformed_stream_cut_count} recoveries — waiting for user input{RESET}")
                print(f"{DIM}    {detail}{RESET}")
                return TURN_DONE
            print(f"{YELLOW}  ✂ MALFORMED TOOL CALL — {detail}; "
                  f"discarded partial tool args and retrying cleanly "
                  f"({malformed_stream_cut_count + 1}/{retry_limit}){RESET}")
            _nudge_current_user_turn(
                messages,
                "Your previous native tool call had malformed or truncated JSON "
                "arguments, so it was discarded before execution. Retry the same "
                "task from the current conversation state, but emit a complete, "
                "valid tool call with all required arguments. For large file "
                "writes, emit only the tool call and skip any explanatory prose.")
            return TURN_STREAM_CUT
        messages.append({"role": "assistant", "content": text or None, "tool_calls": [
            {"id": c["id"], "type": "function", "function": {"name": c["name"], "arguments": c["args"]}}
            for c in ordered]})
        esc_action = None
        for idx, (c, args) in enumerate(zip(ordered, parsed_args)):
            try:
                result = run_tool(c["name"], args)
            except _EscToChat as e:
                esc_action = e.args[0] if e.args else c["name"]
                # Record this call as cancelled, and fill in results for any
                # remaining tool_calls so the message history stays valid
                # (every assistant tool_call needs a matching tool result,
                # or the chat template rejects the context on the next turn).
                messages.append({"role": "tool", "tool_call_id": c["id"],
                                 "content": "CANCELLED by user (Esc) — returned to chat input"})
                for c2 in ordered[idx + 1:]:
                    messages.append({"role": "tool", "tool_call_id": c2["id"],
                                     "content": "SKIPPED (user pressed Esc at an earlier approval)"})
                break
            messages.append({"role": "tool", "tool_call_id": c["id"], "content": result})
        if esc_action is not None:
            print(f"{YELLOW}  ↳ escaped approval of {esc_action!r} — back to chat input{RESET}")
            messages.append({"role": "user", "content":
                "[User pressed Esc at a tool approval prompt and returned to chat to "
                "add more input. Acknowledge briefly and wait for their next message.]"})
            return TURN_ESC
        return TURN_TOOL

    calls = parse_text_calls(text)  # text-fallback path
    if calls:
        messages.append({"role": "assistant", "content": text})
        obs = []
        esc_action = None
        for n, a in calls:
            try:
                obs.append(f"Observation ({n}): {run_tool(n, a)}")
            except _EscToChat as e:
                esc_action = e.args[0] if e.args else n
                obs.append(f"Observation ({n}): CANCELLED by user (Esc)")
                break
        if esc_action is not None:
            obs.append("[User pressed Esc at a tool approval prompt and returned to chat "
                       "to add more input. Acknowledge briefly and wait for their next message.]")
            print(f"{YELLOW}  ↳ escaped approval of {esc_action!r} — back to chat input{RESET}")
            messages.append({"role": "user", "content": "\n".join(obs)})
            return TURN_ESC
        messages.append({"role": "user", "content": "\n".join(obs)})
        return TURN_TOOL

    if forced_final and text.strip() and "length" in finish_reasons:
        print(f"{YELLOW}  ✂ FORCED FINAL ANSWER HIT TOKEN LIMIT — saved partial answer "
              f"with truncation marker{RESET}")
        messages.append({
            "role": "assistant",
            "content": text.rstrip() + "\n\n[Truncated by token limit before completion.]",
        })
        return TURN_DONE

    if not text.strip():
        # Silent stall: the model emitted nothing. In a real agentic loop this is
        # almost never a legitimate "done" — it usually follows a short/empty
        # tool result (the model "knows" the answer but produces no tokens), and
        # returning TURN_DONE here drops the user to the chat prompt mid-task with
        # no explanation. Surface it so the loop can nudge + auto-continue instead.
        #
        # Never auto-continue when:
        #   • forced_final — an empty forced answer already means "give up" (the
        #     forced-final block above printed its own message); stopping is right.
        #   • EMPTY_TURN_RETRY_LIMIT <= 0 — the user opted out; empty = done.
        #   • recovery already exhausted the budget — escalate to a forced answer
        #     (mirrors the reasoning-loop path) instead of spinning forever.
        if (not forced_final and EMPTY_TURN_RETRY_LIMIT > 0
                and empty_turn_count < EMPTY_TURN_RETRY_LIMIT):
            dangling = _last_is_dangling_tool(messages)
            print(f"{YELLOW}  ✂ EMPTY TURN — no content and no tool call"
                  f"{'; dangling tool result awaits a follow-up' if dangling else ''}"
                  f"; nudging the model to continue "
                  f"({empty_turn_count + 1}/{EMPTY_TURN_RETRY_LIMIT}){RESET}")
            _nudge_current_user_turn(messages, EMPTY_TURN_NUDGE)
            return TURN_EMPTY
        if not forced_final and EMPTY_TURN_RETRY_LIMIT > 0:
            print(f"{YELLOW}  ✂ EMPTY TURN — no content after {empty_turn_count} nudge"
                  f"{'s' if empty_turn_count != 1 else ''}; forcing a final answer{RESET}")
            _nudge_current_user_turn(messages, FORCED_FINAL_NUDGE)
            return TURN_FORCE_FINAL
        return TURN_DONE
    messages.append({"role": "assistant", "content": text})
    return TURN_DONE


# Maximum number of model turns (tool calls + retries) per session before the
# loop breaks.  The real bound on work is the per-task timeout (900-3600s);
# each turn costs at minimum a few seconds (API round-trip + generation + tool
# exec), so even 200 turns in a 900s budget means turns averaging <4.5s — that's
# fast spinning, not productive work.  Individual pathologies (reasoning stalls,
# empty turns, malformed streams) already have their own retry limits.  This cap
# is just a total-work backstop so the loop can't run away *inside* the timeout.
# Overridable via env var.  Set very high (e.g. 99999) to effectively disable.
MAX_MODEL_TURNS = _env_int("MINION_MAX_MODEL_TURNS", 200)

def _run_model_turn_loop(messages, force_final=False, recovery_sampling=False,
                         on_step=None):
    steps = 0
    reasoning_loop_cuts = 0
    malformed_stream_cuts = 0
    empty_turn_cuts = 0
    while steps < MAX_MODEL_TURNS:  # cap runaway tool/retry loops
        status = model_turn(messages, reasoning_loop_cuts, malformed_stream_cuts,
                            empty_turn_count=empty_turn_cuts,
                            forced_final=force_final,
                            recovery_sampling=recovery_sampling)
        if on_step is not None:
            on_step()
        force_final = False
        recovery_sampling = False
        if status == TURN_DONE:
            break
        if status == TURN_ESC:
            break  # user pressed Esc at an approval → drop to chat input
        steps += 1
        if status == TURN_STREAM_CUT:
            malformed_stream_cuts += 1
            recovery_sampling = True
            continue
        if status == TURN_EMPTY:
            empty_turn_cuts += 1
            recovery_sampling = True  # more entropy to escape the empty-output attractor
            continue
        if status == TURN_STALL:
            reasoning_loop_cuts += 1
            recovery_sampling = True
            # Full tools retained — the model was nudged to ACT (emit a tool
            # call / write a file) rather than produce a premature final
            # answer.  force_final stays False so model_turn() opens the
            # stream with the complete tool set, not [FINAL_ANSWER_TOOL].
            continue
        if status == TURN_FORCE_FINAL:
            reasoning_loop_cuts += 1
            empty_turn_cuts = 0  # handed off to the forced-final path; reset
            force_final = True
            recovery_sampling = True
            continue
        if status == TURN_TOOL:
            # A tool call is real progress for malformed-stream and empty-turn
            # recovery — those counters are about output pathologies, and a
            # successful tool call means the model is producing again.
            #
            # But reasoning_loop_cuts is different: a single trivial tool call
            # between two 36K-char reasoning spirals does NOT mean the model
            # has recovered from its tendency to over-think.  Resetting the
            # counter here lets the model stall → tool → stall → tool → …
            # forever, and the forced-final-answer safety net (attempt 2/2)
            # never fires.  In headless mode (benchmarks) this burns the
            # entire timeout.  Keep the counter so cumulative stalls escalate.
            malformed_stream_cuts = 0
            empty_turn_cuts = 0
            continue
    # Loop exited because steps >= MAX_MODEL_TURNS.  In interactive mode the
    # user can simply type another instruction; in headless mode (benchmarks)
    # the session would end with whatever partial work exists.  Give the model
    # one last forced-final turn so it can produce a visible answer from the
    # state so far — but only if we haven't already tried (force_final was
    # already consumed by the loop above).
    if steps >= MAX_MODEL_TURNS and not force_final:
        print(f"{YELLOW}  ⚂ MODEL TURN LIMIT ({MAX_MODEL_TURNS}) — "
              f"forcing a final answer from current state{RESET}")
        model_turn(messages, reasoning_loop_cuts, malformed_stream_cuts,
                   empty_turn_count=empty_turn_cuts,
                   forced_final=True,
                   recovery_sampling=recovery_sampling)


# --- multi-line chatbox input ---------------------------------------------
# Replaces the bare `input()` prompt with a framed, multi-line editor:
#   • Enter submits, Alt+Enter (or Ctrl+J) inserts a newline
#   • Paste (bracketed-paste mode) inserts its text verbatim — newlines stay,
#     a trailing newline at the end of paste is stripped so pasting never
#     accidentally submits
#   • Up/Down navigate history, Left/Right move within the current line,
#     Home/End jump to line start/end, Ctrl+U clears the line, Ctrl+C cancels
#   • Long lines word-wrap visually inside the box; the buffer stays one
#     logical string (newlines preserved) so the model sees the real text
# Falls back to plain `input()` when stdin/stdout is not a TTY.

def _chatbox_fallback(prompt):
    """Plain `input()` fallback used when raw terminal mode isn't usable. Single-line only;
    newlines must be typed as the literal `\\n` (rare in practice)."""
    try:
        return input(prompt)
    except (EOFError, KeyboardInterrupt):
        raise


def _char_width(ch):
    if unicodedata.east_asian_width(ch) in ('W', 'F'):
        return 2
    return 1


def _str_width(s):
    return sum(_char_width(c) for c in s)


def _str_slice_by_width(s, max_width):
    """Return the longest prefix of s that fits within max_width display columns."""
    w = 0
    for i, ch in enumerate(s):
        cw = _char_width(ch)
        if w + cw > max_width:
            return s[:i]
        w += cw
    return s


def _raw_read_key(fd):
    b = os.read(fd, 1)
    if not b:
        return ""
    first = b[0]
    if first < 0x80:
        return b.decode("ascii")
    elif first < 0xC0:
        return b.decode("utf-8", "replace")
    elif first < 0xE0:
        need = 1
    elif first < 0xF0:
        need = 2
    else:
        need = 3
    for _ in range(need):
        extra = os.read(fd, 1)
        if not extra:
            break
        b += extra
    return b.decode("utf-8", "replace")


def _raw_read_available(fd, timeout=0.02):
    raw = bytearray()
    while True:
        ready, _, _ = select.select([fd], [], [], timeout)
        if not ready:
            break
        raw += os.read(fd, 1)
        timeout = 0
    return raw.decode("utf-8", "replace")


def _chatbox_raw(initial="", history=None):
    """Normal-scrollback multi-line editor.

    This does not enter the alternate screen. The prompt, streamed model output,
    tool confirmations, and the next prompt all stay in one terminal mode, which
    avoids garbling the REPL after submit.
    """
    history = history or []
    fd = sys.stdin.fileno()
    old = termios.tcgetattr(fd)
    new = old[:]
    new[3] &= ~(termios.ECHO | termios.ICANON)
    new[0] &= ~termios.ICRNL
    new[6][termios.VMIN] = 1
    new[6][termios.VTIME] = 0

    buf = initial.split("\n") if initial else [""]
    row, col = len(buf) - 1, len(buf[-1])
    h_idx = len(history)
    rendered_lines = 0
    cursor_line = 0

    def move_up(n):
        return f"\x1b[{n}A" if n > 0 else ""

    def move_down(n):
        return f"\x1b[{n}B" if n > 0 else ""

    def move_right(n):
        return f"\x1b[{n}C" if n > 0 else ""

    def build_visual(inner_w):
        visual = []
        for bi, line in enumerate(buf):
            if not line:
                visual.append((bi, 0, ""))
                continue
            start = 0
            while start < len(line):
                seg = _str_slice_by_width(line[start:], inner_w)
                visual.append((bi, start, seg))
                start += len(seg)
            if col == len(line) and row == bi and _str_width(line[visual[-1][1]:]) == inner_w:
                visual.append((bi, len(line), ""))
        return visual

    def render():
        nonlocal rendered_lines, cursor_line
        width = shutil.get_terminal_size((80, 24)).columns
        box_w = max(20, min(width - 2, 100))
        inner_w = max(1, box_w - 2)
        visual = build_visual(inner_w)

        cur_vrow = 0
        cur_vcol = 0
        for i, (bi, start, seg) in enumerate(visual):
            if bi != row:
                continue
            if start <= col <= start + len(seg):
                cur_vrow = i
                cur_vcol = _str_width(seg[:col - start])
                break

        hints = "Enter send · Alt+Enter / ^J newline · ^C cancel"
        max_hints = max(0, box_w - 11)
        if _str_width(hints) > max_hints:
            hints = _str_slice_by_width(hints, max_hints - 1) + "…" if max_hints > 1 else "…"
        top_fill = max(0, box_w - 11 - _str_width(hints))
        stats = f"{len(buf)} line{'s' if len(buf) != 1 else ''} · {sum(len(l) for l in buf)} chars"
        max_stats = max(0, box_w - 6)
        if _str_width(stats) > max_stats:
            stats = _str_slice_by_width(stats, max_stats - 1) + "…" if max_stats > 1 else "…"
        bot_fill = max(0, box_w - 4 - _str_width(stats))

        lines = [
            f"{DIM}╭─ {RESET}{CYAN}you{RESET}{DIM} · {hints} {'─' * top_fill}╮{RESET}",
            *[f"{DIM}│{RESET}{seg}{' ' * (inner_w - _str_width(seg))}{DIM}│{RESET}" for _, _, seg in visual],
            f"{DIM}╰─ {stats}{' ' * bot_fill}╯{RESET}",
        ]

        redraw_rows = max(rendered_lines, len(lines))
        out = "\r" + move_up(cursor_line)
        for i in range(redraw_rows):
            out += "\x1b[2K"
            if i < len(lines):
                out += lines[i]
            if i != redraw_rows - 1:
                out += "\n"
        out += "\r" + move_up((redraw_rows - 1) - (cur_vrow + 1)) + move_right(cur_vcol + 1)
        sys.stdout.write(out)
        sys.stdout.flush()
        rendered_lines = len(lines)
        cursor_line = cur_vrow + 1

    def finish():
        sys.stdout.write("\r" + move_down((rendered_lines - 1) - cursor_line) + "\n")
        sys.stdout.write("\x1b[?25h\x1b[?2004l")
        sys.stdout.flush()

    def insert_text(s):
        nonlocal row, col
        if not s:
            return
        parts = s.split("\n")
        cur = buf[row]
        tail = cur[col:]
        if len(parts) == 1:
            buf[row] = cur[:col] + parts[0] + tail
            col += len(parts[0])
            return
        buf[row] = cur[:col] + parts[0]
        new_lines = list(parts[1:-1]) + [parts[-1] + tail]
        buf[row + 1:row + 1] = new_lines
        row += len(new_lines)
        col = len(parts[-1])

    def backspace():
        nonlocal row, col
        if col > 0:
            buf[row] = buf[row][:col - 1] + buf[row][col:]
            col -= 1
        elif row > 0:
            prev = buf[row - 1]
            col = len(prev)
            buf[row - 1] = prev + buf[row]
            del buf[row]
            row -= 1

    def delete_forward():
        line = buf[row]
        if col < len(line):
            buf[row] = line[:col] + line[col + 1:]
        elif row < len(buf) - 1:
            buf[row] = line + buf[row + 1]
            del buf[row + 1]

    def load_from_history(hist_text):
        nonlocal row, col
        buf[:] = hist_text.split("\n") if hist_text else [""]
        row = len(buf) - 1
        col = len(buf[-1])

    def handle_escape(seq):
        nonlocal row, col, h_idx
        if seq.startswith("[200~"):
            paste = seq[5:]
            while "\x1b[201~" not in paste:
                paste += _raw_read_key(fd)
            paste = paste.split("\x1b[201~", 1)[0]
            if paste.endswith("\n") or paste.endswith("\r"):
                paste = paste[:-1]
            insert_text(paste)
            return
        if seq in ("[A", "OA"):
            if history and h_idx > 0:
                h_idx -= 1
                load_from_history(history[h_idx])
            return
        if seq in ("[B", "OB"):
            if history and h_idx < len(history):
                h_idx += 1
                load_from_history("" if h_idx == len(history) else history[h_idx])
            return
        if seq in ("[C", "OC"):
            if col < len(buf[row]):
                col += 1
            elif row < len(buf) - 1:
                row += 1
                col = 0
            return
        if seq in ("[D", "OD"):
            if col > 0:
                col -= 1
            elif row > 0:
                row -= 1
                col = len(buf[row])
            return
        if seq in ("[H", "OH"):
            col = 0
            return
        if seq in ("[F", "OF"):
            col = len(buf[row])
            return
        if seq == "[3~":
            delete_forward()
            return
        if seq in ("\r", "\n"):
            insert_text("\n")

    termios.tcsetattr(fd, termios.TCSADRAIN, new)
    sys.stdout.write("\x1b[?25h\x1b[?2004h\n")
    sys.stdout.flush()
    try:
        render()
        while True:
            c = _raw_read_key(fd)
            if c == "\x1b":
                handle_escape(_raw_read_available(fd))
            elif c == "\r":
                finish()
                return "\n".join(buf)
            elif c == "\n":
                insert_text("\n")
            elif c == "\x03":
                raise KeyboardInterrupt
            elif c == "\x04":
                if not any(buf):
                    raise EOFError
            elif c in ("\x7f", "\x08"):
                backspace()
            elif c == "\x15":
                buf[row] = ""
                col = 0
            elif c >= " " or c == "\t":
                insert_text(c)
            render()
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old)
        sys.stdout.write("\x1b[?25h\x1b[?2004l")
        sys.stdout.flush()


def read_multiline(initial="", history=None):
    """Public entry point. Returns the entered text, or '' on empty submit.
    Raises EOFError / KeyboardInterrupt to match `input()`'s contract."""
    if not sys.stdin.isatty() or not sys.stdout.isatty():
        # Fallback path — single-line input. Same UX as before this feature.
        return _chatbox_fallback(f"{CYAN}you ›{RESET} ")
    try:
        return _chatbox_raw(initial, history)
    except (EOFError, KeyboardInterrupt):
        raise
    except (OSError, termios.error):
        return _chatbox_fallback(f"{CYAN}you ›{RESET} ")


# --- repl -------------------------------------------------------------------
def _banner():
    """Banner shown at startup and after source/approval changes."""
    sep = f"{DIM} · {RESET}"
    parts = [f"{BOLD}minion{RESET}", f"{CYAN}{MODEL}{RESET}"]
    if len(SOURCES) > 1:
        parts.append(f"{MAGENTA}{ACTIVE.name}{RESET}")
    if YOLO:
        parts.append(f"{GREEN}yolo{RESET}")
    elif APPROVE_LEVEL is None:
        parts.append(f"{YELLOW}prompt:all{RESET}")
    elif APPROVE_LEVEL == "high":
        parts.append(f"{GREEN}auto:high{RESET}")
    elif APPROVE_LEVEL == "medium":
        parts.append(f"{YELLOW}auto:medium{RESET}")
    else:
        parts.append(f"{DIM}auto:low{RESET}")
    parts.append(f"{DIM}{client.base_url}{RESET}")
    return sep.join(parts)


# --- one-shot (non-interactive) mode ----------------------------------------
# Activated by --prompt-file (or -f). Reads a task instruction from a file (or
# stdin if the path is "-"), runs a single agent turn loop, and exits — no REPL,
# no banner, no session save. Designed for headless benchmark harnesses that
# pipe a task in and wait for the agent to finish. All of minion's tool-use,
# recovery, and step-cap machinery works exactly as in interactive mode; the
# only difference is that input comes from a file and the process exits after.

def _run_one_shot():
    """Non-interactive entry point for --prompt-file. Reads the instruction,
    runs _run_model_turn_loop once, and exits with a status code."""
    if _PROMPT_FILE is None:
        return False  # not in one-shot mode; caller should proceed to REPL
    # --- read the instruction --------------------------------------------------
    if _PROMPT_FILE == "-":
        instruction = sys.stdin.read()
    else:
        try:
            with open(_PROMPT_FILE, "r", encoding="utf-8") as f:
                instruction = f.read()
        except OSError as e:
            sys.stderr.write(f"minion: cannot read --prompt-file {e.filename!r}: {e}\n")
            sys.exit(2)
    instruction = instruction.strip()
    if not instruction:
        sys.stderr.write("minion: --prompt-file is empty — nothing to do\n")
        sys.exit(2)

    # --- resolve the active source ---------------------------------------------
    # switch_source was already called at import time (line ~898) for the
    # --source flag or the first available source. Verify it landed.
    if ACTIVE is None:
        sys.stderr.write(
            "minion: no source configured — set MINION_SOURCE_* env vars or use --source\n")
        sys.exit(2)

    # Warm the context-window probe in the background (same as the REPL).
    _ensure_ctx_probe(ACTIVE)

    # --- run the turn ----------------------------------------------------------
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": instruction}]
    try:
        _run_model_turn_loop(messages, on_step=None)
    except KeyboardInterrupt:
        sys.stderr.write("\n[minion] interrupted\n")
        sys.exit(130)
    except Exception as e:
        sys.stderr.write(f"\n[minion] error: {type(e).__name__}: {e}\n")
        sys.exit(1)
    sys.exit(0)


def main():
    global YOLO, APPROVE_LEVEL, DEFAULT_APPROVE_LEVEL, AUTOCOMPRESS_PERCENT
    # --prompt-file: headless single-shot mode — run one turn and exit.
    if _run_one_shot():
        return
    # `minion sessions [query]` — discover + exit, no REPL. Checked first so
    # it short-circuits before the banner / network source resolution that the
    # interactive loop depends on.
    if _cli_sessions(sys.argv[1:]):
        return
    # The banner is printed at startup and after source/approval changes.
    print(_banner())
    print()
    # Warm the active source's max-context-window cache before the first
    # prompt: the probe (esp. on a still-loading local server, whose /v1/models
    # + /props 503 while mmapping) can take seconds-to-tens-of-seconds, but the
    # user typically spends that long typing their first message — so by the
    # time turn 1 finishes the cache is often already filled and the footer
    # shows the "/<max>" suffix immediately instead of a turn later. Background,
    # non-blocking; degrades to "no suffix" if it never resolves.
    if ACTIVE is not None:
        _ensure_ctx_probe(ACTIVE)
    messages = [{"role": "system", "content": SYSTEM}]
    history = []  # past user submissions, newest last; Up/Down navigates

    # --- session state ------------------------------------------------------
    # Every minion run has a session id. If the user passed --resume / -r
    # (with an id, prefix, index, or title) or --session <id>, we load that;
    # otherwise a fresh id is minted. The session is auto-saved after each
    # model turn (the messages array is the source of truth — we just write
    # it to ~/.minion/sessions/<id>.json atomically). A session that never
    # got any user input is never written to disk.
    session_id = _session_id_from_args()
    _resume_requested = any(a in ("--resume", "-r") for a in sys.argv[1:])
    if session_id is None:
        if _resume_requested:
            # bare `minion --resume` but no saved sessions exist — start fresh.
            print(f"{DIM}  no saved sessions to resume — starting fresh{RESET}")
        session_id = _new_session_id()
    session_dirty = False  # has the in-memory context diverged from disk?
    if _resume_requested or any(a == "--session" for a in sys.argv[1:]):
        # Only attempt to load when the user explicitly asked to resume a
        # specific (or most-recent) session. A plain `minion` run mints a
        # fresh id above and starts with an empty context.
        data = _load_session(session_id)
        if data and isinstance(data.get("messages"), list):
            messages = data["messages"]
            # Drop the old system prompt and inject the current one so a
            # resumed session picks up any SYSTEM edits / tool changes.
            messages = [m for m in messages if m.get("role") != "system"]
            messages.insert(0, {"role": "system", "content": SYSTEM})
            n_msgs = len([m for m in messages if m.get("role") != "system"])
            title = data.get("title") or "(untitled)"
            print(f"{DIM}  ↻ resumed session {session_id} ({title!r}, "
                  f"{n_msgs} messages){RESET}")
            print()
            # Show the conversation history so the user immediately remembers
            # what the session was about before the first prompt.
            if n_msgs:
                print(f"{DIM}  ── transcript ──{RESET}")
                _print_transcript(messages)
                print(f"{DIM}  ──────────────{RESET}")
                print()
            # Restore the active source (and any model override) the session
            # was started on, so a resume lands on the same endpoint + model.
            src_name = data.get("source")
            if src_name:
                restore_source(src_name, model=data.get("model"))
                # Re-apply provider routing that was in effect when the session
                # was saved, so an OpenRouter /provider choice survives a
                # resume. A saved "provider" list rebuilds the routing object.
                saved_prov = data.get("provider")
                if isinstance(saved_prov, list) and ACTIVE and src_name in SOURCES:
                    _set_provider_order(SOURCES[src_name], saved_prov)
            session_dirty = False  # just loaded — in sync with disk
        elif _resume_requested:
            print(f"{YELLOW}  ✗ no saved session found for {session_id!r}; "
                  f"starting fresh{RESET}")

    # --- usage metrics -------------------------------------------------------
    # Build the cumulative-token recorder for this session. When resuming, we
    # reload its running totals from the saved session JSON so they keep
    # climbing across restarts instead of resetting to zero. The optional
    # HTTP push (MINION_METRICS_URL) is off by default — see _Metrics.
    global _METRICS
    push_url = _env_metrics_url()
    _METRICS = _Metrics(
        session_id, model=MODEL,
        source=ACTIVE.name if ACTIVE else None,
        push_url=push_url)
    if _resume_requested or any(a == "--session" for a in sys.argv[1:]):
        saved = _load_session(session_id) or {}
        if isinstance(saved.get("input_tokens"), int):
            _METRICS.input_tokens = saved.get("input_tokens") or 0
            _METRICS.output_tokens = saved.get("output_tokens") or 0
            _METRICS.cache_read_tokens = saved.get("cache_read_tokens") or 0
            _METRICS.reasoning_tokens = saved.get("reasoning_tokens") or 0
            _METRICS.api_calls = saved.get("api_calls") or 0
            if saved.get("started_at"):
                _METRICS.started_at = saved["started_at"]

    def _save_current(meta=None):
        """Persist the in-memory `messages` to the current session file.

        Title is auto-derived from the first user message (unless the user
        has set one explicitly with /save <title>). Source and cwd are
        recorded so a resume can re-select the right endpoint.
        """
        if session_id is None:
            return
        m = dict(meta or {})
        if "title" not in m:
            first_user = ""
            for msg in messages:
                if msg.get("role") == "user" and isinstance(msg.get("content"), str):
                    first_user = msg["content"]
                    break
            t = _safe_title(first_user)
            if t:
                m["title"] = t
        m.setdefault("source", ACTIVE.name if ACTIVE else None)
        m.setdefault("cwd", os.getcwd())
        m.setdefault("model", MODEL)
        # Record the active provider routing (if any) so a /resume re-applies
        # the same OpenRouter upstream choice. Stored as a bare `order` list
        # (not the whole extra_body) to keep the session JSON minimal and to
        # avoid persisting unrelated request-body keys a future feature might add.
        if ACTIVE:
            order = _provider_order(ACTIVE)
            if order:
                m["provider"] = order
        # Merge cumulative token totals into the session JSON so they persist
        # across restarts and are available to any metrics consumer that reads
        # the session files (e.g. a dashboard). This is always-on — the numbers
        # are already in hand from the stats footer, so storing them costs
        # nothing and a local session log is useful regardless of a push sink.
        if _METRICS is not None:
            m.update(_METRICS.as_meta())
        try:
            _write_session(session_id, messages, m)
        except OSError as e:
            print(f"{RED}  ✗ couldn't save session: {type(e).__name__}: {e}{RESET}")
        # Fire-and-forget push to the optional metrics endpoint. Only does
        # anything if MINION_METRICS_URL is set; self-disables on first
        # failure so a down/absent dashboard never slows the REPL.
        if _METRICS is not None:
            _METRICS.flush()

    while True:
        try:
            user = read_multiline(history=history)
        except (EOFError, KeyboardInterrupt):
            print()
            # Flush the current session on Ctrl-D / Ctrl-C exit so the last
            # turn isn't lost (matches Hermes's exit-path flush).
            if session_dirty:
                _save_current()
            # Show a grey resume hint so the user can pick up right where they
            # left off without having to run `minion sessions` first.
            if session_id:
                print(f"{DIM}  resume with: minion --resume {session_id}{RESET}")
            break
        user = user.strip()
        if not user:
            continue
        if user == "/quit":
            if session_dirty:
                _save_current()
            break
        if user == "/source" or user.startswith("/source "):
            sp = user.split()
            if len(sp) == 1:
                print(f"{DIM}  sources:{RESET}")
                for sname in SOURCE_ORDER:
                    src = SOURCES[sname]
                    mark = f"{GREEN}★{RESET}" if sname == ACTIVE.name else f"{DIM} ·{RESET}"
                    m = src.display_model()
                    ctx = src._context_window
                    if isinstance(ctx, int) and ctx > 0:
                        ctx_s = f" · {_abbr(ctx)} ctx"
                    else:
                        ctx_s = ""
                        _ensure_ctx_probe(src)  # warm the cache for next time
                    # OpenRouter provider routing, if pinned on this source.
                    prov = _provider_order(src)
                    prov_s = f" · via {','.join(prov)}" if prov else ""
                    print(f"  {mark} {CYAN}{sname:<12}{RESET} {DIM}{m} @ {src.base_url}{ctx_s}{prov_s}{RESET}")
                print(f"{DIM}  /source <name> [model] to switch · "
                      "context preserved (use /reset to clear){RESET}")
            else:
                target = sp[1]
                # Optional second token: a model id to pin for this switch.
                # Useful for multi-model hosts (e.g. Together) so you can do
                #   /source together                    → default (GLM-5.2)
                #   /source together zai-org/GLM-4.6     → that model, same host
                # The override is non-sticky: a later bare /source <name>
                # always falls back to the source's configured default.
                model_override = sp[2] if len(sp) >= 3 else None
                if target not in SOURCES:
                    avail = ", ".join(SOURCE_ORDER)
                    print(f"{RED}  ✗ unknown source {target!r} — available: {avail}{RESET}")
                    continue
                if target == ACTIVE.name and not model_override:
                    print(f"{DIM}  already on {target}{RESET}")
                    continue
                switch_source(target, model_override=model_override)
                src = SOURCES[target]
                # An explicit /source switch is user-initiated, so a synchronous
                # probe here is the right trade: the user asked for this source
                # and will want to see its context limit. Caches after the first
                # switch, so later switches to a known source are instant.
                ctx = src.resolve_context_window()
                ctx_s = f" · {_abbr(ctx)} ctx" if isinstance(ctx, int) and ctx > 0 else ""
                print(f"{BOLD}{YELLOW}  → switched to {target}{RESET} {DIM}({MODEL} @ {src.base_url}{ctx_s}){RESET}")
                print(_banner())
                session_dirty = True  # source/model change is worth persisting
                if _METRICS is not None:
                    _METRICS.model = MODEL
                    _METRICS.source = target
            continue
        if user == "/provider" or user.startswith("/provider "):
            # /provider — control OpenRouter upstream-provider routing for the
            # active source. OpenRouter fronts many providers behind one model
            # id; routing picks which of them serves the request (different
            # providers back the same endpoint at different prices / precisions
            # / latencies). The `provider.order` list rides in the request body
            # via extra_body (see Source.extra_request_kwargs). This only
            # meaningfully routes on OpenRouter, but is safe to call on any
            # source — non-consumers ignore unknown body keys.
            #
            #   /provider                       show the active source's routing
            #   /provider openrouter           show routing for <source>
            #   /provider parasail/fp8         pin to that provider (active src)
            #   /provider Together,DeepInfra    ordered: try Together, then DeepInfra
            #   /provider off                   clear routing (let OpenRouter pick)
            #   /provider openrouter parasail   set routing on <source>, not active
            parts = user.split()
            if len(parts) == 1:
                # /provider with no args → show the active source's routing.
                cur = _provider_order(ACTIVE)
                if cur:
                    print(f"{DIM}  provider order: {', '.join(cur)}"
                          f"{RESET} {DIM}(on {ACTIVE.name}){RESET}")
                else:
                    print(f"{DIM}  no provider routing set on {ACTIVE.name}"
                          f"{RESET}")
                print(f"{DIM}  /provider <a,b,…> to pin · "
                      "/provider <source> <a,b,…> · /provider off{RESET}")
                continue
            arg = parts[1]
            # /provider <source> [order]  vs  /provider <order>
            if arg in SOURCES and len(parts) >= 3:
                target_src = SOURCES[arg]
                order_str = parts[2]
            else:
                target_src = ACTIVE
                order_str = arg
            if order_str.lower() in ("off", "none", "clear", ""):
                _set_provider_order(target_src, [])
                print(f"{DIM}  ✓ provider routing cleared on "
                      f"{target_src.name}{RESET}")
            else:
                order = [p.strip() for p in order_str.split(",") if p.strip()]
                _set_provider_order(target_src, order)
                print(f"{DIM}  ✓ provider order: {', '.join(order)}"
                      f"  (on {target_src.name}, no fallback){RESET}")
            session_dirty = True  # routing change is worth persisting
            continue
        if user == "/yolo":
            YOLO = not YOLO
            if YOLO:
                APPROVE_LEVEL = None  # never prompt
            else:
                APPROVE_LEVEL = DEFAULT_APPROVE_LEVEL
            print(f"{DIM}  yolo={YOLO}  approval={_approval_display()}{RESET}")
            print(_banner())
            continue
        if user.startswith("/approval"):
            parts = user.split()
            if len(parts) == 1:
                # /approval with no arg → show current setting
                print(f"{DIM}  approval={_approval_display()}  (all|low|medium|high|yolo){RESET}")
                continue
            arg = parts[1]
            kind, resolved = _normalize_approval(arg)
            if kind:
                _apply_approval(kind, resolved, update_default=(kind != "yolo"))
                if kind == "prompt_all":
                    print(f"{DIM}  approval=all (prompt every classified action){RESET}")
                elif kind == "level":
                    print(f"{DIM}  approval={resolved} (auto-allow ≤ {resolved}){RESET}")
                else:
                    print(f"{DIM}  approval=off (yolo — never prompt){RESET}")
                print(_banner())
            else:
                print(f"{YELLOW}  unknown level {arg!r} — want all|low|medium|high|yolo{RESET}")
            continue
        if user in ("/reset", "/clear", "/new"):
            messages = [{"role": "system", "content": SYSTEM}]
            print(f"{DIM}  context cleared{RESET}")
            # A reset forks a new session id so the fresh chat isn't written
            # over the old one (mirrors Hermes's "new session on /new").
            session_id = _new_session_id()
            session_dirty = False
            # Fresh recorder for the fresh session.
            _METRICS = _Metrics(
                session_id, model=MODEL,
                source=ACTIVE.name if ACTIVE else None,
                push_url=_env_metrics_url())
            print(f"{DIM}  new session {session_id}{RESET}")
            continue
        if user == "/recover" or user.startswith("/recover "):
            note = user.split(None, 1)[1].strip() if " " in user else ""
            content = f"[Runtime note: {MANUAL_RECOVERY_NUDGE}"
            if note:
                content += f" User note: {note}"
            content += "]"
            messages.append({"role": "user", "content": content})
            print(f"{YELLOW}  ↳ manual recovery requested — forcing visible checkpoint{RESET}")
            _run_model_turn_loop(messages, force_final=True, recovery_sampling=True,
                                 on_step=_save_current)
            session_dirty = True
            continue
        if user in ("/compress", "/compact"):
            # nothing to compress if we're under (system + KEEP) turns
            body_len = len(messages) - (1 if messages and messages[0].get("role") == "system" else 0)
            if body_len <= COMPRESS_KEEP:
                print(f"{DIM}  nothing to compress ({body_len} turn{'s' if body_len != 1 else ''} in context){RESET}")
                continue
            try:
                ok = _confirm(f"compress {body_len - COMPRESS_KEEP} older turns (keep last {COMPRESS_KEEP})")
            except _EscToChat:
                print(f"{YELLOW}  ↳ escaped compress approval — back to chat input{RESET}")
                continue
            if not ok:
                print(f"{DIM}  cancelled{RESET}")
                continue
            print(f"{DIM}  compressing…{RESET}")
            result = compress(messages)
            if result is None:
                continue  # error already printed
            kept_n, summarized_n, summary_chars = result
            print(f"{DIM}  └ compressed {summarized_n} turns → 1 summary "
                  f"({summary_chars} chars), kept last {kept_n} verbatim{RESET}")
            session_dirty = True
            continue
        if user == "/autocompress" or user.startswith("/autocompress "):
            arg = user[len("/autocompress"):].strip().lower()
            if not arg:
                if AUTOCOMPRESS_PERCENT <= 0:
                    print(f"{DIM}  auto-compress: off{RESET}")
                else:
                    print(f"{DIM}  auto-compress: {AUTOCOMPRESS_PERCENT}% "
                          f"(auto-compress when context ≥ {AUTOCOMPRESS_PERCENT}% full){RESET}")
                continue
            if arg in ("off", "disable", "0"):
                AUTOCOMPRESS_PERCENT = 0
                print(f"{DIM}  auto-compress: off (disabled){RESET}")
                continue
            if arg in ("on", "enable"):
                AUTOCOMPRESS_PERCENT = AUTOCOMPRESS_DEFAULT
                print(f"{DIM}  auto-compress: {AUTOCOMPRESS_PERCENT}% "
                      f"(auto-compress when context ≥ {AUTOCOMPRESS_PERCENT}% full){RESET}")
                continue
            try:
                val = int(arg)
            except ValueError:
                print(f"{YELLOW}  usage: /autocompress [<1-100|off|on>] — "
                      f"current: {AUTOCOMPRESS_PERCENT}%{RESET}")
                continue
            if not (1 <= val <= 100):
                print(f"{YELLOW}  {val} is out of range — want 1–100 (or off / on){RESET}")
                continue
            AUTOCOMPRESS_PERCENT = val
            print(f"{DIM}  auto-compress: {val}% "
                  f"(auto-compress when context ≥ {val}% full){RESET}")
            continue
        # --- session commands -------------------------------------------------
        if user == "/sessions" or user.startswith("/sessions "):
            _cmd_sessions(user, session_id)
            continue
        if user == "/resume" or user.startswith("/resume "):
            new_sid = _cmd_resume(user, session_id)
            if new_sid:
                # Persist the outgoing session before switching away.
                if session_dirty:
                    _save_current()
                session_id = new_sid
                messages[:] = _session_messages(session_id)
                session_dirty = False
                # Reload the cumulative-token recorder from the resumed
                # session's saved totals so they keep climbing from here.
                _METRICS = _Metrics(
                    session_id, model=MODEL,
                    source=ACTIVE.name if ACTIVE else None,
                    push_url=_env_metrics_url())
                saved = _load_session(session_id) or {}
                if isinstance(saved.get("input_tokens"), int):
                    _METRICS.input_tokens = saved.get("input_tokens") or 0
                    _METRICS.output_tokens = saved.get("output_tokens") or 0
                    _METRICS.cache_read_tokens = saved.get("cache_read_tokens") or 0
                    _METRICS.reasoning_tokens = saved.get("reasoning_tokens") or 0
                    _METRICS.api_calls = saved.get("api_calls") or 0
                    if saved.get("started_at"):
                        _METRICS.started_at = saved["started_at"]
            continue
        if user == "/save" or user.startswith("/save "):
            title = user.split(None, 1)[1].strip() if " " in user else None
            _save_current({"title": title} if title else {})
            session_dirty = False
            shown = f" as {title!r}" if title else ""
            print(f"{DIM}  ✓ saved session {session_id}{shown}{RESET}")
            continue
        if user == "/memory" or user.startswith("/memory "):
            _cmd_memory(messages, user)
            continue
        if user == "/delete" or user.startswith("/delete "):
            target = user.split(None, 1)[1].strip() if " " in user else None
            _cmd_delete(target, session_id)
            continue
        # record for history (skip duplicates of the very last entry so
        # Up doesn't immediately re-show what was just submitted)
        if not history or history[-1] != user:
            history.append(user)
        print()  # breathing room before the spinner/text starts
        messages.append({"role": "user", "content": user})
        _run_model_turn_loop(messages, on_step=_save_current)
        # Auto-save after the turn settles (whether it ended cleanly, hit the
        # step cap, or was escaped). This is the Hermes pattern: persist every
        # turn so a crash / accidental close never loses work.
        session_dirty = True
        _save_current()
        # Maybe auto-compress: if the context window is over the configured
        # fill threshold, silently fold the older turns so the chat can keep
        # going. Done after the save so the pre-compression state is on disk;
        # the compressed result is persisted by the _save_current() below.
        if _maybe_autocompress(messages, _LAST_PROMPT_TOKENS):
            _save_current()
        # Maybe refresh the model-generated description (a cheap non-streaming
        # call every SESSION_DESC_REFRESH turns). Done after the save so the
        # turn's messages are already on disk; the description lands as a
        # second write. Failures are silent — the title is always present.
        _maybe_refresh_description(session_id, messages)
        session_dirty = False


def _session_id_from_args():
    """Resolve a starting session id from CLI flags: --resume/-r and --session.

    `--resume` with no following target resumes the most recent session
    (the natural "pick up where I left off" intent). With a target it resolves
    by index/id/prefix/title. `--session <id>` forces an exact id (creates if
    absent — useful for scripts that want a stable id)."""
    args = sys.argv[1:]
    out = None
    i = 0
    while i < len(args):
        a = args[i]
        if a in ("--resume", "-r"):
            if i + 1 < len(args) and not args[i + 1].startswith("-"):
                # --resume <target>
                target = args[i + 1]
                sessions = _list_sessions(limit=50)
                out = _resolve_session(target, sessions) or target
                i += 2
            else:
                # bare --resume → most recent session, if any
                sessions = _list_sessions(limit=1)
                out = sessions[0]["id"] if sessions else None
                i += 1
            continue
        if a == "--session" and i + 1 < len(args):
            out = args[i + 1]
            i += 2
            continue
        i += 1
    return out


def _session_messages(session_id):
    """Load just the messages list for a session, with the live SYSTEM prepended."""
    data = _load_session(session_id) or {}
    msgs = [m for m in data.get("messages", []) if m.get("role") != "system"]
    msgs.insert(0, {"role": "system", "content": SYSTEM})
    return msgs


def _fmt_when(ts):
    """Compact relative timestamp for the session list."""
    if not ts:
        return "?"
    delta = time.time() - ts
    if delta < 60:
        return "just now"
    if delta < 3600:
        return f"{int(delta // 60)}m ago"
    if delta < 86400:
        return f"{int(delta // 3600)}h ago"
    if delta < 86400 * 7:
        return f"{int(delta // 86400)}d ago"
    return time.strftime("%Y-%m-%d", time.localtime(ts))


def _print_transcript(messages, max_chars=120):
    """Render a session's message history as a one-line-per-message recap.

    Used on resume so the user immediately sees what the conversation was
    about, and by `/sessions <id>` for its detail view. Each message is
    collapsed to a single line (newlines → spaces) and truncated so the whole
    history reads as a scannable timeline rather than a wall of text. Tool
    calls render as `→ name(...)` so you can see what ran. System messages
    are skipped. Returns the number of lines printed."""
    printed = 0
    for m in messages:
        role = m.get("role", "?")
        if role == "system":
            continue
        content = m.get("content")
        if content is None and m.get("tool_calls"):
            calls = ", ".join(
                f"{tc['function']['name']}(...)" for tc in m["tool_calls"])
            content = f"→ {calls}"
        elif isinstance(content, list):
            content = " ".join(p.get("text", "") for p in content if isinstance(p, dict))
        content = (content or "").strip().replace("\n", " ")
        if len(content) > max_chars:
            content = content[:max_chars - 1] + "…"
        color = {"user": CYAN, "assistant": GREEN, "tool": DIM}.get(role, "")
        print(f"  {color}{role:>9}{RESET}  {content}")
        printed += 1
    return printed


def _session_list_page_options(args, default_limit=SESSION_LIST_DEFAULT_LIMIT):
    """Parse paging flags and leave everything else as the search query."""
    page = 1
    limit = default_limit
    query_parts = []
    warnings = []
    i = 0

    def positive_int(value, fallback, name):
        try:
            n = int(value)
        except (TypeError, ValueError):
            warnings.append(f"ignored invalid {name}: {value!r}")
            return fallback
        if n < 1:
            warnings.append(f"ignored invalid {name}: {value!r}")
            return fallback
        return n

    while i < len(args):
        a = args[i]
        if a in ("--page", "-p"):
            if i + 1 >= len(args):
                warnings.append(f"ignored missing value for {a}")
                i += 1
                continue
            page = positive_int(args[i + 1], page, "page")
            i += 2
            continue
        if a.startswith("--page="):
            page = positive_int(a.split("=", 1)[1], page, "page")
            i += 1
            continue
        if a in ("--limit", "-n"):
            if i + 1 >= len(args):
                warnings.append(f"ignored missing value for {a}")
                i += 1
                continue
            limit = positive_int(args[i + 1], limit, "limit")
            i += 2
            continue
        if a.startswith("--limit="):
            limit = positive_int(a.split("=", 1)[1], limit, "limit")
            i += 1
            continue
        query_parts.append(a)
        i += 1

    limit = max(1, min(limit, SESSION_LIST_MAX_LIMIT))
    query = " ".join(query_parts).strip() or None
    return query, page, limit, warnings


def _print_session_list(sessions, start_index=1, current_id=None):
    for offset, s in enumerate(sessions):
        i = start_index + offset
        title = s["title"] or "(empty)"
        if current_id is not None:
            mark = f"{GREEN}●{RESET}" if s["id"] == current_id else f"{DIM}{i:>3}{RESET}"
            prefix = f"  {mark} "
        else:
            prefix = f"  {DIM}{i:>3}{RESET}  "
        sid = s.get("id") or ""
        print(f"{prefix}{MAGENTA}{sid}{RESET}  {title}")

        meta = [f"{s['n']} msg", _fmt_when(s["updated_at"])]
        if s.get("source"):
            meta.append(f"source {s['source']}")
        if s.get("model"):
            meta.append(f"model {s['model']}")
        cwd = _display_path(s.get("cwd"))
        if cwd:
            meta.append(f"cwd {cwd}")
        print(f"       {DIM}{' · '.join(meta)}{RESET}")

        detail = s.get("description")
        label = "desc"
        if not detail and s.get("preview"):
            detail = s["preview"]
            label = "first"
        if detail:
            print(f"       {DIM}{label}: {detail}{RESET}")


def _session_next_hint(command, page, limit, query=None):
    quoted_query = f" {shlex.quote(query)}" if query else ""
    return f"{command}{quoted_query} --page {page + 1} --limit {limit}"


def _display_path(path):
    if not path:
        return None
    path = os.path.expanduser(str(path))
    home = os.path.expanduser("~")
    if path == home:
        return "~"
    if path.startswith(home + os.sep):
        return "~" + path[len(home):]
    return path


def _cli_sessions(args):
    """`minion sessions [query]` — list saved sessions and exit (no REPL).

    With no query: prints the first page of recent sessions. With a query:
    substring-filters (case-insensitive) across title, description, preview,
    and id, so once
    you have dozens of sessions you can narrow with e.g.
    `minion sessions refactor`. Add `--page/-p` and `--limit/-n` to page
    through the list. Returns True if it handled a listing request (caller
    should then exit); False if this wasn't a sessions invocation.

    Deliberately a top-level verb rather than `--resume list`: "list" looks
    like it could be a session title, and a "resume" flag doing two different
    things (list vs. switch) depending on whether it has an arg is the kind
    of inconsistency that confuses. `sessions` = discover, `--resume` = enter.
    """
    if not args:
        return False
    if args[0] not in ("sessions", "--sessions", "ls", "list"):
        return False
    query, page, limit, warnings = _session_list_page_options(args[1:])
    offset = (page - 1) * limit
    fetched = _list_sessions(limit=limit + 1, offset=offset, query=query)
    sessions = fetched[:limit]
    has_next = len(fetched) > limit
    for warning in warnings:
        print(f"{YELLOW}  {warning}{RESET}")
    if not sessions:
        if query:
            print(f"  no sessions matching {query!r} on page {page}")
        else:
            if page > 1:
                print(f"  no sessions on page {page}  ({_sessions_dir()})")
            else:
                print(f"  no saved sessions yet  ({_sessions_dir()})")
        return True
    where = f" matching {query!r}" if query else ""
    print(f"{DIM}  sessions{where} · page {page} · {_sessions_dir()}{RESET}")
    _print_session_list(sessions, start_index=offset + 1)
    print(f"{DIM}  resume with: minion --resume <n|short-id|title>{RESET}")
    if has_next:
        print(f"{DIM}  next page: {_session_next_hint('minion sessions', page, limit, query)}{RESET}")
    return True


def _cmd_sessions(user, current_id):
    """Handle /sessions [n] — list recent sessions, or show one in detail."""
    parts = user.split()
    args = parts[1:]
    list_requested = not args or args[0] in ("--page", "-p", "--limit", "-n")
    list_requested = list_requested or (args and (
        args[0].startswith("--page=")
        or args[0].startswith("--limit=")
        or args[0] in ("page", "p")
    ))
    if list_requested:
        page_args = args
        if page_args and page_args[0] in ("page", "p"):
            page_args = ["--page", *page_args[1:]]
        query, page, limit, warnings = _session_list_page_options(page_args)
        offset = (page - 1) * limit
        fetched = _list_sessions(limit=limit + 1, offset=offset, query=query)
        sessions = fetched[:limit]
        has_next = len(fetched) > limit
        for warning in warnings:
            print(f"{YELLOW}  {warning}{RESET}")
        if not sessions:
            if query:
                print(f"{DIM}  no sessions matching {query!r} on page {page}{RESET}")
            elif page > 1:
                print(f"{DIM}  no sessions on page {page} ({_sessions_dir()}){RESET}")
            else:
                print(f"{DIM}  no saved sessions yet (saved to {SESSION_HOME}/sessions){RESET}")
            return
        where = f" matching {query!r}" if query else ""
        print(f"{DIM}  recent sessions{where} · page {page} ({_sessions_dir()}):{RESET}")
        _print_session_list(sessions, start_index=offset + 1, current_id=current_id)
        print(f"{DIM}  /resume <n|short-id|title> to switch · /delete <n|id> to remove{RESET}")
        if has_next:
            query_tail = f" {shlex.quote(query)}" if query else ""
            print(f"{DIM}  next page: /sessions --page {page + 1} --limit {limit}{query_tail}{RESET}")
        return
    # /sessions <target> — show detail for one session
    arg = " ".join(args).strip()
    sessions = _list_sessions(limit=50)
    sid = _resolve_session(arg, sessions)
    if not sid:
        print(f"{YELLOW}  no session matching {arg!r}{RESET}")
        return
    data = _load_session(sid)
    if not data:
        print(f"{YELLOW}  ✗ couldn't read session {sid}{RESET}")
        return
    msgs = data.get("messages", [])
    title = data.get("title") or "(untitled)"
    print(f"{DIM}  session {sid}{RESET}")
    print(f"{DIM}  title: {title}{RESET}")
    print(f"{DIM}  messages: {len(msgs)} · updated {_fmt_when(data.get('updated_at', 0))}{RESET}")
    print(f"{DIM}  ── transcript ──{RESET}")
    _print_transcript(msgs)


def _cmd_resume(user, current_id):
    """Handle /resume [target] — pick a session to resume. Returns the new id
    or None if nothing changed."""
    parts = user.split(None, 1)
    target = parts[1].strip() if len(parts) > 1 else ""
    sessions = _list_sessions(limit=15)
    if not target:
        if not sessions:
            print(f"{DIM}  no saved sessions to resume{RESET}")
            return None
        print(f"{DIM}  recent sessions:{RESET}")
        for i, s in enumerate(sessions, 1):
            mark = f"{GREEN}●{RESET}" if s["id"] == current_id else f"{DIM}{i:>2}{RESET}"
            title = s["title"] or "(empty)"
            print(f"  {mark} {MAGENTA}{s.get('short', _short_id(s['id']))}{RESET}  "
                  f"{title}{DIM} · {s['n']} msg · {_fmt_when(s['updated_at'])}{RESET}")
            desc = s.get("description") or (s.get("preview") if s.get("preview") != title else None)
            if desc:
                print(f"        {DIM}{desc}{RESET}")
        print(f"{DIM}  /resume <n|short-id|prefix|title> to switch{RESET}")
        return None
    target = target.strip()
    # Let the user strip surrounding <> [] "" they may have copied from help.
    if len(target) >= 2 and target[0] in "<[\"'" and target[-1] in ">]\"'":
        target = target[1:-1].strip()
    sessions = _list_sessions(limit=50)
    sid = _resolve_session(target, sessions)
    if not sid:
        print(f"{YELLOW}  ✗ no session matching {target!r}{RESET}")
        return None
    if sid == current_id:
        print(f"{DIM}  already on that session{RESET}")
        return None
    data = _load_session(sid)
    if not data:
        print(f"{YELLOW}  ✗ couldn't read session {sid}{RESET}")
        return None
    # Reselect the recorded source (and model override) if configured, so a
    # resume lands on the same endpoint + model the session started on.
    src = data.get("source")
    if src and restore_source(src, model=data.get("model")):
        print(f"{BOLD}{YELLOW}  → source {src}{RESET} {DIM}({MODEL}){RESET}")
    n_msgs = len([m for m in data.get("messages", []) if m.get("role") != "system"])
    title = data.get("title") or "(untitled)"
    print(f"{DIM}  ↻ resumed {sid} ({title!r}, {n_msgs} messages){RESET}")
    print()
    return sid


def _cmd_delete(target, current_id):
    """Handle /delete [target] — remove a session file."""
    if not target:
        print(f"{YELLOW}  usage: /delete <n|id|prefix>  (use /sessions to list){RESET}")
        return
    sessions = _list_sessions(limit=50)
    sid = _resolve_session(target, sessions)
    if not sid:
        print(f"{YELLOW}  ✗ no session matching {target!r}{RESET}")
        return
    if sid == current_id:
        print(f"{YELLOW}  can't delete the session you're currently in — /reset first{RESET}")
        return
    if _delete_session(sid):
        print(f"{DIM}  ✓ deleted session {sid}{RESET}")
    else:
        print(f"{YELLOW}  ✗ couldn't delete session {sid}{RESET}")


def _cmd_memory(messages, user_input):
    """Handle /memory save|remember|list  — manual memories.

    /memory save [focus...]      — distill session into a memory (optional focus
                                   guidance for the model: e.g. "focus on config")
    /memory remember <query>     — keyword search through saved memories
    /memory list                 — list all saved memories
    """
    parts = user_input.split(None, 2)
    sub = parts[1].lower() if len(parts) >= 2 else ""

    if sub == "list":
        _cmd_memory_list()
        return

    if sub == "remember":
        query = parts[2] if len(parts) >= 3 else ""
        _cmd_memory_remember(query)
        return

    if sub == "save":
        notes = parts[2] if len(parts) >= 3 else None
        _cmd_memory_save(messages, notes)
        return

    # Bare /memory — show usage
    print(f"{DIM}  usage:{RESET}")
    print(f"{DIM}    /memory save [focus...]   save a memory (optional focus guidance){RESET}")
    print(f"{DIM}    /memory remember <query>   search memories{RESET}")
    print(f"{DIM}    /memory list               list all memories{RESET}")


def _slugify(text, maxlen=60):
    """Turn a title into a filesystem-safe slug."""
    s = text.lower().strip()
    s = re.sub(r"[^a-z0-9]+", "-", s)
    s = s.strip("-")
    if len(s) > maxlen:
        s = s[:maxlen].rstrip("-")
    return s or "untitled"


def _cmd_memory_save(messages, notes=None):
    """Save a memory to ~/.minion/memories/<slug>.md.

    The model always distills the session into a markdown memory note.
    ``notes``, if provided, serves as guidance: "focus on the build system",
    "capture the API endpoint details", etc.
    """
    memories_dir = _memories_dir()
    os.makedirs(memories_dir, exist_ok=True)

    stamp = time.strftime("%Y-%m-%d %H:%M:%S")
    fname_stamp = time.strftime("%Y%m%d-%H%M%S")

    print(f"{DIM}  distilling session into a memory…{RESET}")
    transcript = _render_transcript_for_memory(messages)

    focus = f"\nThe user specifically asks to focus on: {notes}\n" if notes else ""
    prompt = (
        "You are a developer's memory-keeper. Given the following coding session transcript, "
        "extract the key learnings, discoveries, and solutions that are worth remembering "
        "for future work on this project. Focus on:\n"
        "- What problem was being solved\n"
        "- What approach worked (and what didn't)\n"
        "- Specific file paths, commands, config values, or API quirks that were discovered\n"
        "- Why things work the way they do (the reasoning, not just the result)\n"
        f"{focus}"
        "Write this as a concise, knowledge-rich markdown note that another developer "
        "(or yourself weeks from now) can read and immediately understand. "
        "Use backticks for code, paths, and commands. "
        "Start with a `# Title` line that captures the topic.\n\n"
        f"---\n{transcript}\n---"
    )
    try:
        _log_event("req", {"model": MODEL, "messages": [{"role": "user", "content": prompt}],
                           "stream": False, "_purpose": "memory_save"})
        resp = client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "user", "content": prompt}],
            stream=False, timeout=30,
            **(ACTIVE.extra_request_kwargs() if ACTIVE else {}))
        try:
            _log_event("resp", {"_purpose": "memory_save", "data": resp.model_dump()})
        except Exception:
            pass
    except APIConnectionError:
        print(f"{RED}  ✗ can't reach {client.base_url} — memory not saved{RESET}")
        return
    except Exception as e:
        print(f"{RED}  ✗ memory save failed: {type(e).__name__}: {e}{RESET}")
        return

    body = (resp.choices[0].message.content or "").strip()
    if not body:
        print(f"{RED}  ✗ model returned empty — memory not saved{RESET}")
        return

    # Extract title from the markdown H1
    title = ""
    for line in body.split("\n"):
        if line.startswith("# "):
            title = line[2:].strip()
            break
    if not title:
        title = body.split("\n")[0][:80].strip() or f"Memory {fname_stamp}"

    # Write the file — timestamp_slug.md so it's human-readable on the filesystem
    slug = _slugify(title)
    fname = f"{fname_stamp}_{slug}.md"
    content = f"{body}\n\n---\n_Saved on {stamp}_\n"
    path = os.path.join(memories_dir, fname)
    try:
        with open(path, "w", encoding="utf-8") as f:
            f.write(content)
        print(f"{DIM}  ✓ saved memory: {title}{RESET}")
        print(f"{DIM}    {path}{RESET}")
    except OSError as e:
        print(f"{RED}  ✗ couldn't write memory: {e}{RESET}")


def _cmd_memory_remember(query):
    """Keyword-search through saved memories."""
    memories_dir = _memories_dir()
    if not os.path.isdir(memories_dir):
        print(f"{DIM}  no memories saved yet{RESET}")
        return

    if not query:
        print(f"{YELLOW}  usage: /memory remember <search terms>{RESET}")
        return

    q = query.lower()
    files = []
    try:
        for name in sorted(os.listdir(memories_dir), reverse=True):
            if not name.endswith(".md"):
                continue
            path = os.path.join(memories_dir, name)
            try:
                with open(path, encoding="utf-8") as f:
                    content = f.read()
            except OSError:
                continue
            if q in content.lower():
                # Extract title from first H1 or first line
                title = ""
                for line in content.split("\n"):
                    if line.startswith("# "):
                        title = line[2:].strip()
                        break
                if not title:
                    title = name
                files.append((name, title, content))
    except OSError:
        print(f"{DIM}  no memories saved yet{RESET}")
        return

    if not files:
        print(f"{DIM}  no memories matching {query!r}{RESET}")
        return

    print(f"{DIM}  memories matching {query!r}:{RESET}")
    for name, title, content in files:
        print(f"\n{CYAN}  ┌─ {title}{RESET}  {DIM}({name}){RESET}")
        # Show first ~10 lines as preview, strip the title line
        lines = content.split("\n")
        shown = 0
        for line in lines:
            if line.startswith("# ") and shown == 0:
                continue  # skip the title line in preview (already shown)
            if shown >= 10:
                print(f"{DIM}  │  …{RESET}")
                break
            print(f"{DIM}  │{RESET} {line[:100]}")
            shown += 1


def _cmd_memory_list():
    """List all saved memories with titles and dates."""
    memories_dir = _memories_dir()
    if not os.path.isdir(memories_dir):
        print(f"{DIM}  no memories saved yet{RESET}")
        return

    files = []
    try:
        for name in os.listdir(memories_dir):
            if not name.endswith(".md"):
                continue
            path = os.path.join(memories_dir, name)
            try:
                mtime = os.path.getmtime(path)
                with open(path, encoding="utf-8") as f:
                    first_line = f.readline().strip()
                title = first_line[2:] if first_line.startswith("# ") else name
                files.append((mtime, name, title))
            except OSError:
                continue
    except OSError:
        print(f"{DIM}  no memories saved yet{RESET}")
        return

    if not files:
        print(f"{DIM}  no memories saved yet{RESET}")
        return

    files.sort(key=lambda x: x[0], reverse=True)
    print(f"{DIM}  memories ({len(files)}):{RESET}")
    for i, (mtime, name, title) in enumerate(files, 1):
        date = time.strftime("%Y-%m-%d %H:%M", time.localtime(mtime))
        print(f"  {GREEN}{i:>2}.{RESET} {title}  {DIM}({date}){RESET}")


def _render_transcript_for_memory(messages):
    """Render session messages into a compact text transcript for the model.

    Skips the system prompt, truncates long tool outputs, and shows tool calls.
    """
    out = []
    for m in messages:
        if m.get("role") == "system":
            continue
        role = m.get("role", "?")
        content = m.get("content")
        if content is None and m.get("tool_calls"):
            calls = ", ".join(
                f"{c['function']['name']}({c['function']['arguments'][:100]})"
                for c in m["tool_calls"]
            )
            out.append(f"[{role}] → {calls}")
        elif isinstance(content, list):
            text = "".join(p.get("text", "") for p in content if isinstance(p, dict))
            out.append(f"[{role}] {(text or '')[:2000]}")
        else:
            out.append(f"[{role}] {(content or '')[:2000]}")
    return "\n\n".join(out)


if __name__ == "__main__":
    main()