refactor(node): move the CLI out of daemon.py into cli/

Each `if args.command == ...` branch of main() becomes a function in
cli/ (one module per family of verbs); the parser, the process setup
and a VERBS table go to cli/parser.py and cli/dispatch.py. daemon.main
runs the verb or starts the daemon. Tests patch the CLI through
conftest.patch_cli; the CLI golden is unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

12 files changed, 1545 insertions, 1416 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/cli/__init__.py b/packages/meshbay-node/src/meshbay_node/cli/__init__.py
new file mode 100644
index 0000000..45bb893
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/__init__.py
@@ -0,0 +1,2 @@
+"""`meshbay-node`'s command line: the verbs that run and exit, as opposed to
+the daemon that `daemon.main` starts when none is given."""
diff --git a/packages/meshbay-node/src/meshbay_node/cli/api.py b/packages/meshbay-node/src/meshbay_node/cli/api.py
new file mode 100644
index 0000000..a11ab0a
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/api.py
@@ -0,0 +1,131 @@
+"""What every verb needs: the daemon's loopback API, the group an argument
+names, and systemd for the lifecycle verbs."""
+
+import sys
+
+from meshbay_node.config import DEFAULT_CONFIG_PATH, Config
+
+
+def _daemon_api(cfg: Config, path: str, method: str = "GET",
+                timeout: int = 30, body: dict | None = None) -> dict:
+    """
+    Call the daemon's loopback API.
+
+    The daemon owns the roster, the hub session and the live group contexts, so
+    the CLI asks it to act rather than opening its databases behind its back. It
+    also means every operator action goes through the control API's per-run
+    session token (11.5.3), the same gate the desktop client's Node page passes.
+    """
+    import json as _json
+    import urllib.error
+    import urllib.parse
+    import urllib.request
+
+    token_file = cfg.data_dir / "ui-token"
+    if not token_file.exists():
+        print("Node is not running — start it with: meshbay-node")
+        sys.exit(1)
+
+    sep = "&" if "?" in path else "?"
+    url = (f"http://127.0.0.1:{cfg.node.ui_port}{path}"
+           f"{sep}t={token_file.read_text(encoding="utf-8").strip()}")
+    try:
+        data = _json.dumps(body).encode() if body is not None else None
+        req = urllib.request.Request(
+            url, method=method, data=data,
+            headers={"Content-Type": "application/json"} if data else {})
+        with urllib.request.urlopen(req, timeout=timeout) as r:
+            return _json.loads(r.read())
+    except urllib.error.HTTPError as e:
+        raw = e.read().decode()[:600]
+        try:
+            parsed = _json.loads(raw)
+            detail = parsed.get("error", raw)
+            # Endpoints that refuse a name offer the ones that would work; a bare
+            # "no such group" leaves the operator guessing at a UUID.
+            for row in parsed.get("available", []):
+                detail += f"\n  {row.get('name', ''):<24} {row.get('id', '')}"
+        except Exception:
+            detail = raw
+        print(f"failed: {detail}")
+        sys.exit(1)
+    except Exception as e:
+        print(f"failed: {e}")
+        sys.exit(1)
+
+
+def _resolve_group(cfg: Config, group: str | None) -> str:
+    """
+    The group argument as an id, or the only configured one.
+
+    Accepts a name as well, because node.toml already gives every group one and
+    nobody remembers a UUID. A name that matches nothing configured says so, and
+    lists what is — silently passing it through produced a 404 from the daemon
+    that read like the group did not exist on the hub.
+    """
+    if group:
+        by_id = [g for g in cfg.groups if g.id == group]
+        if by_id:
+            return by_id[0].id
+        by_name = [g for g in cfg.groups if g.name == group and g.id]
+        if len(by_name) == 1:
+            return by_name[0].id
+        if len(by_name) > 1:
+            print(f"several groups in node.toml are named {group!r} — use the id")
+            sys.exit(1)
+        # An id this node does not host is still worth passing on: the daemon
+        # gives the better error, naming the group it does host.
+        if "-" in group and len(group) == 36:
+            return group
+        print(f"no group named {group!r} in {DEFAULT_CONFIG_PATH}")
+        if cfg.groups:
+            print("configured groups:")
+            for g in cfg.groups:
+                print(f"  {g.name or '(unnamed)':<24} {g.id or '(no id yet)'}")
+        sys.exit(1)
+    configured = [g.id for g in cfg.groups if g.id]
+    if len(configured) == 1:
+        return configured[0]
+    print("--group is required (several groups configured)"
+          if configured else "no group configured in node.toml")
+    sys.exit(1)
+
+
+def _systemctl_user(verb: str, unit: str, *, not_running_hint: str,
+                    success: str, watch: str | None) -> None:
+    """
+    Run `systemctl --user <verb> <unit>` and report the result.
+
+    The lifecycle authority is the unit, not this process: systemd already
+    knows which PID it started, restarts it on failure (`Restart=on-failure`
+    in the unit) and reloads it correctly (`ExecReload=`). Anything this CLI
+    did instead — finding a process by pattern-matching its command line,
+    signalling it, respawning it — is a second, worse implementation of what
+    systemd is already doing, and pattern-matching a process list has already
+    hit a real developer's real running node by accident.
+
+    Reloads the user manager's view of unit files first. The package
+    installers (deb postinst, rpm %post) run as root and can only reload the
+    *system* manager — a different process from any signed-in user's *user*
+    manager, which is the one that actually owns this unit — so a package
+    upgrade leaves that manager still holding the old unit file and prints a
+    warning naming the exact fix. Doing it here runs it under the right
+    privilege, the user's own, right before the command that would otherwise
+    act on a stale definition. Best-effort and unchecked: a reload the
+    manager did not need must never block what the operator actually asked
+    for, and a genuine problem still surfaces from the verb below.
+    """
+    import subprocess
+
+    subprocess.run(["systemctl", "--user", "daemon-reload"],
+                   capture_output=True, text=True)
+
+    result = subprocess.run(["systemctl", "--user", verb, unit],
+                            capture_output=True, text=True)
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()
+        print(detail or not_running_hint)
+        sys.exit(1)
+    print(success)
+    if watch:
+        print(watch)
diff --git a/packages/meshbay-node/src/meshbay_node/cli/content.py b/packages/meshbay-node/src/meshbay_node/cli/content.py
new file mode 100644
index 0000000..16f1df8
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/content.py
@@ -0,0 +1,133 @@
+"""A group's content: its files, its video matches, its chat."""
+
+import sys
+
+from meshbay_common import MNP_VERSION
+
+from meshbay_node.cli.api import _daemon_api, _resolve_group
+from meshbay_node.config import DEFAULT_CONFIG_PATH, load_config
+
+
+def file(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    sub = args.subcommand or "list"
+    group_id = _resolve_group(cfg, args.group)
+
+    if sub == "list":
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/files")
+        files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"]))
+        if not files:
+            print("no files indexed")
+            return
+        for f in files:
+            print(f"  {f['id'][:12]}  {f['size']:>12}  {f['path']}/{f['name']}")
+        print(f"\n{len(files)} file(s). Remove one with: "
+              f"meshbay-node file rm <id>")
+        return
+
+    if sub == "rm":
+        # Milestone 14.11 — the last operator action that needed a browser.
+        if not args.target:
+            print("usage: meshbay-node file rm <file-id> [--group NAME]")
+            sys.exit(1)
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/files",)
+        matches = [f for f in out.get("files", [])
+                   if f["id"].startswith(args.target)]
+        if not matches:
+            print(f"no file whose id starts with {args.target!r}")
+            sys.exit(1)
+        if len(matches) > 1:
+            print(f"{args.target!r} matches {len(matches)} files — be more specific:")
+            for f in matches[:10]:
+                print(f"  {f['id'][:16]}  {f['path']}/{f['name']}")
+            sys.exit(1)
+        target = matches[0]
+        if not args.yes:
+            print(f"Delete {target['path']}/{target['name']} "
+                  f"({target['size']} bytes) from disk?")
+            print("This removes the file itself, not just the listing.")
+            if input("delete? [y/N] ").strip().lower() not in ("y", "yes"):
+                print("cancelled")
+                return
+        _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}",
+                    method="DELETE")
+        print(f"deleted {target['path']}/{target['name']}")
+        return
+
+    print("usage: meshbay-node file list|rm <id> [--group NAME] [--yes]")
+    sys.exit(1)
+
+
+def video(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    if (args.subcommand or "") != "rematch":
+        print("usage: meshbay-node video rematch [--group NAME] [--yes]")
+        sys.exit(1)
+    group_id = _resolve_group(cfg, args.group)
+    if not args.yes:
+        print("Re-resolve every automatic TMDB match for this group's videos?")
+        print("Manual 'Fix match' corrections are kept. Re-resolution is lazy —")
+        print("each poster re-queries TMDB the next time it is opened.")
+        if input("proceed? [y/N] ").strip().lower() not in ("y", "yes"):
+            print("cancelled")
+            return
+    out = _daemon_api(cfg, f"/api/groups/{group_id}/video/rematch", method="POST")
+    print(f"cleared {out.get('removed', 0)} automatic match(es) "
+          f"across {out.get('videos', 0)} video file(s)")
+    return
+
+
+def chat(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    sub = args.subcommand or "status"
+    group_id = _resolve_group(cfg, args.group)
+
+    if sub == "status":
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/chat")
+        print(f"encryption   always on (MNP {MNP_VERSION})")
+        print(f"epoch        {out.get('epoch', 0)}")
+        print(f"messages     {out.get('encrypted_messages', 0)} encrypted, "
+              f"{out.get('plaintext_messages', 0)} in the clear")
+        if out.get("plaintext_messages"):
+            print("\nThose messages were written before this node spoke MNP "
+                  "2.0 and are\nstill readable off this disk. "
+                  "`chat encrypt-history` converts them.")
+        return
+
+    if sub == "rotate":
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/epoch",
+                          method="POST")
+        print(f"chat epoch {out['epoch']} opened")
+        print("Everyone still in the group keeps reading the history; "
+              "whoever left\ncannot read what is written from now on.")
+        return
+
+    if sub == "encrypt-history":
+        if not args.yes:
+            print("This rewrites the only copy of this group's older "
+                  "messages.")
+            print("A backup of chat.db is taken first, beside it.")
+            if input("re-encrypt now? [y/N] ").strip().lower() not in ("y", "yes"):
+                print("cancelled")
+                return
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/encrypt-history",
+                          method="POST", timeout=300)
+        print(f"re-encrypted {out['converted']} message(s) under epoch "
+              f"{out['epoch']}")
+        print(f"backup       {out['backup']}")
+        return
+
+    if sub == "prune":
+        days = int(args.target or 0)
+        if days < 1:
+            print("usage: meshbay-node chat prune <days> [--group G]")
+            sys.exit(1)
+        out = _daemon_api(
+            cfg, f"/api/groups/{group_id}/chat/prune?max_age_days={days}",
+            method="POST")
+        print(f"removed {out['removed']} message(s) older than {days} day(s)")
+        return
+
+    print("usage: meshbay-node chat "
+          "status|rotate|encrypt-history|prune [--group G]")
+    sys.exit(1)
diff --git a/packages/meshbay-node/src/meshbay_node/cli/dispatch.py b/packages/meshbay-node/src/meshbay_node/cli/dispatch.py
new file mode 100644
index 0000000..849d589
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/dispatch.py
@@ -0,0 +1,68 @@
+"""From the command line to the verb that runs it."""
+
+import logging
+
+from meshbay_node.cli import content, groups, lifecycle, members, settings, setup, status
+from meshbay_node.cli.parser import build_parser
+from meshbay_node.platform import config_dir
+
+
+def start():
+    """Set up the process and read its command line."""
+    from meshbay_node.platform import configure_event_loop, force_utf8_stdio, load_node_env
+    force_utf8_stdio()
+    configure_event_loop()
+    # Before anything reads the environment. On Linux systemd has usually loaded
+    # the same file already via EnvironmentFile=; this is what makes a Windows
+    # run (Startup-folder .vbs, no systemd) and a bare `meshbay-node` behave the
+    # same. Already-set variables are left alone, so it cannot undo either.
+    load_node_env(config_dir())
+
+    parser = build_parser()
+    args = parser.parse_args()
+
+    # Query commands print a report; library logging would interleave with it.
+    quiet = args.command in ("status", "gek-init", "gek", "operator",
+                             "member", "group", "root", "file", "video", "chat",
+                             "denylist", "stun", "reload", "restart-daemon",
+                             "reset")
+    logging.basicConfig(
+        level=logging.ERROR if quiet else getattr(logging, args.log_level),
+        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
+    )
+    return args
+
+
+# Each verb and what runs it. A command line naming none of them starts the
+# daemon, which is daemon.main's to do.
+VERBS = {
+    "init": setup.init,
+    "reset": setup.reset,
+    "status": status.status,
+    "gek-init": groups.gek,
+    "gek": groups.gek,
+    "operator": members.operator,
+    "member": members.member,
+    "group": groups.group,
+    "root": groups.root,
+    "file": content.file,
+    "video": content.video,
+    "chat": content.chat,
+    "denylist": settings.denylist,
+    "stun": settings.stun,
+    "transfers": settings.transfers,
+    "reload": lifecycle.reload,
+    "restart-daemon": lifecycle.restart_daemon,
+    "autostart": lifecycle.autostart,
+    "service": lifecycle.service,
+    "calibrate-argon2": setup.calibrate,
+}
+
+
+def run(args) -> bool:
+    """Run the verb on the command line; False when it names none."""
+    verb = VERBS.get(args.command)
+    if verb is None:
+        return False
+    verb(args)
+    return True
diff --git a/packages/meshbay-node/src/meshbay_node/cli/groups.py b/packages/meshbay-node/src/meshbay_node/cli/groups.py
new file mode 100644
index 0000000..3fab4b9
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/groups.py
@@ -0,0 +1,247 @@
+"""What the node hosts: groups, their directories (roots) and their keys."""
+
+import sys
+from pathlib import Path
+
+from meshbay_node.cli.api import _daemon_api, _resolve_group
+from meshbay_node.config import DEFAULT_CONFIG_PATH, load_config
+
+
+def group(args) -> None:
+    if args.subcommand in (None, "list"):
+        # Milestone 14.2.
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        out = _daemon_api(cfg, "/api/groups")
+        groups = out.get("groups", [])
+        if not groups:
+            print("no groups hosted — add one with: "
+                  "meshbay-node group add <name> --dir <path>")
+            return
+        for g in groups:
+            key = "GEK" if g.get("has_gek") else "NO KEY"
+            print(f"  {g['name']}  [{g['visibility']}/{g.get('join_policy')}]  "
+                  f"{key}  {g['file_count']} file(s)  "
+                  f"{g.get('peers', 0)} peer(s)")
+            print(f"      {g['id']}")
+            for r in g.get("roots", []):
+                flags = []
+                if r.get("writable"):
+                    flags.append("rw")
+                else:
+                    flags.append("ro")
+                if r.get("removable"):
+                    flags.append("removable")
+                if r.get("ejected"):
+                    flags.append("ejected")
+                flag_str = f" ({', '.join(flags)})" if flags else ""
+                live = "" if r.get("available", True) else "  [UNAVAILABLE]"
+                print(f"      root {r['name']}{flag_str}{live}")
+            if not g.get("has_gek"):
+                print(f"      give it a key:  meshbay-node gek init "
+                      f"--group {g['name']}")
+        return
+
+    if args.subcommand == "remove":
+        if not args.target:
+            print("usage: meshbay-node group remove <name>")
+            sys.exit(1)
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        if not args.yes:
+            answer = input(f"Remove group '{args.target}' from this node? [y/N] ")
+            if answer.lower() not in ("y", "yes"):
+                print("cancelled")
+                return
+        out = _daemon_api(cfg, "/api/groups/detach", method="POST",
+                          body={"name": args.target})
+        print(f"{out['name']} ({out['group_id'][:8]}) removed from {out['config']}")
+        print()
+        print("Restart the daemon to stop hosting it:")
+        print("  meshbay-node restart-daemon")
+        return
+
+    if args.subcommand != "add":
+        print("usage: meshbay-node group list|add|remove <name>")
+        sys.exit(1)
+    if not args.target or not args.dir:
+        print("usage: meshbay-node group add <name> --dir <path> "
+              "[--no-writable]")
+        print()
+        print("The group must already exist on the hub and be yours. This")
+        print("only tells the node to host it, and picks its first")
+        print("directory, which accepts uploads unless --no-writable.")
+        print("Add more with:  meshbay-node root add <path> [--writable]")
+        sys.exit(1)
+
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    # Writable unless the operator says otherwise: a brand-new group that
+    # cannot receive a single file until its owner finds a second command
+    # is not a working group. Every root added *later* is read-only by
+    # default, which is the opposite rule and the right one there.
+    writable = args.writable is not False
+    body = {"name": args.target, "shared_dir": args.dir,
+            "writable": writable}
+    out = _daemon_api(cfg, "/api/groups/attach", method="POST", body=body)
+    print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}")
+    print(f"  shared_dir   {out['shared_dir']}"
+          f"  ({'read-write' if writable else 'read-only'})")
+    print()
+    print("Tell the daemon to re-read its config, then give the group a key:")
+    print("  meshbay-node reload")
+    print(f"  meshbay-node gek init --group {out['name']}")
+    print()
+    print("The key is this group's own — members of your other groups cannot")
+    print("read it, and joining one says nothing about the other.")
+    return
+
+
+def root(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    sub = args.subcommand or "list"
+    group_id = _resolve_group(cfg, args.group)
+
+    if sub == "list":
+        out = _daemon_api(cfg, "/api/groups")
+        group = next((g for g in out.get("groups", [])
+                      if g["id"] == group_id), None)
+        if not group:
+            print(f"group {group_id[:8]} not hosted on this node")
+            sys.exit(1)
+        roots = group.get("roots", [])
+        if not roots:
+            print("no roots configured")
+            print(f"add one:  meshbay-node root add /path/to/dir --group {group_id}")
+            return
+        for r in roots:
+            flags = []
+            if r.get("writable"):
+                flags.append("rw")
+            else:
+                flags.append("ro")
+            if r.get("removable"):
+                flags.append("removable")
+            if r.get("ejected"):
+                flags.append("EJECTED")
+            avail = "available" if r.get("available", True) else "UNAVAILABLE"
+            flags.append(avail)
+            print(f"  {r['name']:<20} {', '.join(flags)}")
+            print(f"    {r.get('path', '?')}")
+        return
+
+    if sub == "add":
+        path = args.target
+        if not path:
+            print("usage: meshbay-node root add <path> [--name NAME] "
+                  "[--writable] [--removable] [--group NAME]")
+            sys.exit(1)
+        body = {
+            "path": path,
+            "name": args.name or Path(path).name,
+            "writable": args.writable if args.writable is not None else True,
+            "removable": bool(args.removable),
+        }
+        _daemon_api(cfg, f"/api/groups/{group_id}/roots",
+                    method="POST", body=body)
+        w = "rw" if body["writable"] else "ro"
+        rm = ", removable" if body["removable"] else ""
+        print(f"added root {body['name']} → {path} ({w}{rm})")
+        print("reload the daemon to start indexing:")
+        print("  meshbay-node reload")
+        return
+
+    if sub == "remove":
+        name = args.target
+        if not name:
+            print("usage: meshbay-node root remove <name> [--group NAME]")
+            sys.exit(1)
+        if not args.yes:
+            print(f"Remove root '{name}' from group {group_id[:8]}?")
+            print("Files on disk are untouched; only the node config changes.")
+            if input("remove? [y/N] ").strip().lower() not in ("y", "yes"):
+                print("cancelled")
+                return
+        _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}",
+                    method="DELETE")
+        print(f"removed root {name}")
+        print("reload the daemon to apply:")
+        print("  meshbay-node reload")
+        return
+
+    if sub == "set":
+        name = args.target
+        if not name:
+            print("usage: meshbay-node root set <name> "
+                  "[--writable|--no-writable] "
+                  "[--removable|--no-removable] [--group NAME]")
+            sys.exit(1)
+        body = {}
+        if args.writable is not None:
+            body["writable"] = args.writable
+        if args.removable is not None:
+            body["removable"] = args.removable
+        if not body:
+            print("nothing to change — pass --writable/--no-writable "
+                  "or --removable/--no-removable")
+            sys.exit(1)
+        _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}",
+                    method="PATCH", body=body)
+        changes = ", ".join(f"{k}={v}" for k, v in body.items())
+        print(f"updated root {name}: {changes}")
+        return
+
+    if sub == "eject":
+        name = args.target
+        if not name:
+            print("usage: meshbay-node root eject <name> [--group NAME]")
+            sys.exit(1)
+        _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/eject",
+                    method="PUT")
+        print(f"ejected root {name} — files are hidden until plugged back")
+        return
+
+    if sub == "plug":
+        name = args.target
+        if not name:
+            print("usage: meshbay-node root plug <name> [--group NAME]")
+            sys.exit(1)
+        _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/plug",
+                    method="PUT")
+        print(f"plugged root {name} — files are visible again")
+        return
+
+    print("usage: meshbay-node root list|add|remove|set|eject|plug [name] "
+          "[--group NAME]")
+    sys.exit(1)
+
+
+def gek(args) -> None:
+    # `gek-init` is the original spelling and still works. `gek rotate` is
+    # the one that matters after a revocation: the ex-member holds the
+    # current key and nothing else takes it from them.
+    sub = "init" if args.command == "gek-init" else (args.subcommand or "init")
+    if sub not in ("init", "rotate"):
+        print("usage: meshbay-node gek init|rotate [--group NAME]")
+        sys.exit(1)
+
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    group_id = _resolve_group(cfg, args.group)
+
+    if sub == "rotate" and not args.yes:
+        print("Rotating replaces this group's key.")
+        print("  · every member re-receives it automatically on their next connect")
+        print("  · anyone revoked keeps the OLD key and loses access to new content")
+        print("  · content already downloaded stays readable to whoever has it")
+        if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"):
+            print("cancelled")
+            return
+
+    out = _daemon_api(cfg, f"/api/groups/{group_id}/gek"
+                           f"{'?rotate=true' if sub == 'rotate' else ''}",
+                      method="POST", timeout=60)
+
+    verb = "rotated" if out.get("rotated") else "ready"
+    print(f"GEK {verb} for {group_id}")
+    print(f"  {out.get('authorized_members', 0)} authorized member(s) — each "
+          f"receives the key on connect")
+    for err in out.get("errors") or []:
+        print(f"  ! {err}")
+    return
diff --git a/packages/meshbay-node/src/meshbay_node/cli/lifecycle.py b/packages/meshbay-node/src/meshbay_node/cli/lifecycle.py
new file mode 100644
index 0000000..6341561
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/lifecycle.py
@@ -0,0 +1,135 @@
+"""Starting, stopping and reloading the node: systemd here, the Startup
+launcher or the service task on Windows."""
+
+import sys
+
+from meshbay_node.cli.api import _daemon_api, _systemctl_user
+from meshbay_node.config import DEFAULT_CONFIG_PATH, load_config
+
+
+def reload(args) -> None:
+    if sys.platform == "win32":
+        # No systemd, no SIGHUP: the daemon exposes a hot reload on its
+        # own loopback API (the same one ops.reload_config drives).
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        _daemon_api(cfg, "/api/reload", method="POST")
+        print("sent reload to the running node")
+        return
+    _systemctl_user(
+        "reload", "meshbay-node",
+        not_running_hint="Node is not running as a systemd unit — start it "
+                          "with: systemctl --user start meshbay-node",
+        success="sent reload to meshbay-node",
+        watch="watch the result:  journalctl --user -u meshbay-node -f")
+    return
+
+
+def restart_daemon(args) -> None:
+    if sys.platform == "win32":
+        from meshbay_node.platform import (
+            autostart_end,
+            autostart_run,
+            service_end,
+            service_run,
+            service_status,
+        )
+        if service_status()["installed"]:
+            service_end()
+            service_run()
+            print("restarted the node (service task)")
+            return
+        autostart_end()   # kill whatever is running now
+        try:
+            autostart_run()
+        except RuntimeError as e:
+            print(f"Could not restart: {e}. Stop the daemon (Ctrl+C) and "
+                  "relaunch it from where meshbay-node is on PATH.")
+            sys.exit(1)
+        print("restarted the node")
+        return
+    _systemctl_user(
+        "restart", "meshbay-node",
+        not_running_hint="meshbay-node is not installed as a systemd unit — "
+                          "see packaging/systemd/",
+        success="meshbay-node restarted via systemd",
+        watch="check status:  systemctl --user status meshbay-node\n"
+              "watch logs:    journalctl --user -u meshbay-node -f")
+    return
+
+
+def autostart(args) -> None:
+    from meshbay_node import platform as _plat
+    if sys.platform != "win32":
+        print("autostart is Windows-only — elsewhere use "
+              "'systemctl --user enable --now meshbay-node'.")
+        sys.exit(1)
+    sub = args.subcommand or "status"
+    if sub == "install":
+        _plat.autostart_install()
+        print("Installed the Startup launcher — meshbay-node starts at "
+              "each sign-in (no window, no admin).")
+        print("Start it now with: meshbay-node autostart start")
+    elif sub == "remove":
+        _plat.autostart_remove()
+        print("Removed the Startup launcher.")
+    elif sub == "start":
+        try:
+            _plat.autostart_run()
+        except RuntimeError as e:
+            print(f"Could not start: {e}")
+            sys.exit(1)
+        print("started")
+    elif sub == "stop":
+        _plat.autostart_end()
+        print("stopped")
+    elif sub == "status":
+        st = _plat.autostart_status()
+        if st["installed"]:
+            print("autostart    installed — runs meshbay-node at sign-in")
+        else:
+            print("autostart    not installed — meshbay-node autostart install")
+    else:
+        print("autostart: install | remove | start | stop | status")
+        sys.exit(1)
+    return
+
+
+def service(args) -> None:
+    from meshbay_node import platform as _plat
+    if sys.platform != "win32":
+        print("service mode is Windows-only — elsewhere use "
+              "'systemctl --user enable --now meshbay-node'.")
+        sys.exit(1)
+    sub = args.subcommand or "status"
+    if sub == "install":
+        try:
+            _plat.service_install()
+        except RuntimeError as e:
+            print(f"Could not install: {e}")
+            if "denied" in str(e).lower():
+                print("Run this from an elevated (Administrator) prompt.")
+            sys.exit(1)
+        print(f"Registered the {_plat.TASK_NAME!r} scheduled task — it starts "
+              "meshbay-node at boot, as this user, whether or not you have "
+              "signed in yet (no password stored).")
+        print("Start it now with: meshbay-node service start")
+    elif sub == "remove":
+        _plat.service_remove()
+        print(f"Removed the {_plat.TASK_NAME!r} scheduled task.")
+    elif sub == "start":
+        _plat.service_run()
+        print("started")
+    elif sub == "stop":
+        _plat.service_end()
+        print("stopped")
+    elif sub == "status":
+        st = _plat.service_status()
+        if st["installed"]:
+            print(f"service       installed — {st['state'] or 'unknown state'}")
+        else:
+            print("service       not installed — meshbay-node service install "
+                  "(needs an elevated prompt)")
+    else:
+        print("service: install | remove | start | stop | status")
+        sys.exit(1)
+    return
diff --git a/packages/meshbay-node/src/meshbay_node/cli/members.py b/packages/meshbay-node/src/meshbay_node/cli/members.py
new file mode 100644
index 0000000..ba68056
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/members.py
@@ -0,0 +1,181 @@
+"""Who may use the node: members and the operator's own pairing."""
+
+import sys
+from urllib.parse import quote
+
+from meshbay_node.cli.api import _daemon_api, _resolve_group
+from meshbay_node.config import DEFAULT_CONFIG_PATH, load_config
+
+
+def member(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    sub = args.subcommand or "list"
+
+    if sub == "list":
+        group = args.group or ""
+        out = _daemon_api(
+            cfg, f"/api/roster?group_id={group}" if group else "/api/roster")
+        identities = {i["user_id"]: i for i in out.get("identities", [])}
+
+        members = out.get("members", [])
+        if not members:
+            print("no members admitted yet")
+            print("invite someone:  meshbay-node member invite <username>")
+        for m in members:
+            ident = identities.get(m["user_id"], {})
+            scope = m["group_id"][:8] if m["group_id"] else "node-wide"
+            print(f"{(ident.get('username') or m['user_id'])[:20]:20} "
+                  f"{m['role']:9} {m['status']:8} {scope:10} "
+                  f"pinned {ident.get('pinned_at', '?')} "
+                  f"({ident.get('pinned_via', '?')})")
+
+        invites = out.get("invites", [])
+        if invites:
+            print()
+            for i in invites:
+                # A link names nobody until it is used, so its handle is what
+                # identifies it — and what `cancel` takes.
+                who = (f"link {i['invite_id']}" if i.get("kind") == "link"
+                       else f"user {i['user_id'][:12]}")
+                print(f"pending invite  {who}  "
+                      f"group {(i['group_id'] or 'node-wide')[:8]}  "
+                      f"expires {i['expires_at']}")
+        return
+
+    # `member upload` is gone: whether uploads are accepted is `writable`
+    # on the root they would land in, not a per-group switch. Named
+    # explicitly rather than left to the usage line below, which offered a
+    # username for a verb that no longer takes one — an operator following
+    # it would have got "unknown subcommand" and no idea what replaced it.
+    if sub == "upload":
+        print("`member upload` is gone. Uploads are decided per directory "
+              "now:")
+        print()
+        print("  meshbay-node root list                    "
+              "# which are read-write")
+        print("  meshbay-node root set <name> --writable   "
+              "# accept uploads there")
+        print("  meshbay-node root set <name> --no-writable  # stop them")
+        print()
+        print("A group whose directories are all read-only accepts no "
+              "uploads at all,")
+        print("which is what turning the old switch off meant.")
+        sys.exit(1)
+
+    if not args.target:
+        print(f"usage: meshbay-node member {sub} <username>")
+        sys.exit(1)
+
+    if sub == "invite" and args.link:
+        # A link for someone who may have no account yet, bound to their
+        # address on the hub. The code and the ticket are both in it, so
+        # it goes to them and to nobody else — the CLI mails nothing.
+        group_id = _resolve_group(cfg, args.group)
+        out = _daemon_api(
+            cfg, f"/api/groups/{group_id}/invite-links?email={quote(args.target)}",
+            method="POST")
+        from meshbay_node.roster import write_code_file
+        write_code_file(cfg.data_dir, out["link"], out.get("expires_at", ""),
+                        name="invite-link")
+        print(f"INVITATION LINK  {out['link']}")
+        print(f"valid until      {out.get('expires_at', '?')}")
+        print(f"cancel with      meshbay-node member cancel {out['invite_id']}")
+        print()
+        print(f"Send it to {args.target} yourself. It works once, and only for an")
+        print("account registered with that address: they open it, create their")
+        print("account or sign in, and land in the group without typing a code.")
+        return
+
+    if sub == "invite":
+        group_id = _resolve_group(cfg, args.group)
+        out = _daemon_api(
+            cfg, f"/api/groups/{group_id}/invites?username={args.target}",
+            method="POST")
+        from meshbay_node.roster import write_code_file
+        path = write_code_file(cfg.data_dir, out["code"],
+                               out.get("expires_at", ""), name="invite-code")
+        print(f"INVITATION CODE  {out['code']}")
+        print(f"valid until      {out.get('expires_at', '?')}")
+        print()
+        print(f"Send it to {args.target} however you normally talk. It works")
+        print("once, for that account only, and never passes through the hub.")
+        print("They enter it the first time they open the group — you do not")
+        print("need to be online then.")
+        print()
+        print(f"also written to {path}")
+        return
+
+    if sub == "cancel":
+        # Takes back a link that has not been used, on the node and the hub.
+        group_id = _resolve_group(cfg, args.group)
+        out = _daemon_api(
+            cfg, f"/api/groups/{group_id}/invite-links/{quote(args.target)}",
+            method="DELETE")
+        print(f"invitation link {args.target[:8]} cancelled")
+        if not out.get("hub", True):
+            print("The hub's half could not be reached; it expires on its own.")
+        return
+
+    # revoke and unpin both name a person; the daemon resolves the account.
+    # It tries its own roster first and falls back to the hub, so a node that
+    # pinned someone before invitations carried a name is still manageable.
+    match = _daemon_api(cfg, f"/api/resolve?username={args.target}")
+
+    if sub == "revoke":
+        group_id = _resolve_group(cfg, args.group)
+        out = _daemon_api(
+            cfg, f"/api/members/{match['user_id']}/revoke?group_id={group_id}",
+            method="POST")
+        print(f"{args.target} revoked from {group_id[:8]}")
+        if out.get("invites_dropped"):
+            print("Their unredeemed invitation was cancelled.")
+        # The node decides whether a rotation is warranted and says so in
+        # the reminder — somebody who never redeemed a code never held the
+        # key, and advising a rotation there is advice to ignore the next
+        # time it is real. Deciding it again here is the second
+        # implementation this file exists not to have.
+        if out.get("reminder"):
+            print("They stop receiving the group key on their next connection.")
+            print("They still hold the current one — rotate it:")
+            print(f"  meshbay-node gek-init --group {group_id}")
+        return
+
+    if sub == "unpin":
+        _daemon_api(cfg, f"/api/members/{match['user_id']}/unpin", method="POST")
+        print(f"{args.target} unpinned — they can pair again with a new key")
+        print(f"issue a code:  meshbay-node member invite {args.target}")
+        return
+
+    print("usage: meshbay-node member list|invite|cancel|revoke|unpin")
+    sys.exit(1)
+
+
+def operator(args) -> None:
+    if args.subcommand != "pair":
+        print("usage: meshbay-node operator pair")
+        sys.exit(1)
+    if args.group:
+        # Silently ignoring it invited the reading that a code belongs to a
+        # group, and then that pairing had not worked because the group did
+        # not change.
+        print("operator pair takes no --group: pairing is node-wide.")
+        print("One paired browser can invite to, and delete files in, every")
+        print("group this node hosts.")
+        sys.exit(1)
+
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    out = _daemon_api(cfg, "/api/operator/pair", method="POST")
+
+    from meshbay_node.roster import write_code_file
+    path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", ""))
+
+    print(f"PAIRING CODE   {out['code']}")
+    print(f"valid until    {out.get('expires_at', '?')}")
+    print()
+    print("Sign in to the web app as this node's operator, open one of your")
+    print("groups, go to the Members tab and enter the code there.")
+    print("It works once, for that account only, and authorizes invites and")
+    print("file deletion from that browser.")
+    print()
+    print(f"also written to {path}")
+    return
diff --git a/packages/meshbay-node/src/meshbay_node/cli/parser.py b/packages/meshbay-node/src/meshbay_node/cli/parser.py
new file mode 100644
index 0000000..c1f15be
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/parser.py
@@ -0,0 +1,94 @@
+"""The argument parser: every verb, flag and help line `meshbay-node --help`
+prints."""
+
+import argparse
+from pathlib import Path
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="MeshBay Node daemon")
+    parser.add_argument("command", nargs="?",
+                        choices=["init", "reset", "status", "gek-init",
+                                 "gek", "operator", "member", "group", "root",
+                                 "file", "video", "chat", "denylist", "stun",
+                                 "transfers",
+                                 "reload",
+                                 "restart-daemon", "autostart", "service",
+                                 "calibrate-argon2"],
+                        help="init: provision config + keystore | reset: erase all "
+                             "node state | status: node state and keys "
+                             "| operator pair: pair a "
+                             "browser with this node | member list|invite|cancel|revoke|unpin "
+                             "| group list|add|remove "
+                             "| root list|add|remove|set|eject|plug "
+                             "| gek init|rotate | file list|rm "
+                             "| video rematch: re-resolve TMDB matches for a group's "
+                             "videos "
+                             "| chat status|rotate|encrypt-history|prune "
+                             "| denylist show|clear "
+                             "| stun list|add|remove|reset "
+                             "| transfers show|set|max-size|per-member: live "
+                             "transfer slots, the node-wide caps, the largest "
+                             "single upload, and how many one member may run "
+                             "at once in a group "
+                             "| reload: re-read node.toml (hot; systemd or the "
+                             "loopback API) | restart-daemon: restart the node "
+                             "(systemd unit, the Windows autostart launcher, or the "
+                             "service task, whichever applies) "
+                             "| autostart install|remove|start|stop|status "
+                             "(Windows: run meshbay-node at each sign-in, no admin) "
+                             "| service install|remove|start|stop|status "
+                             "(Windows: run at boot, before sign-in, needs admin "
+                             "once to install) "
+                             "| calibrate-argon2: benchmark")
+    parser.add_argument("subcommand", nargs="?",
+                        help="'pair' for operator; list|invite|revoke|unpin for "
+                             "member; list|add|remove for group; "
+                             "list|add|remove|set|eject|plug for root; "
+                             "init|rotate for gek; "
+                             "list|rm for file; rematch for video; show|clear for "
+                             "denylist; list|add|remove|reset for stun; "
+                             "show|set|max-size|per-member for transfers; "
+                             "install|remove|start|stop|status for autostart and "
+                             "for service")
+    parser.add_argument("target", nargs="?",
+                        help="username for member invite|revoke|unpin (an e-mail address with "
+                             "--link, a link id for member cancel); group name "
+                             "for group add; file id for file rm; identifier for "
+                             "denylist clear; download cap for transfers set; "
+                             "size in GB for transfers max-size")
+    parser.add_argument("value", nargs="?",
+                        help="the second value where a verb takes two: the "
+                             "upload cap for transfers set")
+    parser.add_argument("--hub-url", default=None,
+                        help="hub URL, for init (e.g. https://meshbay.org)")
+    parser.add_argument("--username", default=None,
+                        help="hub username, for init")
+    parser.add_argument("--dir", default=None,
+                        help="shared directory, for group add")
+    parser.add_argument("--yes", action="store_true",
+                        help="skip the confirmation for destructive commands")
+    parser.add_argument("--config", type=Path, default=None,
+                        help="Config file path")
+    parser.add_argument("--group", default=None,
+                        help="group id (optional if only one is configured)")
+    parser.add_argument("--link", action="store_true",
+                        help="member invite: an invitation link for this e-mail "
+                             "address, for someone who may have no account yet")
+    parser.add_argument("--writable", action="store_true", default=None,
+                        dest="writable",
+                        help="root accepts member uploads (root add/set)")
+    parser.add_argument("--no-writable", action="store_false",
+                        dest="writable",
+                        help="root is read-only (root add/set, group add)")
+    parser.add_argument("--removable", action="store_true", default=None,
+                        dest="removable",
+                        help="mark root as removable (root set/add)")
+    parser.add_argument("--no-removable", action="store_false",
+                        dest="removable",
+                        help="mark root as not removable (root set)")
+    parser.add_argument("--name", default=None,
+                        help="root name (root add; defaults to directory basename)")
+    parser.add_argument("--log-level", default="INFO",
+                        choices=["DEBUG", "INFO", "WARNING", "ERROR"])
+    return parser
diff --git a/packages/meshbay-node/src/meshbay_node/cli/settings.py b/packages/meshbay-node/src/meshbay_node/cli/settings.py
new file mode 100644
index 0000000..94c7a1b
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/settings.py
@@ -0,0 +1,227 @@
+"""Node-wide settings: the denylist, STUN servers, transfer caps."""
+
+import sys
+
+from meshbay_node.cli.api import _daemon_api, _resolve_group
+from meshbay_node.config import DEFAULT_CONFIG_PATH, load_config
+
+
+def denylist(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    sub = args.subcommand or "show"
+
+    if sub == "show":
+        out = _daemon_api(cfg, "/api/denylist")
+        total = out.get("count", 0)
+        if not total:
+            print("denylist     empty — nothing is being refused")
+            return
+        for kind in ("users", "groups", "jtis"):
+            for entry in out.get(kind, []):
+                print(f"  {kind[:-1]:<6} {entry}")
+        print(f"\n{total} entr(y/ies). These survive a restart (finding H4).")
+        return
+
+    if sub == "clear":
+        if not args.yes:
+            what = args.target or "EVERY entry"
+            print(f"Clearing the denylist re-admits {what}.")
+            print("A revocation the hub sent will not come back on its own.")
+            if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"):
+                print("cancelled")
+                return
+        out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}",
+                          method="POST")
+        print(f"removed {out['removed']} entr(y/ies) ({out['subject']})")
+        return
+
+    print("usage: meshbay-node denylist show|clear [identifier] [--yes]")
+    sys.exit(1)
+
+
+def stun(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    sub = args.subcommand or "list"
+
+    if sub == "list":
+        out = _daemon_api(cfg, "/api/node-settings")
+        servers = out.get("stun_servers", [])
+        if not servers:
+            print("stun servers     (none configured)")
+            return
+        for i, s in enumerate(servers, 1):
+            print(f"  {i}. {s}")
+        return
+
+    if sub == "add":
+        url = args.target
+        if not url:
+            print("usage: meshbay-node stun add <stun:host:port>")
+            sys.exit(1)
+        if not url.startswith("stun:"):
+            print(f"error: STUN URL must start with stun: — got {url!r}")
+            sys.exit(1)
+        out = _daemon_api(cfg, "/api/node-settings")
+        servers = out.get("stun_servers", [])
+        if url in servers:
+            print(f"already present: {url}")
+            return
+        servers.append(url)
+        _daemon_api(cfg, "/api/node-settings", method="PUT",
+                    body={"stun_servers": servers})
+        print(f"added {url} ({len(servers)} servers total)")
+        return
+
+    if sub == "remove":
+        url = args.target
+        if not url:
+            print("usage: meshbay-node stun remove <stun:host:port>")
+            sys.exit(1)
+        out = _daemon_api(cfg, "/api/node-settings")
+        servers = out.get("stun_servers", [])
+        if url not in servers:
+            print(f"not found: {url}")
+            sys.exit(1)
+        servers.remove(url)
+        _daemon_api(cfg, "/api/node-settings", method="PUT",
+                    body={"stun_servers": servers})
+        print(f"removed {url} ({len(servers)} servers remaining)")
+        return
+
+    if sub == "reset":
+        from meshbay_node.config import DEFAULT_STUN_SERVERS
+        _daemon_api(cfg, "/api/node-settings", method="PUT",
+                    body={"stun_servers": list(DEFAULT_STUN_SERVERS)})
+        print("STUN servers reset to defaults:")
+        for s in DEFAULT_STUN_SERVERS:
+            print(f"  {s}")
+        return
+
+    print("usage: meshbay-node stun list|add|remove|reset [url]")
+    sys.exit(1)
+
+
+def transfers(args) -> None:
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    sub = args.subcommand or "show"
+
+    if sub == "show":
+        out = _daemon_api(cfg, "/api/transfers")
+        for kind, pool in out.get("pools", {}).items():
+            print(f"  {kind:<9} {pool['in_use']}/{pool['cap']} in use, "
+                  f"{pool['queued']} queued   (node-wide)")
+        # From the daemon, not from node.toml: a change made on the Node
+        # page is live before it is written back, and the number to print
+        # is the one being enforced. The local file is the fallback rather
+        # than a literal, so there is no second copy of the default here.
+        settings = _daemon_api(cfg, "/api/node-settings")
+        gb = settings.get("max_upload_gb") or cfg.node.max_upload_gb
+        print(f"\n  largest single upload: {gb:g} GB per file "
+              f"(meshbay-node transfers max-size <GB>)")
+        # Per group, because that is the cap that decides how many one
+        # person runs at once — and it is not the node-wide number. An
+        # operator raising `transfers set 8 8` and still seeing two at a
+        # time is looking at this line, which used to print the node's
+        # default and say nothing about where it came from.
+        groups = out.get("groups") or []
+        if groups:
+            print("\n  per member, per group "
+                  "(meshbay-node transfers per-member <dl> <ul> --group X):")
+            for g in groups:
+                how = "set" if g["set"] else "default"
+                print(f"    {g['name']:<20} {g['download']} download(s), "
+                      f"{g['upload']} upload(s)   [{how}]")
+        leases = out.get("leases", [])
+        if not leases:
+            print("\n  nothing transferring")
+            return
+        print(f"\n  {'transfer':<14}{'kind':<10}{'state':<9}"
+              f"{'user':<12}{'bytes':>12}")
+        for x in leases:
+            where = f"  (#{x['ahead'] + 1} in queue)" if x["state"] == "queued" else ""
+            print(f"  {x['tr']:<14}{x['kind']:<10}{x['state']:<9}"
+                  f"{x['user_id'][:10]:<12}{x['bytes']:>12}{where}")
+        return
+
+    if sub == "set":
+        # `transfers set 4 2` — downloads, then uploads. Node-wide; the
+        # per-member cap is a group's setting and is signed, so it is not
+        # settable from here (see `ops.set_transfer_limits`).
+        values = [v for v in (args.target, args.value) if v]
+        if len(values) != 2:
+            print("usage: meshbay-node transfers set <downloads> <uploads>")
+            sys.exit(1)
+        try:
+            downloads, uploads = int(values[0]), int(values[1])
+        except ValueError:
+            print("error: both values must be whole numbers")
+            sys.exit(1)
+        if downloads < 1 or uploads < 1:
+            print("error: a cap below 1 is not 'unlimited'; it would stop "
+                  "every transfer. Revoke the member instead.")
+            sys.exit(1)
+        out = _daemon_api(cfg, "/api/node-settings", method="PUT",
+                          body={"max_concurrent_downloads": downloads,
+                                "max_concurrent_uploads": uploads})
+        print(f"downloads: {downloads}, uploads: {uploads} "
+              f"(applied now, and kept in node.toml)")
+        return
+
+    if sub == "max-size":
+        # The largest single file a member may upload here. Not a
+        # concurrency cap like `set` — it is the one limit that bounds what
+        # a member writes to the operator's disk, which is why it lives
+        # beside them rather than under a verb of its own.
+        if not args.target:
+            print("usage: meshbay-node transfers max-size <GB>")
+            sys.exit(1)
+        try:
+            gb = float(args.target)
+        except ValueError:
+            print("error: the size must be a number of GB (e.g. 8, or 0.5)")
+            sys.exit(1)
+        if gb <= 0:
+            print("error: a ceiling of zero is not 'unlimited'; it would "
+                  "refuse every upload. Make the root read-only instead.")
+            sys.exit(1)
+        _daemon_api(cfg, "/api/node-settings", method="PUT",
+                    body={"max_upload_gb": gb})
+        print(f"largest single upload: {gb:g} GB per file "
+              f"(applied now, and kept in node.toml)")
+        return
+
+    if sub == "per-member":
+        # How many transfers ONE member may run at once in this group. Not
+        # the same knob as `set`, which is the machine's total — and the
+        # reason "I set 8 8 and still only get two" is the commonest
+        # confusion here: per-member is checked first, by design.
+        values = [v for v in (args.target, args.value) if v]
+        if len(values) != 2:
+            print("usage: meshbay-node transfers per-member <downloads> "
+                  "<uploads> [--group NAME]")
+            sys.exit(1)
+        try:
+            downloads, uploads = int(values[0]), int(values[1])
+        except ValueError:
+            print("error: both values must be whole numbers")
+            sys.exit(1)
+        if downloads < 1 or uploads < 1:
+            print("error: a cap below 1 is not 'unlimited'; it would stop "
+                  "every transfer for that member. Revoke them instead.")
+            sys.exit(1)
+        group_id = _resolve_group(cfg, args.group)
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/transfer-limits",
+                          method="PUT",
+                          body={"downloads": downloads, "uploads": uploads})
+        got = out.get("limits", {})
+        started = out.get("started") or []
+        print(f"each member of this group may now run "
+              f"{got.get('download')} download(s) and "
+              f"{got.get('upload')} upload(s) at once")
+        if started:
+            print(f"{len(started)} waiting transfer(s) started at once")
+        return
+
+    print("usage: meshbay-node transfers show|set <downloads> <uploads>|"
+          "max-size <GB>|per-member <downloads> <uploads> [--group NAME]")
+    sys.exit(1)
diff --git a/packages/meshbay-node/src/meshbay_node/cli/setup.py b/packages/meshbay-node/src/meshbay_node/cli/setup.py
new file mode 100644
index 0000000..b2a8e83
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/setup.py
@@ -0,0 +1,195 @@
+"""Provisioning a node and taking it back down: init, reset, calibrate-argon2."""
+
+import sys
+
+from meshbay_node.config import DEFAULT_CONFIG_PATH, Config, load_config
+from meshbay_node.keystore import create_keystore, load_keystore
+from meshbay_node.platform import chmod_private, config_dir, data_dir, state_dir
+
+
+def calibrate_argon2(target_ms: int = 500) -> None:
+    """Benchmark Argon2id and suggest parameters targeting ~target_ms."""
+    import os
+    import time
+
+    print(f"Calibrating Argon2id (target: {target_ms}ms) ...")
+    salt = os.urandom(16)
+    for mem in [65536, 131072, 262144, 524288]:
+        from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
+        t0 = time.perf_counter()
+        Argon2id(salt=salt, length=32, iterations=3,
+                 lanes=4, memory_cost=mem).derive(b"benchmark")
+        elapsed_ms = (time.perf_counter() - t0) * 1000
+        print(f"  memory_cost={mem:>7} ({mem//1024:>4}MB): {elapsed_ms:.0f}ms", end="")
+        if abs(elapsed_ms - target_ms) < target_ms * 0.3:
+            print("  ← recommended")
+        else:
+            print()
+    print("Set memory_cost in meshbay_common/crypto.py: ARGON2_MEMORY_COST")
+
+
+def init(args) -> None:
+    cfg_path = args.config or DEFAULT_CONFIG_PATH
+    cfg_dir = cfg_path.parent
+    cfg_dir.mkdir(parents=True, exist_ok=True)
+
+    from meshbay_node.platform import install_node_env
+    env_written = install_node_env(cfg_dir)
+    if env_written:
+        print(f"Wrote {env_written} (packaged defaults).")
+
+    hub_url = args.hub_url
+    username = args.username
+
+    if not hub_url:
+        hub_url = input("Hub URL [https://meshbay.org]: ").strip() or "https://meshbay.org"
+    if not username:
+        username = input("Hub username: ").strip()
+    if not username:
+        print("Username is required.")
+        sys.exit(1)
+
+    if cfg_path.exists():
+        existing = cfg_path.read_text(encoding="utf-8")
+        import re as _re
+        m = _re.search(r'username\s*=\s*"([^"]*)"', existing)
+        existing_user = m.group(1) if m else ""
+        if existing_user and existing_user != "myusername" and existing_user != username:
+            print(f"Config already exists with username {existing_user!r}.")
+            print("This node belongs to another operator. Use 'meshbay-node reset' first.")
+            sys.exit(1)
+        if existing_user in ("", "myusername"):
+            updated = _re.sub(
+                r'(username\s*=\s*)"[^"]*"', rf'\1"{username}"', existing)
+            updated = _re.sub(
+                r'(url\s*=\s*)"[^"]*"', rf'\1"{hub_url}"', updated, count=1)
+            cfg_path.write_text(updated, encoding="utf-8", newline="\n")
+            print(f"Config updated: username={username}, hub={hub_url}")
+        else:
+            print(f"Config already exists: {cfg_path}")
+    else:
+        unlock_file = cfg_dir / "unlock.key"
+        toml_lines = [
+            "[hub]",
+            f'url      = "{hub_url}"',
+            f'username = "{username}"',
+            "",
+            "[node]",
+            "quic_enabled = false  # QUIC direct path; no client uses it yet",
+            "quic_port = 19010",
+            "ui_port   = 18000",
+            "",
+            "[keystore]",
+            # Forward slashes: a Windows path in a TOML basic string is a
+            # parse error (`\U`, `\a`, ... are escape sequences).
+            f'unlock_file = "{unlock_file.as_posix()}"',
+            "",
+        ]
+        cfg_path.write_text("\n".join(toml_lines) + "\n", encoding="utf-8", newline="\n")
+        chmod_private(cfg_path)
+        print(f"Config written to {cfg_path}")
+
+    unlock_file = cfg_dir / "unlock.key"
+    if not unlock_file.exists():
+        import secrets
+        key = secrets.token_urlsafe(32)
+        unlock_file.write_text(key + "\n", encoding="utf-8", newline="\n")
+        chmod_private(unlock_file)
+        print(f"Unlock key created: {unlock_file}")
+
+    cfg = load_config(cfg_path)
+    if cfg.keystore.path.exists():
+        print(f"Keystore already exists: {cfg.keystore.path}")
+        keys = load_keystore(
+            path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
+    else:
+        keys = create_keystore(
+            path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
+        print(f"Keystore created: {cfg.keystore.path}")
+
+    print(f"Node key: {keys.pk_ed25519_b64}")
+    print()
+    print("Next steps:")
+    print(f"  1. Link this node key on {hub_url} → Settings → Link Node")
+    if sys.platform == "win32":
+        print("  2. meshbay-node autostart install   (run at each sign-in)")
+        print("     — or just: meshbay-node          (start it now, this session)")
+    else:
+        print("  2. systemctl --user enable --now meshbay-node")
+    print("  3. meshbay-node group add <name> --dir /path/to/files")
+    print("  4. meshbay-node gek init")
+    print("  5. meshbay-node operator pair")
+    return
+
+
+def reset(args) -> None:
+    import shutil
+
+    config_dir_ = config_dir()
+    data_dir_ = data_dir()
+    state_dir_ = state_dir()
+
+    items = []
+    for d in (config_dir_, data_dir_):
+        if d.exists():
+            for child in sorted(d.iterdir()):
+                items.append(child)
+
+    if not items:
+        print("Nothing to reset — no node state found.")
+        return
+
+    print("This will permanently erase all node state:")
+    for p in items:
+        print(f"  {p}")
+    print()
+    print("WARNING: a new keystore means a new identity. All group")
+    print("memberships, operator pairings, and invitations are lost.")
+
+    if not args.yes:
+        answer = input("\nProceed? [y/N] ").strip().lower()
+        if answer != "y":
+            print("Aborted.")
+            return
+
+    import json as _json
+    import subprocess as _sp
+    import urllib.error
+    import urllib.request
+
+    token_file = data_dir_ / "ui-token"
+    if token_file.exists():
+        try:
+            cfg = Config(config_dir_ / "node.toml")
+            tok = token_file.read_text(encoding="utf-8").strip()
+            url = (f"http://127.0.0.1:{cfg.node.ui_port}"
+                   f"/api/unlink?t={tok}")
+            req = urllib.request.Request(url, method="DELETE")
+            with urllib.request.urlopen(req, timeout=5) as r:
+                _json.loads(r.read())
+            print("Unlinked node key from hub.")
+        except Exception:
+            print("Could not unlink from hub (daemon not reachable).")
+
+    if sys.platform == "win32":
+        from meshbay_node.platform import autostart_remove, service_remove
+        autostart_remove()
+        service_remove()   # no-op, silently, if not elevated or not installed
+    else:
+        _sp.run(["systemctl", "--user", "disable", "--now", "meshbay-node"],
+                capture_output=True)
+
+    for d in (config_dir_, data_dir_):
+        if d.exists():
+            shutil.rmtree(d)
+            print(f"Removed {d}")
+    if state_dir_.is_dir():
+        shutil.rmtree(state_dir_)
+        print(f"Removed {state_dir_}")
+    print("Node state erased. Run 'meshbay-node init' to start over.")
+    return
+
+
+def calibrate(args) -> None:
+    calibrate_argon2()
+    return
diff --git a/packages/meshbay-node/src/meshbay_node/cli/status.py b/packages/meshbay-node/src/meshbay_node/cli/status.py
new file mode 100644
index 0000000..162bd46
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/cli/status.py
@@ -0,0 +1,127 @@
+"""`meshbay-node status`: what the node is, read even while it is stopped."""
+
+from pathlib import Path
+
+from meshbay_node.config import DEFAULT_CONFIG_PATH, load_config
+from meshbay_node.keystore import load_keystore
+
+
+def status(args) -> None:
+    import json as _json
+    import urllib.request
+
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+    print(f"hub          {cfg.hub.url}  (user {cfg.hub.username or '—'})")
+
+    try:
+        keys = load_keystore(
+            path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
+        print(f"node key     {keys.pk_ed25519_b64}")
+    except FileNotFoundError:
+        print("node key     <no keystore — run: meshbay-node init>")
+    except Exception as e:
+        print(f"node key     <keystore locked: {e}>")
+
+    token_file = cfg.data_dir / "ui-token"
+    live = None
+    if token_file.exists():
+        try:
+            url = (f"http://127.0.0.1:{cfg.node.ui_port}"
+                   f"/api/status?t={token_file.read_text(encoding="utf-8").strip()}")
+            with urllib.request.urlopen(url, timeout=3) as r:
+                live = _json.loads(r.read())
+        except Exception:
+            live = None
+
+    if live:
+        print(f"daemon       running — {live.get('status')}")
+        print(f"node_id      {live.get('endpoint_hint') or '—'}")
+        print(f"groups       {live.get('group_count', 0)}"
+              f"  files {live.get('total_files', 0)}"
+              f"  peers {live.get('webrtc_peers', 0)}")
+
+        needs = live.get("needs", [])
+        if needs:
+            _GUIDANCE = {
+                "node_key_link": (
+                    "Link node key",
+                    f"Copy the node key above and paste it in "
+                    f"Settings → Link Node on {cfg.hub.url}"),
+                "group_add": (
+                    "Add a group",
+                    "meshbay-node group add <name> --dir /path/to/files"),
+                "operator_pair": (
+                    "Pair as operator",
+                    "meshbay-node operator pair"),
+            }
+            print()
+            print("action needed:")
+            for need in needs:
+                if need.startswith("gek_init:"):
+                    name = need.split(":", 1)[1]
+                    print(f"  → Initialize group key for {name}")
+                    print(f"    meshbay-node gek init --group \"{name}\"")
+                elif need in _GUIDANCE:
+                    label, hint = _GUIDANCE[need]
+                    print(f"  → {label}")
+                    print(f"    {hint}")
+                else:
+                    print(f"  → {need}")
+    else:
+        print("daemon       not running")
+
+    print(f"config       {DEFAULT_CONFIG_PATH}")
+    if not cfg.groups:
+        print("groups       none configured — create a group on the hub, then add")
+        print("             a [[groups]] entry with its id and a directory")
+    else:
+        for g in cfg.groups:
+            print(f"  group      {g.name} [{g.visibility}] {g.id or '<no id>'}")
+            if not g.roots:
+                print("             <no directory configured>")
+            for r in g.roots:
+                label = r.name or Path(r.path).name
+                flags = []
+                if getattr(r, 'writable', False) or getattr(r, 'upload', False):
+                    flags.append("rw")
+                else:
+                    flags.append("ro")
+                if getattr(r, 'removable', False):
+                    flags.append("removable")
+                flag_str = f"  ({', '.join(flags)})" if flags else ""
+                live = "" if Path(r.path).expanduser().is_dir() else "  [UNAVAILABLE]"
+                print(f"             {label} → {r.path}{flag_str}{live}")
+    # Node authority: the roster is the source of truth, node.toml the legacy
+    # form. Read the DB directly so this reports correctly while the daemon is
+    # stopped — the state an operator is most often in when checking.
+    import asyncio as _asyncio
+
+    from meshbay_node.roster import Roster as _Roster
+
+    async def _read_roster() -> tuple[list, int]:
+        r = _Roster(db_path=cfg.data_dir / "roster.db")
+        await r.open()
+        try:
+            return (await r.list_members()), len(await r.list_invites())
+        finally:
+            await r.close()
+
+    try:
+        members, pending = _asyncio.run(_read_roster())
+    except Exception as e:
+        members, pending = [], 0
+        print(f"roster       <unreadable: {e}>")
+
+    operators = [m for m in members if m["role"] == "operator"
+                 and m["status"] == "active"]
+    if operators:
+        for op in operators:
+            print(f"operator     {op.get('username') or op['user_id'][:8]}"
+                  f"  key {(op.get('pk_ed25519') or '')[:16]}…"
+                  f"  paired {op.get('pinned_at', '?')}")
+    else:
+        print("operator     NONE PAIRED — file deletion and member invites are")
+        print("             refused. Run: meshbay-node operator pair")
+    if pending:
+        print(f"invites      {pending} pending code(s)")
+    return
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 9faa09c..a81aced 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -33,7 +33,6 @@ import sys
 import time
 from dataclasses import asdict, replace
 from pathlib import Path
-from urllib.parse import quote
 
 import uvicorn
 from meshbay_common import MNP_VERSION
@@ -46,16 +45,17 @@ from meshbay_node.audit import RETENTION_DAYS as AUDIT_RETENTION_DAYS
 from meshbay_node.audit import AuditStore
 from meshbay_node.bundle_store import BundleStore
 from meshbay_node.chat.store import ChatStore
+from meshbay_node.cli.dispatch import run, start
 from meshbay_node.config import DEFAULT_CONFIG_PATH, Config, load_config
 from meshbay_node.hub_client import HubClient, HubConfig
 from meshbay_node.indexer import DirectoryIndexer, GroupIndex, IndexCache
 from meshbay_node.indexer.enrich import Enricher
 from meshbay_node.indexer.enrich_audio import AudioEnricher
 from meshbay_node.indexer.enrich_photo import PhotoEnricher
-from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
+from meshbay_node.keystore import load_or_create_keystore
 from meshbay_node.media_cache import MediaCache
 from meshbay_node.musicbrainz import MusicBrainzClient
-from meshbay_node.platform import chmod_private, config_dir, data_dir, state_dir
+from meshbay_node.platform import chmod_private
 from meshbay_node.roots import RootError, RootSet, entry_abs_path, off_disk
 from meshbay_node.roster import Roster
 from meshbay_node.tmdb import TmdbClient
@@ -106,29 +106,6 @@ def _owning_directory(path: str, directories: list[str]) -> str | None:
     return best
 
 
-# ── Argon2id calibration ──────────────────────────────────────────────────────
-
-def calibrate_argon2(target_ms: int = 500) -> None:
-    """Benchmark Argon2id and suggest parameters targeting ~target_ms."""
-    import os
-    import time
-
-    print(f"Calibrating Argon2id (target: {target_ms}ms) ...")
-    salt = os.urandom(16)
-    for mem in [65536, 131072, 262144, 524288]:
-        from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
-        t0 = time.perf_counter()
-        Argon2id(salt=salt, length=32, iterations=3,
-                 lanes=4, memory_cost=mem).derive(b"benchmark")
-        elapsed_ms = (time.perf_counter() - t0) * 1000
-        print(f"  memory_cost={mem:>7} ({mem//1024:>4}MB): {elapsed_ms:.0f}ms", end="")
-        if abs(elapsed_ms - target_ms) < target_ms * 0.3:
-            print("  ← recommended")
-        else:
-            print()
-    print("Set memory_cost in meshbay_common/crypto.py: ARGON2_MEMORY_COST")
-
-
 # ── Hub WS sender bridge ─────────────────────────────────────────────────────
 
 class _WsSender:
@@ -1807,1399 +1784,11 @@ class NodeDaemon:
         log.info("Node stopped")
 
 
-# ── CLI helpers ───────────────────────────────────────────────────────────────
-
-def _daemon_api(cfg: Config, path: str, method: str = "GET",
-                timeout: int = 30, body: dict | None = None) -> dict:
-    """
-    Call the daemon's loopback API.
-
-    The daemon owns the roster, the hub session and the live group contexts, so
-    the CLI asks it to act rather than opening its databases behind its back. It
-    also means every operator action goes through the control API's per-run
-    session token (11.5.3), the same gate the desktop client's Node page passes.
-    """
-    import json as _json
-    import urllib.error
-    import urllib.parse
-    import urllib.request
-
-    token_file = cfg.data_dir / "ui-token"
-    if not token_file.exists():
-        print("Node is not running — start it with: meshbay-node")
-        sys.exit(1)
-
-    sep = "&" if "?" in path else "?"
-    url = (f"http://127.0.0.1:{cfg.node.ui_port}{path}"
-           f"{sep}t={token_file.read_text(encoding="utf-8").strip()}")
-    try:
-        data = _json.dumps(body).encode() if body is not None else None
-        req = urllib.request.Request(
-            url, method=method, data=data,
-            headers={"Content-Type": "application/json"} if data else {})
-        with urllib.request.urlopen(req, timeout=timeout) as r:
-            return _json.loads(r.read())
-    except urllib.error.HTTPError as e:
-        raw = e.read().decode()[:600]
-        try:
-            parsed = _json.loads(raw)
-            detail = parsed.get("error", raw)
-            # Endpoints that refuse a name offer the ones that would work; a bare
-            # "no such group" leaves the operator guessing at a UUID.
-            for row in parsed.get("available", []):
-                detail += f"\n  {row.get('name', ''):<24} {row.get('id', '')}"
-        except Exception:
-            detail = raw
-        print(f"failed: {detail}")
-        sys.exit(1)
-    except Exception as e:
-        print(f"failed: {e}")
-        sys.exit(1)
-
-
-def _resolve_group(cfg: Config, group: str | None) -> str:
-    """
-    The group argument as an id, or the only configured one.
-
-    Accepts a name as well, because node.toml already gives every group one and
-    nobody remembers a UUID. A name that matches nothing configured says so, and
-    lists what is — silently passing it through produced a 404 from the daemon
-    that read like the group did not exist on the hub.
-    """
-    if group:
-        by_id = [g for g in cfg.groups if g.id == group]
-        if by_id:
-            return by_id[0].id
-        by_name = [g for g in cfg.groups if g.name == group and g.id]
-        if len(by_name) == 1:
-            return by_name[0].id
-        if len(by_name) > 1:
-            print(f"several groups in node.toml are named {group!r} — use the id")
-            sys.exit(1)
-        # An id this node does not host is still worth passing on: the daemon
-        # gives the better error, naming the group it does host.
-        if "-" in group and len(group) == 36:
-            return group
-        print(f"no group named {group!r} in {DEFAULT_CONFIG_PATH}")
-        if cfg.groups:
-            print("configured groups:")
-            for g in cfg.groups:
-                print(f"  {g.name or '(unnamed)':<24} {g.id or '(no id yet)'}")
-        sys.exit(1)
-    configured = [g.id for g in cfg.groups if g.id]
-    if len(configured) == 1:
-        return configured[0]
-    print("--group is required (several groups configured)"
-          if configured else "no group configured in node.toml")
-    sys.exit(1)
-
-
-def _systemctl_user(verb: str, unit: str, *, not_running_hint: str,
-                    success: str, watch: str | None) -> None:
-    """
-    Run `systemctl --user <verb> <unit>` and report the result.
-
-    The lifecycle authority is the unit, not this process: systemd already
-    knows which PID it started, restarts it on failure (`Restart=on-failure`
-    in the unit) and reloads it correctly (`ExecReload=`). Anything this CLI
-    did instead — finding a process by pattern-matching its command line,
-    signalling it, respawning it — is a second, worse implementation of what
-    systemd is already doing, and pattern-matching a process list has already
-    hit a real developer's real running node by accident.
-
-    Reloads the user manager's view of unit files first. The package
-    installers (deb postinst, rpm %post) run as root and can only reload the
-    *system* manager — a different process from any signed-in user's *user*
-    manager, which is the one that actually owns this unit — so a package
-    upgrade leaves that manager still holding the old unit file and prints a
-    warning naming the exact fix. Doing it here runs it under the right
-    privilege, the user's own, right before the command that would otherwise
-    act on a stale definition. Best-effort and unchecked: a reload the
-    manager did not need must never block what the operator actually asked
-    for, and a genuine problem still surfaces from the verb below.
-    """
-    import subprocess
-
-    subprocess.run(["systemctl", "--user", "daemon-reload"],
-                   capture_output=True, text=True)
-
-    result = subprocess.run(["systemctl", "--user", verb, unit],
-                            capture_output=True, text=True)
-    if result.returncode != 0:
-        detail = (result.stderr or result.stdout).strip()
-        print(detail or not_running_hint)
-        sys.exit(1)
-    print(success)
-    if watch:
-        print(watch)
-
-
 # ── Entry point ───────────────────────────────────────────────────────────────
 
 def main() -> None:
-    import argparse
-
-    from meshbay_node.platform import configure_event_loop, force_utf8_stdio, load_node_env
-    force_utf8_stdio()
-    configure_event_loop()
-    # Before anything reads the environment. On Linux systemd has usually loaded
-    # the same file already via EnvironmentFile=; this is what makes a Windows
-    # run (Startup-folder .vbs, no systemd) and a bare `meshbay-node` behave the
-    # same. Already-set variables are left alone, so it cannot undo either.
-    load_node_env(config_dir())
-
-    parser = argparse.ArgumentParser(description="MeshBay Node daemon")
-    parser.add_argument("command", nargs="?",
-                        choices=["init", "reset", "status", "gek-init",
-                                 "gek", "operator", "member", "group", "root",
-                                 "file", "video", "chat", "denylist", "stun",
-                                 "transfers",
-                                 "reload",
-                                 "restart-daemon", "autostart", "service",
-                                 "calibrate-argon2"],
-                        help="init: provision config + keystore | reset: erase all "
-                             "node state | status: node state and keys "
-                             "| operator pair: pair a "
-                             "browser with this node | member list|invite|cancel|revoke|unpin "
-                             "| group list|add|remove "
-                             "| root list|add|remove|set|eject|plug "
-                             "| gek init|rotate | file list|rm "
-                             "| video rematch: re-resolve TMDB matches for a group's "
-                             "videos "
-                             "| chat status|rotate|encrypt-history|prune "
-                             "| denylist show|clear "
-                             "| stun list|add|remove|reset "
-                             "| transfers show|set|max-size|per-member: live "
-                             "transfer slots, the node-wide caps, the largest "
-                             "single upload, and how many one member may run "
-                             "at once in a group "
-                             "| reload: re-read node.toml (hot; systemd or the "
-                             "loopback API) | restart-daemon: restart the node "
-                             "(systemd unit, the Windows autostart launcher, or the "
-                             "service task, whichever applies) "
-                             "| autostart install|remove|start|stop|status "
-                             "(Windows: run meshbay-node at each sign-in, no admin) "
-                             "| service install|remove|start|stop|status "
-                             "(Windows: run at boot, before sign-in, needs admin "
-                             "once to install) "
-                             "| calibrate-argon2: benchmark")
-    parser.add_argument("subcommand", nargs="?",
-                        help="'pair' for operator; list|invite|revoke|unpin for "
-                             "member; list|add|remove for group; "
-                             "list|add|remove|set|eject|plug for root; "
-                             "init|rotate for gek; "
-                             "list|rm for file; rematch for video; show|clear for "
-                             "denylist; list|add|remove|reset for stun; "
-                             "show|set|max-size|per-member for transfers; "
-                             "install|remove|start|stop|status for autostart and "
-                             "for service")
-    parser.add_argument("target", nargs="?",
-                        help="username for member invite|revoke|unpin (an e-mail address with "
-                             "--link, a link id for member cancel); group name "
-                             "for group add; file id for file rm; identifier for "
-                             "denylist clear; download cap for transfers set; "
-                             "size in GB for transfers max-size")
-    parser.add_argument("value", nargs="?",
-                        help="the second value where a verb takes two: the "
-                             "upload cap for transfers set")
-    parser.add_argument("--hub-url", default=None,
-                        help="hub URL, for init (e.g. https://meshbay.org)")
-    parser.add_argument("--username", default=None,
-                        help="hub username, for init")
-    parser.add_argument("--dir", default=None,
-                        help="shared directory, for group add")
-    parser.add_argument("--yes", action="store_true",
-                        help="skip the confirmation for destructive commands")
-    parser.add_argument("--config", type=Path, default=None,
-                        help="Config file path")
-    parser.add_argument("--group", default=None,
-                        help="group id (optional if only one is configured)")
-    parser.add_argument("--link", action="store_true",
-                        help="member invite: an invitation link for this e-mail "
-                             "address, for someone who may have no account yet")
-    parser.add_argument("--writable", action="store_true", default=None,
-                        dest="writable",
-                        help="root accepts member uploads (root add/set)")
-    parser.add_argument("--no-writable", action="store_false",
-                        dest="writable",
-                        help="root is read-only (root add/set, group add)")
-    parser.add_argument("--removable", action="store_true", default=None,
-                        dest="removable",
-                        help="mark root as removable (root set/add)")
-    parser.add_argument("--no-removable", action="store_false",
-                        dest="removable",
-                        help="mark root as not removable (root set)")
-    parser.add_argument("--name", default=None,
-                        help="root name (root add; defaults to directory basename)")
-    parser.add_argument("--log-level", default="INFO",
-                        choices=["DEBUG", "INFO", "WARNING", "ERROR"])
-    args = parser.parse_args()
-
-    # Query commands print a report; library logging would interleave with it.
-    quiet = args.command in ("status", "gek-init", "gek", "operator",
-                             "member", "group", "root", "file", "video", "chat",
-                             "denylist", "stun", "reload", "restart-daemon",
-                             "reset")
-    logging.basicConfig(
-        level=logging.ERROR if quiet else getattr(logging, args.log_level),
-        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
-    )
-
-    if args.command == "init":
-        cfg_path = args.config or DEFAULT_CONFIG_PATH
-        cfg_dir = cfg_path.parent
-        cfg_dir.mkdir(parents=True, exist_ok=True)
-
-        from meshbay_node.platform import install_node_env
-        env_written = install_node_env(cfg_dir)
-        if env_written:
-            print(f"Wrote {env_written} (packaged defaults).")
-
-        hub_url = args.hub_url
-        username = args.username
-
-        if not hub_url:
-            hub_url = input("Hub URL [https://meshbay.org]: ").strip() or "https://meshbay.org"
-        if not username:
-            username = input("Hub username: ").strip()
-        if not username:
-            print("Username is required.")
-            sys.exit(1)
-
-        if cfg_path.exists():
-            existing = cfg_path.read_text(encoding="utf-8")
-            import re as _re
-            m = _re.search(r'username\s*=\s*"([^"]*)"', existing)
-            existing_user = m.group(1) if m else ""
-            if existing_user and existing_user != "myusername" and existing_user != username:
-                print(f"Config already exists with username {existing_user!r}.")
-                print("This node belongs to another operator. Use 'meshbay-node reset' first.")
-                sys.exit(1)
-            if existing_user in ("", "myusername"):
-                updated = _re.sub(
-                    r'(username\s*=\s*)"[^"]*"', rf'\1"{username}"', existing)
-                updated = _re.sub(
-                    r'(url\s*=\s*)"[^"]*"', rf'\1"{hub_url}"', updated, count=1)
-                cfg_path.write_text(updated, encoding="utf-8", newline="\n")
-                print(f"Config updated: username={username}, hub={hub_url}")
-            else:
-                print(f"Config already exists: {cfg_path}")
-        else:
-            unlock_file = cfg_dir / "unlock.key"
-            toml_lines = [
-                "[hub]",
-                f'url      = "{hub_url}"',
-                f'username = "{username}"',
-                "",
-                "[node]",
-                "quic_enabled = false  # QUIC direct path; no client uses it yet",
-                "quic_port = 19010",
-                "ui_port   = 18000",
-                "",
-                "[keystore]",
-                # Forward slashes: a Windows path in a TOML basic string is a
-                # parse error (`\U`, `\a`, ... are escape sequences).
-                f'unlock_file = "{unlock_file.as_posix()}"',
-                "",
-            ]
-            cfg_path.write_text("\n".join(toml_lines) + "\n", encoding="utf-8", newline="\n")
-            chmod_private(cfg_path)
-            print(f"Config written to {cfg_path}")
-
-        unlock_file = cfg_dir / "unlock.key"
-        if not unlock_file.exists():
-            import secrets
-            key = secrets.token_urlsafe(32)
-            unlock_file.write_text(key + "\n", encoding="utf-8", newline="\n")
-            chmod_private(unlock_file)
-            print(f"Unlock key created: {unlock_file}")
-
-        cfg = load_config(cfg_path)
-        if cfg.keystore.path.exists():
-            print(f"Keystore already exists: {cfg.keystore.path}")
-            keys = load_keystore(
-                path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
-        else:
-            keys = create_keystore(
-                path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
-            print(f"Keystore created: {cfg.keystore.path}")
-
-        print(f"Node key: {keys.pk_ed25519_b64}")
-        print()
-        print("Next steps:")
-        print(f"  1. Link this node key on {hub_url} → Settings → Link Node")
-        if sys.platform == "win32":
-            print("  2. meshbay-node autostart install   (run at each sign-in)")
-            print("     — or just: meshbay-node          (start it now, this session)")
-        else:
-            print("  2. systemctl --user enable --now meshbay-node")
-        print("  3. meshbay-node group add <name> --dir /path/to/files")
-        print("  4. meshbay-node gek init")
-        print("  5. meshbay-node operator pair")
-        return
-
-    if args.command == "reset":
-        import shutil
-
-        config_dir_ = config_dir()
-        data_dir_ = data_dir()
-        state_dir_ = state_dir()
-
-        items = []
-        for d in (config_dir_, data_dir_):
-            if d.exists():
-                for child in sorted(d.iterdir()):
-                    items.append(child)
-
-        if not items:
-            print("Nothing to reset — no node state found.")
-            return
-
-        print("This will permanently erase all node state:")
-        for p in items:
-            print(f"  {p}")
-        print()
-        print("WARNING: a new keystore means a new identity. All group")
-        print("memberships, operator pairings, and invitations are lost.")
-
-        if not args.yes:
-            answer = input("\nProceed? [y/N] ").strip().lower()
-            if answer != "y":
-                print("Aborted.")
-                return
-
-        import json as _json
-        import subprocess as _sp
-        import urllib.error
-        import urllib.request
-
-        token_file = data_dir_ / "ui-token"
-        if token_file.exists():
-            try:
-                cfg = Config(config_dir_ / "node.toml")
-                tok = token_file.read_text(encoding="utf-8").strip()
-                url = (f"http://127.0.0.1:{cfg.node.ui_port}"
-                       f"/api/unlink?t={tok}")
-                req = urllib.request.Request(url, method="DELETE")
-                with urllib.request.urlopen(req, timeout=5) as r:
-                    _json.loads(r.read())
-                print("Unlinked node key from hub.")
-            except Exception:
-                print("Could not unlink from hub (daemon not reachable).")
-
-        if sys.platform == "win32":
-            from meshbay_node.platform import autostart_remove, service_remove
-            autostart_remove()
-            service_remove()   # no-op, silently, if not elevated or not installed
-        else:
-            _sp.run(["systemctl", "--user", "disable", "--now", "meshbay-node"],
-                    capture_output=True)
-
-        for d in (config_dir_, data_dir_):
-            if d.exists():
-                shutil.rmtree(d)
-                print(f"Removed {d}")
-        if state_dir_.is_dir():
-            shutil.rmtree(state_dir_)
-            print(f"Removed {state_dir_}")
-        print("Node state erased. Run 'meshbay-node init' to start over.")
-        return
-
-    if args.command == "calibrate-argon2":
-        calibrate_argon2()
-        return
-
-    if args.command == "status":
-        import json as _json
-        import urllib.request
-
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        print(f"hub          {cfg.hub.url}  (user {cfg.hub.username or '—'})")
-
-        try:
-            keys = load_keystore(
-                path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
-            print(f"node key     {keys.pk_ed25519_b64}")
-        except FileNotFoundError:
-            print("node key     <no keystore — run: meshbay-node init>")
-        except Exception as e:
-            print(f"node key     <keystore locked: {e}>")
-
-        token_file = cfg.data_dir / "ui-token"
-        live = None
-        if token_file.exists():
-            try:
-                url = (f"http://127.0.0.1:{cfg.node.ui_port}"
-                       f"/api/status?t={token_file.read_text(encoding="utf-8").strip()}")
-                with urllib.request.urlopen(url, timeout=3) as r:
-                    live = _json.loads(r.read())
-            except Exception:
-                live = None
-
-        if live:
-            print(f"daemon       running — {live.get('status')}")
-            print(f"node_id      {live.get('endpoint_hint') or '—'}")
-            print(f"groups       {live.get('group_count', 0)}"
-                  f"  files {live.get('total_files', 0)}"
-                  f"  peers {live.get('webrtc_peers', 0)}")
-
-            needs = live.get("needs", [])
-            if needs:
-                _GUIDANCE = {
-                    "node_key_link": (
-                        "Link node key",
-                        f"Copy the node key above and paste it in "
-                        f"Settings → Link Node on {cfg.hub.url}"),
-                    "group_add": (
-                        "Add a group",
-                        "meshbay-node group add <name> --dir /path/to/files"),
-                    "operator_pair": (
-                        "Pair as operator",
-                        "meshbay-node operator pair"),
-                }
-                print()
-                print("action needed:")
-                for need in needs:
-                    if need.startswith("gek_init:"):
-                        name = need.split(":", 1)[1]
-                        print(f"  → Initialize group key for {name}")
-                        print(f"    meshbay-node gek init --group \"{name}\"")
-                    elif need in _GUIDANCE:
-                        label, hint = _GUIDANCE[need]
-                        print(f"  → {label}")
-                        print(f"    {hint}")
-                    else:
-                        print(f"  → {need}")
-        else:
-            print("daemon       not running")
-
-        print(f"config       {DEFAULT_CONFIG_PATH}")
-        if not cfg.groups:
-            print("groups       none configured — create a group on the hub, then add")
-            print("             a [[groups]] entry with its id and a directory")
-        else:
-            for g in cfg.groups:
-                print(f"  group      {g.name} [{g.visibility}] {g.id or '<no id>'}")
-                if not g.roots:
-                    print("             <no directory configured>")
-                for r in g.roots:
-                    label = r.name or Path(r.path).name
-                    flags = []
-                    if getattr(r, 'writable', False) or getattr(r, 'upload', False):
-                        flags.append("rw")
-                    else:
-                        flags.append("ro")
-                    if getattr(r, 'removable', False):
-                        flags.append("removable")
-                    flag_str = f"  ({', '.join(flags)})" if flags else ""
-                    live = "" if Path(r.path).expanduser().is_dir() else "  [UNAVAILABLE]"
-                    print(f"             {label} → {r.path}{flag_str}{live}")
-        # Node authority: the roster is the source of truth, node.toml the legacy
-        # form. Read the DB directly so this reports correctly while the daemon is
-        # stopped — the state an operator is most often in when checking.
-        import asyncio as _asyncio
-
-        from meshbay_node.roster import Roster as _Roster
-
-        async def _read_roster() -> tuple[list, int]:
-            r = _Roster(db_path=cfg.data_dir / "roster.db")
-            await r.open()
-            try:
-                return (await r.list_members()), len(await r.list_invites())
-            finally:
-                await r.close()
-
-        try:
-            members, pending = _asyncio.run(_read_roster())
-        except Exception as e:
-            members, pending = [], 0
-            print(f"roster       <unreadable: {e}>")
-
-        operators = [m for m in members if m["role"] == "operator"
-                     and m["status"] == "active"]
-        if operators:
-            for op in operators:
-                print(f"operator     {op.get('username') or op['user_id'][:8]}"
-                      f"  key {(op.get('pk_ed25519') or '')[:16]}…"
-                      f"  paired {op.get('pinned_at', '?')}")
-        else:
-            print("operator     NONE PAIRED — file deletion and member invites are")
-            print("             refused. Run: meshbay-node operator pair")
-        if pending:
-            print(f"invites      {pending} pending code(s)")
-        return
-
-    if args.command == "member":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        sub = args.subcommand or "list"
-
-        if sub == "list":
-            group = args.group or ""
-            out = _daemon_api(
-                cfg, f"/api/roster?group_id={group}" if group else "/api/roster")
-            identities = {i["user_id"]: i for i in out.get("identities", [])}
-
-            members = out.get("members", [])
-            if not members:
-                print("no members admitted yet")
-                print("invite someone:  meshbay-node member invite <username>")
-            for m in members:
-                ident = identities.get(m["user_id"], {})
-                scope = m["group_id"][:8] if m["group_id"] else "node-wide"
-                print(f"{(ident.get('username') or m['user_id'])[:20]:20} "
-                      f"{m['role']:9} {m['status']:8} {scope:10} "
-                      f"pinned {ident.get('pinned_at', '?')} "
-                      f"({ident.get('pinned_via', '?')})")
-
-            invites = out.get("invites", [])
-            if invites:
-                print()
-                for i in invites:
-                    # A link names nobody until it is used, so its handle is what
-                    # identifies it — and what `cancel` takes.
-                    who = (f"link {i['invite_id']}" if i.get("kind") == "link"
-                           else f"user {i['user_id'][:12]}")
-                    print(f"pending invite  {who}  "
-                          f"group {(i['group_id'] or 'node-wide')[:8]}  "
-                          f"expires {i['expires_at']}")
-            return
-
-        # `member upload` is gone: whether uploads are accepted is `writable`
-        # on the root they would land in, not a per-group switch. Named
-        # explicitly rather than left to the usage line below, which offered a
-        # username for a verb that no longer takes one — an operator following
-        # it would have got "unknown subcommand" and no idea what replaced it.
-        if sub == "upload":
-            print("`member upload` is gone. Uploads are decided per directory "
-                  "now:")
-            print()
-            print("  meshbay-node root list                    "
-                  "# which are read-write")
-            print("  meshbay-node root set <name> --writable   "
-                  "# accept uploads there")
-            print("  meshbay-node root set <name> --no-writable  # stop them")
-            print()
-            print("A group whose directories are all read-only accepts no "
-                  "uploads at all,")
-            print("which is what turning the old switch off meant.")
-            sys.exit(1)
-
-        if not args.target:
-            print(f"usage: meshbay-node member {sub} <username>")
-            sys.exit(1)
-
-        if sub == "invite" and args.link:
-            # A link for someone who may have no account yet, bound to their
-            # address on the hub. The code and the ticket are both in it, so
-            # it goes to them and to nobody else — the CLI mails nothing.
-            group_id = _resolve_group(cfg, args.group)
-            out = _daemon_api(
-                cfg, f"/api/groups/{group_id}/invite-links?email={quote(args.target)}",
-                method="POST")
-            from meshbay_node.roster import write_code_file
-            write_code_file(cfg.data_dir, out["link"], out.get("expires_at", ""),
-                            name="invite-link")
-            print(f"INVITATION LINK  {out['link']}")
-            print(f"valid until      {out.get('expires_at', '?')}")
-            print(f"cancel with      meshbay-node member cancel {out['invite_id']}")
-            print()
-            print(f"Send it to {args.target} yourself. It works once, and only for an")
-            print("account registered with that address: they open it, create their")
-            print("account or sign in, and land in the group without typing a code.")
-            return
-
-        if sub == "invite":
-            group_id = _resolve_group(cfg, args.group)
-            out = _daemon_api(
-                cfg, f"/api/groups/{group_id}/invites?username={args.target}",
-                method="POST")
-            from meshbay_node.roster import write_code_file
-            path = write_code_file(cfg.data_dir, out["code"],
-                                   out.get("expires_at", ""), name="invite-code")
-            print(f"INVITATION CODE  {out['code']}")
-            print(f"valid until      {out.get('expires_at', '?')}")
-            print()
-            print(f"Send it to {args.target} however you normally talk. It works")
-            print("once, for that account only, and never passes through the hub.")
-            print("They enter it the first time they open the group — you do not")
-            print("need to be online then.")
-            print()
-            print(f"also written to {path}")
-            return
-
-        if sub == "cancel":
-            # Takes back a link that has not been used, on the node and the hub.
-            group_id = _resolve_group(cfg, args.group)
-            out = _daemon_api(
-                cfg, f"/api/groups/{group_id}/invite-links/{quote(args.target)}",
-                method="DELETE")
-            print(f"invitation link {args.target[:8]} cancelled")
-            if not out.get("hub", True):
-                print("The hub's half could not be reached; it expires on its own.")
-            return
-
-        # revoke and unpin both name a person; the daemon resolves the account.
-        # It tries its own roster first and falls back to the hub, so a node that
-        # pinned someone before invitations carried a name is still manageable.
-        match = _daemon_api(cfg, f"/api/resolve?username={args.target}")
-
-        if sub == "revoke":
-            group_id = _resolve_group(cfg, args.group)
-            out = _daemon_api(
-                cfg, f"/api/members/{match['user_id']}/revoke?group_id={group_id}",
-                method="POST")
-            print(f"{args.target} revoked from {group_id[:8]}")
-            if out.get("invites_dropped"):
-                print("Their unredeemed invitation was cancelled.")
-            # The node decides whether a rotation is warranted and says so in
-            # the reminder — somebody who never redeemed a code never held the
-            # key, and advising a rotation there is advice to ignore the next
-            # time it is real. Deciding it again here is the second
-            # implementation this file exists not to have.
-            if out.get("reminder"):
-                print("They stop receiving the group key on their next connection.")
-                print("They still hold the current one — rotate it:")
-                print(f"  meshbay-node gek-init --group {group_id}")
-            return
-
-        if sub == "unpin":
-            _daemon_api(cfg, f"/api/members/{match['user_id']}/unpin", method="POST")
-            print(f"{args.target} unpinned — they can pair again with a new key")
-            print(f"issue a code:  meshbay-node member invite {args.target}")
-            return
-
-        print("usage: meshbay-node member list|invite|cancel|revoke|unpin")
-        sys.exit(1)
-
-    if args.command in ("gek-init", "gek"):
-        # `gek-init` is the original spelling and still works. `gek rotate` is
-        # the one that matters after a revocation: the ex-member holds the
-        # current key and nothing else takes it from them.
-        sub = "init" if args.command == "gek-init" else (args.subcommand or "init")
-        if sub not in ("init", "rotate"):
-            print("usage: meshbay-node gek init|rotate [--group NAME]")
-            sys.exit(1)
-
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        group_id = _resolve_group(cfg, args.group)
-
-        if sub == "rotate" and not args.yes:
-            print("Rotating replaces this group's key.")
-            print("  · every member re-receives it automatically on their next connect")
-            print("  · anyone revoked keeps the OLD key and loses access to new content")
-            print("  · content already downloaded stays readable to whoever has it")
-            if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"):
-                print("cancelled")
-                return
-
-        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek"
-                               f"{'?rotate=true' if sub == 'rotate' else ''}",
-                          method="POST", timeout=60)
-
-        verb = "rotated" if out.get("rotated") else "ready"
-        print(f"GEK {verb} for {group_id}")
-        print(f"  {out.get('authorized_members', 0)} authorized member(s) — each "
-              f"receives the key on connect")
-        for err in out.get("errors") or []:
-            print(f"  ! {err}")
-        return
-
-    if args.command == "reload":
-        if sys.platform == "win32":
-            # No systemd, no SIGHUP: the daemon exposes a hot reload on its
-            # own loopback API (the same one ops.reload_config drives).
-            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-            _daemon_api(cfg, "/api/reload", method="POST")
-            print("sent reload to the running node")
-            return
-        _systemctl_user(
-            "reload", "meshbay-node",
-            not_running_hint="Node is not running as a systemd unit — start it "
-                              "with: systemctl --user start meshbay-node",
-            success="sent reload to meshbay-node",
-            watch="watch the result:  journalctl --user -u meshbay-node -f")
-        return
-
-    if args.command == "restart-daemon":
-        if sys.platform == "win32":
-            from meshbay_node.platform import (
-                autostart_end,
-                autostart_run,
-                service_end,
-                service_run,
-                service_status,
-            )
-            if service_status()["installed"]:
-                service_end()
-                service_run()
-                print("restarted the node (service task)")
-                return
-            autostart_end()   # kill whatever is running now
-            try:
-                autostart_run()
-            except RuntimeError as e:
-                print(f"Could not restart: {e}. Stop the daemon (Ctrl+C) and "
-                      "relaunch it from where meshbay-node is on PATH.")
-                sys.exit(1)
-            print("restarted the node")
-            return
-        _systemctl_user(
-            "restart", "meshbay-node",
-            not_running_hint="meshbay-node is not installed as a systemd unit — "
-                              "see packaging/systemd/",
-            success="meshbay-node restarted via systemd",
-            watch="check status:  systemctl --user status meshbay-node\n"
-                  "watch logs:    journalctl --user -u meshbay-node -f")
-        return
-
-    if args.command == "autostart":
-        from meshbay_node import platform as _plat
-        if sys.platform != "win32":
-            print("autostart is Windows-only — elsewhere use "
-                  "'systemctl --user enable --now meshbay-node'.")
-            sys.exit(1)
-        sub = args.subcommand or "status"
-        if sub == "install":
-            _plat.autostart_install()
-            print("Installed the Startup launcher — meshbay-node starts at "
-                  "each sign-in (no window, no admin).")
-            print("Start it now with: meshbay-node autostart start")
-        elif sub == "remove":
-            _plat.autostart_remove()
-            print("Removed the Startup launcher.")
-        elif sub == "start":
-            try:
-                _plat.autostart_run()
-            except RuntimeError as e:
-                print(f"Could not start: {e}")
-                sys.exit(1)
-            print("started")
-        elif sub == "stop":
-            _plat.autostart_end()
-            print("stopped")
-        elif sub == "status":
-            st = _plat.autostart_status()
-            if st["installed"]:
-                print("autostart    installed — runs meshbay-node at sign-in")
-            else:
-                print("autostart    not installed — meshbay-node autostart install")
-        else:
-            print("autostart: install | remove | start | stop | status")
-            sys.exit(1)
-        return
-
-    if args.command == "service":
-        from meshbay_node import platform as _plat
-        if sys.platform != "win32":
-            print("service mode is Windows-only — elsewhere use "
-                  "'systemctl --user enable --now meshbay-node'.")
-            sys.exit(1)
-        sub = args.subcommand or "status"
-        if sub == "install":
-            try:
-                _plat.service_install()
-            except RuntimeError as e:
-                print(f"Could not install: {e}")
-                if "denied" in str(e).lower():
-                    print("Run this from an elevated (Administrator) prompt.")
-                sys.exit(1)
-            print(f"Registered the {_plat.TASK_NAME!r} scheduled task — it starts "
-                  "meshbay-node at boot, as this user, whether or not you have "
-                  "signed in yet (no password stored).")
-            print("Start it now with: meshbay-node service start")
-        elif sub == "remove":
-            _plat.service_remove()
-            print(f"Removed the {_plat.TASK_NAME!r} scheduled task.")
-        elif sub == "start":
-            _plat.service_run()
-            print("started")
-        elif sub == "stop":
-            _plat.service_end()
-            print("stopped")
-        elif sub == "status":
-            st = _plat.service_status()
-            if st["installed"]:
-                print(f"service       installed — {st['state'] or 'unknown state'}")
-            else:
-                print("service       not installed — meshbay-node service install "
-                      "(needs an elevated prompt)")
-        else:
-            print("service: install | remove | start | stop | status")
-            sys.exit(1)
-        return
-
-    if args.command == "chat":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        sub = args.subcommand or "status"
-        group_id = _resolve_group(cfg, args.group)
-
-        if sub == "status":
-            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat")
-            print(f"encryption   always on (MNP {MNP_VERSION})")
-            print(f"epoch        {out.get('epoch', 0)}")
-            print(f"messages     {out.get('encrypted_messages', 0)} encrypted, "
-                  f"{out.get('plaintext_messages', 0)} in the clear")
-            if out.get("plaintext_messages"):
-                print("\nThose messages were written before this node spoke MNP "
-                      "2.0 and are\nstill readable off this disk. "
-                      "`chat encrypt-history` converts them.")
-            return
-
-        if sub == "rotate":
-            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/epoch",
-                              method="POST")
-            print(f"chat epoch {out['epoch']} opened")
-            print("Everyone still in the group keeps reading the history; "
-                  "whoever left\ncannot read what is written from now on.")
-            return
-
-        if sub == "encrypt-history":
-            if not args.yes:
-                print("This rewrites the only copy of this group's older "
-                      "messages.")
-                print("A backup of chat.db is taken first, beside it.")
-                if input("re-encrypt now? [y/N] ").strip().lower() not in ("y", "yes"):
-                    print("cancelled")
-                    return
-            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/encrypt-history",
-                              method="POST", timeout=300)
-            print(f"re-encrypted {out['converted']} message(s) under epoch "
-                  f"{out['epoch']}")
-            print(f"backup       {out['backup']}")
-            return
-
-        if sub == "prune":
-            days = int(args.target or 0)
-            if days < 1:
-                print("usage: meshbay-node chat prune <days> [--group G]")
-                sys.exit(1)
-            out = _daemon_api(
-                cfg, f"/api/groups/{group_id}/chat/prune?max_age_days={days}",
-                method="POST")
-            print(f"removed {out['removed']} message(s) older than {days} day(s)")
-            return
-
-        print("usage: meshbay-node chat "
-              "status|rotate|encrypt-history|prune [--group G]")
-        sys.exit(1)
-
-    if args.command == "denylist":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        sub = args.subcommand or "show"
-
-        if sub == "show":
-            out = _daemon_api(cfg, "/api/denylist")
-            total = out.get("count", 0)
-            if not total:
-                print("denylist     empty — nothing is being refused")
-                return
-            for kind in ("users", "groups", "jtis"):
-                for entry in out.get(kind, []):
-                    print(f"  {kind[:-1]:<6} {entry}")
-            print(f"\n{total} entr(y/ies). These survive a restart (finding H4).")
-            return
-
-        if sub == "clear":
-            if not args.yes:
-                what = args.target or "EVERY entry"
-                print(f"Clearing the denylist re-admits {what}.")
-                print("A revocation the hub sent will not come back on its own.")
-                if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"):
-                    print("cancelled")
-                    return
-            out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}",
-                              method="POST")
-            print(f"removed {out['removed']} entr(y/ies) ({out['subject']})")
-            return
-
-        print("usage: meshbay-node denylist show|clear [identifier] [--yes]")
-        sys.exit(1)
-
-    if args.command == "stun":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        sub = args.subcommand or "list"
-
-        if sub == "list":
-            out = _daemon_api(cfg, "/api/node-settings")
-            servers = out.get("stun_servers", [])
-            if not servers:
-                print("stun servers     (none configured)")
-                return
-            for i, s in enumerate(servers, 1):
-                print(f"  {i}. {s}")
-            return
-
-        if sub == "add":
-            url = args.target
-            if not url:
-                print("usage: meshbay-node stun add <stun:host:port>")
-                sys.exit(1)
-            if not url.startswith("stun:"):
-                print(f"error: STUN URL must start with stun: — got {url!r}")
-                sys.exit(1)
-            out = _daemon_api(cfg, "/api/node-settings")
-            servers = out.get("stun_servers", [])
-            if url in servers:
-                print(f"already present: {url}")
-                return
-            servers.append(url)
-            _daemon_api(cfg, "/api/node-settings", method="PUT",
-                        body={"stun_servers": servers})
-            print(f"added {url} ({len(servers)} servers total)")
-            return
-
-        if sub == "remove":
-            url = args.target
-            if not url:
-                print("usage: meshbay-node stun remove <stun:host:port>")
-                sys.exit(1)
-            out = _daemon_api(cfg, "/api/node-settings")
-            servers = out.get("stun_servers", [])
-            if url not in servers:
-                print(f"not found: {url}")
-                sys.exit(1)
-            servers.remove(url)
-            _daemon_api(cfg, "/api/node-settings", method="PUT",
-                        body={"stun_servers": servers})
-            print(f"removed {url} ({len(servers)} servers remaining)")
-            return
-
-        if sub == "reset":
-            from meshbay_node.config import DEFAULT_STUN_SERVERS
-            _daemon_api(cfg, "/api/node-settings", method="PUT",
-                        body={"stun_servers": list(DEFAULT_STUN_SERVERS)})
-            print("STUN servers reset to defaults:")
-            for s in DEFAULT_STUN_SERVERS:
-                print(f"  {s}")
-            return
-
-        print("usage: meshbay-node stun list|add|remove|reset [url]")
-        sys.exit(1)
-
-    if args.command == "transfers":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        sub = args.subcommand or "show"
-
-        if sub == "show":
-            out = _daemon_api(cfg, "/api/transfers")
-            for kind, pool in out.get("pools", {}).items():
-                print(f"  {kind:<9} {pool['in_use']}/{pool['cap']} in use, "
-                      f"{pool['queued']} queued   (node-wide)")
-            # From the daemon, not from node.toml: a change made on the Node
-            # page is live before it is written back, and the number to print
-            # is the one being enforced. The local file is the fallback rather
-            # than a literal, so there is no second copy of the default here.
-            settings = _daemon_api(cfg, "/api/node-settings")
-            gb = settings.get("max_upload_gb") or cfg.node.max_upload_gb
-            print(f"\n  largest single upload: {gb:g} GB per file "
-                  f"(meshbay-node transfers max-size <GB>)")
-            # Per group, because that is the cap that decides how many one
-            # person runs at once — and it is not the node-wide number. An
-            # operator raising `transfers set 8 8` and still seeing two at a
-            # time is looking at this line, which used to print the node's
-            # default and say nothing about where it came from.
-            groups = out.get("groups") or []
-            if groups:
-                print("\n  per member, per group "
-                      "(meshbay-node transfers per-member <dl> <ul> --group X):")
-                for g in groups:
-                    how = "set" if g["set"] else "default"
-                    print(f"    {g['name']:<20} {g['download']} download(s), "
-                          f"{g['upload']} upload(s)   [{how}]")
-            leases = out.get("leases", [])
-            if not leases:
-                print("\n  nothing transferring")
-                return
-            print(f"\n  {'transfer':<14}{'kind':<10}{'state':<9}"
-                  f"{'user':<12}{'bytes':>12}")
-            for x in leases:
-                where = f"  (#{x['ahead'] + 1} in queue)" if x["state"] == "queued" else ""
-                print(f"  {x['tr']:<14}{x['kind']:<10}{x['state']:<9}"
-                      f"{x['user_id'][:10]:<12}{x['bytes']:>12}{where}")
-            return
-
-        if sub == "set":
-            # `transfers set 4 2` — downloads, then uploads. Node-wide; the
-            # per-member cap is a group's setting and is signed, so it is not
-            # settable from here (see `ops.set_transfer_limits`).
-            values = [v for v in (args.target, args.value) if v]
-            if len(values) != 2:
-                print("usage: meshbay-node transfers set <downloads> <uploads>")
-                sys.exit(1)
-            try:
-                downloads, uploads = int(values[0]), int(values[1])
-            except ValueError:
-                print("error: both values must be whole numbers")
-                sys.exit(1)
-            if downloads < 1 or uploads < 1:
-                print("error: a cap below 1 is not 'unlimited'; it would stop "
-                      "every transfer. Revoke the member instead.")
-                sys.exit(1)
-            out = _daemon_api(cfg, "/api/node-settings", method="PUT",
-                              body={"max_concurrent_downloads": downloads,
-                                    "max_concurrent_uploads": uploads})
-            print(f"downloads: {downloads}, uploads: {uploads} "
-                  f"(applied now, and kept in node.toml)")
-            return
-
-        if sub == "max-size":
-            # The largest single file a member may upload here. Not a
-            # concurrency cap like `set` — it is the one limit that bounds what
-            # a member writes to the operator's disk, which is why it lives
-            # beside them rather than under a verb of its own.
-            if not args.target:
-                print("usage: meshbay-node transfers max-size <GB>")
-                sys.exit(1)
-            try:
-                gb = float(args.target)
-            except ValueError:
-                print("error: the size must be a number of GB (e.g. 8, or 0.5)")
-                sys.exit(1)
-            if gb <= 0:
-                print("error: a ceiling of zero is not 'unlimited'; it would "
-                      "refuse every upload. Make the root read-only instead.")
-                sys.exit(1)
-            _daemon_api(cfg, "/api/node-settings", method="PUT",
-                        body={"max_upload_gb": gb})
-            print(f"largest single upload: {gb:g} GB per file "
-                  f"(applied now, and kept in node.toml)")
-            return
-
-        if sub == "per-member":
-            # How many transfers ONE member may run at once in this group. Not
-            # the same knob as `set`, which is the machine's total — and the
-            # reason "I set 8 8 and still only get two" is the commonest
-            # confusion here: per-member is checked first, by design.
-            values = [v for v in (args.target, args.value) if v]
-            if len(values) != 2:
-                print("usage: meshbay-node transfers per-member <downloads> "
-                      "<uploads> [--group NAME]")
-                sys.exit(1)
-            try:
-                downloads, uploads = int(values[0]), int(values[1])
-            except ValueError:
-                print("error: both values must be whole numbers")
-                sys.exit(1)
-            if downloads < 1 or uploads < 1:
-                print("error: a cap below 1 is not 'unlimited'; it would stop "
-                      "every transfer for that member. Revoke them instead.")
-                sys.exit(1)
-            group_id = _resolve_group(cfg, args.group)
-            out = _daemon_api(cfg, f"/api/groups/{group_id}/transfer-limits",
-                              method="PUT",
-                              body={"downloads": downloads, "uploads": uploads})
-            got = out.get("limits", {})
-            started = out.get("started") or []
-            print(f"each member of this group may now run "
-                  f"{got.get('download')} download(s) and "
-                  f"{got.get('upload')} upload(s) at once")
-            if started:
-                print(f"{len(started)} waiting transfer(s) started at once")
-            return
-
-        print("usage: meshbay-node transfers show|set <downloads> <uploads>|"
-              "max-size <GB>|per-member <downloads> <uploads> [--group NAME]")
-        sys.exit(1)
-
-    if args.command == "file":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        sub = args.subcommand or "list"
-        group_id = _resolve_group(cfg, args.group)
-
-        if sub == "list":
-            out = _daemon_api(cfg, f"/api/groups/{group_id}/files")
-            files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"]))
-            if not files:
-                print("no files indexed")
-                return
-            for f in files:
-                print(f"  {f['id'][:12]}  {f['size']:>12}  {f['path']}/{f['name']}")
-            print(f"\n{len(files)} file(s). Remove one with: "
-                  f"meshbay-node file rm <id>")
-            return
-
-        if sub == "rm":
-            # Milestone 14.11 — the last operator action that needed a browser.
-            if not args.target:
-                print("usage: meshbay-node file rm <file-id> [--group NAME]")
-                sys.exit(1)
-            out = _daemon_api(cfg, f"/api/groups/{group_id}/files",)
-            matches = [f for f in out.get("files", [])
-                       if f["id"].startswith(args.target)]
-            if not matches:
-                print(f"no file whose id starts with {args.target!r}")
-                sys.exit(1)
-            if len(matches) > 1:
-                print(f"{args.target!r} matches {len(matches)} files — be more specific:")
-                for f in matches[:10]:
-                    print(f"  {f['id'][:16]}  {f['path']}/{f['name']}")
-                sys.exit(1)
-            target = matches[0]
-            if not args.yes:
-                print(f"Delete {target['path']}/{target['name']} "
-                      f"({target['size']} bytes) from disk?")
-                print("This removes the file itself, not just the listing.")
-                if input("delete? [y/N] ").strip().lower() not in ("y", "yes"):
-                    print("cancelled")
-                    return
-            _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}",
-                        method="DELETE")
-            print(f"deleted {target['path']}/{target['name']}")
-            return
-
-        print("usage: meshbay-node file list|rm <id> [--group NAME] [--yes]")
-        sys.exit(1)
-
-    if args.command == "video":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        if (args.subcommand or "") != "rematch":
-            print("usage: meshbay-node video rematch [--group NAME] [--yes]")
-            sys.exit(1)
-        group_id = _resolve_group(cfg, args.group)
-        if not args.yes:
-            print("Re-resolve every automatic TMDB match for this group's videos?")
-            print("Manual 'Fix match' corrections are kept. Re-resolution is lazy —")
-            print("each poster re-queries TMDB the next time it is opened.")
-            if input("proceed? [y/N] ").strip().lower() not in ("y", "yes"):
-                print("cancelled")
-                return
-        out = _daemon_api(cfg, f"/api/groups/{group_id}/video/rematch", method="POST")
-        print(f"cleared {out.get('removed', 0)} automatic match(es) "
-              f"across {out.get('videos', 0)} video file(s)")
-        return
-
-    if args.command == "group":
-        if args.subcommand in (None, "list"):
-            # Milestone 14.2.
-            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-            out = _daemon_api(cfg, "/api/groups")
-            groups = out.get("groups", [])
-            if not groups:
-                print("no groups hosted — add one with: "
-                      "meshbay-node group add <name> --dir <path>")
-                return
-            for g in groups:
-                key = "GEK" if g.get("has_gek") else "NO KEY"
-                print(f"  {g['name']}  [{g['visibility']}/{g.get('join_policy')}]  "
-                      f"{key}  {g['file_count']} file(s)  "
-                      f"{g.get('peers', 0)} peer(s)")
-                print(f"      {g['id']}")
-                for r in g.get("roots", []):
-                    flags = []
-                    if r.get("writable"):
-                        flags.append("rw")
-                    else:
-                        flags.append("ro")
-                    if r.get("removable"):
-                        flags.append("removable")
-                    if r.get("ejected"):
-                        flags.append("ejected")
-                    flag_str = f" ({', '.join(flags)})" if flags else ""
-                    live = "" if r.get("available", True) else "  [UNAVAILABLE]"
-                    print(f"      root {r['name']}{flag_str}{live}")
-                if not g.get("has_gek"):
-                    print(f"      give it a key:  meshbay-node gek init "
-                          f"--group {g['name']}")
-            return
-
-        if args.subcommand == "remove":
-            if not args.target:
-                print("usage: meshbay-node group remove <name>")
-                sys.exit(1)
-            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-            if not args.yes:
-                answer = input(f"Remove group '{args.target}' from this node? [y/N] ")
-                if answer.lower() not in ("y", "yes"):
-                    print("cancelled")
-                    return
-            out = _daemon_api(cfg, "/api/groups/detach", method="POST",
-                              body={"name": args.target})
-            print(f"{out['name']} ({out['group_id'][:8]}) removed from {out['config']}")
-            print()
-            print("Restart the daemon to stop hosting it:")
-            print("  meshbay-node restart-daemon")
-            return
-
-        if args.subcommand != "add":
-            print("usage: meshbay-node group list|add|remove <name>")
-            sys.exit(1)
-        if not args.target or not args.dir:
-            print("usage: meshbay-node group add <name> --dir <path> "
-                  "[--no-writable]")
-            print()
-            print("The group must already exist on the hub and be yours. This")
-            print("only tells the node to host it, and picks its first")
-            print("directory, which accepts uploads unless --no-writable.")
-            print("Add more with:  meshbay-node root add <path> [--writable]")
-            sys.exit(1)
-
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        # Writable unless the operator says otherwise: a brand-new group that
-        # cannot receive a single file until its owner finds a second command
-        # is not a working group. Every root added *later* is read-only by
-        # default, which is the opposite rule and the right one there.
-        writable = args.writable is not False
-        body = {"name": args.target, "shared_dir": args.dir,
-                "writable": writable}
-        out = _daemon_api(cfg, "/api/groups/attach", method="POST", body=body)
-        print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}")
-        print(f"  shared_dir   {out['shared_dir']}"
-              f"  ({'read-write' if writable else 'read-only'})")
-        print()
-        print("Tell the daemon to re-read its config, then give the group a key:")
-        print("  meshbay-node reload")
-        print(f"  meshbay-node gek init --group {out['name']}")
-        print()
-        print("The key is this group's own — members of your other groups cannot")
-        print("read it, and joining one says nothing about the other.")
-        return
-
-    if args.command == "root":
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        sub = args.subcommand or "list"
-        group_id = _resolve_group(cfg, args.group)
-
-        if sub == "list":
-            out = _daemon_api(cfg, "/api/groups")
-            group = next((g for g in out.get("groups", [])
-                          if g["id"] == group_id), None)
-            if not group:
-                print(f"group {group_id[:8]} not hosted on this node")
-                sys.exit(1)
-            roots = group.get("roots", [])
-            if not roots:
-                print("no roots configured")
-                print(f"add one:  meshbay-node root add /path/to/dir --group {group_id}")
-                return
-            for r in roots:
-                flags = []
-                if r.get("writable"):
-                    flags.append("rw")
-                else:
-                    flags.append("ro")
-                if r.get("removable"):
-                    flags.append("removable")
-                if r.get("ejected"):
-                    flags.append("EJECTED")
-                avail = "available" if r.get("available", True) else "UNAVAILABLE"
-                flags.append(avail)
-                print(f"  {r['name']:<20} {', '.join(flags)}")
-                print(f"    {r.get('path', '?')}")
-            return
-
-        if sub == "add":
-            path = args.target
-            if not path:
-                print("usage: meshbay-node root add <path> [--name NAME] "
-                      "[--writable] [--removable] [--group NAME]")
-                sys.exit(1)
-            body = {
-                "path": path,
-                "name": args.name or Path(path).name,
-                "writable": args.writable if args.writable is not None else True,
-                "removable": bool(args.removable),
-            }
-            _daemon_api(cfg, f"/api/groups/{group_id}/roots",
-                        method="POST", body=body)
-            w = "rw" if body["writable"] else "ro"
-            rm = ", removable" if body["removable"] else ""
-            print(f"added root {body['name']} → {path} ({w}{rm})")
-            print("reload the daemon to start indexing:")
-            print("  meshbay-node reload")
-            return
-
-        if sub == "remove":
-            name = args.target
-            if not name:
-                print("usage: meshbay-node root remove <name> [--group NAME]")
-                sys.exit(1)
-            if not args.yes:
-                print(f"Remove root '{name}' from group {group_id[:8]}?")
-                print("Files on disk are untouched; only the node config changes.")
-                if input("remove? [y/N] ").strip().lower() not in ("y", "yes"):
-                    print("cancelled")
-                    return
-            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}",
-                        method="DELETE")
-            print(f"removed root {name}")
-            print("reload the daemon to apply:")
-            print("  meshbay-node reload")
-            return
-
-        if sub == "set":
-            name = args.target
-            if not name:
-                print("usage: meshbay-node root set <name> "
-                      "[--writable|--no-writable] "
-                      "[--removable|--no-removable] [--group NAME]")
-                sys.exit(1)
-            body = {}
-            if args.writable is not None:
-                body["writable"] = args.writable
-            if args.removable is not None:
-                body["removable"] = args.removable
-            if not body:
-                print("nothing to change — pass --writable/--no-writable "
-                      "or --removable/--no-removable")
-                sys.exit(1)
-            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}",
-                        method="PATCH", body=body)
-            changes = ", ".join(f"{k}={v}" for k, v in body.items())
-            print(f"updated root {name}: {changes}")
-            return
-
-        if sub == "eject":
-            name = args.target
-            if not name:
-                print("usage: meshbay-node root eject <name> [--group NAME]")
-                sys.exit(1)
-            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/eject",
-                        method="PUT")
-            print(f"ejected root {name} — files are hidden until plugged back")
-            return
-
-        if sub == "plug":
-            name = args.target
-            if not name:
-                print("usage: meshbay-node root plug <name> [--group NAME]")
-                sys.exit(1)
-            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/plug",
-                        method="PUT")
-            print(f"plugged root {name} — files are visible again")
-            return
-
-        print("usage: meshbay-node root list|add|remove|set|eject|plug [name] "
-              "[--group NAME]")
-        sys.exit(1)
-
-    if args.command == "operator":
-        if args.subcommand != "pair":
-            print("usage: meshbay-node operator pair")
-            sys.exit(1)
-        if args.group:
-            # Silently ignoring it invited the reading that a code belongs to a
-            # group, and then that pairing had not worked because the group did
-            # not change.
-            print("operator pair takes no --group: pairing is node-wide.")
-            print("One paired browser can invite to, and delete files in, every")
-            print("group this node hosts.")
-            sys.exit(1)
-
-        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        out = _daemon_api(cfg, "/api/operator/pair", method="POST")
-
-        from meshbay_node.roster import write_code_file
-        path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", ""))
-
-        print(f"PAIRING CODE   {out['code']}")
-        print(f"valid until    {out.get('expires_at', '?')}")
-        print()
-        print("Sign in to the web app as this node's operator, open one of your")
-        print("groups, go to the Members tab and enter the code there.")
-        print("It works once, for that account only, and authorizes invites and")
-        print("file deletion from that browser.")
-        print()
-        print(f"also written to {path}")
+    args = start()
+    if run(args):
         return
 
     cfg = load_config(args.config or DEFAULT_CONFIG_PATH)