butler/gateway/locked_phases/finalize_phase.py

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

from __future__ import annotations

from typing import TYPE_CHECKING

from butler.core.auto_continue import capture_auto_continue_pending
from butler.core.best_effort import safe_best_effort
from butler.core.compaction_status import promote_compaction_diagnostics_to_health
from butler.core.loop_types import LoopStatus
from butler.core.memory_source_surface import snapshot_last_turn_memory_sources
from butler.core.transform_feedback import maybe_apply_turn_feedback
from butler.memory.memory_metrics import get_collector as _memory_metrics_collector
from butler.memory.metrics_persist import flush_memory_metrics
from butler.memory.prefetch_retrieval_metrics import finalize_prefetch_retrieval_metrics
from butler.ops import langfuse_tracer
from butler.ops.eval_turn import extract_tools_used, push_turn_scores
from butler.session.lifecycle import queue_prefetch_after_turn, sync_turn_memory

from .state import LockedTurnState

if TYPE_CHECKING:
    from butler.gateway.message_handler import ButlerMessageHandler


def _phase_finalize_loop_diagnostics(state: LockedTurnState) -> None:
    loop_diag = dict(getattr(state.result, "diagnostics", {}) or {})
    state.health["loop"] = loop_diag
    if getattr(state.result, "transition_reason", ""):
        state.health["loop_transition_reason"] = state.result.transition_reason

    def _promote_diag() -> None:
        promote_compaction_diagnostics_to_health(state.health, loop_diag)
        mm = _memory_metrics_collector().get_session_metrics(state.session_key)
        if "error" not in mm:
            state.health["memory_metrics"] = mm.get("computed") or {}
            for key in (
                "facts_pre_compact",
                "facts_post_compact",
                "anchor_facts_pre",
                "anchor_facts_post",
                "prefetch_turns",
                "prefetch_hits",
            ):
                if key in mm:
                    state.health[key] = mm[key]

    safe_best_effort(_promote_diag, label="locked_phases.compaction_diag")


def _phase_finalize_interrupt_capture(state: LockedTurnState) -> None:
    if state.result.status != LoopStatus.INTERRUPTED:
        return

    def _capture_interrupt() -> None:
        capture_auto_continue_pending(
            state.session_key,
            user_preview=state.augmented,
            reason="interrupt",
            diagnostics=state.health.get("loop")
            if isinstance(state.health.get("loop"), dict)
            else None,
        )

    safe_best_effort(_capture_interrupt, label="locked_phases.auto_continue_capture")


def _phase_finalize_memory_sync(
    handler: "ButlerMessageHandler",
    state: LockedTurnState,
) -> None:
    sync_result = sync_turn_memory(
        handler._orchestrator,
        state.text,
        state.result.final_response or "",
        interrupted=state.result.status == LoopStatus.INTERRUPTED,
        status=state.result.status,
        session_id=state.session_key,
    )
    state.health["memory_sync"] = sync_result
    queue_prefetch_after_turn(
        handler._orchestrator,
        state.text,
        role=state.loop_role,
        session_id=state.session_key,
    )


def _phase_finalize_eval_observability(state: LockedTurnState) -> None:
    def _langfuse_turn_end() -> None:
        if not langfuse_tracer.langfuse_enabled():
            return
        trace_id = ""
        ctx = langfuse_tracer.get_current_trace(session_key=state.session_key)
        if ctx is not None:
            trace_id = ctx.trace_id
        multi, eval_report = push_turn_scores(
            user_text=state.text,
            response_text=state.result.final_response or "",
            tools_used=extract_tools_used(getattr(state.result, "diagnostics", None)),
            session_id=state.session_key,
            trace_id=trace_id,
        )
        state.health["eval_turn"] = {
            "overall": round(multi.overall, 3),
            "dims": multi.by_dimension(),
            "scores_pushed": eval_report.scores_pushed,
        }

        def _transform_feedback() -> None:
            provider = ""
            loop = getattr(state.result, "loop", None)
            client = getattr(loop, "client", None) if loop else None
            if client is not None:
                provider = str(getattr(client, "provider_name", "") or "")
            actions = maybe_apply_turn_feedback(
                multi.by_dimension(),
                provider=provider,
            )
            if actions:
                state.health["transform_feedback"] = actions

        safe_best_effort(_transform_feedback, label="locked_phases.transform_feedback")
        langfuse_tracer.end_trace(session_key=state.session_key, result=state.result)
        langfuse_tracer.flush_langfuse()

    safe_best_effort(_langfuse_turn_end, label="locked_phases.langfuse_turn_end")
    safe_best_effort(
        lambda: flush_memory_metrics(force=True),
        label="locked_phases.memory_metrics_flush",
    )


def _phase_finalize_prefetch_pr(state: LockedTurnState) -> None:
    def _finalize() -> None:
        finalize_prefetch_retrieval_metrics(
            state.session_key,
            state.result.final_response or "",
            state.health,
        )

    safe_best_effort(_finalize, label="locked_phases.prefetch_pr_finalize")


def _phase_finalize_turn(
    handler: "ButlerMessageHandler",
    state: LockedTurnState,
) -> None:
    _phase_finalize_loop_diagnostics(state)
    _phase_finalize_interrupt_capture(state)
    _phase_finalize_memory_sync(handler, state)
    _phase_finalize_prefetch_pr(state)

    def _snapshot_sources() -> None:
        snapshot_last_turn_memory_sources(state.health)

    safe_best_effort(_snapshot_sources, label="locked_phases.memory_sources_snapshot")
    handler._session_registry.set_health(state.session_key, state.health)
    _phase_finalize_eval_observability(state)