#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["cryptography>=43,<47", "websockets>=15,<17"]
# ///
"""Native yink CLI. See docs/plan/PYTHON-CLI.md for capabilities and limits.

Run the downloaded file with `uv run yink.py --help`, or install the package
in a dedicated Python 3.11+ virtual environment. No Node runtime is used.
"""
from __future__ import annotations

import asyncio
import base64
import codecs
import contextlib
import datetime as dt
import getpass
import hashlib
import hmac
import ipaddress
import json
import os
from pathlib import Path
import re
import signal
import socket
import stat
import struct
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

DEFAULT_ORIGIN = "https://yink.io"
MAX_PLAIN = 24 * 1024
MAX_INT = 2**53 - 1
TTLS = {"1h": 3600, "1d": 86400, "3d": 259200, "7d": 604800}
USAGE = """yink — quick shares, secure sends and pair-room terminals (Python 3.11+)

  yink share <paths...> [--note TEXT] [--relay-only]
  yink share --resume <complete-link>
  yink pull <complete-link> [--list] [--note] [--out DIR] [--only N]
  yink term "six words" [--shell PATH]
  yink send [--title-file PATH|- | --title-env NAME] [--text-file PATH|-]
            [--field-file LABEL=PATH]... [--field-env LABEL=NAME]...
            [--expires 1h|1d|3d|7d] [--max-reads N] [--no-recipient-delete]
  yink receive '<recipient-link>' [--delete]
  yink revoke '<management-link>'
  yink install                     (alias: yink --install)

Passphrase: --passphrase-env NAME | --passphrase-file PATH|- | --passphrase-prompt
Global: --origin URL --json --help --rtc-backend auto|relay
State/transfer: --state FILE --timeout MS

Content and passphrases come from stdin, files, environment or a hidden prompt.
Links are bearer secrets: quote them and keep them out of shell history.
Python shares currently use the encrypted relay, subject to its byte budgets.
Terminal hosting requires a POSIX PTY; Windows term is explicitly unavailable.
"""


class CliError(Exception):
    def __init__(self, code, message=None, class_name="local", details=None):
        super().__init__(message or code)
        self.code, self.class_name, self.details = code, class_name, details

    @property
    def exit_code(self):
        return {"usage": 2, "remote": 3, "transfer": 4, "local": 5}[self.class_name]


def usage(code, message):
    return CliError(code, message, "usage")


def json_text(value):
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False)


def js_length(text):
    return len(text.encode("utf-16-le", "surrogatepass")) // 2


def uint(value, limit=MAX_INT):
    return type(value) is int and 0 <= value <= limit


def safe_text(text, keep_lines=False):
    return re.sub(r"[\x00-\x1f\x7f-\x9f\u200b-\u200f\u202a-\u202e\u2066-\u2069\ufeff\u00ad]",
                  lambda m: m[0] if keep_lines and m[0] in "\n\t" else f"\\u{ord(m[0]):04x}", str(text))


def timestamp(ms=None):
    value = dt.datetime.now(dt.timezone.utc) if ms is None else dt.datetime.fromtimestamp(ms/1000, dt.timezone.utc)
    return value.isoformat(timespec="milliseconds").replace("+00:00", "Z")


class Reporter:
    def __init__(self, json_mode=False, stdout=None, stderr=None):
        self.json = json_mode
        self.stdout, self.stderr = stdout or sys.stdout, stderr or sys.stderr
        self.tty = self.stdout.isatty()

    def emit(self, event, fields=None, human=None, exact=False):
        body = {"event": event, "at": timestamp(), **(fields or {})}
        if self.json:
            text = re.sub(r"[\u0080-\u009f\u061c\u200e\u200f\u2028\u2029\u202a-\u202e\u2066-\u2069]",
                          lambda m: f"\\u{ord(m[0]):04x}", json_text(body))
            self.stdout.write(text + "\n")
        elif human is not None:
            self.stdout.write(str(human) if exact else re.sub(r"\r?\n$", "", str(human)) + "\n")
        self.stdout.flush()
        return body

    def info(self, text):
        if not self.json:
            self.stderr.write(text + "\n")
            self.stderr.flush()

    def error(self, error):
        if self.json:
            self.emit("error", {"code": error.code, "class": error.class_name, "message": safe_text(str(error)),
                                **({"details": error.details} if error.details is not None else {})})
        else:
            self.stderr.write(f"error [{error.code}]: {safe_text(str(error))}\n")


PASSPHRASE = {"passphrase-env", "passphrase-file", "passphrase-prompt"}
GLOBAL = {"origin", "json", "help", "rtc-backend"}
ALLOWED = {
    "share": GLOBAL | {"timeout", "state", "note", "relay-only", "resume"},
    "pull": GLOBAL | {"timeout", "state", "out", "only", "list", "note"},
    "term": GLOBAL | {"shell"},
    "send": {"origin", "timeout", "json", "help", "title-file", "title-env", "text-file", "field-file", "field-env",
             "expires", "max-reads", "no-recipient-delete"} | PASSPHRASE,
    "receive": {"origin", "timeout", "json", "help", "delete"} | PASSPHRASE,
    "revoke": {"origin", "timeout", "json", "help"},
}
BOOL = {"json", "help", "relay-only", "list", "no-recipient-delete", "passphrase-prompt", "delete"}
VALUE = set.union(*ALLOWED.values()) - BOOL - {"note"}


def parse_args(argv):
    if argv and argv[0] in ("install", "--install"):
        if any(x not in ("--json", "--help") for x in argv[1:]) or len(set(argv[1:])) != len(argv[1:]):
            raise usage("bad-install-option", "install accepts only --json or --help")
        return {"command": "install", "json": "--json" in argv, "help": "--help" in argv}
    if not argv or argv[0] in ("help", "--help"):
        return {"command": "help", "help": True}
    command = argv[0]
    if command not in ALLOWED:
        raise usage("unknown-command", f"unknown command: {command}")
    opts = {"command": command, "paths": [], "json": False, "help": False}
    i = 1
    while i < len(argv):
        raw = argv[i]
        if not raw.startswith("--"):
            opts["paths"].append(raw)
            i += 1
            continue
        key, sep, value = raw[2:].partition("=")
        boolean = key in BOOL or (key == "note" and command == "pull")
        if not boolean and key not in VALUE and key != "note":
            raise usage("unknown-option", f"unknown option --{key}")
        if boolean:
            if sep:
                raise usage("bad-option", f"--{key} does not take a value")
            value = True
        elif not sep:
            i += 1
            if i == len(argv) or argv[i].startswith("--"):
                raise usage("missing-option-value", f"--{key} needs a value")
            value = argv[i]
        elif not value:
            raise usage("missing-option-value", f"--{key} needs a value")
        if key not in ALLOWED[command]:
            raise usage(f"bad-{command}-option", f"--{key} does not apply to {command}")
        if key in ("field-file", "field-env"):
            label, eq, source = value.partition("=")
            if not eq or not source:
                raise usage("bad-field", f"--{key} needs LABEL={'NAME' if key == 'field-env' else 'PATH'}")
            opts.setdefault("fields", []).append({"kind": "env" if key == "field-env" else "file", "label": label, "source": source})
        else:
            if key in opts and opts[key] is not False:
                raise usage("duplicate-option", f"--{key} was given twice")
            opts[key] = value
        i += 1
    for key in ("timeout", "only"):
        if key in opts:
            try:
                # Number() accepts exponent notation and hexadecimal, like npm.
                raw = opts[key].strip()
                number = int(raw, 0) if re.fullmatch(r"0[xX][0-9a-fA-F]+|0[bB][01]+|0[oO][0-7]+", raw) else float(raw or "0")
                if number != int(number) or number > MAX_INT or number < (1 if key == "timeout" else 0):
                    raise ValueError()
                opts[key] = int(number)
            except (ValueError, OverflowError):
                raise usage("bad-timeout" if key == "timeout" else "bad-item",
                            "--timeout must be a positive integer in milliseconds" if key == "timeout" else "--only must be a non-negative item id")
    if "origin" in opts:
        opts["origin"] = secure_origin(opts["origin"])
    if opts["help"]:
        return opts
    paths = opts["paths"]
    if command == "share":
        if opts.get("resume"):
            if paths:
                raise usage("resume-with-paths", "share --resume cannot also take file paths")
            if "note" in opts:
                raise usage("resume-with-note", "share --resume restores the saved note")
        elif not paths:
            raise usage("missing-path", "share needs at least one path, or --resume <complete-link>")
    if command in ("pull", "receive", "revoke"):
        if len(paths) != 1:
            raise usage("missing-link", f"{command} needs exactly one complete {'share' if command == 'pull' else 'Secure Send'} link")
        opts["link"] = paths[0]
    if command == "term":
        if len(paths) != 1:
            raise usage("missing-code", "term needs one quoted six-word code")
        opts["code"] = paths[0]
    if command in ("send", "receive") and len(PASSPHRASE & opts.keys()) > 1:
        raise usage("duplicate-passphrase", "give at most one of --passphrase-env, --passphrase-file and --passphrase-prompt")
    if command == "send":
        if paths:
            raise usage("send-value-on-command-line", "send never takes values as arguments; use stdin, --text-file, --field-file or --field-env")
        opts.setdefault("fields", [])
        if opts.get("expires", "3d") not in TTLS:
            raise usage("bad-expires", "--expires must be one of 1h, 1d, 3d or 7d")
        opts["ttl"] = TTLS[opts.get("expires", "3d")]
        cap = opts.get("max-reads")
        if cap is not None and (not re.fullmatch(r"[0-9]+", cap) or not 1 <= int(cap) <= 100):
            raise usage("bad-max-reads", "--max-reads must be an integer from 1 to 100")
        opts["maxReads"] = int(cap) if cap is not None else None
        if "title-file" in opts and "title-env" in opts:
            raise usage("duplicate-title", "give at most one of --title-file and --title-env")
        sources = [opts.get(k) for k in ("title-file", "text-file", "passphrase-file")]
        sources += [f["source"] for f in opts["fields"] if f["kind"] == "file"]
        if sources.count("-") > 1:
            raise usage("stdin-reused", "only one option can read from stdin (-)")
    return opts


def secure_origin(raw):
    try:
        url = urllib.parse.urlsplit(raw)
        if url.scheme not in ("http", "https") or not url.hostname or url.username or url.password:
            raise ValueError()
        port = url.port
    except (ValueError, TypeError):
        raise usage("bad-origin", "--origin must be a complete http(s) URL without credentials")
    if url.scheme == "http":
        try:
            loopback = ipaddress.ip_address(url.hostname).is_loopback
        except ValueError:
            loopback = url.hostname.lower() == "localhost"
        if not loopback:
            raise usage("insecure-send-origin", "HTTPS is required; HTTP is allowed only for loopback development origins")
    host = url.hostname.encode("idna").decode().lower()
    if ":" in host:
        host = f"[{host}]"
    suffix = f":{port}" if port and port != (443 if url.scheme == "https" else 80) else ""
    return f"{url.scheme}://{host}{suffix}"


def encode64(value):
    return base64.urlsafe_b64encode(value).decode().rstrip("=")


def decode64(value, length=None):
    if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", value) or len(value) % 4 == 1:
        raise ValueError("Invalid encoded data.")
    result = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
    if (length is not None and len(result) != length) or encode64(result) != value:
        raise ValueError("Invalid encoded data.")
    return result


def crypto_imports():
    try:
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import ec, ed25519
        from cryptography.hazmat.primitives.ciphers.aead import AESGCM
        from cryptography.hazmat.primitives.kdf.hkdf import HKDF
        return hashes, serialization, ec, ed25519, AESGCM, HKDF
    except ImportError:
        raise CliError("dependency-unavailable", "cryptography is required; run with uv run yink.py, or install yink-cli in a dedicated virtual environment")


def hkdf(material, salt, info, length=32):
    hashes, _, _, _, _, HKDF = crypto_imports()
    return HKDF(algorithm=hashes.SHA256(), length=length, salt=salt, info=info).derive(material)


def aes(key):
    return crypto_imports()[4](key)


POLICY_KEYS = ("v", "id", "ttl", "maxReads", "passphrase", "salt", "readHash", "revokeHash", "recipientHash")


def policy(value):
    return {key: value[key] for key in POLICY_KEYS}


def capability_hash(token):
    return hashlib.sha256(decode64(token, 32)).hexdigest()


def validate_envelope(value, ident):
    if not isinstance(value, dict) or set(value) != set(POLICY_KEYS) | {"nonce", "ciphertext"}:
        raise ValueError("Invalid send fields.")
    if type(value["v"]) is not int or value["v"] != 1 or not re.fullmatch(r"[0-9a-f]{32}", value["id"] or "") or value["id"] != ident or type(value["ttl"]) is not int or value["ttl"] not in TTLS.values():
        raise ValueError("Invalid send policy.")
    if value["maxReads"] is not None and (not uint(value["maxReads"], 100) or value["maxReads"] < 1):
        raise ValueError("Invalid retrieval cap.")
    if type(value["passphrase"]) is not bool:
        raise ValueError("Invalid capabilities.")
    for key in ("readHash", "revokeHash", "recipientHash"):
        if key == "recipientHash" and value[key] is None:
            continue
        if not isinstance(value[key], str) or not re.fullmatch(r"[0-9a-f]{64}", value[key]):
            raise ValueError("Invalid capabilities.")
    decode64(value["salt"], 16)
    decode64(value["nonce"], 12)
    cipher = value["ciphertext"]
    if not isinstance(cipher, str) or len(cipher) > (MAX_PLAIN + 16)*4//3 + 1 or not 17 <= len(decode64(cipher)) <= MAX_PLAIN + 16:
        raise ValueError("Send is too large.")
    return {**policy(value), "nonce": value["nonce"], "ciphertext": cipher}


def parse_send_fragment(fragment):
    if len(fragment) > 300:
        raise ValueError("Invalid Secure Send link.")
    pairs = urllib.parse.parse_qsl(fragment.lstrip("#"), keep_blank_values=True)
    p = dict(pairs)
    if len(p) != len(pairs) or p.get("v") != "1" or not re.fullmatch(r"[0-9a-f]{32}", p.get("i", "")):
        raise ValueError("Invalid Secure Send link.")
    if "d" in p:
        if set(p) != {"v", "i", "d"}:
            raise ValueError("Invalid revoke link.")
        decode64(p["d"], 32)
        return {"id": p["i"], "revoke": p["d"]}
    if set(p) != {"v", "i", "k", "r", "p"} or p["p"] not in ("0", "1"):
        raise ValueError("Invalid Secure Send link.")
    decode64(p["k"], 32)
    decode64(p["r"], 32)
    return {"id": p["i"], "secret": p["k"], "read": p["r"], "passphrase": p["p"] == "1"}


