hermes-agent/tools/session_search_tool.py

#!/usr/bin/env python3
"""Session Search Tool - long-term conversation recall over the SQLite session DB.

Single-shape tool; the mode is inferred from the args: DISCOVERY (``query``;
FTS5 deduped by lineage, adaptive detail hydrates only the top result),
SCROLL (``session_id`` + ``around_message_id``; ±window around the anchor),
READ (``session_id`` alone; whole session or head/tail), BROWSE (no args).
No LLM calls — every shape returns actual DB messages.
"""

import json
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional, Union

from hermes_state_common import _RESET_END_REASONS

# Hidden from browsing/searching: integrations (HERMES_SESSION_SOURCE=tool),
# delegate subagent runs, kanban workers — not the user's history.
_HIDDEN_SESSION_SOURCES = ("kanban", "subagent", "tool")

# Searchable but DEMOTED below interactive sessions: cron sessions' repetitive
# vocabulary dominates bare BM25 and starves out the user's own sessions
# ("recall blindness"). Demoting keeps them reachable when they're the only match.
_DEMOTED_SESSION_SOURCES = ("cron",)

# FTS rows scanned before dedup-by-lineage — well above the handful of distinct
# sessions a query returns, so interactive matches buried under cron hits are
# still in hand for the demotion pass.
_DISCOVER_SCAN_LIMIT = 300

# Raw FTS rows are only a discovery-plan input; the response hydrates its own
# anchored window and bookends after lineage dedup.
_DISCOVER_SEARCH_FIELDS = ("id", "session_id", "role", "snippet", "source", "model", "session_started")

# Generated context-compaction handoff summaries (agent/context_compressor.py);
# excluded from bookends so huge compaction payloads aren't re-introduced.
_COMPACTION_PREFIXES = ("[CONTEXT COMPACTION", "[CONTEXT SUMMARY]:")

# /new, /reset, idle/daily expiry and CLI /new ("new_session") end the
# predecessor WITHOUT carrying its transcript forward — unlike compression
# continuations and live delegation children. Derived from the gateway set so
# this tool and the recovery fence cannot drift.
_FRESH_RESET_END_REASONS = frozenset(_RESET_END_REASONS) | {"new_session"}


def _quiet(fn, default, msg, *log_args, with_exc: bool = False):
    """Call ``fn()``; on any exception debug-log *msg* (appending the exception
    as a final ``%s`` arg when *with_exc*) and return *default*."""
    try:
        return fn()
    except Exception as e:
        logging.debug(msg, *(log_args + (e,) if with_exc else log_args), exc_info=True)
        return default


def _format_timestamp(ts: Union[int, float, str, None]) -> str:
    """Unix timestamp (number or numeric string) or ISO string -> readable date.
    "unknown" for None; str(ts) if conversion fails."""
    if ts is None:
        return "unknown"
    try:
        value = ts
        if isinstance(ts, str):
            if not ts.replace(".", "").replace("-", "").isdigit():
                return ts
            value = float(ts)
        if isinstance(value, (int, float)):
            return datetime.fromtimestamp(value).strftime("%B %d, %Y at %I:%M %p")
    except (ValueError, OSError, OverflowError) as e:
        logging.debug("Failed to format timestamp %s: %s", ts, e, exc_info=True)
    except Exception as e:
        logging.debug("Unexpected error formatting timestamp %s: %s", ts, e, exc_info=True)
    return str(ts)


def _session_meta_block(meta: Dict[str, Any]) -> Dict[str, Any]:
    """The ``session_meta`` sub-object shared by read/scroll responses."""
    return {"when": _format_timestamp(meta.get("started_at")), "source": meta.get("source"),
            "model": meta.get("model"), "title": meta.get("title")}


def _ok(**payload) -> str:
    """Serialize a successful tool result (``success`` first, then *payload* in order)."""
    return json.dumps({"success": True, **payload}, ensure_ascii=False)


def _is_compaction_summary(content: str) -> bool:
    """Return True if *content* looks like a generated compaction handoff."""
    return bool(content) and content.lstrip().startswith(_COMPACTION_PREFIXES)


def _resolve_to_parent(db, session_id: str) -> tuple[str, bool]:
    """Walk parent_session_id to the lineage root -> ``(root_id, has_compression_hop)``.
    The flag distinguishes a compression-split lineage (parent content summarised
    away) from a delegation lineage (child content still visible to the parent).
    Falls back to ``(session_id, False)`` on errors."""
    if not session_id:
        return session_id, False
    visited: set[str] = set()
    cur, has_compression = session_id, False
    while cur and cur not in visited:
        visited.add(cur)
        s = _quiet(lambda: db.get_session(cur), None, "Error resolving parent for %s: %s", cur, with_exc=True)
        if not s:
            break
        if s.get("end_reason") == "compression":
            has_compression = True
        if not s.get("parent_session_id"):
            break
        cur = s["parent_session_id"]
    return cur, has_compression


