butler/tools/network_search_policy_ops.py

Rama main · réplica de lectura del repositorio original.

"""Network search policy best-effort helpers (P0-A)."""

from __future__ import annotations

from typing import Any

from butler.core.best_effort import safe_best_effort


def is_github_mcp_intent_safe(query: str) -> bool:
    def _run() -> bool:
        from butler.mcp.github_grounding import (
            is_github_issue_list_intent,
            is_github_repo_list_intent,
        )

        return bool(
            is_github_repo_list_intent(query) or is_github_issue_list_intent(query)
        )

    result = safe_best_effort(
        _run,
        label="network_search.github_mcp_intent",
        default=False,
    )
    return bool(result)


def get_turn_bridge_safe() -> Any | None:
    def _run() -> Any:
        from butler.execution_context import get_current_turn_bridge

        return get_current_turn_bridge()

    return safe_best_effort(
        _run,
        label="network_search.turn_bridge",
        default=None,
    )


def epoch_user_query_safe() -> str:
    def _run() -> str:
        from butler.core.session_epoch import last_user_query_in_epoch
        from butler.execution_context import get_current_session_key

        sk = str(get_current_session_key() or "").strip()
        if sk:
            return str(last_user_query_in_epoch(sk) or "")
        return ""

    result = safe_best_effort(
        _run,
        label="network_search.epoch_query",
        default="",
    )
    return str(result or "")


def web_search_in_current_toolset_safe(*, fallback: bool) -> bool:
    def _run() -> bool:
        from butler.execution_context import get_current_orchestrator, get_current_session_key
        from butler.tools.project_tools import get_tool_definitions_for_project

        orch = get_current_orchestrator()
        if orch is None:
            return True
        pm = getattr(orch, "project_manager", None)
        if pm is None:
            return True
        proj = pm.get_current(session_key=str(get_current_session_key() or ""))
        for role in ("lead", "butler", "content", "dev", "review"):
            tools = get_tool_definitions_for_project(proj, role=role)
            names = {
                str((t.get("function") or {}).get("name") or "")
                for t in tools
                if isinstance(t, dict)
            }
            if "web_search" in names:
                return True
        return False

    result = safe_best_effort(
        _run,
        label="network_search.toolset_web_search",
        default=fallback,
    )
    return bool(result)