butler/transport/streaming_signal.py

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

"""Transport-level streaming-signal dispatch.

This module replaces the former ``butler.core.streaming_tools`` dependency
that ``butler/transport/llm_client.py`` had at lines 360, 383, 476. The
audit (``docs/reviews/project-deep-audit-2026-06-r1to8.md`` §R1-1)
classified that reach as a layering violation: ``transport/`` is the
lowest layer of the v4 architecture and must not import from ``core/``.

The dispatch logic itself is pure (it inspects a collected tool-call dict
and fires a callback once per complete, streaming-eligible call), so it
lives naturally in transport. The helpers here intentionally mirror the
ones still used by ``butler.core.streaming_tools`` for core-side concerns
(``batch_eligible_for_streaming`` etc.) — they are small, self-contained,
and keep the layering boundary clean.
"""

from __future__ import annotations

import json
import logging
from typing import Any, Callable

from butler.utilities.env_parse import env_truthy
from butler.defaults.env_defaults import TRANSPORT_STREAMING_TOOLS_DEFAULT

logger = logging.getLogger(__name__)

_STREAMING_TOOL_NAMES = frozenset({
    "read_file",
    "grep",
    "search_files",
    "list_directory",
    "glob",
    "list_dir",
})


def _streaming_tools_enabled() -> bool:
    return bool(env_truthy("BUTLER_STREAMING_TOOLS", default=TRANSPORT_STREAMING_TOOLS_DEFAULT))


def _is_streaming_tool(name: str) -> bool:
    n = (name or "").strip().lower()
    if n in _STREAMING_TOOL_NAMES:
        return True
    return n.startswith(("read_", "search_", "grep"))


def _try_parse_tool_arguments(args_str: str) -> dict[str, Any] | None:
    text = (args_str or "").strip()
    if not text:
        return None
    try:
        parsed = json.loads(text)
    except (TypeError, json.JSONDecodeError):
        return None
    return parsed if isinstance(parsed, dict) else None


def notify_complete_tool_calls_from_stream(
    collected: dict[int, dict[str, Any]],
    on_tool_call_ready: Callable[[int, str, str, dict[str, Any]], None] | None,
) -> None:
    """Invoke ``on_tool_call_ready`` for each tool call whose JSON args are complete (once).

    Mirrors the contract of the former core-side implementation so callers
    in ``llm_client.py`` do not need to know which layer hosts the helper.
    """
    if not on_tool_call_ready or not _streaming_tools_enabled():
        return
    for idx in sorted(collected.keys()):
        entry = collected[idx]
        if entry.get("_stream_dispatched"):
            continue
        name = str(entry.get("name") or "").strip()
        if not _is_streaming_tool(name):
            continue
        args = _try_parse_tool_arguments(str(entry.get("arguments") or ""))
        if args is None:
            continue
        entry["_stream_dispatched"] = True
        tool_id = str(entry.get("id") or f"call_{idx}")
        from butler.transport.streaming_signal_ops import invoke_streaming_tool_callback_safe

        invoke_streaming_tool_callback_safe(on_tool_call_ready, idx, tool_id, name, args)