"""Durable, peer-safe mesh relationship memory.

This module is intentionally separate from long-term memory (MEMORY.md / memory/shards/), USER.md, chat storage,
and session search. Its inputs are only mesh-visible social text after local
redaction, keyed by the peer's cryptographic fingerprint.

Regex-based redaction reduces obvious PII leakage; it cannot guarantee zero
narrative-identifiable detail. Mesh prompt policy is the layered control.
"""
from __future__ import annotations

import hashlib
import json
import logging
import os
import re
import threading
import time
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Any

from persistence import MUGI_DIR, secure_atomic_write


MESH_DIR = os.path.join(MUGI_DIR, "mesh")
RELATIONSHIPS_DIR = os.path.join(MESH_DIR, "relationships")
RAPPORT_MUTES_PATH = os.path.join(MESH_DIR, "rapport_mutes.json")
SCHEMA_VERSION = 2
DEFAULT_RECENT_TURN_LIMIT = 12
DEFAULT_MAX_SUMMARY_CHARS = 280
DEFAULT_CONTEXT_MAX_CHARS = 1800
DEFAULT_DIGEST_MAX_CHARS = 1200
STATS_CACHE_TTL_SECONDS = 3.0

# Defaults when config omits quotas (0 = unlimited for that dimension).
DEFAULT_GLOBAL_MAX_BYTES = 40 * 1024 * 1024
DEFAULT_PER_RECORD_MAX_BYTES = 1024 * 1024

_log = logging.getLogger(__name__)
_lock = threading.RLock()
_stats_cache_lock = threading.Lock()
_stats_cache: tuple[float, dict[str, Any]] | None = None
_rapport_writes_window: list[float] = []
_rapport_exports_window: list[float] = []

_EMAIL_RE = re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.IGNORECASE)
_HOME_PATH_RE = re.compile(r"(?:/Users|/home)/[^\s'\"<>)]{1,240}")
_WINDOWS_PATH_RE = re.compile(r"\b[A-Za-z]:\\[^\s'\"<>)]{1,240}")
_SECRET_ASSIGNMENT_RE = re.compile(
    r"(?i)\b(api[_-]?key|token|secret|password|passwd|bearer)\b\s*[:=]\s*[^\s'\"<>)]{4,}"
)
_LONG_NUMBER_RE = re.compile(r"\b\d[\d -]{7,}\d\b")
_URL_WITH_QUERY_RE = re.compile(r"https?://[^\s'\"<>)]{1,240}")
_CONTROL_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f]")
_WHITESPACE_RE = re.compile(r"\s+")


@dataclass(frozen=True)
class RelationshipWriteResult:
    ok: bool
    turn_index: int = 0
    path: str = ""
    error: str = ""


def _debug_enabled() -> bool:
    return str(os.environ.get("MUGI_MESH_LEDGER_DEBUG") or "").strip().lower() in {"1", "true", "yes", "on"}


def _now_iso() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def _fingerprint_key(fingerprint: str) -> str:
    fp = (fingerprint or "").strip()
    if not fp:
        return ""
    return hashlib.sha256(fp.encode("utf-8")).hexdigest()


def fingerprint_suffix(fingerprint: str, n: int = 8) -> str:
    fp = (fingerprint or "").strip()
    return fp[-n:] if len(fp) >= n else fp


def _mesh_quota_config() -> tuple[int, int, int, float]:
    """Returns global_max_bytes, per_record_max_bytes, stale_days, digest_brief_chars_scale."""
    try:
        from core.config import get_mesh_config

        mc = get_mesh_config()
        g = int(mc.get("mesh_rapport_global_max_bytes") or DEFAULT_GLOBAL_MAX_BYTES)
        r = int(mc.get("mesh_rapport_per_record_max_bytes") or DEFAULT_PER_RECORD_MAX_BYTES)
        stale = int(mc.get("mesh_rapport_stale_days") or 30)
        return (
            max(0, g),
            max(0, r),
            max(0, stale),
            0.75,
        )
    except Exception:
        return DEFAULT_GLOBAL_MAX_BYTES, DEFAULT_PER_RECORD_MAX_BYTES, 30, 0.75


