# tools/macos_spotlight_tools.py — Spotlight (`mdfind`) search for the Quick Panel launcher surface.
#
# macOS only. We call `/usr/bin/mdfind` as a subprocess so we can use the same
# metadata index Finder uses without writing a native Swift helper. The tool
# is intentionally conservative: strict argument validation, a subprocess
# timeout, a max result cap, and default path scopes that keep system
# directories out of the result set unless the caller explicitly opts in.
#
# macOS TCC (Full Disk Access, Files & Folders) may silently omit results in
# protected locations (Desktop, Documents, iCloud placeholders). We surface a
# permissions hint when zero results come back so users are not misled into
# thinking the file simply does not exist.
from __future__ import annotations

import os
import subprocess
import sys
import time
from threading import Lock

from langchain_core.tools import tool

from tools._logging import _debug_log, _emit_input_status


_MDFIND_PATH = "/usr/bin/mdfind"
_MDFIND_TIMEOUT_SECONDS = 8
_MAX_RESULTS = 50
_MAX_QUERY_LEN = 512

# Per-process rate limit: a small cooldown between full-disk spotlight calls
# prevents a runaway agent loop from pegging the CPU. This is not security —
# it is resource protection for the user's machine.
_SPOTLIGHT_COOLDOWN_SECONDS = 0.4
_last_call_lock = Lock()
_last_call_ts = 0.0


def _is_macos() -> bool:
    return sys.platform == "darwin"


def _default_scopes() -> list[str]:
    """Default search roots: the user's home tree. Keeps `/System` etc. out by default."""
    home = os.path.expanduser("~")
    candidates = [home]
    return [p for p in candidates if os.path.isdir(p)]


def _validate_scope(path: str) -> tuple[bool, str, str]:
    """Return (ok, expanded_path, error) for a user-provided scope path.

    We require an existing directory and forbid the root `/` (too broad, and
    almost always a mistake from a local model parroting examples).
    """
    if not path or not isinstance(path, str):
        return False, "", "scope path must be a non-empty string"
    expanded = os.path.expanduser(path.strip())
    if not expanded or expanded == "/":
        return False, "", "scope '/' is not allowed — name a directory (e.g. '~', '~/Documents')"
    if not os.path.isabs(expanded):
        expanded = os.path.abspath(expanded)
    if not os.path.isdir(expanded):
        return False, expanded, f"scope not found or not a directory: {expanded}"
    return True, expanded, ""


def _respect_cooldown() -> None:
    global _last_call_ts
    with _last_call_lock:
        now = time.monotonic()
        delta = now - _last_call_ts
        if delta < _SPOTLIGHT_COOLDOWN_SECONDS:
            sleep_for = _SPOTLIGHT_COOLDOWN_SECONDS - delta
            time.sleep(sleep_for)
        _last_call_ts = time.monotonic()


def _truncate(s: str, n: int) -> str:
    return s if len(s) <= n else s[: n - 1] + "…"


def _escape_mdfind_value(value: str) -> str:
    """Escape value-string metacharacters for Spotlight metadata expressions."""
    escaped = value.replace("\\", "\\\\")
    escaped = escaped.replace('"', '\\"')
    escaped = escaped.replace("'", "\\'")
    return escaped


def _format_hit(path: str) -> str:
    """One-line result row: basename + parent directory for context."""
    home = os.path.expanduser("~")
    display = path
    if display.startswith(home):
        display = "~" + display[len(home):]
    return display


def _run_mdfind(args: list[str]) -> tuple[bool, list[str], str]:
    try:
        proc = subprocess.run(
            [_MDFIND_PATH, *args],
            capture_output=True,
            text=True,
            timeout=_MDFIND_TIMEOUT_SECONDS,
            check=False,
        )
    except subprocess.TimeoutExpired:
        return False, [], f"Spotlight timed out after {_MDFIND_TIMEOUT_SECONDS}s."
    except FileNotFoundError:
        return False, [], f"Spotlight helper not found at {_MDFIND_PATH}."
    except Exception as exc:
        return False, [], f"Spotlight failed: {exc}"
    if proc.returncode != 0:
        err = (proc.stderr or "").strip() or f"exit {proc.returncode}"
        return False, [], f"Spotlight error: {err}"
    lines = [ln for ln in (proc.stdout or "").splitlines() if ln.strip()]
    return True, lines, ""


