#!/usr/bin/env python3
"""kabu: notify a human and ask them through the Kabu app, from any script or coding agent.

Commands:
  kabu send [TITLE] MESSAGE     notify; `kabu "message"` with no command does the same
  kabu ask QUESTION [MESSAGE]   ask and wait for the answer; `kabu --ask "Q?"` does the same
  kabu status MSG_ID            a message as the API has it (--wait N holds while an ask is pending)
  kabu whoami                   what the key can do, and the project and agent kabu would send
  kabu session start|update|end a session and its Live Activity (--headline, --progress N/M when measured)
  kabu login                    save --api-key KEY, or connect this computer with --device
  kabu mcp                      Kabu's MCP tools on stdin and stdout, for an agent harness to start
  kabu guide                    a copy-paste cheat sheet for coding agents (also: kabu help agents)
  kabu help COMMAND             one command's options and examples (same as kabu COMMAND --help)

Presence (on by default; KABU_PRESENCE=0 or kabu presence off turns it off): kabu presence status shows whether your
agents are live, as their hooks wrote it on this machine and one kabu presence run reports it; kabu presence install
--harness claude|copilot|codex|gemini|cursor|pi adds the hook (pi: an extension) to a harness's settings (the Kabu
plugin has it for Claude Code and Copilot CLI), and uninstall takes it out; kabu presence herdr on adds Herdr's view.

Output is one JSON object on stdout. An error is one JSON line on stderr, {"error": {"code", "message", "hint"}};
the hint is the next step, usually a corrected command. Nothing is ever coloured.
Exit codes: 0 ok (an ask approved, or a choice or text answered), 2 usage or validation error (nothing was sent),
3 network or API error, or an answer kabu does not know, 10 rejected, 20 expired, 21 cancelled or deleted, 130 interrupted.
Environment: KABU_API_KEY, KABU_BASE_URL (default https://usekabu.com), KABU_PROJECT, KABU_AGENT, KABU_SOURCE,
KABU_SESSION, KABU_AUTO_SESSION=1 (an automatic session per host, project and agent), KABU_CONFIG_PATH (default
~/.config/kabu/config.json, written by kabu login, mode 600), KABU_PRESENCE=0 (presence off), KABU_HOME (default
~/.kabu: presence's reporter identity), KABU_RUNTIME_DIR (default XDG_RUNTIME_DIR or TMPDIR: presence's records).
"""
from __future__ import annotations

import argparse
import calendar
from dataclasses import dataclass, field
import difflib
import functools
import getpass
import hashlib
import hmac
import http.client
import json
import mimetypes
import os
from pathlib import Path
import re
import shlex
import shutil
import signal
import socket
import stat
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from typing import Any, Callable, Dict, Iterable, List, Mapping, NamedTuple, NoReturn, Optional, Sequence, Set, Tuple

VERSION = "2.6.0"
EXIT_OK, EXIT_USAGE, EXIT_TRANSPORT = 0, 2, 3
EXIT_REJECTED, EXIT_EXPIRED, EXIT_CANCELLED = 10, 20, 21  # an ask's outcomes; login --device: denied, expired
DEFAULT_BASE_URL = "https://usekabu.com"
WAIT = 25  # seconds the API may hold one request for a pending ask (the contract allows 0 to 50)
GRACE = 120  # seconds kabu keeps waiting past an ask's expiry, for the API to settle it
RETRIES = 3  # after the first try: network errors, 5xx, 429 with Retry-After, 409 idempotency_key_in_use
MAX_RETRY_AFTER = 30.0  # a longer Retry-After fails now rather than block the caller
TITLE_MAX, BODY_MAX = 120, 8192  # MessageCreate maxLength, in characters
REQUEST_MAX = 64 << 10  # the API's cap on a JSON request, in bytes
JSON = Dict[str, Any]
# The type an upload declares, by extension: the container types players open first (AVFoundation will not open
# audio/mp4a-latm, the codec Python's mimetypes names .m4a on macOS; .aac is raw ADTS, audio/aac), then mimetypes.
UPLOAD_TYPES = {
    "m4a": "audio/mp4", "m4b": "audio/mp4", "aac": "audio/aac", "mp4": "video/mp4", "m4v": "video/mp4",
    "mp3": "audio/mpeg", "wav": "audio/wav", "ogg": "audio/ogg", "oga": "audio/ogg", "opus": "audio/ogg",
    "webm": "video/webm", "mov": "video/quicktime", "png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg",
    "gif": "image/gif", "heic": "image/heic", "pdf": "application/pdf", "md": "text/markdown", "txt": "text/plain",
    "json": "application/json", "csv": "text/csv",
}


@dataclass
class CliError(Exception):
    """A failure, printed as one JSON line on stderr: {"error": {"code", "message", "hint", ...extra}}."""

    code: str
    message: str
    hint: str
    exit: int = EXIT_USAGE
    extra: JSON = field(default_factory=dict)


def error_line(e: CliError) -> str:
    """The one JSON line an error prints on stderr. No key ever shows in it: kabu_live_sk_… keeps only its prefix."""
    line = json.dumps({"error": {"code": e.code, "message": e.message, "hint": e.hint, **e.extra}})
    return re.sub(r"kabu_(live|test)_sk_\w+", r"kabu_\1_sk_***", line)


def emit(obj: JSON, code: int = EXIT_OK, sort: bool = False) -> int:
    """Print a command's one JSON object, and return its exit code."""
    print(json.dumps(obj, sort_keys=sort))
    return code


def utf8_json(obj: JSON) -> bytes:
    """A request body as UTF-8 JSON: \\u escapes cost up to 12 bytes a character against the API's 64 KiB cap."""
    return json.dumps(obj, ensure_ascii=False).encode("utf-8", "replace")  # replace: a lone surrogate from argv


def default_config_path() -> Path:
    if os.environ.get("KABU_CONFIG_PATH"):
        return Path(os.environ["KABU_CONFIG_PATH"])
    return Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") / "kabu" / "config.json"


