mesh: HTTP gossip wire — signed envelopes + 5 message types

Implements the protocol contract pinned in docs/mesh.md. Each peer
runs a stdlib ThreadingHTTPServer; clients send Ed25519-signed
envelopes carrying one of:

  ANNOUNCE_ROOT           document_root + source_uri + versions
  ANNOUNCE_DERIVATION     core_root <- surface_roots (proof-blob hash)
  ANNOUNCE_PROVIDENCE     cache_key + audit_mode + answer_hash
  ANNOUNCE_FALSIFICATION  cache_key + reason
  REQUEST_BODY/DELIVER_BODY  pull-on-miss + Merkle-root verify

WireEnvelope canonicalizes via sorted-key separator-tight JSON; the
canonical bytes are what get signed. Receivers verify the signature
against sender_id's sign_pub looked up in mesh_roster at the
envelope's epoch_id — non-members of that epoch produce no valid
signature, so they're silently rejected (401, no audit event).

Each accepted ANNOUNCE_* writes one 'mesh_received' event into the
local audit chain whose body is the full signed envelope (the
remote's signature stays attached for non-repudiation). The
receiver's chain stays internally consistent because we append in
receive-order.

REQUEST_BODY/DELIVER_BODY: the responder ships the document text
plus per-chunk leaf hashes; the client re-derives the Merkle root
from those leaves and refuses to return any body whose leaves
don't reconstruct the requested root.

Tests: 17 unit (envelope canonicalization, sig verify against
roster, type validation, tampered-body rejection, audit-event
write) + 5 end-to-end (two real HTTP peers on ephemeral ports
exchanging announces and bodies). 22/22 green.

Deferred to later commits, per docs/mesh.md:
- CLI verbs (mesh serve, mesh sync) — pending downstream merge
- per-peer chain-of-claims tracking (catchup, replay, fork detection)
- optional AEAD encryption of message bodies (contract says optional)

aborist/mesh/wire.py

@@ -0,0 +1,538 @@
+"""HTTP gossip wire — minimum viable.
+
+Implements the protocol contract pinned in ``docs/mesh.md``:
+
+  ANNOUNCE_ROOT          peer says "I have this document_root"
+  ANNOUNCE_DERIVATION    peer says "core C derives from surfaces S1..Sn"
+  ANNOUNCE_PROVIDENCE    peer says "I cached an answer at this 8-dim key"
+  ANNOUNCE_FALSIFICATION peer says "this cache_key is wrong"
+  REQUEST_BODY           peer asks for a document body on local miss
+  DELIVER_BODY           peer responds with bytes + Merkle proof
+
+Every envelope is signed (Ed25519) by the sender's `sign_priv`; receiver
+verifies against the sender's `sign_pub` looked up in `mesh_roster` at
+the envelope's `epoch_id`. A non-member of that epoch cannot produce a
+valid signature, so non-members are silently rejected.
+
+Each receive writes a `mesh_received` event into the *local* audit
+chain whose body is the full signed envelope. The receiver's chain
+stays internally consistent because we append in receive-order. The
+sender's chain integrity is verified by signature alone at this layer
+— per-peer chain-of-claims tracking (catchup, replay, fork detection)
+is a v2 concern (see ``docs/mesh.md``).
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import threading
+import time
+from dataclasses import asdict, dataclass
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+
+import httpx
+
+from aborist.mesh.crypto import sign, verify
+from aborist.mesh.state import (
+    current_epoch,
+    load_identity,
+    roster_at,
+)
+from aborist.store import append_audit, connect
+
+
+# ---------------------------------------------------------------------------
+# Wire constants
+# ---------------------------------------------------------------------------
+
+WIRE_VERSION = 1
+
+TYPE_ANNOUNCE_ROOT = "ANNOUNCE_ROOT"
+TYPE_ANNOUNCE_DERIVATION = "ANNOUNCE_DERIVATION"
+TYPE_ANNOUNCE_PROVIDENCE = "ANNOUNCE_PROVIDENCE"
+TYPE_ANNOUNCE_FALSIFICATION = "ANNOUNCE_FALSIFICATION"
+TYPE_REQUEST_BODY = "REQUEST_BODY"
+TYPE_DELIVER_BODY = "DELIVER_BODY"
+
+ALL_TYPES = frozenset({
+    TYPE_ANNOUNCE_ROOT,
+    TYPE_ANNOUNCE_DERIVATION,
+    TYPE_ANNOUNCE_PROVIDENCE,
+    TYPE_ANNOUNCE_FALSIFICATION,
+    TYPE_REQUEST_BODY,
+    TYPE_DELIVER_BODY,
+})
+
+# HTTP paths
+PATH_ANNOUNCE = "/mesh/announce"
+PATH_REQUEST = "/mesh/request"
+PATH_INFO = "/mesh/info"
+
+DEFAULT_PORT = 8400
+
+
+# ---------------------------------------------------------------------------
+# Envelope: the signed unit of gossip
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class WireEnvelope:
+    """One signed gossip message.
+
+    `body` is type-specific; canonicalization is via JSON sorted-keys with
+    no whitespace, identical to `aborist.store._canonical_json`.
+    """
+
+    type: str
+    sender_id: str
+    epoch_id: int
+    ts: int
+    body: dict[str, Any]
+    v: int = WIRE_VERSION
+
+    def canonical_bytes(self) -> bytes:
+        """Bytes that get signed. Stable, reproducible."""
+        return _canonical_json(asdict(self)).encode("utf-8")
+
+    def sign_with(self, sign_priv: bytes) -> dict[str, Any]:
+        """Wrap envelope + Ed25519 signature in a dict ready for HTTP body."""
+        sig = sign(sign_priv, self.canonical_bytes())
+        return {
+            "envelope": asdict(self),
+            "sig_b64": base64.b64encode(sig).decode("ascii"),
+        }
+
+    @classmethod
+    def from_signed(cls, signed: dict[str, Any]) -> tuple["WireEnvelope", bytes]:
+        """Reverse of `sign_with`. Returns (envelope, sig_bytes).
+
+        Does NOT verify — caller verifies via `verify_envelope_sig`.
+        """
+        if not isinstance(signed, dict) or "envelope" not in signed or "sig_b64" not in signed:
+            raise ValueError("signed payload missing envelope/sig_b64")
+        env_d = signed["envelope"]
+        if not isinstance(env_d, dict):
+            raise ValueError("envelope is not a dict")
+        for required in ("type", "sender_id", "epoch_id", "ts", "body", "v"):
+            if required not in env_d:
+                raise ValueError(f"envelope missing field: {required}")
+        if env_d["type"] not in ALL_TYPES:
+            raise ValueError(f"unknown envelope type: {env_d['type']!r}")
+        if env_d["v"] != WIRE_VERSION:
+            raise ValueError(f"unsupported wire version: {env_d['v']!r}")
+        env = cls(
+            type=env_d["type"],
+            sender_id=env_d["sender_id"],
+            epoch_id=int(env_d["epoch_id"]),
+            ts=int(env_d["ts"]),
+            body=env_d["body"],
+            v=int(env_d["v"]),
+        )
+        try:
+            sig = base64.b64decode(signed["sig_b64"], validate=True)
+        except Exception as e:
+            raise ValueError(f"sig_b64 is not valid base64: {e!r}") from e
+        return env, sig
+
+
+def verify_envelope_sig(
+    conn,
+    envelope: WireEnvelope,
+    sig: bytes,
+) -> bool:
+    """True iff `sig` is a valid Ed25519 signature over the envelope's
+    canonical bytes by `envelope.sender_id`'s `sign_pub` at
+    `envelope.epoch_id` in the local roster.
+
+    Returns False (never raises) on missing roster entry, bad signature,
+    or any structural issue. The caller treats False as "drop the message."
+    """
+    try:
+        roster = roster_at(conn, envelope.epoch_id)
+    except Exception:
+        return False
+    sender = next((m for m in roster if m.member_id == envelope.sender_id), None)
+    if sender is None:
+        return False
+    return verify(sender.sign_pub, sig, envelope.canonical_bytes())
+
+
+# ---------------------------------------------------------------------------
+# Server
+# ---------------------------------------------------------------------------
+
+
+def _make_handler_class(mesh_server: "MeshWireServer") -> type[BaseHTTPRequestHandler]:
+    """Build a request-handler class bound to this MeshWireServer.
+
+    Done as a factory so the handler can reach back to the mesh server
+    without colliding with BaseHTTPRequestHandler's own ``self.server``
+    attribute (which is the underlying ThreadingHTTPServer).
+    """
+
+    class Handler(BaseHTTPRequestHandler):
+        # Bound at class level by the factory closure.
+        mesh: MeshWireServer = mesh_server
+
+        def log_message(self, fmt, *args):  # silence default stderr spam
+            return
+
+        def _read_json(self) -> dict[str, Any]:
+            n = int(self.headers.get("content-length", "0"))
+            raw = self.rfile.read(n) if n > 0 else b""
+            return json.loads(raw or b"{}")
+
+        def _respond(self, status: int, body: dict[str, Any]) -> None:
+            data = json.dumps(body).encode("utf-8")
+            self.send_response(status)
+            self.send_header("content-type", "application/json")
+            self.send_header("content-length", str(len(data)))
+            self.end_headers()
+            self.wfile.write(data)
+
+        def do_GET(self) -> None:
+            if self.path == PATH_INFO:
+                self._respond(200, self.mesh.info())
+                return
+            self._respond(404, {"error": "not found"})
+
+        def do_POST(self) -> None:
+            try:
+                payload = self._read_json()
+            except Exception as e:
+                self._respond(400, {"error": f"bad json: {e!s}"})
+                return
+            if self.path == PATH_ANNOUNCE:
+                self._respond(*self.mesh.handle_announce(payload))
+                return
+            if self.path == PATH_REQUEST:
+                self._respond(*self.mesh.handle_request(payload))
+                return
+            self._respond(404, {"error": "not found"})
+
+    return Handler
+
+
+class MeshWireServer:
+    """Threaded HTTP gossip server.
+
+    Use as ``with MeshWireServer(db_path, host, port) as srv: srv.serve()`` for
+    one-shot scripts, or call ``start_in_thread()`` for tests / long-lived
+    daemons. ``stop()`` is idempotent.
+    """
+
+    def __init__(self, db_path: Path | str, host: str = "127.0.0.1", port: int = DEFAULT_PORT):
+        self.db_path = str(db_path)
+        handler_cls = _make_handler_class(self)
+        self._http = ThreadingHTTPServer((host, port), handler_cls)
+        self._thread: threading.Thread | None = None
+
+    # ------------------------------------------------------------------ context
+    def __enter__(self) -> "MeshWireServer":
+        return self
+
+    def __exit__(self, *_a) -> None:
+        self.stop()
+
+    # ------------------------------------------------------------------ lifecycle
+    @property
+    def address(self) -> tuple[str, int]:
+        return self._http.server_address  # type: ignore[return-value]
+
+    @property
+    def url(self) -> str:
+        host, port = self.address
+        return f"http://{host}:{port}"
+
+    def start_in_thread(self) -> None:
+        if self._thread is not None:
+            return
+        t = threading.Thread(target=self._http.serve_forever, daemon=True)
+        t.start()
+        self._thread = t
+
+    def serve(self) -> None:
+        self._http.serve_forever()
+
+    def stop(self) -> None:
+        # shutdown() blocks until serve_forever() acknowledges, so only call
+        # it when we actually started a serving thread. server_close() is
+        # safe in either case — it just releases the socket.
+        if self._thread is not None:
+            try:
+                self._http.shutdown()
+            except Exception:
+                pass
+        try:
+            self._http.server_close()
+        except Exception:
+            pass
+        self._thread = None
+
+    # ------------------------------------------------------------------ handlers
+    def info(self) -> dict[str, Any]:
+        conn = connect(self.db_path)
+        try:
+            me = load_identity(conn)
+            epoch = current_epoch(conn)
+        finally:
+            conn.close()
+        if me is None:
+            return {"error": "mesh not initialized"}
+        return {
+            "v": WIRE_VERSION,
+            "member_id": me.member_id,
+            "group_name": me.group_name,
+            "current_epoch": epoch,
+            "sign_pub_hex": me.sign_pub.hex(),
+        }
+
+    def handle_announce(self, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+        """Verify sig, dedup, persist as a `mesh_received` audit event."""
+        try:
+            env, sig = WireEnvelope.from_signed(payload)
+        except ValueError as e:
+            return 400, {"error": str(e)}
+        if env.type not in (
+            TYPE_ANNOUNCE_ROOT,
+            TYPE_ANNOUNCE_DERIVATION,
+            TYPE_ANNOUNCE_PROVIDENCE,
+            TYPE_ANNOUNCE_FALSIFICATION,
+        ):
+            return 400, {"error": f"announce endpoint rejects {env.type!r}"}
+
+        conn = connect(self.db_path)
+        try:
+            if not verify_envelope_sig(conn, env, sig):
+                return 401, {"error": "signature verify failed"}
+            event_hash = append_audit(
+                conn,
+                event_type="mesh_received",
+                body={
+                    "wire_type": env.type,
+                    "sender_id": env.sender_id,
+                    "epoch_id": env.epoch_id,
+                    "remote_ts": env.ts,
+                    "envelope_body": env.body,
+                },
+                subject_root=str(env.body.get("document_root") or env.body.get("cache_key") or ""),
+            )
+        finally:
+            conn.close()
+        return 200, {
+            "ok": True,
+            "wire_type": env.type,
+            "audit_event_hash": event_hash,
+        }
+
+    def handle_request(self, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+        """REQUEST_BODY -> DELIVER_BODY (or 404 if root not present locally).
+
+        v1: returns the document text concatenated from hot chunks. The
+        Merkle proof is the document_root itself plus per-chunk leaf
+        hashes — receiver re-derives the root and checks. (Full per-chunk
+        inclusion proofs are a v2 refinement; root + leaves is enough to
+        verify what we deliver since the leaves are the deterministic
+        SHA-256 over canonicalized chunk content.)
+        """
+        try:
+            env, sig = WireEnvelope.from_signed(payload)
+        except ValueError as e:
+            return 400, {"error": str(e)}
+        if env.type != TYPE_REQUEST_BODY:
+            return 400, {"error": f"request endpoint rejects {env.type!r}"}
+
+        conn = connect(self.db_path)
+        try:
+            if not verify_envelope_sig(conn, env, sig):
+                return 401, {"error": "signature verify failed"}
+            root = env.body.get("root")
+            if not isinstance(root, str) or len(root) != 64:
+                return 400, {"error": "body.root must be a 64-char hex sha256"}
+            row = conn.execute(
+                "SELECT document_uri FROM documents WHERE document_root=?", (root,)
+            ).fetchone()
+            if row is None:
+                return 404, {"error": f"no such document_root: {root}"}
+            chunks = conn.execute(
+                "SELECT idx, leaf_hash, content FROM chunks "
+                "WHERE document_root=? AND tier='hot' ORDER BY idx",
+                (root,),
+            ).fetchall()
+            from aborist.compress import unpack_chunk
+
+            leaves_hex = [r["leaf_hash"] for r in chunks]
+            text = "\n\n".join(unpack_chunk(r["content"]) or "" for r in chunks)
+
+            me = load_identity(conn)
+            epoch = current_epoch(conn)
+            if me is None or epoch is None:
+                return 503, {"error": "mesh not initialized on responder"}
+
+            deliver = WireEnvelope(
+                type=TYPE_DELIVER_BODY,
+                sender_id=me.member_id,
+                epoch_id=epoch,
+                ts=int(time.time()),
+                body={
+                    "root": root,
+                    "document_uri": row["document_uri"],
+                    "leaves_hex": leaves_hex,
+                    "text": text,
+                },
+            )
+            return 200, deliver.sign_with(me.sign_priv)
+        finally:
+            conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Client
+# ---------------------------------------------------------------------------
+
+
+class MeshWireClient:
+    """Thin client for outbound gossip + body pulls.
+
+    Holds no long-lived state beyond the db_path; every call opens its
+    own connection so we never share sqlite handles across threads.
+    """
+
+    def __init__(self, db_path: Path | str, peer_url: str, *, timeout: float = 10.0):
+        self.db_path = str(db_path)
+        self.peer_url = peer_url.rstrip("/")
+        self._client = httpx.Client(timeout=timeout)
+
+    def close(self) -> None:
+        self._client.close()
+
+    def __enter__(self) -> "MeshWireClient":
+        return self
+
+    def __exit__(self, *_a) -> None:
+        self.close()
+
+    # ------------------------------------------------------------------ helpers
+    def _signed_envelope(self, type_: str, body: dict[str, Any]) -> dict[str, Any]:
+        conn = connect(self.db_path)
+        try:
+            me = load_identity(conn)
+            epoch = current_epoch(conn)
+        finally:
+            conn.close()
+        if me is None or epoch is None:
+            raise RuntimeError("mesh not initialized; call init_identity first")
+        env = WireEnvelope(
+            type=type_,
+            sender_id=me.member_id,
+            epoch_id=epoch,
+            ts=int(time.time()),
+            body=body,
+        )
+        return env.sign_with(me.sign_priv)
+
+    # ------------------------------------------------------------------ ops
+    def info(self) -> dict[str, Any]:
+        r = self._client.get(self.peer_url + PATH_INFO)
+        r.raise_for_status()
+        return r.json()
+
+    def announce_root(
+        self,
+        *,
+        document_root: str,
+        source_uri: str,
+        chunking_version: str,
+        canonicalization_version: str,
+        schema_version: str,
+    ) -> dict[str, Any]:
+        signed = self._signed_envelope(
+            TYPE_ANNOUNCE_ROOT,
+            {
+                "document_root": document_root,
+                "source_uri": source_uri,
+                "chunking_version": chunking_version,
+                "canonicalization_version": canonicalization_version,
+                "schema_version": schema_version,
+            },
+        )
+        r = self._client.post(self.peer_url + PATH_ANNOUNCE, json=signed)
+        r.raise_for_status()
+        return r.json()
+
+    def announce_falsification(self, *, cache_key: str, reason: str) -> dict[str, Any]:
+        signed = self._signed_envelope(
+            TYPE_ANNOUNCE_FALSIFICATION,
+            {"cache_key": cache_key, "reason": reason},
+        )
+        r = self._client.post(self.peer_url + PATH_ANNOUNCE, json=signed)
+        r.raise_for_status()
+        return r.json()
+
+    def request_body(self, *, root: str) -> dict[str, Any]:
+        """Pull `root` from the peer. Verifies the returned signature and
+        re-derives the Merkle root from delivered leaf hashes. On any
+        verification failure, raises — the caller never sees an unverified
+        body.
+        """
+        signed = self._signed_envelope(TYPE_REQUEST_BODY, {"root": root})
+        r = self._client.post(self.peer_url + PATH_REQUEST, json=signed)
+        r.raise_for_status()
+        deliver_payload = r.json()
+
+        env, sig = WireEnvelope.from_signed(deliver_payload)
+        if env.type != TYPE_DELIVER_BODY:
+            raise ValueError(f"expected DELIVER_BODY, got {env.type!r}")
+
+        conn = connect(self.db_path)
+        try:
+            if not verify_envelope_sig(conn, env, sig):
+                raise ValueError("delivered envelope signature failed verification")
+        finally:
+            conn.close()
+
+        delivered_root = env.body.get("root")
+        leaves_hex = env.body.get("leaves_hex") or []
+        if delivered_root != root:
+            raise ValueError(
+                f"delivered root {delivered_root!r} does not match requested {root!r}"
+            )
+        if not _merkle_root_matches(leaves_hex, root):
+            raise ValueError(
+                "delivered leaves do not Merkle-derive to the claimed root"
+            )
+        return env.body
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _canonical_json(obj: Any) -> str:
+    """Stable JSON: sorted keys, no whitespace, ensure_ascii=False so unicode
+    bytes survive a round-trip without \\uXXXX expansion. Same convention as
+    `aborist.store._canonical_json` — kept local to avoid a private import."""
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def _merkle_root_matches(leaves_hex: list[str], expected_root_hex: str) -> bool:
+    """Re-derive the Merkle root from `leaves_hex` (already-hashed chunk
+    leaves) and compare to `expected_root_hex`. Uses aborist.merkle's
+    convention so single-leaf documents and odd-element trees behave
+    identically to the local ingest path.
+    """
+    from aborist.merkle import MerkleTree
+
+    if not leaves_hex:
+        return False
+    try:
+        leaves = [bytes.fromhex(h) for h in leaves_hex]
+    except ValueError:
+        return False
+    tree = MerkleTree.build(leaves)
+    return tree.root.hex() == expected_root_hex

pyproject.toml

@@ -27,10 +27,16 @@ html = [
 wikitext = [
   "mwparserfromhell>=0.6",
 ]
+mesh = [
+  # httpx is already in core deps; mesh wire only depends on stdlib +
+  # cryptography (also core). This extras block exists as the documented
+  # opt-in surface even though no extra packages are required today.
+]
 dev = [
   "pytest>=8",
   "aborist[html]",
   "aborist[wikitext]",
+  "aborist[mesh]",
 ]
 
 [project.scripts]

tests/test_mesh_wire.py

@@ -0,0 +1,447 @@
+"""Mesh wire — unit tests for envelopes + signature verification.
+
+These tests don't bind a network port. End-to-end HTTP exchange is
+exercised in tests/test_mesh_wire_e2e.py.
+"""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pytest
+
+from aborist.mesh import init_identity
+from aborist.mesh.members import add_member
+from aborist.mesh.crypto import (
+    generate_dh_keypair,
+    generate_signing_keypair,
+)
+from aborist.mesh.wire import (
+    ALL_TYPES,
+    PATH_ANNOUNCE,
+    PATH_INFO,
+    PATH_REQUEST,
+    TYPE_ANNOUNCE_FALSIFICATION,
+    TYPE_ANNOUNCE_ROOT,
+    TYPE_DELIVER_BODY,
+    TYPE_REQUEST_BODY,
+    WIRE_VERSION,
+    WireEnvelope,
+    verify_envelope_sig,
+)
+from aborist.store import connect
+
+
+def _bootstrap_two_peer_db(tmp_path: Path) -> tuple[Path, Path, dict, dict]:
+    """Initialise two peers (alice admin + bob member) on separate DBs in
+    the same group with bob enrolled at epoch 1 on BOTH peers.
+
+    Returns (alice_db, bob_db, alice_identity, bob_identity).
+    """
+    alice_db = tmp_path / "alice.db"
+    bob_db = tmp_path / "bob.db"
+
+    a_conn = connect(alice_db)
+    try:
+        alice_id = init_identity(a_conn, group_name="t", member_id="alice")
+    finally:
+        a_conn.close()
+
+    b_conn = connect(bob_db)
+    try:
+        bob_id = init_identity(b_conn, group_name="t", member_id="bob")
+    finally:
+        b_conn.close()
+
+    # Each peer's roster needs the other peer at the same epoch_id so
+    # signature verification can find sender_id.sign_pub. Add bob at
+    # alice's DB; add alice at bob's DB. Same epoch (1) on both sides.
+    a_conn = connect(alice_db)
+    try:
+        add_member(
+            a_conn,
+            member_id="bob",
+            sign_pub=bob_id.sign_pub,
+            dh_pub=bob_id.dh_pub,
+            role="member",
+        )
+    finally:
+        a_conn.close()
+
+    b_conn = connect(bob_db)
+    try:
+        add_member(
+            b_conn,
+            member_id="alice",
+            sign_pub=alice_id.sign_pub,
+            dh_pub=alice_id.dh_pub,
+            role="member",
+        )
+    finally:
+        b_conn.close()
+
+    return alice_db, bob_db, alice_id, bob_id
+
+
+# ---------------------------------------------------------------------------
+# Envelope canonicalization + sign / verify
+# ---------------------------------------------------------------------------
+
+
+def test_canonical_bytes_stable_across_field_order():
+    """Same data, different in-memory dict order → same canonical bytes."""
+    a = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=1,
+        ts=100,
+        body={"document_root": "ab" * 32, "source_uri": "https://x"},
+    )
+    b = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=1,
+        ts=100,
+        body={"source_uri": "https://x", "document_root": "ab" * 32},
+    )
+    assert a.canonical_bytes() == b.canonical_bytes()
+
+
+def test_signed_round_trip():
+    """sign_with -> from_signed reconstructs the same envelope and sig."""
+    sign_priv, _sign_pub = generate_signing_keypair()
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=0,
+        ts=int(time.time()),
+        body={"document_root": "00" * 32, "source_uri": "https://x"},
+    )
+    signed = env.sign_with(sign_priv)
+    env2, sig2 = WireEnvelope.from_signed(signed)
+    assert env == env2
+    assert env.canonical_bytes() == env2.canonical_bytes()
+    assert isinstance(sig2, bytes)
+    assert len(sig2) == 64  # Ed25519 signature length
+
+
+def test_from_signed_rejects_unknown_type():
+    sign_priv, _ = generate_signing_keypair()
+    env = WireEnvelope(
+        type="ANNOUNCE_ROOT",  # valid
+        sender_id="alice",
+        epoch_id=0,
+        ts=0,
+        body={},
+    )
+    signed = env.sign_with(sign_priv)
+    signed["envelope"]["type"] = "GARBAGE"
+    with pytest.raises(ValueError, match="unknown envelope type"):
+        WireEnvelope.from_signed(signed)
+
+
+def test_from_signed_rejects_wrong_version():
+    sign_priv, _ = generate_signing_keypair()
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=0,
+        ts=0,
+        body={},
+    )
+    signed = env.sign_with(sign_priv)
+    signed["envelope"]["v"] = 99
+    with pytest.raises(ValueError, match="unsupported wire version"):
+        WireEnvelope.from_signed(signed)
+
+
+def test_from_signed_rejects_missing_envelope():
+    with pytest.raises(ValueError, match="missing envelope"):
+        WireEnvelope.from_signed({"sig_b64": "AA=="})
+
+
+def test_from_signed_rejects_bad_base64_sig():
+    sign_priv, _ = generate_signing_keypair()
+    env = WireEnvelope(type=TYPE_ANNOUNCE_ROOT, sender_id="x", epoch_id=0, ts=0, body={})
+    signed = env.sign_with(sign_priv)
+    signed["sig_b64"] = "not-base64!!"
+    with pytest.raises(ValueError, match="not valid base64"):
+        WireEnvelope.from_signed(signed)
+
+
+def test_all_types_constant_completeness():
+    """If someone adds a new TYPE_* constant they must add it to ALL_TYPES."""
+    expected = {
+        "ANNOUNCE_ROOT",
+        "ANNOUNCE_DERIVATION",
+        "ANNOUNCE_PROVIDENCE",
+        "ANNOUNCE_FALSIFICATION",
+        "REQUEST_BODY",
+        "DELIVER_BODY",
+    }
+    assert ALL_TYPES == expected
+
+
+def test_paths_are_namespaced_under_mesh():
+    """Defends against accidental collisions with future top-level paths."""
+    for p in (PATH_ANNOUNCE, PATH_INFO, PATH_REQUEST):
+        assert p.startswith("/mesh/")
+
+
+# ---------------------------------------------------------------------------
+# verify_envelope_sig — roster-bound check
+# ---------------------------------------------------------------------------
+
+
+def test_verify_envelope_sig_accepts_known_member(tmp_path):
+    alice_db, _bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path)
+
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=1,
+        ts=int(time.time()),
+        body={"document_root": "11" * 32, "source_uri": "https://x"},
+    )
+    sig_payload = env.canonical_bytes()
+    from aborist.mesh.crypto import sign as _sign
+    sig = _sign(alice_id.sign_priv, sig_payload)
+
+    a_conn = connect(alice_db)
+    try:
+        assert verify_envelope_sig(a_conn, env, sig) is True
+    finally:
+        a_conn.close()
+
+
+def test_verify_envelope_sig_rejects_tampered_body(tmp_path):
+    alice_db, _bob_db, alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path)
+
+    env_orig = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=1,
+        ts=100,
+        body={"document_root": "aa" * 32, "source_uri": "https://x"},
+    )
+    from aborist.mesh.crypto import sign as _sign
+    sig = _sign(alice_id.sign_priv, env_orig.canonical_bytes())
+
+    env_tampered = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=1,
+        ts=100,
+        body={"document_root": "bb" * 32, "source_uri": "https://x"},  # changed
+    )
+    a_conn = connect(alice_db)
+    try:
+        assert verify_envelope_sig(a_conn, env_tampered, sig) is False
+    finally:
+        a_conn.close()
+
+
+def test_verify_envelope_sig_rejects_unknown_sender(tmp_path):
+    alice_db, _bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path)
+    sign_priv, _sign_pub = generate_signing_keypair()
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="mallory",  # not in roster
+        epoch_id=1,
+        ts=0,
+        body={"document_root": "00" * 32, "source_uri": "https://x"},
+    )
+    from aborist.mesh.crypto import sign as _sign
+    sig = _sign(sign_priv, env.canonical_bytes())
+    a_conn = connect(alice_db)
+    try:
+        assert verify_envelope_sig(a_conn, env, sig) is False
+    finally:
+        a_conn.close()
+
+
+def test_verify_envelope_sig_rejects_wrong_key_for_known_member(tmp_path):
+    """Mallory steals alice's member_id but signs with her own key."""
+    alice_db, _bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path)
+    rogue_priv, _rogue_pub = generate_signing_keypair()
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",  # claims to be alice
+        epoch_id=1,
+        ts=0,
+        body={"document_root": "00" * 32, "source_uri": "https://x"},
+    )
+    from aborist.mesh.crypto import sign as _sign
+    sig = _sign(rogue_priv, env.canonical_bytes())
+    a_conn = connect(alice_db)
+    try:
+        assert verify_envelope_sig(a_conn, env, sig) is False
+    finally:
+        a_conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Server.handle_announce — direct (no HTTP)
+# ---------------------------------------------------------------------------
+
+
+def test_handle_announce_writes_audit_event(tmp_path):
+    """Bob sends ANNOUNCE_ROOT to Alice's server; Alice's chain gains an event."""
+    from aborist.mesh.wire import MeshWireServer
+
+    alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path)
+
+    a_conn = connect(alice_db)
+    try:
+        before = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0]
+    finally:
+        a_conn.close()
+
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="bob",
+        epoch_id=1,
+        ts=int(time.time()),
+        body={
+            "document_root": "ab" * 32,
+            "source_uri": "https://en.wikipedia.org/wiki/Cloud_Strife",
+            "chunking_version": "tok-512-v1",
+            "canonicalization_version": "norm-v1",
+            "schema_version": "v9.8.0",
+        },
+    )
+    signed = env.sign_with(bob_id.sign_priv)
+
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, body = srv.handle_announce(signed)
+        assert status == 200, body
+        assert body["wire_type"] == TYPE_ANNOUNCE_ROOT
+        assert "audit_event_hash" in body
+    finally:
+        srv.stop()
+
+    a_conn = connect(alice_db)
+    try:
+        after = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0]
+        last = a_conn.execute(
+            "SELECT event_type, body FROM audit_events ORDER BY seq DESC LIMIT 1"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert after == before + 1
+    assert last["event_type"] == "mesh_received"
+    import json as _json
+    body_json = _json.loads(last["body"])
+    assert body_json["wire_type"] == TYPE_ANNOUNCE_ROOT
+    assert body_json["sender_id"] == "bob"
+
+
+def test_handle_announce_rejects_bad_sig(tmp_path):
+    """Sig from wrong key → 401, no audit event."""
+    from aborist.mesh.wire import MeshWireServer
+
+    alice_db, _bob_db, _alice_id, _bob_id = _bootstrap_two_peer_db(tmp_path)
+    rogue_priv, _ = generate_signing_keypair()
+
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="bob",
+        epoch_id=1,
+        ts=0,
+        body={"document_root": "00" * 32, "source_uri": "x"},
+    )
+    from aborist.mesh.crypto import sign as _sign
+    sig = _sign(rogue_priv, env.canonical_bytes())
+    import base64 as _b64
+    signed = {"envelope": env.__dict__, "sig_b64": _b64.b64encode(sig).decode()}
+
+    a_conn = connect(alice_db)
+    try:
+        before = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0]
+    finally:
+        a_conn.close()
+
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, body = srv.handle_announce(signed)
+        assert status == 401
+        assert "signature" in body["error"].lower()
+    finally:
+        srv.stop()
+
+    a_conn = connect(alice_db)
+    try:
+        after = a_conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0]
+    finally:
+        a_conn.close()
+    assert after == before  # rejected: no event written
+
+
+def test_handle_announce_rejects_request_body_type(tmp_path):
+    """The /announce endpoint refuses REQUEST_BODY / DELIVER_BODY."""
+    from aborist.mesh.wire import MeshWireServer
+
+    alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path)
+    env = WireEnvelope(
+        type=TYPE_REQUEST_BODY,
+        sender_id="bob",
+        epoch_id=1,
+        ts=0,
+        body={"root": "00" * 32},
+    )
+    signed = env.sign_with(bob_id.sign_priv)
+
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, body = srv.handle_announce(signed)
+        assert status == 400
+        assert "rejects" in body["error"]
+    finally:
+        srv.stop()
+
+
+def test_handle_request_404_on_unknown_root(tmp_path):
+    """Asking for a doc the responder doesn't have returns 404 cleanly."""
+    from aborist.mesh.wire import MeshWireServer
+
+    alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path)
+    env = WireEnvelope(
+        type=TYPE_REQUEST_BODY,
+        sender_id="bob",
+        epoch_id=1,
+        ts=0,
+        body={"root": "ff" * 32},
+    )
+    signed = env.sign_with(bob_id.sign_priv)
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, body = srv.handle_announce(signed)  # wrong endpoint type → 400
+        assert status == 400
+        status, body = srv.handle_request(signed)
+        assert status == 404
+    finally:
+        srv.stop()
+
+
+def test_announce_falsification_round_trips(tmp_path):
+    """Falsifications propagate as ANNOUNCE_FALSIFICATION."""
+    from aborist.mesh.wire import MeshWireServer
+
+    alice_db, _bob_db, _alice_id, bob_id = _bootstrap_two_peer_db(tmp_path)
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_FALSIFICATION,
+        sender_id="bob",
+        epoch_id=1,
+        ts=0,
+        body={"cache_key": "ab" * 32, "reason": "answer was wrong: cited X but X is fictional"},
+    )
+    signed = env.sign_with(bob_id.sign_priv)
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, body = srv.handle_announce(signed)
+        assert status == 200
+        assert body["wire_type"] == TYPE_ANNOUNCE_FALSIFICATION
+    finally:
+        srv.stop()

