"""Mesh encrypted file transfer: registry, AES-GCM framing, HKDF auth tickets."""
from __future__ import annotations

import base64
import hashlib
import hmac
import json
import logging
import os
import re
import shutil
import struct
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, BinaryIO, Iterator

import requests
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF

from core.config import get_mesh_config
from mesh.audit import append_audit
from mesh.crypto import encrypt_payload
from mesh.protocol import MeshMessage
from mesh.session import MeshSession

SCHEMA_V1 = 1
TRANSFER_AUTH_INFO = b"mugi-mesh-transfer-auth-v1"
_log = logging.getLogger(__name__)
FRAME_MAGIC = b"MFT1"
SERVE_PATH_PREFIX = "/mesh/file_transfers/serve"
AUTH_HEADER_PREFIX = "Mugi-Transfer v1 "


def host_for_url(host: str) -> str:
    h = (host or "").strip()
    if ":" in h and not h.startswith("["):
        return f"[{h}]"
    return h


def derive_auth_key(transfer_key: bytes, transfer_id: str) -> bytes:
    hkdf = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=transfer_id.encode("utf-8"),
        info=TRANSFER_AUTH_INFO,
    )
    return hkdf.derive(transfer_key)


def build_auth_ticket(
    auth_key: bytes,
    *,
    transfer_id: str,
    file_id: str,
    receiver_fingerprint: str,
    expiry_unix: int,
) -> str:
    msg = f"{transfer_id}\n{file_id}\n{receiver_fingerprint}\n{expiry_unix}".encode("utf-8")
    return hmac.new(auth_key, msg, hashlib.sha256).hexdigest()


def verify_auth_ticket(
    auth_key: bytes,
    ticket_hex: str,
    *,
    transfer_id: str,
    file_id: str,
    receiver_fingerprint: str,
    expiry_unix: int,
) -> bool:
    try:
        expected = build_auth_ticket(
            auth_key,
            transfer_id=transfer_id,
            file_id=file_id,
            receiver_fingerprint=receiver_fingerprint,
            expiry_unix=expiry_unix,
        )
        return hmac.compare_digest(expected.lower(), (ticket_hex or "").strip().lower())
    except Exception:
        return False


def parse_transfer_auth_header(value: str | None) -> str | None:
    if not value:
        return None
    v = value.strip()
    if not v.startswith(AUTH_HEADER_PREFIX):
        return None
    rest = v[len(AUTH_HEADER_PREFIX) :].strip()
    return rest or None


def sha256_file(path: str, chunk: int = 1024 * 1024) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        while True:
            b = f.read(chunk)
            if not b:
                break
            h.update(b)
    return h.hexdigest()


def _aad(transfer_id: str, file_id: str, chunk_index: int, total_size: int) -> bytes:
    return f"{transfer_id}|{file_id}|{chunk_index}|{total_size}".encode("utf-8")


def stream_cipher_body(
    *,
    transfer_id: str,
    file_id: str,
    total_size: int,
    path: str,
    key: bytes,
    chunk_size: int,
) -> Iterator[bytes]:
    yield FRAME_MAGIC
    aes = AESGCM(key)
    with open(path, "rb") as f:
        idx = 0
        while True:
            buf = f.read(chunk_size)
            if not buf:
                break
            nonce = os.urandom(12)
            aad = _aad(transfer_id, file_id, idx, total_size)
            ct = aes.encrypt(nonce, buf, aad)
            blob = nonce + ct
            yield struct.pack(">I", len(blob)) + blob
            idx += 1


_ILLEGAL_NAME = re.compile(r"[\x00-\x1f<>:\"|?*\\]")


def _pathconf_limits(path_for_fs: str) -> tuple[int, int]:
    """Return (PATH_MAX, NAME_MAX) for the filesystem containing *path_for_fs*.

    Uses :func:`os.pathconf` with conservative fallbacks when unavailable
    (e.g. exotic paths).
    """
    try:
        p = os.path.realpath(os.path.expanduser(path_for_fs))
        base = p if os.path.isdir(p) else (os.path.dirname(p) or p)
        pm = int(os.pathconf(base, "PC_PATH_MAX"))
        nm = int(os.pathconf(base, "PC_NAME_MAX"))
        if pm <= 0:
            pm = 1024
        if nm <= 0:
            nm = 255
        return pm, nm
    except OSError:
        return 1024, 255


def _relpath_components_respect_name_max(relpath: str, name_max: int) -> str | None:
    """Return an error string if any UTF-8 path component exceeds NAME_MAX."""
    for seg in (relpath or "").split("/"):
        if not seg:
            continue
        if len(seg.encode("utf-8")) > name_max:
            return f"Path component exceeds NAME_MAX ({name_max} bytes): {seg[:80]!r}"
    return None


def _abs_path_respects_path_max(abs_path: str, path_max: int) -> str | None:
    """Return an error string if the absolute path is too long for PATH_MAX."""
    try:
        b = os.fsencode(abs_path)
    except Exception:
        b = abs_path.encode("utf-8", errors="surrogateescape")
    if len(b) > path_max:
        return f"Absolute path exceeds PATH_MAX ({path_max} bytes)."
    return None


def _sanitize_relpath_component(name: str) -> str:
    n = (name or "").strip()
    if not n or n in {".", ".."}:
        raise ValueError("illegal path component")
    if "/" in n or "\\" in n:
        raise ValueError("path separators not allowed in manifest component")
    if _ILLEGAL_NAME.search(n):
        raise ValueError("illegal characters in file name")
    return n


def sanitize_relative_path(relpath: str) -> str:
    parts: list[str] = []
    for seg in (relpath or "").replace("\\", "/").split("/"):
        seg = seg.strip()
        if not seg:
            continue
        parts.append(_sanitize_relpath_component(seg))
    if not parts:
        raise ValueError("empty relative path")
    return "/".join(parts)


