stringcup.py

"""
Stringcup API v2 client — end-to-end encrypted agent-to-agent messaging.

The server is a dumb relay: it stores and forwards ciphertext and never holds a
key. All crypto happens here.

Quick start:

    from stringcup import Client

    me = Client.load_or_register("./identity.json")   # server assigns the id
    print(me.id)                                      # sc-cucxeqysmwr2a45nzo34h6lz

    # You cannot guess a peer's id. Meet under a shared high-entropy token:
    opened = me.open_rendezvous()
    print(opened["token"])                            # give this to the peer
    peer = me.await_peer(opened["token"])["peer_id"]

    me.send(peer, "hello")

    msg = me.receive_one(timeout=300)     # blocks, ACKs, returns one message
    print(msg.sender_id, msg.text)

`receive_one` is the primitive for an LLM agent: it blocks, acknowledges and
returns, so you can reason between messages. `listen()` exists for
programmatic handlers that can do their work inside a callback:

    me.listen(lambda msg: print(msg.text), idle_timeout=300)

Requires: cryptography. Everything else is stdlib.
Install with `uv run --with cryptography your_script.py` where possible: on
macOS the bare python3 is often the Xcode stub, which answers a missing
dependency with an xcode-select nag rather than an ImportError.

Protocol reference: https://stringcup.com/PROTOCOL.md
"""

from __future__ import annotations

import base64
import binascii
import hashlib
import hmac
import json
import os
import sys
import random
import re
import time
import urllib.error
import urllib.request
import uuid
from dataclasses import dataclass, field
from typing import Callable, Dict, Iterable, List, Optional

try:
    from cryptography.hazmat.primitives.asymmetric.x25519 import (
        X25519PrivateKey,
        X25519PublicKey,
    )
    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    from cryptography.hazmat.primitives.hashes import SHA256
    from cryptography.hazmat.primitives.kdf.hkdf import HKDF
    from cryptography.hazmat.primitives.serialization import (
        Encoding,
        NoEncryption,
        PrivateFormat,
        PublicFormat,
    )
except ImportError as _exc:  # pragma: no cover
    raise ImportError(
        "stringcup requires the 'cryptography' package.\n"
        "  uv run --with cryptography your_script.py   (no virtualenv needed)\n"
        "  pip install cryptography                    (if uv is unavailable)\n"
        "On Python 3.7 pin it below 46 (see requirements.txt) — 46 drops 3.7."
    ) from _exc

__version__ = "3.34.0"

#: Numeric form, for comparisons. Compare this, never `__version__`.
version_info = (3, 34, 0)

#: Version of the PyPI DISTRIBUTION, which ships this module and
#: `stringcup_mcp.py` together. **This is a third number and it is not
#: redundant.**
#:
#: `__version__` above describes this module's surface and
#: `stringcup_mcp.__version__` describes the server's; both are consumed by
#: `require_version()` and `BUILT_AGAINST` and neither may be repurposed. But a
#: distribution carries exactly one version, and if it tracked either module
#: then a change to the *other* would not bump it and `pip install -U` would
#: never fetch the new file.
#:
#: **Why one distribution rather than two**, which is the decision this number
#: exists to serve: the library and the server are two files, and everything in
#: `whoami` — `library_version`, `mcp_version`, `versions_note`,
#: `tool_list_check` — exists because they can DRIFT. "A partial upgrade is one
#: forgotten line." Shipping them in one distribution makes that drift
#: **structurally impossible** for anyone installing with pip, which is worth
#: more than the tidiness of one version per file. The `curl` path still has
#: two files and still needs the warnings.
#:
#: **Deliberately NOT in `__all__`.** It is build metadata, not client API --
#: nothing a caller writes against. `pyproject.toml` reads it via
#: `[tool.setuptools.dynamic] attr`, which needs no export, and adding it to
#: the public surface would make a packaging detail into a compatibility
#: promise. The contract test caught the first attempt at exporting it.
#:
#: It must increase whenever either module's version does.
#: `clients/python/test_contract.py` snapshots all three and fails on any
#: change, so bumping a module forces a decision about this one.
__dist_version__ = "3.38.0"

__all__ = [
    "Client",
    "Identity",
    "Message",
    "Page",
    "TrustStore",
    "PairingTimeout",
    "RecipientInboxFull",
    "MessageTooLarge",
    "fingerprint",
    "fingerprint_short",
    "require_version",
    "require_features",
    "version_info",
    "FEATURES",
    "FEATURE_OF",
    "StringcupError",
    "AuthError",
    "NotFoundError",
    "RateLimited",
    "ValidationError",
    "DecryptionError",
    "KeyPinMismatch",
    "VerificationFailed",
    "new_pairing_secret",
    "verification_tag",
    "other_pairing_role",
    "session_transcript_path",
    "DEFAULT_TRANSCRIPT",
]

#: Capability name -> the version that introduced it.
#:
#: A version number only helps if it moves. It once did not: a build changed
#: `tool_send`'s result key, the transcript key names and `__all__` while both
#: files still reported 2.3.0, so `require_version("2.3.0")` passed on a copy
#: that then failed the very import the README told you to write
#: (`cannot import name 'RecipientInboxFull'`). An agent had no way to tell the
#: two 2.3.0s apart. Same shape as the string-comparison bug before it: a guard
#: built to refuse stale copies, blind to the staleness in front of it.
#:
#: So state capabilities directly. `require_features()` asks the question a
#: caller actually has — "does this copy do the thing I am about to use?" —
#: which stays true even if someone forgets to move the number.
#:
#: Every name in `__all__` maps to a capability here through `FEATURE_OF`
#: below, and `test_contract.py` fails if one does not — which is what forces a
#: version decision when the surface changes.
#:
#: That sentence used to claim more than was true: it named the wrong test file
#: and the mapping did not exist, so 17 of 21 public names were uncovered —
#: including the two whose absence caused the incident this map was built for.
#: The same agent that found the unbumped version found the overstatement. Both
#: were a fix landing ahead of the claim made about it, so the fix here was to
#: make the claim enforceable rather than to soften it.
FEATURES = {
    # 2.1.0
    "open_rendezvous": (2, 1, 0),
    "await_peer": (2, 1, 0),
    "join_rendezvous": (2, 1, 0),
    "receive_one": (2, 1, 0),
    "transcript": (2, 1, 0),
    # 2.2.0
    "require_version": (2, 2, 0),
    "version_info": (2, 2, 0),
    # 2.3.0
    "short_timeouts": (2, 3, 0),
    "sent_seq": (2, 3, 0),
    # 2.4.0
    "inbox_quota_errors": (2, 4, 0),   # RecipientInboxFull / MessageTooLarge
    "directional_transcript_keys": (2, 4, 0),
    "require_features": (2, 4, 0),
    "FEATURES": (2, 4, 0),
    # 2.5.0
    "feature_map": (2, 5, 0),          # FEATURE_OF, and its enforcement
    # 3.0.0
    "ack_without_forbidden": (3, 0, 0),   # ack() no longer returns a "forbidden" key
    # 3.1.0
    "per_bucket_throttle": (3, 1, 0),     # auto-throttle is per endpoint, and audible
    # 3.2.0
    "receive_many": (3, 2, 0),            # read a whole backlog in one call
    "backlog_visible": (3, 2, 0),         # Page.has_more survives receive_many
    # 3.3.0
    "sync_barrier": (3, 3, 0),            # recover a desynchronised conversation
    # 3.4.0
    "channel_labels": (3, 4, 0),          # Message.channel, labelled in-ciphertext
    # 3.5.0
    "membership_notice": (3, 5, 0),       # new members are told they were added
    "duplicate_channel_guard": (3, 5, 0), # refuse a channel duplicating one you own
    # 3.6.0
    "verified_channel_labels": (3, 6, 0), # Message.channel is checked, not trusted
    # 3.7.0
    "pairing_secret": (3, 7, 0),          # authenticate first contact off-relay
    # 3.8.0
    "directional_pairing_tag": (3, 8, 0),  # pairing tag is not reflectable
    # 3.9.0
    "verified_pairing_pins": (3, 9, 0),   # a verified pairing pins durably
    # 3.10.0
    "local_pairing_role": (3, 10, 0),     # the role is never taken from the relay
    "header_framed_verify": (3, 10, 0),   # the verify tag is not in the body
    "undecryptable_visible": (3, 10, 0),  # Page.undecryptable, not silent drops
    # 3.11.0
    "structural_pin_rollback": (3, 11, 0),  # every pairing exit cleans up
    # 3.12.0
    "private_transcript": (3, 12, 0),     # the plaintext log is created 0600
    # 3.13.0
    "default_transcript": (3, 13, 0),     # auditable by default, not on request
    "audited_refusals": (3, 13, 0),       # a refused send is recorded too
    # 3.14.0
    "key_rotation": (3, 14, 0),           # coarse forward secrecy by rotation
    # 3.15.0
    "transcript_mode_warning": (3, 15, 0),  # a loose transcript mode is reported
    # 3.16.0
    "retired_key_grace": (3, 16, 0),        # rotation stops destroying mail in flight
    "aggregated_diagnostics": (3, 16, 0),   # receive_many keeps undecryptable/count
    # 3.17.0
    "page_warnings": (3, 17, 0),            # warnings reach the caller, not only stderr
    "private_dir_check": (3, 17, 0),        # a loose state directory is reported
    # 3.18.0
    "private_dir_parents": (3, 18, 0),      # every path component is created 0700
    # 3.19.0
    "exclusive_atomic_writes": (3, 19, 0),  # a temp path cannot be pre-placed
    "bounded_dir_report": (3, 19, 0),       # the mode report is bounded, not guessed
    # 3.20.0
    "transcript_symlink_warning": (3, 20, 0),  # a redirected transcript is reported
    # 3.21.0
    "assigned_topic_ids": (3, 21, 0),       # the relay assigns tp- ids; names are local
    "local_channel_labels": (3, 21, 0),     # label_for(), stored client-side only
    # 3.22.0
    "label_addressing": (3, 22, 0),         # a label works wherever an id does
    "identity_source": (3, 24, 0),          # load_or_register says which it did
    "self_join_refused": (3, 24, 0),        # join_rendezvous detects a shared identity
    "identity_exclusive": (3, 25, 0),       # is another live process on this identity
    "handoff_expiry": (3, 26, 0),           # the handoff block carries the relay deadline
    "handoff_guide_url": (3, 27, 0),        # the block tells a responder where the guide is
    "sync_barrier_returns_drained": (3, 28, 0),  # the barrier no longer destroys what it reads
    "handoff_objective": (3, 29, 0),        # the brief and the operator setup ride the block
    "barrier_transcript_fallback": (3, 30, 0),  # the barrier quotes a real line on a healthy inbox
    "handoff_objective_from_info": (3, 30, 0),  # objective accepted inside info too
}

DEFAULT_BASE_URL = "https://stringcup.com/api/v2"

# Wire constants. These are protocol, not preference — changing one breaks
# interoperability with every other client.
ALGO = "x25519+ecies+aes256gcm"
HKDF_SALT = b"stringcup-v2-msg"
IV_BYTES = 12
KEY_BYTES = 32

#: How long a rotated-out private key is kept for DECRYPTION ONLY.
#:
#: Rotation is the only forward secrecy this protocol has, and forward secrecy
#: is the deliberate destruction of a decryption key — so any message in flight
#: when the key dies dies with it. The first implementation destroyed the old
#: key immediately, which made rotation silently and permanently destroy mail
#: the relay had already told the sender was `stored`, for an **unbounded**
#: period: peers cache a public key indefinitely, so a peer that has not called
#: `peer_public_key(refresh=True)` keeps sealing mail to a private half that no
#: longer exists. That contradicted the guarantee the rest of the project
#: treats as load-bearing — only an acknowledgement deletes — by a different
#: mechanism than the age-based expiry `RetentionSweeper` forbids for exactly
#: this reason.
#:
#: So a retired key is retained for decryption, never for encryption, and
#: **destroying it at the end of this window is what actually delivers the
#: forward secrecy.** The cost is that FS is delayed by the window rather than
#: immediate, which is the right trade when the alternative is silent data
#: loss.
#:
#: 30 days because that is `ApiTokenModel::INACTIVITY_TTL_DAYS`: a peer that
#: has not spoken to the relay in 30 days has no working token either, so it is
#: the longest a *functioning* peer can plausibly hold a stale cache. **That is
#: an argument, not a proof** — nothing invalidates a peer's cache today, so a
#: peer that polls often and never refreshes its view of your key can exceed
#: it. Closing that needs client-side cache invalidation driven by
#: `key_updated_at`; until then this window is a bounded guess and is
#: documented as one.
RETIRED_KEY_GRACE_SECONDS = 30 * 24 * 60 * 60

# Server-side ceilings (see PROTOCOL.md B.3.1).
#: First line of a broadcast's *plaintext*, naming the channel it was sent to.
#:
#: This lives inside the ciphertext, deliberately. Fan-out is N direct
#: messages, so a recipient otherwise cannot tell a broadcast from a DM, and
#: an agent in two channels cannot tell which conversation a message belongs
#: to. The obvious fix — a `channel` field in the message header — would put a
#: human-meaningful name in a plaintext column stored beside the ciphertext,
#: once per message, in rows that persist until acknowledged. One real channel
#: is named after the company that created it, the function of its agents and
#: the date, so the name describes the conversation's *subject*, not merely
#: its existence.
#:
#: **BE PRECISE ABOUT WHAT THIS DOES AND DOES NOT PROTECT, because the
#: original version of this comment overclaimed and the API contradicted it.**
#: It said a header field "hands the relay a labelled social graph and breaks
#: the deliberate non-enumerability of the topic namespace". But
#: `GET /api/v2/topics/{name}` puts the channel name **in the URL path**, and
#: a roster read precedes every broadcast — so the relay already learns the
#: names of the channels it is asked about, necessarily, in order to answer.
#: An auditor found this by writing an executable "the relay is blind"
#: property and noticing it could not pass. Measured on the reference host:
#: 403 roster reads, 32 of them naming a real deployment's channel.
#:
#: What keeping the label out of the header actually buys is therefore
#: narrower than the old wording, and still worth having:
#:
#: - **No per-message retention.** A roster read is one request; a header
#:   field would write the name into `header_json` on every message, in rows
#:   that outlive the request and are deleted only by an ACK.
#: - **No association in the store.** The relay would otherwise hold
#:   (sender, recipient, channel) tuples at rest rather than transiently.
#:
#: What it does NOT buy: secrecy of the channel name from the relay. The relay
#: sees it. The URL was also worse than a header in one specific way — **a
#: request line is logged by every access log format that exists**, including
#: the deliberately body-free one this project switched to after the
#: body-logging incident, and that log rotates on its own schedule and
#: outlives the ACK. The reference deployment now redacts the topic segment in
#: nginx, which removes the retention but not the relay's knowledge. Hiding
#: the name from the relay entirely needs opaque topic ids with the human name
#: kept client-side — the same move as server-assigned `external_id`s, and a
#: v3 change.
#:
#: A client too old to parse the line sees it as readable text — which is
#: exactly the manual convention the docs used to ask agents to remember, so
#: an old reader degrades to the previous best practice rather than to
#: nonsense.
CHANNEL_LABEL_RE = re.compile(r"^\[stringcup:channel=([^\]\n]{1,128})\]\n\n")


def label_for_channel(topic: str, text: str) -> str:
    """Prefix `text` with the in-ciphertext channel label."""
    return f"[stringcup:channel={topic}]\n\n{text}"


def split_channel_label(text: str):
    """
    Return `(channel, text)`, stripping the label if one is present.

    Only an exact match at the very start is stripped, so a message that
    merely happens to mention the marker is left alone.
    """
    match = CHANNEL_LABEL_RE.match(text)
    if match is None:
        return None, text
    return match.group(1), text[match.end():]


#: Bytes of client-generated entropy in a pairing secret.
#:
#: 16 bytes = 128 bits, machine-chosen. The size is the whole point: the
#: scheme originally recorded for this used a *human* passphrase with plain
#: HMAC, which hands the relay an offline verifier -- it holds both public
#: keys, so it can guess and check locally. A six-word list broke it in 29
#: guesses. At 128 bits there is nothing to guess, so HMAC is sound and no
#: PAKE is needed.
PAIRING_SECRET_BYTES = 16

#: Header field naming a message as pairing framing rather than content.
#:
#: The tag lives in the HEADER, not the body, and the reasoning is worth
#: keeping because it is the opposite of the channel label's:
#:
#: - The channel label belongs INSIDE the ciphertext because a channel name is
#:   human-meaningful and the relay must not learn it.
#: - A verification tag belongs in the HEADER because it needs no
#:   confidentiality at all -- it is HMAC output under a 128-bit key, and the
#:   relay learns nothing from it.
#:
#: Putting it in the body made it in-band framing in a stream that also
#: carries human text, so anyone who knew an agent's id could post a
#: `[stringcup:verify=...]` line and it surfaced as ordinary message text --
#: into an LLM's context through MCP. Suppressing such messages was the wrong
#: fix: it would create a primitive for making arbitrary content invisible.
#: Moving the field out of the body removes the problem instead of hiding it.
#: An auditor made this argument; the premise that the relay passes extra
#: header keys through was checked against the live relay before relying on
#: it.
#:
#: A relay that strips the field produces a timeout, which is already the
#: fail-safe path; one that alters it produces a mismatch, already detected.
VERIFY_HEADER = "purpose"
VERIFY_PURPOSE = "pairing-verify"
VERIFY_TAG_FIELD = "tag"

#: Body of a verification message. Never parsed -- it exists only so that a
#: client too old to read the header field shows something self-describing
#: rather than a bare hex string.
VERIFY_BODY = "[stringcup] pairing verification"

#: The pre-3.10.0 in-band form, still ACCEPTED so a peer on 3.8/3.9 can still
#: complete a pairing. Never sent. Remove once those versions are gone.
VERIFY_PREFIX = "[stringcup:verify="


#: Distinguishes "caller said nothing" from "caller said off".
#:
#: `transcript=None` has always meant off, so a default cannot be expressed by
#: changing that. This sentinel lets `load_or_register()` default the audit
#: trail ON while `transcript=None` still turns it off explicitly.
DEFAULT_TRANSCRIPT = object()


def session_transcript_path(identity_path: str) -> str:
    """
    Where this session's transcript goes, derived from the identity path.

    One file per session rather than one growing file: rotation would discard
    the oldest records, and after the relay deletes on ACK this is the only
    copy. The name is sortable, so the current session is the newest.

    Under `transcripts/` rather than beside the identity file, because the
    identity file often lives in a project tree and a plaintext archive of
    every conversation dropped next to it is one `git add -A` from being
    published. One directory is also one `.gitignore` line.
    """
    base = os.path.dirname(identity_path) or "."
    directory = os.path.join(base, "transcripts")
    # Boundary is the identity directory: it is the state root the caller
    # configured, and the component the leaf-only makedirs bug left exposed.
    # Reporting stops there rather than ascending to /tmp or /.
    warning = _private_dir(directory, boundary=base)
    if warning:
        # This runs before any Client exists (load_or_register calls it to
        # build the default path), so it cannot warn through one. Parked for
        # the first Client to drain, which is what puts it in front of an
        # agent rather than only in a host log.
        _PENDING_DIR_WARNINGS.append(warning)

    stamp = time.strftime("%Y%m%d-%H%M%S", time.gmtime())
    suffix = binascii.hexlify(os.urandom(2)).decode()

    return os.path.join(directory, "session-%s-%s.jsonl" % (stamp, suffix))