def load_config() -> JSON:
    try:
        data = json.loads(default_config_path().read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return data if isinstance(data, dict) else {}


def save_config(cfg: JSON) -> Path:
    """Write the config with mode 600 from the first byte: the key never sits in a readable file."""
    path = default_config_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    with os.fdopen(os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w", encoding="utf-8") as out:
        out.write(json.dumps(cfg, indent=2, sort_keys=True) + "\n")
    os.chmod(tmp, 0o600)
    os.replace(tmp, path)
    return path


def resolve_setting(flag: Optional[str], env_keys: Sequence[str], cfg: JSON, cfg_key: str, default: str = "") -> str:
    """The flag, else the first environment variable set, else the saved setting, else the default."""
    for candidate in (flag, *(os.environ.get(k) for k in env_keys), cfg.get(cfg_key)):
        if isinstance(candidate, str) and candidate:
            return candidate
    return default


def base_url(flag: Optional[str], cfg: JSON) -> str:
    url = resolve_setting(flag, ("KABU_BASE_URL", "KABU_API_BASE_URL"), cfg, "base_url", DEFAULT_BASE_URL).rstrip("/")
    if not re.fullmatch(r"https?://[^/\s]+(/\S*)?", url):
        raise CliError("invalid_value", f"The base URL {url!r} is not an http:// or https:// URL.", "Fix --base-url, KABU_BASE_URL "
                       "or the saved one, with: kabu login --base-url https://usekabu.com")
    return url[:-3] if url.endswith("/v2") else url


PROJECT_ENV_KEYS = ("KABU_PROJECT", "AGENT_NOTIFIER_PROJECT")
TEMP_DIRS = tuple({Path(d).resolve() for d in ("/tmp", "/var/tmp", "/var/folders", os.environ.get("TMPDIR") or "/tmp")})
KEPT_DAYS = 7  # an agent session's kept project is removed once unused this long
PROC = Path("/proc")  # Linux's process files


def _git(cwd: Path, *args: str) -> str:
    try:
        out = subprocess.run(["git", *args], cwd=str(cwd), capture_output=True, text=True, timeout=3)
    except (OSError, subprocess.SubprocessError):
        return ""
    return out.stdout.strip() if out.returncode == 0 else ""


def _read_dotenv_project(path: Path) -> str:
    """KABU_PROJECT, else AGENT_NOTIFIER_PROJECT, from one .env file (export, quotes and trailing comments allowed)."""
    try:
        lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError:
        return ""
    found: Dict[str, str] = {}
    for raw in lines:
        line = raw.strip()
        key, sep, val = (line[7:].lstrip() if line.startswith("export ") else line).partition("=")
        key, val = key.strip(), val.strip()
        if line.startswith("#") or not sep or key not in PROJECT_ENV_KEYS:
            continue
        if val[:1] in ("'", '"'):
            end = val.find(val[0], 1)
            val = val[1:end] if end > 0 else val[1:]
        else:
            val = val.split(" #", 1)[0].strip()
        if val:
            found[key] = val
    return next((found[k] for k in PROJECT_ENV_KEYS if found.get(k)), "")


def _git_roots(cwd: Path) -> Tuple[Optional[Path], Optional[Path]]:
    """Return (this worktree's top level, the main worktree's top level)."""
    top = _git(cwd, "rev-parse", "--show-toplevel")
    if not top:
        return None, None
    common = _git(cwd, "rev-parse", "--path-format=absolute", "--git-common-dir")
    main = Path(common).parent if common and Path(common).name == ".git" else Path(top)
    return Path(top).resolve(), main.resolve()


def is_temp(folder: Path) -> bool:
    """A temp folder, or one inside it: it never names a project."""
    return any(folder == d or d in folder.parents for d in TEMP_DIRS)


def named_project(folder: Path) -> Tuple[str, str]:
    """The project a folder names, and how: KABU_PROJECT in .env.local or .env from the folder up to its repository's top,
    then the main worktree's ("dotenv"); else the git repository's name ("git"). ("", "") when it names none, as a temp
    folder does unless it is a worktree of a repository whose main worktree is outside the temp folders."""
    top, main = _git_roots(folder)
    if is_temp(folder) and (main is None or is_temp(main)):
        return "", ""
    dirs = [folder]  # from here up to the repository's top, then the main worktree's
    while top is not None and dirs[-1] != top and dirs[-1].parent != dirs[-1]:
        dirs.append(dirs[-1].parent)
    if main is not None and main not in dirs:
        dirs.append(main)
    for d in dirs:
        for name in (".env.local", ".env"):
            found = _read_dotenv_project(d / name)
            if found:
                return found, "dotenv"
    if main is None:
        return "", ""
    name = _git(folder, "config", "--get", "remote.origin.url").rstrip("/").rsplit("/", 1)[-1].rsplit(":", 1)[-1]
    return (name[:-4] if name.endswith(".git") else name) or main.name, "git"


def resolve_project(cwd: Optional[Path] = None) -> Tuple[Optional[str], JSON]:
    """The project to send without --project (None: the API picks), and the rule that named it (CONTRACT.md, Project
    resolution): the environment, the agent session's root, what the session kept, this folder, the login, a name."""
    for k in PROJECT_ENV_KEYS:
        if os.environ.get(k, "").strip():
            return os.environ[k].strip(), {"rule": "env", "variable": k}
    agent = harness()
    seen: JSON = {"harness": agent[2]} if agent else {}
    variable = {"variable": "CLAUDE_PROJECT_DIR"} if os.environ.get("CLAUDE_PROJECT_DIR") else {}
    found = Path(os.environ["CLAUDE_PROJECT_DIR"]) if variable else process_cwd(agent[0]) if agent else None
    root = str(found.resolve()) if found else ""
    kept = cache_dir() / "sessions" / f"{session_key(agent)}.json" if session_key(agent) else None
    if kept is not None:
        try:
            entry = json.loads(kept.read_text(encoding="utf-8"))
            if entry["root"] == root:  # read back while the root is the folder it was named under; moved, resolve again
                kept.touch()  # in use: kept for another KEPT_DAYS
                return str(entry["project"]), {**entry["source"], "cached": True}
        except (OSError, ValueError, KeyError, TypeError):
            pass  # not kept yet, or unreadable: resolve again
    folders = [("session_root", Path(root))] if root else []
    try:
        folders.append(("cwd", (cwd or Path.cwd()).resolve()))
    except OSError:
        pass  # the current folder was removed: it names nothing, and the root may still
    for rule, folder in folders:
        name, via = named_project(folder)
        if name:
            source = {"rule": rule, **seen, "dir": str(folder), "via": via, **(variable if rule == "session_root" else {})}
            if kept is not None:  # for the session's later calls; files no session used for KEPT_DAYS go
                write_cache(kept, json.dumps({"project": name, "source": source, "root": root}), time.time() - KEPT_DAYS * 86400)
            return name, source
    saved = load_config().get("project")
    if isinstance(saved, str) and saved.strip():
        return saved.strip(), {"rule": "login", **seen}
    home, host = Path(os.path.expanduser("~")).resolve(), socket.gethostname().split(".", 1)[0] or "host"
    for rule, folder in folders:  # never a name for /, the home folder, a missing folder (Linux: "… (deleted)") or a temp one
        if folder.parent != folder and folder != home and folder.is_dir() and not is_temp(folder):
            return f"{folder.name}@{host}", {"rule": "folder", **seen, "dir": str(folder)}
    return None, {"rule": "default", **seen}


def harness() -> Optional[Tuple[int, str, str]]:
    """The agent session kabu runs in: its harness, the nearest ancestor named as one (process id, start time, name)."""
    return next((p for p in ancestors() if p[2] in AGENT_PROCESSES), None)


def session_key(agent: Optional[Tuple[int, str, str]]) -> str:
    """An agent session's key: its harness's process id and a hash of its start time, so a reused id never matches;
    "" when there is no harness or its start time is unknown."""
    return f"{agent[0]}-{hashlib.sha256(agent[1].encode()).hexdigest()[:16]}" if agent and agent[1] else ""


def cache_dir() -> Path:
    return Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache") / "kabu"


def write_cache(path: Path, text: str, stale: float = 0) -> None:
    """Write a file of kabu's cache, after removing the files of its folder unused since `stale`, a time. The cache only
    saves work, so a failure, or a file read half written, only means resolving again."""
    try:
        path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
        for old in path.parent.iterdir() if stale else ():
            try:
                if old.stat().st_mtime < stale:
                    old.unlink()
            except OSError:
                pass  # gone meanwhile, by another call's clean-up, or a folder: never in the way of the write
        path.write_text(text, encoding="utf-8")
    except OSError:
        pass


def gone_sessions(base: str) -> Tuple[Path, List[str]]:
    """The sessions the API at `base` reported gone (deleted with their project), which kabu never names again, and their
    file: one per base URL, as one API's ids mean nothing to another's."""
    path = cache_dir() / f"gone-{hashlib.sha256(base.encode()).hexdigest()[:12]}"
    try:
        return path, path.read_text(encoding="utf-8").split()
    except OSError:
        return path, []


def process_cwd(pid: int) -> Optional[Path]:
    """A process's working folder, or None: /proc on Linux; on macOS, libproc's proc_pidinfo, which lsof calls, in 0.3 ms
    where lsof took 0.3 s and more."""
    try:
        if sys.platform != "darwin":
            return Path(os.readlink(f"/proc/{pid}/cwd"))
        import ctypes
        size, at = 2352, 152  # struct proc_vnodepathinfo: the cwd's struct vnode_info, then its path, MAXPATHLEN 1024
        info = ctypes.create_string_buffer(size)
        if ctypes.CDLL("/usr/lib/libproc.dylib").proc_pidinfo(pid, 9, 0, info, size) != size:  # 9: PROC_PIDVNODEPATHINFO
            return None
        path = info.raw[at:at + 1024].split(b"\0", 1)[0]
        return Path(os.fsdecode(path)) if path else None
    except (OSError, ValueError, ImportError, AttributeError):  # gone, another user's, no libproc, or a Python without ctypes
        return None


# Process names of agent harnesses, looked for among kabu's ancestors.
AGENT_PROCESSES = {"claude": "claude-code", "copilot": "copilot-cli", "codex": "codex", "cursor-agent": "cursor", "gemini": "gemini-cli",
                   "opencode": "opencode", "aider": "aider", "goose": "goose", "amp": "amp", "pi": "pi"}
# Variables harnesses set for the commands they run; never their configuration, such as CODEX_HOME.
AGENT_VARIABLES = {"CLAUDECODE": "claude-code", "CLAUDE_CODE_ENTRYPOINT": "claude-code", "CURSOR_AGENT": "cursor",
                   "CURSOR_TRACE_ID": "cursor", "GEMINI_CLI": "gemini-cli", "CODEX_THREAD_ID": "codex", "CODEX_SANDBOX": "codex",
                   "PI_CODING_AGENT": "pi"}


class Proc(NamedTuple):
    """A process: its parent, start time ("" when unknown: it tells a process from a later one that reuses its id),
    executable name, and whether it is a zombie (exited, its parent has not reaped it)."""

    ppid: int
    start: str
    name: str
    zombie: bool = False


def processes(pids: Optional[Sequence[int]] = None) -> Dict[int, Proc]:
    """Processes by id: those named, else kabu's ancestors. Names only: arguments can hold secrets. Linux: /proc, read
    with the boot id; elsewhere one ps call, which may not show start times (BusyBox), and names are enough then."""
    procs: Dict[int, Proc] = {}
    if (PROC / "self" / "stat").exists():
        try:
            boot = (PROC / "sys" / "kernel" / "random" / "boot_id").read_text().strip()
        except OSError:
            boot = ""
        todo = list(pids) if pids is not None else [os.getppid()]
        while todo and len(procs) < 32 + len(pids or ()):
            pid = todo.pop()
            if pid <= 0 or pid in procs:
                continue
            try:
                stat = (PROC / str(pid) / "stat").read_text(errors="replace")
            except OSError:
                continue
            rest = stat[stat.rindex(")") + 2:].split()  # the fields after the name, which may hold spaces or parentheses
            procs[pid] = Proc(int(rest[1]), f"{boot}:{rest[19]}" if boot else "", stat[stat.index("(") + 1:stat.rindex(")")].lower(),
                              rest[0] == "Z")
            todo += [] if pids is not None else [procs[pid].ppid]
        return procs
    which, state = (["-A"], "") if pids is None else (["-p", ",".join(map(str, pids))], "stat=,")  # zombies, for the reporter
    starts = ((f"pid=,ppid=,{state}lstart=,comm=", r"(\w+ +\w+ +\d+ +[\d:]+ +\d+) +"), (f"pid=,ppid=,{state}comm=", "()"))  # with, then without
    for columns, start in starts if pids is None or pids else ():
        try:
            listing = subprocess.run(["ps", *which, "-o", columns], capture_output=True, text=True, errors="replace", timeout=2,
                                     env={**os.environ, "LC_ALL": "C", "TZ": "UTC0"}).stdout  # one format for start times
        except (OSError, subprocess.SubprocessError):
            break
        rows = re.findall(r"^\s*(\d+)\s+(\d+)\s+" + (r"(\S+)\s+" if state else "()") + start + r"(.+)$", listing, re.MULTILINE)
        procs = {int(pid): Proc(int(ppid), " ".join(when.split()), Path(comm).name.lower(), flags.startswith("Z"))
                 for pid, ppid, flags, when, comm in rows}
        if procs:
            break
    return procs


@functools.lru_cache(maxsize=None)
def ancestors() -> Tuple[Tuple[int, str, str], ...]:
    """kabu's ancestor processes, nearest first: process id, start time ("" when unknown) and executable name."""
    procs = processes()
    found: List[Tuple[int, str, str]] = []
    pid = os.getppid()
    while pid in procs and len(found) < 32:
        found.append((pid, procs[pid].start, procs[pid].name))
        pid = procs[pid].ppid
    return tuple(found)


def detect_agent(env: Mapping[str, str], names: Sequence[str]) -> str:
    """The harness running kabu: the nearest ancestor it knows, else AI_AGENT, else a variable a harness set, else kabu-cli."""
    found = [AGENT_PROCESSES[n] for n in names if n in AGENT_PROCESSES]
    found.append(re.sub(r"[^a-z0-9]+", "-", env.get("AI_AGENT", "").lower()).strip("-"))
    found += [agent for var, agent in AGENT_VARIABLES.items() if env.get(var, "").lower() not in ("", "0", "false")]
    return next((agent for agent in found if agent), "kabu-cli")[:64]


def resolve_agent(flag: Optional[str] = None) -> str:
    """--agent, KABU_AGENT (or AGENT_NOTIFIER_AGENT), the agent saved by kabu login, then the harness detected."""
    return resolve_setting(flag, ("KABU_AGENT", "AGENT_NOTIFIER_AGENT"), load_config(), "agent") or detect_agent(
        os.environ, [name for _, _, name in ancestors()])


class NoRedirect(urllib.request.HTTPRedirectHandler):
    """A 3xx is an HTTP error, never followed: urllib would send the bearer key on to wherever it points."""

    def redirect_request(self, req: urllib.request.Request, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None:
        return None


OPEN = urllib.request.build_opener(NoRedirect).open


@dataclass
class Api:
    """Kabu API v2 over urllib: JSON in and out, bounded retries; an API problem becomes a CliError (exit 3)."""

    base: str
    key: str = ""
    timeout: float = 30.0
    retries: int = RETRIES

    @classmethod
    def connect(cls, a: JSON) -> Api:
        """The client for --api-key and --base-url, else the environment, else the saved login."""
        cfg = load_config()
        key = resolve_setting(a.get("api_key"), ("KABU_API_KEY", "AGENT_NOTIFIER_API_KEY"), cfg, "api_key").strip()
        if not key:
            raise CliError("missing_api_key", "No Kabu API key.", "Set KABU_API_KEY=kabu_live_sk_…, or run: kabu login --api-key KEY")
        if not re.fullmatch(r"[\x21-\x7e]+", key):  # printable ASCII only: a header error would print the key whole
            raise CliError("invalid_api_key", "The API key (--api-key, KABU_API_KEY or the saved one) has a space, line break or "
                           "other character no key has.", "Save it again with: kabu login --api-key - < key.txt")
        return cls(base_url(a.get("base_url"), cfg), key, a.get("timeout") or 30.0)

    def call(self, method: str, path: str, body: Optional[JSON] = None, *, wait: int = 0, idem: str = "",
             form: Optional[Dict[str, str]] = None) -> JSON:
        """One API call; `idem` is the Idempotency-Key that makes a retried create return the first result."""
        headers = {"Accept": "application/json", "User-Agent": f"kabu-cli/{VERSION}",
                   **({"Authorization": f"Bearer {self.key}"} if self.key else {}), **({"Idempotency-Key": idem} if idem else {})}
        data = None
        if body is not None:
            data, headers["Content-Type"] = utf8_json(body), "application/json"
        if form is not None:
            data, headers["Content-Type"] = urllib.parse.urlencode(form).encode(), "application/x-www-form-urlencoded"
        raw = self.request(method, self.base + path + (f"?wait={wait}" if wait else ""), data, headers, self.timeout + wait)
        try:
            parsed = json.loads(raw) if raw.strip() else {}
        except ValueError:
            raise CliError("bad_response", f"{method} {path} did not return JSON.",
                           "Check that KABU_BASE_URL points at the Kabu API.", EXIT_TRANSPORT) from None
        return parsed if isinstance(parsed, dict) else {"data": parsed}

    def request(self, method: str, url: str, data: Optional[bytes], headers: Dict[str, str], timeout: float) -> bytes:
        """Send, and retry with the same headers, so the same Idempotency-Key, up to self.retries times."""
        attempt = 0
        while True:
            try:
                with OPEN(urllib.request.Request(url, data, headers, method=method), timeout=timeout) as resp:
                    return bytes(resp.read())
            except urllib.error.HTTPError as exc:
                error, delay = self.problem(exc, method, url, attempt)
            except (OSError, http.client.HTTPException) as exc:  # refused, reset, timed out
                error = CliError("network_error", f"{method} {url} failed: {getattr(exc, 'reason', exc)}.",
                                 "Check the network and KABU_BASE_URL, then run the same command again.", EXIT_TRANSPORT)
                delay = 0.5 * 2 ** attempt
            if delay is None or attempt == self.retries:
                raise error
            time.sleep(delay)
            attempt += 1

    @staticmethod
    def problem(exc: urllib.error.HTTPError, method: str, url: str, attempt: int) -> Tuple[CliError, Optional[float]]:
        """The API's problem (RFC 9457, or an OAuth error) as a CliError, and the seconds to wait before a retry (None: none)."""
        try:
            problem = json.loads(exc.read() or b"{}")
        except (ValueError, OSError):
            problem = {}
        problem = problem if isinstance(problem, dict) else {}
        status, after = exc.code, exc.headers.get("Retry-After") or ""
        code = str(problem.get("code") or problem.get("error") or f"http_{status}")
        message = problem.get("detail") or problem.get("error_description") or problem.get("title") or f"{method} {url}: HTTP {status}."
        fallback = "Run the same command again shortly." if status >= 500 else "Check the request, and the key and base URL (kabu whoami)."
        extra = {"status": status, **{k: problem[k] for k in ("param", "request_id") if problem.get(k)}}
        error = CliError(code, str(message), str(problem.get("hint") or fallback), EXIT_TRANSPORT, extra)
        if status < 500 and code != "idempotency_key_in_use" and not (status == 429 and after):
            return error, None  # quota_exceeded is a 429 without Retry-After: it lasts until the plan resets
        delay = float(after) if re.fullmatch(r"\d+(\.\d+)?", after) else 0.5 * 2 ** attempt
        return error, (delay if delay <= MAX_RETRY_AFTER else None)

    def upload(self, path: Path, idem: str) -> str:
        """Upload one file (POST /v2/media/uploads, then PUT the bytes) and return its med_ id."""
        raw = path.read_bytes()
        ctype = upload_type(path.name)
        up = self.call("POST", "/v2/media/uploads", {"filename": path.name, "content_type": ctype, "size": len(raw),
                                                      "sha256": hashlib.sha256(raw).hexdigest()}, idem=idem)
        media, url = up.get("media") or {}, up.get("upload_url")
        if not isinstance(url, str) or not isinstance(media.get("id"), str):
            raise CliError("bad_response", f"The upload of {path.name} returned no upload_url or media id.",
                           "Run the same command again.", EXIT_TRANSPORT)
        if media.get("status") != "ready":  # a replayed key may find the bytes already in place
            headers = {"Content-Type": ctype, **{k: v for k, v in (up.get("headers") or {}).items() if isinstance(v, str)}}
            self.request("PUT", self.base + url if url.startswith("/") else url, raw, headers, self.timeout)
        return str(media["id"])


def upload_type(name: str) -> str:
    """The Content-Type an upload of the file called name declares: UPLOAD_TYPES, mimetypes, then octet-stream."""
    ext = name.rpartition(".")[2].lower() if "." in name else ""
    return UPLOAD_TYPES.get(ext) or mimetypes.guess_type(name)[0] or "application/octet-stream"


def field_pair(cmd: str, item: str) -> JSON:
    """--field KEY=VALUE as the contract's {label, value}."""
    key, sep, val = (s.strip() for s in item.partition("="))
    if not sep or not 0 < len(key) <= 40 or len(val) > 200:
        raise CliError("invalid_value", f"--field {item!r} is not KEY=VALUE with a key of 1 to 40 characters and a value of up to 200.",
                       f"Example: kabu {SAMPLES[cmd]} --field env=prod")
    return {"label": key, "value": val}


def link(cmd: str, item: str) -> JSON:
    """--link LABEL=URL, or a bare URL; a bare URL's label is filled in from --url-title, else Open."""
    name, sep, url = item.partition("=")
    bare = not sep or "://" in name
    name, url = ("", item) if bare else (name.strip(), url)
    if not (bare or 0 < len(name) <= 40) or len(url) > 2048 or not re.fullmatch(r"[A-Za-z][A-Za-z0-9+.-]*:\S+", url):
        raise CliError("invalid_value", f"--link {item!r} is not LABEL=URL with a label of 1 to 40 characters and an "
                       "absolute URL of up to 2048.", f'Example: kabu {SAMPLES[cmd]} --link "PR=https://github.com/o/r/pull/7"')
    return {"label": name, "url": url}


def json_object(cmd: str, raw: str) -> JSON:
    """--args: one JSON object, finite numbers only, up to the contract's 16 KiB."""
    hint = f"""Example: kabu {SAMPLES[cmd]} --action db.migrate --args '{{"version": 42}}'"""
    try:
        args = json.loads(raw)
        size = len(json.dumps(args, allow_nan=False, ensure_ascii=False, separators=(",", ":")).encode())
    except ValueError as exc:
        raise CliError("invalid_value", f"--args is not valid JSON: {exc}.", hint) from None
    if not isinstance(args, dict) or size > 16 << 10:
        raise CliError("invalid_value", "--args must be one JSON object of up to 16 KiB.", hint)
    return args


def attachment(cmd: str, raw: str) -> Path:
    path = Path(os.path.expanduser(raw))  # unlike Path.expanduser, never raises for ~nosuchuser
    size = path.stat().st_size if path.is_file() and os.access(path, os.R_OK) else -1
    if size < 0:
        raise CliError("file_not_found", f"--file {raw}: no such file, or kabu cannot read it.",
                       f"Check the path. Example: kabu {SAMPLES[cmd]} --file ./a.png")
    if not 0 < size <= 2 ** 31:
        raise CliError("invalid_file", f"--file {raw} is {size} bytes; an attachment takes 1 byte to 2 GiB.",
                       "Attach a non-empty file under 2 GiB.")
    return path


@dataclass(frozen=True)
class Opt:
    """One option: how it parses, what it accepts, its default and where it lands in the request body."""

    flags: Tuple[str, ...]  # the first long flag names the value
    help: str
    example: str = ""  # a valid value, for help and error hints
    path: str = ""  # request body field, "a.b" to nest; empty when code builds it
    kind: str = "text"  # text, int, float, flag or list
    limit: int = 0  # most characters (text) or uses (list)
    bounds: Tuple[float, float] = (0, 0)  # range of an int or float
    choices: Tuple[str, ...] = ()
    pattern: str = ""  # a regex the whole value must match
    needs: str = ""  # an option this one requires
    ask: bool = False  # only on an ask
    metavar: Optional[str] = None
    default: Optional[Callable[[], Any]] = None  # the value when the option is not given
    convert: Optional[Callable[[str, str], Any]] = None  # (command, checked value) to what the request carries

    @property
    def dest(self) -> str:
        return self.flags[0].lstrip("-").replace("-", "_")


PRIORITY_NUMBERS = {"-1": "low", "0": "normal", "1": "high", "2": "high"}  # kabu.py's; agent keys cannot send urgent
ACCENTS = ("neutral", "blue", "green", "amber", "red", "purple", "teal", "pink", "orange", "indigo", "mint", "brown", "slate")
CONNECTION = (
    Opt(("--api-key",), "API key (default: KABU_API_KEY, then the saved login)", "kabu_live_sk_…"),
    Opt(("--base-url",), "API base URL (default: KABU_BASE_URL, the saved login, https://usekabu.com)", "https://usekabu.com"),
    Opt(("--timeout",), "seconds to wait for each response, 1 to 600 (default 30)", "30", kind="float", bounds=(1, 600), metavar="SECONDS"),
    Opt(("--json",), "ignored: kabu always prints one JSON object (kabu.py's flag)", kind="flag"),
)
PROJECT = Opt(("--project", "-p"), "project, max 64 (default: KABU_PROJECT, else your agent session's repo: see kabu whoami)",
              "payments-api", path="project", limit=64)
AGENT = Opt(("--agent", "-a"), "agent, max 64 (default: KABU_AGENT, the login's, the harness)", "claude-code", path="agent", limit=64,
            default=resolve_agent)
MESSAGE = (
    Opt(("--title", "-t"), "title, max 120 characters (default: the message's first line)", "Build passed"),
    Opt(("--message", "-m", "--body"), "message, Markdown, max 8192 characters; - reads stdin", "All 412 tests passed"),
    Opt(("--ask",), "the question (kabu.py's form: kabu --ask Q is kabu ask Q)", "Deploy to prod?", ask=True, metavar="QUESTION"),
    Opt(("--ask-text",), "a question answered by typing: --ask Q --type text", "Release name?", ask=True, metavar="QUESTION"),
    PROJECT, AGENT,
    Opt(("--source", "-s"), "machine label, max 120 (default: KABU_SOURCE, the hostname)", "ci-runner-3", path="source", limit=120,
        default=lambda: resolve_setting(None, ("KABU_SOURCE",), load_config(), "source", socket.gethostname()[:120])),
    Opt(("--session",), "a ses_… session to show it in (default: KABU_SESSION)", "ses_01J9Z", path="session", pattern=r"ses_\S+",
        default=lambda: os.environ.get("KABU_SESSION")),
    Opt(("--priority",), "low, normal, high or urgent; or -1, 0, 1, 2", "high", path="priority",
        choices=("low", "normal", "high", "urgent", *PRIORITY_NUMBERS), convert=lambda cmd, v: PRIORITY_NUMBERS.get(v, v)),
    Opt(("--expires-in", "--ask-timeout"), "seconds until it expires, 60 to 2592000 (an ask: 300)", "900", path="expires_in",
        kind="int", bounds=(60, 2592000), metavar="SECONDS"),
    Opt(("--sensitive",), "hide the content on lock screens and previews", path="sensitive", kind="flag"),
    Opt(("--collapse-key",), "a later notify with this key replaces it, max 64", "nightly", path="collapse_key", limit=64, metavar="KEY"),
    Opt(("--accent",), "colour: " + ", ".join(ACCENTS), "teal", path="appearance.accent", choices=ACCENTS),
    Opt(("--icon",), "this message's icon, max 64: an SF Symbol name, which the app draws (lowercase words and dots), or a med_… id",
        "checkmark.seal.fill", path="appearance.icon", limit=64, pattern=r"med_\S+|[a-z0-9]+(\.[a-z0-9]+)*"),
    Opt(("--field",), "a key fact, up to 12 (key max 40, value max 200)", "env=prod", path="fields", kind="list", limit=12,
        metavar="KEY=VALUE", convert=field_pair),
    Opt(("--link", "--url"), "a link, up to 4: LABEL=URL, or a bare URL labelled by --url-title (default Open)",
        "PR=https://github.com/o/r/pull/7", path="links", kind="list", limit=4, metavar="LABEL=URL", convert=link),
    Opt(("--url-title",), "the label of links given without one, max 40", "Open PR", needs="link", limit=40, metavar="LABEL"),
    Opt(("--code",), "a code block, max 8000 characters; - reads stdin", "print('hi')", limit=8000),
    Opt(("--language",), "the code block's language, max 30", "python", needs="code", limit=30),
    Opt(("--diff",), "a diff block, max 16000 characters; - reads stdin", "-", limit=16000),
    Opt(("--diff-path",), "the diff's file name, max 300 (a label: nothing is read)", "src/app.py", needs="diff", limit=300, metavar="PATH"),
    Opt(("--quote",), "a quote block, max 2000 characters; - reads stdin", "Ship it", limit=2000),
    Opt(("--action",), "the action to approve, max 100", "db.migrate", path="action.name", limit=100),
    Opt(("--args",), "the action's arguments: a JSON object, max 16 KiB", '{"version": 42}', path="action.args", needs="action",
        metavar="JSON", convert=json_object),
    Opt(("--action-description",), "what it does, max 2000", "Adds an index", path="action.description", needs="action", limit=2000),
    Opt(("--as-audio",), "also speak it (generated from the message)", path="voice.speak", kind="flag"),
    Opt(("--voice", "--tts-voice"), "the voice to speak with, max 60 (implies --as-audio)", "alloy", path="voice.voice", limit=60),
    Opt(("--speed", "--tts-speed"), "speech speed, 0.5 to 2.0 (implies --as-audio)", "1.25", path="voice.speed", kind="float", bounds=(0.5, 2)),
    Opt(("--file", "--file-path", "--attach", "--attachment", "--image"), "attach a file, up to 10 (a spoken notify: 9)", "report.png",
        kind="list", limit=10, metavar="PATH", convert=attachment),
    Opt(("--type",), "confirm (approve or reject), choice, multi or text (default: choice with -c, else confirm)", "choice",
        choices=("confirm", "choice", "multi", "text"), ask=True),
    Opt(("--choice", "-c", "--option"), "an option, 2 to 12, label max 40; ID:LABEL sets its id", "staging", kind="list", limit=12, ask=True),
    Opt(("--allow-reply", "--allow-free-form"), "the human may type a reply instead", path="ask.allow_reply", kind="flag", ask=True),
    Opt(("--irreversible",), "the human must unlock their device; never decided on expiry", path="ask.irreversible", kind="flag", ask=True),
    Opt(("--no-wait",), "do not wait: print the id, then poll with kabu status ID --wait 50", kind="flag", ask=True),
    Opt(("--dry-run",), "the API validates it and shows the push and Live Activity text; nothing is sent", kind="flag"),
) + CONNECTION
STATUS = (Opt(("--wait",), "hold up to N seconds, 0 to 50, while an ask is pending", "50", kind="int", bounds=(0, 50)),) + CONNECTION
PROGRESS_RULE = ("Report progress only when it is measurable (items done of a known total) or for known steps; set\n"
                 "--progress-label to what is counted; never estimate a percentage; otherwise send only a --headline.")
SESSION_UPDATE = (
    Opt(("--headline",), "the Live Activity's line, max 80", "Running tests", path="headline", limit=80),
    Opt(("--progress",), "items done of a known total: N/M, such as 54/172; only when measured, never an estimated percentage", "54/172", path="progress", pattern=r"\d+(\.\d+)?/(?!0+(\.0*)?$)\d+(\.\d+)?",
        metavar="N/M", convert=lambda cmd, v: {"value": float(v.split("/")[0]), "total": float(v.split("/")[1])}),
    Opt(("--progress-label",), "what is counted, such as projects, max 60", "projects", path="progress.label", needs="progress", limit=60, metavar="TEXT"),
)
SESSION_CREATE = (
    Opt(("--title",), "what the session is for, max 80 (required)", "Fix flaky auth test", path="title", limit=80),
    PROJECT, AGENT,
    Opt(("--external-id",), "your run id, max 120: start gets or creates the session of project, agent and id", "run-77",
        path="external_id", limit=120, metavar="ID"),
)
SESSION_END = (
    Opt(("--outcome",), "success, failure or cancelled", "success", path="outcome", choices=("success", "failure", "cancelled")),
    Opt(("--summary",), "how it went, max 500", "PR opened", path="summary", limit=500),
    Opt(("--dismiss-after",), "seconds the ended Live Activity stays, 0 to 14400", "600", path="dismiss_after", kind="int", bounds=(0, 14400)),
) + CONNECTION
KEY = Opt(("--api-key",), "the key to save (kabu_live_sk_… or kabu_test_sk_…); - reads it from stdin", "kabu_live_sk_…",
          pattern=r"kabu_[A-Za-z0-9_]+")
LOGIN = (
    KEY,
    Opt(("--device",), "connect this computer: enter the code it prints in the Kabu app's Connect a computer; its own key is saved",
        kind="flag"),
    Opt(("--name",), "the computer's name shown when approving, max 80 (default: the hostname)", "work-laptop", limit=80,
        default=lambda: socket.gethostname().split(".")[0][:80]),
    Opt(("--base-url",), "API base URL to save (default: KABU_BASE_URL, the saved one, https://usekabu.com)", "https://usekabu.com"),
    Opt(("--project", "-p"), "default project to save, max 64", "payments-api", limit=64),
    Opt(("--agent", "-a"), "default agent to save, max 64", "claude-code", limit=64),
)
PRESENCE_SETTINGS = (
    Opt(("--harness",), "the harness whose settings get the hook: claude (Claude Code), copilot (Copilot CLI), codex, gemini "
        "(Gemini CLI), cursor (Cursor and its CLI) or pi (an extension)", "claude", choices=("claude", "copilot", "codex", "gemini", "cursor", "pi")),
    Opt(("--settings",), "its settings file (default: CLAUDE_CONFIG_DIR's or ~/.claude's settings.json; COPILOT_HOME's or "
        "~/.copilot's; CODEX_HOME's or ~/.codex's hooks.json; ~/.gemini/settings.json; ~/.cursor/hooks.json; "
        "PI_CODING_AGENT_DIR's or ~/.pi/agent's extensions/kabu-presence.js)", "~/.claude/settings.json", metavar="PATH"),
)
ENV_HELP = "\n\nEnvironment: KABU_API_KEY, KABU_BASE_URL, KABU_PROJECT, KABU_AGENT, KABU_SESSION and more: kabu --help."
SESSION_ID = ("id", "?", "a ses_… id (default: KABU_SESSION)")
TEXT_ARG = ("text", "*", "MESSAGE, or TITLE MESSAGE; - reads the message from stdin")
SAMPLES = {"send": 'send "Build passed"', "ask": 'ask "Deploy to prod?"'}
SUBCOMMAND_SAMPLES = {"session": 'start --title "Fix auth"', "presence": "status"}
NAMES = ("send", "ask", "status", "whoami", "session", "login", "mcp", "presence", "guide", "help")
LEGACY = ("daemon", "listen", "service", "steer", "identity", "project")  # kabu.py's commands that kabu does not have
YES_NO = {("yes", "no"), ("y", "n"), ("approve", "reject")}  # the -c pairs that make a bare --ask kabu.py's yes or no
RENAMED = {"--check-status": "Use: kabu status MSG_ID", "--progress": "Use: kabu session update --progress 3/8",
           "--quiet": "Drop the JSON: kabu send … > /dev/null", "-q": "Drop the JSON: kabu send … > /dev/null",
           "--id-only": "Read the id: kabu send … | jq -r .id", "--fyi": "An FYI is a notify: kabu send TITLE MESSAGE",
           "--category": "Use --field category=VALUE.", "--metadata": "Use --field KEY=VALUE, a fact the human sees.",
           "--max-retries": "kabu retries 3 times by itself.", "--client-event-id": "kabu sends an Idempotency-Key by itself."}
GUIDE = """\nFor coding agents:
- One argument is the message, two are TITLE MESSAGE; quote multi-word text; - reads stdin.
- Limits: title 120, message 8192, --field 12 (key 40, value 200), --link 4, --file 10, -c 2 to 12 (label 40).
- Check first: --dry-run has the API validate it and sends nothing; kabu whoami shows the key, project and agent.
- Yes or no: kabu ask "Deploy?" exits 0 only when approved (10 rejected); --type confirm -c Ship -c Wait names the buttons.
- With -c it is a choice, which exits 0 whatever is chosen: branch on .option, never on the exit code.
- If a tool call may time out before a human answers: kabu ask … --no-wait, then kabu status ID --wait 50.
  kabu status exits 0 whatever the answer: read .status and .resolution.decision.
- Exit 3 with an error that names an ask's id: the ask is still open, so wait with kabu status ID --wait 50, never ask
  again. Exit 21 with ask_gone: it was deleted with its project and will never be answered; ask again if still needed.
- The project is your agent session's (the repo it started in) whatever folder kabu runs in, never a temp folder's;
  .project_source says why. KABU_PROJECT and KABU_AGENT override; KABU_SESSION puts messages in a session.
- kabu.py's calls work too: kabu --title T "message" --file-path X; kabu --ask Q -c Yes -c No is yes or no (No exits 10)."""


def example(cmd: str, *opts: Opt) -> str:
    """A corrected command line for an error hint."""
    uses = [o.flags[0] + (f" {shlex.quote(o.example)}" if o.example else "") for o in opts]
    return " ".join(["Example: kabu", SAMPLES.get(cmd, cmd), *uses])


class Parser(argparse.ArgumentParser):
    """argparse that raises a CliError instead of printing usage and exiting."""

    color: bool  # Python 3.14+ colours help on a terminal; kabu never colours

    def error(self, message: str) -> NoReturn:
        raise CliError("usage", f"{self.prog}: {message}.", f"Run: {self.prog} --help")


def parse(cmd: str, args: Sequence[str]) -> JSON:
    """Parse one command's arguments; an unknown flag fails with the closest real one."""
    spec = COMMANDS[cmd]
    p = Parser(prog=f"kabu {cmd}", description=spec.about, epilog="examples:\n" + spec.examples + ENV_HELP, allow_abbrev=False,
               formatter_class=argparse.RawDescriptionHelpFormatter)
    p.color = False
    if spec.positional:
        name, nargs, help_ = spec.positional
        p.add_argument(name, nargs=nargs, help=help_, metavar=name.upper())
    for o in spec.opts:
        kw: JSON = {"action": "store_true"} if o.kind == "flag" else {"metavar": o.metavar}
        if o.kind == "list":
            kw.update(action="append", default=[])
        p.add_argument(*o.flags, dest=o.dest, help=argparse.SUPPRESS if o.ask and cmd == "send" else o.help, **kw)
    ns, extra = p.parse_known_intermixed_args(list(args))
    if extra:
        flag = extra[0].split("=", 1)[0]
        if not flag.startswith("-"):
            shown = "" if cmd == "login" else f" {extra[0]!r}"  # never echo what may be a key
            raise CliError("unexpected_argument", f"Unexpected argument{shown}.", f"Run: kabu {cmd} --help")
        by_flag = {f: o for o in spec.opts for f in o.flags}
        close = [] if flag in RENAMED else difflib.get_close_matches(flag, by_flag, n=1, cutoff=0.8)
        raise CliError("unknown_flag", f"kabu {cmd} has no {flag} flag." + (f" Did you mean {close[0]}?" if close else ""),
                       RENAMED.get(flag) or (example(cmd, by_flag[close[0]]) if close else f"Run: kabu {cmd} --help"))
    return vars(ns)


def given(value: Any) -> bool:
    return value is not None and value is not False and value != []


def check(cmd: str, opts: Sequence[Opt], a: JSON, asking: bool = False) -> JSON:
    """Fill in each row's default and check every option given against its row, before any HTTP call. Converts the values
    in place, and returns the request body fields the rows name (a.b nests)."""
    by_dest = {o.dest: o for o in opts}
    body: JSON = {}
    for o in opts:
        if o.default and not given(a.get(o.dest)):
            a[o.dest] = o.default()
        if not given(a.get(o.dest)):
            continue
        if o.ask and not asking:
            raise CliError("ask_only", f"{o.flags[0]} only applies to an ask.", example("ask", o))
        if o.needs and not given(a.get(o.needs)):
            raise CliError("missing_flag", f"{o.flags[0]} needs {by_dest[o.needs].flags[0]}.", example(cmd, by_dest[o.needs], o))
        a[o.dest] = value(cmd, o, a[o.dest])
        if o.path:
            parent, _, key = o.path.rpartition(".")
            node = body.setdefault(parent, {}) if parent else body
            node[key] = a[o.dest]
    return body


def value(cmd: str, o: Opt, raw: Any) -> Any:
    """One option's value, checked against its limit, range, choices or pattern, then converted."""
    flag = o.flags[0]
    if o.kind == "flag":
        return True
    if o.kind == "list":
        if o.limit and len(raw) > o.limit:
            raise CliError("too_many", f"{flag} is given {len(raw)} times; the limit is {o.limit}.", example(cmd, o))
        return [o.convert(cmd, item) for item in raw] if o.convert else raw
    text = str(raw)
    if o.kind in ("int", "float"):
        lo, hi = o.bounds
        number = float(text) if re.fullmatch(r"\d+" if o.kind == "int" else r"\d+(\.\d+)?", text) else float("nan")
        if not lo <= number <= hi:
            kind = "a whole number" if o.kind == "int" else "a number"
            raise CliError("invalid_value", f"{flag} must be {kind} from {lo} to {hi}, not {text!r}.", example(cmd, o))
        return int(number) if o.kind == "int" else number
    if o.choices and text not in o.choices:
        close = difflib.get_close_matches(text, o.choices, n=1)
        raise CliError("invalid_value", f"{flag} {text!r} is not one of {', '.join(o.choices)}."
                       + (f" Did you mean {close[0]!r}?" if close else ""), example(cmd, o))
    if o.limit and len(text) > o.limit:
        raise CliError("too_long", f"{flag} is {len(text)} characters; the limit is {o.limit}.", example(cmd, o))
    if o.pattern and not re.fullmatch(o.pattern, text):
        shown = "" if flag == "--api-key" else f" {text!r}"  # never echo a key
        raise CliError("invalid_value", f"{flag}{shown} is not valid; it looks like {o.example}.", example(cmd, o))
    return o.convert(cmd, text) if o.convert else text


def texts(cmd: str, a: JSON) -> Tuple[str, str]:
    """The title (default: the message's first line) and the body. One argument is the message, two are TITLE MESSAGE,
    and a value of - is read from stdin, for one value at most."""
    named = [k for k in ("title", "ask", "ask_text") if a[k] is not None]
    if len(named) > 1:
        raise CliError("conflicting_flags", "--title, --ask and --ask-text all set the title: give one.", f'{example("ask")} "Canary is green"')
    a["title"] = a[named[0]] if named else None
    words, free = a["text"], [k for k in ("title", "message") if a[k] is None]
    if len(words) > len(free):
        raise CliError("unexpected_argument", f"Too many arguments: {' '.join(map(repr, words))}.",
                       f'One argument is the message, two are TITLE MESSAGE; quote multi-word text. {example(cmd)} "412 passed"')
    a.update(zip(free[-len(words):], words))
    dashes = [k for k in ("message", "code", "diff", "quote") if a[k] == "-"]
    if len(dashes) > 1:
        raise CliError("conflicting_flags", f"Only one value can read stdin (-), not {len(dashes)}: {', '.join(dashes)}.",
                       f"Pass the others as text. Example: git diff | kabu {SAMPLES[cmd]} --diff - --quote 'Looks good'")
    if dashes:
        a[dashes[0]] = sys.stdin.buffer.read().decode("utf-8", "replace")
    message = (a["message"] or "").strip()
    title = (a["title"] or "").strip() or message.split("\n", 1)[0][:TITLE_MAX].strip()
    if not title:
        raise CliError("missing_text", "Nothing to send: give a message.", example(cmd))
    if len(title) > TITLE_MAX:
        raise CliError("too_long", f"The title is {len(title)} characters; the limit is {TITLE_MAX}.",
                       f'Put the detail in the message. Example: kabu {cmd} "Deploy to prod?" "Canary is green"')
    if len(message) > BODY_MAX:
        raise CliError("too_long", f"The message is {len(message)} characters; the limit is {BODY_MAX}.",
                       f'Shorten it, or attach the whole text. Example: kabu {cmd} "Nightly report" "Summary" --file report.md')
    return title, message or title


def option(index: int, choice: str) -> Dict[str, str]:
    """-c LABEL, or -c ID:LABEL when ID is an id (a letter, then letters, digits, _ . or -) and no space follows the colon,
    so 10:00 and "Retry: once" stay whole labels. The id defaults to the label in lower case, with _ for anything else."""
    split = re.fullmatch(r"([A-Za-z][A-Za-z0-9_.-]{0,63}):(\S.*)", choice.strip())
    label = (split.group(2) if split else choice).strip()
    oid = split.group(1) if split else "".join(ch if ch.isascii() and ch.isalnum() else "_" for ch in label.lower()).strip("_")[:64]
    if not 0 < len(label) <= 40:
        raise CliError("invalid_value", f"-c {choice!r} needs a label of 1 to 40 characters.",
                       'Example: kabu ask "Which env?" -c staging -c "prod:Production (EU)"')
    return {"id": oid or f"opt_{index}", "label": label}


def ask_body(a: JSON) -> JSON:
    """The ask: confirm by default, choice with -c, or the --type given."""
    choices: List[str] = a["choice"]
    kind = a["type"] or ("text" if a["ask_text"] else "choice" if choices else "confirm")
    if kind == "text" or a["ask_text"]:
        if choices or kind != "text":
            raise CliError("conflicting_flags", "A text ask takes no -c and no other --type.", 'Example: kabu ask "Release name?" --type text')
        return {"type": kind}
    if kind == "confirm":
        if choices and (len(choices) != 2 or any(not 0 < len(c) <= 40 for c in choices)):
            raise CliError("invalid_value", "A confirm ask takes no -c, or two labels of up to 40 characters: approve, then reject.",
                           'Example: kabu ask "Ship it?" --type confirm -c Ship -c Wait')
        options = [{"id": d, "label": c, "decision": d} for d, c in zip(("approve", "reject"), choices)]
        return {"type": kind, **({"options": options} if options else {})}
    if len(choices) < 2:
        raise CliError("too_few", f"A {kind} ask needs 2 to 12 -c options, not {len(choices)}.",
                       'Example: kabu ask "Which env?" -c staging -c prod')
    options = [option(i, c) for i, c in enumerate(choices, 1)]
    if len({o["id"] for o in options}) < len(options):
        raise CliError("invalid_value", "Two -c options have the same id.", "Give each its own id: -c stg:Staging -c prd:Production")
    return {"type": kind, "options": options, **({"min": 1, "max": len(options)} if kind == "multi" else {})}


def build_message(cmd: str, a: JSON) -> Tuple[JSON, List[Path], JSON]:
    """The validated request body and files for send or ask, and the project and its source. No HTTP call."""
    if a["project"] in PRIORITY_NUMBERS:  # kabu.py's -p was its priority
        raise CliError("invalid_value", f"--project {a['project']!r} looks like kabu.py's -p priority; in kabu, -p is --project.",
                       f"Example: kabu {SAMPLES[cmd]} --priority {PRIORITY_NUMBERS[a['project']]}")
    asking = cmd == "ask"
    title, message = texts(cmd, a)
    body: JSON = {"title": title, "body": message, **check(cmd, MESSAGE, a, asking)}
    for each in body.get("links", []):
        each["label"] = each["label"] or a["url_title"] or "Open"
    for kind, extra, key in (("code", "language", "language"), ("diff", "diff_path", "path"), ("quote", "", "")):
        if given(a[kind]):
            body.setdefault("blocks", []).append({"type": kind, "text": a[kind], **({key: a[extra]} if extra and a[extra] else {})})
    if "voice" in body:
        body["voice"]["speak"] = True  # --voice and --speed imply --as-audio
        if not asking and len(a["file"]) > 9:
            raise CliError("too_many", "A spoken notify takes 9 files: speech fills the tenth attachment.", "Attach up to 9 files.")
    if asking:
        body["ask"] = {**ask_body(a), **body.get("ask", {})}
        body.setdefault("expires_in", 300)
    source = with_project(cmd, body)  # last, for a valid command only: it is the slowest step
    size = len(utf8_json(body)) + 64 * (len(a["file"]) + 1)  # and room for the media and session ids added later
    if size > REQUEST_MAX:
        raise CliError("too_large", f"The request comes to {size} bytes; the API takes up to {REQUEST_MAX} (64 KiB of UTF-8 JSON).",
                       f'Shorten the text, or attach it. Example: kabu {cmd} "Nightly report" "Summary" --file report.md')
    return body, a["file"], {"project": body.get("project"), "project_source": source}


def with_project(cmd: str, body: JSON) -> JSON:
    """Put the resolved project in a request body that has none from --project, checked as --project is; its source."""
    if "project" in body:
        return {"rule": "flag"}
    project, source = resolve_project()
    body.update({"project": value(cmd, PROJECT, project)} if project else {})
    return source


def the_id(cmd: str, raw: Optional[str], prefix: str, env: str = "") -> str:
    """A msg_ or ses_ id from the argument, else the environment variable: checked before any HTTP call, quoted for a URL."""
    got = raw or (os.environ.get(env) if env else None) or ""
    if not re.fullmatch(prefix + r"\S+", got):
        raise CliError("invalid_id", f"kabu {cmd} needs a {prefix}… id" + (f" (or {env})" if env else "")
                       + (f", not {got!r}." if got else "."), f"Example: kabu {cmd} {prefix}01J9Z")
    return urllib.parse.quote(got, safe="")


def classify_ask_outcome(msg: JSON, kind: str) -> Tuple[JSON, int]:
    """The printed answer of an ask, and its exit code: 0 only for an explicit approve, or the answer to a choice, multi
    or text ask; 10 for a reject, or a typed reply to a confirm (kind); 3 for any other decision, or none."""
    status, resolution = str(msg.get("status") or "unknown"), msg.get("resolution")
    r: JSON = resolution if isinstance(resolution, dict) else {}
    decision = "reject" if r.get("decision") == "deny" else r.get("decision")  # deny: the legacy name; never inferred
    seen = {"id": msg.get("id"), "outcome": status, "status": status, "decision": decision}
    if status in ("expired", "cancelled"):
        return seen, EXIT_EXPIRED if status == "expired" else EXIT_CANCELLED
    option, options = r.get("option"), r.get("options")
    choices = [str(x) for x in options] if isinstance(options, list) else [option] if isinstance(option, str) and option else None
    answer = {**seen, "option": option, "choices": choices, "text": r.get("text"), "values": r.get("values"), "via": r.get("via")}
    if status == "answered" and (decision == "approve" or decision == "respond" and kind != "confirm"):
        return answer, EXIT_OK
    return answer, EXIT_REJECTED if status == "answered" and decision in ("reject", "respond") else EXIT_TRANSPORT


def cmd_message(cmd: str, a: JSON) -> int:
    """send or ask: upload the files, create the message and, for an ask, wait until it settles."""
    body, files, context = build_message(cmd, a)
    api = Api.connect(a)
    gone = body.pop("session") if "session" in body and body["session"] in gone_sessions(api.base)[1] else None  # one stands in
    if a["dry_run"]:  # the API validates it and renders each surface; nothing is uploaded or sent
        return emit({"ok": True, "dry_run": True, "request": body, "files": [str(f) for f in files],
                     "preview": api.call("POST", "/v2/messages/preview", body), "project_source": context["project_source"]})
    key = str(uuid.uuid4())  # one Idempotency-Key per invocation: a retry never sends twice
    joined, folder, lane = presence_join(api) if "session" not in body and not gone else (None, None, "")
    if joined:
        body["session"] = joined  # the agent session's own, which its hooks report live
    if "session" not in body and (gone or os.environ.get("KABU_AUTO_SESSION") == "1"):
        auto_session(api, body, key, session_key(harness()) if gone else "")  # a gone session's stand-in: this session's
    if files:
        body["attachments"] = [{"media": api.upload(f, f"{key}-file-{i}")} for i, f in enumerate(files, 1)]
    try:
        msg = api.call("POST", "/v2/messages", body, idem=key)  # an ask waits below, by its id: a 404 there is ask_gone
    except CliError as e:
        if not joined or e.code != "session_ended":
            raise
        if folder is not None:
            lane_ended(folder, lane)  # the presence session was ended: the message goes without it, and the next one gets a new one
        body.pop("session")
        msg = api.call("POST", "/v2/messages", body, idem=key + "-alone")
    if body.get("session") and msg.get("session", body["session"]) is None:  # the API dropped it: deleted with its project
        gone = body["session"]
        path, known = gone_sessions(api.base)
        write_cache(path, "\n".join([*known, gone][-100:]) + "\n")  # the last 100
    context.update(session=msg.get("session"), **({"session_gone": gone} if gone else {}))
    if cmd == "send" or a["no_wait"]:
        done: JSON = {"attachments": len(files)} if cmd == "send" else {"outcome": "pending"}
        return emit({"ok": True, "id": msg.get("id"), "status": msg.get("status"), **done, **context})
    mid, deadline = str(msg.get("id")), time.monotonic() + body["expires_in"] + GRACE  # the API settles an ask at expiry
    hint = f"The ask is still open, so never ask it again; wait for its answer with: kabu status {mid} --wait 50"
    while msg.get("status") == "pending":
        if time.monotonic() > deadline:
            raise CliError("ask_pending", f"No answer to {mid} by its expiry; kabu stopped waiting.", hint, EXIT_TRANSPORT, {"id": mid})
        try:
            msg = api.call("GET", "/v2/messages/" + urllib.parse.quote(mid, safe=""), wait=WAIT)
        except CliError as e:  # a network blip, 5xx or 429 only delays the wait; a refusal ends it
            if e.extra.get("status") == 404:  # deleted, as with its project: no answer will ever come, so never wait on
                raise CliError("ask_gone", f"{mid} was deleted (as when its project is), so it will never be answered.",
                               "Ask again if you still need it.", EXIT_CANCELLED, {**e.extra, "id": mid}) from None
            if 400 <= e.extra.get("status", 0) < 500 and e.extra["status"] != 429:
                raise CliError(e.code, e.message, hint, e.exit, {**e.extra, "id": mid}) from None
            time.sleep(5)
    summary, code = classify_ask_outcome(msg, body["ask"]["type"])
    return emit({**summary, **context}, code, sort=True)


def auto_session(api: Api, body: JSON, key: str, scope: str = "") -> None:
    """KABU_AUTO_SESSION=1: one automatic session per host, project and agent groups their messages (ADR 022): it claims
    no run, so it stays idle and has no Live Activity of its own; each ask in it has its own. `kabu session start` is
    the explicit, live run. One try of up to 3 s; an ended or expired session (the API returns a run's session even
    then), or any failure, sends the message without one. With no project in the message, the session names none
    either: the API picks it as for the message (a bound key's). A `scope`, this agent session's key, makes it this
    session's own, as a gone session's stand-in is."""
    host, project, agent = socket.gethostname().split(".")[0], body.get("project"), body["agent"]
    tail = f":{scope}" if scope else ""
    try:
        session = Api(api.base, api.key, 3.0, 0).call("POST", "/v2/sessions", {
            "title": f"{agent} on {host}{f' · {project}' if project else ''}"[:80], **({"project": project} if project else {}),
            "agent": agent, "external_id": f"auto:{host}:{project or ''}:{agent}"[:120 - len(tail)] + tail,
            "automatic": True}, idem=f"{key}-session")
    except CliError:
        return  # a session only groups messages: it never stops one
    if isinstance(session.get("id"), str) and session.get("status") not in ("ended", "expired") and not session.get("ended_at"):
        body["session"] = session["id"]


def cmd_session(cmd: str, a: JSON) -> int:
    """session start (get or create, with a Live Activity), update or end."""
    body, action = check(cmd, COMMANDS[cmd].opts, a), cmd.split()[1]
    update = {k: body.pop(k) for k in ("headline", "progress") if k in body}  # what session update takes
    if action != "start":
        sid = the_id(cmd, a["id"], "ses_", "KABU_SESSION")
        if action == "update" and not update:
            raise CliError("missing_flag", "Nothing to update.", "Example: kabu session update --headline 'Running tests' --progress 3/8")
        api, another = Api.connect(a), 'start another: kabu session start --title "…"'
        if sid in gone_sessions(api.base)[1]:  # the API reported it gone, deleted with its project: no call can reach it
            raise CliError("session_gone", f"{sid} no longer exists: it was deleted (as when its project is).",
                           another.capitalize() + ".", EXIT_TRANSPORT, {"id": sid})
        try:
            return emit(api.call("POST", f"/v2/sessions/{sid}/end", body, idem=str(uuid.uuid4())) if action == "end"
                        else api.call("PATCH", f"/v2/sessions/{sid}", update))
        except CliError as e:
            raise if_gone(e, another) from None
    if not a["title"]:
        raise CliError("missing_flag", "kabu session start needs --title.", 'Example: kabu session start --title "Fix flaky auth test"')
    with_project(cmd, body)
    api = Api.connect(a)
    session = api.call("POST", "/v2/sessions", {**body, "live": {"enabled": True}}, idem=str(uuid.uuid4()))
    if session.get("status") == "ended":  # get-or-create found this --external-id's session, already over
        raise CliError("session_ended", f"The session of --external-id {a['external_id']} has ended.",
                       "Start another with a new --external-id.", EXIT_TRANSPORT)
    if update:
        session = api.call("PATCH", "/v2/sessions/" + urllib.parse.quote(str(session.get("id")), safe=""), update)
    return emit(session)


def cmd_status(cmd: str, a: JSON) -> int:
    check(cmd, STATUS, a)
    try:
        return emit(Api.connect(a).call("GET", "/v2/messages/" + the_id(cmd, a["id"], "msg_"), wait=a["wait"] or 0))
    except CliError as e:
        raise if_gone(e, "it will never be answered, so do not wait on it; ask again if you still need it") from None


def if_gone(e: CliError, then: str) -> CliError:
    """The API's 404 for a message or session, with a hint that says what it means; any other error as it is."""
    return CliError(e.code, e.message, f"It was deleted (as when its project is) or never existed: {then}.", e.exit,
                    e.extra) if e.extra.get("status") == 404 else e


def cmd_whoami(cmd: str, a: JSON) -> int:
    check(cmd, CONNECTION, a)
    api = Api.connect(a)
    capabilities = api.call("GET", "/v2")
    try:
        org: Optional[JSON] = api.call("GET", "/v2/org")
    except CliError:
        org = None  # GET /v2/org needs org:write, which agent keys lack; GET /v2/me takes only a signed-in human
    project, source = resolve_project()
    return emit({"ok": True, "capabilities": capabilities, "org": org, "org_name": (org or {}).get("name") or load_config().get("org_name"),
                 "project": project, "project_source": source, "agent": resolve_agent(), "base_url": api.base,
                 "config_path": str(default_config_path())})


def device_login(api: Api, name: str) -> JSON:
    """RFC 8628: print a code for a person to approve in Kabu, then poll for this computer's own key."""
    auth = api.call("POST", "/v2/device_authorizations", {"name": name})
    link = auth.get("verification_uri_complete") or auth.get("verification_uri")
    print(f"In the Kabu app, open Connect a computer and enter the code: {auth.get('user_code')}"
          + (f"\nOr approve it in a browser: {link}" if link else "") + "\nOnly an admin or owner of the workspace can approve it.",
          file=sys.stderr, flush=True)
    form = {"device_code": str(auth.get("device_code", "")), "grant_type": "urn:ietf:params:oauth:grant-type:device_code"}
    interval, deadline = float(auth.get("interval", 5)), time.monotonic() + float(auth.get("expires_in", 600))
    while time.monotonic() < deadline:
        time.sleep(interval)
        try:
            return api.call("POST", "/v2/device_authorizations/poll", form=form)
        except CliError as e:  # the API answers a poll with an OAuth error code only
            if e.code == "access_denied":
                raise CliError(e.code, "The code was denied in Kabu.", "Run kabu login --device again.", EXIT_REJECTED) from None
            if e.code == "expired_token":
                break
            if e.code not in ("authorization_pending", "slow_down"):
                raise
            interval += 5 if e.code == "slow_down" else 0
    raise CliError("expired_token", "The code expired before anyone approved it.", "Run kabu login --device again.", EXIT_EXPIRED)


def cmd_login(cmd: str, a: JSON) -> int:
    """Save a key (--api-key, or typed at a terminal) or get this computer's own (--device), and the defaults given."""
    if a["api_key"] == "-":
        a["api_key"] = sys.stdin.buffer.readline().decode("utf-8", "replace")
    a["api_key"] = a["api_key"] and a["api_key"].strip()  # a key file with Windows line endings leaves a \r
    check(cmd, LOGIN, a)
    if a["api_key"] and a["device"]:
        raise CliError("conflicting_flags", "Give --api-key or --device, not both.", "Example: kabu login --device")
    cfg = load_config()
    cfg.update(base_url=base_url(a["base_url"], cfg), **{k: a[k] for k in ("project", "agent") if a[k]})
    known: JSON = {}
    if a["device"]:
        token = device_login(Api(cfg["base_url"]), a["name"])
        known = {k: token[k] for k in ("org_name", "project") if token.get(k)}
        cfg.update(api_key=str(token.get("access_token", "")), **({"org_name": known["org_name"]} if "org_name" in known else {}))
    else:
        typed = "" if a["api_key"] or not sys.stdin.isatty() else getpass.getpass("Kabu API key: ").strip()
        if not a["api_key"] and not typed:
            raise CliError("missing_flag", "kabu login needs --device or --api-key.", "Example: kabu login --device")
        cfg["api_key"] = a["api_key"] or value(cmd, KEY, typed)
    return emit({"ok": True, **known, "config_path": str(save_config(cfg))})


def tool_project() -> Optional[str]:
    """The project kabu mcp adds to a tool call, resolved for that call as kabu send would: one rules 1 to 5 name (the
    environment, the session, the folder, the login), never a <folder>@<host> name, nor none."""
    project, source = resolve_project()
    return project if source["rule"] not in ("folder", "default") else None


def cmd_mcp(cmd: str, a: JSON) -> int:
    """Bridge a harness's stdio MCP to the API's /mcp with the saved key (CONTRACT.md, MCP bridge). Logs go to stderr."""
    import threading

    check(cmd, CONNECTION, a)
    api = Api.connect(a)
    url, out = api.base + "/mcp", threading.Lock()
    session: Dict[str, str] = {}  # the Mcp-Session-Id the server issued, sent on every later request
    takes: Set[str] = set()  # the tools whose input has a project, from tools/list
    try:  # a key bound to a project sends as it, and another project would be refused (403): then add none
        bound = bool(Api(api.base, api.key, 3.0, 0).call("GET", "/v2").get("project"))
    except CliError:
        bound = True  # not known: add none, as before 2.4.0

    def write(reply: Any) -> None:
        with out:
            try:
                print(json.dumps(reply), flush=True)
            except OSError:
                pass  # the harness has gone, and stdin ends next

    def relay(line: bytes, show: Callable[[Any], None]) -> None:
        """POST one message, and show each message of the reply as it comes: the JSON body, or each event of a stream."""
        headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream",
                   "Authorization": f"Bearer {api.key}", "User-Agent": f"kabu-cli/{VERSION}", **session}
        with OPEN(urllib.request.Request(url, line, headers, method="POST"), timeout=api.timeout + WAIT) as resp:
            if resp.headers.get("Mcp-Session-Id"):
                session["Mcp-Session-Id"] = resp.headers["Mcp-Session-Id"]
            if "text/event-stream" not in resp.headers.get("Content-Type", ""):
                body = resp.read()
                if body.strip():  # a notification's 202 has none
                    show(json.loads(body))
                return
            data: List[str] = []
            for raw in resp:  # an event stream: the data lines of each event are one message
                text = raw.decode().rstrip("\r\n")
                if text.startswith("data:"):
                    data.append(text[5:].removeprefix(" "))
                elif not text and data:
                    show(json.loads("\n".join(data)))
                    data = []

    def forward(line: bytes, msg: Any) -> None:
        """Relay one message. On failure, log it, and answer a request that has no answer yet with a JSON-RPC error:
        the harness never hangs, and never gets two answers for one id."""
        rid = msg.get("id") if isinstance(msg, dict) and "method" in msg else None  # a request's id: it wants an answer
        answered = False

        def show(reply: Any) -> None:
            nonlocal answered
            if isinstance(msg, dict) and msg.get("method") == "tools/list":
                try:
                    takes.update(t["name"] for t in reply["result"]["tools"] if "project" in t["inputSchema"]["properties"])
                except (KeyError, TypeError):
                    pass  # a list without schemas: no tool gets a project
            write(reply)
            answered = answered or isinstance(reply, dict) and "method" not in reply and reply.get("id") == rid

        try:
            relay(line, show)
            if answered or rid is None:
                return
            raise ValueError("no answer")  # an empty body, or a stream that ended without the response
        except urllib.error.HTTPError as exc:
            e = Api.problem(exc, "POST", url, RETRIES)[0]
        except ValueError:
            e = CliError("bad_response", f"POST {url} did not answer with JSON-RPC.", "Check that KABU_BASE_URL points at the Kabu API.")
        except (OSError, http.client.HTTPException) as exc:  # refused, reset, timed out
            e = CliError("network_error", f"POST {url} failed: {getattr(exc, 'reason', exc)}.", "Check the network and KABU_BASE_URL.")
        printed = error_line(e)
        print(printed, file=sys.stderr, flush=True)
        if rid is not None and not answered:
            shown = json.loads(printed)["error"]  # as redacted: message and hint, then the code and the rest as data
            write({"jsonrpc": "2.0", "id": rid, "error": {"code": -32000, "message": f"{shown.pop('message')} {shown.pop('hint')}",
                                                          "data": shown}})

    pending: List[threading.Thread] = []
    presence = McpPresence(api)
    before = signal.signal(signal.SIGTERM, lambda *_: sys.exit(EXIT_OK))  # a harness that stops it so: offline too, below
    try:
        for line in filter(None, (raw.strip() for raw in sys.stdin.buffer)):
            try:
                msg = json.loads(line)
            except ValueError:
                msg = None  # sent as it is: the server answers with a parse error
            call = msg.get("params") if isinstance(msg, dict) and msg.get("method") == "tools/call" else None
            call = call if isinstance(call, dict) else {}
            tool = call.get("name")  # a str, unless the harness sent something odd
            ses = presence.start() if isinstance(msg, dict) and msg.get("method") == "tools/call" else None  # after a call
            args = call.setdefault("arguments", {}) if isinstance(tool, str) and tool in takes else None
            project = tool_project() if isinstance(args, dict) and not args.get("project") and not bound else None
            if project and isinstance(args, dict):
                args["project"] = project
            if ses and isinstance(args, dict) and not args.get("session"):
                args["session"] = ses  # the harness's automatic session, which its hooks and kabu send share
            if project or ses and isinstance(args, dict) and args.get("session") == ses:
                line = json.dumps(msg).encode()
            if isinstance(msg, dict) and "method" in msg and "id" in msg:  # a request runs alongside the others
                pending = [t for t in pending if t.is_alive()] + [threading.Thread(target=forward, args=(line, msg), daemon=True)]
                pending[-1].start()
            else:  # notifications and replies keep their order
                forward(line, msg)
        for thread in pending:
            thread.join()
    finally:
        presence.stop()  # its lane offline; the session is never ended: hooks share it, and a resumed run gets it back
        signal.signal(signal.SIGTERM, before)
    return EXIT_OK


class McpPresence:
    """kabu mcp's presence (ADR 021, 022; CONTRACT.md, MCP bridge). Nothing before the first tool call; then the
    harness's automatic session, got or created once on the hooks' HMAC ladder (one ses_ per run), and an online
    record from source mcp, which stop() turns offline. It never ends the session."""

    def __init__(self, api: Api) -> None:
        self.api, self.started = api, False
        self.session: Optional[str] = None
        self.folder: Optional[Path] = None
        self.pid, self.sid = 0, ""

    def start(self) -> Optional[str]:
        if self.started:
            return self.session
        self.started = True
        parent = ancestors()[:1]  # only when a harness started it, as its MCP server: never a kabu mcp a script runs
        agent, folder = (parent[0] if parent and parent[0][2] in AGENT_PROCESSES else None), runtime_dir()
        if agent is None or not presence_on() or not own_dir(folder, create=True):
            return None
        pid, lane, started = agent[0], f"l{agent[0]}", proc_start(agent[1])
        got = fold(folder, pid).get(lane)
        if got is not None and got.rec["runtime"] != "offline" and started is not None and got.mtime >= started - 1:
            self.sid = str(got.rec.get("sid") or "")  # the hooks' session id, so the same session
        elif os.environ.get("CLAUDE_PID") == str(pid):
            self.sid = os.environ.get("CLAUDE_CODE_SESSION_ID", "") if re.fullmatch(r"[A-Za-z0-9._:-]{1,64}", os.environ.get(
                "CLAUDE_CODE_SESSION_ID", "")) else ""
        try:
            self.session = lane_session(self.api, installation(), folder, lane, {"pid": pid, "sid": self.sid}, agent[1], agent[2], 3.0)
        except (OSError, ValueError):
            self.session = None
        if put_record(folder, pid, {"event": "mcp", "runtime": "online", "attention": None, "source": "mcp", "sid": self.sid, "pane": ""}):
            self.folder, self.pid = folder, pid
            spawn_reporter(folder, self.api)
        return self.session

    def stop(self) -> None:
        if self.folder is not None:
            put_record(self.folder, self.pid, {"event": "end", "runtime": "offline", "attention": None, "source": "mcp", "sid": self.sid,
                                               "pane": ""})
            self.folder = None


def spawn_reporter(folder: Path, api: Api) -> None:
    """Start kabu presence run when none runs, as the hook does: the one process that outlives this one."""
    try:
        pid = int((read_private(folder / "run.pid") or b"0").strip() or 0)
    except ValueError:
        pid = 0
    if pid > 1 and alive(pid):
        return
    try:
        subprocess.Popen([sys.executable, os.path.abspath(__file__), "presence", "run"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                         stderr=subprocess.DEVNULL, start_new_session=True, env={**os.environ, "KABU_API_KEY": api.key, "KABU_BASE_URL": api.base})
    except OSError:
        pass


# ---------------------------------------------------------------------------------------------------- presence --
# Is the agent live? (ADR 021) A hook writes a record at each event of an agent session on this machine; one
# `kabu presence run` folds them into PUT /v2/presence: a batch a minute, and one at each change. What leaves the
# machine: state enums, HMAC'd ids, the harness's name and the project the session runs in, as kabu send sends.
PRESENCE_TICK = 2.0  # seconds between looks at the records and at their harnesses' processes
PRESENCE_SETTLE = 5.0  # working <-> idle waits this long: a turn that ends as the next one starts sends nothing
PRESENCE_BEAT = 60  # seconds between batches while nothing changes, unless the API's next_beat_in says otherwise
PRESENCE_PER_MINUTE = 10  # batches in any 60 s: the API takes 12 a minute from one reporter
PRESENCE_DECAY = 900.0  # a turn with no event for 15 minutes, and no tool running, reads idle
PRESENCE_EXIT = 600.0  # kabu presence run exits after 10 minutes with no agent session to report
PRESENCE_RECHECK = 10.0  # seconds between checks that a harness is no zombie, and still the process that started
PRESENCE_REPORTS = 20  # the most reports one batch takes
PRESENCE_FORGET = 300.0  # an exited session that could not be reported offline is dropped after this; its lease lapses
PRESENCE_TORN = 10.0  # an empty or unreadable record this old is a hook stopped mid-write: removed
HOOK_EVENTS = ("SessionStart", "UserPromptSubmit", "PreToolUse", "PostToolUse", "Notification", "Stop", "SessionEnd")
COPILOT_EVENTS = {"SessionStart": "sessionStart", "UserPromptSubmit": "userPromptSubmitted", "PreToolUse": "preToolUse",
                  "PostToolUse": "postToolUse", "Notification": "notification", "Stop": "agentStop", "SessionEnd": "sessionEnd"}
# Each harness kabu presence install knows: its agent slug, its settings (the variable naming its folder, the folder
# under the home, the file) and the hook events it gets, each in the harness's own name, from its current docs (pi has
# an in-process extension instead: PI_EXTENSION). Cursor gets no preToolUse: a permission hook there blocks the tool
# on an empty answer. The hook maps each name to a state (PRESENCE_HOOK's first case).
HARNESS_SLUGS = {"claude": "claude-code", "copilot": "copilot-cli", "codex": "codex", "gemini": "gemini-cli", "cursor": "cursor",
                 "pi": "pi"}
HARNESS_FILES = {"claude": ("CLAUDE_CONFIG_DIR", ".claude", "settings.json"), "copilot": ("COPILOT_HOME", ".copilot", "settings.json"),
                 "codex": ("CODEX_HOME", ".codex", "hooks.json"), "gemini": ("", ".gemini", "settings.json"),
                 "cursor": ("", ".cursor", "hooks.json"), "pi": ("PI_CODING_AGENT_DIR", ".pi/agent", "extensions/kabu-presence.js")}
HARNESS_EVENTS = {
    "codex": ("SessionStart", "UserPromptSubmit", "PreToolUse", "PostToolUse", "PermissionRequest", "Stop", "Interrupt", "SessionEnd"),
    "gemini": ("SessionStart", "BeforeAgent", "BeforeTool", "AfterTool", "Notification", "AfterAgent", "SessionEnd"),
    "cursor": ("sessionStart", "beforeSubmitPrompt", "postToolUse", "postToolUseFailure", "stop", "sessionEnd"),
    "pi": ("session_start", "agent_start", "tool_call", "ui_prompt_start", "ui_prompt_end", "agent_settled", "session_shutdown"),
}
NEEDS_INPUT = "permission_prompt|elicitation_dialog|elicitation_url_dialog|worker_permission_prompt|agent_needs_input"
HOOK_FILE = "presence-hook.sh"  # kabu presence install writes the hook here, in KABU_HOME, and knows its entries by it
HOOK_CLI = 'kabu_cli="${CLAUDE_PLUGIN_ROOT:-.}/bin/kabu"'  # the plugin's; install puts this CLI's path in its place
RECORD = re.compile(r"l(\d+)-(\d+)\.json")  # a hook's record: the harness's process id, and the event's seq
# The hook: a harness's shell sources it in a subshell ((. FILE) || true), so it runs no program of its own, which
# costs 5 to 10 ms on a Mac. Each event is a new file, created no-clobber (O_EXCL: never through a link, never over
# another file) at the next free seq, so concurrent hooks never lose one another's events and the order is the seq
# order. It never reads its input, which a harness may hold open: the session id is Claude Code's CLAUDE_CODE_SESSION_ID
# (else the harness's process id and start time stand for it), and the Notification matcher picks the types that need
# the user. scripts/plugin.py writes it to plugins/kabu/hooks/presence.sh.
PRESENCE_HOOK = r"""# Kabu presence hook for Claude Code, Copilot CLI, Codex, Gemini CLI and Cursor (ADR 021). Generated by kabu:
# do not edit. Hook command: kabu_event=EVENT [kabu_harness=SLUG]; (. THIS_FILE) || true   It starts no program but
# the reporter, never reads its input (so it never waits for it), has no unbounded loop, and always exits 0.
# Each event is a new 0600 record, created no-clobber in this user's machine-local folder (0700, never a link):
# ${KABU_RUNTIME_DIR:-${XDG_RUNTIME_DIR:-${TMPDIR:-/tmp}}}/kabu-presence-$USER/l<harness pid>-<seq>.json: the state,
# the harness's process and session ids and the Herdr pane. Never a prompt, path or tool name, never the network.
# It starts `kabu presence run`, which reports, when none runs. Off: KABU_PRESENCE=0 or `kabu presence off`.
kabu_cli="${CLAUDE_PLUGIN_ROOT:-.}/bin/kabu"
[ -z "${ZSH_VERSION:-}" ] || emulate sh
umask 077
exec 2>/dev/null
[ "${KABU_PRESENCE:-}" = 0 ] && exit 0
[ -e "${KABU_HOME:-$HOME/.kabu}/presence/off" ] && exit 0
kabu_att=null
case "${kabu_event:-}" in
  SessionStart|sessionStart) kabu_rt=idle kabu_ev=start ;;
  UserPromptSubmit|userPromptSubmitted|BeforeAgent|beforeSubmitPrompt) kabu_rt=working kabu_ev=prompt ;;
  PreToolUse|preToolUse|BeforeTool) kabu_rt=working kabu_ev=tool ;;
  PostToolUse|postToolUse|AfterTool|postToolUseFailure) kabu_rt=working kabu_ev=post ;;
  Notification|notification|PermissionRequest) kabu_rt=working kabu_ev=attention kabu_att='"terminal"' ;;
  Stop|agentStop|AfterAgent|Interrupt|stop) kabu_rt=idle kabu_ev=stop ;;
  SessionEnd|sessionEnd) kabu_rt=offline kabu_ev=end ;;
  *) exit 0 ;;
esac
case "${kabu_harness:-}" in codex|gemini-cli|cursor) ;; *) kabu_harness= ;; esac
kabu_dir=${KABU_RUNTIME_DIR:-${XDG_RUNTIME_DIR:-${TMPDIR:-/tmp}}}
kabu_dir="${kabu_dir%/}/kabu-presence-${USER:-${LOGNAME:-}}"
[ -L "$kabu_dir" ] && exit 0
[ -d "$kabu_dir" ] || mkdir -m 700 "$kabu_dir" || exit 0
[ -O "$kabu_dir" ] && [ ! -L "$kabu_dir" ] || exit 0
kabu_sid=
[ "${CLAUDE_PID:-}" = "$PPID" ] && kabu_sid=${CLAUDE_CODE_SESSION_ID:-}
[ "$kabu_harness" = gemini-cli ] && kabu_sid=${GEMINI_SESSION_ID:-}
kabu_pane=${HERDR_PANE_ID:-}
case $kabu_pane in *[!A-Za-z0-9._:-]*) kabu_pane= ;; esac
[ ${#kabu_pane} -le 40 ] || kabu_pane=
kabu_put() {
  case $kabu_sid in *[!A-Za-z0-9._:-]*) kabu_sid= ;; esac
  [ ${#kabu_sid} -le 64 ] || kabu_sid=
  kabu_max=0 kabu_n=0
  for kabu_f in "$kabu_dir"/l"$PPID"-*.json; do
    kabu_s=${kabu_f##*-}; kabu_s=${kabu_s%.json}
    case $kabu_s in ''|*[!0-9]*) continue ;; esac
    kabu_n=$((kabu_n + 1))
    [ "$kabu_s" -gt "$kabu_max" ] && kabu_max=$kabu_s
  done
  [ "$kabu_n" -lt 200 ] || return 0
  set -C
  kabu_seq=$kabu_max
  while [ "$kabu_seq" -lt $((kabu_max + 32)) ]; do
    kabu_seq=$((kabu_seq + 1))
    kabu_f="$kabu_dir/l$PPID-$kabu_seq.json"
    { [ -e "$kabu_f" ] || [ -L "$kabu_f" ]; } && continue
    printf '%s\n' "{\"v\":2,\"event\":\"$kabu_ev\",\"runtime\":\"$kabu_rt\",\"attention\":$kabu_att,\"seq\":$kabu_seq,\"pid\":$PPID,\"sid\":\"$kabu_sid\",\"pane\":\"$kabu_pane\",\"harness\":\"$kabu_harness\"}" > "$kabu_f" && return 0
  done
}
kabu_put
case $kabu_ev in tool|post) exit 0 ;; esac
kabu_run=0
[ -f "$kabu_dir/run.pid" ] && [ ! -L "$kabu_dir/run.pid" ] && { IFS= read -r kabu_run < "$kabu_dir/run.pid" || :; }
case $kabu_run in ''|*[!0-9]*) kabu_run=0 ;; esac
if [ "$kabu_ev" = start ] || [ "$kabu_run" -le 1 ] || ! kill -0 "$kabu_run"; then
  (python3 "$kabu_cli" presence run </dev/null >/dev/null 2>&1 &)
fi
exit 0
"""
# pi's presence (kabu presence install --harness pi): an extension pi loads in its own process, so it needs no hook.
# Each event is a record written as the hook writes one (a new 0600 file, created no-clobber, never through a link, in
# the same folder), with pi's own process id and, as its session id, the session file Herdr names it by (else pi's
# session id). Every handler catches all and returns nothing: a tool_call handler that threw would block the tool.
PI_EXTENSION = r"""// Kabu presence for pi (ADR 021). Generated by kabu presence install --harness pi: do not edit.
// On each event it writes one 0600 record, no-clobber, in this user's machine-local folder (0700, never a link):
// the state, pi's process and session ids and the Herdr pane. Never a prompt, path or tool name, never the network.
// It never throws into pi and never blocks a tool. It starts `kabu presence run`, which reports, when none runs.
// Off: KABU_PRESENCE=0 or `kabu presence off`.
import fs from "node:fs";
import os from "node:os";
import { spawn } from "node:child_process";

const KABU_CLI = "";
const C = fs.constants;
let running = false;

function folder() {
  const env = process.env;
  if (env.KABU_PRESENCE === "0") return null;
  try {
    fs.lstatSync((env.KABU_HOME || os.homedir() + "/.kabu") + "/presence/off");
    return null;
  } catch (e) {}
  const base = (env.KABU_RUNTIME_DIR || env.XDG_RUNTIME_DIR || env.TMPDIR || "/tmp").replace(/\/$/, "");
  const dir = base + "/kabu-presence-" + (env.USER || env.LOGNAME || "");
  try { fs.mkdirSync(dir, 0o700); } catch (e) {}
  const st = fs.lstatSync(dir);
  if (!st.isDirectory() || st.isSymbolicLink() || st.uid !== process.getuid()) return null;
  if (st.mode & 0o077) fs.chmodSync(dir, 0o700);
  return dir;
}

function sessionId(ctx) {
  let sid = "";
  try {
    const file = ctx && ctx.sessionManager && ctx.sessionManager.getSessionFile();
    sid = typeof file === "string" && file.startsWith("/") ? file : String(ctx.sessionManager.getSessionId() || "");
  } catch (e) {}
  return sid.length <= 4096 && !/[\u0000-\u001f\u007f]/.test(sid) ? sid : "";
}

function put(dir, ev, runtime, attention, ctx) {
  const pid = process.pid;
  let max = 0, n = 0;
  for (const name of fs.readdirSync(dir)) {
    const m = /^l(\d+)-(\d+)\.json$/.exec(name);
    if (m && Number(m[1]) === pid) { n += 1; max = Math.max(max, Number(m[2])); }
  }
  if (n >= 200) return;
  const pane = /^[A-Za-z0-9._:-]{1,40}$/.test(process.env.HERDR_PANE_ID || "") ? process.env.HERDR_PANE_ID : "";
  for (let seq = max + 1; seq <= max + 32; seq += 1) {
    let fd;
    try {
      fd = fs.openSync(dir + "/l" + pid + "-" + seq + ".json", C.O_WRONLY | C.O_CREAT | C.O_EXCL | C.O_NOFOLLOW, 0o600);
    } catch (e) {
      if (e && e.code === "EEXIST") continue;
      return;
    }
    try {
      fs.writeSync(fd, JSON.stringify({ v: 2, event: ev, runtime, attention, seq, pid, sid: sessionId(ctx), pane, harness: "pi" }) + "\n");
    } finally {
      fs.closeSync(fd);
    }
    return;
  }
}

function reporter(dir) {
  let pid = 0;
  try {
    const fd = fs.openSync(dir + "/run.pid", C.O_RDONLY | C.O_NOFOLLOW | C.O_NONBLOCK);
    try {
      if (fs.fstatSync(fd).isFile()) pid = parseInt(fs.readFileSync(fd, "utf8").trim(), 10) || 0;
    } finally {
      fs.closeSync(fd);
    }
  } catch (e) {}
  if (pid > 1) {
    try { process.kill(pid, 0); return; } catch (e) { if (e && e.code === "EPERM") return; }
  }
  const child = spawn("python3", [KABU_CLI, "presence", "run"], { detached: true, stdio: "ignore" });
  child.on("error", () => {});
  child.unref();
}

function record(ev, runtime, attention, ctx) {
  try {
    const dir = folder();
    if (dir === null) return;
    put(dir, ev, runtime, attention, ctx);
    if (ev !== "tool") reporter(dir);
  } catch (e) {}
}

export default function (pi) {
  pi.on("session_start", (event, ctx) => { running = false; record("start", "idle", null, ctx); });
  pi.on("agent_start", (event, ctx) => { running = true; record("prompt", "working", null, ctx); });
  pi.on("tool_call", (event, ctx) => { record("tool", "working", null, ctx); });
  pi.on("ui_prompt_start", (event, ctx) => { record("attention", "working", "terminal", ctx); });
  pi.on("ui_prompt_end", (event, ctx) => { record(running ? "post" : "stop", running ? "working" : "idle", null, ctx); });
  pi.on("agent_settled", (event, ctx) => { running = false; record("stop", "idle", null, ctx); });
  pi.on("session_shutdown", (event, ctx) => {
    if (!event || event.reason !== "reload") record("end", "offline", null, ctx);
  });
}
"""


def kabu_home() -> Path:
    """kabu's state that lasts (KABU_HOME, else ~/.kabu): presence's off switch, reporter identities and the hook."""
    return Path(os.environ.get("KABU_HOME") or Path.home() / ".kabu")


def presence_dir() -> Path:
    return kabu_home() / "presence"


def runtime_dirs() -> List[Path]:
    """This user's machine-local presence folder, as the hook names it: KABU_RUNTIME_DIR, XDG_RUNTIME_DIR, TMPDIR or
    /tmp, then kabu-presence-$USER; on a Mac also its per-user temp folder, which a harness may not pass as TMPDIR."""
    name = "kabu-presence-" + (os.environ.get("USER") or os.environ.get("LOGNAME") or "")
    base = os.environ.get("KABU_RUNTIME_DIR") or os.environ.get("XDG_RUNTIME_DIR") or os.environ.get("TMPDIR") or "/tmp"
    found = [Path(base.rstrip("/") or "/") / name]
    if sys.platform == "darwin" and not os.environ.get("KABU_RUNTIME_DIR"):
        try:
            mac = Path(os.confstr("CS_DARWIN_USER_TEMP_DIR") or "") / name
        except (ValueError, OSError):
            mac = found[0]
        found += [mac] if mac.is_absolute() and mac != found[0] else []
    return found


def runtime_dir() -> Path:
    return runtime_dirs()[0]


def own_dir(path: Path, create: bool = False) -> bool:
    """Whether path is a folder of this user's, never a link, made or set 0700: presence uses no other."""
    try:
        if create and not os.path.lexists(path):
            os.mkdir(path, 0o700)
    except FileExistsError:
        pass
    except OSError:
        return False
    try:
        st = os.lstat(path)
        if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid():
            return False
        if st.st_mode & 0o077:
            os.chmod(path, 0o700)
    except OSError:
        return False
    return True


def presence_on() -> bool:
    """Presence is on unless KABU_PRESENCE=0 or `kabu presence off`."""
    return os.environ.get("KABU_PRESENCE") != "0" and not os.path.lexists(presence_dir() / "off")


def open_dir(path: Path) -> int:
    """A folder of this user's, opened never through a link: work done in it through this descriptor cannot be sent
    elsewhere by a link swapped in later."""
    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    st = os.fstat(fd)
    if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid():
        os.close(fd)
        raise OSError(f"{path} is not a folder of this user's")
    return fd


def write_private(path: Path, value: Any, keep: bool = False) -> None:
    """Write JSON (or bytes) as a new 0600 file, never through a link, then put it in place at once: a rename replaces
    whatever the name was, a link or another file, without writing to it. keep: only if no file is there yet. All of it
    happens in the folder as opened once (open_dir), so a link swapped in for the folder meanwhile changes nothing."""
    folder = open_dir(path.parent)
    tmp = f".{path.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}"
    try:
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=folder)
        with os.fdopen(fd, "wb") as out:
            out.write(value if isinstance(value, bytes) else (json.dumps(value) + "\n").encode())
        try:
            if keep:
                os.link(tmp, path.name, src_dir_fd=folder, dst_dir_fd=folder, follow_symlinks=False)
            else:
                os.rename(tmp, path.name, src_dir_fd=folder, dst_dir_fd=folder)
        except FileExistsError:
            pass
    finally:
        try:
            os.unlink(tmp, dir_fd=folder)
        except OSError:
            pass
        os.close(folder)


def read_private(path: Path, limit: int = 1 << 16) -> Optional[bytes]:
    """A file of this user's, never through a link (the file's or its folder's), and only a regular one (never a FIFO
    that would block), or None."""
    try:
        folder = open_dir(path.parent)
    except OSError:
        return None
    try:
        fd = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=folder)
    except OSError:
        return None
    finally:
        os.close(folder)
    try:
        st = os.fstat(fd)
        if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or st.st_size > limit:
            return None
        return os.read(fd, limit)
    except OSError:
        return None
    finally:
        os.close(fd)


def read_json(path: Path) -> JSON:
    try:
        value = json.loads((read_private(path) or b"").decode("utf-8", "replace").split("\n", 1)[0])
    except ValueError:
        return {}
    return value if isinstance(value, dict) else {}


def machine_id() -> str:
    """This machine, which a home shared with others (NFS) does not share: KABU_MACHINE_ID, Linux's machine-id, the
    Mac's hardware UUID, else the host name. Only its hash names a file."""
    if os.environ.get("KABU_MACHINE_ID"):
        return os.environ["KABU_MACHINE_ID"]
    for name in ("/etc/machine-id", "/var/lib/dbus/machine-id"):
        try:
            text = Path(name).read_text().strip()
        except OSError:
            continue
        if text:
            return text
    if sys.platform == "darwin":
        try:
            import ctypes
            uid, wait = ctypes.create_string_buffer(16), (ctypes.c_long * 2)(0, 0)
            if ctypes.CDLL("/usr/lib/libSystem.B.dylib").gethostuuid(uid, wait) == 0:
                return uid.raw.hex()
        except (OSError, AttributeError, ValueError, ImportError):
            pass
    return socket.gethostname()


def installation(bump: bool = False) -> JSON:
    """This machine's reporter (ADR 021): its instance id and secret, made once and kept (mode 600), and its restart
    counter, which bump raises by one, in presence/reporter-<machine>.json (~/.kabu/installation.json is legacy
    kabu.py's): machines that share a home have their own. The secret never leaves the machine: it keys every HMAC."""
    if not (own_dir(kabu_home(), create=True) or os.path.isdir(kabu_home())) or not own_dir(presence_dir(), create=True):
        raise OSError(f"{presence_dir()} is not a folder of this user's (or is a link): presence uses none other")
    path = presence_dir() / f"reporter-{hashlib.sha256(machine_id().encode()).hexdigest()[:16]}.json"
    inst = read_json(path)
    if not (re.fullmatch(r"pin_[0-9A-Za-z]{8,40}", str(inst.get("instance"))) and re.fullmatch(r"[0-9a-f]{64}", str(inst.get("secret")))):
        fresh = {"instance": "pin_" + uuid.uuid4().hex[:24], "secret": os.urandom(32).hex(), "generation": 0}
        write_private(path, fresh, keep=not os.path.lexists(path))
        inst = read_json(path) or fresh
    inst["generation"] = int(inst.get("generation") or 0)
    if bump:
        inst["generation"] += 1
        write_private(path, inst)
    inst["path"] = str(path)
    return inst


def keyed(inst: JSON, *parts: str) -> str:
    """An id that names local things without showing them: HMAC(installation secret, parts), in hex."""
    return hmac.new(bytes.fromhex(str(inst["secret"])), "\0".join(parts).encode(), hashlib.sha256).hexdigest()


def proc_start(token: str) -> Optional[float]:
    """When a process started, as epoch seconds, from its start token (processes()): a Linux boot id and clock ticks,
    or ps's lstart in UTC. None when unknown."""
    try:
        if ":" in token and " " not in token:
            btime = next(float(line.split()[1]) for line in (PROC / "stat").read_text().splitlines() if line.startswith("btime "))
            return btime + int(token.rsplit(":", 1)[1]) / os.sysconf("SC_CLK_TCK")
        return float(calendar.timegm(time.strptime(token, "%a %b %d %H:%M:%S %Y"))) if token else None
    except (OSError, ValueError, StopIteration):
        return None


@dataclass
class Folded:
    """An agent session's records, folded: the latest event's state, the files that made it, and the latest file of
    each source, which stay."""

    rec: JSON
    mtime: float
    files: List[Path]
    latest: List[Path] = field(default_factory=list)


def fold(folder: Path, pid: Optional[int] = None, now: Optional[float] = None) -> Dict[str, Folded]:
    """Each harness's records in folder, by lane (l<pid>), folded in seq order: the state is the latest event's, the
    session id the latest one given. A record still empty or unreadable is skipped; one PRESENCE_TORN old (a hook
    stopped mid-write) is removed when now is given."""
    lanes: Dict[str, List[Tuple[int, Path, Optional[JSON], float]]] = {}
    try:
        names = os.listdir(folder)
    except OSError:
        return {}
    for name in names:
        m = RECORD.fullmatch(name)
        if not m or pid is not None and int(m.group(1)) != pid:
            continue
        path = folder / name
        try:
            mtime = os.lstat(path).st_mtime
        except OSError:
            continue
        rec = record(path, int(m.group(1)))
        if rec is None and now is not None and now - mtime > PRESENCE_TORN:
            unlink(path)
            continue
        lanes.setdefault(f"l{m.group(1)}", []).append((int(m.group(2)), path, rec, mtime))
    out: Dict[str, Folded] = {}
    for lane, items in lanes.items():
        valid = sorted((i for i in items if i[2] is not None), key=lambda i: i[0])
        hooks = [i for i in valid if (i[2] or {}).get("source", "hook") == "hook"]
        mcp = [i for i in valid if (i[2] or {}).get("source") == "mcp"]
        top = hooks[-1] if hooks else mcp[-1] if mcp else None  # the hooks' working and idle beat kabu mcp's online
        if top is None:
            continue
        rec = dict(top[2] or {})
        rec["sid"] = next((str(i[2]["sid"]) for i in reversed(valid) if i[2] and i[2].get("sid")), "")
        rec["harness"] = next((str(i[2]["harness"]) for i in reversed(valid) if i[2] and i[2].get("harness")), "")
        out[lane] = Folded(rec, top[3], [i[1] for i in items], [last[-1][1] for last in (hooks, mcp) if last])
    return out


def record(path: Path, pid: int) -> Optional[JSON]:
    """One record, a hook's or kabu mcp's (source mcp: online, then offline at its EOF), or None while it is being
    written or when it is not one. Its harness, when it names one, is one kabu knows; its session id is local only."""
    rec = read_json(path)
    sid = rec.get("sid", "")
    ok = rec.get("pid") == pid and isinstance(rec.get("seq"), int) and rec.get("attention") in (None, "terminal") \
        and isinstance(sid, str) and len(sid) <= 4096 and not re.search(r"[\x00-\x1f\x7f]", sid) \
        and rec.get("harness", "") in ("", *HARNESS_SLUGS.values()) \
        and (rec.get("runtime") in ("working", "idle", "offline") and rec.get("source", "hook") == "hook"
             or rec.get("runtime") in ("online", "offline") and rec.get("source") == "mcp")
    return rec if ok else None


def put_record(folder: Path, pid: int, rec: JSON) -> bool:
    """Write a record as the hook does (kabu mcp's): a new 0600 file at the next free seq of the harness's lane, created
    no-clobber in the folder as opened once, never through a link."""
    try:
        dfd = open_dir(folder)
    except OSError:
        return False
    try:
        top = max((seq_of(Path(n)) for n in os.listdir(dfd) if n.startswith(f"l{pid}-") and RECORD.fullmatch(n)), default=0)
        for seq in range(top + 1, top + 33):
            try:
                fd = os.open(f"l{pid}-{seq}.json", os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=dfd)
            except FileExistsError:
                continue
            with os.fdopen(fd, "w", encoding="utf-8") as out:
                out.write(json.dumps({"v": 2, **rec, "seq": seq, "pid": pid}) + "\n")
            return True
        return False
    except OSError:
        return False
    finally:
        os.close(dfd)


def seq_of(path: Path) -> int:
    m = RECORD.fullmatch(path.name)
    return int(m.group(2)) if m else 0


def unlink(path: Path) -> None:
    try:
        os.unlink(path)
    except OSError:
        pass


def harness_slug(name: str) -> str:
    """The agent a harness's process name stands for: claude is claude-code; one kabu does not know, as a slug."""
    return AGENT_PROCESSES.get(name) or re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")[:64] or "agent"


def api_tag(api: Api) -> str:
    """Which API and key a kept session belongs to: one API's ids mean nothing to another's."""
    return hashlib.sha256(f"{api.base}\0{api.key}".encode()).hexdigest()[:16]


def lane_session(api: Api, inst: JSON, folder: Path, lane: str, rec: JSON, start: str, name: str,
                 timeout: float = 10.0) -> Optional[str]:
    """The Kabu session of an agent session: the one kept beside its records for this API, else get or create the
    session whose external_id is HMAC(installation secret, harness ‖ native session id), automatic (ADR 022: it
    claims no run, so it stays idle with no Live Activity of its own). Its
    title is constant ('<harness> session'), and its project the one the harness's folder names, as kabu send's. The
    reporter and a kabu send in that agent session so find the same one. None when the API cannot say."""
    native = str(rec.get("sid") or "") or f"pid:{rec['pid']}:{start}"
    slug, path = str(rec.get("harness") or "") or harness_slug(name), folder / f"{lane}.session.json"
    base = keyed(inst, "session", slug, native)[:40]
    kept = read_json(path)
    turn = int(kept.get("round") or 0) if kept.get("id") == base else 0  # a session that ended: the next one
    if kept.get("api") == api_tag(api) and kept.get("id") == base and str(kept.get("session", "")).startswith("ses_") \
            and kept["session"] not in gone_sessions(api.base)[1]:
        return str(kept["session"])
    try:
        root = process_cwd(int(rec["pid"]))
        project = named_project(root.resolve())[0] if root else ""  # never a temp folder's, nor <folder>@<host>
    except OSError:
        project = ""
    body: JSON = {"title": f"{slug} session", "agent": slug, "automatic": True, **({"project": project} if project else {})}
    for n in range(turn, turn + 3):
        try:
            got = Api(api.base, api.key, timeout, 0).call("POST", "/v2/sessions", {**body, "external_id": f"kp_{base}" + (f"-{n}" if n else "")},
                                                          idem=str(uuid.uuid4()))
        except CliError:
            return None
        if isinstance(got.get("id"), str) and got.get("status") not in ("ended", "expired") and not got.get("ended_at"):
            write_private(path, {"api": api_tag(api), "id": base, "round": n, "session": got["id"]})
            return str(got["id"])
    return None


def lane_ended(folder: Path, lane: str) -> None:
    """The API refused a lane's session (ended, or gone): the next one is a new session."""
    path = folder / f"{lane}.session.json"
    kept = read_json(path)
    if kept:
        write_private(path, {"api": kept.get("api"), "id": kept.get("id"), "round": int(kept.get("round") or 0) + 1})


def presence_join(api: Api) -> Tuple[Optional[str], Optional[Path], str]:
    """The session a kabu send or ask in an agent session that reports presence goes in (CONTRACT.md, Presence), with
    its folder and lane; None when presence is off, no hook wrote a record for this harness since it started, or the
    API cannot say."""
    agent = harness()
    if agent is None or not presence_on():
        return None, None, ""
    lane = f"l{agent[0]}"
    for folder in runtime_dirs():
        got = fold(folder, agent[0]).get(lane) if own_dir(folder) else None
        if got is None:
            continue
        started = proc_start(agent[1])
        if got.rec["runtime"] == "offline" or started is None or got.mtime < started - 1:
            return None, None, ""  # ended, or records left by an earlier process that had this id
        try:
            return lane_session(api, installation(), folder, lane, got.rec, agent[1], agent[2], timeout=3.0), folder, lane
        except (OSError, ValueError):
            return None, None, ""
    return None, None, ""


@dataclass
class Lane:
    """One agent session as the reporter sees it: its records, its harness's process, and what the API last took."""

    rec: JSON
    mtime: float
    start: str
    name: str
    checked: float
    session: Optional[str] = None
    sent: Optional[Tuple[str, Optional[str]]] = None  # the runtime and attention the API last took
    want: Optional[Tuple[str, Optional[str]]] = None  # what it should read, since `since`
    since: float = 0.0
    gone: float = 0.0  # when its harness exited or its session ended: offline once, then it is forgotten
    source: str = "hook"  # where what it reads comes from: its records' source, or herdr when Herdr's answer is used


# The optional Herdr bridge (kabu presence herdr on; ADR 021): Herdr's own view of each agent, from `herdr agent list`
# (never its socket), joined to a lane only by the exact session id Herdr names (agent_session.value), reading only
# agent_status, through this one table; a status it lacks says nothing. Herdr's answer beats the hooks' working and idle;
# a session that ended, an exited harness and the API's own facts (an ask, the end) beat Herdr.
HERDR_STATES: Dict[str, Tuple[str, Optional[str]]] = {"working": ("working", None), "blocked": ("working", "terminal"),
                                                      "idle": ("idle", None), "done": ("idle", None)}
HERDR_POLL = 5.0  # seconds between looks at herdr agent list, while the bridge is on, herdr is on PATH and a lane has a session id
HERDR_STALE = 15.0  # an answer this old (herdr failing since, or printing what kabu cannot read) is no longer used
HERDR_WAIT = 3.0  # herdr agent list gets this long, and 4 MiB, to answer; then it is killed: no answer, which changes nothing


def herdr_on() -> bool:
    """Whether the Herdr bridge is on (kabu presence herdr on): off by default."""
    return os.path.lexists(presence_dir() / "herdr")


def herdr_agents(out: bytes) -> Optional[Dict[str, str]]:
    """What herdr agent list printed, as {agent_session.value: agent_status}, or None when it is not that list. Nothing
    else is read from it; a session id two agents name says nothing."""
    try:
        doc = json.loads(out.decode("utf-8"))
    except (ValueError, UnicodeDecodeError, RecursionError):  # nested past Python's limit: not herdr's list either
        return None
    result = doc.get("result") if isinstance(doc, dict) else None
    agents = result.get("agents") if isinstance(result, dict) else None
    if not isinstance(agents, list):
        return None
    found: Dict[str, str] = {}
    shared: Set[str] = set()
    for agent in agents:
        ses = agent.get("agent_session") if isinstance(agent, dict) else None
        value, status = (ses.get("value") if isinstance(ses, dict) else None), (agent.get("agent_status") if isinstance(agent, dict) else None)
        if isinstance(value, str) and value and isinstance(status, str):
            shared |= {value} if value in found else set()
            found[value] = status
    return {value: status for value, status in found.items() if value not in shared}


def herdr_list(exe: str) -> Optional[bytes]:
    """herdr agent list's output, when it exits 0 within HERDR_WAIT and 4 MiB; else None, and it is killed with whatever
    it started. It is always reaped: the bridge leaves no process behind."""
    import select
    try:
        proc = subprocess.Popen([exe, "agent", "list"], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                                start_new_session=True)
    except OSError:
        return None
    assert proc.stdout is not None
    out, ended, deadline = b"", False, time.monotonic() + HERDR_WAIT
    try:
        while len(out) <= 4 << 20:
            left = deadline - time.monotonic()
            if left <= 0 or not select.select([proc.stdout], [], [], left)[0]:
                break
            chunk = os.read(proc.stdout.fileno(), 1 << 16)
            ended = not chunk
            if ended:
                break
            out += chunk
        if ended:
            proc.wait(timeout=max(0.1, deadline - time.monotonic()))
    except (OSError, ValueError, subprocess.TimeoutExpired):
        ended = False
    finally:
        if proc.poll() is None or not ended:
            try:
                os.killpg(proc.pid, signal.SIGKILL)
            except OSError:
                pass
        proc.stdout.close()
        proc.wait()
    return out if ended and proc.returncode == 0 else None


class Reporter:
    """kabu presence run's state, one step() every PRESENCE_TICK: fold the records, check the harnesses, and send a
    batch when something changed or a beat is due. `clock` is the wall clock: records carry their file's mtime."""

    def __init__(self, connect: Callable[[], Optional[Api]], inst: JSON, folder: Path, clock: Callable[[], float] = time.time) -> None:
        self.connect, self.inst, self.folder, self.clock = connect, inst, folder, clock
        self.lanes: Dict[str, Lane] = {}
        self.seq = 0
        self.sent_at: List[float] = []  # when each batch of the last minute went
        self.next_beat = self.retry_at = 0.0
        self.failures = 0
        self.quiet_since = clock()
        self.status: JSON = {}
        self.herdr: Dict[str, str] = {}  # Herdr's last answer, {session id: agent_status}, and when it came
        self.herdr_at = self.herdr_next = 0.0
        self.herdr_status: JSON = {}

    def report(self) -> JSON:
        """What status.json shows: the last batch, and the bridge's last look (never what Herdr printed)."""
        return {**self.status, **({"herdr": self.herdr_status} if self.herdr_status else {})}

    def scan(self, now: float) -> None:
        """Fold every lane's records and keep only its latest. A new lane is adopted only while its harness runs, is no
        zombie, and started before its latest record was written: a record older than the process with its id is
        another process's, left behind."""
        seen = fold(self.folder, now=now)
        new = [key for key in seen if key not in self.lanes]
        info = processes([int(seen[key].rec["pid"]) for key in new]) if new else {}
        for key in new:
            first, proc = seen[key], info.get(int(seen[key].rec["pid"]))
            started = proc_start(proc.start) if proc else None
            if proc is None or proc.zombie or first.rec["runtime"] == "offline" or started is None or first.mtime < started - 1:
                self.forget(key, first.files)  # exited, ended or stale before anyone reported it: nothing to say
                continue
            self.lanes[key] = Lane(first.rec, first.mtime, proc.start, proc.name, checked=now)
        retired: Dict[str, Lane] = {}
        for key, lane in self.lanes.items():
            got = seen.get(key)
            if got is None:
                lane.gone = lane.gone or now  # its records were removed: offline
                continue
            for path in got.files:  # the latest of each source stays, so the next record's seq goes on from it
                if path not in got.latest and record(path, int(got.rec["pid"])):
                    unlink(path)
            if got.rec["seq"] != lane.rec["seq"] or got.mtime != lane.mtime:
                if lane.session and lane.rec.get("sid") and got.rec.get("sid") and got.rec["sid"] != lane.rec["sid"]:
                    # Another session in the same process (pi's /new or /resume): the one before goes offline once, and
                    # this lane gets or creates the new one's.
                    retired[f"{key}~{lane.rec['seq']}"] = Lane(lane.rec, lane.mtime, lane.start, lane.name, now, lane.session,
                                                               lane.sent, gone=now)
                    lane.session = lane.sent = lane.want = None
                lane.rec, lane.mtime = got.rec, got.mtime
                if lane.gone and got.rec["runtime"] != "offline" and alive(int(got.rec["pid"])):
                    lane.gone = 0.0  # the session went on in the same process, as after Claude Code's /clear
        self.lanes.update(retired)

    def check(self, now: float) -> None:
        """A harness whose process exited, is a zombie, or whose id another process took, or whose session ended,
        reads offline: kill -0 at every look, ps every PRESENCE_RECHECK."""
        due = {id(lane) for lane in self.lanes.values() if not lane.gone and now - lane.checked >= PRESENCE_RECHECK}
        info = processes([int(lane.rec["pid"]) for lane in self.lanes.values() if id(lane) in due]) if due else {}
        for lane in self.lanes.values():
            if lane.gone:
                continue
            dead = not alive(int(lane.rec["pid"])) or lane.rec["runtime"] == "offline"
            if id(lane) in due:
                lane.checked = now
                proc = info.get(int(lane.rec["pid"]))
                dead = dead or proc is None or proc.zombie or proc.start != lane.start
            lane.gone = now if dead else 0.0

    def value(self, lane: Lane, now: float) -> Optional[Tuple[str, Optional[str]]]:
        """What a lane reports in a batch now, or None: what it reads, but the last value taken while working and idle
        settle, and nothing for an exited session that was never reported or is reported offline already."""
        if lane.gone:  # ended, or its harness exited: offline, whatever Herdr says
            lane.source = str(lane.rec.get("source") or "hook")
            return None if lane.sent in (None, ("offline", None)) else ("offline", None)
        runtime, attention = str(lane.rec["runtime"]), lane.rec.get("attention")
        herdr = self.herdr_value(lane, now)
        lane.source = "herdr" if herdr else str(lane.rec.get("source") or "hook")
        if herdr:
            runtime, attention = herdr  # Herdr's view beats the hooks' working and idle
        elif runtime == "working" and attention is None and lane.rec.get("event") != "tool" and now - lane.mtime > PRESENCE_DECAY:
            runtime = "idle"  # a turn the hooks lost the end of (an interrupt): no event for 15 minutes, no tool running
        want = (runtime, attention)
        if lane.want != want:
            lane.want, lane.since = want, now
        settling = lane.sent is not None and {lane.sent[0], runtime} == {"working", "idle"} and lane.sent[1] is None and attention is None
        return lane.sent if settling and now - lane.since < PRESENCE_SETTLE else want

    def look_herdr(self, now: float) -> None:
        """Every HERDR_POLL, while the bridge is on, herdr is on PATH and a lane has a session id to join on: run herdr
        agent list. An answer kabu cannot read (or none) changes nothing: the last one stands until HERDR_STALE."""
        if now < self.herdr_next:
            return
        self.herdr_next = now + HERDR_POLL
        exe = shutil.which("herdr") if herdr_on() else None
        if exe is None or not any(lane.rec.get("sid") for lane in self.lanes.values() if not lane.gone):
            self.herdr, self.herdr_at, self.herdr_status = {}, 0.0, {"on": herdr_on(), "found": exe is not None} if herdr_on() else {}
            return
        out = herdr_list(exe)
        got = herdr_agents(out) if out is not None else None
        if got is not None:
            self.herdr, self.herdr_at = got, now
        self.herdr_status = {"on": True, "found": True, "at": iso(now), "read": got is not None,
                             "joined": sum(bool(self.herdr_value(lane, now)) for lane in self.lanes.values())}

    def herdr_value(self, lane: Lane, now: float) -> Optional[Tuple[str, Optional[str]]]:
        """Herdr's state for a lane, through HERDR_STATES: only for the exact session id, only from an answer Herdr gave
        after the lane's latest record (a prompt, a permission prompt or a stop since then beats what Herdr saw before
        it), and only while that answer is recent: a look that failed since never makes it newer."""
        sid = str(lane.rec.get("sid") or "")
        fresh = bool(sid) and self.herdr_at > lane.mtime and now - self.herdr_at <= HERDR_STALE
        return HERDR_STATES.get(self.herdr.get(sid, "")) if fresh else None

    def step(self) -> bool:
        """One look and, when due, one batch. False once the reporter should exit: presence was turned off, or there
        was no agent session to report for PRESENCE_EXIT."""
        now = self.clock()
        on = presence_on() and own_dir(self.folder) and os.path.isdir(kabu_home())  # gone, as a test's folders go: exit
        self.scan(now)
        self.check(now)
        for lane in self.lanes.values():
            lane.gone = lane.gone or (0.0 if on else now)
        self.look_herdr(now)
        values = {key: self.value(lane, now) for key, lane in self.lanes.items()}
        changed = any(v is not None and v != self.lanes[key].sent for key, v in values.items())
        beat = any(v is not None for v in values.values()) and now >= self.next_beat
        self.sent_at = [t for t in self.sent_at if now - t < 60]
        if (changed or beat) and now >= self.retry_at and len(self.sent_at) < PRESENCE_PER_MINUTE:
            self.send(now, values)
        for key, lane in list(self.lanes.items()):
            if lane.gone and (lane.sent in (None, ("offline", None)) or now - lane.gone > PRESENCE_FORGET):
                pid, got = int(lane.rec["pid"]), fold(self.folder, int(lane.rec["pid"])).get(key)
                self.forget(key, [p for p in (got.files if got else []) if not alive(pid) or seq_of(p) <= lane.rec["seq"]])
        if self.lanes:
            self.quiet_since = now
        return on and now - self.quiet_since < PRESENCE_EXIT

    def forget(self, key: str, files: Iterable[Path]) -> None:
        """Drop a lane, the records it was folded from and its session; a record written since stays."""
        self.lanes.pop(key, None)
        for path in [*files, self.folder / f"{key}.session.json"]:
            unlink(path)

    def send(self, now: float, values: Dict[str, Optional[Tuple[str, Optional[str]]]]) -> None:
        """One batch: every live session of this machine (the API takes a session left out offline), at most 20."""
        api = self.connect()
        if api is None:
            self.status = {**self.status, "error": {"code": "missing_api_key", "at": iso(now)}}
            self.retry_at = now + PRESENCE_BEAT
            return
        reports: List[Tuple[str, Lane, Tuple[str, Optional[str]]]] = []
        for key, lane in sorted(self.lanes.items(), key=lambda kv: (not kv[1].gone, -kv[1].mtime)):
            value = values.get(key)
            if value is None:
                continue
            if lane.session is None and not lane.gone:
                lane.session = lane_session(api, self.inst, self.folder, key, lane.rec, lane.start, lane.name)
            if lane.session is not None and len(reports) < PRESENCE_REPORTS:
                reports.append((key, lane, value))
        if not reports:
            return
        self.seq += 1
        body = {"instance": self.inst["instance"], "generation": self.inst["generation"], "seq": self.seq,
                "reports": [{"session": lane.session, "lane": "ln_" + keyed(self.inst, "lane", key, lane.start)[:32],
                             "runtime": value[0], "attention": value[1], "source": lane.source}
                            for key, lane, value in reports]}
        self.sent_at.append(now)
        beat: JSON = {"at": iso(now), "generation": body["generation"], "seq": self.seq, "reports": len(reports)}
        try:
            ack = Api(api.base, api.key, 10.0, 0).call("PUT", "/v2/presence", body)
        except CliError as e:
            self.failures += 1
            self.retry_at = now + min(PRESENCE_BEAT, 5.0 * 2 ** self.failures)
            self.status = {**self.status, "last_beat": {**beat, "error": {"code": e.code, "status": e.extra.get("status")}}}
            return
        self.failures, self.retry_at = 0, 0.0
        self.next_beat = now + max(1, int(ack.get("next_beat_in") or PRESENCE_BEAT))
        results = [r if isinstance(r, dict) else {} for r in ack.get("results") or []]
        for (key, lane, value), result in zip(reports, results):
            if result.get("reason") not in ("session_ended", "not_found", "stale"):
                lane.sent = value  # taken, or fenced by another reporter's lease: sent again at the next beat
            elif result.get("reason") != "stale":
                lane_ended(self.folder, key)
                lane.session, lane.sent = None, None
                if lane.gone:
                    lane.sent = ("offline", None)  # nothing left to take offline
        if any(r.get("reason") == "stale" for r in results):  # this counter was lost or reused: resume above the API's
            self.inst["generation"] = max(int(ack.get("generation") or 0), int(self.inst["generation"])) + 1
            self.seq = 0
            write_private(Path(self.inst["path"]), {k: v for k, v in self.inst.items() if k != "path"})
            self.retry_at = 0.0
            self.next_beat = now
        self.status = {"last_beat": beat, "ack": {k: ack.get(k) for k in ("server_time", "next_beat_in", "generation", "seq",
                                                                            "expires_at", "results")}}


def alive(pid: int) -> bool:
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except OSError:
        pass  # another user's process
    return True


def iso(at: float) -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(at))


