"""Pairing: the deterministic code derivation AND the invite/redeem ceremony (§2.4).

The **derivation slice** (mint/normalize/derive) is pure — no network, no PyNaCl —
and was frozen first (the §10.1 amendment). The **ceremony** below builds on it:
each owner runs one side, they exchange a five-word code out of band, and the two
brains meet at a relay rendezvous to swap sealed offer/answer blobs, ending with
mutual peer entries + a verification-word check. The ceremony needs PyNaCl
(``SecretBox``) and the relay client, so it imports :func:`identity.sodium` lazily.

A code is five words, spoken or typed between two owners out of band. Both
sides derive the SAME rendezvous address and SecretBox key from the code alone::

    km  = pbkdf2_hmac("sha256", code, b"axon-pair-v1", 600_000, dklen=64)
    rid = base32(sha256(b"rid" + km[:32]).digest()[:16]).lower().rstrip("=")
    key = km[32:]

PBKDF2-HMAC-SHA256, NOT scrypt: `hashlib.scrypt` is missing from the macOS
system Python 3.9 (a LibreSSL build) that is this repo's real floor, and the
two ends of a pairing must compute identical bytes across platforms and
versions — explicit stdlib parameters can't drift the way libsodium's
opslimit/memlimit heuristics could. The parameters below are wire constants:
changing any of them strands every in-flight pairing code, so they are frozen
by a pinned test vector (tests/test_federation_pairing.py) exactly like the
wordlists.

Security margin (§2.4 rationale, §10.1 math): ~55 bits of code space, online
guessing limited to a single-use, 10-minute, rate-limited rendezvous — and even
the relay-compromise OFFLINE case costs ≈2^55 × 600k ≈ 2^74 SHA-256 operations.
"""
import base64
import hashlib
import re
import secrets

from alexandria_vault.federation.words import WORDS2048

CODE_WORDS = 5
# Wire constants — frozen (see module docstring). The salt shares the ``axon-``
# domain prefix that identity's verify domain already uses, so the constants stay
# consistent under one namespace and a future package rename (§9.5) really is
# names-only, not a format change.
PAIR_SALT = b"axon-pair-v1"
PBKDF2_ITERATIONS = 600_000
_WORD_RE = re.compile(r"^[a-z]+$")


def mint_code():
    """Five random words from the 2048-list, normalized form (CSPRNG; ~55 bits)."""
    return "-".join(secrets.choice(WORDS2048) for _ in range(CODE_WORDS))


def normalize_code(code):
    """Canonical form of a typed/spoken code: lowercase, five words, joined by '-'.

    Tolerates the ways humans transcribe — any mix of spaces/hyphens/newlines,
    stray whitespace, capitals. Refuses anything that isn't exactly five
    alphabetic words; membership in WORDS2048 is deliberately NOT required
    here (the redeem side derives from whatever was typed — use
    :func:`unknown_words` to warn about likely typos before burning time on
    the KDF or the rendezvous).
    """
    words = str(code).lower().replace("-", " ").split()
    if len(words) != CODE_WORDS:
        raise ValueError(f"a pairing code is exactly {CODE_WORDS} words, got {len(words)}")
    for w in words:
        if not _WORD_RE.match(w):
            raise ValueError(f"pairing-code words are letters only, got {w!r}")
    return "-".join(words)


def unknown_words(code):
    """Words in `code` that aren't in WORDS2048 — the CLI's typo warning."""
    lexicon = set(WORDS2048)
    return [w for w in normalize_code(code).split("-") if w not in lexicon]


def derive(code):
    """``(rid, key)`` from a pairing code — identical on both ends, by construction.

    ``rid`` is the public rendezvous address (26 base32 chars — safe to send to
    the relay: it reveals nothing about the code without the full KDF work).
    ``key`` is the 32-byte SecretBox key that seals the offer/answer blobs; it
    exists only in the two processes that know the code. Deliberately slow
    (~a few hundred ms): this runs once per pairing, and every attacker guess
    pays the same price.
    """
    code = normalize_code(code)
    km = hashlib.pbkdf2_hmac("sha256", code.encode("ascii"), PAIR_SALT,
                             PBKDF2_ITERATIONS, dklen=64)
    rid_raw = hashlib.sha256(b"rid" + km[:32]).digest()[:16]
    rid = base64.b32encode(rid_raw).decode("ascii").lower().rstrip("=")
    return rid, km[32:]


