#!/usr/bin/env python3
"""
MODULE: havre_remote.py
ROLE: thin client CLI that drives the live control-plane API over HTTPS
LAYER: Adapters
WHEN: run by a Havre client on their own laptop; not the operator tool

Receives shell commands (signup/login/whoami/put/get/list/delete) plus Silo device and sync
commands, seals and unseals files locally with the user's own X25519 key (which never leaves
the machine), and exchanges only sealed objects and ciphertext with https://havre.cloud over
HTTPS. The server stays blind. Silo private sync uses durable resumable uploads and an ordered
change cursor; public library mode is an explicit P0 choice for rights-cleared books or music.
Sibling of havre_cli.py (the operator's direct-to-OVH tool); both reuse envelope.py.
"""

from __future__ import annotations

import argparse
import base64
import datetime
import getpass
import hashlib
import hmac
import json
import math
import os
import pathlib
import sys
import tempfile
import time
import uuid
import urllib.error
import urllib.request
from urllib.parse import urlencode, urlparse

from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import envelope  # noqa: E402
import silo_service  # noqa: E402

DEFAULT_SERVER = "https://havre.cloud"
RECOVERY_BUNDLE_VERSION = 1
RECOVERY_AAD = b"havre-silo-recovery-v1"
# 16 MiB working set: compatible with the production-like OpenSSL memory cap
# while remaining substantially more expensive than a fast password hash.
RECOVERY_SCRYPT = {"n": 16384, "r": 8, "p": 1, "dklen": 32}
DEFAULT_WATCH_RETRY_SECONDS = 15.0
DEFAULT_WATCH_MAX_FAILURES = 5
MAX_WATCH_RETRY_SECONDS = 300.0


# ── LOCAL STATE: identity key + session ───────────────────────────────────────
# The X25519 private key and the bearer token live under HAVRE_HOME (default
# ~/.havre), both mode 600. The key is the user's identity and never leaves the
# machine; the session holds only the server URL, email, and bearer token.

def _home() -> pathlib.Path:
    return pathlib.Path(os.environ.get("HAVRE_HOME") or (pathlib.Path.home() / ".havre"))


def _key_path() -> pathlib.Path:
    return _home() / "key.json"


def _session_path() -> pathlib.Path:
    return _home() / "session.json"


def ensure_key() -> dict:
    path = _key_path()
    if path.exists():
        return json.loads(path.read_text())
    sk, pk = envelope.gen_recipient_keypair()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps({"private": sk, "public": pk}))
    os.chmod(path, 0o600)
    return {"private": sk, "public": pk}


def load_session() -> dict:
    p = _session_path()
    return json.loads(p.read_text()) if p.exists() else {}


def save_session(session: dict) -> None:
    p = _session_path()
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps(session))
    os.chmod(p, 0o600)


def clear_session() -> None:
    p = _session_path()
    if p.exists():
        p.unlink()


def _silo_state_path() -> pathlib.Path:
    return _home() / "silo.json"


def _silo_upload_dir() -> pathlib.Path:
    return _home() / "silo_uploads"


def load_silo_state() -> dict:
    path = _silo_state_path()
    if not path.exists():
        return {"device_id": "", "device_name": "", "device_token": "", "roots": {}}
    try:
        state = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise RuntimeError(f"cannot read {path}: {exc}") from exc
    if not isinstance(state, dict):
        raise RuntimeError(f"{path} must contain a JSON object")
    state.setdefault("device_id", "")
    state.setdefault("device_name", "")
    state.setdefault("device_token", "")
    state.setdefault("roots", {})
    return state


def _silo_active_key(fallback: dict) -> dict:
    """Return the active Silo key, keeping it separate from the account key."""
    state = load_silo_state()
    private = state.get("silo_private")
    public = state.get("silo_public")
    if not private and not public:
        return fallback
    if not isinstance(private, str) or not isinstance(public, str):
        raise RuntimeError("local Silo key state is incomplete; re-import the recovery bundle")
    try:
        derived = envelope.public_key_from_private(private)
    except (TypeError, ValueError) as exc:
        raise RuntimeError("local Silo private key is invalid; re-import the recovery bundle") from exc
    if not hmac.compare_digest(derived, public):
        raise RuntimeError("local Silo public/private keys do not match; re-import the recovery bundle")
    return {"private": private, "public": public}


def _recovery_password(confirm: bool = False) -> str:
    password = getpass.getpass("Silo recovery password: ")
    if len(password) < 12:
        raise ValueError("Silo recovery passwords must be at least 12 characters")
    if confirm:
        repeated = getpass.getpass("Repeat Silo recovery password: ")
        if not hmac.compare_digest(password, repeated):
            raise ValueError("Silo recovery passwords do not match")
    return password


def _derive_recovery_key(password: str, salt: bytes) -> bytes:
    if not isinstance(password, str) or len(password) < 12:
        raise ValueError("Silo recovery passwords must be at least 12 characters")
    return hashlib.scrypt(password.encode("utf-8"), salt=salt, **RECOVERY_SCRYPT)


def _seal_recovery_bundle(payload: dict, password: str) -> dict:
    """Encrypt recovery device material; only KDF metadata and ciphertext leave this function."""
    if not isinstance(payload, dict):
        raise ValueError("recovery payload must be a JSON object")
    salt = os.urandom(16)
    nonce = os.urandom(12)
    plaintext = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ciphertext = AESGCM(_derive_recovery_key(password, salt)).encrypt(
        nonce, plaintext, RECOVERY_AAD)
    return {
        "version": RECOVERY_BUNDLE_VERSION,
        "kdf": {"name": "scrypt", **RECOVERY_SCRYPT},
        "salt_b64": base64.b64encode(salt).decode("ascii"),
        "nonce_b64": base64.b64encode(nonce).decode("ascii"),
        "ciphertext_b64": base64.b64encode(ciphertext).decode("ascii"),
    }


def _recovery_server_is_valid(server: str) -> bool:
    parsed = urlparse(server)
    return parsed.scheme in {"http", "https"} and bool(parsed.netloc)


def _open_recovery_bundle(bundle: dict, password: str) -> dict:
    """Decrypt and validate a recovery bundle without contacting Havre."""
    if not isinstance(bundle, dict) or bundle.get("version") != RECOVERY_BUNDLE_VERSION:
        raise ValueError("unsupported Silo recovery bundle version")
    kdf = bundle.get("kdf")
    expected_kdf = {"name": "scrypt", **RECOVERY_SCRYPT}
    if kdf != expected_kdf:
        raise ValueError("unsupported Silo recovery KDF parameters")
    try:
        salt = base64.b64decode(bundle["salt_b64"], validate=True)
        nonce = base64.b64decode(bundle["nonce_b64"], validate=True)
        ciphertext = base64.b64decode(bundle["ciphertext_b64"], validate=True)
    except (KeyError, TypeError, ValueError) as exc:
        raise ValueError("invalid Silo recovery bundle encoding") from exc
    if len(salt) != 16 or len(nonce) != 12 or not ciphertext:
        raise ValueError("invalid Silo recovery bundle lengths")
    try:
        raw = AESGCM(_derive_recovery_key(password, salt)).decrypt(
            nonce, ciphertext, RECOVERY_AAD)
    except (InvalidTag, ValueError) as exc:
        raise ValueError("wrong Silo recovery password or tampered bundle") from exc
    try:
        payload = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, ValueError) as exc:
        raise ValueError("recovery bundle payload is not valid JSON") from exc
    if not isinstance(payload, dict):
        raise ValueError("recovery bundle payload must be a JSON object")
    required = ("device_id", "device_name", "device_token", "server",
                "private_key", "public_key", "purpose")
    if any(not isinstance(payload.get(name), str) or not payload.get(name)
           for name in required):
        raise ValueError("recovery bundle is missing device material")
    if payload["purpose"] != "silo-device-recovery":
        raise ValueError("recovery bundle purpose is invalid")
    if not _recovery_server_is_valid(payload["server"]):
        raise ValueError("recovery bundle server URL is invalid")
    try:
        derived = envelope.public_key_from_private(payload["private_key"])
    except (TypeError, ValueError) as exc:
        raise ValueError("recovery bundle private key is invalid") from exc
    if not hmac.compare_digest(derived, payload["public_key"]):
        raise ValueError("recovery bundle public/private keys do not match")
    return payload