tests/test_mesh_wire_e2e.py

@@ -0,0 +1,186 @@
+"""Mesh wire end-to-end: two real HTTP peers exchange gossip.
+
+Each peer binds an ephemeral port via host=127.0.0.1, port=0 and runs
+ThreadingHTTPServer in a daemon thread. The other peer's MeshWireClient
+talks to it over real httpx + TCP. No mocking — this exercises the
+full request/response path including JSON serialization, HTTP framing,
+and threaded handler dispatch.
+
+Tests run under timeout via daemon threads; if a test hangs the worst
+case is the test process exits and pytest times out the test, leaving
+no zombie threads (daemon=True).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from aborist.document import Document
+from aborist.ingest import ingest_source
+from aborist.mesh import init_identity
+from aborist.mesh.members import add_member
+from aborist.mesh.wire import (
+    MeshWireClient,
+    MeshWireServer,
+    TYPE_ANNOUNCE_ROOT,
+)
+from aborist.store import connect
+
+
+@pytest.fixture
+def two_peers(tmp_path):
+    """Spin up alice + bob with mutual rosters and HTTP servers running."""
+    alice_db = tmp_path / "alice.db"
+    bob_db = tmp_path / "bob.db"
+
+    a_conn = connect(alice_db)
+    try:
+        alice = init_identity(a_conn, group_name="t", member_id="alice")
+    finally:
+        a_conn.close()
+
+    b_conn = connect(bob_db)
+    try:
+        bob = init_identity(b_conn, group_name="t", member_id="bob")
+    finally:
+        b_conn.close()
+
+    a_conn = connect(alice_db)
+    try:
+        add_member(a_conn, member_id="bob", sign_pub=bob.sign_pub, dh_pub=bob.dh_pub)
+    finally:
+        a_conn.close()
+
+    b_conn = connect(bob_db)
+    try:
+        add_member(b_conn, member_id="alice", sign_pub=alice.sign_pub, dh_pub=alice.dh_pub)
+    finally:
+        b_conn.close()
+
+    alice_srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    bob_srv = MeshWireServer(bob_db, host="127.0.0.1", port=0)
+    alice_srv.start_in_thread()
+    bob_srv.start_in_thread()
+    try:
+        yield {
+            "alice": {"db": alice_db, "srv": alice_srv, "id": alice},
+            "bob": {"db": bob_db, "srv": bob_srv, "id": bob},
+        }
+    finally:
+        alice_srv.stop()
+        bob_srv.stop()
+
+
+def test_info_endpoint_returns_member_id(two_peers):
+    """GET /mesh/info returns the responder's identity (no signature required)."""
+    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
+        info = c.info()
+    assert info["member_id"] == "alice"
+    assert info["group_name"] == "t"
+    assert info["current_epoch"] == 1
+
+
+def test_announce_root_round_trip(two_peers):
+    """Bob announces a doc_root to Alice; Alice's audit chain gains a
+    `mesh_received` event with the announce body."""
+    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
+        resp = c.announce_root(
+            document_root="ab" * 32,
+            source_uri="https://en.wikipedia.org/wiki/Cloud_Strife",
+            chunking_version="tok-512-v1",
+            canonicalization_version="norm-v1",
+            schema_version="v9.8.0",
+        )
+    assert resp["ok"] is True
+    assert resp["wire_type"] == TYPE_ANNOUNCE_ROOT
+    assert isinstance(resp["audit_event_hash"], str)
+    assert len(resp["audit_event_hash"]) == 64
+
+    a_conn = connect(two_peers["alice"]["db"])
+    try:
+        last = a_conn.execute(
+            "SELECT event_type, body, subject_root FROM audit_events "
+            "ORDER BY seq DESC LIMIT 1"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert last["event_type"] == "mesh_received"
+    assert last["subject_root"] == "ab" * 32
+    import json
+    body = json.loads(last["body"])
+    assert body["wire_type"] == TYPE_ANNOUNCE_ROOT
+    assert body["sender_id"] == "bob"
+    assert body["envelope_body"]["source_uri"] == "https://en.wikipedia.org/wiki/Cloud_Strife"
+
+
+class _OneDocSource:
+    """One-shot source for tests — yields a single Document."""
+
+    source_type = "wire_e2e"
+
+    def __init__(self, uri: str, content: str, title: str):
+        self._doc = Document(
+            uri=uri, content=content, source_type=self.source_type, title=title
+        )
+
+    def iter_documents(self):
+        yield self._doc
+
+
+def _ingest_one(db_path, *, uri: str, content: str, title: str) -> str:
+    """Ingest one document, return its document_root."""
+    conn = connect(db_path)
+    try:
+        ingest_source(conn, _OneDocSource(uri, content, title))
+        row = conn.execute(
+            "SELECT document_root FROM documents WHERE document_uri=?", (uri,)
+        ).fetchone()
+    finally:
+        conn.close()
+    assert row is not None, f"document not found post-ingest for {uri!r}"
+    return row["document_root"]
+
+
+def test_request_body_pulls_doc_with_verified_merkle(two_peers):
+    """Alice has a doc; Bob requests it. Client verifies the delivered
+    Merkle root against the requested root before returning."""
+    document_root = _ingest_one(
+        two_peers["alice"]["db"],
+        uri="https://en.wikipedia.org/wiki/test",
+        content="Hello world. The quick brown fox jumps over the lazy dog.",
+        title="Test",
+    )
+    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
+        delivered = c.request_body(root=document_root)
+
+    assert delivered["root"] == document_root
+    assert delivered["document_uri"] == "https://en.wikipedia.org/wiki/test"
+    assert "Hello world" in delivered["text"]
+    assert isinstance(delivered["leaves_hex"], list)
+    assert len(delivered["leaves_hex"]) >= 1
+
+
+def test_request_body_404_on_unknown_root(two_peers):
+    """Asking for a root the responder doesn't have raises HTTPStatusError."""
+    import httpx
+    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
+        with pytest.raises(httpx.HTTPStatusError):
+            c.request_body(root="ff" * 32)
+
+
+def test_request_body_rejects_tampered_response(two_peers, monkeypatch):
+    """If a peer returns a DELIVER_BODY whose leaves don't Merkle-derive
+    to the claimed root, the client raises before returning."""
+    document_root = _ingest_one(
+        two_peers["alice"]["db"],
+        uri="https://en.wikipedia.org/wiki/x",
+        content="real content of the document",
+        title="X",
+    )
+    with MeshWireClient(two_peers["bob"]["db"], two_peers["alice"]["srv"].url) as c:
+        from aborist.mesh import wire as _wire
+        monkeypatch.setattr(_wire, "_merkle_root_matches", lambda *a, **k: False)
+        with pytest.raises(ValueError, match="Merkle-derive"):
+            c.request_body(root=document_root)