# ============================================================================
# The invite / redeem ceremony (§2.4, §5.4) — builds on the derivation above.
# ============================================================================
import datetime as dt        # noqa: E402
import json                  # noqa: E402

from alexandria_vault.federation import sodium              # noqa: E402
from alexandria_vault.federation import identity, state, envelope, outbox  # noqa: E402

DEPOSIT_TOKEN_BYTES = 32
# Default of what a freshly-paired peer may write into OUR vault — the owner
# tightens/loosens per peer in config afterwards. No area restriction (all areas);
# `event` is omitted so a new peer can't stamp the shared events note until trusted.
DEFAULT_PERMISSIONS = {"intents": ["todo", "note", "knowledge"]}


class PairingError(RuntimeError):
    """A pairing step failed in a way worth showing the owner verbatim."""


class RevokeUnconfirmed(PairingError):
    """:func:`remove_peer` could not get the relay to CONFIRM the deposit-token
    revoke — so the peer was NOT removed. Config entry and the stored
    ``minted_token_sha256`` are kept intact: that sha is the only handle that can
    ever name the token again (the relay has no list-tokens API), so dropping it
    on an unconfirmed revoke would leave the peer's access live forever."""


def _now_iso(now=None):
    return envelope._utc_ts(now)   # accepts None / epoch float / datetime


def _sha256_hex(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def _b64(raw):
    return base64.b64encode(bytes(raw)).decode("ascii")


def mint_deposit_token():
    """A fresh per-pairing token (b64 of 32 random bytes) the OTHER side presents
    when depositing into our mailbox."""
    return _b64(secrets.token_bytes(DEPOSIT_TOKEN_BYTES))


def _seal_blob(key, payload):
    box = sodium().secret.SecretBox(key)
    return bytes(box.encrypt(json.dumps(payload, ensure_ascii=False).encode("utf-8")))


def _open_blob(key, blob):
    box = sodium().secret.SecretBox(key)
    return json.loads(box.decrypt(bytes(blob)).decode("utf-8"))


# ---- our own identity, as the peer will file it -----------------------------
def _owner_name(cfg):
    owner = cfg.get("owner") or {}
    return (owner.get("name") if isinstance(owner, dict) else owner) or "peer"


def _slug(text):
    s = re.sub(r"[^a-z0-9]+", "-", (text or "").lower()).strip("-")
    s = re.sub(r"^[^a-z]+", "", s) or "peer"
    return s[:24]


def _our_handle(cfg):
    fed = cfg.get("federation") or {}
    return fed.get("handle") or _slug(_owner_name(cfg))


def _relay_url(cfg):
    url = (cfg.get("federation") or {}).get("relay_url")
    if not url:
        raise PairingError("no federation.relay_url in config — set it before pairing")
    return url


# ---- vault config peers[] ---------------------------------------------------
def _config_path(vault):
    from pathlib import Path
    return Path(vault) / "Agent" / "config.json"


def _write_config_atomically(path, cfg):
    """Write Agent/config.json via temp + os.replace, so a concurrent reader (the
    serve-thread reloading config on its interval) never sees a truncated file."""
    import os
    import tempfile
    from pathlib import Path
    path = Path(path)
    body = (json.dumps(cfg, indent=2, ensure_ascii=False) + "\n").encode("utf-8")
    fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix="." + path.name + ".")
    try:
        os.write(fd, body)
        os.fsync(fd)
        os.close(fd)
        os.replace(tmp, str(path))
    except BaseException:
        try:
            os.close(fd)
        except OSError:
            pass
        try:
            os.unlink(tmp)
        except FileNotFoundError:
            pass
        raise


def _unique_handle(peers, handle, fingerprint):
    """A handle not already used by a DIFFERENT peer — suffix -2, -3, … on collision,
    so a derived handle can never overwrite an established peer (dedup is by
    fingerprint; the handle is only a label)."""
    taken = {p.get("handle") for p in peers if p.get("fingerprint") != fingerprint}
    if handle not in taken:
        return handle
    i = 2
    while f"{handle}-{i}" in taken:
        i += 1
    return f"{handle}-{i}"


