# tools/session_search_tools.py — Search past conversations for cross-session recall.
# Uses the FTS5 index in session_search_db.py for fast full-text search,
# then summarizes matched conversations via subagent LLM calls.
from __future__ import annotations

import json
import os
import re
from collections import defaultdict
from dataclasses import dataclass
from typing import Literal

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from tools._logging import _debug_log, _emit_input_status
from utils.conversation_sidebar import thread_id_allowed_in_main_conversation_sidebar
from utils.recall_signals import MUGI_THREAD_PREFIX_RE, should_browse_sessions_query
from utils.transcript_roles import is_tool_result_carrier_message

SEARCH_SESSIONS_ERROR_PREFIX = "[search_sessions_error]"

_THREAD_UUID_FULL_RE = re.compile(
    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    re.IGNORECASE,
)
_UUID_IN_TEXT_RE = re.compile(
    r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b",
    re.IGNORECASE,
)

_BLOCKED_DIRECT_LOOKUP_PREFIXES = (
    "mesh",
    "group",
    "peer",
    "btw",
    "synthetic",
    "heartbeat",
    "background_heartbeat",
    "drive_action",
    "drive_discovery_",
    "skill_eval_",
)


@dataclass(frozen=True)
class RecallPreflightResult:
    status: str
    query: str
    text: str
    result_chars: int
    truncated: bool = False
    reason: str = ""
    thread_count: int = 0


def _normalize_thread_id(raw: str) -> str:
    """Strip and fix UUID-shaped thread ids when the model inserts accidental spaces."""
    s = (raw or "").strip()
    if not s:
        return ""
    compact = re.sub(r"\s+", "", s)
    if _THREAD_UUID_FULL_RE.fullmatch(compact):
        return compact.lower()
    return s


def _search_sessions_error(message: str) -> str:
    return f"{SEARCH_SESSIONS_ERROR_PREFIX} {message.strip()}"


def is_search_sessions_success(result: str) -> bool:
    """True only for real session hits, not instructional error strings."""
    text = (result or "").strip()
    if not text or text.startswith(SEARCH_SESSIONS_ERROR_PREFIX):
        return False
    lowered = text.lower()
    if lowered.startswith("session search is not available"):
        return False
    if lowered.startswith("search index is rebuilding"):
        return False
    if lowered.startswith("provide a non-empty search query"):
        return False
    if lowered.startswith("no conversations found"):
        return False
    if lowered.startswith("no stored conversation found"):
        return False
    if lowered.startswith("no past conversations found"):
        return False
    if lowered.startswith("no relevant past conversations found"):
        return False
    return text.startswith("Found ") and "conversation" in lowered


_DATE_BOUND_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")


def _chats_db_path() -> str:
    from persistence import MUGI_DIR

    return os.path.join(MUGI_DIR, "chats_db.json")


def _normalize_date_bound(raw: str | None) -> str:
    """Normalize since/until to YYYY-MM-DD or full ISO prefix for lexicographic compare."""
    s = (raw or "").strip()
    if not s:
        return ""
    if _DATE_BOUND_RE.fullmatch(s):
        return s
    if len(s) >= 10 and s[4] == "-" and s[7] == "-":
        return s[:10] if len(s) == 10 else s
    return ""


def _created_at_within_bounds(created_at: str, *, since: str = "", until: str = "") -> bool:
    ca = (created_at or "").strip()
    if not ca:
        return not (since or until)
    if since:
        key = ca[:10] if len(since) == 10 else ca
        if key < since:
            return False
    if until:
        key = ca[:10] if len(until) == 10 else ca
        if key > until:
            return False
    return True


def _load_browseable_conversations(*, include_archived: bool = False) -> list[dict[str, str]]:
    """Sidebar-visible conversations from chats_db.json (newest first)."""
    path = _chats_db_path()
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError, OSError) as exc:
        _debug_log(f"[search_sessions] chats_db read failed: {exc}", "TOOL")
        return []
    if not isinstance(data, dict):
        return []
    meta = data.get("conversation_meta")
    if not isinstance(meta, dict):
        return []

    branch_ids: set[str] = set()
    try:
        from chat_storage import is_branch

        for tid in meta:
            if isinstance(tid, str) and is_branch(tid):
                branch_ids.add(tid)
    except Exception:
        pass

    convos: list[dict[str, str]] = []
    for tid, m in meta.items():
        if not isinstance(tid, str) or tid in branch_ids:
            continue
        if not thread_id_allowed_in_main_conversation_sidebar(tid):
            continue
        m2 = dict(m) if isinstance(m, dict) else {}
        if bool(m2.get("trashed")):
            continue
        if not include_archived and bool(m2.get("archived")):
            continue
        convos.append(
            {
                "thread_id": tid,
                "title": str(m2.get("title") or "Untitled"),
                "created_at": str(m2.get("created_at") or ""),
            }
        )
    convos.sort(key=lambda c: c.get("created_at", ""), reverse=True)
    return convos