def unique_leaf_names(paths: list[str]) -> dict[str, str]:
    """Map absolute path -> single-directory relative name; disambiguate duplicate basenames."""
    out: dict[str, str] = {}
    basename_count: dict[str, int] = {}
    for p in paths:
        ap = os.path.realpath(os.path.expanduser(p))
        base = os.path.basename(ap)
        if not base:
            raise ValueError(f"invalid path: {p}")
        n = basename_count.get(base, 0) + 1
        basename_count[base] = n
        if n == 1:
            name = base
        else:
            root, ext = os.path.splitext(base)
            name = f"{root} ({n}){ext}"
        out[ap] = name
    return out


def _manifest_leaf_for_disk_file(abs_path: str, disk_leaf: str) -> str:
    """Manifest filename for transfer: chat upload original_name when path is a registered attachment.

    Temp uploads often use UUID (or other opaque) on-disk names; the registry keeps the
    user-visible ``original_name`` from the picker / drag-drop.
    """
    try:
        from attachment_registry import get_record_by_path
    except ImportError:
        return disk_leaf
    rec = get_record_by_path(abs_path)
    if rec is None:
        return disk_leaf
    on = (rec.original_name or "").strip()
    if not on:
        return disk_leaf
    leaf = os.path.basename(on.replace("\\", "/")) or disk_leaf
    if not leaf:
        return disk_leaf
    try:
        return _sanitize_relpath_component(leaf)
    except ValueError:
        return disk_leaf


def _disambiguate_transfer_relpaths(entries: list[tuple[str, str]]) -> dict[str, str]:
    """Map absolute path -> unique manifest relpath (stable order)."""
    counts: dict[str, int] = {}
    out: dict[str, str] = {}
    for ap, rel in entries:
        c = counts.get(rel, 0) + 1
        counts[rel] = c
        if c == 1:
            final = rel
        else:
            if "/" in rel:
                parent, leaf = rel.rsplit("/", 1)
            else:
                parent, leaf = "", rel
            base, ext = os.path.splitext(leaf)
            new_leaf = f"{base} ({c}){ext}"
            final = f"{parent}/{new_leaf}" if parent else new_leaf
        out[ap] = final
    return out


def expand_paths_for_outbound(
    raw_paths: list[str],
    *,
    max_files: int,
    max_walk_depth: int,
    skip_dotfiles: bool,
    max_relpath_bytes: int = 8000,
) -> tuple[list[tuple[str, str]], str | None]:
    """Expand files and directories to (abs_path, logical_relpath) pairs.

    Symlinks at roots and symlink files are rejected. Directory walk does not follow links.
    Returns (entries, error_message).
    """
    roots = list(
        dict.fromkeys(os.path.realpath(os.path.expanduser(p)) for p in raw_paths if str(p).strip())
    )
    if not roots:
        return [], "No file paths given."
    acc: list[tuple[str, str]] = []

    for ap in roots:
        if not os.path.exists(ap):
            return [], f"Path does not exist: {ap}"
        if os.path.islink(ap):
            return [], f"Symlinks are not allowed at transfer root: {ap}"
        path_max, name_max = _pathconf_limits(ap)

        if os.path.isfile(ap):
            if err := _abs_path_respects_path_max(ap, path_max):
                return [], err
            base = os.path.basename(ap)
            if not base:
                return [], f"invalid path: {ap}"
            manifest_leaf = _manifest_leaf_for_disk_file(ap, base)
            if err := _relpath_components_respect_name_max(manifest_leaf, name_max):
                return [], err
            try:
                rel = sanitize_relative_path(manifest_leaf)
            except ValueError as e:
                return [], str(e)
            acc.append((ap, rel))
            if len(acc) > max_files:
                return [], f"Too many files (max {max_files})."
            continue

        if os.path.isdir(ap):
            if err := _abs_path_respects_path_max(ap, path_max):
                return [], err
            root_label = os.path.basename(ap.rstrip(os.sep)) or "folder"
            try:
                root_prefix = sanitize_relative_path(root_label)
            except ValueError:
                root_prefix = "folder"
            if err := _relpath_components_respect_name_max(root_prefix, name_max):
                return [], err

            for dirpath, dirnames, filenames in os.walk(ap, topdown=True, followlinks=False):
                if skip_dotfiles:
                    dirnames[:] = [d for d in dirnames if not d.startswith(".")]
                rel_dir = os.path.relpath(dirpath, ap)
                if rel_dir == ".":
                    sub = ""
                else:
                    parts_rel = rel_dir.split(os.sep)
                    if skip_dotfiles and any(p.startswith(".") for p in parts_rel):
                        continue
                    try:
                        sub = "/".join(_sanitize_relpath_component(p) for p in parts_rel)
                    except ValueError as e:
                        return [], str(e)
                    if sub and (err := _relpath_components_respect_name_max(sub, name_max)):
                        return [], err

                for fn in filenames:
                    if skip_dotfiles and fn.startswith("."):
                        continue
                    fp = os.path.join(dirpath, fn)
                    if not os.path.isfile(fp):
                        continue
                    if os.path.islink(fp):
                        return [], f"Symlinks are not allowed: {fp}"
                    pm_file, nm_file = _pathconf_limits(fp)
                    if err := _abs_path_respects_path_max(fp, pm_file):
                        return [], err
                    manifest_fn = _manifest_leaf_for_disk_file(fp, fn)
                    try:
                        _sanitize_relpath_component(manifest_fn)
                    except ValueError as e:
                        return [], str(e)
                    if err := _relpath_components_respect_name_max(manifest_fn, nm_file):
                        return [], err
                    if sub:
                        rel_combined = f"{root_prefix}/{sub}/{manifest_fn}"
                    else:
                        rel_combined = f"{root_prefix}/{manifest_fn}"
                    try:
                        rel_final = sanitize_relative_path(rel_combined)
                    except ValueError as e:
                        return [], str(e)
                    if len(rel_final.split("/")) > max_walk_depth:
                        return [], f"Path too deep (max depth {max_walk_depth}): {rel_final}"
                    if len(rel_final.encode("utf-8")) > max_relpath_bytes:
                        return [], "Relative path too long for manifest."
                    if err := _relpath_components_respect_name_max(rel_final, nm_file):
                        return [], err
                    acc.append((fp, rel_final))
                    if len(acc) > max_files:
                        return [], f"Too many files (max {max_files})."
            continue

        return [], f"Not a file or directory: {ap}"

    if not acc:
        return [], "No files to transfer (empty folders or only skipped items)."
    rel_map = _disambiguate_transfer_relpaths(acc)
    ordered = [(ap, rel_map[ap]) for ap, _ in acc]
    return ordered, None