def presence_run(a: JSON) -> int:
    """The machine's one reporter: exits at once when another holds the lock, presence is off, its folder is not this
    user's own (or is a link) or this is no POSIX system; else reports until presence is turned off or no agent session
    was left for 10 minutes."""
    folder = runtime_dir()
    if not presence_on() or not own_dir(folder, create=True):
        return EXIT_OK
    try:
        import fcntl
    except ImportError:
        return EXIT_OK
    try:
        fd = os.open(folder / "run.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
    except OSError:
        return EXIT_OK  # a link, or not ours to open
    with os.fdopen(fd, "r+b") as lock:
        st = os.fstat(fd)
        if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid():
            return EXIT_OK
        try:
            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            return EXIT_OK  # another reporter runs on this machine
        try:
            os.setsid()  # outlives the hook that started it
        except OSError:
            pass
        write_private(folder / "run.pid", f"{os.getpid()}\n".encode())

        def connect() -> Optional[Api]:
            try:
                return Api.connect(a)  # the key and base URL again each time: kabu login may come later
            except CliError:
                return None

        try:
            inst = installation(bump=True)
        except OSError:
            return EXIT_OK
        reporter, shown = Reporter(connect, inst, folder), None
        try:
            while reporter.step():
                if reporter.report() != shown:  # written when it changes, not every tick
                    shown = reporter.report()
                    write_private(folder / "status.json", shown)
                time.sleep(PRESENCE_TICK)
            write_private(folder / "status.json", reporter.report())
        finally:
            if (read_private(folder / "run.pid") or b"").strip() == str(os.getpid()).encode():
                unlink(folder / "run.pid")
        del lock
    return EXIT_OK


def presence_status() -> JSON:
    """What kabu presence status prints: the switch, the reporter, each agent session's state and the last batch."""
    folder = runtime_dir()
    lanes = []
    for key, got in sorted(fold(folder).items()) if own_dir(folder) else ():
        lanes.append({"lane": key, "runtime": got.rec["runtime"], "attention": got.rec.get("attention"), "event": got.rec.get("event"),
                      "seq": got.rec["seq"], "pid": got.rec["pid"], "alive": alive(int(got.rec["pid"])), "updated_at": iso(got.mtime),
                      "session": read_json(folder / f"{key}.session.json").get("session")})
    run = read_json(folder / "status.json")
    try:
        pid = int((read_private(folder / "run.pid") or b"").strip() or 0)
    except ValueError:
        pid = 0
    try:
        inst = {k: v for k, v in installation().items() if k in ("instance", "generation")}
    except OSError:
        inst = {}
    return {"ok": True, "enabled": presence_on(), "off": os.path.lexists(presence_dir() / "off"),
            "opted_out": os.environ.get("KABU_PRESENCE") == "0", "running": pid > 1 and alive(pid), "pid": pid if pid > 1 and alive(pid) else None,
            "instance": inst.get("instance"), "generation": inst.get("generation"), "dir": str(folder), "lanes": lanes,
            "last_beat": run.get("last_beat"), "ack": run.get("ack"),
            "herdr": {"on": herdr_on(), "found": shutil.which("herdr") is not None, "last": run.get("herdr")}}


def settings_file(harness_name: str, given_path: Optional[str]) -> Path:
    """A harness's user settings (HARNESS_FILES): Claude Code's CLAUDE_CONFIG_DIR (~/.claude) settings.json, Copilot
    CLI's COPILOT_HOME (~/.copilot) settings.json, Codex's CODEX_HOME (~/.codex) hooks.json, Gemini CLI's
    ~/.gemini/settings.json, Cursor's ~/.cursor/hooks.json, pi's PI_CODING_AGENT_DIR (~/.pi/agent) extension file."""
    if given_path:
        return Path(os.path.expanduser(given_path))
    variable, folder, name = HARNESS_FILES[harness_name]
    return Path(os.environ.get(variable) or Path.home() / folder if variable else Path.home() / folder) / name


def hook_command(event: str, slug: str, hook: Path) -> str:
    """The hook command of a harness that runs it in a shell kabu does not choose (Codex: $SHELL -lc, which may be zsh
    or fish; Gemini CLI: bash; Cursor): /bin/sh, exec'd so that the harness stays its parent ($PPID, the lane). Cursor's
    beforeSubmitPrompt also answers {"continue": true}, whatever the hook did."""
    inner = f'kabu_event={event} kabu_harness={slug}; [ ! -r "{hook}" ] || (. "{hook}") || true'
    return f"exec /bin/sh -c '{inner}" + ('; echo {\\"continue\\":true}' if event == "beforeSubmitPrompt" else "") + "'"


def hook_entries(harness_name: str, hook: Path) -> Dict[str, JSON]:
    """The hook's entry for each event, under the harness's own name for it, in its settings' own shape: Claude Code's
    matcher groups; Copilot CLI's native keys with flat {type, bash, timeoutSec}; Codex's groups (timeouts in seconds,
    at most 3 for SessionEnd and Interrupt); Gemini CLI's named groups (timeouts in milliseconds); Cursor's flat
    {command, timeout}."""
    entries: Dict[str, JSON] = {}
    for event in HARNESS_EVENTS.get(harness_name, HOOK_EVENTS):
        name = COPILOT_EVENTS[event] if harness_name == "copilot" else event
        command = f"kabu_event={name}; [ ! -r {shlex.quote(str(hook))} ] || (. {shlex.quote(str(hook))}) || true"
        if harness_name in ("codex", "gemini", "cursor"):
            command = hook_command(event, HARNESS_SLUGS[harness_name], hook)
        if harness_name == "copilot":
            entries[name] = {"type": "command", "bash": command, "timeoutSec": 5}
        elif harness_name == "codex":
            entries[name] = {"hooks": [{"type": "command", "command": command, "timeout": 3 if event in ("SessionEnd", "Interrupt") else 5}]}
        elif harness_name == "gemini":
            entries[name] = {"hooks": [{"name": f"kabu-presence-{event}", "type": "command", "command": command, "timeout": 5000}]}
        elif harness_name == "cursor":
            entries[name] = {"command": command, "timeout": 5}
        else:
            entries[name] = {**({"matcher": NEEDS_INPUT} if event == "Notification" else {}),
                             "hooks": [{"type": "command", "command": command, "timeout": 5}]}
    return entries


PI_MARKER = PI_EXTENSION.split("\n", 1)[0].encode()


def pi_install(a: JSON, install: bool) -> JSON:
    """Write pi's presence extension (PI_EXTENSION, with this CLI's path) into its extensions folder (made 0700 when
    missing), or take it out, after a backup of what was there; never through a link. A file there that kabu did not
    write is left alone (exit 2)."""
    path = settings_file("pi", a["settings"])
    text = PI_EXTENSION.replace('const KABU_CLI = "";', f"const KABU_CLI = {json.dumps(os.path.abspath(__file__))};", 1).encode()
    target = SettingsFile.open(path, create=install)
    raw, saved, changed = None, None, False
    if target is not None:
        try:
            raw = target.read()[0]
            if raw is not None and not raw.startswith(PI_MARKER):
                raise CliError("invalid_settings", f"{path} is not kabu's pi extension, so kabu left it as it is.",
                               "Move it away, or name another file with --settings, then run this again.")
            changed = raw != text if install else raw is not None
            saved = target.backup(raw) if changed and raw is not None else None
            if changed and install:
                target.replace(text, 0o600)
            elif changed:
                target.remove()
        finally:
            target.close()
    return {"ok": True, "harness": "pi", "settings": str(path), "hook": str(path), "installed": install,
            "events": list(HARNESS_EVENTS["pi"]) if install else [], "changed": changed, "backup": saved}


def ours(entry: Any) -> bool:
    """An entry kabu presence install wrote: its command runs the installed hook."""
    return HOOK_FILE in json.dumps(entry)


def digest(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


def settings_dir(folder: Path, create: bool = False) -> Optional[int]:
    """A settings file's folder, opened one component at a time from /, following no link but the system's own (root
    owns /var and /tmp on a Mac; no user process can plant one of those), each missing one made 0700 when create, else
    None; the folder itself this user's. Every read and write of the file then goes through this descriptor."""
    parts = Path(os.path.abspath(folder)).parts
    fd = os.open("/", os.O_RDONLY | os.O_DIRECTORY)
    try:
        for i, part in enumerate(parts[1:], 2):
            shown = Path(*parts[:i])
            try:
                st = os.stat(part, dir_fd=fd, follow_symlinks=False)
            except FileNotFoundError:
                if not create:
                    os.close(fd)
                    return None
                os.mkdir(part, 0o700, dir_fd=fd)
                st = os.stat(part, dir_fd=fd, follow_symlinks=False)
            link = stat.S_ISLNK(st.st_mode)
            if link and st.st_uid != 0:
                raise CliError("unsafe_settings", f"{shown} is a link, and kabu never writes through one.",
                               f"Pass the real path with --settings (under {os.path.realpath(shown)}), then run this again.")
            try:
                nxt = os.open(part, os.O_RDONLY | os.O_DIRECTORY | (0 if link else os.O_NOFOLLOW), dir_fd=fd)
            except OSError as exc:
                raise CliError("unsafe_settings", f"{shown} is not a folder kabu can use ({exc.strerror}).",
                               "Pass another path with --settings, then run this again.") from None
            os.close(fd)
            fd = nxt
        if os.fstat(fd).st_uid != os.getuid():
            raise CliError("unsafe_settings", f"{folder} is not a folder of this user's.", "Pass another path with --settings, then run this again.")
        return fd
    except BaseException:
        try:
            os.close(fd)
        except OSError:
            pass
        raise


@dataclass
class SettingsFile:
    """A harness's settings file (or pi's extension), read, backed up, replaced and removed only through its folder's
    verified descriptor (settings_dir): never through a link, the file's own or a folder's."""

    path: Path
    dir: int

    @classmethod
    def open(cls, path: Path, create: bool = False) -> Optional[SettingsFile]:
        fd = settings_dir(path.parent, create)
        return None if fd is None else cls(path, fd)

    def close(self) -> None:
        os.close(self.dir)

    def read(self, name: str = "") -> Tuple[Optional[bytes], int]:
        """The file's bytes and mode (or a backup's, by name, beside it); None when there is none. A link, or anything
        but a regular file of this user's, is refused."""
        name = name or self.path.name
        try:
            fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=self.dir)
        except FileNotFoundError:
            return None, 0o600
        except OSError:
            raise CliError("unsafe_settings", f"{self.path.parent / name} is a link, and kabu never writes through one.",
                           "Pass the real path with --settings, then run this again.") from None
        try:
            st = os.fstat(fd)
            if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid():
                raise CliError("unsafe_settings", f"{self.path.parent / name} is not a file of this user's.", "Pass another path with --settings.")
            chunks: List[bytes] = []
            while True:
                chunk = os.read(fd, 1 << 20)
                if not chunk:
                    return b"".join(chunks), st.st_mode & 0o777
                chunks.append(chunk)
        finally:
            os.close(fd)

    def write(self, name: str, raw: bytes, mode: int = 0o600) -> None:
        fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=self.dir)
        with os.fdopen(fd, "wb") as out:
            out.write(raw)
            os.fchmod(out.fileno(), mode & 0o600)

    def backup(self, raw: bytes) -> str:
        """A new 0600 copy of the file's bytes, beside it, under a name no other backup has."""
        name = f"{self.path.name}.bak-kabu-presence-{time.strftime('%Y%m%d-%H%M%S')}-{time.time_ns() % 10**9:09d}-{uuid.uuid4().hex[:6]}"
        self.write(name, raw)
        return str(self.path.with_name(name))

    def replace(self, raw: bytes, mode: int) -> None:
        """Put new bytes in place at once (a new file renamed over the name: a link there is replaced, never written
        through), never looser than 0600."""
        tmp = f".{self.path.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}"
        self.write(tmp, raw, mode)
        os.rename(tmp, self.path.name, src_dir_fd=self.dir, dst_dir_fd=self.dir)

    def remove(self) -> None:
        os.unlink(self.path.name, dir_fd=self.dir)


