#!/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
import zipfile
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_book_scanner  # 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_collection_records(base: str, device_token: str) -> list[dict]:
    """Read safe collection metadata through the active device credential."""
    status, out, _ = _silo_api(
        base, "GET", "/api/silo/collections", device_token=device_token)
    if status != 200:
        raise RuntimeError(f"Silo collection list failed: {_silo_map_error(status, out)}")
    records = out.get("collections")
    if not isinstance(records, list) or any(not isinstance(item, dict) for item in records):
        raise RuntimeError("Silo collection list returned invalid metadata")
    return records


def _silo_collection_prefix(args, base: str, device_token: str) -> str:
    """Resolve one collection id or display name to its server-owned stable root."""
    selected = str(getattr(args, "collection", "") or "").strip()
    if not selected:
        raise ValueError("Silo collection selection cannot be empty")
    cached = getattr(args, "_silo_collection_resolved", None)
    if (isinstance(cached, tuple) and len(cached) == 2
            and cached[0] == selected and cached[1]):
        return cached[1]
    records = _silo_collection_records(base, device_token)
    by_id = [item for item in records if str(item.get("id") or "") == selected]
    if len(by_id) == 1:
        match = by_id[0]
    else:
        wanted = selected.casefold()
        by_name = [item for item in records
                    if str(item.get("name") or "").casefold() == wanted]
        if len(by_name) != 1:
            if not by_name:
                raise ValueError(f"Silo collection was not found: '{selected}'")
            raise ValueError(f"Silo collection name is ambiguous: '{selected}'")
        match = by_name[0]
    prefix = _silo_normalize_prefix(str(match.get("root_prefix") or ""))
    if not prefix.startswith("Collections/"):
        raise ValueError("Silo collection metadata returned an unsafe root")
    setattr(args, "_silo_collection_resolved", (selected, prefix))
    return prefix


def _silo_effective_prefixes(args, root_state: dict, *, base: str | None = None,
                             device_token: str | None = None) -> tuple[str, ...]:
    """Resolve collection, explicit, saved, or full-library selection in that order."""
    prefixes, _ = _silo_selection_context(
        args, root_state, base=base, device_token=device_token)
    return prefixes


def _silo_selection_context(args, root_state: dict, *, base: str | None = None,
                            device_token: str | None = None) -> tuple[tuple[str, ...], str]:
    """Resolve the remote selection and its optional local collection mount."""
    if getattr(args, "collection", None):
        if not base or not device_token:
            raise ValueError("Silo collection selection requires an active device")
        prefix = _silo_collection_prefix(args, base, device_token)
        return (prefix,), prefix
    if getattr(args, "include_all", False):
        return (), ""
    explicit = getattr(args, "include_prefix", None)
    if explicit is not None:
        return _silo_normalize_prefixes(explicit), ""
    prefixes = _silo_normalize_prefixes(root_state.get("include_prefixes", []))
    saved_local_prefix = str(root_state.get("collection_root_prefix") or "").strip()
    if saved_local_prefix:
        saved_local_prefix = _silo_normalize_prefix(saved_local_prefix)
        if prefixes == (saved_local_prefix,):
            return prefixes, saved_local_prefix
    return prefixes, ""


def _silo_set_selection(root_state: dict, prefixes: tuple[str, ...],
                        local_prefix: str) -> None:
    """Persist a selection and whether its remote root is mounted locally."""
    root_state["include_prefixes"] = list(prefixes)
    root_state["collection_root_prefix"] = str(local_prefix or "")
    root_state["selection_initialized"] = True