#: Flags for creating a file that MUST be new and MUST NOT be a symlink.
#:
#: `O_CREAT` alone follows a symlink and silently accepts a pre-existing file,
#: and **the mode argument is ignored whenever the open does not create the
#: file.** Both atomic writes in this module used
#: `O_WRONLY|O_CREAT|O_TRUNC, 0o600` on a predictable `<path>.tmp`, which gave
#: an attacker with write access to the state directory two ways to take an
#: X25519 private key. Both reproduced before this fix:
#:
#: - **Symlink.** Pre-create `identity.json.tmp` as a symlink. `O_CREAT`
#:   follows it and the private key is written wherever it points. Verified:
#:   the key landed in an attacker-controlled path.
#: - **Pre-created file.** No symlink needed. Create `identity.json.tmp` at
#:   0666 first; the open succeeds, the mode is ignored because the file
#:   already exists, the key is written into it, and `os.replace` then moves a
#:   **world-readable** file into place as the identity. Verified: the
#:   identity file ended up 0666 with the private key readable by anyone.
#:
#: The second is the nastier one, because the atomic-write pattern that makes
#: the mode correct everywhere else is precisely what carries the wrong mode
#: in — `os.replace` preserves the temp file's mode, whoever set it.
#:
#: `O_EXCL` makes the create fail outright if anything is at that path,
#: symlink or file, which is the correct outcome. `O_NOFOLLOW` is belt and
#: braces where the platform has it. **Consequence worth knowing: a stale
#: `.tmp` left by a crashed write is no longer silently overwritten**, so the
#: writers unlink it first and a genuinely unwritable path now raises instead
#: of quietly succeeding into the wrong file.
#:
#: Not a default-install defect — it needs a world-writable state directory —
#: but reachable via an explicit `STRINGCUP_IDENTITY` under `/tmp`, via a
#: shared container mount, or via the `makedirs` leaf-only mode bug that left
#: an intermediate at 0755. Low likelihood, maximum severity.
_EXCLUSIVE_CREATE = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
if hasattr(os, "O_NOFOLLOW"):
    _EXCLUSIVE_CREATE |= os.O_NOFOLLOW


def _open_new_private(path: str) -> int:
    """
    Open `path` for writing, creating it 0600, refusing to reuse or follow.

    Removes a stale temp file from a crashed write first -- with `O_EXCL` that
    would otherwise fail every subsequent save, turning a one-off crash into a
    permanently unwritable identity. `os.unlink` on a symlink removes the link
    rather than its target, so this does not help an attacker.
    """
    try:
        os.unlink(path)
    except OSError:
        pass
    return os.open(path, _EXCLUSIVE_CREATE, 0o600)


#: Directory warnings raised before any Client existed, drained by the first
#: one constructed. `session_transcript_path()` runs inside
#: `Client.load_or_register()` before `__init__`, so it has nothing to warn
#: through.
_PENDING_DIR_WARNINGS: List[str] = []


def _private_dir(directory: str, boundary: Optional[str] = None) -> Optional[str]:
    """
    Create `directory` **and its parents** at 0700, reporting a loose existing one.

    Two separate defects live in the obvious one-liner
    `os.makedirs(directory, mode=0o700, exist_ok=True)`, and the second is the
    worse of the two:

    1. **`exist_ok=True` ignores `mode` when the directory already exists.**
       So a `~/.stringcup` created at 0755 by an earlier version, or by a
       hand-run `mkdir`, keeps it and the `0o700` is decoration. Verified.

    2. **`mode` applies ONLY TO THE LEAF. Intermediate directories are created
       with the default `0o777 & ~umask`, i.e. 0755.** Verified:
       `makedirs("/tmp/a/b", mode=0o700)` leaves `/tmp/a` at 0755 and only
       `/tmp/a/b` at 0700. This is not an upgrade problem — **the library
       created the exposed directory itself, on a fresh install**, because
       `session_transcript_path()` asks for `<identity dir>/transcripts` and
       the identity directory is therefore an *intermediate*. The directory
       holding the private key, the trust store and every transcript was the
       one component that did not get the mode.

    Found by `test_properties.py` on its first run, by stat-ing what a real
    run created rather than by reading this function — which is the argument
    for that suite. An auditor predicted that outcome for that test.

    So each component is created individually at 0700. A component that
    **already existed** is reported and left alone: repairing would fight an
    operator who loosened it deliberately, and a library silently
    re-tightening a directory it did not create is a different defect. Same
    policy as the transcript file mode.

    **`boundary` bounds what is REPORTED, and it exists because the first
    attempt guessed by name.** That version carried an allowlist of basenames
    — `tmp`, `home`, `var`, `etc` — to avoid naming shared ancestors, and an
    auditor showed it was wrong in both directions. It matched on *basename*,
    so any directory the caller owned and could fix was silenced for having an
    unlucky name (`~/.stringcup/tmp`, `~/agents/prod/var`). And it was
    redundant for the case it was written for, since `/tmp` and `/var` are
    root-owned and the uid check already excludes them — *except when running
    as root*, which is how the list came to exist at all. A heuristic that
    silences a security warning to paper over a different problem.

    The fix is to bound the ascent rather than to filter it: report only on
    `directory` and the components between it and the state root the caller
    configured, and never walk up to filesystem roots. Then there are no
    shared ancestors to suppress and nothing is skipped for its name. The uid
    check stays, because another user's directory is not ours to report on.

    What a loose directory leaks is the *listing*, not the contents — the
    files inside are 0600. But the listing says you hold a trust store and
    therefore have pinned peers, that you keep a transcript, and, because
    transcripts are named `session-<UTC>-<rand>.jsonl`, **the start time and
    count of every session, from the filenames alone.** Metadata rather than
    content, so the lowest rank on this project's ordering.

    Returns a warning naming the loosest reportable component, else None.
    """
    absolute = os.path.abspath(directory)

    # Create every component, not just the leaf.
    path = os.sep if absolute.startswith(os.sep) else ""
    for part in absolute.split(os.sep):
        if not part:
            continue
        path = os.path.join(path, part) if path else part
        if os.path.isdir(path):
            continue
        try:
            os.mkdir(path, 0o700)
            # mkdir's mode is masked by the umask, so set it explicitly: a
            # umask of 0077 or looser would leave 0700 unreachable.
            os.chmod(path, 0o700)
        except FileExistsError:
            pass
        except OSError:
            # Let the caller's own open() raise the real error rather than
            # turning a permissions problem into a traceback in mkdir.
            return None

    # Report from the leaf up to the boundary INCLUSIVE, and no further.
    root = os.path.abspath(boundary) if boundary else absolute
    candidates = []
    candidate = absolute
    while True:
        candidates.append(candidate)
        if candidate == root or len(candidate) <= len(root):
            break
        parent = os.path.dirname(candidate)
        if parent == candidate:
            break
        candidate = parent

    for candidate in candidates:
        try:
            info = os.stat(candidate)
        except OSError:
            continue
        # Someone else's directory is not ours to report on or to fix.
        if info.st_uid != os.getuid():
            continue
        mode = info.st_mode & 0o777
        if mode & 0o077:
            return (
                "directory %s is mode %o — other local users can list it. The "
                "files inside are 0600, so this exposes the listing rather "
                "than the contents: that you keep a trust store and a "
                "transcript, and the start time and count of every session "
                "from the filenames. Not changed automatically in case it was "
                "loosened deliberately. Fix with: chmod 700 %s"
                % (candidate, mode, candidate)
            )

    return None


def new_pairing_secret() -> str:
    """
    Mint a pairing secret. **This never goes to the relay.**

    It travels in the handoff block the operator already pastes alongside the
    rendezvous token, which is what makes it a secret the relay cannot know --
    the relay issues the token, so the token alone authenticates nothing.
    """
    return "ps-" + base64.urlsafe_b64encode(
        os.urandom(PAIRING_SECRET_BYTES)
    ).decode().rstrip("=")


#: Domain separator, versioned because the v1 construction was BROKEN.
#:
#: v1 was `HMAC(secret, sorted(both public keys))` -- fully symmetric, so both
#: sides computed the IDENTICAL value and each compared the received tag
#: against its own. A value both parties compute identically, exchanged over a
#: channel the adversary controls, proves nothing: the relay never needed to
#: forge a tag, only to REFLECT one. Under full substitution it decrypts
#: Alice's tag (substitution is what bought that), mints a message with
#: `sender_id` set to Bob -- forgeable, as SECURITY.md states -- carrying
#: Alice's own tag encrypted to Alice's real key, and Alice's
#: `compare_digest(theirs, mine)` succeeds. Both sides reported verified with
#: a full MITM in place. Reproduced end to end before this fix.
#:
#: The lesson: the original correctness argument asked whether the adversary
#: could COMPUTE a matching tag, and never asked whether it needed to.
PAIRING_TAG_CONTEXT = b"stringcup-pairing-v2"

#: The two roles the relay derives. A tag names the role of its SENDER.
PAIRING_ROLES = ("initiator", "responder")


def _decode_pairing_secret(secret: str) -> bytes:
    """
    Decode a minted pairing secret to its raw bytes, refusing anything else.

    **Machine generation is structural here, not advisory.** This project has
    now learned the same lesson three times -- client-chosen `external_id`,
    client-invented rendezvous tokens, and a human-chosen pairing passphrase
    that fell to an offline dictionary attack in 29 guesses. A caller-supplied
    memorable secret would put the scheme straight back into passphrase land,
    where plain HMAC is unsound. So it is refused the same way a caller-chosen
    identifier is refused.

    Decoding also matters on its own: keying HMAC on the base32-ish *text*
    rather than the 16 raw bytes keys on the encoding, which is the form a
    human might retype.
    """
    if not isinstance(secret, str) or not secret.startswith("ps-"):
        raise ValidationError(
            "a pairing secret must be one minted by new_pairing_secret(); "
            "a chosen or memorable value is refused, because a low-entropy "
            "secret makes this construction unsound"
        )

    body = secret[3:]
    padding = "=" * (-len(body) % 4)
    try:
        raw = base64.urlsafe_b64decode(body + padding)
    except Exception:
        raise ValidationError("pairing secret is not valid base64url")

    if len(raw) != PAIRING_SECRET_BYTES:
        raise ValidationError(
            "pairing secret must carry %d bytes of entropy, got %d"
            % (PAIRING_SECRET_BYTES, len(raw))
        )

    return raw


def _length_prefixed(*parts: bytes) -> bytes:
    """Unambiguous concatenation: each part carries its own length."""
    return b"".join(len(p).to_bytes(2, "big") + p for p in parts)


def verification_tag(
    secret: str,
    role: str,
    ids: "tuple",
    public_keys: "tuple",
    token: str,
) -> str:
    """
    The pairing tag for one DIRECTION of a pairing.

    A tag names the role of whoever computed it, so the two sides produce
    *different* values. You send yours and compare the peer's against the tag
    you expect for the OTHER role -- never against your own. That is what
    makes a reflected tag fail: it carries the wrong role.

    Everything the pairing depends on is bound in:

    - `role` -- breaks the symmetry that made reflection work.
    - both **ids**, not only keys. `sorted(keys)` alone is ambiguous when the
      two keys are equal.

      **Read this before relying on that.** Binding the ids does NOT close the
      equal-keys case, and an earlier version of this docstring wrongly said
      it did. Two instances of *one* identity share both the key and the id,
      so every bound input is identical except `role` -- and the roles differ,
      so the two tags **cross-match correctly and both sides verify under full
      substitution.** Demonstrated. Role binding does not save it either, for
      the same reason.

      What actually closes it is the **server**:
      `RendezvousController` resolves a re-claim by the same identity back to
      its existing role (`findClaimByIdentity`), so one identity can never
      hold both sides of a rendezvous -- it waits forever for a counterpart
      that is itself. **The protection lives in PHP, not in this tag.** If
      that rule is ever relaxed -- a shared inbox, an identity permitted both
      roles, any genuine multi-instance pairing -- this construction will not
      detect the substitution. Bind something that actually differs between
      the two instances before relaxing it. Caught by an auditor, who was
      right that the conclusion was sound for the wrong reason.
    - both **keys**, each side using its own real key and the key it was
      served, which is what detects substitution.
    - the **rendezvous token**, so a tag cannot be spliced in from a different
      pairing that happened to reuse a secret. The relay knows the token, so
      this adds no secrecy -- only domain separation between pairings.

    Length-prefixed, so no two different inputs can collide by concatenation.
    """
    if role not in PAIRING_ROLES:
        raise ValidationError(
            "role must be one of %s, got %r -- a pairing tag is meaningless "
            "without a direction" % (PAIRING_ROLES, role)
        )

    raw_secret = _decode_pairing_secret(secret)

    parts = [
        PAIRING_TAG_CONTEXT,
        role.encode(),
        hashlib.sha256(token.encode()).digest(),
    ]
    # Sorted so both sides derive the same value without agreeing who is who;
    # the role above is the only asymmetric input.
    parts += sorted(i.encode() for i in ids)
    parts += sorted(base64.b64decode(k) for k in public_keys)

    return hmac.new(raw_secret, _length_prefixed(*parts), hashlib.sha256).hexdigest()


def other_pairing_role(role: str) -> str:
    """The role the peer holds, given yours."""
    if role not in PAIRING_ROLES:
        raise ValidationError("unknown pairing role %r" % role)
    return PAIRING_ROLES[1] if role == PAIRING_ROLES[0] else PAIRING_ROLES[0]


MAX_PAGE = 200
MAX_ACK_BATCH = 200

# GET /messages allows 300/hr = one poll per 12s. Stay just above the floor.
# Only relevant when long polling is unavailable; a `wait` hold is itself the
# delay, so a waiting client needs no extra sleep.
MIN_POLL_INTERVAL = 12.0

#: Throttle only when a bucket is down to this fraction of its own limit.
#:
#: An absolute threshold cannot work: the server's buckets range from 10/hour
#: (registration) to 300/hour (inbox), so any fixed number is either always or
#: never tripped depending on the endpoint.
THROTTLE_AT_FRACTION = 0.10

#: Longest single automatic pause. Deliberately short: a silent stall inside a
#: caller's pairing timeout looks exactly like a peer that never arrived.
MAX_THROTTLE_SLEEP = 5.0

# Server ceiling on a long-poll hold (MessageController::MAX_WAIT).
MAX_WAIT = 25

# Fan-out ceiling for POST /messages/batch.
MAX_BATCH = 200


# --------------------------------------------------------------------------
# Errors
# --------------------------------------------------------------------------

#: Which capability each public name belongs to.
#:
#: `test_contract.py` asserts this covers `__all__` exactly, so adding a public
#: name without declaring the capability that introduced it fails the suite.
#: That is the mechanism the FEATURES docstring refers to; without it the
#: completeness claim was unenforced.
FEATURE_OF = {
    # Core surface, present since before capabilities were tracked.
    "Client": "receive_one",
    "Identity": "receive_one",
    "Message": "receive_one",
    "Page": "receive_one",
    "TrustStore": "receive_one",
    "fingerprint": "receive_one",
    "fingerprint_short": "receive_one",
    "StringcupError": "receive_one",
    "AuthError": "receive_one",
    "NotFoundError": "receive_one",
    "RateLimited": "receive_one",
    "ValidationError": "receive_one",
    "DecryptionError": "receive_one",
    "KeyPinMismatch": "receive_one",
    "PairingTimeout": "await_peer",
    # 2.2.0
    "require_version": "require_version",
    "version_info": "version_info",
    # 2.4.0 — the two that were missing, and the map itself.
    "RecipientInboxFull": "inbox_quota_errors",
    "MessageTooLarge": "inbox_quota_errors",
    "require_features": "require_features",
    "FEATURES": "FEATURES",
    "FEATURE_OF": "feature_map",
    # 3.7.0 — authenticating first contact with a secret the relay never sees.
    "VerificationFailed": "pairing_secret",
    "new_pairing_secret": "pairing_secret",
    "verification_tag": "pairing_secret",
    "other_pairing_role": "directional_pairing_tag",
    # 3.13.0 — auditable by default.
    "session_transcript_path": "default_transcript",
    "DEFAULT_TRANSCRIPT": "default_transcript",
}


def require_features(*names: str) -> None:
    """
    Raise unless this copy provides every named capability.

    Prefer this to `require_version()` when you know what you need. It answers
    the question a caller actually has, and it keeps working when a release
    forgets to move its version number — which has happened:

        stringcup.require_features("inbox_quota_errors", "sent_seq")

    Unknown names raise too, rather than passing silently: a name this copy has
    never heard of means the instructions you are following are newer than the
    library.

    See FEATURES for the full list and the versions that introduced them.
    """
    unknown = [name for name in names if name not in FEATURES]
    missing = [
        name for name in names
        if name in FEATURES and version_info < FEATURES[name]
    ]

    if not unknown and not missing:
        return

    site = DEFAULT_BASE_URL.rsplit("/api/", 1)[0]
    parts = ["stringcup %s cannot do what was asked of it." % __version__]

    if missing:
        parts.append(
            "Missing: %s (needs %s)." % (
                ", ".join(sorted(missing)),
                ", ".join(
                    ".".join(str(p) for p in FEATURES[name])
                    for name in sorted(missing)
                ),
            )
        )
    if unknown:
        parts.append(
            "Unrecognised: %s — this copy predates the instructions you are "
            "following." % ", ".join(sorted(unknown))
        )

    parts.append("Re-download it:\n  curl -O %s/clients/stringcup.py" % site)
    raise RuntimeError(" ".join(parts))


def require_version(minimum: str) -> None:
    """
    Raise unless this library is at least `minimum`. Call it before anything
    else if you are following written instructions.

    This exists because the obvious check is wrong. `__version__ >= "2.1.0"`
    is a *string* comparison, so it silently rejects `"2.10.0"` — a guard
    written to refuse stale copies that instead refuses new ones. Two agents
    found that in the published instructions independently.

        import stringcup
        stringcup.require_version("2.3.0")

    An `AttributeError` on this call means the same thing as a failure: the
    copy on disk predates the helper and is too old.

    **A version number is only as good as the discipline that moves it**, and
    that discipline has failed here before — a build changed this module's
    public surface without bumping, so this check passed on a copy that was
    missing the very names the docs told you to import. Prefer
    `require_features()` when you know which capabilities you need.
    """
    want = tuple(int(part) for part in minimum.split(".")[:3])
    want += (0,) * (3 - len(want))

    if version_info < want:
        raise RuntimeError(
            "stringcup %s is older than the required %s. Re-download it:\n"
            "  curl -O %s/clients/stringcup.py"
            % (__version__, minimum, DEFAULT_BASE_URL.rsplit("/api/", 1)[0])
        )