def upsert_peer(vault, peer):
    """Add or update a peer in Agent/config.json, dedup by FINGERPRINT only (the
    stable id). A handle collision with a DIFFERENT peer never clobbers it — the new
    peer's handle is disambiguated instead.

    A re-pair (same fingerprint) refreshes only transport/identity-neutral fields
    (pubkey, relay, name/labels) and PRESERVES the stored ``permissions`` and
    ``can_send``: the ceremony entry always carries the open DEFAULT_PERMISSIONS,
    and overlaying it would silently WIDEN a peer the owner had narrowed — a
    privilege regression the peer could trigger just by re-pairing. Only a
    brand-new fingerprint gets the defaults. Returns the entry as persisted."""
    path = _config_path(vault)
    cfg = json.loads(path.read_text(encoding="utf-8"))
    peers = cfg.setdefault("peers", [])
    for i, p in enumerate(peers):
        if p.get("fingerprint") == peer["fingerprint"]:
            merged = {**p, **peer, "handle": p.get("handle", peer["handle"])}
            for grant in ("permissions", "can_send"):
                if grant in p:                 # a re-pair never loosens the allowlist
                    merged[grant] = p[grant]
            peers[i] = merged
            peer = merged
            break
    else:
        peer = {**peer, "handle": _unique_handle(peers, peer["handle"], peer["fingerprint"])}
        peers.append(peer)
    _write_config_atomically(path, cfg)
    return peer


def _resolve_people_note(vault, name):
    """F17 — the dedup guard. Resolve the peer's name against the vault's People notes and
    return the note TITLE to link, or None. Uses the engine's resolve machinery scoped to
    ``People/`` by path (People notes carry no ``type: person``), and links ONLY when
    exactly one People note matches the name — unambiguous. A miss or a collision returns
    None, and provenance falls back to plain text (F7). Best-effort: never raises, and
    never CREATES a note (linking is only ever to a People note the owner already keeps)."""
    from pathlib import Path
    if not name:
        return None
    try:
        from alexandria_vault.vault_core.query import resolve as resolver
        res = resolver.resolve(Path(vault), name, kind="any", limit=20)
        people = [c for c in res.get("candidates", []) if c["rel"].startswith("People/")]
    except Exception:  # noqa: BLE001 — query machinery absent/unhappy → just don't link
        return None
    return people[0]["title"] if len(people) == 1 else None


def remove_peer(vault, handle, *, relay_client=None, cfg=None):
    """Drop a peer (by handle) — revoke its deposit token at the relay FIRST (the
    block button), and drop the config entry + stored secret only once the relay
    CONFIRMS the revoke. If we minted a token for this peer and the revoke can't
    be confirmed (relay unreachable, non-2xx, or no client/bearer to even ask),
    raise :class:`RevokeUnconfirmed` and change NOTHING — the stored sha is the
    only handle that can ever revoke that token, so it must outlive the failure
    for a retry. Returns the peer on success."""
    path = _config_path(vault)
    conf = json.loads(path.read_text(encoding="utf-8"))
    peers = conf.get("peers") or []
    match = next((p for p in peers if p.get("handle") == handle), None)
    if not match:
        raise PairingError(f"no peer with handle {handle!r}")
    fp = match.get("fingerprint")
    # Revoke the token WE minted for this peer so they can no longer deposit into our
    # mailbox — and require confirmation BEFORE we discard the sha we'd need to name
    # it again. (No sha stored means nothing of ours to revoke — plain removal.)
    sha = (state.peer_secret_get(fp) or {}).get("minted_token_sha256") if fp else None
    if sha:
        def _unconfirmed(why):
            return RevokeUnconfirmed(
                f"@{handle} NOT removed: the relay did not confirm revoking its "
                f"deposit token ({why}) — the peer entry and the revoke handle were "
                "kept; retry `peer remove` when the relay is reachable")
        try:
            mb = state.mailbox_for(_relay_url(cfg or conf))
        except PairingError:                       # no relay_url in config
            mb = {}
        if relay_client is None or not mb.get("bearer"):
            raise _unconfirmed("no relay client or mailbox bearer to authenticate "
                               "the revoke")
        _, pub = identity.load_or_create()
        our_fp = identity.fingerprint(pub)
        try:
            status, _, _ = _owner_op(
                cfg or conf, relay_client, mb["bearer"],
                lambda b: relay_client.delete_token(our_fp, b, sha))
        except PairingError as exc:                 # re-register refused (409 etc.)
            raise _unconfirmed(str(exc)) from exc
        except Exception as exc:  # noqa: BLE001 — unreachable: revoke NOT confirmed
            raise _unconfirmed(f"relay unreachable: {exc}") from exc
        if status not in (200, 204):
            raise _unconfirmed(f"relay answered {status}")
    conf["peers"] = [p for p in peers if p.get("handle") != handle]
    _write_config_atomically(path, conf)
    if fp:
        state.peer_secret_remove(fp)
        # Forget the fingerprint-keyed dedupe window + review counter, so re-pairing
        # the same brain later gets a fresh review_first_n buffer (not skipped).
        from alexandria_vault.engine import federation_state as fstate
        fstate.clear_peer(fp)
    return match


