from __future__ import annotations

import json
import logging
import re
import threading

from langchain_core.tools import tool
from pydantic import BaseModel, Field

_log = logging.getLogger(__name__)

# Process-wide fallback when no per-thread catalog is bound (single-threaded harness).
_TOOL_CATALOG: list = []
_catalog_local = threading.local()


def set_tool_catalog(tools: list) -> None:
    """Set the tool catalog for the current thread's agent run."""
    global _TOOL_CATALOG
    dedup: dict[str, object] = {}
    for t in tools or []:
        name = getattr(t, "name", None) or ""
        if name and name not in dedup:
            dedup[name] = t
    catalog = list(dedup.values())
    _catalog_local.catalog = catalog
    # Worker threads must not clobber the process-wide harness fallback.
    if threading.current_thread() is threading.main_thread():
        _TOOL_CATALOG = catalog


def get_tool_catalog() -> list:
    """Return the catalog for this thread, else the process fallback."""
    cat = getattr(_catalog_local, "catalog", None)
    if cat is not None:
        return list(cat)
    return list(_TOOL_CATALOG)


def reset_tool_catalog_for_tests() -> None:
    """Clear thread-local and global catalogs (tests only)."""
    global _TOOL_CATALOG
    _TOOL_CATALOG = []
    if hasattr(_catalog_local, "catalog"):
        del _catalog_local.catalog


def _short_description(text: str, max_chars: int = 110) -> str:
    clean = re.sub(r"\s+", " ", (text or "").strip())
    if not clean:
        return "No description."
    sentence = re.split(r"(?<=[.!?])\s", clean, maxsplit=1)[0]
    if len(sentence) <= max_chars:
        return sentence
    return sentence[: max_chars - 1].rstrip() + "…"


def _tool_to_openai_schema(tool_obj) -> dict | None:
    """Local schema converter to avoid importing agent_loop (circular import)."""
    name = getattr(tool_obj, "name", None) or "unknown"
    description = getattr(tool_obj, "description", None) or ""
    params = {"type": "object", "properties": {}, "additionalProperties": False}
    schema = getattr(tool_obj, "args_schema", None)
    if schema is not None:
        try:
            if isinstance(schema, dict):
                raw = dict(schema)
            elif hasattr(schema, "model_json_schema"):
                raw = schema.model_json_schema()
            elif hasattr(schema, "schema"):
                raw = schema.schema()
            else:
                raw = {}
            if isinstance(raw, dict):
                params = raw.copy()
                params.pop("title", None)
                params.setdefault("type", "object")
                params.setdefault("properties", {})
        except Exception as exc:
            _log.warning("find_tools schema conversion failed for %s: %s", name, exc)
            return None
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": params,
        },
    }


def build_tool_index(
    tools: list | None = None,
    *,
    exclude_names: set[str] | frozenset[str] | None = None,
) -> str:
    """Return compact name + short-description index for discoverable tools.

    When ``exclude_names`` is set (tiered parent), omit tools that already
    have full schemas in the API ``tools`` list — avoids duplicating tier-0
    entries in the system prompt index.
    """
    src = list(tools) if tools is not None else get_tool_catalog()
    if not src:
        return ""
    skip = {str(n).strip() for n in (exclude_names or ()) if str(n or "").strip()}
    header = (
        "Specialized tools (default set already has full schemas; use find_tools for full schemas):"
        if skip
        else "Available tools (use find_tools to get full schemas):"
    )
    rows: list[str] = []
    for t in sorted(src, key=lambda x: (getattr(x, "name", "") or "")):
        name = getattr(t, "name", None) or ""
        if not name or (skip and name in skip):
            continue
        desc = _short_description(getattr(t, "description", None) or "")
        rows.append(f"- {name}: {desc}")
    if not rows:
        return ""
    return header + "\n" + "\n".join(rows)


# Stable first-line contract for find_tools results. The agent loop parses this
# to promote matches into the per-thread working set — keep prefix and helper
# in lockstep (tested in tests/test_tool_index.py).
_MATCHES_PREFIX = "Matches: "