def _invalidate_stats_cache_locked() -> None:
    global _stats_cache
    _stats_cache = None


def _invalidate_stats_cache() -> None:
    with _stats_cache_lock:
        _invalidate_stats_cache_locked()


def _realpaths_under(base: str, path: str) -> bool:
    try:
        br = os.path.realpath(base)
        pr = os.path.realpath(path)
        common = os.path.commonpath([br, pr])
    except ValueError:
        return False
    return common == br


def iter_relationship_json_files(base_dir: str) -> list[str]:
    """List *.json ledger files under base_dir, skipping symlinks and paths outside base."""
    if not base_dir or not os.path.isdir(base_dir):
        return []
    base_real = os.path.realpath(base_dir)
    found: list[str] = []
    for root, _dirs, files in os.walk(base_dir):
        root_real = os.path.realpath(root)
        try:
            if os.path.commonpath([base_real, root_real]) != base_real:
                continue
        except ValueError:
            continue
        for name in files:
            if not name.endswith(".json"):
                continue
            path = os.path.join(root, name)
            try:
                if os.path.islink(path):
                    continue
                if not os.path.isfile(path):
                    continue
                if not _realpaths_under(base_real, path):
                    continue
            except OSError:
                continue
            found.append(path)
    return found


def total_bytes_relationship_dir(base_dir: str) -> int:
    total = 0
    for path in iter_relationship_json_files(base_dir):
        try:
            total += os.path.getsize(path)
        except OSError:
            continue
    return total


def _record_json_size(record: dict[str, Any]) -> int:
    try:
        return len(json.dumps(record, ensure_ascii=False).encode("utf-8"))
    except Exception:
        return 0


def _migrate_record(data: dict[str, Any]) -> dict[str, Any]:
    ver = int(data.get("schema_version") or 0)
    if ver < 2:
        data.setdefault("last_compact_at", "")
        data.setdefault("source_lineage", {})
        data["schema_version"] = SCHEMA_VERSION
    return data


def _quarantine_corrupt_file(path: str) -> None:
    try:
        bad = f"{path}.bad.{int(time.time())}"
        os.replace(path, bad)
        _log.warning("mesh relationship ledger quarantined corrupt file -> %s", os.path.basename(bad))
    except Exception:
        _log.debug("failed to quarantine corrupt ledger %s", path, exc_info=True)


def _days_since_iso_ts(iso_ts: str) -> float:
    try:
        s = (iso_ts or "").strip()
        if not s:
            return 0.0
        if s.endswith("Z"):
            s = s[:-1] + "+00:00"
        dt = datetime.fromisoformat(s)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return (datetime.now(timezone.utc) - dt).total_seconds() / 86400.0
    except Exception:
        return 0.0


def _load_rapport_mutes_unlocked() -> set[str]:
    if not os.path.isfile(RAPPORT_MUTES_PATH):
        return set()
    try:
        with open(RAPPORT_MUTES_PATH, "r", encoding="utf-8") as f:
            raw = json.load(f)
        if not isinstance(raw, dict):
            return set()
        fps = raw.get("fingerprints")
        if not isinstance(fps, list):
            return set()
        return {str(x).strip() for x in fps if str(x).strip()}
    except Exception:
        return set()


def _save_rapport_mutes(muted: set[str]) -> None:
    os.makedirs(os.path.dirname(RAPPORT_MUTES_PATH), exist_ok=True)
    secure_atomic_write(RAPPORT_MUTES_PATH, json.dumps({"fingerprints": sorted(muted)}, indent=2, ensure_ascii=False))


def is_rapport_muted(fingerprint: str) -> bool:
    fp = (fingerprint or "").strip()
    if not fp:
        return False
    with _lock:
        muted = _load_rapport_mutes_unlocked()
        return fp in muted