def parse_send_link(raw, kind):
    try:
        url = urllib.parse.urlsplit(raw)
        if url.scheme not in ("http", "https") or not url.hostname or url.username or url.password or url.path != "/send" or url.query:
            raise ValueError()
    except ValueError:
        raise usage("bad-send-link", "expected a complete Secure Send link (https://…/send#…)")
    origin = secure_origin(raw)
    try:
        link = parse_send_fragment(url.fragment)
    except (ValueError, TypeError):
        raise usage("bad-send-link", "the Secure Send link is incomplete or damaged")
    if kind == "recipient" and "revoke" in link:
        raise usage("management-link", "that is a sender management link; use `yink revoke` with it")
    if kind == "management" and "revoke" not in link:
        raise usage("recipient-link", "that is a recipient link; `yink revoke` needs the sender management link")
    return origin, link


def send_key(secret, passphrase, envelope):
    salt, material = decode64(envelope["salt"], 16), decode64(secret, 32)
    if envelope["passphrase"]:
        if not isinstance(passphrase, str) or not passphrase or len(passphrase.encode()) > 1024:
            raise ValueError("Enter the passphrase (up to 1,024 UTF-8 bytes).")
        material += hashlib.pbkdf2_hmac("sha256", passphrase.encode(), b"yink-send-pass|v1|" + salt, 600000, 32)
    return hkdf(material, salt, ("yink-send-key|v1|" + envelope["id"]).encode())


def validate_content(content, recipient_hash):
    legacy = isinstance(content, dict) and set(content) == {"title", "fields", "recipientToken"}
    if not isinstance(content, dict) or (not legacy and set(content) != {"title", "text", "fields", "recipientToken"}):
        raise ValueError("Invalid encrypted content.")
    if not isinstance(content["title"], str) or js_length(content["title"]) > 256 or (not legacy and not isinstance(content["text"], str)):
        raise ValueError("Invalid encrypted content.")
    fields = content["fields"]
    if not isinstance(fields, list) or not (1 if legacy else 0) <= len(fields) <= 8:
        raise ValueError("Invalid encrypted content.")
    for field in fields:
        if not isinstance(field, dict) or set(field) != {"label", "value"} or not isinstance(field["label"], str) or js_length(field["label"]) > 128 or not isinstance(field["value"], str):
            raise ValueError("Invalid encrypted field.")
    if recipient_hash is None:
        if content["recipientToken"] is not None:
            raise ValueError("Invalid deletion policy.")
    else:
        decode64(content["recipientToken"], 32)


def create_send(title, text, fields, ttl=259200, max_reads=None, recipient_delete=True, passphrase="", random=os.urandom):
    if type(recipient_delete) is not bool or not isinstance(passphrase, str) or len(passphrase.encode()) > 1024:
        raise ValueError("Invalid options or passphrase too long.")
    if any(f.get("value") == "" for f in fields):
        raise ValueError("Each added field needs a value. Fill it in or remove it.")
    if not any(re.search(r"\S", value) for value in [title, text] + [f["value"] for f in fields]):
        raise ValueError("Add a title, some text or a field before creating a send.")
    ident, secret, read, revoke = random(16).hex(), encode64(random(32)), encode64(random(32)), encode64(random(32))
    recipient = encode64(random(32)) if recipient_delete else None
    envelope = {"v": 1, "id": ident, "ttl": ttl, "maxReads": max_reads, "passphrase": bool(passphrase), "salt": encode64(random(16)),
                "readHash": capability_hash(read), "revokeHash": capability_hash(revoke), "recipientHash": capability_hash(recipient) if recipient else None}
    content = {"title": title, "text": text, "fields": fields, "recipientToken": recipient}
    validate_content(content, envelope["recipientHash"])
    plain = json_text(content).encode()
    if len(plain) > MAX_PLAIN:
        raise ValueError("Keep the title, text, labels and values within 24 KiB of encoded text.")
    nonce = random(12)
    envelope.update(nonce=encode64(nonce), ciphertext=encode64(aes(send_key(secret, passphrase, envelope)).encrypt(nonce, plain, json_text(policy(envelope)).encode())))
    return {"envelope": validate_envelope(envelope, ident), "revoke": revoke,
            "recipientFragment": f"v=1&i={ident}&k={secret}&r={read}&p={int(bool(passphrase))}", "senderFragment": f"v=1&i={ident}&d={revoke}"}


def open_send(envelope, link, passphrase=""):
    value = validate_envelope(envelope, link["id"])
    key = send_key(link["secret"], passphrase, value)
    try:
        plain = aes(key).decrypt(decode64(value["nonce"], 12), decode64(value["ciphertext"]), json_text(policy(value)).encode())
        content = json.loads(plain.decode("utf-8-sig"))
        validate_content(content, value["recipientHash"])
        if content["recipientToken"] and capability_hash(content["recipientToken"]) != value["recipientHash"]:
            raise ValueError()
        return {"title": content["title"], "text": content.get("text", ""), "fields": content["fields"], "recipientToken": content["recipientToken"]}
    except Exception:
        raise ValueError("Could not unlock. Check the passphrase and original link." if value["passphrase"] else "Could not unlock. Check the original link.")


def read_text(source, what):
    try:
        if source == "-":
            data = sys.stdin.buffer.read(4*MAX_PLAIN + 1)
        else:
            with open(source, "rb") as stream:
                data = stream.read(4*MAX_PLAIN + 1)
    except OSError as error:
        raise CliError("read-failed", f"cannot read {what}: {error.strerror}")
    if len(data) > 4*MAX_PLAIN:
        raise usage("send-too-large", f"{what} is larger than a Secure Send can hold")
    try:
        return data.decode("utf-8-sig")
    except UnicodeError:
        raise usage("not-utf8", f"{what} is not UTF-8 text; Secure Send carries text only")


def env_text(name):
    if name not in os.environ:
        raise usage("missing-env", f"environment variable {safe_text(name)} is not set")
    return os.environ[name]


def check_passphrase(value):
    if not value:
        raise usage("empty-passphrase", "the passphrase is empty")
    if "\r" in value or "\n" in value:
        raise usage("bad-passphrase", "a passphrase cannot contain a line break")
    if len(value.encode()) > 1024:
        raise usage("passphrase-too-long", "a passphrase can be at most 1,024 UTF-8 bytes")
    return value


def prompt_hidden(question):
    if not sys.stdin.isatty():
        raise usage("passphrase-prompt-unavailable", "a passphrase prompt needs an interactive terminal; use --passphrase-env or --passphrase-file")
    # Never let getpass fall back to an echoed prompt.
    import warnings
    with warnings.catch_warnings():
        warnings.simplefilter("error", getpass.GetPassWarning)
        try:
            return getpass.getpass(question, stream=sys.stderr)
        except getpass.GetPassWarning:
            raise usage("passphrase-prompt-unavailable", "a hidden passphrase prompt is unavailable")
        except KeyboardInterrupt:
            raise CliError("cancelled", "cancelled at the passphrase prompt")


def passphrase_from(args, confirm=False):
    if "passphrase-env" in args:
        return check_passphrase(env_text(args["passphrase-env"]))
    if "passphrase-file" in args:
        return check_passphrase(re.sub(r"\r?\n$", "", read_text(args["passphrase-file"], "the passphrase file")))
    if args.get("passphrase-prompt"):
        value = check_passphrase(prompt_hidden("Passphrase: "))
        if confirm and prompt_hidden("Repeat passphrase: ") != value:
            raise usage("passphrase-mismatch", "the two passphrases differ")
        return value
    return None


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


class SendNetworkError(Exception):
    def __init__(self, message="Send unavailable. Try again.", status=None):
        super().__init__(message)
        self.status = status


def send_request(origin, ident, method, token, body=None, action="", timeout=30000):
    # No cookies, redirects, query capabilities, or response-body diagnostics.
    if not re.fullmatch(r"[0-9a-f]{32}", ident) or action not in ("", "revoke", "recipient"):
        raise ValueError("Invalid request.")
    request = urllib.request.Request(f"{secure_origin(origin)}/api/send/{ident}" + (f"/{action}" if action else ""),
                                     data=json_text(body).encode() if body is not None else None,
                                     headers={"Authorization": f"Bearer {token}", "Cache-Control": "no-store",
                                              **({"Content-Type": "application/json"} if body is not None else {})}, method=method)
    try:
        response = urllib.request.build_opener(NoRedirect).open(request, timeout=timeout/1000)
    except urllib.error.HTTPError as error:
        response = error
    except (OSError, urllib.error.URLError):
        raise SendNetworkError()
    with response:
        status_code = response.code
        try:
            data = response.read(64*1024 + 1)
            value = json.loads(data) if len(data) <= 64*1024 else None
        except (ValueError, OSError):
            value = None
    if not 200 <= status_code < 300 or not isinstance(value, dict):
        message = value.get("error") if isinstance(value, dict) else None
        raise SendNetworkError(message if not 200 <= status_code < 300 and isinstance(message, str) else "Send unavailable. Try again.", status_code)
    return value


def remote_error(error, overrides=None):
    codes = {400: "send-rejected", 403: "send-forbidden", 404: "send-unavailable", 405: "send-rejected", 409: "send-conflict",
             410: "send-exhausted", 429: "send-busy", 503: "send-full", **(overrides or {})}
    return CliError(codes.get(error.status, "send-network"), safe_text(str(error))[:200], "remote",
                    {"status": error.status} if error.status else None)


def render_content(content, tty):
    value = lambda text: safe_text(text, True) if tty else text
    has_text = bool(re.search(r"\S", content.get("text", "")))
    items = ([{"label": "text", "value": content["text"]}] if has_text else []) + [
        {"label": f["label"] or f"field {i+1}", "bare": not f["label"], "value": f["value"]} for i, f in enumerate(content["fields"])]
    items = [f for f in items if not (f.get("bare") and f["value"] == "")]
    if not content["title"] and len(items) == 1 and (items[0].get("bare") or has_text):
        return value(items[0]["value"])
    out = ["title: " + safe_text(content["title"])] if content["title"] else []
    for item in items:
        out.append(safe_text(item["label"]) + (":\n" if "\n" in item["value"] else ": ") + value(item["value"]))
    return "\n".join(out)


def policy_text(p):
    return " ".join([f"Expires {timestamp(p['expiresAt'])}.", "No retrieval cap." if p["maxReads"] is None else f"Maximum {p['maxReads']} ciphertext retrieval{'s' if p['maxReads'] != 1 else ''}.",
                     *([f"This was retrieval {p['reads']}" + (f" of {p['maxReads']}" if p["maxReads"] is not None else "") + "."] if "reads" in p else []),
                     f"Recipient deletion {'enabled' if p['recipientDelete'] else 'disabled'}.",
                     "Passphrase required; deliver it separately from the link." if p["passphrase"] else "No passphrase."])


def run_send(args, reporter):
    origin = args.get("origin", DEFAULT_ORIGIN)
    title = env_text(args["title-env"]) if "title-env" in args else re.sub(r"\r?\n$", "", read_text(args["title-file"], "the title file")) if "title-file" in args else ""
    if js_length(title) > 256:
        raise usage("title-too-long", "the title can be at most 256 characters")
    if re.search(r"[\x00-\x1f\x7f]", title):
        raise usage("bad-title", "the title cannot contain line breaks or control characters")
    fields = []
    for spec in args["fields"]:
        label = spec["label"]
        if js_length(label) > 128:
            raise usage("label-too-long", "a field label can be at most 128 characters")
        if re.search(r"[\x00-\x1f\x7f]", label):
            raise usage("bad-label", "a field label cannot contain line breaks or control characters")
        value = env_text(spec["source"]) if spec["kind"] == "env" else read_text(spec["source"], f'the file for field "{safe_text(label)}"')
        if value == "":
            raise usage("empty-field", f'field "{safe_text(label)}" is empty; every field needs a value')
        fields.append({"label": label, "value": value})
    text = ""
    if "text-file" in args:
        text = read_text(args["text-file"], "the text file")
    elif "title-env" not in args and "title-file" not in args and not fields and args.get("passphrase-file") != "-" and not sys.stdin.isatty():
        text = read_text("-", "stdin")
    if not title and not text and not fields:
        raise usage("empty-send", "nothing to send: pipe text in, or use --text-file, --title-file/--title-env, --field-file or --field-env")
    password = passphrase_from(args, True) or ""
    if len(fields) > 8:
        raise usage("too-many-fields", "a send holds at most 8 fields")
    recipient_delete = not args.get("no-recipient-delete", False)
    try:
        prepared = create_send(title, text, fields, args["ttl"], args["maxReads"], recipient_delete, password)
    except ValueError as error:
        raise usage("send-too-large" if "24 KiB" in str(error) else "send-invalid", str(error))
    check = open_send(prepared["envelope"], parse_send_fragment(prepared["recipientFragment"]), password)
    if {key: check[key] for key in ("title", "text", "fields")} != {"title": title, "text": text, "fields": fields}:
        raise CliError("send-self-check", "the encrypted send did not round-trip; nothing was uploaded")
    management = f"{origin}/send#{prepared['senderFragment']}"
    for attempt in range(1, 4):
        try:
            result = send_request(origin, prepared["envelope"]["id"], "PUT", prepared["revoke"], prepared["envelope"], timeout=args.get("timeout", 30000))
            break
        except SendNetworkError as error:
            if error.status in (400, 403, 405, 429, 503):
                raise remote_error(error)
            if error.status != 409 and attempt < 3:
                time.sleep(attempt)
                continue
            raise CliError("send-uncertain", f"upload outcome unknown ({safe_text(str(error))}); the send may have been stored. To make sure it is gone, run: yink revoke {management}", "remote",
                           {"id": prepared["envelope"]["id"], "managementLink": management, **({"status": error.status} if error.status else {})})
    if type(result.get("expiresAt")) not in (int, float):
        raise CliError("send-uncertain", f"the server's reply was malformed; the send may have been stored. To make sure it is gone, run: yink revoke {management}", "remote", {"managementLink": management})
    link = f"{origin}/send#{prepared['recipientFragment']}"
    p = {"expiresAt": result["expiresAt"], "maxReads": args["maxReads"], "passphrase": bool(password), "recipientDelete": recipient_delete}
    reporter.emit("created", {"id": prepared["envelope"]["id"], "link": link, "managementLink": management, "ttl": args["ttl"], **p}, link)
    reporter.info(f"Management link (private; revokes this send): {management}")
    reporter.info(policy_text(p))


