"""Agent middleware helpers — retry + transit path normalisation.

Exports:
  - ``should_retry_transient_model_error``: classifier for retry middleware
  - ``build_model_retry_middleware``: factory returning a ModelRetryMiddleware
  - ``set_retry_notifier`` / ``clear_retry_notifier``: bridge so the UI can
    show retry updates without a hard coupling to the middleware
"""

from __future__ import annotations

import asyncio
import contextvars
import time
from collections.abc import Awaitable, Callable, Mapping
from pathlib import Path
from typing import Any

from langchain.agents.middleware import (
    AgentMiddleware,
    AgentState,
    ModelRetryMiddleware,
)
from langchain.agents.middleware._retry import calculate_delay, should_retry_exception
from langchain.agents.middleware.types import ModelRequest, ModelResponse
from langchain_core.messages import ToolMessage

from synapse.runtime.pathing import rewrite_tool_args_paths

# ---------------------------------------------------------------------------
#  Transient error markers (text-based, no status code)
# ---------------------------------------------------------------------------
_TRANSIENT_MODEL_ERROR_MARKERS = (
    "empty model output",
    "overloaded",
    "temporarily unavailable",
    "service unavailable",
    "upstream timeout",
    "upstream request timeout",
    "rate limit",
    "rate_limit",
)

# ---------------------------------------------------------------------------
#  Transient *server* errors: 5xx status codes whose body indicates a
#  recoverable infra hiccup (not a hard client / auth error).
# ---------------------------------------------------------------------------
_RETRYABLE_5XX_MARKERS = (
    "auth_unavailable",
    "overloaded",
    "temporarily unavailable",
    "service unavailable",
    "upstream timeout",
    "upstream request timeout",
)

_RETRYABLE_5XX_STATUSES = frozenset({429, 502, 503, 504})


# ---------------------------------------------------------------------------
#  Module-level retry notifier (set by stream / TUI before each turn)
# ---------------------------------------------------------------------------
_retry_notifier: Callable[[int, float, str], None] | None = None


def set_retry_notifier(fn: Callable[[int, float, str], None] | None) -> None:
    """Install a callback invoked before each retry delay.

    ``fn(attempt, delay, reason)`` where *attempt* is 1-indexed,
    *delay* is seconds about to be slept, and *reason* summarises
    the exception that triggered the retry.
    ``None`` clears the notifier.
    """
    global _retry_notifier
    _retry_notifier = fn


def clear_retry_notifier() -> None:
    """Remove any installed retry notifier."""
    set_retry_notifier(None)


def _model_error_text(exc: Exception) -> str:
    """Collect provider error text, including nested SSE error bodies."""

    parts = [str(exc)]
    body = getattr(exc, "body", None)

    def _collect(value: Any) -> None:
        if isinstance(value, Mapping):
            for nested in value.values():
                _collect(nested)
        elif isinstance(value, (list, tuple)):
            for nested in value:
                _collect(nested)
        elif value is not None:
            parts.append(str(value))

    _collect(body)
    return " ".join(parts).lower()


def should_retry_transient_model_error(exc: Exception) -> bool:
    """Return ``True`` when *exc* is a recoverable transient model error.

    Retried:
      - Provider errors **without** an HTTP status code whose error text
        matches a known transient marker (e.g. ``overloaded``).
      - 5xx server errors (502 / 503 / 504) when the body text contains an
        explicitly-recognised infrastructure marker such as
        ``auth_unavailable`` — these are short-lived auth-infra hiccups that
        recover within seconds.

    **Not** retried:
      - Any error carrying a 4xx status code (401, 429, …).
      - 5xx errors whose body does not match a known marker (non-transient).
    """

    status_code = getattr(exc, "status_code", None)
    if isinstance(status_code, int):
        if status_code in _RETRYABLE_5XX_STATUSES:
            text = _model_error_text(exc)
            if any(marker in text for marker in _RETRYABLE_5XX_MARKERS):
                return True
        return False
    text = _model_error_text(exc)
    return any(marker in text for marker in _TRANSIENT_MODEL_ERROR_MARKERS)


def _format_retry_reason(exc: Exception) -> str:
    """One-line summary of *exc* suitable for the status bar."""
    status_code = getattr(exc, "status_code", None)
    msg = str(exc)
    body = getattr(exc, "body", None)
    if isinstance(body, Mapping):
        err = body.get("error") or {}
        if isinstance(err, Mapping):
            inner = err.get("message") or ""
            if inner:
                msg = str(inner)
    short = msg.split("\n")[0].strip()
    if len(short) > 120:
        short = short[:117] + "..."
    if isinstance(status_code, int):
        return f"[{status_code}] {short}"
    return short