def _write_private_json(path: pathlib.Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
    os.close(fd)
    temporary_path = pathlib.Path(temporary)
    try:
        temporary_path.write_text(
            json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        os.chmod(temporary_path, 0o600)
        os.replace(temporary_path, path)
    finally:
        if temporary_path.exists():
            temporary_path.unlink()


def save_silo_state(state: dict) -> None:
    _write_private_json(_silo_state_path(), state)


def _private_file_ok(path: pathlib.Path) -> bool:
    """Windows does not expose POSIX chmod bits; the file is still user-local there."""
    return path.exists() and (os.name == "nt" or (os.stat(path).st_mode & 0o777) == 0o600)


def _root_state(state: dict, root: pathlib.Path) -> dict:
    key = str(root.resolve())
    roots = state.setdefault("roots", {})
    item = roots.setdefault(key, {"cursor": 0, "files": {}})
    item.setdefault("cursor", 0)
    item.setdefault("files", {})
    return item


def _silo_normalize_prefix(value: str) -> str:
    """Normalize one device-retention prefix without allowing path traversal."""
    prefix = str(value or "").strip().replace("\\", "/").strip("/")
    parts = prefix.split("/")
    if (not prefix or any(part in ("", ".", "..") for part in parts)):
        raise ValueError("Silo include prefixes must be non-empty relative folders")
    return prefix


def _silo_normalize_prefixes(values) -> tuple[str, ...]:
    """Return stable, non-overlapping device-retention prefixes."""
    if values is None:
        return ()
    if isinstance(values, str):
        values = [values]
    normalized = sorted({
        _silo_normalize_prefix(value)
        for value in values
    }, key=lambda value: (len(value), value))
    selected = []
    for prefix in normalized:
        if any(prefix == parent or prefix.startswith(parent + "/")
               for parent in selected):
            continue
        selected.append(prefix)
    return tuple(sorted(selected))


def _silo_effective_prefixes(args, root_state: dict) -> tuple[str, ...]:
    """Resolve explicit CLI selection, saved selection, or the full library default."""
    if getattr(args, "include_all", False):
        return ()
    explicit = getattr(args, "include_prefix", None)
    if explicit is None:
        explicit = root_state.get("include_prefixes", [])
    return _silo_normalize_prefixes(explicit)


def _silo_path_selected(remote_path: str, prefixes: tuple[str, ...]) -> bool:
    if not prefixes:
        return True
    normalized = str(remote_path or "").replace("\\", "/").strip("/")
    return any(normalized == prefix or normalized.startswith(prefix + "/")
               for prefix in prefixes)


def _silo_device_required() -> tuple[str, dict]:
    state = load_silo_state()
    if not state.get("device_id") or not state.get("device_token"):
        print("no Silo device is registered; run 'silo-device register' first",
              file=sys.stderr)
        raise SystemExit(2)
    return state["device_token"], state


def server_url(args) -> str:
    return (getattr(args, "server", None) or os.environ.get("HAVRE_SERVER")
            or load_session().get("server") or load_silo_state().get("server")
            or DEFAULT_SERVER)


# ── HTTP CORE ─────────────────────────────────────────────────────────────────
# One stdlib urllib helper. Returns (status, dict) and never raises on an HTTP
# error code, so each command maps status to a clean human message itself.

def _api(base: str, method: str, path: str, body=None, token=None):
    data = json.dumps(body).encode("utf-8") if body is not None else None
    req = urllib.request.Request(base + path, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    if token:
        req.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(req) as resp:
            return resp.status, json.loads(resp.read() or b"{}")
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read() or b"{}")
        except Exception:  # noqa: BLE001
            return e.code, {"error": e.reason}
    except urllib.error.URLError as e:
        return 0, {"error": f"cannot reach {base}: {e.reason}"}


def _silo_api(base: str, method: str, path: str, body=None, *, token=None,
              device_token=None, headers=None):
    """Call a Silo JSON or byte endpoint and return (status, body, headers)."""
    if isinstance(body, (bytes, bytearray)):
        data = bytes(body)
        content_type = "application/octet-stream"
    elif body is None:
        data = None
        content_type = "application/json"
    else:
        data = json.dumps(body).encode("utf-8")
        content_type = "application/json"
    req = urllib.request.Request(base + path, data=data, method=method)
    req.add_header("Content-Type", content_type)
    if token:
        req.add_header("Authorization", "Bearer " + token)
    if device_token:
        req.add_header("X-Silo-Device-Token", device_token)
    for key, value in (headers or {}).items():
        req.add_header(key, str(value))
    try:
        with urllib.request.urlopen(req) as resp:
            raw = resp.read()
            try:
                output = json.loads(raw or b"{}")
            except (TypeError, ValueError):
                output = {"raw": raw}
            return resp.status, output, dict(resp.headers)
    except urllib.error.HTTPError as exc:
        raw = exc.read()
        try:
            output = json.loads(raw or b"{}")
        except (TypeError, ValueError):
            output = {"raw": raw}
        return exc.code, output, dict(exc.headers)
    except urllib.error.URLError as exc:
        return 0, {"error": f"cannot reach {base}: {exc.reason}"}, {}


def _require_token(args) -> tuple[str, str]:
    s = load_session()
    if not s.get("token"):
        print("not logged in; run 'signup' or 'login' first", file=sys.stderr)
        raise SystemExit(2)
    return server_url(args), s["token"]


# ── AUTH COMMANDS ─────────────────────────────────────────────────────────────

def _password(args) -> str:
    return args.password or getpass.getpass("password: ")


def cmd_signup(args, key) -> int:
    base = server_url(args)
    body = {
        "email": args.email, "password": _password(args), "public_key": key["public"],
        "residency": args.residency, "use": args.use,
        "sector": args.sector, "sensitivity": args.sensitivity, "invite": args.invite,
    }
    status, out = _api(base, "POST", "/api/signup", body)
    if status == 201:
        save_session({"server": base, "email": args.email, "token": out["token"]})
        print(f"account created for {args.email} (posture {out['account'].get('posture_default')})")
        print(f"public key fingerprint: {envelope.fingerprint(key['public'])}")
        return 0
    print(f"signup failed ({status}): {out.get('error', out)}", file=sys.stderr)
    return 1


def cmd_login(args, key) -> int:
    base = server_url(args)
    status, out = _api(base, "POST", "/api/login", {"email": args.email, "password": _password(args)})
    if status == 200:
        save_session({"server": base, "email": args.email, "token": out["token"]})
        print(f"logged in as {args.email}")
        return 0
    print(f"login failed ({status}): {out.get('error', out)}", file=sys.stderr)
    return 1


def cmd_whoami(args, key) -> int:
    base, token = _require_token(args)
    status, out = _api(base, "GET", "/api/me", token=token)
    if status == 200:
        print(json.dumps(out, indent=2))
        return 0
    print(f"whoami failed ({status}): {out.get('error', out)}", file=sys.stderr)
    return 1


def cmd_logout(args, key) -> int:
    clear_session()
    print("session cleared (your key.json is untouched)")
    return 0


# ── OBJECT COMMANDS: seal locally, ship ciphertext ────────────────────────────

def _map_object_error(status, out) -> str:
    if status == 402:
        return "payment required (billing not active for this account)"
    if status == 401:
        return "unauthorized; your session may have expired, run 'login'"
    return f"({status}) {out.get('error', out)}"


def cmd_put(args, key) -> int:
    base, token = _require_token(args)
    data = pathlib.Path(args.file).read_bytes()
    oid = args.id or pathlib.Path(args.file).name
    meta = {
        "filename": pathlib.Path(args.file).name,
        "size": len(data),
        "created": datetime.datetime.now().isoformat(timespec="seconds"),
    }
    if args.plain:
        obj, blob = envelope.seal(data, posture="P0", storage_class="hot",
                                  residency_scope=args.scope, obj_id=oid, metadata=meta)
    else:
        obj, blob = envelope.seal(data, posture="P2", storage_class="hot",
                                  residency_scope=args.scope, obj_id=oid,
                                  recipients=[key["public"]], metadata=meta)
    body = {"object": obj, "blob_b64": base64.b64encode(blob).decode("ascii")}
    status, out = _api(base, "POST", "/api/objects", body, token=token)
    if status == 201:
        print(f"stored '{out['id']}' ({len(data)} bytes, {'plain' if args.plain else 'sealed'}, {args.scope})")
        return 0
    print(f"put failed: {_map_object_error(status, out)}", file=sys.stderr)
    return 1


def cmd_get(args, key) -> int:
    base, token = _require_token(args)
    status, out = _api(base, "GET", "/api/objects/" + args.id, token=token)
    if status != 200:
        print(f"get failed: {_map_object_error(status, out)}", file=sys.stderr)
        return 1
    blob = base64.b64decode(out["blob_b64"])
    sk = None if out["object"]["posture"] == "P0" else key["private"]
    data = envelope.unseal(out["object"], blob, sk)
    pathlib.Path(args.out).write_bytes(data)
    print(f"retrieved '{args.id}' -> {args.out} ({len(data)} bytes), integrity verified")
    return 0


def cmd_list(args, key) -> int:
    base, token = _require_token(args)
    status, out = _api(base, "GET", "/api/objects", token=token)
    if status != 200:
        print(f"list failed: {_map_object_error(status, out)}", file=sys.stderr)
        return 1
    ids = out.get("ids", [])
    if not ids:
        print("(empty)")
    for i in ids:
        print(" ", i)
    return 0


def cmd_delete(args, key) -> int:
    base, token = _require_token(args)
    status, out = _api(base, "DELETE", "/api/objects/" + args.id, token=token)
    if status == 200:
        print(f"deleted '{args.id}'")
        return 0
    print(f"delete failed: {_map_object_error(status, out)}", file=sys.stderr)
    return 1


def _silo_map_error(status, out) -> str:
    if status == 0:
        return out.get("error", "network error")
    if status == 401:
        return "unauthorized; run 'login' or register a new Silo device"
    if status == 402:
        return "payment required; an active Silo package is required"
    if status == 503:
        return out.get("error", "Silo is not configured for checkout or sync")
    return f"({status}) {out.get('error', out)}"


def _silo_local_hash(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _silo_file_hash(path: pathlib.Path, chunk_size: int = 1024 * 1024) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        while True:
            chunk = handle.read(max(1, int(chunk_size)))
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest()


def _silo_download_part(root: pathlib.Path, object_id: str, version_id: str) -> pathlib.Path:
    """Return a private, content-addressed partial ciphertext path for one change."""
    identity = hashlib.sha256(
        f"{object_id}:{version_id}".encode("utf-8")).hexdigest()
    path = root / ".silo" / "downloads" / f"{identity}.ciphertext.part"
    path.parent.mkdir(parents=True, exist_ok=True)
    return path


def _silo_header(headers, name: str, default=""):
    wanted = name.lower()
    for key, value in (headers or {}).items():
        if str(key).lower() == wanted:
            return value
    return default


def _silo_download_to_file(base: str, path: str, device_token: str,
                           destination: pathlib.Path, expected_size: int,
                           expected_sha256: str, *, chunk_size: int = 1024 * 1024):
    """Resume one ciphertext download into a private part file and verify it."""
    destination.parent.mkdir(parents=True, exist_ok=True)
    expected_size = int(expected_size)
    if expected_size < 0:
        raise ValueError("Silo download size cannot be negative")
    offset = destination.stat().st_size if destination.exists() else 0
    if offset > expected_size:
        destination.unlink()
        offset = 0
    request = urllib.request.Request(base + path, method="GET")
    request.add_header("Content-Type", "application/octet-stream")
    request.add_header("X-Silo-Device-Token", device_token)
    if offset:
        request.add_header("Range", f"bytes={offset}-")
    try:
        response = urllib.request.urlopen(request)
    except urllib.error.HTTPError as exc:
        raw = exc.read()
        try:
            body = json.loads(raw or b"{}")
        except (TypeError, ValueError):
            body = {"error": exc.reason}
        return exc.code, body, dict(exc.headers)
    with response:
        headers = dict(response.headers)
        if offset:
            if response.status != 206:
                raise RuntimeError("Silo server did not honor the download range")
            content_range = str(_silo_header(headers, "Content-Range"))
            expected_prefix = f"bytes {offset}-"
            if not content_range.startswith(expected_prefix) or not content_range.endswith(
                    f"/{expected_size}"):
                raise RuntimeError("Silo download returned an unexpected content range")
        elif response.status != 200:
            raise RuntimeError(f"Silo download returned unexpected status {response.status}")
        header_size = _silo_header(headers, "X-Silo-Ciphertext-Size")
        header_hash = _silo_header(headers, "X-Silo-Ciphertext-SHA256")
        if int(header_size or expected_size) != expected_size:
            raise RuntimeError("Silo download size header does not match metadata")
        if header_hash and str(header_hash) != expected_sha256:
            raise RuntimeError("Silo download hash header does not match metadata")
        announced = _silo_header(headers, "Content-Length")
        expected_response_size = expected_size - offset
        if announced and int(announced) != expected_response_size:
            raise RuntimeError("Silo download content length does not match its range")
        written = 0
        with destination.open("ab" if offset else "wb") as handle:
            while True:
                chunk = response.read(max(1, int(chunk_size)))
                if not chunk:
                    break
                handle.write(chunk)
                written += len(chunk)
            handle.flush()
            os.fsync(handle.fileno())
        if written != expected_response_size or destination.stat().st_size != expected_size:
            raise RuntimeError("Silo download ended before the complete ciphertext arrived")
    actual = _silo_file_hash(destination, chunk_size=chunk_size)
    if actual != expected_sha256:
        try:
            destination.unlink()
        except FileNotFoundError:
            pass
        raise ValueError("Silo ciphertext hash verification failed")
    return 200 if not offset else 206, {
        "size_bytes": expected_size, "sha256": actual,
    }, headers


def _silo_content_identity(key: dict, data: bytes, public_library: bool) -> str:
    digest = _silo_local_hash(data)
    if public_library:
        return "sha256:" + digest
    private = base64.b64decode(key["private"].encode("ascii"))
    return "hmac-v1:" + hmac.new(private, data, hashlib.sha256).hexdigest()


def _silo_file_digests(key: dict, path: pathlib.Path, public_library: bool,
                       chunk_size: int = 1024 * 1024) -> tuple[int, str, str]:
    """Hash a local file without reading the file into memory."""
    source_hash = hashlib.sha256()
    private_hmac = None
    if not public_library:
        private = base64.b64decode(key["private"].encode("ascii"))
        private_hmac = hmac.new(private, digestmod=hashlib.sha256)
    size = 0
    with path.open("rb") as handle:
        while True:
            chunk = handle.read(max(1, int(chunk_size)))
            if not chunk:
                break
            size += len(chunk)
            source_hash.update(chunk)
            if private_hmac is not None:
                private_hmac.update(chunk)
    digest = source_hash.hexdigest()
    identity = ("sha256:" + digest if public_library else
                "hmac-v1:" + private_hmac.hexdigest())
    return size, digest, identity


def _silo_safe_target(root: pathlib.Path, remote_path: str) -> pathlib.Path:
    normalized = str(remote_path or "").replace("\\", "/")
    parts = normalized.split("/")
    if (not normalized or normalized.startswith("/")
            or any(part in ("", ".", "..") for part in parts)):
        raise ValueError("Silo returned an unsafe relative path")
    target = (root / pathlib.Path(*parts)).resolve()
    resolved_root = root.resolve()
    if target != resolved_root and resolved_root not in target.parents:
        raise ValueError("Silo path escaped the local sync root")
    return target


def _silo_unique_variant(target: pathlib.Path, suffix: str) -> pathlib.Path:
    """Choose a non-destructive sibling path for a preserved local file."""
    candidate = target.with_name(target.name + suffix)
    counter = 2
    while candidate.exists():
        candidate = target.with_name(target.name + suffix + f"-{counter}")
        counter += 1
    return candidate


def _silo_preserve_local_target(target: pathlib.Path, suffix: str) -> pathlib.Path:
    preserved = _silo_unique_variant(target, suffix)
    os.replace(target, preserved)
    return preserved


def _silo_remove_empty_parents(path: pathlib.Path, root: pathlib.Path) -> None:
    """Remove empty directories left behind by a safe remote file move."""
    current = path
    while current != root and current.is_dir():
        try:
            current.rmdir()
        except OSError:
            return
        current = current.parent


def _silo_remove_tracked_local(root: pathlib.Path, relative: str, record: dict,
                               suffix: str) -> bool:
    """Remove an unchanged tracked file, or preserve a local edit visibly."""
    target = _silo_safe_target(root, relative)
    if not target.exists():
        return False
    unchanged = False
    if target.is_file() and record.get("local_sha256"):
        unchanged = _silo_file_hash(target) == record.get("local_sha256")
    if unchanged:
        target.unlink()
        print(f"removed unchanged local copy '{relative}'")
    else:
        preserved = _silo_preserve_local_target(target, suffix)
        print(f"preserved local changes as '{preserved.relative_to(root)}'")
    return True


def _silo_current_snapshot(base: str, device_token: str,
                           prefixes: tuple[str, ...]) -> tuple[dict[str, dict] | None, str]:
    """Read current safe metadata for a selection without requesting private bytes."""
    snapshot: dict[str, dict] = {}
    queries = prefixes or ("")
    for prefix in queries:
        path = "/api/silo/files"
        if prefix:
            path += "?" + urlencode({"prefix": prefix})
        status, out, _ = _silo_api(base, "GET", path, device_token=device_token)
        if status != 200:
            return None, f"Silo selection snapshot failed: {_silo_map_error(status, out)}"
        rows = out.get("files")
        if not isinstance(rows, list):
            return None, "Silo selection snapshot returned invalid file metadata"
        for item in rows:
            if not isinstance(item, dict):
                return None, "Silo selection snapshot contained invalid file metadata"
            remote_path = str(item.get("path") or "")
            if not _silo_path_selected(remote_path, prefixes):
                return None, "Silo selection snapshot returned a file outside its prefix"
            if not item.get("version_id") or not item.get("object_id"):
                return None, "Silo selection snapshot is missing a file version or object"
            prior = snapshot.get(remote_path)
            if prior is not None and prior != item:
                return None, "Silo selection snapshot returned conflicting file metadata"
            snapshot[remote_path] = item
    return snapshot, ""


def _silo_materialize_remote_file(base: str, device_token: str, key: dict,
                                  root: pathlib.Path, root_state: dict,
                                  change: dict, local_key: str | None = None,
                                  local_record: dict | None = None) -> None:
    """Download, authenticate, decrypt, and track one selected remote file."""
    remote_path = str(change.get("path") or "")
    version_id = str(change.get("version_id") or "")
    object_id = str(change.get("object_id") or "")
    if not remote_path or not version_id or not object_id:
        raise ValueError("Silo change is missing its path, version, or object")
    change_cursor = int(change.get("cursor") or root_state.get("cursor") or 0)
    status, sealed, _ = _silo_api(
        base,
        "GET", "/api/silo/objects/" + object_id + "/metadata",
        device_token=device_token,
    )
    if status != 200:
        raise RuntimeError(_silo_map_error(status, sealed))
    sealed_object = sealed.get("object") or {}
    if sealed_object.get("id") != object_id:
        raise ValueError("Silo metadata object id does not match the change")
    expected_size = int(sealed.get("size_bytes"))
    expected_sha256 = str(
        sealed_object.get("integrity", {}).get("ciphertext_sha256", ""))
    if len(expected_sha256) != 64:
        raise ValueError("Silo metadata is missing a ciphertext hash")
    ciphertext_part = _silo_download_part(root, object_id, version_id)
    download_status, download_out, _ = _silo_download_to_file(
        base, "/api/silo/objects/" + object_id + "/download",
        device_token, ciphertext_part, expected_size, expected_sha256)
    if download_status not in (200, 206):
        raise RuntimeError(_silo_map_error(download_status, download_out))
    private = (None if sealed_object.get("posture") == "P0"
               else key["private"])
    target = _silo_safe_target(root, remote_path)
    target.parent.mkdir(parents=True, exist_ok=True)
    if target.exists():
        local_hash = _silo_file_hash(target)
        if not local_record or local_hash != local_record.get("local_sha256"):
            suffix = (f".local-conflict-{change_cursor}"
                      if change_cursor else ".local-conflict-snapshot")
            target = _silo_unique_variant(target, suffix)
            print(f"preserved local changes as '{target.relative_to(root)}'")
    try:
        _, plaintext_size, plaintext_hash = envelope.unseal_file(
            sealed_object, ciphertext_part, target, private)
    finally:
        try:
            ciphertext_part.unlink()
        except FileNotFoundError:
            pass
    local_path = target.relative_to(root).as_posix()
    if local_key and local_key != local_path:
        root_state["files"].pop(local_key, None)
    root_state["files"][local_path] = {
        "file_id": change.get("file_id") or (local_record or {}).get("file_id", ""),
        "version_id": version_id,
        "object_id": object_id,
        "path": local_path,
        "remote_path": remote_path,
        "local_sha256": plaintext_hash,
        "size_bytes": plaintext_size,
        "mode": ("public-library" if sealed_object.get("posture") == "P0"
                 else "private"),
        "change_cursor": change_cursor,
        "change_device_id": str(change.get("device_id") or ""),
    }
    verb = "bootstrapped" if change.get("snapshot") else "pulled"
    print(f"{verb} '{remote_path}'")


def _silo_apply_remote_move(base: str, device_token: str, key: dict,
                            root: pathlib.Path, root_state: dict,
                            change: dict, local_key: str | None = None,
                            local_record: dict | None = None) -> None:
    """Apply a remote path move without downloading unchanged ciphertext."""
    remote_path = str(change.get("path") or "")
    if not remote_path:
        raise ValueError("Silo move is missing its destination path")
    if local_key and local_record:
        old_target = _silo_safe_target(root, local_key)
        new_target = _silo_safe_target(root, remote_path)
        unchanged = False
        if old_target.exists() and old_target.is_file():
            expected_hash = str(local_record.get("local_sha256") or "")
            unchanged = bool(expected_hash and _silo_file_hash(old_target) == expected_hash)
        if unchanged and (not new_target.exists() or new_target == old_target):
            new_target.parent.mkdir(parents=True, exist_ok=True)
            if new_target != old_target:
                os.replace(old_target, new_target)
                _silo_remove_empty_parents(old_target.parent, root)
            updated = dict(local_record)
            updated.update({
                "path": remote_path,
                "remote_path": remote_path,
                "change_cursor": int(change.get("cursor") or 0),
                "change_device_id": str(change.get("device_id") or ""),
            })
            root_state["files"].pop(local_key, None)
            root_state["files"][remote_path] = updated
            print(f"moved '{local_key}' to '{remote_path}'")
            return
        if old_target.exists() and old_target.is_file() and not unchanged:
            preserved = _silo_preserve_local_target(
                old_target, f".local-before-move-{int(change.get('cursor') or 0)}")
            print(f"preserved local changes as '{preserved.relative_to(root)}'")
    _silo_materialize_remote_file(
        base, device_token, key, root, root_state, change,
        local_key=local_key, local_record=local_record)


def _silo_reconcile_selection(base: str, device_token: str, state: dict,
                              root: pathlib.Path, root_state: dict, key: dict,
                              prefixes: tuple[str, ...]) -> bool:
    """Reconcile a changed device selection before consuming its global cursor."""
    try:
        previous = _silo_normalize_prefixes(root_state.get("include_prefixes", []))
    except (TypeError, ValueError) as exc:
        print(f"Silo saved selection is invalid: {exc}", file=sys.stderr)
        return False
    if previous == prefixes:
        return True
    snapshot, detail = _silo_current_snapshot(base, device_token, prefixes)
    if snapshot is None:
        print(detail, file=sys.stderr)
        return False
    cursor = int(root_state.get("cursor") or 0)
    for remote_path in sorted(snapshot):
        item = snapshot[remote_path]
        local_key, local_record = _silo_find_record(
            root_state["files"], remote_path, str(item.get("version_id") or ""),
            str(item.get("file_id") or ""))
        if (local_record
                and local_record.get("version_id") == item.get("version_id")
                and local_record.get("object_id") == item.get("object_id")):
            target = _silo_safe_target(root, local_key or remote_path)
            if target.exists():
                continue
        change = dict(item)
        change.update({
            "path": remote_path,
            "cursor": cursor,
            "action": item.get("action") or "update",
            "device_id": "",
            "snapshot": True,
        })
        try:
            _silo_materialize_remote_file(
                base, device_token, key, root, root_state, change,
                local_key=local_key, local_record=local_record)
        except (OSError, ValueError, RuntimeError, KeyError) as exc:
            print(f"Silo selection bootstrap failed for '{remote_path}': {exc}",
                  file=sys.stderr)
            return False

    for relative, record in list(root_state["files"].items()):
        if not isinstance(relative, str) or not isinstance(record, dict):
            continue
        remote_path = str(record.get("remote_path") or record.get("path") or relative)
        if _silo_path_selected(remote_path, prefixes) and remote_path in snapshot:
            continue
        try:
            _silo_remove_tracked_local(
                root, relative, record, ".local-before-selection-change")
        except (OSError, ValueError) as exc:
            print(f"Silo selection could not reconcile '{relative}': {exc}",
                  file=sys.stderr)
            return False
        root_state["files"].pop(relative, None)
    root_state["include_prefixes"] = list(prefixes)
    root_state["selection_initialized"] = True
    save_silo_state(state)
    selected = ", ".join(prefixes) if prefixes else "the full remote library"
    print(f"Silo local retention selection: {selected}")
    return True


def _silo_write_bytes(path: pathlib.Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
    temporary_path = pathlib.Path(temporary)
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary_path, path)
    finally:
        if temporary_path.exists():
            temporary_path.unlink()


def _silo_stage_upload(root: pathlib.Path, relative: str, source: pathlib.Path,
                       previous: dict, key: dict, args,
                       recipient_keys: list[str] | None = None,
                       source_size_bytes: int | None = None,
                       source_hash: str | None = None,
                       content_identity: str | None = None) -> tuple[dict, pathlib.Path, pathlib.Path]:
    public_library = bool(getattr(args, "public_library", False))
    if source_size_bytes is None or source_hash is None or content_identity is None:
        source_size_bytes, source_hash, content_identity = _silo_file_digests(
            key, source, public_library)
    mode = "public-library" if public_library else "private"
    recipients = [] if public_library else list(recipient_keys or [key["public"]])
    recipient_refs = sorted(envelope.fingerprint(public) for public in recipients)
    version_seed = f"{relative}\0{source_hash}\0{mode}".encode("utf-8")
    version_id = "ver-" + hashlib.sha256(version_seed).hexdigest()[:48]
    file_id = str(previous.get("file_id") or ("file-" + uuid.uuid4().hex))
    upload_dir = _silo_upload_dir()
    upload_dir.mkdir(parents=True, exist_ok=True)
    meta_path = upload_dir / f"{version_id}.json"
    blob_path = upload_dir / f"{version_id}.blob"
    if meta_path.exists() and blob_path.exists():
        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            meta = {}
        if (meta.get("source_sha256") == source_hash
                and meta.get("path") == relative
                and meta.get("mode") == mode
                and sorted(meta.get("recipient_refs", [])) == recipient_refs):
            return meta, meta_path, blob_path

    obj_id = "obj-" + hashlib.sha256(
        f"{file_id}\0{version_id}".encode("utf-8")).hexdigest()[:32]
    blob_temp_fd, blob_temp_name = tempfile.mkstemp(
        prefix=f".{version_id}.", dir=str(upload_dir))
    blob_temp_path = pathlib.Path(blob_temp_name)
    try:
        os.close(blob_temp_fd)
        obj, actual_source_size, ciphertext_size, sealed_source_hash = envelope.seal_file(
            source, blob_temp_path,
            posture="P0" if public_library else "P2",
            storage_class="hot",
            residency_scope=getattr(args, "scope", "CA"),
            obj_id=obj_id,
            recipients=recipients,
            metadata={
                "filename": pathlib.Path(relative).name,
                "size": source_size_bytes,
                "client": "havre-remote-silo/1",
                "mode": mode,
            },
        )
        if actual_source_size != source_size_bytes or sealed_source_hash != source_hash:
            raise ValueError("source changed while Silo encrypted it")
        os.chmod(blob_temp_path, 0o600)
        os.replace(blob_temp_path, blob_path)
    finally:
        if blob_temp_path.exists():
            blob_temp_path.unlink()

    meta = {
        "path": relative,
        "file_id": file_id,
        "version_id": version_id,
        "base_version_id": previous.get("version_id") or None,
        "source_sha256": source_hash,
        "content_identity": content_identity,
        "mode": mode,
        "recipient_refs": recipient_refs,
        "object": obj,
        "blob_sha256": obj["integrity"]["ciphertext_sha256"],
        "expected_size_bytes": ciphertext_size,
    }
    _write_private_json(meta_path, meta)
    return meta, meta_path, blob_path


def _silo_upload_staged(base: str, device_token: str, meta: dict,
                        blob_path: pathlib.Path, args) -> tuple[bool, dict]:
    blob_size = blob_path.stat().st_size
    if _silo_file_hash(blob_path) != meta["blob_sha256"]:
        print(f"staged ciphertext is corrupt for '{meta['path']}'", file=sys.stderr)
        return False, {}
    part_size = max(1, int(getattr(args, "part_size_bytes", 8 * 1024 * 1024)))
    start_body = {
        "object": meta["object"],
        "path": meta["path"],
        "file_id": meta["file_id"],
        "version_id": meta["version_id"],
        "base_version_id": meta.get("base_version_id"),
        "expected_size_bytes": blob_size,
        "expected_sha256": meta["blob_sha256"],
        "content_identity": meta["content_identity"],
        "part_size_bytes": part_size,
    }
    status, out, _ = _silo_api(
        base, "POST", "/api/silo/uploads", start_body, device_token=device_token)
    if status not in (200, 201):
        print(f"Silo upload start failed: {_silo_map_error(status, out)}", file=sys.stderr)
        return False, {}
    upload = out.get("upload", out)
    upload_id = upload.get("upload_id")
    if not upload_id:
        return True, out

    offset = int(upload.get("next_offset", 0))
    with blob_path.open("rb") as handle:
        while offset < blob_size:
            handle.seek(offset)
            chunk = handle.read(min(part_size, blob_size - offset))
            if not chunk:
                print("Silo staged ciphertext ended before the declared size", file=sys.stderr)
                return False, {}
            end = offset + len(chunk)
            status, chunk_out, headers = _silo_api(
                base,
                "PATCH",
                f"/api/silo/uploads/{upload_id}",
                chunk,
                device_token=device_token,
                headers={"Upload-Offset": str(offset)},
            )
            if status == 200:
                upload = chunk_out.get("upload", chunk_out)
                next_offset = int(upload.get("next_offset", end))
                if next_offset <= offset or next_offset > blob_size:
                    print("Silo upload made invalid forward progress", file=sys.stderr)
                    return False, {}
                offset = next_offset
                continue
            if status == 409:
                recovery = headers.get("Upload-Offset") or (
                    chunk_out.get("upload", {}).get("next_offset")
                    if isinstance(chunk_out.get("upload"), dict) else None)
                if recovery is None:
                    print(f"Silo upload offset conflict: {_silo_map_error(status, chunk_out)}",
                          file=sys.stderr)
                    return False, {}
                recovered_offset = int(recovery)
                if recovered_offset < 0 or recovered_offset > blob_size:
                    print("Silo returned an invalid resume offset", file=sys.stderr)
                    return False, {}
                offset = recovered_offset
                continue
            if status == 0:
                probe_status, probe, _ = _silo_api(
                    base, "GET", f"/api/silo/uploads/{upload_id}", device_token=device_token)
                if probe_status == 200:
                    offset = int(probe.get("upload", {}).get("next_offset", offset))
                    if offset < 0 or offset > blob_size:
                        print("Silo returned an invalid resume offset", file=sys.stderr)
                        return False, {}
                    continue
            print(f"Silo upload chunk failed: {_silo_map_error(status, chunk_out)}",
                  file=sys.stderr)
            return False, {}

    status, completed, _ = _silo_api(
        base, "POST", f"/api/silo/uploads/{upload_id}/complete",
        device_token=device_token)
    if status not in (200, 201):
        print(f"Silo upload completion failed: {_silo_map_error(status, completed)}",
              file=sys.stderr)
        return False, {}
    return True, completed


def _silo_record_from_upload(meta: dict, result: dict, source_hash: str,
                             previous: dict | None = None) -> dict:
    version = result.get("version") if isinstance(result.get("version"), dict) else {}
    change = result.get("change") if isinstance(result.get("change"), dict) else {}
    previous = previous or {}
    return {
        "file_id": version.get("file_id") or meta["file_id"],
        "version_id": version.get("version_id") or meta["version_id"],
        "object_id": version.get("object_id") or result.get("id") or meta["object"]["id"],
        "path": version.get("path") or meta["path"],
        "remote_path": version.get("path") or meta["path"],
        "local_sha256": source_hash,
        "mode": meta["mode"],
        "change_cursor": int(change.get("cursor") or previous.get("change_cursor") or 0),
        "change_device_id": str(change.get("device_id")
                                  or previous.get("change_device_id") or ""),
    }


def _silo_change_touches_record(change: dict, record: dict) -> bool:
    """Return whether a remote change is newer than one local tracking record."""
    try:
        change_cursor = int(change.get("cursor") or 0)
        known_cursor = int(record.get("change_cursor") or 0)
    except (AttributeError, TypeError, ValueError):
        return False
    if change_cursor <= known_cursor:
        return False
    remote_path = str(record.get("remote_path") or record.get("path") or "")
    object_id = str(record.get("object_id") or "")
    return bool(
        (remote_path and str(change.get("path") or "") == remote_path)
        or (object_id and str(change.get("object_id") or "") == object_id)
    )


def _silo_changes_for_delete_guard(base: str, device_token: str,
                                   root_state: dict) -> tuple[bool, list[dict], str]:
    """Read one complete change page before propagating local deletions.

    A full page is required for a safe decision. If the feed has more pages, the
    normal pull cycle will catch up first and a later push may retry the deletion.
    """
    try:
        cursor = max(0, int(root_state.get("cursor") or 0))
    except (AttributeError, TypeError, ValueError):
        return False, [], "local Silo cursor is invalid"
    status, out, _ = _silo_api(
        base, "GET", "/api/silo/changes?" + urlencode({"cursor": cursor, "limit": 500}),
        device_token=device_token,
    )
    if status != 200:
        return False, [], f"Silo deletion guard could not read the change feed: {_silo_map_error(status, out)}"
    changes = out.get("changes")
    if not isinstance(changes, list) or not isinstance(out.get("has_more"), bool):
        return False, [], "Silo deletion guard received an invalid change feed"
    if out.get("has_more"):
        return False, changes, "Silo deletion guard deferred while the remote change feed has more pages"
    return True, [change for change in changes if isinstance(change, dict)], ""


def _silo_propagate_missing(base: str, device_token: str, state: dict,
                            root: pathlib.Path, root_state: dict,
                            prefixes: tuple[str, ...] = ()) -> tuple[bool, int, int]:
    """Propagate selected tracked removals without deleting a newer remote change."""
    tracked = root_state.get("files") or {}
    if not isinstance(tracked, dict):
        print("Silo local file state is invalid; refusing deletion propagation", file=sys.stderr)
        return False, 0, 1
    missing = []
    for relative, record in list(tracked.items()):
        if not isinstance(relative, str) or not isinstance(record, dict):
            continue
        remote_path = str(record.get("remote_path") or record.get("path") or relative)
        if not _silo_path_selected(remote_path, prefixes):
            continue
        try:
            target = _silo_safe_target(root, relative)
        except ValueError as exc:
            print(f"Silo local file state is unsafe for '{relative}': {exc}", file=sys.stderr)
            return False, 0, 1
        if not target.exists():
            missing.append((relative, record))
    if not missing:
        return True, 0, 0

    guard_ready, changes, detail = _silo_changes_for_delete_guard(
        base, device_token, root_state)
    if not guard_ready:
        if detail.endswith("more pages"):
            print(detail)
            return True, 0, 0
        print(detail, file=sys.stderr)
        return False, 0, 1

    deleted = failed = 0
    for relative, record in missing:
        if any(_silo_change_touches_record(change, record) for change in changes):
            print(f"deferred local deletion '{relative}' until the newer remote change is pulled")
            continue
        object_id = str(record.get("object_id") or "")
        if not object_id:
            print(f"Silo cannot propagate local deletion '{relative}': object identity is missing",
                  file=sys.stderr)
            failed += 1
            continue
        status, out, _ = _silo_api(
            base,
            "DELETE",
            "/api/silo/objects/" + object_id,
            {"device_id": state.get("device_id"), "path": relative},
            device_token=device_token,
        )
        if status in (200, 404):
            root_state["files"].pop(relative, None)
            deleted += 1
            if status == 200:
                print(f"propagated local deletion '{relative}' (recoverable remote tombstone)")
            else:
                print(f"reconciled already-deleted remote file '{relative}'")
            continue
        failed += 1
        print(f"Silo deletion failed for '{relative}': {_silo_map_error(status, out)}",
              file=sys.stderr)
    if deleted:
        save_silo_state(state)
    return failed == 0, deleted, failed


# ── SILO COMMANDS ─────────────────────────────────────────────────────────────

def cmd_silo_catalog(args, key) -> int:
    base = server_url(args)
    status, out, _ = _silo_api(base, "GET", "/api/silo/catalog")
    if status == 200:
        print(json.dumps(out, indent=2))
        return 0
    print(f"Silo catalogue failed: {_silo_map_error(status, out)}", file=sys.stderr)
    return 1


def _silo_grant_existing_objects(base: str, device_token: str, source_key: dict,
                                 target_id: str, target_public: str) -> dict:
    """Add a target device wrapper to every existing private object."""
    summary = {"granted": 0, "skipped": 0, "failed": 0}
    if not target_id or not target_public:
        print("target device metadata is incomplete", file=sys.stderr)
        summary["failed"] = 1
        return summary
    status, objects_out, _ = _silo_api(
        base, "GET", "/api/silo/objects", device_token=device_token)
    if status != 200:
        print(f"Silo object list failed: {_silo_map_error(status, objects_out)}",
              file=sys.stderr)
        summary["failed"] = 1
        return summary
    for object_id in objects_out.get("ids", []):
        status, sealed, _ = _silo_api(
            base, "GET", f"/api/silo/objects/{object_id}", device_token=device_token)
        if status != 200:
            summary["failed"] += 1
            print(f"could not read '{object_id}': {_silo_map_error(status, sealed)}",
                  file=sys.stderr)
            continue
        obj = sealed.get("object") or {}
        if obj.get("posture") == "P0":
            summary["skipped"] += 1
            continue
        try:
            updated = envelope.add_recipient(obj, source_key["private"], target_public)
        except (TypeError, ValueError) as exc:
            summary["failed"] += 1
            print(f"could not re-wrap '{object_id}': {exc}", file=sys.stderr)
            continue
        if updated == obj:
            summary["skipped"] += 1
            continue
        status, result, _ = _silo_api(
            base,
            "POST",
            f"/api/silo/objects/{object_id}/grant",
            {"target_device_id": target_id, "object": updated},
            device_token=device_token,
        )
        if status not in (200, 201):
            summary["failed"] += 1
            print(f"grant failed for '{object_id}': {_silo_map_error(status, result)}",
                  file=sys.stderr)
            continue
        summary["granted"] += 1
    return summary


def _recovery_output_path(args) -> pathlib.Path:
    return pathlib.Path(getattr(args, "output", None) or (_home() / "silo-recovery.json")) \
        .expanduser().resolve()


def _cmd_silo_recovery_create(args, key, base: str, token: str, state: dict) -> int:
    output_path = _recovery_output_path(args)
    if output_path.exists():
        print(f"recovery bundle already exists: {output_path} (choose another --output)",
              file=sys.stderr)
        return 2
    try:
        password = _recovery_password(confirm=True)
    except (EOFError, KeyboardInterrupt, ValueError) as exc:
        print(f"Silo recovery bundle was not created: {exc}", file=sys.stderr)
        return 2

    private_key, public_key = envelope.gen_recipient_keypair()
    status, out, _ = _silo_api(
        base,
        "POST",
        "/api/silo/devices",
        {"name": args.name, "public_key": public_key},
        token=token,
    )
    if status != 201:
        print(f"Silo recovery device registration failed: {_silo_map_error(status, out)}",
              file=sys.stderr)
        return 1
    device = out.get("device", {})
    device_id = str(device.get("id") or "")
    device_token = str(device.get("device_token") or "")
    device_name = str(device.get("name") or args.name)
    if not device_id or not device_token:
        print("Silo recovery device registration returned incomplete credentials",
              file=sys.stderr)
        return 1
    payload = {
        "device_id": device_id,
        "device_name": device_name,
        "device_token": device_token,
        "server": base,
        "private_key": private_key,
        "public_key": public_key,
        "created_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "purpose": "silo-device-recovery",
    }
    try:
        bundle = _seal_recovery_bundle(payload, password)
        _write_private_json(output_path, bundle)
    except (OSError, ValueError, RuntimeError) as exc:
        # The one-time device token must not be left active if the local bundle
        # could not be written. The token itself is never printed.
        rollback_status, _ = _api(
            base, "POST", f"/api/silo/devices/{device_id}/revoke", token=token)
        if rollback_status == 200:
            detail = "registration was rolled back"
        else:
            detail = (f"registration may still be active; revoke device {device_id} "
                      "from the account")
        print(f"Silo recovery bundle could not be saved; {detail}: {exc}", file=sys.stderr)
        return 1

    summary = {"granted": 0, "skipped": 0, "failed": 0}
    if state.get("device_id") and state.get("device_token"):
        try:
            source_key = _silo_active_key(key)
            summary = _silo_grant_existing_objects(
                base, state["device_token"], source_key, device_id, public_key)
        except RuntimeError as exc:
            print(f"existing Silo objects were not granted: {exc}", file=sys.stderr)
            summary["failed"] = 1
    else:
        state.update({
            "device_id": device_id,
            "device_name": device_name,
            "device_token": device_token,
            "silo_private": private_key,
            "silo_public": public_key,
            "server": base,
            "roots": {},
        })
        save_silo_state(state)

    print(f"created encrypted Silo recovery bundle: {output_path}")
    print(f"recovery device: {device_id} ({envelope.fingerprint(public_key)})")
    if state.get("device_id") == device_id:
        print("recovery device is now the active local Silo device")
    else:
        print(f"existing objects: {summary['granted']} granted, "
              f"{summary['skipped']} skipped, {summary['failed']} failed")
    return 1 if summary["failed"] else 0


def _cmd_silo_recovery_import(args) -> int:
    raw_bundle_path = getattr(args, "bundle", "") or ""
    if not raw_bundle_path:
        print("--bundle is required for recovery-import", file=sys.stderr)
        return 2
    bundle_path = pathlib.Path(raw_bundle_path).expanduser().resolve()
    try:
        bundle = json.loads(bundle_path.read_text(encoding="utf-8"))
        password = _recovery_password(confirm=False)
        payload = _open_recovery_bundle(bundle, password)
    except (OSError, ValueError, UnicodeDecodeError, EOFError, KeyboardInterrupt) as exc:
        print(f"Silo recovery import failed: {exc}", file=sys.stderr)
        return 1
    current = load_silo_state()
    if any(current.get(field) for field in ("device_id", "device_token", "silo_private")) \
            and not getattr(args, "replace", False):
        print("an active local Silo device already exists; use --replace only after preserving it",
              file=sys.stderr)
        return 2
    save_silo_state({
        "device_id": payload["device_id"],
        "device_name": payload["device_name"],
        "device_token": payload["device_token"],
        "silo_private": payload["private_key"],
        "silo_public": payload["public_key"],
        "server": payload["server"],
        "roots": {},
    })
    print(f"imported Silo recovery device {payload['device_id']} "
          f"({envelope.fingerprint(payload['public_key'])})")
    print("local sync roots were reset; run silo-pull or silo-sync against a new root")
    return 0


def cmd_silo_device(args, key) -> int:
    if args.action == "recovery-import":
        return _cmd_silo_recovery_import(args)
    base, token = _require_token(args)
    state = load_silo_state()
    if args.action == "register":
        device_key = _silo_active_key(key)
        status, out, _ = _silo_api(
            base,
            "POST",
            "/api/silo/devices",
            {"name": args.name, "public_key": device_key["public"]},
            token=token,
        )
        if status != 201:
            print(f"Silo device registration failed: {_silo_map_error(status, out)}",
                  file=sys.stderr)
            return 1
        device = out.get("device", {})
        state.update({
            "device_id": device.get("id", ""),
            "device_name": device.get("name", args.name),
            "device_token": device.get("device_token", ""),
            "silo_private": device_key["private"],
            "silo_public": device_key["public"],
            "server": base,
        })
        save_silo_state(state)
        print(f"registered Silo device '{device.get('name', args.name)}' ({device.get('id')})")
        print(f"key fingerprint: {envelope.fingerprint(device_key['public'])}")
        return 0
    if args.action == "recovery-create":
        return _cmd_silo_recovery_create(args, key, base, token, state)
    if args.action == "grant":
        target_id = args.id or ""
        if not target_id:
            print("a target device id is required for grant", file=sys.stderr)
            return 2
        try:
            device_token, state = _silo_device_required()
            source_key = _silo_active_key(key)
        except SystemExit:
            return 2
        except RuntimeError as exc:
            print(f"Silo device grant failed: {exc}", file=sys.stderr)
            return 1
        status, devices_out, _ = _silo_api(
            base, "GET", "/api/silo/devices", device_token=device_token)
        if status != 200:
            print(f"Silo device grant failed: {_silo_map_error(status, devices_out)}",
                  file=sys.stderr)
            return 1
        target = next((item for item in devices_out.get("devices", [])
                       if item.get("id") == target_id and item.get("status") == "active"), None)
        if not target:
            print("active target device was not found", file=sys.stderr)
            return 1
        summary = _silo_grant_existing_objects(
            base, device_token, source_key, target_id, str(target.get("public_key") or ""))
        print(f"Silo device grant: {summary['granted']} granted, "
              f"{summary['skipped']} skipped, {summary['failed']} failed")
        return 1 if summary["failed"] else 0
    if args.action == "list":
        status, out, _ = _silo_api(base, "GET", "/api/silo/devices", token=token)
        if status == 200:
            print(json.dumps(out, indent=2))
            return 0
        print(f"Silo device list failed: {_silo_map_error(status, out)}", file=sys.stderr)
        return 1
    device_id = args.id or state.get("device_id")
    if not device_id:
        print("a device id is required", file=sys.stderr)
        return 2
    status, out, _ = _silo_api(
        base, "POST", f"/api/silo/devices/{device_id}/revoke", token=token)
    if status != 200:
        print(f"Silo device revoke failed: {_silo_map_error(status, out)}", file=sys.stderr)
        return 1
    if device_id == state.get("device_id"):
        state.update({"device_id": "", "device_name": "", "device_token": "",
                      "silo_private": "", "silo_public": ""})
        save_silo_state(state)
    print(f"revoked Silo device {device_id}")
    return 0


def cmd_silo_status(args, key) -> int:
    base, token = _require_token(args)
    status, out, _ = _silo_api(base, "GET", "/api/silo/space", token=token)
    if status != 200:
        print(f"Silo status failed: {_silo_map_error(status, out)}", file=sys.stderr)
        return 1
    state = load_silo_state()
    print(json.dumps({
        "space": out.get("space"),
        "checkout_ready": out.get("checkout_ready"),
        "device": {
            "id": state.get("device_id"),
            "name": state.get("device_name"),
            "registered": bool(state.get("device_token")),
        },
    }, indent=2))
    return 0


def _silo_normalize_local_path(value: str) -> str:
    normalized = str(value or "").strip().replace("\\", "/").strip("/")
    parts = normalized.split("/")
    if (not normalized or any(part in ("", ".", "..") for part in parts)):
        raise ValueError("Silo local paths must be non-empty and relative")
    return normalized


def _silo_local_prefix_move_plan(root: pathlib.Path, root_state: dict,
                                 old_prefix: str, new_prefix: str,
                                 remote_files: list[dict]) -> list[tuple[str, dict]]:
    """Validate local files before a remote folder move is committed."""
    old_target = _silo_safe_target(root, old_prefix)
    new_target = _silo_safe_target(root, new_prefix)
    if not old_target.exists() or not old_target.is_dir():
        raise ValueError(f"Silo folder is not a local directory: '{old_prefix}'")
    if new_target.exists():
        raise ValueError(f"Silo destination folder already exists: '{new_prefix}'")
    remote_by_path = {str(item.get("path") or ""): item for item in remote_files}
    if any(not path.startswith(old_prefix + "/") for path in remote_by_path):
        raise ValueError("Silo folder snapshot contains a path outside the source folder")

    local_files = []
    for candidate in old_target.rglob("*"):
        if candidate.is_symlink():
            raise ValueError("Silo folder moves do not follow symbolic links")
        if candidate.is_file():
            local_files.append(candidate.relative_to(root).as_posix())

    plan = []
    for local_path in local_files:
        record = root_state.get("files", {}).get(local_path)
        if not isinstance(record, dict):
            raise ValueError(f"Silo folder contains an unsynced local file: '{local_path}'")
        remote_path = str(record.get("remote_path") or record.get("path") or local_path)
        if remote_path not in remote_by_path:
            raise ValueError(f"Silo local state is stale for '{local_path}'")
        expected_hash = str(record.get("local_sha256") or "")
        if not expected_hash or _silo_file_hash(root / local_path) != expected_hash:
            raise ValueError(f"Silo move refused: sync the changed file before moving '{local_path}'")
        plan.append((local_path, record))

    for remote_path, remote in remote_by_path.items():
        local_key, record = _silo_find_record(
            root_state.get("files", {}), remote_path,
            str(remote.get("version_id") or ""), str(remote.get("file_id") or ""))
        if record is not None and not str(local_key or "").startswith(old_prefix + "/"):
            raise ValueError(f"Silo file '{remote_path}' is retained at another local path")
    return plan


def _silo_apply_local_prefix_move(root: pathlib.Path, root_state: dict,
                                  old_prefix: str, new_prefix: str,
                                  plan: list[tuple[str, dict]],
                                  changes: list[dict]) -> None:
    """Rename the local folder and update local cursor records after the server move."""
    old_target = _silo_safe_target(root, old_prefix)
    new_target = _silo_safe_target(root, new_prefix)
    new_target.parent.mkdir(parents=True, exist_ok=True)
    os.replace(old_target, new_target)
    changes_by_file = {
        str(item.get("file_id") or ""): item for item in changes
        if isinstance(item, dict) and item.get("file_id")
    }
    for local_path, record in plan:
        suffix = local_path[len(old_prefix) + 1:]
        new_local_path = new_prefix + "/" + suffix
        remote_path = str(record.get("remote_path") or record.get("path") or local_path)
        if remote_path.startswith(old_prefix + "/"):
            remote_path = new_prefix + "/" + remote_path[len(old_prefix) + 1:]
        updated = dict(record)
        updated.update({"path": new_local_path, "remote_path": remote_path})
        change = changes_by_file.get(str(record.get("file_id") or ""))
        if change:
            updated["change_cursor"] = int(change.get("cursor") or 0)
            updated["change_device_id"] = str(change.get("device_id") or "")
        root_state["files"].pop(local_path, None)
        root_state["files"][new_local_path] = updated


def cmd_silo_move_dir(args, key) -> int:
    """Move a tracked local folder through one atomic remote prefix operation."""
    del key
    base = server_url(args)
    device_token, state = _silo_device_required()
    root = pathlib.Path(args.root).expanduser().resolve()
    if not root.exists() or not root.is_dir():
        print("Silo folder move root must be an existing directory", file=sys.stderr)
        return 2
    try:
        old_prefix = _silo_normalize_local_path(args.old_path)
        new_prefix = _silo_normalize_local_path(args.new_path)
        if old_prefix == new_prefix:
            print("Silo folder move skipped: source and destination are the same")
            return 0
        if (old_prefix.startswith(new_prefix + "/")
                or new_prefix.startswith(old_prefix + "/")):
            raise ValueError("source and destination folders cannot overlap")
        _silo_safe_target(root, old_prefix)
        _silo_safe_target(root, new_prefix)
    except ValueError as exc:
        print(f"Silo folder move cannot start: {exc}", file=sys.stderr)
        return 2

    root_state = _root_state(state, root)
    snapshot, detail = _silo_current_snapshot(base, device_token, (old_prefix,))
    if snapshot is None:
        print(detail, file=sys.stderr)
        return 1
    remote_files = list(snapshot.values())
    if not remote_files:
        # A lost response can leave the local folder in place after the server already
        # completed the move. Recognize that exact destination state and finish locally.
        destination, destination_detail = _silo_current_snapshot(
            base, device_token, (new_prefix,))
        if destination is not None:
            mapped = []
            for item in destination.values():
                path = str(item.get("path") or "")
                if path.startswith(new_prefix + "/"):
                    mapped_item = dict(item)
                    mapped_item["path"] = old_prefix + "/" + path[len(new_prefix) + 1:]
                    mapped.append(mapped_item)
            if mapped:
                old_target = _silo_safe_target(root, old_prefix)
                new_target = _silo_safe_target(root, new_prefix)
                if (not old_target.exists() and new_target.is_dir()
                        and any(str(path).startswith(new_prefix + "/")
                                for path in root_state.get("files", {}))):
                    print(f"Silo folder move already completed: '{new_prefix}'")
                    return 0
                try:
                    plan = _silo_local_prefix_move_plan(
                        root, root_state, old_prefix, new_prefix, mapped)
                    if plan:
                        _silo_apply_local_prefix_move(
                            root, root_state, old_prefix, new_prefix, plan, [])
                        save_silo_state(state)
                        print(f"completed local folder move after an idempotent server retry: '{old_prefix}' to '{new_prefix}'")
                        return 0
                except (OSError, ValueError) as exc:
                    print(f"Silo folder move retry could not finish locally: {exc}", file=sys.stderr)
                    return 1
        if destination is None and destination_detail:
            print(destination_detail, file=sys.stderr)
            return 1
        print(f"Silo folder move skipped: no synced files under '{old_prefix}'")
        return 0

    for item in remote_files:
        if not item.get("file_id") or not item.get("object_id") or not item.get("version_id"):
            print("Silo folder move refused: the server returned incomplete file metadata", file=sys.stderr)
            return 1
    try:
        plan = _silo_local_prefix_move_plan(
            root, root_state, old_prefix, new_prefix, remote_files)
    except (OSError, ValueError) as exc:
        print(f"Silo folder move refused: {exc}", file=sys.stderr)
        return 1
    payload = {
        "device_id": state.get("device_id"),
        "old_prefix": old_prefix,
        "new_prefix": new_prefix,
        "files": [
            {
                "file_id": item["file_id"],
                "path": item["path"],
                "object_id": item["object_id"],
                "version_id": item["version_id"],
            }
            for item in remote_files
        ],
    }
    status, out, _ = _silo_api(
        base, "POST", "/api/silo/move-prefix", payload, device_token=device_token)
    if status not in (200, 201):
        print(f"Silo folder move failed: {_silo_map_error(status, out)}", file=sys.stderr)
        return 1
    move = out.get("move") if isinstance(out.get("move"), dict) else {}
    try:
        _silo_apply_local_prefix_move(
            root, root_state, old_prefix, new_prefix, plan,
            out.get("changes") if isinstance(out.get("changes"), list) else [])
    except OSError as exc:
        print(
            "Silo moved the remote folder but could not rename the local folder; "
            f"run silo-pull to reconcile it: {exc}", file=sys.stderr)
        return 1
    save_silo_state(state)
    print(f"moved folder '{old_prefix}' to '{new_prefix}' ({len(move.get('moved', []))} files)")
    return 0


def cmd_silo_move(args, key) -> int:
    """Move one tracked local file and its remote Silo membership atomically enough for retry."""
    del key
    base = server_url(args)
    device_token, state = _silo_device_required()
    root = pathlib.Path(args.root).expanduser().resolve()
    if not root.exists() or not root.is_dir():
        print("Silo move root must be an existing directory", file=sys.stderr)
        return 2
    try:
        old_relative = _silo_normalize_local_path(args.old_path)
        new_relative = _silo_normalize_local_path(args.new_path)
        old_target = _silo_safe_target(root, old_relative)
        new_target = _silo_safe_target(root, new_relative)
    except ValueError as exc:
        print(f"Silo move cannot start: {exc}", file=sys.stderr)
        return 2
    if old_relative == new_relative:
        print("Silo move skipped: source and destination are the same")
        return 0
    if not old_target.exists() or not old_target.is_file():
        print(f"Silo move source is not a file: '{old_relative}'", file=sys.stderr)
        return 2
    if new_target.exists():
        print(f"Silo move destination already exists: '{new_relative}'", file=sys.stderr)
        return 2
    root_state = _root_state(state, root)
    record = root_state.get("files", {}).get(old_relative)
    if not isinstance(record, dict):
        _, record = _silo_find_record(root_state.get("files", {}), old_relative)
    if not isinstance(record, dict):
        print("Silo move requires a file that has completed one sync", file=sys.stderr)
        return 2
    expected_hash = str(record.get("local_sha256") or "")
    try:
        unchanged = bool(expected_hash and _silo_file_hash(old_target) == expected_hash)
    except OSError as exc:
        print(f"Silo move could not verify the local file: {exc}", file=sys.stderr)
        return 1
    if not unchanged:
        print("Silo move refused: sync the changed file before moving it", file=sys.stderr)
        return 1
    object_id = str(record.get("object_id") or "")
    if not object_id or not record.get("file_id") or not record.get("version_id"):
        print("Silo move refused: local Silo file identity is incomplete", file=sys.stderr)
        return 1
    status, out, _ = _silo_api(
        base,
        "POST",
        "/api/silo/objects/" + object_id + "/move",
        {
            "device_id": state.get("device_id"),
            "file_id": record["file_id"],
            "path": old_relative,
            "new_path": new_relative,
            "version_id": record["version_id"],
        },
        device_token=device_token,
    )
    if status not in (200, 201):
        print(f"Silo move failed: {_silo_map_error(status, out)}", file=sys.stderr)
        return 1
    try:
        new_target.parent.mkdir(parents=True, exist_ok=True)
        os.replace(old_target, new_target)
    except OSError as exc:
        print(
            "Silo moved the remote path but could not rename the local file; "
            f"run silo-pull to reconcile it: {exc}",
            file=sys.stderr,
        )
        return 1
    updated = dict(record)
    updated.update({"path": new_relative, "remote_path": new_relative})
    change = out.get("change") if isinstance(out.get("change"), dict) else {}
    if change.get("device_id"):
        updated["change_device_id"] = change["device_id"]
    root_state["files"].pop(old_relative, None)
    root_state["files"][new_relative] = updated
    save_silo_state(state)
    print(f"moved '{old_relative}' to '{new_relative}'")
    return 0


def _silo_doctor_report(onboarding: dict, state: dict, *, server_devices: list[dict],
                        server_devices_status: int, key_ready: bool,
                        root: pathlib.Path | None = None) -> dict:
    """Combine safe server onboarding facts with local, read-only client checks."""
    checks = []
    for step in onboarding.get("steps", []):
        if not isinstance(step, dict) or not step.get("id"):
            continue
        checks.append({
            "id": f"server-{step['id']}",
            "label": step.get("label", step["id"]),
            "ready": bool(step.get("ready")),
            "required": bool(step.get("required", False)),
            "detail": step.get("detail", ""),
        })
    checks.extend([
        {
            "id": "local-key", "label": "Local Silo key",
            "ready": bool(key_ready), "required": True,
            "detail": ("The local Silo key can derive its matching public key."
                        if key_ready else
                        "The local Silo key is missing or inconsistent; re-import recovery material."),
        },
        {
            "id": "local-session", "label": "Local session files",
            "ready": _private_file_ok(_session_path()) and _private_file_ok(_key_path()),
            "required": True,
            "detail": "The local session and account key files are present with private permissions.",
        },
        {
            "id": "local-device", "label": "Local device state",
            "ready": bool(state.get("device_id") and state.get("device_token")
                          and state.get("silo_private") and state.get("silo_public")),
            "required": True,
            "detail": ("The local Silo device credential and independent key are available."
                        if state.get("device_id") and state.get("device_token")
                        and state.get("silo_private") and state.get("silo_public") else
                        "Register this computer with `silo-device register`."),
        },
        {
            "id": "server-device", "label": "Server device registration",
            "ready": server_devices_status == 200 and any(
                device.get("id") == state.get("device_id")
                and device.get("status") == "active" for device in server_devices),
            "required": True,
            "detail": ("The local device is active in the tenant device list."
                        if server_devices_status == 200 and any(
                            device.get("id") == state.get("device_id")
                            and device.get("status") == "active" for device in server_devices)
                        else "The local device is not active on the Silo server."),
        },
    ])
    if root is not None:
        checks.append({
            "id": "sync-root", "label": "Sync folder",
            "ready": root.exists() and root.is_dir(), "required": True,
            "detail": ("The selected sync folder exists and is a directory."
                        if root.exists() and root.is_dir() else
                        "Create or choose a local directory for Silo sync."),
        })
    ready = all(check["ready"] for check in checks if check["required"])
    next_action = next((check["id"] for check in checks
                        if check["required"] and not check["ready"]), "run-sync")
    return {
        "product": "silo",
        "ready": ready,
        "next_action": next_action,
        "server": {
            "environment": onboarding.get("environment", "unknown"),
            "sync_ready": bool(onboarding.get("sync_ready")),
        },
        "root_checked": str(root) if root is not None else None,
        "checks": checks,
    }


def cmd_silo_doctor(args, key) -> int:
    """Verify Silo setup without creating keys, changing state, or transferring files."""
    base, token = _require_token(args)
    status, onboarding, _ = _silo_api(
        base, "GET", "/api/silo/onboarding", token=token)
    if status != 200:
        print(f"Silo doctor could not read onboarding state: "
              f"{_silo_map_error(status, onboarding)}", file=sys.stderr)
        return 1
    device_status, devices_out, _ = _silo_api(
        base, "GET", "/api/silo/devices", token=token)
    state = {"device_id": "", "device_token": "", "silo_private": "",
             "silo_public": "", "roots": {}}
    key_error = ""
    try:
        state = load_silo_state()
        if not state.get("silo_private") or not state.get("silo_public"):
            raise RuntimeError("no local Silo device key is registered; run 'silo-device register'")
        _silo_active_key(key)
    except (RuntimeError, OSError, ValueError) as exc:
        key_error = str(exc)
    key_ready = not key_error
    root_value = getattr(args, "root", None)
    root = pathlib.Path(root_value).expanduser().resolve() if root_value else None
    report = _silo_doctor_report(
        onboarding, state,
        server_devices=devices_out.get("devices", []) if device_status == 200 else [],
        server_devices_status=device_status,
        key_ready=key_ready,
        root=root,
    )
    if key_error:
        for check in report["checks"]:
            if check["id"] == "local-key":
                check["detail"] = key_error
                break
    if getattr(args, "json", False):
        print(json.dumps(report, indent=2))
    else:
        print(f"Silo doctor: {'READY' if report['ready'] else 'BLOCKED'}")
        for check in report["checks"]:
            marker = "PASS" if check["ready"] else "BLOCKED"
            suffix = " *" if check["required"] else ""
            print(f"  [{marker}] {check['label']}{suffix}: {check['detail']}")
        print(f"Next action: {report['next_action']}")
    return 0 if report["ready"] else 1


def cmd_silo_push(args, key) -> int:
    base = server_url(args)
    device_token, state = _silo_device_required()
    try:
        key = _silo_active_key(key)
    except RuntimeError as exc:
        print(f"Silo push cannot use the local device key: {exc}", file=sys.stderr)
        return 1
    recipient_keys = [key["public"]]
    if not getattr(args, "public_library", False):
        status, devices_out, _ = _silo_api(
            base, "GET", "/api/silo/devices", device_token=device_token)
        if status != 200:
            print(f"Silo device key list failed: {_silo_map_error(status, devices_out)}",
                  file=sys.stderr)
            return 1
        recipient_keys = [
            str(device.get("public_key"))
            for device in devices_out.get("devices", [])
            if device.get("status") == "active" and device.get("public_key")
        ]
        if key["public"] not in recipient_keys:
            recipient_keys.append(key["public"])
    root = pathlib.Path(args.root).expanduser().resolve()
    root.mkdir(parents=True, exist_ok=True)
    root_state = _root_state(state, root)
    try:
        prefixes = _silo_effective_prefixes(args, root_state)
    except (TypeError, ValueError) as exc:
        print(f"Silo push selection is invalid: {exc}", file=sys.stderr)
        return 2
    paths = []
    for path in sorted(root.rglob("*")):
        if path.is_symlink() or not path.is_file():
            continue
        relative = path.relative_to(root).as_posix()
        if relative.startswith(".havre/") or relative.startswith(".silo/"):
            continue
        if not _silo_path_selected(relative, prefixes):
            continue
        paths.append((relative, path))
    deletion_ready, deleted, failed = _silo_propagate_missing(
        base, device_token, state, root, root_state, prefixes)
    if not deletion_ready:
        print("Silo push stopped before uploading: local deletion safety could not be established",
              file=sys.stderr)
        return 1
    uploaded = skipped = 0
    for relative, source in paths:
        try:
            source_size, source_hash, content_identity = _silo_file_digests(
                key, source, bool(getattr(args, "public_library", False)))
            previous = root_state["files"].get(relative, {})
            mode = "public-library" if args.public_library else "private"
            if (previous.get("local_sha256") == source_hash
                    and previous.get("mode") == mode):
                skipped += 1
                continue
            meta, meta_path, blob_path = _silo_stage_upload(
                root, relative, source, previous, key, args, recipient_keys,
                source_size_bytes=source_size, source_hash=source_hash,
                content_identity=content_identity)
            ok, result = _silo_upload_staged(base, device_token, meta, blob_path, args)
            if not ok:
                failed += 1
                continue
            root_state["files"][relative] = _silo_record_from_upload(
                meta, result, source_hash, previous)
            save_silo_state(state)
            try:
                meta_path.unlink()
                blob_path.unlink()
            except FileNotFoundError:
                pass
            uploaded += 1
            print(f"uploaded '{relative}'")
        except (OSError, ValueError, RuntimeError) as exc:
            failed += 1
            print(f"Silo push failed for '{relative}': {exc}", file=sys.stderr)
    save_silo_state(state)
    print(f"Silo push: {uploaded} uploaded, {deleted} deleted, {skipped} unchanged, {failed} failed")
    return 1 if failed else 0


def _silo_find_record(files: dict, remote_path: str, version_id: str = "",
                      file_id: str = ""):
    direct = files.get(remote_path)
    if direct:
        return remote_path, direct
    for local_path, record in files.items():
        if (file_id and record.get("file_id") == file_id
                or record.get("remote_path") == remote_path
                or (version_id and record.get("version_id") == version_id)):
            return local_path, record
    return None, None


def cmd_silo_pull(args, key) -> int:
    base = server_url(args)
    device_token, state = _silo_device_required()
    try:
        key = _silo_active_key(key)
    except RuntimeError as exc:
        print(f"Silo pull cannot use the local device key: {exc}", file=sys.stderr)
        return 1
    root = pathlib.Path(args.root).expanduser().resolve()
    root.mkdir(parents=True, exist_ok=True)
    root_state = _root_state(state, root)
    try:
        prefixes = _silo_effective_prefixes(args, root_state)
    except (TypeError, ValueError) as exc:
        print(f"Silo pull selection is invalid: {exc}", file=sys.stderr)
        return 2
    if not _silo_reconcile_selection(
            base, device_token, state, root, root_state, key, prefixes):
        return 1
    cursor = int(root_state.get("cursor", 0))
    applied = skipped = failed = 0
    limit = max(1, min(int(getattr(args, "limit", 100)), 500))
    while True:
        query = urlencode({"cursor": cursor, "limit": limit})
        status, out, _ = _silo_api(
            base, "GET", "/api/silo/changes?" + query, device_token=device_token)
        if status != 200:
            print(f"Silo change feed failed: {_silo_map_error(status, out)}", file=sys.stderr)
            return 1
        changes = out.get("changes", [])
        for change in changes:
            change_cursor = int(change.get("cursor", cursor))
            remote_path = str(change.get("path") or "")
            version_id = str(change.get("version_id") or "")
            local_key, local_record = _silo_find_record(
                root_state["files"], remote_path, version_id,
                str(change.get("file_id") or ""))
            known_cursor = int((local_record or {}).get("change_cursor") or 0)
            same_version = (local_record
                            and local_record.get("version_id") == version_id)
            same_remote_path = bool(
                local_record
                and str(local_record.get("remote_path") or local_record.get("path") or "")
                == remote_path
            )
            remote_selected = _silo_path_selected(remote_path, prefixes)
            tracked_selected = bool(
                local_record and _silo_path_selected(
                    str(local_record.get("remote_path")
                        or local_record.get("path") or local_key or ""), prefixes))
            if (local_record and known_cursor and change_cursor <= known_cursor) or (
                    remote_selected
                    and change.get("action") != "delete"
                    and change.get("device_id") == state.get("device_id")
                    and same_version
                    and (change.get("action") != "move" or same_remote_path)):
                cursor = change_cursor
                root_state["cursor"] = cursor
                save_silo_state(state)
                skipped += 1
                continue
            if not remote_selected and not tracked_selected:
                cursor = change_cursor
                root_state["cursor"] = cursor
                save_silo_state(state)
                skipped += 1
                continue
            if not remote_selected:
                try:
                    if local_key:
                        _silo_remove_tracked_local(
                            root, local_key, local_record,
                            f".local-before-unselected-{change_cursor}")
                        root_state["files"].pop(local_key, None)
                    cursor = change_cursor
                    root_state["cursor"] = cursor
                    save_silo_state(state)
                    applied += 1
                    print(f"removed unselected remote path '{remote_path}' from local retention")
                except (OSError, ValueError) as exc:
                    failed += 1
                    print(f"Silo pull failed at cursor {change_cursor}: {exc}", file=sys.stderr)
                    return 1
                continue
            try:
                target = _silo_safe_target(root, remote_path)
                if change.get("action") == "delete":
                    unchanged = (target.exists() and local_record
                                 and _silo_file_hash(target)
                                 == local_record.get("local_sha256"))
                    if target.exists() and not unchanged:
                        preserved = target.with_name(
                            target.name + f".local-before-delete-{change_cursor}")
                        os.replace(target, preserved)
                        print(f"preserved local changes as '{preserved.relative_to(root)}'")
                    elif unchanged:
                        target.unlink()
                    if local_key:
                        root_state["files"].pop(local_key, None)
                    print(f"applied remote deletion '{remote_path}'")
                elif change.get("action") == "move":
                    _silo_apply_remote_move(
                        base, device_token, key, root, root_state, change,
                        local_key=local_key, local_record=local_record)
                else:
                    _silo_materialize_remote_file(
                        base, device_token, key, root, root_state, change,
                        local_key=local_key, local_record=local_record)
                cursor = change_cursor
                root_state["cursor"] = cursor
                save_silo_state(state)
                applied += 1
            except (OSError, ValueError, RuntimeError, KeyError) as exc:
                failed += 1
                print(f"Silo pull failed at cursor {change_cursor}: {exc}", file=sys.stderr)
                return 1
        if not out.get("has_more"):
            break
    save_silo_state(state)
    print(f"Silo pull: {applied} applied, {skipped} already local, {failed} failed, cursor {cursor}")
    return 1 if failed else 0


def cmd_silo_sync(args, key) -> int:
    status = 0
    if not getattr(args, "pull_only", False):
        status = max(status, cmd_silo_push(args, key))
    if not getattr(args, "push_only", False):
        status = max(status, cmd_silo_pull(args, key))
    return status


def _run_silo_watch(args, key, *, sync_fn=None, sleep_fn=None) -> int:
    """Poll sync with bounded exponential retry for temporary failures."""
    try:
        interval = float(getattr(args, "interval_seconds", 30.0))
    except (TypeError, ValueError):
        print("Silo watch interval must be a finite positive number", file=sys.stderr)
        return 2
    if not math.isfinite(interval) or interval <= 0:
        print("Silo watch interval must be a finite positive number", file=sys.stderr)
        return 2
    try:
        retry_seconds = float(getattr(args, "retry_seconds", DEFAULT_WATCH_RETRY_SECONDS))
    except (TypeError, ValueError):
        print("Silo watch retry delay must be a finite positive number", file=sys.stderr)
        return 2
    if not math.isfinite(retry_seconds) or retry_seconds <= 0:
        print("Silo watch retry delay must be a finite positive number", file=sys.stderr)
        return 2
    try:
        max_failures = int(getattr(args, "max_failures", DEFAULT_WATCH_MAX_FAILURES))
    except (TypeError, ValueError):
        print("Silo watch retry limit must be a positive integer", file=sys.stderr)
        return 2
    if max_failures <= 0:
        print("Silo watch retry limit must be a positive integer", file=sys.stderr)
        return 2
    sync_fn = sync_fn or cmd_silo_sync
    sleep_fn = sleep_fn or time.sleep
    max_cycles = getattr(args, "_max_cycles", None)
    if max_cycles is not None:
        try:
            max_cycles = int(max_cycles)
        except (TypeError, ValueError):
            print("Silo watch cycle limit must be a positive integer", file=sys.stderr)
            return 2
        if max_cycles <= 0:
            print("Silo watch cycle limit must be a positive integer", file=sys.stderr)
            return 2

    cycles = 0
    failures = 0
    try:
        while True:
            cycles += 1
            print(f"Silo watch: sync cycle {cycles}")
            status = sync_fn(args, key)
            if status:
                if getattr(args, "once", False):
                    print("Silo watch stopped after a sync failure", file=sys.stderr)
                    return status
                failures += 1
                if failures >= max_failures:
                    print(f"Silo watch stopped after {failures} consecutive sync failures",
                          file=sys.stderr)
                    return status
                delay = min(retry_seconds * (2 ** (failures - 1)), MAX_WATCH_RETRY_SECONDS)
                print(f"Silo watch: sync failed; retrying in {delay:g}s "
                      f"({failures}/{max_failures})", file=sys.stderr)
                sleep_fn(delay)
                continue
            failures = 0
            if getattr(args, "once", False) or (
                    max_cycles is not None and cycles >= max_cycles):
                print(f"Silo watch: completed {cycles} cycle(s)")
                return 0
            sleep_fn(interval)
    except KeyboardInterrupt:
        print(f"Silo watch stopped cleanly after {cycles} cycle(s)")
        return 0


def cmd_silo_watch(args, key) -> int:
    """Keep one local sync root current while the command remains running."""
    return _run_silo_watch(args, key)


def cmd_silo_service(args, key) -> int:
    """Install, inspect, or remove a user-level Silo background sync service."""
    del key
    try:
        if args.action == "install":
            if not args.root:
                print("a local folder is required for Silo service install", file=sys.stderr)
                return 2
            result = silo_service.install(
                client_path=pathlib.Path(__file__).resolve(),
                server=server_url(args),
                root=args.root,
                scope=args.scope,
                interval_seconds=args.interval_seconds,
                retry_seconds=args.retry_seconds,
                max_failures=args.max_failures,
                public_library=args.public_library,
                include_prefixes=args.include_prefix,
                include_all=args.include_all,
            )
        elif args.action == "uninstall":
            result = silo_service.uninstall()
        else:
            result = silo_service.status()
    except (OSError, RuntimeError, ValueError) as exc:
        print(f"Silo background service failed: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, indent=2, sort_keys=True))
    return 0


# ── ARG PARSING + DISPATCH ────────────────────────────────────────────────────

DISPATCH = {"signup": cmd_signup, "login": cmd_login, "whoami": cmd_whoami,
            "logout": cmd_logout, "put": cmd_put, "get": cmd_get,
            "list": cmd_list, "delete": cmd_delete,
            "silo-catalog": cmd_silo_catalog, "silo-device": cmd_silo_device,
            "silo-status": cmd_silo_status, "silo-doctor": cmd_silo_doctor,
            "silo-move": cmd_silo_move,
            "silo-move-dir": cmd_silo_move_dir,
            "silo-push": cmd_silo_push,
            "silo-pull": cmd_silo_pull, "silo-sync": cmd_silo_sync,
            "silo-watch": cmd_silo_watch, "silo-service": cmd_silo_service}


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="havre-remote",
                                description="Havre client: your files, sealed on this machine, stored sovereign.")
    p.add_argument("--server", help=f"control-plane base URL (default {DEFAULT_SERVER})")
    sub = p.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("signup", help="create an account on the live API")
    sp.add_argument("email")
    sp.add_argument("--password")
    sp.add_argument("--invite", required=True)
    sp.add_argument("--residency", default="CA")
    sp.add_argument("--use", default="personal")
    sp.add_argument("--sector")
    sp.add_argument("--sensitivity", default="general")

    lp = sub.add_parser("login", help="log in to an existing account")
    lp.add_argument("email")
    lp.add_argument("--password")

    sub.add_parser("whoami", help="show the logged-in account")
    sub.add_parser("logout", help="clear the local session token")

    pp = sub.add_parser("put", help="seal a file locally and upload the ciphertext")
    pp.add_argument("file")
    pp.add_argument("--id")
    pp.add_argument("--scope", default="CA")
    pp.add_argument("--plain", action="store_true", help="store unencrypted P0 (not recommended)")

    gp = sub.add_parser("get", help="download and decrypt a file locally")
    gp.add_argument("id")
    gp.add_argument("out")

    sub.add_parser("list", help="list your object ids")

    dp = sub.add_parser("delete", help="delete an object")
    dp.add_argument("id")

    sub.add_parser("silo-catalog", help="show the public Silo package catalogue")

    sd = sub.add_parser("silo-device", help="register, list, revoke, grant, or recover a Silo device")
    sd.add_argument("action", choices=("register", "list", "revoke", "grant",
                                        "recovery-create", "recovery-import"))
    sd.add_argument("--name", default="Havre remote client",
                    help="device name for register or recovery-create")
    sd.add_argument("--id", help="device id for revoke or target device for grant; defaults to the local device for revoke")
    sd.add_argument("--output", help="encrypted bundle path for recovery-create")
    sd.add_argument("--bundle", help="encrypted bundle path for recovery-import")
    sd.add_argument("--replace", action="store_true",
                    help="replace an existing local Silo device during recovery-import")

    sub.add_parser("silo-status", help="show Silo quota and local device status")

    move = sub.add_parser(
        "silo-move",
        help="move one already-synced local file without re-uploading its ciphertext",
    )
    move.add_argument("root", help="local Silo sync directory")
    move.add_argument("old_path", help="tracked relative file path")
    move.add_argument("new_path", help="new relative file path")

    move_dir = sub.add_parser(
        "silo-move-dir",
        help="move a complete already-synced local folder without re-uploading files",
    )
    move_dir.add_argument("root", help="local Silo sync directory")
    move_dir.add_argument("old_path", help="tracked relative folder path")
    move_dir.add_argument("new_path", help="new relative folder path")

    doctor = sub.add_parser(
        "silo-doctor",
        help="verify Silo account, device, key, and optional sync-folder readiness",
    )
    doctor.add_argument("--root", help="optional local directory to verify as the sync folder")
    doctor.add_argument("--json", action="store_true", help="print machine-readable checks")

    def add_sync_args(parser):
        parser.add_argument("root", help="local directory to synchronize")
        parser.add_argument("--scope", default="CA", help="sealed-object residency scope")
        parser.add_argument(
            "--public-library",
            action="store_true",
            help="use P0 for rights-cleared books or music that may be public",
        )
        parser.add_argument(
            "--part-size-bytes",
            type=int,
            default=8 * 1024 * 1024,
            help="ciphertext chunk size for resumable uploads",
        )

    def add_selection_args(parser):
        selection = parser.add_mutually_exclusive_group()
        selection.add_argument(
            "--include-prefix",
            action="append",
            dest="include_prefix",
            metavar="FOLDER",
            help=("retain only this remote folder prefix locally; repeat for multiple "
                  "folders (the choice is saved per sync root)"),
        )
        selection.add_argument(
            "--include-all",
            action="store_true",
            help="retain the full remote library and clear a saved selection",
        )

    push = sub.add_parser("silo-push", help="seal local files and resume them into Silo")
    add_sync_args(push)
    add_selection_args(push)

    pull = sub.add_parser("silo-pull", help="pull ordered Silo changes and decrypt locally")
    pull.add_argument("root", help="local directory to synchronize")
    pull.add_argument("--limit", type=int, default=100)
    add_selection_args(pull)

    sync = sub.add_parser("silo-sync", help="push local changes, then pull Silo changes")
    add_sync_args(sync)
    add_selection_args(sync)
    sync.add_argument("--pull-only", action="store_true")
    sync.add_argument("--push-only", action="store_true")

    watch = sub.add_parser(
        "silo-watch",
        help="poll push and pull for a local Silo folder until interrupted",
    )
    add_sync_args(watch)
    add_selection_args(watch)
    watch.add_argument(
        "--interval-seconds",
        type=float,
        default=30.0,
        help="seconds between completed sync cycles (default: 30)",
    )
    watch.add_argument(
        "--retry-seconds",
        type=float,
        default=DEFAULT_WATCH_RETRY_SECONDS,
        help="initial delay after a temporary sync failure (default: 15)",
    )
    watch.add_argument(
        "--max-failures",
        type=int,
        default=DEFAULT_WATCH_MAX_FAILURES,
        help="consecutive failures before the watcher stops (default: 5)",
    )
    watch.add_argument("--once", action="store_true", help="run one sync cycle and exit")
    watch.add_argument("--pull-only", action="store_true")
    watch.add_argument("--push-only", action="store_true")
    watch.add_argument("--limit", type=int, default=100)

    service = sub.add_parser(
        "silo-service",
        help="install, inspect, or remove user-level background Silo sync",
    )
    service.add_argument("action", choices=("install", "status", "uninstall"))
    service.add_argument("root", nargs="?", help="local directory to keep synchronized for install")
    service.add_argument("--scope", default="CA", help="sealed-object residency scope")
    service.add_argument("--public-library", action="store_true",
                         help="use P0 for rights-cleared books or music that may be public")
    service.add_argument("--interval-seconds", type=float, default=30.0,
                         help="seconds between completed sync cycles (default: 30)")
    service.add_argument("--retry-seconds", type=float, default=DEFAULT_WATCH_RETRY_SECONDS,
                         help="initial delay after a temporary sync failure (default: 15)")
    service.add_argument("--max-failures", type=int, default=DEFAULT_WATCH_MAX_FAILURES,
                         help="consecutive failures before the watcher stops (default: 5)")
    service_selection = service.add_mutually_exclusive_group()
    service_selection.add_argument("--include-prefix", action="append", dest="include_prefix",
                                   metavar="FOLDER", help="retain this remote folder prefix locally")
    service_selection.add_argument("--include-all", action="store_true",
                                   help="retain the full remote library and clear a saved selection")
    return p


def main(argv=None) -> int:
    argv = argv if argv is not None else sys.argv[1:]
    if "--selftest" in argv:
        return _selftest()
    args = build_parser().parse_args(argv)
    key = None if args.cmd == "silo-service" else ensure_key()
    return DISPATCH[args.cmd](args, key)


# ── SELFTEST: client against an in-process mock control plane ──────────────────

def _package_selftest() -> int:
    """Run the dependency-light smoke checks available in the public client bundle."""
    import tempfile

    home = pathlib.Path(tempfile.mkdtemp(prefix="havre_silo_client_"))
    previous_home = os.environ.get("HAVRE_HOME")
    os.environ["HAVRE_HOME"] = str(home)
    try:
        key = ensure_key()
        session = {"server": DEFAULT_SERVER, "email": "pilot@example.invalid",
                   "token": "local-smoke-only"}
        save_session(session)
        sealed_object, ciphertext = envelope.seal(
            b"Silo client package smoke test", posture="P2", storage_class="hot",
            residency_scope="CA", obj_id="package-smoke", recipients=[key["public"]])
        passed = [
            ("local key is created with private permissions", _private_file_ok(_key_path())),
            ("session state round-trips locally", load_session() == session),
            ("client-side encryption round-trips", envelope.unseal(
                sealed_object, ciphertext, key["private"]) == b"Silo client package smoke test"),
        ]
        print("havre-remote package smoke test")
        for label, result in passed:
            print(f"  [{'PASS' if result else 'FAIL'}] {label}")
        return 0 if all(result for _, result in passed) else 1
    finally:
        if previous_home is None:
            os.environ.pop("HAVRE_HOME", None)
        else:
            os.environ["HAVRE_HOME"] = previous_home

def _selftest() -> int:
    import tempfile
    import threading
    from http.server import ThreadingHTTPServer
    try:
        import control_plane
    except ModuleNotFoundError as exc:
        # The public bundle contains the client and crypto modules, not the private
        # in-process mock control plane. Keep --selftest useful after installation.
        if exc.name == "control_plane":
            return _package_selftest()
        raise

    passed = failed = 0

    def check(name, cond, detail=""):
        nonlocal passed, failed
        if cond:
            passed += 1
            print(f"  [PASS] {name}")
        else:
            failed += 1
            print(f"  [FAIL] {name}{('  ' + detail) if detail else ''}")

    home = pathlib.Path(tempfile.mkdtemp(prefix="havre_remote_"))
    os.environ["HAVRE_HOME"] = str(home)

    # In-process mock control plane (blind, fake storage), invite-gated.
    svc = control_plane.build_services("mock", invite_codes={"INV-TEST"})
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), control_plane.make_handler(svc))
    port = httpd.server_address[1]
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    base = f"http://127.0.0.1:{port}"
    os.environ["HAVRE_SERVER"] = base

    key = ensure_key()
    check("key.json is stored as a private local file", _private_file_ok(_key_path()))

    # Recovery bundles are local artifacts: the server must never see their
    # private key or one-time device token in plaintext.
    recovery_private, recovery_public = envelope.gen_recipient_keypair()
    recovery_token = "device-token-that-must-stay-encrypted"
    recovery_payload = {
        "device_id": "dev-recovery-selftest",
        "device_name": "Recovery self-test",
        "device_token": recovery_token,
        "server": base,
        "private_key": recovery_private,
        "public_key": recovery_public,
        "created_at": "2026-11-02T12:00:00+00:00",
        "purpose": "silo-device-recovery",
    }
    recovery_password = "correct horse battery staple"
    recovery_bundle = _seal_recovery_bundle(recovery_payload, recovery_password)
    serialized_bundle = json.dumps(recovery_bundle)
    check("recovery bundle excludes plaintext private key and token",
          recovery_private not in serialized_bundle and recovery_token not in serialized_bundle)
    check("recovery bundle imports with the correct password",
          _open_recovery_bundle(recovery_bundle, recovery_password)["public_key"] == recovery_public)
    try:
        _open_recovery_bundle(recovery_bundle, "wrong recovery password")
        check("wrong recovery password is refused", False)
    except ValueError:
        check("wrong recovery password is refused", True)
    tampered_bundle = dict(recovery_bundle)
    tampered_ciphertext = bytearray(base64.b64decode(tampered_bundle["ciphertext_b64"]))
    tampered_ciphertext[-1] ^= 1
    tampered_bundle["ciphertext_b64"] = base64.b64encode(tampered_ciphertext).decode("ascii")
    try:
        _open_recovery_bundle(tampered_bundle, recovery_password)
        check("tampered recovery bundle is refused", False)
    except ValueError:
        check("tampered recovery bundle is refused", True)

    # Step 1: no-auth /api/me is 401 and does not raise.
    status, _ = _api(base, "GET", "/api/me")
    check("_api returns 401 on no-auth /api/me", status == 401)

    class A:  # tiny args holder
        server = None

    # Step 2: signup with a wrong invite is rejected cleanly.
    bad = A(); bad.email = "x@firm.ca"; bad.password = "pw"; bad.invite = "WRONG"
    bad.residency = "CA"; bad.use = "business"; bad.sector = "legal"; bad.sensitivity = "sensitive"
    rc = cmd_signup(bad, key)
    check("wrong invite is rejected (nonzero, no traceback)", rc == 1 and not load_session())

    # Step 2: real signup persists a session.
    su = A(); su.email = "client@firm.ca"; su.password = "pw"; su.invite = "INV-TEST"
    su.residency = "CA"; su.use = "business"; su.sector = "legal"; su.sensitivity = "sensitive"
    check("signup succeeds", cmd_signup(su, key) == 0)
    check("session token is stored as a private local file",
          bool(load_session().get("token")) and _private_file_ok(_session_path()))

    # Step 3: seal/upload/list/get/delete round trip, byte-identical, server blind.
    secret = b"the client's private memo; nobody at Havre can read this"
    src = home / "memo.txt"; src.write_bytes(secret)
    pa = A(); pa.server = None; pa.file = str(src); pa.id = "memo1"; pa.scope = "CA"; pa.plain = False
    check("put sealed object", cmd_put(pa, key) == 0)

    # The wire upload must carry no plaintext.
    obj, blob = envelope.seal(secret, posture="P2", storage_class="hot",
                              residency_scope="CA", obj_id="memo1", recipients=[key["public"]])
    wire = json.dumps(obj) + base64.b64encode(blob).decode("ascii")
    check("uploaded object/blob contain no plaintext", secret.decode() not in wire)

    la = A(); la.server = None
    status, out = _api(base, "GET", "/api/objects", token=load_session()["token"])
    check("list shows the object", out.get("ids") == ["memo1"])

    ga = A(); ga.server = None; ga.id = "memo1"; ga.out = str(home / "out.txt")
    check("get decrypts locally", cmd_get(ga, key) == 0 and (home / "out.txt").read_bytes() == secret)

    da = A(); da.server = None; da.id = "memo1"
    check("delete works", cmd_delete(da, key) == 0)
    status, out = _api(base, "GET", "/api/objects", token=load_session()["token"])
    check("empty after delete", out.get("ids") == [])

    # Step 4: Silo CLI registration, encrypted resumable push, ordered pull, and public
    # library deduplication. The mock billing backend is deliberately used only in-process.
    os.environ["HAVRE_SILO_PRICE_SEED_CENTS"] = "999"
    silo_auth = {
        "terms_version": "silo-terms-selftest",
        "terms_accepted_at": "2026-11-02T12:00:00Z",
    }
    code, silo_purchase = _api(
        base,
        "POST",
        "/api/silo/checkout",
        {"order_id": "silo-cli-selftest", "package_id": "seed",
         "authorization": silo_auth, "auto_renew": False},
        token=load_session()["token"],
    )
    check("Silo CLI test package is settled", code == 201
          and silo_purchase.get("order", {}).get("status") == "active")

    register = A(); register.server = None; register.action = "register"
    register.name = "CLI self-test"; register.id = None
    check("Silo CLI registers a device", cmd_silo_device(register, key) == 0
          and bool(load_silo_state().get("device_token")))

    source_root = home / "silo-source"
    source_root.mkdir()
    doctor = A(); doctor.server = None; doctor.root = str(source_root); doctor.json = False
    check("Silo doctor passes after package, device, key, and folder setup",
          cmd_silo_doctor(doctor, key) == 0)
    blocked_doctor = A(); blocked_doctor.server = None
    blocked_doctor.root = str(home / "missing-silo-root"); blocked_doctor.json = True
    check("Silo doctor blocks a missing sync root",
          cmd_silo_doctor(blocked_doctor, key) == 1)
    source_file = source_root / "Documents" / "memo.txt"
    source_file.parent.mkdir()
    source_file.write_bytes(b"encrypted Silo client payload")
    push = A(); push.server = None; push.root = str(source_root); push.scope = "CA"
    push.public_library = False; push.part_size_bytes = 3
    check("Silo CLI pushes encrypted files with resumable chunks",
          cmd_silo_push(push, key) == 0)
    changes_status, changes = _api(
        base, "GET", "/api/silo/changes?cursor=0", token=load_session()["token"])
    first_change = next((item for item in changes.get("changes", [])
                         if item.get("path") == "Documents/memo.txt"), None)
    object_status, sealed = _api(
        base, "GET", "/api/silo/objects/" + str(first_change.get("object_id")),
        token=load_session()["token"]) if first_change else (0, {})
    check("Silo CLI sends ciphertext, not plaintext",
          changes_status == 200 and object_status == 200
          and sealed.get("object", {}).get("posture") == "P2"
          and b"encrypted Silo client payload" not in base64.b64decode(
              sealed.get("blob_b64", "")))
    metadata_status, metadata_out, _ = _silo_api(
        base, "GET", "/api/silo/objects/" + str(first_change.get("object_id")) + "/metadata",
        device_token=load_silo_state()["device_token"])
    metadata_object = metadata_out.get("object", {})
    metadata_size = int(metadata_out.get("size_bytes", 0) or 0)
    metadata_hash = str(metadata_object.get("integrity", {}).get("ciphertext_sha256", ""))
    binary_part = _silo_download_part(
        home / "binary-download-test", str(first_change.get("object_id")),
        str(first_change.get("version_id")))
    binary_part.write_bytes(base64.b64decode(sealed.get("blob_b64", ""))[:7])
    binary_status, binary_out, _ = _silo_download_to_file(
        base, "/api/silo/objects/" + str(first_change.get("object_id")) + "/download",
        load_silo_state()["device_token"], binary_part, metadata_size, metadata_hash,
        chunk_size=5)
    check("Silo binary metadata omits ciphertext and ranged download resumes",
          metadata_status == 200 and "blob_b64" not in metadata_out
          and metadata_object.get("id") == first_change.get("object_id")
          and binary_status == 206 and binary_out.get("sha256") == metadata_hash
          and binary_part.stat().st_size == metadata_size)
    binary_part.unlink()

    recovery_status, recovery_out, _ = _silo_api(
        base,
        "POST",
        "/api/silo/devices",
        {"name": "CLI recovery self-test", "public_key": recovery_public},
        token=load_session()["token"],
    )
    recovery_device = recovery_out.get("device", {})
    check("recovery device receives a one-time device credential",
          recovery_status == 201 and bool(recovery_device.get("id"))
          and bool(recovery_device.get("device_token")))
    recovery_grant_status = 0
    if recovery_status == 201 and first_change and object_status == 200:
        recovery_updated_object = envelope.add_recipient(
            sealed.get("object", {}), key["private"], recovery_public)
        recovery_grant_status, _, _ = _silo_api(
            base,
            "POST",
            f"/api/silo/objects/{first_change.get('object_id')}/grant",
            {"target_device_id": recovery_device.get("id"),
             "object": recovery_updated_object},
            device_token=load_silo_state()["device_token"],
        )
    recovery_object_status, recovery_object, _ = _silo_api(
        base,
        "GET",
        "/api/silo/objects/" + str(first_change.get("object_id")) if first_change else "/api/silo/objects/missing",
        token=load_session()["token"],
    )
    recovery_blob = base64.b64decode(recovery_object.get("blob_b64", "")) \
        if recovery_object_status == 200 else b""
    recovery_decrypts = False
    if recovery_object_status == 200:
        try:
            recovery_decrypts = (
                envelope.unseal(recovery_object["object"], recovery_blob, recovery_private)
                == b"encrypted Silo client payload")
        except (KeyError, TypeError, ValueError):
            recovery_decrypts = False
    check("recovery device can decrypt an existing granted object",
          recovery_grant_status in (200, 201) and recovery_object_status == 200
          and recovery_decrypts)

    mirror_root = home / "silo-mirror"
    pull = A(); pull.server = None; pull.root = str(mirror_root); pull.limit = 100
    check("Silo CLI pulls and decrypts ordered changes",
          cmd_silo_pull(pull, key) == 0
          and (mirror_root / "Documents" / "memo.txt").read_bytes()
          == b"encrypted Silo client payload")

    source_file.write_bytes(b"encrypted Silo client update")
    check("Silo CLI records a new local version", cmd_silo_push(push, key) == 0)
    check("Silo CLI applies an update on the mirror",
          cmd_silo_pull(pull, key) == 0
          and (mirror_root / "Documents" / "memo.txt").read_bytes()
          == b"encrypted Silo client update")

    watch = A(); watch.server = None; watch.root = str(source_root)
    watch.scope = "CA"; watch.public_library = False; watch.part_size_bytes = 3
    watch.pull_only = False; watch.push_only = False; watch.limit = 100
    watch.once = True; watch.interval_seconds = 30
    check("Silo watch one-shot reuses the encrypted sync cycle",
          cmd_silo_watch(watch, key) == 0)
    source_record = load_silo_state()["roots"][str(source_root.resolve())]["files"] \
        ["Documents/memo.txt"]
    check("Silo watch skips historical versions already represented locally",
          source_file.read_bytes() == b"encrypted Silo client update"
          and source_record.get("change_cursor") == 2)
    check("newer remote changes are recognized before a local delete",
          _silo_change_touches_record(
              {"cursor": 3, "path": "Documents/memo.txt", "object_id": "new-object"},
              source_record))
    source_file.unlink()
    check("Silo CLI propagates a tracked local deletion",
          cmd_silo_push(push, key) == 0 and not source_file.exists())
    deletion_changes = _api(
        base, "GET", "/api/silo/changes?cursor=2", token=load_session()["token"])[1]
    deletion_change = next((item for item in deletion_changes.get("changes", [])
                            if item.get("path") == "Documents/memo.txt"
                            and item.get("action") == "delete"), None)
    check("local deletion emits a recoverable remote tombstone",
          bool(deletion_change) and deletion_change.get("cursor") == 3)
    check("Silo mirror applies the recoverable deletion",
          cmd_silo_pull(pull, key) == 0
          and not (mirror_root / "Documents" / "memo.txt").exists())

    move_source = source_root / "Documents" / "move.txt"
    move_source.write_bytes(b"unchanged move payload")
    check("Silo CLI prepares a tracked file for explicit move",
          cmd_silo_push(push, key) == 0)
    move_args = A(); move_args.server = None; move_args.root = str(source_root)
    move_args.old_path = "Documents/move.txt"
    move_args.new_path = "Documents/renamed-move.txt"
    check("Silo CLI moves an unchanged file without re-uploading it",
          cmd_silo_move(move_args, key) == 0
          and not move_source.exists()
          and (source_root / "Documents" / "renamed-move.txt").read_bytes()
          == b"unchanged move payload")
    check("Silo mirror applies a remote move without downloading a new version",
          cmd_silo_pull(pull, key) == 0
          and not (mirror_root / "Documents" / "move.txt").exists()
          and (mirror_root / "Documents" / "renamed-move.txt").read_bytes()
          == b"unchanged move payload")
    (mirror_root / "Documents" / "renamed-move.txt").write_bytes(b"local move edit")
    move_again = A(); move_again.server = None; move_again.root = str(source_root)
    move_again.old_path = "Documents/renamed-move.txt"
    move_again.new_path = "Documents/final-move.txt"
    check("Silo CLI moves the same file a second time",
          cmd_silo_move(move_again, key) == 0)
    check("Silo mirror preserves an edited local path during a remote move",
          cmd_silo_pull(pull, key) == 0
          and (mirror_root / "Documents" / "final-move.txt").read_bytes()
          == b"unchanged move payload"
          and any(
              item.read_bytes() == b"local move edit"
              for item in (mirror_root / "Documents").glob(
                  "renamed-move.txt.local-before-move-*")))

    folder_source = source_root / "FolderMove"
    folder_source.mkdir()
    (folder_source / "one.txt").write_bytes(b"folder move one")
    (folder_source / "two.txt").write_bytes(b"folder move two")
    check("Silo CLI stages a folder with multiple tracked files",
          cmd_silo_push(push, key) == 0)
    folder_move = A(); folder_move.server = None; folder_move.root = str(source_root)
    folder_move.old_path = "FolderMove"
    folder_move.new_path = "Archive/FolderMove"
    check("Silo CLI moves a complete folder without re-uploading files",
          cmd_silo_move_dir(folder_move, key) == 0
          and not folder_source.exists()
          and (source_root / "Archive" / "FolderMove" / "one.txt").read_bytes()
          == b"folder move one"
          and (source_root / "Archive" / "FolderMove" / "two.txt").read_bytes()
          == b"folder move two")
    check("Silo folder move is safe to retry after the local rename",
          cmd_silo_move_dir(folder_move, key) == 0)
    check("Silo mirror applies a folder move as compact file events",
          cmd_silo_pull(pull, key) == 0
          and not (mirror_root / "FolderMove").exists()
          and (mirror_root / "Archive" / "FolderMove" / "one.txt").read_bytes()
          == b"folder move one"
          and (mirror_root / "Archive" / "FolderMove" / "two.txt").read_bytes()
          == b"folder move two")

    # A selective root must advance the tenant-wide cursor without materializing
    # excluded paths, then bootstrap a newly selected prefix from current metadata.
    selected_file = source_root / "Documents" / "selected.txt"
    selected_file.write_bytes(b"selected device folder")
    photo_file = source_root / "Photos" / "photo.txt"
    photo_file.parent.mkdir()
    photo_file.write_bytes(b"space-saving photo")
    check("Silo CLI uploads files for two selectable prefixes",
          cmd_silo_push(push, key) == 0)
    selective_root = home / "silo-selective"
    selective_pull = A(); selective_pull.server = None
    selective_pull.root = str(selective_root); selective_pull.limit = 100
    selective_pull.include_prefix = ["Documents"]
    check("Silo selective pull retains only the requested prefix",
          cmd_silo_pull(selective_pull, key) == 0
          and (selective_root / "Documents" / "selected.txt").read_bytes()
          == b"selected device folder"
          and not (selective_root / "Photos" / "photo.txt").exists())
    selective_state = load_silo_state()["roots"][str(selective_root.resolve())]
    check("Silo selective pull advances the global cursor for excluded events",
          selective_state.get("cursor", 0) >= 5
          and selective_state.get("include_prefixes") == ["Documents"])
    selective_pull.include_prefix = ["Documents", "Photos"]
    check("Silo selection expansion bootstraps current remote metadata",
          cmd_silo_pull(selective_pull, key) == 0
          and (selective_root / "Photos" / "photo.txt").read_bytes()
          == b"space-saving photo")
    selective_pull.include_prefix = ["Documents"]
    remote_photo_status, remote_photos = _api(
        base, "GET", "/api/silo/files?prefix=Photos", token=load_session()["token"])
    check("Silo narrowing frees only the local copy, not the remote file",
          cmd_silo_pull(selective_pull, key) == 0
          and not (selective_root / "Photos" / "photo.txt").exists()
          and remote_photo_status == 200
          and any(item.get("path") == "Photos/photo.txt"
                  for item in remote_photos.get("files", [])))

    watch_args = A(); watch_args.interval_seconds = 0.25
    watch_args.once = False; watch_args._max_cycles = 2
    watch_calls = []; sleep_calls = []

    def fake_sync(_args, _key):
        watch_calls.append(True)
        return 0

    def fake_sleep(seconds):
        sleep_calls.append(seconds)

    check("Silo watch repeats bounded polling cycles",
          _run_silo_watch(watch_args, key, sync_fn=fake_sync, sleep_fn=fake_sleep) == 0
          and len(watch_calls) == 2 and sleep_calls == [0.25])

    retry_args = A(); retry_args.interval_seconds = 0.25
    retry_args.retry_seconds = 0.5; retry_args.max_failures = 3
    retry_args.once = False; retry_args._max_cycles = 3
    retry_calls = []; retry_sleeps = []
    retry_results = iter((1, 0, 0))

    def retrying_sync(_args, _key):
        retry_calls.append(True)
        return next(retry_results)

    check("Silo watch retries a temporary failure and resumes after success",
          _run_silo_watch(retry_args, key, sync_fn=retrying_sync,
                          sleep_fn=retry_sleeps.append) == 0
          and len(retry_calls) == 3 and retry_sleeps == [0.5, 0.25])

    interrupt_args = A(); interrupt_args.interval_seconds = 0.25
    interrupt_args.once = False

    def interrupt_sleep(_seconds):
        raise KeyboardInterrupt

    check("Silo watch stops cleanly on interrupt",
          _run_silo_watch(interrupt_args, key, sync_fn=fake_sync,
                          sleep_fn=interrupt_sleep) == 0
          and len(watch_calls) == 3)

    failure_args = A(); failure_args.interval_seconds = 0.25
    failure_args.once = False; failure_args.max_failures = 1
    failure_calls = []

    def failing_sync(_args, _key):
        failure_calls.append(True)
        return 1

    check("Silo watch stops after a failed sync cycle",
          _run_silo_watch(failure_args, key, sync_fn=failing_sync,
                          sleep_fn=fake_sleep) == 1
          and len(failure_calls) == 1)

    invalid_args = A(); invalid_args.interval_seconds = 0
    invalid_args.once = False
    check("Silo watch rejects a non-positive interval",
          _run_silo_watch(invalid_args, key, sync_fn=fake_sync,
                          sleep_fn=fake_sleep) == 2)
    invalid_retry_args = A(); invalid_retry_args.interval_seconds = 0.25
    invalid_retry_args.retry_seconds = 0; invalid_retry_args.once = False
    check("Silo watch rejects a non-positive retry delay",
          _run_silo_watch(invalid_retry_args, key, sync_fn=fake_sync,
                          sleep_fn=fake_sleep) == 2)

    library_root = home / "silo-library"
    library_root.mkdir()
    (library_root / "one.epub").write_bytes(b"same public book edition")
    (library_root / "copy.epub").write_bytes(b"same public book edition")
    library_push = A(); library_push.server = None; library_push.root = str(library_root)
    library_push.scope = "CA"; library_push.public_library = True; library_push.part_size_bytes = 5
    usage_before = _api(
        base, "GET", "/api/silo/usage", token=load_session()["token"])[1]
    check("Silo CLI deduplicates identical public-library books",
          cmd_silo_push(library_push, key) == 0
          and _api(base, "GET", "/api/silo/usage", token=load_session()["token"])[1]
          .get("usage", {}).get("used_bytes", 0) > usage_before.get("usage", {}).get("used_bytes", 0))
    library_changes = _api(
        base, "GET", "/api/silo/changes?cursor=0", token=load_session()["token"])[1]
    library_ids = {
        item.get("object_id") for item in library_changes.get("changes", [])
        if item.get("path") in {"one.epub", "copy.epub"}
    }
    check("identical library copies share one physical object",
          len(library_ids) == 1)

    # Exercise the actual CLI command paths without interactive input. The test
    # uses a temporary replacement home for import and never writes a real user's
    # recovery bundle.
    command_bundle = home / "silo-command-recovery.json"
    previous_getpass = getpass.getpass
    previous_home = os.environ["HAVRE_HOME"]
    getpass.getpass = lambda _prompt: recovery_password
    try:
        recovery_create = A(); recovery_create.server = None
        recovery_create.name = "CLI command recovery"; recovery_create.output = str(command_bundle)
        check("recovery-create writes an encrypted bundle",
              _cmd_silo_recovery_create(
                  recovery_create, key, base, load_session()["token"], load_silo_state()) == 0
              and command_bundle.exists())
        replacement_home = home / "recovery-import-home"
        os.environ["HAVRE_HOME"] = str(replacement_home)
        recovery_import = A(); recovery_import.bundle = str(command_bundle)
        recovery_import.replace = True
        check("recovery-import installs the recovery device offline",
              _cmd_silo_recovery_import(recovery_import) == 0
              and bool(load_silo_state().get("silo_private"))
              and load_silo_state().get("server") == base)
    finally:
        getpass.getpass = previous_getpass
        os.environ["HAVRE_HOME"] = previous_home

    httpd.shutdown()
    print(f"\n{passed} passed, {failed} failed")
    return 1 if failed else 0


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