def run_receive(args, reporter):
    origin, link = parse_send_link(args["link"], "recipient")
    origin = args.get("origin", origin)
    interactive = sys.stdin.isatty() and "passphrase-env" not in args and "passphrase-file" not in args
    password = passphrase_from(args)
    if not link["passphrase"] and password is not None:
        raise usage("passphrase-unexpected", "this link says the send has no passphrase")
    if link["passphrase"] and password is None:
        if not interactive:
            raise usage("passphrase-required", "this send needs a passphrase: use --passphrase-env, --passphrase-file or --passphrase-prompt")
        password = check_passphrase(prompt_hidden("Passphrase: "))
    try:
        delivery = send_request(origin, link["id"], "GET", link["read"], timeout=args.get("timeout", 30000))
    except SendNetworkError as error:
        if error.status in (404, 410):
            raise CliError("send-unavailable" if error.status == 404 else "send-exhausted",
                           "send unavailable: expired, deleted, or unknown" if error.status == 404 else "the retrieval cap has been reached; this send can no longer be opened", "remote", {"status": error.status})
        raise remote_error(error)
    if not isinstance(delivery.get("envelope"), dict) or type(delivery.get("expiresAt")) not in (int, float) or not uint(delivery.get("reads")):
        raise CliError("send-network", "the server's reply was malformed", "remote")
    envelope = delivery["envelope"]
    for attempt in range(1, 4):
        try:
            if envelope.get("passphrase") and not password:
                if not interactive:
                    raise usage("passphrase-required", "this send needs a passphrase (its link said otherwise); this retrieval was counted")
                password = check_passphrase(prompt_hidden("Passphrase: "))
            content = open_send(envelope, link, password if envelope.get("passphrase") else "")
            break
        except CliError:
            raise
        except (ValueError, TypeError, KeyError) as error:
            if interactive and envelope.get("passphrase") and attempt < 3:
                reporter.stderr.write("Could not unlock. Check the passphrase.\n")
                password = check_passphrase(prompt_hidden("Passphrase: "))
                continue
            raise CliError("unlock-failed", f"{safe_text(str(error))} This retrieval was counted.", "transfer")
    p = {"expiresAt": delivery["expiresAt"], "maxReads": envelope["maxReads"], "reads": delivery["reads"], "passphrase": envelope["passphrase"], "recipientDelete": bool(content["recipientToken"])}
    reporter.emit("received", {"id": link["id"], "title": content["title"], "text": content["text"], "fields": content["fields"], **p}, render_content(content, reporter.tty), exact=not reporter.tty)
    reporter.info(policy_text(p))
    if p["maxReads"] is not None and p["reads"] >= p["maxReads"]:
        reporter.info("That was the final allowed retrieval; the stored ciphertext is gone.")
    if args.get("delete"):
        if not content["recipientToken"]:
            raise CliError("delete-disabled", "the sender disabled recipient deletion; only the sender can revoke this send", "remote")
        try:
            send_request(origin, link["id"], "DELETE", content["recipientToken"], action="recipient", timeout=args.get("timeout", 30000))
        except SendNetworkError as error:
            raise remote_error(error, {404: "delete-unavailable"})
        reporter.emit("deleted", {"id": link["id"], "by": "recipient"})
        reporter.info("Send deleted for everyone. Future retrievals are unavailable; copies already saved cannot be erased.")


def run_revoke(args, reporter):
    origin, link = parse_send_link(args["link"], "management")
    try:
        send_request(args.get("origin", origin), link["id"], "DELETE", link["revoke"], action="revoke", timeout=args.get("timeout", 30000))
    except SendNetworkError as error:
        if error.status == 404:
            raise CliError("send-unavailable", "send unavailable: already deleted, expired, or this revoke link is invalid", "remote", {"status": 404})
        raise remote_error(error)
    reporter.emit("revoked", {"id": link["id"]}, "Send revoked. Future retrievals are unavailable; copies already saved cannot be erased.")


# Restricted canonical CBOR, identical to public/js/chat/cbor.js. No general
# CBOR dependency; no floats, tags, duplicate/prototype keys or indefinite sizes.
def cbor_head(major, number):
    prefix = major << 5
    if number < 24:
        return bytes([prefix | number])
    for code, length in ((24, 1), (25, 2), (26, 4), (27, 8)):
        if number < 1 << (length * 8):
            return bytes([prefix | code]) + number.to_bytes(length, "big")
    raise ValueError("cbor: integer out of range")


def cbor_encode(value, depth=0):
    if depth > 8:
        raise ValueError("cbor: nesting too deep")
    if value is None:
        return b"\xf6"
    if type(value) is bool:
        return b"\xf5" if value else b"\xf4"
    if type(value) is int and -2**53 <= value <= MAX_INT:
        return cbor_head(0, value) if value >= 0 else cbor_head(1, -value-1)
    if isinstance(value, bytes):
        return cbor_head(2, len(value)) + value
    if isinstance(value, str):
        data = value.encode()
        return cbor_head(3, len(data)) + data
    if isinstance(value, list):
        return cbor_head(4, len(value)) + b"".join(cbor_encode(x, depth+1) for x in value)
    if isinstance(value, dict):
        if any(not isinstance(k, str) or k == "__proto__" for k in value):
            raise ValueError("cbor: forbidden map key")
        entries = sorted((cbor_encode(k, depth+1), cbor_encode(v, depth+1)) for k, v in value.items())
        return cbor_head(5, len(entries)) + b"".join(k+v for k, v in entries)
    raise ValueError("cbor: unsupported value")