def _resolve_lineage(db, session_id: str) -> str:
    """Return only the lineage root (ignores the compression hop)."""
    return _resolve_to_parent(db, session_id)[0]


def _session_end_reason(db, session_id: str) -> Optional[str]:
    """Return the session's ``end_reason``, or None if missing/unended/error."""
    if not session_id:
        return None
    try:
        s = db.get_session(session_id)
    except Exception:
        return None
    return (s.get("end_reason") or None) if s else None


def _session_left_live_context(db, session_id: str) -> bool:
    """True when *session_id*'s transcript is no longer in anyone's live context:
    ``compression`` (summarised into the continuation child) or a fresh reset
    (child starts empty). Everything else stays excluded from same-lineage
    recall — live delegation children (``end_reason is None``) are visible to
    the parent agent, and ``branched`` parents were copied verbatim into the
    branch child, so their content IS the current context."""
    end_reason = _session_end_reason(db, session_id)
    return end_reason == "compression" or end_reason in _FRESH_RESET_END_REASONS


def _get_message_storage_state(db, message_id) -> Optional[Dict[str, Any]]:
    """Return the owning session and visibility flags for *message_id*."""
    if not message_id:
        return None

    def _lookup():
        with db._lock:
            return db._conn.execute(
                "SELECT session_id, active, compacted FROM messages WHERE id = ?", (message_id,)
            ).fetchone()

    row = _quiet(_lookup, None, "message storage-state lookup failed for %s", message_id)
    return dict(row) if row is not None else None


def _is_compacted_state(state: Optional[Dict[str, Any]]) -> bool:
    """Compaction archives are ``active=0, compacted=1`` (content summarised
    away by archive_and_compact). Rewind/undo rows are ``active=0, compacted=0``
    and must stay hidden."""
    return state is not None and state["active"] == 0 and state["compacted"] == 1


def _is_compacted_message(db, message_id) -> bool:
    """True if *message_id* is a compaction-archived row — pre-compaction content
    no longer in live context, so it should stay discoverable even on the
    current session. False on any error (caller falls back to skipping)."""
    return _is_compacted_state(_get_message_storage_state(db, message_id))


def _annotate_rebuild_status(db, payload: Dict[str, Any]) -> None:
    """Add a rebuild-progress note while the deferred FTS backfill is running,
    so the agent can explain thin/slow results instead of treating them as
    ground truth. No-op (never raises) when no rebuild is pending."""
    try:
        status = db.fts_rebuild_status()
    except Exception:
        return
    if status is None:
        return
    payload["index_rebuild"] = {"percent": status["percent"], "note": (
        f"The search index is rebuilding in the background ({status['percent']}% done, "
        f"{status['indexed']:,} of {status['total']:,} messages). Results from older messages "
        f"may be incomplete until it finishes.")}