def presence_install(a: JSON, install: bool) -> JSON:
    """Add the presence hook to a harness's settings, or take it out, after a backup. Install is idempotent; uninstall
    puts back the exact bytes from before the install when the file is as kabu left it, else takes out only kabu's
    entries. The hook file goes once no settings use it (presence/installs.json keeps what was installed where)."""
    if a["harness"] == "pi":
        return pi_install(a, install)
    path, hook, manifest_path = settings_file(a["harness"], a["settings"]), kabu_home() / HOOK_FILE, presence_dir() / "installs.json"
    if not (os.path.isdir(kabu_home()) or own_dir(kabu_home(), create=True)) or not own_dir(presence_dir(), create=True):
        raise CliError("unsafe_folder", f"{presence_dir()} is not a folder of this user's, or is a link.", "Remove it, then run this again.")
    if re.search(r"[\x00-\x1f'\"$`\\]", os.path.abspath(hook)):  # it goes inside the quotes of hook_command
        raise CliError("unsafe_folder", f"{hook} has a quote, $, ` or \\ in its path, which a hook command cannot hold.",
                       "Set KABU_HOME to a folder without them, then run this again.")
    target = SettingsFile.open(path)
    if target is None and install:
        raise CliError("missing_folder", f"{path.parent} does not exist, so kabu has no settings to add to.",
                       "Start the harness once, or name the settings file with --settings, then run this again.")
    if target is None:  # nothing to take out: only what kabu kept of it goes
        manifest = read_json(manifest_path)
        manifest.pop(str(path.resolve()), None)
        if not manifest:
            unlink(hook)
        write_private(manifest_path, manifest)
        return {"ok": True, "harness": a["harness"], "settings": str(path), "hook": str(hook), "installed": False, "events": [],
                "changed": False, "backup": None}
    try:
        return settings_install(a, install, path, hook, manifest_path, target)
    finally:
        target.close()