def cbor_decode(data):
    offset = 0

    def take(length):
        nonlocal offset
        if length > len(data)-offset:
            raise ValueError("cbor: truncated")
        result = data[offset:offset+length]
        offset += length
        return result

    def decode(depth):
        if depth > 8:
            raise ValueError("cbor: nesting too deep")
        first = take(1)[0]
        major, ai = first >> 5, first & 31
        if major == 7:
            if ai not in (20, 21, 22):
                raise ValueError("cbor: forbidden simple value")
            return {20: False, 21: True, 22: None}[ai]
        if ai < 24:
            arg = ai
        elif ai in (24, 25, 26, 27):
            length = 1 << (ai - 24)
            arg = int.from_bytes(take(length), "big")
            if arg < (24 if length == 1 else 1 << (8*(length//2))):
                raise ValueError("cbor: non-shortest length")
        else:
            raise ValueError("cbor: forbidden additional info")
        if major in (0, 1):
            if arg > MAX_INT:
                raise ValueError("cbor: integer out of range")
            return arg if major == 0 else -arg-1
        if arg > 0xffffffff:
            raise ValueError("cbor: length too large")
        if major == 2:
            return take(arg)
        if major == 3:
            return take(arg).decode("utf-8-sig")
        if major == 4:
            if arg > len(data)-offset:
                raise ValueError("cbor: truncated array")
            return [decode(depth+1) for _ in range(arg)]
        if major == 5:
            if arg > (len(data)-offset)//2:
                raise ValueError("cbor: truncated map")
            result, previous = {}, None
            for _ in range(arg):
                start = offset
                key = decode(depth+1)
                encoded = data[start:offset]
                if not isinstance(key, str) or key == "__proto__" or (previous is not None and encoded <= previous):
                    raise ValueError("cbor: invalid map key/order")
                previous = encoded
                result[key] = decode(depth+1)
            return result
        raise ValueError("cbor: unsupported major type")

    result = decode(0)
    if offset != len(data):
        raise ValueError("cbor: trailing bytes")
    return result


def derive_share(secret):
    if len(secret) != 32:
        raise ValueError("share crypto: S must be 32 bytes")
    text = secret.hex().encode()
    return {"secret": secret, "id": hashlib.sha256(b"yink-share-room|" + text).digest()[:16],
            "auth": hashlib.sha256(b"yink-share-auth|" + text).hexdigest(), "key": hkdf(secret, b"", b"yink-share-content|v1")}


def message_seal(room, ad, body, salt=None):
    salt = os.urandom(32) if salt is None else salt
    if len(salt) != 32 or len(ad) != 4:
        raise ValueError("share crypto: bad message inputs")
    key = hkdf(room["key"], salt, b"yink-share-msg|v1|" + room["id"].hex().encode())
    cipher = salt + aes(key).encrypt(bytes(12), cbor_encode(body), ad + room["id"])
    if len(cipher) > 240*1024:
        raise ValueError("share crypto: message too large")
    return cipher


def message_open(room, ad, cipher):
    if not 32 < len(cipher) <= 240*1024:
        raise ValueError("share crypto: bad message ciphertext")
    key = hkdf(room["key"], cipher[:32], b"yink-share-msg|v1|" + room["id"].hex().encode())
    return cbor_decode(aes(key).decrypt(bytes(12), cipher[32:], ad + room["id"]))


def xfer_key(room, n, rid):
    if not uint(n, 0xffffffff) or len(rid) != 8:
        raise ValueError("share crypto: bad transfer key inputs")
    return hkdf(room["key"], b"", b"yink-share-xfer|v1|" + room["id"] + struct.pack(">I", n) + rid)


def chunk_seal(key, room, ad, off, size, plain):
    return aes(key).encrypt(bytes(4) + struct.pack(">Q", off), plain, ad + room["id"] + struct.pack(">Q", size))


def chunk_open(key, room, ad, off, size, cipher):
    return aes(key).decrypt(bytes(4) + struct.pack(">Q", off), cipher, ad + room["id"] + struct.pack(">Q", size))


def frame_encode(src, dst, ptype, cipher=b"", flags=2):
    data = bytes([1, flags, src, dst, ptype, 0, 0, 0]) + cipher
    frame_decode(data)
    return data


def frame_decode(data):
    if not isinstance(data, bytes) or not 8 <= len(data) <= 240*1024+8:
        raise ValueError("share wire: bad frame size")
    version, flags, src, dst, ptype, a, b, c = data[:8]
    if version != 1 or flags & 0xf8 or a or b or c or ptype not in (1, 2) or src > 32:
        raise ValueError("share wire: bad frame")
    if bool(flags & 1) != (dst == 255) or (dst == 255 and src != 0) or (src == 0 and dst != 255 and not 1 <= dst <= 32) or (src > 0 and dst != 0):
        raise ValueError("share wire: bad direction")
    return {"flags": flags, "src": src, "dst": dst, "ptype": ptype, "ad": bytes([version, flags, dst, ptype]), "cipher": data[8:]}


SHARE_SHAPES = {
    "share/hello": ("r", {"v", "caps"}, {"mv"}),
    "share/manifest": ("h", {"v", "mv", "note", "files", "sig"}, set()),
    "share/get": ("r", {"n", "rid", "from", "to"}, set()),
    "share/queued": ("h", {"rid", "pos"}, set()),
    "share/ack": ("r", {"rid", "off"}, set()),
    "share/end": ("h", {"rid", "n", "size"}, {"hash"}),
    "share/no": ("h", {"reason"}, {"rid", "n"}),
    "share/cancel": ("r", {"rid", "reason"}, set()),
    "share/bye": ("h", {"reason"}, set()),
    "rtc/offer": ("r", {"epoch", "sdp"}, set()),
    "rtc/answer": ("h", {"epoch", "sdp"}, set()),
    "rtc/ice": ("b", {"epoch", "cand"}, set()),
    "rtc/no": ("h", {"epoch", "reason"}, set()),
    "rtc/ping": ("r", {"id"}, set()),
    "rtc/pong": ("h", {"id"}, set()),
}


def validate_share_message(message, src, am_host):
    if not isinstance(message, dict) or message.get("k") not in SHARE_SHAPES:
        raise ValueError("share proto: unknown-kind")
    direction, required, optional = SHARE_SHAPES[message["k"]]
    keys = set(message) - {"k"}
    if not required <= keys or keys - required - optional:
        raise ValueError("share proto: shape")
    if (direction == "r" and (src == 0 or not am_host)) or (direction == "h" and (src != 0 or am_host)) or (direction == "b" and ((src == 0) == am_host)):
        raise ValueError("share proto: direction")
    for key, value in message.items():
        if key in ("n", "mv", "from", "to", "pos", "off", "size", "epoch") and not uint(value):
            raise ValueError("share proto: bad-" + key)
        if key == "v" and (type(value) is not int or value != 1):
            raise ValueError("share proto: bad-v")
        if key in ("k", "note", "reason", "sdp", "cand") and not isinstance(value, str):
            raise ValueError("share proto: bad-" + key)
        if key == "rid" and (not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{11}", value)):
            raise ValueError("share proto: bad-rid")
        if key == "id" and (not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_-]{22}", value)):
            raise ValueError("share proto: bad-id")
        if key == "caps" and (not isinstance(value, list) or any(not isinstance(x, str) for x in value)):
            raise ValueError("share proto: bad-caps")
        if key == "files" and not isinstance(value, list):
            raise ValueError("share proto: bad-files")
        if key == "sig" and (not isinstance(value, bytes) or len(value) != 64):
            raise ValueError("share proto: bad-sig")
        if key == "hash" and value is not None and (not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value)):
            raise ValueError("share proto: bad-hash")
    if message["k"] == "share/no" and message["reason"] not in ("revoked", "gone", "changed-on-disk", "over-budget", "rate", "unknown-item", "host-busy", "too-large"):
        raise ValueError("share proto: bad-reason")
    if message["k"] == "rtc/no" and message["reason"] not in ("relay-only", "busy", "unsupported"):
        raise ValueError("share proto: bad-reason")
    return message


def parse_share_link(raw):
    try:
        url = urllib.parse.urlsplit(raw)
        origin = secure_origin(raw)
        pairs = urllib.parse.parse_qsl(url.fragment, keep_blank_values=True)
        p = dict(pairs)
        if len(p) != len(pairs) or p.get("k") != "s" or url.path != "/s" or url.query:
            raise ValueError()
        secret, host_public = decode64(p.get("s"), 32), decode64(p.get("h"), 32)
        return origin, derive_share(secret), host_public
    except CliError:
        raise
    except (ValueError, TypeError):
        raise usage("bad-share-link", "share link is missing a valid secret or host key")


def validate_manifest(m):
    if not isinstance(m, dict) or set(m) != {"v", "mv", "note", "files"} or type(m["v"]) is not int or m["v"] != 1 or not uint(m["mv"]) or m["mv"] < 1:
        raise ValueError("manifest: shape/version")
    if not isinstance(m["note"], str) or len(m["note"].encode()) > 16*1024 or not isinstance(m["files"], list) or len(m["files"]) > 256:
        raise ValueError("manifest: note/files")
    ids = set()
    for item in m["files"]:
        if not isinstance(item, dict) or set(item) != {"n", "name", "size", "type", "hash"}:
            raise ValueError("manifest: item-shape")
        if not uint(item["n"], 0xffffffff) or item["n"] in ids or not uint(item["size"], 2**40):
            raise ValueError("manifest: item-id/size")
        ids.add(item["n"])
        if not isinstance(item["name"], str) or not item["name"] or len(item["name"].encode()) > 4096 or not isinstance(item["type"], str) or js_length(item["type"]) > 1024:
            raise ValueError("manifest: name/type")
        if item["hash"] is not None and (not isinstance(item["hash"], str) or not re.fullmatch(r"[0-9a-f]{64}", item["hash"])):
            raise ValueError("manifest: hash")
    if len(cbor_encode(m)) > 240*1024 - 1024:
        raise ValueError("manifest: too large — remove files or shorten names")
    return m


def manifest_signable(m, room):
    return b"yink-share-manifest|v1|" + room["id"] + cbor_encode(m)


def sign_manifest(m, seed, room):
    validate_manifest(m)
    ed25519 = crypto_imports()[3]
    return {**m, "sig": ed25519.Ed25519PrivateKey.from_private_bytes(seed).sign(manifest_signable(m, room))}


def adopt_manifest(message, current, host_public, room):
    try:
        m = validate_manifest({key: message[key] for key in ("v", "mv", "note", "files")})
        crypto_imports()[3].Ed25519PublicKey.from_public_bytes(host_public).verify(message["sig"], manifest_signable(m, room))
    except Exception:
        raise CliError("manifest-invalid", "manifest refused: invalid shape or signature", "transfer")
    if current:
        if m["mv"] < current["mv"]:
            return current
        if m["mv"] == current["mv"] and m != current:
            raise CliError("manifest-invalid", "manifest refused: revision-conflict", "transfer")
    return m


def lstat_or_none(path):
    try:
        return os.lstat(path)
    except FileNotFoundError:
        return None


def same_file(a, b):
    return a is not None and b is not None and stat.S_ISREG(a.st_mode) and stat.S_ISREG(b.st_mode) and (a.st_dev, a.st_ino) == (b.st_dev, b.st_ino)


def open_regular(path, flags, mode=0o600):
    before = lstat_or_none(path)
    if before is not None and not stat.S_ISREG(before.st_mode):
        raise CliError("unsafe-local-file", f"refusing non-regular file: {path}")
    fd = os.open(path, flags | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0), mode)
    after = os.fstat(fd)
    if not stat.S_ISREG(after.st_mode) or not same_file(after, os.lstat(path)) or (before is not None and not same_file(before, after)):
        os.close(fd)
        raise CliError("local-file-race", f"file changed while opening: {path}")
    return fd


class FileLock:
    """OS-owned lock: death releases it. The lock inode is never unlinked."""
    def __init__(self, path, timeout=10):
        self.path, self.timeout, self.fd = Path(path), timeout, None

    def __enter__(self):
        self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        self.fd = open_regular(self.path, os.O_RDWR | os.O_CREAT)
        if os.fstat(self.fd).st_size == 0:
            os.write(self.fd, b"0")
        deadline = time.monotonic() + self.timeout
        while True:
            try:
                if os.name == "nt":
                    import msvcrt
                    os.lseek(self.fd, 0, 0)
                    msvcrt.locking(self.fd, msvcrt.LK_NBLCK, 1)
                else:
                    import fcntl
                    fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                if not same_file(os.fstat(self.fd), os.lstat(self.path)):
                    os.close(self.fd)
                    self.fd = None
                    raise CliError("state-lock-failed", "lock file identity changed")
                return self
            except OSError:
                if time.monotonic() >= deadline:
                    os.close(self.fd)
                    self.fd = None
                    raise CliError("state-lock-failed", f"could not lock: {self.path}")
                time.sleep(.02)

    def __exit__(self, *args):
        if self.fd is not None:
            os.close(self.fd)
            self.fd = None


def restrict_private(path):
    if os.name == "nt":
        import subprocess
        account = os.environ.get("USERNAME")
        if not account:
            raise CliError("state-write-failed", "Windows account identity is unavailable")
        if os.environ.get("USERDOMAIN"):
            account = os.environ["USERDOMAIN"] + "\\" + account
        result = subprocess.run(["icacls.exe", str(path), "/inheritance:r", "/grant:r", account + ":F"],
                                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW)
        if result.returncode:
            raise CliError("state-write-failed", "could not restrict private state permissions")
    else:
        os.chmod(path, 0o600)


def state_path(args):
    selected = args.get("state") or os.environ.get("YINK_PYTHON_STATE") or str(Path.home()/".yink"/"shares-python.json")
    if not args.get("state") and "YINK_PYTHON_STATE" in os.environ and not Path(selected).is_absolute():
        raise CliError("bad-state-path", "YINK_PYTHON_STATE must be a complete absolute file path")
    return Path(os.path.abspath(selected))


def load_state(path):
    if lstat_or_none(path) is None:
        return {"v": 1, "hosts": {}, "pulls": {}}
    try:
        fd = open_regular(path, os.O_RDONLY)
        with os.fdopen(fd, "rb") as stream:
            data = stream.read(4*1024*1024 + 1)
        value = json.loads(data) if len(data) <= 4*1024*1024 else None
        if not isinstance(value, dict) or set(value) != {"v", "hosts", "pulls"} or value["v"] != 1 or not isinstance(value["hosts"], dict) or not isinstance(value["pulls"], dict):
            raise ValueError()
        return value
    except (OSError, ValueError):
        raise CliError("state-invalid", f"state file is invalid: {path}")


def update_state(path, section, key, record):
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    # A distinct suffix and state format avoid mixing Python and npm locking.
    with FileLock(str(path) + ".python-lock"):
        value = load_state(path)
        if record is None:
            value[section].pop(key, None)
        else:
            value[section][key] = record
        temp = path.with_name(f".{path.name}.{os.getpid()}.{os.urandom(8).hex()}.tmp")
        try:
            with os.fdopen(open_regular(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL), "wb") as stream:
                stream.write(json_text(value).encode() + b"\n")
                stream.flush()
                os.fsync(stream.fileno())
            restrict_private(temp)
            if lstat_or_none(path) is not None and not stat.S_ISREG(os.lstat(path).st_mode):
                raise CliError("state-write-failed", "state path is no longer a regular file")
            os.replace(temp, path)
        except Exception as error:
            with contextlib.suppress(FileNotFoundError):
                temp.unlink()
            if isinstance(error, CliError):
                raise
            raise CliError("state-write-failed", f"could not save state: {path}") from error


class Source:
    def __init__(self, path, expected=None):
        self.path = str(Path(path).absolute())
        self.fd = None
        try:
            # Preserve the npm source policy: regular target, symlinks allowed.
            self.fd = os.open(self.path, os.O_RDONLY | getattr(os, "O_BINARY", 0))
            st = os.fstat(self.fd)
            if not stat.S_ISREG(st.st_mode):
                raise usage("source-not-file", f"source is not a regular file: {self.path}")
            self.identity = self.stat_identity(st)
            if expected is not None and self.identity != expected:
                raise CliError("resume-source-changed", f"saved source changed: {self.path}")
            self.name, self.size, self.type = Path(self.path).name, st.st_size, ""
            self.hash = self.hash_all() if self.size <= 16*1024*1024 else None
            self.check()
        except Exception as error:
            self.close()
            if isinstance(error, CliError):
                raise
            raise CliError("source-open-failed", f"cannot open source: {self.path}") from error

    @staticmethod
    def stat_identity(st):
        return {"size": st.st_size, "mtimeNs": str(st.st_mtime_ns), "dev": str(st.st_dev), "ino": str(st.st_ino)}

    def check(self):
        if self.identity != self.stat_identity(os.fstat(self.fd)):
            raise CliError("changed-on-disk", "source changed on disk", "transfer")

    def read(self, start, end):
        self.check()
        if not 0 <= start <= end <= self.size:
            raise CliError("changed-on-disk", "source range changed", "transfer")
        os.lseek(self.fd, start, 0)
        data = bytearray()
        while len(data) < end-start:
            piece = os.read(self.fd, end-start-len(data))
            if not piece:
                raise CliError("changed-on-disk", "source ended early", "transfer")
            data.extend(piece)
        self.check()
        return bytes(data)

    def hash_all(self):
        digest = hashlib.sha256()
        for start in range(0, self.size, 256*1024):
            digest.update(self.read(start, min(self.size, start+256*1024)))
        return digest.hexdigest()

    def record(self, n):
        return {"n": n, "path": self.path, "name": self.name, "type": self.type, "hash": self.hash, "identity": self.identity}

    def close(self):
        if self.fd is not None:
            os.close(self.fd)
            self.fd = None


def safe_filename(name):
    if not isinstance(name, str) or not name or name in (".", "..") or re.search(r'[\\/\x00-\x1f\x7f:*?"<>|]', name) or name.endswith((".", " ")) or re.match(r"^(con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?$", name, re.I) or len(name.encode()) > 240:
        raise CliError("unsafe-file-name", "refusing unsafe file name: " + json_text(name), "transfer")
    return name


def part_identity(room, host_public, item):
    return {"shareIdHex": room["id"].hex(), "hostPublic": host_public.hex(), "n": item["n"], "name": safe_filename(item["name"]), "size": item["size"], "hash": item["hash"]}


def part_key(identity):
    return hashlib.sha256(b"yink-share-part-key|v1|" + json_text(identity).encode()).hexdigest()


def part_mac(secret, identity, received, prefix_hash):
    return hmac.new(secret, b"yink-share-part|v2|" + json_text(identity).encode() + f"|{received}|{prefix_hash}".encode(), "sha256").hexdigest()


class Sink:
    """npm v2 authenticated checkpoints, streaming prefix hash, no-overwrite publish."""
    RETAIN = {"host-away", "host-left", "remote-unavailable", "connection-closed", "gone", "remote-timeout"}

    def __init__(self, directory, room, host_public, item):
        self.identity = part_identity(room, host_public, item)
        self.secret, self.key = room["secret"], part_key(self.identity)
        self.directory = Path(directory).absolute()
        self.directory.mkdir(parents=True, exist_ok=True)
        self.final = self.directory/self.identity["name"]
        self.part = self.directory/f".yink-{self.key}.part"
        self.meta = self.directory/f".yink-{self.key}.json"
        self.lock = FileLock(self.directory/f".yink-{self.key}.python-lock", timeout=.1)
        self.lock.__enter__()
        self.fd = self.meta_fd = None
        self.part_stat = self.meta_stat = None
        self.offset = self.checkpoint_offset = 0
        self.digest = hashlib.sha256()
        self.closed = False
        part_exists = meta_exists = None
        try:
            if lstat_or_none(self.final):
                raise CliError("output-collision", f"output already exists: {self.final}")
            part_exists, meta_exists = lstat_or_none(self.part), lstat_or_none(self.meta)
            if part_exists or meta_exists:
                if not part_exists or not meta_exists or not stat.S_ISREG(part_exists.st_mode) or not stat.S_ISREG(meta_exists.st_mode):
                    raise CliError("partial-collision", f"unusable partial exists for {self.identity['name']}")
                self.meta_fd = open_regular(self.meta, os.O_RDWR)
                self.meta_stat = os.fstat(self.meta_fd)
                latest, last_complete = None, 0
                with os.fdopen(os.dup(self.meta_fd), "rb") as stream:
                    while True:
                        line = stream.readline(8193)
                        if not line:
                            break
                        if len(line) > 8192:
                            raise CliError("partial-metadata-invalid", "partial checkpoint line too large", "transfer")
                        if not line.endswith(b"\n"):
                            break
                        last_complete = stream.tell()
                        try:
                            m = json.loads(line)
                            if set(m) != {"v", "identity", "received", "prefixHash", "mac"} or m["v"] != 2 or m["identity"] != self.identity or not uint(m["received"], self.identity["size"]) or not re.fullmatch(r"[0-9a-f]{64}", m["prefixHash"]):
                                continue
                            expected = part_mac(self.secret, self.identity, m["received"], m["prefixHash"])
                            if isinstance(m["mac"], str) and hmac.compare_digest(expected, m["mac"]) and (latest is None or m["received"] >= latest["received"]):
                                latest = m
                        except (ValueError, KeyError, TypeError):
                            continue
                if latest is None or part_exists.st_size > self.identity["size"] or latest["received"] > part_exists.st_size:
                    raise CliError("partial-identity-mismatch", f"partial does not match {self.identity['name']}", "transfer")
                self.fd = open_regular(self.part, os.O_RDWR)
                self.part_stat = os.fstat(self.fd)
                left = latest["received"]
                while left:
                    data = os.read(self.fd, min(left, 256*1024))
                    if not data:
                        raise CliError("partial-prefix-mismatch", "partial prefix ended early", "transfer")
                    self.digest.update(data)
                    left -= len(data)
                if self.digest.hexdigest() != latest["prefixHash"]:
                    raise CliError("partial-prefix-mismatch", "partial prefix integrity failed", "transfer")
                self.offset = self.checkpoint_offset = latest["received"]
                os.ftruncate(self.fd, self.offset)
                os.lseek(self.fd, self.offset, 0)
                os.ftruncate(self.meta_fd, last_complete)
                os.lseek(self.meta_fd, last_complete, 0)
            else:
                self.fd = open_regular(self.part, os.O_RDWR | os.O_CREAT | os.O_EXCL)
                self.part_stat = os.fstat(self.fd)
                self.meta_fd = open_regular(self.meta, os.O_RDWR | os.O_CREAT | os.O_EXCL)
                self.meta_stat = os.fstat(self.meta_fd)
                self.checkpoint(True)
        except Exception:
            self.close_handles()
            self.lock.__exit__()
            # Never remove pre-existing evidence or another writer's file.
            if not part_exists and self.part_stat:
                self.remove_owned(self.part, self.part_stat)
            if not meta_exists and self.meta_stat:
                self.remove_owned(self.meta, self.meta_stat)
            raise

    @staticmethod
    def write_all(fd, data):
        view = memoryview(data)
        while view:
            written = os.write(fd, view)
            if not written:
                raise CliError("output-write-failed", "output write made no progress")
            view = view[written:]

    def checkpoint(self, force=False):
        if not force and self.offset-self.checkpoint_offset < 1024*1024:
            return
        os.fsync(self.fd)
        prefix = self.digest.hexdigest()
        m = {"v": 2, "identity": self.identity, "received": self.offset, "prefixHash": prefix,
             "mac": part_mac(self.secret, self.identity, self.offset, prefix)}
        self.write_all(self.meta_fd, json_text(m).encode() + b"\n")
        os.fsync(self.meta_fd)
        self.checkpoint_offset = self.offset

    def write(self, data):
        if self.closed:
            raise CliError("sink-closed", "output sink is not writable")
        if self.offset+len(data) > self.identity["size"]:
            raise CliError("size-mismatch", "received more bytes than declared", "transfer")
        self.write_all(self.fd, data)
        self.offset += len(data)
        self.digest.update(data)
        self.checkpoint()

    def state_record(self):
        return {**self.identity, "partPath": str(self.part), "metaPath": str(self.meta), "received": self.offset, "updatedAt": int(time.time()*1000)}

    def close_handles(self):
        for name in ("fd", "meta_fd"):
            fd = getattr(self, name)
            if fd is not None:
                os.close(fd)
                setattr(self, name, None)

    @staticmethod
    def remove_owned(path, expected):
        if same_file(expected, lstat_or_none(path)):
            with contextlib.suppress(FileNotFoundError):
                os.unlink(path)

    def finish(self):
        if self.offset != self.identity["size"]:
            raise CliError("size-mismatch", "received size differs from manifest", "transfer")
        self.checkpoint(True)
        current = os.fstat(self.fd)
        if current.st_size != self.identity["size"]:
            raise CliError("partial-size-mismatch", "partial size changed before publication", "transfer")
        # Re-read actual disk bytes, rather than trusting the incremental hash
        # if another local process edited the open inode behind our back.
        os.lseek(self.fd, 0, 0)
        digest = hashlib.sha256()
        while True:
            chunk = os.read(self.fd, 256*1024)
            if not chunk:
                break
            digest.update(chunk)
        if digest.hexdigest() != self.digest.hexdigest() or (self.identity["hash"] and digest.hexdigest() != self.identity["hash"]):
            self.abort("hash-mismatch")
            raise CliError("hash-mismatch", "hash verification failed", "transfer")
        if not same_file(current, lstat_or_none(self.part)):
            raise CliError("partial-collision", "partial changed before publication")
        try:
            os.link(self.part, self.final, follow_symlinks=False)
        except FileExistsError:
            raise CliError("output-collision", f"output already exists: {self.final}")
        except OSError:
            raise CliError("output-publish-failed", f"could not publish {self.final}; the filesystem must support hard links")
        if not same_file(current, lstat_or_none(self.final)):
            raise CliError("output-publish-race", "published file identity changed")
        self.close_handles()
        self.closed = True
        self.remove_owned(self.part, current)
        self.remove_owned(self.meta, self.meta_stat)
        self.lock.__exit__()

    def abort(self, reason):
        if self.closed:
            return False
        keep = reason in self.RETAIN
        try:
            if keep:
                self.checkpoint(True)
        except OSError:
            keep = False
        self.close_handles()
        self.closed = True
        if not keep:
            self.remove_owned(self.part, self.part_stat)
            self.remove_owned(self.meta, self.meta_stat)
        self.lock.__exit__()
        return keep


def select_backend(args, reporter):
    mode = args.get("rtc-backend", os.environ.get("YINK_RTC_BACKEND", "auto"))
    if mode not in ("auto", "relay"):
        if mode in ("node-datachannel", "werift", "aiortc"):
            raise CliError("rtc-backend-unavailable", f"{mode} is unavailable in this Python build; use --rtc-backend relay", details={"backend": mode})
        raise usage("bad-rtc-backend", f"unknown RTC backend: {mode}")
    reporter.emit("backend", {"backend": "relay", "capabilities": {"direct": False, "resume": True}}, "RTC backend: relay (Python build; relay byte budgets apply)")


async def ws_connect(origin, path, timeout=30):
    try:
        from websockets.asyncio.client import connect
    except ImportError:
        raise CliError("dependency-unavailable", "websockets is required; run with uv run yink.py or install in a dedicated virtual environment")
    url = secure_origin(origin).replace("https://", "wss://", 1).replace("http://", "ws://", 1) + path
    try:
        # proxy=None avoids accidentally presenting private routing metadata to
        # an environment-configured HTTP proxy; no application ping wakes a DO.
        return await connect(url, open_timeout=timeout, max_size=256*1024, max_queue=32, ping_interval=None, close_timeout=2, proxy=None)
    except Exception as error:
        raise CliError("remote-connection-failed", "WebSocket connection failed", "remote") from error


def websocket_closed(error):
    try:
        from websockets.exceptions import ConnectionClosed
        return isinstance(error, ConnectionClosed)
    except ImportError:
        return False


class ShareConnection:
    def __init__(self, origin, room, reporter):
        self.origin, self.room, self.reporter = origin, room, reporter
        self.ws = None
        self.sid = None
        self.joined = None

    async def send(self, kind, fields, dst=0):
        body = {"k": kind, **fields}
        validate_share_message(body, self.sid, self.sid != 0)
        bare = frame_encode(self.sid, dst, 1)
        cipher = message_seal(self.room, frame_decode(bare)["ad"], body)
        await self.ws.send(frame_encode(self.sid, dst, 1, cipher))

    async def close(self):
        if self.ws:
            await self.ws.close()

    async def connect(self, hello, timeout=30):
        self.ws = await ws_connect(self.origin, "/ws/share?room=" + self.room["id"].hex(), timeout)
        await self.ws.send(json_text(hello))
        try:
            async with asyncio.timeout(timeout):
                while True:
                    value = await self.ws.recv()
                    if not isinstance(value, str):
                        continue
                    m = json.loads(value)
                    if m.get("t") in ("hosted", "joined"):
                        sid = 0 if m["t"] == "hosted" else m.get("sid")
                        if not uint(sid, 32) or (m["t"] == "joined" and sid == 0):
                            raise CliError("remote-connection-failed", "invalid relay seating reply", "remote")
                        self.sid, self.joined = sid, m
                        return m
                    if m.get("t") in ("err", "gone", "full"):
                        raise CliError("room-full" if m["t"] == "full" else m.get("code") or m.get("reason") or "remote-unavailable", "share relay refused: " + safe_text(m.get("reason") or m.get("code") or m["t"]), "remote")
        except TimeoutError:
            raise CliError("remote-timeout", "timed out waiting for the share", "remote")

    async def event(self, timeout=None):
        try:
            raw = await asyncio.wait_for(self.ws.recv(), timeout) if timeout else await self.ws.recv()
        except TimeoutError:
            raise CliError("remote-timeout", "timed out waiting for the share", "remote")
        if isinstance(raw, str):
            value = json.loads(raw)
            if not isinstance(value, dict):
                raise CliError("frame-invalid", "invalid relay control", "transfer")
            return "control", value
        frame = frame_decode(raw)
        if frame["dst"] not in (self.sid, 255) or (self.sid > 0 and frame["src"] != 0) or (self.sid == 0 and frame["src"] == 0):
            raise CliError("frame-invalid", "invalid share direction", "transfer")
        if frame["ptype"] == 2:
            return "chunk", frame
        body = message_open(self.room, frame["ad"], frame["cipher"])
        validate_share_message(body, frame["src"], self.sid == 0)
        return "message", (frame["src"], body)


def prepare_host(args):
    path = state_path(args)
    sources = {}
    try:
        if args.get("resume"):
            origin, room, host_public = parse_share_link(args["resume"])
            record = load_state(path)["hosts"].get(room["id"].hex())
            if record is None:
                raise CliError("resume-not-found", "no saved host state matches that link")
            if decode64(record["secret"], 32) != room["secret"]:
                raise CliError("resume-secret-mismatch", "saved host secret does not match the link")
            seed = decode64(record["hostSeed"], 32)
            actual_public = crypto_imports()[3].Ed25519PrivateKey.from_private_bytes(seed).public_key().public_bytes_raw()
            if host_public != actual_public or host_public != decode64(record["hostPublic"], 32):
                raise CliError("resume-host-key-mismatch", "saved signing identity does not match the link")
            manifest = validate_manifest(record["manifest"])
            if not uint(record["nextId"], 2**32) or record["nextId"] <= max((x["n"] for x in manifest["files"]), default=-1) or not isinstance(record["sources"], list) or len(record["sources"]) != len(manifest["files"]):
                raise CliError("state-host-invalid", "saved host state is incomplete")
            for saved in record["sources"]:
                source = Source(saved["path"], expected=saved["identity"])
                sources[saved["n"]] = source
                item = next((x for x in manifest["files"] if x["n"] == saved["n"]), None)
                if not item or source.name != item["name"] or source.size != item["size"] or source.hash != item["hash"] or saved["type"] != item["type"]:
                    raise CliError("resume-source-changed", f"saved source changed: {saved['path']}")
            if len(sources) != len(manifest["files"]) or type(record["relayOnly"]) is not bool or (record["token"] is not None and not re.fullmatch(r"[0-9a-f]{32}", record["token"])):
                raise CliError("state-host-invalid", "saved host state is incomplete")
        else:
            origin = args.get("origin", DEFAULT_ORIGIN)
            room = derive_share(os.urandom(32))
            seed = os.urandom(32)
            host_public = crypto_imports()[3].Ed25519PrivateKey.from_private_bytes(seed).public_key().public_bytes_raw()
            note = args.get("note", "")
            manifest = {"v": 1, "mv": 1 + int(bool(note)), "note": note, "files": []}
            for n, source_path in enumerate(args["paths"]):
                source = Source(source_path)
                sources[n] = source
                manifest["files"].append({"n": n, "name": source.name, "size": source.size, "type": source.type, "hash": source.hash})
                manifest["mv"] += 1
                validate_manifest(manifest)
            record = {"secret": encode64(room["secret"]), "hostSeed": encode64(seed), "hostPublic": encode64(host_public), "token": None,
                      "relayOnly": True, "manifest": manifest, "nextId": len(sources), "sources": [s.record(n) for n, s in sources.items()], "updatedAt": int(time.time()*1000)}
        origin = args.get("origin", origin)
        # Never reseat a previously direct-enabled share as relay-only: the DO
        # owns that immutable property. New Python hosts announce relay-only.
        record["updatedAt"] = int(time.time()*1000)
        update_state(path, "hosts", room["id"].hex(), record)
        return origin, room, seed, host_public, manifest, sources, record, path
    except Exception:
        for source in sources.values():
            source.close()
        raise


class ShareHost(ShareConnection):
    def __init__(self, origin, room, reporter, seed, manifest, sources):
        super().__init__(origin, room, reporter)
        self.seed, self.manifest, self.sources = seed, manifest, sources
        self.signed = sign_manifest(manifest, seed, room)
        self.transfers = {}
        self.used = set()
        self.slots = asyncio.Semaphore(4)
        self.bulk_lock = asyncio.Lock()
        self.next_bulk_at = 0

    async def cancel_transfers(self, sid=None):
        tasks = [value["task"] for value in self.transfers.values() if sid is None or value["sid"] == sid]
        for task in tasks:
            task.cancel()
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)

    async def start_transfer(self, sid, body):
        rid = body["rid"]
        try:
            raw_rid = decode64(rid, 8)
        except ValueError:
            return
        item = next((x for x in self.manifest["files"] if x["n"] == body["n"]), None)
        if not item:
            await self.send("share/no", {"rid": rid, "n": body["n"], "reason": "unknown-item"}, sid)
            return
        if rid in self.used or len(self.used) >= 65536 or any(t["sid"] == sid for t in self.transfers.values()):
            await self.send("share/no", {"rid": rid, "n": body["n"], "reason": "host-busy"}, sid)
            return
        if not body["from"] <= body["to"] <= item["size"]:
            await self.send("share/no", {"rid": rid, "n": body["n"], "reason": "too-large"}, sid)
            return
        self.used.add(rid)
        value = {"sid": sid, "acked": body["from"], "sent": body["from"], "changed": asyncio.Event(), "task": None}
        self.transfers[rid] = value
        value["task"] = asyncio.create_task(self.serve(sid, body, raw_rid, item, value))

    async def serve(self, sid, body, raw_rid, item, value):
        rid, n = body["rid"], body["n"]
        try:
            if self.slots.locked():
                await self.send("share/queued", {"rid": rid, "pos": max(1, len(self.transfers)-4)}, sid)
                self.reporter.emit("queued", {"rid": rid, "pos": max(1, len(self.transfers)-4)})
            async with self.slots:
                key = xfer_key(self.room, n, raw_rid)
                while value["sent"] < body["to"]:
                    if value["sent"]-value["acked"] >= 32*32768:
                        value["changed"].clear()
                        await asyncio.wait_for(value["changed"].wait(), 30)
                        continue
                    start = value["sent"]
                    end = min(body["to"], start+32768)
                    plain = self.sources[n].read(start, end)
                    bare = frame_encode(0, sid, 2, flags=4)
                    cipher = chunk_seal(key, self.room, frame_decode(bare)["ad"], start, item["size"], plain)
                    # The DO charges the host SID across ALL readers, rather
                    # than a separate allowance per transfer. Keep one pacing
                    # clock and leave headroom for network delivery jitter.
                    async with self.bulk_lock:
                        await asyncio.sleep(max(0, self.next_bulk_at-time.monotonic()))
                        value["sent"] = end
                        await self.ws.send(frame_encode(0, sid, 2, cipher, flags=4))
                        self.next_bulk_at = time.monotonic() + (len(cipher)+8)/(384*1024)
                    self.reporter.emit("progress", {"rid": rid, "n": n, "transferred": end, "size": item["size"], "direction": "upload", "path": "relay", "sid": sid, "paint": True})
                while value["acked"] < body["to"]:
                    value["changed"].clear()
                    await asyncio.wait_for(value["changed"].wait(), 30)
                await self.send("share/end", {"rid": rid, "n": n, "size": item["size"], **({"hash": item["hash"]} if item["hash"] else {})}, sid)
                self.reporter.emit("completed", {"rid": rid, "n": n, "side": "host"})
        except asyncio.CancelledError:
            raise
        except Exception as error:
            reason = "changed-on-disk" if isinstance(error, CliError) and error.code == "changed-on-disk" else "gone"
            with contextlib.suppress(Exception):
                await self.send("share/no", {"rid": rid, "n": n, "reason": reason}, sid)
            self.reporter.emit("transfer-error", {"rid": rid, "n": n, "reason": reason, "sid": sid})
        finally:
            if self.transfers.get(rid) is value:
                self.transfers.pop(rid, None)

    async def process(self, event, value):
        if event == "control":
            kind = value.get("t")
            if kind in ("reader-up", "reader-gone"):
                self.reporter.emit("reader", {"sid": value.get("sid"), "state": "connected" if kind == "reader-up" else "gone", "rung": "relay", "paint": True})
                if kind == "reader-gone":
                    await self.cancel_transfers(value.get("sid"))
            if kind == "err":
                self.reporter.emit("status", {"code": value.get("code"), "reason": value.get("reason")})
                if value.get("code") == "tombstoned":
                    raise CliError("tombstoned", "share was ended", "remote")
            if kind == "relay-warn":
                self.reporter.emit("status", {"code": "relay-warn", "used": value.get("used"), "cap": value.get("cap")})
            return
        if event != "message":
            return
        sid, body = value
        kind = body["k"]
        if kind == "share/hello":
            await self.send("share/manifest", self.signed, sid)
        elif kind == "rtc/offer":
            await self.send("rtc/no", {"epoch": body["epoch"], "reason": "relay-only"}, sid)
        elif kind == "share/get":
            await self.start_transfer(sid, body)
        elif kind in ("share/ack", "share/cancel"):
            active = self.transfers.get(body["rid"])
            if active and active["sid"] == sid:
                if kind == "share/cancel":
                    active["task"].cancel()
                    await asyncio.gather(active["task"], return_exceptions=True)
                elif active["acked"] <= body["off"] <= active["sent"]:
                    active["acked"] = body["off"]
                    active["changed"].set()


async def run_share(args, reporter):
    select_backend(args, reporter)
    origin, room, seed, host_public, manifest, sources, record, path = prepare_host(args)
    host = ShareHost(origin, room, reporter, seed, manifest, sources)
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    previous_handlers = {}

    def stopping(signum, frame):
        loop.call_soon_threadsafe(stop.set)

    for signum in (signal.SIGINT, signal.SIGTERM):
        previous_handlers[signum] = signal.signal(signum, stopping)
    waiter = asyncio.create_task(stop.wait())
    try:
        hello = {"t": "hello-host", "proto": 1, "auth": room["auth"], "relayOnly": record["relayOnly"], **({"token": record["token"]} if record["token"] else {})}
        admitted = await host.connect(hello, args.get("timeout", 30000)/1000)
        token = admitted.get("token")
        if not isinstance(token, str) or not re.fullmatch(r"[0-9a-f]{32}", token):
            raise CliError("remote-connection-failed", "invalid host token", "remote")
        record["token"] = hello["token"] = token
        update_state(path, "hosts", room["id"].hex(), record)
        link = f"{origin}/s#s={encode64(room['secret'])}&k=s&h={encode64(host_public)}"
        reporter.emit("hosted", {"link": link, "shareId": room["id"].hex(), "resumed": bool(args.get("resume"))}, link)
        while not stop.is_set():
            receive = asyncio.create_task(host.event())
            done, _ = await asyncio.wait({receive, waiter}, return_when=asyncio.FIRST_COMPLETED)
            if waiter in done:
                receive.cancel()
                await asyncio.gather(receive, return_exceptions=True)
                break
            try:
                event, value = receive.result()
                await host.process(event, value)
            except Exception as error:
                if not websocket_closed(error):
                    if isinstance(error, CliError):
                        raise
                    # One bad reader frame must not take down every reader.
                    reporter.emit("status", {"code": "frame-invalid"})
                    continue
                code = getattr(getattr(error, "rcvd", None), "code", None)
                if code == 4001:
                    reporter.emit("status", {"code": "superseded"})
                    break
                await host.cancel_transfers()
                reporter.emit("status", {"code": "host-away"})
                deadline = time.monotonic()+29
                delay = .2
                while not stop.is_set():
                    try:
                        await host.close()
                        await host.connect(hello, max(.1, min(3, deadline-time.monotonic())))
                        reporter.emit("status", {"code": "host-back"})
                        break
                    except Exception:
                        if time.monotonic() >= deadline:
                            raise CliError("host-timeout", "host reconnect grace expired", "remote")
                        await asyncio.sleep(delay)
                        delay = min(delay*2, 2)
    finally:
        waiter.cancel()
        await asyncio.gather(waiter, return_exceptions=True)
        await host.cancel_transfers()
        await host.close()
        for source in sources.values():
            source.close()
        for signum, handler in previous_handlers.items():
            signal.signal(signum, handler)


class ShareReader(ShareConnection):
    def __init__(self, origin, room, reporter, host_public, timeout):
        super().__init__(origin, room, reporter)
        self.host_public, self.timeout = host_public, timeout
        self.manifest = None
        self.host_up = False
        self.needs_request = False

    async def hello(self):
        await self.send("share/hello", {"v": 1, "caps": ["stream", "resume"], **({"mv": self.manifest["mv"]} if self.manifest else {})})

    async def open(self):
        m = await self.connect({"t": "hello-reader", "proto": 1, "auth": self.room["auth"]}, self.timeout)
        self.host_up = bool(m.get("hostUp"))
        if self.host_up:
            await self.hello()

    async def receive(self):
        try:
            event, value = await self.event(self.timeout)
        except Exception as error:
            if not websocket_closed(error):
                raise
            self.host_up = False
            self.needs_request = True
            self.reporter.emit("status", {"code": "relay-down"})
            deadline, delay = time.monotonic()+29, .2
            while True:
                try:
                    await self.close()
                    await self.open()
                    return "paused", "relay-down"
                except CliError as err:
                    if err.code in ("tombstoned", "host-left", "host-timeout", "gone", "room-full") or time.monotonic() >= deadline:
                        raise CliError("remote-unavailable", "share reconnect grace expired", "remote")
                    await asyncio.sleep(delay)
                    delay = min(delay*2, 2)
        if event == "control":
            kind = value.get("t")
            if kind == "host-away":
                self.host_up = False
                self.needs_request = True
                self.reporter.emit("status", {"code": "host-away", "graceMs": value.get("graceMs")})
                return "paused", "host-away"
            if kind == "host-back":
                self.host_up = True
                self.needs_request = True
                await self.hello()
                return "control", value
            if kind == "gone":
                raise CliError(value.get("reason") or "remote-unavailable", "share unavailable: " + safe_text(value.get("reason") or "gone"), "remote")
            if kind == "err":
                if value.get("code") == "host-away":
                    self.host_up = False
                    self.needs_request = True
                    return "paused", "host-away"
                raise CliError(value.get("code") or "remote-unavailable", "share refused: " + safe_text(value.get("reason") or value.get("code") or "unknown"), "remote")
            if kind == "full":
                raise CliError("room-full", "the share has no reader seat", "remote")
        if event == "message":
            _, message = value
            if message["k"] == "share/manifest":
                self.manifest = adopt_manifest(message, self.manifest, self.host_public, self.room)
                return "manifest", self.manifest
            if message["k"] == "share/bye":
                raise CliError("host-left", "the host explicitly ended this share", "remote")
            return "message", message
        return event, value

    async def wait_manifest(self):
        async with asyncio.timeout(self.timeout):
            while self.manifest is None:
                await self.receive()
        return self.manifest

    async def pull(self, item, sink):
        rid, key, count = None, None, 0

        async def request():
            nonlocal rid, key, count
            raw = os.urandom(8)
            rid, key, count = encode64(raw), xfer_key(self.room, item["n"], raw), 0
            await self.send("share/get", {"n": item["n"], "rid": rid, "from": sink.offset, "to": item["size"]})
            self.needs_request = False

        await request()
        try:
            while True:
                event, value = await self.receive()
                if event == "paused":
                    key = None
                    sink.checkpoint(True)
                    self.reporter.emit("paused", {"n": item["n"], "committed": sink.offset, "size": item["size"], "reason": value})
                elif event == "manifest":
                    latest = next((x for x in self.manifest["files"] if x["n"] == item["n"]), None)
                    if latest != item:
                        raise CliError("revoked", "transfer item was revoked or changed", "transfer")
                    if self.needs_request:
                        await request()
                elif event == "chunk" and key is not None and self.host_up:
                    try:
                        plain = chunk_open(key, self.room, value["ad"], sink.offset, item["size"], value["cipher"])
                    except Exception:
                        raise CliError("gone", "chunk authentication failed", "transfer")
                    if not plain:
                        raise CliError("gone", "empty transfer chunk", "transfer")
                    sink.write(plain)
                    count += 1
                    self.reporter.emit("progress", {"rid": rid, "n": item["n"], "transferred": sink.offset, "size": item["size"], "direction": "download", "path": "relay", "sid": 0, "paint": True})
                    if count % 16 == 0 or sink.offset == item["size"]:
                        await self.send("share/ack", {"rid": rid, "off": sink.offset})
                elif event == "message" and value.get("rid", rid) == rid:
                    if value["k"] == "share/queued":
                        self.reporter.emit("queued", {"rid": rid, "pos": value["pos"]})
                    elif value["k"] == "share/no":
                        raise CliError(value["reason"], "transfer failed: " + value["reason"], "transfer")
                    elif value["k"] == "share/end" and key is not None:
                        if value["n"] != item["n"] or value["size"] != item["size"] or sink.offset != item["size"]:
                            raise CliError("gone", "transfer ended with a mismatched identity or size", "transfer")
                        sink.finish()
                        return
        finally:
            if not sink.closed and rid:
                with contextlib.suppress(Exception):
                    await self.send("share/cancel", {"rid": rid, "reason": "connection-closed"})


def select_manifest(manifest, only=None):
    if only is None:
        return manifest["files"]
    item = next((x for x in manifest["files"] if x["n"] == only), None)
    if item is None:
        raise CliError("item-not-found", f"share has no item {only}", "remote")
    return [item]


async def run_pull(args, reporter):
    origin, room, host_public = parse_share_link(args["link"])
    select_backend(args, reporter)
    reader = ShareReader(args.get("origin", origin), room, reporter, host_public, args.get("timeout", 30000)/1000)
    path = state_path(args)
    try:
        await reader.open()
        try:
            manifest = await reader.wait_manifest()
        except TimeoutError:
            raise CliError("remote-timeout", "timed out waiting for the share", "remote")
        items = select_manifest(manifest, args.get("only"))
        if args.get("list") or args.get("note"):
            reporter.emit("manifest", {"v": manifest["v"], "mv": manifest["mv"], "count": len(manifest["files"])})
            if args.get("note"):
                reporter.emit("note", {"note": manifest["note"]}, safe_text(manifest["note"]) if reporter.tty else manifest["note"])
            if args.get("list"):
                reporter.emit("list", {"files": items}, "\n".join(f"{x['n']}\t{x['size']}\t{safe_text(x['name'])}" for x in items))
            return
        for item in items:
            # A direct-only large file cannot fit the relay's lifetime cap.
            needed = item["size"] + ((item["size"]+32767)//32768)*24
            if needed > 64*1024*1024:
                raise CliError("over-budget", "this file exceeds the relay reader budget; this Python build has no direct WebRTC transport", "transfer")
            sink = Sink(args.get("out", "."), room, host_public, item)
            try:
                update_state(path, "pulls", sink.key, sink.state_record())
                reporter.emit("progress", {"rid": None, "n": item["n"], "transferred": sink.offset, "size": item["size"], "direction": "download", "path": None, "sid": 0, "paint": True})
                await reader.pull(item, sink)
                update_state(path, "pulls", sink.key, None)
                reporter.emit("completed", {"item": item["n"], "name": item["name"], "path": str(sink.final), "bytes": item["size"]}, safe_text(sink.final))
            except BaseException as error:
                reason = error.code if isinstance(error, CliError) else "connection-closed" if isinstance(error, (KeyboardInterrupt, asyncio.CancelledError)) else "write-failed"
                retained = sink.abort(reason)
                with contextlib.suppress(Exception):
                    update_state(path, "pulls", sink.key, sink.state_record() if retained else None)
                raise
    except (ValueError, KeyError, TypeError) as error:
        raise CliError("frame-invalid", "invalid encrypted share frame", "transfer") from error
    finally:
        await reader.close()


# BEGIN GENERATED WORDS (tools/ports/python/sync-download.py)
WORDS = frozenset("""
acid acorn acre acts afar affix aged agent agile aging agony ahead aide aids aim ajar
alarm alias alibi alien alike alive aloe aloft aloha alone amend amino ample amuse angel anger
angle ankle apple april apron aqua area arena argue arise armed armor army aroma array arson
art ashen ashes atlas atom attic audio avert avoid awake award awoke axis bacon badge bagel
baggy baked baker balmy banjo barge barn bash basil bask batch bath baton bats blade blank
blast blaze bleak blend bless blimp blink bloat blob blog blot blunt blurt blush boast boat
body boil bok bolt boned boney bonus bony book booth boots boss botch both boxer breed
bribe brick bride brim bring brink brisk broad broil broke brook broom brush buck bud buggy
bulge bulk bully bunch bunny bunt bush bust busy buzz cable cache cadet cage cake calm
cameo canal candy cane canon cape card cargo carol carry carve case cash cause cedar chain
chair chant chaos charm chase cheek cheer chef chess chest chew chief chili chill chip chomp
chop chow chuck chump chunk churn chute cider cinch city civic civil clad claim clamp clap
clash clasp class claw clay clean clear cleat cleft clerk click cling clink clip cloak clock
clone cloth cloud clump coach coast coat cod coil coke cola cold colt coma come comic
comma cone cope copy coral cork cost cot couch cough cover cozy craft cramp crane crank
crate crave crawl crazy creme crepe crept crib cried crisp crook crop cross crowd crown crumb
crush crust cub cult cupid cure curl curry curse curve curvy cushy cut cycle dab dad
daily dairy daisy dance dandy darn dart dash data date dawn deaf deal dean debit debt
debug decaf decal decay deck decor decoy deed delay denim dense dent depth derby desk dial
diary dice dig dill dime dimly diner dingy disco dish disk ditch ditzy dizzy dock dodge
doing doll dome donor donut dose dot dove down dowry doze drab drama drank draw dress
dried drift drill drive drone droop drove drown drum dry duck duct dude dug duke duo
dusk dust duty dwarf dwell eagle early earth easel east eaten eats ebay ebony ebook echo
edge eel eject elbow elder elf elk elm elope elude elves email emit empty emu enter
entry envoy equal erase error erupt essay etch evade even evict evil evoke exact exit fable
faced fact fade fall false fancy fang fax feast feed femur fence fend ferry fetal fetch
fever fiber fifth fifty film filth final finch fit five flag flaky flame flap flask fled
flick fling flint flip flirt float flock flop floss flyer foam foe fog foil folic folk
food fool found fox foyer frail frame fray fresh fried frill frisk from front frost froth
frown froze fruit gag gains gala game gap gas gave gear gecko geek gem genre gift
gig gills given giver glad glass glide gloss glove glow glue goal going golf gong good
gooey goofy gore gown grab grain grant grape graph grasp grass grave gravy gray green greet
grew grid grief grill grip grit groom grope growl grub grunt guide gulf gulp gummy guru
gush gut guy habit half halo halt happy harm hash hasty hatch hate haven hazel hazy
heap heat heave hedge hefty help herbs hers hub hug hula hull human humid hump hung
hunk hunt hurry hurt hush hut ice icing icon icy igloo image ion iron islam issue
item ivory ivy jab jam jaws jazz jeep jelly jet jiffy job jog jolly jolt jot
joy judge juice juicy july jumbo jump junky juror jury keep keg kept kick kilt king
kite kitty kiwi knee knelt koala kung ladle lady lair lake lance land lapel large lash
lasso last latch late lazy left legal lemon lend lens lent level lever lid life lift
lilac lily limb limes line lint lion lip list lived liver lunar lunch lung lurch lure
lurk lying lyric mace maker malt mama mango manor many map march mardi marry mash match
mate math moan mocha moist mold mom moody mop morse most motor motto mount mouse mousy
mouth move movie mower mud mug mulch mule mull mumbo mummy mural muse music musky mute
nacho nag nail name nanny nap navy near neat neon nerd nest net next niece ninth
nutty oak oasis oat ocean oil old olive omen onion only ooze opal open opera opt
otter ouch ounce outer oval oven owl ozone pace pagan pager palm panda panic pants panty
paper park party pasta patch path patio payer pecan penny pep perch perky perm pest petal
petri petty photo plank plant plaza plead plot plow pluck plug plus poach pod poem poet
pogo point poise poker polar polio polka polo pond pony poppy pork poser pouch pound pout
power prank press print prior prism prize probe prong proof props prude prune pry pug pull
pulp pulse puma punch punk pupil puppy purr purse push putt quack quake query quiet quill
quilt quit quota quote rabid race rack radar radio raft rage raid rail rake rally ramp
ranch range rank rant rash raven reach react ream rebel recap relax relay relic remix repay
repel reply rerun reset rhyme rice rich ride rigid rigor rinse riot ripen rise risk ritzy
rival river roast robe robin rock rogue roman romp rope rover royal ruby rug ruin rule
runny rush rust rut sadly sage said saint salad salon salsa salt same sandy santa satin
sauna saved savor sax say scale scam scan scare scarf scary scoff scold scoop scoot scope
score scorn scout scowl scrap scrub scuba scuff sect sedan self send sepia serve set seven
shack shade shady shaft shaky sham shape share sharp shed sheep sheet shelf shell shine shiny
ship shirt shock shop shore shout shove shown showy shred shrug shun shush shut shy sift
silk silly silo sip siren sixth size skate skew skid skier skies skip skirt skit sky
slab slack slain slam slang slash slate slaw sled sleek sleep sleet slept slice slick slimy
sling slip slit slob slot slug slum slurp slush small smash smell smile smirk smog snack
snap snare snarl sneak sneer sniff snore snort snout snowy snub snuff speak speed spend spent
spew spied spill spiny spoil spoke spoof spool spoon sport spot spout spray spree spur squad
squat squid stack staff stage stain stall stamp stand stank stark start stash state stays steam
steep stem step stew stick sting stir stock stole stomp stony stood stool stoop stop storm
stout stove straw stray strut stuck stud stuff stump stung stunt suds sugar sulk surf sushi
swab swan swarm sway swear sweat sweep swell swept swim swing swipe swirl swoop swore syrup
tacky taco tag take tall talon tamer tank taper taps tarot tart task taste tasty taunt
thank thaw theft theme thigh thing think thong thorn those throb thud thumb thump thus tiara
tidal tidy tiger tile tilt tint tiny trace track trade train trait trap trash tray treat
tree trek trend trial tribe trick trio trout truce truck trump trunk try tug tulip tummy
turf tusk tutor tutu tux tweak tweet twice twine twins twirl twist uncle uncut undo unify
union unit untie upon upper urban used user usher utter value vapor vegan venue verse vest
veto vice video view viral virus visa visor vixen vocal voice void volt voter vowel wad
wafer wager wages wagon wake walk wand wasp watch water wavy wheat whiff whole whoop wick
widen widow width wife wifi wilt wimp wind wing wink wipe wired wiry wise wish wispy
wok wolf womb wool woozy word work worry wound woven wrath wreck wrist xerox yahoo yam
yard year yeast yelp yield yolk yodel yoga yoyo yummy zebra zero zesty zippy zone zoom
""".split())
EMOJI = ['🦊', '🐙', '🐢', '🦉', '🐝', '🐋', '🦁', '🐸', '🦄', '🐞', '🦋', '🐌', '🐊', '🦀', '🐡', '🦜', '🐳', '🦖', '🐚', '🪶', '🌵', '🌲', '🍄', '🌻', '🍀', '🌊', '🔥', '🌙', '⭐', '🌈', '🪐', '❄️', '🍎', '🍋', '🍇', '🍉', '🥑', '🍒', '🍩', '🥨', '🧀', '🥕', '🍕', '🫐', '🎩', '🎈', '🎲', '🎸', '🎺', '🥁', '🚀', '⚓', '🚲', '🛸', '⏰', '🔑', '🔔', '🧲', '💎', '📌', '✂️', '🖍️', '☂️', '🧦']
# END GENERATED WORDS


def parse_words(text):
    words = [w for w in re.split(r"[^a-z]+", str(text).lower().strip()) if w]
    return "-".join(words) if len(words) == 6 and all(w in WORDS for w in words) else None


def pair_stretch(code):
    return hashlib.pbkdf2_hmac("sha256", code.encode(), b"yink-kdf-v2", 600000, 32).hex()


def pair_room(code):
    return hashlib.sha256(("yink-room|" + pair_stretch(code)).encode()).hexdigest()[:32]


def pair_hello(private=None, nonce=None):
    _, serialization, ec, _, _, _ = crypto_imports()
    private = private or ec.generate_private_key(ec.SECP256R1())
    pub = private.public_key().public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
    return {"private": private, "pub": base64.b64encode(pub).decode(), "nonce": base64.b64encode(nonce or os.urandom(16)).decode()}


def pair_session(code, role, mine, theirs, pin=None):
    _, _, ec, _, _, _ = crypto_imports()
    pub = base64.b64decode(theirs["pub"], validate=True)
    nonce = base64.b64decode(theirs["nonce"], validate=True)
    if len(pub) != 65 or len(nonce) != 16:
        raise ValueError("invalid pair hello")
    remote = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), pub)
    shared = mine["private"].exchange(ec.ECDH(), remote)
    khex = pair_stretch(code)
    base = ("yink-salt|" + khex).encode()
    salt = hmac.new(pin, base, "sha256").digest() if pin else hashlib.sha256(base).digest()
    room = hashlib.sha256(("yink-room|" + khex).encode()).hexdigest()[:32]
    a, b = (mine, theirs) if role == "a" else (theirs, mine)
    info = f"yink-v2|{room}|{a['pub']}|{a['nonce']}|{b['pub']}|{b['nonce']}".encode()
    bits = hkdf(shared, salt, info, 168)
    return {"role": role, "sendKey": bits[:32] if role == "a" else bits[32:64], "recvKey": bits[32:64] if role == "a" else bits[:32],
            "confSend": bits[64:96] if role == "a" else bits[96:128], "confRecv": bits[96:128] if role == "a" else bits[64:96], "fp": bits[128:136], "pin": bits[136:]}