def _first_user_preview(thread_id: str, *, max_chars: int = 120) -> str:
    try:
        from chat_storage import get_messages

        for m in get_messages(thread_id) or []:
            if (m.get("role") or "").strip() != "user":
                continue
            content = m.get("content", "")
            if isinstance(content, list):
                parts = [
                    p.get("text", "")
                    for p in content
                    if isinstance(p, dict) and p.get("type") == "text"
                ]
                content = " ".join(parts)
            if not isinstance(content, str) or not content.strip():
                continue
            text = " ".join(content.split())
            if len(text) > max_chars:
                return text[: max_chars - 3].rstrip() + "..."
            return text
    except Exception:
        pass
    return ""


def _format_browse_sessions_result(
    rows: list[dict[str, str]],
    *,
    max_sessions: int,
    since: str = "",
    until: str = "",
) -> str:
    shown = rows[:max_sessions]
    if not shown:
        if since or until:
            return _search_sessions_error(
                "No conversations found in history for the requested date range."
            )
        return _search_sessions_error("No conversations found in history.")

    range_note = ""
    if since and until:
        range_note = f" ({since} to {until})"
    elif since:
        range_note = f" (since {since})"
    elif until:
        range_note = f" (until {until})"

    blocks: list[str] = []
    for i, row in enumerate(shown, 1):
        tid = row["thread_id"]
        title = (row.get("title") or "Untitled").replace('"', "'")
        created = row.get("created_at") or ""
        date_str = created[:10] if len(created) >= 10 else (created or "unknown date")
        preview = _first_user_preview(tid)
        line = f'{i}. [{date_str}] "{title}"'
        if preview:
            line += f" — {preview}"
        line += f"\n   Open: target_thread_id={tid}"
        blocks.append(line)

    header = f"Found {len(blocks)} recent conversation(s) (newest first){range_note}:"
    return header + "\n\n" + "\n\n".join(blocks)


def _search_sessions_browse(
    *,
    max_sessions: int,
    exclude_thread_id: str = "",
    since: str | None = None,
    until: str | None = None,
) -> str:
    since_n = _normalize_date_bound(since)
    until_n = _normalize_date_bound(until)
    excl = (exclude_thread_id or "").strip().lower()

    rows: list[dict[str, str]] = []
    for row in _load_browseable_conversations():
        tid = (row.get("thread_id") or "").strip()
        if not tid or tid.lower() == excl:
            continue
        if not _created_at_within_bounds(row.get("created_at") or "", since=since_n, until=until_n):
            continue
        rows.append(row)

    return _format_browse_sessions_result(
        rows,
        max_sessions=max(1, min(int(max_sessions or 3), 5)),
        since=since_n,
        until=until_n,
    )


def _direct_thread_lookup_allowed(tid: str, *, include_background_threads: bool = False) -> bool:
    t = (tid or "").strip().lower()
    if not t or t in ("default", "__quick__"):
        return False
    if any(t.startswith(p) for p in _BLOCKED_DIRECT_LOOKUP_PREFIXES):
        return False
    if not include_background_threads:
        try:
            from session_search_db import _thread_id_is_background

            if _thread_id_is_background(t):
                return False
        except Exception:
            pass
    return True


def _oldest_foreground_thread_id(*, exclude_thread_id: str = "") -> str | None:
    """Return the earliest indexed foreground thread id, excluding synthetic/background ids."""
    try:
        from session_search_db import _connection
    except Exception:
        return None
    conn = _connection()
    if conn is None:
        return None
    excl = (exclude_thread_id or "").strip().lower()
    try:
        rows = conn.execute(
            """
            SELECT thread_id, MIN(created_at) AS first_ts
            FROM messages_fts
            GROUP BY thread_id
            ORDER BY first_ts ASC
            LIMIT 80
            """
        ).fetchall()
    except Exception:
        return None
    finally:
        try:
            conn.close()
        except Exception:
            pass
    for tid, _first_ts in rows:
        t = str(tid or "").strip().lower()
        if not t or t == excl:
            continue
        if _direct_thread_lookup_allowed(t):
            return t
    return None


def _redact_thread_ids(text: str) -> str:
    return _UUID_IN_TEXT_RE.sub("[thread id redacted]", text or "")


def _format_recall_excerpt_for_injection(date_str: str, excerpt: str, *, header: str) -> str:
    date_label = date_str[:10] if len(str(date_str or "")) >= 10 else (date_str or "unknown date")
    body = _redact_thread_ids(excerpt)
    return f"{header}\n\n[prior conversation, {date_label}]\n{body}"