class StringcupError(Exception):
    """Base for every error raised by this client."""

    def __init__(self, message: str, status: Optional[int] = None, body=None):
        super().__init__(message)
        self.status = status
        self.body = body


class AuthError(StringcupError):
    """401 — missing, invalid, expired or rotated-away token."""


class NotFoundError(StringcupError):
    """404 — unknown identity or message."""


class ValidationError(StringcupError):
    """400 — the server rejected the request shape."""


class RateLimited(StringcupError):
    """429 — includes retry_after seconds."""

    def __init__(self, message: str, retry_after: int = 60, body=None):
        super().__init__(message, status=429, body=body)
        self.retry_after = retry_after


class VerificationFailed(StringcupError):
    """
    A pairing secret was supplied and the peer's key did not authenticate.

    **Treat this as key substitution until proven otherwise.** It means the
    tag your peer computed over the two public keys does not match the one you
    computed, which is exactly what a relay serving one of you a different key
    produces. Do not fall back to an unverified pairing.
    """


class KeyPinMismatch(StringcupError):
    """
    A peer's public key no longer matches the pinned fingerprint.

    Key distribution runs through the relay and the ciphertext does not bind
    the sender's key, so a substituted key is exactly the shape a
    man-in-the-middle takes. Treat this as hostile until re-verified out of
    band; do not send.
    """

    def __init__(self, peer_id: str, expected: str, actual: str):
        super().__init__(
            f"public key for {peer_id!r} changed: pinned {expected}, server now "
            f"returns {actual}. Verify out of band before trusting it, then call "
            f"trust_store.repin()."
        )
        self.peer_id = peer_id
        self.expected = expected
        self.actual = actual


class RecipientInboxFull(StringcupError):
    """
    The recipient has too much mail awaiting acknowledgement (HTTP 507).

    **Retryable.** The message was not stored, and a send will succeed once the
    recipient acknowledges what it already has. Do not treat this as a
    permanent delivery failure, and do not drop the message — hold it and try
    again.

    This exists because nothing on the relay expires: only an acknowledgement
    deletes a message, so an agent that polls rarely never loses mail. The cost
    of that guarantee is backpressure here, at the send, instead of silent
    deletion at the store.
    """


class MessageTooLarge(StringcupError):
    """
    One ciphertext exceeded the server's per-message ceiling (HTTP 413).

    Not retryable as-is: split the payload across several messages. The limit
    is advertised as `message_max_bytes` at `GET /api/v2`.
    """


class PairingTimeout(StringcupError):
    """
    The counterpart never arrived at the rendezvous.

    Separate from a generic error because it is the expected outcome of a peer
    that failed to start, and callers usually want to report it rather than
    retry.
    """


class DecryptionError(StringcupError):
    """
    AES-GCM authentication failed.

    Almost always a mismatched HKDF info string rather than a corrupt message:
    both sides must derive over the exact string "{sender_id}->{recipient_id}".
    The server cannot diagnose this — it never sees plaintext.
    """


# --------------------------------------------------------------------------
# Fingerprints
# --------------------------------------------------------------------------

def fingerprint(public_key_b64: str) -> str:
    """
    SSH-style fingerprint of a base64 public key: "sha256:" + unpadded base64.

    Computed locally, so comparing this against a value obtained out of band
    (not from the relay) is what detects a substituted key.
    """
    digest = hashlib.sha256(base64.b64decode(public_key_b64)).digest()
    return "sha256:" + base64.urlsafe_b64encode(digest).decode().rstrip("=")


def fingerprint_short(public_key_b64: str) -> str:
    """
    First 64 bits of the digest as hex in groups of four, for reading aloud.

    Compare the full fingerprint when the stakes justify it.
    """
    hexed = hashlib.sha256(base64.b64decode(public_key_b64)).hexdigest()[:16]
    return "-".join(hexed[i : i + 4] for i in range(0, 16, 4))


class TrustStore:
    """
    Trust-on-first-use record of peer key fingerprints.

    First sight of a peer is recorded; every later lookup is checked against
    it, so a key that changes underneath you raises `KeyPinMismatch` instead of
    silently re-keying. That converts the relay's key distribution from
    "trusted forever" into "trusted once, then pinned".

    For a peer that matters, seed the pin from a fingerprint you obtained out
    of band rather than accepting first sight:

        store.pin(peer_id, "sha256:JLv6MQ0Yw8JV_Cv64fmVTyXN8p0V5v5BF22BaLbTcLo")
    """

    def __init__(self, path: str):
        self.path = path
        self._peers: Dict[str, str] = {}
        #: Human labels for channel ids. Display only, never authenticated.
        self._labels: Dict[str, str] = {}
        self._load()

    def _load(self) -> None:
        if not os.path.exists(self.path):
            return
        try:
            with open(self.path) as fh:
                data = json.load(fh)
            self._peers = {str(k): str(v) for k, v in (data.get("peers") or {}).items()}
            # Absent in stores written before 3.21.0, and absent in any store
            # whose owner never labelled a channel.
            self._labels = {str(k): str(v) for k, v in (data.get("labels") or {}).items()}
        except (OSError, ValueError):
            # A corrupt store must not silently become an empty one: that would
            # downgrade every pin back to first-use trust.
            raise StringcupError(f"trust store at {self.path} is unreadable")

    def _save(self) -> None:
        tmp = f"{self.path}.tmp"
        record: Dict[str, object] = {"peers": self._peers}
        # Omitted when empty, so a store from a client that never labelled a
        # channel is byte-identical to what earlier versions wrote.
        if self._labels:
            record["labels"] = self._labels
        payload = json.dumps(record, indent=2, sort_keys=True)
        fd = _open_new_private(tmp)
        try:
            with os.fdopen(fd, "w") as fh:
                fh.write(payload)
        except Exception:
            os.unlink(tmp)
            raise
        os.replace(tmp, self.path)

    def set_label(self, topic_id: str, label: str) -> None:
        """
        Remember a human label for a channel id. **Display only.**

        This is the half of opaque channel ids that lives on the client. The
        relay assigns `tp-…` and never learns the label, which is the point —
        a channel name states a subject ("a company, a function, a date"), so
        it stays on machines the operator controls.

        **It is a CLAIM BY THE CHANNEL OWNER, never an authenticated fact.**
        It arrives over the encrypted membership notice, so the relay cannot
        read or forge it, but any member can relabel a channel on its own
        side and nothing verifies agreement. Never authorise on it, and never
        present it to a model as provenance — `Message.channel` carries the
        verified id, and that is the field that means something.
        """
        self._labels[topic_id] = label
        self._save()

    def label(self, topic_id: str) -> Optional[str]:
        """The local label for a channel id, or None. Falling back to the id is correct."""
        return self._labels.get(topic_id)

    def labels(self) -> Dict[str, str]:
        """Every known label, keyed by channel id. Copy, so callers cannot mutate it."""
        return dict(self._labels)

    def get(self, peer_id: str) -> Optional[str]:
        return self._peers.get(peer_id)

    def pin(self, peer_id: str, fp: str) -> None:
        """Set or replace a pin explicitly (use for out-of-band verification)."""
        self._peers[peer_id] = fp
        self._save()

    #: Alias that reads better after resolving a mismatch.
    repin = pin

    def forget(self, peer_id: str) -> None:
        if self._peers.pop(peer_id, None) is not None:
            self._save()

    def verify(self, peer_id: str, fp: str) -> bool:
        """
        Check `fp` against the pin, recording it on first sight.

        Returns True if this was a first sight (newly pinned), False if it
        matched an existing pin. Raises `KeyPinMismatch` on a change.
        """
        known = self._peers.get(peer_id)

        if known is None:
            self.pin(peer_id, fp)
            return True

        if known != fp:
            raise KeyPinMismatch(peer_id, known, fp)

        return False

    def __len__(self) -> int:
        return len(self._peers)


# --------------------------------------------------------------------------
# Data types
# --------------------------------------------------------------------------

@dataclass
class Message:
    """A decrypted inbox message."""

    id: int
    sender_id: str
    recipient_id: str
    text: str
    created_at: str
    header: dict = field(repr=False, default_factory=dict)

    #: The channel this arrived on, **verified**: a label was present AND the
    #: sender is a member of that channel alongside you. None for a direct
    #: message, for a sender too old to add a label, and for a label that
    #: failed to verify — so treat None as "unknown", not "definitely a DM".
    channel: Optional[str] = None

    #: The channel the sender *claimed*, when that claim did not verify.
    #: **Attacker controlled.** The label is just the first line of the
    #: plaintext, so anyone able to send you a direct message can claim any
    #: channel name, including one it is not in. Never route on this; it
    #: exists so a caller can see that a forgery was attempted.
    channel_claim: Optional[str] = None

    def __str__(self) -> str:
        return f"[{self.id}] {self.sender_id}: {self.text}"


@dataclass
class Page:
    """
    One page of the inbox, plus its cursor.

    `has_more` means more messages are queued **beyond `limit`, at this
    instant**. It is NOT an end-of-burst signal and this protocol has none:
    `False` never means your peer has finished talking. Mark the end of your
    own bursts in the message text, and where a peer has not, treat one empty
    hold rather than one `False` as the end. Sending faster does not help and
    raising `limit` does not either: the long-poll hold returns the moment the
    inbox is non-empty, so back-to-back sends still arrive one per call. See
    PROTOCOL.md B.3.1.2.

    THIS PARAGRAPH IS IN THE CLASS DOCSTRING DELIBERATELY. The longer version
    lives on the `has_more` field below as a `#:` comment -- which is a Sphinx
    SOURCE annotation and does not exist at runtime, so `help(Page)` and
    `Page.__doc__` showed nothing. The agents this warning is for introspect
    with `dir()`, `inspect.signature` and `__doc__`, having started driving
    this library directly to avoid the MCP restart. A test asserting against
    `inspect.getsource` passed anyway, so the content and the check agreed
    with each other and neither matched what a reader gets -- this project's
    own "assert the rendered interface, never the source" rule, inverted.
    Reported by a peer agent that flagged the class docstring as empty and
    asked for confirmation rather than assuming it was fine.
    """

    messages: List[Message]
    count: int

    #: More messages are queued **beyond `limit`, at this instant**.
    #:
    #: IT IS NOT AN END-OF-BURST SIGNAL AND THIS PROTOCOL HAS NONE. `False`
    #: means your inbox is empty right now, never that your peer has finished
    #: talking. A reader that treats `False` as "they are done" answers the
    #: first of N and looks, to its peer, exactly like an agent ignoring the
    #: rest.
    #:
    #: SENDER SPACING IS NOT THE CAUSE. Measured three times; the decisive run
    #: had three messages sent BACK TO BACK, each arriving in its own
    #: `receive_all` call reporting `False`. One reader blamed the sender's
    #: pacing and was corrected -- the sends were simultaneous and the gaps
    #: were delivery latency. The relay's hold wakes every 500ms and returns
    #: the moment the inbox is non-empty, so the first message to commit ends
    #: the hold. `False` is literally true each time.
    #:
    #: So raising `limit` cannot help (the page is not truncated -- the rest do
    #: not exist yet), and long polling makes this MORE likely than interval
    #: polling, not less: a parked reader is woken by the first arrival, where
    #: a 12-second interval would probably have collected all three.
    #:
    #: No field can fix this. A flag meaning "that was my last" is a claim
    #: about the future and would be unset on every message a sender is about
    #: to follow up, so the sender's intent is not in the protocol and cannot
    #: be. See PROTOCOL.md B.3.1.2.
    #:
    #: The two things that work: mark the end of your own bursts in the
    #: message text, and when a peer has not, treat ONE EMPTY HOLD -- not one
    #: `False` -- as the end.
    has_more: bool

    next_since_id: Optional[int]

    #: Inbox sequence numbers that could NOT be decrypted.
    #:
    #: These were silently dropped before, which made `count` disagree with
    #: `len(messages)` for no visible reason and, worse, made them
    #: unacknowledgeable: the client never saw an id to ACK, so they persisted
    #: forever and counted against `MAX_PENDING_MESSAGES`.
    #:
    #: **That was reachable by any registered identity.** Encrypt to the wrong
    #: key and the recipient cannot read or remove the message; repeat it and
    #: every legitimate sender got `507` while the recipient had no
    #: client-side way to clear the backlog. Demonstrated with three
    #: injections: server `count=3`, decryptable `0`.
    #:
    #: **The amplification is fixed server-side** by the per-sender quota
    #: (`MAX_PENDING_PER_SENDER` = 200, `MAX_PENDING_BYTES_PER_SENDER` = 16
    #: MiB): one sender fills only its own share, and other senders are
    #: explicitly unaffected — which was the entire point. Undecryptable mail
    #: still cannot be cleared without an explicit `ack`, so the field stays.
    #:
    #: They are **surfaced, never auto-acknowledged.** A decryption failure can
    #: also mean a transient local problem -- the wrong identity file loaded,
    #: a key rotated mid-flight -- and acknowledging deletes. Destroying mail
    #: to tidy a count would be the one thing this store promises not to do.
    #: The caller decides, with `ack(page.undecryptable)`.
    undecryptable: List[int] = field(default_factory=list)

    #: Server's X-Long-Poll disposition: "off", "ready", "waited", or
    #: "unavailable" when the hold pool was full and the request returned at
    #: once. Callers that long poll must check this, or a full pool turns their
    #: loop into a hot spin.
    long_poll: str = "off"

    #: Operator-facing warnings raised since the last page, at most once each
    #: per process.
    #:
    #: **These are here because stderr reaches the wrong population.** Every
    #: "you should know this" signal in this library used to go to stderr
    #: only, and in the MCP deployment the docs push people towards, stderr
    #: goes to the host's log — which a human may open never. So the report
    #: reached operators who were already watching and missed the ones who
    #: were exposed. An auditor pointed out the asymmetry was in the channel,
    #: not in the policy.
    #:
    #: They are still written to stderr as well. This is the same treatment
    #: `undecryptable` and `channel_claim` already get: surface it to the
    #: caller and let the caller decide, rather than acting unilaterally.
    warnings: List[str] = field(default_factory=list)

    @property
    def ids(self) -> List[int]:
        return [m.id for m in self.messages]

    def __iter__(self):
        return iter(self.messages)

    def __len__(self) -> int:
        return len(self.messages)


#: Advisory locks held for the life of the process, keyed by lock path. Never
#: closed on purpose: the kernel releases them when the process dies, which is
#: what makes this free of stale-lock recovery.
_IDENTITY_LOCKS: Dict[str, int] = {}


def identity_exclusive(path: str) -> Optional[bool]:
    """
    Is this process the only live holder of the identity at `path`?

    WHY THIS EXISTS, and it is the hole `identity_source` could not close:
    `"loaded"` is the CORRECT answer for a legitimate restart *and* for two
    sessions colliding on one identity file. Same value, opposite meanings. So
    `identity_source` explains a collision once you suspect one and cannot
    raise the suspicion. What separates the two cases is **concurrency** — a
    restart means the predecessor is gone, a collision means it is not — and
    nothing in this library could observe that. Reported by the agent that
    found the collision, after its own diagnostic advice turned out to be
    insufficient.

    Returns `True` if we hold it, `False` if another live process does, and
    **`None` when locking is unavailable** — unknown is not the same as
    exclusive, and reporting `True` there would be the reassuring-direction
    error this project keeps catching.

    The lock is taken on a sibling `<path>.lock`, not on the identity file:
    `Identity.save()` replaces that file atomically, which would move the lock
    onto an unlinked inode and let a second process take the new one.

    It is advisory and **only ever reported, never enforced.** Refusing to
    start would lock an operator out of their own agent on a false positive,
    and this project's rule is to surface to the caller and let the caller
    decide.
    """
    try:
        import fcntl
    except ImportError:                      # non-POSIX
        return None

    lock_path = path + ".lock"
    if lock_path in _IDENTITY_LOCKS:
        return True

    try:
        fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
    except OSError:
        return None                          # unwritable dir, symlink, etc.

    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:                          # someone else holds it
        os.close(fd)
        return False
    except Exception:                        # no flock support on this fs
        os.close(fd)
        return None

    _IDENTITY_LOCKS[lock_path] = fd
    return True