class NotifyingModelRetryMiddleware(ModelRetryMiddleware):
    """``ModelRetryMiddleware`` subclass that fires a notifier on each retry."""

    def _notify_retry(self, attempt: int, delay: float, exc: Exception) -> None:
        if _retry_notifier is not None:
            try:
                reason = _format_retry_reason(exc)
                _retry_notifier(attempt, delay, reason)
            except Exception:  # noqa: BLE001
                pass

    # -- sync ----------------------------------------------------------------
    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        for attempt in range(self.max_retries + 1):
            try:
                return handler(request)
            except Exception as exc:
                attempts_made = attempt + 1
                if not should_retry_exception(exc, self.retry_on):
                    return self._handle_failure(exc, attempts_made)
                if attempt < self.max_retries:
                    delay = calculate_delay(
                        attempt,
                        backoff_factor=self.backoff_factor,
                        initial_delay=self.initial_delay,
                        max_delay=self.max_delay,
                        jitter=self.jitter,
                    )
                    self._notify_retry(attempts_made, delay, exc)
                    if delay > 0:
                        time.sleep(delay)
                else:
                    return self._handle_failure(exc, attempts_made)
        msg = "Unexpected: retry loop completed without returning"
        raise RuntimeError(msg)

    # -- async ---------------------------------------------------------------
    async def awrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
    ) -> ModelResponse:
        for attempt in range(self.max_retries + 1):
            try:
                return await handler(request)
            except Exception as exc:
                attempts_made = attempt + 1
                if not should_retry_exception(exc, self.retry_on):
                    return self._handle_failure(exc, attempts_made)
                if attempt < self.max_retries:
                    delay = calculate_delay(
                        attempt,
                        backoff_factor=self.backoff_factor,
                        initial_delay=self.initial_delay,
                        max_delay=self.max_delay,
                        jitter=self.jitter,
                    )
                    self._notify_retry(attempts_made, delay, exc)
                    if delay > 0:
                        await asyncio.sleep(delay)
                else:
                    return self._handle_failure(exc, attempts_made)
        msg = "Unexpected: retry loop completed without returning"
        raise RuntimeError(msg)


def build_model_retry_middleware() -> NotifyingModelRetryMiddleware:
    """Retry recoverable model failures (stream + HTTP 5xx) with backoff."""
    return NotifyingModelRetryMiddleware(
        max_retries=999,
        retry_on=should_retry_transient_model_error,
        on_failure="error",
        initial_delay=1.0,
        backoff_factor=2.0,
        max_delay=8.0,
        jitter=True,
    )


def _dual_wrap_model_call(*, name: str, apply):
    """Build model-call middleware with both sync and async hooks.

    ``apply(request) -> request`` mutates/overrides the request; the handler is
    always invoked (sync or async). Required for ``astream`` / ``ainvoke``.
    """

    def wrap_model_call(self, request, handler):  # noqa: ANN001, ARG001
        return handler(apply(request))

    async def awrap_model_call(self, request, handler):  # noqa: ANN001, ARG001
        return await handler(apply(request))

    return type(
        name,
        (AgentMiddleware,),
        {
            "state_schema": AgentState,
            "tools": [],
            "wrap_model_call": wrap_model_call,
            "awrap_model_call": awrap_model_call,
        },
    )()


def _dual_wrap_tool_call(*, name: str, apply):
    """Build tool-call middleware with both sync and async hooks."""

    def wrap_tool_call(self, request, handler):  # noqa: ANN001, ARG001
        return handler(apply(request))

    async def awrap_tool_call(self, request, handler):  # noqa: ANN001, ARG001
        return await handler(apply(request))

    return type(
        name,
        (AgentMiddleware,),
        {
            "state_schema": AgentState,
            "tools": [],
            "wrap_tool_call": wrap_tool_call,
            "awrap_tool_call": awrap_tool_call,
        },
    )()