# ---- mailbox (lazy registration) --------------------------------------------
def ensure_mailbox(cfg, relay_client, *, now=None):
    """Return (our_fp, bearer, our_pub), registering our relay mailbox on first
    need (§5.4) and caching the bearer in mailbox.json."""
    _, pub = identity.load_or_create()
    our_fp = identity.fingerprint(pub)
    relay_url = _relay_url(cfg)
    mb = state.mailbox_for(relay_url)          # per-relay: a relay_url change never orphans
    # Reuse the cached bearer only if it was registered for THIS identity. A rotated /
    # restored brain.key has a new fingerprint, so the old bearer is for an orphaned
    # mailbox — re-register a fresh one rather than hand back a bearer that will 401.
    if mb.get("bearer") and mb.get("fingerprint") in (None, our_fp):
        return our_fp, mb["bearer"], pub
    status, payload = relay_client.create_mailbox(our_fp, _b64(pub))
    if status == 201:
        bearer = payload["bearer"]
    elif status == 409:
        raise PairingError(
            "the relay already has a mailbox for this brain but the local bearer is "
            "missing (mailbox.json lost) — a bearer is minted once and can't be "
            "recovered; contact the relay operator or rotate the identity")
    else:
        raise PairingError(f"mailbox registration failed ({status}): {payload}")
    state.mailbox_save(relay_url, bearer, _now_iso(now), fingerprint=our_fp)
    return our_fp, bearer, pub


def reregister_mailbox(cfg, relay_client, *, now=None, stale=None):
    """The relay answered **401** to our cached bearer ``stale``: its idle GC reclaimed
    the mailbox (no tokens, no blobs, 30 days — e.g. an invite nobody redeemed, whose
    token the reaper revoked). Register afresh (RELAY-REVIEW F4) — and NEVER forget the
    cached bearer before holding its replacement (§10 #1): a bearer is minted once, so
    a transient failure here (a 429 on registration, a blip) must leave the file as it
    was, and two pollers racing the same 401 must not destroy each other's fresh
    bearer. On a 409 the mailbox exists: if the cached bearer has meanwhile CHANGED
    from ``stale`` a sibling re-registered first and its bearer is the answer;
    otherwise this is the genuine "bearer lost" case."""
    _, pub = identity.load_or_create()
    our_fp = identity.fingerprint(pub)
    relay_url = _relay_url(cfg)
    status, payload = relay_client.create_mailbox(our_fp, _b64(pub))
    if status == 201:
        bearer = payload["bearer"]
        state.mailbox_save(relay_url, bearer, _now_iso(now), fingerprint=our_fp)
        return our_fp, bearer, pub
    if status == 409:
        current = state.mailbox_for(relay_url)
        if (current.get("bearer") and current.get("bearer") != stale
                and current.get("fingerprint") in (None, our_fp)):
            return our_fp, current["bearer"], pub          # a sibling got there first
        raise PairingError(
            "the relay already has a mailbox for this brain but the local bearer is "
            "refused (401) — a bearer is minted once and can't be recovered; contact the "
            "relay operator or rotate the identity")
    raise PairingError(f"mailbox re-registration failed ({status}): {payload}")


def _owner_op(cfg, relay_client, bearer, op, *, now=None):
    """Run ``op(bearer) -> (status, payload)``; on a 401, re-register the mailbox and
    retry ONCE with the fresh bearer. Returns ``(status, payload, bearer)`` so the
    caller keeps using the bearer that worked. Every owner-authenticated relay call
    in this module goes through here — a stale bearer must never be a dead end."""
    status, payload = op(bearer)
    if status != 401:
        return status, payload, bearer
    _, bearer, _ = reregister_mailbox(cfg, relay_client, now=now, stale=bearer)
    status, payload = op(bearer)
    return status, payload, bearer


