diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 3f3404b40b02..e0c51fb35388 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -3076,6 +3076,7 @@ def _execute(next_args: dict) -> Any:
                     around_message_id=next_args.get("around_message_id"),
                     window=next_args.get("window", 5),
                     sort=next_args.get("sort"),
+                    detail=next_args.get("detail", "adaptive"),
                     db=session_db,
                     current_session_id=agent.session_id,
                 ),
diff --git a/agent/tool_executor.py b/agent/tool_executor.py
index d0ce5621aab7..bdf4efc23585 100644
--- a/agent/tool_executor.py
+++ b/agent/tool_executor.py
@@ -1975,6 +1975,7 @@ def _execute(next_args: dict) -> Any:
                     around_message_id=next_args.get("around_message_id"),
                     window=next_args.get("window", 5),
                     sort=next_args.get("sort"),
+                    detail=next_args.get("detail", "adaptive"),
                     db=session_db,
                     current_session_id=agent.session_id,
                 )
diff --git a/contributors/emails/lepetitprince716@gmail.com b/contributors/emails/lepetitprince716@gmail.com
new file mode 100644
index 000000000000..568b68869a76
--- /dev/null
+++ b/contributors/emails/lepetitprince716@gmail.com
@@ -0,0 +1 @@
+lepetitprince716-prog
diff --git a/hermes_cli/model_switch.py b/hermes_cli/model_switch.py
index 3299843d2ab6..9b88f8465fb3 100644
--- a/hermes_cli/model_switch.py
+++ b/hermes_cli/model_switch.py
@@ -21,7 +21,9 @@
 from __future__ import annotations
 
 import logging
+import os
 import re
+import time
 from dataclasses import dataclass
 from typing import Any, List, NamedTuple, Optional
 
@@ -2054,6 +2056,262 @@ def _scoped_key_env(name: str) -> str:
         return ""
 
 