def _split_mugi_thread_prefix(raw_query: str) -> tuple[str, str]:
    """Strip `mugi-thread:<uuid>` bridge prefixes; return (cleaned_query, thread_id_or_empty)."""
    q = (raw_query or "").strip()
    if not q:
        return "", ""
    m = MUGI_THREAD_PREFIX_RE.search(q)
    if not m:
        return q, ""
    tid = _normalize_thread_id(str(m.group(1)))
    cleaned = (q[: m.start()] + q[m.end() :]).strip()
    return cleaned, tid


def _resolve_effective_thread_id(
    query: str,
    target_thread_id: str | None,
) -> tuple[str, str]:
    """Return (search_query, effective_tid) including bare UUID + hint extraction."""
    from utils.recall_signals import extract_direct_thread_id

    cleaned_query, auto_tid = _split_mugi_thread_prefix(query or "")
    explicit_tid = _normalize_thread_id(target_thread_id or "")
    effective_tid = explicit_tid or auto_tid
    if not effective_tid:
        extracted = extract_direct_thread_id(query or "")
        if extracted:
            effective_tid = _normalize_thread_id(extracted)
    return cleaned_query.strip(), effective_tid


def _cap_text(text: str, max_chars: int) -> tuple[str, bool]:
    cap = max(256, int(max_chars or 0))
    s = (text or "").strip()
    if len(s) <= cap:
        return s, False
    return s[:cap].rstrip() + "\n[...truncated]", True


def _message_lines_from_thread(thread_id: str) -> tuple[str, list[str]]:
    """Load formatted message lines from chat_storage."""
    from chat_storage import get_messages

    msgs = get_messages(thread_id)
    if not msgs:
        return "unknown date", []
    first_date = "unknown date"
    parts: list[str] = []
    for m in msgs:
        if (m.get("role") or "").strip() == "tool" or is_tool_result_carrier_message(m):
            continue
        role = (m.get("role") or "?").upper()
        content = m.get("content", "")
        if isinstance(content, list):
            text_parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"]
            content = " ".join(text_parts)
        if not content or not isinstance(content, str):
            continue
        if first_date == "unknown date":
            ts = (m.get("timestamp") or "").strip()
            if ts:
                first_date = ts
        if len(content) > 500:
            content = content[:500] + "..."
        parts.append(f"{role}: {content}")
    return first_date, parts


def _effective_semantic_search_flags(
    semantic: bool | None,
    hybrid: bool | None,
) -> tuple[bool | None, bool | None]:
    """Apply memory policy + retrieval tier to semantic/hybrid search toggles."""
    semantic_effective = semantic
    hybrid_effective = hybrid
    try:
        from agent_loop_config import get_memory_policy_prefs
        from utils.memory_retrieval_tier import MemoryRetrievalTier, get_current_retrieval_tier

        prefs = get_memory_policy_prefs() or {}
        if semantic_effective is None:
            semantic_effective = bool(prefs.get("memory_semantic_search_enabled", True))
        snap = get_current_retrieval_tier()
        if snap is not None and snap.tier == MemoryRetrievalTier.LOCAL:
            semantic_effective = False
            hybrid_effective = False
    except Exception:
        pass
    return semantic_effective, hybrid_effective


def _search_within_thread(
    thread_id: str,
    query: str,
    *,
    max_results: int = 8,
    semantic: bool | None = None,
    hybrid: bool | None = None,
    exclude_background_threads: bool | None = None,
) -> list[dict]:
    """FTS/semantic search scoped to a single thread."""
    from session_search_db import search

    q = (query or "").strip()
    if not q or not (thread_id or "").strip():
        return []

    def _run(*, fts_mode: str) -> list:
        return search(
            q,
            include_thread_id=thread_id,
            max_results=max_results,
            exclude_background_threads=exclude_background_threads,
            semantic=semantic,
            hybrid=hybrid,
            fts_mode=fts_mode,
        )

    results = _run(fts_mode="and")
    if not results and len(q.split()) >= 2:
        results = _run(fts_mode="and_content")
    if not results:
        results = _run(fts_mode="or")
    return results


def _format_in_thread_search_body(
    thread_id: str,
    hits: list[dict],
    *,
    max_chars: int,
) -> tuple[str, str]:
    """Build (date_str, excerpt body) from in-thread FTS hits."""
    date_str = "unknown date"
    if hits:
        ca = str(hits[0].get("created_at") or "")
        if ca:
            date_str = ca[:10] if len(ca) >= 10 else ca
    if date_str == "unknown date":
        date_str, _ = _message_lines_from_thread(thread_id)

    blocks: list[str] = []
    seen: set[str] = set()
    for hit in hits:
        snippet = (hit.get("snippet") or "").strip()
        if not snippet or snippet in seen:
            continue
        seen.add(snippet)
        role = (hit.get("role") or "").strip().upper()
        blocks.append(f"{role}: {snippet}" if role else snippet)

    body = "\n\n".join(blocks)
    cap = max(512, int(max_chars or 3500))
    if len(body) > cap:
        body = body[:cap].rstrip() + "\n[...truncated]"
    return date_str, body


