mesh: per-peer audit chain-of-claims tracking on the wire

Receiver now rejects gossip envelopes whose prev_event_hash doesn't
extend the sender's last-seen event_hash — fork detection lives at the
wire layer, not just in docs/mesh.md.

Schema: new mesh_peer_chains table (peer_member_id PK, last_event_hash,
last_seq, last_seen_at) with idempotent _migrate_mesh_peer_chains
matching the _migrate_audit_mode pattern.

Envelope: WireEnvelope gains prev_event_hash + event_hash, both
optional for back-compat. canonical_bytes() excludes event_hash to
avoid the circular dependency (the hash is computed FROM the canonical
bytes). New WireEnvelope.with_chain(prev_event_hash=...) builds a
linked envelope with event_hash = sha256(prev || canonical_bytes).

Sender (MeshWireClient._signed_envelope, MeshWireServer.handle_request
DELIVER_BODY): reads our latest_event_hash, builds the linked
envelope, then appends a 'mesh_sent' audit event whose body is the
envelope's canonical (event-hash-excluded) dict. By construction the
audit event_hash equals the envelope event_hash — wire chain-of-claims
and local audit chain stay in lockstep, so back-to-back sends advance
the chain naturally.

Receiver (handle_announce): when an envelope carries chain fields,
recompute event_hash and 400 on mismatch; look up
mesh_peer_chains[sender] and 409 on prev mismatch; on accept update
the row + write the existing mesh_received event.

Backward compat: legacy envelopes (event_hash=None) bypass chain
enforcement — kept the existing test_mesh_wire.py fixtures verbatim
since they construct WireEnvelope directly without with_chain(). All
production traffic goes through MeshWireClient and is always tracked.

v1 deferred (documented in module docstring): no multi-event catchup;
operator retries on 409. No cross-peer reconciliation.

Tests: 12 new in tests/test_mesh_chain.py covering the 7 required
cases plus canonical-bytes exclusion, migration idempotence, and
legacy-envelope fallback. 246 passed, 1 skipped overall.

aborist/mesh/wire.py