def _path_is_inside_directory(dir_real: str, candidate: str) -> bool:
    """True if realpath(candidate) is dir_real or nested inside it."""
    d = os.path.realpath(dir_real)
    c = os.path.realpath(candidate)
    if c == d:
        return True
    prefix = d if d.endswith(os.sep) else d + os.sep
    return c.startswith(prefix)


def decrypt_frame_stream(
    raw: BinaryIO,
    *,
    transfer_id: str,
    file_id: str,
    total_size: int,
    key: bytes,
) -> Iterator[bytes]:
    magic = raw.read(4)
    if magic != FRAME_MAGIC:
        raise ValueError("bad stream magic")
    aes = AESGCM(key)
    idx = 0
    while True:
        hdr = raw.read(4)
        if not hdr:
            break
        if len(hdr) < 4:
            raise ValueError("truncated frame length")
        (ln,) = struct.unpack(">I", hdr)
        if ln < 12 or ln > 16 * 1024 * 1024:
            raise ValueError("bad frame length")
        blob = raw.read(ln)
        if len(blob) != ln:
            raise ValueError("truncated frame body")
        nonce, ct = blob[:12], blob[12:]
        aad = _aad(transfer_id, file_id, idx, total_size)
        pt = aes.decrypt(nonce, ct, aad)
        yield pt
        idx += 1


@dataclass
class OutboundFile:
    file_id: str
    abs_path: str
    relpath: str
    size: int
    sha256_hex: str


@dataclass
class OutboundTransfer:
    transfer_id: str
    session_id: str
    receiver_fingerprint: str
    key: bytes
    expiry_unix: int
    host: str
    port: int
    files: list[OutboundFile]
    created_ts: float = field(default_factory=time.time)
    cancelled: bool = False
    client_source: str | None = None


@dataclass
class InboundOffer:
    transfer_id: str
    session_id: str
    sender_fingerprint: str
    sender_hostname: str
    receiver_fingerprint: str
    key: bytes
    expiry_unix: int
    host: str
    port: int
    files: list[dict[str, Any]]
    received_ts: float = field(default_factory=time.time)
    last_error: str | None = None
    # Staged receive (prepare + per-file fetch) for progress UIs.
    mesh_accept_sent: bool = False
    accept_dest: str | None = None
    accept_planned: list[tuple[str, str, dict[str, Any]]] | None = None
    fetch_completed_ids: list[str] = field(default_factory=list)


