#!/usr/bin/python3.12
"""
pyrepld -- keeps named Python sessions alive between agent conversations.

Rules: the namespace lives as long as a process, never longer; nothing is
replayed; what was lost is written at death and reported at birth; every
reply reminds you of runs still going elsewhere.

Design, and why (README.md has more):
  * Sessions. Each name is its own worker process (worker.py), created on
    first use. Two conversations no longer share one namespace by accident,
    one session can be interrupted or killed without touching the others,
    and a session can run under any interpreter (a venv, say).
  * Unix socket in $XDG_RUNTIME_DIR, peer uid checked on both ends. A TCP
    port on localhost with no auth would let any local account run code as you.
  * No guessing which code is slow. Every run starts synchronously; if it is
    still going after `timeout` seconds the call returns and the run carries
    on, collectable with `wait`. (Guessing from the text misfires: ", ".join(x)
    is not a background job.)
  * Output is captured, so the caller sees what the code printed.
  * Interrupt is real: SIGINT to the worker, KeyboardInterrupt in your code.
  * Sessions can live on other hosts (ssh, one multiplexed connection per
    host, optional jump hosts) or run as root (sudo -n, or sudo -S with the
    password relayed from the caller's askpass). Both talk the worker
    protocol over stdin/stdout instead of inherited pipes. A remote host
    needs nothing but sh and some python3: the code is sent over (boot.py).
  * Per-host settings come from ~/.config/pyrepl/config.json (CONFIG).
    The user's own modules sit next to it in modules/; sessions import
    them by name and the source comes from here (hub_module).
  * The daemon is a hub: code in one session can open another and copy
    objects to it, from it, or call functions in it (put/get/call). The
    worker asks over its control pipe, so remote sessions can do it too.
  * Idle sessions are swapped: a day unused, the namespace is pickled to
    disk and the process ends; the next use brings it back. Not a replay:
    what doesn't pickle is lost and said so (see "swap" below).

Wire format: one JSON request line, one JSON reply line, one connection per
call, on $XDG_RUNTIME_DIR/pyrepl/sock. Ops: exec, wait, interrupt, kill,
sessions, swap, shutdown. Normally started on demand by `pyrepl` (the client) as a
transient systemd --user unit named pyrepld; see README.md.
"""
import base64
import codecs
import fcntl
import fnmatch
import glob
import hashlib
import io
import json
import os
import re
import shlex
import signal
import socket
import stat
import struct
import subprocess
import sys
import threading
import time
import zlib

HERE = os.path.dirname(os.path.abspath(__file__))
WORKER = os.path.join(HERE, "worker.py")
HOST = socket.gethostname().split(".")[0]

RUNDIR = os.path.join(os.environ.get("XDG_RUNTIME_DIR") or "/tmp/pyrepl-%d" % os.getuid(), "pyrepl")
SOCK = os.path.join(RUNDIR, "sock")
# The home may be shared between hosts (NFS) while each host has its own
# daemon, so state is kept per host or two machines' records would collide.
STATE_ROOT = os.path.join(os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state"),
                          "pyrepl")
STATE = os.path.join(STATE_ROOT, HOST)
# Set by `pyrepl` when it starts us as a systemd unit with memory limits.
LIMITED = os.environ.get("PYREPLD_LIMITED") == "1"


CONFIG = os.path.join(os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config"),
                      "pyrepl", "config.json")


def config_for(host):
    """Settings for host. Top-level keys are the defaults; every pattern
    under "hosts" (fnmatch) that matches host overrides them, in file order.
    Read at each session start, so an edit needs no restart. Keys:
      pythons    extra places to look for python: paths, directories, globs
      jump       jump hosts, in order (ssh -J)
      sudo       auto (sudo -n, else ask), nopasswd, askpass (always ask)
      ssh        the ssh command and its options, replacing SSH_OPTS
      multiplex  false if that ssh is not OpenSSH (no ControlMaster)
    Read for this host only (top level and its own patterns):
      swap_after_hours  idle time before a session is swapped to disk (24; 0 = never)
      swap_keep_days    how long an unrestored swap is kept (7)"""
    try:
        with open(CONFIG) as fh:
            raw = json.load(fh)
    except FileNotFoundError:
        return {}
    except ValueError as exc:
        raise RuntimeError("%s is not valid JSON: %s" % (CONFIG, exc))
    out = dict((k, v) for k, v in raw.items() if k != "hosts")
    for pat, over in (raw.get("hosts") or {}).items():
        if fnmatch.fnmatch(host, pat):
            out.update(over)
    return out


# Also run by sh on remote hosts, so it must parse under python 2 as well.
PROBE = 'import sys; v = sys.version_info; sys.stdout.write("%d.%d" % v[:2] if v >= (3, 6) else "old")'


def py_minor(p):
    """'3.11' for an interpreter that runs and is 3.6 or later, else None."""
    try:
        out = subprocess.run([p, "-c", PROBE], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                             timeout=10).stdout.decode()
    except (OSError, subprocess.SubprocessError):
        return None
    return out if out.startswith("3.") else None


def _default_python():
    """Interpreter for local sessions: $PYREPL_PYTHON, else the first of the
    config's pythons that runs, else ours. Its minor version is the one
    remote sessions look for, so functions copy between them."""
    for e in [os.environ.get("PYREPL_PYTHON") or ""] + list(config_for(HOST).get("pythons") or []):
        for p in sorted(glob.glob(e)) if "/" in e else []:
            for q in [os.path.join(p, n) for n in ("python3", "python")] if os.path.isdir(p) else [p]:
                minor = py_minor(q)
                if minor:
                    return q, minor
    return sys.executable, "%d.%d" % sys.version_info[:2]


DEFAULT_PYTHON, COPY_MINOR = _default_python()   # other minor versions: data copies, code doesn't


def _json_escape(err):
    # Same as worker.py: lone surrogates (in code a client sent, say) leave as
    # a visible "\udce9" inside the JSON string instead of failing the reply.
    return "".join("\\\\u%04x" % ord(c) for c in err.object[err.start:err.end]), err.end


codecs.register_error("pyrepl-json", _json_escape)
NAME_RE = re.compile(r"^[A-Za-z0-9_.-]{1,64}$")
HOST_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.@-]{0,127}$")
PATH_RE = re.compile(r"^[A-Za-z0-9_./+-]{1,256}$")
GLOB_RE = re.compile(r"^[A-Za-z0-9_./+*?-]{1,256}$")
JUMP_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.@-]{0,127}(:[0-9]{1,5})?$")
SUDO = os.environ.get("PYREPL_SUDO", "sudo")      # tests point these at tests/fake-sudo
REMOTE_SUDO = os.environ.get("PYREPL_REMOTE_SUDO")
SSH_OPTS = ["-T", "-o", "BatchMode=yes",
            "-o", "ServerAliveInterval=30", "-o", "ServerAliveCountMax=3"]
HOP_TIMEOUT = 15     # ssh ConnectTimeout per hop: through jumps it covers them all
# Looked for after the caller's and the config's places, newest first.
REMOTE_PYTHONS = ["python3.%d" % m for m in range(14, 5, -1)] + [
    "python3", "/usr/libexec/platform-python", "python"]
OUT_STORE_CAP = 2000000      # chars of undelivered output kept per session
OUT_REPLY_CAP = 30000        # chars of output returned in one reply
LONG_RUN = 300

SESSIONS = {}
SESSIONS_LOCK = threading.Lock()
CREATE_LOCK = threading.Lock()