def _our_blob(role, cfg, our_fp, our_pub, deposit_token, now):
    return {"role": role, "pubkey": _b64(our_pub), "fingerprint": our_fp,
            "name": _owner_name(cfg), "handle": _our_handle(cfg),
            "relay": _relay_url(cfg), "deposit_token": deposit_token,
            "ts": _now_iso(now)}


def _peer_entry(blob, label_name, label_handle, owner_named=True):
    # `_owner_named` (transient; popped before persist) records whether the peer's LABEL
    # was chosen by the owner vs defaulted from the peer's own self-declared name. F17
    # provenance-linking keys off it: a peer must not be able to forge a working [[link]]
    # to a real People note by declaring someone else's name (see confirm()).
    return {"name": label_name or blob.get("name") or "peer",
            "handle": label_handle or blob.get("handle") or "peer",
            "owner": blob.get("name"), "pubkey": blob["pubkey"],
            "fingerprint": blob["fingerprint"], "relay": blob.get("relay"),
            "permissions": dict(DEFAULT_PERMISSIONS), "can_send": True,
            "_owner_named": bool(owner_named)}


def _handshake(cfg, our_fp, our_pub, bearer, peer_blob, label_name, label_handle, our_token,
               *, owner_named=True):
    """Shared tail of both sides — compute the pairing result WITHOUT persisting.

    The verification words are the trust anchor (§2.2): the two owners must read
    them aloud and agree before anything goes live. So this only verifies the
    peer's fingerprint binds to its key and computes the words + the peer entry; it
    writes nothing. The owner then either :func:`confirm`s (words matched → persist)
    or :func:`abort`s (mismatch → revoke the token we exposed, persist nothing)."""
    peer_pub = base64.b64decode(peer_blob["pubkey"], validate=True)
    if identity.fingerprint(peer_pub) != peer_blob["fingerprint"]:
        raise PairingError("peer fingerprint does not match its public key — aborting")
    return {"peer": _peer_entry(peer_blob, label_name, label_handle, owner_named),
            "words": identity.verification_words(our_pub, peer_pub),
            "peer_fingerprint": peer_blob["fingerprint"],
            "peer_deposit_token": peer_blob["deposit_token"],
            "our_fp": our_fp, "our_token": our_token, "bearer": bearer}


def confirm(vault, handshake, *, now=None):
    """The owner confirmed the verification words match — NOW persist: store the
    peer's deposit token (state home), write the peer to config, and fire the
    pairing ping. Returns the persisted peer entry."""
    hs = handshake
    state.peer_secret_set(hs["peer_fingerprint"], deposit_token=hs["peer_deposit_token"],
                          minted_token_sha256=_sha256_hex(hs["our_token"]))
    peer_in = dict(hs["peer"])
    # F17 provenance link — ONLY when the owner chose this peer's label. A peer's own
    # self-declared name is attacker-controlled, so linking on it would let a peer forge a
    # working [[link]] (and backlinks) to a real People note (e.g. the owner's boss). The
    # invite side always owner-names; the join side only when the owner passed a name.
    owner_named = peer_in.pop("_owner_named", False)
    if owner_named:
        note = _resolve_people_note(vault, peer_in.get("name"))
        if note:
            peer_in["people_note"] = note
    peer = upsert_peer(vault, peer_in)
    pid = outbox.stage(hs["peer_fingerprint"], "ping", envelope.ping_body(), now=now)
    outbox.fire(pid, now=now)             # a pairing ping is owner-initiated egress
    return peer


def abort(relay_client, handshake, cfg=None):
    """The owner rejected the pairing (verification words did NOT match — a possible
    man-in-the-middle). Revoke the deposit token we registered so the unverified party
    can't deposit into our mailbox, and persist nothing. Returns True iff the revoke
    was confirmed by the relay (so the caller doesn't falsely claim it revoked).

    A 401 is confirmed ONLY when re-registering the mailbox succeeds (the relay had
    reclaimed it, and every token with it); a 401 from a wrong cached bearer against a
    live mailbox — where the token is still honoured — re-registers to a 409 and is
    NOT confirmed (RELAY-REVIEW §10 #4). Without ``cfg`` (no relay_url to re-register
    against) a 401 is not confirmed either."""
    hs = handshake
    if relay_client is None or not hs.get("bearer") or not hs.get("our_token"):
        return False
    sha = _sha256_hex(hs["our_token"])
    try:
        if cfg is not None:
            status, _, _ = _owner_op(cfg, relay_client, hs["bearer"],
                                     lambda b: relay_client.delete_token(hs["our_fp"], b, sha))
        else:
            status, _ = relay_client.delete_token(hs["our_fp"], hs["bearer"], sha)
        return status in (200, 204)
    except Exception:  # noqa: BLE001 — unreachable, or re-register refused: NOT confirmed
        return False