def build_task_namespace_middleware():
    """Give each concurrent ``task`` invocation a distinct subgraph namespace."""

    def _call_id(request: Any) -> str:
        call = request.tool_call
        if isinstance(call, dict):
            return str(call.get("id") or "")
        return str(getattr(call, "id", None) or "")

    def _name(request: Any) -> str:
        call = request.tool_call
        if isinstance(call, dict):
            return str(call.get("name") or "")
        return str(getattr(call, "name", None) or "")

    def _config(request: Any, call_id: str) -> dict[str, Any]:
        config = dict(request.runtime.config or {})
        configurable = dict(config.get("configurable") or {})
        parent_ns = str(configurable.get("checkpoint_ns") or "")
        segment = f"task_call:{call_id}"
        configurable["checkpoint_ns"] = f"{parent_ns}|{segment}" if parent_ns else segment
        config["configurable"] = configurable
        return config

    def wrap_tool_call(self, request, handler):  # noqa: ANN001, ARG001
        call_id = _call_id(request)
        if _name(request) != "task" or not call_id:
            return handler(request)
        from langchain_core.runnables.config import set_config_context

        with set_config_context(_config(request, call_id)) as ctx:
            return ctx.run(handler, request)

    async def awrap_tool_call(self, request, handler):  # noqa: ANN001, ARG001
        call_id = _call_id(request)
        if _name(request) != "task" or not call_id:
            return await handler(request)
        from langchain_core.runnables.config import set_config_context

        with set_config_context(_config(request, call_id)) as ctx:
            task = ctx.run(asyncio.create_task, handler(request))
            return await task

    return type(
        "scope_task_subgraphs",
        (AgentMiddleware,),
        {
            "state_schema": AgentState,
            "tools": [],
            "wrap_tool_call": wrap_tool_call,
            "awrap_tool_call": awrap_tool_call,
        },
    )()


# Required model-facing field: short purpose shown in the timeline UI.
TOOL_INTENT_KEY = "intent"
TOOL_INTENT_DESCRIPTION = (
    "Required. One short sentence describing WHY this tool is being called "
    "(user-facing intent for the timeline UI). Prefer Chinese. "
    "Example: 'inspect pytest config' / 'locate login failure'. "
    "Do not dump raw args or only restate the tool name."
)


def build_path_normalize_middleware(workspace: Path):
    """Rewrite host/Windows paths in tool args to virtual ``/`` paths."""

    root = Path(workspace).resolve()

    def _apply(request):  # type: ignore[no-untyped-def]
        tool_call = request.tool_call
        # tool_call may be dict-like
        if isinstance(tool_call, dict):
            args = dict(tool_call.get("args") or {})
            new_args = rewrite_tool_args_paths(args, root)
            if new_args != args:
                new_call = {**tool_call, "args": new_args}
                return request.override(tool_call=new_call)
            return request
        args = dict(getattr(tool_call, "args", None) or {})
        new_args = rewrite_tool_args_paths(args, root)
        if new_args != args:
            # Best-effort for object-style tool_call
            try:
                new_call = {
                    "name": getattr(tool_call, "name", None),
                    "args": new_args,
                    "id": getattr(tool_call, "id", None),
                    "type": getattr(tool_call, "type", "tool_call"),
                }
                return request.override(tool_call=new_call)
            except Exception:  # noqa: BLE001
                return request
        return request

    return _dual_wrap_tool_call(name="normalize_virtual_paths", apply=_apply)


def build_tool_error_recovery_middleware():
    """Return tool failures to the model instead of terminating the agent graph."""

    def _error_message(request, exc: Exception) -> ToolMessage:  # type: ignore[no-untyped-def]
        tool_call = request.tool_call
        if isinstance(tool_call, dict):
            name = str(tool_call.get("name") or "tool")
            call_id = str(tool_call.get("id") or "unknown")
        else:
            name = str(getattr(tool_call, "name", None) or "tool")
            call_id = str(getattr(tool_call, "id", None) or "unknown")
        return ToolMessage(
            content=(
                f"Error: {name} failed ({type(exc).__name__}): {exc}\n"
                "The tool call failed. Continue the task by correcting the arguments "
                "or choosing another safe tool."
            ),
            tool_call_id=call_id,
            name=name,
            status="error",
        )

    def wrap_tool_call(self, request, handler):  # noqa: ANN001, ARG001
        try:
            return handler(request)
        except Exception as exc:  # noqa: BLE001
            return _error_message(request, exc)

    async def awrap_tool_call(self, request, handler):  # noqa: ANN001, ARG001
        try:
            return await handler(request)
        except Exception as exc:  # noqa: BLE001
            return _error_message(request, exc)

    return type(
        "recover_tool_errors",
        (AgentMiddleware,),
        {
            "state_schema": AgentState,
            "tools": [],
            "wrap_tool_call": wrap_tool_call,
            "awrap_tool_call": awrap_tool_call,
        },
    )()