def log(msg):
    # Also to a file: the user's own journal isn't readable everywhere.
    line = "%s %s" % (time.strftime("%Y-%m-%dT%H:%M:%S"), msg)
    print(line, file=sys.stderr, flush=True)
    try:
        with open(os.path.join(RUNDIR, "pyrepld.log"), "a") as fh:
            fh.write(line + "\n")
    except OSError:
        pass


def ago(t):
    s = int(time.time() - t)
    if s < 120:
        return "%ds" % s
    if s < 7200:
        return "%dm" % (s // 60)
    if s < 172800:
        return "%dh" % (s // 3600)
    return "%dd" % (s // 86400)


class Run(object):
    def __init__(self, cell, code, op="exec"):
        self.cell, self.code, self.op = cell, code, op
        self.started = time.time()
        self.done = False
        self.delivered = False
        self.result = self.error = self.seconds = self.blob = None


def remote_argv(host, jump, cfg):
    """ssh command line for a worker on host. The remote command is always
    `exec /bin/sh -s`: what it runs comes over stdin (remote_script), and
    nothing from a cell is ever part of either."""
    if not HOST_RE.match(host or ""):
        raise ValueError("bad host name %r" % host)
    for j in jump:
        if not JUMP_RE.match(j):
            raise ValueError("bad jump host %r" % j)
    ssh = cfg.get("ssh") or ["ssh"] + SSH_OPTS + [
        "-o", "ConnectTimeout=%d" % (HOP_TIMEOUT * (1 + len(jump)))]
    if not (isinstance(ssh, list) and all(isinstance(a, str) for a in ssh)):
        raise ValueError("config: ssh must be a list of strings")
    if cfg.get("multiplex", True):
        # A route through jump hosts gets its own master, not a direct one's.
        tag = "-" + hashlib.sha1(",".join(jump).encode()).hexdigest()[:8] if jump else ""
        ssh = ssh + ["-o", "ControlMaster=auto", "-o", "ControlPersist=600",
                     "-o", "ControlPath=" + os.path.join(RUNDIR, "ssh-%C" + tag)]
    return ssh + (["-J", ",".join(jump)] if jump else []) + [host, "exec /bin/sh -s"]


# Finds an interpreter (the first of the candidates on COPY_MINOR, else the
# first that runs 3.6+), then execs it on BOOT_STUB, announcing PYREPL-BOOT
# first: only then does the daemon send the payload, so no shell can read
# ahead into it. Candidates are paths, directories, globs or bare names.
REMOTE_SCRIPT = r"""
want=@WANT@ mode=@MODE@ sudo=@SUDO@ probe=@PROBE@ stub=@STUB@
best= bestv=
try_py() {
    [ -f "$1" ] && [ -x "$1" ] || return 1
    v=$("$1" -c "$probe" 2>/dev/null) || return 1
    case $v in 3.*) ;; *) return 1 ;; esac
    [ -n "$best" ] || { best=$1; bestv=$v; }
    [ "$v" = "$want" ] || return 1
    best=$1; bestv=$v
}
look() {
    case $1 in
        */*) if [ -d "$1" ]; then
                 for n in "python$want" python3 python; do try_py "$1/$n" && return 0; done
                 return 1
             fi
             try_py "$1" ;;
        *)   p=$(command -v "$1" 2>/dev/null) && try_py "$p" ;;
    esac
}
for c in @CANDS@; do look "$c" && break; done
[ -n "$best" ] || { echo "no python 3.6+ on $(hostname) in: @CANDS@" >&2; exit 127; }
case $mode in
    user)    cd "$HOME" 2>/dev/null || cd /
             echo "PYREPL-BOOT $bestv $best"
             exec "$best" -u -c "$stub" ;;
    root-n)  "$sudo" -n true 2>/dev/null || { echo PYREPL-NEEDPW; exit 3; }
             cd /
             echo "PYREPL-BOOT $bestv $best"
             exec "$sudo" -n -- "$best" -u -c "$stub" ;;
    root-pw) cd /
             echo "PYREPL-BOOT $bestv $best"
             exec "$sudo" -k -S -p '' -- "$best" -u -c "$stub" ;;
esac
"""

# Reads "<length>\n<payload>" byte-exact from fd 0, so the protocol after it
# stays where the worker will read it.
BOOT_STUB = """import base64, os, zlib
def _r(n):
    b = b""
    while len(b) < n:
        c = os.read(0, n - len(b))
        if not c:
            raise SystemExit("pyrepl: the session code was cut short")
        b += c
    return b
_h = b""
while not _h.endswith(b"\\n"):
    _h += _r(1)
exec(zlib.decompress(base64.b64decode(_r(int(_h)))))
"""


def remote_script(mode, candidates):
    for c in candidates:
        if not GLOB_RE.match(c):
            raise ValueError("bad python place %r (letters, digits, _ . / + - * ? only)" % c)
    if REMOTE_SUDO and not PATH_RE.match(REMOTE_SUDO):
        raise ValueError("bad PYREPL_REMOTE_SUDO")
    fill = {"@WANT@": COPY_MINOR, "@MODE@": mode, "@SUDO@": shlex.quote(REMOTE_SUDO or "sudo"),
            "@PROBE@": shlex.quote(PROBE), "@STUB@": shlex.quote(BOOT_STUB),
            "@CANDS@": " ".join(candidates)}     # unquoted on purpose: globs expand there
    return re.sub("|".join(fill), lambda m: fill[m.group(0)], REMOTE_SCRIPT)


def boot_payload():
    """boot.py with worker.py, tk.py and vendor/ as its FILES, for BOOT_STUB.
    Built per session, so edits reach new remote sessions like local ones."""
    files = {}
    for rel in ("worker.py", "tk.py"):
        with open(os.path.join(HERE, rel), encoding="utf-8") as fh:
            files[rel] = fh.read()
    for dp, dns, fns in os.walk(os.path.join(HERE, "vendor")):
        dns[:] = [d for d in dns if d != "__pycache__"]
        for fn in fns:
            if fn.endswith(".py"):
                p = os.path.join(dp, fn)
                with open(p, encoding="utf-8") as fh:
                    files[os.path.relpath(p, HERE)] = fh.read()
    with open(os.path.join(HERE, "boot.py"), encoding="utf-8") as fh:
        src = "FILES = %r\n" % files + fh.read()
    data = base64.b64encode(zlib.compress(src.encode("utf-8"), 9)).decode("ascii")
    return "%d\n%s" % (len(data), data)


class NeedPassword(Exception):
    """A root session can't start without the sudo password: ask, then retry."""


class Session(object):
    def __init__(self, name, cwd, env, python, host=None, root=False, password=None,
                 pythons=(), jump=(), cfg=None):
        self.name = name
        self.host, self.root, self.jump = host, root, list(jump)
        self.cond = threading.Condition()
        self.wlock = threading.Lock()
        self.run = None
        self.cells = 0
        self.names = []
        self.buf = []            # output not yet handed to a caller, as chunks
        self.buflen = 0
        self.prebuf = ""         # what came before "ready": rc noise, banners, start errors
        self.created = self.last_used = time.time()
        self.dead = None         # reason string once the worker is gone
        self.registered = False  # only registered sessions leave a death record
        self.swap_tried = 0      # last failed swap_out
        self.transcript = os.path.join(STATE, name + ".py")
        self.info = {}
        # Remote and root workers can't inherit our pipes (ssh can't carry
        # them, sudo closes them), so they speak the protocol on stdin/stdout
        # and their raw output comes back on stderr.
        self.stdio = bool(host or root)
        self.password = password          # sent once, then dropped
        if self.stdio:
            if host:
                argv = remote_argv(host, self.jump, cfg or {})
                mode = "root-pw" if root and password else "root-n" if root else "user"
                script = remote_script(mode, [python] if python else list(pythons) + REMOTE_PYTHONS)
            elif password:
                # -k: always ask, so the password line is always sudo's to read
                argv = [SUDO, "-k", "-S", "-p", "", "--", python, "-u", WORKER, "--stdio"]
            else:
                if subprocess.run([SUDO, "-n", "true"], stdin=subprocess.DEVNULL,
                                  stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode:
                    raise NeedPassword("sudo here wants a password")
                argv = [SUDO, "-n", "--", python, "-u", WORKER, "--stdio"]
            self.proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                         stderr=subprocess.PIPE, cwd=cwd if not host else None,
                                         env=env if not host else None, start_new_session=True)
            self.ctl_w = io.TextIOWrapper(self.proc.stdin, encoding="utf-8")
            self.ctl_r = io.TextIOWrapper(self.proc.stdout, encoding="utf-8", errors="replace")
            self.raw = self.proc.stderr
            if host:
                self.send_raw(script)
            elif password:
                self.send_raw(password + "\n")      # read by sudo -S, never logged
                self.password = None
        else:
            ci_r, ci_w = os.pipe()
            co_r, co_w = os.pipe()
            try:
                self.proc = subprocess.Popen(
                    [python, "-u", WORKER, str(ci_r), str(co_w)],
                    stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                    pass_fds=(ci_r, co_w), cwd=cwd, env=env, start_new_session=True)
            except BaseException:
                os.close(ci_w)
                os.close(co_r)
                raise
            finally:
                os.close(ci_r)
                os.close(co_w)
            self.ctl_w = os.fdopen(ci_w, "w", encoding="utf-8")
            self.ctl_r = os.fdopen(co_r, "r", encoding="utf-8")
            self.raw = self.proc.stdout
        ready = threading.Event()
        self.ctl_reader = threading.Thread(target=self._read_ctl, args=(ready,), daemon=True)
        self.ctl_reader.start()
        threading.Thread(target=self._read_raw, daemon=True).start()
        threading.Thread(target=self._wait_exit, args=(ready,), daemon=True).start()
        if not ready.wait((40 + HOP_TIMEOUT * len(self.jump)) if self.stdio else 20) or self.dead:
            self._stop_proc()
            self.password = None
            if "PYREPL-NEEDPW" in self.prebuf:
                raise NeedPassword("sudo on %s wants a password" % host)
            why = (self.prebuf + "".join(self.buf)).strip()[-2000:] or "no output"
            if root and not self.dead:
                why += " (no answer from sudo: a wrong password makes it ask again)"
            raise RuntimeError("worker did not start (%s): %s" % (self.where(), why))

    def where(self):
        return "%s%s" % ("root " if self.root else "", "on " + self.host if self.host else "here")

    def label(self):
        return self.name + ("@" + self.host if self.host else "") + (" [root]" if self.root else "")

    def _boot(self):
        """The remote sh found python and is about to exec it: send what it reads."""
        pw, self.password = self.password, None
        self.send_raw((pw + "\n" if pw else "") + boot_payload())   # pw: read by sudo -S

    # -- worker -> daemon -------------------------------------------------
    def _append(self, s):
        """Caller holds self.cond."""
        self.buf.append(s)
        self.buflen += len(s)
        if self.buflen > OUT_STORE_CAP:
            buf = "".join(self.buf)
            cut = len(buf) - OUT_STORE_CAP // 2
            buf = (buf[:OUT_STORE_CAP // 4]
                   + "\n[... pyrepl dropped %d chars of output nobody had read ...]\n" % cut
                   + buf[OUT_STORE_CAP // 4 + cut:])
            self.buf, self.buflen = [buf], len(buf)

    def _read_ctl(self, ready):
        # Nothing in here may end this thread early: a session whose "done"
        # never gets read shows as running forever.
        for line in self.ctl_r:
            finished = None
            try:
                try:
                    msg = json.loads(line)
                except ValueError:
                    msg = None
                if not self.info and (not isinstance(msg, dict) or msg.get("t") != "ready"):
                    # Before "ready" a remote stdout may carry rc-file chatter.
                    if line.startswith("PYREPL-BOOT ") and self.host:
                        try:
                            self._boot()
                        except OSError:
                            pass                # the link is gone; the reader will see EOF
                    with self.cond:
                        if isinstance(msg, dict) and msg.get("t") == "out":
                            self._append(str(msg.get("s", "")))   # the worker's own early words
                        else:
                            self.prebuf += line
                    continue
                if not isinstance(msg, dict):
                    raise ValueError("not a JSON object")
                t = msg.get("t")
                if t == "hub":
                    threading.Thread(target=self._hub, args=(msg,), daemon=True).start()
                    continue
                with self.cond:
                    if t == "out":
                        self._append(str(msg["s"]))
                    elif t == "ready":
                        self.info = msg
                        ready.set()
                    elif t == "done" and self.run:
                        r = finished = self.run
                        r.result, r.error, r.seconds = msg["result"], msg["error"], msg["seconds"]
                        r.blob = msg.get("blob")
                        r.done = True
                        self.names = msg.get("names") or []
                        self.cond.notify_all()
            except Exception as exc:  # noqa: BLE001
                # Not ours (something wrote to the control fd). Show it rather
                # than lose it.
                log("session %s: bad control line (%s): %r" % (self.name, exc, line[:200]))
                with self.cond:
                    self._append(line)
            if finished:
                self._log_cell(finished)

    def _wait_exit(self, ready):
        # Death is the process exiting, not EOF on the control pipe: a child
        # the cell forked can hold the pipe open long after the worker is gone.
        rc = self.proc.wait()
        self.ctl_reader.join(1.0)        # let a last "done" through first
        if self.host and rc == 255:
            reason = "lost its ssh link to %s" % self.host
        elif self.host:
            reason = "ended on %s (ssh exit %d)" % (self.host, rc)
        elif rc >= 0:
            reason = "exited with status %d" % rc
        else:
            try:
                reason = "killed by %s" % signal.Signals(-rc).name
            except ValueError:
                reason = "killed by signal %d" % -rc
        if rc == -9 and LIMITED and not self.host:
            reason += (" (if nobody killed it, probably the pyrepld memory limit, which all "
                       "sessions share)")
        self._died(reason)
        ready.set()

    def _read_raw(self):
        fd = self.raw.fileno()
        dec = codecs.getincrementaldecoder("utf-8")("replace")
        while True:
            chunk = os.read(fd, 65536)
            text = dec.decode(chunk, final=not chunk)
            if text:
                with self.cond:
                    if self.info or not self.stdio:     # local: sitecustomize warnings etc.
                        self._append(text)
                    else:         # ssh banners and login noise, shown only if start fails
                        self.prebuf += text
            if not chunk:
                return

    def _hub(self, msg):
        """A request from code running in this session; answered on its control pipe."""
        req = msg.get("req") or {}
        try:
            op = req.get("op")
            if op not in HUB_OPS:
                raise ValueError("unknown hub op %r" % op)
            reply = HUB_OPS[op](req, self)
        except Exception as exc:  # noqa: BLE001
            reply = {"error": "%s: %s" % (type(exc).__name__, exc)}
        try:
            self.send({"op": "reply", "id": msg.get("id"), "reply": reply})
        except (OSError, ValueError):
            pass

    def _log_cell(self, r):
        try:
            with open(self.transcript, "a", encoding="utf-8", errors="backslashreplace") as fh:
                fh.write("# --- cell %d  %s  %s  %.2fs\n%s\n"
                         % (r.cell, time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(r.started)),
                            "error" if r.error else "ok", r.seconds or 0, r.code.rstrip()))
        except Exception as exc:  # noqa: BLE001
            log("transcript append failed for %s: %s" % (self.name, exc))

    def _died(self, reason):
        with SESSIONS_LOCK:
            if SESSIONS.get(self.name) is self:
                del SESSIONS[self.name]
        with self.cond:
            if self.dead:
                return
            self.dead = reason
            if self.run and not self.run.done:
                self.run.done = True
                self.run.error = "the session process %s while this was running\n" % reason
            self.cond.notify_all()
        if self.registered:
            write_death(self, reason)
        log("session %s %s" % (self.name, reason))

    # -- daemon -> worker -------------------------------------------------
    def send_raw(self, text):
        with self.wlock:
            self.ctl_w.write(text)
            self.ctl_w.flush()

    def send(self, msg):
        # ASCII JSON: a lone surrogate in a cell must not fail the strict UTF-8 pipe.
        self.send_raw(json.dumps(msg) + "\n")

    def start(self, code, op="exec", **fields):
        """Caller holds self.cond. code is what the transcript records."""
        self.cells += 1
        self.run = Run(self.cells, code, op)
        self.last_used = time.time()
        msg = {"op": op, "cell": self.cells}
        msg.update(fields)
        if op == "exec":
            msg["code"] = code
        try:
            self.send(msg)
        except (OSError, ValueError) as exc:
            # The worker died between our liveness check and now; the exit
            # watcher reports how.
            self.run.done = True
            self.run.error = "could not hand the cell to the session process: %s\n" % exc

    def interrupt(self):
        if self.stdio:          # a remote or root worker is out of reach of our signals
            self.send({"op": "interrupt"})
        else:
            os.kill(self.proc.pid, signal.SIGINT)

    def _stop_proc(self):
        if self.stdio:
            # ssh/sudo only: killing the group could take a shared ssh master
            # with it. The worker sees EOF and exits.
            try:
                self.proc.kill()
            except ProcessLookupError:
                pass
            try:
                self.ctl_w.close()
            except (OSError, ValueError):
                pass
        else:
            try:
                os.killpg(self.proc.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass

    def kill(self, reason):
        self._died(reason)      # first, so the exit watcher's "killed by SIGKILL" loses
        self._stop_proc()

    def take_output(self):
        out = "".join(self.buf)
        self.buf, self.buflen = [], 0
        if len(out) > OUT_REPLY_CAP:
            head, tail = OUT_REPLY_CAP // 3, OUT_REPLY_CAP * 2 // 3
            out = (out[:head] + "\n[... %d chars omitted ...]\n" % (len(out) - head - tail)
                   + out[-tail:])
        return out

    def rss_mb(self):
        if self.stdio:
            return None
        try:
            with open("/proc/%d/status" % self.proc.pid) as fh:
                for line in fh:
                    if line.startswith("VmRSS:"):
                        return round(int(line.split()[1]) / 1024)
        except OSError:
            return None


# -- death records: written at death, reported at the next birth -------------
def death_path(name):
    return os.path.join(STATE, name + ".death.json")


def write_death(s, reason):
    try:
        with open(death_path(s.name), "w") as fh:
            json.dump({"at": time.time(), "reason": reason, "cells": s.cells,
                       "names": [n for n, _ in s.names], "transcript": s.transcript,
                       "host": s.host, "root": s.root}, fh)
    except OSError as exc:
        log("death record failed for %s: %s" % (s.name, exc))


def check_rebirth(name, req):
    """A dead session comes back only where it was: a cell meant for a
    remote (or root) session must not quietly run here as you instead."""
    try:
        with open(death_path(name)) as fh:
            d = json.load(fh)
    except (OSError, ValueError):
        return
    if d.get("reason") == "killed on request":      # a deliberate end frees the name
        return
    was = (d.get("host") or None, bool(d.get("root")))
    now = (req.get("host") or None, bool(req.get("root") or req.get("sudo")))
    if was != now:
        where = ("on %s" % was[0] if was[0] else "on this host") + (" as root" if was[1] else "")
        raise ValueError("session %r ran %s and ended %s ago (%s). Start it again the same way "
                         "(%s), or use another name." % (
                             name, where, ago(d["at"]), d["reason"],
                             ", ".join(filter(None, ["host=%r" % was[0] if was[0] else "",
                                                     "sudo=True" if was[1] else ""])) or "no host"))


def read_death(name, consume=True):
    p = death_path(name)
    try:
        with open(p) as fh:
            d = json.load(fh)
        if consume:
            os.unlink(p)
    except (OSError, ValueError):
        return None
    if not consume:
        return ("Session %r ended %s ago (%s) after %d cell(s). Its code is in %s."
                % (name, ago(d["at"]), d["reason"], d.get("cells", 0), d.get("transcript")))
    names = d.get("names") or []
    shown = ", ".join(names[:15]) + (" ..." if len(names) > 15 else "")
    return ("New session. A previous session named %r ended %s ago (%s) after %d cell(s)%s. "
            "Nothing was carried over; its code is in %s."
            % (name, ago(d["at"]), d["reason"], d.get("cells", 0),
               ", holding %d name(s): %s" % (len(names), shown) if names else "",
               d.get("transcript")))


# -- swap: idle sessions go to disk and come back on next use ----------------
# A session nobody has used for swap_after_hours (config, default 24) is
# pickled to SWAP and its process ends, giving its memory back. The next exec
# or tk.session() under that name starts it again where it was (same host,
# python, cwd) and loads what pickled; the note says what didn't. A swap not
# restored within swap_keep_days (default 7) is deleted and leaves a death
# record instead. Root sessions are never swapped: their namespace would sit
# in the user's files, and bringing one back could need the sudo password.
# A graceful stop swaps every user session too (stop_swapping), so they
# outlive a daemon restart. The sweep runs every SWEEP_EVERY seconds, and
# on demand (op swap without a session).
SWAP = os.path.join(STATE, "swap")
SWEEP_EVERY = 600


def swap_settings():
    """(seconds idle before a swap or None for never, seconds a swap is kept)."""
    cfg = config_for(HOST)
    after = cfg.get("swap_after_hours", 24)
    keep = cfg.get("swap_keep_days", 7)
    return (float(after) * 3600 if after not in (None, False, 0) else None,
            float(keep) * 86400)


def swap_path(name, ext):
    return os.path.join(SWAP, name + ext)


# A swap is code: unpickling the blob runs whatever it says, and the record
# names the python to start. So both are written 0600 in a 0700 dir, and
# read only if they still are: ours, regular files (no links), no group or
# other bits. Anything else is refused, never loaded.
def write_private(path, data):
    tmp = path + ".tmp"
    try:
        os.unlink(tmp)                  # a leftover from a swap cut short
    except FileNotFoundError:
        pass
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, "w") as fh:
        os.fchmod(fd, 0o600)            # whatever the umask
        fh.write(data)
    os.rename(tmp, path)


def read_private(path):
    """The file's text, or OSError if it isn't ours and private."""
    st = os.lstat(SWAP)
    if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid() or st.st_mode & 0o077:
        raise PermissionError("%s is not a private directory of ours; not loading from it" % SWAP)
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
    with os.fdopen(fd) as fh:
        st = os.fstat(fd)
        if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or st.st_mode & 0o077:
            raise PermissionError("%s is not a private file of ours (mode %o, uid %d); not "
                                  "loading it" % (path, st.st_mode & 0o7777, st.st_uid))
        return fh.read()


def read_swap(name):
    """The swap record for name, or None if there is none or it can't be trusted."""
    try:
        rec = json.loads(read_private(swap_path(name, ".json")))
        if not (isinstance(rec, dict) and all(isinstance(rec.get(k), (int, float))
                                              for k in ("at", "idle_since", "size"))):
            raise ValueError("not a swap record")
        return rec
    except FileNotFoundError:
        return None
    except (OSError, ValueError) as exc:
        log("swap record of %s ignored: %s" % (name, exc))
        return None


def drop_swap(name):
    for ext in (".json", ".blob"):
        try:
            os.unlink(swap_path(name, ext))
        except FileNotFoundError:
            pass


def swap_out(s, why):
    """Pickle s's namespace to disk and end it. None when done, else why not."""
    if s.root:
        return "root sessions are not swapped"
    with s.cond:
        if s.dead:
            return "already ended"
        r = s.run
        if r and not r.done:
            return "busy with cell %d" % r.cell
        before, unread = r, None
        if r and not r.delivered:
            unread = (("error: " + r.error.strip().splitlines()[-1]) if r.error
                      else "result: %s" % (r.result if r.result is not None else "None")[:500])
            unread = "cell %d finished unread, %s" % (r.cell, unread)
        pending = s.take_output()
        idle = s.last_used
        s.start("# swapped out to disk (%s)" % why, "swapout")
        while not s.run.done and not s.dead:
            s.cond.wait(1.0)
        r = s.run
        r.delivered = True
        s.take_output()                 # pickling chatter (warnings) isn't the user's
        s.last_used = idle              # a swap attempt is not use
        if s.dead or r.error:
            s.swap_tried = time.time()  # the sweep tries again a full swap period later
            if before and not s.dead:   # what nobody had read is still theirs to read
                s.run = before
                s.buf.insert(0, pending)
                s.buflen += len(pending)
            err = s.dead or r.error.strip().splitlines()[-1]
            log("session %s not swapped: %s" % (s.name, err))
            return err
        info = json.loads(r.result)
        rec = {"at": time.time(), "why": why, "idle_since": idle, "cells": s.cells,
               "names": s.names, "lost": info["lost"], "threads": info["threads"],
               "cwd": info["cwd"], "host": s.host, "jump": s.jump,
               "python": s.info.get("executable"), "transcript": s.transcript,
               "unread": unread, "output": pending[-2000:] if pending.strip() else None,
               "size": len(r.blob)}
        try:
            private_dir(SWAP)
            write_private(swap_path(s.name, ".blob"), r.blob)
            write_private(swap_path(s.name, ".json"), json.dumps(rec))   # last: marks a whole swap
        except OSError as exc:
            drop_swap(s.name)
            log("session %s not swapped, can't write %s: %s" % (s.name, SWAP, exc))
            return "can't write the swap file: %s" % exc
        s.registered = False            # a swap is not a death: no death record
        s.kill("swapped to disk")
    return None


def check_swap(name, req):
    """A swapped session comes back only as what it was, like a dead one."""
    rec = read_swap(name)
    if rec is None:
        return None
    if req.get("root") or req.get("sudo"):
        raise ValueError("session %r is a swapped-out user session; use it without sudo, or "
                         "`kill` it to free the name" % name)
    if "host" in req and (req.get("host") or None) != rec.get("host"):
        raise ValueError("session %r is swapped out and ran %s; start it again without host "
                         "or with host=%r, or use another name"
                         % (name, "on " + rec["host"] if rec.get("host") else "on this host",
                            rec.get("host")))
    return rec


def restore(s, rec):
    """Load a swap into the fresh session s (not yet registered). Returns the note."""
    try:
        blob = read_private(swap_path(s.name, ".blob"))
    except OSError as exc:
        log("swap of %s not loaded: %s" % (s.name, exc))
        return "Session %r was swapped out but its swap file was not loaded (%s); this is a " \
               "new, empty session." % (s.name, exc)
    with s.cond:
        s.cells = rec.get("cells", 0)
        s.start("# restored from swap (swapped out %s)"
                % time.strftime("%Y-%m-%d %H:%M", time.localtime(rec["at"])), "swapin",
                blob=blob, cwd=rec.get("cwd"))
        while not s.run.done and not s.dead:
            s.cond.wait(1.0)
        s.run.delivered = True
        s.take_output()
        err = s.dead or s.run.error
        if not err:
            s.info["cwd"] = json.loads(s.run.blob)["cwd"]
    if err:
        return ("Session %r was swapped out %s ago but would not load (%s); this is a new, "
                "empty session. The swap is kept in %s until it expires."
                % (s.name, ago(rec["at"]), err.strip().splitlines()[-1], SWAP))
    drop_swap(s.name)
    note = "Restored session %r from disk: it was swapped out %s ago after %s idle, %d name(s) " \
           "back." % (s.name, ago(rec["at"]), ago_span(rec["at"] - rec["idle_since"]),
                      len(s.names))
    if rec.get("lost"):
        note += " Lost (would not pickle): %s." % ", ".join(
            "%s (%s)" % (n, t) for n, t, _ in rec["lost"])
    if rec.get("threads"):
        note += " Threads that were running are gone: %s." % ", ".join(rec["threads"])
    if rec.get("cwd") and rec["cwd"] != s.info.get("cwd"):
        note += " Its cwd %s is gone; now in %s." % (rec["cwd"], s.info.get("cwd"))
    if rec.get("unread"):
        note += " At swap time, %s." % rec["unread"]
    if rec.get("output"):
        note += " Output nobody had read at swap time:\n" + rec["output"].rstrip()
    return note


def ago_span(sec):
    return ago(time.time() - sec)


def purge_swaps(keep):
    """Swaps older than keep go, leaving a death record to say so."""
    try:
        files = os.listdir(SWAP)
    except FileNotFoundError:
        return
    now = time.time()
    for fn in files:
        if fn.endswith(".tmp"):
            try:
                if now - os.path.getmtime(os.path.join(SWAP, fn)) > 3600:
                    os.unlink(os.path.join(SWAP, fn))   # a swap cut short by a daemon stop
            except FileNotFoundError:
                pass
            continue
        if fn.endswith(".blob") and not os.path.exists(os.path.join(SWAP, fn[:-5] + ".json")):
            try:                    # a blob whose record never got written
                if now - os.path.getmtime(os.path.join(SWAP, fn)) > 3600:
                    os.unlink(os.path.join(SWAP, fn))
            except FileNotFoundError:
                pass
            continue
        if not fn.endswith(".json"):
            continue
        name = fn[:-5]
        rec = read_swap(name)
        if rec is None or now - rec["at"] < keep:
            continue
        with CREATE_LOCK:           # not while an exec is bringing it back
            if read_swap(name) is None:
                continue
            try:
                with open(death_path(name), "w") as fh:
                    json.dump({"at": rec["at"], "cells": rec.get("cells", 0),
                               "reason": "swapped to disk after %s idle, then deleted unrestored "
                                         "%s later" % (ago_span(rec["at"] - rec["idle_since"]),
                                                       ago_span(now - rec["at"])),
                               "names": [n for n, _ in rec.get("names") or []],
                               "transcript": rec.get("transcript"), "host": rec.get("host"),
                               "root": False}, fh)
            except OSError as exc:
                log("death record failed for %s: %s" % (name, exc))
            drop_swap(name)
        log("swap of %s deleted, unrestored since %s"
            % (name, time.strftime("%Y-%m-%d %H:%M", time.localtime(rec["at"]))))


def sweep():
    """Swap out idle sessions, delete old swaps. Returns what it swapped."""
    after, keep = swap_settings()
    done = []
    if after:
        with SESSIONS_LOCK:
            live = list(SESSIONS.values())
        for s in live:
            if (not s.root and not s.dead
                    and time.time() - max(s.last_used, s.swap_tried) >= after):
                why = "idle %s" % ago(s.last_used)
                if swap_out(s, why) is None:
                    log("session %s swapped to disk (%s)" % (s.name, why))
                    done.append(s.name)
    purge_swaps(keep)
    return done


def swapper():
    """sweep() in the background, every SWEEP_EVERY seconds."""
    while True:
        time.sleep(SWEEP_EVERY)
        try:
            sweep()
        except Exception as exc:  # noqa: BLE001 -- a bad config must not end the sweeps
            log("swap sweep failed: %s: %s" % (type(exc).__name__, exc))


def swapped_list(want=None):
    try:
        names = sorted(fn[:-5] for fn in os.listdir(SWAP)
                       if fn.endswith(".json") and NAME_RE.match(fn[:-5]))
    except FileNotFoundError:
        return []
    out = []
    for name in names:
        if want and name != want:
            continue
        rec = read_swap(name)
        if rec:
            out.append({"session": name, "host": rec.get("host"), "swapped": ago(rec["at"]),
                        "cells": rec.get("cells"), "size_mb": round(rec.get("size", 0) / 1e6, 1),
                        "names": [n for n, _ in rec.get("names") or []][:40],
                        "name_count": len(rec.get("names") or [])})
    return out


# -- ops ---------------------------------------------------------------------
def get_session(name, req=None):
    """Existing session, or a new one if req (an exec request) is given."""
    if not NAME_RE.match(name or ""):
        raise ValueError("session names are 1-64 chars of A-Z a-z 0-9 _ . - (got %r)" % name)
    with SESSIONS_LOCK:
        s = SESSIONS.get(name)
    if s and s.dead:
        s = None
    if s and req is not None and "host" in req and (req.get("host") or None) != s.host:
        raise ValueError("session %r already exists %s; pick another name"
                         % (name, "on " + s.host if s.host else "on this host"))
    if s or req is None:
        return s, None
    with CREATE_LOCK:
        with SESSIONS_LOCK:
            s = SESSIONS.get(name)
        if s:                         # someone created it while we waited
            return s, None
        swapped = check_swap(name, req)
        if swapped:
            read_death(name)          # an older death is old news
            note = None
            req = dict(req, host=swapped.get("host"), jump=swapped.get("jump") or [])
            if not req.get("python") and swapped.get("python"):
                if swapped.get("host"):
                    if GLOB_RE.match(swapped["python"]):
                        req["pythons"] = [swapped["python"]] + _list(req.get("pythons"))
                elif os.access(swapped["python"], os.X_OK):
                    req["python"] = swapped["python"]
            if swapped.get("cwd") and not swapped.get("host") and os.path.isdir(swapped["cwd"]):
                req["cwd"] = swapped["cwd"]
        else:
            check_rebirth(name, req)
            note = read_death(name)
        host = req.get("host") or None
        root = bool(req.get("root"))
        cfg = config_for(host or HOST)
        sudo_mode = cfg.get("sudo", "auto")
        if sudo_mode not in ("auto", "nopasswd", "askpass"):
            raise ValueError("config: sudo is auto, nopasswd or askpass, not %r" % sudo_mode)
        password = req.get("password")
        if root and not password and sudo_mode == "askpass":
            raise NeedPassword("root sessions %s ask for the password (config: sudo=askpass)"
                               % ("on " + host if host else "here"))
        python = req.get("python") or (None if host else DEFAULT_PYTHON)
        jump = req.get("jump") if req.get("jump") is not None else cfg.get("jump") or []
        try:
            s = Session(name, req.get("cwd") or os.path.expanduser("~"), req.get("env"), python,
                        host=host, root=root, password=password, cfg=cfg,
                        pythons=_list(req.get("pythons")) + _list(cfg.get("pythons")),
                        jump=_list(jump))
        except NeedPassword as exc:
            if sudo_mode == "nopasswd":
                raise RuntimeError("%s, but the config says sudo=nopasswd for it" % exc)
            raise
        if swapped:
            note = restore(s, swapped)
        s.registered = True
        with SESSIONS_LOCK:
            SESSIONS[name] = s
    try:
        with open(s.transcript, "a") as fh:
            fh.write("\n# === session %r started %s on %s%s, pid %s, python %s, cwd %s\n"
                     % (name, time.strftime("%Y-%m-%d %H:%M:%S"), s.info.get("host", HOST),
                        " as root" if s.root else "", s.info.get("pid"),
                        s.info.get("python"), s.info.get("cwd")))
    except OSError:
        pass
    born = "Created session %r %s(python %s%s, cwd %s)." % (
        name, "on %s %s" % (s.host, "via %s " % " > ".join(s.jump) if s.jump else "") if s.host else "",
        s.info.get("python"), " at " + s.info["executable"] if s.host else "", s.info.get("cwd"))
    if s.root:
        born += " It runs as root."
    if not minor_ok(s):
        born += (" Not Python %s: functions won't copy between this session and others "
                 "(put/get/call); plain data still does.%s" % (
                     COPY_MINOR, " If %s has a %s somewhere, kill this session and name the "
                     "place in `pythons`." % (s.host, COPY_MINOR) if s.host else ""))
    return s, (note + " " + born) if note else born


def _list(v):
    """A list from a list, a comma-separated string, or nothing."""
    if not v:
        return []
    return [x.strip() for x in v.split(",") if x.strip()] if isinstance(v, str) else list(v)


def minor_ok(s):
    return (s.info.get("python") or "").startswith(COPY_MINOR + ".")


def report(s, notes=None):
    """Reply for the session's current run. Caller holds s.cond."""
    r = s.run
    out = {"session": s.name, "output": s.take_output()}
    if r:
        out["cell"] = r.cell
        if r.done:
            out.update(state="done", result=r.result, error=r.error, seconds=r.seconds)
            r.delivered = True
        else:
            out.update(state="running", elapsed=round(time.time() - r.started, 1))
    else:
        out["state"] = "idle"
    if s.dead:
        out["dead"] = s.dead
    if notes:
        out["notes"] = notes
    return out


def wait_done(s, timeout):
    deadline = time.time() + max(0.0, timeout)
    while not s.run.done and not s.dead:
        left = deadline - time.time()
        if left <= 0:
            break
        s.cond.wait(min(left, 1.0))


def timeout_of(req, default):
    t = float(req.get("timeout", default))
    if t != t:                                    # NaN
        raise ValueError("timeout must be a number of seconds")
    return min(max(t, 0.0), 600.0)


def op_exec(req):
    code = req.get("code")
    if not isinstance(code, str) or not code.strip():
        raise ValueError("empty code")
    timeout = timeout_of(req, 30)
    notes = []
    for _ in range(2):
        try:
            s, born = get_session(req.get("session"), req)
        except NeedPassword as exc:
            return {"session": req.get("session"), "state": "absent", "need_password": True,
                    "error": str(exc), "notes": notes}
        if born:
            notes.append(born)
        s.cond.acquire()
        if not s.dead:
            break
        s.cond.release()       # killed between lookup and lock: start a fresh one
    else:
        return {"session": s.name, "state": "absent", "notes": notes,
                "error": "the session died as soon as it started: %s" % s.dead}
    try:
        r = s.run
        if r and not r.done:
            return {"session": s.name, "state": "busy", "cell": r.cell,
                    "elapsed": round(time.time() - r.started, 1), "code_head": r.code[:200],
                    "error": "session %r is still running cell %d. Use wait or interrupt, "
                             "or work in another session." % (s.name, r.cell)}
        if r and not r.delivered:
            notes.append("Cell %d finished while nobody was looking: %s" % (
                r.cell, ("error: " + r.error.strip().splitlines()[-1]) if r.error
                else "result: %s" % (r.result if r.result is not None else "None")[:500]))
        pending = s.take_output()
        if pending.strip():
            notes.append("Output that arrived between calls:\n" + pending.rstrip())
        s.start(code)
        wait_done(s, timeout)
        return report(s, notes)
    finally:
        s.cond.release()


def absent(name):
    out = {"session": name, "state": "absent", "error": "no live session by that name"}
    rec = read_swap(name or "") if NAME_RE.match(name or "") else None
    last = read_death(name, consume=False)
    if rec:
        out["notes"] = ["Session %r was swapped to disk %s ago; an exec in it (or "
                        "tk.session(%r)) brings it back." % (name, ago(rec["at"]), name)]
    elif last:
        out["notes"] = [last]
    return out


def op_wait(req):
    s, _ = get_session(req.get("session"))
    if not s:
        return absent(req.get("session"))
    with s.cond:
        if not s.run:
            return report(s)
        if s.run.done and s.run.delivered:
            out = report(s)
            out["notes"] = ["Nothing running; cell %d's result was already delivered." % s.run.cell]
            return out
        wait_done(s, timeout_of(req, 60))
        return report(s)


def op_interrupt(req):
    s, _ = get_session(req.get("session"))
    if not s:
        return absent(req.get("session"))
    with s.cond:
        if not s.run or s.run.done:
            if s.run and not s.run.delivered:
                return report(s, ["Nothing was running; this cell had already finished."])
            return {"session": s.name, "state": "idle", "notes": ["Nothing was running."]}
        s.interrupt()
        wait_done(s, 5)
        if s.run.done:
            return report(s)
        return report(s, ["Sent SIGINT 5s ago and the cell is still running -- it is probably "
                          "inside C code that does not check for signals. `kill` the session "
                          "if it has to stop now (the namespace goes with it)."])


def op_kill(req):
    s, _ = get_session(req.get("session"))
    if not s:
        rec = read_swap(req["session"])
        if not rec:
            return absent(req.get("session"))
        with CREATE_LOCK:
            drop_swap(req["session"])
        return {"session": req["session"], "state": "killed",
                "notes": ["Deleted its swap: %d name(s)%s. Transcript: %s" % (
                    len(rec.get("names") or []), (": " + ", ".join(
                        n for n, _ in rec["names"][:30])) if rec.get("names") else "",
                    rec.get("transcript"))]}
    with s.cond:
        names = [n for n, _ in s.names]
    s.kill("killed on request")
    return {"session": s.name, "state": "killed",
            "notes": ["Discarded %d name(s)%s. Transcript: %s"
                      % (len(names), (": " + ", ".join(names[:30])) if names else "",
                         s.transcript)]}


def describe(s, detail):
    with s.cond:
        r = s.run
        d = {"session": s.name, "host": s.host, "jump": s.jump, "root": s.root,
             "pid": s.info.get("pid") if s.stdio else s.proc.pid, "python": s.info.get("python"),
             "cwd": s.info.get("cwd"), "age": ago(s.created), "idle": ago(s.last_used),
             "cells": s.cells, "rss_mb": s.rss_mb(), "transcript": s.transcript}
        if r and not r.done:
            d["running"] = {"cell": r.cell, "elapsed": round(time.time() - r.started),
                            "code_head": r.code[:120]}
        elif r and not r.delivered:
            d["unread_result"] = r.cell
        d["names"] = s.names if detail else [n for n, _ in s.names][:40]
        d["name_count"] = len(s.names)
    return d


def op_sessions(req):
    with SESSIONS_LOCK:
        live = list(SESSIONS.values())
    want = req.get("session")
    if want:
        live = [s for s in live if s.name == want]
    return {"host": HOST, "pid": os.getpid(), "state_dir": STATE,
            "sessions": [describe(s, bool(want)) for s in sorted(live, key=lambda s: s.name)],
            "swapped": swapped_list(want)}


def op_swap(req):
    """Swap a session to disk now; without a name, run the sweep now."""
    if not req.get("session"):
        done = sweep()
        return {"state": "idle", "notes": ["Swept: %s." % (
            "swapped " + ", ".join(done) if done else "nothing idle long enough to swap")]}
    s, _ = get_session(req.get("session"))
    if not s:
        return absent(req.get("session"))
    why = swap_out(s, "on request")
    if why:
        return {"session": s.name, "state": "idle", "error": "not swapped: %s" % why}
    rec = read_swap(s.name)
    return {"session": s.name, "state": "swapped",
            "notes": ["Swapped to disk (%.1f MB, %d name(s)%s); the next exec brings it back."
                      % (rec["size"] / 1e6, len(rec["names"]), ", lost: " + ", ".join(
                          n for n, _, _ in rec["lost"]) if rec["lost"] else "")]}


def elsewhere(exclude):
    """The house-repl rule: never let a reply forget what is still going."""
    with SESSIONS_LOCK:
        live = list(SESSIONS.values())
    out = []
    for s in live:
        if s.name == exclude:
            continue
        r = s.run
        if r and not r.done:
            age = time.time() - r.started
            item = "%s: cell %d running %s" % (s.name, r.cell, ago(r.started))
            if age > LONG_RUN:
                item += " -- STILL RUNNING, check it"
            out.append(item)
        elif r and not r.delivered:
            out.append("%s: cell %d finished, result unread" % (s.name, r.cell))
    return out


OPS = {"exec": op_exec, "wait": op_wait, "interrupt": op_interrupt, "kill": op_kill,
       "sessions": op_sessions, "swap": op_swap}


# -- hub: requests from code running in a session -----------------------------
def hub_open(req, caller):
    """Get or create a session for tk.session(). cwd/env come from the caller."""
    try:
        s, born = get_session(req.get("session"), req)
    except NeedPassword as exc:
        return {"need_password": True, "error": str(exc)}
    return {"session": s.name, "host": s.host, "root": s.root, "python": s.info.get("python"),
            "cwd": s.info.get("cwd"), "notes": [born] if born else []}


def hub_transfer(req, caller):
    """put/get/call: run the op in the target session and hand back what it made.

    The target is busy for the duration, like a cell, and the transcript
    records it. Output the target prints meanwhile goes to the caller.
    """
    s, _ = get_session(req.get("session"))
    if not s:
        return {"error": "no live session %r; open it with tk.session()" % req.get("session")}
    if s is caller:
        return {"error": "%r is the session this code runs in" % s.name}
    op = req.get("op")
    head = "# %s from session %r" % (op, caller.name)
    label = (req.get("label") or "").rstrip()
    with s.cond:
        r = s.run
        if r and not r.done:
            return {"error": "session %r is busy with cell %d (%s)"
                    % (s.name, r.cell, r.code.split("\n")[0][:80])}
        mark = len(s.buf)          # chunks: what the target prints from here on is the caller's
        fields = dict((k, req[k]) for k in ("blob", "names") if k in req)
        s.start(head + ("\n" + label if label else ""), op, **fields)
        while not s.run.done:
            s.cond.wait(1.0)
        r = s.run
        r.delivered = True
        out = "".join(s.buf[mark:])
        s.buf = s.buf[:mark]
        s.buflen = sum(len(c) for c in s.buf)
        reply = {"result": r.result, "error": r.error, "blob": r.blob, "output": out,
                 "seconds": r.seconds}
        if s.dead:
            reply["dead"] = s.dead
    if r.error and not (minor_ok(s) and minor_ok(caller)):
        reply["error"] += ("(%s runs Python %s and %s runs %s: code objects don't load across "
                           "minor versions; only plain data does)\n"
                           % (caller.name, caller.info.get("python"), s.name, s.info.get("python")))
    return reply


def hub_module(req, caller):
    """pyrepl's modules (pyrepl.NAME), from this host's folders so every
    session sees the same ones (worker._Modules imports them, tk.modules()
    lists them). With a name: its source. Without: the user's folder and
    the names, and with index=True each one's docstring line too. The
    folder logic is tk's."""
    import tk
    name = req.get("name")
    if name is None:
        idx = tk.module_index(docs=bool(req.get("index")))
        return {"folder": tk.MODULES, "names": [n for n, _ in idx], "index": idx}
    if not name.isidentifier():
        return {"error": "bad module name %r" % name}
    path = tk.module_path(name)
    if path is None:
        return {"error": "no module %r in %s or %s" % (name, tk.MODULES, tk.SHIPPED)}
    try:
        with open(path, encoding="utf-8") as fh:
            return {"source": fh.read(), "path": path}
    except FileNotFoundError:
        return {"error": "no module %r in %s" % (name, tk.MODULES)}
    except (OSError, UnicodeDecodeError) as exc:
        return {"error": "can't read %s: %s" % (path, exc)}


def hub_ssh(req, caller):
    """The ssh line a session for host would use (config ssh and jump, the
    shared connection), for tk.sh(host=): one-off commands get the same route
    and the same shell-agnostic remote command. That route starts here as
    this user: a remote or root caller gets {"plain": True} and uses plain
    ssh from where it is, as who it is."""
    if caller.host or caller.root:
        return {"plain": True}
    host = req.get("host")
    cfg = config_for(host or "")
    jump = req.get("jump") if req.get("jump") is not None else cfg.get("jump") or []
    return {"argv": remote_argv(host, _list(jump), cfg)}


HUB_OPS = {"open": hub_open, "put": hub_transfer, "get": hub_transfer, "call": hub_transfer,
           "module": hub_module, "ssh": hub_ssh,
           "interrupt": lambda req, caller: op_interrupt(req),
           "kill": lambda req, caller: op_kill(req),
           "sessions": lambda req, caller: op_sessions(req)}


def serve(conn):
    try:
        uid = struct.unpack("3i", conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED,
                                                  struct.calcsize("3i")))[1]
        if uid != os.getuid():
            return
        f = conn.makefile("rwb")
        line = f.readline()
        if not line:
            return
        try:
            req = json.loads(line)
            op = req.get("op")
            if op == "shutdown":
                swapped, lost = stop_swapping()
                f.write((json.dumps({"state": "stopping", "swapped": swapped, "lost": lost})
                         + "\n").encode())
                f.flush()
                os.kill(os.getpid(), signal.SIGTERM)
                return
            if op not in OPS:
                raise ValueError("unknown op %r" % op)
            out = OPS[op](req)
            others = elsewhere(req.get("session")) if op != "sessions" else None
            if others:
                out["other_sessions"] = others
        except Exception as exc:  # noqa: BLE001
            out = {"error": "%s: %s" % (type(exc).__name__, exc)}
        f.write((json.dumps(out, ensure_ascii=False) + "\n").encode("utf-8", "pyrepl-json"))
        f.flush()
    except OSError:
        pass
    finally:
        conn.close()


STOPPING = threading.Lock()
STOP_SWAP_DEADLINE = 90     # the unit's TimeoutStopSec (in pyrepl) leaves room for it
STOP_SWAP_THREADS = 4       # pickling all at once could hit the memory cap


def stop_swapping():
    """A graceful stop swaps every user session out first, so the next
    daemon brings them back. Runs once: a second SIGTERM means stop now.
    Returns (swapped names, "name: why not" for the rest)."""
    if not STOPPING.acquire(False):
        return [], []
    try:
        os.unlink(SOCK)           # nothing new comes in while we swap
    except OSError:
        pass
    with SESSIONS_LOCK:
        live = [s for s in SESSIONS.values() if not s.dead]
    todo = [s for s in live if not s.root]
    why = dict((s.name, "root sessions are not swapped") for s in live if s.root)
    deadline = time.time() + STOP_SWAP_DEADLINE

    def work():
        while time.time() < deadline:
            with SESSIONS_LOCK:
                if not todo:
                    return
                s = todo.pop(0)
            why[s.name] = swap_out(s, "pyrepld stopped")

    threads = [threading.Thread(target=work, daemon=True)
               for _ in range(min(STOP_SWAP_THREADS, len(todo)))]
    for t in threads:
        t.start()
    for t in threads:
        t.join(max(0, deadline - time.time()))
    for s in live:
        why.setdefault(s.name, "no time left to swap it")
    swapped = sorted(n for n, w in why.items() if w is None)
    lost = sorted("%s: %s" % (n, w) for n, w in why.items() if w is not None)
    log("stopping: %d session(s) swapped to disk%s" % (
        len(swapped), "; not swapped: " + "; ".join(lost) if lost else ""))
    return swapped, lost


def shutdown(*_):
    stop_swapping()
    with SESSIONS_LOCK:
        live = list(SESSIONS.values())
    for s in live:
        s.kill("pyrepld stopped")
    try:
        os.unlink(SOCK)
    except OSError:
        pass
    log("stopped; %d session(s) discarded" % len(live))
    os._exit(0)


def private_dir(path):
    """Create path (and parents) as 0700, then insist that it is ours and not a link."""
    parent = os.path.dirname(path)
    if not os.path.isdir(parent):
        private_dir(parent)
    try:
        os.mkdir(path, 0o700)
    except FileExistsError:
        pass
    st = os.lstat(path)
    if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.getuid():
        raise SystemExit("pyrepld: %s is not a directory owned by uid %d; refusing to use it"
                         % (path, os.getuid()))
    os.chmod(path, 0o700)


def main():
    if not os.environ.get("XDG_RUNTIME_DIR"):
        private_dir(os.path.dirname(RUNDIR))   # /tmp/pyrepl-UID: nobody else may own it
    private_dir(RUNDIR)
    os.makedirs(os.path.dirname(STATE_ROOT), exist_ok=True)
    private_dir(STATE_ROOT)        # transcripts hold code, and code holds secrets
    private_dir(STATE)
    lock = open(os.path.join(RUNDIR, "lock"), "w")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        log("another pyrepld already holds %s; exiting" % RUNDIR)
        return 0
    try:
        os.unlink(SOCK)
    except FileNotFoundError:
        pass
    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    srv.bind(SOCK)
    os.chmod(SOCK, 0o600)
    srv.listen(32)
    signal.signal(signal.SIGTERM, shutdown)
    signal.signal(signal.SIGINT, shutdown)
    threading.Thread(target=swapper, daemon=True).start()
    log("pyrepld listening on %s | state %s | default python %s | pid %d"
        % (SOCK, STATE, DEFAULT_PYTHON, os.getpid()))
    while True:
        conn, _ = srv.accept()
        threading.Thread(target=serve, args=(conn,), daemon=True).start()


if __name__ == "__main__":
    sys.exit(main())