class FileTransferRegistry:
    """Outbound/inbound state shared by mesh handler and HTTP routes."""

    def __init__(self, manager: Any) -> None:
        self._mgr = manager
        self._lock = threading.RLock()
        self._outbound: dict[str, OutboundTransfer] = {}
        self._inbound: dict[str, InboundOffer] = {}

    def _cfg(self) -> dict[str, Any]:
        return get_mesh_config()

    def is_enabled(self) -> bool:
        return bool(self._cfg().get("file_transfer_enabled", True))

    def advertise_host_port(self) -> tuple[str, int]:
        cfg = self._cfg()
        port = int(getattr(self._mgr, "port", 0) or 0)
        configured = str(cfg.get("advertise_host") or "").strip()
        if configured:
            host = configured
        else:
            host = str(getattr(self._mgr.identity, "hostname", "") or "127.0.0.1")
            try:
                import socket

                candidates: set[str] = set()
                for name in (socket.gethostname(), socket.getfqdn()):
                    if not name:
                        continue
                    try:
                        _n, _a, ips = socket.gethostbyname_ex(name)
                        for ip in ips:
                            if ip and not ip.startswith("127."):
                                candidates.add(ip)
                    except Exception:
                        pass
                if candidates:
                    host = sorted(candidates)[0]
            except Exception:
                pass
            local_addrs = getattr(self._mgr, "_local_addresses", None) or set()
            if host.startswith("127.") or host in {"::1", "localhost"}:
                non_loop = sorted(
                    ip for ip in local_addrs if ip and not ip.startswith("127.") and ip not in {"::1"}
                )
                if non_loop:
                    host = non_loop[0]
        return host, port

    def _send_file_transfer_control(
        self, *, session_id: str, message: MeshMessage, op: str
    ) -> None:
        for attempt in range(2):
            try:
                self._mgr.send_message(session_id=session_id, message=message, source="file_transfer")
                return
            except Exception as exc:
                _log.debug(
                    "file_transfer %s notify failed (attempt %d): %s",
                    op,
                    attempt + 1,
                    exc,
                )
                if attempt == 0:
                    time.sleep(0.25)

    def _verify_offer_fits_mesh_wire(
        self,
        *,
        session_id: str,
        transfer_id: str,
        receiver_fingerprint: str,
        expiry_unix: int,
        host: str,
        port: int,
        key: bytes,
        files: list[OutboundFile],
    ) -> str | None:
        """Return an error string if the offer cannot be sent on the mesh (size limits)."""
        session = self._mgr._get_session(session_id)
        if session is None:
            return "Connect to the peer in Mesh first (no active session)."
        get_max = getattr(self._mgr, "_get_message_max_bytes", None)
        max_msg = int(get_max()) if callable(get_max) else 4 * 1024 * 1024
        payload = {
            "schema": SCHEMA_V1,
            "transfer_id": transfer_id,
            "sender_fingerprint": self._mgr.identity.fingerprint,
            "receiver_fingerprint": receiver_fingerprint.strip(),
            "expiry_unix": expiry_unix,
            "host": host,
            "port": port,
            "serve_path_prefix": SERVE_PATH_PREFIX,
            "key_b64": base64.b64encode(key).decode("ascii"),
            "files": [
                {"id": f.file_id, "relpath": f.relpath, "size": f.size, "sha256_hex": f.sha256_hex}
                for f in files
            ],
        }
        ttl = int(self._cfg().get("file_transfer_offer_ttl_seconds", 3600))
        msg = MeshMessage(
            type="file_transfer_offer",
            conversation_id=f"file-transfer:{transfer_id}",
            conversation_type="task",
            message_number=1,
            payload=payload,
            ttl=min(3600, max(300, ttl)),
        )
        try:
            raw = json.dumps(msg.model_dump(), ensure_ascii=False).encode("utf-8")
        except Exception as exc:
            return f"invalid offer message: {exc}"
        if len(raw) > max_msg:
            return (
                "Mesh offer exceeds max payload size; send fewer files or raise [mesh] message_max_bytes."
            )
        try:
            enc = encrypt_payload(session.session_key, msg.model_dump(), 0)
            wire_obj = {"session_id": session_id, **enc}
            wire_raw = json.dumps(wire_obj).encode("utf-8")
        except Exception as exc:
            return f"could not build encrypted mesh payload: {exc}"
        if len(wire_raw) > max_msg:
            return (
                "Encrypted mesh offer exceeds max payload size; raise [mesh] message_max_bytes."
            )
        return None

    def prepare_outbound(
        self,
        *,
        session_id: str,
        receiver_fingerprint: str,
        paths: list[str],
        client_source: str | None = None,
    ) -> dict[str, Any]:
        if not self.is_enabled():
            return {"ok": False, "error": "file_transfer disabled in config ([mesh] file_transfer_enabled)."}
        cfg = self._cfg()
        max_bytes = int(cfg.get("file_transfer_max_bytes", 100 * 1024 * 1024 * 1024))
        max_files = int(cfg.get("file_transfer_max_files", 50))
        ttl = int(cfg.get("file_transfer_offer_ttl_seconds", 3600))

        session = self._mgr._get_session(session_id)
        if session is None:
            return {"ok": False, "error": "Connect to the peer in Mesh first (no active session)."}
        if session.peer_fingerprint.strip() != (receiver_fingerprint or "").strip():
            return {"ok": False, "error": "Receiver fingerprint does not match this session's peer."}

        max_conc = int(cfg.get("file_transfer_max_concurrent_transfers", 4))
        walk_depth = int(cfg.get("file_transfer_max_walk_depth", 64))
        skip_dots = bool(cfg.get("file_transfer_skip_dotfiles", True))
        with self._lock:
            self._prune_expired_locked()
            now_t = time.time()
            active_out = sum(
                1 for o in self._outbound.values() if (not o.cancelled) and o.expiry_unix > now_t
            )
        if active_out >= max_conc:
            return {
                "ok": False,
                "error": f"Too many concurrent outbound transfers (max {max_conc}).",
            }

        ordered, expand_err = expand_paths_for_outbound(
            list(paths),
            max_files=max_files,
            max_walk_depth=walk_depth,
            skip_dotfiles=skip_dots,
        )
        if expand_err:
            return {"ok": False, "error": expand_err}

        files: list[OutboundFile] = []
        total = 0
        for ap, relp in ordered:
            if not os.path.isfile(ap):
                return {"ok": False, "error": f"Not a regular file: {ap}"}
            if os.path.islink(ap):
                return {"ok": False, "error": f"Symlinks are not allowed: {ap}"}
            st = os.stat(ap)
            sz = int(st.st_size)
            total += sz
            if total > max_bytes:
                return {"ok": False, "error": f"Total size exceeds cap ({max_bytes} bytes)."}
            try:
                h = sha256_file(ap)
            except OSError as exc:
                return {"ok": False, "error": f"Could not read file (offline or permissions?): {exc}"}
            fid = f"f{len(files)}"
            files.append(OutboundFile(file_id=fid, abs_path=ap, relpath=relp, size=sz, sha256_hex=h))

        tid = uuid.uuid4().hex
        key = os.urandom(32)
        expiry = int(time.time()) + max(60, ttl)
        host, port = self.advertise_host_port()

        wire_err = self._verify_offer_fits_mesh_wire(
            session_id=session_id,
            transfer_id=tid,
            receiver_fingerprint=receiver_fingerprint,
            expiry_unix=expiry,
            host=host,
            port=port,
            key=key,
            files=files,
        )
        if wire_err:
            return {"ok": False, "error": wire_err}

        src = (client_source or "").strip() or None
        ot = OutboundTransfer(
            transfer_id=tid,
            session_id=session_id,
            receiver_fingerprint=receiver_fingerprint.strip(),
            key=key,
            expiry_unix=expiry,
            host=host,
            port=port,
            files=files,
            client_source=src,
        )
        with self._lock:
            self._prune_expired_locked()
            self._outbound[tid] = ot

        return {
            "ok": True,
            "transfer_id": tid,
            "expiry_unix": expiry,
            "host": host,
            "port": port,
            "serve_path_prefix": SERVE_PATH_PREFIX,
            "sender_fingerprint": self._mgr.identity.fingerprint,
            "receiver_fingerprint": ot.receiver_fingerprint,
            "files": [
                {"id": f.file_id, "relpath": f.relpath, "size": f.size, "sha256_hex": f.sha256_hex}
                for f in files
            ],
            "session_id": session_id,
        }

    def commit_outbound(self, *, transfer_id: str, session_id: str) -> dict[str, Any]:
        if not self.is_enabled():
            return {"ok": False, "error": "file_transfer disabled"}
        with self._lock:
            self._prune_expired_locked()
            ot = self._outbound.get(transfer_id)
        if ot is None or ot.cancelled:
            return {"ok": False, "error": "Unknown or cancelled transfer."}
        if ot.session_id != session_id:
            return {"ok": False, "error": "Session mismatch."}
        if time.time() > ot.expiry_unix:
            return {"ok": False, "error": "Transfer draft expired; prepare again."}

        payload = {
            "schema": SCHEMA_V1,
            "transfer_id": ot.transfer_id,
            "sender_fingerprint": self._mgr.identity.fingerprint,
            "receiver_fingerprint": ot.receiver_fingerprint,
            "expiry_unix": ot.expiry_unix,
            "host": ot.host,
            "port": ot.port,
            "serve_path_prefix": SERVE_PATH_PREFIX,
            "key_b64": base64.b64encode(ot.key).decode("ascii"),
            "files": [
                {"id": f.file_id, "relpath": f.relpath, "size": f.size, "sha256_hex": f.sha256_hex}
                for f in ot.files
            ],
        }
        convo_id = f"file-transfer:{ot.transfer_id}"
        ttl = int(self._cfg().get("file_transfer_offer_ttl_seconds", 3600))
        msg = MeshMessage(
            type="file_transfer_offer",
            conversation_id=convo_id,
            conversation_type="task",
            message_number=1,
            payload=payload,
            ttl=min(3600, max(300, ttl)),
        )
        out = self._mgr.send_message(session_id=session_id, message=msg, source="file_transfer")
        if (
            out.get("ok")
            and (ot.client_source == "quick_panel")
            and bool(self._cfg().get("file_transfer_audit_quick_panel", False))
        ):
            sess = self._mgr._get_session(session_id)
            peer_name = getattr(sess, "peer_hostname", "") if sess else ""
            append_audit(
                {
                    "dir": "out",
                    "source": "quick_panel",
                    "peer": ot.receiver_fingerprint,
                    "peer_name": peer_name,
                    "type": "file_transfer_quick_panel_send",
                    "conversation_id": convo_id,
                    "summary": f"offer {ot.transfer_id} ({len(ot.files)} files)",
                    "bytes": 0,
                }
            )
        return out

    def cancel_outbound(
        self, *, transfer_id: str, session_id: str, notify_peer: bool = True
    ) -> dict[str, Any]:
        client_src: str | None = None
        recv_fp = ""
        with self._lock:
            ot = self._outbound.get(transfer_id)
            if ot is None or ot.session_id != session_id:
                return {"ok": False, "error": "Unknown transfer."}
            client_src = ot.client_source
            recv_fp = ot.receiver_fingerprint
            ot.cancelled = True
            del self._outbound[transfer_id]
        if (
            (not notify_peer)
            and client_src == "quick_panel"
            and bool(self._cfg().get("file_transfer_audit_quick_panel", False))
        ):
            sess = self._mgr._get_session(session_id)
            peer_name = getattr(sess, "peer_hostname", "") if sess else ""
            append_audit(
                {
                    "dir": "out",
                    "source": "quick_panel",
                    "peer": recv_fp,
                    "peer_name": peer_name,
                    "type": "file_transfer_quick_panel_draft_cancelled",
                    "conversation_id": f"file-transfer:{transfer_id}",
                    "summary": f"draft {transfer_id} discarded (no offer sent)",
                    "bytes": 0,
                }
            )
        if notify_peer:
            msg = MeshMessage(
                type="file_transfer_cancel",
                conversation_id=f"file-transfer:{transfer_id}",
                conversation_type="task",
                message_number=1,
                payload={"schema": SCHEMA_V1, "transfer_id": transfer_id},
                ttl=300,
            )
            self._send_file_transfer_control(session_id=session_id, message=msg, op="cancel")
        return {"ok": True}

    def serve_file_stream(
        self,
        *,
        transfer_id: str,
        file_id: str,
        ticket_hex: str,
    ) -> tuple[int, Iterator[bytes] | None, str]:
        cfg = self._cfg()
        chunk_size = int(cfg.get("file_transfer_serve_chunk_size", 256 * 1024))
        if not ticket_hex:
            return 401, None, "unauthorized"
        with self._lock:
            self._prune_expired_locked()
            ot = self._outbound.get(transfer_id)
        if ot is None or ot.cancelled:
            return 404, None, "not found"
        if time.time() > ot.expiry_unix:
            return 410, None, "expired"
        fe = next((f for f in ot.files if f.file_id == file_id), None)
        if fe is None:
            return 404, None, "file not in transfer"
        auth_key = derive_auth_key(ot.key, transfer_id)
        if not verify_auth_ticket(
            auth_key,
            ticket_hex,
            transfer_id=transfer_id,
            file_id=file_id,
            receiver_fingerprint=ot.receiver_fingerprint,
            expiry_unix=ot.expiry_unix,
        ):
            return 403, None, "bad ticket"
        try:
            live_hash = sha256_file(fe.abs_path)
        except OSError:
            return 500, None, "read error"
        if live_hash.lower() != fe.sha256_hex.lower():
            return 409, None, "file changed since offer"

        def gen() -> Iterator[bytes]:
            try:
                yield from stream_cipher_body(
                    transfer_id=transfer_id,
                    file_id=file_id,
                    total_size=fe.size,
                    path=fe.abs_path,
                    key=ot.key,
                    chunk_size=chunk_size,
                )
            except Exception as exc:
                _log.error(
                    "file_transfer serve stream failed transfer=%s file=%s: %s",
                    transfer_id,
                    file_id,
                    exc,
                    exc_info=True,
                )
                raise

        return 200, gen(), ""

    def _prune_expired_locked(self) -> None:
        now = time.time()
        for tid in [t for t, o in self._outbound.items() if o.expiry_unix <= now]:
            self._outbound.pop(tid, None)
        for tid in [t for t, o in self._inbound.items() if o.expiry_unix <= now]:
            self._inbound.pop(tid, None)

    @staticmethod
    def _is_loopback_serve_host(host: str) -> bool:
        h = (host or "").strip().lower()
        if not h:
            return True
        if h in ("127.0.0.1", "localhost", "::1", "[::1]"):
            return True
        return h.startswith("127.")

    def _coerce_serve_host_port(
        self, host: str, port: int, session: MeshSession
    ) -> tuple[str, int]:
        """Prefer the mesh session peer address when the offer lists loopback or bad port.

        The sender's ``advertise_host_port()`` can resolve to 127.0.0.1; the
        receiver must fetch files from the *peer* HTTP endpoint (same as mesh).
        """
        h = (host or "").strip()
        p = int(port)
        pa = (session.peer_address or "").strip()
        pp = int(session.peer_port or 0)
        if (not h or self._is_loopback_serve_host(h)) and pa:
            h = pa
        if p <= 0 and pp > 0:
            p = pp
        if not h and pa:
            h = pa
        return h, p

    def send_outbound(
        self,
        *,
        session_id: str,
        receiver_fingerprint: str,
        paths: list[str],
        client_source: str | None = None,
    ) -> dict[str, Any]:
        """Prepare and commit in one step (user-facing "send" — same work as before)."""
        prep = self.prepare_outbound(
            session_id=session_id,
            receiver_fingerprint=receiver_fingerprint,
            paths=paths,
            client_source=client_source,
        )
        if not prep.get("ok"):
            return prep
        tid = str(prep.get("transfer_id") or "")
        cmt = self.commit_outbound(transfer_id=tid, session_id=session_id)
        if not cmt.get("ok"):
            try:
                self.cancel_outbound(transfer_id=tid, session_id=session_id, notify_peer=False)
            except Exception:
                pass
            return cmt
        return {
            "ok": True,
            "transfer_id": tid,
            "files": prep.get("files"),
            "host": prep.get("host"),
            "port": prep.get("port"),
        }

    def on_mesh_offer(self, session: MeshSession, msg: MeshMessage) -> None:
        payload = msg.payload or {}
        if int(payload.get("schema") or 0) != SCHEMA_V1:
            return
        tid = str(payload.get("transfer_id") or "")
        sfp = str(payload.get("sender_fingerprint") or "")
        rfp = str(payload.get("receiver_fingerprint") or "")
        if not tid or not sfp or not rfp:
            return
        if sfp != session.peer_fingerprint:
            return
        if rfp != self._mgr.identity.fingerprint:
            return
        try:
            key = base64.b64decode(str(payload.get("key_b64") or ""), validate=True)
        except Exception:
            return
        if len(key) != 32:
            return
        expiry = int(payload.get("expiry_unix") or 0)
        if expiry <= int(time.time()):
            return
        files = payload.get("files") or []
        if not isinstance(files, list) or not files:
            return
        raw_host = str(payload.get("host") or "")
        raw_port = int(payload.get("port") or 0)
        host, port = self._coerce_serve_host_port(raw_host, raw_port, session)
        if not host or port <= 0:
            append_audit(
                {
                    "dir": "in",
                    "source": "peer",
                    "peer": session.peer_fingerprint,
                    "peer_name": session.peer_hostname,
                    "type": "file_transfer_offer_ignored",
                    "conversation_id": msg.conversation_id,
                    "summary": f"unusable serve address ({raw_host!r}:{raw_port!r})",
                    "bytes": 0,
                }
            )
            return
        offer = InboundOffer(
            transfer_id=tid,
            session_id=session.session_id,
            sender_fingerprint=sfp,
            sender_hostname=session.peer_hostname,
            receiver_fingerprint=rfp,
            key=key,
            expiry_unix=expiry,
            host=host,
            port=port,
            files=list(files),
        )
        with self._lock:
            self._prune_expired_locked()
            self._inbound[tid] = offer
        append_audit(
            {
                "dir": "in",
                "source": "peer",
                "peer": session.peer_fingerprint,
                "peer_name": session.peer_hostname,
                "type": "file_transfer_offer",
                "conversation_id": msg.conversation_id,
                "summary": f"offer {tid} ({len(files)} files)",
                "bytes": 0,
            }
        )

    def list_inbound_pending(self) -> list[dict[str, Any]]:
        now = time.time()
        with self._lock:
            self._prune_expired_locked()
            items = list(self._inbound.values())
        out: list[dict[str, Any]] = []
        for o in items:
            if o.expiry_unix <= now:
                continue
            row: dict[str, Any] = {
                "transfer_id": o.transfer_id,
                "sender_fingerprint": o.sender_fingerprint,
                "sender_hostname": o.sender_hostname,
                "receiver_fingerprint": o.receiver_fingerprint,
                "expiry_unix": o.expiry_unix,
                "files": o.files,
                "session_id": o.session_id,
            }
            if o.last_error:
                row["last_error"] = o.last_error
            out.append(row)
        return out

    def decline_inbound(self, *, transfer_id: str, session_id: str | None = None) -> dict[str, Any]:
        with self._lock:
            offer = self._inbound.pop(transfer_id, None)
        if offer is None:
            return {"ok": False, "error": "Unknown offer."}
        if session_id and offer.session_id != session_id:
            return {"ok": False, "error": "Session mismatch."}
        msg = MeshMessage(
            type="file_transfer_decline",
            conversation_id=f"file-transfer:{transfer_id}",
            conversation_type="task",
            message_number=1,
            payload={"schema": SCHEMA_V1, "transfer_id": transfer_id},
            ttl=300,
        )
        self._send_file_transfer_control(session_id=offer.session_id, message=msg, op="decline")
        return {"ok": True}

    def _get_inbound_offer_unlocked(
        self, transfer_id: str, session_id: str | None
    ) -> tuple[InboundOffer | None, dict[str, Any] | None]:
        """Return (offer, error_response) — one of them is always set."""
        self._prune_expired_locked()
        offer = self._inbound.get(transfer_id)
        if offer is None:
            return None, {"ok": False, "error": "Unknown offer."}
        if session_id and offer.session_id != session_id:
            return None, {"ok": False, "error": "Session mismatch."}
        if time.time() > offer.expiry_unix:
            return None, {"ok": False, "error": "Offer expired."}
        return offer, None

    def _build_inbound_planned_writes(
        self, offer: InboundOffer, dest: str
    ) -> list[tuple[str, str, dict[str, Any]]] | dict[str, Any]:
        cfg = self._cfg()
        max_bytes = int(cfg.get("file_transfer_max_bytes", 100 * 1024 * 1024 * 1024))
        total_need = sum(int(f.get("size") or 0) for f in offer.files if isinstance(f, dict))
        if total_need > max_bytes:
            return {"ok": False, "error": "Offer exceeds local size cap."}
        dest_real = os.path.realpath(os.path.expanduser(dest))
        du = shutil.disk_usage(dest_real)
        if du.free < total_need + 65536:
            return {"ok": False, "error": "Insufficient disk space."}
        planned: list[tuple[str, str, dict[str, Any]]] = []
        for fe in offer.files:
            if not isinstance(fe, dict):
                continue
            fid = str(fe.get("id") or "")
            rel_raw = str(fe.get("relpath") or "")
            try:
                safe_rel = sanitize_relative_path(rel_raw)
            except ValueError as e:
                return {"ok": False, "error": f"Bad manifest path: {e}"}
            path_max, name_max = _pathconf_limits(dest_real)
            if err := _relpath_components_respect_name_max(safe_rel, name_max):
                return {"ok": False, "error": err}
            rel_parts = safe_rel.split("/")
            out_path = dest_real
            for part in rel_parts:
                out_path = os.path.join(out_path, part)
            try:
                out_path = os.path.realpath(out_path)
            except OSError as e:
                return {"ok": False, "error": f"Bad destination path: {e}"}
            if err := _abs_path_respects_path_max(out_path, path_max):
                return {"ok": False, "error": err}
            if not _path_is_inside_directory(dest_real, out_path):
                return {"ok": False, "error": "Manifest path escapes destination directory."}
            planned.append((fid, out_path, fe))
        return planned

    def _send_mesh_accept_if_needed(self, offer: InboundOffer) -> None:
        if offer.mesh_accept_sent:
            return
        acc = MeshMessage(
            type="file_transfer_accept",
            conversation_id=f"file-transfer:{offer.transfer_id}",
            conversation_type="task",
            message_number=1,
            payload={"schema": SCHEMA_V1, "transfer_id": offer.transfer_id},
            ttl=300,
        )
        try:
            self._mgr.send_message(session_id=offer.session_id, message=acc, source="file_transfer")
        except Exception:
            pass
        offer.mesh_accept_sent = True

    def _download_one_inbound_file(
        self,
        offer: InboundOffer,
        *,
        transfer_id: str,
        fid: str,
        out_path: str,
        meta: dict[str, Any],
    ) -> str | None:
        """Return None on success, or an error string."""
        auth_key = derive_auth_key(offer.key, transfer_id)
        base = f"http://{host_for_url(offer.host)}:{offer.port}"
        ticket = build_auth_ticket(
            auth_key,
            transfer_id=transfer_id,
            file_id=fid,
            receiver_fingerprint=offer.receiver_fingerprint,
            expiry_unix=offer.expiry_unix,
        )
        url = f"{base}{SERVE_PATH_PREFIX}/{transfer_id}/{fid}"
        headers = {"Authorization": f"{AUTH_HEADER_PREFIX}{ticket}"}
        total_size = int(meta.get("size") or 0)
        expect_hash = str(meta.get("sha256_hex") or "")
        part_path = out_path + ".part"
        try:
            with requests.get(url, headers=headers, stream=True, timeout=120) as resp:
                if resp.status_code != 200:
                    return f"{fid}: HTTP {resp.status_code}"
                h = hashlib.sha256()
                with open(part_path, "wb") as out_f:
                    raw_reader = resp.raw
                    magic = raw_reader.read(4)
                    if magic != FRAME_MAGIC:
                        raise ValueError("bad ciphertext stream")
                    aes = AESGCM(offer.key)
                    chunk_index = 0
                    while True:
                        hdr = raw_reader.read(4)
                        if not hdr:
                            break
                        if len(hdr) < 4:
                            raise ValueError("truncated stream")
                        (ln,) = struct.unpack(">I", hdr)
                        blob = raw_reader.read(ln)
                        if len(blob) != ln:
                            raise ValueError("truncated stream")
                        nonce, ct = blob[:12], blob[12:]
                        aad = _aad(transfer_id, fid, chunk_index, total_size)
                        pt = aes.decrypt(nonce, ct, aad)
                        out_f.write(pt)
                        h.update(pt)
                        chunk_index += 1
                digest = h.hexdigest()
                if digest.lower() != expect_hash.lower():
                    try:
                        os.remove(part_path)
                    except OSError:
                        pass
                    return f"{fid}: sha256 mismatch"
                os.replace(part_path, out_path)
        except Exception as exc:
            try:
                if os.path.isfile(part_path):
                    os.remove(part_path)
            except OSError:
                pass
            return f"{fid}: {exc}"
        return None

    def accept_inbound_prepare(
        self,
        *,
        transfer_id: str,
        destination_dir: str,
        session_id: str | None = None,
    ) -> dict[str, Any]:
        """Validate destination, send mesh accept once, store planned paths for staged fetches."""
        if not self.is_enabled():
            return {"ok": False, "error": "file_transfer disabled"}
        with self._lock:
            offer, err = self._get_inbound_offer_unlocked(transfer_id, session_id)
            if err is not None:
                return err
            assert offer is not None
            offer.last_error = None
            dest = os.path.realpath(os.path.expanduser(destination_dir))
            if not os.path.isdir(dest):
                return {"ok": False, "error": "Destination is not a directory."}
            planned_or_err = self._build_inbound_planned_writes(offer, dest)
            if isinstance(planned_or_err, dict):
                return planned_or_err
            planned = planned_or_err
            if not planned:
                return {"ok": False, "error": "No files in this transfer."}
            for _fid, out_p, _meta in planned:
                parent = os.path.dirname(out_p)
                if parent:
                    try:
                        os.makedirs(parent, exist_ok=True)
                    except OSError as e:
                        return {"ok": False, "error": f"Could not create folder: {e}"}
            self._send_mesh_accept_if_needed(offer)
            offer.accept_dest = dest
            offer.accept_planned = planned
            offer.fetch_completed_ids.clear()
            total_bytes = sum(int(m.get("size") or 0) for _, _, m in planned)
            file_rows: list[dict[str, Any]] = []
            for fid, out_path, meta in planned:
                file_rows.append(
                    {
                        "id": fid,
                        "relpath": str(meta.get("relpath") or ""),
                        "size": int(meta.get("size") or 0),
                        "local_name": os.path.basename(out_path),
                    }
                )
            return {
                "ok": True,
                "destination": dest,
                "files": file_rows,
                "total_bytes": total_bytes,
                "total_files": len(planned),
            }

    def accept_inbound_fetch_one(
        self,
        *,
        transfer_id: str,
        file_id: str,
        session_id: str | None = None,
    ) -> dict[str, Any]:
        """Download a single file after :meth:`accept_inbound_prepare`."""
        if not self.is_enabled():
            return {"ok": False, "error": "file_transfer disabled"}
        fid = (file_id or "").strip()
        if not fid:
            return {"ok": False, "error": "Missing file_id."}
        with self._lock:
            offer, err = self._get_inbound_offer_unlocked(transfer_id, session_id)
            if err is not None:
                return err
            assert offer is not None
            planned = offer.accept_planned
            if not planned:
                return {"ok": False, "error": "Call accept prepare first."}
            if fid in offer.fetch_completed_ids:
                done = len(offer.fetch_completed_ids)
                total = len(planned)
                paths = [p for i, p, _ in planned if i in offer.fetch_completed_ids]
                return {
                    "ok": True,
                    "skipped": True,
                    "path": next((p for i, p, _ in planned if i == fid), ""),
                    "completed_count": done,
                    "total_count": total,
                    "done_all": done >= total,
                    "paths_written": paths,
                }
            trip: tuple[str, str, dict[str, Any]] | None = None
            for i, p, m in planned:
                if i == fid:
                    trip = (i, p, m)
                    break
            if trip is None:
                return {"ok": False, "error": "Unknown file_id for this transfer."}

        err_one = self._download_one_inbound_file(
            offer,
            transfer_id=transfer_id,
            fid=trip[0],
            out_path=trip[1],
            meta=trip[2],
        )
        with self._lock:
            if transfer_id not in self._inbound:
                return {"ok": False, "error": "Offer was removed."}
            offer2 = self._inbound[transfer_id]
            if err_one is not None:
                offer2.last_error = err_one
                written = [p for i, p, _ in (offer2.accept_planned or []) if i in offer2.fetch_completed_ids]
                return {
                    "ok": False,
                    "error": err_one,
                    "completed_count": len(offer2.fetch_completed_ids),
                    "total_count": len(offer2.accept_planned or []),
                    "paths_written": written,
                }
            if fid not in offer2.fetch_completed_ids:
                offer2.fetch_completed_ids.append(fid)
            done = len(offer2.fetch_completed_ids)
            total = len(offer2.accept_planned or [])
            paths = [p for i, p, _ in (offer2.accept_planned or []) if i in offer2.fetch_completed_ids]
            all_done = done >= total
            dest = offer2.accept_dest or ""
            if all_done:
                self._inbound.pop(transfer_id, None)
            return {
                "ok": True,
                "path": trip[1],
                "completed_count": done,
                "total_count": total,
                "done_all": all_done,
                "destination": dest,
                "paths_written": paths,
            }

    def accept_inbound_impl(
        self,
        *,
        transfer_id: str,
        destination_dir: str,
        session_id: str | None = None,
    ) -> dict[str, Any]:
        prep = self.accept_inbound_prepare(
            transfer_id=transfer_id,
            destination_dir=destination_dir,
            session_id=session_id,
        )
        if not prep.get("ok"):
            return prep
        files_manifest = prep.get("files") or []
        written: list[str] = []
        errors: list[str] = []
        for row in files_manifest:
            fid = str(row.get("id") or "")
            one = self.accept_inbound_fetch_one(transfer_id=transfer_id, file_id=fid, session_id=session_id)
            if not one.get("ok"):
                errors.append(str(one.get("error") or "fetch failed"))
                written = list(one.get("paths_written") or written)
                err_text = "; ".join(errors) if errors else None
                with self._lock:
                    if transfer_id in self._inbound:
                        self._inbound[transfer_id].last_error = err_text
                return {"ok": False, "error": err_text, "paths_written": written}
            written = list(one.get("paths_written") or written)
        return {"ok": True, "destination": prep.get("destination"), "paths_written": written}

    def clear_all(self) -> None:
        with self._lock:
            self._outbound.clear()
            self._inbound.clear()

    def on_mesh_control(self, session: MeshSession, msg: MeshMessage) -> None:
        payload = msg.payload or {}
        tid = str(payload.get("transfer_id") or "")
        if not tid:
            return
        if msg.type == "file_transfer_accept":
            append_audit(
                {
                    "dir": "in",
                    "source": "peer",
                    "peer": session.peer_fingerprint,
                    "peer_name": session.peer_hostname,
                    "type": "file_transfer_accept",
                    "conversation_id": msg.conversation_id,
                    "summary": f"accept {tid}",
                    "bytes": 0,
                }
            )
        elif msg.type == "file_transfer_decline":
            with self._lock:
                self._outbound.pop(tid, None)
        elif msg.type == "file_transfer_cancel":
            with self._lock:
                self._inbound.pop(tid, None)