def _order_for_recall(raw_results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Stable-sort FTS rows so interactive sessions rank above automation.
    BM25 order is preserved within each class; only cross-class order changes,
    so a cron hit never displaces an interactive hit during lineage dedup."""
    return sorted(raw_results, key=lambda r: 1 if (r.get("source") or "") in _DEMOTED_SESSION_SOURCES else 0)


def _shape_message(m: Dict[str, Any], anchor_id: Optional[int] = None,
                   max_content_len: Optional[int] = None) -> Dict[str, Any]:
    """Slim a message row for the tool response. Keeps content even if empty
    (absent content is meaningful — tool-call-only assistant turns). With
    *max_content_len*, content is truncated and ``content_truncated`` /
    ``original_content_chars`` added."""
    content = m.get("content")
    if isinstance(content, str) and "\x1b" in content:
        # Recalled messages can carry ANSI escapes (archived terminal output).
        from tools.ansi_strip import strip_ansi

        content = strip_ansi(content)
    original_chars = None
    if max_content_len and content and len(content) > max_content_len:
        original_chars = len(content)
        content = content[:max_content_len] + "…"
    entry = {"id": m.get("id"), "role": m.get("role"), "content": content, "timestamp": m.get("timestamp")}
    entry.update({k: m.get(k) for k in ("tool_name", "tool_calls", "tool_call_id") if m.get(k)})
    if anchor_id is not None and m.get("id") == anchor_id:
        entry["anchor"] = True
    if original_chars is not None:
        entry["content_truncated"] = True
        entry["original_content_chars"] = original_chars
    return {k: v for k, v in entry.items() if v is not None or k == "content"}


def _session_link(session_id: str, profile: str = None) -> str:
    """The reference the agent writes to point the user at a session — same
    value the desktop composer emits, so it renders as a titled link. The
    profile segment is omitted when it can't be named confidently (a bare id
    still resolves, it just can't disambiguate across profiles)."""
    name = (profile or "").strip()
    if not name:
        def _active():
            from hermes_cli.profiles import get_active_profile_name

            resolved = get_active_profile_name()
            return "" if resolved == "custom" else resolved

        name = _quiet(_active, "", "get_active_profile_name failed for session link")
    return f"@session:{name}/{session_id}" if name else f"@session:{session_id}"


def _normalize_title_query(query: str) -> str:
    """Strip common quoting the model may include around a remembered title."""
    return query.strip().strip("`'\"")


def _title_match_result(db, query: str, current_lineage_root: Optional[str]) -> Optional[Dict[str, Any]]:
    """Return a discovery-shaped result when the query matches a session title."""
    title_query = _normalize_title_query(query)
    if not title_query:
        return None
    session_id = _quiet(lambda: db.resolve_session_by_title(title_query), None,
                        "resolve_session_by_title failed for %r", title_query)
    if not session_id:
        return None
    lineage_root = _resolve_lineage(db, session_id)
    # Same-lineage title hits are in-context only while the session is live;
    # /new-reset and compression-ended parents are not.
    if (
        current_lineage_root
        and lineage_root == current_lineage_root
        and not _session_left_live_context(db, session_id)):
        return None

    session_meta = _quiet(lambda: db.get_session(lineage_root) or db.get_session(session_id), None,
                          "get_session failed for title match %s", session_id) or {}
    if session_meta.get("source") in _HIDDEN_SESSION_SOURCES:
        return None
    messages = _quiet(lambda: db.get_messages(session_id), [], "get_messages failed for title match %s", session_id)
    anchor_id = messages[0].get("id") if messages else None
    view = {}
    if anchor_id is not None:
        view = _quiet(lambda: db.get_anchored_view(session_id, anchor_id, window=5, bookend=3), {},
                      "get_anchored_view failed for title match %s/%s", session_id, anchor_id)
    entry = {
        "session_id": session_id, "when": _format_timestamp(session_meta.get("started_at")),
        "source": session_meta.get("source", "unknown"), "model": session_meta.get("model") or "unknown",
        "title": session_meta.get("title") or title_query, "matched_role": "session_title",
        "match_message_id": anchor_id,
        "snippet": f"Session title matched: {session_meta.get('title') or title_query}",
        "bookend_start": [_shape_message(m) for m in (view.get("bookend_start") or messages[:3])],
        "messages": [_shape_message(m, anchor_id=anchor_id) for m in (view.get("window") or messages[:5])],
        "bookend_end": [_shape_message(m) for m in (view.get("bookend_end") or messages[-3:])],
        "messages_before": view.get("messages_before", 0),
        "messages_after": view.get("messages_after", max(len(messages) - 5, 0)),
        "detail": "full", "_lineage_root": lineage_root}
    if lineage_root and lineage_root != session_id:
        entry["parent_session_id"] = lineage_root
    return entry


def _discover_payload(db, query: str, detail: str, results: list, **extra) -> str:
    payload = {"success": True, "mode": "discover", "query": query, "detail": detail,
               "results": results, "count": len(results), **extra}
    _annotate_rebuild_status(db, payload)
    return json.dumps(payload, ensure_ascii=False)


def _dedupe_by_lineage(db, raw_results, limit, seen_sessions, current_session_id, current_lineage_root) -> None:
    """Fill *seen_sessions* (lineage_root -> first surviving FTS row) up to *limit*.

    The raw owning session_id stays on the row — only it pairs validly with the
    FTS match id for the anchored window. Current-lineage hits are skipped
    UNLESS the transcript left live context: compression-ended session, /new-
    reset predecessor (hiding it made gateway recall blind after every /new),
    or an in-place compacted row on the SAME session_id. A live delegation
    child has end_reason=None, so it stays excluded.
    """
    for r in raw_results:
        if len(seen_sessions) >= limit:
            break
        raw_sid = r["session_id"]
        resolved_sid, _ = _resolve_to_parent(db, raw_sid)
        is_compacted_hit = _is_compacted_message(db, r.get("id"))
        is_ended_session = _session_left_live_context(db, raw_sid)
        if (
            current_lineage_root
            and resolved_sid == current_lineage_root
            and not (is_ended_session or is_compacted_hit)):
            continue
        if current_session_id and raw_sid == current_session_id and not is_compacted_hit:
            continue
        seen_sessions.setdefault(resolved_sid, {**r, "_lineage_root": resolved_sid})


def _bookend(view: Dict[str, Any], key: str) -> List[Dict[str, Any]]:
    return [_shape_message(m, max_content_len=1200) for m in (view.get(key) or [])
            if not _is_compaction_summary(m.get("content", ""))]


def _hydrate_hit(db, lineage_root: str, match_info: Dict[str, Any], result_detail: str) -> Optional[Dict[str, Any]]:
    """Build one discovery result from a surviving FTS row; None if the anchored
    view can't be loaded (the hit is dropped)."""
    hit_sid = match_info.get("session_id") or lineage_root
    msg_id = match_info.get("id")
    try:
        view = db.get_anchored_view(hit_sid, msg_id, window=5, bookend=3)
    except Exception as e:
        logging.warning("get_anchored_view failed for %s/%s: %s", hit_sid, msg_id, e, exc_info=True)
        return None
    session_meta = _quiet(lambda: db.get_session(lineage_root), None, "get_session failed for %s", lineage_root) or {}
    full = result_detail == "full"
    window_messages = view.get("window") or []
    if not full:
        window_messages = [m for m in window_messages if m.get("id") == msg_id]
    entry = {
        "session_id": hit_sid,
        "when": _format_timestamp(session_meta.get("started_at") or match_info.get("session_started")),
        "source": session_meta.get("source") or match_info.get("source", "unknown"),
        "model": session_meta.get("model") or match_info.get("model") or "unknown",
        "title": session_meta.get("title") or None, "matched_role": match_info.get("role"),
        "match_message_id": msg_id, "snippet": match_info.get("snippet") or "",
        "bookend_start": _bookend(view, "bookend_start") if full else [],
        "messages": [_shape_message(m, anchor_id=msg_id, max_content_len=4000) for m in window_messages],
        "bookend_end": _bookend(view, "bookend_end") if full else [],
        "messages_before": view.get("messages_before", 0), "messages_after": view.get("messages_after", 0),
        "detail": result_detail}
    if lineage_root and lineage_root != hit_sid:
        entry["parent_session_id"] = lineage_root
    return entry


def _discover(db, query: str, role_filter: Optional[List[str]], limit: int, sort: Optional[str],
              detail: str, current_session_id: str = None, link_profile: str = None) -> str:
    """Discovery shape: FTS5 plus adaptive or full result hydration."""
    role_list = role_filter if role_filter else ["user", "assistant"]
    current_lineage_root = _resolve_lineage(db, current_session_id) if current_session_id else None
    title_result = _title_match_result(db, query, current_lineage_root)

    try:
        raw_results = db.search_messages(
            query=query, role_filter=role_list, exclude_sources=list(_HIDDEN_SESSION_SOURCES),
            limit=_DISCOVER_SCAN_LIMIT, offset=0, sort=sort, fields=_DISCOVER_SEARCH_FIELDS)
    except Exception as e:
        logging.error("FTS5 search failed: %s", e, exc_info=True)
        return tool_error(f"Search failed: {e}", success=False)

    # Demote cron rows below interactive ones BEFORE dedup so a high-volume
    # cron corpus can't starve the user's own sessions out of the top `limit`.
    raw_results = _order_for_recall(raw_results)

    if not raw_results and not title_result:
        return _discover_payload(db, query, detail, [], message=(
            "No matching sessions found. FTS5 ANDs all terms by default — "
            "broaden with OR (`alpha OR beta`), exact-match with quoted "
            "phrases, exclude with NOT, or prefix-match with `deploy*`."))

    seen_sessions: Dict[str, Dict[str, Any]] = {}
    results = []
    if title_result:
        title_lineage = title_result.pop("_lineage_root", None)
        if title_lineage:
            seen_sessions[title_lineage] = {"_title_only": True}
        results.append(title_result)
    _dedupe_by_lineage(db, raw_results, limit, seen_sessions, current_session_id, current_lineage_root)

    for lineage_root, match_info in seen_sessions.items():
        if match_info.get("_title_only"):
            continue
        # Adaptive: only the top-ranked result is fully hydrated.
        entry = _hydrate_hit(db, lineage_root, match_info, "full" if detail == "full" or not results else "compact")
        if entry is not None:
            results.append(entry)
    for entry in results:
        entry["link"] = _session_link(entry["session_id"], link_profile)
    return _discover_payload(db, query, detail, results, sessions_searched=len(seen_sessions), link_hint=(
        "When referring the user to a session, write its `link` value "
        "verbatim inline mid-sentence (it renders as a titled link) — never "
        "as markdown, in backticks, on its own line, or next to the "
        "title/id/date. To read more around a compact result, scroll: "
        "session_search(session_id=..., around_message_id=match_message_id)."))


def _resolve_profile_db(profile: str):
    """Open another profile's ``state.db`` read-only (no write lock — safe on a
    live DB), or None for the current profile."""
    if profile is None or not str(profile).strip():
        return None
    from hermes_cli import profiles as profiles_mod
    from hermes_state import SessionDB

    canon = profiles_mod.normalize_profile_name(profile)
    profiles_mod.validate_profile_name(canon)
    if not profiles_mod.profile_exists(canon):
        raise ValueError(f"profile '{canon}' does not exist")
    return SessionDB(db_path=profiles_mod.get_profile_dir(canon) / "state.db", read_only=True)


def _locate_session_db(session_id: str):
    """Scan every profile's ``state.db`` for a session id -> ``(db, profile_name)``
    or ``(None, None)``. Ids are globally unique, so the first hit is authoritative."""
    from pathlib import Path

    try:
        from hermes_cli import profiles as profiles_mod
        from hermes_state import SessionDB
    except Exception:
        return None, None
    targets = [("default", profiles_mod.get_profile_dir("default"))]
    targets += _quiet(lambda: [(info.name, info.path) for info in profiles_mod.list_profiles()],
                      [], "list_profiles failed during session locate")
    seen: set = set()
    for name, home in targets:
        db_path = Path(home) / "state.db"
        if str(db_path) in seen or not db_path.exists():
            continue
        seen.add(str(db_path))
        try:
            pdb = SessionDB(db_path=db_path, read_only=True)
        except Exception:
            continue
        if _quiet(lambda: pdb.get_session(session_id), None,
                  "get_session probe failed for %s in %s", session_id, name):
            return pdb, name
        pdb.close()
    return None, None


def _get_session_meta(db, session_id: str) -> dict:
    """``db.get_session`` that degrades to ``{}`` on error."""
    return _quiet(lambda: db.get_session(session_id), None,
                  "get_session failed for %s: %s", session_id, with_exc=True) or {}


def _read_session(db, session_id: str, head: int = 20, tail: int = 10, link_profile: str = None) -> str:
    """Read shape: whole session, or first ``head`` + last ``tail`` messages
    with a pointer to scroll the middle."""
    meta = _get_session_meta(db, session_id)
    if not meta:
        return tool_error(f"session_id not found: {session_id}", success=False)

    try:
        rows = db.get_messages(session_id)
    except Exception as e:
        logging.error("get_messages failed for %s: %s", session_id, e, exc_info=True)
        return tool_error(f"failed to load session: {e}", success=False)

    shaped = [_shape_message(m) for m in rows]
    total = len(shaped)
    truncated = total > head + tail
    extra = {"message": (f"Session has {total} messages; showing first {head} + last {tail}. "
                         "Pass around_message_id (any id above) to scroll the middle.")} if truncated else {}
    return _ok(mode="read", session_id=session_id, link=_session_link(session_id, link_profile),
               session_meta=_session_meta_block(meta), message_count=total, truncated=truncated,
               messages=shaped[:head] + shaped[-tail:] if truncated else shaped, **extra)


def _list_recent_sessions(db, limit: int, current_session_id: str = None, link_profile: str = None) -> str:
    """Browse shape: metadata for the most recent sessions (no LLM, no FTS5)."""
    try:
        # list_sessions_rich (include_children=False) already applies the
        # canonical child classifier: roots, /branch children and /new-reset
        # children are admitted, delegation/compression children hidden.
        # Re-classifying here re-hid legacy reset children — trust the query.
        sessions = db.list_sessions_rich(
            limit=limit + 15,  # extra so we can skip current / compression roots
            exclude_sources=list(_HIDDEN_SESSION_SOURCES),
            order_by_last_active=True)

        current_root, has_compression_hop = (
            _resolve_to_parent(db, current_session_id) if current_session_id else (None, False))
        results = []
        for s in sessions:
            sid = s.get("id", "")
            if sid == current_session_id:
                continue
            # Compression continuation: the root's turns were summarised into
            # the live child, so hide the root. /new-reset children share a
            # root but carry no transcript — keep that root browsable.
            if has_compression_hop and current_root and sid == current_root:
                continue
            results.append({
                "session_id": sid, "link": _session_link(sid, link_profile), "title": s.get("title") or None,
                "source": s.get("source", ""), "started_at": s.get("started_at", ""),
                "last_active": s.get("last_active", ""), "message_count": s.get("message_count", 0),
                "preview": s.get("preview", "")})
            if len(results) >= limit:
                break
        return _ok(mode="browse", results=results, count=len(results), message=(
            f"Showing {len(results)} most recent sessions. Pass a query= to search, "
            "or session_id+around_message_id to scroll."))
    except Exception as e:
        logging.error("Error listing recent sessions: %s", e, exc_info=True)
        return tool_error(f"Failed to list recent sessions: {e}", success=False)


def _clamp_int(value, default: int, lo: int, hi: int) -> int:
    if not isinstance(value, int):
        try:
            value = int(value)
        except (TypeError, ValueError):
            value = default
    return max(lo, min(value, hi))


def _anchor_in_live_context(db, anchor_state, anchor_session_id: str, current_session_id: str) -> bool:
    """True when the scroll anchor is still in the caller's active context and
    must be rejected. Same-lineage history that has LEFT live context (compacted
    rows, compression-ended parents, /new-reset predecessors) is allowed, so
    scroll never rejects a result discovery just returned."""
    a_root = _resolve_lineage(db, anchor_session_id)
    c_root = _resolve_lineage(db, current_session_id)
    if not (a_root and c_root and a_root == c_root):
        return False
    if _is_compacted_state(anchor_state):
        return False
    # Rewind/undo rows (active=0, compacted!=1) never count as out-of-context history.
    is_inactive_non_compacted = (
        anchor_state is not None and anchor_state["active"] == 0 and anchor_state["compacted"] != 1)
    return is_inactive_non_compacted or not _session_left_live_context(db, anchor_session_id)


def _rebind_to_owner(db, session_id: str, owning: str, around_message_id: int, window: int):
    """Lineage rebind: the caller paired a parent session_id with a message id
    that lives in a descendant (compaction / delegation create child sessions).
    Returns ``(view, warning)`` from the owning session, or ``(None, None)``."""
    a_root = _resolve_lineage(db, session_id)
    o_root = _resolve_lineage(db, owning)
    if not (a_root and o_root and a_root == o_root):
        return None, None
    rebind_view = _quiet(lambda: db.get_messages_around(owning, around_message_id, window=window),
                         None, "rebind get_messages_around failed: %s", with_exc=True)
    if not (rebind_view and rebind_view.get("window")):
        return None, None
    return rebind_view, f"around_message_id {around_message_id} lives in {owning} (child of {session_id}); rebound transparently"


def _scroll(db, session_id: str, around_message_id: int, window: int = 5,
            current_session_id: str = None) -> str:
    """Scroll shape: a window of messages centered on an anchor (no FTS5, no
    bookends). Rebinds silently if the anchor lives in a same-lineage child."""
    if not isinstance(session_id, str) or not session_id.strip():
        return tool_error("scroll requires session_id", success=False)
    session_id = session_id.strip()
    try:
        around_message_id = int(around_message_id)
    except (TypeError, ValueError):
        return tool_error("scroll requires integer around_message_id", success=False)
    window = _clamp_int(window, 5, 1, 20)

    # Locate the anchor BEFORE the current-lineage guard (see _anchor_in_live_context).
    anchor_state = _get_message_storage_state(db, around_message_id)
    owning_session_id = anchor_state.get("session_id") if anchor_state is not None else None

    if current_session_id and _anchor_in_live_context(
        db, anchor_state, owning_session_id or session_id, current_session_id):
        return tool_error("scroll rejected: anchor lives in the current session lineage (already in your active context)", success=False)

    session_meta = _get_session_meta(db, session_id)
    if not session_meta:
        return tool_error(f"session_id not found: {session_id}", success=False)

    try:
        view = db.get_messages_around(session_id, around_message_id, window=window)
    except Exception as e:
        logging.error("get_messages_around failed: %s", e, exc_info=True)
        return tool_error(f"failed to load messages: {e}", success=False)

    messages = view.get("window") or []
    rebind_warning = None
    if not messages and owning_session_id and owning_session_id != session_id:
        rebind_view, rebind_warning = _rebind_to_owner(
            db, session_id, owning_session_id, around_message_id, window)
        if rebind_view is not None:
            view = rebind_view
            messages = view["window"]
            session_meta = _get_session_meta(db, owning_session_id) or session_meta
            session_id = owning_session_id

    if not messages:
        return tool_error(f"around_message_id {around_message_id} not in session_id {session_id}", success=False)

    return _ok(
        mode="scroll", session_id=session_id, around_message_id=around_message_id,
        session_meta=_session_meta_block(session_meta), window=window,
        messages=[_shape_message(m, anchor_id=around_message_id) for m in messages],
        messages_before=view.get("messages_before", 0), messages_after=view.get("messages_after", 0),
        hint=("Scroll forward: re-call with around_message_id = the LAST message's "
              "id; backward: the FIRST message's id (the boundary message repeats "
              "as an orientation marker). messages_before/messages_after < window "
              "means you've hit that end of the session."),
        **({"warning": rebind_warning} if rebind_warning else {}))


def _read_with_profile_fallback(db, sid: str, profile: Optional[str]) -> str:
    """Read shape. On a miss in the target profile, scan every profile (the
    model may have dropped the owning profile from the link) and tag the result
    with the profile it was found in."""
    result = _read_session(db, sid, link_profile=profile)
    if json.loads(result).get("success"):
        return result
    located, owner = _locate_session_db(sid)
    if located is not None:
        try:
            found = json.loads(_read_session(located, sid, link_profile=owner))
        finally:
            located.close()
        if found.get("success"):
            found["profile"] = owner
            return json.dumps(found, ensure_ascii=False)
    return result


def _dispatch(query, role_filter, limit, db, current_session_id, session_id,
              around_message_id, window, sort, profile, detail, owned_dbs) -> str:
    """Mode dispatch (see module docstring). Scroll wins over read/discovery when
    an anchor is set — the agent asked for a specific slice. Profile DBs opened
    here are appended to *owned_dbs* for the caller to close."""
    # A raw `@session:<profile>/<id>` link passed as session_id: ids never
    # contain "/", so a slash means profile/id — strip the prefix and adopt the
    # embedded profile only when none was passed explicitly.
    if isinstance(session_id, str) and "/" in session_id:
        emb_profile, _, emb_id = session_id.partition("/")
        if emb_id:
            session_id = emb_id
            if emb_profile and (profile is None or not str(profile).strip()):
                profile = emb_profile

    # Cross-profile read: swap in the named profile's DB (read-only) for every
    # shape. Current-lineage guards key off ids that won't collide, so they
    # stay inert.
    if profile is not None and str(profile).strip():
        try:
            profile_db = _resolve_profile_db(profile)
        except Exception as e:
            return tool_error(f"profile '{profile}': {e}", success=False)
        if profile_db is not None:
            db, current_session_id = profile_db, None
            owned_dbs.append(profile_db)

    has_session = isinstance(session_id, str) and bool(session_id.strip())
    if has_session and around_message_id is not None:
        return _scroll(db, session_id, around_message_id, window, current_session_id)
    if has_session:
        return _read_with_profile_fallback(db, session_id.strip(), profile)

    limit = _clamp_int(limit, 3, 1, 10)
    if not query or not isinstance(query, str) or not query.strip():
        return _list_recent_sessions(db, limit, current_session_id, link_profile=profile)

    role_list = ([r.strip() for r in role_filter.split(",") if r.strip()] or None) if isinstance(role_filter, str) else None
    sort_norm = sort.strip().lower() if isinstance(sort, str) else None
    if sort_norm not in ("newest", "oldest"):
        sort_norm = None
    detail_norm = "full" if isinstance(detail, str) and detail.strip().lower() == "full" else "adaptive"
    return _discover(
        db=db, query=query.strip(), role_filter=role_list, limit=limit, sort=sort_norm,
        detail=detail_norm, current_session_id=current_session_id, link_profile=profile)


def session_search(query: str = "", role_filter: str = None, limit: int = 3, db=None,
                   current_session_id: str = None, session_id: str = None, around_message_id: int = None,
                   window: int = 5, sort: str = None, profile: str = None, detail: str = "adaptive") -> str:
    """Run session search and close databases opened by this invocation.
    Parameter order is positional-compatible with older callers."""
    owned_dbs: List[Any] = []
    if db is None:
        try:
            from hermes_state import get_shared_session_db

            db = get_shared_session_db()
            owned_dbs.append(db)
        except Exception:
            logging.debug("SessionDB unavailable for session_search", exc_info=True)
            from hermes_state import format_session_db_unavailable

            return tool_error(format_session_db_unavailable(), success=False)

    try:
        return _dispatch(query, role_filter, limit, db, current_session_id, session_id,
                         around_message_id, window, sort, profile, detail, owned_dbs)
    finally:
        from hermes_state import release_or_close

        for owned_db in reversed(owned_dbs):
            _quiet(lambda: release_or_close(owned_db), None, "Failed to close session_search SessionDB")


def check_session_search_requirements() -> bool:
    """Requires the SQLite state database."""
    try:
        from hermes_state import _default_db_path
        return _default_db_path().parent.exists()
    except ImportError:
        return False


SESSION_SEARCH_SCHEMA = {
    "name": "session_search",
    "description": (
        "Recall past conversations: search or read old Hermes sessions (FTS5), or "
        "scroll inside one. Four shapes, picked by args: `query` = discovery "
        "(top-N matching sessions, top result fully hydrated); `session_id` + "
        "`around_message_id` = scroll (window of messages around an anchor); "
        "`session_id` alone = read a whole session — how you resolve an "
        "`@session:<profile>/<id>` link (split on '/' into profile + id); no "
        "args = browse recent sessions. Results are actual DB messages, no LLM. "
        "Searches conversation history ONLY — when the user gave a direct "
        "source (URL, file, contact, live system), inspect that first; never "
        "conclude 'not found' from history alone. Use for questions about past "
        "conversations: 'what did we do about X', 'where did we leave Y'. When "
        "referring the user to a session, write its `link` value verbatim "
        "inline (it renders as a titled link)."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": (
                    "Search query (discovery shape). Keywords, phrases, or boolean "
                    "expressions to find in past sessions. Omit to browse recent "
                    "sessions. Ignored when session_id + around_message_id are set "
                    "(scroll shape)."
                ),
            },
            "limit": {
                "type": "integer",
                "description": (
                    "Discovery shape only. Max sessions to return (default 3, max 10). "
                    "Bump to 5–10 when the topic likely spans several sessions and you "
                    "want to pick the right one to scroll into."
                ),
                "default": 3,
            },
            "sort": {
                "type": "string",
                "enum": ["newest", "oldest"],
                "description": (
                    "Discovery shape only. Temporal bias on top of FTS5 ranking: omit "
                    "for relevance-only (exploratory recall), 'newest' for "
                    "\"where did we leave X\", 'oldest' for \"how did X start\"."
                ),
            },
            "detail": {
                "type": "string",
                "enum": ["adaptive", "full"],
                "description": (
                    "Discovery shape only. 'adaptive' (default) fully hydrates the "
                    "top-ranked result and returns only the exact anchor message for "
                    "lower-ranked results. 'full' returns bookends and the complete "
                    "anchored window for every result."
                ),
                "default": "adaptive",
            },
            "session_id": {
                "type": "string",
                "description": (
                    "Scroll shape. Session to read inside. Use the session_id returned "
                    "from a prior discovery call. Must be paired with "
                    "around_message_id."
                ),
            },
            "around_message_id": {
                "type": "integer",
                "description": (
                    "Scroll shape. Message id to center the window on — use "
                    "match_message_id from a discovery result, or any id from a "
                    "prior window."
                ),
            },
            "window": {
                "type": "integer",
                "description": (
                    "Scroll shape only. Messages to return on each side of the anchor "
                    "(anchor itself always included). Clamped to [1, 20]. Default 5."
                ),
                "default": 5,
            },
            "role_filter": {
                "type": "string",
                "description": (
                    "Optional. Comma-separated roles to include. Discovery defaults to "
                    "'user,assistant' (tool output is usually noise). Pass "
                    "'user,assistant,tool' to include tool output (debugging tool "
                    "behaviour) or 'tool' to search tool output only."
                ),
            },
            "profile": {
                "type": "string",
                "description": (
                    "Optional. Read sessions from another Hermes profile's database "
                    "(read-only). Use when resolving an `@session:<profile>/<id>` link: "
                    "pass the profile segment here with session_id as the id segment. "
                    "Omit to use the current profile."
                ),
            },
        },
        "required": [],
    },
}


# --- Registry ---
from tools.registry import registry, tool_error

registry.register(
    name="session_search",
    toolset="session_search",
    schema=SESSION_SEARCH_SCHEMA,
    handler=lambda args, **kw: session_search(
        query=args.get("query") or "", limit=args.get("limit", 3), window=args.get("window", 5),
        detail=args.get("detail", "adaptive"), db=kw.get("db"), current_session_id=kw.get("current_session_id"),
        **{k: args.get(k) for k in ("role_filter", "session_id", "around_message_id", "sort", "profile")}),
    check_fn=check_session_search_requirements,
    emoji="🔍")