def _silo_normalize_relative_path(value: str, label: 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(f"{label} must be non-empty and relative")
    return normalized


def _silo_local_path_for_remote(remote_path: str, local_prefix: str = "") -> str:
    """Map a full remote path into the selected local sync root."""
    normalized = _silo_normalize_relative_path(remote_path, "Silo remote paths")
    prefix = str(local_prefix or "").strip()
    if prefix:
        prefix = _silo_normalize_prefix(prefix)
        if normalized == prefix:
            raise ValueError("Silo collection root cannot be materialized as a file")
        if not normalized.startswith(prefix + "/"):
            raise ValueError("Silo remote path is outside the selected collection")
        normalized = normalized[len(prefix) + 1:]
    return _silo_normalize_relative_path(normalized, "Silo local paths")


def _silo_remote_path_for_local(local_path: str, local_prefix: str = "") -> str:
    """Map a local sync-root path into the selected full remote path."""
    normalized = _silo_normalize_relative_path(local_path, "Silo local paths")
    prefix = str(local_prefix or "").strip()
    if not prefix:
        return normalized
    return _silo_normalize_prefix(prefix) + "/" + normalized


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 _new_silo_sync_metrics() -> dict:
    return {
        "push_status": "not_run",
        "pull_status": "not_run",
        "uploaded_files": 0,
        "downloaded_files": 0,
        "deleted_files": 0,
        "conflict_files": 0,
        "skipped_files": 0,
        "bytes_uploaded": 0,
        "bytes_downloaded": 0,
        "cursor": 0,
        "error_code": "",
    }


def _silo_sync_metrics(args) -> dict:
    metrics = getattr(args, "_silo_sync_metrics", None)
    if not isinstance(metrics, dict):
        metrics = _new_silo_sync_metrics()
        setattr(args, "_silo_sync_metrics", metrics)
    return metrics


def _silo_sync_receipt_status(metrics: dict) -> str:
    phases = [metrics.get("push_status"), metrics.get("pull_status")]
    ran = [phase for phase in phases if phase != "not_run"]
    failures = sum(phase == "failed" for phase in ran)
    successes = sum(phase == "completed" for phase in ran)
    if not failures:
        return "completed"
    return "partial" if successes else "failed"


def _silo_submit_sync_receipt(base: str, device_token: str, args,
                              started_at: str, metrics: dict) -> None:
    """Submit safe sync counters without making a completed data transfer disappear."""
    finished_at = datetime.datetime.now(datetime.timezone.utc).isoformat().replace(
        "+00:00", "Z")
    try:
        root = pathlib.Path(args.root).expanduser().resolve()
        local_state = load_silo_state()
        root_state = local_state.get("roots", {}).get(str(root), {})
        metrics["cursor"] = max(0, int(root_state.get("cursor") or 0))
    except (OSError, TypeError, ValueError):
        metrics["error_code"] = metrics.get("error_code") or "local_state"
    payload = {
        "receipt_id": "sr_" + uuid.uuid4().hex,
        "status": _silo_sync_receipt_status(metrics),
        "push_status": metrics.get("push_status", "not_run"),
        "pull_status": metrics.get("pull_status", "not_run"),
        "started_at": started_at,
        "finished_at": finished_at,
        "cursor": metrics.get("cursor", 0),
        "uploaded_files": metrics.get("uploaded_files", 0),
        "downloaded_files": metrics.get("downloaded_files", 0),
        "deleted_files": metrics.get("deleted_files", 0),
        "conflict_files": metrics.get("conflict_files", 0),
        "skipped_files": metrics.get("skipped_files", 0),
        "bytes_uploaded": metrics.get("bytes_uploaded", 0),
        "bytes_downloaded": metrics.get("bytes_downloaded", 0),
        "error_code": metrics.get("error_code", ""),
    }
    status, out, _ = _silo_api(
        base, "POST", "/api/silo/sync-receipts", payload,
        device_token=device_token)
    if status in (200, 201):
        receipt = out.get("receipt") or {}
        print(f"Silo sync receipt: {receipt.get('receipt_id', payload['receipt_id'])} "
              f"({payload['status']})")
    else:
        # Data sync remains the primary operation. Old servers and transient control-plane
        # failures are visible, but do not turn a successful encrypted transfer into a lie.
        print(f"Silo sync receipt was not saved: {_silo_map_error(status, out)}",
              file=sys.stderr)


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_ensure_remote_folder(root: pathlib.Path, target: pathlib.Path,
                               suffix: str) -> None:
    """Create a remote directory path while preserving local file collisions."""
    resolved_root = root.resolve()
    resolved_root.mkdir(parents=True, exist_ok=True)
    current = resolved_root
    for part in target.relative_to(resolved_root).parts:
        current = current / part
        if current.is_symlink() or (current.exists() and not current.is_dir()):
            preserved = _silo_preserve_local_target(current, suffix)
            print(f"preserved local collision as '{preserved.relative_to(resolved_root)}'")
        current.mkdir(exist_ok=True)


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,
                                  local_prefix: str = "") -> 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")
    local_path = _silo_local_path_for_remote(remote_path, local_prefix)
    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")
    target = _silo_safe_target(root, local_path)
    legacy_target = None
    if local_key and local_key != local_path:
        legacy_target = _silo_safe_target(root, local_key)
        if (local_record
                and local_record.get("version_id") == version_id
                and local_record.get("object_id") == object_id
                and not target.exists()
                and legacy_target.is_file()
                and local_record.get("local_sha256")
                and _silo_file_hash(legacy_target) == local_record.get("local_sha256")):
            target.parent.mkdir(parents=True, exist_ok=True)
            os.replace(legacy_target, target)
            _silo_remove_empty_parents(legacy_target.parent, root)
            updated = dict(local_record)
            updated.update({
                "path": local_path,
                "remote_path": remote_path,
                "change_cursor": change_cursor,
                "change_device_id": str(change.get("device_id") or ""),
            })
            root_state["files"].pop(local_key, None)
            root_state["files"][local_path] = updated
            print(f"remapped '{remote_path}' to local root")
            return
    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.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 legacy_target and legacy_target.exists() and legacy_target != target:
        legacy_unchanged = bool(
            local_record and local_record.get("local_sha256")
            and legacy_target.is_file()
            and _silo_file_hash(legacy_target) == local_record.get("local_sha256"))
        if legacy_unchanged:
            legacy_target.unlink()
            _silo_remove_empty_parents(legacy_target.parent, root)
        elif legacy_target.is_file():
            preserved = _silo_preserve_local_target(
                legacy_target, f".local-before-remap-{change_cursor}")
            print(f"preserved local changes as '{preserved.relative_to(root)}'")
    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,
                            local_prefix: str = "") -> 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")
    local_path = _silo_local_path_for_remote(remote_path, local_prefix)
    if local_key and local_record:
        old_target = _silo_safe_target(root, local_key)
        new_target = _silo_safe_target(root, local_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": local_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"][local_path] = updated
            print(f"moved '{local_key}' to '{local_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, local_prefix=local_prefix)


def _silo_apply_remote_folder_change(root: pathlib.Path, change: dict,
                                     local_prefix: str = "",
                                     source_in_scope: bool = True) -> None:
    """Materialize a metadata-only folder event without touching ciphertext."""
    action = str(change.get("action") or "")
    remote_path = str(change.get("path") or "")
    if action not in {"folder_create", "folder_move", "folder_delete", "folder_restore"} \
            or not remote_path:
        raise ValueError("Silo folder change is missing its action or path")
    local_path = _silo_local_path_for_remote(remote_path, local_prefix)
    target = _silo_safe_target(root, local_path)
    cursor = int(change.get("cursor") or 0)
    if action == "folder_delete":
        if not target.exists() and not target.is_symlink():
            return
        if target.is_dir() and not target.is_symlink():
            try:
                next(target.iterdir())
            except StopIteration:
                target.rmdir()
                print(f"removed remote folder '{remote_path}'")
                return
            preserved = _silo_unique_variant(
                target, f".local-before-folder-delete-{cursor}")
            os.replace(target, preserved)
            print(f"preserved local folder as '{preserved.relative_to(root)}'")
            return
        preserved = _silo_preserve_local_target(
            target, f".local-before-folder-delete-{cursor}")
        print(f"preserved local file as '{preserved.relative_to(root)}'")
        return
    if action in {"folder_create", "folder_restore"}:
        _silo_ensure_remote_folder(
            root, target, f".local-before-folder-{cursor}")
        verb = "restored" if action == "folder_restore" else "created"
        print(f"{verb} remote folder '{remote_path}'")
        return

    previous_path = str(change.get("previous_path") or "")
    if not previous_path:
        raise ValueError("Silo folder move is missing its previous path")
    if not source_in_scope:
        target.mkdir(parents=True, exist_ok=True)
        print(f"created remote folder '{remote_path}'")
        return
    old_target = _silo_safe_target(
        root, _silo_local_path_for_remote(previous_path, local_prefix))
    if old_target == target:
        target.mkdir(parents=True, exist_ok=True)
        return
    if old_target.exists():
        if old_target.is_dir():
            if target.exists():
                if not target.is_dir():
                    preserved = _silo_preserve_local_target(
                        target, f".local-before-folder-move-{cursor}")
                    print(f"preserved local file as '{preserved.relative_to(root)}'")
                    target.mkdir(parents=True, exist_ok=True)
                else:
                    try:
                        next(old_target.iterdir())
                    except StopIteration:
                        old_target.rmdir()
                    else:
                        preserved = _silo_unique_variant(
                            old_target, f".local-before-folder-move-{cursor}")
                        os.replace(old_target, preserved)
                        print(f"preserved local folder as '{preserved.relative_to(root)}'")
                        target.mkdir(parents=True, exist_ok=True)
            else:
                target.parent.mkdir(parents=True, exist_ok=True)
                os.replace(old_target, target)
        else:
            preserved = _silo_preserve_local_target(
                old_target, f".local-before-folder-move-{cursor}")
            print(f"preserved local file as '{preserved.relative_to(root)}'")
            target.mkdir(parents=True, exist_ok=True)
    else:
        target.mkdir(parents=True, exist_ok=True)
    print(f"moved remote folder '{previous_path}' to '{remote_path}'")


def _silo_remove_remote_folder_change(root: pathlib.Path, change: dict,
                                      local_prefix: str = "") -> None:
    """Remove a moved folder that left the selected local retention scope."""
    previous_path = str(change.get("previous_path") or "")
    if not previous_path:
        raise ValueError("Silo folder move is missing its previous path")
    old_target = _silo_safe_target(
        root, _silo_local_path_for_remote(previous_path, local_prefix))
    if not old_target.exists():
        return
    cursor = int(change.get("cursor") or 0)
    if old_target.is_dir():
        try:
            next(old_target.iterdir())
        except StopIteration:
            old_target.rmdir()
            _silo_remove_empty_parents(old_target.parent, root)
            print(f"removed unselected remote folder '{previous_path}'")
            return
        preserved = _silo_unique_variant(
            old_target, f".local-before-unselected-folder-{cursor}")
        os.replace(old_target, preserved)
        print(f"preserved local folder as '{preserved.relative_to(root)}'")
        return
    preserved = _silo_preserve_local_target(
        old_target, f".local-before-unselected-folder-{cursor}")
    print(f"preserved local file as '{preserved.relative_to(root)}'")


def _silo_reconcile_selection(base: str, device_token: str, state: dict,
                              root: pathlib.Path, root_state: dict, key: dict,
                              prefixes: tuple[str, ...],
                              local_prefix: 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
    try:
        previous_local_prefix = _silo_normalize_prefix(
            str(root_state.get("collection_root_prefix") or "")) \
            if root_state.get("collection_root_prefix") else ""
    except ValueError as exc:
        print(f"Silo saved collection mount is invalid: {exc}", file=sys.stderr)
        return False
    if previous == prefixes and previous_local_prefix == local_prefix:
        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, _silo_local_path_for_remote(remote_path, local_prefix))
            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,
                local_prefix=local_prefix)
        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)
    _silo_set_selection(root_state, prefixes, local_prefix)
    save_silo_state(state)
    selected = ", ".join(prefixes) if prefixes else "the full remote library"
    suffix = " (mounted at local sync root)" if local_prefix else ""
    print(f"Silo local retention selection: {selected}{suffix}")
    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,
                       book_intake_id: str | None = None,
                       remote_path: str | None = None) -> tuple[dict, pathlib.Path, pathlib.Path]:
    public_library = bool(getattr(args, "public_library", False))
    book_intake_id = str(book_intake_id or "").strip() or None
    remote_path = (_silo_normalize_relative_path(remote_path, "Silo remote paths")
                   if remote_path else
                   _silo_normalize_relative_path(relative, "Silo local paths"))
    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"{remote_path}\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") == remote_path
                and meta.get("mode") == mode
                and meta.get("book_intake_id") == book_intake_id
                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": remote_path,
        "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,
        "book_intake_id": book_intake_id,
        "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"],
        "book_intake_id": meta.get("book_intake_id"),
        "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,
                             local_path: str | 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 {}
    remote_path = version.get("path") or meta["path"]
    local_path = local_path or remote_path
    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": local_path,
        "remote_path": remote_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": remote_path},
            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 cmd_silo_collections(args, key) -> int:
    """List safe tenant collections using only the active device credential."""
    del key
    try:
        device_token, _ = _silo_device_required()
        records = _silo_collection_records(server_url(args), device_token)
    except SystemExit:
        return 2
    except (RuntimeError, ValueError) as exc:
        print(str(exc), file=sys.stderr)
        return 1
    if getattr(args, "json", False):
        print(json.dumps({"collections": records}, indent=2))
        return 0
    print("Silo collections:")
    if not records:
        print("  (none)")
        return 0
    for record in records:
        name = str(record.get("name") or "Unnamed collection")
        collection_id = str(record.get("id") or "unknown")
        kind = str(record.get("kind") or "unknown")
        mode = str(record.get("mode") or "unknown")
        file_count = int(record.get("file_count") or 0)
        used_bytes = int(record.get("used_bytes") or 0)
        print(f"  {name} ({collection_id}) [{kind}/{mode}] "
              f"{file_count} file(s), {used_bytes} byte(s)")
    return 0