def _format_direct_thread_lookup(
    thread_id: str,
    *,
    search_query: str = "",
    max_chars: int = 3500,
    include_background_threads: bool = False,
    semantic: bool | None = None,
    hybrid: bool | None = None,
) -> tuple[str, str] | None:
    """Return (formatted_text, reason) for a direct thread open, or None if unavailable.

    When *search_query* is non-empty, search within the thread first (FTS snippets from
    the matching region). Falls back to head+tail excerpt when there are no hits.
    """
    if not _direct_thread_lookup_allowed(thread_id, include_background_threads=include_background_threads):
        return None

    max_chars_i = max(512, min(int(max_chars or 3500), 8000))
    sq = (search_query or "").strip()
    exclude_bg: bool | None = False if include_background_threads else None

    if sq:
        hits = _search_within_thread(
            thread_id,
            sq,
            max_results=8,
            semantic=semantic,
            hybrid=hybrid,
            exclude_background_threads=exclude_bg,
        )
        if hits:
            date_str, body = _format_in_thread_search_body(thread_id, hits, max_chars=max_chars_i)
            if body:
                formatted = _format_recall_excerpt_for_injection(
                    date_str,
                    body,
                    header=f"Found {len(hits)} matching passage(s) in conversation:",
                )
                return formatted, "direct_thread_search"

    date_str, excerpt = _thread_excerpt_head_tail(thread_id, max_chars=max_chars_i)
    if not excerpt:
        return None
    if sq:
        excerpt = (
            f"No keyword matches for '{sq}' in this thread; showing conversation start/end:\n\n"
            + excerpt
        )
    formatted = _format_recall_excerpt_for_injection(
        date_str,
        excerpt,
        header="Found 1 conversation (direct thread lookup):",
    )
    return formatted, "direct_thread"