def settings_install(a: JSON, install: bool, path: Path, hook: Path, manifest_path: Path, target: SettingsFile) -> JSON:
    """presence_install's work on a settings file reached through its verified folder (SettingsFile)."""
    raw, mode = target.read()
    try:
        settings = json.loads(raw) if raw and raw.strip() else {}
    except ValueError as exc:
        raise CliError("invalid_settings", f"{path} is not JSON ({exc}), so kabu left it as it is.", "Fix the file, then run this again.") from None
    if not isinstance(settings, dict) or not isinstance(settings.get("hooks", {}), dict):
        raise CliError("invalid_settings", f"{path} has no hooks object kabu can add to.", "Fix the file, then run this again.")
    manifest = read_json(manifest_path)
    key, entry = str(path.resolve()), manifest.get(str(path.resolve())) or {}
    hooks: JSON = {}
    for event, entries in settings.get("hooks", {}).items():
        kept = [e for e in entries if not ours(e)] if isinstance(entries, list) else entries
        if kept != []:
            hooks[event] = kept
    saved: List[str] = []
    if install:
        write_private(hook, PRESENCE_HOOK.replace(HOOK_CLI, f"kabu_cli={shlex.quote(os.path.abspath(__file__))}", 1).encode())
        for event, added in hook_entries(a["harness"], hook).items():
            hooks[event] = [*(hooks.get(event) or []), added]
        updated = {**({"version": 1} if a["harness"] == "cursor" else {}), **settings, "hooks": hooks}  # Cursor needs its version
        new = (json.dumps(updated, indent=2, ensure_ascii=False) + "\n").encode()
        changed = updated != settings
        if changed:
            saved += [target.backup(raw)] if raw is not None else []
            target.replace(new, mode)
            manifest[key] = {"harness": a["harness"], "before": entry.get("before", digest(raw) if raw is not None else None),
                             "backup": entry.get("backup", saved[0] if saved else None), "after": digest(new)}
    else:
        changed = raw is not None and (key in manifest or any(ours(e) for v in settings.get("hooks", {}).values() for e in (v if isinstance(v, list) else [])))
        if changed and raw is not None:
            saved.append(target.backup(raw))
            kept_backup = Path(str(entry.get("backup") or ""))  # read beside the file, through its folder, or not at all
            first = target.read(kept_backup.name)[0] if entry.get("backup") and kept_backup.parent == path.parent else None
            if entry and digest(raw) == entry.get("after") and entry.get("before") is None:
                target.remove()  # kabu made the file: as kabu left it, it goes
            elif entry and digest(raw) == entry.get("after") and first is not None and digest(first) == entry.get("before"):
                target.replace(first, mode)  # as kabu left it: the bytes from before, exactly
            else:  # edited since: only kabu's entries go
                updated = {**settings, "hooks": hooks}
                if not hooks and "hooks" not in json.loads(first or b"{}"):
                    updated.pop("hooks")
                target.replace((json.dumps(updated, indent=2, ensure_ascii=False) + "\n").encode(), mode)
        manifest.pop(key, None)
        if not manifest:
            unlink(hook)
    write_private(manifest_path, manifest)
    return {"ok": True, "harness": a["harness"], "settings": str(path), "hook": str(hook), "installed": install,
            "events": list(hook_entries(a["harness"], hook)) if install else [], "changed": changed,
            "backup": saved[0] if saved else None}