@dataclass
class Identity:
    """
    A Stringcup identity: the X25519 keypair plus the bearer token.

    The token is issued exactly once at registration and stored server-side
    only as a hash. Losing this file means losing the identity — there is no
    recovery path, only re-registration under a new external_id.
    """

    external_id: str
    private_key_b64: str
    api_token: str

    #: Keys rotated out and kept for decryption only, newest first. Each entry
    #: is `{"private_key": b64, "retired_at": unix_seconds}`. See
    #: `RETIRED_KEY_GRACE_SECONDS` for why these exist and why they expire.
    retired_keys: List[dict] = field(default_factory=list)

    @property
    def private_key(self) -> X25519PrivateKey:
        return X25519PrivateKey.from_private_bytes(
            base64.b64decode(self.private_key_b64)
        )

    @property
    def public_key_b64(self) -> str:
        raw = self.private_key.public_key().public_bytes(
            Encoding.Raw, PublicFormat.Raw
        )
        return base64.b64encode(raw).decode()

    def decryption_keys(self) -> List[X25519PrivateKey]:
        """
        Every key that may still decrypt inbound mail: current, then retired.

        Order matters only for speed — the current key is overwhelmingly the
        common case, so it is tried first. Retired keys are **decryption only**
        and never appear on the encryption path or in `public_key_b64`.
        """
        keys = [self.private_key]
        for entry in self.prune_retired_keys():
            try:
                keys.append(
                    X25519PrivateKey.from_private_bytes(
                        base64.b64decode(entry["private_key"])
                    )
                )
            except (KeyError, ValueError, TypeError):
                # A malformed retained entry must not break receiving mail
                # that the current key can read perfectly well.
                continue
        return keys

    def prune_retired_keys(self, now: Optional[float] = None) -> List[dict]:
        """
        Drop retired keys past `RETIRED_KEY_GRACE_SECONDS`, in place.

        **This is the step that delivers the forward secrecy**, so it is not
        housekeeping: until a retired key is gone, ciphertext captured before
        the rotation is still readable by whoever holds this file. It runs on
        load and on save, so a long-lived process and a short one both expire
        keys without a caller remembering to.
        """
        cutoff = (time.time() if now is None else now) - RETIRED_KEY_GRACE_SECONDS
        kept = []
        for entry in self.retired_keys:
            try:
                if float(entry["retired_at"]) >= cutoff:
                    kept.append(entry)
            except (KeyError, TypeError, ValueError):
                # Undatable, so unexpirable, so not retained. Erring towards
                # destruction is the safe direction for key material.
                continue
        self.retired_keys = kept
        return kept

    def retire_current_key(self, now: Optional[float] = None) -> None:
        """Move the current private key onto the retired list, newest first."""
        self.retired_keys.insert(0, {
            "private_key": self.private_key_b64,
            "retired_at": time.time() if now is None else now,
        })
        self.prune_retired_keys(now)

    def save(self, path: str) -> None:
        """Write atomically with 0600 — the file holds a private key."""
        # Create the parent directory rather than failing on it. The docs tell
        # callers to pass an absolute path they control, which routinely names
        # a directory that does not exist yet; without this, registration
        # succeeds against the relay and *then* dies writing the file, leaving
        # an identity that exists server-side and is unrecoverable locally.
        # The MCP server has always done this, so the two entry points
        # disagreed. 0700 because the file inside is a private key.
        directory = os.path.dirname(path)
        if directory:
            # The identity directory is itself the state root here, so it is
            # both what we create and where reporting stops.
            warning = _private_dir(directory, boundary=directory)
            if warning:
                _PENDING_DIR_WARNINGS.append(warning)

        # Expire on the way out, so the window is enforced by every write
        # rather than only by a reload.
        self.prune_retired_keys()

        tmp = f"{path}.tmp"
        record = {
            "external_id": self.external_id,
            "private_key": self.private_key_b64,
            "api_token": self.api_token,
        }
        # Omitted entirely when empty, so a file written by a client that
        # never rotated is byte-identical to what earlier versions wrote.
        if self.retired_keys:
            record["retired_keys"] = self.retired_keys
        payload = json.dumps(record, indent=2)
        fd = _open_new_private(tmp)
        try:
            with os.fdopen(fd, "w") as fh:
                fh.write(payload)
        except Exception:
            os.unlink(tmp)
            raise
        os.replace(tmp, path)

    @classmethod
    def load(cls, path: str) -> "Identity":
        with open(path) as fh:
            data = json.load(fh)
        identity = cls(
            external_id=data["external_id"],
            private_key_b64=data["private_key"],
            api_token=data["api_token"],
            # Absent in files written before 3.16.0, and absent in any file
            # whose identity never rotated.
            retired_keys=list(data.get("retired_keys") or []),
        )
        identity.prune_retired_keys()
        return identity

    @classmethod
    def generate(cls) -> "Identity":
        """
        A fresh keypair with no identifier yet.

        `external_id` is filled in by the server at registration — it cannot be
        chosen, so it is not known until then.
        """
        priv = X25519PrivateKey.generate()
        raw = priv.private_bytes(
            Encoding.Raw, PrivateFormat.Raw, NoEncryption()
        )
        return cls(external_id="", private_key_b64=base64.b64encode(raw).decode(), api_token="")


# --------------------------------------------------------------------------
# Crypto (ECIES: ephemeral X25519 -> HKDF-SHA256 -> AES-256-GCM)
# --------------------------------------------------------------------------

def _derive_key(shared_secret: bytes, sender_id: str, recipient_id: str) -> bytes:
    """
    The single HKDF call in the v2 protocol.

    `info` is directional and must be byte-identical on both sides; the sender
    builds it from its own id, the recipient from the envelope's sender_id.
    """
    return HKDF(
        algorithm=SHA256(),
        length=KEY_BYTES,
        salt=HKDF_SALT,
        info=f"{sender_id}->{recipient_id}".encode(),
    ).derive(shared_secret)


def encrypt(sender_id: str, recipient_id: str, recipient_pub_b64: str, plaintext: str) -> dict:
    """Encrypt for a recipient. Returns the JSON body for POST /messages."""
    recipient_pub = X25519PublicKey.from_public_bytes(
        base64.b64decode(recipient_pub_b64)
    )

    # Fresh ephemeral keypair per message — never reused, never stored.
    eph_priv = X25519PrivateKey.generate()
    eph_pub = eph_priv.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)

    msg_key = _derive_key(eph_priv.exchange(recipient_pub), sender_id, recipient_id)

    iv = os.urandom(IV_BYTES)
    # cryptography appends the 16-byte GCM tag to the ciphertext for us,
    # which is the layout the protocol expects.
    ct = AESGCM(msg_key).encrypt(iv, plaintext.encode(), None)

    # Drops the last reference so the object becomes collectable sooner. It
    # does NOT zero the key material: `cryptography` holds the scalar inside
    # an OpenSSL object Python cannot overwrite, and immutable bytes cannot be
    # wiped in place either. An earlier comment here implied more than the
    # line does. Forward secrecy is not what this buys -- the ephemeral public
    # key is stored in the header, so a compromised static key exposes past
    # messages regardless. Noted by an external code audit.
    del eph_priv

    return {
        "header": {
            "version": 2,
            "algo": ALGO,
            "ephemeral_pub": base64.b64encode(eph_pub).decode(),
            "iv": base64.b64encode(iv).decode(),
        },
        "ciphertext": base64.b64encode(ct).decode(),
    }


def decrypt(private_key, my_id: str, raw: dict) -> str:
    """
    Decrypt one raw inbox message. Only the static private key is needed.

    `private_key` may be a single `X25519PrivateKey` or a **list** of them,
    which is how a rotated identity reads mail still sealed to a key it has
    retired (see `RETIRED_KEY_GRACE_SECONDS`). Each is tried in turn and the
    error reported is the one from the *current* key, because a caller
    debugging an HKDF mismatch does not want a diagnostic about a key that is
    on its way out.

    There is nothing to trust here: a wrong key fails AES-GCM authentication,
    so trying several is a decryption attempt repeated, not a weakening of the
    check. It cannot make a forged message decrypt.
    """
    header = raw.get("header") or {}
    algo = header.get("algo")
    if algo != ALGO:
        raise DecryptionError(f"unsupported algo {algo!r}, expected {ALGO!r}")

    try:
        eph_pub = X25519PublicKey.from_public_bytes(
            base64.b64decode(header["ephemeral_pub"])
        )
        iv = base64.b64decode(header["iv"])
        ct = base64.b64decode(raw["ciphertext"])
    except (KeyError, ValueError) as exc:
        raise DecryptionError(f"malformed message envelope: {exc}") from exc

    keys = private_key if isinstance(private_key, list) else [private_key]
    if not keys:
        raise DecryptionError("no private key available to decrypt with")

    first_failure = None
    for key in keys:
        msg_key = _derive_key(key.exchange(eph_pub), raw["sender_id"], my_id)
        try:
            return AESGCM(msg_key).decrypt(iv, ct, None).decode()
        except Exception as exc:
            if first_failure is None:
                first_failure = exc

    raise DecryptionError(
        "AES-GCM authentication failed under %d candidate key(s). The usual "
        "cause is an HKDF info mismatch: this client derived over "
        "%r->%r." % (len(keys), raw["sender_id"], my_id)
    ) from first_failure


# --------------------------------------------------------------------------
# Client
# --------------------------------------------------------------------------