def _thread_excerpt_head_tail(thread_id: str, *, max_chars: int = 3000) -> tuple[str, str]:
    """Load excerpt using head+tail strategy for long conversations."""
    first_date, lines = _message_lines_from_thread(thread_id)
    if not lines:
        return "unknown date", ""
    full = "\n".join(lines)
    cap = max(512, int(max_chars or 3000))
    if len(full) <= cap:
        return first_date, full
    omit_marker = "\n[… middle omitted …]\n"
    tail_budget = int(cap * 0.45)
    head_budget = cap - tail_budget - len(omit_marker)
    if head_budget < 128:
        head_budget = max(128, cap // 2 - len(omit_marker) // 2)
        tail_budget = cap - head_budget - len(omit_marker)
    head_lines: list[str] = []
    head_len = 0
    for line in lines:
        add = len(line) + (1 if head_lines else 0)
        if head_len + add > head_budget:
            break
        head_lines.append(line)
        head_len += add
    head_end_idx = len(head_lines)
    remaining = lines[head_end_idx:]
    if not remaining:
        return first_date, "\n".join(head_lines)
    tail_lines: list[str] = []
    tail_len = 0
    for line in reversed(remaining):
        add = len(line) + (1 if tail_lines else 0)
        if tail_len + add > tail_budget:
            break
        tail_lines.insert(0, line)
        tail_len += add
    excerpt = "\n".join(head_lines) + omit_marker + "\n".join(tail_lines)
    return first_date, excerpt


def _thread_excerpt_for_summary(thread_id: str, *, max_chars: int = 3000) -> tuple[str, str]:
    """Load plaintext excerpt from chat_storage; returns (date_prefix, excerpt)."""
    return _thread_excerpt_head_tail(thread_id, max_chars=max_chars)

def recall_preflight_search(
    query: str,
    *,
    max_sessions: int = 3,
    max_chars: int = 3500,
    include_background_threads: bool = False,
    semantic: bool | None = None,
    hybrid: bool | None = None,
    target_thread_id: str | None = None,
    current_thread_id: str | None = None,
    browse_recent: bool = False,
    since: str | None = None,
    until: str | None = None,
) -> RecallPreflightResult:
    """Bounded, non-LLM recall lookup for explicit continuity turns.

    This is deliberately cheaper than the public ``search_sessions`` tool: it
    uses the same FTS/semantic search backend and privacy filters, but returns
    capped excerpts/snippets without asking the main model to summarize.
    """
    try:
        max_sessions_i = max(1, min(int(max_sessions or 3), 5))
    except (TypeError, ValueError):
        max_sessions_i = 3
    max_chars_i = max(512, min(int(max_chars or 3500), 8000))

    search_query, effective_tid = _resolve_effective_thread_id(query or "", target_thread_id)

    try:
        from session_search_db import search, is_rebuilding, ensure_index_exists
    except Exception as exc:
        return RecallPreflightResult(
            status="error",
            query=search_query,
            text="",
            result_chars=0,
            reason=f"session_search_unavailable:{type(exc).__name__}",
        )

    if effective_tid:
        if not _direct_thread_lookup_allowed(effective_tid, include_background_threads=include_background_threads):
            return RecallPreflightResult(
                status="empty",
                query=search_query,
                text="",
                result_chars=0,
                reason="background_thread_excluded",
            )
        semantic_effective, hybrid_effective = _effective_semantic_search_flags(semantic, hybrid)
        if (search_query or "").strip():
            try:
                ensure_index_exists()
                if is_rebuilding():
                    return RecallPreflightResult(
                        status="error",
                        query=search_query,
                        text="",
                        result_chars=0,
                        reason="index_rebuilding",
                    )
            except Exception as exc:
                return RecallPreflightResult(
                    status="error",
                    query=search_query,
                    text="",
                    result_chars=0,
                    reason=f"index_unavailable:{type(exc).__name__}",
                )
        lookup = _format_direct_thread_lookup(
            effective_tid,
            search_query=search_query,
            max_chars=max_chars_i,
            include_background_threads=include_background_threads,
            semantic=semantic_effective,
            hybrid=hybrid_effective,
        )
        if lookup is None:
            return RecallPreflightResult(
                status="empty",
                query=search_query,
                text="",
                result_chars=0,
                reason="direct_thread_not_found",
            )
        formatted, lookup_reason = lookup
        capped, truncated = _cap_text(formatted, max_chars_i)
        return RecallPreflightResult(
            status="ran",
            query=search_query,
            text=capped,
            result_chars=len(capped),
            truncated=truncated,
            reason=lookup_reason,
            thread_count=1,
        )

    current_thread = (current_thread_id or "").strip()
    if not current_thread:
        try:
            from persistence import get_current_agent_thread_id

            current_thread = get_current_agent_thread_id() or ""
        except Exception:
            current_thread = ""

    use_browse = bool(browse_recent) or (
        bool(search_query) and should_browse_sessions_query(search_query)
    )
    if use_browse:
        browse_text = _search_sessions_browse(
            max_sessions=max_sessions_i,
            exclude_thread_id=current_thread,
            since=since,
            until=until,
        )
        if not browse_text.startswith(SEARCH_SESSIONS_ERROR_PREFIX):
            capped, truncated = _cap_text(browse_text, max_chars_i)
            count_match = re.search(r"Found (\d+) recent conversation", capped)
            thread_count = int(count_match.group(1)) if count_match else max_sessions_i
            return RecallPreflightResult(
                status="ran",
                query=search_query,
                text=capped,
                result_chars=len(capped),
                truncated=truncated,
                reason="browse_recent",
                thread_count=thread_count,
            )
        if bool(browse_recent) or not search_query:
            return RecallPreflightResult(
                status="empty",
                query=search_query,
                text="",
                result_chars=0,
                reason="browse_empty",
            )

    if not search_query:
        return RecallPreflightResult(status="skipped", query="", text="", result_chars=0, reason="empty_query")

    try:
        ensure_index_exists()
        if is_rebuilding():
            return RecallPreflightResult(
                status="error",
                query=search_query,
                text="",
                result_chars=0,
                reason="index_rebuilding",
            )
    except Exception as exc:
        return RecallPreflightResult(
            status="error",
            query=search_query,
            text="",
            result_chars=0,
            reason=f"index_unavailable:{type(exc).__name__}",
        )

    exclude_bg: bool | None = None
    if include_background_threads:
        exclude_bg = False

    semantic_effective = semantic
    hybrid_effective = hybrid
    try:
        from agent_loop_config import get_memory_policy_prefs
        from utils.memory_retrieval_tier import (
            MemoryRetrievalTier,
            apply_search_sessions_tier_cap,
            get_current_retrieval_tier,
        )

        prefs = get_memory_policy_prefs() or {}
        if semantic_effective is None:
            semantic_effective = bool(prefs.get("memory_semantic_search_enabled", True))
        snap = get_current_retrieval_tier()
        if snap is not None:
            max_sessions_i = int(apply_search_sessions_tier_cap(
                {"max_sessions": max_sessions_i},
                snap.max_sessions_cap,
            ).get("max_sessions") or max_sessions_i)
            if snap.tier == MemoryRetrievalTier.LOCAL:
                semantic_effective = False
                hybrid_effective = False
    except Exception:
        pass

    def _run_search(q: str, *, fts_mode: str) -> list:
        return search(
            q,
            exclude_thread_id=current_thread,
            max_results=max_sessions_i * 5,
            exclude_background_threads=exclude_bg,
            semantic=semantic_effective,
            hybrid=hybrid_effective,
            fts_mode=fts_mode,
        )

    try:
        results = _run_search(search_query, fts_mode="and")
        if not results and len(search_query.split()) >= 2:
            results = _run_search(search_query, fts_mode="and_content")
        if not results:
            results = _run_search(search_query, fts_mode="or")
    except Exception as exc:
        return RecallPreflightResult(
            status="error",
            query=search_query,
            text="",
            result_chars=0,
            reason=f"search_failed:{type(exc).__name__}",
        )

    if not results:
        try:
            from utils.recall_signals import continuity_anchor_recall_query

            if continuity_anchor_recall_query(search_query):
                browse_text = _search_sessions_browse(
                    max_sessions=max_sessions_i,
                    exclude_thread_id=current_thread,
                    since=since,
                    until=until,
                )
                if not browse_text.startswith(SEARCH_SESSIONS_ERROR_PREFIX):
                    capped, truncated = _cap_text(browse_text, max_chars_i)
                    count_match = re.search(r"Found (\d+) recent conversation", capped)
                    thread_count = int(count_match.group(1)) if count_match else max_sessions_i
                    return RecallPreflightResult(
                        status="ran",
                        query=search_query,
                        text=capped,
                        result_chars=len(capped),
                        truncated=truncated,
                        reason="browse_after_empty_fts",
                        thread_count=thread_count,
                    )
                global_oldest = _oldest_foreground_thread_id(exclude_thread_id="")
                ct = current_thread.strip().lower()
                oldest_tid = _oldest_foreground_thread_id(exclude_thread_id=current_thread)
                if global_oldest and ct == global_oldest:
                    oldest_tid = global_oldest
                elif not oldest_tid:
                    oldest_tid = global_oldest
                if oldest_tid:
                    date_str, excerpt = _thread_excerpt_head_tail(oldest_tid, max_chars=max_chars_i)
                    if excerpt:
                        formatted = _format_recall_excerpt_for_injection(
                            date_str,
                            excerpt,
                            header="Found earliest indexed conversation (continuity fallback):",
                        )
                        capped, truncated = _cap_text(formatted, max_chars_i)
                        return RecallPreflightResult(
                            status="ran",
                            query=search_query,
                            text=capped,
                            result_chars=len(capped),
                            truncated=truncated,
                            reason="continuity_oldest_thread",
                            thread_count=1,
                        )
        except Exception as exc:
            _debug_log(f"[recall_preflight] continuity fallback failed: {exc}", "RECALL")
        return RecallPreflightResult(status="empty", query=search_query, text="", result_chars=0, reason="no_results")

    threads: dict[str, list] = defaultdict(list)
    for r in results:
        tid = str(r.get("thread_id") or "")
        if not tid:
            continue
        if len(threads) >= max_sessions_i and tid not in threads:
            continue
        threads[tid].append(r)

    blocks: list[str] = []
    for tid, hits in list(threads.items())[:max_sessions_i]:
        first_date = str(hits[0].get("created_at") or "unknown date")
        date_str = first_date[:10] if len(first_date) >= 10 else first_date
        try:
            date_str2, excerpt = _thread_excerpt_for_summary(tid, max_chars=1100)
            if date_str2 != "unknown date":
                date_str = date_str2
        except Exception:
            excerpt = ""
        if not excerpt:
            snippets = [str(h.get("snippet") or "").strip() for h in hits[:3]]
            excerpt = "\n".join(s for s in snippets if s)
        if excerpt:
            date_label = date_str[:10] if len(date_str) >= 10 else date_str
            blocks.append(f"[prior conversation, {date_label}]\n{_redact_thread_ids(excerpt)}")

    if not blocks:
        return RecallPreflightResult(status="empty", query=search_query, text="", result_chars=0, reason="empty_excerpts")

    text, truncated = _cap_text(
        f"Found {len(blocks)} relevant past conversation(s) for recall:\n\n" + "\n\n---\n\n".join(blocks),
        max_chars_i,
    )
    return RecallPreflightResult(
        status="ran",
        query=search_query,
        text=text,
        result_chars=len(text),
        truncated=truncated,
        reason="ok",
        thread_count=len(blocks),
    )


class SearchSessionsInput(BaseModel):
    query: str = Field(
        default="",
        description="Search query — keywords, phrases, or FTS5 syntax (OR, NOT, prefix*). "
        "Leave empty with browse_recent=true when the user points at a past chat without a topic.",
    )
    max_sessions: int = Field(default=3, ge=1, le=5, description="Maximum number of past conversations to return (1-5).")
    include_background_threads: bool = Field(
        default=False,
        description="If True, include internal threads (heartbeat, drive discovery). Default False uses config [session_search].exclude_background_threads.",
    )
    semantic: bool | None = Field(
        default=None,
        description="Enable semantic vector search. None uses [session_search].semantic_search_enabled and tier policy.",
    )
    hybrid: bool | None = Field(
        default=None,
        description="When semantic search is active, merge lexical + semantic rankings. None uses [session_search].hybrid_search_enabled.",
    )
    target_thread_id: str | None = Field(
        default=None,
        description="Optional chat thread UUID to load directly (bypasses keyword search). "
        "When combined with query, searches within that thread for matching passages instead of "
        "only returning the conversation start/end. "
        "Also accepted via alias keys thread_id or conversation_id, or mugi-thread: prefix in query.",
    )
    order_by: Literal["relevance", "created_at"] = Field(
        default="relevance",
        description="Keyword hits: relevance (default) or created_at (newest matching thread first). Browse mode always uses created_at.",
    )
    browse_recent: bool = Field(
        default=False,
        description="List recent conversations by date (title + preview) instead of keyword search. "
        "Use when the user says 'that conversation', 'go back', or gives a date without a topic.",
    )
    since: str | None = Field(
        default=None,
        description="Optional start date (YYYY-MM-DD or ISO) when browsing or filtering by date.",
    )
    until: str | None = Field(
        default=None,
        description="Optional end date (YYYY-MM-DD or ISO) when browsing or filtering by date.",
    )


@tool(args_schema=SearchSessionsInput)
def search_sessions(
    query: str = "",
    max_sessions: int = 3,
    include_background_threads: bool = False,
    semantic: bool | None = None,
    hybrid: bool | None = None,
    target_thread_id: str | None = None,
    order_by: Literal["relevance", "created_at"] = "relevance",
    browse_recent: bool = False,
    since: str | None = None,
    until: str | None = None,
) -> str:
    """Search past conversations for relevant context. Use when the user references
    something from a previous conversation, says 'remember when...', or when you need
    to recall details from earlier work. For vague 'that conversation' cues without
    a topic, set browse_recent=true (optionally since/until). Returns excerpts or a
    date-sorted list with target_thread_id to open a thread."""
    _debug_log(
        f"[TOOL] search_sessions called: query_len={len((query or '').strip())}, max_sessions={max_sessions}",
        "TOOL",
    )
    _emit_input_status("Searching past conversations…")

    try:
        from session_search_db import search, is_rebuilding, ensure_index_exists
    except ImportError:
        _emit_input_status("")
        return _search_sessions_error("Session search is not available (session_search_db module not found).")

    search_query, effective_tid = _resolve_effective_thread_id(query or "", target_thread_id)

    if effective_tid:
        if not _direct_thread_lookup_allowed(effective_tid, include_background_threads=include_background_threads):
            _emit_input_status("")
            return _search_sessions_error(
                "Direct thread lookup is not allowed for internal or background threads."
            )
        semantic_effective, hybrid_effective = _effective_semantic_search_flags(semantic, hybrid)
        if (search_query or "").strip():
            ensure_index_exists()
            if is_rebuilding():
                _emit_input_status("")
                return _search_sessions_error(
                    "Search index is rebuilding from chat history. Try again in a moment."
                )
        lookup = _format_direct_thread_lookup(
            effective_tid,
            search_query=search_query,
            max_chars=3500,
            include_background_threads=include_background_threads,
            semantic=semantic_effective,
            hybrid=hybrid_effective,
        )
        if lookup is None:
            _emit_input_status("")
            return _search_sessions_error(
                "No stored conversation found for that thread id. "
                "It may be unsaved, from another profile, or the id may be wrong."
            )
        formatted, _reason = lookup
        _emit_input_status("")
        return formatted

    current_thread = ""
    try:
        from persistence import get_current_agent_thread_id

        current_thread = get_current_agent_thread_id() or ""
    except Exception:
        pass

    use_browse = bool(browse_recent) or (
        bool(search_query) and should_browse_sessions_query(search_query)
    )
    if use_browse:
        _emit_input_status("Listing recent conversations…")
        browse_out = _search_sessions_browse(
            max_sessions=max_sessions,
            exclude_thread_id=current_thread,
            since=since,
            until=until,
        )
        _emit_input_status("")
        if not browse_out.startswith(SEARCH_SESSIONS_ERROR_PREFIX):
            return browse_out
        if browse_recent or not search_query:
            return browse_out

    ensure_index_exists()

    if is_rebuilding():
        _emit_input_status("")
        return _search_sessions_error("Search index is rebuilding from chat history. Try again in a moment.")

    if not search_query:
        _emit_input_status("")
        return _search_sessions_error(
            "Provide a non-empty search query, pass target_thread_id, or set browse_recent=true "
            "to list recent conversations by date."
        )

    # Get current thread to exclude it from results (may refresh if empty above)
    if not current_thread:
        try:
            from persistence import get_current_agent_thread_id

            current_thread = get_current_agent_thread_id() or ""
        except Exception:
            pass

    exclude_bg: bool | None = None
    if include_background_threads:
        exclude_bg = False
    semantic_effective, hybrid_effective = _effective_semantic_search_flags(semantic, hybrid)

    def _run_search(q: str, *, fts_mode: str) -> list:
        return search(
            q,
            exclude_thread_id=current_thread,
            max_results=max_sessions * 5,
            exclude_background_threads=exclude_bg,
            semantic=semantic_effective,
            hybrid=hybrid_effective,
            fts_mode=fts_mode,
        )

    results = _run_search(search_query, fts_mode="and")
    if not results and len(search_query.split()) >= 2:
        results = _run_search(search_query, fts_mode="and_content")
    if not results:
        results = _run_search(search_query, fts_mode="or")

    if not results:
        if should_browse_sessions_query(search_query):
            _emit_input_status("Listing recent conversations…")
            browse_out = _search_sessions_browse(
                max_sessions=max_sessions,
                exclude_thread_id=current_thread,
                since=since,
                until=until,
            )
            _emit_input_status("")
            if not browse_out.startswith(SEARCH_SESSIONS_ERROR_PREFIX):
                return browse_out
        _emit_input_status("")
        return _search_sessions_error(f"No past conversations found matching '{search_query}'.")

    # Group by thread_id, take top N unique threads
    threads: dict[str, list] = defaultdict(list)
    for r in results:
        tid = r["thread_id"]
        if len(threads) >= max_sessions and tid not in threads:
            continue
        threads[tid].append(r)

    thread_items = list(threads.items())
    if (order_by or "relevance").strip().lower() == "created_at":
        def _thread_sort_key(item: tuple[str, list]) -> str:
            hits = item[1]
            dates = [str(h.get("created_at") or "") for h in hits]
            return max(dates) if dates else ""

        thread_items.sort(key=_thread_sort_key, reverse=True)

    # Build summaries for each matched thread
    summaries: list[str] = []
    max_summaries = max_sessions
    if len(results) > max_sessions * 3:
        # Avoid stacking utility summarization cost on very large result sets.
        max_summaries = max(1, min(max_sessions, 2))

    for tid, hits in thread_items[:max_summaries]:
        snippets = [h["snippet"] for h in hits[:3]]
        first_date = hits[0].get("created_at", "unknown date")
        date_str = first_date[:10] if len(first_date) >= 10 else first_date

        # Load full conversation for richer context (truncated)
        full_context = ""
        try:
            date_str2, excerpt = _thread_excerpt_for_summary(tid)
            full_context = excerpt
            if date_str2 != "unknown date":
                date_str = date_str2
        except Exception:
            full_context = "\n".join(snippets)

        # Summarize using the LLM if we have enough context
        summary = _summarize_thread(search_query, full_context, date_str)
        summaries.append(summary)

    _emit_input_status("")
    if not summaries:
        return _search_sessions_error(f"No relevant past conversations found for '{search_query}'.")

    header = f"Found {len(summaries)} relevant past conversation(s):"
    return header + "\n\n" + "\n\n---\n\n".join(summaries)


def _summarize_thread(query: str, context: str, date_str: str) -> str:
    """Summarize a conversation excerpt relevant to the query.

    Uses a lightweight LLM call if available; falls back to returning raw snippets.
    """
    if len(context) < 100:
        return f"[{date_str}] {context}"

    prompt_messages = [
        {"role": "system", "content": (
            "You are a conversation summarizer. Produce a concise 2-4 sentence summary "
            "of the conversation excerpt below, focusing on details relevant to the query. "
            "Include specific facts, file paths, commands, or decisions. "
            "Do NOT include preamble or meta-commentary. Start with the summary directly."
        )},
        {"role": "user", "content": (
            f"Query: {query}\n\n"
            f"Conversation excerpt (from {date_str}):\n\n{context[:4000]}\n\n"
            "Summarize the parts relevant to the query."
        )},
    ]

    try:
        from utility_model import utility_complete
        summary = (
            utility_complete(
                prompt_messages,
                max_tokens=256,
                temperature=0.3,
                activity_operation="session_search_summary",
            ) or ""
        ).strip()
        if summary:
            return f"[{date_str}] {summary}"
    except Exception:
        pass

    try:
        from agent_loop import _get_openai_client, get_backend_base_url, get_llmster_model
        base_url = get_backend_base_url()
        is_lmstudio = "1234" in (base_url or "")
        model = (get_llmster_model() or "local-model") if is_lmstudio else "local-model"
        api_key = "lm-studio"

        client = _get_openai_client(base_url, api_key)
        response = client.chat.completions.create(
            model=model,
            messages=prompt_messages,
            max_tokens=256,
            temperature=0.3,
            stream=False,
        )
        summary = (response.choices[0].message.content or "").strip()
        if summary:
            return f"[{date_str}] {summary}"
    except Exception:
        pass

    # Fallback: return truncated context
    truncated = context[:500] + ("..." if len(context) > 500 else "")
    return f"[{date_str}] {truncated}"