def cmd_presence(cmd: str, a: JSON) -> int:
    """presence run, status, on, off, install and uninstall (CONTRACT.md, Presence)."""
    action = cmd.split()[1]
    check(cmd, COMMANDS[cmd].opts, a)
    if action == "run":
        return presence_run(a)
    if action == "herdr":
        if a["state"] not in ("on", "off"):
            raise CliError("missing_argument" if a["state"] is None else "invalid_value", "kabu presence herdr takes on or off.",
                           "Example: kabu presence herdr on")
        if not (os.path.isdir(kabu_home()) or own_dir(kabu_home(), create=True)) or not own_dir(presence_dir(), create=True):
            raise CliError("unsafe_folder", f"{presence_dir()} is not a folder of this user's, or is a link.", "Remove it, then run this again.")
        if a["state"] == "on":
            write_private(presence_dir() / "herdr", {"herdr": True})
        else:
            unlink(presence_dir() / "herdr")
    if action in ("on", "off"):
        if not (os.path.isdir(kabu_home()) or own_dir(kabu_home(), create=True)) or not own_dir(presence_dir(), create=True):
            raise CliError("unsafe_folder", f"{presence_dir()} is not a folder of this user's, or is a link.", "Remove it, then run this again.")
        if action == "off":
            write_private(presence_dir() / "off", {"off": True})
        else:
            unlink(presence_dir() / "off")
    if action in ("install", "uninstall"):
        if not a["harness"]:
            raise CliError("missing_flag", f"kabu presence {action} needs --harness claude, copilot, codex, gemini, cursor or pi.",
                           f"Example: kabu presence {action} --harness claude")
        return emit(presence_install(a, action == "install"))
    return emit(presence_status())