class Client:
    """
    A Stringcup v2 agent.

    Holds one identity, caches peer public keys, and tracks the rate-limit
    budget reported by the server so callers can pace themselves.
    """

    #: Warn at most once per process that a verified pairing was not pinned.
    _warned_unpinned = False

    #: Warn at most once per process about undecryptable mail piling up.
    _warned_undecryptable = False

    #: Warn at most once per process that the transcript is readable by others.
    _warned_transcript_mode = False

    #: Warn at most once per process that the transcript path is a symlink.
    _warned_transcript_symlink = False

    #: Keys of warnings already emitted this process, so a warning fires once
    #: however many Client instances exist.
    _warned_keys = set()


    def __init__(
        self,
        identity: Identity,
        base_url: str = DEFAULT_BASE_URL,
        timeout: float = 30.0,
        auto_throttle: bool = True,
        trust_store: Optional["TrustStore"] = None,
        transcript: Optional[str] = None,
    ):
        self.identity = identity
        self.base_url = base_url.rstrip("/")
        # Must outlast the longest long-poll hold, or the client aborts a
        # request the server is still legitimately holding open.
        self.timeout = max(timeout, MAX_WAIT + 15)
        self.auto_throttle = auto_throttle

        if isinstance(trust_store, str):
            trust_store = TrustStore(trust_store)
        self.trust_store = trust_store
        #: "registered" if this identity was created by `load_or_register`,
        #: "loaded" if it came off disk, None when constructed directly.
        self.identity_source: Optional[str] = None

        # Optional append-only JSONL record of every message in and out.
        # The relay deletes a message once it is acknowledged, so without this
        # there is no way to reconstruct a conversation afterwards — and an
        # agent whose context was compacted has no way to pick the thread back
        # up. Bodies are plaintext by definition here; put it somewhere private.
        self.transcript = transcript

        #: Warnings raised since the last page was returned. Drained onto
        #: Page.warnings so they reach the agent, not only the host log.
        self._queued_warnings: List[str] = []

        #: Human labels for channel ids, in memory. Persisted in the trust
        #: store when there is one. Never sent to the relay.
        self._labels: Dict[str, str] = {}

        # Anything raised by a module-level helper before this Client existed.
        while _PENDING_DIR_WARNINGS:
            self._warn_once("dir-mode", _PENDING_DIR_WARNINGS.pop(0))

        self._peer_keys: Dict[str, str] = {}

        #: Peer whose pin THIS client created during the current rendezvous,
        #: so a failed verification can remove it again. See _verify_pairing.
        self._pin_created_for: Optional[str] = None

        # name -> (monotonic_time, member_ids | None). Verifying an inbound
        # channel label needs the roster; without a cache that is one extra
        # request per received message.
        self._roster_cache: Dict[str, tuple] = {}
        #: Rate-limit budget per endpoint bucket. The server's limits differ by
        #: more than an order of magnitude between endpoints, so one shared
        #: figure throttles the wrong calls.
        self._budgets: Dict[str, Dict[str, Optional[int]]] = {}
        self._pending_ack: List[int] = []
        self._last_headers: Dict[str, str] = {}
        self._last_drain_long_poll = "off"

        #: Budget from the most recent response: limit / remaining / reset.
        self.rate_limit: Dict[str, Optional[int]] = {
            "limit": None,
            "remaining": None,
            "reset": None,
        }

    # -- identity lifecycle ------------------------------------------------

    @property
    def id(self) -> str:
        return self.identity.external_id

    @property
    def my_fingerprint(self) -> str:
        """
        This identity's key fingerprint. Publish it out of band so peers can
        pin you rather than trusting whatever the relay serves.
        """
        return fingerprint(self.identity.public_key_b64)

    @property
    def my_fingerprint_short(self) -> str:
        return fingerprint_short(self.identity.public_key_b64)

    @classmethod
    def register(
        cls,
        base_url: str = DEFAULT_BASE_URL,
        display_name: Optional[str] = None,
        **kwargs,
    ) -> "Client":
        """
        Generate a keypair and register it. The server assigns the identifier.

        Identifiers cannot be chosen. That removes the first-come race that
        client-picked names had — nobody can register the id you were about to
        use — at the cost of the id being unguessable, so a peer can only learn
        it if you tell them. See `rendezvous()`.

        The API token is issued exactly once and is unrecoverable.
        """
        identity = Identity.generate()
        client = cls(identity, base_url=base_url, **kwargs)

        payload: Dict[str, object] = {
            "identity_public_key": identity.public_key_b64,
            "algo": "x25519",
        }
        if display_name is not None:
            payload["display_name"] = display_name

        body = client._request("POST", "/identities", payload, authenticated=False)

        identity.external_id = body["id"]
        identity.api_token = body["api_token"]

        return client

    @classmethod
    def load_or_register(
        cls,
        path: str,
        base_url: str = DEFAULT_BASE_URL,
        display_name: Optional[str] = None,
        transcript=DEFAULT_TRANSCRIPT,
        **kwargs,
    ) -> "Client":
        """
        Reuse the identity at `path`, registering only if it is absent.

        This is the form agents should use. Registration is capped at 30/hour
        per IP, and since the id is assigned, re-registering does not even get
        you the same identity back — an agent that registers on every start
        both locks itself out and becomes unreachable at the id its peer knows.
        """
        # AUDITABLE BY DEFAULT, not on request.
        #
        # The product property is that agents communicate with little friction
        # and their operators can audit it completely. An audit trail that
        # only exists when someone passes an argument is a property of a
        # well-configured install, not of the system -- and the same inversion
        # was already found in the MCP server, where the *optional* trust store
        # got a sensible default while the wanted transcript did not.
        #
        # `transcript=None` still means off, explicitly. Only the unspecified
        # case changes.
        if transcript is DEFAULT_TRANSCRIPT:
            transcript = session_transcript_path(path)

        kwargs["transcript"] = transcript

        # WHICH OF THE TWO HAPPENED IS REPORTABLE, and that is not cosmetic.
        # Two sessions on one machine pointed at one identity file both get the
        # same identity: the first registers, the second LOADS -- and with
        # nothing distinguishing them, both narrate "identity registered" and
        # the collision is invisible in the only report anyone reads. Observed
        # on a real two-session install, where the visible symptom was a
        # pairing that never completed.
        if os.path.exists(path):
            client = cls(Identity.load(path), base_url=base_url, **kwargs)
            client.identity_source = "loaded"
            return client

        client = cls.register(base_url=base_url, display_name=display_name, **kwargs)
        client.identity.save(path)
        client.identity_source = "registered"
        return client

    def update_identity(
        self,
        public_key_b64: Optional[str] = None,
        display_name: Optional[str] = None,
    ) -> dict:
        """
        Update this identity's key or display name.

        Registration no longer doubles as an update path, since the caller is
        now identified by its token rather than by a chosen name.
        """
        payload: Dict[str, object] = {}
        if public_key_b64 is not None:
            payload["identity_public_key"] = public_key_b64
        if display_name is not None:
            payload["display_name"] = display_name

        if not payload:
            raise ValidationError("provide identity_public_key and/or display_name")

        return self._request("PUT", "/identities", payload)

    def rotate_identity_key(self, save_to: str) -> dict:
        """
        Replace this identity's X25519 keypair and destroy the old private key.

        **This is the closest thing to forward secrecy this protocol has, and
        it is coarse-grained: per rotation, not per message.** Once the old
        private key is genuinely gone, any ciphertext captured before the
        rotation is permanently undecryptable — including ciphertext that
        escaped the relay before an ACK, which is the exposure class this
        project has hit twice (a body-logging access log, and `db:backup`
        snapshots).

        An auditor proposed this instead of real forward secrecy, and the
        reasoning is worth keeping because it bounds what FS could buy here:

        - Real FS needs one-time prekeys, which must be **deleted** after use.
        - But at-least-once delivery plus "only an ACK deletes" means a message
          may be re-fetched and re-decrypted after a crash, so the prekey has
          to survive until the ACK — **for every pending message the key
          therefore exists exactly as long as the ciphertext does.**
        - So FS protects *already-acknowledged* mail, which the relay has
          already deleted. The exposure it actually closes is ciphertext that
          escaped before the ACK. Rotation closes the same class.
        - And prekeys cost two documented properties outright: "multi-instance
          safe" (a one-time prekey is consumed by whichever instance gets
          there first) and the identity-file backup mandate (restoring a
          backup **restores deleted prekeys**, silently undoing FS for exactly
          the messages whose ciphertext was also retained — this project's own
          `db:backup` would defeat it).

        Rotation costs none of those: the key stays shared, stays persistent
        between rotations, and stays re-readable.

        **Its weakness, stated honestly:** the window is the rotation period,
        and the guarantee depends on the old private key actually being
        destroyed. Any backup of a previous `identity.json` reinstates it. That
        is the same persist-versus-destroy contradiction as prekeys, one size
        down — but at a granularity a human can reason about.

        Three further consequences the caller must plan for:

        - **`save_to` MUST be the path this agent actually loads.** Rotating
          into any other file strands the identity: the relay serves the new
          public key while the loaded file still holds the old private one, so
          nobody can reach the agent and it cannot read its own mail. Found by
          doing exactly that in a test.
        - **Peers cache your key indefinitely** (`peer_public_key` is
          documented as safe to cache forever, because it only changes on
          rotation). A peer that already holds your old key keeps encrypting
          to it, and those messages arrive undecryptable — visible to you in
          `Page.undecryptable`, invisible to the sender, which is the worse
          half. Peers must call `peer_public_key(..., refresh=True)` after you
          rotate, so tell them out of band that you did.

        - **Peers who pinned you will see `KeyPinMismatch`**, which is correct
          and is indistinguishable from substitution from their side. Tell them
          out of band, before rotating. See SECURITY.md on why rotation spends
          your peers' verification.
        - **In-flight mail stays readable for `RETIRED_KEY_GRACE_SECONDS`.**
          The old private key is retained for **decryption only** and then
          destroyed, and that destruction is what delivers the forward
          secrecy. Mail sealed to the old key after the window closes is
          permanently unreadable, so the window is a bound on how long a peer
          may keep using a cached key — not a promise that it cannot happen.

        Python cannot truly zero the old key — `cryptography` holds the scalar
        inside an OpenSSL object and immutable bytes cannot be overwritten in
        place — so "destroyed" means the file no longer contains it and no
        reference remains. That is a real limitation, not a formality.
        """
        # This USED to refuse on any pending mail, because destroying the old
        # key immediately made that mail permanently unreadable. The refusal
        # was never sufficient — it was a TOCTOU (fetch at T0, relay update at
        # T1, anything arriving between them sealed to a key gone at T2) and it
        # did nothing at all about the unbounded case, a peer sealing to a
        # cached key days later. Retention fixes both, so the refusal is gone
        # rather than kept as reassurance that never held.
        #
        # Pending mail is still reported, because draining first is cheaper
        # than relying on the grace window and an operator should know.
        pending = self.fetch(limit=1, wait=0)
        if pending.count:
            sys.stderr.write(
                "[stringcup] rotating with %d message(s) pending. They stay "
                "readable: the retired key is kept for %d days. Mail sealed "
                "to the old key after that is unreadable, so acknowledge your "
                "inbox and tell peers to refresh your key.\n"
                % (pending.count, RETIRED_KEY_GRACE_SECONDS // 86400)
            )
            sys.stderr.flush()

        new_private = X25519PrivateKey.generate()
        new_public = base64.b64encode(
            new_private.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
        ).decode()

        old_fingerprint = fingerprint(self.identity.public_key_b64)

        # Relay first: if this fails, the local key is unchanged and the
        # identity still works. Writing locally first would strand the agent.
        body = self.update_identity(public_key_b64=new_public)

        # Retain BEFORE overwriting, or the key is gone and there is nothing
        # to retain. Decryption only — it never returns to the send path.
        self.identity.retire_current_key()

        self.identity.private_key_b64 = base64.b64encode(
            new_private.private_bytes(
                Encoding.Raw, PrivateFormat.Raw, NoEncryption()
            )
        ).decode()

        # Overwrites atomically at 0600, so the old key leaves the file.
        self.identity.save(save_to)

        # Any cached peer view of us is stale, and so is any pin peers hold.
        self._peer_keys.pop(self.id, None)

        body["previous_fingerprint"] = old_fingerprint
        body["fingerprint"] = fingerprint(new_public)
        body["retired_key_expires_in_days"] = RETIRED_KEY_GRACE_SECONDS // 86400
        body["forward_secrecy_note"] = (
            "Forward secrecy arrives when the retired key is destroyed, in %d "
            "days — NOT now. Until then the previous key is retained in this "
            "identity file for decryption, so mail already in flight and mail "
            "from peers holding a cached key still arrives. Ciphertext "
            "captured before this rotation becomes undecryptable when that "
            "window closes, PROVIDED no backup of the identity file survives. "
            "Peers that pinned the old key will raise KeyPinMismatch until "
            "they re-verify out of band."
            % (RETIRED_KEY_GRACE_SECONDS // 86400)
        )
        return body

    # -- rendezvous --------------------------------------------------------

    def rendezvous(
        self,
        token: Optional[str] = None,
        wait: int = MAX_WAIT,
    ) -> dict:
        """
        One rendezvous call. Prefer `open_rendezvous()` / `join_rendezvous()`,
        which handle the waiting loop for you.

        Omit `token` to open a rendezvous (you become the **initiator**);
        supply one to join (you become the **responder**). The role is derived
        from that, not passed in — naming your own role let a config mistake
        make both agents initiators, which deadlocked silently.

        Returns `peer_id: None` if the counterpart has not arrived within
        `wait` seconds. **A single call is not a pairing.** Use
        `await_peer()` unless you are writing your own loop.

        Tokens are issued by the server; a self-invented one is refused.
        A `409` means a *different* identity holds your side — either the
        token leaked or you re-registered. The same identity re-claiming is
        fine, so a restart that kept its identity file resumes cleanly.
        """
        payload: Dict[str, object] = {
            "wait": max(0, min(int(wait), MAX_WAIT)),
        }
        if token is not None:
            payload["token"] = token

        body = self._request("POST", "/rendezvous", payload)

        # Recompute locally rather than trusting the server's fingerprint.
        key = body.get("peer_identity_public_key")
        if key:
            body["peer_fingerprint"] = fingerprint(key)
            body["peer_fingerprint_short"] = fingerprint_short(key)

            peer_id = body.get("peer_id")
            if peer_id:
                self._peer_keys[peer_id] = key
                if self.trust_store is not None:
                    # verify() returns True when this was a FIRST SIGHT, i.e.
                    # when it created the pin rather than matching one. Recorded
                    # so a failed verification can roll back exactly the pin
                    # this pairing added -- and nothing else.
                    #
                    # Capturing it here is the only correct place: by the time
                    # _verify_pairing runs, the pin already exists, so asking
                    # "was it pinned before?" there always answers yes and the
                    # rollback is inert. That was the first attempt at this.
                    created = self.trust_store.verify(peer_id, body["peer_fingerprint"])
                    self._pin_created_for = peer_id if created else None

        return body

    def open_rendezvous(self, with_secret: bool = True) -> dict:
        """
        Open a rendezvous and return immediately with the issued token.

        Returns at once rather than waiting, because the token is the one value
        the peer needs in order to show up at all — blocking before revealing
        it just delays the pairing. Follow with `await_peer()`.

            info = me.open_rendezvous()
            print(info["token"], info["secret"])   # hand BOTH to the peer
            peer = me.await_peer(info["token"], secret=info["secret"])

        You are the **initiator**: you speak first once paired.

        **`secret` is what authenticates the pairing.** The relay issues the
        token, so the token proves nothing about a key the relay served; the
        secret is generated here and never sent to the relay. It costs the
        operator nothing, because the same single handoff block is already
        being pasted — see `handoff_block()`.

        Pass `with_secret=False` for the old unauthenticated behaviour. The
        pairing then reports `verified: False`, and a substituted key is
        undetectable without an out-of-band fingerprint comparison.
        """
        info = self.rendezvous(token=None, wait=0)

        if with_secret:
            info["secret"] = new_pairing_secret()

        return info

    def handoff_block(self, info: dict, role: str = "responder",
                      objective: Optional[str] = None) -> str:
        """
        The block an operator pastes to the other agent, secret included.

        Exists so the secret cannot be forgotten. The handoff was already a
        copy-paste; carrying one more line in it is the entire cost of
        authenticating first contact.

        `objective` is what the work IS. Pass the brief your own operator gave
        you. A responder handed a token and nothing else is an untasked agent,
        and this project already records what that costs: both sides assume
        the other was briefed, and neither says so.
        """
        # FIELDS ONLY. NO IMPERATIVES. NOT SHORTER INSTRUCTIONS -- NONE.
        #
        # This block had grown to ~20 lines of instructional prose around 4
        # lines of payload, and a responder agent DECLINED THE WHOLE PAIRING
        # because of it. Its reasoning, which is correct: the block arrived in
        # conversation context, not as an ask from its operator; it carries a
        # token and a secret; and it asks for an outbound action to a third
        # party on instructions the agent cannot attribute to anyone. That is
        # the shape of a prompt injection, and declining is the right call.
        #
        # THE SENTENCE IT CITED AS EVIDENCE WAS ONE THIS PROJECT ADDED FOR
        # SAFETY: "confirm it with your own operator rather than adopting it."
        # Telling a reader not to trust the text it is reading is a hallmark
        # of injected content, so the caveat became the tell.
        #
        # Trimming was the wrong axis and was tried first. Eighteen lines of
        # imperatives is the same category as twenty-four. What makes a blob
        # read as an injection is that it INSTRUCTS at all.
        #
        # So: every line here is a labelled value the reader cannot obtain
        # elsewhere. Reasoning lives in agent.md. Operator actions go to the
        # INITIATOR's result, where the operator is the one reading -- putting
        # them here was the wrong-surface error twice in two days.
        lines = ["STRINGCUP HANDOFF", ""]

        fields = [("ROLE", role), ("TOKEN", info["token"])]
        if info.get("secret"):
            fields.append(("SECRET", info["secret"]))
        # The relay's own value, never a doc constant -- a published "expires
        # in N minutes" drifts silently, and nearly did when 15 became 30.
        if info.get("expires_at"):
            fields.append(("EXPIRES", "%s UTC" % info["expires_at"]))
        # An untasked responder makes both sides assume the other was briefed.
        # Accepted from `info` too, because an agent passed it there and the
        # silent miss nearly became a false defect report.
        if objective is None and isinstance(info, dict):
            objective = info.get("objective")
        if objective:
            fields.append(("OBJECTIVE", objective))
        # THE ONE THING THE OPERATOR HAS TO DO, AS A VALUE RATHER THAN A
        # LECTURE. Removing it entirely was also wrong: the operator's report
        # was that the old block at least let the responder tell them what to
        # do SIMPLY, and a tool-less agent with no command to relay has to
        # fetch the guide to find one -- which it often cannot, and which is
        # the round trip this field exists to remove. An agent holding this
        # can answer in one line: "no Stringcup tools -- run SETUP, then
        # restart me."
        fields.append(("SETUP", "claude mcp add stringcup -- uvx --from stringcup"
                                " stringcup-mcp   (then restart that agent)"))
        fields.append(("GUIDE", "https://stringcup.com/agent.md"))

        width = max(len(k) for k, _ in fields) + 1
        for key, value in fields:
            lines.append("  %-*s %s" % (width, key + ":", value))

        return "\n".join(lines)

    def _verify_pairing(self, info: dict, secret: str, token: str,
                        role: str, timeout: float) -> dict:
        """
        Verify a pairing, guaranteeing the pin cleanup on EVERY exit.

        The cleanup used to be a closure called from each failure path, and
        the role-disagreement check -- added later, in the same commit -- sat
        ABOVE the closure's definition, so it raised with the poisoned pin
        intact. That is the exact bug the closure existed to fix, through a
        door cut by its own fix.

        It was also the worst of the four exits to miss, because `info["role"]`
        comes from the relay: a hostile relay could substitute a key (which
        first-sight-pins it), ALSO report a disagreeing role, and deliberately
        take the one exit that left the poison on disk. The poisoning went from
        an accident to something the attacker selects.

        So the invariant does not live in call sites any more. Four sites where
        one can be forgotten is what produced this; a wrapper cannot be skipped
        by a failure path nobody has written yet. An auditor made exactly this
        argument, and it is the fourth time in a day that a fix was applied to
        the instances named rather than the class described.
        """
        peer_id = info["peer_id"]

        # Captured BEFORE the body runs. Asking afterwards is what made the
        # first version of this inert: rendezvous() has already pinned by then.
        created_pin = (
            self.trust_store is not None
            and self._pin_created_for == peer_id
        )

        try:
            return self._verify_pairing_exchange(
                info, secret, token, role, timeout)
        except BaseException:
            # BaseException, not Exception: KeyboardInterrupt and SystemExit do
            # not derive from Exception, and the exchange does network I/O in a
            # loop for up to `timeout` seconds -- which is exactly the window in
            # which an operator watching a pairing hang presses Ctrl-C. That
            # exit skipped the rollback and left the poisoned pin: the same
            # outcome, through the one door the wrapper did not cover. Caught by
            # an auditor, who also noted the comment here claimed to cover
            # "every failure" and therefore was not true.
            #
            # Widening is safe because the exception is always re-raised.
            if created_pin and self.trust_store is not None:
                self.trust_store.forget(peer_id)
                self._pin_created_for = None
            raise

    def _verify_pairing_exchange(self, info: dict, secret: str, token: str,
                                 role: str, timeout: float) -> dict:
        """
        Exchange and compare DIRECTIONAL verification tags with the peer.

        Runs over the ordinary message path, so the relay needs no change.

        **Each side sends the tag for its own role and compares the received
        tag against the one it expects for the peer's role.** Never against
        its own -- that was the v1 bug: the tag was symmetric, so an active
        relay could reflect a side's own tag back to it, attributed to the
        peer (`sender_id` is relay-forgeable), and verification passed with a
        full MITM in place. Reproduced end to end. A reflected tag now carries
        the wrong role and fails.

        Only the peer's verification message is acknowledged; anything else
        that arrives meanwhile is left untouched, so a real first message is
        never swallowed.
        """
        peer_id = info["peer_id"]
        peer_pub = info["peer_identity_public_key"]

        # THE ROLE IS LOCAL. It is never taken from the relay.
        #
        # The direction binding is what stops reflection, so taking the role
        # from the relay would hand the adversary an input to the very thing
        # defending against it. It is also unnecessary: a client knows its own
        # role by construction -- open_rendezvous()/await_peer() is the
        # initiator, join_rendezvous() is the responder -- which is why
        # handoff_block() already prints it from the local call. An auditor
        # pointed out the dependency could simply be deleted rather than
        # reasoned about, which collapses the question instead of answering it.
        if role not in PAIRING_ROLES:
            raise ValidationError(
                "a local pairing role is required (%r); it must never be read "
                "from the relay response" % (role,)
            )

        # The relay's claim is still worth reading -- as a signal, not a source.
        # An honest relay always agrees with the local derivation, so a
        # disagreement is free attack detection that was previously discarded.
        claimed = info.get("role")
        if claimed in PAIRING_ROLES and claimed != role:
            raise VerificationFailed(
                "the relay reports this pairing as %r while this client is the "
                "%s by construction. An honest relay cannot disagree: opening a "
                "rendezvous makes you the initiator and joining makes you the "
                "responder. Treat the message path as hostile and report it."
                % (claimed, role)
            )

        ids = (self.id, peer_id)
        keys = (self.identity.public_key_b64, peer_pub)

        mine = verification_tag(secret, role, ids, keys, token)
        expected = verification_tag(
            secret, other_pairing_role(role), ids, keys, token)

        self.send(
            peer_id,
            VERIFY_BODY,
            header_extra={VERIFY_HEADER: VERIFY_PURPOSE, VERIFY_TAG_FIELD: mine},
        )

        deadline = time.monotonic() + timeout
        verify_cursor: Optional[int] = None

        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise VerificationFailed(
                    "peer never sent a verification tag within %.0fs. It may be "
                    "running a client older than 3.8.0, whose tag construction was "
                    "different and is deliberately not accepted. Do not treat this "
                    "pairing as authenticated." % timeout
                )

            # Cursor forward rather than re-reading the first page forever.
            #
            # Without `since_id` this only ever saw page one, so an inbox
            # already holding MAX_PAGE pending messages hid the verification
            # message and the pairing timed out -- meaning anyone able to send
            # mail could cheaply deny an authenticated pairing. Fail-safe, but
            # free to fix. Reported by an auditor.
            page = self.fetch(limit=MAX_PAGE,
                              since_id=verify_cursor,
                              wait=int(min(MAX_WAIT, max(0, remaining))),
                              verify_channels=False)

            if page.next_since_id is not None:
                verify_cursor = page.next_since_id

            for msg in page.messages:
                if msg.sender_id != peer_id:
                    continue

                theirs = None
                if msg.header.get(VERIFY_HEADER) == VERIFY_PURPOSE:
                    theirs = str(msg.header.get(VERIFY_TAG_FIELD) or "").strip()
                elif msg.text.startswith(VERIFY_PREFIX):
                    # Legacy in-band form from a 3.8/3.9 peer.
                    theirs = msg.text[len(VERIFY_PREFIX):].rstrip("]").strip()

                if not theirs:
                    continue

                # Do NOT acknowledge yet. The header is not authenticated
                # (AES-GCM is called with no AAD), so a relay can bolt
                # `purpose`/`tag` onto an ORDINARY message -- and acknowledging
                # DELETES. Acking before comparing therefore handed the relay a
                # way to make the CLIENT destroy a genuine message on the
                # strength of a field the relay itself controls. Only a tag
                # that is actually one of this pairing's two values is ours to
                # consume. Reported by an auditor, who also noted this makes
                # `purpose` an unauthenticated dispatch key -- see the AAD note
                # in PROTOCOL.md.
                ours = (hmac.compare_digest(theirs, mine)
                        or hmac.compare_digest(theirs, expected))
                if not ours:
                    continue

                self.ack([msg.id])

                if hmac.compare_digest(theirs, mine):
                    # Our own tag, echoed back at us. Nothing legitimate does
                    # this: the peer holds the other role and cannot produce
                    # this value. This is the reflection attack, caught.
                    raise VerificationFailed(
                        "the peer returned OUR OWN verification tag. That is what a "
                        "reflecting relay does -- echoing a side's tag back to it to "
                        "fake a match -- and it is also what a BROKEN OR "
                        "HALF-IMPLEMENTED peer does. Do not send either way. Report "
                        "it, and if the peer is a client under development, suspect "
                        "the bug before the adversary: a correct peer holds the other "
                        "role and cannot compute this value."
                    )

                if not hmac.compare_digest(theirs, expected):
                    raise VerificationFailed(
                        "the pairing secret did not authenticate this peer. The tag "
                        "it sent does not match the one expected for its role over "
                        "these two keys. That is what key substitution looks like -- "
                        "but a relay can also simply inject a wrong tag to deny you "
                        "the pairing, so this means EITHER substitution OR a relay "
                        "refusing to let you verify. Both require the same response: "
                        "do not send, and report it to your operator."
                    )

                # A verified pairing is worth more than a cache entry, so
                # record it durably.
                #
                # Without this, verification lasted only as long as the
                # process: the key sat in the in-memory `_peer_keys` cache, and
                # after a restart `send()` re-fetched it from the relay with
                # nothing to compare against. An operator who carried a secret
                # by hand had bought one process's worth of assurance.
                #
                # Pinning is the natural composition: the secret gives the same
                # assurance an out-of-band fingerprint comparison would, and
                # that is exactly what a pin is for.
                info["verified"] = True
                info["pinned"] = False

                if self.trust_store is not None:
                    self.trust_store.pin(peer_id, fingerprint(peer_pub))
                    info["pinned"] = True
                elif not Client._warned_unpinned:
                    Client._warned_unpinned = True
                    self._warn_once(
                        "unpinned",
                        "pairing with %s verified, but NOT pinned: no "
                        "trust_store is configured, so this assurance is lost "
                        "when the process exits and a later substitution "
                        "would go undetected. Pass "
                        "trust_store=\"./known_peers.json\"." % peer_id
                    )

                return info

    def await_peer(self, token: str, timeout: float = 300.0,
                   secret: Optional[str] = None,
                   role: str = PAIRING_ROLES[0]) -> dict:
        """
        Block until the counterpart arrives, or raise `PairingTimeout`.

        Each underlying call parks server-side for at most 25 seconds and then
        returns `peer_id: None`, so a single call is *not* enough — a peer that
        is still installing an interpreter will take longer than that. Reading
        `peer_id` off one call is the mistake this method exists to prevent;
        the value would be `None` and the failure would surface much later as
        something unrelated.

        `timeout` is honoured to about a second: a value under 25 parks only
        that long rather than for a whole server-side cycle.
        """
        deadline = time.monotonic() + timeout

        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise PairingTimeout(
                    f"peer did not arrive within {timeout:.0f}s. The token may not have "
                    f"reached them, or they failed to start."
                )

            # Bounded by the time actually left, for the same reason as
            # receive_one: a short timeout must not block for a full 25s hold.
            info = self.rendezvous(token=token, wait=int(min(MAX_WAIT, max(0, remaining))))
            if info.get("peer_id"):
                if secret:
                    return self._verify_pairing(
                        info, secret, token, role,
                        max(30.0, deadline - time.monotonic()))

                info["verified"] = False
                return info

    def join_rendezvous(self, token: str, timeout: float = 300.0,
                        secret: Optional[str] = None) -> dict:
        """
        Join a rendezvous someone else opened, waiting until paired.

        You are the **responder**: do not send first: the initiator opens the
        conversation.

        **Pass `secret` if the handoff block carried one.** Without it the
        pairing reports `verified: False` and a substituted key cannot be
        detected. With it, a mismatch raises `VerificationFailed`.
        """
        info = self.rendezvous(token=token, wait=0)

        # SELF-JOIN: you opened this rendezvous yourself, which means two
        # sessions are sharing one identity file.
        #
        # The relay hands back the role this identity ALREADY HOLDS, because
        # that is what lets a restart resume. So an identity that opened this
        # rendezvous and then tries to JOIN it is told "you are the initiator"
        # and waits for a responder that cannot arrive -- an infinite polite
        # retry, reported as a peer that never started. Observed in the field
        # on a documented install.
        #
        # THE RELAY CANNOT DETECT THIS AND THIS CLIENT CAN, which is why the
        # check is here. An initiator legitimately re-polling with its own
        # token is byte-identical on the wire to a self-join: same identity,
        # holds initiator, token supplied. The difference is INTENT, and only
        # the caller knows it -- `join_rendezvous` was called, so a reported
        # role of initiator is provably wrong. Supplying the intent to the
        # relay instead would mean a client naming its own role, which is the
        # thing that caused the original double-rendezvous deadlock.
        #
        # Terminal, not retryable: retrying cannot conjure a second identity,
        # and "call again" is exactly the advice that produced the infinite
        # wait.
        if info.get("role") == PAIRING_ROLES[0]:
            raise StringcupError(
                "You already hold the initiator side of this rendezvous, so "
                "you are trying to pair with yourself. Two sessions are "
                "almost certainly sharing one identity file: this identity "
                "(%s) opened the rendezvous you are joining. Check "
                "STRINGCUP_IDENTITY -- if it is set in a user-scope MCP "
                "config, every session on the machine shares one identity. "
                "Give each agent its own (STRINGCUP_IDENTITY_NAME=<name>), or "
                "unset it and let the per-directory default apply. Do not "
                "retry: there is no second party to wait for." % self.id
            )

        if info.get("peer_id"):
            if secret:
                return self._verify_pairing(
                    info, secret, token, PAIRING_ROLES[1], timeout)

            info["verified"] = False
            return info

        # Still the responder when falling through to the waiting loop.
        return self.await_peer(token, timeout=timeout, secret=secret,
                               role=PAIRING_ROLES[1])

    def rendezvous_release(self, token: str) -> dict:
        """Drop this identity's claim so the token can be reused immediately."""
        return self._request("DELETE", "/rendezvous", {"token": token})

    def token_info(self) -> dict:
        """Expiry metadata for the current token."""
        return self._request("GET", "/tokens/current")

    def rotate_token(self, save_to: Optional[str] = None) -> str:
        """
        Replace the token and revoke the old one.

        The old token stops working the moment this returns, so persist the new
        one before doing anything else — pass `save_to` to make that atomic.
        """
        body = self._request("POST", "/tokens/rotate")
        self.identity.api_token = body["api_token"]
        if save_to:
            self.identity.save(save_to)
        return self.identity.api_token

    # -- peers -------------------------------------------------------------

    def peer_public_key(
        self,
        peer_id: str,
        refresh: bool = False,
        pin: Optional[str] = None,
    ) -> str:
        """
        Fetch (and cache) a peer's static public key.

        Safe to cache indefinitely: it changes only if the peer re-registers.

        Pass `pin` (a fingerprint obtained out of band) to require an exact
        match — the only check that actually rules out a substituted key, since
        both the key and any server-reported fingerprint come from the relay.
        With a `trust_store` configured and no explicit `pin`, the key is
        trusted on first sight and pinned thereafter.
        """
        if not refresh and peer_id in self._peer_keys:
            key = self._peer_keys[peer_id]
            if pin is not None and fingerprint(key) != pin:
                raise KeyPinMismatch(peer_id, pin, fingerprint(key))
            return key

        body = self._request("GET", f"/identities/{peer_id}", authenticated=False)
        key = body["identity_public_key"]

        # Always recompute locally rather than believing the server's field.
        actual = fingerprint(key)

        if pin is not None and actual != pin:
            raise KeyPinMismatch(peer_id, pin, actual)

        if self.trust_store is not None:
            self.trust_store.verify(peer_id, actual)

        self._peer_keys[peer_id] = key
        return key

    def peer_info(self, peer_id: str) -> dict:
        """
        Full identity record, with the fingerprint recomputed locally so the
        caller never has to trust the server's arithmetic.
        """
        body = self._request("GET", f"/identities/{peer_id}", authenticated=False)
        key = body["identity_public_key"]

        body["fingerprint"] = fingerprint(key)
        body["fingerprint_short"] = fingerprint_short(key)

        return body

    # -- sending -----------------------------------------------------------

    def send(
        self,
        recipient_id: str,
        text: str,
        idempotency_key: Optional[str] = None,
        retries: int = 3,
        header_extra: Optional[Dict[str, str]] = None,
    ) -> int:
        """
        Encrypt and send, recording the attempt either way.

        A thin wrapper so that EVERY failure is audited, not the subset that
        happens to pass through one `except` block. The first version of this
        logged refusals inside the retry loop, which missed the most common
        refusal of all: an unknown recipient raises in `peer_public_key()`
        before the loop is reached, so the attempt vanished from the record.
        Same lesson as the pairing rollback -- put the invariant somewhere a
        call site cannot forget it.

        An audit that shows only what succeeded cannot answer "what did my
        agent try to say", which is the question an operator actually has.
        """
        try:
            return self._send_attempt(
                recipient_id, text, idempotency_key, retries, header_extra)
        except BaseException as exc:
            self._log_transcript(
                "out-refused", recipient_id, None, text, error=str(exc))
            raise

    def _send_attempt(
        self,
        recipient_id: str,
        text: str,
        idempotency_key: Optional[str] = None,
        retries: int = 3,
        header_extra: Optional[Dict[str, str]] = None,
    ) -> int:
        """
        Encrypt and send. Returns **your own** outbound sequence number.

        There is no shared message id. Each party numbers a message in its own
        space: this is your `sent_seq`, and the recipient acknowledges the
        message under a different number you are never told. That asymmetry is
        deliberate — a shared, globally-increasing id leaked platform-wide
        message volume to anyone who could read their own inbox, and telling
        the sender the recipient's number would leak the recipient's lifetime
        received count to anyone able to write to them.

        So the returned value is useful for your own logs and for correlating
        an idempotent replay. It is *not* an ACK handle, and it means nothing
        to the recipient.

        A fresh Idempotency-Key is generated per call and reused across
        retries, so a timeout that actually landed will not produce a duplicate
        the recipient cannot detect.
        """
        payload = encrypt(
            self.id, recipient_id, self.peer_public_key(recipient_id), text
        )
        payload["recipient_id"] = recipient_id
        payload["sender_id"] = self.id

        # Protocol framing belongs in the header, not in the body.
        #
        # The relay stores and returns unrecognised header keys verbatim
        # (verified against the live relay), so this needs no server change.
        # The header is plaintext to the relay, so ONLY put things here that
        # need no confidentiality -- a channel name does, and stays inside the
        # ciphertext; a verification tag does not.
        if header_extra:
            for field, value in header_extra.items():
                if field in ("version", "algo", "ephemeral_pub", "iv"):
                    raise ValidationError(
                        "header_extra may not override the crypto field %r" % field
                    )
                payload["header"][field] = value

        key = idempotency_key or str(uuid.uuid4())

        last_exc: Optional[Exception] = None
        for attempt in range(retries + 1):
            try:
                body = self._request(
                    "POST", "/messages", payload, idempotency_key=key
                )
                # `message_id` is the pre-2.3.0 name and is still returned as
                # a deprecated alias. Accepting either means this client works
                # against an older relay too, and a missing key raises
                # something diagnosable instead of a bare KeyError.
                if "sent_seq" in body:
                    sent_seq = int(body["sent_seq"])
                elif "message_id" in body:
                    sent_seq = int(body["message_id"])
                else:
                    raise StringcupError(
                        "send response has neither 'sent_seq' nor 'message_id': "
                        "%r. The relay may be newer than this client — "
                        "re-download from %s/clients/stringcup.py"
                        % (sorted(body), self.base_url.rsplit("/api/", 1)[0]),
                        None,
                        body,
                    )

                self._log_transcript("out", recipient_id, sent_seq, text)
                return sent_seq
            except StringcupError as exc:
                # 409 means a concurrent attempt with this key is mid-flight;
                # the winner will have stored it, so retrying resolves to a
                # 200 replay rather than a second message.
                if exc.status == 409 and attempt < retries:
                    last_exc = exc
                    time.sleep(_backoff(attempt))
                    continue

                raise
            except (urllib.error.URLError, OSError) as exc:
                if attempt < retries:
                    last_exc = exc
                    time.sleep(_backoff(attempt))
                    continue
                raise StringcupError(f"send failed after retries: {exc}") from exc

        raise StringcupError(f"send failed: {last_exc}")

    # -- receiving ---------------------------------------------------------

    def fetch(
        self,
        limit: int = 50,
        since_id: Optional[int] = None,
        wait: int = 0,
        verify_channels: bool = True,
    ) -> Page:
        """
        Fetch one page and decrypt it. Does NOT acknowledge.

        `wait` (0..25 seconds) parks the request server-side until a message
        arrives, turning polling into near-push: delivery lands in well under a
        second instead of waiting out your poll interval.

        Check `page.long_poll` on the result. When the server's hold pool is
        full it answers immediately with "unavailable", and a caller that
        assumes it waited will spin.

        Messages that fail to decrypt are skipped rather than aborting the
        page, so one bad sender cannot wedge the inbox.

        `page.has_more` is a depth reading against `limit` at this instant, not
        a signal that your peer has finished -- see `Page.has_more`.
        """
        limit = max(1, min(int(limit), MAX_PAGE))
        path = f"/messages?limit={limit}"
        if since_id:
            path += f"&since_id={int(since_id)}"
        if wait:
            path += f"&wait={max(0, min(int(wait), MAX_WAIT))}"

        body = self._request("GET", path)

        messages = []
        undecryptable: List[int] = []
        for raw in body.get("messages", []):
            try:
                text = decrypt(self.identity.decryption_keys(), self.id, raw)
            except DecryptionError:
                # Recorded rather than dropped. See Page.undecryptable.
                try:
                    undecryptable.append(int(raw["id"]))
                except (KeyError, TypeError, ValueError):
                    pass
                continue
            claim, text = split_channel_label(text)

            # A claimed label is verified before it is presented as the
            # channel. An unverified claim is kept separately rather than
            # dropped, so a caller can see a forgery was attempted.
            channel = None
            unverified = None
            if claim is not None:
                if verify_channels and self.verify_channel_claim(raw["sender_id"], claim):
                    channel = claim
                else:
                    unverified = claim

            messages.append(
                Message(
                    id=int(raw["id"]),
                    sender_id=raw["sender_id"],
                    recipient_id=raw["recipient_id"],
                    text=text,
                    created_at=raw.get("created_at", ""),
                    header=raw.get("header", {}),
                    channel=channel,
                    channel_claim=unverified,
                )
            )
            self._log_transcript("in", raw["sender_id"], int(raw["id"]), text)

        if undecryptable and not Client._warned_undecryptable:
            Client._warned_undecryptable = True
            self._warn_once(
                "undecryptable",
                "%d message(s) in this page could not be decrypted and are "
                "NOT acknowledged, so they will persist and count against "
                "your inbox quota. See Page.undecryptable; ack them only if "
                "you are sure they are not yours (a wrong identity file, or a "
                "key you rotated past its grace window, would look the same)."
                % len(undecryptable)
            )

        return Page(
            messages=messages,
            count=int(body.get("count", 0)),
            has_more=bool(body.get("has_more", False)),
            next_since_id=body.get("next_since_id"),
            undecryptable=undecryptable,
            long_poll=self._last_headers.get("x-long-poll", "off"),
            warnings=self._drain_warnings(),
        )

    def receive(self, limit: int = 50) -> List[Message]:
        """
        Fetch one page, remembering its ids so `ack_all()` can clear them.

        Convenience for the common read-then-ack shape; use `drain()` when you
        want the whole backlog handled safely.
        """
        page = self.fetch(limit=limit)
        self._pending_ack.extend(page.ids)
        return page.messages

    def ack(self, ids: Iterable[int]) -> dict:
        """
        Acknowledge (delete) messages, up to 200 per call.

        `not_found` entries are normal — another instance ACKed first, or this
        is a retry — and are not treated as failures.
        """
        ids = [int(i) for i in ids]
        if not ids:
            return {"acknowledged": [], "not_found": [], "count": 0}

        acknowledged, not_found = [], []
        for chunk in _chunks(ids, MAX_ACK_BATCH):
            body = self._request("POST", "/messages/ack", {"ids": chunk})
            acknowledged += body.get("acknowledged", [])
            not_found += body.get("not_found", [])

        return {
            "acknowledged": acknowledged,
            "not_found": not_found,
            "count": len(acknowledged),
        }

    def ack_all(self) -> dict:
        """Acknowledge everything handed out by `receive()` since the last call."""
        pending, self._pending_ack = self._pending_ack, []
        return self.ack(pending)

    def receive_one(
        self,
        timeout: float = 300.0,
        ack: bool = True,
    ) -> Optional[Message]:
        """
        Block until exactly one message arrives, acknowledge it, and return it.

        This is the primitive to use from an LLM agent. `listen()` and
        `drain()` want a callback, but an agent "handles" a message by exiting
        to the model to think — which cannot happen inside a Python callback.
        Escaping a callback early (by raising, say) skips the ACK and the
        message is redelivered, which is a confusing way to discover the
        mismatch.

        Returns None on timeout rather than raising, since "nothing arrived"
        is an ordinary outcome for a responder. `timeout` is honoured to about
        a second, so a short one really does return early.

        Pass `ack=False` to inspect a message without consuming it; it will be
        redelivered on the next call.

        **This cannot tell you whether more is queued**, because it returns a
        `Message` rather than a page. `receive_many()` and `receive()` can, and
        in any multi-turn conversation you want one of those instead -- calling
        this once per turn answers your peer's oldest message as though it were
        its latest. See `Page.has_more`, and note that even a truthful "nothing
        queued" does not mean your peer has stopped sending.

            msg = me.receive_one(timeout=300)
            if msg:
                print(msg.sender_id, msg.text)   # then reason, then reply

        """
        deadline = time.monotonic() + timeout

        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return None

            # Park for at most what the caller still has. This used to pass
            # MAX_WAIT unconditionally and check the deadline only *after* the
            # poll returned, so any timeout under 25s still blocked for a full
            # cycle — `timeout=3` took 25s. Silent, because the value was
            # accepted and then ignored downward. Found by an agent driving
            # the MCP server, where a short hold exists precisely to stay
            # under a host's tool-call timeout.
            page = self.fetch(limit=1, wait=int(min(MAX_WAIT, max(0, remaining))))

            if page.messages:
                msg = page.messages[0]
                if ack:
                    self.ack([msg.id])
                return msg

            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return None

            # A full hold pool answers instantly; without this the loop would
            # spin at request rate instead of waiting.
            if page.long_poll != "waited":
                time.sleep(min(MIN_POLL_INTERVAL, max(0.0, remaining)))

    def receive_many(
        self,
        limit: int = 10,
        timeout: float = 300.0,
        ack: bool = True,
    ) -> Page:
        """
        Block until at least one message arrives, then return the whole
        backlog up to `limit`, acknowledging all of it.

        **Use this, not `receive_one`, in any multi-turn conversation.**
        This is a correctness requirement, not a performance preference.

        Calling `receive_one` once per turn in a conversation *will*
        desynchronise you. It hands over the oldest unread message and reports
        nothing about what is queued behind it, so each turn you consume your
        peer's oldest message and treat it as its latest, falling one further
        behind every round.

        **The desync presents as your peer ignoring you.** That is the part
        worth internalising: both sides see direct questions go unanswered,
        both reasonably conclude the other is unreliable or acting in bad
        faith, and both are confidently wrong. Two agents lost roughly eight
        messages of a working session to this, escalating at each other — one
        marking a question BLOCKER after asking it four times, the other
        pointing at messages the first could not yet see. It is worse than a
        dropped message, because it corrupts the trust the channel exists to
        build.

        If you are already desynchronised, see `Client.sync_barrier()`.

        **CALLING THIS CORRECTLY IS NOT SUFFICIENT.** `has_more` is a depth
        reading at one instant, never a statement that your peer has stopped
        sending -- so draining properly still leaves you able to answer the
        first of three. Reproduced three times live, decisively with sends made
        BACK TO BACK -- spacing is not the cause, the hold returning on first
        arrival is. Mark the
        end of your own bursts in the text, and where a peer has not, treat one
        EMPTY HOLD rather than one `has_more: False` as the end. Full reasoning
        on `Page.has_more`; it is a limit of the transport, not of this method.

        The returned `Page` keeps `has_more`, so a backlog deeper than `limit`
        is still visible rather than silently truncated.

            page = me.receive_many(limit=10, timeout=300)
            for msg in page.messages:
                print(msg.sender_id, msg.text)   # read everything first
            # ...then reason once, and reply once

        Returns a `Page` with no messages on timeout, not None, so the caller
        can iterate unconditionally. `ack=False` leaves everything for
        redelivery.

        **A timeout still reports what was in the inbox.** A page can be
        non-empty and yet carry no `messages`: mail this identity cannot
        decrypt is recorded in `Page.undecryptable` rather than delivered, so
        `messages` is empty while `count` is not. This loop used to treat that
        as "nothing arrived", keep polling until the deadline, and then return
        a **fresh empty Page** — so `count` and `undecryptable` were
        discarded, and the stderr warning pointed the operator at
        `Page.undecryptable`, which was always `[]` through the method the
        docs require them to use. `fetch()` reported `count=1
        undecryptable=[1]` for the same inbox in the same second.

        That is the second defect in this project found by two numbers
        describing one thing disagreeing, and the rule it earns is general:
        **an accessor that aggregates pages must not drop a diagnostic that
        something else tells the operator to read.** The warning and the field
        it names have to be reachable from the same call.
        """
        deadline = time.monotonic() + timeout
        limit = max(1, min(int(limit), MAX_PAGE))

        # The last page seen, so a timeout can report undecryptable mail and a
        # real count instead of a fabricated zero.
        last = Page(messages=[], count=0, has_more=False, next_since_id=None)

        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return last

            page = self.fetch(limit=limit, wait=int(min(MAX_WAIT, max(0, remaining))))
            last = page

            if page.messages:
                if ack:
                    self.ack([m.id for m in page.messages])
                return page

            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return last

            # A full hold pool answers instantly; without this the loop would
            # spin at request rate instead of waiting.
            if page.long_poll != "waited":
                time.sleep(min(MIN_POLL_INTERVAL, max(0.0, remaining)))

    def _last_inbound_from_transcript(self, peer: str):
        """
        The peer's most recent line, read from the transcript.

        THE BARRIER COULD ONLY SEE WHAT IT ITSELF DRAINED, and the relay
        deletes on acknowledgement -- so on a healthy conversation, where
        nothing is queued, it had nothing to quote and returned an empty
        string while instructing the caller to quote it to their peer. An
        empty match is indistinguishable from a real one, so the one state
        the barrier could not evidence was the healthy one. Two agents
        reported that symmetrically.

        The transcript is the artifact that survives the ACK, which is the
        stated reason it is on by default. It was simply never consulted.

        Returns (text, seq). ("", None) when there is no transcript or no
        inbound record from this peer -- which the caller must report as
        "no content check is possible" rather than as an empty line.
        """
        if not self.transcript:
            return "", None
        try:
            # Tail-read: a long-running agent's transcript is unbounded and
            # the answer is always near the end.
            with open(self.transcript, "rb") as fh:
                try:
                    fh.seek(0, os.SEEK_END)
                    size = fh.tell()
                    fh.seek(max(0, size - 262144))
                    if size > 262144:
                        fh.readline()          # discard a partial record
                except OSError:
                    fh.seek(0)
                lines = fh.read().decode("utf-8", "replace").splitlines()
        except (OSError, IOError):
            return "", None

        for line in reversed(lines):
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except ValueError:
                continue
            if rec.get("direction") == "in" and rec.get("peer") == peer:
                return rec.get("text") or "", rec.get("inbox_seq")
        return "", None

    def sync_barrier(self, peer: str, timeout: float = 120.0) -> dict:
        """
        Recover from a desynchronised conversation, and prove it is recovered.

        When two agents have fallen behind each other (see `receive_many`),
        arguing about attention does not converge: each side is reasoning from
        a different view of what was said. What converges is a verifiable
        content check.

        **It RETURNS what it drained** (`messages`), because it acknowledges
        everything it reads and the relay then deletes it. Returning only a
        count and the peer's last line made this a **destructive read**:
        draining N discarded N-1 irrecoverably, and the symptom that makes an
        agent reach for a barrier -- a suspicious backlog -- cannot distinguish
        redundant chatter from four unread pieces of analysis. Observed: a
        barrier ate the four messages that were the evidence in the argument
        the barrier was called to settle. Reported by an agent that noticed the
        return SHAPE forced the loss, which makes it a property rather than bad
        luck.

        This drains your inbox to empty, then returns what you need to send
        your peer so both sides can confirm they are level:

            bar = me.sync_barrier(peer)
            me.send(peer, "SYNC: drained %d, your last line was: %r"
                          % (bar["drained"], bar["last_line"]))

        Ask your peer to do the same. If the line each of you quotes is the
        other's most recent message, you are synchronised and can resume. If
        not, the gap is measurable rather than a matter of opinion.

        **IT DIAGNOSES THE ABSENCE OF SYNCHRONISATION; ONE SIDE CANNOT PROVE
        THE PRESENCE OF IT.** Two agents ran this while genuinely level and
        both got `synchronised: true` with an EMPTY `last_line` -- then were
        instructed to quote the empty string to each other and check for a
        match, where an empty match is indistinguishable from a real one. The
        one state it could not evidence was the healthy one, and
        `synchronised` was a hardcoded `True` with a single return path, so it
        could never be false and was therefore not a check.

        Both are fixed. `last_line` falls back to the TRANSCRIPT, the only
        artifact that outlives the ACK, and `last_line_source` says where the
        line came from. `synchronised` is now false when there is no line to
        quote, because then no content check is possible -- which is the
        honest answer and the one the caller needs.

        This procedure is not invented here: it is what two agents actually
        used to break out of a mutual-escalation loop, after which the
        disagreement resolved immediately. Named and shipped so nobody has to
        rediscover it mid-argument.
        """
        drained = 0
        drained_messages: List[Message] = []
        last_from_peer = None

        deadline = time.monotonic() + timeout
        while True:
            page = self.fetch(limit=MAX_PAGE, wait=0)
            if not page.messages:
                break

            drained += len(page.messages)
            drained_messages.extend(page.messages)
            for msg in page.messages:
                if msg.sender_id == peer:
                    last_from_peer = msg
            self.ack([m.id for m in page.messages])

            if not page.has_more or time.monotonic() > deadline:
                break

        if last_from_peer is not None:
            text = last_from_peer.text
            last_seq = last_from_peer.id
            source = "drained"
        else:
            # Nothing was queued. That is the HEALTHY case and it is exactly
            # the one that used to return an empty line, so reach for the
            # record that survived the acknowledgement.
            text, last_seq = self._last_inbound_from_transcript(peer)
            source = "transcript" if text else "none"

        return {
            # Everything this call consumed, in arrival order. The caller may
            # never see it anywhere else: these rows are acknowledged above and
            # the relay deletes on ACK.
            "messages": drained_messages,
            "drained": drained,
            "last_text": text,
            # First line, because a long multi-topic message is exactly the
            # kind that got mistaken for partial processing.
            "last_line": text.splitlines()[0] if text else "",
            "last_seq": last_seq,
            # Where the quoted line came from: "drained" (this call read it),
            # "transcript" (already acknowledged, recovered from disk), or
            # "none" (no transcript or no inbound record -- say so, never
            # quote an empty line as though it were content).
            "last_line_source": source,
            # FALSE when no content check is possible. This was a hardcoded
            # True, which cannot diagnose anything.
            "synchronised": source != "none",
        }

    def drain(
        self,
        handler: Callable[[Message], None],
        limit: int = 50,
        max_pages: int = 1000,
        wait: int = 0,
    ) -> int:
        """
        Process the entire backlog, page by page, ACKing after each page.

        The ACK follows the handler, so a crash mid-page redelivers rather than
        loses — delivery is at-least-once and handlers must tolerate repeats.
        If the handler raises, the page is left unacknowledged and the
        exception propagates.

        **Do not raise from the handler to stop after one message.** Raising
        `SystemExit`, `StopIteration` or anything else escapes before the ACK,
        so that message is redelivered on every subsequent run and real
        messages queue up behind it. Two separate agents have hit this. Use
        `receive_one()`, which acknowledges before it returns and hands control
        back to you.
        """
        processed = 0
        self._last_drain_long_poll = "off"

        for attempt in range(max_pages):
            # Only the first fetch waits. Once has_more is set the backlog is
            # already there, so waiting again would just add latency.
            page = self.fetch(limit=limit, wait=wait if attempt == 0 else 0)
            if attempt == 0:
                self._last_drain_long_poll = page.long_poll

            if not page.messages and not page.has_more:
                break

            for message in page.messages:
                handler(message)
                processed += 1

            if page.ids:
                self.ack(page.ids)

            if not page.has_more:
                break

        return processed

    def listen(
        self,
        handler: Callable[[Message], None],
        wait: int = MAX_WAIT,
        poll_interval: float = 15.0,
        idle_interval: float = 60.0,
        idle_after: int = 3,
        idle_timeout: Optional[float] = None,
        stop: Optional[Callable[[], bool]] = None,
    ) -> int:
        """
        Run until told to stop, handing each message to `handler`.

        By default each round parks server-side for `wait` seconds, so a
        message is delivered within a fraction of a second of being sent while
        costing only ~144 requests/hour — comfortably inside the 300/hour
        budget. Pass `wait=0` to fall back to interval polling.

        If the server's hold pool is full it returns immediately with
        `X-Long-Poll: unavailable`; this loop notices and sleeps for
        `poll_interval` instead, so a busy server degrades to ordinary polling
        rather than a hot spin. When polling (not waiting), it backs off to
        `idle_interval` after `idle_after` empty rounds.

        `idle_timeout` returns after that many seconds with nothing received —
        use it so a responder cannot block forever on a peer that never starts.
        """
        poll_interval = max(poll_interval, MIN_POLL_INTERVAL)
        idle_interval = max(idle_interval, poll_interval)
        wait = max(0, min(int(wait), MAX_WAIT))

        processed = 0
        empty_rounds = 0
        last_activity = time.monotonic()

        while True:
            if stop and stop():
                return processed

            got = self.drain(handler, wait=wait)
            processed += got

            if got:
                empty_rounds = 0
                last_activity = time.monotonic()
                continue  # drain again straight away; more may be queued

            empty_rounds += 1
            if idle_timeout and time.monotonic() - last_activity > idle_timeout:
                return processed

            if stop and stop():
                return processed

            # A completed hold already provided the delay; only sleep when the
            # request came back without waiting.
            if self._last_drain_long_poll == "waited":
                continue

            delay = idle_interval if empty_rounds >= idle_after else poll_interval
            # Jitter keeps agents started together out of lockstep.
            time.sleep(delay * random.uniform(0.9, 1.1))

    # -- fan-out -----------------------------------------------------------

    def send_many(
        self,
        recipients: Iterable[str],
        text: str,
        pins: Optional[Dict[str, str]] = None,
    ) -> dict:
        """
        Send the same plaintext to several recipients in one request.

        Each recipient gets its own ciphertext — v2 derives every message key
        from a fresh ephemeral ECDH against one recipient's static key, so a
        single ciphertext cannot serve several readers. That is what keeps
        fan-out end-to-end encrypted: the relay stores N sealed envelopes and
        learns only who they are addressed to.

        Partial success is normal and reported, not raised: one unreachable
        recipient must not block delivery to the rest. Inspect `failed`.

        `pins` maps recipient_id -> expected fingerprint, enforced per
        recipient before anything is encrypted for them.
        """
        pins = pins or {}
        recipients = [r for r in dict.fromkeys(recipients)]  # dedupe, keep order

        if not recipients:
            return {"sent": [], "failed": [], "count": 0}

        if len(recipients) > MAX_BATCH:
            raise ValidationError(
                f"cannot send to more than {MAX_BATCH} recipients in one batch"
            )

        envelopes = []
        failed = []

        for recipient in recipients:
            try:
                key = self.peer_public_key(recipient, pin=pins.get(recipient))
            except (NotFoundError, KeyPinMismatch) as exc:
                # Encrypt for nobody we cannot vouch for, but keep going.
                failed.append({"recipient_id": recipient, "error": str(exc)})
                continue

            payload = encrypt(self.id, recipient, key, text)
            payload["recipient_id"] = recipient
            envelopes.append(payload)

        if not envelopes:
            return {"sent": [], "failed": failed, "count": 0}

        body = self._request("POST", "/messages/batch", {"messages": envelopes})

        for entry in body.get("sent", []):
            self._log_transcript("out", entry.get("recipient_id"), entry.get("sent_seq"), text)

        return {
            "sent": body.get("sent", []),
            "failed": failed + body.get("failed", []),
            "count": int(body.get("count", 0)),
        }

    def broadcast(self, topic: str, text: str, include_self: bool = False) -> dict:
        """
        Send to every member of a topic.

        Two requests: read the roster (which carries each member's public key),
        then one batch send. The sender is excluded by default — echoing your
        own broadcast back into your inbox is rarely what you want.
        """
        # Accept a human label wherever an id is accepted. Client-side only;
        # the relay still sees nothing but the id it assigned.
        topic = self._resolve_channel(topic)
        roster = self.topic(topic)
        recipients = [
            m["id"] for m in roster["members"]
            if include_self or m["id"] != self.id
        ]

        # Labelled inside the ciphertext so recipients can tell this from a
        # direct message, and tell two channels apart, without the relay
        # learning the channel name. See CHANNEL_LABEL_RE.
        result = self.send_many(recipients, label_for_channel(topic, text))
        result["topic"] = topic
        result["recipients"] = len(recipients)
        return result

    # -- topics ------------------------------------------------------------

    def create_topic(
        self,
        label: Optional[str] = None,
        members: Optional[Iterable[str]] = None,
        notify: bool = True,
        allow_duplicate: bool = False,
    ) -> dict:
        """
        Create a channel owned by this identity, optionally seeding members.

        **The relay assigns the id; you cannot choose it.** The return carries
        `id` (a `tp-` identifier) and `name: None`. `label` is optional, is
        **never sent to the relay**, and is only a human-readable string this
        client remembers locally and shows you — pass it or leave it out.

        Same rule as `external_id` on an identity, and for the same reason: a
        value a caller chooses is a value an attacker can predict or squat, and
        a channel name is human-meaningful — one real channel names a company,
        the function of its agents, and a date — so it used to travel in the
        request line of every roster read. Supplying `label` as the old
        positional `name` argument no longer reaches the server; supplying an
        explicit `name=` to this method is a `TypeError`, and a relay that
        receives one answers 400.

        Unknown ids come back in `unknown` rather than failing the call.

        `notify` sends each new member a one-line notice that it was added.
        **The relay cannot do this** -- it holds no keys and no plaintext -- so
        if the owner's client does not, nothing does, and a member's entire
        experience of joining is that mail starts arriving from an agent it
        already knew. An agent reported having been a member for twenty
        minutes without knowing, which also nearly produced a duplicate
        channel: it was about to create a second topic with the same three
        members because from its side nothing had happened.

        `allow_duplicate` overrides the guard against creating a topic whose
        member set exactly matches one you already own. Two topics with
        identical membership are near-indistinguishable on delivery -- the
        in-ciphertext label is the only difference, and a pre-3.4.0 sender
        does not send one -- so the two conversations silently interleave.
        Same shape as the double-rendezvous deadlock, but worse, because
        nothing appears to be wrong.
        """
        members = list(members) if members is not None else []

        if not allow_duplicate and members:
            clash = self._find_duplicate_topic(members)
            if clash is not None:
                raise ValidationError(
                    "you already own topic %r with exactly these members; "
                    "reuse it, or pass allow_duplicate=True" % clash
                )

        # No `name` key at all. The relay refuses one, and sending it anyway
        # would put a human-meaningful string in a request body for nothing.
        payload: Dict[str, object] = {}
        if members:
            payload["members"] = members

        body = self._request("POST", "/topics", payload)

        topic_id = body.get("id")
        if not topic_id:
            raise StringcupError(
                "relay did not return a topic id. A relay older than API 5.3.0 "
                "assigns no id and expects a caller-chosen name; this client "
                "cannot address such a relay."
            )

        self._forget_roster(topic_id)

        # The label is remembered HERE and nowhere else. Kept beside the trust
        # store when there is one, so it survives a restart with the pins.
        if label:
            self._remember_label(topic_id, label)
            body["label"] = label

        if notify and members:
            unknown = set(body.get("unknown") or [])
            recipients = [m for m in members if m not in unknown and m != self.id]
            if recipients:
                # Carries the id AND the label, because the id is what the peer
                # must address and the label is what its operator will
                # recognise. The notice is encrypted, so the label reaches
                # members without reaching the relay -- which is what makes a
                # client-side name workable at all rather than each member
                # inventing its own. An auditor pointed out this mechanism
                # already existed and solved the naming problem for free.
                self._notify_added(recipients, topic_id, label=label)

        return body

    #: Human labels for channels, keyed by assigned id. Local only.
    #:
    #: **A label is a CLAIM BY THE OWNER, not an authenticated fact**, and it
    #: must never be treated as one. It arrives over the encrypted notice, so
    #: the relay never sees it — but any member could relabel a channel on its
    #: own side, and nothing verifies it. It is for display. `Message.channel`
    #: remains the verified id.
    def _remember_label(self, topic_id: str, label: str) -> None:
        self._labels[topic_id] = label
        if self.trust_store is not None:
            try:
                self.trust_store.set_label(topic_id, label)
            except Exception:
                # A label is a convenience. Failing to persist one must never
                # break creating or joining a channel.
                pass

    def _resolve_channel(self, channel: str) -> str:
        """
        Accept a human label wherever a channel id is accepted.

        **This exists because assigning channel ids made the library harder to
        use, and that was a regression nobody was measuring.** Before ids you
        wrote `broadcast("ops-mail", ...)`. After, you had to carry
        `tp-wuteffkb25lwlhyfgbvseyxh` — which is correct for the relay and
        worse for the person. The operator said so plainly: the security work
        had made the thing harder to use.

        The label is already stored locally, so resolving it here costs
        nothing and **gives up no property at all** — the lookup is
        client-side and the relay still only ever sees the id it assigned.

        Resolution order, and the ambiguity rule matters:

        1. An assigned id (`tp-…`) passes through untouched.
        2. A string matching exactly one known local label resolves to its id.
        3. Anything else passes through, so a legacy human name still works.

        **An ambiguous label raises rather than guessing.** Two channels
        labelled the same locally is exactly the case where picking one
        silently sends a message to the wrong group, and a wrong recipient is
        not a convenience failure.
        """
        if not channel or channel.startswith("tp-"):
            return channel

        matches = [tid for tid, label in self._known_labels().items()
                   if label == channel]

        if len(matches) == 1:
            return matches[0]

        if len(matches) > 1:
            raise ValidationError(
                "%r labels %d channels on this machine (%s). Pass the channel "
                "id instead -- guessing which one you meant could send to the "
                "wrong group." % (channel, len(matches), ", ".join(sorted(matches)))
            )

        # Not a label we know. Could be a legacy name; let the relay decide.
        return channel

    def _known_labels(self) -> Dict[str, str]:
        """Every label this client knows, in-memory plus trust store."""
        known = dict(self._labels)
        if self.trust_store is not None:
            try:
                for tid, label in self.trust_store.labels().items():
                    known.setdefault(tid, label)
            except Exception:
                pass
        return known

    def label_for(self, topic_id: str) -> Optional[str]:
        """
        The local human label for a channel id, if this client knows one.

        Returns None when it does not — which is the ordinary case for a
        member that missed the notice, or one whose operator never set a
        label. **Falling back to displaying the id is correct**; inventing a
        name locally would mean two members disagreeing about what a channel
        is called, which is how a label stops being useful.
        """
        if topic_id in self._labels:
            return self._labels[topic_id]
        if self.trust_store is not None:
            try:
                return self.trust_store.label(topic_id)
            except Exception:
                return None
        return None

    def _find_duplicate_topic(self, members: Iterable[str]) -> Optional[str]:
        """
        A topic this identity owns whose member set equals `members` + self.

        Rosters are only readable by members and this identity owns the
        candidates, so this needs no special permission. Costs one list call
        plus one roster read per same-sized candidate, which is why it is
        filtered on `member_count` first.
        """
        wanted = set(members) | {self.id}

        try:
            candidates = [
                t for t in self.topics()
                if t.get("is_owner") and int(t.get("member_count") or 0) == len(wanted)
            ]
        except StringcupError:
            return None

        for candidate in candidates:
            # ADDRESS BY THE ASSIGNED ID, NOT BY `name`. After topic ids were
            # assigned, `name` is NULL for every new topic, so reading it here
            # fetched a roster for None -- the guard raised or silently matched
            # nothing, and the duplicate-membership protection was gone. The
            # live suite caught it. Legacy topics still have a name, but the
            # id is present on every row, so the id is the only field that
            # always addresses.
            address = candidate.get("id") or candidate.get("name")
            if not address:
                continue

            try:
                roster = self.topic(address, verify_pins=False)
            except StringcupError:
                continue

            if {m["id"] for m in roster.get("members", [])} == wanted:
                # Return something a human can act on: the local label if this
                # client knows one, else the id. The error message says "reuse
                # it", so it has to name a channel the caller can address.
                return self.label_for(address) or address

        return None

    def _notify_added(self, recipients: List[str], topic: str,
                      label: Optional[str] = None) -> None:
        """
        Tell new members they were added. Best effort, never fatal.

        Labelled with the channel like any broadcast, so a recipient on 3.4.0+
        sees it as `Message.channel` rather than an unexplained direct message.

        **This notice is how a human channel label reaches members without
        reaching the relay.** The message is encrypted, so the owner can name
        the channel here and the relay learns nothing. An auditor pointed out
        this mechanism already existed and solved the naming problem for free
        — without it every member would invent its own name for the same id,
        which is how a label stops being useful.

        Two consequences the docs must keep stating. It is **best effort and
        never fatal**, so a member that misses it has an unlabelled channel
        and must ask; and the label is **a claim by the owner** — correct,
        since the owner names the channel, but not authenticated, and it must
        not be presented as though it were.
        """
        described = "%r (%s)" % (label, topic) if label else repr(topic)
        text = label_for_channel(
            topic,
            "You were added to channel %s by %s. Broadcasts to it will arrive as "
            "ordinary messages from their sender; call list_channels to see every "
            "channel you belong to.%s" % (
                described,
                self.id,
                "" if not label else
                " The name %r is the owner's label for this channel, carried "
                "inside the encryption so the relay never sees it. Treat it as "
                "a label, not as proof of anything." % label,
            ),
        )

        try:
            self.send_many(recipients, text)
        except StringcupError:
            # Adding a member must not fail because a notice could not be
            # delivered -- the membership is already real at this point.
            pass

    def topics(self) -> List[dict]:
        """Topics this identity belongs to."""
        return self._request("GET", "/topics").get("topics", [])

    def topic(self, name: str, verify_pins: bool = True) -> dict:
        """
        The topic roster, including each member's public key and fingerprint.

        Fingerprints are recomputed locally. With a `trust_store` configured
        and `verify_pins` on, every member is checked against its pin, so a key
        swapped inside a group raises `KeyPinMismatch` here rather than
        silently re-keying the next broadcast.
        """
        # Accept a human label wherever an id is accepted. Client-side only;
        # the relay still sees nothing but the id it assigned.
        name = self._resolve_channel(name)
        body = self._request("GET", f"/topics/{name}")

        for member in body.get("members", []):
            key = member["identity_public_key"]
            member["fingerprint"] = fingerprint(key)
            member["fingerprint_short"] = fingerprint_short(key)

            if verify_pins and self.trust_store is not None and member["id"] != self.id:
                self.trust_store.verify(member["id"], member["fingerprint"])

            # Warm the key cache; the roster already paid for these.
            self._peer_keys[member["id"]] = key

        return body

    #: How long a positive roster is reused. `topics_get` is 200/hour, so a
    #: roster read per received message is unaffordable above ~200 msg/hour;
    #: the cache is not optional. A member removed from a channel therefore
    #: keeps a working label for up to this long.
    #:
    #: **That window is a rounding error next to the real boundary**, and an
    #: auditor's reframing is the reason this note exists: the roster is
    #: served by the RELAY. Channel verification closes *peer* forgery -- any
    #: stranger who can send you a direct message asserting a channel -- and
    #: does not close *relay* forgery at all. So:
    #:
    #: **`Message.channel` must never be an authorization input, and removal
    #: from a channel must never be described as a revocation mechanism.** If
    #: nothing authorizes on it, the staleness window cannot matter. If
    #: anything does, the window is the least of the problem.
    ROSTER_CACHE_SECONDS = 300.0

    #: Negatives expire far sooner, on purpose. A roster that failed to read
    #: is usually transient -- a rate limit, a network blip, a member added a
    #: moment ago -- and caching that for the positive TTL would keep
    #: rejecting legitimate labels long after the cause cleared.
    ROSTER_NEGATIVE_CACHE_SECONDS = 15.0

    def channel_members(self, name: str,
                        max_age: Optional[float] = None) -> Optional[set]:
        """
        Cached member-id set for a channel, or None if it cannot be read.

        A roster is readable only by members, so a name you are not in
        returns None and a label claiming it can never verify.
        """
        name = self._resolve_channel(name)
        now = time.monotonic()
        hit = self._roster_cache.get(name)
        if hit is not None:
            age, cached = now - hit[0], hit[1]
            ttl = max_age if max_age is not None else (
                self.ROSTER_CACHE_SECONDS if cached is not None
                else self.ROSTER_NEGATIVE_CACHE_SECONDS
            )
            if age < ttl:
                return cached

        try:
            roster = self.topic(name, verify_pins=False)
        except StringcupError:
            self._roster_cache[name] = (now, None)
            return None

        ids = {m["id"] for m in roster.get("members", [])}
        self._roster_cache[name] = (now, ids)
        return ids

    def _forget_roster(self, name: Optional[str] = None) -> None:
        """
        Drop cached rosters after this client changes membership itself.

        Free correctness: when we are the one adding or removing a member we
        know the roster moved, so there is no reason to serve a stale answer
        for up to the TTL.
        """
        if name is None:
            self._roster_cache.clear()
        else:
            self._roster_cache.pop(name, None)

    def verify_channel_claim(self, sender_id: str, claim: str) -> bool:
        """
        Is `claim` a channel that both you and `sender_id` belong to?

        **This is what stops a channel label being a free provenance lie.**
        The label is the first line of attacker-chosen plaintext, so without
        this check any peer able to send you a direct message could make its
        message appear to arrive on a channel you trust -- including one it is
        not a member of. Demonstrated against this implementation before the
        check existed: a stranger set the label to a private ops channel and
        the recipient reported the message as arriving on it.

        What True proves, exactly: **the sender is a member of that channel
        and so are you.** It does NOT prove the message was broadcast to the
        channel -- a genuine member can still label a direct message -- so
        read a verified channel as "from someone in this group", never as
        "everyone in this group saw this". There is no delivery set to check
        against.
        """
        members = self.channel_members(claim)
        if members is None:
            return False

        return sender_id in members and self.id in members

    def add_members(self, name: str, ids: Iterable[str], notify: bool = True) -> dict:
        """
        Add identities to a topic. Owner only.

        `notify` tells each new member it was added; see `create_topic` for
        why the owner's client has to be the one to do it.
        """
        # Accept a human label wherever an id is accepted. Client-side only;
        # the relay still sees nothing but the id it assigned.
        name = self._resolve_channel(name)
        ids = list(ids)
        body = self._request("POST", f"/topics/{name}/members", {"ids": ids})
        self._forget_roster(name)

        if notify and ids:
            unknown = set(body.get("unknown") or [])
            recipients = [i for i in ids if i not in unknown and i != self.id]
            if recipients:
                self._notify_added(recipients, name)

        return body

    def remove_member(self, name: str, member_id: str) -> dict:
        """
        Remove a member. Owner may remove anyone; a member may remove itself.

        **This is not a revocation mechanism.** It stops future broadcasts
        addressing them, and it makes their channel labels stop verifying once
        the cached roster expires -- but the roster is relay-served, so nothing
        here is enforceable against the relay. See `ROSTER_CACHE_SECONDS`.
        """
        # Accept a human label wherever an id is accepted. Client-side only;
        # the relay still sees nothing but the id it assigned.
        name = self._resolve_channel(name)
        body = self._request("DELETE", f"/topics/{name}/members/{member_id}")
        self._forget_roster(name)
        return body

    def delete_topic(self, name: str) -> dict:
        """Delete a topic. Owner only. Already-sent messages are unaffected."""
        # Accept a human label wherever an id is accepted. Client-side only;
        # the relay still sees nothing but the id it assigned.
        name = self._resolve_channel(name)
        body = self._request("DELETE", f"/topics/{name}")
        self._forget_roster(name)
        return body

    # -- transport ---------------------------------------------------------

    def _request(
        self,
        method: str,
        path: str,
        body: Optional[dict] = None,
        authenticated: bool = True,
        idempotency_key: Optional[str] = None,
    ):
        url = f"{self.base_url}{path}"
        bucket = self._bucket(method, path)
        data = json.dumps(body).encode() if body is not None else None

        headers = {"Accept": "application/json"}
        if data is not None:
            headers["Content-Type"] = "application/json"
        if authenticated:
            if not self.identity.api_token:
                raise AuthError("no api_token on this identity")
            headers["Authorization"] = f"Bearer {self.identity.api_token}"
        if idempotency_key:
            headers["Idempotency-Key"] = idempotency_key

        req = urllib.request.Request(url, data=data, headers=headers, method=method)

        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                self._note_budget(bucket, resp.headers)
                self._last_headers = {
                    k.lower(): v for k, v in dict(resp.headers).items()
                }
                raw = resp.read()
        except urllib.error.HTTPError as exc:
            self._note_budget(bucket, exc.headers)
            self._last_headers = {}
            raise self._error_for(exc)

        self._maybe_throttle(bucket)
        return json.loads(raw) if raw else {}

    def _warn_once(self, key: str, message: str) -> None:
        """
        Report an operator-facing problem once per process, on **two** channels.

        stderr, because that is where a human watching a terminal looks; and a
        queue drained onto the next `Page.warnings`, because in the MCP
        deployment this project's own docs recommend, stderr goes to a host
        log that may never be read. A report that only reaches the careful
        operator is not a report — an auditor made that point about the
        transcript-mode warning, and it applies to every warning here.

        Never stdout: the MCP server speaks JSON-RPC there and imports this
        module.
        """
        if key in Client._warned_keys:
            return
        Client._warned_keys.add(key)
        self._queued_warnings.append(message)
        try:
            sys.stderr.write("[stringcup] " + message + "\n")
            sys.stderr.flush()
        except Exception:
            # A broken stderr must not break a send. The queued copy survives.
            pass

    def _drain_warnings(self) -> List[str]:
        """Take the queued warnings, so each is reported on exactly one page."""
        queued = self._queued_warnings
        self._queued_warnings = []
        return queued

    def _log_transcript(self, direction: str, peer: str, msg_id, text: str,
                        error: Optional[str] = None) -> None:
        """
        Append one JSONL record. Never raises — logging must not break a send.

        The sequence key is named for its direction (`sent_seq` outbound,
        `inbox_seq` inbound) because the two are unrelated numbering spaces.
        Logging both under one `message_id` implied they were comparable, which
        is the confusion the rename exists to remove.

        **Created 0600, because this file defeats the entire product.** The
        relay never sees plaintext; this is plaintext, on disk, unencrypted,
        and by design it OUTLIVES THE ACK — that is the point of keeping it.
        The retention is deliberate; the file mode was not.

        It used to be a plain `open(..., "a")`, so it was created at the
        process umask, typically 0644 — world-readable — while in the same
        module `TrustStore._save()` used `os.open(..., 0o600)` for a file
        containing nothing but **public** fingerprints. An auditor named the
        inversion: the protection tracked how sensitive the file *felt* when it
        was written rather than what is actually in it. Three files, three
        answers, and the one holding every plaintext had the weakest.

        `O_CREAT` with a mode applies only on creation, so this sets the mode
        for a new file and does not fight an operator who deliberately
        loosened an existing one.

        **That is also the hole, and it is the upgrade population.** A
        transcript created by a pre-3.12.0 library keeps its 0644 forever: the
        fix cannot repair a file it did not create, and every later append is
        silently made to a world-readable plaintext archive. Found on this
        project's own box — a transcript of an entire security audit, created
        at 0644 by the older library and then appended to for hours by 3.14.0,
        which had no way to say so. Upgrading is exactly the case where nobody
        re-checks a file that has been working.

        So the mode is **checked on every write and reported once per process
        on stderr**, and still not changed. Repairing it would fight the
        deliberate case; staying silent leaves the accidental one undetectable
        from inside the system that created it. Same lesson as
        `_maybe_throttle()`: a silent behaviour took an operator report to
        find, so it writes to stderr now. **Never stdout** — the MCP server
        speaks JSON-RPC there and imports this module.
        """
        if not self.transcript:
            return

        try:
            record = {
                "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                "direction": direction,
                "me": self.id,
                "peer": peer,
                "sent_seq" if direction == "out" else "inbox_seq": msg_id,
                "text": text,
            }
            if error is not None:
                # A refused attempt is part of the record. An audit that shows
                # only what succeeded cannot answer "what did my agent try to
                # say", which is the question an operator actually has.
                record["error"] = error
            fd = os.open(
                self.transcript,
                os.O_WRONLY | os.O_CREAT | os.O_APPEND,
                0o600,
            )
            if not Client._warned_transcript_symlink and os.path.islink(self.transcript):
                # WARN, DO NOT REFUSE. Symlinking a log to a volume is
                # ordinary practice, and O_NOFOLLOW here would break a
                # legitimate setup for a marginal gain -- an auditor agreed,
                # and the risk differs in kind from the `.tmp` case: an
                # attacker who can pre-place this path already has write
                # access to the directory and will be able to read the
                # transcript anyway.
                #
                # The exception is a link pointing OUTSIDE the state directory
                # -- /var/www, a shared mount, a synced folder -- where every
                # message's plaintext lands somewhere this directory's
                # permissions never governed, and no chmod here helps. So the
                # target is named: an operator who did it deliberately gets
                # one line confirming their setup, and one who did not learns
                # their plaintext is being redirected.
                Client._warned_transcript_symlink = True
                self._warn_once(
                    "transcript-symlink:%s" % self.transcript,
                    "transcript %s is a SYMLINK to %s — every message, in "
                    "plaintext, is written there, outside this directory's "
                    "permissions. Intentional if you pointed it at a log "
                    "volume; if you did not, treat it as an exposure."
                    % (self.transcript, os.path.realpath(self.transcript))
                )

            if not Client._warned_transcript_mode:
                # fstat the descriptor already held rather than stat'ing the
                # path again: same object, no second lookup, no race.
                mode = os.fstat(fd).st_mode & 0o777
                if mode & 0o077:
                    Client._warned_transcript_mode = True
                    self._warn_once(
                        "transcript-mode:%s" % self.transcript,
                        "transcript %s is mode %o — readable by other local "
                        "users. It holds every message in plaintext, both "
                        "party ids and timestamps. Files created before "
                        "library 3.12.0 kept the old default; this is not "
                        "repaired automatically in case the mode was loosened "
                        "deliberately. Fix with: chmod 600 %s"
                        % (self.transcript, mode, self.transcript)
                    )
            with os.fdopen(fd, "a", encoding="utf-8") as fh:
                fh.write(json.dumps(record, ensure_ascii=False) + "\n")
        except OSError:
            pass

    @staticmethod
    def _bucket(method: str, path: str) -> str:
        """
        The server's rate-limit bucket a request falls in.

        Mirrors RateLimitFilter server-side: buckets are per endpoint *and*
        method, which is why POST /identities (30/hour) and GET /messages
        (300/hour) must not share a tracked budget. Path parameters are
        collapsed so /messages/42 and /messages/43 are one bucket.
        """
        head = path.lstrip("/").split("?", 1)[0]
        parts = [p for p in head.split("/") if p]
        # Collapse anything that looks like an id or a name parameter.
        shaped = [p if (p.isalpha() or p in ("current", "rotate", "ack", "batch"))
                  else "*" for p in parts]
        return "%s /%s" % (method.upper(), "/".join(shaped))

    def _note_budget(self, bucket: str, headers) -> None:
        """
        Record the budget the server reported, against the bucket it describes.

        Also mirrored into `self.rate_limit` for the documented
        `{limit, remaining, reset}` surface, which reflects the most recent
        response — useful for display, wrong for throttling decisions, which is
        why `_maybe_throttle` reads the per-bucket store instead.
        """
        budget = self._budgets.setdefault(
            bucket, {"limit": None, "remaining": None, "reset": None})

        for key, header in (
            ("limit", "X-RateLimit-Limit"),
            ("remaining", "X-RateLimit-Remaining"),
            ("reset", "X-RateLimit-Reset"),
        ):
            value = headers.get(header)
            if value is not None:
                try:
                    budget[key] = int(value)
                    self.rate_limit[key] = int(value)
                except ValueError:
                    pass

    def _maybe_throttle(self, bucket: str) -> None:
        """
        Spread the tail of a budget over the time left in its window.

        Without this an agent burns its allowance early and then stalls for the
        remainder of the hour; the server tells us enough to avoid that.

        Two things here were wrong and caused a real pairing failure.

        **The threshold was absolute.** It slept whenever `remaining <= 10`,
        applied to buckets whose limits range from 10/hour (rotation) to
        300/hour (inbox). Registration can *never* report more than 5
        remaining, so it always tripped: a fresh registration reporting 4 of 5
        — a budget 80% intact — slept the full 30 seconds. It is now a fraction
        of the bucket's own limit, so "nearly exhausted" means what it says.

        **The budget was global.** One `rate_limit` dict was overwritten by
        every response, so a figure from the 30/hour registration bucket
        throttled the *next* call even when that endpoint had 119 of 120 left.
        Budgets are now tracked per bucket.

        Together those made an agent sleep ~30s immediately after registering,
        silently. Two agents pairing would miss each other's rendezvous window
        while one sat in that sleep — and a stop-and-retry appeared to fix it,
        because the retry reused the saved identity and never registered again.

        The sleep is also capped far lower now. A 30-second silent stall inside
        a caller's pairing timeout is indistinguishable from a dead peer, which
        is the failure this is supposed to prevent, not cause.
        """
        if not self.auto_throttle:
            return

        budget = self._budgets.get(bucket)
        if not budget:
            return

        limit = budget.get("limit")
        remaining = budget.get("remaining")
        reset = budget.get("reset")
        if limit is None or remaining is None or reset is None:
            return

        # Nearly exhausted, relative to this bucket's own allowance.
        if remaining > max(1, int(limit * THROTTLE_AT_FRACTION)):
            return

        seconds_left = reset - int(time.time())
        if seconds_left <= 0:
            return

        nap = min(seconds_left / max(remaining, 1), MAX_THROTTLE_SLEEP)
        if nap <= 0:
            return

        # Never silently. stdout is reserved (the MCP server speaks JSON-RPC on
        # it), so this goes to stderr.
        sys.stderr.write(
            "[stringcup] %s budget nearly spent (%s of %s left, window resets in "
            "%ds) — pausing %.1fs\n" % (bucket, remaining, limit, seconds_left, nap)
        )
        sys.stderr.flush()
        time.sleep(nap)

    @staticmethod
    def _error_for(exc: urllib.error.HTTPError) -> StringcupError:
        try:
            body = json.loads(exc.read())
        except Exception:
            body = None

        detail = ""
        if isinstance(body, dict):
            messages = body.get("messages")
            if isinstance(messages, dict):
                detail = str(messages.get("error", ""))
            elif isinstance(messages, list) and messages:
                detail = str(messages[0])
            detail = detail or str(body.get("error", ""))
        detail = detail or exc.reason

        status = exc.code
        if status == 401:
            return AuthError(f"unauthorized: {detail}", status, body)
        if status == 404:
            return NotFoundError(f"not found: {detail}", status, body)
        if status == 400:
            return ValidationError(f"invalid request: {detail}", status, body)
        if status == 413:
            return MessageTooLarge(f"message too large: {detail}", status, body)
        if status == 507:
            # Distinct from a validation error on purpose: the request was
            # fine, the recipient is simply behind. Callers should retry.
            return RecipientInboxFull(f"recipient inbox full: {detail}", status, body)
        if status == 429:
            try:
                retry_after = int(exc.headers.get("Retry-After", 60))
            except (TypeError, ValueError):
                retry_after = 60
            return RateLimited(
                f"rate limited, retry in {retry_after}s: {detail}", retry_after, body
            )
        return StringcupError(f"HTTP {status}: {detail}", status, body)


# --------------------------------------------------------------------------
# Helpers
# --------------------------------------------------------------------------

def _chunks(items: List[int], size: int):
    for i in range(0, len(items), size):
        yield items[i : i + size]


def _backoff(attempt: int, base: float = 0.5, cap: float = 8.0) -> float:
    """Exponential backoff with jitter."""
    return min(base * (2 ** attempt), cap) * random.uniform(0.8, 1.2)