diff --git a/cli/README.md b/cli/README.md
new file mode 100644
index 0000000..741fb30
--- /dev/null
+++ b/cli/README.md
@@ -0,0 +1,121 @@
+# zkac-node CLI
+
+Command-line interface for [ZKAC](../README.md) using the **Python bindings only** (`zkac` package). It runs a **registry-capable server** (management + client-managed registries + optional issuance relay) and **per-user** material under `~/.zkac/` (or `$ZKAC_HOME`).
+
+## Prerequisites
+
+- Python ≥ 3.9  
+- The **`zkac`** extension built and installed from the repository root, for example:
+
+  ```bash
+  cd /path/to/ZKAC
+  maturin develop
+  # or: pip install -e .
+  ```
+
+## Installation
+
+```bash
+cd /path/to/ZKAC/cli
+pip install -e .
+```
+
+This installs the **`zkac-node`** console script.
+
+## Environment
+
+| Variable    | Meaning |
+|------------|---------|
+| `ZKAC_HOME` | Base directory for users (default: `~/.zkac`). Each user lives at `$ZKAC_HOME/<userid>/`. |
+
+## Server vs client
+
+- **Server** (`zkac-node serve`): a node that can **accept registry create/update** from an operator with the **`zkac.mgmt`** credential. It also serves **managed** sessions (BBS+ auth against stored client-managed registries) and optionally a **relay** port for blind issuance queues.
+- **Client**: a **userid** with files under `$ZKAC_HOME/<userid>/` (transport key, registries, credentials).
+
+## Ports (defaults)
+
+| Port role   | Default | Purpose |
+|------------|---------|---------|
+| Management | 7400    | ZKAC + static role `zkac.mgmt`; JSON commands (create/update registry, issuance peek/grant). |
+| Managed    | 7401    | ZKAC + `RegistryManager`; member proves a role in a client-managed registry. |
+| Relay      | 7402    | Optional **plaintext** JSON line protocol for enqueue/poll of issuance requests. Use `--relay-port 0` on `serve` to disable. Binds with `--relay-bind` (default `127.0.0.1`). |
+
+## Layout on disk
+
+**Per user** (`$ZKAC_HOME/<userid>/`):
+
+- `transport.json` — Ristretto **client** transport keypair (`zkac.Keypair`).
+- `profile.json` — `userid` and metadata.
+- `registries/<slug>/` — one directory per logical registry:
+  - `admin.json` / `registry.json` — produced by `registry-init` (admin issuer material + public state + state cert).
+  - `roles/<name>.json` — member credential payloads for `connect`.
+  - `issued/` — files from `issue-member` (handoff).
+- `pending/<request_id>.json` — saved by `issuance-request` until `issuance-poll` finalizes.
+- `servers/<label>/server.json` — from `pin-server` (pinned server pubkey + host/ports).
+
+**Server data** (`serve --data <dir>`):
+
+- `transport.json` — server transport keypair.
+- `mgmt_issuer.json` — BBS issuer secret for the static `zkac.mgmt` role only.
+- `mgmt_member.json` — member credential for `zkac.mgmt` (give to operators who push registries).
+- `registry_events.json` — append-only log to rebuild `RegistryManager` after restart (registry state; **not** the issuance queue).
+
+## Commands
+
+Run `zkac-node --help` and `zkac-node <command> --help` for full flags.
+
+| Command | Description |
+|---------|-------------|
+| `serve` | Start server; `--init` creates keys and mgmt credential in `--data`. |
+| `user-create` | Create a new userid directory with a client transport key. |
+| `pin-server` | Save server pubkey + host/ports for a user (from server’s `transport.json`). |
+| `registry-init` | Offline: build a client-managed registry (`RegistryState` + state cert) under the user’s `registries/<slug>/`. |
+| `registry-push` | Push create (or `--update`) to the server over the **management** port using `mgmt_member.json` from the server data dir. |
+| `registry-import-public` | Copy `registry.json` from another user (public metadata for peers). |
+| `connect` | Open a **managed** ZKAC session and send a JSON command (`--command`, default `whoami`). |
+| `issue-member` | Admin: blind-issue a role credential to a file under `issued/` (out-of-band handoff). |
+| `import-credential` | Install a member JSON into `roles/<name>.json`. |
+| `issuance-request` | Enqueue a blind commitment on the **relay** (see caveats below). |
+| `issuance-grant` | Admin: list pending via mgmt peek, `issue_blind`, `grant_credential`. |
+| `issuance-poll` | Poll relay for blind signature; finalize credential into `roles/<role>.json` when ready. |
+
+## Typical flows
+
+**Operator: first-time server**
+
+```bash
+export ZKAC_HOME=~/.zkac   # optional
+zkac-node serve --data ./myserver --init --mgmt-port 7400 --managed-port 7401 --relay-port 7402
+```
+
+Distribute **`myserver/transport.json`** (public key) and **`myserver/mgmt_member.json`** (sensitive) to admins who push registries.
+
+**Admin: create registry offline and push**
+
+```bash
+zkac-node user-create alice
+zkac-node registry-init --user alice --slug demo --roles analyst reader
+zkac-node registry-push --user alice --slug demo --server-data ./myserver --host 127.0.0.1 --mgmt-port 7400
+```
+
+**Member: pin server, import public registry + credential, connect**
+
+```bash
+zkac-node pin-server --user bob --name prod --transport-file ./myserver/transport.json \
+  --host 10.0.0.1 --mgmt-port 7400 --managed-port 7401 --relay-port 7402
+zkac-node registry-import-public --from-user alice --to-user bob --slug demo
+# copy or issue credential file into bob/registries/demo/roles/analyst.json
+zkac-node connect --user bob --slug demo --credential roles/analyst.json \
+  --server-file ~/.zkac/bob/servers/prod/server.json --host 10.0.0.1 --managed-port 7401
+```
+
+## Caveats
+
+1. **Issuance relay queue is in-memory.** Pending grants are lost if `serve` restarts before `issuance-grant` / `issuance-poll` complete. Registry snapshots are persisted via `registry_events.json`; the relay queue is not.
+2. **Relay is a localhost-oriented, plaintext JSON line protocol** (not the full E2E design in the Rust layer). Use `--relay-bind 127.0.0.1` and do not expose the relay port untrusted networks without additional protection.
+3. **`zkac.mgmt`** is a **separate** BBS role from the client-managed registry admin; it only authorizes **management RPCs** to the server (push registry, peek/grant issuance). Registry state integrity still comes from BBS+ state certificates inside `RegistryManager`.
+
+## Development
+
+The CLI package name is **`zkac-node-cli`** (see `pyproject.toml`). It declares no PyPI dependency on `zkac`; install the `zkac` wheel from the parent project first.
diff --git a/cli/pyproject.toml b/cli/pyproject.toml
new file mode 100644
index 0000000..ab97ed7
--- /dev/null
+++ b/cli/pyproject.toml
@@ -0,0 +1,17 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "zkac-node-cli"
+version = "0.1.0"
+description = "ZKAC node CLI (server + client) using the zkac Python bindings"
+readme = "README.md"
+requires-python = ">=3.9"
+dependencies = []
+
+[project.scripts]
+zkac-node = "zkac_cli.main:main"
+
+[tool.hatch.build.targets.wheel]
+packages = ["zkac_cli"]
diff --git a/cli/zkac_cli/__init__.py b/cli/zkac_cli/__init__.py
new file mode 100644
index 0000000..d7e6f28
--- /dev/null
+++ b/cli/zkac_cli/__init__.py
@@ -0,0 +1,3 @@
+"""ZKAC node CLI — server (registry-capable) and client commands."""
+
+__version__ = "0.1.0"
diff --git a/cli/zkac_cli/__pycache__/__init__.cpython-314.pyc b/cli/zkac_cli/__pycache__/__init__.cpython-314.pyc
new file mode 100644
index 0000000..04e06b0
Binary files /dev/null and b/cli/zkac_cli/__pycache__/__init__.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/client_ops.cpython-314.pyc b/cli/zkac_cli/__pycache__/client_ops.cpython-314.pyc
new file mode 100644
index 0000000..3ac448c
Binary files /dev/null and b/cli/zkac_cli/__pycache__/client_ops.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/creds.cpython-314.pyc b/cli/zkac_cli/__pycache__/creds.cpython-314.pyc
new file mode 100644
index 0000000..d464f06
Binary files /dev/null and b/cli/zkac_cli/__pycache__/creds.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/issuance_util.cpython-314.pyc b/cli/zkac_cli/__pycache__/issuance_util.cpython-314.pyc
new file mode 100644
index 0000000..405ed7a
Binary files /dev/null and b/cli/zkac_cli/__pycache__/issuance_util.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/main.cpython-314.pyc b/cli/zkac_cli/__pycache__/main.cpython-314.pyc
new file mode 100644
index 0000000..389222e
Binary files /dev/null and b/cli/zkac_cli/__pycache__/main.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/paths.cpython-314.pyc b/cli/zkac_cli/__pycache__/paths.cpython-314.pyc
new file mode 100644
index 0000000..6d77798
Binary files /dev/null and b/cli/zkac_cli/__pycache__/paths.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/registry_local.cpython-314.pyc b/cli/zkac_cli/__pycache__/registry_local.cpython-314.pyc
new file mode 100644
index 0000000..9ed2013
Binary files /dev/null and b/cli/zkac_cli/__pycache__/registry_local.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/registry_log.cpython-314.pyc b/cli/zkac_cli/__pycache__/registry_log.cpython-314.pyc
new file mode 100644
index 0000000..fe4ca07
Binary files /dev/null and b/cli/zkac_cli/__pycache__/registry_log.cpython-314.pyc differ
diff --git a/cli/zkac_cli/__pycache__/server_app.cpython-314.pyc b/cli/zkac_cli/__pycache__/server_app.cpython-314.pyc
new file mode 100644
index 0000000..b5d84d8
Binary files /dev/null and b/cli/zkac_cli/__pycache__/server_app.cpython-314.pyc differ
diff --git a/cli/zkac_cli/client_ops.py b/cli/zkac_cli/client_ops.py
new file mode 100644
index 0000000..183cd61
--- /dev/null
+++ b/cli/zkac_cli/client_ops.py
@@ -0,0 +1,64 @@
+"""TCP clients for management, managed, and relay channels."""
+
+from __future__ import annotations
+
+import json
+import socket
+from typing import Any
+
+import zkac
+from zkac.tcp import FramedSession, client_handshake, client_handshake_managed
+
+
+def mgmt_call(
+    host: str,
+    port: int,
+    server_pk: zkac.PublicKey,
+    mgmt_cred: zkac.Credential,
+    cmd: dict[str, Any],
+) -> dict[str, Any]:
+    node = zkac.Node(zkac.Keypair())
+    sock = socket.create_connection((host, port))
+    try:
+        session = client_handshake(sock, node, server_pk, mgmt_cred)
+        framed = FramedSession(sock, session)
+        framed.send(json.dumps(cmd).encode("utf-8"))
+        return json.loads(framed.recv().decode("utf-8"))
+    finally:
+        sock.close()
+
+
+def managed_call(
+    host: str,
+    port: int,
+    server_pk: zkac.PublicKey,
+    cred: zkac.Credential,
+    registry_id: bytes,
+    cmd: dict[str, Any],
+) -> dict[str, Any]:
+    node = zkac.Node(zkac.Keypair())
+    sock = socket.create_connection((host, port))
+    try:
+        session = client_handshake_managed(sock, node, server_pk, cred, registry_id)
+        framed = FramedSession(sock, session)
+        framed.send(json.dumps(cmd).encode("utf-8"))
+        return json.loads(framed.recv().decode("utf-8"))
+    finally:
+        sock.close()
+
+
+def relay_line(host: str, port: int, obj: dict[str, Any]) -> dict[str, Any]:
+    payload = (json.dumps(obj) + "\n").encode("utf-8")
+    sock = socket.create_connection((host, port))
+    try:
+        sock.sendall(payload)
+        buf = b""
+        while b"\n" not in buf:
+            chunk = sock.recv(4096)
+            if not chunk:
+                break
+            buf += chunk
+        line = buf.split(b"\n", 1)[0]
+        return json.loads(line.decode("utf-8"))
+    finally:
+        sock.close()
diff --git a/cli/zkac_cli/creds.py b/cli/zkac_cli/creds.py
new file mode 100644
index 0000000..1d903f9
--- /dev/null
+++ b/cli/zkac_cli/creds.py
@@ -0,0 +1,84 @@
+"""Serialize / restore BBS+ member credentials and transport keys."""
+
+from __future__ import annotations
+
+import base64
+import json
+from pathlib import Path
+from typing import Any
+
+import zkac
+
+
+def save_json(path: Path, obj: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(obj, indent=2), encoding="utf-8")
+
+
+def load_json(path: Path) -> Any:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def member_payload(
+    blind_sig: bytes,
+    req: zkac.BlindRequest,
+    role_id: bytes,
+    epoch: int,
+    issuer_pk: zkac.BbsPublicKey,
+) -> dict[str, Any]:
+    return {
+        "role_id_hex": role_id.hex(),
+        "epoch": epoch,
+        "blind_sig_b64": base64.b64encode(blind_sig).decode(),
+        "member_secret_b64": base64.b64encode(req.member_secret()).decode(),
+        "prover_blind_b64": base64.b64encode(req.prover_blind()).decode(),
+        "issuer_public_key_b64": base64.b64encode(issuer_pk.to_bytes()).decode(),
+    }
+
+
+def load_member_credential(data: dict[str, Any]) -> zkac.Credential:
+    pk = zkac.BbsPublicKey.from_bytes(base64.b64decode(data["issuer_public_key_b64"]))
+    rid = bytes.fromhex(data["role_id_hex"])
+    return zkac.Credential.finalize(
+        base64.b64decode(data["blind_sig_b64"]),
+        base64.b64decode(data["member_secret_b64"]),
+        base64.b64decode(data["prover_blind_b64"]),
+        rid,
+        int(data["epoch"]),
+        pk,
+    )
+
+
+def save_transport_keypair(path: Path, kp: zkac.Keypair) -> None:
+    save_json(
+        path,
+        {
+            "secret_key_b64": base64.b64encode(kp.secret_key_bytes()).decode(),
+            "public_key_b64": base64.b64encode(kp.public_key().to_bytes()).decode(),
+        },
+    )
+
+
+def load_transport_keypair(path: Path) -> zkac.Keypair:
+    t = load_json(path)
+    return zkac.Keypair.from_secret_key(base64.b64decode(t["secret_key_b64"]))
+
+
+def save_server_transport(path: Path, kp: zkac.Keypair) -> None:
+    save_json(
+        path,
+        {
+            "server_secret_key_b64": base64.b64encode(kp.secret_key_bytes()).decode(),
+            "server_public_key_b64": base64.b64encode(kp.public_key().to_bytes()).decode(),
+        },
+    )
+
+
+def load_server_public_key(path: Path) -> zkac.PublicKey:
+    t = load_json(path)
+    return zkac.PublicKey.from_bytes(base64.b64decode(t["server_public_key_b64"]))
+
+
+def load_server_keypair(path: Path) -> zkac.Keypair:
+    t = load_json(path)
+    return zkac.Keypair.from_secret_key(base64.b64decode(t["server_secret_key_b64"]))
diff --git a/cli/zkac_cli/issuance_util.py b/cli/zkac_cli/issuance_util.py
new file mode 100644
index 0000000..f59eb97
--- /dev/null
+++ b/cli/zkac_cli/issuance_util.py
@@ -0,0 +1,26 @@
+"""Helpers around RegistryManager issuance queues (Python API drains on take)."""
+
+from __future__ import annotations
+
+import zkac
+
+
+def peek_pending_requests(
+    mgr: zkac.RegistryManager, registry_id: bytes
+) -> list[tuple[bytes, bytes, bytes, bytes]]:
+    """Return pending (request_id, role_id, eph_pk, ciphertext) and re-queue them."""
+    items = mgr.take_pending_requests(registry_id)
+    out: list[tuple[bytes, bytes, bytes, bytes]] = []
+    for tup in items:
+        req_id, role_id, eph_pk, enc = tup
+        out.append((req_id, role_id, eph_pk, enc))
+        mgr.queue_issuance_request(registry_id, req_id, role_id, eph_pk, enc)
+    return out
+
+
+def take_pending_requests(
+    mgr: zkac.RegistryManager, registry_id: bytes
+) -> list[tuple[bytes, bytes, bytes, bytes]]:
+    """Drain pending queue (admin processes and must grant or lose requests)."""
+    items = mgr.take_pending_requests(registry_id)
+    return [(a, b, c, d) for a, b, c, d in items]
diff --git a/cli/zkac_cli/main.py b/cli/zkac_cli/main.py
new file mode 100644
index 0000000..0a3e003
--- /dev/null
+++ b/cli/zkac_cli/main.py
@@ -0,0 +1,416 @@
+"""zkac-node CLI entrypoint."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import secrets
+import shutil
+import sys
+from pathlib import Path
+
+import zkac
+
+from zkac_cli import creds
+from zkac_cli import client_ops
+from zkac_cli import paths
+from zkac_cli import registry_local
+from zkac_cli import server_app
+
+
+MGMT_ROLE = server_app.MGMT_ROLE
+
+
+def cmd_serve(args: argparse.Namespace) -> None:
+    data = Path(args.data).resolve()
+    if args.init or not (data / "transport.json").is_file():
+        data.mkdir(parents=True, exist_ok=True)
+        transport = zkac.Keypair()
+        issuer = zkac.BbsIssuer()
+        pk = issuer.public_key()
+        mg_rid = zkac.role_id(MGMT_ROLE)
+        req = zkac.prepare_blind_request()
+        sig = issuer.issue_blind(req.commitment_with_proof(), mg_rid, 1)
+        _ = zkac.Credential.finalize(
+            sig, req.member_secret(), req.prover_blind(), mg_rid, 1, pk
+        )
+        creds.save_server_transport(data / "transport.json", transport)
+        creds.save_json(
+            data / "mgmt_issuer.json",
+            {"issuer_secret_b64": base64.b64encode(issuer.secret_key_bytes()).decode()},
+        )
+        m = creds.member_payload(sig, req, mg_rid, 1, pk)
+        creds.save_json(data / "mgmt_member.json", m)
+        print(f"Initialized server data in {data}", flush=True)
+        print(
+            "Copy mgmt_member.json for operators; distribute server_public_key from transport.json",
+            flush=True,
+        )
+    relay = None if args.relay_port == 0 else args.relay_port
+    relay_bind = args.relay_bind
+    print(
+        f"Starting: mgmt={args.mgmt_port} managed={args.managed_port} relay={relay!r} bind={relay_bind}",
+        flush=True,
+    )
+    server_app.serve(data, args.mgmt_port, args.managed_port, relay, relay_bind)
+
+
+def cmd_user_create(args: argparse.Namespace) -> None:
+    uid = args.userid
+    d = paths.ensure_user(uid)
+    kp = zkac.Keypair()
+    creds.save_transport_keypair(d / "transport.json", kp)
+    creds.save_json(d / "profile.json", {"userid": uid})
+    print(f"User {uid!r}: wrote {d / 'transport.json'}")
+
+
+def cmd_registry_init(args: argparse.Namespace) -> None:
+    base = paths.ensure_user(args.user)
+    out = (base / "registries" / args.slug).resolve()
+    r = registry_local.create_registry_bundle(args.slug, list(args.roles), out)
+    print(json.dumps(r, indent=2))
+
+
+def cmd_registry_push(args: argparse.Namespace) -> None:
+    server_data = Path(args.server_data).resolve()
+    reg_dir = (paths.user_dir(args.user) / "registries" / args.slug).resolve()
+    reg = creds.load_json(reg_dir / "registry.json")
+    mg = creds.load_json(server_data / "mgmt_member.json")
+    t = creds.load_json(server_data / "transport.json")
+    server_pk = zkac.PublicKey.from_bytes(base64.b64decode(t["server_public_key_b64"]))
+    mgmt_cred = creds.load_member_credential(mg)
+    cmd = {"cmd": "create_registry", "state_b64": reg["state_bytes_b64"], "state_cert_b64": reg["state_cert_b64"]}
+    if args.update:
+        cmd = {"cmd": "update_registry", "state_b64": reg["state_bytes_b64"], "state_cert_b64": reg["state_cert_b64"]}
+    out = client_ops.mgmt_call(args.host, args.mgmt_port, server_pk, mgmt_cred, cmd)
+    print(json.dumps(out, indent=2))
+
+
+def cmd_connect(args: argparse.Namespace) -> None:
+    reg_dir = (paths.user_dir(args.user) / "registries" / args.slug).resolve()
+    reg_path = reg_dir / "registry.json"
+    if reg_path.is_file():
+        reg = creds.load_json(reg_path)
+    elif args.registry_id:
+        reg = {"registry_id_hex": args.registry_id}
+    else:
+        print(
+            "Missing registry.json. Copy from the admin (public) or pass --registry-id.",
+            file=sys.stderr,
+        )
+        raise SystemExit(1)
+    cred_path = (reg_dir / args.credential).resolve()
+    m = creds.load_json(cred_path)
+    credential = creds.load_member_credential(m)
+    server = creds.load_json(Path(args.server_file).expanduser().resolve())
+    server_pk = zkac.PublicKey.from_bytes(base64.b64decode(server["server_public_key_b64"]))
+    registry_id = bytes.fromhex(reg["registry_id_hex"])
+    body = client_ops.managed_call(
+        args.host,
+        args.managed_port,
+        server_pk,
+        credential,
+        registry_id,
+        {"cmd": args.command},
+    )
+    print(json.dumps(body, indent=2))
+
+
+def cmd_issuance_request(args: argparse.Namespace) -> None:
+    """Queue a blind issuance request (payload is commitment bytes; localhost relay only)."""
+    reg = creds.load_json(paths.user_dir(args.user) / "registries" / args.slug / "registry.json")
+    issuance_pk = base64.b64decode(reg["issuance_public_key_b64"])
+    req = zkac.prepare_blind_request()
+    payload = req.commitment_with_proof()
+    req_id = secrets.token_bytes(32)
+    eph_pk = secrets.token_bytes(32)
+    # Store locally for finalize
+    pending = {
+        "request_id_hex": req_id.hex(),
+        "role_name": args.role,
+        "member_secret_b64": base64.b64encode(req.member_secret()).decode(),
+        "prover_blind_b64": base64.b64encode(req.prover_blind()).decode(),
+        "issuer_public_key_b64": reg["admin_issuer_public_key_b64"],
+        "epoch": 1,
+    }
+    pdir = paths.user_dir(args.user) / "pending"
+    pdir.mkdir(parents=True, exist_ok=True)
+    creds.save_json(pdir / f"{req_id.hex()}.json", pending)
+
+    obj = {
+        "cmd": "enqueue",
+        "registry_id_hex": reg["registry_id_hex"],
+        "role_name": args.role,
+        "request_id_hex": req_id.hex(),
+        "eph_pk_hex": eph_pk.hex(),
+        "payload_b64": base64.b64encode(payload).decode(),
+    }
+    out = client_ops.relay_line(args.host, args.relay_port, obj)
+    print(json.dumps(out, indent=2))
+    print(f"Saved pending material to {pdir / (req_id.hex() + '.json')}")
+
+
+def cmd_issuance_poll(args: argparse.Namespace) -> None:
+    reg = creds.load_json(paths.user_dir(args.user) / "registries" / args.slug / "registry.json")
+    obj = {
+        "cmd": "poll",
+        "registry_id_hex": reg["registry_id_hex"],
+        "request_id_hex": args.request_id,
+    }
+    out = client_ops.relay_line(args.host, args.relay_port, obj)
+    print(json.dumps(out, indent=2))
+    if out.get("status") == "ready":
+        pend = creds.load_json(paths.user_dir(args.user) / "pending" / f"{args.request_id}.json")
+        pk = zkac.BbsPublicKey.from_bytes(base64.b64decode(pend["issuer_public_key_b64"]))
+        rid = zkac.role_id(pend["role_name"])
+        blind_sig = base64.b64decode(out["blind_sig_b64"])
+        zkac.Credential.finalize(
+            blind_sig,
+            base64.b64decode(pend["member_secret_b64"]),
+            base64.b64decode(pend["prover_blind_b64"]),
+            rid,
+            int(pend["epoch"]),
+            pk,
+        )
+        out_dir = paths.user_dir(args.user) / "registries" / args.slug / "roles"
+        out_dir.mkdir(parents=True, exist_ok=True)
+        member = {
+            "role_id_hex": rid.hex(),
+            "epoch": int(pend["epoch"]),
+            "blind_sig_b64": base64.b64encode(blind_sig).decode(),
+            "member_secret_b64": pend["member_secret_b64"],
+            "prover_blind_b64": pend["prover_blind_b64"],
+            "issuer_public_key_b64": pend["issuer_public_key_b64"],
+        }
+        dest = out_dir / f"{pend['role_name']}.json"
+        creds.save_json(dest, member)
+        print(f"You have role {pend['role_name']!r}; credential saved to {dest}")
+        print("Use: zkac-node connect ... --credential roles/{0}.json".format(pend["role_name"]))
+
+
+def cmd_issuance_grant(args: argparse.Namespace) -> None:
+    server_data = Path(args.server_data).resolve()
+    reg_dir = (paths.user_dir(args.admin_user) / "registries" / args.slug).resolve()
+    admin = creds.load_json(reg_dir / "admin.json")
+    issuer = zkac.BbsIssuer.from_secret_key(base64.b64decode(admin["admin_issuer_secret_b64"]))
+    t = creds.load_json(server_data / "transport.json")
+    mg = creds.load_json(server_data / "mgmt_member.json")
+    server_pk = zkac.PublicKey.from_bytes(base64.b64decode(t["server_public_key_b64"]))
+    mgmt_cred = creds.load_member_credential(mg)
+
+    peek = client_ops.mgmt_call(
+        args.host,
+        args.mgmt_port,
+        server_pk,
+        mgmt_cred,
+        {"cmd": "issuance_peek", "registry_id_hex": args.registry_id},
+    )
+    pending = peek.get("pending") or []
+    target = None
+    for p in pending:
+        if p["request_id_hex"] == args.request_id:
+            target = p
+            break
+    if target is None:
+        print("request_id not in pending queue", file=sys.stderr)
+        raise SystemExit(1)
+    commit = base64.b64decode(target["payload_b64"])
+    role_id = bytes.fromhex(target["role_id_hex"])
+    blind = issuer.issue_blind(commit, role_id, 1)
+    out = client_ops.mgmt_call(
+        args.host,
+        args.mgmt_port,
+        server_pk,
+        mgmt_cred,
+        {
+            "cmd": "issuance_grant",
+            "registry_id_hex": args.registry_id,
+            "request_id_hex": args.request_id,
+            "blind_sig_b64": base64.b64encode(blind).decode(),
+        },
+    )
+    print(json.dumps(out, indent=2))
+
+
+def cmd_issue_member_file(args: argparse.Namespace) -> None:
+    """Admin issues a role credential locally (out-of-band handoff)."""
+    reg_dir = (paths.user_dir(args.admin_user) / "registries" / args.slug).resolve()
+    admin = creds.load_json(reg_dir / "admin.json")
+    issuer = zkac.BbsIssuer.from_secret_key(base64.b64decode(admin["admin_issuer_secret_b64"]))
+    pk = issuer.public_key()
+    rid = zkac.role_id(args.role)
+    req = zkac.prepare_blind_request()
+    sig = issuer.issue_blind(req.commitment_with_proof(), rid, 1)
+    payload = creds.member_payload(sig, req, rid, 1, pk)
+    out_path = reg_dir / "issued" / f"{args.role}_{args.target_user}.json"
+    creds.save_json(out_path, payload)
+    print(f"Wrote {out_path}")
+
+
+def cmd_registry_import_public(args: argparse.Namespace) -> None:
+    """Copy public registry.json (state + cert summary) from another local user."""
+    src = (paths.user_dir(args.from_user) / "registries" / args.slug / "registry.json").resolve()
+    dst_dir = paths.ensure_user(args.to_user) / "registries" / args.slug
+    dst_dir.mkdir(parents=True, exist_ok=True)
+    shutil.copy(src, dst_dir / "registry.json")
+    print(f"Copied {src} -> {dst_dir / 'registry.json'}")
+
+
+def cmd_import_credential(args: argparse.Namespace) -> None:
+    src = Path(args.file).resolve()
+    data = creds.load_json(src)
+    dest = paths.user_dir(args.user) / "registries" / args.slug / "roles" / f"{args.name}.json"
+    creds.save_json(dest, data)
+    print(f"Imported credential to {dest}")
+
+
+def cmd_pin_server(args: argparse.Namespace) -> None:
+    """Save server public key + ports for a user (from server's transport.json)."""
+    tpath = Path(args.transport_file).resolve()
+    t = creds.load_json(tpath)
+    d = paths.ensure_user(args.user) / "servers" / args.name
+    d.mkdir(parents=True, exist_ok=True)
+    creds.save_json(
+        d / "server.json",
+        {
+            "server_public_key_b64": t["server_public_key_b64"],
+            "host": args.host,
+            "mgmt_port": args.mgmt_port,
+            "managed_port": args.managed_port,
+            "relay_port": args.relay_port,
+        },
+    )
+    print(f"Pinned server {args.name!r} for user {args.user!r} -> {d / 'server.json'}")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(
+        prog="zkac-node",
+        description="ZKAC node CLI (server + client). "
+        "Requires the 'zkac' package from this repo (maturin develop). "
+        "Set ZKAC_HOME to override ~/.zkac/. "
+        "Issuance relay queues are in-memory — use the same server process between request and grant.",
+    )
+    sub = p.add_subparsers(dest="command", required=True)
+
+    ps = sub.add_parser("serve", help="Run server (registry-capable node)")
+    ps.add_argument("--data", type=Path, required=True, help="Server data directory")
+    ps.add_argument("--init", action="store_true", help="Initialize data dir (transport + mgmt issuer)")
+    ps.add_argument("--mgmt-port", type=int, default=7400)
+    ps.add_argument("--managed-port", type=int, default=7401)
+    ps.add_argument(
+        "--relay-port",
+        type=int,
+        default=7402,
+        help="Plaintext relay for issuance queue (localhost only). Use 0 to disable.",
+    )
+    ps.add_argument("--relay-bind", default="127.0.0.1")
+    ps.set_defaults(func=cmd_serve)
+
+    su = sub.add_parser("user-create", help="Create a user directory under ~/.zkac/")
+    su.add_argument("userid")
+    su.set_defaults(func=cmd_user_create)
+
+    pi = sub.add_parser("pin-server", help="Save server connection info for a user")
+    pi.add_argument("--user", required=True)
+    pi.add_argument("--name", required=True, help="Label for this server")
+    pi.add_argument("--transport-file", required=True, help="Path to server's transport.json")
+    pi.add_argument("--host", required=True)
+    pi.add_argument("--mgmt-port", type=int, default=7400)
+    pi.add_argument("--managed-port", type=int, default=7401)
+    pi.add_argument("--relay-port", type=int, default=7402)
+    pi.set_defaults(func=cmd_pin_server)
+
+    ri = sub.add_parser("registry-init", help="Create a client-managed registry offline")
+    ri.add_argument("--user", required=True)
+    ri.add_argument("--slug", required=True)
+    ri.add_argument("--roles", nargs="+", required=True)
+    ri.set_defaults(func=cmd_registry_init)
+
+    rp = sub.add_parser("registry-push", help="Push registry snapshot to server (mgmt channel)")
+    rp.add_argument("--user", required=True)
+    rp.add_argument("--slug", required=True)
+    rp.add_argument("--server-data", type=Path, required=True, help="Server data dir (transport + mgmt_member)")
+    rp.add_argument("--host", default="127.0.0.1")
+    rp.add_argument("--mgmt-port", type=int, default=7400)
+    rp.add_argument("--update", action="store_true", help="Send update_registry instead of create")
+    rp.set_defaults(func=cmd_registry_push)
+
+    cc = sub.add_parser("connect", help="Managed ZKAC session test (whoami/get_registry)")
+    cc.add_argument("--user", required=True)
+    cc.add_argument("--slug", required=True)
+    cc.add_argument("--credential", default="roles/member.json", help="Relative path under registries/<slug>/")
+    cc.add_argument(
+        "--registry-id",
+        default="",
+        help="If registry.json is absent, hex registry id (pin public registry first)",
+    )
+    cc.add_argument("--server-file", required=True, help="Pinned server.json from pin-server")
+    cc.add_argument("--host", required=True)
+    cc.add_argument("--managed-port", type=int, default=7401)
+    cc.add_argument("--command", default="whoami")
+    cc.set_defaults(func=cmd_connect)
+
+    ir = sub.add_parser(
+        "issuance-request",
+        help="Enqueue blind commitment on relay (localhost; see help text)",
+    )
+    ir.add_argument("--user", required=True)
+    ir.add_argument("--slug", required=True)
+    ir.add_argument("--role", required=True)
+    ir.add_argument("--host", default="127.0.0.1")
+    ir.add_argument("--relay-port", type=int, default=7402)
+    ir.set_defaults(func=cmd_issuance_request)
+
+    ip = sub.add_parser("issuance-poll", help="Poll relay for blind signature")
+    ip.add_argument("--user", required=True)
+    ip.add_argument("--slug", required=True)
+    ip.add_argument("--request-id", required=True, dest="request_id")
+    ip.add_argument("--host", default="127.0.0.1")
+    ip.add_argument("--relay-port", type=int, default=7402)
+    ip.set_defaults(func=cmd_issuance_poll)
+
+    ig = sub.add_parser("issuance-grant", help="Admin: issue blind sig and push grant to server")
+    ig.add_argument("--admin-user", required=True)
+    ig.add_argument("--slug", required=True)
+    ig.add_argument("--server-data", type=Path, required=True)
+    ig.add_argument("--registry-id", required=True, dest="registry_id")
+    ig.add_argument("--request-id", required=True, dest="request_id")
+    ig.add_argument("--host", default="127.0.0.1")
+    ig.add_argument("--mgmt-port", type=int, default=7400)
+    ig.set_defaults(func=cmd_issuance_grant)
+
+    im = sub.add_parser("issue-member", help="Admin: issue credential to file for handoff")
+    im.add_argument("--admin-user", required=True)
+    im.add_argument("--slug", required=True)
+    im.add_argument("--role", required=True)
+    im.add_argument("--target-user", required=True)
+    im.set_defaults(func=cmd_issue_member_file)
+
+    irp = sub.add_parser(
+        "registry-import-public",
+        help="Copy registry.json from another user (public state + cert metadata)",
+    )
+    irp.add_argument("--from-user", required=True, dest="from_user")
+    irp.add_argument("--to-user", required=True, dest="to_user")
+    irp.add_argument("--slug", required=True)
+    irp.set_defaults(func=cmd_registry_import_public)
+
+    ii = sub.add_parser("import-credential", help="Copy a member json into a user's registry folder")
+    ii.add_argument("--user", required=True)
+    ii.add_argument("--slug", required=True)
+    ii.add_argument("--name", required=True, help="Filename base (e.g. analyst)")
+    ii.add_argument("file", type=Path)
+    ii.set_defaults(func=cmd_import_credential)
+
+    return p
+
+
+def main() -> None:
+    args = build_parser().parse_args()
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/cli/zkac_cli/paths.py b/cli/zkac_cli/paths.py
new file mode 100644
index 0000000..4748614
--- /dev/null
+++ b/cli/zkac_cli/paths.py
@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+
+def zkac_home() -> Path:
+    return Path(os.environ.get("ZKAC_HOME", Path.home() / ".zkac"))
+
+
+def user_dir(userid: str) -> Path:
+    return zkac_home() / userid
+
+
+def ensure_user(userid: str) -> Path:
+    zkac_home().mkdir(parents=True, exist_ok=True)
+    d = user_dir(userid)
+    d.mkdir(parents=True, exist_ok=True)
+    return d
diff --git a/cli/zkac_cli/registry_local.py b/cli/zkac_cli/registry_local.py
new file mode 100644
index 0000000..5bfde57
--- /dev/null
+++ b/cli/zkac_cli/registry_local.py
@@ -0,0 +1,68 @@
+"""Offline client-managed registry creation (same structure as demo/setup_managed_demo)."""
+
+from __future__ import annotations
+
+import base64
+import json
+from pathlib import Path
+
+import zkac
+
+from zkac_cli import creds
+
+
+def create_registry_bundle(
+    slug: str,
+    role_names: list[str],
+    out_dir: Path,
+) -> dict[str, str]:
+    """
+    Build a new registry with admin issuer = role issuer for all listed roles.
+    Writes admin.json, registry.json under out_dir.
+    """
+    out_dir.mkdir(parents=True, exist_ok=True)
+    admin_issuer = zkac.BbsIssuer()
+    admin_pk = admin_issuer.public_key()
+    admin_rid = zkac.admin_role_id()
+
+    req = zkac.prepare_blind_request()
+    sig = admin_issuer.issue_blind(req.commitment_with_proof(), admin_rid, 0)
+    admin_cred = zkac.Credential.finalize(
+        sig, req.member_secret(), req.prover_blind(), admin_rid, 0, admin_pk
+    )
+
+    issuance_kp = zkac.IssuanceKeypair()
+    role_entries = [(zkac.role_id(name), admin_pk, 1) for name in role_names]
+
+    state = zkac.RegistryState.build(
+        admin_pk, issuance_kp.public_key_bytes(), 1, b"\x00" * 32, role_entries
+    )
+    state_bytes = state.serialize()
+    state_cert = state.certify(admin_cred)
+    registry_id = state.registry_id()
+
+    admin_payload = {
+        "slug": slug,
+        "admin_issuer_secret_b64": base64.b64encode(admin_issuer.secret_key_bytes()).decode(),
+        "admin_issuer_public_key_b64": base64.b64encode(admin_pk.to_bytes()).decode(),
+        "admin_member_secret_b64": base64.b64encode(req.member_secret()).decode(),
+        "admin_prover_blind_b64": base64.b64encode(req.prover_blind()).decode(),
+        "admin_blind_sig_b64": base64.b64encode(sig).decode(),
+        "issuance_secret_b64": base64.b64encode(issuance_kp.secret_bytes()).decode(),
+        "issuance_public_key_b64": base64.b64encode(issuance_kp.public_key_bytes()).decode(),
+        "registry_id_hex": registry_id.hex(),
+    }
+    creds.save_json(out_dir / "admin.json", admin_payload)
+
+    reg_payload = {
+        "slug": slug,
+        "registry_id_hex": registry_id.hex(),
+        "state_bytes_b64": base64.b64encode(state_bytes).decode(),
+        "state_cert_b64": base64.b64encode(bytes(state_cert)).decode(),
+        "admin_issuer_public_key_b64": base64.b64encode(admin_pk.to_bytes()).decode(),
+        "issuance_public_key_b64": base64.b64encode(issuance_kp.public_key_bytes()).decode(),
+        "roles": role_names,
+    }
+    creds.save_json(out_dir / "registry.json", reg_payload)
+
+    return {"registry_id_hex": registry_id.hex(), "path": str(out_dir)}
diff --git a/cli/zkac_cli/registry_log.py b/cli/zkac_cli/registry_log.py
new file mode 100644
index 0000000..ac67709
--- /dev/null
+++ b/cli/zkac_cli/registry_log.py
@@ -0,0 +1,59 @@
+"""Persist RegistryManager by replaying create/update events."""
+
+from __future__ import annotations
+
+import base64
+import json
+from pathlib import Path
+from typing import Any
+
+import zkac
+
+
+def save_events(path: Path, events: list[dict[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({"events": events}, indent=2), encoding="utf-8")
+
+
+def load_events(path: Path) -> list[dict[str, Any]]:
+    if not path.is_file():
+        return []
+    data = json.loads(path.read_text(encoding="utf-8"))
+    return list(data.get("events", []))
+
+
+def replay_manager(events: list[dict[str, Any]]) -> zkac.RegistryManager:
+    """Rebuild manager from persisted events (multiple registries supported)."""
+    mgr = zkac.RegistryManager()
+    by_rid: dict[bytes, list[dict[str, Any]]] = {}
+    for e in events:
+        st = base64.b64decode(e["state_b64"])
+        st_obj = zkac.RegistryState.deserialize(st)
+        rid = st_obj.registry_id()
+        by_rid.setdefault(rid, []).append(e)
+    for rid, evs in by_rid.items():
+        evs.sort(
+            key=lambda e: zkac.RegistryState.deserialize(
+                base64.b64decode(e["state_b64"])
+            ).version()
+        )
+        for i, e in enumerate(evs):
+            st = base64.b64decode(e["state_b64"])
+            cert = base64.b64decode(e["state_cert_b64"])
+            if i == 0:
+                mgr.create(st, cert)
+            else:
+                mgr.update(rid, st, cert)
+    return mgr
+
+
+def append_event(path: Path, op: str, state_bytes: bytes, state_cert: bytes) -> None:
+    events = load_events(path)
+    events.append(
+        {
+            "op": op,
+            "state_b64": base64.b64encode(state_bytes).decode(),
+            "state_cert_b64": base64.b64encode(state_cert).decode(),
+        }
+    )
+    save_events(path, events)
diff --git a/cli/zkac_cli/server_app.py b/cli/zkac_cli/server_app.py
new file mode 100644
index 0000000..d43346c
--- /dev/null
+++ b/cli/zkac_cli/server_app.py
@@ -0,0 +1,272 @@
+"""TCP servers: ZKAC management, ZKAC managed app, localhost relay (credential queue)."""
+
+from __future__ import annotations
+
+import base64
+import json
+import socket
+import threading
+import traceback
+from pathlib import Path
+from typing import Callable
+
+import zkac
+from zkac.tcp import FramedSession, server_handshake, server_handshake_managed
+
+from zkac_cli import issuance_util
+from zkac_cli import registry_log
+
+
+MGMT_ROLE = "zkac.mgmt"
+
+
+def build_static_registry(issuer_pk: zkac.BbsPublicKey) -> zkac.RoleRegistry:
+    reg = zkac.RoleRegistry()
+    reg.register_role(zkac.role_id(MGMT_ROLE), issuer_pk, 1)
+    return reg
+
+
+def handle_mgmt_session(
+    conn: socket.socket,
+    data_dir: Path,
+    mgr: zkac.RegistryManager,
+    registry_ids: set[bytes],
+    events_path: Path,
+    save_registry: Callable[[], None],
+) -> None:
+    from zkac_cli import creds
+
+    framed = None
+    try:
+        sk = creds.load_server_keypair(data_dir / "transport.json")
+        node = zkac.Node(sk)
+        issuer = zkac.BbsIssuer.from_secret_key(
+            base64.b64decode(creds.load_json(data_dir / "mgmt_issuer.json")["issuer_secret_b64"])
+        )
+        static_reg = build_static_registry(issuer.public_key())
+        session, role_id = server_handshake(conn, node, static_reg)
+        if role_id != zkac.role_id(MGMT_ROLE):
+            return
+        framed = FramedSession(conn, session)
+        raw = framed.recv()
+        msg = json.loads(raw.decode("utf-8"))
+        cmd = msg["cmd"]
+        if cmd == "create_registry":
+            st = base64.b64decode(msg["state_b64"])
+            cert = base64.b64decode(msg["state_cert_b64"])
+            rid = mgr.create(st, cert)
+            registry_ids.add(rid)
+            registry_log.append_event(events_path, "create", st, cert)
+            save_registry()
+            out = {"ok": True, "registry_id_hex": rid.hex()}
+        elif cmd == "update_registry":
+            st = base64.b64decode(msg["state_b64"])
+            cert = base64.b64decode(msg["state_cert_b64"])
+            st_o = zkac.RegistryState.deserialize(st)
+            rid = st_o.registry_id()
+            mgr.update(rid, st, cert)
+            registry_log.append_event(events_path, "update", st, cert)
+            save_registry()
+            out = {"ok": True, "registry_id_hex": rid.hex()}
+        elif cmd == "get_registry":
+            rid = bytes.fromhex(msg["registry_id_hex"])
+            st, c = mgr.get(rid)
+            out = {
+                "ok": True,
+                "state_b64": base64.b64encode(st).decode(),
+                "state_cert_b64": base64.b64encode(c).decode(),
+            }
+        elif cmd == "list_registries":
+            out = {"ok": True, "registry_ids_hex": [h.hex() for h in sorted(registry_ids)]}
+        elif cmd == "issuance_peek":
+            rid = bytes.fromhex(msg["registry_id_hex"])
+            pending = issuance_util.peek_pending_requests(mgr, rid)
+            out = {
+                "ok": True,
+                "pending": [
+                    {
+                        "request_id_hex": a.hex(),
+                        "role_id_hex": b.hex(),
+                        "eph_pk_hex": c.hex(),
+                        "payload_b64": base64.b64encode(d).decode(),
+                    }
+                    for a, b, c, d in pending
+                ],
+            }
+        elif cmd == "issuance_grant":
+            rid = bytes.fromhex(msg["registry_id_hex"])
+            req_id = bytes.fromhex(msg["request_id_hex"])
+            blind = base64.b64decode(msg["blind_sig_b64"])
+            mgr.grant_credential(rid, req_id, blind)
+            save_registry()
+            out = {"ok": True}
+        else:
+            out = {"ok": False, "error": f"unknown cmd {cmd}"}
+        framed.send(json.dumps(out).encode("utf-8"))
+    except Exception:
+        traceback.print_exc()
+    finally:
+        conn.close()
+
+
+def run_managed_handler(
+    conn: socket.socket,
+    data_dir: Path,
+    mgr: zkac.RegistryManager,
+) -> None:
+    from zkac_cli import creds
+
+    try:
+        sk = creds.load_server_keypair(data_dir / "transport.json")
+        node = zkac.Node(sk)
+        session, registry_id, role_id = server_handshake_managed(conn, node, mgr)
+        framed = FramedSession(conn, session)
+        raw = framed.recv()
+        msg = json.loads(raw.decode("utf-8"))
+        cmd = msg.get("cmd", "ping")
+        if cmd == "get_registry":
+            rid = bytes.fromhex(msg["registry_id_hex"])
+            st, c = mgr.get(rid)
+            body = {
+                "ok": True,
+                "state_b64": base64.b64encode(st).decode(),
+                "state_cert_b64": base64.b64encode(c).decode(),
+            }
+        elif cmd == "whoami":
+            body = {
+                "ok": True,
+                "registry_id_hex": registry_id.hex(),
+                "role_id_hex": role_id.hex(),
+            }
+        else:
+            body = {
+                "ok": True,
+                "registry_id_hex": registry_id.hex(),
+                "role_id_hex": role_id.hex(),
+                "note": "authenticated managed session",
+            }
+        framed.send(json.dumps(body).encode("utf-8"))
+    except Exception:
+        traceback.print_exc()
+    finally:
+        conn.close()
+
+
+def handle_relay_session(
+    conn: socket.socket,
+    mgr: zkac.RegistryManager,
+    save_registry: Callable[[], None],
+) -> None:
+    try:
+        buf = b""
+        while b"\n" not in buf:
+            chunk = conn.recv(4096)
+            if not chunk:
+                return
+            buf += chunk
+        line, _, _ = buf.partition(b"\n")
+        msg = json.loads(line.decode("utf-8"))
+        cmd = msg["cmd"]
+        if cmd == "enqueue":
+            rid = bytes.fromhex(msg["registry_id_hex"])
+            role = zkac.role_id(msg["role_name"])
+            req_id = bytes.fromhex(msg["request_id_hex"])
+            eph = bytes.fromhex(msg["eph_pk_hex"])
+            blob = base64.b64decode(msg["payload_b64"])
+            mgr.queue_issuance_request(rid, req_id, role, eph, blob)
+            save_registry()
+            out = {"ok": True, "status": "queued"}
+        elif cmd == "poll":
+            rid = bytes.fromhex(msg["registry_id_hex"])
+            req_id = bytes.fromhex(msg["request_id_hex"])
+            g = mgr.take_granted_credential(rid, req_id)
+            save_registry()
+            if g is None:
+                out = {"ok": True, "status": "pending"}
+            else:
+                out = {
+                    "ok": True,
+                    "status": "ready",
+                    "blind_sig_b64": base64.b64encode(g).decode(),
+                }
+        else:
+            out = {"ok": False, "error": "unknown relay cmd"}
+        conn.sendall((json.dumps(out) + "\n").encode("utf-8"))
+    except Exception:
+        traceback.print_exc()
+    finally:
+        conn.close()
+
+
+def serve(
+    data_dir: Path,
+    mgmt_port: int,
+    managed_port: int,
+    relay_port: int | None,
+    relay_bind: str,
+) -> None:
+    events_path = data_dir / "registry_events.json"
+    registry_ids: set[bytes] = set()
+    if events_path.is_file():
+        mgr = registry_log.replay_manager(registry_log.load_events(events_path))
+        for e in registry_log.load_events(events_path):
+            st = base64.b64decode(e["state_b64"])
+            registry_ids.add(zkac.RegistryState.deserialize(st).registry_id())
+    else:
+        mgr = zkac.RegistryManager()
+
+    def save_registry() -> None:
+        return
+
+    def mgmt_loop() -> None:
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        s.bind(("0.0.0.0", mgmt_port))
+        s.listen(8)
+        print(f"[mgmt] ZKAC listening on 0.0.0.0:{mgmt_port}")
+        while True:
+            c, a = s.accept()
+            print(f"[mgmt] connect {a}")
+            threading.Thread(
+                target=handle_mgmt_session,
+                args=(c, data_dir, mgr, registry_ids, events_path, save_registry),
+                daemon=True,
+            ).start()
+
+    def managed_loop() -> None:
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        s.bind(("0.0.0.0", managed_port))
+        s.listen(8)
+        print(f"[managed] ZKAC listening on 0.0.0.0:{managed_port}")
+        while True:
+            c, a = s.accept()
+            print(f"[managed] connect {a}")
+            threading.Thread(
+                target=run_managed_handler,
+                args=(c, data_dir, mgr),
+                daemon=True,
+            ).start()
+
+    def relay_loop() -> None:
+        if relay_port is None:
+            return
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        s.bind((relay_bind, relay_port))
+        s.listen(8)
+        print(f"[relay] plaintext JSON lines on {relay_bind}:{relay_port}")
+        while True:
+            c, a = s.accept()
+            print(f"[relay] connect {a}")
+            threading.Thread(
+                target=handle_relay_session,
+                args=(c, mgr, save_registry),
+                daemon=True,
+            ).start()
+
+    threading.Thread(target=mgmt_loop, daemon=True).start()
+    threading.Thread(target=managed_loop, daemon=True).start()
+    if relay_port is not None:
+        threading.Thread(target=relay_loop, daemon=True).start()
+    threading.Event().wait()