class PairCipher:
    def __init__(self, session):
        self.session = session
        self.send_ctr, self.recv_ctr = [0, 0], [0, 0]

    def seal(self, plain, channel=0):
        self.send_ctr[channel] += 1
        counter = self.send_ctr[channel]
        prefix = bytes([0xa0 if self.session["role"] == "a" else 0xb0, channel, 0, 0])
        nonce = prefix + struct.pack(">Q", counter)
        return struct.pack(">Q", counter) + aes(self.session["sendKey"]).encrypt(nonce, plain, None)

    def open(self, wire, channel=0):
        if len(wire) < 24:
            raise ValueError("short frame")
        counter = struct.unpack(">Q", wire[:8])[0]
        if counter <= self.recv_ctr[channel]:
            raise ValueError("replayed frame")
        prefix = bytes([0xb0 if self.session["role"] == "a" else 0xa0, channel, 0, 0])
        plain = aes(self.session["recvKey"]).decrypt(prefix + wire[:8], wire[8:], None)
        self.recv_ctr[channel] = counter
        return plain


class PairConnection:
    """PairLink's one-shot terminal handshake, all application frames on WS."""
    def __init__(self, code, origin, reporter):
        self.code, self.origin, self.reporter = code, origin, reporter
        self.ws = None
        self.role = self.mine = self.theirs = self.session = self.tag = self.cipher = None
        self.confirmed = False
        self.send_lock = asyncio.Lock()

    async def hello(self):
        if self.mine is None and self.role:
            self.mine = pair_hello()
            await self.ws.send(json_text({"t": "msg", "d": {"k": "hello", "pub": self.mine["pub"], "n": self.mine["nonce"]}}))

    async def open(self):
        self.ws = await ws_connect(self.origin, "/ws?room=" + pair_room(self.code))
        await self.ws.send(json_text({"t": "join", "kind": "term"}))
        while not self.confirmed:
            message = await self.receive()
            if message and message[0] == "application":
                raise CliError("pairing-failed", "application data arrived before key confirmation", "remote")

    async def receive(self):
        try:
            raw = await self.ws.recv()
        except Exception as error:
            if websocket_closed(error):
                raise CliError("remote-unavailable", "peer left", "remote")
            raise
        if isinstance(raw, bytes):
            if not self.confirmed:
                return None
            try:
                plain = self.cipher.open(raw)
            except Exception:
                raise CliError("pair-frame-invalid", "terminal frame authentication failed", "transfer")
            if not plain:
                return None
            if plain[0] == 1:
                m = json.loads(plain[1:])
                if not isinstance(m, dict):
                    raise CliError("pair-frame-invalid", "invalid terminal message", "transfer")
                # Terminal never negotiates a direct application channel.
                if m.get("k") in ("sdp", "ice"):
                    return None
                return "json", m
            if plain[0] == 3:
                return "data", plain[1:]
            return None
        m = json.loads(raw)
        kind = m.get("t")
        if kind == "joined":
            if m.get("role") not in ("a", "b"):
                raise CliError("pairing-failed", "invalid pair role", "remote")
            self.role = m["role"]
            self.reporter.emit("status", {"message": f"joined as '{self.role}'"}, f"joined as '{self.role}'")
            if m.get("peer"):
                await self.hello()
        elif kind == "peer-joined":
            await self.hello()
        elif kind in ("full", "burned") or (kind == "peer-left" and self.confirmed):
            raise CliError("remote-unavailable", "room already has two clients" if kind == "full" else "room was burned" if kind == "burned" else "peer left", "remote")
        elif kind == "msg":
            data = m.get("d", {})
            if data.get("k") == "hello":
                if data.get("pin"):
                    raise CliError("pin-mismatch", "this terminal cannot join a continuity-pinned agent room", "remote")
                fresh = {"pub": data.get("pub"), "nonce": data.get("n")}
                if self.theirs and self.theirs != fresh:
                    raise CliError("peer-rekeyed", "peer changed keys; terminal connection ended", "remote")
                self.theirs = fresh
                await self.hello()
            elif data.get("k") == "confirm":
                self.tag = data.get("tag")
            if self.mine and self.theirs and self.session is None:
                self.session = pair_session(self.code, self.role, self.mine, self.theirs)
                tag = base64.b64encode(hmac.new(self.session["confSend"], ("yink-confirm|" + self.role).encode(), "sha256").digest()).decode()
                await self.ws.send(json_text({"t": "msg", "d": {"k": "confirm", "tag": tag}}))
            if self.session and self.tag and not self.confirmed:
                other = "b" if self.role == "a" else "a"
                expected = hmac.new(self.session["confRecv"], ("yink-confirm|" + other).encode(), "sha256").digest()
                try:
                    good = hmac.compare_digest(expected, base64.b64decode(self.tag, validate=True))
                except (ValueError, TypeError):
                    good = False
                if not good:
                    await self.ws.send(json_text({"t": "burn"}))
                    raise CliError("pairing-failed", "key confirmation FAILED - the two sides did not agree on the code", "remote")
                self.confirmed, self.cipher = True, PairCipher(self.session)
                fp = self.session["fp"]
                seal = {"emoji": EMOJI[fp[0] % 64] + EMOJI[fp[1] % 64], "code": fp[2:4].hex().upper()}
                self.reporter.emit("seal", seal, f"SEAL: {seal['emoji']} {seal['code']}")
                self.reporter.emit("rung", {"rung": "relay"}, "transport: relay")
        return None

    async def send_json(self, value):
        async with self.send_lock:
            await self.ws.send(self.cipher.seal(b"\x01" + json_text(value).encode()))

    async def send_data(self, data):
        async with self.send_lock:
            for start in range(0, len(data), 192*1024):
                await self.ws.send(self.cipher.seal(b"\x03" + data[start:start+192*1024]))