def _tool_name(tool: Any) -> str:
    name = getattr(tool, "name", None)
    if name:
        return str(name)
    return str(getattr(tool, "__name__", tool))


def build_tool_exclusion_middleware(excluded: set[str] | frozenset[str] | list[str]):
    """Hide tools from the model request (LocalShell-safe alternative to permissions).

    deepagents ``FilesystemPermission`` cannot be combined with backends that
    implement command execution. Use this middleware for product isolation.
    """
    blocked = frozenset(str(x) for x in excluded if x)

    def _apply(request):  # type: ignore[no-untyped-def]
        if not blocked:
            return request
        tools = getattr(request, "tools", None) or []
        filtered = [t for t in tools if _tool_name(t) not in blocked]
        if len(filtered) != len(tools):
            return request.override(tools=filtered)
        return request

    return _dual_wrap_model_call(name="exclude_tools", apply=_apply)


def _strip_intent_from_tool_call(tool_call: Any) -> Any:
    """Remove intent from args before the real tool schema validates."""
    if isinstance(tool_call, dict):
        args = dict(tool_call.get("args") or {})
        if TOOL_INTENT_KEY not in args:
            return tool_call
        args.pop(TOOL_INTENT_KEY, None)
        return {**tool_call, "args": args}

    args = dict(getattr(tool_call, "args", None) or {})
    if TOOL_INTENT_KEY not in args:
        return tool_call
    args.pop(TOOL_INTENT_KEY, None)
    try:
        return {
            "name": getattr(tool_call, "name", None),
            "args": args,
            "id": getattr(tool_call, "id", None),
            "type": getattr(tool_call, "type", "tool_call"),
        }
    except Exception:  # noqa: BLE001
        return tool_call


def _field_definitions_with_intent(schema: Any) -> dict[str, Any] | None:
    """Build create_model field map with required intent first."""
    from pydantic import Field
    from pydantic_core import PydanticUndefined

    fields = getattr(schema, "model_fields", None)
    if not fields:
        return None
    if TOOL_INTENT_KEY in fields:
        return None

    defs: dict[str, Any] = {
        TOOL_INTENT_KEY: (
            str,
            Field(..., description=TOOL_INTENT_DESCRIPTION),
        )
    }
    for name, finfo in fields.items():
        ann = finfo.annotation
        desc = finfo.description or name
        if finfo.is_required():
            defs[name] = (ann, Field(..., description=desc))
            continue
        default = finfo.default
        default_factory = getattr(finfo, "default_factory", None)
        if default_factory is not None and default is PydanticUndefined:
            defs[name] = (ann, Field(default_factory=default_factory, description=desc))
        elif default is PydanticUndefined:
            defs[name] = (ann, Field(default=None, description=desc))
        else:
            defs[name] = (ann, Field(default=default, description=desc))
    return defs


def add_intent_to_tool(tool: Any, *, cache: dict[int, Any] | None = None) -> Any:
    """Return a model-facing tool clone with required ``intent`` in args schema.

    Tools may include injected runtime args (e.g. ``ToolRuntime`` / ``BaseStore``).
    Those annotations are not JSON-serializable; create the schema with
    ``arbitrary_types_allowed=True`` so wrapping does not crash the agent.
    LangChain's ``tool_call_schema`` still hides injected fields from the model.
    """
    if cache is not None:
        cached = cache.get(id(tool))
        if cached is not None:
            return cached

    get_schema = getattr(tool, "get_input_schema", None)
    if not callable(get_schema):
        return tool
    try:
        schema = get_schema()
    except Exception:  # noqa: BLE001
        return tool

    defs = _field_definitions_with_intent(schema)
    if defs is None:
        return tool

    from langchain_core.tools import StructuredTool
    from pydantic import ConfigDict, create_model

    title = getattr(schema, "__name__", None) or f"{_tool_name(tool)}Schema"
    try:
        # compact_conversation etc. carry ToolRuntime (contains BaseStore).
        NewModel = create_model(
            f"{title}WithIntent",
            __config__=ConfigDict(arbitrary_types_allowed=True),
            **defs,
        )
    except Exception:  # noqa: BLE001
        # Never break the whole turn because one tool schema is exotic.
        return tool

    def _sync(**kwargs: Any) -> Any:
        data = dict(kwargs)
        data.pop(TOOL_INTENT_KEY, None)
        return tool.invoke(data)

    async def _async(**kwargs: Any) -> Any:
        data = dict(kwargs)
        data.pop(TOOL_INTENT_KEY, None)
        return await tool.ainvoke(data)

    try:
        wrapped = StructuredTool.from_function(
            func=_sync,
            coroutine=_async,
            name=_tool_name(tool),
            description=getattr(tool, "description", None) or _tool_name(tool),
            args_schema=NewModel,
        )
    except Exception:  # noqa: BLE001
        return tool

    if cache is not None:
        cache[id(tool)] = wrapped
    return wrapped


