"""R1-6 inbound pipeline phases for ``ButlerMessageHandler.handle_message``.

Audit source: ``docs/reviews/project-deep-audit-2026-06-r1to8.md`` §R1-6

The original ``handle_message`` (243-617, ~334 non-blank lines) was a
god method that mixed session resolution, text transformation, guard
checks (io_guardrail, human_gate, injection_guard, injection_llm,
bot_loop_guard, two_phase_confirm), pre-dispatch rewrites,
idempotency, session initializing, queue dispatch, and reply
admission — all in one linear flow with no test seams.

This module exposes the **pre-session** pipeline as composable phase
functions. Each takes the handler instance plus the per-turn state it
needs and returns either:

* ``None``  — pipeline should continue to the next phase
* ``str``   — terminal user-facing reply (early return)

Post-condition: every phase helper is a thin orchestrator under 50
source lines (R1-5.2 size contract — see
``tests/test_message_handler_split.py``).
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, Optional, cast

if TYPE_CHECKING:
    from butler.gateway.message_handler import ButlerMessageHandler

logger = logging.getLogger(__name__)

from butler.core.message_ir import inbound_text_from_gateway
from butler.core.session_transcript import append_transcript_entry
from butler.core.steer import format_steer_gateway_reply, is_run_active, steer
from butler.core.two_phase_confirm import cancel_pending_unless_confirm, try_execute_pending_confirm
from butler.gateway.bot_loop_guard import record_and_should_suppress
from butler.gateway.handler_helpers import (
    _is_prequeue_interrupt_command,
    apply_auto_continue_rewrite,
)
from butler.gateway.hooks import apply_pre_gateway_dispatch
from butler.resilience.inbound_idempotency import check_and_reserve_inbound, record_duplicate_skip
from butler.gateway.message_pipelines_fail_closed import (
    apply_human_gate_fail_closed,
    apply_injection_guard_fail_closed,
    apply_injection_llm_fail_closed,
    apply_io_guardrail_fail_closed,
)
from butler.resilience.message_queue import (
    enqueue_inbound,
    format_queued_ack,
    pending_count,
)
from butler.gateway.owner_gate import is_gateway_owner, owner_required_message
from butler.gateway.queue_settings import get_queue_mode, session_drop_policy
from butler.gateway.reply_admission import try_admit
from butler.gateway.session_lifecycle import (
    format_initializing_ack,
    session_initializing_enabled,
    try_enter_session,
)
from butler.mcp.profiles import mcp_profiles_enabled, select_profile_for_text, set_session_profile
from butler.session.keys import chat_id_from_session_key
from butler.skills.similarity import _ensure_jieba
from butler.core.best_effort import safe_best_effort


def _record_injection_transcript(
    session_key: str,
    entry_type: str,
    score: float,
    text: str,
    *,
    label: str,
) -> None:

    def _run() -> None:
        append_transcript_entry(
            session_key,
            entry_type,
            {"score": score, "preview": text[:120]},
        )

    safe_best_effort(_run, label=label, default=None)


def _warm_session_skills(orchestrator: Any) -> None:
    def _jieba() -> None:
        _ensure_jieba()

    safe_best_effort(_jieba, label="message_pipelines.jieba_warmup", default=None)
    mgr = getattr(orchestrator, "_skill_manager", None)
    if mgr is not None:
        mgr.list_skills()


# ---------------------------------------------------------------------------
# Phase 1 — resolve the canonical ``platform:chat_id:project`` session key.
# ---------------------------------------------------------------------------

def _phase_resolve_session_key(
    handler: "ButlerMessageHandler",
    *,
    platform: str,
    external_id: str | None,
    session_key: str | None,
) -> str:
    """Phase: normalize inbound args to a canonical session key."""
    return cast(
        str,
        handler.resolve_session_key(
            platform=platform,
            external_id=external_id,
            session_key=session_key,
        ),
    )


# ---------------------------------------------------------------------------
# Phase 2 — apply platform-specific text transformation (message_ir).
# ---------------------------------------------------------------------------

def _phase_transform_inbound_text(
    text: str,
    *,
    platform: str,
    external_id: str | None,
    session_key: str,
) -> str:
    """Phase: delegate to ``message_ir.inbound_text_from_gateway``.

    Wrapped in try/except to preserve the original fail-open behavior:
    if the transformer is missing or raises, the raw text is used.
    """
    def _run() -> str:
        return cast(
            str,
            inbound_text_from_gateway(
                text,
                platform=platform,
                external_id=external_id,
                session_key=session_key,
            ),
        )

    transformed = safe_best_effort(
        _run,
        label="message_pipelines.inbound_text_transform",
        default=None,
    )
    return cast(str, transformed if transformed is not None else text)


# ---------------------------------------------------------------------------
# Phase 3 — select the MCP profile for the session based on the text.
# ---------------------------------------------------------------------------

def _phase_apply_mcp_profile(text: str, session_key: str) -> None:
    """Phase: select and bind the MCP profile for this session."""

    def _run() -> None:
        if mcp_profiles_enabled() and text.strip():
            set_session_profile(session_key, select_profile_for_text(text))

    safe_best_effort(_run, label="message_pipelines.mcp_profile", default=None)


# ---------------------------------------------------------------------------
# Phase 4 — io guardrail (fail-closed on exception).
# ---------------------------------------------------------------------------

def _phase_apply_io_guardrail(text: str) -> Optional[str]:
    """Phase: io guardrail tripwire. Returns a blocking reply or None."""
    return cast(Optional[str], apply_io_guardrail_fail_closed(text))


# ---------------------------------------------------------------------------
# Phase 5 — human gate (with owner check on gate reply).
# ---------------------------------------------------------------------------

def _phase_apply_human_gate(
    text: str,
    session_key: str,
    *,
    platform: str,
    external_id: str | None,
) -> Optional[str]:
    """Phase: human gate — blocks (with owner check) if pending review."""
    return cast(
        Optional[str],
        apply_human_gate_fail_closed(
            text,
            session_key,
            platform=platform,
            external_id=external_id,
        ),
    )


# ---------------------------------------------------------------------------
# Phase 6 — injection risk scoring (heuristic).
# ---------------------------------------------------------------------------

def _phase_apply_injection_guard(
    text: str, session_key: str,
) -> tuple[str, Optional[str]]:
    """Phase: heuristic injection scoring + adversarial marker rewrite.

    Returns ``(new_text, block_or_None)``. When ``block`` is set,
    the caller must short-circuit and return it as the final reply
    (fail-closed semantics from the original ``handle_message``).
    """
    return cast(
        tuple[str, Optional[str]],
        apply_injection_guard_fail_closed(
            text,
            session_key,
            record_injection_transcript=_record_injection_transcript,
        ),
    )


# ---------------------------------------------------------------------------
# Phase 7 — LLM-based injection gate (with bypass/pending handling).
# ---------------------------------------------------------------------------

def _phase_apply_injection_llm(text: str, session_key: str) -> Optional[str]:
    """Phase: LLM-driven injection gate. Returns a blocking reply or None."""
    return cast(
        Optional[str],
        apply_injection_llm_fail_closed(
            text,
            session_key,
            record_injection_transcript=_record_injection_transcript,
        ),
    )


# ---------------------------------------------------------------------------
# Phase 8 — bot loop guard (group chat echo loop prevention).
# ---------------------------------------------------------------------------

def _phase_apply_bot_loop_guard(
    text: str,
    session_key: str,
    *,
    external_id: str | None,
) -> Optional[str]:
    """Phase: suppress group-chat bot echo loops."""
    def _run() -> Optional[str]:
        chat_id = session_key.split(":")[-1] if ":" in session_key else session_key
        suppress, _reason = record_and_should_suppress(
            chat_id=chat_id,
            sender_id=str(external_id or ""),
            text=text,
        )
        if suppress:
            return "（已忽略：群聊 bot 互回复环防护）"
        return None

    return cast(
        Optional[str],
        safe_best_effort(_run, label="message_pipelines.bot_loop_guard", default=None),
    )


# ---------------------------------------------------------------------------
# Phase 9 — two-phase confirm (with owner check on confirm reply).
# ---------------------------------------------------------------------------

def _phase_apply_two_phase_confirm(
    text: str,
    session_key: str,
    *,
    platform: str,
    external_id: str | None,
) -> Optional[str]:
    """Phase: dispatch / confirm / cancel two-phase confirm flow."""
    def _run() -> Optional[str]:
        pending_reply = try_execute_pending_confirm(text, session_key=session_key)
        if pending_reply is not None:
            if not is_gateway_owner(platform=platform, external_id=external_id):
                return cast(str, owner_required_message())
            return cast(str, pending_reply)
        cancel_note = cancel_pending_unless_confirm(text, session_key=session_key)
        if cancel_note:
            return f"{cancel_note}\n\n{text}"
        return None

    return cast(
        Optional[str],
        safe_best_effort(
            _run,
            label="message_pipelines.two_phase_confirm",
            default=None,
        ),
    )


# ---------------------------------------------------------------------------
# Phase 10 — prequeue interrupt command.
# ---------------------------------------------------------------------------

def _phase_apply_prequeue_interrupt(
    text: str,
    session_key: str,
    handler: "ButlerMessageHandler",
) -> Optional[str]:
    """Phase: prequeue interrupt command (e.g. /停止)."""
    if _is_prequeue_interrupt_command(text):
        return cast(str, handler._format_prequeue_interrupt_reply(session_key))
    return None


# ---------------------------------------------------------------------------
# Phase 11 — pre-dispatch text rewrites (auto_continue + pre_gateway_dispatch).
# ---------------------------------------------------------------------------

def _phase_apply_pre_dispatch_rewrites(
    text: str,
    session_key: str,
    *,
    platform: str,
) -> Optional[str]:
    """Phase: rewrite text via auto_continue and pre_gateway_dispatch hooks.

    Returns ``None`` to mean "use original text", ``""`` to mean
    "drop this turn", or a non-empty string to mean "use as new text".

    Note: the original code did **not** wrap ``apply_pre_gateway_dispatch``
    in try/except — a hook failure propagated. We preserve that
    fail-loud behavior here on purpose.
    """
    continued = apply_auto_continue_rewrite(session_key, text)
    if continued:
        text = continued
    rewritten = apply_pre_gateway_dispatch(text, session_key=session_key, platform=platform)
    if rewritten is None:
        return None
    if not rewritten.strip():
        return ""
    return cast(Optional[str], rewritten)


# ---------------------------------------------------------------------------
# Phase 12 — inbound idempotency (Sprint 14 REL-11-1 reservation).
# ---------------------------------------------------------------------------

def _phase_apply_idempotency(
    text: str,
    session_key: str,
    *,
    external_id: str | None,
) -> tuple[Optional[str], bool, str]:
    """Phase: idempotency reservation.

    Returns ``(early_reply_or_None, reserved, inbound_id)``.
    ``reserved=True`` means the caller is responsible for calling
    ``complete_inbound`` in the ``finally`` block.
    """
    def _run() -> tuple[Optional[str], bool, str]:
        inbound_id = str(external_id or "").strip()
        if inbound_id and inbound_id == chat_id_from_session_key(session_key):
            inbound_id = ""
        _idem = check_and_reserve_inbound(
            session_key,
            inbound_id,
            text_preview=text[:80],
        )
        if _idem.skip:
            record_duplicate_skip(
                session_key,
                reason=_idem.reason,
                external_id=inbound_id,
                preview=text[:80],
            )
            return (_idem.user_reply or "（重复消息已忽略。）"), False, inbound_id
        return None, True, inbound_id

    return cast(
        tuple[Optional[str], bool, str],
        safe_best_effort(
            _run,
            label="message_pipelines.idempotency",
            default=(None, False, ""),
        ),
    )


# ---------------------------------------------------------------------------
# Phase 13 — session initializing (warmup + queue while spinning up).
# ---------------------------------------------------------------------------

def _phase_apply_session_initializing(
    text: str,
    session_key: str,
    *,
    platform: str,
    external_id: str | None,
    orchestrator: Any,
) -> Optional[str]:
    """Phase: session-initializing warmup. Returns early ack or None."""
    def _run() -> Optional[str]:
        if not session_initializing_enabled():
            return None

        state = try_enter_session(session_key, lambda: _warm_session_skills(orchestrator))
        if state != "queued":
            return None
        if enqueue_inbound(
            session_key,
            text,
            platform=platform,
            external_id=external_id or "",
        ):
            return cast(
                str,
                format_initializing_ack(pending=pending_count(session_key)),
            )
        return cast(str, format_initializing_ack())

    return cast(
        Optional[str],
        safe_best_effort(
            _run,
            label="message_pipelines.session_initializing",
            default=None,
        ),
    )


# ---------------------------------------------------------------------------
# Phase 14 — queue inbound (steer / interrupt / collect / followup).
# ---------------------------------------------------------------------------

def _phase_apply_queue_inbound(
    text: str,
    session_key: str,
    *,
    platform: str,
    external_id: str | None,
    handler: "ButlerMessageHandler",
) -> Optional[str]:
    """Phase: enqueue or steer per queue_mode; return ack or None.

    Note: the original code did **not** wrap the queue logic in
    try/except — a queue failure propagated. We preserve that
    fail-loud behavior here on purpose.
    """
    if not handler._should_queue_inbound(session_key, text):
        return None
    mode = get_queue_mode(session_key)
    if mode == "steer":
        if is_run_active(session_key) and steer(text, session_key=session_key):
            return cast(str, format_steer_gateway_reply(accepted=True, active=True))
        return None
    if mode == "interrupt":
        handler._format_prequeue_interrupt_reply(session_key)
        return None
    if enqueue_inbound(
        session_key,
        text,
        platform=platform,
        external_id=external_id or "",
    ):
        return cast(
            str,
            format_queued_ack(
                pending=pending_count(session_key),
                session_key=session_key,
            ),
        )
    if session_drop_policy(session_key) == "new":
        return "队列已满，最新消息未入队。可发 /queue 调整 cap 或 /诊断 查看。"
    return cast(
        str,
        format_queued_ack(pending=pending_count(session_key), session_key=session_key),
    )


# ---------------------------------------------------------------------------
# Phase 15 — reply admission (single-flight gate).
# ---------------------------------------------------------------------------

def _phase_apply_admission(text: str, session_key: str) -> Optional[Any]:
    """Phase: try_admit. Returns admission token (caller releases) or None."""
    def _run() -> Any:
        return try_admit(session_key)

    return safe_best_effort(
        _run,
        label="message_pipelines.reply_admission",
        default=True,
    )


def queue_inbound_for_admission_failure(
    text: str,
    session_key: str,
    *,
    platform: str,
    external_id: str | None,
) -> str:
    """When ``try_admit`` returns ``None`` we either enqueue or reply."""
    def _run() -> Optional[str]:
        if enqueue_inbound(
            session_key,
            text,
            platform=platform,
            external_id=external_id or "",
        ):
            return cast(
                str,
                format_queued_ack(
                    pending=pending_count(session_key),
                    session_key=session_key,
                ),
            )
        return None

    queued = safe_best_effort(
        _run,
        label="message_pipelines.admission_failure_enqueue",
        default=None,
    )
    if queued:
        return cast(str, queued)
    return "会话处理中，请稍候…"