@@ -16,15 +16,51 @@ 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``).
+stays internally consistent because we append in receive-order.
+
+**Per-peer chain-of-claims tracking.** Every envelope built via the
+sender helper (``MeshWireClient`` outbound traffic, plus
+``WireEnvelope.with_chain``) carries two extra fields:
+
+- ``prev_event_hash`` — the sender's previous chain-of-claims hash, or
+  ``None`` for the first envelope a peer ever broadcasts.
+- ``event_hash`` — ``sha256(prev_event_hash || canonical_envelope_bytes)``.
+
+The receiver:
+1. Verifies the Ed25519 signature (existing).
+2. Recomputes ``event_hash`` from ``prev_event_hash`` + the envelope's
+   canonical bytes (excluding ``event_hash`` itself, which is circular).
+   Mismatch → 400.
+3. Looks up the sender's last-seen ``event_hash`` in
+   ``mesh_peer_chains``. If a row exists, the envelope's
+   ``prev_event_hash`` MUST equal it; otherwise the chain has forked
+   and we return 409 Conflict. If no row exists, this is the first
+   envelope we've seen from this peer; any ``prev_event_hash`` is
+   accepted (we have no prior to compare against).
+4. On accept, ``mesh_peer_chains`` is updated to the new
+   ``event_hash`` + bumped ``last_seq`` + ``now``.
+
+Backward compatibility: envelopes that arrive WITHOUT
+``prev_event_hash`` and ``event_hash`` (legacy, hand-built test fixtures
+from before the chain-tracking rollout) skip per-peer chain enforcement.
+The signature check still runs. New code paths always carry chain
+fields, so this fallback only matters for legacy test envelopes.
+
+v1 limitations (deferred to a later commit):
+
+- **No multi-event catchup / replay.** If a receiver missed event N
+  (peer sent N+1 with prev=N's hash but receiver still has prev=N-1),
+  we reject with 409 and rely on an operator to retry or run a catchup
+  pass. v2 will fetch the missing range from the peer and replay.
+- **No cross-peer reconciliation.** Each peer's chain is tracked
+  independently — alice's broadcasts to bob and bob's broadcasts to
+  carol are unrelated streams.
 """
 
 from __future__ import annotations
 
 import base64
+import hashlib
 import json
 import os
 import threading
@@ -43,7 +79,7 @@ from aborist.mesh.state import (
     recover_epoch_secret,
     roster_at,
 )
-from aborist.store import append_audit, connect
+from aborist.store import append_audit, connect, latest_event_hash
 
 
 # ---------------------------------------------------------------------------
@@ -95,6 +131,17 @@ class WireEnvelope:
     binds the ciphertext: a MITM cannot swap `encrypted_body` without
     invalidating the signature. Receivers detect encryption by the
     presence of `encrypted_body` and decrypt before handler dispatch.
+
+    `prev_event_hash` and `event_hash` carry the sender's chain-of-claims
+    state. Both default to ``None`` for backward compatibility with hand-
+    built test envelopes. Production traffic is built via
+    :meth:`with_chain` (or `MeshWireClient`) which always populates them.
+
+    `event_hash` is intentionally NOT included in the signed canonical
+    bytes — it is computed FROM those bytes, so embedding it would be
+    circular. The receiver re-derives ``event_hash`` from
+    ``sha256(prev_event_hash || canonical_bytes_without_event_hash)``
+    and asserts equality before accepting.
     """
 
     type: str
@@ -104,21 +151,73 @@ class WireEnvelope:
     body: dict[str, Any]
     v: int = WIRE_VERSION
     encrypted_body: dict[str, str] | None = None
+    prev_event_hash: str | None = None
+    event_hash: str | None = None
 
     def canonical_bytes(self) -> bytes:
         """Bytes that get signed. Stable, reproducible.
 
-        Includes `encrypted_body` whenever it is set, so signatures bind
-        the ciphertext as well as the metadata. We omit the field
-        entirely from the canonical form when it is None to keep
-        cleartext-only envelopes byte-identical to the v1 wire format
-        (forward-compat for receivers that don't speak AEAD yet).
+        Excludes ``event_hash`` (circular: it's the hash OF these bytes)
+        but includes ``prev_event_hash`` (linkage proof — the sender is
+        committing to a specific predecessor) and ``encrypted_body``
+        (so signatures bind the ciphertext too). The ``encrypted_body``
+        field is dropped from the canonical form when ``None`` so
+        cleartext-only envelopes stay byte-identical to the v1 wire
+        format — forward-compat for receivers that don't speak AEAD yet.
         """
         d = asdict(self)
+        d.pop("event_hash", None)
         if d.get("encrypted_body") is None:
             d.pop("encrypted_body", None)
         return _canonical_json(d).encode("utf-8")
 
+    def compute_event_hash(self) -> str:
+        """Deterministic hash of this envelope as a chain-of-claims node.
+
+        ``sha256(prev_event_hash_bytes || canonical_bytes)`` where
+        ``prev_event_hash_bytes`` is the raw 32-byte digest if
+        ``prev_event_hash`` is set, else empty.
+        """
+        h = hashlib.sha256()
+        if self.prev_event_hash is not None:
+            h.update(bytes.fromhex(self.prev_event_hash))
+        h.update(self.canonical_bytes())
+        return h.hexdigest()
+
+    def with_chain(self, *, prev_event_hash: str | None) -> "WireEnvelope":
+        """Return a copy with chain fields populated.
+
+        ``event_hash`` is computed deterministically from
+        ``prev_event_hash`` + this envelope's canonical bytes. The
+        returned envelope is the one the sender signs and broadcasts.
+        """
+        # Construct the linked envelope first so canonical_bytes() sees
+        # the new prev_event_hash. event_hash itself is excluded from the
+        # canonical bytes, so we set it last.
+        linked = WireEnvelope(
+            type=self.type,
+            sender_id=self.sender_id,
+            epoch_id=self.epoch_id,
+            ts=self.ts,
+            body=self.body,
+            v=self.v,
+            encrypted_body=self.encrypted_body,
+            prev_event_hash=prev_event_hash,
+            event_hash=None,
+        )
+        eh = linked.compute_event_hash()
+        return WireEnvelope(
+            type=linked.type,
+            sender_id=linked.sender_id,
+            epoch_id=linked.epoch_id,
+            ts=linked.ts,
+            body=linked.body,
+            v=linked.v,
+            encrypted_body=linked.encrypted_body,
+            prev_event_hash=linked.prev_event_hash,
+            event_hash=eh,
+        )
+
     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())