def build_intent_schema_middleware():
    """Inject required ``intent`` into every tool schema the model sees.

    Execution path strips ``intent`` so original deepagents tools keep working.
    The stream/UI still sees ``intent`` on AI tool_calls and can render it.
    """
    cache: dict[int, Any] = {}

    def _apply_inject(request):  # type: ignore[no-untyped-def]
        tools = list(getattr(request, "tools", None) or [])
        if not tools:
            return request
        rewritten = [add_intent_to_tool(t, cache=cache) for t in tools]
        if rewritten != tools:
            return request.override(tools=rewritten)
        return request

    def _apply_strip(request):  # type: ignore[no-untyped-def]
        tool_call = getattr(request, "tool_call", None)
        if tool_call is None:
            return request
        new_call = _strip_intent_from_tool_call(tool_call)
        if new_call is not tool_call:
            try:
                return request.override(tool_call=new_call)
            except Exception:  # noqa: BLE001
                return request
        return request

    # Stack as a list: model-facing schema rewrite + tool-exec intent strip.
    return [
        _dual_wrap_model_call(name="require_tool_intent", apply=_apply_inject),
        _dual_wrap_tool_call(name="strip_tool_intent", apply=_apply_strip),
    ]


_prompt_cleanup_saved_tokens: contextvars.ContextVar[int] = contextvars.ContextVar(
    "synapse_prompt_cleanup_saved_tokens", default=0
)


def current_prompt_cleanup_saved_tokens() -> int:
    """Return prompt-cleanup savings for the current model-call context."""
    return max(0, int(_prompt_cleanup_saved_tokens.get() or 0))


# ---------------------------------------------------------------------------
#  Strip redundant prompt blocks injected by deepagents built-in middleware
# ---------------------------------------------------------------------------

# Text block prefixes injected by deepagents middleware.  Each block
# duplicates content already present in the corresponding tool definition
# (e.g. ``write_todos`` description, filesystem tool docs).  Stripping them
# saves ~720 tokens per model call without losing any capability.

_REDUNDANT_BLOCK_PREFIXES: tuple[str, ...] = (
    "\n\n## `write_todos`",
    "\n\n## Skills System",
    "\n\n## Following Conventions",
)