def _spotlight_search_impl(
    *,
    query: str,
    name_contains: str,
    within_days: int,
    scope: str,
    limit: int,
) -> str:
    if not _is_macos():
        return (
            "spotlight_search is macOS-only (uses /usr/bin/mdfind). "
            "On this platform, try search_filesystem instead."
        )

    q = (query or "").strip()
    name = (name_contains or "").strip()
    if not q and not name:
        return (
            "Provide a query (full-text search) or name_contains (filename match). "
            "Example: spotlight_search(query='cereal', within_days=7)."
        )
    if len(q) > _MAX_QUERY_LEN:
        return f"Query too long (>{_MAX_QUERY_LEN} chars). Shorten and retry."
    if len(name) > _MAX_QUERY_LEN:
        return f"name_contains too long (>{_MAX_QUERY_LEN} chars). Shorten and retry."

    try:
        limit = int(limit)
    except (TypeError, ValueError):
        limit = _MAX_RESULTS
    limit = max(1, min(_MAX_RESULTS, limit))

    try:
        within_days = int(within_days)
    except (TypeError, ValueError):
        within_days = 0
    within_days = max(0, within_days)

    # Build the mdfind query. Users pass natural-language fragments; we wrap
    # them into a safe metadata query rather than shelling out raw input.
    clauses: list[str] = []
    if q:
        # Case-insensitive, diacritic-insensitive full-text/metadata match.
        escaped_q = _escape_mdfind_value(q)
        clauses.append(
            f"kMDItemTextContent == \"*{escaped_q}*\"cd || "
            f"kMDItemDisplayName == \"*{escaped_q}*\"cd"
        )
    if name:
        escaped_name = _escape_mdfind_value(name)
        clauses.append(f"kMDItemDisplayName == \"*{escaped_name}*\"cd")
    if within_days > 0:
        # `$time.today(-N)` is a builtin mdfind token for "N days ago".
        clauses.append(f"kMDItemContentModificationDate >= $time.today(-{within_days})")
    expr = " && ".join(f"({c})" for c in clauses) if len(clauses) > 1 else clauses[0]

    scopes: list[str]
    if scope and scope.strip():
        ok, expanded, err = _validate_scope(scope)
        if not ok:
            return f"ERROR: {err}"
        scopes = [expanded]
    else:
        scopes = _default_scopes()

    _emit_input_status("Spotlight searching…")
    _debug_log(f"[spotlight] expr={expr!r} scopes={scopes} limit={limit}")
    _respect_cooldown()

    args: list[str] = []
    for s in scopes:
        args.extend(["-onlyin", s])
    args.append(expr)

    ok, lines, err = _run_mdfind(args)
    if not ok:
        return f"ERROR: {err}"

    if not lines:
        hint = (
            "Spotlight returned no matches. Possible causes:\n"
            "- The index may still be catching up for recently-modified files.\n"
            "- macOS may restrict results for Desktop, Documents, or iCloud Drive "
            "until Mugi is granted Full Disk Access (System Settings → Privacy & Security → Full Disk Access).\n"
            "- iCloud Drive placeholders are excluded until the file has been downloaded."
        )
        summary = f"No files matched query={q!r}"
        if name:
            summary += f", name_contains={name!r}"
        if within_days > 0:
            summary += f", within_days={within_days}"
        return f"{summary}.\n\n{hint}"

    rows = lines[:limit]
    header = f"Spotlight: {len(rows)} result(s)"
    if len(lines) > len(rows):
        header += f" (of {len(lines)}+ — limit {_MAX_RESULTS})"
    if scope:
        header += f" in {scope}"
    formatted = "\n".join(f"- {_format_hit(p)}" for p in rows)
    return f"{header}\n{formatted}"


@tool
def spotlight_search(
    query: str = "",
    name_contains: str = "",
    within_days: int = 0,
    scope: str = "",
    limit: int = 20,
) -> str:
    """Search local files with macOS Spotlight (mdfind).

    Use this for "find any file with <text>", "files I changed this week named …",
    or any OS-facing disk lookup from the Quick Panel. Do NOT use this to search
    Mugi chat history (use search_sessions) or workspace long-term memory
    (use search_memory).

    Args:
        query: Full-text or metadata fragment to match. Example: "cereal".
        name_contains: Optional filename substring (case/diacritic-insensitive).
        within_days: Optional recency filter. 0 = any age. 7 = changed in the last 7 days.
        scope: Optional directory to search (e.g. "~/Documents"). Defaults to the user's home tree; `/` is rejected.
        limit: Max rows to return (1–50).

    Returns a short, human-readable list of paths. On macOS only; on other
    platforms it explains the limitation and points to search_filesystem.
    """
    return _spotlight_search_impl(
        query=query,
        name_contains=name_contains,
        within_days=within_days,
        scope=scope,
        limit=limit,
    )