def set_rapport_muted(fingerprint: str, muted: bool) -> bool:
    fp = (fingerprint or "").strip()
    if not fp:
        return False
    with _lock:
        cur = _load_rapport_mutes_unlocked()
        if muted:
            cur.add(fp)
        else:
            cur.discard(fp)
        _save_rapport_mutes(cur)
    _invalidate_stats_cache()
    return True


def list_rapport_mutes() -> list[str]:
    with _lock:
        return sorted(_load_rapport_mutes_unlocked())


def _path_for_fingerprint(fingerprint: str) -> str:
    key = _fingerprint_key(fingerprint)
    if not key:
        raise ValueError("peer fingerprint is required")
    return os.path.join(RELATIONSHIPS_DIR, key[:2], f"{key}.json")


def _safe_text(raw: str, *, max_chars: int = DEFAULT_MAX_SUMMARY_CHARS) -> str:
    text = _CONTROL_RE.sub("", str(raw or ""))
    text = _EMAIL_RE.sub("[redacted-email]", text)
    text = _HOME_PATH_RE.sub("[redacted-path]", text)
    text = _WINDOWS_PATH_RE.sub("[redacted-path]", text)
    text = _SECRET_ASSIGNMENT_RE.sub(lambda m: f"{m.group(1)}=[redacted-secret]", text)
    text = _LONG_NUMBER_RE.sub("[redacted-number]", text)

    def _redact_url(match: re.Match[str]) -> str:
        url = match.group(0)
        return "[redacted-url]" if "?" in url else url[:120]

    text = _URL_WITH_QUERY_RE.sub(_redact_url, text)
    text = _WHITESPACE_RE.sub(" ", text).strip()
    if len(text) > max_chars:
        text = text[: max(0, max_chars - 1)].rstrip() + "…"
    return text


def sanitize_peer_visible_text(raw: str, *, max_chars: int = DEFAULT_MAX_SUMMARY_CHARS) -> str:
    """Return a bounded, peer-safe summary/excerpt for ledger persistence."""
    from utils.llm_output_sanitize import strip_reasoning_preamble_leak, strip_think_tags, strip_untagged_cot

    t = str(raw or "")
    t = strip_reasoning_preamble_leak(strip_untagged_cot(strip_think_tags(t)))
    return _safe_text(t, max_chars=max_chars)


def _mentions_local_long_term_memory(text: str) -> bool:
    t = str(text or "")
    tl = t.lower()
    if "memory.md" in tl:
        return True
    if "memory/shards" in tl or "memory\\shards" in tl:
        return True
    return False


def contains_peer_sensitive_content(raw: str) -> bool:
    """Conservative guard for text that should not be sent to mesh peers."""
    text = str(raw or "")
    tl = text.lower()
    return bool(
        _EMAIL_RE.search(text)
        or _HOME_PATH_RE.search(text)
        or _WINDOWS_PATH_RE.search(text)
        or _SECRET_ASSIGNMENT_RE.search(text)
        or _LONG_NUMBER_RE.search(text)
        or _mentions_local_long_term_memory(text)
        or ("user.md" in tl)
    )


def _append_rapport_event_window(window: list[float], now: float) -> None:
    window[:] = [t for t in window if now - t < 3600.0]
    window.append(now)


def rapport_session_counters() -> dict[str, int]:
    now = time.time()
    with _lock:
        w = list(_rapport_writes_window)
        e = list(_rapport_exports_window)
    w = [t for t in w if now - t < 3600.0]
    e = [t for t in e if now - t < 3600.0]
    return {"rapport_writes_last_hour": len(w), "rapport_exports_last_hour": len(e)}


def record_rapport_export_event() -> None:
    with _lock:
        _append_rapport_event_window(_rapport_exports_window, time.time())