def build_strip_redundant_prompt_blocks():
    """Remove middleware-injected prompt blocks that duplicate tool definitions."""

    def _apply(request):  # type: ignore[no-untyped-def]
        _prompt_cleanup_saved_tokens.set(0)
        msg = getattr(request, "system_message", None)
        if msg is None or not hasattr(msg, "content_blocks"):
            return request
        blocks = msg.content_blocks
        if not blocks:
            return request

        filtered = []
        removed_chars = 0
        changed = False
        for block in blocks:
            text = block.get("text", "") if isinstance(block, dict) else ""
            if any(text.startswith(p) for p in _REDUNDANT_BLOCK_PREFIXES):
                changed = True
                removed_chars += len(text)
                continue
            filtered.append(block)

        if not changed:
            return request

        new_msg = msg.__class__(content_blocks=filtered)
        _prompt_cleanup_saved_tokens.set(max(0, (removed_chars + 3) // 4))
        return request.override(system_message=new_msg)

    return _dual_wrap_model_call(name="strip_redundant_prompt", apply=_apply)


# ---------------------------------------------------------------------------
#  Compact tool descriptions — replace verbose upstream descriptions with
#  concise alternatives to reduce prompt-cache-polluting tool-schema tokens.
# ---------------------------------------------------------------------------

# Per-tool short descriptions.  The upstream LangChain / deepagents defaults
# embed full-blown usage guides inside tool descriptions (3-13 KB each),
# pushing tool-schema tokens to ~134K per request.  Replacing them with
# one-sentence summaries keeps the essential signal while saving ~30K+ tokens
# of schema overhead (most of which is cached, but shorter schemas still
# reduce bandwidth and cache-eviction pressure).
_COMPACT_TOOL_DESCRIPTIONS: dict[str, str] = {
    "write_todos": (
        "Create and manage a structured task list for your current work session. "
        "Each todo has content and status: pending, in_progress, or completed. "
        "Only use for complex multi-step tasks (3+ steps); skip for trivial tasks."
    ),
    "execute": (
        "Execute a shell command in a sandbox environment. "
        "Returns combined stdout/stderr with exit code. "
        "Use timeout=SECONDS for long commands. "
        "Join multiple commands with && or ; (not newlines). "
        "Use glob/grep/read_file instead of find/grep/cat."
    ),
    "read_file": (
        "Read a file from the filesystem and return content with cat -n line numbers. "
        "Use pagination (offset/limit) for large files: read_file(path, offset=0, limit=100). "
        "Always read a file before editing it. "
        "Supports images, audio, video, and PDF via multimodal reads."
    ),
}


def build_compact_tool_descriptions(
    overrides: dict[str, str] | None = None,
) -> AgentMiddleware:
    """Build middleware that replaces verbose upstream tool descriptions.

    Args:
        overrides: Additional per-tool short descriptions merged on top of
            the built-in ``_COMPACT_TOOL_DESCRIPTIONS`` map.
    """
    descriptions = dict(_COMPACT_TOOL_DESCRIPTIONS)
    if overrides:
        descriptions.update(overrides)

    if not descriptions:
        return _dual_wrap_model_call(name="compact_tool_desc_noop", apply=lambda r: r)

    _compact_tool_desc_saved_tokens = contextvars.ContextVar[int](
        "compact_tool_desc_saved_tokens", default=0
    )

    def _apply(request):  # type: ignore[no-untyped-def]
        tools = getattr(request, "tools", None)
        if not tools:
            return request

        saved_chars = 0
        changed = False
        new_tools: list[Any] = []

        for tool in tools:
            if _is_tool_dict(tool):
                name, updated, chars = _compact_dict_tool(tool, descriptions)
            elif hasattr(tool, "name") and hasattr(tool, "description"):
                name, updated, chars = _compact_base_tool(tool, descriptions)
            else:
                new_tools.append(tool)
                continue

            if updated:
                changed = True
                saved_chars += chars
                new_tools.append(updated)
            else:
                new_tools.append(tool)

        if not changed:
            return request

        _compact_tool_desc_saved_tokens.set(
            max(0, (saved_chars + 3) // 4)
        )
        return request.override(tools=new_tools)

    return _dual_wrap_model_call(
        name="compact_tool_descriptions", apply=_apply
    )


def _is_tool_dict(obj: Any) -> bool:
    return isinstance(obj, dict) and (
        "function" in obj or "name" in obj
    )


def _compact_dict_tool(
    tool: dict[str, Any],
    descriptions: dict[str, str],
) -> tuple[str, Any, int]:
    """Replace description for OpenAI-function-format tools."""
    # OpenAI format: {"type": "function", "function": {"name": "...", "description": "..."}}
    fn = tool.get("function") if isinstance(tool.get("function"), dict) else tool
    name = str(fn.get("name", ""))
    short = descriptions.get(name)
    if short is None:
        return name, None, 0
    old_desc = str(fn.get("description", ""))
    if old_desc == short:
        return name, None, 0
    saved = max(0, len(old_desc.encode("utf-8")) - len(short.encode("utf-8")))
    if isinstance(tool.get("function"), dict):
        new_fn = dict(fn, description=short)
        return name, {**tool, "function": new_fn}, saved
    else:
        return name, {**tool, "description": short}, saved


def _compact_base_tool(
    tool: Any,
    descriptions: dict[str, str],
) -> tuple[str, Any, int]:
    """Replace description for LangChain BaseTool objects."""
    name = getattr(tool, "name", "")
    short = descriptions.get(name)
    if short is None:
        return name, None, 0
    old_desc = getattr(tool, "description", "")
    if old_desc == short:
        return name, None, 0
    saved = max(0, len(old_desc.encode("utf-8")) - len(short.encode("utf-8")))
    new_tool = tool.model_copy(update={"description": short})
    return name, new_tool, saved