def _silo_grant_existing_objects(base: str, device_token: str, source_key: dict,
                                 target_id: str, target_public: str,
                                 collection_prefix: str = "") -> dict:
    """Add a target device wrapper to existing private objects in an optional scope."""
    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
    if collection_prefix:
        try:
            collection_prefix = _silo_normalize_prefix(collection_prefix)
        except ValueError as exc:
            print(f"Silo collection grant scope is invalid: {exc}", file=sys.stderr)
            summary["failed"] = 1
            return summary
        status, files_out, _ = _silo_api(
            base, "GET", "/api/silo/files?" + urlencode({"prefix": collection_prefix}),
            device_token=device_token)
        if status != 200:
            print(f"Silo collection file list failed: {_silo_map_error(status, files_out)}",
                  file=sys.stderr)
            summary["failed"] = 1
            return summary
        object_ids = sorted({
            str(item.get("object_id") or "").strip()
            for item in files_out.get("files", [])
            if isinstance(item, dict) and item.get("object_id")
        })
    else:
        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
        object_ids = sorted({
            str(object_id).strip() for object_id in objects_out.get("ids", [])
            if str(object_id).strip()
        })
    for object_id in object_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_enroll(args) -> int:
    code = str(getattr(args, "code", "") or "").strip()
    if not code:
        print("a browser enrollment code is required", file=sys.stderr)
        return 2
    base = server_url(args)
    state = load_silo_state()
    if any(state.get(field) for field in
           ("device_id", "device_token", "silo_private", "silo_public")) \
            and not bool(getattr(args, "replace", False)):
        print("a local Silo device already exists; use --replace only after preserving it",
              file=sys.stderr)
        return 2
    try:
        device_key = envelope.gen_recipient_keypair()
        status, out, _ = _silo_api(
            base, "POST", "/api/silo/device-enrollment/redeem",
            {"code": code, "name": args.name, "public_key": device_key[1]},
        )
    except (OSError, ValueError, RuntimeError) as exc:
        print(f"Silo device enrollment failed: {exc}", file=sys.stderr)
        return 1
    if status != 201:
        print(f"Silo device enrollment failed: {_silo_map_error(status, out)}",
              file=sys.stderr)
        return 1
    device = out.get("device", {})
    if not device.get("id") or not device.get("device_token"):
        print("Silo device enrollment returned incomplete credentials", file=sys.stderr)
        return 1
    state.update({
        "device_id": device.get("id", ""),
        "device_name": device.get("name", args.name),
        "device_token": device.get("device_token", ""),
        "silo_private": device_key[0],
        "silo_public": device_key[1],
        "server": base,
    })
    save_silo_state(state)
    print(f"enrolled Silo device '{device.get('name', args.name)}' ({device.get('id')})")
    print(f"key fingerprint: {envelope.fingerprint(device_key[1])}")
    return 0


def cmd_silo_device(args, key) -> int:
    if args.action == "recovery-import":
        return _cmd_silo_recovery_import(args)
    if args.action == "setup":
        root = str(getattr(args, "root", "") or "").strip()
        if not root:
            print("--root is required for silo-device setup", file=sys.stderr)
            return 2
    if args.action in {"enroll", "setup"}:
        enrollment_status = _cmd_silo_enroll(args)
        if enrollment_status != 0 or args.action == "enroll":
            return enrollment_status
        sync_args = argparse.Namespace(
            server=getattr(args, "server", None),
            root=root,
            scope=getattr(args, "scope", "CA"),
            public_library=bool(getattr(args, "public_library", False)),
            part_size_bytes=int(getattr(args, "part_size_bytes", 8 * 1024 * 1024)),
            collection=getattr(args, "collection", None),
            include_prefix=None,
            include_all=False,
            pull_only=False,
            push_only=False,
            limit=100,
        )
        print("starting the first encrypted Silo sync")
        return cmd_silo_sync(sync_args, key)
    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
        collection_prefix = ""
        if getattr(args, "collection", None):
            try:
                collection_prefix = _silo_collection_prefix(args, base, device_token)
            except (RuntimeError, ValueError) 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 ""),
            collection_prefix=collection_prefix)
        scope_note = f" in {collection_prefix}" if collection_prefix else ""
        print(f"Silo device grant: {summary['granted']} granted, "
              f"{summary['skipped']} skipped, {summary['failed']} failed{scope_note}")
        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:
    return _silo_normalize_relative_path(value, "Silo local paths")