+# --- Parallel prefetch for provider model catalogs -----------------------
+#
+# When the 1h disk cache lapses (or on first cold open), list_authenticated_providers()
+# calls cached_provider_model_ids() serially for each authed provider.  Each call
+# that misses the cache blocks on a live /v1/models HTTP round-trip (1-8s per
+# provider depending on endpoint latency).  With 10+ authed providers the
+# cumulative serial blocking time is 15-30+ seconds.
+#
+# This prefetch function runs those same cached_provider_model_ids() calls in
+# parallel via ThreadPoolExecutor before the main picker build loop starts.
+# The main loop then hits warm cache entries instead of blocking on live
+# fetches.  Providers whose cache was already fresh (SWR or within TTL) are
+# skipped entirely — no wasted network calls.
+#
+# Net effect on a 13-provider setup with an expired cache:
+#   Before: ~20s serial blocking (sum of all provider latencies)
+#   After:  ~8s parallel (max single provider latency), rest served from cache
+
+_PARALLEL_PREFETCH_WORKERS = 8
+
+
+def _prefetch_provider_models_parallel(provider_slugs: list[str]) -> None:
+    """Fetch model catalogs for multiple providers in parallel.
+
+    Only providers whose cache entry is stale or missing are fetched; fresh
+    entries are skipped to avoid unnecessary network calls.  Each worker uses
+    :func:`update_provider_cache_entry` (thread-safe) to persist its result,
+    so concurrent writes to ``provider_models_cache.json`` don't clobber each
+    other.
+
+    :param provider_slugs: Hermes provider IDs to prefetch (e.g. ``["openrouter",
+        "anthropic", "deepseek"]``).  Unknown providers are silently skipped.
+    """
+    from hermes_cli.models import cached_provider_model_ids
+
+    # Quick-stale-check: skip providers whose cache is already fresh so we
+    # don't waste network calls on a warm cache.  We check staleness the same
+    # way cached_provider_model_ids does internally: load the cache, compare
+    # age to TTL.  This is a read-only check — if the cache file changes
+    # between this check and the actual fetch, cached_provider_model_ids will
+    # still do the right thing (it re-reads the cache internally).
+    from hermes_cli.models import (
+        _load_provider_models_cache,
+        _credential_fingerprint,
+        _PROVIDER_MODELS_CACHE_TTL,
+        normalize_provider,
+    )
+
+    now = time.time()
+    stale_slugs: list[str] = []
+    cache = _load_provider_models_cache()
+    for slug in provider_slugs:
+        normalized = normalize_provider(slug) or (slug or "")
+        if not normalized:
+            continue
+        entry = cache.get(normalized)
+        fp = _credential_fingerprint(normalized)
+        if (
+            isinstance(entry, dict)
+            and entry.get("fp") == fp
+            and isinstance(entry.get("models"), list)
+            and entry["models"]
+        ):
+            age = now - float(entry.get("at", 0))
+            if age < _PROVIDER_MODELS_CACHE_TTL:
+                continue  # fresh, skip
+        stale_slugs.append(normalized)
+
+    if not stale_slugs:
+        return
+
+    import concurrent.futures
+
+    def _fetch_one(slug: str) -> None:
+        try:
+            models = cached_provider_model_ids(slug, force_refresh=True)
+            # cached_provider_model_ids already persists the result, but in a
+            # non-locked read-modify-write.  Re-persist via the thread-safe
+            # path to guarantee no lost writes under concurrency.
+            if models:
+                from hermes_cli.models import update_provider_cache_entry
+                update_provider_cache_entry(slug, models)
+        except Exception:
+            pass  # best-effort; picker falls back to curated list
+
+    with concurrent.futures.ThreadPoolExecutor(
+        max_workers=min(_PARALLEL_PREFETCH_WORKERS, len(stale_slugs)),
+        thread_name_prefix="model-cache-prefetch",
+    ) as executor:
+        list(executor.map(_fetch_one, stale_slugs))
+
+
+def _collect_authed_provider_slugs(
+    models_dev_data: dict,
+    curated: dict[str, list[str]],
+    excluded: list[str],
+) -> list[str]:
+    """Quick-scan which providers have credentials, without fetching model lists.
+
+    Mirrors the credential-check logic from sections 1, 2, and 2b of
+    :func:`list_authenticated_providers` but **only** collects the provider
+    slugs — it never calls ``cached_provider_model_ids``.  The returned list
+    is consumed by :func:`_prefetch_provider_models_parallel` to warm the disk
+    cache in parallel before the serial picker build loop starts.
+
+    :param models_dev_data: The models.dev registry dict (from ``fetch_models_dev()``).
+    :param curated: The curated model-lists dict (``_PROVIDER_MODELS`` + extras).
+    :param excluded: Provider slugs to exclude (from ``model_catalog.excluded_providers``).
+    :returns: List of normalized provider slugs that have credentials.
+    """
+    import os
+    from agent.models_dev import PROVIDER_TO_MODELS_DEV
+    from hermes_cli.auth import PROVIDER_REGISTRY, _load_auth_store
+    from hermes_cli.providers import HERMES_OVERLAYS, ALIASES as _PROVIDER_ALIAS_TABLE
+    from hermes_cli.models import _AGGREGATOR_PROVIDERS as _AGG_PROVIDERS, CANONICAL_PROVIDERS
+
+    _excluded_set = {str(p).strip().lower() for p in excluded if p}
+    slugs: list[str] = []
+    seen: set[str] = set()
+
+    # --- Section 1: Hermes-mapped providers (PROVIDER_TO_MODELS_DEV) ---
+    for hermes_id, mdev_id in PROVIDER_TO_MODELS_DEV.items():
+        _alias_target = _PROVIDER_ALIAS_TABLE.get(hermes_id)
+        if (
+            _alias_target
+            and _alias_target != hermes_id
+            and _alias_target in _AGG_PROVIDERS
+        ):
+            continue
+        _canonical = hermes_id
+        try:
+            from providers import get_provider_profile as _gpp
+            _prof = _gpp(hermes_id)
+            if _prof is not None:
+                _canonical = _prof.name
+        except Exception:
+            pass
+        if _canonical != hermes_id:
+            continue
+        if hermes_id.lower() in seen:
+            continue
+        if hermes_id.lower() in _excluded_set or mdev_id.lower() in _excluded_set:
+            continue
+        pdata = models_dev_data.get(mdev_id)
+        if not isinstance(pdata, dict):
+            continue
+        pconfig = PROVIDER_REGISTRY.get(hermes_id)
+        if pconfig and pconfig.auth_type != "api_key":
+            continue
+        from hermes_cli.auth import is_runtime_provider_routable
+        if not is_runtime_provider_routable(hermes_id):
+            continue
+        if pconfig and pconfig.api_key_env_vars:
+            env_vars = list(pconfig.api_key_env_vars)
+        else:
+            env_vars = pdata.get("env", [])
+            if not isinstance(env_vars, list):
+                continue
+        has_creds = any(_scoped_key_env(ev) for ev in env_vars)
+        if not has_creds:
+            try:
+                store = _load_auth_store()
+                raw_pool_present = bool(
+                    store and store.get("credential_pool", {}).get(hermes_id)
+                )
+                if raw_pool_present:
+                    has_creds = _credential_pool_is_usable(
+                        hermes_id, raw_pool_present=True
+                    )
+            except Exception:
+                pass
+        if has_creds:
+            slugs.append(hermes_id)
+            seen.add(hermes_id.lower())
+
+    # --- Section 2: Hermes-only providers (HERMES_OVERLAYS) ---
+    _mdev_to_hermes = {v: k for k, v in PROVIDER_TO_MODELS_DEV.items()}
+    for pid, overlay in HERMES_OVERLAYS.items():
+        if pid.lower() in seen:
+            continue
+        hermes_slug = _mdev_to_hermes.get(pid, pid)
+        if hermes_slug.lower() in seen:
+            continue
+        if pid.lower() in _excluded_set or hermes_slug.lower() in _excluded_set:
+            continue
+        has_creds = False
+        if overlay.auth_type == "aws_sdk":
+            # Skip AWS SDK providers in prefetch — credential detection is heavier
+            continue
+        elif overlay.auth_type == "vertex":
+            try:
+                from agent.vertex_adapter import has_vertex_credentials
+                has_creds = has_vertex_credentials()
+            except Exception:
+                pass
+        elif overlay.extra_env_vars:
+            has_creds = any(_scoped_key_env(ev) for ev in overlay.extra_env_vars)
+        if not has_creds and overlay.auth_type == "api_key":
+            for _key in (pid, hermes_slug):
+                pcfg = PROVIDER_REGISTRY.get(_key)
+                if pcfg and pcfg.api_key_env_vars:
+                    if any(_scoped_key_env(ev) for ev in pcfg.api_key_env_vars):
+                        has_creds = True
+                        break
+        if not has_creds:
+            try:
+                store = _load_auth_store()
+                providers_store = store.get("providers", {}) if store else {}
+                if pid in providers_store or hermes_slug in providers_store:
+                    has_creds = True
+            except Exception:
+                pass
+        if not has_creds:
+            try:
+                if _credential_pool_is_usable(hermes_slug):
+                    has_creds = True
+            except Exception:
+                pass
+        if has_creds:
+            slugs.append(hermes_slug)
+            seen.add(pid.lower())
+            seen.add(hermes_slug.lower())
+
+    # --- Section 2b: Canonical providers cross-check ---
+    for _cp in CANONICAL_PROVIDERS:
+        if _cp.slug.lower() in seen:
+            continue
+        if _cp.slug.lower() in _excluded_set:
+            continue
+        _cp_config = PROVIDER_REGISTRY.get(_cp.slug)
+        _cp_has_creds = False
+        if _cp_config and _cp_config.api_key_env_vars:
+            _cp_has_creds = any(_scoped_key_env(ev) for ev in _cp_config.api_key_env_vars)
+        if not _cp_has_creds:
+            try:
+                _cp_store = _load_auth_store()
+                _cp_providers_store = _cp_store.get("providers", {}) if _cp_store else {}
+                if _cp.slug in _cp_providers_store:
+                    _cp_has_creds = True
+            except Exception:
+                pass
+        if not _cp_has_creds:
+            try:
+                if _credential_pool_is_usable(_cp.slug):
+                    _cp_has_creds = True
+            except Exception:
+                pass
+        if not _cp_has_creds and _cp_config and getattr(_cp_config, "auth_type", "") == "aws_sdk":
+            continue  # skip AWS SDK in prefetch
+        if _cp_has_creds:
+            slugs.append(_cp.slug)
+            seen.add(_cp.slug.lower())
+
+    return slugs
+
+
 def list_authenticated_providers(
     current_provider: str = "",
     current_base_url: str = "",
@@ -2129,7 +2387,6 @@ def list_authenticated_providers(
         except Exception:
             pass
 
-
     results: List[dict] = []
     seen_slugs: set = set()  # lowercase-normalized to catch case variants (#9545)
     _current_provider_norm = str(current_provider or "").strip().lower()
@@ -2258,6 +2515,29 @@ def _has_aws_sdk_creds_for_listing(slug: str) -> bool:
             live = [current_model]
         curated["lmstudio"] = live
 
+    # --- Parallel cache prefetch ---------------------------------------------
+    # The serial loops below (sections 1, 2, 2b) each call
+    # cached_provider_model_ids(slug) which blocks on a live /v1/models HTTP
+    # round-trip when the disk cache is stale or missing.  With many authed
+    # providers those serial round-trips stack to 15-30s on a cold/expired
+    # cache.  Pre-scanning which providers have credentials (without fetching
+    # their model lists) and warming their cache entries in parallel makes
+    # the subsequent serial calls hit fresh cache entries instead.
+    #
+    # Skipped entirely when refresh=True (the serial path already force-refreshes)
+    # and when there are 3 or fewer authed providers (serial is fast enough;
+    # avoids thread-pool overhead for the common 1-2 provider case).
+    _prefetch_slugs: list[str] = []
+    if not refresh:
+        _prefetch_slugs = _collect_authed_provider_slugs(
+            data, curated, excluded_providers or []
+        )
+    if len(_prefetch_slugs) > 3:
+        try:
+            _prefetch_provider_models_parallel(_prefetch_slugs)
+        except Exception:
+            pass  # best-effort; serial path still works as fallback
+
     # --- 1. Check Hermes-mapped providers ---
     from hermes_cli.models import _AGGREGATOR_PROVIDERS as _AGG_PROVIDERS
     from hermes_cli.providers import ALIASES as _PROVIDER_ALIAS_TABLE
diff --git a/hermes_cli/models.py b/hermes_cli/models.py
index d10e02b41278..2050fb53934c 100644
--- a/hermes_cli/models.py
+++ b/hermes_cli/models.py
@@ -3563,6 +3563,9 @@ def _load_provider_models_cache() -> dict:
         return {}
 
 
+_cache_write_lock = threading.Lock()
+
+
 def _save_provider_models_cache(data: dict) -> None:
     """Persist the cache dict. Best-effort — silent on any error."""
     try:
@@ -3574,6 +3577,31 @@ def _save_provider_models_cache(data: dict) -> None:
         pass
 
 
+def update_provider_cache_entry(provider: str, models: list[str]) -> None:
+    """Thread-safe single-entry update of the provider-models disk cache.
+
+    Used by parallel prefetch workers so concurrent fetches don't clobber
+    each other's writes via read-modify-write races on the shared JSON file.
+    Each worker loads the latest cache state under the lock, writes its own
+    entry, and saves — best-effort, silent on any error.
+    """
+    try:
+        normalized = normalize_provider(provider) or (provider or "")
+        if not normalized or not models:
+            return
+        fp = _credential_fingerprint(normalized)
+        with _cache_write_lock:
+            cache = _load_provider_models_cache()
+            cache[normalized] = {
+                "fp": fp,
+                "at": time.time(),
+                "models": list(models),
+            }
+            _save_provider_models_cache(cache)
+    except Exception:
+        pass
+
+
 def cached_provider_model_ids(
     provider: Optional[str],
     *,
diff --git a/hermes_state_search.py b/hermes_state_search.py
index b1cf669210fe..e8d29f413ee8 100644
--- a/hermes_state_search.py
+++ b/hermes_state_search.py
@@ -1390,7 +1390,6 @@ def _run_trigram_search(
                 m.session_id,
                 m.role,
                 snippet({table}, -1, '>>>', '<<<', '...', 40) AS snippet,
-                m.content,
                 m.timestamp,
                 m.tool_name,
                 s.source,
@@ -1583,7 +1582,7 @@ def _search_messages_like_fallback(
         sql = f"""
             SELECT m.id, m.session_id, m.role,
                    substr(m.content, max(1, instr(m.content, ?) - 40), 120) AS snippet,
-                   m.content, m.timestamp, m.tool_name,
+                   m.timestamp, m.tool_name,
                    s.source, s.model, s.started_at AS session_started
             FROM messages m
             JOIN sessions s ON s.id = m.session_id
@@ -1689,7 +1688,12 @@ def _finalize_search_matches(
             except Exception:
                 match["context"] = []
 
-        # Remove full content from result (snippet is enough, saves tokens)
+        # Full message content is never selected by any search route: every
+        # SELECT returns snippet + metadata only (saves I/O on multi-MB tool
+        # rows and the tokens a content column would cost downstream). The
+        # context query above re-fetches its 3-message window by id, so
+        # nothing reads content from the match rows themselves. The pop stays
+        # as a belt-and-braces guard for any future route that selects it.
         for match in matches:
             match.pop("content", None)
 
@@ -1823,7 +1827,6 @@ def _search_messages_impl(
                 m.session_id,
                 m.role,
                 snippet(messages_fts, -1, '>>>', '<<<', '...', 40) AS snippet,
-                m.content,
                 m.timestamp,
                 m.tool_name,
                 s.source,
@@ -1913,7 +1916,6 @@ def _search_messages_impl(
                         m.session_id,
                         m.role,
                         snippet(messages_fts_cjk, -1, '>>>', '<<<', '...', 40) AS snippet,
-                        m.content,
                         m.timestamp,
                         m.tool_name,
                         s.source,
@@ -2002,7 +2004,6 @@ def _search_messages_impl(
                         m.session_id,
                         m.role,
                         snippet(messages_fts_trigram, -1, '>>>', '<<<', '...', 40) AS snippet,
-                        m.content,
                         m.timestamp,
                         m.tool_name,
                         s.source,
@@ -2095,7 +2096,7 @@ def _search_messages_impl(
                            substr(m.content,
                                   max(1, instr(m.content, ?) - 40),
                                   120) AS snippet,
-                           m.content, m.timestamp, m.tool_name,
+                           m.timestamp, m.tool_name,
                            s.source, s.model, s.started_at AS session_started
                     FROM messages m
                     JOIN sessions s ON s.id = m.session_id
@@ -2272,7 +2273,7 @@ def _search_unindexed_gap(
                    substr(m.content,
                           max(1, instr(m.content, ?) - 40),
                           120) AS snippet,
-                   m.content, m.timestamp, m.tool_name,
+                   m.timestamp, m.tool_name,
                    s.source, s.model, s.started_at AS session_started
             FROM messages m
             JOIN sessions s ON s.id = m.session_id
diff --git a/run_agent.py b/run_agent.py
index 86aee2e53ae4..b6f655c355aa 100644
--- a/run_agent.py
+++ b/run_agent.py
@@ -4385,6 +4385,10 @@ def release_clients(self) -> None:
             self._close_cached_request_openai_client(reason="cache_evict")
         except Exception:
             pass
+        try:
+            self._close_cached_request_anthropic_client(reason="cache_evict")
+        except Exception:
+            pass
 
     def close(self) -> None:
         """Release all resources held by this agent instance.
@@ -4472,6 +4476,10 @@ def close(self) -> None:
             self._close_cached_request_openai_client(reason="agent_close")
         except Exception:
             pass
+        try:
+            self._close_cached_request_anthropic_client(reason="agent_close")
+        except Exception:
+            pass
 
         # 6c. Close the Codex app-server session. The runtime already drops
         # it on turn crash / retirement (agent/codex_runtime.py), but hard
@@ -5386,8 +5394,31 @@ def _abort_request_openai_client(self, client: Any, *, reason: str) -> None:
                 exc,
             )
 
+    def _request_anthropic_client_cache_ref(self) -> dict:
+        # Lazy init — tests build agents via AIAgent.__new__ without __init__.
+        cache = getattr(self, "_request_anthropic_client_cache", None)
+        if cache is None:
+            cache = {"client": None, "key": None, "poisoned": False, "in_use": False}
+            self._request_anthropic_client_cache = cache
+        return cache
+
+    def _request_anthropic_client_key(self) -> tuple:
+        """Cache key covering everything that forces a fresh client: credential
+        rotation, base URL / region changes, timeout changes (model switch),
+        and the 1M-context beta flag."""
+        if getattr(self, "provider", None) == "bedrock":
+            region = getattr(self, "_bedrock_region", "us-east-1") or "us-east-1"
+            return ("bedrock", region)
+        return (
+            "direct",
+            self._anthropic_api_key,
+            getattr(self, "_anthropic_base_url", None),
+            get_provider_request_timeout(self.provider, self.model),
+            bool(getattr(self, "_oauth_1m_beta_disabled", False)),
+        )
+
     def _create_request_anthropic_client(self, *, reason: str) -> Any:
-        """Build a request-local Anthropic client for one in-flight call.
+        """Build (or reuse) a request-local Anthropic client for one in-flight call.
 
         The shared ``_anthropic_client`` stays the long-lived primary, but the
         stale/interrupt watchdog runs on the poll thread and must never call
@@ -5399,24 +5430,56 @@ def _create_request_anthropic_client(self, *, reason: str) -> Any:
         worker performs the SDK-level close from its own context — the same
         ownership contract the OpenAI-wire path already uses.
 
+        Also mirrors the OpenAI-wire path's single-slot cache
+        (``_create_request_openai_client``): building ``anthropic.Anthropic``
+        means a fresh httpx pool and TCP+TLS handshake per call, so the client
+        is kept warm across sequential calls whose cache key (credentials,
+        base URL/region, timeout, 1M-beta flag) hasn't changed. ``in_use``
+        keeps a second concurrent call from sharing one pool's close/abort
+        lifecycle — it gets a fresh untracked client instead.
+
         Mirrors ``_rebuild_anthropic_client`` construction (direct + Bedrock,
-        1M-beta drop) but returns a fresh client instead of swapping the shared
-        one.
+        1M-beta drop) but returns a fresh/cached client instead of swapping
+        the shared one.
         """
         if self.api_mode == "anthropic_messages":
             self._try_refresh_anthropic_client_credentials()
-        _drop_1m = bool(getattr(self, "_oauth_1m_beta_disabled", False))
-        if getattr(self, "provider", None) == "bedrock":
+        key = self._request_anthropic_client_key()
+
+        stale = None
+        with self._openai_client_lock():
+            cache = self._request_anthropic_client_cache_ref()
+            cached = cache["client"]
+            if cached is not None and not cache["in_use"]:
+                if (
+                    not cache["poisoned"]
+                    and cache["key"] == key
+                    and not self._is_openai_client_closed(cached)
+                ):
+                    cache["in_use"] = True
+                    return cached
+                # Key changed (credential rotation, base URL/region, timeout,
+                # 1M-beta flip), poisoned by a cross-thread abort, or
+                # externally closed — never reuse; discard and rebuild below.
+                stale = cached
+                cache["client"] = None
+                cache["key"] = None
+                cache["poisoned"] = False
+        if stale is not None:
+            # Safe to close from this thread: in_use was False, so no worker
+            # thread owns the pool's FDs (same #29507 reasoning as OpenAI).
+            self._close_request_anthropic_client(stale, reason=f"reuse_evict:{reason}")
+
+        if key[0] == "bedrock":
             from agent.anthropic_adapter import build_anthropic_bedrock_client
-            region = getattr(self, "_bedrock_region", "us-east-1") or "us-east-1"
-            client = build_anthropic_bedrock_client(region)
+            client = build_anthropic_bedrock_client(key[1])
         else:
             from agent.anthropic_adapter import build_anthropic_client
             client = build_anthropic_client(
                 self._anthropic_api_key,
                 getattr(self, "_anthropic_base_url", None),
                 timeout=get_provider_request_timeout(self.provider, self.model),
-                drop_context_1m_beta=_drop_1m,
+                drop_context_1m_beta=key[4],
             )
         logger.debug(
             "Anthropic request client created (%s, shared=False) provider=%s model=%s",
@@ -5424,17 +5487,41 @@ def _create_request_anthropic_client(self, *, reason: str) -> Any:
             getattr(self, "provider", None),
             getattr(self, "model", None),
         )
+        with self._openai_client_lock():
+            cache = self._request_anthropic_client_cache_ref()
+            if cache["client"] is None:
+                cache["client"] = client
+                cache["key"] = key
+                cache["poisoned"] = False
+                cache["in_use"] = True
+            # else: a concurrent call holds the slot — hand this client out
+            # untracked; _close_request_anthropic_client fully closes
+            # untracked clients, preserving the per-request lifecycle.
         return client
 
     def _close_request_anthropic_client(self, client: Any, *, reason: str) -> None:
-        """Owner-thread full close of a request-local Anthropic client.
-
-        Force-closes the pool's TCP sockets first (CLOSE-WAIT hygiene, parity
-        with ``_close_openai_client``), then does the graceful SDK close. Safe
+        """Owner-thread close of a request-local Anthropic client.
+
+        On a clean finish (``reason`` in ``_REQUEST_CLIENT_REUSE_REASONS``)
+        the pool is kept warm in the cache slot for the next sequential call,
+        mirroring ``_close_request_openai_client``. Any other outcome
+        (error / kill / abort / stale-slot eviction) force-closes the pool's
+        TCP sockets first (CLOSE-WAIT hygiene, parity with
+        ``_close_openai_client``), then does the graceful SDK close. Safe
         because the caller owns the connection.
         """
         if client is None:
             return
+        with self._openai_client_lock():
+            cache = self._request_anthropic_client_cache_ref()
+            if cache["client"] is client:
+                if reason in self._REQUEST_CLIENT_REUSE_REASONS and not cache["poisoned"]:
+                    cache["in_use"] = False
+                    return
+                cache["client"] = None
+                cache["key"] = None
+                cache["poisoned"] = False
+                cache["in_use"] = False
         try:
             self._force_close_tcp_sockets(client)
             client.close()
@@ -5453,6 +5540,30 @@ def _close_request_anthropic_client(self, client: Any, *, reason: str) -> None:
                 exc,
             )
 
+    def _close_cached_request_anthropic_client(self, *, reason: str) -> None:
+        """Teardown hook: really close the cached per-request Anthropic client."""
+        with self._openai_client_lock():
+            cache = getattr(self, "_request_anthropic_client_cache", None)
+            client = cache["client"] if cache else None
+            in_use = bool(cache["in_use"]) if cache else False
+            if cache is not None:
+                cache["client"] = None
+                cache["key"] = None
+                cache["poisoned"] = False
+                cache["in_use"] = False
+        if client is None:
+            return
+        if in_use:
+            # A worker thread has this client checked out for an in-flight
+            # request — same #29507 reasoning as the OpenAI teardown hook.
+            self._abort_request_anthropic_client(client, reason=f"{reason}_in_flight")
+            return
+        try:
+            self._force_close_tcp_sockets(client)
+            client.close()
+        except Exception:
+            pass
+
     def _abort_request_anthropic_client(self, client: Any, *, reason: str) -> None:
         """Cross-thread abort for request-local Anthropic clients.
 
@@ -5464,6 +5575,13 @@ def _abort_request_anthropic_client(self, client: Any, *, reason: str) -> None:
         """
         if client is None:
             return
+        # A pool whose sockets were shut down from a stranger thread must
+        # never be reused: poison the cache slot so the owner-thread close
+        # discards it and the next create builds a fresh client.
+        with self._openai_client_lock():
+            cache = self._request_anthropic_client_cache_ref()
+            if cache["client"] is client:
+                cache["poisoned"] = True
         try:
             shutdown_count = self._force_close_tcp_sockets(client)
             # Same visibility contract as the OpenAI abort path (#72975):
diff --git a/tests/agent/test_anthropic_request_client_reuse.py b/tests/agent/test_anthropic_request_client_reuse.py
new file mode 100644
index 000000000000..d5983d62ec30
--- /dev/null
+++ b/tests/agent/test_anthropic_request_client_reuse.py
@@ -0,0 +1,157 @@
+"""Per-request Anthropic wire client reuse across sequential LLM calls.
+
+Mirrors ``tests/agent/test_request_client_reuse.py`` (the OpenAI-wire cache)
+for the Anthropic request-local client. Before this cache existed,
+``_create_request_anthropic_client`` built a fresh ``anthropic.Anthropic``
+(and its httpx pool) on every single LLM call and ``_close_request_anthropic_client``
+always fully closed it — no reuse across a turn's sequential tool-loop calls,
+unlike the OpenAI-wire path.
+
+- identical cache key (credentials, base URL, timeout, 1M-beta flag) → same
+  client object handed back (the reuse win);
+- key changes (credential rotation, base URL change) → evict + rebuild;
+- cross-thread abort poisons the slot → the owner-thread close does a real
+  close and the next create rebuilds;
+- non-reuse close reasons (error cleanups, stale/interrupt kills) discard —
+  only request_complete / stream_request_complete reuse;
+- teardown (release_clients / close) really closes the cached client, or
+  detaches it to the in-flight worker's own close when checked out.
+"""
+
+from unittest.mock import MagicMock, patch
+
+from run_agent import AIAgent
+
+
+class _StubClient:
+    """Minimal non-Mock client: _is_openai_client_closed reads ``is_closed``."""
+
+    def __init__(self):
+        self.is_closed = False
+
+    def close(self):
+        self.is_closed = True
+
+
+def _make_agent(provider="anthropic", base_url="https://api.anthropic.com", model="claude-sonnet-5"):
+    agent = AIAgent.__new__(AIAgent)
+    agent.provider = provider
+    agent.model = model
+    agent.api_mode = "anthropic_messages"
+    agent._anthropic_api_key = "sk-ant-test"
+    agent._anthropic_base_url = base_url
+    agent._oauth_1m_beta_disabled = False
+    # Real credential-refresh reaches auth/network state we don't need here;
+    # the cache logic under test is agnostic to it.
+    agent._try_refresh_anthropic_client_credentials = MagicMock(return_value=False)
+    return agent
+
+
+class _Harness:
+    """Patch the Anthropic client build/socket seams and record calls."""
+
+    def __init__(self, agent):
+        self.agent = agent
+        self.built = []  # reason
+        self._patchers = []
+
+    def __enter__(self):
+        def _fake_build(*a, **k):
+            self.built.append(k.get("drop_context_1m_beta"))
+            return _StubClient()
+
+        self._patchers = [
+            patch("agent.anthropic_adapter.build_anthropic_client", side_effect=_fake_build),
+            patch.object(self.agent, "_force_close_tcp_sockets", return_value=0),
+        ]
+        for p in self._patchers:
+            p.start()
+        return self
+
+    def __exit__(self, *exc):
+        for p in self._patchers:
+            p.stop()
+
+
+def test_reuse_on_identical_key_same_object():
+    agent = _make_agent()
+    with _Harness(agent) as h:
+        a = agent._create_request_anthropic_client(reason="chat_completion_request")
+        agent._close_request_anthropic_client(a, reason="request_complete")
+        assert not a.is_closed  # kept for reuse, not really closed
+
+        b = agent._create_request_anthropic_client(reason="chat_completion_request")
+        assert b is a
+        assert len(h.built) == 1
+
+
+def test_rebuild_on_credential_rotation():
+    agent = _make_agent()
+    with _Harness(agent):
+        a = agent._create_request_anthropic_client(reason="r")
+        agent._close_request_anthropic_client(a, reason="request_complete")
+
+        agent._anthropic_api_key = "sk-ant-rotated"
+        b = agent._create_request_anthropic_client(reason="r")
+        assert b is not a
+        assert a.is_closed  # stale slot really closed on eviction
+
+        agent._close_request_anthropic_client(b, reason="request_complete")
+        c = agent._create_request_anthropic_client(reason="r")
+        assert c is b
+
+
+def test_non_reuse_reason_discards_client():
+    agent = _make_agent()
+    with _Harness(agent):
+        a = agent._create_request_anthropic_client(reason="r")
+        agent._close_request_anthropic_client(a, reason="request_error_cleanup")
+        assert a.is_closed
+
+        b = agent._create_request_anthropic_client(reason="r")
+        assert b is not a
+
+
+def test_cross_thread_abort_poisons_slot():
+    agent = _make_agent()
+    with _Harness(agent):
+        a = agent._create_request_anthropic_client(reason="r")
+        agent._abort_request_anthropic_client(a, reason="interrupt")
+        # Owner thread's close now sees the poisoned slot and really closes.
+        agent._close_request_anthropic_client(a, reason="request_complete")
+        assert a.is_closed
+
+        b = agent._create_request_anthropic_client(reason="r")
+        assert b is not a
+
+
+def test_concurrent_call_gets_untracked_client():
+    agent = _make_agent()
+    with _Harness(agent):
+        a = agent._create_request_anthropic_client(reason="r")
+        # Slot still checked out (in_use=True) — a second concurrent call
+        # must not share it.
+        b = agent._create_request_anthropic_client(reason="r")
+        assert b is not a
+
+        # Finishing the untracked one does a real close, not a slot release.
+        agent._close_request_anthropic_client(b, reason="request_complete")
+        assert b.is_closed
+        # The tracked slot is unaffected and still reusable.
+        agent._close_request_anthropic_client(a, reason="request_complete")
+        c = agent._create_request_anthropic_client(reason="r")
+        assert c is a
+
+
+def test_agent_close_closes_cached_request_client():
+    agent = _make_agent()
+    with _Harness(agent):
+        a = agent._create_request_anthropic_client(reason="r")
+        agent._close_request_anthropic_client(a, reason="request_complete")
+        assert not a.is_closed
+
+        agent._close_cached_request_anthropic_client(reason="agent_close")
+        assert a.is_closed
+
+        # Idempotent: a second teardown must not error or double-act.
+        agent._close_cached_request_anthropic_client(reason="agent_close")
diff --git a/tests/hermes_cli/test_model_cache_parallel_prefetch.py b/tests/hermes_cli/test_model_cache_parallel_prefetch.py
new file mode 100644
index 000000000000..17a2ec202eff
--- /dev/null
+++ b/tests/hermes_cli/test_model_cache_parallel_prefetch.py
@@ -0,0 +1,250 @@
+"""Tests for parallel model-catalog prefetch and thread-safe cache writes.
+
+Regression tests for the serial /v1/models bottleneck: when the 1h disk cache
+lapses, ``list_authenticated_providers()`` previously fetched each authed
+provider's model list serially. With 10+ providers this stacked to 15-30s of
+blocking HTTP round-trips. The parallel prefetch warms stale cache entries
+concurrently via ThreadPoolExecutor before the serial picker loop starts.
+"""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Thread-safe cache entry update (hermes_cli/models.py)
+# ---------------------------------------------------------------------------
+
+class TestUpdateProviderCacheEntry:
+    """Verify ``update_provider_cache_entry`` writes safely under concurrency."""
+
+    def test_writes_new_entry(self, tmp_path, monkeypatch):
+        """A new entry is persisted to the cache file."""
+        import hermes_cli.models as mod
+
+        cache_path = tmp_path / "provider_models_cache.json"
+        monkeypatch.setattr(mod, "_provider_models_cache_path", lambda: cache_path)
+
+        with patch.object(mod, "_credential_fingerprint", return_value="fp1"):
+            mod.update_provider_cache_entry("openrouter", ["m1", "m2"])
+
+        cache = mod._load_provider_models_cache()
+        assert "openrouter" in cache
+        assert cache["openrouter"]["models"] == ["m1", "m2"]
+        assert cache["openrouter"]["fp"] == "fp1"
+
+    def test_does_not_clobber_other_entries(self, tmp_path, monkeypatch):
+        """Concurrent writes to different providers don't lose entries."""
+        import hermes_cli.models as mod
+
+        cache_path = tmp_path / "provider_models_cache.json"
+        monkeypatch.setattr(mod, "_provider_models_cache_path", lambda: cache_path)
+
+        # Seed with one entry
+        with patch.object(mod, "_credential_fingerprint", return_value="fp_a"):
+            mod.update_provider_cache_entry("provider_a", ["a1"])
+
+        # Write a second entry
+        with patch.object(mod, "_credential_fingerprint", return_value="fp_b"):
+            mod.update_provider_cache_entry("provider_b", ["b1"])
+
+        cache = mod._load_provider_models_cache()
+        assert "provider_a" in cache
+        assert cache["provider_a"]["models"] == ["a1"]
+        assert "provider_b" in cache
+        assert cache["provider_b"]["models"] == ["b1"]
+
+    def test_skips_empty_models(self, tmp_path, monkeypatch):
+        """Empty model lists are not written to cache."""
+        import hermes_cli.models as mod
+
+        cache_path = tmp_path / "provider_models_cache.json"
+        monkeypatch.setattr(mod, "_provider_models_cache_path", lambda: cache_path)
+
+        mod.update_provider_cache_entry("empty_provider", [])
+        cache = mod._load_provider_models_cache()
+        assert "empty_provider" not in cache
+
+    def test_concurrent_writes_no_lost_entries(self, tmp_path, monkeypatch):
+        """Multiple threads writing different providers concurrently — all land."""
+        import hermes_cli.models as mod
+        import concurrent.futures
+
+        cache_path = tmp_path / "provider_models_cache.json"
+        monkeypatch.setattr(mod, "_provider_models_cache_path", lambda: cache_path)
+
+        providers = [f"prov_{i}" for i in range(10)]
+
+        with patch.object(mod, "_credential_fingerprint", side_effect=lambda p: f"fp_{p}"):
+            with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
+                list(executor.map(
+                    lambda p: mod.update_provider_cache_entry(p, [f"model_{p}"]),
+                    providers,
+                ))
+
+        cache = mod._load_provider_models_cache()
+        for p in providers:
+            assert p in cache, f"{p} was lost in concurrent write"
+            assert cache[p]["models"] == [f"model_{p}"]
+
+
+# ---------------------------------------------------------------------------
+# Parallel prefetch (hermes_cli/model_switch.py)
+# ---------------------------------------------------------------------------
+
+class TestPrefetchProviderModelsParallel:
+    """Verify ``_prefetch_provider_models_parallel`` fetches concurrently."""
+
+    def test_skips_all_fresh_entries(self, monkeypatch):
+        """When all cache entries are fresh, no fetch is made."""
+        from hermes_cli.model_switch import _prefetch_provider_models_parallel
+
+        fresh_cache = {
+            "openrouter": {"fp": "fp", "at": time.time(), "models": ["m1"]},
+            "anthropic": {"fp": "fp", "at": time.time(), "models": ["m2"]},
+        }
+
+        with patch("hermes_cli.models._load_provider_models_cache", return_value=fresh_cache), \
+             patch("hermes_cli.models._credential_fingerprint", return_value="fp"), \
+             patch("hermes_cli.models.cached_provider_model_ids") as fetch:
+            _prefetch_provider_models_parallel(["openrouter", "anthropic"])
+
+        fetch.assert_not_called()
+
+    def test_fetches_only_stale_entries(self, monkeypatch):
+        """Only providers with stale/missing cache entries are fetched."""
+        from hermes_cli.model_switch import _prefetch_provider_models_parallel
+
+        cache = {
+            "fresh_prov": {"fp": "fp_f", "at": time.time(), "models": ["m1"]},
+        }
+
+        fetch_calls = []
+
+        def mock_fetch(slug, force_refresh=False):
+            fetch_calls.append(slug)
+            return [f"model_{slug}"]
+
+        with patch("hermes_cli.models._load_provider_models_cache", return_value=cache), \
+             patch("hermes_cli.models._credential_fingerprint", return_value="fp_f"), \
+             patch("hermes_cli.models.cached_provider_model_ids", side_effect=mock_fetch), \
+             patch("hermes_cli.models.update_provider_cache_entry"):
+            _prefetch_provider_models_parallel(["fresh_prov", "stale_prov"])
+
+        assert "fresh_prov" not in fetch_calls
+        assert "stale_prov" in fetch_calls
+
+    def test_fetches_in_parallel(self, monkeypatch):
+        """Multiple providers are fetched concurrently, not serially."""
+        from hermes_cli.model_switch import _prefetch_provider_models_parallel
+
+        # Track overlap: if serial, no two fetches should overlap in time.
+        active = []
+        max_concurrent = [0]
+        lock = __import__("threading").Lock()
+
+        def mock_fetch(slug, force_refresh=False):
+            with lock:
+                active.append(slug)
+                max_concurrent[0] = max(max_concurrent[0], len(active))
+            time.sleep(0.05)  # simulate network latency
+            with lock:
+                active.remove(slug)
+            return [f"model_{slug}"]
+
+        slugs = [f"prov_{i}" for i in range(6)]
+
+        with patch("hermes_cli.models._load_provider_models_cache", return_value={}), \
+             patch("hermes_cli.models._credential_fingerprint", return_value="fp"), \
+             patch("hermes_cli.models.cached_provider_model_ids", side_effect=mock_fetch), \
+             patch("hermes_cli.models.update_provider_cache_entry"):
+            _prefetch_provider_models_parallel(slugs)
+
+        assert max_concurrent[0] > 1, "fetches were serial, not parallel"
+
+    def test_swallows_exceptions(self):
+        """A failing provider fetch doesn't raise — best-effort."""
+        from hermes_cli.model_switch import _prefetch_provider_models_parallel
+
+        def mock_fetch(slug, force_refresh=False):
+            raise ConnectionError("simulated network failure")
+
+        with patch("hermes_cli.models._load_provider_models_cache", return_value={}), \
+             patch("hermes_cli.models._credential_fingerprint", return_value="fp"), \
+             patch("hermes_cli.models.cached_provider_model_ids", side_effect=mock_fetch), \
+             patch("hermes_cli.models.update_provider_cache_entry"):
+            # Should not raise
+            _prefetch_provider_models_parallel(["failing_prov"])
+
+    def test_empty_list_is_noop(self):
+        """Empty provider list does nothing."""
+        from hermes_cli.model_switch import _prefetch_provider_models_parallel
+
+        with patch("hermes_cli.models.cached_provider_model_ids") as fetch:
+            _prefetch_provider_models_parallel([])
+        fetch.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Integration: prefetch is called from list_authenticated_providers
+# ---------------------------------------------------------------------------
+
+class TestPrefetchIntegration:
+    """Verify ``list_authenticated_providers`` triggers parallel prefetch."""
+
+    def test_prefetch_called_with_more_than_3_providers(self):
+        """When >3 providers are authed, parallel prefetch is invoked."""
+        from hermes_cli import model_switch
+
+        slugs = [f"prov_{i}" for i in range(5)]
+        captured_slugs = []
+
+        def mock_collect(data, curated, excluded):
+            return slugs
+
+        with patch.object(model_switch, "_collect_authed_provider_slugs", side_effect=mock_collect), \
+             patch.object(model_switch, "_prefetch_provider_models_parallel") as prefetch:
+            try:
+                model_switch.list_authenticated_providers()
+            except Exception:
+                pass  # we only care about the prefetch call
+            captured_slugs = prefetch.call_args[0][0] if prefetch.called else []
+
+        assert prefetch.called
+        assert captured_slugs == slugs
+
+    def test_prefetch_skipped_with_3_or_fewer_providers(self):
+        """When ≤3 providers are authed, parallel prefetch is skipped."""
+        from hermes_cli import model_switch
+
+        slugs = ["prov_a", "prov_b"]
+
+        def mock_collect(data, curated, excluded):
+            return slugs
+
+        with patch.object(model_switch, "_collect_authed_provider_slugs", side_effect=mock_collect), \
+             patch.object(model_switch, "_prefetch_provider_models_parallel") as prefetch:
+            try:
+                model_switch.list_authenticated_providers()
+            except Exception:
+                pass
+
+        prefetch.assert_not_called()
+
+    def test_prefetch_skipped_on_refresh(self):
+        """When refresh=True, prefetch is skipped (serial path force-refreshes)."""
+        from hermes_cli import model_switch
+
+        with patch.object(model_switch, "_collect_authed_provider_slugs") as collect, \
+             patch.object(model_switch, "_prefetch_provider_models_parallel") as prefetch:
+            try:
+                model_switch.list_authenticated_providers(refresh=True)
+            except Exception:
+                pass
+
+        collect.assert_not_called()
+        prefetch.assert_not_called()
diff --git a/tests/run_agent/test_token_persistence_non_cli.py b/tests/run_agent/test_token_persistence_non_cli.py
index dd82395d237b..7479c2af85ca 100644
--- a/tests/run_agent/test_token_persistence_non_cli.py
+++ b/tests/run_agent/test_token_persistence_non_cli.py
@@ -80,9 +80,49 @@ def fake_session_search(**kwargs):
     monkeypatch.setitem(sys.modules, "tools.session_search_tool", session_search_mod)
 
     agent = _make_agent(None, platform="acp")
-    result = json.loads(agent._invoke_tool("session_search", {"query": "Hermes"}, "task-id"))
+    result = json.loads(agent._invoke_tool(
+        "session_search",
+        {"query": "Hermes", "detail": "full"},
+        "task-id",
+    ))
 
     assert result["success"] is True
     assert captured["db"] is sentinel_db
     assert captured["query"] == "Hermes"
+    assert captured["detail"] == "full"
     assert agent._session_db is sentinel_db
+
+
+def test_sequential_session_search_forwards_detail(monkeypatch):
+    session_db = MagicMock()
+    captured = {}
+
+    session_search_mod = ModuleType("tools.session_search_tool")
+
+    def fake_session_search(**kwargs):
+        captured.update(kwargs)
+        return json.dumps({"success": True, "results": []})
+
+    session_search_mod.session_search = fake_session_search
+    monkeypatch.setitem(sys.modules, "tools.session_search_tool", session_search_mod)
+
+    agent = _make_agent(session_db, platform="acp")
+    tool_call = SimpleNamespace(
+        id="search-1",
+        function=SimpleNamespace(
+            name="session_search",
+            arguments=json.dumps({"query": "Hermes", "detail": "full"}),
+        ),
+    )
+    assistant_message = SimpleNamespace(tool_calls=[tool_call])
+    messages = []
+
+    agent._execute_tool_calls_sequential(
+        assistant_message,
+        messages,
+        "task-id",
+    )
+
+    assert captured["db"] is session_db
+    assert captured["query"] == "Hermes"
+    assert captured["detail"] == "full"
diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py
index 1b2f688ae76d..2d720669784e 100644
--- a/tests/test_hermes_state.py
+++ b/tests/test_hermes_state.py
@@ -473,6 +473,7 @@ def connect_without_trigram(*args, **kwargs):
             results = db.search_messages("大别山")
             assert len(results) == 1
             # Note: search_messages strips 'content' from results; use 'snippet'.
+            assert "content" not in results[0]
             assert "大别山" in results[0]["snippet"]
         finally:
             db.close()
@@ -739,6 +740,8 @@ def test_search_finds_content(self, db):
         # At least one result should mention docker
         snippets = [r.get("snippet", "") for r in results]
         assert any("docker" in s.lower() or "Docker" in s for s in snippets)
+        # Results never carry full content; snippet + metadata only.
+        assert all("content" not in r for r in results)
 
 
 
diff --git a/tests/tools/test_session_search.py b/tests/tools/test_session_search.py
index af36d90dc249..fb61db973f72 100644
--- a/tests/tools/test_session_search.py
+++ b/tests/tools/test_session_search.py
@@ -1,12 +1,14 @@
 """Tests for the single-shape session_search tool.
 
-Three calling shapes:
-  1. DISCOVERY — pass query → FTS5 + anchored window + bookends per hit
+Four calling shapes:
+  1. DISCOVERY — pass query → FTS5 + adaptive/full hydration
   2. SCROLL    — pass session_id + around_message_id → just the window
-  3. BROWSE    — no args → recent sessions chronologically
+  3. READ      — pass session_id → whole or head/tail-truncated session
+  4. BROWSE    — no args → recent sessions chronologically
 
 All run zero LLM calls.
 """
+import inspect
 import json
 import time
 
@@ -72,6 +74,8 @@ def test_schema_params_cover_every_shape(self):
         assert "query" in params
         assert "limit" in params
         assert params["sort"]["enum"] == ["newest", "oldest"]
+        assert params["detail"]["enum"] == ["adaptive", "full"]
+        assert params["detail"]["default"] == "adaptive"
         # Scroll shape
         assert "session_id" in params
         assert "around_message_id" in params
@@ -81,6 +85,22 @@ def test_schema_params_cover_every_shape(self):
         # Mode is inferred from which args are set — no explicit mode param
         assert "mode" not in params
 
+    def test_detail_parameter_is_appended_for_positional_compatibility(self):
+        parameters = list(inspect.signature(session_search).parameters)
+        historical_prefix = [
+            "query",
+            "role_filter",
+            "limit",
+            "db",
+            "current_session_id",
+            "session_id",
+            "around_message_id",
+            "window",
+            "sort",
+            "profile",
+        ]
+        assert parameters == [*historical_prefix, "detail"]
+
 
 class TestFormatTimestamp:
     def test_formats_unix_and_passes_through_the_rest(self):
@@ -176,17 +196,22 @@ def search_spy(*args, **kwargs):
         assert "context" not in requested_fields
         assert len(result["results"]) == 1
         hit = result["results"][0]
+        assert hit["detail"] == "full"
         assert "bookend_start" in hit
         assert hit["messages"]
         assert "bookend_end" in hit
 
-    def test_discovery_result_has_bookends_and_window(self, db):
+    def test_full_detail_returns_bookends_and_window_for_every_hit(self, db):
         _seed_modpack_sessions(db)
-        result = json.loads(session_search(query="modpack", limit=3, db=db))
+        result = json.loads(session_search(
+            query="modpack", limit=3, detail="full", db=db
+        ))
         assert result["success"] is True
         assert result["mode"] == "discover"
+        assert result["detail"] == "full"
         assert result["count"] >= 1
         for hit in result["results"]:
+            assert hit["detail"] == "full"
             assert "bookend_start" in hit
             assert "messages" in hit
             assert "bookend_end" in hit
@@ -195,6 +220,72 @@ def test_discovery_result_has_bookends_and_window(self, db):
             assert "messages_before" in hit
             assert "messages_after" in hit
 
+    def test_default_discovery_keeps_top_full_and_compacts_lower_hits(self, db):
+        _seed_modpack_sessions(db)
+
+        result = json.loads(session_search(query="modpack", limit=3, db=db))
+
+        assert result["success"] is True
+        assert result["detail"] == "adaptive"
+        assert len(result["results"]) == 3
+
+        top, *lower = result["results"]
+        assert top["detail"] == "full"
+        assert "bookend_start" in top
+        assert len(top["messages"]) > 1
+        assert "bookend_end" in top
+
+        for hit in lower:
+            assert hit["detail"] == "compact"
+            assert hit["bookend_start"] == []
+            assert len(hit["messages"]) == 1
+            assert hit["messages"][0]["id"] == hit["match_message_id"]
+            assert hit["messages"][0]["anchor"] is True
+            assert hit["bookend_end"] == []
+
+    def test_adaptive_detail_preserves_ranking_and_reduces_payload(self, db):
+        now = int(time.time())
+        for session_index in range(3):
+            session_id = f"payload_{session_index}"
+            db.create_session(session_id, source="cli")
+            db._conn.execute(
+                "UPDATE sessions SET started_at = ? WHERE id = ?",
+                (now - session_index, session_id),
+            )
+            for message_index in range(8):
+                db.append_message(
+                    session_id,
+                    role="user" if message_index % 2 == 0 else "assistant",
+                    content=f"opening {session_index}-{message_index} " + "o" * 2500,
+                )
+            db.append_message(
+                session_id,
+                role="user",
+                content=f"payloadneedle anchor {session_index} " + "a" * 3500,
+            )
+            for message_index in range(8):
+                db.append_message(
+                    session_id,
+                    role="assistant" if message_index % 2 == 0 else "user",
+                    content=f"closing {session_index}-{message_index} " + "c" * 2500,
+                )
+        db._conn.commit()
+
+        adaptive_json = session_search(query="payloadneedle", limit=3, db=db)
+        full_json = session_search(
+            query="payloadneedle", limit=3, detail="full", db=db
+        )
+        adaptive = json.loads(adaptive_json)
+        full = json.loads(full_json)
+
+        assert [r["session_id"] for r in adaptive["results"]] == [
+            r["session_id"] for r in full["results"]
+        ]
+        assert [r["match_message_id"] for r in adaptive["results"]] == [
+            r["match_message_id"] for r in full["results"]
+        ]
+        assert len(adaptive_json.encode("utf-8")) < len(full_json.encode("utf-8")) * 0.6
+
 
     def test_current_session_filtered_out(self, db):
         _seed_modpack_sessions(db)
diff --git a/tools/session_search_tool.py b/tools/session_search_tool.py
index 759ed1c44f6d..c5752f5ca4ad 100644
--- a/tools/session_search_tool.py
+++ b/tools/session_search_tool.py
@@ -2,23 +2,27 @@
 """
 Session Search Tool - Long-Term Conversation Recall
 
-Single-shape tool with three calling modes (inferred from args, no explicit
+Single-shape tool with four calling modes (inferred from args, no explicit
 mode parameter):
 
-  1. DISCOVERY — pass ``query``. Runs FTS5, dedupes hits by session lineage,
-     returns top N sessions each with: snippet, ±5 message window around the
-     match, plus bookend_start (first 3 user+assistant msgs of session) and
-     bookend_end (last 3). Zero LLM cost.
+  1. DISCOVERY — pass ``query``. Runs FTS5 and dedupes hits by session lineage.
+     Adaptive detail (the default) fully hydrates the top result with a ±5
+     message window and bookends, while lower-ranked results keep the exact
+     anchor message plus metadata. Pass ``detail="full"`` to fully hydrate
+     every result. Zero LLM cost.
 
   2. SCROLL — pass ``session_id`` + ``around_message_id``. Returns a window
      of ±window messages centered on the anchor, no FTS5, no bookends. To
      scroll forward / backward, re-anchor on the last / first message id of
      the returned window.
 
-  3. BROWSE — no args. Returns recent sessions chronologically (titles,
+  3. READ — pass ``session_id`` without an anchor. Returns the whole session,
+     or a bounded head/tail view for large sessions.
+
+  4. BROWSE — no args. Returns recent sessions chronologically (titles,
      previews, timestamps).
 
-All three modes operate on the SQLite session DB via the FTS5 index and
+All four modes operate on the SQLite session DB via the FTS5 index and
 the get_anchored_view / get_messages_around primitives in hermes_state.
 No LLM calls anywhere — every shape returns actual messages from the DB.
 
@@ -740,6 +744,7 @@ def _title_match_result(
         "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:
@@ -753,10 +758,11 @@ def _discover(
     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 + anchored window + bookends per hit. Single call."""
+    """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)
@@ -788,6 +794,7 @@ def _discover(
             "success": True,
             "mode": "discover",
             "query": query,
+            "detail": detail,
             "results": [],
             "count": 0,
             "message": "No matching sessions found.",
@@ -864,6 +871,11 @@ def _discover(
         except Exception:
             session_meta = {}
 
+        result_detail = "full" if detail == "full" or not results else "compact"
+        window_messages = view.get("window") or []
+        if result_detail == "compact":
+            window_messages = [m for m in window_messages if m.get("id") == msg_id]
+
         entry = {
             "session_id": hit_sid,
             "when": _format_timestamp(
@@ -875,19 +887,31 @@ def _discover(
             "matched_role": match_info.get("role"),
             "match_message_id": msg_id,
             "snippet": match_info.get("snippet") or "",
-            "bookend_start": [
-                _shape_message(m, max_content_len=1200)
-                for m in (view.get("bookend_start") or [])
-                if not _is_compaction_summary(m.get("content", ""))
-            ],
-            "messages": [_shape_message(m, anchor_id=msg_id, max_content_len=4000) for m in (view.get("window") or [])],
-            "bookend_end": [
-                _shape_message(m, max_content_len=1200)
-                for m in (view.get("bookend_end") or [])
-                if not _is_compaction_summary(m.get("content", ""))
+            "bookend_start": (
+                [
+                    _shape_message(m, max_content_len=1200)
+                    for m in (view.get("bookend_start") or [])
+                    if not _is_compaction_summary(m.get("content", ""))
+                ]
+                if result_detail == "full"
+                else []
+            ),
+            "messages": [
+                _shape_message(m, anchor_id=msg_id, max_content_len=4000)
+                for m in window_messages
             ],
+            "bookend_end": (
+                [
+                    _shape_message(m, max_content_len=1200)
+                    for m in (view.get("bookend_end") or [])
+                    if not _is_compaction_summary(m.get("content", ""))
+                ]
+                if result_detail == "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
@@ -900,6 +924,7 @@ def _discover(
         "success": True,
         "mode": "discover",
         "query": query,
+        "detail": detail,
         "results": results,
         "count": len(results),
         "sessions_searched": len(seen_sessions),
@@ -922,12 +947,14 @@ def _session_search_impl(
     sort: str = None,
     # Cross-profile (any shape)
     profile: str = None,
+    # Discovery result shaping (appended to preserve positional compatibility)
+    detail: str = "adaptive",
     *,
     _owned_dbs: Optional[List[Any]] = None,
 ) -> str:
     """Single-shape tool. Mode inferred from which args are set.
 
-    Discovery: pass ``query``.
+    Discovery: pass ``query``; ``detail="full"`` hydrates every result.
     Scroll:    pass ``session_id`` + ``around_message_id``.
     Read:      pass ``session_id`` (no anchor) — dumps the whole session.
     Browse:    pass nothing.
@@ -1017,12 +1044,19 @@ def _session_search_impl(
         if candidate in ("newest", "oldest"):
             sort_norm = candidate
 
+    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,
     )
@@ -1042,6 +1076,8 @@ def session_search(
     sort: str = None,
     # Cross-profile (any shape)
     profile: str = None,
+    # Discovery result shaping (appended to preserve positional compatibility)
+    detail: str = "adaptive",
 ) -> str:
     """Run session search and close databases opened by this invocation."""
     owned_dbs: List[Any] = []
@@ -1069,6 +1105,7 @@ def session_search(
             window=window,
             sort=sort,
             profile=profile,
+            detail=detail,
             _owned_dbs=owned_dbs,
         )
     finally:
@@ -1108,19 +1145,21 @@ def check_session_search_requirements() -> bool:
         "FOUR CALLING SHAPES\n\n"
         "  1) DISCOVERY — pass `query`:\n"
         "     session_search(query=\"auth refactor\", limit=3)\n"
-        "     Runs FTS5, dedupes hits by session lineage, returns the top N sessions. "
-        "Each result carries:\n"
+        "     Runs FTS5, dedupes hits by session lineage, and returns the top N "
+        "sessions. Adaptive detail is the default: the top-ranked result carries "
+        "full context, while lower-ranked results stay compact. Pass `detail=\"full\"` "
+        "to fully hydrate every result. Every result carries:\n"
         "       - session_id, title, when, source\n"
         "       - snippet: FTS5-highlighted match excerpt\n"
-        "       - bookend_start: first 3 user+assistant messages of the session "
-        "(the goal / kickoff)\n"
-        "       - messages: ±5 messages around the FTS5 match, with the anchor message "
-        "flagged (the hit in context)\n"
-        "       - bookend_end: last 3 user+assistant messages of the session "
-        "(the resolution / decisions)\n"
+        "       - detail: `full` or `compact`\n"
+        "       - bookend_start/bookend_end: the first/last 3 user+assistant messages "
+        "for full results; empty lists for compact results\n"
+        "       - messages: ±5 messages around the FTS5 match for full results; only "
+        "the flagged anchor message for compact results\n"
         "       - match_message_id, messages_before, messages_after\n"
-        "     Bookends + window together let you reconstruct goal → match → resolution "
-        "without paying for the whole transcript.\n\n"
+        "     The top result's bookends + window let you reconstruct goal → match → "
+        "resolution immediately. Scroll a compact result when another session looks "
+        "more promising.\n\n"
         "  2) SCROLL — pass `session_id` + `around_message_id`:\n"
         "     session_search(session_id=\"...\", around_message_id=12345, window=10)\n"
         "     Returns a window of ±`window` messages centered on the anchor. No FTS5, "
@@ -1197,6 +1236,17 @@ def check_session_search_requirements() -> bool:
                     "and browse shapes."
                 ),
             },