@dataclass(frozen=True)
class Command:
    handler: Callable[[str, JSON], int]
    about: str
    examples: str
    opts: Tuple[Opt, ...]
    positional: Tuple[str, ...] = ()  # name, nargs, help


COMMANDS = {
    "send": Command(cmd_message, "Notify a human. Prints {ok, id, status, attachments}.", """\
  kabu send "Tests passed" "412 passed, 0 failed" --file report.png --link "PR=https://github.com/o/r/pull/7"
  kabu send "Nightly report" - < report.md
  git diff | kabu send "Review this" --diff - --diff-path src/app.py""", MESSAGE, TEXT_ARG),
    "ask": Command(cmd_message, "Ask a human and wait. Exit 0 approved or a choice or text given; 10 rejected; 20 expired; "
                   "21 cancelled or deleted.", """\
  kabu ask "Deploy to prod?"                  approve (exit 0) or reject (exit 10)
  kabu ask "Which env?" -c staging -c prod    a choice: exit 0 whatever is chosen, so branch on .option
  kabu ask "Release name?" --type text        the answer's text is what they typed
  kabu ask "Run it?" --action db.migrate --args '{"version": 42}' --irreversible""", MESSAGE, TEXT_ARG),
    "status": Command(cmd_status, "Print a message as the API has it; exit 0 whatever the answer, so read .resolution.decision.", """\
  kabu status msg_01J9Z --wait 50     hold up to 50 s while an ask is pending""", STATUS, ("id", "?", "a msg_… id")),
    "whoami": Command(cmd_whoami, "Print what the key can do, its org, and the project (and why) and agent kabu would send.", """\
  kabu whoami""", CONNECTION),
    "session start": Command(cmd_session, "Get or create a running session with a Live Activity. Prints the session.", """\
  export KABU_SESSION=$(kabu session start --title "Fix flaky auth test" | jq -r .id)
  kabu session start --title "Nightly build" --external-id run-77 --headline "Compiling" --progress 1/4""",
                             SESSION_CREATE + SESSION_UPDATE + CONNECTION),
    "session update": Command(cmd_session, "Update a session's Live Activity. Prints the session.", """\
  kabu session update --headline "Deleting projects" --progress 54/172 --progress-label projects

{G}""".replace("{G}", PROGRESS_RULE), SESSION_UPDATE + CONNECTION, SESSION_ID),
    "session end": Command(cmd_session, "End a session; its pending asks are cancelled. Prints the session.", """\
  kabu session end --outcome success --summary "PR opened"
  kabu session end ses_01J9Z --outcome failure --dismiss-after 0""", SESSION_END, SESSION_ID),
    "login": Command(cmd_login, "Save a key and defaults to the config file (mode 600). Exit 10: code denied; 20: expired.", """\
  kabu login --device                   prints a code to enter in the Kabu app, then saves this computer's key
  kabu login --api-key - < key.txt      no prompt; with no flag at a terminal, kabu asks for the key""", LOGIN),
    "mcp": Command(cmd_mcp, "Serve Kabu's MCP tools on stdin and stdout for a harness that starts it, through the API's /mcp "
                   "and the saved key. Exit 0 when stdin closes.", """\
  claude mcp add kabu -- kabu mcp       Claude Code; the kabu plugin adds it for you
  copilot mcp add kabu -- kabu mcp      GitHub Copilot CLI
  echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' | kabu mcp""", CONNECTION),
    "presence status": Command(cmd_presence, "Print whether presence is on, the reporter, each agent session's record on this "
                               "machine and the last batch the API took.", """\
  kabu presence status | jq '.lanes[] | {runtime, session}'""", ()),
    "presence on": Command(cmd_presence, "Turn presence on again on this machine (it is on by default). Prints the status.", """\
  kabu presence on""", ()),
    "presence off": Command(cmd_presence, "Turn presence off on this machine: the hooks write nothing, the reporter takes its "
                            "sessions offline and exits, and kabu send joins no presence session. Prints the status.", """\
  kabu presence off                     KABU_PRESENCE=0 does the same for one agent's environment""", ()),
    "presence run": Command(cmd_presence, "Report this machine's agent sessions to Kabu until none is left for 10 minutes. The "
                            "hook starts it; a second one exits at once.", """\
  kabu presence run &""", CONNECTION),
    "presence herdr": Command(cmd_presence, "Turn the Herdr bridge on or off on this machine (off by default): while on and herdr "
                             "is on PATH, the reporter reads herdr agent list every 5 s, and Herdr's state for a session it names "
                             "by the same id beats the hooks' working and idle. Prints the status.", """\
  kabu presence herdr on""", (), ("state", "?", "on or off")),
    "presence install": Command(cmd_presence, "Add the presence hook to a harness's settings, after a backup; running it again "
                                "changes nothing. Not needed with the Kabu plugin, which has the hook.", """\
  kabu presence install --harness claude
  kabu presence install --harness copilot --settings ~/.copilot-work/settings.json
  kabu presence install --harness codex       Codex then asks you to trust its hooks once, in /hooks
  kabu presence install --harness pi          an extension pi loads at its next start""", PRESENCE_SETTINGS),
    "presence uninstall": Command(cmd_presence, "Take the presence hook out of a harness's settings, after a backup.", """\
  kabu presence uninstall --harness claude""", PRESENCE_SETTINGS),
}