async def read_fd(fd, count=32768):
    loop = asyncio.get_running_loop()
    ready = loop.create_future()
    loop.add_reader(fd, lambda: None if ready.done() else ready.set_result(None))
    try:
        await ready
        return os.read(fd, count)
    finally:
        loop.remove_reader(fd)


async def run_term(args, reporter):
    code = parse_words(args["code"])
    if code is None:
        raise usage("bad-pair-code", "term needs exactly six valid yink words")
    if os.name != "posix":
        raise CliError("term-unavailable", "this Python build requires a POSIX PTY for term; Windows ConPTY is not implemented", details={"dependency": "POSIX PTY"})
    if not sys.stdin.isatty():
        raise CliError("term-consent-required", "terminal sharing requires an interactive TTY for consent")
    import fcntl
    import pty
    import termios
    import tty
    import shutil

    select_backend(args, reporter)
    link = PairConnection(code, args.get("origin", DEFAULT_ORIGIN), reporter)
    pid = master = old_attrs = None
    tasks = []
    resize_tasks = set()
    previous_signals = {}
    reason = "peer-ended"
    try:
        await link.open()
        remote_size = None
        while remote_size is None:
            result = await link.receive()
            if result and result[0] == "json":
                m = result[1]
                if m.get("k") == "hello-helper":
                    remote_size = (max(40, int(m.get("cols") or 100)), max(10, int(m.get("rows") or 30)))
                elif m.get("k") == "bye":
                    return
        reporter.stderr.write("Compare the seal above. Press ENTER to grant control, or Ctrl+C to abort: ")
        reporter.stderr.flush()
        async def consent_input():
            while True:
                data = await read_fd(sys.stdin.fileno(), 1)
                if not data:
                    raise CliError("term-consent-required", "terminal consent was not granted")
                if data == b"\n":
                    return

        async def before_consent():
            nonlocal remote_size
            while True:
                # Keep decrypting while the host compares the seal. Never
                # queue pre-consent keystrokes for the future child shell.
                result = await link.receive()
                if result and result[0] == "json":
                    value = result[1]
                    if value.get("k") == "bye":
                        return
                    if value.get("k") in ("resize", "hello-helper"):
                        remote_size = (max(40, int(value.get("cols") or 100)), max(10, int(value.get("rows") or 30)))

        consent = asyncio.create_task(consent_input())
        discard = asyncio.create_task(before_consent())
        try:
            done, _ = await asyncio.wait((consent, discard), return_when=asyncio.FIRST_COMPLETED)
            if discard in done:
                discard.result()
                return
            consent.result()
            await asyncio.sleep(0)
        finally:
            for task in (consent, discard):
                task.cancel()
            await asyncio.gather(consent, discard, return_exceptions=True)
        size = shutil.get_terminal_size((100, 30))
        cols, rows = max(40, min(size.columns, remote_size[0])), max(10, min(size.lines, remote_size[1]))
        shell = args.get("shell") or os.environ.get("SHELL") or "/bin/sh"
        # Consent must precede process creation and every remote input write.
        try:
            pid, master = pty.fork()
            if pid == 0:
                try:
                    os.execvpe(shell, [shell], {**os.environ, "TERM": "xterm-256color"})
                except OSError:
                    os.write(2, b"yink: could not start shell\n")
                    os._exit(127)
        except OSError:
            raise CliError("pty-start-failed", f"could not start {shell}")
        fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
        await link.send_json({"k": "hello-host", "cols": cols, "rows": rows, "user": getpass.getuser(), "machine": socket.gethostname(), "shell": shell})
        await link.send_json({"k": "size", "cols": cols, "rows": rows})
        old_attrs = termios.tcgetattr(sys.stdin.fileno())
        tty.setraw(sys.stdin.fileno())
        loop = asyncio.get_running_loop()
        ended = asyncio.Event()

        async def resize():
            size = shutil.get_terminal_size((100, 30))
            cols = max(40, min(size.columns, remote_size[0]))
            rows = max(10, min(size.lines, remote_size[1]))
            fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
            await link.send_json({"k": "size", "cols": cols, "rows": rows})

        def on_resize():
            task = asyncio.create_task(resize())
            resize_tasks.add(task)
            task.add_done_callback(resize_tasks.discard)

        for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGWINCH):
            previous_signals[signum] = signal.getsignal(signum)
            loop.add_signal_handler(signum, on_resize if signum == signal.SIGWINCH else ended.set)
        reporter.emit("term-live", {"cols": cols, "rows": rows, "shell": shell}, f"terminal live at {cols}x{rows}; Ctrl+] ends it")
        mirror = reporter.stderr if reporter.json else reporter.stdout
        decoder = codecs.getincrementaldecoder("utf-8")("replace")

        async def output():
            while True:
                try:
                    data = await read_fd(master)
                except OSError:
                    return "shell-exited"
                if not data:
                    return "shell-exited"
                mirror.write(decoder.decode(data))
                mirror.flush()
                await link.send_data(data)

        async def local_input():
            while True:
                data = await read_fd(sys.stdin.fileno())
                if not data or b"\x1d" in data:
                    return "host-ended"
                Sink.write_all(master, data)

        async def remote_input():
            nonlocal remote_size
            while True:
                result = await link.receive()
                if not result:
                    continue
                kind, value = result
                if kind == "data":
                    Sink.write_all(master, value)
                elif kind == "json" and value.get("k") == "bye":
                    return "peer-ended"
                elif kind == "json" and value.get("k") == "resize":
                    remote_size = (max(40, int(value.get("cols") or 100)), max(10, int(value.get("rows") or 30)))
                    await resize()

        async def signal_end():
            await ended.wait()
            return "host-ended"

        tasks = [asyncio.create_task(x()) for x in (output, local_input, remote_input, signal_end)]
        done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        reason = next(iter(done)).result()
    finally:
        for signum, handler in previous_signals.items():
            asyncio.get_running_loop().remove_signal_handler(signum)
            signal.signal(signum, handler)
        for task in resize_tasks:
            task.cancel()
        if resize_tasks:
            await asyncio.gather(*resize_tasks, return_exceptions=True)
        for task in tasks:
            task.cancel()
        if tasks:
            await asyncio.gather(*tasks, return_exceptions=True)
        if old_attrs is not None:
            termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_attrs)
        if master is not None:
            os.close(master)
        if pid:
            with contextlib.suppress(ProcessLookupError):
                os.kill(pid, signal.SIGHUP)
            # Bounded cleanup; do not leave a shell running after connection loss.
            for _ in range(20):
                if os.waitpid(pid, os.WNOHANG)[0]:
                    break
                await asyncio.sleep(.05)
            else:
                with contextlib.suppress(ProcessLookupError):
                    os.kill(pid, signal.SIGKILL)
                os.waitpid(pid, 0)
        if link.ws:
            if link.confirmed:
                with contextlib.suppress(Exception):
                    await link.send_json({"k": "ended", "reason": reason})
            await link.ws.close()
    reporter.emit("completed", {"reason": reason}, "terminal ended: " + reason)