# ---- invite (side A) --------------------------------------------------------
def invite(vault, cfg, relay_client, name, handle=None, *, now=None):
    """Mint a code, register our mailbox + a deposit token, seal and deposit the
    offer at the rendezvous, and persist the pending invite. Returns the pending
    dict (carrying the code + verification is completed by :func:`wait`)."""
    sodium()   # preflight PyNaCl locally BEFORE any relay call — a missing dep must
    #            fail fast, never mid-ceremony after the network has been touched (F2)
    our_fp, bearer, pub = ensure_mailbox(cfg, relay_client, now=now)
    code = mint_code()
    rid, key = derive(code)
    our_token = mint_deposit_token()
    status, _, bearer = _owner_op(
        cfg, relay_client, bearer,
        lambda b: relay_client.add_token(our_fp, b, _sha256_hex(our_token)), now=now)
    if status != 201:
        raise PairingError(f"could not register the pairing deposit token ({status})")
    offer = _our_blob("offer", cfg, our_fp, pub, our_token, now)
    status, _ = relay_client.pair_put(rid, "offer", _seal_blob(key, offer))
    if status != 201:
        raise PairingError(f"could not deposit the pairing offer ({status})")
    pending = {"code": code, "rid": rid, "key": _b64(key), "our_token": our_token,
               "name": name, "handle": handle,
               "created": now if isinstance(now, (int, float)) else __import__("time").time()}
    state.pending_save(code, pending)
    return pending


def reap_pending(cfg, relay_client, *, now=None):
    """Drop pending invites AND stashed handshakes older than the TTL and REVOKE the
    deposit token each exposed (an abandoned invite or an un-confirmed handshake
    otherwise leaks a live token that pins our relay mailbox). Returns the number
    reaped. Call before a new invite."""
    import time
    t = now if isinstance(now, (int, float)) else time.time()
    expired = state.pending_gc(t)
    expired_ready = state.ready_gc(t)
    tokens = [p.get("our_token") for p in expired]
    tokens += [(r.get("handshake") or {}).get("our_token") for r in expired_ready]
    tokens = [tok for tok in tokens if tok]
    if not tokens:
        return 0
    _, pub = identity.load_or_create()
    our_fp = identity.fingerprint(pub)
    bearer = state.mailbox_for(_relay_url(cfg)).get("bearer")
    if bearer and relay_client is not None:
        for tok in tokens:
            try:
                _, _, bearer = _owner_op(
                    cfg, relay_client, bearer,
                    lambda b, tok=tok: relay_client.delete_token(our_fp, b, _sha256_hex(tok)),
                    now=now)
            except Exception:  # noqa: BLE001 — best-effort
                pass
    return len(expired) + len(expired_ready)


def wait(vault, cfg, relay_client, pending, *, now=None):
    """Poll the rendezvous ONCE for the answer; when it arrives, return the pairing
    HANDSHAKE (peer + verification words) WITHOUT persisting — the owner then
    confirm()s or abort()s after comparing words. Returns None while still
    unanswered (the CLI loops; tests call once). Raises on an expired rendezvous."""
    rid = pending["rid"]
    key = base64.b64decode(pending["key"])
    status, payload = relay_client.pair_get(rid, "answer")
    if status == 404:
        return None                        # not answered yet (or rendezvous lapsed)
    if status == 429:
        # The relay meters rendezvous reads per IP (30 / 10 min) and every check costs
        # one: too-eager polling is "still waiting", never an error that makes the
        # owner abandon a live invite (RELAY-REVIEW F5). Check once a minute.
        return None
    if status != 200:
        raise PairingError(f"rendezvous read failed ({status})")
    answer = _open_blob(key, relay_client.blob_of(payload))
    if answer.get("role") != "answer":
        raise PairingError("malformed answer blob")
    _, pub = identity.load_or_create()
    our_fp = identity.fingerprint(pub)
    bearer = state.mailbox_for(_relay_url(cfg)).get("bearer")
    hs = _handshake(cfg, our_fp, pub, bearer, answer, pending.get("name"),
                    pending.get("handle"), pending["our_token"],
                    owner_named=bool(pending.get("name")))    # the inviter typed the name
    state.pending_clear(pending["code"])
    return hs