@@ -134,7 +233,11 @@ class WireEnvelope:
     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`.
+        Does NOT verify — caller verifies via `verify_envelope_sig` and
+        (when chain fields are present) via the receiver's chain check.
+        ``prev_event_hash`` and ``event_hash`` are optional in the wire
+        payload — legacy envelopes that pre-date chain tracking decode
+        with both as ``None``.
         """
         if not isinstance(signed, dict) or "envelope" not in signed or "sig_b64" not in signed:
             raise ValueError("signed payload missing envelope/sig_b64")
@@ -155,6 +258,12 @@ class WireEnvelope:
             for k in ("nonce_b64", "ct_b64"):
                 if k not in encrypted_body:
                     raise ValueError(f"encrypted_body missing field: {k}")
+        prev = env_d.get("prev_event_hash")
+        eh = env_d.get("event_hash")
+        if prev is not None and not isinstance(prev, str):
+            raise ValueError("prev_event_hash must be str or null")
+        if eh is not None and not isinstance(eh, str):
+            raise ValueError("event_hash must be str or null")
         env = cls(
             type=env_d["type"],
             sender_id=env_d["sender_id"],
@@ -163,6 +272,8 @@ class WireEnvelope:
             body=env_d["body"],
             v=int(env_d["v"]),
             encrypted_body=encrypted_body,
+            prev_event_hash=prev,
+            event_hash=eh,
         )
         try:
             sig = base64.b64decode(signed["sig_b64"], validate=True)
@@ -248,6 +359,44 @@ def decrypt_envelope_body(
     return decoded
 
 
+def _peer_chain_row(conn, peer_member_id: str) -> tuple[str, int] | None:
+    """Return (last_event_hash, last_seq) for a peer, or None if unseen."""
+    row = conn.execute(
+        "SELECT last_event_hash, last_seq FROM mesh_peer_chains "
+        "WHERE peer_member_id = ?",
+        (peer_member_id,),
+    ).fetchone()
+    if row is None:
+        return None
+    return row["last_event_hash"], int(row["last_seq"])
+
+
+def _update_peer_chain(
+    conn,
+    peer_member_id: str,
+    new_event_hash: str,
+    *,
+    now: int,
+) -> None:
+    """Upsert mesh_peer_chains[peer] = new_event_hash, last_seq+=1, now."""
+    existing = _peer_chain_row(conn, peer_member_id)
+    if existing is None:
+        conn.execute(
+            "INSERT INTO mesh_peer_chains "
+            "(peer_member_id, last_event_hash, last_seq, last_seen_at) "
+            "VALUES (?, ?, ?, ?)",
+            (peer_member_id, new_event_hash, 1, now),
+        )
+    else:
+        _, last_seq = existing
+        conn.execute(
+            "UPDATE mesh_peer_chains "
+            "SET last_event_hash = ?, last_seq = ?, last_seen_at = ? "
+            "WHERE peer_member_id = ?",
+            (new_event_hash, last_seq + 1, now, peer_member_id),
+        )
+
+
 def verify_envelope_sig(
     conn,
     envelope: WireEnvelope,
@@ -425,6 +574,40 @@ class MeshWireServer:
         try:
             if not verify_envelope_sig(conn, env, sig):
                 return 401, {"error": "signature verify failed"}
+
+            # Per-peer chain-of-claims tracking. Only enforced when the
+            # sender attached chain fields. Hand-built legacy envelopes
+            # (env.event_hash is None) skip enforcement — signature alone.
+            if env.event_hash is not None:
+                expected_eh = env.compute_event_hash()
+                if expected_eh != env.event_hash:
+                    return 400, {
+                        "error": "envelope event_hash mismatch: "
+                        f"computed {expected_eh}, claimed {env.event_hash}"
+                    }
+                existing = _peer_chain_row(conn, env.sender_id)
+                if existing is not None:
+                    last_eh, _ = existing
+                    if env.prev_event_hash != last_eh:
+                        return 409, {
+                            "error": (
+                                f"fork detected: peer {env.sender_id} "
+                                f"expected prev_event_hash {last_eh}, "
+                                f"got {env.prev_event_hash}"
+                            )
+                        }
+                # else: first envelope from this peer; any prev_event_hash
+                # (None or arbitrary string) is accepted — we have no
+                # prior to compare against.
+                _update_peer_chain(
+                    conn,
+                    env.sender_id,
+                    env.event_hash,
+                    now=int(time.time()),
+                )
+
+            # AEAD decrypt happens after sig + chain checks so we don't
+            # waste a secret recovery on rejected traffic.
             effective_body = env.body
             if env.encrypted_body is not None:
                 try:
@@ -432,6 +615,7 @@ class MeshWireServer:
                     effective_body = decrypt_envelope_body(env, epoch_secret)
                 except (ValueError, RuntimeError) as e:
                     return 401, {"error": f"AEAD decrypt failed: {e!s}"}
+
             event_hash = append_audit(
                 conn,
                 event_type="mesh_received",
@@ -442,6 +626,11 @@ class MeshWireServer:
                     "remote_ts": env.ts,
                     "envelope_body": effective_body,
                     "encrypted": env.encrypted_body is not None,
+                    # Sender's chain-of-claims hash (if present). Records
+                    # what we accepted so future audits can reconstruct
+                    # the sender's claimed history.
+                    "sender_event_hash": env.event_hash,
+                    "sender_prev_event_hash": env.prev_event_hash,
                 },
                 subject_root=str(
                     effective_body.get("document_root")
@@ -501,6 +690,7 @@ class MeshWireServer:
             if me is None or epoch is None:
                 return 503, {"error": "mesh not initialized on responder"}
 
+            prev = latest_event_hash(conn)
             deliver = WireEnvelope(
                 type=TYPE_DELIVER_BODY,
                 sender_id=me.member_id,
@@ -512,7 +702,25 @@ class MeshWireServer:
                     "leaves_hex": leaves_hex,
                     "text": text,
                 },
+            ).with_chain(prev_event_hash=prev)
+
+            # Mirror the client's audit-log of every outbound chain-linked
+            # envelope so DELIVER_BODY also extends the responder's chain.
+            from dataclasses import asdict as _asdict
+            audit_body = _asdict(deliver)
+            audit_body.pop("event_hash", None)
+            # Mirror canonical_bytes(): drop None encrypted_body so the
+            # audit hash matches event_hash for cleartext envelopes.
+            if audit_body.get("encrypted_body") is None:
+                audit_body.pop("encrypted_body", None)
+            audit_eh = append_audit(
+                conn,
+                event_type="mesh_sent",
+                body=audit_body,
+                subject_root=root,
+                ts=deliver.ts,
             )
+            assert audit_eh == deliver.event_hash
             return 200, deliver.sign_with(me.sign_priv)
         finally:
             conn.close()
@@ -552,56 +760,93 @@ class MeshWireClient:
         *,
         encrypt: bool = False,
     ) -> dict[str, Any]:
-        """Build a signed envelope. When `encrypt` is True, AEAD-encrypt
-        the body under this peer's recovered epoch secret before signing.
+        """Build, chain-link, optionally encrypt, sign, and audit-log one
+        outbound envelope.
 
-        Sender-side secret recovery uses `recover_epoch_secret`, which
-        looks up the rotator from the audit chain and re-unwraps this
-        peer's slot — no separate sender-side cache is required.
+        Three responsibilities, in order:
+
+        1. **Confidentiality (optional).** When ``encrypt=True``, the
+           body is AEAD-encrypted under this peer's recovered epoch
+           secret. Sender-side recovery uses ``recover_epoch_secret``
+           which re-unwraps our own slot — no sender-side cache.
+        2. **Chain-of-claims.** Populate ``prev_event_hash`` from our
+           own ``latest_event_hash`` and compute ``event_hash`` so the
+           receiver can enforce per-peer chain consistency.
+        3. **Local audit log.** Append a ``mesh_sent`` event whose body
+           is the envelope's canonical (event-hash-excluded) dict.
+           By construction its ``audit_event.event_hash`` equals the
+           envelope's ``event_hash``, unifying the wire chain-of-claims
+           with the local audit chain. ``latest_event_hash`` then
+           advances to this envelope's ``event_hash``, so back-to-back
+           sends extend a single linear chain.
+
+        Drop the same dict shape into both wire and audit so a future
+        forensic replay only needs the audit body to reproduce the
+        envelope bytes (modulo signature).
         """
+        from dataclasses import asdict as _asdict
+
         conn = connect(self.db_path)
         try:
             me = load_identity(conn)
             epoch = current_epoch(conn)
-            secret: bytes | None = None
+            if me is None or epoch is None:
+                raise RuntimeError("mesh not initialized; call init_identity first")
+
+            ts = int(time.time())
             if encrypt:
-                if me is None or epoch is None:
-                    raise RuntimeError(
-                        "mesh not initialized; call init_identity first"
-                    )
                 secret = recover_epoch_secret(conn, epoch_id=epoch)
+                encrypted = encrypt_body_for_envelope(
+                    type_=type_,
+                    sender_id=me.member_id,
+                    epoch_id=epoch,
+                    ts=ts,
+                    body=body,
+                    epoch_secret=secret,
+                )
+                base_env = WireEnvelope(
+                    type=type_,
+                    sender_id=me.member_id,
+                    epoch_id=epoch,
+                    ts=ts,
+                    body={},  # placeholder; real body lives in encrypted_body
+                    encrypted_body=encrypted,
+                )
+            else:
+                base_env = WireEnvelope(
+                    type=type_,
+                    sender_id=me.member_id,
+                    epoch_id=epoch,
+                    ts=ts,
+                    body=body,
+                )
+
+            prev = latest_event_hash(conn)
+            env = base_env.with_chain(prev_event_hash=prev)
+
+            audit_body = _asdict(env)
+            audit_body.pop("event_hash", None)
+            # Drop None-valued encrypted_body so cleartext audit bodies
+            # stay byte-identical to the v1 audit shape (mirrors
+            # canonical_bytes' behavior — keeps audit hash == event_hash
+            # for cleartext envelopes built by older callers).
+            if audit_body.get("encrypted_body") is None:
+                audit_body.pop("encrypted_body", None)
+            audit_event_hash = append_audit(
+                conn,
+                event_type="mesh_sent",
+                body=audit_body,
+                subject_root=str(body.get("document_root") or body.get("cache_key") or ""),
+                ts=env.ts,
+            )
+            assert audit_event_hash == env.event_hash, (
+                f"chain unification broke: audit hash {audit_event_hash} "
+                f"!= envelope event_hash {env.event_hash}"
+            )
+            signed = env.sign_with(me.sign_priv)
         finally:
             conn.close()
-        if me is None or epoch is None:
-            raise RuntimeError("mesh not initialized; call init_identity first")
-        ts = int(time.time())
-        if encrypt:
-            assert secret is not None  # invariant from the branch above
-            encrypted = encrypt_body_for_envelope(
-                type_=type_,
-                sender_id=me.member_id,
-                epoch_id=epoch,
-                ts=ts,
-                body=body,
-                epoch_secret=secret,
-            )
-            env = WireEnvelope(
-                type=type_,
-                sender_id=me.member_id,
-                epoch_id=epoch,
-                ts=ts,
-                body={},  # placeholder; real body is in encrypted_body
-                encrypted_body=encrypted,
-            )
-        else:
-            env = WireEnvelope(
-                type=type_,
-                sender_id=me.member_id,
-                epoch_id=epoch,
-                ts=ts,
-                body=body,
-            )
-        return env.sign_with(me.sign_priv)
+        return signed
 
     # ------------------------------------------------------------------ ops
     def info(self) -> dict[str, Any]:

aborist/store.py

@@ -258,6 +258,19 @@ CREATE TABLE IF NOT EXISTS mesh_epochs (
     reason             TEXT
 );
 
+-- Per-peer audit-chain tracking. Each row records the most recent
+-- event_hash a given peer has broadcast to us; we enforce that every
+-- subsequent gossip envelope carries `prev_event_hash == last_event_hash`
+-- of that peer. A mismatch is a fork — the gossip is rejected (409).
+-- last_seq is the local count of accepted envelopes from that peer
+-- (informational; the canonical chain identity is last_event_hash).
+CREATE TABLE IF NOT EXISTS mesh_peer_chains (
+    peer_member_id    TEXT PRIMARY KEY,
+    last_event_hash   TEXT NOT NULL,
+    last_seq          INTEGER NOT NULL,
+    last_seen_at      INTEGER NOT NULL
+);
+
 -- FTS5 over chunk content for UNGROUNDED-mode keyword search.
 --
 -- Contentless mode (`content=''`): FTS5 stores ONLY the inverted index, no
@@ -296,6 +309,7 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection:
     conn.row_factory = sqlite3.Row
     conn.executescript(SCHEMA_SQL)
     _migrate_audit_mode(conn)
+    _migrate_mesh_peer_chains(conn)
     conn.execute("PRAGMA synchronous = NORMAL")
     conn.execute("PRAGMA cache_size = -65536")
     conn.execute("PRAGMA temp_store = MEMORY")
@@ -355,6 +369,30 @@ def _migrate_audit_mode(conn: sqlite3.Connection) -> None:
     )
 
 
+def _migrate_mesh_peer_chains(conn: sqlite3.Connection) -> None:
+    """Forward-migrate pre-mesh-fork-detection shards.
+
+    Adds the `mesh_peer_chains` table to DBs that pre-date per-peer
+    audit-chain tracking on the mesh wire. CREATE TABLE IF NOT EXISTS
+    in SCHEMA_SQL covers brand-new shards; this migration is a belt-
+    and-suspenders idempotency check for callers that bypass the full
+    SCHEMA_SQL pass (cross-shard query views, etc.). Idempotent —
+    PRAGMA-checks before issuing CREATE.
+    """
+    row = conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table' AND name='mesh_peer_chains'"
+    ).fetchone()
+    if row is None:
+        conn.execute(
+            "CREATE TABLE mesh_peer_chains ("
+            " peer_member_id    TEXT PRIMARY KEY,"
+            " last_event_hash   TEXT NOT NULL,"
+            " last_seq          INTEGER NOT NULL,"
+            " last_seen_at      INTEGER NOT NULL"
+            ")"
+        )
+
+
 def _rebuild_providence_cache_ungrounded(conn: sqlite3.Connection) -> None:
     """One-time table rebuild: rename audit_mode value VISUAL → UNGROUNDED.
 

tests/test_mesh_chain.py

@@ -0,0 +1,494 @@
+"""Mesh wire — per-peer audit-chain tracking.
+
+Receiver-side enforcement that gossip envelopes from a given peer
+extend a single linear chain of claims. Catches:
+- Tampered ``event_hash`` (recomputation mismatch -> 400).
+- Forks / replays where ``prev_event_hash`` doesn't match the last
+  envelope we accepted from that peer (-> 409).
+- Cross-talk between peers' chains (alice's broadcasts must not
+  affect bob's chain state and vice versa).
+
+Implements the spec from ``docs/mesh.md``:
+  > Audit chains merge by prev_event_hash. A gossip insert that does
+  > not extend a chain consistently is rejected — mesh never silently
+  > forks history.
+"""
+
+from __future__ import annotations
+
+import time
+from pathlib import Path
+
+import pytest
+
+from aborist.mesh import init_identity
+from aborist.mesh.crypto import sign as _sign
+from aborist.mesh.members import add_member
+from aborist.mesh.wire import (
+    MeshWireServer,
+    TYPE_ANNOUNCE_ROOT,
+    WireEnvelope,
+)
+from aborist.store import connect
+
+
+def _bootstrap_three_peer_db(tmp_path: Path):
+    """Alice (admin) + bob + carol on alice's DB, both enrolled at the
+    same current_epoch (each ``add_member`` bumps the epoch by one).
+
+    Returns (alice_db, current_epoch, alice_id, bob_id, carol_id). Alice's
+    roster at ``current_epoch`` contains bob and carol so signature
+    verification finds their sign_pubs when receiving from either.
+    """
+    from aborist.mesh.state import current_epoch as _current_epoch
+
+    alice_db = tmp_path / "alice.db"
+    bob_db = tmp_path / "bob.db"
+    carol_db = tmp_path / "carol.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()
+
+    c_conn = connect(carol_db)
+    try:
+        carol = init_identity(c_conn, group_name="t", member_id="carol")
+    finally:
+        c_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)
+        add_member(a_conn, member_id="carol", sign_pub=carol.sign_pub, dh_pub=carol.dh_pub)
+        epoch = _current_epoch(a_conn)
+    finally:
+        a_conn.close()
+
+    return alice_db, epoch, alice, bob, carol
+
+
+def _signed_envelope(
+    *,
+    sign_priv: bytes,
+    sender_id: str,
+    body: dict,
+    prev_event_hash: str | None,
+    epoch_id: int,
+    ts: int | None = None,
+) -> tuple[WireEnvelope, dict]:
+    """Build a chain-linked, signed envelope. Returns (env, signed_dict)."""
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id=sender_id,
+        epoch_id=epoch_id,
+        ts=int(time.time()) if ts is None else ts,
+        body=body,
+    ).with_chain(prev_event_hash=prev_event_hash)
+    return env, env.sign_with(sign_priv)
+
+
+def _root_for(seed: str) -> str:
+    """Deterministic 64-char hex doc-root marker for tests (not a real hash)."""
+    return (seed * 64)[:64]
+
+
+# ---------------------------------------------------------------------------
+# Sender-side determinism
+# ---------------------------------------------------------------------------
+
+
+def test_with_chain_produces_deterministic_event_hash():
+    """Same envelope + same prev_event_hash -> same event_hash, every time."""
+    body = {
+        "document_root": _root_for("a"),
+        "source_uri": "https://x",
+        "chunking_version": "tok-512-v1",
+        "canonicalization_version": "norm-v1",
+        "schema_version": "v9.8.0",
+    }
+    env_base = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=1,
+        ts=1700_000_000,
+        body=body,
+    )
+    a = env_base.with_chain(prev_event_hash=None)
+    b = env_base.with_chain(prev_event_hash=None)
+    assert a.event_hash == b.event_hash
+    assert a.event_hash is not None
+    assert len(a.event_hash) == 64
+
+    # Different prev -> different hash.
+    c = env_base.with_chain(prev_event_hash="aa" * 32)
+    assert c.event_hash != a.event_hash
+
+
+def test_event_hash_excluded_from_canonical_bytes():
+    """canonical_bytes() must NOT contain event_hash (avoids circularity).
+
+    The literal hex of event_hash also must not appear, since the JSON
+    serialization is the only path it could leak through. (The substring
+    ``"event_hash"`` does appear because ``prev_event_hash`` is included;
+    that's expected — only the *value* of event_hash is excluded.)
+    """
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="alice",
+        epoch_id=1,
+        ts=0,
+        body={"document_root": _root_for("a")},
+    ).with_chain(prev_event_hash=None)
+    cb = env.canonical_bytes()
+    assert env.event_hash is not None
+    assert env.event_hash.encode() not in cb
+    # The "event_hash" key itself is NOT serialized — only "prev_event_hash" is.
+    assert b'"event_hash"' not in cb
+    assert b'"prev_event_hash"' in cb
+
+
+# ---------------------------------------------------------------------------
+# Receiver-side: first envelope, chain extension, fork detection
+# ---------------------------------------------------------------------------
+
+
+def test_first_announce_with_prev_none_accepted(tmp_path):
+    """A peer's very first envelope, prev_event_hash=None, is accepted."""
+    alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path)
+    _env, signed = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("b"), "source_uri": "https://x"},
+        prev_event_hash=None,
+        epoch_id=epoch,
+    )
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, body = srv.handle_announce(signed)
+    finally:
+        srv.stop()
+    assert status == 200, body
+
+    a_conn = connect(alice_db)
+    try:
+        row = a_conn.execute(
+            "SELECT last_event_hash, last_seq FROM mesh_peer_chains "
+            "WHERE peer_member_id = 'bob'"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert row is not None
+    assert row["last_event_hash"] == _env.event_hash
+    assert row["last_seq"] == 1
+
+
+def test_chain_extension_accepted(tmp_path):
+    """Second envelope with prev=first.event_hash extends the chain."""
+    alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path)
+    env1, signed1 = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("1"), "source_uri": "https://x1"},
+        prev_event_hash=None,
+        epoch_id=epoch,
+    )
+    env2, signed2 = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("2"), "source_uri": "https://x2"},
+        prev_event_hash=env1.event_hash,
+        epoch_id=epoch,
+    )
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        s1, _ = srv.handle_announce(signed1)
+        s2, _ = srv.handle_announce(signed2)
+    finally:
+        srv.stop()
+    assert s1 == 200
+    assert s2 == 200
+
+    a_conn = connect(alice_db)
+    try:
+        row = a_conn.execute(
+            "SELECT last_event_hash, last_seq FROM mesh_peer_chains "
+            "WHERE peer_member_id = 'bob'"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert row["last_event_hash"] == env2.event_hash
+    assert row["last_seq"] == 2
+
+
+def test_fork_detection_rejects_409(tmp_path):
+    """Second envelope with prev=GARBAGE is rejected — chain forked."""
+    alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path)
+    env1, signed1 = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("1"), "source_uri": "https://x"},
+        prev_event_hash=None,
+        epoch_id=epoch,
+    )
+    # Bob (or someone using bob's key) sends a second envelope claiming
+    # prev = arbitrary garbage instead of env1.event_hash.
+    _env2, signed2 = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("2"), "source_uri": "https://x"},
+        prev_event_hash="ff" * 32,
+        epoch_id=epoch,
+    )
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        s1, _ = srv.handle_announce(signed1)
+        s2, body = srv.handle_announce(signed2)
+    finally:
+        srv.stop()
+    assert s1 == 200
+    assert s2 == 409
+    assert "fork detected" in body["error"]
+    assert "bob" in body["error"]
+
+    # The fork rejection must NOT advance the peer's tracked chain.
+    a_conn = connect(alice_db)
+    try:
+        row = a_conn.execute(
+            "SELECT last_event_hash, last_seq FROM mesh_peer_chains "
+            "WHERE peer_member_id = 'bob'"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert row["last_event_hash"] == env1.event_hash
+    assert row["last_seq"] == 1
+
+
+def test_tampered_event_hash_rejected_400(tmp_path):
+    """An envelope whose event_hash doesn't match recomputation is 400."""
+    alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path)
+    env, signed = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("a"), "source_uri": "https://x"},
+        prev_event_hash=None,
+        epoch_id=epoch,
+    )
+    # Tamper with the event_hash field without re-signing. The envelope
+    # signature itself remains valid (event_hash is excluded from the
+    # signed canonical bytes), so the receiver catches this at the
+    # event_hash recomputation step, not the signature step.
+    signed["envelope"]["event_hash"] = "deadbeef" * 8
+    assert signed["envelope"]["event_hash"] != env.event_hash
+
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, body = srv.handle_announce(signed)
+    finally:
+        srv.stop()
+    assert status == 400
+    assert "event_hash mismatch" in body["error"]
+
+    # No peer-chain row written for a rejected envelope.
+    a_conn = connect(alice_db)
+    try:
+        row = a_conn.execute(
+            "SELECT * FROM mesh_peer_chains WHERE peer_member_id = 'bob'"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert row is None
+
+
+def test_chain_tracking_advances_across_multiple_announces(tmp_path):
+    """Receiver tracks last_event_hash + last_seq correctly over N announces."""
+    alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path)
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        prev = None
+        last_eh = None
+        for i in range(5):
+            env, signed = _signed_envelope(
+                sign_priv=bob.sign_priv,
+                sender_id="bob",
+                body={"document_root": _root_for(str(i)), "source_uri": f"https://x/{i}"},
+                prev_event_hash=prev,
+                epoch_id=epoch,
+            )
+            status, _ = srv.handle_announce(signed)
+            assert status == 200, f"announce {i} failed"
+            prev = env.event_hash
+            last_eh = env.event_hash
+    finally:
+        srv.stop()
+
+    a_conn = connect(alice_db)
+    try:
+        row = a_conn.execute(
+            "SELECT last_event_hash, last_seq FROM mesh_peer_chains "
+            "WHERE peer_member_id = 'bob'"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert row["last_event_hash"] == last_eh
+    assert row["last_seq"] == 5
+
+
+def test_two_peers_have_independent_chains(tmp_path):
+    """Bob's announces don't affect carol's chain state and vice versa."""
+    alice_db, epoch, _alice, bob, carol = _bootstrap_three_peer_db(tmp_path)
+
+    # Bob: 2 envelopes
+    bob_env1, bob_signed1 = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("b1"), "source_uri": "https://b1"},
+        prev_event_hash=None,
+        epoch_id=epoch,
+    )
+    bob_env2, bob_signed2 = _signed_envelope(
+        sign_priv=bob.sign_priv,
+        sender_id="bob",
+        body={"document_root": _root_for("b2"), "source_uri": "https://b2"},
+        prev_event_hash=bob_env1.event_hash,
+        epoch_id=epoch,
+    )
+    # Carol: 1 envelope, prev=None (her chain starts fresh — should be
+    # accepted even though bob has already posted).
+    carol_env1, carol_signed1 = _signed_envelope(
+        sign_priv=carol.sign_priv,
+        sender_id="carol",
+        body={"document_root": _root_for("c1"), "source_uri": "https://c1"},
+        prev_event_hash=None,
+        epoch_id=epoch,
+    )
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        assert srv.handle_announce(bob_signed1)[0] == 200
+        assert srv.handle_announce(carol_signed1)[0] == 200  # carol fresh start
+        assert srv.handle_announce(bob_signed2)[0] == 200    # bob extends his own chain
+    finally:
+        srv.stop()
+
+    a_conn = connect(alice_db)
+    try:
+        bob_row = a_conn.execute(
+            "SELECT last_event_hash, last_seq FROM mesh_peer_chains "
+            "WHERE peer_member_id = 'bob'"
+        ).fetchone()
+        carol_row = a_conn.execute(
+            "SELECT last_event_hash, last_seq FROM mesh_peer_chains "
+            "WHERE peer_member_id = 'carol'"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert bob_row["last_event_hash"] == bob_env2.event_hash
+    assert bob_row["last_seq"] == 2
+    assert carol_row["last_event_hash"] == carol_env1.event_hash
+    assert carol_row["last_seq"] == 1
+
+
+# ---------------------------------------------------------------------------
+# Schema migration: peer-chains table appears on legacy DBs
+# ---------------------------------------------------------------------------
+
+
+def test_mesh_peer_chains_table_exists_after_connect(tmp_path):
+    """Fresh connect() creates mesh_peer_chains via SCHEMA_SQL + migration."""
+    db = tmp_path / "fresh.db"
+    conn = connect(db)
+    try:
+        row = conn.execute(
+            "SELECT name FROM sqlite_master "
+            "WHERE type='table' AND name='mesh_peer_chains'"
+        ).fetchone()
+    finally:
+        conn.close()
+    assert row is not None
+
+
+def test_mesh_peer_chains_migration_idempotent(tmp_path):
+    """Two connects in a row don't ALTER twice."""
+    db = tmp_path / "twice.db"
+    connect(db).close()
+    conn = connect(db)
+    try:
+        cnt = conn.execute(
+            "SELECT COUNT(*) FROM sqlite_master "
+            "WHERE type='table' AND name='mesh_peer_chains'"
+        ).fetchone()[0]
+    finally:
+        conn.close()
+    assert cnt == 1
+
+
+def test_legacy_db_without_peer_chains_migrates(tmp_path):
+    """A DB created before the peer-chain rollout gains the table on open."""
+    import sqlite3 as _s
+    db = tmp_path / "legacy.db"
+    raw = _s.connect(db)
+    raw.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
+    raw.commit()
+    raw.close()
+
+    # Pre-state: no mesh_peer_chains.
+    raw = _s.connect(db)
+    pre = raw.execute(
+        "SELECT name FROM sqlite_master WHERE name='mesh_peer_chains'"
+    ).fetchone()
+    raw.close()
+    assert pre is None
+
+    # Open via connect() — runs SCHEMA_SQL + migration.
+    conn = connect(db)
+    try:
+        post = conn.execute(
+            "SELECT name FROM sqlite_master WHERE name='mesh_peer_chains'"
+        ).fetchone()
+    finally:
+        conn.close()
+    assert post is not None
+
+
+# ---------------------------------------------------------------------------
+# Backward compat: legacy envelopes (no chain fields) still accepted
+# ---------------------------------------------------------------------------
+
+
+def test_legacy_envelope_without_chain_fields_accepted(tmp_path):
+    """A WireEnvelope with event_hash=None (legacy fixture path) is accepted
+    without per-peer chain enforcement. Sig check still runs."""
+    alice_db, epoch, _alice, bob, _carol = _bootstrap_three_peer_db(tmp_path)
+    # Built directly without with_chain() — both prev and event_hash are None.
+    env = WireEnvelope(
+        type=TYPE_ANNOUNCE_ROOT,
+        sender_id="bob",
+        epoch_id=epoch,
+        ts=int(time.time()),
+        body={"document_root": _root_for("a"), "source_uri": "https://x"},
+    )
+    assert env.event_hash is None
+    signed = env.sign_with(bob.sign_priv)
+    srv = MeshWireServer(alice_db, host="127.0.0.1", port=0)
+    try:
+        status, _ = srv.handle_announce(signed)
+    finally:
+        srv.stop()
+    assert status == 200
+
+    # Legacy path doesn't write a peer-chain row.
+    a_conn = connect(alice_db)
+    try:
+        row = a_conn.execute(
+            "SELECT * FROM mesh_peer_chains WHERE peer_member_id='bob'"
+        ).fetchone()
+    finally:
+        a_conn.close()
+    assert row is None