def _silo_local_prefix_move_plan(root: pathlib.Path, root_state: dict,
                                 old_prefix: str, new_prefix: str,
                                 remote_files: list[dict],
                                 remote_old_prefix: str | None = None) -> list[tuple[str, dict]]:
    """Validate local files before a remote folder move is committed."""
    remote_old_prefix = remote_old_prefix or old_prefix
    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(remote_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],
                                  remote_old_prefix: str | None = None,
                                  remote_new_prefix: str | None = None) -> None:
    """Rename the local folder and update local cursor records after the server move."""
    remote_old_prefix = remote_old_prefix or old_prefix
    remote_new_prefix = remote_new_prefix or new_prefix
    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(remote_old_prefix + "/"):
            remote_path = remote_new_prefix + "/" + remote_path[len(remote_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)
    try:
        _, local_prefix = _silo_selection_context(
            args, root_state, base=base, device_token=device_token)
        remote_old_prefix = _silo_remote_path_for_local(old_prefix, local_prefix)
        remote_new_prefix = _silo_remote_path_for_local(new_prefix, local_prefix)
    except (TypeError, ValueError, RuntimeError) as exc:
        print(f"Silo folder move selection is invalid: {exc}", file=sys.stderr)
        return 2
    snapshot, detail = _silo_current_snapshot(
        base, device_token, (remote_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, (remote_new_prefix,))
        if destination is not None:
            mapped = []
            for item in destination.values():
                path = str(item.get("path") or "")
                if path.startswith(remote_new_prefix + "/"):
                    mapped_item = dict(item)
                    mapped_item["path"] = (
                        remote_old_prefix + "/" + path[len(remote_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,
                        remote_old_prefix=remote_old_prefix)
                    if plan:
                        _silo_apply_local_prefix_move(
                            root, root_state, old_prefix, new_prefix, plan, [],
                            remote_old_prefix=remote_old_prefix,
                            remote_new_prefix=remote_new_prefix)
                        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,
            remote_old_prefix=remote_old_prefix)
    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": remote_old_prefix,
        "new_prefix": remote_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 [],
            remote_old_prefix=remote_old_prefix,
            remote_new_prefix=remote_new_prefix)
    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)
    try:
        prefixes, local_prefix = _silo_selection_context(
            args, root_state, base=base, device_token=device_token)
        remote_old_path = _silo_remote_path_for_local(old_relative, local_prefix)
        remote_new_path = _silo_remote_path_for_local(new_relative, local_prefix)
    except (TypeError, ValueError, RuntimeError) as exc:
        print(f"Silo move selection is invalid: {exc}", file=sys.stderr)
        return 2
    record = root_state.get("files", {}).get(old_relative)
    if not isinstance(record, dict):
        _, record = _silo_find_record(
            root_state.get("files", {}), remote_old_path)
    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": remote_old_path,
            "new_path": remote_new_path,
            "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": remote_new_path})
    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
    _silo_set_selection(root_state, prefixes, local_prefix)
    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:
    metrics = _silo_sync_metrics(args)
    metrics["push_status"] = "running"
    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, local_prefix = _silo_selection_context(
            args, root_state, base=base, device_token=device_token)
    except (TypeError, ValueError, RuntimeError) as exc:
        print(f"Silo push selection is invalid: {exc}", file=sys.stderr)
        return 2
    _silo_set_selection(root_state, prefixes, local_prefix)
    paths = []
    only_path = str(getattr(args, "only_path", "") or "").replace("\\", "/").strip("/")
    for path in sorted(root.rglob("*")):
        if path.is_symlink() or not path.is_file():
            continue
        relative = path.relative_to(root).as_posix()
        if only_path and relative != only_path:
            continue
        if relative.startswith(".havre/") or relative.startswith(".silo/"):
            continue
        remote_path = _silo_remote_path_for_local(relative, local_prefix)
        if not _silo_path_selected(remote_path, prefixes):
            continue
        paths.append((relative, path, remote_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
    book_intake_id = str(getattr(args, "book_intake_id", "") or "").strip() or None
    for relative, source, remote_path 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
                    and str(previous.get("remote_path") or previous.get("path") or relative)
                    == remote_path):
                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,
                book_intake_id=book_intake_id, remote_path=remote_path)
            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, local_path=relative)
            save_silo_state(state)
            try:
                meta_path.unlink()
                blob_path.unlink()
            except FileNotFoundError:
                pass
            uploaded += 1
            metrics["bytes_uploaded"] += int(meta.get("expected_size_bytes") or 0)
            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)
    metrics["uploaded_files"] = uploaded
    metrics["deleted_files"] += deleted
    metrics["skipped_files"] += skipped
    metrics["push_status"] = "failed" if failed else "completed"
    if failed and not metrics.get("error_code"):
        metrics["error_code"] = "push_failed"
    print(f"Silo push: {uploaded} uploaded, {deleted} deleted, {skipped} unchanged, {failed} failed")
    return 1 if failed else 0


def _silo_infer_book_format(path: pathlib.Path) -> str:
    suffix = path.suffix.casefold()
    return {
        ".pdf": "pdf",
        ".epub": "epub",
        ".mobi": "ebook",
        ".azw": "ebook",
        ".azw3": "ebook",
        ".cbz": "comic",
        ".cbr": "comic",
    }.get(suffix, "scan")