def parse_find_tools_matches(result: str) -> list[str]:
    """Parse tool names from a find_tools result's stable 'Matches: a, b' first line."""
    text = str(result or "").strip()
    if not text:
        return []
    first = text.splitlines()[0]
    if not first.startswith(_MATCHES_PREFIX):
        return []
    return [n.strip() for n in first[len(_MATCHES_PREFIX):].split(",") if n.strip()]


def _working_set_promotion_active() -> bool:
    """True when the core diet / working-set promotion path is enabled in config."""
    try:
        from core.config import get_agent_loop_tool_context_config

        return bool(get_agent_loop_tool_context_config().get("core_toolset_enabled", True))
    except Exception:
        return False


class FindToolsInput(BaseModel):
    query: str = Field(description="Keywords for the task (e.g. 'ocr pdf', 'calendar reminders').")
    max_results: int = Field(default=5, ge=1, le=12, description="Maximum number of matching tools to return.")


@tool(args_schema=FindToolsInput)
def find_tools(query: str, max_results: int = 5) -> str:
    """Find relevant tools by keywords and return their full function schemas.

    Examples: `ocr`, `calendar`, `vault` / `pattern inbox` / `capture` for Mugi Vault notes, `email`, `remind`.
    For HTML compositions rendered to MP4/WebM locally, use `hyperframes` or `html video` (install via Settings > Software > Hyperframes).
    For interactive Playwright automation (forms, pagination), use `web script`, `playwright`, or `run_web_script` (requires Settings > Advanced > Allow web automation scripts).
    """
    q = (query or "").strip().lower()
    if not q:
        return "Provide a non-empty query."
    catalog = get_tool_catalog()
    if not catalog:
        return "Tool catalog is not initialized yet."

    terms = [t for t in re.split(r"[^a-z0-9_]+", q) if t]
    ranked: list[tuple[int, object]] = []
    for tool_obj in catalog:
        name = (getattr(tool_obj, "name", None) or "").lower()
        normalized_name = name
        if normalized_name.startswith("mcp__"):
            parts = normalized_name.split("__", 2)
            normalized_name = parts[2] if len(parts) > 2 else normalized_name
        desc = (getattr(tool_obj, "description", None) or "").lower()
        haystack = f"{name} {normalized_name} {desc}"
        score = 0
        for term in terms:
            if term == normalized_name or term == name:
                score += 8
            elif term in normalized_name or term in name:
                score += 4
            elif term in haystack:
                score += 1
        if score > 0:
            ranked.append((score, tool_obj))
    ranked.sort(key=lambda x: (-x[0], getattr(x[1], "name", "")))
    chosen = [tool_obj for _, tool_obj in ranked[: max(1, int(max_results))]]
    if not chosen:
        return "No matching tools found. Try broader keywords."

    payload: list[dict] = []
    included: list[object] = []
    for tool_obj in chosen:
        schema = _tool_to_openai_schema(tool_obj)
        if schema is None:
            continue
        payload.append(schema)
        included.append(tool_obj)
    if not payload:
        return "Matching tools found but schemas could not be built."
    names = ", ".join(getattr(t, "name", "unknown") for t in included)
    if _working_set_promotion_active():
        # Core diet active: the agent loop promotes these matches into the
        # native schema set before the next round, so dumping full JSON
        # schemas here would be paid twice. Names + one-line summaries are
        # enough for the model to pick; exact argument schemas arrive natively.
        lines = []
        for tool_obj, schema in zip(included, payload):
            fn = schema.get("function") or {}
            params = (fn.get("parameters") or {}).get("properties") or {}
            required = set((fn.get("parameters") or {}).get("required") or [])
            args = ", ".join(
                f"{k}{'' if k in required else '?'}" for k in params
            )
            lines.append(
                f"- {getattr(tool_obj, 'name', 'unknown')}({args}): "
                + _short_description(getattr(tool_obj, "description", "") or "")
            )
        return (
            f"{_MATCHES_PREFIX}{names}\n"
            "These tools are now loaded natively — call them directly as your next step "
            "(exact argument schemas are attached to the request).\n"
            + "\n".join(lines)
        )
    return (
        f"{_MATCHES_PREFIX}{names}\n"
        "Use the function.name exactly as shown below, and match argument keys exactly to each returned schema.\n\n"
        + json.dumps(payload, indent=2)
    )