class RelationshipLedger:
    def __init__(self, *, base_dir: str | None = None, recent_turn_limit: int = DEFAULT_RECENT_TURN_LIMIT):
        self.base_dir = base_dir or RELATIONSHIPS_DIR
        self.recent_turn_limit = max(4, int(recent_turn_limit or DEFAULT_RECENT_TURN_LIMIT))
        self._last_corrupt_file_count = 0

    def _path(self, fingerprint: str) -> str:
        if self.base_dir == RELATIONSHIPS_DIR:
            return _path_for_fingerprint(fingerprint)
        key = _fingerprint_key(fingerprint)
        if not key:
            raise ValueError("peer fingerprint is required")
        return os.path.join(self.base_dir, key[:2], f"{key}.json")

    def _empty_record(self, fingerprint: str, *, peer_display: str = "", peer_hostname: str = "") -> dict[str, Any]:
        now = _now_iso()
        display = sanitize_peer_visible_text(peer_display or peer_hostname or "peer", max_chars=80) or "peer"
        return {
            "schema_version": SCHEMA_VERSION,
            "peer_fingerprint": fingerprint,
            "peer_display_resolved": display,
            "peer_hostname": sanitize_peer_visible_text(peer_hostname, max_chars=120),
            "rolling_summary": "",
            "recent_turns": [],
            "message_count": 0,
            "first_seen": now,
            "last_seen": now,
            "last_compact_at": "",
            "source_lineage": {
                "allowed_sources": ["mesh-visible-social-text"],
                "forbidden_sources": ["MEMORY.md", "memory/shards/", "USER.md", "main-chat", "session-search", "tool-private-data"],
            },
        }

    def read(self, fingerprint: str) -> dict[str, Any] | None:
        try:
            path = self._path(fingerprint)
        except Exception:
            return None
        if not os.path.isfile(path):
            if _debug_enabled():
                _log.info("mesh relationship ledger miss fp_suffix=%s", fingerprint_suffix(fingerprint))
            return None
        if os.path.islink(path):
            return None
        try:
            with open(path, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception:
            _log.warning("ledger read failed fp_suffix=%s; quarantine", fingerprint_suffix(fingerprint), exc_info=_debug_enabled())
            _quarantine_corrupt_file(path)
            return None
        if not isinstance(data, dict):
            _quarantine_corrupt_file(path)
            return None
        data = _migrate_record(dict(data))
        if _debug_enabled():
            _log.info(
                "mesh relationship ledger hit fp_suffix=%s chars=%d",
                fingerprint_suffix(fingerprint),
                len(json.dumps(data, ensure_ascii=False)),
            )
        return data

    def write(self, record: dict[str, Any]) -> str:
        fp = str(record.get("peer_fingerprint") or "").strip()
        path = self._path(fp)
        os.makedirs(os.path.dirname(path), exist_ok=True)
        record.setdefault("schema_version", SCHEMA_VERSION)
        secure_atomic_write(path, json.dumps(record, indent=2, ensure_ascii=False))
        _invalidate_stats_cache()
        return path

    def delete(self, fingerprint: str) -> bool:
        try:
            path = self._path(fingerprint)
        except Exception:
            return False
        try:
            if os.path.isfile(path):
                os.remove(path)
                _invalidate_stats_cache()
                return True
        except Exception:
            _log.debug("failed to delete mesh relationship ledger file", exc_info=True)
        return False

    def clear_all(self) -> int:
        count = 0
        for path in iter_relationship_json_files(self.base_dir):
            try:
                os.remove(path)
                count += 1
            except Exception:
                _log.debug("failed to remove mesh relationship ledger file %s", path, exc_info=True)
        _invalidate_stats_cache()
        return count

    def _enforce_record_byte_budget(self, record: dict[str, Any], budget: int) -> None:
        if budget <= 0:
            return
        recent = record.get("recent_turns")
        if not isinstance(recent, list):
            recent = []
        guard = 0
        while _record_json_size(record) > budget and len(recent) > 1 and guard < 200:
            guard += 1
            recent.pop(0)
            record["recent_turns"] = recent
        rolling = str(record.get("rolling_summary") or "")
        while _record_json_size(record) > budget and len(rolling) > 120:
            rolling = rolling[: max(80, len(rolling) // 2)].rstrip() + "…"
            record["rolling_summary"] = rolling
        guard = 0
        while _record_json_size(record) > budget and len(recent) > 1 and guard < 400:
            guard += 1
            recent.pop(0)
            record["recent_turns"] = recent

    def _project_bytes_if_write(self, path_write: str, record: dict[str, Any]) -> int:
        total = total_bytes_relationship_dir(self.base_dir)
        old = os.path.getsize(path_write) if os.path.isfile(path_write) else 0
        return total - old + _record_json_size(record)

    def _evict_one_lru_other(self, exclude_path: str) -> bool:
        metas: list[tuple[str, str]] = []
        for jp in iter_relationship_json_files(self.base_dir):
            if os.path.abspath(jp) == os.path.abspath(exclude_path):
                continue
            ls = ""
            try:
                with open(jp, "r", encoding="utf-8") as f:
                    d = json.load(f)
                if isinstance(d, dict):
                    ls = str(d.get("last_seen") or "")
            except Exception:
                ls = ""
            metas.append((ls, jp))
        metas.sort(key=lambda x: x[0])
        if not metas:
            return False
        _ls, jp = metas[0]
        try:
            os.remove(jp)
            _invalidate_stats_cache()
            _log.info("ledger evicted LRU for global quota tail=%s", jp[-52:])
            return True
        except Exception:
            return False

    def append_turn(
        self,
        *,
        fingerprint: str,
        direction: str,
        text: str,
        peer_display: str = "",
        peer_hostname: str = "",
        conversation_id: str = "",
        message_type: str = "",
        source: str = "",
    ) -> RelationshipWriteResult:
        fp = (fingerprint or "").strip()
        if not fp:
            return RelationshipWriteResult(ok=False, error="missing peer fingerprint")
        summary = sanitize_peer_visible_text(text)
        if not summary:
            return RelationshipWriteResult(ok=False, error="empty sanitized text")
        safe_direction = direction if direction in {"in", "out", "sys"} else "sys"

        with _lock:
            mutes_pre = _load_rapport_mutes_unlocked()
            if fp in mutes_pre:
                _log.info(
                    "mesh rapport mute skip fp_suffix=%s",
                    fingerprint_suffix(fp),
                    extra={
                        "event": "mesh_rapport_ledger",
                        "ledger_action": "refused",
                        "peer_fp_suffix": fingerprint_suffix(fp),
                        "reason": "rapport_muted",
                    },
                )
                return RelationshipWriteResult(ok=False, error="rapport_muted")

            global_max_eff, record_max_eff, stale_days_eff, _ = _mesh_quota_config()
            if global_max_eff <= 0:
                global_max_eff = DEFAULT_GLOBAL_MAX_BYTES
            if record_max_eff <= 0:
                record_max_eff = DEFAULT_PER_RECORD_MAX_BYTES

            record = self.read(fp) or self._empty_record(fp, peer_display=peer_display, peer_hostname=peer_hostname)

            prev_last_seen = str(record.get("last_seen") or "")
            if peer_display or peer_hostname:
                record["peer_display_resolved"] = sanitize_peer_visible_text(
                    peer_display or record.get("peer_display_resolved") or peer_hostname or "peer",
                    max_chars=80,
                ) or "peer"
                if peer_hostname:
                    record["peer_hostname"] = sanitize_peer_visible_text(peer_hostname, max_chars=120)
            now = _now_iso()
            next_index = int(record.get("message_count") or 0) + 1
            forced_compact = bool(
                stale_days_eff > 0
                and bool(prev_last_seen)
                and _days_since_iso_ts(prev_last_seen) >= float(stale_days_eff)
            )

            turn = {
                "index": next_index,
                "ts": now,
                "dir": safe_direction,
                "text_summary": summary,
                "conversation_id": sanitize_peer_visible_text(conversation_id, max_chars=120),
                "message_type": sanitize_peer_visible_text(message_type, max_chars=40),
                "source": sanitize_peer_visible_text(source, max_chars=40),
            }
            recent = record.get("recent_turns")
            if not isinstance(recent, list):
                recent = []
            recent.append(turn)
            record["recent_turns"] = recent[-self.recent_turn_limit :]
            record["message_count"] = next_index
            record["last_seen"] = now

            rolling_empty = not str(record.get("rolling_summary") or "").strip()
            if rolling_empty or forced_compact:
                record["rolling_summary"] = self._derive_lightweight_summary(record)
                if forced_compact:
                    record["last_compact_at"] = now
            elif next_index % self.recent_turn_limit == 0:
                record["rolling_summary"] = self._derive_lightweight_summary(record)
                record["last_compact_at"] = now
            else:
                pass

            self._enforce_record_byte_budget(record, record_max_eff)
            record["schema_version"] = SCHEMA_VERSION

            try:
                path = self._path(fp)
            except Exception as exc:
                return RelationshipWriteResult(ok=False, turn_index=next_index, error=str(exc))

            loop_guard = 0
            while global_max_eff and self._project_bytes_if_write(path, record) > global_max_eff and loop_guard < 400:
                loop_guard += 1
                if self._evict_one_lru_other(path):
                    continue
                self._enforce_record_byte_budget(record, max(record_max_eff // 2, 4096))
                if _record_json_size(record) <= 8192:
                    break

            if global_max_eff and self._project_bytes_if_write(path, record) > global_max_eff:
                return RelationshipWriteResult(ok=False, turn_index=next_index, error="mesh rapport global quota exceeded")

            try:
                path_written = self.write(record)
            except Exception as exc:
                _log.debug("failed to append mesh relationship ledger turn", exc_info=True)
                return RelationshipWriteResult(ok=False, turn_index=next_index, error=str(exc))

            _append_rapport_event_window(_rapport_writes_window, time.time())

        if _debug_enabled():
            _log.info(
                "mesh relationship ledger write fp_suffix=%s dir=%s turn=%s chars=%s",
                fp[-8:],
                safe_direction,
                next_index,
                len(summary),
                extra={
                    "event": "mesh_rapport_ledger",
                    "ledger_action": "append",
                    "peer_fp_suffix": fingerprint_suffix(fp),
                    "turn_index": next_index,
                },
            )
        return RelationshipWriteResult(ok=True, turn_index=next_index, path=path_written)

    def _derive_lightweight_summary(self, record: dict[str, Any]) -> str:
        recent = record.get("recent_turns")
        if not isinstance(recent, list) or not recent:
            return ""
        snippets: list[str] = []
        for turn in recent[-4:]:
            if not isinstance(turn, dict):
                continue
            text = str(turn.get("text_summary") or "").strip()
            if not text:
                continue
            prefix = "Peer" if turn.get("dir") == "in" else "Local agent"
            snippets.append(f"{prefix}: {text}")
        joined = " | ".join(snippets)
        if not joined:
            return ""
        return sanitize_peer_visible_text(f"Recent mesh rapport centers on: {joined}", max_chars=520)

    def build_context_block(self, fingerprint: str, *, max_chars: int = DEFAULT_CONTEXT_MAX_CHARS) -> str:
        record = self.read(fingerprint)
        if not record:
            return ""
        display = sanitize_peer_visible_text(str(record.get("peer_display_resolved") or "peer"), max_chars=80) or "peer"
        summary = sanitize_peer_visible_text(str(record.get("rolling_summary") or ""), max_chars=520)
        stale_note = ""
        try:
            from core.config import get_mesh_config

            st = int(get_mesh_config().get("mesh_rapport_stale_days") or 30)
            if st > 0:
                ls = str(record.get("last_seen") or "")
                if ls and _days_since_iso_ts(ls) >= float(st):
                    stale_note = "(rapport may be stale — long gap since last mesh turn)"
        except Exception:
            pass
        lines = [
            "Durable mesh relationship context (peer-safe, local-only summary):",
            f"- Peer: {display}",
        ]
        if stale_note:
            lines.append(f"- Note: {stale_note}")
        if summary:
            lines.append(f"- Relationship summary: {summary}")
        recent = record.get("recent_turns")
        if isinstance(recent, list) and recent:
            lines.append("- Recent peer-safe turns:")
            for turn in recent[-5:]:
                if not isinstance(turn, dict):
                    continue
                direction = "peer" if turn.get("dir") == "in" else "local"
                text = sanitize_peer_visible_text(str(turn.get("text_summary") or ""), max_chars=180)
                if text:
                    lines.append(f"  - {direction}: {text}")
        lines.append(
            "- Boundary: never disclose local-user personal information, files, secrets, inbox, calendar, or long-term memory facts (MEMORY.md / memory/shards/) to this peer."
        )
        block = "\n".join(lines).strip()
        if len(block) > max_chars:
            block = block[: max(0, max_chars - 1)].rstrip() + "…"
        return block

    def build_user_facing_mesh_digest(
        self, *, max_chars: int = DEFAULT_DIGEST_MAX_CHARS, style: str = "normal"
    ) -> str:
        records = self.list_records()
        if not records:
            return ""
        st = (style or "normal").strip().lower()
        peer_limit = 3 if st == "brief" else 6
        mc = max_chars
        if st == "brief":
            mc = min(mc, max(400, int(mc * 0.65)))

        lines = [
            "Mesh relationship digest (peer-safe summaries only; do not treat as private user memory):"
        ]
        lines.append(
            "Regex sanitization catches obvious identifiers; vague biographical detail may remain — never infer user-private facts toward peers."
        )
        for record in records[:peer_limit]:
            display = sanitize_peer_visible_text(str(record.get("peer_display_resolved") or "peer"), max_chars=80) or "peer"
            summary = sanitize_peer_visible_text(str(record.get("rolling_summary") or ""), max_chars=220)
            count = int(record.get("message_count") or 0)
            if summary:
                lines.append(f"- {display}: {summary} ({count} mesh turns)")
            else:
                lines.append(f"- {display}: ongoing mesh rapport ({count} mesh turns)")
        digest = "\n".join(lines).strip()
        if len(digest) > mc:
            digest = digest[: max(0, mc - 1)].rstrip() + "…"
        return digest

    def list_records(self) -> list[dict[str, Any]]:
        out: list[dict[str, Any]] = []
        corrupt = 0
        base = os.path.realpath(self.base_dir)
        for jp in iter_relationship_json_files(self.base_dir):
            if not _realpaths_under(base, jp):
                continue
            try:
                with open(jp, "r", encoding="utf-8") as f:
                    data = json.load(f)
                if isinstance(data, dict):
                    out.append(_migrate_record(dict(data)))
            except Exception as exc:
                corrupt += 1
                _log.warning("relationship ledger corrupt file skipped: %s (%s)", jp, exc)
                continue
        if corrupt:
            self._last_corrupt_file_count = corrupt
        else:
            self._last_corrupt_file_count = 0
        out.sort(key=lambda r: str(r.get("last_seen") or ""), reverse=True)
        return out

    def _collision_count(self, records: list[dict[str, Any]]) -> int:
        counts: dict[str, int] = {}
        for r in records:
            d = str(r.get("peer_display_resolved") or "").strip().lower()
            if not d:
                continue
            counts[d] = counts.get(d, 0) + 1
        return sum(max(0, c - 1) for c in counts.values())

    def stats(self) -> dict[str, Any]:
        global _stats_cache
        now_m = time.time()
        with _stats_cache_lock:
            if _stats_cache and (now_m - _stats_cache[0]) < STATS_CACHE_TTL_SECONDS:
                return dict(_stats_cache[1])
        payload = self._compute_stats_uncached()
        with _stats_cache_lock:
            _stats_cache = (now_m, payload)
        return dict(payload)

    def _compute_stats_uncached(self) -> dict[str, Any]:
        records = self.list_records()
        last_seen = ""
        if records:
            last_seen = str(records[0].get("last_seen") or "")
        total_b = total_bytes_relationship_dir(self.base_dir)
        try:
            gmax, _, _, _ = _mesh_quota_config()
            if gmax <= 0:
                gmax = DEFAULT_GLOBAL_MAX_BYTES
        except Exception:
            gmax = DEFAULT_GLOBAL_MAX_BYTES
        pressure = "ok"
        if gmax > 0:
            ratio = total_b / float(gmax)
            if ratio >= 0.98:
                pressure = "full"
            elif ratio >= 0.85:
                pressure = "warn"
        cts = rapport_session_counters()

        collisions = self._collision_count(records)

        return {
            "relationship_ledger_peer_count": len(records),
            "relationship_ledger_last_seen": last_seen,
            "relationship_ledger_dir": self.base_dir,
            "relationship_ledger_total_bytes": total_b,
            "relationship_ledger_collision_count": collisions,
            "relationship_ledger_corrupt_file_count": int(self._last_corrupt_file_count or 0),
            "relationship_quota_pressure": pressure,
            "relationship_ledger_cached_at": _now_iso(),
            **cts,
        }

    def export_payload(
        self, *, max_records: int = 200, omit_full_fingerprint: bool = True
    ) -> dict[str, Any]:
        recs_in = self.list_records()[: max(1, max_records)]
        recs_out: list[dict[str, Any]] = []
        for r in recs_in:
            item = dict(r)
            fp = str(item.pop("peer_fingerprint", "") or "")
            item["schema_version_export"] = 1
            item["summary"] = "sanitized rapport metadata; layered prompt policy applies for peer privacy."
            if omit_full_fingerprint:
                item["peer_fingerprint_suffix"] = fingerprint_suffix(fp, 16)
                item["peer_fingerprint_sha256"] = _fingerprint_key(fp)[:24] + "…"
            else:
                item["peer_fingerprint"] = fp
            recs_out.append(item)
        return {
            "sensitivity": "rapport_metadata_peer_safe",
            "records": recs_out,
            "exported_count": len(recs_out),
        }


_default_ledger = RelationshipLedger()


def get_relationship_ledger() -> RelationshipLedger:
    return _default_ledger


def append_turn(**kwargs: Any) -> RelationshipWriteResult:
    return _default_ledger.append_turn(**kwargs)


def build_context_block(fingerprint: str, *, max_chars: int = DEFAULT_CONTEXT_MAX_CHARS) -> str:
    return _default_ledger.build_context_block(fingerprint, max_chars=max_chars)


def build_user_facing_mesh_digest(*, max_chars: int = DEFAULT_DIGEST_MAX_CHARS, style: str | None = None) -> str:
    st = style
    if st is None:
        try:
            from core.config import get_mesh_config

            st = str(get_mesh_config().get("mesh_rapport_digest_style") or "normal")
        except Exception:
            st = "normal"
    return _default_ledger.build_user_facing_mesh_digest(max_chars=max_chars, style=st)


def delete_relationship(fingerprint: str) -> bool:
    return _default_ledger.delete(fingerprint)


def clear_all_relationships() -> int:
    out = _default_ledger.clear_all()
    _invalidate_stats_cache()
    return out


def relationship_stats() -> dict[str, Any]:
    return _default_ledger.stats()


def rapport_export_bundle(
    *,
    max_records: int = 200,
    omit_full_fingerprint: bool = True,
) -> dict[str, Any]:
    """JSON-serializable export dict for authenticated GET /mesh/relationships/export."""
    return _default_ledger.export_payload(
        max_records=max_records, omit_full_fingerprint=omit_full_fingerprint
    )