def cmd_silo_book_push(args, key) -> int:
    """Scan and preflight one private book locally, then upload and bind its edition."""
    base, token = _require_token(args)
    try:
        _silo_device_required()
        active_key = _silo_active_key(key)
    except (RuntimeError, SystemExit) as exc:
        print(f"Silo book push cannot use the local device: {exc}", file=sys.stderr)
        return 1
    root = pathlib.Path(args.root).expanduser().resolve()
    source = pathlib.Path(args.file).expanduser().resolve()
    if not source.is_file():
        print(f"book file does not exist: {source}", file=sys.stderr)
        return 2
    try:
        relative = source.relative_to(root).as_posix()
    except ValueError:
        print("book file must be inside the selected Silo sync root", file=sys.stderr)
        return 2
    scan = {}
    if not bool(getattr(args, "no_scan", False)):
        try:
            scan = silo_book_scanner.scan_book(source)
        except (OSError, ValueError) as exc:
            print(f"local book scan failed: {exc}", file=sys.stderr)
            return 2
        candidates = list(scan.get("isbn_candidates") or [])
        explicit_isbn = str(getattr(args, "isbn", "") or "").strip()
        if len(candidates) > 1 and not explicit_isbn:
            print(
                "book scan found multiple ISBNs; specify --isbn for the edition "
                "before any upload",
                file=sys.stderr,
            )
            return 2
        if explicit_isbn:
            try:
                normalized_explicit = silo_book_scanner.normalize_isbn(explicit_isbn)
            except ValueError as exc:
                print(f"invalid --isbn: {exc}", file=sys.stderr)
                return 2
            if candidates and normalized_explicit not in candidates:
                print(
                    "book scan ISBN conflicts with --isbn; review the edition before "
                    "any upload",
                    file=sys.stderr,
                )
                return 2
            explicit_isbn = normalized_explicit
        if not explicit_isbn and len(candidates) == 1:
            args.isbn = candidates[0]
        elif explicit_isbn:
            args.isbn = explicit_isbn
        if not str(getattr(args, "title", "") or "").strip() and scan.get("title"):
            args.title = scan["title"]
        if not str(getattr(args, "author", "") or "").strip() and scan.get("author"):
            args.author = scan["author"]
        if not str(getattr(args, "publisher", "") or "").strip() and scan.get("publisher"):
            args.publisher = scan["publisher"]
        if (not str(getattr(args, "language", "") or "").strip()
                or str(getattr(args, "language", "")).strip() == "und") and scan.get("language"):
            args.language = scan["language"]
        if getattr(args, "publication_year", None) is None and scan.get("publication_year"):
            args.publication_year = scan["publication_year"]
        if not str(getattr(args, "format", "") or "").strip() and scan.get("format"):
            args.format = scan["format"]
        if scan.get("isbn"):
            print(f"book identified locally: ISBN {scan['isbn']} ({', '.join(scan.get('evidence', []))})")
        elif scan.get("warning"):
            print(f"book scan: {scan['warning']}")

    if not str(getattr(args, "title", "") or "").strip() or not str(
            getattr(args, "author", "") or "").strip():
        print(
            "silo-book-push needs --title and --author when local book metadata "
            "does not provide them",
            file=sys.stderr,
        )
        return 2
    _, _, content_identity = _silo_file_digests(active_key, source, False)
    title = str(args.title).strip()
    author = str(args.author).strip()
    format_name = str(getattr(args, "format", "") or "").strip() \
        or _silo_infer_book_format(source)
    descriptor = "\0".join((
        content_identity,
        title,
        author,
        str(getattr(args, "isbn", "") or "").strip(),
        str(getattr(args, "language", "") or "und").strip() or "und",
        str(getattr(args, "publisher", "") or "").strip(),
        str(getattr(args, "publication_year", "") or ""),
        format_name,
    ))
    intake_id = "cli-book-" + hashlib.sha256(descriptor.encode("utf-8")).hexdigest()[:48]
    intake_body = {
        "intake_id": intake_id,
        "title": title,
        "author": author,
        "isbn": str(getattr(args, "isbn", "") or "").strip(),
        "language": str(getattr(args, "language", "") or "und").strip() or "und",
        "publisher": str(getattr(args, "publisher", "") or "").strip(),
        "publication_year": getattr(args, "publication_year", None),
        "format": format_name,
        "source_kind": "private",
        "source_url": "",
        "rights_declaration": "",
        "confidence": float(getattr(args, "confidence", 0.99) or 0.99),
        "content_identity": content_identity,
    }
    status, outcome = _api(base, "POST", "/api/silo/books/intake", intake_body, token=token)
    intake = outcome.get("intake") if isinstance(outcome.get("intake"), dict) else {}
    decision = str(intake.get("decision") or "")
    intake_status = str(intake.get("status") or "")
    if intake_status == "linked" or (status == 200 and decision == "exact-duplicate"):
        print(f"book duplicate skipped: '{relative}' already has a tenant-local object")
        return 0
    if status == 202 or decision == "needs-review" or intake_status == "review":
        print("book identity needs review; no book bytes were uploaded", file=sys.stderr)
        return 1
    if status not in (200, 201) or decision not in {"new-book", "new-edition"} \
            or intake_status != "awaiting_upload":
        print(f"book preflight failed: {_silo_map_error(status, outcome)}", file=sys.stderr)
        return 1

    args.only_path = relative
    args.book_intake_id = str(intake.get("intake_id") or intake_id)
    args.public_library = False
    args.include_prefix = None
    args.include_all = False
    push_status = cmd_silo_push(args, active_key)
    if push_status == 0:
        print(f"book upload bound to edition '{args.book_intake_id}'")
    return push_status


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:
    metrics = _silo_sync_metrics(args)
    metrics["pull_status"] = "running"
    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, local_prefix = _silo_selection_context(
            args, root_state, base=base, device_token=device_token)
    except (TypeError, ValueError, RuntimeError) 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,
            local_prefix):
        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 "")
            action = str(change.get("action") or "")
            folder_change = action in {
                "folder_create", "folder_move", "folder_delete", "folder_restore"}
            previous_remote_path = str(change.get("previous_path") 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)
            previous_selected = bool(
                previous_remote_path
                and _silo_path_selected(previous_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
            folder_leaving_scope = (
                folder_change and action == "folder_move" and previous_selected)
            if not remote_selected and not tracked_selected and not folder_leaving_scope:
                cursor = change_cursor
                root_state["cursor"] = cursor
                save_silo_state(state)
                skipped += 1
                continue
            if not remote_selected:
                try:
                    if folder_leaving_scope:
                        _silo_remove_remote_folder_change(
                            root, change, local_prefix=local_prefix)
                    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:
                local_path = _silo_local_path_for_remote(remote_path, local_prefix)
                target = _silo_safe_target(root, local_key or local_path)
                if folder_change:
                    _silo_apply_remote_folder_change(
                        root, change, local_prefix=local_prefix,
                        source_in_scope=previous_selected)
                elif 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}'")
                    metrics["deleted_files"] += 1
                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,
                        local_prefix=local_prefix)
                else:
                    _silo_materialize_remote_file(
                        base, device_token, key, root, root_state, change,
                        local_key=local_key, local_record=local_record,
                        local_prefix=local_prefix)
                    metrics["downloaded_files"] += 1
                    metrics["bytes_downloaded"] += max(0, int(change.get("size_bytes") or 0))
                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)
    metrics["cursor"] = cursor
    metrics["skipped_files"] += skipped
    metrics["pull_status"] = "failed" if failed else "completed"
    if failed and not metrics.get("error_code"):
        metrics["error_code"] = "pull_failed"
    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:
    metrics = _new_silo_sync_metrics()
    setattr(args, "_silo_sync_metrics", metrics)
    if hasattr(args, "_silo_collection_resolved"):
        delattr(args, "_silo_collection_resolved")
    started_at = datetime.datetime.now(datetime.timezone.utc).isoformat().replace(
        "+00:00", "Z")
    state = load_silo_state()
    device_token = str(state.get("device_token") or "")
    status = 0
    try:
        if not getattr(args, "pull_only", False):
            push_status = cmd_silo_push(args, key)
            status = max(status, push_status)
            if push_status and metrics.get("push_status") == "running":
                metrics["push_status"] = "failed"
                metrics["error_code"] = metrics.get("error_code") or "push_failed"
        if not getattr(args, "push_only", False):
            pull_status = cmd_silo_pull(args, key)
            status = max(status, pull_status)
            if pull_status and metrics.get("pull_status") == "running":
                metrics["pull_status"] = "failed"
                metrics["error_code"] = metrics.get("error_code") or "pull_failed"
        return status
    finally:
        if metrics.get("push_status") == "running":
            metrics["push_status"] = "failed"
            metrics["error_code"] = metrics.get("error_code") or "push_failed"
        if metrics.get("pull_status") == "running":
            metrics["pull_status"] = "failed"
            metrics["error_code"] = metrics.get("error_code") or "pull_failed"
        if device_token:
            _silo_submit_sync_receipt(
                server_url(args), device_token, args, started_at, metrics)


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,
                collection=args.collection,
                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-collections": cmd_silo_collections,
            "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-book-push": cmd_silo_book_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")
    collections = sub.add_parser(
        "silo-collections",
        help="list server-backed Silo collections using the active device",
    )
    collections.add_argument("--json", action="store_true",
                             help="print machine-readable collection metadata")

    sd = sub.add_parser("silo-device", help="register, enroll, setup, list, revoke, grant, or recover a Silo device")
    sd.add_argument("action", choices=("register", "enroll", "setup", "list", "revoke", "grant",
                                        "recovery-create", "recovery-import"))
    sd.add_argument("--name", default="Havre remote client",
                    help="device name for register, enroll, setup, or recovery-create")
    sd.add_argument("--code", help="one-time code copied from the authenticated Silo browser for enroll or setup")
    sd.add_argument("--root", help="local Silo sync directory for setup")
    sd.add_argument("--scope", default="CA", help="sealed-object residency scope for setup")
    sd.add_argument(
        "--public-library",
        action="store_true",
        help="use P0 for rights-cleared books or music that may be public during setup",
    )
    sd.add_argument(
        "--part-size-bytes",
        type=int,
        default=8 * 1024 * 1024,
        help="ciphertext chunk size for resumable uploads during setup",
    )
    sd.add_argument("--id", help="device id for revoke or target device for grant; defaults to the local device for revoke")
    sd.add_argument(
        "--collection", metavar="ID_OR_NAME",
        help=("for setup, sync only this server-backed collection; for grant, authorize "
              "only existing private objects in this collection"),
    )
    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 enrollment or 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.add_argument(
        "--collection", metavar="ID_OR_NAME",
        help="resolve a collection root for the local-to-remote path mapping",
    )

    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")
    move_dir.add_argument(
        "--collection", metavar="ID_OR_NAME",
        help="resolve a collection root for the local-to-remote path mapping",
    )

    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(
            "--collection",
            metavar="ID_OR_NAME",
            help=("select one server-backed Silo collection by id or case-insensitive "
                  "name; the resolved stable root is saved for this sync root"),
        )
        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)

    book_push = sub.add_parser(
        "silo-book-push",
        help="scan and identify one private book locally, then encrypt, upload, and bind its edition",
    )
    book_push.add_argument("root", help="local Silo sync directory")
    book_push.add_argument("file", help="book file inside the sync directory")
    book_push.add_argument("--title", default="", help="book title; EPUB metadata may supply it")
    book_push.add_argument("--author", default="", help="book author; EPUB metadata may supply it")
    book_push.add_argument("--isbn", help="ISBN-10 or ISBN-13 from the cover barcode")
    book_push.add_argument("--language", default="und")
    book_push.add_argument("--publisher", default="")
    book_push.add_argument("--publication-year", type=int)
    book_push.add_argument("--format", default="")
    book_push.add_argument("--confidence", type=float, default=0.99)
    book_push.add_argument(
        "--no-scan", action="store_true",
        help="skip the bounded local ISBN/EPUB metadata scan; manual metadata is required",
    )
    book_push.add_argument("--scope", default="CA", help="sealed-object residency scope")
    book_push.add_argument(
        "--part-size-bytes", type=int, default=8 * 1024 * 1024,
        help="ciphertext chunk size for resumable uploads",
    )
    book_push.add_argument(
        "--collection",
        metavar="ID_OR_NAME",
        help=("select one server-backed Silo collection by id or case-insensitive "
              "name; book contents are mounted at the local sync root"),
    )
    book_push.set_defaults(
        public_library=False, include_prefix=None, include_all=False, collection=None)

    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(
        "--collection", metavar="ID_OR_NAME",
        help="select one server-backed Silo collection by id or name",
    )
    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 in {"silo-service", "silo-collections"} 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)
    session = load_session()
    check("session token is stored as a private local file",
          bool(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")

    enrollment_status, enrollment_out = _api(
        base, "POST", "/api/silo/device-enrollment", {}, token=load_session()["token"])
    enroll = A(); enroll.server = None; enroll.action = "enroll"
    enroll.name = "CLI enrollment self-test"; enroll.id = None
    enroll.code = enrollment_out.get("enrollment", {}).get("code", "")
    enroll.replace = False
    check("Silo CLI redeems a browser enrollment code with an independent local key",
          enrollment_status == 201 and bool(enroll.code)
          and cmd_silo_device(enroll, key) == 0
          and bool(load_silo_state().get("device_token"))
          and load_silo_state().get("silo_private") != key["private"])

    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")
    setup_enrollment_status, setup_enrollment_out = _api(
        base, "POST", "/api/silo/device-enrollment", {}, token=load_session()["token"])
    setup = A(); setup.server = None; setup.action = "setup"
    setup.name = "CLI first-sync self-test"; setup.id = None
    setup.code = setup_enrollment_out.get("enrollment", {}).get("code", "")
    setup.root = str(source_root); setup.scope = "CA"
    setup.public_library = False; setup.part_size_bytes = 3
    setup.collection = None; setup.include_prefix = None; setup.include_all = False
    setup.pull_only = False; setup.push_only = False; setup.replace = True
    check("Silo CLI setup enrolls and runs the first encrypted sync",
          setup_enrollment_status == 201 and bool(setup.code)
          and cmd_silo_device(setup, key) == 0
          and (source_root / "Documents" / "memo.txt").read_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:
        active_silo_private = load_silo_state().get("silo_private", "")
        recovery_updated_object = envelope.add_recipient(
            sealed.get("object", {}), active_silo_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)

    # Exercise the actual second-device sync path with a separate local home and
    # recipient key. This is deliberately still a deterministic mock pilot, but it
    # catches the gap between a grant that can decrypt one object and a replacement
    # device that can initialize a real sync root and save its own receipt.
    primary_home = os.environ["HAVRE_HOME"]
    second_device_sync_ok = False
    second_device_receipt_ok = False
    if recovery_status == 201 and recovery_device.get("id") and recovery_device.get("device_token"):
        second_home = home / "silo-second-device"
        second_root = second_home / "silo-mirror"
        second_home.mkdir(parents=True, exist_ok=True)
        try:
            os.environ["HAVRE_HOME"] = str(second_home)
            save_session(session)
            save_silo_state({
                "device_id": recovery_device["id"],
                "device_name": recovery_device.get("name", "CLI recovery self-test"),
                "device_token": recovery_device["device_token"],
                "silo_private": recovery_private,
                "silo_public": recovery_public,
                "roots": {},
            })
            second_sync = A()
            second_sync.server = None
            second_sync.root = str(second_root)
            second_sync.scope = "CA"
            second_sync.public_library = False
            second_sync.part_size_bytes = 3
            second_sync.pull_only = False
            second_sync.push_only = False
            second_sync.limit = 100
            second_sync.include_prefix = None
            second_sync.include_all = False
            second_device_sync_ok = (
                cmd_silo_sync(second_sync, {"private": recovery_private,
                                             "public": recovery_public}) == 0
                and (second_root / "Documents" / "memo.txt").read_bytes()
                == b"encrypted Silo client payload"
            )
            receipt_status, receipt_out, _ = _silo_api(
                base, "GET", "/api/silo/sync-receipts?limit=20",
                token=session["token"])
            second_device_receipt_ok = (
                receipt_status == 200
                and any(
                    item.get("device_name") == recovery_device.get(
                        "name", "CLI recovery self-test")
                    and item.get("status") == "completed"
                    and item.get("downloaded_files", 0) >= 1
                    for item in receipt_out.get("receipts", [])
                )
            )
        except (OSError, RuntimeError, ValueError, KeyError, TypeError):
            second_device_sync_ok = False
            second_device_receipt_ok = False
        finally:
            os.environ["HAVRE_HOME"] = primary_home
    check("Silo second device syncs and decrypts a granted file",
          second_device_sync_ok)
    check("Silo second device saves a bounded sync receipt",
          second_device_receipt_ok)

    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)
    receipt_status, receipt_out, _ = _silo_api(
        base, "GET", "/api/silo/sync-receipts?limit=5",
        token=load_session()["token"])
    latest_receipt = (receipt_out.get("receipts") or [{}])[0]
    check("Silo sync persists a safe customer-visible receipt",
          receipt_status == 200
          and latest_receipt.get("status") == "completed"
          and latest_receipt.get("pull_status") == "completed"
          and "device_token" not in latest_receipt
          and "tenant_id" not in latest_receipt)
    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")
    metadata_folder_root = home / "silo-metadata-folders"
    _silo_apply_remote_folder_change(
        metadata_folder_root,
        {"action": "folder_create", "path": "Empty/Child", "cursor": 11})
    check("Silo pull materializes an empty folder without an object download",
          (metadata_folder_root / "Empty" / "Child").is_dir())
    _silo_apply_remote_folder_change(
        metadata_folder_root,
        {"action": "folder_move", "previous_path": "Empty/Child",
         "path": "Archive/Child", "cursor": 12})
    check("Silo pull moves an empty folder without changing file state",
          not (metadata_folder_root / "Empty" / "Child").exists()
          and (metadata_folder_root / "Archive" / "Child").is_dir())
    folder_trash_root = home / "silo-folder-trash"
    _silo_apply_remote_folder_change(
        folder_trash_root,
        {"action": "folder_create", "path": "Empty/TrashMe", "cursor": 15})
    _silo_apply_remote_folder_change(
        folder_trash_root,
        {"action": "folder_delete", "path": "Empty/TrashMe", "cursor": 16})
    _silo_apply_remote_folder_change(
        folder_trash_root,
        {"action": "folder_restore", "path": "Empty/TrashMe", "cursor": 17})
    check("Silo pull removes and restores an empty folder without object access",
          (folder_trash_root / "Empty" / "TrashMe").is_dir())
    (folder_trash_root / "Empty" / "TrashMe" / "local.txt").write_bytes(b"keep me")
    _silo_apply_remote_folder_change(
        folder_trash_root,
        {"action": "folder_delete", "path": "Empty/TrashMe", "cursor": 18})
    check("Silo pull preserves unexpected local folder contents on Trash",
          not (folder_trash_root / "Empty" / "TrashMe").exists()
          and any(
              (item / "local.txt").read_bytes() == b"keep me"
              for item in (folder_trash_root / "Empty").glob(
                  "TrashMe.local-before-folder-delete-*")))
    (folder_trash_root / "Empty" / "Parent" / "Child").mkdir(parents=True)
    _silo_apply_remote_folder_change(
        folder_trash_root,
        {"action": "folder_delete", "path": "Empty/Parent/Child", "cursor": 19})
    check("Silo folder Trash preserves an empty parent directory",
          (folder_trash_root / "Empty" / "Parent").is_dir()
          and not (folder_trash_root / "Empty" / "Parent" / "Child").exists())
    collision_root = home / "silo-folder-collision"
    collision_root.mkdir()
    (collision_root / "Blocker").write_bytes(b"local file")
    _silo_apply_remote_folder_change(
        collision_root,
        {"action": "folder_restore", "path": "Blocker/Restored", "cursor": 20})
    check("Silo folder restore preserves a blocking local parent file",
          (collision_root / "Blocker" / "Restored").is_dir()
          and any(
              item.read_bytes() == b"local file"
              for item in collision_root.glob("Blocker.local-before-folder-*")))
    scoped_folder_root = home / "silo-scoped-folders"
    _silo_apply_remote_folder_change(
        scoped_folder_root,
        {"action": "folder_move", "previous_path": "Photos/Empty",
         "path": "Documents/Empty", "cursor": 13},
        local_prefix="Documents", source_in_scope=False)
    check("Silo selective pull materializes a folder moving into scope",
          (scoped_folder_root / "Empty").is_dir()
          and not (scoped_folder_root / "Photos" / "Empty").exists())
    leaving_folder_root = home / "silo-leaving-folder"
    (leaving_folder_root / "Documents" / "Empty").mkdir(parents=True)
    _silo_remove_remote_folder_change(
        leaving_folder_root,
        {"action": "folder_move", "previous_path": "Documents/Empty",
         "path": "Photos/Empty", "cursor": 14})
    check("Silo selective pull removes an empty folder leaving scope",
          not (leaving_folder_root / "Documents" / "Empty").exists())

    # 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)

    # Private book bytes follow the same local-identity rule, but the client
    # must preflight the edition before it stages any ciphertext. The second
    # copy intentionally lives under another sync root to prove that the
    # tenant-local identity, not the local path, controls deduplication.
    book_root = home / "silo-book"
    book_root.mkdir()
    book_file = book_root / "Scans" / "book.pdf"
    book_file.parent.mkdir()
    book_file.write_bytes(b"private scanned book edition one")
    book_push = A(); book_push.server = None
    book_push.root = str(book_root); book_push.file = str(book_file)
    book_push.title = "The Pilot Book"; book_push.author = "A Havre Reader"
    book_push.isbn = ""; book_push.language = "eng"; book_push.publisher = "Havre Press"
    book_push.publication_year = 2026; book_push.format = "pdf"
    book_push.confidence = 0.99; book_push.scope = "CA"; book_push.part_size_bytes = 5
    book_push.public_library = False; book_push.include_prefix = None
    book_push.include_all = False
    book_identity = _silo_file_digests(_silo_active_key(key), book_file, False)[2]
    book_intake_id = "cli-book-" + hashlib.sha256(
        "\0".join((
            book_identity,
            book_push.title,
            book_push.author,
            book_push.isbn,
            book_push.language,
            book_push.publisher,
            str(book_push.publication_year),
            book_push.format,
        )).encode("utf-8")).hexdigest()[:48]
    check("Silo CLI preflights and binds a private scanned book",
          cmd_silo_book_push(book_push, key) == 0
          and _api(base, "GET", "/api/silo/files?prefix=Scans",
                   token=load_session()["token"])[1].get("files", [])
          and _api(base, "POST", "/api/silo/books/intake", {
              "intake_id": book_intake_id,
              "title": book_push.title,
              "author": book_push.author,
              "language": book_push.language,
              "publisher": book_push.publisher,
              "publication_year": book_push.publication_year,
              "format": book_push.format,
              "source_kind": "private",
              "confidence": book_push.confidence,
              "content_identity": book_identity,
          }, token=load_session()["token"])[1].get("intake", {}).get("status")
          == "linked")

    auto_book = book_root / "Books" / "auto-identified.epub"
    auto_book.parent.mkdir()
    with zipfile.ZipFile(auto_book, "w", compression=zipfile.ZIP_DEFLATED) as archive:
        archive.writestr(
            "META-INF/container.xml",
            '<?xml version="1.0"?><container><rootfiles><rootfile '
            'full-path="OEBPS/content.opf"/></rootfiles></container>',
        )
        archive.writestr(
            "OEBPS/content.opf",
            '<package><metadata xmlns:dc="http://purl.org/dc/elements/1.1/">'
            '<dc:title>Auto Identified Book</dc:title><dc:creator>Local Scanner</dc:creator>'
            '<dc:publisher>Havre Press</dc:publisher><dc:language>eng</dc:language>'
            '<dc:identifier>urn:isbn:9780451524935</dc:identifier>'
            '</metadata></package>',
        )
    auto_push = A(); auto_push.server = None
    auto_push.root = str(book_root); auto_push.file = str(auto_book)
    auto_push.title = ""; auto_push.author = ""; auto_push.isbn = ""
    auto_push.language = "und"; auto_push.publisher = ""; auto_push.publication_year = None
    auto_push.format = ""; auto_push.confidence = 0.99; auto_push.scope = "CA"
    auto_push.part_size_bytes = 5; auto_push.public_library = False
    auto_push.include_prefix = None; auto_push.include_all = False; auto_push.no_scan = False
    auto_files = _api(base, "GET", "/api/silo/files?prefix=Books",
                      token=load_session()["token"])[1].get("files", [])
    check("Silo CLI fills metadata from an EPUB before upload",
          cmd_silo_book_push(auto_push, key) == 0
          and any(item.get("path") == "Books/auto-identified.epub" for item in auto_files
                  + _api(base, "GET", "/api/silo/files?prefix=Books",
                         token=load_session()["token"])[1].get("files", [])))
    duplicate_root = home / "silo-book-duplicate"
    duplicate_root.mkdir()
    duplicate_file = duplicate_root / "Copies" / "book.pdf"
    duplicate_file.parent.mkdir()
    duplicate_file.write_bytes(book_file.read_bytes())
    duplicate_push = A(); duplicate_push.server = None
    duplicate_push.root = str(duplicate_root); duplicate_push.file = str(duplicate_file)
    duplicate_push.title = book_push.title; duplicate_push.author = book_push.author
    duplicate_push.isbn = book_push.isbn; duplicate_push.language = book_push.language
    duplicate_push.publisher = book_push.publisher
    duplicate_push.publication_year = book_push.publication_year
    duplicate_push.format = book_push.format; duplicate_push.confidence = book_push.confidence
    duplicate_push.scope = "CA"; duplicate_push.part_size_bytes = 5
    duplicate_push.public_library = False; duplicate_push.include_prefix = None
    duplicate_push.include_all = False
    usage_before_book_duplicate = _api(
        base, "GET", "/api/silo/usage", token=load_session()["token"])[1]
    duplicate_status = cmd_silo_book_push(duplicate_push, key)
    usage_after_book_duplicate = _api(
        base, "GET", "/api/silo/usage", token=load_session()["token"])[1]
    duplicate_files = _api(
        base, "GET", "/api/silo/files?prefix=Copies",
        token=load_session()["token"])[1].get("files", [])
    check("Silo CLI skips an exact book duplicate before a second upload",
          duplicate_status == 0
          and usage_after_book_duplicate.get("usage", {}).get("used_bytes")
          == usage_before_book_duplicate.get("usage", {}).get("used_bytes")
          and not any(item.get("path") == "Copies/book.pdf" for item in duplicate_files))

    collection_status, collection_out = _api(
        base, "POST", "/api/silo/collections",
        {"name": "Private scans", "kind": "documents", "mode": "private"},
        token=load_session()["token"],
    )
    pilot_collection = collection_out.get("collection", {})
    collection_list = A(); collection_list.server = None; collection_list.json = True
    check("Silo CLI lists safe collections with its device credential",
          collection_status == 201 and bool(pilot_collection.get("id"))
          and cmd_silo_collections(collection_list, key) == 0)
    collection_source = home / "silo-collection-source"
    collection_source.mkdir()
    (collection_source / "scan.txt").write_bytes(b"selected collection payload")
    collection_push = A(); collection_push.server = None
    collection_push.root = str(collection_source); collection_push.scope = "CA"
    collection_push.public_library = False; collection_push.part_size_bytes = 3
    collection_push.collection = "Private scans"
    collection_push.include_prefix = None; collection_push.include_all = False
    collection_state = lambda: load_silo_state()["roots"].get(
        str(collection_source.resolve()), {})
    check("Silo CLI selects a collection by name for encrypted push",
          cmd_silo_push(collection_push, key) == 0
          and collection_state().get("files", {}).get("scan.txt", {}).get("remote_path")
          == pilot_collection["root_prefix"] + "/scan.txt")
    outside_root = home / "silo-outside-scope"
    outside_root.mkdir()
    outside_file = outside_root / "Outside" / "other.txt"
    outside_file.parent.mkdir()
    outside_file.write_bytes(b"outside collection grant scope")
    outside_push = A(); outside_push.server = None
    outside_push.root = str(outside_root); outside_push.scope = "CA"
    outside_push.public_library = False; outside_push.part_size_bytes = 3
    outside_push.collection = None; outside_push.include_prefix = None
    outside_push.include_all = False
    check("Silo CLI creates an object outside the selected collection",
          cmd_silo_push(outside_push, key) == 0)
    scoped_private, scoped_public = envelope.gen_recipient_keypair()
    scoped_device_status, scoped_device_out, _ = _silo_api(
        base, "POST", "/api/silo/devices",
        {"name": "Scoped grant self-test", "public_key": scoped_public},
        token=load_session()["token"],
    )
    scoped_device = scoped_device_out.get("device", {})
    scoped_grant = A(); scoped_grant.server = None; scoped_grant.action = "grant"
    scoped_grant.name = "Scoped grant self-test"; scoped_grant.id = scoped_device.get("id")
    scoped_grant.collection = "Private scans"; scoped_grant.code = None
    scoped_grant.output = None; scoped_grant.bundle = None; scoped_grant.replace = False
    check("Silo CLI scopes an existing-device grant to one collection",
          scoped_device_status == 201
          and bool(scoped_device.get("device_token"))
          and cmd_silo_device(scoped_grant, key) == 0)
    collection_files = _api(
        base, "GET", "/api/silo/files?" + urlencode(
            {"prefix": pilot_collection.get("root_prefix", "")}),
        token=load_session()["token"],
    )[1].get("files", [])
    outside_files = _api(
        base, "GET", "/api/silo/files?prefix=Outside",
        token=load_session()["token"],
    )[1].get("files", [])
    scoped_collection_status = scoped_outside_status = 0
    scoped_collection_object = scoped_outside_object = {}
    if collection_files and outside_files and scoped_device.get("device_token"):
        scoped_collection_status, scoped_collection_object, _ = _silo_api(
            base, "GET", f"/api/silo/objects/{collection_files[0].get('object_id')}",
            device_token=scoped_device["device_token"],
        )
        scoped_outside_status, scoped_outside_object, _ = _silo_api(
            base, "GET", f"/api/silo/objects/{outside_files[0].get('object_id')}",
            device_token=scoped_device["device_token"],
        )
    scoped_target_ref = envelope.fingerprint(scoped_public)
    collection_wrappers = (scoped_collection_object.get("object") or {}).get("wrapped_deks", [])
    outside_wrappers = (scoped_outside_object.get("object") or {}).get("wrapped_deks", [])
    check("Silo scoped grant reaches the selected collection but not another object",
          scoped_collection_status == 200 and scoped_outside_status == 200
          and any(item.get("recipient_ref") == scoped_target_ref
                  for item in collection_wrappers if isinstance(item, dict))
          and not any(item.get("recipient_ref") == scoped_target_ref
                      for item in outside_wrappers if isinstance(item, dict)))
    collection_mirror = home / "silo-collection-mirror"
    collection_pull = A(); collection_pull.server = None
    collection_pull.root = str(collection_mirror); collection_pull.limit = 100
    collection_pull.collection = pilot_collection["id"]
    collection_pull.include_prefix = None; collection_pull.include_all = False
    collection_mirror_file = collection_mirror / "scan.txt"
    check("Silo CLI selects a collection by id for encrypted pull",
          cmd_silo_pull(collection_pull, key) == 0
          and collection_mirror_file.read_bytes() == b"selected collection payload"
          and not (collection_mirror / "Collections").exists())
    collection_move = A(); collection_move.server = None
    collection_move.root = str(collection_mirror)
    collection_move.old_path = "scan.txt"; collection_move.new_path = "renamed-scan.txt"
    collection_move.collection = pilot_collection["id"]
    check("Silo CLI moves a selected collection file at its local root",
          cmd_silo_move(collection_move, key) == 0
          and (collection_mirror / "renamed-scan.txt").read_bytes()
          == b"selected collection payload"
          and not collection_mirror_file.exists()
          and cmd_silo_pull(collection_pull, key) == 0)
    collection_delete = A(); collection_delete.server = None
    collection_delete.root = str(collection_mirror); collection_delete.scope = "CA"
    collection_delete.public_library = False; collection_delete.part_size_bytes = 3
    collection_delete.collection = pilot_collection["id"]
    collection_delete.include_prefix = None; collection_delete.include_all = False
    (collection_mirror / "renamed-scan.txt").unlink()
    collection_delete_status = cmd_silo_push(collection_delete, key)
    collection_files_after_delete = _api(
        base, "GET", "/api/silo/files?" + urlencode(
            {"prefix": pilot_collection["root_prefix"]}),
        token=load_session()["token"],
    )[1].get("files", [])
    check("Silo CLI propagates a selected collection deletion by remote path",
          collection_delete_status == 0
          and not any(item.get("path") == pilot_collection["root_prefix"]
                      + "/renamed-scan.txt"
                      for item in collection_files_after_delete))

    # 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())