butler/gateway/locked_phases/format_phase.py

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

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from butler.core.best_effort import safe_best_effort
from butler.core.memory_recap_line import maybe_prepend_memory_recap
from butler.core.session_hydration import recovery_notice_text
from butler.core.turn_summary_line import maybe_prepend_turn_summary
from butler.gateway.item_event_sink import recent_thread_items
from butler.gateway.outbound_bridge import get_current_bridge
from butler.ops import langfuse_tracer

from .state import LockedTurnState

if TYPE_CHECKING:
    from butler.gateway.message_handler import ButlerMessageHandler

logger = logging.getLogger(__name__)


def _record_format_turn_langfuse(state: LockedTurnState) -> None:
    def _record() -> None:
        if langfuse_tracer.langfuse_enabled():
            ctx = langfuse_tracer.get_current_trace(session_key=state.session_key)
            if ctx is not None:
                ctx.on_gateway_outbound(state.session_key, len(state.out or ""), state.turn_elapsed)

    safe_best_effort(_record, label="locked_phases.langfuse_outbound")


def _append_format_turn_extras(state: LockedTurnState, welcome_prefix: str = "") -> None:
    if welcome_prefix:
        state.out = f"{welcome_prefix}\n\n---\n\n{state.out}" if state.out else welcome_prefix

    def _recovery_notice() -> None:
        if getattr(state.loop, "_session_recovery_pending", None) is not True:
            return
        note = recovery_notice_text()
        state.out = f"{note}\n\n{state.out}" if state.out else note
        setattr(state.loop, "_session_recovery_pending", False)
        state.health["session_recovery_notice"] = True

    safe_best_effort(_recovery_notice, label="locked_phases.session_recovery_notice")

    def _turn_summary() -> None:
        state.out = maybe_prepend_turn_summary(state.session_key, state.out or "")

    safe_best_effort(_turn_summary, label="locked_phases.turn_summary_line")

    def _memory_recap() -> None:
        state.out = maybe_prepend_memory_recap(
            state.session_key,
            state.out or "",
            health=state.health,
        )

    safe_best_effort(_memory_recap, label="locked_phases.memory_recap_line")


def _phase_format_turn_response(
    handler: "ButlerMessageHandler",
    state: LockedTurnState,
    welcome_prefix: str = "",
) -> None:
    import time as _time

    state.out = handler._format_response(state.result, state.platform)
    state.turn_elapsed = _time.monotonic() - state.turn_started
    br = get_current_bridge()
    if br is not None:
        br.record_turn_elapsed(state.turn_elapsed)
        state.health["outbound_events"] = br.recent_outbound_events()[-8:]

    def _thread_items() -> None:
        items = recent_thread_items(8)
        if items:
            state.health["thread_items"] = items

    safe_best_effort(_thread_items, label="locked_phases.thread_items")
    logger.info(
        "Gateway turn done session=%s elapsed=%.1fs out_len=%d",
        state.session_key,
        state.turn_elapsed,
        len(state.out or ""),
    )
    _record_format_turn_langfuse(state)
    _append_format_turn_extras(state, welcome_prefix)