# ---- redeem (side B) --------------------------------------------------------
def redeem(vault, cfg, relay_client, code, *, name=None, handle=None, now=None):
    """Fetch the offer at the code's rendezvous, register our own mailbox + token,
    seal and deposit the answer, and return the pairing HANDSHAKE (peer +
    verification words) WITHOUT persisting — the owner then confirm()s or abort()s
    after comparing words. ``name``/``handle`` are OUR label for the peer (what to call
    them in this vault); either defaults to the name/handle the peer declared for itself."""
    sodium()   # preflight PyNaCl locally BEFORE the rendezvous fetch — never half-run the
    #            ceremony (or burn a single-use code) on a missing local dep (F2)
    rid, key = derive(code)
    status, payload = relay_client.pair_get(rid, "offer")
    if status == 404:
        raise PairingError("no offer at this code — it expired (10 min) or is wrong")
    if status == 429:
        raise PairingError("the relay is metering rendezvous reads from this network "
                           "(30 per 10 minutes) — wait a minute, then try the code again; "
                           "it is still valid")
    if status != 200:
        raise PairingError(f"rendezvous read failed ({status})")
    offer = _open_blob(key, relay_client.blob_of(payload))
    if offer.get("role") != "offer":
        raise PairingError("malformed offer blob")
    our_fp, bearer, pub = ensure_mailbox(cfg, relay_client, now=now)
    our_token = mint_deposit_token()
    status, _, bearer = _owner_op(
        cfg, relay_client, bearer,
        lambda b: relay_client.add_token(our_fp, b, _sha256_hex(our_token)), now=now)
    if status != 201:
        raise PairingError(f"could not register the pairing deposit token ({status})")
    answer = _our_blob("answer", cfg, our_fp, pub, our_token, now)
    status, payload = relay_client.pair_put(rid, "answer", _seal_blob(key, answer))
    if status != 201:
        # The token we just registered is now a token nobody holds: revoke it (best
        # effort) so failed redeems can't pin the mailbox or eat the 64-token cap
        # (RELAY-REVIEW §10 #7).
        try:
            relay_client.delete_token(our_fp, bearer, _sha256_hex(our_token))
        except Exception:  # noqa: BLE001
            pass
    if status == 409:
        # The relay says which (S2): the rid lapsed between our read and our answer,
        # or the answer slot is already taken.
        if (payload or {}).get("error") == "expired":
            raise PairingError("this code expired (10 min) while answering — ask for a "
                               "fresh one")
        raise PairingError("this code was already answered — pairing codes are single-use")
    if status != 201:
        raise PairingError(f"could not deposit the pairing answer ({status})")
    return _handshake(cfg, our_fp, pub, bearer, offer, name or offer.get("name"),
                      handle or offer.get("handle"), our_token,
                      owner_named=name is not None)    # link a People note only if the
    #                    OWNER named this peer — never on the peer's self-declared name (F17)


# ============================================================================
# Resumable, non-blocking pairing sessions (§6 — the friendly walkthrough)
# ============================================================================
# Thin orchestration over the steps above so an AGENT can drive the ceremony one
# call at a time — no step blocks, and the verification-word confirm is an explicit
# later call (confirm_pending / abort_pending), never a terminal y/n prompt. The
# handshake is stashed in the state home (0600) between "show the words" and the
# confirm, so the steps can run in separate processes (a CLI call now, an MCP tool
# next) and on either side of the human word-check. Keyed by the peer's local handle.

def _created_ts(now):
    import time
    return now if isinstance(now, (int, float)) else time.time()


def _pending_by_handle(handle):
    for code, p in state.pending_all().items():
        if isinstance(p, dict) and p.get("handle") == handle:
            return code, p
    return None, None


def start_invite(vault, cfg, relay_client, name, handle=None, *, now=None):
    """Side A, step 1 — NON-BLOCKING. Mint a code + post the offer; return the code to
    relay out of band. The pending invite is persisted (keyed by code, carrying the
    handle) so :func:`poll_invite` can pick up the answer later. Returns
    ``{code, handle, name}``."""
    handle = handle or _slug(name)
    reap_pending(cfg, relay_client, now=now)
    pending = invite(vault, cfg, relay_client, name, handle=handle, now=now)
    return {"code": pending["code"], "handle": handle, "name": name}