+            "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": (
@@ -1261,6 +1311,7 @@ def check_session_search_requirements() -> bool:
         around_message_id=args.get("around_message_id"),
         window=args.get("window", 5),
         sort=args.get("sort"),
+        detail=args.get("detail", "adaptive"),
         profile=args.get("profile"),
         db=kw.get("db"),
         current_session_id=kw.get("current_session_id"),
diff --git a/website/docs/user-guide/sessions.md b/website/docs/user-guide/sessions.md
index 3f02dd2be20b..7d36dfa256e1 100644
--- a/website/docs/user-guide/sessions.md
+++ b/website/docs/user-guide/sessions.md
@@ -612,9 +612,9 @@ routing is the only thing the repair changes. Back up first
 
 ## Session Search Tool
 
-The agent has a built-in `session_search` tool that performs full-text search across all past conversations using SQLite's FTS5 engine — and lets the agent scroll through any session it finds. No LLM calls, no summarization, no truncation. Every shape returns actual messages from the DB.
+The agent has a built-in `session_search` tool that performs full-text search across all past conversations using SQLite's FTS5 engine — and lets the agent scroll through any session it finds. It makes no LLM calls and returns views of actual messages from the DB rather than generating summaries.
 
-### Three calling shapes
+### Four calling shapes
 
 The tool infers what you want from which arguments you set. There's no `mode` parameter.
 
@@ -624,16 +624,18 @@ The tool infers what you want from which arguments you set. There's no `mode` pa
 session_search(query="auth refactor", limit=3)
 ```
 
-Runs FTS5, dedupes hits by session lineage, returns the top N sessions. Each result carries:
+Runs FTS5, dedupes hits by session lineage, and returns the top N sessions. Discovery uses adaptive detail by default: the highest-ranked result includes its full context window and bookends, while lower-ranked results stay compact. Pass `detail="full"` to fully hydrate every result.
+
+Each result carries:
 
 - `session_id`, `title`, `when`, `source`
 - `snippet` — FTS5-highlighted match excerpt
-- `bookend_start` — first 3 user+assistant messages of the session (the goal/kickoff)
-- `messages` — ±5 messages around the FTS5 match, with the anchor message flagged (the hit in context)
-- `bookend_end` — last 3 user+assistant messages of the session (the resolution/decisions)
+- `detail` — `full` or `compact`
+- `bookend_start` / `bookend_end` — first/last 3 user+assistant messages for full results; empty lists for compact results
+- `messages` — ±5 messages around the FTS5 match for full results; only the flagged anchor message for compact results
 - `match_message_id`, `messages_before`, `messages_after`
 
-Bookends + window together reconstruct goal → match → resolution without paying for the whole transcript. Typical wall time: 15–50ms on a real session DB.
+The top result reconstructs goal → match → resolution immediately. If another compact result looks more promising, use its session and message IDs with the scroll shape. Typical wall time is tens of milliseconds on a real session DB.
 
 **2. Scroll — pass `session_id` + `around_message_id`:**
 
@@ -650,7 +652,15 @@ Returns a window of ±`window` messages centered on the anchor. No FTS5, no book
 
 Typical wall time: 1–2ms per scroll call.
 
-**3. Browse — no args:**
+**3. Read — pass `session_id` without an anchor:**
+
+```python
+session_search(session_id="20260510_174648_805cc2")
+```
+
+Returns the whole session, or a bounded head/tail view for large sessions. This shape is also used to resolve an `@session:<profile>/<id>` link.
+
+**4. Browse — no args:**
 
 ```python
 session_search()
@@ -670,6 +680,7 @@ The keyword mode supports standard FTS5 query syntax:
 ### Optional parameters
 
 - `sort` — `newest` or `oldest`, on top of FTS5 ranking. Omit for relevance-only ordering (the default; suitable for exploratory recall). Use `newest` for "where did we leave X" questions, `oldest` for "how did X start" questions.
+- `detail` — `adaptive` (default) fully hydrates only the top discovery result; `full` hydrates every discovery result.
 - `role_filter` — 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.
 
 ### When It's Used