def run(args: List[str]) -> int:
    """A command, its help, the version or the guide; with no command, kabu.py's form."""
    if not args:
        raise CliError("missing_command", "kabu needs a command or a message.", 'Run: kabu --help, or: kabu send "Build passed"')
    head = args[0]
    if head in ("-h", "--help") or args[:2] in (["session", "--help"], ["presence", "--help"]):
        print(__doc__)
        return EXIT_OK
    if head == "--version":
        print(f"kabu {VERSION}")
        return EXIT_OK
    if head == "guide" or args[:2] == ["help", "agents"]:  # the rules, then every command's examples
        print(__doc__ + GUIDE + "".join(f"\n\nkabu {name}: {c.about}\n{c.examples}" for name, c in COMMANDS.items()))
        return EXIT_OK
    if head == "help":
        if args[1:] and args[1] not in NAMES:
            raise CliError("unknown_command", f"kabu has no {args[1]!r} command.", "Run: kabu --help")
        return run(args[1:] + ["--help"])
    name = " ".join(args[:2]) if head in ("session", "presence") else head
    if name in COMMANDS:
        return COMMANDS[name].handler(name, parse(name, args[len(name.split()):]))
    if head in ("session", "presence"):
        subs = [n.split()[1] for n in COMMANDS if n.startswith(head + " ")]
        close = difflib.get_close_matches(" ".join(args[1:2]), subs, n=1)
        raise CliError("unknown_command", f"kabu {head} takes {', '.join(subs[:-1])} or {subs[-1]}, not {' '.join(args[1:2])!r}."
                       + (f" Did you mean {close[0]!r}?" if close else ""), f"Example: kabu {head} {SUBCOMMAND_SAMPLES[head]}")
    if head in LEGACY:  # never a message: a steering LaunchAgent may still run `kabu.py daemon`
        raise CliError("unknown_command", f"kabu has no {head!r} command: kabu.py's steering ({', '.join(LEGACY[:5])}) and project "
                       "settings are not in kabu.", f"Use the legacy kabu.py 0.2.0 for it. To send the word itself: kabu send {head}")
    close = difflib.get_close_matches(head, NAMES, n=1, cutoff=0.75) if head.isalpha() else []
    if close:
        raise CliError("unknown_command", f"kabu has no {head!r} command. Did you mean {close[0]!r}?",
                       f"Run: kabu {close[0]} --help (to send the word itself: kabu send {head})")
    cmd = "ask" if {arg.split("=", 1)[0] for arg in args} & {"--ask", "--ask-text"} else "send"
    a = parse(cmd, args)
    if a["ask"] and not a["type"] and not a["allow_reply"] and tuple(c.strip().lower() for c in a["choice"]) in YES_NO:
        a["type"] = "confirm"  # the first approves (exit 0), the second rejects (exit 10); any other pair stays a choice
    return cmd_message(cmd, a)


def main(argv: Optional[Sequence[str]] = None) -> int:
    try:
        return run(list(sys.argv[1:] if argv is None else argv))
    except CliError as e:
        error = e
    except (OSError, ValueError, RuntimeError) as exc:  # a local failure kabu could not foresee, such as an unwritable config
        error = CliError("local_error", f"{type(exc).__name__}: {exc}.", "Check the file, permission or setting it names.")
    except SystemExit as e:  # argparse after --help
        return e.code if isinstance(e.code, int) else EXIT_OK
    except KeyboardInterrupt:
        return 130
    print(error_line(error), file=sys.stderr)
    return error.exit


if __name__ == "__main__":
    raise SystemExit(main())