def poll_invite(vault, cfg, relay_client, handle, *, now=None):
    """Side A, step 2 — NON-BLOCKING. Check the rendezvous ONCE for the peer's answer.
    Returns ``{status: "waiting"}`` until it arrives, then ``{status: "ready", words,
    peer_name, handle}`` and stashes the handshake for confirm/abort. Idempotent once
    ready (returns the stash again)."""
    stashed = state.ready_get(handle)
    if stashed:
        hs = stashed["handshake"]
        return {"status": "ready", "words": hs["words"], "grant": hs["peer"].get("permissions"),
                "peer_name": hs["peer"].get("name"), "handle": handle}
    code, pending = _pending_by_handle(handle)
    if not pending:
        raise PairingError(f"no pending invite for @{handle} — run an invite first")
    hs = wait(vault, cfg, relay_client, pending, now=now)
    if hs is None:
        return {"status": "waiting", "handle": handle}
    state.ready_save(handle, {"handshake": hs, "name": hs["peer"].get("name"),
                              "handle": handle, "role": "invite", "created": _created_ts(now)})
    return {"status": "ready", "words": hs["words"], "grant": hs["peer"].get("permissions"),
            "peer_name": hs["peer"].get("name"), "handle": handle}


def join(vault, cfg, relay_client, code, *, name=None, handle=None, now=None):
    """Side B — NON-BLOCKING. Redeem a code: post the answer and return the words. The
    handshake is stashed for confirm/abort. Returns ``{status: "ready", words,
    peer_name, handle}``. ``name``/``handle`` are OUR label for the peer; either defaults
    to what the peer declared for itself, so the redeemer can name them at pairing time
    (`confirm`/`abort` then reference the effective handle)."""
    try:
        reap_pending(cfg, relay_client, now=now)   # revoke tokens from abandoned pairings —
    except Exception:  # noqa: BLE001         a join-only brain never calls invite, which
        pass          #                        is the other reap trigger (best-effort)
    hs = redeem(vault, cfg, relay_client, code, name=name, handle=handle, now=now)
    handle = hs["peer"].get("handle") or "peer"
    state.ready_save(handle, {"handshake": hs, "name": hs["peer"].get("name"),
                              "handle": handle, "role": "join", "created": _created_ts(now)})
    return {"status": "ready", "words": hs["words"], "grant": hs["peer"].get("permissions"),
            "peer_name": hs["peer"].get("name"), "handle": handle}


def confirm_pending(vault, handle, *, now=None):
    """Finalize a pairing whose words the owner confirmed match. Loads the stashed
    handshake, persists the peer + fires the pairing ping, clears the stash. Returns
    the persisted peer entry. Raises if nothing is awaiting confirmation for ``handle``."""
    r = state.ready_get(handle)
    if not r:
        raise PairingError(f"no pairing awaiting confirmation for @{handle}")
    peer = confirm(vault, r["handshake"], now=now)
    state.ready_clear(handle)
    return peer


def abort_pending(vault, relay_client, handle):
    """Reject a pairing (verification words did NOT match — a possible MITM). Revoke the
    token we exposed and, on a CONFIRMED revoke, clear the stash. Returns True iff the
    relay confirmed the revoke; on failure the stash is KEPT so the owner can retry."""
    r = state.ready_get(handle)
    if not r:
        raise PairingError(f"no pairing awaiting confirmation for @{handle}")
    try:
        cfg = json.loads(_config_path(vault).read_text(encoding="utf-8"))
    except Exception:  # noqa: BLE001 — no config to re-register against: a 401 stays unconfirmed
        cfg = None
    ok = abort(relay_client, r["handshake"], cfg=cfg)
    if ok:
        state.ready_clear(handle)
    return ok


def pending_pairings(vault):
    """Every in-flight pairing for status/UX — waiting invites + handshakes awaiting the
    word check: ``[{handle, name, role, status, words?}]`` (words present only once the
    answer has arrived)."""
    out = []
    for code, p in state.pending_all().items():
        if isinstance(p, dict):
            out.append({"handle": p.get("handle"), "name": p.get("name"),
                        "role": "invite", "status": "waiting"})
    for handle, r in state.ready_all().items():
        if isinstance(r, dict):
            out.append({"handle": handle, "name": r.get("name"),
                        "role": r.get("role"), "status": "ready",
                        "words": (r.get("handshake") or {}).get("words")})
    return out