def main(argv=None):
    argv = sys.argv[1:] if argv is None else argv
    # Newline translation on Windows must not mutate piped received content.
    for stream in (sys.stdout, sys.stderr):
        if hasattr(stream, "reconfigure"):
            stream.reconfigure(encoding="utf-8", newline="")
    reporter = Reporter("--json" in argv)
    try:
        if sys.version_info < (3, 11):
            raise CliError("python-unavailable", "Python 3.11 or newer is required")
        args = parse_args(argv)
        reporter.json = args.get("json", False)
        if args.get("help"):
            reporter.stdout.write(USAGE)
            return 0
        command = args["command"]
        if command == "install":
            run_install(reporter)
            return 0
        if command in ("send", "receive", "revoke"):
            {"send": run_send, "receive": run_receive, "revoke": run_revoke}[command](args, reporter)
        else:
            asyncio.run({"share": run_share, "pull": run_pull, "term": run_term}[command](args, reporter))
        return 0
    except CliError as error:
        reporter.error(error)
        return error.exit_code
    except KeyboardInterrupt:
        reporter.error(CliError("cancelled", "cancelled"))
        return 5
    except BrokenPipeError:
        return 5
    except Exception as error:
        # Never dump a traceback, a raw network body, or an accidental URL
        # containing capabilities. Local OS diagnostics are still useful.
        message = error.strerror if isinstance(error, OSError) else "local operation failed"
        reporter.error(CliError("local-failure", message))
        return 5


def install_bytes(data, target, executable=False):
    """Publish owned bytes atomically without following a target symlink."""
    target = Path(target)
    current = lstat_or_none(target)
    if current is not None and not stat.S_ISREG(current.st_mode):
        raise CliError("install-collision", f"refusing non-regular install target: {target}")
    if current is not None and target.read_bytes() == data:
        return False
    temp = target.with_name(f".{target.name}.{os.getpid()}.{os.urandom(6).hex()}.tmp")
    try:
        with os.fdopen(open_regular(temp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o700 if executable else 0o600), "wb") as output:
            output.write(data)
            output.flush()
            os.fsync(output.fileno())
        current = lstat_or_none(target)
        if current is not None and not stat.S_ISREG(current.st_mode):
            raise CliError("install-collision", f"refusing non-regular install target: {target}")
        os.replace(temp, target)
        if executable and os.name != "nt":
            os.chmod(target, 0o700)
    finally:
        with contextlib.suppress(FileNotFoundError):
            temp.unlink()
    return True


def install_file(source, target, executable=False):
    """Copy exact trusted local bytes, atomically; never download a replacement."""
    return install_bytes(Path(source).read_bytes(), target, executable)


def installed_hash(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def validate_owned_install(path, expected_hash):
    current = lstat_or_none(path)
    if current is None:
        return
    if not stat.S_ISREG(current.st_mode):
        raise CliError("install-collision", f"refusing non-regular install file: {path}")
    if not isinstance(expected_hash, str) or not re.fullmatch(r"[0-9a-f]{64}", expected_hash) or installed_hash(path) != expected_hash:
        raise CliError("install-collision", f"refusing to replace a changed or unmanaged file: {path}")


def run_install(reporter):
    import shlex
    frozen = bool(getattr(sys, "frozen", False))
    source = Path(sys.executable if frozen else __file__).absolute()
    if os.name == "nt":
        local = os.environ.get("LOCALAPPDATA")
        if not local or not Path(local).is_absolute():
            raise CliError("install-unavailable", "LOCALAPPDATA must name an absolute per-user directory")
        directory = Path(local)/"Programs"/"yink"
        target = directory/("yink.exe" if frozen else "yink.py")
        wrapper = None if frozen else directory/"yink.cmd"
    else:
        directory = Path.home()/".local"/"bin"
        target = directory/("yink" if frozen else "yink.py")
        wrapper = None if frozen else directory/"yink"
    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
    if directory.is_symlink():
        raise CliError("install-collision", "install directory must not be a symlink")
    marker = directory/".yink-python-install.json"
    with FileLock(str(marker) + ".lock"):
        digest = hashlib.sha256(source.read_bytes()).hexdigest()
        if wrapper is not None:
            if os.name == "nt":
                # % in a Python location would be expanded by cmd.exe; fail rather
                # than silently executing a different interpreter.
                if any(char in sys.executable for char in '%\r\n"'):
                    raise CliError("install-unavailable", "the Python interpreter path cannot be represented safely in a cmd wrapper")
                body = ('@echo off\r\n"' + sys.executable + '" "%~dp0yink.py" %*\r\n').encode()
            else:
                body = ("#!/bin/sh\nexec " + shlex.quote(sys.executable) + " " + shlex.quote(str(target)) + ' "$@"\n').encode()
            wrapper_digest = hashlib.sha256(body).hexdigest()
        else:
            body = None
            wrapper_digest = None
        marker_stat = lstat_or_none(marker)
        if marker_stat is not None:
            try:
                if not stat.S_ISREG(marker_stat.st_mode):
                    raise ValueError()
                with os.fdopen(open_regular(marker, os.O_RDONLY), "rb") as stream:
                    raw = stream.read(65537)
                if len(raw) > 65536:
                    raise ValueError()
                old = json.loads(raw)
                if (not isinstance(old, dict) or old.get("owner") != "yink-python-cli" or old.get("directory") != str(directory)
                        or old.get("artifact") != str(target) or old.get("wrapper") != (str(wrapper) if wrapper else None)
                        or old.get("bundledRuntime") is not frozen):
                    raise ValueError()
            except (ValueError, OSError, UnicodeError):
                raise CliError("install-collision", "existing install marker is not owned by this yink installation")
            validate_owned_install(target, old.get("sha256"))
            if wrapper is not None:
                legacy_wrapper = old.get("wrapperSha256") is None
                validate_owned_install(wrapper, wrapper_digest if legacy_wrapper else old.get("wrapperSha256"))
        else:
            for file in (target, wrapper):
                if file is not None and lstat_or_none(file) is not None and os.path.abspath(file) != os.path.abspath(source):
                    raise CliError("install-collision", f"refusing to replace an unmanaged file: {file}")
        changed = install_file(source, target, executable=frozen)
        if wrapper is not None:
            changed = install_bytes(body, wrapper, executable=os.name != "nt") or changed
        metadata = {"owner": "yink-python-cli", "directory": str(directory), "artifact": str(target), "wrapper": str(wrapper) if wrapper else None, "sha256": digest, "wrapperSha256": wrapper_digest, "bundledRuntime": frozen}
        install_bytes(json_text(metadata).encode() + b"\n", marker)
    paths = [os.path.normcase(os.path.abspath(x)) for x in os.environ.get("PATH", "").split(os.pathsep) if x]
    on_path = os.path.normcase(str(directory.absolute())) in paths
    if os.name == "nt":
        quoted = str(directory).replace("'", "''")
        instruction = f"$env:Path = '{quoted};' + $env:Path"
    else:
        instruction = 'export PATH="$HOME/.local/bin:$PATH"'
    reporter.emit("installed", {"path": str(wrapper or target), "directory": str(directory), "changed": changed, "onPath": on_path, "pathInstruction": None if on_path else instruction, "bundledRuntime": frozen},
                  f"{'Installed' if changed else 'Already installed'} yink: {wrapper or target}")
    if not on_path:
        reporter.info("Make yink available in this shell: " + instruction)
        reporter.info("For future shells, add this directory to your user PATH (or the export line to your shell profile).")
    if not frozen:
        reporter.info("This source install uses the current Python interpreter and its existing dependencies; keep that environment available.")


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